"""
Pattern 104: Memory Replay & Rehearsal
Description:
The Memory Replay & Rehearsal pattern implements mechanisms for re-experiencing
and strengthening memories through replay, similar to how the human brain
consolidates memories during sleep and practice. This pattern is crucial for
learning, skill retention, and memory consolidation in long-running agent systems.
Memory replay involves re-activating stored experiences to strengthen neural
patterns, transfer information from short-term to long-term storage, and discover
new insights through recombination. Rehearsal can be exact (episodic replay) or
generative (synthetic variations), and can occur during downtime, triggered by
context, or scheduled periodically.
This pattern supports various replay strategies including experience replay
(reinforcement learning), prioritized replay (focus on important memories),
interleaved replay (mixing old and new), and generative replay (creating
variations). It implements replay scheduling, priority calculation, and
consolidation effects.
Key Components:
1. Memory Buffer: Store of experiences to replay
2. Replay Strategy: Algorithm for selecting memories
3. Replay Scheduler: When to perform replay
4. Priority Calculator: Importance scoring
5. Consolidation Engine: Strengthening mechanism
6. Synthetic Generator: Create variations
7. Replay Metrics: Track effectiveness
Replay Types:
1. Experience Replay: Re-live past experiences
2. Prioritized Replay: Focus on important/surprising
3. Interleaved Replay: Mix old and new memories
4. Generative Replay: Create synthetic variations
5. Forward Replay: Chronological order
6. Reverse Replay: Reverse chronological
7. Random Replay: Random sampling
Replay Timing:
1. Offline Replay: During downtime/sleep
2. Online Replay: During active tasks
3. Triggered Replay: By context or cues
4. Periodic Replay: Scheduled intervals
5. Event-driven Replay: After significant events
Use Cases:
- Reinforcement learning agents
- Skill learning and retention
- Memory consolidation systems
- Experience-based learning
- Pattern recognition improvement
- Knowledge transfer and generalization
- Continuous learning systems
Advantages:
- Strengthens important memories
- Improves learning efficiency
- Enables offline learning
- Prevents catastrophic forgetting
- Discovers new patterns
- Transfers knowledge
- Simulates "sleep consolidation"
Challenges:
- Computational cost of replay
- Selecting which memories to replay
- Balancing old vs. new experiences
- Avoiding overfitting to replayed memories
- Determining replay frequency
- Managing replay buffer size
LangChain Implementation:
This implementation uses LangChain for:
- LLM-based memory importance assessment
- Synthetic memory generation
- Pattern extraction from replay
- Learning from replayed experiences
Production Considerations:
- Schedule replay during low-activity periods
- Implement priority-based sampling
- Monitor replay effectiveness metrics
- Limit replay buffer size
- Balance replay with new experiences
- Track memory strengthening
- Implement incremental consolidation
- Consider computational budget
"""
import os
import time
from typing import List, Dict, Any, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from collections import deque
import random
from dotenv import load_dotenv
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
load_dotenv()
class ReplayStrategy(Enum):
"""Strategy for replaying memories."""
PRIORITIZED = "prioritized" # High-priority first
RANDOM = "random" # Random sampling
RECENT = "recent" # Most recent first
OLDEST = "oldest" # Oldest first
INTERLEAVED = "interleaved" # Mix old and new
REVERSE = "reverse" # Reverse chronological
class ReplayTrigger(Enum):
"""When replay is triggered."""
MANUAL = "manual"
PERIODIC = "periodic"
THRESHOLD = "threshold" # When buffer is full
CONTEXTUAL = "contextual" # By context match
EVENT_DRIVEN = "event_driven"
@dataclass
class MemoryTrace:
"""
Memory trace for replay.
Attributes:
trace_id: Unique identifier
content: Memory content
context: Context when created
timestamp: Creation time
importance: Importance score
replay_count: Times replayed
strength: Memory strength
last_replayed: Last replay time
success_metric: Performance metric
metadata: Additional information
"""
trace_id: str
content: str
context: Dict[str, Any]
timestamp: datetime
importance: float
replay_count: int = 0
strength: float = 0.5
last_replayed: Optional[datetime] = None
success_metric: float = 0.5
metadata: Dict[str, Any] = field(default_factory=dict)
class MemoryReplaySystem:
"""
Memory replay and rehearsal system.
This system manages the replay of stored memories for consolidation,
learning, and pattern discovery.
"""
def __init__(
self,
buffer_size: int = 1000,
replay_batch_size: int = 10,
strategy: ReplayStrategy = ReplayStrategy.PRIORITIZED,
strengthening_rate: float = 0.1,
temperature: float = 0.3
):
"""
Initialize memory replay system.
Args:
buffer_size: Maximum memories to store
replay_batch_size: Memories per replay session
strategy: Replay selection strategy
strengthening_rate: How much to strengthen on replay
temperature: LLM temperature
"""
self.buffer_size = buffer_size
self.replay_batch_size = replay_batch_size
self.strategy = strategy
self.strengthening_rate = strengthening_rate
self.memory_buffer: deque = deque(maxlen=buffer_size)
self.replay_history: List[Dict[str, Any]] = []
self.llm = ChatOpenAI(temperature=temperature, model="gpt-3.5-turbo")
self.trace_counter = 0
def store_memory(
self,
content: str,
context: Dict[str, Any],
importance: float = 0.5,
success_metric: float = 0.5
) -> MemoryTrace:
"""
Store memory for potential replay.
Args:
content: Memory content
context: Context information
importance: Importance score
success_metric: Success/performance metric
Returns:
Created memory trace
"""
self.trace_counter += 1
trace = MemoryTrace(
trace_id=f"trace_{self.trace_counter}",
content=content,
context=context,
timestamp=datetime.now(),
importance=importance,
success_metric=success_metric
)
self.memory_buffer.append(trace)
return trace
def calculate_priority(self, trace: MemoryTrace) -> float:
"""
Calculate replay priority for memory.
Args:
trace: Memory trace
Returns:
Priority score
"""
# Factors: importance, recency, success, replay count
now = datetime.now()
# Recency (newer = higher priority)
time_since = (now - trace.timestamp).total_seconds() / 3600
recency_score = 1.0 / (1.0 + time_since)
# Inverse replay count (less replayed = higher priority)
replay_score = 1.0 / (1.0 + trace.replay_count)
# Combine factors
priority = (
trace.importance * 0.4 +
recency_score * 0.2 +
trace.success_metric * 0.2 +
replay_score * 0.2
)
return priority
def select_memories_for_replay(
self,
n: Optional[int] = None
) -> List[MemoryTrace]:
"""
Select memories to replay based on strategy.
Args:
n: Number of memories (default: replay_batch_size)
Returns:
List of memory traces to replay
"""
if n is None:
n = self.replay_batch_size
if not self.memory_buffer:
return []
n = min(n, len(self.memory_buffer))
memories = list(self.memory_buffer)
if self.strategy == ReplayStrategy.PRIORITIZED:
# Sort by priority
priorities = [(trace, self.calculate_priority(trace)) for trace in memories]
priorities.sort(key=lambda x: x[1], reverse=True)
selected = [trace for trace, _ in priorities[:n]]
elif self.strategy == ReplayStrategy.RANDOM:
selected = random.sample(memories, n)
elif self.strategy == ReplayStrategy.RECENT:
selected = sorted(memories, key=lambda x: x.timestamp, reverse=True)[:n]
elif self.strategy == ReplayStrategy.OLDEST:
selected = sorted(memories, key=lambda x: x.timestamp)[:n]
elif self.strategy == ReplayStrategy.INTERLEAVED:
# Mix old and new
sorted_by_time = sorted(memories, key=lambda x: x.timestamp)
selected = []
for i in range(n):
if i % 2 == 0 and i // 2 = 0:
selected.append(sorted_by_time[-(i // 2 + 1)]) # New
elif self.strategy == ReplayStrategy.REVERSE:
selected = sorted(memories, key=lambda x: x.timestamp, reverse=True)[:n]
else:
selected = memories[:n]
return selected
def replay(
self,
n: Optional[int] = None,
consolidate: bool = True
) -> Dict[str, Any]:
"""
Perform memory replay.
Args:
n: Number of memories to replay
consolidate: Whether to strengthen memories
Returns:
Replay statistics
"""
selected = self.select_memories_for_replay(n)
if not selected:
return {"replayed": 0, "strengthened": 0}
replayed_count = 0
strengthened_count = 0
for trace in selected:
# Update replay statistics
trace.replay_count += 1
trace.last_replayed = datetime.now()
replayed_count += 1
# Consolidate (strengthen) memory
if consolidate:
old_strength = trace.strength
trace.strength = min(1.0, trace.strength + self.strengthening_rate)
if trace.strength > old_strength:
strengthened_count += 1
# Record replay session
session = {
"timestamp": datetime.now(),
"replayed_count": replayed_count,
"strengthened_count": strengthened_count,
"strategy": self.strategy.value,
"traces": [t.trace_id for t in selected]
}
self.replay_history.append(session)
return {
"replayed": replayed_count,
"strengthened": strengthened_count,
"avg_strength_increase": self.strengthening_rate if consolidate else 0
}
def generate_synthetic_memory(
self,
base_memory: MemoryTrace,
variation: str = "similar"
) -> MemoryTrace:
"""
Generate synthetic memory variation.
Args:
base_memory: Memory to vary
variation: Type of variation
Returns:
New synthetic memory
"""
# Generate variation using LLM
prompt = ChatPromptTemplate.from_template(
"Given this memory: {content}\n\n"
"Generate a {variation} variation of this memory. "
"Keep the core concept but change details or perspective. "
"Return only the new memory content.\n\n"
"Variation:"
)
chain = prompt | self.llm | StrOutputParser()
synthetic_content = chain.invoke({
"content": base_memory.content,
"variation": variation
})
# Create synthetic memory
synthetic = self.store_memory(
content=synthetic_content.strip(),
context={**base_memory.context, "synthetic": True, "base": base_memory.trace_id},
importance=base_memory.importance * 0.8,
success_metric=base_memory.success_metric
)
return synthetic
def generative_replay(
self,
n_synthetic: int = 5,
min_importance: float = 0.6
) -> List[MemoryTrace]:
"""
Perform generative replay with synthetic variations.
Args:
n_synthetic: Number of synthetic memories to generate
min_importance: Minimum importance for base memories
Returns:
List of generated synthetic memories
"""
# Select high-importance memories as bases
important_memories = [
trace for trace in self.memory_buffer
if trace.importance >= min_importance
]
if not important_memories:
return []
# Generate synthetic variations
synthetic_memories = []
for _ in range(n_synthetic):
base = random.choice(important_memories)
synthetic = self.generate_synthetic_memory(base)
synthetic_memories.append(synthetic)
return synthetic_memories
def get_statistics(self) -> Dict[str, Any]:
"""Get replay statistics."""
if not self.memory_buffer:
return {
"total_memories": 0,
"total_replays": 0,
"avg_strength": 0,
"avg_replay_count": 0
}
memories = list(self.memory_buffer)
return {
"total_memories": len(memories),
"total_replays": len(self.replay_history),
"avg_strength": sum(m.strength for m in memories) / len(memories),
"avg_replay_count": sum(m.replay_count for m in memories) / len(memories),
"buffer_usage": len(memories) / self.buffer_size,
"strategy": self.strategy.value
}
class ReplayLearningAgent:
"""
Agent that learns through memory replay.
This agent uses replay to consolidate memories and improve
performance through rehearsal.
"""
def __init__(
self,
buffer_size: int = 500,
replay_interval: int = 10,
temperature: float = 0.5
):
"""
Initialize replay learning agent.
Args:
buffer_size: Memory buffer size
replay_interval: Interactions between replays
temperature: LLM temperature
"""
self.replay_system = MemoryReplaySystem(
buffer_size=buffer_size,
strategy=ReplayStrategy.PRIORITIZED
)
self.replay_interval = replay_interval
self.interaction_count = 0
self.llm = ChatOpenAI(temperature=temperature, model="gpt-3.5-turbo")
def experience(
self,
content: str,
context: Dict[str, Any],
importance: float = 0.5,
success: float = 0.5
):
"""
Record an experience.
Args:
content: Experience content
context: Context information
importance: Importance score
success: Success metric
"""
self.replay_system.store_memory(content, context, importance, success)
self.interaction_count += 1
# Periodic replay
if self.interaction_count % self.replay_interval == 0:
self.replay_system.replay()
def consolidate(self):
"""Perform memory consolidation through replay."""
self.replay_system.replay(consolidate=True)
def get_strong_memories(self, n: int = 10) -> List[MemoryTrace]:
"""
Get strongest memories.
Args:
n: Number of memories
Returns:
List of strong memories
"""
memories = list(self.replay_system.memory_buffer)
memories.sort(key=lambda x: x.strength, reverse=True)
return memories[:n]
def demonstrate_memory_replay():
"""Demonstrate memory replay and rehearsal pattern."""
print("=" * 80)
print("MEMORY REPLAY & REHEARSAL PATTERN DEMONSTRATION")
print("=" * 80)
# Example 1: Basic replay system
print("\n" + "=" * 80)
print("Example 1: Memory Storage and Replay")
print("=" * 80)
replay_system = MemoryReplaySystem(
buffer_size=50,
replay_batch_size=5,
strategy=ReplayStrategy.PRIORITIZED
)
print("\nStoring experiences...")
experiences = [
("Learned Python basics", {"topic": "python"}, 0.7, 0.8),
("Made a coding mistake", {"topic": "error"}, 0.9, 0.3),
("Successfully debugged issue", {"topic": "debug"}, 0.8, 0.9),
("Read documentation", {"topic": "learning"}, 0.5, 0.6),
("Completed project milestone", {"topic": "achievement"}, 0.9, 0.9),
]
for content, context, importance, success in experiences:
replay_system.store_memory(content, context, importance, success)
print(f" ✓ Stored: {content}")
stats = replay_system.get_statistics()
print(f"\nBuffer status: {stats['total_memories']} memories stored")
# Perform replay
print("\nPerforming prioritized replay...")
result = replay_system.replay(n=3, consolidate=True)
print(f" Replayed: {result['replayed']} memories")
print(f" Strengthened: {result['strengthened']} memories")
# Example 2: Different replay strategies
print("\n" + "=" * 80)
print("Example 2: Comparing Replay Strategies")
print("=" * 80)
strategies = [
ReplayStrategy.PRIORITIZED,
ReplayStrategy.RANDOM,
ReplayStrategy.RECENT,
ReplayStrategy.OLDEST
]
for strategy in strategies:
test_system = MemoryReplaySystem(
buffer_size=20,
replay_batch_size=3,
strategy=strategy
)
# Add memories with timestamps
for i in range(10):
test_system.store_memory(
f"Memory {i}",
{"index": i},
importance=0.3 + (i * 0.05),
success_metric=0.5
)
time.sleep(0.01)
print(f"\nStrategy: {strategy.value}")
selected = test_system.select_memories_for_replay(3)
for trace in selected:
print(f" → {trace.content} (importance: {trace.importance:.2f})")
# Example 3: Strengthening through replay
print("\n" + "=" * 80)
print("Example 3: Memory Strengthening Through Replay")
print("=" * 80)
strengthen_system = MemoryReplaySystem(
strengthening_rate=0.15
)
# Store important memory
trace = strengthen_system.store_memory(
"Critical information to remember",
{"type": "important"},
importance=0.8,
success_metric=0.9
)
print(f"\nMemory: '{trace.content}'")
print(f"Initial strength: {trace.strength:.2f}")
print(f"\nReplaying memory multiple times...")
for i in range(1, 6):
strengthen_system.replay(n=1, consolidate=True)
print(f" After replay {i}: strength = {trace.strength:.2f}, "
f"replay_count = {trace.replay_count}")
# Example 4: Priority-based replay
print("\n" + "=" * 80)
print("Example 4: Priority-Based Memory Selection")
print("=" * 80)
priority_system = MemoryReplaySystem(
strategy=ReplayStrategy.PRIORITIZED
)
# Add memories with different priorities
memory_data = [
("Low priority memory", 0.2, 0.3),
("Medium priority memory", 0.5, 0.5),
("High priority memory", 0.9, 0.8),
("Critical memory", 0.95, 0.9),
("Another low priority", 0.3, 0.4),
]
traces = []
for content, importance, success in memory_data:
trace = priority_system.store_memory(
content,
{"type": "test"},
importance,
success
)
traces.append(trace)
time.sleep(0.01)
print("\nMemories and their priorities:")
for trace in traces:
priority = priority_system.calculate_priority(trace)
print(f" [{priority:.3f}] {trace.content}")
print(f"\nSelecting top 3 for replay...")
selected = priority_system.select_memories_for_replay(3)
for trace in selected:
print(f" ✓ {trace.content}")
# Example 5: Interleaved replay
print("\n" + "=" * 80)
print("Example 5: Interleaved Replay (Old + New)")
print("=" * 80)
interleaved_system = MemoryReplaySystem(
strategy=ReplayStrategy.INTERLEAVED,
replay_batch_size=6
)
# Add memories over time
for i in range(10):
interleaved_system.store_memory(
f"Memory from time {i}",
{"time": i},
importance=0.5
)
time.sleep(0.01)
print("\nInterleaved replay (mixing old and new):")
selected = interleaved_system.select_memories_for_replay(6)
for trace in selected:
print(f" → {trace.content}")
# Example 6: Generative replay
print("\n" + "=" * 80)
print("Example 6: Generative Replay with Synthetic Memories")
print("=" * 80)
gen_system = MemoryReplaySystem()
# Store base memories
base_memories = [
("Python is good for data science", {"topic": "python"}, 0.8),
("Machine learning requires clean data", {"topic": "ml"}, 0.9),
]
print("\nBase memories:")
for content, context, importance in base_memories:
trace = gen_system.store_memory(content, context, importance, 0.7)
print(f" • {trace.content}")
print("\nGenerating synthetic variations...")
synthetic = gen_system.generative_replay(n_synthetic=3, min_importance=0.7)
print(f"\nGenerated {len(synthetic)} synthetic memories:")
for trace in synthetic:
base_id = trace.context.get("base", "unknown")
print(f" • {trace.content}")
print(f" (from: {base_id})")
# Example 7: Replay learning agent
print("\n" + "=" * 80)
print("Example 7: Agent with Periodic Replay")
print("=" * 80)
agent = ReplayLearningAgent(
buffer_size=50,
replay_interval=5
)
print("\nAgent learning through experience...")
experiences = [
("Learned concept A", {"subject": "learning"}, 0.7, 0.8),
("Applied concept A successfully", {"subject": "application"}, 0.8, 0.9),
("Struggled with concept B", {"subject": "challenge"}, 0.9, 0.4),
("Reviewed fundamentals", {"subject": "review"}, 0.6, 0.7),
("Mastered concept B", {"subject": "mastery"}, 0.9, 0.9),
("Completed practice exercises", {"subject": "practice"}, 0.7, 0.8),
("Taught concept to peer", {"subject": "teaching"}, 0.8, 0.9),
("Received positive feedback", {"subject": "feedback"}, 0.7, 0.8),
("Started new project", {"subject": "project"}, 0.8, 0.7),
("Made significant progress", {"subject": "progress"}, 0.8, 0.8),
]
for i, (content, context, importance, success) in enumerate(experiences, 1):
agent.experience(content, context, importance, success)
print(f" {i}. Experienced: {content}")
if i % agent.replay_interval == 0:
print(f" → Automatic replay triggered at interaction {i}")
# Check strongest memories
print("\nStrongest consolidated memories:")
strong = agent.get_strong_memories(5)
for trace in strong:
print(f" [{trace.strength:.2f}] {trace.content} "
f"(replayed {trace.replay_count}x)")
# Example 8: Replay statistics
print("\n" + "=" * 80)
print("Example 8: Replay System Statistics")
print("=" * 80)
stats_system = MemoryReplaySystem(buffer_size=100)
# Simulate learning session
print("\nSimulating learning session...")
for i in range(50):
stats_system.store_memory(
f"Experience {i}",
{"index": i},
importance=random.uniform(0.3, 0.9),
success_metric=random.uniform(0.4, 1.0)
)
# Multiple replay sessions
for _ in range(5):
stats_system.replay()
stats = stats_system.get_statistics()
print(f"\nReplay Statistics:")
print(f" Total Memories: {stats['total_memories']}")
print(f" Total Replay Sessions: {stats['total_replays']}")
print(f" Average Memory Strength: {stats['avg_strength']:.2f}")
print(f" Average Replay Count: {stats['avg_replay_count']:.2f}")
print(f" Buffer Usage: {stats['buffer_usage']:.1%}")
print(f" Strategy: {stats['strategy']}")
# Summary
print("\n" + "=" * 80)
print("SUMMARY: Memory Replay & Rehearsal Pattern")
print("=" * 80)
summary = """
The Memory Replay & Rehearsal pattern demonstrated:
1. BASIC REPLAY (Example 1):
- Memory storage in replay buffer
- Selection and replay of memories
- Memory strengthening through consolidation
- Replay statistics tracking
2. REPLAY STRATEGIES (Example 2):
- Prioritized: High-importance first
- Random: Random sampling
- Recent: Newest memories
- Oldest: Oldest memories
- Strategy comparison and selection
3. STRENGTHENING (Example 3):
- Progressive strength increase
- Multiple replay sessions
- Replay count tracking
- Consolidation effects
4. PRIORITY CALCULATION (Example 4):
- Importance weighting
- Recency consideration
- Success metric integration
- Replay count balancing
- Automatic priority ranking
5. INTERLEAVED REPLAY (Example 5):
- Mixing old and new memories
- Balanced temporal distribution
- Prevents recency bias
- Catastrophic forgetting prevention
6. GENERATIVE REPLAY (Example 6):
- Synthetic memory generation
- LLM-based variations
- Knowledge expansion
- Creative recombination
7. AGENT INTEGRATION (Example 7):
- Periodic automatic replay
- Experience-based learning
- Strongest memory identification
- Continuous consolidation
8. STATISTICS TRACKING (Example 8):
- Buffer usage monitoring
- Replay session counting
- Strength progression
- System performance metrics
KEY BENEFITS:
✓ Strengthens important memories
✓ Improves learning efficiency
✓ Enables offline consolidation
✓ Prevents catastrophic forgetting
✓ Discovers new patterns
✓ Simulates sleep consolidation
✓ Supports continuous learning
USE CASES:
• Reinforcement learning agents
• Skill learning and retention
• Memory consolidation systems
• Experience-based learning
• Pattern recognition improvement
• Knowledge transfer and generalization
• Continuous learning systems
BEST PRACTICES:
1. Schedule replay during low-activity periods
2. Use priority-based selection for efficiency
3. Balance old and new experiences
4. Monitor strengthening effectiveness
5. Limit replay buffer size appropriately
6. Track replay metrics for tuning
7. Consider computational budget
8. Implement gradual strengthening
TRADE-OFFS:
• Computational cost vs. learning benefit
• Replay frequency vs. new experiences
• Buffer size vs. memory usage
• Exact vs. generative replay
PRODUCTION CONSIDERATIONS:
→ Schedule replay during off-peak hours
→ Implement priority-based sampling for efficiency
→ Monitor replay effectiveness metrics
→ Set appropriate buffer size limits
→ Balance replay with new experience learning
→ Track memory strength progression
→ Implement incremental consolidation
→ Consider distributed replay for scale
→ Use batch processing for efficiency
→ Monitor computational costs
This pattern enables efficient memory consolidation and learning
through strategic replay and rehearsal, mimicking human memory
consolidation during sleep and practice.
"""
print(summary)
if __name__ == "__main__":
demonstrate_memory_replay()
Agentic Design Pattern: Memory Replay
Pattern 104: Memory Replay – Line-by-Line Explanation
Overview
This document provides a detailed line-by-line explanation of the Memory Replay pattern implementation.
Pattern Description
Pattern 104: Memory Replay & Rehearsal
Description: The Memory Replay & Rehearsal pattern implements mechanisms for re-experiencing and strengthening memories through replay, similar to how the human brain consolidates memories during sleep and practice. This pattern is crucial for learning, skill retention, and memory consolidation in long-running agent systems.
Memory replay involves re-activating stored experiences to strengthen neural
patterns, transfer information from short-term to long-term storage, and discover
new insights through recombination. Rehearsal can be exact (episodic replay) or
generative (synthetic variations), and can occur during downtime, triggered by
context, or scheduled periodically.
This pattern supports various replay strategies including experience replay
(reinforcement learning), prioritized replay (focus on important memories),
interleaved replay (mixing old and new), and generative replay (creating
variations). It implements replay scheduling, priority calculation, and
consolidation effects.
Key Components: 1. Memory Buffer: Store of experiences to replay 2. Replay Strategy: Algorithm for selecting memories 3. Replay Scheduler: When to perform replay 4. Priority Calculator: Importance scoring 5. Consolidation Engine: Strengthening mechanism 6. Synthetic Generator: Create variations 7. Replay Metrics: Track effectiveness
Replay Types: 1. Experience Replay: Re-live past experiences 2. Prioritized Replay: Focus on important/surprising 3. Interleaved Replay: Mix old and new memories 4. Generative Replay: Create synthetic variations 5. Forward Replay: Chronological order 6. Reverse Replay: Reverse chronological 7. Random Replay: Random sampling
Replay Timing: 1. Offline Replay: During downtime/sleep 2. Online Replay: During active tasks 3. Triggered Replay: By context or cues 4. Periodic Replay: Scheduled intervals 5. Event-driven Replay: After significant events
Use Cases: – Reinforcement learning agents – Skill learning and retention – Memory consolidation systems – Experience-based learning – Pattern recognition improvement – Knowledge transfer and generalization – Continuous learning systems
Advantages: – Strengthens important memories – Improves learning efficiency – Enables offline learning – Prevents catastrophic forgetting – Discovers new patterns – Transfers knowledge – Simulates “sleep consolidation”
Challenges: – Computational cost of replay – Selecting which memories to replay – Balancing old vs. new experiences – Avoiding overfitting to replayed memories – Determining replay frequency – Managing replay buffer size
LangChain Implementation: This implementation uses LangChain for: – LLM-based memory importance assessment – Synthetic memory generation – Pattern extraction from replay – Learning from replayed experiences
Production Considerations: – Schedule replay during low-activity periods – Implement priority-based sampling – Monitor replay effectiveness metrics – Limit replay buffer size – Balance replay with new experiences – Track memory strengthening – Implement incremental consolidation – Consider computational budget
File Statistics
- Total Lines: 903
- Classes: 5
- Functions: 1
- Constants: 0
Module Documentation
"""
Pattern 104: Memory Replay & Rehearsal
Description:
The Memory Replay & Rehearsal pattern implements mechanisms for re-experiencing
and strengthening memories through replay, similar to how the human brain
consolidates memories during sleep and practice. This pattern is crucial for
learning, skill retention, and memory consolidation in long-running agent systems.
Memory replay involves re-activating stored experiences to strengthen neural
patterns, transfer information from short-term to long-term storage, and discover
new insights through recombination. Rehearsal can be exact (episodic replay) or
generative (synthetic variations), and can occur during downtime, triggered by
context, or scheduled periodically.
This pattern supports various replay strategies including experience replay
(reinforcement learning), prioritized replay (focus on important memories),
interleaved replay (mixing old and new), and generative replay (creating
variations). It implements replay scheduling, priority calculation, and
consolidation effects.
Key Components:
1. Memory Buffer: Store of experiences to replay
2. Replay Strategy: Algorithm for selecting memories
3. Replay Scheduler: When to perform replay
4. Priority Calculator: Importance scoring
5. Consolidation Engine: Strengthening mechanism
6. Synthetic Generator: Create variations
7. Replay Metrics: Track effectiveness
Replay Types:
1. Experience Replay: Re-live past experiences
2. Prioritized Replay: Focus on important/surprising
3. Interleaved Replay: Mix old and new memories
4. Generative Replay: Create synthetic variations
5. Forward Replay: Chronological order
6. Reverse Replay: Reverse chronological
7. Random Replay: Random sampling
Replay Timing:
1. Offline Replay: During downtime/sleep
2. Online Replay: During active tasks
3. Triggered Replay: By context or cues
4. Periodic Replay: Scheduled intervals
5. Event-driven Replay: After significant events
Use Cases:
- Reinforcement learning agents
- Skill learning and retention
- Memory consolidation systems
- Experience-based learning
- Pattern recognition improvement
- Knowledge transfer and generalization
- Continuous learning systems
Advantages:
- Strengthens important memories
- Improves learning efficiency
- Enables offline learning
- Prevents catastrophic forgetting
- Discovers new patterns
- Transfers knowledge
- Simulates "sleep consolidation"
Challenges:
- Computational cost of replay
- Selecting which memories to replay
- Balancing old vs. new experiences
- Avoiding overfitting to replayed memories
- Determining replay frequency
- Managing replay buffer size
LangChain Implementation:
This implementation uses LangChain for:
- LLM-based memory importance assessment
- Synthetic memory generation
- Pattern extraction from replay
- Learning from replayed experiences
Production Considerations:
- Schedule replay during low-activity periods
- Implement priority-based sampling
- Monitor replay effectiveness metrics
- Limit replay buffer size
- Balance replay with new experiences
- Track memory strengthening
- Implement incremental consolidation
- Consider computational budget
"""
Purpose: This module implements the Memory Replay pattern, providing a structured approach to agent design and implementation.
Import Statements
Standard Library Imports
import collections.deque
import dataclasses.dataclass
import dataclasses.field
import datetime.datetime
import datetime.timedelta
import enum.Enum
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import random
import time
import typing.Any
import typing.Callable
import typing.Dict
import typing.List
import typing.Optional
Purpose: Core Python functionality:
– collections.deque: Specialized container datatypes
– dataclasses.dataclass: Simplified class definitions with @dataclass
– dataclasses.field: Simplified class definitions with @dataclass
– datetime.datetime: Date and time handling
– datetime.timedelta: Date and time handling
– enum.Enum: Enumeration support
– langchain_core.output_parsers.StrOutputParser: Regular expression operations
– langchain_core.prompts.ChatPromptTemplate: Regular expression operations
– os: Environment variable access and system operations
– random: Random number generation
– time: Time-related functions
– typing.Any: Type hints for better code documentation (List, Dict, Any, Optional)
– typing.Callable: Type hints for better code documentation (List, Dict, Any, Optional)
– typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional)
– typing.List: Type hints for better code documentation (List, Dict, Any, Optional)
– typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional)
LangChain Imports
from langchain_openai.ChatOpenAI
Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows
Other Imports
from dotenv.load_dotenv
Purpose: Additional dependencies for specialized functionality.
Classes
Class: ReplayStrategy
Lines 108-115
Inherits from: Enum
Purpose:
Strategy for replaying memories.
Class: ReplayTrigger
Lines 118-124
Inherits from: Enum
Purpose:
When replay is triggered.
Class: MemoryTrace
Lines 128-153
Purpose:
Memory trace for replay.
Attributes:
trace_id: Unique identifier
content: Memory content
context: Context when created
timestamp: Creation time
importance: Importance score
replay_count: Times replayed
strength: Memory strength
last_replayed: Last replay time
success_metric: Performance metric
metadata: Additional information
Class: MemoryReplaySystem
Lines 156-453
Purpose:
Memory replay and rehearsal system.
This system manages the replay of stored memories for consolidation,
learning, and pattern discovery.
Methods:
__init__(self, buffer_size, replay_batch_size, strategy, strengthening_rate, temperature)
Lines 164-190
Purpose:
Initialize memory replay system.
Args:
buffer_size: Maximum memories to store
replay_batch_size: Memories per replay session
strategy: Replay selection strategy
strengthening_rate: How much to strengthen on replay
temperature: LLM temperature
Parameters:
– buffer_size: Buffer size
– replay_batch_size: Replay batch size
– strategy: Strategy
– strengthening_rate: Strengthening rate
– temperature: Sampling temperature (0-1)
store_memory(self, content, context, importance, success_metric)
Lines 192-222
Purpose:
Store memory for potential replay.
Args:
content: Memory content
context: Context information
importance: Importance score
success_metric: Success/performance metric
Returns:
Created memory trace
Parameters:
– content: Content
– context: Context
– importance: Importance
– success_metric: Success metric
Returns: MemoryTrace
calculate_priority(self, trace)
Lines 224-252
Purpose:
Calculate replay priority for memory.
Args:
trace: Memory trace
Returns:
Priority score
Parameters:
– trace: Trace
Returns: float
select_memories_for_replay(self, n)
Lines 254-307
Purpose:
Select memories to replay based on strategy.
Args:
n: Number of memories (default: replay_batch_size)
Returns:
List of memory traces to replay
Parameters:
– n: N
Returns: <ast.Subscript object at 0x000001A06D71A410>
replay(self, n, consolidate)
Lines 309-359
Purpose:
Perform memory replay.
Args:
n: Number of memories to replay
consolidate: Whether to strengthen memories
Returns:
Replay statistics
Parameters:
– n: N
– consolidate: Consolidate
Returns: <ast.Subscript object at 0x000001A06D701710>
generate_synthetic_memory(self, base_memory, variation)
Lines 361-399
Purpose:
Generate synthetic memory variation.
Args:
base_memory: Memory to vary
variation: Type of variation
Returns:
New synthetic memory
Parameters:
– base_memory: Memory store for conversation history
– variation: Variation
Returns: MemoryTrace
generative_replay(self, n_synthetic, min_importance)
Lines 401-432
Purpose:
Perform generative replay with synthetic variations.
Args:
n_synthetic: Number of synthetic memories to generate
min_importance: Minimum importance for base memories
Returns:
List of generated synthetic memories
Parameters:
– n_synthetic: N synthetic
– min_importance: Min importance
Returns: <ast.Subscript object at 0x000001A06D6624D0>
get_statistics(self)
Lines 434-453
Purpose:
Get replay statistics.
Parameters:
Returns: <ast.Subscript object at 0x000001A06D83CF50>
Class: ReplayLearningAgent
Lines 456-525
Purpose:
Agent that learns through memory replay.
This agent uses replay to consolidate memories and improve
performance through rehearsal.
Methods:
__init__(self, buffer_size, replay_interval, temperature)
Lines 464-484
Purpose:
Initialize replay learning agent.
Args:
buffer_size: Memory buffer size
replay_interval: Interactions between replays
temperature: LLM temperature
Parameters:
– buffer_size: Buffer size
– replay_interval: Replay interval
– temperature: Sampling temperature (0-1)
experience(self, content, context, importance, success)
Lines 486-507
Purpose:
Record an experience.
Args:
content: Experience content
context: Context information
importance: Importance score
success: Success metric
Parameters:
– content: Content
– context: Context
– importance: Importance
– success: Success
consolidate(self)
Lines 509-511
Purpose:
Perform memory consolidation through replay.
Parameters:
get_strong_memories(self, n)
Lines 513-525
Purpose:
Get strongest memories.
Args:
n: Number of memories
Returns:
List of strong memories
Parameters:
– n: N
Returns: <ast.Subscript object at 0x000001A06D88A710>
Functions
demonstrate_memory_replay()
Lines 528-898
Purpose:
Demonstrate memory replay and rehearsal pattern.
Key Concepts Summary
Pattern Overview
The Memory Replay pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.
Core Components
- Agent Class: Main implementation of the pattern logic
- Prompt Engineering: Carefully crafted prompts to guide LLM behavior
- Tool Integration: Optional tools for extended functionality
- Memory Management: State handling across interactions
- Output Parsing: Structured response processing
When to Use
✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability
❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)
Best Practices Demonstrated
1. Environment Configuration
- ✅ Use
.envfiles for API keys - ✅ Never hardcode credentials
- ✅ Validate environment variables on startup
2. Error Handling
- ✅ Graceful degradation on failures
- ✅ Informative error messages
- ✅ Logging for debugging
3. Prompt Engineering
- ✅ Clear, specific instructions
- ✅ Structured output formats
- ✅ Few-shot examples when beneficial
4. Code Organization
- ✅ Separation of concerns (class-based design)
- ✅ Type hints for clarity
- ✅ Docstrings for documentation
5. Performance Optimization
- ✅ Appropriate temperature settings
- ✅ Token budget management
- ✅ Caching where applicable
LangChain Components Used
- ChatOpenAI: OpenAI model interface
- Chains: Sequential component composition
Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development
Execution Flow
1. Initialize agent with configuration
↓
2. Load environment and dependencies
↓
3. Prepare prompt with input data
↓
4. Invoke LLM through LangChain
↓
5. Parse and validate response
↓
6. Return formatted result
Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation
Performance Considerations
Computational Cost
- LLM Calls: Primary cost factor (based on tokens)
- Latency: Depends on model and input size
- Memory: Minimal for most patterns
Optimization Strategies
- Prompt Optimization: Minimize tokens while maintaining clarity
- Caching: Cache repeated queries/responses
- Model Selection: Use smaller models when appropriate
- Batch Processing: Group similar requests
- Streaming: For long responses, stream tokens
Cost Estimates (Approximate)
- GPT-3.5-turbo: ~$0.002 per request
- GPT-4: ~$0.03 per request
- GPT-4-turbo: ~$0.01 per request
Actual costs vary based on input/output token counts
Common Pitfalls to Avoid
1. Prompt Design Issues
❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts
2. Error Handling
❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation
3. Token Management
❌ Exceeding model token limits
✅ Truncate inputs, chunk processing
4. Security
❌ Exposing API keys in code
✅ Use environment variables
5. Testing
❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs
6. Cost Management
❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring
Further Improvements
Production Enhancements
- Monitoring: Add logging, metrics, tracing
- Validation: Input/output schema validation
- Rate Limiting: Implement request throttling
- Caching: Redis or in-memory cache for responses
- Async Support: Non-blocking operations for scalability
Feature Extensions
- Multi-Model Support: Switch between providers
- Fallback Models: Graceful degradation strategy
- Custom Tools: Domain-specific function calling
- Advanced Memory: Vector stores for long-term memory
- Evaluation: Automated testing and benchmarking
Code Quality
- Testing: Unit tests, integration tests
- Documentation: API docs, usage examples
- Type Safety: Stricter type hints
- Configuration: Externalized config files
Related Patterns
- Pattern 103
- Pattern 105
Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…
Summary
The Memory Replay pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:
✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns
🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability
📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/
File: 104_memory_replay.py
Total Lines: 903
Pattern Category: AI Agent Design Pattern #104
Generated from actual implementation code
For educational and reference purposes
Leave a Reply