Оцінка штучних інтелектуальних агентів: золоті набори, судді з 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 Evaluations пропонує інфраструктуру для керованих оцінок агентів на платформі 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
]
Справжній контроль якості
Канал розгортання, який не блокує погані розгортання, є лише монітором. Детектор регресії пов’язує результати офлайн-тестування з процесом розгортання:
# 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,
}
Hook 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}",
}
)
Що зазвичай йде не так
Інфраструктура оцінювання не є привабливою та ніколи не використовується повністю. Типові проблеми:
Судді не є нейтральними. Модель судді має упередження. Вона схильна віддавати перевагу довшим, більш впевненим відповідям, написаним у стилі, близькому до даних її навчання, тож коротка правильна відповідь може отримати нижчий бал, ніж довга, ледь помітно неправильна. Час від часу порівнюйте оцінки судді з оцінками людей та коригуйте критерії оцінювання, коли з’являються систематичні упередження.
Золоті набори даних стають застарілими. Розподіл завдань у продакшені змінюється. Набір даних, створений шість місяців тому, може не відображати ситуацію, з якою зараз стикається агент. Проводьте регулярні сеанси підтримки якості, особливо після значних змін у сфері продукту, якою опікується агент.
Рівень проходження приховує деталі. Система може підтримувати високий рівень проходження, при цьому погіршуючи результати у найважливіших випадках. Відстежуйте невдачі окремо за тегами та рівнем складності. Погіршення результатів у складних випадках є більш тривожним, ніж у простих, навіть якщо загальний показник залишається стабільним.
Тести типу A/B вимагають більшого обсягу трафіку, ніж очікується. Щоб показати статистичну значущість покращення якості на 3%, потрібно щонайменше сотні зразків. При низькому обсязі трафіку тести можуть тривати тижнями. Визначайте тривалість заздалегідь та не оголошуйте переможця раніше часу.
Процес оцінки може містити помилки. Запити для оцінювання можуть бути трохи неправильними, ідеальні випадки можуть містити неправильно задану очікувану поведінку, а код оцінювання — помилки типу „один занадто“. Ставіться до цього так само серйозно, як і до коду агента: тестуйте його, перевіряйте та стежте за дивними розподілами балів, які вказують на помилку оцінювання, а не на справжню зміну якості.
Вся система
Повна архітектура, від золотого набору даних через офлайн- та онлайн-оцінювання до контрольних точок 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: вбудовування фактів користувача, їх повернення через ідентифікатори потоків та додавання фільтрів навантаження для ізоляції кількох користувачів.