SSE 流式输出实战：给 LLM 应用做打字机效果

大模型的回答是逐 token 生成的，但非流式接口要等整个回答生成完才一次性返回。生成 10 秒，用户就对着转圈占位符干等 10 秒。流式输出的做法是：服务端每拿到一小段内容就立刻推给浏览器，文字像打字机一样逐段蹦出来——这是所有主流 AI 聊天产品的标配交互，而承载它的主流协议就是 SSE（Server-Sent Events，服务器发送事件）。本文把协议细节、Node.js 转发实现和 nginx 反代下的生产坑一次讲透，文中的服务端与解析代码均在本地实际跑通（Node v24.4.1 + Express 5.2.1）。

先算一笔感知延迟的账

流式并不会让总生成时长变短，它压缩的是 TTFT（Time To First Token，首 token 延迟）——用户「等到第一个字」的时间。OpenAI 官方 Cookbook 有一组直观的示例数据：让模型从 1 数到 100，非流式要等约 1.88 秒才一次性拿到全部结果；改用流式，第一个 token 约 0.1 秒到达，之后每个 token 间隔仅 0.01–0.02 秒（截至 2026-10-07 查询的官方示例，实测数值随模型与负载浮动，但「TTFT 远小于总时长」的量级关系普遍成立）。

总时长几乎没变，体验却是天壤之别：0.1 秒就有内容开始滚动，用户感知是「立刻在回答我」；1.88 秒的白屏感知是「卡了」。流式还有个次要收益：用户能提前发现答非所问、及时点停止，省掉后续无效生成的费用。

WebSocket、SSE、分块轮询怎么选

给 LLM 应用做流式，常见三条路：

方案	通信方向	断线处理	协议开销	适配 LLM 流式
分块轮询 / 长轮询	客户端拉	自己处理	反复建连，HTTP 头开销大	勉强，延迟与浪费都大
WebSocket	双向	自研心跳与重连	升级后的二进制帧	可以，但通常只用到「服务器推」这一半
SSE	服务器推	浏览器内置自动重连	纯 HTTP 文本流	天然契合

LLM 对话的形状是：请求一次（普通 HTTP POST 带上 prompt），响应是一条持续下行的 token 流。SSE 的单向推送正好匹配这个形状；它跑在普通 HTTP 上，文本协议便于日志排查、网关透传和复用现有鉴权；浏览器端 EventSource 还内置自动重连。WebSocket 的双向能力在「AI 回复」场景里基本闲置，心跳、重连、帧解析都要自己养。所以 OpenAI 兼容 API 的流式响应干脆就是 SSE。（需要双向低频交互——语音通话、协作编辑——才值得上 WebSocket，这是本文作者判断。）

SSE 有个值得提前知道的约束：MDN 指出 HTTP/1.1 下浏览器对同一域名的连接数上限很低（6 个），Chrome 和 Firefox 均标记为 Won't fix；HTTP/2 多路复用下默认可协商 100 条流。生产环境上 HTTP/2 即可化解，后文多租户一节还会回到这点。

SSE 协议五分钟

SSE 的「协议」就是一段 UTF-8 文本流，响应的 MIME 类型必须是 text/event-stream。事件之间用空行分隔，每行是「字段名: 值」，标准字段只有四个：

• data: 消息数据。多条连续 data 行会被拼接成一条消息（行间插入换行）。
• event: 事件类型，客户端用 addEventListener 监听；缺省触发 message。
• id: 事件编号。浏览器记住最后一个 id，自动重连时通过 Last-Event-ID 请求头带回给服务端。
• retry: 告诉浏览器自动重连前等待的毫秒数。

冒号开头的行是注释，客户端忽略，常用来做心跳保活。除此之外没有任何二进制握手——这也是它好排查、好代理的原因。

OpenAI 兼容 API 在请求参数里设 stream: true 后，返回的就是这样一条流（Chat Completions 流式格式，以官方文档与 SDK 源码为准）：

HTTP/1.1 200 OK
Content-Type: text/event-stream

data: {"id":"chatcmpl-...","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"role":"assistant"},"finish_reason":null}]}

data: {"id":"chatcmpl-...","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"你"},"finish_reason":null}]}

data: {"id":"chatcmpl-...","object":"chat.completion.chunk","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}

data: [DONE]

三个要点：object 是 chat.completion.chunk，区别于非流式的 chat.completion；增量内容在 choices[0].delta.content 里，而不是非流式的 message；流以一条 data: [DONE] 收尾——OpenAI 官方 SDK 的流解析器就是靠它判断终止的（openai-python 的 _streaming.py 中有一行 if sse.data.startswith("[DONE]"): break）。另外，首条 delta 通常只有 role，末条 delta 是空对象且 finish_reason 为 stop；想要 token 用量统计，需设 stream_options: {"include_usage": true}，用量挂在最后一个 choices 为空数组的 chunk 上。

服务端：Express 把上游流转发给浏览器

下面是一个把「上游 LLM 流」转发给浏览器的 SSE 端点。真实场景里上游是对 OpenAI 兼容 API 的流式 fetch，这里按任务最小化原则用 async generator 逐 token 模拟，只验证转发链路本身：

// server.js — Express SSE 端点：把「上游 LLM 流」转发给浏览器
// 运行：node server.js
const express = require("express");

const app = express();

// —— 假的上游：逐 token 产出的 async generator ——
// 真实场景这里是对 OpenAI 兼容 API 的流式 fetch，
// 本文只验证浏览器这一段链路，用定时器模拟推理间隔。
async function* fakeUpstreamStream() {
  const tokens = ["流式", "输出的", "价值", "在于", "：", "让用户", "在第一秒", "就看到", "反馈", "。"];
  for (const token of tokens) {
    await new Promise((r) => setTimeout(r, 80)); // 模拟逐 token 的推理间隔
    yield token;
  }
}

app.get("/api/chat/stream", async (req, res) => {
  // 1. SSE 必需的响应头
  res.set({
    "Content-Type": "text/event-stream; charset=utf-8",
    "Cache-Control": "no-cache",
    Connection: "keep-alive",
    "X-Accel-Buffering": "no", // nginx 收到此头即不再缓冲该响应
  });
  res.flushHeaders();

  // 2. 心跳：冒号开头的注释行，客户端不会当事件处理，
  //    但能让代理与客户端确认连接仍然存活。
  const heartbeatMs = Number(process.env.SSE_HEARTBEAT_MS) || 15000;
  const heartbeat = setInterval(() => res.write(": ping\n\n"), heartbeatMs);

  // 3. 客户端中途断开时 close 触发，停止无谓的上游消费
  let closed = false;
  res.on("close", () => {
    closed = true;
    clearInterval(heartbeat);
  });

  // 4. 转发：上游每吐一个 token，就写一条 data: 事件
  try {
    for await (const token of fakeUpstreamStream()) {
      if (closed) break;
      const chunk = {
        id: "chatcmpl-demo",
        object: "chat.completion.chunk",
        choices: [{ index: 0, delta: { content: token }, finish_reason: null }],
      };
      res.write(`data: ${JSON.stringify(chunk)}\n\n`);
    }
    if (!closed) res.write("data: [DONE]\n\n");
  } catch (err) {
    if (!closed) {
      res.write(`data: ${JSON.stringify({ error: String(err) })}\n\n`);
      res.write("data: [DONE]\n\n");
    }
  } finally {
    clearInterval(heartbeat);
    res.end();
  }
});

app.listen(8787, () => console.log("SSE demo listening on http://127.0.0.1:8787"));

五个实现要点：

1. 响应头四件套：Content-Type: text/event-stream 是协议硬要求（顺手带上 charset=utf-8，SSE 规定按 UTF-8 解码）；Cache-Control: no-cache 禁止中间层缓存；Connection: keep-alive 维持长连接；X-Accel-Buffering: no 是写给 nginx 的——按 nginx 官方文档，upstream 响应头里这个字段的优先级高于 proxy_buffering 指令。
2. res.flushHeaders() 立刻发出响应头，让浏览器尽早建立 EventSource 连接、进入接收状态。
3. 心跳用注释行：: ping 不会派发成事件，却能把「空闲间隔」压到各层超时阈值以内。
4. 监听 close 感知断开：用户关掉页面后继续消费上游就是白烧 token，closed 标志让循环及时 break。
5. 正常结束写 data: [DONE] 再 end()，与 OpenAI 风格客户端的终止约定保持一致。

curl 验证

起服务后用 curl -N（--no-buffer，禁掉 curl 自身的输出缓冲）验证：

curl -sN "http://127.0.0.1:8787/api/chat/stream?q=hi"

实际输出（节选，本地真实跑出的结果）：

data: {"id":"chatcmpl-demo","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"流式"},"finish_reason":null}]}

data: {"id":"chatcmpl-demo","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"输出的"},"finish_reason":null}]}

data: {"id":"chatcmpl-demo","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"。"},"finish_reason":null}]}

data: [DONE]

心跳默认 15 秒一条，demo 流总共不到一秒跑完，平时看不到它。验证时我把心跳间隔调成 30 毫秒（环境变量 SSE_HEARTBEAT_MS=30），输出里就能看到 : ping 穿插在 data 事件之间——注释行确实原样透传且不影响事件流。另外用 curl -m 0.25 模拟客户端中途断开，服务端日志干净、进程健在，后续请求照常服务，说明 close 清理逻辑正常。

前端：EventSource 的天花板与 fetch 替代

浏览器原生的 EventSource 三行代码就能收流，还自带自动重连：

const es = new EventSource("/api/chat/stream?q=hi");
es.onmessage = (e) => appendToChat(JSON.parse(e.data).choices[0].delta.content);

但它有两个写死在 HTML 规范里的限制：只能发 GET，不能带任何自定义请求头。这意味着带不了 Authorization 头——除非把 token 塞进 URL（会落进访问日志，不安全）或改用 Cookie。LLM 网关几乎都要鉴权头，于是生产上更常用 fetch + ReadableStream 自己解析：

// client.mjs — 用 fetch + ReadableStream 消费 SSE
// EventSource 只发 GET、带不了自定义 header；
// 需要带 Authorization 或 POST 请求体时，用 fetch 自己解析流。
const resp = await fetch("http://127.0.0.1:8787/api/chat/stream?q=hi", {
  headers: { Authorization: "Bearer test-token" }, // EventSource 加不了这个头
});
if (!resp.ok || !resp.headers.get("content-type")?.startsWith("text/event-stream")) {
  throw new Error(`unexpected response: ${resp.status} ${resp.headers.get("content-type")}`);
}

const reader = resp.body.getReader();
const decoder = new TextDecoder(); // SSE 规定 UTF-8 解码
let buffer = "";
let text = "";

while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  buffer += decoder.decode(value, { stream: true });

  // 事件之间用空行（\n\n）分隔
  let sep;
  while ((sep = buffer.indexOf("\n\n")) !== -1) {
    const rawEvent = buffer.slice(0, sep);
    buffer = buffer.slice(sep + 2);

    // 逐行取 data: 字段；多个 data 行按规范用换行拼接
    const dataLines = [];
    for (const line of rawEvent.split("\n")) {
      if (line.startsWith(":")) continue; // 注释行（心跳）
      if (line.startsWith("data:")) dataLines.push(line.slice(5).replace(/^ /, ""));
    }
    const data = dataLines.join("\n");

    if (data === "[DONE]") {
      console.log("\n[DONE] received");
      console.log("final text:", text);
      process.exit(0);
    }
    if (data) {
      const chunk = JSON.parse(data);
      const delta = chunk.choices?.[0]?.delta?.content ?? "";
      text += delta;
      process.stdout.write(delta); // 打字机效果：来一段渲染一段
    }
  }
}
console.log("final text:", text);

解析器只做三件事：TextDecoder 以流式模式解码（必须传 { stream: true }，否则多字节中文被 chunk 切开时会解出乱码）；按空行切事件、逐行提取 data: 字段，多个 data 行按规范用换行拼接；跳过冒号开头的注释行。本地实测，这段代码对着上面的服务端逐段打出完整句子、正确略过心跳、收到 [DONE] 干净退出。代价是：EventSource 免费送的自动重连与 Last-Event-ID 回传，现在要自己做。

生产环境五个坑

1. nginx 反代缓冲。proxy_buffering 默认 on：nginx 会尽量攒满缓冲区再发给客户端，SSE 就从「打字机」退化成「憋半天一口气全出来」。解法是让后端响应头带 X-Accel-Buffering: no（优先级高于指令，也是最省事的方案），或在该 location 显式 proxy_buffering off;；反过来，配置里若有 proxy_ignore_headers X-Accel-Buffering; 则会把它忽略掉，别误伤。
2. gzip 压缩会缓冲。开了 gzip 的 location 上，压缩器要攒够数据才输出，连 X-Accel-Buffering: no 也救不了它。必须对 SSE 路由 gzip off;。这是最隐蔽的一坑：本地直连后端一切正常，挂上 nginx 才「卡住、打包送达」。
3. 连接超时。nginx 的 proxy_read_timeout 默认 60 秒，语义是「两次成功读之间」的超时：模型长时间不吐 token（工具调用、深度思考）就会中招。对策是组合拳：心跳注释行把空闲间隔压进阈值，同时按业务调大该值（如 proxy_read_timeout 3600s;）。
4. 断线重连与 Last-Event-ID。EventSource 断线后自动重连（retry: 可调间隔），并把最后的 id: 放进 Last-Event-ID 请求头。但 LLM 的 token 流通常是「丢了就丢了」——严格续传需要服务端缓存整段生成结果按 id 重放；更常见的务实做法是：重连后重新发起生成，把已收到的文本作为上下文回显，或把生成任务落库、前端改走进度查询。
5. 多租户下的连接数管理。SSE 是长连接：HTTP/1.1 下浏览器对同域名只有 6 条并发连接（多个标签页很快耗尽），服务端每条连接也各占一个 socket 与内存。对策：全站上 HTTP/2（单连接多路复用，默认可协商 100 条流）；网关层按用户做并发连接配额，对闲置连接超时主动 res.end()，防止被刷爆。

回到开头的账：网关决定「怎么调度一次 LLM 调用」，SSE 链路决定「怎么把结果流畅地送出去」，两者拼起来才是完整的 LLM 服务端体验。这套链路里没有黑魔法——一个 text/event-stream、几行 data:、一版诚实的缓冲配置，打字机效果就稳了。

参考资料

1. MDN Web Docs — Using server-sent events：https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events
2. WHATWG HTML Standard — Server-sent events（协议权威规范）：https://html.spec.whatwg.org/multipage/server-sent-events.html
3. OpenAI Cookbook — How to stream completions（TTFT 示例数据与 delta 结构）：https://developers.openai.com/cookbook/examples/how_to_stream_completions
4. openai-python — src/openai/_streaming.py（data: [DONE] 终止符与 SSE 解析实现）：https://github.com/openai/openai-python/blob/main/src/openai/_streaming.py
5. nginx 官方文档 — ngx_http_proxy_module（proxy_buffering / X-Accel-Buffering 语义）：https://nginx.org/en/docs/http/ngx_http_proxy_module.html