too much
This commit is contained in:
277
character_explorer.py
Normal file
277
character_explorer.py
Normal file
@@ -0,0 +1,277 @@
|
||||
import asyncio
|
||||
import os
|
||||
from pprint import pprint
|
||||
|
||||
import yaml
|
||||
from pathlib import Path
|
||||
from dotenv import load_dotenv
|
||||
from living_agents import CharacterAgent, PromptManager
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class CharacterExplorer:
|
||||
"""Interactive explorer for testing CharacterAgent functionality"""
|
||||
|
||||
def __init__(self, character_agent: CharacterAgent):
|
||||
self.agent = character_agent
|
||||
|
||||
async def start_interactive_session(self):
|
||||
"""Start the interactive exploration menu"""
|
||||
|
||||
print(f"\n🎭 Character Explorer - {self.agent.character.name}")
|
||||
print("=" * 50)
|
||||
|
||||
while True:
|
||||
self._show_menu()
|
||||
|
||||
try:
|
||||
choice = input("\nChoose option (1-6): ").strip()
|
||||
|
||||
if choice == "1":
|
||||
await self._handle_ask_question()
|
||||
elif choice == "2":
|
||||
await self._handle_list_memories()
|
||||
elif choice == "3":
|
||||
await self._handle_view_memory()
|
||||
elif choice == "4":
|
||||
await self._handle_memory_stats()
|
||||
elif choice == "5":
|
||||
await self._handle_character_summary()
|
||||
elif choice == "6":
|
||||
print("👋 Goodbye!")
|
||||
break
|
||||
else:
|
||||
print("❌ Invalid choice. Please enter 1-6.")
|
||||
|
||||
except KeyboardInterrupt:
|
||||
print("\n👋 Goodbye!")
|
||||
break
|
||||
except Exception as e:
|
||||
print(f"❌ Error: {e}")
|
||||
|
||||
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")
|
||||
|
||||
async def _handle_ask_question(self):
|
||||
"""Handle asking questions to the character"""
|
||||
question = input(f"\n💬 Ask {self.agent.character.name}: ").strip()
|
||||
if not question:
|
||||
return
|
||||
|
||||
print(f"\n🤔 {self.agent.character.name} is thinking...")
|
||||
try:
|
||||
response = await self.agent.react_to_situation(question)
|
||||
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(question, 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:")
|
||||
print("1. All memories")
|
||||
print("2. Observations only")
|
||||
print("3. Reflections only")
|
||||
print("4. Plans only")
|
||||
print("5. By importance (high to low)")
|
||||
print("6. By recency (newest first)")
|
||||
|
||||
filter_choice = input("Choose filter (1-6): ").strip()
|
||||
|
||||
memories = self.agent.memory_stream.memories.copy()
|
||||
|
||||
if filter_choice == "2":
|
||||
memories = [m for m in memories if m.memory_type == "observation"]
|
||||
title = "Observations"
|
||||
elif filter_choice == "3":
|
||||
memories = [m for m in memories if m.memory_type == "reflection"]
|
||||
title = "Reflections"
|
||||
elif filter_choice == "4":
|
||||
memories = [m for m in memories if m.memory_type == "plan"]
|
||||
title = "Plans"
|
||||
elif filter_choice == "5":
|
||||
memories = sorted(memories, key=lambda m: m.importance_score, reverse=True)
|
||||
title = "All Memories (by importance)"
|
||||
elif filter_choice == "6":
|
||||
memories = sorted(memories, key=lambda m: m.creation_time, reverse=True)
|
||||
title = "All Memories (by recency)"
|
||||
else:
|
||||
title = "All Memories"
|
||||
|
||||
print(f"\n📋 {title} ({len(memories)} total):")
|
||||
for i, memory in enumerate(memories, 1):
|
||||
age_hours = (memory.last_accessed - memory.creation_time).total_seconds() / 3600
|
||||
print(
|
||||
f"{i:3d}. [{memory.memory_type[:4]}] [imp:{memory.importance_score}] [age:{age_hours:.1f}h] {memory.description}")
|
||||
|
||||
if len(memories) > 20:
|
||||
print(f"\n... showing first 20 of {len(memories)} memories")
|
||||
print("💡 Tip: Use option 3 to view specific memories in detail")
|
||||
|
||||
async def _handle_view_memory(self):
|
||||
"""View a specific memory with its related memories"""
|
||||
try:
|
||||
memory_num = int(input(f"\nEnter memory number (1-{len(self.agent.memory_stream.memories)}): ").strip())
|
||||
|
||||
if 1 <= memory_num <= len(self.agent.memory_stream.memories):
|
||||
memory = self.agent.memory_stream.memories[memory_num - 1]
|
||||
|
||||
print(f"\n🔍 Memory #{memory_num} Details:")
|
||||
print(f" Type: {memory.memory_type}")
|
||||
print(f" Importance: {memory.importance_score}/10")
|
||||
print(f" Created: {memory.creation_time.strftime('%Y-%m-%d %H:%M:%S')}")
|
||||
print(f" Last accessed: {memory.last_accessed.strftime('%Y-%m-%d %H:%M:%S')}")
|
||||
print(f" Description: {memory.description}")
|
||||
|
||||
# Show related memories using embeddings
|
||||
print(f"\n🔗 Related memories (by similarity):")
|
||||
try:
|
||||
related = await self.agent._get_related_memories_for_scoring(
|
||||
memory.description,
|
||||
exclude_self=memory,
|
||||
k=5
|
||||
)
|
||||
for i, rel_mem in enumerate(related, 1):
|
||||
rel_index = self.agent.memory_stream.memories.index(rel_mem) + 1
|
||||
print(
|
||||
f" {i}. [#{rel_index}] [{rel_mem.memory_type}] {rel_mem.description[:70]}{'...' if len(rel_mem.description) > 70 else ''}")
|
||||
|
||||
if not related:
|
||||
print(" (No related memories found)")
|
||||
|
||||
except Exception as e:
|
||||
print(f" ❌ Error finding related memories: {e}")
|
||||
|
||||
else:
|
||||
print(f"❌ Invalid memory number. Range: 1-{len(self.agent.memory_stream.memories)}")
|
||||
|
||||
except ValueError:
|
||||
print("❌ Please enter a valid number")
|
||||
|
||||
async def _handle_memory_stats(self):
|
||||
"""Show detailed memory statistics"""
|
||||
memories = self.agent.memory_stream.memories
|
||||
|
||||
print(f"\n📊 Memory Statistics for {self.agent.character.name}:")
|
||||
print(f" Total memories: {len(memories)}")
|
||||
|
||||
# Breakdown by type
|
||||
by_type = {}
|
||||
for mem in memories:
|
||||
by_type[mem.memory_type] = by_type.get(mem.memory_type, 0) + 1
|
||||
|
||||
print(f"\n📂 Memory Types:")
|
||||
for mem_type, count in by_type.items():
|
||||
percentage = (count / len(memories)) * 100
|
||||
print(f" {mem_type.title()}: {count} ({percentage:.1f}%)")
|
||||
|
||||
# Importance distribution
|
||||
importance_scores = [m.importance_score for m in memories]
|
||||
if importance_scores:
|
||||
print(f"\n📈 Importance Distribution:")
|
||||
print(f" Average: {sum(importance_scores) / len(importance_scores):.1f}")
|
||||
print(f" Range: {min(importance_scores)}-{max(importance_scores)}")
|
||||
|
||||
# Visual histogram
|
||||
print(f" Score distribution:")
|
||||
for score in range(1, 11):
|
||||
count = importance_scores.count(score)
|
||||
bar = "█" * max(1, count // 2) if count > 0 else ""
|
||||
print(f" {score:2d}: {count:2d} {bar}")
|
||||
|
||||
# Recency info
|
||||
if memories:
|
||||
oldest = min(memories, key=lambda m: m.creation_time)
|
||||
newest = max(memories, key=lambda m: m.creation_time)
|
||||
print(f"\n⏰ Time Span:")
|
||||
print(f" Oldest: {oldest.creation_time.strftime('%Y-%m-%d %H:%M')}")
|
||||
print(f" Newest: {newest.creation_time.strftime('%Y-%m-%d %H:%M')}")
|
||||
|
||||
async def _handle_character_summary(self):
|
||||
"""Generate and show character summary"""
|
||||
print(f"\n👤 Generating summary for {self.agent.character.name}...")
|
||||
try:
|
||||
# summary = await self.agent.get_summary()
|
||||
print(f"\n📝 AI-Generated Character Summary:")
|
||||
# print(f" {summary}")
|
||||
|
||||
print(f"\n📋 Structured Character Data:")
|
||||
print(f" Name: {self.agent.character.name}")
|
||||
print(f" Age: {self.agent.character.age}")
|
||||
print(f" Personality: {self.agent.character.personality}")
|
||||
print(f" Occupation: {self.agent.character.occupation}")
|
||||
print(f" Location: {self.agent.character.location}")
|
||||
|
||||
if self.agent.character.relationships:
|
||||
print(f" Relationships:")
|
||||
for person, relationship in self.agent.character.relationships.items():
|
||||
print(f" • {person}: {relationship}")
|
||||
|
||||
if self.agent.character.goals:
|
||||
print(f" Goals:")
|
||||
for goal in self.agent.character.goals:
|
||||
print(f" • {goal}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Error generating summary: {e}")
|
||||
|
||||
|
||||
async def load_and_explore_character(template_path: str):
|
||||
"""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:
|
||||
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
|
||||
|
||||
print(f"📁 Loading character from {template_path}...")
|
||||
print(f"🤖 Creating memories and scoring importance...")
|
||||
|
||||
try:
|
||||
# Create character agent
|
||||
agent = await CharacterAgent.create_from_template(template)
|
||||
print(f"✅ {agent.character.name} loaded successfully!")
|
||||
|
||||
# Start explorer
|
||||
explorer = CharacterExplorer(agent)
|
||||
await explorer.start_interactive_session()
|
||||
|
||||
return agent
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Error creating character: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return None
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
template = Path(__file__).parent / 'character_templates' / 'Alice.yml'
|
||||
asyncio.run(load_and_explore_character(str(template)))
|
||||
Reference in New Issue
Block a user