Trace grading : noter les exécutions
Trace grading : noter les exécutions
Le tracing enregistre ce que fait votre agent. 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.
Grading automatique avec des règles
Le niveau le plus simple : des règles programmatiques qui notent chaque trace :
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,
)Grading par LLM
Pour une évaluation plus fine, utilisez un modèle pour noter les traces :
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.4",
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_outputPipeline de grading en production
Mettez en place un pipeline qui note automatiquement les traces :
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())Tableau de bord de qualité
Suivez l’évolution de la qualité dans le temps :
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 FalsePoints 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