Sortie en streaming

L'utilisateur pose une question, l'écran reste vide, trois secondes passent, et toute la réponse apparaît d'un coup. Toujours trois secondes, mais si le premier caractère apparaît après une demi-seconde et que la suite s'affiche caractère après caractère, l'utilisateur le vit bien mieux : il sait que le programme travaille, et il peut lire en attendant.

C'est la sortie en streaming (streaming). La leçon 2 du module 01 l'a montré : le modèle génère de toute façon token par token. Le streaming consiste simplement à vous envoyer chaque petit morceau dès qu'il est généré, au lieu d'attendre d'avoir tout pour tout envoyer.

Activer le streaming

En ajoutant stream=True à la requête, on ne reçoit plus une réponse complète, mais un flux de données que l'on parcourt avec une boucle for :

stream = client.chat.completions.create(
    model=MODEL,
    messages=[{"role": "user", "content": "用大约 200 字介绍 httpx 和 requests 的主要区别。"}],
    stream=True,
    extra_body={"thinking": {"type": "disabled"}},
)
for chunk in stream:
    if chunk.choices and chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)
print()

Chaque chunk est un petit morceau de données, et le texte nouvellement généré se trouve dans chunk.choices[0].delta.content. delta signifie « incrément » : il ne contient que le nouveau contenu de ce morceau, pas tout le contenu jusqu'ici. C'est à vous de les assembler.

Le end="" de print colle les morceaux sans retour à la ligne, et flush=True les affiche immédiatement au lieu d'attendre que le tampon soit plein. Sans flush=True, le texte apparaît par paquets, et l'effet de streaming disparaît.

Mesure : combien plus rapide ?

code/03-llm-apps/streaming.py pose la même question en streaming et sans, note le moment où apparaît le premier caractère et celui où tout est fini, avec et sans réflexion :

def streaming(thinking, show=False):
    start = time.time()
    first_content = None
    stream = client.chat.completions.create(
        model=MODEL, messages=QUESTION, stream=True,
        stream_options={"include_usage": True},  # 让最后一个数据块带上 usage
        extra_body={"thinking": {"type": "enabled" if thinking else "disabled"}},
    )
    usage = None
    for chunk in stream:
        if chunk.usage:
            usage = chunk.usage
        if not chunk.choices:
            continue
        delta = chunk.choices[0].delta
        if delta.content:
            if first_content is None:
                first_content = time.time() - start
            if show:
                print(delta.content, end="", flush=True)
    if show:
        print()
    return first_content, time.time() - start, usage.completion_tokens

Mon résultat (le script affiche d'abord une sortie en streaming ; ici je ne garde que les mesures) :

不思考 非流式：第一个字 1.82 秒，全部完成 1.82 秒，输出 153 词元
不思考 流式：  第一个字 0.67 秒，全部完成 1.71 秒，输出 202 词元
开思考 非流式：第一个字 2.48 秒，全部完成 2.48 秒，输出 330 词元
开思考 流式：  第一个字 1.82 秒，全部完成 2.75 秒，输出 379 词元

Sans réflexion, le premier caractère apparaît en streaming à 0,67 seconde, alors que sans streaming, il faut attendre 1,82 seconde pour voir quoi que ce soit.

Remarquez que le temps de « fin complète » est à peu près le même dans les deux cas (1,71 contre 1,82 seconde, et les deux réponses n'ont pas la même longueur). Le streaming ne fait pas générer le modèle plus vite ; le temps total est inchangé, seul change le moment où l'utilisateur commence à voir du contenu. Plus la réponse est longue, plus la différence est nette : pour une longue réponse qui met 20 secondes à se générer, sans streaming, l'utilisateur fixe un écran vide pendant 20 secondes.

Avec la réflexion, le premier caractère n'apparaît en streaming qu'à 1,82 seconde, car le modèle réfléchit d'abord et ne commence la réponse officielle qu'ensuite. La réflexion est elle aussi renvoyée en streaming, dans delta.reasoning_content ; pour montrer à l'utilisateur « réflexion en cours… », vous pouvez l'afficher, ou seulement afficher un indicateur.

Obtenir usage en streaming

Sans streaming, response.usage indique directement le nombre de tokens utilisés. En streaming, cette information n'est pas fournie par défaut.

Avec stream_options={"include_usage": True}, le serveur envoie à la fin du flux un morceau supplémentaire contenant usage, mais dont choices est une liste vide. D'où le if not chunk.choices: continue du code ci-dessus ; sans cette ligne, accéder à chunk.choices[0] lève une IndexError.

Quelques points d'attention en streaming

finish_reason est dans le dernier morceau. En streaming, le finish_reason des premiers morceaux vaut None ; seul le dernier morceau avec contenu prend une valeur comme stop ou length. Pour vérifier si la réponse a été tronquée, notez-le dans la boucle.

Une erreur peut survenir en cours de route. Un appel sans streaming réussit ou échoue d'un bloc. Un appel en streaming peut se couper après avoir produit la moitié de la réponse, que l'utilisateur a déjà vue. Réessayer génère alors une nouvelle réponse depuis le début, qui ne correspondra pas forcément à la moitié déjà affichée. La pratique courante : une erreur pendant l'établissement de la connexion peut être réessayée ; une erreur après le début de la sortie se signale à l'utilisateur (« la réponse a été interrompue »), qui décide s'il repose la question. C'est ce que fait RepoBot à la leçon 5.

Il faut reconstituer soi-même la réponse complète. Dans une conversation à plusieurs tours, la réponse du modèle doit être stockée dans l'historique comme message assistant. En streaming, il n'y a pas de réponse complète toute prête : il faut assembler tous les delta.content.

Envoyer le streaming au navigateur

En ligne de commande, print suffit. Pour une page web, il faut relayer en temps réel la sortie du modèle vers le navigateur via votre backend. La méthode la plus courante est SSE (Server-Sent Events) : un moyen, pris en charge nativement par les navigateurs, pour que le serveur pousse continuellement des messages vers le navigateur. Son format est très simple : chaque message est une ligne data: contenu, suivie d'une ligne vide.

Un exemple minimal avec FastAPI (code/03-llm-apps/streaming_web.py) :

import json
import os

from fastapi import FastAPI
from fastapi.responses import HTMLResponse, StreamingResponse
from openai import AsyncOpenAI

client = AsyncOpenAI(  # 网页服务要同时应付很多请求，用异步客户端
    api_key=os.environ["LLM_API_KEY"],
    base_url=os.environ.get("LLM_BASE_URL", "https://api.deepseek.com"),
)
MODEL = os.environ.get("LLM_MODEL", "deepseek-flash")
app = FastAPI()


@app.get("/chat")
async def chat(q: str):
    async def events():
        stream = await client.chat.completions.create(
            model=MODEL,
            messages=[{"role": "user", "content": q}],
            stream=True,
            extra_body={"thinking": {"type": "disabled"}},
        )
        async for chunk in stream:
            if chunk.choices and chunk.choices[0].delta.content:
                # SSE 的格式：每条消息以 "data: " 开头，以空行结尾
                yield f"data: {json.dumps(chunk.choices[0].delta.content, ensure_ascii=False)}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(events(), media_type="text/event-stream")

Quelques points clés :

• Utiliser AsyncOpenAI et non OpenAI. Un service web doit traiter de nombreux utilisateurs à la fois ; un client synchrone bloque tout le service pendant qu'il attend le modèle, tandis qu'un client asynchrone peut traiter d'autres requêtes pendant l'attente.
• StreamingResponse reçoit un générateur ; à chaque yield, un morceau de données part vers le navigateur.
• Chaque morceau est encodé avec json.dumps. La sortie du modèle peut contenir des retours à la ligne, et SSE sépare les messages par des retours à la ligne ; les mettre tels quels casserait le format, alors qu'en chaîne JSON le problème disparaît.
• On envoie un [DONE] final pour signaler la fin au navigateur.

Côté navigateur, on reçoit avec EventSource :

const source = new EventSource("/chat?q=" + encodeURIComponent(question));
source.onmessage = (e) => {
  if (e.data === "[DONE]") { source.close(); return; }
  out.textContent += JSON.parse(e.data);
};

Le code complet de la page est dans streaming_web.py. Installer les dépendances et démarrer :

uv add fastapi uvicorn
uvicorn streaming_web:app --port 8000

Ouvrez http://127.0.0.1:8000 dans le navigateur pour essayer. Vous pouvez aussi voir les données SSE brutes avec curl ; le paramètre -N fait afficher chaque morceau dès sa réception :

curl -N "http://127.0.0.1:8000/chat?q=用一句话介绍httpx"

Voici les premiers messages que j'ai vus :

data: "HTTP"

data: "X"

data: " "

data: "是一个"

data: "功能"

Chaque message fait un ou deux tokens. À chaque message reçu, le navigateur l'ajoute à la page.

EventSource ne peut envoyer que des requêtes GET : la question doit tenir dans l'URL, de longueur limitée, et on ne peut pas facilement y joindre l'historique de conversation. Dans un vrai projet, on envoie généralement une requête POST avec fetch et on lit le flux renvoyé. C'est ce que fait la leçon 6 du module 06 lors du déploiement de RepoBot.

Quand ne pas utiliser le streaming

• Quand le résultat est destiné à un programme. Par exemple l'extraction JSON de la leçon 4 du module 02 : le programme a besoin du JSON complet pour l'analyser, le streaming ne sert à rien.
• Pour les traitements par lots en arrière-plan. Personne n'attend devant l'écran ; le streaming ne ferait que compliquer le code.

Partout où quelqu'un attend une réponse devant l'écran, il faut utiliser le streaming.

Exercices

1. Dans streaming.py, remplacez la question par « écris un article de 800 caractères présentant httpx », et comparez de nouveau le temps du premier caractère avec et sans streaming. L'écart s'est-il creusé ?
2. Modifiez la fonction streaming pour afficher aussi delta.reasoning_content en gris (ou avec une autre marque) quand la réflexion est activée, afin que l'utilisateur voie à quoi « pense » le modèle.
3. Ajoutez une fonction à streaming_web.py : à la fin du flux, envoyer un message supplémentaire indiquant au navigateur combien de tokens ont été utilisés (pensez à stream_options).

Auto-test

1. Le streaming permet-il au modèle de générer plus vite une réponse complète ?

Non. Le temps total pour générer une réponse complète ne change pratiquement pas. Ce que change le streaming, c'est le moment où l'utilisateur voit le premier caractère : le contenu s'affiche au fur et à mesure, sans attendre la fin de la génération.

2. En streaming avec stream_options={"include_usage": True}, le programme lève une IndexError sur la ligne chunk.choices[0]. Pourquoi ?

Avec include_usage, le serveur envoie à la fin du flux un morceau ne contenant que usage, dont choices est une liste vide. Il faut vérifier que chunk.choices n'est pas vide avant d'accéder à choices[0].

3. Pourquoi un backend web doit-il utiliser AsyncOpenAI plutôt que OpenAI ?

Un client synchrone bloque pendant qu'il attend la réponse du modèle, et le serveur ne peut alors pas traiter les requêtes d'autres utilisateurs. Un client asynchrone peut passer à d'autres requêtes pendant l'attente, si bien qu'un seul processus sert de nombreux utilisateurs à la fois.