# 流式响应 (/docs/cloud/services/model/guides/streaming)



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

本页使用 `POST /openai/v1/chat/completions` 的 `deepseek-flash`，并提供 `POST /anthropic/v1/messages` 的 `claude-opus-5` 示例；两者于 2026-10-12 验证通过。

## 1. 准备终端 [#1-准备终端]

将 Praxis 云 API Key 放入环境变量 `PRAXIS_API_KEY`；获取方式见 [Key 与鉴权](/docs/cloud/account/api-keys)。不要把 Key 写入代码。准备 Python 和 [uv](https://docs.astral.sh/uv/getting-started/installation/)；TypeScript 示例使用 [Bun](https://bun.sh/docs/installation)。

```bash
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 接收文本和末块用量 [#2-用-python-接收文本和末块用量]

保存为 `streaming.py`，运行 `uv run -q --no-project --with openai --with anthropic python streaming.py`。

```python
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` 应进入[工具调用](/docs/cloud/services/model/guides/tool-calling)流程。
3. 设置 `stream_options.include_usage=true` 后，继续读取末块 `usage`。这一块的 `choices` 可以为空，不能直接访问 `choices[0]`。
4. HTTP 事件流以 `data: [DONE]` 收尾；SDK 负责处理该标记并结束迭代。

不要在收到 `finish_reason` 时立即跳出循环，否则会漏掉之后的用量。断开连接不等于模型完成；没有终态或末块用量时，不应把部分文本和未知用量记为完整结果。

## 3. 用 cURL 检查事件顺序 [#3-用-curl-检查事件顺序]

在同一终端执行：

```bash
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 [#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`。

```python
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](https://developers.openai.com/api/docs/guides/streaming-responses) · [Anthropic](https://platform.claude.com/docs/en/build-with-claude/streaming)。

[API 参考 →](/docs/api/compatible/openai) · [Cookbook →](/docs/cookbook)