OpenAI Assistants API #
什么是 Assistants API？ #
Assistants API 是 OpenAI 提供的一套用于构建 AI 助手的 API。它封装了状态管理、对话历史、工具调用等功能，让开发者能够更轻松地构建复杂的 AI 应用。
text
┌─────────────────────────────────────────────────────────────┐
│ Assistants API 架构 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Assistant │ │
│ │ ┌───────────┐ ┌───────────┐ ┌───────────┐ │ │
│ │ │ 指令设置 │ │ 模型选择 │ │ 工具配置 │ │ │
│ │ └───────────┘ └───────────┘ └───────────┘ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Thread │ │
│ │ ┌─────────────────────────────────────────────┐ │ │
│ │ │ Messages: [用户消息, AI回复, 工具结果...] │ │ │
│ │ └─────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Run │ │
│ │ 执行对话，调用工具，生成回复 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘
核心概念 #
text
┌─────────────────────────────────────────────────────────────┐
│ 核心概念 │
├─────────────────────────────────────────────────────────────┤
│ │
│ Assistant（助手） │
│ ───────────────────────────────────────────────────────── │
│ 定义 AI 助手的身份、能力和行为 │
│ 包含：指令、模型、工具 │
│ │
│ Thread（线程） │
│ ───────────────────────────────────────────────────────── │
│ 对话会话，存储消息历史 │
│ 支持持久化，可跨请求保持状态 │
│ │
│ Message（消息） │
│ ───────────────────────────────────────────────────────── │
│ 对话中的单条消息 │
│ 可以是用户消息或助手回复 │
│ │
│ Run（运行） │
│ ───────────────────────────────────────────────────────── │
│ 执行一次对话处理 │
│ 管理状态流转和工具调用 │
│ │
└─────────────────────────────────────────────────────────────┘
基本用法 #
创建助手 #
python
from openai import OpenAI
client = OpenAI()
assistant = client.beta.assistants.create(
name="数学辅导助手",
instructions="你是一个专业的数学辅导老师。用清晰、易懂的方式解释数学概念，并提供练习题。",
model="gpt-4o-mini"
)
print(f"助手 ID: {assistant.id}")
创建线程并添加消息 #
python
thread = client.beta.threads.create()
message = client.beta.threads.messages.create(
thread_id=thread.id,
role="user",
content="请解释什么是勾股定理？"
)
print(f"线程 ID: {thread.id}")
print(f"消息 ID: {message.id}")
运行助手 #
python
run = client.beta.threads.runs.create(
thread_id=thread.id,
assistant_id=assistant.id
)
print(f"运行 ID: {run.id}")
print(f"状态: {run.status}")
获取回复 #
python
import time
while True:
run = client.beta.threads.runs.retrieve(
thread_id=thread.id,
run_id=run.id
)
if run.status == "completed":
break
elif run.status == "failed":
print("运行失败")
break
time.sleep(1)
messages = client.beta.threads.messages.list(
thread_id=thread.id
)
for msg in messages.data:
if msg.role == "assistant":
print(msg.content[0].text.value)
完整示例 #
python
from openai import OpenAI
import time
client = OpenAI()
class AssistantChat:
def __init__(self, name: str, instructions: str, model: str = "gpt-4o-mini"):
self.client = OpenAI()
self.assistant = self.client.beta.assistants.create(
name=name,
instructions=instructions,
model=model
)
self.thread = self.client.beta.threads.create()
def chat(self, user_input: str) -> str:
self.client.beta.threads.messages.create(
thread_id=self.thread.id,
role="user",
content=user_input
)
run = self.client.beta.threads.runs.create(
thread_id=self.thread.id,
assistant_id=self.assistant.id
)
while True:
run = self.client.beta.threads.runs.retrieve(
thread_id=self.thread.id,
run_id=run.id
)
if run.status == "completed":
break
elif run.status in ["failed", "cancelled", "expired"]:
raise Exception(f"运行失败: {run.status}")
time.sleep(0.5)
messages = self.client.beta.threads.messages.list(
thread_id=self.thread.id,
limit=1,
order="desc"
)
return messages.data[0].content[0].text.value
def get_history(self):
messages = self.client.beta.threads.messages.list(
thread_id=self.thread.id
)
return messages.data
chat = AssistantChat(
name="编程助手",
instructions="你是一个专业的编程助手，帮助用户解决编程问题。"
)
print(chat.chat("什么是 Python 的列表推导式？"))
print(chat.chat("能给我一个例子吗？"))
工具功能 #
代码解释器 #
python
assistant = client.beta.assistants.create(
name="数据分析助手",
instructions="你是一个数据分析专家，使用 Python 进行数据分析。",
model="gpt-4o-mini",
tools=[{"type": "code_interpreter"}]
)
thread = client.beta.threads.create()
client.beta.threads.messages.create(
thread_id=thread.id,
role="user",
content="请计算 1 到 100 的和，并画出图表。"
)
run = client.beta.threads.runs.create(
thread_id=thread.id,
assistant_id=assistant.id
)
文件检索 #
python
file = client.files.create(
file=open("document.pdf", "rb"),
purpose="assistants"
)
assistant = client.beta.assistants.create(
name="文档助手",
instructions="根据上传的文档回答问题。",
model="gpt-4o-mini",
tools=[{"type": "file_search"}],
tool_resources={
"file_search": {
"vector_stores": [{
"file_ids": [file.id]
}]
}
}
)
函数调用 #
python
assistant = client.beta.assistants.create(
name="天气助手",
instructions="你是一个天气助手，帮助用户查询天气信息。",
model="gpt-4o-mini",
tools=[
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取指定城市的天气",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称"
}
},
"required": ["city"]
}
}
}
]
)
处理工具调用 #
python
import json
def handle_tool_calls(run, thread_id):
tool_calls = run.required_action.submit_tool_outputs.tool_calls
tool_outputs = []
for tool_call in tool_calls:
if tool_call.function.name == "get_weather":
args = json.loads(tool_call.function.arguments)
result = get_weather(args["city"])
tool_outputs.append({
"tool_call_id": tool_call.id,
"output": json.dumps(result)
})
client.beta.threads.runs.submit_tool_outputs(
thread_id=thread_id,
run_id=run.id,
tool_outputs=tool_outputs
)
def get_weather(city: str) -> dict:
return {"city": city, "temp": 25, "condition": "晴"}
def run_with_tools(thread_id: str, assistant_id: str):
run = client.beta.threads.runs.create(
thread_id=thread_id,
assistant_id=assistant_id
)
while True:
run = client.beta.threads.runs.retrieve(
thread_id=thread_id,
run_id=run.id
)
if run.status == "completed":
break
elif run.status == "requires_action":
handle_tool_calls(run, thread_id)
elif run.status in ["failed", "cancelled", "expired"]:
raise Exception(f"运行失败: {run.status}")
time.sleep(0.5)
messages = client.beta.threads.messages.list(
thread_id=thread_id,
limit=1,
order="desc"
)
return messages.data[0].content[0].text.value
文件处理 #
上传文件 #
python
file = client.files.create(
file=open("data.csv", "rb"),
purpose="assistants"
)
print(f"文件 ID: {file.id}")
print(f"文件名: {file.filename}")
附加文件到消息 #
python
file = client.files.create(
file=open("report.pdf", "rb"),
purpose="assistants"
)
message = client.beta.threads.messages.create(
thread_id=thread.id,
role="user",
content="请分析这个报告",
attachments=[
{
"file_id": file.id,
"tools": [{"type": "file_search"}]
}
]
)
获取生成的文件 #
python
messages = client.beta.threads.messages.list(thread_id=thread.id)
for message in messages.data:
for content in message.content:
if content.type == "image_file":
file_id = content.image_file.file_id
image_data = client.files.content(file_id)
with open("generated_image.png", "wb") as f:
f.write(image_data.content)
流式响应 #
使用流式输出 #
python
from openai import OpenAI
client = OpenAI()
def stream_assistant_response(thread_id: str, assistant_id: str):
with client.beta.threads.runs.stream(
thread_id=thread_id,
assistant_id=assistant_id
) as stream:
for event in stream:
if event.event == "thread.message.delta":
if hasattr(event.data.delta.content[0], 'text'):
text = event.data.delta.content[0].text.value
print(text, end="", flush=True)
thread = client.beta.threads.create()
client.beta.threads.messages.create(
thread_id=thread.id,
role="user",
content="写一个关于人工智能的故事"
)
stream_assistant_response(thread.id, assistant.id)
状态管理 #
运行状态 #
text
┌─────────────────────────────────────────────────────────────┐
│ Run 状态流转 │
├─────────────────────────────────────────────────────────────┤
│ │
│ queued │
│ │ │
│ ▼ │
│ in_progress │
│ │ │
│ ├──► requires_action ──► (提交工具输出) ──► in_progress │
│ │ │
│ ├──► completed │
│ ├──► failed │
│ ├──► cancelled │
│ └──► expired │
│ │
└─────────────────────────────────────────────────────────────┘
取消运行 #
python
run = client.beta.threads.runs.cancel(
thread_id=thread.id,
run_id=run.id
)
完整应用示例 #
智能客服助手 #
python
from openai import OpenAI
import time
import json
client = OpenAI()
class CustomerServiceAssistant:
def __init__(self):
self.assistant = client.beta.assistants.create(
name="智能客服",
instructions="""你是一个专业的客服助手。
职责：
1. 回答产品相关问题
2. 处理订单查询
3. 解决用户投诉
4. 提供售后支持
行为准则：
- 礼貌、耐心、专业
- 准确回答问题
- 无法解决时引导转人工
""",
model="gpt-4o-mini",
tools=[
{
"type": "function",
"function": {
"name": "query_order",
"description": "查询订单状态",
"parameters": {
"type": "object",
"properties": {
"order_id": {
"type": "string",
"description": "订单号"
}
},
"required": ["order_id"]
}
}
},
{
"type": "function",
"function": {
"name": "check_inventory",
"description": "检查商品库存",
"parameters": {
"type": "object",
"properties": {
"product_id": {
"type": "string",
"description": "商品ID"
}
},
"required": ["product_id"]
}
}
}
]
)
self.threads = {}
def _handle_tool_call(self, tool_call):
if tool_call.function.name == "query_order":
args = json.loads(tool_call.function.arguments)
return self._query_order(args["order_id"])
elif tool_call.function.name == "check_inventory":
args = json.loads(tool_call.function.arguments)
return self._check_inventory(args["product_id"])
return {"error": "Unknown function"}
def _query_order(self, order_id: str) -> dict:
orders = {
"ORD001": {"status": "已发货", "tracking": "SF123456"},
"ORD002": {"status": "处理中", "tracking": None}
}
return orders.get(order_id, {"status": "未找到订单"})
def _check_inventory(self, product_id: str) -> dict:
inventory = {
"PROD001": {"stock": 100, "price": 99.9},
"PROD002": {"stock": 0, "price": 199.9}
}
return inventory.get(product_id, {"stock": 0, "price": 0})
def chat(self, user_id: str, message: str) -> str:
if user_id not in self.threads:
self.threads[user_id] = client.beta.threads.create()
thread = self.threads[user_id]
client.beta.threads.messages.create(
thread_id=thread.id,
role="user",
content=message
)
run = client.beta.threads.runs.create(
thread_id=thread.id,
assistant_id=self.assistant.id
)
while True:
run = client.beta.threads.runs.retrieve(
thread_id=thread.id,
run_id=run.id
)
if run.status == "completed":
break
elif run.status == "requires_action":
tool_calls = run.required_action.submit_tool_outputs.tool_calls
tool_outputs = []
for tool_call in tool_calls:
result = self._handle_tool_call(tool_call)
tool_outputs.append({
"tool_call_id": tool_call.id,
"output": json.dumps(result, ensure_ascii=False)
})
client.beta.threads.runs.submit_tool_outputs(
thread_id=thread.id,
run_id=run.id,
tool_outputs=tool_outputs
)
elif run.status in ["failed", "cancelled", "expired"]:
return f"服务暂时不可用: {run.status}"
time.sleep(0.5)
messages = client.beta.threads.messages.list(
thread_id=thread.id,
limit=1,
order="desc"
)
return messages.data[0].content[0].text.value
assistant = CustomerServiceAssistant()
print(assistant.chat("user1", "我的订单 ORD001 到哪里了？"))
print(assistant.chat("user1", "商品 PROD001 有库存吗？"))
数据分析助手 #
python
class DataAnalysisAssistant:
def __init__(self):
self.client = OpenAI()
self.assistant = self.client.beta.assistants.create(
name="数据分析助手",
instructions="""你是一个数据分析专家。
使用 Python 进行数据分析：
1. 数据清洗和预处理
2. 统计分析
3. 数据可视化
4. 生成报告
请使用代码解释器执行分析任务。
""",
model="gpt-4o-mini",
tools=[{"type": "code_interpreter"}]
)
self.thread = self.client.beta.threads.create()
def analyze(self, data_description: str, question: str) -> str:
self.client.beta.threads.messages.create(
thread_id=self.thread.id,
role="user",
content=f"数据：{data_description}\n\n问题：{question}"
)
run = self.client.beta.threads.runs.create(
thread_id=self.thread.id,
assistant_id=self.assistant.id
)
while True:
run = self.client.beta.threads.runs.retrieve(
thread_id=self.thread.id,
run_id=run.id
)
if run.status == "completed":
break
elif run.status in ["failed", "cancelled", "expired"]:
return f"分析失败: {run.status}"
time.sleep(0.5)
messages = self.client.beta.threads.messages.list(
thread_id=self.thread.id,
limit=1,
order="desc"
)
return messages.data[0].content[0].text.value
analyzer = DataAnalysisAssistant()
result = analyzer.analyze(
"销售数据：1月100万，2月120万，3月150万，4月130万",
"分析销售趋势并预测下个月"
)
print(result)
最佳实践 #
1. 助手指令设计 #
python
assistant = client.beta.assistants.create(
name="专业助手",
instructions="""你是一个专业的助手。
## 角色定义
[描述助手的角色和定位]
## 核心能力
[列出助手的主要能力]
## 行为准则
[定义助手的行为规范]
## 回复格式
[指定回复的格式要求]
## 限制说明
[说明助手不能做什么]
""",
model="gpt-4o-mini"
)
2. 错误处理 #
python
def safe_run(thread_id: str, assistant_id: str, timeout: int = 300):
run = client.beta.threads.runs.create(
thread_id=thread_id,
assistant_id=assistant_id
)
start_time = time.time()
while True:
run = client.beta.threads.runs.retrieve(
thread_id=thread_id,
run_id=run.id
)
if run.status == "completed":
return run
elif run.status == "failed":
raise Exception(f"运行失败: {run.last_error}")
elif run.status == "cancelled":
raise Exception("运行被取消")
elif run.status == "expired":
raise Exception("运行超时")
if time.time() - start_time > timeout:
client.beta.threads.runs.cancel(
thread_id=thread_id,
run_id=run.id
)
raise Exception("运行超时")
time.sleep(1)
3. 资源管理 #
python
def cleanup_assistant(assistant_id: str):
"""删除助手"""
client.beta.assistants.delete(assistant_id)
def cleanup_thread(thread_id: str):
"""删除线程"""
client.beta.threads.delete(thread_id)
def cleanup_file(file_id: str):
"""删除文件"""
client.files.delete(file_id)
下一步 #
现在你已经掌握了 Assistants API 的使用方法，接下来学习 最佳实践，了解如何在生产环境中高效、安全地使用 OpenAI API！
最后更新：2026-03-29