OpenAI Assistants API完整教程2026

为什么选择Assistants API？

功能Chat APIAssistants API

对话记忆需自己管理自动（Threads） 代码执行不支持Code Interpreter 文件搜索不支持File Search（RAG） 异步处理不支持Runs状态机

核心概念

Assistant：定义AI角色，一次创建多次复用

Thread：一次对话，保存所有消息历史

Run：执行一次Assistant调用（queued → in_progress → completed）

完整示例

python
from openai import OpenAI
client = OpenAI()
创建助手
assistant = client.beta.assistants.create(
    name="数据分析助手",
    instructions="你是专业的数据分析师。",
    model="gpt-4o",
    tools=[{"type": "code_interpreter"}, {"type": "file_search"}]
)
创建线程和发送消息
thread = client.beta.threads.create()
client.beta.threads.messages.create(
    thread_id=thread.id, role="user",
    content="分析销售数据，找出增长最快的类别"
)
运行并获取结果
run = client.beta.threads.runs.create_and_poll(
    thread_id=thread.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)

文件搜索（内置RAG）

python
file = client.files.create(file=open("docs.pdf", "rb"), purpose="assistants")
vs = client.beta.vector_stores.create(name="知识库")
client.beta.vector_stores.files.create(vector_store_id=vs.id, file_id=file.id)
client.beta.assistants.update(
    assistant_id=assistant.id,
    tool_resources={"file_search": {"vector_store_ids": [vs.id]}}
)

流式输出

python
with client.beta.threads.runs.stream(thread_id=thread.id, assistant_id=assistant.id) as stream:
    for text in stream.text_deltas:
        print(text, end="", flush=True)

生产环境注意事项

设置 max_completion_tokens 控制费用

Run执行需要几十秒，需要polling或webhook

定期清理长期不活跃的Thread