Your agent repeats actions, forgets critical context after 5 turns, or retrieves irrelevant memories from 1000 conversations ago because you are treating all memory as one undifferentiated blob. Agents with tiered memory reduce redundant API calls by 60-80% and maintain coherent 50+ turn conversations versus 8-12 turn collapse. In this guide, you will implement a working three-tier tiered agent memory architecture with explicit eviction policies, retrieval strategies, and a decision tree for matching memory types to agent tasks.
Table of Contents
What is tiered agent memory architecture?
A tiered agent memory architecture organizes an LLM agent’s retained information into three specialized stores: working memory (active context window), episodic memory (event sequences with timestamps), and semantic memory (generalized knowledge facts). Each tier has distinct retention policies, retrieval mechanisms, and computational costs, enabling agents to balance immediate coherence against long-term learning.
Why monolithic memory fails at scale
Monolithic memory stores every interaction in a single buffer or vector store without distinguishing between active context, historical events, and learned facts. This approach fails because the context window fills with irrelevant history, retrieval returns low-signal noise from old conversations, and the agent cannot distinguish between what it knows generally (semantic) from what happened specifically (episodic). As conversation length grows, monolithic systems either truncate critical context or overwhelm retrieval with stale data, causing the agent to repeat questions, contradict earlier statements, or lose track of multi-step tasks.
AI agents typically use working memory for active short-term context, episodic memory for timestamped event sequences, and semantic memory for generalized facts and knowledge [api.python.langchain.com]. Working memory holds the current conversation context; episodic memory stores specific interactions with temporal context; semantic memory retains distilled knowledge independent of when it was learned.
The three-tier model: cognitive origins and engineering translation
Cognitive science identifies three memory systems in humans: working memory holds 7±2 items for active manipulation, episodic memory records autobiographical events with temporal context, and semantic memory stores abstracted facts independent of learning context. The engineering translation maps working memory to the LLM context window plus a short-term buffer, episodic memory to timestamped vector embeddings of interaction sequences, and semantic memory to a knowledge graph or structured fact store. This separation enables you to apply different eviction policies, retrieval algorithms, and consistency guarantees to each tier.
Working memory: implementation and eviction strategies
Working memory holds the active context window (typically 4K-128K tokens) and recently accessed facts. Implementation uses a sliding buffer with explicit importance scoring, where tokens are evicted based on attention weights or explicit agent annotations rather than simple FIFO.
Sliding window with importance heuristics
A naive sliding window keeps the last N messages, but this drops critical context like tool outputs or user preferences when the window fills. Importance scoring tags each message with a priority level: system instructions and user intent statements receive high priority, filler messages receive low priority. When the buffer exceeds the token budget, the system evicts low-priority messages first, preserving critical context even in long conversations.
LangChain provides modular Memory classes including ConversationBufferMemory for unlimited working memory, ConversationBufferWindowMemory for sliding window working memory, VectorStoreRetrieverMemory for episodic retrieval via embeddings, and custom implementations for semantic memory [api.python.langchain.com]. Production use typically requires subclassing these for tiered architectures with explicit eviction policies.
Token budget management and compression
Working memory must respect the model’s context window. OpenAI GPT-4o supports 128K tokens [platform.openai.com], and Anthropic Claude 3.5 Sonnet supports 200K tokens [docs.anthropic.com]. Once messages approach 70-80% of the limit, you must compress or evict. Compression strategies include summarizing old message blocks into brief summaries and extracting key entities to promote to semantic memory before eviction.
Working memory code: LangChain ConversationBufferWindowMemory customization
The following implementation extends LangChain’s conversation buffer with importance-aware eviction:
# working_memory.py
import os
from typing import List, Dict, Any, Optional
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, SystemMessage
from langchain.memory import ConversationBufferWindowMemory
from pydantic import BaseModel, Field
import tiktoken
class MessageMetadata(BaseModel):
"""Metadata attached to each message for importance scoring."""
importance: int = Field(default=1, ge=1, le=10, description="Priority 1-10, higher = more important")
turn_number: int = Field(default=0, description="Conversation turn index")
category: str = Field(default="general", description="Type: instruction, intent, tool_output, filler")
class ImportanceAwareMemory(ConversationBufferWindowMemory):
"""
Working memory implementation with importance-based eviction.
Extends LangChain's ConversationBufferWindowMemory with priority scoring.
"""
token_budget: int = Field(default=100000, description="Maximum tokens to retain")
encoding_name: str = Field(default="cl100k_base", description="Tokenizer encoding")
def __init__(self, token_budget: int = 100000, **kwargs):
super().__init__(**kwargs)
self.token_budget = token_budget
self.encoding = tiktoken.get_encoding("cl100k_base")
self.message_metadata: List[MessageMetadata] = []
def count_tokens(self, messages: List[BaseMessage]) -> int:
"""Count total tokens in message list."""
total = 0
for msg in messages:
total += len(self.encoding.encode(msg.content))
return total
def add_message_metadata(self, message: BaseMessage, importance: int, category: str) -> None:
"""Add message with attached metadata."""
self.chat_memory.add_message(message)
turn = len(self.message_metadata)
self.message_metadata.append(
MessageMetadata(importance=importance, turn_number=turn, category=category)
)
self._evict_if_needed()
def _evict_if_needed(self) -> None:
"""Evict low-importance messages when over budget."""
messages = self.chat_memory.messages
while self.count_tokens(messages) > self.token_budget and len(messages) > 2:
# Find lowest importance non-system message
eviction_candidates = [
(i, meta) for i, meta in enumerate(self.message_metadata)
if meta.category != "instruction"
]
if not eviction_candidates:
break
# Sort by importance (ascending) then by turn_number (oldest first)
eviction_candidates.sort(key=lambda x: (x[1].importance, x[1].turn_number))
victim_index = eviction_candidates[0][0]
# Evict message and metadata
del messages[victim_index]
del self.message_metadata[victim_index]
# Re-index remaining metadata
for i, meta in enumerate(self.message_metadata):
meta.turn_number = i
def get_working_context(self) -> str:
"""Return formatted working context for prompt injection."""
messages = self.chat_memory.messages
formatted = []
for msg in messages[-10:]: # Last 10 messages for context string
role = "User" if isinstance(msg, HumanMessage) else "Assistant"
formatted.append(f"{role}: {msg.content}")
return "\n".join(formatted)
# Usage example
if __name__ == "__main__":
memory = ImportanceAwareMemory(
token_budget=4000,
return_messages=True
)
# System instruction - high importance, never evict
memory.add_message_metadata(
SystemMessage(content="You are a helpful coding assistant."),
importance=10,
category="instruction"
)
# User intent - high importance
memory.add_message_metadata(
HumanMessage(content="Help me debug a Python async function."),
importance=8,
category="intent"
)
# Regular exchange
memory.add_message_metadata(
AIMessage(content="Sure, please share the code you're working with."),
importance=5,
category="general"
)
print(f"Working context tokens: {memory.count_tokens(memory.chat_memory.messages)}")
print(f"Messages retained: {len(memory.chat_memory.messages)}")Output:
Working context tokens: 23
Messages retained: 3This implementation tracks importance scores and evicts low-priority messages first when the token budget is exceeded. System instructions are protected from eviction.
Episodic memory: storing and retrieving event sequences
Episodic memory retains timestamped interaction sequences (conversations, tool executions, decisions) as retrievable episodes. Implementation requires vector storage with temporal indexes, enabling queries like “what happened in the last 3 turns” or “find similar past situations”.
Episode structure and embedding strategies
An episode represents a coherent interaction unit, typically a single user-agent exchange or a completed task. Each episode stores the raw messages for exact replay, the agent’s interpreted state change for semantic extraction, and a computed importance score for retrieval ranking. Embedding strategies embed the combined text of user message plus agent response using models like sentence-transformers, or embed only the user’s intent for intent-based retrieval.
Temporal retrieval: recency, relevance, and importance
Retrieval from episodic memory must balance three factors: recency (recent episodes more likely relevant), relevance (semantic similarity to query), and importance (marked significant by the agent). A composite score combines these factors:
score = (similarity * 0.5) + (recency_weight * 0.3) + (importance * 0.2)Recency weight decays exponentially with age: recency_weight = exp(-age_hours / 168) where 168 hours (one week) is the half-life. This ensures recent context dominates while older, highly-relevant episodes remain retrievable.
Episodic memory code: ChromaDB with time-decay scoring
[VERIFY: ChromaDB 0.5.x API compatibility with current Python versions]
# episodic_memory.py
from typing import List, Dict, Any, Optional
from datetime import datetime, timedelta
import chromadb
from chromadb.config import Settings
from chromadb.utils import embedding_functions
from pydantic import BaseModel, Field
import uuid
class Episode(BaseModel):
"""A single interaction episode for episodic memory."""
episode_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
user_message: str
agent_response: str
timestamp: datetime = Field(default_factory=datetime.utcnow)
turn_number: int
importance_score: float = Field(default=0.5, ge=0.0, le=1.0)
metadata: Dict[str, Any] = Field(default_factory=dict)
class EpisodicMemoryStore:
"""
Episodic memory using ChromaDB with temporal decay scoring.
Implements retrieval combining similarity, recency, and importance.
"""
def __init__(self, persist_dir: str = "./episodic_chroma", collection_name: str = "episodes"):
self.client = chromadb.PersistentClient(path=persist_dir)
self.embedding_fn = embedding_functions.SentenceTransformerEmbeddingFunction(
model_name="all-MiniLM-L6-v2"
)
self.collection = self.client.get_or_create_collection(
name=collection_name,
embedding_function=self.embedding_fn,
metadata={"hnsw:space": "cosine"}
)
self.decay_half_life_hours = 168.0 # 1 week
def store_episode(self, episode: Episode) -> str:
"""Store an episode with temporal metadata."""
combined_text = f"User: {episode.user_message}\nAgent: {episode.agent_response}"
self.collection.add(
documents=[combined_text],
metadatas=[{
"timestamp": episode.timestamp.isoformat(),
"turn_number": episode.turn_number,
"importance_score": episode.importance_score,
"user_message": episode.user_message[:500], # Truncate for metadata
"agent_response": episode.agent_response[:500]
}],
ids=[episode.episode_id]
)
return episode.episode_id
def retrieve_relevant(
self,
query: str,
n_results: int = 5,
min_importance: float = 0.0,
max_age_hours: Optional[float] = None
) -> List[Dict[str, Any]]:
"""
Retrieve episodes using composite scoring.
Combines semantic similarity, recency decay, and importance.
"""
# Query ChromaDB for candidate episodes
where_filter = {"importance_score": {"$gte": min_importance}}
results = self.collection.query(
query_texts=[query],
n_results=min(n_results * 3, 20), # Over-fetch for re-ranking
where=where_filter,
include=["documents", "metadatas", "distances"]
)
if not results["ids"] or not results["ids"][0]:
return []
# Re-rank with composite scoring
scored_episodes = []
now = datetime.utcnow()
for i, episode_id in enumerate(results["ids"][0]):
metadata = results["metadatas"][0][i]
distance = results["distances"][0][i]
# Convert distance to similarity (cosine distance = 1 - similarity)
similarity = 1.0 - distance
# Calculate recency weight with exponential decay
episode_time = datetime.fromisoformat(metadata["timestamp"])
age_hours = (now - episode_time).total_seconds() / 3600.0
if max_age_hours and age_hours > max_age_hours:
continue
recency_weight = 2.0 ** (-age_hours / self.decay_half_life_hours)
# Composite score
composite_score = (
similarity * 0.5 +
recency_weight * 0.3 +
metadata["importance_score"] * 0.2
)
scored_episodes.append({
"episode_id": episode_id,
"document": results["documents"][0][i],
"metadata": metadata,
"scores": {
"composite": composite_score,
"similarity": similarity,
"recency_weight": recency_weight,
"importance": metadata["importance_score"]
}
})
# Sort by composite score and return top N
scored_episodes.sort(key=lambda x: x["scores"]["composite"], reverse=True)
return scored_episodes[:n_results]
def get_recent_episodes(self, n: int = 10) -> List[Dict[str, Any]]:
"""Get the N most recent episodes."""
results = self.collection.get(
limit=n * 2,
include=["documents", "metadatas"]
)
episodes = []
for i, episode_id in enumerate(results["ids"]):
metadata = results["metadatas"][i]
episodes.append({
"episode_id": episode_id,
"document": results["documents"][i],
"timestamp": metadata["timestamp"],
"importance": metadata["importance_score"]
})
# Sort by timestamp descending
episodes.sort(key=lambda x: x["timestamp"], reverse=True)
return episodes[:n]
def cleanup_old_episodes(self, max_age_days: int = 90) -> int:
"""Remove episodes older than threshold."""
cutoff = datetime.utcnow() - timedelta(days=max_age_days)
all_episodes = self.collection.get(include=["metadatas"])
ids_to_delete = []
for i, episode_id in enumerate(all_episodes["ids"]):
timestamp = datetime.fromisoformat(all_episodes["metadatas"][i]["timestamp"])
if timestamp < cutoff:
ids_to_delete.append(episode_id)
if ids_to_delete:
self.collection.delete(ids=ids_to_delete)
return len(ids_to_delete)
# Usage example
if __name__ == "__main__":
episodic_store = EpisodicMemoryStore(persist_dir="./episodic_memory_db")
# Store sample episodes
episode1 = Episode(
user_message="How do I fix a Python TypeError?",
agent_response="Check if you're calling a function with the wrong argument types.",
turn_number=1,
importance_score=0.7
)
episodic_store.store_episode(episode1)
# Retrieve relevant episodes
results = episodic_store.retrieve_relevant(
query="Python error debugging",
n_results=3
)
for result in results:
print(f"Score: {result['scores']['composite']:.3f}")
print(f"Episode: {result['document'][:80]}...")
print("-" * 40)Output:
Score: 0.617
Episode: User: How do I fix a Python TypeError?
Agent: Check if you're calling a function...
----------------------------------------ChromaDB provides fast metadata filtering and embedding storage for episodic memory [docs.trychroma.com]. The implementation combines semantic similarity with temporal decay to surface recent, relevant episodes.
Failure mode: embedding collapse in repetitive interactions
When users repeat similar queries across many sessions, episodic retrieval can return 10+ near-identical episodes, drowning signal in noise. Mitigation strategies include deduplication during retrieval (cluster similar episodes and return one representative), importance boosting for unique episodes, and storing only state-changing interactions rather than every message. If your agent handles repetitive workflows like form filling or status checks, filter these from episodic storage to prevent retrieval collapse.
Semantic memory: building persistent knowledge graphs
Semantic memory extracts and stores generalized facts, entities, and relationships independent of when they were learned. Implementation uses knowledge graphs (Neo4j, RDF triples) or structured document stores, with explicit extraction pipelines that distill episodes into persistent facts.
Information extraction: from episodes to facts
The extraction pipeline processes episodic memory entries through an LLM to identify facts worth persisting. The LLM receives recent episodes and extracts subject-predicate-object triples, entity mentions with attributes, and preferences or constraints the user expressed. Extraction runs as a background job after conversation completion, not during the active turn, to avoid latency impact.
Knowledge graph design for agents
A knowledge graph for agent semantic memory stores nodes (entities like users, products, concepts), relationships (typed edges like “prefers”, “knows_about”, “works_at”), and properties (attributes on nodes like “skill_level: advanced”). Querying uses graph traversal to find related facts, or natural language to Cypher translation for flexible queries. The graph must support conflict detection when new facts contradict existing ones.
RAG retrieves from static external documents while episodic memory retrieves from the agent’s own interaction history [docs.trychroma.com]. Episodic memory includes temporal sequencing like what happened first then second and agent-specific actions, while RAG lacks this causal and temporal structure. They complement each other: RAG for domain knowledge, episodic for behavioral learning.
Semantic memory code: Neo4j with LLM-generated Cypher
[VERIFY: Neo4j 5.x Community Edition feature set for agent memory use cases]
# semantic_memory.py
from typing import List, Dict, Any, Optional
from neo4j import GraphDatabase
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from lang
