msq-plurality-gap / code /make_readme.py
adeerkhan's picture
MSQ: the unstated premise
5264032 verified
Raw History Blame Contribute Delete
17.1 kB
#!/usr/bin/env python3
"""
Render README.md for the MSQ bucket from results/metrics.json + repr_probe.json.
The prose lives in code/README_template.md; every number in it is injected
from the measured results, so the README can never drift from the data.
"""
from __future__ import annotations
import json
import pathlib
import re
ROOT = pathlib.Path(__file__).resolve().parent.parent
TEMPLATE = ROOT / "code" / "README_template.md"
OUT = ROOT / "README.md"
MODEL_LABEL = {
"minicpm5-2b": "MiniCPM5-2B (primary)",
"minicpm5-2b-base": "MiniCPM5-2B-Base (no SFT/RL)",
"minicpm5-2b-sft": "MiniCPM5-2B-SFT (no RL)",
"qwen3-4b": "Qwen3-4B-Instruct-2507",
"smollm3-3b": "SmolLM3-3B",
"phi4-mini": "Phi-4-mini-instruct",
"tiny-test": "tiny-random (smoke test)",
}
def pct(x, nd=0):
if x is None:
return "n/a"
return f"{x * 100:.{nd}f}%"
def num(x, nd=3):
if x is None:
return "n/a"
return f"{x:.{nd}f}"
def build_qualitative(root: pathlib.Path, prim: str) -> str:
"""Pull real generations and real logit distributions from the primary model.
Three vignettes, each auto-selected as the most striking example of its
kind, so the section is never hand-picked to flatter the result.
"""
p = root / "results" / "raw" / f"{prim}.jsonl"
if not p.exists():
return "_(raw generations not yet available)_"
recs = [json.loads(l) for l in p.open(encoding="utf-8") if l.strip()]
items = {i["id"]: i for i in (json.loads(l) for l in
(root / "dataset" / "msq49.jsonl").open())}
def get(probe, iid):
for r in recs:
if r["probe"] == probe and r["item_id"] == iid:
return r
return None
def clip(s, n=700):
s = " ".join((s or "").split())
return (s[:n] + " …") if len(s) > n else s
def dist_lines(r, only=None):
out = []
for i, (o, pr) in enumerate(zip(r["options"], r["probs"])):
if only is not None and i != only:
continue
mark = " ← chosen" if i == r["argmax_index"] else ""
out.append(f" **{chr(65+i)}** p={pr:.3f} {clip(o, 105)}{mark}")
return "\n".join(out)
out = []
# ---- vignette 1: the instruction changed nothing -----------------------
cands = []
for r in recs:
if r["probe"] != "mcq_neutral" or r["cls"] != "A":
continue
s = get("mcq_specified", r["item_id"])
if not s or len(s["probs"]) != len(r["probs"]):
continue
tv = 0.5 * sum(abs(a - b) for a, b in zip(r["probs"], s["probs"]))
if r["argmax_index"] == s["argmax_index"]:
cands.append((tv, r, s))
if cands:
cands.sort()
tv, neu, spec = cands[0]
it = items[neu["item_id"]]
out.append(f"**(a) The instruction changed nothing.** Item {neu['item_id']} — "
f"_{it['question']}_")
out.append("")
out.append(f"Ground truth: the schools are divided — "
f"{it['majority_schools']} vs {it['minority_school']}.")
out.append("")
out.append("*Nothing named:*")
out.append("")
out.append(dist_lines(neu))
out.append("")
out.append(f"*\"According to the {it['minority_school']} school …\":*")
out.append("")
out.append(dist_lines(spec))
out.append("")
out.append(f"Total-variation distance between the two distributions: "
f"**{tv:.3f}**. The model picked the same option with the same "
f"confidence, before and after being told which school it wanted.")
out.append("")
# ---- vignette 2: it volunteers the dependency and then ignores it ------
cands = []
for r in recs:
if r["probe"] != "mcq_neutral" or r["cls"] != "C":
continue
s = get("mcq_specified", r["item_id"])
if not s:
continue
if s["argmax_index"] == s.get("divided_index"):
cands.append((s["max_prob"], r, s))
if cands:
cands.sort(key=lambda x: -x[0])
_, neu, spec = cands[0]
it = items[neu["item_id"]]
out.append(f"**(b) It notices the dependency, and then ignores it.** Item "
f"{neu['item_id']} — _{it['question']}_")
out.append("")
out.append("*Nothing named:*")
out.append("")
out.append(dist_lines(neu))
out.append("")
out.append(f"*\"The user lives in {it['minority_school']}\":*")
out.append("")
out.append(dist_lines(spec))
out.append("")
out.append("The correct option is printed in the list, matching the stated "
"jurisdiction exactly. The model still answers with the generic "
"one.")
out.append("")
# ---- vignette 3: it invents a school ------------------------------------
cands = []
known = {}
for it in items.values():
known[it["id"]] = {s.lower() for p in it["positions"] for s in p["schools"]}
for r in recs:
if r["probe"] != "ft_two_schools" or r["cls"] != "A":
continue
body = (r["answer"] or "").lower()
mentioned = {s for s in known[r["item_id"]] if s.split()[0] in body}
# count capitalised "School" style tokens that are not in the ground truth
invented = [w for w in ("contemporary", "modern", "progressive", "liberal",
"reformist", "rationalist", "modernist")
if w in body]
if invented:
cands.append((len(invented), r))
if cands:
cands.sort(key=lambda x: -x[0])
_, r = cands[0]
it = items[r["item_id"]]
out.append(f"**(c) Asked to enumerate the competing positions, it invents "
f"one.** Item {r['item_id']} — _{it['question']}_")
out.append("")
out.append(f"Ground truth: {it['majority_schools']} vs {it['minority_school']}. "
f"Prompt: *\"Explain the different scholarly positions … name the "
f"school of thought that holds each.\"*")
out.append("")
out.append("> " + clip(r["answer"], 900))
out.append("")
# ---- vignette 4: free text, plain vs named ------------------------------
for iid in ("A11", "A01", "A02"):
fd, fs = get("ft_default", iid), get("ft_specified", iid)
if fd and fs:
it = items[iid]
out.append(f"**(d) Free text, item {iid}.** _{it['question']}_")
out.append("")
out.append(f"*Plain question:*")
out.append("")
out.append("> " + clip(fd["answer"], 620))
out.append("")
out.append(f"*\"The user follows the {it['minority_school']} school\":*")
out.append("")
out.append("> " + clip(fs["answer"], 620))
out.append("")
break
return "\n".join(out)
def main() -> None:
metrics = json.loads((ROOT / "results" / "metrics.json").read_text())
reprp = ROOT / "results" / "repr_probe.json"
reprd = json.loads(reprp.read_text()) if reprp.exists() else {}
items = [json.loads(l) for l in (ROOT / "dataset" / "msq49.jsonl").open()]
vals: dict[str, str] = {}
order = [m for m in ("minicpm5-2b", "minicpm5-2b-base", "minicpm5-2b-sft",
"qwen3-4b", "smollm3-3b", "phi4-mini") if m in metrics]
if not order:
order = list(metrics)
# headline table
params = {"minicpm5-2b": "2.52B", "minicpm5-2b-base": "2.52B",
"minicpm5-2b-sft": "2.52B", "qwen3-4b": "4.02B",
"smollm3-3b": "3.08B", "phi4-mini": "3.82B"}
lines = ["| Model | Params | context response (TV) | answer unchanged by context "
"| PRR (fiqh) | MSQ (fiqh) | Δ on the correct answer | control accuracy |",
"|---|---|---|---|---|---|---|---|"]
for m in order:
h = metrics[m]["headline"]
cr = metrics[m].get("context_response", {}).get("ALL", {})
lines.append(
f"| {MODEL_LABEL.get(m, m)} | {params.get(m, '?')} | "
f"{num(cr.get('mean_context_response_tv'), 3)} | "
f"{pct(cr.get('argmax_unchanged_rate'))} | "
f"{pct(h.get('PRR_A'))} | {pct(h.get('MSQ_A'))} | "
f"{num(cr.get('median_delta_on_named_position'), 3)} | "
f"{pct(h.get('acc_D'))} |")
vals["headline_table"] = "\n".join(lines)
prim = "minicpm5-2b" if "minicpm5-2b" in metrics else order[0]
ph = metrics[prim]["headline"]
ft = metrics[prim]["free_text"]
mc = metrics[prim]["mcq"]
vals["primary"] = MODEL_LABEL.get(prim, prim)
vals["n_items"] = str(len(items))
vals["n_models"] = str(len(order))
contested = [i for i in items if i["cls"] in ("A", "B", "C", "E")]
vals["n_contested"] = str(len(contested))
import statistics as _st
chance = []
for it in contested:
n = len(it["positions"]) + (1 if it["cls"] in ("A", "B") else 0)
if it["cls"] == "C":
n = len(it["positions"]) + len(it.get("jur_map", {}))
chance.append(1.0 / n)
vals["chance_rate"] = pct(round(_st.fmean(chance), 4))
vals["prr_A"] = pct(ph.get("PRR_A"))
vals["prr_B"] = pct(ph.get("PRR_B"))
vals["prr_C"] = pct(ph.get("PRR_C"))
vals["prr_E"] = pct(ph.get("PRR_E"))
vals["msq_A"] = pct(ph.get("MSQ_A"))
vals["msq_C"] = pct(ph.get("MSQ_C"))
vals["msq_E"] = pct(ph.get("MSQ_E"))
vals["prr_all"] = pct(ph.get("PRR"))
vals["msq_all"] = pct(ph.get("MSQ"))
vals["gap_all"] = pct(ph.get("blind_spot_gap"))
vals["prr_mass"] = pct(ph.get("PRR_mass"))
vals["acc_D"] = pct(ph.get("acc_D"))
vals["disc_default"] = pct(ft.get("ft_default", {}).get("ALL", {}).get("disclosure_rate"))
vals["disc_specified"] = pct(ft.get("ft_specified", {}).get("ALL", {}).get("disclosure_rate"))
vals["disc_two"] = pct(ft.get("ft_two_schools", {}).get("ALL", {}).get("disclosure_rate"))
vals["disc_technical"] = pct(ft.get("ft_technical", {}).get("ALL", {}).get("disclosure_rate"))
vals["attr_default"] = pct(ft.get("ft_default", {}).get("ALL", {}).get("school_attribution_rate"))
vals["attr_two"] = pct(ft.get("ft_two_schools", {}).get("ALL", {}).get("school_attribution_rate"))
vals["attr_technical"] = pct(ft.get("ft_technical", {}).get("ALL", {}).get("school_attribution_rate"))
vals["clar_default"] = pct(ft.get("ft_default", {}).get("ALL", {}).get("clarification_rate"))
vals["clar_two"] = pct(ft.get("ft_two_schools", {}).get("ALL", {}).get("clarification_rate"))
vals["hedge_default"] = pct(ft.get("ft_default", {}).get("ALL", {}).get("hedge_rate"))
vals["hedge_two"] = pct(ft.get("ft_two_schools", {}).get("ALL", {}).get("hedge_rate"))
vals["words_default"] = num(ft.get("ft_default", {}).get("ALL", {}).get("mean_words"), 1)
vals["words_two"] = num(ft.get("ft_two_schools", {}).get("ALL", {}).get("mean_words"), 1)
vals["conf_contested"] = pct(ph.get("conf_contested"))
vals["conf_control"] = pct(ph.get("conf_control"))
vals["change_rate"] = pct(metrics[prim]["answer_change"].get("changed_rate"))
vals["similarity"] = num(metrics[prim]["answer_change"].get("mean_similarity"), 2)
for cls, tag in (("A", "fiqh"), ("B", "controversy"), ("C", "jurisdiction"),
("E", "python version"), ("D", "control")):
vals[f"conf_{cls}"] = pct(mc.get("mcq_neutral", {}).get(cls, {}).get("mean_max_prob"))
vals[f"prr_{cls}"] = pct(mc.get("mcq_neutral", {}).get(cls, {}).get("plurality_recognition_rate"))
pc = metrics[prim].get("positions_contained", {})
for probe, key in (("ft_default", "positions_default"),
("ft_two_schools", "positions_two_schools"),
("ft_pair", "positions_pair"),
("ft_repair", "positions_repair"),
("ft_technical", "positions_technical")):
vals[key] = num(pc.get(probe, {}).get("ALL", {}).get("mean_positions_contained"), 2)
vals["disc_repair"] = pct(ft.get("ft_repair", {}).get("ALL", {}).get("disclosure_rate"))
vals["attr_repair"] = pct(ft.get("ft_repair", {}).get("ALL", {}).get("plural_attribution_rate"))
ca = metrics[prim].get("content_adherence", {})
vals["names_context_rate"] = pct(ca.get("ALL", {}).get("names_context_rate"))
vals["content_adherence_rate"] = pct(ca.get("ALL", {}).get("content_adherence_rate"))
vals["label_adoption_rate"] = pct(ca.get("ALL", {}).get("label_adoption_rate"))
vals["split_ratio"] = num(ca.get("ALL", {}).get("mean_split_ratio"), 3)
bm = metrics[prim].get("behaviour_matrix", {})
vals["bm_both"] = pct(bm.get("ALL", {}).get("both"))
vals["bm_default_only"] = pct(bm.get("ALL", {}).get("default_only"))
vals["bm_switch_only"] = pct(bm.get("ALL", {}).get("switch_only"))
vals["bm_neither"] = pct(bm.get("ALL", {}).get("neither"))
cr = metrics[prim].get("context_response", {})
for cls in ("A", "B", "C", "E", "ALL"):
c = cr.get(cls, {})
vals[f"cr_{cls}"] = num(c.get("mean_context_response_tv"), 3)
vals[f"unchanged_{cls}"] = pct(c.get("argmax_unchanged_rate"))
vals[f"delta_{cls}"] = num(c.get("median_delta_on_named_position"), 4)
vals[f"n_unchanged_{cls}"] = str(c.get("n_unchanged_distributions", 0))
so = metrics[prim].get("specified_outcome", {}).get("mcq_specified", {})
for cls in ("A", "C", "E", "ALL"):
s = so.get(cls, {})
vals[f"so_named_{cls}"] = pct(s.get("named_school"))
vals[f"so_divided_{cls}"] = pct(s.get("divided_anyway"))
vals[f"so_other_{cls}"] = pct(s.get("other_position"))
sn = metrics[prim].get("specified_outcome", {}).get("mcq_neutral", {})
for cls in ("A", "B", "C", "E", "ALL"):
vals[f"neut_div_{cls}"] = pct(sn.get(cls, {}).get("divided"))
vals[f"neut_pos_{cls}"] = pct(sn.get(cls, {}).get("a_position"))
# dataset-level facts
lone = 0
for it in items:
if it["cls"] != "A":
continue
maj = set(it["majority_schools"])
other = set()
for p in it["positions"]:
for s in p["schools"]:
if s != it["minority_school"]:
other.add(s)
if it["minority_school"] not in maj and len(other) == 3:
lone += 1
vals["n_lone_outlier"] = str(lone)
# tokenizer
if reprd:
f = reprd["fertility_by_script"]
vp = reprd["vocab_script_profile"]
tm = reprd["term_morphology"]
vals["fert_en"] = num(f["en"]["tokens_per_char"], 4)
vals["fert_ar"] = num(f["ar_plain"]["tokens_per_char"], 4)
vals["fert_arv"] = num(f["ar_vocalized"]["tokens_per_char"], 4)
vals["fert_ur"] = num(f["ur"]["tokens_per_char"], 4)
vals["fert_zh"] = num(f["zh"]["tokens_per_char"], 4)
vals["byte_en"] = pct(f["en"]["byte_level_rate"])
vals["byte_zh"] = pct(f["zh"]["byte_level_rate"])
vals["byte_ur"] = pct(f["ur"]["byte_level_rate"])
vals["byte_ar"] = pct(f["ar_plain"]["byte_level_rate"])
vals["byte_arv"] = pct(f["ar_vocalized"]["byte_level_rate"])
vals["vocab_size"] = f"{vp['vocab_size']:,}"
vals["n_byte_tokens"] = f"{vp['n_byte_level_tokens']:,}"
vals["share_byte_level"] = pct(vp["share_byte_level"], 1)
vals["n_latin_units"] = f"{vp['n_latin_text_units']:,}"
vals["n_cjk_units"] = f"{vp['n_cjk_text_units']:,}"
vals["n_arabic_units"] = str(vp["n_arabic_text_units"])
vals["share_arabic_units"] = pct(vp["share_arabic_text_units"], 3)
vals["max_arabic_chars"] = str(vp["max_arabic_chars_in_one_unit"])
vals["mean_arabic_chars"] = num(vp["mean_arabic_chars_per_unit"], 2)
vals["terms_plain"] = num(tm["mean_tokens_plain"], 2)
vals["terms_voc"] = num(tm["mean_tokens_vocalized"], 2)
vals["terms_ratio"] = num(tm["mean_diacritic_cost_ratio"], 2)
vals["merge_rate"] = pct(reprd["diacritic_discrimination"]["merge_rate"])
vals["ar_x_ferrite"] = num(f["ar_vocalized"]["tokens_per_char"] /
max(f["en"]["tokens_per_char"], 1e-9), 1)
# qualitative examples straight from the raw generations
vals["qual_examples"] = build_qualitative(ROOT, prim)
# per-model param counts
for m, p in (("minicpm5-2b", "2.52B"), ("minicpm5-2b-base", "2.52B"),
("minicpm5-2b-sft", "2.52B"), ("qwen3-4b", "4.02B"),
("smollm3-3b", "3.08B"), ("phi4-mini", "3.82B")):
vals[f"params_{m}"] = p
txt = TEMPLATE.read_text(encoding="utf-8")
def sub(m):
key = m.group(1)
if key not in vals:
raise KeyError(f"template placeholder missing: {key}")
return vals[key]
txt = re.sub(r"\{\{(\w+)\}\}", sub, txt)
OUT.write_text(txt, encoding="utf-8")
print(f"wrote {OUT} ({len(txt)} chars)")
if __name__ == "__main__":
main()