266 lines
10 KiB
Python
266 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import random
|
|
import re
|
|
import sys
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from engine_lib.models import TurnResult
|
|
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
|
|
|
|
|
|
class GameEngine:
|
|
def __init__(self, session_dir: str | Path | None = None):
|
|
self.config = config.load_config()
|
|
|
|
def generate_turn(
|
|
self,
|
|
player_action: str | None = None,
|
|
recent_narrative: str | None = None,
|
|
on_thought: callable = None,
|
|
on_action: callable = None,
|
|
on_player_roll: 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")
|
|
|
|
system = build_system_prompt(recent_narrative=recent_narrative, recent_log=session_log)
|
|
base_parts = []
|
|
if player_action:
|
|
base_parts.append(f"## Player's Request\n{player_action}")
|
|
if not player_action and not recent_narrative:
|
|
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."
|
|
)
|
|
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
|
|
|
|
for attempt in range(MAX_RETRIES + 1):
|
|
total_attempts = attempt + 1
|
|
user = base_user
|
|
if attempt > 0:
|
|
user += f"\n\n---\n\n## Turn Generation 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 = "Your response was empty. Generate a complete turn with narrative and state changes."
|
|
state.append_llm_log("\n[RETRY] empty response")
|
|
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")
|
|
|
|
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
|
|
state_changes: list[dict] = []
|
|
|
|
for tc in tool_calls:
|
|
name = tc.get("tool", "")
|
|
args = tc.get("args", {})
|
|
|
|
if name == "narrative":
|
|
text = args.get("text", "")
|
|
if text:
|
|
book_log = (book_log + "\n\n" + text) if book_log else text
|
|
elif name == "finalize_turn":
|
|
if args.get("ambience"):
|
|
ambience = args["ambience"]
|
|
if args.get("log_entry"):
|
|
log_entry = args["log_entry"]
|
|
elif name == "player_roll":
|
|
pass
|
|
elif name not in ("roll",):
|
|
state_changes.append(tc)
|
|
|
|
# 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,
|
|
)
|
|
|
|
if valid:
|
|
state.append_llm_log(f"\n[TURN VALID] {reason}")
|
|
elif reason == "Unrecognized":
|
|
if attempt < MAX_RETRIES:
|
|
feedback = "The validation system could not process the previous turn. Please regenerate."
|
|
state.append_llm_log(f"\n[TURN REGENERATE] (unrecognized) attempt {attempt + 2}")
|
|
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}")
|
|
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
|
|
|
|
# Second pass — execute all tool calls
|
|
extr_start = datetime.now()
|
|
|
|
for tc in tool_calls:
|
|
name = tc.get("tool", "")
|
|
args = tc.get("args", {})
|
|
|
|
if name == "narrative":
|
|
pass
|
|
elif name == "finalize_turn":
|
|
pass
|
|
elif name == "player_roll" and on_player_roll:
|
|
dice = args.get("dice", "1d6")
|
|
reason = args.get("reason", "a check")
|
|
roll_val = on_player_roll(dice, reason)
|
|
result = f"Player rolled {dice} for '{reason}': {roll_val}"
|
|
else:
|
|
result = execute_tool(name, args)
|
|
|
|
if name not in ("narrative", "finalize_turn", "player_roll"):
|
|
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
|
|
|
|
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,
|
|
)
|
|
|
|
return TurnResult(
|
|
book_log=book_log,
|
|
log_entry=log_entry,
|
|
ambience=ambience,
|
|
debug_info="; ".join(errors) if errors else "",
|
|
changes=changes,
|
|
)
|
|
|
|
|
|
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()
|