Le streaming en temps réel

Sans streaming, votre code attend que Claude ait fini d’écrire avant de recevoir quoi que ce soit. Avec le streaming, la réponse arrive morceau par morceau, comme sur claude.ai. Cette leçon montre les deux façons de diffuser une réponse avec le SDK Python, l’ordre des événements reçus, ce que la réflexion des modèles actuels y ajoute, et le gain mesurable : le temps avant le premier jeton.

Pourquoi diffuser

Une requête ordinaire ne rend la main qu’une fois toute la réponse générée. Pour une application interactive, c’est une attente muette : le carnet mesurait 4,2 secondes avant la moindre lettre pour un essai de 500 jetons sur Claude 3 Haiku, et 47 secondes pour 1 000 jetons sur Claude 3 Opus. En streaming, le premier fragment arrivait en 0,5 seconde dans le premier cas, 1,9 dans le second. Le streaming ne raccourcit pas la génération totale, il fait arriver le début plus tôt : c’est le temps avant le premier jeton (TTFT). Il sert aussi aux longues réponses, que le SDK refuse de traiter sans streaming au-delà d’une dizaine de minutes estimées.

Le flux brut : stream=True

Passer stream=True à messages.create renvoie un itérateur d’événements serveur (SSE) au lieu d’un objet Message.

from anthropic import Anthropic

client = Anthropic()

stream = client.messages.create(
    model="claude-opus-5-5",
    max_tokens=4096,
    messages=[{"role": "user", "content": "Écris une phrase de trois mots, sans préambule."}],
    stream=True,
)
for event in stream:
    print(event.type)

Les événements arrivent dans un ordre fixe : un message_start, qui porte un Message encore vide et les jetons d’entrée ; puis, pour chaque bloc de contenu, un content_block_start, un ou plusieurs content_block_delta et un content_block_stop ; puis un ou plusieurs message_delta, qui portent stop_reason et les jetons de sortie ; enfin un message_stop. Des événements ping peuvent s’intercaler, et une erreur, par exemple overloaded_error en période de forte charge, peut arriver dans le flux sous forme d’événement error. La documentation prévient que de nouveaux types d’événements peuvent apparaître : votre code doit ignorer ceux qu’il ne connaît pas.

Afficher le texte, et seulement le texte

Le carnet affichait event.delta.text pour chaque content_block_delta. Sur les modèles actuels, ce code plante : les deltas ont plusieurs types. Un bloc de réflexion envoie des thinking_delta (avec un champ thinking, vide par défaut sur la famille 5.5) suivis d’un signature_delta ; un appel d’outil envoie des input_json_delta ; seul le texte envoie des text_delta. On filtre donc sur le type du delta.

stream = client.messages.create(
    model="claude-opus-5-5",
    max_tokens=4096,
    messages=[{"role": "user", "content": "Comment fonctionnent les grands modèles de langage ?"}],
    stream=True,
)
for event in stream:
    if event.type == "message_start":
        print(f"[jetons d'entrée : {event.message.usage.input_tokens}]")
    elif event.type == "content_block_delta" and event.delta.type == "text_delta":
        print(event.delta.text, end="", flush=True)
    elif event.type == "message_delta":
        print(f"\n[arrêt : {event.delta.stop_reason}, jetons de sortie : {event.usage.output_tokens}]")

end="" évite le retour à la ligne après chaque fragment, flush=True force l’affichage immédiat. Les jetons d’entrée viennent du message_start, les jetons de sortie du message_delta, réflexion comprise.

Les helpers du SDK

client.messages.stream() est la méthode que recommande le SDK : un gestionnaire de contexte qui accumule la réponse et expose deux outils. stream.text_stream itère sur les seuls fragments de texte, sans tri à faire ; stream.get_final_message() renvoie, une fois le flux terminé, le Message complet, comme l’aurait fait messages.create.

with client.messages.stream(
    model="claude-opus-5-5",
    max_tokens=4096,
    messages=[{"role": "user", "content": "Écris un sonnet sur les orchidées."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    message = stream.get_final_message()

print(f"\n\n{message.stop_reason}, {message.usage.output_tokens} jetons de sortie")

Le carnet utilisait la version asynchrone, AsyncAnthropic avec async with et async for ; elle s’écrit de la même façon et convient aux serveurs web. Si vous n’avez pas besoin d’afficher le texte au fil de l’eau mais voulez éviter les délais d’expiration des longues requêtes, get_final_message() seul suffit : vous diffusez sans rien afficher et récupérez le message complet.

Mesurer le temps avant le premier jeton

import time

def measure_ttft(model, prompt, max_tokens=4096):
    start = time.time()
    first = None
    with client.messages.stream(model=model, max_tokens=max_tokens,
                                messages=[{"role": "user", "content": prompt}]) as stream:
        for _ in stream.text_stream:
            if first is None:
                first = time.time() - start
        message = stream.get_final_message()
    total = time.time() - start
    print(f"{model} : premier jeton de texte {first:.2f} s, total {total:.2f} s, "
          f"{message.usage.output_tokens} jetons de sortie")

measure_ttft("claude-haiku-5-5", "Écris un long essai sur l'histoire de la Révolution française.")

Avec un modèle qui réfléchit, le premier jeton de texte arrive après la réflexion : le temps mesuré inclut ce temps de réflexion, d’autant plus long que l’effort est élevé. Par défaut, la famille 5.5 n’envoie pas le texte de sa réflexion (affichage « omitted ») : le serveur saute ces jetons dans le flux, ce qui accélère l’arrivée du texte. Les jetons de réflexion restent facturés.

À vous de jouer

Un chatbot qui écrit en direct. Reprenez le chatbot de la leçon 2 et diffusez chaque réponse avec client.messages.stream() : affichez text_stream au fil de l’eau, puis ajoutez stream.get_final_message().content à l’historique comme message assistant.

Comparer. Mesurez le temps avant le premier jeton et le temps total de la même demande sur Haiku 5.5 et Opus 5.5, avec et sans streaming (pour la version sans streaming, le premier jeton arrive avec le dernier). Recommencez avec output_config={"effort": "low"} sur Opus 5.5.

Sources officielles : 05_Streaming.ipynb (anthropics/courses, archivé) · Streaming messages et Thinking (documentation Claude : types d’événements, deltas de réflexion, affichage omis)

Points clés à retenir

• Le streaming ne raccourcit pas la génération, il réduit le temps avant le premier jeton, et il est conseillé pour les requêtes longues.
• Ordre des événements : message_start, puis par bloc content_block_start, content_block_delta, content_block_stop, puis message_delta et message_stop ; ping et error peuvent s’intercaler.
• Les deltas ont plusieurs types (text_delta, thinking_delta, signature_delta, input_json_delta) : on n’affiche que les text_delta.
• client.messages.stream() fournit text_stream et get_final_message() ; c’est la méthode recommandée.
• Sur un modèle qui réfléchit, le premier jeton de texte arrive après la réflexion, dont les jetons restent facturés.