69 lines
3.2 KiB
Python
69 lines
3.2 KiB
Python
"""Inspect only the frozen calibration baseline; no held-out inputs accepted."""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import math
|
|
from pathlib import Path
|
|
|
|
|
|
BASELINE_SHA256 = "56f42c523527dd748918bb8f17be6bd7818158227e692fccaf387d915f9d0a94"
|
|
|
|
|
|
def analyze(raw, policy):
|
|
if hashlib.sha256(raw).hexdigest() != BASELINE_SHA256:
|
|
raise ValueError("Expected exact frozen calibration baseline")
|
|
report = json.loads(raw)
|
|
if not report["complete"] or report["split"] != "calibration" or report["held_out_evaluated"]:
|
|
raise ValueError("Only complete calibration evidence is accepted")
|
|
if report["fixture_sha256"] != policy["fixture_sha256"]:
|
|
raise ValueError("Fixture differs from frozen hypothesis")
|
|
threshold = policy["semantic_cosine_minimum"]
|
|
if not math.isfinite(threshold) or not -1 <= threshold <= 1:
|
|
raise ValueError("Invalid cosine minimum")
|
|
scopes = {}
|
|
rows = []
|
|
for row in report["rows"]:
|
|
if row["query"]["split"] != "calibration":
|
|
raise ValueError("Non-calibration query in report")
|
|
label = row["query"]["classification"]
|
|
if label not in ("positive", "absent_artifact"):
|
|
raise ValueError("Unsupported relevance label")
|
|
score = row["semantic_raw"][0]["score"]
|
|
if not math.isfinite(score) or not -1 <= score <= 1:
|
|
raise ValueError("Invalid raw semantic cosine")
|
|
accepted = score >= threshold
|
|
group = scopes.setdefault(row["scope"], {
|
|
"positive_scores": [], "absent_scores": [],
|
|
"retained_positive_top_results": 0, "absent_queries_with_retained_neighbors": 0,
|
|
})
|
|
group["positive_scores" if label == "positive" else "absent_scores"].append(score)
|
|
if accepted:
|
|
group["retained_positive_top_results" if label == "positive"
|
|
else "absent_queries_with_retained_neighbors"] += 1
|
|
rows.append({"id": row["query"]["id"], "scope": row["scope"], "label": label,
|
|
"top_cosine": score, "accepted_by_hypothesis": accepted})
|
|
for group in scopes.values():
|
|
positive = group.pop("positive_scores")
|
|
absent = group.pop("absent_scores")
|
|
group.update(positive_requests=len(positive), absent_requests=len(absent),
|
|
minimum_positive_top_cosine=min(positive),
|
|
maximum_absent_top_cosine=max(absent),
|
|
observed_separation=min(positive) - max(absent))
|
|
return {"policy_id": policy["id"], "calibration_only": True,
|
|
"held_out_evaluated": False, "production_policy_changed": False,
|
|
"scopes": scopes, "rows": rows}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("report", type=Path)
|
|
parser.add_argument("output", type=Path)
|
|
args = parser.parse_args()
|
|
policy_path = Path(__file__).parent / "policies/cosine-v1.json"
|
|
policy_bytes = policy_path.read_bytes()
|
|
result = analyze(args.report.read_bytes(), json.loads(policy_bytes))
|
|
result["policy_sha256"] = hashlib.sha256(policy_bytes).hexdigest()
|
|
with args.output.open("x", encoding="utf-8") as stream:
|
|
json.dump(result, stream, indent=2)
|
|
print(json.dumps(result["scopes"], indent=2))
|