# 流式响应

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

来源：https://zhonkemodel.dflop.top/docs/guides/streaming

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

## 基本用法

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

### OpenAI SDK

```python
from openai import OpenAI

client = OpenAI(
    base_url="https://zhonkezhonkeapi.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

```python
from anthropic import Anthropic

client = Anthropic(
    base_url="https://zhonkezhonkeapi.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

```python
from google import genai

client = genai.Client(
    api_key="sk-gpushare-xxx",
    http_options={"base_url": "https://zhonkezhonkeapi.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 时)

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

JavaScript fetch 解析:

```javascript
const resp = await fetch("https://zhonkezhonkeapi.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.5",
    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`:

```python
stream = client.chat.completions.create(
    model="gpt-5.5",
    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)

```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)

```javascript
const controller = new AbortController();

const resp = await fetch("https://zhonkezhonkeapi.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,按非流式错误处理即可。完整错误码见[错误处理](../reference/errors.md)。

### 开流后 (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 参考](../reference/api-reference.md)。

## 反向代理 / Nginx 注意

本网关在响应里附加:
```http
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 是最直观的:

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

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