流式输出（SSE）

流式输出解决什么问题

默认情况下，调用聊天补全接口会等待模型把整段回答生成完，再一次性返回。这带来两个问题：

• 首字延迟高：用户要盯着空白等待，交互感差。
• 长回答体验差：回答越长，等待越久。

开启流式输出后，服务端把回答拆成许多小片段，生成一小段就立刻推送一段。客户端可以边收边显示，用户几乎立刻看到文字开始出现。

在 OpenAI 兼容接口里，这个开关就是一个请求体字段：stream: true。底层通常使用 SSE（Server-Sent Events） 作为传输格式。

请求：加上 stream=true

绝大多数聚合站与官方接口都遵循 OpenAI 兼容格式，请求大致如下：

POST /v1/chat/completions
{
  "model": "<你的模型名>",
  "stream": true,
  "messages": [
    { "role": "user", "content": "用一句话解释什么是 SSE" }
  ]
}

关键点：

• stream: true 打开流式。
• 请求头 Accept 一般可保持默认；服务端会返回 Content-Type: text/event-stream。
• 除了多一个字段，请求体和普通调用完全一致。

响应不再是单个 JSON，而是持续到来的文本流。

SSE 分片长什么样

SSE 是一种简单的文本协议：一个事件由若干行组成，字段形式为 字段: 值，事件之间用空行分隔。LLM 接口里，每个数据行通常以 data: 开头：

data: {"choices":[{"delta":{"content":"你"}}]}

data: {"choices":[{"delta":{"content":"好"}}]}

data: {"choices":[{"delta":{"content":"。"}}]}

data: [DONE]

需要理解的几点：

• 每个分片是一段增量：增量文本通常在 choices[0].delta.content，而不是 message.content。
• 分片不保证对齐：一个分片可能切在词中间，甚至切在多字节字符中间（不过正规服务端会保证 UTF-8 完整）。不要假设每个分片是一个完整的词或句子。
• 结束标记是 data: [DONE]：看到它表示流结束，此时应停止读取并关闭连接。
• 可能夹杂非内容事件：例如只带 role 的首个分片、带 finish_reason 的末尾分片、或心跳注释行（以 : 开头），解析时都应容错。

消费流程（概念步骤）

无论用什么语言，消费逻辑都遵循同一套步骤：

1. 发起带 stream: true 的 HTTP 请求，拿到可读的响应流（而非等待整体完成）。
2. 按行读取，累积到缓冲区，按换行符切分。
3. 对每一行：

• 跳过空行与注释行（以 : 开头）。
• 去掉 data: 前缀。
• 如果是 [DONE]，结束。
• 否则解析为 JSON，取出 delta 中的增量文本。

1. 把增量追加到本地累积字符串，并立即渲染给用户。
2. 流结束时，用累积的内容作为最终完整回答。

注意第 2 步：网络并不保证一次 "按行" 到达。一个 TCP 分片可能含有多行，也可能只有半行。因此必须自己做行缓冲，不能假设「一次读取 = 一行」。这是自己手写解析时最常见的 bug 来源。

最小、框架无关的示例（JavaScript / fetch）

下面示例只依赖浏览器或 Node 自带的 fetch 与 TextDecoder，不依赖任何 SDK：

async function streamChat(prompt, onDelta) {
  const res = await fetch("/v1/chat/completions", {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      "Authorization": "Bearer " + API_KEY,
    },
    body: JSON.stringify({
      model: MODEL,
      stream: true,
      messages: [{ role: "user", content: prompt }],
    }),
  });

  const reader = res.body.getReader();
  const decoder = new TextDecoder();
  let buffer = "";
  let full = "";

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

    buffer += decoder.decode(value, { stream: true });

    // 按行切分，最后一段可能是半行，留在 buffer 里
    const lines = buffer.split("\n");
    buffer = lines.pop();

    for (const line of lines) {
      const trimmed = line.trim();
      if (!trimmed || trimmed.startsWith(":")) continue;
      if (!trimmed.startsWith("data:")) continue;

      const payload = trimmed.slice(5).trim();
      if (payload === "[DONE]") continue;

      try {
        const json = JSON.parse(payload);
        const delta = json.choices?.[0]?.delta?.content;
        if (delta) {
          full += delta;
          onDelta(delta);
        }
      } catch {
        // 忽略无法解析的分片，避免因个别异常中断整条流
      }
    }
  }

  return full;
}

几个值得说明的细节：

• decoder.decode(value, { stream: true })：告诉解码器后续还有数据，避免把一个多字节字符在中途误判为损坏。
• buffer = lines.pop()：把最后可能不完整的一行留到下次拼接，这是处理「半行」的关键。
• trim().startsWith("data:")：同时兼容 data: 与 data: 两种写法。
• try/catch：流式场景下个别分片解析失败不应让整个回答作废。

Python 的最小写法

如果不使用 SDK，可以直接读原始字节流：

import json, requests

def stream_chat(prompt, on_delta):
    with requests.post(
        "https://your-endpoint/v1/chat/completions",
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={
            "model": MODEL,
            "stream": True,
            "messages": [{"role": "user", "content": prompt}],
        },
        stream=True,
    ) as r:
        for raw in r.iter_lines():
            if not raw:
                continue
            line = raw.decode("utf-8", errors="ignore").strip()
            if not line.startswith("data:"):
                continue
            payload = line[5:].strip()
            if payload == "[DONE]":
                break
            try:
                delta = json.loads(payload)["choices"][0]["delta"].get("content")
            except (json.JSONDecodeError, KeyError, IndexError):
                continue
            if delta:
                on_delta(delta)

    # 调用方负责累积 full 文本

requests 的 stream=True 与 iterlines() 已经帮你处理了行缓冲，但注意 iterlines() 的切分行为在不同版本下略有差异，生产代码里仍建议保留容错。

常见坑与建议

• 忘记缓冲半行：网络分片与 SSE 行不对齐，必须自己拼接。
• 读错字段：流式用的是 delta.content，非流式用 message.content，两者不同。
• 忽略 [DONE]：不识别结束标记会一直读到连接超时。
• 错误地假设分片是完整词：前端如果按字符逐个追加没有问题；如果按「词」做高亮或翻译，需要先把累积文本再分词。
• 不处理错误分片：一段坏数据不该中断整条流。
• 忘记取消：用户中途退出发送新问题，应调用 AbortController（JS）或关闭响应（Python），否则会残留连接与计费。
• 计费与计量：流式与普通调用在计费口径上通常一致，仍以实际 token 计算。具体规则以你所使用服务的说明为准。

何时不该用流式

• 需要拿到完整结果再做后处理（例如 JSON 结构化输出、需要校验完整对象）时，非流式更简单。
• 后台批处理、无人值守的调用：没有交互体验要求，流式只增加复杂度。

流式是改善交互的手段，不是默认必选项。理解 SSE 分片的形状与消费方式，你就能在需要时把它接得稳、接得对。