AI API 流式输出 SSE 实现与常见坑

几乎所有 AI 聊天产品都在用流式输出：模型一边生成，用户一边看到文字逐字出现。它的技术基础是 SSE（Server-Sent Events），实现起来看似只是加一个 stream: true，但上线后的问题五花八门：本地好好的，部署后变成「卡十几秒再一次性全部吐出来」；偶尔输出到一半就停了，前端却以为已经结束；中文时不时出现乱码；用户点了停止，账单上还在扣费。

这篇把流式输出从协议到部署完整走一遍：事件格式怎么读、OpenAI 和 Anthropic 有什么区别、nginx 为什么会把流「憋住」、断流怎么判断和重试、前端怎么正确读流，以及流式请求的计费口径。SDK 层面的快速上手可以先看 Node.js 调用 Claude、GPT、Gemini API 教程。

SSE 是什么：一个不关闭的 HTTP 响应

SSE 本质上是一个持续写入的普通 HTTP 响应。服务端返回 Content-Type: text/event-stream，然后不断往连接里写文本行，每个事件由若干 字段: 值 行组成，事件之间用一个空行分隔：

event: message
data: {"text": "你好"}

data: {"text": "，世界"}

规则只有几条：以 data: 开头的行是数据，event: 是可选的事件类型，空行表示一个事件结束，以冒号开头的行是注释（常被用作心跳）。和 WebSocket 相比，SSE 是单向的（服务端到客户端）、基于普通 HTTP、穿透代理和防火墙更容易，正好契合「发一次请求、收一长串输出」的大模型场景。

开启方式就是在请求体里加 "stream": true：

curl -N https://zivv.pro/v1/chat/completions \
  -H "Authorization: Bearer sk-your-key" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "claude-sonnet-5",
    "stream": true,
    "messages": [{"role": "user", "content": "数到五"}]
  }'

-N 关闭 curl 自己的输出缓冲，你会看到事件一行行地出现。排查流式问题时，这条命令是最好的基准：curl 直连是逐字出现的，而你的应用里不是，问题就在你自己的链路上。

事件格式：OpenAI 与 Anthropic 的差异

不同协议的 SSE 事件结构差别很大，自己解析时必须分清。

OpenAI 兼容格式（/v1/chat/completions）只用 data: 行，每个事件是一个 chat.completion.chunk，以 data: [DONE] 结束：

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

data: {"choices":[{"index":0,"delta":{"content":"一、二"},"finish_reason":null}]}

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

data: [DONE]

Anthropic 原生格式（/v1/messages）同时使用 event: 和 data:，按消息生命周期分成多种事件：

event: message_start
data: {"type":"message_start","message":{"id":"msg_...","usage":{"input_tokens":12,"output_tokens":1}}}

event: content_block_start
data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}

event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"一、二"}}

event: content_block_stop
data: {"type":"content_block_stop","index":0}

event: message_delta
data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":15}}

event: message_stop
data: {"type":"message_stop"}

另外，/v1/responses 接口（Codex 等工具使用）的事件又是另一套命名，比如 response.output_text.delta 和 response.completed；Gemini 兼容接口的流式也有自己的结构。结论：能用官方 SDK 就别手写解析器，SDK 已经处理好了这些差异；必须手写时，按协议分别实现，不要指望一个通用解析逻辑吃遍所有格式。

还有一个容易踩的坑：Anthropic 流里可能出现非文本块，比如开启 thinking 时的 thinking_delta，或工具调用时的 input_json_delta。只处理 text_delta 的解析器要显式忽略其他类型，而不是报错。

最常见的坑：nginx 缓冲让流「卡住」

「本地逐字输出，上线后一次性吐出」，十有八九是反向代理在缓冲。nginx 默认开启 proxy_buffering：它会先把上游响应攒到缓冲区里，攒够了或响应结束了才发给客户端。对普通网页这是优化，对 SSE 就是灾难。

在代理 AI 接口的 location 里关闭缓冲：

location /api/chat {
    proxy_pass http://127.0.0.1:3000;

    proxy_http_version 1.1;
    proxy_set_header Connection "";

    proxy_buffering off;        # 关键：不缓冲上游响应
    proxy_cache off;            # 流式响应不能缓存
    gzip off;                   # 压缩也会攒数据

    proxy_read_timeout 300s;    # 相邻两次数据之间的最大间隔
    proxy_send_timeout 300s;
}

如果不方便改 nginx 全局配置，也可以由应用在响应头里加 X-Accel-Buffering: no，nginx 会对这条响应单独关闭缓冲。除了 nginx，以下几层同样会缓冲，排查时逐一确认：

• 压缩中间件：Express 的 compression 等中间件会等数据攒够再压缩输出，对 SSE 路由要跳过
• CDN 与 WAF：部分 CDN 默认缓冲或缓存响应，需要对流式路径关闭缓存、开启流式透传
• Serverless 平台：部分平台的函数默认把响应完整生成后再返回，需要使用平台提供的流式响应方式
• 应用框架本身：确认你的路由真的在边收边写，而不是把上游流全部读完后再 res.send

proxy_read_timeout 也值得单独说一句：它限制的是两次读取之间的间隔，而不是总时长。模型在输出前可能「思考」一段时间不发数据，这个值设得太小（nginx 默认 60 秒），长思考任务会在中间层被掐断。更多超时分层的设置见 529 与超时、流式中断排查。

前端用 fetch 读流

浏览器原生的 EventSource 只支持 GET 请求、不能自定义请求体和鉴权头，不适合调用大模型接口。实际做法是用 fetch 发 POST，再手动读 response.body：

async function streamChat(messages: unknown[], onText: (t: string) => void, signal?: AbortSignal) {
  const res = await fetch("/api/chat", {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ messages, stream: true }),
    signal,
  });
  if (!res.ok || !res.body) throw new Error("HTTP " + res.status);

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

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

    const events = buffer.split("\n\n");
    buffer = events.pop() ?? ""; // 最后一段可能不完整，留到下一轮

    for (const evt of events) {
      for (const line of evt.split("\n")) {
        if (!line.startsWith("data:")) continue;
        const data = line.slice(5).trim();
        if (data === "[DONE]") { finished = true; continue; }
        const json = JSON.parse(data);
        const text = json.choices?.[0]?.delta?.content;
        if (text) onText(text);
      }
    }
  }
  if (!finished) throw new Error("stream ended without [DONE]");
}

这段代码处理了三个最常见的前端问题：

1. 事件被切断：网络数据块的边界和 SSE 事件边界没有任何关系，一个 JSON 可能被切成两半。必须用缓冲区累积，按空行切分，把最后不完整的部分留到下一轮
2. 中文乱码：TextDecoder 必须用 { stream: true }，否则一个多字节字符被切在两个块之间时会解码失败
3. 假结束：读到 done 不等于正常完成，只有收到 [DONE]（或 Anthropic 的 message_stop）才算完整；否则应该提示用户「输出中断」

另外，注意浏览器不应该直接持有 API Key。上面的 /api/chat 应该是你自己的服务端路由，由服务端带上 Key 去请求 https://zivv.pro/v1。

停止生成：AbortController

「停止生成」按钮用 AbortController 实现：

const controller = new AbortController();
stopButton.onclick = () => controller.abort();
await streamChat(messages, appendToUI, controller.signal);

只在浏览器端中断还不够：如果你的服务端在转发，要把客户端断开传递给上游请求（Node.js 中把请求的 signal 传给 SDK），否则浏览器已经不看了，服务端还在继续读上游、继续消耗 Token。

断流判断与重连

流式请求中途断开的原因很多：移动网络切换、代理空闲回收、上游过载。处理原则：

• 先判断是否真的断了：以结束标志为准，OpenAI 格式看 [DONE] 与 finish_reason，Anthropic 格式看 message_stop。没有结束标志就关闭的连接一律视为中断
• 区分「还没出字」和「出了一半」：首个数据到达前失败，可以直接自动重试；已经展示了一部分内容，自动重试会造成重复文本，更好的做法是保留已有内容并提供「继续生成」或「重新生成」
• 续写而不是重来：长文和代码场景，把已收到的内容作为 assistant 消息放回上下文，让模型从断点继续，比整段重新生成更省 Token
• 流中的错误事件：Anthropic 协议可能在流中途下发 event: error（如 overloaded_error），这时 HTTP 状态码已经是 200，必须在事件层面识别并处理

需要特别说明：大模型 API 的 SSE 不支持 Last-Event-ID 那种「从某个事件续传」，所谓重连就是发起一个新请求。重试的退避策略可以直接复用 429 排查文章 里的指数退避实现。

流式请求怎么计费

很多人以为流式更贵或更便宜，实际上计费与是否流式无关：

• 按实际 Token 计费：输入 Token 加输出 Token，价格为分组倍率乘以官方模型单价，流式和非流式完全一致（详见 计费说明）
• 中途停止或断开：已经生成的输出 Token 通常仍会计入，停止越早，消耗越少。这也是要把客户端中断传递给上游的原因
• 怎么拿到流式用量：OpenAI 格式在请求中加 "stream_options": {"include_usage": true}，最后一个数据块会带 usage（此时 choices 为空数组，解析时要容错）；Anthropic 格式从 message_start 和 message_delta 中读取
• 以平台用量为准：自己统计的数字用于实时展示和预算控制，最终对账以控制台的用量记录为准

想估算一个流式聊天功能的月成本，可以参考 Token 成本计算指南。

上线前的流式检查清单

常见问题 FAQ

Q：为什么流式输出本地正常，部署后一次性全部出来？ A：几乎都是中间层缓冲：nginx 的 proxy_buffering、压缩中间件、CDN 或 Serverless 平台。按本文 nginx 一节关闭缓冲，或在响应头加 X-Accel-Buffering: no，再用 curl -N 逐层验证。

Q：前端能用 EventSource 调用大模型接口吗？ A：不适合。EventSource 只能发 GET 请求，不能带请求体和自定义鉴权头。用 fetch 发 POST，再读 response.body 解析 SSE 是标准做法。

Q：流式输出到一半断了怎么办？ A：先用结束标志确认是中断；首字前失败可自动重试，中途断开则保留已有内容并提供续写或重新生成。上游过载导致的中断可以参考 529 排查 做退避和降级。

Q：流式输出比非流式更贵吗？ A：不会。两种方式都按实际输入输出 Token 计费，流式只是改变了返回方式。

Q：OpenAI 格式和 Anthropic 格式的流可以用同一个解析器吗？ A：不建议。两者的事件名、文本字段位置和结束标志都不同。用 OpenAI 兼容入口 https://zivv.pro/v1 调用所有模型可以只维护一种解析逻辑；需要 Claude 原生事件时再单独实现 Anthropic 解析。

Q：Claude Code、Cursor 这类工具也需要配置流式吗？ A：不需要，这些工具内部已经默认使用流式，你只需配置好 Base URL 和 Key，参考 Cursor 自定义 API 教程。如果在它们前面自建了代理，同样要关闭代理缓冲。

写在最后

流式输出的难点从来不在 stream: true 这一行，而在链路上的每一层：代理要透传、超时要按间隔设置、前端要正确切分事件、断流要靠结束标志识别、停止要传到上游。把本文的检查清单逐项过一遍，绝大多数「卡顿」「断流」「乱码」问题都能在上线前解决。想直接动手测试，注册 Zivv 后用开头的 curl 命令即可看到逐字输出，模型列表见 模型广场。