Implementing Memory in Our Agent › Starting Point: A Complete Memory System

class ConversationManager:
    """
    Manages short-term conversation memory with automatic windowing.
    """
    
    def __init__(self, max_messages: int = 20):
        self.messages = []
        self.max_messages = max_messages
    
    def add_user_message(self, content: str):
        """Add a user message to history."""
        self.messages.append({
            "role": "user",
            "content": content
        })
        self._trim_if_needed()
    
    def add_assistant_message(self, content: str):
        """Add an assistant message to history."""
        self.messages.append({
            "role": "assistant",
            "content": content
        })
        self._trim_if_needed()
    
    def _trim_if_needed(self):
        """Keep only the most recent messages."""
        if len(self.messages) > self.max_messages:
            self.messages = self.messages[-self.max_messages:]
    
    def get_messages(self) -> list:
        """Get all current messages."""
        return self.messages.copy()
    
    def clear(self):
        """Start a fresh conversation."""
        self.messages = []

import json
import os
from datetime import datetime

class KnowledgeStore:
    """
    Stores and retrieves long-term facts about the user.
    """
    
    def __init__(self, storage_file: str = "knowledge.json"):
        self.storage_file = storage_file
        self.facts = self._load_facts()
    
    def _load_facts(self) -> list:
        """Load facts from disk."""
        if os.path.exists(self.storage_file):
            with open(self.storage_file, 'r') as f:
                return json.load(f)
        return []
    
    def _save_facts(self):
        """Save facts to disk."""
        with open(self.storage_file, 'w') as f:
            json.dump(self.facts, f, indent=2)
    
    def add_fact(self, fact: str, category: str = None):
        """Store a new fact."""
        entry = {
            "fact": fact,
            "category": category,
            "timestamp": datetime.now().isoformat()
        }
        self.facts.append(entry)
        self._save_facts()
    
    def search(self, query: str) -> list:
        """
        Search for relevant facts.
        Simple keyword matching for now.
        """
        query_lower = query.lower()
        results = []
        
        for entry in self.facts:
            fact_lower = entry["fact"].lower()
            if query_lower in fact_lower:
                results.append(entry["fact"])
            elif entry["category"] and query_lower in entry["category"].lower():
                results.append(entry["fact"])
        
        return results
    
    def get_all_facts(self) -> list:
        """Get all stored facts."""
        return [entry["fact"] for entry in self.facts]

import os
from anthropic import Anthropic

class PersonalAssistant:
    """
    A personal assistant with both short-term and long-term memory.
    """
    
    def __init__(self, api_key: str):
        # Using Claude Sonnet 4.5 for its excellent agent reasoning capabilities
        self.client = Anthropic(api_key=api_key)
        self.model = "claude-sonnet-4-5"
        
        # Memory systems
        self.conversation = ConversationManager(max_messages=20)
        self.knowledge = KnowledgeStore()
    
    def chat(self, user_message: str) -> str:
        """
        Process a user message and return a response.
        Handles both conversation and knowledge storage.
        """
        # Check if this is a memory command
        if self._is_memory_command(user_message):
            return self._handle_memory_command(user_message)
        
        # Search for relevant facts
        relevant_facts = self.knowledge.search(user_message)
        
        # Build system prompt with context
        system_prompt = self._build_system_prompt(relevant_facts)
        
        # Add user message to conversation
        self.conversation.add_user_message(user_message)
        
        # Get response from Claude
        response = self.client.messages.create(
            model=self.model,
            max_tokens=1024,
            system=system_prompt,
            messages=self.conversation.get_messages()
        )
        
        # Extract and store response
        assistant_message = response.content[0].text
        self.conversation.add_assistant_message(assistant_message)
        
        return assistant_message
    
    def _is_memory_command(self, message: str) -> bool:
        """Detect if user wants to store something."""
        keywords = ["remember", "store", "save", "keep in mind", "note that"]
        message_lower = message.lower()
        return any(keyword in message_lower for keyword in keywords)
    
    def _handle_memory_command(self, message: str) -> str:
        """Extract and store information."""
        # Use Claude to extract the fact
        response = self.client.messages.create(
            model=self.model,
            max_tokens=256,
            system="Extract the key fact the user wants you to remember. Return only the fact as a clear, concise statement.",
            messages=[{"role": "user", "content": message}]
        )
        
        fact = response.content[0].text
        self.knowledge.add_fact(fact)
        
        return f"Got it! I'll remember that {fact.lower()}"
    
    def _build_system_prompt(self, relevant_facts: list) -> str:
        """Build system prompt with retrieved knowledge."""
        base_prompt = "You are a helpful personal assistant."
        
        if not relevant_facts:
            return base_prompt
        
        facts_text = "\n".join(f"- {fact}" for fact in relevant_facts)
        return f"""{base_prompt}

You have access to the following information about the user:

{facts_text}

Use this information when relevant to provide personalized responses."""
    
    def start_new_conversation(self):
        """Clear conversation history but keep long-term knowledge."""
        self.conversation.clear()

## Create the assistant
assistant = PersonalAssistant(api_key=os.getenv("ANTHROPIC_API_KEY"))

## Store some facts
print(assistant.chat("Remember that my birthday is July 20"))
## Output: Got it! I'll remember that your birthday is july 20

print(assistant.chat("Remember that I'm allergic to peanuts"))
## Output: Got it! I'll remember that you're allergic to peanuts

print(assistant.chat("Remember that I prefer Italian food"))
## Output: Got it! I'll remember that you prefer italian food

## Now ask questions that use this knowledge
print(assistant.chat("What should I be careful about when eating out?"))
## Output: Based on what I know, you should be careful about peanuts since 
## you're allergic to them. Always inform restaurant staff about your allergy 
## and ask about ingredients, especially in sauces and desserts where peanuts 
## might be hidden...

print(assistant.chat("Suggest a restaurant for my birthday dinner"))
## Output: For your birthday on July 20, I'd suggest an Italian restaurant 
## since you prefer Italian food. Make sure to mention your peanut allergy 
## when making the reservation...