File size: 7,243 Bytes
3ee6a05 | 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 | """Measure Verdict LUT8 against FP16 on fixed Decision Index requests without using gold labels."""
from __future__ import annotations
import argparse
import gzip
import json
import platform
import statistics
import subprocess
import time
from collections import Counter
from pathlib import Path
import coremltools as ct
import numpy as np
from transformers import AutoTokenizer
from assets import ROOT, sha256, verify_assets
from decision_index_engine import adapt_question, as_text
from native_reference import build_request, decode, load_calibrator
from preprocessing import prepare
def selected_requests(rows_path: Path, tokenizer, limit: int, length: int):
"""First ten eligible rows per family, before inspecting either model's output."""
family_counts = Counter()
skipped = Counter()
selected = []
with gzip.open(rows_path, "rt") as stream:
for line in stream:
row = json.loads(line)
family = row["family"]
if family_counts[family] >= 10:
continue
if len(row["questions"]) != 1:
skipped["multiple_questions"] += 1
continue
question = next(iter(row["questions"].values()))
if question["type"] not in ("choice", "noul"):
skipped["unsupported_question_type"] += 1
continue
try:
request = build_request(as_text(row["state"]), adapt_question(question))
except ValueError:
skipped["invalid_or_overcapacity"] += 1
continue
tokens = len(tokenizer(request.text, truncation=False)["input_ids"])
if tokens > length:
skipped["overlength"] += 1
continue
selected.append((row["id"], family, request, tokens))
family_counts[family] += 1
if len(selected) >= limit:
break
return selected, family_counts, skipped
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--rows", type=Path, required=True, help="pinned public Decision Index selected-rows.jsonl.gz")
parser.add_argument("--limit", type=int, default=100)
parser.add_argument("--length", type=int, default=128)
parser.add_argument("--repeats", type=int, default=5)
args = parser.parse_args()
if args.length != 128:
raise ValueError("the predeclared LUT8 validation protocol is L128 only")
if args.limit != 100:
raise ValueError("the predeclared validation manifest uses exactly 100 requests")
source = verify_assets(required=("config.json", "tokenizer.json", "tokenizer_config.json", "calibrator.json"))
tokenizer = AutoTokenizer.from_pretrained(source)
config = json.loads((source / "config.json").read_text())
calibrator = load_calibrator(source)
selected, families, skipped = selected_requests(args.rows, tokenizer, args.limit, args.length)
if len(selected) != args.limit:
raise ValueError(f"only {len(selected)} eligible fixed requests; expected {args.limit}")
packages = {
"fp16": ROOT / "build" / "verdict_fp16_L128_candidates25.mlpackage",
"lut8": ROOT / "build" / "verdict_lut8_kmeans_per_tensor_L128_candidates25.mlpackage",
}
models = {name: ct.models.MLModel(str(path), compute_units=ct.ComputeUnit.ALL) for name, path in packages.items()}
for _, _, request, _ in selected[:2]:
arrays = prepare(tokenizer, config["class_token_index"], request.text, args.length, 25)
for model in models.values():
model.predict(arrays)
rows = []
timings = {name: [] for name in models}
for index, (row_id, family, request, tokens) in enumerate(selected):
arrays = prepare(tokenizer, config["class_token_index"], request.text, args.length, 25)
results = {}
for name, model in models.items():
output = model.predict(arrays)
results[name] = decode(output["logits"], request, calibrator)
if index < 20:
for _ in range(args.repeats):
start = time.perf_counter()
model.predict(arrays)
timings[name].append((time.perf_counter() - start) * 1000)
reference = results["fp16"]
compressed = results["lut8"]
differences = [abs(reference["probabilities"][key] - compressed["probabilities"][key]) for key in request.ids]
rows.append(
{
"id": row_id,
"family": family,
"tokens": tokens,
"candidates": len(request.ids),
"fp16_selected_id": reference["selected_id"],
"lut8_selected_id": compressed["selected_id"],
"selection_agrees": reference["selected_id"] == compressed["selected_id"],
"abstention_agrees": reference["is_abstention"] == compressed["is_abstention"],
"max_probability_error": max(differences),
}
)
errors = np.array([row["max_probability_error"] for row in rows])
selection_agreement = sum(row["selection_agrees"] for row in rows) / len(rows)
abstention_agreement = sum(row["abstention_agrees"] for row in rows) / len(rows)
gates = {
"min_selection_agreement": 0.99,
"min_abstention_agreement": 0.99,
"max_p95_probability_error": 0.02,
"max_worst_probability_error": 0.10,
}
report = {
"suite_file": args.rows.name,
"suite_sha256": sha256(args.rows),
"selection_protocol": "First 10 eligible rows per family in suite order, 100 total; no gold labels used",
"selected_row_ids": [row["id"] for row in rows],
"families": dict(families),
"skipped_before_limit": dict(skipped),
"packages": {name: path.name for name, path in packages.items()},
"hardware": {
"chip": subprocess.run(
["sysctl", "-n", "machdep.cpu.brand_string"], capture_output=True, text=True
).stdout.strip(),
"macos": platform.mac_ver()[0],
},
"questions": len(rows),
"selection_agreement": selection_agreement,
"abstention_agreement": abstention_agreement,
"p95_probability_error": float(np.percentile(errors, 95)),
"worst_probability_error": float(errors.max()),
"median_model_call_ms": {name: statistics.median(values) for name, values in timings.items()},
"gates": gates,
"rows": rows,
}
report["passed"] = (
selection_agreement >= gates["min_selection_agreement"]
and abstention_agreement >= gates["min_abstention_agreement"]
and report["p95_probability_error"] <= gates["max_p95_probability_error"]
and report["worst_probability_error"] <= gates["max_worst_probability_error"]
)
target = ROOT / "reports" / "lut8-L128-suite-parity.json"
target.write_text(json.dumps(report, indent=2) + "\n")
print(
json.dumps({key: value for key, value in report.items() if key not in ("rows", "selected_row_ids")}, indent=2)
)
if not report["passed"]:
raise SystemExit(1)
if __name__ == "__main__":
main()
|