Évaluation des agents d’IA : ensembles parfaits, juges LLM et portes de régression CI
Évaluation en mode hors ligne et en ligne pour les agents grâce à AgentCore, LangFuse, Ragas et DeepEval, ainsi que des ensembles de données d’or, un système de contrôle CI, des tests A/B et un retour d’information basé sur la mémoire.
Sources: Agentic Architectures — Article 8: Evaluation and Continuous Improvement Pipelines
Tôt ou tard dans un projet d’agent, quelqu’un pose une question qui semble difficile mais l’est en réalité peu : est-ce que cela s’améliore vraiment ? Un prompt système est réécrit, un cycle de vérification est ajusté, un nouveau niveau de mémoire est ajouté, et l’agent semble mieux fonctionner lors des tests. Un meilleur rendu apparent n’est pas une mesure fiable. Sans mesures en environnement de production, les dégradations restent cachées jusqu’à ce qu’un utilisateur les découvre, et personne ne peut déterminer si les modifications apportées ont été utiles. L’optimisation par intuition convient comme point de départ, mais c’est une méthode insuffisante pour aller plus loin.
Les outils d’évaluation ont beaucoup évolué. AWS AgentCore Evaluations, LangFuse, Ragas et DeepEval fournissent chacun des éléments solides pour leur construction. Cependant, la tâche consistant à les assembler en un pipeline capable de détecter les problèmes avant même que les utilisateurs ne s’en aperçoivent reste à la charge de l’équipe. Les sections suivantes décrivent ce processus d’assemblage, en utilisant un agent de traitement de documents comme exemple concret, et relient le champ success_rate de la mémoire procédurale, issu d’un design de mémoire antérieur, à un véritable système de notation.
Ce qui est abordé
- Évaluation hors ligne versus en ligne. Ces deux méthodes nécessitent des infrastructures différentes, mesurent des éléments distincts et détectent des types d’échecs différents. Les équipes qui n’utilisent qu’une seule méthode restent surprises par les problèmes de l’autre.
- LLM en tant que juge à grande échelle. Une couche de notation systématique appliquée à chaque trace de production qui signale automatiquement les dégradations, au lieu d’une vérification ponctuelle.
- Comparaison des outils. AgentCore Evaluations, LangFuse, Ragas et DeepEval couvrent chacun un niveau différent, et du code fonctionnel indique quand utiliser l’un ou l’autre.
- Ensembles de données d’or. Origine des cas de test, méthode pour les sélectionner à partir des traces de production et façon de prévenir leur obsolétisation.
- Détection des dégradations dans le CI. Toute modification du framework qui diminue la qualité doit empêcher le déploiement.
- Test A/B des versions d’agent. Diriger une partie du trafic en direct vers une nouvelle configuration et comparer les résultats avant de valider les changements.
- Intégration des résultats dans la mémoire. Le champ
success_ratedispose enfin d’une logique de mesure réelle.
Évaluation hors ligne et en ligne
Ces deux modes ne sont pas interchangeables. Ils sont tous deux nécessaires, et voici comment ils se comparent :
+---------------------------+-----------------------------+-----------------------------+
| Dimension | Offline Evaluation | Online Evaluation |
+---------------------------+-----------------------------+-----------------------------+
| When it runs | Before deployment | Continuously in production |
+---------------------------+-----------------------------+-----------------------------+
| Input source | Curated golden dataset | Real user requests |
+---------------------------+-----------------------------+-----------------------------+
| Ground truth | Hand-labeled expected | Implicit (user feedback, |
| | outputs | downstream outcomes) |
+---------------------------+-----------------------------+-----------------------------+
| Speed | Can be slow, batch is fine | Must be fast, async |
+---------------------------+-----------------------------+-----------------------------+
| What it catches | Regressions vs known cases | Distribution shift, |
| | Logic errors, edge cases | real-world failure modes |
+---------------------------+-----------------------------+-----------------------------+
| Primary tools | Ragas, DeepEval, | LangFuse, AgentCore, |
| | AgentCore Evaluations | LLM-as-judge async |
+---------------------------+-----------------------------+-----------------------------+
| Output | Pass/fail per test case, | Metric trends, anomaly |
| | blocks deployment if fails | alerts, drift detection |
+---------------------------+-----------------------------+-----------------------------+
Résultats d’une source vers l’autre :
Golden Dataset
|
v
+------------------+ fails +------------------+
| Offline Eval |--------------->| Block Deploy |
| (pre-deploy) | | Alert team |
+------------------+ +------------------+
|
| passes
v
+------------------+
| Production |
| Deployment |
+------------------+
|
v
+------------------+ anomaly +------------------+
| Online Eval |--------------->| Alert team |
| (continuous) | | Rollback flag |
+------------------+ +------------------+
|
| good traces
v
+------------------+
| Golden Dataset |
| Update (curate) |
+------------------+
La boucle est intentionnelle. Les traces de production montrant des cas difficiles ou intéressants sont intégrées dans l’ensemble de données d’or, afin que les exécutions hors ligne ultérieures reflètent mieux l’utilisation réelle.
Ensembles de données d’or
Tout ce qui se passe hors ligne dépend de la qualité des cas de test. Un ensemble d’or de faible qualité engendre une confiance trompeuse : tous les tests réussissent, la version est déployée, mais il s’avère que les véritables modes de défaillance n’ont jamais été testés. Les cas proviennent de trois sources : des cas rédigés manuellement pour des cas limites connus, des cas sélectionnés parmi les traces de production où l’agent a eu du mal, et des cas représentant des parcours réussis importants qui doivent continuer de fonctionner.
# harness/evaluation/golden_dataset.py
import boto3
import json
import time
import uuid
from typing import List, Optional
from dataclasses import dataclass, field
@dataclass
class GoldenCase:
case_id: str
input: str # the task given to the agent
expected_output: Optional[str] # expected final answer (can be None for open-ended)
expected_behaviors: List[str] # things the agent should do, e.g. "calls security_review tool"
forbidden_behaviors: List[str] # things the agent must not do
tags: List[str] # "edge_case", "regression", "happy_path", etc.
difficulty: str # EASY | MEDIUM | HARD
source: str # "hand_crafted" | "production_trace" | "failure_case"
created_at: float
last_used_at: Optional[float]
times_failed: int = 0
class GoldenDatasetStore:
"""
Manages the golden dataset used for offline evaluation.
Supports curation from production traces and manual additions.
"""
def __init__(
self,
table_name: str = "agent-golden-dataset",
region: str = "us-east-1",
):
dynamodb = boto3.resource("dynamodb", region_name=region)
self.table = dynamodb.Table(table_name)
def add_case(self, case: GoldenCase):
self.table.put_item(Item={
"case_id": case.case_id,
"input": case.input,
"expected_output": case.expected_output,
"expected_behaviors": case.expected_behaviors,
"forbidden_behaviors": case.forbidden_behaviors,
"tags": case.tags,
"difficulty": case.difficulty,
"source": case.source,
"created_at": case.created_at,
"last_used_at": case.last_used_at,
"times_failed": case.times_failed,
})
def curate_from_production_trace(
self,
trace: dict,
agent_outcome: str,
human_rating: Optional[float] = None,
extractor_model=None,
) -> Optional[GoldenCase]:
"""
Promotes a production trace to the golden dataset.
Only call this for traces that represent interesting or important cases.
Human rating 0.0-1.0 is optional but strongly recommended.
"""
if not extractor_model:
return None
from langchain_core.messages import SystemMessage, HumanMessage
response = extractor_model.invoke([
SystemMessage(content="""
From this production trace, extract a golden test case.
Return JSON:
{
"input": "the original task",
"expected_behaviors": ["list of things a correct agent should do"],
"forbidden_behaviors": ["list of things the agent must not do"],
"difficulty": "EASY | MEDIUM | HARD",
"worth_adding": true | false
}
Set worth_adding to false if the case is too simple or redundant.
"""),
HumanMessage(content=json.dumps(trace)[:4000])
])
try:
extracted = json.loads(response.content)
except json.JSONDecodeError:
return None
if not extracted.get("worth_adding", False):
return None
return GoldenCase(
case_id=str(uuid.uuid4()),
input=extracted["input"],
expected_output=None, # open-ended, behavior-based eval
expected_behaviors=extracted.get("expected_behaviors", []),
forbidden_behaviors=extracted.get("forbidden_behaviors", []),
tags=["production_trace", agent_outcome.lower()],
difficulty=extracted.get("difficulty", "MEDIUM"),
source="production_trace",
created_at=time.time(),
last_used_at=None,
times_failed=0,
)
def get_cases(
self,
tags: Optional[List[str]] = None,
difficulty: Optional[str] = None,
limit: int = 100,
) -> List[GoldenCase]:
"""Retrieves test cases, optionally filtered by tag or difficulty."""
response = self.table.scan(Limit=limit)
items = response.get("Items", [])
if tags:
items = [i for i in items if any(t in i.get("tags", []) for t in tags)]
if difficulty:
items = [i for i in items if i.get("difficulty") == difficulty]
return [GoldenCase(**item) for item in items]
def record_failure(self, case_id: str):
"""Tracks which cases are frequently failing for prioritization."""
self.table.update_item(
Key={"case_id": case_id},
UpdateExpression="SET times_failed = if_not_exists(times_failed, :z) + :inc",
ExpressionAttributeValues={":z": 0, ":inc": 1},
)
LLM en tant que juge
Les réponses générées par les agents ne disposent souvent pas d’une seule réponse correcte. Un agent d’analyse de contrats peut produire trois résumés valables d’un même contrat, et un agent d’examen de code peut signaler les mêmes vulnérabilités dans des ordres et des formulations différents. L’évaluation par correspondance exacte échoue dans ce cas. La solution pratique consiste à utiliser un modèle distinct qui évalue les réponses selon des critères prédéfinis et fournit une note structurée. Le juge a besoin de critères explicites et sans équivoque, car des critères vagues entraînent des notes incohérentes, inutiles pour détecter les dégradations.
# harness/evaluation/llm_judge.py
import boto3
import json
from typing import List, Optional
from dataclasses import dataclass
from langchain_aws import ChatBedrock
from langchain_core.messages import SystemMessage, HumanMessage
@dataclass
class JudgementResult:
case_id: str
score: float # 0.0 to 1.0
passed: bool
dimension_scores: dict # per-rubric scores
reasoning: str
flags: List[str] # "HALLUCINATION", "INCOMPLETE", "FORBIDDEN_BEHAVIOR"
JUDGE_SYSTEM_PROMPT = """
You are an evaluation judge for an AI agent system. You assess agent outputs
against explicit rubrics. You are strict, consistent, and evidence-based.
You must:
- Score each dimension independently
- Cite specific evidence from the agent output for each score
- Flag hallucinations, incomplete responses, and forbidden behaviors explicitly
- Never give partial credit for behaviors that were required but missing
Return only valid JSON matching the schema provided.
"""
class LLMJudge:
"""
Evaluates agent outputs against a structured rubric.
Designed to run both in offline eval (against golden dataset)
and online (against production traces, asynchronously).
"""
PASS_THRESHOLD = 0.75
def __init__(self, region: str = "us-east-1"):
bedrock = boto3.client("bedrock-runtime", region_name=region)
# Judge uses extended thinking for careful, consistent evaluation
self.judge_model = ChatBedrock(
client=bedrock,
model_id="anthropic.claude-3-7-sonnet-20250219-v1:0",
model_kwargs={
"temperature": 0,
"max_tokens": 4000,
"thinking": {"type": "enabled", "budget_tokens": 3000},
}
)
def evaluate(
self,
case_id: str,
task_input: str,
agent_output: str,
agent_trace: List[dict], # tool calls, intermediate steps
expected_behaviors: List[str],
forbidden_behaviors: List[str],
expected_output: Optional[str] = None,
) -> JudgementResult:
# Build tool call summary from trace
tool_calls_summary = "\n".join([
f"- {step.get('tool')}: {str(step.get('input', ''))[:200]}"
for step in agent_trace
if step.get("type") == "tool_call"
])
rubric_sections = []
for i, behavior in enumerate(expected_behaviors, 1):
rubric_sections.append(f" B{i}: {behavior}")
for i, behavior in enumerate(forbidden_behaviors, 1):
rubric_sections.append(f" F{i}: MUST NOT {behavior}")
evaluation_prompt = f"""
Task given to agent:
{task_input}
Agent final output:
{agent_output}
Tools called during the run:
{tool_calls_summary or "none recorded"}
{f"Expected output (for reference, not exact match):{chr(10)}{expected_output}" if expected_output else ""}
Rubric:
{chr(10).join(rubric_sections)}
Evaluate and return JSON:
{{
"dimension_scores": {{
"task_completion": 0.0,
"behavior_compliance": 0.0,
"output_quality": 0.0,
"no_forbidden_behaviors": 0.0
}},
"flags": [],
"reasoning": "brief explanation of scores",
"overall_score": 0.0
}}
Score each dimension 0.0 to 1.0.
Flags can include: HALLUCINATION, INCOMPLETE, FORBIDDEN_BEHAVIOR, WRONG_TOOLS, CORRECT.
"""
response = self.judge_model.invoke([
SystemMessage(content=JUDGE_SYSTEM_PROMPT),
HumanMessage(content=evaluation_prompt)
])
try:
result = json.loads(response.content)
except json.JSONDecodeError:
return JudgementResult(
case_id=case_id,
score=0.0,
passed=False,
dimension_scores={},
reasoning="Judge returned unparseable output",
flags=["JUDGE_ERROR"],
)
overall = result.get("overall_score", 0.0)
return JudgementResult(
case_id=case_id,
score=overall,
passed=overall >= self.PASS_THRESHOLD,
dimension_scores=result.get("dimension_scores", {}),
reasoning=result.get("reasoning", ""),
flags=result.get("flags", []),
)
Évaluations AgentCore
AWS AgentCore Evaluations propose une infrastructure gérée pour les évaluations des agents sur Bedrock. Elle orchestre les exécutions d’évaluation, stocke les résultats et agrège les métriques. Elle ne crée pas de grilles d’évaluation, qui restent la responsabilité de l’équipe.
# harness/evaluation/agentcore_eval.py
import boto3
import json
import time
from typing import List, Optional
class AgentCoreEvaluator:
"""
Wraps AWS AgentCore Evaluations for structured offline eval runs.
Best suited for systematic evaluation against a fixed golden dataset.
"""
def __init__(self, region: str = "us-east-1"):
self.client = boto3.client("bedrock-agent", region_name=region)
self.runtime = boto3.client("bedrock-agent-runtime", region_name=region)
self.region = region
def create_evaluation_job(
self,
job_name: str,
agent_id: str,
agent_alias_id: str,
golden_cases: List[dict],
output_s3_uri: str,
role_arn: str,
) -> str:
"""
Submits an evaluation job against AgentCore.
Returns the evaluation job ID.
"""
response = self.client.create_agent_alias(
agentId=agent_id,
agentAliasName=f"eval-{job_name}",
) if not agent_alias_id else {"agentAlias": {"agentAliasId": agent_alias_id}}
eval_config = {
"agentEvaluatorConfiguration": {
"agentId": agent_id,
"agentAliasId": agent_alias_id,
}
}
dataset = {
"name": job_name,
"datasetType": "Inline",
"dataset": {
"items": [
{
"conversationInput": {
"turns": [{"role": "user", "content": case["input"]}]
},
"referenceResponses": [case.get("expected_output", "")]
if case.get("expected_output") else [],
}
for case in golden_cases
]
}
}
job_response = self.client.create_evaluation_job(
jobName=job_name,
evaluationConfig=eval_config,
inferenceConfig={"models": [{"bedrockModel": {"modelIdentifier": "anthropic.claude-3-5-sonnet-20241022-v2:0"}}]},
outputDataConfig={"s3Uri": output_s3_uri},
roleArn=role_arn,
)
return job_response["jobArn"]
def wait_for_completion(self, job_arn: str, poll_interval: int = 30) -> dict:
"""Polls until evaluation completes and returns results summary."""
while True:
response = self.client.get_evaluation_job(jobIdentifier=job_arn)
status = response.get("status")
if status == "Completed":
return response
elif status in ("Failed", "Stopped"):
raise RuntimeError(f"Evaluation job {job_arn} ended with status: {status}")
time.sleep(poll_interval)
LangFuse pour l’évaluation en ligne
LangFuse gère la partie en ligne. Chaque trace de production est évaluée en arrière-plan, les scores sont regroupés en lignes de tendance et des baisses anormales déclenchent des alertes. L’évaluation doit rester asynchrone, sinon elle retarde les réponses de l’agent.
# harness/evaluation/langfuse_scorer.py
import asyncio
from typing import Optional
from langfuse import Langfuse
from harness.evaluation.llm_judge import LLMJudge, JudgementResult
class LangFuseOnlineScorer:
"""
Scores production traces asynchronously using LangFuse.
The scoring happens after the agent run completes, never in the critical path.
"""
def __init__(
self,
public_key: str,
secret_key: str,
host: str = "https://cloud.langfuse.com",
region: str = "us-east-1",
):
self.langfuse = Langfuse(
public_key=public_key,
secret_key=secret_key,
host=host,
)
self.judge = LLMJudge(region=region)
def score_trace_async(
self,
trace_id: str,
task_input: str,
agent_output: str,
agent_trace: list,
expected_behaviors: list,
forbidden_behaviors: list,
):
"""
Submits scoring as a background task.
Call this after the agent run completes, fire-and-forget style.
"""
asyncio.create_task(
self._score_and_record(
trace_id, task_input, agent_output,
agent_trace, expected_behaviors, forbidden_behaviors
)
)
async def _score_and_record(
self,
trace_id: str,
task_input: str,
agent_output: str,
agent_trace: list,
expected_behaviors: list,
forbidden_behaviors: list,
):
try:
result = self.judge.evaluate(
case_id=trace_id,
task_input=task_input,
agent_output=agent_output,
agent_trace=agent_trace,
expected_behaviors=expected_behaviors,
forbidden_behaviors=forbidden_behaviors,
)
# Write overall score to LangFuse
self.langfuse.score(
trace_id=trace_id,
name="overall_quality",
value=result.score,
comment=result.reasoning,
)
# Write per-dimension scores
for dimension, score in result.dimension_scores.items():
self.langfuse.score(
trace_id=trace_id,
name=dimension,
value=score,
)
# Write flags as categorical scores
for flag in result.flags:
self.langfuse.score(
trace_id=trace_id,
name=f"flag_{flag.lower()}",
value=1.0,
)
except Exception as e:
# Scoring failure must never surface to the user
print(f"Online scoring failed for trace {trace_id}: {e}")
def get_metric_trends(
self,
metric_name: str = "overall_quality",
days: int = 7,
) -> dict:
"""
Retrieves score trend data for dashboard or alerting.
"""
scores = self.langfuse.get_scores(name=metric_name, limit=500)
if not scores:
return {"mean": None, "trend": "insufficient_data"}
values = [s.value for s in scores if s.value is not None]
if len(values) < 10:
return {"mean": sum(values) / len(values), "trend": "insufficient_data"}
recent = values[:len(values)//2]
older = values[len(values)//2:]
recent_mean = sum(recent) / len(recent)
older_mean = sum(older) / len(older)
trend = "stable"
if recent_mean < older_mean - 0.05:
trend = "degrading"
elif recent_mean > older_mean + 0.05:
trend = "improving"
return {
"mean": recent_mean,
"trend": trend,
"sample_size": len(values),
}
Ragas et DeepEval
Un juge général évalue la qualité globale, tandis que Ragas et DeepEval prennent en compte des métriques spécialisées que des critères généraux ont du mal à mesurer.
Ragas vise l’évaluation des systèmes RAG. Pour un agent qui récupère des informations depuis les bases de connaissances Bedrock, il fournit directement des indicateurs de fidélité, de pertinence des réponses et de précision du contexte :
# harness/evaluation/ragas_eval.py
from ragas import evaluate
from ragas.metrics import (
faithfulness,
answer_relevancy,
context_precision,
context_recall,
)
from datasets import Dataset
from typing import List
def evaluate_rag_quality(
questions: List[str],
answers: List[str],
contexts: List[List[str]],
ground_truths: Optional[List[str]] = None,
) -> dict:
"""
Evaluates RAG quality using Ragas metrics.
Call this when the agent's output depends heavily on retrieved context,
e.g. agents using Bedrock Knowledge Bases as semantic memory.
"""
data = {
"question": questions,
"answer": answers,
"contexts": contexts,
}
if ground_truths:
data["ground_truth"] = ground_truths
dataset = Dataset.from_dict(data)
metrics = [faithfulness, answer_relevancy, context_precision]
if ground_truths:
metrics.append(context_recall)
results = evaluate(dataset, metrics=metrics)
return results.to_pandas().mean().to_dict()
DeepEval couvre un éventail plus large du comportement des agents, en évaluant des éléments tels que les hallucinations, le degré de justesse d’une réponse et le fait que les outils appropriés aient été appelés avec les arguments corrects :
# harness/evaluation/deepeval_eval.py
from deepeval import evaluate as deepeval_evaluate
from deepeval.metrics import (
HallucinationMetric,
AnswerRelevancyMetric,
ToolCorrectnessMetric,
)
from deepeval.test_case import LLMTestCase, LLMTestCaseParams
from typing import List
def evaluate_agent_correctness(
inputs: List[str],
actual_outputs: List[str],
expected_tools: List[List[str]],
actual_tools: List[List[str]],
contexts: Optional[List[List[str]]] = None,
) -> List[dict]:
"""
Evaluates agent outputs for hallucination, relevancy, and tool correctness.
expected_tools: the tools the agent should have called
actual_tools: the tools the agent actually called
"""
metrics = [
HallucinationMetric(threshold=0.5),
AnswerRelevancyMetric(threshold=0.7),
ToolCorrectnessMetric(threshold=0.8),
]
test_cases = []
for i, (inp, out) in enumerate(zip(inputs, actual_outputs)):
test_case = LLMTestCase(
input=inp,
actual_output=out,
expected_tools=expected_tools[i] if expected_tools else [],
tools_called=actual_tools[i] if actual_tools else [],
context=contexts[i] if contexts else [],
)
test_cases.append(test_case)
results = deepeval_evaluate(test_cases, metrics)
return [
{
"input": tc.input,
"passed": all(m.is_successful() for m in metrics),
"scores": {m.__class__.__name__: m.score for m in metrics},
}
for tc in test_cases
]
Un véritable point de contrôle qualité
Un pipeline qui ne bloque pas les déploiements défectueux n’est qu’un outil de surveillance. Le détecteur de régressions relie les résultats hors ligne au pipeline de déploiement :
# harness/evaluation/regression_detector.py
import json
import boto3
from typing import List, Optional
from dataclasses import dataclass
@dataclass
class EvalRunResult:
run_id: str
agent_version: str
overall_pass_rate: float
dimension_scores: dict
failed_cases: List[str]
timestamp: float
class RegressionDetector:
"""
Compares current eval results against baseline and flags regressions.
Designed to run in CI before deployment.
Returns a clear pass/fail signal that can block a deploy.
"""
# How much worse the new version can be before blocking
REGRESSION_THRESHOLDS = {
"overall_pass_rate": 0.05, # 5% drop blocks deploy
"task_completion": 0.05,
"behavior_compliance": 0.03, # stricter on behavior
"no_forbidden_behaviors": 0.01,# very strict on safety
}
def __init__(
self,
table_name: str = "agent-eval-history",
region: str = "us-east-1",
):
dynamodb = boto3.resource("dynamodb", region_name=region)
self.table = dynamodb.Table(table_name)
def store_eval_result(self, result: EvalRunResult):
self.table.put_item(Item={
"run_id": result.run_id,
"agent_version": result.agent_version,
"overall_pass_rate": str(result.overall_pass_rate),
"dimension_scores": {k: str(v) for k, v in result.dimension_scores.items()},
"failed_cases": result.failed_cases,
"timestamp": int(result.timestamp),
})
def get_baseline(self, agent_version: str) -> Optional[EvalRunResult]:
"""
Retrieves the most recent eval result for the current production version.
This is the baseline the new version is compared against.
"""
response = self.table.query(
KeyConditionExpression="agent_version = :v",
ExpressionAttributeValues={":v": agent_version},
ScanIndexForward=False,
Limit=1,
)
items = response.get("Items", [])
if not items:
return None
item = items[0]
return EvalRunResult(
run_id=item["run_id"],
agent_version=item["agent_version"],
overall_pass_rate=float(item["overall_pass_rate"]),
dimension_scores={k: float(v) for k, v in item.get("dimension_scores", {}).items()},
failed_cases=item.get("failed_cases", []),
timestamp=float(item["timestamp"]),
)
def check_for_regression(
self,
current_result: EvalRunResult,
baseline: EvalRunResult,
) -> dict:
"""
Compares current eval results against baseline.
Returns a report with pass/fail verdict and specific regressions found.
"""
regressions = []
warnings = []
# Check overall pass rate
delta = baseline.overall_pass_rate - current_result.overall_pass_rate
if delta > self.REGRESSION_THRESHOLDS["overall_pass_rate"]:
regressions.append(
f"Overall pass rate dropped {delta:.1%}: "
f"{baseline.overall_pass_rate:.1%} -> {current_result.overall_pass_rate:.1%}"
)
elif delta > 0:
warnings.append(
f"Overall pass rate slightly lower: "
f"{baseline.overall_pass_rate:.1%} -> {current_result.overall_pass_rate:.1%}"
)
# Check per-dimension scores
for dimension, threshold in self.REGRESSION_THRESHOLDS.items():
if dimension == "overall_pass_rate":
continue
baseline_score = baseline.dimension_scores.get(dimension, 1.0)
current_score = current_result.dimension_scores.get(dimension, 1.0)
delta = baseline_score - current_score
if delta > threshold:
regressions.append(
f"{dimension} dropped {delta:.1%}: "
f"{baseline_score:.1%} -> {current_score:.1%}"
)
# Check for new failing cases that were previously passing
new_failures = set(current_result.failed_cases) - set(baseline.failed_cases)
if new_failures:
regressions.append(
f"{len(new_failures)} cases that previously passed are now failing: "
f"{', '.join(list(new_failures)[:5])}"
)
return {
"verdict": "BLOCK" if regressions else ("WARN" if warnings else "PASS"),
"regressions": regressions,
"warnings": warnings,
"current_pass_rate": current_result.overall_pass_rate,
"baseline_pass_rate": baseline.overall_pass_rate,
}
L’hook CI est un petit script qui s’arrête avec un code différent de zéro lorsqu’une régression apparaît, ce qui suffit à interrompre la plupart des pipelines :
# scripts/eval_gate.py
import sys
import json
from harness.evaluation.regression_detector import RegressionDetector, EvalRunResult
def run_eval_gate(current_result_path: str, baseline_version: str):
with open(current_result_path) as f:
data = json.load(f)
current = EvalRunResult(**data)
detector = RegressionDetector()
baseline = detector.get_baseline(baseline_version)
if not baseline:
print("No baseline found. Storing current result as new baseline.")
detector.store_eval_result(current)
sys.exit(0)
report = detector.check_for_regression(current, baseline)
print(json.dumps(report, indent=2))
if report["verdict"] == "BLOCK":
print("DEPLOY BLOCKED: regressions detected.")
sys.exit(1)
elif report["verdict"] == "WARN":
print("DEPLOY ALLOWED WITH WARNINGS.")
sys.exit(0)
else:
detector.store_eval_result(current)
print("DEPLOY APPROVED.")
sys.exit(0)
if __name__ == "__main__":
run_eval_gate(sys.argv[1], sys.argv[2])
Tests A/B des versions
Un test A/B met en parallèle deux configurations d’agent sur du trafic réel et compare leurs performances avant que la nouvelle version ne soit pleinement adoptée :
# harness/evaluation/ab_test.py
import boto3
import random
import time
from typing import Optional
from dataclasses import dataclass
@dataclass
class ABTestConfig:
experiment_id: str
control_version: str # current production harness
treatment_version: str # new harness being tested
traffic_split: float # 0.0 to 1.0, fraction going to treatment
start_time: float
end_time: Optional[float]
min_samples: int = 100 # minimum samples before declaring a winner
class ABTestRouter:
"""
Routes production requests to control or treatment agent versions.
Records which version handled each request for later comparison.
"""
def __init__(
self,
table_name: str = "agent-ab-test-assignments",
region: str = "us-east-1",
):
dynamodb = boto3.resource("dynamodb", region_name=region)
self.table = dynamodb.Table(table_name)
def get_version_for_request(
self,
request_id: str,
config: ABTestConfig,
) -> str:
"""
Assigns a version to this request and records the assignment.
Deterministic for the same request_id (consistent user experience).
"""
# Use request_id hash for deterministic assignment
import hashlib
hash_val = int(hashlib.md5(request_id.encode()).hexdigest(), 16)
use_treatment = (hash_val % 100) < (config.traffic_split * 100)
version = config.treatment_version if use_treatment else config.control_version
self.table.put_item(Item={
"request_id": request_id,
"experiment_id": config.experiment_id,
"assigned_version": version,
"timestamp": int(time.time()),
})
return version
def get_experiment_results(
self,
experiment_id: str,
eval_history_table: str = "agent-eval-history",
region: str = "us-east-1",
) -> dict:
"""
Aggregates eval scores by version for the experiment period.
Call this to determine which version performed better.
"""
response = self.table.query(
KeyConditionExpression="experiment_id = :eid",
ExpressionAttributeValues={":eid": experiment_id},
)
assignments = response.get("Items", [])
control_requests = [a["request_id"] for a in assignments if "control" in a["assigned_version"]]
treatment_requests = [a["request_id"] for a in assignments if "treatment" in a["assigned_version"]]
return {
"experiment_id": experiment_id,
"control_sample_size": len(control_requests),
"treatment_sample_size": len(treatment_requests),
"sufficient_data": min(len(control_requests), len(treatment_requests)) >= 100,
}
Fermer le cercle avec la mémoire
Le champ de mémoire procédurale success_rate était auparavant mis à jour après chaque exécution, mais cette mesure était rudimentaire : l’exécution se terminait soit oui, soit non. Avec un pipeline d’évaluation adéquat en place, la connexion devient directe, et les scores des juges peuvent être intégrés dans le stockage mémoire :
# harness/memory/procedural_eval_bridge.py
from harness.evaluation.llm_judge import LLMJudge
from harness.memory.procedural import ProceduralMemoryStore
class ProceduralMemoryEvalBridge:
"""
Updates procedural memory success rates using LLM judge scores
rather than naive completion flags.
"""
def __init__(self, region: str = "us-east-1"):
self.judge = LLMJudge(region=region)
self.procedural = ProceduralMemoryStore(region=region)
self.QUALITY_THRESHOLD = 0.75
def evaluate_and_update(
self,
procedure_id: str,
task_input: str,
agent_output: str,
agent_trace: list,
expected_behaviors: list,
forbidden_behaviors: list,
):
"""
Runs judge evaluation on a procedure-guided run and updates
the success rate with a quality-weighted result.
"""
judgement = self.judge.evaluate(
case_id=procedure_id,
task_input=task_input,
agent_output=agent_output,
agent_trace=agent_trace,
expected_behaviors=expected_behaviors,
forbidden_behaviors=forbidden_behaviors,
)
# Success is now quality-gated, not just completion-gated
succeeded = judgement.score >= self.QUALITY_THRESHOLD
self.procedural.record_application_result(
procedure_id=procedure_id,
succeeded=succeeded,
)
# If a validated procedure's success rate drops, flag for review
if not succeeded and "HALLUCINATION" in judgement.flags:
self.procedural.table.update_item(
Key={"procedure_id": procedure_id},
UpdateExpression="SET needs_review = :v, review_reason = :r",
ExpressionAttributeValues={
":v": True,
":r": f"Hallucination detected in judge eval. Score: {judgement.score:.2f}",
}
)
Ce qui a tendance à mal se passer
L’infrastructure d’évaluation n’est pas glamour et n’est jamais mise en œuvre. Problèmes typiques :
Les juges ne sont pas neutres. Un modèle de juge présente des biais. Il a tendance à privilégier les réponses plus longues et plus assurées, écrites dans un style proche de ses données d’entraînement ; ainsi, une réponse courte et correcte peut obtenir un score inférieur à celui d’une réponse longue mais légèrement erronée. Comparez régulièrement ces juges avec les évaluations humaines et modifiez les critères d’évaluation lorsque des biais systématiques apparaissent.
Les ensembles de données d’ référence deviennent obsolètes. La répartition des tâches en production évolue. Un ensemble construit il y a six mois peut ne plus refléter ce à quoi l’agent est confronté aujourd’hui. Organisez régulièrement des sessions de mise à jour, en particulier après de grands changements dans le domaine couvert par l’agent.
Les taux de réussite cachent des détails importants. Un système peut maintenir un taux de réussite élevé tout en se détériorant sur les cas les plus cruciaux. Suivez les échecs séparément par catégorie et par niveau de difficulté. Une baisse des performances sur les cas difficiles est plus inquiétante qu’une baisse sur les cas faciles, même si le taux global reste stable.
Les tests A/B nécessitent plus de trafic que prévu. Pour démontrer une signification statistique avec une amélioration de qualité de 3 %, il faut au moins des centaines d’échantillons. Avec un faible trafic, les tests peuvent durer des semaines. Déterminez la durée à l’avance et ne proclamez pas de gagnant prématurément.
Le processus d’évaluation peut contenir des erreurs. Les prompts utilisés pour l’évaluation peuvent être légèrement erronés, les cas de référence peuvent définir un comportement attendu incorrect, et le code de notation peut présenter des erreurs de type « off-by-one ». Traitez-le avec autant de sérieux que le code de l’agent : testez-le, examinez-le attentivement, et surveillez les distributions de notes inhabituelles qui indiquent une erreur de notation plutôt qu’un véritable changement de qualité.
Tout le système
L’architecture complète, allant du jeu de données d’or en passant par l’évaluation hors ligne et en ligne, jusqu’aux points de contrôle CI et au feedback mémoire :
+----------------------------+
| Golden Dataset |
| DynamoDB |
| hand-crafted + curated |
+-------------+--------------+
|
+------------------------+------------------------+
| |
v v
+-----------------------+ +------------------------+
| Offline Eval (CI) | | Online Eval (Prod) |
| Ragas + DeepEval | | LangFuse + Judge |
| AgentCore Eval | | Async scoring |
| LLM Judge | | Trend monitoring |
+-----------+-----------+ +----------+-------------+
| |
v v
+-----------------------+ +------------------------+
| Regression Detector | | Score History DB |
| PASS / WARN / BLOCK | | DynamoDB |
+-----------+-----------+ +----------+-------------+
| |
v v
+-----------+-----------+ +----------+-------------+
| Deploy Pipeline | | Procedural Memory |
| blocked on BLOCK | | success_rate update |
+-----------------------+ +------------------------+
|
| production traces (good cases)
v
+-----------------------+
| Golden Dataset |
| Curation |
| (loop closes) |
+-----------------------+
Et la technologie derrière chaque composant :
+-----------------------------+---------------------+------------------------------+
| Component | Technology | Role |
+-----------------------------+---------------------+------------------------------+
| Golden Dataset | DynamoDB | Test cases, curated from |
| | | production and hand-crafted |
+-----------------------------+---------------------+------------------------------+
| LLM-as-Judge | Claude 3.7 Sonnet | Structured rubric scoring, |
| | (extended thinking) | offline and online eval |
+-----------------------------+---------------------+------------------------------+
| Offline Eval Orchestration | AWS AgentCore | Managed batch eval runs, |
| | Evaluations | result storage |
+-----------------------------+---------------------+------------------------------+
| Online Eval and Tracing | LangFuse | Production trace scoring, |
| | | metric trends, dashboards |
+-----------------------------+---------------------+------------------------------+
| RAG Quality Metrics | Ragas | Faithfulness, relevancy, |
| | | context precision/recall |
+-----------------------------+---------------------+------------------------------+
| Agent-Specific Metrics | DeepEval | Hallucination, tool |
| | | correctness, answer quality |
+-----------------------------+---------------------+------------------------------+
| Regression Detection | Custom (DynamoDB) | Baseline comparison, |
| | | CI gate, deploy blocking |
+-----------------------------+---------------------+------------------------------+
| A/B Test Routing | Custom (DynamoDB) | Traffic split, deterministic |
| | | assignment, result rollup |
+-----------------------------+---------------------+------------------------------+
| Procedural Memory Bridge | Custom | Connects eval scores to |
| | | Article 7 success_rate field |
+-----------------------------+---------------------+------------------------------+
| PII Scrubbing in Eval | AWS Comprehend | Before golden dataset |
| | | storage from prod traces |
+-----------------------------+---------------------+------------------------------+
| Eval Result History | DynamoDB | Baseline storage, trend |
| | | analysis, A/B results |
+-----------------------------+---------------------+------------------------------+
| Local Dev Alternative | Ollama + ChromaDB | Full eval stack runnable |
| | Docker Compose | without AWS or paid APIs |
+-----------------------------+---------------------+------------------------------+
| Infrastructure as Code | Terraform | DynamoDB tables, S3 buckets, |
| | | IAM roles for eval jobs |
+-----------------------------+---------------------+------------------------------+
Lectures complémentaires
- Orchestration multi-agent sur Bedrock et LangGraph : quatre modèles — Modèles superviseur-travailleur, pipeline, diffusion et débat avec LangGraph sur AWS Bedrock, ainsi que mécanismes de transfert, de contrôle budgétaire, de traçage et de gestion des pannes en production.
- Mémoire agentes sémantique dans LangGraph avec Qdrant et rappel entre threads — Fournir à un agent LangGraph une mémoire à long terme grâce à Qdrant : intégrer des faits utilisateur, les rappeler selon les identifiants de threads, et ajouter des filtres de charge utile pour l’isolation multi-locataires.