Retour au blog

API & SDKGuide

Réponses en streaming en Python

Affichez la réponse du modèle au fur et à mesure de sa génération avec le package openai officiel — exemples sync et async, sortie de raisonnement, usage des tokens et gestion des erreurs.

Par CheapTokens TeamPublié le 3 min de lecture

Code Python avec une boucle for sur les chunks du stream et du texte apparaissant token par token
Sur cette page
  1. Configuration
  2. Exemple minimal
  3. Modèles de raisonnement
  4. Usage des tokens
  5. asyncio
  6. Arrêt anticipé et timeouts
  7. Erreurs
  8. Ce qui circule sur le réseau

Avec le streaming, la réponse arrive morceau par morceau pendant que le modèle la génère encore, vos utilisateurs voient donc le texte immédiatement au lieu d'attendre la réponse complète. L'API utilise le format de streaming standard d'OpenAI (server-sent events), et le package Python officiel openai s'en charge pour vous.

Chaque exemple ci-dessous est complet : copiez-le dans un fichier et exécutez-le. Les exemples ont été vérifiés avec Python 3.10.12 et openai 3.31.0 contre un serveur de test local qui reproduit le format de streaming de l'API (syntaxe et exécution).

Configuration#

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

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

Deux détails comptent :

  • Ignorez les chunks sans choices. Certains chunks ne contiennent pas de texte — par exemple le chunk final avec l'usage des tokens.
  • delta.content peut être vide. Le premier chunk ne contient souvent que le rôle, et le dernier seulement finish_reason. N'affichez le texte que lorsqu'il est présent.

Modèles de raisonnement#

zai-org/GLM-5.3-Flash raisonne avant de répondre. Le raisonnement arrive dans un champ séparé du delta, reasoning_content, et la réponse elle-même dans content. Le SDK conserve les champs inconnus, vous pouvez donc le lire avec getattr. Voici un exemple complet avec son propre client et son propre 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()

Le raisonnement peut prendre un certain temps avant que le premier mot de la réponse n'apparaisse — une raison de plus de streamer.

Usage des tokens#

Demandez l'usage avec stream_options ; il arrive dans le dernier chunk, qui n'a pas de 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}")

Multipliez ces nombres par les tarifs du modèle pour obtenir le coût de la requête — voir Comment estimer le coût d'un agent IA.

asyncio#

En code async, utilisez AsyncOpenAI et async for. Le client a son propre nom, async_client, pour ne pas entrer en conflit avec le client synchrone des autres exemples :

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

Arrêt anticipé et timeouts#

Quittez la boucle pour arrêter la génération. Ouvrir le stream avec with ferme la connexion à la fin du bloc. Le timeout du client est défini explicitement ; le SDK réessaie aussi certaines requêtes échouées de lui-même (max_retries, 2 par défaut).

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)

Erreurs#

Les erreurs qui surviennent avant le premier chunk — mauvaise clé, solde vide, modèle inconnu — sont levées sous forme d'exceptions lorsque vous créez le 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)

Une liste complète des codes de statut et des correctifs se trouve dans Erreurs de connexion courantes et comment les corriger.

Ce qui circule sur le réseau#

Pour déboguer, examinez le stream brut avec curl (-N désactive la mise en tampon) :

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

Chaque événement est une ligne data: {…} avec un chunk JSON ; le stream se termine par data: [DONE]. Les lignes qui commencent par un deux-points sont des commentaires keep-alive envoyés pendant les longues pauses — les clients les ignorent.

Étiquettes: