aishastr.live

Agentic Architecture Pattern: Short Term Memory

Written by

admin

in


    """
Pattern 65: Short-Term Memory Pattern

Agents maintain temporary memory with rapid decay for immediate task context.
Holds information briefly (seconds to minutes) for active use.
"""

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


@dataclass
class ShortTermItem:
    """Item in short-term memory with automatic decay."""
    item_id: str
    content: Any
    created_at: datetime = field(default_factory=datetime.now)
    ttl_seconds: float = 30.0  # Time to live in seconds
    access_count: int = 0
    
    def is_expired(self) -> bool:
        """Check if item has expired."""
        age = (datetime.now() - self.created_at).total_seconds()
        return age > self.ttl_seconds
    
    def time_remaining(self) -> float:
        """Get remaining time before expiration."""
        age = (datetime.now() - self.created_at).total_seconds()
        return max(0, self.ttl_seconds - age)


class ShortTermMemory:
    """
    Short-term memory with rapid decay and limited retention.
    """
    
    def __init__(self, capacity: int = 7, default_ttl: float = 30.0):
        self.capacity = capacity
        self.default_ttl = default_ttl
        self.items: Dict[str, ShortTermItem] = {}
        self.access_log: Deque[str] = deque(maxlen=100)
    
    def store(self, item_id: str, content: Any, ttl: Optional[float] = None):
        """Store item in short-term memory."""
        # Clean expired items first
        self._clean_expired()
        
        # Check capacity
        if len(self.items) >= self.capacity and item_id not in self.items:
            self._evict_oldest()
        
        # Store item
        item = ShortTermItem(
            item_id=item_id,
            content=content,
            ttl_seconds=ttl if ttl is not None else self.default_ttl
        )
        self.items[item_id] = item
        self.access_log.append(f"store:{item_id}")
    
    def retrieve(self, item_id: str) -> Optional[Any]:
        """Retrieve item from short-term memory."""
        # Clean expired items
        self._clean_expired()
        
        if item_id not in self.items:
            self.access_log.append(f"miss:{item_id}")
            return None
        
        item = self.items[item_id]
        
        if item.is_expired():
            del self.items[item_id]
            self.access_log.append(f"expired:{item_id}")
            return None
        
        item.access_count += 1
        self.access_log.append(f"hit:{item_id}")
        return item.content
    
    def _clean_expired(self):
        """Remove expired items."""
        expired = [iid for iid, item in self.items.items() if item.is_expired()]
        for iid in expired:
            del self.items[iid]
    
    def _evict_oldest(self):
        """Evict oldest item."""
        if not self.items:
            return
        
        oldest_id = min(self.items.keys(), 
                       key=lambda k: self.items[k].created_at)
        del self.items[oldest_id]
        self.access_log.append(f"evict:{oldest_id}")
    
    def get_active_items(self) -> List[ShortTermItem]:
        """Get all non-expired items."""
        self._clean_expired()
        return list(self.items.values())
    
    def clear(self):
        """Clear all short-term memory."""
        self.items.clear()
        self.access_log.append("clear:all")


class ShortTermMemoryAgent:
    """Agent with short-term memory for immediate context."""
    
    def __init__(self, name: str, memory_capacity: int = 5, 
                 default_ttl: float = 30.0, model_name: str = "gpt-4"):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.stm = ShortTermMemory(capacity=memory_capacity, default_ttl=default_ttl)
        self.conversation_buffer: Deque[str] = deque(maxlen=10)
    
    def note(self, key: str, value: Any, ttl: Optional[float] = None):
        """Quickly note something in short-term memory."""
        self.stm.store(key, value, ttl)
        print(f"   šŸ“ {self.name} noted: {key} (expires in {ttl or self.stm.default_ttl}s)")
    
    def remember(self, key: str) -> Optional[Any]:
        """Try to remember from short-term memory."""
        value = self.stm.retrieve(key)
        if value:
            item = self.stm.items.get(key)
            remaining = item.time_remaining() if item else 0
            print(f"   āœ“ {self.name} remembered: {key} ({remaining:.1f}s remaining)")
        else:
            print(f"   āœ— {self.name} forgot: {key} (expired or not found)")
        return value
    
    def process_immediate_context(self, input_text: str) -> str:
        """Process input with immediate short-term memory context."""
        # Store input in conversation buffer
        self.conversation_buffer.append(f"Input: {input_text}")
        
        # Get active memory items
        active_items = self.stm.get_active_items()
        
        if not active_items:
            context = "No immediate context in short-term memory."
        else:
            context = "\n".join([
                f"- {item.item_id}: {item.content} ({item.time_remaining():.1f}s left)"
                for item in sorted(active_items, 
                                  key=lambda x: x.time_remaining(), 
                                  reverse=True)
            ])
        
        # Recent conversation
        recent_conv = "\n".join(list(self.conversation_buffer)[-3:])
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}. You have short-term memory for immediate context. "
                      f"Memory items decay quickly (within seconds)."),
            ("human", f"Short-term memory:\n{context}\n\n"
                     f"Recent conversation:\n{recent_conv}\n\n"
                     f"Current input: {input_text}\n\n"
                     f"Respond briefly (1-2 sentences) using your short-term memory.")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        response_text = str(response.content)[:200]
        self.conversation_buffer.append(f"Response: {response_text}")
        
        return response_text


class ShortTermMemoryPattern:
    """
    Short-Term Memory Pattern - Temporary rapid-decay memory.
    
    Agents hold information briefly for immediate use, with automatic
    expiration and capacity limits mimicking human short-term memory.
    """
    
    def __init__(self):
        self.agents: Dict[str, ShortTermMemoryAgent] = {}
    
    def register_agent(self, agent: ShortTermMemoryAgent):
        """Register agent with short-term memory."""
        self.agents[agent.name] = agent
    
    def run(self, scenario: str) -> Dict[str, Any]:
        """Execute short-term memory pattern."""
        total_items = sum(len(agent.stm.items) for agent in self.agents.values())
        total_accesses = sum(len(agent.stm.access_log) for agent in self.agents.values())
        
        return {
            "pattern": "ShortTermMemory",
            "scenario": scenario,
            "total_agents": len(self.agents),
            "active_items": total_items,
            "total_accesses": total_accesses
        }


# Example usage
if __name__ == "__main__":
    import time
    
    print("=== Short-Term Memory Pattern Demo ===\n")
    
    # Create pattern
    pattern = ShortTermMemoryPattern()
    
    # Create agent with short-term memory (30 second TTL)
    print("1. Creating agent with short-term memory...")
    agent = ShortTermMemoryAgent("Alice", memory_capacity=5, default_ttl=15.0)
    pattern.register_agent(agent)
    
    # Store items with different TTLs
    print("\n2. Storing items in short-term memory...")
    agent.note("customer_name", "John Doe", ttl=20.0)
    agent.note("order_number", "ORD-12345", ttl=15.0)
    agent.note("total_amount", "$125.50", ttl=10.0)
    agent.note("shipping_address", "123 Main St", ttl=15.0)
    
    # Immediate recall
    print("\n3. Immediate recall (should work)...")
    agent.remember("customer_name")
    agent.remember("order_number")
    
    # Check active items
    print("\n4. Active items in short-term memory:")
    active = agent.stm.get_active_items()
    for item in sorted(active, key=lambda x: x.time_remaining(), reverse=True):
        print(f"   - {item.item_id}: {item.time_remaining():.1f}s remaining")
    
    # Wait a bit
    print("\n5. Waiting 8 seconds...")
    time.sleep(8)
    
    # Try to recall again
    print("\n6. Recall after 8 seconds...")
    agent.remember("customer_name")  # Should still work (20s TTL)
    agent.remember("total_amount")  # May be expired (10s TTL)
    
    # Process with context
    print("\n7. Processing with short-term memory context...")
    queries = [
        "What's the customer's name?",
        "What's the order total?",
        "Where should we ship it?"
    ]
    
    for query in queries:
        print(f"\n   Q: {query}")
        response = agent.process_immediate_context(query)
        print(f"   A: {response}")
        time.sleep(2)  # Brief pause between queries
    
    # Wait for more expiration
    print("\n8. Waiting 10 more seconds for memory decay...")
    time.sleep(10)
    
    # Try to recall again
    print("\n9. Recall after total 18 seconds...")
    agent.remember("customer_name")  # May be expired
    agent.remember("order_number")  # Likely expired
    
    # Check what's left
    print("\n10. Remaining items in short-term memory:")
    remaining = agent.stm.get_active_items()
    if remaining:
        for item in remaining:
            print(f"   - {item.item_id}: {item.time_remaining():.1f}s remaining")
    else:
        print("   (No items remaining - all expired)")
    
    # Statistics
    print("\n11. Memory Statistics:")
    summary = pattern.run("Customer order processing")
    print(f"   Active items: {summary['active_items']}")
    print(f"   Total accesses: {summary['total_accesses']}")
    
    # Access log analysis
    log = list(agent.stm.access_log)
    hit_count = sum(1 for entry in log if entry.startswith("hit:"))
    miss_count = sum(1 for entry in log if entry.startswith("miss:"))
    expired_count = sum(1 for entry in log if entry.startswith("expired:"))
    
    print(f"\n12. Access Log Analysis:")
    print(f"   Hits: {hit_count}")
    print(f"   Misses: {miss_count}")
    print(f"   Expired: {expired_count}")
    if hit_count + miss_count > 0:
        hit_rate = hit_count / (hit_count + miss_count) * 100
        print(f"   Hit rate: {hit_rate:.1f}%")
    
    print("\nāœ“ Short-term memory pattern demonstrated rapid decay!")

Agentic Architecture Pattern: Short Term Memory

Pattern 65: 65 Short Term Memory – Line-by-Line Explanation

Overview

Pattern 65: Short-Term Memory Pattern

Agents maintain temporary memory with rapid decay for immediate task context. Holds information briefly (seconds to minutes) for active use.

Code Structure Analysis

File Information

  • Pattern Number: 65
  • Pattern Name: 65 Short Term Memory
  • File Path: patterns\memory_knowledge\65_short_term_memory.py
  • Total Lines: 300

Detailed Code Explanation

Module Docstring

"""
Pattern 65: Short-Term Memory Pattern

Agents maintain temporary memory with rapid decay for immediate task context.
Holds information briefly (seconds to minutes) for active use.
"""

Explanation: This module docstring describes the purpose and key features of the 65 Short Term 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 langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
    import time  # 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 langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Classes

ShortTermItem (Line 17)

Description: Item in short-term memory with automatic decay.

Methods:

is_expired (Line 25)
def is_expired(self):
    """Check if item has expired."""

Purpose: Check if item has expired.

Parameters: – self: Instance reference

time_remaining (Line 30)
def time_remaining(self):
    """Get remaining time before expiration."""

Purpose: Get remaining time before expiration.

Parameters: – self: Instance reference

ShortTermMemory (Line 36)

Description: Short-term memory with rapid decay and limited retention.

Methods:

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

Purpose: No method docstring.

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

store (Line 47)
def store(self, item_id, content, ttl):
    """Store item in short-term memory."""

Purpose: Store item in short-term memory.

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

retrieve (Line 65)
def retrieve(self, item_id):
    """Retrieve item from short-term memory."""

Purpose: Retrieve item from short-term memory.

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

_clean_expired (Line 85)
def _clean_expired(self):
    """Remove expired items."""

Purpose: Remove expired items.

Parameters: – self: Instance reference

_evict_oldest (Line 91)
def _evict_oldest(self):
    """Evict oldest item."""

Purpose: Evict oldest item.

Parameters: – self: Instance reference

get_active_items (Line 101)
def get_active_items(self):
    """Get all non-expired items."""

Purpose: Get all non-expired items.

Parameters: – self: Instance reference

clear (Line 106)
def clear(self):
    """Clear all short-term memory."""

Purpose: Clear all short-term memory.

Parameters: – self: Instance reference

ShortTermMemoryAgent (Line 112)

Description: Agent with short-term memory for immediate context.

Methods:

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

Purpose: No method docstring.

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

note (Line 122)
def note(self, key, value, ttl):
    """Quickly note something in short-term memory."""

Purpose: Quickly note something in short-term memory.

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

remember (Line 127)
def remember(self, key):
    """Try to remember from short-term memory."""

Purpose: Try to remember from short-term memory.

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

process_immediate_context (Line 138)
def process_immediate_context(self, input_text):
    """Process input with immediate short-term memory context."""

Purpose: Process input with immediate short-term memory context.

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

ShortTermMemoryPattern (Line 177)

Description: Short-Term Memory Pattern – Temporary rapid-decay memory.

Agents hold information briefly for immediate use, with automatic expiration and capacity limits mimicking human short-term memory.

Methods:

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

Purpose: No method docstring.

Parameters: – self: Instance reference

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

Purpose: Register agent with short-term memory.

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

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

Purpose: Execute short-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 = ShortTermItem(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