OpenAI Assistants API：状态化助手与工具集成

FreeGuideOnline 最新 2026-06-14

什么是OpenAI Assistants API

Assistants API 是 OpenAI 提供的一套用于构建有状态、可持久化AI 助手的接口。与无状态的 Chat Completions API 不同，Assistants API 会替你管理对话历史、线程状态以及工具调用的上下文。你可以把它理解为：你定义一位“助手”（Assistant），设定好指令、模型和工具；接下来每个用户会话都运行在一条“线程”（Thread）上；线程中不断追加用户消息、触发助手运行（Run），最终读取助手的回复。整个过程中，OpenAI 帮你自动处理上下文窗口管理、工具调用循环和文件引用。

借助 Assistants API，开发者能够快速为产品加入以下能力：

  • 长期记忆的对话体验（线程自动保留完整历史）
  • 代码解释器（Code Interpreter）让助手能写代码、运行代码并返回结果
  • 文件搜索（File Search）基于上传的文档进行 RAG 式的问答
  • 自定义函数调用（Function Calling）连接外部业务系统

核心概念速览

在使用 API 之前，必须先理清这几个对象之间的关系：

  • Assistant（助手）：定义了 AI 的行为。包括系统指令（instructions）、使用的模型（如 gpt-4-turbo）、启用的工具（tools）以及可引用的文件。
  • Thread（线程）：代表一个与用户的对话会话。线程内保存了用户消息和助手消息，不会因 API 调用结束而丢失。
  • Message（消息）：线程中的一条记录，可以是用户发来的，也可以是助手生成的。消息可包含文本、图片文件等。
  • Run（运行）：在某个线程上执行一次助手调用。你需要创建一个 Run，让助手读取线程中的新消息并进行响应。Run 有状态（排队中、进行中、需要操作、已完成等），可以异步轮询或等待。

简单来说，一次典型的调用流程是：

  1. 创建或选择一个 Assistant
  2. 创建 Thread，并向其中添加一条用户 Message
  3. 在 Thread 上创建并启动一个 Run
  4. 轮询 Run 状态，直到 completed，或处理中间状态（如 requires_action）
  5. 提取助手生成的最新 Message 返回给用户

快速上手指南：Python 环境

以下使用 Python 和 OpenAI 官方库。安装依赖：

pip install openai

设置 API 密钥：

import openai
openai.api_key = "your-api-key"
# 或使用 client = openai.OpenAI(api_key="...")

第一步：创建助手

我们可以创建一个具备代码解释器和文件搜索能力的助手，并设定指令。

from openai import OpenAI
client = OpenAI()

assistant = client.beta.assistants.create(
    name="数据分析助手",
    instructions="你是一位资深数据分析师。回答问题时优先使用文件搜索工具从上传的文档中寻找证据；如果需要进行计算或绘图，使用代码解释器。回答应专业、简洁。",
    model="gpt-4-turbo",
    tools=[
        {"type": "code_interpreter"},
        {"type": "file_search"}
    ]
)
print(assistant.id)  # 保存此 ID，后续复用

第二步：开启线程并发送用户消息

thread = client.beta.threads.create()
print(thread.id)  # 保存线程 ID

message = client.beta.threads.messages.create(
    thread_id=thread.id,
    role="user",
    content="根据 sales_report.pdf 分析 Q2 的销售趋势并画折线图"
)

提示：要使用文件搜索，需要先将 PDF 文件上传并关联到助手或消息。此时我们可先上传文件，然后在创建消息时传入附件（具体见下一节工具集成）。

第三步：创建并启动 Run

run = client.beta.threads.runs.create(
    thread_id=thread.id,
    assistant_id=assistant.id
)

第四步：轮询 Run 状态并获取回复

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 in ["failed", "cancelled", "expired"]:
        print(f"Run 异常终止，状态：{run.status}")
        break
    time.sleep(1)

# 获取最新助手消息
messages = client.beta.threads.messages.list(thread_id=thread.id, order="desc", limit=1)
if messages.data:
    # 消息内容可能是文本，也可能是代码解释器的图片输出等
    for content_block in messages.data[0].content:
        if content_block.type == "text":
            print(content_block.text.value)

助手会按指令调用工具，最终返回分析文本和图片（图片通过 file_ids 返回，可下载展示）。

工具集成详解

Assistants API 当前支持三种内置工具和自定义函数调用。工具在创建 Assistant 时通过 tools 参数配置。

代码解释器（Code Interpreter）

激活后，助手可以编写并执行 Python 代码，并使用生成的输出。典型场景包括：

  • 数学计算、复杂公式求解
  • 数据分析、绘制图表（matplotlib）
  • 文件格式转换（如 CSV 转 JSON）

助手会自动在沙箱环境中运行代码，并将生成的文件（如图片、CSV）附加到回复中。你无需自行管理执行环境。

配置示例：

assistant = client.beta.assistants.create(
    model="gpt-4-turbo",
    tools=[{"type": "code_interpreter"}]
)

在 Run 进行中，如果触发了代码解释器，Run 状态会变为 in_progress，内部步骤（steps）中会包含 tool_calls 详情。最终返回的消息可能包含 image_file 类型的 content block，你可通过 file_id 下载图片。

文件搜索实现了基于向量的语义搜索，让助手能够引用你上传的文档内容。它类似于“内置的 RAG”，无需你自行构建检索管道。

使用步骤：

  1. 上传文件并获取 file.id

    file = client.files.create(
    file=open("knowledge_base.pdf", "rb"),
    purpose="assistants"
)

  2. 将文件关联到助手（或者也可以只给某个线程/消息添加附件，但通常推荐关联到助手以便全局使用）

    assistant = client.beta.assistants.update(
    assistant_id=assistant.id,
    tool_resources={"file_search": {"vector_store_ids": [vector_store_id]}}
)

    注意：文件搜索需要先创建一个向量存储（Vector Store），将文件加入其中。便捷方式：

    vector_store = client.beta.vector_stores.create(name="企业知识库")
file_batch = client.beta.vector_stores.file_batches.create(
    vector_store_id=vector_store.id,
    file_ids=[file.id]
)

    然后将 vector_store.id 填入助手的 tool_resources 配置。

  3. 在消息中也可以添加临时文件：当用户上传某个需要立即检索的文档时，可在创建消息时传入 attachments 参数，指定 file_search 工具。

message = client.beta.threads.messages.create(
    thread_id=thread.id,
    role="user",
    content="总结这份文档的要点",
    attachments=[{"file_id": file.id, "tools": [{"type": "file_search"}]}]
)

当 Run 执行时，助手会自动从关联的向量存储或消息附件中检索相关片段，并基于此生成回答。

函数调用（Function Calling）

你可以定义自定义函数，让助手在合适的时机调用这些函数获取实时数据或执行业务操作。这实现了助手与外部 API 的深度集成。

定义函数示例：查询天气

assistant = client.beta.assistants.create(
    model="gpt-4-turbo",
    tools=[{
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的当前天气",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {"type": "string", "description": "城市，如 Beijing"}
                },
                "required": ["location"]
            }
        }
    }]
)

处理函数调用：当 Run 状态变为 requires_action 时，你需要提取 required_action 中的函数调用信息，执行函数后将输出提交回去。

if run.status == "requires_action":
    tool_calls = run.required_action.submit_tool_outputs.tool_calls
    tool_outputs = []
    for call in tool_calls:
        if call.function.name == "get_weather":
            import json
            args = json.loads(call.function.arguments)
            # 模拟调用你的内部天气服务
            weather_result = {"temperature": "22°C", "condition": "晴"}  # 实际应调用真实API
            tool_outputs.append({
                "tool_call_id": call.id,
                "output": json.dumps(weather_result)
            })
    # 提交函数输出并让 Run 继续
    run = client.beta.threads.runs.submit_tool_outputs(
        thread_id=thread.id,
        run_id=run.id,
        tool_outputs=tool_outputs
    )

之后 Run 会继续执行，并基于函数输出生成最终回复。

管理状态与线程持久化

Assistant API 的最大优势是自动维护对话状态。每个 Thread 独立保存全部消息历史，你可以随时暂停、恢复会话，完全不用担心上下文遗忘。这非常适合构建长时间运行、多轮交互的应用。

线程操作示例：

  • 列出所有消息：client.beta.threads.messages.list(thread_id=thread.id)
  • 检索单个消息：client.beta.threads.messages.retrieve(thread_id=..., message_id=...)
  • 删除线程（结束会话）：client.beta.threads.delete(thread_id=thread.id)

最佳实践：用户第一次对话时创建 Thread，将 thread.id 保存在你的后端关联到该用户；后续用户所有消息都追加到同一 Thread 下。这样助手就会一直记住上下文。

如果你需要“遗忘”，可以用 client.beta.threads.messages.delete() 删除特定消息，或者直接创建新 Thread。

Run 生命周期与错误处理

Run 的状态转换主要有：

  • queued：等待资源
  • in_progress：助手正在处理（含调用工具）
  • requires_action：需要你提交函数输出
  • completed：成功完成
  • failed / cancelled / expired：运行失败、被取消或超时

建议使用轮询或 Webhook （需企业版支持） 来监听状态。生产中可加入超时和重试逻辑。另外，可以利用 stream 参数创建流式 Run，获得实时事件，体验更佳（类似 Chat Completions 的流式）。

run = client.beta.threads.runs.create(
    thread_id=thread.id,
    assistant_id=assistant.id,
    stream=True
)
for event in run:
    # 处理 event 对象，例如 event.event == "thread.message.delta"
    pass

实际场景：构建客户支持助手

假设你要构建一个能够查询订单、退款和知识库的助手。组合技术如下：

  1. 助手定义：指令为“你是客服助手，礼貌专业。”启用 file_search（知识库）和两个函数：lookup_order、refund_order。
  2. 知识库：将产品手册、FAQ 等上传至向量存储。
  3. 函数实现：在 requires_action 阶段，调用内部订单系统 API 并返回 JSON 结果。
  4. 线程持久化：每个用户在其线程内连续提问，助手能关联之前的订单查询结果。

关键代码结构与前述一致，不再重复。

成本与优化建议

  • 按 tokens 计费：模型推理、代码解释器会话、文件搜索的存储都产生费用。具体参考 OpenAI 定价页。
  • 优化：
    • 删掉不再使用的 Thread 和 Vector Store 以减少存储费。
    • 利用 truncation_strategy 参数控制对话历史的截断方式，避免上下文过长。
    • 对于高频相同知识请求，缓存检索结果或使用助手的 temperature 参数调低随机性。

总结

OpenAI Assistants API 将过去需要自行拼接的状态管理、工具调用循环和多模态文件处理，抽象为简单的对象模型。初学者只需掌握 Assistant、Thread、Run 三者的关系，即可快速构建出强大的 AI 应用。学会核心流程，再逐步深入各工具的配置细节，你就能为产品赋予智能对话、数据分析、文档问答以及复杂的函数交互能力。

现在就去创建你的第一个助手并实验各种工具组合吧。