File size: 3,673 Bytes
210f9af | 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 | """Serving parity and repeat-run determinism (docs/classifier-plan.md §5.5 item 6, §6.4). Pure Python.
- ``serving_parity``: node12 endpoint logits vs pod reference logits on the same posts (the 1,000 test posts):
per-field gated label flip rate and the largest calibrated gated probability difference; passes with <= 1% flips
per field and a max difference <= 0.05 (twitter HANDOFF rule).
- ``determinism``: two runs of the same checkpoint on the same posts: max probability difference <= 1e-3.
Run as ``python -m training.parity serving|determinism A.jsonl B.jsonl --taxonomy DIR [--temperatures T.json]``.
"""
from __future__ import annotations
import argparse
import json
from pathlib import Path
from typing import Mapping
from .calibrate import gated, max_prob_diff
from .taxonomy import Taxonomy, argmax
FLIPS_MAX = 0.01
PROB_MAX = 0.05
DETERMINISM_MAX = 1e-3
def _paired(a_rows, b_rows) -> list[tuple[dict, dict]]:
a = {r["id"]: r for r in a_rows}
b = {r["id"]: r for r in b_rows}
if len(a) != len(list(a_rows)) or len(b) != len(list(b_rows)):
raise ValueError("duplicate ids")
if set(a) != set(b):
raise ValueError(f"id sets differ ({len(set(a) ^ set(b))} ids in only one side)")
return [(a[i], b[i]) for i in sorted(a)]
def compare(a_rows, b_rows, tax: Taxonomy, temps: Mapping[str, float] | None = None) -> dict:
pairs = _paired(list(a_rows), list(b_rows))
flips = {f: 0 for f in tax.names}
worst = 0.0
for a, b in pairs:
pa, pb = gated(a["logits"], tax, temps), gated(b["logits"], tax, temps)
worst = max(worst, max_prob_diff(pa, pb))
la, lb = argmax(pa), argmax(pb)
for f in tax.names:
flips[f] += la[f] != lb[f]
n = len(pairs)
return {"n": n, "flips": flips, "flip_rate": {f: (v / n if n else None) for f, v in flips.items()},
"max_prob_diff": worst}
def serving_parity(reference_rows, served_rows, tax: Taxonomy, temps: Mapping[str, float] | None = None) -> dict:
out = compare(reference_rows, served_rows, tax, temps)
out["pass"] = bool(out["n"]) and all(v <= FLIPS_MAX for v in out["flip_rate"].values()) \
and out["max_prob_diff"] <= PROB_MAX
out["thresholds"] = {"flip_rate": FLIPS_MAX, "max_prob_diff": PROB_MAX}
return out
def determinism(run_a_rows, run_b_rows, tax: Taxonomy, temps: Mapping[str, float] | None = None) -> dict:
out = compare(run_a_rows, run_b_rows, tax, temps)
out["argmax_identical"] = all(v == 0 for v in out["flips"].values())
out["pass"] = bool(out["n"]) and out["max_prob_diff"] <= DETERMINISM_MAX
out["threshold"] = DETERMINISM_MAX
return out
def main(argv=None) -> int:
from .data import load_jsonl
p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("kind", choices=["serving", "determinism"])
p.add_argument("a", help="Reference (or first run) predictions JSONL")
p.add_argument("b", help="Served (or second run) predictions JSONL")
p.add_argument("--taxonomy", required=True, help="taxonomy.json or a directory holding it")
p.add_argument("--temperatures")
p.add_argument("--out")
args = p.parse_args(argv)
tax = Taxonomy.load(args.taxonomy)
temps = json.loads(Path(args.temperatures).read_text()) if args.temperatures else None
fn = serving_parity if args.kind == "serving" else determinism
res = fn(load_jsonl(args.a), load_jsonl(args.b), tax, temps)
text = json.dumps(res, indent=1)
if args.out:
Path(args.out).write_text(text + "\n")
print(text)
return 0 if res["pass"] else 1
if __name__ == "__main__":
raise SystemExit(main())
|