aishastr.live

Agentic Architecture Pattern: Long Term Memory

Written by

admin

in


    """
Pattern 64: Long-Term Memory Pattern

Agents maintain persistent memory across sessions with consolidation and retrieval.
Combines episodic, semantic, and procedural memory for long-term storage.
"""

from typing import Dict, Any, List, Optional, Set
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
import json
from pathlib import Path
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI


class MemoryType(Enum):
    """Type of long-term memory."""
    EPISODIC = "episodic"  # Personal experiences
    SEMANTIC = "semantic"  # Facts and concepts
    PROCEDURAL = "procedural"  # Skills and procedures
    AUTOBIOGRAPHICAL = "autobiographical"  # Personal history


class MemoryStrength(Enum):
    """Memory consolidation strength."""
    WEAK = 0.3
    MODERATE = 0.6
    STRONG = 0.9
    PERMANENT = 1.0


@dataclass
class LongTermMemoryEntry:
    """Entry in long-term memory."""
    memory_id: str
    memory_type: MemoryType
    content: Any
    context: Dict[str, Any]
    strength: float = 0.5  # Consolidation strength
    created_at: datetime = field(default_factory=datetime.now)
    last_accessed: datetime = field(default_factory=datetime.now)
    access_count: int = 0
    tags: Set[str] = field(default_factory=set)
    associations: Set[str] = field(default_factory=set)  # IDs of related memories


@dataclass
class MemoryConsolidation:
    """Record of memory consolidation process."""
    consolidation_id: str
    memories_consolidated: List[str]
    original_strength: float
    new_strength: float
    timestamp: datetime = field(default_factory=datetime.now)
    reason: str = ""


class LongTermMemoryStore:
    """Persistent long-term memory storage."""
    
    def __init__(self, storage_path: Optional[Path] = None):
        self.storage_path = storage_path
        self.memories: Dict[str, LongTermMemoryEntry] = {}
        self.consolidation_history: List[MemoryConsolidation] = []
        self.indexes: Dict[str, Dict[Any, Set[str]]] = {
            "type": {},
            "tags": {},
            "date": {}
        }
    
    def store(self, memory: LongTermMemoryEntry):
        """Store memory in long-term storage."""
        self.memories[memory.memory_id] = memory
        
        # Update indexes
        mem_type = memory.memory_type.value
        if mem_type not in self.indexes["type"]:
            self.indexes["type"][mem_type] = set()
        self.indexes["type"][mem_type].add(memory.memory_id)
        
        # Tag index
        for tag in memory.tags:
            if tag not in self.indexes["tags"]:
                self.indexes["tags"][tag] = set()
            self.indexes["tags"][tag].add(memory.memory_id)
        
        # Date index (by day)
        date_key = memory.created_at.date()
        if date_key not in self.indexes["date"]:
            self.indexes["date"][date_key] = set()
        self.indexes["date"][date_key].add(memory.memory_id)
    
    def retrieve(self, memory_id: str) -> Optional[LongTermMemoryEntry]:
        """Retrieve memory and update access metadata."""
        if memory_id not in self.memories:
            return None
        
        memory = self.memories[memory_id]
        memory.last_accessed = datetime.now()
        memory.access_count += 1
        
        # Strengthen memory through retrieval
        self._strengthen_memory(memory_id, 0.05)
        
        return memory
    
    def search_by_type(self, memory_type: MemoryType) -> List[LongTermMemoryEntry]:
        """Search memories by type."""
        memory_ids = self.indexes["type"].get(memory_type.value, set())
        return [self.memories[mid] for mid in memory_ids if mid in self.memories]
    
    def search_by_tags(self, tags: Set[str]) -> List[LongTermMemoryEntry]:
        """Search memories by tags."""
        result_ids = None
        for tag in tags:
            tag_ids = self.indexes["tags"].get(tag, set())
            if result_ids is None:
                result_ids = tag_ids
            else:
                result_ids = result_ids & tag_ids
        
        if result_ids is None:
            return []
        
        return [self.memories[mid] for mid in result_ids if mid in self.memories]
    
    def search_by_date_range(self, start: datetime, end: datetime) -> List[LongTermMemoryEntry]:
        """Search memories by date range."""
        results = []
        current_date = start.date()
        end_date = end.date()
        
        while current_date  7:  # Only decay after a week
                decay = decay_rate * (days_since_access / 7)
                memory.strength = max(0.1, memory.strength - decay)


class LongTermMemoryAgent:
    """Agent with long-term persistent memory."""
    
    def __init__(self, name: str, storage_path: Optional[Path] = None, model_name: str = "gpt-4"):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        self.memory_store = LongTermMemoryStore(storage_path)
    
    def remember(self, memory_id: str, content: Any, memory_type: MemoryType,
                 context: Optional[Dict[str, Any]] = None, tags: Optional[Set[str]] = None):
        """Store in long-term memory."""
        memory = LongTermMemoryEntry(
            memory_id=memory_id,
            memory_type=memory_type,
            content=content,
            context=context or {},
            tags=tags or set(),
            strength=0.5
        )
        
        self.memory_store.store(memory)
        print(f"   šŸ’¾ {self.name} stored long-term memory: {memory_id} ({memory_type.value})")
    
    def recall(self, memory_id: str) -> Optional[Any]:
        """Recall from long-term memory."""
        memory = self.memory_store.retrieve(memory_id)
        if memory:
            print(f"   āœ“ {self.name} recalled: {memory_id} (strength: {memory.strength:.2f}, accessed {memory.access_count}x)")
            return memory.content
        else:
            print(f"   āœ— {self.name} couldn't recall: {memory_id}")
            return None
    
    def reflect_on_memories(self, memory_type: MemoryType, limit: int = 5) -> str:
        """Reflect on memories using LLM."""
        memories = self.memory_store.search_by_type(memory_type)[:limit]
        
        if not memories:
            return f"No {memory_type.value} memories to reflect on."
        
        memories_text = "\n".join([
            f"- [{m.memory_id}] {m.content} (strength: {m.strength:.2f}, accessed: {m.access_count}x)"
            for m in sorted(memories, key=lambda x: x.strength, reverse=True)
        ])
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name} reflecting on your long-term memories."),
            ("human", f"Your {memory_type.value} memories:\n{memories_text}\n\n"
                     f"Reflect on these memories. What patterns or insights do you notice? "
                     f"(2-3 sentences)")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        return str(response.content)[:300]


class LongTermMemoryPattern:
    """
    Long-Term Memory Pattern - Persistent cross-session memory.
    
    Agents maintain memories across sessions with consolidation,
    strengthening through retrieval, and decay of unused memories.
    """
    
    def __init__(self):
        self.agents: Dict[str, LongTermMemoryAgent] = {}
    
    def register_agent(self, agent: LongTermMemoryAgent):
        """Register agent with long-term memory."""
        self.agents[agent.name] = agent
    
    def run(self, scenario: str) -> Dict[str, Any]:
        """Execute long-term memory pattern."""
        total_memories = sum(len(agent.memory_store.memories) for agent in self.agents.values())
        total_consolidations = sum(len(agent.memory_store.consolidation_history) for agent in self.agents.values())
        
        avg_strength = 0
        if total_memories > 0:
            all_strengths = [m.strength for agent in self.agents.values() 
                           for m in agent.memory_store.memories.values()]
            avg_strength = sum(all_strengths) / len(all_strengths)
        
        return {
            "pattern": "LongTermMemory",
            "scenario": scenario,
            "total_agents": len(self.agents),
            "total_memories": total_memories,
            "average_strength": avg_strength,
            "total_consolidations": total_consolidations
        }


# Example usage
if __name__ == "__main__":
    print("=== Long-Term Memory Pattern Demo ===\n")
    
    # Create pattern
    pattern = LongTermMemoryPattern()
    
    # Create agent with long-term memory
    print("1. Creating agent with long-term memory...")
    agent = LongTermMemoryAgent("Sage")
    pattern.register_agent(agent)
    
    # Store various types of memories
    print("\n2. Storing long-term memories...")
    
    # Episodic memories
    agent.remember("first_project", "Led AI chatbot project in 2023", MemoryType.EPISODIC,
                   context={"year": 2023, "role": "lead"}, tags={"work", "project", "ai"})
    
    agent.remember("conference_talk", "Presented at AI Conference 2024", MemoryType.EPISODIC,
                   context={"year": 2024, "type": "presentation"}, tags={"conference", "public_speaking"})
    
    # Semantic memories
    agent.remember("python_knowledge", "Python is a high-level programming language", MemoryType.SEMANTIC,
                   tags={"programming", "knowledge"})
    
    agent.remember("ml_concept", "Machine learning enables computers to learn from data", MemoryType.SEMANTIC,
                   tags={"ai", "knowledge", "ml"})
    
    # Procedural memories
    agent.remember("debug_process", "Check logs, isolate issue, fix, test", MemoryType.PROCEDURAL,
                   tags={"programming", "skill"})
    
    # Recall memories
    print("\n3. Recalling memories...")
    agent.recall("first_project")
    agent.recall("python_knowledge")
    agent.recall("debug_process")
    agent.recall("first_project")  # Second recall strengthens
    
    # Search by type
    print("\n4. Searching memories by type...")
    episodic = agent.memory_store.search_by_type(MemoryType.EPISODIC)
    print(f"   Episodic memories: {len(episodic)}")
    for mem in episodic:
        print(f"   - {mem.memory_id}: {mem.content} (strength: {mem.strength:.2f})")
    
    # Search by tags
    print("\n5. Searching memories by tags...")
    ai_memories = agent.memory_store.search_by_tags({"ai"})
    print(f"   Memories tagged 'ai': {len(ai_memories)}")
    for mem in ai_memories:
        print(f"   - {mem.memory_id}: {mem.content}")
    
    # Consolidate important memories
    print("\n6. Consolidating memories...")
    important_ids = ["first_project", "ml_concept"]
    agent.memory_store.consolidate_memories(important_ids, reason="frequently accessed")
    print(f"   Consolidated {len(important_ids)} memories")
    
    # Check strength after consolidation
    for mid in important_ids:
        mem = agent.memory_store.memories.get(mid)
        if mem:
            print(f"   - {mid}: strength now {mem.strength:.2f}")
    
    # Reflection
    print("\n7. Reflecting on memories...")
    reflection = agent.reflect_on_memories(MemoryType.EPISODIC)
    print(f"   Reflection on episodic memories:")
    print(f"   {reflection}")
    
    # Statistics
    print("\n8. Memory Statistics:")
    summary = pattern.run("Personal knowledge and experience")
    print(f"   Total Memories: {summary['total_memories']}")
    print(f"   Average Strength: {summary['average_strength']:.2f}")
    print(f"   Consolidations: {summary['total_consolidations']}")
    
    # Memory distribution
    print("\n9. Memory Distribution:")
    for mem_type in MemoryType:
        count = len(agent.memory_store.search_by_type(mem_type))
        print(f"   {mem_type.value}: {count}")
    
    print("\nāœ“ Long-term memory pattern enabled persistent knowledge storage!")

Agentic Architecture Pattern: Long Term Memory

Pattern 64: 64 Long Term Memory – Line-by-Line Explanation

Overview

Pattern 64: Long-Term Memory Pattern

Agents maintain persistent memory across sessions with consolidation and retrieval. Combines episodic, semantic, and procedural memory for long-term storage.

Code Structure Analysis

File Information

  • Pattern Number: 64
  • Pattern Name: 64 Long Term Memory
  • File Path: patterns\memory_knowledge\64_long_term_memory.py
  • Total Lines: 366

Detailed Code Explanation

Module Docstring

"""
Pattern 64: Long-Term Memory Pattern

Agents maintain persistent memory across sessions with consolidation and retrieval.
Combines episodic, semantic, and procedural memory for long-term storage.
"""

Explanation: This module docstring describes the purpose and key features of the 64 Long Term Memory pattern.

Imports Section

from typing import Dict, Any, List, Optional, Set
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
import json
from pathlib import Path
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional, Set: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime, timedelta: Date and time utilities – from enum import Enum: Enumeration support for fixed sets of values – import json: JSON parsing and serialization – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Enumerations

MemoryType (Line 18)

class MemoryType(Enum):
    EPISODIC
    SEMANTIC
    PROCEDURAL
    AUTOBIOGRAPHICAL

Purpose: Type of long-term memory.

Values: – EPISODIC: Represents episodic – SEMANTIC: Represents semantic – PROCEDURAL: Represents procedural – AUTOBIOGRAPHICAL: Represents autobiographical

MemoryStrength (Line 26)

class MemoryStrength(Enum):
    WEAK
    MODERATE
    STRONG
    PERMANENT

Purpose: Memory consolidation strength.

Values: – WEAK: Represents weak – MODERATE: Represents moderate – STRONG: Represents strong – PERMANENT: Represents permanent

Classes

MemoryType (Line 18)

Description: Type of long-term memory.

MemoryStrength (Line 26)

Description: Memory consolidation strength.

LongTermMemoryEntry (Line 35)

Description: Entry in long-term memory.

MemoryConsolidation (Line 50)

Description: Record of memory consolidation process.

LongTermMemoryStore (Line 60)

Description: Persistent long-term memory storage.

Methods:

__init__ (Line 63)
def __init__(self, storage_path):
    """No method docstring."""

Purpose: No method docstring.

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

store (Line 73)
def store(self, memory):
    """Store memory in long-term storage."""

Purpose: Store memory in long-term storage.

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

retrieve (Line 95)
def retrieve(self, memory_id):
    """Retrieve memory and update access metadata."""

Purpose: Retrieve memory and update access metadata.

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

search_by_type (Line 109)
def search_by_type(self, memory_type):
    """Search memories by type."""

Purpose: Search memories by type.

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

search_by_tags (Line 114)
def search_by_tags(self, tags):
    """Search memories by tags."""

Purpose: Search memories by tags.

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

search_by_date_range (Line 129)
def search_by_date_range(self, start, end):
    """Search memories by date range."""

Purpose: Search memories by date range.

Parameters: – self: Instance reference – start: Parameter for start – end: Parameter for end

consolidate_memories (Line 144)
def consolidate_memories(self, memory_ids, reason):
    """Consolidate memories to strengthen retention."""

Purpose: Consolidate memories to strengthen retention.

Parameters: – self: Instance reference – memory_ids: Parameter for memory_ids – reason: Parameter for reason

_strengthen_memory (Line 168)
def _strengthen_memory(self, memory_id, amount):
    """Strengthen memory consolidation."""

Purpose: Strengthen memory consolidation.

Parameters: – self: Instance reference – memory_id: Parameter for memory_id – amount: Parameter for amount

decay_memories (Line 174)
def decay_memories(self, decay_rate):
    """Apply memory decay to unused memories."""

Purpose: Apply memory decay to unused memories.

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

LongTermMemoryAgent (Line 186)

Description: Agent with long-term persistent memory.

Methods:

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

Purpose: No method docstring.

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

remember (Line 194)
def remember(self, memory_id, content, memory_type, context, tags):
    """Store in long-term memory."""

Purpose: Store in long-term memory.

Parameters: – self: Instance reference – memory_id: Parameter for memory_id – content: Parameter for content – memory_type: Parameter for memory_type – context: Parameter for context – tags: Parameter for tags

recall (Line 209)
def recall(self, memory_id):
    """Recall from long-term memory."""

Purpose: Recall from long-term memory.

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

reflect_on_memories (Line 219)
def reflect_on_memories(self, memory_type, limit):
    """Reflect on memories using LLM."""

Purpose: Reflect on memories using LLM.

Parameters: – self: Instance reference – memory_type: Parameter for memory_type – limit: Parameter for limit

LongTermMemoryPattern (Line 244)

Description: Long-Term Memory Pattern – Persistent cross-session memory.

Agents maintain memories across sessions with consolidation, strengthening through retrieval, and decay of unused memories.

Methods:

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

Purpose: No method docstring.

Parameters: – self: Instance reference

register_agent (Line 255)
def register_agent(self, agent):
    """Register agent with long-term memory."""

Purpose: Register agent with long-term memory.

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

run (Line 259)
def run(self, scenario):
    """Execute long-term memory pattern."""

Purpose: Execute long-term 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 = MemoryType(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