6.Integrate a Memory

uv pip install upsonic sqlalchemy aiosqlite
# pip install upsonic sqlalchemy aiosqlite

# Upsonic Docs: Using Memory in your Agent
# https://docs.upsonic.ai/guides/using_memory_in_your_agent

# Imports
from upsonic import Agent
from upsonic import Task
from upsonic.storage import InMemoryStorage, Memory

# Create a storage provider
storage = InMemoryStorage()

# Create a Memory instance with all memory types enabled
memory = Memory(
    storage=storage,
    session_id="user_session_123",
    user_id="user_456",
    full_session_memory=True,      # Remember complete conversations
    summary_memory=True,           # Maintain conversation summaries
    user_analysis_memory=True,     # Build user profiles
    dynamic_user_profile=True,     # Automatically adapt profile schema
    num_last_messages=10,          # Limit context to last 10 messages
    debug=True                     # Enable debug logging
)

# Create an Agent with Memory
personal_assistant = Agent(
    model="anthropic/claude-sonnet-4-5",
    name="Personal Assistant",
    role="Personal AI Assistant with Memory",
    goal="Provide personalized assistance by remembering user preferences and conversation history",
    instructions="""
    Always reference previous conversations when relevant.
    Adapt your responses based on the user's profile and preferences.
    Use the conversation summary to provide context-aware assistance.
    """,
    memory=memory
)

# Example tasks that demonstrate memory capabilities
task1 = Task(
    description="I prefer to be called 'Alex' and I'm interested in machine learning",
)

task2 = Task(
    description="What did we discuss in our previous conversation?",
)

task3 = Task(
    description="Based on what you know about me, suggest a learning path for AI",
)

# Execute tasks to build memory
print("=== Task 1: Building User Profile ===")
result1 = personal_assistant.print_do(task1)
print(f"Response: {result1}")

print("\n=== Task 2: Testing Memory Recall ===")
result2 = personal_assistant.print_do(task2)
print(f"Response: {result2}")

print("\n=== Task 3: Using Memory for Personalization ===")
result3 = personal_assistant.print_do(task3)
print(f"Response: {result3}")

# Check memory statistics
print(f"\n=== Memory Statistics ===")
print(f"Session ID: {memory.session_id}")
print(f"User ID: {memory.user_id}")
print(f"Full Session Memory: {memory.full_session_memory_enabled}")
print(f"Summary Memory: {memory.summary_memory_enabled}")
print(f"User Analysis Memory: {memory.user_analysis_memory_enabled}")

from upsonic.storage import PostgresStorage

postgres_storage = PostgresStorage(
    db_url="postgresql://user:pass@localhost:5432/upsonic_db",
    db_schema="public",
    session_table="agent_sessions",
    user_memory_table="user_profiles"
)

from upsonic.storage import RedisStorage

# Redis Storage for High Performance
redis_storage = RedisStorage(
    db_url="redis://localhost:6379/0",
    db_prefix="upsonic",
    expire=86400  # 24 hours TTL
)

from upsonic.storage import JSONStorage

# JSON Storage for Simple Applications
json_storage = JSONStorage(
    db_path="./data/memory"
)

from pydantic import BaseModel, Field
from upsonic import Agent, Task
from upsonic.storage import Memory
from upsonic.storage.json import JSONStorage

class CustomUserProfile(BaseModel):
    expertise_level: str = Field(description="User's expertise level in the domain")
    preferred_language: str = Field(description="User's preferred programming language")
    learning_goals: list[str] = Field(description="User's learning objectives")
    communication_style: str = Field(description="User's preferred communication style")

# Initialize JSON storage for custom profile
json_storage = JSONStorage("./memory_data")

# Create Memory with custom profile schema
custom_memory = Memory(
    storage=json_storage,
    session_id="session_789",
    user_id="user_101",
    user_analysis_memory=True,
    user_profile_schema=CustomUserProfile,
    dynamic_user_profile=False,  # Use our custom schema instead of dynamic
    model="anthropic/claude-sonnet-4-5"
)

# Create agent with custom memory
profile_agent = Agent(
    model="anthropic/claude-sonnet-4-5",
    name="Profile Builder",
    role="User Profile Specialist",
    memory=custom_memory
)

# Build user profile with structured data
profile_task = Task(
    description="I'm an intermediate Python developer. I prefer Python and want to learn advanced async programming and system design. Please communicate in a technical but friendly way."
)

result = profile_agent.print_do(profile_task)

# The agent now has a structured understanding of the user based on CustomUserProfile schema

memory_with_model = Memory(
    storage=storage,
    session_id="session_456",
    user_id="user_789",
    summary_memory=True,
    user_analysis_memory=True,
    model="anthropic/claude-sonnet-4-5",  # For generating summaries and analyzing traits
    debug=True
)

# Create unique session IDs for different conversation contexts
import uuid

session_id = f"chat_session_{uuid.uuid4()}"
user_id = f"user_{uuid.uuid4()}"

memory = Memory(
    storage=storage,
    session_id=session_id,
    user_id=user_id,
    full_session_memory=True
)

# Check memory statistics
print(f"Memory Configuration:")
print(f"- Full Session Memory: {memory.full_session_memory_enabled}")
print(f"- Summary Memory: {memory.summary_memory_enabled}")
print(f"- User Analysis Memory: {memory.user_analysis_memory_enabled}")
print(f"- Max Messages: {memory.num_last_messages}")
print(f"- Feed Tool Results: {memory.feed_tool_call_results}")

from upsonic import Team, Agent, Task
from upsonic.storage import Memory, InMemoryStorage

# Initialize storage
storage = InMemoryStorage()

# Create a shared memory instance
shared_memory = Memory(
    storage=storage,
    session_id="team_session_123",
    user_id="user_456",
    full_session_memory=True,
    summary_memory=True,
    model="anthropic/claude-sonnet-4-5"  # Required for summary generation
)

# Define tasks for the team
research_task = Task(
    description="Research the latest trends in AI and machine learning for 2024"
)

writing_task = Task(
    description="Write a comprehensive article based on the research findings"
)

# Create agents with shared memory
research_agent = Agent(
    model="anthropic/claude-sonnet-4-5",
    name="Research Agent",
    role="Research Specialist",
    goal="Conduct thorough research on AI trends",
    memory=shared_memory
)

writing_agent = Agent(
    model="anthropic/claude-sonnet-4-5",
    name="Writing Agent", 
    role="Content Writer",
    goal="Create engaging content from research",
    memory=shared_memory
)

# Create a team with shared memory
research_team = Team(
    agents=[research_agent, writing_agent],
    memory=shared_memory,
    tasks=[research_task, writing_task]
)

# Execute team tasks
print("=== Executing Team Tasks with Shared Memory ===")
result = research_team.print_do()

# Both agents can now access the shared conversation history
print(f"\n=== Memory Statistics ===")
print(f"Session ID: {shared_memory.session_id}")
print(f"User ID: {shared_memory.user_id}")
print(f"Full Session Memory: {shared_memory.full_session_memory_enabled}")
print(f"Summary Memory: {shared_memory.summary_memory_enabled}")

memory = Memory(
    storage=storage,
    session_id="debug_session",
    user_id="debug_user",
    full_session_memory=True,
    debug=True  # Enable detailed logging
)