流式输出 SSE：原理与各语言实现

大模型 API 的流式输出基于 Server-Sent Events（SSE）协议：服务端生成一个 token 就立即推送一帧，客户端边收边渲染，用户无需等待完整回复即可开始阅读。对话类、长文生成类应用默认应开启流式。

SSE 协议原理

SSE 是 HTTP 长连接上的单向文本流，服务端以 text/event-stream 内容类型持续发送格式化帧：

data: {"choices":[{"delta":{"content":"你"},"index":0}]}

data: {"choices":[{"delta":{"content":"好"},"index":0}]}

data: [DONE]

每帧以 data: 开头，两个空行表示帧边界，[DONE] 标记流结束。与 WebSocket 相比，SSE 更轻量——只需普通 HTTP，天然支持浏览器 EventSource API 和服务端反向代理，不需要协议升级。

一个真实踩坑：为什么”流式”跑起来却像非流式

自己接过流式接口的人多半遇到过这种诡异现象：代码明明写了 stream=True，浏览器控制台里 EventSource 也确实连上了，可页面上就是没有逐字跳出的效果——等了几秒，字一下子全冒出来，跟没开流式一模一样。

根因十有八九是中间代理把响应体整体缓冲了。SSE 依赖的是”服务端一有数据就立刻 flush 出去”，但 Nginx 默认开着 proxy_buffering on，会把上游（你的后端）返回的内容攒到一定大小或者等上游关闭连接才转发给客户端，这就把一条本该逐帧推送的流硬生生变成了”攒够了再发”。排查方法很简单：直接用 curl -N 打你的接口，如果 curl 端就已经是一次性吐出全部内容而不是逐行滚动，问题就出在你的后端到用户之间的某一层反代，而不是代码本身。

修法是在对应 location 块里显式关闭缓冲，并在响应头里补一个 Nginx 私有头告诉它不要缓冲这个响应：

location /api/ {
    proxy_pass http://127.0.0.1:3000;
    proxy_buffering off;
    proxy_read_timeout 120s;
    add_header X-Accel-Buffering no;
}

X-Accel-Buffering: no 这个响应头是 Nginx 专门识别的信号，哪怕全局配置里 proxy_buffering 没关，只要上游返回了这个头，Nginx 也会对这一次响应单独关闭缓冲。如果你的服务部署在有 CDN 或者多层反代（比如阿里云 SLB + Nginx + 应用服务器）的环境里，这三层都要确认没有偷偷开缓冲，漏了任何一层都会导致”代码没问题但效果不对”。

何时开启流式？

场景	建议	原因
对话界面（chatbot）	开启 stream	用户即见即读，体验更好
长文/报告生成	开启 stream	防止超时，逐步展示进度
批量处理 / 后端管道	不开启	非流式更易处理、结构化
函数调用（function calling）	可选	流式下需拼接 tool_calls delta
测试/调试	不开启	非流式响应更直观

Python 流式实现

使用 openai SDK

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ.get("OPENAI_BASE_URL", "https://api.lidayun.com/v1"),
)

def stream_chat(prompt: str, model: str = "gpt-4o-mini") -> str:
    full_text = ""
    stream = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        stream=True,
        max_tokens=1024,
    )
    for chunk in stream:
        delta = chunk.choices[0].delta.content or ""
        print(delta, end="", flush=True)   # 边收边打印
        full_text += delta
    print()   # 换行
    return full_text

result = stream_chat("用 Python 实现二分查找，并解释思路")

这段代码里 delta = chunk.choices[0].delta.content or "" 这个 or "" 不是随手写的防御式代码，是真会用到——流式响应里有些 chunk 只携带 role 字段（比如流的第一个 chunk 通常是 {"role": "assistant"}，内容为空），或者结尾那个带 finish_reason 的 chunk 的 content 字段本身就是 None。如果不加这个兜底，直接 full_text += delta 会在某次 chunk 上抛出 TypeError: can only concatenate str (not "NoneType") to str，这是新手写流式解析最容易踩的一个空指针坑。

不依赖 SDK，自己用 httpx 解析 SSE 时的编码陷阱

如果你不想引入 openai SDK，想自己用 httpx 或 requests 撸一个流式客户端，切记要按”行”读取，而不是按固定字节数切块。原因是 UTF-8 编码下一个中文汉字占 3 个字节，如果你用 iter_content(chunk_size=1024) 这种按字节数切分的读法，网络传输的分片边界完全可能落在一个汉字的 3 个字节中间，导致你拿到的这一块数据末尾是半个汉字，解码时直接抛 UnicodeDecodeError: 'utf-8' codec can't decode byte 0xe4 in position 1023: unexpected end of data。

requests 的 iter_lines()、httpx 的 iter_lines() 之所以能规避这个问题，是因为它们内部按换行符 \n 做切分再交给你，而换行符本身是单字节的 ASCII 字符，不可能出现在某个多字节 UTF-8 字符编码序列的内部——所以只要你的切分基准是”行”而不是”字节数”，就永远不会把一个汉字腰斩。自己写底层 SSE 解析器时，这条是铁律：宁可多缓冲一点，也别按定长字节切。

在 FastAPI 中向浏览器转发 SSE

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json

app = FastAPI()

@app.get("/api/chat")
async def chat_stream(prompt: str):
    async def generator():
        stream = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            stream=True,
        )
        for chunk in stream:
            delta = chunk.choices[0].delta.content or ""
            if delta:
                data = json.dumps({"content": delta}, ensure_ascii=False)
                yield f"data: {data}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(generator(), media_type="text/event-stream")

JavaScript 流式实现

Node.js（openai SDK）

import OpenAI from "openai";

const client = new OpenAI({
  apiKey: process.env.OPENAI_API_KEY,
  baseURL: process.env.OPENAI_BASE_URL ?? "https://api.lidayun.com/v1",
});

async function streamChat(prompt) {
  const stream = await client.chat.completions.create({
    model: "gpt-4o-mini",
    messages: [{ role: "user", content: prompt }],
    stream: true,
  });

  let fullText = "";
  for await (const chunk of stream) {
    const delta = chunk.choices[0]?.delta?.content ?? "";
    process.stdout.write(delta);   // 边收边输出
    fullText += delta;
  }
  console.log();
  return fullText;
}

await streamChat("解释 JavaScript 事件循环");

浏览器端：用 EventSource 接收后端 SSE

<!-- 前端不应直接调 AI API，通过自己的后端中转 -->
<script>
function startStream(prompt) {
  const output = document.getElementById("output");
  output.textContent = "";

  const url = `/api/chat?prompt=${encodeURIComponent(prompt)}`;
  const source = new EventSource(url);

  source.onmessage = (event) => {
    if (event.data === "[DONE]") {
      source.close();
      return;
    }
    const { content } = JSON.parse(event.data);
    output.textContent += content;   // 边收边渲染
  };

  source.onerror = () => {
    source.close();
    console.error("SSE 连接断开");
  };
}
</script>

断线重连与心跳：生产环境绕不开的一步

Demo 阶段大家都图省事，EventSource 连上就不管了。真上了生产你会发现连接三天两头断——公司出口代理、家庭宽带的 NAT、云厂商的负载均衡，都可能在连接空闲一段时间（常见阈值是 60 秒没有新数据）后悄悄把它掐掉，客户端这边只会收到一个 onerror，既不报具体原因也不会自动重连。

服务端这边的应对是发心跳：每隔 10～15 秒推一条 SSE 注释行（以 : 开头，EventSource 会自动忽略这种行，不会触发 onmessage），维持连接”看起来一直有数据在传”，绕开中间代理的空闲回收判断。客户端这边则该做指数退避重连，而不是断了就狂刷新：

let retryDelay = 1000;

function connectWithRetry(prompt) {
  const source = new EventSource(`/api/chat?prompt=${encodeURIComponent(prompt)}`);

  source.onopen = () => {
    retryDelay = 1000;   // 连上了，重置退避时间
  };

  source.onerror = () => {
    source.close();
    setTimeout(() => connectWithRetry(prompt), retryDelay);
    retryDelay = Math.min(retryDelay * 2, 30000);   // 翻倍，封顶 30s
  };

  return source;
}

这里的关键是 retryDelay 要在重连成功后重置，否则一次短暂抖动之后，即便网络已经恢复，你的客户端还会按上次断线时累积的长间隔傻等，用户体验反而更差。

流式是否影响计费，超时又该怎么设

开流式会不会多花钱？ 不会。主流网关的计费口径是按 prompt token 数 + completion token 数算总量，流式只是把 completion 部分拆成一帧一帧推给你，账单上 usage 字段体现的还是生成完成后的汇总值，不会因为多传了几十个 HTTP 帧就多算钱（以各平台官方计费文档为准，截至 2026-06 主流网关都是这个口径）。但有一点要注意：如果你自己代理层的超时设得太短，在模型还没生成完之前就把连接掐断，模型侧该收的费已经产生了，你这边却没收全内容，这笔钱不会因为你没收到而退——所以配置超时时间不能图省事乱设一个小数字。

该怎么设超时？ 流式场景要分清”连接超时”和”读超时”两个概念，二者含义完全不同：

超时类型	含义	流式场景建议值
connect timeout	建立 TCP/TLS 连接的等待上限	5～10s，长了通常是网络不通
read timeout（每次读取的间隔）	两次收到数据之间允许的最大空闲时间	至少 60s，模型排队繁忙时首 token 延迟（TTFT）可能到 5～10s

真实报错举例：Python 下 httpx.ReadTimeout 或者 Node 下 ETIMEDOUT，八成不是网络挂了，而是你把 httpx.Timeout 或者 axios 的 timeout 参数设成了三五秒——这个值放在非流式请求上没问题，但流式请求的”读超时”衡量的是相邻两帧之间的间隔而不是总时长，只要模型还在持续吐字，连接就不该被判定为超时，读超时该给得比非流式宽松得多。

429 限流怎么处理？ Error: Request failed with status code 429 表示当前 key 的 QPS 或并发数超限，流式的长连接本身也占一次请求配额。正确姿势是读响应头里的 Retry-After（如果平台返回了这个头，就按它给的秒数等待），没有就用前面讲的指数退避重试，绝不能收到 429 立刻重试——那样大概率还是 429，而且频繁重试容易被平台判定为异常调用行为，触发更严格的临时限制。

并发连接数与心跳的一个真实坑

如果你的产品是多用户在线的聊天类应用，每个用户对话时都会占用一条常驻 SSE 长连接。这里有一个容易被忽略的性能坑：如果给每条连接的心跳都各自开一个 setInterval，用户到了几千并发，光是心跳定时器本身的唤醒开销就会明显拖慢 CPU——因为每个定时器到期都要被事件循环单独调度一次。

更稳的做法是维护一个全局的心跳广播定时器，统一每 15 秒触发一次，遍历当前所有活跃连接推一条心跳注释，而不是每条连接各自计时。经验值供参考：单个 2 核 4G 的实例，在合理管理心跳和 buffer 的前提下，扛住三五千条并发 SSE 长连接问题不大；但如果没做集中调度，连接数一上千就可能先被心跳定时器本身拖垮，而不是被真正的业务流量压垮。

常见问题

流式下如何获取总 token 用量？ 部分平台会在最后一个 chunk 的 usage 字段返回用量，但并非所有平台都支持。可在流结束后用 tiktoken 估算，或在后端记录每次请求的 prompt token 数。

SSE 连接超时怎么处理？ 网关/反向代理（如 Nginx）默认有读超时，需将 proxy_read_timeout 调大（建议 120s+）；客户端侧可监听 onerror 事件触发重连。

流式输出能和 function calling 一起用吗？ 可以，但 tool_calls 会以 delta 形式分帧推送，需要在客户端拼接完整的 JSON 参数后再解析，实现稍复杂，可参考 openai SDK 文档中的 streaming with tools 示例。具体拆解一下拼接逻辑：tool_calls 的 delta 是按数组下标 index 分片下发的，同一个函数调用的 name 和 arguments 字符串可能要分几十个 chunk 才能收完整，你必须在本地按 tool_calls[i].index 做累加拼接（arguments 字段是逐字符甚至逐 token 追加的字符串，不是完整 JSON），一直等到某个 chunk 的 finish_reason 变成 "tool_calls"，才代表这个函数的参数拼完了，这时候才能对拼出来的字符串做 JSON.parse。如果你在中途任意一帧就尝试解析，大概率会报 SyntaxError: Unexpected end of JSON input（JS）或者 json.decoder.JSONDecodeError: Expecting value（Python）——这几乎是流式 + function calling 组合下最高频的报错，根因永远是”参数还没收完就急着解析”。

浏览器直接调 AI API 的 SSE 有跨域问题吗？ 有。生产上应通过自己的后端中转，一方面规避 CORS，一方面保护 API key 不暴露在前端。

---

更多接入知识见大模型 API 接入完全指南与接入教程专题。Python 与 Node.js 完整非流式示例分别见Python 调用大模型 API 完整示例和Node.js 接入大模型 API。需要稳定的多模型流式接入？申请力达云聚合 API 内测。