Architecture de production : monitoring, alerting, scaling

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 (observer), l’alerting (réagir) et le scaling (absorber la charge). Cette leçon finale rassemble tous les concepts du cours dans une architecture de production complète.

Monitoring : observer tout

Métriques essentielles

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 dashboard

Middleware de monitoring

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 wrapper

Alerting : réagir vite

Système d’alertes

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"],
                        },
                    )

Scaling : absorber la charge

Circuit breaker

Protégez votre application quand l’API est en panne :

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()
            raise

Fallback entre modèles

class FallbackModeles:
    """Bascule vers un modèle de secours en cas de panne."""

    def __init__(self):
        self.modeles = ["gpt-5.3", "gpt-5.4", "o4-mini"]
        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.")

Auto-scaling basé sur la file d’attente

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_actifs

Checklist de mise en production

Avant de déployer votre application LLM, vérifiez ces points :

• 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
• Les alertes doivent être graduées : warning puis critical
• Le circuit breaker protège contre les cascades de pannes
• Le fallback entre modèles assure la disponibilité
• Une checklist de production évite les oublis critiques