创建日期：2026-09-08 ｜ 最近更新：2026-09-08 本篇所有「帧/事件节选」与运行输出均为本机真实抓包/真实运行：DeepSeek 同一 key，OpenAI 兼容端点与 Anthropic 兼容端点各 stream:true 一次。模型输出逐次略有差异，帧数/字数随运行变化，属正常。

LLM 流式（SSE）与打字机：OpenAI / Anthropic 怎么把一个字一个字推给你

一句话：你在网页/终端看到的「一个字一个字蹦出来」，是 stream: true 的功劳。 上一篇拆了两家非流式的请求/返回；这篇讲另一半——大模型自回归地边算边发，服务端通过 SSE 长连接把增量一帧帧推给你，客户端边收边渲染就成了打字机。我们把 OpenAI（delta 切片）与 Anthropic（content_block_delta 事件）的真实流都抓出来看，最后给一个能真跑的打字机脚本。

阅读前建议先看首篇：OpenAI 与 Anthropic 消息格式——那里的非流式返回，正是把本文流式增量「拼回去」的结果。

0. 为什么会出现「一个字一个字」？

根子在生成方式：大模型一次只算下一个 token，算完就把它加进上下文再算下一个（自回归）。既然天生是逐 token 产出的，服务端没必要憋到全部生成完再回——它每算出一点立刻 flush：

模式	服务端行为	你看到的效果
非流式（默认）	等整段 `end_turn`，一次性返回完整 JSON	卡几秒，然后「啪」地整段出现
流式（`stream: true`）	边算边发，TCP 长连接不断开	逐 token（视觉上近于逐字）往外蹦

传输通道叫 SSE（Server-Sent Events）：一个不关闭的 HTTP 响应，服务端单向持续往 body 里写文本帧。每帧空行分隔，核心是 data: 行。它不是 WebSocket——不需要双向、不需要协议升级，普通 fetch 拿 response.body 就能读。

为什么是一蹦 2~4 个字而不是严格逐字？SSE 按 token 发帧，中文一个 token 常等于 1 个字或一个小词；人眼合起来就像打字机。

1. 先把「一帧」切开

SSE 流的行协议极简，客户端只要按换行把 data: 载荷抠出来：

data: {"choices":[{"delta":{"content":"我"}}]}
<空行>
data: {"choices":[{"delta":{"content":"是"}}]}
<空行>
...
data: [DONE]

一帧 = 若干行 + 空行（有的服务端只有 data: 行、用换行分隔）；
关键坑：HTTP chunk 可能把一个汉字切成两半发过来，所以解码必须用 TextDecoder(…, { stream: true }) 攒着，见 §5。

2. OpenAI 协议流：一整份 JSON 的「增量切片」

请求只多一个字段——stream: true。返回的每一帧 data: 都是同一份 completion 结构的部分：这次多出来的是 choices[0].delta（对应非流式的 message）。把 delta.content 逐帧拼起来 = 非流式的 message.content。

本机真实流（DeepSeek OpenAI 兼容端点）节选——思考阶段走 delta.reasoning_content、正文阶段才走 delta.content：

#1 content=null reasoning=""          ← 首帧先广播 role=assistant
#2 content=null reasoning="我们"
#3 content=null reasoning="只需要"
...                                  ← 思考帧约 60 帧（此处省略）
#61 content="HTTP" reasoning=null
#62 content="是"
#63 content="用于"
...                                  ← 正文帧到 [DONE]
[DONE]
data 帧总数= 70

读这张节的三个要点：

思考与正文分相位：开头只有 reasoning_content 有货、content 是 null；思考完才切到 content（与首篇非流式 message.reasoning_content / message.content 一一对应）；
客户端拼文本：正文阶段把每个 delta.content 累加即可；finish_reason 一般只在最后带 content 的空帧里出现（或干脆没有，以 [DONE] 收尾）；
流式默认不返回 usage——OpenAI 官方要传 stream_options: { include_usage: true } 才会在最后一帧给，DeepSeek 兼容端点也一样不随流给。

3. Anthropic 协议流：按「内容块生命周期」发事件

Anthropic 的流不是「整份 JSON 的切片」，而是一系列有名字的事件（data.type 区分），且 event: 行同时标类型。真实流的开头两行：

event: message_start
data: {"type":"message_start","message":{"id":"784c5840-…","type":"message","role":"assistant","model":"deepseek-v4-flash","content":[],"stop_reason":null,"stop_sequence":null,"usage":{"input_tokens":97,"cache_creation_input_tokens":0,"cache_read_input_"…（截断）

一次「想一下再答」的完整事件序（本机真实抓包，thinking_delta 仅示前 3 条）：

[message_start]          ← 消息开始，usage 只有 input_tokens
[content_block_start]    block=thinking        ← 思考块开
[ping]                   ← 保活帧，客户端要跳过
[delta thinking_delta]   "We"
[delta thinking_delta]   " need"
[delta thinking_delta]   " answer"
[delta signature_delta]                        ← 思考块收尾的签名
[content_block_stop]     ← 思考块关
[content_block_start]    block=text            ← 正文块开
[delta text_delta]       "p"
[content_block_stop]     ← 正文块关
[message_delta]          stop_reason=end_turn  ← 收尾事件带结束原因
[message_stop]           ← 整条消息结束

对应关系：思考、正文各占一个 content_block，块内用 content_block_delta 的 thinking_delta / text_delta 逐 token 喂内容——这跟非流式返回里 content: [{type:'thinking'},{type:'text'}] 的块数组是同构的。收尾在 message_delta（拿 stop_reason）。

实测还踩到一个真坑：max_tokens 会把 thinking 一起算进去。让模型「想太久」时，它会把 token 预算全烧在 thinking_delta 上，最后 message_delta 直接 stop_reason=max_tokens，一个字的正文都没出。给推理模型留 max_tokens 要给足。

4. 一张表对照

维度	OpenAI `chat.completions`	Anthropic `messages`
开启	请求体加 `stream: true`	请求体加 `stream: true`
流的形态	一整份 completion 的增量切片	一系列命名事件
增量字段	`choices[0].delta.content` / `delta.reasoning_content`	`content_block_delta.delta.text` / `.thinking`
思考/正文怎么分	相位：思考在 `reasoning_content`、正文在 `content`	块：thinking 块与 text 块各一个 `content_block`
结束信号	`data: [DONE]`（finish_reason 偶尔出现在末帧）	`message_delta`（带 `stop_reason`）→ `message_stop`
保活帧	无	中间夹 `ping`，客户端要跳过
usage	默认不给，需 `stream_options.include_usage`	收尾 `message_delta`/官方给摘要（兼容端点不一）
累积即非流式	拼 `delta.content` == `message.content`	拼 text 块的 `text_delta` == `text` 块

5. 完整可跑的打字机脚本（真实输出）

下面的 stream.mjs（位于博客配套实验仓库 ioi/mini-agent/）用 Node 原生 fetch 把两家流都打一遍：正文 process.stdout.write 逐字输出——你在终端跑会看到字一个一个蹦。连接配置自动吃 Claude Code 的 settings（见 claude-env.mjs）；若脱离该仓库，把首行 import './claude-env.mjs'; 删掉、自行 export 三个 ANTHROPIC_* 即可。

node stream.mjs "只回复两个字：pong。"

// stream.mjs —— 流式(SSE)「打字机」demo：OpenAI / Anthropic 两种协议各打一发。
import './claude-env.mjs';   // 自动读入 Claude Code settings 里的 ANTHROPIC_*（免手动 export）

// 0. 配置：Anthropic 用 ANTHROPIC_*；OpenAI 端点是“摘掉 /anthropic 尾巴”的兄弟地址
const ANT_BASE = (process.env.ANTHROPIC_BASE_URL || 'https://api.deepseek.com/anthropic').replace(/\/+$/, '');
const KEY = process.env.ANTHROPIC_AUTH_TOKEN || '';
const MODEL = process.env.ANTHROPIC_MODEL || 'deepseek-v4-flash';
const OAI_BASE = (process.env.OPENAI_BASE_URL || ANT_BASE.replace(/\/anthropic$/i, '') || 'https://api.deepseek.com').replace(/\/+$/, '');
// OpenAI 兼容端点只认不带 [1M] 变体的纯模型名，剥掉括号后缀
const OAI_MODEL = (process.env.OPENAI_MODEL || MODEL).replace(/\[\w+\]$/, '');

export const config = { oai: { base: OAI_BASE, model: OAI_MODEL }, ant: { base: ANT_BASE, model: MODEL }, hasKey: !!KEY };
if (!config.hasKey) { console.error('缺少 key：先跑 node claude-env.mjs 看连接配置'); process.exit(1); }

// 1. 通用 SSE 读取：把响应体按行切成 data 帧
async function* sseData(res) {
  const reader = res.body.getReader();
  const dec = new TextDecoder();
  let buf = '';
  for (;;) {
    const { done, value } = await reader.read();
    if (done) break;
    buf += dec.decode(value, { stream: true });   // 关键：防多字节汉字被 chunk 拦腰截断
    let i;
    while ((i = buf.indexOf('\n')) >= 0) {
      const line = buf.slice(0, i).trim();
      buf = buf.slice(i + 1);
      if (line.startsWith('data:')) yield line.slice(5).trim();
    }
  }
}

// 2. Anthropic 协议流（内容块事件：thinking_delta / text_delta）
export async function streamAnthropic(userText, { maxTokens = 2048 } = {}) {
  const res = await fetch(`${ANT_BASE}/v1/messages`, {
    method: 'POST',
    headers: { 'content-type': 'application/json', 'x-api-key': KEY, 'anthropic-version': '2023-06-01' },
    body: JSON.stringify({ model: MODEL, stream: true, max_tokens: maxTokens,
      messages: [{ role: 'user', content: userText }] }),
  });
  if (!res.ok) throw new Error(`Anthropic HTTP ${res.status}: ${(await res.text()).slice(0, 300)}`);

  const events = {};                    // 按事件类型计数
  let text = '', think = '', inThinking = false, stop = null;
  for await (const data of sseData(res)) {
    const ev = JSON.parse(data);
    events[ev.type] = (events[ev.type] ?? 0) + 1;
    if (ev.type === 'content_block_start') inThinking = ev.content_block?.type === 'thinking';
    else if (ev.type === 'content_block_delta') {
      const d = ev.delta ?? {};
      if (d.type === 'text_delta') { text += d.text; process.stdout.write(d.text); }  // 正文：逐字打出
      else if (d.type === 'thinking_delta') { think += d.thinking; }                  // 思考：只攒不打
      // signature_delta 等：忽略
    } else if (ev.type === 'message_delta') stop = ev.delta?.stop_reason ?? stop;
  }
  return { text, think, stop, events };
}

// 3. OpenAI 协议流（delta.reasoning_content 思考 → delta.content 正文）
export async function streamOpenAI(userText, { maxTokens = 2048 } = {}) {
  const res = await fetch(`${OAI_BASE}/v1/chat/completions`, {
    method: 'POST',
    headers: { 'content-type': 'application/json', authorization: `Bearer ${KEY}` },
    body: JSON.stringify({ model: OAI_MODEL, stream: true, max_tokens: maxTokens,
      messages: [{ role: 'user', content: userText }] }),
  });
  if (!res.ok) throw new Error(`OpenAI HTTP ${res.status}: ${(await res.text()).slice(0, 300)}`);

  let text = '', think = '', finish = null, frames = 0;
  for await (const data of sseData(res)) {
    if (data === '[DONE]') break;
    frames++;
    const ch = JSON.parse(data).choices?.[0] ?? {};
    const d = ch.delta ?? {};
    if (d.content) { text += d.content; process.stdout.write(d.content); }       // 正文：逐字打出
    else if (d.reasoning_content) think += d.reasoning_content;                   // 思考：只攒不打
    if (ch.finish_reason) finish = ch.finish_reason;
  }
  return { text, think, finish, frames };
}

// 4. 主流程
const Q = process.argv.slice(2).join(' ') || '只回复两个字：pong。';
console.log(`问题：${Q}\n`);
console.log(`===== Anthropic 协议  ${ANT_BASE}/v1/messages   (stream:true) =====`);
const ant = await streamAnthropic(Q);
console.log();   // 结束打字机行
console.log(`正文：${ant.text}`);
console.log(`思考(隐藏) ${ant.think.length} 字 | stop_reason=${ant.stop} | 事件统计=${JSON.stringify(ant.events)}`);

console.log(`\n===== OpenAI 兼容协议  ${OAI_BASE}/v1/chat/completions   (stream:true) =====`);
const oai = await streamOpenAI(Q);
console.log();
console.log(`正文：${oai.text}`);
console.log(`思考(隐藏) ${oai.think.length} 字 | finish_reason=${oai.finish} | data 帧=${oai.frames}（含 [DONE]）`);

本机真实输出（终端里那两行 pong 是逐字打出来的；文件里是回声与统计）：

问题：只回复两个字：pong。

===== Anthropic 协议  https://api.deepseek.com/anthropic/v1/messages   (stream:true) =====
pong
正文：pong
思考(隐藏) 300 字 | stop_reason=end_turn | 事件统计={"message_start":1,"content_block_start":2,"ping":1,"content_block_delta":85,"content_block_stop":2,"message_delta":1,"message_stop":1}

===== OpenAI 兼容协议  https://api.deepseek.com/v1/chat/completions   (stream:true) =====
pong
正文：pong
思考(隐藏) 352 字 | finish_reason=stop | data 帧=94（含 [DONE]）

怎么读这份输出：

两边正文都只打了 pong，但帧数是两种协议：Anthropic 是 85 个 content_block_delta（思考+正文的增量事件），OpenAI 是 94 个 data 帧——模型在两端「想」的长短这次不同，数值只对本次运行成立；
思考被隐藏（只计数不打字），所以你只看到正文蹦——想连思考一起展示，把 thinking_delta / reasoning_content 也 process.stdout.write 出去即可；
stop_reason=end_turn（Anthropic）与 finish_reason=stop（OpenAI）是两家的「正常结束」信号；Anthropic 那次 ping 出现 1 次，验证了保活帧确实存在。

6. 常见坑清单

#	坑	解法
1	HTTP chunk 把一个汉字拦腰截断，解码乱码	`TextDecoder(…, { stream: true })` 攒 buffer，别每 chunk 单独 decode
2	Anthropic 流里夹 `ping` 保活帧	按 `data.type` 分发，`ping` 直接跳过
3	只该渲染正文，思考不能直接给用户	Anthropic 只写 `text_delta`、跳过 `thinking_delta`；OpenAI 只写 `delta.content`
4	推理模型把 `max_tokens` 烧在思考上，正文被截断	`max_tokens` 给足（实测 256 会被 thinking 吃光、`stop=max_tokens`）
5	想拿 usage 却没数	OpenAI 流默认不给，传 `stream_options: { include_usage: true }`；别指望兼容端点每个收尾帧都带完整 usage
6	`data:` 载荷偶发不是 JSON（如 `[DONE]`、注释）	逐行抠出后先判 `=== '[DONE]'`，再 `JSON.parse`

7. 关联与下一步

配套可跑代码：ioi/mini-agent/stream.mjs（含自动装配连接配置的 claude-env.mjs），node stream.mjs 直接看打字机；
把本系列的 delta 拼回去就是首篇的非流式结构；类型化角度，流式事件比非流式更适合用 zod 的 discriminatedUnion('type')（Anthropic 事件正是按 type 区分的联合）——下一篇可写「收发两家 → zod 归一化成自己的内部消息类型」的适配层，届时 Agent 的 streamAnthropic 就能和frontend-agent 系列的工具循环拼在一起。

动手

cd ioi/mini-agent && node stream.mjs "一句话介绍 HTTP"，在终端亲眼看逐字输出；
改代码把 thinking_delta/reasoning_content 也 process.stdout.write，对比「思考也打字」的效果；
把 maxTokens 调小到 64，跑一遍观察哪边先截断、stop_reason 变成什么。

自测

为什么模型输出会「一个字一个字」出现？流式相对非流式的网络差别是什么？
OpenAI 流里思考与正文分别走哪个 delta 字段？结束信号是什么？
Anthropic 流按什么维度发事件？思考与正文在事件序里是怎么区分开的？
为什么 TextDecoder 要 stream: true？
max_tokens 在推理模型流式下有什么坑？

0. 为什么会出现「一个字一个字」？​

1. 先把「一帧」切开​

2. OpenAI 协议流：一整份 JSON 的「增量切片」​

3. Anthropic 协议流：按「内容块生命周期」发事件​

4. 一张表对照​

5. 完整可跑的打字机脚本（真实输出）​

6. 常见坑清单​

7. 关联与下一步​

动手​

自测​