Episodic memory, short-term vs. long-term storage, vector-based retrieval and prioritized experience replay – how LLM agents learn from past experiences and leverage proven strategies for new tasks.
1. The Memory Problem
Previous articles demonstrated how an agent learns from feedback (Article 02) and how automated tests serve as reward signals (Article 03). Yet a fundamental problem remained unsolved: The agent forgets everything after each task.
Without memory, an agent repeats the same mistakes. A login implementation that succeeded on the third attempt must be rebuilt from scratch for the next login feature. The feedback signal from the previous task is lost.
"Experience replay allows the agent to learn from past experiences by storing transitions in a replay buffer and sampling from them during training." — Mnih et al. (2015), Human-level control through deep reinforcement learning, Nature 518, pp. 529–533
In Deep Reinforcement Learning, Experience Replay solves exactly this problem: past experiences are stored and provided as context for new decisions. For LLM agents, this means: episodes consisting of (task, solution, feedback, outcome) are persistently stored and embedded as context in the prompt when similar tasks arise.
2. Episodes as a Data Type
The first step: defining a clear data structure for experiences. Every agent interaction – from task description through solution to feedback – forms an Episode:
class OutcomeStatus(Enum):
SUCCESS = "success"
PARTIAL = "partial"
FAILURE = "failure"
@dataclass(frozen=True)
class Episode:
episode_id: str
task: TaskDescription
solution: str
feedback: Feedback
outcome: OutcomeStatus
timestamp: float = 0.0
iterations: int = 1
strategy_used: str = "default"
The task itself is also structured – with category and complexity:
class TaskCategory(Enum):
CODE_GENERATION = "code_generation"
DEBUGGING = "debugging"
REFACTORING = "refactoring"
TESTING = "testing"
ARCHITECTURE = "architecture"
UNKNOWN = "unknown"
@dataclass(frozen=True)
class TaskDescription:
description: str
category: TaskCategory = TaskCategory.UNKNOWN
complexity: float = 0.5
The feedback contains textual information alongside the reward from Article 03:
@dataclass(frozen=True)
class Feedback:
reward: float
message: str = ""
tests_passed: int = 0
tests_total: int = 0
@property
def pass_rate(self) -> float:
return self.tests_passed / self.tests_total if self.tests_total > 0 else 0.0
Three design decisions:
- Frozen Dataclasses: Episodes are immutable. What happened stays recorded – no retroactive manipulation.
- Categorization: TaskCategory enables skill tracking (Section 8).
- Timestamps: timestamp is essential for temporal decay (Section 5).
3. Embedding-Based Retrieval
An agent doesn't need all past episodes – only the relevant ones. The core question: how to find the most similar experiences for a new task?
The answer: Vector embeddings and cosine similarity.
"Vector databases store data as high-dimensional vectors, enabling efficient similarity search. They are the backbone of retrieval-augmented generation (RAG) systems." — Ramirez, J. (2024), Generative AI with LangChain, 2nd ed., p. 189
Cosine Similarity
Two texts are transformed into vectors of equal dimension. Their similarity is determined by the cosine of the angle between the vectors:
def cosine_similarity(a: Embedding, b: Embedding) -> float:
mag_a = _magnitude(a)
mag_b = _magnitude(b)
if mag_a == 0.0 or mag_b == 0.0:
return 0.0
return _dot_product(a, b) / (mag_a * mag_b)
| Value | Meaning |
|---|---|
| 1.0 | Identical direction (very similar) |
| 0.0 | Orthogonal (no similarity) |
| −1.0 | Opposite direction |
Embedding Abstraction
The embedding function is designed as an injectable parameter. For tests, a deterministic hash-based function suffices; in production, it is replaced by OpenAI Embeddings, Sentence-Transformers, or similar models:
Embedding = tuple[float, ...]
EmbedFn = Callable[[str], Embedding]
def simple_hash_embedding(text: str, dim: int = 64) -> Embedding:
"""Deterministic pseudo-embedding function for tests."""
values = []
for i in range(dim):
h = hash(text + str(i)) % 10000
values.append((h - 5000) / 5000.0)
mag = _magnitude(tuple(values))
if mag == 0.0:
return tuple(0.0 for _ in range(dim))
return tuple(v / mag for v in values)
This abstraction decouples the architecture from specific embedding models. The entire codebase works without external dependencies – the embedding model is a swappable component.
4. The Embedding Store
The EmbeddingStore manages pairs of (embedding, episode) and enables similarity search:
@dataclass
class EmbeddingStore:
embed_fn: EmbedFn = field(default=simple_hash_embedding)
_entries: list[tuple[Embedding, Episode]] = field(
default_factory=list, init=False
)
def add(self, episode: Episode) -> None:
embedding = self.embed_fn(episode.task.description)
self._entries.append((embedding, episode))
def search(self, query: str, top_k: int = 3) -> tuple[SearchResult, ...]:
if not self._entries:
return ()
query_embedding = self.embed_fn(query)
scored = [
SearchResult(episode=ep, similarity=cosine_similarity(query_embedding, emb))
for emb, ep in self._entries
]
scored.sort(key=lambda r: r.similarity, reverse=True)
return tuple(scored[:top_k])
❌ Naive Approach: All Episodes as Context
Agent receives task → All 500 past episodes into the prompt → Context window overflows
This doesn't work. LLMs have limited context windows, and irrelevant information degrades response quality.
✅ Vector-Based Retrieval: Top-K Most Similar Episodes
Agent receives task → Compute embedding → Retrieve top-3 similar episodes → Focused context
In production, the linear scan is replaced by FAISS, ChromaDB, or Pinecone – the interface remains identical. The EmbedFn abstraction enables this switch without architectural changes.
5. Temporal Decay: Recency Matters
Not all experiences are equally relevant. An episode from yesterday is more valuable than one from three months ago – the codebase has evolved, patterns have changed.
Exponential time decay solves the problem:
def temporal_decay(
similarity: float,
episode_time: float,
current_time: float,
half_life: float = 86400.0,
) -> float:
age = max(0.0, current_time - episode_time)
decay_factor = math.exp(-math.log(2) * age / half_life)
return similarity * decay_factor
The half-life determines when the weight drops to 50%:
| Age | Decay Factor (half-life = 1 day) |
|---|---|
| 0 hours | 1.00 |
| 12 hours | 0.71 |
| 1 day | 0.50 |
| 2 days | 0.25 |
| 1 week | 0.008 |
The concept originates from psychology: forgetting follows an exponential decay curve (Ebbinghaus, 1885). For agents, this means: old experiences don't disappear – they lose weight. A brilliant solution from a month ago is still found, but a similar solution from yesterday is preferred.
6. Prioritized Experience Replay
Semantic similarity alone is insufficient. An episode can be similar but old. Or similar but failed. Four signals determine the value of an experience:
@dataclass(frozen=True)
class ReplayWeights:
similarity: float = 0.4
recency: float = 0.3
reward: float = 0.2
outcome_bonus: float = 0.1
The priority calculation combines all four signals:
def compute_priority(
result: SearchResult,
current_time: float,
weights: ReplayWeights | None = None,
half_life: float = 86400.0,
) -> PrioritizedResult:
w = weights or ReplayWeights()
sim_score = max(0.0, result.similarity)
recency_score = temporal_decay(1.0, result.episode.timestamp, current_time, half_life)
reward_score = result.episode.feedback.reward
outcome_score = 1.0 if result.episode.outcome == OutcomeStatus.SUCCESS else (
0.5 if result.episode.outcome == OutcomeStatus.PARTIAL else 0.0
)
priority = (
w.similarity * sim_score
+ w.recency * recency_score
+ w.reward * reward_score
+ w.outcome_bonus * outcome_score
)
return PrioritizedResult(
episode=result.episode,
similarity=result.similarity,
priority=priority,
)
"Prioritized experience replay samples transitions with probability proportional to their temporal-difference error, allowing the agent to learn more frequently from surprising or informative transitions." — Schaul, T. et al. (2016), Prioritized Experience Replay, ICLR 2016
The weights are configurable as a ReplayWeights dataclass. In practice, the weighting shifts depending on the phase:
| Phase | Similarity | Recency | Reward | Outcome |
|---|---|---|---|---|
| Early phase (little experience) | 0.6 | 0.1 | 0.2 | 0.1 |
| Normal phase | 0.4 | 0.3 | 0.2 | 0.1 |
| Late phase (extensive experience) | 0.3 | 0.4 | 0.2 | 0.1 |
In the early phase, similarity dominates – every relevant experience counts. In the late phase, the weight shifts to recency – with hundreds of episodes, the most recent ones are the most valuable.
7. The Episodic Memory
The building blocks so far assemble into a two-tier memory:
@dataclass
class EpisodicMemory:
short_term_limit: int = 5
embed_fn: EmbedFn = field(default=simple_hash_embedding)
_short_term: list[Episode] = field(default_factory=list, init=False)
_long_term: EmbeddingStore = field(init=False)
def record(self, episode: Episode) -> None:
self._short_term.append(episode)
if len(self._short_term) > self.short_term_limit:
self._short_term = self._short_term[-self.short_term_limit:]
self._long_term.add(episode)
def recall(
self,
query: str,
top_k: int = 3,
current_time: float | None = None,
weights: ReplayWeights | None = None,
half_life: float = 86400.0,
) -> tuple[PrioritizedResult, ...]:
now = current_time if current_time is not None else time.time()
candidates = self._long_term.search(query, top_k=top_k * 2)
prioritized = tuple(
compute_priority(r, now, weights, half_life)
for r in candidates
)
ranked = sorted(prioritized, key=lambda p: p.priority, reverse=True)
return tuple(ranked[:top_k])
Two storage tiers:
| Tier | Capacity | Access | Content |
|---|---|---|---|
| Short-term | Last 5 episodes (FIFO) | Direct | Current context, recent attempts |
| Long-term | Unlimited | Semantic search | All past episodes |
The short-term memory is a sliding window – it always contains the last N episodes, regardless of their relevance. The long-term memory stores everything and enables targeted retrieval via embeddings.
The recall method combines all previous concepts:
1. Semantic search in the embedding store (Section 4)
2. Temporal decay on the results (Section 5)
3. Prioritized replay for the final ranking (Section 6)
8. Experience Replay: Context for the Agent
Raw episodes don't make a usable prompt. The ReplayContext aggregates the relevant information:
@dataclass(frozen=True)
class ReplayContext:
episodes: tuple[Episode, ...]
strategies_used: tuple[str, ...]
avg_reward: float
success_rate: float
The preparation extracts strategies, computes statistics, and deduplicates:
def build_replay_context(results: tuple[PrioritizedResult, ...]) -> ReplayContext:
if not results:
return ReplayContext(episodes=(), strategies_used=(), avg_reward=0.0, success_rate=0.0)
episodes = tuple(r.episode for r in results)
strategies = tuple(dict.fromkeys(ep.strategy_used for ep in episodes))
rewards = [ep.feedback.reward for ep in episodes]
successes = sum(1 for ep in episodes if ep.outcome == OutcomeStatus.SUCCESS)
return ReplayContext(
episodes=episodes,
strategies_used=strategies,
avg_reward=sum(rewards) / len(rewards),
success_rate=successes / len(episodes),
)
The formatted prompt provides the agent with targeted context:
def format_replay_prompt(context: ReplayContext, task: TaskDescription) -> str:
if not context.episodes:
return f"Task: {task.description}\nNo relevant experiences available."
lines = [
f"Task: {task.description}",
f"Category: {task.category.value}",
"",
f"Relevant experiences ({len(context.episodes)}):",
f" Average reward: {context.avg_reward:.2f}",
f" Success rate: {context.success_rate:.0%}",
f" Proven strategies: {', '.join(context.strategies_used)}",
]
# ... episode details
return "\n".join(lines)
The result: the agent sees not only the current task, but also which strategies worked for similar tasks, what the average reward was, and what specific errors occurred in the past.
9. Skill Tracking: Identifying Strengths and Weaknesses
Episodes contain category information (TaskCategory). Aggregated, this yields a competency profile:
@dataclass(frozen=True)
class SkillProfile:
category: TaskCategory
total_episodes: int
successes: int
avg_reward: float
avg_iterations: float
@property
def success_rate(self) -> float:
return self.successes / self.total_episodes if self.total_episodes > 0 else 0.0
The computation groups episodes by category:
def compute_skill_profiles(episodes: tuple[Episode, ...]) -> tuple[SkillProfile, ...]:
by_category: dict[TaskCategory, list[Episode]] = {}
for ep in episodes:
cat = ep.task.category
if cat not in by_category:
by_category[cat] = []
by_category[cat].append(ep)
profiles = []
for cat in sorted(by_category, key=lambda c: c.value):
eps = by_category[cat]
successes = sum(1 for e in eps if e.outcome == OutcomeStatus.SUCCESS)
rewards = [e.feedback.reward for e in eps]
iterations = [e.iterations for e in eps]
profiles.append(SkillProfile(
category=cat,
total_episodes=len(eps),
successes=successes,
avg_reward=sum(rewards) / len(rewards),
avg_iterations=sum(iterations) / len(iterations),
))
return tuple(profiles)
Weak categories are identified and influence strategy selection:
def identify_weak_skills(
profiles: tuple[SkillProfile, ...],
threshold: float = 0.5,
) -> tuple[SkillProfile, ...]:
return tuple(p for p in profiles if p.success_rate < threshold)
| Category | Episodes | Success Rate | Strategy Implication |
|---|---|---|---|
| Code Generation | 50 | 85% | Default strategy |
| Debugging | 20 | 35% | → Defensive strategy, more context |
| Testing | 15 | 70% | Default strategy |
| Refactoring | 8 | 25% | → Defensive strategy, provide examples |
10. Adaptive Strategy Selection
Memory enables experience-based strategy selection. Instead of a fixed strategy, the agent chooses based on three signals:
STRATEGY_POOL: dict[str, str] = {
"conservative": "Generate minimal, safe code. Avoid complexity.",
"exploratory": "Try new approaches. Use modern patterns.",
"defensive": "Focus on error handling and edge cases.",
"iterative": "Start simple, extend incrementally.",
"example_driven": "Follow successful examples.",
}
The selection follows a clear hierarchy:
def select_strategy(
task: TaskDescription,
memory: EpisodicMemory,
current_time: float | None = None,
) -> str:
now = current_time if current_time is not None else time.time()
results = memory.recall(task.description, top_k=5, current_time=now)
context = build_replay_context(results)
# 1. Prefer successful strategies
if context.success_rate > 0.6 and context.strategies_used:
strategy = context.strategies_used[0]
if strategy in STRATEGY_POOL:
return strategy
# 2. Weak categories → defensive strategy
# ... (skill-tracking-based)
# 3. Complexity-based selection
if task.complexity > 0.7:
return "iterative"
if task.complexity < 0.3:
return "exploratory"
return "iterative"
The decision logic: 1. Experience: If similar tasks succeeded with a particular strategy → use the same strategy 2. Weakness: If the task category has been identified as weak → defensive strategy 3. Complexity: High complexity → iterative. Low complexity → exploratory
These three levels form a fallback mechanism: without experience, complexity decides. With experience, knowledge from past episodes dominates.
11. Learning Curve and Diagnostics
The episode history is not just context for the agent – it is a diagnostic tool:
@dataclass(frozen=True)
class LearningSnapshot:
episode_number: int
cumulative_reward: float
rolling_success_rate: float
rolling_avg_reward: float
The learning curve computes rolling averages over a configurable window:
def compute_learning_curve(
episodes: tuple[Episode, ...],
window: int = 5,
) -> tuple[LearningSnapshot, ...]:
snapshots = []
cumulative = 0.0
for i, ep in enumerate(episodes):
cumulative += ep.feedback.reward
start = max(0, i - window + 1)
window_eps = episodes[start:i + 1]
rolling_successes = sum(
1 for e in window_eps if e.outcome == OutcomeStatus.SUCCESS
)
rolling_rewards = [e.feedback.reward for e in window_eps]
snapshots.append(LearningSnapshot(
episode_number=i + 1,
cumulative_reward=cumulative,
rolling_success_rate=rolling_successes / len(window_eps),
rolling_avg_reward=sum(rolling_rewards) / len(rolling_rewards),
))
return tuple(snapshots)
Three metrics indicate whether the memory system is working:
| Metric | Expected with Working Memory | Problem Indicator |
|---|---|---|
| Rolling Success Rate | Increases over time | Stagnates or declines |
| Rolling Avg Reward | Increases monotonically | Oscillates |
| Cumulative Reward | Grows steadily | Grows linearly (no learning effect) |
When the rolling success rate increases, the agent is effectively utilizing past experiences. If it stagnates, the retrieval may not be precise enough – or the stored experiences are not informative.
12. Comparison: Memory Architectures
| Architecture | Strength | Weakness | Use Case |
|---|---|---|---|
| No Memory | Simple, no overhead | No cross-task learning | One-off tasks |
| Context Window | Automatic, no setup | Limited, ephemeral | Short-term context |
| JSON Log | Transparent, debuggable | No semantic retrieval | Prototyping |
| Vector DB | Semantic search, scales | Setup effort | Production |
| Episodic Memory | Short-term + long-term, prioritized | More complex architecture | Complete system |
The recommendation: Start with JSON log and context window. Switch to episodic memory with vector DB once the agent regularly processes similar tasks and the experience base grows.
13. Conclusion
Memory & Experience Replay transform a stateless agent into a learning system:
- Episodes structure experiences as immutable records with task, solution, feedback, and outcome
- Embedding-based retrieval finds relevant experiences via cosine similarity without scanning all episodes
- Temporal decay weights recent experiences higher than outdated ones – analogous to human forgetting
- Prioritized experience replay combines similarity, recency, reward, and outcome into a robust relevance score
- Two-tier memory separates short-term (sliding window) and long-term (vector store) for different access patterns
- Skill tracking identifies strengths and weaknesses to adapt strategy selection
- Learning curves make progress visible and diagnosable
The architecture is deliberately modular: the embedding function is injectable, the store is swappable (from in-memory to FAISS), and the priority weighting is configurable. The intelligence resides not in the storage, but in the combination of retrieval, prioritization, and context preparation.
The next article in the series covers The Virtual Dev Team – how multiple specialized agents with different roles collaborate and improve each other.
Sources & Resources
- Mnih, V. et al. (2015): Human-level control through deep reinforcement learning. Nature 518, pp. 529–533. – Experience replay as a core mechanism in deep RL.
- Schaul, T. et al. (2016): Prioritized Experience Replay. ICLR 2016. – Prioritization by TD-error, transferable to agent systems.
- Ramirez, J. (2024): Generative AI with LangChain. 2nd ed. Packt. pp. 189 ff. – Vector databases, embedding models, RAG architecture.
- Labonne, M. et al. (2025): LLM Engineer's Handbook. Packt. pp. 331 ff. – RLHF, feedback loops, policy optimization.
- Vaid, S. & Lund, B. (2025): LLM Design Patterns. Packt. pp. 452 ff. – Memory patterns for LLM agents, context management.
- Ebbinghaus, H. (1885): Über das Gedächtnis. Duncker & Humblot. – Forgetting curve as the basis for temporal decay.
→ Full source code: src/