From ecd2affac47f42b3effc8ee0717c07cb88467af9 Mon Sep 17 00:00:00 2001
From: wassname <1103714+wassname@users.noreply.github.com>
Date: Thu, 25 Jun 2026 19:14:04 +0800
Subject: [PATCH] ordinal readouts: keep raw lp, add sensitive logit contrast C
+ log-odds
The default ordinal readout was the expected Likert score E = sum k*p_k, which is
insensitive to steering: dE/dl_j = p_j(j-E) vanishes when the model answers confidently
(peaked at the mode), so a steer that reallocates the tails barely moves E. read.py threw
away the raw logprobs after renormalizing, so nothing downstream could recover the signal.
- read.py keeps the raw lp_gather (the primitive) + the think traces on every row.
- readouts.py: pure functions of lp -- expected_score E (human-comparable), logit_contrast
C = sum (k-mid)*lp_k (primary steer signal: dC/dl_j = w_j, no p_j suppression, normalizer-
invariant, dC = w.dl exactly), agree_logodds LO (readable 2-bin direction), entropy.
- per_item_categorical also frame-averages the logprobs (exact for the linear contrast).
- administer returns profile_C alongside profile_E, per-item E/C/LO/entropy with bootstrap
CIs for both, and the raw per-(item,frame) rows with lp + think for downstream reconstruction.
Unit check: on a peaked-at-4 dist under a small disagree steer, dE=-0.11 but dC=-1.80
(=w.dl exactly) and dLO=-0.60; C identical on raw logits vs renormalized logprobs.
Co-Authored-By: Claudypoo <288921227+claudypoo@users.noreply.github.com>
---
src/tinymfv/__init__.py | 3 ++
src/tinymfv/administer.py | 111 ++++++++++++++++++++++++++------------
src/tinymfv/instrument.py | 13 +++--
src/tinymfv/read.py | 13 +++--
src/tinymfv/readouts.py | 68 +++++++++++++++++++++++
5 files changed, 166 insertions(+), 42 deletions(-)
create mode 100644 src/tinymfv/readouts.py
diff --git a/src/tinymfv/__init__.py b/src/tinymfv/__init__.py
index 8c6fe0d..770b2d4 100644
--- a/src/tinymfv/__init__.py
+++ b/src/tinymfv/__init__.py
@@ -26,6 +26,7 @@ from .guided import guided_rollout_forced_choice, _DEFAULT_FORCED_FOUNDATIONS
from .instrument import Instrument, InstrItem, per_item_categorical
from .instruments import get as get_instrument, INSTRUMENTS, build_instrument
from .read import read_items, resolve_answer_ids, build_user_content
+from .readouts import expected_score, logit_contrast, agree_logodds, entropy
from .administer import administer
@@ -44,6 +45,8 @@ def __getattr__(name: str):
__all__ = [
# entrypoints
"evaluate", "administer", "get_instrument", "read_items",
+ # ordinal readouts (pure functions of the raw answer-token logprobs)
+ "expected_score", "logit_contrast", "agree_logodds", "entropy",
# types consumers build / subset
"Instrument", "InstrItem",
# data API
diff --git a/src/tinymfv/administer.py b/src/tinymfv/administer.py
index 74c3298..1563cc0 100644
--- a/src/tinymfv/administer.py
+++ b/src/tinymfv/administer.py
@@ -21,13 +21,19 @@ import numpy as np
from .instrument import Instrument, per_item_categorical, reduce_ordinal, canonicalize_to_forward
from .read import read_items, resolve_answer_ids
+from .readouts import expected_score, logit_contrast, agree_logodds, entropy
class ItemRow(TypedDict):
id: str
foundation: str
- keyed_agreement: float # E, reverse-keyed for sign<0 items
- E: float # expected scale point, 1..scale_max
+ E: float # expected scale point, 1..scale_max (human-comparable, insensitive)
+ keyed_E: float # E reverse-keyed for sign<0 items (== old keyed_agreement)
+ C: float # rank-centered logit contrast (primary steer signal)
+ keyed_C: float # C reverse-keyed (negated) for sign<0 items
+ logodds: float # agree-vs-disagree log-odds (readable direction summary)
+ keyed_logodds: float
+ entropy: float # within-allowed entropy, nats (coherence the pmass gate misses)
pmass_allowed: float
frame_spread: float
@@ -36,18 +42,27 @@ class ItemFrameRow(TypedDict):
id: str
framing: str # forward | inverted | negated
foundation: str
- agreement: float # forward-canonicalized E toward the original statement
- keyed_agreement: float
+ lp: list[float] # raw logprobs at the M scale tokens (presented orientation)
+ E: float # forward-canonicalized E toward the original statement
+ C: float # forward-canonicalized logit contrast
+ keyed_E: float
+ keyed_C: float
pmass_allowed: float
+ think: str # the model's think trace for this (item, frame)
+ n_think: int
+ emitted_close: bool
class AdministerResult(TypedDict):
- profile: np.ndarray # [len(dimensions)] per-factor keyed agreement -- the map input
- dimensions: list[str] # factor order, matches `profile`
- foundations: list[dict] # one per factor: foundation, mean, sd, ci95_lo/hi, framing_spread,
- # + dynamic f_ keys (f_forward/f_inverted/f_negated)
- per_item: list[ItemRow] # one per item, frame-averaged
- per_item_frame: list[ItemFrameRow] # one per (item, frame) -- the granularity the maps bootstrap
+ # profiles: per-factor means in factor order (`dimensions`). E is the human-comparable score;
+ # C (the rank-centered logit contrast) is the steering-legible one (E saturates at confidence).
+ profile_E: np.ndarray # [len(dimensions)] per-factor keyed E (the human-comparison map input)
+ profile_C: np.ndarray # [len(dimensions)] per-factor keyed contrast (the steer map input)
+ profile: np.ndarray # alias of profile_E (kept so existing E maps keep working)
+ dimensions: list[str] # factor order, matches the profiles
+ foundations: list[dict] # one per factor: foundation, mean(E), C, logodds, sd, ci95*, f_
+ per_item: list[ItemRow] # one per item, frame-averaged readouts
+ per_item_frame: list[ItemFrameRow] # one per (item, frame): raw lp + think + readouts
mean_pmass_allowed: float # coherence check (mass on valid answer tokens)
@@ -58,7 +73,6 @@ def administer(model, tok, instr: Instrument, *, batch_size: int = 36,
# it build_user_content would silently emit a bare statement (no legend) and the profile would be
# junk while pmass still looks fine. Fail loud.
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)
# max_think_tokens=64 is the spec's "light" default: the model thinks before the prefilled answer
# slot, so an activation steer accrues over the trace before being read. Floor is 1 (the shared
@@ -67,47 +81,74 @@ def administer(model, tok, instr: Instrument, *, batch_size: int = 36,
max_think_tokens=max_think_tokens, batch_size=batch_size, verbose_first=True)
items = per_item_categorical(per_row, instr.kind) # {id: {p, pmass, dimension, sign, ...}}
- profile = reduce_ordinal(items, instr) # per-factor keyed agreement
+ M = instr.scale_max
+ profile_E = reduce_ordinal(items, instr) # per-factor keyed E (human comparison)
mean_pmass = float(np.mean([it["pmass"] for it in items.values()]))
- # per-item frame-averaged keyed agreement (for the maps' bootstrap uncertainty cross)
+ # per-item frame-averaged readouts. E and entropy come from the averaged probability vector (so the
+ # NaN-at-collapse signal survives); C and log-odds come from the averaged logprobs (the sensitive
+ # log-space readouts). Reverse-keying (sign<0): E reflects to M+1-E, while the midpoint-centered
+ # contrast C and the log-odds simply negate (reflecting the scale negates a centered weight).
per_item_rows = []
for iid, it in items.items():
- E = float((it["p"] * w).sum())
- keyed = (instr.scale_max + 1 - E) if it["sign"] < 0 else E
- per_item_rows.append({"id": iid, "foundation": it["dimension"], "keyed_agreement": keyed,
- "E": E, "pmass_allowed": it["pmass"], "frame_spread": it["frame_spread"]})
+ lp, p, sign = it["lp"], it["p"], it["sign"]
+ E, Cval, LO = expected_score(p, M), logit_contrast(lp, M), agree_logodds(lp, M)
+ per_item_rows.append({
+ "id": iid, "foundation": it["dimension"],
+ "E": E, "keyed_E": (M + 1 - E) if sign < 0 else E,
+ "C": Cval, "keyed_C": -Cval if sign < 0 else Cval,
+ "logodds": LO, "keyed_logodds": -LO if sign < 0 else LO,
+ "entropy": entropy(p, M), "pmass_allowed": it["pmass"], "frame_spread": it["frame_spread"],
+ })
- # per-frame factor means + framing spread (acquiescence/wording diagnostic): for each frame,
- # canonicalize that frame's presented distribution to forward, key it, pool per factor.
- # Also keep the per-(item, frame) rows: experiment analyses (e.g. the MFQ-2 map's framing-bias
- # diagnostic + paired base-vs-steer delta) need the per-framing granularity that per_item
- # averages away. agreement = forward-canonicalized E (agreement toward the original statement);
- # keyed_agreement reflects reverse-keyed items, same as reduce_ordinal.
+ # per-(item, frame) rows: the raw granularity (lp + think + per-frame readouts) downstream analyses
+ # need -- framing-bias diagnostic, paired base-vs-steer deltas, the pro-vs-anti interaction control.
+ # E/C are forward-canonicalized toward the original statement; keyed_* reflect reverse-keyed items.
frames = sorted({r["frame"] for r in per_row})
- by_dim_frame: dict[tuple[str, str], list[float]] = {}
+ by_dim_frame: dict[tuple[str, str], list[float]] = {} # keyed E per (factor, frame): framing diagnostic
per_item_frame: list[dict] = []
for r in per_row:
+ sign = r["sign"]
p_fwd = canonicalize_to_forward(r["p"], r["frame"], instr.kind)
- E = float((p_fwd * w).sum())
- keyed = (instr.scale_max + 1 - E) if r["sign"] < 0 else E
- by_dim_frame.setdefault((r["dimension"], r["frame"]), []).append(keyed)
- per_item_frame.append({"id": r["id"], "framing": r["frame"], "foundation": r["dimension"],
- "agreement": E, "keyed_agreement": keyed,
- "pmass_allowed": r["pmass_allowed"]})
+ lp_fwd = canonicalize_to_forward(r["lp"], r["frame"], instr.kind)
+ E, Cval = expected_score(p_fwd, M), logit_contrast(lp_fwd, M)
+ by_dim_frame.setdefault((r["dimension"], r["frame"]), []).append((M + 1 - E) if sign < 0 else E)
+ per_item_frame.append({
+ "id": r["id"], "framing": r["frame"], "foundation": r["dimension"],
+ "lp": list(map(float, r["lp"])),
+ "E": E, "C": Cval,
+ "keyed_E": (M + 1 - E) if sign < 0 else E,
+ "keyed_C": -Cval if sign < 0 else Cval,
+ "pmass_allowed": r["pmass_allowed"],
+ "think": r["think"], "n_think": r["n_think"], "emitted_close": r["emitted_close"],
+ })
+ # per-factor means + bootstrap CIs for BOTH E and C (the uncertainty the user wants alongside the
+ # mean), the mean log-odds, and the per-frame E means (framing-bias diagnostic).
rng = np.random.default_rng(0)
+ def _ci(vals: np.ndarray) -> tuple[float, float]:
+ boot = rng.choice(vals, size=(2000, len(vals)), replace=True).mean(axis=1)
+ return float(np.percentile(boot, 2.5)), float(np.percentile(boot, 97.5))
+ profile_C = np.zeros(len(instr.dimensions))
foundations = []
for j, d in enumerate(instr.dimensions):
- vals = np.array([row["keyed_agreement"] for row in per_item_rows if row["foundation"] == d])
- boot = rng.choice(vals, size=(2000, len(vals)), replace=True).mean(axis=1)
+ e_vals = np.array([row["keyed_E"] for row in per_item_rows if row["foundation"] == d])
+ c_vals = np.array([row["keyed_C"] for row in per_item_rows if row["foundation"] == d])
+ lo_vals = np.array([row["keyed_logodds"] for row in per_item_rows if row["foundation"] == d])
+ profile_C[j] = float(np.mean(c_vals))
+ e_lo, e_hi = _ci(e_vals); c_lo, c_hi = _ci(c_vals)
per_fr = {fr: float(np.mean(by_dim_frame[(d, fr)])) for fr in frames}
foundations.append({
- "foundation": d, "mean": float(profile[j]), "sd": float(vals.std(ddof=1)),
- "ci95_lo": float(np.percentile(boot, 2.5)), "ci95_hi": float(np.percentile(boot, 97.5)),
+ "foundation": d,
+ "mean": float(profile_E[j]), "sd": float(e_vals.std(ddof=1)),
+ "ci95_lo": e_lo, "ci95_hi": e_hi,
+ "C": float(profile_C[j]), "C_sd": float(c_vals.std(ddof=1)),
+ "C_ci95_lo": c_lo, "C_ci95_hi": c_hi,
+ "logodds": float(np.mean(lo_vals)),
"framing_spread": float(max(per_fr.values()) - min(per_fr.values())),
**{f"f_{fr}": v for fr, v in per_fr.items()},
})
- return {"profile": profile, "dimensions": instr.dimensions, "foundations": foundations,
+ return {"profile_E": profile_E, "profile_C": profile_C, "profile": profile_E,
+ "dimensions": instr.dimensions, "foundations": foundations,
"per_item": per_item_rows, "per_item_frame": per_item_frame,
"mean_pmass_allowed": mean_pmass}
diff --git a/src/tinymfv/instrument.py b/src/tinymfv/instrument.py
index 7413080..b18e002 100644
--- a/src/tinymfv/instrument.py
+++ b/src/tinymfv/instrument.py
@@ -112,10 +112,13 @@ def canonicalize_to_forward(p: np.ndarray, frame: str, kind: Kind) -> np.ndarray
def per_item_categorical(per_row: list[dict], kind: Kind) -> dict[str, dict]:
"""Collapse (item, frame) rows to one forward-orientation categorical per item id.
- Each row has: id, frame, p (renormalized over answer_space, sums to 1), pmass_allowed,
- dimension, sign, human_label. Returns {id: {p, pmass, dimension, sign, human_label, n_frames,
- frame_spread}} where p is the mean of canonicalized frame distributions and frame_spread is
- the max L1 gap between any two canonical frames (the acquiescence/negation diagnostic).
+ Each row has: id, frame, lp (raw logprobs at the M tokens), p (renormalized over answer_space),
+ pmass_allowed, dimension, sign, human_label. Returns {id: {lp, p, pmass, dimension, sign,
+ human_label, n_frames, frame_spread}} where p is the mean of canonicalized frame probability
+ vectors (kept for E + the human-comparison maps, preserving the NaN-at-collapse signal), lp is
+ the mean of canonicalized frame logprobs (the log-space primitive for the contrast C + log-odds;
+ averaging in log space is exact for the linear contrast), and frame_spread is the max L1 gap
+ between any two canonical probability frames (the acquiescence/negation diagnostic).
"""
by_id: dict[str, list[dict]] = defaultdict(list)
for r in per_row:
@@ -134,11 +137,13 @@ def per_item_categorical(per_row: list[dict], kind: Kind) -> dict[str, dict]:
assert len({r.get("sign", 1) for r in rows}) == 1, f"{iid}: inconsistent sign across frames"
assert len({r["frame"] for r in rows}) == len(rows), f"{iid}: duplicate frame in rows"
canon = [canonicalize_to_forward(r["p"], r["frame"], kind) for r in rows]
+ canon_lp = [canonicalize_to_forward(r["lp"], r["frame"], kind) for r in rows]
C = np.stack(canon)
spread = float(max((np.abs(C[i] - C[j]).sum()
for i in range(len(C)) for j in range(i + 1, len(C))), default=0.0))
r0 = rows[0]
out[iid] = {
+ "lp": np.stack(canon_lp).mean(axis=0),
"p": C.mean(axis=0),
"pmass": float(np.mean([r["pmass_allowed"] for r in rows])),
"dimension": r0.get("dimension"),
diff --git a/src/tinymfv/read.py b/src/tinymfv/read.py
index 3935fd4..5d863b2 100644
--- a/src/tinymfv/read.py
+++ b/src/tinymfv/read.py
@@ -84,7 +84,7 @@ def read_items(model, tok, instr: Instrument, items: list[InstrItem], answer_ids
# frame debias is downstream in canonicalize_to_forward). force_only: the "(" prefill is too
# short for natural-emission detection (matches by chance in the think trace), so always read
# the forced answer slot. n_samples=1, temperature=0 -> deterministic.
- _thinks, slots = _rollout_natural_or_forced(
+ thinks, slots = _rollout_natural_or_forced(
model, tok, user_prompts,
schema_hint="", max_think_tokens=max_think_tokens,
scoring_slots=[("Just answer", instr.prefill)],
@@ -94,8 +94,12 @@ def read_items(model, tok, instr: Instrument, items: list[InstrItem], answer_ids
)
for j, it in enumerate(chunk):
slot = slots[j][0]
- # lp_gather[k] is the full-vocab log_softmax logprob of answer token k at the answer slot.
- p_a = np.exp(np.asarray(slot["lp_gather"], dtype=float)) # [A] prob on each answer token
+ # lp = lp_gather: the full-vocab log_softmax logprob of each answer token at the answer
+ # slot. This is the RAW PRIMITIVE -- every readout (E, the logit contrast C, log-odds,
+ # entropy) is a pure function of it, and a steer effect is just a difference of lp. Keep
+ # it; do not throw it away by collapsing to a single number here.
+ lp = np.asarray(slot["lp_gather"], dtype=float) # [A] raw logprobs (full-vocab norm)
+ p_a = np.exp(lp) # [A] prob on each answer token
pmass = float(slot["pmass_allowed"]) # mass on allowed tokens (coherence)
# Renormalize within allowed. INTENTIONALLY NOT NaN-guarded: at full coherence collapse
# pmass -> 0 so p_norm -> NaN and poisons that item's factor. That is the honest signal, a
@@ -103,12 +107,15 @@ def read_items(model, tok, instr: Instrument, items: list[InstrItem], answer_ids
# of 10 != the mean of 130), so it must not be silently turned into a comparable-looking
# number. NaN marks "do not compare". Do not "fix" this with a softmax/eps fallback.
p_norm = p_a / p_a.sum() # [A] within allowed (NaN at collapse, by design)
+ think_text, n_think, emitted_close = thinks[j]
out.append({
"id": it.id, "frame": it.frame,
+ "lp": lp, # raw logprobs at the M scale tokens
"p": p_norm,
"pmass_allowed": pmass,
"dimension": it.dimension, "sign": it.sign,
"human_label": it.human_label,
+ "think": think_text, "n_think": n_think, "emitted_close": emitted_close,
})
if verbose_first and i == 0:
slot0 = slots[0][0]
diff --git a/src/tinymfv/readouts.py b/src/tinymfv/readouts.py
new file mode 100644
index 0000000..66b37ad
--- /dev/null
+++ b/src/tinymfv/readouts.py
@@ -0,0 +1,68 @@
+"""Ordinal Likert readouts: pure functions of the raw answer-token logprobs.
+
+One primitive (`lp`, the full-vocab logprobs at the M scale tokens) -> many summaries. The split
+matters because the summaries answer different questions and have very different steer-sensitivity:
+
+ expected_score E = sum_k k * p_k, in [1, M] human-comparable scale score
+ logit_contrast C = sum_k (k - mid) * lp_k primary steer signal (sensitive, signed)
+ agree_logodds LO = lse(top) - lse(bottom) readable 2-bin direction summary
+ entropy H = -sum_k p_k log p_k within-allowed coherence (uniform = ln M)
+
+Why E hides steering and C/LO do not (the whole reason this module exists):
+
+ dE/dl_j = p_j (j - E) -> vanishes when the model is confident (p_j -> 0 in the tails, and
+ j - E -> 0 at the mode). A peaked distribution sits in a flat spot
+ of E, so a small steer that reallocates the tails barely moves it.
+ dC/dl_j = (k - mid) -> a fixed weight, NO p_j factor. Centered weights (sum = 0) also kill
+ the softmax normalizer, so C is normalizer-invariant (raw logits,
+ full-vocab logprobs, or within-M renormalized logprobs all give the
+ same C) and the steer effect is exactly linear: dC = w . dl.
+
+So C is the log-space analog of E: same `sum (weight_k) * (per-token quantity)` shape, but in
+logprobs with midpoint-centered weights. LO is C's 2-bin special case (weights +-1 on the poles,
+neutral dropped, each pole pooled with logsumexp). Keep E only for landing the model against human
+norms; use C (or LO) for "did the steer move it".
+"""
+from __future__ import annotations
+
+import numpy as np
+
+
+def _logsumexp(x: np.ndarray) -> float:
+ x = np.asarray(x, dtype=float)
+ m = float(np.max(x))
+ return m + float(np.log(np.exp(x - m).sum()))
+
+
+def expected_score(p: np.ndarray, scale_max: int) -> float:
+ """E = sum_k k * p_k, in [1, scale_max]. Human-comparable; INSENSITIVE near a confident answer."""
+ w = np.arange(1, scale_max + 1, dtype=float)
+ return float((np.asarray(p, dtype=float) * w).sum())
+
+
+def logit_contrast(lp: np.ndarray, scale_max: int) -> float:
+ """C = sum_k (k - mid) * lp_k, the rank-centered logit contrast (mid = (1 + scale_max) / 2).
+
+ Primary ordinal steer readout. Sensitive (dC/dl_j = weight_j, no probability suppression),
+ signed, unbounded, normalizer-invariant (weights sum to 0, so any constant offset on `lp`
+ cancels), and linear in the logits so dC across a steer = weights . (lp_steered - lp_base)."""
+ mid = (1 + scale_max) / 2.0
+ w = np.arange(1, scale_max + 1, dtype=float) - mid
+ return float((np.asarray(lp, dtype=float) * w).sum())
+
+
+def agree_logodds(lp: np.ndarray, scale_max: int) -> float:
+ """LO = logsumexp(top n_side) - logsumexp(bottom n_side), n_side = scale_max // 2.
+
+ The readable 2-bin direction summary: nats in favor of agreeing over disagreeing among the
+ non-neutral options. Drops the middle category; less information than C but interpretable."""
+ lp = np.asarray(lp, dtype=float)
+ n_side = scale_max // 2
+ return _logsumexp(lp[-n_side:]) - _logsumexp(lp[:n_side])
+
+
+def entropy(p: np.ndarray, scale_max: int | None = None) -> float:
+ """Shannon entropy of the within-allowed distribution, nats. Coherence the pmass gate misses:
+ a uniform answer has pmass ~ 1 but entropy = ln(scale_max), the max."""
+ p = np.asarray(p, dtype=float)
+ return float(-(p * np.log(p + 1e-12)).sum())