API & SDKHướng dẫn

Phản hồi streaming trong Python

In câu trả lời của mô hình ngay khi nó được tạo bằng gói openai chính thức — ví dụ đồng bộ và async, đầu ra suy luận, sử dụng token và xử lý lỗi.

Bởi CheapTokens TeamĐã xuất bản 15 thg 10, 20263 phút đọc

Mã Python với vòng lặp for qua các chunk stream và văn bản xuất hiện từng token một

Trên trang này

Cài đặt
Ví dụ tối thiểu
Mô hình suy luận
Sử dụng token
asyncio
Dừng sớm và timeout
Lỗi
Trên đường truyền

Với streaming, câu trả lời đến từng phần trong khi mô hình vẫn đang tạo ra nó, vì vậy người dùng của bạn thấy văn bản ngay lập tức thay vì chờ toàn bộ phản hồi. API sử dụng định dạng streaming chuẩn của OpenAI (server-sent events), và gói Python chính thức openai xử lý nó cho bạn.

Mọi ví dụ dưới đây đều hoàn chỉnh: sao chép nó vào một tệp và chạy. Các ví dụ đã được kiểm tra với Python 3.10.12 và openai 3.31.0 trên một máy chủ thử nghiệm cục bộ mô phỏng định dạng streaming của API (cú pháp và thực thi).

Cài đặt#

bash

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

Ví dụ tối thiểu#

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()

Hai chi tiết quan trọng:

Bỏ qua các chunk không có choices. Một số chunk không mang văn bản — ví dụ chunk cuối cùng với thông tin sử dụng token.
delta.content có thể trống. Chunk đầu tiên thường chỉ mang role, và chunk cuối cùng chỉ mang finish_reason. Chỉ in văn bản khi nó có ở đó.

Mô hình suy luận#

zai-org/GLM-5.3-Flash suy luận trước khi trả lời. Phần suy luận đến trong một trường riêng của delta, reasoning_content, và câu trả lời nằm trong content. SDK giữ các trường không xác định, vì vậy bạn có thể đọc nó bằng getattr. Đây là một ví dụ hoàn chỉnh với client riêng và stream riêng:

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()

Suy luận có thể mất một lúc trước khi từ đầu tiên của câu trả lời xuất hiện — một lý do nữa để dùng streaming.

Sử dụng token#

Yêu cầu thông tin sử dụng với stream_options; nó đến trong chunk cuối cùng, chunk này không có 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}")

Nhân các số này với mức giá của mô hình để tính chi phí của yêu cầu — xem Cách ước tính chi phí của một AI agent.

asyncio#

Trong mã async, dùng AsyncOpenAI và async for. Client có tên riêng, async_client, để không xung đột với client đồng bộ từ các ví dụ khác:

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())

Dừng sớm và timeout#

Thoát khỏi vòng lặp để dừng tạo. Mở stream với with sẽ đóng kết nối khi khối kết thúc. Timeout của client được đặt rõ ràng; SDK cũng tự động thử lại một số yêu cầu thất bại (max_retries, mặc định là 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)

Lỗi#

Các lỗi xảy ra trước chunk đầu tiên — sai key, số dư trống, mô hình không xác định — được ném ra dưới dạng ngoại lệ khi bạn tạo stream:

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)

Danh sách đầy đủ các mã trạng thái và cách khắc phục có trong Lỗi kết nối thường gặp và cách khắc phục.

Trên đường truyền#

Để gỡ lỗi, hãy xem stream thô bằng curl (-N tắt buffering):

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"}]}'

Mỗi sự kiện là một dòng data: {…} với một chunk JSON; stream kết thúc bằng data: [DONE]. Các dòng bắt đầu bằng dấu hai chấm là chú thích keep-alive được gửi trong thời gian tạm dừng dài — client bỏ qua chúng.

Thẻ:

Cài đặt#

Ví dụ tối thiểu#

Mô hình suy luận#

Sử dụng token#

asyncio#

Dừng sớm và timeout#

Lỗi#

Trên đường truyền#

Bài viết liên quan

Các lỗi kết nối thường gặp và cách khắc phục

Token trong AI là gì? Giải thích bằng ngôn ngữ đơn giản

Request API đầu tiên của bạn với CheapTokens

Giá API DeepSeek năm 2026 — bạn thực sự trả bao nhiêu