从零实现AI流式对话：一篇搞懂SSE ReadableStream与JSON截断处理

写在前面

最近在做一个AI聊天应用，需求很明确：调用大模型接口时，要让回复内容像ChatGPT那样一个字一个字地“蹦”出来，而不是转圈圈等半天后一次性显示完整答案。

说实话，刚开始我以为这很简单——不就是stream: true嘛，配上前端EventSource就完事了。结果真正上手才发现，坑比想象中多得多。尤其是JSON截断、双重退出逻辑和协议解析这三个点，让我熬夜到凌晨三点。

这篇文章就把我从踩坑到填坑的全过程记录下来，包含最容易被忽略的底层原理，希望能帮你省下几个小时的调试时间。

---

一、先看数据长啥样：服务器返回的原始流示例

在写代码之前，我们必须先搞懂服务器到底吐出来的是什么。下面是一次真实请求中，response.body 流经 TextDecoder 解码后，按顺序收到的原始文本块（已格式化展示）：

1.1 正常情况：一条完整的 data 行

 复制代码data: {"id":"chatcmpl-abc123","object":"chat.completion.chunk","created":1721088000,"model":"deepseek-v4-flash","choices":[{"index":0,"delta":{"content":"从前"},"finish_reason":null}]}

1.2 粘包情况：一次收到多条 data 行

 复制代码data: {"id":"1","choices":[{"index":0,"delta":{"content":"有座"}}]}
data: {"id":"2","choices":[{"index":0,"delta":{"content":"山，"}}]}
data: {"id":"3","choices":[{"index":0,"delta":{"content":"山上"}}]}

1.3 拆包情况（最坑）：一条JSON被物理切成了两半

第一次 reader.read() 拿到：

 复制代码data: {"id":"1","choices":[{"index":0,"delta":{"con

第二次 reader.read() 拿到：

 复制代码tent":"有座庙"},"finish_reason":null}]}

1.4 结束信号：[DONE]

 复制代码data: [DONE]

1.5 解析流程图解

 复制代码原始二进制流（Uint8Array）
        ↓ TextDecoder.decode()
文本字符串（可能包含多个 n）
        ↓ .split('n')
数组：["data: {...}", "data: {...}", ""]
        ↓ .filter(line => line.startsWith('data: '))
有效行：["data: {...}", "data: {...}"]
        ↓ .slice(6) 去掉前缀
裸JSON字符串：["{...}", "{...}"]
        ↓ JSON.parse()（可能失败，进buffer）
JavaScript对象 → 提取 delta.content
        ↓ 累加拼接
最终页面渲染内容

---

二、核心流程全貌（先看图，再读代码）

 复制代码用户点击提交
    ↓
fetch发起POST请求（stream: true）
    ↓
服务器返回ReadableStream二进制流（TCP分片传输）
    ↓
getReader()获取读取器，进入while轮询
    ↓
await reader.read() 嘬一口数据（拿到上面示例中的某个chunk）
    ↓
TextDecoder解码二进制 → 文本字符串
    ↓
【关键】按 n 分割成多行（解决TCP粘包问题）
    ↓
过滤出以 "data: " 开头的行
    ↓
去掉 "data: " 前缀 → 得到裸JSON字符串
    ↓
尝试 JSON.parse() 
    ├─ 成功 → 提取 delta.content → 拼接到页面（增量渲染）
    └─ 失败 → 存入buffer，等下一块拼上再解析（处理拆包截断）
    ↓
遇到 data: [DONE] → 主动退出循环（业务结束）
    ↓
reader.read() 返回 done: true → 被动退出循环（连接关闭）

---

三、代码实战：每一行都是知识点

3.1 基础配置与状态管理

 复制代码import { ref } from 'vue'const question = ref('讲一个中国龙的故事');
const content = ref('');
const stream = ref(true);  // 开关，方便对比流式/非流式

这里用的是Vue3的ref，但核心逻辑和框架无关，你用React的useState、原生JS都行。

3.2 发起请求：别忘了stream参数

 复制代码const response = await fetch(endpoint, {
  method: 'POST',
  headers: {
    'Content-Type': 'application/json',
    Authorization: `Bearer ${import.meta.env.VITE_DEEPSEEK_API_KEY}`
  },
  body: JSON.stringify({
    model: 'deepseek-v4-flash',
    messages: [{ role: 'user', content: question.value }],
    stream: stream.value  // 关键！告诉服务器我要流式
  })
});

关键点：stream: true这个参数是告诉LLM服务端——“别等生成完再返回，生成多少给我多少”。

3.3 流式处理核心（带完整注释，对应上面数据示例的解析逻辑）

 复制代码if (stream.value) {
  content.value = '';
  const reader = response.body?.getReader();
  const decoder = new TextDecoder();
  let done = false;      // 【退出条件1】开关变量
  let buffer = '';       // 存放被截断的“半截JSON行”
  
  while (!done) {
    // 嘬一口，拿到 Uint8Array 二进制块
    const { value, done: doneReading } = await reader.read();
    done = doneReading;  // 【退出条件2】流读取完毕（底层EOF）
    
    //  把上一轮没解析完的buffer拼到当前chunk最前面（解决拆包截断）
    const chunkValue = buffer + decoder.decode(value);
    buffer = '';  // 拼完后立即清空，用完即弃
    
    //  灵魂操作：按 n 分割，解决TCP粘包 
    // 对应数据示例：将 "data: {...}ndata: {...}" 切成 ["data: {...}", "data: {...}"]
    const lines = chunkValue.split('n')
      .filter(line => line.startsWith('data: '));
    
    for (const line of lines) {
      // 切掉 "data: " 头，裸JSON登场
      // 对应数据示例："data: {"id":"1"...}" → "{"id":"1"...}"
      const incoming = line.slice(6);
      
      //  退出条件1：服务端发送的显式结束信号 
      if (incoming === '[DONE]') {
        done = true;
        break;
      }
      
      try {
        // 解析裸JSON字符串 → JavaScript对象
        const data = JSON.parse(incoming);
        // 提取增量补丁 delta.content
        const delta = data.choices[0].delta.content;
        if (delta) {
          content.value += delta;  // 增量追加，逐字渲染
        }
      } catch (e) {
        // JSON解析失败（被截断了，如上面示例的 "data: {"con"）
        // 存到buffer下轮再战，补上后续的 "tent":"有座庙"..."
        buffer = `data: ${incoming}`;
      }
    }
  }
}

---

四、三大灵魂拷问（理解这些，才算真正掌握）

4.1 为什么必须按 n 分割？直接 JSON.parse 不行吗？

这是SSE协议规范 + TCP流特性的双重约束。

1. SSE协议硬规定：大模型厂商都遵循Server-Sent Events规范，要求每条消息以data:开头，行尾用n（或rn）分隔。服务器发送的是“文本行流”，而不是“JSON数组流”。

2. TCP粘包/拆包现实：TCP是流式协议，没有“消息边界”。你调用一次reader.read()，可能收到：

   • 拆包：半条数据（data: {"con）——如示例1.3
   • 粘包：三条数据挤在一起（data: {A}ndata: {B}ndata: {C}）——如示例1.2

   如果不用n切分，你根本无法从data: {"con"}和tent":"hi"}}这种乱流中找回边界。split('n')的本质，是把“不可靠的字节流”还原为“可靠的逻辑消息帧” 。

4.2 返回的 delta 到底是什么？为什么不是全量 content？

delta 是  “增量补丁” （Incremental Patch），而不是“完整快照”。

• 非流式返回的message.content是完整的、最终版的答案。
• 流式返回的delta.content是相对于上一条消息新增的那几个字符。

结合数据示例：

• 第1个chunk：delta: "从前" → 页面显示“从前”
• 第2个chunk：delta: "有座" → 页面显示“从前有座”
• 第3个chunk：delta: "山" → 页面显示“从前有座山”

设计成增量而非全量的原因：

1. 节省带宽：重复传输完整的“从前”+“从前有座”是巨大的浪费。
2. 降低延迟：增量包体积小，传输更快。
3. 前端只需做字符串拼接（content.value += delta），逻辑极简。

4.3 为什么有两个退出条件（done 和 [DONE]）？不重复吗？

这是底层协议（TCP/HTTP）  与应用层协议（SSE）  的双重保险，不仅不重复，而且缺一不可。

退出机制	来源	触发时机	对应数据示例	如果不处理会怎样
doneReading === true	ReadableStream底层读取器（HTTP/TCP层）	服务器主动关闭HTTP连接，或网络中断	无特定数据，突然读不到任何字节了	即使收到[DONE]，若连接没关，循环会永远等待reader.read()，页面卡死
incoming === '[DONE]'	LLM服务端业务层（SSE协议）	大模型生成完最后一段，主动发送结束标记	示例1.4：data: [DONE]	若LLM生成完但没发[DONE]（或延迟发送），即使连接没断，你也不知道回答完了，页面会一直转圈

执行顺序通常是：LLM生成完毕 → 发送最后一条data: [DONE]（业务结束）→ 约几百毫秒后，服务器关闭TCP连接（底层结束）。

你的代码必须同时处理这两种情况：

• 碰到[DONE]立即跳出，提前渲染完成态（不用傻等TCP挥手）。
• 万一网络异常导致没收到[DONE]，连接断了，doneReading也能兜底退出，避免死循环。

---

五、Buffer截断处理详解（最容易被忽略的神器）

这是整个方案里最考验功底、最容易被忽略、但最关键的地方，也是上面示例1.3（拆包）的解决方案。

5.1 问题复现（对应示例1.3）

LLM服务端返回的数据长这样：

 复制代码data: {"id":"xxx","choices":[{"index":0,"delta":{"content":"从前"}}]}
data: {"id":"xxx","choices":[{"index":0,"delta":{"content":"有座"}}]}
data: {"id":"xxx","choices":[{"index":0,"delta":{"content":"山"}}]}
data: [DONE]

每一行都是一个完整的JSON，直接JSON.parse()没问题。

但实际情况往往没这么理想。由于网络传输、TCP分片、服务端缓冲区等因素，你收到的chunk可能长这样：

 复制代码data: {"id":"xxx","choices":[{"index":0,"delta":{"con
tent":"从前有座山"}}]}

看到没？{"con 和 tent":"从前有座山"}}]} 被分到了两个chunk里。

如果直接按行分割并尝试JSON.parse()，第一行解析必然失败。

4.2 buffer的解决方案

buffer的核心逻辑就是 “这行解析不了，我先存着，等下一块数据来了拼在一起再试” 。

 复制代码try {
  const data = JSON.parse(incoming);
  // 解析成功，正常处理
} catch (e) {
  // 解析失败，把这一行原样存到buffer里
  buffer = `data: ${incoming}`;
}

下一轮循环时：

 复制代码const chunkValue = buffer + decoder.decode(value);
// 把上一轮没解析完的拼到新数据前面，再重新分割

说白了就是：解析失败不丢弃，而是作为“未完成的拼图碎片”，等下一块数据来了一起拼。

4.3 为什么不能用简单的字符串拼接？

你可能会想：“那我直接把所有chunk拼成一个完整字符串，最后一次性解析不行吗？”

不行。原因有二：

1. 内存问题：如果回复内容很长（几千字），全拼在内存里会占用大量资源
2. 实时性问题：用户要的是“边生成边显示”，不是“等全生成完再显示”

所以buffer方案的精髓是：只缓存“当前这一行”未完成的部分，不缓存整个响应。

---

六、避坑指南（我替你们踩过的雷）

1. n不止一个：服务器可能一次发多行（示例1.2），必须split('n')后for循环逐行处理。
2. [DONE]可能粘包：它可能和上一条数据挤在同一个chunk里，比如data: {...}ndata: [DONE]，所以循环里判断incoming === '[DONE]'要及时break。
3. decoder.decode(value)可能返回空：流结束时value为undefined，稳妥写法是decoder.decode(value || new Uint8Array())。
4. buffer拼合后记得清空：buffer = ''必须放在拼合语句的下一行，否则会无限累积。
5. data:后面可能有空格：有些服务端实现是data:不带空格，但主流都是data:（带一个空格），slice(6)稳妥。如果怕不兼容，可以用line.replace(/^data:s*/, '')。

---

七、完整可运行代码（Vue3版）

 复制代码

---

八、一图流总结

 复制代码┌─────────────────────────────────────────────────────────────────┐
│                     流式输出终极心法                            │
├─────────────────────────────────────────────────────────────────┤
│   数据样本：data: {"delta":{"content":"你好"}}              │
│  ️  分割术：n 是地桩，split 拆粘包，slice(6) 去头          │
│   容错术：buffer 缓存截断行（如 {"con），拼合后再解析      │
│   数据层：delta 是增量补丁，前端只需累加拼接                │
│   退出层：双重保险——业务 [DONE] + 底层 EOF(doneReading)    │
│   体验层：边生成边渲染，首字延迟从秒级降至毫秒级            │
└─────────────────────────────────────────────────────────────────┘

---