Evaluating AI Agents: Golden Sets, LLM Judges and CI Regression Gates
Offline and online evaluation for agents with AgentCore, LangFuse, Ragas and DeepEval, plus golden datasets, CI gating, A/B tests and memory feedback.
This article is published in English.
Sources: Agentic Architectures — Article 8: Evaluation and Continuous Improvement Pipelines
Sooner or later in an agent project, somebody asks a deceptively hard question: is it actually improving? A system prompt gets rewritten, a verification loop adjusted, a new memory tier added, and the agent seems better in testing. Seeming better isn't a measurement. Without measurement in production, regressions stay hidden until a user finds them, and nobody can tell whether harness changes helped. Optimizing by instinct is fine as a start and a poor way to continue.
Evaluation tooling has matured a great deal. AWS AgentCore Evaluations, LangFuse, Ragas and DeepEval each supply strong building blocks. The conceptual work of assembling them into a pipeline that catches problems before users do still falls to the team. The sections below walk through that assembly, using a document-processing agent as the running example, and tie the procedural memory success_rate field from an earlier memory design to a real scoring system.
What gets covered
- Offline versus online evaluation. They need different infrastructure, measure different things and catch different failures. Teams that run only one keep being surprised by the other's problems.
- LLM-as-judge at scale. A systematic scoring layer over every production trace that flags regressions automatically, in place of a one-off check.
- Tool comparison. AgentCore Evaluations, LangFuse, Ragas and DeepEval each cover a different layer, and working code shows when to reach for which.
- Golden datasets. Where test cases originate, how to curate them from production traces and how to prevent staleness.
- Regression detection in CI. A harness change that lowers quality should block deployment.
- A/B testing agent versions. Route a share of live traffic to a new configuration and compare before committing.
- Feeding results into memory. The
success_ratefield finally gets real measurement logic.
Offline and online evaluation
The two modes aren't interchangeable. Both are needed, and here is how they compare:
+---------------------------+-----------------------------+-----------------------------+
| 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 |
+---------------------------+-----------------------------+-----------------------------+
Results from one feed the other:
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) |
+------------------+
The loop is deliberate. Production traces showing difficult or interesting cases get promoted into the golden dataset, so later offline runs reflect actual usage better.
Golden datasets
Everything offline depends on the quality of test cases. A weak golden set breeds false confidence: every test passes, the release ships, and the real failure modes turn out never to have been tested. Cases come from three places: hand-written ones for known edge cases, ones curated from production traces where the agent struggled, and ones capturing important happy paths that must keep working.
# 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-as-judge
Agent output often has no single right answer. A contract-analysis agent can produce three valid summaries of one contract, and a code-review agent can report the same vulnerabilities in different order and wording. Exact-match scoring fails there. The practical answer is a separate model that grades output against a rubric and returns a structured score. The judge needs explicit, unambiguous criteria, because vague rubrics yield inconsistent scores that are useless for spotting regressions.
# 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", []),
)
AgentCore Evaluations
AWS AgentCore Evaluations offers managed evaluation infrastructure for agents on Bedrock. It orchestrates evaluation runs, stores results and aggregates metrics. It doesn't write rubrics, which remain the team's responsibility.
# 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 for online scoring
LangFuse handles the online side. Each production trace is scored in the background, the scores roll up into trend lines and unusual drops raise alerts. The scoring must stay asynchronous, otherwise it adds delay to the agent's replies.
# 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 and DeepEval
A general judge covers broad quality, while Ragas and DeepEval handle specialized metrics that a general rubric struggles to capture.
Ragas targets RAG evaluation. For an agent retrieving from Bedrock Knowledge Bases, it provides faithfulness, answer relevancy and context precision out of the box:
# 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 casts a wider net over agent behavior, scoring things such as hallucinations, how correct an answer is and whether the right tools were called with the right arguments:
# 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
]
A real quality gate
A pipeline that doesn't block bad deployments is just a monitor. The regression detector links offline results to the deployment pipeline:
# 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,
}
The CI hook is a short script that exits with a non-zero code when a regression appears, which is enough to stop most 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])
A/B testing versions
An A/B test runs two agent configurations side by side on real traffic and compares their scores before the new version is fully adopted:
# 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,
}
Closing the loop with memory
The procedural memory success_rate field used to be updated after each run, but the measure was crude: the run either finished or it didn't. With a proper evaluation pipeline in place, the connection is straightforward, and judge scores can flow into the memory store:
# 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}",
}
)
What tends to go wrong
Evaluation infrastructure isn't glamorous and is never done. Typical problems:
Judges aren't neutral. A judge model has biases. It tends to favor longer, more confident outputs written in a style close to its training data, so a short correct answer may score below a long, subtly wrong one. Compare the judge with human ratings from time to time and revise rubrics when systematic bias shows up.
Golden datasets go stale. Production task distribution drifts. A set built six months ago may not reflect what the agent faces now. Hold regular curation sessions, particularly after big changes to the product area the agent covers.
Pass rates hide detail. A system can keep a high pass rate while worsening on the cases that matter most. Track failures by tag and by difficulty separately. A drop on HARD cases is more alarming than one on EASY cases, even if the aggregate holds.
A/B tests need more traffic than expected. Showing statistical significance for a 3% quality gain takes hundreds of samples at least. With low traffic, tests can run for weeks. Decide the duration up front and don't call a winner early.
The eval pipeline can have bugs. Judge prompts may be subtly wrong, golden cases may encode incorrect expected behavior and scoring code can have off-by-one errors. Treat it as seriously as the agent code: test it, review it, and watch for odd score distributions that point to a scoring bug and not a real quality shift.
The whole system
The full architecture, from golden dataset through offline and online scoring to CI gates and memory feedback:
+----------------------------+
| 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) |
+-----------------------+
And the technology behind each component:
+-----------------------------+---------------------+------------------------------+
| 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 |
+-----------------------------+---------------------+------------------------------+