""" 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 # ========================================================================= # Main Context Building # ========================================================================= 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]] = [] # (content, priority) # Combat section (highest priority if in combat) if manager.in_combat and manager.combat_state: combat_section = self.build_combat_summary(manager.combat_state) sections.append((combat_section, 100)) # Party section (high priority) party_section = self.build_party_summary(manager) if party_section: sections.append((party_section, 90)) # Location section location_section = self._build_location_section(manager) if location_section: sections.append((location_section, 70)) # Recent events section events_section = self._build_events_section(manager) if events_section: sections.append((events_section, 60)) # NPCs section npcs_section = self._build_npcs_section(manager) if npcs_section: sections.append((npcs_section, 50)) # Combine sections within token budget 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 """ # Sort by priority (highest first) 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: # Try to truncate section to fit remaining_tokens = self._max_tokens - current_tokens if remaining_tokens > 100: # Worth including something 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) # ========================================================================= # Combat Summary # ========================================================================= 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] = [] # Header lines.append(f"**COMBAT - Round {combat.round_number}**") # Turn order for i, combatant in enumerate(combat.combatants): # Skip dead/fled combatants if combatant.status in (CombatantStatus.DEAD, CombatantStatus.FLED): continue # Current turn marker is_current = i == combat.turn_index marker = ">" if is_current else " " # Player/Enemy tag tag = "Player" if combatant.is_player else "Enemy" # HP status hp_status = self._get_hp_status_string(combatant.hp_current, combatant.hp_max) hp_display = f"{combatant.hp_current}/{combatant.hp_max} HP" # Conditions conditions_str = "" if combatant.conditions: conditions_str = f" [{', '.join(combatant.conditions)}]" elif combatant.status == CombatantStatus.UNCONSCIOUS: conditions_str = " [Unconscious]" # Status indicator for unconscious status_suffix = "" if combatant.status == CombatantStatus.UNCONSCIOUS: status_suffix = " - DOWN" # Build line 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) # ========================================================================= # Party Summary # ========================================================================= 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: # Active marker is_active = snapshot.character_id == manager.active_character_id active_marker = " (Active)" if is_active else "" # HP status hp_status = self._get_hp_status_string(snapshot.hp_current, snapshot.hp_max) hp_display = f"{snapshot.hp_current}/{snapshot.hp_max} HP" # Conditions conditions_str = "" if snapshot.conditions: conditions_str = f" [Conditions: {', '.join(snapshot.conditions)}]" # Class and level 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) # ========================================================================= # Location Section # ========================================================================= 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] = [] # Location header lines.append(f"**Location: {manager.current_location}**") scene = manager.current_scene if scene: # Description if scene.description: lines.append(scene.description) # Sensory details (if enabled) 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)) # Exits 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) # ========================================================================= # Events Section # ========================================================================= def _build_events_section(self, manager: GameStateManager) -> str: """ Build recent events section. Args: manager: GameStateManager with events Returns: Formatted events section """ # Get significant events first, then recent events = manager.event_logger.get_recent( count=self._max_events, significant_only=False, ) if not events: return "" # Filter to most relevant significant = [e for e in events if e.is_significant] recent = events[:5] # Last 5 events # Combine, preferring significant 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) # ========================================================================= # NPCs Section # ========================================================================= 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: # Name and brief description desc = npc.description[:100] + "..." if len(npc.description) > 100 else npc.description line = f"- {npc.name}: {desc}" # Add personality hint if available 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) # ========================================================================= # Helper Methods # ========================================================================= 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 # Find a good break point truncated = text[:max_chars] # Try to break at a newline last_newline = truncated.rfind("\n") if last_newline > max_chars * 0.5: return truncated[:last_newline] # Try to break at a sentence for punct in [".", "!", "?"]: last_punct = truncated.rfind(punct) if last_punct > max_chars * 0.5: return truncated[: last_punct + 1] # Just truncate with ellipsis return truncated[:-3] + "..." # ========================================================================= # Specialized Context Builders # ========================================================================= 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] = [] # Combat state (required) sections.append(self.build_combat_summary(manager.combat_state)) # Party HP summary party_section = self.build_party_summary(manager) if party_section: sections.append(party_section) # Current location (brief) 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 (detailed) location_section = self._build_location_section(manager) if location_section: sections.append(location_section) # NPCs npcs_section = self._build_npcs_section(manager) if npcs_section: sections.append(npcs_section) # Party status (brief) party_section = self.build_party_summary(manager) if party_section: sections.append(party_section) # Recent events 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 (detailed) 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)) # Location (brief) sections.append(f"**Location:** {manager.current_location}") # Recent dialogue events dialogue_events = manager.event_logger.get_recent( count=5, event_type=None, # Get all, filter below ) 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 status 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] = [] # Combat status 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" ) # Location parts.append(f"Location: {manager.current_location}") # Active character HP 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)