Streaming des reponses

Le probleme d’experience utilisateur

Quand vous envoyez une requete a Claude, le modele peut mettre entre 10 et 30 secondes a generer sa reponse complete. Sans streaming, votre utilisateur fixe un ecran vide pendant tout ce temps, sans savoir si quelque chose se passe.

Le streaming resout ce probleme : au lieu d’attendre la reponse entiere, vous recevez les tokens au fur et a mesure qu’ils sont generes. L’utilisateur voit le texte apparaitre progressivement, exactement comme dans l’interface chat de Claude.

Comment ca marche : les evenements SSE

Le streaming utilise le protocole Server-Sent Events (SSE). Le serveur envoie un flux d’evenements structures, chacun contenant une partie de la reponse.

Voici les types d’evenements dans l’ordre ou ils arrivent :

Un message typique produit cette sequence :

message_start → content_block_start → content_block_delta (x N) → content_block_stop → message_delta → message_stop

Implementation basique avec stream=True

La facon la plus directe d’activer le streaming est de passer stream=True a l’appel API :

import anthropic

client = anthropic.Anthropic()

with client.messages.stream(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Explique le fonctionnement d'un moteur diesel."}]
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

Le parametre flush=True est important : il force l’affichage immediat de chaque fragment, sans attendre un saut de ligne.

La methode simplifiee : client.messages.stream()

Le SDK Python fournit un context manager qui simplifie considerablement le streaming. Plutot que de gerer chaque evenement SSE manuellement, vous utilisez stream.text_stream :

import anthropic

client = anthropic.Anthropic()

with client.messages.stream(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "Ecris un court poeme sur la mer."}
    ]
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

print()  # Saut de ligne final

stream.text_stream est un iterateur qui vous donne uniquement les fragments de texte, sans vous encombrer avec les evenements de structure (start, stop, delta).

Recuperer le message complet apres streaming

Parfois, vous avez besoin a la fois du streaming (pour l’affichage) et du message complet (pour le traiter ensuite). La methode get_final_message() vous donne le message reconstitue apres la fin du flux :

import anthropic

client = anthropic.Anthropic()

with client.messages.stream(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "Donne-moi 3 idees de projets Python pour debutants."}
    ]
) as stream:
    # Affichage en temps reel
    for text in stream.text_stream:
        print(text, end="", flush=True)

    # Recuperation du message complet
    message = stream.get_final_message()

print()
print(f"\nTokens d'entree : {message.usage.input_tokens}")
print(f"Tokens de sortie : {message.usage.output_tokens}")
print(f"Raison d'arret : {message.stop_reason}")

Le message retourne est identique a ce que vous auriez obtenu sans streaming — c’est un objet Message complet avec tout le contenu, l’usage de tokens, et les metadonnees.

Gerer les evenements individuels

Si vous avez besoin d’un controle plus fin (par exemple pour mettre a jour une barre de progression ou detecter le debut d’un bloc outil), vous pouvez ecouter chaque type d’evenement :

import anthropic

client = anthropic.Anthropic()

with client.messages.stream(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "Explique la photosynthese."}
    ]
) as stream:
    for event in stream:
        if event.type == "content_block_start":
            print("[Debut du bloc]")
        elif event.type == "content_block_delta":
            if event.delta.type == "text_delta":
                print(event.delta.text, end="", flush=True)
        elif event.type == "content_block_stop":
            print("\n[Fin du bloc]")
        elif event.type == "message_stop":
            print("\n[Message termine]")

Quand utiliser le streaming

• Applications interactives : chatbots, assistants, interfaces conversationnelles
• Reponses longues : articles, analyses detaillees, generation de code
• Feedback utilisateur : montrer que le systeme travaille

Le streaming n’ajoute aucun cout supplementaire et ne change pas le contenu de la reponse. C’est purement une amelioration de l’experience utilisateur.

Resume