Streaming des Réponses en Python

Pourquoi utiliser le streaming ?

Avec chat.complete(), vous attendez que le modèle génère l’intégralité de sa réponse avant de recevoir quoi que ce soit. Pour une réponse courte, c’est imperceptible. Mais pour une réponse longue (plusieurs paragraphes, du code, une analyse détaillée), l’utilisateur fixe un écran vide pendant plusieurs secondes.

Le streaming résout ce problème : la réponse arrive mot par mot (ou plutôt, chunk par chunk), ce qui permet un affichage progressif similaire à ce que vous voyez sur les interfaces de chat IA. L’expérience utilisateur s’en trouve radicalement améliorée.

Utiliser chat.stream()

Le SDK Mistral V2 fournit la méthode chat.stream() qui retourne un itérateur de chunks :

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()

Différences avec chat.complete()

Anatomie d’un chunk

Chaque chunk du stream contient un delta — un fragment de la réponse :

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 premier chunk contient souvent le rôle (assistant), les chunks suivants contiennent le texte progressivement, et le dernier chunk porte le finish_reason.

Reconstruire la réponse complète

Si vous avez besoin à la fois du streaming pour l’affichage et de la réponse complète pour la stocker en base de données :

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")

Streaming asynchrone

Pour les applications web (FastAPI, aiohttp), utilisez la version asynchrone :

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 version asynchrone est essentielle dans les serveurs web où vous ne voulez pas bloquer le thread principal pendant la génération.

Quand utiliser le streaming ?

• Interfaces de chat : toujours, pour une expérience fluide
• API backend : quand votre frontend supporte les Server-Sent Events (SSE)
• Scripts batch : rarement nécessaire, chat.complete() suffit
• Réponses longues : dès que la réponse dépasse quelques phrases

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