diff --git a/scripts/wvs_map.py b/scripts/wvs_map.py index 0e059a3..efd736b 100644 --- a/scripts/wvs_map.py +++ b/scripts/wvs_map.py @@ -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. diff --git a/src/tinymfv/maps.py b/src/tinymfv/maps.py index 0dfdd83..bd365d0 100644 --- a/src/tinymfv/maps.py +++ b/src/tinymfv/maps.py @@ -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) diff --git a/src/tinymfv/read_api.py b/src/tinymfv/read_api.py index c365f0a..2f89752 100644 --- a/src/tinymfv/read_api.py +++ b/src/tinymfv/read_api.py @@ -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