Kembali ke blog

API & SDKPanduan

Respons streaming di Python

Cetak jawaban model saat dihasilkan dengan paket openai resmi — contoh sinkron dan async, output penalaran, penggunaan token, dan penanganan kesalahan.

Oleh CheapTokens TeamDiterbitkan 3 menit baca

Kode Python dengan for loop atas chunk stream dan teks muncul token demi token
Di halaman ini
  1. Persiapan
  2. Contoh minimal
  3. Model penalaran
  4. Penggunaan token
  5. asyncio
  6. Berhenti lebih awal dan timeout
  7. Kesalahan
  8. Apa yang ada di wire

Dengan streaming, jawaban tiba sepotong demi sepotong saat model masih menghasilkannya, sehingga pengguna Anda langsung melihat teks alih-alih menunggu seluruh respons. API ini menggunakan format streaming standar OpenAI (server-sent events), dan paket Python resmi openai menanganinya untuk Anda.

Setiap contoh di bawah ini lengkap: salin ke file dan jalankan. Contoh-contoh ini diuji dengan Python 3.10.12 dan openai 3.31.0 terhadap server uji lokal yang mereproduksi format streaming API (sintaks dan eksekusi).

Persiapan#

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

Contoh minimal#

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

Dua detail penting:

  • Lewati chunk tanpa choices. Beberapa chunk tidak membawa teks — misalnya chunk terakhir dengan penggunaan token.
  • delta.content bisa kosong. Chunk pertama sering hanya membawa role, dan yang terakhir hanya finish_reason. Cetak teks hanya saat ada.

Model penalaran#

zai-org/GLM-5.3-Flash bernalar sebelum menjawab. Penalaran datang di field terpisah dari delta, reasoning_content, dan jawabannya sendiri di content. SDK mempertahankan field yang tidak dikenal, jadi Anda bisa membacanya dengan getattr. Ini contoh lengkap dengan kliennya sendiri dan streamnya sendiri:

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

Penalaran bisa memakan waktu sebelum kata pertama jawaban muncul — alasan lain untuk streaming.

Penggunaan token#

Minta penggunaan dengan stream_options; itu datang di chunk terakhir, yang tidak memiliki 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}")

Kalikan angka-angka ini dengan tarif model untuk mendapatkan biaya permintaan — lihat Cara memperkirakan biaya agen AI.

asyncio#

Dalam kode async gunakan AsyncOpenAI dan async for. Klien memiliki namanya sendiri, async_client, agar tidak bentrok dengan client sinkron dari contoh lain:

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

Berhenti lebih awal dan timeout#

Tinggalkan loop untuk menghentikan generasi. Membuka stream dengan with menutup koneksi saat blok berakhir. Timeout klien diatur secara eksplisit; SDK juga mencoba ulang beberapa permintaan yang gagal sendiri (max_retries, 2 secara default).

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)

Kesalahan#

Kesalahan yang terjadi sebelum chunk pertama — kunci salah, saldo kosong, model tidak dikenal — dimunculkan sebagai pengecualian saat Anda membuat 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)

Daftar lengkap kode status dan perbaikannya ada di Kesalahan koneksi umum dan cara memperbaikinya.

Apa yang ada di wire#

Untuk debugging, lihat stream mentah dengan curl (-N mematikan 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"}]}'

Setiap event adalah baris data: {…} dengan chunk JSON; stream berakhir dengan data: [DONE]. Baris yang dimulai dengan titik dua adalah komentar keep-alive yang dikirim selama jeda panjang — klien mengabaikannya.

Tag: