如果你的生产服务仍在创建 Assistant、Thread 或 Run,应把 2026 年 8 月 26 日(2026-08-26)当成完成迁移并验证上线的最后期限,而不是开始改代码的日期。OpenAI 已在弃用公告中确定:Assistants API 将在这一天停用,官方替代是 Responses API 与 Conversations API。
本文只处理官方 OpenAI Assistants API → Responses/Conversations 的生产迁移;Chat Completions → Responses、Azure OpenAI,以及第三方 OpenAI-compatible provider 的迁移合同不在本文范围内,不能直接套用以下对象、工具或数据保留结论。
安全路线不是批量替换 endpoint:
- 把 Assistant 的 instructions、model 和 tools 迁回版本控制下的应用配置;
- 根据持久性与数据要求,选择 Conversation、
previous_response_id或应用自管 Items; - 用 Responses 处理输入与输出,并由应用完成自定义工具循环;
- 复用 vector store 前先验证 File Search 的结果、引用与延迟;
- 旧链路保持可回滚,只有影子测试与灰度指标通过后才切生产。
还有一个容易造成二次返工的日期:OpenAI 当前的 Assistants migration guide仍展示 Assistant → Prompt,但官方在 2026 年 6 月 3 日宣布 reusable prompt objects 弃用,v1/prompts 和这些对象计划于 2026 年 11 月 30 日(2026-11-30)停用。新的长期实现不应把 Prompt object 当终点;应把 prompt 内容放在应用代码或自己的配置系统中,再通过 instructions/input 发送给 Responses。
先建立迁移台账,不要先改 URL
对一个真实 Assistant 逐项盘点。没有 owner 的项目,迁移后最容易变成隐性故障。
| 旧资产或行为 | 新 owner | 必须验证 |
|---|---|---|
| Assistant 的 instructions、model、tools | 应用内版本化配置 | 同一测试输入是否仍满足行为边界 |
| Thread | Conversation,或明确选择的其他状态策略 | 跨 turn、跨设备、跨进程是否保持正确上下文 |
| Message | Responses 的 input/output Items | 不能只保存最终文字而丢掉工具或 reasoning Items |
| Run | 一次 Response;真正长任务才考虑 background mode | 完成、失败、超时、取消与重试语义 |
| Run step | typed Items | message、function call、tool output 必须按 type 分流 |
submit_tool_outputs | 下一次 Response 的 function_call_output | call_id 与原调用严格配对 |
| Retrieval / File Search | Responses hosted file_search + vector store | 文件就绪、召回、引用、过滤、延迟和成本 |
| Run polling | 同步响应、streaming,或显式 background:true | 不要把所有旧轮询机械改成后台任务 |
这张表的作用不是记录“代码改过了”,而是记录哪个系统现在拥有哪一种状态,以及怎样证明没有退化。
迁移前至少保存:
- 每个 Assistant 的 instructions、工具 schema、模型配置和关联 vector store;
- 10—30 个代表真实任务的 golden cases,包括失败分支;
- 当前质量、首 token 延迟、总延迟、输入/输出 token、工具成功率和错误率;
- Thread 与应用 session/user 的映射;
- 旧链路的 feature flag 和回滚入口;
- 必须继续访问的活跃历史,以及可以只归档、不转换的历史。
如果尚未有可用的 OpenAI Platform 项目密钥和 API Billing,先完成OpenAI API Key、计费与首次调用验收。创建更多 key 不会解决对象模型或工具循环问题。
Thread 迁到哪里:先选状态策略
OpenAI conversation-state 文档提供三种主要方式。它们不是同一功能的不同语法,而是不同的数据与控制合同。
| 选择 | 适合 | 代价与停止规则 |
|---|---|---|
| Conversation | 跨 session、设备或 job 的持久会话;需要保存 message、tool call 与 tool output | Conversation items 不受普通 Response 30 天 TTL 限制;先确认数据保留、删除和组织政策 |
previous_response_id | 短链、多轮原型或简单 continuation | 顶层 instructions 要在新请求中重新发送;历史输入仍按 input tokens 计费,链持续增长时应停止并改策略 |
应用自管 Items,store:false | ZDR、明确的数据最小化、需要自己裁剪上下文 | 必须保存并回放完整 response.output,包括 reasoning 与工具 Items,不能只保存 output_text |
长会话还需要 compaction,而不是无限追加。官方的Compaction 指南提供服务端 threshold 与显式 /responses/compact 路径。触发规则应来自你自己的 context/token 观测;每次 compact 后都要重跑“关键事实保留、待办保留、工具状态可继续”的测试。
一个实用决策顺序:
- 业务是否需要跨设备或跨任务恢复?需要则优先评估 Conversation。
- 组织是否要求 ZDR 或应用自主管理历史?需要则选择
store:false和完整 Item replay。 - 是否只是几轮短交互?可以先用
previous_response_id,但设置 token 与 turn 上限。 - 会话是否可能长期增长?在上线前定义 compaction 和删除策略。
同一场景对照:Assistants before 与 Responses after
下面两段使用同一个迁移合同:输入都是 Where is ORDER-42?,工具都是严格 schema 的 get_order_status(order_id: string),预期都是工具只执行一次并返回 shipped。任一侧出现错误工具名、参数不合法、调用 ID 未原样返回、超过 6 轮或在无证据时编造状态,都判失败。
这些代码按当前 OpenAI 官方接口与 SDK 示例对齐,但本 run 没有使用读者的凭据、project、Assistant ID 或当前 SDK 实际运行,因此不构成 production parity 证据。复制前要在当前 SDK 的 staging 环境执行,并把 SDK 版本、响应 Items、调用 ID、最终输出和失败分支写进迁移记录。
Before:Assistant、Thread、Run 与 requires_action
旧侧假设 OPENAI_ASSISTANT_ID 指向已存在的生产 Assistant,且其 instructions 与工具 schema 已和下方新侧的 INSTRUCTIONS、TOOLS 核对一致;不要为了运行示例再创建新的 Assistant。
pythonimport json import os from openai import OpenAI client = OpenAI() assistant_id = os.environ["OPENAI_ASSISTANT_ID"] def get_order_status(order_id: str) -> dict: demo = {"ORDER-42": {"status": "shipped", "carrier": "Example Express"}} return demo.get(order_id, {"status": "not_found"}) thread = client.beta.threads.create() client.beta.threads.messages.create( thread_id=thread.id, role="user", content="Where is ORDER-42?", ) run = client.beta.threads.runs.create_and_poll( thread_id=thread.id, assistant_id=assistant_id, ) for _ in range(6): if run.status == "requires_action": calls = run.required_action.submit_tool_outputs.tool_calls outputs = [] for call in calls: if call.function.name != "get_order_status": raise RuntimeError(f"unexpected tool: {call.function.name}") args = json.loads(call.function.arguments) outputs.append({ "tool_call_id": call.id, "output": json.dumps(get_order_status(**args)), }) run = client.beta.threads.runs.submit_tool_outputs_and_poll( thread_id=thread.id, run_id=run.id, tool_outputs=outputs, ) else: break if run.status != "completed": raise RuntimeError(f"legacy run failed: {run.status}")
这里由 Assistant 保存配置,Thread 保存状态,Run 推进执行;应用在 requires_action 时执行函数,并用旧 tool_call_id 提交结果。
After:应用配置、Conversation、Response 与 tool loop
pythonimport json import os from openai import OpenAI client = OpenAI() MODEL = os.environ["OPENAI_MODEL"] INSTRUCTIONS = "Use get_order_status for order lookups. Do not invent status." TOOLS = [{ "type": "function", "name": "get_order_status", "description": "Return the current status for one order.", "strict": True, "parameters": { "type": "object", "properties": {"order_id": {"type": "string"}}, "required": ["order_id"], "additionalProperties": False, }, }] def get_order_status(order_id: str) -> dict: demo = {"ORDER-42": {"status": "shipped", "carrier": "Example Express"}} return demo.get(order_id, {"status": "not_found"}) conversation = client.conversations.create( metadata={"app_session": "migration-smoke-001"} ) response = client.responses.create( model=MODEL, conversation=conversation.id, instructions=INSTRUCTIONS, tools=TOOLS, input="Where is ORDER-42?", ) for _ in range(6): calls = [item for item in response.output if item.type == "function_call"] if not calls: print(response.output_text) break outputs = [] for call in calls: if call.name != "get_order_status": raise RuntimeError(f"unexpected tool: {call.name}") args = json.loads(call.arguments) outputs.append({ "type": "function_call_output", "call_id": call.call_id, "output": json.dumps(get_order_status(**args)), }) response = client.responses.create( model=MODEL, conversation=conversation.id, instructions=INSTRUCTIONS, tools=TOOLS, input=outputs, ) else: raise RuntimeError("tool loop exceeded 6 rounds")
新侧把 instructions、model 与 schema 交给应用版本控制,以 Conversation 保存本场景状态,以 Response 和 typed Items 表达执行。应用必须验证参数、执行 custom function,并把原始 call_id 放进 function_call_output。若选择应用自管历史而非 Conversation,还要回放完整 response.output,不能只保存最终文本。
三类工具的执行责任不能混在一起
| 工具类型 | 谁实际执行 | 应用必须负责 | 不能据此推断 |
|---|---|---|---|
| Custom function | 你的应用;Responses 只返回 function_call | 参数验证、授权、超时、幂等、审计、轮数上限,再回传原始 call_id 对应的 function_call_output | 收到 call 不代表业务动作已执行 |
| Hosted / built-in tool | OpenAI 在 Response 内按该工具的当前合同执行,例如本文的 file_search | 验证目标模型与 project 是否支持、配置范围、结果 Items、引用、错误、成本与数据边界 | HTTP 200 不代表检索或业务结果已通过 |
| Remote MCP | Responses API 调用你配置且信任的远程 MCP server,并把结果形成 mcp_call Item | server/OAuth 信任、allowed_tools、approval、发出数据的审查与日志、第三方 retention 和失败处理 | OpenAI 不替你验证第三方 server,也不保证其行为、数据政策或 side effect |
官方 function-calling 流程定义 custom function 的应用回环;MCP and Connectors 指南说明 remote MCP 的 server、approval 与安全合同。MCP 默认 approval 流程可以被配置改变;敏感动作不要未经评审改成自动批准。
File Search:复用 vector store 只是起点
Responses 的 file_search 是服务端 hosted tool,不使用上面的应用函数循环。官方File Search 指南说明,可以指定 vector store,并通过 include=["file_search_call.results"]取回检索结果;最终 message 还可能带 file citations。
下面的 TypeScript smoke test 从环境读取模型与 vector store:
typescriptimport OpenAI from "openai"; const client = new OpenAI(); const model = process.env.OPENAI_MODEL; const vectorStoreId = process.env.OPENAI_VECTOR_STORE_ID; if (!model || !vectorStoreId) { throw new Error("OPENAI_MODEL and OPENAI_VECTOR_STORE_ID are required"); } const response = await client.responses.create({ model, input: "Which section defines the refund approval limit?", tools: [{ type: "file_search", vector_store_ids: [vectorStoreId], }], include: ["file_search_call.results"], }); if (response.status !== "completed") { throw new Error(`file search failed: ${response.status}`); } console.log(response.output_text); console.dir(response.output, { depth: null });
不要用“请求成功”判定迁移完成。至少检查:
- 目标 vector store 确实属于当前 account/project,文件处理状态为 ready;
- golden query 检索到正确文件和片段;
- 最终答案只引用返回证据,annotations/citations 指向正确文件;
- filters、排序、召回数量与旧链路的业务要求一致;
- 空结果、冲突文件、过期文件和越权文件按预期处理;
- P50/P95 延迟、input tokens、工具调用次数和错误率没有越过 guardrail。
如果召回质量退化,先停止灰度扩张;不要靠扩大 prompt 或无限增加返回片段掩盖索引、过滤或文件质量问题。
streaming、Structured Outputs 与后台任务不能照搬
迁移 Run handler 时按实际行为拆分:
- 普通请求:直接等待 Response;
- 用户需要逐字显示:消费
response.created、response.output_text.delta、response.completed和error等 typed events; - 真正长任务:显式使用
background:true,再按Background mode 文档查询 queued/in_progress/completed/failed,并验证取消路径; - 结构化输出:把旧 schema 迁到 Responses 的
text.format,不要继续发送旧式response_format; - 自定义函数:同时处理
response.function_call_arguments.delta/done等事件,不能把文本 delta handler 复用到所有 Item。
每一类都应有失败样本。若旧系统依赖 Run cancellation、并行 tools、timeout 或重试,必须给新实现定义同等业务结果;“SDK 没抛异常”不是等价验证。
旧 Thread 不要全量强迁
OpenAI 当前迁移指南明确表示:没有自动 Thread → Conversation 工具。推荐先把新聊天放到 Responses/Conversations,再按需要 backfill 旧 Thread。
把历史分成三类:
- 活跃且会继续对话:转换支持的 message content,保留 legacy Thread ID 与新 Conversation ID 映射,然后做人工抽样;
- 需要审计但不会继续:只读归档,应用在 UI 中展示,不必让模型继续消费;
- 过期、重复或无法验证:不 backfill,按既有保留政策处理。
官方示例主要转换 message。附件、tool events、metadata、unsupported content 和业务侧状态必须另外盘点。若无法证明语义等价,就保留只读历史,不要把“迁移了所有行”当成功。
用影子流量和灰度证明可以切换
为旧链路和新链路运行同一组输入,但只把旧链路结果返回给用户;新链路只记录经去标识化的对比指标。进入 canary 前,至少完成:
| 场景 | 通过证据 | 失败时动作 |
|---|---|---|
| 基本文本 | 关键约束与拒答边界一致 | 修配置,不扩大流量 |
| 多轮状态 | 关键事实保留且无跨 session 串话 | 回退并检查状态 owner |
| 自定义函数 | 工具名、参数、call_id、输出均正确 | 停止切流,修工具循环 |
| File Search | 正确文件、片段、引用和无结果行为 | 保持旧检索路径 |
| streaming/background | UI 事件、结束、失败、取消均可观察 | 回退 handler |
| Structured Outputs | schema 验证通过,无静默降级 | 拒绝不合格输出 |
| 长会话 | token 增长与 compaction 后事实保留可控 | 限制 turn 或改状态策略 |
| 数据边界 | store/delete/retention 与组织合同一致 | 阻止生产上线 |
| 运行指标 | 质量不降,P95、成本和错误率在阈值内 | 回滚 feature flag |
推荐切流阶梯是:内部账号 → 1% 低风险流量 → 5% → 25% → 100%。每一级至少跨过一个完整业务周期;任何核心功能不等价、数据边界不清或错误率越线,都回到上一级。旧 Assistant、Thread 和 Run 代码只在稳定观察窗口结束后下线,不能先删除再验证。
最终完成标准很具体:新流量不再创建 Assistant/Thread/Run;状态策略有 owner;工具和 File Search 通过 golden tests;streaming、失败与取消可观察;token、延迟、成本和 retention 有 guardrail;feature flag 能恢复旧路径。少一项,就还没有完成生产迁移。



