05_test_memory.py

#!/usr/bin/env python3
"""
Script to test memory retrieval from AgentCore Memory.

This script retrieves memories for user_001 from the preferences namespace
to show what the agent remembers about the customer.
"""

import json

try:
    from bedrock_agentcore.memory import MemoryClient
except ImportError:
    print("✗ Error: bedrock_agentcore package not found")
    print("  Install with: pip install bedrock-agentcore")
    exit(1)

# Load memory_id from config
with open('memory_config.json') as f:
    config = json.load(f)
    memory_id = config['memory_id']

print("=" * 70)
print("AGENTCORE MEMORY RETRIEVAL TEST")
print("=" * 70)
print(f"Memory ID: {memory_id}")
print(f"Customer: user_001")
print(f"Region: us-west-2")
print(f"Namespace: app/user_001/preferences")
print(f"Search Query: 'customer preferences and communication'")
print("=" * 70)
print()

# Create memory client
memory_client = MemoryClient(region_name='us-west-2')

try:
    # Retrieve memories from preferences namespace
    print("🔍 Searching preferences namespace...")
    memories = memory_client.retrieve_memories(
        memory_id=memory_id,
        namespace="app/user_001/preferences",
        query="customer preferences and communication",
        top_k=3
    )
    
    if memories:
        print(f"✓ Retrieved {len(memories)} preference(s)\n")
        
        print("=" * 70)
        print("WHAT THE AGENT REMEMBERS ABOUT THIS CUSTOMER")
        print("=" * 70)
        
        for i, memory in enumerate(memories, 1):
            print(f"\n📌 Preference {i}:")
            print("─" * 70)
            
            # Extract content
            content = memory.get('content', {})
            if isinstance(content, dict):
                text = content.get('text', 'N/A')
            else:
                text = str(content)
            
            print(f"Content: {text}")
            
            # Show relevance score
            relevance = memory.get('relevanceScore')
            if relevance is not None:
                print(f"Relevance Score: {relevance:.3f}")
            
            # Show metadata if available
            metadata = memory.get('metadata', {})
            if metadata:
                print(f"Metadata: {json.dumps(metadata, indent=2)}")
        
        print("\n" + "=" * 70)
        print("SUMMARY")
        print("=" * 70)
        print(f"✓ Agent has {len(memories)} preference(s) stored for user_001")
        print("✓ These preferences will be used to personalize future interactions")
        print("✓ Expected: Email notification preference should be captured")
        
    else:
        print("⚠️  No memories found in preferences namespace")
        print("\nPossible reasons:")
        print("  1. Memory extraction is still processing (takes 20-30 seconds)")
        print("  2. No preferences were extracted from the conversations")
        print("  3. The search query didn't match stored preferences")
        print("\n💡 Tip: Try running 04_seed_memory.py again and wait longer")
    
    # Also check semantic namespace for facts
    print("\n" + "=" * 70)
    print("CHECKING SEMANTIC NAMESPACE (FACTS)")
    print("=" * 70)
    
    semantic_memories = memory_client.retrieve_memories(
        memory_id=memory_id,
        namespace="app/user_001/semantic",
        query="laptop return defective",
        top_k=3
    )
    
    if semantic_memories:
        print(f"✓ Retrieved {len(semantic_memories)} fact(s)\n")
        
        for i, memory in enumerate(semantic_memories, 1):
            print(f"\n📌 Fact {i}:")
            print("─" * 70)
            
            content = memory.get('content', {})
            if isinstance(content, dict):
                text = content.get('text', 'N/A')
            else:
                text = str(content)
            
            print(f"Content: {text}")
            
            relevance = memory.get('relevanceScore')
            if relevance is not None:
                print(f"Relevance Score: {relevance:.3f}")
    else:
        print("⚠️  No facts found in semantic namespace yet")
        
except Exception as e:
    print(f"❌ Error retrieving memories: {e}")
    import traceback
    traceback.print_exc()
    exit(1)

print("\n" + "=" * 70)
print("✓ Memory retrieval test complete!")
print("=" * 70)