返回博客

API 与 SDK指南

Python 中的流式响应

使用官方 openai 包在模型生成答案时实时打印——同步和异步示例、推理输出、token 用量和错误处理。

作者：CheapTokens Team发布于 3 分钟阅读

Python 代码，包含遍历流块的 for 循环，文本逐 token 出现
本页内容
  1. 设置
  2. 最小示例
  3. 推理模型
  4. Token 用量
  5. asyncio
  6. 提前停止和超时
  7. 错误
  8. 线上传输的内容

使用流式传输时，答案会在模型仍在生成的过程中逐块到达，因此用户能立即看到文本，而无需等待完整响应。API 使用标准的 OpenAI 流式格式（服务器发送事件），官方的 openai Python 包会为你处理这些。

下面的每个示例都是完整的：复制到文件中即可运行。这些示例已使用 Python 3.10.12 和 openai 3.31.0 针对一个复现 API 流式格式的本地测试服务器进行了检查（语法和执行）。

设置#

bash
pip install openai
export CHEAPTOKENS_API_KEY="paste-your-key-here"

最小示例#

python
import os
from openai import OpenAI

client = OpenAI(
    base_url="https://api.cheaptokens.dev/v1",
    api_key=os.environ["CHEAPTOKENS_API_KEY"],
)

stream = client.chat.completions.create(
    model="deepseek-ai/DeepSeek-V4-Flash-0731",
    messages=[{"role": "user", "content": "Write a haiku about latency"}],
    stream=True,
)
for chunk in stream:
    if not chunk.choices:  # service chunk without text, e.g. the final usage chunk
        continue
    text = chunk.choices[0].delta.content
    if text:
        print(text, end="", flush=True)
print()

有两个细节很重要：

  • 跳过没有 choices 的块。 有些块不携带文本——例如带有 token 用量的最后一个块。
  • delta.content 可能为空。 第一个块通常只携带角色，最后一个块只携带 finish_reason。仅当文本存在时才打印它。

推理模型#

zai-org/GLM-5.3-Flash 在回答之前会进行推理。推理内容位于 delta 的单独字段 reasoning_content 中，而答案本身位于 content 中。SDK 会保留未知字段，因此你可以用 getattr 读取它。这是一个带有自己的客户端和自己的流的完整示例：

python
import os
import sys
from openai import OpenAI

client = OpenAI(
    base_url="https://api.cheaptokens.dev/v1",
    api_key=os.environ["CHEAPTOKENS_API_KEY"],
)

stream = client.chat.completions.create(
    model="zai-org/GLM-5.3-Flash",
    messages=[{"role": "user", "content": "Is 1013 a prime number?"}],
    stream=True,
)
for chunk in stream:
    if not chunk.choices:
        continue
    delta = chunk.choices[0].delta
    thinking = getattr(delta, "reasoning_content", None)
    if thinking:
        print(thinking, end="", flush=True, file=sys.stderr)  # reasoning → stderr
    if delta.content:
        print(delta.content, end="", flush=True)              # answer → stdout
print()

推理可能需要一段时间，答案的第一个词才会出现——这是流式传输的另一个理由。

Token 用量#

使用 stream_options 请求用量；它出现在最后一个块中，该块没有 choices：

python
import os
from openai import OpenAI

client = OpenAI(
    base_url="https://api.cheaptokens.dev/v1",
    api_key=os.environ["CHEAPTOKENS_API_KEY"],
)

stream = client.chat.completions.create(
    model="deepseek-ai/DeepSeek-V4-Flash-0731",
    messages=[{"role": "user", "content": "Explain idempotency in one paragraph"}],
    stream=True,
    stream_options={"include_usage": True},
)
usage = None
for chunk in stream:
    if chunk.usage:
        usage = chunk.usage
    if chunk.choices and chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

if usage:
    print(f"\ninput: {usage.prompt_tokens}, output: {usage.completion_tokens}")

将这些数字乘以模型的费率即可得到请求的成本——参见如何估算 AI 代理的成本。

asyncio#

在异步代码中使用 AsyncOpenAI 和 async for。客户端有自己的名称 async_client，因此不会与其他示例中的同步 client 冲突：

python
import asyncio
import os
from openai import AsyncOpenAI

async_client = AsyncOpenAI(
    base_url="https://api.cheaptokens.dev/v1",
    api_key=os.environ["CHEAPTOKENS_API_KEY"],
)

async def main() -> None:
    stream = await async_client.chat.completions.create(
        model="deepseek-ai/DeepSeek-V4-Flash-0731",
        messages=[{"role": "user", "content": "Name three uses of a message queue"}],
        stream=True,
    )
    async for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            print(chunk.choices[0].delta.content, end="", flush=True)
    print()

asyncio.run(main())

提前停止和超时#

离开循环即可停止生成。使用 with 打开流会在块结束时关闭连接。客户端超时是显式设置的；SDK 还会自行重试一些失败的请求（max_retries，默认 2 次）。

python
import os
from openai import OpenAI

client = OpenAI(
    base_url="https://api.cheaptokens.dev/v1",
    api_key=os.environ["CHEAPTOKENS_API_KEY"],
    timeout=120,  # seconds; long answers take time
)

received = ""
with client.chat.completions.create(
    model="deepseek-ai/DeepSeek-V4-Flash-0731",
    messages=[{"role": "user", "content": "Write a long story about a lighthouse"}],
    stream=True,
) as stream:
    for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            received += chunk.choices[0].delta.content
        if len(received) > 300:  # stop early: enough text received
            break
print(received)

错误#

在第一个块之前发生的错误——错误的密钥、余额为空、未知模型——会在你创建流时作为异常抛出：

python
import os
import openai
from openai import OpenAI

client = OpenAI(
    base_url="https://api.cheaptokens.dev/v1",
    api_key=os.environ["CHEAPTOKENS_API_KEY"],
)

try:
    stream = client.chat.completions.create(
        model="deepseek-ai/DeepSeek-V4-Flash-0731",
        messages=[{"role": "user", "content": "Hello"}],
        stream=True,
    )
    for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            print(chunk.choices[0].delta.content, end="", flush=True)
    print()
except openai.AuthenticationError:          # 401: wrong or rotated key
    print("Check the API key")
except openai.BadRequestError as e:         # 400: e.g. unknown model ID
    print("Bad request:", e.message)
except openai.APIStatusError as e:          # other statuses, e.g. 402 when the balance is used up
    print(e.status_code, e.message)

状态码和修复方法的完整列表见常见连接错误及其修复方法。

线上传输的内容#

要调试，请用 curl 查看原始流（-N 关闭缓冲）：

bash
curl -N https://api.cheaptokens.dev/v1/chat/completions \
  -H "Authorization: Bearer $CHEAPTOKENS_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"model": "deepseek-ai/DeepSeek-V4-Flash-0731", "stream": true,
       "messages": [{"role": "user", "content": "Hi"}]}'

每个事件都是一行 data: {…} 加上一个 JSON 块；流以 data: [DONE] 结束。以冒号开头的行是长时间暂停期间发送的保活注释——客户端会忽略它们。

标签: