工程实践
流式输出接入:三种写法与三个坑
长输出任务用非流式请求,用户要盯着空屏幕等几十秒;开了流式,首 token 到达后就能边收边渲染。协议层面只是加一个 "stream": true,真正的工作量在客户端怎么读、怎么兜底。
curl 观察原始事件Python SDK 增量读取Node SDK 异步迭代
写法一:curl 先看原始 SSE 长什么样
调试流式问题时,先用 curl 看原始事件流,能立刻分辨"服务端没发"还是"客户端没收"。-N 关闭 curl 自身的缓冲。
curl -N https://codebridgeapi.top/v1/chat/completions \
-H "Authorization: Bearer YOUR_API_KEY" \
-H "Content-Type: application/json" \
-d '{"model":"<从 /v1/models 复制>","stream":true,
"messages":[{"role":"user","content":"数到 5"}]}'
# 正常输出是一行行 data: {...} 的增量,最后以 data: [DONE] 结束写法二:Python SDK
from openai import OpenAI
client = OpenAI(base_url="https://codebridgeapi.top/v1", api_key="YOUR_API_KEY")
stream = client.chat.completions.create(
model="<从 /v1/models 复制>",
messages=[{"role": "user", "content": "写一首四行小诗"}],
stream=True,
)
parts = []
for chunk in stream:
delta = chunk.choices[0].delta.content or ""
parts.append(delta)
print(delta, end="", flush=True) # 边收边渲染
answer = "".join(parts) # 断流兜底:已收内容始终在手写法三:Node SDK
import OpenAI from "openai";
const client = new OpenAI({
baseURL: "https://codebridgeapi.top/v1",
apiKey: process.env.API_KEY,
});
const stream = await client.chat.completions.create({
model: "<从 /v1/models 复制>",
messages: [{ role: "user", content: "写一首四行小诗" }],
stream: true,
});
for await (const chunk of stream) {
process.stdout.write(chunk.choices[0]?.delta?.content ?? "");
}三个最常踩的坑
坑一:中间层缓冲
本地代理、公司网关、部分抓包工具会把 SSE 攒成整包再放行,表现为"等很久然后全文一次蹦出来"。排查顺序:直连网络 → 关代理 → 关抓包。自建反代要给该路径关闭 proxy_buffering。
坑二:超时设置照抄非流式
流式连接的总时长天然更长,客户端把总超时设成 30 秒会频繁误杀。正确做法是放宽总超时(120 秒以上),另设"首 token 超时"来快速发现真故障。
坑三:漏处理 [DONE] 与半行数据
自己解析 SSE(不用 SDK)时,要处理结束标记 data: [DONE],并且按行缓冲——TCP 分包可能把一个 JSON 切成两半,直接 JSON.parse 会偶发报错。能用 SDK 就用 SDK。