File size: 7,614 Bytes
1d2de8a | 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 | """Compare saved MLX scores with CUDA using identical, frozen temperatures.
No fitting, parameter selection, or changes to inference weights take place.
Partial panels are diagnostic only and can never qualify a release.
"""
import argparse
import json
from collections import defaultdict
from pathlib import Path
import numpy as np
from solomon_mlx._vendor.semantics import listed_probs, p_yes
from solomon_mlx.api import TASKS
from solomon_mlx.artifacts import digest, runtime_identity, sha256
from solomon_mlx.evaluation import compare_rows, load_panel, read_cuda_scores
def decision_probabilities(row, temperature):
"""Parity includes every branch, even when its gold label is not a listed option."""
if row["task"] in ("boolean", "entity", "multilabel"):
p = p_yes(row["letter_logits"], temperature)
return np.array([1 - p, p])
width = row["n"] - 2 if row["head_key"].endswith("choiceR") else row["n"]
return listed_probs(row["letter_logits"], width, temperature)
def compare(panel, scores, cuda_directory, reference, output, *, allow_partial=False):
panel, scores, output = Path(panel), Path(scores), Path(output)
if output.exists():
raise FileExistsError("Parity reports are immutable")
jobs, manifest = load_panel(panel)
identity = json.loads((scores / "identity.json").read_text())
model_binding = json.loads(Path("models/quality/binding.json").read_text())
if identity["runtime"] != runtime_identity(model_binding):
raise ValueError("Scores belong to another MLX runtime")
if identity["panel_sha256"] != manifest["jobs_sha256"]:
raise ValueError("Scores belong to another panel")
groups = defaultdict(list)
for job in jobs:
groups[job["document_key"]].append(job)
rows, files = [], {}
for key, group in groups.items():
path = scores / (key + ".json")
if not path.exists() and allow_partial:
continue
record = json.loads(path.read_text())
body = {k: v for k, v in record.items() if k != "sha256"}
if (
record["sha256"] != digest(body)
or record["identity"] != digest(identity)
or [r["id"] for r in record["rows"]] != [r["id"] for r in group]
):
raise ValueError("Corrupt or mismatched score document")
rows.extend(record["rows"])
files[path.name] = sha256(path)
complete = len(files) == len(groups)
if not allow_partial:
marker = json.loads((scores / "complete.json").read_text())
if marker != {
"identity": digest(identity),
"documents": len(groups),
"branches": len(jobs),
"files": files,
}:
raise ValueError("Incomplete or mismatched completion manifest")
ref = json.loads(Path(reference).read_text())
cuda = read_cuda_scores(cuda_directory, panel, ref["identity"])
selected = {r["id"] for r in rows}
cuda = [r for r in cuda if r["id"] in selected]
binding_path = Path("evaluations/cuda-acceptance/input/serving-binding.json")
source_manifest = json.loads((binding_path.parent / "manifest.json").read_text())
if sha256(binding_path) != source_manifest["files"][binding_path.name]:
raise ValueError("CUDA acceptance binding checksum mismatch")
binding = json.loads(binding_path.read_text())
for key in (
"adapter_sha256",
"trained_heads_sha256",
"model_sha256",
"numerics",
"placement",
"arithmetic",
):
if binding["runtime"][key] != ref["identity"][key]:
raise ValueError("CUDA temperatures belong to another reference")
temperatures = {task: binding["temperatures"]["models"][task]["temperature"] for task in TASKS}
comparisons = {}
cuda_by_id = {r["id"]: r for r in cuda}
for name, temps in (("temperature_one", dict.fromkeys(TASKS, 1.0)), ("cuda_serving", temperatures)):
result = compare_rows(rows, cuda, temperatures=temps, reference_temperatures=temps)
result.pop("quality_gate_passed")
result["accuracy_units"] = result["units"]
result["accuracy_questions"] = result["questions"]
worst, questions = [], defaultdict(list)
for row in rows:
other = {**row, "letter_logits": cuda_by_id[row["id"]]["letter_logits"]}
p = decision_probabilities(row, temps[row["task"]])
q = decision_probabilities(other, temps[row["task"]])
if not np.isfinite(p).all() or not np.isfinite(q).all():
raise ValueError("Nonfinite parity probability")
agrees = int(np.argmax(p)) == int(np.argmax(q))
questions[row["question_id"]].append(agrees)
worst.append(
{
"id": row["id"],
"task": row["task"],
"max_probability_drift": float(np.max(np.abs(p - q))),
"decision_agrees": agrees,
}
)
result.update(
units=len(rows),
questions=len(questions),
unit_decision_agreement=float(np.mean([r["decision_agrees"] for r in worst])),
question_decision_agreement=float(np.mean([all(v) for v in questions.values()])),
max_probability_drift=max(r["max_probability_drift"] for r in worst),
mean_probability_drift=float(np.mean([r["max_probability_drift"] for r in worst])),
)
result["agreement_gate_passed"] = (
result["unit_decision_agreement"] >= 0.999 and result["question_decision_agreement"] >= 0.999
)
result["largest_probability_differences"] = sorted(
worst, key=lambda r: r["max_probability_drift"], reverse=True
)[:10]
comparisons[name] = result
report = {
"scope": "complete text parity panel" if complete else "partial text parity diagnostic",
"complete": complete,
"documents": len(files),
"total_documents": len(groups),
"branches": len(rows),
"total_branches": len(jobs),
"tasks": sorted({r["task"] for r in rows}),
"runtime": identity["runtime"],
"cuda_runtime": ref["identity"],
"panel_sha256": identity["panel_sha256"],
"score_files_sha256": digest(files),
"cuda_serving_binding_sha256": sha256(binding_path),
"temperature_fitting_performed": False,
"temperature_policy": "identical settings on both backends; not MLX calibration",
"comparisons": comparisons,
"parity_gate_passed": complete and all(r["agreement_gate_passed"] for r in comparisons.values()),
"bitwise_equality_claimed": False,
"image_qualification": False,
}
output.parent.mkdir(parents=True, exist_ok=True)
output.write_text(json.dumps(report, indent=2))
print(
json.dumps(
{k: report[k] for k in ("scope", "documents", "branches", "tasks", "parity_gate_passed")},
indent=2,
)
)
return report
if __name__ == "__main__":
parser = argparse.ArgumentParser()
for field in ("panel", "scores", "cuda-directory", "output"):
parser.add_argument("--" + field, required=True)
parser.add_argument("--reference", default="evaluations/bf16-reference-1789901869/report.json")
parser.add_argument("--allow-partial", action="store_true")
args = parser.parse_args()
compare(
args.panel,
args.scores,
args.cuda_directory,
args.reference,
args.output,
allow_partial=args.allow_partial,
)
|