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

fix llm config for ramalama, add safeguards, spinner animation

Browse Source 
- Fix last_narrative -> last_prompt bug in generate() and generate_stream()
- Add **kwargs guard to build_user_message() to catch wrong param names
- Extract _set_llm_env() helper that always sets API key (fallback placeholder)
- Add 30s timeout to all litellm.completion() calls
- Update config model to openai/deepseek-r1 for ramalama
- Replace shifting dots spinner with fixed-character shape morph
This commit is contained in:
Dejvino 2026-06-25 19:21:55 +02:00
parent d265dfc7f7
commit 4c968f8096
3 changed files with 230 additions and 154 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
session/config.json
View File

@ -1,8 +1,8 @@
{
"llm": {
"model": "openai/gpt-4o-mini",
"api_key": "not-needed",
"model": "openai/deepseek-r1",
"api_key": null,
"api_base": "http://localhost:8080/v1",
"temperature": 0.8
"temperature": 0.7
}
}

300
tools/engine.py
View File

@ -35,6 +35,7 @@ TODAY = date.today().isoformat()
# ── Structured output ──────────────────────────────────────────────────────
@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
@ -46,6 +47,20 @@ class GenerationResult:
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
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
debug_info: str = ""
# ── DM System Prompt Template ──────────────────────────────────────────────
SYSTEM_PROMPT = Template("""You are the Dungeon Master for **The Chaos**, a solo card-based rules-light fantasy TTRPG. Your job is to narrate an immersive, responsive story for one player character.
@ -83,50 +98,46 @@ Favourable +1, Risky -1, Desperate -2, Well-prepared +1, Poor visibility -1, Rel
### Exploration
6 ten-minute watches per hour. Each meaningful action advances a watch.
## Output Format
IMPORTANT: End every response with a JSON fenced code block:
## How Turns Work
```json
{
"log_entry": "- **time of day** — brief description of what happened.",
"ambience": "ambience_name_or_null",
"character_updates": null,
"world_updates": null,
"journal_add": [],
"journal_done": []
}
```
Each turn follows this sequence:
1. The player's action or response is given to you.
2. Think about what happens. Read game state files, roll dice, or ask the player to roll.
3. When ready, call **finalize_turn** to complete the turn.
Rules for the JSON block:
- **log_entry**: One-line log entry summarizing this turn's action.
- **ambience**: One of: silence, calm, combat, dungeon, forest, tavern, tension, town, wilds. Set to null to keep current.
- **character_updates**: ONLY include if HP, cash, gear, or stats changed. Provide the FULL updated character sheet markdown. Otherwise null.
- **world_updates**: ONLY include if NPCs, locations, or world state changed. Provide the FULL updated world markdown. Otherwise null.
- **journal_add**: New TODO items to add.
- **journal_done**: TODO items that are now completed.
The **finalize_turn** tool produces all data for this turn:
- **book_log** — Narrative of what happened this turn. Appended to the story book.
- **user_prompt** — What the player sees next: describe the situation and ask what they do.
- **ambience** — One of: silence, calm, combat, dungeon, forest, tavern, tension, town, wilds.
- **character_updates** — Full character sheet (ONLY if HP/cash/gear/stats changed, otherwise omit).
- **world_updates** — Full world state (ONLY if NPCs/locations/threads changed, otherwise omit).
- **journal_add** — New TODO items.
- **journal_done** — Completed TODO items.
IMPORTANT: You MUST call **finalize_turn** to end the turn. Until then you will be called again to continue thinking and gathering information.
## Available Tools
You may use these tools before writing your final JSON block. Tool calls go in their own fenced code block:
Tool calls go in their own fenced code block:
```tool
{"tool": "tool_name", "args": {...}}
```
You can call tools any number of times, one per block. Available tools:
- **read_file** — Read a game state file. `{"file": "character|world|book|log|journal"}`
- **roll** — Roll dice (outcome is shown to player). `{"dice": "2d6", "modifier": "-1"}`
- **think** — Internal reasoning (shown in game status bar). `{"thought": "Your reasoning here"}`
- **player_roll** — Ask the player to roll physical dice. **Only use when the player's action has an uncertain outcome that the dice should decide.** `{"dice": "2d6", "reason": "why"}`
You may also show reasoning inline with a separate fence:
You may also show reasoning inline:
```thought
Your reasoning here
```
Call tools as needed, then end with the final JSON block.
Tools available:
Every tool call **must** include a `"dm_status"` string in `args` — a short, public-facing description of what the DM is doing (e.g. `"consulting the archives"`, `"examining the wound"`, `"calculating the odds"`). The player sees this in the UI. Keep it vague — never reveal what the DM is actually reading or learning.
- **read_file** — Read a game state file. `{"file": "character|world|book|log|journal", "dm_status": "..."}`
- **roll** — Auto-roll dice (outcome shown in status). `{"dice": "2d6", "modifier": "-1", "dm_status": "..."}`
- **player_roll** — Ask the player to roll physical dice. **Use when the outcome is uncertain.** `{"dice": "2d6", "reason": "why", "dm_status": "..."}`
- **finalize_turn** — **Complete the turn.** Provide all turn data as args. **Must include** `"dm_status"`.
When the player makes a choice, resolve it with the dice mechanics above. Describe the action, roll dice implicitly (describe the outcome, don't say "rolling dice"), apply damage/effects, and update state.
@ -202,6 +213,15 @@ class GameEngine:
def temperature(self) -> float:
return self.config.get("llm", {}).get("temperature", 0.8)
def _set_llm_env(self) -> None:
"""Set provider-specific env vars for litellm."""
prefix = self.model.split("/")[0].upper()
import os
key = self.api_key or "sk-placeholder"
os.environ[f"{prefix}_API_KEY"] = key
if self.api_base:
os.environ[f"{prefix}_API_BASE"] = self.api_base
# ── Context Assembly ────────────────────────────────────────────────
def _read_file(self, path: Path) -> str:
@ -243,42 +263,46 @@ class GameEngine:
def build_user_message(
self,
player_action: str | None = None,
last_narrative: 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_narrative:
parts.append(f"## Previously\n{last_narrative}")
if last_prompt:
parts.append(f"## Situation\n{last_prompt}")
if player_action:
parts.append(f"## Player Action\n{player_action}")
has_existing_story = bool(
self._read_file(BOOK_PATH).strip()
) if not last_narrative else True
) if not last_prompt else True
if not player_action and not last_narrative:
if not player_action and not last_prompt:
if has_existing_story:
parts.append(
"## Instructions\n"
"Continue the story from where it left off. Describe "
"what happens next based on the current situation. "
"Use tools as needed, then end with a JSON block."
"Continue the story from where it left off. Think, "
"gather information, then call finalize_turn."
)
else:
parts.append(
"## Instructions\n"
"Establish the opening scene. Dillion is at the "
"Splintered Tankard in the Keep. Describe the "
"setting and let the player decide what to do. "
"Use tools as needed, then end with a JSON block."
"setting, then call finalize_turn."
)
else:
parts.append(
"## Instructions\n"
"Describe the outcome of the player's action using game "
"mechanics where appropriate. Let the player decide their "
"next move freely. Use tools as needed, then end with a "
"JSON block."
"mechanics where appropriate. Think, gather information, "
"then call finalize_turn to complete the turn."
)
return "\n\n".join(parts)
@ -297,7 +321,7 @@ class GameEngine:
"""
system = self.build_system_prompt()
user = self.build_user_message(
player_action=player_action, last_narrative=last_narrative
player_action=player_action, last_prompt=last_narrative
)
messages = [
@ -316,15 +340,7 @@ class GameEngine:
)
# Set API key / base if provided
if self.api_key:
# litellm reads env vars or we can pass via kwargs
os_env_key = f"{self.model.split('/')[0]}_API_KEY".upper()
import os
os.environ[os_env_key] = self.api_key
if self.api_base:
os_env_base = f"{self.model.split('/')[0]}_API_BASE".upper()
import os
os.environ[os_env_base] = self.api_base
self._set_llm_env()
try:
response = litellm.completion(
@ -332,6 +348,7 @@ class GameEngine:
messages=messages,
temperature=self.temperature,
stream=False,
timeout=30,
)
text = response.choices[0].message.content or ""
except Exception as e:
@ -353,7 +370,7 @@ class GameEngine:
"""
system = self.build_system_prompt()
user = self.build_user_message(
player_action=player_action, last_narrative=last_narrative
player_action=player_action, last_prompt=last_narrative
)
messages = [
@ -369,14 +386,7 @@ class GameEngine:
})
return
if self.api_key:
os_env_key = f"{self.model.split('/')[0]}_API_KEY".upper()
import os
os.environ[os_env_key] = self.api_key
if self.api_base:
os_env_base = f"{self.model.split('/')[0]}_API_BASE".upper()
import os
os.environ[os_env_base] = self.api_base
self._set_llm_env()
try:
response = litellm.completion(
@ -384,6 +394,7 @@ class GameEngine:
messages=messages,
temperature=self.temperature,
stream=True,
timeout=30,
)
full_text = ""
for chunk in response:
@ -415,8 +426,24 @@ class GameEngine:
"description": "Ask the player to physically roll dice and enter the result.",
"args": {"dice": "e.g. 2d6+1", "reason": "Why the roll is needed (shown to player)"},
},
"finalize_turn": {
"description": "Complete the turn with all required data.",
"args": {
"book_log": "Narrative of what happened (appended to story book)",
"user_prompt": "What the player sees next — describe and ask what they do",
"ambience": "Optional: soundscape name",
"character_updates": "Optional: full character sheet if changed",
"world_updates": "Optional: full world state if changed",
"journal_add": "Optional: list of new TODO items",
"journal_done": "Optional: list of completed TODO items",
},
},
}
def _tool_think(self, args: dict) -> str:
"""Think tool — content is displayed via dm_status in the status bar."""
return ""
def _tool_read_file(self, args: dict) -> str:
filename = (args or {}).get("file", "")
paths = {
@ -454,7 +481,12 @@ class GameEngine:
@staticmethod
def _describe_tool_action(tool_name: str, args: dict) -> str:
"""Return a user-facing status message for a tool call."""
"""Return a user-facing status message for a tool call.
Prefer the LLM-provided dm_status — otherwise fall back to a generic description."""
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",
@ -474,8 +506,6 @@ class GameEngine:
elif tool_name == "player_roll":
dice = (args or {}).get("dice", "1d6")
desc = f"asking you to roll {dice}"
elif tool_name == "think":
desc = "pausing to think"
else:
desc = f"using {tool_name}"
return f"DM is {desc}..."
@ -484,6 +514,7 @@ class GameEngine:
fn_map = {
"read_file": self._tool_read_file,
"roll": self._tool_roll,
"think": self._tool_think,
}
fn = fn_map.get(tool_name)
if not fn:
@ -526,26 +557,25 @@ class GameEngine:
def generate_with_tools(
self,
player_action: str | None = None,
last_narrative: str | None = None,
last_prompt: str | None = None,
on_thought: callable = None,
on_action: callable = None,
on_player_roll: callable = None,
) -> GenerationResult:
) -> TurnResult:
"""
Multi-turn generation with tool-use loop.
The LLM can output ```thought blocks (reasoning), ```tool blocks
(tool calls), and a final ```json block. If any tool is called,
the result is fed back and the LLM is re-invoked. The loop ends
when the LLM outputs a ```json block with no tool calls.
The LLM can output ```thought blocks, call ```tool blocks, and
MUST call **finalize_turn** to complete the turn. Until then the
loop continues feeding tool results back.
`on_thought` is called for each thought block (may be from a
worker thread — use call_from_thread in the TUI).
`on_thought` / `on_action` may be called from a worker thread —
use call_from_thread in the TUI.
"""
system = self.build_system_prompt()
user = self.build_user_message(
player_action=player_action,
last_narrative=last_narrative,
last_prompt=last_prompt,
)
messages: list[dict] = [
@ -553,47 +583,102 @@ class GameEngine:
{"role": "user", "content": user},
]
self._set_llm_env()
try:
import litellm
except ImportError:
return GenerationResult(narrative="", error="litellm not installed")
if self.api_key:
os_env_key = f"{self.model.split('/')[0]}_API_KEY".upper()
import os
os.environ[os_env_key] = self.api_key
if self.api_base:
os_env_base = f"{self.model.split('/')[0]}_API_BASE".upper()
import os
os.environ[os_env_base] = self.api_base
return TurnResult(error="litellm not installed")
max_rounds = 10
debug_entries: list[str] = []
for round_idx in range(max_rounds):
round_log: list[str] = [f"── Round {round_idx + 1} ──"]
try:
response = litellm.completion(
model=self.model,
messages=messages,
temperature=self.temperature,
stream=False,
timeout=30,
)
text = response.choices[0].message.content or ""
except Exception as e:
return GenerationResult(narrative="", error=f"LLM call failed: {e}")
return TurnResult(error=f"LLM call failed: {e}")
# Thoughts
thoughts = self._extract_thoughts(text)
if thoughts:
round_log.append(f" thoughts: {len(thoughts)}")
for t in thoughts:
if on_thought:
on_thought(t.strip())
# Tool calls
tool_calls = self._extract_tool_calls(text)
final_data = self._extract_final_json(text)
finalize_call: dict | None = None
other_calls: list[dict] = []
if tool_calls:
results = []
for tc in tool_calls:
if tc.get("tool") == "finalize_turn":
finalize_call = tc
else:
other_calls.append(tc)
# Log tool call summary
if tool_calls:
names = [tc.get("tool", "?") for tc in tool_calls]
round_log.append(f" tools: {', '.join(names)}")
# finalize_turn present → validate and return
if finalize_call:
args = finalize_call.get("args", {})
errs = []
if not args.get("dm_status"):
errs.append("dm_status is required")
if not args.get("book_log"):
errs.append("book_log is required")
if not args.get("user_prompt"):
errs.append("user_prompt is required")
if errs:
round_log.append(f" finalize_turn validation errors: {', '.join(errs)}")
debug_entries.append("\n".join(round_log))
messages.append({"role": "assistant", "content": text})
messages.append({
"role": "user",
"content": f"## Validation Error\nfinalize_turn missing: {', '.join(errs)}. Please provide all required fields and call finalize_turn again."
})
continue
return TurnResult(
book_log=args.get("book_log", ""),
user_prompt=args.get("user_prompt", ""),
ambience=args.get("ambience"),
character_updates=args.get("character_updates"),
world_updates=args.get("world_updates"),
journal_add=args.get("journal_add", []),
journal_done=args.get("journal_done", []),
)
# Execute other tools
if other_calls:
results = []
for tc in other_calls:
name = tc.get("tool", "?")
args = tc.get("args", {})
# dm_status is required on every tool call
if not args.get("dm_status"):
err_msg = (
f"**Validation Error:** Tool `{name}` missing required `dm_status`. "
f"Describe what the DM is doing (e.g. "
f'`"dm_status": "consulting the archives"`). Please retry.'
)
results.append(err_msg)
round_log.append(f" {name}: MISSING dm_status")
continue
if on_action:
on_action(self._describe_tool_action(name, args))
if name == "player_roll" and on_player_roll:
@ -604,32 +689,28 @@ class GameEngine:
else:
result = self._execute_tool(name, args)
results.append(f"**Tool:** {name}\n**Args:** {json.dumps(args)}\n**Result:** {result}")
round_log.append(f" {name}: OK")
messages.append({"role": "assistant", "content": text})
messages.append({
"role": "user",
"content": "## Tool Results\n\n" + "\n\n".join(results),
})
continue # Another round
debug_entries.append("\n".join(round_log))
continue
# No tool calls → parse the final JSON
if final_data:
return GenerationResult(
narrative=text[: text.rfind("```json")].strip(),
choices=final_data.get("choices", []),
log_entry=final_data.get("log_entry"),
ambience=final_data.get("ambience"),
character_updates=final_data.get("character_updates"),
world_updates=final_data.get("world_updates"),
journal_add=final_data.get("journal_add", []),
journal_done=final_data.get("journal_done", []),
)
# No tools, no finalize → remind LLM
round_log.append(" no tool calls — prompted to use tools")
debug_entries.append("\n".join(round_log))
messages.append({"role": "assistant", "content": text})
messages.append({
"role": "user",
"content": "## Instructions\nUse tools to gather information or call **finalize_turn** to complete the turn."
})
# Fallback: no tool calls, no JSON → normal parse
return self.parse_response(text)
return GenerationResult(
narrative="",
error="Tool loop exceeded max rounds (10)",
debug_text = "\n\n".join(debug_entries)
return TurnResult(
error=f"Turn loop exceeded max rounds ({max_rounds}). Below is a debug log of what the LLM did each round:\n\n{debug_text}",
debug_info=debug_text,
)
# ── Response Parsing ────────────────────────────────────────────────
@ -686,8 +767,8 @@ class GameEngine:
# ── State Persistence ───────────────────────────────────────────────
def apply_state(self, result: GenerationResult) -> None:
"""Write state changes from a GenerationResult to disk."""
def apply_state(self, result: TurnResult) -> None:
"""Write state changes from a TurnResult to disk."""
if result.character_updates:
CHAR_PATH.write_text(result.character_updates.strip() + "\n")
@ -695,9 +776,6 @@ class GameEngine:
if result.world_updates:
WORLD_PATH.write_text(result.world_updates.strip() + "\n")
if result.log_entry:
self.append_log(result.log_entry)
if result.ambience:
AMBIENCE_PATH.write_text(result.ambience.strip().lower() + "\n")

78
tools/run.py
View File

@ -23,7 +23,7 @@ from rich.markdown import Markdown as RichMarkdown
from rich.theme import Theme
# ── Game engine ─────────────────────────────────────────
from engine import GameEngine, GenerationResult
from engine import GameEngine, GenerationResult, TurnResult
# ── Optional miniaudio ────────────────────────────────────
try:
@ -533,12 +533,13 @@ class ChaosTUI(App):
self.engine = GameEngine()
# Game loop state
self._last_narrative: str = ""
self._last_result: GenerationResult | None = None
self._last_prompt: str = ""
self._last_result: TurnResult | None = None
self._is_processing: bool = False
# Thinking animation
self._thinking_dots = 0
self._spinner_frames = ["◴", "◷", "◶", "◵"]
self._thinking_frame = 0
self._thinking_timer_handle = None
self._dm_action = "DM is weaving the narrative"
@ -605,7 +606,7 @@ class ChaosTUI(App):
# ── Game Loop ─────────────────────────────────────────
def _call_llm(self, player_action: str | None = None):
"""Called when we need new content from the LLM (scene or resolution)."""
"""Called when the player has acted — sends their action to the LLM."""
if self._is_processing:
return
self._is_processing = True
@ -626,7 +627,7 @@ class ChaosTUI(App):
def _run_generation(self, player_action: str | None) -> None:
"""Worker thread: calls engine.generate_with_tools() and posts result back."""
last_narrative = self._last_narrative if self._last_narrative else None
last_prompt = self._last_prompt if self._last_prompt else None
def on_thought(thought: str) -> None:
self.call_from_thread(self._on_thought, thought)
@ -636,7 +637,7 @@ class ChaosTUI(App):
result = self.engine.generate_with_tools(
player_action=player_action,
last_narrative=last_narrative,
last_prompt=last_prompt,
on_thought=on_thought,
on_action=on_action,
on_player_roll=self._on_player_roll,
@ -647,10 +648,11 @@ class ChaosTUI(App):
def _show_thinking(self) -> None:
"""Show the thinking indicator and start the animation timer."""
self._dm_action = "DM is weaving the narrative"
self._thinking_dots = 0
self._thinking_frame = 0
status = self.query_one("#play-status", Static)
status.add_class("processing")
status.update(f"✦ {self._dm_action} ✦")
spinner = self._spinner_frames[0]
status.update(f"✦ {spinner} {self._dm_action} ✦")
self._thinking_timer_handle = self.set_interval(
0.5, self._tick_thinking
)
@ -668,18 +670,20 @@ class ChaosTUI(App):
"""Display a thought from the DM in the status bar."""
display = thought[:60] + "…" if len(thought) > 60 else thought
self._dm_action = display
self._thinking_dots = 0
self._thinking_frame = 0
status = self.query_one("#play-status", Static)
status.add_class("processing")
status.update(f"✦ {display} ✦")
spinner = self._spinner_frames[0]
status.update(f"✦ {spinner} {display} ✦")
def _on_action(self, action: str) -> None:
"""Display a DM action (tool call) in the status bar."""
self._dm_action = action
self._thinking_dots = 0
self._thinking_frame = 0
status = self.query_one("#play-status", Static)
status.add_class("processing")
status.update(f"✦ {action} ✦")
spinner = self._spinner_frames[0]
status.update(f"✦ {spinner} {action} ✦")
def _on_player_roll(self, dice: str, reason: str) -> str:
"""Called from worker thread. Shows roll popup, blocks until player responds."""
@ -702,49 +706,42 @@ class ChaosTUI(App):
self.push_screen(RollModal(dice, reason), on_dismiss)
def _tick_thinking(self) -> None:
"""Animate the thinking dots on the current DM action."""
"""Animate the spinner on the current DM action."""
if not self._is_processing:
return
self._thinking_dots = (self._thinking_dots + 1) % 4
dots = "." * self._thinking_dots
self._thinking_frame = (self._thinking_frame + 1) % len(self._spinner_frames)
spinner = self._spinner_frames[self._thinking_frame]
status = self.query_one("#play-status", Static)
status.update(f"✦ {self._dm_action}{dots} ✦")
status.update(f"✦ {spinner} {self._dm_action} ✦")
def _on_generation_done(
self, result: GenerationResult, player_action: str | None
self, result: TurnResult, player_action: str | None
) -> None:
"""Handle the completed generation on the main thread."""
"""Handle the completed turn on the main thread."""
self._is_processing = False
self._hide_thinking()
if result.error:
self._show_error(result.error)
self._show_error(result.error, result.debug_info)
return
# If this was a resolution (player acted), archive the previous turn
if player_action and self._last_narrative:
archive_text = (
f"{self._last_narrative}\n\n"
f"---\n\n"
f"**Player chose:** {player_action}\n\n"
f"{result.narrative}"
)
self.engine.archive_turn(archive_text)
# Archive the turn's book log
if result.book_log:
self.engine.archive_turn(result.book_log)
# Apply state changes
if result.character_updates or result.world_updates:
self.engine.apply_state(result)
# Display the scene
# Display the next user prompt
self._display_scene(result)
# Store for next turn
self._last_narrative = result.narrative
self._last_prompt = result.user_prompt
self._last_result = result
def _display_scene(self, result: GenerationResult) -> None:
"""Update the UI with a new scene."""
self._set_narrative(result.narrative)
def _display_scene(self, result: TurnResult) -> None:
"""Update the UI with the next user prompt."""
self._set_narrative(result.user_prompt)
self._enable_input()
def _enable_input(self) -> None:
@ -761,11 +758,12 @@ class ChaosTUI(App):
scroll = self.query_one("#play-scroll", VerticalScroll)
scroll.scroll_home(animate=False)
def _show_error(self, error: str) -> None:
self._set_narrative(
f"**Error:** {error}\n\n"
"Check your session/config.json and ensure your LLM provider is running."
)
def _show_error(self, error: str, debug_info: str = "") -> None:
text = f"**Error:** {error}\n\n"
if debug_info:
text += f"**Debug Info:**\n\n{debug_info}\n\n"
text += "Check your session/config.json and ensure your LLM provider is running."
self._set_narrative(text)
self._enable_input()
# ── Input handling ────────────────────────────────────