Streaming SSE

Recevoir les réponses en temps réel

Par défaut, l’API attend que la réponse complète soit générée avant de la retourner. Avec le streaming, vous recevez les tokens au fur et à mesure de leur génération. Cette technique réduit le temps perçu par l’utilisateur et permet d’afficher les réponses progressivement, exactement comme le fait l’interface Grok.

Comment fonctionne le streaming

Le streaming utilise le protocole Server-Sent Events (SSE). Au lieu d’une seule réponse HTTP, le serveur envoie un flux continu d’événements. Chaque événement contient un fragment de la réponse. Le client les assemble pour reconstituer le texte complet.

Le paramètre stream: true active ce mode sur n’importe quel endpoint de génération.

Streaming en Python avec le SDK OpenAI

from openai import OpenAI
import os

client = OpenAI(
    api_key=os.getenv("XAI_API_KEY"),
    base_url="https://api.x.ai/v1"
)

stream = client.chat.completions.create(
    model="grok-4",
    messages=[
        {"role": "user", "content": "Expliquez les principes SOLID en programmation."}
    ],
    stream=True
)

for chunk in stream:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

print()  # Saut de ligne final

Explication du code

• stream=True active le streaming
• La réponse est un itérateur de chunk (fragments)
• Chaque chunk contient un delta avec le nouveau contenu
• end="" empêche les sauts de ligne entre chaque fragment
• flush=True force l’affichage immédiat

Streaming avec l’endpoint /v1/responses

stream = client.responses.create(
    model="grok-4",
    input="Décrivez les étapes pour déployer une application web.",
    stream=True
)

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

print()

L’endpoint /v1/responses utilise des événements typés. Le type response.output_text.delta contient les fragments de texte.

Streaming avec cURL

cURL supporte nativement le streaming SSE. Ajoutez "stream": true au corps de la requête :

curl -N https://api.x.ai/v1/chat/completions \
  -H "Authorization: Bearer $XAI_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "grok-4",
    "messages": [{"role": "user", "content": "Bonjour"}],
    "stream": true
  }'

L’option -N désactive le buffering de cURL pour afficher les événements dès leur réception.

Chaque ligne du flux SSE commence par data: suivi d’un objet JSON. Le dernier événement est data: [DONE] qui signale la fin de la génération.

Quand utiliser le streaming

Recommandé

• Interfaces utilisateur : l’utilisateur voit la réponse apparaître mot par mot
• Réponses longues : évite un temps d’attente de plusieurs secondes avant le premier affichage
• Applications interactives : chatbots, assistants, outils de rédaction

Pas nécessaire

• Traitements batch : vous n’affichez pas le résultat en temps réel
• Pipelines automatisés : vous avez besoin de la réponse complète pour la traiter
• Appels courts : pour une réponse de quelques mots, le streaming ajoute de la complexité sans bénéfice

Gestion d’erreurs en streaming

En mode streaming, les erreurs peuvent survenir au milieu du flux. Votre code doit gérer ces interruptions :

try:
    stream = client.chat.completions.create(
        model="grok-4",
        messages=[{"role": "user", "content": "Test"}],
        stream=True
    )
    for chunk in stream:
        if chunk.choices[0].delta.content:
            print(chunk.choices[0].delta.content, end="")
except Exception as e:
    print(f"\nErreur pendant le streaming : {e}")

Points clés à retenir

• Ajoutez stream=True pour recevoir les tokens au fur et à mesure
• Le protocole utilisé est SSE (Server-Sent Events)
• En Python, itérez sur les chunks et affichez delta.content
• Avec cURL, utilisez -N pour désactiver le buffering
• Réservez le streaming aux cas où l’affichage progressif apporte de la valeur