Coverage for agentos/eval/benchmark.py: 44%
243 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-08 10:59 +0800
1"""
2v1.10.0: External Evaluation Harness — SWE-bench & GAIA benchmark runner.
4Supports:
5- SWE-bench: software engineering task resolution
6- GAIA: multi-step reasoning benchmark
7- Custom eval suites via registry
8- Scoring: pass@k, F1, exact match, semantic similarity
9"""
11from __future__ import annotations
13import json
14import statistics
15import time
16from collections.abc import Callable
17from dataclasses import dataclass, field
18from enum import StrEnum
19from pathlib import Path
20from typing import Any
22# ── Enums & Data Classes ──────────────────────────────────────────
25class EvalMetric(StrEnum):
26 """Supported evaluation metrics."""
28 PASS_AT_K = "pass@k" # Fraction of correct in k generations
29 EXACT_MATCH = "exact_match" # String equality
30 F1 = "f1" # F1 score (token overlap)
31 ROUGE_L = "rouge_l" # ROUGE-L
32 SEMANTIC_SIM = "semantic_sim" # Embedding cosine similarity
33 LLM_AS_JUDGE = "llm_as_judge" # LLM-graded
36class EvalSuite(StrEnum):
37 """Supported benchmark suites."""
39 SWE_BENCH = "swe-bench"
40 SWE_BENCH_LITE = "swe-bench-lite"
41 GAIA = "gaia"
42 GAIA_VAL = "gaia-validation"
43 CUSTOM = "custom"
46@dataclass
47class EvalCase:
48 """A single evaluation case."""
50 id: str
51 suite: EvalSuite
52 prompt: str
53 expected: str
54 repo: str = "" # For SWE-bench: git repo
55 base_commit: str = "" # For SWE-bench: base commit hash
56 test_patch: str = "" # For SWE-bench: test patch
57 metadata: dict[str, Any] = field(default_factory=dict)
60@dataclass
61class EvalSample:
62 """One generation sample for a case."""
64 case_id: str
65 sample_index: int # 0..k-1 for pass@k
66 generated: str
67 score: float = 0.0
68 passed: bool = False
69 latency_ms: float = 0.0
70 metadata: dict[str, Any] = field(default_factory=dict)
73@dataclass
74class EvalResult:
75 """Result for a single evaluation case (aggregated across samples)."""
77 case_id: str
78 suite: EvalSuite
79 metric: EvalMetric
80 score: float # pass@k or single-sample score
81 samples: list[EvalSample] = field(default_factory=list)
82 error: str = ""
85@dataclass
86class EvalReport:
87 """Full evaluation report across all cases."""
89 suite: EvalSuite
90 total_cases: int
91 passed_cases: int
92 avg_score: float
93 scores: list[float] = field(default_factory=list)
94 metric: EvalMetric = EvalMetric.EXACT_MATCH
95 results: list[EvalResult] = field(default_factory=list)
96 duration_s: float = 0.0
97 metadata: dict[str, Any] = field(default_factory=dict)
99 @property
100 def success_rate(self) -> float:
101 return self.passed_cases / max(self.total_cases, 1)
103 @property
104 def median_score(self) -> float:
105 return statistics.median(self.scores) if self.scores else 0.0
107 @property
108 def std_dev(self) -> float:
109 return statistics.stdev(self.scores) if len(self.scores) > 1 else 0.0
112# ── Scorers ────────────────────────────────────────────────────────
115class Scorer:
116 """Base scorer."""
118 def score(self, generated: str, expected: str) -> float:
119 raise NotImplementedError
121 @property
122 def metric(self) -> EvalMetric:
123 raise NotImplementedError
126class ExactMatchScorer(Scorer):
127 """Exact string match scorer."""
129 @property
130 def metric(self) -> EvalMetric:
131 return EvalMetric.EXACT_MATCH
133 def score(self, generated: str, expected: str) -> float:
134 if not expected:
135 return 1.0 if not generated else 0.0
136 return 1.0 if generated.strip() == expected.strip() else 0.0
139class F1Scorer(Scorer):
140 """Token-level F1 scorer."""
142 @property
143 def metric(self) -> EvalMetric:
144 return EvalMetric.F1
146 def score(self, generated: str, expected: str) -> float:
147 if not expected:
148 return 1.0 if not generated else 0.0
150 gen_tokens = set(generated.lower().split())
151 exp_tokens = set(expected.lower().split())
153 if not gen_tokens or not exp_tokens:
154 return 0.0
156 tp = len(gen_tokens & exp_tokens)
157 precision = tp / len(gen_tokens)
158 recall = tp / len(exp_tokens)
160 if precision + recall == 0:
161 return 0.0
162 return 2 * precision * recall / (precision + recall)
165class ROUGELScorer(Scorer):
166 """ROUGE-L scorer (longest common subsequence)."""
168 @property
169 def metric(self) -> EvalMetric:
170 return EvalMetric.ROUGE_L
172 @staticmethod
173 def _lcs_len(a: list[str], b: list[str]) -> int:
174 m, n = len(a), len(b)
175 dp = [[0] * (n + 1) for _ in range(m + 1)]
176 for i in range(1, m + 1):
177 for j in range(1, n + 1):
178 if a[i - 1] == b[j - 1]:
179 dp[i][j] = dp[i - 1][j - 1] + 1
180 else:
181 dp[i][j] = max(dp[i - 1][j], dp[i][j - 1])
182 return dp[m][n]
184 def score(self, generated: str, expected: str) -> float:
185 if not expected:
186 return 1.0 if not generated else 0.0
188 gen_tokens = generated.lower().split()
189 exp_tokens = expected.lower().split()
191 if not gen_tokens or not exp_tokens:
192 return 0.0
194 lcs = self._lcs_len(gen_tokens, exp_tokens)
195 precision = lcs / len(gen_tokens) if gen_tokens else 0
196 recall = lcs / len(exp_tokens) if exp_tokens else 0
198 if precision + recall == 0:
199 return 0.0
200 return 2 * precision * recall / (precision + recall)
203def get_scorer(metric: EvalMetric) -> Scorer:
204 """Factory: get scorer for a given metric."""
205 return {
206 EvalMetric.EXACT_MATCH: ExactMatchScorer(),
207 EvalMetric.F1: F1Scorer(),
208 EvalMetric.ROUGE_L: ROUGELScorer(),
209 }.get(metric, ExactMatchScorer())
212# ── SWE-bench Loader ──────────────────────────────────────────────
215class SWEBenchLoader:
216 """Load and parse SWE-bench dataset.
218 SWE-bench format: each instance is a GitHub issue with a known fix.
219 Agent must produce a patch that passes the test suite.
220 """
222 @staticmethod
223 def load(path: str | Path, subset: str = "lite") -> list[EvalCase]:
224 """Load SWE-bench instances from a JSON/JSONL file."""
225 path = Path(path)
226 cases = []
228 if not path.exists():
229 raise FileNotFoundError(f"SWE-bench dataset not found: {path}")
231 if path.suffix == ".jsonl":
232 with open(path) as f:
233 for line in f:
234 if line.strip():
235 instance = json.loads(line)
236 cases.append(SWEBenchLoader._parse(instance, subset))
237 elif path.suffix == ".json":
238 data = json.loads(path.read_text())
239 instances = data if isinstance(data, list) else data.get("instances", [])
240 for instance in instances:
241 cases.append(SWEBenchLoader._parse(instance, subset))
243 return cases
245 @staticmethod
246 def _parse(instance: dict, subset: str) -> EvalCase:
247 return EvalCase(
248 id=instance.get("instance_id", ""),
249 suite=EvalSuite.SWE_BENCH_LITE if "lite" in subset else EvalSuite.SWE_BENCH,
250 prompt=instance.get("problem_statement", instance.get("issue", "")),
251 expected="", # SWE-bench doesn't have expected text; it has a test patch
252 repo=instance.get("repo", ""),
253 base_commit=instance.get("base_commit", ""),
254 test_patch=instance.get("test_patch", instance.get("patch", "")),
255 metadata={
256 "hints_text": instance.get("hints_text", ""),
257 "version": instance.get("version", ""),
258 },
259 )
262# ── GAIA Loader ────────────────────────────────────────────────────
265class GAIALoader:
266 """Load and parse GAIA benchmark dataset.
268 GAIA: multi-step reasoning benchmark with 466 questions.
269 Levels: L1 (simple), L2 (medium), L3 (complex).
270 """
272 @staticmethod
273 def load(path: str | Path, level: str | None = None) -> list[EvalCase]:
274 """Load GAIA questions from JSON/JSONL."""
275 path = Path(path)
276 cases = []
278 if not path.exists():
279 raise FileNotFoundError(f"GAIA dataset not found: {path}")
281 if path.suffix == ".jsonl":
282 with open(path) as f:
283 for line in f:
284 if line.strip():
285 q = json.loads(line)
286 if level and q.get("Level", "") != level:
287 continue
288 cases.append(GAIALoader._parse(q, "validation" in path.name))
289 elif path.suffix == ".json":
290 data = json.loads(path.read_text())
291 questions = data if isinstance(data, list) else data.get("questions", [])
292 for q in questions:
293 if level and q.get("Level", "") != level:
294 continue
295 cases.append(GAIALoader._parse(q, "validation" in path.name))
297 return cases
299 @staticmethod
300 def _parse(q: dict, is_val: bool) -> EvalCase:
301 return EvalCase(
302 id=q.get("task_id", q.get("id", "")),
303 suite=EvalSuite.GAIA_VAL if is_val else EvalSuite.GAIA,
304 prompt=q.get("Question", q.get("question", "")),
305 expected=q.get("Final answer", q.get("answer", "")),
306 metadata={
307 "level": q.get("Level", ""),
308 "annotator_metadata": q.get("Annotator Metadata", ""),
309 },
310 )
313# ── Evaluation Runner ──────────────────────────────────────────────
316class EvalRunner:
317 """Run evaluations over multiple cases with pass@k support.
319 Usage:
320 runner = EvalRunner(generate_fn=my_agent.generate)
321 report = runner.run(cases, k=3, metric=EvalMetric.EXACT_MATCH)
322 """
324 def __init__(
325 self,
326 generate_fn: Callable[[str], str],
327 scorer: Scorer | None = None,
328 ):
329 """
330 Args:
331 generate_fn: Function (prompt) -> generated_text
332 scorer: Optional scorer override
333 """
334 self.generate = generate_fn
335 self.scorer = scorer
337 def run(
338 self,
339 cases: list[EvalCase],
340 k: int = 1,
341 metric: EvalMetric = EvalMetric.EXACT_MATCH,
342 on_case_start: Callable[[EvalCase], None] | None = None,
343 on_case_end: Callable[[EvalResult], None] | None = None,
344 ) -> EvalReport:
345 """Run evaluation on a list of cases.
347 Args:
348 cases: Evaluation cases
349 k: Number of samples per case (for pass@k)
350 metric: Scoring metric
351 on_case_start: Callback before each case
352 on_case_end: Callback after each case
354 Returns:
355 EvalReport with aggregated results
356 """
357 start_time = time.time()
358 scorer = self.scorer or get_scorer(metric)
359 results: list[EvalResult] = []
361 for case in cases:
362 if on_case_start:
363 on_case_start(case)
365 samples: list[EvalSample] = []
366 scores: list[float] = []
367 error = ""
369 for i in range(k):
370 try:
371 t0 = time.time()
372 generated = self.generate(case.prompt)
373 latency = (time.time() - t0) * 1000
375 s = scorer.score(generated, case.expected)
376 samples.append(
377 EvalSample(
378 case_id=case.id,
379 sample_index=i,
380 generated=generated,
381 score=s,
382 passed=s >= 0.5,
383 latency_ms=latency,
384 )
385 )
386 scores.append(s)
387 except Exception as e:
388 error = str(e)
389 samples.append(
390 EvalSample(
391 case_id=case.id,
392 sample_index=i,
393 generated="",
394 score=0.0,
395 passed=False,
396 latency_ms=0,
397 )
398 )
399 scores.append(0.0)
401 # pass@k: fraction where at least one sample passes
402 any(s.passed for s in samples)
403 # Use max score for the case score
404 case_score = max(scores) if scores else 0.0
406 result = EvalResult(
407 case_id=case.id,
408 suite=case.suite,
409 metric=metric,
410 score=case_score,
411 samples=samples,
412 error=error,
413 )
414 results.append(result)
416 if on_case_end:
417 on_case_end(result)
419 scores_list = [r.score for r in results]
420 passed = sum(1 for s in scores_list if s >= 0.5)
422 report = EvalReport(
423 suite=cases[0].suite if cases else EvalSuite.CUSTOM,
424 total_cases=len(cases),
425 passed_cases=passed,
426 avg_score=sum(scores_list) / max(len(scores_list), 1),
427 scores=scores_list,
428 metric=metric,
429 results=results,
430 duration_s=time.time() - start_time,
431 )
432 return report
434 def run_pass_at_k(
435 self,
436 cases: list[EvalCase],
437 k: int = 5,
438 metric: EvalMetric = EvalMetric.EXACT_MATCH,
439 ) -> EvalReport:
440 """Run pass@k evaluation (shorthand)."""
441 return self.run(cases, k=k, metric=metric)
443 def print_report(self, report: EvalReport) -> str:
444 """Generate a human-readable report string."""
445 lines = [
446 "╔══ Evaluation Report ══╗",
447 f"║ Suite: {report.suite.value:<20} ║",
448 f"║ Metric: {report.metric.value:<20} ║",
449 f"║ Cases: {report.total_cases:<20} ║",
450 f"║ Passed: {report.passed_cases} ({report.success_rate:.1%})",
451 f"║ Avg Score:{report.avg_score:.4f}",
452 f"║ Median: {report.median_score:.4f} ║",
453 f"║ Std Dev: {report.std_dev:.4f} ║",
454 f"║ Time: {report.duration_s:.1f}s",
455 "╚════════════════════════╝",
456 ]
457 if report.results and len(report.results) <= 20:
458 lines.append("\nPer-case scores:")
459 for r in report.results:
460 icon = "✓" if r.score >= 0.5 else "✗"
461 lines.append(f" {icon} {r.case_id[:40]:<42} {r.score:.3f}")
463 return "\n".join(lines)
466# ── Eval Registry ──────────────────────────────────────────────────
469class EvalRegistry:
470 """Registry for custom evaluation suites and scorers."""
472 def __init__(self):
473 self._suites: dict[str, list[EvalCase]] = {}
474 self._scorers: dict[str, Scorer] = {}
476 def register_suite(self, name: str, cases: list[EvalCase]) -> None:
477 self._suites[name] = cases
479 def register_scorer(self, name: str, scorer: Scorer) -> None:
480 self._scorers[name] = scorer
482 def get_suite(self, name: str) -> list[EvalCase]:
483 if name not in self._suites:
484 raise KeyError(f"Unknown eval suite: {name}")
485 return self._suites[name]
487 def get_scorer(self, name: str) -> Scorer:
488 return self._scorers.get(name, get_scorer(EvalMetric.EXACT_MATCH))
490 def list_suites(self) -> list[str]:
491 return list(self._suites.keys())
494# ── Quick Eval Helpers ─────────────────────────────────────────────
497def evaluate_quick(
498 generate_fn: Callable[[str], str],
499 cases: list[dict[str, str]],
500 metric: EvalMetric = EvalMetric.EXACT_MATCH,
501 k: int = 1,
502) -> EvalReport:
503 """Quick evaluation from a list of {prompt, expected} dicts."""
504 eval_cases = [
505 EvalCase(id=str(i), suite=EvalSuite.CUSTOM, prompt=c["prompt"], expected=c["expected"])
506 for i, c in enumerate(cases)
507 ]
508 runner = EvalRunner(generate_fn)
509 return runner.run(eval_cases, k=k, metric=metric)