| """Calibrated decision spine for the FSI suit (Phase 2, harness perfection). |
| |
| Turns N votes (each: verdict + self-reported confidence label) into ONE |
| decision the system can stand behind: |
| |
| 1. map each confidence label to its MEASURED accuracy (calibration table), |
| 2. weight votes by that calibrated reliability (not naive majority), |
| 3. governed abstention: below threshold -> "not enough information" |
| (selective prediction; never fake confidence), |
| 4. emit a full chain-of-custody trace. |
| |
| Formula: p_final = (sum of calibrated accuracies of votes for the winning |
| verdict) / (number of votes). Unanimous HIGH votes -> p = acc(HIGH); |
| a weak-LOW unanimous vote stays weak; a split vote is attenuated. Monotone, |
| defensible, and directly measurable against the probe battery. |
| |
| Why (research): verbalized confidence is systematically anti-calibrated |
| (ORCE 2026-05; Direct Confidence Alignment 2025-12; arXiv 2408.11774); |
| small models cannot self-correct with weak self-critique (arXiv 2404.09931), |
| so this layer is deterministic suit logic, not more generation. Selective |
| prediction with calibrated abstention is the production recipe for SLMs |
| (governance-ready SLM recipe 2025-08; conformal selective prediction 2026-07). |
| |
| Usage: |
| from research.decision import decide, accuracy_vs_coverage, load_table |
| """ |
| import json |
| from collections import Counter, defaultdict |
|
|
| BUCKETS = ["HIGH", "MEDIUM", "LOW", "cannot assess"] |
|
|
| |
| def bucket_for(p): |
| if p >= 0.66: |
| return "HIGH" |
| if p >= 0.40: |
| return "MEDIUM" |
| if p > 0.0: |
| return "LOW" |
| return "cannot assess" |
|
|
|
|
| def load_table(path): |
| """Load logs/calib_summary_<tag>.json -> {bucket: {acc, n}}.""" |
| with open(path) as f: |
| s = json.load(f) |
| combined = s.get("combined", s) |
| table = {} |
| for b in BUCKETS: |
| info = (combined.get("buckets") or {}).get(b) or {} |
| table[b] = {"acc": info.get("acc"), "n": info.get("n", 0)} |
| return table |
|
|
|
|
| def calibrated_prob(conf, table, unknown=0.0): |
| """P(correct) for a confidence label per the calibration table.""" |
| c = (conf or "cannot assess").upper() |
| row = (table or {}).get(c) or {} |
| acc = row.get("acc") |
| if acc is None or row.get("n", 0) == 0: |
| return unknown |
| return acc |
|
|
|
|
| def weighted_tally(votes, table, unknown=0.0): |
| """votes: list of {verdict, conf}. Returns {verdict: total_weight}.""" |
| per = defaultdict(float) |
| for v in votes: |
| w = calibrated_prob(v.get("conf"), table, unknown) |
| per[v["verdict"].strip().lower()] += w |
| return dict(per) |
|
|
|
|
| def decide(votes, table, threshold=0.0, unknown=0.0, |
| abstain_text="not enough information"): |
| """One decision from N votes. Returns a decision dict with trace. |
| |
| threshold: abstain unless p_final >= threshold (selective prediction). |
| """ |
| n = len(votes) |
| per = weighted_tally(votes, table, unknown) |
| if not per: |
| return {"verdict": abstain_text, "confidence": "cannot assess", |
| "p": 0.0, "basis": "no votes", "tally": per, |
| "votes": votes, "abstained": True} |
| winner, w_win = max(per.items(), key=lambda kv: kv[1]) |
| p = w_win / n |
| abstained = threshold > 0 and p < threshold |
| if abstained: |
| return {"verdict": abstain_text, "confidence": "LOW", "p": p, |
| "basis": f"below abstention threshold {threshold:.2f}", |
| "tally": per, "votes": votes, "abstained": True} |
| return {"verdict": winner, "confidence": bucket_for(p), "p": p, |
| "basis": "calibrated weighted vote", "tally": per, |
| "votes": votes, "abstained": False} |
|
|
|
|
| def accuracy_vs_coverage(probes, table, thresholds=(0.0, 0.1, 0.2, 0.3, 0.4, 0.5), |
| unknown=0.0): |
| """Selective-prediction curve. |
| |
| probes: list of {votes: [{verdict, conf}], correct: bool} where correct |
| says whether the probe's canonical verdict matches the winner. |
| Returns [{threshold, decided_n, coverage, correct_n, accuracy}]. |
| """ |
| out = [] |
| for t in thresholds: |
| decided, correct = 0, 0 |
| for pr in probes: |
| d = decide(pr["votes"], table, threshold=t, unknown=unknown) |
| if d["abstained"]: |
| continue |
| decided += 1 |
| correct += int(pr["correct"]) |
| out.append({"threshold": t, |
| "decided_n": decided, |
| "coverage": decided / len(probes) if probes else 0.0, |
| "correct_n": correct, |
| "accuracy": correct / decided if decided else None}) |
| return out |
|
|
|
|
| def bucket_abstention_curve(rows, table, known_buckets=BUCKETS): |
| """Single-vote curve: abstain the worst-calibrated buckets until a target |
| accuracy is reached. rows: [{conf, correct}]. Returns list of |
| {drop_below, coverage, accuracy} stepping from worst to best bucket.""" |
| accs = {b: table.get(b, {}).get("acc") for b in known_buckets} |
| accs = {b: a for b, a in accs.items() if a is not None} |
| order = sorted(accs, key=lambda b: accs[b]) |
| out = [] |
| for k in range(len(order) + 1): |
| kept = set(order[k:]) |
| rows_in = [r for r in rows if (r.get("conf") or "cannot assess").upper() in kept] |
| n = len(rows_in) |
| cor = sum(1 for r in rows_in if r.get("correct")) |
| out.append({"drop_worse_than": order[k - 1] if k else None, |
| "kept_buckets": sorted(kept), |
| "coverage": n / len(rows) if rows else 0.0, |
| "accuracy": cor / n if n else None}) |
| return out |
|
|
|
|
| def trace(decision, sources=(), user_note=""): |
| """Chain-of-custody: every decision is attachable to its evidence trail.""" |
| return {"verdict": decision["verdict"], |
| "confidence": decision["confidence"], |
| "p": decision["p"], |
| "basis": decision["basis"], |
| "abstained": decision["abstained"], |
| "tally": decision["tally"], |
| "votes": [{"verdict": v["verdict"], "conf": v["conf"]} |
| for v in decision["votes"]], |
| "sources": list(sources)[:8], |
| "user_note": user_note} |
|
|
