File size: 11,396 Bytes
e1acc15 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 | """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)
# Amendment 6's explicit exception: if REF cannot demonstrate useful
# adaptation in this cell, missing useful gain there is INCONCLUSIVE.
# It cannot excuse NI, KL, baseline, or another cell's definitive failure.
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()
|