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