RDS Long-Term Memory

#!/usr/bin/env python3
"""
Basic Mem0 SDK test script
Performs create, read, update, and delete operations using MemoryClient

Usage:
    python test_mem0_sdk_basic.py
"""
import os
import json
import time
from mem0 import MemoryClient

# ============================================

def main():
    # Load configuration from environment variables
    api_key = os.environ.get("MEM0_API_KEY")
    host = os.environ.get("MEM0_HOST")

    print("=" * 60)
    print("Mem0 SDK Basic Functionality Test")
    print("=" * 60)
    print(f"Service endpoint: {host}")

    # Initialize client
    client = MemoryClient(host=host, api_key=api_key)

    # Test user
    user_id = f"sdk_test_user_{int(time.time())}"
    print(f"Test user: {user_id}")
    print("=" * 60)

    memory_id = None

    try:
        # =================== 1. Add memory ===================
        print("\n[1] Add memory")
        print("-" * 40)

        messages = [
            {"role": "user", "content": "My name is Zhang San. I work in Beijing and enjoy playing basketball."},
            {"role": "assistant", "content": "Hello Zhang San! Beijing is a great city, and basketball is a fantastic sport."},
        ]

        result = client.add(messages, user_id=user_id)
        print(f"Add result: {json.dumps(result, ensure_ascii=False, indent=2)}")

        # Wait for processing to complete
        time.sleep(2)

        # =================== 2. Get all memories ===================
        print("\n[2] Get all memories")
        print("-" * 40)

        all_memories = client.get_all(user_id=user_id)
        print(f"Memory list: {json.dumps(all_memories, ensure_ascii=False, indent=2)}")

        memories = all_memories.get("results", [])
        print(f"Total memories: {len(memories)}")

        if memories:
            memory_id = memories[0]["id"]
            print(f"First memory ID: {memory_id}")

        # =================== 3. Search memories ===================
        print("\n[3] Search memories")
        print("-" * 40)

        search_result = client.search("Where does Zhang San work?", user_id=user_id)
        print(f"Search result: {json.dumps(search_result, ensure_ascii=False, indent=2)}")

        # =================== 4. Get a single memory ===================
        print("\n[4] Get a single memory")
        print("-" * 40)

        if memory_id:
            single_memory = client.get(memory_id)
            print(f"Memory details: {json.dumps(single_memory, ensure_ascii=False, indent=2)}")
        else:
            print("No memory ID available")

        # =================== 5. Update memory ===================
        print("\n[5] Update memory")
        print("-" * 40)

        if memory_id:
            new_content = "Zhang San now works in Shanghai and still enjoys playing basketball."
            update_result = client.update(memory_id, new_content)
            print(f"Update result: {json.dumps(update_result, ensure_ascii=False, indent=2)}")
        else:
            print("No memory ID available")

        # =================== 6. Get memory history ===================
        print("\n[6] Get memory history")
        print("-" * 40)

        if memory_id:
            history = client.history(memory_id)
            print(f"History: {json.dumps(history, ensure_ascii=False, indent=2)}")
        else:
            print("No memory ID available")

        # =================== 7. Delete a single memory ===================
        print("\n[7] Delete a single memory")
        print("-" * 40)

        if memory_id:
            delete_result = client.delete(memory_id)
            print(f"Delete result: {json.dumps(delete_result, ensure_ascii=False, indent=2)}")
        else:
            print("No memory ID available")

    except Exception as e:
        print(f"\nError: {e}")
        import traceback
        traceback.print_exc()
        return 1

    return 0

if __name__ == "__main__":
    exit(main())

#!/usr/bin/env python3
"""
Mem0 SDK graph search test script
Performs create, read, update, and delete operations with graph-enabled search

Key features:
- Enable graph search with enable_graph=True
- Shows relations (graph relationships) in results
- Compares standard search vs. graph search

Usage:
    python current_file.py
"""

import os
import json
import time
from mem0 import MemoryClient

# ============================================
# Configuration
# ============================================
HOST = "http://<your-host>:80/memory"  # Service endpoint
API_KEY = "<your-api-key>"              # API key

# ============================================


def print_json(data, title=None):
    """Print JSON data in formatted form"""
    if title:
        print(f"\n{title}:")
    print(json.dumps(data, ensure_ascii=False, indent=2))


def print_relations(relations):
    """Print graph relationships in formatted form"""
    if not relations:
        print("  (No graph relationships)")
        return
    
    print(f"  Total relationships: {len(relations)}")
    for i, rel in enumerate(relations, 1):
        source = rel.get("source", "?")
        relationship = rel.get("relationship", "?")
        destination = rel.get("destination") or rel.get("target", "?")
        print(f"    [{i}] {source} --[{relationship}]--> {destination}")


def main():
    # Load configuration from environment variables
    api_key = os.environ.get("MEM0_API_KEY", API_KEY)
    host = os.environ.get("MEM0_HOST", HOST)
    
    print("=" * 70)
    print("Mem0 SDK Graph Search Test")
    print("=" * 70)
    print(f"Service endpoint: {host}")
    
    # Initialize client
    client = MemoryClient(host=host, api_key=api_key)
    
    # Test user
    user_id = "user_001"
    print(f"Test user: {user_id}")
    print("=" * 70)
    
    try:
        # =================== 1. Add multiple memories (build graph relationships) ===================
        print("\n[1] Add multiple memories (build graph relationships)")
        print("-" * 50)
        
        # First conversation: Personal info
        messages1 = [
            {"role": "user", "content": "My name is Zhang San. I work at Alibaba on AI platform development."},
            {"role": "assistant", "content": "Hello Zhang San! Alibaba's AI platform is highly influential."},
        ]
        result1 = client.add(messages1, user_id=user_id)
        print(f"Add result 1: {json.dumps(result1, ensure_ascii=False)[:200]}...")
        
        time.sleep(2)
        
        # Second conversation: Hobbies
        messages2 = [
            {"role": "user", "content": "I enjoy playing basketball and often play with Li Si on weekends."},
            {"role": "assistant", "content": "Basketball is a great sport—more fun with friends!"},
        ]
        result2 = client.add(messages2, user_id=user_id)
        print(f"Add result 2: {json.dumps(result2, ensure_ascii=False)[:200]}...")
        
        time.sleep(2)
        
        # Third conversation: Additional relationships
        messages3 = [
            {"role": "user", "content": "My boss is Manager Wang, head of the Data Intelligence department."},
            {"role": "assistant", "content": "Understood. Manager Wang leads the Data Intelligence department."},
        ]
        result3 = client.add(messages3, user_id=user_id)
        print(f"Add result 3: {json.dumps(result3, ensure_ascii=False)[:200]}...")
        
        # Wait for graph construction to complete
        print("\nWaiting for graph relationships to build...")
        time.sleep(5)
        
        # =================== 2. Get all memories (with graph) ===================
        print("\n[2] Get all memories (enable_graph=True)")
        print("-" * 50)
        
        all_memories = client.get_all(user_id=user_id, enable_graph=True)
        
        results = all_memories.get("results", [])
        relations = all_memories.get("relations", [])
        
        print(f"\nVector search results: {len(results)} memories")
        for i, mem in enumerate(results[:5], 1):
            print(f"  [{i}] {mem.get('memory', '')[:50]}...")
        
        print(f"\nGraph relationships:")
        print_relations(relations)
        
        # =================== 3. Graph search vs. standard search ===================
        print("\n[3] Graph search vs. standard search comparison")
        print("-" * 50)
        
        query = "Where does Zhang San work?"
        print(f"Search query: {query}")
        
        # 3.1 Standard search (graph disabled)
        print("\n【Standard search】enable_graph=False:")
        search_normal = client.search(query, user_id=user_id, enable_graph=False)
        normal_results = search_normal.get("results", [])
        normal_relations = search_normal.get("relations")
        
        print(f"  Vector results: {len(normal_results)}")
        for i, mem in enumerate(normal_results[:3], 1):
            score = mem.get("score", "N/A")
            memory = mem.get("memory", "")[:40]
            print(f"    [{i}] score={score:.4f} | {memory}...")
        print(f"  Graph relationships: {normal_relations}")
        
        # 3.2 Graph search (graph enabled)
        print("\n【Graph search】enable_graph=True:")
        search_graph = client.search(query, user_id=user_id, enable_graph=True)
        graph_results = search_graph.get("results", [])
        graph_relations = search_graph.get("relations", [])
        
        print(f"  Vector results: {len(graph_results)}")
        for i, mem in enumerate(graph_results[:3], 1):
            score = mem.get("score", "N/A")
            memory = mem.get("memory", "")[:40]
            print(f"    [{i}] score={score:.4f} | {memory}...")
        
        print(f"  Graph relationships:")
        print_relations(graph_relations)
        
        # =================== 4. Multi-keyword graph search ===================
        print("\n[4] Multi-keyword graph search")
        print("-" * 50)
        
        queries = [
            "Relationship between Zhang San and Li Si",
            "Who is Manager Wang?",
            "Basketball-related activities",
        ]
        
        for query in queries:
            print(f"\nQuery: {query}")
            result = client.search(query, user_id=user_id, enable_graph=True)
            
            results = result.get("results", [])
            relations = result.get("relations", [])
            
            print(f"  Vector results: {len(results)}")
            if results:
                print(f"    Top1: {results[0].get('memory', '')[:50]}...")
            
            print(f"  Graph relationships:")
            print_relations(relations)
        
    except Exception as e:
        print(f"\nError: {e}")
        import traceback
        traceback.print_exc()
    return 0


if __name__ == "__main__":
    exit(main())