Assistant 架构与核心机制
Assistant（助手） 是一个特定用途的 AI 实体，它包含名称、描述、默认模型、指令、可用工具和关联文件等。可以把 Assistant 看作是一个"预配置"的聊天机器人，例如知识库问答助手或技术顾问。一个 Assistant 在创建时会设置默认的指令（instructions）和模型等属性，这些也可以在运行（Run）时覆写。例如，创建一个数学辅导助手时，指令可能是"你是数学老师，简洁回答问题"。
核心组件
1. 工具（Tools）
Assistant 可以启用不同类型的工具来扩展能力。目前支持的三种工具是：
- 函数调用（Function Calling）：允许模型识别并调用预定义的外部函数
- 文件检索（File Search）：自动对上传的文档进行切片、索引和向量化检索
- 代码解释（Code Interpreter）：让助手执行代码来解决问题
工具需要在创建 Assistant 时声明，例如指定 tools=[{"type":"function"}, {"type":"file_search"}] 并通过 tool_resources 关联具体的文件或向量库。
2. 文件（Files）
文件是帮助 Assistant 回答问题的知识来源。开发者可以通过 API 上传文件（文本、PDF 等）并指定 purpose="assistants"。上传后，文件会被分片并存入向量检索库，供 File Search 工具调用。
文件限制：
- 每个向量库最多支持 10,000 个文件
- 每个助手/对话线程限制一个向量库
- 在对话中，也可以将文件作为消息的附件发送给助手进行参考
3. Thread（线程）
线程表示一次会话的上下文。每个线程包含一系列有序消息（Messages），包括用户消息和助手回复。线程自动保存完整对话历史，并在达到模型上下文长度时自动截断（截取最新消息），无须开发者手动管理所有历史。
线程优势：
- 多轮对话更容易实现
- 应用只需创建一个线程，后续发送新消息时使用相同线程 ID
- OpenAI 会在后台拼接最新的上下文并自动截断过长部分
4. Message（消息）
消息是线程中的一条记录，包含：
- 发信者角色（"user" 或 "assistant"）
- 文本内容
- 可选附件（文件）
例如，用户发送的提问是一条角色为 "user" 的消息，助手回复的一条消息则角色为 "assistant"。消息会被存储在指定的线程下，无需每次请求时都传输整段历史文本。
5. Run（运行）
Run 表示一次助手执行过程。当想让某个 Assistant 对一个已有线程产生回复时，需要创建一个 Run。
Run 执行流程：
- 创建 Run 时，需要指定
thread_id和assistant_id - 可选地覆盖当次运行的指令、模型或工具配置
- Run 会异步执行：首次创建时状态为
queued - 在后台生成回答或调用工具，最终状态变为
completed
重要特性：
- 与传统的 Chat Completion 不同，一次 Run 可能涉及多个步骤
- 助手不仅可以生成文本回复，还可能调用一个或多个工具
- 甚至向同一线程添加多条消息
- Run 的最终结果可能包括一系列消息和工具调用的输入输出
- 开发者可以通过轮询接口（如
threads.runs.retrieve）获取 Run 状态并查看回复内容
核心概念总结表
| 概念 | 说明 |
|---|---|
| Assistant | 特定用途的 AI 助手实体，封装了基础模型、指令、工具和知识文档 |
| Thread | 一次对话会话，存储消息历史并自动截断过长内容，简化上下文管理 |
| Message | 对话中的一条记录，具有角色（用户/助手）和内容，可携带文件附件 |
| Run | 将某个 Assistant 应用于指定 Thread 的执行过程。创建 Run 后，Assistant 会根据当前线程的消息生成回复，可能使用工具并产生多条输出 |
| 工具（Tool） | 助手可选的附加能力。目前支持 function（函数调用）、file_search（检索上传文档）和 code_interpreter（代码执行）等工具 |
| 文件（File） | 上传的知识文档，可与多条消息和助手关联。通过 File Search 工具对其内容进行向量检索，以增强回答能力 |
使用 Assistants API 创建与管理助手
OpenAI 提供了 Assistants API（助理接口），简化了基于对话历史与工具调用的应用开发。下面以 Python SDK 为例介绍典型步骤：
1 创建 Assistant
调用 client.assistants.create()（旧版为 client.beta.assistants.create）定义一个新助手。需要指定助手名称、指令和默认模型。例如：
from openai import OpenAI
client = OpenAI(api_key="API_KEY")
assistant = client.beta.assistants.create(
name="我的AI助手",
instructions="你是一个知识库查询助手，只使用提供的文档回答问题。",
model="gpt-4o",
tools=[{"type": "file_search"}] # 启用文件检索工具
)以上代码创建了一个新的 Assistant，并返回其 ID。创建成功后，可在控制台或 API 响应中查看 assistant.id。
2 创建 Thread（对话线程）
当用户启动新会话时，先用 client.threads.create() 新建一个线程。线程代表一次独立对话，会话结束前一直保持存在。
thread = client.beta.threads.create()新线程创建后会返回一个 thread.id，代表该会话的唯一标识。一般而言，每个用户启动对话时都创建一个新的线程，以便隔离不同用户或会话的上下文。
3 发送用户消息
在获得线程 ID 后，将用户的输入作为一条消息添加到线程里。使用 client.threads.messages.create()，指定线程 ID、角色 "user" 和消息内容。
user_msg = client.beta.threads.messages.create(
thread_id=thread.id,
role="user",
content="请问今天天气如何？"
)如上，将用户提问加入线程。如果需要，也可以在 file_ids=[file.id] 参数中包含先前上传的文件 ID，让助手能够访问这些文件。
4 启动 Run（执行助手）
有了用户消息，就可以让 Assistant 响应了。调用 client.threads.runs.create()，需要传入 thread_id 和 assistant_id，也可附加当次运行要用的新指令或覆盖参数。这个请求是异步的，会立即返回 Run 元数据。
run = client.beta.threads.runs.create(
thread_id=thread.id,
assistant_id=assistant.id
)创建 Run 后，助手开始处理该线程中的消息，包括生成回复或调用工具。运行状态会是 "queued"，之后逐步变为 "in_progress"，直到 "completed"。
5 轮询并获取回复
由于 Run 异步执行，我们需要循环查询其状态，直到完成。完成后，助手的回复会以消息的形式追加到该线程中，可通过 client.threads.messages.list(thread_id, order="asc") 或 runs.retrieve 获得。
# 等待 Run 完成
import time
while run.status in ["queued", "in_progress"]:
time.sleep(0.5)
run = client.beta.threads.runs.retrieve(
thread_id=thread.id,
run_id=run.id
)
# 获取回复消息
messages = client.beta.threads.messages.list(
thread_id=thread.id,
order="asc"
)
response = messages.data[-1].content[0].text.value
print("Assistant:", response)6 （可选）添加更多消息与工具调用
助手完成一次回复后，如果需要继续对话，可再向同一线程添加用户消息并重复上述 Run 流程。若在对话中使用函数调用工具，流程会稍有不同（参见示例代码）。需要注意，一个线程在同一时间只能有一个 Run 在执行。若需要让同一个助手进行多轮复杂互动，它会持续不断地为同一线程创建新的 Run。
以上步骤展示了使用 Assistants API 的基本流程，从创建助手到进行对话。官方文档和社区示例中也使用了类似流程。例如，在官方教程中创建数学辅导助手的代码片段与上述类似；再依次创建线程、消息和运行。
Python 完整示例：函数调用、文件上传与上下文处理
import time, json
from openai import OpenAI
client = OpenAI(api_key="YOUR_API_KEY")
# --- 步骤1：上传文件 (作为知识库) ---
file_resp = client.files.create(
file=open("knowledge.txt", "rb"),
purpose="assistants"
)
file_id = file_resp.id
# --- 步骤2：创建向量库（仅V2，需要将文件打入向量存储） ---
vector_store = client.beta.vector_stores.create(
name="KnowledgeBase",
file_ids=[file_id]
)
# 关联到 file_search 工具
tool_resources = {"file_search": {"vector_store_ids": [vector_store.id]}}
# --- 步骤3：定义自定义函数(工具) ---
def send_email(quote, recipient="user@example.com"):
# 模拟发送邮件的操作
print(f"【模拟发送邮件】主题: 新名言, 内容: {quote}, 收件人: {recipient}")
return f"邮件已发送给 {recipient}"
# 将函数描述转换为 JSON 模式
function_tool = {
"type": "function",
"function": {
"name": "send_email",
"description": "把一条名言通过邮件发送给用户",
"parameters": {
"type": "object",
"properties": {
"quote": {
"type": "string",
"description": "生成的有趣名言"
},
"recipient": {
"type": "string",
"description": "用户邮箱"
}
},
"required": ["quote"]
}
}
}
# --- 步骤4：创建 Assistant，启用 file_search 和 function 工具 ---
assistant = client.beta.assistants.create(
name="FunQuotes助手",
instructions=(
"你是一个聪明的名言生成助手。你可以讲笑话并生成有趣名言；"
"如果用户要求通过邮件发送名言，请使用 send_email 工具。"
),
model="gpt-4o",
tools=[{"type": "file_search"}, function_tool],
tool_resources=tool_resources
)
assistant_id = assistant.id
# --- 步骤5：启动对话线程 ---
thread = client.beta.threads.create()
# --- 步骤6：模拟用户提问 ---
# 示例：要求生成名言并通过邮件发送
user_question = "请告诉我一个有趣的名言并发邮件给我。"
client.beta.threads.messages.create(
thread_id=thread.id,
role="user",
content=user_question
)
# --- 步骤7：执行 Run ---
run = client.beta.threads.runs.create(
thread_id=thread.id,
assistant_id=assistant_id
)
# 等待助手完成
while run.status in ("queued", "in_progress"):
time.sleep(0.5)
run = client.beta.threads.runs.retrieve(
thread_id=thread.id,
run_id=run.id
)
# 检查助手是否请求调用函数
if run.status == "requires_action" and run.required_action:
tool_call = run.required_action.submit_tool_outputs.tool_calls[0]
func_name = tool_call.function.name
args = json.loads(tool_call.function.arguments)
print("助手请求调用函数：", func_name, "参数：", args)
# 执行相应的 Python 函数
result = send_email(**args) # 这里调用 send_email 函数
# 将函数执行结果反馈给助手
run = client.beta.threads.runs.submit_tool_outputs(
thread_id=thread.id,
run_id=run.id,
tool_outputs=[{
"tool_call_id": tool_call.id,
"output": result
}]
)
# 继续等待 Run 完成
while run.status in ("queued", "in_progress"):
time.sleep(0.5)
run = client.beta.threads.runs.retrieve(
thread_id=thread.id,
run_id=run.id
)
# --- 步骤8：获取最终回复 ---
messages = client.beta.threads.messages.list(
thread_id=thread.id,
order="asc"
)
for msg in messages.data:
if msg.role == "assistant":
print("助手回复：", msg.content[0].text.value)代码说明
- 文件上传与向量存储：程序用
client.files.create()上传知识文档，并创建一个向量存储vector_stores.create()以启用 file_search 功能 - 自定义函数定义：定义了一个 Python 函数
send_email，并以 JSON 形式告诉助手该函数的名称、描述及参数 - 助手创建：调用
assistants.create创建助手，启用了文件检索和自定义函数工具 - 对话执行：创建一个对话线程，将用户提问加入该线程，然后用
threads.runs.create启动 Run - 函数调用处理：在等待 Run 完成后，检查
run.status是否为 "requires_action"，这表示助手决定需要调用我们定义的函数。如果是，就从run.required_action中提取函数名和参数，调用本地的send_email函数，再通过runs.submit_tool_outputs将函数输出传回给助手 - 获取最终回复：最后再次等待 Run 完成，并从线程中读取助手的最终回复
这段完整示例展示了如何结合上下文保持、文件检索和函数调用等高级功能来实现智能助手。