Zurück zum Blog

API & SDKAnleitung

Streaming-Antworten in Python

Gib die Antwort des Modells aus, während sie generiert wird, mit dem offiziellen openai-Paket — Sync- und Async-Beispiele, Reasoning-Ausgabe, Token-Nutzung und Fehlerbehandlung.

Von CheapTokens TeamVeröffentlicht am 3 Min. Lesezeit

Python-Code mit einer for-Schleife über Stream-Chunks und Text, der Token für Token erscheint
Auf dieser Seite
  1. Setup
  2. Minimales Beispiel
  3. Reasoning-Modelle
  4. Token-Nutzung
  5. asyncio
  6. Vorzeitig stoppen und Timeouts
  7. Fehler
  8. Was auf der Leitung passiert

Beim Streaming trifft die Antwort Stück für Stück ein, während das Modell sie noch generiert, sodass deine Nutzer sofort Text sehen, anstatt auf die gesamte Antwort zu warten. Die API verwendet das Standard-Streaming-Format von OpenAI (Server-Sent Events), und das offizielle openai Python-Paket übernimmt das für dich.

Jedes Beispiel unten ist vollständig: Kopiere es in eine Datei und führe es aus. Die Beispiele wurden mit Python 3.10.12 und openai 3.31.0 gegen einen lokalen Testserver geprüft, der das Streaming-Format der API reproduziert (Syntax und Ausführung).

Setup#

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

Minimales Beispiel#

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

Zwei Details sind wichtig:

  • Überspringe Chunks ohne choices. Manche Chunks enthalten keinen Text — zum Beispiel der letzte Chunk mit der Token-Nutzung.
  • delta.content kann leer sein. Der erste Chunk enthält oft nur die Rolle, und der letzte nur finish_reason. Gib Text nur aus, wenn er vorhanden ist.

Reasoning-Modelle#

zai-org/GLM-5.3-Flash denkt nach, bevor es antwortet. Das Reasoning kommt in einem separaten Feld des Deltas, reasoning_content, und die Antwort selbst in content. Das SDK behält unbekannte Felder bei, sodass du es mit getattr lesen kannst. Dies ist ein vollständiges Beispiel mit eigenem Client und eigenem 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()

Reasoning kann eine Weile dauern, bevor das erste Wort der Antwort erscheint — ein weiterer Grund zu streamen.

Token-Nutzung#

Fordere die Nutzung mit stream_options an; sie kommt im letzten Chunk, der kein choices hat:

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}")

Multipliziere diese Zahlen mit den Raten des Modells, um die Kosten der Anfrage zu erhalten — siehe How to estimate the cost of an AI agent.

asyncio#

Verwende in asynchronem Code AsyncOpenAI und async for. Der Client hat seinen eigenen Namen, async_client, damit er nicht mit dem synchronen client aus den anderen Beispielen kollidiert:

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

Vorzeitig stoppen und Timeouts#

Verlasse die Schleife, um die Generierung zu stoppen. Das Öffnen des Streams mit with schließt die Verbindung, wenn der Block endet. Der Client-Timeout wird explizit gesetzt; das SDK wiederholt auch einige fehlgeschlagene Anfragen von selbst (max_retries, standardmäßig 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)

Fehler#

Fehler, die vor dem ersten Chunk auftreten — falscher Schlüssel, leeres Guthaben, unbekanntes Modell — werden als Ausnahmen ausgelöst, wenn du den Stream erstellst:

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)

Eine vollständige Liste der Statuscodes und Lösungen findest du in Common connection errors and how to fix them.

Was auf der Leitung passiert#

Zum Debuggen schaue dir den Rohstream mit curl an (-N schaltet die Pufferung aus):

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

Jedes Event ist eine data: {…}-Zeile mit einem JSON-Chunk; der Stream endet mit data: [DONE]. Zeilen, die mit einem Doppelpunkt beginnen, sind Keep-Alive-Kommentare, die während langer Pausen gesendet werden — Clients ignorieren sie.

Tags: