流式响应

逐块读取文本,确认响应结束并获取完整用量。

流式响应把一条回答拆成多个事件。显示收到的文本后,还要读取终态和用量,才能确认本轮完成。

本页使用 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 的读取顺序是:

  1. 收集 choices[].delta.content;role、思考或空增量可能没有正文,不应当作文本显示。
  2. 记录 finish_reason。本例的完整文本回答以 stop 结束;tool_calls 应进入工具调用流程。
  3. 设置 stream_options.include_usage=true 后,继续读取末块 usage。这一块的 choices 可以为空,不能直接访问 choices[0]。
  4. 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 后,文本、终态与末块用量均正常返回。

协议细节:OpenAI · Anthropic。

API 参考 → · Cookbook →

导出

本页内容