Buckets:
| """All numbers of the Evaluation section: readout accuracy and calibration, reasoning accuracy and cost, | |
| agreement, disagreement AUROC, routing curves, latency and the pre-registered hypotheses H1-H4.""" | |
| from __future__ import annotations | |
| import collections | |
| import csv | |
| import json | |
| import math | |
| import os | |
| import numpy as np | |
| import od_score | |
| from common import auroc, brier, ece, fit_temperature, nll, read_jsonl, softmax | |
| PRED = "results/predictions" | |
| READOUTS = ["base", "gold_only", "votes_only", "joint"] | |
| PARTNER = {"base": "base", "gold_only": "base", "votes_only": "base", "joint": "joint"} # readout -> its reasoner | |
| IN_DIST = {"eval_proc", "eval_ts"} | |
| EVAL_SETS = ["eval_proc", "eval_ts", "opendecisions", "localllama"] | |
| FRACTIONS = [round(0.1 * i, 1) for i in range(11)] | |
| N_BOOT = 1000 | |
| def load(name): | |
| path = f"{PRED}/{name}.jsonl" | |
| return {r["id"]: r for r in read_jsonl(path)} if os.path.exists(path) else None | |
| def calibrated(items, lps, temps): | |
| return [softmax(np.asarray(lp), temps[it["qtype"]]) for it, lp in zip(items, lps)] | |
| def fit_temps(cal_items, cal_lps): | |
| temps = {} | |
| for qt in ("choice", "noul", "score"): | |
| idx = [i for i, it in enumerate(cal_items) if it["qtype"] == qt] | |
| temps[qt] = fit_temperature([np.asarray(cal_lps[i]) for i in idx], [cal_items[i]["gold"] for i in idx]) if idx else 1.0 | |
| return temps | |
| def calib_block(items, probs): | |
| gold = [it["gold"] for it in items] | |
| conf = np.array([p.max() for p in probs]) | |
| corr = np.array([int(np.argmax(p)) == g for p, g in zip(probs, gold)]) | |
| return {"n": len(items), "accuracy": float(corr.mean()), "ece15": ece(conf, corr), "nll": nll(probs, gold), | |
| "brier": brier(probs, gold)} | |
| def group_acc(items, correct, key): | |
| g = collections.defaultdict(list) | |
| for it, c in zip(items, correct): | |
| k = key(it) | |
| if k is not None: | |
| g[str(k)].append(c) | |
| return {k: {"acc": float(np.mean(v)), "n": len(v)} for k, v in sorted(g.items())} | |
| def od_capability(items, probs): | |
| tasks = od_score.read_jsonl("data/raw/jgeuter__OpenDecisions/opendecisions.jsonl") | |
| preds = [{"id": it["od_id"], "probs": {lab: float(x) for lab, x in zip(it["od_labels"], p)}} | |
| for it, p in zip(items, probs)] | |
| s = od_score.score(tasks, preds) | |
| return {k: s[k] for k in ("capability", "intelligence", "calibration", "accuracy_per_type", "calibration_items_tvd")} | |
| def ll_block(items, probs): | |
| kl = [float(np.sum(np.where(np.asarray(it["target"]) > 0, np.asarray(it["target"]) * | |
| (np.log(np.maximum(it["target"], 1e-12)) - np.log(np.maximum(p, 1e-12))), 0.0))) | |
| for it, p in zip(items, probs)] | |
| corr = [int(np.argmax(p)) == it["gold"] for it, p in zip(items, probs)] | |
| agree = [i for i, it in enumerate(items) if it.get("panel_agrees")] | |
| return {"accuracy_all": float(np.mean(corr)), "kl_from_gold": float(np.mean(kl)), | |
| "accuracy_panel_agrees": float(np.mean([corr[i] for i in agree])) if agree else None, | |
| "n_panel_agrees": len(agree), "by_workflow": group_acc(items, corr, lambda it: it["family"])} | |
| def margin(p): | |
| s = np.sort(np.log(np.maximum(p, 1e-12)))[::-1] | |
| return float(s[0] - s[1]) | |
| def router_features(it, p): | |
| k = len(p) | |
| ent = float(-(p * np.log(np.maximum(p, 1e-12))).sum() / math.log(k)) | |
| return [p.max(), margin(p), ent, math.log(k), float(it["qtype"] == "choice"), float(it["qtype"] == "noul")] | |
| def fit_logreg(X, y, l2=1.0, iters=2000, lr=0.5): | |
| X = np.asarray(X, float) | |
| mu, sd = X.mean(0), X.std(0) + 1e-9 | |
| Z = np.hstack([(X - mu) / sd, np.ones((len(X), 1))]) | |
| w = np.zeros(Z.shape[1]) | |
| y = np.asarray(y, float) | |
| for _ in range(iters): # plain gradient descent, L2 on weights | |
| p = 1 / (1 + np.exp(-Z @ w)) | |
| g = Z.T @ (p - y) / len(y) + l2 * np.r_[w[:-1], 0] / len(y) | |
| w -= lr * g | |
| return lambda Xn: 1 / (1 + np.exp(-np.hstack([(np.asarray(Xn, float) - mu) / sd, np.ones((len(Xn), 1))]) @ w)) | |
| def routing_curve(r_ok, t_ok, t_tok, route_score): | |
| """route_score: lower = route first. Accuracy and mean generated tokens at each fraction f.""" | |
| n = len(r_ok) | |
| order = np.argsort(route_score, kind="mergesort") | |
| rows = [] | |
| acc_r, acc_t, mean_tok = np.mean(r_ok), np.mean(t_ok), np.mean(t_tok) | |
| for f in FRACTIONS: | |
| m = int(round(f * n)) | |
| routed = np.zeros(n, bool) | |
| routed[order[:m]] = True | |
| acc = float(np.where(routed, t_ok, r_ok).mean()) | |
| rows.append({"f": f, "accuracy": acc, "mean_tokens": float(np.where(routed, t_tok, 0).mean()), | |
| "random_accuracy": float((1 - f) * acc_r + f * acc_t), "random_mean_tokens": float(f * mean_tok)}) | |
| return rows | |
| def gap_recovered(curve, f=0.3): | |
| a0, a1 = curve[0]["accuracy"], curve[-1]["accuracy"] | |
| af = next(r["accuracy"] for r in curve if abs(r["f"] - f) < 1e-9) | |
| return (af - a0) / (a1 - a0) if abs(a1 - a0) > 1e-12 else float("nan") | |
| def boot(fn, n_items, seed=0, n=N_BOOT): | |
| rng = np.random.default_rng(seed) | |
| vals = [fn(rng.integers(0, n_items, n_items)) for _ in range(n)] | |
| vals = [v for v in vals if not (isinstance(v, float) and math.isnan(v))] | |
| return float(np.percentile(vals, 2.5)), float(np.percentile(vals, 97.5)) | |
| def compute_all(out_path="results/metrics.json"): | |
| ev = read_jsonl("data/eval.jsonl") | |
| cal = read_jsonl("data/calib.jsonl") | |
| res = {"readout": {}, "reasoning": {}, "agreement": {}, "routing": {}, "latency": {}, "hypotheses": {}} | |
| probs_of, ok_of = {}, {} | |
| for name in READOUTS: | |
| ro = load(f"readout_{name}") | |
| if ro is None: | |
| continue | |
| cal_lps = [ro[it["id"]]["lp"] for it in cal] | |
| temps = fit_temps(cal, cal_lps) | |
| ev_lps = [ro[it["id"]]["lp"] for it in ev] | |
| probs = calibrated(ev, ev_lps, temps) | |
| raw = calibrated(ev, ev_lps, {"choice": 1.0, "noul": 1.0, "score": 1.0}) | |
| probs_of[name] = probs | |
| ok = np.array([int(np.argmax(p)) == it["gold"] for it, p in zip(ev, probs)]) | |
| ok_of[name] = ok | |
| block = {"temperatures": temps, "by_set": {}, "raw_by_set": {}} | |
| for s in EVAL_SETS: | |
| idx = [i for i, it in enumerate(ev) if it["set"] == s] | |
| block["by_set"][s] = calib_block([ev[i] for i in idx], [probs[i] for i in idx]) | |
| block["raw_by_set"][s] = calib_block([ev[i] for i in idx], [raw[i] for i in idx]) | |
| idx = [i for i, it in enumerate(ev) if it["set"] in IN_DIST] | |
| block["in_distribution"] = calib_block([ev[i] for i in idx], [probs[i] for i in idx]) | |
| proc = [i for i, it in enumerate(ev) if it["set"] == "eval_proc"] | |
| block["proc_by_family"] = group_acc([ev[i] for i in proc], ok[proc], lambda it: it["family"]) | |
| block["proc_by_level"] = group_acc([ev[i] for i in proc], ok[proc], lambda it: it["level"]) | |
| od = [i for i, it in enumerate(ev) if it["set"] == "opendecisions"] | |
| block["od_capability"] = od_capability([ev[i] for i in od], [probs[i] for i in od]) | |
| block["od_capability_raw"] = od_capability([ev[i] for i in od], [raw[i] for i in od]) | |
| block["od_by_tier"] = group_acc([ev[i] for i in od], ok[od], lambda it: it["tier"]) | |
| block["od_by_family"] = group_acc([ev[i] for i in od], ok[od], lambda it: it["family"]) | |
| ll = [i for i, it in enumerate(ev) if it["set"] == "localllama"] | |
| block["localllama"] = ll_block([ev[i] for i in ll], [probs[i] for i in ll]) | |
| res["readout"][name] = block | |
| reason_of = {} | |
| for name in ("base", "joint"): | |
| rr = load(f"reason_{name}") | |
| if rr is None: | |
| continue | |
| rows = [rr.get(it["id"]) for it in ev] | |
| t_ok = np.array([r is not None and r["answer"] is not None and r["answer"] == it["gold"] for r, it in zip(rows, ev)]) | |
| post_ok = np.array([r is not None and r["post_lp"] is not None and int(np.argmax(r["post_lp"])) == it["gold"] | |
| for r, it in zip(rows, ev)]) | |
| toks = np.array([r["tokens"] if r else 0 for r in rows]) | |
| reason_of[name] = (rows, t_ok, toks) | |
| blk = {"n": len(ev), "predicted": sum(r is not None for r in rows)} | |
| for s in EVAL_SETS + ["in_distribution", "all"]: | |
| idx = [i for i, it in enumerate(ev) if (s == "all" or it["set"] == s or (s == "in_distribution" and it["set"] in IN_DIST))] | |
| sub = [rows[i] for i in idx if rows[i] is not None] | |
| blk[s] = {"accuracy": float(t_ok[idx].mean()), "post_readout_accuracy": float(post_ok[idx].mean()), | |
| "tokens_median": float(np.median(toks[idx])), "tokens_p90": float(np.percentile(toks[idx], 90)), | |
| "looped_share": float(np.mean([r["looped"] for r in sub])) if sub else None, | |
| "parse_rate": float(np.mean([r["answer"] is not None for r in sub])) if sub else None, | |
| "letter_matches_post_readout": float(np.mean([r["post_lp"] is not None and r["answer"] is not None and | |
| int(np.argmax(r["post_lp"])) == r["answer"] for r in sub])) if sub else None} | |
| proc = [i for i, it in enumerate(ev) if it["set"] == "eval_proc"] | |
| blk["proc_by_family"] = group_acc([ev[i] for i in proc], t_ok[proc], lambda it: it["family"]) | |
| blk["proc_by_level"] = group_acc([ev[i] for i in proc], t_ok[proc], lambda it: it["level"]) | |
| res["reasoning"][name] = blk | |
| # agreement, AUROCs and routing for each readout with its reasoner | |
| curves = {} | |
| for name, probs in probs_of.items(): | |
| partner = PARTNER[name] | |
| if partner not in reason_of: | |
| continue | |
| rows, t_ok, toks = reason_of[partner] | |
| r_ans = np.array([int(np.argmax(p)) for p in probs]) | |
| has = np.array([r is not None and r["answer"] is not None for r in rows]) | |
| t_ans = np.array([r["answer"] if (r is not None and r["answer"] is not None) else -1 for r in rows]) | |
| agree = (r_ans == t_ans) & has | |
| unsure = np.array([1 - p.max() for p in probs]) | |
| blk = {"agreement": float(agree[has].mean()), | |
| "agreement_by_set": {s: float(agree[[i for i, it in enumerate(ev) if it["set"] == s and has[i]]].mean()) for s in EVAL_SETS}, | |
| "disagreement_auroc": auroc(unsure[has], ~agree[has])} | |
| err = (~ok_of[name]) & has | |
| blk["fixable_auroc"] = auroc(unsure[err], t_ok[err]) | |
| res["agreement"][name] = blk | |
| # routing: answer of routed items = reasoning letter, or the readout's own when the chain gave none | |
| t_ok_route = np.where(has, t_ok, ok_of[name]) | |
| curve_m = routing_curve(ok_of[name], t_ok_route, toks, np.array([margin(p) for p in probs])) | |
| res["routing"][name] = {"margin": curve_m, "gap_recovered_f0.3_margin": gap_recovered(curve_m)} | |
| curves[(name, "margin")] = curve_m | |
| # learned agreement router (LEV-style), fitted on calibration items only | |
| ro = load(f"readout_{name}") | |
| rr = load(f"reason_{partner}") | |
| if ro is not None and rr is not None: | |
| temps = res["readout"][name]["temperatures"] | |
| cx, cy = [], [] | |
| for it in cal: | |
| r = rr.get(it["id"]) | |
| if r is None or r["answer"] is None: | |
| continue | |
| p = softmax(np.asarray(ro[it["id"]]["lp"]), temps[it["qtype"]]) | |
| cx.append(router_features(it, p)) | |
| cy.append(int(np.argmax(p)) == r["answer"]) | |
| if len(set(cy)) == 2: | |
| f = fit_logreg(cx, cy) | |
| score = f([router_features(it, p) for it, p in zip(ev, probs)]) | |
| curve_l = routing_curve(ok_of[name], t_ok_route, toks, score) | |
| res["routing"][name]["learned"] = curve_l | |
| res["routing"][name]["gap_recovered_f0.3_learned"] = gap_recovered(curve_l) | |
| curves[(name, "learned")] = curve_l | |
| # the deployable system: readout by default, thinking for routed items, probabilities from the readout | |
| # after the chain (per-type temperatures fitted on calibration chains only) | |
| res["system"] = {} | |
| for name in ("base", "joint"): | |
| if name not in probs_of or name not in reason_of: | |
| continue | |
| rr = load(f"reason_{name}") | |
| cal_rows = [(it, rr.get(it["id"])) for it in cal] | |
| cal_rows = [(it, r) for it, r in cal_rows if r is not None and r["post_lp"] is not None] | |
| post_t = fit_temps([it for it, _ in cal_rows], [r["post_lp"] for _, r in cal_rows]) if cal_rows else {"choice": 1.0, "noul": 1.0, "score": 1.0} | |
| rows = reason_of[name][0] | |
| read_p = probs_of[name] | |
| post_p = [softmax(np.asarray(r["post_lp"]), post_t[it["qtype"]]) if (r is not None and r["post_lp"] is not None) else rp | |
| for it, r, rp in zip(ev, rows, read_p)] | |
| toks = reason_of[name][2] | |
| od = [i for i, it in enumerate(ev) if it["set"] == "opendecisions"] | |
| blk = {"post_temperatures": post_t} | |
| def score(probs, routed): | |
| out = {"mean_tokens": float(np.where(routed, toks, 0).mean()), "routed_share": float(routed.mean())} | |
| for s in EVAL_SETS + ["all"]: | |
| idx = [i for i, it in enumerate(ev) if s == "all" or it["set"] == s] | |
| out[f"acc_{s}"] = float(np.mean([int(np.argmax(probs[i])) == ev[i]["gold"] for i in idx])) | |
| out["od_capability"] = od_capability([ev[i] for i in od], [probs[i] for i in od]) | |
| return out | |
| blk["readout_only"] = score(read_p, np.zeros(len(ev), bool)) | |
| blk["thinking_always"] = score(post_p, np.ones(len(ev), bool)) | |
| margins = np.array([margin(p) for p in read_p]) | |
| for f in (0.1, 0.2, 0.3, 0.5): | |
| routed = np.zeros(len(ev), bool) | |
| routed[np.argsort(margins, kind="mergesort")[:int(round(f * len(ev)))]] = True | |
| mix = [pp if rt else rp for pp, rp, rt in zip(post_p, read_p, routed)] | |
| blk[f"routed_margin_f{f}"] = score(mix, routed) | |
| res["system"][name] = blk | |
| for name in ("base", "joint"): | |
| path = f"{PRED}/latency_{name}.json" | |
| if os.path.exists(path): | |
| res["latency"][name] = json.load(open(path)) | |
| # ------------------------------------------------------------ hypotheses | |
| H = res["hypotheses"] | |
| idx_in = np.array([i for i, it in enumerate(ev) if it["set"] in IN_DIST]) | |
| if "joint" in ok_of and "gold_only" in ok_of: | |
| a, b = ok_of["joint"][idx_in].astype(float), ok_of["gold_only"][idx_in].astype(float) | |
| lo, hi = boot(lambda s: a[s].mean() - b[s].mean(), len(a), seed=1) | |
| H["H1"] = {"joint_minus_gold_only_acc": float(a.mean() - b.mean()), "ci95": [lo, hi], "confirmed": lo > 0} | |
| if "joint" in res["agreement"]: | |
| def auc_of(name, sel): | |
| probs = probs_of[name] | |
| rows = reason_of[PARTNER[name]][0] | |
| has = np.array([rows[i] is not None and rows[i]["answer"] is not None for i in sel]) | |
| dis = np.array([int(np.argmax(probs[i])) != (rows[i]["answer"] if rows[i] is not None else -1) for i in sel]) | |
| uns = np.array([1 - probs[i].max() for i in sel]) | |
| return auroc(uns[has], dis[has]) | |
| alli = np.arange(len(ev)) | |
| H2 = {} | |
| for other in ("gold_only", "base"): | |
| if other in probs_of: | |
| d = auc_of("joint", alli) - auc_of(other, alli) | |
| lo, hi = boot(lambda s: auc_of("joint", alli[s]) - auc_of(other, alli[s]), len(alli), seed=2, n=300) | |
| H2[f"joint_minus_{other}"] = {"delta": d, "ci95": [lo, hi], "ok": d >= 0.03 and lo > 0} | |
| H2["confirmed"] = all(v["ok"] for v in H2.values()) if H2 else False | |
| H["H2"] = H2 | |
| if "joint" in reason_of and "base" in reason_of: | |
| a, b = reason_of["joint"][1].astype(float), reason_of["base"][1].astype(float) | |
| lo, hi = boot(lambda s: a[s].mean() - b[s].mean(), len(a), seed=3) | |
| d = float(a.mean() - b.mean()) | |
| H["H3"] = {"joint_minus_base_reasoning_acc": d, "ci95": [lo, hi], "confirmed": d >= -0.01} | |
| if ("joint", "margin") in curves: | |
| g = gap_recovered(curves[("joint", "margin")]) | |
| H["H4"] = {"gap_recovered_f0.3": g, "readout_acc": curves[("joint", "margin")][0]["accuracy"], | |
| "reasoning_acc": curves[("joint", "margin")][-1]["accuracy"], "confirmed": bool(g >= 0.8)} | |
| if ("joint", "learned") in curves: | |
| H["H4"]["gap_recovered_f0.3_learned_router"] = gap_recovered(curves[("joint", "learned")]) | |
| os.makedirs(os.path.dirname(out_path), exist_ok=True) | |
| with open(out_path, "w") as f: | |
| json.dump(res, f, indent=1, default=float) | |
| with open("results/routing.csv", "w", newline="") as f: | |
| w = csv.writer(f) | |
| w.writerow(["readout", "router", "f", "accuracy", "mean_tokens", "random_accuracy", "random_mean_tokens"]) | |
| for (name, kind), curve in curves.items(): | |
| for r in curve: | |
| w.writerow([name, kind, r["f"], r["accuracy"], r["mean_tokens"], r["random_accuracy"], r["random_mean_tokens"]]) | |
| print(json.dumps(res["hypotheses"], indent=1, default=float)) | |
| return res | |
Xet Storage Details
- Size:
- 17.2 kB
- Xet hash:
- 42cf15249017648f7a45515804e32b24ddf9511ad5ec5891145ffbd3d92aaeec
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.