782916a283
Give EvalRun.summary a typed RunSummary value (unified RunError, lenient legacy parsing) so readers stop reaching into a schemaless dict, and route every cross-run rollup — dashboard, scenario ranking, trend, campaign report — through one aggregate_runs seam. Fixes the divergence where stats averaged pass_rate over completed-only runs while the campaign report counted faults as 0.0. Cross-run rule (ADR-0004): genuine faults count 0.0, user-cancelled runs are excluded from both denominators.
45 lines
1.8 KiB
Python
45 lines
1.8 KiB
Python
"""The single cross-run aggregation seam (ADR-0004).
|
|
|
|
Every reader that rolls Runs up into pass_rate / availability / latency —
|
|
dashboard, scenario ranking, trend, campaign report — calls ``aggregate_runs``.
|
|
No caller may read ``summary["pass_rate"]`` and re-aggregate on its own.
|
|
|
|
Rules (ADR-0004, extending ADR-0002's service perspective across runs):
|
|
- a genuinely faulted run counts 0.0 in both pass_rate and availability;
|
|
- a user-cancelled run (``summary.error.code == "cancelled_by_user"``) is
|
|
excluded from both denominators — cancellation is a user action, not a
|
|
quality or availability signal of the target;
|
|
- ``run_count`` still reports everything that happened, cancelled included.
|
|
"""
|
|
|
|
from typing import Any
|
|
|
|
from agenteval.models import EvalRun, RunStatus
|
|
|
|
|
|
def aggregate_runs(runs: list[EvalRun]) -> dict[str, Any]:
|
|
scored = [r for r in runs if not (r.summary is not None and r.summary.is_cancelled)]
|
|
n = len(scored)
|
|
if n == 0:
|
|
return {"run_count": len(runs), "pass_rate": None, "availability": None, "avg_latency_ms": None}
|
|
|
|
completed = [r for r in scored if r.status == RunStatus.COMPLETED]
|
|
pass_rates = [_completed_pass_rate(r) if r.status == RunStatus.COMPLETED else 0.0 for r in scored]
|
|
latencies = [
|
|
r.summary.avg_latency_ms
|
|
for r in completed
|
|
if r.summary is not None and r.summary.avg_latency_ms is not None
|
|
]
|
|
return {
|
|
"run_count": len(runs),
|
|
"pass_rate": round(sum(pass_rates) / n, 4),
|
|
"availability": round(len(completed) / n, 4),
|
|
"avg_latency_ms": round(sum(latencies) / len(latencies), 1) if latencies else None,
|
|
}
|
|
|
|
|
|
def _completed_pass_rate(run: EvalRun) -> float:
|
|
if run.summary is None or run.summary.pass_rate is None:
|
|
return 0.0
|
|
return run.summary.pass_rate
|