Bumalik sa blog

API at SDKGabay

Streaming responses sa Python

I-print ang sagot ng modelo habang ginagawa ito gamit ang opisyal na openai package — sync at async na mga halimbawa, reasoning output, token usage at error handling.

Ni CheapTokens TeamNai-publish noong 3 min na pagbabasa

Python code na may for loop sa mga stream chunk at tekstong lumilitaw nang token by token
Sa pahinang ito
  1. Setup
  2. Minimal na halimbawa
  3. Reasoning models
  4. Token usage
  5. asyncio
  6. Maagang paghinto at timeouts
  7. Mga error
  8. Ano ang nasa wire

Sa streaming, dumarating ang sagot nang paisa-isa habang ginagawa pa ito ng modelo, kaya agad na nakikita ng mga user mo ang teksto sa halip na maghintay sa buong tugon. Gumagamit ang API ng standard na OpenAI streaming format (server-sent events), at ang opisyal na openai Python package ang humahawak nito para sa iyo.

Kumpleto ang bawat halimbawa sa ibaba: kopyahin ito sa isang file at patakbuhin. Sinuri ang mga halimbawa gamit ang Python 3.10.12 at openai 3.31.0 laban sa isang lokal na test server na gumagaya sa streaming format ng API (syntax at execution).

Setup#

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

Minimal na halimbawa#

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

Dalawang detalye ang mahalaga:

  • Laktawan ang mga chunk na walang choices. May mga chunk na walang teksto — halimbawa ang huling chunk na may token usage.
  • Maaaring walang laman ang delta.content. Ang unang chunk ay madalas na role lang ang dala, at ang huli ay finish_reason lang. I-print ang teksto kapag naroon lang ito.

Reasoning models#

Nangangatwiran muna ang zai-org/GLM-5.3-Flash bago ito sumagot. Dumarating ang reasoning sa hiwalay na field ng delta, ang reasoning_content, at ang sagot mismo sa content. Pinapanatili ng SDK ang mga hindi kilalang field, kaya mababasa mo ito gamit ang getattr. Ito ay kumpletong halimbawa na may sariling client at sariling stream:

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

Maaaring tumagal ang reasoning bago lumitaw ang unang salita ng sagot — isa pang dahilan para mag-stream.

Token usage#

Humingi ng usage gamit ang stream_options; dumarating ito sa huling chunk, na walang 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}")

I-multiply ang mga numerong ito sa mga rate ng modelo para makuha ang gastos ng request — tingnan ang How to estimate the cost of an AI agent.

asyncio#

Sa async code gamitin ang AsyncOpenAI at async for. May sariling pangalan ang client, async_client, para hindi ito sumalungat sa synchronous na client mula sa ibang mga halimbawa:

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

Maagang paghinto at timeouts#

Lumabas sa loop para ihinto ang generation. Ang pagbubukas ng stream gamit ang with ay nagsasara ng koneksyon kapag natapos ang block. Ang client timeout ay itinakda nang tahasan; awtomatikong nagre-retry din ang SDK ng ilang nabigong request (max_retries, 2 bilang 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)

Mga error#

Ang mga error na nangyayari bago ang unang chunk — maling key, walang laman na balance, hindi kilalang modelo — ay itinatapon bilang exceptions kapag ginawa mo ang 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)

Ang buong listahan ng status codes at mga fix ay nasa Common connection errors and how to fix them.

Ano ang nasa wire#

Para mag-debug, tingnan ang raw stream gamit ang curl (pinapatay ng -N ang 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"}]}'

Ang bawat event ay isang data: {…} na linya na may JSON chunk; nagtatapos ang stream sa data: [DONE]. Ang mga linyang nagsisimula sa colon ay keep-alive comments na ipinapadala sa mahabang paghinto — hindi pinapansin ng mga client ang mga ito.

Mga tag: