Streaming : recevoir les réponses en temps réel

Le streaming permet de recevoir la réponse token par token au lieu d’attendre la génération complète. Ce n’est pas un raffinement d’interface : c’est ce qui sépare un chatbot que l’on trouve lent d’un chatbot que l’on trouve vif, à performance serveur strictement identique.

Pourquoi streamer

Sans streaming, l’utilisateur attend parfois dix à trente secondes devant un écran vide avant que la réponse n’apparaisse d’un bloc. Avec le streaming, les premiers mots s’affichent en moins d’une seconde et la lecture commence pendant que la génération se poursuit. La durée totale est la même ; la perception, non. L’activation tient à un seul paramètre, stream=True, qui transforme le retour de create() en itérateur d’événements.

from openai import OpenAI
client = OpenAI()

# SANS streaming — attente complète
response = client.responses.create(
    model="gpt-5.6-terra",
    input="Écrivez un paragraphe sur l'intelligence artificielle."
)
print(response.output_text)  # Tout d'un coup après l'attente

# AVEC streaming — token par token
stream = client.responses.create(
    model="gpt-5.6-terra",
    input="Écrivez un paragraphe sur l'intelligence artificielle.",
    stream=True
)

for event in stream:
    if event.type == "response.output_text.delta":
        print(event.delta, end="", flush=True)
print()  # Nouvelle ligne à la fin

Des événements typés, pas un flux indifférencié

Le flux ne transporte pas uniquement du texte. Il vous informe de la création de la réponse, de l’ajout de chaque élément de sortie, de l’arrivée des fragments, puis de la complétion, ce dernier événement portant les métriques d’usage. Vous filtrez donc sur event.type pour savoir ce que vous manipulez, et c’est cette structure qui vous permet, par exemple, d’afficher un indicateur d’activité dès response.created et de comptabiliser les tokens à la fin sans second appel.

stream = client.responses.create(
    model="gpt-5.6-terra",
    input="Bonjour !",
    stream=True
)

for event in stream:
    match event.type:
        case "response.created":
            print(f"[Réponse créée] ID: {event.response.id}")
        case "response.output_item.added":
            print(f"[Nouvel élément de sortie]")
        case "response.output_text.delta":
            print(f"[Texte] {event.delta}", end="")
        case "response.output_text.done":
            print(f"\n[Texte complet]")
        case "response.completed":
            print(f"[Terminé] Tokens: {event.response.usage.total_tokens}")

# Résultat :
# [Réponse créée] ID: resp_abc123
# [Nouvel élément de sortie]
# [Texte] Bonjour[Texte]  ![Texte]  Comment[Texte]  puis[Texte] -je...
# [Texte complet]
# [Terminé] Tokens: 25

Dans une application réelle, vous voudrez presque toujours faire deux choses à la fois : afficher les fragments et reconstituer le texte complet, afin de l’archiver ou de le retourner à l’appelant. La règle est simple : accumulez au fil de l’itération, car une fois le flux consommé, il ne peut pas être relu.

def stream_et_collecter(prompt: str) -> str:
    """Stream la réponse tout en la collectant."""
    stream = client.responses.create(
        model="gpt-5.6-terra",
        input=prompt,
        stream=True
    )

    texte_complet = ""
    for event in stream:
        if event.type == "response.output_text.delta":
            texte_complet += event.delta
            print(event.delta, end="", flush=True)

    print()  # Nouvelle ligne
    return texte_complet

resultat = stream_et_collecter("Listez 3 capitales européennes.")
print(f"\nTexte collecté ({len(resultat)} caractères)")

Streaming et function calling

Les appels de fonctions se streament également, avec leurs propres événements : les arguments arrivent par fragments via response.function_call_arguments.delta, et response.function_call_arguments.done signale que le JSON est complet. La conséquence pratique est importante — n’essayez jamais de désérialiser les arguments avant l’événement done, vous n’auriez qu’un JSON tronqué.

stream = client.responses.create(
    model="gpt-5.6-terra",
    input="Quelle heure est-il à Tokyo ?",
    tools=[{
        "type": "function",
        "name": "get_heure",
        "description": "Obtenir l'heure dans une ville",
        "parameters": {
            "type": "object",
            "properties": {
                "ville": {"type": "string"}
            },
            "required": ["ville"]
        }
    }],
    stream=True
)

for event in stream:
    match event.type:
        case "response.output_text.delta":
            print(event.delta, end="")
        case "response.function_call_arguments.delta":
            print(f"[Args] {event.delta}", end="")
        case "response.function_call_arguments.done":
            print(f"\n[Appel de fonction complet]")

# Résultat :
# [Args] {"ville": "Tokyo"}
# [Appel de fonction complet]

Le cas asynchrone

Dès que votre application sert plusieurs utilisateurs simultanément, passez à AsyncOpenAI : le client synchrone monopoliserait la boucle d’événements pendant toute la génération, et vos autres requêtes attendraient. La syntaxe diffère de deux mots — await sur la création, async for sur l’itération.

from openai import AsyncOpenAI
import asyncio

async_client = AsyncOpenAI()

async def stream_async(prompt: str):
    """Stream asynchrone de la réponse."""
    stream = await async_client.responses.create(
        model="gpt-5.6-terra",
        input=prompt,
        stream=True
    )

    async for event in stream:
        if event.type == "response.output_text.delta":
            print(event.delta, end="", flush=True)
    print()

asyncio.run(stream_async("Racontez une blague courte."))

Côté web, le streaming ne sert à rien s’il s’arrête à votre serveur : il faut le prolonger jusqu’au navigateur. FastAPI le permet avec StreamingResponse, alimentée par un générateur qui relaie les fragments au fur et à mesure.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import OpenAI

app = FastAPI()
client = OpenAI()

@app.get("/chat")
async def chat(question: str):
    def generate():
        stream = client.responses.create(
            model="gpt-5.6-terra",
            input=question,
            stream=True
        )
        for event in stream:
            if event.type == "response.output_text.delta":
                yield event.delta

    return StreamingResponse(generate(), media_type="text/plain")

Les erreurs en cours de flux

Une particularité mérite votre vigilance : en streaming, l’erreur peut survenir après que des fragments ont déjà été affichés. L’utilisateur voit alors une réponse commencée puis interrompue. Encadrez donc l’itération elle-même, et pas seulement la création du flux, afin de pouvoir signaler proprement la coupure plutôt que de laisser une phrase en suspens.

from openai import APIError

def stream_avec_erreurs(prompt: str):
    """Stream avec gestion d'erreurs robuste."""
    try:
        stream = client.responses.create(
            model="gpt-5.6-terra",
            input=prompt,
            stream=True
        )

        for event in stream:
            if event.type == "response.output_text.delta":
                print(event.delta, end="", flush=True)
        print()

    except APIError as e:
        print(f"\nErreur API pendant le streaming : {e.message}")
    except Exception as e:
        print(f"\nErreur inattendue : {e}")

stream_avec_erreurs("Expliquez le streaming en 2 phrases.")

Points clés à retenir

Activez le streaming avec stream=True pour une expérience temps réel
Filtrez les événements par event.type pour traiter chaque type de données
response.output_text.delta contient les fragments de texte
response.completed signale la fin et inclut les métriques d’usage
Utilisez AsyncOpenAI pour le streaming dans les applications asynchrones
Gérez toujours les erreurs même en mode streaming