"""
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