Trace grading : noter les exécutions

Le tracing enregistre ce que fait votre agent. C’est indispensable, mais insuffisant : personne ne relit dix mille traces quotidiennes, et les problèmes se cachent précisément dans celles que vous n’ouvrirez jamais. Le trace grading va plus loin : il note automatiquement chaque exécution pour identifier les problèmes, mesurer la qualité dans le temps, et alimenter une boucle d’amélioration continue.

Le concept de trace grading

Chaque trace produite par votre agent contient l’intégralité de l’exécution : le message de l’utilisateur, les appels de tools, les réponses intermédiaires, et la réponse finale. Le trace grading analyse ces traces et attribue des scores selon vos critères. Vous passez ainsi d’un journal que l’on consulte après coup à un indicateur que l’on surveille en continu, et la question « comment se comporte l’agent en ce moment ? » cesse d’appeler une réponse au jugé.

Grading automatique avec des règles

Le niveau le plus simple consiste en des règles programmatiques qui notent chaque trace. Chacune part d’un score parfait de 1,0 et retranche selon ce qu’elle observe. Une exécution qui dépasse dix secondes coûte 0,2 point, non parce que le résultat est faux mais parce que l’utilisateur a probablement abandonné. Plus de quinze tours coûte 0,3 point : c’est la signature d’un agent qui tourne en rond entre deux tools. Chaque erreur de tool retire 0,1 point, les guardrails déclenchés sont signalés sans pénalité — ils ont fait leur travail —, et une réponse de moins de dix caractères coûte à elle seule un demi-point, parce qu’un agent muet est un agent inutile.

from dataclasses import dataclass

@dataclass
class GradeTrace:
    trace_id: str
    score: float  # 0.0 à 1.0
    details: dict
    alertes: list[str]

def noter_trace(trace_data: dict) -> GradeTrace:
    """Note une trace selon des règles métier."""
    score = 1.0
    alertes = []
    details = {}

    # Règle 1 : Latence
    duree = trace_data.get("duration_ms", 0)
    if duree > 10000:
        score -= 0.2
        alertes.append(f"Latence élevée : {duree}ms")
    details["latence_ms"] = duree

    # Règle 2 : Nombre de tours
    nb_tours = trace_data.get("num_turns", 0)
    if nb_tours > 15:
        score -= 0.3
        alertes.append(f"Trop de tours : {nb_tours}")
    details["nb_tours"] = nb_tours

    # Règle 3 : Erreurs de tools
    tool_errors = trace_data.get("tool_errors", 0)
    if tool_errors > 0:
        score -= 0.1 * tool_errors
        alertes.append(f"{tool_errors} erreur(s) de tool")
    details["erreurs_tools"] = tool_errors

    # Règle 4 : Guardrails déclenchés
    guardrails = trace_data.get("guardrails_triggered", 0)
    if guardrails > 0:
        alertes.append(f"{guardrails} guardrail(s) déclenché(s)")
    details["guardrails"] = guardrails

    # Règle 5 : Réponse vide
    reponse = trace_data.get("final_output", "")
    if not reponse or len(reponse) < 10:
        score -= 0.5
        alertes.append("Réponse vide ou trop courte")
    details["longueur_reponse"] = len(reponse)

    return GradeTrace(
        trace_id=trace_data["trace_id"],
        score=max(0.0, score),
        details=details,
        alertes=alertes,
    )

Le dictionnaire details accompagne systématiquement le score : sans lui, vous sauriez qu’une exécution vaut 0,5 sans savoir pourquoi. Ces seuils sont par ailleurs les vôtres et non des constantes universelles — un agent de recherche documentaire tolère quinze tours là où un assistant de caisse doit répondre en trois.

Grading par LLM

Les règles ne voient pas le contenu. Une réponse rapide, courte et polie qui répond à côté de la question obtiendra 1,0. Pour une évaluation plus fine, utilisez un modèle pour noter les traces sur des critères de fond : pertinence, exactitude et ton, chacun de 1 à 5, assortis des problèmes identifiés et de suggestions d’amélioration. Là encore, la sortie structurée via Pydantic rend les notes agrégeables au lieu de produire un commentaire libre.

from pydantic import BaseModel
from agents import Agent, Runner

class GradeLLM(BaseModel):
    score_pertinence: int  # 1-5
    score_exactitude: int  # 1-5
    score_ton: int  # 1-5
    problemes: list[str]
    suggestions: list[str]

agent_grader = Agent(
    name="Grader de traces",
    instructions="""Vous notez les exécutions d'un agent commercial.

    Analysez la trace complète (question, tools appelés, réponse) et notez :
    - Pertinence (1-5) : la réponse répond-elle à la question ?
    - Exactitude (1-5) : les informations sont-elles correctes ?
    - Ton (1-5) : le ton est-il professionnel et adapté ?

    Identifiez les problèmes et proposez des améliorations.""",
    model="gpt-5.6-sol",
    output_type=GradeLLM,
)

async def noter_trace_llm(trace: dict) -> GradeLLM:
    prompt = f"""Trace d'exécution à noter :

    Question utilisateur : {trace['user_message']}
    Tools appelés : {trace['tools_called']}
    Réponse finale : {trace['final_output']}
    Durée : {trace['duration_ms']}ms
    Nombre de tours : {trace['num_turns']}"""

    result = await Runner.run(agent_grader, prompt)
    return result.final_output

Le prompt transmet aussi les tools appelés et la durée : le juge peut ainsi reprocher à l’agent d’avoir répondu sans consulter le catalogue, ce qu’une lecture de la seule réponse ne permettrait pas.

Pipeline de grading en production

Reste à combiner les deux approches sans ruiner votre budget. Noter chaque trace par LLM sur un volume de production revient à multiplier votre facture d’inférence ; ne noter que par règles vous laisse aveugle au fond. Le pipeline ci-dessous applique donc un tri : toutes les traces passent par le grading programmatique, rapide et gratuit, et seules celles qui tombent sous 0,7 sont soumises au juge. Le score du juge, somme de trois notes sur 5, est ramené sur l’échelle 0-1 en divisant par 15 pour rester comparable au premier. En dessous de 0,5, la trace rejoint la liste des alertes critiques.

import asyncio
from collections import defaultdict

class PipelineGrading:
    def __init__(self):
        self.scores = defaultdict(list)
        self.alertes_critiques = []

    async def noter_batch(self, traces: list[dict]):
        """Note un batch de traces."""
        for trace in traces:
            # Grading programmatique (rapide)
            grade_regles = noter_trace(trace)

            # Grading LLM pour les cas douteux
            if grade_regles.score < 0.7:
                grade_llm = await noter_trace_llm(trace)
                score_final = (
                    grade_llm.score_pertinence +
                    grade_llm.score_exactitude +
                    grade_llm.score_ton
                ) / 15.0  # Normaliser sur 0-1
            else:
                score_final = grade_regles.score

            self.scores["global"].append(score_final)

            # Alertes
            if score_final < 0.5:
                self.alertes_critiques.append({
                    "trace_id": trace["trace_id"],
                    "score": score_final,
                    "alertes": grade_regles.alertes,
                })

    def rapport(self):
        """Génère un rapport de qualité."""
        scores = self.scores["global"]
        if not scores:
            return "Aucune trace évaluée."

        import statistics
        return {
            "nb_traces": len(scores),
            "score_moyen": statistics.mean(scores),
            "score_median": statistics.median(scores),
            "score_min": min(scores),
            "pct_critiques": len(self.alertes_critiques) / len(scores),
            "alertes_critiques": self.alertes_critiques[:10],
        }

# Utilisation
pipeline = PipelineGrading()
await pipeline.noter_batch(traces_du_jour)
print(pipeline.rapport())

Le rapport conserve le score minimum et un échantillon des dix pires traces : c’est par là que commence une investigation, jamais par la moyenne.

Tableau de bord de qualité

Un score isolé ne signifie rien ; c’est sa trajectoire qui informe. En écrivant le rapport quotidien dans un fichier JSONL, vous constituez une série temporelle que n’importe quel outil de visualisation sait lire, et vous pouvez comparer automatiquement le dernier point au précédent. Une baisse de plus de 0,1 déclenche une alerte de régression — et le rapprochement avec vos déploiements du jour donne en général le coupable en quelques minutes.

import json
from datetime import datetime

def enregistrer_metriques(rapport: dict, fichier: str = "metriques_agent.jsonl"):
    """Enregistre les métriques quotidiennes."""
    entree = {
        "date": datetime.now().isoformat(),
        **rapport,
    }
    with open(fichier, "a") as f:
        f.write(json.dumps(entree) + "\n")

def detecter_regression(metriques_recentes: list[dict], seuil: float = 0.1):
    """Détecte une baisse significative de qualité."""
    if len(metriques_recentes) < 2:
        return False
    dernier = metriques_recentes[-1]["score_moyen"]
    precedent = metriques_recentes[-2]["score_moyen"]
    baisse = precedent - dernier
    if baisse > seuil:
        print(f"REGRESSION DETECTEE : score passé de {precedent:.2f} à {dernier:.2f}")
        return True
    return False

Comparer deux jours consécutifs reste sensible au bruit : sur un faible volume, préférez une moyenne glissante avant de crier à la régression, sous peine de courir après un écart statistique.

Points clés à retenir

Le trace grading note automatiquement chaque exécution de votre agent
Les règles programmatiques (latence, erreurs, longueur) sont rapides et fiables
Le grading LLM évalue la pertinence, l’exactitude et le ton
Combinez les deux : règles rapides pour le tri, LLM pour les cas douteux
Suivez les scores dans le temps pour détecter les régressions
Les alertes critiques (score < 0.5) doivent déclencher une investigation