AI Agent Memory Systems: Building Persistent Context
How to give AI agents long-term memory that actually works.
Without memory, every conversation with an AI agent starts from scratch. Memory systems allow agents to remember user preferences, past interactions, and learned patterns — making them progressively more useful over time.
Types of AI Agent Memory
┌──────────────────────────────────────────┐
│ Memory Types │
├──────────────────────────────────────────┤
│ │
│ Short-term (Working Memory) │
│ ├── Current conversation context │
│ ├── Last N messages │
│ └── Active tool results │
│ │
│ Medium-term (Episodic Memory) │
│ ├── Past conversations │
│ ├── User interactions history │
│ └── Resolved problems │
│ │
│ Long-term (Semantic Memory) │
│ ├── User preferences and profile │
│ ├── Learned facts and patterns │
│ └── Knowledge base │
│ │
│ Procedural Memory │
│ ├── Learned workflows │
│ ├── Successful strategies │
│ └── Skill definitions │
│ │
└──────────────────────────────────────────┘
Implementing Working Memory
class WorkingMemory:
"""Short-term memory for current conversation."""
def __init__(self, max_messages=20):
self.messages = []
self.max_messages = max_messages
self.active_context = {} # Tool results, current task
def add_message(self, role, content):
self.messages.append({
"role": role,
"content": content,
"timestamp": datetime.utcnow(),
})
# Trim if too long
if len(self.messages) > self.max_messages:
self.messages = self.messages[-self.max_messages:]
def add_context(self, key, value):
"""Add tool results or task context."""
self.active_context[key] = value
def get_context_window(self):
"""Get messages formatted for LLM context."""
return self.messages
Implementing Episodic Memory
class EpisodicMemory:
"""Memory of past conversations and interactions."""
def __init__(self, storage_backend="postgres"):
self.db = get_storage(storage_backend)
async def store_episode(self, user_id, conversation):
"""Store a completed conversation."""
# Summarize for efficient retrieval
summary = await self.summarize(conversation)
# Generate embeddings for semantic search
embedding = await embed(summary)
await self.db.insert("episodes", {
"user_id": user_id,
"summary": summary,
"embedding": embedding,
"full_conversation": conversation,
"timestamp": datetime.utcnow(),
"topics": await self.extract_topics(conversation),
"sentiment": await self.analyze_sentiment(conversation),
})
async def recall(self, user_id, query, limit=5):
"""Find relevant past conversations."""
query_embedding = await embed(query)
results = await self.db.query("""
SELECT summary, full_conversation, timestamp, topics
FROM episodes
WHERE user_id = $1
ORDER BY embedding <-> $2
LIMIT $3
""", user_id, query_embedding, limit)
return results
Implementing Semantic Memory
class SemanticMemory:
"""Long-term knowledge and user profile."""
async def learn_about_user(self, user_id, fact, source="conversation"):
"""Store a fact about the user."""
# Verify fact isn't already stored
existing = await self.db.query(
"user_facts",
{"user_id": user_id, "fact": fact},
)
if not existing:
await self.db.insert("user_facts", {
"user_id": user_id,
"fact": fact,
"source": source,
"confidence": 0.7, # Increases with corroboration
"created": datetime.utcnow(),
"last_accessed": datetime.utcnow(),
})
else:
# Increase confidence if corroborated
await self.db.update(
"user_facts",
{"id": existing[0]["id"]},
{"confidence": min(1.0, existing[0]["confidence"] + 0.1)},
)
async def get_user_profile(self, user_id):
"""Retrieve consolidated user profile."""
facts = await self.db.query(
"user_facts",
{"user_id": user_id, "confidence": {"$gte": 0.5}},
sort="-confidence",
)
return {
"preferences": [f for f in facts if f["category"] == "preference"],
"facts": [f for f in facts if f["category"] == "fact"],
"goals": [f for f in facts if f["category"] == "goal"],
"history": [f for f in facts if f["category"] == "history"],
}
Implementing Procedural Memory
class ProceduralMemory:
"""Memory of learned workflows and strategies."""
async def learn_procedure(self, name, steps, trigger, success_rate=1.0):
"""Store a successful workflow."""
await self.db.insert("procedures", {
"name": name,
"steps": steps,
"trigger": trigger, # When to use this procedure
"success_rate": success_rate,
"times_used": 1,
"created": datetime.utcnow(),
"last_used": None,
})
async def find_procedure(self, task_description):
"""Find a relevant learned procedure."""
# Semantic search for matching procedures
embedding = await embed(task_description)
procedures = await self.db.query("""
SELECT * FROM procedures
WHERE success_rate > 0.7
ORDER BY embedding <-> $1
LIMIT 3
""", embedding)
return procedures[0] if procedures else None
Memory Orchestration
class MemorySystem:
"""Coordinates all memory types."""
def __init__(self):
self.working = WorkingMemory()
self.episodic = EpisodicMemory()
self.semantic = SemanticMemory()
self.procedural = ProceduralMemory()
async def build_context(self, user_id, current_message):
"""Build rich context from all memory types."""
context = {}
# 1. User profile from semantic memory
profile = await self.semantic.get_user_profile(user_id)
context["user_profile"] = profile
# 2. Relevant past conversations
past = await self.episodic.recall(user_id, current_message, limit=3)
context["relevant_history"] = past
# 3. Applicable procedures
procedure = await self.procedural.find_procedure(current_message)
context["applicable_workflow"] = procedure
# 4. Current conversation
context["current_conversation"] = self.working.get_context_window()
return context
async def learn_from_interaction(self, user_id, conversation):
"""Update all memory types after an interaction."""
# Store episode
await self.episodic.store_episode(user_id, conversation)
# Extract and store user facts
facts = await self.extract_facts(conversation)
for fact in facts:
await self.semantic.learn_about_user(user_id, fact)
# Learn procedures if successful
if conversation.was_successful:
procedure = await self.extract_procedure(conversation)
if procedure:
await self.procedural.learn_procedure(**procedure)
Memory Compression
Large memories need compression to fit in context windows:
class MemoryCompressor:
async def compress_history(self, messages):
"""Compress old conversation history."""
if len(messages) <= 10:
return messages
# Keep last 5 messages verbatim
recent = messages[-5:]
# Summarize the rest
old = messages[:-5]
summary = await llm.summarize(old)
return [
{"role": "system", "content": f"[Previous conversation summary: {summary}]"},
*recent,
]
Vector Store Integration
from pinecone import Pinecone
class VectorMemory:
def __init__(self):
self.pc = Pinecone(api_key=os.environ["PINECONE_KEY"])
self.index = self.pc.Index("agent-memory")
async def store(self, user_id, content, metadata=None):
embedding = await embed(content)
self.index.upsert(
id=str(uuid4()),
values=embedding,
metadata={
"user_id": user_id,
"content": content,
"timestamp": datetime.utcnow().isoformat(),
**(metadata or {}),
},
)
async def search(self, user_id, query, top_k=5):
query_embedding = await embed(query)
results = self.index.query(
vector=query_embedding,
filter={"user_id": user_id},
top_k=top_k,
include_metadata=True,
)
return [r["metadata"] for r in results["matches"]]
Privacy and Memory
GDPR Considerations
class GDPRCompliantMemory:
async def delete_user_data(self, user_id):
"""Right to erasure — delete all memory."""
await self.working.clear(user_id)
await self.episodic.delete_all(user_id)
await self.semantic.delete_all(user_id)
await self.procedural.delete_user(user_id)
await self.vector_store.delete(user_id=user_id)
async def export_user_data(self, user_id):
"""Right to portability — export all memory."""
return {
"episodes": await self.episodic.get_all(user_id),
"facts": await self.semantic.get_all(user_id),
"procedures": await self.procedural.get_user(user_id),
}
Conclusion
Effective memory is what separates a chatbot from a truly intelligent agent. By implementing multi-layer memory — working, episodic, semantic, and procedural — you create agents that become more useful with every interaction.
Learn More
- AI Agent Memory Systems: Persistent Context
- RAG vs Fine-Tuning for AI Agents
- Skill-Based AI Architecture
- GDPR Compliance for AI Tools
Find memory and context tools on SkillExchange.