流式响应
逐块读取文本,确认响应结束并获取完整用量。
流式响应把一条回答拆成多个事件。显示收到的文本后,还要读取终态和用量,才能确认本轮完成。
本页使用 POST /openai/v1/chat/completions 的 deepseek-flash,并提供 POST /anthropic/v1/messages 的 claude-opus-5 示例;两者于 2026-10-12 验证通过。
1. 准备终端
将 Praxis 云 API Key 放入环境变量 PRAXIS_API_KEY;获取方式见 Key 与鉴权。不要把 Key 写入代码。准备 Python 和 uv;TypeScript 示例使用 Bun。
test -n "$PRAXIS_API_KEY"期望:退出码为 0。示例默认使用 https://api.deepraxis.ai。切换环境时,将 PRAXIS_API_BASE 设为相应 API 域名,不带末尾斜杠或 API 路径。
将每段 Python 保存为标明的文件名,以 uv run -q --no-project --with openai --with anthropic python <文件名> 执行。各段代码独立运行。
2. 用 Python 接收文本和末块用量
保存为 streaming.py,运行 uv run -q --no-project --with openai --with anthropic python streaming.py。
import os
from openai import OpenAI
client = OpenAI(api_key=os.environ['PRAXIS_API_KEY'],
base_url=os.environ.get('PRAXIS_API_BASE', 'https://api.deepraxis.ai') + '/openai/v1')
text, finish, usage = '', None, None
with client.chat.completions.with_streaming_response.create(
model='deepseek-flash', messages=[{'role':'user','content':'Reply with exactly: Hello streaming.'}],
max_tokens=256, stream=True, stream_options={'include_usage':True},
) as response:
print('X-Request-Id:', response.headers.get('x-request-id'))
for chunk in response.parse():
if chunk.usage is not None:
usage = chunk.usage
for choice in chunk.choices:
piece = choice.delta.content or ''
text += piece
print(piece, end='', flush=True)
if choice.finish_reason is not None:
finish = choice.finish_reason
print('\nfinish_reason:', finish)
assert text.strip() and finish == 'stop', 'Incomplete text response'
assert usage is not None, 'No final usage received'
print('usage:', usage.model_dump())期望:逐块输出 Hello streaming.,随后输出 finish_reason: stop 和包含输入、输出 token 数的 usage;退出码为 0。
Chat Completions 的读取顺序是:
- 收集
choices[].delta.content;role、思考或空增量可能没有正文,不应当作文本显示。 - 记录
finish_reason。本例的完整文本回答以stop结束;tool_calls应进入工具调用流程。 - 设置
stream_options.include_usage=true后,继续读取末块usage。这一块的choices可以为空,不能直接访问choices[0]。 - HTTP 事件流以
data: [DONE]收尾;SDK 负责处理该标记并结束迭代。
不要在收到 finish_reason 时立即跳出循环,否则会漏掉之后的用量。断开连接不等于模型完成;没有终态或末块用量时,不应把部分文本和未知用量记为完整结果。
3. 用 cURL 检查事件顺序
在同一终端执行:
curl --fail-with-body --no-buffer --silent --show-error --include \
"${PRAXIS_API_BASE:-https://api.deepraxis.ai}/openai/v1/chat/completions" \
-H "Authorization: Bearer $PRAXIS_API_KEY" \
-H "Content-Type: application/json" \
-d '{"model":"deepseek-flash","messages":[{"role":"user","content":"Reply with exactly: Hello streaming."}],"max_tokens":256,"stream":true,"stream_options":{"include_usage":true}}'期望:HTTP 200,响应类型为 text/event-stream;正文增量后出现 finish_reason: "stop"、带 usage 的块,最后是 data: [DONE];cURL 退出码为 0。--no-buffer 避免 cURL 缓冲显示,但单独退出码 0 不能证明回答完整,仍需检查终态。
4. 使用 Anthropic Messages
Messages 依次发送 message_start、内容块的 content_block_start / content_block_delta / content_block_stop、message_delta 和 message_stop,中间可能有 ping。不要将它按 Chat Completions 的 choices 解析。输入用量在起始消息中,后续用量更新由 SDK 汇总;get_final_message() 返回完整结果。
保存为 streaming-anthropic.py,运行 uv run -q --no-project --with openai --with anthropic python streaming-anthropic.py。
import os
from anthropic import Anthropic
client = Anthropic(api_key=os.environ['PRAXIS_API_KEY'],
base_url=os.environ.get('PRAXIS_API_BASE', 'https://api.deepraxis.ai') + '/anthropic')
with client.messages.stream(model='claude-opus-5', max_tokens=256,
messages=[{'role':'user','content':'Reply with exactly: Hello streaming.'}]) as stream:
print('X-Request-Id:', stream.response.headers.get('x-request-id'))
for text in stream.text_stream:
print(text, end='', flush=True)
message = stream.get_final_message()
assert message.stop_reason == 'end_turn'
assert any(block.type == 'text' and block.text.strip() for block in message.content)
print('\nusage:', message.usage.model_dump())期望:输出同样的文本和最终用量,stop_reason 为 end_turn,退出码为 0。
实测失败与处理
把 text-embedding-3-small 用在本页聊天请求中时,实测返回 HTTP 404:
openai.chat.completions is not offered for text-embedding-3-small
这是模型不提供聊天端点,不是等待更多事件可以解决的问题。改用本页已验证的 deepseek-flash 后,文本、终态与末块用量均正常返回。