Episodic Memory and Experience Replay
Duration: 15 min
Episodic Memory and Experience Replay
Duration: 15 min
Overview
Episodic memory records specific experiences - complete sequences of actions, observations, and outcomes that an agent has lived through. Unlike semantic memory (which stores facts), episodic memory preserves the narrative: what happened, in what order, and what the result was. This enables agents to learn from past successes and failures, avoid repeating mistakes, and improve decision-making over time.
Key Concepts & Foundations
- What: Episodic memory stores complete experiences as sequences of (state, action, outcome) tuples
- Why: Agents that remember past experiences can avoid repeating mistakes and replicate successful strategies
- How: Record action traces, store with outcomes, retrieve similar past episodes when facing new tasks
Episodic vs. Semantic Memory
| Aspect | Episodic | Semantic | |--------|----------|----------| | Content | Specific experiences | General facts | | Structure | Sequential narrative | Key-value or graph | | Example | "Last time I deployed without tests, it broke" | "Always run tests before deploying" | | Retrieval | By situation similarity | By topic or keyword |
Experience Structure
A complete episode contains:
1. Trigger - What initiated the experience 2. Context - The state of the world when it started 3. Actions - The sequence of steps taken 4. Observations - What the agent observed after each action 5. Outcome - Success/failure and the final result 6. Reflection - What was learned
Hands-On Implementation
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
class Outcome(Enum):
SUCCESS = "success"
FAILURE = "failure"
PARTIAL = "partial"
@dataclass
class ActionStep:
action: str
observation: str
@dataclass
class Episode:
id: str
trigger: str
context: str
steps: list[ActionStep] = field(default_factory=list)
outcome: Outcome = Outcome.PARTIAL
reflection: str = ""
tags: list[str] = field(default_factory=list)
def summary(self) -> str:
steps_desc = " -> ".join(s.action for s in self.steps[:4])
if len(self.steps) > 4:
steps_desc += f" ... (+{len(self.steps) - 4} more)"
return (
f"[{self.outcome.value}] {self.trigger}\n"
f" Steps: {steps_desc}\n"
f" Reflection: {self.reflection}"
)
class EpisodicMemory:
def __init__(self):
self.episodes: list[Episode] = []
self._counter = 0
def start_episode(self, trigger: str, context: str, tags=None) -> Episode:
self._counter += 1
ep = Episode(id=f"ep-{self._counter:04d}", trigger=trigger,
context=context, tags=tags or [])
self.episodes.append(ep)
return ep
def record_step(self, episode: Episode, action: str, observation: str):
episode.steps.append(ActionStep(action=action, observation=observation))
def complete_episode(self, episode: Episode, outcome: Outcome, reflection: str = ""):
episode.outcome = outcome
episode.reflection = reflection
def recall_similar(self, trigger: str, top_k: int = 3) -> list[Episode]:
trigger_words = set(trigger.lower().split())
scored = []
for ep in self.episodes:
overlap = len(trigger_words & set(ep.trigger.lower().split()))
if overlap > 0:
scored.append((ep, overlap))
scored.sort(key=lambda x: x[1], reverse=True)
return [ep for ep, _ in scored[:top_k]]
def get_lessons_learned(self) -> list[str]:
return [f"[{ep.outcome.value}] {ep.reflection}"
for ep in self.episodes if ep.reflection]
Demo: Agent learning from past deployment experiences
memory = EpisodicMemory()ep1 = memory.start_episode(
trigger="Deploy ML model to production",
context="New model version ready, no staging test done",
tags=["deployment", "ml"]
)
memory.record_step(ep1, "Skipped staging tests", "No errors visible yet")
memory.record_step(ep1, "Deployed to production", "Model serving 500 errors")
memory.record_step(ep1, "Rolled back deployment", "Service restored after 15 min")
memory.complete_episode(ep1, Outcome.FAILURE,
reflection="Always run staging tests before production. Model had incompatible schema.")
ep2 = memory.start_episode(
trigger="Deploy ML model to production",
context="New model version ready, staging available",
tags=["deployment", "ml"]
)
memory.record_step(ep2, "Ran full test suite", "All 47 tests passed")
memory.record_step(ep2, "Deployed to staging", "Staging healthy")
memory.record_step(ep2, "Ran integration tests", "All endpoints correct")
memory.record_step(ep2, "Deployed with canary", "No errors at 5% traffic")
memory.complete_episode(ep2, Outcome.SUCCESS,
reflection="Staging + canary deployment is the safe path.")
New task: retrieve similar past experiences
similar = memory.recall_similar("Deploy updated model to production")
print("Past experiences for similar task:")
for ep in similar:
print(ep.summary())
print()
print("Lessons learned:", memory.get_lessons_learned())
Advanced Techniques
1. Outcome-Weighted Retrieval - Prefer successful episodes over failures when generating plans 2. Episode Clustering - Group similar episodes to identify recurring patterns 3. Temporal Decay - Weight recent episodes higher; old ones may be outdated 4. Counterfactual Replay - Ask "what if I had done X instead?" to generate alternatives 5. Episode Compression - Summarize long episodes into concise actionable lessons
Quiz
Q1: How does episodic memory differ from storing facts?
- A) Episodic memory is faster to query
- B) Episodic memory preserves the sequence of actions and outcomes, not just conclusions ✓
- C) Facts require more storage space
- D) Episodic memory only works with LLMs
Q2: What is experience replay in agent memory?
- A) Repeating the same action until it works
- B) Retrieving past experiences to inform current decisions and avoid past mistakes ✓
- C) Playing back audio recordings
- D) Resetting the agent to a previous state
Q3: Why should an agent record reflections after episodes?
- A) To increase storage costs
- B) To distill actionable lessons for future similar situations ✓
- C) To satisfy logging requirements
- D) To make the code more complex