| """Amendment 6: useful acceptance gain and paired family non-inferiority.""" |
| from __future__ import annotations |
|
|
| import argparse |
| from decimal import Decimal |
| import json |
| import math |
| from pathlib import Path |
| from typing import TypedDict, cast |
|
|
| from .data import D2Blocks, sha256 |
| from .pilot import ARMS, Spec, digest, read_spec, validate_inputs, write_json, verify_units |
| from .pilot_statistics import CLUSTER_KEY, FEW_CLUSTERS, DEV_SPLITS, Interval, paired_interval, split_roles |
| from .pilot_eval import EVALUATOR, SPLITS, Evaluation, Metrics, read_arm, validate_sample_evaluation |
| from .pilot_sample import validate_sample |
|
|
| RULE_VERSION = "amendment-6ab-source-prompt-bootstrap-v2" |
| RULE = "Amendments 6/6a/6b: dev E[A] excluding bonus; source_docs components or empty-source prompt_sha256 clusters, then requests; bootstrap 2000 seed 20261024, 90% CI; task labels are strata; REF/FAST gain >=0.05 with lower>0; FAST-REF lower>=-0.03; FAST baseline deficit<=0.01; median per-position KL deficit<=0.01; definitive failures => FAIL; extra seed only if every failing check straddles; REF no useful gain => INCONCLUSIVE" |
|
|
|
|
| def uncertainty(evaluation: Evaluation, points: list[int]) -> dict[str, Interval]: |
| DEV_SPLITS = split_roles(evaluation["baseline"]) |
| SPLITS = tuple(DEV_SPLITS) |
| result: dict[str, Interval] = {} |
| for point in points: |
| for split in SPLITS: |
| base = evaluation["baseline"][split]["request_totals"] |
| ref = evaluation["arms"]["REF"][str(point)][split]["request_totals"] |
| fast = evaluation["arms"]["FAST"][str(point)][split]["request_totals"] |
| prefix = f"{point}/{DEV_SPLITS[split]}" |
| result[prefix+"/REF_gain"] = paired_interval(ref, base) |
| result[prefix+"/FAST_gain"] = paired_interval(fast, base) |
| result[prefix+"/FAST_minus_REF"] = paired_interval(fast, ref) |
| return result |
|
|
|
|
| def status(checks: dict[str, bool]) -> str: |
| if any(not value for key, value in checks.items() if key.endswith("REF_useful_gain")): |
| return "INCONCLUSIVE" |
| return "PASS" if all(checks.values()) else "FAIL" |
|
|
|
|
| def population_report(evaluation: Evaluation) -> dict[str, object]: |
| """Report task strata separately from the sampled source/prompt clusters.""" |
| def strata(metrics: Metrics) -> dict[str, dict[str, float | int]]: |
| groups: dict[str, list[tuple[int, float]]] = {} |
| for row in metrics["request_totals"]: |
| groups.setdefault(row["task_label"], []).append((row["rounds"], row["accepted_sum"])) |
| return {task: {"requests": len(rows), "rounds": sum(n for n, _ in rows), |
| "accepted_length": math.fsum(a for _, a in rows)/sum(n for n, _ in rows)} |
| for task, rows in sorted(groups.items())} |
| DEV_SPLITS = split_roles(evaluation["baseline"]) |
| counts = {DEV_SPLITS[s]: len({r["family"] for r in m["request_totals"]}) |
| for s, m in evaluation["baseline"].items()} |
| return {"cluster_key": CLUSTER_KEY, "clusters_per_split": counts, |
| "caveats": {s: FEW_CLUSTERS for s, n in counts.items() if n < 5}, |
| "task_strata": {"baseline": {DEV_SPLITS[s]: strata(m) for s, m in evaluation["baseline"].items()}, |
| "arms": {a: {p: {DEV_SPLITS[s]: strata(m) for s, m in splits.items()} |
| for p, splits in points.items()} for a, points in evaluation["arms"].items()}}} |
|
|
|
|
|
|
| class Receipt(TypedDict): |
| schema: str |
| passed: bool |
| verdict: str |
| intervals: dict[str, Interval] |
| extra_seed_eligible: bool |
| synthetic: bool |
| rule: str |
| rule_version: str |
| plan: Spec |
| plan_sha256: str |
| evaluation: Evaluation |
| evaluation_sha256: str |
| checks: dict[str, bool] |
| evidence_sha256: str |
| sample_sha256: str |
| population: dict[str, object] |
|
|
|
|
| def finite(value: object) -> None: |
| if isinstance(value, dict): |
| for item in value.values(): |
| finite(item) |
| elif isinstance(value, list): |
| for item in value: |
| finite(item) |
| elif isinstance(value, (float, Decimal)) and not math.isfinite(value): |
| raise ValueError("nonfinite pilot evidence") |
|
|
|
|
| def number(value: float) -> Decimal: |
| result = Decimal(str(value)) |
| if not result.is_finite(): |
| raise ValueError("nonfinite pilot metric") |
| return result |
|
|
|
|
| def validate_metrics(m: Metrics, baseline: Metrics) -> None: |
| if (m["rounds"] <= 0 or m["requests"] <= 0 or len(m["kl_by_position"]) != 5 |
| or m["rounds"] != baseline["rounds"] or m["requests"] != baseline["requests"] |
| or m["population_sha256"] != baseline["population_sha256"] |
| or not Decimal(0) <= number(m["accepted_length"]) <= Decimal(5)): |
| raise ValueError("pilot held-out population or metric mismatch") |
| finite(m) |
| rows = m["request_totals"] |
| if (len(m["kl_median_by_position"]) != 5 or len(rows) != m["requests"] |
| or sum(r["rounds"] for r in rows) != m["rounds"] |
| or abs(math.fsum(r["accepted_sum"] for r in rows)/m["rounds"]-m["accepted_length"]) > 1e-12): |
| raise ValueError("pilot request aggregates do not reproduce E[A]") |
| if (len({r["request_sha256"] for r in rows}) != len(rows) |
| or any(not r["family"] or not r["task_label"] or type(r["rounds"]) is not int or r["rounds"] < 1 |
| or not 0 <= r["accepted_sum"] <= 5*r["rounds"] for r in rows)): |
| raise ValueError("invalid pilot request aggregates/family coverage") |
|
|
|
|
| def outcome(spec: Spec, evaluation: Evaluation) -> dict[str, bool]: |
| finite(evaluation) |
| DEV_SPLITS = split_roles(evaluation["baseline"]) |
| SPLITS = tuple(DEV_SPLITS) |
| if (evaluation["schema"] != "dsv41.d2.pilot-evaluation.v1" or evaluation["synthetic"] != spec["synthetic"] |
| or evaluation["evaluator"] != EVALUATOR |
| or set(evaluation["baseline"]) != set(SPLITS) or set(evaluation["arms"]) != set(ARMS) |
| or set(evaluation["checkpoints"]) != set(ARMS) or set(evaluation["k2_sha256"]) != set(SPLITS)): |
| raise ValueError("missing/mismatched pilot arms, splits or evaluator") |
| if "sample" not in spec: |
| raise ValueError("pilot verdict requires its once-only pre-training sample") |
| validate_sample(spec["sample"], spec["manifest_sha256"]) |
| validate_sample_evaluation(spec, evaluation, SPLITS) |
| points = {str(n) for n in spec["checkpoints"]} |
| for name in ARMS: |
| if set(evaluation["arms"][name]) != points or set(evaluation["checkpoints"][name]) != points: |
| raise ValueError("missing pilot checkpoint") |
| for n in points: |
| if set(evaluation["arms"][name][n]) != set(SPLITS): |
| raise ValueError("missing pilot held-out split") |
| for split in SPLITS: |
| validate_metrics(evaluation["baseline"][split], evaluation["baseline"][split]) |
| validate_metrics(evaluation["arms"][name][n][split], evaluation["baseline"][split]) |
| checks = {} |
| intervals = uncertainty(evaluation, spec["checkpoints"][-2:]) |
| for point in spec["checkpoints"][-2:]: |
| for split in SPLITS: |
| prefix = f"{point}/{DEV_SPLITS[split]}" |
| base = evaluation["baseline"][split] |
| ref, fast = [evaluation["arms"][name][str(point)][split] for name in ("REF", "FAST")] |
| for arm in ("REF", "FAST"): |
| ci = intervals[prefix+f"/{arm}_gain"] |
| checks[prefix+f"/{arm}_useful_gain"] = number(ci["difference"]) >= Decimal(".05") and ci["lower"] > 0 |
| checks[prefix+"/non_inferiority"] = number(intervals[prefix+"/FAST_minus_REF"]["lower"]) >= Decimal("-.03") |
| checks[prefix+"/no_baseline_regression"] = number(fast["accepted_length"])-number(base["accepted_length"]) >= Decimal("-.01") |
| checks[prefix+"/KL_median"] = all(number(f)-number(r) <= Decimal(".01") for f, r in |
| zip(fast["kl_median_by_position"], ref["kl_median_by_position"])) |
|
|
| return checks |
|
|
|
|
| def decision(spec: Spec, evaluation: Evaluation, checks: dict[str, bool]) -> tuple[str, bool]: |
| intervals = uncertainty(evaluation, spec["checkpoints"][-2:]) |
| crossings: list[bool] = [] |
| definitive_failure = False |
| for key, passed in checks.items(): |
| if passed: |
| continue |
| crossing = False |
| if key.endswith("/non_inferiority"): |
| ci = intervals[key.rsplit("/", 1)[0]+"/FAST_minus_REF"] |
| crossing = ci["lower"] < -.03 < ci["upper"] |
| elif key.endswith("_useful_gain"): |
| ci = intervals[key.removesuffix("_useful_gain")+"_gain"] |
| crossing = any(ci["lower"] < bound < ci["upper"] for bound in (0., .05)) |
| crossings.append(crossing) |
| |
| |
| |
| unproved_adaptation = key.endswith("_useful_gain") and not checks[ |
| key.rsplit("/", 1)[0]+"/REF_useful_gain"] |
| definitive_failure |= not crossing and not unproved_adaptation |
| if definitive_failure: |
| return "FAIL", False |
| if crossings and all(crossings): |
| return "INCONCLUSIVE", True |
| return status(checks), False |
|
|
|
|
| def build(plan: Path, evaluation_path: Path, root: Path, manifest: Path, parity: Path, g1: Path) -> Receipt: |
| spec = read_spec(plan, sampled=True) |
| validate_inputs(spec, manifest, parity, g1) |
| evaluation = cast(Evaluation, json.loads(evaluation_path.read_text())) |
| if evaluation["plan_sha256"] != sha256(plan): |
| raise ValueError("pilot evaluation receipt has a mismatched plan") |
| data = D2Blocks(manifest, "train") |
| verify_units(spec, data) |
| for name in ARMS: |
| arm = read_arm(root, name, spec, plan, len(data)) |
| if evaluation["checkpoints"].get(name) != arm["checkpoints"]: |
| raise ValueError("evaluated checkpoint differs from arm receipt") |
| checks = outcome(spec, evaluation) |
| verdict, another_seed = decision(spec, evaluation, checks) |
| evidence = {"plan": spec, "evaluation": evaluation, "checks": checks, "rule_version": RULE_VERSION, "sample_sha256": spec["sample"]["sha256"]} |
| return {"schema": "dsv41.d2.pilot-receipt.v1", "passed": verdict == "PASS", "verdict": verdict, |
| "intervals": uncertainty(evaluation, spec["checkpoints"][-2:]), |
| "extra_seed_eligible": another_seed, "synthetic": spec["synthetic"], |
| "population": population_report(evaluation), |
| "rule": RULE, "rule_version": RULE_VERSION, "plan": spec, "plan_sha256": sha256(plan), "evaluation": evaluation, |
| "evaluation_sha256": sha256(evaluation_path), "checks": checks, "evidence_sha256": digest(evidence), |
| "sample_sha256": spec["sample"]["sha256"]} |
|
|
|
|
| def main() -> None: |
| p = argparse.ArgumentParser(__doc__) |
| for n in ("plan", "evaluation", "root", "manifest", "parity-receipt", "g1-receipt", "out"): |
| p.add_argument("--"+n, required=True, type=Path) |
| a = p.parse_args() |
| result = build(a.plan, a.evaluation, a.root, a.manifest, a.parity_receipt, a.g1_receipt) |
| write_json(a.out, result) |
| print(json.dumps({"passed": result["passed"], "synthetic": result["synthetic"], "checks": result["checks"]}), flush=True) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
