Agent 工程体系 · 第 92/98 篇。内容以 2026 年 9 月可验证的公开规范和稳定接口为基线；框架版本敏感能力会明确标注，不把实验行为写成通用保证。

Agent API 契约：会话、消息、附件、事件、错误和幂等键

Agent API 不是“把一段字符串发给模型，再返回一段字符串”的 HTTP 包装。只要系统支持多轮上下文、文件输入、工具调用、流式输出、断线恢复、人工审批或异步执行，API 实际上同时承担了六类状态的表达：

1. 会话状态：这次交互属于谁，历史上下文是什么；
2. 消息状态：用户、Agent、工具和系统分别产生了哪些输入输出；
3. 附件状态：文件是否上传完成、是否可被模型读取、是否仍然保留；
4. 运行事件：一次 Agent 执行过程中发生了什么；
5. 错误状态：请求失败、执行失败、工具失败和流式连接失败分别如何表示；
6. 幂等状态：客户端重试时，服务端如何避免重复创建消息、重复扣费或重复执行工具。

本文给出一套面向生产系统的 Agent API 契约。文中的资源名称、字段和 HTTP 路径是 WR BLOG Agent 工程基线的建议契约，不是 OpenAI 官方 API 的逐字段复制。OpenAI Agents SDK 主要提供 Agent、工具、交接、护栏、会话和流式运行等运行时能力；它默认使用 Responses API，并在更高层管理运行循环、工具执行和会话状态。(openai.github.io)

---

一、先定义边界：资源 API 与运行时协议不是一回事

一个常见错误是把以下三个对象混为一谈：

• 消息：已经存在于会话历史中的业务事实；
• 运行：Agent 为处理一次输入而进行的一段执行；
• 事件：运行过程中对外发布的时间序列。

它们的关系可以表示为：

flowchart LR
    C[客户端] -->|创建消息| S[会话 Session]
    S -->|启动一次运行| R[运行 Run]
    R --> M[模型调用]
    R --> T[工具调用]
    R --> G[护栏与审批]
    R --> E[事件 Event Stream]
    R -->|完成后写入| S
    E --> C
    S --> C

会话是长期资源，运行是一次执行，事件是运行的观察面。

因此：

• POST /sessions/{session_id}/messages 表示向会话提交一个新输入；
• 服务端可以同步返回最终消息，也可以返回一个 run_id；
• GET /runs/{run_id}/events 或 SSE/WebSocket 用于观察运行过程；
• 最终产生的 Agent 消息才会成为会话历史的一部分；
• 中间的 delta、工具开始、工具结束、usage 和错误事件，不应直接当作普通聊天消息保存。

OpenAI Agents SDK 的会话机制也是类似分层：运行开始前读取历史，运行结束后保存本次产生的新项目，包括用户输入、Agent 响应和工具调用等，而不是让调用方每次手动拼接全部历史。(openai.github.io)

---

二、契约的核心对象

1. Session：会话

会话是具有明确身份、访问边界和历史顺序的交互容器。

建议定义：

{
  "id": "sess_01JABC...",
  "tenant_id": "tenant_demo",
  "subject_id": "user_123",
  "agent_id": "support_agent",
  "status": "active",
  "revision": 7,
  "created_at": "2026-09-01T08:00:00Z",
  "updated_at": "2026-09-01T08:03:12Z",
  "expires_at": null,
  "metadata": {
    "channel": "web",
    "locale": "zh-CN"
  }
}

关键字段含义如下：

字段	含义
id	全局唯一会话 ID，不应使用用户可猜测的连续整数
tenant_id	租户或项目边界，防止跨租户访问
subject_id	最终用户或调用主体
agent_id	处理该会话的逻辑 Agent
status	active、archived、deleted 等生命周期状态
revision	会话历史版本，用于并发控制
metadata	业务标签，不应被默认拼入模型上下文

revision 不是装饰字段。假设两个请求同时读取到版本 7：

请求 A：读取 revision=7，追加消息，写入 revision=8
请求 B：读取 revision=7，追加消息，写入 revision=8

如果数据库没有条件更新，B 可能覆盖 A，或者两条消息的顺序与客户端看到的顺序不一致。

因此至少要满足：

$$\text{update succeeds} \iff \text{stored\_revision} = \text{expected\_revision}$$

写入成功后：

$$\text{new\_revision} = \text{expected\_revision} + 1$$

失败时返回 409 conflict，让客户端重新读取会话状态，而不是盲目重试同一个写操作。

OpenAI Agents SDK 的 Session 负责为特定会话保存历史，并在每次运行前读取历史、运行后保存新项目；SDK 还支持 SQLite、Redis、SQLAlchemy、MongoDB 等不同后端。(openai.github.io)

---

2. Message：消息

消息是会话中的持久化事实。它回答的是“谁在什么时候产生了什么内容”。

建议消息结构：

{
  "id": "msg_01JABD...",
  "session_id": "sess_01JABC...",
  "run_id": "run_01JABE...",
  "role": "user",
  "status": "completed",
  "content": [
    {
      "type": "text",
      "text": "请分析这份合同"
    },
    {
      "type": "attachment_ref",
      "attachment_id": "att_01JABF..."
    }
  ],
  "sequence": 8,
  "created_at": "2026-09-01T08:03:00Z",
  "metadata": {}
}

role 描述消息的生产者，而不是权限：

• user：用户输入；
• assistant：Agent 面向用户的输出；
• tool：工具执行结果；
• system：系统注入的控制信息；
• developer：开发者指令，是否对外暴露取决于产品边界。

content 必须是数组，而不是单一字符串。原因是一次消息可能同时包含文本、图片、文件引用或结构化内容。把内容强行压成字符串，会导致附件、引用和多模态输入无法表达。

消息的 status 至少应区分：

pending   已接受但尚未进入运行
streaming 正在生成
completed 已形成稳定结果
failed    生成失败
cancelled 被主动取消

需要注意，流式输出中的每个文本片段不是一条独立消息。客户端可以通过：

message_id + event_id + delta

逐步构造 UI，但服务端最终只应提交一条具有稳定 message_id 的 assistant 消息。

---

3. Attachment：附件

附件是独立于消息内容的可寻址对象。消息只保存附件引用，不直接保存二进制内容。

建议附件状态机：

stateDiagram-v2
    [*] --> uploading
    uploading --> uploaded: 上传完成
    uploading --> failed: 上传失败
    uploaded --> scanning: 安全检查
    scanning --> ready: 检查通过
    scanning --> rejected: 类型或内容不允许
    ready --> expired: 到期
    ready --> deleted: 删除
    failed --> [*]
    rejected --> [*]

附件对象示例：

{
  "id": "att_01JABF...",
  "filename": "contract.pdf",
  "media_type": "application/pdf",
  "size_bytes": 248193,
  "sha256": "8e6f...",
  "status": "ready",
  "purpose": "agent_input",
  "storage_uri": "internal://objects/...",
  "expires_at": "2026-09-08T08:00:00Z",
  "created_at": "2026-09-01T08:00:11Z"
}

附件上传成功不等于附件可用。至少存在三个时刻：

1. 文件字节已上传；
2. 文件安全扫描和格式识别完成；
3. 下游模型或检索系统已经可以读取它。

只有达到 ready，消息接口才应允许引用它。否则会出现“消息创建成功，但运行时模型找不到文件”的异步失败。

在 OpenAI Responses API 中，文件输入可以通过 input_file 和 file_id 引用；官方示例先上传文件，再将文件 ID 放入用户内容中。这个模式也说明了文件对象和消息对象应当分离。(developers.openai.com)

推荐接口：

POST /v1/attachments
Content-Type: multipart/form-data
Idempotency-Key: upload-contract-20260901-001

响应：

{
  "id": "att_01JABF...",
  "status": "scanning",
  "filename": "contract.pdf",
  "size_bytes": 248193
}

客户端必须轮询或订阅状态：

GET /v1/attachments/att_01JABF...

当状态为 ready 后，再提交消息。不要在服务端收到文件后同步阻塞等待所有解析、OCR、索引和病毒扫描完成，否则大文件会把普通 API 请求变成长连接任务。

---

三、一次 Agent 调用的完整生命周期

推荐将“提交用户输入”和“执行 Agent”视为一个逻辑操作，但在协议上保留独立的 run 资源。

1. 创建消息并启动运行

POST /v1/sessions/sess_01JABC/runs
Content-Type: application/json
Idempotency-Key: turn-user123-20260901-0007

请求：

{
  "input": {
    "role": "user",
    "content": [
      {
        "type": "text",
        "text": "请总结合同中的付款风险"
      },
      {
        "type": "attachment_ref",
        "attachment_id": "att_01JABF"
      }
    ]
  },
  "mode": "stream",
  "agent": {
    "id": "contract_review",
    "version": "2026-08-15"
  },
  "client_context": {
    "locale": "zh-CN"
  }
}

响应：

{
  "run_id": "run_01JABE",
  "session_id": "sess_01JABC",
  "user_message_id": "msg_01JABG",
  "status": "queued",
  "stream": {
    "protocol": "sse",
    "cursor": null
  }
}

这里的 run_id 表示一次 Agent 执行，而 user_message_id 表示已经接受的用户输入。二者不能合并成一个 ID，因为一次消息可能触发重试、恢复、审批或多个底层模型调用。

2. 运行状态

建议状态如下：

queued
running
waiting_for_approval
waiting_for_tool
completed
failed
cancelled
expired

状态转移必须有单向性约束。例如：

queued -> running
running -> waiting_for_approval
waiting_for_approval -> running
running -> completed
running -> failed
running -> cancelled

completed、failed、cancelled 和 expired 是终态。终态运行不能因为客户端重复请求而重新执行。

---

四、事件协议：Delta 不是消息，Finish 才是边界

一次运行的事件流可以抽象为：

事件 = {
  event_id,
  run_id,
  sequence,
  type,
  occurred_at,
  data
}

建议统一事件封装：

{
  "event_id": "evt_00000042",
  "run_id": "run_01JABE",
  "sequence": 42,
  "type": "message.delta",
  "occurred_at": "2026-09-01T08:03:04.120Z",
  "data": {
    "message_id": "msg_01JABH",
    "content_index": 0,
    "delta": "付款"
  }
}

1. Delta 事件

message.delta 表示内容增量：

{
  "type": "message.delta",
  "data": {
    "message_id": "msg_01JABH",
    "delta": "付款"
  }
}

客户端拼接：

"合同"
+ "中的"
+ "付款"
+ "风险"
= "合同中的付款风险"

但客户端不应根据收到的最后一个 delta 判断运行完成。网络断开可能发生在任意片段之后，服务端也可能尚未完成工具调用、usage 统计或会话落库。

OpenAI Responses API 的流式接口使用带类型的语义事件，例如 response.created、response.output_text.delta、response.completed 和 error；Agents SDK 的底层流式事件也可以包装这些原始 Responses 事件。(developers.openai.com)

2. Tool 事件

工具调用不能只暴露成一段普通文本，否则客户端无法展示进度、审计参数或区分工具失败。

建议：

{
  "type": "tool.started",
  "data": {
    "tool_call_id": "call_123",
    "tool_name": "query_invoice",
    "arguments": {
      "invoice_id": "INV-2026-001"
    }
  }
}

工具结束：

{
  "type": "tool.completed",
  "data": {
    "tool_call_id": "call_123",
    "tool_name": "query_invoice",
    "result": {
      "status": "overdue",
      "amount": 12800
    },
    "duration_ms": 431
  }
}

工具失败：

{
  "type": "tool.failed",
  "data": {
    "tool_call_id": "call_123",
    "tool_name": "query_invoice",
    "error": {
      "code": "upstream_timeout",
      "message": "发票服务响应超时",
      "retryable": true
    }
  }
}

工具参数和结果是否直接发送给浏览器，需要由权限策略决定。面向终端用户的 UI 通常只展示工具名称和进度，不展示完整参数，因为参数可能包含令牌、个人信息或内部查询条件。

3. Usage 事件

usage 是计量结果，不应被当作模型内容。

{
  "type": "usage",
  "data": {
    "input_tokens": 1820,
    "output_tokens": 436,
    "cached_input_tokens": 900,
    "tool_calls": 2,
    "estimated_cost": {
      "currency": "USD",
      "amount": "0.0124"
    }
  }
}

Usage 可能在运行结束前不可得，尤其是存在多次模型调用、工具调用或自动压缩时。因此：

• UI 可以显示“正在统计”；
• 计费系统应以服务端最终记录为准；
• 客户端提交的 usage 不可信；
• 失败运行是否计费，必须由服务端明确规定。

Agents SDK 的结果对象提供原始响应、运行产生的新项目和最终输出等不同观察面；这些信息用途不同，不能用 final_output 代替完整审计记录。(openai.github.io)

4. Finish 事件

run.finished 是客户端判断本次流是否完整的唯一业务边界：

{
  "type": "run.finished",
  "data": {
    "status": "completed",
    "final_message_id": "msg_01JABH",
    "finish_reason": "stop",
    "usage_event_id": "evt_00000048"
  }
}

推荐的最小事件顺序：

run.started
message.started
message.delta
tool.started
tool.completed
message.delta
usage
run.finished

允许出现零个或多个 tool.* 事件，但必须满足：

$$\text{run.finished} \Rightarrow \text{所有必需的消息与工具状态已确定}$$

如果客户端只收到 message.delta 而没有收到 run.finished，应把运行标记为“未知”，而不是“成功”。

---

五、SSE、断线与恢复

SSE 示例：

GET /v1/runs/run_01JABE/events
Accept: text/event-stream
Last-Event-ID: evt_00000042

服务端输出：

id: evt_00000043
event: message.delta
data: {"run_id":"run_01JABE","sequence":43,"data":{"delta":"风险"}}

id: evt_00000044
event: usage
data: {"run_id":"run_01JABE","sequence":44,"data":{"output_tokens":436}}

id: evt_00000045
event: run.finished
data: {"run_id":"run_01JABE","sequence":45,"data":{"status":"completed"}}

恢复协议必须明确三个事实：

1. event_id 是否持久化；
2. 客户端从哪个位置恢复；
3. 服务端是否保留足够长的事件历史。

推荐使用单调递增的 sequence 作为恢复游标，并使用 event_id 作为外部标识：

客户端已确认 sequence=42
重连时请求 sequence > 42 的事件

事件投递通常采用 至少一次 语义。原因是服务端发送事件和客户端确认之间存在窗口：

服务端发送 evt_43
客户端已经处理 evt_43
网络在 ACK 前断开
服务端无法知道客户端是否成功

因此客户端必须按 event_id 去重：

seen = set()

def consume(event):
    if event["event_id"] in seen:
        return
    seen.add(event["event_id"])

    if event["type"] == "message.delta":
        append_delta(event["data"]["message_id"], event["data"]["delta"])

这不等于允许事件乱序。服务端仍应保证同一运行内：

$$sequence_{n+1} > sequence_n$$

客户端如果发现跳号，例如收到 45 后直接收到 48，应暂停拼接并请求 46、47；不能把缺失的内容静默忽略。

Agents SDK 明确要求调用方持续消费 stream_events() 直到迭代器结束，因为最后一个可见 token 到达后，会话持久化、审批状态处理或历史压缩仍可能继续执行；只有流结束后，运行才算真正完成。(openai.github.io)

---

六、错误契约：错误不是一个字符串

错误至少分为四层：

层次	示例	是否可重试
HTTP 请求错误	JSON 无法解析、字段缺失	修正请求后重试
资源错误	会话不存在、附件未就绪	通常不可直接重试
运行错误	模型超时、超过最大轮数	视错误类型
流错误	连接中断、游标过期	重新连接或重新获取状态

统一错误结构：

{
  "error": {
    "code": "attachment_not_ready",
    "message": "附件仍在安全扫描中",
    "type": "invalid_state",
    "retryable": true,
    "request_id": "req_01JABK",
    "run_id": null,
    "details": {
      "attachment_id": "att_01JABF",
      "current_status": "scanning",
      "retry_after_ms": 3000
    }
  }
}

字段约束：

• code：稳定、机器可判断，不能依赖 message 做分支；
• message：面向开发者或用户的说明；
• type：错误类别；
• retryable：服务端给出的初步判断；
• request_id：用于日志关联；
• details：结构化诊断信息。

建议错误码：

invalid_request
unauthorized
forbidden
session_not_found
session_conflict
attachment_not_ready
attachment_rejected
quota_exceeded
rate_limited
model_timeout
model_refusal
tool_timeout
tool_failed
approval_required
run_not_found
cursor_expired
idempotency_conflict
internal_error

HTTP 状态码与业务错误码不能互相替代。例如多个原因都可能返回 429：请求速率过高、余额耗尽、项目消费上限或组织使用上限。OpenAI 官方错误指南也将这些情况区分为不同原因，而不是让客户端仅凭 429 判断处理方式。(developers.openai.com)

因此客户端的判断顺序应当是：

先读 HTTP status
再读 error.code
最后根据 retryable、Retry-After 和业务状态决定动作

典型策略：

401/403              不重试，检查认证与权限
400                  不重试，修正请求
404                  不重试，确认资源 ID
409                  重新读取资源或检查幂等键
429 rate_limited     按 Retry-After 或退避重试
429 spend_limit      不重试，等待额度恢复或切换配额
500/503              有上限地退避重试
tool_timeout         只有工具声明幂等时才自动重试

指数退避可以写成：

$$delay_n = \min(delay_{max}, base \times 2^n) + jitter$$

其中 n 是第几次重试，base 是基础等待时间，jitter 是随机扰动。随机扰动用于避免大量客户端在同一时刻再次冲击服务端。

---

七、幂等键：防止重复写入，不保证重复副作用自动安全

幂等键是客户端为一次逻辑请求生成的稳定标识。它解决的是：

客户端发送请求
服务端已经完成
响应在网络中丢失
客户端无法判断是否成功
客户端再次发送

请求头：

Idempotency-Key: turn-user123-20260901-0007

服务端应保存：

{
  "scope": "tenant_demo:user_123",
  "key": "turn-user123-20260901-0007",
  "request_hash": "sha256:...",
  "status": "completed",
  "response_status": 201,
  "response_body": {},
  "expires_at": "2026-09-08T08:03:00Z"
}

处理规则：

第一次请求

不存在 key
-> 计算 request_hash
-> 原子创建幂等记录
-> 执行业务操作
-> 保存响应
-> 返回响应

重复请求且请求体相同

key 已存在
request_hash 相同
-> 不重复执行
-> 返回第一次请求的结果

重复请求但请求体不同

key 已存在
request_hash 不同
-> 返回 409 idempotency_conflict

形式化表示：

$$same(key, scope) \land hash_1 \neq hash_2 \Rightarrow 409$$

幂等键必须绑定作用域。不能只用全局 key，否则用户 A 的 abc-001 可能与用户 B 的 abc-001 冲突。推荐作用域至少包含租户和主体：

scope = tenant_id + ":" + subject_id + ":" + operation

幂等键的边界

幂等键只能保证“同一个 API 操作不会被服务端重复接受”，不能自动让所有外部副作用安全。

例如 Agent 调用支付工具：

Agent -> payment.create(order_id=100)
payment 服务已扣款
Agent 网关超时
Agent 重试

如果工具没有自己的幂等键，支付可能被创建两次。

因此工具调用也必须传递稳定的副作用键：

{
  "tool_name": "create_payment",
  "arguments": {
    "order_id": "100",
    "idempotency_key": "run_01JABE:call_123"
  }
}

工具端再以 order_id 或 run_id + tool_call_id 建立唯一约束。API 幂等和工具幂等是两层机制，不能只实现外层。

---

八、会话并发：同一会话是否允许并行运行

如果同一会话同时收到两个用户请求：

请求 A：总结合同
请求 B：把总结翻译成英文

系统必须先定义语义，而不是交给数据库随机排序。

方案一：串行会话

同一 session_id 同时只允许一个活跃运行：

A running
B -> 409 session_busy

优点是上下文顺序清晰，适合客服和事务型 Agent。缺点是一个慢工具会阻塞后续消息。

方案二：并行会话

允许多个运行并行，但每个运行绑定一个输入版本：

{
  "session_id": "sess_01JABC",
  "base_revision": 7
}

运行完成时要求：

current_revision == base_revision

若不一致，则：

• 丢弃自动提交；
• 生成分支；
• 或要求客户端重新合并。

方案三：逻辑分支

把会话历史视为一棵树：

revision 7
├── run A -> revision 8A
└── run B -> revision 8B

这适合评测、草稿和多方案推理，但会增加 UI、存储和合并复杂度。

普通对话系统通常选择方案一，因为“用户看到的消息顺序”比最大并发量更重要。若使用 Agents SDK 的 Session，必须把 Session 的读写并发策略与外层 API 的会话锁策略统一起来，不能一边在网关层串行，一边在后台直接绕过网关修改同一会话。

---

九、一个最小可运行的 FastAPI 契约骨架

下面的代码展示接口边界，不连接真实模型，便于先验证资源、幂等和事件语义。

from __future__ import annotations

import asyncio
import hashlib
import json
import uuid
from dataclasses import dataclass, field
from typing import Any

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel, Field

app = FastAPI()

sessions: dict[str, dict[str, Any]] = {}
runs: dict[str, dict[str, Any]] = {}
idempotency: dict[tuple[str, str], dict[str, Any]] = {}


class TextContent(BaseModel):
    type: str = "text"
    text: str


class InputMessage(BaseModel):
    role: str = "user"
    content: list[TextContent]


class RunRequest(BaseModel):
    input: InputMessage
    mode: str = Field(default="sync", pattern="^(sync|stream)$")


def request_hash(body: RunRequest) -> str:
    raw = json.dumps(body.model_dump(), sort_keys=True, ensure_ascii=False)
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()


@app.post("/v1/sessions")
async def create_session():
    session_id = "sess_" + uuid.uuid4().hex
    sessions[session_id] = {
        "id": session_id,
        "revision": 0,
        "messages": [],
    }
    return sessions[session_id]


@app.post("/v1/sessions/{session_id}/runs")
async def create_run(
    session_id: str,
    body: RunRequest,
    idempotency_key: str | None = Header(default=None, alias="Idempotency-Key"),
):
    if session_id not in sessions:
        raise HTTPException(
            status_code=404,
            detail={
                "code": "session_not_found",
                "message": "session does not exist",
            },
        )

    if not idempotency_key:
        raise HTTPException(
            status_code=400,
            detail={
                "code": "idempotency_key_required",
                "message": "Idempotency-Key is required",
            },
        )

    scope = f"session:{session_id}:create_run"
    key = (scope, idempotency_key)
    body_hash = request_hash(body)

    previous = idempotency.get(key)
    if previous:
        if previous["request_hash"] != body_hash:
            raise HTTPException(
                status_code=409,
                detail={
                    "code": "idempotency_conflict",
                    "message": "same key was used with a different request",
                },
            )
        return previous["response"]

    run_id = "run_" + uuid.uuid4().hex
    message_id = "msg_" + uuid.uuid4().hex

    session = sessions[session_id]
    session["revision"] += 1
    session["messages"].append({
        "id": message_id,
        "run_id": run_id,
        "role": "user",
        "content": body.input.model_dump()["content"],
        "sequence": session["revision"],
    })

    response = {
        "run_id": run_id,
        "user_message_id": message_id,
        "status": "queued",
    }

    runs[run_id] = {
        "id": run_id,
        "session_id": session_id,
        "status": "queued",
        "events": [],
    }
    idempotency[key] = {
        "request_hash": body_hash,
        "response": response,
    }

    asyncio.create_task(execute_run(run_id))
    return response


async def execute_run(run_id: str):
    run = runs[run_id]
    run["status"] = "running"

    def emit(event_type: str, data: dict[str, Any]):
        sequence = len(run["events"]) + 1
        run["events"].append({
            "event_id": f"{run_id}:evt:{sequence}",
            "sequence": sequence,
            "type": event_type,
            "data": data,
        })

    emit("run.started", {"run_id": run_id})
    emit("message.started", {"message_id": f"msg_{uuid.uuid4().hex}"})

    for delta in ["已收到", "你的请求。"]:
        emit("message.delta", {"delta": delta})
        await asyncio.sleep(0.05)

    emit("usage", {
        "input_tokens": 10,
        "output_tokens": 4,
    })
    emit("run.finished", {
        "status": "completed",
        "finish_reason": "stop",
    })
    run["status"] = "completed"

这个骨架刻意没有把事件流接口塞进后台任务中，因为生产系统还需要处理：

• 事件持久化；
• SSE 连接；
• Last-Event-ID 恢复；
• 多进程之间的事件广播；
• 运行终态查询；
• 任务取消；
• 工具调用和审批；
• 事件过期后的快照恢复。

它展示了三个重要事实：

1. 幂等记录必须在启动后台任务前建立；
2. 用户消息和运行资源必须有不同 ID；
3. 运行完成必须产生明确的 run.finished，不能依赖 HTTP 请求是否仍然保持连接。

---

十、与 Agents SDK 的映射

如果服务端使用 OpenAI Agents SDK，可以按以下方式映射：

WR API 概念	Agents SDK 概念
session_id	Session 实例或底层存储中的 session ID
run_id	一次 Runner.run() 或 Runner.run_streamed()
message	RunItem、输入项或输出项
message.delta	RawResponsesStreamEvent 中的文本增量
tool.*	RunItemStreamEvent 的工具调用和工具输出
run.finished	stream_events() 迭代器结束且结果状态已确定
waiting_for_approval	result.interruptions 与 RunState
运行错误	AgentsException 及其 RunErrorDetails

Agents SDK 的高层运行项事件提供“消息生成”“工具执行”“交接”等语义级事件，而原始响应事件则适合逐 token 推送。两者应该同时保留：前端通常消费语义事件，诊断和协议适配层则保存原始事件。(openai.github.io)

审批暂停不能被当作失败。SDK 的流会结束，并在 result.interruptions 中暴露待处理审批；调用方应转换为 RunState，审批或拒绝后继续运行，而不是创建一条新的用户消息。(openai.github.io)

同样，SDK 的 final_output 在流结束前可能仍为 None，运行也可能因为审批中断而没有最终输出。生产 API 不应在收到首个文本片段时就把运行标记为完成。(openai.github.io)

---

十一、常见错误及诊断方式

把会话 ID 当作消息 ID

表现：

刷新页面后消息重复
多标签页互相覆盖
工具结果无法关联到具体一轮运行

修复：会话、运行、消息、工具调用和事件分别生成 ID，并建立显式关联。

把断线当作运行失败

表现：

客户端断线
服务端任务仍在执行
客户端重连后又创建一次运行
模型调用和工具副作用重复

修复：断线只改变连接状态，不改变运行状态。客户端应先查询：

GET /v1/runs/{run_id}

再使用事件游标恢复。

只做 HTTP 层重试

表现：

POST 重试后出现两条用户消息
同一个外部订单创建两次

修复：写接口必须使用幂等键；有副作用的工具必须有独立的工具幂等键。

用错误消息文本判断错误类型

表现：

英文环境下客户端分支失效
服务端修改文案导致前端无法处理

修复：使用稳定的 error.code，message 仅用于展示和诊断。

把完整模型事件直接暴露给浏览器

表现：

提示词泄露
工具参数包含内部凭据
内部模型错误暴露给最终用户

修复：建立事件投影层：

内部事件流 -> 权限过滤 -> 面向用户的事件流

内部审计流可以保留完整数据，外部流只保留允许公开的字段。

自动压缩与低延迟流式冲突

如果会话历史在每轮后自动压缩，运行可能在最后一个可见 token 后仍等待压缩完成。Agents SDK 文档明确提示，自动压缩可能让流在最后一个输出 token 后继续保持一段时间。低延迟场景可以关闭自动压缩，改为在轮次之间或空闲时间手动执行，但这属于运行策略取舍，不是协议层保证。(openai.github.io)

---

十二、最终契约应保证的性质

一套可用于生产的 Agent API，至少应满足以下不变量：

会话顺序

同一会话中，已提交消息具有稳定顺序：

$$m_i.sequence < m_{i+1}.sequence$$

事件可恢复

对于事件保留期内的游标 c：

$$resume(c) = \{e \mid e.sequence > c\}$$

恢复结果不能依赖客户端是否曾经成功收到某个事件。

终态不可逆

$$status \in \{completed, failed, cancelled, expired\} \Rightarrow \text{不再进入 running}$$

幂等冲突可检测

$$same(scope, key) \land different(request\_hash) \Rightarrow 409$$

消息与增量可关联

每个 message.delta 必须能通过 message_id 归属到一条消息；每个工具事件必须能通过 tool_call_id 归属到一次工具调用。

失败原因可分类

客户端不应只知道“失败了”，还应知道：

是否已经创建用户消息
是否已经执行工具
是否可能产生外部副作用
是否允许恢复
是否需要人工处理

当这些性质被明确写入契约后，Agent API 才能同时支持同步调用、流式输出、断线恢复、人工审批、模型切换、工具重试和生产审计，而不会退化为一组无法推断状态的字符串接口。

---

系列导航与关联阅读

• 系列入口：Agent 工程完整路线：从运行循环、记忆与协议到安全、评测和生产交付
• 上一篇：Agent 流式事件协议：Delta、Tool、Usage、Finish、断线和恢复
• 下一篇：Agent 网关：认证、模型路由、配额、策略、审计和协议适配

官方资料

• OpenAI Agents SDK Documentation
• OpenAI Agents Guide

本文依据 Agent、模型、协议与框架官方资料重新梳理；正文、示例与生产清单由 WR BLOG 编写。