DungeonMaster-AI / src /game /story_context.py
bhupesh-sf's picture
first commit
f8ba6bf verified
"""
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)