|
|
""" |
|
|
DungeonMaster AI - Story Context Builder |
|
|
|
|
|
Builds formatted context strings for LLM prompts from game state. |
|
|
Manages token budgets and prioritizes information for context windows. |
|
|
""" |
|
|
|
|
|
from __future__ import annotations |
|
|
|
|
|
import logging |
|
|
from typing import TYPE_CHECKING |
|
|
|
|
|
from .models import ( |
|
|
CombatState, |
|
|
CombatantStatus, |
|
|
CharacterSnapshot, |
|
|
HPStatus, |
|
|
NPCInfo, |
|
|
SceneInfo, |
|
|
SessionEvent, |
|
|
) |
|
|
|
|
|
if TYPE_CHECKING: |
|
|
from .game_state_manager import GameStateManager |
|
|
|
|
|
logger = logging.getLogger(__name__) |
|
|
|
|
|
|
|
|
class StoryContextBuilder: |
|
|
""" |
|
|
Builds LLM context strings from game state. |
|
|
|
|
|
Formats party status, combat state, location, recent events, |
|
|
and NPCs into a structured context for the DM agent. |
|
|
Manages token budgets and prioritizes important information. |
|
|
""" |
|
|
|
|
|
def __init__( |
|
|
self, |
|
|
max_tokens: int = 2000, |
|
|
max_events: int = 10, |
|
|
include_sensory_details: bool = True, |
|
|
) -> None: |
|
|
""" |
|
|
Initialize the context builder. |
|
|
|
|
|
Args: |
|
|
max_tokens: Maximum estimated tokens for context |
|
|
max_events: Maximum recent events to include |
|
|
include_sensory_details: Whether to include sensory details |
|
|
""" |
|
|
self._max_tokens = max_tokens |
|
|
self._max_events = max_events |
|
|
self._include_sensory = include_sensory_details |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def build_full_context(self, manager: GameStateManager) -> str: |
|
|
""" |
|
|
Build the complete context string for LLM prompts. |
|
|
|
|
|
Args: |
|
|
manager: GameStateManager with current state |
|
|
|
|
|
Returns: |
|
|
Formatted context string |
|
|
""" |
|
|
sections: list[tuple[str, int]] = [] |
|
|
|
|
|
|
|
|
if manager.in_combat and manager.combat_state: |
|
|
combat_section = self.build_combat_summary(manager.combat_state) |
|
|
sections.append((combat_section, 100)) |
|
|
|
|
|
|
|
|
party_section = self.build_party_summary(manager) |
|
|
if party_section: |
|
|
sections.append((party_section, 90)) |
|
|
|
|
|
|
|
|
location_section = self._build_location_section(manager) |
|
|
if location_section: |
|
|
sections.append((location_section, 70)) |
|
|
|
|
|
|
|
|
events_section = self._build_events_section(manager) |
|
|
if events_section: |
|
|
sections.append((events_section, 60)) |
|
|
|
|
|
|
|
|
npcs_section = self._build_npcs_section(manager) |
|
|
if npcs_section: |
|
|
sections.append((npcs_section, 50)) |
|
|
|
|
|
|
|
|
return self._combine_sections(sections) |
|
|
|
|
|
def _combine_sections( |
|
|
self, |
|
|
sections: list[tuple[str, int]], |
|
|
) -> str: |
|
|
""" |
|
|
Combine sections within token budget. |
|
|
|
|
|
Args: |
|
|
sections: List of (content, priority) tuples |
|
|
|
|
|
Returns: |
|
|
Combined context string |
|
|
""" |
|
|
|
|
|
sections.sort(key=lambda x: x[1], reverse=True) |
|
|
|
|
|
result_parts: list[str] = [] |
|
|
current_tokens = 0 |
|
|
|
|
|
for content, priority in sections: |
|
|
section_tokens = self._estimate_tokens(content) |
|
|
|
|
|
if current_tokens + section_tokens <= self._max_tokens: |
|
|
result_parts.append(content) |
|
|
current_tokens += section_tokens |
|
|
else: |
|
|
|
|
|
remaining_tokens = self._max_tokens - current_tokens |
|
|
if remaining_tokens > 100: |
|
|
truncated = self._truncate_to_tokens(content, remaining_tokens) |
|
|
if truncated: |
|
|
result_parts.append(truncated) |
|
|
current_tokens += self._estimate_tokens(truncated) |
|
|
break |
|
|
|
|
|
return "\n\n".join(result_parts) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def build_combat_summary(self, combat: CombatState) -> str: |
|
|
""" |
|
|
Build a combat state summary. |
|
|
|
|
|
Args: |
|
|
combat: Current combat state |
|
|
|
|
|
Returns: |
|
|
Formatted combat summary |
|
|
""" |
|
|
lines: list[str] = [] |
|
|
|
|
|
|
|
|
lines.append(f"**COMBAT - Round {combat.round_number}**") |
|
|
|
|
|
|
|
|
for i, combatant in enumerate(combat.combatants): |
|
|
|
|
|
if combatant.status in (CombatantStatus.DEAD, CombatantStatus.FLED): |
|
|
continue |
|
|
|
|
|
|
|
|
is_current = i == combat.turn_index |
|
|
marker = ">" if is_current else " " |
|
|
|
|
|
|
|
|
tag = "Player" if combatant.is_player else "Enemy" |
|
|
|
|
|
|
|
|
hp_status = self._get_hp_status_string(combatant.hp_current, combatant.hp_max) |
|
|
hp_display = f"{combatant.hp_current}/{combatant.hp_max} HP" |
|
|
|
|
|
|
|
|
conditions_str = "" |
|
|
if combatant.conditions: |
|
|
conditions_str = f" [{', '.join(combatant.conditions)}]" |
|
|
elif combatant.status == CombatantStatus.UNCONSCIOUS: |
|
|
conditions_str = " [Unconscious]" |
|
|
|
|
|
|
|
|
status_suffix = "" |
|
|
if combatant.status == CombatantStatus.UNCONSCIOUS: |
|
|
status_suffix = " - DOWN" |
|
|
|
|
|
|
|
|
turn_indicator = " <- Your Turn" if is_current and combatant.is_player else "" |
|
|
line = ( |
|
|
f"{marker} {combatant.initiative}: {combatant.name} ({tag}) - " |
|
|
f"{hp_display} [{hp_status}]{conditions_str}{status_suffix}{turn_indicator}" |
|
|
) |
|
|
lines.append(line) |
|
|
|
|
|
return "\n".join(lines) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def build_party_summary(self, manager: GameStateManager) -> str: |
|
|
""" |
|
|
Build a party status summary. |
|
|
|
|
|
Args: |
|
|
manager: GameStateManager with party data |
|
|
|
|
|
Returns: |
|
|
Formatted party summary |
|
|
""" |
|
|
snapshots = manager.get_party_snapshots() |
|
|
if not snapshots: |
|
|
return "" |
|
|
|
|
|
lines: list[str] = ["**Party Status:**"] |
|
|
|
|
|
for snapshot in snapshots: |
|
|
|
|
|
is_active = snapshot.character_id == manager.active_character_id |
|
|
active_marker = " (Active)" if is_active else "" |
|
|
|
|
|
|
|
|
hp_status = self._get_hp_status_string(snapshot.hp_current, snapshot.hp_max) |
|
|
hp_display = f"{snapshot.hp_current}/{snapshot.hp_max} HP" |
|
|
|
|
|
|
|
|
conditions_str = "" |
|
|
if snapshot.conditions: |
|
|
conditions_str = f" [Conditions: {', '.join(snapshot.conditions)}]" |
|
|
|
|
|
|
|
|
class_info = f"Lvl {snapshot.level} {snapshot.character_class}" |
|
|
|
|
|
line = ( |
|
|
f"- {snapshot.name}{active_marker}: {hp_display} [{hp_status}] " |
|
|
f"- {class_info}{conditions_str}" |
|
|
) |
|
|
lines.append(line) |
|
|
|
|
|
return "\n".join(lines) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _build_location_section(self, manager: GameStateManager) -> str: |
|
|
""" |
|
|
Build location/scene section. |
|
|
|
|
|
Args: |
|
|
manager: GameStateManager with location data |
|
|
|
|
|
Returns: |
|
|
Formatted location section |
|
|
""" |
|
|
lines: list[str] = [] |
|
|
|
|
|
|
|
|
lines.append(f"**Location: {manager.current_location}**") |
|
|
|
|
|
scene = manager.current_scene |
|
|
if scene: |
|
|
|
|
|
if scene.description: |
|
|
lines.append(scene.description) |
|
|
|
|
|
|
|
|
if self._include_sensory and scene.sensory_details: |
|
|
sensory_parts: list[str] = [] |
|
|
for sense, detail in scene.sensory_details.items(): |
|
|
if detail: |
|
|
sensory_parts.append(f"*{sense.capitalize()}*: {detail}") |
|
|
if sensory_parts: |
|
|
lines.append("\n".join(sensory_parts)) |
|
|
|
|
|
|
|
|
if scene.exits: |
|
|
exits_str = ", ".join( |
|
|
f"{direction} to {dest}" |
|
|
for direction, dest in scene.exits.items() |
|
|
) |
|
|
lines.append(f"*Exits*: {exits_str}") |
|
|
|
|
|
return "\n".join(lines) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _build_events_section(self, manager: GameStateManager) -> str: |
|
|
""" |
|
|
Build recent events section. |
|
|
|
|
|
Args: |
|
|
manager: GameStateManager with events |
|
|
|
|
|
Returns: |
|
|
Formatted events section |
|
|
""" |
|
|
|
|
|
events = manager.event_logger.get_recent( |
|
|
count=self._max_events, |
|
|
significant_only=False, |
|
|
) |
|
|
|
|
|
if not events: |
|
|
return "" |
|
|
|
|
|
|
|
|
significant = [e for e in events if e.is_significant] |
|
|
recent = events[:5] |
|
|
|
|
|
|
|
|
to_show: list[SessionEvent] = [] |
|
|
seen_ids: set[str] = set() |
|
|
|
|
|
for event in significant + recent: |
|
|
if event.event_id not in seen_ids: |
|
|
to_show.append(event) |
|
|
seen_ids.add(event.event_id) |
|
|
if len(to_show) >= self._max_events: |
|
|
break |
|
|
|
|
|
if not to_show: |
|
|
return "" |
|
|
|
|
|
lines: list[str] = ["**Recent Events:**"] |
|
|
for event in to_show: |
|
|
lines.append(f"- {event.description}") |
|
|
|
|
|
return "\n".join(lines) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _build_npcs_section(self, manager: GameStateManager) -> str: |
|
|
""" |
|
|
Build NPCs present section. |
|
|
|
|
|
Args: |
|
|
manager: GameStateManager with NPC data |
|
|
|
|
|
Returns: |
|
|
Formatted NPCs section |
|
|
""" |
|
|
npcs = manager.get_npcs_in_scene() |
|
|
if not npcs: |
|
|
return "" |
|
|
|
|
|
lines: list[str] = ["**NPCs Present:**"] |
|
|
|
|
|
for npc in npcs: |
|
|
|
|
|
desc = npc.description[:100] + "..." if len(npc.description) > 100 else npc.description |
|
|
line = f"- {npc.name}: {desc}" |
|
|
|
|
|
|
|
|
if npc.personality: |
|
|
personality_short = ( |
|
|
npc.personality[:50] + "..." |
|
|
if len(npc.personality) > 50 |
|
|
else npc.personality |
|
|
) |
|
|
line += f" ({personality_short})" |
|
|
|
|
|
lines.append(line) |
|
|
|
|
|
return "\n".join(lines) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _get_hp_status_string(self, hp_current: int, hp_max: int) -> str: |
|
|
""" |
|
|
Get HP status string from current/max HP. |
|
|
|
|
|
Args: |
|
|
hp_current: Current HP |
|
|
hp_max: Maximum HP |
|
|
|
|
|
Returns: |
|
|
Status string (Healthy, Wounded, Critical, Unconscious) |
|
|
""" |
|
|
if hp_current <= 0: |
|
|
return "Unconscious" |
|
|
|
|
|
if hp_max <= 0: |
|
|
return "Unknown" |
|
|
|
|
|
percent = (hp_current / hp_max) * 100 |
|
|
|
|
|
if percent > 50: |
|
|
return "Healthy" |
|
|
elif percent > 25: |
|
|
return "Wounded" |
|
|
else: |
|
|
return "Critical" |
|
|
|
|
|
def _estimate_tokens(self, text: str) -> int: |
|
|
""" |
|
|
Estimate token count for text. |
|
|
|
|
|
Uses rough approximation of 4 characters per token. |
|
|
|
|
|
Args: |
|
|
text: Text to estimate |
|
|
|
|
|
Returns: |
|
|
Estimated token count |
|
|
""" |
|
|
return len(text) // 4 |
|
|
|
|
|
def _truncate_to_tokens(self, text: str, max_tokens: int) -> str: |
|
|
""" |
|
|
Truncate text to fit within token budget. |
|
|
|
|
|
Args: |
|
|
text: Text to truncate |
|
|
max_tokens: Maximum tokens |
|
|
|
|
|
Returns: |
|
|
Truncated text |
|
|
""" |
|
|
max_chars = max_tokens * 4 |
|
|
|
|
|
if len(text) <= max_chars: |
|
|
return text |
|
|
|
|
|
|
|
|
truncated = text[:max_chars] |
|
|
|
|
|
|
|
|
last_newline = truncated.rfind("\n") |
|
|
if last_newline > max_chars * 0.5: |
|
|
return truncated[:last_newline] |
|
|
|
|
|
|
|
|
for punct in [".", "!", "?"]: |
|
|
last_punct = truncated.rfind(punct) |
|
|
if last_punct > max_chars * 0.5: |
|
|
return truncated[: last_punct + 1] |
|
|
|
|
|
|
|
|
return truncated[:-3] + "..." |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def build_combat_context(self, manager: GameStateManager) -> str: |
|
|
""" |
|
|
Build context optimized for combat situations. |
|
|
|
|
|
Prioritizes combat state, current combatant, and party HP. |
|
|
|
|
|
Args: |
|
|
manager: GameStateManager |
|
|
|
|
|
Returns: |
|
|
Combat-focused context |
|
|
""" |
|
|
if not manager.in_combat or not manager.combat_state: |
|
|
return self.build_full_context(manager) |
|
|
|
|
|
sections: list[str] = [] |
|
|
|
|
|
|
|
|
sections.append(self.build_combat_summary(manager.combat_state)) |
|
|
|
|
|
|
|
|
party_section = self.build_party_summary(manager) |
|
|
if party_section: |
|
|
sections.append(party_section) |
|
|
|
|
|
|
|
|
sections.append(f"**Location:** {manager.current_location}") |
|
|
|
|
|
return "\n\n".join(sections) |
|
|
|
|
|
def build_exploration_context(self, manager: GameStateManager) -> str: |
|
|
""" |
|
|
Build context optimized for exploration. |
|
|
|
|
|
Prioritizes location details, sensory info, and NPCs. |
|
|
|
|
|
Args: |
|
|
manager: GameStateManager |
|
|
|
|
|
Returns: |
|
|
Exploration-focused context |
|
|
""" |
|
|
sections: list[str] = [] |
|
|
|
|
|
|
|
|
location_section = self._build_location_section(manager) |
|
|
if location_section: |
|
|
sections.append(location_section) |
|
|
|
|
|
|
|
|
npcs_section = self._build_npcs_section(manager) |
|
|
if npcs_section: |
|
|
sections.append(npcs_section) |
|
|
|
|
|
|
|
|
party_section = self.build_party_summary(manager) |
|
|
if party_section: |
|
|
sections.append(party_section) |
|
|
|
|
|
|
|
|
events_section = self._build_events_section(manager) |
|
|
if events_section: |
|
|
sections.append(events_section) |
|
|
|
|
|
return "\n\n".join(sections) |
|
|
|
|
|
def build_social_context(self, manager: GameStateManager) -> str: |
|
|
""" |
|
|
Build context optimized for social encounters. |
|
|
|
|
|
Prioritizes NPC information and dialogue history. |
|
|
|
|
|
Args: |
|
|
manager: GameStateManager |
|
|
|
|
|
Returns: |
|
|
Social-focused context |
|
|
""" |
|
|
sections: list[str] = [] |
|
|
|
|
|
|
|
|
npcs = manager.get_npcs_in_scene() |
|
|
if npcs: |
|
|
lines: list[str] = ["**NPCs Present:**"] |
|
|
for npc in npcs: |
|
|
lines.append(f"\n**{npc.name}**") |
|
|
if npc.description: |
|
|
lines.append(npc.description) |
|
|
if npc.personality: |
|
|
lines.append(f"*Personality*: {npc.personality}") |
|
|
if npc.dialogue_hooks: |
|
|
lines.append(f"*Might say*: \"{npc.dialogue_hooks[0]}\"") |
|
|
sections.append("\n".join(lines)) |
|
|
|
|
|
|
|
|
sections.append(f"**Location:** {manager.current_location}") |
|
|
|
|
|
|
|
|
dialogue_events = manager.event_logger.get_recent( |
|
|
count=5, |
|
|
event_type=None, |
|
|
) |
|
|
dialogue_lines: list[str] = ["**Recent Conversation:**"] |
|
|
for event in dialogue_events: |
|
|
if "dialogue" in event.event_type.value.lower(): |
|
|
dialogue_lines.append(f"- {event.description}") |
|
|
if len(dialogue_lines) > 1: |
|
|
sections.append("\n".join(dialogue_lines)) |
|
|
|
|
|
|
|
|
party_section = self.build_party_summary(manager) |
|
|
if party_section: |
|
|
sections.append(party_section) |
|
|
|
|
|
return "\n\n".join(sections) |
|
|
|
|
|
def build_minimal_context(self, manager: GameStateManager) -> str: |
|
|
""" |
|
|
Build minimal context for token-constrained situations. |
|
|
|
|
|
Only includes essential information. |
|
|
|
|
|
Args: |
|
|
manager: GameStateManager |
|
|
|
|
|
Returns: |
|
|
Minimal context string |
|
|
""" |
|
|
parts: list[str] = [] |
|
|
|
|
|
|
|
|
if manager.in_combat and manager.combat_state: |
|
|
current = manager.combat_state.current_combatant |
|
|
if current: |
|
|
parts.append( |
|
|
f"Combat Round {manager.combat_state.round_number}, " |
|
|
f"{current.name}'s turn" |
|
|
) |
|
|
|
|
|
|
|
|
parts.append(f"Location: {manager.current_location}") |
|
|
|
|
|
|
|
|
for snapshot in manager.get_party_snapshots(): |
|
|
if snapshot.character_id == manager.active_character_id: |
|
|
parts.append( |
|
|
f"Active: {snapshot.name} " |
|
|
f"({snapshot.hp_current}/{snapshot.hp_max} HP)" |
|
|
) |
|
|
break |
|
|
|
|
|
return " | ".join(parts) |
|
|
|