Long-Term Knowledge Storage and Retrieval: Building Persistent Memory for AI Agents

import json
import os

class SimpleMemory:
    def __init__(self, memory_file="assistant_memory.json"):
        self.memory_file = memory_file
        self.memory = self._load_memory()
    
    def _load_memory(self):
        """Load memory from file if it exists"""
        if os.path.exists(self.memory_file):
            with open(self.memory_file, 'r') as f:
                return json.load(f)
        return {}
    
    def _save_memory(self):
        """Save memory to file"""
        with open(self.memory_file, 'w') as f:
            json.dump(self.memory, f, indent=2)
    
    def store(self, key, value):
        """Store a fact in long-term memory"""
        self.memory[key] = value
        self._save_memory()
        return f"Remembered: {key}"
    
    def retrieve(self, key):
        """Retrieve a fact from long-term memory"""
        return self.memory.get(key, "I don't have that information stored.")

## Example usage
memory = SimpleMemory()
memory.store("user_birthday", "July 20")
memory.store("favorite_restaurant", "Luigi's Italian Kitchen")

## Later, even after restarting the program
print(memory.retrieve("user_birthday"))  # Output: July 20

class SearchableMemory:
    def __init__(self, memory_file="assistant_knowledge.json"):
        self.memory_file = memory_file
        self.facts = self._load_facts()
    
    def _load_facts(self):
        """Load facts from file"""
        if os.path.exists(self.memory_file):
            with open(self.memory_file, 'r') as f:
                return json.load(f)
        return []
    
    def _save_facts(self):
        """Save facts to file"""
        with open(self.memory_file, 'w') as f:
            json.dump(self.facts, f, indent=2)
    
    def add_fact(self, fact, category=None):
        """Add a fact to the knowledge base"""
        entry = {
            "fact": fact,
            "category": category,
            "timestamp": str(datetime.now())
        }
        self.facts.append(entry)
        self._save_facts()
        return f"Stored: {fact}"
    
    def search(self, query):
        """Search for facts containing the query text"""
        results = []
        query_lower = query.lower()
        
        for entry in self.facts:
            # Simple keyword search
            if query_lower in entry["fact"].lower():
                results.append(entry["fact"])
            elif entry["category"] and query_lower in entry["category"].lower():
                results.append(entry["fact"])
        
        return results if results else ["No matching information found."]

## Example usage
from datetime import datetime

kb = SearchableMemory()
kb.add_fact("User's birthday is July 20", category="personal")
kb.add_fact("User prefers Italian food", category="preferences")
kb.add_fact("User is allergic to peanuts", category="preferences")

## Search by keyword
print(kb.search("birthday"))
## Output: ["User's birthday is July 20"]

print(kb.search("food"))
## Output: ["User prefers Italian food"]

print(kb.search("preferences"))
## Output: ["User prefers Italian food", "User is allergic to peanuts"]

from sentence_transformers import SentenceTransformer
import numpy as np

class VectorMemory:
    def __init__(self):
        # Using Claude Sonnet 4.5 for the agent, but a local model for embeddings
        # to keep costs down for frequent similarity searches
        self.encoder = SentenceTransformer('all-MiniLM-L6-v2')
        self.facts = []
        self.vectors = []
    
    def add_fact(self, fact):
        """Add a fact and its vector representation"""
        vector = self.encoder.encode(fact)
        self.facts.append(fact)
        self.vectors.append(vector)
        return f"Stored: {fact}"
    
    def search(self, query, top_k=3):
        """Find the most relevant facts for a query"""
        if not self.facts:
            return ["No information stored yet."]
        
        # Convert query to vector
        query_vector = self.encoder.encode(query)
        
        # Calculate similarity with all stored facts
        similarities = []
        for i, fact_vector in enumerate(self.vectors):
            # Cosine similarity
            similarity = np.dot(query_vector, fact_vector) / (
                np.linalg.norm(query_vector) * np.linalg.norm(fact_vector)
            )
            similarities.append((similarity, self.facts[i]))
        
        # Sort by similarity and return top results
        similarities.sort(reverse=True, key=lambda x: x[0])
        return [fact for score, fact in similarities[:top_k] if score > 0.3]

## Example usage
memory = VectorMemory()
memory.add_fact("User's birthday is July 20")
memory.add_fact("User prefers Italian food")
memory.add_fact("User is allergic to peanuts")
memory.add_fact("User lives in San Francisco")
memory.add_fact("User works as a software engineer")

## Semantic search
print(memory.search("What should I avoid eating?"))
## Output: ["User is allergic to peanuts", "User prefers Italian food"]

print(memory.search("Where does the user live?"))
## Output: ["User lives in San Francisco"]

print(memory.search("What is the user's profession?"))
## Output: ["User works as a software engineer"]

import os
from anthropic import Anthropic
import os

class AssistantWithMemory:
    def __init__(self):
        self.client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
        # Using Claude Sonnet 4.5 for its superior agent reasoning capabilities
        self.model = "claude-sonnet-4-5"
        self.memory = VectorMemory()
        self.conversation_history = []
    
    def process_message(self, user_message):
        """Process a user message, using memory when relevant"""
        # First, check if this is a request to remember something
        if self._is_memory_request(user_message):
            return self._store_memory(user_message)
        
        # Search for relevant facts
        relevant_facts = self.memory.search(user_message, top_k=3)
        
        # Build context with conversation history and retrieved facts
        context = self._build_context(relevant_facts)
        
        # Get response from Claude
        response = self.client.messages.create(
            model=self.model,
            max_tokens=1024,
            system=context,
            messages=self.conversation_history + [
                {"role": "user", "content": user_message}
            ]
        )
        
        assistant_message = response.content[0].text
        
        # Update conversation history
        self.conversation_history.append({"role": "user", "content": user_message})
        self.conversation_history.append({"role": "assistant", "content": assistant_message})
        
        return assistant_message
    
    def _is_memory_request(self, message):
        """Detect if user wants to store information"""
        memory_keywords = ["remember", "store", "save", "keep in mind", "note that"]
        return any(keyword in message.lower() for keyword in memory_keywords)
    
    def _store_memory(self, message):
        """Extract and store information from user message"""
        # Use Claude to extract the fact to remember
        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 statement.",
            messages=[{"role": "user", "content": message}]
        )
        
        fact = response.content[0].text
        self.memory.add_fact(fact)
        return f"Got it! I'll remember that {fact.lower()}"
    
    def _build_context(self, relevant_facts):
        """Build system context with retrieved facts"""
        if not relevant_facts or relevant_facts == ["No information stored yet."]:
            return "You are a helpful personal assistant."
        
        facts_text = "\n".join(f"- {fact}" for fact in relevant_facts)
        return f"""You are a helpful personal assistant. You have access to the following information about the user:

{facts_text}

Use this information when relevant to provide personalized responses."""

## Example conversation
assistant = AssistantWithMemory()

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

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

print(assistant.process_message("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. When eating out, make sure to inform the restaurant staff about 
## your peanut allergy and ask about ingredients in dishes...

def should_remember(self, information):
    """Use Claude to decide if information is worth storing long-term"""
    response = self.client.messages.create(
        model=self.model,
        max_tokens=128,
        system="Decide if this information should be stored in long-term memory. Answer only 'yes' or 'no' with a brief reason.",
        messages=[{"role": "user", "content": f"Should I remember this: {information}"}]
    )
    
    decision = response.content[0].text.lower()
    return "yes" in decision

def update_fact(self, old_fact, new_fact):
    """Replace an old fact with updated information"""
    # Remove old fact
    if old_fact in self.memory.facts:
        idx = self.memory.facts.index(old_fact)
        self.memory.facts.pop(idx)
        self.memory.vectors.pop(idx)
    
    # Add new fact
    self.memory.add_fact(new_fact)
    return f"Updated: {new_fact}"