Model 广场Model 广场

流式响应

SSE 解析, usage chunk, 中断处理, 流式错误与 180s 时长上限, 何时必须 stream=true

本平台支持 SSE (Server-Sent Events) 流式响应。首 token 更早到达,UI 可以边生成边渲染。

基本用法#

client 初始化 (base_url / API Key) 详见快速开始,这里每段只重复最小构造。

OpenAI SDK#

from openai import OpenAI

client = OpenAI(
    base_url="https://api.dflop.top/v1",
    api_key="sk-gpushare-xxx",
)

stream = client.chat.completions.create(
    model="claude-sonnet-4-6",
    messages=[{"role": "user", "content": "Stream a haiku"}],
    stream=True,
)

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

Anthropic SDK#

from anthropic import Anthropic

client = Anthropic(
    base_url="https://api.dflop.top",  # ⚠️ 不带 /v1
    api_key="sk-gpushare-xxx",
)

with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Stream a haiku"}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

Gemini SDK#

from google import genai

client = genai.Client(
    api_key="sk-gpushare-xxx",
    http_options={"base_url": "https://api.dflop.top"},
)

stream = client.models.generate_content_stream(
    model="gemini-2.5-pro",
    contents="Stream a haiku",
)
for chunk in stream:
    print(chunk.text, end="", flush=True)

何时必须用 stream=true#

工具必须 stream原因
image_generation✅上游 WebSocket 流式通道不返回同步响应,只发流
web_search✅同上,搜索过程通过流式 chunk 实时返回
function 工具任意走 HTTP 通道,流 / 非流均可
纯文本对话任意流 / 非流均可

不满足时,gateway 返回 400 invalid_request_error,message 为 Tools `web_search` and `image_generation` require `stream: true`。

SSE 协议格式#

本网关透传上游 SSE,每条 data: 行 = 一个 chunk。

OpenAI Chat 流式 chunk#

data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"Hello"}}]}

data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":" world"}}]}

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

data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[],"usage":{"prompt_tokens":10,"completion_tokens":2,"total_tokens":12}}

data: [DONE]

注意两点 (手写解析器最常踩):

  • finish_reason chunk 不带 usage —— usage 是紧随其后的独立尾 chunk,这是 OpenAI stream_options.include_usage 的标准约定
  • usage chunk 的 choices 是空数组,直接 chunk.choices[0].delta 会越界 —— 先判 chunk.choices 非空再取 delta

Anthropic 流式 events#

event: message_start
data: {"type":"message_start","message":{...}}

event: content_block_delta
data: {"type":"content_block_delta","delta":{"type":"text_delta","text":"Hello"}}

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

Gemini 流式 (SSE)#

data: {"candidates":[{"content":{"parts":[{"text":"Hello"}]}}]}

data: {"candidates":[{"content":{"parts":[{"text":" world"}]},"finishReason":"STOP"}]}

Gemini 协议没有 [DONE] 哨兵 —— 以流自然关闭为结束信号,解析器不要等 [DONE]。

手工解析 (curl + 客户端无 SDK 时)#

curl https://api.dflop.top/v1/chat/completions \
  -H "Authorization: Bearer $PLATFORM_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "gpt-5.4",
    "stream": true,
    "messages": [{"role":"user","content":"Hello"}]
  }' \
  --no-buffer | while read line; do
    echo "$line"
  done

JavaScript fetch 解析:

const resp = await fetch("https://api.dflop.top/v1/chat/completions", {
  method: "POST",
  headers: {
    "Authorization": `Bearer ${process.env.PLATFORM_API_KEY}`,
    "Content-Type": "application/json",
  },
  body: JSON.stringify({
    model: "gpt-5.4",
    stream: true,
    messages: [{ role: "user", content: "Hello" }],
  }),
});

const reader = resp.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");
  buffer = lines.pop() ?? "";

  for (const line of lines) {
    if (!line.startsWith("data: ")) continue;
    const data = line.slice(6);
    if (data === "[DONE]") return;
    const chunk = JSON.parse(data);
    process.stdout.write(chunk.choices[0]?.delta?.content ?? "");
  }
}

拿 token usage#

OpenAI Chat 流式默认不返回 trailing usage chunk。要拿真实 token 数,加 stream_options:

stream = client.chat.completions.create(
    model="gpt-5.4",
    messages=[...],
    stream=True,
    stream_options={"include_usage": True},  # 关键
)

last_chunk = None
for chunk in stream:
    if chunk.choices:
        print(chunk.choices[0].delta.content or "", end="")
    last_chunk = chunk

# 最后一个 chunk 带 usage
print(f"\nTokens: {last_chunk.usage}")

本网关内部对大部分上游通道自动注入 stream_options.include_usage=true (除了 GLM 上游,它本来就发 usage chunk)。所以客户端不强制设也能拿到 token,但显式设更稳。

中断 / 取消#

OpenAI SDK (Python)#

stream = client.chat.completions.create(..., stream=True)
try:
    for chunk in stream:
        if some_user_canceled():
            stream.close()  # 显式关闭,gateway 收到 client disconnect
            break
        print(chunk.choices[0].delta.content or "", end="")
except KeyboardInterrupt:
    stream.close()

fetch (JavaScript)#

const controller = new AbortController();

const resp = await fetch("https://api.dflop.top/v1/chat/completions", {
  signal: controller.signal,
  // ...
});

// 之后想中断:
controller.abort();

客户端断开后,gateway 与上游的连接随之立即关闭 (HTTP / WebSocket 通道均如此) —— 机制是连接 drop,不是显式取消指令,效果上上游随即停止生成。已生成的 token 仍计费;若断开导致 usage 尾 chunk 没收到,gateway 按已流出的字符数估算 output tokens 落账。

错误处理#

开流前: 普通 HTTP 错误,不是 SSE#

鉴权失败 (401)、余额耗尽 (402)、模型不存在 (400 model_not_found)、无可用通道 (503) 等都发生在开流之前,以普通 HTTP JSON 错误体返回 —— 此时还没有 SSE,按非流式错误处理即可。完整错误码见错误处理。

开流后 (HTTP 200): 按路径区分#

工具请求 (web_search / image_generation,走 WS V2 通道): 中途错误会合成一个错误 chunk,后跟 [DONE]:

data: {"error":{"message":"<具体原因文本>","type":"api_error","code":"upstream_error"}}

data: [DONE]

type / code 是固定值 (api_error / upstream_error),不要按 code 分支判断错误类别 —— 具体原因只在 message 文本里。解析时先检查 chunk 是否有 error 字段再处理 delta.content,把 error 当 turn 结束信号。

纯文本对话及 /v1/messages / /v1/responses / /v1beta 全部路径: 上游中途出错时,gateway 不合成错误事件,流会提前终止 —— 不保证发 [DONE]、finish_reason、message_stop 或任何错误帧。健壮的客户端必须把「流结束但没见到终结哨兵」也当作中断处理,不能当作正常完成。(上游协议内自带的错误事件,如 Anthropic 的 event: error,会原样透传 —— 但那是上游发的,gateway 自己不产。)

无论哪种路径,错误 / 中断前已经流出的 content 都是部分有效的,可以保留展示。

~180 秒时长上限#

单次请求 (含流式,覆盖 /v1/chat/completions、/v1/messages、/v1/responses、/v1beta 全部 chat 端点) 有 180 秒总时长上限 —— 流式只是更早拿到首 token,不豁免总时长。超时的表现就是上一节的「流提前终止」: 长回答总在 3 分钟左右被截断,基本就是撞上了这条线。大 max_tokens 长文生成、慢模型、image_generation 长 turn 都可能命中。建议:

  • 调低 max_tokens,或把长任务拆成多个 turn
  • 客户端 SDK timeout 设 ≥ 200s (略大于 gateway 上限,把超时判定留给 gateway)
  • 已产出的 token 照常计费

各端点超时一览见 API 参考。

反向代理 / Nginx 注意#

本网关在响应里附加:

Content-Type: text/event-stream
Cache-Control: no-cache, no-store, no-transform
X-Accel-Buffering: no

如果你在自己的反向代理后面再代理本网关,确保别开 buffering (proxy_buffering off; for Nginx),否则流式整个响应会被缓冲到结束才一次性下发,失去流式意义。

调试#

curl 直接看 raw SSE 是最直观的:

curl https://api.dflop.top/v1/chat/completions \
  -H "Authorization: Bearer $PLATFORM_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"model":"gpt-5.4","stream":true,"messages":[{"role":"user","content":"hi"}]}' \
  --no-buffer 2>&1 | head -20

如果看到的是一次性响应而不是逐 chunk 显示,说明中间某层在缓冲。