Le serveur et le streaming SSE.
Fichier : packages/minilab/server.py
Votre assistant répond dans le terminal. Dernière étape : en faire un vrai service web, que n’importe quelle application peut interroger, avec une réponse qui s’affiche token après token. C’est la leçon 16 en code : streaming avec les Server-Sent Events, détokenisation au fil de l’eau, arrêt quand l’utilisateur s’en va, et mesure du délai du premier token et du débit. À la fin de la page, vous interrogerez votre propre modèle depuis ce navigateur.
Comprendre l’architecture du fichier
Le serveur expose une route, POST /v1/chat. Elle reçoit une liste de messages et quelques réglages, met la conversation en forme et la fait tenir dans la fenêtre avec le code de l’étape 5, puis génère avec celui de l’étape 4. Le modèle chargé est l’assistant fine-tuné, chat.pt, une seule fois au premier appel.
Avec stream: true, la réponse est un flux text/event-stream : un événement data: … par token, puis un dernier événement avec finish_reason, usage et la vitesse mesurée, puis data: [DONE]. C’est le format du laboratoire de la leçon 16. Avec stream: false, un seul objet JSON arrive à la fin. Sans délai du premier token, cette fois : l’utilisateur ne voit rien avant la fin, son attente est le temps total.
Trois détails comptent pour une vraie application. Si le client ferme la connexion, la génération s’arrête : pas de calcul gaspillé. Les erreurs sont explicites : 422 pour une requête mal formée, 400 si la conversation ne tient pas dans la fenêtre, 503 si le modèle n’a pas encore été fine-tuné. Enfin, le serveur autorise les pages de ce cours à l’appeler depuis le navigateur, grâce au CORS (Cross-Origin Resource Sharing).
Construire pas à pas : la requête et le modèle
Deux modèles Pydantic décrivent la requête : FastAPI les utilise pour vérifier chaque appel et renvoyer une erreur 422 claire si un champ manque ou sort des bornes, par exemple max_tokens au-delà de 100. get_engine charge le modèle et le tokenizer une seule fois, grâce à lru_cache : les requêtes suivantes réutilisent le même objet.
"""Étape 6 — un serveur FastAPI qui envoie les réponses en streaming SSE (leçon 16).
POST /v1/chat {"messages": [{"role": "user", "content": "Qui est Passepartout ?"}], "stream": true, "max_tokens": 60, "temperature": 0}
Avec "stream": true, la réponse arrive token par token, un événement par token :
data: {"type": "delta", "index": 0, "id": 412, "text": "Pass"}
puis un dernier événement avec la raison de l'arrêt, l'usage et la vitesse mesurée, etdata: [DONE]. Avec "stream": false, un seul objet JSON arrive, une fois tout terminé.Si le navigateur ferme la connexion, le serveur arrête de générer.
uv run python -m packages.minilab.server # démonstration dans le terminal uv run uvicorn packages.minilab.server:app --port 8001 # le vrai serveur"""
import asyncioimport jsonimport timefrom collections.abc import AsyncIterator, Awaitable, Callablefrom dataclasses import dataclassfrom functools import lru_cachefrom typing import Annotated, Literal
from fastapi import Depends, FastAPI, HTTPException, Requestfrom fastapi.middleware.cors import CORSMiddlewarefrom fastapi.responses import StreamingResponsefrom pydantic import BaseModel, Field
from packages.minilab.chat import CHAT_CHECKPOINT, fit_to_windowfrom packages.minilab.generate import SamplingSettings, generatefrom packages.minilab.model import MiniLabModel, loadfrom packages.minilab.tokenizer import IM_END, StreamDecoder, Tokenizer, get_tokenizer
class ChatMessage(BaseModel): role: Literal["system", "user", "assistant"] content: str = Field(max_length=500)
class ChatBody(BaseModel): messages: list[ChatMessage] = Field(min_length=1, max_length=20) stream: bool = True max_tokens: int = Field(default=60, ge=1, le=100) temperature: float = Field(default=0.0, ge=0, le=2) top_k: int | None = Field(default=None, ge=1) top_p: float | None = Field(default=None, gt=0, le=1) seed: int = 0
@dataclass(frozen=True)class Engine: model: MiniLabModel tokenizer: Tokenizer
@lru_cache(maxsize=1)def get_engine() -> Engine: """Le modèle fine-tuné de l'étape 5, chargé une seule fois.""" if not CHAT_CHECKPOINT.exists(): raise HTTPException( 503, "Modèle absent : lancez d'abord l'étape 5 (packages.minilab.chat)." ) return Engine(load(CHAT_CHECKPOINT), get_tokenizer())Construire pas à pas : les événements de la réponse
answer_events est le cœur du fichier. Ce générateur asynchrone produit un dictionnaire par événement. À chaque token, il vérifie d’abord si le client est toujours là, détokenise avec le StreamDecoder de l’étape 1, qui attend la fin des caractères coupés, puis rend un delta. La ligne await asyncio.sleep(0) rend la main au serveur, pour qu’il envoie l’événement tout de suite au lieu de les accumuler.
Le délai du premier token se mesure entre le début de la requête et le premier delta : il comprend la mise en forme, le prefill et le premier pas. Le débit compte les tokens suivants, pendant le decode. Ce sont les deux mesures de la leçon 15.
def sse(data: dict[str, object] | str) -> str: """Un événement Server-Sent Events : une ligne data:, puis une ligne vide.""" payload = data if isinstance(data, str) else json.dumps(data, ensure_ascii=False) return f"data: {payload}\n\n"
async def answer_events( engine: Engine, body: ChatBody, is_disconnected: Callable[[], Awaitable[bool]]) -> AsyncIterator[dict[str, object]]: """Génère la réponse et la décrit sous forme d'événements : un delta par token, puis done.""" started = time.perf_counter() model, tokenizer = engine.model, engine.tokenizer messages = [message.model_dump() for message in body.messages] try: _, prompt = fit_to_window(messages, tokenizer, model.config.context_length, body.max_tokens) except ValueError as error: raise HTTPException(400, str(error)) from error settings = SamplingSettings(body.temperature, body.top_k, body.top_p, body.seed) end = tokenizer.special[IM_END] decoder = StreamDecoder(tokenizer) first_at = None count, finish_reason = 0, "length" for step in generate(model, prompt, body.max_tokens, settings, stop_id=end): count += 1 if step.token_id == end: # le modèle ferme son tour : un arrêt naturel finish_reason = "stop" break if await is_disconnected(): # le client est parti : on arrête, on économise le calcul return text = decoder.feed(step.token_id) # peut attendre la fin d'un caractère UTF-8 first_at = first_at or time.perf_counter() yield {"type": "delta", "index": count - 1, "id": step.token_id, "text": text} await asyncio.sleep(0) # laisse le serveur envoyer l'événement tout de suite if rest := decoder.flush(): yield {"type": "delta", "index": count, "id": None, "text": rest} done = time.perf_counter() first_at = first_at or done decode_tokens = max(count - 1, 0) yield { "type": "done", "finish_reason": finish_reason, "usage": {"prompt_tokens": len(prompt), "completion_tokens": count}, "ttft_ms": round((first_at - started) * 1000, 1), "tokens_per_second": round(decode_tokens / (done - first_at), 1) if done > first_at else None, "total_ms": round((done - started) * 1000, 1), }Construire pas à pas : l’application et ses routes
Le middleware CORS autorise les pages de ce cours : lancé en local, sur le port 3000, en ligne sur minilab-llm.com, ou sur une prévisualisation en .vercel.app. L’option allow_private_network répond à la vérification que les navigateurs récents envoient avant qu’un site public puisse joindre votre machine, à l’adresse 127.0.0.1.
La route choisit le format. En streaming, elle enveloppe chaque événement avec sse et le renvoie dans une StreamingResponse. Sans streaming, elle reconstitue le texte et renvoie un seul objet.
app = FastAPI(title="MiniLab — serveur du projet")# Autorise les pages du cours à appeler ce serveur local depuis le navigateur : le cours# lancé en local (port 3000), en ligne sur minilab-llm.com, ou sur une prévisualisation# Vercel (*.vercel.app). allow_private_network répond à la vérification que les# navigateurs envoient avant qu'un site public puisse joindre 127.0.0.1.app.add_middleware( CORSMiddleware, allow_origins=[ "http://localhost:3000", "http://127.0.0.1:3000", "https://minilab-llm.com", "https://www.minilab-llm.com", ], allow_origin_regex=r"https://[a-z0-9-]+\.vercel\.app", allow_methods=["GET", "POST"], allow_headers=["Content-Type"], allow_private_network=True,)
@app.get("/health")def health() -> dict[str, str]: return {"status": "ok"}
@app.post("/v1/chat", response_model=None)async def chat( body: ChatBody, request: Request, engine: Annotated[Engine, Depends(get_engine)]) -> StreamingResponse | dict[str, object]: events = answer_events(engine, body, request.is_disconnected) if body.stream:
async def stream() -> AsyncIterator[str]: async for event in events: yield sse(event) yield sse("[DONE]")
return StreamingResponse( stream(), media_type="text/event-stream", headers={"Cache-Control": "no-cache"} ) # Sans streaming : on attend toute la réponse, puis on envoie un seul objet JSON. text, last = "", {} async for event in events: if event["type"] == "delta": text += str(event["text"]) else: last = event # Pas de TTFT ici : l'utilisateur ne voit rien avant la fin, l'attente est le temps total. return {"text": text, **{k: v for k, v in last.items() if k not in ("type", "ttft_ms")}}Construire pas à pas : la démonstration
main interroge le serveur sans le démarrer, avec le client de test de FastAPI : une requête en streaming, puis la même sans. Elle affiche les premiers et derniers événements bruts, le texte reconstitué et les mesures.
def main() -> None: from fastapi.testclient import TestClient
client = TestClient(app) question = "Qui est Passepartout ?" body = {"messages": [{"role": "user", "content": question}], "stream": True} print(f"POST /v1/chat, stream: true — « {question} »\n") with client.stream("POST", "/v1/chat", json=body) as response: if response.status_code != 200: raise SystemExit(response.json()["detail"]) lines = [line for line in response.iter_lines() if line] for line in lines[:3] + ["…"] + lines[-2:]: print(f" {line}") deltas = [json.loads(line[6:]) for line in lines if line.startswith("data: {")] done = deltas.pop() print(f"\nTexte reconstitué : « {''.join(d['text'] for d in deltas)} »") print(f" finish_reason : {done['finish_reason']} · usage : {done['usage']}") print( f" TTFT : {done['ttft_ms']} ms · débit : {done['tokens_per_second']} tokens/s · total : {done['total_ms']} ms" )
body["stream"] = False plain = client.post("/v1/chat", json=body).json() print( f"\nPOST /v1/chat, stream: false → un seul objet JSON :\n {json.dumps(plain, ensure_ascii=False)}" ) print("\nPour le vrai serveur :") print(" uv run uvicorn packages.minilab.server:app --port 8001") print(' curl -N http://127.0.0.1:8001/v1/chat -H "Content-Type: application/json" \\') print(' -d \'{"messages": [{"role": "user", "content": "Qui est Fix ?"}]}\'')
if __name__ == "__main__": main()Tester et observer dans le terminal
- Lancer la démonstration
uv run python -m packages.minilab.server - Lancer les tests de l’étape
uv run pytest tests/minilab/test_server.py
POST /v1/chat, stream: true — « Qui est Passepartout ? »
data: {"type": "delta", "index": 0, "id": 794, "text": "Passepartout"}
data: {"type": "delta", "index": 1, "id": 532, "text": " est"}
data: {"type": "delta", "index": 2, "id": 291, "text": " le"}
…
data: {"type": "done", "finish_reason": "stop", "usage": {"prompt_tokens": 38, "completion_tokens": 25}, "ttft_ms": 1.9, "tokens_per_second": 2071.5, "total_ms": 13.5}
data: [DONE]
Texte reconstitué : « Passepartout est le domestique français de Phileas Fogg, un garçon honnête et débrouillard. »
finish_reason : stop · usage : {'prompt_tokens': 38, 'completion_tokens': 25}
TTFT : 1.9 ms · débit : 2071.5 tokens/s · total : 13.5 ms
POST /v1/chat, stream: false → un seul objet JSON :
{"text": "Passepartout est le domestique français de Phileas Fogg, un garçon honnête et débrouillard.", "finish_reason": "stop", "usage": {"prompt_tokens": 38, "completion_tokens": 25}, "tokens_per_second": 2666.1, "total_ms": 10.1}Chaque token est un événement : « Passepartout » tient en un seul token, appris par le BPE de l’étape 1. La réponse s’arrête d’elle-même, finish_reason vaut stop, et l’usage compte 38 tokens de prompt et 25 de réponse, <|im_end|> compris. Votre modèle est si petit qu’il génère environ 2 000 tokens par seconde : le premier token arrive en 2 ms, quand un grand modèle en ligne met souvent plusieurs centaines de millisecondes.
Lancez maintenant le vrai serveur, puis interrogez-le avec curl. L’option -N désactive la mise en mémoire tampon : les événements s’affichent au fur et à mesure.
- Terminal 1 — démarrer le serveur (il reste ouvert)
uv run uvicorn packages.minilab.server:app --port 8001 - Terminal 2 — envoyer une requête
curl -N http://127.0.0.1:8001/v1/chat -H "Content-Type: application/json" -d '{"messages": [{"role": "user", "content": "Qui est Fix ?"}]}'
Interroger votre modèle depuis cette page
Votre serveur tourne sur le port 8001 ? Cette page peut lui parler directement. La question part de votre navigateur vers votre machine, et la réponse revient en streaming, générée par le modèle que vous avez entraîné vous-même. Essayez aussi la question sur l’Espagne, hors du roman, et le bouton « Arrêter » pendant une réponse.
Tester votre serveur local (port 8001)
Lancez d’abord votre serveur dans un terminal, depuis la racine de votre projet :
- Terminal — démarrer votre serveur
uv run uvicorn packages.minilab.server:app --port 8001
Envoyez la même question avec curl, en ajoutant "temperature": 0.8, "seed": 3 au corps de la requête : la réponse varie, puis revient à l’identique avec la même seed. Avec "max_tokens": 5, la réponse est coupée et finish_reason vaut length.
Le serveur FastAPI expose POST /v1/chat. En streaming, il envoie un événement SSE par token, détokenisé au fil de l’eau, puis un événement final (finish_reason, usage, délai du premier token, débit) et data: [DONE] ; sans streaming, un seul objet JSON. La génération s’arrête si le client se déconnecte, et le CORS permet aux pages du cours d’appeler 127.0.0.1:8001. Votre modèle va du texte brut jusqu’à l’écran : le projet MiniLab est complet.