Architecture de production : monitoring, alerting, scaling
Mis à jour le 28 juillet 2026
Votre application en production : les trois piliers
Déployer une application LLM en production ne s’arrête pas au code. Trois piliers garantissent la fiabilité : le monitoring pour observer, l’alerting pour réagir, le scaling pour absorber la charge. Ils forment une chaîne, et la chaîne vaut son maillon le plus faible : des métriques que personne ne regarde ne servent à rien, une alerte sans procédure de réponse non plus, et une capacité de montée en charge qui se déclenche trop tard équivaut à une panne. Cette leçon finale rassemble tous les concepts du cours dans une architecture de production complète.
Observer : ce qu’il faut mesurer
Quatre familles de mesures suffisent à couvrir l’essentiel : le volume d’appels, la latence, les erreurs et le coût, ventilés par modèle. La ventilation par modèle est ce qui rend le tableau de bord actionnable — un taux d’erreur global de 3 % ne vous apprend rien, alors que le même chiffre concentré sur un seul modèle vous désigne immédiatement le coupable.
La classe ci-dessous accumule ces mesures en mémoire, avec une fenêtre glissante de 10 000 latences pour éviter que le processus ne grossisse indéfiniment. Le calcul de percentile, plutôt que de moyenne, mérite d’être souligné : la latence moyenne d’un service LLM est presque toujours trompeuse, parce qu’une poignée d’appels très lents disparaissent dans la moyenne alors que ce sont eux que vos utilisateurs commentent.
import time
from dataclasses import dataclass, field
from collections import defaultdict
@dataclass
class MetriquesProduction:
"""Collecte les métriques essentielles en production."""
compteurs: dict = field(default_factory=lambda: defaultdict(int))
latences: dict = field(default_factory=lambda: defaultdict(list))
erreurs: dict = field(default_factory=lambda: defaultdict(int))
couts: dict = field(default_factory=lambda: defaultdict(float))
def enregistrer_appel(
self,
modele: str,
latence: float,
tokens_entree: int,
tokens_sortie: int,
cout: float,
succes: bool,
):
self.compteurs[modele] += 1
self.latences[modele].append(latence)
self.couts[modele] += cout
if not succes:
self.erreurs[modele] += 1
# Garder les 10 000 dernières latences
if len(self.latences[modele]) > 10_000:
self.latences[modele] = self.latences[modele][-10_000:]
def percentile(self, modele: str, p: float) -> float:
"""Calcule un percentile de latence."""
valeurs = sorted(self.latences.get(modele, []))
if not valeurs:
return 0.0
idx = int(len(valeurs) * p / 100)
return valeurs[min(idx, len(valeurs) - 1)]
def taux_erreur(self, modele: str) -> float:
total = self.compteurs.get(modele, 0)
if total == 0:
return 0.0
return self.erreurs.get(modele, 0) / total
def tableau_de_bord(self) -> dict:
"""Génère un tableau de bord complet."""
dashboard = {}
for modele in self.compteurs:
dashboard[modele] = {
"appels": self.compteurs[modele],
"latence_p50": f"{self.percentile(modele, 50):.2f}s",
"latence_p99": f"{self.percentile(modele, 99):.2f}s",
"taux_erreur": f"{self.taux_erreur(modele):.2%}",
"cout_total": f"${self.couts[modele]:.2f}",
}
return dashboardEncore faut-il que la collecte soit systématique. Instrumenter à la main chaque appel garantit qu’on en oubliera, et probablement celui qui posera problème. Le décorateur ci-dessous déplace la mesure au niveau de la fonction d’appel : le bloc finally enregistre la latence et le succès quoi qu’il arrive, y compris lorsque l’exception remonte.
import openai
import functools
metriques = MetriquesProduction()
def monitorer(func):
"""Décorateur qui monitore chaque appel API."""
@functools.wraps(func)
def wrapper(*args, **kwargs):
modele = kwargs.get("model", args[0] if args else "inconnu")
debut = time.perf_counter()
succes = True
try:
resultat = func(*args, **kwargs)
return resultat
except Exception as e:
succes = False
raise
finally:
latence = time.perf_counter() - debut
metriques.enregistrer_appel(
modele=modele,
latence=latence,
tokens_entree=0,
tokens_sortie=0,
cout=0,
succes=succes,
)
return wrapperLes compteurs de tokens et le coût restent à zéro dans cette version, faute d’accès à l’objet de réponse au moment de l’enregistrement. Complétez-les en lisant les champs d’usage retournés par l’API : sans cette information, votre tableau de bord affichera fidèlement une colonne de coûts nuls.
Réagir : des alertes graduées
Une alerte n’a de valeur que si elle appelle une action. C’est pourquoi le système distingue deux paliers par métrique : un seuil d’avertissement, qui signale une dérive à examiner dans la journée, et un seuil critique, qui justifie de réveiller quelqu’un. Confondre les deux produit le pire résultat possible — une équipe qui a pris l’habitude d’ignorer les notifications.
from enum import Enum
class NiveauAlerte(Enum):
INFO = "info"
WARNING = "warning"
CRITICAL = "critical"
class SystemeAlertes:
"""Déclenche des alertes basées sur les métriques."""
def __init__(self):
self.seuils = {
"taux_erreur": {"warning": 0.05, "critical": 0.15},
"latence_p99": {"warning": 10.0, "critical": 30.0},
"cout_horaire": {"warning": 50.0, "critical": 200.0},
}
self.alertes_envoyees: list[dict] = []
def verifier(self, metriques: MetriquesProduction) -> list[dict]:
"""Vérifie les métriques et génère des alertes."""
alertes = []
for modele in metriques.compteurs:
# Taux d'erreur
taux = metriques.taux_erreur(modele)
if taux >= self.seuils["taux_erreur"]["critical"]:
alertes.append({
"niveau": NiveauAlerte.CRITICAL,
"message": (
f"Taux d'erreur CRITIQUE pour {modele}: "
f"{taux:.1%}"
),
"modele": modele,
})
elif taux >= self.seuils["taux_erreur"]["warning"]:
alertes.append({
"niveau": NiveauAlerte.WARNING,
"message": (
f"Taux d'erreur élevé pour {modele}: "
f"{taux:.1%}"
),
"modele": modele,
})
# Latence
p99 = metriques.percentile(modele, 99)
if p99 >= self.seuils["latence_p99"]["critical"]:
alertes.append({
"niveau": NiveauAlerte.CRITICAL,
"message": (
f"Latence p99 CRITIQUE pour {modele}: "
f"{p99:.1f}s"
),
"modele": modele,
})
self.alertes_envoyees.extend(alertes)
return alertes
async def notifier(self, alertes: list[dict]):
"""Envoie les notifications (webhook, email, Slack)."""
import httpx
for alerte in alertes:
if alerte["niveau"] == NiveauAlerte.CRITICAL:
# Notification immédiate
async with httpx.AsyncClient() as http:
await http.post(
"https://votre-webhook.example.com/alertes",
json={
"niveau": alerte["niveau"].value,
"message": alerte["message"],
},
)L’alerte sur le coût horaire est celle qu’on oublie le plus souvent, et c’est dommage : une boucle de retry mal bornée ou un traitement par lot lancé deux fois se traduisent par une facture, pas par une erreur. Elle est le seul filet de sécurité qui détecte ce type d’incident pendant qu’il se produit.
Absorber : trois mécanismes de résilience
Quand l’API répond en erreur, réessayer aggrave la situation : vous ajoutez de la charge à un service déjà en difficulté et vous faites attendre vos propres utilisateurs pour rien. Le disjoncteur inverse la logique. Après un nombre défini d’échecs consécutifs, il ouvre le circuit et rejette immédiatement les appels suivants, puis retente au bout du délai de réinitialisation.
class CircuitBreaker:
"""Coupe les appels quand trop d'erreurs se produisent."""
def __init__(
self,
seuil_ouverture: int = 5,
delai_reset: float = 60.0,
):
self.seuil = seuil_ouverture
self.delai_reset = delai_reset
self.echecs_consecutifs = 0
self.ouvert_depuis: float | None = None
@property
def est_ouvert(self) -> bool:
if self.ouvert_depuis is None:
return False
if time.time() - self.ouvert_depuis > self.delai_reset:
# Tenter de fermer le circuit
self.ouvert_depuis = None
self.echecs_consecutifs = 0
return False
return True
def enregistrer_succes(self):
self.echecs_consecutifs = 0
self.ouvert_depuis = None
def enregistrer_echec(self):
self.echecs_consecutifs += 1
if self.echecs_consecutifs >= self.seuil:
self.ouvert_depuis = time.time()
def appeler(self, func, *args, **kwargs):
"""Exécute la fonction si le circuit est fermé."""
if self.est_ouvert:
raise RuntimeError(
"Circuit ouvert — API indisponible. "
f"Retry dans {self.delai_reset}s."
)
try:
resultat = func(*args, **kwargs)
self.enregistrer_succes()
return resultat
except Exception as e:
self.enregistrer_echec()
raiseLe disjoncteur protège votre application ; il ne la fait pas fonctionner pour autant. C’est le rôle du basculement entre modèles, qui essaie les alternatives dans l’ordre en sautant celles dont le circuit est déjà ouvert. Un point de conception à retenir : l’ordre de la liste est une décision produit, pas technique. Placez en premier le modèle qui rend le service attendu et derrière lui ceux qui rendent un service dégradé mais acceptable — mieux vaut une réponse un peu moins fine qu’un message d’indisponibilité.
class FallbackModeles:
"""Bascule vers un modèle de secours en cas de panne."""
def __init__(self):
self.modeles = ["gpt-5.6-terra", "gpt-5.6-sol", "gpt-5.6-luna"]
self.circuits = {
m: CircuitBreaker() for m in self.modeles
}
def appeler(self, prompt: str) -> str:
"""Essaie chaque modèle jusqu'à obtenir une réponse."""
client = openai.OpenAI()
for modele in self.modeles:
circuit = self.circuits[modele]
if circuit.est_ouvert:
continue
try:
response = circuit.appeler(
client.responses.create,
model=modele,
input=prompt,
)
return response.output_text
except Exception:
continue
raise RuntimeError("Tous les modèles sont indisponibles.")Le troisième mécanisme concerne le dimensionnement. Piloter le nombre de workers sur la taille de la file d’attente plutôt que sur la charge processeur est le bon réflexe pour un service qui passe l’essentiel de son temps à attendre des réponses réseau : le processeur reste au repos pendant que la file s’allonge. Notez l’asymétrie de la règle appliquée — on double à la montée et on divise par deux à la descente, avec des seuils très éloignés (100 contre 10). Cet écart empêche l’oscillation : sans lui, le système passerait son temps à créer et détruire des workers autour du point de bascule.
class AutoScaler:
"""Ajuste le nombre de workers selon la charge."""
def __init__(
self,
workers_min: int = 2,
workers_max: int = 50,
seuil_scale_up: int = 100,
seuil_scale_down: int = 10,
):
self.workers_min = workers_min
self.workers_max = workers_max
self.seuil_up = seuil_scale_up
self.seuil_down = seuil_scale_down
self.workers_actifs = workers_min
def ajuster(self, taille_file: int) -> int:
"""Retourne le nombre de workers recommandé."""
if taille_file > self.seuil_up:
self.workers_actifs = min(
self.workers_actifs * 2,
self.workers_max,
)
elif taille_file < self.seuil_down:
self.workers_actifs = max(
self.workers_actifs // 2,
self.workers_min,
)
return self.workers_actifsChecklist de mise en production
Ce cours s’achève sur la liste de vérification à passer avant d’ouvrir votre application au public. Chaque point renvoie à une leçon que vous avez traversée ; si l’un d’eux vous paraît obscur, c’est la leçon correspondante qu’il faut relire avant de déployer.
- Monitoring des latences (p50, p95, p99) et des taux d’erreur
- Alertes configurées sur les seuils critiques
- Circuit breaker pour protéger contre les pannes API
- Fallback entre modèles configuré
- Modération des entrées et sorties activée
- Rate limiting côté client implémenté
- Logs structurés pour le debugging
- Évals automatisées dans la CI/CD
- Budget de coût avec alertes
- Documentation des prompts versionnée
Points clés à retenir
- Le monitoring couvre latence, erreurs, coûts et qualité des réponses
- Suivez les percentiles de latence, jamais la moyenne seule
- Les alertes doivent être graduées : warning puis critical
- Le circuit breaker protège contre les cascades de pannes et le fallback assure la disponibilité
- Une checklist de production évite les oublis critiques
Testez vos connaissances
Latence, coûts, évals : les réflexes de production en cinq questions.
1. Quels sont les grands leviers de réduction de latence ?
Réponse : Le streaming (latence perçue), le prompt caching (préfixes stables), les Predicted Outputs quand la réponse est largement prévisible, et l’optimisation du prompt lui-même — mesurés bout en bout.
2. Quand utiliser la Batch API ?
Réponse : Pour les traitements massifs sans contrainte temps réel : lots asynchrones à coût réduit — le classement, l’enrichissement ou la génération en volume n’ont rien à faire sur le chemin synchrone.
3. Que fait la compaction des conversations ?
Réponse : Elle compresse l’historique long (résumés, élagage) pour tenir dans le contexte et maîtriser les coûts — indispensable aux agents et sessions longues.
4. À quoi sert le framework Evals en CI/CD ?
Réponse : À exécuter automatiquement vos évaluations à chaque changement de prompt ou de modèle : la régression se détecte au commit, pas en production.
5. Que comprend une architecture de production complète ?
Réponse : Monitoring (latence, coûts, qualité), alerting, files asynchrones, modération à l’échelle et plan de scaling — l’IA en production est un système, pas un appel API.
Mesurer, mettre en cache, batcher, évaluer en continu : quatre habitudes qui font les applications rapides, sobres et fiables.