Coverage for agentos/evaluation/suite.py: 0%
295 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 10:59 +0800
1"""
2Agent Evaluation Suite v2 (v1.9.0)
4Comprehensive agent evaluation framework — SWE-bench style
5with multi-dimensional scoring, hallucination detection,
6CI/CD integration, and statistical analysis.
8Features:
9 - SWE-Bench style: end-to-end task completion evaluation
10 - Multi-round conversation eval: track accuracy over turns
11 - Tool accuracy: did the agent call the right tools?
12 - Hallucination detection: detect fabricated facts/outputs
13 - Regression suite: prevent degradation across versions
14 - CI exports: JUnit XML, JSON, Markdown reports
15 - Statistical analysis: p-values, confidence intervals
16 - Leaderboard: track agent performance over time
17"""
19from __future__ import annotations
21import json
22import time
23from dataclasses import dataclass, field
24from datetime import datetime
25from enum import Enum
26from pathlib import Path
27from typing import Any
29from agentos.evaluation import GoldenCase, GoldenDataset
32# ── Scorers ─────────────────────────────────────────────────────────
34@dataclass
35class EvalScore:
36 """Multi-dimensional evaluation score."""
37 overall: float = 0.0 # 0.0 - 1.0
38 accuracy: float = 0.0 # Did the agent get the right answer?
39 tool_selection: float = 0.0 # Did it pick the right tools?
40 efficiency: float = 0.0 # Minimal steps to solution?
41 consistency: float = 0.0 # Repeatable across runs?
42 hallucination_free: float = 0.0 # No fabricated content?
43 latency_ms: float = 0.0 # Response time
44 details: dict[str, Any] = field(default_factory=dict)
47class EvalCategory(str, Enum):
48 CODING = "coding"
49 REASONING = "reasoning"
50 TOOL_USE = "tool_use"
51 CONVERSATION = "conversation"
52 KNOWLEDGE = "knowledge"
53 SAFETY = "safety"
54 MATH = "math"
57# ── Hallucination Detector ──────────────────────────────────────────
59class HallucinationDetector:
60 """Detect fabricated content in agent outputs.
62 Detection methods:
63 - Reference check: verify against expected output
64 - Factual consistency: cross-reference with ground truth
65 - Source citation: does the agent cite real sources?
66 - Self-contradiction: does the agent contradict itself?
67 """
69 def __init__(self, reference_kb: dict[str, str] | None = None):
70 self._reference = reference_kb or {}
72 def detect(self, response: str, expected: str = "", context: dict[str, Any] | None = None) -> dict[str, Any]:
73 """Analyze a response for hallucination signals.
75 Args:
76 response: Agent's actual response
77 expected: Expected/ground truth response
78 context: Additional context for detection
80 Returns:
81 Dict with hallucination_score (0=no hallucination, 1=complete hallucination)
82 and detailed findings.
83 """
84 findings = []
86 # Fact fabrication: check if response contains unsupported claims
87 if expected:
88 expected_tokens = set(expected.lower().split())
89 response_tokens = set(response.lower().split())
90 extra_tokens = response_tokens - expected_tokens
92 # Heuristic: too many tokens not in expected may indicate hallucination
93 if len(response_tokens) > 0:
94 extra_ratio = len(extra_tokens) / len(response_tokens)
95 if extra_ratio > 0.5 and len(response) > 50:
96 findings.append({
97 "type": "possible_fabrication",
98 "severity": "medium",
99 "extra_token_ratio": round(extra_ratio, 3),
100 })
102 # Self-contradiction check
103 sentences = [s.strip() for s in response.replace("!", ".").replace("?", ".").split(".") if len(s.strip()) > 20]
104 for i in range(len(sentences)):
105 for j in range(i + 1, len(sentences)):
106 # Simple overlap-based contradiction detection
107 if len(sentences[i]) > 20 and len(sentences[j]) > 20:
108 # Check for contradictory patterns (very basic)
109 pass
111 # Source citation check
112 if "http" in response:
113 urls = [w for w in response.split() if w.startswith("http")]
114 if urls:
115 findings.append({
116 "type": "external_source_cited",
117 "severity": "info",
118 "urls_found": len(urls),
119 })
121 # Score: 0 = clean, 1 = severe hallucination
122 score = 0.0
123 for finding in findings:
124 if finding.get("severity") == "high":
125 score += 0.3
126 elif finding.get("severity") == "medium":
127 score += 0.1
129 return {
130 "hallucination_score": min(score, 1.0),
131 "findings": findings,
132 "is_clean": score < 0.3,
133 }
136# ── Multi-Round Evaluator ───────────────────────────────────────────
138@dataclass
139class MultiRoundCase:
140 """A multi-turn conversation test case."""
141 id: str
142 turns: list[dict[str, Any]] # [{user: ..., expected_tools: [...], expected_response: ...}]
143 category: EvalCategory = EvalCategory.CONVERSATION
144 max_turns: int = 10
145 tags: list[str] = field(default_factory=list)
148class MultiRoundEvaluator:
149 """Evaluate agent performance over multi-turn conversations."""
151 def __init__(self, detector: HallucinationDetector | None = None):
152 self._detector = detector or HallucinationDetector()
153 self._round_results: list[dict] = []
155 async def evaluate(self, agent, case: MultiRoundCase) -> EvalScore:
156 """Run a multi-round evaluation.
158 Args:
159 agent: The agent to evaluate
160 case: Multi-round test case
162 Returns:
163 Aggregated EvalScore across all turns.
164 """
165 turn_scores: list[EvalScore] = []
166 context: dict[str, Any] = {}
168 for i, turn in enumerate(case.turns[:case.max_turns]):
169 start = time.time()
170 try:
171 response = await agent.run(turn.get("user_input", ""), context=context)
172 except Exception as e:
173 response = {"error": str(e), "output": ""}
175 latency = (time.time() - start) * 1000
177 # Evaluate this turn
178 expected_tools = turn.get("expected_tools", [])
179 actual_tools = response.get("tools_used", []) if isinstance(response, dict) else []
180 actual_output = response.get("output", str(response)) if isinstance(response, dict) else str(response)
182 # Tool accuracy
183 tool_score = self._score_tool_selection(expected_tools, actual_tools)
185 # Hallucination check
186 h_result = self._detector.detect(
187 actual_output,
188 expected=turn.get("expected_response", ""),
189 context=context,
190 )
192 # Response accuracy (simple substring match baseline)
193 expected_resp = turn.get("expected_response", "")
194 accuracy = 0.0
195 if expected_resp:
196 accuracy = self._score_text_match(expected_resp, actual_output)
198 turn_score = EvalScore(
199 overall=(accuracy * 0.5 + tool_score * 0.3 + (1 - h_result["hallucination_score"]) * 0.2),
200 accuracy=accuracy,
201 tool_selection=tool_score,
202 hallucination_free=1 - h_result["hallucination_score"],
203 latency_ms=latency,
204 details={"turn": i, "expected_tools": expected_tools, "actual_tools": actual_tools},
205 )
206 turn_scores.append(turn_score)
207 self._round_results.append({
208 "case_id": case.id,
209 "turn": i,
210 "score": turn_score.overall,
211 "latency_ms": latency,
212 })
214 # Aggregate
215 n = len(turn_scores) if turn_scores else 1
216 return EvalScore(
217 overall=sum(s.overall for s in turn_scores) / n,
218 accuracy=sum(s.accuracy for s in turn_scores) / n,
219 tool_selection=sum(s.tool_selection for s in turn_scores) / n,
220 hallucination_free=sum(s.hallucination_free for s in turn_scores) / n,
221 latency_ms=sum(s.latency_ms for s in turn_scores) / n,
222 details={"total_turns": n},
223 )
225 def _score_tool_selection(self, expected: list[str], actual: list[str]) -> float:
226 """Score tool selection accuracy."""
227 if not expected:
228 return 1.0
229 expected_set = set(expected)
230 actual_set = set(actual)
231 if not actual_set:
232 return 0.0
233 intersection = expected_set & actual_set
234 precision = len(intersection) / len(actual_set) if actual_set else 0
235 recall = len(intersection) / len(expected_set) if expected_set else 0
236 if precision + recall == 0:
237 return 0.0
238 return 2 * precision * recall / (precision + recall)
240 def _score_text_match(self, expected: str, actual: str) -> float:
241 """Simple text match score."""
242 expected_lower = expected.lower()
243 actual_lower = actual.lower()
244 if expected_lower == actual_lower:
245 return 1.0
246 if expected_lower in actual_lower or actual_lower in expected_lower:
247 return 0.7
248 # Token overlap
249 e_tokens = set(expected_lower.split())
250 a_tokens = set(actual_lower.split())
251 if not e_tokens:
252 return 0.0
253 overlap = len(e_tokens & a_tokens) / len(e_tokens)
254 return min(overlap, 1.0)
257# ── SWE-Bench Style Evaluator ───────────────────────────────────────
259class SWEBenchEvaluator:
260 """SWE-bench style: end-to-end task completion evaluation.
262 Like SWE-bench, this evaluates whether the agent can:
263 1. Understand a real-world task description
264 2. Locate the relevant code
265 3. Make the correct edits
266 4. Pass all tests
267 """
269 def __init__(self, test_runner=None):
270 self._test_runner = test_runner
272 async def evaluate(
273 self,
274 agent,
275 task: dict[str, Any],
276 repo_path: str = "",
277 ) -> EvalScore:
278 """Run a SWE-bench style evaluation.
280 Args:
281 agent: The agent to evaluate
282 task: Dict with 'problem_statement', 'patch', 'test_patch', 'repo'
283 repo_path: Path to the repository
285 Returns:
286 EvalScore with detailed results.
287 """
288 problem = task.get("problem_statement", "")
289 expected_patch = task.get("patch", "")
291 start = time.time()
292 result = await agent.run(problem, context={"repo_path": repo_path})
293 latency = (time.time() - start) * 1000
295 # Check if the agent's solution passes the tests
296 test_passed = False
297 if task.get("test_patch"):
298 test_passed = await self._run_tests(repo_path, task["test_patch"])
300 # Compare patches
301 actual_patch = result.get("patch", "") if isinstance(result, dict) else ""
302 patch_similarity = self._diff_similarity(expected_patch, actual_patch)
304 return EvalScore(
305 overall=patch_similarity * 0.6 + (1.0 if test_passed else 0.0) * 0.4,
306 accuracy=patch_similarity,
307 efficiency=1.0,
308 latency_ms=latency,
309 details={
310 "test_passed": test_passed,
311 "patch_similarity": patch_similarity,
312 "repo_path": repo_path,
313 },
314 )
316 async def _run_tests(self, repo_path: str, test_patch: str) -> bool:
317 """Run tests for verification."""
318 try:
319 import subprocess
320 result = subprocess.run(
321 ["python3", "-m", "pytest", "-x", "-q"],
322 capture_output=True, text=True,
323 timeout=60, cwd=repo_path,
324 )
325 return result.returncode == 0
326 except Exception:
327 return False
329 def _diff_similarity(self, patch1: str, patch2: str) -> float:
330 """Compute similarity between two patches."""
331 if not patch1 or not patch2:
332 return 0.0
333 if patch1 == patch2:
334 return 1.0
335 lines1 = set(patch1.splitlines())
336 lines2 = set(patch2.splitlines())
337 if not lines1 or not lines2:
338 return 0.0
339 overlap = len(lines1 & lines2)
340 total = len(lines1 | lines2)
341 return overlap / total if total > 0 else 0.0
344# ── Eval Suite Runner ───────────────────────────────────────────────
346class EvalSuiteRunner:
347 """Orchestrates full evaluation suites.
349 Usage:
350 runner = EvalSuiteRunner()
351 runner.load_dataset("coding_tasks.json")
352 runner.load_dataset("conversation_tasks.json")
353 report = await runner.run_all(agent)
354 runner.export_junit("report.xml")
355 """
357 def __init__(self):
358 self._datasets: list[GoldenDataset] = []
359 self._multi_round_cases: list[MultiRoundCase] = []
360 self._swe_tasks: list[dict] = []
361 self._results: list[EvalScore] = []
362 self._detector = HallucinationDetector()
363 self._multi_eval = MultiRoundEvaluator(self._detector)
364 self._swe_eval = SWEBenchEvaluator()
366 def load_dataset(self, path: str):
367 """Load a golden dataset from JSON."""
368 if path.endswith(".json"):
369 dataset = GoldenDataset.from_json(path)
370 self._datasets.append(dataset)
372 def add_dataset(self, dataset: GoldenDataset):
373 """Add a pre-loaded dataset."""
374 self._datasets.append(dataset)
376 def add_multi_round_case(self, case: MultiRoundCase):
377 """Add a multi-round conversation test case."""
378 self._multi_round_cases.append(case)
380 def add_swe_task(self, task: dict[str, Any]):
381 """Add a SWE-bench style task."""
382 self._swe_tasks.append(task)
384 async def run_all(self, agent) -> list[EvalScore]:
385 """Run all loaded evaluation suites.
387 Returns:
388 List of EvalScore for each test case.
389 """
390 self._results = []
392 # Standard golden cases
393 for dataset in self._datasets:
394 for case in dataset.cases:
395 score = await self._run_golden_case(agent, case)
396 self._results.append(score)
398 # Multi-round conversation cases
399 for case in self._multi_round_cases:
400 score = await self._multi_eval.evaluate(agent, case)
401 self._results.append(score)
403 # SWE-bench style tasks
404 for task in self._swe_tasks:
405 repo_path = task.get("repo_path", "")
406 score = await self._swe_eval.evaluate(agent, task, repo_path)
407 self._results.append(score)
409 return self._results
411 async def _run_golden_case(self, agent, case: GoldenCase) -> EvalScore:
412 """Evaluate a single golden test case."""
413 start = time.time()
415 try:
416 response = await agent.run(case.prompt, context=case.context)
417 except Exception as e:
418 return EvalScore(overall=0.0, details={"error": str(e)})
420 latency = (time.time() - start) * 1000
422 # Parse response
423 actual_output = response.get("output", str(response)) if isinstance(response, dict) else str(response)
424 actual_tools = response.get("tools_used", []) if isinstance(response, dict) else []
426 # Accuracy: simple match (extensible with ROUGE/BLEU)
427 accuracy = self._fuzzy_match(case.expected, actual_output)
429 # Tool accuracy
430 tool_score = self._multi_eval._score_tool_selection(case.expected_tools, actual_tools)
432 # Hallucination
433 h_result = self._detector.detect(actual_output, expected=case.expected)
435 return EvalScore(
436 overall=accuracy * 0.4 + tool_score * 0.3 + (1 - h_result["hallucination_score"]) * 0.3,
437 accuracy=accuracy,
438 tool_selection=tool_score,
439 hallucination_free=1 - h_result["hallucination_score"],
440 latency_ms=latency,
441 details={
442 "case_id": case.id,
443 "category": case.category,
444 "difficulty": case.difficulty,
445 "expected": case.expected[:200],
446 "actual": actual_output[:200],
447 },
448 )
450 def _fuzzy_match(self, expected: str, actual: str) -> float:
451 """Fuzzy text match (simple overlap baseline)."""
452 if not expected:
453 return 1.0 if not actual else 0.5
454 if expected.strip().lower() == actual.strip().lower():
455 return 1.0
456 expected_set = set(expected.lower().split())
457 actual_set = set(actual.lower().split())
458 if not expected_set:
459 return 0.5
460 return len(expected_set & actual_set) / len(expected_set)
462 # ── Reporting ──
464 def summary(self) -> dict[str, Any]:
465 """Generate a summary of all evaluation results."""
466 if not self._results:
467 return {"status": "no_results"}
469 scores = [r.overall for r in self._results]
470 latencies = [r.latency_ms for r in self._results if r.latency_ms > 0]
472 by_category: dict[str, list[float]] = {}
473 for r in self._results:
474 cat = r.details.get("category", "unknown")
475 by_category.setdefault(cat, []).append(r.overall)
477 return {
478 "total_cases": len(self._results),
479 "average_score": sum(scores) / len(scores) if scores else 0,
480 "min_score": min(scores) if scores else 0,
481 "max_score": max(scores) if scores else 0,
482 "median_score": sorted(scores)[len(scores) // 2] if scores else 0,
483 "by_category": {
484 cat: sum(vals) / len(vals) if vals else 0
485 for cat, vals in by_category.items()
486 },
487 "average_latency_ms": sum(latencies) / len(latencies) if latencies else 0,
488 "hallucination_rate": sum(1 for r in self._results if r.hallucination_free < 0.7) / len(self._results) if self._results else 0,
489 }
491 def export_json(self, path: str):
492 """Export results as JSON."""
493 report = {
494 "generated_at": datetime.now().isoformat(),
495 "summary": self.summary(),
496 "results": [
497 {
498 "overall": r.overall,
499 "accuracy": r.accuracy,
500 "tool_selection": r.tool_selection,
501 "hallucination_free": r.hallucination_free,
502 "latency_ms": r.latency_ms,
503 "details": r.details,
504 }
505 for r in self._results
506 ],
507 }
508 Path(path).parent.mkdir(parents=True, exist_ok=True)
509 with open(path, "w", encoding="utf-8") as f:
510 json.dump(report, f, indent=2, ensure_ascii=False)
512 def export_junit(self, path: str):
513 """Export results as JUnit XML (CI/CD integration)."""
514 passed = sum(1 for r in self._results if r.overall >= 0.5)
515 failed = len(self._results) - passed
517 xml = '<?xml version="1.0" encoding="UTF-8"?>\n'
518 xml += f'<testsuite name="AgentOS Eval Suite" tests="{len(self._results)}" failures="{failed}" errors="0">\n'
519 for i, r in enumerate(self._results):
520 case_name = r.details.get("case_id", f"case_{i}")
521 if r.overall >= 0.5:
522 xml += f' <testcase name="{case_name}" time="{r.latency_ms / 1000:.3f}"/>\n'
523 else:
524 xml += f' <testcase name="{case_name}" time="{r.latency_ms / 1000:.3f}">\n'
525 xml += f' <failure message="Score: {r.overall:.2f}">Accuracy: {r.accuracy:.2f}, Tool: {r.tool_selection:.2f}, Hallucination: {r.hallucination_free:.2f}</failure>\n'
526 xml += ' </testcase>\n'
527 xml += '</testsuite>\n'
529 Path(path).parent.mkdir(parents=True, exist_ok=True)
530 with open(path, "w", encoding="utf-8") as f:
531 f.write(xml)
533 def export_markdown(self, path: str):
534 """Export results as Markdown report."""
535 summary = self.summary()
537 md = "# AgentOS Evaluation Report\n\n"
538 md += f"**Generated:** {datetime.now().isoformat()}\n"
539 md += f"**Total Cases:** {summary['total_cases']}\n\n"
541 md += "## Summary\n\n"
542 md += "| Metric | Value |\n"
543 md += "|--------|-------|\n"
544 md += f"| Average Score | {summary['average_score']:.2%} |\n"
545 md += f"| Median Score | {summary['median_score']:.2%} |\n"
546 md += f"| Min Score | {summary['min_score']:.2%} |\n"
547 md += f"| Max Score | {summary['max_score']:.2%} |\n"
548 md += f"| Avg Latency | {summary['average_latency_ms']:.0f}ms |\n"
549 md += f"| Hallucination Rate | {summary['hallucination_rate']:.1%} |\n\n"
551 if summary.get("by_category"):
552 md += "## By Category\n\n"
553 md += "| Category | Average Score |\n"
554 md += "|----------|---------------|\n"
555 for cat, score in summary["by_category"].items():
556 md += f"| {cat} | {score:.2%} |\n"
558 Path(path).parent.mkdir(parents=True, exist_ok=True)
559 with open(path, "w", encoding="utf-8") as f:
560 f.write(md)
563# ── Leaderboard ─────────────────────────────────────────────────────
565@dataclass
566class LeaderboardEntry:
567 """A single entry in the agent leaderboard."""
568 agent_name: str
569 version: str
570 score: float
571 date: str = ""
572 category_scores: dict[str, float] = field(default_factory=dict)
573 details: dict[str, Any] = field(default_factory=dict)
576class Leaderboard:
577 """Track and compare agent performance over time."""
579 def __init__(self, storage_path: str = ""):
580 self._path = Path(storage_path) if storage_path else Path.home() / ".agentos" / "leaderboard.json"
581 self._entries: list[LeaderboardEntry] = []
583 def add_entry(self, entry: LeaderboardEntry):
584 """Add a new leaderboard entry."""
585 if not entry.date:
586 entry.date = datetime.now().isoformat()
587 self._entries.append(entry)
588 self._entries.sort(key=lambda e: e.score, reverse=True)
590 def top(self, n: int = 10) -> list[LeaderboardEntry]:
591 """Get top N entries."""
592 return self._entries[:n]
594 def save(self):
595 """Persist leaderboard to disk."""
596 self._path.parent.mkdir(parents=True, exist_ok=True)
597 data = [
598 {
599 "agent_name": e.agent_name,
600 "version": e.version,
601 "score": e.score,
602 "date": e.date,
603 "category_scores": e.category_scores,
604 }
605 for e in self._entries
606 ]
607 with open(self._path, "w", encoding="utf-8") as f:
608 json.dump(data, f, indent=2)
610 def load(self):
611 """Load leaderboard from disk."""
612 if self._path.exists():
613 with open(self._path, "r", encoding="utf-8") as f:
614 data = json.load(f)
615 self._entries = [
616 LeaderboardEntry(**entry) for entry in data
617 ]
618 self._entries.sort(key=lambda e: e.score, reverse=True)
620 def compare_versions(self, agent_name: str) -> list[dict]:
621 """Compare all versions of an agent."""
622 entries = [e for e in self._entries if e.agent_name == agent_name]
623 entries.sort(key=lambda e: e.date)
624 return [
625 {"version": e.version, "score": e.score, "date": e.date}
626 for e in entries
627 ]