Ацэнка AI-агентаў: эталонныя наборы, LLM-суддзі і рэгрэсійныя фільтры ў CI
Афлайн- і анлайн-ацэнка агентаў з AgentCore, LangFuse, Ragas і DeepEval, а таксама эталонныя наборы, фільтры ў CI, A/B-тэсты і зваротная сувязь праз памяць.
Крыніцы: Agentic Architectures — Article 8: Evaluation and Continuous Improvement Pipelines
Раней чым пазней у праекце з агентамі хтось задае на першы погляд складны запіт: чы рэальна ўлучшэнне? Складзіцца новы запрос для сістэмы, налагоджуецца цыкл парабярання, дадаёцца новы рэверс мемарыяў, і агент здаецца лепшы пад час тэставання. Але проста здача лепшым — гэта не памер. Без памераў у рэальнай працы рэгрэсіі застаюцца непазначанымі, пакуль іх не знайдзе корыстнік, і ніхто не можа сказаць, чы змяны ў інфраструктурэ дапамоглі. Оптымізацыя на інстынкты падходзіць як пачатак, але ёй нельга продаважваць як основны спосаб працы.
Інструменты для аналізу значныя чыглы ў розвіце. AWS AgentCore Evaluations, LangFuse, Ragas і DeepEval кожны з яных ствараюць моцныя базовыя елементы. Аднак концэптуальная праца па ўзаемным з’еднанні іх у алгорытм, які выявляе проблемы ранейш, чым іх пазнаюць корыстувальнікі, як і раней, лягае на команду. У наступных раздзелах показана процедура такога з’еднання на прыкладзе агента для апрацоўкі дакументаў, пры чым полье success_rate з ранейшага дызайну памяці прыўязваецца да рэальной сістэмы оценкі.
Што будзе рассмотрана
- Аналіз па афляйну протыва онлаўн-аналізу. Яны выкарыстоваюць разную інфраструктуру, меркуюць розныя показнікі і выявляюць разныя прычыны збоёў. Команды, якія вядуць толькі аднаго з варыянтаў, продовжваюць страждаць ад проблем другага.
success_rate нарэшце атрымвае рэальную логіку вылічэння.Адміністрування поза мерэжай і ў меражы
Гэтыя два рэжымы не можна заменіць адным другым. Абодва неабходны, і хтоць як воні параджуцца:
+---------------------------+-----------------------------+-----------------------------+
| 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 |
+---------------------------+-----------------------------+-----------------------------+
Рэзультаты з аднаго крыніцы ў іншае:
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) |
+------------------+
Гэты цыкл ўзначальны. Следы з рэжыму працы, якія паказваюць складныя або цікавыя кейсы, прадвайваюцца ў «золатыя» наборы дадзеных, таму пазнейшыя офлайн-тэсты лепшая адбіваюць рэальную сітуацыю ўжывання.
«Золатыя» наборы дадзеных
Усё, што відбываецца офлайн, залежыць ад якосці тэстовых кейсаў. Слабы «золаты» набор стварае хыбную паверу: усі тэсты пройшлі, продукт выпускаецца, а выявляецца, што рэальныя спосабы збою ніколі не тэставаліся. Кейсы прадаюцца з трохоў месца: рукапісныя кейсы для вядомых крайніх сітуацыяў, кейсы, выбраныя з следаў з рэжыму працы, дзе агент страждала, і кейсы, якія фіксуюць важлівыя успішныя сценарыі, якія павінны продовжваць працаваць.
# 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-як суддзя
Расчынкі агента часта не маюць ўніверсальна правільнай адпаведзі. Агент для аналізу кантракта можа стварыць тры правамерныя падсумкі аднаго кантракта, а агент для перагляду коду можа апісаць тыя ж уразлівасці ў разным парадку і з разным формулюванням. Метод аблікавання точных падабенства тут не працуе. Практычныя рашэнні — гэта окольны мадэль, якая оцінюе расчынкі за певным крэтыярам і вяртае структураваны бал. Суддзі неабходны чыстая, без двусмысленнасці крэтыяры, таму што нечыстыя крэтыяры даюць нестабільныя балы, якія не падходзяць для выяўлення рэгрэсій.
# 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
AWS AgentCore адміністрацыя запрошвае інфраструктуру для керавання процэсамі адміністрацыі для агентаў на Bedrock. Яна коордынуе запуск процэсаў адміністрацыі, зберагае рэзультаты і агрэгуе паказнікі. Яна не стварае крэтырыяў оценкі, якія застаюцца адпаведальнасцю каманды.
# 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 для онлайн-ацэнкавання
LangFuse карае захад онлайн-часткі. Кожны трац працы ацэнкуецца у фонавым рэжыме, рэзультаты ацэнкавання сумуецца ў лініях тэндэнцый, а нечыстае падняцца робіць сігналы апавяшэння. Ацэнкаванне павінна застаўся асінхронным, інакш гэта прызводзіць да затрымкі у адпаведзях агента.
# 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 і DeepEval
Звычны суддзі ацэнююць загальную якосць, тады калі Ragas і DeepEval выкарыстоўваюць спецыялізаваныя показнікі, якія важка адобразіць за дапамой стандартных крэтарыў.
Ragas прызначаны для ацэнкі RAG. Для агента, які запошчае інфармацыю з баз знаёмых Bedrock, ён без дадатковых налаштаванняў забезпечвае адраднасць, рэлевантнасць адпаведзей і точнасць кантэксту:
# 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 аналізуе яксця дзейства агента, оцэнюючы такія праметры, як галюцынацыі, ступень правільнасці адпаведзя і тое, чыргульна былі выкорыстаны правыя інструменты з правымі аргументамі:
# 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
]
Праўdziва бар'ерна сістэма якасці
Сістэма, якая не блакуе пад час выканання некальканых развертань, ёсць проста манітар. Дзеяктэр рэгрэсii спаява рэзультаты паўтарных тэстав з працэсам развертання:
# 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,
}
CI хук — гэта короткі скрыпт, які завершаецца з кодам, не равным нулю, калі выявляецца рэгрэсія, і гэтага дастаткова, каб зупініць большасць пайплайнаў:
# 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
Пад час тэставання A/B два канфігурацыі агента працуюць паралельна на рэальным трафіку, і ўпораўнююцься іх рэзультаты прычымо да таго, калі новая версія не будзе ўжо цэлкам апрымлена:
# 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,
}
Закрыць цыкл з памяцю
Поле працэйнай памяці success_rate раней адкладвалася пасля кожнага запуску, але гэты показнік быў прымітны: запуск або завершаліся, або не завершаліся. Калі ёсць правільны працэс ацэнкі, зв’язак стае простым, і балы суддзяў можна перадаць у базу памяці:
# 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}",
}
)
Што частаюць робіць не так
Інфраструктура ацэнкаў не ёсць прыгожай і ніколі не выканана належна. Тыповыя проблемы:
Суддзі не ўніверсальныя. Мадэль суддзі мае уперадзе патэнцыі. Яна часта вярнуе перавагу дэльным, болей упевненым адпаведзенням, напісаным у стыле, блізкам да дадзеных, на якіх яна навчалася, таму короткае правільнае адпаведзенне можа мяць ніжэйшую оцэнку, чым дзёльнае, але незначна некорэктнае. Час а час перацэнюйце працу суддзі па адпаведным людскім оцэнкам і праверайце крытэрыя, калі з’яўляюцца сістэматычныя патэнцыі.
Золатыя наборы дадзеных стаюць застарэлымі. Распадзеўка задач у працэсе вырабоцтва зменяецца. Набор, створаны шэсьць месацоў таму, можа не відбіваць тое, з чым зараз сталкаецца агент. Практыкуйце рэгулярныя сесіі куратуры, особліва пасля значных змян у сферэ продукту, якую агент обрабоцвае.
Рэштаты пасаванняяў скрываюць дакладную інфармацыю. Сістэма можа падтрымваць высокі рэштат пасаванняяў, аднак пагіршаць працу ў найважлівейшых кэсах. Следзіце за няудачамі праз атрыбуты і па ступеню складнасці окрэма. Падыш у рэштатах кэсаў з класам HARD ёсць болей абалоўваючым, чым у кэсах з класам EASY, нават як загальны рэштат застаецца стабільным.
Тэсты тыпу A/B трэбуюць больш трафіку, чым прыпускаецца. Ёсчыленне статыстычнай значамасці паўстання якосці на 3% выкалічваецца ўжо з сотняў прыкладаў. Праз малы трафік тэсты можуць трываць недзеяў. Загадзя выберыце трываласць тэстаў і не прызначайце пазёмніка раней часу.
Працэс ацэнкі можа мець бягы. Запросы для суддзіў могу быць непаўнама правільнымі, ідэальныя кэсы могу мець некорэктна зафіксаванае апэктываўанае паведанне, а код для расчытку балоў можа мець памылкі тыпу «off-by-one». Ставіцеся да гэтага так сама серйозна, як і да коду агента: тэставайце, пераглядзіце і стараўсяцеся зазначыць незвычныя распады балоў, якія можуць свядчыць пра бяг у кодзе для расчытку, а не пра рэальную змяну якосці.
Весь сістэма
Полная архітэктура, начынającыся з базы дадзеных „Golden Dataset“, праходзячы праз офлайн- і онлайн-оцэнкі, да вората CI і адзынакоўвання памяці:
+----------------------------+
| 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) |
+-----------------------+
І тэхналогія, якая стоіць за кожным компанентам:
+-----------------------------+---------------------+------------------------------+
| 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 |
+-----------------------------+---------------------+------------------------------+
Супаўязаныя матэрыялы
- Оркестрацыя калькуляцый колькіма агентамі на Bedrock і LangGraph: чатыры патэрна — патэрны керавання, пайплайны, розповсюджэння і дыскусія з выкарыстоўваннем LangGraph на AWS Bedrock, а таксама механізмы перадачы задач, контролю бюджэту, трэйсінгу і адбору прыбуткаў у разы некальканасловных сітуацыях.
- Семантычная памяць агентаў у LangGraph з выкарыстоўваннем Qdrant і можлівасцю адзыскання дадзеных між ніткамі — стварэнне давялікай памяці для агентаў LangGraph за дапамогою Qdrant: уключэнне фактов корыстувачаў, адзысканне іх па ідэнтыфікаторах нітак і дадзенне фільтраў пакетаў для ізоляціі калькуляцый разным корыстувачам.