OpenAI Assistants API 完整指南

Assistants API 解决了什么问题？

用普通 Chat Completions API 构建对话 AI 时，开发者需要自己：

维护对话历史（每次把所有消息发给 API）

管理上下文长度（超出就要截断）

实现文件上传和解析

处理代码执行的沙箱环境

Assistants API 把这些全包了。

---

核心概念

Assistant（助手）
  ├── 配置项：系统指令、工具权限、模型选择
  └── 可以跨 Thread 复用
Thread（线程）= 一次完整的对话
  ├── 自动管理历史消息
  └── 超出上下文长度时自动截断旧消息
Message（消息）= Thread 中的一条消息
  └── 支持文本、图片、文件
Run（运行）= 让 Assistant 处理 Thread 的一次执行
  └── 状态：queued → in_progress → completed

---

快速开始

python
from openai import OpenAI
client = OpenAI()
第一步：创建 Assistant（只需创建一次，保存 ID 复用）
assistant = client.beta.assistants.create(
    name="个人助手",
    instructions="""你是一个专业的个人助手。
    - 记住用户提到的个人信息和偏好
    - 回答简洁，避免废话
    - 遇到需要计算的问题，使用 code_interpreter 工具
    """,
    tools=[{"type": "code_interpreter"}, {"type": "file_search"}],
    model="gpt-4o",
)
print(f"Assistant ID: {assistant.id}")  # 保存这个ID！

---

完整对话流程

python
每个用户一个 Thread（保存 thread_id 到数据库）
thread = client.beta.threads.create()
print(f"Thread ID: {thread.id}")
用户发消息
client.beta.threads.messages.create(
    thread_id=thread.id,
    role="user",
    content="我叫张三，在北京工作，帮我分析一下北京AI创业公司的趋势"
)
运行 Assistant
run = client.beta.threads.runs.create_and_poll(
    thread_id=thread.id,
    assistant_id=assistant.id,  # 你保存的 Assistant ID
)
获取回复
if run.status == "completed":
    messages = client.beta.threads.messages.list(thread_id=thread.id)
    print(messages.data[0].content[0].text.value)

---

文件分析能力

python
上传文件
with open("financial_report.pdf", "rb") as f:
    file = client.files.create(file=f, purpose="assistants")
创建带文件的消息
client.beta.threads.messages.create(
    thread_id=thread.id,
    role="user",
    content="帮我总结这份财报的关键数字",
    attachments=[
        {"file_id": file.id, "tools": [{"type": "file_search"}]}
    ]
)

---

代码执行（Code Interpreter）

Assistants API 内置 Python 沙箱，AI 可以直接运行代码：

python
client.beta.threads.messages.create(
    thread_id=thread.id,
    role="user",
    content="帮我分析一下这份 CSV 的数据分布，并生成图表"
)
AI 会自动写 Python 代码、运行、返回图表

---

持久化：把对话存起来

对于生产应用，需要把 Thread ID 和用户绑定：

python
import sqlite3
def get_or_create_thread(user_id: str) -> str:
    """为每个用户维护一个持久 Thread"""
    conn = sqlite3.connect("threads.db")
    cursor = conn.cursor()
    
    cursor.execute("SELECT thread_id FROM users WHERE user_id = ?", (user_id,))
    result = cursor.fetchone()
    
    if result:
        return result[0]
    
    # 新用户，创建 Thread
    thread = client.beta.threads.create()
    cursor.execute(
        "INSERT INTO users (user_id, thread_id) VALUES (?, ?)",
        (user_id, thread.id)
    )
    conn.commit()
    return thread.id

---

费用说明

费用项计费方式备注

模型推理按 token与 Chat API 相同 Code Interpreter$0.03/次 Session每次 Run 独立计算 File Search$0.10/GB/天向量存储费用

建议：测试阶段关闭 File Search 节省费用，仅在需要时开启。

---

什么时候用 Assistants API vs Chat API？

场景推荐 API

简单的单次问答Chat Completions 需要多轮对话记忆Assistants API 需要分析上传的文件Assistants API 需要执行代码Assistants API 要求极低延迟Chat Completions 成本极度敏感Chat Completions

---

常见问题

Q：Thread 的消息会一直保存吗？ A：默认保存 60 天，之后自动删除。

Q：可以给 Assistant 设置记忆上限吗？ A：不能直接设置，但可以在系统提示词中要求 AI「总结并丢弃超过 N 轮的对话」。

Q：Assistants API 支持流式输出吗？ A：支持，使用 stream=True 参数。

👉 查看 MCP 工具生态 | 探索 AI Agent 用例