流式解析与事件 · 从 SSE 字节到屏幕

流式解析与事件 · 从 SSE 字节到屏幕

模型的回复是流式的（Server-Sent Events）。这一章顺着字节流，看它怎么变成你屏幕上跳动的字，以及路上那些「出错了怎么办」的机关。

L1：把字节变成类型化事件

SamplingClient（client.rs）拿到 HTTP 响应后，跑一条 SSE 解析流水线：

response.bytes_stream()
  → 去掉 UTF-8 BOM
  → .eventsource()  解出一帧帧 SSE
  → .scan(...) 状态机  处理 [DONE]、doom-loop 检查帧、流内错误
  → 反序列化成后端专属的类型化事件
  → .boxed()

其中 .scan 状态机很值得看：[DONE] 结束；doom-loop 的检查帧要吞掉（否则会让类型反序列化失败；这种帧只出现在 Responses 流里）；流内错误转成 StreamError。反序列化还有个宽容处理——遇到 async-openai 建模不了的 xAI 扩展工具（x_search），先剥掉再重试，绝不因为一个不认识的字段整条流崩掉。

L2：归一成 SamplingEvent

L1 的原始事件进 L2 流变换，吐出统一的 SamplingEvent（events.rs）。这套词汇是上层唯一要认的东西：

这些事件流到会话循环，再由 TUI 画进滚动缓冲——就是你看到的实时输出。

一个精妙点：错误保真跨层

这是我最欣赏的设计。L1 的错误是信息丰富的 SamplingError（含 reqwest::Error），但它不能 Clone、不能序列化。而 L2 的事件要能序列化（要过通道、要落日志），只能带一个「有损」的 SamplingErrorInfo。

问题来了：重试决策需要丰富的错误（才能分清是 401 还是 413 还是限流）。grok 的解法是 tee_errors（actor/request_task.rs）——用一个 Arc<Mutex<Option<SamplingError>>> 单元格，在原始流上「三通」一下，把第一个丰富错误截留进单元格，同时不改变流本身。actor 于是能从单元格里取回丰富错误做精确分类，取不到再用 synthesize_from_info 兜底重建。

两套独立的重试预算

出错重试是 agent 稳定性的命门。采样器把重试分成两套互不相干的预算：

• 传输/空响应预算(默认 DEFAULT_MAX_RETRIES=15，可用 GROK_MAX_RETRIES 覆盖；指数退避从 2 秒起、封顶 30 秒，带 ±20% 抖动)：网络抖动、5xx、空回复走这套。
• doom-loop 预算(DoomLoopRecoveryPolicy 的 max_retries，默认 2 次，可配 0～5)：模型陷入复读机式失控走这套(详见打断与防失控)。

两套分开的意义：一次 doom 重采样不该消耗传输预算，反之亦然。核心是 classify_error(retry.rs)把错误映射成 RetryDecision：

还有双 idle 计时器防「连接活着但不出字」(默认 300 秒)：一个管「流干脆不 yield 了」，一个管「一直发空 keepalive 但没实质进展」，任一超时都报 IdleTimeout。采样器这一层放弃了，也不等于这一轮就失败：会话层对流停滞、传输错误这类瞬时故障还有一道 turn 级重投(见主循环)。

和《从 LLM 到 Coding Agent》对照

对照那本的 streaming 与 error-recovery：最小实现里，流式是 for await chunk，错误恢复是 try/catch + retry。grok 的增量全在「真实网络很脏」这件事上——BOM、宽容反序列化、错误保真、分类重试、连接池自愈、双超时。这些不改变主线逻辑，却决定了产品在烂网络下会不会崩。

上一篇：三种 API 流 · 下一篇：会话状态 · chat-state