diff --git a/character_explorer.py b/character_explorer.py index c921903..480eba8 100644 --- a/character_explorer.py +++ b/character_explorer.py @@ -1,6 +1,9 @@ import asyncio import os +import pickle from pprint import pprint +from datetime import datetime +from typing import Optional, List import yaml from pathlib import Path @@ -8,14 +11,102 @@ from dotenv import load_dotenv from living_agents import CharacterAgent, PromptManager import logging +from llm_connector import LLMMessage + logger = logging.getLogger(__name__) +def save_character_to_cache(agent: CharacterAgent, template_path: Path): + """Save a character agent to cache""" + cache_path = Path('.cache') / f"{template_path.stem}.pkl" + + try: + # Create cache data with metadata + cache_data = { + 'character': agent.character, + 'memories': agent.memory_stream.memories, + 'template_path': template_path, + 'cached_at': datetime.now(), + 'cache_version': '1.0' + } + + with open(cache_path, 'wb') as f: + pickle.dump(cache_data, f) + + print(f"šŸ’¾ Character saved to cache: {cache_path.name}") + + except Exception as e: + print(f"āš ļø Warning: Could not save character to cache: {e}") + + +def load_character_from_cache(template_path: Path) -> Optional[CharacterAgent]: + """Load a character agent from cache, return None if not found or invalid""" + cache_path = Path('.cache') / f"{template_path.stem}.pkl" + + if not cache_path.exists(): + return None + + try: + with open(cache_path, 'rb') as f: + cache_data = pickle.load(f) + + # Verify cache data structure + required_keys = ['character', 'memories', 'template_path', 'cached_at'] + if not all(key in cache_data for key in required_keys): + print(f"āš ļø Invalid cache file format: {cache_path.name}") + return None + + # Check if template file is newer than cache + template_mtime = Path(template_path).stat().st_mtime + cache_time = cache_data['cached_at'].timestamp() + + if template_mtime > cache_time: + print(f"šŸ“ Template file is newer than cache, will recreate character") + return None + + # Reconstruct CharacterAgent + from living_agents import LLMAgent + agent = CharacterAgent(cache_data['character'], LLMAgent()) + agent.memory_stream.memories = cache_data['memories'] + + cached_date = cache_data['cached_at'].strftime('%Y-%m-%d %H:%M:%S') + print(f"šŸ’¾ Loaded character from cache (created: {cached_date})") + + return agent + + except Exception as e: + print(f"āš ļø Could not load character from cache: {e}") + return None + + +def get_cache_info(template_path: Path) -> Optional[dict]: + """Get information about cached character""" + cache_path = Path('.cache') / f"{template_path.stem}.pkl" + + if not cache_path.exists(): + return None + + try: + with open(cache_path, 'rb') as f: + cache_data = pickle.load(f) + + return { + 'cache_file': cache_path.name, + 'cached_at': cache_data['cached_at'], + 'character_name': cache_data['character'].name, + 'memory_count': len(cache_data['memories']) + } + except: + return None + + class CharacterExplorer: """Interactive explorer for testing CharacterAgent functionality""" + chat_history: List[LLMMessage] def __init__(self, character_agent: CharacterAgent): self.agent = character_agent + self.chat_history = [] async def start_interactive_session(self): """Start the interactive exploration menu""" @@ -27,23 +118,29 @@ class CharacterExplorer: self._show_menu() try: - choice = input("\nChoose option (1-6): ").strip() + choice = input("\nChoose option (1-8): ").strip() if choice == "1": await self._handle_ask_question() - elif choice == "2": - await self._handle_list_memories() + if choice == "2": + await self._handle_chat() elif choice == "3": - await self._handle_view_memory() + await self._handle_list_memories() elif choice == "4": - await self._handle_memory_stats() + await self._handle_view_memory() elif choice == "5": - await self._handle_character_summary() + await self._handle_memory_stats() elif choice == "6": + await self._handle_character_summary() + elif choice == "7": + # Regenerate character + if await self._handle_regenerate_character(): + break # Exit to main menu after regeneration + elif choice == "8": print("šŸ‘‹ Goodbye!") break else: - print("āŒ Invalid choice. Please enter 1-6.") + print("āŒ Invalid choice. Please enter 1-8.") except KeyboardInterrupt: print("\nšŸ‘‹ Goodbye!") @@ -51,16 +148,31 @@ class CharacterExplorer: except Exception as e: print(f"āŒ Error: {e}") + def clear_character_cache(self) -> bool: + """Clear cached character for a template""" + cache_file = Path('.cache') / f"{self.agent.character.template_file.stem}.pkl" + + if cache_file.exists(): + try: + cache_file.unlink() + return True + except Exception as e: + print(f"āŒ Could not clear cache: {e}") + return False + return False + def _show_menu(self): """Display the interactive menu options""" print(f"\nšŸŽ­ Character Explorer Menu") print("=" * 30) - print("1. šŸ’¬ Ask a question") - print("2. šŸ“š List all memories") - print("3. šŸ” View specific memory (with related)") - print("4. šŸ“Š Memory statistics") - print("5. šŸ‘¤ Character summary") - print("6. 🚪 Exit") + print(f"1. šŸ’¬ Ask {self.agent.character.name} a question") + print(f"2. šŸ’¬ Chat with {self.agent.character.name}") + print("3. šŸ“š List all memories") + print("4. šŸ” View specific memory (with related)") + print("5. šŸ“Š Memory statistics") + print("6. šŸ‘¤ Character summary") + print("7. šŸ”„ Regenerate character") + print("8. 🚪 Exit") async def _handle_ask_question(self): """Handle asking questions to the character""" @@ -82,6 +194,33 @@ class CharacterExplorer: except Exception as e: print(f"āŒ Error getting response: {e}") + async def _handle_chat(self): + """Handle chat with a character""" + if not self.chat_history: + scenario_content = input(f"\nšŸ’¬ Set a scenario for a new chat with {self.agent.character.name}: ").strip() + if not scenario_content: + return + + self.chat_history.append({'role': 'system', 'content': scenario_content}) + + scenario_content = input(f"\nšŸ’¬ Ask {self.agent.character.name}: ").strip() + if not scenario_content: + return + + print(f"\nšŸ¤” {self.agent.character.name} is thinking...") + try: + response = await self.agent.react_to_situation(scenario_content) + print(f"šŸ’¬ {self.agent.character.name}: {response}") + + # Show which memories were retrieved for this response + relevant_memories = await self.agent.memory_stream.retrieve_related_memories(scenario_content, k=5) + print(f"\n🧠 Memories used for this response:") + for i, mem in enumerate(relevant_memories, 1): + print(f" {i}. [{mem.memory_type}] {mem.description[:80]}{'...' if len(mem.description) > 80 else ''}") + + except Exception as e: + print(f"āŒ Error getting response: {e}") + async def _handle_list_memories(self): """List all memories with filtering options""" print("\nšŸ“š Memory Filter Options:") @@ -204,6 +343,7 @@ class CharacterExplorer: print(f" Occupation: {self.agent.character.occupation}") print(f" Location: {self.agent.character.location}") print("") + print(f" - File: {self.agent.character.template_file}") print(f" Traits") traits_summary = "\n".join([f" - {trait.strength}/10 {trait.name} ({trait.description})" for trait in self.agent.character.traits]) if self.agent.character.traits else "No traits yet." print(traits_summary) @@ -220,47 +360,150 @@ class CharacterExplorer: except Exception as e: print(f"āŒ Error generating summary: {e}") + async def _handle_regenerate_character(self) -> bool: + """Handle regenerating the current character""" + character_name = self.agent.character.name -async def load_and_explore_character(template_path: str): + print(f"\nšŸ”„ Regenerate {character_name}") + print("This will clear the cache and recreate the character from scratch.") + print("All current memories and traits will be lost!") + + # Confirm regeneration + confirm = input(f"Regenerate {character_name}? (y/N): ").strip().lower() + if confirm not in ['y', 'yes']: + print("āŒ Regeneration cancelled") + return False + + template_file = self.agent.character.template_file + if self.agent.character.template_file is None: + print(f"āŒ Could not find template file for {character_name}") + return False + + # Clear the cache + if self.clear_character_cache(): + print(f"šŸ—‘ļø Cleared cache for {character_name}") + else: + print("āŒ Failed to clear cache") + return False + + await load_and_explore_character(template_file) + return True + + +def select_character_template(show_cache_info: bool = True) -> Path: + """Show character template selection menu and return chosen template path""" + + templates_dir = Path(__file__).parent / 'character_templates' + + # Find all .yml files in the character_templates directory + template_files = list(templates_dir.glob('*.yml')) + + if not template_files: + print("āŒ No character templates found in character_templates/") + return None + + print("\nšŸŽ­ Available Character Templates:") + print("=" * 60) + + # Display template options with cache info + for i, template_file in enumerate(template_files, 1): + # Try to load the template to get the name + try: + with open(template_file, 'r', encoding='utf-8') as f: + template_data = yaml.safe_load(f) + character_name = template_data.get('name', template_file.stem) + except: + character_name = template_file.stem + + # Check for cached version if requested + if show_cache_info: + cache_info = get_cache_info(template_file) + if cache_info: + cache_status = f"šŸ’¾ (cached {cache_info['cached_at'].strftime('%m/%d %H:%M')})" + else: + cache_status = "šŸ”„ (will create new)" + print(f"{i}. {character_name} ({template_file.name}) {cache_status}") + else: + print(f"{i}. {character_name} ({template_file.name})") + + print(f"{len(template_files) + 1}. 🚪 Exit") + + # Get user choice + while True: + try: + choice = input(f"\nChoose a character template (1-{len(template_files) + 1}): ").strip() + choice_num = int(choice) + + if choice_num == len(template_files) + 1: + return None # Exit choice + elif 1 <= choice_num <= len(template_files): + return template_files[choice_num - 1] + else: + print(f"āŒ Invalid choice. Please enter 1-{len(template_files) + 1}.") + except ValueError: + print(f"āŒ Please enter a valid number (1-{len(template_files) + 1}).") + except KeyboardInterrupt: + return None + + +async def load_and_explore_character(template_path: Path): """Load a character template and start exploration""" # Load environment - # env_path = Path(__file__).parent.parent / '.env' load_dotenv() - # Load template - if not Path(template_path).exists(): - print(f"āŒ Template file not found: {template_path}") - return None + # Try to load from cache first + print(f"šŸ“ Loading character from {template_path.name}...") + agent = load_character_from_cache(template_path) - try: - with open(template_path, 'r', encoding='utf-8') as f: - template = yaml.safe_load(f) - except Exception as e: - print(f"āŒ Error loading template: {e}") - return None + if agent is None: + # No cache or cache invalid, create from template + try: + with open(template_path, 'r', encoding='utf-8') as f: + template = yaml.safe_load(f) + template['yaml_file'] = template_path + except Exception as e: + print(f"āŒ Error loading template: {e}") + return None - print(f"šŸ“ Loading character from {template_path}...") - print(f"šŸ¤– Creating memories and scoring importance...") + print(f"šŸ¤– Creating memories and scoring importance...") - try: - # Create character agent - agent = await CharacterAgent.create_from_template(template) + try: + # Create character agent from template + agent = await CharacterAgent.create_from_template(template) + print(f"āœ… {agent.character.name} created successfully!") + + # Save to cache for next time + save_character_to_cache(agent, template_path) + + except Exception as e: + print(f"āŒ Error creating character: {e}") + import traceback + traceback.print_exc() + return None + else: print(f"āœ… {agent.character.name} loaded successfully!") - # Start explorer - explorer = CharacterExplorer(agent) - await explorer.start_interactive_session() + # Start explorer + explorer = CharacterExplorer(agent) + await explorer.start_interactive_session() - return agent + return agent - except Exception as e: - print(f"āŒ Error creating character: {e}") - import traceback - traceback.print_exc() - return None + +async def main(): + """Main function with character selection""" + print("šŸŽ­ Welcome to the Living Agents Character Explorer!") + + selected_template = select_character_template(show_cache_info=True) + + if selected_template is None: + print("šŸ‘‹ Goodbye!") + return + + # Load and explore the selected character + await load_and_explore_character(selected_template) if __name__ == '__main__': - template = Path(__file__).parent / 'character_templates' / 'Alice.yml' - asyncio.run(load_and_explore_character(str(template))) + asyncio.run(main()) diff --git a/living_agents/character_agent.py b/living_agents/character_agent.py index 0653070..d6e916e 100644 --- a/living_agents/character_agent.py +++ b/living_agents/character_agent.py @@ -71,6 +71,33 @@ class CharacterAgent: return response + async def chat_with(self, messages: List[LLMMessage]) -> LLMMessage: + """Generate reaction based on memory and character""" + # Retrieve relevant memories + relevant_memories = await self.memory_stream.retrieve_related_memories(situation, k=8) + memory_context = "\n".join([f"- {m.description}" for m in relevant_memories]) + + context = { + 'character': self._get_character_prompt(), + 'character_name': self.character.name, + 'memory_context': memory_context, + 'situation': situation} + prompt = PromptManager.get_prompt('react_to_situation', context) + + response = await self.llm.chat([{"role": "user", "content": prompt}]) + + # create new memories from interaction + interaction_context = { + 'situation': f'I reacted to: \n{situation}', + 'response': f'My response was: \n{response}', + } + prompt, schema = PromptManager.get_prompt_with_schema('extract_interaction_memories', interaction_context) + memories_response = await self.llm.client.get_structured_response([{"role": "user", "content": prompt}], schema) + for new_memory in memories_response['memories']: + await self.perceive(new_memory) + + return response + async def plan_day(self) -> List[str]: """Generate high-level daily plan""" # Retrieve relevant memories about goals, habits, schedule @@ -255,5 +282,7 @@ Summary:""" memory.importance_score = await instance._score_memory_importance(memory) await instance._analyze_trait_impact(memory) + instance.character.template_file = template['yaml_file'] + logger.info(f"Character {instance.character.name} created successfully") return instance diff --git a/living_agents/datatypes.py b/living_agents/datatypes.py index 85f794b..ae14c26 100644 --- a/living_agents/datatypes.py +++ b/living_agents/datatypes.py @@ -2,7 +2,7 @@ from dataclasses import dataclass, field from typing import Dict, List, Optional, Literal, TypedDict from datetime import datetime from uuid import uuid4 -import random +from pathlib import Path class CharacterTemplate(TypedDict): @@ -10,7 +10,7 @@ class CharacterTemplate(TypedDict): observations: List[str] reflections: List[str] plans: List[str] - yaml_file: str + yaml_file: Path @dataclass @@ -72,6 +72,7 @@ class Character: traits: List[CharacterTrait] = field(default_factory=list) relationships: Dict[str, str] = field(default_factory=dict) goals: List[str] = field(default_factory=list) + template_file: Optional[Path] = None _id: str = field(default_factory=lambda: str(uuid4())[:8]) def get_trait(self, trait_name, trait_description) -> CharacterTrait: diff --git a/living_agents/memory_stream.py b/living_agents/memory_stream.py index 3850e14..94d46dd 100644 --- a/living_agents/memory_stream.py +++ b/living_agents/memory_stream.py @@ -41,7 +41,6 @@ class MemoryStream: # Track for reflection trigger self.recent_importance_sum += memory.importance_score - print(f"Recent Importance Sum: {self.recent_importance_sum}") # Trigger reflection if threshold exceeded if self.recent_importance_sum >= self.importance_threshold: diff --git a/living_agents/prompts/chat_with_agent.md b/living_agents/prompts/chat_with_agent.md new file mode 100644 index 0000000..7daa62e --- /dev/null +++ b/living_agents/prompts/chat_with_agent.md @@ -0,0 +1,15 @@ +{{character}} + +Relevant memories from your past: +{{memory_context}} + +Respond as {{character_name}} describing what you did and how you reacted. Write in first person past tense as if this +just happened to you. + +Examples of the response style: + +- "I looked up from my book and smiled nervously..." +- "I felt my heart race and took a deep breath before I said..." +- "I hesitated for a moment, then decided to..." + +Stay completely in character and be specific about your actions, thoughts, and words. \ No newline at end of file