Zero 架构 · 第十一课

抽流:一条 StreamEvent channel,同时喂给「画面」和「循环」

第九课给出了 provider 的出向词汇 —— 一条 StreamEvent 的 channel(9 种事件)。第六课讲了 TUI 怎么把这些事件 实时画上屏。但第一课的循环最终要的不是「流」,而是一份攒齐的结果:这一轮模型说了什么文本、 发起了哪些 tool call、用了多少 token。CollectStreamWithOptions helpers.go:90 就是这台「抽干机」:它一边把流实时转发给回调(喂画面),一边把它攒成一个 CollectedStream(喂循环)。

对齐 mission:流式和非流式是同一份数据的两副面孔。一个 coding agent 既要让用户看着字一个个蹦出来(流), 又要在这一轮结束后拿到完整的tool call 列表去执行(非流)。这一课就是把「流」优雅收束成「一次结果」的那个接缝 —— 一次遍历,两个用途。

核心形状:一个 for-select 抽干 channel,四条出口都走 finish()

函数主体就是一个 for { select {...} } helpers.go:102,把 channel 抽到干。有四种方式结束, 全部经由同一个 finish() 闭包收尾:

finish() helpers.go:94 做两件事:collector.flush() 把攒的 tool call 和文本落定,并在见过 usage 时补发一次 OnUsage。关键在「单一收尾点」: 无论从哪条出口走,flush 恰好被调一次 —— 这是文本只被物化一次、tool call 只被输出一次的保证。

两副面孔:同一个 switch 里,攒 + 转发

每个事件在 switch event.Type helpers.go:125 里同时做两件事 —— 攒进结果 和 转发给回调:

case StreamEventText:
    collector.text.WriteString(event.Content)   // 攒(给循环)
    if options.OnText != nil { options.OnText(...) }  // 转发(给画面)
case StreamEventToolCallStart:
    collector.start(id, name, signature)         // 攒
    if options.OnToolCallStart != nil { ... }    // 转发
case StreamEventToolCallDelta: ...  // 攒参数片段 + 转发
case StreamEventUsage:  collected.Usage = mergeUsageSnapshot(...)
case StreamEventToolCallDropped: collected.DroppedToolCalls++

这正是第六课那些回调(OnText/OnToolCallStart/OnToolCallDelta/OnUsage)的来源: CollectOptions helpers.go:34 就是 TUI 挂进来那组闭包。不需要回调的调用方 (会话标题、recap、压缩)传空 options,抽流就退化成纯粹的「攒」。一个函数,流式和非流式两用。

硬骨头:toolCallCollector —— 为什么「攒」而不是「end 即发」

tool call 是流式到达的:start(给 id+name)、若干 delta(参数片段一点点拼)、end。 最容易想到的是「end 时就把这个 call 发出去」。Zero 偏不 —— end 什么都不发 helpers.go:278,只有 flush 在最后按 start 顺序一次性全发 helpers.go:319。

空 ID 的泥潭:合成 key + delta 先到的收养

有的后端不给 tool call 分配 ID(id=="")。天真地用 ID 当 map key,两个并发的空 ID call 就会合并成一个。 collector 用一套机制把它们分开:

• 空 ID → 合成 key:每个空 ID 的 start 都造一个 \x00synthetic-N 的独立 key,压进一个栈 helpers.go:238;delta/end 用「栈顶那个最近打开的」来路由 helpers.go:294。
• delta 先于 start 到达:一个空 ID 的 delta 若在任何 start 之前就来了,就地开一个合成 call 并记下 pendingEmptyDelta helpers.go:266;紧接着的空 ID start 收养这个 call helpers.go:234,而不是另开一个 —— 于是先到的参数片段不会变成「无主孤儿」。
• 无名 call 被丢弃:flush 时 Name=="" 的 call 不进 ToolCalls,而是计入 DroppedToolCalls helpers.go:326 —— agent 绝不派发一个空工具名。

两处小而正确的决定

• 文本用 strings.Builder 而非 += helpers.go:130:collected.Text += chunk 每次都重新分配整个字符串, 长响应下是 O(n²)。攒进 Builder、flush 时物化一次。
• Usage 逐字段取最新非零:mergeUsageSnapshot helpers.go:165 合并多次 usage 快照(有些 provider 分多次报), 每个字段保留最新的非零值,再走第九课的 NormalizeUsage 校验不变量。

动手回忆

接下来该读的一手源码

less +90  ../zero/internal/zeroruntime/helpers.go   # CollectStreamWithOptions:for-select 抽干 + 四条出口
less +94  ../zero/internal/zeroruntime/helpers.go   # finish():单一收尾点,flush 只调一次
less +204 ../zero/internal/zeroruntime/helpers.go   # toolCallCollector:攒调用、按 start 顺序 flush
less +229 ../zero/internal/zeroruntime/helpers.go   # start/delta/end:空 ID 合成 key + delta 收养
less +165 ../zero/internal/zeroruntime/helpers.go   # mergeUsageSnapshot:逐字段取最新非零

读的顺序:先看主 for-select 的四条出口都汇到 finish() —— 理解「单一收尾点」。再看 switch 里每个 case 「攒 + 转发」的对称。最后啃 toolCallCollector 的空 ID 处理 —— 合成 key、delta 先到的收养、无名丢弃, 这是全函数唯一需要多读两遍的地方,也是把「乱序、缺 ID、并发」的真实流式收得干净的功夫所在。