aishastr.live

Agentic Architecture Pattern: Working Memory

Written by

admin

in


    """
Pattern 63: Working Memory Pattern

Agents maintain temporary, limited-capacity memory for current task execution.
Similar to human working memory: holds active information needed right now.
"""

from typing import Dict, Any, List, Optional, Deque
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from collections import deque
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI


class ItemPriority(Enum):
    """Priority level for working memory items."""
    CRITICAL = 3
    HIGH = 2
    NORMAL = 1
    LOW = 0


@dataclass
class WorkingMemoryItem:
    """Item in working memory."""
    item_id: str
    content: Any
    priority: ItemPriority
    created_at: datetime = field(default_factory=datetime.now)
    last_accessed: datetime = field(default_factory=datetime.now)
    access_count: int = 0
    ttl: Optional[timedelta] = None  # Time to live
    context: Dict[str, Any] = field(default_factory=dict)


@dataclass
class MemoryOperation:
    """Record of memory operation."""
    operation_type: str  # "store", "retrieve", "update", "evict"
    item_id: str
    timestamp: datetime = field(default_factory=datetime.now)
    success: bool = True


class WorkingMemory:
    """
    Limited-capacity working memory with eviction policies.
    """
    
    def __init__(self, capacity: int = 7):  # Miller's law: 7Ā±2 items
        self.capacity = capacity
        self.items: Dict[str, WorkingMemoryItem] = {}
        self.access_queue: Deque[str] = deque()  # For LRU eviction
        self.operations: List[MemoryOperation] = []
    
    def store(self, item: WorkingMemoryItem) -> bool:
        """Store item in working memory."""
        # Check if we need to evict
        if len(self.items) >= self.capacity and item.item_id not in self.items:
            self._evict()
        
        # Store item
        self.items[item.item_id] = item
        
        # Update access queue
        if item.item_id in self.access_queue:
            self.access_queue.remove(item.item_id)
        self.access_queue.append(item.item_id)
        
        # Record operation
        self.operations.append(MemoryOperation("store", item.item_id))
        
        return True
    
    def retrieve(self, item_id: str) -> Optional[WorkingMemoryItem]:
        """Retrieve item from working memory."""
        if item_id not in self.items:
            self.operations.append(MemoryOperation("retrieve", item_id, success=False))
            return None
        
        item = self.items[item_id]
        
        # Check TTL
        if item.ttl and datetime.now() - item.created_at > item.ttl:
            self._remove(item_id)
            return None
        
        # Update access metadata
        item.last_accessed = datetime.now()
        item.access_count += 1
        
        # Update LRU queue
        if item_id in self.access_queue:
            self.access_queue.remove(item_id)
        self.access_queue.append(item_id)
        
        # Record operation
        self.operations.append(MemoryOperation("retrieve", item_id))
        
        return item
    
    def update(self, item_id: str, new_content: Any) -> bool:
        """Update item in working memory."""
        if item_id not in self.items:
            return False
        
        self.items[item_id].content = new_content
        self.items[item_id].last_accessed = datetime.now()
        
        self.operations.append(MemoryOperation("update", item_id))
        
        return True
    
    def _evict(self):
        """Evict item based on priority and recency."""
        if not self.items:
            return
        
        # Find lowest priority, least recently used item
        candidates = []
        for item_id in list(self.access_queue):
            if item_id in self.items:
                item = self.items[item_id]
                # Score: lower is better for eviction
                score = item.priority.value * 100 + item.access_count
                candidates.append((score, item_id))
        
        if candidates:
            candidates.sort()
            item_id_to_evict = candidates[0][1]
            self._remove(item_id_to_evict)
    
    def _remove(self, item_id: str):
        """Remove item from working memory."""
        if item_id in self.items:
            del self.items[item_id]
        if item_id in self.access_queue:
            self.access_queue.remove(item_id)
        
        self.operations.append(MemoryOperation("evict", item_id))
    
    def get_all(self) -> List[WorkingMemoryItem]:
        """Get all items in working memory."""
        return list(self.items.values())
    
    def clear(self):
        """Clear all working memory."""
        self.items.clear()
        self.access_queue.clear()
    
    def is_full(self) -> bool:
        """Check if working memory is at capacity."""
        return len(self.items) >= self.capacity


class WorkingMemoryAgent:
    """Agent with working memory for current task execution."""
    
    def __init__(self, name: str, memory_capacity: int = 7, model_name: str = "gpt-4"):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.working_memory = WorkingMemory(capacity=memory_capacity)
        self.task_context: Dict[str, Any] = {}
    
    def start_task(self, task_name: str, task_data: Dict[str, Any]):
        """Start new task, initializing working memory."""
        self.task_context = {"task": task_name, "data": task_data}
        
        # Store task info in working memory
        task_item = WorkingMemoryItem(
            "current_task",
            task_name,
            ItemPriority.CRITICAL,
            context=task_data
        )
        self.working_memory.store(task_item)
        
        print(f"   šŸŽÆ {self.name} started task: {task_name}")
    
    def remember(self, key: str, value: Any, priority: ItemPriority = ItemPriority.NORMAL,
                 ttl: Optional[timedelta] = None):
        """Store information in working memory."""
        item = WorkingMemoryItem(key, value, priority, ttl=ttl)
        self.working_memory.store(item)
        print(f"   šŸ’­ {self.name} remembered: {key} (priority: {priority.name})")
    
    def recall(self, key: str) -> Optional[Any]:
        """Recall information from working memory."""
        item = self.working_memory.retrieve(key)
        if item:
            print(f"   āœ“ {self.name} recalled: {key}")
            return item.content
        else:
            print(f"   āœ— {self.name} couldn't recall: {key}")
            return None
    
    def process_with_context(self, query: str) -> str:
        """Process query using working memory context."""
        # Get all working memory items
        items = self.working_memory.get_all()
        
        context_text = "\n".join([
            f"- {item.item_id}: {item.content} (priority: {item.priority.name}, accessed: {item.access_count})"
            for item in sorted(items, key=lambda x: x.priority.value, reverse=True)
        ])
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name} working on a task. "
                      f"Use your working memory to inform your response."),
            ("human", f"Current working memory:\n{context_text}\n\n"
                     f"Current task: {self.task_context.get('task', 'None')}\n\n"
                     f"Query: {query}\n\n"
                     f"Respond based on your working memory context. Be brief (2-3 sentences).")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        return str(response.content)[:300]
    
    def get_memory_status(self) -> Dict[str, Any]:
        """Get current working memory status."""
        items = self.working_memory.get_all()
        
        return {
            "capacity": self.working_memory.capacity,
            "used": len(items),
            "available": self.working_memory.capacity - len(items),
            "is_full": self.working_memory.is_full(),
            "items": [
                {
                    "id": item.item_id,
                    "priority": item.priority.name,
                    "access_count": item.access_count
                }
                for item in sorted(items, key=lambda x: x.priority.value, reverse=True)
            ]
        }


class WorkingMemoryPattern:
    """
    Working Memory Pattern - Temporary task-focused memory.
    
    Agents maintain limited-capacity memory for current task execution,
    with automatic eviction based on priority and recency.
    """
    
    def __init__(self):
        self.agents: Dict[str, WorkingMemoryAgent] = {}
    
    def register_agent(self, agent: WorkingMemoryAgent):
        """Register agent with working memory."""
        self.agents[agent.name] = agent
    
    def run(self, scenario: str) -> Dict[str, Any]:
        """Execute working memory pattern."""
        total_items = sum(len(agent.working_memory.items) for agent in self.agents.values())
        total_operations = sum(len(agent.working_memory.operations) for agent in self.agents.values())
        
        return {
            "pattern": "WorkingMemory",
            "scenario": scenario,
            "total_agents": len(self.agents),
            "total_items_in_memory": total_items,
            "total_operations": total_operations
        }


# Example usage
if __name__ == "__main__":
    print("=== Working Memory Pattern Demo ===\n")
    
    # Create pattern
    pattern = WorkingMemoryPattern()
    
    # Create agent with limited working memory
    print("1. Creating agent with working memory (capacity: 5)...")
    agent = WorkingMemoryAgent("Worker", memory_capacity=5)
    pattern.register_agent(agent)
    
    # Start a task
    print("\n2. Starting task...")
    agent.start_task("Data Analysis", {"dataset": "sales_2024", "goal": "find trends"})
    
    # Store items in working memory
    print("\n3. Storing information in working memory...")
    agent.remember("total_records", 10000, ItemPriority.HIGH)
    agent.remember("date_range", "2024-01-01 to 2024-12-31", ItemPriority.NORMAL)
    agent.remember("primary_metric", "revenue", ItemPriority.HIGH)
    agent.remember("secondary_metric", "customer_count", ItemPriority.NORMAL)
    
    # Check memory status
    print("\n4. Working memory status:")
    status = agent.get_memory_status()
    print(f"   Capacity: {status['used']}/{status['capacity']}")
    print(f"   Items:")
    for item in status['items']:
        print(f"   - {item['id']} (priority: {item['priority']}, accessed: {item['access_count']}x)")
    
    # Store more items (will trigger eviction)
    print("\n5. Adding more items (triggers eviction)...")
    agent.remember("temp_calculation", 42, ItemPriority.LOW)
    agent.remember("current_filter", "region=North", ItemPriority.NORMAL)
    
    status = agent.get_memory_status()
    print(f"   Memory now: {status['used']}/{status['capacity']}")
    
    # Recall items
    print("\n6. Recalling items from working memory...")
    agent.recall("primary_metric")
    agent.recall("total_records")
    agent.recall("temp_calculation")  # May have been evicted
    
    # Process with context
    print("\n7. Processing queries with working memory context...")
    queries = [
        "What is the main goal of my current task?",
        "What metrics am I tracking?",
        "Summarize what I'm working on"
    ]
    
    for query in queries:
        print(f"\n   Q: {query}")
        response = agent.process_with_context(query)
        print(f"   A: {response}")
    
    # Memory operations statistics
    print("\n8. Memory operation statistics:")
    ops = agent.working_memory.operations
    op_counts = {}
    for op in ops:
        op_counts[op.operation_type] = op_counts.get(op.operation_type, 0) + 1
    
    print(f"   Total operations: {len(ops)}")
    for op_type, count in op_counts.items():
        print(f"   - {op_type}: {count}")
    
    # Summary
    print("\n9. Pattern Summary:")
    summary = pattern.run("Task-focused data analysis")
    print(f"   Total items in memory: {summary['total_items_in_memory']}")
    print(f"   Total operations: {summary['total_operations']}")
    
    print("\nāœ“ Working memory pattern enabled limited-capacity task execution!")

Agentic Architecture Pattern: Working Memory

Pattern 63: 63 Working Memory – Line-by-Line Explanation

Overview

Pattern 63: Working Memory Pattern

Agents maintain temporary, limited-capacity memory for current task execution. Similar to human working memory: holds active information needed right now.

Code Structure Analysis

File Information

  • Pattern Number: 63
  • Pattern Name: 63 Working Memory
  • File Path: patterns\memory_knowledge\63_working_memory.py
  • Total Lines: 348

Detailed Code Explanation

Module Docstring

"""
Pattern 63: Working Memory Pattern

Agents maintain temporary, limited-capacity memory for current task execution.
Similar to human working memory: holds active information needed right now.
"""

Explanation: This module docstring describes the purpose and key features of the 63 Working Memory pattern.

Imports Section

from typing import Dict, Any, List, Optional, Deque
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from collections import deque
from enum import Enum
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, Deque: 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 – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Enumerations

ItemPriority (Line 17)

class ItemPriority(Enum):
    CRITICAL
    HIGH
    NORMAL
    LOW

Purpose: Priority level for working memory items.

Values: – CRITICAL: Represents critical – HIGH: Represents high – NORMAL: Represents normal – LOW: Represents low

Classes

ItemPriority (Line 17)

Description: Priority level for working memory items.

WorkingMemoryItem (Line 26)

Description: Item in working memory.

MemoryOperation (Line 39)

Description: Record of memory operation.

WorkingMemory (Line 47)

Description: Limited-capacity working memory with eviction policies.

Methods:

__init__ (Line 52)
def __init__(self, capacity):
    """No method docstring."""

Purpose: No method docstring.

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

store (Line 58)
def store(self, item):
    """Store item in working memory."""

Purpose: Store item in working memory.

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

retrieve (Line 77)
def retrieve(self, item_id):
    """Retrieve item from working memory."""

Purpose: Retrieve item from working memory.

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

update (Line 104)
def update(self, item_id, new_content):
    """Update item in working memory."""

Purpose: Update item in working memory.

Parameters: – self: Instance reference – item_id: Parameter for item_id – new_content: Parameter for new_content

_evict (Line 116)
def _evict(self):
    """Evict item based on priority and recency."""

Purpose: Evict item based on priority and recency.

Parameters: – self: Instance reference

_remove (Line 135)
def _remove(self, item_id):
    """Remove item from working memory."""

Purpose: Remove item from working memory.

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

get_all (Line 144)
def get_all(self):
    """Get all items in working memory."""

Purpose: Get all items in working memory.

Parameters: – self: Instance reference

clear (Line 148)
def clear(self):
    """Clear all working memory."""

Purpose: Clear all working memory.

Parameters: – self: Instance reference

is_full (Line 153)
def is_full(self):
    """Check if working memory is at capacity."""

Purpose: Check if working memory is at capacity.

Parameters: – self: Instance reference

WorkingMemoryAgent (Line 158)

Description: Agent with working memory for current task execution.

Methods:

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

Purpose: No method docstring.

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

start_task (Line 167)
def start_task(self, task_name, task_data):
    """Start new task, initializing working memory."""

Purpose: Start new task, initializing working memory.

Parameters: – self: Instance reference – task_name: Parameter for task_name – task_data: Parameter for task_data

remember (Line 182)
def remember(self, key, value, priority, ttl):
    """Store information in working memory."""

Purpose: Store information in working memory.

Parameters: – self: Instance reference – key: Parameter for key – value: Parameter for value – priority: Parameter for priority – ttl: Parameter for ttl

recall (Line 189)
def recall(self, key):
    """Recall information from working memory."""

Purpose: Recall information from working memory.

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

process_with_context (Line 199)
def process_with_context(self, query):
    """Process query using working memory context."""

Purpose: Process query using working memory context.

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

get_memory_status (Line 223)
def get_memory_status(self):
    """Get current working memory status."""

Purpose: Get current working memory status.

Parameters: – self: Instance reference

WorkingMemoryPattern (Line 243)

Description: Working Memory Pattern – Temporary task-focused memory.

Agents maintain limited-capacity memory for current task execution, with automatic eviction based on priority and recency.

Methods:

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

Purpose: No method docstring.

Parameters: – self: Instance reference

register_agent (Line 254)
def register_agent(self, agent):
    """Register agent with working memory."""

Purpose: Register agent with working memory.

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

run (Line 258)
def run(self, scenario):
    """Execute working memory pattern."""

Purpose: Execute working 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 = ItemPriority(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