wassname/moral-maps
1
0
Fork 0
mirror of https://github.com/wassname/moral-maps.git synced 2026-09-09 11:27:22 +08:00
Code Issues Packages Projects Releases Wiki Activity

ordinal readouts: keep raw lp, add sensitive logit contrast C + log-odds

Browse Source 
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>
This commit is contained in:
wassname
2026-06-25 19:14:04 +08:00
co-authored by Claudypoo
parent 3580bffedd
commit ecd2affac4
5 changed files with 166 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/tinymfv/__init__.py
+3
View File
@@ -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
src/tinymfv/administer.py
+76 -35
View File
@@ -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_<frame> 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_<frame>
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}
src/tinymfv/instrument.py
+9 -4
View File
@@ -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"),
src/tinymfv/read.py
+10 -3
View File
@@ -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]
src/tinymfv/readouts.py
+68
View File
@@ -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())