用 Server-Sent Events 流式传输 LLM 响应

LLM API 使用 Server-Sent Events（SSE） 逐个令牌地流式输出内容。SSE 是一种单向文本协议：服务器发送 text/event-stream 数据，客户端在数据到达时逐步读取。当你设置 stream: true 时，响应不再是一个完整的 JSON 主体，而是一连串小事件——每个事件都是一行携带 JSON 片段的 data:，事件之间以空行分隔，并且（对于 Chat Completions）以一行字面量 data: [DONE] 结束。要在令牌生成时就把它们渲染出来，你需要增量地解析这个流，而不是等待整个响应返回。本文将讲清楚这套线路格式、为什么调用 LLM 时你应该用 fetch 而不是浏览器内置的 EventSource，以及如何用 AbortController 干净利落地取消一个流。

流式传输值得为它付出额外的解析成本，因为它能大幅压缩用户感知到的延迟。用户真正在意的是首个令牌到达的时间（time-to-first-token）；流式回答会在几百毫秒内就开始出现，而不必等到整段内容全部生成完毕。代价是你的客户端现在要自己承担一个解析器和一条取消路径。

精确剖析 SSE 线路格式

SSE 由一小组规则定义。服务器以 Content-Type: text/event-stream 响应，主体是 UTF-8 文本，消息之间以一对换行符（\n\n）分隔。每条消息由一行或多行 field: value 组成。协议只定义了四个字段：

字段	含义
data	负载。同一条消息中连续的 data: 行会以换行符相连拼接在一起。
event	事件类型的名称；会被派发给相匹配的监听器。若省略，则为默认的 message。
id	设置客户端的“最后事件 ID”，重连时通过 Last-Event-ID 头回传给服务器。
retry	重连延迟，单位为毫秒。

以冒号开头的行是注释，会被忽略——服务商用它作为保活心跳，防止空闲代理关闭连接。一个原始的 Chat Completions 流在线路上看起来是这样的：

data: {"id":"chatcmpl-x","choices":[{"delta":{"role":"assistant"}}]}

data: {"id":"chatcmpl-x","choices":[{"delta":{"content":"Hello"}}]}

data: {"id":"chatcmpl-x","choices":[{"delta":{"content":" world"}}]}

data: [DONE]

每个数据块都是一个 chat.completion.chunk 对象；文本位于 choices[0].delta.content 中，你需要累加这些增量（delta）来拼出完整消息。流以结束标记 data: [DONE] 收尾，这个标记不是 JSON——你必须在尝试解析前把它当作字面字符串来检查，否则你的 JSON 解码器会在最后一行抛出异常。

两套事件词汇：Chat Completions 与 Responses

传输层完全相同，但事件因端点而异。Chat Completions 发出 chat.completion.chunk 增量和 [DONE] 结束标记。Responses API 则发出带有明确名称的语义事件——response.created、response.output_text.delta、response.completed 等等——每个事件都直接声明了自己是什么。语义事件更易于路由：你不必去检查数据块的结构，只需针对事件名做分支判断。如果你正在两个端点之间做选择，请参阅 Responses API 与 Chat Completions 对比；如果要迁移现有的流式代码，请参阅 将 Chat Completions 迁移到 Responses API。

要在 Chat Completions 流上获得最终的用量统计，请发送 stream_options: {"include_usage": true}；服务商会追加一个最后的数据块，其 usage 对象里包含令牌计数，你可以将它转发到网关的用量日志中用于计费。

为什么不用 EventSource？

浏览器自带一个 SSE 客户端 EventSource，但出于两个具体原因，它是调用 LLM 时的错误工具。第一，EventSource 只能发起 GET 请求，而 LLM 调用是带 JSON 主体的 POST。第二，EventSource 无法设置自定义请求头，因此你无法发送 Authorization: Bearer ...。这两个限制都是结构性的，不是你能通过配置切换掉的。

正确的做法是用 fetch 配合可读流，自己解析 SSE。这样你就能拥有 POST、请求头，以及对取消的完全控制：

const res = await fetch("https://your-gateway.example.com/v1/chat/completions", {
  method: "POST",
  headers: {
    "Content-Type": "application/json",
    Authorization: "Bearer AVE_GATEWAY_API_KEY",
  },
  body: JSON.stringify({ model: "gpt-5.6", stream: true,
    messages: [{ role: "user", content: "Stream a haiku about latency." }] }),
});

const reader = res.body.getReader();
const decoder = new TextDecoder();
let buffer = "";

while (true) {
  const { value, done } = await reader.read();
  if (done) break;
  // Normalize CRLF so the blank-line separator matches regardless of line endings.
  buffer += decoder.decode(value, { stream: true }).replace(/\r\n/g, "\n");

  const events = buffer.split("\n\n");     // messages end at a blank line
  buffer = events.pop();                    // keep the incomplete trailing chunk

  for (const evt of events) {
    // One event can hold several data: lines; join them per the SSE spec.
    const data = evt
      .split("\n")
      .filter((l) => l.startsWith("data:"))
      .map((l) => l.slice(5).replace(/^ /, ""))
      .join("\n");
    if (!data) continue;
    if (data === "[DONE]") return;          // sentinel is not JSON
    const json = JSON.parse(data);
    const token = json.choices?.[0]?.delta?.content;
    if (token) appendToken(token);          // browser: write to a DOM text node
  }
}

appendToken 是你的渲染出口——在浏览器中，可以把 token 追加到文本节点或 React state；在 Node 中，则可以改为写入 process.stdout。有两个解析细节至关重要，却很容易漏掉。首先把 \r\n 归一化为 \n，因为某些中间层会发出 CRLF，若直接按 \n\n 切分，就会错过 \r\n\r\n 边界。其次，要拼接一个事件中的所有 data: 行，而不是只读取第一行：SSE 规范要求连续的 data: 行之间以换行符连接，只保留一行会破坏多行负载。

缓冲同样关键：TCP 不会尊重消息边界，所以一次 read() 可能只送来半个事件，也可能一次送来好几个事件。要在空行分隔符处切分，并把不完整的尾部结转到下一次读取。跳过这一步是最常见的流式传输 bug——它在测试时能跑通，却会在高负载下损坏数据。

用 AbortController 取消流

用户会放弃请求：他们会离开页面、修改提示词，或者叫停一个失控的生成。他们一旦这么做，你就应当立即停止为令牌计费。AbortController 就是实现这一点的机制。创建一个控制器，把它的 signal 传给 fetch，再调用 abort() 来拆除连接：

const controller = new AbortController();
const res = await fetch(url, { method: "POST", signal: controller.signal, /* ... */ });

// stopButton.onclick:
controller.abort();   // fetch promise rejects with a DOMException named "AbortError"

当你调用 abort() 时，正在进行的 fetch 会以一个 name 为 AbortError 的 DOMException 拒绝，所以你可以通过检查 err.name === "AbortError" 来区分一次主动取消和一次真正的故障。如果你想要一个硬超时而不必自己接线控制器，AbortSignal.timeout(ms) 会返回一个在指定毫秒数后自动中止的信号——超时触发时拒绝的是 TimeoutError 而非 AbortError。及时中止是一种直接的成本控制手段，这与 降低 LLM API 成本 密切相关。

重连及其在 LLM 流上的局限

SSE 内置了重连机制：id 字段和 retry 字段让浏览器能在连接断开后，通过重新发送 Last-Event-ID 来恢复。这对事件推送流很强大，但对 LLM 生成只部分适用，因为恢复的连接无法“撤销”已经产生的令牌计费，而且大多数模型流并不是为从任意偏移量重放而设计的。对于 LLM 流量，请把中断的流当作一个不完整的结果并重新生成——同时配合幂等性协调，以免被重复计费，具体见 LLM API 重试与幂等性。还有一条基础设施层面的提示：SSE 走 HTTP/1.1 时，每个域名最多只能有六个并发连接，所以要从同一个浏览器源发起多个并行生成，就需要 HTTP/2，而网关通常会替你终结 HTTP/2 连接。

通过服务器代理 SSE 流

大多数生产环境并不会让浏览器直接与服务商之间流式通信——它们会通过自己的后端或网关来流转，由后者持有服务商密钥并施加鉴权、限流和日志记录。代理 SSE 有几项很容易搞错的要求，一旦出错就会产生一个“卡住”的流，直到整段响应准备就绪才吐出来。

• 禁用响应缓冲。 反向代理和框架往往默认缓冲响应，这会彻底破坏流式传输。要立即把数据块透传出去，并在相关环节禁用代理缓冲，让任何组件都不会在转发前把主体攒起来。
• 设置正确的请求头。 用 Content-Type: text/event-stream 和 Cache-Control: no-cache 来响应，这样中间件就不会缓存或重塑这个流。
• 转发，而非重新序列化。 尽可能原样透传服务商的数据块；对每个令牌重新解析再重新发出，会增加延迟并招来边界 bug。
• 传播取消。 当客户端断开连接时，也要中止上游请求，否则你会继续为没人会读的令牌付费。这是 AbortController 在服务器端的镜像。

网关是承载这一切的天然场所，因为它本就终结客户端连接并持有凭据。把每个服务商注册为它背后的一个渠道，就能让网关从任一健康的上游进行流式传输，同时向每个客户端呈现一致的 SSE 形态，并像记录非流式请求一样记录流式请求的 token 用量。如果你在同一个端点后面运行多个服务商，那么在它们各自的事件格式之间进行流式转换，正是 OpenAI 兼容的 API 网关 为你完成的工作之一。

常见问题

为什么我的 JSON 解析器会在流的最后一行抛出异常？

因为 Chat Completions 流的最后一行是 data: [DONE]，而 [DONE] 是一个字面结束标记，不是 JSON。在把负载交给 JSON.parse 之前，先检查 [DONE] 字符串并停下来。这是最常见的一种流式解析错误。

我能用浏览器的 EventSource 来做 OpenAI 风格的流式传输吗？

不能。EventSource 只支持 GET 且无法设置 Authorization 头，而 LLM 调用是需要鉴权的 POST 请求。请用 fetch 配合 res.body.getReader()，并自己解析 text/event-stream，正如上面的示例所示。

我该如何停止一个流并停止为令牌付费？

给 fetch 挂上一个 AbortController 信号，并调用 controller.abort()。连接会被拆除，fetch 的 promise 会以 AbortError 拒绝。只有当取消信号一路传播到提供商时，中止才会停止计费；在代理架构中，服务器也必须中止自己的上游调用。因此，要把它接到任何 UI 的“停止”控件和客户端断连事件上，并确保取消能端到端传播。

Chat Completions 流和 Responses 流在线路上看起来一样吗？

它们共享 SSE 传输层，但事件不同。Chat Completions 发送以 [DONE] 结尾的 chat.completion.chunk 增量；Responses API 发送像 response.output_text.delta 和 response.completed 这样的具名语义事件。为其中一个写的解析器需要调整才能用于另一个。

接好一条可取消的流

拿上面那段 fetch + reader 代码，把它指向你的网关并带上 stream: true，随着 token 到达把 choices[0].delta.content 渲染到文本节点。然后加入 AbortController 和停止按钮。中止是一种尽力而为的操作：本地记录的输出 token 数在请求取消后应该下降，但这只能视为信号，不能当作证明——断开连接后，上游可能仍会短暂继续生成或计费；如果最终的上游用量没有到达，本地计数也可能只是估算值。要确认取消确实抵达提供商，请将网关的 stream_status（client_gone 标记）与提供商自己的用量或账单交叉核对，而不要只看本地 token 计数。

参考资料

• MDN: Using server-sent events
• MDN: AbortController
• OpenAI Chat Completions 流式传输参考
• OpenAI Responses 流式事件