一次 Responses 流不是一串同级消息,而是响应级生命周期包住若干 Item 级生命周期。Created 说明响应存在,Item Added 建立活动项,Delta 只更新暂态展示,Item Done 交付完整事实,Completed 才关闭采样边界。
具体问题与启用条件
本节建立统一事件状态机和层级关系。文字、Reasoning 与工具 Diff 的专属组装规则分别在 5.14—5.18。
| 条件来源 | 决定字段或状态 | 对本机制的影响 |
|---|---|---|
| 线协议 | SSE 或 Responses WebSocket | 都映射成同一 ResponseEvent |
| 内容类型 | Message、Reasoning、Tool Call 等 | 决定 Item 内可出现的 Delta |
| 终态 | response.completed/failed/incomplete | 决定一次采样成功或失败 |
协议、类型与状态所有权
| 状态或协议 | 所有者 | 生命周期 | 关键不变量 |
|---|---|---|---|
| ResponsesStreamEvent | codex-api parser | 单个线事件 | 保留线协议字段 |
| ResponseEvent | 传输无关映射层 | 交给 Core 的单事件 | SSE/WS 语义一致 |
| active_item | try_run_sampling_request | Item Added 到 Done | 同一时刻最多一个 |
| SamplingRequestResult | 采样循环 | 到 Completed | 汇总 follow-up 与最后消息 |
机制调用链如下:
response.created
→ response.output_item.added(item)
→ item 专属 delta/part events*
→ response.output_item.done(full item)
→ 可重复下一个 item
→ response.completed(response id, usage, end_turn)
→ SamplingRequestResult
机制怎样工作
codex-api 只把认识的线事件提升为 ResponseEvent;未知类型被 trace 后忽略。Core 对 Created 不修改历史。Item Added 尝试把 ResponseItem 投影成客户端 TurnItem,必要时发 ItemStarted,并把它放入 active_item。Delta 必须依附当前活动项。
Item Done 到达时先结束工具 Diff consumer、清走 active item、flush 文本 parser,再处理服务端给出的完整 ResponseItem。一次响应可以包含多个 Done Item,所以 Item Done 不结束采样。Completed 才 flush 所有剩余文本状态、发 RawResponseCompleted、记录 token usage、刷新 diff/token count,并根据 end_turn=false 设置后续采样。response.failed 和 incomplete 在更低层已转成错误,不进入成功状态机。
Python 风格伪代码
这段伪代码保留生产实现中会改变结果的状态、分支和异步边界;认证 SDK、遥测字段和 Rust 所有权样板被折叠为明确的领域对象。
async def consume_response(stream: AsyncIterator[ResponseEvent]) -> SamplingResult:
active: TurnItem | None = None
needs_follow_up = False
last_message: str | None = None
async for event in stream:
match event:
case Created():
pass
case OutputItemAdded(item):
if active is not None:
raise ProtocolError("overlapping active items")
active = await begin_item(item)
case Delta() as delta:
if active is None:
raise ProtocolError("delta without active item")
await publish_transient_delta(active, delta)
case OutputItemDone(item):
await finish_transient_state(active)
active = None
result = await commit_complete_item(item)
needs_follow_up |= result.needs_follow_up
last_message = result.last_message or last_message
case Completed(response_id, usage, end_turn):
await flush_all_transient_state()
await record_response_boundary(response_id, usage)
return SamplingResult(needs_follow_up or end_turn is False, last_message)
raise StreamError("closed before response.completed")
失败、取消与恢复
| 故障或边界 | 已发生的状态 | 对上层的结果 | 能否直接重试 | 恢复动作 |
|---|---|---|---|---|
| Delta 无 active item | 无法确定归属 | 协议/内部状态错误 | 不直接 | 终止或诊断流 |
| Item Done 无可投影 TurnItem | 完整 ResponseItem 仍存在 | 按类型记录或忽略 UI | 不适用 | 保持 History 合同 |
| 多个 Item 后断流 | Done Item 已提交、active 暂态未提交 | Stream 错误 | 是 | 从最新 History 重建 |
| Completed end_turn=false | 响应成功 | needs_follow_up=true | 不是错误重试 | 进入下一采样 |
设计取舍与验证
响应级与 Item 级双层状态机增加了事件类型和暂态管理,却避免把“一个工具调用完成”误当成“整个模型响应完成”。未知事件可忽略利于协议演进;对已知 Delta 缺活动项严格报错,则能尽早发现破坏 UI 归属的序列。
| 可验证契约 | 证据方式 | 预期结果 |
|---|---|---|
| Created→Item→Done→Completed 顺序 | SSE fixture 逐事件断言 | 映射事件顺序不变 |
| 一个响应含多个 Item | items 集成测试 | 每项各自 Started/Completed,采样只结束一次 |
| end_turn=false 继续采样 | 连续响应 fixture | 同 Turn 发起下一模型请求 |
Mini Codex 对照
Mini Codex 应为响应和 Item 分别建状态,不要只用 buffer: str。最小事件联合至少包括 Created、ItemStarted、Delta、ItemDone、Completed 和 Error,并对非法序列写测试。
本节边界
本节只定义事件骨架。5.14 从最常见的 Assistant Message 开始,说明文字为什么既要低延迟转发又要等完整项提交。
评论
登录后即可评论