dejvino/splinter-keep
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

More refactors

Browse Source
This commit is contained in:
Dejvino
2026-06-30 20:03:53 +02:00
parent e9a1187f34
commit 545d3bcac0
20 changed files with 1633 additions and 1686 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/engine_lib/__init__.py
View File
tools/engine_lib/config.py
+74
View File
@@ -0,0 +1,74 @@
#!/usr/bin/env python3
"""
config.py — LLM configuration loading and accessors for The Chaos engine.
"""
from __future__ import annotations
import json
import sys
from pathlib import Path
from .paths import CONFIG_PATH
DEFAULT_CONFIG: dict = {
"llm": {
"model": "ollama/llama3.1",
"api_key": None,
"api_base": None,
"temperature": 0.8,
"max_tokens": 300,
}
}
def load_config(path: Path = CONFIG_PATH) -> dict:
"""Load config from path, creating default if missing. Returns config dict."""
if not path.exists():
print(
"No session/config.json found. Creating default.\n"
"Edit the model field (e.g. 'ollama/llama3.1', 'openai/gpt-4', "
"'anthropic/claude-sonnet-4-20250514') and set api_key if needed.",
file=sys.stderr,
)
cfg = dict(DEFAULT_CONFIG)
save_config(cfg, path)
return cfg
raw = path.read_text()
cfg = json.loads(raw)
llm = cfg.get("llm", {})
if not llm.get("api_key"):
llm["api_key"] = None
return cfg
def save_config(config: dict, path: Path = CONFIG_PATH) -> None:
"""Save config dict to path."""
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(config, indent=2) + "\n")
# ── Accessors ──────────────────────────────────────────────────────────────
def get_model(config: dict) -> str:
return config.get("llm", {}).get("model", "ollama/llama3.1")
def get_api_key(config: dict) -> str | None:
return config.get("llm", {}).get("api_key")
def get_api_base(config: dict) -> str | None:
return config.get("llm", {}).get("api_base")
def get_temperature(config: dict) -> float:
return config.get("llm", {}).get("temperature", 0.8)
def get_max_tokens(config: dict) -> int:
return config.get("llm", {}).get("max_tokens", 512)
def get_timeout(config: dict) -> int:
return config.get("llm", {}).get("timeout", 120)
tools/engine_lib/context.py
+74
View File
@@ -0,0 +1,74 @@
#!/usr/bin/env python3
"""
context.py — System prompt and user message assembly for The Chaos engine.
All functions are standalone — no dependency on GameEngine.
"""
from __future__ import annotations
from .paths import CHAR_PATH, WORLD_PATH, BOOK_PATH
from .prompts import SYSTEM_PROMPT
from . import state
def build_system_prompt() -> str:
"""Assemble the system prompt with current game state."""
char = state.read_file(CHAR_PATH) or "*No character sheet.*"
world = state.truncate_world(state.read_file(WORLD_PATH) or "") or "*No world state.*"
log = state.read_recent_log()
story = state.read_recent_book()
return SYSTEM_PROMPT.substitute(
character=char, world=world, log=log, story=story
)
def build_prose_prompt() -> str:
"""Assemble the prose-generation prompt with current game state."""
from .prompts import PROSE_PROMPT
char = state.read_file(CHAR_PATH) or "*No character sheet.*"
world = state.truncate_world(state.read_file(WORLD_PATH) or "") or "*No world state.*"
log = state.read_recent_log()
story = state.read_recent_book()
return PROSE_PROMPT.substitute(
character=char, world=world, log=log, story=story
)
def build_user_message(
player_action: str | None = None,
last_prompt: str | None = None,
**kwargs: str | None,
) -> str:
"""Build the user message for this turn's LLM call."""
if kwargs:
raise TypeError(
f"build_user_message() got unexpected keyword arguments: "
f"{set(kwargs)}. Did you mean 'last_prompt' instead of one of these?"
)
parts = []
if last_prompt:
parts.append(f"## Situation\n{last_prompt}")
if player_action:
parts.append(f"## Player's Request\n{player_action}")
has_existing_story = bool(
state.read_file(BOOK_PATH).strip()
) if not last_prompt else True
if not player_action and not last_prompt:
if has_existing_story:
raise RuntimeError("User action is required for every turn.")
else:
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."
)
return "\n\n".join(parts)
tools/engine_lib/llm.py
+63
View File
@@ -0,0 +1,63 @@
#!/usr/bin/env python3
"""
llm.py — LLM interaction layer for The Chaos engine.
Provides the low-level call_llm function and environment variable setup
for provider-specific auth.
"""
from __future__ import annotations
import os
from .state import append_llm_log
def set_llm_env(model: str, api_key: str | None, api_base: str | None) -> None:
"""Set provider-specific env vars for litellm."""
prefix = model.split("/")[0].upper()
key = api_key or "sk-placeholder"
os.environ[f"{prefix}_API_KEY"] = key
if api_base:
os.environ[f"{prefix}_API_BASE"] = api_base
def call_llm(
messages: list[dict],
*,
model: str,
temperature: float,
timeout: int,
max_tokens: int,
label: str = "",
on_debug: callable = None,
) -> str | None:
"""Make a single LLM call. Returns content text or None on error."""
try:
import litellm
except ImportError:
if on_debug:
on_debug("llm_error", {"label": label, "error": "litellm not installed"})
return None
try:
response = litellm.completion(
model=model,
messages=messages,
temperature=temperature,
stream=False,
timeout=timeout,
max_tokens=max_tokens,
)
content = getattr(response.choices[0].message, 'content', None) or ""
reasoning = getattr(response.choices[0].message, 'reasoning_content', None) or ""
if reasoning and reasoning not in content:
append_llm_log(f"\n--- {label} [reasoning] ---\n{reasoning}")
text = content or reasoning
append_llm_log(f"\n--- {label} ---\n{text}")
return text
except Exception as e:
err_msg = f"{type(e).__name__}: {e}"
append_llm_log(f"\n--- LLM ERROR ({label}) ---\n{err_msg}")
if on_debug:
on_debug("llm_error", {"label": label, "error": err_msg})
return None
tools/engine_lib/models.py
+35
View File
@@ -0,0 +1,35 @@
#!/usr/bin/env python3
"""
models.py — Data classes for The Chaos game engine.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class GenerationResult:
"""Legacy result — kept for backward compat with CLI main()."""
narrative: str
choices: list[str] = field(default_factory=list)
log_entry: Optional[str] = None
ambience: Optional[str] = None
character_updates: Optional[str] = None
world_updates: Optional[str] = None
journal_add: list[str] = field(default_factory=list)
journal_done: list[str] = field(default_factory=list)
error: Optional[str] = None
@dataclass
class TurnResult:
"""Output of a complete turn via finalize_turn tool."""
book_log: str = ""
user_prompt: str = ""
ambience: Optional[str] = None
log_entry: Optional[str] = None
error: Optional[str] = None
debug_info: str = ""
changes: list[str] = field(default_factory=list)
tools/engine_lib/parsing.py
+121
View File
@@ -0,0 +1,121 @@
#!/usr/bin/env python3
"""
parsing.py — LLM response parsing and turn logging for The Chaos engine.
Standalone functions — no dependency on GameEngine.
"""
from __future__ import annotations
import json
import re
from datetime import datetime
from typing import Optional
from .models import GenerationResult
from . import state
def parse_response(text: str) -> GenerationResult:
"""
Parse a full LLM response into a GenerationResult.
Extracts the JSON block and splits narrative from it.
"""
if text.startswith('{"error":'):
try:
err = json.loads(text).get("error", "Unknown error")
except json.JSONDecodeError:
err = "Unknown error"
return GenerationResult(narrative="", error=err)
json_pattern = r"```json\s*\n?(.*?)\n?```"
matches = re.findall(json_pattern, text, re.DOTALL)
narrative = text
data = {}
if matches:
json_str = matches[-1].strip()
narrative = text[: text.rfind("```json")]
narrative_lines = []
for line in narrative.splitlines():
if not line.lstrip().startswith('book_log:'):
narrative_lines.append(line)
narrative = "\n".join(narrative_lines).strip()
try:
data = json.loads(json_str)
except json.JSONDecodeError:
pass
else:
text_stripped = text.strip()
if text_stripped.startswith("{") and text_stripped.endswith("}"):
try:
data = json.loads(text_stripped)
narrative = data.get("narrative", "")
except json.JSONDecodeError:
pass
return GenerationResult(
narrative=narrative or text,
choices=data.get("choices", []),
log_entry=data.get("log_entry"),
ambience=data.get("ambience"),
character_updates=data.get("character_updates"),
world_updates=data.get("world_updates"),
journal_add=data.get("journal_add", []),
journal_done=data.get("journal_done", []),
)
def log_turn_details(
player_action: str,
last_prompt: str,
strategy_name: str,
die_roll: int,
model: str,
temperature: float,
max_tokens: int,
book_log: str,
log_entry: str,
ambience: Optional[str],
tool_calls: list,
on_debug=None,
) -> None:
"""Write structured turn summary to llm.log and fire TUI debug event."""
ts = datetime.now().isoformat()
output_chars = len(book_log)
output_words = len(book_log.split()) if book_log else 0
applied = len([tc for tc in tool_calls if tc.get("tool") != "finalize_turn"])
state.append_llm_log("")
state.append_llm_log(f"┌─ Turn Details — {ts}")
state.append_llm_log(f"├─ Input: {player_action}")
state.append_llm_log(f"├─ Last Prompt: {last_prompt}")
state.append_llm_log(f"├─ Strategy: {strategy_name}")
state.append_llm_log(f"├─ Dice: {die_roll} (1d6)")
state.append_llm_log(f"├─ Model: {model} | Temp: {temperature} | Tokens: {max_tokens}")
state.append_llm_log(f"├─ Output: {output_chars} chars ({output_words} words)")
state.append_llm_log(f"├─ Log Entry: {log_entry}")
state.append_llm_log(f"├─ Ambience: {ambience or 'None'}")
tools_preview = ", ".join(tc.get("tool", "?") for tc in tool_calls)
state.append_llm_log(f"├─ Tool Calls: {len(tool_calls)} ({tools_preview})")
state.append_llm_log(
"└─────────────────────────────────────────────────────────────────────────────────────────┘"
)
if on_debug:
on_debug("turn_details", {
"timestamp": ts,
"model": model,
"temperature": temperature,
"max_tokens": max_tokens,
"strategy_name": strategy_name,
"die_roll": die_roll,
"player_action": player_action,
"book_log_chars": output_chars,
"book_log_words": output_words,
"ambience": ambience,
"tool_calls_count": len(tool_calls),
"applied_changes_count": applied,
"tool_call_results": tool_calls,
})
tools/engine_lib/paths.py
+25
View File
@@ -0,0 +1,25 @@
#!/usr/bin/env python3
"""
paths.py — Path constants for The Chaos game engine.
"""
from __future__ import annotations
from datetime import date
from pathlib import Path
BASE_DIR = Path(__file__).resolve().parent.parent
SESSION_DIR = BASE_DIR / 'session'
CONFIG_PATH = SESSION_DIR / 'config.json'
CHAR_PATH = SESSION_DIR / 'character.md'
WORLD_PATH = SESSION_DIR / 'world.md'
BOOK_PATH = SESSION_DIR / 'book.md'
JOURNAL_PATH = SESSION_DIR / 'journal.md'
AMBIENCE_PATH = SESSION_DIR / 'ambience.md'
LOG_DIR = SESSION_DIR / 'log'
LLM_LOG_PATH = SESSION_DIR / 'llm.log'
AMBIENCE_OPTIONS_PATH = SESSION_DIR / "ambience_options.md"
CHANGES_PATH = SESSION_DIR / "changes.md"
AUDIO_DIR = SESSION_DIR / "audio"
TODAY = date.today().isoformat()
tools/engine_lib/prompts.py
+86
View File
@@ -0,0 +1,86 @@
#!/usr/bin/env python3
"""
prompts.py — System prompt templates for The Chaos game engine.
"""
from __future__ import annotations
from string import Template
SYSTEM_PROMPT = Template("""You are the DM for "The Chaos". Narrate in 2nd person ("You"), vivid but concise. Player: Dillion.
## Rules
- **Odds**: 1d6, 4+ favourable, 3- trouble.
- **Traits**: 3d6, roll UNDER trait.
- **Combat**: 1d6, 4+ hits. Damage: 1d6 + mod - armour.
- **Wounds at 0 HP**: 1d6 → 1-2 die, 3-4 -1 max HP, 5-6 -1 all rolls until healed.
- **Modifiers**: Favourable +1, Risky -1, Desperate -2.
## Tools (action only)
Wrap in ```tool to perform an action:
```
{"tool": "roll", "args": {"dice": "1d6"}}
```
- **roll** — dice, modifier
- **player_roll** — dice, reason
- **character_update** — content: "full sheet" (if HP/cash/gear/stats change)
- **world_update** — content: "full world" (if NPCs/locations/threads change)
- **journal_update** — add: [...], done: [...]
You have the full state above — no need to look anything up. Just write the story and use tools when the player's action changes something. If a player action is impossible (e.g. they try to use an item they don't have), narrate the failure and DO NOT use any state-changing tools.
## State
### Character
$character
### World
$world
### Log
$log
### Story
$story""")
PROSE_PROMPT = Template("""You are the DM for "The Chaos". Narrate in 2nd person ("You"), vivid but concise. Player: Dillion.
## Rules
- **Odds**: 1d6, 4+ favourable, 3- trouble.
- **Traits**: 3d6, roll UNDER trait.
- **Combat**: 1d6, 4+ hits. Damage: 1d6 + mod - armour.
- **Wounds at 0 HP**: 1d6 → 1-2 die, 3-4 -1 max HP, 5-6 -1 all rolls until healed.
- **Modifiers**: Favourable +1, Risky -1, Desperate -2.
A die is cast at the start of each turn — incorporate it into your narrative.
End your response with a `### Changes` block listing what changed:
### Changes
- Current Health: 3
- Cash: 45 silver
- Added to inventory: Silver key
- Removed from inventory: Torches (10)
- Replaced gear: Mace (1d6+1) → Mace (1d6+2)
- Note: Found a hidden passage
- Journal done: Defeat the demon
- Journal add: Investigate the mine
Only include lines for things that actually changed. Omit unused lines entirely.
## State
### Character
$character
### World
$world
### Log
$log
### Story
$story""")
tools/engine_lib/state.py
+210
View File
@@ -0,0 +1,210 @@
#!/usr/bin/env python3
"""
state.py — Game state persistence and file I/O for The Chaos engine.
Standalone functions that read/write session files. No dependency on
GameEngine or other modules besides paths.
"""
from __future__ import annotations
import re
import sys
from datetime import date, datetime, timedelta
from pathlib import Path
from .paths import (
CHAR_PATH, WORLD_PATH, BOOK_PATH, JOURNAL_PATH, AMBIENCE_PATH,
LOG_DIR, LLM_LOG_PATH, AMBIENCE_OPTIONS_PATH, CHANGES_PATH,
AUDIO_DIR, TODAY,
)
from .models import TurnResult
def read_file(path: Path) -> str:
"""Read a text file, return empty string if missing."""
return path.read_text().strip() if path.exists() else ""
def read_recent_log(max_entries: int = 5) -> str:
"""Read the latest log file and return the last N entries."""
log_path = LOG_DIR / f"{TODAY}.md"
if not log_path.exists():
yesterday = (date.today() - timedelta(days=1)).isoformat()
log_path = LOG_DIR / f"{yesterday}.md"
if not log_path.exists():
return "*No recent events.*"
lines = log_path.read_text().splitlines()
entries = [l for l in lines if l.strip().startswith("- ")]
return "\n".join(entries[-max_entries:]) or "*No recent events.*"
def read_recent_book(max_turns: int = 1) -> str:
"""Return the last N turns from the book as context."""
text = read_file(BOOK_PATH)
if not text:
return "*No prior story.*"
turns = text.split("\n## ")
recent = turns[-max_turns:]
return "\n## ".join(recent) if len(turns) > 1 else recent[0]
def truncate_world(text: str) -> str:
"""Extract key world context: NPCs, factions, active threads, rumours."""
if not text or text == "*No world state.*":
return text
sections = re.split(r"\n(?=## |### )", text)
parts = []
for sec in sections:
header = sec.split("\n")[0].strip() if sec else ""
if "Active Threads" in header:
parts.append(sec)
elif "Notable NPCs" in header or "Factions at Play" in header or "### Rumours" in header:
parts.append(sec)
result = "\n\n".join(parts)
return result or text[:1500] + "\n_(world truncated)_"
def get_valid_ambiences() -> set[str]:
"""Parse ambience_options.md and return set of valid ambience names."""
valid = {"silence"}
if not AMBIENCE_OPTIONS_PATH.exists():
return valid
in_table = False
for line in AMBIENCE_OPTIONS_PATH.read_text().splitlines():
s = line.strip()
if not s.startswith("|") or not s.endswith("|"):
in_table = False
continue
if in_table and all(c in "-:| " for c in s):
continue
parts = [p.strip() for p in s.split("|") if p.strip()]
if not parts:
continue
if not in_table:
in_table = True
continue
name = parts[0].lower()
files_str = parts[1] if len(parts) > 1 else ""
files = [f.strip() for f in files_str.split(",")]
has_files = any((AUDIO_DIR / f).exists() or f for f in files)
if has_files:
valid.add(name)
return valid
def validate_update_size(name: str, new_content: str, path: Path) -> bool:
"""Reject updates more than 30% shorter than existing — likely partial paste."""
if not path.exists():
return True
old = path.read_text().strip()
if not old:
return True
ratio = len(new_content) / len(old)
if ratio < 0.7:
print(
f"WARNING: {name} update rejected ({ratio:.0%} of original size "
f"= {len(new_content)} vs {len(old)} chars) — likely a partial paste.",
file=sys.stderr,
)
return False
return True
def apply_state(result: TurnResult) -> None:
"""Write state changes from a TurnResult to disk."""
if result.ambience:
AMBIENCE_PATH.write_text(result.ambience.strip().lower() + "\n")
if result.changes:
CHANGES_PATH.write_text("\n".join(result.changes) + "\n")
else:
CHANGES_PATH.write_text("")
def archive_turn(narrative: str) -> None:
"""Append the narrative as a new turn in book.md."""
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M")
heading = f"\n\n## Turn — {timestamp}\n\n"
BOOK_PATH.parent.mkdir(parents=True, exist_ok=True)
with open(BOOK_PATH, "a") as f:
f.write(heading + narrative.strip() + "\n")
def append_log(entry: str) -> None:
"""Append a log entry to today's log file."""
LOG_DIR.mkdir(parents=True, exist_ok=True)
log_path = LOG_DIR / f"{TODAY}.md"
if not log_path.exists():
log_path.write_text(f"# Session Log — {TODAY}\n\n")
with open(log_path, "a") as f:
f.write(entry.strip() + "\n")
def append_llm_log(text: str) -> None:
"""Append raw LLM activity to llm.log for debugging."""
LLM_LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
with open(LLM_LOG_PATH, "a") as f:
f.write(text + "\n")
def update_journal(add: list[str] | None = None, done: list[str] | None = None) -> None:
"""Add or complete TODO items in journal.md."""
if not JOURNAL_PATH.exists():
JOURNAL_PATH.write_text("# Journal\n\n## TODO\n\n## DONE\n\n")
lines = JOURNAL_PATH.read_text().splitlines()
todo_items: list[str] = []
done_items: list[str] = []
before_todo: list[str] = []
between: list[str] = []
after_done: list[str] = []
section = "before_todo"
for line in lines:
stripped = line.strip()
if stripped == "## TODO":
section = "todo"
before_todo.append(line)
elif stripped == "## DONE":
section = "done"
between.append(line)
elif section == "before_todo":
before_todo.append(line)
elif section == "todo":
if stripped.startswith("- "):
todo_items.append(stripped[2:])
else:
between.append(line)
elif section == "done":
if stripped.startswith("- "):
done_items.append(stripped[2:])
else:
after_done.append(line)
if done:
done_set = set(done)
todo_items = [i for i in todo_items if i not in done_set]
new_done = [i for i in done if i not in done_items]
done_items.extend(new_done)
if add:
todo_set = set(todo_items)
new_todo = [i for i in add if i not in todo_set]
todo_items = new_todo + todo_items
out = list(before_todo)
for item in todo_items:
out.append(f"- {item}")
out.extend(between)
for item in done_items:
out.append(f"- {item}")
out.extend(after_done)
cleaned = []
prev_blank = False
for line in out:
is_blank = line.strip() == ""
if is_blank and prev_blank:
continue
cleaned.append(line)
prev_blank = is_blank
JOURNAL_PATH.write_text("\n".join(cleaned) + "\n")
tools/engine_lib/strategies.py
+657
View File
@@ -0,0 +1,657 @@
#!/usr/bin/env python3
"""
strategies.py — Generation strategies for The Chaos engine.
Contains the three-phase conversational approach and the single-call
tool-based approach. All functions are standalone — no dependency on
GameEngine (config values and callbacks are passed explicitly).
"""
from __future__ import annotations
import json
import random
import re
from datetime import datetime
from typing import Iterator
from .models import GenerationResult, TurnResult
from .prompts import PROSE_PROMPT
from .llm import set_llm_env, call_llm
from .tools_handler import (
execute_tool, describe_tool_action, describe_change,
parse_changes_block, extract_tool_calls,
)
from .context import build_system_prompt, build_user_message, build_prose_prompt
from .validation import auto_prompt, validate_narrative
from .parsing import parse_response, log_turn_details
from . import state
# ── Synchronous (legacy) ───────────────────────────────────────────────────
def generate(
player_action: str | None = None,
last_narrative: str | None = None,
*,
model: str,
temperature: float,
timeout: int,
max_tokens: int,
api_key: str | None = None,
api_base: str | None = None,
) -> GenerationResult:
"""
Synchronous generation. Calls the LLM, parses the response,
and returns a GenerationResult.
"""
system = build_system_prompt()
user = build_user_message(
player_action=player_action, last_prompt=last_narrative
)
messages = [
{"role": "system", "content": system},
{"role": "user", "content": user},
]
try:
import litellm
except ImportError:
return GenerationResult(
narrative="",
error="litellm is not installed. Run: pip install litellm",
)
set_llm_env(model, api_key, api_base)
try:
response = litellm.completion(
model=model,
messages=messages,
temperature=temperature,
stream=False,
timeout=timeout,
)
text = response.choices[0].message.content or ""
except Exception as e:
return GenerationResult(
narrative="",
error=f"LLM call failed: {e}",
)
return parse_response(text)
# ── Streaming (legacy) ─────────────────────────────────────────────────────
def generate_stream(
player_action: str | None = None,
last_narrative: str | None = None,
*,
model: str,
temperature: float,
timeout: int,
max_tokens: int,
api_key: str | None = None,
api_base: str | None = None,
) -> Iterator[str]:
"""
Streaming generator. Yields text chunks as they arrive from the LLM.
On completion, the final yield is the FULL text (for parsing).
"""
system = build_system_prompt()
user = build_user_message(
player_action=player_action, last_prompt=last_narrative
)
messages = [
{"role": "system", "content": system},
{"role": "user", "content": user},
]
try:
import litellm
except ImportError:
yield json.dumps({
"error": "litellm is not installed. Run: pip install litellm"
})
return
set_llm_env(model, api_key, api_base)
try:
response = litellm.completion(
model=model,
messages=messages,
temperature=temperature,
stream=True,
timeout=timeout,
)
full_text = ""
for chunk in response:
delta = chunk.choices[0].delta.content or ""
if delta:
full_text += delta
yield full_text
yield full_text
except Exception as e:
yield json.dumps({"error": f"LLM call failed: {e}"})
# ── Three-phase generation ─────────────────────────────────────────────────
def generate_with_tools(
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,
*,
model: str,
temperature: float,
timeout: int,
max_tokens: int,
api_key: str | None = None,
api_base: str | None = None,
) -> TurnResult:
"""
Three-phase generation:
1. **Prose** — LLM writes the full book_log from context + player action.
2. **Summarize** — LLM condenses the book_log into one log line.
3. **Extract** — LLM reads the book_log and outputs tool calls for state changes.
"""
set_llm_env(model, api_key, api_base)
datetime_now = datetime.now()
state.append_llm_log(f"\n{'='*60}")
state.append_llm_log(f"=== Turn — {datetime_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)")
book_log = None
changes_block = ""
log_entry = None
user_prompt = auto_prompt("")
ambience = None
debug_info = ""
changes: list[str] = []
for outer_attempt in range(3):
# ── Phase 1: Prose ────────────────────────────────────────────────
if on_action:
on_action(f"Phase 1/3: writing story (dice={die_roll})")
if on_debug:
on_debug("phase", {"phase": 1, "name": "prose", "status": "start", "dice": die_roll, "outer_attempt": outer_attempt + 1})
system = build_prose_prompt()
user = build_user_message(
player_action=player_action,
last_prompt=last_prompt,
)
user += f"\n\n*A die is cast: **{die_roll}** (1d6).*"
text = call_llm([
{"role": "system", "content": system},
{"role": "user", "content": user},
], model=model, temperature=temperature, timeout=timeout,
max_tokens=1024, label=f"Prose attempt {outer_attempt + 1}", on_debug=on_debug)
if not text or not text.strip():
if on_debug:
on_debug("phase", {"phase": 1, "status": "empty", "attempt": outer_attempt + 1})
continue
raw = text.strip()
changes_block = ""
if "### Changes" in raw:
parts = raw.split("### Changes", 1)
book_log = parts[0].strip()
changes_block = "### Changes" + parts[1]
else:
book_log = raw
if on_debug:
preview = book_log[:150].replace("\n", "\\n")
on_debug("phase", {"phase": 1, "status": "done", "chars": len(book_log), "changes": bool(changes_block), "preview": preview})
# ── Validation ────────────────────────────────────────────────────
if on_debug:
on_debug("phase", {"phase": 1, "name": "validation", "status": "start"})
valid, reason = validate_narrative(book_log, model=model, temperature=temperature, timeout=timeout, on_debug=on_debug)
if not valid:
if on_debug:
on_debug("phase", {"phase": 1, "status": "validation_failed", "reason": reason, "outer_attempt": outer_attempt + 1})
book_log = None
continue
# ── Phase 2: Summarize ────────────────────────────────────────────
if on_action:
on_action("Phase 2/3: summarizing story")
if on_debug:
on_debug("phase", {"phase": 2, "name": "summarize", "status": "start"})
log_context = state.read_recent_log()
log_entry = None
for p2_attempt in range(2):
context = book_log
if changes_block:
context += f"\n\n{changes_block}"
text = call_llm([
{"role": "user", "content":
f"Given the session log so far, summarize the new story in one line. "
f"Focus on who was involved (character and NPC names):\n\n"
f"## Session Log\n{log_context}\n\n"
f"## New Story\n{context}"}
], model=model, temperature=temperature, timeout=timeout,
max_tokens=max_tokens, label=f"Summarize attempt {p2_attempt + 1}", on_debug=on_debug)
if text and text.strip():
log_entry = text.strip().split("\n")[0][:300]
if on_debug:
on_debug("phase", {"phase": 2, "status": "done", "summary": log_entry})
break
if not log_entry:
log_entry = book_log.split("\n")[0][:120]
if on_debug:
on_debug("phase", {"phase": 2, "status": "fallback", "summary": log_entry})
# ── Phase 3: Extract state changes ────────────────────────────────
if on_action:
on_action("Phase 3/3: extracting state changes")
if on_debug:
on_debug("phase", {"phase": 3, "name": "extract", "status": "start"})
user_prompt = auto_prompt(book_log)
ambience = None
phase3_errors = []
changes = []
# Step 1: Parse ### Changes block directly
if changes_block.strip():
for tc in parse_changes_block(changes_block):
name = tc["tool"]
args = tc.get("args", {})
if name == "finalize_turn":
continue
result = execute_tool(name, args)
if result.startswith("**Error:") or result.startswith("Tool error") or result.startswith("Unknown"):
phase3_errors.append(f"{name}: {result}")
else:
desc = describe_change(name, args)
if desc:
changes.append(desc)
# Step 2: LLM Phase 3 for finalize_turn + any extra changes
previous_attempt = None
phase3_ok = False
for p3_attempt in range(5):
from paths import CHAR_PATH, WORLD_PATH
current_char = state.read_file(CHAR_PATH) or "*No character.*"
current_world = state.truncate_world(state.read_file(WORLD_PATH) or "") or "*No world.*"
phase3_prompt = (
f"## Current Character\n{current_char}\n\n"
f"## Current World\n{current_world}\n\n"
f"## Story\n{book_log}\n\n"
)
if changes_block.strip():
phase3_prompt += (
f"## Changes already applied\n{changes_block}\n\n"
f"Output the finalize_turn tool to end the turn. "
f"Add extra tool calls if you spot changes the list above missed.\n\n"
)
else:
phase3_prompt += (
f"Read the story and compare with current state. Output tool calls for any changes:\n\n"
)
phase3_prompt += (
f"Output ```tool blocks for changes only. Examples:\n\n"
)
if previous_attempt:
phase3_prompt += (
f"--- PREVIOUS ATTEMPT (had errors) ---\n"
f"{previous_attempt['output']}\n\n"
f"--- FEEDBACK ---\n"
f"{previous_attempt['feedback']}\n\n"
f"Fix the issues above. Output corrected tool calls only.\n\n"
)
text = call_llm([
{"role": "user", "content": phase3_prompt +
f"```tool\n{{\"tool\": \"modify_vitals\", \"args\": {{\"current_hp\": 5, \"cash\": 45}}}}\n```\n"
f"```tool\n{{\"tool\": \"modify_traits\", \"args\": {{\"dex\": 15}}}}\n```\n"
f"```tool\n{{\"tool\": \"add_to_inventory\", \"args\": {{\"item\": \"Silver key\"}}}}\n```\n"
f"```tool\n{{\"tool\": \"remove_from_inventory\", \"args\": {{\"item\": \"Torches (10)\"}}}}\n```\n"
f"```tool\n{{\"tool\": \"replace_gear\", \"args\": {{\"before\": \"Mace (1d6+1)\", \"after\": \"Mace (1d6+2, sharpened)\"}}}}\n```\n"
f"```tool\n{{\"tool\": \"add_note\", \"args\": {{\"note\": \"Found a hidden passage under the temple\"}}}}\n```\n"
f"```tool\n{{\"tool\": \"replace_note\", \"args\": {{\"before\": \"Old note text\", \"after\": \"New note text\"}}}}\n```\n"
f"```tool\n{{\"tool\": \"world_update\", \"args\": {{\"content\": \"# The World\\n\\n...full new world state...\"}}}}\n```\n"
f"```tool\n{{\"tool\": \"journal_update\", \"args\": {{\"add\": [\"Investigate the mine\"], \"done\": [\"Defeat the demon\"]}}}}\n```\n"
f"```tool\n{{\"tool\": \"finalize_turn\", \"args\": {{\"user_prompt\": \"What do you do?\", \"ambience\": \"dungeon\"}}}}\n```\n\n"
f"Only output tools for things that actually changed. Omit unchanged fields."}
], model=model, temperature=temperature, timeout=timeout,
max_tokens=max_tokens, label=f"Extract attempt {p3_attempt + 1}", on_debug=on_debug)
if not text or not text.strip():
if on_debug:
on_debug("phase", {"phase": 3, "status": "empty", "attempt": p3_attempt + 1})
continue
tool_calls_list = extract_tool_calls(
text, round_num=p3_attempt + 1, on_debug=on_debug
)
if on_debug and tool_calls_list:
names = [tc.get("tool", "?") for tc in tool_calls_list if tc.get("tool") != "finalize_turn"]
fin = any(tc.get("tool") == "finalize_turn" for tc in tool_calls_list)
on_debug("phase", {"phase": 3, "status": "tools_found", "tools": names, "has_finalize": fin})
errors: list[str] = []
attempt_changes: list[str] = []
for tc in tool_calls_list:
name = tc.get("tool", "?")
args = tc.get("args", {})
if name == "finalize_turn":
if args.get("user_prompt"):
user_prompt = args["user_prompt"]
if args.get("ambience"):
ambience = args["ambience"]
continue
if on_action:
on_action(f"State: {describe_tool_action(name, args)}")
if on_debug:
on_debug("tool_call", {"round": p3_attempt + 1, "tool": name, "args": args})
if 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 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:
attempt_changes.append(desc)
if on_debug:
on_debug("tool_result", {"round": p3_attempt + 1, "tool": name, "result": result})
if not errors:
phase3_ok = True
debug_info = ""
changes.extend(attempt_changes)
if on_debug:
on_debug("phase", {"phase": 3, "status": "done", "applied": len([tc for tc in tool_calls_list if tc.get("tool") != "finalize_turn"])})
break
phase3_errors = errors
debug_info = "; ".join(errors)
if on_debug:
on_debug("phase", {"phase": 3, "status": "errors", "errors": errors, "attempt": p3_attempt + 1})
feedback_lines = ["The previous tool calls had errors:"]
for e in errors:
feedback_lines.append(f"- {e}")
feedback_lines.append("")
feedback_lines.append("Fix ALL issues above. Use correct tool names, valid JSON, and reasonable values.")
previous_attempt = {"output": text, "feedback": "\n".join(feedback_lines)}
if phase3_ok:
break
if on_debug:
on_debug("phase", {"phase": 3, "status": "exhausted", "errors": phase3_errors})
on_debug("phase", {"phase": 1, "status": "retry_after_phase3_failure", "outer_attempt": outer_attempt + 1})
book_log = None
if not book_log:
return TurnResult(error="Generation failed after exhausting all retries")
if on_action:
on_action("Turn complete")
if on_debug:
on_debug("phase_done", {
"book_log_chars": len(book_log),
"log_entry": log_entry,
"user_prompt": user_prompt,
"ambience": ambience,
"extract_errors": debug_info or None,
})
state.append_llm_log(
f"\n--- FINAL ---\n"
f"book_log: {book_log[:200]}\n"
f"log_entry: {log_entry}\n"
f"user_prompt: {user_prompt}\n"
f"ambience: {ambience}\n"
)
return TurnResult(
book_log=book_log,
log_entry=log_entry,
user_prompt=user_prompt,
ambience=ambience,
debug_info=debug_info,
changes=changes,
)
# ── Single-call generation ─────────────────────────────────────────────────
def generate_with_tools_single(
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,
*,
model: str,
temperature: float,
timeout: int,
max_tokens: int,
api_key: str | None = None,
api_base: str | None = None,
) -> TurnResult:
"""
Single-call generation using tools.
Uses a single LLM call with all tools available — LLM outputs
narrative + tool blocks in one go. No retry loop.
"""
datetime_now = datetime.now()
state.append_llm_log(f"\n{'='*60}")
state.append_llm_log(f"=== Turn — {datetime_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]}")
strategy_name = "tools"
if on_action:
on_action(f"LLM: {model} | temp={temperature} | tokens={max_tokens} | strategy={strategy_name}")
if on_debug:
on_debug("config", {"model": model, "temperature": temperature, "max_tokens": max_tokens, "strategy": strategy_name})
die_roll = random.randint(1, 6)
state.append_llm_log(f"Dice: {die_roll} (1d6)")
system = """You are an RPG dungeon master. The player just took an action.
Output ONLY ```tool blocks — no prose, no reasoning, no explanation outside tool blocks. Every piece of output must be in a tool block.
Use these tools to perform every action. Wrap each in its own ```tool block:
```tool
{"tool": "narrative", "args": {"text": "The full vivid narrative prose goes here."}}
```
```tool
{"tool": "modify_vitals", "args": {"current_hp": 5, "cash": 45}}
```
```tool
{"tool": "modify_traits", "args": {"dex": 15}}
```
```tool
{"tool": "add_to_inventory", "args": {"item": "Silver key"}}
```
```tool
{"tool": "remove_from_inventory", "args": {"item": "Torches (10)"}}
```
```tool
{"tool": "replace_gear", "args": {"before": "Mace (1d6+1)", "after": "Mace (1d6+2, sharpened)"}}
```
```tool
{"tool": "add_note", "args": {"note": "Found a hidden passage under the temple"}}
```
```tool
{"tool": "replace_note", "args": {"before": "Old note text", "after": "New note text"}}
```
```tool
{"tool": "world_update", "args": {"content": "# The World\n\n...full new world state..."}}
```
```tool
{"tool": "journal_update", "args": {"add": ["Investigate the mine"], "done": ["Defeat the demon"]}}
```
```tool
{"tool": "finalize_turn", "args": {"user_prompt": "What do you do?", "ambience": "dungeon"}}
```
"""
system += build_prose_prompt()
user = build_user_message(
player_action=player_action,
last_prompt=last_prompt,
)
user += f"\n\n*A die is cast: **{die_roll}** (1d6).*"
start_time = datetime.now()
set_llm_env(model, api_key, api_base)
state.append_llm_log(f"\n[TOOL] Single call — {len(system)} chars system, {len(user)} chars user")
state.append_llm_log(f"System preview: {system.split(chr(10))[0][:80]}...")
state.append_llm_log(f"User preview: {user.split(chr(10))[0][:80]}...")
text = call_llm(
[{"role": "system", "content": system},
{"role": "user", "content": user}],
model=model, temperature=temperature, timeout=timeout,
max_tokens=4096, label="Single tool call", on_debug=on_debug,
)
total_elapsed = (datetime.now() - start_time).total_seconds() * 1000
if text:
state.append_llm_log(f"\n[TOOL] got {len(text)} chars in {total_elapsed:.1f}ms")
if not text or not text.strip():
return TurnResult(error="Single tool call returned empty response")
raw = text.strip()
book_log = ""
log_entry = None
user_prompt = auto_prompt("")
ambience = None
tool_calls = []
changes: list[str] = []
phase3_errors: list[str] = []
tool_pattern = r"```tool\s*\n?(.*?)\n?```"
matches = re.findall(tool_pattern, text, re.DOTALL)
if matches:
for block in matches:
block = block.strip()
try:
tc = json.loads(block)
tool_calls.append(tc)
name = tc.get("tool", "unknown")
args = tc.get("args", {})
state.append_llm_log(f"\n[EXTRACT] {name}: {json.dumps(args)[:100]}")
if name == "narrative":
book_log = args.get("text", book_log)
elif name == "finalize_turn":
if args.get("user_prompt"):
user_prompt = args["user_prompt"]
if args.get("ambience"):
ambience = args["ambience"]
except json.JSONDecodeError as e:
state.append_llm_log(f"\n[EXTRACT] bad JSON: {e}")
continue
log_entry = None
if book_log:
clean = re.sub(r'\s+', ' ', book_log).strip()
first_sentence = re.split(r'(?<=[.!?])\s+', clean)
if first_sentence:
log_entry = first_sentence[0].strip()[:200]
else:
log_entry = clean[:200]
state.append_llm_log(f"\n[SUMMARY] \"{log_entry}\"")
extr_start = datetime.now()
for tc in tool_calls:
name = tc.get("tool", "unknown")
args = tc.get("args", {})
if name in ("finalize_turn", "narrative"):
continue
result = execute_tool(name, args)
if result.startswith("**Error:") or result.startswith("Tool error") or result.startswith("Unknown"):
phase3_errors.append(f"{name}: {result}")
else:
desc = describe_change(name, args)
if desc:
changes.append(desc)
apply_elapsed = (datetime.now() - extr_start).total_seconds() * 1000
state.append_llm_log(f"\n[APPLY] {len(changes)} changes in {apply_elapsed:.1f}ms")
else:
state.append_llm_log(f"\n[TOOL] no tool blocks found")
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") != "finalize_turn"])
on_debug("phase_done", {
"book_log_chars": len(book_log),
"log_entry": log_entry,
"user_prompt": user_prompt,
"ambience": ambience,
"extract_errors": phase3_errors or None,
"total_elapsed_ms": 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=strategy_name,
die_roll=die_roll,
model=model,
temperature=temperature,
max_tokens=max_tokens,
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(phase3_errors) if phase3_errors else "",
changes=changes,
)
tools/engine_lib/tools_handler.py
+463
View File
@@ -0,0 +1,463 @@
#!/usr/bin/env python3
"""
tools_handler.py — Tool call infrastructure for The Chaos engine.
Handles tool call extraction, execution, and description. All functions
are standalone — no dependency on the GameEngine class.
"""
from __future__ import annotations
import json
import random
import re
from .paths import CHAR_PATH, WORLD_PATH, LOG_DIR, TODAY
from .state import read_file, validate_update_size, update_journal, append_llm_log
# ── Tool Registry ───────────────────────────────────────────────────────────
TOOL_REGISTRY: dict[str, dict] = {
"roll": {"description": "Roll dice.", "args": {"dice": "1d6", "modifier": "+1"}},
"player_roll": {"description": "Ask player to roll.", "args": {"dice": "1d6", "reason": "why"}},
"modify_traits": {"description": "Change STR/DEX/WIL.", "args": {"str": "optional", "dex": "optional", "wil": "optional"}},
"modify_vitals": {"description": "Change HP, cash, weapon, armour.", "args": {"current_hp": "optional", "max_hp": "optional", "cash": "optional", "weapon": "optional", "armour": "optional"}},
"add_to_inventory": {"description": "Add item to gear.", "args": {"item": "item name and stats"}},
"remove_from_inventory": {"description": "Remove item from gear.", "args": {"item": "exact item text"}},
"replace_gear": {"description": "Replace gear by exact match.", "args": {"before": "exact text", "after": "new text"}},
"add_note": {"description": "Add note to sheet.", "args": {"note": "note content"}},
"replace_note": {"description": "Replace note by exact match.", "args": {"before": "exact text", "after": "new text"}},
"world_update": {"description": "Replace world state.", "args": {"content": "full world markdown"}},
"journal_update": {"description": "Update TODO/DONE.", "args": {"add": "[...]", "done": "[...]"}},
"finalize_turn": {"description": "End turn.", "args": {"user_prompt": "question for player", "ambience": "soundscape name"}},
}
# ── Character Sheet Patcher ─────────────────────────────────────────────────
def patch_character(pattern: str, repl: str, count: int = 1, flags: int = 0) -> str:
"""Apply a regex replacement to character.md. Returns error msg or empty string."""
text = CHAR_PATH.read_text()
new, n = re.subn(pattern, repl, text, count=count, flags=flags)
if n == 0:
return f"**Error:** pattern not found:\n{pattern}"
CHAR_PATH.write_text(new)
return ""
# ── Individual Tool Implementations ─────────────────────────────────────────
def tool_think(args: dict) -> str:
return ""
def tool_read_file(args: dict) -> str:
filename = (args or {}).get("file", "")
paths = {
"character": CHAR_PATH,
"world": WORLD_PATH,
"log": LOG_DIR / f"{TODAY}.md",
}
path = paths.get(filename)
if not path:
return f"Unknown file: {filename}. Choose from: {', '.join(paths)}"
return read_file(path) or f"*{filename} is empty.*"
def tool_roll(args: dict) -> str:
import random
dice_str = (args or {}).get("dice", "1d6")
modifier_str = (args or {}).get("modifier", "0")
try:
count, sides = dice_str.lower().split("d")
count = int(count) if count else 1
sides = int(sides)
except (ValueError, TypeError):
return f"Invalid dice: {dice_str}. Use format like '2d6'."
mod = 0
if modifier_str:
try:
mod = int(modifier_str)
except ValueError:
pass
rolls = [random.randint(1, sides) for _ in range(count)]
total = sum(rolls) + mod
mod_str = f" {'+' if mod >= 0 else ''}{mod}" if mod != 0 else ""
return f"Roll: {dice_str}{mod_str} → [{', '.join(str(r) for r in rolls)}] = {total}"
def tool_modify_traits(args: dict) -> str:
errors = []
for stat in ("str", "dex", "wil"):
val = args.get(stat)
if val is not None:
err = patch_character(
rf"^(- \*\*{stat.upper()}:\*\*\s*)\d+", rf"\g<1>{val}", count=1, flags=re.MULTILINE
)
if err:
errors.append(err)
return "; ".join(errors) if errors else "Traits updated."
def tool_modify_vitals(args: dict) -> str:
errors = []
for field, label in [("current_hp", "Current Health"), ("max_hp", "Max Health"),
("cash", "Cash"), ("weapon", "Weapon"), ("armour", "Armour")]:
val = args.get(field)
if val is not None:
err = patch_character(
rf"^(- \*\*{label}:\*\*\s*).*", rf"\g<1>{val}", count=1, flags=re.MULTILINE
)
if err:
errors.append(err)
return "; ".join(errors) if errors else "Vitals updated."
def tool_add_to_inventory(args: dict) -> str:
item = (args or {}).get("item", "")
if not item:
return "**Error:** `item` is required."
text = CHAR_PATH.read_text()
if item in text:
return f"Item already in inventory: {item}"
gear_section = re.search(r"^## Gear\n", text, re.MULTILINE)
if gear_section:
insert_at = gear_section.end()
text = text[:insert_at] + f"- {item}\n" + text[insert_at:]
else:
text += f"\n## Gear\n- {item}\n"
CHAR_PATH.write_text(text)
return f"Added to inventory: {item}"
def tool_remove_from_inventory(args: dict) -> str:
item = (args or {}).get("item", "")
if not item:
return "**Error:** `item` is required."
err = patch_character(rf"^- {re.escape(item)}\n?", "", count=1, flags=re.MULTILINE)
if err:
return f"**Error:** item not found: {item}"
return f"Removed from inventory: {item}"
def tool_replace_gear(args: dict) -> str:
before = (args or {}).get("before", "")
after = (args or {}).get("after", "")
if not before or not after:
return "**Error:** `before` and `after` are required."
err = patch_character(rf"^- {re.escape(before)}", f"- {after}", count=1, flags=re.MULTILINE)
if err:
return f"**Error:** gear not found: {before}"
return f"Gear replaced: {before} → {after}"
def tool_add_note(args: dict) -> str:
note = (args or {}).get("note", "")
if not note:
return "**Error:** `note` is required."
text = CHAR_PATH.read_text()
notes_section = re.search(r"^## Notes & Scribbles\n", text, re.MULTILINE)
if notes_section:
text = text[:notes_section.end()] + f"- {note}\n" + text[notes_section.end():]
else:
text += f"\n## Notes & Scribbles\n- {note}\n"
CHAR_PATH.write_text(text)
return f"Note added: {note}"
def tool_replace_note(args: dict) -> str:
before = (args or {}).get("before", "")
after = (args or {}).get("after", "")
if not before or not after:
return "**Error:** `before` and `after` are required."
err = patch_character(rf"^- {re.escape(before)}", f"- {after}", count=1, flags=re.MULTILINE)
if err:
return f"**Error:** note not found: {before}"
return f"Note replaced."
def tool_world_update(args: dict) -> str:
content = (args or {}).get("content", "")
if not content:
return "**Error:** `content` is required."
if not validate_update_size("world", content, WORLD_PATH):
return "**Error:** Update rejected — content is too short (likely a partial paste)."
WORLD_PATH.write_text(content.strip() + "\n")
return "World state updated."
def tool_journal_update(args: dict) -> str:
add = (args or {}).get("add", [])
done = (args or {}).get("done", [])
if isinstance(add, str):
add = [add]
if isinstance(done, str):
done = [done]
if not add and not done:
return "**Error:** Provide at least one of `add` or `done`."
update_journal(add=add, done=done)
return "Journal updated."
# ── Tool Dispatcher ─────────────────────────────────────────────────────────
def execute_tool(tool_name: str, args: dict) -> str:
"""Execute a tool by name. Returns result string."""
fn_map = {
"roll": tool_roll,
"modify_traits": tool_modify_traits,
"modify_vitals": tool_modify_vitals,
"add_to_inventory": tool_add_to_inventory,
"remove_from_inventory": tool_remove_from_inventory,
"replace_gear": tool_replace_gear,
"add_note": tool_add_note,
"replace_note": tool_replace_note,
"world_update": tool_world_update,
"journal_update": tool_journal_update,
}
fn = fn_map.get(tool_name)
if not fn:
return f"Unknown tool: {tool_name}"
try:
return fn(args)
except Exception as e:
import traceback
tb = traceback.format_exc()
append_llm_log(f"\n--- TOOL ERROR ({tool_name}) ---\n{tb}")
return f"Tool error ({tool_name}): {e}"
# ── Descriptions ────────────────────────────────────────────────────────────
def describe_tool_action(tool_name: str, args: dict) -> str:
"""Return a user-facing status message for a tool call."""
dm_status = (args or {}).get("dm_status")
if dm_status:
return f"DM is {dm_status}..."
read_descriptions = {
"character": "reading the character sheet",
"world": "consulting the world map",
"book": "reviewing the story so far",
"log": "checking the session log",
"journal": "scanning the journal",
}
if tool_name == "read_file":
file = (args or {}).get("file", "")
desc = read_descriptions.get(file, f"reading {file}")
elif tool_name in ("character_get", "world_get", "journal_get"):
file = tool_name.replace("_get", "")
desc = read_descriptions.get(file, f"reading {file}")
elif tool_name in ("character_update", "world_update"):
desc = "updating the records"
elif tool_name == "journal_update":
desc = "updating the journal"
elif tool_name == "roll":
dice = (args or {}).get("dice", "1d6")
mod = (args or {}).get("modifier")
desc = f"rolling {dice}"
if mod:
desc += f" {mod}"
elif tool_name == "player_roll":
dice = (args or {}).get("dice", "1d6")
desc = f"asking you to roll {dice}"
elif tool_name == "modify_traits":
desc = "updating traits"
elif tool_name == "modify_vitals":
desc = "updating vitals"
elif tool_name == "add_to_inventory":
desc = "adding item to inventory"
elif tool_name == "remove_from_inventory":
desc = "removing item from inventory"
elif tool_name == "replace_gear":
desc = "replacing gear"
elif tool_name == "add_note":
desc = "adding note"
elif tool_name == "replace_note":
desc = "replacing note"
else:
desc = f"using {tool_name}"
return f"DM is {desc}..."
def describe_change(tool_name: str, args: dict) -> str:
"""Build a compact human-readable change description from a tool call."""
if tool_name == "modify_vitals":
parts = []
for k, v in args.items():
label = k.replace("_", " ").title()
parts.append(f"{label}: {v}")
return f"⚡ {', '.join(parts)}" if parts else ""
elif tool_name == "modify_traits":
parts = []
for k, v in args.items():
parts.append(f"{k.upper()}: {v}")
return f"⚡ {', '.join(parts)}"
elif tool_name == "add_to_inventory":
return f"+ {args.get('item', '?')}"
elif tool_name == "remove_from_inventory":
return f"− {args.get('item', '?')}"
elif tool_name == "replace_gear":
return f"↻ {args.get('before', '?')} → {args.get('after', '?')}"
elif tool_name == "add_note":
note = args.get("note", "?")
return f"📝 {note[:60]}{'…' if len(note) > 60 else ''}"
elif tool_name == "replace_note":
return f"📝 {args.get('before', '?')[:40]} → {args.get('after', '?')[:40]}"
elif tool_name == "world_update":
return "🌍 World updated"
elif tool_name == "journal_update":
parts = []
for a in args.get("add", []):
parts.append(f"📋 {a}")
for d in args.get("done", []):
parts.append(f"✅ {d}")
return "; ".join(parts) if parts else ""
return ""
# ── Changes Block Parser ────────────────────────────────────────────────────
def parse_changes_block(changes_block: str) -> list[dict]:
"""Parse a ### Changes block into tool call dicts."""
calls = []
for raw_line in changes_block.split("\n"):
line = raw_line.strip()
if not line.startswith("- "):
continue
content = line[2:].strip()
m = re.match(r"Current Health:\s*(\d+)", content, re.IGNORECASE)
if m:
calls.append({"tool": "modify_vitals", "args": {"current_hp": m.group(1)}})
continue
m = re.match(r"Cash:\s*(\d+)", content, re.IGNORECASE)
if m:
calls.append({"tool": "modify_vitals", "args": {"cash": m.group(1)}})
continue
m = re.match(r"Max Health:\s*(\d+)", content, re.IGNORECASE)
if m:
calls.append({"tool": "modify_vitals", "args": {"max_hp": m.group(1)}})
continue
m = re.match(r"Add(?:ed)? to inventory:\s*(.+)", content, re.IGNORECASE)
if m:
for item in [i.strip() for i in m.group(1).split(",") if i.strip()]:
calls.append({"tool": "add_to_inventory", "args": {"item": item}})
continue
m = re.match(r"Remov(?:e|ed) from inventory:\s*(.+)", content, re.IGNORECASE)
if m:
for item in [i.strip() for i in m.group(1).split(",") if i.strip()]:
calls.append({"tool": "remove_from_inventory", "args": {"item": item}})
continue
m = re.match(r"Replace(?:d)? gear:\s*(.+?)\s*[→➜]\s*(.+)", content, re.IGNORECASE)
if m:
calls.append({"tool": "replace_gear", "args": {"before": m.group(1).strip(), "after": m.group(2).strip()}})
continue
m = re.match(r"Note:\s*(.+)", content, re.IGNORECASE)
if m:
calls.append({"tool": "add_note", "args": {"note": m.group(1).strip()}})
continue
m = re.match(r"Journal add:\s*(.+)", content, re.IGNORECASE)
if m:
calls.append({"tool": "journal_update", "args": {"add": [i.strip() for i in m.group(1).split(",") if i.strip()]}})
continue
m = re.match(r"Journal done:\s*(.+)", content, re.IGNORECASE)
if m:
calls.append({"tool": "journal_update", "args": {"done": [i.strip() for i in m.group(1).split(",") if i.strip()]}})
continue
m = re.match(r"Looted from .+:\s*(.+)", content, re.IGNORECASE)
if m:
items_text = m.group(1).strip()
calls.append({"tool": "add_note", "args": {"note": f"Looted: {items_text}"}})
continue
return calls
# ── Extraction Functions ────────────────────────────────────────────────────
def extract_thoughts(text: str) -> list[str]:
pattern = r"```thought\s*\n?(.*?)```"
return re.findall(pattern, text, re.DOTALL)
def extract_tool_calls(text: str, *, round_num: int = 0, on_debug: callable = None) -> list[dict]:
"""Extract tool calls from ```tool and ```json blocks."""
calls = []
seen = set()
def _try_parse(raw: str) -> dict | None:
try:
obj = json.loads(raw)
if isinstance(obj, dict) and "tool" in obj:
return obj
except json.JSONDecodeError:
pass
return None
for m in re.finditer(r"```(?:tool|json|finalize_turn)\s*\n?", text):
fence_type = m.group(0).strip("``` \n\r")
obj = None
try:
decoder = json.JSONDecoder()
obj, end = decoder.raw_decode(text, m.end())
except (json.JSONDecodeError, ValueError, StopIteration):
pass
if obj is None:
close = text.find("```", m.end())
if close > 0:
raw = text[m.end():close].strip()
def _escape_in_strings(s: str) -> str:
return re.sub(r'"(?:[^"\\]|\\.)*"', lambda x: x.group(0).replace("\n", "\\n"), s, flags=re.DOTALL)
repaired = _escape_in_strings(raw)
obj = _try_parse(repaired)
if obj is not None and isinstance(obj, dict):
if fence_type == "finalize_turn":
obj = {"tool": "finalize_turn", "args": obj}
if "tool" not in obj:
obj = None
if obj is not None:
key = (obj["tool"], json.dumps(obj.get("args", {}), sort_keys=True))
if key not in seen:
seen.add(key)
calls.append(obj)
elif on_debug:
preview = text[m.end():m.end() + 120].replace("\n", "\\n")
on_debug("parse_error", {"round": round_num, "content": preview})
return calls
def extract_final_json(text: str) -> dict | None:
pattern = r"```json\s*\n?(.*?)```"
matches = re.findall(pattern, text, re.DOTALL)
if not matches:
return None
try:
return json.loads(matches[-1].strip())
except json.JSONDecodeError:
return None
def strip_tool_blocks(text: str) -> str:
"""Remove ```tool, ```json, finalize_turn blocks from narrative text."""
return re.sub(
r'```(?:tool|json|finalize_turn)\s*\n?.*?```',
'',
text,
flags=re.DOTALL,
).strip()
tools/engine_lib/validation.py
+74
View File
@@ -0,0 +1,74 @@
#!/usr/bin/env python3
"""
validation.py — Narrative quality validation for The Chaos engine.
Standalone functions — no dependency on GameEngine.
"""
from __future__ import annotations
import re
from collections import Counter
from .llm import call_llm
def auto_prompt(book_log: str = "") -> str:
"""Fallback player prompt."""
return "**What do you do?**"
def validate_narrative(
book_log: str,
*,
model: str,
temperature: float,
timeout: int,
on_debug: callable = None,
) -> tuple[bool, str]:
"""Check if book_log is acceptable narrative. Returns (ok, reason)."""
lines = book_log.strip().split("\n")
if not lines:
return False, "Empty narrative"
common = Counter(lines).most_common(1)
if common and common[0][1] >= 5:
return False, f"Repetition: '{common[0][0][:60]}' ×{common[0][1]}"
mech_lines = [l for l in lines if re.match(
r'^\*\*(?:Roll|Damage|Success|Failure|Check|Save|Hit|Miss|'
r'Strenght|Dexterity|Willpower|STR|DEX|WIL|'
r'(?:[A-Z][a-z]+(?: \(\w+\))?:))',
l
)]
if mech_lines:
ratio = len(mech_lines) / len(lines)
if ratio > 0.3:
return False, f"Game mechanics dominate ({len(mech_lines)}/{len(lines)} lines)"
if re.search(r'```(?:tool|json)', book_log):
return False, "Contains unprocessed tool blocks"
prose = re.sub(r'[*_#>`~\-\d]', '', book_log).strip()
if len(prose) < 50:
return False, "Too short to be meaningful"
text = call_llm([
{"role": "user", "content":
f"Rate this RPG narrative quality 1-5.\n"
f"1 = unreadable (spam, repetition, pure mechanics, garbled)\n"
f"2 = poor (mostly mechanics, little story)\n"
f"3 = acceptable (some narrative but rough)\n"
f"4 = good (solid prose, minor issues)\n"
f"5 = excellent (vivid, engaging)\n"
f"Reply with ONLY a single digit 1-5.\n\n"
f"{book_log[:600]}"}
], model=model, temperature=temperature, timeout=timeout,
max_tokens=2, label="Narrative validation", on_debug=on_debug)
if text and text.strip().isdigit():
score = int(text.strip())
if score < 3:
return False, f"Quality score: {score}/5"
return True, ""