拆解 OpenAI 的 SSE 数据流

2026-09-26 curl 8.5.0 / Python 3.12 下面每个字节都是真抓的

"stream": true 这几个字符，落到线路上就是：一个普通 HTTP 响应，Content-Type: text/event-stream，body 永远不结束，服务器一行一行往里写。没有新协议、没有长连接握手、没有分帧协商——它就是 SSE（Server-Sent Events），WHATWG HTML 规范第 9.2.5 节那个老古董。

麻烦在于规范只定义了「行」，各家实现却对「一条事件什么时候算完整」各有默契。这一篇把它拆开：先讲规范里真正规定的东西，再给一组实测抓包，把「文档没写但一定会撞到」的部分钉死。

1 · SSE 的线格式：只有 5 个字段

规范里，事件流是以 \n / \r\n / \r 结尾的行，每行 field: value。字段一共四个加注释：

三条被反复写错的规定：

• 空行才是事件边界，不是换行。收到一行 data: 时事件还没「完成」，必须等一个空行才 dispatch。
• data: 后面那一个空格是协议的一部分——规范说「值若以单个 U+0020 开头则去掉它」。所以 data:{...}（无空格）同样合法，你的解析要能吞掉两种。
• 字段值里出现 : 无所谓，只有第一个 : 是分隔符；field alone（无冒号）表示空值字段。

2 · 真实字节长什么样

一次最小请求（max_tokens: 48），响应头：

HTTP/2 200
content-type: text/event-stream
cache-control: no-cache
x-request-id: a5d7756c6f0674ee5509ca9e85d06a0b
server: cloudflare

同一请求强制走 HTTP/1.1 时，body 是标准的 chunked 编码；走 HTTP/2 时没有 Transfer-Encoding，SSE 事件被塞进 DATA 帧里：

HTTP/1.1 200 OK
Content-Type: text/event-stream
Transfer-Encoding: chunked          # h1：SSE 事件 不是 一个事件一个 chunk，别拿 chunk 当边界
Cache-Control: no-cache
Connection: keep-alive

结论：传输层的分帧和 SSE 的事件边界毫无关系。h1 的 chunk、h2 的 DATA 帧、TLS record、TCP 段，全都是「字节」，不是「事件」。

body 的前 4 行，用 cat -A 显示（$ = LF，^M 会出现如果是 CRLF）：

data: {"id":"39b6f4a051c64151b5656d699d245302","object":"chat.completion.chunk","c$
   "created":1790411641,"model":"agnes-3.0-flash","choices":[{"index":0,"delta":{$
data: {"id":"39b6f4a051c64151b5656d699d245302","object":"chat.completion.chunk","$
   "choices":[{"index":0,"delta":{"content":"Server"}}]}$
$

本次响应共 52 条 data: 行，形状是「一条 JSON + 一个空行」。统计出来的事实：

3 · chunk 的 JSON 形状，以及「别按字符串猜」

三行完整样本，分别是第一行、中间行、倒数第三行：

// 首块：只给你 role，content 是空串
{"id":"39b6f4…","object":"chat.completion.chunk","created":1790411641,"model":"agnes-3.0-flash",
 "choices":[{"index":0,"delta":{"role":"assistant","content":""}}]}

// 中间块：delta 里只有 content，一个词
{"id":"39b6f4…","object":"chat.completion.chunk","created":1790411641,"model":"agnes-3.0-flash",
 "choices":[{"index":0,"delta":{"content":"Server"}}]}

// 收尾块：delta 变空对象，finish_reason 出现
{"id":"39b6f4…","object":"chat.completion.chunk","created":1790411641,"model":"agnes-3.0-flash",
 "choices":[{"index":0,"delta":{},"finish_reason":"length"}]}

然后是最后一块，注意它和上面三块字段顺序不同：

{"id":"39b6f4…","created":1790411641,"model":"agnes-3.0-flash","object":"chat.completion.chunk",
 "choices":[{"index":0,"delta":{}}],
 "usage":{"completion_tokens":48,"prompt_tokens":80,"total_tokens":128,
          "completion_tokens_details":{"reasoning_tokens":0}}}

4 · 三个不同的「结束」，别混

流里有三种收尾信号，语义各不相同：

实测三种 finish_reason 都出现过：

"finish_reason":"stop"     # 正常说完（max_tokens=16，模型只用了 6 token）
"finish_reason":"length"   # 撞到 max_tokens=48 被截断
"finish_reason":"tool_calls" # 见第 7 节

所以判断「这个回答完整吗」不能只看有没有 [DONE]——[DONE] 只代表传输结束，length 代表内容被截了。两个维度都要读。

5 · 读到字节 ≠ 读到事件

这是所有 bug 的源头。用 read(37) 这种奇怪的长度去读同一个流（故意不用行迭代器）：

read#1  37B  t=0.49s  b'data: {"id":"314c8ee02b5e4707a54957a5'
read#2  37B  t=0.49s  b'e0ebc14f","object":"chat.completion.c'
read#3  37B  t=0.49s  b'hunk","created":1790411718,"model":"a'
read#4  37B  t=0.49s  b'gnes-3.0-flash","choices":[{"index":0'
read#5  37B  t=0.50s  b',"delta":{"role":"assistant","content'
read#6  37B  t=0.50s  b'":""}}]}\n\ndata: {"id":"314c8ee02b5e47'   ← 一次读里含完整事件 + 下一条的开头
read#7  37B  t=0.50s  b'07a54957a5e0ebc14f","object":"chat.co'

看 read#6：它同时包含「上一条事件的空行结尾」和「下一条事件的前半截」。任何「一次 read 就是一条 data」的写法都是错的。

还有两个连环坑，都是实测出来的：

坑 A：read(n) 会阻塞到攒够 n 字节

# Python urllib，read(8192)，一个总共约 2 万字节的流
read#1  8192B  t=1.06s   尾部 b'id":"ea88955236a642b3a24'     ← 停在 JSON 中间
read#2  8192B  t=1.56s   尾部 b'hat.completion.chunk","c'      ← 停在单词中间
read#3  3592B  t=1.73s   尾部 b'ns":0}}}\n\ndata: [DONE]\n\n'   # 直到 EOF 才拿到剩下的

每次都刚好 8192 字节 ⇒ read(n) 是「攒够才返回」。一个短回答（<8KB）你会一直等到流结束才看到第一个字节，打字机效果当场消失。要「来多少读多少」就用 read1() / os.read() / reader.read()（fetch）这类单次原始读。

坑 B：一次读会把一个汉字劈成两半

# read1(19)：故意用奇怪长度，检查截断处的字节尾巴
@read20  … {"content":"ä¸            ← 三个字节的中文字符，只送到前两个
@read30  … {"content":"ï¼
@read40  … {"content":"äº

所以「decode() 每块字节」必然偶发 UnicodeDecodeError 或者吞字。要么像我上面那样先攒字节、按 \n\n 切出完整事件、再整块解码（这样最稳，因为事件边界一定在字符边界之后），要么用流式解码器（TextDecoder({stream:true})、Python codecs.getincrementaldecoder("utf-8")()），并保留未消费的余量。

正确骨架只有一个：

buffer = b""
while True:
    chunk = conn.read1(65536)         # 单次原始读，别用 read(n) 攒够才返回
    if not chunk:
        break                         # EOF：buffer 里可能还剩残包，见第 6 节
    buffer += chunk
    while b"\n\n" in buffer:          # 只按空行切事件
        raw, buffer = buffer.split(b"\n\n", 1)
        data_lines = []
        for line in raw.split(b"\n"):
            line = line.rstrip(b"\r")             # 容忍 CRLF
            if not line or line.startswith(b":"):  # 空行 / 心跳注释
                continue
            field, _, value = line.partition(b":")
            if value.startswith(b" "):
                value = value[1:]                 # 协议规定：去掉那一个空格
            if field == b"data":
                data_lines.append(value)
        if data_lines:
            handle(b"\n".join(data_lines))        # 完整事件才解码/parse

6 · [DONE] 没来就 EOF

我起了一个 max_tokens: 400 的长流，然后让客户端提前关掉（head -c 240），看最后落到磁盘的字节：

curl: (23) Failure writing output to destination
bytes=240
--- 尾部 60 字节 ---
ntent":""}}]}$
$
data: {"id":"97b5d283c2794677b3e9d6a33c5052d8     ← 断在这里，无 LF、无 }、无 [DONE]

两个必须处理的分支：

• buffer 里剩下没有空行结尾的残包：不能 parse，只能丢掉或上报。
• EOF 之前没见过 [DONE]：这是异常终止，不是完成。把它当完成，你的 UI 就会停在半句话上还显示「✓ 已完成」。

至于原因：SSE 规范自带重连（浏览器 EventSource 会带着 Last-Event-ID 回来），但 Chat Completions 不发 id: 字段（第 2 节实测 0 条），所以没有任何续传锚点——断了只能重发整个请求，或者接受残缺。

7 · tool_calls 是分片到达的字符串

强制模型调一次工具，抓到的完整流（8 块）：

1  "delta":{"role":"assistant","content":""}
2  "delta":{"tool_calls":[{"id":"call_e44af56fcb6c4fccb2a3903a",
                          "function":{"arguments":"","name":"get_weather"}, … "index":0}]}
3  "delta":{"tool_calls":[{"function":{"arguments":"{"},"type":"function","index":0}]}
4  "delta":{"tool_calls":[{"function":{"arguments":"\"city\": \"北京\""},"type":"function","index":0}]}
5  "delta":{"tool_calls":[{"function":{"arguments":"}"},"type":"function","index":0}]}
6  "delta":{},"finish_reason":"tool_calls"
7  "usage":{"completion_tokens":22,"prompt_tokens":283,"total_tokens":305}
8  data: [DONE]

规则：

• id 和 function.name 只在第 2 块出现一次，后续块只有 arguments 片段。
• arguments 是被切碎的一段 JSON 字符串——"{" + "\"city\": \"北京\"" + "}"。每一块单独 json.loads 都会失败，必须按 tool_calls[].index 分组拼接，全部拼完再解析。
• 并行工具调用靠 index（不是数组下标）区分；finish_reason:"tool_calls" 才是「参数完整了」的信号。
• 所以流式 UI 里想「实时显示正在调用哪个工具」，只能显示 name + 累积参数原文，不能提前 parse。

8 · n>1：delta 会交错

一次 "n": 2 的流，把每块的 choices[] 打出来：

idx1:{"role": "assistant", "content": "   ← 第一块居然是 index 1
idx1:{"content": "苹果"}
idx0:{"content": "苹果"}
idx0:{} fr=stop
idx0:{} fr=None  | usage

两个结论：到达顺序不保证按 index，而且每块只携带发生变化的那一路（不是完整数组）。所以必须用 choices[].index 做 key 维护 N 个累加器，别拿 choices[0] 当「当前这一路」。

9 · 时间维度：TTFB 抖 4 倍，中途还会静默 1.3 秒

同一个脚本、同一个模型，连续 5 次请求，记录每次原始读的到达时间：

#1  reads= 48  TTFB= 0.62s  total=1.08s   gaps: median=10ms   max=  30ms
#2  reads= 29  TTFB= 0.51s  total=2.02s   gaps: median=10ms   max=  581ms
#3  reads= 36  TTFB= 2.02s  total=3.59s   gaps: median= 9ms   max=1326ms
#4  reads= 86  TTFB= 0.65s  total=1.88s   gaps: median=15ms   max=  30ms
#5  reads= 89  TTFB= 1.27s  total=2.16s   gaps: median=10ms   max=  41ms
# 另外单独抓到过一次极端值：TTFB = 14.37s，之后 8 条 data 全挤在 56ms 内到达

这张表能推出四条实操结论：

• 超时要设「空闲」而不是「总计」。中位间隔 10ms，但 P-max 是 1326ms，最长一次首字节 14.37s。任何「总耗时 > 10s 就掐」的策略都会杀掉一个服务端完全正常的请求；反过来「60s 内没新字节才算死」又太宽，用户已经盯着空白屏一分钟了。合理值：首字节 15–30s、块间 20–30s，并且可取消。
• 「逐 token 平滑流出」是幻觉。14.37s 那次说明中间任何一层（网关排队、nginx proxy_buffering、Cloudflare、SDK 批量 flush）都能把流攒起来一次性给你。想要打字机手感，就在客户端本地按固定节奏回放已收到的文本，而不是把网络节奏直接当渲染节奏。
• 间隔抖动 > 100 倍 ⇒ 任何「按 chunk 数估算进度百分比」的逻辑都不可信；进度只能显示「已输出 N 字」。
• 心跳是有用的：链路上有 idle timeout 的代理（云负载均衡默认 60s）会在长静默时掐连接，服务端发 : ping 保活。你的解析器必须把 : 开头的行当空气，而不是当错误。

10 · 错误不会走 SSE

最容易写出「解析器炸在 HTML/JSON 错误页上」的地方。实测两类失败：

# 不存在的模型 —— 注意是 503，不是 4xx
status=503 content-type: application/json; charset=utf-8
{"error":{"code":"model_not_found","message":"分组 default 下模型 … 无可用渠道（distributor）","type":"…_error"}}

# 无效令牌
status=401 content-type: application/json; charset=utf-8
{"error":{"code":"","message":"无效的令牌","type":"…_error"}}

也就是说：

• 进 SSE 循环之前先判 status 和 content-type。只有 text/event-stream 才继续按行解析，否则按普通 JSON 错误处理。
• 错误体形状各家不同（有的 error.code 是空串，有的多 type），别指望固定 schema。
• 200 + application/json 也可能出现——网关把流式请求降级成了一次性返回。看到 object: "chat.completion"（而不是 chat.completion.chunk）就说明它没在流。
• HTTP 200 之后中途断开、或流里出现 data: {"error": …}（部分网关的用法）都是「传输成功但业务失败」，必须单独处理。

11 · 厂商扩展字段：白名单解析

各家会在 chunk 里塞自己的东西，最常见的几类：

规则一句话：解析时只认你需要的字段，其余全部忽略；反过来说，任何「chunk 一定有这些 key」的假设都不成立——第 3 节那个字段顺序变化就是证据。

12 · 一个能直接抄的客户端（两份都真跑过）

下面两段不是伪代码：把它们从本页抠出来直接跑，同一个「必须调用工具」的请求，Python 3.12 与 Bun 都返回 finish="tool_calls" + arguments={"city": "北京"}（参数在流里被切成 { / "city": "北京" / } 三片，见第 7 节）；纯文本请求分别返回 finish="stop" 与 finish="length"（后者撞上 max_tokens: 40），usage 只在最后一块出现。要点全在前面各节，这里只是把它们缝起来。

Python（只用标准库）

import json
import urllib.request


def chat_stream(payload, base, key, idle_timeout=90):
    req = urllib.request.Request(
        base.rstrip("/") + "/chat/completions",
        data=json.dumps({**payload, "stream": True,
                         "stream_options": {"include_usage": True}}).encode(),
        headers={"Authorization": f"Bearer {key}",
                 "Content-Type": "application/json",
                 "Accept": "text/event-stream"})

    content = ""
    tool_calls = {}          # index -> {id, name, arguments(str)}
    usage, finish, saw_done, buf = None, None, False, b""

    with urllib.request.urlopen(req, timeout=idle_timeout) as r:   # 空闲超时
        if "event-stream" not in (r.headers.get("content-type") or ""):
            return json.loads(r.read().decode() or "{}")            # 错误或被降级成一次性 JSON

        while True:
            chunk = r.read1(65536)                                   # 单次原始读，见第 5 节坑 A
            if not chunk:
                break
            buf += chunk
            while b"\n\n" in buf:
                raw, buf = buf.split(b"\n\n", 1)
                lines = []
                for line in raw.split(b"\n"):
                    line = line.rstrip(b"\r")
                    if not line or line.startswith(b":"):             # 心跳 / 空行
                        continue
                    if not line.startswith(b"data:"):
                        continue                                       # 本 API 不发 event:
                    v = line[5:]
                    lines.append(v[1:] if v.startswith(b" ") else v)   # 去掉那一个空格
                if not lines:
                    continue
                data = b"\n".join(lines).decode("utf-8")               # 完整事件才解码
                if data.strip() == "[DONE]":
                    saw_done = True
                    continue
                obj = json.loads(data)
                if obj.get("usage"):
                    usage = obj["usage"]
                for ch in obj.get("choices") or []:                    # 可能压根没有 choices
                    delta = ch.get("delta") or {}
                    if delta.get("content"):
                        content += delta["content"]
                    for tc in delta.get("tool_calls") or []:
                        acc = tool_calls.setdefault(
                            tc.get("index", 0), {"id": None, "name": "", "arguments": ""})
                        acc["id"] = tc.get("id") or acc["id"]
                        fn = tc.get("function") or {}
                        if fn.get("name"):
                            acc["name"] += fn["name"]                  # name 也可能分片
                        if fn.get("arguments"):
                            acc["arguments"] += fn["arguments"]         # 只拼字符串，先别 parse
                    if ch.get("finish_reason"):
                        finish = ch["finish_reason"]

    if not saw_done:
        raise ConnectionError("stream ended without [DONE]")           # 第 6 节：这是异常

    parsed = {i: {"id": v["id"], "name": v["name"],
                  # 只有到这里 arguments 才是合法 JSON
                  "arguments": json.loads(v["arguments"] or "{}")}
              for i, v in tool_calls.items()}
    return {"content": content, "tool_calls": parsed, "usage": usage, "finish": finish}

TypeScript / Bun（fetch + ReadableStream）

type Streamed = {
  content: string;
  toolCalls: Record<number, { id: string | null; name: string; arguments: unknown }>;
  usage: unknown;
  finish: string | null;
};

export async function chatStream(payload: Record<string, unknown>,
                                 base: string, key: string): Promise<Streamed> {
  const res = await fetch(`${base.replace(/\/$/, "")}/chat/completions`, {
    method: "POST",
    headers: { Authorization: `Bearer ${key}`, "Content-Type": "application/json" },
    body: JSON.stringify({ ...payload, stream: true,
                           stream_options: { include_usage: true } }),
  });

  const ct = res.headers.get("content-type") ?? "";
  if (!res.ok || !ct.includes("text/event-stream")) {        // 第 10 节：先判，再进循环
    throw new Error(`HTTP ${res.status} ${ct}: ` +
                    JSON.stringify(await res.json().catch(() => null)));
  }

  const reader = res.body!.getReader();
  const decoder = new TextDecoder();                  // 必须配 {stream:true}，见第 5 节坑 B
  let buf = "", sawDone = false, content = "", usage: unknown = null, finish: string | null = null;
  const calls = new Map<number, { id: string | null; name: string; arguments: string }>();

  for (;;) {
    const { value, done } = await reader.read();      // read() 本来就是「来多少给多少」
    if (done) break;
    buf += decoder.decode(value, { stream: true });

    let sep: number;
    while ((sep = buf.search(/\r?\n\r?\n/)) >= 0) {   // LF 和 CRLF 的空行都算
      const block = buf.slice(0, sep);
      buf = buf.slice(sep).replace(/^\r?\n\r?\n/, ""); // 吃掉分隔符本身

      const dataLines: string[] = [];
      for (const line of block.split(/\r?\n/)) {
        if (!line || line.startsWith(":")) continue;   // 空行 / 心跳
        if (!line.startsWith("data:")) continue;       // 本 API 不发 event:
        const v = line.slice(5);
        dataLines.push(v.startsWith(" ") ? v.slice(1) : v);   // 去掉那一个空格
      }
      if (!dataLines.length) continue;
      const data = dataLines.join("\n");
      if (data.trim() === "[DONE]") { sawDone = true; continue; }  // 不是 JSON，先特判

      const obj = JSON.parse(data);
      if (obj.usage) usage = obj.usage;
      for (const ch of obj.choices ?? []) {              // usage chunk 里可能是 []
        const d = ch.delta ?? {};
        if (d.content) content += d.content;
        for (const tc of d.tool_calls ?? []) {
          const i = tc.index ?? 0;                    // 按 index 分组，不是数组下标
          const acc = calls.get(i) ?? { id: null, name: "", arguments: "" };
          if (tc.id) acc.id = tc.id;
          if (tc.function?.name) acc.name += tc.function.name;
          if (tc.function?.arguments) acc.arguments += tc.function.arguments;
          calls.set(i, acc);                           // 只拼字符串
        }
        if (ch.finish_reason) finish = ch.finish_reason;
      }
    }
  }
  if (!sawDone) throw new Error("stream ended without [DONE]");

  const toolCalls = Object.fromEntries([...calls].map(([i, v]) =>
    [i, { ...v, arguments: JSON.parse(v.arguments || "{}") }]));   // 到这儿才是合法 JSON
  return { content, toolCalls, usage, finish };
}

curl 侧记两件事：流式一定加 -N/--no-buffer，否则你会以为「没输出」；想验证到达节奏，curl -N … | while IFS= read -r l; do printf '%s %s\n' "$(date +%T.%N)" "$l"; done。想直接看有没有 [DONE]：… | tail -c 200 | cat -A。

13 · Checklist

• 先判 status + content-type，再进 SSE 循环；application/json 是错误/降级。
• 按空行切事件，不按换行；buffer 里可能同时有「完整事件 + 半条事件」。
• 容忍 data: 后有无空格、CRLF、: 心跳、多行 data。
• [DONE] 不是 JSON，先特判；EOF 前没见到它就是异常。
• 只 json.loads，绝不按字符串前缀猜；字段顺序会变。
• 累加器按 choices[].index 分组（n>1 会交错），未知字段一律忽略。
• arguments 是切碎的句子，拼完再 parse；finish_reason:"tool_calls" 才代表齐了。
• 区分传输结束（[DONE]）和内容截断（finish_reason:"length"）。
• 用空闲超时，不是总超时；准备收 burst，打字机在本地回放。
• 没有 id: 字段 ⇒ 无法续传；重连 = 重发，注意幂等和二次计费。
• 别用 read(n)（攒够才返回），用单次原始读；完整事件再解码，防止汉字被劈开。
• usage chunk 的 choices 可能是空数组；遍历一律写 obj.get("choices") or []。

· 相关

• Pi 的 RPC 模式与 JSON 事件流把 agent 当后端进程时，事件流长什么样