跨供应商 LLM API 流式传输实战：SSE 差异、Token 缓冲与切换踩坑指南

所有主流 LLM API 都通过 Server-Sent Events (SSE) 支持流式传输。设置 stream: true，token 就会逐步到达而非一次性返回。原理简单，但 OpenAI、Anthropic、DashScope 和 DeepSeek 各自的 SSE 实现存在足够多的差异，切换供应商——或通过网关路由——会让你的流式客户端以意想不到的方式崩溃。

本指南记录了这些具体差异。我们测试了每个供应商的流式输出，记录了线上协议的真实样貌：事件类型、分块结构、终止信号、流中错误行为和工具调用序列化。如果你正在运行多供应商架构或正在考虑，这就是我们构建自己的流式标准化层时希望存在的参考。

OpenAI 兼容指供应商提供一个 chat-completions 接口，其请求与响应结构与 OpenAI API 契约足够接近——只需替换三个值（API key、base URL、模型名），原来的 OpenAI SDK 调用即可直接工作。最小实践面是POST /v1/chat/completions 带 messages、model， 并返回 OpenAI 形式的流式响应。

SSE 流式传输的真实面貌

四家供应商都返回 Content-Type: text/event-stream 并发送换行分隔的数据块。相似之处到此为止。

OpenAI Chat Completions 格式

OpenAI 的 Chat Completions 流式传输发送不含 event: 字段的 data: 行。每个分块是一个 JSON 对象，包含 object: "chat.completion.chunk" 和一个 choices 数组，其中的 delta 对象持有 role、content、tool_calls 或 refusal——永远只有增量部分，不包含完整消息。来源：OpenAI 流式指南，检索于 2026-07-31。

data: {"id":"chatcmpl-abc","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"role":"assistant","content":""},"finish_reason":null}]}

data: {"id":"chatcmpl-abc","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"你好"},"finish_reason":null}]}

data: {"id":"chatcmpl-abc","object":"chat.completion.chunk","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}

data: [DONE]

关键细节：

• 终止信号：字面字符串 data: [DONE] 表示流结束。这不是合法 JSON。
• 用量统计：仅在传入 stream_options: {"include_usage": true} 时包含。用量分块的 choices 数组为空。
• 限流头：在初始 HTTP 响应头中发送（x-ratelimit-limit-requests、x-ratelimit-remaining-tokens 等），不在流中。来源：Simon Willison 的 LLM API 流式调查，检索于 2026-07-31。

OpenAI 较新的 Responses API 使用类型化语义事件（response.created、response.output_text.delta、response.completed）替代 Chat Completions 分块格式，但 Chat Completions 仍然是 OpenAI 兼容供应商模拟的格式。

Anthropic Messages 格式

Anthropic 在 SSE 中同时使用 event: 和 data: 字段——这是与 OpenAI 的关键区别。流遵循生命周期：message_start → content_block_start → content_block_delta（重复）→ content_block_stop → message_delta → message_stop。来源：Anthropic 流式文档，检索于 2026-07-31。

event: message_start
data: {"type":"message_start","message":{"id":"msg_01X","role":"assistant","content":[],"usage":{"input_tokens":25,"output_tokens":1}}}

event: content_block_start
data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}

event: ping
data: {"type":"ping"}

event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"你好"}}

event: content_block_stop
data: {"type":"content_block_stop","index":0}

event: message_delta
data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":15}}

event: message_stop
data: {"type":"message_stop"}

关键细节：

• 无 [DONE]：Anthropic 以 event: message_stop 终止，没有 data: [DONE] 哨兵。
• Ping 事件：Anthropic 发送 event: ping 保活。只预期 data: 行的客户端会出错。
• 用量分两处：输入 token 在 message_start 中；输出 token 在 message_delta 中。
• 内容块索引：index 字段支持单次响应中的多个内容块（文本、tool_use、thinking）。OpenAI 用 choices[0].index 实现类似但结构不同的目的。
• 限流头：使用 anthropic-ratelimit-* 前缀，不是 x-ratelimit-*。

DashScope（OpenAI 兼容模式）

DashScope 在 https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions 暴露 OpenAI 兼容端点。设置 stream: true 时，响应格式与 OpenAI 的 Chat Completions 分块一致：data: 行加 chat.completion.chunk 对象，以 data: [DONE] 终止。来源：阿里云百炼 OpenAI 兼容性，检索于 2026-07-31。

差异比较微妙：

• 认证头：DashScope 的 OpenAI 兼容模式使用与 OpenAI 相同的 Authorization: Bearer <api-key>。
• 思考/推理 token：调用启用思考的 Qwen 模型（如 qwen3.7-max 设置 enable_thinking: true）时，推理 token 单独流式传输。DashScope 在相同的 delta 结构中用 reasoning_content 字段包裹它们。这个字段不在 OpenAI 规范中。
• 分块大小：在我们的测试中，DashScope 每个 SSE 分块发送的 token 略多于 OpenAI——通常每块 2–4 个 token，而 OpenAI 文本模型每块 1 个 token。这意味着相同输出长度的 HTTP 帧更少，可能影响 UI 中的首 token 到达感知。

DeepSeek 格式

DeepSeek API 明确兼容 OpenAI，同时在 https://api.deepseek.com/anthropic 支持 Anthropic 格式端点。OpenAI 兼容路径的流式传输遵循相同的 data: + chat.completion.chunk + data: [DONE] 模式。来源：DeepSeek API 文档，检索于 2026-07-31。

我们观察到的差异：

• 推理内容：与 DashScope 相同，DeepSeek 在启用思考模式时通过 delta 中的 reasoning_content 字段流式传输推理 token。这是与 DashScope 相同的扩展，不在 OpenAI 规范中。
• 双格式支持：DeepSeek 是唯一一个从同一 API 原生支持 OpenAI 和 Anthropic SSE 格式的第一方供应商。/anthropic 路径返回 Anthropic 风格的 event: + data: 生命周期事件。
• 缓存命中指标：DeepSeek 可能在用量分块中包含 cache_creation_input_tokens 和 cache_read_input_tokens，类似 Anthropic 的提示缓存字段。

Token 缓冲差异

供应商刷新 token 的速率不同。这对用户体验很重要——逐字符渲染的聊天 UI 看起来响应迅速，但 3 秒静默后一次性推送 50 个 token 的 UI 感觉已经挂了。

对于所有供应商的推理模型，预期在第一个可见内容 token 之前会有较长的暂停——模型正在先生成思维链。部分供应商流式传输推理 token（DashScope、DeepSeek 的 reasoning_content）；其他供应商则保留到最终答案开始。

流中错误处理

当 SSE 连接已打开且 token 已开始流出后出现问题时，各供应商的处理方式不同。这是多供应商流式传输中最困难的部分——你的 HTTP 状态码已经是 200。

OpenAI

OpenAI 发送带有 error 字段的最终分块或突然关闭连接。Chat Completions SSE 流中没有标准的错误事件类型。如果模型在生成中途触发内容过滤器，最后一个 delta 可能包含 finish_reason: "content_filter" 而非 "stop"。如果服务器崩溃，TCP 连接直接断开，没有 data: [DONE]——你的客户端必须处理不完整的流。

Anthropic

Anthropic 可以发送 event: error，payload 为 data: {"type":"error","error":{"type":"overloaded_error","message":"..."}}。错误作为标准 SSE 事件到达，监听 event: 类型的客户端可以干净地捕获。随后连接关闭。来源：Anthropic API 错误文档，检索于 2026-07-31。

DashScope

DashScope 的 OpenAI 兼容流式传输遵循 OpenAI 的模式：流中错误不常见，表现为连接断开或格式错误的分块。原生 DashScope API 有更结构化的错误事件，但兼容模式路径用 OpenAI 线上兼容性换取了这一点。

DeepSeek

DeepSeek 的 OpenAI 兼容路径遵循 OpenAI 的错误行为。Anthropic 兼容路径遵循 Anthropic 的错误事件模式。

生产规则：始终实现超时和不完整流检测器。如果收到分块但在超时内从未看到终止信号（data: [DONE] 或 event: message_stop），将响应视为失败并记录部分输出以便调试。

工具调用流式传输：最难的标准化问题

流式工具调用意味着逐字符接收函数名和 JSON 参数。单个供应商就已经很棘手了。跨供应商的差异会成倍增加。

OpenAI

工具调用通过 delta.tool_calls[i].function.name（发送一次）和 delta.tool_calls[i].function.arguments（作为增量字符串片段发送）进行流式传输。你必须拼接 arguments 片段，只在 finish_reason: "tool_calls" 之后解析完整 JSON。多个工具调用可以通过 index 交错。

{"delta":{"tool_calls":[{"index":0,"id":"call_abc","type":"function","function":{"name":"get_weather","arguments":""}}]}}
{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\"lo"}}]}}
{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"cation\":"}}]}}
{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"\"Paris\"}"}}]}}

Anthropic

Anthropic 将工具使用作为 type: "tool_use" 的内容块进行流式传输。content_block_start 事件携带 {"type":"tool_use","id":"toolu_abc","name":"get_weather","input":{}}，然后 content_block_delta 事件携带 {"type":"input_json_delta","partial_json":"..."} 片段。JSON 片段遵循相同的拼接模式，但封装结构与 OpenAI 完全不同。

DashScope 和 DeepSeek

两者在使用 OpenAI 兼容端点时都遵循 OpenAI 的工具调用流式格式。DeepSeek 的 Anthropic 兼容端点遵循 Anthropic 的工具使用流式格式。DashScope 不暴露 Anthropic 兼容端点。

网关挑战：接受任意上游格式并输出一致下游格式的流式网关必须维护每连接状态：哪些内容块已打开、哪些工具调用参数正在组装、原始供应商的终止语义是什么。LLM-Rosetta 项目（阿贡国家实验室，2026 年 4 月）将此形式化为一个包含 10 种流事件类型和有状态上下文管理的 schema。来源：LLM-Rosetta 论文，检索于 2026-07-31。

Content-Type 与连接生命周期

所有供应商使用 HTTP/1.1 分块传输编码或 HTTP/2 数据帧。SSE 连接是长期存活的 HTTP 响应——不是 WebSocket。OpenAI 的 Responses API 还提供 WebSocket 模式用于持久连接，但那是独立的传输机制。

浏览器 EventSource API 无法消费这些端点，因为 EventSource 只支持 GET 请求；LLM API 需要 POST。使用 fetch() 配合流式 body reader 或类似 @microsoft/fetch-event-source 的库。

网关流式标准化：统一多种格式

TheRouter 通过配置的供应商路由流式请求时，网关必须将上游 SSE 标准化为一致的下游协议。我们通过配置的供应商路由 OpenAI 兼容请求，并在实际产品路径支持的地方提供供应商/模型路由和回退。

标准化挑战有三个层次：

1. 事件封装：将 Anthropic 的 event: + data: 生命周期转换为 OpenAI 风格的纯 data: 分块，或反向。这包括映射 message_start → 第一个带 role 的分块，content_block_delta → delta.content，message_delta → 带 finish_reason 的最终分块。
2. 扩展字段：剥离或保留供应商特定字段如 reasoning_content、cache_read_input_tokens 或 Anthropic 的 usage 位置。下游客户端不应因意外字段而崩溃，但也不应依赖只有一个上游供应商发送的字段。
3. 终止语义：确保下游始终收到预期的终止信号，无论上游发送什么。如果上游是 Anthropic 而下游预期 OpenAI 格式，网关必须在处理 event: message_stop 后发出 data: [DONE]。

关于供应商在定价、模型和 API 表面上的更广泛比较，参见我们的 LLM API 供应商对比和 OpenAI 兼容 API 供应商指南。

代码示例：最小化跨供应商流式客户端

这个 Python 客户端同时处理 OpenAI 风格和 Anthropic 风格的 SSE 流。它是有意简化的——生产实现需要重试逻辑、超时处理和适当的背压控制。

import httpx
import json

def stream_openai_compatible(base_url: str, api_key: str, model: str, messages: list):
    """从任何 OpenAI 兼容端点（OpenAI、DashScope、DeepSeek）进行流式传输。"""
    with httpx.stream(
        "POST",
        f"{base_url}/chat/completions",
        headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
        json={"model": model, "messages": messages, "stream": True,
              "stream_options": {"include_usage": True}},
        timeout=60.0,
    ) as response:
        buffer = ""
        for line in response.iter_lines():
            if not line or line.startswith(":"):
                continue
            if line == "data: [DONE]":
                break
            if line.startswith("data: "):
                chunk = json.loads(line[6:])
                delta = chunk.get("choices", [{}])[0].get("delta", {})
                if content := delta.get("content"):
                    yield content
                if reasoning := delta.get("reasoning_content"):
                    yield f"[thinking] {reasoning}"

def stream_anthropic(api_key: str, model: str, messages: list):
    """从 Anthropic 原生 API 进行流式传输。"""
    with httpx.stream(
        "POST",
        "https://api.anthropic.com/v1/messages",
        headers={
            "x-api-key": api_key,
            "anthropic-version": "2023-06-01",
            "Content-Type": "application/json",
        },
        json={"model": model, "messages": messages, "stream": True, "max_tokens": 4096},
        timeout=60.0,
    ) as response:
        for line in response.iter_lines():
            if not line:
                continue
            if line.startswith("event: "):
                event_type = line[7:]
                if event_type == "message_stop":
                    break
                continue
            if line.startswith("data: "):
                data = json.loads(line[6:])
                if data.get("type") == "content_block_delta":
                    delta = data.get("delta", {})
                    if delta.get("type") == "text_delta":
                        yield delta.get("text", "")

关于流式供应商请求中途失败时回退路由的工作方式，参见我们的 LLM API 回退路由指南。

生产踩坑

代理和 CDN 缓冲

反向代理（nginx、Cloudflare、AWS ALB）可能缓冲 SSE 响应，以大批次而非逐 token 交付。这会破坏流式体验。缓解措施：

• nginx：proxy_buffering off; 和 X-Accel-Buffering: no 响应头
• Cloudflare：使用 cf-no-transform 头或在 zone 中禁用响应缓冲
• AWS ALB：ALB 在所有配置中并非都原生支持 SSE——考虑 NLB 或直连

超时配置

SSE 连接是长期存活的。默认 HTTP 客户端超时（30 秒）会杀死长输出的流式请求。分别设置读取超时（每块）和连接超时：

# httpx：总超时 vs 读取超时
timeout = httpx.Timeout(connect=10.0, read=120.0, write=10.0, pool=10.0)

背压

如果客户端处理 token 的速度慢于服务器发送速度，TCP 接收缓冲区会填满，服务器的发送最终会阻塞。对于大多数 LLM API 这不是实际问题——token 生成比网络传输慢——但批量流式或缓存响应可能压倒慢客户端。

保活与重连

SSE 连接可能因网络问题静默断开。Anthropic 的 ping 事件有助于快速检测死连接。对于没有 ping 的 OpenAI 风格流，实现读取超时：如果 N 秒内没有数据到达且流未终止，则重连。注意 LLM API 通常不支持从断点恢复流——断开的连接意味着新请求。

检查清单：接入新供应商的流式端点

集成新 LLM 供应商的流式 API 时，逐项验证：

1. 事件格式：供应商使用纯 data:（OpenAI 风格）还是 event: + data:（Anthropic 风格）？
2. 终止信号：是 data: [DONE]、event: message_stop 还是其他？
3. 内容提取路径：文本增量在哪里？choices[0].delta.content？delta.text_delta.text？还是供应商特定的？
4. 工具调用流式：遵循 OpenAI 的 tool_calls[i].function.arguments 模式还是 Anthropic 的 input_json_delta 模式？
5. 用量报告：流中是否包含用量？仅在请求时包含？在第一个分块、最后一个分块还是两者？
6. 扩展字段：供应商是否添加非标准字段如 reasoning_content、cache_read_input_tokens 或自定义元数据？
7. 错误信号：供应商如何在流中发出错误信号？结构化事件？连接断开？格式错误的分块？
8. 保活机制：供应商是否发送 ping 事件，还是在 token 生成间连接静默？
9. 分块大小：供应商发送单个 token 还是批量？这影响感知延迟。

关于非流式场景下不同供应商的错误和状态码差异，参见我们的 API 限流对比和供应商对比。

FAQ

能否使用浏览器 EventSource API 消费 LLM 流式端点？

不能。EventSource API 只支持 GET 请求。所有 LLM 流式 API 需要 POST。使用 fetch() 配合 ReadableStream reader 或类似 @microsoft/fetch-event-source 的库。供应商集成细节参见我们的 OpenAI 兼容 API 供应商指南。

所有 OpenAI 兼容供应商的流式传输都与 OpenAI 完全一致吗？

基本文本流式路径大多一致。data: 行格式和 data: [DONE] 终止是统一的。差异出现在扩展字段（DashScope 和 DeepSeek 的 reasoning_content）、用量报告行为和分块大小。工具调用流式是差异最微妙的领域。

从 OpenAI 切换到 Anthropic 会怎样？

你的流式客户端会崩溃。Anthropic 使用完全不同的事件生命周期（message_start / content_block_delta / message_stop）和命名的 event: 类型。你需要新的客户端实现或一个将 Anthropic 格式标准化为 OpenAI 兼容分块的网关。

如何处理流式中的推理 token？

DashScope 和 DeepSeek 通过 delta 对象中的 reasoning_content 字段流式传输推理 token。Anthropic 将 extended thinking 作为 type: "thinking" 的独立内容块流式传输。OpenAI 不在 Chat Completions 中流式传输推理 token——它们在内部消费。你的客户端必须优雅地处理 reasoning_content 字段（如果意外则忽略，如果 UI 支持则显示）。

是否存在跨供应商流式标准化的标准？

LLM-Rosetta 项目（2026 年 4 月）提出了包含 10 种流事件类型的 hub-and-spoke IR 作为形式化方法。在实践中，大多数网关和 SDK（LiteLLM、Portkey、TheRouter）实现了各自的标准化层。OpenAI Chat Completions 分块格式是大多数兼容供应商瞄准的事实标准。