Coverage for little_loops / ab_writer.py: 47%
47 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-08 15:34 -0500
1"""A/B baseline results aggregation and ab.json writer.
3Provides the ABResults dataclass, summary calculation, and JSON schema
4generation for the A/B baseline comparison feature (FEAT-1790).
5"""
7from __future__ import annotations
9import json
10import statistics
11from dataclasses import dataclass, field
12from pathlib import Path
13from typing import Any
15# ---------------------------------------------------------------------------
16# JSON Schema for ab.json (draft-07)
17# ---------------------------------------------------------------------------
19_AB_SCHEMA: dict[str, Any] = {
20 "$schema": "http://json-schema.org/draft-07/schema#",
21 "$id": "little-loops://ab-results.json",
22 "title": "A/B Baseline Comparison Results",
23 "description": "Per-item and summary results from blind A/B baseline comparison.",
24 "type": "object",
25 "required": ["summary", "items"],
26 "properties": {
27 "summary": {
28 "type": "object",
29 "required": [
30 "harness_pass_rate",
31 "baseline_pass_rate",
32 "delta",
33 "median_tokens_harness",
34 "median_tokens_baseline",
35 "median_duration_harness",
36 "median_duration_baseline",
37 ],
38 "properties": {
39 "harness_pass_rate": {
40 "type": "number",
41 "description": "Harness arm pass rate (0-1)",
42 },
43 "baseline_pass_rate": {
44 "type": "number",
45 "description": "Baseline arm pass rate (0-1)",
46 },
47 "delta": {
48 "type": "number",
49 "description": "Pass-rate delta (harness - baseline)",
50 },
51 "median_tokens_harness": {
52 "type": "integer",
53 "description": "Median token count for harness arm",
54 },
55 "median_tokens_baseline": {
56 "type": "integer",
57 "description": "Median token count for baseline arm",
58 },
59 "median_duration_harness": {
60 "type": "number",
61 "description": "Median duration (ms) for harness arm",
62 },
63 "median_duration_baseline": {
64 "type": "number",
65 "description": "Median duration (ms) for baseline arm",
66 },
67 },
68 "additionalProperties": False,
69 },
70 "items": {
71 "type": "array",
72 "description": "Per-item blind comparison records",
73 "items": {
74 "type": "object",
75 "required": [
76 "index",
77 "harness_pass",
78 "baseline_pass",
79 "harness_tokens",
80 "baseline_tokens",
81 "harness_duration_ms",
82 "baseline_duration_ms",
83 ],
84 "properties": {
85 "index": {"type": "integer", "description": "Zero-based item index"},
86 "harness_pass": {
87 "type": "boolean",
88 "description": "Whether harness arm passed evaluation",
89 },
90 "baseline_pass": {
91 "type": "boolean",
92 "description": "Whether baseline arm passed evaluation",
93 },
94 "harness_tokens": {
95 "type": "integer",
96 "description": "Token count for harness arm",
97 },
98 "baseline_tokens": {
99 "type": "integer",
100 "description": "Token count for baseline arm",
101 },
102 "harness_duration_ms": {
103 "type": "integer",
104 "description": "Duration (ms) for harness arm",
105 },
106 "baseline_duration_ms": {
107 "type": "integer",
108 "description": "Duration (ms) for baseline arm",
109 },
110 "confidence": {
111 "type": "number",
112 "description": "Judge confidence (0-1)",
113 },
114 "reason": {
115 "type": "string",
116 "description": "Judge reasoning for verdicts",
117 },
118 },
119 "additionalProperties": True,
120 },
121 },
122 },
123 "additionalProperties": False,
124}
126# ---------------------------------------------------------------------------
127# Data structures
128# ---------------------------------------------------------------------------
131@dataclass
132class ABResults:
133 """Aggregated A/B comparison results.
135 Attributes:
136 harness_pass_rate: Fraction of items where harness arm passed (0-1)
137 baseline_pass_rate: Fraction of items where baseline arm passed (0-1)
138 delta: Pass-rate difference (harness - baseline)
139 median_tokens_harness: Median token count for harness arm
140 median_tokens_baseline: Median token count for baseline arm
141 median_duration_harness: Median duration (ms) for harness arm
142 median_duration_baseline: Median duration (ms) for baseline arm
143 per_item: List of per-item comparison records
144 """
146 harness_pass_rate: float
147 baseline_pass_rate: float
148 delta: float
149 median_tokens_harness: int
150 median_tokens_baseline: int
151 median_duration_harness: float
152 median_duration_baseline: float
153 per_item: list[dict[str, Any]] = field(default_factory=list)
156# ---------------------------------------------------------------------------
157# Aggregation
158# ---------------------------------------------------------------------------
161def calculate_ab_summary(per_item_results: list[dict[str, Any]]) -> ABResults:
162 """Aggregate per-item verdicts into summary statistics.
164 Args:
165 per_item_results: List of per-item dicts, each with keys:
166 harness_pass, baseline_pass, harness_tokens, baseline_tokens,
167 harness_duration_ms, baseline_duration_ms
169 Returns:
170 ABResults with computed aggregate statistics
171 """
172 if not per_item_results:
173 return ABResults(
174 harness_pass_rate=0.0,
175 baseline_pass_rate=0.0,
176 delta=0.0,
177 median_tokens_harness=0,
178 median_tokens_baseline=0,
179 median_duration_harness=0.0,
180 median_duration_baseline=0.0,
181 per_item=[],
182 )
184 n = len(per_item_results)
185 harness_passes = sum(1 for item in per_item_results if item.get("harness_pass", False))
186 baseline_passes = sum(1 for item in per_item_results if item.get("baseline_pass", False))
188 harness_tokens = [item.get("harness_tokens", 0) for item in per_item_results]
189 baseline_tokens = [item.get("baseline_tokens", 0) for item in per_item_results]
190 harness_durations = [item.get("harness_duration_ms", 0) for item in per_item_results]
191 baseline_durations = [item.get("baseline_duration_ms", 0) for item in per_item_results]
193 return ABResults(
194 harness_pass_rate=harness_passes / n,
195 baseline_pass_rate=baseline_passes / n,
196 delta=(harness_passes - baseline_passes) / n,
197 median_tokens_harness=int(statistics.median(harness_tokens)),
198 median_tokens_baseline=int(statistics.median(baseline_tokens)),
199 median_duration_harness=float(statistics.median(harness_durations)),
200 median_duration_baseline=float(statistics.median(baseline_durations)),
201 per_item=per_item_results,
202 )
205# ---------------------------------------------------------------------------
206# Serialization
207# ---------------------------------------------------------------------------
210def ab_results_to_dict(results: ABResults) -> dict[str, Any]:
211 """Serialize ABResults to the ab.json wire format.
213 Args:
214 results: Aggregated A/B results
216 Returns:
217 Dict matching the ab.json schema
218 """
219 return {
220 "summary": {
221 "harness_pass_rate": results.harness_pass_rate,
222 "baseline_pass_rate": results.baseline_pass_rate,
223 "delta": results.delta,
224 "median_tokens_harness": results.median_tokens_harness,
225 "median_tokens_baseline": results.median_tokens_baseline,
226 "median_duration_harness": results.median_duration_harness,
227 "median_duration_baseline": results.median_duration_baseline,
228 },
229 "items": results.per_item,
230 }
233def write_ab_json(results: ABResults, run_dir: str) -> None:
234 """Write ab.json to the run directory.
236 Args:
237 results: Aggregated A/B results to serialize
238 run_dir: Target directory (created if missing)
239 """
240 Path(run_dir).mkdir(parents=True, exist_ok=True)
241 path = Path(run_dir) / "ab.json"
242 path.write_text(json.dumps(ab_results_to_dict(results), indent=2))
245def read_ab_json(run_dir: str) -> ABResults | None:
246 """Read ab.json from a run directory.
248 Args:
249 run_dir: Directory containing ab.json
251 Returns:
252 ABResults if file exists and is valid, None otherwise
253 """
254 path = Path(run_dir) / "ab.json"
255 if not path.exists():
256 return None
257 try:
258 data = json.loads(path.read_text())
259 except (json.JSONDecodeError, OSError):
260 return None
262 summary = data.get("summary", {})
263 items = data.get("items", [])
264 return ABResults(
265 harness_pass_rate=summary.get("harness_pass_rate", 0.0),
266 baseline_pass_rate=summary.get("baseline_pass_rate", 0.0),
267 delta=summary.get("delta", 0.0),
268 median_tokens_harness=summary.get("median_tokens_harness", 0),
269 median_tokens_baseline=summary.get("median_tokens_baseline", 0),
270 median_duration_harness=summary.get("median_duration_harness", 0.0),
271 median_duration_baseline=summary.get("median_duration_baseline", 0.0),
272 per_item=items,
273 )
276def get_ab_schema() -> dict[str, Any]:
277 """Return the ab.json JSON Schema (draft-07)."""
278 return _AB_SCHEMA