diff --git a/src/agenteval/flaky.py b/src/agenteval/flaky.py index 2f96c07..554ea2b 100644 --- a/src/agenteval/flaky.py +++ b/src/agenteval/flaky.py @@ -7,7 +7,7 @@ import math from dataclasses import dataclass, field -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Sequence from agenteval.models import EvalResult @@ -90,6 +90,42 @@ def aggregate_multi_run( ) +def pass_hat_k(passed: int, total: int, k: int) -> float: + """pass^k — the unbiased probability that **all k** of k trials pass. + + The Tau-bench reliability estimator: sample k of the ``total`` observed + trials (which had ``passed`` successes) without replacement and ask that all + k succeed → ``C(passed, k) / C(total, k)``. Reliability under repetition, the + complement of pass@k (at-least-one). ``k=1`` ⇒ the plain pass rate. + + Returns ``0.0`` when there are fewer than k passes, and ``nan`` when there + aren't k trials to draw from (insufficient data — don't fabricate a number). + """ + if k < 1: + raise ValueError("k must be >= 1") + if total < k: + return float("nan") # insufficient trials to estimate + if passed < k: + return 0.0 + return math.comb(passed, k) / math.comb(total, k) + + +def reliability_at_k(result: MultiRunResult, k: int) -> float: + """pass^k for one case's multi-run result.""" + return pass_hat_k(result.passed_count, result.runs, k) + + +def reliability_summary(cases: List[MultiRunResult], ks: Sequence[int]) -> Dict[str, float]: + """Mean pass^k across cases for each k (cases with too few trials skipped).""" + out: Dict[str, float] = {} + for k in ks: + vals = [reliability_at_k(c, k) for c in cases] + finite = [v for v in vals if not math.isnan(v)] + if finite: + out[f"pass^{k}"] = sum(finite) / len(finite) + return out + + def should_quarantine( result: MultiRunResult, config: Optional[QuarantineConfig] = None, @@ -157,5 +193,7 @@ def build_multi_run_report( "flaky_cases": flaky_count, "quarantined_cases": quarantined_count, "stable_cases": len(cases) - flaky_count, + # pass^k reliability (#13): pass^2 + pass^(all runs), when enough trials. + **reliability_summary(cases, sorted({k for k in (2, num_runs) if k >= 2})), }, ) diff --git a/tests/test_pass_hat_k.py b/tests/test_pass_hat_k.py new file mode 100644 index 0000000..208c346 --- /dev/null +++ b/tests/test_pass_hat_k.py @@ -0,0 +1,78 @@ +"""pass^k multi-trial reliability metric (#13).""" + +import math + +import pytest + +from agenteval.flaky import ( + MultiRunResult, + build_multi_run_report, + pass_hat_k, + reliability_at_k, + reliability_summary, +) +from agenteval.models import EvalResult + + +def test_k1_is_pass_rate(): + assert pass_hat_k(3, 10, 1) == 0.3 + + +def test_all_pass_is_one(): + assert pass_hat_k(5, 5, 3) == 1.0 + + +def test_unbiased_estimator_without_replacement(): + # 4 passes of 5 trials, k=2: C(4,2)/C(5,2) = 6/10 + assert pass_hat_k(4, 5, 2) == pytest.approx(0.6) + # k=3: C(4,3)/C(5,3) = 4/10 + assert pass_hat_k(4, 5, 3) == pytest.approx(0.4) + + +def test_fewer_than_k_passes_is_zero(): + assert pass_hat_k(1, 5, 2) == 0.0 + assert pass_hat_k(0, 5, 1) == 0.0 + + +def test_insufficient_trials_is_nan(): + assert math.isnan(pass_hat_k(2, 2, 3)) # can't draw 3 from 2 + + +def test_k_must_be_positive(): + with pytest.raises(ValueError): + pass_hat_k(3, 5, 0) + + +def test_complement_of_pass_at_k_intuition(): + # All-pass ⇒ pass^k == 1 for any k≤n; a single failure drops pass^n to 0. + assert pass_hat_k(5, 5, 5) == 1.0 + assert pass_hat_k(4, 5, 5) == 0.0 # not enough passes for all-5 + + +def _mr(name, passed, runs): + return MultiRunResult( + case_name=name, runs=runs, passed_count=passed, pass_rate=passed / runs, + mean_score=0.0, stddev_score=0.0, scores=[], consistency_score=0.0, is_flaky=0 < passed < runs, + ) + + +def test_reliability_at_k_and_summary(): + a = _mr("a", 4, 4) # always passes + b = _mr("b", 2, 4) # flaky: pass^2 = C(2,2)/C(4,2) = 1/6 + assert reliability_at_k(a, 2) == 1.0 + assert reliability_at_k(b, 2) == pytest.approx(1 / 6) + summary = reliability_summary([a, b], [2]) + assert summary["pass^2"] == pytest.approx((1.0 + 1 / 6) / 2) + + +def test_report_summary_includes_pass_hat_k(): + def res(passed): + return EvalResult( + case_name="c", passed=passed, score=1.0 if passed else 0.0, details={}, + agent_output="", tools_called=[], tokens_in=0, tokens_out=0, cost_usd=None, latency_ms=0, + ) + + report = build_multi_run_report({"c": [res(True), res(True), res(False)]}, num_runs=3) + # 2 of 3 passed → pass^2 = C(2,2)/C(3,2) = 1/3; pass^3 = 0 (only 2 passes) + assert report.summary["pass^2"] == pytest.approx(1 / 3) + assert report.summary["pass^3"] == 0.0