Files
imbad0202__academic-researc…/scripts/_eval_threshold_gate.py
T
Edward Cheng-I Wu 501366f197 fix(eval-harness): gate binding per-class thresholds + drop pending tasks from lift baseline (#328) (#336)
The absolute-threshold CI gate inspected only aggregate_metric.passed, so a PR
that regressed a binding per-class threshold (e.g. citation_extraction.false.accuracy
below 0.85 while aggregate stayed >= 0.90, or rq_framing_patterns.fnr above 0.30
while balanced_accuracy passed) passed the gate when it should block. run_evals
stamps per_class[].passed against those declared thresholds; failed_tasks() now
iterates per_class too, keyed <task>.<class>.<metric> to match the lift gate.

check_ranking_lift._flatten_report let a pending task's placeholder
aggregate_metric.value: 0.0 (from _pending_result) enter the lift baseline, so
once the task landed its real value read as a zero-baseline change spuriously
needing [ranking-regression-acknowledged]. It now skips non-measured tasks.

Both consumers use the same positive skip-unless-measured guard
(task.get("status", "measured") != "measured") so a future non-measured status
is excluded consistently rather than re-opening the baseline-pollution gap.

Tests: new scripts/test__eval_threshold_gate.py (9 gate cases incl. the
per-class-fail-aggregate-pass regression + symmetry cases) wired into the CI
pytest manifest; 5 _flatten_report status-filter tests in test_check_ranking_lift.py.

Closes #328

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-05 21:40:40 +08:00

74 lines
3.0 KiB
Python

#!/usr/bin/env python3
"""Absolute-threshold verdict for the eval harness (#184 Delta 3 / Fix 1).
Phase 1b has no ``main`` baseline, so the CI gate is an ABSOLUTE-threshold check,
not a lift comparison. This module reads a ``run_evals`` report and returns the
list of measured tasks whose aggregate metric failed its declared threshold.
Kept out of the workflow YAML (no inline heredoc) so the gate's only load-bearing
logic is unit-testable rather than asserted via brittle YAML string matching.
CLI::
python -m scripts._eval_threshold_gate <report.json>
prints a comma-separated list of failed threshold keys —
``<task>.aggregate.<metric>`` and ``<task>.<class>.<metric>`` (empty line if
none) — to stdout.
"""
from __future__ import annotations
import json
import sys
from typing import Any
def failed_tasks(report: dict[str, Any]) -> list[str]:
"""Return one key per measured threshold that failed — aggregate AND per-class.
Only tasks with ``status == "measured"`` that declare a threshold (so the
measurer set ``.passed``) are gated. ``passed is False`` is the failure signal
— a metric without a threshold (``passed`` absent) is not gated, and a
pending/skipped task is never gated.
Both axes are binding: manifests declare aggregate AND per-class thresholds
(e.g. citation_extraction: aggregate ``accuracy >= 0.90`` plus per_class
``accuracy >= 0.85``; rq_framing_patterns: ``balanced_accuracy >= 0.75`` plus
``fnr <= 0.30`` / ``fpr <= 0.20`` as per_class rows). run_evals stamps
``per_class[].passed`` against the per-class threshold, so the gate must honour
it too — else a PR that regresses only a per-class metric passes the gate when
it should block (#328). Aggregate failures use ``<task>.aggregate.<metric>``;
per-class failures use ``<task>.<class_name>.<metric>`` (matching the lift
gate's per-class key shape).
"""
failures: list[str] = []
for task in report.get("per_task", []):
# Skip-unless-measured (symmetric with check_ranking_lift._flatten_report).
if task.get("status", "measured") != "measured":
continue
agg = task.get("aggregate_metric") or {}
if agg.get("passed") is False:
failures.append(f"{task['task_name']}.aggregate.{agg.get('metric', '?')}")
for pc in task.get("per_class", []):
if pc.get("passed") is False:
failures.append(
f"{task['task_name']}.{pc.get('class_name', '?')}."
f"{pc.get('metric', '?')}"
)
return failures
def main(argv: list[str] | None = None) -> int:
args = sys.argv[1:] if argv is None else argv
if len(args) != 1:
print("usage: python -m scripts._eval_threshold_gate <report.json>",
file=sys.stderr)
return 2
report = json.loads(open(args[0], encoding="utf-8").read())
print(",".join(failed_tasks(report)))
return 0
if __name__ == "__main__":
sys.exit(main())