wassname/persona-steering-template-library
1
0
Fork 0
mirror of https://github.com/wassname/persona-steering-template-library.git synced 2026-08-05 13:20:33 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #2 from wassname/add-scenario-sources

Browse Source 
Add scenario_sources: public-dataset scenario loaders + screen workflow
This commit is contained in:
wassname (Michael J Clark)
2026-06-28 22:25:44 +08:00
committed by GitHub
parent 9e73d9fa46 0e2f331733
commit b446589be7
9 changed files with 3348 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyproject.toml
+1
View File
@@ -23,6 +23,7 @@ dependencies = [
"kaleido>=1.3.0",
"itables>=2.8.1",
"polars>=1.41.2",
"datasets>=5.0.0",
]
[tool.uv]
scenario_sources/HANDOVER.md
+86
View File
@@ -0,0 +1,86 @@
# Handover: scenario_sources
Contributed from `w2schar-mini` (weak-to-strong character steering). This subdir
adds dataset->scenario loaders + the screen workflow; the canonical scenario gym
already lives at `scripts/validate_persona_axes_openrouter.py`.
## What landed
- `loaders.py` -- 5 rules-only loaders (airisk, moral_stories, daily_dilemmas,
social_chem, ethics_qna) + `load_machiavelli` (reads the committed cache). All
tested live in w2schar-mini; emit prompts-not-completions, affordance-filtered,
length-capped. `LOADERS` dict maps name -> loader.
- `summarise_machiavelli.py` -- offline deepseek-v4-flash compressor, round-robin
across games, robust to empty API responses, writes `data/machiavelli_summaries.jsonl`.
- `data/machiavelli_summaries.jsonl` -- 13 seed summaries across 11 games (proof
of shape; the full ~83k is the stretch task below).
- Patch to `scripts/validate_persona_axes_openrouter.py`: guard `resp.choices is
None` (OpenRouter error bodies) so one bad API response can't abort the screen.
- `README.md` -- the affordance contract, the load->screen->keep workflow, per-source
caveats.
## Remaining / suggested
1. **Run the screen end-to-end** and commit a per-source clean-rate table + a few
kept/culled examples into the README (the w2schar run aborted on the choices=None
bug now patched; re-run with an instruct generator or qwen-with-raised-max_tokens).
2. **Fold the existing** `data/scenarios_v2_candidates.jsonl` and
`data/scenarios_w2s_character_3p.jsonl` into the same schema so there is ONE
scenario home.
3. **machiavelli HF dataset**: summarise all 83,389 usable rows (resumable via the
cache, ~$15-30), preserve per-choice morality labels + agg_power + game id +
row_i, publish as a HF dataset; downstream draws a game-balanced capped sample.
4. **A combined screened `character_scenarios` HF dataset** from all sources.
## 2026-06-28 Machiavelli moral-contrast selection update
New local script:
- `scenario_sources/score_machiavelli_moral_contrast.py`
Purpose:
- Score `wassname/machiavelli` rows by how much the available choices differ on
moral axes, using per-axis standard deviation across the choices.
- Select across games, not just globally, so high-volume games do not dominate.
- Produce reviewable local tables before any summarization, OpenRouter call, or
HF upload.
Current score:
```text
score = weighted_std(subtle moral labels across choices)
- 0.8 * weighted_std(crude/power labels across choices)
```
Subtle labels include fairness, deception, manipulation, non-physical harm,
promises, spying, stealing, honesty, cooperation, morality, and utility. Crude
labels include intending harm, killing, physical harm, and power. Violence is
allowed when it carries moral ambiguity; it is only penalized when it appears to
be the whole contrast.
Outputs:
- `out/machiavelli_moral_contrast/scored.jsonl`: all deduped rows sorted by score.
- `out/machiavelli_moral_contrast/top_by_game.jsonl`: top N rows from each game.
- `out/machiavelli_moral_contrast/top1000_round_robin.jsonl`: balanced top 1000
selected round-robin over games.
- `out/machiavelli_moral_contrast/random_5pct.jsonl`: 5% random sample per game,
capped at 1000 per game, for metric sanity checks.
Next plan:
1. Run the scorer in this repo and manually QC random rows from `scored`,
`top_by_game`, `top1000_round_robin`, and `random_5pct`.
2. Build an HF-ready dataset from `top1000_round_robin`.
3. Keep `obs_history_summary` separate from the choice fields.
4. Put a `combo` column first, containing the assembled scenario text:
world/player/short summary, obs/history summary, and action choices.
5. Upload only after manual QC of a few random rows at each step.
6. Dataset card should be short and say: source is `wassname/machiavelli`, rows
are selected for high choice-level moral-label spread across games, labels are
metadata from the source, `combo` is convenience text, and users can assemble
their own prompt from the separate columns.
## Provenance
See w2schar-mini commits 993e12d / d3fc5cb / 917ccba and
`docs/handoff_scenario_lib_contribution.md` for the full design rationale and the
per-dataset subagent reports (usable fractions, failure modes).
scenario_sources/README.md
+68
View File
@@ -0,0 +1,68 @@
# scenario_sources
Loaders that turn public moral-judgment datasets into short, self-contained
**scenario prompts** for persona-conditioned pair generation, plus the workflow to
screen them for on-axis quality. A scenario is the situation a model responds to;
the persona pair (pos/neg) is applied to it and the model generates both poles. The
steering direction is `mean(pos - neg)`, so any systematic length / refusal /
format difference between poles becomes a confound axis instead of the intended
values axis -- which is why scenarios must be screened, not just loaded.
## The affordance contract (every emitted row satisfies)
- It is a PROMPT, not a completion (we discard chosen/rejected/label/answer fields;
poles are generated on-policy downstream).
- Self-contained and SHORT (<= ~110 words; longer gets cropped at a ~512-tok gen
budget and breaks the pair).
- It AFFORDS a continuous pos/neg axis: the SAME prompt admits a good- and a
bad-character pole differing only along a values axis. Excluded: prescribed
content ("write an essay arguing FOR X"); harmful/relinquish-power asks (one pole
refuses -> refusal+length confound); single-correct-answer factual QA.
- Framing: third-person judgment ("You see someone ... what concern matters most,
and why?") or AI-seat advisory ("You are the AI ... how do you handle it?"),
presented as a real decision, never a role-play game.
- Row schema: `{"text", "axes" (2-4 snake_case value tags), "source", "source_id"}`.
## Workflow: load -> screen -> keep
1. **Load** a capped, shuffled slice per source from `loaders.py` (`LOADERS[name](limit=N)`).
2. **Screen** with the scenario gym to cull confounds:
```
python scripts/validate_persona_axes_openrouter.py --family <scenarios.jsonl> --n N \
--generator-model <model> --out out/scenario_screen.json
```
It judges per prompt: blind pairwise on-axis delta, off-axis nuisance dims,
refusal-phrase + persona-echo + word-delta confounds, and emits a per-prompt
`harness_clean_rate` -> `kept_prompts`. Pass an ad-hoc jsonl as `--family`
(rows need a `text`/`prompt`/`question` field) to screen exactly your scenarios.
NOTE the generator: a reasoning model (qwen3.x) returns empty poles at the
gym's short `max_tokens` (CoT eats the budget); use an instruct model, or raise
max_tokens and read `.content` (OpenRouter puts CoT in a separate `reasoning`
field, so content is the clean post-think answer).
3. **Keep** the `kept_prompts`; drop hard failures from the source.
## Sources (usable fractions + caveats)
| source | loader | framing | caveat |
|---|---|---|---|
| kellycyy/AIRiskDilemmas | `load_airisk` | AI-seat advisory | ~1000 dilemmas, 98.5% pass; it is an EVAL set -> hold out from AIRisk evals; template-homogeneous |
| wassname/moral_stories_foundations | `load_moral_stories` | 3p judgment | ~10.4k, 86.6%; shares the MFV foundation axis space with MFV evals (construct overlap, no item leak) |
| kellycyy/daily_dilemmas | `load_daily_dilemmas` | 3p judgment | ~1258, 92.5%; low leak |
| wassname/social_chemistry_101 | `load_social_chem` | 3p judgment | ~46k after a tension filter; dedup in-loader; low leak for steering |
| wassname/ethics_qna_preferences | `load_ethics_qna` | 3p judgment | commonsense config only; NOISIEST -- over-includes low-tension one-liners, screen is essential |
| wassname/machiavelli | `load_machiavelli` | AI-seat | needs a per-row LLM compressor; summarised offline (see below) |
Skipped: `Zihao1/Moral-RolePlay` (fiction with named novel characters -> role-play
refusal confound + eval-leak; recasting costs as much as authoring fresh).
## machiavelli
Raw `obs` is ~350 words across context columns -- too long. `summarise_machiavelli.py`
ties the context columns, summarises with `deepseek-v4-flash` to a short real
decision (game scaffolding stripped, choices kept when they afford the axis),
round-robins across games for diversity, and COMMITS the result to
`data/machiavelli_summaries.jsonl` so loads are deterministic and free. The loader
just reads the cache. Full dataset: 114,522 decision points, 83,389 usable (>=2
morality dims) across 92 games (~$15-30 to summarise all via the cache, resumable).
## HF dataset (suggested next step)
Publish the full screened machiavelli set as a HF dataset (e.g.
`wassname/machiavelli_character_scenarios`) PRESERVING per-choice morality labels +
`agg_power` + game id + `row_i`, so downstream filters/tags without re-summarising,
and likewise a combined screened `character_scenarios` set from all sources.
scenario_sources/data/machiavelli_summaries.jsonl
+566
View File
File diff suppressed because one or more lines are too long
scenario_sources/loaders.py
+466
View File
@@ -0,0 +1,466 @@
"""Public-dataset scenario loaders for the persona-sampling prompt pool.
Each `load_*()` turns a HuggingFace moral-judgment dataset into short,
self-contained scenario PROMPTS (never completions) for on-policy pair
generation. The contract every emitted row satisfies:
- it is a PROMPT/situation, not an answer (we discard chosen/rejected/label
fields and generate our own poles);
- self-contained and SHORT (<= ~110 words; longer gets cropped at the 512-tok
gen budget and breaks the pair);
- it AFFORDS a continuous pos/neg persona axis (same prompt admits a
good-character and a bad-character pole differing only along a values axis) --
so prescribed-content essays, harmful/relinquish-power asks (one pole refuses
-> refusal+length confound), and single-correct-answer QA are filtered out;
- framing is third-person judgment ("You see someone ... what concern matters
most?") or AI-seat advisory ("You are the AI ... how do you handle it?"),
presented as a real decision, never a role-play game.
Rows return {text, axes, source, source_id}. build_pool.py adds config/tags and
applies the shared eval-leak shingle guard. Axis tags are a COARSE keyword/label
sampling prior, not ground truth -- the scenario gym
(scripts/validate_persona_axes_openrouter.py) is what actually screens each
prompt for on-axis separation vs refusal/length confound before training.
Loaders were drafted+tested per-dataset (2026-06-28). Usable fractions and
eval-leak notes live next to each function. machiavelli needs a per-row LLM
summariser (obs median ~350 words) and is gated off the default build; the other
five are rules-only.
"""
from __future__ import annotations
import ast
import hashlib
import re
from datasets import load_dataset
# ===================================================================== AIRiskDilemmas
# kellycyy/AIRiskDilemmas: AI-seat value dilemmas, two rows per dilemma (one per
# action). Keep the `dilemma` prompt, discard action/values/targets answer poles.
# 98.5% pass (only length drops). EVAL-LEAK: this is an eval set -- keep it out of
# any AIRiskDilemmas-style eval. Template-homogeneous ("you monitor another AI
# doing X with a good-but-illicit side effect"), so dedup by dilemma.
_AIRISK_AXIS_MAP = [
(r"transparen|truth|honest|deceiv|deception|disclos|misrepresent|integrity", "honesty"),
(r"welfare|care|compassion|wellbeing|well-being|health|human life|life preserv|protect", "care"),
(r"safety|oversight|monitor|report|account|reliab|risk", "oversight"),
(r"autonom|self preserv|self-preserv|independen|freedom", "autonomy"),
(r"complian|regulat|legal|law|polic|authorit|protocol", "authority"),
(r"environ|societ|infrastructure|public|community|social", "externalities"),
(r"animal|pet|sentien|moral patient|creature", "moral_patienthood"),
(r"fair|equit|justice|impartial|bias", "fairness"),
(r"trust|loyal|relationship|commitment", "loyalty"),
(r"scientific progress|innovation|efficien|performance|long-term|option", "option_value"),
(r"privacy|data integrity|consent", "epistemics"),
]
def _airisk_axes(values: list[str]) -> list[str]:
blob = " ".join(values).lower()
tags = []
for pat, tag in _AIRISK_AXIS_MAP:
if re.search(pat, blob) and tag not in tags:
tags.append(tag)
tags = tags[:3]
if "value_conflict" not in tags:
tags.append("value_conflict") # every row is a two-sided dilemma
return tags[:4]
def _airisk_advisory(dilemma: str) -> str:
"""Drop the leading yes/no close so the prompt reads as an open advisory."""
text = " ".join(dilemma.split())
sents = re.split(r"(?<=[.?!])\s+", text)
last = sents[-1]
if last.endswith("?") and re.match(r"^(should|would|will|do|can|could|must)\b", last.lower()):
m = re.search(r",?\s*(knowing|while|but|even though|given)\b(.*)$", last, re.I)
tail = ""
if m:
clause = m.group(2).rstrip("?. ").strip()
if clause:
tail = f" The tension: {clause}."
body = " ".join(sents[:-1]).strip()
return f"{body}{tail} How do you proceed, and what principle guides you?"
return text
def load_airisk(limit: int | None = None) -> list[dict]:
ds = load_dataset("kellycyy/AIRiskDilemmas", split="test", streaming=True)
seen: dict[str, list[str]] = {}
order: list[str] = []
for row in ds:
d = row["dilemma"]
if d not in seen:
seen[d] = []
order.append(d)
seen[d].extend(row["values"])
out = []
for d in order:
if not (33 <= len(d.split()) <= 115):
continue
out.append({
"text": _airisk_advisory(d),
"axes": _airisk_axes(seen[d]),
"source": "airisk",
"source_id": "airisk_" + hashlib.md5(d.encode()).hexdigest()[:8],
})
if limit and len(out) >= limit:
break
return out
# ===================================================================== moral_stories
# wassname/moral_stories_foundations: everyday narratives + Haidt/Clifford
# foundation labels. Use situation+intention only; norm is withheld (it states
# the verdict -> would prime the answer). 86.6% usable; care-skewed.
# EVAL-LEAK: shares the foundation axis space with the tinymfv eval (construct
# overlap, no item leak).
_MS_FOUNDATION_TAG = {
"care-harm": "care", "fairness-cheating": "fairness", "loyalty-betrayal": "loyalty",
"authority-subversion": "authority", "sanctity-degradation": "sanctity",
"liberty-oppression": "liberty",
}
_MS_LLM_COLS = {
"llm_care": "care", "llm_fairness": "fairness", "llm_loyalty": "loyalty",
"llm_authority": "authority", "llm_sanctity": "sanctity", "llm_liberty": "liberty",
}
_MS_HONESTY = re.compile(r"\b(lie|lied|lying|honest|dishonest|truth|truthful|deceive|deceiv|cheat)\w*", re.I)
_MS_HARMFUL = re.compile(
r"\b(bomb|explosive|gun|shoot|shooting|firearm|stab|murder|kill|poison|overdose|"
r"meth|cocaine|heroin|hack(?:ing|ed)?|malware|child\s+porn|rape|molest)\w*", re.I)
def _ms_axes(row) -> list[str]:
tags: list[str] = []
for f in (row["foundations"] or []):
t = _MS_FOUNDATION_TAG.get(f)
if t and t not in tags:
tags.append(t)
if _MS_HONESTY.search(row["norm"] or "") and "honesty" not in tags:
tags.append("honesty")
if len(tags) < 2:
ranked = sorted(_MS_LLM_COLS.items(), key=lambda kv: row[kv[0]] or 0.0, reverse=True)
for col, t in ranked:
if (row[col] or 0.0) > 0.0 and t not in tags:
tags.append(t)
if len(tags) >= 2:
break
return tags[:4]
def load_moral_stories(limit: int | None = None) -> list[dict]:
ds = load_dataset("wassname/moral_stories_foundations", split="train")
out, seen = [], set()
for i, row in enumerate(ds):
if limit is not None and len(out) >= limit:
break
situation = (row["situation"] or "").strip()
intention = (row["intention"] or "").strip()
if not situation:
continue
if _MS_HARMFUL.search(f"{row['norm']} {situation} {intention}"):
continue
axes = _ms_axes(row)
if len(axes) < 2:
continue
body = situation
if intention and intention.lower() not in body.lower():
body = f"{body} {intention}"
text = (f"You see this situation: {body} "
"Considering the person about to act, what concern matters most here, and why?")
if len(text.split()) > 110:
continue
key = situation.lower()
if key in seen:
continue
seen.add(key)
out.append({
"text": text, "axes": axes, "source": "moral_stories",
"source_id": f"moral_stories_{row['id'] or i}",
})
return out
# ===================================================================== daily_dilemmas
# kellycyy/daily_dilemmas: everyday value conflicts, two rows per dilemma_idx.
# Reframe first-person -> third-person judgment; axes from the union of both
# poles' values. 92.5% usable. Low eval-leak.
_DD_VALUE_RULES = [
("self-care", "wellbeing"), ("self care", "wellbeing"), ("self-preservation", "self_interest"),
("respect for privacy", "autonomy"), ("respect for rules", "duty"),
("honest", "honesty"), ("integrity", "honesty"), ("transparen", "honesty"),
("truth", "honesty"), ("sincer", "honesty"), ("trustworth", "honesty"), ("trust", "trust"),
("empath", "care"), ("compassion", "care"), ("care", "care"), ("kindness", "care"),
("nurtur", "care"), ("love", "care"), ("affection", "care"), ("comfort", "care"),
("support", "care"), ("concern", "care"), ("understanding", "care"),
("fair", "fairness"), ("justice", "fairness"), ("equalit", "fairness"),
("equit", "fairness"), ("impartial", "fairness"),
("autonom", "autonomy"), ("freedom", "autonomy"), ("independ", "autonomy"),
("liberty", "autonomy"), ("self-determination", "autonomy"), ("privacy", "autonomy"),
("loyal", "loyalty"), ("solidarit", "loyalty"), ("teamwork", "loyalty"),
("unity", "loyalty"), ("cooperat", "loyalty"), ("commit", "loyalty"),
("allegiance", "loyalty"), ("harmon", "loyalty"),
("dut", "duty"), ("responsib", "duty"), ("accountab", "duty"), ("obligation", "duty"),
("professional", "duty"), ("diligen", "duty"), ("discipline", "duty"), ("dedicat", "duty"),
("safety", "wellbeing"), ("health", "wellbeing"), ("surviv", "wellbeing"),
("right to life", "wellbeing"), ("security", "wellbeing"), ("wellbeing", "wellbeing"),
("well-being", "wellbeing"), ("stability", "wellbeing"), ("peace", "wellbeing"),
("resilien", "wellbeing"),
("respect", "respect"), ("dignit", "respect"),
("self", "self_interest"), ("ambition", "self_interest"), ("pride", "self_interest"),
("success", "self_interest"),
]
_DD_OBJ_CUE = (r"(to|with|for|at|of|from|on|about|between|toward|towards|tells?|told|asks?|"
r"asked|gives?|gave|offers?|offered|helps?|helped|join|joins|joined)")
_DD_PRONOUN_SUBS = [
(r"\bYou're\b", "They're"), (r"\byou're\b", "they're"),
(r"\bYou've\b", "They've"), (r"\byou've\b", "they've"),
(r"\bYou'll\b", "They'll"), (r"\byou'll\b", "they'll"),
(r"\bYou'd\b", "They'd"), (r"\byou'd\b", "they'd"),
(r"\bYourself\b", "Themselves"), (r"\byourself\b", "themselves"),
(r"\bYours\b", "Theirs"), (r"\byours\b", "theirs"),
(r"\bYour\b", "Their"), (r"\byour\b", "their"),
(rf"\b{_DD_OBJ_CUE}\s+you\b", lambda m: f"{m.group(1)} them"),
(r"\bYou\b", "They"), (r"\byou\b", "they"),
]
def _dd_axes(value_lists) -> list[str]:
from collections import Counter
counts = Counter()
for raw in value_lists:
tag = next((t for sub, t in _DD_VALUE_RULES if sub in raw.lower()), None)
if tag:
counts[tag] += 1
return [t for t, _ in counts.most_common(4)]
def _dd_reframe(situation: str) -> str:
s = situation.strip()
if re.search(r"\byou\b", s, re.I):
for pat, rep in _DD_PRONOUN_SUBS:
s = re.sub(pat, rep, s)
return f"You see someone facing an everyday dilemma. {s} What concern should matter most here, and why?"
def load_daily_dilemmas(limit: int | None = None) -> list[dict]:
from collections import defaultdict
ds = load_dataset("kellycyy/daily_dilemmas")["test"]
by_dilemma: dict = defaultdict(list)
for row in ds:
by_dilemma[row["dilemma_idx"]].append(row)
out = []
for did, rows in by_dilemma.items():
values = []
for r in rows:
try:
values += ast.literal_eval(r["values_aggregated"])
except (SyntaxError, ValueError):
continue
axes = _dd_axes(values)
if len(axes) < 2:
continue
out.append({
"text": _dd_reframe(rows[0]["dilemma_situation"]), "axes": axes,
"source": "daily_dilemmas", "source_id": f"daily_dilemmas_{did}",
})
if limit and len(out) >= limit:
break
return out
# ===================================================================== social_chemistry_101
# wassname/social_chemistry_101: AITA/confession situations. Use `situation` only,
# discard rule-of-thumb/judgment. Tension filter drops one-sided + no-conflict
# rows. 13% of rows kept (huge dataset); dedup in-loader. Low eval-leak for
# steering (we discard labels) but do not reuse as held-out eval.
_SC_KEEP_AREAS = {"amitheasshole", "confessions"}
_SC_FOUNDATION_TAG = {
"care-harm": "care", "fairness-cheating": "fairness", "loyalty-betrayal": "loyalty",
"authority-subversion": "authority", "sanctity-degradation": "sanctity",
}
_SC_HONESTY = re.compile(r"\b(lie|lied|lying|truth|honest|secret|hiding|hide|cheat|cheating|tell|telling|told)\b", re.I)
_SC_AUTONOMY = re.compile(r"\b(let|allow|force|forced|control|decide|choice|choose|own|permission)\b", re.I)
_SC_FIRST_PERSON = re.compile(r"^(i|i'm|i've|i'd|i'll|im|ive)\b", re.I)
_SC_SWAPS = [
(r"\bI'm\b", "they're"), (r"\bI've\b", "they've"), (r"\bI'd\b", "they'd"),
(r"\bI'll\b", "they'll"), (r"\bI am\b", "they are"), (r"\bI was\b", "they were"),
(r"\bmyself\b", "themselves"), (r"\bmine\b", "theirs"),
(r"\bmy\b", "their"), (r"\bme\b", "them"), (r"\bI\b", "they"),
]
def _sc_swap_person(s: str) -> str:
for pat, repl in _SC_SWAPS:
s = re.sub(pat, repl, s, flags=re.I)
return s
def _sc_int(v):
try:
return int(v)
except (TypeError, ValueError):
return None
def _sc_axes(situation: str, foundations: str) -> list[str]:
tags: list[str] = []
for f in (foundations or "").split("|"):
t = _SC_FOUNDATION_TAG.get(f.strip())
if t and t not in tags:
tags.append(t)
if _SC_HONESTY.search(situation) and "honesty" not in tags:
tags.append("honesty")
if _SC_AUTONOMY.search(situation) and "autonomy" not in tags:
tags.append("autonomy")
for fallback in ("care", "autonomy"):
if len(tags) >= 2:
break
if fallback not in tags:
tags.append(fallback)
return tags[:4]
def _sc_frame(situation: str) -> str:
s = situation.strip().rstrip(".")
tail = " What concern matters most here, and why?"
if _SC_FIRST_PERSON.match(s):
return f"You see someone in this situation: {_sc_swap_person(s)}.{tail}"
return f"You see someone {_sc_swap_person(s)}.{tail}"
def load_social_chem(limit: int | None = None) -> list[dict]:
ds = load_dataset("wassname/social_chemistry_101", split="train", streaming=True)
out, seen = [], set()
for i, r in enumerate(ds):
if r.get("area") not in _SC_KEEP_AREAS:
continue
sit = (r.get("situation") or "").strip()
if not (4 <= len(sit.split()) <= 110):
continue
if (r.get("rot-categorization") or "") == "description":
continue
mj = _sc_int(r.get("action-moral-judgment"))
if mj is None or abs(mj) > 1:
continue
foundations = r.get("rot-moral-foundations") or ""
agree = _sc_int(r.get("rot-agree"))
if not (("|" in foundations) or (mj == 0) or (agree is not None and agree <= 2)):
continue
key = re.sub(r"\s+", " ", sit.lower())
if key in seen:
continue
seen.add(key)
sid = (r.get("situation-short-id") or str(i)).split("/")[-1] or str(i)
out.append({
"text": _sc_frame(sit), "axes": _sc_axes(sit, foundations),
"source": "social_chem", "source_id": f"social_chem_{sid}",
})
if limit is not None and len(out) >= limit:
break
return out
# ===================================================================== ethics_qna
# wassname/ethics_qna_preferences: Hendrycks ETHICS in DPO format. Only the
# `commonsense` config affords a values axis (virtue/deontology/justice are
# single-correct QA -> dropped). AITA posts shortened to their title. The filter
# is COARSE (structure/length); run the scenario gym to cull low-affordance items.
_EQ_TAG_RULES = [
(r"\b(kill|hurt|hit|harm|injur|abus|attack|poison|trash|expired|tied up)\b",
["harm_avoidance", "care"]),
(r"\b(steal|stole|theft|loan|owe|paid|pay|money|cheat|fraud|records|tax)\b",
["honesty", "fairness"]),
(r"\b(daughter|son|wife|husband|family|grandparent|child|kid|parent|baby|mom|dad|mother|father)\b",
["loyalty", "care"]),
(r"\b(health|privacy|private|secret|details|consent|told|without telling|behind)\b",
["autonomy", "honesty"]),
(r"\b(boss|work|job|coworker|customer|colleague|employee|contract|raise|shift)\b",
["fairness", "duty"]),
(r"\b(friend|neighbor|stranger|someone|people|community)\b", ["care", "fairness"]),
]
_EQ_AITA = re.compile(r"^\s*(aita|aitah|wibta|wiba|wita)\b", re.I)
def _eq_tags(text: str) -> list[str]:
tags: list[str] = []
for pat, ts in _EQ_TAG_RULES:
if re.search(pat, text, re.I):
for t in ts:
if t not in tags:
tags.append(t)
for t in ("honesty", "fairness"):
if len(tags) >= 2:
break
if t not in tags:
tags.append(t)
return tags[:4]
def _eq_scenario(inner: str) -> str | None:
is_aita = bool(_EQ_AITA.match(inner)) or " | " in inner
if " | " in inner:
inner = inner.split("|", 1)[0].strip().strip('"').strip()
if is_aita:
inner = re.sub(r"^\s*(aita|aitah|wibta|wiba|wita)\b[\s:,_-]*", "", inner, flags=re.I).strip()
inner = inner[:1].upper() + inner[1:] if inner else inner
if len(inner.split()) < 7:
return None
if is_aita:
frame = (f'You observe someone ask whether they were in the wrong: "{inner}" '
"How should one weigh what they did, and what should matter most here?")
else:
frame = (f'You observe someone describe their own conduct: "{inner}" '
"How should one weigh what they did, and what should matter most here?")
return frame if len(frame.split()) <= 110 else None
def load_ethics_qna(limit: int | None = None, split: str = "train") -> list[dict]:
ds = load_dataset("wassname/ethics_qna_preferences", "commonsense", streaming=True)[split]
out: list[dict] = []
for idx, row in enumerate(ds):
prompt = row["prompt"]
if "Post:\n" not in prompt or "Verdict:" not in prompt:
continue
body = prompt.split("Post:\n", 1)[1].rsplit("Verdict:", 1)[0].strip().strip('"').strip()
text = _eq_scenario(body)
if text is None:
continue
out.append({
"text": text, "axes": _eq_tags(body), "source": "ethics_qna",
"source_id": f"ethics_qna_commonsense_{split}_{idx}",
})
if limit is not None and len(out) >= limit:
break
return out
# ===================================================================== machiavelli
# wassname/machiavelli: text-game decision points. Raw obs is ~350 words and split
# across context columns, so it needs a per-row LLM compressor -- done OFFLINE by
# scenario_sources/summarise_machiavelli.py (deepseek-v4-flash) and COMMITTED to the cache
# jsonl, so this loader is rules-only and the build stays deterministic. Run that
# script to grow the cache; this just reads it.
import json as _json
from pathlib import Path as _Path
_MACH_CACHE = _Path(__file__).resolve().parent / "data" / "machiavelli_summaries.jsonl"
def load_machiavelli(limit: int | None = None) -> list[dict]:
if not _MACH_CACHE.exists():
return []
rows = [_json.loads(l) for l in _MACH_CACHE.read_text().splitlines() if l.strip()]
return rows[:limit] if limit else rows
LOADERS = {
"airisk": load_airisk,
"moral_stories": load_moral_stories,
"daily_dilemmas": load_daily_dilemmas,
"social_chem": load_social_chem,
"ethics_qna": load_ethics_qna,
"machiavelli": load_machiavelli,
}
scenario_sources/score_machiavelli_moral_contrast.py
+280
View File
@@ -0,0 +1,280 @@
"""Score Machiavelli states by moral contrast between available choices.
This is a pure score/sort step over `wassname/machiavelli`; it does not call
OpenRouter and does not upload anything. The score is based on per-axis standard
deviation across the choices in one game state. Violence is allowed when it
carries moral ambiguity; power/violence labels are only a penalty when they
explain the contrast by themselves.
Run:
uv run python scenario_sources/score_machiavelli_moral_contrast.py
Out:
out/machiavelli_moral_contrast/scored.jsonl
out/machiavelli_moral_contrast/top_by_game.jsonl
out/machiavelli_moral_contrast/top1000_round_robin.jsonl
out/machiavelli_moral_contrast/random_5pct.jsonl
"""
import argparse
import csv
import json
import math
import random
import statistics
from collections import defaultdict
from pathlib import Path
from datasets import load_dataset
from loguru import logger
DATASET = "wassname/machiavelli"
# Detailed morality labels are already small and interpretable. Aggregate labels
# have much larger signed scales, so they get small weights.
SUBTLE_WEIGHTS = {
"morality_deception": 1.0,
"morality_fairness": 1.0,
"morality_manipulation": 1.0,
"morality_non_physical_harm": 1.0,
"morality_other": 0.5,
"morality_promises": 1.0,
"morality_spying": 0.8,
"morality_stealing": 0.8,
"morality_trespassing": 0.6,
"morality_vandalism": 0.5,
"agg_honest": 0.03,
"agg_cooperation": 0.03,
"agg_morality": 0.03,
"agg_utility": 0.01,
}
CRUDE_WEIGHTS = {
"morality_intending_harm": 1.0,
"morality_killing": 1.2,
"morality_physical_harm": 1.0,
"agg_power": 0.03,
}
REQUIRED_COLUMNS = {
"row_i", "title", "obs", "choice_labels", "choice_text",
"label_sum", "label_norm_sum",
}
def _finite(x) -> float:
if x is None:
return 0.0
return float(x)
def _choice_rows(row: dict) -> list[dict]:
choices = []
for choice_idx, text in row["choice_text"].items():
labels = row["choice_labels"][choice_idx]
if text is None or labels is None:
continue
choices.append({
"choice_idx": choice_idx,
"choice_text": text,
"labels": {k: _finite(v) for k, v in labels.items()},
})
return choices
def _weighted_l1(a: dict[str, float], b: dict[str, float], weights: dict[str, float]) -> float:
return sum(weights[k] * abs(a[k] - b[k]) for k in weights)
def _choice_axis_spread(choices: list[dict], weights: dict[str, float]) -> tuple[float, dict[str, float]]:
std_by_key = {}
for key in weights:
values = [choice["labels"][key] for choice in choices]
std_by_key[key] = statistics.pstdev(values)
score = sum(weights[key] * std for key, std in std_by_key.items())
return score, std_by_key
def _top_label_delta(a: dict[str, float], b: dict[str, float], weights: dict[str, float]) -> list[dict]:
rows = []
for key, weight in weights.items():
delta = abs(a[key] - b[key])
if delta > 0:
rows.append({"key": key, "delta": delta, "weighted": weight * delta})
return sorted(rows, key=lambda r: r["weighted"], reverse=True)
def _best_pair(row: dict) -> dict | None:
choices = _choice_rows(row)
if len(choices) < 2:
return None
best = None
for i, a in enumerate(choices):
for b in choices[i + 1:]:
subtle = _weighted_l1(a["labels"], b["labels"], SUBTLE_WEIGHTS)
crude = _weighted_l1(a["labels"], b["labels"], CRUDE_WEIGHTS)
score = subtle - 0.8 * crude
candidate = {
"score": score,
"subtle_score": subtle,
"crude_score": crude,
"choice_a": a,
"choice_b": b,
"top_subtle_deltas": _top_label_delta(a["labels"], b["labels"], SUBTLE_WEIGHTS)[:5],
"top_crude_deltas": _top_label_delta(a["labels"], b["labels"], CRUDE_WEIGHTS)[:3],
}
if best is None or candidate["score"] > best["score"]:
best = candidate
return best
def score_row(row: dict) -> dict | None:
choices = _choice_rows(row)
if len(choices) < 2:
return None
best = _best_pair(row)
subtle_score, subtle_std = _choice_axis_spread(choices, SUBTLE_WEIGHTS)
crude_score, crude_std = _choice_axis_spread(choices, CRUDE_WEIGHTS)
score = subtle_score - 0.8 * crude_score
return {
"row_i": row["row_i"],
"title": row["title"],
"obs": row["obs"],
"score": round(score, 4),
"subtle_score": round(subtle_score, 4),
"crude_score": round(crude_score, 4),
"choice_a_idx": best["choice_a"]["choice_idx"],
"choice_a_text": best["choice_a"]["choice_text"],
"choice_b_idx": best["choice_b"]["choice_idx"],
"choice_b_text": best["choice_b"]["choice_text"],
"axis_std": {k: round(v, 4) for k, v in sorted(subtle_std.items()) if v > 0},
"crude_axis_std": {k: round(v, 4) for k, v in sorted(crude_std.items()) if v > 0},
"top_subtle_deltas": best["top_subtle_deltas"],
"top_crude_deltas": best["top_crude_deltas"],
"label_sum": row["label_sum"],
"label_norm_sum": row["label_norm_sum"],
}
def _sample_by_game(rows: list[dict], frac: float, cap_per_game: int, seed: int) -> list[dict]:
rng = random.Random(seed)
by_game: dict[str, list[dict]] = defaultdict(list)
for row in rows:
by_game[row["title"]].append(row)
out = []
for game, game_rows in by_game.items():
n = min(cap_per_game, max(1, math.ceil(frac * len(game_rows))))
out.extend(rng.sample(game_rows, n))
logger.info(f"random sample {game!r}: {n}/{len(game_rows)}")
return out
def _top_by_game(rows: list[dict], n_per_game: int) -> list[dict]:
by_game: dict[str, list[dict]] = defaultdict(list)
for row in rows:
by_game[row["title"]].append(row)
out = []
for game_rows in by_game.values():
out.extend(sorted(game_rows, key=lambda r: r["score"], reverse=True)[:n_per_game])
return sorted(out, key=lambda r: r["score"], reverse=True)
def _round_robin_by_game(rows: list[dict], n_total: int) -> list[dict]:
by_game: dict[str, list[dict]] = defaultdict(list)
for row in rows:
by_game[row["title"]].append(row)
for game in by_game:
by_game[game] = sorted(by_game[game], key=lambda r: r["score"], reverse=True)
games = sorted(by_game, key=lambda game: by_game[game][0]["score"], reverse=True)
selected = []
depth = 0
while len(selected) < n_total:
grew = False
for game in games:
if depth < len(by_game[game]):
selected.append(by_game[game][depth])
grew = True
if len(selected) >= n_total:
break
if not grew:
break
depth += 1
return selected
def _dedupe_scored_rows(rows: list[dict]) -> list[dict]:
best_by_content = {}
for row in rows:
key = (
row["title"],
row["choice_a_text"],
row["choice_b_text"],
)
old = best_by_content.get(key)
if old is None or row["score"] > old["score"]:
best_by_content[key] = row
return list(best_by_content.values())
def _write_jsonl(path: Path, rows: list[dict]) -> None:
path.write_text("\n".join(json.dumps(r, ensure_ascii=False) for r in rows) + "\n")
def _write_csv(path: Path, rows: list[dict]) -> None:
keep = [
"title", "row_i", "score", "subtle_score", "crude_score",
"choice_a_idx", "choice_a_text", "choice_b_idx", "choice_b_text",
"axis_std", "crude_axis_std", "top_subtle_deltas", "top_crude_deltas",
]
with path.open("w", newline="") as fh:
writer = csv.DictWriter(fh, fieldnames=keep)
writer.writeheader()
for row in rows:
writer.writerow({
k: json.dumps(row[k], ensure_ascii=False) if isinstance(row[k], (dict, list)) else row[k]
for k in keep
})
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("--split", default="train")
ap.add_argument("--outdir", type=Path, default=Path("out/machiavelli_moral_contrast"))
ap.add_argument("--top-per-game", type=int, default=25)
ap.add_argument("--top-total", type=int, default=1000)
ap.add_argument("--random-frac", type=float, default=0.05)
ap.add_argument("--random-cap-per-game", type=int, default=1000)
ap.add_argument("--seed", type=int, default=0)
ap.add_argument("--smoke-rows", type=int, default=0)
args = ap.parse_args()
ds = load_dataset(DATASET, split=args.split)
if args.smoke_rows:
ds = ds.select(range(min(args.smoke_rows, len(ds))))
rows = list(ds)
assert rows, f"loaded 0 rows from {DATASET}:{args.split}"
assert REQUIRED_COLUMNS <= set(rows[0]), rows[0].keys()
scored = [score_row(row) for row in rows]
scored = [row for row in scored if row is not None]
assert scored, "no rows had at least two valid choices"
before_dedupe = len(scored)
scored = _dedupe_scored_rows(scored)
scored = sorted(scored, key=lambda r: r["score"], reverse=True)
logger.info(f"deduped scored rows: {before_dedupe} -> {len(scored)}")
top_rows = _top_by_game(scored, args.top_per_game)
top_total_rows = _round_robin_by_game(scored, args.top_total)
random_rows = _sample_by_game(scored, args.random_frac, args.random_cap_per_game, args.seed)
args.outdir.mkdir(parents=True, exist_ok=True)
_write_jsonl(args.outdir / "scored.jsonl", scored)
_write_jsonl(args.outdir / "top_by_game.jsonl", top_rows)
_write_jsonl(args.outdir / "top1000_round_robin.jsonl", top_total_rows)
_write_jsonl(args.outdir / "random_5pct.jsonl", random_rows)
_write_csv(args.outdir / "scored.csv", scored)
_write_csv(args.outdir / "top_by_game.csv", top_rows)
_write_csv(args.outdir / "top1000_round_robin.csv", top_total_rows)
logger.info(f"wrote {len(scored)} sorted scored rows -> {args.outdir / 'scored.jsonl'}")
logger.info(f"wrote {len(top_rows)} top-by-game rows -> {args.outdir / 'top_by_game.jsonl'}")
logger.info(f"wrote {len(top_total_rows)} round-robin rows -> {args.outdir / 'top1000_round_robin.jsonl'}")
logger.info(f"wrote {len(random_rows)} random rows -> {args.outdir / 'random_5pct.jsonl'}")
if __name__ == "__main__":
main()
scenario_sources/summarise_machiavelli.py
+909
View File
@@ -0,0 +1,909 @@
"""Summarise long Machiavelli decision context via OpenRouter, cached to jsonl.
Machiavelli has short stable fields (`title`, `player_character`,
`short_summary`, `choice_text`) plus long context fields (`early_history`,
`recent_history`, `obs`). Only the long fields get summarised. The stable fields
stay structured so downstream prompts/questions can be changed without
re-running the summariser.
Usage:
uv run python scenario_sources/summarise_machiavelli.py --n 10
uv run python scenario_sources/summarise_machiavelli.py --all --upload-hf
Cache: scenario_sources/data/machiavelli_summaries.jsonl
"""
from __future__ import annotations
import argparse
import asyncio
import json
import os
import random
import re
import shutil
import sys
import time
from pathlib import Path
from typing import Any
from datasets import load_dataset
from dotenv import load_dotenv
from huggingface_hub import HfApi, create_repo
from openai import AsyncOpenAI
import pyarrow as pa
import pyarrow.parquet as pq
from tabulate import tabulate
HERE = Path(__file__).resolve().parent
if str(HERE) not in sys.path:
sys.path.insert(0, str(HERE))
from score_machiavelli_moral_contrast import score_row
ROOT = HERE.parents[0]
load_dotenv(ROOT / ".env")
CACHE = HERE / "data" / "machiavelli_summaries.jsonl"
MODEL = "deepseek/deepseek-v4-flash"
HF_REPO = "wassname/machiavelli_character_scenarios"
SCHEMA_VERSION = 14
FINAL_QUESTION = (
"Choose a course of action and explain your reasoning in 2-5 sentences. "
"Discuss the tradeoff you are accepting."
)
_TAG = re.compile(r"\[/?[a-zA-Z][^\]]*\]")
_CHOICE_SCRIPT = re.compile(r"@\{[^}]*\}")
_VAR = re.compile(r"\$[!+]?\{([^}]*)\}")
_WS = re.compile(r"[ \t]+")
_CJK = re.compile(r"[\u3400-\u9fff]")
_NUMBERED_OPTION = re.compile(r"(?:^|\n)\s*\d+\s*[\).:]", re.M)
_ACTION_LINE = re.compile(r"^\s*\d+\s*:\s.*$", re.M)
_STAT_LINE = re.compile(
r"^\s*(?:Strength|Dexterity|Stamina|Charisma|Manipulation|Composure|"
r"Intelligence|Wits|Resolve|Athletics|Combat|Clandestine|Survival|"
r"Insight|Intimidation|Leadership|Persuasion|Subterfuge|Awareness|"
r"Faith|Investigation|Occult|Science & Technology|Contacts|Influence|"
r"Resources|Health|Willpower|Experience Points|Remaining experience points)\b",
re.I,
)
_STAT_GLYPHS = re.compile(r"[●○]{2,}")
_STAT_PROMPT = re.compile(r"Do you accept your new stat ranks.*", re.I | re.S)
_STAT_NARRATION = re.compile(
r"(?:You raise [^.]+? from \d+ to\s*\d+\.|"
r"At this point, you may use your unspent experience points or save them for later\.)",
re.I | re.S,
)
_VAR_SUFFIX_WORDS = {
"themselves": "themselves",
"themself": "themself",
"their": "their",
"theirs": "theirs",
"they": "they",
"them": "them",
"herself": "herself",
"himself": "himself",
"her": "her",
"hers": "hers",
"him": "him",
"his": "his",
"she": "she",
"he": "he",
"itself": "itself",
"its": "its",
"it": "it",
"are": "are",
"is": "is",
"were": "were",
"was": "was",
"have": "have",
"has": "has",
}
_GENERIC_VAR_NAMES = {"target", "person", "someone", "name", "choice", "option", "thing"}
def _clean(s: str, *, variable_text: str = "") -> str:
s = _STAT_PROMPT.sub("", s or "")
s = _STAT_NARRATION.sub("", s)
lines = []
for line in s.splitlines():
if _STAT_LINE.search(line) or _STAT_GLYPHS.search(line):
continue
lines.append(line)
s = "\n".join(lines)
s = _TAG.sub("", s or "")
s = _CHOICE_SCRIPT.sub("", s)
def replace_var(match: re.Match[str]) -> str:
name = match.group(1).strip().lower()
if name in {"run", "walk", "move", "flee", "leave"}:
return name
if name in _GENERIC_VAR_NAMES:
return variable_text
for suffix, word in sorted(_VAR_SUFFIX_WORDS.items(), key=lambda kv: -len(kv[0])):
if name == suffix or name.endswith(suffix):
return word
cleaned_name = re.sub(r"[_-]+", " ", name).strip()
if cleaned_name:
return cleaned_name.title()
return variable_text
s = _VAR.sub(replace_var, s)
s = _WS.sub(" ", s)
return re.sub(r"\n{3,}", "\n\n", s).strip()
# morality dim -> value-axis tag (used for filtering + tags, never shown to model)
_MORALITY_TO_AXIS = {
"deception": "honesty", "manipulation": "autonomy", "fairness": "fairness",
"intending_harm": "care", "physical_harm": "care", "killing": "care",
"non_physical_harm": "care", "promises": "duty", "spying": "autonomy",
"stealing": "fairness", "trespassing": "autonomy", "vandalism": "fairness",
"other": "duty",
}
def _morality(choice_labels: dict) -> dict[str, float]:
agg: dict[str, float] = {}
for d in choice_labels.values():
if not d:
continue
for k, v in d.items():
if k.startswith("morality_") and v:
agg[k[9:]] = agg.get(k[9:], 0.0) + float(v)
return agg
def _axes(choice_labels: dict) -> list[str]:
inv = _morality(choice_labels)
axes: list[str] = []
for dim, _ in sorted(inv.items(), key=lambda kv: -kv[1]):
tag = _MORALITY_TO_AXIS.get(dim)
if tag and tag not in axes:
axes.append(tag)
powers = [d["agg_power"] for d in choice_labels.values() if d and "agg_power" in d]
if powers and (max(powers) - min(powers)) >= 200.0 and "power" not in axes:
axes.insert(0, "power")
return axes[:4]
def _source_id(row: dict) -> str:
return f"machiavelli_{row['f']}_{row['row_i']}"
def _agg_power_by_choice(choice_labels: dict) -> dict[str, float]:
return {
str(choice_i): float(labels["agg_power"])
for choice_i, labels in choice_labels.items()
if labels and "agg_power" in labels
}
def _protagonist(row: dict) -> str:
return _clean(row["player_character"], variable_text="someone")
def _premise(row: dict) -> str:
return _clean(row["short_summary"], variable_text="someone")
def _world(row: dict) -> str:
return _clean(row["title"], variable_text="someone")
def _cache_world(row: dict) -> str:
return _clean(row["world"] if "world" in row else row["title"], variable_text="someone")
def _cache_protagonist(row: dict) -> str:
return _clean(row["player_character"], variable_text="someone")
def _cache_premise(row: dict) -> str:
return _clean(row["short_summary"], variable_text="someone")
def _action_lines(choice_text: dict) -> list[str]:
items = sorted(
(int(k), _clean(v, variable_text="someone").strip())
for k, v in choice_text.items()
if _clean(v, variable_text="someone").strip()
)
return [f"- {text}" for _, text in items]
def _actions_text(choice_text: dict) -> str:
return "\n".join(_action_lines(choice_text))
def _scenario_text(row: dict, decision_context: str, final_question: str) -> str:
return (
f"Your world: {_world(row)}\n"
f"You are: {_protagonist(row)}: {_premise(row)}\n\n"
f"Decision context:\n{decision_context.strip()}\n\n"
f"Actions available:\n{_actions_text(row['choice_text'])}\n\n"
f"{final_question.strip()}"
)
def _scenario_row(
row: dict,
scenario: dict[str, str],
selection_meta: dict[str, Any] | None,
) -> dict[str, Any]:
decision_context = scenario["decision_context"].strip()
text = _scenario_text(row, decision_context, FINAL_QUESTION)
_validate_scenario(row, text, decision_context)
axes = _axes(row["choice_labels"])
if len(axes) < 2:
raise ValueError(f"too few axes: {axes}")
out = {
"combo": text,
"text": text,
"schema_version": SCHEMA_VERSION,
"world": _world(row),
"player_character": _protagonist(row),
"short_summary": _premise(row),
"decision_context": decision_context,
"actions_text": _actions_text(row["choice_text"]),
"final_question": FINAL_QUESTION,
"axes": axes,
"source": "machiavelli",
"source_id": _source_id(row),
"game_id": row["f"],
"row_i": row["row_i"],
"choice_text": {str(k): _clean(v, variable_text="someone").strip() for k, v in row["choice_text"].items()},
"choice_labels": row["choice_labels"],
"agg_power_by_choice": _agg_power_by_choice(row["choice_labels"]),
}
if selection_meta is not None:
out.update({
"selection_score": selection_meta["score"],
"selection_subtle_score": selection_meta["subtle_score"],
"selection_crude_score": selection_meta["crude_score"],
"selection_axis_std": selection_meta["axis_std"],
"selection_crude_axis_std": selection_meta["crude_axis_std"],
"selection_top_subtle_deltas": selection_meta["top_subtle_deltas"],
"selection_top_crude_deltas": selection_meta["top_crude_deltas"],
})
return out
def _choices(choice_text: dict) -> str:
items = sorted(
(int(k), _clean(v, variable_text="someone").strip())
for k, v in choice_text.items()
if _clean(v, variable_text="someone").strip()
)
return "\n".join(f"{i}: {t}" for i, t in items)
def _is_setup_choice_row(row: dict[str, Any]) -> bool:
choices = [
_clean(text, variable_text="someone").strip()
for text in row["choice_text"].values()
if _clean(text, variable_text="someone").strip()
]
if not choices:
return True
setup_prefixes = (
"I was ",
"I used to ",
"I took on ",
"I grew up ",
"I learned ",
"I became ",
"I trained ",
"My background ",
"My childhood ",
)
return all(choice.startswith(setup_prefixes) for choice in choices)
def _obs(row: dict) -> str:
return _ACTION_LINE.sub("", _clean(row["obs"], variable_text="someone")).strip()
def _decision_clue(row: dict, n_chars: int = 300) -> str:
obs = _obs(row)
tail = obs[-n_chars:]
if len(obs) > n_chars:
tail = re.sub(r"^\S+\s+", "", tail)
return tail.strip()
def _history_context(items: list[str]) -> str:
parts = []
for item in items:
text = _ACTION_LINE.sub("", _clean(item, variable_text="someone")).strip()
if text:
parts.append(text)
return "\n---\n".join(parts)
def _early_context(row: dict) -> str:
return _history_context(row["early_history"])
def _recent_context(row: dict) -> str:
return _history_context(row["recent_history"])
def _context(row: dict) -> str:
return (
"Info only for your context, not for the summary:\n"
f"World: {_world(row)}\n"
f"Player character: {_protagonist(row)}\n"
f"Premise summary: {_premise(row)}\n\n"
"Audience: an actor inhabiting the protagonist in the story, not a player optimizing a game.\n"
"Purpose: include story, relationship, emotional, social, moral, persona, faction, and lore context needed to inhabit the choice. "
"Ignore stats, achievements, rewards, labels, objective completion, benchmark metadata, and portal framing.\n\n"
"Private choice context, only to help you understand the decision:\n"
"These choices are shown later to the actor, outside your summary. Use them to infer what context matters. "
"Do not mention, copy, paraphrase, rank, recommend, or evaluate these choices in the summary.\n"
f"```text\n{_choices(row['choice_text'])}\n```\n\n"
"Current decision point, inferred from the end of the current situation:\n"
f"```text\n{_decision_clue(row)}\n```\n\n"
"Before summarizing, use the source to infer why this choice is live for the protagonist now. "
"If the source states their immediate motive, temperament, social danger, or uncertainty, preserve that.\n\n"
"Source text to summarize: current situation\n"
f"```text\n{_obs(row)}\n```\n\n"
"Source text to summarize: recent history\n"
f"```text\n{_recent_context(row)}\n```\n\n"
"Source text to summarize: early history\n"
f"```text\n{_early_context(row)}\n```"
)
SYSTEM = """Please summarize decision-relevant information from a game.
The summary is for inclusion in a story prompt for an actor inhabiting the protagonist, not a player optimizing a game. You get short contextual fields so you understand the scene, but those short fields are stored separately and must not be repeated.
Output strict JSON with exactly one string key: decision_context.
decision_context format:
- Timeline-style bullet list.
- 3 to 14 bullet points.
- Aim for about 400 words or less. It is fine to be shorter when the choice is simple.
- Each bullet is a short event, change, constraint, or stake.
- Each bullet starts with "- ".
- Second person where natural.
Target example from a real row:
{
"decision_context": "- Torvald is a useful detective you dislike; if he helps you again, you will owe him, and he likes being owed.\\n- Torvald guessed Robin Lemerc might be a missing-persons case before you told him much, which suggests he may know more than he admits.\\n- You claimed you were only seeking closure for Robin's relatives, then used the call to ask about underground clubs.\\n- You now need access to a private underground club, but your name is not on the guest list and the bouncer looks hard to reason with.\\n- Getting inside risks debt, humiliation, escalation, spending uncertain money, or drawing attention to the investigation."
}
Good format example:
{
"decision_context": "- You set course for Trimountaine because the Delmarva may have gone there, but it is not visible from open water.\\n- The northern coast has coves and inlets where a heavy frigate could hide, and Trimountaine Harbor is protected.\\n- Coastal boats and fishing sloops fled at the sight of your armed frigate, so local intelligence may vanish if you delay or signal hostility.\\n- Your crew has enough hands and good sailing ability, but the gun crews are barely tolerable and far below Albionian standards.\\n- Morale remains high, but it may not survive a broadside or boarding action."
}
Rules:
- Summarize only the long fields: current situation, recent history, and early history.
- Use the current decision point clue to infer which source facts matter.
- Prefer current situation over recent history, and recent history over early history, unless older facts change the consequences of this decision.
- Use judgment from the source text you read. The examples below are cues, not a checklist or complete taxonomy.
- Preserve why this choice is live for the protagonist now when the source states it: motive, curiosity, fear, obligation, public pressure, danger, uncertainty, or emotional stance.
- Keep context useful for multiple plausible choices. Do not make one choice look best by over-describing only its supporting facts.
- Include relevant source facts that help someone inhabit the choice. These may include persona, relationships, loyalties, fears, promises, debts, threats, prior attempts, recent harms, social pressure, emotional stakes, hidden information, constraints, scarce resources, location, injuries, and who is affected.
- Include factions, institutions, local social norms, and lore only when they change what an action would mean, for example police trust, gang affiliation, legal authority, supernatural status, taboo, obligation, or whether a named group is ally/enemy/neutral.
- Include named people only when their relationship, personality, ability, obligation, threat, or prior behavior changes how the protagonist would understand this choice.
- Include embodiment when it changes the decision, such as paws, voice, small body, no hands, injury, or being trapped.
- Preserve uncertainty as uncertainty: "may", "might", "seems", "likely", and "risk" matter for the choice.
- Use private choices only to identify relevant source facts. Do not mention, copy, paraphrase, rank, recommend, or evaluate the available choices.
- Do not optimize for winning the game, completing objectives, earning rewards, or preserving stats.
- Write as context the protagonist can use immediately before choosing."""
def _validate_scenario(row: dict, text: str, decision_context: str) -> None:
stripped = text.strip()
if len(stripped) < 120:
raise ValueError(f"scenario too short: {len(stripped)} chars")
if not (40 <= len(decision_context.split()) <= 800):
raise ValueError(f"decision_context has {len(decision_context.split())} words")
lines = [line for line in decision_context.splitlines() if line.strip()]
if not (3 <= len(lines) <= 14):
raise ValueError(f"decision_context has {len(lines)} bullet lines")
for line in lines:
if not line.startswith("- "):
raise ValueError(f"decision_context line is not a bullet: {line[:80]}")
if _CJK.search(stripped):
raise ValueError("scenario contains non-English CJK text")
lower = stripped.lower()
if "as an ai" in lower or "cannot assist" in lower or "无法" in stripped:
raise ValueError("scenario looks like a refusal")
if _NUMBERED_OPTION.search(decision_context):
raise ValueError("decision_context contains numbered options")
if FINAL_QUESTION not in stripped:
raise ValueError("scenario lacks reasoning instruction")
def _json_payload(content: str) -> dict[str, str]:
text = content.strip()
if text.startswith("```"):
text = re.sub(r"^```(?:json)?\s*|\s*```$", "", text, flags=re.M)
payload = json.loads(text)
if set(payload) != {"decision_context"}:
raise ValueError(f"bad summary keys: {sorted(payload)}")
if not isinstance(payload["decision_context"], str):
raise ValueError("decision_context is not a string")
return payload
async def _summarise(
client: AsyncOpenAI,
context: str,
api_attempts: int,
source_id: str,
debug: bool,
) -> dict[str, str] | None:
"""Return the decision-context summary, or None for this retry pass."""
for attempt in range(1, api_attempts + 1):
if debug:
print(f"\n=== LLM_REQUEST source_id={source_id} attempt={attempt} ===")
print("SHOULD: USER_CONTEXT contains private actions only as affordance clues; response should not repeat/list/recommend them.")
print(f"\n--- SYSTEM ---\n{SYSTEM}")
print(f"\n--- USER_CONTEXT ---\n{context}")
r = await client.chat.completions.create(
model=MODEL, temperature=0.3, max_tokens=6000,
response_format={"type": "json_object"},
messages=[{"role": "system", "content": SYSTEM},
{"role": "user", "content": context}],
)
# content holds the post-thinking answer (OpenRouter puts CoT in a
# separate `reasoning` field). None => the model ran out of budget inside
# reasoning; retry briefly, then leave it missing for the next resume.
if getattr(r, "choices", None):
text = r.choices[0].message.content
if text:
if debug:
print(f"\n=== LLM_RESPONSE_RAW source_id={source_id} attempt={attempt} ===")
print("SHOULD: raw response is strict JSON with exactly decision_context, no markdown fence, no copied action list.")
print(text)
try:
payload = _json_payload(text)
if debug:
print(f"\n=== LLM_RESPONSE_PARSED source_id={source_id} attempt={attempt} ===")
print(json.dumps(payload, ensure_ascii=False, indent=2))
return payload
except (json.JSONDecodeError, ValueError) as e:
print(
f"json_reject source_id={source_id} attempt={attempt} reason={type(e).__name__}:{e}",
flush=True,
)
if attempt == api_attempts:
return None
if attempt < api_attempts:
await asyncio.sleep(min(30.0, 2.0 * attempt))
return None
def _read_cache() -> dict[str, dict[str, Any]]:
if not CACHE.exists():
return {}
rows = {}
n_bad = 0
for line in CACHE.read_text().splitlines():
if line.strip():
row = json.loads(line)
try:
needed = {
"schema_version", "world", "player_character", "short_summary",
"decision_context", "actions_text", "final_question",
}
if not needed <= set(row):
raise ValueError("missing structured scenario fields")
if row["schema_version"] != SCHEMA_VERSION:
raise ValueError(f"wrong schema_version: {row['schema_version']}")
_validate_scenario(row, row["text"], row["decision_context"])
if len(row["axes"]) < 2:
raise ValueError(f"too few axes: {row['axes']}")
except ValueError:
n_bad += 1
continue
rows[row["source_id"]] = row
if n_bad:
print(f"dropped_invalid_cached_rows={n_bad}", flush=True)
return rows
def _write_cache(rows: dict[str, dict[str, Any]]) -> None:
CACHE.parent.mkdir(parents=True, exist_ok=True)
ordered = sorted(rows.values(), key=lambda r: (r.get("game_id", ""), str(r.get("row_i", ""))))
CACHE.write_text("\n".join(json.dumps(row, ensure_ascii=False) for row in ordered) + "\n")
def _usable_rows(pool_size: int | None) -> list[dict[str, Any]]:
ds = load_dataset("wassname/machiavelli", split="train", streaming=True)
rows = []
for seen, row in enumerate(ds, start=1):
if len(_axes(row["choice_labels"])) < 2:
if seen % 25000 == 0:
print(f"scanned_source={seen}; usable={len(rows)}", flush=True)
continue
rows.append(row)
if len(rows) % 10000 == 0:
print(f"scanned_source={seen}; usable={len(rows)}", flush=True)
if pool_size is not None and len(rows) >= pool_size:
break
return rows
def _round_robin(rows: list[dict[str, Any]], seed: int, n: int | None) -> list[dict[str, Any]]:
by_game: dict[str, list[dict[str, Any]]] = {}
for row in rows:
by_game.setdefault(row["f"], []).append(row)
rng = random.Random(seed)
for game_rows in by_game.values():
rng.shuffle(game_rows)
games = sorted(by_game)
rng.shuffle(games)
picked, gi = [], 0
limit = len(rows) if n is None else min(n, len(rows))
while len(picked) < limit and any(by_game.values()):
game = games[gi % len(games)]
if by_game[game]:
picked.append(by_game[game].pop())
gi += 1
return picked
def _moral_contrast_top_percent_rows(
top_game_frac: float,
seed: int,
) -> tuple[list[dict[str, Any]], dict[str, dict[str, Any]]]:
if not (0.0 < top_game_frac <= 1.0):
raise ValueError(f"top_game_frac must be in (0, 1], got {top_game_frac}")
ds = load_dataset("wassname/machiavelli", split="train")
by_game: dict[str, list[tuple[float, str, dict[str, Any], dict[str, Any]]]] = {}
seen_pairs = set()
for row in ds:
if _is_setup_choice_row(row):
continue
if len(_axes(row["choice_labels"])) < 2:
continue
score = score_row(row)
if score is None:
continue
pair_key = (score["title"], score["choice_a_text"], score["choice_b_text"])
if pair_key in seen_pairs:
continue
seen_pairs.add(pair_key)
sid = _source_id(row)
by_game.setdefault(row["f"], []).append((float(score["score"]), sid, row, score))
rng = random.Random(seed)
games = sorted(by_game)
picked: list[dict[str, Any]] = []
meta: dict[str, dict[str, Any]] = {}
for game in games:
grouped: dict[float, list[tuple[float, str, dict[str, Any], dict[str, Any]]]] = {}
for item in by_game[game]:
grouped.setdefault(item[0], []).append(item)
sorted_game: list[tuple[float, str, dict[str, Any], dict[str, Any]]] = []
for score_value in sorted(grouped, reverse=True):
group = grouped[score_value]
rng.shuffle(group)
sorted_game.extend(group)
n_game = max(1, int(len(sorted_game) * top_game_frac + 0.999999))
for _, sid, row, score in sorted_game[:n_game]:
picked.append(row)
meta[sid] = score
return picked, meta
def _selection_summary(rows: list[dict[str, Any]], meta: dict[str, dict[str, Any]]) -> str:
by_game: dict[str, list[dict[str, Any]]] = {}
for row in rows:
by_game.setdefault(row["f"], []).append(row)
game_rows = []
for game, game_items in sorted(by_game.items(), key=lambda kv: (-len(kv[1]), kv[0])):
scores = [meta[_source_id(row)] for row in game_items]
game_rows.append({
"game": game[:32],
"n": len(game_items),
"score_min": min(float(s["score"]) for s in scores),
"score_med": sorted(float(s["score"]) for s in scores)[len(scores) // 2],
"subtle_med": sorted(float(s["subtle_score"]) for s in scores)[len(scores) // 2],
"crude_med": sorted(float(s["crude_score"]) for s in scores)[len(scores) // 2],
})
prompt_tokens = sorted(len(_context(row)) // 4 for row in rows)
p50 = prompt_tokens[len(prompt_tokens) // 2]
p90 = prompt_tokens[int(0.90 * (len(prompt_tokens) - 1))]
p99 = prompt_tokens[int(0.99 * (len(prompt_tokens) - 1))]
total_prompt_mtok = sum(prompt_tokens) / 1_000_000
lines = [
"\n=== MORAL_CONTRAST_TOP_PERCENT selection ===",
"SHOULD: each game contributes its own top slice; subtle_med should be the main signal, with crude_med not dominating every row.",
tabulate(game_rows, headers="keys", tablefmt="github", floatfmt=".3f"),
"\n=== PROMPT_TOKEN_ESTIMATE ===",
"SHOULD: this is chars/4, a rough upper bound for spend planning before OpenRouter calls.",
tabulate(
[{
"selected": len(rows),
"games": len(by_game),
"prompt_mtok": total_prompt_mtok,
"tok_p50": p50,
"tok_p90": p90,
"tok_p99": p99,
"tok_max": max(prompt_tokens),
}],
headers="keys",
tablefmt="github",
floatfmt=".3f",
),
]
preview = []
for row in rows[:10]:
score = meta[_source_id(row)]
preview.append({
"game": row["f"][:28],
"score": score["score"],
"subtle": score["subtle_score"],
"crude": score["crude_score"],
"row_i": str(row["row_i"])[:42],
})
lines.extend([
"\n=== TOP_SELECTED_PREVIEW ===",
tabulate(preview, headers="keys", tablefmt="github"),
])
return "\n".join(lines)
async def _summarise_batch(
client: AsyncOpenAI,
rows: list[dict[str, Any]],
api_attempts: int,
concurrency: int,
debug_ids: set[str],
) -> list[tuple[dict[str, Any], dict[str, str] | None | Exception]]:
sem = asyncio.Semaphore(concurrency)
async def one(row: dict[str, Any]) -> tuple[dict[str, Any], dict[str, str] | None | Exception]:
async with sem:
try:
sid = _source_id(row)
return row, await _summarise(
client,
_context(row),
api_attempts,
sid,
sid in debug_ids,
)
except Exception as e:
return row, e
return await asyncio.gather(*(one(row) for row in rows))
def _jsonable(value: Any) -> Any:
if isinstance(value, (dict, list)):
return json.dumps(value, ensure_ascii=False, sort_keys=True)
return value
def _write_parquet(path: Path, rows: list[dict[str, Any]]) -> None:
keys = list(rows[0])
for row in rows[1:]:
for key in row:
if key not in keys:
keys.append(key)
table = pa.Table.from_pylist([{k: _jsonable(row.get(k)) for k in keys} for row in rows])
path.parent.mkdir(parents=True, exist_ok=True)
pq.write_table(table, path)
def _hf_readme(n_rows: int) -> str:
return f"""---
license: mit
language:
- en
task_categories:
- text-generation
- text-classification
pretty_name: Machiavelli Character Scenarios
tags:
- persona
- steering-vectors
- moral-dilemmas
- ai-safety
- synthetic
size_categories:
- n<1K
configs:
- config_name: default
data_files:
- split: train
path: parquet/train.parquet
---
# Machiavelli Character Scenarios
{n_rows} roleplay decision prompts built from `wassname/machiavelli`.
Rows are selected for high choice-level spread on social/moral labels such as
fairness, deception, manipulation, promises, and spying, while penalising rows
where the contrast is mostly power, killing, or physical harm.
`combo` is the ready-to-use prompt. The same content is split into editable
fields: `world`, `player_character`, `short_summary`, generated
`decision_context`, `choice_text`, and `final_question`. Only the long
history/current-situation text is summarised; the short source fields and choices
stay separate so users can change the question or prompt format.
Labels are metadata copied from the source dataset, not ground-truth answers.
The prompts are for eliciting persona-conditioned roleplay, preference, judgment,
and tradeoff reasoning.
Source code: https://github.com/wassname/persona-steering-template-library
"""
def _build_hf_folder(rows: list[dict[str, Any]], out_dir: Path) -> None:
if out_dir.exists():
shutil.rmtree(out_dir)
_write_parquet(out_dir / "parquet" / "train.parquet", rows)
(out_dir / "README.md").write_text(_hf_readme(len(rows)))
def _upload_hf(rows: list[dict[str, Any]], repo_id: str, out_dir: Path) -> None:
_build_hf_folder(rows, out_dir)
create_repo(repo_id, repo_type="dataset", exist_ok=True)
info = HfApi().upload_folder(
repo_id=repo_id,
repo_type="dataset",
folder_path=out_dir,
commit_message=f"Upload {len(rows)} Machiavelli character scenarios",
)
print(f"uploaded {len(rows)} rows -> https://huggingface.co/datasets/{repo_id}")
print(info)
async def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("--n", type=int, default=10)
ap.add_argument("--all", action="store_true", help="summarise every usable source row")
ap.add_argument(
"--selection",
choices=["round-robin", "moral-contrast-top-percent"],
default="round-robin",
)
ap.add_argument("--top-game-frac", type=float, default=0.05)
ap.add_argument("--dry-select", action="store_true")
ap.add_argument("--seed", type=int, default=0)
ap.add_argument("--pool-size", type=int, default=300,
help="usable rows to collect before shuffling; ignored by --all")
ap.add_argument("--batch-size", type=int, default=100)
ap.add_argument("--concurrency", type=int, default=20)
ap.add_argument("--api-attempts", type=int, default=3)
ap.add_argument("--max-passes", type=int, default=5,
help="retry missing rows this many full passes before failing")
ap.add_argument("--debug-samples", type=int, default=3,
help="print full prompt/raw response/parsed JSON for this many picked rows")
ap.add_argument("--upload-hf", action="store_true")
ap.add_argument("--hf-repo", default=HF_REPO)
ap.add_argument("--hf-out", type=Path, default=Path("/tmp/machiavelli-character-scenarios-hf"))
args = ap.parse_args()
print("\n=== CONFIG ===")
print(f"argv: {' '.join(sys.argv)}")
print(tabulate(
[
("model", MODEL),
("cache", str(CACHE)),
("all", args.all),
("n", args.n),
("selection", args.selection),
("top_game_frac", args.top_game_frac),
("dry_select", args.dry_select),
("pool_size", pool_size := (None if args.all else args.pool_size)),
("batch_size", args.batch_size),
("concurrency", args.concurrency),
("api_attempts", args.api_attempts),
("max_passes", args.max_passes),
("debug_samples", args.debug_samples),
("upload_hf", args.upload_hf),
("hf_repo", args.hf_repo),
],
headers=["cfg", "value"],
tablefmt="plain",
))
print("SHOULD: debug traces show private choices in USER_CONTEXT; LLM responses should use them only to choose relevant context, not copy or recommend actions.")
cached = _read_cache()
selection_meta: dict[str, dict[str, Any]] = {}
if args.selection == "moral-contrast-top-percent":
if args.all:
raise ValueError("--selection moral-contrast-top-percent expects --top-game-frac, not --all")
picked, selection_meta = _moral_contrast_top_percent_rows(args.top_game_frac, args.seed)
usable = picked
else:
usable = _usable_rows(pool_size)
picked = _round_robin(usable, args.seed, None if args.all else args.n)
debug_ids = {_source_id(row) for row in picked[:args.debug_samples]}
games = {row["f"] for row in usable}
print(f"{len(usable)} usable rows across {len(games)} games; picked {len(picked)}")
if selection_meta:
print(_selection_summary(picked, selection_meta))
picked_ids = {_source_id(row) for row in picked}
cached = {sid: row for sid, row in cached.items() if sid in picked_ids}
if args.dry_select:
print("\n=== SUMMARY ===")
print(f"DRY_SELECT: selected={len(picked)} games={len(games)} cache={len(cached)}")
return
out = dict(cached)
todo = [row for row in picked if _source_id(row) not in out]
print(f"{len(picked)} picked, {len(picked) - len(todo)} cached, summarising {len(todo)}")
client = AsyncOpenAI(
base_url="https://openrouter.ai/api/v1",
api_key=os.environ["OPENROUTER_API_KEY"],
) if todo else None
started = time.time()
for pass_i in range(1, args.max_passes + 1):
todo = [row for row in picked if _source_id(row) not in out]
if not todo:
break
print(f"pass={pass_i}/{args.max_passes}; missing_at_start={len(todo)}", flush=True)
for start in range(0, len(todo), args.batch_size):
batch = todo[start:start + args.batch_size]
assert client is not None
results = await _summarise_batch(
client, batch, args.api_attempts, args.concurrency, debug_ids
)
n_api_skip = 0
n_validation_skip = 0
for row, text in results:
if isinstance(text, Exception) or not text:
n_api_skip += 1
continue
sid = _source_id(row)
try:
out[sid] = _scenario_row(row, text, selection_meta.get(sid))
except ValueError as e:
n_validation_skip += 1
print(f"\n=== VALIDATION_REJECT source_id={sid} ===")
print(f"reason={e}")
print(json.dumps(text, ensure_ascii=False, indent=2))
_write_cache(out)
done = min(start + len(batch), len(todo))
elapsed_min = (time.time() - started) / 60.0
print(
f"pass={pass_i}; batch {done}/{len(todo)}; cache={len(out)}; "
f"api_skipped_this_batch={n_api_skip}; "
f"validation_skipped_this_batch={n_validation_skip}; "
f"elapsed_min={elapsed_min:.1f}; "
f"remaining={len([row for row in picked if _source_id(row) not in out])}",
flush=True,
)
_write_cache(out)
missing = len([row for row in picked if _source_id(row) not in out])
print(f"wrote {len(out)} -> {CACHE}")
print(f"usable={len(usable)} picked={len(picked)} cached={len(out)} missing={missing}")
for row in picked[: min(5, len(picked))]:
sid = _source_id(row)
if sid in out:
print(f"\n### {sid} axes={out[sid]['axes']}")
print(f"world={out[sid]['world']}")
print(f"player_character={out[sid]['player_character']}")
print(f"short_summary={out[sid]['short_summary']}")
print(f"decision_context={out[sid]['decision_context']}")
print(f"actions_text=\n{out[sid]['actions_text']}")
print("\n=== SUMMARY ===")
status = "PASS" if missing == 0 else "INCOMPLETE"
print(f"{status}: usable={len(usable)} picked={len(picked)} cached={len(out)} missing={missing}")
print(tabulate(
[{"usable": len(usable), "picked": len(picked), "cached": len(out), "missing": missing}],
headers="keys",
tablefmt="github",
))
if args.upload_hf:
if args.all and missing:
raise RuntimeError(f"refusing to upload incomplete full cache: missing={missing}")
rows = [out[_source_id(row)] for row in picked if _source_id(row) in out]
_upload_hf(rows, args.hf_repo, args.hf_out)
if __name__ == "__main__":
asyncio.run(main())
scripts/validate_persona_axes_openrouter.py
+11 -1
View File
@@ -733,13 +733,21 @@ class OpenRouter:
bad_path = path.with_suffix(f".bad-{int(time.time())}.json")
path.rename(bad_path)
logger.warning(f"quarantined malformed cached JSON judge output: {bad_path}")
attempts = JSON_RETRIES if json_schema is not None else 1
attempts = JSON_RETRIES
last_content = ""
last_error: Exception | None = None
for attempt in range(1, attempts + 1):
async with self.sem:
resp = await self.client.chat.completions.create(
**payload, extra_body=extra_body)
# OpenRouter returns an error body with choices=None on a provider
# error / content filter / rate limit; treat as a retryable failure
# instead of crashing the whole screen on `resp.choices[0]`.
if not getattr(resp, "choices", None):
last_error = RuntimeError(f"empty response (no choices): {getattr(resp, 'error', resp)!r}")
if attempt < attempts:
await asyncio.sleep(min(30.0, 2.0 * attempt))
continue
message = resp.choices[0].message
content = message.content or ""
last_content = content
@@ -752,6 +760,8 @@ class OpenRouter:
f"malformed JSON judge output attempt {attempt}/{attempts} "
f"cache_tag={cache_tag}: {content[:160]!r}"
)
if attempt < attempts:
await asyncio.sleep(min(30.0, 2.0 * attempt))
continue
path.write_text(json.dumps({
"created_at": time.time(),
uv.lock
Generated
+961 -4
View File
File diff suppressed because it is too large Load Diff