Download code/figures.py from adeerkhan/msq-plurality-gap: direct link, hf CLI and curl.
- Browser
- Download file 13.2 kB
-
https://huggingface.co/adeerkhan/msq-plurality-gap/resolve/main/code/figures.py
- Command line
-
hf download hf://adeerkhan/msq-plurality-gap/code/figures.py
-
curl -L -o figures.py https://huggingface.co/adeerkhan/msq-plurality-gap/resolve/main/code/figures.py
13.2 kB
| #!/usr/bin/env python3 | |
| """Figures for the MSQ study. Reads results/metrics.json and results/raw/*.jsonl.""" | |
| from __future__ import annotations | |
| import json | |
| import pathlib | |
| import sys | |
| import matplotlib | |
| matplotlib.use("Agg") | |
| import matplotlib.pyplot as plt | |
| import numpy as np | |
| sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent)) | |
| import msq # noqa: E402 | |
| ROOT = pathlib.Path(__file__).resolve().parent.parent | |
| RES = ROOT / "results" | |
| FIG = ROOT / "figures" | |
| INK = "#1f2933" | |
| ACCENT = "#1d6fd0" | |
| WARM = "#e85d4c" | |
| TEAL = "#2a9d8f" | |
| AMBER = "#e09f3e" | |
| GREY = "#9aa5b1" | |
| def style(ax): | |
| ax.spines["top"].set_visible(False) | |
| ax.spines["right"].set_visible(False) | |
| ax.spines["left"].set_color(GREY) | |
| ax.spines["bottom"].set_color(GREY) | |
| ax.tick_params(colors=INK, labelsize=9) | |
| ax.grid(axis="y", color="#e4e7eb", lw=0.8) | |
| ax.set_axisbelow(True) | |
| def load_metrics(): | |
| p = RES / "metrics.json" | |
| return json.loads(p.read_text()) if p.exists() else {} | |
| def load_raw(model): | |
| p = RES / "raw" / f"{model}.jsonl" | |
| if not p.exists(): | |
| return [] | |
| return [json.loads(l) for l in p.open(encoding="utf-8") if l.strip()] | |
| def fig_prr_msq(metrics, primary): | |
| """PRR and MSQ by class, for every model: the three-symptom structure.""" | |
| models = [m for m in metrics if metrics[m].get("mcq")] | |
| if not models: | |
| return | |
| classes = ["A", "B", "C", "E"] | |
| labels = ["A\nfiqh\n(madhhab)", "B\nlive\ncontroversy", "C\njurisdiction", | |
| "E\nPython\nversion"] | |
| fig, axes = plt.subplots(1, 3, figsize=(13.0, 4.3)) | |
| width = 0.8 / max(len(models), 1) | |
| def draw(ax, section, key, title, ylabel): | |
| for k, m in enumerate(models): | |
| vals = [] | |
| for c in classes: | |
| v = metrics[m]["mcq"].get(section, {}).get(c, {}).get(key) | |
| vals.append((v or 0) * 100) | |
| xs = [i + (k - (len(models) - 1) / 2) * width for i in range(len(classes))] | |
| ax.bar(xs, vals, width=width * 0.9, | |
| color=ACCENT if m == primary else TEAL, | |
| label=m if k == 0 else None) | |
| ax.set_xticks(range(len(classes))) | |
| ax.set_xticklabels(labels, fontsize=8) | |
| ax.set_title(title, fontsize=10, color=INK, pad=10, loc="left") | |
| ax.set_ylabel(ylabel) | |
| ax.set_ylim(0, 108) | |
| style(ax) | |
| draw(axes[0], "mcq_neutral", "plurality_recognition_rate", | |
| "Does it NOTICE the question is conditional?", "% choosing 'it depends'") | |
| draw(axes[1], "mcq_specified", "school_adherence_rate", | |
| "Does it USE the context it was given?", "% following the named context") | |
| ax = axes[2] | |
| for k, m in enumerate(models): | |
| cr = metrics[m].get("context_response", {}).get("ALL", {}) | |
| v = (cr.get("mean_context_response_tv") or 0) | |
| ax.bar([k], [v], width=0.5, color=ACCENT if m == primary else TEAL) | |
| ax.text(k, v + 0.012, f"{v:.3f}", ha="center", fontsize=9, color=INK) | |
| u = cr.get("argmax_unchanged_rate") | |
| if u is not None: | |
| ax.text(k, 0.012, f"{u*100:.0f}% unchanged", ha="center", fontsize=8, | |
| color="white", rotation=90, va="bottom") | |
| ax.set_xticks(range(len(models))) | |
| ax.set_xticklabels([m.replace("-", "\n") for m in models], fontsize=8) | |
| ax.set_title("How far does naming the context move the answer?", | |
| fontsize=10, color=INK, pad=10, loc="left") | |
| ax.set_ylabel("total-variation distance between the two conditions") | |
| ax.set_ylim(0, 0.55) | |
| style(ax) | |
| handles = [plt.Rectangle((0, 0), 1, 1, color=ACCENT), | |
| plt.Rectangle((0, 0), 1, 1, color=TEAL)] | |
| fig.legend(handles, ["primary", "comparison models"], loc="upper right", | |
| frameon=False, fontsize=9, ncol=2) | |
| fig.tight_layout() | |
| fig.savefig(FIG / "fig1_prr_msq.png", dpi=170, bbox_inches="tight") | |
| plt.close(fig) | |
| def fig_disclosure(metrics, primary): | |
| if primary not in metrics: | |
| return | |
| ft = metrics[primary]["free_text"] | |
| pc = metrics[primary].get("positions_contained", {}) | |
| probes = ["ft_default", "ft_repair", "ft_technical", "ft_two_schools", "ft_pair"] | |
| labels = ["plain\nquestion", "one-line\nrepair", "technical\nregister", | |
| "asked to\nenumerate", "asked for two\nnamed schools"] | |
| series = [ | |
| ("disclosure_rate", "signals that the answer is context-dependent", ACCENT, None), | |
| ("plural_attribution_rate", "names two or more competing schools", WARM, None), | |
| ("mean_positions_contained", "positions actually carried", TEAL, "pc"), | |
| ] | |
| fig, axes = plt.subplots(1, 3, figsize=(12.4, 4.0), sharey=False) | |
| for ax, (key, title, col, kind) in zip(axes, series): | |
| vals = [] | |
| for p in probes: | |
| if kind == "pc": | |
| v = pc.get(p, {}).get("ALL", {}).get("mean_positions_contained") | |
| else: | |
| v = ft.get(p, {}).get("ALL", {}).get(key) | |
| vals.append(v or 0) | |
| if kind == "pc": | |
| vals = [v * 2 for v in vals] # scale to a 0-2 "positions" axis | |
| top = 2.1 | |
| else: | |
| vals = [v * 100 for v in vals] | |
| top = 105 | |
| bars = ax.bar(range(len(probes)), vals, color=col, width=0.62) | |
| for b, v in zip(bars, vals): | |
| ax.text(b.get_x() + b.get_width() / 2, v + top * 0.02, | |
| f"{v:.2f}" if kind == "pc" else f"{v:.0f}%", | |
| ha="center", fontsize=8.5, color=INK) | |
| ax.set_xticks(range(len(probes))) | |
| ax.set_xticklabels(labels, fontsize=7.6) | |
| ax.set_title(title, fontsize=9.5, color=INK, pad=10, loc="left") | |
| ax.set_ylim(0, top) | |
| if kind == "pc": | |
| ax.set_yticks([0, 1, 2]) | |
| ax.set_yticklabels(["0", "1", "2"]) | |
| style(ax) | |
| axes[0].set_ylabel("% of answers") | |
| fig.suptitle(f"Disclosure behaviour on contested questions · {primary}", | |
| fontsize=11.5, color=INK, x=0.005, ha="left", y=1.02) | |
| fig.tight_layout() | |
| fig.savefig(FIG / "fig2_disclosure.png", dpi=170, bbox_inches="tight") | |
| plt.close(fig) | |
| def fig_confidence(metrics, primary): | |
| if primary not in metrics: | |
| return | |
| mn = metrics[primary]["mcq"]["mcq_neutral"] | |
| so = metrics[primary].get("specified_outcome", {}).get("mcq_specified", {}) | |
| cr = metrics[primary].get("context_response", {}) | |
| classes = ["A", "B", "C", "E"] | |
| labels = ["A\nfiqh", "B\ncontroversy", "C\njurisdiction", "E\nPython\nversion"] | |
| fig, axes = plt.subplots(1, 3, figsize=(13.4, 4.2)) | |
| # panel 1 -- what it picks once the context is named | |
| ax = axes[0] | |
| keys = [("named_school", "the context it was given", ACCENT), | |
| ("divided_anyway", "'it depends' anyway", AMBER), | |
| ("other_position", "some other position", GREY)] | |
| bottoms = [0.0] * len(classes) | |
| for key, name, col in keys: | |
| vals = [] | |
| for c in classes: | |
| v = so.get(c, {}).get(key) | |
| vals.append((v or 0) * 100) | |
| ax.bar(range(len(classes)), vals, bottom=bottoms, color=col, width=0.6, | |
| label=name) | |
| bottoms = [b + v for b, v in zip(bottoms, vals)] | |
| ax.set_xticks(range(len(classes))) | |
| ax.set_xticklabels(labels, fontsize=8.5) | |
| ax.set_ylabel("% of items") | |
| ax.set_title("Once the context IS named, what does it pick?", fontsize=10, | |
| color=INK, pad=10, loc="left") | |
| ax.set_ylim(0, 100) | |
| ax.legend(fontsize=8, frameon=False, loc="upper center", | |
| bbox_to_anchor=(0.5, -0.18), ncol=1) | |
| style(ax) | |
| # panel 2 -- how far the instruction moved the answer | |
| ax = axes[1] | |
| for c, col in zip(classes, [WARM, AMBER, TEAL, "#8e7dbe"]): | |
| d = cr.get(c, {}).get("median_delta_on_named_position") | |
| if d is None: | |
| continue | |
| ax.bar([classes.index(c)], [d], width=0.55, color=col) | |
| ax.text(classes.index(c), d + 0.004, f"{d:.3f}", ha="center", fontsize=9, | |
| color=INK) | |
| ax.set_xticks(range(len(classes))) | |
| ax.set_xticklabels(labels, fontsize=8.5) | |
| ax.set_ylabel("median change in probability on the correct answer") | |
| ax.set_title("Naming the context barely moves the answer", fontsize=10, | |
| color=INK, pad=10, loc="left") | |
| ax.set_ylim(0, 0.42) | |
| style(ax) | |
| # panel 3 -- confidence | |
| ax = axes[2] | |
| allc = classes + ["D"] | |
| vals = [(mn.get(c, {}).get("mean_max_prob") or 0) * 100 for c in allc] | |
| cols = [WARM, AMBER, TEAL, "#8e7dbe", GREY] | |
| bars = ax.bar(range(len(allc)), vals, color=cols, width=0.6) | |
| for b, v in zip(bars, vals): | |
| ax.text(b.get_x() + b.get_width() / 2, v + 1.5, f"{v:.0f}%", ha="center", | |
| fontsize=9.5, color=INK) | |
| ax.set_xticks(range(len(allc))) | |
| ax.set_xticklabels(["A\nfiqh", "B\ncontroversy", "C\njurisdiction", | |
| "E\nPython", "D\ncontrol"], fontsize=8.5) | |
| ax.set_ylabel("mean confidence in its own chosen option") | |
| ax.set_title("Confidence never drops", fontsize=10, color=INK, pad=10, loc="left") | |
| ax.set_ylim(0, 112) | |
| style(ax) | |
| fig.tight_layout() | |
| fig.savefig(FIG / "fig3_confidence_prr.png", dpi=170, bbox_inches="tight") | |
| plt.close(fig) | |
| def fig_per_item(metrics, primary): | |
| recs = load_raw(primary) | |
| if not recs: | |
| return | |
| items = {i["id"]: i for i in msq.load_items()} | |
| rows = [r for r in recs if r["probe"] == "mcq_neutral" and r["cls"] in ("A", "B")] | |
| if not rows: | |
| return | |
| rows.sort(key=lambda r: r["item_id"]) | |
| fig, ax = plt.subplots(figsize=(9.6, 5.2)) | |
| xs, ys, cs = [], [], [] | |
| for r in rows: | |
| it = items[r["item_id"]] | |
| divided = max((i for i, o in enumerate(r["options"]) | |
| if o.lower().startswith(("the scholars are divided", | |
| "this is a contested matter", | |
| "it depends on the jurisdiction"))), | |
| default=None) | |
| if divided is None: | |
| continue | |
| xs.append(r["item_id"]) | |
| ys.append(r["probs"][divided] * 100) | |
| arg = r["argmax_index"] | |
| cs.append("#2a9d8f" if arg == divided else "#e85d4c") | |
| ax.bar(range(len(xs)), ys, color=cs, width=0.7) | |
| ax.set_xticks(range(len(xs))) | |
| ax.set_xticklabels(xs, rotation=90, fontsize=7) | |
| ax.set_ylabel("probability assigned to the 'divided' option") | |
| ax.set_title("Per-item probability of the plurality answer (teal = the model " | |
| "chose it; red = it chose one school)", | |
| fontsize=10.5, color=INK, pad=10, loc="left") | |
| style(ax) | |
| fig.tight_layout() | |
| fig.savefig(FIG / "fig4_per_item.png", dpi=170) | |
| plt.close(fig) | |
| def fig_tokenizer(): | |
| p = RES / "repr_probe.json" | |
| if not p.exists(): | |
| return | |
| rep = json.loads(p.read_text()) | |
| fert = rep.get("fertility_by_script", {}) | |
| names = {"en": "English", "zh": "Chinese", "ar_vocalized": "Arabic\n(vocalised)", | |
| "ar_plain": "Arabic\n(plain)", "ur": "Urdu", "fa": "Persian", | |
| "hi": "Hindi", "arabizi": "Romanised\nArabic"} | |
| fig, axes = plt.subplots(1, 2, figsize=(10.8, 4.2)) | |
| keys = [k for k in names if k in fert] | |
| ax = axes[0] | |
| vals = [fert[k]["tokens_per_char"] * 100 for k in keys] | |
| colors = [] | |
| for k in keys: | |
| if k.startswith("ar"): | |
| colors.append(WARM) | |
| elif k in ("ur", "fa", "hi"): | |
| colors.append(TEAL) | |
| else: | |
| colors.append(GREY) | |
| bars = ax.bar(range(len(keys)), vals, color=colors, width=0.62) | |
| for b, v in zip(bars, vals): | |
| ax.text(b.get_x() + b.get_width() / 2, v + 0.6, f"{v:.1f}", ha="center", | |
| fontsize=8.5, color=INK) | |
| ax.set_xticks(range(len(keys))) | |
| ax.set_xticklabels([names[k] for k in keys], fontsize=8) | |
| ax.set_ylabel("tokens per 100 characters") | |
| ax.set_title("Token cost by script", fontsize=10.5, color=INK, pad=10, loc="left") | |
| style(ax) | |
| ax = axes[1] | |
| keys2 = [k for k in keys] | |
| vals2 = [fert[k]["byte_level_rate"] * 100 for k in keys2] | |
| bars = ax.bar(range(len(keys2)), vals2, color=colors, width=0.62) | |
| for b, v in zip(bars, vals2): | |
| ax.text(b.get_x() + b.get_width() / 2, v + 1.2, f"{v:.0f}%", ha="center", | |
| fontsize=8.5, color=INK) | |
| ax.set_xticks(range(len(keys2))) | |
| ax.set_xticklabels([names[k] for k in keys2], fontsize=8) | |
| ax.set_ylabel("% of tokens that are raw byte fragments") | |
| ax.set_title("Where the vocabulary gives up", fontsize=10.5, color=INK, pad=10, | |
| loc="left") | |
| ax.set_ylim(0, 68) | |
| style(ax) | |
| fig.tight_layout() | |
| fig.savefig(FIG / "fig5_tokenizer.png", dpi=170) | |
| plt.close(fig) | |
| def main(): | |
| FIG.mkdir(parents=True, exist_ok=True) | |
| metrics = load_metrics() | |
| primary = "minicpm5-2b" if "minicpm5-2b" in metrics else next(iter(metrics), None) | |
| if metrics: | |
| fig_prr_msq(metrics, primary) | |
| fig_disclosure(metrics, primary) | |
| fig_confidence(metrics, primary) | |
| fig_per_item(metrics, primary) | |
| fig_tokenizer() | |
| print("figures:", sorted(p.name for p in FIG.glob("*.png"))) | |
| if __name__ == "__main__": | |
| main() | |