aishastr.live

Agentic Architecture Pattern: Episodic Memory

Written by

admin

in


    """
Pattern 61: Episodic Memory Pattern

Agents store and recall specific experiences/episodes with temporal context.
Like human episodic memory: "I remember when X happened at time Y."
"""

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage


@dataclass
class Episode:
    """Single episode in memory."""
    episode_id: str
    description: str
    context: Dict[str, Any]
    participants: List[str]
    location: Optional[str] = None
    outcome: Optional[str] = None
    emotional_valence: float = 0.0  # -1 (negative) to +1 (positive)
    importance: float = 0.5  # 0 to 1
    timestamp: datetime = field(default_factory=datetime.now)
    tags: List[str] = field(default_factory=list)


@dataclass
class MemoryRecall:
    """Result of memory recall."""
    query: str
    recalled_episodes: List[Episode]
    relevance_scores: Dict[str, float]
    recall_time: datetime = field(default_factory=datetime.now)


class EpisodicMemoryAgent:
    """Agent with episodic memory capabilities."""
    
    def __init__(self, name: str, model_name: str = "gpt-4"):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.episodes: List[Episode] = []
        self.recall_history: List[MemoryRecall] = []
    
    def store_episode(self, episode: Episode):
        """Store new episode in memory."""
        self.episodes.append(episode)
        print(f"   šŸ“ {self.name} stored: {episode.description}")
    
    def recall_by_time(self, start_time: datetime, end_time: datetime) -> List[Episode]:
        """Recall episodes within time range."""
        return [ep for ep in self.episodes 
                if start_time <= ep.timestamp  List[Episode]:
        """Recall episodes with specific context."""
        return [ep for ep in self.episodes 
                if ep.context.get(context_key) == context_value]
    
    def recall_by_participant(self, participant: str) -> List[Episode]:
        """Recall episodes involving specific participant."""
        return [ep for ep in self.episodes 
                if participant in ep.participants]
    
    def recall_similar(self, query: str, top_k: int = 5) -> MemoryRecall:
        """Recall episodes similar to query using LLM."""
        if not self.episodes:
            return MemoryRecall(query, [], {})
        
        # Create episode descriptions
        episodes_text = "\n".join([
            f"{i+1}. [{ep.timestamp.strftime('%Y-%m-%d %H:%M')}] {ep.description} "
            f"(Participants: {', '.join(ep.participants)})"
            for i, ep in enumerate(self.episodes)
        ])
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name} recalling past experiences. "
                      f"Review your episodic memories and identify which ones are most relevant to the query."),
            ("human", f"Your episodic memories:\n{episodes_text}\n\n"
                     f"Query: {query}\n\n"
                     f"List the episode numbers (1-{len(self.episodes)}) most relevant to this query, "
                     f"in order of relevance. Format: just the numbers separated by commas.")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        # Parse episode numbers
        content = str(response.content)
        try:
            import re
            numbers = re.findall(r'\d+', content)
            indices = [int(n) - 1 for n in numbers if 0 < int(n) <= len(self.episodes)][:top_k]
            
            recalled = [self.episodes[i] for i in indices if i  str:
        """Reflect on stored episode using LLM."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name} reflecting on a past experience."),
            ("human", f"Experience: {episode.description}\n"
                     f"When: {episode.timestamp.strftime('%Y-%m-%d %H:%M')}\n"
                     f"Context: {episode.context}\n"
                     f"Outcome: {episode.outcome}\n\n"
                     f"What did you learn from this experience? (2-3 sentences)")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        return str(response.content)[:200]


class EpisodicMemoryPattern:
    """
    Episodic Memory Pattern - Experience-based memory system.
    
    Agents store specific experiences with temporal and contextual information,
    enabling recall of "what happened when" and learning from past experiences.
    """
    
    def __init__(self):
        self.agents: Dict[str, EpisodicMemoryAgent] = {}
    
    def register_agent(self, agent: EpisodicMemoryAgent):
        """Register agent with episodic memory."""
        self.agents[agent.name] = agent
    
    def create_shared_episode(self, episode: Episode):
        """Create episode shared by all agents."""
        for agent in self.agents.values():
            agent.store_episode(episode)
    
    def run(self, scenario: str) -> Dict[str, Any]:
        """Execute episodic memory pattern."""
        total_episodes = sum(len(agent.episodes) for agent in self.agents.values())
        total_recalls = sum(len(agent.recall_history) for agent in self.agents.values())
        
        return {
            "pattern": "EpisodicMemory",
            "total_agents": len(self.agents),
            "total_episodes": total_episodes,
            "total_recalls": total_recalls,
            "scenario": scenario
        }


# Example usage
if __name__ == "__main__":
    print("=== Episodic Memory Pattern Demo ===\n")
    
    # Create pattern
    pattern = EpisodicMemoryPattern()
    
    # Create agent with episodic memory
    print("1. Creating agent with episodic memory...")
    agent = EpisodicMemoryAgent("Alice")
    pattern.register_agent(agent)
    
    # Store episodes
    print("\n2. Storing episodic memories...")
    
    episodes = [
        Episode(
            "ep1",
            "First meeting with project team",
            context={"project": "AI Platform", "phase": "kickoff"},
            participants=["Alice", "Bob", "Carol"],
            location="Conference Room A",
            outcome="Agreed on project timeline",
            emotional_valence=0.8,
            importance=0.9,
            tags=["meeting", "project", "team"]
        ),
        Episode(
            "ep2",
            "Debugged critical production issue",
            context={"project": "AI Platform", "phase": "deployment"},
            participants=["Alice", "Dave"],
            location="Remote",
            outcome="Fixed database connection bug",
            emotional_valence=-0.3,
            importance=0.95,
            tags=["debugging", "production", "urgent"]
        ),
        Episode(
            "ep3",
            "Celebrated successful product launch",
            context={"project": "AI Platform", "phase": "launch"},
            participants=["Alice", "Bob", "Carol", "Dave", "Eve"],
            location="Office",
            outcome="Product launched successfully",
            emotional_valence=1.0,
            importance=1.0,
            tags=["celebration", "launch", "success"]
        ),
        Episode(
            "ep4",
            "Code review session with junior developer",
            context={"project": "AI Platform", "phase": "development"},
            participants=["Alice", "Frank"],
            location="Video call",
            outcome="Improved code quality and mentored junior dev",
            emotional_valence=0.6,
            importance=0.7,
            tags=["code_review", "mentoring", "learning"]
        ),
    ]
    
    for episode in episodes:
        agent.store_episode(episode)
    
    # Recall by different criteria
    print("\n3. Recalling memories by participant...")
    bob_episodes = agent.recall_by_participant("Bob")
    print(f"   Episodes with Bob: {len(bob_episodes)}")
    for ep in bob_episodes:
        print(f"   - {ep.description}")
    
    print("\n4. Recalling memories by context...")
    launch_episodes = agent.recall_by_context("phase", "launch")
    print(f"   Episodes in launch phase: {len(launch_episodes)}")
    for ep in launch_episodes:
        print(f"   - {ep.description}")
    
    print("\n5. Recalling similar memories (semantic search)...")
    queries = [
        "Tell me about times when we had problems",
        "What were our team celebrations?",
        "Experiences with code and development"
    ]
    
    for query in queries:
        print(f"\n   Query: {query}")
        recall = agent.recall_similar(query, top_k=2)
        for ep in recall.recalled_episodes:
            print(f"   - {ep.description} (relevance: {recall.relevance_scores.get(ep.episode_id, 0):.1f})")
    
    # Reflection
    print("\n6. Reflecting on experience...")
    important_ep = max(agent.episodes, key=lambda e: e.importance)
    reflection = agent.reflect_on_experience(important_ep)
    print(f"   Reflecting on: {important_ep.description}")
    print(f"   Reflection: {reflection}")
    
    # Summary
    print("\n7. Memory Statistics:")
    summary = pattern.run("Personal experience tracking")
    print(f"   Total Episodes: {summary['total_episodes']}")
    print(f"   Total Recalls: {summary['total_recalls']}")
    print(f"   Average importance: {sum(ep.importance for ep in agent.episodes) / len(agent.episodes):.2f}")
    print(f"   Average emotional valence: {sum(ep.emotional_valence for ep in agent.episodes) / len(agent.episodes):.2f}")
    
    print("\nāœ“ Episodic memory pattern enabled experience-based recall!")

Agentic Architecture Pattern: Episodic Memory

Pattern 61: 61 Episodic Memory – Line-by-Line Explanation

Overview

Pattern 61: Episodic Memory Pattern

Agents store and recall specific experiences/episodes with temporal context. Like human episodic memory: “I remember when X happened at time Y.”

Code Structure Analysis

File Information

  • Pattern Number: 61
  • Pattern Name: 61 Episodic Memory
  • File Path: patterns\memory_knowledge\61_episodic_memory.py
  • Total Lines: 268

Detailed Code Explanation

Module Docstring

"""
Pattern 61: Episodic Memory Pattern

Agents store and recall specific experiences/episodes with temporal context.
Like human episodic memory: "I remember when X happened at time Y."
"""

Explanation: This module docstring describes the purpose and key features of the 61 Episodic Memory pattern.

Imports Section

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
            import re  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime: Date and time utilities – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration – from langchain_core.messages import HumanMessage, AIMessage: LangChain framework integration

Classes

Episode (Line 17)

Description: Single episode in memory.

MemoryRecall (Line 32)

Description: Result of memory recall.

EpisodicMemoryAgent (Line 40)

Description: Agent with episodic memory capabilities.

Methods:

__init__ (Line 43)
def __init__(self, name, model_name):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – model_name: LLM model to use (e.g., “gpt-4”)

store_episode (Line 49)
def store_episode(self, episode):
    """Store new episode in memory."""

Purpose: Store new episode in memory.

Parameters: – self: Instance reference – episode: Parameter for episode

recall_by_time (Line 54)
def recall_by_time(self, start_time, end_time):
    """Recall episodes within time range."""

Purpose: Recall episodes within time range.

Parameters: – self: Instance reference – start_time: Parameter for start_time – end_time: Parameter for end_time

recall_by_context (Line 59)
def recall_by_context(self, context_key, context_value):
    """Recall episodes with specific context."""

Purpose: Recall episodes with specific context.

Parameters: – self: Instance reference – context_key: Parameter for context_key – context_value: Parameter for context_value

recall_by_participant (Line 64)
def recall_by_participant(self, participant):
    """Recall episodes involving specific participant."""

Purpose: Recall episodes involving specific participant.

Parameters: – self: Instance reference – participant: Parameter for participant

recall_similar (Line 69)
def recall_similar(self, query, top_k):
    """Recall episodes similar to query using LLM."""

Purpose: Recall episodes similar to query using LLM.

Parameters: – self: Instance reference – query: Parameter for query – top_k: Parameter for top_k

reflect_on_experience (Line 110)
def reflect_on_experience(self, episode):
    """Reflect on stored episode using LLM."""

Purpose: Reflect on stored episode using LLM.

Parameters: – self: Instance reference – episode: Parameter for episode

EpisodicMemoryPattern (Line 127)

Description: Episodic Memory Pattern – Experience-based memory system.

Agents store specific experiences with temporal and contextual information, enabling recall of “what happened when” and learning from past experiences.

Methods:

__init__ (Line 135)
def __init__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

register_agent (Line 138)
def register_agent(self, agent):
    """Register agent with episodic memory."""

Purpose: Register agent with episodic memory.

Parameters: – self: Instance reference – agent: Parameter for agent

create_shared_episode (Line 142)
def create_shared_episode(self, episode):
    """Create episode shared by all agents."""

Purpose: Create episode shared by all agents.

Parameters: – self: Instance reference – episode: Parameter for episode

run (Line 147)
def run(self, scenario):
    """Execute episodic memory pattern."""

Purpose: Execute episodic memory pattern.

Parameters: – self: Instance reference – scenario: Parameter for scenario

Design Patterns Used

  • Strategy Pattern: Different execution strategies for various scenarios
  • State Pattern: Managing different states in the workflow
  • Chain of Responsibility: Sequential processing through workflow nodes
  • Observer Pattern: Tracking and monitoring pattern execution
  • Factory Pattern: Creating instances with proper configuration

Usage Example

# Initialize the pattern
pattern = Episode(name="example", model_name="gpt-4")

# Configure and execute
result = pattern.execute(data={"key": "value"})

# Check results
print(result)

Key Concepts

  • LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
  • Workflow Orchestration: LangGraph manages execution flow through nodes
  • State Management: TypedDict defines workflow state structure
  • Error Handling: Comprehensive error handling and fallback mechanisms
  • Extensibility: Designed for easy customization and extension

Extension Points

This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *

More posts