387 lines
17 KiB
Python
387 lines
17 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import random
|
|
import re
|
|
import sys
|
|
from datetime import datetime
|
|
from difflib import SequenceMatcher
|
|
from pathlib import Path
|
|
|
|
from engine_lib.models import TurnResult, END_MARKER
|
|
from engine_lib import config
|
|
from engine_lib.context import build_system_prompt
|
|
from engine_lib.validation import validate_turn
|
|
from engine_lib.tools_handler import execute_tool, describe_change, extract_tool_calls
|
|
from engine_lib.parsing import log_turn_details
|
|
from engine_lib import state
|
|
from engine_lib.llm import call_llm
|
|
from engine_lib.paths import CHARACTER_CREATION_PATH, RULES_INJECTION_PATH
|
|
|
|
|
|
class GameEngine:
|
|
REQUIRED_TOOL_ARGS: dict[str, list[str]] = {
|
|
"modify_inventory": ["operation", "item"],
|
|
"modify_note": ["operation"],
|
|
"modify_world": ["operation", "value"],
|
|
"modify_journal": ["operation", "value"],
|
|
}
|
|
|
|
def __init__(self, session_dir: str | Path | None = None):
|
|
self.config = config.load_config()
|
|
|
|
def _check_required_tool_args(self, state_changes: list[dict]) -> str:
|
|
"""Check that all state-changing tool calls have required args. Returns empty string if OK, or a description of what's missing."""
|
|
missing = []
|
|
for tc in state_changes:
|
|
name = tc.get("tool", "")
|
|
req = self.REQUIRED_TOOL_ARGS.get(name)
|
|
if not req:
|
|
continue
|
|
args = tc.get("args") or {k: v for k, v in tc.items() if k != "tool"}
|
|
for arg in req:
|
|
val = args.get(arg)
|
|
if val is None or (isinstance(val, str) and not val.strip()):
|
|
missing.append(f"{name}: missing required `{arg}`")
|
|
return "; ".join(missing)
|
|
|
|
def generate_turn(
|
|
self,
|
|
player_action: str | None = None,
|
|
recent_narrative: str | None = None,
|
|
on_thought: callable = None,
|
|
on_action: callable = None,
|
|
) -> TurnResult:
|
|
now = datetime.now()
|
|
state.append_llm_log(f"\n{'='*60}")
|
|
state.append_llm_log(f"=== Turn — {now.strftime('%Y-%m-%d %H:%M:%S')} ===")
|
|
state.append_llm_log(f"{'='*60}")
|
|
if player_action:
|
|
state.append_llm_log(f"Player: {player_action}")
|
|
if recent_narrative is None:
|
|
recent_narrative = state.read_recent_book(2)
|
|
session_log = state.read_recent_log()
|
|
|
|
die_roll = random.randint(1, 6)
|
|
state.append_llm_log(f"Dice: {die_roll} (1d6)")
|
|
|
|
lm = self.config.get("llm", {})
|
|
model = lm.get("model", "ollama/llama3.1")
|
|
|
|
if on_action:
|
|
on_action("DM is preparing a response")
|
|
|
|
is_new_game = not player_action and not recent_narrative
|
|
system = build_system_prompt(recent_narrative=recent_narrative, recent_log=session_log)
|
|
if is_new_game:
|
|
cc = state.read_file(CHARACTER_CREATION_PATH)
|
|
if cc:
|
|
system += f"\n\n## Character Creation Reference\n{cc}"
|
|
state.append_llm_log(f"\n[NEW GAME] injected character_creation.md ({len(cc)} chars)")
|
|
|
|
is_meta = bool(player_action and player_action.strip().startswith(">"))
|
|
|
|
base_parts = []
|
|
if player_action:
|
|
base_parts.append(f"## Player's Request\n{player_action}")
|
|
if is_meta:
|
|
base_parts.append(
|
|
"## Instructions\n"
|
|
"The player's message starts with `>` — this is a meta out-of-character question to the DM. "
|
|
"Do NOT advance the story. Respond as the DM in meta language, starting the response with `>`. "
|
|
"Use the `narrative` tool to output your meta response. Do NOT call any other tools (no modify_journal, no finalize_turn, no rolls, no state changes)."
|
|
)
|
|
elif is_new_game:
|
|
base_parts.append(
|
|
"## Instructions\n"
|
|
"This is a new story. Welcome the player and guide them through the game setup."
|
|
)
|
|
else:
|
|
base_parts.append(
|
|
"## Instructions\n"
|
|
"Advance the story based on the player's request. "
|
|
"All state is shown above — write the outcome directly."
|
|
)
|
|
if not is_meta:
|
|
base_parts.append(f"\n*A die is cast: **{die_roll}** (1d6).*")
|
|
base_user = "\n\n".join(base_parts)
|
|
|
|
MAX_RETRIES = 2
|
|
tool_calls = []
|
|
book_log = ""
|
|
ambience = None
|
|
errors: list[str] = []
|
|
changes: list[str] = []
|
|
start_time = datetime.now()
|
|
total_attempts = 0
|
|
prev_raw = ""
|
|
|
|
_ = None # placeholder
|
|
|
|
for attempt in range(MAX_RETRIES + 1):
|
|
total_attempts = attempt + 1
|
|
user = base_user
|
|
if attempt > 0:
|
|
user += f"\n\n---\n\n## Your Previous Response\n\n```\n{prev_raw}\n```\n\n---\n\n## Feedback\n{feedback}"
|
|
|
|
state.append_llm_log(f"\n[TOOL] Attempt {attempt + 1}/{MAX_RETRIES + 1} — {len(system)} chars system, {len(user)} chars user")
|
|
|
|
text = call_llm(
|
|
[{"role": "system", "content": system}, {"role": "user", "content": user}],
|
|
label="Turn generation",
|
|
)
|
|
|
|
if not text or not text.strip():
|
|
if attempt < MAX_RETRIES:
|
|
feedback = f"Your response was empty. Generate a complete turn with narrative and state changes."
|
|
state.append_llm_log("\n[RETRY] empty response")
|
|
if on_action:
|
|
on_action("DM is weaving the tale...")
|
|
continue
|
|
return TurnResult(error="LLM returned empty response after retries")
|
|
|
|
raw = text.strip()
|
|
state.append_llm_log(f"\n[TOOL] got {len(raw)} chars in {(datetime.now() - start_time).total_seconds() * 1000:.1f}ms")
|
|
prev_raw = raw
|
|
|
|
tool_calls = extract_tool_calls(raw)
|
|
if not tool_calls:
|
|
state.append_llm_log("\n[TOOL] no tool blocks found")
|
|
|
|
# First pass — extract narrative + identify state changes (don't execute yet)
|
|
book_log = ""
|
|
ambience = None
|
|
log_entry = None
|
|
meta_log = ""
|
|
state_changes: list[dict] = []
|
|
|
|
for tc in tool_calls:
|
|
name = tc.get("tool", "")
|
|
args = tc.get("args") or {k: v for k, v in tc.items() if k != "tool"}
|
|
|
|
if name == "narrative":
|
|
text = args.get("text", "") or ""
|
|
if text:
|
|
book_log = (book_log + "\n\n" + text) if book_log else text
|
|
elif name == "finalize_turn":
|
|
raw = (args.get("ambience") or "").strip().lower()
|
|
if raw:
|
|
valid = state.get_valid_ambiences()
|
|
if raw in valid:
|
|
ambience = raw
|
|
else:
|
|
state.append_llm_log(f"\n[WARN] invalid ambience '{raw}'")
|
|
ambience = None
|
|
if args.get("log_entry"):
|
|
log_entry = args["log_entry"]
|
|
if args.get("meta_log"):
|
|
meta_log = args["meta_log"]
|
|
elif name == "read_rules":
|
|
cat = args.get("category", "mechanics")
|
|
result = execute_tool("read_rules", {"category": cat})
|
|
state.append_llm_log(f"\n[READ RULES] loaded {len(result)} chars")
|
|
RULES_INJECTION_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
RULES_INJECTION_PATH.write_text(result)
|
|
else:
|
|
state_changes.append(tc)
|
|
|
|
# Required args check — reject if any state-changing tool is missing required arguments
|
|
missing_args = self._check_required_tool_args(state_changes)
|
|
if missing_args:
|
|
state.append_llm_log(f"\n[TURN MISSING ARGS] {missing_args}")
|
|
if attempt < MAX_RETRIES:
|
|
feedback = f"The following tool calls are missing required arguments: {missing_args}. Include all required fields for each tool and regenerate."
|
|
state.append_llm_log(f"\n[TURN REGENERATE] (missing args) attempt {attempt + 2}")
|
|
if on_action:
|
|
on_action("DM is consulting the fates...")
|
|
continue
|
|
state.append_llm_log(f"\n[TURN MISSING ARGS EXCEEDED] accepting despite missing args")
|
|
|
|
# Meta check — reject if state changes produced for a meta action
|
|
if is_meta and state_changes:
|
|
state.append_llm_log(f"\n[TURN META REJECTED] state changes not allowed for meta action")
|
|
if attempt < MAX_RETRIES:
|
|
feedback = f"This is a meta action. Do NOT call any state-changing tools. Respond only with meta text (starting with `>`) and no tool calls beyond a finalize_turn."
|
|
state.append_llm_log(f"\n[TURN REGENERATE] (meta) attempt {attempt + 2}")
|
|
if on_action:
|
|
on_action("DM is consulting the fates...")
|
|
continue
|
|
state.append_llm_log(f"\n[TURN META EXCEEDED] accepting despite state changes")
|
|
|
|
# Narrative check — reject if finalized with log_entry but no narrative
|
|
if not is_meta and log_entry and not book_log:
|
|
state.append_llm_log(f"\n[TURN NO NARRATIVE] finalized with log_entry but no narrative")
|
|
if attempt < MAX_RETRIES:
|
|
feedback = f"You called finalize_turn with a log_entry but produced no narrative. Every turn must include a `narrative` tool block with the story. Regenerate with both narrative and log_entry."
|
|
state.append_llm_log(f"\n[TURN REGENERATE] (no narrative) attempt {attempt + 2}")
|
|
if on_action:
|
|
on_action("DM is weaving the tale...")
|
|
continue
|
|
state.append_llm_log(f"\n[TURN NO NARRATIVE EXCEEDED] accepting despite missing narrative")
|
|
|
|
# Duplicate check — reject if narrative is 80%+ similar to last book entry
|
|
if not is_meta and book_log:
|
|
prev = state.read_recent_book(1)
|
|
if prev and prev not in ("*No prior story.*",):
|
|
prev_text = re.sub(r"^## Turn \d+\n\n", "", prev, flags=re.MULTILINE).strip()
|
|
ratio = SequenceMatcher(None, book_log, prev_text).ratio()
|
|
if ratio >= 0.8:
|
|
state.append_llm_log(f"\n[TURN DUPLICATE] {ratio:.0%} match with previous turn")
|
|
if attempt < MAX_RETRIES:
|
|
feedback = f"The narrative is nearly identical to the previous turn. Generate something new and different."
|
|
state.append_llm_log(f"\n[TURN REGENERATE] (duplicate) attempt {attempt + 2}")
|
|
if on_action:
|
|
on_action("DM is weaving the tale...")
|
|
continue
|
|
state.append_llm_log(f"\n[TURN DUPLICATE EXCEEDED] cannot generate unique narrative")
|
|
return TurnResult(
|
|
book_log="",
|
|
log_entry="Your action was rejected — could not generate a unique narrative.",
|
|
user_prompt=f"*Your action:\n\t\"{player_action}\"\nwas rejected:\nengine failed to produce unique narrative*",
|
|
)
|
|
|
|
# Validate the generated turn
|
|
if on_action:
|
|
on_action("DM is validating the response")
|
|
if player_action and book_log:
|
|
valid, reason, action = validate_turn(
|
|
player_action,
|
|
narrative=book_log,
|
|
log_entry=log_entry or "",
|
|
changes=state_changes,
|
|
story=recent_narrative,
|
|
log=session_log,
|
|
meta=is_meta,
|
|
)
|
|
|
|
if valid:
|
|
state.append_llm_log(f"\n[TURN VALID] {reason}")
|
|
elif reason == "Unrecognized":
|
|
if attempt < MAX_RETRIES:
|
|
feedback = f"The validation system could not process the previous turn. Please regenerate."
|
|
state.append_llm_log(f"\n[TURN REGENERATE] (unrecognized) attempt {attempt + 2}")
|
|
if on_action:
|
|
on_action("DM is consulting the fates...")
|
|
continue
|
|
state.append_llm_log(f"\n[TURN UNRECOGNIZED] cannot validate turn")
|
|
return TurnResult(
|
|
book_log="",
|
|
log_entry="Your action was rejected — cannot validate turn.",
|
|
user_prompt=f"*Your action:\n\t\"{player_action}\"\nwas rejected:\ncannot validate turn*",
|
|
)
|
|
elif action == "reject":
|
|
state.append_llm_log(f"\n[TURN REJECTED] {reason}")
|
|
return TurnResult(
|
|
book_log="",
|
|
log_entry=f"Your action was rejected — {reason}.",
|
|
user_prompt=f"*Your action:\n\t\"{player_action}\"\nwas rejected:\n\t{reason}*",
|
|
)
|
|
elif action == "regenerate" and attempt < MAX_RETRIES:
|
|
validator_tool = json.dumps({"tool": "validate", "args": {"valid": False, "reason": reason, "action": "regenerate"}})
|
|
feedback = f"The validation tool returned:\n```tool\n{validator_tool}\n```\n\nPlease regenerate the turn addressing the issues above. Keep the same player action but fix the problems described."
|
|
state.append_llm_log(f"\n[TURN REGENERATE] attempt {attempt + 2}: {reason}")
|
|
if on_action:
|
|
on_action("DM is searching for inspiration...")
|
|
continue
|
|
else:
|
|
state.append_llm_log(f"\n[TURN REGENERATE EXCEEDED] accepting despite: {reason}")
|
|
else:
|
|
state.append_llm_log("\n[TURN SKIP VALIDATION] no player action or no narrative")
|
|
|
|
# Accept this turn — execute all tool calls
|
|
break
|
|
|
|
if is_meta:
|
|
tool_calls = [tc for tc in tool_calls if tc.get("tool") == "narrative"]
|
|
|
|
# Second pass — execute all tool calls
|
|
extr_start = datetime.now()
|
|
|
|
for tc in tool_calls:
|
|
name = tc.get("tool", "")
|
|
args = tc.get("args") or {k: v for k, v in tc.items() if k != "tool"}
|
|
|
|
if name in ("narrative", "read_rules"):
|
|
pass
|
|
else:
|
|
result = execute_tool(name, args)
|
|
|
|
if name not in ("narrative", "finalize_turn", "player_roll", "read_rules"):
|
|
if result.startswith("**Error:") or result.startswith("Tool error") or result.startswith("Unknown"):
|
|
errors.append(f"{name}: {result}")
|
|
else:
|
|
desc = describe_change(name, args)
|
|
if desc:
|
|
changes.append(desc)
|
|
|
|
if not log_entry and book_log:
|
|
clean = re.sub(r'\s+', ' ', book_log).strip()
|
|
sentences = re.split(r'(?<=[.!?])\s+', clean)
|
|
log_entry = sentences[0][:200] if sentences else clean[:200]
|
|
|
|
apply_elapsed = (datetime.now() - extr_start).total_seconds() * 1000
|
|
state.append_llm_log(f"\n[APPLY] {len(changes)} changes in {apply_elapsed:.1f}ms")
|
|
|
|
total_elapsed = (datetime.now() - start_time).total_seconds() * 1000
|
|
|
|
game_over = END_MARKER in book_log
|
|
|
|
if on_action:
|
|
on_action("Turn complete")
|
|
|
|
log_turn_details(
|
|
player_action=player_action or "",
|
|
last_prompt=recent_narrative or "",
|
|
strategy_name="tools",
|
|
die_roll=die_roll,
|
|
model=model,
|
|
temperature=lm.get("temperature", 0.8),
|
|
max_tokens=lm.get("max_tokens", 4096),
|
|
book_log=book_log,
|
|
log_entry=log_entry or "",
|
|
ambience=ambience,
|
|
tool_calls=tool_calls,
|
|
meta_log=meta_log,
|
|
)
|
|
|
|
return TurnResult(
|
|
book_log=book_log,
|
|
log_entry=log_entry,
|
|
ambience=ambience,
|
|
debug_info="; ".join(errors) if errors else "",
|
|
changes=changes,
|
|
is_meta=is_meta,
|
|
game_over=game_over,
|
|
meta_log=meta_log,
|
|
)
|
|
|
|
|
|
def main():
|
|
"""Generate a turn from the command line (debug/testing)."""
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(description="The Chaos Game Engine (CLI)")
|
|
parser.add_argument("--action", "-a", help="Player action text")
|
|
args = parser.parse_args()
|
|
|
|
engine = GameEngine()
|
|
result = engine.generate_turn(
|
|
player_action=args.action,
|
|
)
|
|
|
|
if result.error:
|
|
print(f"ERROR: {result.error}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
print(result.book_log)
|
|
if result.user_prompt:
|
|
print(f"\n{result.user_prompt}")
|
|
if result.log_entry:
|
|
print(f"\n[Log] {result.log_entry}")
|
|
if result.ambience:
|
|
print(f"[Ambience] {result.ambience}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|