From 0bef8b0d3d0b3e00b43b9cff19c4ef763e8ae3b7 Mon Sep 17 00:00:00 2001 From: wassname <1103714+wassname@users.noreply.github.com> Date: Tue, 23 Jun 2026 09:41:36 +0800 Subject: [PATCH] instrument: fail-fast asserts from external review (#57) deepseek-v4-pro review of the admin->tinymfv port (algebra proven correct, diff 8.25e-08 = float noise). Closes its flagged fail-fast gaps: - per_item_categorical asserts uniform frame count per item (else the per-item average would silently reweight a factor) - reduce_ordinal asserts dimension is not None (no phantom-factor pooling) - administer asserts every ordinal item carries meta['task'] (else build_prompt would silently drop the response-scale legend and the profile would be junk) Also commits maps.py + viz deps from the stage-3 port. Co-Authored-By: Claudypoo <288921227+claudypoo@users.noreply.github.com> --- pyproject.toml | 3 + src/tinymfv/__init__.py | 2 + src/tinymfv/administer.py | 4 + src/tinymfv/instrument.py | 7 + src/tinymfv/maps.py | 308 ++++++++++++++++++++++++++++++++++++++ 5 files changed, 324 insertions(+) create mode 100644 src/tinymfv/maps.py diff --git a/pyproject.toml b/pyproject.toml index 777368b..514a120 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -16,6 +16,9 @@ dependencies = [ "tqdm", "tabulate", "datasets", + "numpy", + "matplotlib>=3.8", # tinymfv.maps: culture-map + range viz + "textalloc>=1.2.3", # non-overlapping label placement on the maps ] [dependency-groups] diff --git a/src/tinymfv/__init__.py b/src/tinymfv/__init__.py index 9d55b00..eed942b 100644 --- a/src/tinymfv/__init__.py +++ b/src/tinymfv/__init__.py @@ -27,6 +27,7 @@ from .instrument import Instrument, InstrItem, per_item_categorical, REDUCERS from .instruments import get as get_instrument, INSTRUMENTS, build_instrument from .read import read_items, resolve_answer_ids, build_prompt from .administer import administer +from . import maps __all__ = [ "CONDITIONS", "CONFIGS", "ConfigName", @@ -37,5 +38,6 @@ __all__ = [ "Instrument", "InstrItem", "per_item_categorical", "REDUCERS", "get_instrument", "INSTRUMENTS", "build_instrument", "read_items", "resolve_answer_ids", "build_prompt", "administer", + "maps", ] diff --git a/src/tinymfv/administer.py b/src/tinymfv/administer.py index 474d6d5..2acec26 100644 --- a/src/tinymfv/administer.py +++ b/src/tinymfv/administer.py @@ -23,6 +23,10 @@ from .read import read_items, resolve_answer_ids def administer(model, tok, instr: Instrument, *, batch_size: int = 36) -> dict: assert instr.kind == "ordinal", "administer() is the ordinal survey readout; use evaluate() for nominal MFV" + # Every ordinal item must carry its frame-specific response-scale legend in meta['task']; without + # it build_prompt would silently emit a bare statement (no legend) and the profile would be junk + # while pmass still looks fine. Fail loud. (External review #57: build_prompt silent fallback.) + assert all("task" in it.meta for it in instr.items), f"{instr.name}: ordinal items need meta['task']" w = np.arange(1, instr.scale_max + 1, dtype=float) answer_ids = resolve_answer_ids(tok, instr.answer_space) per_row = read_items(model, tok, instr, instr.items, answer_ids, diff --git a/src/tinymfv/instrument.py b/src/tinymfv/instrument.py index 43f6f4c..abf3668 100644 --- a/src/tinymfv/instrument.py +++ b/src/tinymfv/instrument.py @@ -118,6 +118,12 @@ def per_item_categorical(per_row: list[dict], kind: Kind) -> dict[str, dict]: by_id: dict[str, list[dict]] = defaultdict(list) for r in per_row: by_id[r["id"]].append(r) + # Each item collapses to ONE averaged distribution, so every item contributes EQUALLY to the + # factor mean. That matches pooling all (item, frame) rows only if every item has the same frame + # count. Assert it loudly rather than silently reweighting a factor if a future instrument gives + # some items fewer frames. (External review #57: dormant weighting asymmetry.) + frame_counts = {len(rows) for rows in by_id.values()} + assert len(frame_counts) == 1, f"heterogeneous frame counts per item: {frame_counts}" out: dict[str, dict] = {} for iid, rows in by_id.items(): canon = [canonicalize_to_forward(r["p"], r["frame"], kind) for r in rows] @@ -159,6 +165,7 @@ def reduce_ordinal(items: dict[str, dict], instr: Instrument) -> np.ndarray: w = np.arange(1, instr.scale_max + 1, dtype=float) by_dim: dict[str, list[float]] = defaultdict(list) for it in items.values(): + assert it["dimension"] is not None, "ordinal item has dimension=None; it would pool into a phantom factor" E = float((it["p"] * w).sum()) agr = (instr.scale_max + 1 - E) if it["sign"] < 0 else E by_dim[it["dimension"]].append(agr) diff --git a/src/tinymfv/maps.py b/src/tinymfv/maps.py new file mode 100644 index 0000000..b979dd0 --- /dev/null +++ b/src/tinymfv/maps.py @@ -0,0 +1,308 @@ +"""Instrument-parameterized culture-map + range viz: the visual proof of the survey eval. + +Two plot families, both pure (numpy arrays + an Instrument in, matplotlib Figure out). All data +loading -- LLM profiles, calibration windows, the human cross-cultural CSVs -- stays in the calling +experiment; this module only draws, so it serves every instrument (MFQ-2 / Big5 / 16PF / HSQ) the +same way. + + plot_ipsative_pca : where the model sits among human cultures. Each society's profile is + row-centred (its overall endorsement level removed) then PCA'd, so the axes + are RELATIVE emphasis across factors, not acquiescence. The model's base + + steered poles are projected into the same space; a compass inset shows how + each factor loads. + plot_range : per-factor, the steer as a directed c-sweep (tail at the -c pole, single + arrowhead at the +c pole) against the strip of human societies. With a zoomed + small-multiple companion (plot_range_zoom), one panel per factor on its own + y-axis. + +Ported from the weight_steer_honesty experiment (mft_honesty.maps / mapviz + scripts/ +fig_profile_sweeps.py). The experiment's MFQ-2-specific overlays (MFT theory axes, delta table) +stay there -- they are about the steering vectors, not the instrument. +""" +from __future__ import annotations +from pathlib import Path + +import numpy as np +import matplotlib.pyplot as plt + +from .instrument import Instrument + +# range-plot palette + geometry (shared with the experiment's prior fig_profile_sweeps look) +CLOUD_GREY = "0.78" # individual respondents (subtle backdrop) +COUNTRY_GREY = "0.38" # society means (the named dots) +MEDIAN_GREY = "0.15" +POS_COL, NEG_COL = "#c0392b", "#2c6fbb" # +c side (red) / -c side (blue) of the sweep +DX_HUMAN, DX_STEER = -0.17, 0.18 +GROUP_PITCH = 1.55 # x-distance between factors; > pair width so each (societies, steer) reads as one unit +# ipsative-map palette +C_BASE, C_HON, C_DIS, C_HUM = "#33688f", "#c2702f", "#8a5a9c", "#888888" + + +def save_both(fig, fig_dir: Path, stem: str, dpi: int = 200) -> Path: + fig_dir.mkdir(parents=True, exist_ok=True) + fig.savefig(fig_dir / f"{stem}.png", dpi=dpi, bbox_inches="tight") + fig.savefig(fig_dir / f"{stem}.svg", bbox_inches="tight") + return fig_dir / f"{stem}.png" + + +# --- ipsative PCA culture map ------------------------------------------------------------------- + +def row_centre_op(K: int) -> np.ndarray: + """Ipsative operator: M @ Pc subtracts each row's own mean across the K factors, removing the + overall-endorsement/acquiescence level that would otherwise dominate PC1.""" + return np.eye(K) - np.ones((K, K)) / K + + +def ipsative_pca(M: np.ndarray, k: int = 2): + """Row-centre each row of M (societies x K), then PCA across rows. + Returns (P, Vt, var, mu, Pc); project a new point v via ((v @ Pc) - mu) @ Vt[:k].T.""" + Pc = row_centre_op(M.shape[1]) + Mp = M @ Pc + mu = Mp.mean(axis=0) + Mc = Mp - mu + _, S, Vt = np.linalg.svd(Mc, full_matrices=False) + var = (S ** 2) / (S ** 2).sum() + return Mc @ Vt[:k].T, Vt, var, mu, Pc + + +def compass(ax_main, L: np.ndarray, labels: list[str], title: str = "compass", + box=(0.62, 0.70, 0.30, 0.27), color: str = "#3a6b35") -> None: + """Compass-rose inset (axes-fraction): each factor is an arrow (direction = 2D loading, length = + magnitude); reference circle at 80% of the shortest arrow so labels sit at the tips.""" + lens = np.linalg.norm(L, axis=1) + L = L / (lens.max() + 1e-9) + circle_r = 0.8 * lens.min() / lens.max() + cax = ax_main.inset_axes(list(box)) + cax.patch.set_alpha(0.0) + cax.add_patch(plt.Circle((0, 0), circle_r, fill=False, color="#bbbbbb", lw=0.7)) + for j, lab in enumerate(labels): + x, y = L[j] + cax.annotate("", xy=(x, y), xytext=(0, 0), arrowprops=dict(arrowstyle="->", color=color, lw=1.1)) + r = np.hypot(x, y) + cax.text(x / r * (r + 0.07), y / r * (r + 0.07), lab.capitalize(), fontsize=8, + fontweight="bold", color=color, ha="left" if x >= 0 else "right", + va="bottom" if y >= 0 else "top", clip_on=False) + cax.set_xlim(-1.5, 1.5); cax.set_ylim(-1.5, 1.5) + cax.set_aspect("equal"); cax.axis("off") + cax.set_title(title, fontsize=10, fontweight="bold", color=color, pad=3) + + +def plot_ipsative_pca(instr: Instrument, dims: list[str], countries: list[str], M: np.ndarray, + base: np.ndarray, hon: np.ndarray | None, dis: np.ndarray | None, + *, boots: dict | None = None, pad=(0.18, 0.16)): + """Ipsative culture map. M is societies x K (0-1 fraction); base/hon/dis are length-K fraction + vectors (or None). `boots` optionally maps 'base'/'honest'/'dis' -> (n x K) bootstrap fraction + matrices for the uncertainty cross. Returns the Figure.""" + try: + import textalloc as ta + except ImportError: + ta = None + P, Vt, var, mu, Pc = ipsative_pca(M) + if Vt[0, 0] < 0: # stabilise SVD sign: factor[0] loads +PC1 + Vt[0] = -Vt[0] + if Vt[1, 1 % len(dims)] < 0: + Vt[1] = -Vt[1] + P = (M @ Pc - mu) @ Vt[:2].T + + def proj(v): + return ((v @ Pc) - mu) @ Vt[:2].T if v is not None else None + pb, ph, pf = proj(base), proj(hon), proj(dis) + + fig, ax = plt.subplots(figsize=(8.5, 7.5)) + ax.set_facecolor("#faf8f2") + ax.grid(True, color="#eceadf", lw=0.3, zorder=0) + ax.scatter(P[:, 0], P[:, 1], s=26, c=C_HUM, alpha=0.7, edgecolors="white", linewidths=0.5, zorder=3) + if ta is not None: + try: + ta.allocate_text(fig, ax, P[:, 0], P[:, 1], countries, x_scatter=P[:, 0], y_scatter=P[:, 1], + textsize=7.5, linecolor="#c2bca8", linewidth=0.5, textcolor="#555555") + except Exception: + ta = None + if ta is None: + for i, c in enumerate(countries): + ax.annotate(c, (P[i, 0], P[i, 1]), fontsize=7, color="#555555", + xytext=(3, 2), textcoords="offset points") + for key, col, pt in [("base", C_BASE, pb), ("honest", C_HON, ph), ("dis", C_DIS, pf)]: + if boots and key in boots and pt is not None: + bp = (np.asarray(boots[key]) @ Pc - mu) @ Vt[:2].T + e1, e2 = 1.96 * bp.std(0) + ax.errorbar(pt[0], pt[1], xerr=e1, yerr=e2, fmt="none", ecolor=col, + elinewidth=0.7, alpha=0.55, capsize=2.5, capthick=0.8, zorder=4) + for pt, col, mk, lab, dxy, ha in [(ph, C_HON, "s", "honest (c=+2)", (9, 9), "left"), + (pf, C_DIS, "^", "dishonest (c=-2)", (-9, -1), "right"), + (pb, C_BASE, "o", "baseline (c=0)", (9, -13), "left")]: + if pt is None: + continue + if pt is not pb: + ax.annotate("", xy=pt, xytext=pb, arrowprops=dict(arrowstyle="-|>", color=col, lw=2.0), zorder=5) + ax.scatter(*pt, s=120, c=col, marker=mk, edgecolors="white", linewidths=1.2, zorder=7) + ax.annotate(lab, pt, xytext=dxy, textcoords="offset points", fontsize=9, color=col, + fontweight="bold", ha=ha, va="center", zorder=8) + compass(ax, Vt[:2].T, dims, title=f"{instr.display} compass") + x0, x1 = ax.get_xlim(); y0, y1 = ax.get_ylim() # modest top-right headroom for the compass inset + ax.set_xlim(x0, x1 + pad[0] * (x1 - x0)); ax.set_ylim(y0, y1 + pad[1] * (y1 - y0)) + ax.set_xlabel(f"PC1 ({var[0]*100:.0f}% var, relative emphasis)") + ax.set_ylabel(f"PC2 ({var[1]*100:.0f}% var, relative emphasis)") + ax.set_title(f"{instr.name}: ipsative culture map ({len(countries)} societies)", fontsize=10) + return fig + + +# --- per-vector steer range --------------------------------------------------------------------- + +def draw_range_panel(ax, instr: Instrument, dims: list[str], cs: list[float], prof: dict, + humans: dict, cloud: np.ndarray | None, vec: str) -> tuple[float, float]: + """One vector's range panel: respondent cloud + named society dots + the steer c-sweep drawn as + a directed -c->+c axis (tail dot at -c, interior dots, single arrowhead at +c). The widest steer + carries a '-N {vec} / base / +N {vec}' in-plot key. Returns the cropped (ymin, ymax).""" + rng = np.random.default_rng(0) + ys: list[float] = [] + spans = [float(np.ptp([prof[c][i] for c in cs])) for i in range(len(dims))] + label_i = int(np.argmax(spans)) if len(cs) >= 2 else -1 + + for i, d in enumerate(dims): + gx = i * GROUP_PITCH + xh = gx + DX_HUMAN + if cloud is not None: + col = cloud[:, i] + jit = (rng.random(len(col)) - 0.5) * 0.30 + yj = (rng.random(len(col)) - 0.5) * 0.17 + ax.scatter(xh + jit, col + yj, s=2, color=CLOUD_GREY, alpha=0.13, edgecolor="none", zorder=0) + means = np.array([m for _, m in humans[d]]) + names = [c for c, _ in humans[d]] + jit2 = (rng.random(len(means)) - 0.5) * 0.16 + ax.scatter(xh + jit2, means, s=24, color=COUNTRY_GREY, alpha=0.9, + edgecolor="white", linewidth=0.3, zorder=3) + ax.plot([xh - 0.15, xh + 0.15], [np.median(means)] * 2, color=MEDIAN_GREY, lw=1.5, zorder=4) + ys += means.tolist() + for idx, va, dy in [(int(means.argmax()), "bottom", 0.03), (int(means.argmin()), "top", -0.03)]: + ax.annotate(names[idx], (xh, means[idx] + dy), fontsize=7, ha="center", va=va, color="0.25", zorder=5) + + xs = gx + DX_STEER + yv = np.array([prof[c][i] for c in cs]) + ys += yv.tolist() + for ca, cb, ya, yb in zip(cs[:-1], cs[1:], yv[:-1], yv[1:]): + ax.plot([xs, xs], [ya, yb], color=(NEG_COL if cb <= 0 else POS_COL), + lw=2.0, zorder=6, solid_capstyle="round") + for k, (c, y) in enumerate(zip(cs, yv)): # tail + interior get dots; +c end is the arrowhead + if k == len(cs) - 1: + continue + ax.plot(xs, y, "o", ms=3.5, zorder=7, + color="black" if c == 0 else (POS_COL if c > 0 else NEG_COL)) + if len(cs) >= 2: + ax.plot(xs, yv[-1], marker=("^" if yv[-1] >= yv[-2] else "v"), + color=POS_COL, ms=6.5, markeredgecolor="none", zorder=8) + if i == label_i: + base_y = float(yv[list(cs).index(0.0)]) + for c_end, y_end in [(cs[0], yv[0]), (0.0, base_y), (cs[-1], yv[-1])]: + txt = "base" if c_end == 0 else f"{int(c_end):+d} {vec}" + ax.annotate(txt, (xs + 0.10, y_end), fontsize=6.8, ha="left", va="center", + color=(MEDIAN_GREY if c_end == 0 else (POS_COL if c_end > 0 else NEG_COL)), zorder=9) + + pad = 0.10 * (max(ys) - min(ys)) + ax.set_ylim(min(ys) - pad, max(ys) + pad) + ax.set_title(f"steer c-sweep {int(min(cs)):+d}..{int(max(cs)):+d} (coherent only); " + f"tail = -c, arrow = +c (more {vec})", fontsize=9) + ax.set_xticks(np.arange(len(dims)) * GROUP_PITCH) + ax.set_xticklabels(dims, rotation=30, ha="right", fontsize=8) + ax.set_xlim(-0.6, (len(dims) - 1) * GROUP_PITCH + 0.6) + ax.grid(axis="y", alpha=0.15) + ax.spines[["top", "right"]].set_visible(False) + return min(ys) - pad, max(ys) + pad + + +def plot_range(instr: Instrument, dims: list[str], cs: list[float], prof: dict, humans: dict, + cloud: np.ndarray | None, vec: str, key_text: str): + """Single-axes range figure for one steering vector. Returns the Figure.""" + figw = max(6.4, 1.5 * len(dims) + 1.5) + fig, ax = plt.subplots(figsize=(figw, 4.8)) + draw_range_panel(ax, instr, dims, cs, prof, humans, cloud, vec) + ax.set_ylabel(f"{instr.display} mean (1-{instr.scale_max})") + fig.suptitle(f"Steered {instr.display} range: {vec}\n{key_text}", fontsize=7.5, y=1.0) + fig.tight_layout() + return fig + + +def plot_range_zoom(instr: Instrument, dims: list[str], cs: list[float], prof: dict, humans: dict, vec: str): + """Zoomed companion: one subplot per factor with its OWN y-axis, so the steer (small vs the + human spread) is legible. Societies near the steer named; off-range extremes in the corners. + Returns the Figure.""" + try: + import textalloc as ta + except ImportError: + ta = None + n = len(dims) + ncol = min(3, n) + nrow = (n + ncol - 1) // ncol + fig, axes = plt.subplots(nrow, ncol, figsize=(3.9 * ncol, 3.2 * nrow), squeeze=False) + rng = np.random.default_rng(0) + for i, d in enumerate(dims): + ax = axes[i // ncol][i % ncol] + yv = np.array([prof[c][i] for c in cs]) + soc = humans[d] + soc_vals = np.array([m for _, m in soc]) + q1, q3 = np.percentile(soc_vals, [25, 75]) + lo, hi = min(yv.min(), q1), max(yv.max(), q3) + m = max(0.10, 0.30 * (hi - lo)) + ylo, yhi = lo - m, hi + m + near = [(name, v) for name, v in soc if ylo <= v <= yhi] + soc_x = (-0.42 + (rng.random(len(near)) - 0.5) * 0.16) if near else np.array([]) + if near: + ax.scatter(soc_x, [v for _, v in near], s=20, color=COUNTRY_GREY, + edgecolor="white", linewidth=0.3, zorder=3) + med = float(np.median(soc_vals)) + if ylo <= med <= yhi: + ax.plot([-0.60, -0.24], [med, med], color=MEDIAN_GREY, lw=1.2, zorder=4) + + xs = 0.30 + for ca, cb, ya, yb in zip(cs[:-1], cs[1:], yv[:-1], yv[1:]): + ax.plot([xs, xs], [ya, yb], color=(NEG_COL if cb <= 0 else POS_COL), lw=2.4, zorder=6, solid_capstyle="round") + for k, (c, y) in enumerate(zip(cs, yv)): # tail + interior get dots; +c end is the arrowhead + if k != len(cs) - 1: + ax.plot(xs, y, "o", ms=4.5, color="black" if c == 0 else (POS_COL if c > 0 else NEG_COL), zorder=7) + if len(cs) >= 2: + ax.plot(xs, yv[-1], marker=("^" if yv[-1] >= yv[-2] else "v"), color=POS_COL, ms=7, markeredgecolor="none", zorder=8) + + base_y = float(yv[list(cs).index(0.0)]) + named = {} + if near: + name_xy = {nm: (float(x), v) for (nm, v), x in zip(near, soc_x)} + for ref in (base_y, float(yv.max()), float(yv.min())): + named[min(near, key=lambda t: abs(t[1] - ref))[0]] = ref + tx = [xs] * len(cs) + [name_xy[nm][0] for nm in named] if near else [xs] * len(cs) + ty = list(map(float, yv)) + [name_xy[nm][1] for nm in named] if near else list(map(float, yv)) + txt = [("c=0" if c == 0 else f"c={int(c):+d}") for c in cs] + list(named) + dot_x = [xs] * len(cs) + (list(soc_x) if near else []) + dot_y = list(map(float, yv)) + ([v for _, v in near] if near else []) + placed = False + if ta is not None: + try: + ta.allocate_text(fig, ax, tx, ty, txt, x_scatter=dot_x, y_scatter=dot_y, + textsize=6.5, linecolor="#bbbbbb", linewidth=0.4, textcolor="#333333") + placed = True + except Exception: + placed = False + if not placed: + for x, y, t in zip(tx, ty, txt): + ax.annotate(t, (x + (0.12 if x >= 0 else -0.12), y), fontsize=6.5, + ha="left" if x >= 0 else "right", va="center", color="#333333") + mx_name, mx_val = max(soc, key=lambda t: t[1]) + mn_name, mn_val = min(soc, key=lambda t: t[1]) + if mx_val > yhi: + ax.text(0.02, 0.99, f"↑ {mx_name} {mx_val:.2f}", transform=ax.transAxes, + fontsize=6.3, ha="left", va="top", color=MEDIAN_GREY) + if mn_val < ylo: + ax.text(0.02, 0.01, f"↓ {mn_name} {mn_val:.2f}", transform=ax.transAxes, + fontsize=6.3, ha="left", va="bottom", color=MEDIAN_GREY) + ax.set_title(d, fontsize=9) + ax.set_xlim(-1.05, 1.25) + ax.set_ylim(ylo, yhi) + ax.set_xticks([]) + ax.tick_params(labelsize=7) + ax.spines[["top", "right", "bottom"]].set_visible(False) + for k in range(n, nrow * ncol): + axes[k // ncol][k % ncol].axis("off") + fig.suptitle(f"Steered {instr.display}, zoomed per subscale: {vec}\n" + f"grey = nearby societies (named: nearest the base / +end / -end; corner ^v = the 2 off-range " + f"extremes); coloured line = steer (tail = -c, arrow = +c); each panel has its OWN y-axis", fontsize=8) + fig.tight_layout() + return fig