mirror of
https://github.com/wassname/moral-maps.git
synced 2026-09-12 12:32:34 +08:00
instrument: Option-3 spec (one reader, reducer-only variation) + canonicalize-at-reader
Unifies forced-choice (nominal) and Likert (ordinal, expectation over the integer distribution) on tinymfv's answer-token reader. Per scientist panel (docs/reviews/ sci_ma_*.md): canonicalize every frame's distribution to one forward orientation before metrics+reducer (fixes the asymmetric nominal-reader/ordinal-reducer reflection), renormalize p with pmass kept as canary, add ordinal |E-error| metric, keying applied only in the profile reducer (proven orthogonal to framing, not a double-flip), cross-scale guard, negative-control shuffle. 8 pure-function unit tests pass. Co-Authored-By: Claudypoo <288921227+claudypoo@users.noreply.github.com>
This commit is contained in:
wassname
@@ -0,0 +1,180 @@
|
||||
"""Instrument spec: one answer-token reader, many questionnaires (Option 3).
|
||||
|
||||
Every instrument is the SAME measurement: a softmax over an answer-token set at a prefilled
|
||||
slot, with a think budget, debias, BMA over sampled traces, a pmass coherence canary, and
|
||||
temperature calibration to human soft-labels. Per-instrument variation is small:
|
||||
|
||||
1. answer_space : tokens gathered. Nominal = foundation words (forced-choice MFV);
|
||||
ordinal = scale points ['1'..'M'] (MFQ-2, Big5, 16PF, HSQ).
|
||||
2. reducer : per-item canonical distribution -> profile over `dimensions`.
|
||||
Nominal -> mean choice frequency (the answer IS the dimension).
|
||||
Ordinal -> E[scale point] grouped by item dimension (expectation over the
|
||||
integer distribution, not argmax), with reverse-keying applied HERE only.
|
||||
3. scaffold : prompt + assistant prefill that forces the answer slot.
|
||||
4. human_label : per-item forward-orientation distribution over answer_space.
|
||||
|
||||
Design corrected after a frontier scientist panel (docs/reviews/sci_ma_*.md). The panel's
|
||||
central, agreed flaw: leaving the frame reflection in the reducer made nominal and ordinal
|
||||
debias asymmetric and risked double-flipping. The fix unifies them:
|
||||
|
||||
CANONICALIZE-AT-READER. Every frame's gathered distribution is mapped to ONE forward
|
||||
orientation before anything else (`canonicalize_to_forward`): nominal reorder frames reindex
|
||||
to canonical answer order (the reader already gathers in canonical order, so this is identity);
|
||||
ordinal `inverted`/`negated` frames reverse the probability vector (the distribution-level
|
||||
analog of agreement = M+1 - E). The per-item object is then the mean of these canonical
|
||||
distributions over frames -- a single forward-orientation categorical, used IDENTICALLY for:
|
||||
- metrics: soft-NLL / JS / temperature T vs the forward human histogram (both kinds), plus
|
||||
an ORDINAL metric (mean |E_model - E_human|) so the certifying metric is sensitive to the
|
||||
expectation the profile actually uses; top1 / informedness only for nominal (an argmax
|
||||
flip metric is meaningless on an ordered scale: 4-vs-5 != 1-vs-5).
|
||||
- profile: the reducer. Keying (reverse-keyed items) is applied ONLY in reduce_ordinal,
|
||||
because it is a pooling-into-factor correction, orthogonal to and composable with the
|
||||
framing canonicalization -- not a double-correction.
|
||||
|
||||
`p` everywhere is renormalized over the allowed answer tokens (sums to 1); `pmass_allowed` is
|
||||
the separate coherence canary, never mixed into the distribution.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
from collections import defaultdict
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Literal
|
||||
|
||||
import numpy as np
|
||||
|
||||
Kind = Literal["nominal", "ordinal"]
|
||||
ORDINAL_REFLECT_FRAMES = {"inverted", "negated"} # frames whose meaning is flipped vs forward
|
||||
|
||||
|
||||
@dataclass
|
||||
class InstrItem:
|
||||
id: str
|
||||
prompt: str # user-turn content (vignette, or task + statement)
|
||||
dimension: str | None = None # ordinal: the factor; nominal: None (read from answer)
|
||||
sign: int = 1 # ordinal keying: -1 = reverse-keyed (pool-reflect, profile only)
|
||||
frame: str = "forward" # forward | inverted | negated (canonicalized before use)
|
||||
human_label: np.ndarray | None = None # forward-orientation dist over answer_space, sums to 1
|
||||
meta: dict = field(default_factory=dict)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Instrument:
|
||||
name: str
|
||||
construct: str # map-layer tag: "salience" | "endorsement" | "wrongness" ...
|
||||
kind: Kind
|
||||
answer_space: list[str] # tokens gathered at the answer slot
|
||||
dimensions: list[str] # profile axes for the map
|
||||
items: list[InstrItem]
|
||||
prefill: str # assistant prefill that forces the answer slot
|
||||
schema_hint: str | None = None # instruction appended to the user turn
|
||||
answer_to_dim: dict[str, str] | None = None # nominal only: answer token -> dimension
|
||||
scale_max: int = 5 # ordinal: top of the Likert scale (== len(answer_space))
|
||||
human_scale_max: int = 5 # human reference scale; HSQ humans on 1-7 (see caveat below)
|
||||
|
||||
def __post_init__(self):
|
||||
if self.kind == "ordinal":
|
||||
assert len(self.answer_space) == self.scale_max, "ordinal answer_space must be 1..scale_max"
|
||||
if self.kind == "nominal" and self.answer_to_dim is None:
|
||||
self.answer_to_dim = {a: a for a in self.answer_space}
|
||||
# Cross-scale caveat (panel: human_scale_max was unused): a 1-7 human histogram (HSQ)
|
||||
# cannot share a 5-way soft-NLL/JS with a 1-5 model directly. Calibration for such
|
||||
# instruments must project both to a common support (or report 0-1 endorsement only).
|
||||
# Enforced loudly rather than silently mis-comparing:
|
||||
if self.kind == "ordinal" and self.human_scale_max != self.scale_max:
|
||||
assert all(it.human_label is None for it in self.items), (
|
||||
f"{self.name}: human_scale_max={self.human_scale_max} != scale_max={self.scale_max}; "
|
||||
"per-item categorical calibration needs a common support. Project the human "
|
||||
"histogram to the model scale before attaching human_label, or leave it None.")
|
||||
|
||||
|
||||
# --- canonicalization: every frame -> one forward orientation, BEFORE metrics or reducer ---
|
||||
|
||||
def canonicalize_to_forward(p: np.ndarray, frame: str, kind: Kind) -> np.ndarray:
|
||||
"""Map a presented-orientation answer distribution to forward orientation.
|
||||
|
||||
nominal: reader gathers in canonical answer order already -> identity (reorder frames are
|
||||
averaged as aligned probability vectors upstream).
|
||||
ordinal: `inverted` (scale legend reversed) and `negated` (content negated) flip meaning, so
|
||||
reverse the vector: p_forward[d] = p_presented[M+1-d]. This is the distribution-level
|
||||
form of agreement = M+1 - E. NB negation is not a guaranteed-exact semantic
|
||||
complement (panel caveat); the frame-disagreement diagnostic surfaces items where it
|
||||
breaks. `forward` -> identity.
|
||||
"""
|
||||
p = np.asarray(p, dtype=float)
|
||||
if kind == "ordinal" and frame in ORDINAL_REFLECT_FRAMES:
|
||||
return p[::-1].copy()
|
||||
return p
|
||||
|
||||
|
||||
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).
|
||||
"""
|
||||
by_id: dict[str, list[dict]] = defaultdict(list)
|
||||
for r in per_row:
|
||||
by_id[r["id"]].append(r)
|
||||
out: dict[str, dict] = {}
|
||||
for iid, rows in by_id.items():
|
||||
canon = [canonicalize_to_forward(r["p"], 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] = {
|
||||
"p": C.mean(axis=0),
|
||||
"pmass": float(np.mean([r["pmass_allowed"] for r in rows])),
|
||||
"dimension": r0.get("dimension"),
|
||||
"sign": r0.get("sign", 1),
|
||||
"human_label": r0.get("human_label"),
|
||||
"n_frames": len(rows),
|
||||
"frame_spread": spread,
|
||||
}
|
||||
return out
|
||||
|
||||
|
||||
# --- profile reducers: the only kind-specific summary ---
|
||||
|
||||
def reduce_nominal(items: dict[str, dict], instr: Instrument) -> np.ndarray:
|
||||
"""Forced-choice profile = mean choice frequency over the answer space, folded to dimensions."""
|
||||
dim_idx = {d: j for j, d in enumerate(instr.dimensions)}
|
||||
acc = np.zeros((len(items), len(instr.dimensions)))
|
||||
for i, it in enumerate(items.values()):
|
||||
for a, pa in zip(instr.answer_space, it["p"]):
|
||||
acc[i, dim_idx[instr.answer_to_dim[a]]] += pa
|
||||
return acc.mean(axis=0)
|
||||
|
||||
|
||||
def reduce_ordinal(items: dict[str, dict], instr: Instrument) -> np.ndarray:
|
||||
"""Likert profile = mean keyed agreement per dimension; agreement = E[scale point].
|
||||
|
||||
Keying lives here and ONLY here: a reverse-keyed item (sign<0) means agreeing with it scores
|
||||
LOW on the factor, so reflect agreement = M+1 - E. The frame canonicalization already happened
|
||||
in per_item_categorical, so this is the single, orthogonal keying step (no double-flip).
|
||||
"""
|
||||
w = np.arange(1, instr.scale_max + 1, dtype=float)
|
||||
by_dim: dict[str, list[float]] = defaultdict(list)
|
||||
for it in items.values():
|
||||
E = float((it["p"] * w).sum())
|
||||
agr = (instr.scale_max + 1 - E) if it["sign"] < 0 else E
|
||||
by_dim[it["dimension"]].append(agr)
|
||||
return np.array([float(np.mean(by_dim[d])) for d in instr.dimensions])
|
||||
|
||||
|
||||
REDUCERS = {"nominal": reduce_nominal, "ordinal": reduce_ordinal}
|
||||
|
||||
|
||||
def expected_value(p: np.ndarray, scale_max: int) -> float:
|
||||
return float((np.asarray(p) * np.arange(1, scale_max + 1)).sum())
|
||||
|
||||
|
||||
# --- negative control: shuffle item->dimension (ordinal) / answer->dim (nominal); profile
|
||||
# correlation with the true profile must collapse to chance. A non-null result is only
|
||||
# interpretable if this control passes (all three reviewers required it). ---
|
||||
|
||||
def shuffle_dimensions(items: dict[str, dict], rng: np.random.Generator) -> dict[str, dict]:
|
||||
dims = [it["dimension"] for it in items.values()]
|
||||
rng.shuffle(dims)
|
||||
return {iid: {**it, "dimension": d} for (iid, it), d in zip(items.items(), dims)}
|
||||
@@ -0,0 +1,111 @@
|
||||
"""Unit tests for the instrument abstraction's pure functions (no model needed).
|
||||
|
||||
Covers the panel's flagged risks: frame canonicalization, the keying-vs-framing composition
|
||||
(NOT a double-flip), renormalized p, ordinal expectation, nominal choice frequency, and the
|
||||
negative control.
|
||||
"""
|
||||
import numpy as np
|
||||
|
||||
from tinymfv.instrument import (
|
||||
Instrument, InstrItem, canonicalize_to_forward, per_item_categorical,
|
||||
reduce_nominal, reduce_ordinal, expected_value, shuffle_dimensions,
|
||||
)
|
||||
|
||||
M = 5
|
||||
ONEHOT = {d: np.eye(M)[d - 1] for d in range(1, M + 1)} # ONEHOT[4] = mass on scale point 4
|
||||
|
||||
|
||||
def _ord_instr(items):
|
||||
return Instrument("t", "endorsement", "ordinal", ["1", "2", "3", "4", "5"],
|
||||
["care", "authority"], items, prefill="(")
|
||||
|
||||
|
||||
def test_canonicalize():
|
||||
# forward + nominal -> identity; ordinal inverted/negated -> reversed vector
|
||||
p = ONEHOT[5]
|
||||
assert np.array_equal(canonicalize_to_forward(p, "forward", "ordinal"), p)
|
||||
assert np.array_equal(canonicalize_to_forward(p, "inverted", "ordinal"), ONEHOT[1])
|
||||
assert np.array_equal(canonicalize_to_forward(p, "negated", "ordinal"), ONEHOT[1])
|
||||
assert np.array_equal(canonicalize_to_forward(p, "inverted", "nominal"), p) # nominal identity
|
||||
|
||||
|
||||
def test_forward_expectation():
|
||||
rows = [{"id": "1", "frame": "forward", "p": ONEHOT[4], "pmass_allowed": 1.0,
|
||||
"dimension": "care", "sign": 1, "human_label": None}]
|
||||
items = per_item_categorical(rows, "ordinal")
|
||||
assert abs(expected_value(items["1"]["p"], M) - 4.0) < 1e-9
|
||||
prof = reduce_ordinal(items, _ord_instr([]))
|
||||
assert abs(prof[0] - 4.0) < 1e-9 # care
|
||||
assert np.isnan(prof[1]) # authority has no items
|
||||
|
||||
|
||||
def test_frame_consistency():
|
||||
# Same item, forward vs inverted: after canonicalization the per-item categorical must agree.
|
||||
fwd = [{"id": "1", "frame": "forward", "p": ONEHOT[1], "pmass_allowed": 1.0,
|
||||
"dimension": "care", "sign": 1, "human_label": None}]
|
||||
inv = [{"id": "1", "frame": "inverted", "p": ONEHOT[5], "pmass_allowed": 1.0,
|
||||
"dimension": "care", "sign": 1, "human_label": None}]
|
||||
e_fwd = expected_value(per_item_categorical(fwd, "ordinal")["1"]["p"], M)
|
||||
e_inv = expected_value(per_item_categorical(inv, "ordinal")["1"]["p"], M)
|
||||
assert abs(e_fwd - 1.0) < 1e-9 and abs(e_inv - 1.0) < 1e-9 # canonicalization makes them agree
|
||||
|
||||
|
||||
def test_keying_is_not_double_flip():
|
||||
# Reverse-keyed item ("I keep in the background", sign=-1) on an INVERTED scale.
|
||||
# Model puts mass on presented "5". Panel claimed frame+keying double-corrects; show it does not.
|
||||
# inverted: presented 5 -> canonical 1 (disagrees with the item) -> E_canon = 1.
|
||||
# keying sign<0 (pool-reflect): agreement = M+1 - 1 = 5 = HIGH on the factor. Correct:
|
||||
# disagreeing with "I keep in the background" == extraverted.
|
||||
rows = [{"id": "x", "frame": "inverted", "p": ONEHOT[5], "pmass_allowed": 1.0,
|
||||
"dimension": "care", "sign": -1, "human_label": None}]
|
||||
items = per_item_categorical(rows, "ordinal")
|
||||
assert abs(expected_value(items["x"]["p"], M) - 1.0) < 1e-9 # canonical agreement-with-item = 1
|
||||
prof = reduce_ordinal(items, _ord_instr([]))
|
||||
assert abs(prof[0] - 5.0) < 1e-9 # keyed factor score = 5, not 1 (no double flip)
|
||||
|
||||
# And it matches the SAME reverse-keyed item shown forward with the mirrored answer (mass on 1):
|
||||
rows_fwd = [{"id": "x", "frame": "forward", "p": ONEHOT[1], "pmass_allowed": 1.0,
|
||||
"dimension": "care", "sign": -1, "human_label": None}]
|
||||
prof_fwd = reduce_ordinal(per_item_categorical(rows_fwd, "ordinal"), _ord_instr([]))
|
||||
assert abs(prof[0] - prof_fwd[0]) < 1e-9
|
||||
|
||||
|
||||
def test_frame_spread_diagnostic():
|
||||
# forward mass on 1, inverted mass on presented 1 (-> canonical 5): canonical frames disagree
|
||||
# maximally -> frame_spread = 2.0 (L1 between two disjoint one-hots).
|
||||
rows = [{"id": "1", "frame": "forward", "p": ONEHOT[1], "pmass_allowed": 1.0,
|
||||
"dimension": "care", "sign": 1, "human_label": None},
|
||||
{"id": "1", "frame": "inverted", "p": ONEHOT[1], "pmass_allowed": 1.0,
|
||||
"dimension": "care", "sign": 1, "human_label": None}]
|
||||
items = per_item_categorical(rows, "ordinal")
|
||||
assert abs(items["1"]["frame_spread"] - 2.0) < 1e-9
|
||||
|
||||
|
||||
def test_reduce_nominal():
|
||||
instr = Instrument("mfv", "salience", "nominal", ["care", "authority"],
|
||||
["care", "authority"], [InstrItem("1", "v1")], prefill="(")
|
||||
rows = [{"id": "1", "frame": "forward", "p": np.array([0.8, 0.2]), "pmass_allowed": 1.0},
|
||||
{"id": "2", "frame": "forward", "p": np.array([0.4, 0.6]), "pmass_allowed": 1.0}]
|
||||
prof = reduce_nominal(per_item_categorical(rows, "nominal"), instr)
|
||||
assert np.allclose(prof, [0.6, 0.4]) # mean choice frequency
|
||||
|
||||
|
||||
def test_negative_control_shuffle():
|
||||
rng = np.random.default_rng(0)
|
||||
rows = [{"id": str(i), "frame": "forward", "p": ONEHOT[(i % 5) + 1], "pmass_allowed": 1.0,
|
||||
"dimension": "care" if i < 5 else "authority", "sign": 1, "human_label": None}
|
||||
for i in range(10)]
|
||||
items = per_item_categorical(rows, "ordinal")
|
||||
shuffled = shuffle_dimensions(items, rng)
|
||||
# same item ids, dimensions permuted
|
||||
assert set(shuffled) == set(items)
|
||||
assert [shuffled[k]["dimension"] for k in items] != [items[k]["dimension"] for k in items]
|
||||
|
||||
|
||||
def test_cross_scale_guard():
|
||||
# HSQ-like: human on 1-7, model on 1-5, with a human_label attached -> must raise.
|
||||
import pytest
|
||||
items = [InstrItem("1", "q", dimension="care", human_label=np.ones(5) / 5)]
|
||||
with pytest.raises(AssertionError):
|
||||
Instrument("hsq", "endorsement", "ordinal", ["1", "2", "3", "4", "5"],
|
||||
["care"], items, prefill="(", human_scale_max=7)
|
||||
Reference in New Issue
Block a user