Tag: ai

  • Agentic Design Pattern: Active Learning

    
        """
    Pattern 034: Active Learning
    
    Description:
        Active Learning enables agents to strategically request human input for the most
        informative or uncertain cases. Rather than passively learning from all data,
        the agent actively selects examples that will maximize learning efficiency,
        focusing human effort on cases where it matters most.
    
    Components:
        - Uncertainty Estimator: Measures confidence in predictions
        - Query Strategy: Selects most informative examples
        - Oracle: Human expert providing labels/feedback
        - Learning Loop: Incorporates feedback to improve
        - Sample Selector: Chooses examples for annotation
    
    Use Cases:
        - Training data collection with limited budget
        - Model improvement in production
        - Handling ambiguous or edge cases
        - Domain adaptation
        - Quality assurance and verification
    
    LangChain Implementation:
        Uses confidence estimation and strategic sampling to identify cases where
        human feedback will be most valuable, optimizing the learning process.
    """
    
    import os
    from typing import List, Dict, Any, Optional, Tuple
    from dataclasses import dataclass, field
    from datetime import datetime
    from enum import Enum
    import random
    from collections import defaultdict
    from dotenv import load_dotenv
    
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser
    from langchain_openai import ChatOpenAI
    
    load_dotenv()
    
    
    class QueryStrategy(Enum):
        """Strategies for selecting samples for human annotation."""
        UNCERTAINTY_SAMPLING = "uncertainty_sampling"  # Least confident examples
        MARGIN_SAMPLING = "margin_sampling"  # Smallest margin between top predictions
        ENTROPY_SAMPLING = "entropy_sampling"  # Highest entropy
        QUERY_BY_COMMITTEE = "query_by_committee"  # Maximum disagreement
        DIVERSITY_SAMPLING = "diversity_sampling"  # Representative diversity
        EXPECTED_ERROR_REDUCTION = "expected_error_reduction"  # Max expected improvement
    
    
    @dataclass
    class Prediction:
        """A prediction with confidence information."""
        input: str
        output: str
        confidence: float
        alternatives: List[Tuple[str, float]] = field(default_factory=list)  # (output, confidence)
        metadata: Dict[str, Any] = field(default_factory=dict)
    
    
    @dataclass
    class AnnotationRequest:
        """Request for human annotation."""
        id: str
        input: str
        predicted_output: str
        confidence: float
        query_strategy: QueryStrategy
        timestamp: datetime = field(default_factory=datetime.now)
        
        # Human response
        human_label: Optional[str] = None
        feedback: Optional[str] = None
        annotation_time: Optional[datetime] = None
    
    
    @dataclass
    class TrainingExample:
        """Labeled training example."""
        input: str
        output: str
        source: str  # "human", "synthetic", "model"
        confidence: float = 1.0
        created_at: datetime = field(default_factory=datetime.now)
    
    
    class Oracle:
        """
        Simulates human expert for annotation (in production, would be real human).
        """
        
        def __init__(self, auto_annotate: bool = True):
            self.auto_annotate = auto_annotate
            self.annotation_count = 0
        
        def annotate(self, request: AnnotationRequest) -> AnnotationRequest:
            """
            Provide annotation for the request.
            
            In simulation mode, provides realistic labels.
            In production, would wait for human annotation.
            """
            self.annotation_count += 1
            
            if self.auto_annotate:
                # Simulate human annotation
                # In practice, this would involve actual human judgment
                request.human_label = f"[HUMAN ANNOTATED] {request.predicted_output}"
                request.feedback = "Annotation provided"
                request.annotation_time = datetime.now()
            
            return request
    
    
    class ActiveLearningAgent:
        """
        Agent that uses active learning to improve through strategic human feedback.
        
        Features:
        - Multiple query strategies
        - Confidence estimation
        - Strategic sample selection
        - Learning from feedback
        """
        
        def __init__(
            self,
            query_strategy: QueryStrategy = QueryStrategy.UNCERTAINTY_SAMPLING,
            oracle: Optional[Oracle] = None,
            temperature: float = 0.7
        ):
            self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=temperature)
            self.query_strategy = query_strategy
            self.oracle = oracle or Oracle()
            
            # Training data
            self.training_examples: List[TrainingExample] = []
            self.unlabeled_pool: List[str] = []
            self.annotation_history: List[AnnotationRequest] = []
            
            self._next_request_id = 1
            
            # Prompt for making predictions with confidence
            self.prediction_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are an AI assistant that provides predictions with confidence scores.
    
    Previous examples (for reference):
    {examples}
    
    Task: {input}
    
    Provide your answer and estimate your confidence (0.0-1.0).
    
    Format:
    ANSWER: [your answer]
    CONFIDENCE: [0.0-1.0]
    ALTERNATIVES: [alternative answer 1] (confidence), [alternative answer 2] (confidence)"""),
                ("user", "{input}")
            ])
        
        def predict_with_confidence(self, input_text: str) -> Prediction:
            """
            Make a prediction with confidence estimation.
            """
            # Format examples from training data
            examples_text = self._format_examples(self.training_examples[:5])
            
            # Generate prediction
            chain = self.prediction_prompt | self.llm | StrOutputParser()
            result = chain.invoke({
                "input": input_text,
                "examples": examples_text
            })
            
            # Parse result
            answer, confidence, alternatives = self._parse_prediction(result)
            
            return Prediction(
                input=input_text,
                output=answer,
                confidence=confidence,
                alternatives=alternatives
            )
        
        def select_for_annotation(
            self,
            predictions: List[Prediction],
            budget: int = 5
        ) -> List[Prediction]:
            """
            Select most informative examples for human annotation.
            
            Args:
                predictions: List of predictions to choose from
                budget: Number of examples to select
            
            Returns:
                Selected predictions for annotation
            """
            if not predictions:
                return []
            
            # Apply query strategy
            if self.query_strategy == QueryStrategy.UNCERTAINTY_SAMPLING:
                # Select examples with lowest confidence
                selected = sorted(predictions, key=lambda p: p.confidence)[:budget]
            
            elif self.query_strategy == QueryStrategy.MARGIN_SAMPLING:
                # Select examples with smallest margin between top 2 predictions
                selected = []
                for pred in predictions:
                    if len(pred.alternatives) >= 2:
                        margin = abs(pred.alternatives[0][1] - pred.alternatives[1][1])
                        selected.append((margin, pred))
                    else:
                        selected.append((0.0, pred))
                
                selected = [p for _, p in sorted(selected, key=lambda x: x[0])[:budget]]
            
            elif self.query_strategy == QueryStrategy.DIVERSITY_SAMPLING:
                # Select diverse examples (simple: random sampling)
                selected = random.sample(predictions, min(budget, len(predictions)))
            
            else:
                # Default: uncertainty sampling
                selected = sorted(predictions, key=lambda p: p.confidence)[:budget]
            
            return selected
        
        def request_annotations(
            self,
            predictions: List[Prediction]
        ) -> List[AnnotationRequest]:
            """
            Request human annotations for selected predictions.
            """
            requests = []
            
            for pred in predictions:
                request = AnnotationRequest(
                    id=f"ann_{self._next_request_id:04d}",
                    input=pred.input,
                    predicted_output=pred.output,
                    confidence=pred.confidence,
                    query_strategy=self.query_strategy
                )
                self._next_request_id += 1
                
                # Get annotation from oracle
                request = self.oracle.annotate(request)
                
                requests.append(request)
                self.annotation_history.append(request)
            
            return requests
        
        def incorporate_feedback(
            self,
            annotations: List[AnnotationRequest]
        ) -> int:
            """
            Incorporate human annotations into training data.
            
            Returns number of examples added.
            """
            added = 0
            
            for annotation in annotations:
                if annotation.human_label:
                    example = TrainingExample(
                        input=annotation.input,
                        output=annotation.human_label,
                        source="human",
                        confidence=1.0
                    )
                    self.training_examples.append(example)
                    added += 1
            
            return added
        
        def active_learning_cycle(
            self,
            unlabeled_data: List[str],
            annotation_budget: int = 5
        ) -> Dict[str, Any]:
            """
            Perform one cycle of active learning.
            
            Steps:
            1. Make predictions on unlabeled data
            2. Select most informative examples
            3. Request human annotations
            4. Incorporate feedback into training data
            
            Returns:
                Statistics about the cycle
            """
            # Step 1: Predict on unlabeled data
            predictions = []
            for input_text in unlabeled_data:
                pred = self.predict_with_confidence(input_text)
                predictions.append(pred)
            
            # Step 2: Select for annotation
            selected = self.select_for_annotation(predictions, annotation_budget)
            
            # Step 3: Request annotations
            annotations = self.request_annotations(selected)
            
            # Step 4: Incorporate feedback
            added = self.incorporate_feedback(annotations)
            
            return {
                "cycle_complete": True,
                "predictions_made": len(predictions),
                "selected_for_annotation": len(selected),
                "annotations_received": len(annotations),
                "examples_added": added,
                "total_training_examples": len(self.training_examples),
                "avg_confidence": sum(p.confidence for p in predictions) / len(predictions) if predictions else 0
            }
        
        def _format_examples(self, examples: List[TrainingExample]) -> str:
            """Format training examples for prompt."""
            if not examples:
                return "No examples yet."
            
            lines = []
            for i, ex in enumerate(examples, 1):
                lines.append(f"Example {i}:")
                lines.append(f"  Input: {ex.input}")
                lines.append(f"  Output: {ex.output}")
            
            return "\n".join(lines)
        
        def _parse_prediction(self, result: str) -> Tuple[str, float, List[Tuple[str, float]]]:
            """Parse prediction result."""
            lines = result.strip().split('\n')
            
            answer = ""
            confidence = 0.5
            alternatives = []
            
            for line in lines:
                if line.startswith("ANSWER:"):
                    answer = line.replace("ANSWER:", "").strip()
                elif line.startswith("CONFIDENCE:"):
                    try:
                        conf_str = line.replace("CONFIDENCE:", "").strip()
                        confidence = float(conf_str)
                    except ValueError:
                        pass
                elif line.startswith("ALTERNATIVES:"):
                    # Parse alternatives (simple parsing)
                    alt_str = line.replace("ALTERNATIVES:", "").strip()
                    # This is simplified - in practice, would need better parsing
            
            if not answer:
                answer = result
            
            return answer, confidence, alternatives
        
        def get_statistics(self) -> Dict[str, Any]:
            """Get active learning statistics."""
            if not self.annotation_history:
                return {
                    "total_annotations": 0,
                    "training_examples": len(self.training_examples),
                    "avg_confidence_at_annotation": 0.0
                }
            
            avg_conf = sum(a.confidence for a in self.annotation_history) / len(self.annotation_history)
            
            return {
                "total_annotations": len(self.annotation_history),
                "training_examples": len(self.training_examples),
                "avg_confidence_at_annotation": avg_conf,
                "oracle_calls": self.oracle.annotation_count,
                "query_strategy": self.query_strategy.value
            }
    
    
    def demonstrate_active_learning():
        """
        Demonstrates active learning with strategic sample selection
        and human feedback integration.
        """
        print("=" * 80)
        print("ACTIVE LEARNING DEMONSTRATION")
        print("=" * 80)
        
        # Create oracle (simulated human expert)
        oracle = Oracle(auto_annotate=True)
        
        # Test 1: Uncertainty Sampling
        print("\n" + "=" * 80)
        print("Test 1: Uncertainty Sampling Strategy")
        print("=" * 80)
        
        agent1 = ActiveLearningAgent(
            query_strategy=QueryStrategy.UNCERTAINTY_SAMPLING,
            oracle=oracle
        )
        
        # Simulated unlabeled data
        unlabeled_data = [
            "What is machine learning?",
            "Explain neural networks",
            "How does backpropagation work?",
            "What is deep learning?",
            "Describe convolutional neural networks",
            "What is transfer learning?",
            "Explain reinforcement learning",
            "What are transformers in AI?"
        ]
        
        print(f"\nUnlabeled data pool: {len(unlabeled_data)} examples")
        print("\nStarting active learning cycle...")
        
        result1 = agent1.active_learning_cycle(unlabeled_data, annotation_budget=3)
        
        print(f"\nCycle Results:")
        print(f"  Predictions made: {result1['predictions_made']}")
        print(f"  Selected for annotation: {result1['selected_for_annotation']}")
        print(f"  Annotations received: {result1['annotations_received']}")
        print(f"  Examples added to training: {result1['examples_added']}")
        print(f"  Total training examples: {result1['total_training_examples']}")
        print(f"  Average confidence: {result1['avg_confidence']:.2f}")
        
        # Show which examples were selected
        print(f"\nSelected examples (lowest confidence):")
        for i, annotation in enumerate(agent1.annotation_history[:3], 1):
            print(f"  {i}. \"{annotation.input}\" (confidence: {annotation.confidence:.2f})")
        
        # Test 2: Multiple cycles
        print("\n" + "=" * 80)
        print("Test 2: Multiple Active Learning Cycles")
        print("=" * 80)
        
        agent2 = ActiveLearningAgent(
            query_strategy=QueryStrategy.UNCERTAINTY_SAMPLING,
            oracle=oracle
        )
        
        # Run 3 cycles
        for cycle in range(1, 4):
            print(f"\n--- Cycle {cycle} ---")
            result = agent2.active_learning_cycle(unlabeled_data, annotation_budget=2)
            print(f"Training examples: {result['total_training_examples']}")
            print(f"Average confidence: {result['avg_confidence']:.2f}")
        
        # Test 3: Compare strategies
        print("\n" + "=" * 80)
        print("Test 3: Comparing Query Strategies")
        print("=" * 80)
        
        strategies = [
            QueryStrategy.UNCERTAINTY_SAMPLING,
            QueryStrategy.MARGIN_SAMPLING,
            QueryStrategy.DIVERSITY_SAMPLING
        ]
        
        for strategy in strategies:
            agent = ActiveLearningAgent(
                query_strategy=strategy,
                oracle=Oracle(auto_annotate=True)
            )
            
            result = agent.active_learning_cycle(unlabeled_data[:5], annotation_budget=2)
            
            print(f"\n{strategy.value}:")
            print(f"  Examples added: {result['examples_added']}")
            print(f"  Avg confidence: {result['avg_confidence']:.2f}")
        
        # Show final statistics
        print("\n" + "=" * 80)
        print("Final Statistics")
        print("=" * 80)
        
        stats = agent1.get_statistics()
        print(f"\nAgent 1 (Uncertainty Sampling):")
        print(f"  Total annotations: {stats['total_annotations']}")
        print(f"  Training examples: {stats['training_examples']}")
        print(f"  Avg confidence at annotation: {stats['avg_confidence_at_annotation']:.2f}")
        print(f"  Oracle calls: {stats['oracle_calls']}")
        
        # Summary
        print("\n" + "=" * 80)
        print("SUMMARY")
        print("=" * 80)
        print("""
    Active Learning provides:
    ✓ Strategic sample selection for annotation
    ✓ Confidence-based uncertainty estimation
    ✓ Multiple query strategies
    ✓ Efficient use of human expertise
    ✓ Continuous improvement through feedback
    ✓ Cost-effective training data collection
    
    This pattern excels at:
    - Maximizing learning from limited annotation budget
    - Focusing human effort on most informative examples
    - Handling ambiguous or edge cases
    - Improving model performance efficiently
    - Domain adaptation with minimal labels
    
    Query strategies:
    1. UNCERTAINTY_SAMPLING: Select least confident predictions
    2. MARGIN_SAMPLING: Smallest margin between top predictions
    3. ENTROPY_SAMPLING: Highest prediction entropy
    4. QUERY_BY_COMMITTEE: Maximum disagreement among models
    5. DIVERSITY_SAMPLING: Representative sample diversity
    6. EXPECTED_ERROR_REDUCTION: Maximum expected improvement
    
    Active learning cycle:
    1. Make predictions on unlabeled data
    2. Estimate confidence for each prediction
    3. Select most informative examples using query strategy
    4. Request human annotations for selected examples
    5. Incorporate feedback into training data
    6. Repeat until budget exhausted or performance satisfactory
    
    Benefits:
    - Efficiency: Fewer labels needed for same performance
    - Focus: Human effort on valuable examples
    - Adaptability: Continuously improves with feedback
    - Cost-effective: Reduces annotation costs
    - Quality: Better than random sampling
    
    Use active learning when you need:
    - Limited annotation budget
    - Efficient model training
    - Production model improvement
    - Domain adaptation
    - Edge case handling
    - Cost-effective data collection
    """)
    
    
    if __name__ == "__main__":
        demonstrate_active_learning()
    
        

    Agentic Design Pattern: Active Learning

    Pattern 034: Active Learning – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Active Learning pattern implementation.


    Pattern Description

    Pattern 034: Active Learning

    Description: Active Learning enables agents to strategically request human input for the most informative or uncertain cases. Rather than passively learning from all data, the agent actively selects examples that will maximize learning efficiency, focusing human effort on cases where it matters most.

    Components: – Uncertainty Estimator: Measures confidence in predictions – Query Strategy: Selects most informative examples – Oracle: Human expert providing labels/feedback – Learning Loop: Incorporates feedback to improve – Sample Selector: Chooses examples for annotation

    Use Cases: – Training data collection with limited budget – Model improvement in production – Handling ambiguous or edge cases – Domain adaptation – Quality assurance and verification

    LangChain Implementation: Uses confidence estimation and strategic sampling to identify cases where human feedback will be most valuable, optimizing the learning process.


    File Statistics

    • Total Lines: 546
    • Classes: 6
    • Functions: 1
    • Constants: 0

    Module Documentation

    """
    Pattern 034: Active Learning
    
    Description:
        Active Learning enables agents to strategically request human input for the most
        informative or uncertain cases. Rather than passively learning from all data,
        the agent actively selects examples that will maximize learning efficiency,
        focusing human effort on cases where it matters most.
    
    Components:
        - Uncertainty Estimator: Measures confidence in predictions
        - Query Strategy: Selects most informative examples
        - Oracle: Human expert providing labels/feedback
        - Learning Loop: Incorporates feedback to improve
        - Sample Selector: Chooses examples for annotation
    
    Use Cases:
        - Training data collection with limited budget
        - Model improvement in production
        - Handling ambiguous or edge cases
        - Domain adaptation
        - Quality assurance and verification
    
    LangChain Implementation:
        Uses confidence estimation and strategic sampling to identify cases where
        human feedback will be most valuable, optimizing the learning process.
    """
    

    Purpose: This module implements the Active Learning pattern, providing a structured approach to agent design and implementation.


    Import Statements

    Standard Library Imports

    import collections.defaultdict
    import dataclasses.dataclass
    import dataclasses.field
    import datetime.datetime
    import enum.Enum
    import langchain_core.output_parsers.StrOutputParser
    import langchain_core.prompts.ChatPromptTemplate
    import os
    import random
    import typing.Any
    import typing.Dict
    import typing.List
    import typing.Optional
    import typing.Tuple
    

    Purpose: Core Python functionality: – collections.defaultdict: Specialized container datatypes – 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.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – random: Random number generation – 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) – typing.Tuple: 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: QueryStrategy

    Lines 45-52

    Inherits from: Enum

    Purpose:

    Strategies for selecting samples for human annotation.
    

    Class: Prediction

    Lines 56-62

    Purpose:

    A prediction with confidence information.
    

    Class: AnnotationRequest

    Lines 66-78

    Purpose:

    Request for human annotation.
    

    Class: TrainingExample

    Lines 82-88

    Purpose:

    Labeled training example.
    

    Class: Oracle

    Lines 91-116

    Purpose:

    Simulates human expert for annotation (in production, would be real human).
    

    Methods:

    __init__(self, auto_annotate)

    Lines 96-98

    Purpose: Initialize the class instance with required parameters.

    Parameters: – auto_annotate: Auto annotate

    annotate(self, request)

    Lines 100-116

    Purpose:

    Provide annotation for the request.
    
    In simulation mode, provides realistic labels.
    In production, would wait for human annotation.
    

    Parameters: – request: Request

    Returns: AnnotationRequest


    Class: ActiveLearningAgent

    Lines 119-384

    Purpose:

    Agent that uses active learning to improve through strategic human feedback.
    
    Features:
    - Multiple query strategies
    - Confidence estimation
    - Strategic sample selection
    - Learning from feedback
    

    Methods:

    __init__(self, query_strategy, oracle, temperature)

    Lines 130-163

    Purpose: Initialize the class instance with required parameters.

    Parameters: – query_strategy: Input query or question – oracle: Oracle – temperature: Sampling temperature (0-1)

    predict_with_confidence(self, input_text)

    Lines 165-187

    Purpose:

    Make a prediction with confidence estimation.
    

    Parameters: – input_text: Input data

    Returns: Prediction

    select_for_annotation(self, predictions, budget)

    Lines 189-232

    Purpose:

    Select most informative examples for human annotation.
    
    Args:
        predictions: List of predictions to choose from
        budget: Number of examples to select
    
    Returns:
        Selected predictions for annotation
    

    Parameters: – predictions: Predictions – budget: Budget

    Returns: <ast.Subscript object at 0x000001A06D179A90>

    request_annotations(self, predictions)

    Lines 234-259

    Purpose:

    Request human annotations for selected predictions.
    

    Parameters: – predictions: Predictions

    Returns: <ast.Subscript object at 0x000001A06D75B090>

    incorporate_feedback(self, annotations)

    Lines 261-283

    Purpose:

    Incorporate human annotations into training data.
    
    Returns number of examples added.
    

    Parameters: – annotations: Annotations

    Returns: int

    active_learning_cycle(self, unlabeled_data, annotation_budget)

    Lines 285-325

    Purpose:

    Perform one cycle of active learning.
    
    Steps:
    1. Make predictions on unlabeled data
    2. Select most informative examples
    3. Request human annotations
    4. Incorporate feedback into training data
    
    Returns:
        Statistics about the cycle
    

    Parameters: – unlabeled_data: Unlabeled data – annotation_budget: Annotation budget

    Returns: <ast.Subscript object at 0x000001A06D6AA490>

    _format_examples(self, examples)

    Lines 327-338

    Purpose:

    Format training examples for prompt.
    

    Parameters: – examples: Examples

    Returns: str

    _parse_prediction(self, result)

    Lines 340-365

    Purpose:

    Parse prediction result.
    

    Parameters: – result: Result

    Returns: <ast.Subscript object at 0x000001A06D681C10>

    get_statistics(self)

    Lines 367-384

    Purpose:

    Get active learning statistics.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D625450>


    Functions

    demonstrate_active_learning()

    Lines 387-541

    Purpose:

    Demonstrates active learning with strategic sample selection
    and human feedback integration.
    

    Key Concepts Summary

    Pattern Overview

    The Active Learning 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

    1. Agent Class: Main implementation of the pattern logic
    2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
    3. Tool Integration: Optional tools for extended functionality
    4. Memory Management: State handling across interactions
    5. 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

    1. Prompt Optimization: Minimize tokens while maintaining clarity
    2. Caching: Cache repeated queries/responses
    3. Model Selection: Use smaller models when appropriate
    4. Batch Processing: Group similar requests
    5. 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

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

    Feature Extensions

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

    Code Quality

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

    Related Patterns

    • Pattern 033
    • Pattern 035

    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 Active Learning 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: 034_active_learning.py
    Total Lines: 546
    Pattern Category: AI Agent Design Pattern #034


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Human In The Loop

    
        """
    Pattern 033: Human-in-the-Loop (HITL)
    
    Description:
        Human-in-the-Loop integrates human oversight and intervention at critical
        decision points in agent workflows. This pattern enables humans to approve,
        correct, or guide agent actions, ensuring safety, quality, and alignment
        with human values in high-stakes or ambiguous situations.
    
    Components:
        - Intervention Points: Designated moments for human input
        - Approval Gates: Require human approval before proceeding
        - Correction Mechanism: Allow humans to fix agent errors
        - Guidance System: Humans provide direction when agent is uncertain
        - Escalation Logic: Automatic escalation on confidence thresholds
        - Feedback Integration: Learn from human corrections
    
    Use Cases:
        - High-stakes decision making (medical, financial, legal)
        - Content moderation and review
        - Quality assurance for generated content
        - Training data collection and refinement
        - Compliance and regulatory requirements
        - Handling edge cases and ambiguity
    
    LangChain Implementation:
        Uses callback handlers and approval mechanisms to integrate human
        oversight into agent workflows, with configurable intervention points
        and escalation criteria.
    """
    
    import os
    from typing import List, Dict, Any, Optional, Callable
    from dataclasses import dataclass, field
    from datetime import datetime
    from enum import Enum
    from dotenv import load_dotenv
    
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser
    from langchain_openai import ChatOpenAI
    
    load_dotenv()
    
    
    class InterventionType(Enum):
        """Types of human intervention."""
        APPROVAL = "approval"  # Human must approve action
        CORRECTION = "correction"  # Human can correct output
        GUIDANCE = "guidance"  # Human provides direction
        REVIEW = "review"  # Human reviews but doesn't block
        ESCALATION = "escalation"  # Automatic escalation to human
    
    
    class InterventionStatus(Enum):
        """Status of intervention request."""
        PENDING = "pending"
        APPROVED = "approved"
        REJECTED = "rejected"
        CORRECTED = "corrected"
        SKIPPED = "skipped"
    
    
    @dataclass
    class InterventionRequest:
        """Request for human intervention."""
        id: str
        type: InterventionType
        context: str
        proposed_action: str
        confidence: float
        timestamp: datetime = field(default_factory=datetime.now)
        metadata: Dict[str, Any] = field(default_factory=dict)
        
        # Response
        status: InterventionStatus = InterventionStatus.PENDING
        human_response: Optional[str] = None
        correction: Optional[str] = None
        response_time: Optional[datetime] = None
    
    
    @dataclass
    class EscalationCriteria:
        """Criteria for automatic escalation to human."""
        min_confidence: float = 0.7  # Below this, escalate
        sensitive_keywords: List[str] = field(default_factory=list)
        high_stakes_actions: List[str] = field(default_factory=list)
        require_approval_for: List[str] = field(default_factory=list)
    
    
    class HumanInterface:
        """
        Interface for human interaction (simulated or real).
        
        In production, this would connect to a UI, API, or messaging system.
        """
        
        def __init__(self, auto_approve: bool = False):
            self.auto_approve = auto_approve
            self.pending_requests: List[InterventionRequest] = []
        
        def request_intervention(
            self,
            request: InterventionRequest
        ) -> InterventionRequest:
            """
            Request human intervention.
            
            In simulation mode, auto-responds. In production, would wait for human.
            """
            self.pending_requests.append(request)
            
            if self.auto_approve:
                # Simulate human response
                if request.type == InterventionType.APPROVAL:
                    request.status = InterventionStatus.APPROVED
                    request.human_response = "Approved by human"
                elif request.type == InterventionType.CORRECTION:
                    # Simulate a correction
                    request.status = InterventionStatus.CORRECTED
                    request.correction = f"Corrected version: {request.proposed_action} [HUMAN EDITED]"
                elif request.type == InterventionType.GUIDANCE:
                    request.status = InterventionStatus.APPROVED
                    request.human_response = "Proceed with suggested approach"
                else:
                    request.status = InterventionStatus.APPROVED
                
                request.response_time = datetime.now()
            
            return request
        
        def get_pending_count(self) -> int:
            """Get number of pending intervention requests."""
            return sum(1 for r in self.pending_requests if r.status == InterventionStatus.PENDING)
        
        def get_all_requests(self) -> List[InterventionRequest]:
            """Get all intervention requests."""
            return self.pending_requests.copy()
    
    
    class HITLAgent:
        """
        Agent with Human-in-the-Loop integration.
        
        Features:
        - Configurable intervention points
        - Automatic escalation based on criteria
        - Approval gates for sensitive actions
        - Learning from human feedback
        """
        
        def __init__(
            self,
            human_interface: HumanInterface,
            escalation_criteria: Optional[EscalationCriteria] = None,
            temperature: float = 0.7
        ):
            self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=temperature)
            self.human_interface = human_interface
            self.escalation_criteria = escalation_criteria or EscalationCriteria()
            
            self._next_request_id = 1
            self.intervention_history: List[InterventionRequest] = []
            
            # Prompt for generating response
            self.response_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are an AI assistant that works with human oversight.
    
    Task: {task}
    
    Generate a response and provide a confidence score (0.0-1.0) for your answer.
    
    Respond in this format:
    CONFIDENCE: [score]
    RESPONSE: [your response]"""),
                ("user", "{task}")
            ])
            
            # Prompt for confidence assessment
            self.confidence_prompt = ChatPromptTemplate.from_messages([
                ("system", """Assess your confidence in this response on a scale of 0.0 to 1.0.
    
    Task: {task}
    Response: {response}
    
    Consider:
    - Certainty of facts
    - Complexity of task
    - Potential for error
    - Stakes involved
    
    Respond with just a number between 0.0 and 1.0."""),
                ("user", "Assess confidence")
            ])
        
        def process_with_oversight(
            self,
            task: str,
            require_approval: bool = False
        ) -> Dict[str, Any]:
            """
            Process task with human oversight.
            
            Args:
                task: Task to process
                require_approval: Force human approval regardless of confidence
            
            Returns:
                Dictionary with response and intervention details
            """
            # Generate initial response
            chain = self.response_prompt | self.llm | StrOutputParser()
            result = chain.invoke({"task": task})
            
            # Parse response
            response, confidence = self._parse_response(result)
            
            # Check if escalation is needed
            should_escalate = (
                require_approval or
                self._should_escalate(task, response, confidence)
            )
            
            if should_escalate:
                # Request human intervention
                intervention_type = (
                    InterventionType.APPROVAL if require_approval or confidence  tuple[str, float]:
            """Parse response and confidence from LLM output."""
            lines = result.strip().split('\n')
            confidence = 0.5  # Default
            response = result
            
            for line in lines:
                if line.startswith("CONFIDENCE:"):
                    try:
                        conf_str = line.replace("CONFIDENCE:", "").strip()
                        confidence = float(conf_str)
                    except ValueError:
                        pass
                elif line.startswith("RESPONSE:"):
                    response = line.replace("RESPONSE:", "").strip()
            
            # If format wasn't followed, estimate confidence
            if confidence == 0.5 and "CONFIDENCE:" not in result:
                confidence = self._estimate_confidence(result)
            
            return response, confidence
        
        def _estimate_confidence(self, response: str) -> float:
            """Estimate confidence from response content."""
            # Simple heuristic based on uncertainty phrases
            uncertainty_phrases = [
                "i think", "maybe", "possibly", "uncertain", "not sure",
                "might", "could be", "perhaps", "probably"
            ]
            
            response_lower = response.lower()
            uncertainty_count = sum(1 for phrase in uncertainty_phrases if phrase in response_lower)
            
            # More uncertainty phrases = lower confidence
            confidence = max(0.3, 1.0 - (uncertainty_count * 0.15))
            
            return confidence
        
        def _should_escalate(self, task: str, response: str, confidence: float) -> bool:
            """Determine if task should be escalated to human."""
            # Low confidence
            if confidence  Dict[str, Any]:
            """
            Request human correction for a specific issue.
            """
            request = InterventionRequest(
                id=f"req_{self._next_request_id:04d}",
                type=InterventionType.CORRECTION,
                context=f"Task: {task}\nIssue: {issue_description}",
                proposed_action=initial_response,
                confidence=0.0,  # Explicitly requesting correction
                metadata={"issue": issue_description}
            )
            self._next_request_id += 1
            
            # Get human correction
            request = self.human_interface.request_intervention(request)
            self.intervention_history.append(request)
            
            return {
                "original": initial_response,
                "corrected": request.correction or initial_response,
                "status": request.status.value
            }
        
        def get_statistics(self) -> Dict[str, Any]:
            """Get statistics about human interventions."""
            if not self.intervention_history:
                return {
                    "total_interventions": 0,
                    "by_type": {},
                    "by_status": {},
                    "avg_confidence": 0.0
                }
            
            by_type = {}
            by_status = {}
            
            for request in self.intervention_history:
                # Count by type
                type_key = request.type.value
                by_type[type_key] = by_type.get(type_key, 0) + 1
                
                # Count by status
                status_key = request.status.value
                by_status[status_key] = by_status.get(status_key, 0) + 1
            
            avg_confidence = sum(r.confidence for r in self.intervention_history) / len(self.intervention_history)
            
            return {
                "total_interventions": len(self.intervention_history),
                "by_type": by_type,
                "by_status": by_status,
                "avg_confidence": avg_confidence
            }
    
    
    def demonstrate_human_in_the_loop():
        """
        Demonstrates Human-in-the-Loop with various intervention scenarios.
        """
        print("=" * 80)
        print("HUMAN-IN-THE-LOOP DEMONSTRATION")
        print("=" * 80)
        
        # Create human interface (simulated)
        human_interface = HumanInterface(auto_approve=True)
        
        # Configure escalation criteria
        escalation_criteria = EscalationCriteria(
            min_confidence=0.7,
            sensitive_keywords=["medical", "financial", "legal", "delete", "confidential"],
            high_stakes_actions=["approve payment", "delete data", "grant access"]
        )
        
        # Create HITL agent
        agent = HITLAgent(
            human_interface=human_interface,
            escalation_criteria=escalation_criteria
        )
        
        # Test 1: High confidence task (no intervention)
        print("\n" + "=" * 80)
        print("Test 1: High Confidence Task (No Intervention Expected)")
        print("=" * 80)
        
        task1 = "What is the capital of France?"
        print(f"\nTask: {task1}")
        result1 = agent.process_with_oversight(task1)
        print(f"Response: {result1['response']}")
        print(f"Confidence: {result1['confidence']:.2f}")
        print(f"HITL Used: {result1['used_hitl']}")
        print(f"HITL Result: {result1['hitl_result']}")
        
        # Test 2: Sensitive keyword (automatic escalation)
        print("\n" + "=" * 80)
        print("Test 2: Sensitive Content (Automatic Escalation)")
        print("=" * 80)
        
        task2 = "Should I approve this medical procedure for the patient?"
        print(f"\nTask: {task2}")
        result2 = agent.process_with_oversight(task2)
        print(f"Response: {result2['response']}")
        print(f"Confidence: {result2['confidence']:.2f}")
        print(f"HITL Used: {result2['used_hitl']}")
        print(f"HITL Result: {result2['hitl_result']}")
        if result2['original_response']:
            print(f"Original Response: {result2['original_response'][:100]}...")
        
        # Test 3: Forced approval
        print("\n" + "=" * 80)
        print("Test 3: Forced Human Approval")
        print("=" * 80)
        
        task3 = "Delete all user data from the database"
        print(f"\nTask: {task3}")
        result3 = agent.process_with_oversight(task3, require_approval=True)
        print(f"Response: {result3['response']}")
        print(f"Confidence: {result3['confidence']:.2f}")
        print(f"HITL Used: {result3['used_hitl']}")
        print(f"HITL Result: {result3['hitl_result']}")
        
        # Test 4: Request correction
        print("\n" + "=" * 80)
        print("Test 4: Human Correction Request")
        print("=" * 80)
        
        task4 = "Calculate the total revenue for Q4 2024"
        initial_response = "The total revenue was $1.5M"
        issue = "The calculation seems incorrect based on our records"
        
        print(f"\nTask: {task4}")
        print(f"Initial Response: {initial_response}")
        print(f"Issue: {issue}")
        
        correction_result = agent.request_correction(task4, initial_response, issue)
        print(f"Corrected Response: {correction_result['corrected']}")
        print(f"Status: {correction_result['status']}")
        
        # Test 5: Low confidence (escalation)
        print("\n" + "=" * 80)
        print("Test 5: Low Confidence Response")
        print("=" * 80)
        
        task5 = "I'm not sure about this complex legal interpretation..."
        print(f"\nTask: {task5}")
        result5 = agent.process_with_oversight(task5)
        print(f"Response: {result5['response']}")
        print(f"Confidence: {result5['confidence']:.2f}")
        print(f"HITL Used: {result5['used_hitl']}")
        print(f"HITL Result: {result5['hitl_result']}")
        
        # Show statistics
        print("\n" + "=" * 80)
        print("Intervention Statistics")
        print("=" * 80)
        
        stats = agent.get_statistics()
        print(f"\nTotal Interventions: {stats['total_interventions']}")
        print(f"By Type: {stats['by_type']}")
        print(f"By Status: {stats['by_status']}")
        print(f"Average Confidence: {stats['avg_confidence']:.2f}")
        
        print(f"\nPending Requests: {human_interface.get_pending_count()}")
        
        # Summary
        print("\n" + "=" * 80)
        print("SUMMARY")
        print("=" * 80)
        print("""
    Human-in-the-Loop (HITL) provides:
    ✓ Human oversight at critical decision points
    ✓ Automatic escalation based on confidence
    ✓ Approval gates for sensitive actions
    ✓ Correction mechanisms for error fixing
    ✓ Guidance integration for ambiguous cases
    ✓ Audit trail of human interventions
    
    This pattern excels at:
    - High-stakes decision making (medical, financial, legal)
    - Quality assurance and content moderation
    - Handling edge cases and ambiguity
    - Regulatory compliance requirements
    - Training data collection
    - Building trust and accountability
    
    Intervention types:
    1. APPROVAL: Human must approve before proceeding
    2. CORRECTION: Human can fix errors or improve output
    3. GUIDANCE: Human provides direction when uncertain
    4. REVIEW: Human reviews but doesn't block execution
    5. ESCALATION: Automatic routing to human based on criteria
    
    Escalation triggers:
    - Low confidence scores (< 0.7 by default)
    - Sensitive keywords (medical, financial, legal, etc.)
    - High-stakes actions (delete data, approve payments, etc.)
    - Explicit requirement for human approval
    
    Benefits:
    - Safety: Prevents high-risk errors
    - Quality: Human expertise improves outputs
    - Trust: Builds confidence through oversight
    - Learning: Agent learns from corrections
    - Compliance: Meets regulatory requirements
    - Accountability: Clear audit trail
    
    Use HITL when you need:
    - Safety in high-stakes scenarios
    - Quality assurance for critical outputs
    - Regulatory compliance
    - Human expertise for edge cases
    - Trust and transparency
    - Gradual automation with human backup
    """)
    
    
    if __name__ == "__main__":
        demonstrate_human_in_the_loop()
    
        

    Agentic Design Pattern: Human In The Loop

    Pattern 033: Human In The Loop – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Human In The Loop pattern implementation.


    Pattern Description

    Pattern 033: Human-in-the-Loop (HITL)

    Description: Human-in-the-Loop integrates human oversight and intervention at critical decision points in agent workflows. This pattern enables humans to approve, correct, or guide agent actions, ensuring safety, quality, and alignment with human values in high-stakes or ambiguous situations.

    Components: – Intervention Points: Designated moments for human input – Approval Gates: Require human approval before proceeding – Correction Mechanism: Allow humans to fix agent errors – Guidance System: Humans provide direction when agent is uncertain – Escalation Logic: Automatic escalation on confidence thresholds – Feedback Integration: Learn from human corrections

    Use Cases: – High-stakes decision making (medical, financial, legal) – Content moderation and review – Quality assurance for generated content – Training data collection and refinement – Compliance and regulatory requirements – Handling edge cases and ambiguity

    LangChain Implementation: Uses callback handlers and approval mechanisms to integrate human oversight into agent workflows, with configurable intervention points and escalation criteria.


    File Statistics

    • Total Lines: 555
    • Classes: 6
    • Functions: 1
    • Constants: 0

    Module Documentation

    """
    Pattern 033: Human-in-the-Loop (HITL)
    
    Description:
        Human-in-the-Loop integrates human oversight and intervention at critical
        decision points in agent workflows. This pattern enables humans to approve,
        correct, or guide agent actions, ensuring safety, quality, and alignment
        with human values in high-stakes or ambiguous situations.
    
    Components:
        - Intervention Points: Designated moments for human input
        - Approval Gates: Require human approval before proceeding
        - Correction Mechanism: Allow humans to fix agent errors
        - Guidance System: Humans provide direction when agent is uncertain
        - Escalation Logic: Automatic escalation on confidence thresholds
        - Feedback Integration: Learn from human corrections
    
    Use Cases:
        - High-stakes decision making (medical, financial, legal)
        - Content moderation and review
        - Quality assurance for generated content
        - Training data collection and refinement
        - Compliance and regulatory requirements
        - Handling edge cases and ambiguity
    
    LangChain Implementation:
        Uses callback handlers and approval mechanisms to integrate human
        oversight into agent workflows, with configurable intervention points
        and escalation criteria.
    """
    

    Purpose: This module implements the Human In The Loop 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.output_parsers.StrOutputParser
    import langchain_core.prompts.ChatPromptTemplate
    import os
    import typing.Any
    import typing.Callable
    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.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Callable: 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: InterventionType

    Lines 46-52

    Inherits from: Enum

    Purpose:

    Types of human intervention.
    

    Class: InterventionStatus

    Lines 55-61

    Inherits from: Enum

    Purpose:

    Status of intervention request.
    

    Class: InterventionRequest

    Lines 65-79

    Purpose:

    Request for human intervention.
    

    Class: EscalationCriteria

    Lines 83-88

    Purpose:

    Criteria for automatic escalation to human.
    

    Class: HumanInterface

    Lines 91-138

    Purpose:

    Interface for human interaction (simulated or real).
    
    In production, this would connect to a UI, API, or messaging system.
    

    Methods:

    __init__(self, auto_approve)

    Lines 98-100

    Purpose: Initialize the class instance with required parameters.

    Parameters: – auto_approve: Auto approve

    request_intervention(self, request)

    Lines 102-130

    Purpose:

    Request human intervention.
    
    In simulation mode, auto-responds. In production, would wait for human.
    

    Parameters: – request: Request

    Returns: InterventionRequest

    get_pending_count(self)

    Lines 132-134

    Purpose:

    Get number of pending intervention requests.
    

    Parameters:

    Returns: int

    get_all_requests(self)

    Lines 136-138

    Purpose:

    Get all intervention requests.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D6EBA10>


    Class: HITLAgent

    Lines 141-390

    Purpose:

    Agent with Human-in-the-Loop integration.
    
    Features:
    - Configurable intervention points
    - Automatic escalation based on criteria
    - Approval gates for sensitive actions
    - Learning from human feedback
    

    Methods:

    __init__(self, human_interface, escalation_criteria, temperature)

    Lines 152-194

    Purpose: Initialize the class instance with required parameters.

    Parameters: – human_interface: Human interface – escalation_criteria: Escalation criteria – temperature: Sampling temperature (0-1)

    process_with_oversight(self, task, require_approval)

    Lines 196-273

    Purpose:

    Process task with human oversight.
    
    Args:
        task: Task to process
        require_approval: Force human approval regardless of confidence
    
    Returns:
        Dictionary with response and intervention details
    

    Parameters: – task: Task – require_approval: Require approval

    Returns: <ast.Subscript object at 0x000001A06D64E9D0>

    _parse_response(self, result)

    Lines 275-295

    Purpose:

    Parse response and confidence from LLM output.
    

    Parameters: – result: Result

    Returns: <ast.Subscript object at 0x000001A06D616050>

    _estimate_confidence(self, response)

    Lines 297-311

    Purpose:

    Estimate confidence from response content.
    

    Parameters: – response: Response

    Returns: float

    _should_escalate(self, task, response, confidence)

    Lines 313-330

    Purpose:

    Determine if task should be escalated to human.
    

    Parameters: – task: Task – response: Response – confidence: Confidence

    Returns: bool

    request_correction(self, task, initial_response, issue_description)

    Lines 332-359

    Purpose:

    Request human correction for a specific issue.
    

    Parameters: – task: Task – initial_response: Initial response – issue_description: Issue description

    Returns: <ast.Subscript object at 0x000001A06D159850>

    get_statistics(self)

    Lines 361-390

    Purpose:

    Get statistics about human interventions.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D70FE10>


    Functions

    demonstrate_human_in_the_loop()

    Lines 393-550

    Purpose:

    Demonstrates Human-in-the-Loop with various intervention scenarios.
    

    Key Concepts Summary

    Pattern Overview

    The Human In The Loop 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

    1. Agent Class: Main implementation of the pattern logic
    2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
    3. Tool Integration: Optional tools for extended functionality
    4. Memory Management: State handling across interactions
    5. 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

    1. Prompt Optimization: Minimize tokens while maintaining clarity
    2. Caching: Cache repeated queries/responses
    3. Model Selection: Use smaller models when appropriate
    4. Batch Processing: Group similar requests
    5. 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

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

    Feature Extensions

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

    Code Quality

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

    Related Patterns

    • Pattern 032
    • Pattern 034

    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 Human In The Loop 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: 033_human_in_the_loop.py
    Total Lines: 555
    Pattern Category: AI Agent Design Pattern #033


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: State Machine Agent

    
        """
    Pattern 032: State Machine Agent
    
    Description:
        State machine agents define behavior through explicit states and transitions.
        This pattern provides deterministic, predictable agent behavior with clear
        state management, transition conditions, and event handling. Ideal for
        workflows requiring strict control and auditability.
    
    Components:
        - State: Discrete condition of the agent
        - Transition: Rule for moving between states
        - Event: Trigger that causes transitions
        - Guard: Condition that must be met for transition
        - Action: Operation performed during transition
        - State Context: Data maintained across states
    
    Use Cases:
        - Workflow automation
        - Conversation management
        - Game AI with distinct phases
        - Process orchestration
        - Regulatory compliance scenarios
    
    LangChain Implementation:
        Uses state definitions, transition rules, and event handling to create
        predictable agent behavior. Can be integrated with LangGraph for
        visual workflow representation.
    """
    
    import os
    from typing import List, Dict, Any, Optional, Callable, Set
    from dataclasses import dataclass, field
    from enum import Enum
    from datetime import datetime
    from collections import defaultdict
    import json
    from dotenv import load_dotenv
    
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser
    from langchain_openai import ChatOpenAI
    
    load_dotenv()
    
    
    class EventType(Enum):
        """Types of events that can trigger transitions."""
        USER_INPUT = "user_input"
        TIMEOUT = "timeout"
        COMPLETION = "completion"
        ERROR = "error"
        EXTERNAL = "external"
        INTERNAL = "internal"
    
    
    @dataclass
    class Event:
        """An event that can trigger state transitions."""
        type: EventType
        name: str
        data: Dict[str, Any] = field(default_factory=dict)
        timestamp: datetime = field(default_factory=datetime.now)
    
    
    @dataclass
    class State:
        """
        Represents a discrete state in the state machine.
        
        States define what the agent can do and how it responds.
        """
        name: str
        description: str
        entry_action: Optional[Callable] = None  # Called when entering state
        exit_action: Optional[Callable] = None  # Called when leaving state
        allowed_events: Set[str] = field(default_factory=set)  # Events accepted in this state
        metadata: Dict[str, Any] = field(default_factory=dict)
        
        def __hash__(self):
            return hash(self.name)
        
        def __eq__(self, other):
            return isinstance(other, State) and self.name == other.name
    
    
    @dataclass
    class Transition:
        """
        Defines a transition between states.
        
        Transitions specify how and when the agent moves between states.
        """
        from_state: str
        to_state: str
        event: str
        guard: Optional[Callable[[Dict[str, Any]], bool]] = None  # Condition function
        action: Optional[Callable[[Dict[str, Any]], Dict[str, Any]]] = None  # Transition action
        priority: int = 0  # Higher priority transitions checked first
        description: str = ""
    
    
    @dataclass
    class StateContext:
        """
        Maintains data across state transitions.
        
        The context is preserved and passed through the state machine.
        """
        data: Dict[str, Any] = field(default_factory=dict)
        history: List[str] = field(default_factory=list)  # State history
        event_log: List[Event] = field(default_factory=list)  # Event history
        
        def set(self, key: str, value: Any):
            """Set context value."""
            self.data[key] = value
        
        def get(self, key: str, default: Any = None) -> Any:
            """Get context value."""
            return self.data.get(key, default)
        
        def log_event(self, event: Event):
            """Log an event."""
            self.event_log.append(event)
        
        def log_state(self, state_name: str):
            """Log state transition."""
            self.history.append(state_name)
    
    
    class StateMachine:
        """
        Implements a finite state machine.
        
        Features:
        - State management
        - Event-driven transitions
        - Guard conditions
        - Entry/exit actions
        - Context preservation
        """
        
        def __init__(self, initial_state: str):
            self.states: Dict[str, State] = {}
            self.transitions: List[Transition] = []
            self.current_state: Optional[str] = None
            self.initial_state = initial_state
            self.context = StateContext()
            
            # Index transitions by from_state and event for efficiency
            self.transition_index: Dict[tuple, List[Transition]] = defaultdict(list)
        
        def add_state(self, state: State):
            """Add a state to the machine."""
            self.states[state.name] = state
        
        def add_transition(self, transition: Transition):
            """Add a transition to the machine."""
            self.transitions.append(transition)
            # Index by (from_state, event) for quick lookup
            key = (transition.from_state, transition.event)
            self.transition_index[key].append(transition)
            # Sort by priority
            self.transition_index[key].sort(key=lambda t: t.priority, reverse=True)
        
        def start(self):
            """Initialize the state machine."""
            if self.initial_state not in self.states:
                raise ValueError(f"Initial state '{self.initial_state}' not found")
            
            self.current_state = self.initial_state
            self.context.log_state(self.current_state)
            
            # Execute entry action
            state = self.states[self.current_state]
            if state.entry_action:
                state.entry_action(self.context)
        
        def handle_event(self, event: Event) -> bool:
            """
            Process an event and potentially transition to a new state.
            
            Returns:
                True if a transition occurred, False otherwise
            """
            if not self.current_state:
                raise RuntimeError("State machine not started")
            
            # Log event
            self.context.log_event(event)
            
            # Check if event is allowed in current state
            current_state_obj = self.states[self.current_state]
            if current_state_obj.allowed_events and event.name not in current_state_obj.allowed_events:
                return False
            
            # Find applicable transition
            key = (self.current_state, event.name)
            possible_transitions = self.transition_index.get(key, [])
            
            for transition in possible_transitions:
                # Check guard condition
                if transition.guard and not transition.guard(self.context.data):
                    continue
                
                # Transition is valid - execute it
                self._execute_transition(transition, event)
                return True
            
            return False
        
        def _execute_transition(self, transition: Transition, event: Event):
            """Execute a state transition."""
            old_state = self.current_state
            new_state = transition.to_state
            
            # Exit action of current state
            if self.states[old_state].exit_action:
                self.states[old_state].exit_action(self.context)
            
            # Transition action
            if transition.action:
                result = transition.action(self.context.data)
                if result:
                    self.context.data.update(result)
            
            # Update current state
            self.current_state = new_state
            self.context.log_state(new_state)
            
            # Entry action of new state
            if self.states[new_state].entry_action:
                self.states[new_state].entry_action(self.context)
        
        def get_current_state(self) -> Optional[State]:
            """Get the current state object."""
            if self.current_state:
                return self.states[self.current_state]
            return None
        
        def get_available_events(self) -> Set[str]:
            """Get events that can be handled in current state."""
            if not self.current_state:
                return set()
            
            # Get events from transitions
            events = set()
            for transition in self.transitions:
                if transition.from_state == self.current_state:
                    events.add(transition.event)
            
            return events
        
        def is_in_state(self, state_name: str) -> bool:
            """Check if currently in specified state."""
            return self.current_state == state_name
        
        def get_state_history(self) -> List[str]:
            """Get history of states visited."""
            return self.context.history.copy()
    
    
    class StateMachineAgent:
        """
        An agent that uses a state machine for behavior control.
        
        Combines state machine determinism with LLM flexibility for
        decision-making within states.
        """
        
        def __init__(self, state_machine: StateMachine, temperature: float = 0.3):
            self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=temperature)
            self.state_machine = state_machine
            
            # Prompt for processing within a state
            self.process_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are an AI agent operating in a state machine.
    
    Current State: {state_name}
    State Description: {state_description}
    Available Events: {available_events}
    Context: {context}
    
    User Input: {user_input}
    
    Based on the current state and user input, determine:
    1. What event should be triggered (choose from available events)
    2. What data should be included with the event
    
    Respond in JSON format:
    {{
      "event": "event_name",
      "data": {{...}}
    }}"""),
                ("user", "{user_input}")
            ])
        
        def start(self):
            """Start the agent's state machine."""
            self.state_machine.start()
            print(f"Agent started in state: {self.state_machine.current_state}")
        
        def process_input(self, user_input: str) -> Dict[str, Any]:
            """
            Process user input within current state context.
            
            Uses LLM to determine appropriate event to trigger.
            """
            current_state = self.state_machine.get_current_state()
            
            if not current_state:
                return {"error": "State machine not started"}
            
            available_events = self.state_machine.get_available_events()
            
            # Use LLM to determine event and data
            chain = self.process_prompt | self.llm | StrOutputParser()
            response = chain.invoke({
                "state_name": current_state.name,
                "state_description": current_state.description,
                "available_events": ", ".join(available_events),
                "context": json.dumps(self.state_machine.context.data),
                "user_input": user_input
            })
            
            # Parse response
            try:
                # Extract JSON from response
                start_idx = response.find('{')
                end_idx = response.rfind('}') + 1
                if start_idx != -1 and end_idx != 0:
                    json_str = response[start_idx:end_idx]
                    parsed = json.loads(json_str)
                else:
                    parsed = json.loads(response)
                
                event_name = parsed.get("event")
                event_data = parsed.get("data", {})
                
                # Create and handle event
                event = Event(
                    type=EventType.USER_INPUT,
                    name=event_name,
                    data=event_data
                )
                
                transitioned = self.state_machine.handle_event(event)
                
                return {
                    "success": True,
                    "event": event_name,
                    "transitioned": transitioned,
                    "new_state": self.state_machine.current_state,
                    "response": self._generate_response()
                }
                
            except Exception as e:
                return {
                    "success": False,
                    "error": str(e),
                    "raw_response": response
                }
        
        def _generate_response(self) -> str:
            """Generate a response based on current state."""
            current_state = self.state_machine.get_current_state()
            
            if not current_state:
                return "I'm not in a valid state."
            
            # Generate contextual response
            response_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are in state: {state_name}
    {state_description}
    
    Generate a brief, helpful response acknowledging the state and guiding the user."""),
                ("user", "Generate response")
            ])
            
            chain = response_prompt | self.llm | StrOutputParser()
            response = chain.invoke({
                "state_name": current_state.name,
                "state_description": current_state.description
            })
            
            return response
        
        def get_status(self) -> Dict[str, Any]:
            """Get current agent status."""
            current_state = self.state_machine.get_current_state()
            
            return {
                "current_state": self.state_machine.current_state,
                "state_description": current_state.description if current_state else "",
                "available_events": list(self.state_machine.get_available_events()),
                "context": self.state_machine.context.data,
                "state_history": self.state_machine.get_state_history()
            }
    
    
    def create_conversation_agent() -> StateMachineAgent:
        """
        Create a conversation agent with state machine.
        
        States: greeting → collecting_info → processing → providing_result → farewell
        """
        # Create state machine
        sm = StateMachine(initial_state="greeting")
        
        # Define states
        states = [
            State(
                name="greeting",
                description="Initial greeting state, welcoming the user",
                allowed_events={"start_conversation", "ask_question"}
            ),
            State(
                name="collecting_info",
                description="Collecting information from the user",
                allowed_events={"provide_info", "clarify", "cancel"}
            ),
            State(
                name="processing",
                description="Processing user request",
                allowed_events={"complete", "error"}
            ),
            State(
                name="providing_result",
                description="Providing results to user",
                allowed_events={"ask_followup", "satisfied", "retry"}
            ),
            State(
                name="farewell",
                description="Ending the conversation",
                allowed_events={"restart"}
            ),
            State(
                name="error",
                description="Error state",
                allowed_events={"retry", "cancel"}
            )
        ]
        
        for state in states:
            sm.add_state(state)
        
        # Define transitions
        transitions = [
            # From greeting
            Transition("greeting", "collecting_info", "start_conversation",
                      description="User starts conversation"),
            Transition("greeting", "providing_result", "ask_question",
                      description="User asks direct question"),
            
            # From collecting_info
            Transition("collecting_info", "processing", "provide_info",
                      description="User provides required info"),
            Transition("collecting_info", "collecting_info", "clarify",
                      description="User needs clarification"),
            Transition("collecting_info", "farewell", "cancel",
                      description="User cancels"),
            
            # From processing
            Transition("processing", "providing_result", "complete",
                      description="Processing completed successfully"),
            Transition("processing", "error", "error",
                      description="Error during processing"),
            
            # From providing_result
            Transition("providing_result", "collecting_info", "ask_followup",
                      description="User has follow-up questions"),
            Transition("providing_result", "farewell", "satisfied",
                      description="User is satisfied"),
            Transition("providing_result", "processing", "retry",
                      description="User wants to retry"),
            
            # From error
            Transition("error", "processing", "retry",
                      description="Retry after error"),
            Transition("error", "farewell", "cancel",
                      description="Cancel after error"),
            
            # From farewell
            Transition("farewell", "greeting", "restart",
                      description="Start new conversation")
        ]
        
        for transition in transitions:
            sm.add_transition(transition)
        
        # Create agent
        agent = StateMachineAgent(sm)
        
        return agent
    
    
    def demonstrate_state_machine_agent():
        """
        Demonstrates state machine agent with conversation flow.
        """
        print("=" * 80)
        print("STATE MACHINE AGENT DEMONSTRATION")
        print("=" * 80)
        
        # Create conversation agent
        agent = create_conversation_agent()
        
        # Test 1: Start and follow conversation flow
        print("\n" + "=" * 80)
        print("Test 1: Conversation Flow")
        print("=" * 80)
        
        # Start agent
        agent.start()
        
        # Simulate conversation
        conversation = [
            "Hello, I need help with something",
            "I want to learn about Python programming",
            "I've provided the information",
            "That's helpful, thank you!",
            "Goodbye"
        ]
        
        for user_input in conversation:
            print(f"\nUser: {user_input}")
            
            # Get status before
            status = agent.get_status()
            print(f"Current State: {status['current_state']}")
            print(f"Available Events: {', '.join(status['available_events'])}")
            
            # Process input
            result = agent.process_input(user_input)
            
            if result.get("success"):
                print(f"Event Triggered: {result['event']}")
                print(f"Transitioned: {result['transitioned']}")
                print(f"New State: {result['new_state']}")
                print(f"Agent: {result['response']}")
            else:
                print(f"Error: {result.get('error')}")
        
        # Test 2: Show state history
        print("\n" + "=" * 80)
        print("Test 2: State History")
        print("=" * 80)
        
        history = agent.state_machine.get_state_history()
        print(f"\nState transitions: {' → '.join(history)}")
        
        # Test 3: Event log
        print("\n" + "=" * 80)
        print("Test 3: Event Log")
        print("=" * 80)
        
        print(f"\nTotal events: {len(agent.state_machine.context.event_log)}")
        for i, event in enumerate(agent.state_machine.context.event_log, 1):
            print(f"{i}. {event.name} ({event.type.value}) at {event.timestamp.strftime('%H:%M:%S')}")
        
        # Test 4: Create different state machine (order processing)
        print("\n" + "=" * 80)
        print("Test 4: Order Processing State Machine")
        print("=" * 80)
        
        # Create order processing state machine
        order_sm = StateMachine(initial_state="cart")
        
        order_states = [
            State("cart", "Shopping cart - adding items", allowed_events={"checkout", "cancel"}),
            State("payment", "Payment processing", allowed_events={"pay", "cancel"}),
            State("fulfillment", "Order fulfillment", allowed_events={"ship", "cancel"}),
            State("shipped", "Order shipped", allowed_events={"deliver", "return"}),
            State("completed", "Order completed", allowed_events={"review"}),
            State("cancelled", "Order cancelled", allowed_events={"restart"})
        ]
        
        for state in order_states:
            order_sm.add_state(state)
        
        order_transitions = [
            Transition("cart", "payment", "checkout"),
            Transition("cart", "cancelled", "cancel"),
            Transition("payment", "fulfillment", "pay"),
            Transition("payment", "cancelled", "cancel"),
            Transition("fulfillment", "shipped", "ship"),
            Transition("fulfillment", "cancelled", "cancel"),
            Transition("shipped", "completed", "deliver"),
            Transition("shipped", "cancelled", "return"),
            Transition("completed", "cart", "review"),
            Transition("cancelled", "cart", "restart")
        ]
        
        for transition in order_transitions:
            order_sm.add_transition(transition)
        
        # Start and simulate order flow
        order_sm.start()
        print(f"Starting state: {order_sm.current_state}")
        
        order_events = [
            ("checkout", "Proceed to checkout"),
            ("pay", "Payment successful"),
            ("ship", "Order shipped"),
            ("deliver", "Order delivered")
        ]
        
        for event_name, description in order_events:
            event = Event(EventType.INTERNAL, event_name)
            success = order_sm.handle_event(event)
            print(f"✓ {description} → {order_sm.current_state}")
        
        print(f"\nFinal state: {order_sm.current_state}")
        print(f"Order flow: {' → '.join(order_sm.get_state_history())}")
        
        # Summary
        print("\n" + "=" * 80)
        print("SUMMARY")
        print("=" * 80)
        print("""
    State Machine Agent provides:
    ✓ Explicit state definitions
    ✓ Event-driven transitions
    ✓ Guard conditions for validation
    ✓ Entry/exit actions
    ✓ Context preservation across states
    ✓ Deterministic behavior
    
    This pattern excels at:
    - Workflow automation with clear phases
    - Conversation management with distinct stages
    - Process orchestration
    - Game AI with discrete states
    - Compliance and auditing requirements
    
    Key components:
    1. States: Discrete conditions with specific behaviors
    2. Transitions: Rules for moving between states
    3. Events: Triggers that cause transitions
    4. Guards: Conditions that must be met
    5. Actions: Operations during transitions
    6. Context: Data maintained across states
    
    Advantages:
    - Predictable behavior
    - Easy to visualize and debug
    - Clear state tracking
    - Audit trail through history
    - Testable transitions
    - Integration with workflow tools
    
    State machine patterns:
    - Linear: greeting → processing → farewell
    - Branching: processing → success | error
    - Cyclic: processing → review → processing
    - Hierarchical: nested sub-state machines
    
    Use state machine agents when you need:
    - Deterministic, predictable behavior
    - Clear workflow phases
    - Audit trails and compliance
    - Visual workflow representation
    - Easy testing and debugging
    - Formal verification of behavior
    """)
    
    
    if __name__ == "__main__":
        demonstrate_state_machine_agent()
    
        

    Agentic Design Pattern: State Machine Agent

    Pattern 032: State Machine Agent – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the State Machine Agent pattern implementation.


    Pattern Description

    Pattern 032: State Machine Agent

    Description: State machine agents define behavior through explicit states and transitions. This pattern provides deterministic, predictable agent behavior with clear state management, transition conditions, and event handling. Ideal for workflows requiring strict control and auditability.

    Components: – State: Discrete condition of the agent – Transition: Rule for moving between states – Event: Trigger that causes transitions – Guard: Condition that must be met for transition – Action: Operation performed during transition – State Context: Data maintained across states

    Use Cases: – Workflow automation – Conversation management – Game AI with distinct phases – Process orchestration – Regulatory compliance scenarios

    LangChain Implementation: Uses state definitions, transition rules, and event handling to create predictable agent behavior. Can be integrated with LangGraph for visual workflow representation.


    File Statistics

    • Total Lines: 670
    • Classes: 7
    • Functions: 2
    • Constants: 0

    Module Documentation

    """
    Pattern 032: State Machine Agent
    
    Description:
        State machine agents define behavior through explicit states and transitions.
        This pattern provides deterministic, predictable agent behavior with clear
        state management, transition conditions, and event handling. Ideal for
        workflows requiring strict control and auditability.
    
    Components:
        - State: Discrete condition of the agent
        - Transition: Rule for moving between states
        - Event: Trigger that causes transitions
        - Guard: Condition that must be met for transition
        - Action: Operation performed during transition
        - State Context: Data maintained across states
    
    Use Cases:
        - Workflow automation
        - Conversation management
        - Game AI with distinct phases
        - Process orchestration
        - Regulatory compliance scenarios
    
    LangChain Implementation:
        Uses state definitions, transition rules, and event handling to create
        predictable agent behavior. Can be integrated with LangGraph for
        visual workflow representation.
    """
    

    Purpose: This module implements the State Machine Agent pattern, providing a structured approach to agent design and implementation.


    Import Statements

    Standard Library Imports

    import collections.defaultdict
    import dataclasses.dataclass
    import dataclasses.field
    import datetime.datetime
    import enum.Enum
    import json
    import langchain_core.output_parsers.StrOutputParser
    import langchain_core.prompts.ChatPromptTemplate
    import os
    import typing.Any
    import typing.Callable
    import typing.Dict
    import typing.List
    import typing.Optional
    import typing.Set
    

    Purpose: Core Python functionality: – collections.defaultdict: Specialized container datatypes – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – datetime.datetime: Date and time handling – enum.Enum: Enumeration support – json: JSON serialization/deserialization – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Callable: 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) – typing.Set: 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: EventType

    Lines 47-54

    Inherits from: Enum

    Purpose:

    Types of events that can trigger transitions.
    

    Class: Event

    Lines 58-63

    Purpose:

    An event that can trigger state transitions.
    

    Class: State

    Lines 67-84

    Purpose:

    Represents a discrete state in the state machine.
    
    States define what the agent can do and how it responds.
    

    Methods:

    __hash__(self)

    Lines 80-81

    Purpose: Hash operation.

    Parameters:

    __eq__(self, other)

    Lines 83-84

    Purpose: Eq operation.

    Parameters: – other: Other


    Class: Transition

    Lines 88-100

    Purpose:

    Defines a transition between states.
    
    Transitions specify how and when the agent moves between states.
    

    Class: StateContext

    Lines 104-128

    Purpose:

    Maintains data across state transitions.
    
    The context is preserved and passed through the state machine.
    

    Methods:

    set(self, key, value)

    Lines 114-116

    Purpose:

    Set context value.
    

    Parameters: – key: Key – value: Value

    get(self, key, default)

    Lines 118-120

    Purpose:

    Get context value.
    

    Parameters: – key: Key – default: Default

    Returns: Any

    log_event(self, event)

    Lines 122-124

    Purpose:

    Log an event.
    

    Parameters: – event: Event

    log_state(self, state_name)

    Lines 126-128

    Purpose:

    Log state transition.
    

    Parameters: – state_name: State name


    Class: StateMachine

    Lines 131-260

    Purpose:

    Implements a finite state machine.
    
    Features:
    - State management
    - Event-driven transitions
    - Guard conditions
    - Entry/exit actions
    - Context preservation
    

    Methods:

    __init__(self, initial_state)

    Lines 143-151

    Purpose: Initialize the class instance with required parameters.

    Parameters: – initial_state: Initial state

    add_state(self, state)

    Lines 153-155

    Purpose:

    Add a state to the machine.
    

    Parameters: – state: State

    add_transition(self, transition)

    Lines 157-164

    Purpose:

    Add a transition to the machine.
    

    Parameters: – transition: Transition

    start(self)

    Lines 166-177

    Purpose:

    Initialize the state machine.
    

    Parameters:

    handle_event(self, event)

    Lines 179-210

    Purpose:

    Process an event and potentially transition to a new state.
    
    Returns:
        True if a transition occurred, False otherwise
    

    Parameters: – event: Event

    Returns: bool

    _execute_transition(self, transition, event)

    Lines 212-233

    Purpose:

    Execute a state transition.
    

    Parameters: – transition: Transition – event: Event

    get_current_state(self)

    Lines 235-239

    Purpose:

    Get the current state object.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D738C10>

    get_available_events(self)

    Lines 241-252

    Purpose:

    Get events that can be handled in current state.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D6046D0>

    is_in_state(self, state_name)

    Lines 254-256

    Purpose:

    Check if currently in specified state.
    

    Parameters: – state_name: State name

    Returns: bool

    get_state_history(self)

    Lines 258-260

    Purpose:

    Get history of states visited.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D607090>


    Class: StateMachineAgent

    Lines 263-398

    Purpose:

    An agent that uses a state machine for behavior control.
    
    Combines state machine determinism with LLM flexibility for
    decision-making within states.
    

    Methods:

    __init__(self, state_machine, temperature)

    Lines 271-296

    Purpose: Initialize the class instance with required parameters.

    Parameters: – state_machine: State machine – temperature: Sampling temperature (0-1)

    start(self)

    Lines 298-301

    Purpose:

    Start the agent's state machine.
    

    Parameters:

    process_input(self, user_input)

    Lines 303-362

    Purpose:

    Process user input within current state context.
    
    Uses LLM to determine appropriate event to trigger.
    

    Parameters: – user_input: Input data

    Returns: <ast.Subscript object at 0x000001A06D609A90>

    _generate_response(self)

    Lines 364-386

    Purpose:

    Generate a response based on current state.
    

    Parameters:

    Returns: str

    get_status(self)

    Lines 388-398

    Purpose:

    Get current agent status.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D682D10>


    Functions

    create_conversation_agent()

    Lines 401-494

    Purpose:

    Create a conversation agent with state machine.
    
    States: greeting → collecting_info → processing → providing_result → farewell
    

    Returns: StateMachineAgent

    demonstrate_state_machine_agent()

    Lines 497-665

    Purpose:

    Demonstrates state machine agent with conversation flow.
    

    Key Concepts Summary

    Pattern Overview

    The State Machine Agent 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

    1. Agent Class: Main implementation of the pattern logic
    2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
    3. Tool Integration: Optional tools for extended functionality
    4. Memory Management: State handling across interactions
    5. 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

    1. Prompt Optimization: Minimize tokens while maintaining clarity
    2. Caching: Cache repeated queries/responses
    3. Model Selection: Use smaller models when appropriate
    4. Batch Processing: Group similar requests
    5. 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

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

    Feature Extensions

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

    Code Quality

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

    Related Patterns

    • Pattern 031
    • Pattern 033

    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 State Machine Agent 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: 032_state_machine_agent.py
    Total Lines: 670
    Pattern Category: AI Agent Design Pattern #032


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Memory Consolidation

    
        """
    Pattern 031: Memory Consolidation
    
    Description:
        Memory consolidation processes and organizes memories over time through
        summarization, compression, and hierarchical structuring. This pattern prevents
        memory overflow, improves retrieval efficiency, and maintains the most important
        information while discarding or compressing less relevant details.
    
    Components:
        - Consolidation Scheduler: Triggers consolidation processes
        - Summarizer: Creates compressed representations of memories
        - Importance Evaluator: Scores memory significance
        - Memory Pruner: Removes or archives low-value memories
        - Hierarchical Organizer: Creates multi-level memory structures
        - Index Builder: Creates efficient retrieval structures
    
    Use Cases:
        - Long-running agents with accumulating memory
        - Managing limited memory resources
        - Improving retrieval speed
        - Creating knowledge hierarchies
        - Intelligent forgetting
    
    LangChain Implementation:
        Uses LLM-based summarization, importance scoring, and hierarchical
        organization to maintain an efficient and useful memory structure.
    """
    
    import os
    from typing import List, Dict, Any, Optional, Set
    from dataclasses import dataclass, field
    from datetime import datetime, timedelta
    from enum import Enum
    import json
    from collections import defaultdict
    from dotenv import load_dotenv
    
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser
    from langchain_openai import ChatOpenAI
    
    load_dotenv()
    
    
    class MemoryLevel(Enum):
        """Hierarchical levels of memory consolidation."""
        RAW = "raw"  # Original, unconsolidated memories
        CONSOLIDATED = "consolidated"  # Grouped and summarized
        ABSTRACTED = "abstracted"  # High-level abstractions
        ARCHIVED = "archived"  # Long-term storage
    
    
    class ImportanceCriterion(Enum):
        """Criteria for evaluating memory importance."""
        FREQUENCY = "frequency"  # How often accessed
        RECENCY = "recency"  # How recent
        EMOTIONAL = "emotional"  # Emotional significance
        RELEVANCE = "relevance"  # Topic relevance
        UNIQUENESS = "uniqueness"  # How unique/rare
        OUTCOME = "outcome"  # Success/failure impact
    
    
    @dataclass
    class Memory:
        """Individual memory unit."""
        id: str
        content: str
        timestamp: datetime
        level: MemoryLevel
        importance_score: float = 0.5
        access_count: int = 0
        last_accessed: Optional[datetime] = None
        tags: Set[str] = field(default_factory=set)
        metadata: Dict[str, Any] = field(default_factory=dict)
        consolidated_from: List[str] = field(default_factory=list)  # Source memory IDs
        
        def mark_accessed(self):
            """Mark memory as accessed."""
            self.access_count += 1
            self.last_accessed = datetime.now()
    
    
    @dataclass
    class ConsolidatedMemory(Memory):
        """Consolidated memory containing summarized information."""
        summary: str = ""
        source_count: int = 0
        time_span: Optional[timedelta] = None
        
        def __post_init__(self):
            self.level = MemoryLevel.CONSOLIDATED
    
    
    @dataclass
    class ConsolidationResult:
        """Result of a consolidation operation."""
        memories_processed: int
        memories_consolidated: int
        memories_pruned: int
        memories_archived: int
        new_memories_created: int
        space_saved: float  # Percentage
    
    
    class ImportanceEvaluator:
        """
        Evaluates memory importance using multiple criteria.
        
        Combines various factors to produce a single importance score.
        """
        
        def __init__(self, llm: ChatOpenAI):
            self.llm = llm
            
            self.evaluation_prompt = ChatPromptTemplate.from_messages([
                ("system", """Evaluate the importance of this memory on a scale of 0.0 to 1.0.
    
    Memory: {content}
    Context: {context}
    
    Consider:
    - Relevance to long-term goals
    - Uniqueness of information
    - Emotional or practical significance
    - Potential future utility
    
    Respond with just a number between 0.0 and 1.0."""),
                ("user", "Evaluate importance")
            ])
        
        def evaluate(
            self,
            memory: Memory,
            current_time: datetime,
            recency_weight: float = 0.3,
            access_weight: float = 0.3,
            llm_weight: float = 0.4
        ) -> float:
            """
            Calculate importance score for a memory.
            
            Combines multiple factors with configurable weights.
            """
            # Recency score (exponential decay)
            hours_ago = (current_time - memory.timestamp).total_seconds() / 3600
            recency_score = max(0.0, 1.0 - (hours_ago / (24 * 7)))  # Decay over 1 week
            
            # Access frequency score (normalized)
            access_score = min(1.0, memory.access_count / 10.0)
            
            # LLM-based semantic importance
            llm_score = self._evaluate_semantic_importance(memory)
            
            # Weighted combination
            importance = (
                recency_weight * recency_score +
                access_weight * access_score +
                llm_weight * llm_score
            )
            
            return importance
        
        def _evaluate_semantic_importance(self, memory: Memory) -> float:
            """Use LLM to evaluate semantic importance."""
            try:
                chain = self.evaluation_prompt | self.llm | StrOutputParser()
                result = chain.invoke({
                    "content": memory.content,
                    "context": json.dumps(memory.metadata)
                })
                
                # Extract number from response
                score_str = result.strip()
                score = float(score_str)
                return max(0.0, min(1.0, score))  # Clamp to [0, 1]
            except Exception:
                # Fallback to default
                return 0.5
    
    
    class MemoryConsolidator:
        """
        Performs memory consolidation operations.
        
        Capabilities:
        - Group related memories
        - Create summaries
        - Build hierarchies
        - Prune low-value memories
        """
        
        def __init__(self, llm: ChatOpenAI):
            self.llm = llm
            
            # Prompt for summarizing multiple memories
            self.summary_prompt = ChatPromptTemplate.from_messages([
                ("system", """Summarize these related memories into a concise consolidated memory.
    
    Memories to consolidate:
    {memories}
    
    Create a summary that captures the essential information while being more concise.
    Focus on patterns, key facts, and important outcomes."""),
                ("user", "Create consolidated summary")
            ])
            
            # Prompt for finding relationships
            self.grouping_prompt = ChatPromptTemplate.from_messages([
                ("system", """Analyze these memories and identify which ones are related and should be grouped together.
    
    Memories:
    {memories}
    
    Return groups as JSON array of arrays containing memory indices.
    Example: [[0, 1, 3], [2, 4], [5]]"""),
                ("user", "Group related memories")
            ])
        
        def consolidate_memories(
            self,
            memories: List[Memory],
            importance_threshold: float = 0.3,
            max_age_days: int = 30
        ) -> ConsolidationResult:
            """
            Perform full consolidation cycle on memories.
            
            Steps:
            1. Evaluate importance
            2. Group related memories
            3. Summarize groups
            4. Prune low-value memories
            5. Update memory levels
            """
            current_time = datetime.now()
            evaluator = ImportanceEvaluator(self.llm)
            
            result = ConsolidationResult(
                memories_processed=len(memories),
                memories_consolidated=0,
                memories_pruned=0,
                memories_archived=0,
                new_memories_created=0,
                space_saved=0.0
            )
            
            # Step 1: Update importance scores
            for memory in memories:
                memory.importance_score = evaluator.evaluate(memory, current_time)
            
            # Step 2: Identify memories to prune
            prunable = [
                m for m in memories
                if m.importance_score < importance_threshold
                and m.level == MemoryLevel.RAW
            ]
            result.memories_pruned = len(prunable)
            
            # Step 3: Group related memories for consolidation
            consolidatable = [
                m for m in memories
                if m.level == MemoryLevel.RAW
                and m not in prunable
                and (current_time - m.timestamp).days = 2:
                groups = self._group_memories(consolidatable)
                
                # Step 4: Create consolidated memories
                for group in groups:
                    if len(group) >= 2:  # Only consolidate groups of 2+
                        consolidated = self._create_consolidated_memory(group)
                        result.new_memories_created += 1
                        result.memories_consolidated += len(group)
            
            # Step 5: Archive old memories
            archivable = [
                m for m in memories
                if (current_time - m.timestamp).days > max_age_days
                and m.level == MemoryLevel.CONSOLIDATED
            ]
            result.memories_archived = len(archivable)
            
            # Calculate space saved
            if memories:
                original_size = sum(len(m.content) for m in memories)
                pruned_size = sum(len(m.content) for m in prunable)
                if original_size > 0:
                    result.space_saved = (pruned_size / original_size) * 100
            
            return result
        
        def _group_memories(self, memories: List[Memory]) -> List[List[Memory]]:
            """
            Group related memories together.
            
            Uses tag similarity and temporal proximity.
            """
            if len(memories) = 1:  # At least one common tag
                            group.append(other)
                            processed.add(j)
                            continue
                    
                    # Check temporal proximity (within 1 day)
                    time_diff = abs((memory.timestamp - other.timestamp).total_seconds())
                    if time_diff  ConsolidatedMemory:
            """
            Create a consolidated memory from a group.
            """
            # Prepare memory texts
            memory_texts = "\n\n".join([
                f"Memory {i+1} ({m.timestamp.strftime('%Y-%m-%d %H:%M')}):\n{m.content}"
                for i, m in enumerate(group)
            ])
            
            # Generate summary
            chain = self.summary_prompt | self.llm | StrOutputParser()
            summary = chain.invoke({"memories": memory_texts})
            
            # Combine metadata
            combined_tags = set()
            for memory in group:
                combined_tags.update(memory.tags)
            
            # Calculate time span
            timestamps = [m.timestamp for m in group]
            time_span = max(timestamps) - min(timestamps)
            
            # Calculate average importance
            avg_importance = sum(m.importance_score for m in group) / len(group)
            
            # Create consolidated memory
            consolidated = ConsolidatedMemory(
                id=f"consolidated_{group[0].id}",
                content=summary,
                summary=summary,
                timestamp=min(timestamps),  # Use earliest timestamp
                level=MemoryLevel.CONSOLIDATED,
                importance_score=avg_importance,
                tags=combined_tags,
                consolidated_from=[m.id for m in group],
                source_count=len(group),
                time_span=time_span
            )
            
            return consolidated
    
    
    class ConsolidationScheduler:
        """
        Manages when and how consolidation is triggered.
        
        Supports multiple triggering strategies:
        - Time-based (periodic)
        - Size-based (memory threshold)
        - Demand-based (on query)
        """
        
        def __init__(
            self,
            consolidation_interval_hours: int = 24,
            size_threshold: int = 100,
            importance_threshold: float = 0.3
        ):
            self.consolidation_interval_hours = consolidation_interval_hours
            self.size_threshold = size_threshold
            self.importance_threshold = importance_threshold
            self.last_consolidation = datetime.now()
        
        def should_consolidate(
            self,
            memory_count: int,
            time_since_last: Optional[timedelta] = None
        ) -> bool:
            """
            Determine if consolidation should be triggered.
            """
            # Size-based trigger
            if memory_count >= self.size_threshold:
                return True
            
            # Time-based trigger
            if time_since_last is None:
                time_since_last = datetime.now() - self.last_consolidation
            
            hours_since = time_since_last.total_seconds() / 3600
            if hours_since >= self.consolidation_interval_hours:
                return True
            
            return False
        
        def mark_consolidated(self):
            """Mark that consolidation has occurred."""
            self.last_consolidation = datetime.now()
    
    
    class MemoryConsolidationAgent:
        """
        Agent that manages memory with automatic consolidation.
        
        Features:
        - Automatic memory consolidation
        - Importance-based retention
        - Hierarchical memory organization
        - Efficient retrieval
        """
        
        def __init__(self, temperature: float = 0.3):
            self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=temperature)
            self.consolidator = MemoryConsolidator(self.llm)
            self.scheduler = ConsolidationScheduler()
            
            self.raw_memories: List[Memory] = []
            self.consolidated_memories: List[ConsolidatedMemory] = []
            self.archived_memories: List[Memory] = []
            
            self._next_id = 1
        
        def add_memory(
            self,
            content: str,
            tags: Optional[Set[str]] = None,
            metadata: Optional[Dict[str, Any]] = None
        ) -> str:
            """Add a new memory and check if consolidation is needed."""
            memory = Memory(
                id=f"mem_{self._next_id:04d}",
                content=content,
                timestamp=datetime.now(),
                level=MemoryLevel.RAW,
                tags=tags or set(),
                metadata=metadata or {}
            )
            
            self._next_id += 1
            self.raw_memories.append(memory)
            
            # Check if consolidation should be triggered
            if self.scheduler.should_consolidate(len(self.raw_memories)):
                self.consolidate()
            
            return memory.id
        
        def consolidate(self) -> ConsolidationResult:
            """
            Perform memory consolidation.
            
            This processes raw memories, creates consolidated versions,
            and prunes low-value memories.
            """
            print("\n🔄 Starting memory consolidation...")
            
            result = self.consolidator.consolidate_memories(
                self.raw_memories,
                importance_threshold=self.scheduler.importance_threshold
            )
            
            # Apply consolidation results
            # Remove pruned memories
            self.raw_memories = [
                m for m in self.raw_memories
                if m.importance_score >= self.scheduler.importance_threshold
            ]
            
            # Mark scheduler
            self.scheduler.mark_consolidated()
            
            print(f"✓ Consolidation complete: {result.memories_consolidated} memories consolidated, "
                  f"{result.memories_pruned} pruned, {result.space_saved:.1f}% space saved")
            
            return result
        
        def retrieve_memories(
            self,
            query: str,
            include_consolidated: bool = True,
            max_results: int = 5
        ) -> List[Memory]:
            """
            Retrieve relevant memories.
            
            Searches across raw and consolidated memories.
            """
            query_lower = query.lower()
            all_memories = self.raw_memories[:]
            
            if include_consolidated:
                all_memories.extend(self.consolidated_memories)
            
            # Simple keyword matching
            relevant = []
            for memory in all_memories:
                if query_lower in memory.content.lower():
                    memory.mark_accessed()
                    relevant.append(memory)
            
            # Sort by importance * recency
            current_time = datetime.now()
            relevant.sort(
                key=lambda m: m.importance_score * (
                    1.0 / (1.0 + (current_time - m.timestamp).days)
                ),
                reverse=True
            )
            
            return relevant[:max_results]
        
        def get_statistics(self) -> Dict[str, Any]:
            """Get memory statistics."""
            total_memories = (
                len(self.raw_memories) +
                len(self.consolidated_memories) +
                len(self.archived_memories)
            )
            
            consolidated_count = sum(
                len(m.consolidated_from)
                for m in self.consolidated_memories
            )
            
            return {
                "total_memories": total_memories,
                "raw_memories": len(self.raw_memories),
                "consolidated_memories": len(self.consolidated_memories),
                "archived_memories": len(self.archived_memories),
                "consolidation_ratio": (
                    consolidated_count / total_memories if total_memories > 0 else 0
                ),
                "avg_importance": (
                    sum(m.importance_score for m in self.raw_memories) / len(self.raw_memories)
                    if self.raw_memories else 0
                )
            }
    
    
    def demonstrate_memory_consolidation():
        """
        Demonstrates memory consolidation with automatic processing,
        importance evaluation, and hierarchical organization.
        """
        print("=" * 80)
        print("MEMORY CONSOLIDATION DEMONSTRATION")
        print("=" * 80)
        
        # Create agent
        agent = MemoryConsolidationAgent()
        
        # Test 1: Add many memories to trigger consolidation
        print("\n" + "=" * 80)
        print("Test 1: Adding Memories (Will Trigger Consolidation)")
        print("=" * 80)
        
        # Add memories in groups
        python_memories = [
            ("Learned about Python list comprehensions", {"python", "programming"}),
            ("Practiced Python decorators", {"python", "programming"}),
            ("Studied Python generators", {"python", "programming"}),
        ]
        
        database_memories = [
            ("Learned SQL JOIN operations", {"sql", "database"}),
            ("Practiced database indexing", {"sql", "database"}),
            ("Studied database normalization", {"sql", "database"}),
        ]
        
        misc_memories = [
            ("Read about machine learning basics", {"ml", "learning"}),
            ("Explored neural networks", {"ml", "learning"}),
            ("Had coffee at 3pm", {"personal"}),
            ("Answered user questions about APIs", {"work", "api"}),
        ]
        
        all_memory_groups = [python_memories, database_memories, misc_memories]
        
        for memory_group in all_memory_groups:
            for content, tags in memory_group:
                agent.add_memory(content, tags=tags)
                print(f"✓ Added: {content}")
        
        # Show initial statistics
        stats = agent.get_statistics()
        print(f"\nInitial statistics:")
        print(f"  Raw memories: {stats['raw_memories']}")
        print(f"  Consolidated memories: {stats['consolidated_memories']}")
        
        # Test 2: Manual consolidation
        print("\n" + "=" * 80)
        print("Test 2: Manual Consolidation Trigger")
        print("=" * 80)
        
        result = agent.consolidate()
        print(f"\nConsolidation results:")
        print(f"  Memories processed: {result.memories_processed}")
        print(f"  Memories consolidated: {result.memories_consolidated}")
        print(f"  Memories pruned: {result.memories_pruned}")
        print(f"  New consolidated memories: {result.new_memories_created}")
        print(f"  Space saved: {result.space_saved:.1f}%")
        
        # Show updated statistics
        stats = agent.get_statistics()
        print(f"\nUpdated statistics:")
        print(f"  Total memories: {stats['total_memories']}")
        print(f"  Raw memories: {stats['raw_memories']}")
        print(f"  Consolidated memories: {stats['consolidated_memories']}")
        print(f"  Consolidation ratio: {stats['consolidation_ratio']:.2f}")
        print(f"  Average importance: {stats['avg_importance']:.2f}")
        
        # Test 3: Retrieve memories
        print("\n" + "=" * 80)
        print("Test 3: Memory Retrieval After Consolidation")
        print("=" * 80)
        
        queries = ["Python", "database", "machine learning"]
        
        for query in queries:
            print(f"\nQuery: '{query}'")
            memories = agent.retrieve_memories(query, max_results=3)
            print(f"Found {len(memories)} relevant memories:")
            for memory in memories:
                level_icon = "📝" if memory.level == MemoryLevel.RAW else "📚"
                print(f"  {level_icon} [{memory.importance_score:.2f}] {memory.content[:60]}...")
                if isinstance(memory, ConsolidatedMemory) and memory.consolidated_from:
                    print(f"      (Consolidated from {len(memory.consolidated_from)} memories)")
        
        # Summary
        print("\n" + "=" * 80)
        print("SUMMARY")
        print("=" * 80)
        print("""
    Memory Consolidation provides:
    ✓ Automatic memory organization and compression
    ✓ Importance-based retention
    ✓ Hierarchical memory structures
    ✓ Intelligent forgetting (pruning)
    ✓ Improved retrieval efficiency
    
    This pattern excels at:
    - Managing long-running agents with accumulating memory
    - Preventing memory overflow
    - Maintaining important information while discarding noise
    - Creating knowledge hierarchies
    - Balancing detail with efficiency
    
    Key consolidation operations:
    1. Importance Evaluation: Score memories based on multiple factors
    2. Grouping: Identify related memories
    3. Summarization: Create compressed representations
    4. Pruning: Remove low-value memories
    5. Archiving: Move old memories to long-term storage
    
    Triggering strategies:
    - Time-based: Periodic consolidation (every 24 hours)
    - Size-based: When memory count exceeds threshold (100 items)
    - Demand-based: Before retrieval operations
    
    Benefits:
    - Reduced memory footprint
    - Faster retrieval
    - Better organization
    - Preserved important information
    - Natural forgetting of irrelevant details
    
    Use memory consolidation when you need to:
    - Run agents for extended periods
    - Manage limited memory resources
    - Maintain retrieval performance
    - Build knowledge hierarchies
    - Implement realistic forgetting
    """)
    
    
    if __name__ == "__main__":
        demonstrate_memory_consolidation()
    
        

    Agentic Design Pattern: Memory Consolidation

    Pattern 031: Memory Consolidation – Line-by-Line Explanation

    Overview

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


    Pattern Description

    Pattern 031: Memory Consolidation

    Description: Memory consolidation processes and organizes memories over time through summarization, compression, and hierarchical structuring. This pattern prevents memory overflow, improves retrieval efficiency, and maintains the most important information while discarding or compressing less relevant details.

    Components: – Consolidation Scheduler: Triggers consolidation processes – Summarizer: Creates compressed representations of memories – Importance Evaluator: Scores memory significance – Memory Pruner: Removes or archives low-value memories – Hierarchical Organizer: Creates multi-level memory structures – Index Builder: Creates efficient retrieval structures

    Use Cases: – Long-running agents with accumulating memory – Managing limited memory resources – Improving retrieval speed – Creating knowledge hierarchies – Intelligent forgetting

    LangChain Implementation: Uses LLM-based summarization, importance scoring, and hierarchical organization to maintain an efficient and useful memory structure.


    File Statistics

    • Total Lines: 705
    • Classes: 9
    • Functions: 1
    • Constants: 0

    Module Documentation

    """
    Pattern 031: Memory Consolidation
    
    Description:
        Memory consolidation processes and organizes memories over time through
        summarization, compression, and hierarchical structuring. This pattern prevents
        memory overflow, improves retrieval efficiency, and maintains the most important
        information while discarding or compressing less relevant details.
    
    Components:
        - Consolidation Scheduler: Triggers consolidation processes
        - Summarizer: Creates compressed representations of memories
        - Importance Evaluator: Scores memory significance
        - Memory Pruner: Removes or archives low-value memories
        - Hierarchical Organizer: Creates multi-level memory structures
        - Index Builder: Creates efficient retrieval structures
    
    Use Cases:
        - Long-running agents with accumulating memory
        - Managing limited memory resources
        - Improving retrieval speed
        - Creating knowledge hierarchies
        - Intelligent forgetting
    
    LangChain Implementation:
        Uses LLM-based summarization, importance scoring, and hierarchical
        organization to maintain an efficient and useful memory structure.
    """
    

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


    Import Statements

    Standard Library Imports

    import collections.defaultdict
    import dataclasses.dataclass
    import dataclasses.field
    import datetime.datetime
    import datetime.timedelta
    import enum.Enum
    import json
    import langchain_core.output_parsers.StrOutputParser
    import langchain_core.prompts.ChatPromptTemplate
    import os
    import typing.Any
    import typing.Dict
    import typing.List
    import typing.Optional
    import typing.Set
    

    Purpose: Core Python functionality: – collections.defaultdict: Specialized container datatypes – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – datetime.datetime: Date and time handling – datetime.timedelta: Date and time handling – enum.Enum: Enumeration support – json: JSON serialization/deserialization – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: 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) – typing.Set: 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: MemoryLevel

    Lines 46-51

    Inherits from: Enum

    Purpose:

    Hierarchical levels of memory consolidation.
    

    Class: ImportanceCriterion

    Lines 54-61

    Inherits from: Enum

    Purpose:

    Criteria for evaluating memory importance.
    

    Class: Memory

    Lines 65-81

    Purpose:

    Individual memory unit.
    

    Methods:

    mark_accessed(self)

    Lines 78-81

    Purpose:

    Mark memory as accessed.
    

    Parameters:


    Class: ConsolidatedMemory

    Lines 85-92

    Inherits from: Memory

    Purpose:

    Consolidated memory containing summarized information.
    

    Methods:

    __post_init__(self)

    Lines 91-92

    Purpose: Post Init operation.

    Parameters:


    Class: ConsolidationResult

    Lines 96-103

    Purpose:

    Result of a consolidation operation.
    

    Class: ImportanceEvaluator

    Lines 106-179

    Purpose:

    Evaluates memory importance using multiple criteria.
    
    Combines various factors to produce a single importance score.
    

    Methods:

    __init__(self, llm)

    Lines 113-130

    Purpose: Initialize the class instance with required parameters.

    Parameters: – llm: Language model

    evaluate(self, memory, current_time, recency_weight, access_weight, llm_weight)

    Lines 132-162

    Purpose:

    Calculate importance score for a memory.
    
    Combines multiple factors with configurable weights.
    

    Parameters: – memory: Memory store for conversation history – current_time: Current time – recency_weight: Recency weight – access_weight: Access weight – llm_weight: Language model

    Returns: float

    _evaluate_semantic_importance(self, memory)

    Lines 164-179

    Purpose:

    Use LLM to evaluate semantic importance.
    

    Parameters: – memory: Memory store for conversation history

    Returns: float


    Class: MemoryConsolidator

    Lines 182-377

    Purpose:

    Performs memory consolidation operations.
    
    Capabilities:
    - Group related memories
    - Create summaries
    - Build hierarchies
    - Prune low-value memories
    

    Methods:

    __init__(self, llm)

    Lines 193-218

    Purpose: Initialize the class instance with required parameters.

    Parameters: – llm: Language model

    consolidate_memories(self, memories, importance_threshold, max_age_days)

    Lines 220-293

    Purpose:

    Perform full consolidation cycle on memories.
    
    Steps:
    1. Evaluate importance
    2. Group related memories
    3. Summarize groups
    4. Prune low-value memories
    5. Update memory levels
    

    Parameters: – memories: Memories – importance_threshold: Importance threshold – max_age_days: Max age days

    Returns: ConsolidationResult

    _group_memories(self, memories)

    Lines 295-335

    Purpose:

    Group related memories together.
    
    Uses tag similarity and temporal proximity.
    

    Parameters: – memories: Memories

    Returns: <ast.Subscript object at 0x000001A06D17D250>

    _create_consolidated_memory(self, group)

    Lines 337-377

    Purpose:

    Create a consolidated memory from a group.
    

    Parameters: – group: Group

    Returns: ConsolidatedMemory


    Class: ConsolidationScheduler

    Lines 380-425

    Purpose:

    Manages when and how consolidation is triggered.
    
    Supports multiple triggering strategies:
    - Time-based (periodic)
    - Size-based (memory threshold)
    - Demand-based (on query)
    

    Methods:

    __init__(self, consolidation_interval_hours, size_threshold, importance_threshold)

    Lines 390-399

    Purpose: Initialize the class instance with required parameters.

    Parameters: – consolidation_interval_hours: Consolidation interval hours – size_threshold: Size threshold – importance_threshold: Importance threshold

    should_consolidate(self, memory_count, time_since_last)

    Lines 401-421

    Purpose:

    Determine if consolidation should be triggered.
    

    Parameters: – memory_count: Memory store for conversation history – time_since_last: Time since last

    Returns: bool

    mark_consolidated(self)

    Lines 423-425

    Purpose:

    Mark that consolidation has occurred.
    

    Parameters:


    Class: MemoryConsolidationAgent

    Lines 428-564

    Purpose:

    Agent that manages memory with automatic consolidation.
    
    Features:
    - Automatic memory consolidation
    - Importance-based retention
    - Hierarchical memory organization
    - Efficient retrieval
    

    Methods:

    __init__(self, temperature)

    Lines 439-448

    Purpose: Initialize the class instance with required parameters.

    Parameters: – temperature: Sampling temperature (0-1)

    add_memory(self, content, tags, metadata)

    Lines 450-473

    Purpose:

    Add a new memory and check if consolidation is needed.
    

    Parameters: – content: Content – tags: Tags – metadata: Metadata

    Returns: str

    consolidate(self)

    Lines 475-502

    Purpose:

    Perform memory consolidation.
    
    This processes raw memories, creates consolidated versions,
    and prunes low-value memories.
    

    Parameters:

    Returns: ConsolidationResult

    retrieve_memories(self, query, include_consolidated, max_results)

    Lines 504-537

    Purpose:

    Retrieve relevant memories.
    
    Searches across raw and consolidated memories.
    

    Parameters: – query: Input query or question – include_consolidated: Include consolidated – max_results: Max results

    Returns: <ast.Subscript object at 0x000001A06D6140D0>

    get_statistics(self)

    Lines 539-564

    Purpose:

    Get memory statistics.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D645D90>


    Functions

    demonstrate_memory_consolidation()

    Lines 567-700

    Purpose:

    Demonstrates memory consolidation with automatic processing,
    importance evaluation, and hierarchical organization.
    

    Key Concepts Summary

    Pattern Overview

    The Memory Consolidation 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

    1. Agent Class: Main implementation of the pattern logic
    2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
    3. Tool Integration: Optional tools for extended functionality
    4. Memory Management: State handling across interactions
    5. 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

    1. Prompt Optimization: Minimize tokens while maintaining clarity
    2. Caching: Cache repeated queries/responses
    3. Model Selection: Use smaller models when appropriate
    4. Batch Processing: Group similar requests
    5. 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

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

    Feature Extensions

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

    Code Quality

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

    Related Patterns

    • Pattern 030
    • Pattern 032

    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 Memory Consolidation 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: 031_memory_consolidation.py
    Total Lines: 705
    Pattern Category: AI Agent Design Pattern #031


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Episodic Memory Retrieval

    
        """
    Pattern 030: Episodic Memory & Retrieval
    
    Description:
        Episodic memory stores and retrieves specific experiences or events with temporal
        and contextual information. This pattern enables agents to recall relevant past
        experiences based on similarity, temporal proximity, and contextual relevance,
        supporting experience-based reasoning and learning.
    
    Components:
        - Episode: Time-stamped experience with context, actions, and outcomes
        - Episode Store: Storage system for episodes with indexing
        - Retrieval System: Similarity-based and temporal filtering
        - Context Matcher: Identifies relevant episodes for current situation
        - Experience Reconstructor: Rebuilds complete episode details
    
    Use Cases:
        - Learning from past interactions
        - Experience-based decision making
        - Personalization through history
        - Debugging and analysis
        - Temporal reasoning
    
    LangChain Implementation:
        Uses vector similarity search for content-based retrieval combined with
        temporal and contextual filtering to find relevant past experiences.
    """
    
    import os
    from typing import List, Dict, Any, Optional, Tuple
    from dataclasses import dataclass, field
    from datetime import datetime, timedelta
    from enum import Enum
    import json
    import math
    from dotenv import load_dotenv
    
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser
    from langchain_openai import ChatOpenAI
    
    load_dotenv()
    
    
    class EpisodeType(Enum):
        """Types of episodes that can be stored."""
        CONVERSATION = "conversation"
        TASK_EXECUTION = "task_execution"
        DECISION = "decision"
        ERROR = "error"
        SUCCESS = "success"
        LEARNING = "learning"
        INTERACTION = "interaction"
    
    
    @dataclass
    class Episode:
        """
        Represents a single episodic memory.
        
        An episode captures a complete experience including context,
        actions taken, outcomes, and temporal information.
        """
        id: str
        type: EpisodeType
        timestamp: datetime
        context: Dict[str, Any]  # Situational context
        content: str  # Main content/description
        actions: List[str] = field(default_factory=list)  # Actions taken
        outcome: Optional[str] = None  # Result of the episode
        metadata: Dict[str, Any] = field(default_factory=dict)
        tags: List[str] = field(default_factory=list)
        importance: float = 0.5  # 0-1 scale
        access_count: int = 0
        last_accessed: Optional[datetime] = None
        
        def to_text(self) -> str:
            """Convert episode to searchable text representation."""
            parts = [
                f"Type: {self.type.value}",
                f"Content: {self.content}",
            ]
            
            if self.context:
                parts.append(f"Context: {json.dumps(self.context)}")
            
            if self.actions:
                parts.append(f"Actions: {', '.join(self.actions)}")
            
            if self.outcome:
                parts.append(f"Outcome: {self.outcome}")
            
            if self.tags:
                parts.append(f"Tags: {', '.join(self.tags)}")
            
            return " | ".join(parts)
        
        def mark_accessed(self):
            """Mark episode as accessed."""
            self.access_count += 1
            self.last_accessed = datetime.now()
    
    
    @dataclass
    class RetrievalQuery:
        """Query for retrieving episodes."""
        query: str
        episode_types: Optional[List[EpisodeType]] = None
        time_range: Optional[Tuple[datetime, datetime]] = None
        min_importance: float = 0.0
        max_results: int = 5
        tags: Optional[List[str]] = None
    
    
    @dataclass
    class RetrievalResult:
        """Result of episode retrieval with relevance scoring."""
        episode: Episode
        relevance_score: float  # Overall relevance (0-1)
        similarity_score: float  # Content similarity (0-1)
        recency_score: float  # Temporal relevance (0-1)
        importance_score: float  # Importance weight (0-1)
    
    
    class EpisodicMemoryStore:
        """
        Stores and retrieves episodic memories.
        
        Supports multiple retrieval strategies:
        - Similarity-based (content matching)
        - Temporal filtering (time-based)
        - Importance weighting
        - Tag-based filtering
        """
        
        def __init__(self, recency_decay_hours: float = 168.0):  # Default 1 week
            self.episodes: Dict[str, Episode] = {}
            self.recency_decay_hours = recency_decay_hours
            self._next_id = 1
        
        def add_episode(self, episode: Episode) -> str:
            """Add an episode to memory."""
            if not episode.id:
                episode.id = f"ep_{self._next_id:04d}"
                self._next_id += 1
            
            self.episodes[episode.id] = episode
            return episode.id
        
        def get_episode(self, episode_id: str) -> Optional[Episode]:
            """Retrieve a specific episode by ID."""
            episode = self.episodes.get(episode_id)
            if episode:
                episode.mark_accessed()
            return episode
        
        def retrieve(self, query: RetrievalQuery) -> List[RetrievalResult]:
            """
            Retrieve relevant episodes based on query.
            
            Combines multiple scoring factors:
            - Similarity to query content
            - Recency (temporal decay)
            - Importance weighting
            """
            candidates = self._filter_episodes(query)
            
            if not candidates:
                return []
            
            # Score each candidate
            results = []
            current_time = datetime.now()
            
            for episode in candidates:
                # Similarity score (simple keyword matching)
                similarity_score = self._compute_similarity(query.query, episode)
                
                # Recency score (exponential decay)
                recency_score = self._compute_recency(episode.timestamp, current_time)
                
                # Importance score
                importance_score = episode.importance
                
                # Combined relevance score (weighted average)
                relevance_score = (
                    0.5 * similarity_score +
                    0.3 * recency_score +
                    0.2 * importance_score
                )
                
                result = RetrievalResult(
                    episode=episode,
                    relevance_score=relevance_score,
                    similarity_score=similarity_score,
                    recency_score=recency_score,
                    importance_score=importance_score
                )
                results.append(result)
            
            # Sort by relevance and limit results
            results.sort(key=lambda x: x.relevance_score, reverse=True)
            
            # Mark accessed
            for result in results[:query.max_results]:
                result.episode.mark_accessed()
            
            return results[:query.max_results]
        
        def _filter_episodes(self, query: RetrievalQuery) -> List[Episode]:
            """Apply filters to episodes."""
            filtered = []
            
            for episode in self.episodes.values():
                # Type filter
                if query.episode_types and episode.type not in query.episode_types:
                    continue
                
                # Importance filter
                if episode.importance < query.min_importance:
                    continue
                
                # Time range filter
                if query.time_range:
                    start_time, end_time = query.time_range
                    if not (start_time <= episode.timestamp  float:
            """
            Compute similarity between query and episode.
            Simple keyword-based similarity (can be replaced with embeddings).
            """
            query_words = set(query.lower().split())
            episode_text = episode.to_text().lower()
            episode_words = set(episode_text.split())
            
            if not query_words:
                return 0.0
            
            # Jaccard similarity
            intersection = query_words & episode_words
            union = query_words | episode_words
            
            if not union:
                return 0.0
            
            return len(intersection) / len(union)
        
        def _compute_recency(self, timestamp: datetime, current_time: datetime) -> float:
            """
            Compute recency score with exponential decay.
            More recent episodes score higher.
            """
            hours_ago = (current_time - timestamp).total_seconds() / 3600
            
            if hours_ago  Dict[str, Any]:
            """Get statistics about episodic memory."""
            if not self.episodes:
                return {
                    "total_episodes": 0,
                    "episode_types": {},
                    "avg_importance": 0.0,
                    "most_accessed": None
                }
            
            type_counts = {}
            for episode in self.episodes.values():
                type_counts[episode.type.value] = type_counts.get(episode.type.value, 0) + 1
            
            avg_importance = sum(e.importance for e in self.episodes.values()) / len(self.episodes)
            
            most_accessed = max(self.episodes.values(), key=lambda e: e.access_count)
            
            return {
                "total_episodes": len(self.episodes),
                "episode_types": type_counts,
                "avg_importance": avg_importance,
                "most_accessed": {
                    "id": most_accessed.id,
                    "access_count": most_accessed.access_count,
                    "content": most_accessed.content[:50] + "..."
                }
            }
    
    
    class EpisodicMemoryAgent:
        """
        Agent that uses episodic memory to inform its responses.
        
        Capabilities:
        - Store experiences as episodes
        - Retrieve relevant past experiences
        - Learn from past successes and failures
        - Provide context-aware responses
        """
        
        def __init__(self, temperature: float = 0.7):
            self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=temperature)
            self.memory_store = EpisodicMemoryStore()
            
            # Prompt for responding with episodic context
            self.response_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are an AI assistant with episodic memory.
    You can remember past experiences and use them to inform your responses.
    
    Relevant past experiences:
    {episodes}
    
    Use these experiences to provide context-aware, informed responses.
    Reference past experiences when relevant."""),
                ("user", "{query}")
            ])
            
            # Prompt for extracting learnings
            self.learning_prompt = ChatPromptTemplate.from_messages([
                ("system", """Analyze this experience and extract key learnings.
    
    Experience: {experience}
    
    What are the key takeaways? What worked? What didn't?
    Provide 2-3 concise learnings."""),
                ("user", "Extract learnings")
            ])
        
        def record_episode(
            self,
            episode_type: EpisodeType,
            content: str,
            context: Optional[Dict[str, Any]] = None,
            actions: Optional[List[str]] = None,
            outcome: Optional[str] = None,
            importance: float = 0.5,
            tags: Optional[List[str]] = None
        ) -> str:
            """Record a new episodic memory."""
            episode = Episode(
                id="",  # Will be auto-generated
                type=episode_type,
                timestamp=datetime.now(),
                context=context or {},
                content=content,
                actions=actions or [],
                outcome=outcome,
                importance=importance,
                tags=tags or []
            )
            
            return self.memory_store.add_episode(episode)
        
        def respond_with_memory(self, query: str) -> Tuple[str, List[Episode]]:
            """
            Generate response using relevant episodic memories.
            
            Returns:
                Tuple of (response, relevant_episodes)
            """
            # Retrieve relevant episodes
            retrieval_query = RetrievalQuery(
                query=query,
                max_results=3
            )
            
            results = self.memory_store.retrieve(retrieval_query)
            
            # Format episodes for context
            if results:
                episodes_text = "\n\n".join([
                    f"Episode {i+1} (relevance: {r.relevance_score:.2f}):\n"
                    f"Type: {r.episode.type.value}\n"
                    f"Time: {r.episode.timestamp.strftime('%Y-%m-%d %H:%M')}\n"
                    f"Content: {r.episode.content}\n"
                    f"Outcome: {r.episode.outcome or 'N/A'}"
                    for i, r in enumerate(results)
                ])
            else:
                episodes_text = "No relevant past experiences found."
            
            # Generate response
            chain = self.response_prompt | self.llm | StrOutputParser()
            response = chain.invoke({
                "episodes": episodes_text,
                "query": query
            })
            
            # Extract episodes
            episodes = [r.episode for r in results]
            
            return response, episodes
        
        def learn_from_experience(self, episode_id: str) -> Optional[str]:
            """
            Extract learnings from a specific episode.
            """
            episode = self.memory_store.get_episode(episode_id)
            
            if not episode:
                return None
            
            # Format experience
            experience_text = (
                f"Type: {episode.type.value}\n"
                f"Context: {json.dumps(episode.context)}\n"
                f"Content: {episode.content}\n"
                f"Actions: {', '.join(episode.actions)}\n"
                f"Outcome: {episode.outcome or 'N/A'}"
            )
            
            # Extract learnings
            chain = self.learning_prompt | self.llm | StrOutputParser()
            learnings = chain.invoke({"experience": experience_text})
            
            return learnings
        
        def get_similar_experiences(
            self,
            reference_episode_id: str,
            max_results: int = 5
        ) -> List[RetrievalResult]:
            """
            Find episodes similar to a reference episode.
            """
            reference = self.memory_store.get_episode(reference_episode_id)
            
            if not reference:
                return []
            
            query = RetrievalQuery(
                query=reference.content,
                episode_types=[reference.type],
                max_results=max_results
            )
            
            results = self.memory_store.retrieve(query)
            
            # Exclude the reference episode itself
            return [r for r in results if r.episode.id != reference_episode_id]
        
        def get_recent_episodes(
            self,
            hours: int = 24,
            episode_type: Optional[EpisodeType] = None
        ) -> List[Episode]:
            """Get episodes from recent time period."""
            current_time = datetime.now()
            start_time = current_time - timedelta(hours=hours)
            
            query = RetrievalQuery(
                query="",  # Match all
                episode_types=[episode_type] if episode_type else None,
                time_range=(start_time, current_time),
                max_results=100  # Get all recent
            )
            
            results = self.memory_store.retrieve(query)
            return [r.episode for r in results]
    
    
    def demonstrate_episodic_memory():
        """
        Demonstrates episodic memory with storage, retrieval,
        and experience-based reasoning.
        """
        print("=" * 80)
        print("EPISODIC MEMORY & RETRIEVAL DEMONSTRATION")
        print("=" * 80)
        
        # Create agent
        agent = EpisodicMemoryAgent()
        
        # Test 1: Record various episodes
        print("\n" + "=" * 80)
        print("Test 1: Recording Episodes")
        print("=" * 80)
        
        episodes_to_record = [
            {
                "type": EpisodeType.CONVERSATION,
                "content": "User asked about Python programming best practices",
                "context": {"topic": "programming", "language": "Python"},
                "actions": ["Provided advice on PEP 8", "Recommended tools"],
                "outcome": "User found the advice helpful",
                "importance": 0.7,
                "tags": ["programming", "python", "advice"]
            },
            {
                "type": EpisodeType.TASK_EXECUTION,
                "content": "Helped user debug a database connection issue",
                "context": {"topic": "debugging", "database": "PostgreSQL"},
                "actions": ["Checked connection string", "Verified credentials", "Tested connection"],
                "outcome": "Successfully resolved the connection issue",
                "importance": 0.8,
                "tags": ["debugging", "database", "postgresql"]
            },
            {
                "type": EpisodeType.ERROR,
                "content": "Failed to parse user's malformed JSON input",
                "context": {"error_type": "JSONDecodeError"},
                "actions": ["Attempted to parse JSON", "Provided error message"],
                "outcome": "User corrected the JSON format",
                "importance": 0.6,
                "tags": ["error", "json", "parsing"]
            },
            {
                "type": EpisodeType.SUCCESS,
                "content": "Successfully explained machine learning concepts to beginner",
                "context": {"topic": "machine learning", "user_level": "beginner"},
                "actions": ["Used simple analogies", "Provided examples", "Answered follow-up questions"],
                "outcome": "User understood the concepts clearly",
                "importance": 0.9,
                "tags": ["teaching", "machine_learning", "success"]
            },
            {
                "type": EpisodeType.CONVERSATION,
                "content": "Discussed best practices for API design",
                "context": {"topic": "api design", "focus": "RESTful APIs"},
                "actions": ["Explained REST principles", "Discussed versioning", "Covered authentication"],
                "outcome": "User appreciated the comprehensive explanation",
                "importance": 0.75,
                "tags": ["api", "design", "rest"]
            }
        ]
        
        episode_ids = []
        for ep_data in episodes_to_record:
            ep_id = agent.record_episode(**ep_data)
            episode_ids.append(ep_id)
            print(f"✓ Recorded episode: {ep_data['content'][:50]}...")
        
        # Show statistics
        stats = agent.memory_store.get_statistics()
        print(f"\nMemory statistics:")
        print(f"  Total episodes: {stats['total_episodes']}")
        print(f"  Episode types: {stats['episode_types']}")
        print(f"  Average importance: {stats['avg_importance']:.2f}")
        
        # Test 2: Query with episodic memory
        print("\n" + "=" * 80)
        print("Test 2: Responding with Episodic Memory")
        print("=" * 80)
        
        queries = [
            "How should I approach learning programming?",
            "I'm having database connection problems, any advice?",
            "Can you explain machine learning to me?"
        ]
        
        for query in queries:
            print(f"\nUser: {query}")
            response, relevant_episodes = agent.respond_with_memory(query)
            print(f"\nAgent: {response}")
            print(f"\nRelevant past experiences: {len(relevant_episodes)}")
            for ep in relevant_episodes:
                print(f"  - {ep.type.value}: {ep.content[:60]}...")
        
        # Test 3: Learn from experiences
        print("\n" + "=" * 80)
        print("Test 3: Learning from Past Experiences")
        print("=" * 80)
        
        print(f"\nAnalyzing episode: {episode_ids[1]}")
        learnings = agent.learn_from_experience(episode_ids[1])
        if learnings:
            print(f"Learnings extracted:\n{learnings}")
        
        # Test 4: Find similar experiences
        print("\n" + "=" * 80)
        print("Test 4: Finding Similar Experiences")
        print("=" * 80)
        
        reference_id = episode_ids[0]
        print(f"\nFinding experiences similar to: {episode_ids[0]}")
        similar = agent.get_similar_experiences(reference_id, max_results=3)
        
        if similar:
            print(f"Found {len(similar)} similar experiences:")
            for result in similar:
                print(f"  - Relevance: {result.relevance_score:.2f}")
                print(f"    {result.episode.content[:70]}...")
        else:
            print("No similar experiences found")
        
        # Test 5: Recent activity
        print("\n" + "=" * 80)
        print("Test 5: Recent Activity Review")
        print("=" * 80)
        
        recent = agent.get_recent_episodes(hours=24)
        print(f"\nEpisodes from last 24 hours: {len(recent)}")
        for ep in recent[:3]:
            print(f"  - [{ep.timestamp.strftime('%H:%M')}] {ep.type.value}: {ep.content[:60]}...")
        
        # Summary
        print("\n" + "=" * 80)
        print("SUMMARY")
        print("=" * 80)
        print("""
    Episodic Memory & Retrieval provides:
    ✓ Time-stamped experience storage
    ✓ Multi-factor relevance scoring
    ✓ Context-aware retrieval
    ✓ Learning from past experiences
    ✓ Temporal and semantic filtering
    
    This pattern excels at:
    - Personalizing responses based on history
    - Learning from successes and failures
    - Providing context-aware recommendations
    - Supporting temporal reasoning
    - Building experiential knowledge
    
    Key advantages:
    1. Rich contextual information
    2. Temporal awareness
    3. Experience-based learning
    4. Flexible retrieval strategies
    5. Automatic relevance scoring
    
    Retrieval factors:
    - Similarity: Content matching (50% weight)
    - Recency: Temporal decay (30% weight)
    - Importance: Episode significance (20% weight)
    
    Use episodic memory when you need to:
    - Remember specific past interactions
    - Learn from historical outcomes
    - Provide personalized experiences
    - Support case-based reasoning
    - Track agent development over time
    """)
    
    
    if __name__ == "__main__":
        demonstrate_episodic_memory()
    
        

    Agentic Design Pattern: Episodic Memory Retrieval

    Pattern 030: Episodic Memory Retrieval – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Episodic Memory Retrieval pattern implementation.


    Pattern Description

    Pattern 030: Episodic Memory & Retrieval

    Description: Episodic memory stores and retrieves specific experiences or events with temporal and contextual information. This pattern enables agents to recall relevant past experiences based on similarity, temporal proximity, and contextual relevance, supporting experience-based reasoning and learning.

    Components: – Episode: Time-stamped experience with context, actions, and outcomes – Episode Store: Storage system for episodes with indexing – Retrieval System: Similarity-based and temporal filtering – Context Matcher: Identifies relevant episodes for current situation – Experience Reconstructor: Rebuilds complete episode details

    Use Cases: – Learning from past interactions – Experience-based decision making – Personalization through history – Debugging and analysis – Temporal reasoning

    LangChain Implementation: Uses vector similarity search for content-based retrieval combined with temporal and contextual filtering to find relevant past experiences.


    File Statistics

    • Total Lines: 651
    • Classes: 6
    • Functions: 1
    • Constants: 0

    Module Documentation

    """
    Pattern 030: Episodic Memory & Retrieval
    
    Description:
        Episodic memory stores and retrieves specific experiences or events with temporal
        and contextual information. This pattern enables agents to recall relevant past
        experiences based on similarity, temporal proximity, and contextual relevance,
        supporting experience-based reasoning and learning.
    
    Components:
        - Episode: Time-stamped experience with context, actions, and outcomes
        - Episode Store: Storage system for episodes with indexing
        - Retrieval System: Similarity-based and temporal filtering
        - Context Matcher: Identifies relevant episodes for current situation
        - Experience Reconstructor: Rebuilds complete episode details
    
    Use Cases:
        - Learning from past interactions
        - Experience-based decision making
        - Personalization through history
        - Debugging and analysis
        - Temporal reasoning
    
    LangChain Implementation:
        Uses vector similarity search for content-based retrieval combined with
        temporal and contextual filtering to find relevant past experiences.
    """
    

    Purpose: This module implements the Episodic Memory Retrieval 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 datetime.timedelta
    import enum.Enum
    import json
    import langchain_core.output_parsers.StrOutputParser
    import langchain_core.prompts.ChatPromptTemplate
    import os
    import typing.Any
    import typing.Dict
    import typing.List
    import typing.Optional
    import typing.Tuple
    

    Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – datetime.datetime: Date and time handling – datetime.timedelta: Date and time handling – enum.Enum: Enumeration support – json: JSON serialization/deserialization – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: 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) – typing.Tuple: 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
    from math
    

    Purpose: Additional dependencies for specialized functionality.


    Classes

    Class: EpisodeType

    Lines 45-53

    Inherits from: Enum

    Purpose:

    Types of episodes that can be stored.
    

    Class: Episode

    Lines 57-101

    Purpose:

    Represents a single episodic memory.
    
    An episode captures a complete experience including context,
    actions taken, outcomes, and temporal information.
    

    Methods:

    to_text(self)

    Lines 77-96

    Purpose:

    Convert episode to searchable text representation.
    

    Parameters:

    Returns: str

    mark_accessed(self)

    Lines 98-101

    Purpose:

    Mark episode as accessed.
    

    Parameters:


    Class: RetrievalQuery

    Lines 105-112

    Purpose:

    Query for retrieving episodes.
    

    Class: RetrievalResult

    Lines 116-122

    Purpose:

    Result of episode retrieval with relevance scoring.
    

    Class: EpisodicMemoryStore

    Lines 125-300

    Purpose:

    Stores and retrieves episodic memories.
    
    Supports multiple retrieval strategies:
    - Similarity-based (content matching)
    - Temporal filtering (time-based)
    - Importance weighting
    - Tag-based filtering
    

    Methods:

    __init__(self, recency_decay_hours)

    Lines 136-139

    Purpose: Initialize the class instance with required parameters.

    Parameters: – recency_decay_hours: Recency decay hours

    add_episode(self, episode)

    Lines 141-148

    Purpose:

    Add an episode to memory.
    

    Parameters: – episode: Episode

    Returns: str

    get_episode(self, episode_id)

    Lines 150-155

    Purpose:

    Retrieve a specific episode by ID.
    

    Parameters: – episode_id: Episode id

    Returns: <ast.Subscript object at 0x000001A06D6A9950>

    retrieve(self, query)

    Lines 157-208

    Purpose:

    Retrieve relevant episodes based on query.
    
    Combines multiple scoring factors:
    - Similarity to query content
    - Recency (temporal decay)
    - Importance weighting
    

    Parameters: – query: Input query or question

    Returns: <ast.Subscript object at 0x000001A06D6F6AD0>

    _filter_episodes(self, query)

    Lines 210-236

    Purpose:

    Apply filters to episodes.
    

    Parameters: – query: Input query or question

    Returns: <ast.Subscript object at 0x000001A06D62B6D0>

    _compute_similarity(self, query, episode)

    Lines 238-257

    Purpose:

    Compute similarity between query and episode.
    Simple keyword-based similarity (can be replaced with embeddings).
    

    Parameters: – query: Input query or question – episode: Episode

    Returns: float

    _compute_recency(self, timestamp, current_time)

    Lines 259-271

    Purpose:

    Compute recency score with exponential decay.
    More recent episodes score higher.
    

    Parameters: – timestamp: Timestamp – current_time: Current time

    Returns: float

    get_statistics(self)

    Lines 273-300

    Purpose:

    Get statistics about episodic memory.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D6066D0>


    Class: EpisodicMemoryAgent

    Lines 303-472

    Purpose:

    Agent that uses episodic memory to inform its responses.
    
    Capabilities:
    - Store experiences as episodes
    - Retrieve relevant past experiences
    - Learn from past successes and failures
    - Provide context-aware responses
    

    Methods:

    __init__(self, temperature)

    Lines 314-340

    Purpose: Initialize the class instance with required parameters.

    Parameters: – temperature: Sampling temperature (0-1)

    record_episode(self, episode_type, content, context, actions, outcome, importance, tags)

    Lines 342-365

    Purpose:

    Record a new episodic memory.
    

    Parameters: – episode_type: Episode type – content: Content – context: Context – actions: Actions – outcome: Outcome – importance: Importance – tags: Tags

    Returns: str

    respond_with_memory(self, query)

    Lines 367-405

    Purpose:

    Generate response using relevant episodic memories.
    
    Returns:
        Tuple of (response, relevant_episodes)
    

    Parameters: – query: Input query or question

    Returns: <ast.Subscript object at 0x000001A06D156E50>

    learn_from_experience(self, episode_id)

    Lines 407-429

    Purpose:

    Extract learnings from a specific episode.
    

    Parameters: – episode_id: Episode id

    Returns: <ast.Subscript object at 0x000001A06D72A890>

    get_similar_experiences(self, reference_episode_id, max_results)

    Lines 431-453

    Purpose:

    Find episodes similar to a reference episode.
    

    Parameters: – reference_episode_id: Reference episode id – max_results: Max results

    Returns: <ast.Subscript object at 0x000001A06D72B850>

    get_recent_episodes(self, hours, episode_type)

    Lines 455-472

    Purpose:

    Get episodes from recent time period.
    

    Parameters: – hours: Hours – episode_type: Episode type

    Returns: <ast.Subscript object at 0x000001A06D169750>


    Functions

    demonstrate_episodic_memory()

    Lines 475-646

    Purpose:

    Demonstrates episodic memory with storage, retrieval,
    and experience-based reasoning.
    

    Key Concepts Summary

    Pattern Overview

    The Episodic Memory Retrieval 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

    1. Agent Class: Main implementation of the pattern logic
    2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
    3. Tool Integration: Optional tools for extended functionality
    4. Memory Management: State handling across interactions
    5. 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

    1. Prompt Optimization: Minimize tokens while maintaining clarity
    2. Caching: Cache repeated queries/responses
    3. Model Selection: Use smaller models when appropriate
    4. Batch Processing: Group similar requests
    5. 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

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

    Feature Extensions

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

    Code Quality

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

    Related Patterns

    • Pattern 029
    • Pattern 031

    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 Episodic Memory Retrieval 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: 030_episodic_memory_retrieval.py
    Total Lines: 651
    Pattern Category: AI Agent Design Pattern #030


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Semantic Memory Networks

    
        """
    Pattern 029: Semantic Memory Networks
    
    Description:
        Semantic memory networks represent structured knowledge as graphs with entities,
        relationships, and attributes. This pattern enables sophisticated reasoning over
        interconnected concepts, supports graph-based queries, and facilitates knowledge
        discovery through relationship traversal.
    
    Components:
        - Entity: Nodes representing concepts, objects, or individuals
        - Relationship: Edges connecting entities with typed connections
        - Attributes: Properties associated with entities
        - Graph Store: Storage and indexing of the knowledge graph
        - Query Engine: Traversal and pattern matching over the graph
        - Reasoning Module: Inference and relationship discovery
    
    Use Cases:
        - Knowledge management systems
        - Question answering over structured knowledge
        - Relationship discovery and analysis
        - Semantic search and retrieval
        - Expert systems and decision support
    
    LangChain Implementation:
        Uses graph data structures, relationship modeling, and LLM-based query
        interpretation to create a semantic network that supports both explicit
        knowledge storage and implicit reasoning.
    """
    
    import os
    from typing import List, Dict, Any, Optional, Set, Tuple
    from dataclasses import dataclass, field
    from enum import Enum
    from datetime import datetime
    from collections import defaultdict
    import json
    from dotenv import load_dotenv
    
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser
    from langchain_openai import ChatOpenAI
    
    load_dotenv()
    
    
    class RelationType(Enum):
        """Types of relationships between entities."""
        IS_A = "is_a"  # Inheritance/taxonomy
        PART_OF = "part_of"  # Composition
        RELATED_TO = "related_to"  # General association
        CAUSES = "causes"  # Causal relationship
        PRECEDES = "precedes"  # Temporal relationship
        LOCATED_IN = "located_in"  # Spatial relationship
        OWNS = "owns"  # Ownership
        CREATED_BY = "created_by"  # Authorship
        USES = "uses"  # Functional relationship
        SIMILAR_TO = "similar_to"  # Similarity
    
    
    @dataclass
    class Entity:
        """Represents a node in the semantic network."""
        id: str
        name: str
        type: str
        attributes: Dict[str, Any] = field(default_factory=dict)
        created_at: datetime = field(default_factory=datetime.now)
        
        def __hash__(self):
            return hash(self.id)
        
        def __eq__(self, other):
            return isinstance(other, Entity) and self.id == other.id
    
    
    @dataclass
    class Relationship:
        """Represents an edge in the semantic network."""
        source_id: str
        target_id: str
        type: RelationType
        attributes: Dict[str, Any] = field(default_factory=dict)
        weight: float = 1.0
        created_at: datetime = field(default_factory=datetime.now)
    
    
    @dataclass
    class QueryResult:
        """Result of a semantic query."""
        entities: List[Entity]
        relationships: List[Relationship]
        paths: List[List[str]] = field(default_factory=list)
        explanation: str = ""
    
    
    class SemanticGraph:
        """
        Graph data structure for storing semantic knowledge.
        
        Supports efficient entity lookup, relationship traversal,
        and pattern matching operations.
        """
        
        def __init__(self):
            self.entities: Dict[str, Entity] = {}
            self.relationships: List[Relationship] = []
            # Adjacency lists for efficient traversal
            self.outgoing: Dict[str, List[Relationship]] = defaultdict(list)
            self.incoming: Dict[str, List[Relationship]] = defaultdict(list)
        
        def add_entity(self, entity: Entity) -> None:
            """Add an entity to the graph."""
            self.entities[entity.id] = entity
        
        def add_relationship(self, relationship: Relationship) -> None:
            """Add a relationship to the graph."""
            # Validate entities exist
            if relationship.source_id not in self.entities:
                raise ValueError(f"Source entity {relationship.source_id} not found")
            if relationship.target_id not in self.entities:
                raise ValueError(f"Target entity {relationship.target_id} not found")
            
            self.relationships.append(relationship)
            self.outgoing[relationship.source_id].append(relationship)
            self.incoming[relationship.target_id].append(relationship)
        
        def get_entity(self, entity_id: str) -> Optional[Entity]:
            """Retrieve an entity by ID."""
            return self.entities.get(entity_id)
        
        def get_related_entities(
            self,
            entity_id: str,
            relation_type: Optional[RelationType] = None,
            direction: str = "outgoing"
        ) -> List[Entity]:
            """
            Get entities related to the given entity.
            
            Args:
                entity_id: Source entity ID
                relation_type: Optional filter by relationship type
                direction: "outgoing", "incoming", or "both"
            """
            related = []
            
            if direction in ["outgoing", "both"]:
                for rel in self.outgoing[entity_id]:
                    if relation_type is None or rel.type == relation_type:
                        related.append(self.entities[rel.target_id])
            
            if direction in ["incoming", "both"]:
                for rel in self.incoming[entity_id]:
                    if relation_type is None or rel.type == relation_type:
                        related.append(self.entities[rel.source_id])
            
            return related
        
        def find_path(
            self,
            start_id: str,
            end_id: str,
            max_depth: int = 5
        ) -> Optional[List[str]]:
            """
            Find a path between two entities using BFS.
            
            Returns list of entity IDs forming the path, or None if no path exists.
            """
            if start_id not in self.entities or end_id not in self.entities:
                return None
            
            if start_id == end_id:
                return [start_id]
            
            # BFS
            queue = [(start_id, [start_id])]
            visited = {start_id}
            
            while queue:
                current_id, path = queue.pop(0)
                
                if len(path) > max_depth:
                    continue
                
                # Explore neighbors
                for rel in self.outgoing[current_id]:
                    neighbor_id = rel.target_id
                    
                    if neighbor_id == end_id:
                        return path + [neighbor_id]
                    
                    if neighbor_id not in visited:
                        visited.add(neighbor_id)
                        queue.append((neighbor_id, path + [neighbor_id]))
            
            return None
        
        def get_subgraph(
            self,
            entity_ids: List[str],
            depth: int = 1
        ) -> Tuple[List[Entity], List[Relationship]]:
            """
            Extract a subgraph around given entities up to specified depth.
            """
            included_entities = set(entity_ids)
            
            # Expand to neighbors
            for _ in range(depth):
                new_entities = set()
                for entity_id in included_entities:
                    neighbors = self.get_related_entities(entity_id, direction="both")
                    new_entities.update(e.id for e in neighbors)
                included_entities.update(new_entities)
            
            # Get all entities
            entities = [self.entities[eid] for eid in included_entities if eid in self.entities]
            
            # Get relationships between included entities
            relationships = [
                rel for rel in self.relationships
                if rel.source_id in included_entities and rel.target_id in included_entities
            ]
            
            return entities, relationships
        
        def search_entities(
            self,
            query: str,
            entity_type: Optional[str] = None
        ) -> List[Entity]:
            """
            Search entities by name or attributes.
            Simple keyword-based search.
            """
            query_lower = query.lower()
            results = []
            
            for entity in self.entities.values():
                # Filter by type if specified
                if entity_type and entity.type != entity_type:
                    continue
                
                # Check name
                if query_lower in entity.name.lower():
                    results.append(entity)
                    continue
                
                # Check attributes
                for value in entity.attributes.values():
                    if isinstance(value, str) and query_lower in value.lower():
                        results.append(entity)
                        break
            
            return results
    
    
    class SemanticMemoryAgent:
        """
        Agent that maintains and reasons over a semantic memory network.
        
        Capabilities:
        - Extract entities and relationships from text
        - Store knowledge in graph structure
        - Answer questions using graph traversal
        - Discover implicit relationships
        """
        
        def __init__(self, temperature: float = 0.1):
            self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=temperature)
            self.graph = SemanticGraph()
            
            # Prompt for entity extraction
            self.extraction_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are an expert at extracting structured knowledge from text.
    Extract entities and relationships from the following text.
    
    Format your response as JSON:
    {{
      "entities": [
        {{"id": "unique_id", "name": "Entity Name", "type": "entity_type", "attributes": {{}}}},
        ...
      ],
      "relationships": [
        {{"source": "source_id", "target": "target_id", "type": "relationship_type"}},
        ...
      ]
    }}
    
    Valid relationship types: is_a, part_of, related_to, causes, precedes, located_in, owns, created_by, uses, similar_to
    
    Be specific and extract all meaningful entities and relationships."""),
                ("user", "{text}")
            ])
            
            # Prompt for query understanding
            self.query_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are an expert at answering questions using a knowledge graph.
    
    Given the following knowledge graph elements:
    {graph_context}
    
    Answer this question: {question}
    
    Provide a clear, concise answer based on the graph information."""),
                ("user", "{question}")
            ])
        
        def add_knowledge(self, text: str) -> Dict[str, Any]:
            """
            Extract and add knowledge from text to the semantic network.
            """
            # Extract entities and relationships
            chain = self.extraction_prompt | self.llm | StrOutputParser()
            result = chain.invoke({"text": text})
            
            # Parse JSON response
            try:
                # Try to extract JSON from response
                start_idx = result.find('{')
                end_idx = result.rfind('}') + 1
                if start_idx != -1 and end_idx != 0:
                    json_str = result[start_idx:end_idx]
                    knowledge = json.loads(json_str)
                else:
                    knowledge = json.loads(result)
            except json.JSONDecodeError:
                # If parsing fails, return error
                return {
                    "success": False,
                    "error": "Failed to parse extracted knowledge",
                    "raw_response": result
                }
            
            # Add entities
            entities_added = 0
            for entity_data in knowledge.get("entities", []):
                entity = Entity(
                    id=entity_data["id"],
                    name=entity_data["name"],
                    type=entity_data["type"],
                    attributes=entity_data.get("attributes", {})
                )
                self.graph.add_entity(entity)
                entities_added += 1
            
            # Add relationships
            relationships_added = 0
            for rel_data in knowledge.get("relationships", []):
                try:
                    # Map string to enum
                    rel_type_str = rel_data["type"].upper()
                    rel_type = RelationType[rel_type_str]
                    
                    relationship = Relationship(
                        source_id=rel_data["source"],
                        target_id=rel_data["target"],
                        type=rel_type
                    )
                    self.graph.add_relationship(relationship)
                    relationships_added += 1
                except (KeyError, ValueError) as e:
                    # Skip invalid relationships
                    continue
            
            return {
                "success": True,
                "entities_added": entities_added,
                "relationships_added": relationships_added
            }
        
        def query(self, question: str) -> str:
            """
            Answer a question using the semantic network.
            """
            # Search for relevant entities in the question
            words = question.lower().split()
            relevant_entities = set()
            
            for word in words:
                entities = self.graph.search_entities(word)
                relevant_entities.update(e.id for e in entities)
            
            if not relevant_entities:
                return "I don't have enough knowledge to answer that question."
            
            # Get subgraph around relevant entities
            entities, relationships = self.graph.get_subgraph(list(relevant_entities), depth=2)
            
            # Format graph context
            graph_context = self._format_graph_context(entities, relationships)
            
            # Generate answer
            chain = self.query_prompt | self.llm | StrOutputParser()
            answer = chain.invoke({
                "graph_context": graph_context,
                "question": question
            })
            
            return answer
        
        def find_relationship(
            self,
            entity1: str,
            entity2: str
        ) -> Optional[List[str]]:
            """
            Find how two entities are related through the graph.
            """
            # Find entities by name
            entities1 = self.graph.search_entities(entity1)
            entities2 = self.graph.search_entities(entity2)
            
            if not entities1 or not entities2:
                return None
            
            # Find shortest path
            path = self.graph.find_path(entities1[0].id, entities2[0].id)
            
            if path:
                # Convert IDs to names
                return [self.graph.get_entity(eid).name for eid in path]
            
            return None
        
        def get_related_concepts(
            self,
            concept: str,
            relation_type: Optional[str] = None
        ) -> List[str]:
            """
            Get concepts related to the given concept.
            """
            entities = self.graph.search_entities(concept)
            
            if not entities:
                return []
            
            # Convert relation type string to enum if provided
            rel_type_enum = None
            if relation_type:
                try:
                    rel_type_enum = RelationType[relation_type.upper()]
                except KeyError:
                    pass
            
            # Get related entities
            related = self.graph.get_related_entities(
                entities[0].id,
                relation_type=rel_type_enum,
                direction="both"
            )
            
            return [e.name for e in related]
        
        def _format_graph_context(
            self,
            entities: List[Entity],
            relationships: List[Relationship]
        ) -> str:
            """Format graph elements as readable context."""
            lines = ["Entities:"]
            for entity in entities[:10]:  # Limit to avoid token overflow
                attrs_str = ", ".join(f"{k}={v}" for k, v in entity.attributes.items())
                lines.append(f"  - {entity.name} ({entity.type})" + (f": {attrs_str}" if attrs_str else ""))
            
            lines.append("\nRelationships:")
            for rel in relationships[:20]:  # Limit relationships
                source = self.graph.get_entity(rel.source_id)
                target = self.graph.get_entity(rel.target_id)
                if source and target:
                    lines.append(f"  - {source.name} {rel.type.value} {target.name}")
            
            return "\n".join(lines)
        
        def get_statistics(self) -> Dict[str, Any]:
            """Get statistics about the semantic network."""
            entity_types = defaultdict(int)
            relation_types = defaultdict(int)
            
            for entity in self.graph.entities.values():
                entity_types[entity.type] += 1
            
            for rel in self.graph.relationships:
                relation_types[rel.type.value] += 1
            
            return {
                "total_entities": len(self.graph.entities),
                "total_relationships": len(self.graph.relationships),
                "entity_types": dict(entity_types),
                "relation_types": dict(relation_types)
            }
    
    
    def demonstrate_semantic_memory_networks():
        """
        Demonstrates semantic memory networks with knowledge extraction,
        graph-based queries, and relationship discovery.
        """
        print("=" * 80)
        print("SEMANTIC MEMORY NETWORKS DEMONSTRATION")
        print("=" * 80)
        
        # Create agent
        agent = SemanticMemoryAgent()
        
        # Test 1: Build knowledge base
        print("\n" + "=" * 80)
        print("Test 1: Building Knowledge Base")
        print("=" * 80)
        
        knowledge_texts = [
            """
            Python is a programming language created by Guido van Rossum.
            Python is used for web development, data analysis, and artificial intelligence.
            Python is similar to Ruby and JavaScript in terms of ease of use.
            """,
            """
            Machine Learning is a part of Artificial Intelligence.
            Machine Learning uses algorithms to analyze data and make predictions.
            TensorFlow and PyTorch are popular Machine Learning frameworks.
            TensorFlow was created by Google.
            """,
            """
            San Francisco is located in California.
            California is part of the United States.
            San Francisco is known for its technology industry.
            Many tech companies are located in San Francisco.
            """
        ]
        
        for i, text in enumerate(knowledge_texts, 1):
            print(f"\nAdding knowledge batch {i}...")
            result = agent.add_knowledge(text)
            if result["success"]:
                print(f"✓ Added {result['entities_added']} entities and {result['relationships_added']} relationships")
            else:
                print(f"✗ Error: {result.get('error', 'Unknown error')}")
        
        # Show statistics
        stats = agent.get_statistics()
        print(f"\nKnowledge base statistics:")
        print(f"  Total entities: {stats['total_entities']}")
        print(f"  Total relationships: {stats['total_relationships']}")
        print(f"  Entity types: {stats['entity_types']}")
        print(f"  Relation types: {stats['relation_types']}")
        
        # Test 2: Query knowledge
        print("\n" + "=" * 80)
        print("Test 2: Querying Knowledge Base")
        print("=" * 80)
        
        questions = [
            "What is Python used for?",
            "Who created Python?",
            "What are some Machine Learning frameworks?",
            "Where is San Francisco located?"
        ]
        
        for question in questions:
            print(f"\nQuestion: {question}")
            answer = agent.query(question)
            print(f"Answer: {answer}")
        
        # Test 3: Relationship discovery
        print("\n" + "=" * 80)
        print("Test 3: Discovering Relationships")
        print("=" * 80)
        
        concept_pairs = [
            ("Python", "Artificial Intelligence"),
            ("TensorFlow", "Google"),
            ("San Francisco", "United States")
        ]
        
        for concept1, concept2 in concept_pairs:
            print(f"\nFinding relationship between '{concept1}' and '{concept2}':")
            path = agent.find_relationship(concept1, concept2)
            if path:
                print(f"  Path: {' → '.join(path)}")
            else:
                print("  No direct relationship found")
        
        # Test 4: Related concepts
        print("\n" + "=" * 80)
        print("Test 4: Finding Related Concepts")
        print("=" * 80)
        
        concepts = ["Python", "Machine Learning", "San Francisco"]
        
        for concept in concepts:
            print(f"\nConcepts related to '{concept}':")
            related = agent.get_related_concepts(concept)
            if related:
                for rel_concept in related[:5]:  # Show top 5
                    print(f"  - {rel_concept}")
            else:
                print("  No related concepts found")
        
        # Summary
        print("\n" + "=" * 80)
        print("SUMMARY")
        print("=" * 80)
        print("""
    Semantic Memory Networks provide:
    ✓ Structured knowledge representation as graphs
    ✓ Entity and relationship extraction from text
    ✓ Graph-based querying and reasoning
    ✓ Relationship discovery through path finding
    ✓ Concept exploration through traversal
    
    This pattern excels at:
    - Knowledge management and organization
    - Question answering over structured data
    - Discovering implicit connections
    - Semantic search and retrieval
    - Building expert systems
    
    Key advantages:
    1. Explicit relationship modeling
    2. Efficient graph traversal
    3. Multi-hop reasoning support
    4. Scalable knowledge storage
    5. Interpretable knowledge representation
    
    Use semantic networks when you need to:
    - Maintain structured knowledge bases
    - Answer complex relational queries
    - Discover connections between concepts
    - Build domain-specific knowledge systems
    - Support inference and reasoning tasks
    """)
    
    
    if __name__ == "__main__":
        demonstrate_semantic_memory_networks()
    
        

    Agentic Design Pattern: Semantic Memory Networks

    Pattern 029: Semantic Memory Networks – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Semantic Memory Networks pattern implementation.


    Pattern Description

    Pattern 029: Semantic Memory Networks

    Description: Semantic memory networks represent structured knowledge as graphs with entities, relationships, and attributes. This pattern enables sophisticated reasoning over interconnected concepts, supports graph-based queries, and facilitates knowledge discovery through relationship traversal.

    Components: – Entity: Nodes representing concepts, objects, or individuals – Relationship: Edges connecting entities with typed connections – Attributes: Properties associated with entities – Graph Store: Storage and indexing of the knowledge graph – Query Engine: Traversal and pattern matching over the graph – Reasoning Module: Inference and relationship discovery

    Use Cases: – Knowledge management systems – Question answering over structured knowledge – Relationship discovery and analysis – Semantic search and retrieval – Expert systems and decision support

    LangChain Implementation: Uses graph data structures, relationship modeling, and LLM-based query interpretation to create a semantic network that supports both explicit knowledge storage and implicit reasoning.


    File Statistics

    • Total Lines: 639
    • Classes: 6
    • Functions: 1
    • Constants: 0

    Module Documentation

    """
    Pattern 029: Semantic Memory Networks
    
    Description:
        Semantic memory networks represent structured knowledge as graphs with entities,
        relationships, and attributes. This pattern enables sophisticated reasoning over
        interconnected concepts, supports graph-based queries, and facilitates knowledge
        discovery through relationship traversal.
    
    Components:
        - Entity: Nodes representing concepts, objects, or individuals
        - Relationship: Edges connecting entities with typed connections
        - Attributes: Properties associated with entities
        - Graph Store: Storage and indexing of the knowledge graph
        - Query Engine: Traversal and pattern matching over the graph
        - Reasoning Module: Inference and relationship discovery
    
    Use Cases:
        - Knowledge management systems
        - Question answering over structured knowledge
        - Relationship discovery and analysis
        - Semantic search and retrieval
        - Expert systems and decision support
    
    LangChain Implementation:
        Uses graph data structures, relationship modeling, and LLM-based query
        interpretation to create a semantic network that supports both explicit
        knowledge storage and implicit reasoning.
    """
    

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


    Import Statements

    Standard Library Imports

    import collections.defaultdict
    import dataclasses.dataclass
    import dataclasses.field
    import datetime.datetime
    import enum.Enum
    import json
    import langchain_core.output_parsers.StrOutputParser
    import langchain_core.prompts.ChatPromptTemplate
    import os
    import typing.Any
    import typing.Dict
    import typing.List
    import typing.Optional
    import typing.Set
    import typing.Tuple
    

    Purpose: Core Python functionality: – collections.defaultdict: Specialized container datatypes – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – datetime.datetime: Date and time handling – enum.Enum: Enumeration support – json: JSON serialization/deserialization – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: 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) – typing.Set: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Tuple: 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: RelationType

    Lines 47-58

    Inherits from: Enum

    Purpose:

    Types of relationships between entities.
    

    Class: Entity

    Lines 62-74

    Purpose:

    Represents a node in the semantic network.
    

    Methods:

    __hash__(self)

    Lines 70-71

    Purpose: Hash operation.

    Parameters:

    __eq__(self, other)

    Lines 73-74

    Purpose: Eq operation.

    Parameters: – other: Other


    Class: Relationship

    Lines 78-85

    Purpose:

    Represents an edge in the semantic network.
    

    Class: QueryResult

    Lines 89-94

    Purpose:

    Result of a semantic query.
    

    Class: SemanticGraph

    Lines 97-257

    Purpose:

    Graph data structure for storing semantic knowledge.
    
    Supports efficient entity lookup, relationship traversal,
    and pattern matching operations.
    

    Methods:

    __init__(self)

    Lines 105-110

    Purpose: Initialize the class instance with required parameters.

    Parameters:

    add_entity(self, entity)

    Lines 112-114

    Purpose:

    Add an entity to the graph.
    

    Parameters: – entity: Entity

    Returns: None

    add_relationship(self, relationship)

    Lines 116-126

    Purpose:

    Add a relationship to the graph.
    

    Parameters: – relationship: Relationship

    Returns: None

    get_entity(self, entity_id)

    Lines 128-130

    Purpose:

    Retrieve an entity by ID.
    

    Parameters: – entity_id: Entity id

    Returns: <ast.Subscript object at 0x000001A06D749A10>

    get_related_entities(self, entity_id, relation_type, direction)

    Lines 132-158

    Purpose:

    Get entities related to the given entity.
    
    Args:
        entity_id: Source entity ID
        relation_type: Optional filter by relationship type
        direction: "outgoing", "incoming", or "both"
    

    Parameters: – entity_id: Entity id – relation_type: Relation type – direction: Direction

    Returns: <ast.Subscript object at 0x000001A06D74BC50>

    find_path(self, start_id, end_id, max_depth)

    Lines 160-198

    Purpose:

    Find a path between two entities using BFS.
    
    Returns list of entity IDs forming the path, or None if no path exists.
    

    Parameters: – start_id: Start id – end_id: End id – max_depth: Max depth

    Returns: <ast.Subscript object at 0x000001A06D751A10>

    get_subgraph(self, entity_ids, depth)

    Lines 200-227

    Purpose:

    Extract a subgraph around given entities up to specified depth.
    

    Parameters: – entity_ids: Entity ids – depth: Depth

    Returns: <ast.Subscript object at 0x000001A06D615850>

    search_entities(self, query, entity_type)

    Lines 229-257

    Purpose:

    Search entities by name or attributes.
    Simple keyword-based search.
    

    Parameters: – query: Input query or question – entity_type: Entity type

    Returns: <ast.Subscript object at 0x000001A06D620F90>


    Class: SemanticMemoryAgent

    Lines 260-494

    Purpose:

    Agent that maintains and reasons over a semantic memory network.
    
    Capabilities:
    - Extract entities and relationships from text
    - Store knowledge in graph structure
    - Answer questions using graph traversal
    - Discover implicit relationships
    

    Methods:

    __init__(self, temperature)

    Lines 271-309

    Purpose: Initialize the class instance with required parameters.

    Parameters: – temperature: Sampling temperature (0-1)

    add_knowledge(self, text)

    Lines 311-372

    Purpose:

    Extract and add knowledge from text to the semantic network.
    

    Parameters: – text: Text

    Returns: <ast.Subscript object at 0x000001A06D64E590>

    query(self, question)

    Lines 374-402

    Purpose:

    Answer a question using the semantic network.
    

    Parameters: – question: Question to answer

    Returns: str

    find_relationship(self, entity1, entity2)

    Lines 404-426

    Purpose:

    Find how two entities are related through the graph.
    

    Parameters: – entity1: Entity1 – entity2: Entity2

    Returns: <ast.Subscript object at 0x000001A06D15B2D0>

    get_related_concepts(self, concept, relation_type)

    Lines 428-456

    Purpose:

    Get concepts related to the given concept.
    

    Parameters: – concept: Concept – relation_type: Relation type

    Returns: <ast.Subscript object at 0x000001A06D703450>

    _format_graph_context(self, entities, relationships)

    Lines 458-476

    Purpose:

    Format graph elements as readable context.
    

    Parameters: – entities: Entities – relationships: Relationships

    Returns: str

    get_statistics(self)

    Lines 478-494

    Purpose:

    Get statistics about the semantic network.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D15CD50>


    Functions

    demonstrate_semantic_memory_networks()

    Lines 497-634

    Purpose:

    Demonstrates semantic memory networks with knowledge extraction,
    graph-based queries, and relationship discovery.
    

    Key Concepts Summary

    Pattern Overview

    The Semantic Memory Networks 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

    1. Agent Class: Main implementation of the pattern logic
    2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
    3. Tool Integration: Optional tools for extended functionality
    4. Memory Management: State handling across interactions
    5. 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

    1. Prompt Optimization: Minimize tokens while maintaining clarity
    2. Caching: Cache repeated queries/responses
    3. Model Selection: Use smaller models when appropriate
    4. Batch Processing: Group similar requests
    5. 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

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

    Feature Extensions

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

    Code Quality

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

    Related Patterns

    • Pattern 028
    • Pattern 030

    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 Semantic Memory Networks 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: 029_semantic_memory_networks.py
    Total Lines: 639
    Pattern Category: AI Agent Design Pattern #029


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Working Memory

    
        """
    Pattern 028: Working Memory
    
    Description:
        The Working Memory pattern provides an active workspace for the agent's current
        task processing. Unlike short-term memory (conversation history) or long-term
        memory (persistent storage), working memory holds intermediate results, active
        goals, and context needed for the current cognitive task. It's dynamically
        updated as the agent progresses through multi-step reasoning.
    
    Components:
        - Active Goals: Current objectives being pursued
        - Intermediate Results: Partial solutions and computations
        - Task Context: Relevant information for current task
        - Attention Focus: What the agent is currently processing
        - Scratch Space: Temporary storage for calculations
    
    Use Cases:
        - Multi-step problem solving
        - Complex reasoning tasks
        - Task decomposition and execution
        - Mathematical computations
        - Planning and execution monitoring
    
    LangChain Implementation:
        Uses structured state management to track active goals, intermediate
        results, and task progress. Implements focus mechanisms to maintain
        relevant context during complex operations.
    
    Key Features:
        - Task-scoped memory
        - Goal tracking
        - Intermediate result storage
        - Dynamic updates
        - Context relevance filtering
    """
    
    import os
    from typing import List, Dict, Any, Optional, Set
    from dataclasses import dataclass, field
    from enum import Enum
    from datetime import datetime
    from dotenv import load_dotenv
    
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser
    from langchain_openai import ChatOpenAI
    
    load_dotenv()
    
    
    class GoalStatus(Enum):
        """Status of a goal in working memory."""
        ACTIVE = "active"
        IN_PROGRESS = "in_progress"
        COMPLETED = "completed"
        BLOCKED = "blocked"
        ABANDONED = "abandoned"
    
    
    @dataclass
    class Goal:
        """Represents a goal in working memory."""
        id: str
        description: str
        status: GoalStatus = GoalStatus.ACTIVE
        parent_goal_id: Optional[str] = None
        sub_goals: List[str] = field(default_factory=list)
        result: Optional[Any] = None
        metadata: Dict[str, Any] = field(default_factory=dict)
        created_at: datetime = field(default_factory=datetime.now)
        
        def to_dict(self) -> Dict[str, Any]:
            """Convert to dictionary."""
            return {
                "id": self.id,
                "description": self.description,
                "status": self.status.value,
                "parent_goal_id": self.parent_goal_id,
                "sub_goals": self.sub_goals,
                "result": str(self.result) if self.result is not None else None,
                "metadata": self.metadata
            }
    
    
    @dataclass
    class IntermediateResult:
        """Stores intermediate computation results."""
        id: str
        description: str
        value: Any
        goal_id: str
        timestamp: datetime = field(default_factory=datetime.now)
        
        def to_dict(self) -> Dict[str, Any]:
            """Convert to dictionary."""
            return {
                "id": self.id,
                "description": self.description,
                "value": str(self.value),
                "goal_id": self.goal_id,
                "timestamp": self.timestamp.isoformat()
            }
    
    
    class WorkingMemorySpace:
        """
        Working memory workspace for active task processing.
        """
        
        def __init__(self, capacity: int = 7):
            """
            Initialize working memory.
            
            Args:
                capacity: Maximum number of active items (Miller's Law: 7±2)
            """
            self.capacity = capacity
            self.goals: Dict[str, Goal] = {}
            self.intermediate_results: Dict[str, IntermediateResult] = {}
            self.task_context: Dict[str, Any] = {}
            self.attention_focus: Optional[str] = None  # Current goal ID
            self.scratch_space: Dict[str, Any] = {}
            
        def add_goal(
            self,
            goal_id: str,
            description: str,
            parent_goal_id: Optional[str] = None
        ) -> Goal:
            """
            Add a new goal to working memory.
            
            Args:
                goal_id: Unique goal identifier
                description: Goal description
                parent_goal_id: Parent goal if this is a sub-goal
                
            Returns:
                Created goal
            """
            goal = Goal(
                id=goal_id,
                description=description,
                parent_goal_id=parent_goal_id
            )
            
            self.goals[goal_id] = goal
            
            # Link to parent
            if parent_goal_id and parent_goal_id in self.goals:
                self.goals[parent_goal_id].sub_goals.append(goal_id)
            
            # Set as focus if first goal
            if self.attention_focus is None:
                self.attention_focus = goal_id
            
            return goal
        
        def update_goal_status(self, goal_id: str, status: GoalStatus, result: Any = None):
            """Update goal status and optionally store result."""
            if goal_id in self.goals:
                self.goals[goal_id].status = status
                if result is not None:
                    self.goals[goal_id].result = result
        
        def store_intermediate(
            self,
            result_id: str,
            description: str,
            value: Any,
            goal_id: str
        ):
            """Store an intermediate result."""
            result = IntermediateResult(
                id=result_id,
                description=description,
                value=value,
                goal_id=goal_id
            )
            self.intermediate_results[result_id] = result
        
        def get_active_goals(self) -> List[Goal]:
            """Get all active goals."""
            return [
                goal for goal in self.goals.values()
                if goal.status in [GoalStatus.ACTIVE, GoalStatus.IN_PROGRESS]
            ]
        
        def get_current_focus(self) -> Optional[Goal]:
            """Get currently focused goal."""
            if self.attention_focus and self.attention_focus in self.goals:
                return self.goals[self.attention_focus]
            return None
        
        def shift_focus(self, goal_id: str):
            """Shift attention to a different goal."""
            if goal_id in self.goals:
                self.attention_focus = goal_id
        
        def get_goal_context(self, goal_id: str) -> Dict[str, Any]:
            """Get all relevant context for a goal."""
            if goal_id not in self.goals:
                return {}
            
            goal = self.goals[goal_id]
            context = {
                "goal": goal.to_dict(),
                "intermediate_results": [
                    ir.to_dict() for ir in self.intermediate_results.values()
                    if ir.goal_id == goal_id
                ],
                "sub_goals": [
                    self.goals[sg_id].to_dict()
                    for sg_id in goal.sub_goals
                    if sg_id in self.goals
                ]
            }
            
            return context
        
        def get_capacity_usage(self) -> float:
            """Get percentage of working memory capacity used."""
            active_items = len(self.get_active_goals()) + len(self.intermediate_results)
            return min(100.0, (active_items / self.capacity) * 100)
        
        def clear(self):
            """Clear working memory."""
            self.goals.clear()
            self.intermediate_results.clear()
            self.task_context.clear()
            self.attention_focus = None
            self.scratch_space.clear()
        
        def get_memory_state(self) -> Dict[str, Any]:
            """Get complete state of working memory."""
            return {
                "capacity": self.capacity,
                "capacity_usage": self.get_capacity_usage(),
                "total_goals": len(self.goals),
                "active_goals": len(self.get_active_goals()),
                "intermediate_results": len(self.intermediate_results),
                "current_focus": self.attention_focus,
                "goals": [g.to_dict() for g in self.goals.values()],
                "results": [ir.to_dict() for ir in self.intermediate_results.values()]
            }
    
    
    class WorkingMemoryAgent:
        """
        Agent with working memory for complex task processing.
        """
        
        def __init__(self, model: str = "gpt-3.5-turbo"):
            """
            Initialize agent with working memory.
            
            Args:
                model: LLM model to use
            """
            self.llm = ChatOpenAI(model=model, temperature=0.3)
            self.working_memory = WorkingMemorySpace()
            self.task_counter = 0
        
        def solve_problem(self, problem: str) -> Dict[str, Any]:
            """
            Solve a problem using working memory.
            
            Args:
                problem: Problem description
                
            Returns:
                Solution with working memory trace
            """
            print(f"\n[Agent] Solving problem: {problem}")
            print(f"[Agent] Initializing working memory...\n")
            
            # Clear working memory for new task
            self.working_memory.clear()
            
            # Create main goal
            main_goal_id = "goal_0"
            self.working_memory.add_goal(main_goal_id, problem)
            
            # Decompose problem into sub-goals
            print("[Step 1] Decomposing problem into sub-goals...")
            sub_goals = self._decompose_problem(problem)
            
            for i, sub_goal in enumerate(sub_goals):
                sub_goal_id = f"goal_{i+1}"
                self.working_memory.add_goal(sub_goal_id, sub_goal, main_goal_id)
                print(f"  Sub-goal {i+1}: {sub_goal}")
            
            print(f"\n[Working Memory] Capacity: {self.working_memory.get_capacity_usage():.1f}%\n")
            
            # Process each sub-goal
            for i, sub_goal in enumerate(sub_goals):
                sub_goal_id = f"goal_{i+1}"
                
                # Shift focus to this sub-goal
                self.working_memory.shift_focus(sub_goal_id)
                current_focus = self.working_memory.get_current_focus()
                
                print(f"[Step {i+2}] Processing: {current_focus.description}")
                
                # Mark as in progress
                self.working_memory.update_goal_status(sub_goal_id, GoalStatus.IN_PROGRESS)
                
                # Solve sub-goal
                result = self._solve_sub_goal(sub_goal)
                
                # Store result
                self.working_memory.update_goal_status(
                    sub_goal_id,
                    GoalStatus.COMPLETED,
                    result
                )
                
                # Store as intermediate result
                self.working_memory.store_intermediate(
                    f"result_{i}",
                    f"Solution to: {sub_goal}",
                    result,
                    sub_goal_id
                )
                
                print(f"  ✓ Result: {result}")
                print(f"  [Working Memory] Capacity: {self.working_memory.get_capacity_usage():.1f}%\n")
            
            # Synthesize final solution
            print("[Final Step] Synthesizing solution from sub-goals...")
            final_solution = self._synthesize_solution(problem)
            
            # Mark main goal as completed
            self.working_memory.update_goal_status(
                main_goal_id,
                GoalStatus.COMPLETED,
                final_solution
            )
            
            print(f"\n✓ Solution: {final_solution}\n")
            
            # Get memory state
            memory_state = self.working_memory.get_memory_state()
            
            return {
                "problem": problem,
                "solution": final_solution,
                "sub_goals": sub_goals,
                "working_memory_trace": memory_state
            }
        
        def _decompose_problem(self, problem: str) -> List[str]:
            """Decompose problem into sub-goals."""
            prompt = ChatPromptTemplate.from_messages([
                ("system", """You are an expert at problem decomposition. Break down
    the given problem into 3-5 clear, actionable sub-goals."""),
                ("user", """Problem: {problem}
    
    List 3-5 sub-goals needed to solve this problem. Be specific and actionable.
    Respond with one sub-goal per line.""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            
            response = chain.invoke({"problem": problem})
            
            # Parse sub-goals
            lines = response.strip().split("\n")
            sub_goals = []
            for line in lines:
                line = line.strip()
                # Remove numbering
                for prefix in ["1.", "2.", "3.", "4.", "5.", "-", "*", "•"]:
                    if line.startswith(prefix):
                        line = line[len(prefix):].strip()
                if line:
                    sub_goals.append(line)
            
            return sub_goals[:5]  # Limit to 5 sub-goals
        
        def _solve_sub_goal(self, sub_goal: str) -> str:
            """Solve a single sub-goal."""
            # Get context from working memory
            focus = self.working_memory.get_current_focus()
            context = self.working_memory.get_goal_context(focus.id) if focus else {}
            
            # Get previously completed results
            previous_results = [
                ir for ir in self.working_memory.intermediate_results.values()
            ]
            
            previous_context = ""
            if previous_results:
                previous_context = "\n".join([
                    f"- {ir.description}: {ir.value}"
                    for ir in previous_results
                ])
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You solve sub-goals clearly and concisely."),
                ("user", """Sub-goal: {sub_goal}
    
    Previous Results:
    {previous_context}
    
    Provide a clear, concise solution:""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            
            result = chain.invoke({
                "sub_goal": sub_goal,
                "previous_context": previous_context if previous_context else "None yet"
            })
            
            return result.strip()
        
        def _synthesize_solution(self, problem: str) -> str:
            """Synthesize final solution from sub-goal results."""
            # Get all intermediate results
            results_text = "\n".join([
                f"{i+1}. {ir.description}\n   {ir.value}"
                for i, ir in enumerate(self.working_memory.intermediate_results.values())
            ])
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You synthesize complete solutions from sub-goal results."),
                ("user", """Original Problem: {problem}
    
    Sub-goal Results:
    {results_text}
    
    Provide a complete, coherent solution to the original problem:""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            
            solution = chain.invoke({
                "problem": problem,
                "results_text": results_text
            })
            
            return solution.strip()
    
    
    def demonstrate_working_memory():
        """Demonstrate the Working Memory pattern."""
        
        print("=" * 80)
        print("WORKING MEMORY PATTERN DEMONSTRATION")
        print("=" * 80)
        
        agent = WorkingMemoryAgent()
        
        # Test 1: Multi-step math problem
        print("\n" + "=" * 80)
        print("TEST 1: Multi-Step Mathematical Problem")
        print("=" * 80)
        
        problem1 = """A store sells apples for $2 each and oranges for $3 each. 
    If you buy 5 apples and 4 oranges, get a 10% discount, and pay 8% tax, 
    what is the final price?"""
        
        result1 = agent.solve_problem(problem1)
        
        print("=" * 80)
        print("WORKING MEMORY TRACE")
        print("=" * 80)
        print(f"Goals tracked: {result1['working_memory_trace']['total_goals']}")
        print(f"Active computations: {result1['working_memory_trace']['intermediate_results']}")
        print(f"Capacity used: {result1['working_memory_trace']['capacity_usage']:.1f}%")
        
        # Test 2: Planning task
        print("\n\n" + "=" * 80)
        print("TEST 2: Planning Task")
        print("=" * 80)
        
        problem2 = """Plan a weekend trip to a nearby city including:
    transportation, accommodation, activities, and budget estimation."""
        
        result2 = agent.solve_problem(problem2)
        
        print("=" * 80)
        print("WORKING MEMORY TRACE")
        print("=" * 80)
        print(f"Goals tracked: {result2['working_memory_trace']['total_goals']}")
        print(f"Sub-goals completed: {len(result2['sub_goals'])}")
        
        # Test 3: Analytical task
        print("\n\n" + "=" * 80)
        print("TEST 3: Analytical Reasoning")
        print("=" * 80)
        
        problem3 = """Analyze the pros and cons of remote work and provide
    a recommendation for a tech company."""
        
        result3 = agent.solve_problem(problem3)
        
        print("=" * 80)
        print("WORKING MEMORY TRACE")
        print("=" * 80)
        for goal in result3['working_memory_trace']['goals']:
            print(f"\nGoal: {goal['description']}")
            print(f"  Status: {goal['status']}")
            if goal['result']:
                preview = goal['result'][:100] + "..." if len(goal['result']) > 100 else goal['result']
                print(f"  Result: {preview}")
        
        # Summary
        print("\n\n" + "=" * 80)
        print("PATTERN SUMMARY")
        print("=" * 80)
        print("""
    The Working Memory pattern demonstrates:
    
    1. **Active Workspace**: Maintains current task context
    2. **Goal Tracking**: Monitors progress on objectives
    3. **Intermediate Storage**: Stores partial results
    4. **Attention Focus**: Tracks current processing target
    5. **Capacity Management**: Limited capacity (Miller's Law)
    
    Key Components:
    
    Goals:
    - Hierarchical goal structure (main goal → sub-goals)
    - Status tracking (active, in-progress, completed, blocked)
    - Result storage
    - Parent-child relationships
    
    Intermediate Results:
    - Temporary computation storage
    - Linked to specific goals
    - Time-stamped for ordering
    - Available for later steps
    
    Attention Focus:
    - Single goal actively processed
    - Context switching between goals
    - Maintains processing coherence
    
    Working Memory Characteristics:
    
    Capacity Limitation (Miller's Law):
    - 7±2 items simultaneously
    - Prevents cognitive overload
    - Forces prioritization
    - Matches human working memory
    
    Dynamic Updates:
    - Content changes as task progresses
    - Old items removed when complete
    - New items added as needed
    - Maintains relevance
    
    Task-Scoped:
    - Cleared between tasks
    - Fresh workspace for each problem
    - No cross-task contamination
    
    Differences from Other Memory Types:
    
    vs Short-Term Memory:
    - Working: Active processing workspace
    - Short-term: Conversation history buffer
    - Working is goal-oriented, short-term is sequential
    
    vs Long-Term Memory:
    - Working: Temporary, task-scoped
    - Long-term: Permanent, cross-session
    - Working is volatile, long-term persists
    
    Use Cases:
    - Multi-step problem solving
    - Complex calculations
    - Task planning and execution
    - Reasoning chains
    - Resource allocation
    
    Benefits:
    - **Organized Processing**: Structured approach to complexity
    - **Progress Tracking**: Monitor goal completion
    - **Context Maintenance**: Relevant information available
    - **Intermediate Access**: Use partial results
    - **Cognitive Modeling**: Mimics human working memory
    
    This pattern is essential for agents tackling complex, multi-step
    tasks that require maintaining and manipulating intermediate state.
    """)
    
    
    if __name__ == "__main__":
        demonstrate_working_memory()
    
        

    Agentic Design Pattern: Working Memory

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

    Overview

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


    Pattern Description

    Pattern 028: Working Memory

    Description: The Working Memory pattern provides an active workspace for the agent’s current task processing. Unlike short-term memory (conversation history) or long-term memory (persistent storage), working memory holds intermediate results, active goals, and context needed for the current cognitive task. It’s dynamically updated as the agent progresses through multi-step reasoning.

    Components: – Active Goals: Current objectives being pursued – Intermediate Results: Partial solutions and computations – Task Context: Relevant information for current task – Attention Focus: What the agent is currently processing – Scratch Space: Temporary storage for calculations

    Use Cases: – Multi-step problem solving – Complex reasoning tasks – Task decomposition and execution – Mathematical computations – Planning and execution monitoring

    LangChain Implementation: Uses structured state management to track active goals, intermediate results, and task progress. Implements focus mechanisms to maintain relevant context during complex operations.

    Key Features: – Task-scoped memory – Goal tracking – Intermediate result storage – Dynamic updates – Context relevance filtering


    File Statistics

    • Total Lines: 594
    • Classes: 5
    • Functions: 1
    • Constants: 0

    Module Documentation

    """
    Pattern 028: Working Memory
    
    Description:
        The Working Memory pattern provides an active workspace for the agent's current
        task processing. Unlike short-term memory (conversation history) or long-term
        memory (persistent storage), working memory holds intermediate results, active
        goals, and context needed for the current cognitive task. It's dynamically
        updated as the agent progresses through multi-step reasoning.
    
    Components:
        - Active Goals: Current objectives being pursued
        - Intermediate Results: Partial solutions and computations
        - Task Context: Relevant information for current task
        - Attention Focus: What the agent is currently processing
        - Scratch Space: Temporary storage for calculations
    
    Use Cases:
        - Multi-step problem solving
        - Complex reasoning tasks
        - Task decomposition and execution
        - Mathematical computations
        - Planning and execution monitoring
    
    LangChain Implementation:
        Uses structured state management to track active goals, intermediate
        results, and task progress. Implements focus mechanisms to maintain
        relevant context during complex operations.
    
    Key Features:
        - Task-scoped memory
        - Goal tracking
        - Intermediate result storage
        - Dynamic updates
        - Context relevance filtering
    """
    

    Purpose: This module implements the Working 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.output_parsers.StrOutputParser
    import langchain_core.prompts.ChatPromptTemplate
    import os
    import typing.Any
    import typing.Dict
    import typing.List
    import typing.Optional
    import typing.Set
    

    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.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: 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) – typing.Set: 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: GoalStatus

    Lines 52-58

    Inherits from: Enum

    Purpose:

    Status of a goal in working memory.
    

    Class: Goal

    Lines 62-83

    Purpose:

    Represents a goal in working memory.
    

    Methods:

    to_dict(self)

    Lines 73-83

    Purpose:

    Convert to dictionary.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D66AB90>


    Class: IntermediateResult

    Lines 87-103

    Purpose:

    Stores intermediate computation results.
    

    Methods:

    to_dict(self)

    Lines 95-103

    Purpose:

    Convert to dictionary.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D6A3390>


    Class: WorkingMemorySpace

    Lines 106-246

    Purpose:

    Working memory workspace for active task processing.
    

    Methods:

    __init__(self, capacity)

    Lines 111-123

    Purpose:

    Initialize working memory.
    
    Args:
        capacity: Maximum number of active items (Miller's Law: 7±2)
    

    Parameters: – capacity: Capacity

    add_goal(self, goal_id, description, parent_goal_id)

    Lines 125-158

    Purpose:

    Add a new goal to working memory.
    
    Args:
        goal_id: Unique goal identifier
        description: Goal description
        parent_goal_id: Parent goal if this is a sub-goal
    
    Returns:
        Created goal
    

    Parameters: – goal_id: Goal id – description: Description – parent_goal_id: Parent goal id

    Returns: Goal

    update_goal_status(self, goal_id, status, result)

    Lines 160-165

    Purpose:

    Update goal status and optionally store result.
    

    Parameters: – goal_id: Goal id – status: Status – result: Result

    store_intermediate(self, result_id, description, value, goal_id)

    Lines 167-181

    Purpose:

    Store an intermediate result.
    

    Parameters: – result_id: Result id – description: Description – value: Value – goal_id: Goal id

    get_active_goals(self)

    Lines 183-188

    Purpose:

    Get all active goals.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D156A50>

    get_current_focus(self)

    Lines 190-194

    Purpose:

    Get currently focused goal.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D658710>

    shift_focus(self, goal_id)

    Lines 196-199

    Purpose:

    Shift attention to a different goal.
    

    Parameters: – goal_id: Goal id

    get_goal_context(self, goal_id)

    Lines 201-220

    Purpose:

    Get all relevant context for a goal.
    

    Parameters: – goal_id: Goal id

    Returns: <ast.Subscript object at 0x000001A06D739FD0>

    get_capacity_usage(self)

    Lines 222-225

    Purpose:

    Get percentage of working memory capacity used.
    

    Parameters:

    Returns: float

    clear(self)

    Lines 227-233

    Purpose:

    Clear working memory.
    

    Parameters:

    get_memory_state(self)

    Lines 235-246

    Purpose:

    Get complete state of working memory.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D62ABD0>


    Class: WorkingMemoryAgent

    Lines 249-444

    Purpose:

    Agent with working memory for complex task processing.
    

    Methods:

    __init__(self, model)

    Lines 254-263

    Purpose:

    Initialize agent with working memory.
    
    Args:
        model: LLM model to use
    

    Parameters: – model: Language model instance

    solve_problem(self, problem)

    Lines 265-351

    Purpose:

    Solve a problem using working memory.
    
    Args:
        problem: Problem description
    
    Returns:
        Solution with working memory trace
    

    Parameters: – problem: Problem

    Returns: <ast.Subscript object at 0x000001A06D60A890>

    _decompose_problem(self, problem)

    Lines 353-380

    Purpose:

    Decompose problem into sub-goals.
    

    Parameters: – problem: Problem

    Returns: <ast.Subscript object at 0x000001A06D5F7C50>

    _solve_sub_goal(self, sub_goal)

    Lines 382-417

    Purpose:

    Solve a single sub-goal.
    

    Parameters: – sub_goal: Sub goal

    Returns: str

    _synthesize_solution(self, problem)

    Lines 419-444

    Purpose:

    Synthesize final solution from sub-goal results.
    

    Parameters: – problem: Problem

    Returns: str


    Functions

    demonstrate_working_memory()

    Lines 447-589

    Purpose:

    Demonstrate the Working Memory pattern.
    

    Key Concepts Summary

    Pattern Overview

    The Working 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

    1. Agent Class: Main implementation of the pattern logic
    2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
    3. Tool Integration: Optional tools for extended functionality
    4. Memory Management: State handling across interactions
    5. 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

    1. Prompt Optimization: Minimize tokens while maintaining clarity
    2. Caching: Cache repeated queries/responses
    3. Model Selection: Use smaller models when appropriate
    4. Batch Processing: Group similar requests
    5. 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

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

    Feature Extensions

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

    Code Quality

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

    Related Patterns

    • Pattern 027
    • Pattern 029

    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 Working 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: 028_working_memory.py
    Total Lines: 594
    Pattern Category: AI Agent Design Pattern #028


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Long Term Memory

    
        """
    Pattern 027: Long-Term Memory
    
    Description:
        The Long-Term Memory pattern enables agents to store and retrieve information
        across multiple sessions and conversations. Unlike short-term memory that exists
        only within a session, long-term memory persists indefinitely, allowing agents
        to remember user preferences, past interactions, learned facts, and accumulated
        knowledge over time.
    
    Components:
        - Persistent Storage: Vector databases, traditional databases, file systems
        - Memory Indexing: Efficient retrieval mechanisms
        - Memory Types: Episodic (events), Semantic (facts), Procedural (skills)
        - Retrieval System: Similarity search, keyword matching
        - Memory Consolidation: Organizing and optimizing stored memories
    
    Use Cases:
        - Personalized user experiences
        - Learning from past interactions
        - Knowledge accumulation
        - User preference tracking
        - Long-term relationship building
    
    LangChain Implementation:
        Uses vector stores (Chroma, FAISS) for semantic memory storage,
        implements retrieval strategies, and combines with conversation
        context for personalized interactions.
    
    Key Features:
        - Cross-session persistence
        - Semantic similarity search
        - Memory categorization
        - Efficient retrieval
        - Memory aging and importance
    """
    
    import os
    from typing import List, Dict, Any, Optional, Tuple
    from dataclasses import dataclass, field
    from enum import Enum
    from datetime import datetime
    import json
    from pathlib import Path
    from dotenv import load_dotenv
    
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser
    from langchain_openai import ChatOpenAI, OpenAIEmbeddings
    
    load_dotenv()
    
    
    class MemoryType(Enum):
        """Types of long-term memories."""
        EPISODIC = "episodic"  # Specific events and experiences
        SEMANTIC = "semantic"  # Facts and knowledge
        PROCEDURAL = "procedural"  # Skills and procedures
        PREFERENCE = "preference"  # User preferences
        INTERACTION = "interaction"  # Past interactions
    
    
    @dataclass
    class Memory:
        """Represents a long-term memory."""
        id: str
        content: str
        memory_type: MemoryType
        timestamp: datetime
        importance: float = 1.0  # 0-1 scale
        metadata: Dict[str, Any] = field(default_factory=dict)
        access_count: int = 0
        last_accessed: Optional[datetime] = None
        
        def to_dict(self) -> Dict[str, Any]:
            """Convert to dictionary."""
            return {
                "id": self.id,
                "content": self.content,
                "memory_type": self.memory_type.value,
                "timestamp": self.timestamp.isoformat(),
                "importance": self.importance,
                "metadata": self.metadata,
                "access_count": self.access_count,
                "last_accessed": self.last_accessed.isoformat() if self.last_accessed else None
            }
        
        @classmethod
        def from_dict(cls, data: Dict[str, Any]) -> 'Memory':
            """Create from dictionary."""
            return cls(
                id=data["id"],
                content=data["content"],
                memory_type=MemoryType(data["memory_type"]),
                timestamp=datetime.fromisoformat(data["timestamp"]),
                importance=data.get("importance", 1.0),
                metadata=data.get("metadata", {}),
                access_count=data.get("access_count", 0),
                last_accessed=datetime.fromisoformat(data["last_accessed"]) if data.get("last_accessed") else None
            )
    
    
    class SimpleVectorStore:
        """
        Simple vector store for semantic memory storage (mock implementation).
        In production, would use Chroma, FAISS, or Pinecone.
        """
        
        def __init__(self):
            """Initialize vector store."""
            self.memories: List[Memory] = []
            self.embeddings_model = None  # Would use OpenAIEmbeddings() in production
        
        def add_memory(self, memory: Memory):
            """Add a memory to the store."""
            self.memories.append(memory)
        
        def search(
            self,
            query: str,
            memory_type: Optional[MemoryType] = None,
            k: int = 5
        ) -> List[Memory]:
            """
            Search for relevant memories.
            
            Args:
                query: Search query
                memory_type: Filter by memory type
                k: Number of results
                
            Returns:
                List of relevant memories
            """
            # Filter by type if specified
            candidates = self.memories
            if memory_type:
                candidates = [m for m in candidates if m.memory_type == memory_type]
            
            # Simple keyword matching (in production, would use vector similarity)
            query_lower = query.lower()
            scored_memories = []
            
            for memory in candidates:
                # Simple scoring based on keyword matches
                content_lower = memory.content.lower()
                score = 0.0
                
                for word in query_lower.split():
                    if word in content_lower:
                        score += 1.0
                
                # Boost by importance and recency
                score *= memory.importance
                
                days_old = (datetime.now() - memory.timestamp).days
                recency_factor = 1.0 / (1.0 + days_old * 0.01)
                score *= recency_factor
                
                if score > 0:
                    scored_memories.append((score, memory))
            
            # Sort by score and return top k
            scored_memories.sort(reverse=True, key=lambda x: x[0])
            results = [memory for _, memory in scored_memories[:k]]
            
            # Update access statistics
            for memory in results:
                memory.access_count += 1
                memory.last_accessed = datetime.now()
            
            return results
        
        def get_all_memories(self) -> List[Memory]:
            """Get all memories."""
            return self.memories.copy()
        
        def delete_memory(self, memory_id: str):
            """Delete a memory by ID."""
            self.memories = [m for m in self.memories if m.id != memory_id]
    
    
    class PersistentMemoryStore:
        """
        Persistent storage for long-term memory (file-based mock).
        In production, would use proper database.
        """
        
        def __init__(self, storage_path: str = "memory_store.json"):
            """
            Initialize persistent store.
            
            Args:
                storage_path: Path to storage file
            """
            self.storage_path = Path(storage_path)
            self.vector_store = SimpleVectorStore()
            self._load()
        
        def _load(self):
            """Load memories from disk."""
            if self.storage_path.exists():
                try:
                    with open(self.storage_path, 'r') as f:
                        data = json.load(f)
                        for memory_data in data:
                            memory = Memory.from_dict(memory_data)
                            self.vector_store.add_memory(memory)
                except Exception as e:
                    print(f"Warning: Could not load memories: {e}")
        
        def _save(self):
            """Save memories to disk."""
            try:
                data = [m.to_dict() for m in self.vector_store.get_all_memories()]
                with open(self.storage_path, 'w') as f:
                    json.dump(data, f, indent=2)
            except Exception as e:
                print(f"Warning: Could not save memories: {e}")
        
        def add_memory(self, memory: Memory):
            """Add and persist a memory."""
            self.vector_store.add_memory(memory)
            self._save()
        
        def search(
            self,
            query: str,
            memory_type: Optional[MemoryType] = None,
            k: int = 5
        ) -> List[Memory]:
            """Search for relevant memories."""
            return self.vector_store.search(query, memory_type, k)
        
        def get_all_memories(self) -> List[Memory]:
            """Get all stored memories."""
            return self.vector_store.get_all_memories()
        
        def clear(self):
            """Clear all memories."""
            self.vector_store.memories.clear()
            self._save()
    
    
    class LongTermMemoryAgent:
        """
        Agent with long-term memory capabilities.
        """
        
        def __init__(
            self,
            user_id: str,
            storage_path: Optional[str] = None,
            model: str = "gpt-3.5-turbo"
        ):
            """
            Initialize agent with long-term memory.
            
            Args:
                user_id: Unique user identifier
                storage_path: Path to memory storage
                model: LLM model to use
            """
            self.user_id = user_id
            self.llm = ChatOpenAI(model=model, temperature=0.7)
            
            # Initialize memory store
            if storage_path is None:
                storage_path = f"memory_{user_id}.json"
            self.memory_store = PersistentMemoryStore(storage_path)
            
            self.interaction_count = 0
        
        def store_memory(
            self,
            content: str,
            memory_type: MemoryType,
            importance: float = 1.0,
            metadata: Optional[Dict] = None
        ):
            """
            Store a new memory.
            
            Args:
                content: Memory content
                memory_type: Type of memory
                importance: Importance score (0-1)
                metadata: Additional metadata
            """
            memory = Memory(
                id=f"{self.user_id}_{datetime.now().timestamp()}",
                content=content,
                memory_type=memory_type,
                timestamp=datetime.now(),
                importance=importance,
                metadata=metadata or {}
            )
            self.memory_store.add_memory(memory)
        
        def recall_memories(
            self,
            query: str,
            memory_type: Optional[MemoryType] = None,
            k: int = 5
        ) -> List[Memory]:
            """
            Recall relevant memories.
            
            Args:
                query: Query for memory retrieval
                memory_type: Filter by type
                k: Number of memories to retrieve
                
            Returns:
                List of relevant memories
            """
            return self.memory_store.search(query, memory_type, k)
        
        def chat(self, user_message: str) -> str:
            """
            Chat with memory-augmented responses.
            
            Args:
                user_message: User's message
                
            Returns:
                Agent's response
            """
            self.interaction_count += 1
            
            # Recall relevant memories
            relevant_memories = self.recall_memories(user_message, k=3)
            
            # Format memories for context
            memory_context = ""
            if relevant_memories:
                memory_lines = []
                for mem in relevant_memories:
                    memory_lines.append(
                        f"[{mem.memory_type.value}] {mem.content} "
                        f"(from {mem.timestamp.strftime('%Y-%m-%d')})"
                    )
                memory_context = "\n".join(memory_lines)
            
            # Create prompt with memory context
            prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a helpful AI assistant with long-term memory.
    Use the relevant memories to personalize your responses and maintain continuity
    across conversations."""),
                ("user", """Relevant Memories:
    {memory_context}
    
    Current Message: {message}
    
    Provide a helpful, personalized response:""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            
            response = chain.invoke({
                "memory_context": memory_context if memory_context else "No relevant memories found.",
                "message": user_message
            })
            
            # Store this interaction
            self.store_memory(
                content=f"User: {user_message} | AI: {response}",
                memory_type=MemoryType.INTERACTION,
                importance=0.5
            )
            
            # Extract and store important information
            self._extract_and_store_info(user_message)
            
            return response.strip()
        
        def _extract_and_store_info(self, message: str):
            """Extract and store important information from message."""
            # Simple extraction logic (in production, would use NER or LLM)
            message_lower = message.lower()
            
            # Detect preferences
            if any(word in message_lower for word in ["like", "prefer", "favorite", "love"]):
                self.store_memory(
                    content=message,
                    memory_type=MemoryType.PREFERENCE,
                    importance=0.8
                )
            
            # Detect facts
            if any(word in message_lower for word in ["my name is", "i am", "i work", "i live"]):
                self.store_memory(
                    content=message,
                    memory_type=MemoryType.SEMANTIC,
                    importance=0.9
                )
        
        def get_memory_summary(self) -> Dict[str, Any]:
            """Get summary of stored memories."""
            all_memories = self.memory_store.get_all_memories()
            
            memory_by_type = {}
            for mem_type in MemoryType:
                memory_by_type[mem_type.value] = sum(
                    1 for m in all_memories if m.memory_type == mem_type
                )
            
            total_accesses = sum(m.access_count for m in all_memories)
            
            return {
                "user_id": self.user_id,
                "total_memories": len(all_memories),
                "by_type": memory_by_type,
                "total_accesses": total_accesses,
                "interactions": self.interaction_count
            }
    
    
    def demonstrate_long_term_memory():
        """Demonstrate the Long-Term Memory pattern."""
        
        print("=" * 80)
        print("LONG-TERM MEMORY PATTERN DEMONSTRATION")
        print("=" * 80)
        
        # Create agent for user
        user_id = "alice_demo"
        agent = LongTermMemoryAgent(user_id=user_id, storage_path=f"memory_{user_id}_demo.json")
        
        # Clear any existing memories
        agent.memory_store.clear()
        
        # Session 1: Initial interaction
        print("\n" + "=" * 80)
        print("SESSION 1: Initial Interaction")
        print("=" * 80)
        
        messages_session1 = [
            "Hi, my name is Alice and I'm a software engineer.",
            "I really love Python programming and machine learning.",
            "My favorite food is sushi."
        ]
        
        for msg in messages_session1:
            print(f"\n👤 User: {msg}")
            response = agent.chat(msg)
            print(f"🤖 Agent: {response}")
        
        print("\n📊 Memory Summary after Session 1:")
        summary = agent.get_memory_summary()
        print(json.dumps(summary, indent=2))
        
        # Simulate new session (create new agent instance)
        print("\n\n" + "=" * 80)
        print("SESSION 2: Later Conversation (Testing Persistence)")
        print("=" * 80)
        print("(Simulating new session - creating fresh agent instance)")
        
        agent2 = LongTermMemoryAgent(user_id=user_id, storage_path=f"memory_{user_id}_demo.json")
        
        messages_session2 = [
            "What's my name?",
            "What do I do for work?",
            "What are my interests?",
            "What food do I like?"
        ]
        
        for msg in messages_session2:
            print(f"\n👤 User: {msg}")
            response = agent2.chat(msg)
            print(f"🤖 Agent: {response}")
        
        print("\n📊 Memory Summary after Session 2:")
        summary2 = agent2.get_memory_summary()
        print(json.dumps(summary2, indent=2))
        
        # Session 3: Memory accumulation
        print("\n\n" + "=" * 80)
        print("SESSION 3: Adding More Memories")
        print("=" * 80)
        
        agent3 = LongTermMemoryAgent(user_id=user_id, storage_path=f"memory_{user_id}_demo.json")
        
        messages_session3 = [
            "I recently started learning about transformers.",
            "I'm working on a project about recommendation systems.",
            "Can you suggest something based on my interests?"
        ]
        
        for msg in messages_session3:
            print(f"\n👤 User: {msg}")
            response = agent3.chat(msg)
            print(f"🤖 Agent: {response}")
        
        # Show memory details
        print("\n\n" + "=" * 80)
        print("STORED MEMORIES ANALYSIS")
        print("=" * 80)
        
        all_memories = agent3.memory_store.get_all_memories()
        
        print(f"\nTotal memories: {len(all_memories)}")
        
        for mem_type in MemoryType:
            type_memories = [m for m in all_memories if m.memory_type == mem_type]
            if type_memories:
                print(f"\n{mem_type.value.upper()} Memories ({len(type_memories)}):")
                for mem in type_memories[:3]:  # Show first 3
                    print(f"  - {mem.content[:80]}...")
                    print(f"    Importance: {mem.importance:.1f}, Accessed: {mem.access_count} times")
        
        # Clean up demo file
        try:
            os.remove(f"memory_{user_id}_demo.json")
        except:
            pass
        
        # Summary
        print("\n\n" + "=" * 80)
        print("PATTERN SUMMARY")
        print("=" * 80)
        print("""
    The Long-Term Memory pattern demonstrates:
    
    1. **Cross-Session Persistence**: Memories survive across sessions
    2. **Memory Categorization**: Different types (episodic, semantic, preference)
    3. **Semantic Retrieval**: Find relevant memories via similarity
    4. **Memory Statistics**: Track access patterns and importance
    5. **Personalization**: Use past memories for context-aware responses
    
    Memory Types:
    
    Episodic Memory:
    - Stores specific events and experiences
    - "When I visited Paris..."
    - Time and context-stamped
    
    Semantic Memory:
    - Stores facts and knowledge
    - "Paris is the capital of France"
    - General information
    
    Procedural Memory:
    - Stores skills and procedures
    - "How to make coffee"
    - Step-by-step knowledge
    
    Preference Memory:
    - Stores user preferences
    - "I prefer dark mode"
    - Likes and dislikes
    
    Interaction Memory:
    - Stores past conversations
    - Complete interaction history
    - Enables continuity
    
    Key Benefits:
    - **Personalization**: Tailored responses based on history
    - **Continuity**: Maintains context across sessions
    - **Learning**: Accumulates knowledge over time
    - **Relationship**: Builds long-term user relationships
    - **Efficiency**: Doesn't repeat questions
    
    Implementation Considerations:
    - Vector databases for semantic search (Chroma, Pinecone, FAISS)
    - Memory importance and decay
    - Privacy and data retention
    - Storage optimization
    - Cross-device synchronization
    
    Use Cases:
    - Personal assistants with memory of user
    - Educational tutors tracking student progress
    - Healthcare agents remembering patient history
    - Customer service with account context
    - Long-term collaborative projects
    
    This pattern transforms agents from stateless responders to
    persistent entities with accumulated knowledge and relationships.
    """)
    
    
    if __name__ == "__main__":
        demonstrate_long_term_memory()
    
        

    Agentic Design Pattern: Long Term Memory

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

    Overview

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


    Pattern Description

    Pattern 027: Long-Term Memory

    Description: The Long-Term Memory pattern enables agents to store and retrieve information across multiple sessions and conversations. Unlike short-term memory that exists only within a session, long-term memory persists indefinitely, allowing agents to remember user preferences, past interactions, learned facts, and accumulated knowledge over time.

    Components: – Persistent Storage: Vector databases, traditional databases, file systems – Memory Indexing: Efficient retrieval mechanisms – Memory Types: Episodic (events), Semantic (facts), Procedural (skills) – Retrieval System: Similarity search, keyword matching – Memory Consolidation: Organizing and optimizing stored memories

    Use Cases: – Personalized user experiences – Learning from past interactions – Knowledge accumulation – User preference tracking – Long-term relationship building

    LangChain Implementation: Uses vector stores (Chroma, FAISS) for semantic memory storage, implements retrieval strategies, and combines with conversation context for personalized interactions.

    Key Features: – Cross-session persistence – Semantic similarity search – Memory categorization – Efficient retrieval – Memory aging and importance


    File Statistics

    • Total Lines: 586
    • Classes: 5
    • Functions: 1
    • Constants: 0

    Module Documentation

    """
    Pattern 027: Long-Term Memory
    
    Description:
        The Long-Term Memory pattern enables agents to store and retrieve information
        across multiple sessions and conversations. Unlike short-term memory that exists
        only within a session, long-term memory persists indefinitely, allowing agents
        to remember user preferences, past interactions, learned facts, and accumulated
        knowledge over time.
    
    Components:
        - Persistent Storage: Vector databases, traditional databases, file systems
        - Memory Indexing: Efficient retrieval mechanisms
        - Memory Types: Episodic (events), Semantic (facts), Procedural (skills)
        - Retrieval System: Similarity search, keyword matching
        - Memory Consolidation: Organizing and optimizing stored memories
    
    Use Cases:
        - Personalized user experiences
        - Learning from past interactions
        - Knowledge accumulation
        - User preference tracking
        - Long-term relationship building
    
    LangChain Implementation:
        Uses vector stores (Chroma, FAISS) for semantic memory storage,
        implements retrieval strategies, and combines with conversation
        context for personalized interactions.
    
    Key Features:
        - Cross-session persistence
        - Semantic similarity search
        - Memory categorization
        - Efficient retrieval
        - Memory aging and importance
    """
    

    Purpose: This module implements the Long 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 json
    import langchain_core.output_parsers.StrOutputParser
    import langchain_core.prompts.ChatPromptTemplate
    import os
    import pathlib.Path
    import typing.Any
    import typing.Dict
    import typing.List
    import typing.Optional
    import typing.Tuple
    

    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 – json: JSON serialization/deserialization – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – pathlib.Path: Object-oriented filesystem paths – 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) – typing.Tuple: Type hints for better code documentation (List, Dict, Any, Optional)

    LangChain Imports

    from langchain_openai.ChatOpenAI
    from langchain_openai.OpenAIEmbeddings
    

    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: MemoryType

    Lines 54-60

    Inherits from: Enum

    Purpose:

    Types of long-term memories.
    

    Class: Memory

    Lines 64-100

    Purpose:

    Represents a long-term memory.
    

    Methods:

    to_dict(self)

    Lines 75-86

    Purpose:

    Convert to dictionary.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D64A3D0>

    from_dict(cls, data)

    Lines 89-100

    Decorators: classmethod

    Purpose:

    Create from dictionary.
    

    Parameters: – cls: Cls – data: Data

    Returns: Memory


    Class: SimpleVectorStore

    Lines 103-180

    Purpose:

    Simple vector store for semantic memory storage (mock implementation).
    In production, would use Chroma, FAISS, or Pinecone.
    

    Methods:

    __init__(self)

    Lines 109-112

    Purpose:

    Initialize vector store.
    

    Parameters:

    add_memory(self, memory)

    Lines 114-116

    Purpose:

    Add a memory to the store.
    

    Parameters: – memory: Memory store for conversation history

    search(self, query, memory_type, k)

    Lines 118-172

    Purpose:

    Search for relevant memories.
    
    Args:
        query: Search query
        memory_type: Filter by memory type
        k: Number of results
    
    Returns:
        List of relevant memories
    

    Parameters: – query: Input query or question – memory_type: Memory store for conversation history – k: K

    Returns: <ast.Subscript object at 0x000001A06D6EA550>

    get_all_memories(self)

    Lines 174-176

    Purpose:

    Get all memories.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D6E9950>

    delete_memory(self, memory_id)

    Lines 178-180

    Purpose:

    Delete a memory by ID.
    

    Parameters: – memory_id: Memory store for conversation history


    Class: PersistentMemoryStore

    Lines 183-242

    Purpose:

    Persistent storage for long-term memory (file-based mock).
    In production, would use proper database.
    

    Methods:

    __init__(self, storage_path)

    Lines 189-198

    Purpose:

    Initialize persistent store.
    
    Args:
        storage_path: Path to storage file
    

    Parameters: – storage_path: Storage path

    _load(self)

    Lines 200-210

    Purpose:

    Load memories from disk.
    

    Parameters:

    _save(self)

    Lines 212-219

    Purpose:

    Save memories to disk.
    

    Parameters:

    add_memory(self, memory)

    Lines 221-224

    Purpose:

    Add and persist a memory.
    

    Parameters: – memory: Memory store for conversation history

    search(self, query, memory_type, k)

    Lines 226-233

    Purpose:

    Search for relevant memories.
    

    Parameters: – query: Input query or question – memory_type: Memory store for conversation history – k: K

    Returns: <ast.Subscript object at 0x000001A06D70CD50>

    get_all_memories(self)

    Lines 235-237

    Purpose:

    Get all stored memories.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D70E550>

    clear(self)

    Lines 239-242

    Purpose:

    Clear all memories.
    

    Parameters:


    Class: LongTermMemoryAgent

    Lines 245-416

    Purpose:

    Agent with long-term memory capabilities.
    

    Methods:

    __init__(self, user_id, storage_path, model)

    Lines 250-272

    Purpose:

    Initialize agent with long-term memory.
    
    Args:
        user_id: Unique user identifier
        storage_path: Path to memory storage
        model: LLM model to use
    

    Parameters: – user_id: User id – storage_path: Storage path – model: Language model instance

    store_memory(self, content, memory_type, importance, metadata)

    Lines 274-298

    Purpose:

    Store a new memory.
    
    Args:
        content: Memory content
        memory_type: Type of memory
        importance: Importance score (0-1)
        metadata: Additional metadata
    

    Parameters: – content: Content – memory_type: Memory store for conversation history – importance: Importance – metadata: Metadata

    recall_memories(self, query, memory_type, k)

    Lines 300-317

    Purpose:

    Recall relevant memories.
    
    Args:
        query: Query for memory retrieval
        memory_type: Filter by type
        k: Number of memories to retrieve
    
    Returns:
        List of relevant memories
    

    Parameters: – query: Input query or question – memory_type: Memory store for conversation history – k: K

    Returns: <ast.Subscript object at 0x000001A06D15B6D0>

    chat(self, user_message)

    Lines 319-375

    Purpose:

    Chat with memory-augmented responses.
    
    Args:
        user_message: User's message
    
    Returns:
        Agent's response
    

    Parameters: – user_message: User message

    Returns: str

    _extract_and_store_info(self, message)

    Lines 377-396

    Purpose:

    Extract and store important information from message.
    

    Parameters: – message: Message

    get_memory_summary(self)

    Lines 398-416

    Purpose:

    Get summary of stored memories.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D659D90>


    Functions

    demonstrate_long_term_memory()

    Lines 419-581

    Purpose:

    Demonstrate the Long-Term Memory pattern.
    

    Key Concepts Summary

    Pattern Overview

    The Long 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

    1. Agent Class: Main implementation of the pattern logic
    2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
    3. Tool Integration: Optional tools for extended functionality
    4. Memory Management: State handling across interactions
    5. 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

    1. Prompt Optimization: Minimize tokens while maintaining clarity
    2. Caching: Cache repeated queries/responses
    3. Model Selection: Use smaller models when appropriate
    4. Batch Processing: Group similar requests
    5. 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

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

    Feature Extensions

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

    Code Quality

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

    Related Patterns

    • Pattern 026
    • Pattern 028

    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 Long 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: 027_long_term_memory.py
    Total Lines: 586
    Pattern Category: AI Agent Design Pattern #027


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Short Term Memory

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

    1. Agent Class: Main implementation of the pattern logic
    2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
    3. Tool Integration: Optional tools for extended functionality
    4. Memory Management: State handling across interactions
    5. 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

    1. Prompt Optimization: Minimize tokens while maintaining clarity
    2. Caching: Cache repeated queries/responses
    3. Model Selection: Use smaller models when appropriate
    4. Batch Processing: Group similar requests
    5. 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

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

    Feature Extensions

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

    Code Quality

    1. Testing: Unit tests, integration tests
    2. Documentation: API docs, usage examples
    3. Type Safety: Stricter type hints
    4. 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: Action Sequence Planning

    
        """
    Pattern 025: Action Sequence Planning
    
    Description:
        The Action Sequence Planning pattern enables agents to decompose complex goals
        into ordered sequences of actions with preconditions, effects, and dependencies.
        The agent analyzes the goal, identifies required actions, determines ordering
        constraints, validates feasibility, and executes the plan step-by-step with
        monitoring and replanning capabilities.
    
    Components:
        - Action Library: Catalog of available actions with specifications
        - Planner: Generates action sequences to achieve goals
        - Dependency Analyzer: Identifies action dependencies and constraints
        - Feasibility Checker: Validates plan viability
        - Executor: Executes plans with monitoring and error handling
    
    Use Cases:
        - Multi-step task automation
        - Workflow orchestration
        - Robot task planning
        - Complex problem decomposition
        - Project planning and execution
    
    LangChain Implementation:
        Uses LLM for action selection and ordering, implements STRIPS-like
        planning with preconditions and effects, handles dynamic replanning.
    
    Key Features:
        - Action dependency analysis
        - Precondition and effect modeling
        - Parallel action detection
        - Plan validation
        - Dynamic replanning on failures
    """
    
    import os
    from typing import List, Dict, Any, Optional, Set
    from dataclasses import dataclass, field
    from enum import Enum
    from datetime import datetime
    from dotenv import load_dotenv
    
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser
    from langchain_openai import ChatOpenAI
    
    load_dotenv()
    
    
    class ActionStatus(Enum):
        """Status of action execution."""
        PENDING = "pending"
        READY = "ready"
        EXECUTING = "executing"
        COMPLETED = "completed"
        FAILED = "failed"
        SKIPPED = "skipped"
    
    
    @dataclass
    class ActionDefinition:
        """Definition of an available action."""
        name: str
        description: str
        preconditions: List[str]
        effects: List[str]
        duration: float = 1.0  # Estimated duration
        
        def to_dict(self) -> Dict[str, Any]:
            """Convert to dictionary for LLM context."""
            return {
                "name": self.name,
                "description": self.description,
                "preconditions": self.preconditions,
                "effects": self.effects
            }
    
    
    @dataclass
    class PlannedAction:
        """An action in a plan."""
        action: ActionDefinition
        step_number: int
        dependencies: List[int] = field(default_factory=list)
        status: ActionStatus = ActionStatus.PENDING
        result: Optional[str] = None
        error: Optional[str] = None
        start_time: Optional[datetime] = None
        end_time: Optional[datetime] = None
    
    
    @dataclass
    class ActionPlan:
        """Complete action plan."""
        goal: str
        actions: List[PlannedAction]
        initial_state: Set[str]
        final_state: Set[str]
        timestamp: datetime = field(default_factory=datetime.now)
        
        def get_ready_actions(self) -> List[PlannedAction]:
            """Get actions that are ready to execute."""
            ready = []
            for action in self.actions:
                if action.status == ActionStatus.PENDING:
                    # Check if all dependencies completed
                    dependencies_met = all(
                        self.actions[dep].status == ActionStatus.COMPLETED
                        for dep in action.dependencies
                    )
                    if dependencies_met:
                        ready.append(action)
            return ready
        
        def is_complete(self) -> bool:
            """Check if plan is complete."""
            return all(
                action.status in [ActionStatus.COMPLETED, ActionStatus.SKIPPED]
                for action in self.actions
            )
    
    
    class ActionLibrary:
        """
        Library of available actions.
        """
        
        def __init__(self):
            """Initialize action library with common actions."""
            self.actions: Dict[str, ActionDefinition] = {}
            self._register_builtin_actions()
        
        def _register_builtin_actions(self):
            """Register built-in actions."""
            
            # Data collection actions
            self.register_action(ActionDefinition(
                name="gather_requirements",
                description="Collect and document project requirements",
                preconditions=["project_defined"],
                effects=["requirements_gathered"]
            ))
            
            self.register_action(ActionDefinition(
                name="research_topic",
                description="Research and gather information on a topic",
                preconditions=[],
                effects=["research_completed", "information_gathered"]
            ))
            
            # Analysis actions
            self.register_action(ActionDefinition(
                name="analyze_data",
                description="Analyze collected data or information",
                preconditions=["data_available"],
                effects=["analysis_completed", "insights_generated"]
            ))
            
            self.register_action(ActionDefinition(
                name="identify_risks",
                description="Identify and assess potential risks",
                preconditions=["requirements_gathered"],
                effects=["risks_identified"]
            ))
            
            # Design actions
            self.register_action(ActionDefinition(
                name="create_design",
                description="Create system or solution design",
                preconditions=["requirements_gathered", "research_completed"],
                effects=["design_created"]
            ))
            
            self.register_action(ActionDefinition(
                name="review_design",
                description="Review and validate design",
                preconditions=["design_created"],
                effects=["design_reviewed"]
            ))
            
            # Implementation actions
            self.register_action(ActionDefinition(
                name="implement_solution",
                description="Implement the designed solution",
                preconditions=["design_reviewed"],
                effects=["solution_implemented"]
            ))
            
            self.register_action(ActionDefinition(
                name="write_documentation",
                description="Write technical documentation",
                preconditions=["solution_implemented"],
                effects=["documentation_written"]
            ))
            
            # Testing actions
            self.register_action(ActionDefinition(
                name="create_test_plan",
                description="Create comprehensive test plan",
                preconditions=["design_created"],
                effects=["test_plan_created"]
            ))
            
            self.register_action(ActionDefinition(
                name="execute_tests",
                description="Execute test cases and validate solution",
                preconditions=["solution_implemented", "test_plan_created"],
                effects=["tests_executed", "quality_validated"]
            ))
            
            # Deployment actions
            self.register_action(ActionDefinition(
                name="prepare_deployment",
                description="Prepare solution for deployment",
                preconditions=["tests_executed", "documentation_written"],
                effects=["deployment_ready"]
            ))
            
            self.register_action(ActionDefinition(
                name="deploy_solution",
                description="Deploy solution to production",
                preconditions=["deployment_ready"],
                effects=["solution_deployed"]
            ))
        
        def register_action(self, action: ActionDefinition):
            """Register a new action."""
            self.actions[action.name] = action
        
        def get_action(self, name: str) -> Optional[ActionDefinition]:
            """Get action by name."""
            return self.actions.get(name)
        
        def get_all_actions_description(self) -> str:
            """Get formatted description of all actions."""
            descriptions = []
            for action in self.actions.values():
                precond = ", ".join(action.preconditions) if action.preconditions else "none"
                effects = ", ".join(action.effects)
                descriptions.append(
                    f"- {action.name}: {action.description}\n"
                    f"  Preconditions: {precond}\n"
                    f"  Effects: {effects}"
                )
            return "\n\n".join(descriptions)
    
    
    class ActionPlanner:
        """
        Plans action sequences to achieve goals.
        """
        
        def __init__(
            self,
            action_library: ActionLibrary,
            model: str = "gpt-3.5-turbo"
        ):
            """
            Initialize action planner.
            
            Args:
                action_library: Library of available actions
                model: LLM model to use
            """
            self.action_library = action_library
            self.llm = ChatOpenAI(model=model, temperature=0.3)
        
        def create_plan(
            self,
            goal: str,
            initial_state: Optional[Set[str]] = None
        ) -> ActionPlan:
            """
            Create action plan to achieve goal.
            
            Args:
                goal: Goal to achieve
                initial_state: Initial world state
                
            Returns:
                Action plan
            """
            if initial_state is None:
                initial_state = set()
            
            actions_desc = self.action_library.get_all_actions_description()
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", """You are an expert action planner. Create a sequence of actions
    to achieve the given goal, respecting preconditions and dependencies.
    
    Available Actions:
    {actions_description}
    
    Your response should list actions in order, one per line, using exact action names.
    Only use actions from the available list."""),
                ("user", """Goal: {goal}
    
    Initial State: {initial_state}
    
    Create an action plan listing actions in execution order:""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            
            initial_state_str = ", ".join(initial_state) if initial_state else "none"
            
            response = chain.invoke({
                "actions_description": actions_desc,
                "goal": goal,
                "initial_state": initial_state_str
            })
            
            # Parse action names
            action_names = self._parse_action_names(response)
            
            # Build plan with dependencies
            planned_actions = self._build_plan_with_dependencies(
                action_names,
                initial_state
            )
            
            # Calculate final state
            final_state = self._calculate_final_state(planned_actions, initial_state)
            
            return ActionPlan(
                goal=goal,
                actions=planned_actions,
                initial_state=initial_state,
                final_state=final_state
            )
        
        def _parse_action_names(self, response: str) -> List[str]:
            """Parse action names from LLM response."""
            lines = response.strip().split("\n")
            action_names = []
            
            for line in lines:
                line = line.strip()
                # Remove numbering, bullets, etc.
                for prefix in ["1.", "2.", "3.", "4.", "5.", "6.", "7.", "8.", "9.", "-", "*", "•"]:
                    if line.startswith(prefix):
                        line = line[len(prefix):].strip()
                
                # Check if this is a valid action name
                if self.action_library.get_action(line):
                    action_names.append(line)
            
            return action_names
        
        def _build_plan_with_dependencies(
            self,
            action_names: List[str],
            initial_state: Set[str]
        ) -> List[PlannedAction]:
            """Build plan with dependency analysis."""
            planned_actions = []
            current_state = initial_state.copy()
            
            for step_num, action_name in enumerate(action_names):
                action_def = self.action_library.get_action(action_name)
                if not action_def:
                    continue
                
                # Find dependencies (actions that provide required preconditions)
                dependencies = []
                for precond in action_def.preconditions:
                    if precond not in current_state:
                        # Find which previous action provides this
                        for i, prev_action in enumerate(planned_actions):
                            if precond in prev_action.action.effects:
                                dependencies.append(i)
                                break
                
                planned_action = PlannedAction(
                    action=action_def,
                    step_number=step_num,
                    dependencies=dependencies
                )
                
                planned_actions.append(planned_action)
                
                # Update state with effects
                current_state.update(action_def.effects)
            
            return planned_actions
        
        def _calculate_final_state(
            self,
            planned_actions: List[PlannedAction],
            initial_state: Set[str]
        ) -> Set[str]:
            """Calculate final state after all actions."""
            final_state = initial_state.copy()
            
            for planned_action in planned_actions:
                final_state.update(planned_action.action.effects)
            
            return final_state
    
    
    class PlanExecutor:
        """
        Executes action plans.
        """
        
        def __init__(self):
            """Initialize plan executor."""
            pass
        
        def execute_plan(self, plan: ActionPlan) -> Dict[str, Any]:
            """
            Execute action plan.
            
            Args:
                plan: Action plan to execute
                
            Returns:
                Execution results
            """
            print(f"\n[Executor] Executing plan for goal: {plan.goal}")
            print(f"[Executor] Total actions: {len(plan.actions)}\n")
            
            completed = 0
            failed = 0
            
            while not plan.is_complete():
                # Get ready actions
                ready_actions = plan.get_ready_actions()
                
                if not ready_actions:
                    # Check if we have pending actions that can't proceed
                    pending = [a for a in plan.actions if a.status == ActionStatus.PENDING]
                    if pending:
                        print("[Executor] ✗ Deadlock: Actions have unmet dependencies")
                        for action in pending:
                            action.status = ActionStatus.FAILED
                            action.error = "Unmet dependencies"
                            failed += 1
                    break
                
                # Execute ready actions
                for action in ready_actions:
                    self._execute_action(action)
                    
                    if action.status == ActionStatus.COMPLETED:
                        completed += 1
                    elif action.status == ActionStatus.FAILED:
                        failed += 1
            
            success_rate = completed / len(plan.actions) * 100 if plan.actions else 0
            
            return {
                "goal": plan.goal,
                "total_actions": len(plan.actions),
                "completed": completed,
                "failed": failed,
                "success_rate": success_rate,
                "plan": plan
            }
        
        def _execute_action(self, action: PlannedAction):
            """Execute a single action."""
            action.status = ActionStatus.EXECUTING
            action.start_time = datetime.now()
            
            print(f"[Step {action.step_number + 1}] Executing: {action.action.name}")
            print(f"  Description: {action.action.description}")
            
            if action.dependencies:
                deps_str = ", ".join([str(d + 1) for d in action.dependencies])
                print(f"  Dependencies: Steps {deps_str}")
            
            # Simulate action execution (in real system, would call actual function)
            try:
                # Simulate success
                action.result = f"Successfully completed {action.action.name}"
                action.status = ActionStatus.COMPLETED
                print(f"  ✓ Completed")
                print(f"  Effects: {', '.join(action.action.effects)}")
                
            except Exception as e:
                action.error = str(e)
                action.status = ActionStatus.FAILED
                print(f"  ✗ Failed: {action.error}")
            
            action.end_time = datetime.now()
            print()
    
    
    class ActionSequencePlanningAgent:
        """
        Agent that plans and executes action sequences.
        """
        
        def __init__(self, model: str = "gpt-3.5-turbo"):
            """
            Initialize action sequence planning agent.
            
            Args:
                model: LLM model to use
            """
            self.action_library = ActionLibrary()
            self.planner = ActionPlanner(self.action_library, model)
            self.executor = PlanExecutor()
        
        def achieve_goal(
            self,
            goal: str,
            initial_state: Optional[Set[str]] = None
        ) -> Dict[str, Any]:
            """
            Plan and execute actions to achieve goal.
            
            Args:
                goal: Goal to achieve
                initial_state: Initial world state
                
            Returns:
                Execution results
            """
            print(f"\n{'=' * 80}")
            print(f"GOAL: {goal}")
            print("=" * 80)
            
            # Create plan
            print("\n[Agent] Creating action plan...")
            plan = self.planner.create_plan(goal, initial_state)
            
            print(f"\n[Agent] Plan created with {len(plan.actions)} actions")
            print("\nACTION PLAN:")
            print("-" * 80)
            
            for action in plan.actions:
                deps = ""
                if action.dependencies:
                    deps = f" (depends on: {[d+1 for d in action.dependencies]})"
                print(f"{action.step_number + 1}. {action.action.name}{deps}")
                print(f"   {action.action.description}")
            
            print("-" * 80)
            print(f"\nInitial state: {', '.join(plan.initial_state) if plan.initial_state else 'none'}")
            print(f"Expected final state: {', '.join(plan.final_state)}")
            
            # Execute plan
            print(f"\n{'=' * 80}")
            print("EXECUTION")
            print("=" * 80)
            
            result = self.executor.execute_plan(plan)
            
            # Summary
            print("=" * 80)
            print("EXECUTION SUMMARY")
            print("=" * 80)
            print(f"Goal: {result['goal']}")
            print(f"Total actions: {result['total_actions']}")
            print(f"Completed: {result['completed']}")
            print(f"Failed: {result['failed']}")
            print(f"Success rate: {result['success_rate']:.0f}%")
            
            return result
    
    
    def demonstrate_action_sequence_planning():
        """Demonstrate the Action Sequence Planning pattern."""
        
        print("=" * 80)
        print("ACTION SEQUENCE PLANNING PATTERN DEMONSTRATION")
        print("=" * 80)
        
        agent = ActionSequencePlanningAgent()
        
        # Show available actions
        print("\n" + "=" * 80)
        print("AVAILABLE ACTIONS")
        print("=" * 80)
        print(agent.action_library.get_all_actions_description())
        
        # Test 1: Software project workflow
        print("\n\n" + "=" * 80)
        print("TEST 1: Software Development Project")
        print("=" * 80)
        
        goal1 = "Complete a software development project from requirements to deployment"
        initial_state1 = {"project_defined"}
        
        result1 = agent.achieve_goal(goal1, initial_state1)
        
        # Test 2: Research and analysis task
        print("\n\n" + "=" * 80)
        print("TEST 2: Research and Analysis Task")
        print("=" * 80)
        
        goal2 = "Research a topic, analyze findings, and create a report"
        initial_state2 = set()
        
        result2 = agent.achieve_goal(goal2, initial_state2)
        
        # Test 3: Design and implementation
        print("\n\n" + "=" * 80)
        print("TEST 3: Design and Implementation")
        print("=" * 80)
        
        goal3 = "Design, implement, and test a new feature"
        initial_state3 = {"requirements_gathered", "research_completed"}
        
        result3 = agent.achieve_goal(goal3, initial_state3)
        
        # Overall summary
        print("\n\n" + "=" * 80)
        print("PATTERN SUMMARY")
        print("=" * 80)
        print("""
    The Action Sequence Planning pattern demonstrates:
    
    1. **Goal Decomposition**: Breaks complex goals into action sequences
    2. **Dependency Analysis**: Identifies action dependencies and ordering
    3. **Precondition Checking**: Ensures actions are feasible
    4. **Effect Modeling**: Tracks state changes from actions
    5. **Sequential Execution**: Executes plan respecting dependencies
    
    Key Benefits:
    - **Structured Approach**: Systematic goal achievement
    - **Dependency Management**: Proper action ordering
    - **State Tracking**: Monitors world state changes
    - **Validation**: Ensures plan feasibility
    - **Transparency**: Clear action sequences and reasoning
    
    Use Cases:
    - Multi-step task automation
    - Workflow orchestration
    - Project planning and execution
    - Robot task planning
    - Business process automation
    
    Planning Principles:
    - Actions have clear preconditions and effects
    - Dependencies are automatically inferred
    - Plans respect causal relationships
    - State evolves through action effects
    - Parallel execution possible when no dependencies
    
    This pattern enables agents to tackle complex, multi-step goals
    through structured planning and systematic execution.
    """)
    
    
    if __name__ == "__main__":
        demonstrate_action_sequence_planning()
    
        

    Agentic Design Pattern: Action Sequence Planning

    Pattern 025: Action Sequence Planning – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Action Sequence Planning pattern implementation.


    Pattern Description

    Pattern 025: Action Sequence Planning

    Description: The Action Sequence Planning pattern enables agents to decompose complex goals into ordered sequences of actions with preconditions, effects, and dependencies. The agent analyzes the goal, identifies required actions, determines ordering constraints, validates feasibility, and executes the plan step-by-step with monitoring and replanning capabilities.

    Components: – Action Library: Catalog of available actions with specifications – Planner: Generates action sequences to achieve goals – Dependency Analyzer: Identifies action dependencies and constraints – Feasibility Checker: Validates plan viability – Executor: Executes plans with monitoring and error handling

    Use Cases: – Multi-step task automation – Workflow orchestration – Robot task planning – Complex problem decomposition – Project planning and execution

    LangChain Implementation: Uses LLM for action selection and ordering, implements STRIPS-like planning with preconditions and effects, handles dynamic replanning.

    Key Features: – Action dependency analysis – Precondition and effect modeling – Parallel action detection – Plan validation – Dynamic replanning on failures


    File Statistics

    • Total Lines: 652
    • Classes: 8
    • Functions: 1
    • Constants: 0

    Module Documentation

    """
    Pattern 025: Action Sequence Planning
    
    Description:
        The Action Sequence Planning pattern enables agents to decompose complex goals
        into ordered sequences of actions with preconditions, effects, and dependencies.
        The agent analyzes the goal, identifies required actions, determines ordering
        constraints, validates feasibility, and executes the plan step-by-step with
        monitoring and replanning capabilities.
    
    Components:
        - Action Library: Catalog of available actions with specifications
        - Planner: Generates action sequences to achieve goals
        - Dependency Analyzer: Identifies action dependencies and constraints
        - Feasibility Checker: Validates plan viability
        - Executor: Executes plans with monitoring and error handling
    
    Use Cases:
        - Multi-step task automation
        - Workflow orchestration
        - Robot task planning
        - Complex problem decomposition
        - Project planning and execution
    
    LangChain Implementation:
        Uses LLM for action selection and ordering, implements STRIPS-like
        planning with preconditions and effects, handles dynamic replanning.
    
    Key Features:
        - Action dependency analysis
        - Precondition and effect modeling
        - Parallel action detection
        - Plan validation
        - Dynamic replanning on failures
    """
    

    Purpose: This module implements the Action Sequence Planning 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.output_parsers.StrOutputParser
    import langchain_core.prompts.ChatPromptTemplate
    import os
    import typing.Any
    import typing.Dict
    import typing.List
    import typing.Optional
    import typing.Set
    

    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.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: 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) – typing.Set: 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: ActionStatus

    Lines 51-58

    Inherits from: Enum

    Purpose:

    Status of action execution.
    

    Class: ActionDefinition

    Lines 62-77

    Purpose:

    Definition of an available action.
    

    Methods:

    to_dict(self)

    Lines 70-77

    Purpose:

    Convert to dictionary for LLM context.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D60F5D0>


    Class: PlannedAction

    Lines 81-90

    Purpose:

    An action in a plan.
    

    Class: ActionPlan

    Lines 94-121

    Purpose:

    Complete action plan.
    

    Methods:

    get_ready_actions(self)

    Lines 102-114

    Purpose:

    Get actions that are ready to execute.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D645450>

    is_complete(self)

    Lines 116-121

    Purpose:

    Check if plan is complete.
    

    Parameters:

    Returns: bool


    Class: ActionLibrary

    Lines 124-246

    Purpose:

    Library of available actions.
    

    Methods:

    __init__(self)

    Lines 129-132

    Purpose:

    Initialize action library with common actions.
    

    Parameters:

    _register_builtin_actions(self)

    Lines 134-225

    Purpose:

    Register built-in actions.
    

    Parameters:

    register_action(self, action)

    Lines 227-229

    Purpose:

    Register a new action.
    

    Parameters: – action: Action

    get_action(self, name)

    Lines 231-233

    Purpose:

    Get action by name.
    

    Parameters: – name: Name

    Returns: <ast.Subscript object at 0x000001A06D7512D0>

    get_all_actions_description(self)

    Lines 235-246

    Purpose:

    Get formatted description of all actions.
    

    Parameters:

    Returns: str


    Class: ActionPlanner

    Lines 249-400

    Purpose:

    Plans action sequences to achieve goals.
    

    Methods:

    __init__(self, action_library, model)

    Lines 254-267

    Purpose:

    Initialize action planner.
    
    Args:
        action_library: Library of available actions
        model: LLM model to use
    

    Parameters: – action_library: Action library – model: Language model instance

    create_plan(self, goal, initial_state)

    Lines 269-332

    Purpose:

    Create action plan to achieve goal.
    
    Args:
        goal: Goal to achieve
        initial_state: Initial world state
    
    Returns:
        Action plan
    

    Parameters: – goal: Goal – initial_state: Initial state

    Returns: ActionPlan

    _parse_action_names(self, response)

    Lines 334-350

    Purpose:

    Parse action names from LLM response.
    

    Parameters: – response: Response

    Returns: <ast.Subscript object at 0x000001A06D648550>

    _build_plan_with_dependencies(self, action_names, initial_state)

    Lines 352-387

    Purpose:

    Build plan with dependency analysis.
    

    Parameters: – action_names: Action names – initial_state: Initial state

    Returns: <ast.Subscript object at 0x000001A06D63C510>

    _calculate_final_state(self, planned_actions, initial_state)

    Lines 389-400

    Purpose:

    Calculate final state after all actions.
    

    Parameters: – planned_actions: Planned actions – initial_state: Initial state

    Returns: <ast.Subscript object at 0x000001A06D6B2710>


    Class: PlanExecutor

    Lines 403-489

    Purpose:

    Executes action plans.
    

    Methods:

    __init__(self)

    Lines 408-410

    Purpose:

    Initialize plan executor.
    

    Parameters:

    execute_plan(self, plan)

    Lines 412-461

    Purpose:

    Execute action plan.
    
    Args:
        plan: Action plan to execute
    
    Returns:
        Execution results
    

    Parameters: – plan: Plan

    Returns: <ast.Subscript object at 0x000001A06D64FF10>

    _execute_action(self, action)

    Lines 463-489

    Purpose:

    Execute a single action.
    

    Parameters: – action: Action


    Class: ActionSequencePlanningAgent

    Lines 492-563

    Purpose:

    Agent that plans and executes action sequences.
    

    Methods:

    __init__(self, model)

    Lines 497-506

    Purpose:

    Initialize action sequence planning agent.
    
    Args:
        model: LLM model to use
    

    Parameters: – model: Language model instance

    achieve_goal(self, goal, initial_state)

    Lines 508-563

    Purpose:

    Plan and execute actions to achieve goal.
    
    Args:
        goal: Goal to achieve
        initial_state: Initial world state
    
    Returns:
        Execution results
    

    Parameters: – goal: Goal – initial_state: Initial state

    Returns: <ast.Subscript object at 0x000001A06D6A8650>


    Functions

    demonstrate_action_sequence_planning()

    Lines 566-647

    Purpose:

    Demonstrate the Action Sequence Planning pattern.
    

    Key Concepts Summary

    Pattern Overview

    The Action Sequence Planning 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

    1. Agent Class: Main implementation of the pattern logic
    2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
    3. Tool Integration: Optional tools for extended functionality
    4. Memory Management: State handling across interactions
    5. 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

    1. Prompt Optimization: Minimize tokens while maintaining clarity
    2. Caching: Cache repeated queries/responses
    3. Model Selection: Use smaller models when appropriate
    4. Batch Processing: Group similar requests
    5. 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

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

    Feature Extensions

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

    Code Quality

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

    Related Patterns

    • Pattern 024
    • Pattern 026
    • Other Tool Use Patterns (020-025)

    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 Action Sequence Planning 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: 025_action_sequence_planning.py
    Total Lines: 652
    Pattern Category: AI Agent Design Pattern #025


    Generated from actual implementation code
    For educational and reference purposes