Coverage for agentos/evaluation/regression.py: 0%
150 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-08 12:20 +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
9import json
10import math
11import xml.etree.ElementTree as ET
12from dataclasses import dataclass, field
14from agentos.evaluation import EvalReport, Evaluator, ScoreDetail
17@dataclass
18class RegressionCheck:
19 """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."""
34 baseline_label: str = "baseline"
35 current_label: str = "current"
36 baseline: EvalReport = None
37 current: EvalReport = None
38 checks: list[RegressionCheck] = field(default_factory=list)
39 total_regressions: int = 0
40 total_improvements: int = 0
41 pass_delta: float = 0.0
42 score_delta: float = 0.0
43 verdict: str = "OK" # OK | WARN | FAIL
45 def to_markdown(self) -> str:
46 lines = [
47 f"# Regression Report: {self.baseline_label} → {self.current_label}",
48 "",
49 f"**Verdict**: `{self.verdict}`",
50 f"**Pass Rate**: {self.baseline.pass_rate:.1%} → {self.current.pass_rate:.1%} (Δ={self.pass_delta:+.1%})",
51 f"**Avg Score**: {self.baseline.avg_score:.1f} → {self.current.avg_score:.1f} (Δ={self.score_delta:+.1f})",
52 f"**Regressions**: {self.total_regressions} | **Improvements**: {self.total_improvements}",
53 "",
54 ]
56 if self.checks:
57 lines.append("## Detail")
58 lines.append("| Case ID | Baseline | Current | Δ | Verdict |")
59 lines.append("|---------|----------|---------|---|---------|")
60 for c in self.checks:
61 icon = "RECRESSION" if c.regression else "IMPROVED" if c.delta > 0 else "SAME"
62 lines.append(
63 f"| {c.case_id} | {c.baseline_score:.1f} | {c.current_score:.1f} | "
64 f"{c.delta:+.1f} | {icon} |"
65 )
67 return "\n".join(lines)
70class RegressionRunner:
71 """Detect regressions by comparing baseline and current evaluation runs.
73 Usage:
74 runner = RegressionRunner(evaluator, baseline=report)
75 report = await runner.check(current_report)
76 # or sync:
77 report = runner.check_sync(current_report)
78 """
80 def __init__(
81 self,
82 evaluator: Evaluator,
83 baseline: EvalReport | None = None,
84 threshold: float = 5.0,
85 severe_threshold: float = 20.0,
86 ):
87 self.evaluator = evaluator
88 self.baseline = baseline
89 self.threshold = threshold
90 self.severe_threshold = severe_threshold
92 async def run_baseline(self) -> EvalReport:
93 """Run and store the baseline."""
94 self.baseline = await self.evaluator.run()
95 return self.baseline
97 async def check(self, current: EvalReport | None = None) -> RegressionReport:
98 """Compare current against baseline. If current not given, run it."""
99 if current is None:
100 current = await self.evaluator.run()
101 return self._compare(current)
103 def check_sync(self, current: EvalReport) -> RegressionReport:
104 """Synchronous version for testing."""
105 return self._compare(current)
107 def _compare(self, current: EvalReport) -> RegressionReport:
109 if self.baseline is None:
110 raise ValueError("No baseline set. Call run_baseline() first.")
112 report = RegressionReport(
113 baseline=self.baseline,
114 current=current,
115 pass_delta=current.pass_rate - self.baseline.pass_rate,
116 score_delta=current.avg_score - self.baseline.avg_score,
117 )
119 # Build lookup from baseline results
120 baseline_map: dict[str, ScoreDetail] = {r.case_id: r for r in self.baseline.results}
122 for current_result in current.results:
123 cid = current_result.case_id
124 baseline_result = baseline_map.get(cid)
126 if baseline_result is None:
127 # New case, no baseline comparison
128 report.checks.append(
129 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 )
137 continue
139 delta = current_result.total_score - baseline_result.total_score
140 regression = delta < -self.threshold
142 severity = "none"
143 if delta < -self.severe_threshold:
144 severity = "severe"
145 elif delta < -self.threshold:
146 severity = "moderate"
147 elif delta < 0:
148 severity = "minor"
150 if regression:
151 report.total_regressions += 1
152 elif delta > self.threshold:
153 report.total_improvements += 1
155 report.checks.append(
156 RegressionCheck(
157 case_id=cid,
158 baseline_score=baseline_result.total_score,
159 current_score=current_result.total_score,
160 delta=round(delta, 1),
161 regression=regression,
162 severity=severity,
163 )
164 )
166 # Verdict
167 if report.total_regressions > 0:
168 has_severe = any(c.severity == "severe" for c in report.checks)
169 report.verdict = "FAIL" if has_severe else "WARN"
171 return report
174# --- Statistical Runner ---
177@dataclass
178class StatResult:
179 """Statistical summary of N evaluation runs."""
181 trials: int = 0
182 pass_rates: list[float] = field(default_factory=list)
183 avg_scores: list[float] = field(default_factory=list)
184 mean_pass_rate: float = 0.0
185 std_pass_rate: float = 0.0
186 mean_score: float = 0.0
187 std_score: float = 0.0
188 ci95_pass_rate: tuple[float, float] = (0.0, 0.0)
189 ci95_score: tuple[float, float] = (0.0, 0.0)
191 def to_dict(self) -> dict:
192 return {
193 "trials": self.trials,
194 "mean_pass_rate": round(self.mean_pass_rate, 4),
195 "std_pass_rate": round(self.std_pass_rate, 4),
196 "ci95_pass_rate": [round(x, 4) for x in self.ci95_pass_rate],
197 "mean_score": round(self.mean_score, 2),
198 "std_score": round(self.std_score, 2),
199 "ci95_score": [round(x, 2) for x in self.ci95_score],
200 }
203class StatisticalRunner:
204 """Run evaluation N times and compute statistics.
206 Usage:
207 srunner = StatisticalRunner(evaluator, trials=10)
208 stats = await srunner.run()
209 """
211 def __init__(self, evaluator: Evaluator, trials: int = 5):
212 self.evaluator = evaluator
213 self.trials = max(trials, 2)
215 async def run(self) -> StatResult:
216 pass_rates = []
217 avg_scores = []
219 for i in range(self.trials):
220 report = await self.evaluator.run()
221 pass_rates.append(report.pass_rate)
222 avg_scores.append(report.avg_score)
224 return self._compute(pass_rates, avg_scores)
226 def _compute(self, pass_rates: list[float], avg_scores: list[float]) -> StatResult:
227 n = len(pass_rates)
228 mean_pr = sum(pass_rates) / n
229 var_pr = sum((x - mean_pr) ** 2 for x in pass_rates) / (n - 1)
230 std_pr = math.sqrt(var_pr) if var_pr > 0 else 0
232 mean_s = sum(avg_scores) / n
233 var_s = sum((x - mean_s) ** 2 for x in avg_scores) / (n - 1)
234 std_s = math.sqrt(var_s) if var_s > 0 else 0
236 # 95% CI using t-distribution approximation (z for simplicity)
237 z = 1.96
238 ci_pr = (
239 max(0, mean_pr - z * std_pr / math.sqrt(n)),
240 min(1, mean_pr + z * std_pr / math.sqrt(n)),
241 )
242 ci_s = (
243 mean_s - z * std_s / math.sqrt(n),
244 mean_s + z * std_s / math.sqrt(n),
245 )
247 return StatResult(
248 trials=n,
249 pass_rates=pass_rates,
250 avg_scores=avg_scores,
251 mean_pass_rate=round(mean_pr, 4),
252 std_pass_rate=round(std_pr, 4),
253 mean_score=round(mean_s, 2),
254 std_score=round(std_s, 2),
255 ci95_pass_rate=(round(ci_pr[0], 4), round(ci_pr[1], 4)),
256 ci95_score=(round(ci_s[0], 2), round(ci_s[1], 2)),
257 )
260# --- CI Exports ---
263def to_junit_xml(report: EvalReport, suite_name: str = "AgentOS Eval") -> str:
264 """Convert evaluation report to JUnit XML for CI integration (GitHub Actions, Jenkins, etc.)."""
265 total = len(report.results)
266 failed_count = sum(1 for r in report.results if not r.passed)
267 total_time = sum(r.duration_ms for r in report.results) / 1000
269 suite = ET.Element(
270 "testsuite",
271 {
272 "name": suite_name,
273 "tests": str(total),
274 "failures": str(failed_count),
275 "errors": "0",
276 "skipped": "0",
277 "time": f"{total_time:.3f}",
278 "timestamp": report.timestamp or "",
279 },
280 )
282 for result in report.results:
283 testcase = ET.SubElement(
284 suite,
285 "testcase",
286 {
287 "classname": f"AgentOS.{report.dataset_name}",
288 "name": result.case_id,
289 "time": f"{result.duration_ms / 1000:.3f}",
290 },
291 )
293 if not result.passed:
294 failure = ET.SubElement(
295 testcase,
296 "failure",
297 {
298 "type": "ScoreBelowThreshold",
299 "message": f"Score {result.total_score:.1f}: {result.details}",
300 },
301 )
302 failure.text = f"Actual: {result.actual_output[:500]}\nErrors: {result.errors}"
304 # Add score as property
305 props = ET.SubElement(testcase, "properties")
306 ET.SubElement(props, "property", {"name": "score", "value": f"{result.total_score:.1f}"})
307 ET.SubElement(props, "property", {"name": "metrics", "value": json.dumps(result.metrics)})
309 return ET.tostring(suite, encoding="unicode")
312def to_json(report: EvalReport, indent: int = 2) -> str:
313 """Serialize EvalReport to JSON string."""
314 import dataclasses
316 class ReportEncoder(json.JSONEncoder):
317 def default(self, obj):
318 if dataclasses.is_dataclass(obj):
319 return dataclasses.asdict(obj)
320 return super().default(obj)
322 return json.dumps(report, cls=ReportEncoder, indent=indent, ensure_ascii=False)
325def save_report(report: EvalReport, path: str, format: str = "markdown"):
326 """Save evaluation report to file (markdown / json / junit)."""
327 if format == "markdown":
328 content = report.to_markdown()
329 elif format == "json":
330 content = to_json(report)
331 elif format == "junit":
332 content = to_junit_xml(report)
333 else:
334 raise ValueError(f"Unknown format: {format}")
336 with open(path, "w", encoding="utf-8") as f:
337 f.write(content)