OpenAI Assistants API 迁移到 Responses API：停用后的生产切换

{
  "thread_id": "thread_archived_example",
  "messages": [
    {
      "id": "msg_saved_1",
      "role": "user",
      "created_at": 1780000000,
      "attachments": [],
      "content": [
        {"type": "text", "text": {"value": "订单是 DEMO-7", "annotations": []}}
      ]
    },
    {
      "id": "msg_saved_2",
      "role": "assistant",
      "created_at": 1780000001,
      "attachments": [],
      "content": [
        {"type": "text", "text": {"value": "请问要查询配送状态吗？", "annotations": []}}
      ]
    }
  ],
  "tool_events": []
}

import json
import sys
from pathlib import Path


def convert(archive):
    items, blockers = [], []
    messages = archive.get("messages")
    if not isinstance(messages, list) or not messages:
        raise ValueError("需要完整消息记录，只有 Thread ID 不够")
    previous_time = None
    for index, message in enumerate(messages):
        label = message.get("id", f"message-{index}")
        timestamp = message.get("created_at")
        if type(timestamp) is not int:
            raise ValueError(f"{label}: 缺少有效时间")
        if previous_time is not None and timestamp < previous_time:
            raise ValueError("消息不是时间升序，请先核对原始顺序")
        previous_time = timestamp
        role = message.get("role")
        if role not in ("user", "assistant"):
            blockers.append(f"{label}: 不支持的角色")
            continue
        parts = message.get("content")
        if not isinstance(parts, list) or not parts:
            blockers.append(f"{label}: 缺少内容")
            continue
        content = []
        if message.get("attachments"):
            blockers.append(f"{label}: 附件需要重新确认访问与内容")
        for part in parts:
            if part.get("type") != "text":
                blockers.append(f"{label}: 非文本内容需要单独转换")
                continue
            text = part.get("text", {})
            if not isinstance(text.get("value"), str):
                blockers.append(f"{label}: 文本内容无效")
                continue
            if text.get("annotations"):
                blockers.append(f"{label}: 引用需要核对原文件")
            content.append({
                "type": "input_text" if role == "user" else "output_text",
                "text": text["value"],
            })
        if content:
            items.append({"role": role, "content": content})
    if archive.get("tool_events"):
        blockers.append("工具事件需确认结果、授权和业务动作是否已完成")
    return {
        "legacy_thread_id": archive.get("thread_id"),
        "ready_to_resume": not blockers,
        "input": items,
        "blockers": blockers,
        "original": archive,
    }


if __name__ == "__main__":
    source, destination = map(Path, sys.argv[1:3])
    if destination.exists():
        raise FileExistsError("不覆盖已有转换结果")
    result = convert(json.loads(source.read_text(encoding="utf-8")))
    destination.write_text(
        json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8"
    )
    print(json.dumps({
        "ready_to_resume": result["ready_to_resume"],
        "blockers": result["blockers"],
    }, ensure_ascii=False))

import json
import sys
from pathlib import Path
from openai import OpenAI

MODEL, QUESTION = sys.argv[1:3]
STATE_PATH = Path("session.json")
INSTRUCTIONS = (
    "你是订单客服。订单状态必须来自 get_order_status，不能猜测。"
    "只查询当前用户已获授权的订单；工具返回错误时直接说明无法查询。"
)
TOOLS = [{
    "type": "function",
    "name": "get_order_status",
    "description": "查询当前用户有权限访问的订单配送状态。",
    "strict": True,
    "parameters": {
        "type": "object",
        "properties": {"order_id": {"type": "string"}},
        "required": ["order_id"],
        "additionalProperties": False,
    },
}]

# 合成的当前用户权限；生产中由已认证身份及业务数据库提供。
ALLOWED_ORDERS = {"DEMO-7"}
ORDERS = {"DEMO-7": {"status": "shipped", "carrier": "演示快递"}}


def save_state(history, ready):
    temporary = STATE_PATH.with_suffix(".tmp")
    temporary.write_text(json.dumps({
        "ready": ready, "config_version": "orders-v1", "history": history,
    }, ensure_ascii=False, indent=2), encoding="utf-8")
    temporary.replace(STATE_PATH)


def execute(call):
    if call.name != "get_order_status":
        raise ValueError("未知工具，停止执行")
    args = json.loads(call.arguments)
    if (not isinstance(args, dict) or set(args) != {"order_id"}
            or not isinstance(args["order_id"], str)):
        raise ValueError("工具参数无效")
    order_id = args["order_id"]
    if order_id not in ALLOWED_ORDERS:
        return {"error": "not_authorized"}
    return ORDERS.get(order_id, {"error": "not_found"})


if STATE_PATH.exists():
    state = json.loads(STATE_PATH.read_text(encoding="utf-8"))
    if not state["ready"] or state["config_version"] != "orders-v1":
        raise RuntimeError("会话未完成或配置不匹配，请先人工核对")
    history = state["history"]
else:
    converted = json.loads(Path("converted.json").read_text(encoding="utf-8"))
    if not converted["ready_to_resume"]:
        raise RuntimeError("历史存在未处理内容，不能自动续聊")
    history = converted["input"]

save_state(history, True)
Path("last-ready.json").write_text(
    STATE_PATH.read_text(encoding="utf-8"), encoding="utf-8"
)
history.append({"role": "user", "content": QUESTION})
save_state(history, False)
client = OpenAI()
seen_calls = set()

for _ in range(6):
    response = client.responses.create(
        model=MODEL,
        instructions=INSTRUCTIONS,
        tools=TOOLS,
        input=history,
        store=False,
        parallel_tool_calls=False,
        include=["reasoning.encrypted_content"],
    )
    if response.status != "completed":
        raise RuntimeError(f"Response 未成功完成：{response.status}")
    history.extend(item.model_dump(exclude_none=True) for item in response.output)
    save_state(history, False)
    calls = [item for item in response.output if item.type == "function_call"]
    if calls:
        for call in calls:
            if not call.call_id or call.call_id in seen_calls:
                raise RuntimeError("调用标识缺失或重复")
            seen_calls.add(call.call_id)
            result = execute(call)
            history.append({
                "type": "function_call_output",
                "call_id": call.call_id,
                "output": json.dumps(result, ensure_ascii=False),
            })
        save_state(history, False)
        continue
    if any(item.type != "message" and item.type != "reasoning"
           for item in response.output):
        raise RuntimeError("出现本示例未处理的输出类型")
    texts = [part.text for item in response.output
             if item.type == "message"
             and getattr(item, "phase", None) in (None, "final_answer")
             for part in item.content if part.type == "output_text"]
    if not texts or not "".join(texts).strip():
        raise RuntimeError("没有可显示的最终文本，请检查拒绝或中间输出")
    save_state(history, True)
    print("".join(texts))
    break
else:
    raise RuntimeError("超过工具循环上限，保留现场并停止")

VECTOR_STORE_ID = "YOUR_VERIFIED_VECTOR_STORE_ID"
response = client.responses.create(
    model=MODEL,
    store=False,
    input="退款审批额度在什么文件的哪一节？请依据检索内容回答。",
    tools=[{
        "type": "file_search",
        "vector_store_ids": [VECTOR_STORE_ID],
    }],
    include=["file_search_call.results"],
)
if response.status != "completed":
    raise RuntimeError(f"检索请求未成功：{response.status}")
searches = [item for item in response.output if item.type == "file_search_call"]
if not searches or not any(getattr(item, "results", None) for item in searches):
    raise RuntimeError("没有可核对的检索结果，不能判为迁移成功")
print(response.output_text)
print(response.model_dump_json(indent=2))