ctan
e5c4892a57
Representative 7-combatant caster-heavy encounter on sample_arena, measured through the real engine path (engine._build_observation at each policy call, delegating to default_policy): - observation: 64,204 bytes at first call, 67,712 max mid-battle (~16.9k tokens) - policy calls/battle: 124 (7 combatants, round_cap=20), 12 for a 2v2 duel - projected calls per sweep cell: 124 / 12,400 / 124,000 at runs 1/100/1000 Loose bounds (obs < 200 KB, calls/battle < 200) + seed-determinism pinned.
184 lines
6.7 KiB
Python
184 lines
6.7 KiB
Python
"""Phase-2 measurement spike (docs/phase-2-llm-architecture.md section 12.b).
|
|
|
|
Two numbers decide whether a per-turn LLM is viable for the Monte Carlo
|
|
sweep: the OBSERVATION SIZE (bytes / rough tokens) and the POLICY CALLS PER
|
|
BATTLE (hence per sweep cell). This module measures both through the real
|
|
engine path on a representative caster-heavy encounter, prints a summary
|
|
block for the orchestrator, and pins LOOSE upper bounds so a regression
|
|
(bigger snapshot schema, policy called twice per turn, ...) fails loudly.
|
|
No engine behaviour is changed here: the recorder only reads
|
|
``engine._build_observation`` (pure read, consumes no RNG) at each policy
|
|
invocation and delegates to ``default_policy``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import TYPE_CHECKING
|
|
|
|
from pf1e_simulator.combat import Action, CombatantState, CombatEngine, default_policy
|
|
from pf1e_simulator.grid import Grid
|
|
from pf1e_simulator.loaders import load_combatant
|
|
from pf1e_simulator.map import load_map
|
|
from pf1e_simulator.rng import SeededRng
|
|
from pf1e_simulator.runner import EncounterSpec, Side, build_states
|
|
from pf1e_simulator.spells import SPELLS_DIR, SpellSpec, load_spell_registry
|
|
|
|
if TYPE_CHECKING:
|
|
import pytest
|
|
|
|
from pf1e_simulator.models import Combatant
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
MAP = ROOT / "data" / "maps" / "sample_arena.yaml"
|
|
|
|
# Loose ceilings: a regression past these fails loudly; the real values are
|
|
# far below (printed by the tests for the orchestrator to quote).
|
|
MAX_OBS_BYTES = 200_000
|
|
MAX_CALLS_PER_BATTLE = 200
|
|
|
|
# Round cap for the measurement battles: generous enough to exercise buffs,
|
|
# conditions and summons, small enough to bound the calls/battle figure.
|
|
MEASURE_ROUND_CAP = 20
|
|
|
|
|
|
class _Recorder:
|
|
"""Counting policy wrapper: counts invocations, snapshots observation size.
|
|
|
|
Delegates every decision to ``default_policy`` so the battle trajectory
|
|
(and therefore the call count) is the real engine path. The observation
|
|
is rebuilt at each call via ``engine._build_observation(state)`` - the
|
|
same snapshot the engine hands to an ``LLMPolicy`` - and its JSON size
|
|
is tracked (first call = start of battle, max = fullest mid-battle
|
|
snapshot with buffs/conditions active).
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self.calls = 0
|
|
self.first_bytes: int | None = None
|
|
self.max_bytes = 0
|
|
|
|
def policy(self, engine: CombatEngine, state: CombatantState) -> tuple[Action, ...]:
|
|
self.calls += 1
|
|
obs = engine._build_observation(state).to_dict()
|
|
size = len(json.dumps(obs))
|
|
if self.first_bytes is None:
|
|
self.first_bytes = size
|
|
self.max_bytes = max(self.max_bytes, size)
|
|
return default_policy(engine, state)
|
|
|
|
|
|
def _c(path: str) -> Combatant:
|
|
return load_combatant(ROOT / "data" / path)
|
|
|
|
|
|
def _representative_spec() -> EncounterSpec:
|
|
"""Caster-heavy 3v4 on the sample arena (walls, rubble, water, pillars).
|
|
|
|
Oni (channel + heal + summon), misty (spell combat), nairda (hexes) vs
|
|
the lich (high-level caster, fear aura, paralysis touch), a thrune
|
|
agent, an orc and a goblin: buffs, conditions, spells and pairs all
|
|
load the snapshot.
|
|
"""
|
|
return EncounterSpec(
|
|
map=load_map(MAP),
|
|
round_cap=MEASURE_ROUND_CAP,
|
|
sides=(
|
|
Side(
|
|
name="players",
|
|
combatants=(_c("pcs/oni.json"), _c("pcs/misty.json"), _c("pcs/nairda.json")),
|
|
zone="A",
|
|
),
|
|
Side(
|
|
name="monsters",
|
|
combatants=(
|
|
_c("monsters/lich.json"),
|
|
_c("monsters/thrune_agent.json"),
|
|
_c("monsters/orc.json"),
|
|
_c("monsters/goblin.json"),
|
|
),
|
|
zone="B",
|
|
),
|
|
),
|
|
)
|
|
|
|
|
|
def _duel_spec() -> EncounterSpec:
|
|
"""Small 2v2 baseline to show how calls/battle scales with combatants."""
|
|
return EncounterSpec(
|
|
map=load_map(MAP),
|
|
round_cap=MEASURE_ROUND_CAP,
|
|
sides=(
|
|
Side(name="players", combatants=(_c("monsters/goblin.json"),) * 2, zone="A"),
|
|
Side(name="monsters", combatants=(_c("monsters/orc.json"),) * 2, zone="B"),
|
|
),
|
|
)
|
|
|
|
|
|
def _run_battle(
|
|
spec: EncounterSpec, seed: int, registry: dict[str, SpellSpec]
|
|
) -> _Recorder:
|
|
grid = Grid.from_spec(spec.map)
|
|
states = build_states(spec)
|
|
rec = _Recorder()
|
|
CombatEngine(
|
|
SeededRng(seed),
|
|
grid,
|
|
states,
|
|
round_cap=spec.round_cap,
|
|
policy=rec.policy,
|
|
spell_registry=registry,
|
|
).run()
|
|
return rec
|
|
|
|
|
|
def _emit(line: str) -> None:
|
|
sys.stdout.write(line + "\n")
|
|
|
|
|
|
def test_observation_size_and_policy_calls(capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""Measure both numbers on the representative battle; pin loose bounds."""
|
|
registry = load_spell_registry(SPELLS_DIR)
|
|
spec = _representative_spec()
|
|
rec = _run_battle(spec, seed=1, registry=registry)
|
|
assert rec.first_bytes is not None
|
|
|
|
# Determinism: same seed -> same trajectory -> same numbers.
|
|
rec2 = _run_battle(spec, seed=1, registry=registry)
|
|
assert rec2.calls == rec.calls
|
|
assert rec2.first_bytes == rec.first_bytes
|
|
assert rec2.max_bytes == rec.max_bytes
|
|
|
|
# Loose bounds (regression guards, not exact pins).
|
|
assert rec.first_bytes < MAX_OBS_BYTES
|
|
assert rec.max_bytes < MAX_OBS_BYTES
|
|
assert rec.calls < MAX_CALLS_PER_BATTLE
|
|
|
|
duel = _run_battle(_duel_spec(), seed=1, registry=registry)
|
|
assert duel.calls < MAX_CALLS_PER_BATTLE
|
|
assert duel.first_bytes is not None
|
|
|
|
first_tokens = rec.first_bytes // 4
|
|
max_tokens = rec.max_bytes // 4
|
|
_emit("=== PHASE2 SUMMARY (measurement spike) ===")
|
|
_emit(f"encounter: 7 combatants, round_cap={MEASURE_ROUND_CAP}, seed=1")
|
|
_emit(f"observation bytes (first call): {rec.first_bytes} (~{first_tokens} tokens)")
|
|
_emit(f"observation bytes (max in battle): {rec.max_bytes} (~{max_tokens} tokens)")
|
|
_emit(f"policy calls/battle (7 combatants): {rec.calls}")
|
|
_emit(f"policy calls/battle (2v2 duel): {duel.calls}")
|
|
_emit(f"observation 2v2 duel bytes: {duel.first_bytes} (~{duel.first_bytes // 4} tokens)")
|
|
for runs in (1, 100, 1000):
|
|
_emit(f"projected calls per sweep cell at runs={runs}: {rec.calls * runs}")
|
|
out = capsys.readouterr().out
|
|
assert "PHASE2 SUMMARY" in out
|
|
|
|
|
|
def test_projection_is_linear_in_runs() -> None:
|
|
"""Calls per sweep cell = calls/battle x runs (fixed per-battle count)."""
|
|
registry = load_spell_registry(SPELLS_DIR)
|
|
rec = _run_battle(_representative_spec(), seed=7, registry=registry)
|
|
assert rec.calls > 0
|
|
assert rec.calls * 1000 == (rec.calls * 100) * 10
|