结论：智能体流式输出不是”把 stream=true 打开”就完事，真正要处理的是三件事——按 sequence_number 有序拼接增量、把”工具调用中”这类中间状态显式暴露给用户、以及断流后能续跑。只渲染 delta 文本而忽略事件类型和游标，你得到的会是一个偶尔乱序、中断即丢、看不到进度的半成品。
为什么智能体尤其需要流式
普通对话流式只是提前展示文本；智能体的每一次运行里，模型可能在多轮工具调用之间来回，单次响应耗时从几秒到几分钟不等。没有流式时用户面对的是一个空白转圈，无法判断”是在思考、在调工具、还是卡住了”。
流式的第二个价值是可中断：用户看到方向不对可以立刻叫停，省掉整轮 token 。第三个价值是可续跑：拿到游标后即使连接断开，也能从断点继续消费事件。
两种流式粒度
| 粒度 | 消费的东西 | 适合场景 | 复杂度 |
|---|---|---|---|
| 文本增量（delta） | output_text.delta 一类事件 | 聊天界面、纯生成型回答 | 低 |
| 运行事件流（run events） | 工具调用、 handoff 、审批暂停、完成等完整事件 | 智能体控制台、需要显示进度与中间态 | 中 |
做智能体产品，第二种才是应该选的。只消费 delta，你看不到”正在调用天气 API”这种状态，界面就只能干等。
最小实现：消费运行事件流
以 OpenAI Agents SDK 为例，流式运行用的是同一套循环和状态策略，区别只是在运行还没结束时就能消费事件：
import asyncio
from openai.types.responses import ResponseTextDeltaEvent
from agents import Agent, Runner
agent = Agent(name="Planet guide", instructions="Answer with short facts.")
async def main():
stream = Runner.run_streamed(agent, "Give me three short facts about Saturn.")
async for event in stream.stream_events():
if event.type == "raw_response_event" and isinstance(event.data, ResponseTextDeltaEvent):
print(event.data.delta, end="", flush=True)
print(f"\nFinal: {stream.final_output}")
asyncio.run(main())JavaScript 版的形状一致：
import { Agent, run } from "@openai/agents";
const stream = await run(agent, "Give me three short facts about Saturn.", { stream: true });
for await (const event of stream) {
if (event.type === "raw_model_stream_event" && event.data.type === "output_text_delta") {
process.stdout.write(event.data.delta);
}
}
await stream.completed;
console.log("\nFinal:", stream.finalOutput);事件模型的三层结构
把 Responses API 的事件流按层级理解，实现时不容易乱：
- 信封层：描述整个响应对象状态，如
response.created、response.in_progress、response.completed、response.failed。真正的 token 用量只在response.completed里给出。 - 输出项层：
response.output_item.added/.done，一个输出项可以是消息、工具调用或推理。 - 内容块层：
response.content_part.added→response.output_text.delta→response.output_text.done。文本就在这层以增量形式到达。
- 按
output_index+item_id+content_index三级建缓冲区，收到的 delta 追加到对应缓冲区，绝不要直接往一个全局字符串上拼。 - 严格按
sequence_number排序消费。乱序到达在网络层是常态，靠序号纠正比靠到达顺序可靠。 - 把非文本事件（工具调用、 handoff 、审批请求）映射成界面上的显式状态，让用户知道智能体在做什么，而不是对着空屏等。
- 在
response.completed（或failed/incomplete）时落库并结算用量；中途断开则保存游标，用starting_after续消费。
断流与续跑
这是流式实现里最容易被跳过、也最容易在演示时翻车的一环。要点如下：
- 记录游标：每个事件都带
sequence_number，持续更新最后一个值。 - 支持从断点续读：以 Responses API 为例，可用
?stream=true&starting_after=<cursor>重新接入；后台模式下响应会继续跑，重连即可接上。 - 取消是幂等的：重复调用 cancel 只会返回最终状态，可以放心重试。
- 流结束才算结束：务必等
stream.completed再把这次运行当作已定稿，否则你会拿到半截结果。
后台模式 + 流式的组合值得单独说一句：同时设 background=true 与 stream=true，可以立刻开始消费事件，同时保留”客户端掉线后重连”的能力。代价是首 token 延迟比同步模式略高。
中断不是新一轮对话
这是最常犯的设计错误。用户在流式进行中点”停止”，或者运行因为等待人工审批而暂停，都属于暂停，不是新一轮。
区别很实际：如果当成新的一轮，轮次计数、历史记录和服务端的续接 ID 都会错位，恢复后上下文断裂。正确做法是把审批视为”暂停的运行”，从保存的状态恢复继续执行——这一点在 智能体人机协作（HITL） 和 断点续跑 里有更完整的展开。
和结构化输出怎么共存
一个常见误解是”流式了就没法要结构化结果”。其实两者可以并存：界面消费事件流，落库时用最终的结构化对象。做法是先定义 JSON Schema 约束输出（见 智能体结构化输出），流式阶段照常渲染增量文本，收到 response.completed 后拿完整输出做校验；校验失败按重试策略处理，而不是把半截 JSON 丢给用户。
要点是：流式负责体验，结构化负责正确性，两者由 completed 事件衔接。
上线检查清单
- 缓冲区按三级索引隔离，delta 不乱串。
- 按
sequence_number有序消费，能容忍乱序与重复。 - 工具调用、审批暂停有显式 UI 状态。
- 游标持久化，断线可续；取消操作幂等。
completed/failed/incomplete三种终态都有处理分支。- 用量只在
completed结算，中途不估算。
如果你刚开始搭智能体，可以先跑通 OpenAI Agents SDK 入门，再回来补流式这一层。
评论 (0)