AI Agent Memory Systems

Real-World Scenario

A personal assistant agent that schedules meetings, answers questions about your calendar, and learns your preferences over time. On day 1, you tell it you prefer morning meetings and dislike Friday afternoons. On day 15, when scheduling a new meeting, it applies these preferences automatically — because they’re stored in long-term memory, not the conversation history.

Short-Term Memory: Conversation History

import anthropic
from typing import Optional
from dataclasses import dataclass, field

client = anthropic.Anthropic()

@dataclass
class ConversationMemory:
    """Manages conversation history with optional summarization."""
    messages: list[dict] = field(default_factory=list)
    max_messages: int = 20   # keep last N messages

    def add(self, role: str, content: str) -> None:
        self.messages.append({"role": role, "content": content})
        # Trim old messages to stay within context budget
        if len(self.messages) > self.max_messages:
            # Keep system context + recent messages
            self.messages = self.messages[-self.max_messages:]

    def get_context(self) -> list[dict]:
        return self.messages.copy()

    def summarize_if_needed(self, token_threshold: int = 4000) -> None:
        """Compress old messages into a summary when context grows large."""
        if len(self.messages) < 10:
            return

        # Summarize all but the last 4 messages
        to_summarize = self.messages[:-4]
        history_text = "\n".join(
            f"{m['role'].title()}: {m['content']}"
            for m in to_summarize
        )

        response = client.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=512,
            messages=[{
                "role": "user",
                "content": f"Summarize this conversation concisely, preserving key facts:\n\n{history_text}"
            }]
        )
        summary = response.content[0].text

        # Replace old messages with summary
        self.messages = [
            {"role": "user",      "content": f"[Previous conversation summary: {summary}]"},
            {"role": "assistant", "content": "Understood, I have the context from our previous discussion."},
        ] + self.messages[-4:]

Long-Term Memory: Key-Value Store

import json
import sqlite3
from datetime import datetime
from pathlib import Path

class AgentMemoryStore:
    """Persistent key-value memory backed by SQLite."""

    def __init__(self, db_path: str = "./agent_memory.db"):
        self.conn = sqlite3.connect(db_path)
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS memories (
                key       TEXT PRIMARY KEY,
                value     TEXT NOT NULL,
                category  TEXT DEFAULT 'general',
                created   TEXT NOT NULL,
                updated   TEXT NOT NULL
            )
        """)
        self.conn.commit()

    def remember(self, key: str, value: str, category: str = "general") -> None:
        """Store or update a memory."""
        now = datetime.utcnow().isoformat()
        self.conn.execute("""
            INSERT INTO memories (key, value, category, created, updated)
            VALUES (?, ?, ?, ?, ?)
            ON CONFLICT(key) DO UPDATE SET
                value   = excluded.value,
                updated = excluded.updated
        """, (key, value, category, now, now))
        self.conn.commit()

    def recall(self, key: str) -> str | None:
        """Retrieve a specific memory by key."""
        row = self.conn.execute(
            "SELECT value FROM memories WHERE key = ?", (key,)
        ).fetchone()
        return row[0] if row else None

    def recall_category(self, category: str) -> dict[str, str]:
        """Retrieve all memories in a category."""
        rows = self.conn.execute(
            "SELECT key, value FROM memories WHERE category = ?", (category,)
        ).fetchall()
        return {row[0]: row[1] for row in rows}

    def forget(self, key: str) -> None:
        self.conn.execute("DELETE FROM memories WHERE key = ?", (key,))
        self.conn.commit()

    def all_memories(self) -> list[dict]:
        rows = self.conn.execute(
            "SELECT key, value, category, updated FROM memories ORDER BY updated DESC"
        ).fetchall()
        return [{"key": r[0], "value": r[1], "category": r[2], "updated": r[3]} for r in rows]

    def to_context_string(self) -> str:
        """Format all memories for injection into system prompt."""
        memories = self.all_memories()
        if not memories:
            return ""
        lines = ["## What I remember about you:\n"]
        for m in memories:
            lines.append(f"- {m['key']}: {m['value']}")
        return "\n".join(lines)

Agent with Full Memory Integration

import anthropic
import json

client = anthropic.Anthropic()

# Memory tools the agent can use to read/write its own memory
MEMORY_TOOLS = [
    {
        "name": "remember",
        "description": "Save a fact or preference to long-term memory. Use when the user tells you something important about themselves or their preferences.",
        "input_schema": {
            "type": "object",
            "properties": {
                "key":      {"type": "string", "description": "A descriptive key, e.g. 'preferred_language', 'work_schedule'"},
                "value":    {"type": "string", "description": "The value to remember"},
                "category": {"type": "string", "enum": ["preferences", "facts", "tasks", "general"]},
            },
            "required": ["key", "value"]
        }
    },
    {
        "name": "recall",
        "description": "Retrieve a specific memory by key.",
        "input_schema": {
            "type": "object",
            "properties": {
                "key": {"type": "string"}
            },
            "required": ["key"]
        }
    },
    {
        "name": "list_memories",
        "description": "List all stored memories. Use to review what you know about the user.",
        "input_schema": {"type": "object", "properties": {}}
    }
]

class MemoryAgent:
    def __init__(self):
        self.memory = AgentMemoryStore()
        self.history = ConversationMemory()

    def _build_system(self) -> str:
        memory_ctx = self.memory.to_context_string()
        return f"""You are a helpful personal assistant with long-term memory.

{memory_ctx}

When users share preferences, facts about themselves, or important information,
use the 'remember' tool to store it for future conversations.
Proactively apply stored preferences without being asked."""

    def chat(self, user_message: str) -> str:
        self.history.add("user", user_message)
        messages = self.history.get_context()

        while True:
            response = client.messages.create(
                model="claude-sonnet-4-6",
                max_tokens=1024,
                system=self._build_system(),
                tools=MEMORY_TOOLS,
                messages=messages,
            )

            if response.stop_reason == "end_turn":
                reply = next(b.text for b in response.content if hasattr(b, "text"))
                self.history.add("assistant", reply)
                return reply

            messages.append({"role": "assistant", "content": response.content})
            tool_results = []

            for block in response.content:
                if block.type != "tool_use":
                    continue

                if block.name == "remember":
                    self.memory.remember(
                        block.input["key"],
                        block.input["value"],
                        block.input.get("category", "general")
                    )
                    result = f"Remembered: {block.input['key']} = {block.input['value']}"

                elif block.name == "recall":
                    value = self.memory.recall(block.input["key"])
                    result = value or f"No memory found for key: {block.input['key']}"

                elif block.name == "list_memories":
                    mems = self.memory.all_memories()
                    result = json.dumps(mems, indent=2) if mems else "No memories stored yet."

                else:
                    result = "Unknown tool"

                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": result,
                })

            messages.append({"role": "user", "content": tool_results})


# Example session
agent = MemoryAgent()

turns = [
    "Hi! I'm a Python developer and I prefer short, code-focused explanations.",
    "I work best in the mornings and prefer async Python over threading.",
    "Can you explain the asyncio event loop?",   # should apply stored preferences
]

for turn in turns:
    print(f"\nUser: {turn}")
    reply = agent.chat(turn)
    print(f"Agent: {reply[:300]}...")