正确处理流式聊天补全：SSE、用量分块与重试

太长不看: 流式传输 LLM API 响应通过实时发送生成的令牌来降低感知延迟，但需要正确解析服务器发送事件 (SSE) 并显式设置 stream_options 才能接收用量数据。处理连接中断、速率限制和余额不足错误对于构建健壮的实现至关重要。TokShop API 支持此 OpenAI 兼容的流式标准。

如何在流式传输中处理错误和重试？

流式连接很脆弱，可能被网络问题、服务器重启或速率限制中断。为了处理这种情况，需要实现一个重试策略，在失败时重新发送整个请求，因为 API 是无状态的。像 HTTP 402（余额不足）或 HTTP 429（速率限制）这样的特定错误应该被捕获并适当处理，OpenAI SDK 为某些情况提供了内置的重试逻辑。

理解聊天补全的 SSE 格式

服务器发送事件是一个简单的基于文本的协议，每个事件都是一行，以 data: 开头，后跟一个 JSON 负载。对于 OpenAI 兼容的流式传输，流会发送多个事件：

• 内容分块 — data: {"choices":[{"delta":{"content":"Hello"}}]}
• 完成原因 — data: {"choices":[{"delta":{},"finish_reason":"stop"}]}
• 用量分块 — data: {"usage":{"prompt_tokens":10,"completion_tokens":5}}（仅在设置 stream_options={"include_usage": true} 时发送）
• 流结束 — data: [DONE]

一个常见的错误是假设用量分块总是会出现。默认情况下，OpenAI 兼容的 API 在流式模式下会省略用量数据以减少延迟。要获取令牌计数，你必须在请求中显式设置 stream_options。

包含用量跟踪的示例请求

import openai

client = openai.OpenAI(
    api_key="sk-tok-your-key-here",
    base_url="https://tokshop.xyz/v1"
)

response = client.chat.completions.create(
    model="deepseek-v3.2",
    messages=[{"role": "user", "content": "Explain SSE in one sentence."}],
    stream=True,
    stream_options={"include_usage": True}  # <-- 关键
)

没有 include_usage，流会在最终内容分块后结束，你将无法知道使用了多少令牌。有了它，倒数第二个事件（在 [DONE] 之前）会包含用量对象。

正确解析流

OpenAI Python SDK 在你迭代响应时内部处理 SSE 解析。但如果你使用原始 HTTP 或其他语言，则需要逐行解析。

Python：使用 SDK（推荐）

full_content = ""
usage = None

for chunk in response:
    if chunk.choices and chunk.choices[0].delta.content:
        full_content += chunk.choices[0].delta.content
        print(chunk.choices[0].delta.content, end="")
    if chunk.usage:
        usage = chunk.usage
        print(f"\n\nUsage: {usage.prompt_tokens} prompt + {usage.completion_tokens} completion tokens")

原始 HTTP 解析（适用于非 Python 客户端）

import requests
import json

resp = requests.post(
    "https://tokshop.xyz/v1/chat/completions",
    headers={
        "Authorization": "Bearer sk-tok-your-key-here",
        "Content-Type": "application/json"
    },
    json={
        "model": "deepseek-v3.2",
        "messages": [{"role": "user", "content": "Hi"}],
        "stream": True,
        "stream_options": {"include_usage": True}
    },
    stream=True
)

for line in resp.iter_lines():
    if line:
        decoded = line.decode("utf-8")
        if decoded.startswith("data: "):
            payload = decoded[6:]
            if payload == "[DONE]":
                break
            data = json.loads(payload)
            if "choices" in data and data["choices"]:
                delta = data["choices"][0].get("delta", {})
                if "content" in delta:
                    print(delta["content"], end="")
            if "usage" in data:
                print(f"\nUsage: {data['usage']}")

注意：用量分块可能在最后一个内容分块之后但在 [DONE] 之前到达。务必在每个事件中检查 usage 键。

处理流式传输中的错误和重试

流式连接比非流式连接更脆弱。网络波动、服务器重启或速率限制都可能中断流式响应。以下是处理常见场景的方法。

HTTP 402：余额不足

TokShop 使用预付费美元额度。如果你的余额在流式传输过程中耗尽，服务器将关闭连接并返回 HTTP 402 和 JSON 主体 {"error": {"code": "insufficient_balance"}}。你的客户端应该捕获此错误，并提示用户充值或切换到更便宜的模型。

try:
    response = client.chat.completions.create(...)
    for chunk in response:
        # process chunk
        pass
except openai.APIStatusError as e:
    if e.status_code == 402:
        print("Insufficient balance. Visit https://tokshop.xyz/pricing to add credits.")
    else:
        raise

流式传输过程中连接中断

如果 TCP 连接中断（例如，WiFi 故障），你将收到 APIConnectionError。最安全的重试策略是使用相同的消息历史记录重新发送整个请求。这是可行的，因为 API 是无状态的——每个请求都是独立的。

import time
from openai import APIError

def stream_with_retry(client, model, messages, max_retries=3):
    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model=model,
                messages=messages,
                stream=True,
                stream_options={"include_usage": True}
            )
            for chunk in response:
                yield chunk
            return  # success, exit retry loop
        except (APIError, ConnectionError) as e:
            if attempt == max_retries - 1:
                raise
            wait = 2 ** attempt  # exponential backoff: 1s, 2s, 4s
            print(f"Stream failed (attempt {attempt+1}), retrying in {wait}s...")
            time.sleep(wait)

速率限制（HTTP 429）

如果你触达速率限制，服务器会返回 429。OpenAI SDK 默认会自动使用指数退避进行重试（可通过 max_retries 配置）。对于流式传输，同样适用——但请注意，重试会重置整个流，因此你将丢失已接收的任何令牌。

client = openai.OpenAI(
    api_key="sk-tok-...",
    base_url="https://tokshop.xyz/v1",
    max_retries=2  # default is 2
)

为流式传输选择合适的模型

不同的模型有不同的生成速度，这会影响流式延迟。根据最近的报告，像 DeepSeek V3.2 这样的小模型往往比具有更长上下文窗口的大模型逐令牌流式传输更快。

TokShop 以具有竞争力的每令牌价格提供多种模型。对于成本重要的流式传输用例，请考虑：

• DeepSeek V3.2 — 输入令牌 $0.42/M，输出令牌 $0.63/M，128K 上下文。速度和能力之间的良好平衡。
• GLM 4.6 — 输入令牌 $0.9/M，输出令牌 $3.3/M，200K 上下文。速度较慢，但能处理非常长的对话。
• Qwen3 Coder — 输入令牌 $2.25/M，输出令牌 $11.25/M，262K 上下文。针对代码生成进行了优化。

如果你正在构建一个流式传输响应的聊天界面，可以从更快、更便宜的模型开始，仅在任务需要时才升级到更大的模型。查看定价页面获取最新的模型列表。

生产环境流式传输的实用模式

1. 始终在流式传输中请求用量

在每个流式请求中设置 stream_options={"include_usage": True}。没有它，你将失去对成本的可见性。TokShop 仪表板会记录每次调用及其令牌计数，但流式用量分块能提供实时反馈。

2. 为显示缓冲部分内容

不要单独渲染每个字符——将小分块批量处理（例如，10 毫秒间隔）以减少 Web UI 中的 DOM 更新。SDK 已经在 API 层面做到了这一点，但如果你正在构建前端，请实现一个简单的防抖。

3. 处理 [DONE] 标记

[DONE] 事件标志着流的结束。如果你错过了它（例如，由于解析错误），你的循环可能会挂起等待更多数据。在原始 HTTP 方法中，在遇到 [DONE] 时显式中断。

4. 记录流中断

如果流在响应过程中失败，请记录部分内容和错误。这有助于调试问题是网络相关还是服务器端问题。你还可以通过要求用户重新表述查询来实现“恢复”功能。

结论

一旦理解了事件格式和常见故障模式，使用 SSE 进行流式聊天补全就很简单了。始终使用 stream_options 请求用量数据，正确解析事件（包括最终的用量分块），并为瞬时错误实现指数退避的重试逻辑。

对于生产环境使用，请针对不同的模型和网络条件测试你的流式实现。TokShop API 遵循 OpenAI 标准，因此为 OpenAI 编写的代码无需修改即可工作——只需将基础 URL 指向 https://tokshop.xyz/v1。有关模型定价和速率限制的更多详细信息，请访问文档。

常见问题

如何在流式传输时获取令牌用量数据？

要在流式响应中接收令牌用量数据（prompt_tokens 和 completion_tokens），你必须在请求中显式设置 stream_options={"include_usage": True}。没有此参数，默认会省略用量分块以减少延迟。

如果流式连接中断该怎么办？

如果 TCP 连接在流式传输过程中中断，你应该实现一个重试策略，重新发送整个原始请求，因为 API 是无状态的。使用指数退避（例如，1秒、2秒、4秒）进行重试是处理瞬时故障的推荐做法。

为什么 [DONE] 标记在 SSE 流中很重要？

data: [DONE] 事件标志着服务器发送事件流的明确结束。正确解析并在收到此标记时中断读取循环至关重要，可以防止你的客户端在等待永远不会到达的更多数据时挂起。