流式返回

流式（Streaming） 让服务器一边生成一边把 tokens 逐字往回吐。用户端体验更好（不用等完全生成完再显示），也让长请求不容易超时。

打开流式

在请求 body 里加 "stream": true：

{
  "model": "claude-opus-4-8",
  "messages": [...],
  "stream": true
}

SSE 格式

响应用 Server-Sent Events：

Content-Type: text/event-stream

data: {"id":"...","choices":[{"delta":{"role":"assistant"}}]}

data: {"id":"...","choices":[{"delta":{"content":"Hel"}}]}

data: {"id":"...","choices":[{"delta":{"content":"lo"}}]}

data: {"id":"...","choices":[{"delta":{"content":" world"}}]}

data: {"id":"...","choices":[{"finish_reason":"stop"}],"usage":{"prompt_tokens":10,"completion_tokens":3,"total_tokens":13}}

data: [DONE]

关键点：

• 每个事件是一行 data: {json}\n\n
• 最后一个 data 里带完整的 usage（tokens 统计）
• 结束标志是字符串 data: [DONE]

Python 解析

用 OpenAI SDK 最省事，它自己解析：

stream = client.chat.completions.create(
    model="claude-opus-4-8",
    messages=[{"role":"user","content":"讲个故事"}],
    stream=True,
    stream_options={"include_usage": True},  # 让最后一 chunk 带 usage
)

full = ""
for chunk in stream:
    delta = chunk.choices[0].delta.content if chunk.choices else None
    if delta:
        full += delta
        print(delta, end="", flush=True)
    if chunk.usage:
        print(f"\n\nusage: {chunk.usage}")

print(f"\ntotal length: {len(full)}")

Node.js 解析

const stream = await client.chat.completions.create({
  model: "claude-opus-4-8",
  messages: [{ role: "user", content: "讲个故事" }],
  stream: true,
});

for await (const chunk of stream) {
  const text = chunk.choices[0]?.delta?.content;
  if (text) process.stdout.write(text);
}

手写 SSE 解析（无 SDK）

import httpx, json

with httpx.stream(
    "POST",
    "https://api.modexflow.cc/v1/chat/completions",
    headers={"Authorization": f"Bearer {KEY}"},
    json={
        "model": "claude-opus-4-8",
        "messages": [{"role":"user","content":"hi"}],
        "stream": True,
    },
    timeout=None,
) as r:
    for line in r.iter_lines():
        if not line.startswith("data: "):
            continue
        payload = line[6:]
        if payload == "[DONE]":
            break
        chunk = json.loads(payload)
        delta = chunk["choices"][0]["delta"].get("content", "")
        print(delta, end="", flush=True)

Reasoning tokens

DeepSeek Pro / GPT-5.5 会在正式输出前"思考"，思考过程也算 completion tokens：

{
  "delta": {
    "content": null,
    "reasoning_content": "让我先分析一下这个问题..."
  }
}

• delta.content = 面向用户的答案
• delta.reasoning_content = 思考过程（一般不要展示给用户）

usage 里会有：

{
  "prompt_tokens": 10,
  "completion_tokens": 200,
  "completion_tokens_details": {
    "reasoning_tokens": 180
  }
}

注意：completion_tokens 包含 reasoning tokens。所以如果 max_tokens: 100 但模型花了 80 tokens 思考，实际只剩 20 tokens 给你输出——很可能得到空的 content。给推理型模型的 max_tokens 要慷慨（Claude/GPT 至少 500，DeepSeek Pro 至少 1000）。

Prompt Caching 提示

Claude-ESpeed 分组的响应 usage 里会带：

{
  "prompt_tokens": 5000,
  "prompt_tokens_details": {
    "cached_tokens": 4500
  }
}

cached_tokens > 0 说明命中缓存了——省钱。

要触发缓存：相同的前缀（system prompt + 前面若干轮对话）需要在 5 分钟内被再次调用。所以：

• 建议：把长而稳定的内容（system prompt、代码库、文档）放在最前面
• 避免：每次改动 messages 顶部
• 不生效：temperature、max_tokens 等参数变化不影响缓存命中

客户端超时

• 首字节等 30-60s 是正常上限（Claude 长 prompt / GPT 深度思考时）
• 流式一旦开始，chunk 间隔一般 < 1s
• 整个请求上限 30 分钟（我们网关和 Caddy 都设成 30m 读/写超时）

中断请求

客户端断开连接 → 网关同步断开对上游的请求。上游那边（比如 Anthropic）会立即停止生成，不再产生 tokens。

但已经生成到那一刻的 tokens 会算钱。如果生成到一半你按 Ctrl+C，可能已经用了几千 tokens。

常见问题

Q: 我看不到 usage 字段？
在请求里加 "stream_options": {"include_usage": true}（OpenAI SDK 默认已加）。

Q: 手写 SSE 收到不完整的 line？
你的 HTTP 客户端 buffer 太大或没关。用 curl -N / httpx.stream() / fetch with reader，别用 .text() 一把梭。

Q: 流式突然断了？
看客户端超时是不是设太短。用 include_usage 或 finish_reason 判断是否正常结束——没收到 [DONE] 就是异常中断。