PricingLoginStart free
AI Engineering Academy · Leçon

Diffusion dans FastAPI avec les événements envoyés par le serveur

Créez un point d’accès FastAPI qui relaie les réponses diffusées du LLM vers un client navigateur à l’aide de StreamingResponse et du type de contenu text/event-stream.

Leçon 3 sur 413 étapes

Diffusion dans FastAPI avec les événements envoyés par le serveur est une leçon AI Engineering Academy gratuite sur CoddyKit. Ceci est la leçon 3 sur 4. Tu peux lire la leçon complète ci-dessous gratuitement — puis la pratiquer en direct dans le navigateur avec un éditeur de code intégré et un tuteur IA 24/7. Elle fait partie du parcours d'apprentissage AI Engineering Academy, et ta progression se synchronise sur le web et l'application CoddyKit. Le cours AI Engineering Academy comprend 4 leçons au total.

Pourquoi utiliser les événements envoyés par le serveur pour le streaming de LLM

Les événements envoyés par le serveur (SSE) sont une norme du W3C qui permet à un serveur d’envoyer un flux d’événements textuels à un navigateur au moyen d’une seule connexion HTTP persistante. Contrairement à WebSockets, SSE est unidirectionnel, du serveur vers le client, fonctionne avec HTTP/1.1 standard, se reconnecte automatiquement en cas de déconnexion et ne nécessite aucune bibliothèque spéciale dans le navigateur. Ces caractéristiques en font le moyen de transport idéal pour diffuser les tokens d’un LLM depuis un serveur FastAPI vers une interface web.

Format filaire SSE

SSE envoie des données textuelles sous la forme d’une série de champs séparés par des sauts de ligne. Chaque événement contient un champ de type event facultatif, un champ data contenant la charge utile et un id facultatif pour la reconnexion. Les événements sont séparés par une ligne vide. Pour le streaming de LLM, envoyez chaque token sous la forme d’une ligne data: token_text\n\n, puis envoyez à la fin un événement spécial data: [DONE]\n\n pour signaler la fin du flux.

# SSE wire format example
'''
data: The\n\n
data:  capital\n\n
data:  of\n\n
data:  France\n\n
data:  is\n\n
data:  Paris\n\n
data: [DONE]\n\n
'''

# Each 'data:' line is one event.
# The double newline (\n\n) terminates each event.
# The client receives these as EventSource message events.
# The content-type must be 'text/event-stream'.

StreamingResponse dans FastAPI

StreamingResponse de FastAPI accepte une génératrice asynchrone qui produit des chaînes et les diffuse vers le client. En définissant media_type sur 'text/event-stream' et en formatant chaque chaîne produite comme un événement SSE, vous transformez n’importe quelle génératrice asynchrone en véritable flux SSE. FastAPI gère automatiquement le cycle de vie de la connexion, la transmission des données et les en-têtes HTTP.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import AsyncOpenAI
import asyncio

app = FastAPI()
async_client = AsyncOpenAI()

async def token_generator(prompt: str):
    stream = await async_client.chat.completions.create(
        model='gpt-4o-mini',
        messages=[{'role': 'user', 'content': prompt}],
        stream=True,
    )
    async for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            yield f'data: {delta}\n\n'  # SSE format
    yield 'data: [DONE]\n\n'

@app.get('/stream')
async def stream_endpoint(prompt: str):
    return StreamingResponse(
        token_generator(prompt),
        media_type='text/event-stream',
        headers={'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no'},
    )

En-têtes HTTP importants pour SSE

Trois en-têtes HTTP sont essentiels au bon fonctionnement de SSE via les mandataires et les CDN. Cache-Control: no-cache empêche les intermédiaires de mettre le flux en cache. Connection: keep-alive maintient la connexion TCP ouverte. X-Accel-Buffering: no désactive la mise en mémoire tampon des réponses par Nginx, qui regrouperait sinon les fragments et annulerait l’effet du streaming. Sans ce dernier en-tête, Nginx met en mémoire tampon toute la sortie avant de la transmettre au navigateur.

from fastapi.responses import StreamingResponse

SSE_HEADERS = {
    'Cache-Control': 'no-cache',
    'Connection': 'keep-alive',
    'X-Accel-Buffering': 'no',   # disable nginx buffering
    'Access-Control-Allow-Origin': '*',  # CORS for cross-origin clients
}

@app.get('/chat')
async def chat_stream(prompt: str):
    return StreamingResponse(
        token_generator(prompt),
        media_type='text/event-stream',
        headers=SSE_HEADERS,
    )

Événements SSE structurés avec des charges JSON

Pour des API de streaming plus riches, encodez la charge utile de chaque événement au format JSON plutôt qu’en texte brut. Vous pouvez ainsi inclure des métadonnées avec le token, par exemple son type, contenu ou appel d’outil, un identifiant de message ou un horodatage de latence. Le client du navigateur analyse le JSON de chaque événement et achemine les différents types d’événements vers différents composants de l’interface.

import json
import time

async def json_token_generator(prompt: str, session_id: str):
    t_start = time.perf_counter()
    stream = await async_client.chat.completions.create(
        model='gpt-4o-mini',
        messages=[{'role': 'user', 'content': prompt}],
        stream=True,
    )
    async for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            payload = json.dumps({
                'type': 'token',
                'content': delta,
                'session_id': session_id,
                't_ms': round((time.perf_counter() - t_start) * 1000),
            })
            yield f'data: {payload}\n\n'
    # Send completion event
    yield f'data: {json.dumps({"type": "done", "session_id": session_id})}\n\n'

Consommer SSE dans un navigateur (JavaScript)

L’API EventSource côté navigateur se connecte à un point d’accès SSE et déclenche des événements à mesure de leur arrivée. Pour le streaming de tokens, écoutez l’événement message par défaut, analysez les données comme du JSON ou traitez-les comme une chaîne brute, puis ajoutez chaque token au DOM. Gérez le marqueur [DONE] en fermant la connexion EventSource.

// Browser-side JavaScript
const prompt = 'Explain hybrid search in one paragraph.';
const url = '/stream?prompt=' + encodeURIComponent(prompt);

const source = new EventSource(url);
const output = document.getElementById('output');

source.onmessage = (event) => {
  if (event.data === '[DONE]') {
    source.close();  // stop listening
    return;
  }
  output.textContent += event.data;  // append each token
};

source.onerror = (err) => {
  console.error('SSE error:', err);
  source.close();
};

Requêtes POST avec fetch pour le streaming

EventSource prend uniquement en charge les requêtes GET, ce qui est limitant pour les invites complexes. Pour les requêtes POST, qui envoient un corps JSON contenant l’historique de la conversation, utilisez l’API fetch du navigateur avec l’API Streams afin de lire progressivement le corps de la réponse. Ce modèle est utilisé par l’interface web de ChatGPT et par la plupart des interfaces de discussion LLM en production.

// Browser-side: POST with fetch and ReadableStream
async function streamPost(messages) {
  const response = await fetch('/chat', {
    method: 'POST',
    headers: {'Content-Type': 'application/json'},
    body: JSON.stringify({messages}),
  });

  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  const output = document.getElementById('output');

  while (true) {
    const {done, value} = await reader.read();
    if (done) break;
    const text = decoder.decode(value, {stream: true});
    // Parse SSE lines
    for (const line of text.split('\n')) {
      if (line.startsWith('data: ') && line !== 'data: [DONE]') {
        output.textContent += line.slice(6);
      }
    }
  }
}

Point d’accès POST FastAPI pour le streaming de discussion

Pour le streaming de discussions basé sur POST, définissez un modèle Pydantic pour le corps de la requête, acceptez une liste de messages et diffusez la réponse du LLM. Vous pouvez ainsi transmettre l’historique complet de la conversation à chaque requête et prendre en charge les applications de discussion à plusieurs tours. Le modèle est identique à celui du streaming GET, à ceci près que vous extrayez l’invite du corps de la requête.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

class ChatRequest(BaseModel):
    messages: list[dict]
    model: str = 'gpt-4o-mini'

@app.post('/chat')
async def chat_post(request: ChatRequest):
    async def generate():
        stream = await async_client.chat.completions.create(
            model=request.model,
            messages=request.messages,
            stream=True,
        )
        async for chunk in stream:
            delta = chunk.choices[0].delta.content
            if delta:
                yield f'data: {delta}\n\n'
        yield 'data: [DONE]\n\n'

    return StreamingResponse(
        generate(),
        media_type='text/event-stream',
        headers=SSE_HEADERS,
    )

Gérer les déconnexions des clients

Lorsqu’un utilisateur quitte une page ou ferme l’onglet, la connexion HTTP se ferme et FastAPI déclenche asyncio.CancelledError dans la génératrice de streaming. Gérez toujours ce cas afin d’éviter de laisser ouvertes des requêtes de streaming LLM et d’engendrer des coûts d’API inutiles. Encapsulez votre génératrice dans un bloc try/except pour CancelledError et annulez le flux OpenAI lorsqu’une déconnexion est détectée.

from fastapi import Request

@app.get('/stream')
async def stream_with_disconnect(prompt: str, request: Request):
    async def generate_with_cancel():
        try:
            stream = await async_client.chat.completions.create(
                model='gpt-4o-mini',
                messages=[{'role': 'user', 'content': prompt}],
                stream=True,
            )
            async for chunk in stream:
                if await request.is_disconnected():
                    break  # client gone, stop generating
                delta = chunk.choices[0].delta.content
                if delta:
                    yield f'data: {delta}\n\n'
        except asyncio.CancelledError:
            pass  # client disconnected
        finally:
            yield 'data: [DONE]\n\n'

    return StreamingResponse(generate_with_cancel(), media_type='text/event-stream')

Ajouter l’authentification des requêtes

Les points d’accès de streaming en production doivent authentifier les requêtes afin d’empêcher toute utilisation non autorisée du LLM. Utilisez Depends de FastAPI avec une clé d’API ou une vérification d’en-tête JWT. L’authentification a lieu avant le démarrage de la génératrice, la surcharge est donc minimale et le flux ne commence qu’une fois l’utilisateur vérifié.

from fastapi import Header, HTTPException, Depends

VALID_API_KEYS = {'sk-demo-key-1', 'sk-demo-key-2'}

async def verify_api_key(x_api_key: str = Header(None)):
    if x_api_key not in VALID_API_KEYS:
        raise HTTPException(status_code=401, detail='Invalid API key')
    return x_api_key

@app.post('/chat')
async def authenticated_chat(
    request: ChatRequest,
    api_key: str = Depends(verify_api_key),
):
    async def generate():
        stream = await async_client.chat.completions.create(
            model=request.model,
            messages=request.messages,
            stream=True,
        )
        async for chunk in stream:
            delta = chunk.choices[0].delta.content
            if delta:
                yield f'data: {delta}\n\n'
        yield 'data: [DONE]\n\n'

    return StreamingResponse(generate(), media_type='text/event-stream', headers=SSE_HEADERS)

Tester les points d’accès SSE

Testez les points d’accès de streaming avec TestClient de FastAPI en mode streaming. Utilisez with client.stream('GET', '/stream', params={...}) as r et parcourez r.iter_lines() pour recevoir les événements SSE. Vous pouvez ainsi vérifier que les tokens sont correctement formatés, que le marqueur DONE est envoyé et que les cas d’erreur produisent des événements d’erreur SSE appropriés.

from fastapi.testclient import TestClient

def test_sse_endpoint():
    with TestClient(app) as client:
        with client.stream('GET', '/stream', params={'prompt': 'Say hi'}) as r:
            assert r.status_code == 200
            assert 'text/event-stream' in r.headers['content-type']

            events = []
            for line in r.iter_lines():
                if line.startswith('data: '):
                    events.append(line[6:])

            assert events[-1] == '[DONE]'
            full_text = ''.join(e for e in events if e != '[DONE]')
            assert len(full_text) > 0

Vérification rapide

Vérifiez votre compréhension du streaming FastAPI avec SSE présenté dans cette leçon.

Réponds-y dans la leçon interactive

Récapitulatif de la leçon

Dans cette leçon, vous avez appris que les événements envoyés par le serveur constituent le moyen de transport HTTP standard pour diffuser les tokens d’un LLM vers les clients de navigateur, que StreamingResponse avec text/event-stream transforme toute génératrice asynchrone en flux SSE dans FastAPI, et que des en-têtes essentiels, notamment X-Accel-Buffering et Cache-Control, sont nécessaires pour garantir un comportement correct derrière des mandataires. Gérez les déconnexions des clients afin d’éviter les appels d’API LLM orphelins. Nous allons maintenant étudier le streaming de réponses contenant des appels d’outils.

Gratuit pour commencer

Apprends Python avec un tuteur IA — gratuit

Écris et exécute du vrai code dans ton navigateur, obtiens de l'aide instantanée d'un tuteur IA disponible 24h/24, et reprends là où tu t'es arrêté sur le web ou dans l'app.

Cours
30
Leçons
120
Commencer gratuitementVoir le cours complet

Questions Fréquemment Posées

La leçon « Diffusion dans FastAPI avec les événements envoyés par le serveur » est-elle gratuite ?

Oui — le texte complet de « Diffusion dans FastAPI avec les événements envoyés par le serveur » est gratuit à lire ici sur le web. Pour la pratiquer de manière interactive (un éditeur de code intégré et un tuteur IA 24/7) et déverrouiller le reste du cours AI Engineering Academy, passe à CoddyKit PRO. Le cours AI Engineering Academy comprend 4 leçons au total.

Qu'est-ce que j'apprendrai dans « Diffusion dans FastAPI avec les événements envoyés par le serveur » ?

Créez un point d’accès FastAPI qui relaie les réponses diffusées du LLM vers un client navigateur à l’aide de StreamingResponse et du type de contenu text/event-stream. Tu pratiques AI Engineering Academy avec du code pratique que tu exécutes directement dans le navigateur, et un tuteur IA 24/7 répond à tes questions au fur et à mesure que tu avances dans la leçon.

Dois-je avoir de l'expérience pour commencer AI Engineering Academy ?

Aucune expérience préalable n'est requise. AI Engineering Academy sur CoddyKit est structuré pour les débutants jusqu'aux apprenants avancés, donc tu peux commencer ici ou depuis le début et avancer à ton rythme. Ceci est la leçon 3 sur 4.

Combien de temps prend la leçon « Diffusion dans FastAPI avec les événements envoyés par le serveur » ?

La plupart des leçons CoddyKit prennent environ 5–10 minutes. Chacune est courte et interactive, tu progresses régulièrement et tu repiques exactement où tu t'es arrêté sur le web et l'app.

Peux-tu écrire et exécuter du code dans cette leçon AI Engineering Academy ?

Oui. Chaque leçon AI Engineering Academy inclut un éditeur de code intégré, tu écris et exécutes du vrai code directement dans ton navigateur et tu reçois des retours IA instantanés — aucune configuration locale requise.

Toutes les leçons de ce cours

  1. Comprendre la diffusion des jetons
  2. Consommer des flux avec le SDK Python
  3. Diffusion dans FastAPI avec les événements envoyés par le serveur
  4. Gérer les appels d’outils dans les réponses diffusées
← Retour à AI Engineering Academy