Buckets:
| """Yes/no bias check: one logit offset for yes/no questions, fitted on our calibration items (never S1MB or eval | |
| labels), applied to our eval's yes/no questions and to S1MB. Writes results/yesno_offset.json.""" | |
| import json | |
| import numpy as np | |
| from scipy.optimize import minimize_scalar | |
| from common import read_jsonl | |
| from jevals_eval import PRED, temps_for_model | |
| from s1mb_eval import D, QT, bench_metrics, task_avg | |
| items = read_jsonl(f"{D}/items.jsonl") | |
| tasks = {r["bench"]: r["task"] for r in items} | |
| cal = [x for x in read_jsonl("data/calib.jsonl") if x["qtype"] == "noul"] | |
| ev = [x for x in read_jsonl("runs/v2/data/eval.jsonl") if x["qtype"] == "noul"] | |
| out = {} | |
| for name in ("rl60", "v1_joint", "base"): | |
| pred = {r["id"]: np.array(r["lp"]) for r in read_jsonl(PRED[name])} | |
| T = temps_for_model(PRED[name]) | |
| z = lambda xs: np.array([(pred[x["id"]][0] - pred[x["id"]][1]) / T["noul"] for x in xs]) # option 0 = true | |
| y = lambda xs: np.array([1.0 if int(x["gold"]) == 0 else 0.0 for x in xs]) | |
| zc, yc = z(cal), y(cal) | |
| sig = lambda v: 1 / (1 + np.exp(-v)) | |
| b = minimize_scalar(lambda b: -np.mean(yc * np.log(sig(zc + b) + 1e-12) + (1 - yc) * np.log(1 - sig(zc + b) + 1e-12)), | |
| bounds=(-5, 5), method="bounded").x | |
| evx = [x for x in ev if x["id"] in pred] | |
| ze, ye = z(evx), y(evx) | |
| ba = lambda zz, yy: float(((zz >= 0)[yy == 1].mean() + (zz < 0)[yy == 0].mean()) / 2) | |
| R = json.load(open(f"{D}/runs/{name}.json"))["answers"] | |
| res = {} | |
| for label, off in (("calibrated", 0.0), ("calibrated+offset", b)): | |
| per = {} | |
| for r, a in zip(items, R): | |
| qt = QT[r["task"]] | |
| lp = np.asarray(a["lp"]) / T[qt] | |
| if r["task"] == "noul": | |
| lp = lp + np.array([off, 0.0]) | |
| p = np.exp(lp - lp.max()); p /= p.sum() | |
| opts = r["options"] if r["task"] != "noul" else sorted(r["options"], key=lambda o: o["id"] != "true") | |
| ids = [o["id"] for o in opts] | |
| per.setdefault(r["bench"], []).append({"probs": {ids[j]: float(pj) for j, pj in zip(a["cands"], p)}, | |
| "target": {o["id"]: t for o, t in zip(r["options"], r["target"])}, | |
| "order": [o["id"] for o in r["options"]], "values": {o["id"]: o["value"] for o in r["options"]}}) | |
| adj = {} | |
| for bn, qs in per.items(): | |
| m = bench_metrics(tasks[bn], qs) | |
| adj[bn] = min(1.0, max(0.0, m["skill"])) if m["eligible"] else None | |
| yes = np.mean([q["probs"]["true"] >= 0.5 for bn, qs in per.items() if tasks[bn] == "noul" for q in qs]) | |
| res[label] = {**{k: (v * 100 if v is not None else None) for k, v in task_avg(adj, tasks).items()}, "s1mb_yes_rate": float(yes), | |
| "adjusted": adj} | |
| out[name] = {"offset": float(b), "calib_n": len(cal), "eval_noul_n": len(evx), | |
| "eval_yes_rate": float(np.mean(ze >= 0)), "eval_yes_rate_offset": float(np.mean(ze + b >= 0)), "eval_gold_yes": float(ye.mean()), | |
| "eval_ba": ba(ze, ye), "eval_ba_offset": ba(ze + b, ye), "s1mb": res} | |
| s = res | |
| print(f"{name}: offset {b:+.2f} | eval BA {ba(ze, ye):.3f} -> {ba(ze + b, ye):.3f} | S1MB yes/no {s['calibrated']['noul']:.1f} -> " | |
| f"{s['calibrated+offset']['noul']:.1f}, Task Avg {s['calibrated']['task_avg']:.1f} -> {s['calibrated+offset']['task_avg']:.1f}, " | |
| f"S1MB yes-rate {s['calibrated']['s1mb_yes_rate']:.3f} -> {s['calibrated+offset']['s1mb_yes_rate']:.3f}") | |
| json.dump(out, open("results/yesno_offset.json", "w"), indent=1) | |
Xet Storage Details
- Size:
- 3.62 kB
- Xet hash:
- 341ef72e49160b33e91167194f611dd207a2197686d7197771311a70e8fb65ee
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.