Le streaming en temps réel

Objectifs

• Comprendre le fonctionnement du streaming
• Travailler avec les événements de flux

Pourquoi le streaming ?

Jusqu’ici, tous vos appels ont suivi le même schéma : client.messages.create() renvoie le contenu une fois que toute la génération est terminée. Sur une réponse d’une phrase, personne ne le remarque. Sur un essai de huit cents mots, l’utilisateur fixe un écran vide pendant plusieurs secondes, sans le moindre signe que quelque chose se passe — et une part non négligeable d’entre eux conclura que l’application est cassée.

Le streaming résout ce problème en envoyant le contenu au fur et à mesure de sa génération, exactement comme sur claude.ai. Rappelez-vous que le modèle produit sa réponse un token à la fois : le texte existe donc déjà côté serveur bien avant que le message soit complet. Le streaming consiste simplement à ne plus attendre pour vous le transmettre.

Activer le streaming

Un seul argument suffit à basculer d’un mode à l’autre.

stream = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=500,
    stream=True,
    messages=[{"role": "user", "content": "Écris un court essai sur l'ocean"}]
)

Attention à ce que vous récupérez : ce n’est plus un objet Message mais un générateur qui émet des événements server-sent (SSE). Tenter d’accéder directement à stream.content[0].text échouera, puisque le contenu n’existe pas encore au moment où la fonction retourne. Vous devez parcourir le flux.

Les types d’événements

Ce flux n’est pas une simple suite de fragments de texte. Il est structuré, et chaque envoi arrive dans un ordre garanti.

1. MessageStartEvent : début du message (contenu vide)
2. ContentBlockStartEvent : début d’un bloc de contenu
3. ContentBlockDeltaEvent (un ou plusieurs) : les morceaux de texte généré
4. ContentBlockStopEvent : fin du bloc de contenu
5. MessageDeltaEvent : informations de fin (tokens utilisés)
6. MessageStopEvent : fin du message

Cette structure reproduit celle de la réponse complète que vous connaissez déjà : un message qui contient des blocs, lesquels contiennent du texte. Le texte généré se trouve dans les ContentBlockDeltaEvent, accessibles via event.delta.text ; les autres événements servent au cadrage et, pour le MessageDeltaEvent, à récupérer la consommation de tokens en fin de course.

Extraire le texte

Le traitement minimal consiste donc à filtrer les événements de type content_block_delta et à afficher leur contenu.

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

Les deux arguments de print ne sont pas décoratifs. end="" supprime le retour à la ligne automatique, sans quoi chaque fragment s’afficherait sur sa propre ligne ; flush=True force l’écriture immédiate sur la sortie standard au lieu d’attendre que le tampon se remplisse — sans lui, le texte apparaîtrait par blocs et vous perdriez précisément l’effet recherché.

Impact sur le Time to First Token (TTFT)

Le bénéfice se mesure sur une métrique précise, le temps avant le premier token, et non sur la durée totale.

Le temps total reste le même, mais l’utilisateur voit du contenu 8x plus vite. C’est un gain de perception, pas de performance brute — et c’est exactement ce qui compte dans une interface conversationnelle, où l’attente ressentie pèse davantage que la seconde réellement gagnée.

Les helpers du SDK

Écrire soi-même la boucle de filtrage devient vite répétitif. Le SDK propose client.messages.stream(), un gestionnaire de contexte qui expose directement un itérateur de texte et se charge de fermer proprement la connexion en sortie de bloc.

with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=500,
    messages=[{"role": "user", "content": "Raconte une histoire courte"}]
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

get_final_message()

Le streaming ne vous prive pas des métadonnées. Une fois le flux épuisé, get_final_message() reconstitue l’objet Message complet, avec son stop_reason et son usage — indispensable si vous journalisez la consommation ou si vous devez archiver la réponse entière.

with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=500,
    messages=[{"role": "user", "content": "Raconte une histoire"}]
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    final = stream.get_final_message()
    print(f"\nTokens utilises : {final.usage}")

Handlers d’événements personnalisés

Pour les applications plus élaborées, vous pouvez définir une classe dotée de gestionnaires d’événements plutôt que d’écrire une boucle : le SDK appellera vos méthodes au bon moment. on_text(text, snapshot) se déclenche à chaque nouveau texte, on_stream_event(event) pour tout événement sans distinction, on_message(message) une fois le message complet, on_exception(exception) en cas d’erreur, et on_end() ferme la marche comme dernier événement du flux.

Prenez une interface de support qui affiche la réponse en direct : on_text pousse chaque fragment vers le navigateur, on_message journalise la conversation et sa consommation dès que la génération est terminée, on_exception bascule sur un message de repli si la connexion lâche en cours de route. Trois responsabilités séparées, trois méthodes, là où une boucle unique vous obligerait à imbriquer les conditions les unes dans les autres.

Pour vous entraîner

Reprenez le chatbot en ligne de commande de la deuxième leçon et remplacez son appel bloquant par un appel en streaming, de sorte que les réponses de Claude s’affichent en temps réel, token par token. Une subtilité vous attend : vous devez toujours ajouter la réponse complète à l’historique de conversation, ce qui suppose d’accumuler les fragments au fil de l’affichage ou de passer par get_final_message().