| """Evals on eval-owned fixtures (see eval_fixtures.py — not the shipping scenarios), |
| staged so the right answer can't be argued. |
| |
| Disposition: run the real CharacterTurn on one staged line, diff its before/after stance |
| with the neutral DispositionShift comparator. Referee: one real Referee call per staged |
| turn, asserted on its outcome label. Driver voice: one real SituationDriver call per |
| staged turn, asserted with a deterministic regex — no judge. |
| |
| uv run evaluate.py quick (1 run) |
| uv run evaluate.py --strict all-of-3 |
| uv run evaluate.py [--strict] ferry one fixture only |
| """ |
|
|
| import re |
| import sys |
|
|
| import dspy |
|
|
| import dispositions |
| import engine |
| from eval_fixtures import ( |
| EVAL_FIXTURES, |
| GATE, |
| GATE_SCENE, |
| OUTCOME_CASES, |
| SHIFT_CASES, |
| VOICE_CASES, |
| ) |
| from signatures import Referee, SituationDriver |
|
|
|
|
| def mark(ok): |
| return "✅" if ok else "❌" |
|
|
|
|
| class DispositionShift(dspy.Signature): |
| """Neutral before/after diff: which way did this stance move? Sees only the two stance |
| cells — no persona, goal, or rules.""" |
|
|
| previous_disposition: str = dspy.InputField( |
| desc="the character's stance before the line" |
| ) |
| next_disposition: str = dspy.InputField( |
| desc="the character's stance after the line" |
| ) |
| shift: str = dspy.OutputField( |
| desc="exactly one word. 'better' = warmer, 'worse' = harder, 'neutral' = unchanged." |
| ) |
|
|
|
|
| |
| SECOND_PERSON = re.compile(r"\byou\b|\byours?\b", re.IGNORECASE) |
| PLAYER_IN_THIRD = re.compile(r"\bthe player\b|\bthe traveller\b", re.IGNORECASE) |
|
|
|
|
| def run_suite(title, label, cases, passes, describe, runs): |
| """The all-of-N harness every suite shares: run each case `runs` times, demand every |
| run pass, print one line per case and a section summary, return the section verdict.""" |
| print(f"\n{'=' * 70}\n {title} [all-of-{runs}]\n{'=' * 70}") |
| passed = 0 |
| for case in cases: |
| oks = [passes(case) for _ in range(runs)] |
| all_ok = all(oks) |
| passed += all_ok |
| print(f" {mark(all_ok)} {describe(case)} {sum(oks)}/{runs}") |
| ok = passed == len(cases) |
| print(f"\n {mark(ok)} {label}: {passed}/{len(cases)} passed all {runs} runs") |
| return ok |
|
|
|
|
| def voice_passes(case, driver): |
| pred = driver( |
| goal=GATE.goal, scene=GATE_SCENE, this_turn=case.this_turn, room=case.room |
| ) |
| beat = pred.next_scene |
| return bool(SECOND_PERSON.search(beat)) and not PLAYER_IN_THIRD.search(beat) |
|
|
|
|
| def evaluate_voice(driver, runs): |
| return run_suite( |
| "Gate — driver voice", |
| "voice", |
| VOICE_CASES, |
| lambda case: voice_passes(case, driver), |
| lambda case: f"{case.note[:50]:<50} want 'you'-voice ", |
| runs, |
| ) |
|
|
|
|
| def outcome_passes(case, referee): |
| pred = referee( |
| goal=GATE.goal, |
| situation=GATE_SCENE, |
| this_turn=case.this_turn, |
| room=case.room, |
| player_so_far=case.player_so_far, |
| final_turn=case.final_turn, |
| ) |
| return engine.norm_label(pred.outcome) in case.expect |
|
|
|
|
| def evaluate_outcomes(referee, runs): |
| return run_suite( |
| "Gate — referee outcome", |
| "outcome", |
| OUTCOME_CASES, |
| lambda case: outcome_passes(case, referee), |
| lambda case: f"{case.note[:50]:<50} want {'/'.join(sorted(case.expect)):<14}", |
| runs, |
| ) |
|
|
|
|
| def shift_passes(scen, case, actor, comparator): |
| """Real character turn, then diff its before/after stance toward the player.""" |
| char = next(c for c in scen.characters if c.name.lower() == case.char.lower()) |
| keys = dispositions.row_keys(scen) |
| prev = dispositions.merge_row({}, case.disposition or char.disposition, keys) |
| turn = actor( |
| persona=char.persona, |
| disposition=dispositions.render_row(char.name, prev, keys), |
| scene=case.scene or scen.intro, |
| since_you_spoke=case.since or f"PLAYER: {case.line}", |
| ) |
| new = dispositions.merge_row(prev, turn.updated_dispositions, keys) |
| verdict = comparator( |
| previous_disposition=prev["Player"], |
| next_disposition=new["Player"], |
| ) |
| return engine.norm_label(verdict.shift) == case.expect |
|
|
|
|
| def evaluate_suite(scen, actor, comparator, runs): |
| return run_suite( |
| f"{scen.title} — disposition (id: {scen.id})", |
| "disposition", |
| SHIFT_CASES.get(scen.id, []), |
| lambda case: shift_passes(scen, case, actor, comparator), |
| lambda case: ( |
| f"[{case.char:<8}] {(case.note or case.line)[:42]:<42} want {case.expect:<8}" |
| ), |
| runs, |
| ) |
|
|
|
|
| def main(): |
| args = sys.argv[1:] |
| runs = 1 |
| if "--strict" in args: |
| args.remove("--strict") |
| runs = 3 |
| lm = engine.make_lm(0.2, cache=False) |
|
|
| registry = {s.id: s for s in EVAL_FIXTURES} |
| only = args[0] if args else None |
| scenarios = [registry[only]] if only else list(registry.values()) |
| print(f" mode: all-of-{runs}, cache off") |
|
|
| comparator = dspy.Predict(DispositionShift) |
| results = {} |
| for scen in scenarios: |
| if scen.id not in SHIFT_CASES: |
| continue |
| with dspy.context(lm=lm): |
| actor = dspy.Predict(engine.character_signature(scen)) |
| results[f"{scen.id}/disposition"] = evaluate_suite( |
| scen, actor, comparator, runs |
| ) |
|
|
| if not only or only == "gate": |
| with dspy.context(lm=lm): |
| referee = dspy.Predict(Referee.with_instructions(GATE.director_rules)) |
| results["gate/outcome"] = evaluate_outcomes(referee, runs) |
| driver = dspy.Predict( |
| SituationDriver.with_instructions(GATE.director_rules) |
| ) |
| results["gate/voice"] = evaluate_voice(driver, runs) |
|
|
| print(f"\n{'=' * 70}\n SUMMARY") |
| for key, ok in results.items(): |
| print(f" {mark(ok)} {key}") |
| sys.exit(0 if all(results.values()) else 1) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
