
    """
Pattern 026: Short-Term Memory

Description:
    The Short-Term Memory pattern enables agents to maintain context within the current
    conversation or task session. It provides immediate access to recent interactions,
    allowing the agent to reference previous messages, maintain conversation flow, and
    build upon earlier context without requiring persistent storage.

Components:
    - Conversation Buffer: Stores recent message history
    - Context Window Manager: Manages token limits
    - Message Formatter: Structures messages for LLM
    - Relevance Filter: Keeps most relevant context
    - Memory Summarizer: Condenses long conversations

Use Cases:
    - Multi-turn conversations
    - Task continuity within session
    - Context-aware responses
    - Follow-up questions
    - Anaphora resolution

LangChain Implementation:
    Uses LangChain's memory abstractions including ConversationBufferMemory,
    ConversationBufferWindowMemory, and ConversationSummaryMemory for
    different short-term memory strategies.

Key Features:
    - Automatic message history tracking
    - Context window management
    - Memory clearing and reset
    - Selective memory retention
    - Token-aware buffering
"""

import os
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime
from dotenv import load_dotenv

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_openai import ChatOpenAI

load_dotenv()


class MemoryStrategy(Enum):
    """Strategy for managing short-term memory."""
    BUFFER = "buffer"  # Keep all messages
    WINDOW = "window"  # Keep last N messages
    TOKEN_LIMIT = "token_limit"  # Keep within token limit
    SUMMARY = "summary"  # Summarize older messages


@dataclass
class Message:
    """Represents a conversation message."""
    role: str  # "human", "ai", "system"
    content: str
    timestamp: datetime = field(default_factory=datetime.now)
    metadata: Dict[str, Any] = field(default_factory=dict)
    
    def to_langchain_message(self):
        """Convert to LangChain message format."""
        if self.role == "human":
            return HumanMessage(content=self.content)
        elif self.role == "ai":
            return AIMessage(content=self.content)
        else:
            return SystemMessage(content=self.content)


@dataclass
class ConversationContext:
    """Context for current conversation."""
    messages: List[Message] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    started_at: datetime = field(default_factory=datetime.now)
    
    def add_message(self, role: str, content: str, metadata: Optional[Dict] = None):
        """Add a message to the conversation."""
        message = Message(
            role=role,
            content=content,
            metadata=metadata or {}
        )
        self.messages.append(message)
    
    def get_recent_messages(self, n: int) -> List[Message]:
        """Get the N most recent messages."""
        return self.messages[-n:] if n > 0 else []
    
    def get_messages_by_role(self, role: str) -> List[Message]:
        """Get all messages from a specific role."""
        return [msg for msg in self.messages if msg.role == role]
    
    def clear(self):
        """Clear all messages."""
        self.messages.clear()
        self.started_at = datetime.now()


class BufferMemory:
    """
    Simple buffer that stores all conversation messages.
    """
    
    def __init__(self, max_messages: Optional[int] = None):
        """
        Initialize buffer memory.
        
        Args:
            max_messages: Maximum number of messages to store (None = unlimited)
        """
        self.max_messages = max_messages
        self.context = ConversationContext()
    
    def add_interaction(self, human_message: str, ai_message: str):
        """Add a human-AI interaction pair."""
        self.context.add_message("human", human_message)
        self.context.add_message("ai", ai_message)
        
        # Trim if exceeding max
        if self.max_messages and len(self.context.messages) > self.max_messages:
            excess = len(self.context.messages) - self.max_messages
            self.context.messages = self.context.messages[excess:]
    
    def get_context_string(self) -> str:
        """Get conversation history as formatted string."""
        lines = []
        for msg in self.context.messages:
            role = msg.role.capitalize()
            lines.append(f"{role}: {msg.content}")
        return "\n".join(lines)
    
    def get_langchain_messages(self) -> List:
        """Get messages in LangChain format."""
        return [msg.to_langchain_message() for msg in self.context.messages]
    
    def clear(self):
        """Clear memory."""
        self.context.clear()


class WindowMemory:
    """
    Sliding window memory that keeps only the last N messages.
    """
    
    def __init__(self, window_size: int = 10):
        """
        Initialize window memory.
        
        Args:
            window_size: Number of recent messages to keep
        """
        self.window_size = window_size
        self.context = ConversationContext()
    
    def add_interaction(self, human_message: str, ai_message: str):
        """Add interaction and maintain window."""
        self.context.add_message("human", human_message)
        self.context.add_message("ai", ai_message)
        
        # Keep only window_size messages
        if len(self.context.messages) > self.window_size:
            self.context.messages = self.context.messages[-self.window_size:]
    
    def get_context_string(self) -> str:
        """Get recent conversation history."""
        lines = []
        for msg in self.context.messages:
            role = msg.role.capitalize()
            lines.append(f"{role}: {msg.content}")
        return "\n".join(lines)
    
    def get_langchain_messages(self) -> List:
        """Get messages in LangChain format."""
        return [msg.to_langchain_message() for msg in self.context.messages]


class TokenLimitMemory:
    """
    Memory that keeps messages within a token budget.
    """
    
    def __init__(self, max_tokens: int = 2000):
        """
        Initialize token-limit memory.
        
        Args:
            max_tokens: Maximum tokens to keep in memory
        """
        self.max_tokens = max_tokens
        self.context = ConversationContext()
    
    def _estimate_tokens(self, text: str) -> int:
        """Estimate token count (rough approximation)."""
        # Simple estimation: ~4 characters per token
        return len(text) // 4
    
    def add_interaction(self, human_message: str, ai_message: str):
        """Add interaction and maintain token limit."""
        self.context.add_message("human", human_message)
        self.context.add_message("ai", ai_message)
        
        # Remove oldest messages until under token limit
        while self._get_total_tokens() > self.max_tokens and len(self.context.messages) > 2:
            self.context.messages.pop(0)
    
    def _get_total_tokens(self) -> int:
        """Calculate total tokens in memory."""
        total = 0
        for msg in self.context.messages:
            total += self._estimate_tokens(msg.content)
        return total
    
    def get_context_string(self) -> str:
        """Get conversation history."""
        lines = []
        for msg in self.context.messages:
            role = msg.role.capitalize()
            lines.append(f"{role}: {msg.content}")
        return "\n".join(lines)
    
    def get_token_count(self) -> int:
        """Get current token count."""
        return self._get_total_tokens()


class SummaryMemory:
    """
    Memory that summarizes older messages to save space.
    """
    
    def __init__(
        self,
        recent_messages: int = 4,
        model: str = "gpt-3.5-turbo"
    ):
        """
        Initialize summary memory.
        
        Args:
            recent_messages: Number of recent messages to keep verbatim
            model: LLM model for summarization
        """
        self.recent_messages = recent_messages
        self.llm = ChatOpenAI(model=model, temperature=0.3)
        self.context = ConversationContext()
        self.summary = ""
    
    def add_interaction(self, human_message: str, ai_message: str):
        """Add interaction and update summary if needed."""
        self.context.add_message("human", human_message)
        self.context.add_message("ai", ai_message)
        
        # If we have more than recent_messages, summarize older ones
        if len(self.context.messages) > self.recent_messages:
            messages_to_summarize = self.context.messages[:-self.recent_messages]
            self.summary = self._create_summary(messages_to_summarize)
            self.context.messages = self.context.messages[-self.recent_messages:]
    
    def _create_summary(self, messages: List[Message]) -> str:
        """Create summary of messages."""
        if not messages:
            return self.summary
        
        # Format messages
        formatted = "\n".join([
            f"{msg.role.capitalize()}: {msg.content}"
            for msg in messages
        ])
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", """Summarize the following conversation, preserving key information
and context. Be concise but comprehensive."""),
            ("user", "Previous summary: {summary}\n\nNew messages:\n{messages}\n\nProvide updated summary:")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        
        new_summary = chain.invoke({
            "summary": self.summary if self.summary else "None",
            "messages": formatted
        })
        
        return new_summary.strip()
    
    def get_context_string(self) -> str:
        """Get conversation context with summary."""
        parts = []
        
        if self.summary:
            parts.append(f"[Summary of earlier conversation: {self.summary}]\n")
        
        for msg in self.context.messages:
            role = msg.role.capitalize()
            parts.append(f"{role}: {msg.content}")
        
        return "\n".join(parts)


class ShortTermMemoryAgent:
    """
    Agent with short-term memory for conversations.
    """
    
    def __init__(
        self,
        memory_strategy: MemoryStrategy = MemoryStrategy.WINDOW,
        model: str = "gpt-3.5-turbo",
        **memory_kwargs
    ):
        """
        Initialize agent with short-term memory.
        
        Args:
            memory_strategy: Strategy for memory management
            model: LLM model to use
            **memory_kwargs: Arguments for memory initialization
        """
        self.llm = ChatOpenAI(model=model, temperature=0.7)
        self.memory_strategy = memory_strategy
        
        # Initialize appropriate memory type
        if memory_strategy == MemoryStrategy.BUFFER:
            self.memory = BufferMemory(**memory_kwargs)
        elif memory_strategy == MemoryStrategy.WINDOW:
            self.memory = WindowMemory(**memory_kwargs)
        elif memory_strategy == MemoryStrategy.TOKEN_LIMIT:
            self.memory = TokenLimitMemory(**memory_kwargs)
        elif memory_strategy == MemoryStrategy.SUMMARY:
            self.memory = SummaryMemory(**memory_kwargs)
    
    def chat(self, user_message: str) -> str:
        """
        Chat with the agent.
        
        Args:
            user_message: User's message
            
        Returns:
            Agent's response
        """
        # Get conversation context
        context = self.memory.get_context_string()
        
        # Create prompt with context
        prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a helpful AI assistant. Use the conversation history
to provide context-aware responses. Reference previous messages when relevant."""),
            ("user", "{context}\n\nHuman: {message}\n\nProvide a helpful response:")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        
        # Generate response
        response = chain.invoke({
            "context": context if context else "No previous conversation.",
            "message": user_message
        })
        
        # Store interaction
        self.memory.add_interaction(user_message, response)
        
        return response.strip()
    
    def get_conversation_summary(self) -> Dict[str, Any]:
        """Get summary of current conversation."""
        message_count = len(self.memory.context.messages)
        
        summary = {
            "strategy": self.memory_strategy.value,
            "message_count": message_count,
            "human_messages": len(self.memory.context.get_messages_by_role("human")),
            "ai_messages": len(self.memory.context.get_messages_by_role("ai")),
            "started_at": self.memory.context.started_at.isoformat()
        }
        
        # Add strategy-specific info
        if hasattr(self.memory, 'get_token_count'):
            summary["token_count"] = self.memory.get_token_count()
        
        if hasattr(self.memory, 'summary') and self.memory.summary:
            summary["has_summary"] = True
            summary["summary_length"] = len(self.memory.summary)
        
        return summary
    
    def clear_memory(self):
        """Clear conversation memory."""
        self.memory.context.clear()


def demonstrate_short_term_memory():
    """Demonstrate the Short-Term Memory pattern."""
    
    print("=" * 80)
    print("SHORT-TERM MEMORY PATTERN DEMONSTRATION")
    print("=" * 80)
    
    # Test 1: Buffer Memory
    print("\n" + "=" * 80)
    print("TEST 1: Buffer Memory (Unlimited)")
    print("=" * 80)
    
    agent1 = ShortTermMemoryAgent(
        memory_strategy=MemoryStrategy.BUFFER,
        max_messages=None
    )
    
    conversation1 = [
        "Hi, my name is Alice.",
        "What's my name?",
        "I'm interested in learning about AI.",
        "What did I say I was interested in?"
    ]
    
    for msg in conversation1:
        print(f"\n👤 Human: {msg}")
        response = agent1.chat(msg)
        print(f"🤖 AI: {response}")
    
    summary1 = agent1.get_conversation_summary()
    print(f"\n📊 Conversation Summary:")
    print(f"   Messages: {summary1['message_count']}")
    print(f"   Strategy: {summary1['strategy']}")
    
    # Test 2: Window Memory
    print("\n\n" + "=" * 80)
    print("TEST 2: Window Memory (Last 4 Messages)")
    print("=" * 80)
    
    agent2 = ShortTermMemoryAgent(
        memory_strategy=MemoryStrategy.WINDOW,
        window_size=4
    )
    
    conversation2 = [
        "My favorite color is blue.",
        "I live in New York.",
        "I work as a teacher.",
        "I like pizza.",
        "What's my favorite color?",  # This should be forgotten
        "Where do I live?",  # This should be remembered
    ]
    
    for msg in conversation2:
        print(f"\n👤 Human: {msg}")
        response = agent2.chat(msg)
        print(f"🤖 AI: {response}")
    
    summary2 = agent2.get_conversation_summary()
    print(f"\n📊 Conversation Summary:")
    print(f"   Messages in window: {summary2['message_count']}")
    print(f"   Strategy: {summary2['strategy']}")
    
    # Test 3: Token Limit Memory
    print("\n\n" + "=" * 80)
    print("TEST 3: Token Limit Memory (2000 tokens)")
    print("=" * 80)
    
    agent3 = ShortTermMemoryAgent(
        memory_strategy=MemoryStrategy.TOKEN_LIMIT,
        max_tokens=2000
    )
    
    conversation3 = [
        "Let me tell you about my day. " * 20,  # Long message
        "What did I just tell you about?",
        "Here's another long story. " * 20,  # Another long message
        "Can you remember both stories?"
    ]
    
    for msg in conversation3:
        msg_preview = msg[:60] + "..." if len(msg) > 60 else msg
        print(f"\n👤 Human: {msg_preview}")
        response = agent3.chat(msg)
        response_preview = response[:100] + "..." if len(response) > 100 else response
        print(f"🤖 AI: {response_preview}")
    
    summary3 = agent3.get_conversation_summary()
    print(f"\n📊 Conversation Summary:")
    print(f"   Messages: {summary3['message_count']}")
    print(f"   Token count: {summary3['token_count']}")
    print(f"   Strategy: {summary3['strategy']}")
    
    # Summary
    print("\n\n" + "=" * 80)
    print("PATTERN SUMMARY")
    print("=" * 80)
    print("""
The Short-Term Memory pattern demonstrates:

1. **Context Continuity**: Maintains conversation flow
2. **Multiple Strategies**: Different approaches for different needs
3. **Resource Management**: Controls memory usage
4. **Context Awareness**: References previous messages
5. **Session Scoped**: Memory persists within session

Memory Strategies Compared:

Buffer Memory:
- ✓ Complete conversation history
- ✓ Perfect recall
- ✗ Unbounded growth
- Use: Short conversations, debugging

Window Memory:
- ✓ Fixed memory size
- ✓ Recent context preserved
- ✗ Forgets older messages
- Use: Long conversations, resource constraints

Token Limit Memory:
- ✓ Token budget control
- ✓ Adapts to message length
- ✗ May lose context unpredictably
- Use: API token limits, cost control

Summary Memory:
- ✓ Compact representation
- ✓ Preserves key information
- ✗ Lossy compression
- Use: Very long conversations

Key Benefits:
- **Natural Conversations**: Agents remember context
- **Anaphora Resolution**: Handle "it", "that", "they" references
- **Follow-up Questions**: Build on previous topics
- **Resource Control**: Manage memory/token usage
- **Session Management**: Clear boundaries

Use Cases:
- Customer service chatbots
- Personal assistants
- Tutorial/teaching agents
- Technical support
- Conversational interfaces

This pattern is fundamental for any conversational AI system,
enabling natural multi-turn interactions with context awareness.
""")


if __name__ == "__main__":
    demonstrate_short_term_memory()

Agentic Design Pattern: Short Term Memory

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

Overview

This document provides a detailed line-by-line explanation of the Short Term Memory pattern implementation.

Pattern Description

Pattern 026: Short-Term Memory

Description: The Short-Term Memory pattern enables agents to maintain context within the current conversation or task session. It provides immediate access to recent interactions, allowing the agent to reference previous messages, maintain conversation flow, and build upon earlier context without requiring persistent storage.

Components: – Conversation Buffer: Stores recent message history – Context Window Manager: Manages token limits – Message Formatter: Structures messages for LLM – Relevance Filter: Keeps most relevant context – Memory Summarizer: Condenses long conversations

Use Cases: – Multi-turn conversations – Task continuity within session – Context-aware responses – Follow-up questions – Anaphora resolution

LangChain Implementation: Uses LangChain’s memory abstractions including ConversationBufferMemory, ConversationBufferWindowMemory, and ConversationSummaryMemory for different short-term memory strategies.

Key Features: – Automatic message history tracking – Context window management – Memory clearing and reset – Selective memory retention – Token-aware buffering

File Statistics

Total Lines: 554
Classes: 8
Functions: 1
Constants: 0

Module Documentation

"""
Pattern 026: Short-Term Memory

Description:
    The Short-Term Memory pattern enables agents to maintain context within the current
    conversation or task session. It provides immediate access to recent interactions,
    allowing the agent to reference previous messages, maintain conversation flow, and
    build upon earlier context without requiring persistent storage.

Components:
    - Conversation Buffer: Stores recent message history
    - Context Window Manager: Manages token limits
    - Message Formatter: Structures messages for LLM
    - Relevance Filter: Keeps most relevant context
    - Memory Summarizer: Condenses long conversations

Use Cases:
    - Multi-turn conversations
    - Task continuity within session
    - Context-aware responses
    - Follow-up questions
    - Anaphora resolution

LangChain Implementation:
    Uses LangChain's memory abstractions including ConversationBufferMemory,
    ConversationBufferWindowMemory, and ConversationSummaryMemory for
    different short-term memory strategies.

Key Features:
    - Automatic message history tracking
    - Context window management
    - Memory clearing and reset
    - Selective memory retention
    - Token-aware buffering
"""

Purpose: This module implements the Short Term Memory pattern, providing a structured approach to agent design and implementation.

Import Statements

Standard Library Imports

import dataclasses.dataclass
import dataclasses.field
import datetime.datetime
import enum.Enum
import langchain_core.messages.AIMessage
import langchain_core.messages.HumanMessage
import langchain_core.messages.SystemMessage
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import langchain_core.prompts.MessagesPlaceholder
import os
import typing.Any
import typing.Dict
import typing.List
import typing.Optional

Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – datetime.datetime: Date and time handling – enum.Enum: Enumeration support – langchain_core.messages.AIMessage: Regular expression operations – langchain_core.messages.HumanMessage: Regular expression operations – langchain_core.messages.SystemMessage: Regular expression operations – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – langchain_core.prompts.MessagesPlaceholder: Regular expression operations – os: Environment variable access and system operations – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI

Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows

Other Imports

from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

Classes

Class: `MemoryStrategy`

Lines 52-57

Inherits from: Enum

Purpose:

Strategy for managing short-term memory.

Class: `Message`

Lines 61-75

Purpose:

Represents a conversation message.

Methods:

`to_langchain_message(self)`

Lines 68-75

Purpose:

Convert to LangChain message format.

Parameters:

Class: `ConversationContext`

Lines 79-105

Purpose:

Context for current conversation.

Methods:

`add_message(self, role, content, metadata)`

Lines 85-92

Purpose:

Add a message to the conversation.

Parameters: – role: Role – content: Content – metadata: Metadata

`get_recent_messages(self, n)`

Lines 94-96

Purpose:

Get the N most recent messages.

Parameters: – n: N

Returns: <ast.Subscript object at 0x000001A06D17A5D0>

`get_messages_by_role(self, role)`

Lines 98-100

Purpose:

Get all messages from a specific role.

Parameters: – role: Role

Returns: <ast.Subscript object at 0x000001A06D1789D0>

`clear(self)`

Lines 102-105

Purpose:

Clear all messages.

Parameters:

Class: `BufferMemory`

Lines 108-147

Purpose:

Simple buffer that stores all conversation messages.

Methods:

`init(self, max_messages)`

Lines 113-121

Purpose:

Initialize buffer memory.

Args:
    max_messages: Maximum number of messages to store (None = unlimited)

Parameters: – max_messages: Max messages

`add_interaction(self, human_message, ai_message)`

Lines 123-131

Purpose:

Add a human-AI interaction pair.

Parameters: – human_message: Human message – ai_message: Ai message

`get_context_string(self)`

Lines 133-139

Purpose:

Get conversation history as formatted string.

Parameters:

Returns: str

`get_langchain_messages(self)`

Lines 141-143

Purpose:

Get messages in LangChain format.

Parameters:

Returns: List

`clear(self)`

Lines 145-147

Purpose:

Clear memory.

Parameters:

Class: `WindowMemory`

Lines 150-184

Purpose:

Sliding window memory that keeps only the last N messages.

Methods:

`init(self, window_size)`

Lines 155-163

Purpose:

Initialize window memory.

Args:
    window_size: Number of recent messages to keep

Parameters: – window_size: Window size

`add_interaction(self, human_message, ai_message)`

Lines 165-172

Purpose:

Add interaction and maintain window.

Parameters: – human_message: Human message – ai_message: Ai message

`get_context_string(self)`

Lines 174-180

Purpose:

Get recent conversation history.

Parameters:

Returns: str

`get_langchain_messages(self)`

Lines 182-184

Purpose:

Get messages in LangChain format.

Parameters:

Returns: List

Class: `TokenLimitMemory`

Lines 187-233

Purpose:

Memory that keeps messages within a token budget.

Methods:

`init(self, max_tokens)`

Lines 192-200

Purpose:

Initialize token-limit memory.

Args:
    max_tokens: Maximum tokens to keep in memory

Parameters: – max_tokens: Maximum tokens to generate

`_estimate_tokens(self, text)`

Lines 202-205

Purpose:

Estimate token count (rough approximation).

Parameters: – text: Text

Returns: int

`add_interaction(self, human_message, ai_message)`

Lines 207-214

Purpose:

Add interaction and maintain token limit.

Parameters: – human_message: Human message – ai_message: Ai message

`_get_total_tokens(self)`

Lines 216-221

Purpose:

Calculate total tokens in memory.

Parameters:

Returns: int

`get_context_string(self)`

Lines 223-229

Purpose:

Get conversation history.

Parameters:

Returns: str

`get_token_count(self)`

Lines 231-233

Purpose:

Get current token count.

Parameters:

Returns: int

Class: `SummaryMemory`

Lines 236-306

Purpose:

Memory that summarizes older messages to save space.

Methods:

`init(self, recent_messages, model)`

Lines 241-256

Purpose:

Initialize summary memory.

Args:
    recent_messages: Number of recent messages to keep verbatim
    model: LLM model for summarization

Parameters: – recent_messages: Recent messages – model: Language model instance

`add_interaction(self, human_message, ai_message)`

Lines 258-267

Purpose:

Add interaction and update summary if needed.

Parameters: – human_message: Human message – ai_message: Ai message

`_create_summary(self, messages)`

Lines 269-293

Purpose:

Create summary of messages.

Parameters: – messages: Messages

Returns: str

`get_context_string(self)`

Lines 295-306

Purpose:

Get conversation context with summary.

Parameters:

Returns: str

Class: `ShortTermMemoryAgent`

Lines 309-398

Purpose:

Agent with short-term memory for conversations.

Methods:

`init(self, memory_strategy, model)`

Lines 314-339

Purpose:

Initialize agent with short-term memory.

Args:
    memory_strategy: Strategy for memory management
    model: LLM model to use
    **memory_kwargs: Arguments for memory initialization

Parameters: – memory_strategy: Memory store for conversation history – model: Language model instance

`chat(self, user_message)`

Lines 341-372

Purpose:

Chat with the agent.

Args:
    user_message: User's message

Returns:
    Agent's response

Parameters: – user_message: User message

Returns: str

`get_conversation_summary(self)`

Lines 374-394

Purpose:

Get summary of current conversation.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D165C50>

`clear_memory(self)`

Lines 396-398

Purpose:

Clear conversation memory.

Parameters:

Functions

`demonstrate_short_term_memory()`

Lines 401-549

Purpose:

Demonstrate the Short-Term Memory pattern.

Key Concepts Summary

Pattern Overview

The Short Term Memory pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

Agent Class: Main implementation of the pattern logic
Prompt Engineering: Carefully crafted prompts to guide LLM behavior
Tool Integration: Optional tools for extended functionality
Memory Management: State handling across interactions
Output Parsing: Structured response processing

When to Use

✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability

❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)

Best Practices Demonstrated

1. Environment Configuration

✅ Use .env files for API keys
✅ Never hardcode credentials
✅ Validate environment variables on startup

2. Error Handling

✅ Graceful degradation on failures
✅ Informative error messages
✅ Logging for debugging

3. Prompt Engineering

✅ Clear, specific instructions
✅ Structured output formats
✅ Few-shot examples when beneficial

4. Code Organization

✅ Separation of concerns (class-based design)
✅ Type hints for clarity
✅ Docstrings for documentation

5. Performance Optimization

✅ Appropriate temperature settings
✅ Token budget management
✅ Caching where applicable

LangChain Components Used

ChatOpenAI: OpenAI model interface
Chains: Sequential component composition

Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development

Execution Flow

1. Initialize agent with configuration
   ↓
2. Load environment and dependencies
   ↓
3. Prepare prompt with input data
   ↓
4. Invoke LLM through LangChain
   ↓
5. Parse and validate response
   ↓
6. Return formatted result

Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation

Performance Considerations

Computational Cost

LLM Calls: Primary cost factor (based on tokens)
Latency: Depends on model and input size
Memory: Minimal for most patterns

Optimization Strategies

Prompt Optimization: Minimize tokens while maintaining clarity
Caching: Cache repeated queries/responses
Model Selection: Use smaller models when appropriate
Batch Processing: Group similar requests
Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

GPT-3.5-turbo: ~$0.002 per request
GPT-4: ~$0.03 per request
GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

Further Improvements

Production Enhancements

Monitoring: Add logging, metrics, tracing
Validation: Input/output schema validation
Rate Limiting: Implement request throttling
Caching: Redis or in-memory cache for responses
Async Support: Non-blocking operations for scalability

Feature Extensions

Multi-Model Support: Switch between providers
Fallback Models: Graceful degradation strategy
Custom Tools: Domain-specific function calling
Advanced Memory: Vector stores for long-term memory
Evaluation: Automated testing and benchmarking

Code Quality

Testing: Unit tests, integration tests
Documentation: API docs, usage examples
Type Safety: Stricter type hints
Configuration: Externalized config files

Related Patterns

Pattern 025
Pattern 027

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

Summary

The Short Term Memory pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns

🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability

📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/

File: 026_short_term_memory.py
Total Lines: 554
Pattern Category: AI Agent Design Pattern #026

Generated from actual implementation code
For educational and reference purposes

Agentic Design Pattern: Short Term Memory

Agentic Design Pattern: Short Term Memory

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

Overview

Pattern Description

File Statistics

Module Documentation

Import Statements

Standard Library Imports

LangChain Imports

Other Imports

Classes

Class: MemoryStrategy

Class: Message

to_langchain_message(self)

Class: ConversationContext

add_message(self, role, content, metadata)

get_recent_messages(self, n)

get_messages_by_role(self, role)

clear(self)

Class: BufferMemory

__init__(self, max_messages)

add_interaction(self, human_message, ai_message)

get_context_string(self)

get_langchain_messages(self)

clear(self)

Class: WindowMemory

__init__(self, window_size)

add_interaction(self, human_message, ai_message)

get_context_string(self)

get_langchain_messages(self)

Class: TokenLimitMemory

__init__(self, max_tokens)

_estimate_tokens(self, text)

add_interaction(self, human_message, ai_message)

_get_total_tokens(self)

get_context_string(self)

get_token_count(self)

Class: SummaryMemory

__init__(self, recent_messages, model)

add_interaction(self, human_message, ai_message)

_create_summary(self, messages)

get_context_string(self)

Class: ShortTermMemoryAgent

__init__(self, memory_strategy, model)

chat(self, user_message)

get_conversation_summary(self)

clear_memory(self)

Functions

demonstrate_short_term_memory()

Key Concepts Summary

Pattern Overview

Core Components

When to Use

Best Practices Demonstrated

1. Environment Configuration

2. Error Handling

3. Prompt Engineering

4. Code Organization

5. Performance Optimization

LangChain Components Used

Execution Flow

Performance Considerations

Computational Cost

Optimization Strategies

Cost Estimates (Approximate)

Common Pitfalls to Avoid

1. Prompt Design Issues

2. Error Handling

3. Token Management

4. Security

5. Testing

6. Cost Management

Further Improvements

Production Enhancements

Feature Extensions

Code Quality

Related Patterns

Summary

Comments

Class: `MemoryStrategy`

Class: `Message`

`to_langchain_message(self)`

Class: `ConversationContext`

`add_message(self, role, content, metadata)`

`get_recent_messages(self, n)`

`get_messages_by_role(self, role)`

`clear(self)`

Class: `BufferMemory`

`init(self, max_messages)`

`add_interaction(self, human_message, ai_message)`

`get_context_string(self)`

`get_langchain_messages(self)`

`clear(self)`

Class: `WindowMemory`

`init(self, window_size)`

`add_interaction(self, human_message, ai_message)`

`get_context_string(self)`

`get_langchain_messages(self)`

Class: `TokenLimitMemory`

`init(self, max_tokens)`

`_estimate_tokens(self, text)`

`add_interaction(self, human_message, ai_message)`

`_get_total_tokens(self)`

`get_context_string(self)`

`get_token_count(self)`

Class: `SummaryMemory`

`init(self, recent_messages, model)`

`add_interaction(self, human_message, ai_message)`

`_create_summary(self, messages)`

`get_context_string(self)`

Class: `ShortTermMemoryAgent`

`init(self, memory_strategy, model)`

`chat(self, user_message)`

`get_conversation_summary(self)`

`clear_memory(self)`

`demonstrate_short_term_memory()`