Agent 实现方法论 From Scratch

流式输出：手写 SSE 客户端

流式输出：手写 SSE 客户端

第 02 章我们学会了用 fetch 裸调 LLM，但拿到的是一个「一次性 JSON」。这一章我们让响应**流式**地到达：不等模型把整段话想完再一次性发给你，它每生成一个字，服务器就吐一个「字」，客户端边收边显示。你在 ChatGPT 里看到的打字机效果，也是「能用」和「好用」的分水岭。

本章目标

读完本章并做完配套练习后，你应该能够：

开场反例：等 30 秒，还是一个字一个字地出来？

想象同样的 30 秒生成过程，两种体验：

体验 A（一次性）： 你问模型一个问题，屏幕上什么都没有。你盯着空白的终端，5 秒、10 秒、20 秒……心里开始打鼓：它是不是卡了？我的 API key 是不是没带对？到了第 30 秒，一大段文本「啪」地整块砸在屏幕上。

体验 B（流式）： 你按下回车，1 秒内屏幕上就出现了第一个字，然后一个字一个字地往外蹦。你看着它「思考」、逐词输出，第 30 秒文本打完收工。

两次生成的都是同一段文字，花的都是 30 秒。但体验 B 让你感觉「它活着、它在回应我」，体验 A 让你焦虑。首字节延迟（TTFB）决定了用户对「卡不卡」的第一印象，而流式把「总延迟」转化为「渐进可见的进度」。

这是用户侧的动机。而开发者侧的动机更直接：如果你要做一个 Agent，工具调用、多步推理、审批交互全都依赖流式的中间态（第 13、15 章你会看到整套事件流协议）。现在不把流式吃透，后面寸步难行。

还有一层容易被忽略的经济账：长任务里，用户的耐心比模型的速度更贵。哪怕你改变不了「模型总共要算 30 秒」这个事实，流式也能让你在第二秒就告诉用户「进展如何」，而不是让他在第 29 秒关掉页面。对需要跑好几步工具调用的 Agent 来说，每一个中间结果都是一次「我还活着、正在干活」的广播——没有它，用户只会看到一片死寂然后怀疑程序挂了。

那么，直接给 fetch 加一个 stream: true 就完事了吗？

const res = await fetch(ENDPOINT, {
  method: 'POST',
  body: JSON.stringify({ model: 'gpt-4o', messages, stream: true }),
});
const data = await res.json(); // ❌ 直接炸掉
console.log(data.choices[0].message.content);

开了 stream: true 之后，服务端不再返回一个完整的 JSON，而是一段连续推送的文本流。如果你不会解析它，只会干瞪眼：打印 res 的 body，你会看到一堆这样的东西：

data: {"choices":[{"delta":{"content":"你"}}]}

data: {"choices":[{"delta":{"content":"好"}}]}

data: [DONE]

把这些直接喂给 JSON.parse 必然报错。它既不是 JSON，也不是平文本，而是一个有自己格式的协议，这就是 SSE（Server-Sent Events）。

SSE 协议：text/event-stream 解剖

SSE（Server-Sent Events，服务器推送事件）是一种基于 HTTP 的、单向的服务器→客户端推送协议。服务器把响应的 content-type 设为 text/event-stream，然后一条接一条地把事件推过来；连接保持打开，直到服务器主动关闭或客户端断开。

每个事件叫一个 frame，格式很简单。若干以 键: 值 形式书写的行，用空行（\n\n）分隔两个事件：

data: 第一行负载
data: 第二行负载

event: named-event
id: 42
data: 带名称和序号的事件

: 这是注释行，客户端忽略

四个常用字段：

字段含义
data:事件的负载。可以多行，多行会被拼接（中间换行）。我们最常用的就是它
event:事件类型（默认 message）。OpenAI 的流没用它，只靠 data
id:事件序号，配合浏览器自动断线重连的 Last-Event-ID 机制
:注释，客户端直接忽略（常用于心跳保活）

注意一个关键点：连接是保持打开的。服务端发完一个事件不会就此结束响应，会继续推下一个；HTTP 层靠 分块传输编码（chunked transfer） 把整段推送打包进同一个响应。对客户端来说，fetch 拿到的是一个「能持续读、不知道什么时候才算完」的 ReadableStream——下一节要啃的就是它。也正因如此，SSE 是单向的：只有服务器往客户端推，客户端要发指令就得另开请求（比如发个「停止」走另一个接口）。

而 OpenAI 兼容的流式接口，全部信息都塞在 data: 行里，负载本身又是一段 JSON（下一节解剖）。

你可能会问：浏览器不是原生支持 SSE 吗？EventSource 一个对象就搞定，为什么要手写？

const es = new EventSource('/v1/chat/completions'); // ⚠️ 只支持 GET

问题就在这：EventSource 只能用 GET，且无法自定义请求头。而 LLM API 是 POST + Authorization: Bearer <key>。所以浏览器原生的 EventSource 在 LLM 场景下基本用不上，只能自己用 fetch 拉一个 POST 的流，然后手写一个解析器。这正好也是本教程的风格：把「框架魔法」拆开，自己组装。

ReadableStream 逐 chunk 解析

用 fetch 发起流式请求后，响应体是一个 ReadableStream。它就像一根水管，数据分成一块一块（chunk）流过来，不会一次给完。客户端的工作是：

  1. res.body.getReader() 拿到读取器；
  2. while 循环里 await reader.read()，每次得到 { value, done }；
  3. 把字节解码成字符串，攒出完整的 SSE 事件。
const reader = res.body.getReader();
const decoder = new TextDecoder(); // 流式模式：{ stream: true }

let buffer = ''; // ← 半行缓冲，本章最重要的变量
while (true) {
  const { value, done } = await reader.read();
  if (done) break;
  buffer += decoder.decode(value, { stream: true });

  // 空行分帧：只要 buffer 里出现 '\n\n'，就切出一个完整事件
  let sep = buffer.indexOf('\n\n');
  while (sep !== -1) {
    const frame = buffer.slice(0, sep);
    buffer = buffer.slice(sep + 2); // 剩下的部分留回 buffer
    handleFrame(frame);
    sep = buffer.indexOf('\n\n');
  }
}

这段代码的核心是 buffer 变量。为什么必须有它？

最大的坑：chunk 边界不对齐行边界。

网络从不保证「一个 chunk = 一个完整事件」。你请求服务端吐 10 个字，它可能一次发来 data: {"choices":[{"delta":{"conten，下一个 TCP 包里才是 t":"你"}}]}\n\n——一个 JSON 行被切成了两半。你如果天真地认为每次 read() 拿到的都是一整行，就会把 conten 当成一个事件去 JSON.parse，直接报错。

解决之道就是攒着：不管 chunk 从哪切，全部先拼进 buffer，只在看到 \n\n（事件结束的标志）时才切出来处理；切完之后剩下的残渣继续留在 buffer 里等下一个 chunk 来补。这套做法，就是「半行 buffering」。

再补一个细节：decoder.decode(value, { stream: true }) 的 stream 模式是干嘛的？TextDecoder 如果不开 stream 模式，遇到「一个多字节字符恰好被切成两半」的情况，会把它解码成 \uFFFD（乱码豆腐块）。开了 stream 模式，它会把没凑齐的字节先缓存起来，等下一个 chunk 到齐再输出完整的字符。所以 stream 模式解决的是「半个字符」问题，buffer 解决的是「半个事件」问题，两者都要。

delta 累积：从 chunk 拼出完整文本

那么，data: 行里的 JSON 长什么样？OpenAI 风格的流式响应，每个事件是一个 chat.completion.chunk 对象，字段和普通响应很像，但 choices[0].message 变成了 choices[0].delta，每一帧只携带「增量」：

data: {"id":"chatcmpl-8x4...","object":"chat.completion.chunk","created":1723456789,"model":"gpt-4o","choices":[{"index":0,"delta":{"role":"assistant"},"finish_reason":null}]}

data: {"choices":[{"index":0,"delta":{"content":"你"},"finish_reason":null}]}

data: {"choices":[{"index":0,"delta":{"content":"好"},"finish_reason":null}]}

data: {"choices":[{"index":0,"delta":{"content":"世界"},"finish_reason":null}]}

data: {"choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}

data: {"usage":{"prompt_tokens":24,"completion_tokens":8,"total_tokens":32}}

data: [DONE]

逐帧解读：

完整文本 = 累积所有 delta.content。 客户端要维护一个累积变量，每帧把 delta.content 追加进去，同时（可选）通过回调把它交给 UI 实时渲染：

let fullText = '';
function handleFrame(frame: string) {
  for (const line of frame.split('\n')) {
    const trimmed = line.trim();
    if (!trimmed.startsWith('data:')) continue;     // 跳过 event:/id:/注释行
    const payload = trimmed.slice(5).trim();
    if (payload === '[DONE]') return;               // 哨兵：结束
    const data = JSON.parse(payload);
    const delta = data.choices?.[0]?.delta?.content;
    if (typeof delta === 'string') {
      fullText += delta;
      onDelta(delta);                                // 逐字交给 UI
    }
  }
}

注意三处防守：只认 data: 行；先判断 [DONE] 再 JSON.parse；delta.content 必须是字符串才处理（role 帧、finish_reason 帧没有 content，直接跳过）。拼完之后，fullText 就是与普通响应 choices[0].message.content 完全等价的完整文本。流式只是传输方式变了，语义没变。

把这个过程在脑子里走一遍：收到 "你" → fullText 变成 "你"，UI 打出「你」；收到 "好" → fullText 变成 "你好"；收到 "世界" → fullText 变成 "你好世界"。你看，delta.content 是一小块一小块内容，fullText 是它们的线性拼接，顺序不能乱，谁先来谁在前。所以 onDelta 回调必须在累加之后立刻调用：UI 渲染的时序必须和文本的生成时序一致，否则会出现「先显示后半个词」的诡异现象。

另外两个细节值得留意。其一，role 帧虽然没有 content，但有时携带的信息有用，比如有的模型允许在流式响应中途修改 role，或后续章节要做工具调用时，delta.tool_calls 增量也是同一套机制（第 05 章）。其二，usage 帧在 [DONE] 之前出现，且不是每个兼容实现都会发，所以解析时把它当可选信息，拿到就记账、没拿到也别崩。

终止与中止：三种「结束」的区别

一个流式请求可能以三种方式结束，别把它们混为一谈：

1. 正常结束（[DONE] 哨兵）。 模型把话说完，服务端发来 data: [DONE] 然后关闭连接。客户端看到哨兵就该停止解析、收工。有的实现会顺手把 usage 帧也收集起来用于成本核算（第 18 章）。

2. 客户端主动中止（AbortController）。 用户在 UI 上点了「停止生成」，或请求超时、或高优先级任务抢断了当前请求。这是客户端主动发起的取消。标准姿势：

const controller = new AbortController();
// 给 fetch 传 signal
const res = await fetch(ENDPOINT, { ..., signal: controller.signal });

// 某个时机（用户点停止 / 收到第 N 个 delta / 超时）
controller.abort();

对「友好中止」而言，拿到部分文本后应该 resolve 当前已累积的文本（不要抛错），并且 reader.cancel() 停止继续读取连接：

let fullText = '';
const onAbort = () => { reader.cancel(); resolve(fullText); };
signal.addEventListener('abort', onAbort, { once: true });

3. 连接中断（传输层失败）。 网络断了、服务器崩了。这不是 abort——reader.read() 会以 reject（抛错） 的形式告诉你「读不出来了」，你拿不到优雅的部分文本。判断口诀：abort 是你主动喊停、read 会安静返回；中断是客观故障、read 会抛错。前者要「温柔收尾」，后者要「妥善报错 + 考虑重试」。

常见坑

坑一：忘记半行 buffer，一个 chunk 当一条事件解析。 这是新手最常见也最容易困惑的错：单测/单机跑得好好的，一上真实网络就间歇性崩。因为本地 mock 有时恰好一个 chunk 就是完整一行，掩盖了问题。记住：只要不是自己构造的流，永远假设 chunk 会切在行中间，一律进 buffer 再分帧。

坑二：把 [DONE] 当 JSON 解析。 看到 data: [DONE] 直接 JSON.parse('[DONE]')，SyntaxError 当场崩溃。[DONE] 不是 JSON，是哨兵文本，必须先判断再解析。

坑三：abort 了但忘记 reader.cancel()，连接泄漏。 只 controller.abort() 让 fetch 层面的信号中止，却不处理已经打开的 body 流，底层连接可能不会立刻释放。中止时要记得 reader.cancel()，把水管真正关掉。

坑四：把 abort 和中途报错混为一谈。 abort 是「用户主动喊停」，不该以异常形式冒泡打扰 UI；网络中断是客观错误，该 rethrow 交给上层重试。区分信号来源，别一刀切地 catch 掉所有东西。

小结

下一章（ch04）我们把目光从「协议细节」拉回「工程抽象」：OpenAI 和 Anthropic 的接口长得不一样，如何抽一个最小 Provider 接口同时兼容它们。那时你会发现，流式（包括它的中断与错误）是 Provider 接口里最难啃的一块。

延伸阅读

完成阅读，去做练习 →