Production Deployment - Déployer l'IA

Le code de la leçon précédente tourne dans ton terminal et personne d'autre que toi n'y a accès. Déployer, c'est troquer le carnet de terrain contre une station radio : il ne suffit plus que ça émette, il faut aussi savoir combien ça coûte et voir tout de suite quand le silence s'installe.

L'endpoint /chat : exposer le modèle simplement

FastAPI décrit un endpoint avec le décorateur @app.post, et la forme des données d'entrée et de sortie avec une classe Pydantic BaseModel. Tu n'as donc pas à vérifier à la main que le client a bien envoyé un champ message : la bibliothèque rejette d'elle-même une requête mal formée. Le bloc ci-dessous montre l'intégration complète d'un LLM dans un endpoint /chat qui accepte un message et renvoie la réponse de l'IA.

1from fastapi import FastAPI
2from pydantic import BaseModel
3from openai import AsyncOpenAI
4
5app = FastAPI()
6client = AsyncOpenAI()
7
8class ChatRequest(BaseModel):
9    message: str
10
11class ChatResponse(BaseModel):
12    response: str
13
14@app.post("/chat", response_model=ChatResponse)
15async def chat(request: ChatRequest):
16    completion = await client.chat.completions.create(
17        model="gpt-5",
18        messages=[{"role": "user", "content": request.message}]
19    )
20    return ChatResponse(response=completion.choices[0].message.content)

ChatRequest décrit ce qui entre, ChatResponse ce qui sort, et response_model veille à ce que la réponse ait exactement cette forme. La classe AsyncOpenAI autorise le await, si bien que le serveur continue de servir d'autres clients pendant l'attente. En revanche, l'instruction return rend toute la réponse d'un seul coup : l'utilisateur fixe le vide une quinzaine de secondes avant de voir le premier mot.

Le streaming : StreamingResponse et générateur async

Un return classique ne se laissera jamais convaincre de faire du streaming. Ce n'est pas une affaire de réglage, mais de définition : return termine la fonction et rend une valeur, donc cette valeur doit déjà être complète au moment où elle part. Les fichiers temporaires n'y changent rien non plus, puisque l'écriture n'a lieu qu'une fois le texte entier reçu : le client attend tout autant, avec un détour par le disque en prime. La chose est pourtant parfaitement possible, et elle tient en deux pièces : une fonction qui rend le texte par morceaux, et la classe StreamingResponse qui l'emballe dans une réponse HTTP.

1from fastapi.responses import StreamingResponse
2
3@app.post("/chat/stream")
4async def chat_stream(request: ChatRequest):
5    async def generate():
6        stream = await client.chat.completions.create(
7            model="gpt-5",
8            messages=[{"role": "user", "content": request.message}],
9            stream=True
10        )
11        async for chunk in stream:
12            content = chunk.choices[0].delta.content
13            if content:
14                yield f"data: {content}\n\n"
15        yield "data: [DONE]\n\n"
16
17    return StreamingResponse(generate(), media_type="text/event-stream")

Toute la différence tient dans le mot yield. La fonction generate ne renvoie pas de résultat et ne se termine pas : elle cède un fragment, puis se met en pause jusqu'à ce qu'on lui en réclame un autre. Une telle fonction est un générateur async, et StreamingResponse pousse ses fragments vers le client au fil de l'eau, sans attendre le dernier.

Deux pannes silencieuses méritent d'être retenues car aucune des deux ne lève d'erreur. Sans la condition if content:, le mot None atterrit au milieu du flux, parce que les fragments de début et de fin portent un delta.content vide. Et sans le yield final avec [DONE], le client fait tourner son indicateur de chargement alors que le serveur a terminé son travail depuis longtemps.

WebSocket : quand un seul sens ne suffit plus

Le streaming ne règle que la moitié du problème. Le serveur parle plus longtemps, mais toujours en réponse à une question posée. HTTP fonctionne sur le schéma requête-réponse et le serveur n'a aucun moyen de prendre la parole en premier : or, dans les chatbots d'IA, tu veux pouvoir pousser une notification ou le résultat d'une tâche que personne n'est en train de réclamer.

WebSocket maintient une seule connexion ouverte dans laquelle les deux côtés envoient des messages à n'importe quel moment : c'est exactement cela, une communication bidirectionnelle en temps réel. Et c'est la seule vraie raison d'y recourir. Ce n'est pas plus rapide que HTTP, puisque les deux circulent sur le même TCP et que l'écart sur une requête isolée reste indécelable. Ce n'est pas plus sécurisé non plus : ws:// voyage en clair, et wss:// utilise le même TLS que https, ni plus ni moins. Et il est encore plus faux de croire que ça ne nécessite pas de serveur : ça en exige davantage, puisque le serveur garde la connexion ouverte pendant toute la conversation au lieu de la refermer aussitôt.

1from fastapi import WebSocket, WebSocketDisconnect
2
3active_connections: dict[str, WebSocket] = {}
4
5@app.websocket("/ws/{client_id}")
6async def websocket_endpoint(websocket: WebSocket, client_id: str):
7    await websocket.accept()
8    active_connections[client_id] = websocket
9    try:
10        while True:
11            question = await websocket.receive_text()
12            answer = await ask_model(question)
13            await websocket.send_text(answer)
14    except WebSocketDisconnect:
15        del active_connections[client_id]

La fonction ask_model n'est autre que l'appel au modèle du premier exemple, et la boucle while True traite les messages successifs sans jamais rompre la connexion. Le dictionnaire active_connections conserve les connexions par identifiant de client : c'est précisément ce qui te permet d'écrire à quelqu'un qui n'a rien demandé. Oublier de supprimer l'entrée dans WebSocketDisconnect est une fuite de mémoire : le serveur tourne des semaines et le dictionnaire enfle de connexions mortes que rien ne nettoie.

Mise en cache et limites de débit

Dans les applications d'IA, chaque appel au modèle est facturé au token : si cent personnes posent la même question, tu paies cent fois la même réponse. La mise en cache réduit les coûts d'API et accélère les réponses, puisqu'un résultat déjà mémorisé revient en quelques millisecondes, sans toucher au réseau. Prétendre qu'il n'y a aucun avantage serait donc faux sur les deux tableaux, celui de la facture comme celui de la latence.

1from cachetools import TTLCache
2from slowapi import Limiter
3from slowapi.util import get_remote_address
4
5limiter = Limiter(key_func=get_remote_address)
6response_cache = TTLCache(maxsize=1000, ttl=3600)
7
8@app.post("/chat/cached")
9@limiter.limit("10/minute")
10async def chat_cached(request: ChatRequest):
11    key = request.message.strip().lower()
12    if key in response_cache:
13        return {"response": response_cache[key], "cached": True}
14
15    response_cache[key] = await ask_model(request.message)
16    return {"response": response_cache[key], "cached": False}

TTLCache supprime de lui-même les entrées au bout d'une heure (ttl=3600), ce qui évite aux réponses de vieillir définitivement, et la clé est le texte de la question : deux questions identiques tombent sur la même entrée.

Il faut savoir en revanche que la mise en cache n'améliore pas la qualité des réponses. Elle rend le texte qui est déjà tombé une fois, donc une réponse médiocre revient tout aussi médiocre, simplement plus vite. Elle n'est pas non plus exigée par la loi : aucun texte ne t'oblige à mettre quoi que ce soit en cache, c'est une décision de coût qui t'appartient. Le décorateur @limiter.limit("10/minute") de la bibliothèque slowapi plafonne, lui, les requêtes venant d'une même adresse, pour qu'un seul client n'épuise pas ton budget en un quart d'heure.

Avant de l'exposer au monde

Avant de déployer un LLM en production, trois choses comptent plus que tout : tester, monitorer et sécuriser les clés d'API. La clé doit venir d'une variable d'environnement, jamais du code : écrite en dur, elle part dans le dépôt, et de là sur la facture de quelqu'un d'autre. Le monitoring, lui, commence par un endpoint /health grâce auquel l'hébergeur sait que le processus est encore en vie.

1import logging
2import os
3
4logging.basicConfig(level=logging.INFO)
5logger = logging.getLogger("safari-ai")
6
7client = AsyncOpenAI(api_key=os.environ["OPENAI_API_KEY"])
8
9@app.get("/health")
10async def health_check():
11    logger.info("health ok")
12    return {"status": "healthy"}

Lire os.environ["OPENAI_API_KEY"] présente un avantage supplémentaire : quand la variable manque, l'application refuse de démarrer, au lieu d'accepter du trafic et de ne se plaindre qu'à la première question posée. Il vaut aussi la peine d'entourer l'appel au modèle d'un try, pour qu'une panne du fournisseur revienne au client sous forme de HTTPException pendant que logger.exception en garde la trace dans les logs.

Trois tentations, ensuite, auxquelles il faut résister. Utiliser le dernier modèle sans le tester paraît anodin, puisque le nom du modèle n'est qu'une chaîne de caractères, sauf qu'un nouveau modèle formate ses réponses autrement et casse en silence l'analyse que tu fais de son texte. Désactiver la journalisation pour gagner en vitesse t'économise une fraction de milliseconde et te prive de la seule source d'information sur ce qui s'est passé pendant la nuit. Exposer l'API sans limites de débit finit toujours sur une facture : la clé est la tienne, donc les requêtes des autres, c'est encore toi qui les paies.

Résumé

• Le streaming dans FastAPI s'implémente avec StreamingResponse et un générateur async, c'est-à-dire une fonction qui utilise yield.
• Un return classique ne donne pas de streaming, car il rend une valeur complète et termine la fonction. Les fichiers temporaires n'aident pas davantage, et l'idée que ce n'est pas possible est fausse : c'est parfaitement faisable.
• On choisit WebSocket parce qu'il permet une communication bidirectionnelle en temps réel. Il n'est pas plus rapide que HTTP, il n'est pas plus sécurisé, et il nécessite bel et bien un serveur qui garde la connexion ouverte.
• L'endpoint /chat repose sur @app.post et sur des modèles Pydantic : ChatRequest en entrée, ChatResponse en sortie via response_model.
• Avant de déployer un LLM en production, l'essentiel est de tester, monitorer et sécuriser les clés d'API dans des variables d'environnement. Le dernier modèle utilisé sans être testé, la journalisation désactivée et une API exposée sans limites de débit sont les trois façons les plus simples de casser la production.
• La mise en cache réduit les coûts d'API et accélère les réponses. Elle n'améliore pas leur qualité, elle n'est pas exigée par la loi, et dire qu'elle n'a aucun avantage revient à ignorer ta propre facture.

La prochaine leçon présente le patron ReAct, où l'agent alterne raisonnement (Reasoning) et action (Acting) en répétant le cycle Thought - Action - Observation. Retiens pour l'instant ceci : le déploiement n'est pas le dernier commit, c'est le premier jour où quelqu'un d'autre paie tes erreurs avec ta clé.

1# ===========================================
2# Labo Safari : Intégration FastAPI + LLM
3# ===========================================
4# Construis une API REST qui se connecte à un LLM
5# pour fournir des informations sur la faune du safari.
6
7from fastapi import FastAPI, HTTPException
8from pydantic import BaseModel
9from openai import OpenAI
10
11app = FastAPI(title="Safari AI Guide API")
12client = OpenAI(api_key="your-api-key-here")
13
14# --- Modèles requête/réponse ---
15class AnimalQuery(BaseModel):
16    animal: str
17    question: str
18    max_tokens: int = 150
19
20class GuideResponse(BaseModel):
21    animal: str
22    answer: str
23    tokens_used: int
24
25# --- Endpoint du guide de safari ---
26@app.post("/ask-guide", response_model=GuideResponse)
27async def ask_safari_guide(query: AnimalQuery):
28    """Pose une question sur n'importe quel animal au guide de safari IA."""
29    try:
30        response = client.chat.completions.create(
31            model="gpt-4",
32            messages=[
33                {
34                    "role": "system",
35                    "content": "Tu es Darwin, un guide de safari expert. "
36                               "Donne des réponses concises et factuelles sur la faune."
37                },
38                {
39                    "role": "user",
40                    "content": f"À propos du {query.animal} : {query.question}"
41                }
42            ],
43            max_tokens=query.max_tokens,
44            temperature=0.7
45        )
46
47        return GuideResponse(
48            animal=query.animal,
49            answer=response.choices[0].message.content,
50            tokens_used=response.usage.total_tokens
51        )
52    except Exception as e:
53        raise HTTPException(status_code=500, detail=str(e))
54
55# --- Endpoint de health check ---
56@app.get("/health")
57async def health_check():
58    return {"status": "active", "service": "Safari AI Guide"}
59
60# --- Lister les espèces disponibles ---
61@app.get("/species")
62async def list_species():
63    return {
64        "species": [
65            "Lion", "Éléphant", "Girafe", "Zèbre",
66            "Guépard", "Hippopotame", "Rhinocéros", "Léopard"
67        ]
68    }
69
70# TODO : Ajoute un endpoint pour les requêtes d'animaux par lot
71# class BatchQuery(BaseModel):
72#     animals: list[str]
73#     question: str
74#
75# @app.post("/ask-guide/batch")
76# async def batch_query(query: BatchQuery):
77#     results = []
78#     for animal in query.animals:
79#         single_query = AnimalQuery(animal=animal, question=query.question)
80#         result = await ask_safari_guide(single_query)
81#         results.append(result)
82#     return {"results": results}
83
84# TODO : Ajoute un endpoint de streaming
85# from fastapi.responses import StreamingResponse
86#
87# @app.post("/ask-guide/stream")
88# async def stream_guide(query: AnimalQuery):
89#     async def generate():
90#         stream = client.chat.completions.create(
91#             model="gpt-4",
92#             messages=[{"role": "user", "content": query.question}],
93#             stream=True
94#         )
95#         for chunk in stream:
96#             if chunk.choices[0].delta.content:
97#                 yield chunk.choices[0].delta.content
98#     return StreamingResponse(generate(), media_type="text/plain")
99
100print("=== Safari AI Guide API ===")
101print("Endpoints :")
102print("  POST /ask-guide - Poser une question sur un animal")
103print("  GET  /health    - Health check")
104print("  GET  /species   - Lister les espèces")
105print("\nLance avec : uvicorn main:app --reload")
106