Evaluación de agentes de IA: conjuntos dorados, jueces de LLM y puertas de regresión de CI
Evaluación offline y en línea para agentes con AgentCore, LangFuse, Ragas y DeepEval, además de conjuntos de datos de referencia, control de integración continua, pruebas A/B y retroalimentación basada en la memoria.
Fuentes: Agentic Architectures — Article 8: Evaluation and Continuous Improvement Pipelines
Tarde o temprano en un proyecto con agentes, alguien plantea una pregunta aparentemente difícil: ¿realmente está mejorando? Se reescribe un prompt del sistema, se ajusta un bucle de verificación, se añade un nuevo nivel de memoria, y el agente parece funcionar mejor en las pruebas. Que parezca mejorar no constituye una medida real. Sin mediciones en entornos de producción, las regresiones permanecen ocultas hasta que un usuario las descubre, y nadie puede determinar si los cambios realizados fueron útiles. Optimizar por instinto está bien como punto de partida, pero es una forma deficiente para continuar con el proceso.
Las herramientas de evaluación han evolucionado enormemente. AWS AgentCore Evaluations, LangFuse, Ragas y DeepEval ofrecen cada uno componentes sólidos para su desarrollo. Sin embargo, la tarea de integrarlos en un proceso que detecte problemas antes de que los usuarios lo hagan sigue correspondiendo al equipo. Las secciones siguientes explican dicho proceso, utilizando como ejemplo un agente de procesamiento de documentos, y vinculan el campo success_rate de la memoria procedural de un diseño anterior con un sistema real de puntuación.
Qué se aborda
- Evaluación offline versus online. Requieren infraestructuras diferentes, miden aspectos distintos y detectan tipos de fallos variados. Los equipos que solo utilizan uno u otro siguen sorprendiéndose con los problemas del otro.
- LLM como juez a gran escala. Una capa de puntuación sistemática sobre cada rastro de producción que identifica automáticamente las regresiones, en lugar de una verificación puntual.
- Comparación de herramientas. AgentCore Evaluations, LangFuse, Ragas y DeepEval abarcan cada uno una capa diferente, y el código de ejemplo muestra cuándo utilizar cada una.
- Conjuntos de datos de referencia. De dónde provienen los casos de prueba, cómo seleccionarlos de los rastros de producción y cómo evitar que se vuelvan obsoletos.
- Detección de regresiones en CI. Cualquier cambio en el entorno que reduzca la calidad debe impedir el despliegue.
- Pruebas A/B de versiones del agente. Dirigir una parte del tráfico en tiempo real a una nueva configuración y comparar resultados antes de aplicar los cambios.
- Almacenamiento de resultados en memoria. El campo
success_ratefinalmente cuenta con una lógica de medición real.
Evaluación offline y online
Los dos modos no son intercambiables. Se necesitan ambos, y aquí está cómo se comparan:
+---------------------------+-----------------------------+-----------------------------+
| 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 |
+---------------------------+-----------------------------+-----------------------------+
Resultados de un feed al otro:
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) |
+------------------+
El bucle es intencional. Los registros de producción que muestran casos difíciles o interesantes se incluyen en el conjunto de datos dorado, de modo que las ejecuciones posteriores sin conexión reflejen mejor el uso real.
Conjuntos de datos dorados
Todo lo que se hace sin conexión depende de la calidad de los casos de prueba. Un conjunto dorado deficiente genera una falsa sensación de confianza: todos los tests pasan, se lanza la versión y resulta que los verdaderos modos de fallo nunca fueron probados. Los casos provienen de tres fuentes: aquellos escritos a mano para casos límite conocidos, otros seleccionados de los registros de producción donde el agente tuvo dificultades, y aquellos que capturan rutas exitosas importantes que deben seguir funcionando.
# 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 como juez
A menudo, los resultados generados por un agente no tienen una única respuesta correcta. Un agente de análisis de contratos puede producir tres resúmenes válidos de un mismo contrato, y un agente de revisión de código puede señalar las mismas vulnerabilidades en diferentes ordenes y con distintas formulaciones. La evaluación por coincidencia exacta falla en estos casos. La solución práctica es utilizar un modelo separado que califique los resultados según una rúbrica y devuelva una puntuación estructurada. El juez necesita criterios explícitos y sin ambigüedades, ya que las rúbricas vagas generan puntuaciones inconsistentes que resultan inútiles para detectar regresiones.
# 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", []),
)
Evaluaciones de AgentCore
AWS AgentCore Evaluations proporciona una infraestructura gestionada para la evaluación de agentes en Bedrock. Orquesta las ejecuciones de evaluación, almacena los resultados y agrega métricas. No crea rúbricas, las cuales siguen siendo responsabilidad del equipo.
# 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 para la puntuación en línea
LangFuse se encarga de la parte en línea. Cada rastro de producción es puntuado en segundo plano; las puntuaciones se agrupan en líneas de tendencia y las caídas inusuales generan alertas. La puntuación debe mantenerse asíncrona, ya que de lo contrario se genera retraso en las respuestas del agente.
# 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 y DeepEval
Un juez general evalúa la calidad en su conjunto, mientras que Ragas y DeepEval se encargan de métricas especializadas que resultan difíciles de medir con una rúbrica general.
Ragas está orientado a la evaluación de RAG. Para un agente que recupera información de las bases de conocimiento Bedrock, ofrece de forma predeterminada criterios como la fidelidad, la relevancia de las respuestas y la precisión del contexto:
# 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 analiza de manera más exhaustiva el comportamiento del agente, evaluando aspectos como las alucinaciones, cuán correcta es una respuesta y si se han utilizado las herramientas adecuadas con los argumentos pertinentes:
# 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
]
Una verdadera puerta de calidad
Un pipeline que no bloquea las implementaciones defectuosas es simplemente un monitor. El detector de regresiones relaciona los resultados fuera de línea con el pipeline de implementación:
# 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,
}
El gancho CI es un script breve que sale con un código distinto de cero cuando aparece una regresión, lo cual es suficiente para detener la mayoría de los 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])
Pruebas A/B de versiones
Una prueba A/B ejecuta dos configuraciones del agente uno al lado del otro con tráfico real y compara sus resultados antes de que la nueva versión sea adoptada por completo:
# 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,
}
Cerrando el ciclo con la memoria
El campo de memoria procedural success_rate solía actualizarse después de cada ejecución, pero la medida era rudimentaria: la ejecución terminaba o no. Con un pipeline de evaluación adecuado en su lugar, la conexión es sencilla, y las puntuaciones de los jueces pueden integrarse en el almacenamiento de memoria:
# 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}",
}
)
Qué suele salir mal
La infraestructura de evaluación no es algo llamativo y nunca se lleva a cabo por completo. Problemas típicos:
Los jueces no son neutrales. Los modelos de juez tienen prejuicios. Tienden a favorecer respuestas más largas y seguras, escritas en un estilo similar al de sus datos de entrenamiento; por lo tanto, una respuesta corta y correcta puede obtener una calificación más baja que una larga pero ligeramente incorrecta. Compare periódicamente al juez con las calificaciones humanas y revise las rúbricas cuando aparezcan prejuicios sistemáticos.
Los conjuntos de datos ideales se vuelven obsoletos. La distribución de tareas en producción cambia con el tiempo. Un conjunto creado hace seis meses podría no reflejar lo que enfrenta actualmente el agente. Realice sesiones regulares de curaduría, especialmente después de cambios importantes en el área del producto que maneja el agente.
Las tasas de aprobación ocultan detalles importantes. Un sistema puede mantener una alta tasa de aprobación mientras empeora en los casos que son más cruciales. Haga un seguimiento de los fallos por categoría y por nivel de dificultad por separado. Una disminución en los casos DIFÍCILES es más preocupante que una en los casos FÁCILES, incluso si la tasa general se mantiene estable.
Las pruebas A/B requieren más tráfico del esperado. Para demostrar significancia estadística con una mejora de calidad del 3%, se necesitan al menos cientos de muestras. Con poco tráfico, las pruebas pueden durar semanas. Decida la duración de antemano y no declare un ganador prematuramente.
El proceso de evaluación puede contener errores. Los mensajes de evaluación pueden estar ligeramente incorrectos, los casos de referencia podrían codificar un comportamiento esperado erróneo y el código de puntuación puede presentar errores de tipo “off-by-one”. Trátelo con la misma seriedad que el código del agente: pruébelo, revíselo y esté atento a distribuciones de puntuaciones inusuales que indiquen un error en la puntuación y no un cambio real en la calidad.
Todo el sistema
Toda la arquitectura, desde el conjunto de datos de referencia pasando por la evaluación offline y online hasta las puertas de control integrado y el feedback de memoria:
+----------------------------+
| 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) |
+-----------------------+
Y la tecnología detrás de cada componente:
+-----------------------------+---------------------+------------------------------+
| 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 |
+-----------------------------+---------------------+------------------------------+
Lecturas relacionadas
- Orquestación de múltiples agentes en Bedrock y LangGraph: cuatro patrones — Patrones de supervisor-trabajador, pipeline, difusión y debate con LangGraph en AWS Bedrock, además de mecanismos de transferencia de tareas, protección de presupuestos, seguimiento y modos de fallo en producción.
- Memoria semántica de agentes en LangGraph con Qdrant y recuperación entre hilos — Proporcionar a un agente de LangGraph memoria a largo plazo mediante Qdrant: incrustar datos del usuario, recuperarlos entre diferentes identificadores de hilos y agregar filtros de carga útil para lograr aislamiento multiinquilino.