pi-ai（二）：流式事件架构

TL;DR：pi-ai 的每次模型调用都同步返回一条 AssistantMessageEventStream：适配器在后台异步任务里读厂商的 SSE/WebSocket，原地修改一个 output 消息，并把变化翻译成统一事件 start → text/thinking/toolcall 的 start/delta/end → done | error 推进去；调用方用 for await 拉取增量，用 result() 拿最终权威消息。底座 EventStream 是一个"push 生产者 + 异步迭代消费者"的队列，终止事件同时兑现 result promise。核心约定是错误是事件而不是异常：网络错误、认证失败、用户中止都变成带部分内容和 usage 的 error 事件，所以成功、失败、取消只有一条处理路径。上层 pi-agent-core 把这些事件映射成 message_start/update/end 广播给 TUI 和会话持久化。

本篇讲 pi-ai 的流式部分，整体定位和请求主链路见 pi-ai（一）：统一模型调用层。

为什么要流式，为什么要自己的事件协议

不用流式的话，调用方必须等服务端把整段响应生成完才能拿到数据。对 coding agent 来说这不可接受：一次回复可能有几十秒的思考、几百行的文件写入，用户需要实时看到推理和输出，需要在参数还没传完时就看到"正在写 src/index.ts"，还需要随时按 Esc 中止并保留已经生成的内容。

流式本身不难，难在统一。从底向上看有三层：

第三层：自己的事件流（pi-ai 的 AssistantMessageEventStream）  ← pi、Claude Code 这类 Agent
第二层：厂商 SDK 的流式接口（openai、@anthropic-ai/sdk）        ← 大多数项目
第一层：裸 HTTP 流式（fetch + ReadableStream 逐行解析 SSE）      ← 追求极致控制的少数项目

SDK 已经处理了请求格式化、SSE 逐行解析、重试和认证头，但每家 SDK 吐出的事件形状都不一样：Anthropic 是 content_block_start/delta/stop，OpenAI Responses 是一串 response.* 事件，Chat Completions 是不断追加 choices[0].delta 的 chunk，Gemini 每个 chunk 带完整的 parts。pi-ai 在 SDK 之上再加一层，把这些全部统一成一种事件，同时统一 abort 信号的传递、部分结果的保存、成本计算，以及 thinking 和工具参数的增量解析。这样上层的 Agent 循环和 UI 只需要针对一种事件协议写一次。

统一事件协议

一条流的事件顺序固定为：

start → (text_* | thinking_* | toolcall_*)* → done
start → (更新事件)* → error            生成开始后失败
error                                    请求准备阶段就失败（认证、模块加载），没有 start

更新事件按内容块分三组，每组都是 *_start、*_delta、*_end 三段：文本块的 delta 是新增文字，text_end 给出完整文本；thinking 块同理（被厂商加密的 redacted thinking 可能在 start 时就完整，没有 delta）；工具调用块的 delta 是参数 JSON 的一个片段，toolcall_end 给出完整的 ToolCall。终止事件只有两种：done 的 reason 是 stop | length | toolUse | deferred，带最终消息；error 的 reason 是 error | aborted，带一条同样完整的 AssistantMessage，里面有已生成的部分内容、errorMessage 和已经产生的 usage。

有两个字段决定了消费者怎么写：

contentIndex 指明事件属于最终消息 content 数组的哪个块。不同块的事件不保证连续：厂商可能在同一个 chunk 里同时给出文本和工具参数，pi 也会交错地发出 text_start, text_delta, toolcall_start, text_delta, toolcall_delta。消费者必须按 contentIndex 归属事件，不能假设一个块的三段事件不会被打断。

partial 是"到目前为止的响应"，每个非终止事件都带。注意它是共享的活对象，不是事件发生时的快照：适配器一直在原地修改同一个 output，队列里还没被消费的旧事件看到的 partial 也已经是更新后的样子。所以处理事件时可以读它，但不能把它当历史状态保存。这个设计让 TUI 每次只需要"拿 partial 重新渲染"，不用自己累积状态，代价是消费者要清楚它的可变语义。

EventStream：push 生产者与 async iterator 消费者

生产者（网络回调）和消费者（UI 渲染、Agent 循环）的节奏不同：网络可能一次来三个 token，也可能半天不来；消费者可能很慢，也可能早就在等。需要一个中间缓冲协调两边。EventStream<T, R> 只有一百多行，核心是两个队列：

push(event)         生产者调用：有消费者在等就直接交给它，否则放进事件队列；
                    若是终止事件，标记 done 并兑现 result promise
for await (...)     消费者：队列里有事件就取，已结束就退出，否则挂一个 resolver 到等待队列
end(result?)        结束流，唤醒所有还在等的消费者
result()            返回 Promise<R>，在终止事件到来时兑现

"何为终止事件、如何从中取结果"由构造参数决定。AssistantMessageEventStream 把 done 和 error 定义为终止事件，结果分别取 event.message 和 event.error。它在此之上只多做一件事（1.1 起）：在终止时用单调时钟给最终消息写入 durationMs，所以所有适配器、包括直接调用 API 实现的路径都有统一的耗时。1.1 起 stream 函数也必须返回这个类型本身，手写的 EventStream 子类不再能冒充。

这个结构带来两个好处。一是流式和非流式统一：不关心过程的调用方直接 await s.result()，complete() 就是这么实现的，底层只维护一套机制。二是失败也有结果：因为 error 也是终止事件，result() 永远不会 reject，总能拿到一条消息，成功的或失败的。

**为什么是事件流而不是 for 循环加回调。**一般的简单 Agent 写法是同步的：

for chunk in stream:                 # 边收边处理
    if delta.content: on_token(...)  # 回调里打印，处理和消费绑死在一起
    if delta.tool_calls: args += ... # 累积参数，等流结束再执行工具

这种写法事件只有两种、消费者只有一个终端打印，够用。但 pi 的事件类型多得多，消费者也多：UI 关心 text/thinking 增量，会话持久化关心最终消息，扩展关心工具调用，遥测关心首包和块大小。把这些都塞进一个 on_token 回调会让生产者认识所有消费者，任何一个出错都会连累其他的。事件流让生产者（适配器）只管 push，完全不知道谁在消费；消费者也不知道事件来自哪家厂商。测试时不需要真实 LLM，手动 push 一串事件（或用内置的 faux provider）就能测 UI 和 Agent 循环。另外 TypeScript 的 I/O 本来就是强制异步的（不像 Python 里异步是可选的），用 async iterator 表达"会陆续到达的数据"是最自然的写法。

要澄清一点：EventStream 本身是单消费者的。它是一个队列，每个事件只会交给一个等待者，两个地方同时 for await 同一条流会把事件分走。真正的"多个消费者各自独立订阅"发生在上一层：pi-agent-core 是这条流唯一的消费者，它把事件翻译成 Agent 事件后，再通过 Agent.subscribe() 广播给所有监听者。

适配器如何把 SSE/WebSocket 归一

所有适配器的骨架几乎一样，以 anthropic-messages.ts 为例：

stream(model, transcript, options):
  stream = new AssistantMessageEventStream()
  启动后台 async 任务，然后立即 return stream
    output = { role: "assistant", content: [], usage: 0, stopReason: "pending", ... }   ← 唯一的"工作区"
    try:
      转换消息（transformMessages）→ 构造 payload（onPayload 可改）→ 发请求（带 signal）
      onResponse → push(start)
      for await 每个厂商事件:
        await onProviderStreamEvent?.(event)
        按厂商的块 index 找到对应的 content 块 → 原地修改 output → push 对应的 *_start/_delta/_end
        遇到 usage 就更新 output.usage 并 calculateCost()
      检查：已中止？没有 stop reason？ → throw
      push(done) ; end()
    catch(error):
      output.stopReason = signal.aborted ? "aborted" : "error"
      output.errorMessage = error.message
      push(error) ; end()

一句话概括：在后台任务里只维护一个 output 对象并不断原地修改，通过 push 把增量实时推给消费者；成功时 push done，失败或中止时 push 带着部分内容的 error。

各适配器的差异只集中在几处：消息和工具 schema 怎么转换、厂商的块编号怎么映射到 contentIndex、thinking 和签名怎么表示、usage/缓存/价格怎么计算。几个值得一提的点：

• 厂商的块编号和流式中间状态（工具参数原始文本、块 index 等）放在 content 块之外的 Map 里，不挂在 content 上。1.5 之前这些字段挂在块上，会漏进 partial 和失败流的 transcript，现在不会了。
• start 在拿到 HTTP 响应之后才 push，所以连接阶段的失败表现为"只有 error 没有 start"。Anthropic 在 message_start 时就记下输入 token，这样即使很早被中止也有准确的输入 usage。
• 传输方式不限于 SSE。OpenAI Codex 适配器可以走 WebSocket（transport 选项），归一化逻辑一样；Gemini 不流式传输工具参数，会一次性给出一个带完整参数的 toolcall_delta。
• pi-messages 是 pi 自己的线协议：把 { model, context, options } 一次 POST 给后端，后端用 SSE 回传已经序列化的 pi 事件。Radius 网关就说这种协议，所以统一事件协议本身也可以跨网络传输。
• onProviderStreamEvent 在归一化之前看到解析后的原始事件，是读取 pi 没保留的厂商字段（如 OpenRouter 的元数据）的官方出口；它按顺序 await，抛错会让请求失败。

工具参数的增量解析

工具参数是一段逐步到达的 JSON 文本，比如 {"path": "src/in 然后 dex.ts", "content": "...。为了让 UI 能在 toolcall_delta 时就显示文件路径，每个 delta 后 partial 里的 arguments 都应该是"到目前为止能解析出的对象"。

早期做法是每来一个 delta 就把累积的全部文本重新解析一遍，复杂度随参数大小平方增长：一次 90KB 的 write 在 Anthropic 上峰值约 140MB 堆，500KB 的要跑好几分钟。1.5 引入 StreamingJsonParser 和其上的 ToolCallArgumentsStream，所有 Provider 统一使用：每个片段只读一次，总代价与文本长度线性相关；它是容错的（未闭合的字符串显示已读部分，未开始的成员省略，非法转义保留为字面量，从不抛错）；每次返回的快照之后不再改变，已完成的子结构在快照间共享。append() 还会返回这一片带来的变化（某路径被设值、某字符串被追加），供自己维护副本的写入方使用。

增量解析只服务于预览。toolcall_end 时会对完整文本做一次严格解析：如果最终文本不是合法 JSON，ToolCall 带上 argumentsError，校验阶段直接失败并把原始文本回给模型，而不是拿解析了一半的字段去执行。toolcall_end 的参数完整但未经 schema 校验，校验由上层用 validateToolCall() 完成。

中止、错误与 usage 语义

错误是事件，不是异常。stream 函数的契约是：必须返回事件流；一旦被调用，请求、模型和运行时的失败都编码进流里，而不是抛出；错误终止必须产出一条 stopReason 为 error 或 aborted、带 errorMessage 的 AssistantMessage。lazyStream() 把同一契约延伸到了请求准备阶段：认证失败、Provider 不存在、模块加载失败也都变成 error 事件。唯一的例外是直接调用 API 实现的 streamSimple() 而又缺少认证时，会同步抛错，因为那条路径没有 Models 帮你解析认证。

好处是调用方只有一套处理逻辑，不需要 try-catch 加事件处理两套；失败消息和成功消息结构一样，可以照常写进会话历史。

中止的完整链路是这样的：用户在 TUI 按 Esc → Agent.abort() 调用本次运行的 AbortController.abort() → 同一个 signal 已经沿着 agent-loop 传进了 streamFn 和适配器 → 适配器把 signal 交给 SDK/fetch，底层连接被取消；同时流式循环每处理一个厂商事件都会检查 signal.aborted → 抛出异常进入 catch → 因为 signal 已中止，stopReason 记为 aborted → push error 事件 → Agent 循环收到后结束这一轮。

部分结果是怎么保留的：从流创建到结束只有一个 output 对象，所有修改都是原地进行的，所以 catch 块里直接把这个 output 标上错误状态推出去，已经收到的文本、thinking、工具调用和 usage 都还在。消费者在循环里可以通过 event.partial 实时看，流结束后通过 result() 拿到含部分内容的最终消息。中止的消息可以放回上下文，用户说"继续"后接着聊；不过在重放历史时，error/aborted 的 assistant 消息会被 transformMessages() 跳过，避免把不完整的推理或调用发给厂商。

usage 语义：适配器在流中随时更新 usage 并用模型的价格表计算成本，所以失败和中止的消息也带有已产生的 token 和费用。某个字段为 0 可能只是厂商没提供这一类计量，不代表确定没有。1.5 新增 cost.totalFromProvider 区分"厂商报告的实际账单"（如 OpenRouter）和"按目录价格的估算"。

上层如何消费

pi-agent-core 是事件流的直接消费者。它的 streamAssistantResponse() 先把 Agent 消息转换成 LLM 消息并 normalizeContext()，再调用 streamFn（coding-agent 注入的实现内部就是 modelRuntime.streamSimple()），然后逐个翻译事件：start 时把 partial 推进上下文并发出 message_start；每个 text/thinking/toolcall 事件都替换上下文里的那条 partial 并发出 message_update，原始的 pi-ai 事件附在 assistantMessageEvent 字段里；done/error 时取 result() 作为权威最终消息替换进上下文，发出 message_end。注意当前实现不会边流式边执行工具：工具执行在整条 assistant 流收敛成最终消息之后才开始，tool_execution_* 事件也在那之后发出。详见 pi-agent-core。

广播给多个消费者发生在 Agent 这一层：Agent.subscribe() 注册的每个监听者都会收到上述事件，并且会被依次 await。在 Pi 里，交互模式的 TUI 在 message_update 时用最新的 assistant 消息重新渲染流式组件，并在看到新的 toolCall 块时提前创建工具执行组件，所以参数还在流入时界面上已经有了工具卡片；AgentSession 只在 message_end 时调用 sessionManager.appendMessage() 写入会话文件。也就是说，流式中间态只活在内存和界面上，落盘的永远是最终消息。

streamProxy 展示了这套协议的另一个好处。有些应用不想把厂商 key 放在客户端，而是让服务端代理 LLM 调用。pi-agent-core 提供的 streamProxy() 就是一个可以作为 streamFn 传给 Agent 的流函数：它把模型、上下文和可序列化的选项 POST 给代理服务器，服务器以 SSE 回传事件，但为了省带宽去掉了每个事件里的 partial；客户端拿着这些精简事件自己重建 partial（工具参数同样用 ToolCallArgumentsStream 增量解析），再推进一条本地的 AssistantMessageEventStream。对 Agent 循环来说，它和直连厂商没有任何区别。

取舍

得到的东西：成功、失败、取消只有一条处理路径；流式和非流式共用一套机制；生产者和消费者完全解耦，换厂商、加消费者、做代理、写测试都不需要改另一侧；partial 让渲染逻辑不用自己累积状态；事件和消息都可序列化，便于跨网络传输（pi-messages、streamProxy）和会话持久化。

付出的代价：消费事件流比消费一个 Promise<string> 复杂，消费者要理解 start/delta/end 的状态、事件交错和 partial 的可变语义；"错误不抛出"意味着忘记检查 stopReason 的调用方会把失败当成正常的空回复；EventStream 是单消费者的，需要多路分发时必须在上层自己做广播。

常见追问

为什么 partial 不做成不可变快照？ 每个 delta 都深拷贝一次整条消息，代价随消息长度增长，长回复下非常浪费；而大多数消费者只需要"当下最新的状态"。用共享活对象换性能，再在文档里写清"不要保存 partial"，是有意的取舍。

为什么错误不直接 reject result()？ 失败的回复同样是对话历史的一部分：它带着部分内容和已经计费的 usage，Agent 需要把它记进上下文和会话、显示给用户、决定是否重试。如果是 reject，这些信息就只能从异常对象里绕回来，调用方也要写两套逻辑。

事件流里有背压吗？ 没有。push() 从不阻塞，消费者慢时事件堆在内存队列里。对 LLM 响应这种规模（最多几 MB 文本）这不是问题，换来的是生产者实现的简单。

为什么不边流式边执行工具？ 一轮回复里可能有多个工具调用，参数要完整并通过校验才能安全执行，而且模型的 stopReason 要到最后才知道。先收敛成最终消息再执行，语义更清楚，也让审批、并行执行和中止处理都基于确定的数据。

源码入口（v1.5）：packages/ai/src/utils/event-stream.ts、packages/ai/src/api/lazy.ts、packages/ai/src/api/anthropic-messages.ts、packages/ai/src/utils/streaming-json.ts、packages/agent/src/agent-loop.ts、packages/agent/src/proxy.ts