LLM 流式输出协议深度:SSE、Data Stream 与解析

流式输出是 AI 应用的体感命门,这件事现在没人反对了——首字 1 秒 + 边生成边吐,比憋 10 秒一次性吐体验好得多,哪怕总时间更长。

但"流式"这词被说烂了,底下到底在流什么,很多人是糊涂的。打开 Network 面板,你会看到一条 text/event-stream 的响应,里面一行行 data: {...}——这不是 WebSocket,而是一套叫 SSE 的协议。AI SDK 又在它之上叠了一层"data stream 协议",把 token、tool call、错误、中止全塞进带类型标记的事件流。再往上还有 Nginx/Cloudflare,它们一个缓冲设置没配对,你的"流式"就退化成"一次性"。

这篇就讲这几层协议:从 HTTP 长连接到 SSE 语法,到 data stream 的类型事件,到流式 tool calling 的复杂度,到可中断流和代理缓冲的坑。读完你应该能手写一个能解析、能中断、能穿透 CDN 的流式客户端,而不是只会调 useChat。

一、为什么是 SSE:HTTP 长连接上的单向流

先说结论:LLM 流式的事实标准是 SSE(Server-Sent Events),不是 WebSocket。

SSE 本质是一个普通的 HTTP/1.1 GET(或 POST)响应,只是服务端不结束它,而是一直往里写数据。三个关键点:

1. Content-Type: text/event-stream,告诉浏览器"这是事件流,别等响应结束再给我"。
2. 单向:只有服务端往客户端推,客户端不通过这条连接回传。聊天场景里用户输入是另发一个 POST,不需要双向。
3. 基于 HTTP:走标准 HTTP 基础设施,天然过 CDN、过 H2/H3、过鉴权头、过 cookie,不需要协议升级握手。

对比 WebSocket:

LLM 场景 99% 是"客户端发一次问 → 服务端流式回一大段",纯单向。用 WebSocket 等于杀鸡用牛刀,还要扛握手、代理兼容、自实现重连的负担。SSE 赢了,不是因为它更强大,而是因为它更贴合场景。

一个值得知道的点:浏览器内置的 EventSource 只支持 GET、不能自定义请求头,所以 AI 应用里(要 POST body、要带 Authorization)几乎都是用 fetch 手撸 SSE 解析,而不是用 EventSource。这也是为什么下一节要讲手解析。

二、原始 SSE 解析:ReadableStream + TextDecoder + 按 \n\n 分块

先把 SSE 协议本身的语法说清。SSE 流的格式是这样的:

data: {"v":1}

data: {"v":2}

event: ping
data: keep-alive

id: 42
data: {"v":3}

规则就几条:每个事件由若干行组成,事件之间用一个空行(也就是 \n\n)分隔;每行是 field: value 的形式;data: 是数据体,event: 是事件类型,id: 是事件 ID,retry: 是建议的重连间隔(ms)。 一个事件的 data: 可以有多行,拼接起来才是完整数据。

fetch 拿到这种流后,response.body 是个 ReadableStream<Uint8Array>。解码 + 分块的核心代码:

async function parseSSE(
  res: Response,
  onData: (data: string, event?: string) => void,
) {
  const reader = res.body!
    .pipeThrough(new TextDecoderStream())   // bytes -> string
    .getReader();
  let buffer = "";
  let event = undefined;

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

    // SSE 以空行(\n\n)分隔事件
    let sep: number;
    while ((sep = buffer.indexOf("\n\n")) !== -1) {
      const raw = buffer.slice(0, sep);
      buffer = buffer.slice(sep + 2);

      const dataLines: string[] = [];
      for (const line of raw.split("\n")) {
        if (line.startsWith("data:")) dataLines.push(line.slice(5).trimStart());
        else if (line.startsWith("event:")) event = line.slice(6).trim();
        // id: / retry: 同理,需要可自行收集
      }
      if (dataLines.length) onData(dataLines.join("\n"), event);
      event = undefined;
    }
  }
}

两个最容易踩的坑都在这段代码里:

• 必须用 buffer 缓存半截。 reader.read() 一次返回的 chunk 可能在 data: 中间断开。直接 split 会把一条事件劈成两半。正确做法是只在 buffer 里找完整的 \n\n 分隔,处理完把尾巴留到下一次。
• 用 indexOf("\n\n") 而不是 split。 长流里 split 每次都遍历整段 buffer,indexOf 找到一处处理一处,性能更好。
• 换行符要小心:规范是 \n,但有些服务端发 \r\n\r\n。生产代码里最好先 replaceAll("\r\n", "\n") 归一化。

这是最底层的能力:拿到任意 SSE 流,你能正确还原出一条条事件。下面看 AI SDK 在这上面叠了什么。

三、AI SDK 的 data stream 协议:在 SSE 之上加类型标记

裸 SSE 只能告诉你"有一条数据",但这条数据是文本 token、还是 tool 调用、还是错误,得你自己约定。AI SDK 的 data stream 协议就是这套约定:把模型生成的所有东西按类型打成带 type 字段的 JSON 事件,塞进 SSE 的 data: 里。

注意一个坑:这个协议改过版。 AI SDK v3/v4 用的是"行前缀"格式——0:"文本",2:[{...}],9:{tool_call},行首数字代表类型,很多老博客和教程里还是这套。v5 起换成了更标准的 SSE + JSON 格式,而且更重要的好处是:支持 ping 心跳、可重连、缓存友好。下面以当前版本为准。

当前 data stream 的事件类型(每个都是 data: {...} 一行):

注意一个设计:文本不是一条 text 事件,而是 start/delta/end 三段,而且每个文本块有唯一 id。 这是为了支持一条消息里多个文本块、思维链、tool 调用交错出现——前端按 id 把 delta 归位到正确的块上。

后端用 AI SDK 发流,当前 API 是 toUIMessageStream + createUIMessageStreamResponse:

// app/api/chat/route.ts (Next.js)
import { convertToModelMessages, createUIMessageStreamResponse,
  streamText, toUIMessageStream, UIMessage } from "ai";

export async function POST(req: Request) {
  const { messages }: { messages: UIMessage[] } = await req.json();
  const result = streamText({
    model,                                  // 你的 provider
    messages: await convertToModelMessages(messages),
  });
  return createUIMessageStreamResponse({
    stream: toUIMessageStream({ stream: result.stream }),
  });
}

前端用 @ai-sdk/react 的 useChat 直接消费,内部就是把上面那张表的事件还原成 message.parts。如果你要写非 JS 后端(Python FastAPI 等)对接 useChat,照这张表手发事件即可,记得带上响应头 x-vercel-ai-ui-message-stream: v1,否则前端会按旧协议解析。

(还有一个更简单的 text stream 协议,就是纯文本 chunk、没有类型标记,适合只需要吐文字、不要 tool call 的场景,用 toTextStream / createTextStreamResponse。下文主要讲功能更全的 data stream。)

四、流式 tool calling:模型边生成边吐 tool 参数

流式 tool calling 是整个协议里最精巧、也最容易翻车的部分。

先看模型在不流式时怎么调 tool:生成完一整段 JSON 参数,解析,执行。简单粗暴。但在流式模式下,模型是逐 token 输出的,你会看到 {"city":"San 这样的半截 JSON。协议要解决的问题是:怎么在参数还没生成完时,就让前端把"正在调用 getWeather"这个状态显示出来,而且生成完了能拿到完整参数。

data stream 的解法是三段式:

data: {"type":"tool-input-start","toolCallId":"call_abc","toolName":"getWeather"}

data: {"type":"tool-input-delta","toolCallId":"call_abc","inputTextDelta":"{\"city\":\"San"}

data: {"type":"tool-input-delta","toolCallId":"call_abc","inputTextDelta":" Francisco\"}"}

data: {"type":"tool-input-available","toolCallId":"call_abc","toolName":"getWeather","input":{"city":"San Francisco"}}

• tool-input-start:前端立刻可以渲染一个"正在调用 getWeather…"的占位卡片,体验远好于干等。
• tool-input-delta:把参数碎片拼起来,可以做"参数正在生成"的实时预览(注意:半截 JSON 不能 JSON.parse,要么只展示原文,要么用容错的增量 JSON 解析器)。
• tool-input-available:这才是可靠的参数来源。 服务端已经把完整参数 parse 好了,直接用,别信自己拼出来的字符串。

复杂度来自几个地方:多 tool 并行(一条消息里模型可能并发调多个 tool,前端要按 toolCallId 分别归位);参数中途改主意(模型吐到一半改结构,预览会"跳变",要做平滑);tool 执行本身也流式(tool-output-available 之前,长任务 tool 可以自己推 data-* 进度事件)。这些都是为什么协议里文本块、tool 块都要带 id 而不是单纯追加。

实战经验:前端 UI 一定要区分"参数预览"和"最终参数"两态。 预览态(收到 delta 但还没 available)用低对比度/打字光标,available 一到就切换成正式卡片,否则用户盯着一个不断跳变的半截 JSON,观感很差。

五、可中断流:AbortController + 后端感知取消

可中断是流式的标配——用户点"停止",生成必须立刻停,而且已经吐出来的内容要保留。这要求前后端配合。

前端用 AbortController:

const controller = new AbortController();
fetch("/api/chat", { method: "POST", body: ..., signal: controller.signal });
// 用户点停止
controller.abort();   // fetch 抛 AbortError,reader 也一并断开

useChat().stop() 内部干的就是这个。关键在后端要感知到这个取消。Next.js Route Handler 里,req.signal 就是和客户端连接绑定的 AbortSignal:

export async function POST(req: Request) {
  const result = streamText({
    model,
    messages,
    abortSignal: req.signal,        // 把客户端取消信号透传给模型
  });
  return createUIMessageStreamResponse({
    stream: toUIMessageStream({ stream: result.stream }),
  });
}

客户端 abort() → TCP 连接断 → 服务端 req.signal 触发 → streamText 收到信号 → 停止向 provider 续 pull token。这条链路任何一个环节没接上,中断就是假的: 用户点了停止,后端还在傻傻烧 token 烧钱。

几个坑:

• 中断不是错误,别按错误处理。 fetch abort 抛的是 AbortError(name 是 "AbortError"),要单独 catch,UI 上保留已生成文本而不是显示"出错"。
• 保留部分输出。 data stream 协议在客户端中断时,服务端可以发一条 {"type":"abort","reason":"user cancelled"},前端据此把消息标记为"已中止"而不是"已完成"。已收到的 text-delta 不要丢。
• provider 的流要及时关。 abortSignal 触发后,要确保对 provider(OpenAI/Anthropic)那条上游连接也主动 close,否则你停了客户端、上游还在计费。

六、代理/CDN 缓冲:流式失效的头号杀手

这是真实生产里最坑的一类问题:代码完全正确,但流式就是退化成"等很久一次性全到"。99% 是中间层把流缓冲了。

SSE 要工作,数据必须一产生就转发,中间任何一层攒着不发,体感就崩。常见的缓冲来源:

1. Nginx 反向代理。 默认 proxy_buffering on,会把响应攒到一定大小或上游结束才转发。解法是给流式 location 关掉:

location /api/chat {
  proxy_pass http://upstream;
  proxy_buffering off;          # 关键
  proxy_cache off;
  proxy_http_version 1.1;       # 长连接
  proxy_read_timeout 600s;      # LLM 慢,别用默认 60s
  add_header X-Accel-Buffering no;
}

更省心的做法是让后端响应直接带 X-Accel-Buffering: no 头,Nginx 看到它会对这条响应禁用缓冲——不用改 Nginx 配置,尤其适合 Serverless。

2. 压缩中间件。 gzip/br 压缩必然要攒一批字节才有意义,开启后流式基本失效。对 text/event-stream 响应必须跳过压缩。

3. Cloudflare 等 CDN。 CF 默认会缓冲,免费版在某些情况下会等响应结束才下发,流式体感全无。排查时直接绕过 CF 打源站对比,立刻能定位。

4. HTTP/2 的流控。 H2 理论上对流式友好,但某些中间层的 H2 实现会做聚合。一般保留 H1.1 + proxy_http_version 1.1 最稳。

自测方法:Network 里看那条请求的响应体出现时机——流式正常时,发起后 1 秒内就会陆续到达(size 持续增长);缓冲了的话,会卡很久一次性全到。DevTools 的 EventStream 面板还能直接看到一条条事件。

七、错误流与重连

流比一次性请求多两种失败模式:中途出错和连接断开,协议都要管。

中途出错。 模型生成到一半挂了(限流、内容审核、provider 内部错误),data stream 会推一条 {"type":"error","errorText":"..."},然后通常跟着结束。前端处理这条事件时,保留已生成的文本,把消息标为"部分生成 + 出错",给个重试按钮——而不是清空重来,那是用户体验灾难。

客户端断连。 区分两类:用户主动 abort(上节的 AbortError)和网络波动被动断开。SSE 规范自带重连机制:服务端发 id: 123,浏览器 EventSource 断线后会带 Last-Event-ID: 123 自动重连,retry: 3000 可告诉浏览器 3 秒后再连。但 AI 应用大多用 fetch 不用 EventSource,所以重连得自己写:记录最后收到的消息位置,断开时指数退避重发。

实际工程里,多数产品对"中途断开"采取轻量重试而非严格续传——模型流式不好"从中间接上"(provider 侧要支持、上下文要重建)。务实做法:断开就提示"网络中断,点此重试",重试时把已生成的半截保留为 assistant 消息,重新生成。严格无感续传性价比通常不高,除非是超长生成的硬场景。

小结

流式输出的"流"字,底下是几层叠加的协议,每层都有要讲清的细节:

• SSE 是底座:HTTP 长连接、单向、text/event-stream、按 \n\n 分事件。LLM 场景几乎都用 fetch 手解析而非 EventSource(要 POST + 自定义头)。
• data stream 是约定:在 SSE 的 data: 里塞带 type 的 JSON 事件(text/tool-input/data/error/finish…)。v5 起是 JSON-SSE 格式,取代了老的 0:/9: 行前缀;非 JS 后端对接照表手发即可,记得带 x-vercel-ai-ui-message-stream: v1 头。
• 流式 tool calling 最精巧:参数三段式(start/delta/available),前端要区分"预览态"和"最终态",最终参数只信 available,半截 JSON 别硬 parse。
• 可中断要端到端:AbortController → req.signal → abortSignal 透传给 provider,任一环节断了就是假中断;中断不是错误,已生成内容要保留。
• 缓冲是头号杀手:Nginx proxy_buffering off、响应头 X-Accel-Buffering: no、对 text/event-stream 跳过 gzip、排查 Cloudflare——代码没错但流不动,九成在这。

一句话:真正的流式,是"数据一产生就到用户屏幕上",中间任何一层攒一下,体感就垮。 把这条链路从浏览器一直摸到 provider,每一层都不缓冲、能中断、会重连,你的 AI 应用才配叫"实时"——这一项的功夫,是区分"能跑的 demo"和"可用的产品"的分水岭。