流式输出
默认情况下接口会等模型全部写完才一次性返回。加上 "stream": true,就会逐字返回,用户不用干等。
请求
只在请求体里加一个字段:
json
{
"model": "deepseek-v4-pro-0813",
"messages": [{ "role": "user", "content": "讲个短故事" }],
"stream": true
}响应格式
返回的是 text/event-stream,每行形如:
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":{"content":"前"},"finish_reason":null}]}
data: {"id":"chatcmpl-abc","object":"chat.completion.chunk","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}
data: [DONE]要点:
| 现象 | 说明 |
|---|---|
delta.content | 这一次新增的文字片段,要累加起来 |
delta 是空对象 | 通常出现在开头(role)和结尾 |
finish_reason: "stop" | 生成结束 |
data: [DONE] | 流结束标记,收到就可以断开连接 |
不要用换行符切分
SSE 的一条消息可能跨多行,正确做法是按 \n\n(空行)切分事件,再取 data: 后面的内容。
示例
bash
curl -N https://api.blueworkshop.top/v1/chat/completions \
-H "Authorization: Bearer $CYAN_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"model": "deepseek-v4-pro-0813",
"messages": [{"role": "user", "content": "讲个短故事"}],
"stream": true
}'python
from openai import OpenAI
client = OpenAI(
api_key="sk-你的密钥",
base_url="https://api.blueworkshop.top/v1",
timeout=300.0,
)
stream = client.chat.completions.create(
model="deepseek-v4-pro-0813",
messages=[{"role": "user", "content": "讲个短故事"}],
stream=True,
)
for chunk in stream:
if not chunk.choices:
continue
delta = chunk.choices[0].delta.content
if delta:
print(delta, end="", flush=True)
print()javascript
import OpenAI from 'openai'
const client = new OpenAI({
apiKey: process.env.CYAN_API_KEY,
baseURL: 'https://api.blueworkshop.top/v1',
})
const stream = await client.chat.completions.create({
model: 'deepseek-v4-pro-0813',
messages: [{ role: 'user', content: '讲个短故事' }],
stream: true,
})
for await (const chunk of stream) {
const delta = chunk.choices[0]?.delta?.content
if (delta) process.stdout.write(delta)
}手动解析(不用 SDK 时)
python
import json
import requests
resp = requests.post(
"https://api.blueworkshop.top/v1/chat/completions",
headers={
"Authorization": "Bearer sk-你的密钥",
"Content-Type": "application/json",
},
json={
"model": "deepseek-v4-pro-0813",
"messages": [{"role": "user", "content": "讲个短故事"}],
"stream": True,
},
stream=True,
timeout=300,
)
for raw in resp.iter_lines():
if not raw:
continue
line = raw.decode("utf-8")
if not line.startswith("data:"):
continue
payload = line[5:].strip()
if payload == "[DONE]":
break
chunk = json.loads(payload)
delta = chunk["choices"][0]["delta"].get("content")
if delta:
print(delta, end="", flush=True)流式下的 usage
默认情况下流式返回不包含 token 用量。想要的话加:
json
{
"stream": true,
"stream_options": { "include_usage": true }
}这样最后会多推一个 chunk,choices 是空数组,usage 里有完整统计:
json
{ "choices": [], "usage": { "prompt_tokens": 12, "completion_tokens": 88, "total_tokens": 100 } }计费不受影响
不管开不开 include_usage,用量都在服务端照实记录,只是返回不返回给你的区别。
常见坑
| 问题 | 原因 |
|---|---|
| 前端一个字都不显示 | 中间有缓冲。nginx / Cloudflare / 客户端都要关缓冲;我们的网关已关 |
| 中文显示成乱码 | 没按 UTF-8 解码,或者按字节切断了多字节字符 |
| 收到一半就断了 | 读超时太短。流式要设 300 秒以上 |
| 一直不出结果 | 推理模型(o 系列 / R1)会先"想"很久才吐第一个字,这是正常的 |
| 累积文字重复 | 把 delta 当成全量覆盖了,实际是增量追加 |
推理模型的流式很特殊
o1 / R1 这类模型会先生成大量内部推理(不返回给你),然后才开始输出正文。所以首字延迟可能长达几十秒。
界面上要做区分:显示「正在思考…」而不是「无响应」。