Persistent Memory Architectures

Duration: 15 min

Persistent Memory Architectures

Duration: 15 min

Overview

Production agents need memory that survives crashes, restarts, and scaling events. This module covers the architecture patterns for building persistent agent memory: choosing storage backends, designing schemas, handling concurrency, and implementing memory as a service that multiple agent instances can share.

Key Concepts & Foundations

  • What: Persistent memory architecture ensures agent state survives beyond a single process lifetime
  • Why: Production agents restart, scale horizontally, and serve multiple users - memory must be durable and shared
  • How: Use database-backed storage with proper schemas, caching layers, and access patterns

Architecture Patterns

| Pattern | Description | Use Case | |---------|-------------|----------| | Embedded | Memory in agent process (SQLite, in-memory) | Single-user, prototyping | | Client-Server | Agent connects to external DB | Multi-instance, production | | Memory Service | Dedicated microservice for memory | Multi-agent systems | | Event-Sourced | Store all events, derive state | Audit trails, replay |

Storage Backend Selection

  • Redis - Fast, good for session/working memory, TTL support
  • PostgreSQL + pgvector - Relational + vector search, good all-rounder
  • MongoDB - Flexible schemas, good for episodic memory documents
  • DynamoDB - Serverless, scales automatically, good for AWS-native agents
  • SQLite - Zero-config, embedded, perfect for single-user desktop agents

Schema Design Principles

1. User isolation - Each user's memories are completely separate 2. Temporal ordering - Every memory has a timestamp for recency queries 3. Type tagging - Label memories by type (fact, preference, episode, etc.) 4. Version tracking - Track when memories are updated or corrected

Hands-On Implementation

import json
import sqlite3
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path

@dataclass
class StoredMemory:
    id: int
    user_id: str
    content: str
    memory_type: str
    importance: float
    created_at: str
    updated_at: str
    metadata: dict

class PersistentMemoryStore:
    """SQLite-backed persistent memory for agents."""
    def __init__(self, db_path: str = ":memory:"):
        self.db_path = db_path
        self.conn = sqlite3.connect(db_path)
        self.conn.row_factory = sqlite3.Row
        self._create_tables()
    def _create_tables(self):
        self.conn.executescript("""
            CREATE TABLE IF NOT EXISTS memories (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id TEXT NOT NULL,
                content TEXT NOT NULL,
                memory_type TEXT DEFAULT 'general',
                importance REAL DEFAULT 0.5,
                created_at TEXT DEFAULT (datetime('now')),
                updated_at TEXT DEFAULT (datetime('now')),
                metadata TEXT DEFAULT '{}'
            );
            CREATE INDEX IF NOT EXISTS idx_user ON memories(user_id);
            CREATE INDEX IF NOT EXISTS idx_type ON memories(user_id, memory_type);
            CREATE INDEX IF NOT EXISTS idx_importance ON memories(user_id, importance DESC);
        """)
        self.conn.commit()
    def store(self, user_id: str, content: str, memory_type: str = "general",
              importance: float = 0.5, metadata: dict = None) -> int:
        cursor = self.conn.execute(
            """INSERT INTO memories (user_id, content, memory_type, importance, metadata)
               VALUES (?, ?, ?, ?, ?)""",
            (user_id, content, memory_type, importance, json.dumps(metadata or {}))
        )
        self.conn.commit()
        return cursor.lastrowid
    def query(self, user_id: str, memory_type: str = None,
              min_importance: float = 0.0, limit: int = 10) -> list[StoredMemory]:
        sql = "SELECT * FROM memories WHERE user_id = ? AND importance >= ?"
        params = [user_id, min_importance]
        if memory_type:
            sql += " AND memory_type = ?"
            params.append(memory_type)
        sql += " ORDER BY importance DESC, created_at DESC LIMIT ?"
        params.append(limit)
        rows = self.conn.execute(sql, params).fetchall()
        return [StoredMemory(
            id=r["id"], user_id=r["user_id"], content=r["content"],
            memory_type=r["memory_type"], importance=r["importance"],
            created_at=r["created_at"], updated_at=r["updated_at"],
            metadata=json.loads(r["metadata"])
        ) for r in rows]
    def update(self, memory_id: int, content: str = None, importance: float = None):
        updates = ["updated_at = datetime('now')"]
        params = []
        if content is not None:
            updates.append("content = ?")
            params.append(content)
        if importance is not None:
            updates.append("importance = ?")
            params.append(importance)
        params.append(memory_id)
        self.conn.execute(
            f"UPDATE memories SET {', '.join(updates)} WHERE id = ?", params
        )
        self.conn.commit()
    def delete_user_memories(self, user_id: str):
        self.conn.execute("DELETE FROM memories WHERE user_id = ?", (user_id,))
        self.conn.commit()
    def get_stats(self, user_id: str) -> dict:
        row = self.conn.execute(
            """SELECT COUNT(*) as total,
                      COUNT(DISTINCT memory_type) as types,
                      AVG(importance) as avg_importance
               FROM memories WHERE user_id = ?""", (user_id,)
        ).fetchone()
        return {"total": row["total"], "types": row["types"],
                "avg_importance": round(row["avg_importance"] or 0, 2)}


Demo: Persistent memory across agent restarts

store = PersistentMemoryStore()  # Using :memory: for demo
Store memories for a user

store.store("user-123", "Name is Alex", "identity", importance=0.95)
store.store("user-123", "Prefers Python", "preference", importance=0.8)
store.store("user-123", "Works at startup", "fact", importance=0.6)
store.store("user-123", "Building a RAG pipeline", "project", importance=0.85)
store.store("user-123", "Had trouble with Docker", "struggle", importance=0.4)
Query memories

print("High-importance memories:")
for m in store.query("user-123", min_importance=0.7):
    print(f"  [{m.memory_type}] {m.content} (importance={m.importance})")
print("\nPreferences only:")
for m in store.query("user-123", memory_type="preference"):
    print(f"  {m.content}")
print("\nStats:", store.get_stats("user-123"))

Advanced Techniques

1. Write-Ahead Logging - Buffer writes and flush periodically for better performance 2. Memory Sharding - Partition by user ID for horizontal scaling 3. Cache Layer - Keep hot memories (high access frequency) in Redis, cold in PostgreSQL 4. Conflict Resolution - When multiple agent instances write simultaneously, use last-write-wins or merge 5. Backup and Migration - Regular snapshots and schema versioning for safe upgrades

Quiz

Q1: Why do production agents need persistent memory instead of in-process storage?

  • A) It is cheaper
  • B) Agents restart, scale horizontally, and must share state across instances ✓
  • C) In-process storage is slower
  • D) Databases have better syntax

Q2: Which storage backend is best for a serverless agent on AWS?

  • A) SQLite (requires persistent filesystem)
  • B) DynamoDB (serverless, auto-scaling, no instance management) ✓
  • C) Redis (requires always-on instance)
  • D) Local file system

Q3: What does user isolation mean in memory architecture?

  • A) Users cannot access the internet
  • B) Each user's memories are completely separate and inaccessible to other users ✓
  • C) Users are isolated from the agent
  • D) Only one user can use the system at a time

← Back ▶ Run in Colab