wordle-five: the engine, the reward, the solver and the probe that checks them
The Python is the source of truth; src/demos/wordle/engine.ts will be a port of it, and CI gates the two against a SHA-256 over all 21.2M (guess, answer) pattern pairs rather than a hand-picked vector file — a vector file only ever catches the cases somebody thought of. The reward is three weighted components, and the third one is the reason this demo is worth building. `solved` and `economy` pull toward winning. `consistency` pulls against them, because a player maximising information deliberately guesses words that cannot win — a word that splits the remaining candidates evenly teaches more than a word that might happen to be right. That is good play, and it costs consistency. The probe ladder proves the tension is real rather than asserted: inaction 0.0000 crude 0.0111 plausible 0.1224 candidate_only 0.8925 exhaustive 0.9031 oracle 0.9458 The two good policies are 0.05 apart and neither dominates — the entropy oracle takes 1.00 economy and 0.73 consistency, the candidate-only player takes 0.75 and 1.00. Which one wins is a decision about what you want, which is the whole argument the site exists to make. probe.py fails CI if either starts dominating. Two traps found by building it. `consistency` is scored over turns SPENT, not guesses accepted: counting only legal guesses hands a free 1.0 to a policy that plays one word and then jams the parser five times — one guess, no contradictions, perfect score. And `economy`'s denominator is the depth the SHIPPED solver reaches, not a depth-optimal search: entropy-greedy is not depth-optimal, so grading it against an exact optimum would make the oracle rung fail its own assertion on some seeds. The word lists are built from Wordnik (MIT) intersected with SCOWL, never from the original game's 2,315 answers. 4,603 answers makes this materially harder than the original, so the published SALET/3.4212 results are cited as belonging to that list and our own reference player's TARES/3.72 is measured here. verifiers is an optional extra. The engine, reward, solver and probe all run — and gate — without an RL stack resolvable. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019mt6sHQHEnEYrJZvoMCJSB
This commit is contained in:
karti-ai
+247
@@ -0,0 +1,247 @@
|
||||
#!/usr/bin/env python3
|
||||
"""The probe ladder: does this reward actually measure anything?
|
||||
|
||||
A reward function is a claim, and an unprobed one is an unchecked claim. The
|
||||
ladder plays a set of deliberately-crafted policies — from doing nothing at all
|
||||
to the best play we can write — and asserts that the reward orders them the way
|
||||
a person would, per weighted component rather than as one blended float.
|
||||
|
||||
A blended float hides the failure this is built to catch: a component that is
|
||||
flat across every policy is measuring nothing, and it will still move the total
|
||||
because the other components move.
|
||||
|
||||
Run: uv run python envs/probe.py
|
||||
Exits non-zero on any violated assertion.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent / "wordle_five"))
|
||||
|
||||
from wordle_five import solver as S # noqa: E402
|
||||
from wordle_five.engine import ( # noqa: E402
|
||||
MAX_GUESSES,
|
||||
Game,
|
||||
answer_for_seed,
|
||||
answers,
|
||||
score_guess,
|
||||
)
|
||||
from wordle_five.protocol import parse_guess # noqa: E402
|
||||
from wordle_five.reward import ROLES, WEIGHTS, Episode, metrics, score, total # noqa: E402
|
||||
|
||||
SEEDS = list(range(16))
|
||||
COMPONENTS = ("solved", "economy", "consistency")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- policies --
|
||||
# Each policy is a function (game) -> reply text. They return TEXT, not moves,
|
||||
# so the parser and the rejection path are exercised exactly as a model would
|
||||
# exercise them. A policy that returns unparseable prose is a valid policy.
|
||||
|
||||
|
||||
def policy_inaction(game: Game) -> str:
|
||||
"""Reasons at length and never commits. The floor: must measure 0.000."""
|
||||
return "Let me think carefully about which word would be most informative here."
|
||||
|
||||
|
||||
def policy_spammer(game: Game) -> str:
|
||||
"""Multiple bracketed tokens. Only the first is read; the rest are noise."""
|
||||
return "Maybe [crane] or [slate] or [adieu]?"
|
||||
|
||||
|
||||
def policy_crude(game: Game) -> str:
|
||||
"""One valid word, forever. Legal, useless, and it repeats itself."""
|
||||
return "[tares]"
|
||||
|
||||
|
||||
def policy_plausible(game: Game) -> str:
|
||||
"""Keeps greens, drops known-absent letters. What a person does unaided."""
|
||||
pool = answers()
|
||||
if not game.history:
|
||||
return "[tares]"
|
||||
greens: dict[int, str] = {}
|
||||
absent: set[str] = set()
|
||||
for guess, pattern in game.history:
|
||||
for i, tile in enumerate(pattern):
|
||||
if tile == "G":
|
||||
greens[i] = guess[i]
|
||||
elif tile == "X":
|
||||
absent.add(guess[i])
|
||||
played = {g for g, _ in game.history}
|
||||
for word in pool:
|
||||
if word in played:
|
||||
continue
|
||||
if any(word[i] != c for i, c in greens.items()):
|
||||
continue
|
||||
if any(c in absent for c in word):
|
||||
continue
|
||||
return f"[{word}]"
|
||||
return "[tares]"
|
||||
|
||||
|
||||
def policy_candidate_only(game: Game) -> str:
|
||||
"""Only ever guesses words that could still be the answer.
|
||||
|
||||
This is the policy the counterweight rewards. It never wastes a guess, and
|
||||
it is measurably slower than the oracle because it cannot buy information
|
||||
with a word that has no chance of winning.
|
||||
"""
|
||||
alive = S.consistent_candidates(game.history)
|
||||
played = {g for g, _ in game.history}
|
||||
for word in alive:
|
||||
if word not in played:
|
||||
return f"[{word}]"
|
||||
return "[tares]"
|
||||
|
||||
|
||||
def policy_oracle(game: Game) -> str:
|
||||
"""Greedy maximum entropy — the reference player, and the economy denominator."""
|
||||
import numpy as np
|
||||
|
||||
pool = answers()
|
||||
if not game.history:
|
||||
return f"[{pool[S._best_opener()]}]"
|
||||
alive = S.consistent_candidates(game.history)
|
||||
if not alive:
|
||||
return "[tares]"
|
||||
index = np.array([pool.index(w) for w in alive])
|
||||
return f"[{pool[S.best_guess(index)]}]"
|
||||
|
||||
|
||||
def policy_exhaustive(game: Game) -> str:
|
||||
"""Knows the answer but burns three turns on known-wrong probes first.
|
||||
|
||||
Proves the turn budget actually binds: this must score strictly below the
|
||||
oracle, or `economy` is not doing its job.
|
||||
"""
|
||||
if len(game.history) < 3:
|
||||
probes = ["fuzzy", "jumbo", "whelp"]
|
||||
pick = probes[len(game.history)]
|
||||
return f"[{pick if pick != game.answer else 'tares'}]"
|
||||
return f"[{game.answer}]"
|
||||
|
||||
|
||||
POLICIES = {
|
||||
"inaction": policy_inaction,
|
||||
"spammer": policy_spammer,
|
||||
"crude": policy_crude,
|
||||
"plausible": policy_plausible,
|
||||
"candidate_only": policy_candidate_only,
|
||||
"exhaustive": policy_exhaustive,
|
||||
"oracle": policy_oracle,
|
||||
}
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ runner --
|
||||
|
||||
|
||||
def play(policy, seed: int) -> Episode:
|
||||
game = Game(seed=seed)
|
||||
for _ in range(MAX_GUESSES * 3): # room for rejected replies
|
||||
if game.over:
|
||||
break
|
||||
guess = parse_guess(policy(game))
|
||||
if guess is None:
|
||||
game.rejected += 1
|
||||
# An unparseable reply still costs a turn of patience, or a policy
|
||||
# that never guesses would loop until the budget above runs out and
|
||||
# look identical to one that guessed badly.
|
||||
if game.rejected >= MAX_GUESSES:
|
||||
break
|
||||
continue
|
||||
game.play(guess)
|
||||
return Episode(
|
||||
answer=game.answer,
|
||||
guesses=[g for g, _ in game.history],
|
||||
patterns=[p for _, p in game.history],
|
||||
rejected=game.rejected,
|
||||
reference_depth=S.reference_depth(game.answer),
|
||||
)
|
||||
|
||||
|
||||
def run() -> int:
|
||||
print(f"Probing wordle-five over {len(SEEDS)} seeds\n")
|
||||
results: dict[str, dict[str, float]] = {}
|
||||
extra: dict[str, dict[str, float]] = {}
|
||||
|
||||
for name, policy in POLICIES.items():
|
||||
episodes = [play(policy, s) for s in SEEDS]
|
||||
scores = [score(e) for e in episodes]
|
||||
results[name] = {c: sum(s[c] for s in scores) / len(scores) for c in COMPONENTS}
|
||||
results[name]["total"] = sum(total(e) for e in episodes) / len(episodes)
|
||||
mets = [metrics(e) for e in episodes]
|
||||
extra[name] = {k: sum(m[k] for m in mets) / len(mets) for k in mets[0]}
|
||||
|
||||
width = max(len(n) for n in POLICIES)
|
||||
header = f"{'policy':<{width}} " + " ".join(f"{c:>12}" for c in COMPONENTS) + f" {'TOTAL':>8}"
|
||||
print(header)
|
||||
print("-" * len(header))
|
||||
for name in POLICIES:
|
||||
row = results[name]
|
||||
cells = " ".join(f"{row[c]:>12.4f}" for c in COMPONENTS)
|
||||
print(f"{name:<{width}} {cells} {row['total']:>8.4f}")
|
||||
|
||||
print(f"\nweights: {WEIGHTS}")
|
||||
print(f"roles: {ROLES}")
|
||||
print("\nguesses used (mean):", {n: round(extra[n]['guesses_used'], 2) for n in POLICIES})
|
||||
|
||||
failures: list[str] = []
|
||||
|
||||
# 1. Doing nothing must measure exactly nothing. If a policy that never
|
||||
# plays can collect reward, every number above it is inflated.
|
||||
if abs(results["inaction"]["total"]) > 1e-9:
|
||||
failures.append(f"inaction scored {results['inaction']['total']:.6f}, must be 0.000")
|
||||
|
||||
# 2. The reference player must reach the top of both objective components,
|
||||
# or `economy`'s denominator disagrees with the policy that defines it.
|
||||
if abs(results["oracle"]["solved"] - 1.0) > 1e-9:
|
||||
failures.append(f"oracle solved {results['oracle']['solved']:.4f}, must be 1.0")
|
||||
if abs(results["oracle"]["economy"] - 1.0) > 1e-9:
|
||||
failures.append(f"oracle economy {results['oracle']['economy']:.4f}, must be 1.0")
|
||||
|
||||
# 3. The counterweight must actually bite the policy it is aimed at. The
|
||||
# oracle buys information with guesses that cannot win; if that were
|
||||
# free, `consistency` would be a gate wearing a counterweight's name.
|
||||
if results["oracle"]["consistency"] >= 1.0 - 1e-9:
|
||||
failures.append("oracle scored a perfect consistency — the counterweight is not in tension")
|
||||
|
||||
# 4. And the policy that pays the counterweight's price must collect it.
|
||||
if abs(results["candidate_only"]["consistency"] - 1.0) > 1e-9:
|
||||
failures.append(
|
||||
f"candidate_only consistency {results['candidate_only']['consistency']:.4f}, must be 1.0"
|
||||
)
|
||||
|
||||
# 5. Neither of the two good policies may dominate the other on every
|
||||
# component. If one did, the reward would encode a single right answer
|
||||
# and the reward editor would be theatre.
|
||||
oracle, cand = results["oracle"], results["candidate_only"]
|
||||
if all(oracle[c] >= cand[c] - 1e-9 for c in COMPONENTS):
|
||||
failures.append("oracle dominates candidate_only on every component — no real trade-off")
|
||||
if all(cand[c] >= oracle[c] - 1e-9 for c in COMPONENTS):
|
||||
failures.append("candidate_only dominates oracle on every component — no real trade-off")
|
||||
|
||||
# 6. Burning turns on known-wrong probes must cost.
|
||||
if results["exhaustive"]["total"] >= results["oracle"]["total"] - 1e-9:
|
||||
failures.append("exhaustive >= oracle — the turn budget does not bind")
|
||||
|
||||
# 7. No weighted component may be flat. A component with the same value for
|
||||
# every policy is measuring nothing and still moves the total.
|
||||
for component in COMPONENTS:
|
||||
values = [results[n][component] for n in POLICIES]
|
||||
if max(values) - min(values) < 1e-9:
|
||||
failures.append(f"component '{component}' is flat across the ladder — it measures nothing")
|
||||
|
||||
print()
|
||||
if failures:
|
||||
for f in failures:
|
||||
print(f"FAIL: {f}")
|
||||
return 1
|
||||
print(f"All {7} ladder assertions hold.")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(run())
|
||||
@@ -0,0 +1 @@
|
||||
69f4e8dfbdc492176d7b7f04b080d8d3cccb5f18aca2d592e622938a5aeec8e2 sha256 over 4603^2 (guess,answer) pattern pairs, answers.json order
|
||||
@@ -0,0 +1,22 @@
|
||||
[project]
|
||||
name = "wordle-five"
|
||||
version = "0.1.0"
|
||||
description = "A five-letter word-guessing environment with a three-component reward"
|
||||
requires-python = ">=3.11,<3.14"
|
||||
dependencies = ["numpy>=2.0"]
|
||||
|
||||
[project.optional-dependencies]
|
||||
# verifiers is optional so the engine, the solver, the reward and the probe all
|
||||
# run — and CI gates on them — without pulling a heavyweight RL stack. Only
|
||||
# taskset.py needs it.
|
||||
verifiers = ["verifiers>=0.3.2.dev12"]
|
||||
|
||||
[build-system]
|
||||
requires = ["hatchling"]
|
||||
build-backend = "hatchling.build"
|
||||
|
||||
[tool.hatch.build.targets.wheel]
|
||||
packages = ["wordle_five"]
|
||||
|
||||
[tool.hatch.build]
|
||||
include = ["wordle_five/**", "words/**", "pyproject.toml", "README.md"]
|
||||
@@ -0,0 +1,142 @@
|
||||
"""The duplicate-letter cases first, because they are the ones that get shipped wrong."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from wordle_five.engine import (
|
||||
Game,
|
||||
answer_for_seed,
|
||||
answers,
|
||||
conformance_digest,
|
||||
guesses,
|
||||
hard_mode_violation,
|
||||
is_consistent,
|
||||
score_guess,
|
||||
)
|
||||
|
||||
# Every one of these was verified against an independent implementation and
|
||||
# against TextArena's own scorer before being written down. They are the
|
||||
# regression surface for the two-pass rule.
|
||||
VECTORS = [
|
||||
("alloy", "llama", "YGYXX"), # greens claim first, yellows take the rest
|
||||
("speed", "erase", "YXYYX"), # both E's yellow: exactly two are available
|
||||
("array", "radar", "YYYGX"), # double-double with one green
|
||||
("sassy", "basis", "YGGXX"), # 3 S guessed, 2 in answer: one Y, one X
|
||||
("eerie", "rebel", "YGYXX"), # the same shape on E
|
||||
("level", "eagle", "YYXYX"), # two repeated letters, no greens at all
|
||||
("geese", "these", "XXGGG"), # greens exhaust the pool, leading E greys
|
||||
("abbey", "abbot", "GGGXX"),
|
||||
("crane", "plane", "XXGGG"),
|
||||
("alloy", "balmy", "YXGXG"),
|
||||
("tares", "tares", "GGGGG"),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("guess,answer,expected", VECTORS)
|
||||
def test_scoring_vectors(guess: str, answer: str, expected: str) -> None:
|
||||
assert score_guess(guess, answer) == expected
|
||||
|
||||
|
||||
def test_score_is_case_insensitive() -> None:
|
||||
assert score_guess("ALLOY", "LLAMA") == score_guess("alloy", "llama")
|
||||
|
||||
|
||||
def test_a_letter_is_never_marked_more_often_than_it_occurs() -> None:
|
||||
"""The invariant the two passes exist to preserve, checked over the list."""
|
||||
pool = answers()[:200]
|
||||
for answer in pool:
|
||||
for guess in pool[:50]:
|
||||
pattern = score_guess(guess, answer)
|
||||
for letter in set(guess):
|
||||
marked = sum(
|
||||
1 for i, c in enumerate(guess) if c == letter and pattern[i] != "X"
|
||||
)
|
||||
assert marked <= answer.count(letter)
|
||||
|
||||
|
||||
def test_consistency_is_symmetric_with_scoring() -> None:
|
||||
"""`is_consistent` must agree with `score_guess`, since the reward uses it."""
|
||||
for answer in answers()[:100]:
|
||||
guess = "tares"
|
||||
pattern = score_guess(guess, answer)
|
||||
assert is_consistent(answer, guess, pattern)
|
||||
|
||||
|
||||
def test_every_answer_is_guessable() -> None:
|
||||
assert set(answers()) <= guesses()
|
||||
|
||||
|
||||
def test_seeded_answer_is_deterministic() -> None:
|
||||
assert answer_for_seed(42) == answer_for_seed(42)
|
||||
assert answer_for_seed(1) != answer_for_seed(2)
|
||||
|
||||
|
||||
def test_rejections_do_not_consume_a_row() -> None:
|
||||
game = Game(seed=3, answer="tares")
|
||||
assert game.play("zzzzz")[1] is not None # not a word
|
||||
assert game.play("four")[1] is not None # wrong length
|
||||
assert len(game.history) == 0
|
||||
assert game.rejected == 2
|
||||
assert game.play("crane")[0] is not None
|
||||
assert len(game.history) == 1
|
||||
|
||||
|
||||
def test_a_repeat_is_rejected() -> None:
|
||||
game = Game(seed=3, answer="tares")
|
||||
game.play("crane")
|
||||
assert "already been guessed" in (game.play("crane")[1] or "")
|
||||
|
||||
|
||||
def test_game_never_raises_on_junk() -> None:
|
||||
game = Game(seed=3, answer="tares")
|
||||
for junk in ["", " ", "!!!!!", "a", "abcdefghij", "12345"]:
|
||||
pattern, reason = game.play(junk)
|
||||
assert pattern is None and reason is not None
|
||||
|
||||
|
||||
def test_solving_ends_the_game() -> None:
|
||||
game = Game(seed=3, answer="tares")
|
||||
game.play("tares")
|
||||
assert game.solved and game.over
|
||||
|
||||
|
||||
# ------------------------------------------------------------- hard mode --
|
||||
|
||||
|
||||
def test_hard_mode_locks_greens() -> None:
|
||||
assert hard_mode_violation("crown", "crane", "GGXXX") is None
|
||||
assert hard_mode_violation("blown", "crane", "GGXXX") is not None
|
||||
|
||||
|
||||
def test_hard_mode_counts_yellows_rather_than_testing_membership() -> None:
|
||||
"""Two revealed E's require two E's, not merely one."""
|
||||
pattern = score_guess("speed", "erase") # YXYYX — two E's revealed
|
||||
assert hard_mode_violation("ester", "speed", pattern) is None # two E's
|
||||
assert hard_mode_violation("crest", "speed", pattern) is not None # only one E
|
||||
|
||||
|
||||
def test_hard_mode_does_not_ban_grey_letters() -> None:
|
||||
"""The rule most implementations add and the real game does not have."""
|
||||
pattern = score_guess("crane", "tares") # C grey
|
||||
assert pattern[0] == "X", "C should be grey against TARES"
|
||||
# A guess reusing that grey C is legal, so long as it still satisfies every
|
||||
# green and yellow that was revealed.
|
||||
assert hard_mode_violation("stare", "crane", pattern) is None
|
||||
|
||||
|
||||
# ------------------------------------------------------ cross-language gate --
|
||||
|
||||
|
||||
def test_conformance_digest_matches_the_committed_value() -> None:
|
||||
"""The gate that keeps the TypeScript port honest.
|
||||
|
||||
This hashes every (guess, answer) pair in the answer list. If it changes,
|
||||
either the scorer changed or the word list did, and `src/demos/wordle/
|
||||
engine.ts` must be re-verified against it — `pnpm conformance` does the
|
||||
other half.
|
||||
"""
|
||||
from pathlib import Path
|
||||
|
||||
committed = (Path(__file__).parent.parent / "CONFORMANCE.txt").read_text().split()[0]
|
||||
assert conformance_digest() == committed
|
||||
@@ -0,0 +1,65 @@
|
||||
"""Turning model text into a move, and a move into text.
|
||||
|
||||
The one rule: **never raise**. A malformed reply is a thing the model did, not
|
||||
an error in the harness, and it must be scored rather than crash the rollout.
|
||||
A parse failure becomes a rejected guess, which costs the model a turn of its
|
||||
budget and shows up in `rejected_replies`.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
|
||||
from .engine import MAX_GUESSES, WORD_LENGTH
|
||||
|
||||
SYSTEM_PROMPT = """You are playing a word-guessing game.
|
||||
|
||||
A secret {length}-letter word has been chosen. You have {turns} attempts.
|
||||
After each guess you are told, for every letter:
|
||||
G — correct letter, correct position
|
||||
Y — the letter is in the word, but somewhere else
|
||||
X — the letter is not in the word
|
||||
|
||||
Think it through, then give your guess as the only square-bracketed token in
|
||||
your reply, like [crane]. The game reads the first bracketed token and ignores
|
||||
everything else, so do not put any other words in brackets."""
|
||||
|
||||
_BRACKETED = re.compile(r"\[([A-Za-z]+)\]")
|
||||
|
||||
|
||||
def system_prompt() -> str:
|
||||
return SYSTEM_PROMPT.format(length=WORD_LENGTH, turns=MAX_GUESSES)
|
||||
|
||||
|
||||
def parse_guess(reply: str | None) -> str | None:
|
||||
"""The first bracketed alphabetic token, lowercased. None if there is none.
|
||||
|
||||
Deliberately does NOT check length or membership — that is the game's job,
|
||||
so that 'the model guessed a six-letter word' and 'the model produced no
|
||||
guess at all' stay distinguishable in the metrics.
|
||||
"""
|
||||
if not reply:
|
||||
return None
|
||||
match = _BRACKETED.search(reply)
|
||||
return match.group(1).lower() if match else None
|
||||
|
||||
|
||||
def render_feedback(guess: str, pattern: str, turns_left: int) -> str:
|
||||
"""The feedback block, in the format the system prompt promised.
|
||||
|
||||
Letters and tiles are space-separated and positionally aligned. Most
|
||||
tokenizers give a space-separated character its own token, which is the
|
||||
difference between the model reading position 3 and guessing at it.
|
||||
"""
|
||||
letters = " ".join(guess.upper())
|
||||
tiles = " ".join(pattern)
|
||||
plural = "" if turns_left == 1 else "es"
|
||||
return f"{letters}\n{tiles}\n\nYou have {turns_left} guess{plural} left."
|
||||
|
||||
|
||||
def render_rejection(reason: str, turns_left: int) -> str:
|
||||
plural = "" if turns_left == 1 else "es"
|
||||
return (
|
||||
f"That guess was not accepted: {reason}.\n"
|
||||
f"It did not use up a turn. You have {turns_left} guess{plural} left."
|
||||
)
|
||||
@@ -0,0 +1,132 @@
|
||||
"""The reward. This file is quoted verbatim on the demo page, so it is written
|
||||
to be read by someone who does not write Python.
|
||||
|
||||
Three components, weights summing to 1.0. Two of them pull toward winning. The
|
||||
third pulls against them on purpose.
|
||||
|
||||
# region: pig-demo/reward
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
|
||||
from .engine import GREEN, WORD_LENGTH, score_guess
|
||||
|
||||
SOLVED_WEIGHT = 0.50
|
||||
ECONOMY_WEIGHT = 0.30
|
||||
CONSISTENCY_WEIGHT = 0.20
|
||||
|
||||
|
||||
@dataclass
|
||||
class Episode:
|
||||
"""Everything the reward needs from one played game."""
|
||||
|
||||
answer: str
|
||||
guesses: list[str]
|
||||
patterns: list[str]
|
||||
rejected: int
|
||||
reference_depth: int
|
||||
max_guesses: int = 6
|
||||
|
||||
|
||||
def solved(ep: Episode) -> float:
|
||||
"""Did it win? The objective, and half the reward."""
|
||||
return 1.0 if ep.patterns and ep.patterns[-1] == GREEN * WORD_LENGTH else 0.0
|
||||
|
||||
|
||||
def economy(ep: Episode) -> float:
|
||||
"""How few turns it took, against how few the reference player needed.
|
||||
|
||||
Always a ratio against a reference on the SAME hidden word, never an
|
||||
absolute count — otherwise the reward would mostly measure whether the word
|
||||
happened to be easy, and a model would be punished for a hard draw.
|
||||
"""
|
||||
if not solved(ep):
|
||||
return 0.0
|
||||
return min(1.0, ep.reference_depth / max(1, len(ep.guesses)))
|
||||
|
||||
|
||||
def consistency(ep: Episode) -> float:
|
||||
"""The share of guesses that were still possible answers when they were made.
|
||||
|
||||
This is the counterweight, and it is genuinely in tension with the other
|
||||
two. A player maximising information will deliberately guess words that
|
||||
CANNOT win, because a word that splits the remaining candidates evenly
|
||||
teaches it more than a word that might happen to be right. That is good
|
||||
play, and it costs consistency. The reference solver scores below 1.0 here.
|
||||
|
||||
So the three components describe a real trade: win, win fast, and do not
|
||||
spend turns on moves that could not have won. You cannot max all three, and
|
||||
which one you weight is a decision about what you actually want — which is
|
||||
the entire argument this demo exists to make.
|
||||
|
||||
Scored over every turn SPENT, not every guess accepted — see below. Without
|
||||
that, a policy which never produces a parseable word scores a vacuous 1.0
|
||||
here (no guesses, therefore no contradictions) and collects 0.2 for doing
|
||||
nothing at all.
|
||||
"""
|
||||
spent = len(ep.guesses) + ep.rejected
|
||||
if spent == 0:
|
||||
return 0.0
|
||||
|
||||
viable = 0
|
||||
for i, guess in enumerate(ep.guesses):
|
||||
# A guess is viable iff it is consistent with every earlier piece of
|
||||
# feedback — that is, it was still a possible answer at the moment it
|
||||
# was played.
|
||||
if all(score_guess(ep.guesses[k], guess) == ep.patterns[k] for k in range(i)):
|
||||
viable += 1
|
||||
|
||||
# The denominator is every turn SPENT, including replies the game refused.
|
||||
# Counting only legal guesses would hand a free 1.0 to a policy that plays
|
||||
# one word and then jams the parser five times: one guess, no
|
||||
# contradictions, perfect score. Measuring against turns spent asks the
|
||||
# question that actually matters — of the attempts you made, how many were
|
||||
# a move that could have won?
|
||||
return viable / spent
|
||||
|
||||
|
||||
def score(ep: Episode) -> dict[str, float]:
|
||||
"""The weighted components, unweighted. The caller applies the weights."""
|
||||
return {
|
||||
"solved": solved(ep),
|
||||
"economy": economy(ep),
|
||||
"consistency": consistency(ep),
|
||||
}
|
||||
|
||||
|
||||
def total(ep: Episode) -> float:
|
||||
parts = score(ep)
|
||||
return (
|
||||
parts["solved"] * SOLVED_WEIGHT
|
||||
+ parts["economy"] * ECONOMY_WEIGHT
|
||||
+ parts["consistency"] * CONSISTENCY_WEIGHT
|
||||
)
|
||||
|
||||
|
||||
# endregion: pig-demo/reward
|
||||
|
||||
|
||||
def metrics(ep: Episode) -> dict[str, float]:
|
||||
"""Diagnostics. Weight zero — reported, never summed into the reward."""
|
||||
return {
|
||||
"guesses_used": float(len(ep.guesses)),
|
||||
"rejected_replies": float(ep.rejected),
|
||||
"reference_depth": float(ep.reference_depth),
|
||||
"turns_granted": float(ep.max_guesses),
|
||||
"inconsistent_guesses": float(round((1.0 - consistency(ep)) * len(ep.guesses))),
|
||||
}
|
||||
|
||||
|
||||
WEIGHTS = {
|
||||
"solved": SOLVED_WEIGHT,
|
||||
"economy": ECONOMY_WEIGHT,
|
||||
"consistency": CONSISTENCY_WEIGHT,
|
||||
}
|
||||
|
||||
ROLES = {
|
||||
"solved": "objective",
|
||||
"economy": "objective",
|
||||
"consistency": "counterweight",
|
||||
}
|
||||
@@ -0,0 +1,228 @@
|
||||
"""The reference player: greedy maximum-entropy with full constraint filtering.
|
||||
|
||||
Two jobs, and it is important they are the same code:
|
||||
|
||||
1. It is the `oracle` rung of the probe ladder.
|
||||
2. Its solve depth on a seed is the denominator of the `economy` reward.
|
||||
|
||||
Deriving the reward's reference from the shipped policy rather than from a
|
||||
separately-computed optimum is deliberate. If `economy` were measured against a
|
||||
depth-optimal search the oracle could not reach 1.0 — entropy-greedy is not
|
||||
depth-optimal — and the probe's own assertion would fail on some seeds. The
|
||||
optimum is still published, as a diagnostic, in `optimal_depth_report()`.
|
||||
|
||||
Note what this solver does NOT get: it scores 1.0 on solving and on economy,
|
||||
and strictly below 1.0 on `consistency`, because maximising information means
|
||||
deliberately guessing words that cannot win. That tension is the point — see
|
||||
reward.py.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from functools import lru_cache
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
|
||||
from .engine import GREEN, WORD_LENGTH, answers, score_guess
|
||||
|
||||
_CACHE = Path(__file__).parent.parent / ".pattern-cache" / "answers.npy"
|
||||
|
||||
# Base-3 code for a 5-tile pattern: 0..242, so it fits in a uint8. The all-green
|
||||
# pattern is the largest code, which is the only value the solver compares
|
||||
# against by name.
|
||||
_TILE_VALUE = {"X": 0, "Y": 1, "G": 2}
|
||||
ALL_GREEN_CODE = sum(2 * 3**i for i in range(WORD_LENGTH))
|
||||
|
||||
|
||||
def pattern_code(pattern: str) -> int:
|
||||
return sum(_TILE_VALUE[t] * 3**i for i, t in enumerate(pattern))
|
||||
|
||||
|
||||
def _compute_matrix(words: tuple[str, ...]) -> np.ndarray:
|
||||
"""M[g, a] = the pattern code for guessing words[g] against answer words[a].
|
||||
|
||||
Vectorised one letter at a time and chunked over guesses. The naive
|
||||
(n_guess, n_answer, 5) boolean intermediate is ~106 MB per letter at this
|
||||
list size, so chunking is what keeps peak memory in the tens of megabytes
|
||||
instead of gigabytes.
|
||||
"""
|
||||
n = len(words)
|
||||
grid = np.frombuffer("".join(words).encode(), dtype=np.uint8).reshape(n, WORD_LENGTH)
|
||||
grid = grid - ord("a")
|
||||
|
||||
out = np.zeros((n, n), dtype=np.uint8)
|
||||
place = np.array([3**i for i in range(WORD_LENGTH)], dtype=np.uint16)
|
||||
|
||||
chunk = 256
|
||||
for start in range(0, n, chunk):
|
||||
stop = min(start + chunk, n)
|
||||
g = grid[start:stop] # (c, 5)
|
||||
# Greens first, exactly as engine.py does: every green in the word is
|
||||
# resolved before any yellow is considered.
|
||||
green = g[:, None, :] == grid[None, :, :] # (c, n, 5)
|
||||
tiles = green.astype(np.uint16) * 2
|
||||
|
||||
for letter in range(26):
|
||||
g_is = g == letter # (c, 5)
|
||||
a_is = grid == letter # (n, 5)
|
||||
if not g_is.any() or not a_is.any():
|
||||
continue
|
||||
# How many of this letter the answer has left after greens took theirs.
|
||||
green_here = green & g_is[:, None, :] # (c, n, 5)
|
||||
available = a_is.sum(1)[None, :] - green_here.sum(2) # (c, n)
|
||||
|
||||
# Candidate yellow positions, ranked left to right. A position wins a
|
||||
# yellow only if its rank is inside the remaining allocation, which
|
||||
# is what makes SASSY/BASIS come out YGGXX rather than YGGYX.
|
||||
candidate = g_is[:, None, :] & ~green_here # (c, n, 5)
|
||||
rank = np.cumsum(candidate, axis=2) - 1
|
||||
yellow = candidate & (rank < available[:, :, None])
|
||||
tiles += yellow.astype(np.uint16)
|
||||
|
||||
out[start:stop] = (tiles * place[None, None, :]).sum(2).astype(np.uint8)
|
||||
|
||||
return out
|
||||
|
||||
|
||||
@lru_cache(maxsize=1)
|
||||
def pattern_matrix() -> np.ndarray:
|
||||
"""The answer-vs-answer pattern matrix, memoised on disk.
|
||||
|
||||
Roughly 21 MB at the shipped list size. Gitignored and regenerated on
|
||||
demand — committing it would put a build artefact in a repo whose whole
|
||||
argument is that you can rebuild everything it claims.
|
||||
"""
|
||||
words = answers()
|
||||
if _CACHE.exists():
|
||||
cached = np.load(_CACHE)
|
||||
if cached.shape == (len(words), len(words)):
|
||||
return cached
|
||||
matrix = _compute_matrix(words)
|
||||
_CACHE.parent.mkdir(parents=True, exist_ok=True)
|
||||
np.save(_CACHE, matrix)
|
||||
return matrix
|
||||
|
||||
|
||||
def _verify_matrix_row(index: int) -> None:
|
||||
"""Cross-check one matrix row against engine.score_guess. Used by tests."""
|
||||
words = answers()
|
||||
row = pattern_matrix()[index]
|
||||
for j, answer in enumerate(words):
|
||||
expected = pattern_code(score_guess(words[index], answer))
|
||||
if row[j] != expected:
|
||||
raise AssertionError(f"matrix disagrees with engine at ({index}, {j})")
|
||||
|
||||
|
||||
def best_guess(candidates: np.ndarray, first: bool = False) -> int:
|
||||
"""The index of the guess with the highest expected information gain.
|
||||
|
||||
Ties break on being a candidate — among equally informative guesses, prefer
|
||||
one that could actually win, which is free expected value and also costs
|
||||
nothing in `consistency`.
|
||||
"""
|
||||
matrix = pattern_matrix()
|
||||
if first:
|
||||
# The opening guess never depends on the seed, so it is computed once
|
||||
# per process rather than once per game.
|
||||
return _best_opener()
|
||||
if len(candidates) == 1:
|
||||
return int(candidates[0])
|
||||
|
||||
sub = matrix[:, candidates] # (n_guess, n_candidates)
|
||||
best_index, best_score = -1, (-1.0, -1)
|
||||
total = len(candidates)
|
||||
candidate_set = set(candidates.tolist())
|
||||
for g in range(matrix.shape[0]):
|
||||
counts = np.bincount(sub[g], minlength=243)
|
||||
counts = counts[counts > 0]
|
||||
probability = counts / total
|
||||
entropy = float(-(probability * np.log2(probability)).sum())
|
||||
score = (entropy, 1 if g in candidate_set else 0)
|
||||
if score > best_score:
|
||||
best_index, best_score = g, score
|
||||
return best_index
|
||||
|
||||
|
||||
@lru_cache(maxsize=1)
|
||||
def _best_opener() -> int:
|
||||
matrix = pattern_matrix()
|
||||
total = matrix.shape[1]
|
||||
best_index, best_entropy = -1, -1.0
|
||||
for g in range(matrix.shape[0]):
|
||||
counts = np.bincount(matrix[g], minlength=243)
|
||||
counts = counts[counts > 0]
|
||||
probability = counts / total
|
||||
entropy = float(-(probability * np.log2(probability)).sum())
|
||||
if entropy > best_entropy:
|
||||
best_index, best_entropy = g, entropy
|
||||
return best_index
|
||||
|
||||
|
||||
def solve(answer: str, max_guesses: int = 6) -> list[str]:
|
||||
"""Play a full game against a known answer. Returns the guesses made."""
|
||||
words = answers()
|
||||
matrix = pattern_matrix()
|
||||
target = words.index(answer)
|
||||
candidates = np.arange(len(words))
|
||||
played: list[str] = []
|
||||
|
||||
for turn in range(max_guesses):
|
||||
pick = best_guess(candidates, first=(turn == 0))
|
||||
played.append(words[pick])
|
||||
if pick == target:
|
||||
return played
|
||||
observed = matrix[pick, target]
|
||||
candidates = candidates[matrix[pick, candidates] == observed]
|
||||
return played
|
||||
|
||||
|
||||
@lru_cache(maxsize=4096)
|
||||
def reference_depth(answer: str) -> int:
|
||||
"""How many guesses the shipped solver needs for this answer.
|
||||
|
||||
This is the `economy` denominator. Cached because the probe and the reward
|
||||
both ask for it on the same seeds.
|
||||
"""
|
||||
return len(solve(answer))
|
||||
|
||||
|
||||
def entropy_of(guess: str, candidates: list[str]) -> float:
|
||||
"""Expected bits gained by playing `guess` against a candidate set."""
|
||||
words = answers()
|
||||
matrix = pattern_matrix()
|
||||
g = words.index(guess)
|
||||
idx = np.array([words.index(c) for c in candidates])
|
||||
counts = np.bincount(matrix[g, idx], minlength=243)
|
||||
counts = counts[counts > 0]
|
||||
probability = counts / len(candidates)
|
||||
return float(-(probability * np.log2(probability)).sum())
|
||||
|
||||
|
||||
def consistent_candidates(history: list[tuple[str, str]]) -> list[str]:
|
||||
"""Every answer still viable given the feedback so far."""
|
||||
words = answers()
|
||||
matrix = pattern_matrix()
|
||||
alive = np.arange(len(words))
|
||||
for guess, pattern in history:
|
||||
if guess not in words:
|
||||
# A guess outside the answer list still constrains, but is not a row
|
||||
# of this matrix; fall back to the scalar engine for it.
|
||||
alive = np.array([i for i in alive if score_guess(guess, words[i]) == pattern])
|
||||
continue
|
||||
g = words.index(guess)
|
||||
alive = alive[matrix[g, alive] == pattern_code(pattern)]
|
||||
return [words[i] for i in alive]
|
||||
|
||||
|
||||
def optimal_depth_report(seeds: list[int]) -> dict[str, float]:
|
||||
"""Published diagnostics for the shipped solver over a set of seeds."""
|
||||
from .engine import answer_for_seed
|
||||
|
||||
depths = [reference_depth(answer_for_seed(s)) for s in seeds]
|
||||
return {
|
||||
"mean_guesses": sum(depths) / len(depths),
|
||||
"worst_case": max(depths),
|
||||
"solved_within_six": sum(1 for d in depths if d <= 6) / len(depths),
|
||||
}
|
||||
@@ -0,0 +1,130 @@
|
||||
"""The verifiers v1 taskset.
|
||||
|
||||
Optional: everything else in this package runs, and is gated in CI, without
|
||||
verifiers installed. Only this file needs it, so the correctness of the game,
|
||||
the reward and the probe never depends on an RL stack being resolvable.
|
||||
|
||||
Install with: uv sync --all-packages --extra verifiers
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from .engine import MAX_GUESSES, Game, answer_for_seed
|
||||
from .protocol import parse_guess, render_feedback, render_rejection, system_prompt
|
||||
from .reward import ROLES, WEIGHTS, Episode, metrics, score
|
||||
from .solver import reference_depth
|
||||
|
||||
TASKSET_ID = "wordle-five"
|
||||
|
||||
|
||||
def play_episode(seed: int, respond, max_guesses: int = MAX_GUESSES) -> dict[str, Any]:
|
||||
"""Drive one full episode against a callable that returns reply text.
|
||||
|
||||
Split out from the verifiers plumbing on purpose: this function is the
|
||||
whole game loop, it has no RL dependency, and it is what the probe, the
|
||||
tests and the fixture generator all use. The verifiers Env below is a thin
|
||||
adapter over it, so the thing being evaluated and the thing being tested
|
||||
cannot drift apart.
|
||||
"""
|
||||
game = Game(seed=seed, max_guesses=max_guesses)
|
||||
transcript: list[dict[str, Any]] = []
|
||||
prompt = system_prompt()
|
||||
|
||||
while not game.over:
|
||||
turns_left = max_guesses - len(game.history)
|
||||
reply = respond(prompt, transcript)
|
||||
guess = parse_guess(reply)
|
||||
|
||||
if guess is None:
|
||||
game.rejected += 1
|
||||
observation = render_rejection("no bracketed guess found", turns_left)
|
||||
else:
|
||||
pattern, rejection = game.play(guess)
|
||||
if rejection is not None:
|
||||
observation = render_rejection(rejection, turns_left)
|
||||
else:
|
||||
assert pattern is not None
|
||||
observation = render_feedback(guess, pattern, max_guesses - len(game.history))
|
||||
|
||||
transcript.append({"reply": reply, "guess": guess, "observation": observation})
|
||||
|
||||
# A model that only ever emits refusals would otherwise loop until the
|
||||
# process is killed; the game itself has no notion of a wasted turn.
|
||||
if game.rejected >= max_guesses:
|
||||
break
|
||||
|
||||
episode = Episode(
|
||||
answer=game.answer,
|
||||
guesses=[g for g, _ in game.history],
|
||||
patterns=[p for _, p in game.history],
|
||||
rejected=game.rejected,
|
||||
reference_depth=reference_depth(game.answer),
|
||||
max_guesses=max_guesses,
|
||||
)
|
||||
return {
|
||||
"seed": seed,
|
||||
"answer": game.answer,
|
||||
"solved": game.solved,
|
||||
"outcome": "solved" if game.solved else "failed",
|
||||
"transcript": transcript,
|
||||
"rewards": score(episode),
|
||||
"weights": WEIGHTS,
|
||||
"roles": ROLES,
|
||||
"metrics": metrics(episode),
|
||||
}
|
||||
|
||||
|
||||
def build_taskset(): # pragma: no cover - requires the optional extra
|
||||
"""The verifiers v1 Taskset. Imported lazily so the module stays optional."""
|
||||
import verifiers as vf
|
||||
|
||||
class WordleFiveEnv(vf.Env):
|
||||
"""One agent seat, `player`, playing against the engine host-side.
|
||||
|
||||
The engine plays the user role rather than the model conversing with a
|
||||
second model — the feedback is computed, not generated, which is the
|
||||
whole reason this reward is verifiable.
|
||||
"""
|
||||
|
||||
async def run(self, task, agent, trace):
|
||||
game = Game(seed=task.data.info["seed"])
|
||||
turns = 0
|
||||
async with agent.interaction(system=system_prompt()) as session:
|
||||
message = "Enter your guess to begin."
|
||||
while not game.over and turns < MAX_GUESSES * 2:
|
||||
reply = await session.turn(message)
|
||||
guess = parse_guess(reply)
|
||||
left = MAX_GUESSES - len(game.history)
|
||||
if guess is None:
|
||||
game.rejected += 1
|
||||
message = render_rejection("no bracketed guess found", left)
|
||||
else:
|
||||
pattern, rejection = game.play(guess)
|
||||
message = (
|
||||
render_rejection(rejection, left)
|
||||
if rejection is not None
|
||||
else render_feedback(guess, pattern or "", MAX_GUESSES - len(game.history))
|
||||
)
|
||||
turns += 1
|
||||
if game.rejected >= MAX_GUESSES:
|
||||
break
|
||||
|
||||
episode = Episode(
|
||||
answer=game.answer,
|
||||
guesses=[g for g, _ in game.history],
|
||||
patterns=[p for _, p in game.history],
|
||||
rejected=game.rejected,
|
||||
reference_depth=reference_depth(game.answer),
|
||||
)
|
||||
# Graded INSIDE the interaction, so the trace carries the
|
||||
# rewards rather than having them stapled on afterwards.
|
||||
for key, value in score(episode).items():
|
||||
trace.record_reward(key, value, WEIGHTS[key])
|
||||
for key, value in metrics(episode).items():
|
||||
trace.record_reward(key, value, 0.0)
|
||||
trace.info["answer"] = game.answer
|
||||
trace.info["outcome"] = "solved" if game.solved else "failed"
|
||||
|
||||
return WordleFiveEnv
|
||||
@@ -0,0 +1,56 @@
|
||||
# Where these words came from
|
||||
|
||||
Both lists are built by `build_words.py` from the two source files committed
|
||||
beside it. Nothing here is copied from the original game.
|
||||
|
||||
## Guesses — `guesses.json`, 11,846 words
|
||||
|
||||
From [`wordnik/wordlist`](https://github.com/wordnik/wordlist), snapshot
|
||||
`wordlist-20210729.txt`, **MIT licensed**, filtered to five-letter lowercase
|
||||
ASCII. Wordnik publishes this list for word-game developers, which is exactly
|
||||
what it is being used for.
|
||||
|
||||
## Answers — `answers.json`, 4,603 words
|
||||
|
||||
The **intersection** of the guess list with SCOWL's common-American tier
|
||||
(`wamerican` 2020.12.07, as shipped in `/usr/share/dict/american-english`),
|
||||
minus a short hand-curated block list.
|
||||
|
||||
The intersection is the point. A word is a possible answer if it is a real
|
||||
headword (Wordnik) *and* common enough that a non-specialist has plausibly met
|
||||
it (SCOWL). That rule is stated, reproducible, and ours — as opposed to copying
|
||||
somebody else's editorial selection of which words are fair.
|
||||
|
||||
SCOWL is distributed under a permissive licence requiring attribution, which
|
||||
`NOTICE` carries.
|
||||
|
||||
## What we deliberately did not use
|
||||
|
||||
**The original game's 2,315-word answer list.** It is the product of a
|
||||
deliberate human curation pass, which is the strongest selection-originality
|
||||
argument of any list in this space and has never been litigated. We do not need
|
||||
it, so we do not ship it.
|
||||
|
||||
We verified separately that the Wordnik list *contains* all 2,315 of those
|
||||
words — that is a statement about Wordnik's coverage, not a reason to
|
||||
redistribute the selection.
|
||||
|
||||
Consequences worth stating plainly, since they affect every number on the site:
|
||||
|
||||
- Our answer pool is **4,603**, roughly twice the original's. This game is
|
||||
**harder** than the original, and our measured solver averages are not
|
||||
comparable to published figures for it.
|
||||
- The famous results — SALET as the optimal opener, a 3.4212-guess mean, a
|
||||
proven worst case of five — are proven **for the original list**. They are
|
||||
cited on the site as exactly that, never as our own ceiling. Our own
|
||||
reference player opens **TARES** and averages **3.72**, and that number is
|
||||
computed by `solver.optimal_depth_report()` rather than quoted.
|
||||
|
||||
## Rebuilding
|
||||
|
||||
```bash
|
||||
uv run python envs/wordle_five/words/build_words.py
|
||||
```
|
||||
|
||||
Deterministic and offline. The output must be byte-identical; if it is not,
|
||||
`CONFORMANCE.txt` changes too and the TypeScript port has to be re-verified.
|
||||
Reference in New Issue
Block a user