Streaming des Réponses en Python

Le problème que résout le streaming

Avec chat.complete(), vous attendez que le modèle ait généré l’intégralité de sa réponse avant d’en recevoir le moindre caractère. Sur une réponse courte, l’attente passe inaperçue. Sur une analyse détaillée, un long extrait de code ou un raisonnement en plusieurs paragraphes, l’utilisateur reste devant un écran vide pendant plusieurs secondes — et quelques secondes de silence dans une interface de chat suffisent à faire croire que l’application est plantée.

Le streaming change cette perception : la réponse arrive au fil de l’eau, chunk par chunk, et s’affiche progressivement comme dans les interfaces de chat que vous connaissez. Le temps total de génération ne diminue pas d’une milliseconde, mais le premier mot apparaît presque immédiatement, et c’est ce délai-là que l’utilisateur ressent.

Utiliser chat.stream()

Le SDK Mistral V2 expose la méthode chat.stream(), qui retourne un itérateur de chunks au lieu d’un objet unique :

import os
from mistralai import Mistral

client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])

stream = client.chat.stream(
    model="mistral-small-latest",
    messages=[
        {
            "role": "user",
            "content": "Décrivez les étapes pour déployer une application Flask sur un serveur Linux."
        }
    ]
)

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

# Retour à la ligne final
print()

Deux détails font toute la différence dans cette boucle. Le test if content est indispensable, car certains chunks arrivent sans texte et provoqueraient un affichage de None. Et flush=True force Python à écrire immédiatement sur la sortie plutôt que d’attendre le remplissage de son tampon — sans lui, tout le texte apparaîtrait d’un bloc à la fin, ce qui annulerait exactement le bénéfice recherché.

Ce que contient un chunk

Chaque chunk transporte un delta, c’est-à-dire un fragment de la réponse en cours de construction. En l’inspectant, vous accédez aussi bien au texte qu’aux métadonnées de fin :

for chunk in stream:
    delta = chunk.data.choices[0].delta

    # Le contenu textuel (peut être None pour certains chunks)
    texte = delta.content

    # La raison d'arrêt (None sauf pour le dernier chunk)
    fin = chunk.data.choices[0].finish_reason

    if texte:
        print(texte, end="", flush=True)

    if fin:
        print(f"\n[Fin du stream : {fin}]")

Le flux suit toujours la même chorégraphie : le premier chunk porte généralement le rôle assistant, les chunks intermédiaires apportent le texte par petits morceaux, et le dernier livre le finish_reason. C’est donc sur ce dernier chunk que vous saurez si la réponse s’est terminée normalement ou si elle a été coupée par max_tokens.

Afficher et conserver en même temps

Le streaming a un revers : le texte défile, mais vous ne le détenez nulle part. Or une application réelle doit souvent archiver la réponse en base de données pour l’historique de conversation. La solution consiste à accumuler les fragments pendant l’affichage :

import os
from mistralai import Mistral

client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])

stream = client.chat.stream(
    model="mistral-small-latest",
    messages=[
        {"role": "user", "content": "Listez cinq frameworks Python pour le web."}
    ]
)

reponse_complete = []

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

print()

# La réponse complète reconstituée
texte_final = "".join(reponse_complete)
print(f"\nLongueur totale : {len(texte_final)} caractères")

Notez le passage par une liste puis un "".join() final plutôt qu’une concaténation de chaînes à chaque tour de boucle : sur une réponse longue, la seconde approche recopie tout le texte accumulé à chaque fragment reçu.

La variante asynchrone

Dans un serveur web — FastAPI, aiohttp — la version synchrone bloque le thread pendant toute la génération, ce qui gèle les autres requêtes en cours. Le SDK fournit donc chat.stream_async() :

import asyncio
import os
from mistralai import Mistral

async def generer_reponse():
    client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])

    stream = await client.chat.stream_async(
        model="mistral-small-latest",
        messages=[
            {"role": "user", "content": "Expliquez le concept de coroutine en Python."}
        ]
    )

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

    print()

asyncio.run(generer_reponse())

La structure est identique à la version synchrone, aux mots-clés await et async for près. Dès que votre code tourne dans un serveur web, c’est cette variante qu’il faut employer.

Quand streamer, quand s’en passer

Dans une interface de chat, le streaming s’impose toujours : c’est le confort de lecture qui est en jeu. Côté API backend, il vaut d’être activé dès lors que votre frontend sait consommer des Server-Sent Events. Pour toute réponse dépassant quelques phrases, le gain de latence perçue justifie la complexité supplémentaire. En revanche, dans un script batch qui traite mille documents et écrit les résultats dans un fichier, personne ne regarde l’écran : chat.complete() reste plus simple et parfaitement adapté.

Points clés à retenir

• chat.stream() retourne un itérateur de chunks pour un affichage progressif
• Le contenu se trouve dans chunk.data.choices[0].delta.content
• Utilisez flush=True dans vos print() pour un affichage en temps réel
• Reconstruisez la réponse complète en accumulant les chunks si nécessaire
• La version asynchrone chat.stream_async() est recommandée pour les applications web