223 lines
7.9 KiB
Python
223 lines
7.9 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_action
|
|
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,
|
|
last_prompt: str | None = None,
|
|
on_thought: callable = None,
|
|
on_action: callable = None,
|
|
on_player_roll: callable = None,
|
|
on_debug: 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}")
|
|
elif last_prompt:
|
|
state.append_llm_log(f"Resume from: {last_prompt[:120]}")
|
|
|
|
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(f"LLM: {model} | temp={lm.get('temperature')}")
|
|
if on_debug:
|
|
on_debug("config", {"model": model, "temperature": lm.get("temperature"), "max_tokens": lm.get("max_tokens"), "strategy": "tools"})
|
|
|
|
if player_action:
|
|
story = state.read_recent_book()
|
|
log = state.read_recent_log()
|
|
valid, reason = validate_action(player_action, story=story, log=log, on_debug=on_debug)
|
|
if valid:
|
|
state.append_llm_log(f"\n[VALIDATION PASSED] {reason}")
|
|
else:
|
|
state.append_llm_log(f"\n[VALIDATION REJECTED] {reason}")
|
|
return TurnResult(
|
|
book_log="",
|
|
log_entry=f"You can't do that — {reason}.",
|
|
user_prompt=f"*Your action:\n\t\"{player_action}\"\nwas rejected:\n\t{reason}*",
|
|
)
|
|
|
|
system = build_system_prompt()
|
|
parts = []
|
|
if last_prompt:
|
|
parts.append(f"## Situation\n{last_prompt}")
|
|
if player_action:
|
|
parts.append(f"## Player's Request\n{player_action}")
|
|
if not player_action and not last_prompt:
|
|
parts.append(
|
|
"## Instructions\n"
|
|
"This is a new story. Welcome the player and guide them through the game setup."
|
|
)
|
|
else:
|
|
parts.append(
|
|
"## Instructions\n"
|
|
"Advance the story based on the player's request. "
|
|
"All state is shown above — write the outcome directly."
|
|
)
|
|
parts.append(f"\n*A die is cast: **{die_roll}** (1d6).*")
|
|
user = "\n\n".join(parts)
|
|
|
|
start_time = datetime.now()
|
|
state.append_llm_log(f"\n[TOOL] Single call — {len(system)} chars system, {len(user)} chars user")
|
|
|
|
text = call_llm(
|
|
[{"role": "system", "content": system}, {"role": "user", "content": user}],
|
|
label="Turn generation",
|
|
on_debug=on_debug,
|
|
)
|
|
|
|
if not text or not text.strip():
|
|
return TurnResult(error="LLM returned empty response")
|
|
|
|
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, on_debug=on_debug)
|
|
if not tool_calls:
|
|
state.append_llm_log("\n[TOOL] no tool blocks found")
|
|
|
|
book_log = ""
|
|
log_entry = None
|
|
user_prompt = ""
|
|
ambience = None
|
|
changes: list[str] = []
|
|
errors: list[str] = []
|
|
|
|
extr_start = datetime.now()
|
|
|
|
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"]
|
|
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 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")
|
|
if on_debug:
|
|
applied = len([tc for tc in tool_calls if tc.get("tool") not in ("finalize_turn", "narrative")])
|
|
on_debug("phase_done", {
|
|
"book_log_chars": len(book_log),
|
|
"log_entry": log_entry,
|
|
"user_prompt": user_prompt,
|
|
"ambience": ambience,
|
|
"extract_errors": errors or None,
|
|
"total_elapsed_ms": total_elapsed,
|
|
"tool_calls_count": len(tool_calls),
|
|
"applied_changes_count": applied,
|
|
"tool_call_results": tool_calls,
|
|
})
|
|
|
|
log_turn_details(
|
|
player_action=player_action or last_prompt or "",
|
|
last_prompt=last_prompt 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,
|
|
on_debug=on_debug,
|
|
)
|
|
|
|
return TurnResult(
|
|
book_log=book_log,
|
|
log_entry=log_entry,
|
|
user_prompt=user_prompt,
|
|
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")
|
|
parser.add_argument("--last", "-l", help="Last narrative text")
|
|
args = parser.parse_args()
|
|
|
|
engine = GameEngine()
|
|
result = engine.generate_turn(
|
|
player_action=args.action,
|
|
last_prompt=args.last,
|
|
)
|
|
|
|
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()
|