Long-Term Memory: Persistent Storage Systems¶

Overview¶

Long-term memory is persistent storage that survives across conversations and sessions. It enables agents to learn over time and build up knowledge.

Unlike short-term memory (measured in tokens, cleared each turn), long-term memory is measured in GB/TB and keeps growing.

Types of Long-Term Memory¶

1. Episodic Memory (What Happened)¶

Event: "User asked about quantum computing on July 15"
Storage: Vector DB indexed by time + semantics
Purpose: Learn from past interactions
Access: "Retrieve similar past conversations"

Structure:

class EpisodEvent:
    timestamp: datetime
    user_input: str
    agent_response: str
    outcome: str  # Success/failure/quality
    embedding: np.array  # For semantic search
    tags: List[str]  # For filtering

2. Semantic Memory (Knowledge Base)¶

Fact: "Claude is an AI assistant made by Anthropic"
Storage: Knowledge graph or structured DB
Purpose: General knowledge
Access: "What is Claude?"

Structure:

class SemanticFact:
    subject: str  # "Claude"
    predicate: str  # "is_made_by"
    object: str  # "Anthropic"
    confidence: float  # 0.0-1.0
    source: str  # "Training data" / "User told me"

3. Procedural Memory (How to Do Things)¶

Procedure: "To analyze a dataset: 1) Load, 2) Explore, 3) Clean, 4) Analyze"
Storage: Procedure library with metrics
Purpose: Skill/strategy storage
Access: "How do I analyze data?"

Structure:

class Procedure:
    name: str  # "analyze_dataset"
    steps: List[str]
    success_rate: float  # Historical success
    prerequisites: List[str]
    estimated_tokens: int
    tags: List[str]

Storage Technologies¶

Option 1: Vector Database¶

Best for: Episodic memory (past events)

Characteristics: - Semantic search capability - Fast retrieval (using indices) - Good for similarity matching - Scales to millions of items

Examples: Chroma, Weaviate, Milvus, LanceDB, Pinecone

Implementation:

from chromadb import Client

vector_db = Client()
collection = vector_db.get_or_create_collection("episodic_memory")

# Store an episode
collection.add(
    documents=[user_input],
    embeddings=[embedding],
    metadatas=[{
        "timestamp": timestamp,
        "agent_response": response,
        "quality": 0.85
    }],
    ids=[episode_id]
)

# Retrieve similar episodes
results = collection.query(
    query_embeddings=[query_embedding],
    n_results=5
)

Option 2: Knowledge Graph¶

Best for: Semantic memory (facts and relationships)

Characteristics: - Structured relationships - Reasoning over relationships - Good for complex queries - Moderate scalability

Examples: Neo4j, ArangoDB, MemGraph

Implementation:

from neo4j import GraphDatabase

driver = GraphDatabase.driver("bolt://localhost:7687")

# Store a fact
with driver.session() as session:
    session.run(
        """
        CREATE (a:Person {name: 'Alice'})
        CREATE (b:Company {name: 'Acme'})
        CREATE (a)-[:WORKS_AT]->(b)
        """
    )

# Query relationships
with driver.session() as session:
    result = session.run(
        "MATCH (p:Person)-[:WORKS_AT]->(c:Company) RETURN p, c"
    )

Option 3: Traditional Database¶

Best for: Procedural memory (procedures) and structured data

Characteristics: - Reliable and proven - Good for structured data - ACID transactions - Complex queries

Examples: PostgreSQL, MongoDB, MySQL

Implementation:

import sqlite3

conn = sqlite3.connect('memory.db')
cursor = conn.cursor()

# Store a procedure
cursor.execute('''
    INSERT INTO procedures (name, steps, success_rate)
    VALUES (?, ?, ?)
''', ('analyze_data', json.dumps(steps), 0.85))

# Query procedures
cursor.execute('SELECT * FROM procedures WHERE success_rate > 0.8')
results = cursor.fetchall()

conn.commit()

Hybrid Architecture¶

Most production systems use a hybrid approach:

- ┌─────────────────────────────────────────┐
    - Long-Term Memory System          │
  - ┤
│                                         │
    - Vector DB                              │
      - Episodic memories (events)         │
      - Semantic similarity search         │
      - ~1-10M items                       │
│                                         │
    - Knowledge Graph                        │
      - Semantic facts (relationships)     │
      - Relationship reasoning             │
      - ~100K-1M triples                   │
│                                         │
    - Relational DB                          │
      - Structured data                    │
      - Procedures & policies              │
      - User profiles                      │
│                                         │
    - File Storage                           │
      - Large documents                    │
      - Full conversations                 │
      - Backups                            │
│                                         │
  - ┘

Memory Consolidation¶

Process of organizing and compressing memories over time:

Stage 1: Encoding (Recording)¶

New experience → Encode into vectors → Store in vector DB

Stage 2: Consolidation (Organizing)¶

Similar memories → Group together → Extract patterns

Stage 3: Compression (Compacting)¶

1000 similar memories → Summarize into 10 concepts

Implementation:

class MemoryConsolidator:
    def consolidate(self, memory_db):
        """Run consolidation process"""

        # Stage 1: Clustering
        # Group similar memories
        clusters = self.cluster_similar_memories(memory_db)

        # Stage 2: Extracting Patterns
        # What patterns emerge?
        patterns = self.extract_patterns(clusters)

        # Stage 3: Summarization
        # Convert to semantic facts
        for pattern in patterns:
            semantic_fact = self.pattern_to_fact(pattern)
            self.semantic_db.add(semantic_fact)

        # Stage 4: Compression
        # Old specific memories can be deleted
        for cluster in clusters:
            self.mark_for_deletion(cluster)

Retrieval Strategies¶

Strategy 1: Similarity Search¶

# Find similar past episodes
query = "User asked about budget planning"
query_embedding = embed(query)

similar_episodes = vector_db.search(
    query_embedding,
    top_k=5
)

# Result: Past times we discussed budgets

# Follow relationships in knowledge graph
query = """
  Find all companies that Alice knows about
  through her colleagues
"""

results = knowledge_graph.query(query)

Strategy 3: Filtering¶

# Filter by metadata
recent_high_quality = memory_db.filter(
    timestamp__after=datetime(2025, 1, 1),
    quality__gte=0.8,
    tags__in=["research", "important"]
)

Practical Implementation¶

Complete Long-Term Memory System¶

class LongTermMemorySystem:
    def __init__(self):
        # Vector DB for episodic
        self.episodic = Chroma(collection_name="episodes")

        # Knowledge graph for semantic
        self.semantic = Neo4jDB()

        # SQL DB for procedures
        self.procedural = SQLiteDB("procedures.db")

    def remember_event(self, event: dict):
        """Store an episodic memory"""
        # Embed and store
        embedding = embed_text(event['description'])

        self.episodic.add(
            documents=[event['description']],
            embeddings=[embedding],
            metadatas=[{
                'timestamp': event['timestamp'],
                'outcome': event['outcome'],
                'tags': event['tags']
            }]
        )

    def remember_fact(self, subject, predicate, obj, confidence=1.0):
        """Store a semantic memory"""
        self.semantic.add_triple(
            subject=subject,
            predicate=predicate,
            object=obj,
            confidence=confidence
        )

    def remember_procedure(self, procedure: dict):
        """Store a procedural memory"""
        self.procedural.insert(
            name=procedure['name'],
            steps=json.dumps(procedure['steps']),
            success_rate=procedure['success_rate'],
            tags=','.join(procedure['tags'])
        )

    def recall_similar(self, query: str, k=5):
        """Recall similar past events"""
        query_embedding = embed_text(query)
        return self.episodic.search(query_embedding, top_k=k)

    def recall_fact(self, subject, predicate):
        """Recall a fact"""
        return self.semantic.query(subject, predicate)

    def recall_procedure(self, task_name):
        """Recall how to do something"""
        return self.procedural.query(
            "SELECT * FROM procedures WHERE name = ?",
            (task_name,)
        )

    def consolidate(self):
        """Periodically consolidate memories"""
        # Cluster similar episodes
        clusters = self.episodic.cluster_similar()

        # Extract patterns into semantic facts
        for cluster in clusters:
            pattern = self.extract_pattern(cluster)
            self.remember_fact(
                subject=pattern['subject'],
                predicate=pattern['predicate'],
                obj=pattern['object'],
                confidence=0.9
            )

Scaling Long-Term Memory¶

Challenge: Storage Growth¶

Day 1:   10 memories    (0.1 MB)
Day 30:  300 memories   (3 MB)
Month 6: 5,000 memories (50 MB)
Year 1:  15,000 memories (150 MB)
Year 5:  75,000 memories (750 MB)

Solutions¶

1. Partitioning by Time¶

vector_db_2025_q1 = Chroma("episodes_2025_q1")
vector_db_2025_q2 = Chroma("episodes_2025_q2")

# Query recent
recent = vector_db_2025_q2.search(query)

# Query historical (if needed)
historical = vector_db_2025_q1.search(query)

2. Partitioning by User¶

user_memories = {
    "alice": Chroma("alice_episodes"),
    "bob": Chroma("bob_episodes")
}

3. Pruning Old Memories¶

def prune_old_memories(memory_db, days=90):
    """Delete old, low-quality memories"""

    cutoff = now() - timedelta(days=days)

    old_memories = memory_db.filter(
        timestamp__before=cutoff
    )

    for memory in old_memories:
        if memory['quality'] < 0.5:
            memory_db.delete(memory['id'])

Privacy & Security¶

Challenge: PII in Memories¶

Dangerous to store:
  ✗ Full credit card numbers
  ✗ Social security numbers
  ✗ Passwords
  ✗ Medical information

Solution: Anonymization¶

def anonymize_event(event):
    """Remove PII before storing"""
    # Replace credit cards with "CARD_XXXX"
    event['text'] = re.sub(
        r'\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}',
        'CARD_XXXX',
        event['text']
    )

    # Replace emails with "USER_EMAIL"
    event['text'] = re.sub(
        r'\S+@\S+',
        'USER_EMAIL',
        event['text']
    )

    return event

Monitoring Long-Term Memory¶

def monitor_memory_health(memory_system):
    """Check memory system health"""

    print(f"Vector DB size: {memory_system.episodic.size()}")
    print(f"Knowledge graph nodes: {memory_system.semantic.count_nodes()}")
    print(f"Procedures stored: {memory_system.procedural.count()}")

    # Check retrieval quality
    test_queries = [
        "Tell me about past research discussions",
        "What facts do we know about AI?",
        "How do we analyze data?"
    ]

    for query in test_queries:
        results = memory_system.recall_similar(query)
        if not results:
            print(f"WARNING: No results for '{query}'")

    # Check memory freshness
    avg_age = memory_system.episodic.average_age_days()
    print(f"Average memory age: {avg_age} days")

Best Practices¶

1. Structure Matters¶

# ✅ Good: Clean structure
memory = {
    "timestamp": datetime.now(),
    "event_type": "research_completed",
    "subject": "quantum_computing",
    "quality": 0.9
}

# ❌ Bad: Messy
memory = {"data": "stuff happened"}

2. Version Control Memories¶

# ✅ Good: Track updates
memory.update(
    fact="New discovery",
    version=2,  # Updated
    previous_version=1
)

# ❌ Bad: Overwrite without history
memory["fact"] = "New discovery"  # Old value lost

3. Index Frequently Accessed¶

# ✅ Good: Index for speed
memory_db.create_index("timestamp")
memory_db.create_index("quality")
memory_db.create_index("tags")

# ❌ Bad: No indexing
# Each query scans everything

4. Regular Maintenance¶

# ✅ Good: Scheduled consolidation
schedule.every().day.at("02:00").do(
    memory_system.consolidate
)
schedule.every().week.at("03:00").do(
    memory_system.prune_old_memories
)

# ❌ Bad: Never clean up
# Memory bloats, gets slow

Key Takeaways¶

Multiple storage types needed - Vector DB, Knowledge Graph, SQL DB
Consolidation is critical - Compress memories over time
Hybrid architecture - Use right tool for each memory type
Scalability challenges exist - Plan for growth
Privacy matters - Anonymize before storing
Monitor health - Check retrieval quality regularly

Next Steps¶

Read Vector Stores & Retrieval - Semantic search
Read Memory Compression - Making memories efficient

Last Updated: August 9, 2026

Long-Term Memory: Persistent Storage Systems¶

Overview¶

Types of Long-Term Memory¶

1. Episodic Memory (What Happened)¶

2. Semantic Memory (Knowledge Base)¶

3. Procedural Memory (How to Do Things)¶

Storage Technologies¶

Option 1: Vector Database¶

Option 2: Knowledge Graph¶

Option 3: Traditional Database¶

Hybrid Architecture¶

Memory Consolidation¶

Stage 1: Encoding (Recording)¶

Stage 2: Consolidation (Organizing)¶

Stage 3: Compression (Compacting)¶

Retrieval Strategies¶

Strategy 1: Similarity Search¶

Strategy 2: Relationship Navigation¶

Strategy 3: Filtering¶

Practical Implementation¶

Complete Long-Term Memory System¶

Scaling Long-Term Memory¶

Challenge: Storage Growth¶

Solutions¶

1. Partitioning by Time¶

2. Partitioning by User¶

3. Pruning Old Memories¶

Privacy & Security¶

Challenge: PII in Memories¶

Solution: Anonymization¶

Monitoring Long-Term Memory¶

Best Practices¶

1. Structure Matters¶

2. Version Control Memories¶

3. Index Frequently Accessed¶

4. Regular Maintenance¶

Key Takeaways¶

Next Steps¶