AI API 流式响应(SSE/Streaming)实战指南：从原理到断线重连全解析

如果你的 AI 应用等用户看着转圈等半分钟才出结果,体验一定很差。流式响应(Streaming) 是几乎所有生产级 AI 应用的标配——用户能实时看到文字一个个"打字机式"地出现,大幅降低了感知延迟。但流式接口的实现比一次性获取结果复杂得多,尤其是接入中转站之后,断流、乱码、超时等问题层出不穷。本文系统讲清楚原理和排坑方法。

流式响应的底层原理:SSE

主流 AI API 的流式输出基于 Server-Sent Events (SSE) 协议,本质是 HTTP 长连接 + 分块传输编码(chunked transfer encoding)。服务端不断往同一个连接里推送小片段数据,客户端逐块解析。

一个典型的 SSE 数据流长这样:

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

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

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

data: [DONE]

每一行 data: xxx 是一个事件,末尾的 data: [DONE] 标志流结束。客户端需要按行解析 JSON,并拼接 delta.content 字段来还原完整文本。

Python 实现示例

from openai import OpenAI

client = OpenAI(
    api_key="sk-xxxx",
    base_url="https://你的中转站域名/v1"
)

stream = client.chat.completions.create(
    model="gpt-5.6-sol",
    messages=[{"role": "user", "content": "写一首关于秋天的诗"}],
    stream=True
)

full_text = ""
for chunk in stream:
    delta = chunk.choices[0].delta.content
    if delta:
        full_text += delta
        print(delta, end="", flush=True)

print("\n\n完整内容:", full_text)

如果不用官方 SDK,直接用 requests 手动处理 SSE 流也很常见,尤其是对接一些 SDK 支持不完善的中转站:

import requests
import json

resp = requests.post(
    "https://你的中转站域名/v1/chat/completions",
    headers={"Authorization": "Bearer sk-xxxx"},
    json={"model": "gpt-5.6-sol", "messages": [...], "stream": True},
    stream=True
)

for line in resp.iter_lines():
    if not line:
        continue
    line = line.decode("utf-8")
    if line.startswith("data: "):
        payload = line[6:]
        if payload == "[DONE]":
            break
        chunk = json.loads(payload)
        delta = chunk["choices"][0]["delta"].get("content", "")
        print(delta, end="", flush=True)

Node.js 实现示例

import OpenAI from "openai";

const client = new OpenAI({
  apiKey: "sk-xxxx",
  baseURL: "https://你的中转站域名/v1",
});

const stream = await client.chat.completions.create({
  model: "gpt-5.6-sol",
  messages: [{ role: "user", content: "写一段产品介绍文案" }],
  stream: true,
});

let fullText = "";
for await (const chunk of stream) {
  const delta = chunk.choices[0]?.delta?.content || "";
  fullText += delta;
  process.stdout.write(delta);
}

前端直接消费流式接口

如果不想经过后端中转,前端也可以用 fetch + ReadableStream 直接消费 SSE(需要 API Key 走后端代理以免泄露,但流式转发逻辑可以放前端):

const response = await fetch("/api/chat-proxy", {
  method: "POST",
  body: JSON.stringify({ messages }),
});

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

while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  buffer += decoder.decode(value, { stream: true });

  const lines = buffer.split("\n\n");
  buffer = lines.pop(); // 保留未完整的最后一段
  for (const line of lines) {
    if (line.startsWith("data: ") && line !== "data: [DONE]") {
      const json = JSON.parse(line.slice(6));
      const delta = json.choices[0]?.delta?.content || "";
      appendToUI(delta);
    }
  }
}

中转站场景下的常见坑

1. 反向代理缓冲导致"假流式":很多中转站或者你自己部署的 Nginx/反代如果没关闭 buffering,会把整个流缓存完再一次性发给客户端,表现上就像没开流式一样。排查方法:在 Nginx 配置里加 proxy_buffering off;,同时确认中转站官方文档是否提示需要额外的 header(如 X-Accel-Buffering: no)

2. 中途断流没有重试机制:长文本生成时网络抖动很容易导致流中断,生产环境必须实现"断点续传"或至少"失败后完整重试",不能让用户看到半截文字就卡死

3. 中文多字节字符被截断导致乱码:如果按固定字节数切分 chunk 而不是按行/按事件解析,可能把一个中文字符的 UTF-8 编码切在中间,导致解析报错或乱码。务必用行边界(\n\n)而不是固定长度来切分

4. 超时设置过短:流式请求的"总耗时"可能远超普通请求,如果客户端或网关设置了固定的整体超时(比如 30 秒),长回复可能被强制掐断。应该设置基于"多久没有新数据"的空闲超时,而不是总耗时超时

5. first token 延迟(TTFT)是重要指标:流式场景下用户体验的关键不是总耗时,而是"多久出现第一个字"。测试中转站质量时,建议专门监控 TTFT 指标,而不仅仅看总响应时间

断线重连的实战模式

import time

def stream_with_retry(client, model, messages, max_retries=3):
    for attempt in range(max_retries):
        try:
            full_text = ""
            stream = client.chat.completions.create(
                model=model, messages=messages, stream=True
            )
            for chunk in stream:
                delta = chunk.choices[0].delta.content
                if delta:
                    full_text += delta
                    yield delta
            return  # 正常结束
        except Exception as e:
            print(f"第{attempt+1}次尝试失败: {e}")
            if attempt < max_retries - 1:
                time.sleep(2 ** attempt)  # 指数退避
            else:
                raise

注意:重试时如果已经输出了部分内容,简单粗暴地整段重来会导致重复文字。生产环境更成熟的方案是把已生成的部分文字作为上下文,让模型"接着写"而不是从头再来,但这需要模型支持续写且成本会略微增加。

总结

流式响应看起来只是加一个 stream: True 参数,但要在生产环境稳定运行,需要处理反代缓冲、断线重试、多字节字符切分、超时策略等一系列细节。选择中转站时,建议实测其流式接口在弱网环境下的表现,而不只是看官方文档写了"支持 streaming"就直接上线。

想查看更多AI API开发实战技巧,可以参考站内 AI API 重试与熔断机制最佳实践 和 baseURL 切换实战与代码示例大全,或直接前往 openairouter.net 查看各中转站的流式接口稳定性评测和速度排行榜。