mirror of
https://github.com/wassname/moral-maps.git
synced 2026-09-09 11:27:22 +08:00
wvs map: dense Likert rated readout with bootstrap CIs (more reliable)
Replace the single forced-choice sampling reader (1 bit/call, truncated verbose models to empty at max_tokens=32, positionally biased) with a dense readout: rate every option 1-5 as JSON, N samples, binary items order-balanced to cancel positional bias, ratings mapped back to canonical order and normalized to a distribution. All of a model's calls fire concurrently (asyncio.gather over the async openrouter_wrapper; per_call=1 since providers ignore n>1). Each model carries a bootstrap 95% CI (over items + samples) drawn as error bars, so mushy models (deepseek-flash was near coin-flip) read as uncertain, not confident dots. One flaky provider is skipped, not fatal. Co-Authored-By: Claudypoo <288921227+claudypoo@users.noreply.github.com>
This commit is contained in:
wassname
+90
-37
@@ -6,11 +6,15 @@ Economist chart), on the two named IW dimensions instead of a blind PCA.
|
||||
|
||||
Each axis is a small hand-picked battery of GlobalOpinionQA WVS items (tinymfv.iw_axes), every item
|
||||
oriented to its axis-positive pole by reading the option order. A country's coordinate is the mean
|
||||
`positiveness` (0-1) over that axis's items from the human WVS distribution; a model's coordinate is
|
||||
the SAME items administered through the answer-token reader (open models: read_items; logprob-less
|
||||
API models: read_items_sampled), reduced identically. This is an APPROXIMATE IW (3 themes/axis, not
|
||||
the canonical 5 -- national pride/authority/materialism are absent from GlobalOpinionQA), not a
|
||||
verbatim reproduction; the caveat is printed on the figure.
|
||||
`positiveness` (0-1) over that axis's items from the human WVS choice frequencies. A model's
|
||||
coordinate is the SAME items administered as a dense Likert readout (read_items_rated: rate every
|
||||
option 1-5 as JSON, binary options order-permuted to cancel positional bias, N samples, normalized
|
||||
mean rating -> distribution), reduced identically; open models can instead use the answer-token
|
||||
logprob reader (read_items). Each model also carries a bootstrap 95% CI (over items + samples) drawn
|
||||
as error bars, so mushy / uncertain placements read as uncertain rather than confident dots. NB the
|
||||
model coordinate is a rating-derived pseudo-distribution while the human one is a real choice
|
||||
frequency -- a documented proxy. This is an APPROXIMATE IW (3 themes/axis, not the canonical 5 --
|
||||
national pride/authority/materialism are absent from GlobalOpinionQA), not a verbatim reproduction.
|
||||
|
||||
uv run python scripts/wvs_map.py --local-model Qwen/Qwen3-0.6B \
|
||||
--api-models meta-llama/llama-3.1-8b-instruct openai/gpt-4o-mini
|
||||
@@ -40,7 +44,7 @@ from tinymfv import maps
|
||||
from tinymfv.zones import zones_for, zone_of
|
||||
from tinymfv.instrument import Instrument, InstrItem
|
||||
from tinymfv.read import read_items, resolve_answer_ids
|
||||
from tinymfv.read_api import read_items_sampled
|
||||
from tinymfv.read_api import read_items_rated
|
||||
from tinymfv.iw_axes import AXIS_ITEMS, X_AXIS, Y_AXIS, SKIP, resolve_items, positiveness
|
||||
|
||||
# option labels are single digits 0..n-1 -- single-token (unlike '10' on the justifiable scale) and
|
||||
@@ -130,22 +134,51 @@ def model_axis_scores(vecs: dict[str, np.ndarray], meta: dict[str, dict],
|
||||
|
||||
|
||||
def read_model(rows: list[dict], meta: dict[str, dict]) -> dict[str, np.ndarray]:
|
||||
"""rows from read_items / read_items_sampled -> {suffix: p over that item's options}. NaN p (read
|
||||
collapse) fails loud later via positiveness rather than being imputed."""
|
||||
"""rows from read_items -> {suffix: p over that item's options}. NaN p (read collapse) fails loud
|
||||
later via positiveness rather than being imputed."""
|
||||
return {r["id"]: np.asarray(r["p"], float)[: meta[r["id"]]["n"]] for r in rows}
|
||||
|
||||
|
||||
def model_coord_ci(psamples: dict[str, np.ndarray], resolved: dict[str, list[dict]],
|
||||
rng: np.random.Generator, B: int = 500) -> tuple[float, float, float, float]:
|
||||
"""(x, y, x_se, y_se). Point estimate = mean positiveness over each axis's items on the mean-over-
|
||||
samples p. The SE is a bootstrap over BOTH noise sources: resample the axis's items with
|
||||
replacement (item-set noise, only ~3-7 items/axis) and, per item, draw one of its N rating samples
|
||||
(readout noise). std over B replicates -> the CI drawn as error bars on the map."""
|
||||
def axis_coords(getp) -> list[float]:
|
||||
return [float(np.mean([positiveness(getp(it), it["pole_idx"], it["n"]) for it in resolved[axis]]))
|
||||
for axis in (X_AXIS, Y_AXIS)]
|
||||
x, y = axis_coords(lambda it: psamples[it["suffix"]].mean(0))
|
||||
bx, by = [], []
|
||||
for _ in range(B):
|
||||
xy = []
|
||||
for axis in (X_AXIS, Y_AXIS):
|
||||
items = resolved[axis]
|
||||
idx = rng.integers(0, len(items), len(items))
|
||||
vals = []
|
||||
for j in idx:
|
||||
it = items[j]
|
||||
ps = psamples[it["suffix"]]
|
||||
vals.append(positiveness(ps[int(rng.integers(0, len(ps)))], it["pole_idx"], it["n"]))
|
||||
xy.append(float(np.mean(vals)))
|
||||
bx.append(xy[0]); by.append(xy[1])
|
||||
return x, y, float(np.std(bx)), float(np.std(by))
|
||||
|
||||
|
||||
def main() -> None:
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--local-model", default="Qwen/Qwen3-0.6B")
|
||||
ap.add_argument("--api-models", nargs="*", default=[])
|
||||
ap.add_argument("--api-samples", type=int, default=20)
|
||||
ap.add_argument("--api-samples", type=int, default=12,
|
||||
help="rating samples per item (each dense: every option rated), binary items order-balanced")
|
||||
ap.add_argument("--api-max-tokens", type=int, default=1024,
|
||||
help="output budget per rating call; large enough that a reasoning model finishes the JSON")
|
||||
ap.add_argument("--max-think-tokens", type=int, default=64)
|
||||
ap.add_argument("--device", default="cuda" if torch.cuda.is_available() else "cpu")
|
||||
ap.add_argument("--out", default="/tmp/claude-1000/wvs_map_iw.png")
|
||||
ap.add_argument("--cache", default="/tmp/claude-1000/wvs_iw_vectors.json",
|
||||
help="cache model per-item p vectors so re-styling skips the API/model calls")
|
||||
ap.add_argument("--responses", default="/tmp/claude-1000/wvs_iw_responses.jsonl",
|
||||
ap.add_argument("--cache", default="/tmp/claude-1000/wvs_iw_rated.json",
|
||||
help="cache model (x,y[,x_se,y_se]) coords so re-styling skips the API/model calls")
|
||||
ap.add_argument("--responses", default="/tmp/claude-1000/wvs_iw_rated_responses.jsonl",
|
||||
help="append every raw model response here (audit trail; API calls cost money)")
|
||||
args = ap.parse_args()
|
||||
|
||||
@@ -157,38 +190,49 @@ def main() -> None:
|
||||
countries, P = human_axis_scores(resolved)
|
||||
logger.info(f"{len(recs)} WVS questions -> {len(countries)} countries on 2 IW axes")
|
||||
|
||||
instrs, meta = build_instruments(resolved)
|
||||
# DETERMINISTIC key over the item set: Python's builtin hash() is salted per process
|
||||
# (PYTHONHASHSEED), so it changes every run and the cache never hits across processes -- costing a
|
||||
# fresh API call each time. hashlib is stable.
|
||||
sig = hashlib.md5(repr(sorted((s, m["n"]) for s, m in meta.items())).encode()).hexdigest()[:8]
|
||||
# One rated item per distinct WVS question (canonical option order), administered to every API model.
|
||||
rated_items, seen = [], set()
|
||||
for axis in (X_AXIS, Y_AXIS):
|
||||
for it in resolved[axis]:
|
||||
if it["suffix"] in seen:
|
||||
continue
|
||||
seen.add(it["suffix"])
|
||||
rated_items.append({"id": it["suffix"], "question": it["rec"]["q"],
|
||||
"options": it["rec"]["opts"], "n": it["n"]})
|
||||
|
||||
# DETERMINISTIC cache key over the item set: Python's builtin hash() is salted per process
|
||||
# (PYTHONHASHSEED), so it changes every run and the cache never hits -- costing a fresh API call
|
||||
# each time. hashlib is stable. cache value = (x, y[, x_se, y_se]) per model.
|
||||
sig = hashlib.md5(repr(sorted((it["id"], it["n"]) for it in rated_items)).encode()).hexdigest()[:8]
|
||||
cpath = Path(args.cache)
|
||||
cpath.parent.mkdir(parents=True, exist_ok=True)
|
||||
cache = json.loads(cpath.read_text()).get(sig, {}) if cpath.exists() else {}
|
||||
vecs: dict[str, dict[str, np.ndarray]] = {k: {s: np.array(p) for s, p in v.items()}
|
||||
for k, v in cache.items()}
|
||||
models: dict[str, tuple] = {k: tuple(v) for k, v in cache.items()}
|
||||
|
||||
def save_cache() -> None:
|
||||
"""Persist after EACH model so a killed run (session teardown) keeps every finished model
|
||||
(kill-safe incremental write, not write-once-at-end)."""
|
||||
"""Persist after EACH model so a killed run keeps every finished model (kill-safe)."""
|
||||
allc = json.loads(cpath.read_text()) if cpath.exists() else {}
|
||||
allc[sig] = {k: {s: p.tolist() for s, p in v.items()} for k, v in vecs.items()}
|
||||
allc[sig] = {k: list(v) for k, v in models.items()}
|
||||
cpath.write_text(json.dumps(allc))
|
||||
|
||||
rpath = Path(args.responses)
|
||||
rpath.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
def save_responses(key: str, rows: list[dict]) -> None:
|
||||
"""Append every raw sampled response (audit trail -- these API calls cost money)."""
|
||||
"""Append every raw rated response (audit trail -- these API calls cost money)."""
|
||||
with rpath.open("a") as fh:
|
||||
for r in rows:
|
||||
fh.write(json.dumps({"model": key, "sig": sig, "item": r["id"],
|
||||
"prompt": r.get("prompt"), "texts": r.get("texts"),
|
||||
"p": np.asarray(r["p"]).tolist(), "pmass": r["pmass_allowed"]}) + "\n")
|
||||
|
||||
rng = np.random.default_rng(0) # deterministic bootstrap
|
||||
|
||||
# Open local model: answer-token logprob reader (single-choice categorical) -> (x, y), no CI.
|
||||
if args.local_model:
|
||||
key = args.local_model.split("/")[-1] + " (lp)"
|
||||
if key not in vecs:
|
||||
if key not in models:
|
||||
instrs, meta = build_instruments(resolved)
|
||||
tok = AutoTokenizer.from_pretrained(args.local_model)
|
||||
if tok.pad_token is None:
|
||||
tok.pad_token = tok.eos_token
|
||||
@@ -200,21 +244,30 @@ def main() -> None:
|
||||
resolve_answer_ids(tok, instr.answer_space),
|
||||
max_think_tokens=args.max_think_tokens, batch_size=16,
|
||||
verbose_first=(k == 0))
|
||||
vecs[key] = read_model(rows, meta)
|
||||
models[key] = model_axis_scores(read_model(rows, meta), meta, resolved)
|
||||
save_cache()
|
||||
for m in args.api_models:
|
||||
key = m.split("/")[-1] + " (sampled)"
|
||||
if key not in vecs:
|
||||
rows = []
|
||||
for k, instr in enumerate(instrs):
|
||||
rows += read_items_sampled(m, instr, instr.items, n_samples=args.api_samples,
|
||||
verbose_first=(k == 0))
|
||||
save_responses(key, rows) # raw answers first (before reducing to p vectors)
|
||||
vecs[key] = read_model(rows, meta)
|
||||
save_cache() # persist this model before starting the next (kill-safe)
|
||||
logger.info(f"cached {key}")
|
||||
|
||||
models = {k: model_axis_scores(v, meta, resolved) for k, v in vecs.items()}
|
||||
# API models: dense rated readout -> (x, y, x_se, y_se) with bootstrap CI.
|
||||
for m in args.api_models:
|
||||
key = m.split("/")[-1] + " (rated)"
|
||||
if key in models:
|
||||
continue
|
||||
try: # one flaky provider / network blip must not abort the panel
|
||||
rows = read_items_rated(m, rated_items, n_samples=args.api_samples,
|
||||
max_tokens=args.api_max_tokens, verbose_first=True)
|
||||
except Exception as e:
|
||||
logger.warning(f"{key}: read failed ({type(e).__name__}: {e}) -> skipping (not cached)")
|
||||
continue
|
||||
save_responses(key, rows) # raw answers first (before reducing)
|
||||
psamples = {r["id"]: np.array(r["p_samples"]) for r in rows}
|
||||
collapsed = [k for k, v in psamples.items() if v.size == 0]
|
||||
if collapsed: # a refusing / off-format model: skip, keep the panel going
|
||||
logger.warning(f"{key}: parse collapse on {collapsed} -> skipping (not cached)")
|
||||
continue
|
||||
models[key] = model_coord_ci(psamples, resolved, rng)
|
||||
save_cache() # persist this model before the next (kill-safe)
|
||||
x, y, xs, ys = models[key]
|
||||
logger.info(f"cached {key}: ({x:.2f}, {y:.2f}) +-({1.96*xs:.02f}, {1.96*ys:.02f}) 95% CI")
|
||||
|
||||
# Render through the SHARED value-map renderer (same one the instrument value maps use): pole
|
||||
# signposts through the human median, 4 auto-selected zone hulls, textalloc labels, model stars.
|
||||
|
||||
+11
-4
@@ -255,12 +255,19 @@ def plot_value_map(display: str, countries: list[str], P: np.ndarray,
|
||||
tcol = ["#111"] * len(lab_i)
|
||||
if models:
|
||||
mnames = list(models)
|
||||
mpts = np.array([models[k] for k in mnames])
|
||||
ax.scatter(mpts[:, 0], mpts[:, 1], s=150, marker="o", c=MODEL_RED, # Economist: bigger red dots
|
||||
mx = np.array([models[k][0] for k in mnames])
|
||||
my = np.array([models[k][1] for k in mnames])
|
||||
# optional bootstrap SE (rated models carry (x, y, x_se, y_se); logprob models just (x, y))
|
||||
xse = np.array([models[k][2] if len(models[k]) > 2 else 0.0 for k in mnames])
|
||||
yse = np.array([models[k][3] if len(models[k]) > 3 else 0.0 for k in mnames])
|
||||
if (xse > 0).any() or (yse > 0).any(): # 95% CI -> a mushy model reads as uncertain
|
||||
ax.errorbar(mx, my, xerr=1.96 * xse, yerr=1.96 * yse, fmt="none", ecolor=MODEL_RED,
|
||||
elinewidth=1.0, alpha=0.4, capsize=2.5, capthick=0.8, zorder=7)
|
||||
ax.scatter(mx, my, s=150, marker="o", c=MODEL_RED, # Economist: bigger red dots
|
||||
edgecolors="white", linewidths=1.0, zorder=8)
|
||||
tx += list(mpts[:, 0]); ty += list(mpts[:, 1]); txt += mnames
|
||||
tx += list(mx); ty += list(my); txt += mnames
|
||||
tcol += [MODEL_RED] * len(mnames)
|
||||
sx = list(P[:, 0]) + list(mpts[:, 0]); sy = list(P[:, 1]) + list(mpts[:, 1])
|
||||
sx = list(P[:, 0]) + list(mx); sy = list(P[:, 1]) + list(my)
|
||||
else:
|
||||
sx, sy = list(P[:, 0]), list(P[:, 1])
|
||||
ax.margins(0.13)
|
||||
|
||||
+117
-1
@@ -25,6 +25,7 @@ and E degenerates to an integer.
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import re
|
||||
from math import inf
|
||||
|
||||
@@ -58,8 +59,10 @@ def _sample_texts(model: str, prompt: str, n_samples: int, temperature: float,
|
||||
errors), requesting up to 8 per call via `n` and topping up until N."""
|
||||
texts: list[str] = []
|
||||
while len(texts) < n_samples:
|
||||
# cap n at 4/call: a big n * max_tokens response is slow enough to trip httpx read timeouts
|
||||
# (which then exhaust the wrapper's retry budget); smaller requests are more reliable.
|
||||
payload = {"model": model, "messages": [{"role": "user", "content": prompt}],
|
||||
"temperature": temperature, "n": min(n_samples - len(texts), 8),
|
||||
"temperature": temperature, "n": min(n_samples - len(texts), 4),
|
||||
"max_tokens": max_tokens}
|
||||
data = asyncio.run(openrouter_request(payload))
|
||||
texts.extend((c["message"].get("content") or "") for c in data["choices"])
|
||||
@@ -107,3 +110,116 @@ def read_items_sampled(model: str, instr: Instrument, items: list[InstrItem], *,
|
||||
f"SHOULD: replies are a bare token in {space}; parse rate near 1.0 -> coherent. "
|
||||
f"ELSE the model is refusing / adding prose / the option set is off.\n")
|
||||
return out
|
||||
|
||||
|
||||
# --- rated readout (dense, positional-bias-controlled) ------------------------------------------
|
||||
_RATE_PROMPT = (
|
||||
"{question}\n\n"
|
||||
"Rate how strongly you personally agree with or endorse EACH of the following answers, on a "
|
||||
"scale of 1 (strongly reject) to 5 (strongly endorse):\n{legend}\n\n"
|
||||
'Respond with ONLY a JSON object mapping each answer\'s number to its 1-5 rating, e.g. '
|
||||
'{{"0": 2, "1": 5}}. Rate every answer, and output nothing but the JSON.')
|
||||
|
||||
|
||||
def _parse_ratings(text: str, n: int) -> dict[int, float] | None:
|
||||
"""The LAST brace-object in `text` parsed as {presented_label -> 1-5 rating}, or None if it is
|
||||
malformed / missing a label / out of range. LAST so a reasoning model that echoes the schema
|
||||
earlier doesn't win over its final answer."""
|
||||
objs = re.findall(r"\{[^{}]*\}", text)
|
||||
if not objs:
|
||||
return None
|
||||
try:
|
||||
raw = json.loads(objs[-1])
|
||||
except json.JSONDecodeError:
|
||||
return None
|
||||
out = {}
|
||||
for k in range(n):
|
||||
v = raw.get(str(k), raw.get(k))
|
||||
if v is None or not (1 <= float(v) <= 5):
|
||||
return None
|
||||
out[k] = float(v)
|
||||
return out
|
||||
|
||||
|
||||
def _rate_plan(items: list[dict], n_samples: int, per_call: int = 1) -> list[dict]:
|
||||
"""Flatten (item, presented-order, count) into a list of <=per_call requests. A binary item splits
|
||||
its draws between the two orders (positional-bias control); an ordinal item keeps natural order
|
||||
(shuffling "Never..Always" is nonsense). per_call=1: OpenRouter providers do NOT reliably honour
|
||||
n>1 (they return a single completion), so one request per sample -- fine, they fire concurrently."""
|
||||
plan = []
|
||||
for i, it in enumerate(items):
|
||||
opts, n = it["options"], it["n"]
|
||||
groups = [([0, 1], (n_samples + 1) // 2), ([1, 0], n_samples // 2)] if n == 2 \
|
||||
else [(list(range(n)), n_samples)]
|
||||
for perm, tot in groups:
|
||||
legend = "\n".join(f"{j}) {opts[perm[j]]}" for j in range(n))
|
||||
prompt = _RATE_PROMPT.format(question=it["question"], legend=legend)
|
||||
while tot > 0:
|
||||
k = min(tot, per_call); tot -= k
|
||||
plan.append({"i": i, "perm": perm, "prompt": prompt, "cnt": k})
|
||||
return plan
|
||||
|
||||
|
||||
def read_items_rated(model: str, items: list[dict], *, n_samples: int = 12, temperature: float = 1.0,
|
||||
max_tokens: int = 512, concurrency: int = 8, verbose_first: bool = False) -> list[dict]:
|
||||
"""Dense Likert readout: per item, ask the model to rate EVERY option 1-5 as JSON, N times, and
|
||||
normalize the mean rating to a per-option distribution `p`. Higher signal per call than a single
|
||||
forced choice, and positional bias is controlled by permuting the PRESENTED order of BINARY items
|
||||
(n==2) across samples then mapping ratings back to the canonical option order. All requests for the
|
||||
model fire CONCURRENTLY (asyncio.gather, capped at `concurrency`) so a 12-item panel is ~1 round
|
||||
trip, not 24 sequential ones.
|
||||
|
||||
`items`: [{"id", "question", "options"(canonical), "n"}]. Returns per item: id, p (mean over valid
|
||||
samples, canonical order), p_samples (per-sample canonical p arrays for bootstrap CIs),
|
||||
pmass_allowed (valid-JSON fraction), prompt, texts. p is NaN at total parse collapse (do not
|
||||
compare), matching the logprob reader. A failed request (network) just drops its samples."""
|
||||
plan = _rate_plan(items, n_samples)
|
||||
|
||||
async def run_all() -> list:
|
||||
sem = asyncio.Semaphore(concurrency)
|
||||
async def call(req):
|
||||
async with sem:
|
||||
payload = {"model": model, "messages": [{"role": "user", "content": req["prompt"]}],
|
||||
"temperature": temperature, "n": req["cnt"], "max_tokens": max_tokens}
|
||||
data = await openrouter_request(payload)
|
||||
return [(c["message"].get("content") or "") for c in data["choices"]]
|
||||
return await asyncio.gather(*(call(r) for r in plan), return_exceptions=True)
|
||||
|
||||
results = asyncio.run(run_all())
|
||||
agg = {i: {"p_samples": [], "texts": [], "prompt": ""} for i in range(len(items))}
|
||||
n_fail = 0
|
||||
for req, res in zip(plan, results):
|
||||
i, n, perm = req["i"], items[req["i"]]["n"], req["perm"]
|
||||
agg[i]["prompt"] = req["prompt"]
|
||||
if isinstance(res, Exception):
|
||||
n_fail += 1
|
||||
continue
|
||||
for text in res:
|
||||
agg[i]["texts"].append(text)
|
||||
rated = _parse_ratings(text, n)
|
||||
if rated is None:
|
||||
continue
|
||||
r_canon = np.zeros(n)
|
||||
for j in range(n):
|
||||
r_canon[perm[j]] = rated[j] # map presented label -> canonical option
|
||||
agg[i]["p_samples"].append(r_canon / r_canon.sum())
|
||||
if n_fail:
|
||||
logger.warning(f"{model}: {n_fail}/{len(plan)} rating calls failed (network) -> fewer samples")
|
||||
|
||||
out = []
|
||||
for i, it in enumerate(items):
|
||||
ps = agg[i]["p_samples"]
|
||||
n = it["n"]
|
||||
p = np.mean(ps, axis=0) if ps else np.full(n, np.nan)
|
||||
out.append({"id": it["id"], "p": p, "p_samples": [x.tolist() for x in ps],
|
||||
"pmass_allowed": len(ps) / n_samples, "n_samples": n_samples,
|
||||
"prompt": agg[i]["prompt"], "texts": agg[i]["texts"]})
|
||||
if verbose_first and i == 0:
|
||||
logger.debug(
|
||||
f"\n=== TRACE read_items_rated first item ({model}, N={n_samples}) ===\n"
|
||||
f"--- prompt ---\n{agg[i]['prompt']}\n"
|
||||
f"--- first 2 raw replies ---\n{agg[i]['texts'][:2]}\n"
|
||||
f"--- mean p over {it['options']} ---\n{np.round(p, 3).tolist()} valid={len(ps)}/{n_samples}\n"
|
||||
f"SHOULD: replies are a bare JSON dict of 1-5 ratings; valid rate near 1.0 -> coherent. "
|
||||
f"ELSE the model is refusing / adding prose / max_tokens too small (empty content).\n")
|
||||
return out
|
||||
|
||||
Reference in New Issue
Block a user