""" Character Memory System for BaoLife Provides long-term memory for NPCs by extracting and storing important facts from conversations. Enables characters to remember player preferences, past events, and relationship history across sessions. """ import asyncio from openai import AsyncOpenAI # Avoid circular import with functions.py by importing directly from source from database.db_operations import get_database_connection from config import config # Create OpenAI async client openai_client = AsyncOpenAI(api_key=config.OPENAI_API_KEY) # Import API tracker for cost monitoring try: from api_usage_tracker import tracker as api_tracker except ImportError: api_tracker = None # Gracefully handle if not available class CharacterMemory: """ Persistent memory system for NPC characters. Extracts and stores important facts from conversations. """ def __init__(self, character_id, player_id): """ Initialize character memory. Args: character_id: ID of the character player_id: ID of the player """ self.character_id = character_id self.player_id = player_id self.facts = [] # Cached facts self.loaded = False async def extract_facts(self, conversation, character): """ Extract important facts from recent conversation messages. Called periodically (e.g., every 5 messages) to update memory. Args: conversation: conversationObj instance character: personClass instance """ # Only extract if we have enough new messages if len(conversation.conversation) < 5: return # Get last 5 messages for fact extraction recent_messages = conversation.conversation[-5:] # Format messages for extraction conversation_text = self._format_messages_for_extraction(recent_messages, character) # Create extraction prompt extraction_prompt = f"""Extract key facts from this conversation that {character.firstname} should remember about the player. Conversation: {conversation_text} Extract facts in this format (one per line): - [Fact about player's preferences, plans, or important information] Only extract genuinely important facts (max 3). If nothing important, respond with "None". Facts should be specific and actionable for future conversations. Facts:""" try: # Call API for fact extraction result = await asyncio.wait_for( openai_client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": extraction_prompt}], max_tokens=200, temperature=0.2 # Low temperature for factual extraction ), timeout=5.0 ) facts_text = result.choices[0].message.content.strip() # Track API usage for fact extraction if api_tracker: try: conversation_id = conversation.id if hasattr(conversation, 'id') else None cost = api_tracker.track_usage( self.player_id, conversation_id, 'gpt-4o-mini', { 'prompt_tokens': result.usage.prompt_tokens, 'completion_tokens': result.usage.completion_tokens, 'total_tokens': result.usage.total_tokens }, purpose='fact_extraction' ) print(f"Fact extraction cost: ${cost:.6f}") except Exception as e: print(f"Failed to track fact extraction usage: {e}") if facts_text.lower() != "none" and facts_text: # Parse facts from response facts = [ f.strip()[2:].strip() # Remove "- " prefix for f in facts_text.split('\n') if f.strip().startswith('-') ] if facts: # Save facts to database self._save_facts(facts, conversation) self.facts.extend(facts) print(f"Extracted {len(facts)} new facts for {character.firstname}") return facts except Exception as e: print(f"Failed to extract facts: {e}") return [] def _format_messages_for_extraction(self, messages, character): """Format messages for fact extraction""" formatted = [] for msg in messages: speaker = "Player" if msg.sender != character.id else character.firstname formatted.append(f"{speaker}: {msg.message}") return "\n".join(formatted) def load_facts(self, limit=10): """ Load character's memory facts from database. Args: limit: Maximum number of facts to load (most recent) Returns: List of fact strings """ if self.loaded and self.facts: return self.facts mydb = get_database_connection() try: with mydb.cursor(dictionary=True) as cursor: cursor.execute(""" SELECT fact, importance, learned_date FROM character_memory WHERE character_id = %s AND player_id = %s ORDER BY importance DESC, learned_date DESC LIMIT %s """, (self.character_id, self.player_id, limit)) results = cursor.fetchall() self.facts = [row['fact'] for row in results] self.loaded = True print(f"Loaded {len(self.facts)} facts for character {self.character_id}") return self.facts except Exception as e: print(f"Failed to load character memory: {e}") # Table might not exist yet, return empty list return [] finally: mydb.close() def get_relevant_facts(self, max_facts=5): """ Get relevant facts for context injection. Args: max_facts: Maximum number of facts to return Returns: List of fact strings """ if not self.loaded: self.load_facts() # Return most recent/important facts return self.facts[:max_facts] def get_memory_context(self): """ Get formatted memory context for prompt injection. Returns: String containing memory context, or empty string if no memories """ facts = self.get_relevant_facts() if not facts: return "" # Format facts for prompt facts_text = "; ".join(facts) return f"You remember these things about the player: {facts_text}" def _save_facts(self, facts, conversation): """ Save extracted facts to database. Args: facts: List of fact strings conversation: conversationObj instance for metadata """ mydb = get_database_connection() try: with mydb.cursor() as cursor: # Prepare batch insert values = [] for fact in facts: # Calculate importance (simple heuristic: longer = more important) importance = min(10, max(3, len(fact) // 10)) values.append(( self.character_id, self.player_id, fact, importance, conversation.id if hasattr(conversation, 'id') else None )) # Batch insert all facts cursor.executemany(""" INSERT INTO character_memory (character_id, player_id, fact, importance, conversation_id, learned_date) VALUES (%s, %s, %s, %s, %s, NOW()) """, values) mydb.commit() print(f"Saved {len(facts)} facts to database") except Exception as e: print(f"Failed to save facts to database: {e}") print(f"Facts were: {facts}") # Don't crash if table doesn't exist yet finally: mydb.close() def clear_facts(self): """Clear all facts for this character-player relationship""" mydb = get_database_connection() try: with mydb.cursor() as cursor: cursor.execute(""" DELETE FROM character_memory WHERE character_id = %s AND player_id = %s """, (self.character_id, self.player_id)) mydb.commit() self.facts = [] self.loaded = False print(f"Cleared all facts for character {self.character_id}") except Exception as e: print(f"Failed to clear facts: {e}") finally: mydb.close() def create_character_memory_table(): """ Create character_memory table if it doesn't exist. Call this during server startup or migration. """ mydb = get_database_connection() try: with mydb.cursor() as cursor: cursor.execute(""" CREATE TABLE IF NOT EXISTS character_memory ( id INT AUTO_INCREMENT PRIMARY KEY, character_id VARCHAR(36) NOT NULL, player_id VARCHAR(36) NOT NULL, fact TEXT NOT NULL, learned_date DATETIME NOT NULL, importance TINYINT DEFAULT 5, conversation_id VARCHAR(36), INDEX idx_char_player (character_id, player_id), INDEX idx_importance (importance DESC), INDEX idx_learned_date (learned_date DESC) ) """) mydb.commit() print("Character memory table created/verified") except Exception as e: print(f"Failed to create character_memory table: {e}") finally: mydb.close() # NOTE: Do not initialize table on import to avoid creating DB connections at import time. # The table will be created on first use if needed, or can be created explicitly # by calling create_character_memory_table() during server startup.