Agent Memory Patterns
MemoryArchitectureTutorialBest Practices
Implement persistent memory systems for AI agents with proven architectural patterns.
Agent Memory Patterns
Overview
Agent memory is crucial for building AI agents that remember context, learn from interactions, and maintain continuity across sessions. This tutorial explores proven patterns for implementing persistent memory in AI agents.
Memory Architecture Overview
┌─────────────────────────────────────────────────────────────────┐
│ Memory Layers │
├─────────────────────────────────────────────────────────────────┤
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ Short-Term │───▶│ Working │───▶│ Long-Term │ │
│ │ (Context) │ │ Memory │ │ Memory │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ Session │ │ Task │ │ Vector │ │
│ │ Context │ │ State │ │ Store │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
└─────────────────────────────────────────────────────────────────┘
Pattern 1: Hierarchical Memory
Implementation
# src/memory/hierarchical.py
from dataclasses import dataclass, field
from typing import Any
from datetime import datetime
@dataclass
class MemoryEntry:
content: str
category: str
importance: float # 0-1
timestamp: datetime
metadata: dict = field(default_factory=dict)
embeddings: list[float] = field(default_factory=list)
class HierarchicalMemory:
"""Multi-level memory system."""
def __init__(self):
# Short-term: Last N interactions
self.short_term: list[MemoryEntry] = []
self.short_term_limit = 20
# Working: Current task context
self.working: dict[str, Any] = {}
# Long-term: Persistent storage
self.long_term: list[MemoryEntry] = []
def add(self, content: str, category: str, importance: float = 0.5):
"""Add to appropriate memory level."""
entry = MemoryEntry(
content=content,
category=category,
importance=importance,
timestamp=datetime.now()
)
# Add to short-term
self.short_term.append(entry)
if len(self.short_term) > self.short_term_limit:
self.short_term.pop(0)
# If important, promote to long-term
if importance > 0.7:
self.long_term.append(entry)
self._consolidate_long_term()
return entry
def retrieve(self, query: str, level: str = "all") -> list[MemoryEntry]:
"""Retrieve relevant memories."""
results = []
if level in ["short_term", "all"]:
results.extend(self._search_short_term(query))
if level in ["long_term", "all"]:
results.extend(self._search_long_term(query))
return sorted(results, key=lambda e: e.importance, reverse=True)
def _search_short_term(self, query: str) -> list[MemoryEntry]:
"""Simple keyword search in short-term."""
query_lower = query.lower()
return [
e for e in self.short_term
if query_lower in e.content.lower()
]
def _search_long_term(self, query: str) -> list[MemoryEntry]:
"""Semantic search in long-term (requires embeddings)."""
# Use vector similarity
query_embedding = generate_embedding(query)
scores = cosine_similarity(query_embedding, [e.embeddings for e in self.long_term])
return [
entry for entry, score in zip(self.long_term, scores)
if score > 0.7
]
def _consolidate_long_term(self):
"""Merge similar memories to prevent bloat."""
# Group by category and merge similar entries
...
Pattern 2: Entity-Based Memory
Implementation
# src/memory/entity.py
from collections import defaultdict
from dataclasses import dataclass, field
@dataclass
class Entity:
name: str
entity_type: str # person, project, concept, etc.
attributes: dict[str, Any]
relationships: list[str] = field(default_factory=list)
last_seen: datetime = field(default_factory=datetime.now)
mention_count: int = 0
class EntityMemory:
"""Memory organized around entities."""
def __init__(self):
self.entities: dict[str, Entity] = {}
self.relationships: dict[tuple[str, str], list[str]] = defaultdict(list)
def extract_entities(self, text: str) -> list[Entity]:
"""Extract entities from text using LLM."""
prompt = f"""Extract entities from this text.
Text: {text}
Return JSON with entities and their types:
- person: People mentioned
- project: Projects or tasks
- concept: Technical concepts
- organization: Companies or groups
"""
# Call LLM to extract entities
...
def update_entity(self, name: str, updates: dict):
"""Update or create entity."""
if name not in self.entities:
self.entities[name] = Entity(
name=name,
entity_type=updates.get('type', 'unknown'),
attributes={}
)
entity = self.entities[name]
entity.attributes.update(updates.get('attributes', {}))
entity.last_seen = datetime.now()
entity.mention_count += 1
def add_relationship(self, entity1: str, entity2: str, relation: str):
"""Add relationship between entities."""
self.relationships[(entity1, entity2)].append(relation)
self.relationships[(entity2, entity1)].append(f"related_to_{relation}")
def get_entity_context(self, entity_name: str) -> dict:
"""Get full context for an entity."""
entity = self.entities.get(entity_name)
if not entity:
return {}
# Find related entities
related = []
for (e1, e2), relations in self.relationships.items():
if e1 == entity_name:
related.append((e2, relations))
return {
'entity': entity,
'related_entities': related,
'context_prompt': self._build_context_prompt(entity, related)
}
def _build_context_prompt(self, entity: Entity, related: list) -> str:
"""Build context prompt for LLM."""
lines = [f"About {entity.name} ({entity.entity_type}):"]
for attr, value in entity.attributes.items():
lines.append(f"- {attr}: {value}")
if related:
lines.append("\nRelated to:")
for name, relations in related:
lines.append(f"- {name}: {', '.join(relations)}")
return "\n".join(lines)
Pattern 3: Time-Weighted Memory
Implementation
# src/memory/temporal.py
from datetime import datetime, timedelta
import math
class TemporalMemory:
"""Memory with time-based decay and recency weighting."""
def __init__(
self,
half_life_days: float = 30.0,
recency_weight: float = 0.3,
importance_weight: float = 0.7
):
self.entries: list[MemoryEntry] = []
self.half_life = timedelta(days=half_life_days)
self.recency_weight = recency_weight
self.importance_weight = importance_weight
def add(self, entry: MemoryEntry):
"""Add entry with timestamp."""
entry.timestamp = datetime.now()
self.entries.append(entry)
def get_weight(self, entry: MemoryEntry) -> float:
"""Calculate time-weighted importance."""
# Recency factor (decays over time)
age = datetime.now() - entry.timestamp
recency_score = math.exp(-age / self.half_life)
# Combined score
return (
self.recency_weight * recency_score +
self.importance_weight * entry.importance
)
def retrieve(self, query: str, k: int = 10) -> list[MemoryEntry]:
"""Retrieve top-k memories by weighted score."""
# Filter by semantic similarity
candidates = self._semantic_search(query)
# Score by time-weighted importance
scored = [
(entry, self.get_weight(entry))
for entry in candidates
]
# Return top-k
scored.sort(key=lambda x: x[1], reverse=True)
return [entry for entry, _ in scored[:k]]
def prune(self, min_weight: float = 0.1):
"""Remove low-weight entries."""
self.entries = [
e for e in self.entries
if self.get_weight(e) >= min_weight
]
Pattern 4: Memory Graph
Implementation
# src/memory/graph.py
from collections import defaultdict
class MemoryGraph:
"""Graph-based memory with relationships."""
def __init__(self):
self.nodes: dict[str, dict] = {}
self.edges: dict[tuple[str, str], dict] = {}
def add_node(self, node_id: str, node_type: str, properties: dict):
"""Add a memory node."""
self.nodes[node_id] = {
'id': node_id,
'type': node_type,
'properties': properties,
'created': datetime.now()
}
def add_edge(self, from_id: str, to_id: str, relation: str, weight: float = 1.0):
"""Add relationship between nodes."""
self.edges[(from_id, to_id)] = {
'relation': relation,
'weight': weight,
'created': datetime.now()
}
def traverse(self, start_id: str, max_depth: int = 3) -> list[dict]:
"""Traverse graph from starting node."""
results = []
visited = set()
def dfs(node_id: str, depth: int, path: list):
if depth > max_depth or node_id in visited:
return
visited.add(node_id)
results.append({
'node': self.nodes[node_id],
'depth': depth,
'path': path + [node_id]
})
# Follow outgoing edges
for (from_id, to_id), edge in self.edges.items():
if from_id == node_id:
dfs(
to_id,
depth + 1,
path + [node_id]
)
dfs(start_id, 0, [])
return results
def find_related(self, node_id: str, relation_type: str = None) -> list[dict]:
"""Find all nodes related to a given node."""
related = []
for (from_id, to_id), edge in self.edges.items():
if relation_type and edge['relation'] != relation_type:
continue
if from_id == node_id:
related.append({
'node': self.nodes[to_id],
'relation': edge['relation'],
'direction': 'outgoing'
})
elif to_id == node_id:
related.append({
'node': self.nodes[from_id],
'relation': edge['relation'],
'direction': 'incoming'
})
return related
Pattern 5: Memory Consolidation
Implementation
# src/memory/consolidation.py
from difflib import SequenceMatcher
class MemoryConsolidator:
"""Consolidate and summarize memories."""
def __init__(self, similarity_threshold: float = 0.8):
self.threshold = similarity_threshold
def find_similar(self, entries: list[MemoryEntry]) -> list[list[MemoryEntry]]:
"""Group similar entries."""
groups = []
used = set()
for i, entry in enumerate(entries):
if i in used:
continue
group = [entry]
used.add(i)
for j, other in enumerate(entries[i+1:], i+1):
if j in used:
continue
similarity = self._similarity(entry.content, other.content)
if similarity > self.threshold:
group.append(other)
used.add(j)
if len(group) > 1:
groups.append(group)
return groups
def _similarity(self, a: str, b: str) -> float:
"""Calculate text similarity."""
return SequenceMatcher(None, a.lower(), b.lower()).ratio()
def consolidate_group(self, group: list[MemoryEntry]) -> MemoryEntry:
"""Merge similar entries into one."""
# Use LLM to summarize
contents = [e.content for e in group]
prompt = f"""Summarize these similar memories into one:
{chr(10).join(f'{i+1}. {c}' for i, c in enumerate(contents))}
Summary:"""
summary = call_llm(prompt)
# Keep highest importance and most recent
best = max(group, key=lambda e: (e.importance, e.timestamp))
return MemoryEntry(
content=summary,
category=best.category,
importance=best.importance,
timestamp=datetime.now(),
metadata={'consolidated_from': [e.timestamp for e in group]}
)
def run_consolidation(self, memory: HierarchicalMemory) -> int:
"""Run consolidation on long-term memory."""
similar_groups = self.find_similar(memory.long_term)
original_count = len(memory.long_term)
for group in similar_groups:
consolidated = self.consolidate_group(group)
memory.long_term.remove(group[0]) # Remove first
for entry in group[1:]:
memory.long_term.remove(entry)
memory.long_term.append(consolidated)
return original_count - len(memory.long_term)
Complete Memory System
Integration Example
# src/memory/system.py
class AgentMemorySystem:
"""Complete memory system combining all patterns."""
def __init__(self):
self.hierarchical = HierarchicalMemory()
self.entity = EntityMemory()
self.temporal = TemporalMemory()
self.graph = MemoryGraph()
self.consolidator = MemoryConsolidator()
def process_interaction(self, user_input: str, agent_response: str):
"""Process a complete interaction."""
# 1. Extract entities
entities = self.entity.extract_entities(user_input)
for entity in entities:
self.entity.update_entity(entity.name, {
'type': entity.entity_type,
'attributes': {'last_interaction': datetime.now()}
})
# 2. Add to hierarchical memory
self.hierarchical.add(
content=f"User: {user_input}\nAgent: {agent_response}",
category="conversation",
importance=self._calculate_importance(user_input, agent_response)
)
# 3. Add to temporal memory
self.temporal.add(MemoryEntry(
content=agent_response,
category="agent_output",
importance=0.5,
timestamp=datetime.now()
))
# 4. Update graph
self._update_memory_graph(user_input, agent_response)
def get_context(self, query: str) -> str:
"""Get relevant context for a query."""
# Retrieve from all memory systems
hierarchical_results = self.hierarchical.retrieve(query)
temporal_results = self.temporal.retrieve(query)
# Merge and deduplicate
all_results = self._merge_results(
hierarchical_results,
temporal_results
)
# Build context prompt
context_lines = ["Relevant context from memory:\n"]
for i, entry in enumerate(all_results[:5], 1):
context_lines.append(f"{i}. {entry.content}")
return "\n".join(context_lines)
def _calculate_importance(self, user_input: str, agent_response: str) -> float:
"""Calculate importance of interaction."""
# High importance indicators
high_importance_keywords = [
'important', 'urgent', 'critical', 'deadline',
'remember', 'save', 'favorite', 'error', 'bug'
]
input_lower = user_input.lower()
if any(kw in input_lower for kw in high_importance_keywords):
return 0.9
# Question about preferences
if '?' in user_input and any(
kw in input_lower for kw in ['like', 'prefer', 'want', 'need']
):
return 0.8
return 0.5
def _update_memory_graph(self, user_input: str, agent_response: str):
"""Update memory graph with new relationships."""
# Extract entities and create graph nodes/edges
...
def _merge_results(self, *results_lists) -> list[MemoryEntry]:
"""Merge results from multiple sources, deduplicating."""
seen = set()
merged = []
for results in results_lists:
for entry in results:
key = entry.content[:100] # Use first 100 chars as key
if key not in seen:
seen.add(key)
merged.append(entry)
return merged
def run_maintenance(self):
"""Periodic maintenance tasks."""
# Prune old memories
self.temporal.prune()
# Consolidate similar memories
self.consolidator.run_consolidation(self.hierarchical)
Best Practices
1. Memory Budget
class BoundedMemory:
"""Memory with size limits."""
MAX_ENTRIES = 1000
MAX_BYTES = 10_000_000 # ~10MB
def add(self, entry: MemoryEntry):
# Check limits before adding
if len(self.entries) >= self.MAX_ENTRIES:
self._evict_oldest()
self.entries.append(entry)
def _evict_oldest(self):
"""Remove oldest entries to make room."""
self.entries.sort(key=lambda e: e.timestamp)
self.entries = self.entries[-self.MAX_ENTRIES:]
2. Selective Storage
def should_store(user_input: str, agent_response: str) -> bool:
"""Decide if interaction should be stored."""
# Don't store trivial exchanges
trivial_patterns = [
'hello', 'hi', 'thanks', 'thank you',
'goodbye', 'bye', 'okay', 'ok'
]
if any(p in user_input.lower() for p in trivial_patterns):
return False
# Store if it contains new information
if '?' in user_input or any(
kw in user_input.lower() for kw in ['what', 'how', 'why', 'explain']
):
return True
return False
3. Privacy Considerations
class PrivateMemory:
"""Memory with privacy controls."""
SENSITIVE_PATTERNS = [
r'\b\d{3}-\d{2}-\d{4}\b', # SSN
r'\b\d{16}\b', # Credit card
r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', # Email
]
def sanitize(self, text: str) -> str:
"""Remove sensitive information."""
import re
for pattern in self.SENSITIVE_PATTERNS:
text = re.sub(pattern, '[REDACTED]', text)
return text
def add(self, entry: MemoryEntry):
entry.content = self.sanitize(entry.content)
# Then add to memory
Resources
- Mem0: https://mem0.ai/
- AgentMemory: https://agentmemory.dev/
- Zep: https://www.getzep.com/
- LangChain Memory: https://python.langchain.com/docs/modules/memory/
Last updated: May 2026