Coverage for agentos/evaluation/regression.py: 36%
151 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 10:59 +0800
1"""Evaluation regression testing for AgentOS.
3Compare evaluation runs, detect regressions, generate CI artifacts.
4Builds on top of agentos.evaluation core (GoldenDataset, Evaluator, EvalReport).
5"""
7from __future__ import annotations
9from dataclasses import dataclass, field
10from typing import Dict, List, Optional, Tuple
11import json
12import math
13import xml.etree.ElementTree as ET
15from agentos.evaluation import Evaluator, EvalReport, ScoreDetail
18@dataclass
19class RegressionCheck:
20 """A single regression check result."""
21 case_id: str
22 baseline_score: float
23 current_score: float
24 delta: float
25 regression: bool = False
26 severity: str = "none" # none | minor | moderate | severe
27 details: str = ""
30@dataclass
31class RegressionReport:
32 """Comparison report between baseline and current evaluation."""
33 baseline_label: str = "baseline"
34 current_label: str = "current"
35 baseline: EvalReport = None
36 current: EvalReport = None
37 checks: List[RegressionCheck] = field(default_factory=list)
38 total_regressions: int = 0
39 total_improvements: int = 0
40 pass_delta: float = 0.0
41 score_delta: float = 0.0
42 verdict: str = "OK" # OK | WARN | FAIL
44 def to_markdown(self) -> str:
45 lines = [
46 f"# Regression Report: {self.baseline_label} → {self.current_label}",
47 "",
48 f"**Verdict**: `{self.verdict}`",
49 f"**Pass Rate**: {self.baseline.pass_rate:.1%} → {self.current.pass_rate:.1%} (Δ={self.pass_delta:+.1%})",
50 f"**Avg Score**: {self.baseline.avg_score:.1f} → {self.current.avg_score:.1f} (Δ={self.score_delta:+.1f})",
51 f"**Regressions**: {self.total_regressions} | **Improvements**: {self.total_improvements}",
52 "",
53 ]
55 if self.checks:
56 lines.append("## Detail")
57 lines.append("| Case ID | Baseline | Current | Δ | Verdict |")
58 lines.append("|---------|----------|---------|---|---------|")
59 for c in self.checks:
60 icon = "RECRESSION" if c.regression else "IMPROVED" if c.delta > 0 else "SAME"
61 lines.append(
62 f"| {c.case_id} | {c.baseline_score:.1f} | {c.current_score:.1f} | "
63 f"{c.delta:+.1f} | {icon} |"
64 )
66 return "\n".join(lines)
69class RegressionRunner:
70 """Detect regressions by comparing baseline and current evaluation runs.
72 Usage:
73 runner = RegressionRunner(evaluator, baseline=report)
74 report = await runner.check(current_report)
75 # or sync:
76 report = runner.check_sync(current_report)
77 """
79 def __init__(
80 self,
81 evaluator: Evaluator,
82 baseline: Optional[EvalReport] = None,
83 threshold: float = 5.0,
84 severe_threshold: float = 20.0,
85 ):
86 self.evaluator = evaluator
87 self.baseline = baseline
88 self.threshold = threshold
89 self.severe_threshold = severe_threshold
91 async def run_baseline(self) -> EvalReport:
92 """Run and store the baseline."""
93 self.baseline = await self.evaluator.run()
94 return self.baseline
96 async def check(self, current: Optional[EvalReport] = None) -> RegressionReport:
97 """Compare current against baseline. If current not given, run it."""
98 if current is None:
99 current = await self.evaluator.run()
100 return self._compare(current)
102 def check_sync(self, current: EvalReport) -> RegressionReport:
103 """Synchronous version for testing."""
104 return self._compare(current)
106 def _compare(self, current: EvalReport) -> RegressionReport:
108 if self.baseline is None:
109 raise ValueError("No baseline set. Call run_baseline() first.")
111 report = RegressionReport(
112 baseline=self.baseline,
113 current=current,
114 pass_delta=current.pass_rate - self.baseline.pass_rate,
115 score_delta=current.avg_score - self.baseline.avg_score,
116 )
118 # Build lookup from baseline results
119 baseline_map: Dict[str, ScoreDetail] = {
120 r.case_id: r for r in self.baseline.results
121 }
123 for current_result in current.results:
124 cid = current_result.case_id
125 baseline_result = baseline_map.get(cid)
127 if baseline_result is None:
128 # New case, no baseline comparison
129 report.checks.append(RegressionCheck(
130 case_id=cid,
131 baseline_score=0,
132 current_score=current_result.total_score,
133 delta=0,
134 details="new case",
135 ))
136 continue
138 delta = current_result.total_score - baseline_result.total_score
139 regression = delta < -self.threshold
141 severity = "none"
142 if delta < -self.severe_threshold:
143 severity = "severe"
144 elif delta < -self.threshold:
145 severity = "moderate"
146 elif delta < 0:
147 severity = "minor"
149 if regression:
150 report.total_regressions += 1
151 elif delta > self.threshold:
152 report.total_improvements += 1
154 report.checks.append(RegressionCheck(
155 case_id=cid,
156 baseline_score=baseline_result.total_score,
157 current_score=current_result.total_score,
158 delta=round(delta, 1),
159 regression=regression,
160 severity=severity,
161 ))
163 # Verdict
164 if report.total_regressions > 0:
165 has_severe = any(c.severity == "severe" for c in report.checks)
166 report.verdict = "FAIL" if has_severe else "WARN"
168 return report
171# --- Statistical Runner ---
174@dataclass
175class StatResult:
176 """Statistical summary of N evaluation runs."""
177 trials: int = 0
178 pass_rates: List[float] = field(default_factory=list)
179 avg_scores: List[float] = field(default_factory=list)
180 mean_pass_rate: float = 0.0
181 std_pass_rate: float = 0.0
182 mean_score: float = 0.0
183 std_score: float = 0.0
184 ci95_pass_rate: Tuple[float, float] = (0.0, 0.0)
185 ci95_score: Tuple[float, float] = (0.0, 0.0)
187 def to_dict(self) -> dict:
188 return {
189 "trials": self.trials,
190 "mean_pass_rate": round(self.mean_pass_rate, 4),
191 "std_pass_rate": round(self.std_pass_rate, 4),
192 "ci95_pass_rate": [round(x, 4) for x in self.ci95_pass_rate],
193 "mean_score": round(self.mean_score, 2),
194 "std_score": round(self.std_score, 2),
195 "ci95_score": [round(x, 2) for x in self.ci95_score],
196 }
199class StatisticalRunner:
200 """Run evaluation N times and compute statistics.
202 Usage:
203 srunner = StatisticalRunner(evaluator, trials=10)
204 stats = await srunner.run()
205 """
207 def __init__(self, evaluator: Evaluator, trials: int = 5):
208 self.evaluator = evaluator
209 self.trials = max(trials, 2)
211 async def run(self) -> StatResult:
212 pass_rates = []
213 avg_scores = []
215 for i in range(self.trials):
216 report = await self.evaluator.run()
217 pass_rates.append(report.pass_rate)
218 avg_scores.append(report.avg_score)
220 return self._compute(pass_rates, avg_scores)
222 def _compute(self, pass_rates: List[float], avg_scores: List[float]) -> StatResult:
223 n = len(pass_rates)
224 mean_pr = sum(pass_rates) / n
225 var_pr = sum((x - mean_pr) ** 2 for x in pass_rates) / (n - 1)
226 std_pr = math.sqrt(var_pr) if var_pr > 0 else 0
228 mean_s = sum(avg_scores) / n
229 var_s = sum((x - mean_s) ** 2 for x in avg_scores) / (n - 1)
230 std_s = math.sqrt(var_s) if var_s > 0 else 0
232 # 95% CI using t-distribution approximation (z for simplicity)
233 z = 1.96
234 ci_pr = (
235 max(0, mean_pr - z * std_pr / math.sqrt(n)),
236 min(1, mean_pr + z * std_pr / math.sqrt(n)),
237 )
238 ci_s = (
239 mean_s - z * std_s / math.sqrt(n),
240 mean_s + z * std_s / math.sqrt(n),
241 )
243 return StatResult(
244 trials=n,
245 pass_rates=pass_rates,
246 avg_scores=avg_scores,
247 mean_pass_rate=round(mean_pr, 4),
248 std_pass_rate=round(std_pr, 4),
249 mean_score=round(mean_s, 2),
250 std_score=round(std_s, 2),
251 ci95_pass_rate=(round(ci_pr[0], 4), round(ci_pr[1], 4)),
252 ci95_score=(round(ci_s[0], 2), round(ci_s[1], 2)),
253 )
256# --- CI Exports ---
259def to_junit_xml(report: EvalReport, suite_name: str = "AgentOS Eval") -> str:
260 """Convert evaluation report to JUnit XML for CI integration (GitHub Actions, Jenkins, etc.)."""
261 total = len(report.results)
262 failed_count = sum(1 for r in report.results if not r.passed)
263 total_time = sum(r.duration_ms for r in report.results) / 1000
265 suite = ET.Element("testsuite", {
266 "name": suite_name,
267 "tests": str(total),
268 "failures": str(failed_count),
269 "errors": "0",
270 "skipped": "0",
271 "time": f"{total_time:.3f}",
272 "timestamp": report.timestamp or "",
273 })
275 for result in report.results:
276 testcase = ET.SubElement(suite, "testcase", {
277 "classname": f"AgentOS.{report.dataset_name}",
278 "name": result.case_id,
279 "time": f"{result.duration_ms / 1000:.3f}",
280 })
282 if not result.passed:
283 failure = ET.SubElement(testcase, "failure", {
284 "type": "ScoreBelowThreshold",
285 "message": f"Score {result.total_score:.1f}: {result.details}",
286 })
287 failure.text = f"Actual: {result.actual_output[:500]}\nErrors: {result.errors}"
289 # Add score as property
290 props = ET.SubElement(testcase, "properties")
291 ET.SubElement(props, "property", {
292 "name": "score", "value": f"{result.total_score:.1f}"
293 })
294 ET.SubElement(props, "property", {
295 "name": "metrics", "value": json.dumps(result.metrics)
296 })
298 return ET.tostring(suite, encoding="unicode")
301def to_json(report: EvalReport, indent: int = 2) -> str:
302 """Serialize EvalReport to JSON string."""
303 import dataclasses
305 class ReportEncoder(json.JSONEncoder):
306 def default(self, obj):
307 if dataclasses.is_dataclass(obj):
308 return dataclasses.asdict(obj)
309 return super().default(obj)
311 return json.dumps(report, cls=ReportEncoder, indent=indent, ensure_ascii=False)
314def save_report(report: EvalReport, path: str, format: str = "markdown"):
315 """Save evaluation report to file (markdown / json / junit)."""
316 if format == "markdown":
317 content = report.to_markdown()
318 elif format == "json":
319 content = to_json(report)
320 elif format == "junit":
321 content = to_junit_xml(report)
322 else:
323 raise ValueError(f"Unknown format: {format}")
325 with open(path, "w", encoding="utf-8") as f:
326 f.write(content)