Tableau de bord temps réel

Trois semaines après l'ouverture, la directrice de Lin & Lumière pose quatre questions : combien de clients l'assistant a-t-il servis seul, combien de temps attendent ceux qu'il transfère, combien coûte une conversation, et est-ce que ça ralentit le soir. Si la réponse demande un export et un tableur, personne ne la regardera. Le tableau de bord répond à ces quatre questions en direct, à partir de la base qui sert déjà au routage.

Le mécanisme

Aucun système de métriques séparé : chaque réponse de l'agent est enregistrée avec sa latence, ses jetons, son coût et les outils appelés. Les indicateurs sont recalculés depuis SQLite et poussés au navigateur par Server-Sent Events (SSE) : une réponse HTTP qui reste ouverte et dans laquelle le serveur écrit un événement data: … à chaque changement.

flowchart LR
  T["Tour d'agent"] -->|"latence, jetons, coût, outils"| M[("messages")]
  E["Escalade, prise en charge"] -->|"escalade_le, prise_le"| C[("conversations")]
  M --> I["indicateurs() toutes les 1,5 s"]
  C --> I
  I -->|"changé ?"| D{"diff"}
  D -->|"oui : data: {...}"| N["EventSource du navigateur"]
  D -->|"non : : ping"| N
  N --> V["6 tuiles, activité par minute, outils, dernières conversations"]

Mise en œuvre

Mesurer au bon endroit

Le coût se calcule dans le tour d'agent, à partir de usage_metadata de chaque événement ADK. Un tour avec appel d'outil fait deux appels au modèle : les deux sont comptés.

if ev.usage_metadata:
    um = ev.usage_metadata
    j_in += um.prompt_token_count or 0
    j_out += (um.candidates_token_count or 0) + (um.thoughts_token_count or 0)
...
cout = (j_in * self.prix[0] + j_out * self.prix[1]) / 1e6

Les jetons de raisonnement (thoughts_token_count) sont facturés au prix de sortie ; les oublier sous-estime le coût dès que le modèle raisonne avant de répondre. Les prix viennent d'une table datée dans config.py, surchargée par PRIX_ENTREE et PRIX_SORTIE. Un modèle inconnu produit un avertissement au démarrage plutôt qu'un coût faux en silence.

Calculer depuis la base

def centile(valeurs: list[float], p: float) -> float:
    """Centile par rang le plus proche ; 0 si vide."""
    if not valeurs:
        return 0
    v = sorted(valeurs)
    return v[max(0, math.ceil(p / 100 * len(v)) - 1)]

Le centile par rang renvoie toujours une latence réellement observée, jamais une interpolation entre deux mesures : sur 20 réponses, la p95 est la 19ᵉ plus lente.

Pousser au navigateur

async def flux():
    precedent = None
    while not await request.is_disconnected():
        courant = json.dumps(indicateurs(store))
        if courant != precedent:
            yield f"data: {courant}\n\n"
            precedent = courant
        else:
            yield ": ping\n\n"  # garde la connexion ouverte derrière les proxys
        await asyncio.sleep(1.5)

Côté navigateur, quatre lignes : new EventSource(url), onmessage qui met à jour les tuiles, onerror qui masque la pastille « en direct ». La reconnexion après une coupure est faite par le navigateur lui-même. Le graphique d'activité est un SVG dessiné à la main dans tableau.js : 30 barres, une infobulle, pas de bibliothèque.

Pourquoi SSE plutôt que WebSocket : le flux ne va que du serveur au navigateur, SSE passe par HTTP ordinaire (proxys, Cloud Run, authentification par jeton dans l'URL) et se reconnecte seul. Pourquoi pas un simple sondage toutes les 2 s, comme la console : avec un écran ouvert toute la journée, SSE n'envoie qu'un commentaire : ping de quelques octets quand rien ne change, au lieu de renvoyer tout l'état à chaque requête.

make run      # puis http://localhost:8080/tableau-de-bord?token=<ADMIN_TOKEN>
make e2e      # l'étape 5 lit /api/metrics et vérifie +4 conversations, +1 escalade

Résultat obtenu

Étape 5 de make e2e (Vertex AI, gemini-3.5-flash-lite, 2026-10-16), après 4 conversations dont une escaladée :

{
 "conversations": 4,
 "par_canal": {"web": 2, "whatsapp": 2},
 "escalades": 1,
 "taux_resolution": 0.75,
 "reponses_agent": 5,
 "latence_p50_ms": 2806,
 "latence_p95_ms": 3321,
 "cout_usd": 0.00548,
 "jetons": 14825,
 "outils": {"rechercher_aide": 2, "suivre_colis": 1, "consulter_commande": 1,
            "demander_retour": 1, "transferer_a_un_conseiller": 1}
}
  OK  indicateurs cohérents (+4 conversations, +1 escalade)

Environ 3 000 jetons et 0,0011 $ par réponse de l'agent : la consigne et les déclarations des cinq outils pèsent plus que la question du client.

Ce qui casse en production

• Le taux de résolution flatte. Un client qui ferme le widget sans réponse utile compte comme « résolu sans humain ». Ajoutez un signal explicite (« cette réponse vous a aidé ? ») ou mesurez les clients qui reviennent dans les 48 h pour le même motif.
• Le flux SSE se coupe toutes les deux minutes. Sur Cloud Run, la durée maximale d'une requête (--timeout 120 dans make cloud-up) s'applique au flux. EventSource se reconnecte, le tableau clignote. Montez le délai jusqu'à 3600 s pour la page de supervision, ou acceptez la reconnexion.
• Les événements arrivent par paquets, avec 30 s de retard. Un proxy (nginx, certains équilibreurs) met la réponse en mémoire tampon. Cache-Control: no-cache ne suffit pas toujours : derrière nginx, ajoutez X-Accel-Buffering: no.
• La base ralentit avec dix écrans ouverts. Chaque flux recalcule tout, toutes les 1,5 s, en relisant 24 h de messages. Simple et suffisant pour une boutique de 1 200 commandes par mois ; au-delà, un seul calcul partagé par tous les flux, ou des agrégats par minute.
• Le coût affiché est à 0. Le modèle a changé (variable MODEL) et n'est pas dans la table de prix. Le journal de démarrage le signale : « prix inconnu pour … ».

La règle de décision

Mesurez d'abord ce qui se trouve déjà sur le chemin de la requête (latence, jetons, outils) et stockez-le avec le message : une métrique calculée à côté finit par diverger. Ajoutez un outil d'observabilité dédié (Cloud Monitoring, OpenTelemetry) le jour où vous avez plusieurs instances ou besoin d'alertes ; pas avant.

Pratique

Exercice. Ajoutez une tuile « Escalades pour panne » qui distingue les escalades décidées par l'agent de celles causées par une erreur technique. Indice : regardez ce que le service écrit dans messages.outils quand un tour échoue.

pannes = [c for c in escaladees if c["motif_escalade"] == "Erreur technique de l'assistant"]
...
"escalades_panne": len(pannes),

• Je sais définir précisément chaque indicateur et son biais
• Je sais compter le coût d'un tour, jetons de raisonnement compris
• Je sais pousser des indicateurs par SSE et ce qui coupe le flux derrière un proxy
• J'ai ouvert le tableau de bord et vu les tuiles changer pendant une conversation