开发接入
流式输出
加上 stream=true 参数,边生成边返回,适合聊天界面并避免长回答超时。
加上 stream: true,模型会边生成边返回,适合聊天界面,也能避免长回答等待过久。
import os
from openai import OpenAI
client = OpenAI(
api_key=os.environ["ATHANDRA_API_KEY"],
base_url="https://ai.athandra.com/v1",
)
stream = client.chat.completions.create(
model=os.environ.get("ATHANDRA_MODEL", "deepseek-v3.2"),
messages=[{"role": "user", "content": "从 1 数到 5,每个数字一行"}],
stream=True,
)
for chunk in stream:
# 部分分片没有 choices 或 content(例如首尾分片),要先判断
if chunk.choices and chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="", flush=True)
print()
返回格式
服务端以 SSE(Server-Sent Events)逐块推送,每一行以 data: 开头,内容是一个 JSON 分片,增量文字在 choices[0].delta.content:
data: {"id":"chatcmpl-...","object":"chat.completion.chunk","model":"deepseek-v3.2","choices":[{"index":0,"delta":{"content":"好的"},"finish_reason":null}]}
data: {"id":"chatcmpl-...","object":"chat.completion.chunk","model":"deepseek-v3.2","choices":[{"index":0,"delta":{"content":",我们来"},"finish_reason":null}]}
data: [DONE]使用 SDK 时不用自己解析,直接遍历即可。
注意事项
- 首个分片的
content可能是空字符串,部分分片没有choices,代码里要先判断(示例已处理)。 - curl 要加
-N,否则输出会被缓冲,看起来像“一次性返回”。 - 经过 Nginx 等反向代理时,需要关闭代理缓冲,否则流式会被攒成一块。
- 流式也是避免超时的好办法,见 错误排查:超时。