AI 对话里的 SSE

聊聊 AI 对话里的 SSE

打开任何一个 AI 对话产品，你都会看到那个熟悉的「打字机效果」——文字一个字一个字地蹦出来。这背后几乎清一色是同一个技术：SSE（Server-Sent Events）。

为什么是 SSE 而不是 WebSocket？浏览器原生的 SSE 为什么生产环境不用？断线了怎么续上？工具调用又是怎么塞进流里的？这篇把这些一次讲清楚。

读完这篇你能收获什么

• ✅ 说清楚 AI 对话为什么选 SSE，以及它和短轮询 / 长轮询 / WebSocket 的取舍
• ✅ 看懂 SSE 协议的真身，知道 data / event / id / 空行分隔各是干嘛的
• ✅ 明白原生 EventSource 为什么上不了生产，以及生产为什么改用 fetch + ReadableStream
• ✅ 会自己解析 SSE 流，避开粘包拆包、中文乱码这几个必踩的坑
• ✅ 掌握异常分类、断线续传、心跳保活，让对话通道扛得住真实网络
• ✅ 理解流式工具调用（Tool Calling）和 Agent 循环，让对话从「会聊天」进化到「会干活」

---

一、为什么 AI 对话偏偏选了 SSE

先看 AI 对话的本质需求。大模型是逐 token 生成的，一段回答算完要好几秒甚至几十秒。如果等全部算完再返回，用户就得盯着空白屏幕干等——体验灾难。

所以核心诉求就八个字：服务端边生成、客户端边渲染。

抓住两个关键词：单向（数据主要是服务端往客户端推）+ 流式（一次请求、持续返回片段）。带着这两点看候选方案：

短轮询   客户端每秒问一次「好了吗」   →  延迟高、空请求爆炸        ❌
长轮询   服务端挂起连接有数据才返回   →  每段都要重建连接、啰嗦    🤔
WebSocket 全双工长连接              →  能力最强，但对单向场景过重  🤔
SSE      基于 HTTP 的服务端单向推送  →  严丝合缝                   ✅

WebSocket 为什么被淘汰？它是全双工的，而 AI 对话里客户端只在开头发一次问题，之后几乎不往上推——全双工是浪费。更麻烦的是它走独立协议（ws://），需要单独握手升级，CDN、鉴权网关、Gzip、监控这些 HTTP 基础设施对它支持不好，断线重连、心跳都得自己造轮子。它适合聊天室、协同编辑这种真·双向高频场景。

而 SSE 就是为「服务端→客户端单向流」而生的，它走标准 HTTP，整套 HTTP 生态无缝复用，协议还极简。

维度	短轮询	长轮询	WebSocket	SSE
数据方向	拉	拉	全双工	服务端→客户端
底层协议	HTTP	HTTP	独立协议	HTTP
实时性	差	中	高	高
基础设施兼容	好	好	差	好
实现复杂度	低	中	高	低
适配 AI 对话	差	凑合	过重	最佳

这就是 OpenAI、Anthropic 以及国内各家大模型 API 全都用 SSE 的原因。

---

二、SSE 协议长什么样

在写代码前，先看它在网络上的真身。响应头必须声明：

Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive

响应体是一行行纯文本，以空行 \n\n 分隔一个个事件：

data: 你好

data: 我是 AI

event: done
data: [DONE]

字段就这几个，记住即可：

• data: —— 数据内容
• event: —— 事件名，客户端可按名分发
• id: —— 事件 ID，断线续传的关键
• retry: —— 断线后重连间隔（毫秒）
• : 开头 —— 注释行，常用来做心跳保活
• 空行 \n\n —— 事件结束的分隔符，解析全靠它

---

三、原生 EventSource：能跑，但上不了生产

浏览器自带了 SSE 客户端 EventSource，代码少到感人：

const es = new EventSource('/chat');
es.onmessage = (e) => appendToUI(e.data);   // 逐段追加，打字机效果
es.addEventListener('done', () => es.close());

浏览器帮你做了解析、自动重连、Last-Event-ID 回传。看起来很美——但一到真实 AI 产品就全是坑：

❌ 只能 GET，不能 POST。 AI 对话要发一大坨 JSON（几千 token 的上下文、model、temperature），GET 只能塞 query string，根本装不下。

❌ 不能自定义请求头。 现代鉴权靠 Authorization: Bearer <token>，EventSource 不支持加 header，你只能把 token 塞进 URL——不安全。

❌ 重连不可控。 浏览器断线后会无脑用原始 URL 重发 GET。对一个生成到一半的对话，重发只会得到一个全新的回答，而不是「接着刚才继续」。这时候自动重连反而是 bug。

❌ 错误信息为零。 onerror 里拿不到 HTTP 状态码，不知道是 401 还是 500 还是断网，没法精细处理。

一句话：EventSource 是「只读一个公开 GET 接口」的玩具。AI 对话需要带鉴权的 POST + 大 body + 可控错误处理，它一个都满足不了。

---

四、生产标准姿势：fetch + ReadableStream

既然原生客户端不够用，那就自己当 SSE 客户端：用 fetch 发请求（想 POST 就 POST、想加 header 就加），再手动读取响应的可读流，自己解析 SSE。

async function chat(messages, onEvent) {
  const resp = await fetch('/v1/chat/completions', {
    method: 'POST',                                   // ✅ 任意方法
    headers: {
      'Content-Type': 'application/json',
      'Authorization': `Bearer ${token}`,             // ✅ 自定义鉴权头
    },
    body: JSON.stringify({ model, messages, stream: true }),  // ✅ 大 body
  });

  if (!resp.ok) throw new Error(`HTTP ${resp.status}`); // ✅ 拿得到状态码

  const reader = resp.body.getReader();
  const decoder = new TextDecoder('utf-8');
  let buffer = '';                                      // ⭐ 关键：跨块缓冲

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

    // 按空行切出「完整事件」，最后一段可能不完整，留到下一轮
    const parts = buffer.split('\n\n');
    buffer = parts.pop();
    for (const raw of parts) {
      const data = raw.replace(/^data: /, '');
      if (data === '[DONE]') return;
      onEvent(JSON.parse(data));                        // 解出真正的 token
    }
  }
}

对比原生 EventSource，用「自己解析协议」这一点代价，换回了全部控制权：

能力	原生 EventSource	fetch + ReadableStream
请求方法	只能 GET	任意（POST）
自定义 Header	❌	✅ 可带鉴权
请求体 body	❌	✅ 大 JSON
HTTP 状态码	❌	✅
重连策略	浏览器托管、不可控	完全自己掌控
取消请求	.close()	AbortController

Vercel 的 ai SDK、@microsoft/fetch-event-source、LangChain 前端，底层全是这个套路。

⚠️ 解析时的三个必踩坑

自己解析 SSE，有三个几乎人人会栽的坑：

1. 一定要用 buffer 处理「粘包 / 拆包」。 TCP 是字节流，read() 返回的一块数据边界完全不可靠——可能是半个事件，也可能是两个半事件。必须累积后按 \n\n 切分，别指望「一个 chunk 就是一个事件」。

第 1 块:  data: 你好\n\ndata: 我  ← 一个半事件（粘包 + 拆包）
第 2 块:  是 AI\n\n              ← 补全上一个

处理: buffer 累积  →  split('\n\n')  →  最后一段(半个)留回 buffer

2. TextDecoder 必须开 stream: true。 一个 UTF-8 中文字符可能横跨两个 chunk，不加它会解出乱码「�」。

3. data: 冒号后约定有一个空格，解析时要去掉。

---

五、生产级容错：异常、断连重续、心跳

真实网络是不可靠的：wifi 切 4G、隧道、服务端超时、token 过期……这些都得兜住。

异常要分类，别一把 catch 了事

if (resp.status === 401) refreshTokenAndRetry();   // 鉴权失败：刷 token
if (resp.status === 429) backoffRetry();           // 限流：退避重试
if (resp.status >= 500) retry();                   // 服务端错误：可重试
if (err.name === 'AbortError') silentStop();       // 用户主动停：静默

有个特别关键的点：「正常结束」的唯一标志是收到约定的结束事件（如 [DONE]），而不是 done === true。 连接断了 done 也会是 true。必须用哨兵标记区分「正常收完」和「半路断了」,否则断线会被当成正常结束，用户就看到一句没说完的话。

断线续传：别让用户重问一遍

天真的重连 = 重发请求 = 模型从头再答一遍，用户看着回答「闪一下重来」，还白烧一次 token。正确做法是接着断掉的地方继续，两种思路：

① SSE 原生的 Last-Event-ID（连接级续传）

服务端给每个事件带 id:，客户端重连时把最后收到的 id 通过 Last-Event-ID 头回传，服务端从该 id 之后继续推。用 fetch 需要手动维护：

let lastEventId = null;
// 解析时记录：if (event.id) lastEventId = event.id;
// 重连时回传：headers['Last-Event-ID'] = lastEventId;

② 业务层续传（更常用）

很多大模型 API 不支持连接级续传（一次生成是原子的）。这时在业务层做：把「已生成的部分」落库或缓存，重连时从库里补齐缺失部分，或作为上下文让模型「继续之前未完成的回答」。

配合指数退避 + 抖动，避免断网时的重连风暴：

const delay = Math.min(1000 * 2 ** attempt, 30_000) * (1 + Math.random() * 0.3);

心跳保活：防止「假死」连接

模型「思考」时可能十几秒不吐字，中间的代理/网关会把这条「空闲」连接判定为死连接掐掉。解法是服务端空闲时定期发注释行心跳，客户端解析时自动忽略：

: ping

: ping

---

六、进阶：让流式对话「会干活」——工具调用

现代 AI 不只是吐文字，还要调用工具（查天气、搜索、执行代码）。难点在于：工具调用的参数本身也是流式、一点点吐出来的。

模型不会一次给你完整的调用，而是分片吐 tool_calls 的 delta：

delta: {tool_calls:[{index:0, id:"call_x", function:{name:"get_weather", arguments:""}}]}
delta: {tool_calls:[{index:0, function:{arguments:"{\"ci"}}]}       ← 参数一个字一个字拼
delta: {tool_calls:[{index:0, function:{arguments:"ty\":\"北京\"}"}}]}
delta: {finish_reason:"tool_calls"}                                 ← 拼完了

函数名一次给全，但 arguments（一个 JSON 字符串）是逐字符拼出来的。所以要按 index 累积，直到 finish_reason: "tool_calls" 才 JSON.parse：

const toolCalls = {};
function handleDelta(delta) {
  if (delta.content) appendToUI(delta.content);        // 普通文本直接渲染
  for (const tc of delta.tool_calls || []) {
    const i = tc.index;
    toolCalls[i] ??= { id: '', name: '', arguments: '' };
    if (tc.id) toolCalls[i].id = tc.id;
    if (tc.function?.name) toolCalls[i].name = tc.function.name;
    if (tc.function?.arguments) toolCalls[i].arguments += tc.function.arguments; // 累加!
  }
}

而工具调用不是终点，它是一个多轮循环——这也是 Agent 的核心：

  ┌─────────────────────────────────────────────┐
  │                                             ↓
用户提问 → 请求模型(stream) → 流式解析 → 收到 finish_reason?
                                              │
                        ┌─────────────────────┴──────────────────┐
                     "stop"                                "tool_calls"
                        │                                        │
                    回答结束 ✅                          本地执行工具拿结果
                                                                 │
                                          把 tool_calls + 工具结果追加进 messages
                                                                 │
                                                          带新 messages 再问一次 ──┘

async function runAgent(messages) {
  while (true) {
    const { text, toolCalls, finishReason } = await streamOnce(messages);
    if (finishReason === 'stop') return text;          // 结束

    messages.push({ role: 'assistant', content: text, tool_calls: toolCalls });
    for (const call of toolCalls) {
      const result = await executeTool(call.name, JSON.parse(call.arguments));
      messages.push({ role: 'tool', tool_call_id: call.id, content: JSON.stringify(result) });
    }
    // 带着工具结果再问一次，模型基于结果继续生成
  }
}

几个实战要点：tool_calls 数组可能一次有多个（并行调用），要并发执行并按 tool_call_id 对齐结果；工具执行失败也要作为 tool 消息回传，让模型自己决定重试还是换方案，别直接抛给用户。

---

写在最后

一条生产级的 AI 对话通道，本质就是三层：

• 选对协议：AI 对话是「单向 + 流式」，SSE 用最小复杂度完美匹配，还白嫖整套 HTTP 生态。
• 可控的客户端：弃用玩具级的 EventSource，改用 fetch + ReadableStream，付出「自己解析协议」的代价换回完全控制权——记得用 buffer 处理粘包拆包。
• 鲁棒的连接管理 + Agent 循环：异常分类、断线续传、心跳保活兜住不稳定网络；再套一层工具调用循环，就从「会聊天」进化成「会干活」。

看懂这三层，你就看懂了市面上所有 AI 对话产品的流式底座。