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

WVS map: labeled Inglehart-Welzel axes + Economist-style convex-hull zones

Browse Source 
Replace the blind ipsative-PCA projection with two named IW axes built from
GlobalOpinionQA WVS items (tinymfv.iw_axes): X = Survival<->Self-expression
(homosexuality, trust, political action), Y = Traditional<->Secular-Rational
(religion importance+belief, abortion, child autonomy). Each item is oriented to
its axis-positive pole by reading the option order, so a reversed row can't flip a
country. Human anchors land where the published IW map puts them (Sweden top-right,
Nigeria/Pakistan bottom-left, East Asia secular-but-survival top-left).

Models answer the same items via the answer-token reader (single-digit option labels
so the 1-10 justifiable scale stays single-token); coords are the same axis-mean.

Add maps.draw_zone_hulls: tight rounded convex hulls (not inflated disc unions),
zone-coloured dots, outlier-only labels, white-haloed region names -- the Economist
grammar. draw_zone_regions stays for the instrument maps' within-country spread.

Co-Authored-By: Claudypoo <288921227+claudypoo@users.noreply.github.com>
This commit is contained in:
wassname
2026-07-04 21:26:25 +08:00
co-authored by Claudypoo
parent ea796dee9a
commit abe43ad593
4 changed files with 306 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/probe_iw_themes.py
+53
View File
@@ -0,0 +1,53 @@
"""Dump FULL (untruncated) text + options + coverage for WVS questions matching each IW theme, so
the precise per-item selector and axis-positive pole can be locked by eye. -- authored by Claude
uv run python scripts/probe_iw_themes.py
"""
from __future__ import annotations
import ast
import re
from datasets import load_dataset
from tinymfv.zones import zone_of
SKIP = re.compile(r"don'?t know|no answer|refus|decline|none of|not applicable|^other|missing|inap",
re.I)
THEMES = {
"Y_religion": r"how important it is in your life|believe in any|importance of god|attend religious",
"Y_abortion": r"\babortion\b",
"Y_childaut": r"encouraged to learn at home",
"X_homosex": r"homosexual",
"X_trust": r"most people can be trusted",
"X_politaction": r"forms of political action",
}
def main() -> None:
ds = load_dataset("Anthropic/llm_global_opinions", split="train")
recs = []
for r in ds:
if r["source"] != "WVS" or not r["question"]:
continue
opts = ast.literal_eval(r["options"]) if isinstance(r["options"], str) else r["options"]
keep = [i for i, o in enumerate(opts) if not SKIP.search(o)]
sel = ast.literal_eval(re.search(r"\{.*\}", r["selections"], re.S).group(0))
ncov = sum(1 for c in sel if zone_of(c) and sum(sel[c][i] for i in keep) > 0)
recs.append({"q": r["question"], "opts": [opts[i] for i in keep], "ncov": ncov})
for theme, pat in THEMES.items():
rx = re.compile(pat, re.I)
hits = sorted([r for r in recs if rx.search(r["q"])], key=lambda r: -r["ncov"])
print(f"\n{'='*90}\n{theme}: {len(hits)} matches\n{'='*90}")
for r in hits:
if r["ncov"] < 40:
continue
tail = r["q"][-90:].replace("\n", " / ")
print(f"[cov {r['ncov']:3d}] n_opts={len(r['opts'])} ...TAIL: {tail!r}")
print(f" opts: {r['opts']}")
if __name__ == "__main__":
main()
scripts/wvs_map.py
+131 -97
View File
@@ -1,11 +1,16 @@
"""WVS / Inglehart-Welzel style culture map: place LLMs among human societies (the Economist chart).
"""WVS Inglehart-Welzel culture map with LABELED axes: place LLMs among human societies (the
Economist chart), on the two named IW dimensions instead of a blind PCA.
Runs tinymfv on the WVS subset of Anthropic/llm_global_opinions "like the others": each 4-option WVS
question is one ordinal item; a country x question matrix of expected-score E is ipsative-PCA'd (the
same maps.ipsative_pca the instrument maps use) into 2 axes, with IW zone ellipses. Each model is
administered the SAME questions -- open models via the logprob reader (read_items), logprob-less API
models via the sampling reader (read_items_sampled) -- reduced to an E-vector and projected onto the
country axes as a labelled dot.
X = Survival <-> Self-expression (homosexuality tolerance, interpersonal trust, political action)
Y = Traditional <-> Secular-Rational (religion importance + belief, abortion, child autonomy)
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.
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
@@ -16,7 +21,6 @@ import argparse
import ast
import json
import re
from collections import Counter
from pathlib import Path
import dotenv
@@ -32,31 +36,36 @@ from datasets import load_dataset
from transformers import AutoModelForCausalLM, AutoTokenizer
from tinymfv import maps
from tinymfv.zones import zones_for, zone_of, ECONOMIST_OUTLIERS
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.readouts import expected_score
from tinymfv.iw_axes import AXIS_ITEMS, X_AXIS, Y_AXIS, SKIP, resolve_items, positiveness
SKIP = re.compile(r"don'?t know|no answer|refus|decline|none of|not applicable|^other", re.I)
MODEL_COLORS = ["#c0392b", "#8e44ad", "#16a085", "#d35400", "#2980b9", "#c2185b"]
# option labels are single digits 0..n-1 -- single-token (unlike '10' on the justifiable scale) and
# the format the answer-token reader is tuned for (a bare digit, not a letter the model ignores in
# favour of the option word).
DIGITS = "0123456789"
def load_wvs(n_opts: int) -> list[dict]:
"""WVS questions with exactly `n_opts` substantive options (DK/No-answer dropped), each carrying
its per-country human distribution renormalized over the substantive options."""
def load_wvs_all() -> list[dict]:
"""Every WVS question with its substantive options (DK/refusal/Missing/INAP dropped) and each
zone-mapped country's distribution renormalized over those options."""
ds = load_dataset("Anthropic/llm_global_opinions", split="train")
out = []
for r in ds:
if r["source"] != "WVS":
if r["source"] != "WVS" or not r["question"]:
continue
opts = ast.literal_eval(r["options"]) if isinstance(r["options"], str) else r["options"]
keep = [i for i, o in enumerate(opts) if not SKIP.search(o)]
if len(keep) != n_opts:
if len(keep) < 2:
continue
sel = ast.literal_eval(re.search(r"\{.*\}", r["selections"], re.S).group(0))
dist = {}
for c, ps in sel.items():
if not zone_of(c):
continue
v = np.array([ps[i] for i in keep], float)
if v.sum() > 0:
dist[c] = v / v.sum()
@@ -64,86 +73,94 @@ def load_wvs(n_opts: int) -> list[dict]:
return out
def dense_block(recs: list[dict]) -> tuple[list[str], list[dict]]:
"""Largest complete country x question block (no imputation): start from every zone-mapped
country x every question, then greedily drop whichever row or column has the most missing cells
until no cell is missing. WVS coverage is patchy per-question, so a complete block needs this
trim rather than a fixed coverage threshold."""
countries = sorted({c for r in recs for c in r["dist"] if zone_of(c)})
A = np.array([[c in r["dist"] for r in recs] for c in countries]) # C x Q presence
rows, cols = list(range(len(countries))), list(range(len(recs)))
while True:
sub = A[np.ix_(rows, cols)]
if sub.all():
break
frac_r, frac_c = (~sub).mean(1), (~sub).mean(0) # missing FRACTION per country / per question
if frac_r.max() >= frac_c.max():
rows.pop(int(np.argmax(frac_r)))
else:
cols.pop(int(np.argmax(frac_c)))
return [countries[i] for i in rows], [recs[j] for j in cols]
def human_axis_scores(resolved: dict[str, list[dict]]) -> tuple[list[str], np.ndarray]:
"""Per-country (X, Y). A country is kept if it covers at least half of each axis's items; its
axis value is the mean positiveness over the items it does cover."""
countries = sorted({c for items in resolved.values() for it in items for c in it["rec"]["dist"]})
rows, keep = [], []
for c in countries:
xy, ok = [], True
for axis in (X_AXIS, Y_AXIS):
vals = [positiveness(it["rec"]["dist"][c], it["pole_idx"], it["n"])
for it in resolved[axis] if c in it["rec"]["dist"]]
if len(vals) < (len(resolved[axis]) + 1) // 2:
ok = False
break
xy.append(float(np.mean(vals)))
if ok:
keep.append(c)
rows.append(xy)
return keep, np.array(rows)
def human_matrix(countries: list[str], block: list[dict], n_opts: int) -> np.ndarray:
"""countries x questions matrix of expected-score E as a 0-1 fraction (E-1)/(n_opts-1)."""
w = np.arange(1, n_opts + 1)
M = np.array([[float((block[q]["dist"][c] * w).sum()) for q in range(len(block))] for c in countries])
return (M - 1) / (n_opts - 1)
def build_instruments(resolved: dict[str, list[dict]]) -> tuple[list[Instrument], dict[str, dict]]:
"""One nominal Instrument per distinct option-count (answer_space = single letters), covering the
union of both axes' items. Returns the instruments + a {suffix: {pole_idx, n, axis}} index."""
items_by_n: dict[int, list[InstrItem]] = {}
meta: dict[str, dict] = {}
seen: set[str] = set()
for axis, items in resolved.items():
for it in items:
s = it["suffix"]
meta[s] = {"pole_idx": it["pole_idx"], "n": it["n"], "axis": axis}
if s in seen:
continue
seen.add(s)
n, opts = it["n"], it["rec"]["opts"]
legend = "; ".join(f"{DIGITS[k]}) {o}" for k, o in enumerate(opts))
task = f"Answer options: {legend}. Respond with only the number."
items_by_n.setdefault(n, []).append(
InstrItem(id=s, prompt=it["rec"]["q"], dimension="iw", sign=1,
frame="forward", meta={"task": task}))
instrs = [Instrument(name=f"wvs_iw_n{n}", construct="opinion", kind="nominal",
answer_space=list(DIGITS[:n]), dimensions=["iw"], items=its,
prefill="(", display="WVS-IW")
for n, its in sorted(items_by_n.items())]
return instrs, meta
def build_instrument(block: list[dict], n_opts: int) -> Instrument:
"""One ordinal Instrument over the WVS questions: answer_space digits 1..n_opts, each item the
question + its numbered options legend."""
items = []
for i, r in enumerate(block):
legend = "; ".join(f"{k+1}) {o}" for k, o in enumerate(r["opts"]))
task = f"Answer options: {legend}. Respond with only the number."
items.append(InstrItem(id=str(i), prompt=r["q"], dimension="wvs", sign=1,
frame="forward", meta={"task": task}))
return Instrument(name="wvs", construct="opinion", kind="ordinal",
answer_space=[str(i) for i in range(1, n_opts + 1)],
dimensions=["wvs"], items=items, prefill="(", scale_max=n_opts,
human_scale_max=n_opts, display="WVS")
def model_axis_scores(vecs: dict[str, np.ndarray], meta: dict[str, dict],
resolved: dict[str, list[dict]]) -> tuple[float, float]:
"""(X, Y) for one model from its per-item p vectors (suffix -> p over options)."""
xy = []
for axis in (X_AXIS, Y_AXIS):
vals = [positiveness(vecs[it["suffix"]], it["pole_idx"], it["n"]) for it in resolved[axis]]
xy.append(float(np.mean(vals)))
return xy[0], xy[1]
def model_vector(rows: list[dict], n: int, n_opts: int) -> np.ndarray:
"""Per-question E as a 0-1 fraction, in item order. NaN if a question's read collapsed."""
by_id = {int(r["id"]): r for r in rows}
out = np.full(n, np.nan)
for i in range(n):
p = np.asarray(by_id[i]["p"], float)
if np.isfinite(p).all() and abs(p.sum() - 1) < 1e-6:
out[i] = (expected_score(p, n_opts) - 1) / (n_opts - 1)
return out
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."""
return {r["id"]: np.asarray(r["p"], float)[: meta[r["id"]]["n"]] for r in rows}
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("--n-opts", type=int, default=4)
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("--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.png")
ap.add_argument("--cache", default="/tmp/claude-1000/wvs_model_vectors.json",
help="cache model E-vectors so re-plotting (new zone style) skips the API calls")
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")
args = ap.parse_args()
recs = load_wvs(args.n_opts)
countries, block = dense_block(recs)
logger.info(f"{len(recs)} {args.n_opts}-option WVS questions -> dense block "
f"{len(countries)} countries x {len(block)} questions")
Hfrac = human_matrix(countries, block, args.n_opts)
P, Vt, var, mu, Pc = maps.ipsative_pca(Hfrac)
instr = build_instrument(block, args.n_opts)
recs = load_wvs_all()
resolved = resolve_items(recs)
for axis, items in resolved.items():
logger.info(f"{axis}: " + ", ".join(f"{it['suffix']}[n{it['n']},pole{it['pole_idx']}]"
for it in items))
countries, P = human_axis_scores(resolved)
logger.info(f"{len(recs)} WVS questions -> {len(countries)} countries on 2 IW axes")
# cache keyed by the exact question block: same block -> reuse E-vectors, only re-projecting +
# re-drawing (so iterating on the zone style costs no API calls).
sig = f"{args.n_opts}:{len(block)}:{hash(tuple(r['q'] for r in block)) & 0xffffffff}"
instrs, meta = build_instruments(resolved)
sig = str(hash(tuple(sorted((s, m["n"]) for s, m in meta.items()))) & 0xffffffff)
cpath = Path(args.cache)
cache = json.loads(cpath.read_text()).get(sig, {}) if cpath.exists() else {}
vecs: dict[str, np.ndarray] = {k: np.array(v) for k, v in cache.items()}
vecs: dict[str, dict[str, np.ndarray]] = {k: {s: np.array(p) for s, p in v.items()}
for k, v in cache.items()}
if args.local_model:
key = args.local_model.split("/")[-1] + " (lp)"
@@ -153,41 +170,58 @@ def main() -> None:
tok.pad_token = tok.eos_token
tok.padding_side = "left"
lm = AutoModelForCausalLM.from_pretrained(args.local_model, dtype=torch.bfloat16).to(args.device).eval()
rows = read_items(lm, tok, instr, instr.items, resolve_answer_ids(tok, instr.answer_space),
max_think_tokens=args.max_think_tokens, batch_size=16, verbose_first=True)
vecs[key] = model_vector(rows, len(block), args.n_opts)
rows = []
for k, instr in enumerate(instrs):
rows += read_items(lm, tok, instr, instr.items,
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)
for m in args.api_models:
key = m.split("/")[-1] + " (sampled)"
if key not in vecs:
rows = read_items_sampled(m, instr, instr.items, n_samples=args.api_samples, verbose_first=True)
vecs[key] = model_vector(rows, len(block), args.n_opts)
rows = []
for k, instr in enumerate(instrs):
rows += read_items_sampled(m, instr, instr.items, n_samples=args.api_samples,
verbose_first=(k == 0))
vecs[key] = read_model(rows, meta)
cpath.parent.mkdir(parents=True, exist_ok=True)
allc = json.loads(cpath.read_text()) if cpath.exists() else {}
allc[sig] = {k: v.tolist() for k, v in vecs.items()}
allc[sig] = {k: {s: p.tolist() for s, p in v.items()} for k, v in vecs.items()}
cpath.write_text(json.dumps(allc))
hmean = Hfrac.mean(0) # a question the model didn't answer coherently -> the human average there
models = {k: ((np.where(np.isnan(v), hmean, v) @ Pc) - mu) @ Vt[:2].T for k, v in vecs.items()}
models = {k: model_axis_scores(v, meta, resolved) for k, v in vecs.items()}
zones, emph = zones_for(countries)
fig, ax = plt.subplots(figsize=(10, 8))
zone_of_c = {c: z for z, members in zones.items() for c in members}
dot_cols = [maps.ZONE_COLORS.get(zone_of_c[c], "#888888") for c in countries]
fig, ax = plt.subplots(figsize=(11, 9))
ax.set_facecolor("#faf8f2")
ax.grid(True, color="#eceadf", lw=0.3, zorder=0)
maps.draw_zone_regions(ax, P, countries, zones)
ax.scatter(P[:, 0], P[:, 1], s=22, c=maps.C_HUM, alpha=0.7, edgecolors="white", linewidths=0.4, zorder=3)
ax.axhline(0.5, color="#d9d5c6", lw=0.8, zorder=1)
ax.axvline(0.5, color="#d9d5c6", lw=0.8, zorder=1)
maps.draw_zone_hulls(ax, P, countries, zones)
# Dots coloured by zone (region shown by colour, like the Economist) + only the named outliers
# labelled -- 90 country labels is the clutter the user flagged; the region name (drawn by
# draw_zone_regions at each zone centroid) carries the rest.
ax.scatter(P[:, 0], P[:, 1], s=28, c=dot_cols, alpha=0.85, edgecolors="white", linewidths=0.5, zorder=3)
for i, c in enumerate(countries):
is_e = c in emph
ax.annotate(c, (P[i, 0], P[i, 1]), fontsize=7.5 if is_e else 6, xytext=(3, 2),
textcoords="offset points", color="#111" if is_e else "#666",
fontweight="bold" if is_e else "normal", zorder=6)
if c in emph:
ax.annotate(c, (P[i, 0], P[i, 1]), fontsize=9, xytext=(4, 3),
textcoords="offset points", color="#111", fontweight="bold", zorder=6)
for (name, pt), col in zip(models.items(), MODEL_COLORS):
ax.scatter(*pt, s=120, marker="*", c=col, edgecolors="white", linewidths=1.0, zorder=8)
ax.annotate(name, pt, xytext=(6, 4), textcoords="offset points", fontsize=9,
ax.scatter(*pt, s=150, marker="*", c=col, edgecolors="white", linewidths=1.0, zorder=8)
ax.annotate(name, pt, xytext=(7, 4), textcoords="offset points", fontsize=9,
fontweight="bold", color=col, zorder=9)
ax.set_xlabel(f"PC1 ({var[0]*100:.0f}% var)")
ax.set_ylabel(f"PC2 ({var[1]*100:.0f}% var)")
ax.set_title(f"WVS values map: LLMs among {len(countries)} human societies "
f"({len(block)} WVS questions, ipsative PCA)", fontsize=11)
ax.set_xlabel(f"{X_AXIS} (right = self-expression)")
ax.set_ylabel(f"{Y_AXIS} (up = secular-rational)")
ax.set_title(f"WVS Inglehart-Welzel map: LLMs among {len(countries)} human societies "
f"(approximate IW axes)", fontsize=12)
ax.text(0.01, 0.01,
"Approximate IW: axes built from GlobalOpinionQA WVS items (3 themes/axis, not the\n"
"canonical 5; national pride / authority / materialism absent). Not a verbatim WVS "
"factor score.",
transform=ax.transAxes, fontsize=6.5, color="#888", va="bottom", ha="left", zorder=10)
fig.tight_layout()
fig.savefig(args.out, dpi=200, bbox_inches="tight")
logger.info(f"wrote {args.out}")