Streaming optimisé

Ce que le streaming change vraiment

Le streaming ne rend pas le modèle plus rapide : il rend l’attente supportable. Sur une réponse qui met cinq secondes à se former, l’utilisateur sans streaming fixe un écran vide pendant cinq secondes, puis reçoit un bloc de texte. Avec le streaming, il voit les premiers mots après quelques centaines de millisecondes et lit pendant que le modèle écrit. La durée totale est identique, la perception ne l’est pas du tout. C’est pourquoi, en production, se passer du streaming est presque toujours une erreur de conception plutôt qu’un choix.

Encore faut-il que le flux traverse toute votre chaîne. Un flux généré correctement par l’API, mais accumulé par votre serveur applicatif ou mis en tampon par votre reverse proxy, arrive à l’utilisateur d’un seul bloc : vous avez payé la complexité du streaming sans en toucher le bénéfice. Cette leçon suit donc le chemin complet, de l’appel API jusqu’au navigateur.

Consommer le flux côté client

Avec la Responses API, activer stream=True transforme la réponse en une suite d’événements typés que vous parcourez au fur et à mesure. Chaque fragment de texte arrive dans un événement response.output_text.delta ; la fin du flux est signalée par response.completed, qui porte l’objet réponse complet, utile pour récupérer l’usage en tokens.

import openai

client = openai.OpenAI()

def generer_en_streaming(prompt: str):
    """Génération en streaming avec la Responses API."""
    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)
        elif event.type == "response.completed":
            print()  # Nouvelle ligne à la fin
            return event.response

Le flush=True n’est pas cosmétique : sans lui, Python met lui-même la sortie en tampon et vous retrouvez le comportement bloc que vous cherchiez à éviter. Ce détail se rejoue à chaque couche de la chaîne, et c’est la principale difficulté du streaming.

Le mécanisme couvre également les appels de fonctions. Les arguments d’un outil se construisent progressivement, et vous pouvez les afficher au fil de la génération pour signaler à l’utilisateur qu’une recherche est en train d’être formulée, plutôt que de le laisser devant une interface figée pendant que le modèle prépare son appel.

outils = [
    {
        "type": "function",
        "name": "rechercher_produit",
        "description": "Recherche un produit dans le catalogue.",
        "parameters": {
            "type": "object",
            "properties": {
                "requête": {"type": "string"},
                "categorie": {"type": "string"},
            },
            "required": ["requête"],
        },
    }
]

def streaming_avec_outils(prompt: str):
    stream = client.responses.create(
        model="gpt-5.6-terra",
        input=prompt,
        tools=outils,
        stream=True,
    )

    for event in stream:
        match event.type:
            case "response.output_text.delta":
                print(event.delta, end="", flush=True)
            case "response.function_call_arguments.delta":
                # Arguments de la fonction en cours de génération
                print(f"[arg: {event.delta}]", end="")
            case "response.function_call_arguments.done":
                print(f"\n[Appel fonction terminé]")

Relayer le flux jusqu’au navigateur

Votre frontend ne parle pas à l’API OpenAI directement — il parle à votre serveur, qui doit donc relayer le flux. Les Server-Sent Events sont le format naturel pour cela : une réponse HTTP qui ne se ferme pas et dont chaque ligne data: porte un fragment JSON. L’exemple suivant relaie le texte au fil de l’eau, puis émet un dernier message contenant la comptabilité des tokens, ce qui vous permet de facturer ou de journaliser sans second appel.

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

app = FastAPI()
client = openai.OpenAI()

@app.post("/api/chat")
async def chat_streaming(requete: dict):
    prompt = requete["message"]

    async def generer():
        stream = client.responses.create(
            model="gpt-5.6-terra",
            input=prompt,
            stream=True,
        )

        for event in stream:
            if event.type == "response.output_text.delta":
                donnees = json.dumps({"texte": event.delta})
                yield f"data: {donnees}\n\n"
            elif event.type == "response.completed":
                usage = event.response.usage
                donnees = json.dumps({
                    "fin": True,
                    "tokens_entree": usage.input_tokens,
                    "tokens_sortie": usage.output_tokens,
                })
                yield f"data: {donnees}\n\n"

    return StreamingResponse(
        generer(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",  # Important pour nginx
        },
    )

Reste le reverse proxy, coupable habituel des streamings qui « ne marchent pas en production alors que ça marchait en local ». Par défaut, nginx accumule la réponse d’un backend avant de la transmettre. La configuration ci-dessous désactive ce comportement pour la route concernée uniquement.

location /api/chat {
    proxy_pass http://127.0.0.1:8000;
    proxy_buffering off;
    proxy_cache off;
    proxy_set_header Connection "";
    chunked_transfer_encoding on;
}

Tenir dans la durée

Un flux ouvert pendant plusieurs dizaines de secondes est plus exposé qu’un appel court : une coupure réseau, un quota atteint, et la génération s’interrompt au milieu d’une phrase. La stratégie de reprise doit distinguer les deux cas, parce qu’ils n’appellent pas la même patience. Une erreur de connexion se retente avec un backoff exponentiel, qui laisse au réseau le temps de se rétablir sans marteler l’API ; un dépassement de quota se retente après une attente fixe, puisque la fenêtre de limitation se réinitialise à intervalle régulier.

import openai

def streaming_resilient(prompt: str, max_tentatives: int = 3):
    """Streaming avec gestion d'erreurs et retry."""
    for tentative in range(max_tentatives):
        try:
            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
                    yield event.delta

            return  # Succès, on sort

        except openai.APIConnectionError:
            if tentative < max_tentatives - 1:
                import time
                time.sleep(2 ** tentative)  # Backoff exponentiel
            else:
                raise
        except openai.RateLimitError:
            import time
            time.sleep(5)  # Attente fixe pour rate limit

Une dernière retouche améliore le confort de lecture. Émettre chaque token produit un texte qui tressaute, surtout si votre interface applique une animation ou si le résultat part vers une synthèse vocale. Accumuler jusqu’à la fin de phrase donne un rythme plus proche de la lecture humaine, au prix d’un léger délai supplémentaire sur le premier fragment.

def streaming_par_phrase(prompt: str):
    """Accumule et yield par phrase complète."""
    buffer = ""

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

    for event in stream:
        if event.type == "response.output_text.delta":
            buffer += event.delta
            # Chercher une fin de phrase
            for separateur in [".\n", "!\n", "?\n", ". ", "! ", "? "]:
                if separateur in buffer:
                    parties = buffer.split(separateur, 1)
                    yield parties[0] + separateur.rstrip()
                    buffer = parties[1] if len(parties) > 1 else ""

    if buffer.strip():
        yield buffer.strip()

Notez le if buffer.strip() final : sans lui, une réponse qui ne se termine pas par une ponctuation forte perdrait ses derniers mots. Ce genre d’oubli passe inaperçu en test et se manifeste en production sur les réponses tronquées par max_output_tokens.

Points clés à retenir

• Utilisez toujours le streaming en production pour réduire la latence perçue
• Configurez nginx avec proxy_buffering off pour le SSE
• Gérez les erreurs avec retry et backoff exponentiel
• Accumulez par phrases pour un affichage plus naturel