#!/usr/bin/env python3 """ calibrate-decision-layer.py — is "inverted confidence" a property of constrained- token softmax, or an artifact of one prompt? And does temperature scaling fix it? Design, fixed before any of these calls were made (2026-09-21): Question 1 Does the high-confidence/low-confidence accuracy gap seen with letters A-D in fixed order survive (a) shuffling which class gets which letter, per item, and (b) answering with the label word instead of a letter? Question 2 After temperature scaling fitted on a disjoint holdout, what is the ECE on the original 200 test cases? Data TEST = data/decision-cases.jsonl (the published 200, unchanged) HOLDOUT = 200 further AG News test items, 50 per class, seed 20260922, excluding every TEST id. Written to data/decision-holdout.jsonl. Variants letters-fixed A=world B=sports C=business D=scitech (as published) letters-shuffled class-to-letter mapping permuted per item (seeded by id) words answer token is the label word itself Models gpt-4.1-nano, gpt-4o-mini (the two that accept logit_bias) Confidence probabilities are read ONLY from the four allowed tokens and renormalised over them. (The published run normalised over the top-4 logprobs without filtering; with a +100 bias those are the four letters in practice, but it was not enforced.) Reported accuracy, per-class recall, ECE (10 equal-width bins), accuracy at conf >= 0.9 vs < 0.8 with a two-sided Fisher exact p, then the same after temperature scaling (single T, minimising holdout NLL). Nothing here calls Jev. Every number this script produces is about OpenAI models. Usage python scripts/calibrate-decision-layer.py --out public/data/decision-calibration.csv """ from __future__ import annotations import argparse import csv import json import math import os import random import sys import time import urllib.request from collections import Counter, defaultdict from concurrent.futures import ThreadPoolExecutor import httpx import tiktoken CLASSES = ["world", "sports", "business", "scitech"] DESCRIPTIONS = { "world": "World news: international affairs, politics, conflicts, diplomacy, governments", "sports": "Sports: games, matches, athletes, teams, leagues, tournaments", "business": "Business: companies, markets, economy, finance, earnings, deals", "scitech": "Science and technology: research, space, computing, internet, software, gadgets", } WORDS = {"world": " World", "sports": " Sports", "business": " Business", "scitech": " Science"} LETTERS = ["A", "B", "C", "D"] TASK = "Classify this news article into one topic." AG_NAMES = {0: "world", 1: "sports", 2: "business", 3: "scitech"} def load_env() -> None: for f in (".env.local", ".env"): if os.path.exists(f): for line in open(f): line = line.strip() if line and not line.startswith("#") and "=" in line: k, v = line.removeprefix("export ").split("=", 1) os.environ.setdefault(k.strip(), v.strip().strip("\"'")) def build_holdout(test_ids: set[str], path: str) -> list[dict]: if os.path.exists(path): return [json.loads(line) for line in open(path)] rows = [] for off in range(200, 7600, 400): # different pages from the TEST pull url = ("https://datasets-server.huggingface.co/rows?dataset=fancyzhx%2Fag_news" f"&config=default&split=test&offset={off}&length=100") for attempt in range(12): # the datasets server returns 500 "busier than usual" in bursts try: data = json.load(urllib.request.urlopen(url, timeout=30)) break except Exception: time.sleep(min(5 * (attempt + 1), 30)) else: raise SystemExit(f"could not fetch AG News page {off}") for r in data["rows"]: rid = f"ag-{r['row_idx']:05d}" if rid not in test_ids: rows.append({"id": rid, "state": r["row"]["text"], "label": AG_NAMES[r["row"]["label"]]}) rng = random.Random(20260922) out = [] for c in CLASSES: out += rng.sample([r for r in rows if r["label"] == c], 50) rng.shuffle(out) with open(path, "w") as fh: for r in out: fh.write(json.dumps(r) + "\n") return out def make_prompt(variant: str, case_id: str): """Return (system prompt, {token_text: class}).""" if variant == "words": lines = "\n".join(f"{WORDS[c].strip()} = {DESCRIPTIONS[c]}" for c in CLASSES) return (f"{TASK} Answer with exactly one word.\n\n{lines}", {WORDS[c]: c for c in CLASSES}) order = list(CLASSES) if variant == "letters-shuffled": random.Random(f"shuffle:{case_id}").shuffle(order) lines = "\n".join(f"{LETTERS[i]} = {c}: {DESCRIPTIONS[c]}" for i, c in enumerate(order)) return (f"{TASK} Answer with exactly one letter.\n\n{lines}", {LETTERS[i]: c for i, c in enumerate(order)}) def call(client: httpx.Client, model: str, variant: str, case: dict, enc) -> dict: system, token_to_class = make_prompt(variant, case["id"]) bias = {} for tok in token_to_class: ids = enc.encode(tok) if len(ids) != 1: raise SystemExit(f"{tok!r} is not a single token") bias[str(ids[0])] = 100 body = { "model": model, "messages": [{"role": "system", "content": system}, {"role": "user", "content": case["state"]}, {"role": "assistant", "content": "Answer:"}], "logit_bias": bias, "max_tokens": 1, "temperature": 0, "logprobs": True, "top_logprobs": 8, } for attempt in range(6): r = client.post("https://api.openai.com/v1/chat/completions", json=body, timeout=60) if r.status_code == 200: break if r.status_code in (429, 500, 502, 503): time.sleep(1.5 * (attempt + 1)) continue r.raise_for_status() else: r.raise_for_status() top = r.json()["choices"][0]["logprobs"]["content"][0]["top_logprobs"] # Only the four allowed tokens count. Match on exact token text; a token # missing from the top-8 gets a floor probability rather than zero. logp = {c: -30.0 for c in CLASSES} for t in top: cls = token_to_class.get(t["token"]) if cls is None and variant != "words": cls = token_to_class.get(t["token"].strip()) if cls is not None: logp[cls] = max(logp[cls], t["logprob"]) return {"id": case["id"], "label": case["label"], **{f"lp_{c}": logp[c] for c in CLASSES}} def softmax_T(logps: list[float], T: float) -> list[float]: z = [lp / T for lp in logps] m = max(z) e = [math.exp(v - m) for v in z] s = sum(e) return [v / s for v in e] def fit_temperature(rows: list[dict]) -> float: def nll(T: float) -> float: total = 0.0 for r in rows: p = softmax_T([r[f"lp_{c}"] for c in CLASSES], T) total -= math.log(max(p[CLASSES.index(r["label"])], 1e-12)) return total / len(rows) grid = [0.25 * 1.06 ** i for i in range(90)] # 0.25 .. ~45 return min(grid, key=nll) def fisher_two_sided(a: int, b: int, c: int, d: int) -> float: n, r1, c1 = a + b + c + d, a + b, a + c if min(r1, n - r1, c1, n - c1) == 0: return 1.0 def p(x): return math.comb(r1, x) * math.comb(n - r1, c1 - x) / math.comb(n, c1) p0 = p(a) return min(1.0, sum(p(x) for x in range(max(0, c1 - (n - r1)), min(r1, c1) + 1) if p(x) <= p0 * (1 + 1e-9))) def evaluate(rows: list[dict], T: float) -> dict: preds = [] for r in rows: p = softmax_T([r[f"lp_{c}"] for c in CLASSES], T) k = max(range(4), key=lambda i: p[i]) preds.append((p[k], CLASSES[k] == r["label"], CLASSES[k], r["label"])) n = len(preds) bins = defaultdict(list) for conf, ok, *_ in preds: bins[min(int(conf * 10), 9)].append((conf, ok)) ece = sum(len(b) / n * abs(sum(o for _, o in b) / len(b) - sum(c for c, _ in b) / len(b)) for b in bins.values()) hi = [ok for conf, ok, *_ in preds if conf >= 0.9] lo = [ok for conf, ok, *_ in preds if conf < 0.8] recall = {c: sum(ok for _, ok, _, lab in preds if lab == c) / max(1, sum(1 for *_, lab in preds if lab == c)) for c in CLASSES} return { "n": n, "accuracy": sum(ok for _, ok, *_ in preds) / n, "ece": ece, "mean_conf": sum(c for c, *_ in preds) / n, "hi_n": len(hi), "hi_acc": (sum(hi) / len(hi)) if hi else float("nan"), "lo_n": len(lo), "lo_acc": (sum(lo) / len(lo)) if lo else float("nan"), "fisher_p": fisher_two_sided(sum(hi), len(hi) - sum(hi), sum(lo), len(lo) - sum(lo)) if hi and lo else float("nan"), "pred_counts": dict(Counter(p for _, _, p, _ in preds)), **{f"recall_{c}": recall[c] for c in CLASSES}, } def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--test", default="data/decision-cases.jsonl") ap.add_argument("--holdout", default="data/decision-holdout.jsonl") ap.add_argument("--models", default="gpt-4.1-nano,gpt-4o-mini") ap.add_argument("--variants", default="letters-fixed,letters-shuffled,words") ap.add_argument("--out", default="public/data/decision-calibration.csv") ap.add_argument("--raw-out", default="public/data/decision-calibration-raw.csv") ap.add_argument("--workers", type=int, default=8) args = ap.parse_args() load_env() test = [json.loads(line) for line in open(args.test)] holdout = build_holdout({c["id"] for c in test}, args.holdout) assert not ({c["id"] for c in test} & {c["id"] for c in holdout}), "holdout overlaps test" print(f"test={len(test)} holdout={len(holdout)} (disjoint)", file=sys.stderr) enc = tiktoken.get_encoding("o200k_base") client = httpx.Client(headers={"Authorization": f"Bearer {os.environ['OPENAI_API_KEY']}"}) summary, raw = [], [] for model in args.models.split(","): for variant in args.variants.split(","): t0 = time.time() with ThreadPoolExecutor(args.workers) as pool: h = list(pool.map(lambda c: call(client, model, variant, c, enc), holdout)) t = list(pool.map(lambda c: call(client, model, variant, c, enc), test)) T = fit_temperature(h) before, after = evaluate(t, 1.0), evaluate(t, T) print(f"{model:13} {variant:17} acc={before['accuracy']:.3f} " f"ECE {before['ece']:.3f}->{after['ece']:.3f} (T={T:.2f}) " f"hi {before['hi_acc']:.3f} (n={before['hi_n']}) vs lo {before['lo_acc']:.3f} (n={before['lo_n']}) " f"p={before['fisher_p']:.3f} scitech recall={before['recall_scitech']:.2f} [{time.time()-t0:.0f}s]", file=sys.stderr) for stage, ev in (("raw", before), ("temperature_scaled", after)): summary.append({"model": model, "variant": variant, "stage": stage, "temperature": round(T if stage != "raw" else 1.0, 3), **{k: (round(v, 4) if isinstance(v, float) else v) for k, v in ev.items() if k != "pred_counts"}, "pred_counts": json.dumps(ev["pred_counts"], sort_keys=True)}) for split, rows in (("holdout", h), ("test", t)): for r in rows: raw.append({"model": model, "variant": variant, "split": split, **r}) for path, rows in ((args.out, summary), (args.raw_out, raw)): with open(path, "w", newline="") as fh: w = csv.DictWriter(fh, fieldnames=list(rows[0])) w.writeheader() w.writerows(rows) print(f"wrote {path} ({len(rows)} rows)", file=sys.stderr) return 0 if __name__ == "__main__": raise SystemExit(main())