Add Memory to AI Agent: Python Tutorial for Beginners

import json
import os
from datetime import datetime
from typing import Dict, List, Optional

class AgentMemory:
    """
    Memory system for AI agents with short-term and long-term storage.
    Short-term: In-memory conversation history for current session
    Long-term: Persistent JSON storage across sessions
    """
    
    def __init__(self, agent_id: str, memory_file: str = "agent_memory.json"):
        self.agent_id = agent_id
        self.memory_file = memory_file
        
        # Short-term memory: current conversation
        self.conversation_history: List[Dict] = []
        
        # Long-term memory: persistent user data
        self.long_term_memory: Dict = self._load_long_term_memory()
    
    def _load_long_term_memory(self) -> Dict:
        """Load persistent memory from disk or create new"""
        if os.path.exists(self.memory_file):
            with open(self.memory_file, 'r') as f:
                all_memories = json.load(f)
                return all_memories.get(self.agent_id, {
                    "user_preferences": {},
                    "facts": [],
                    "context": {}
                })
        return {"user_preferences": {}, "facts": [], "context": {}}
    
    def _save_long_term_memory(self):
        """Persist memory to disk"""
        all_memories = {}
        if os.path.exists(self.memory_file):
            with open(self.memory_file, 'r') as f:
                all_memories = json.load(f)
        
        all_memories[self.agent_id] = self.long_term_memory
        
        with open(self.memory_file, 'w') as f:
            json.dump(all_memories, f, indent=2)
    
    def add_to_conversation(self, role: str, content: str):
        """Add message to short-term conversation history"""
        self.conversation_history.append({
            "role": role,
            "content": content,
            "timestamp": datetime.now().isoformat()
        })
    
    def store_fact(self, fact: str):
        """Store important information in long-term memory"""
        if fact not in self.long_term_memory["facts"]:
            self.long_term_memory["facts"].append({
                "content": fact,
                "stored_at": datetime.now().isoformat()
            })
            self._save_long_term_memory()
    
    def update_preference(self, key: str, value: str):
        """Update user preferences in long-term memory"""
        self.long_term_memory["user_preferences"][key] = value
        self._save_long_term_memory()
    
    def get_context_for_llm(self, max_history: int = 10) -> str:
        """
        Format memory into context string for LLM prompts.
        Combines recent conversation + relevant long-term memory.
        """
        context_parts = []
        
        # Add long-term facts if available
        if self.long_term_memory["facts"]:
            facts_str = "\n".join([f"- {f['content']}" for f in self.long_term_memory["facts"][-5:]])
            context_parts.append(f"Relevant facts about the user:\n{facts_str}")
        
        # Add user preferences
        if self.long_term_memory["user_preferences"]:
            prefs_str = "\n".join([f"- {k}: {v}" for k, v in self.long_term_memory["user_preferences"].items()])
            context_parts.append(f"User preferences:\n{prefs_str}")
        
        # Add recent conversation history
        recent_messages = self.conversation_history[-max_history:]
        if recent_messages:
            conv_str = "\n".join([f"{m['role']}: {m['content']}" for m in recent_messages])
            context_parts.append(f"Recent conversation:\n{conv_str}")
        
        return "\n\n".join(context_parts)
    
    def clear_conversation(self):
        """Clear short-term memory (new conversation session)"""
        self.conversation_history = []

import anthropic
import os

class MemoryEnabledAgent:
    """
    AI Agent with conversation and long-term memory capabilities.
    Extends basic agent architecture with persistent context.
    """
    
    def __init__(self, agent_id: str = "default_agent"):
        self.client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
        self.memory = AgentMemory(agent_id)
        self.model = "claude-sonnet-4-20250514"
    
    def chat(self, user_message: str, store_facts: bool = True) -> str:
        """
        Process user message with full memory context.
        
        Args:
            user_message: The user's input
            store_facts: Whether to auto-extract facts for long-term storage
        
        Returns:
            Agent's response string
        """
        # Add user message to conversation history
        self.memory.add_to_conversation("user", user_message)
        
        # Build context-aware system prompt
        memory_context = self.memory.get_context_for_llm()
        system_prompt = f"""You are a helpful AI assistant with memory capabilities.
        
{memory_context}

When users share important information about themselves, their projects, or preferences, acknowledge it naturally.
Refer back to previous context when relevant to show you remember past interactions."""

        # Prepare messages for API call
        messages = [{"role": "user", "content": user_message}]
        
        # Call Claude API with memory-enriched context
        response = self.client.messages.create(
            model=self.model,
            max_tokens=1024,
            system=system_prompt,
            messages=messages
        )
        
        assistant_reply = response.content[0].text
        
        # Store assistant response in conversation history
        self.memory.add_to_conversation("assistant", assistant_reply)
        
        # Optional: Auto-extract important facts (simple version)
        if store_facts and self._seems_important(user_message):
            self.memory.store_fact(user_message)
        
        return assistant_reply
    
    def _seems_important(self, message: str) -> bool:
        """Simple heuristic to detect if message contains storable information"""
        importance_keywords = ["my name is", "i work on", "i prefer", "remember that", "my project"]
        return any(keyword in message.lower() for keyword in importance_keywords)
    
    def remember_preference(self, key: str, value: str):
        """Explicitly store a user preference"""
        self.memory.update_preference(key, value)
    
    def new_conversation(self):
        """Start fresh conversation while keeping long-term memory"""
        self.memory.clear_conversation()


# Usage example
if __name__ == "__main__":
    agent = MemoryEnabledAgent(agent_id="user_john")
    
    # First interaction - agent learns about user
    response1 = agent.chat("Hi! My name is John and I'm working on a Python web scraper project.")
    print(f"Agent: {response1}\n")
    
    # Second interaction - agent remembers context
    response2 = agent.chat("What was I working on again?")
    print(f"Agent: {response2}\n")
    
    # Explicitly store preference
    agent.remember_preference("coding_style", "PEP 8 strict")
    
    # New session (simulated) - long-term memory persists
    agent.new_conversation()
    response3 = agent.chat("What do you know about my coding preferences?")
    print(f"Agent: {response3}")