File size: 3,838 Bytes
566718d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 | """Headless live-LLM self-play.
Drives the *real* runtime pipeline against the configured Modal endpoints (no
fake completer) for N ticks, then prints the ledger path and a behaviour
summary. Used to audit how the real NPC model reacts to the prompt briefing.
Usage:
.venv\\Scripts\\python.exe scripts\\live_sim.py --ticks 60
"""
from __future__ import annotations
import argparse
import json
from collections import Counter
from pathlib import Path
import sys
REPO_ROOT = Path(__file__).resolve().parents[1]
SRC_ROOT = REPO_ROOT / "src"
if str(SRC_ROOT) not in sys.path:
sys.path.insert(0, str(SRC_ROOT))
def _load_env() -> None:
env_path = REPO_ROOT / ".env"
if not env_path.is_file():
return
import os
for line in env_path.read_text(encoding="utf-8").splitlines():
line = line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, value = line.split("=", 1)
os.environ.setdefault(key.strip(), value.strip())
def main() -> None:
_load_env()
from world_simulator.api.runtime import create_game_runtime
from world_simulator.config import load_game_config
from world_simulator.simulation.spawning import create_world
parser = argparse.ArgumentParser(description="Headless live-LLM self-play.")
parser.add_argument("--ticks", type=int, default=60)
parser.add_argument(
"--config", type=Path, default=REPO_ROOT / "config" / "game.modal.local.json"
)
args = parser.parse_args()
config = load_game_config(args.config)
world = create_world(config)
runtime = create_game_runtime(world=world, config=config)
print(
f"LIVE start config={args.config.name} ticks={args.ticks} "
f"simulator={runtime.simulator_name} npcs={len(world.npcs)}"
)
for i in range(args.ticks):
status, _payload = runtime.tick()
if int(status) != 200:
print(f"tick {i} failed: {status}")
break
if (i + 1) % 10 == 0:
print(f" ...tick {i + 1} done")
ledger_path = runtime._ledger.ledger_path # noqa: SLF001
print(f"LIVE done. ledger={ledger_path}")
_summarize(ledger_path)
def _summarize(ledger_path: Path) -> None:
actions: Counter[str] = Counter()
verdicts: Counter[str] = Counter()
fallbacks: Counter[str] = Counter()
event_types: Counter[str] = Counter()
sources: Counter[str] = Counter()
with ledger_path.open("r", encoding="utf-8") as handle:
for line in handle:
if not line.strip():
continue
rec = json.loads(line)
phase = rec.get("phase")
if phase == "npc_response":
verdict = rec.get("validator_verdict") or {}
verdicts[str(verdict.get("status"))] += 1
parsed = rec.get("parsed_action") or {}
if isinstance(parsed, dict) and parsed.get("action"):
actions[str(parsed["action"])] += 1
elif phase == "npc_fallback":
fallbacks[str(rec.get("reason"))] += 1
elif phase == "engine_events":
for event in rec.get("events") or []:
event_types[str(event.get("type"))] += 1
print("\n== parsed LLM actions ==")
for action, count in actions.most_common():
print(f" {action}: {count}")
print("\n== validator verdicts ==")
for status, count in verdicts.most_common():
print(f" {status}: {count}")
print("\n== fallback reasons ==")
for reason, count in fallbacks.most_common() or [("none", 0)]:
print(f" {reason}: {count}")
print("\n== engine event types ==")
for event_type, count in event_types.most_common():
print(f" {event_type}: {count}")
if __name__ == "__main__":
main()
|