Tag: design

  • Agentic Design Pattern: Inductive Reasoning

    
        """
    Pattern 160: Inductive Reasoning
    
    Description:
        Inductive reasoning generalizes from specific observations to broader patterns
        or rules. It moves from particular instances to general conclusions, inferring
        patterns that likely hold across similar cases.
    
    Components:
        - Example collection
        - Pattern identification
        - Rule generalization
        - Confidence assessment
        - Counter-example checking
    
    Use Cases:
        - Pattern discovery
        - Rule learning
        - Trend analysis
        - Scientific generalization
        - Predictive modeling
    
    Benefits:
        - Discovers patterns from data
        - Enables prediction
        - Learns from examples
        - Practical applicability
    
    Trade-offs:
        - Conclusions not guaranteed
        - Vulnerable to biased samples
        - Requires sufficient examples
        - May over-generalize
    
    LangChain Implementation:
        Uses ChatOpenAI for pattern recognition, rule extraction,
        and validation through additional examples
    """
    
    import os
    from typing import List, Dict, Any, Optional
    from dataclasses import dataclass, field
    from dotenv import load_dotenv
    
    from langchain_openai import ChatOpenAI
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser
    import json
    
    load_dotenv()
    
    
    @dataclass
    class Example:
        """Single example/observation"""
        input: str
        output: str
        features: Dict[str, Any] = field(default_factory=dict)
        metadata: Dict[str, Any] = field(default_factory=dict)
    
    
    @dataclass
    class Pattern:
        """Discovered pattern"""
        description: str
        rule: str
        confidence: float  # 0-1
        supporting_examples: List[str] = field(default_factory=list)
        counter_examples: List[str] = field(default_factory=list)
        generality: float = 0.5  # How general vs specific
    
    
    @dataclass
    class InductiveAnalysis:
        """Result of inductive reasoning"""
        examples: List[Example]
        patterns: List[Pattern]
        generalizations: List[str] = field(default_factory=list)
        confidence: float = 0.0
        reasoning_trace: List[str] = field(default_factory=list)
    
    
    class InductiveReasoningAgent:
        """Agent that performs inductive reasoning"""
        
        def __init__(self, model_name: str = "gpt-3.5-turbo"):
            """
            Initialize inductive reasoning agent
            
            Args:
                model_name: LLM model to use
            """
            self.llm = ChatOpenAI(model=model_name, temperature=0.7)
            self.validator_llm = ChatOpenAI(model=model_name, temperature=0)
        
        def reason(self, examples: List[Example],
                   domain: Optional[str] = None,
                   min_confidence: float = 0.5) -> InductiveAnalysis:
            """
            Perform inductive reasoning on examples
            
            Args:
                examples: List of examples to learn from
                domain: Optional domain context
                min_confidence: Minimum confidence threshold
                
            Returns:
                InductiveAnalysis with discovered patterns
            """
            reasoning_trace = []
            
            # Step 1: Identify patterns
            reasoning_trace.append(f"Analyzing {len(examples)} examples...")
            patterns = self._identify_patterns(examples, domain)
            reasoning_trace.append(f"Identified {len(patterns)} potential patterns")
            
            # Step 2: Validate patterns
            reasoning_trace.append("Validating patterns against examples...")
            for pattern in patterns:
                self._validate_pattern(pattern, examples)
            
            # Step 3: Filter by confidence
            patterns = [p for p in patterns if p.confidence >= min_confidence]
            reasoning_trace.append(f"Retained {len(patterns)} high-confidence patterns")
            
            # Step 4: Generate generalizations
            generalizations = self._generate_generalizations(patterns, examples)
            reasoning_trace.append(f"Generated {len(generalizations)} generalizations")
            
            # Step 5: Calculate overall confidence
            if patterns:
                avg_confidence = sum(p.confidence for p in patterns) / len(patterns)
            else:
                avg_confidence = 0.0
            
            return InductiveAnalysis(
                examples=examples,
                patterns=patterns,
                generalizations=generalizations,
                confidence=avg_confidence,
                reasoning_trace=reasoning_trace
            )
        
        def _identify_patterns(self, examples: List[Example],
                              domain: Optional[str]) -> List[Pattern]:
            """Identify patterns from examples"""
            # Format examples
            examples_text = "\n".join([
                f"Input: {ex.input} → Output: {ex.output}"
                for ex in examples
            ])
            
            # Create pattern identification prompt
            prompt = ChatPromptTemplate.from_messages([
                ("system", """You are an expert at identifying patterns from examples.
    Analyze the given input-output examples and identify patterns or rules.
    
    {domain_context}
    
    For each pattern, provide:
    1. Clear description
    2. General rule
    3. How general vs specific it is (0-1)
    
    Return as JSON array: [{{"description": "...", "rule": "...", "generality": 0.0-1.0}}]"""),
                ("user", "Examples:\n{examples}\n\nIdentify patterns:")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            
            domain_context = f"Domain: {domain}" if domain else "Consider general patterns."
            
            response = chain.invoke({
                "examples": examples_text,
                "domain_context": domain_context
            })
            
            # Parse patterns
            patterns = self._parse_patterns(response)
            return patterns
        
        def _parse_patterns(self, response: str) -> List[Pattern]:
            """Parse patterns from LLM response"""
            try:
                # Try to extract JSON
                start = response.find('[')
                end = response.rfind(']') + 1
                if start >= 0 and end > start:
                    json_str = response[start:end]
                    data = json.loads(json_str)
                    
                    patterns = []
                    for item in data:
                        pattern = Pattern(
                            description=item.get('description', ''),
                            rule=item.get('rule', ''),
                            confidence=0.5,  # Initial, will validate
                            generality=float(item.get('generality', 0.5))
                        )
                        patterns.append(pattern)
                    return patterns
            except Exception as e:
                print(f"Error parsing patterns: {e}")
            
            # Fallback parsing
            patterns = []
            lines = response.split('\n')
            current_pattern = {}
            
            for line in lines:
                line = line.strip()
                if line.startswith(('1.', '2.', '3.', '4.', '5.')):
                    if current_pattern.get('description'):
                        patterns.append(Pattern(
                            description=current_pattern.get('description', ''),
                            rule=current_pattern.get('rule', ''),
                            confidence=0.5,
                            generality=0.5
                        ))
                    current_pattern = {'description': line[2:].strip()}
                elif 'rule:' in line.lower():
                    current_pattern['rule'] = line.split(':', 1)[1].strip()
            
            # Add last pattern
            if current_pattern.get('description'):
                patterns.append(Pattern(
                    description=current_pattern.get('description', ''),
                    rule=current_pattern.get('rule', ''),
                    confidence=0.5,
                    generality=0.5
                ))
            
            return patterns[:5]  # Limit to 5
        
        def _validate_pattern(self, pattern: Pattern, examples: List[Example]):
            """Validate pattern against examples"""
            supporting = 0
            contradicting = 0
            
            for example in examples:
                # Check if pattern applies to this example
                prompt = ChatPromptTemplate.from_messages([
                    ("system", "Determine if the pattern/rule applies to this example. Respond with 'yes' or 'no'."),
                    ("user", """Pattern: {pattern}
    Rule: {rule}
    
    Example - Input: {input}, Output: {output}
    
    Does the pattern apply?""")
                ])
                
                chain = prompt | self.validator_llm | StrOutputParser()
                
                response = chain.invoke({
                    "pattern": pattern.description,
                    "rule": pattern.rule,
                    "input": example.input,
                    "output": example.output
                })
                
                if 'yes' in response.lower():
                    supporting += 1
                    pattern.supporting_examples.append(f"{example.input} → {example.output}")
                else:
                    contradicting += 1
                    pattern.counter_examples.append(f"{example.input} → {example.output}")
            
            # Calculate confidence
            total = supporting + contradicting
            if total > 0:
                pattern.confidence = supporting / total
            else:
                pattern.confidence = 0.0
        
        def _generate_generalizations(self, patterns: List[Pattern],
                                     examples: List[Example]) -> List[str]:
            """Generate general rules from patterns"""
            if not patterns:
                return []
            
            # Format patterns
            patterns_text = "\n".join([
                f"- {p.description} (confidence: {p.confidence:.0%})"
                for p in patterns
            ])
            
            # Create generalization prompt
            prompt = ChatPromptTemplate.from_messages([
                ("system", "Synthesize the patterns into broader generalizations or principles."),
                ("user", """Patterns discovered:
    {patterns}
    
    Generate 2-3 general rules or principles that capture these patterns:""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            
            response = chain.invoke({"patterns": patterns_text})
            
            # Parse generalizations
            generalizations = []
            for line in response.split('\n'):
                line = line.strip()
                if line and (line[0].isdigit() or line.startswith('-')):
                    clean_line = line.lstrip('0123456789.-) ')
                    if clean_line:
                        generalizations.append(clean_line)
            
            return generalizations
        
        def predict(self, analysis: InductiveAnalysis, new_input: str) -> Dict[str, Any]:
            """Use learned patterns to make a prediction"""
            # Format patterns
            patterns_text = "\n".join([
                f"- {p.rule} (confidence: {p.confidence:.0%})"
                for p in analysis.patterns
            ])
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", """Use the learned patterns to predict the output for the new input.
    Explain which pattern(s) you're applying."""),
                ("user", """Learned patterns:
    {patterns}
    
    New input: {input}
    
    Predict the output and explain your reasoning:""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            
            response = chain.invoke({
                "patterns": patterns_text,
                "input": new_input
            })
            
            return {
                "input": new_input,
                "prediction": response,
                "based_on_patterns": len(analysis.patterns),
                "confidence": analysis.confidence
            }
    
    
    def demonstrate_inductive_reasoning():
        """Demonstrate inductive reasoning pattern"""
        print("=" * 80)
        print("INDUCTIVE REASONING PATTERN DEMONSTRATION")
        print("=" * 80)
        
        agent = InductiveReasoningAgent()
        
        # Example 1: Number patterns
        print("\n" + "="*80)
        print("EXAMPLE 1: Number Pattern Discovery")
        print("="*80)
        examples = [
            Example(input="2", output="4"),
            Example(input="3", output="9"),
            Example(input="4", output="16"),
            Example(input="5", output="25"),
            Example(input="6", output="36"),
        ]
        
        analysis = agent.reason(examples, domain="Mathematics")
        
        print("Examples:")
        for ex in examples:
            print(f"  {ex.input} → {ex.output}")
        
        print(f"\nDiscovered Patterns:")
        for i, pattern in enumerate(analysis.patterns, 1):
            print(f"\n{i}. {pattern.description}")
            print(f"   Rule: {pattern.rule}")
            print(f"   Confidence: {pattern.confidence:.0%}")
            print(f"   Generality: {pattern.generality:.1f}")
        
        # Test prediction
        if analysis.patterns:
            prediction = agent.predict(analysis, "7")
            print(f"\nPrediction for '7': {prediction['prediction'][:200]}...")
        
        # Example 2: Text transformation
        print("\n" + "="*80)
        print("EXAMPLE 2: Text Transformation Pattern")
        print("="*80)
        examples = [
            Example(input="cat", output="cats"),
            Example(input="dog", output="dogs"),
            Example(input="house", output="houses"),
            Example(input="car", output="cars"),
            Example(input="book", output="books"),
        ]
        
        analysis = agent.reason(examples, domain="English grammar")
        
        print("Examples:")
        for ex in examples:
            print(f"  {ex.input} → {ex.output}")
        
        print(f"\nPattern Analysis:")
        for pattern in analysis.patterns:
            print(f"\n- {pattern.description}")
            print(f"  Confidence: {pattern.confidence:.0%}")
            if pattern.supporting_examples:
                print(f"  Supporting: {len(pattern.supporting_examples)} examples")
        
        print(f"\nGeneralizations:")
        for gen in analysis.generalizations:
            print(f"  - {gen}")
        
        # Example 3: Category classification
        print("\n" + "="*80)
        print("EXAMPLE 3: Classification Pattern")
        print("="*80)
        examples = [
            Example(input="apple", output="fruit"),
            Example(input="banana", output="fruit"),
            Example(input="carrot", output="vegetable"),
            Example(input="broccoli", output="vegetable"),
            Example(input="orange", output="fruit"),
            Example(input="spinach", output="vegetable"),
        ]
        
        analysis = agent.reason(examples, domain="Food classification")
        
        print("Examples:")
        for ex in examples:
            print(f"  {ex.input} → {ex.output}")
        
        print(f"\nDiscovered Rules:")
        for pattern in analysis.patterns:
            print(f"\n- {pattern.rule}")
            print(f"  Confidence: {pattern.confidence:.0%}")
        
        # Test with new input
        prediction = agent.predict(analysis, "tomato")
        print(f"\nPrediction for 'tomato':")
        print(f"  {prediction['prediction'][:300]}...")
        
        # Example 4: Sequence patterns
        print("\n" + "="*80)
        print("EXAMPLE 4: Sequence Pattern")
        print("="*80)
        examples = [
            Example(input="position 1", output="2"),
            Example(input="position 2", output="4"),
            Example(input="position 3", output="6"),
            Example(input="position 4", output="8"),
            Example(input="position 5", output="10"),
        ]
        
        analysis = agent.reason(examples, domain="Sequences")
        
        print("Examples:")
        for ex in examples:
            print(f"  {ex.input} → {ex.output}")
        
        print(f"\nPattern:")
        if analysis.patterns:
            pattern = analysis.patterns[0]
            print(f"  {pattern.description}")
            print(f"  Rule: {pattern.rule}")
            print(f"  Confidence: {pattern.confidence:.0%}")
        
        # Example 5: Behavior pattern
        print("\n" + "="*80)
        print("EXAMPLE 5: Behavioral Pattern")
        print("="*80)
        examples = [
            Example(input="Monday morning", output="heavy traffic"),
            Example(input="Tuesday morning", output="heavy traffic"),
            Example(input="Wednesday morning", output="heavy traffic"),
            Example(input="Saturday morning", output="light traffic"),
            Example(input="Sunday morning", output="light traffic"),
        ]
        
        analysis = agent.reason(examples, domain="Traffic patterns")
        
        print("Examples:")
        for ex in examples:
            print(f"  {ex.input} → {ex.output}")
        
        print(f"\nInduced Patterns:")
        for i, pattern in enumerate(analysis.patterns, 1):
            print(f"\n{i}. {pattern.description}")
            print(f"   Confidence: {pattern.confidence:.0%}")
        
        print(f"\nGeneralizations:")
        for gen in analysis.generalizations:
            print(f"  - {gen}")
        
        # Prediction
        prediction = agent.predict(analysis, "Thursday morning")
        print(f"\nPrediction for 'Thursday morning':")
        print(f"  {prediction['prediction'][:200]}...")
        
        # Summary
        print("\n" + "="*80)
        print("SUMMARY: Inductive Reasoning Best Practices")
        print("="*80)
        print("""
    1. EXAMPLE COLLECTION:
       - Gather diverse, representative examples
       - Ensure sufficient quantity
       - Include edge cases
       - Verify example quality
    
    2. PATTERN IDENTIFICATION:
       - Look for commonalities
       - Consider multiple levels of abstraction
       - Identify exceptions
       - Test alternative patterns
    
    3. VALIDATION:
       - Check against all examples
       - Count supporting vs contradicting evidence
       - Calculate confidence scores
       - Seek counter-examples
    
    4. GENERALIZATION:
       - Move from specific to general
       - Define scope of applicability
       - Acknowledge limitations
       - Avoid over-generalization
    
    5. CONFIDENCE ASSESSMENT:
       - Based on supporting evidence ratio
       - Consider sample size
       - Account for example diversity
       - Update with new evidence
    
    6. APPLICATIONS:
       - Pattern discovery in data
       - Rule learning from examples
       - Trend analysis
       - Predictive modeling
       - Scientific generalization
    
    Benefits:
    ✓ Discovers patterns from data
    ✓ Learns from examples
    ✓ Enables predictions
    ✓ Practical and applicable
    ✓ Foundation for machine learning
    
    Limitations:
    - Conclusions not guaranteed
    - Sensitive to sample bias
    - May over-generalize
    - Requires sufficient examples
    - Exception handling needed
        """)
    
    
    if __name__ == "__main__":
        demonstrate_inductive_reasoning()
    
        

    Agentic Design Pattern: Inductive Reasoning

    Pattern 160: Inductive Reasoning – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Inductive Reasoning pattern implementation.


    Pattern Description

    Pattern 160: Inductive Reasoning

    Description: Inductive reasoning generalizes from specific observations to broader patterns or rules. It moves from particular instances to general conclusions, inferring patterns that likely hold across similar cases.

    Components: – Example collection – Pattern identification – Rule generalization – Confidence assessment – Counter-example checking

    Use Cases: – Pattern discovery – Rule learning – Trend analysis – Scientific generalization – Predictive modeling

    Benefits: – Discovers patterns from data – Enables prediction – Learns from examples – Practical applicability

    Trade-offs: – Conclusions not guaranteed – Vulnerable to biased samples – Requires sufficient examples – May over-generalize

    LangChain Implementation: Uses ChatOpenAI for pattern recognition, rule extraction, and validation through additional examples


    File Statistics

    • Total Lines: 558
    • Classes: 4
    • Functions: 1
    • Constants: 0

    Module Documentation

    """
    Pattern 160: Inductive Reasoning
    
    Description:
        Inductive reasoning generalizes from specific observations to broader patterns
        or rules. It moves from particular instances to general conclusions, inferring
        patterns that likely hold across similar cases.
    
    Components:
        - Example collection
        - Pattern identification
        - Rule generalization
        - Confidence assessment
        - Counter-example checking
    
    Use Cases:
        - Pattern discovery
        - Rule learning
        - Trend analysis
        - Scientific generalization
        - Predictive modeling
    
    Benefits:
        - Discovers patterns from data
        - Enables prediction
        - Learns from examples
        - Practical applicability
    
    Trade-offs:
        - Conclusions not guaranteed
        - Vulnerable to biased samples
        - Requires sufficient examples
        - May over-generalize
    
    LangChain Implementation:
        Uses ChatOpenAI for pattern recognition, rule extraction,
        and validation through additional examples
    """
    

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


    Import Statements

    Standard Library Imports

    import dataclasses.dataclass
    import dataclasses.field
    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
    

    Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – 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)

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

    Lines 54-59

    Purpose:

    Single example/observation
    

    Class: Pattern

    Lines 63-70

    Purpose:

    Discovered pattern
    

    Class: InductiveAnalysis

    Lines 74-80

    Purpose:

    Result of inductive reasoning
    

    Class: InductiveReasoningAgent

    Lines 83-342

    Purpose:

    Agent that performs inductive reasoning
    

    Methods:

    __init__(self, model_name)

    Lines 86-94

    Purpose:

    Initialize inductive reasoning agent
    
    Args:
        model_name: LLM model to use
    

    Parameters: – model_name: Language model instance

    reason(self, examples, domain, min_confidence)

    Lines 96-142

    Purpose:

    Perform inductive reasoning on examples
    
    Args:
        examples: List of examples to learn from
        domain: Optional domain context
        min_confidence: Minimum confidence threshold
    
    Returns:
        InductiveAnalysis with discovered patterns
    

    Parameters: – examples: Examples – domain: Domain – min_confidence: Min confidence

    Returns: InductiveAnalysis

    _identify_patterns(self, examples, domain)

    Lines 144-180

    Purpose:

    Identify patterns from examples
    

    Parameters: – examples: Examples – domain: Domain

    Returns: <ast.Subscript object at 0x000001A06D684350>

    _parse_patterns(self, response)

    Lines 182-233

    Purpose:

    Parse patterns from LLM response
    

    Parameters: – response: Response

    Returns: <ast.Subscript object at 0x000001A06D615FD0>

    _validate_pattern(self, pattern, examples)

    Lines 235-273

    Purpose:

    Validate pattern against examples
    

    Parameters: – pattern: Pattern – examples: Examples

    _generate_generalizations(self, patterns, examples)

    Lines 275-309

    Purpose:

    Generate general rules from patterns
    

    Parameters: – patterns: Patterns – examples: Examples

    Returns: <ast.Subscript object at 0x000001A06D7CC990>

    predict(self, analysis, new_input)

    Lines 311-342

    Purpose:

    Use learned patterns to make a prediction
    

    Parameters: – analysis: Analysis – new_input: Input data

    Returns: <ast.Subscript object at 0x000001A06D8550D0>


    Functions

    demonstrate_inductive_reasoning()

    Lines 345-553

    Purpose:

    Demonstrate inductive reasoning pattern
    

    Key Concepts Summary

    Pattern Overview

    The Inductive Reasoning 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 159
    • Pattern 161

    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 Inductive Reasoning 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: 160_inductive_reasoning.py
    Total Lines: 558
    Pattern Category: AI Agent Design Pattern #160


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Abductive Reasoning

    
        """
    Pattern 159: Abductive Reasoning
    
    Description:
        Abductive reasoning infers the best explanation for a set of observations.
        Unlike deductive reasoning (which derives consequences from premises) or
        inductive reasoning (which generalizes from examples), abduction works
        backwards from observations to find the most likely cause or explanation.
    
    Components:
        - Observation collection
        - Hypothesis generation
        - Explanation evaluation
        - Likelihood ranking
        - Plausibility assessment
    
    Use Cases:
        - Diagnostic reasoning (medical, technical)
        - Root cause analysis
        - Mystery solving
        - Troubleshooting
        - Scientific hypothesis formation
    
    Benefits:
        - Finds likely explanations
        - Handles incomplete information
        - Supports decision-making
        - Enables inference to best explanation
    
    Trade-offs:
        - Conclusions not guaranteed
        - Multiple possible explanations
        - Requires domain knowledge
        - Computationally intensive
    
    LangChain Implementation:
        Uses ChatOpenAI for hypothesis generation, evaluation scoring,
        and ranking of explanations by plausibility
    """
    
    import os
    from typing import List, Dict, Any, Optional
    from dataclasses import dataclass, field
    from enum import Enum
    from dotenv import load_dotenv
    
    from langchain_openai import ChatOpenAI
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser
    import json
    
    load_dotenv()
    
    
    class ConfidenceLevel(Enum):
        """Confidence levels for hypotheses"""
        VERY_HIGH = "very_high"
        HIGH = "high"
        MEDIUM = "medium"
        LOW = "low"
        VERY_LOW = "very_low"
    
    
    @dataclass
    class Observation:
        """Single observation or evidence"""
        description: str
        importance: float = 1.0  # 0-1 scale
        timestamp: Optional[str] = None
        source: Optional[str] = None
    
    
    @dataclass
    class Hypothesis:
        """Potential explanation"""
        explanation: str
        confidence: float  # 0-1 scale
        supporting_evidence: List[str] = field(default_factory=list)
        contradicting_evidence: List[str] = field(default_factory=list)
        assumptions: List[str] = field(default_factory=list)
        plausibility_score: float = 0.0
    
    
    @dataclass
    class AbductiveAnalysis:
        """Result of abductive reasoning"""
        observations: List[Observation]
        hypotheses: List[Hypothesis]
        best_explanation: Optional[Hypothesis]
        reasoning_trace: List[str] = field(default_factory=list)
    
    
    class AbductiveReasoningAgent:
        """Agent that performs abductive reasoning"""
        
        def __init__(self, model_name: str = "gpt-3.5-turbo"):
            """
            Initialize abductive reasoning agent
            
            Args:
                model_name: LLM model to use
            """
            self.llm = ChatOpenAI(model=model_name, temperature=0.7)
            self.evaluator_llm = ChatOpenAI(model=model_name, temperature=0)
        
        def reason(self, observations: List[Observation], 
                   domain_context: Optional[str] = None,
                   num_hypotheses: int = 5) -> AbductiveAnalysis:
            """
            Perform abductive reasoning on observations
            
            Args:
                observations: List of observations
                domain_context: Optional domain-specific context
                num_hypotheses: Number of hypotheses to generate
                
            Returns:
                AbductiveAnalysis with ranked hypotheses
            """
            reasoning_trace = []
            
            # Step 1: Generate hypotheses
            reasoning_trace.append("Generating hypotheses from observations...")
            hypotheses = self._generate_hypotheses(observations, domain_context, num_hypotheses)
            reasoning_trace.append(f"Generated {len(hypotheses)} hypotheses")
            
            # Step 2: Evaluate each hypothesis
            reasoning_trace.append("Evaluating hypotheses...")
            for hypothesis in hypotheses:
                self._evaluate_hypothesis(hypothesis, observations)
            
            # Step 3: Rank hypotheses
            hypotheses.sort(key=lambda h: h.plausibility_score, reverse=True)
            reasoning_trace.append("Ranked hypotheses by plausibility")
            
            # Step 4: Select best explanation
            best_explanation = hypotheses[0] if hypotheses else None
            
            return AbductiveAnalysis(
                observations=observations,
                hypotheses=hypotheses,
                best_explanation=best_explanation,
                reasoning_trace=reasoning_trace
            )
        
        def _generate_hypotheses(self, observations: List[Observation],
                                domain_context: Optional[str],
                                num_hypotheses: int) -> List[Hypothesis]:
            """Generate potential hypotheses"""
            # Format observations
            obs_text = "\n".join([
                f"- {obs.description} (importance: {obs.importance:.1f})"
                for obs in observations
            ])
            
            # Create prompt for hypothesis generation
            prompt = ChatPromptTemplate.from_messages([
                ("system", """You are an expert at abductive reasoning - inferring the best explanation for observations.
    Generate {num_hypotheses} plausible hypotheses that could explain the given observations.
    For each hypothesis, provide:
    1. Clear explanation
    2. Supporting reasoning
    3. Key assumptions
    
    {domain_context}
    
    Return as JSON array: [{{"explanation": "...", "reasoning": "...", "assumptions": ["..."]}}]"""),
                ("user", "Observations:\n{observations}\n\nGenerate {num_hypotheses} hypotheses:")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            
            context = f"Domain context: {domain_context}" if domain_context else "Consider general knowledge."
            
            response = chain.invoke({
                "observations": obs_text,
                "num_hypotheses": num_hypotheses,
                "domain_context": context
            })
            
            # Parse response
            hypotheses = self._parse_hypotheses(response)
            return hypotheses
        
        def _parse_hypotheses(self, response: str) -> List[Hypothesis]:
            """Parse hypotheses from LLM response"""
            try:
                # Try to extract JSON
                start = response.find('[')
                end = response.rfind(']') + 1
                if start >= 0 and end > start:
                    json_str = response[start:end]
                    data = json.loads(json_str)
                    
                    hypotheses = []
                    for item in data:
                        hypothesis = Hypothesis(
                            explanation=item.get('explanation', ''),
                            confidence=0.5,  # Initial neutral confidence
                            assumptions=item.get('assumptions', [])
                        )
                        hypotheses.append(hypothesis)
                    return hypotheses
            except Exception as e:
                print(f"Error parsing hypotheses: {e}")
            
            # Fallback: split by numbered items
            lines = response.split('\n')
            hypotheses = []
            current_explanation = []
            
            for line in lines:
                line = line.strip()
                if line and (line[0].isdigit() or line.startswith('-')):
                    if current_explanation:
                        explanation_text = ' '.join(current_explanation)
                        if explanation_text:
                            hypotheses.append(Hypothesis(
                                explanation=explanation_text,
                                confidence=0.5
                            ))
                        current_explanation = []
                    # Remove numbering/bullet
                    clean_line = line.lstrip('0123456789.-) ')
                    current_explanation.append(clean_line)
                elif current_explanation:
                    current_explanation.append(line)
            
            # Add last hypothesis
            if current_explanation:
                explanation_text = ' '.join(current_explanation)
                if explanation_text:
                    hypotheses.append(Hypothesis(
                        explanation=explanation_text,
                        confidence=0.5
                    ))
            
            return hypotheses[:5]  # Limit to 5
        
        def _evaluate_hypothesis(self, hypothesis: Hypothesis, 
                                observations: List[Observation]):
            """Evaluate a hypothesis against observations"""
            # Format observations
            obs_text = "\n".join([obs.description for obs in observations])
            
            # Create evaluation prompt
            prompt = ChatPromptTemplate.from_messages([
                ("system", """Evaluate how well this hypothesis explains the observations.
    Provide:
    1. Supporting evidence (which observations support it)
    2. Contradicting evidence (which observations contradict it)
    3. Confidence score (0-1)
    4. Plausibility assessment
    
    Return as JSON: {{
        "supporting": ["..."],
        "contradicting": ["..."],
        "confidence": 0.0-1.0,
        "plausibility": 0.0-1.0,
        "reasoning": "..."
    }}"""),
                ("user", """Hypothesis: {hypothesis}
    
    Observations:
    {observations}
    
    Evaluate:""")
            ])
            
            chain = prompt | self.evaluator_llm | StrOutputParser()
            
            response = chain.invoke({
                "hypothesis": hypothesis.explanation,
                "observations": obs_text
            })
            
            # Parse evaluation
            try:
                start = response.find('{')
                end = response.rfind('}') + 1
                if start >= 0 and end > start:
                    json_str = response[start:end]
                    data = json.loads(json_str)
                    
                    hypothesis.supporting_evidence = data.get('supporting', [])
                    hypothesis.contradicting_evidence = data.get('contradicting', [])
                    hypothesis.confidence = float(data.get('confidence', 0.5))
                    hypothesis.plausibility_score = float(data.get('plausibility', 0.5))
            except Exception as e:
                print(f"Error parsing evaluation: {e}")
                # Use default values
                hypothesis.confidence = 0.5
                hypothesis.plausibility_score = 0.5
        
        def explain_reasoning(self, analysis: AbductiveAnalysis) -> str:
            """Generate natural language explanation of reasoning"""
            prompt = ChatPromptTemplate.from_messages([
                ("system", "Explain the abductive reasoning process in clear, natural language."),
                ("user", """Observations:
    {observations}
    
    Best Explanation: {best_explanation}
    Confidence: {confidence:.2%}
    
    Supporting Evidence:
    {supporting}
    
    Explain why this is the best explanation:""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            
            obs_text = "\n".join([f"- {obs.description}" for obs in analysis.observations])
            supporting_text = "\n".join([f"- {s}" for s in analysis.best_explanation.supporting_evidence])
            
            explanation = chain.invoke({
                "observations": obs_text,
                "best_explanation": analysis.best_explanation.explanation,
                "confidence": analysis.best_explanation.confidence,
                "supporting": supporting_text
            })
            
            return explanation
    
    
    def demonstrate_abductive_reasoning():
        """Demonstrate abductive reasoning pattern"""
        print("=" * 80)
        print("ABDUCTIVE REASONING PATTERN DEMONSTRATION")
        print("=" * 80)
        
        agent = AbductiveReasoningAgent()
        
        # Example 1: Medical diagnosis
        print("\n" + "="*80)
        print("EXAMPLE 1: Medical Diagnosis")
        print("="*80)
        observations = [
            Observation("Patient has a fever of 101°F", importance=0.9),
            Observation("Patient reports sore throat", importance=0.8),
            Observation("Patient has swollen lymph nodes", importance=0.7),
            Observation("Patient has been in contact with sick individuals", importance=0.6)
        ]
        
        analysis = agent.reason(
            observations,
            domain_context="Medical diagnosis",
            num_hypotheses=4
        )
        
        print("Observations:")
        for obs in observations:
            print(f"  - {obs.description}")
        
        print(f"\nGenerated {len(analysis.hypotheses)} hypotheses:")
        for i, hyp in enumerate(analysis.hypotheses, 1):
            print(f"\n{i}. {hyp.explanation}")
            print(f"   Confidence: {hyp.confidence:.2%}")
            print(f"   Plausibility: {hyp.plausibility_score:.2%}")
        
        print(f"\nBest Explanation:")
        print(f"  {analysis.best_explanation.explanation}")
        print(f"  Confidence: {analysis.best_explanation.confidence:.2%}")
        
        # Example 2: Technical troubleshooting
        print("\n" + "="*80)
        print("EXAMPLE 2: Technical Troubleshooting")
        print("="*80)
        observations = [
            Observation("Website is loading slowly", importance=1.0),
            Observation("Database query times are normal", importance=0.8),
            Observation("CPU usage is at 95%", importance=0.9),
            Observation("Recent code deployment yesterday", importance=0.7),
            Observation("Memory usage is normal", importance=0.6)
        ]
        
        analysis = agent.reason(
            observations,
            domain_context="Software system performance troubleshooting",
            num_hypotheses=3
        )
        
        print("Observations:")
        for obs in observations:
            print(f"  - {obs.description}")
        
        print(f"\nTop 3 Hypotheses:")
        for i, hyp in enumerate(analysis.hypotheses[:3], 1):
            print(f"\n{i}. {hyp.explanation}")
            print(f"   Plausibility: {hyp.plausibility_score:.2%}")
            if hyp.supporting_evidence:
                print(f"   Supporting: {', '.join(hyp.supporting_evidence[:2])}")
        
        # Example 3: Mystery scenario
        print("\n" + "="*80)
        print("EXAMPLE 3: Mystery Investigation")
        print("="*80)
        observations = [
            Observation("Window is broken from the inside", importance=0.9),
            Observation("Valuable items are missing", importance=1.0),
            Observation("No signs of forced entry", importance=0.8),
            Observation("Alarm system was deactivated", importance=0.9),
            Observation("Security code was used correctly", importance=0.95)
        ]
        
        analysis = agent.reason(
            observations,
            domain_context="Criminal investigation",
            num_hypotheses=4
        )
        
        print("Observations:")
        for obs in observations:
            print(f"  - {obs.description}")
        
        print(f"\nBest Explanation:")
        best = analysis.best_explanation
        print(f"  {best.explanation}")
        print(f"  Confidence: {best.confidence:.2%}")
        print(f"  Plausibility: {best.plausibility_score:.2%}")
        
        # Generate natural language explanation
        explanation = agent.explain_reasoning(analysis)
        print(f"\nReasoning:")
        print(f"  {explanation[:300]}...")
        
        # Example 4: Business analysis
        print("\n" + "="*80)
        print("EXAMPLE 4: Business Analysis")
        print("="*80)
        observations = [
            Observation("Sales dropped 30% this month", importance=1.0),
            Observation("Competitor launched new product", importance=0.8),
            Observation("Customer complaints increased", importance=0.7),
            Observation("Website traffic is unchanged", importance=0.6),
            Observation("Pricing remained the same", importance=0.5)
        ]
        
        analysis = agent.reason(
            observations,
            domain_context="Business performance analysis",
            num_hypotheses=3
        )
        
        print("Observations:")
        for obs in observations:
            print(f"  - {obs.description}")
        
        print(f"\nHypotheses (ranked by plausibility):")
        for i, hyp in enumerate(analysis.hypotheses, 1):
            print(f"\n{i}. {hyp.explanation}")
            print(f"   Plausibility: {hyp.plausibility_score:.2%}")
        
        # Example 5: Scientific reasoning
        print("\n" + "="*80)
        print("EXAMPLE 5: Scientific Observation")
        print("="*80)
        observations = [
            Observation("Plant leaves are turning yellow", importance=0.9),
            Observation("Soil is consistently wet", importance=0.8),
            Observation("Plant is in low light area", importance=0.7),
            Observation("No visible pests or diseases", importance=0.6)
        ]
        
        analysis = agent.reason(
            observations,
            domain_context="Plant biology and care",
            num_hypotheses=4
        )
        
        print("Observations:")
        for obs in observations:
            print(f"  - {obs.description}")
        
        print(f"\nTop Hypothesis:")
        best = analysis.hypotheses[0]
        print(f"  Explanation: {best.explanation}")
        print(f"  Confidence: {best.confidence:.2%}")
        if best.assumptions:
            print(f"  Assumptions: {', '.join(best.assumptions)}")
        
        # Summary
        print("\n" + "="*80)
        print("SUMMARY: Abductive Reasoning Best Practices")
        print("="*80)
        print("""
    1. OBSERVATION COLLECTION:
       - Gather all available evidence
       - Note importance/relevance of each observation
       - Include timing and context
       - Consider source reliability
    
    2. HYPOTHESIS GENERATION:
       - Generate multiple plausible explanations
       - Consider diverse possibilities
       - Don't prematurely eliminate options
       - Use domain knowledge
    
    3. EVALUATION CRITERIA:
       - Explanatory power (how well it explains observations)
       - Simplicity (Occam's Razor)
       - Consistency with known facts
       - Plausibility given context
    
    4. RANKING FACTORS:
       - Supporting evidence count and quality
       - Contradicting evidence
       - Prior probability
       - Completeness of explanation
    
    5. REASONING PROCESS:
       - Work backwards from observations
       - Consider alternative explanations
       - Evaluate each hypothesis rigorously
       - Update beliefs with new evidence
    
    6. APPLICATIONS:
       - Diagnostic reasoning
       - Root cause analysis
       - Mystery solving
       - Scientific hypothesis formation
       - Decision support systems
    
    Benefits:
    ✓ Finds most likely explanations
    ✓ Handles incomplete information
    ✓ Supports diagnostic tasks
    ✓ Enables inference to best explanation
    ✓ Practical for real-world problems
    
    Limitations:
    - Conclusions not guaranteed correct
    - Multiple explanations may be equally plausible
    - Requires domain knowledge
    - Sensitive to observation quality
    - May miss novel explanations
        """)
    
    
    if __name__ == "__main__":
        demonstrate_abductive_reasoning()
    
        

    Agentic Design Pattern: Abductive Reasoning

    Pattern 159: Abductive Reasoning – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Abductive Reasoning pattern implementation.


    Pattern Description

    Pattern 159: Abductive Reasoning

    Description: Abductive reasoning infers the best explanation for a set of observations. Unlike deductive reasoning (which derives consequences from premises) or inductive reasoning (which generalizes from examples), abduction works backwards from observations to find the most likely cause or explanation.

    Components: – Observation collection – Hypothesis generation – Explanation evaluation – Likelihood ranking – Plausibility assessment

    Use Cases: – Diagnostic reasoning (medical, technical) – Root cause analysis – Mystery solving – Troubleshooting – Scientific hypothesis formation

    Benefits: – Finds likely explanations – Handles incomplete information – Supports decision-making – Enables inference to best explanation

    Trade-offs: – Conclusions not guaranteed – Multiple possible explanations – Requires domain knowledge – Computationally intensive

    LangChain Implementation: Uses ChatOpenAI for hypothesis generation, evaluation scoring, and ranking of explanations by plausibility


    File Statistics

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

    Module Documentation

    """
    Pattern 159: Abductive Reasoning
    
    Description:
        Abductive reasoning infers the best explanation for a set of observations.
        Unlike deductive reasoning (which derives consequences from premises) or
        inductive reasoning (which generalizes from examples), abduction works
        backwards from observations to find the most likely cause or explanation.
    
    Components:
        - Observation collection
        - Hypothesis generation
        - Explanation evaluation
        - Likelihood ranking
        - Plausibility assessment
    
    Use Cases:
        - Diagnostic reasoning (medical, technical)
        - Root cause analysis
        - Mystery solving
        - Troubleshooting
        - Scientific hypothesis formation
    
    Benefits:
        - Finds likely explanations
        - Handles incomplete information
        - Supports decision-making
        - Enables inference to best explanation
    
    Trade-offs:
        - Conclusions not guaranteed
        - Multiple possible explanations
        - Requires domain knowledge
        - Computationally intensive
    
    LangChain Implementation:
        Uses ChatOpenAI for hypothesis generation, evaluation scoring,
        and ranking of explanations by plausibility
    """
    

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


    Import Statements

    Standard Library Imports

    import dataclasses.dataclass
    import dataclasses.field
    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
    

    Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – 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)

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

    Lines 55-61

    Inherits from: Enum

    Purpose:

    Confidence levels for hypotheses
    

    Class: Observation

    Lines 65-70

    Purpose:

    Single observation or evidence
    

    Class: Hypothesis

    Lines 74-81

    Purpose:

    Potential explanation
    

    Class: AbductiveAnalysis

    Lines 85-90

    Purpose:

    Result of abductive reasoning
    

    Class: AbductiveReasoningAgent

    Lines 93-323

    Purpose:

    Agent that performs abductive reasoning
    

    Methods:

    __init__(self, model_name)

    Lines 96-104

    Purpose:

    Initialize abductive reasoning agent
    
    Args:
        model_name: LLM model to use
    

    Parameters: – model_name: Language model instance

    reason(self, observations, domain_context, num_hypotheses)

    Lines 106-144

    Purpose:

    Perform abductive reasoning on observations
    
    Args:
        observations: List of observations
        domain_context: Optional domain-specific context
        num_hypotheses: Number of hypotheses to generate
    
    Returns:
        AbductiveAnalysis with ranked hypotheses
    

    Parameters: – observations: Observations – domain_context: Domain context – num_hypotheses: Num hypotheses

    Returns: AbductiveAnalysis

    _generate_hypotheses(self, observations, domain_context, num_hypotheses)

    Lines 146-183

    Purpose:

    Generate potential hypotheses
    

    Parameters: – observations: Observations – domain_context: Domain context – num_hypotheses: Num hypotheses

    Returns: <ast.Subscript object at 0x000001A06E0AB450>

    _parse_hypotheses(self, response)

    Lines 185-238

    Purpose:

    Parse hypotheses from LLM response
    

    Parameters: – response: Response

    Returns: <ast.Subscript object at 0x000001A06D6304D0>

    _evaluate_hypothesis(self, hypothesis, observations)

    Lines 240-293

    Purpose:

    Evaluate a hypothesis against observations
    

    Parameters: – hypothesis: Hypothesis – observations: Observations

    explain_reasoning(self, analysis)

    Lines 295-323

    Purpose:

    Generate natural language explanation of reasoning
    

    Parameters: – analysis: Analysis

    Returns: str


    Functions

    demonstrate_abductive_reasoning()

    Lines 326-537

    Purpose:

    Demonstrate abductive reasoning pattern
    

    Key Concepts Summary

    Pattern Overview

    The Abductive Reasoning 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 158
    • Pattern 160

    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 Abductive Reasoning 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: 159_abductive_reasoning.py
    Total Lines: 542
    Pattern Category: AI Agent Design Pattern #159


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Plugin Extension

    
        """
    Pattern 158: Plugin/Extension Architecture
    
    Description:
        Enables extensibility through a plugin system where new capabilities can be
        added dynamically without modifying core code. Plugins follow a standard
        interface and can be loaded, configured, and executed at runtime.
    
    Components:
        - Plugin interface/protocol
        - Plugin registry
        - Plugin loader
        - Lifecycle management
        - Dependency injection
        - Configuration system
    
    Use Cases:
        - Custom tool integration
        - Domain-specific capabilities
        - Third-party extensions
        - Modular agent systems
        - Dynamic feature addition
    
    Benefits:
        - Extensibility without core changes
        - Modular architecture
        - Third-party contributions
        - Dynamic capability loading
        - Separation of concerns
    
    Trade-offs:
        - Complexity overhead
        - Plugin management burden
        - Potential security risks
        - Version compatibility issues
    
    LangChain Implementation:
        Uses protocol classes for interfaces, dynamic loading,
        registry pattern, and lifecycle hooks
    """
    
    import os
    import importlib
    import inspect
    from typing import Dict, Any, List, Optional, Callable, Protocol
    from dataclasses import dataclass, field
    from enum import Enum
    from abc import ABC, abstractmethod
    from dotenv import load_dotenv
    
    from langchain_openai import ChatOpenAI
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser
    
    load_dotenv()
    
    
    class PluginStatus(Enum):
        """Plugin status"""
        UNLOADED = "unloaded"
        LOADED = "loaded"
        INITIALIZED = "initialized"
        ACTIVE = "active"
        DISABLED = "disabled"
        ERROR = "error"
    
    
    class PluginType(Enum):
        """Plugin types"""
        TOOL = "tool"
        PROCESSOR = "processor"
        FILTER = "filter"
        ANALYZER = "analyzer"
        GENERATOR = "generator"
    
    
    @dataclass
    class PluginMetadata:
        """Plugin metadata"""
        name: str
        version: str
        author: str
        description: str
        plugin_type: PluginType
        dependencies: List[str] = field(default_factory=list)
        config_schema: Dict[str, Any] = field(default_factory=dict)
    
    
    @dataclass
    class PluginConfig:
        """Plugin configuration"""
        enabled: bool = True
        priority: int = 0
        settings: Dict[str, Any] = field(default_factory=dict)
    
    
    class Plugin(Protocol):
        """Plugin interface using Protocol (structural typing)"""
        
        @property
        def metadata(self) -> PluginMetadata:
            """Get plugin metadata"""
            ...
        
        def initialize(self, config: PluginConfig) -> bool:
            """Initialize plugin"""
            ...
        
        def execute(self, input_data: Any, context: Dict[str, Any]) -> Any:
            """Execute plugin functionality"""
            ...
        
        def shutdown(self) -> bool:
            """Cleanup and shutdown"""
            ...
    
    
    class BasePlugin(ABC):
        """Base class for plugins"""
        
        def __init__(self):
            """Initialize base plugin"""
            self.config: Optional[PluginConfig] = None
            self.status = PluginStatus.UNLOADED
            self._metadata: Optional[PluginMetadata] = None
        
        @property
        def metadata(self) -> PluginMetadata:
            """Get plugin metadata"""
            if self._metadata is None:
                self._metadata = self._create_metadata()
            return self._metadata
        
        @abstractmethod
        def _create_metadata(self) -> PluginMetadata:
            """Create plugin metadata"""
            pass
        
        def initialize(self, config: PluginConfig) -> bool:
            """Initialize plugin"""
            try:
                self.config = config
                self._on_initialize()
                self.status = PluginStatus.INITIALIZED
                if config.enabled:
                    self.status = PluginStatus.ACTIVE
                return True
            except Exception as e:
                self.status = PluginStatus.ERROR
                print(f"Plugin initialization failed: {e}")
                return False
        
        def _on_initialize(self):
            """Override for custom initialization"""
            pass
        
        @abstractmethod
        def execute(self, input_data: Any, context: Dict[str, Any]) -> Any:
            """Execute plugin functionality"""
            pass
        
        def shutdown(self) -> bool:
            """Cleanup and shutdown"""
            try:
                self._on_shutdown()
                self.status = PluginStatus.UNLOADED
                return True
            except Exception as e:
                print(f"Plugin shutdown failed: {e}")
                return False
        
        def _on_shutdown(self):
            """Override for custom shutdown"""
            pass
    
    
    # Example Plugin Implementations
    
    class SentimentAnalysisPlugin(BasePlugin):
        """Plugin for sentiment analysis"""
        
        def _create_metadata(self) -> PluginMetadata:
            return PluginMetadata(
                name="sentiment_analyzer",
                version="1.0.0",
                author="Example Corp",
                description="Analyzes sentiment of text using LLM",
                plugin_type=PluginType.ANALYZER
            )
        
        def _on_initialize(self):
            """Initialize LLM"""
            self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
        
        def execute(self, input_data: Any, context: Dict[str, Any]) -> Any:
            """Analyze sentiment"""
            text = input_data if isinstance(input_data, str) else str(input_data)
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", "Analyze the sentiment of the given text. Respond with: positive, negative, or neutral."),
                ("user", "{text}")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            sentiment = chain.invoke({"text": text})
            
            return {
                "sentiment": sentiment.strip().lower(),
                "text": text,
                "plugin": self.metadata.name
            }
    
    
    class TextSummarizerPlugin(BasePlugin):
        """Plugin for text summarization"""
        
        def _create_metadata(self) -> PluginMetadata:
            return PluginMetadata(
                name="text_summarizer",
                version="1.0.0",
                author="Example Corp",
                description="Summarizes text content",
                plugin_type=PluginType.PROCESSOR
            )
        
        def _on_initialize(self):
            """Initialize LLM"""
            max_words = self.config.settings.get("max_words", 50) if self.config else 50
            self.max_words = max_words
            self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.3)
        
        def execute(self, input_data: Any, context: Dict[str, Any]) -> Any:
            """Summarize text"""
            text = input_data if isinstance(input_data, str) else str(input_data)
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", "Summarize the following text in {max_words} words or less."),
                ("user", "{text}")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            summary = chain.invoke({"text": text, "max_words": self.max_words})
            
            return {
                "summary": summary,
                "original_length": len(text.split()),
                "summary_length": len(summary.split()),
                "plugin": self.metadata.name
            }
    
    
    class KeywordExtractorPlugin(BasePlugin):
        """Plugin for keyword extraction"""
        
        def _create_metadata(self) -> PluginMetadata:
            return PluginMetadata(
                name="keyword_extractor",
                version="1.0.0",
                author="Example Corp",
                description="Extracts keywords from text",
                plugin_type=PluginType.ANALYZER
            )
        
        def _on_initialize(self):
            """Initialize LLM"""
            self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
        
        def execute(self, input_data: Any, context: Dict[str, Any]) -> Any:
            """Extract keywords"""
            text = input_data if isinstance(input_data, str) else str(input_data)
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", "Extract 5-7 key terms or phrases from the text. Return as comma-separated list."),
                ("user", "{text}")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            keywords_str = chain.invoke({"text": text})
            
            keywords = [k.strip() for k in keywords_str.split(",")]
            
            return {
                "keywords": keywords,
                "count": len(keywords),
                "plugin": self.metadata.name
            }
    
    
    class ProfanityFilterPlugin(BasePlugin):
        """Plugin for filtering profanity"""
        
        def _create_metadata(self) -> PluginMetadata:
            return PluginMetadata(
                name="profanity_filter",
                version="1.0.0",
                author="Example Corp",
                description="Filters inappropriate content",
                plugin_type=PluginType.FILTER
            )
        
        def _on_initialize(self):
            """Initialize filter"""
            self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
        
        def execute(self, input_data: Any, context: Dict[str, Any]) -> Any:
            """Filter content"""
            text = input_data if isinstance(input_data, str) else str(input_data)
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", "Analyze if the text contains profanity or inappropriate content. Respond with 'yes' or 'no'."),
                ("user", "{text}")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            contains_profanity = chain.invoke({"text": text})
            
            is_inappropriate = "yes" in contains_profanity.lower()
            
            return {
                "is_appropriate": not is_inappropriate,
                "filtered": is_inappropriate,
                "text": text if not is_inappropriate else "[CONTENT FILTERED]",
                "plugin": self.metadata.name
            }
    
    
    class PluginRegistry:
        """Manages plugin registration and lifecycle"""
        
        def __init__(self):
            """Initialize plugin registry"""
            self.plugins: Dict[str, BasePlugin] = {}
            self.configs: Dict[str, PluginConfig] = {}
            self.execution_order: List[str] = []
        
        def register_plugin(self, plugin: BasePlugin, config: Optional[PluginConfig] = None):
            """Register a plugin"""
            if config is None:
                config = PluginConfig()
            
            plugin_name = plugin.metadata.name
            
            # Check dependencies
            for dep in plugin.metadata.dependencies:
                if dep not in self.plugins:
                    print(f"Warning: Missing dependency '{dep}' for plugin '{plugin_name}'")
            
            # Register plugin
            self.plugins[plugin_name] = plugin
            self.configs[plugin_name] = config
            
            # Initialize plugin
            success = plugin.initialize(config)
            
            if success:
                # Add to execution order based on priority
                self._update_execution_order(plugin_name, config.priority)
                print(f"✓ Registered and initialized plugin: {plugin_name}")
            else:
                print(f"✗ Failed to initialize plugin: {plugin_name}")
            
            return success
        
        def _update_execution_order(self, plugin_name: str, priority: int):
            """Update execution order based on priority"""
            if plugin_name in self.execution_order:
                self.execution_order.remove(plugin_name)
            
            # Insert based on priority (higher priority first)
            inserted = False
            for i, name in enumerate(self.execution_order):
                if self.configs[name].priority  bool:
            """Unregister a plugin"""
            if plugin_name not in self.plugins:
                return False
            
            plugin = self.plugins[plugin_name]
            plugin.shutdown()
            
            del self.plugins[plugin_name]
            del self.configs[plugin_name]
            self.execution_order.remove(plugin_name)
            
            print(f"✓ Unregistered plugin: {plugin_name}")
            return True
        
        def get_plugin(self, plugin_name: str) -> Optional[BasePlugin]:
            """Get plugin by name"""
            return self.plugins.get(plugin_name)
        
        def execute_plugin(self, plugin_name: str, input_data: Any, 
                          context: Optional[Dict[str, Any]] = None) -> Any:
            """Execute specific plugin"""
            if context is None:
                context = {}
            
            plugin = self.plugins.get(plugin_name)
            if not plugin:
                raise ValueError(f"Plugin not found: {plugin_name}")
            
            if plugin.status != PluginStatus.ACTIVE:
                raise RuntimeError(f"Plugin not active: {plugin_name}")
            
            return plugin.execute(input_data, context)
        
        def execute_pipeline(self, input_data: Any, 
                            plugin_types: Optional[List[PluginType]] = None,
                            context: Optional[Dict[str, Any]] = None) -> Any:
            """Execute plugins in pipeline"""
            if context is None:
                context = {}
            
            result = input_data
            
            for plugin_name in self.execution_order:
                plugin = self.plugins[plugin_name]
                config = self.configs[plugin_name]
                
                # Skip if not enabled
                if not config.enabled or plugin.status != PluginStatus.ACTIVE:
                    continue
                
                # Filter by type if specified
                if plugin_types and plugin.metadata.plugin_type not in plugin_types:
                    continue
                
                # Execute plugin
                result = plugin.execute(result, context)
            
            return result
        
        def list_plugins(self) -> List[Dict[str, Any]]:
            """List all registered plugins"""
            return [
                {
                    "name": plugin.metadata.name,
                    "version": plugin.metadata.version,
                    "type": plugin.metadata.plugin_type.value,
                    "status": plugin.status.value,
                    "enabled": self.configs[name].enabled,
                    "priority": self.configs[name].priority
                }
                for name, plugin in self.plugins.items()
            ]
        
        def shutdown_all(self):
            """Shutdown all plugins"""
            for plugin_name in list(self.plugins.keys()):
                self.unregister_plugin(plugin_name)
    
    
    def demonstrate_plugin_architecture():
        """Demonstrate plugin/extension architecture"""
        print("=" * 80)
        print("PLUGIN/EXTENSION ARCHITECTURE DEMONSTRATION")
        print("=" * 80)
        
        # Initialize registry
        registry = PluginRegistry()
        
        # Example 1: Register plugins
        print("\n" + "="*80)
        print("EXAMPLE 1: Registering Plugins")
        print("="*80)
        
        # Register sentiment analyzer
        sentiment_plugin = SentimentAnalysisPlugin()
        registry.register_plugin(sentiment_plugin, PluginConfig(
            enabled=True,
            priority=10
        ))
        
        # Register summarizer
        summarizer_plugin = TextSummarizerPlugin()
        registry.register_plugin(summarizer_plugin, PluginConfig(
            enabled=True,
            priority=5,
            settings={"max_words": 30}
        ))
        
        # Register keyword extractor
        keyword_plugin = KeywordExtractorPlugin()
        registry.register_plugin(keyword_plugin, PluginConfig(
            enabled=True,
            priority=3
        ))
        
        # Register profanity filter
        filter_plugin = ProfanityFilterPlugin()
        registry.register_plugin(filter_plugin, PluginConfig(
            enabled=True,
            priority=20  # High priority - run first
        ))
        
        # Example 2: List plugins
        print("\n" + "="*80)
        print("EXAMPLE 2: List Registered Plugins")
        print("="*80)
        plugins = registry.list_plugins()
        for plugin in plugins:
            print(f"- {plugin['name']} v{plugin['version']}")
            print(f"  Type: {plugin['type']}, Status: {plugin['status']}")
            print(f"  Priority: {plugin['priority']}, Enabled: {plugin['enabled']}")
        
        # Example 3: Execute individual plugin
        print("\n" + "="*80)
        print("EXAMPLE 3: Execute Individual Plugin")
        print("="*80)
        text = "This product is amazing! I love it so much!"
        result = registry.execute_plugin("sentiment_analyzer", text)
        print(f"Input: {text}")
        print(f"Result: {result}")
        
        # Example 4: Execute multiple plugins
        print("\n" + "="*80)
        print("EXAMPLE 4: Execute Multiple Plugins")
        print("="*80)
        text = "Artificial intelligence is transforming how we work and live. Machine learning models can now process vast amounts of data and make predictions with high accuracy."
        
        print(f"Input: {text[:100]}...\n")
        
        # Sentiment
        sentiment_result = registry.execute_plugin("sentiment_analyzer", text)
        print(f"Sentiment: {sentiment_result['sentiment']}")
        
        # Summary
        summary_result = registry.execute_plugin("text_summarizer", text)
        print(f"Summary: {summary_result['summary']}")
        
        # Keywords
        keyword_result = registry.execute_plugin("keyword_extractor", text)
        print(f"Keywords: {', '.join(keyword_result['keywords'])}")
        
        # Example 5: Filter by plugin type
        print("\n" + "="*80)
        print("EXAMPLE 5: Execute by Plugin Type")
        print("="*80)
        text = "The company announced record profits this quarter."
        
        # Execute only analyzers
        print("Executing ANALYZER plugins:")
        result = registry.execute_pipeline(text, plugin_types=[PluginType.ANALYZER])
        print(f"Result: {result}")
        
        # Example 6: Content filtering
        print("\n" + "="*80)
        print("EXAMPLE 6: Content Filtering")
        print("="*80)
        clean_text = "This is a nice and appropriate message."
        result = registry.execute_plugin("profanity_filter", clean_text)
        print(f"Text: {clean_text}")
        print(f"Is Appropriate: {result['is_appropriate']}")
        print(f"Output: {result['text']}")
        
        # Example 7: Execution order (priority)
        print("\n" + "="*80)
        print("EXAMPLE 7: Plugin Execution Order")
        print("="*80)
        print("Execution order (by priority):")
        for i, plugin_name in enumerate(registry.execution_order, 1):
            plugin = registry.get_plugin(plugin_name)
            priority = registry.configs[plugin_name].priority
            print(f"{i}. {plugin_name} (priority: {priority})")
        
        # Cleanup
        print("\n" + "="*80)
        print("CLEANUP: Shutting Down Plugins")
        print("="*80)
        registry.shutdown_all()
        
        # Summary
        print("\n" + "="*80)
        print("SUMMARY: Plugin Architecture Best Practices")
        print("="*80)
        print("""
    1. PLUGIN INTERFACE:
       - Define clear plugin interface/protocol
       - Standardize lifecycle methods (initialize, execute, shutdown)
       - Use metadata for plugin description
       - Support configuration schemas
    
    2. REGISTRY MANAGEMENT:
       - Centralized plugin registration
       - Dependency checking
       - Priority-based execution
       - Enable/disable plugins dynamically
    
    3. LIFECYCLE MANAGEMENT:
       - Proper initialization
       - Resource cleanup on shutdown
       - Error handling and status tracking
       - Graceful failure handling
    
    4. EXECUTION PATTERNS:
       - Individual plugin execution
       - Pipeline execution
       - Type-based filtering
       - Context passing between plugins
    
    5. CONFIGURATION:
       - Plugin-specific settings
       - Priority control
       - Enable/disable flags
       - Runtime configuration updates
    
    6. EXTENSIBILITY:
       - Easy to add new plugins
       - No core code modification
       - Third-party plugin support
       - Hot-loading capabilities (advanced)
    
    Benefits:
    ✓ Modular and extensible architecture
    ✓ Separation of concerns
    ✓ Easy feature addition
    ✓ Third-party contributions
    ✓ Dynamic capability enhancement
    ✓ Clean core codebase
    
    Challenges:
    - Plugin management complexity
    - Version compatibility
    - Security considerations for third-party plugins
    - Performance overhead
    - Dependency management
    - Testing across plugin combinations
        """)
    
    
    if __name__ == "__main__":
        demonstrate_plugin_architecture()
    
        

    Agentic Design Pattern: Plugin Extension

    Pattern 158: Plugin Extension – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Plugin Extension pattern implementation.


    Pattern Description

    Pattern 158: Plugin/Extension Architecture

    Description: Enables extensibility through a plugin system where new capabilities can be added dynamically without modifying core code. Plugins follow a standard interface and can be loaded, configured, and executed at runtime.

    Components: – Plugin interface/protocol – Plugin registry – Plugin loader – Lifecycle management – Dependency injection – Configuration system

    Use Cases: – Custom tool integration – Domain-specific capabilities – Third-party extensions – Modular agent systems – Dynamic feature addition

    Benefits: – Extensibility without core changes – Modular architecture – Third-party contributions – Dynamic capability loading – Separation of concerns

    Trade-offs: – Complexity overhead – Plugin management burden – Potential security risks – Version compatibility issues

    LangChain Implementation: Uses protocol classes for interfaces, dynamic loading, registry pattern, and lifecycle hooks


    File Statistics

    • Total Lines: 640
    • Classes: 11
    • Functions: 1
    • Constants: 0

    Module Documentation

    """
    Pattern 158: Plugin/Extension Architecture
    
    Description:
        Enables extensibility through a plugin system where new capabilities can be
        added dynamically without modifying core code. Plugins follow a standard
        interface and can be loaded, configured, and executed at runtime.
    
    Components:
        - Plugin interface/protocol
        - Plugin registry
        - Plugin loader
        - Lifecycle management
        - Dependency injection
        - Configuration system
    
    Use Cases:
        - Custom tool integration
        - Domain-specific capabilities
        - Third-party extensions
        - Modular agent systems
        - Dynamic feature addition
    
    Benefits:
        - Extensibility without core changes
        - Modular architecture
        - Third-party contributions
        - Dynamic capability loading
        - Separation of concerns
    
    Trade-offs:
        - Complexity overhead
        - Plugin management burden
        - Potential security risks
        - Version compatibility issues
    
    LangChain Implementation:
        Uses protocol classes for interfaces, dynamic loading,
        registry pattern, and lifecycle hooks
    """
    

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


    Import Statements

    Standard Library Imports

    import dataclasses.dataclass
    import dataclasses.field
    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
    import typing.Protocol
    

    Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – 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) – typing.Protocol: 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 abc.ABC
    from abc.abstractmethod
    from dotenv.load_dotenv
    from importlib
    from inspect
    

    Purpose: Additional dependencies for specialized functionality.


    Classes

    Class: PluginStatus

    Lines 58-65

    Inherits from: Enum

    Purpose:

    Plugin status
    

    Class: PluginType

    Lines 68-74

    Inherits from: Enum

    Purpose:

    Plugin types
    

    Class: PluginMetadata

    Lines 78-86

    Purpose:

    Plugin metadata
    

    Class: PluginConfig

    Lines 90-94

    Purpose:

    Plugin configuration
    

    Class: Plugin

    Lines 97-115

    Inherits from: Protocol

    Purpose:

    Plugin interface using Protocol (structural typing)
    

    Methods:

    metadata(self)

    Lines 101-103

    Decorators: property

    Purpose:

    Get plugin metadata
    

    Parameters:

    Returns: PluginMetadata

    initialize(self, config)

    Lines 105-107

    Purpose:

    Initialize plugin
    

    Parameters: – config: Config

    Returns: bool

    execute(self, input_data, context)

    Lines 109-111

    Purpose:

    Execute plugin functionality
    

    Parameters: – input_data: Input data – context: Context

    Returns: Any

    shutdown(self)

    Lines 113-115

    Purpose:

    Cleanup and shutdown
    

    Parameters:

    Returns: bool


    Class: BasePlugin

    Lines 118-174

    Inherits from: ABC

    Purpose:

    Base class for plugins
    

    Methods:

    __init__(self)

    Lines 121-125

    Purpose:

    Initialize base plugin
    

    Parameters:

    metadata(self)

    Lines 128-132

    Decorators: property

    Purpose:

    Get plugin metadata
    

    Parameters:

    Returns: PluginMetadata

    _create_metadata(self)

    Lines 135-137

    Decorators: abstractmethod

    Purpose:

    Create plugin metadata
    

    Parameters:

    Returns: PluginMetadata

    initialize(self, config)

    Lines 139-151

    Purpose:

    Initialize plugin
    

    Parameters: – config: Config

    Returns: bool

    _on_initialize(self)

    Lines 153-155

    Purpose:

    Override for custom initialization
    

    Parameters:

    execute(self, input_data, context)

    Lines 158-160

    Decorators: abstractmethod

    Purpose:

    Execute plugin functionality
    

    Parameters: – input_data: Input data – context: Context

    Returns: Any

    shutdown(self)

    Lines 162-170

    Purpose:

    Cleanup and shutdown
    

    Parameters:

    Returns: bool

    _on_shutdown(self)

    Lines 172-174

    Purpose:

    Override for custom shutdown
    

    Parameters:


    Class: SentimentAnalysisPlugin

    Lines 179-211

    Inherits from: BasePlugin

    Purpose:

    Plugin for sentiment analysis
    

    Methods:

    _create_metadata(self)

    Lines 182-189

    Purpose: Create and return a new instance or configuration.

    Parameters:

    Returns: PluginMetadata

    _on_initialize(self)

    Lines 191-193

    Purpose:

    Initialize LLM
    

    Parameters:

    execute(self, input_data, context)

    Lines 195-211

    Purpose:

    Analyze sentiment
    

    Parameters: – input_data: Input data – context: Context

    Returns: Any


    Class: TextSummarizerPlugin

    Lines 214-249

    Inherits from: BasePlugin

    Purpose:

    Plugin for text summarization
    

    Methods:

    _create_metadata(self)

    Lines 217-224

    Purpose: Create and return a new instance or configuration.

    Parameters:

    Returns: PluginMetadata

    _on_initialize(self)

    Lines 226-230

    Purpose:

    Initialize LLM
    

    Parameters:

    execute(self, input_data, context)

    Lines 232-249

    Purpose:

    Summarize text
    

    Parameters: – input_data: Input data – context: Context

    Returns: Any


    Class: KeywordExtractorPlugin

    Lines 252-286

    Inherits from: BasePlugin

    Purpose:

    Plugin for keyword extraction
    

    Methods:

    _create_metadata(self)

    Lines 255-262

    Purpose: Create and return a new instance or configuration.

    Parameters:

    Returns: PluginMetadata

    _on_initialize(self)

    Lines 264-266

    Purpose:

    Initialize LLM
    

    Parameters:

    execute(self, input_data, context)

    Lines 268-286

    Purpose:

    Extract keywords
    

    Parameters: – input_data: Input data – context: Context

    Returns: Any


    Class: ProfanityFilterPlugin

    Lines 289-324

    Inherits from: BasePlugin

    Purpose:

    Plugin for filtering profanity
    

    Methods:

    _create_metadata(self)

    Lines 292-299

    Purpose: Create and return a new instance or configuration.

    Parameters:

    Returns: PluginMetadata

    _on_initialize(self)

    Lines 301-303

    Purpose:

    Initialize filter
    

    Parameters:

    execute(self, input_data, context)

    Lines 305-324

    Purpose:

    Filter content
    

    Parameters: – input_data: Input data – context: Context

    Returns: Any


    Class: PluginRegistry

    Lines 327-457

    Purpose:

    Manages plugin registration and lifecycle
    

    Methods:

    __init__(self)

    Lines 330-334

    Purpose:

    Initialize plugin registry
    

    Parameters:

    register_plugin(self, plugin, config)

    Lines 336-362

    Purpose:

    Register a plugin
    

    Parameters: – plugin: Plugin – config: Config

    _update_execution_order(self, plugin_name, priority)

    Lines 364-378

    Purpose:

    Update execution order based on priority
    

    Parameters: – plugin_name: Plugin name – priority: Priority

    unregister_plugin(self, plugin_name)

    Lines 380-393

    Purpose:

    Unregister a plugin
    

    Parameters: – plugin_name: Plugin name

    Returns: bool

    get_plugin(self, plugin_name)

    Lines 395-397

    Purpose:

    Get plugin by name
    

    Parameters: – plugin_name: Plugin name

    Returns: <ast.Subscript object at 0x000001A06D7CC850>

    execute_plugin(self, plugin_name, input_data, context)

    Lines 399-412

    Purpose:

    Execute specific plugin
    

    Parameters: – plugin_name: Plugin name – input_data: Input data – context: Context

    Returns: Any

    execute_pipeline(self, input_data, plugin_types, context)

    Lines 414-438

    Purpose:

    Execute plugins in pipeline
    

    Parameters: – input_data: Input data – plugin_types: Plugin types – context: Context

    Returns: Any

    list_plugins(self)

    Lines 440-452

    Purpose:

    List all registered plugins
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D6F6290>

    shutdown_all(self)

    Lines 454-457

    Purpose:

    Shutdown all plugins
    

    Parameters:


    Functions

    demonstrate_plugin_architecture()

    Lines 460-635

    Purpose:

    Demonstrate plugin/extension architecture
    

    Key Concepts Summary

    Pattern Overview

    The Plugin Extension 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 157
    • Pattern 159

    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 Plugin Extension 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: 158_plugin_extension.py
    Total Lines: 640
    Pattern Category: AI Agent Design Pattern #158


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Adapter Wrapper

    
        """
    Pattern 157: Adapter/Wrapper Pattern
    
    Description:
        Provides unified interfaces to different LLM providers and models, abstracting
        away provider-specific details. Enables easy switching between models without
        changing application code.
    
    Components:
        - Base interface definition
        - Provider-specific adapters
        - Response normalization
        - Error handling translation
        - Configuration management
    
    Use Cases:
        - Multi-provider support
        - Model switching
        - Vendor independence
        - Testing with different models
        - Cost optimization
    
    Benefits:
        - Provider independence
        - Simplified integration
        - Easy model switching
        - Consistent interface
        - Testability
    
    Trade-offs:
        - Abstraction overhead
        - May limit provider-specific features
        - Mapping complexity
        - Performance overhead
    
    LangChain Implementation:
        Uses abstract base classes, provider-specific implementations,
        and normalization layers for consistent responses
    """
    
    import os
    from typing import Dict, Any, List, Optional, Protocol
    from dataclasses import dataclass
    from abc import ABC, abstractmethod
    from enum import Enum
    from dotenv import load_dotenv
    
    from langchain_openai import ChatOpenAI
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser
    from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, SystemMessage
    
    load_dotenv()
    
    
    class ModelProvider(Enum):
        """Supported model providers"""
        OPENAI = "openai"
        ANTHROPIC = "anthropic"
        GOOGLE = "google"
        LOCAL = "local"
    
    
    @dataclass
    class ModelConfig:
        """Model configuration"""
        provider: ModelProvider
        model_name: str
        temperature: float = 0.7
        max_tokens: Optional[int] = None
        api_key: Optional[str] = None
    
    
    @dataclass
    class Message:
        """Standardized message format"""
        role: str  # system, user, assistant
        content: str
        metadata: Dict[str, Any] = None
        
        def __post_init__(self):
            if self.metadata is None:
                self.metadata = {}
    
    
    @dataclass
    class CompletionResponse:
        """Standardized completion response"""
        content: str
        model: str
        provider: ModelProvider
        tokens_used: Optional[int] = None
        finish_reason: Optional[str] = None
        metadata: Dict[str, Any] = None
        
        def __post_init__(self):
            if self.metadata is None:
                self.metadata = {}
    
    
    class LLMAdapter(ABC):
        """Abstract base class for LLM adapters"""
        
        @abstractmethod
        def complete(self, messages: List[Message], **kwargs) -> CompletionResponse:
            """Generate completion from messages"""
            pass
        
        @abstractmethod
        def stream(self, messages: List[Message], **kwargs):
            """Stream completion"""
            pass
        
        @abstractmethod
        def get_model_info(self) -> Dict[str, Any]:
            """Get model information"""
            pass
    
    
    class OpenAIAdapter(LLMAdapter):
        """Adapter for OpenAI models"""
        
        def __init__(self, config: ModelConfig):
            """Initialize OpenAI adapter"""
            self.config = config
            self.llm = ChatOpenAI(
                model=config.model_name,
                temperature=config.temperature,
                max_tokens=config.max_tokens,
                api_key=config.api_key or os.getenv("OPENAI_API_KEY")
            )
        
        def complete(self, messages: List[Message], **kwargs) -> CompletionResponse:
            """Generate completion"""
            # Convert messages to LangChain format
            lc_messages = self._convert_messages(messages)
            
            try:
                # Generate response
                response = self.llm.invoke(lc_messages)
                
                return CompletionResponse(
                    content=response.content,
                    model=self.config.model_name,
                    provider=ModelProvider.OPENAI,
                    finish_reason="stop",
                    metadata={
                        "response_type": type(response).__name__
                    }
                )
            except Exception as e:
                raise RuntimeError(f"OpenAI completion failed: {str(e)}")
        
        def stream(self, messages: List[Message], **kwargs):
            """Stream completion"""
            lc_messages = self._convert_messages(messages)
            
            try:
                for chunk in self.llm.stream(lc_messages):
                    yield chunk.content
            except Exception as e:
                raise RuntimeError(f"OpenAI streaming failed: {str(e)}")
        
        def get_model_info(self) -> Dict[str, Any]:
            """Get model information"""
            return {
                "provider": ModelProvider.OPENAI.value,
                "model": self.config.model_name,
                "temperature": self.config.temperature,
                "max_tokens": self.config.max_tokens
            }
        
        def _convert_messages(self, messages: List[Message]) -> List[BaseMessage]:
            """Convert standard messages to LangChain messages"""
            lc_messages = []
            for msg in messages:
                if msg.role == "system":
                    lc_messages.append(SystemMessage(content=msg.content))
                elif msg.role == "user":
                    lc_messages.append(HumanMessage(content=msg.content))
                elif msg.role == "assistant":
                    lc_messages.append(AIMessage(content=msg.content))
            return lc_messages
    
    
    class AnthropicAdapter(LLMAdapter):
        """Adapter for Anthropic models (simulated)"""
        
        def __init__(self, config: ModelConfig):
            """Initialize Anthropic adapter"""
            self.config = config
            # In real implementation, use actual Anthropic client
            # For demo, we'll use OpenAI as fallback
            self.llm = ChatOpenAI(
                model="gpt-3.5-turbo",
                temperature=config.temperature
            )
        
        def complete(self, messages: List[Message], **kwargs) -> CompletionResponse:
            """Generate completion"""
            # Convert to format expected by Anthropic (simulated)
            prompt = self._format_anthropic_prompt(messages)
            
            try:
                response = self.llm.invoke([HumanMessage(content=prompt)])
                
                return CompletionResponse(
                    content=response.content,
                    model=self.config.model_name,
                    provider=ModelProvider.ANTHROPIC,
                    finish_reason="end_turn",
                    metadata={
                        "simulated": True
                    }
                )
            except Exception as e:
                raise RuntimeError(f"Anthropic completion failed: {str(e)}")
        
        def stream(self, messages: List[Message], **kwargs):
            """Stream completion"""
            prompt = self._format_anthropic_prompt(messages)
            
            try:
                for chunk in self.llm.stream([HumanMessage(content=prompt)]):
                    yield chunk.content
            except Exception as e:
                raise RuntimeError(f"Anthropic streaming failed: {str(e)}")
        
        def get_model_info(self) -> Dict[str, Any]:
            """Get model information"""
            return {
                "provider": ModelProvider.ANTHROPIC.value,
                "model": self.config.model_name,
                "temperature": self.config.temperature,
                "simulated": True
            }
        
        def _format_anthropic_prompt(self, messages: List[Message]) -> str:
            """Format messages for Anthropic API"""
            # Anthropic uses a different format
            formatted = []
            for msg in messages:
                if msg.role == "system":
                    formatted.append(f"System: {msg.content}")
                elif msg.role == "user":
                    formatted.append(f"Human: {msg.content}")
                elif msg.role == "assistant":
                    formatted.append(f"Assistant: {msg.content}")
            formatted.append("Assistant:")
            return "\n\n".join(formatted)
    
    
    class GoogleAdapter(LLMAdapter):
        """Adapter for Google models (simulated)"""
        
        def __init__(self, config: ModelConfig):
            """Initialize Google adapter"""
            self.config = config
            # In real implementation, use actual Google client
            self.llm = ChatOpenAI(
                model="gpt-3.5-turbo",
                temperature=config.temperature
            )
        
        def complete(self, messages: List[Message], **kwargs) -> CompletionResponse:
            """Generate completion"""
            # Convert to Google format (simulated)
            lc_messages = []
            for msg in messages:
                if msg.role == "system":
                    lc_messages.append(SystemMessage(content=msg.content))
                elif msg.role == "user":
                    lc_messages.append(HumanMessage(content=msg.content))
            
            try:
                response = self.llm.invoke(lc_messages)
                
                return CompletionResponse(
                    content=response.content,
                    model=self.config.model_name,
                    provider=ModelProvider.GOOGLE,
                    finish_reason="STOP",
                    metadata={
                        "simulated": True
                    }
                )
            except Exception as e:
                raise RuntimeError(f"Google completion failed: {str(e)}")
        
        def stream(self, messages: List[Message], **kwargs):
            """Stream completion"""
            lc_messages = []
            for msg in messages:
                if msg.role == "user":
                    lc_messages.append(HumanMessage(content=msg.content))
            
            try:
                for chunk in self.llm.stream(lc_messages):
                    yield chunk.content
            except Exception as e:
                raise RuntimeError(f"Google streaming failed: {str(e)}")
        
        def get_model_info(self) -> Dict[str, Any]:
            """Get model information"""
            return {
                "provider": ModelProvider.GOOGLE.value,
                "model": self.config.model_name,
                "temperature": self.config.temperature,
                "simulated": True
            }
    
    
    class UnifiedLLMInterface:
        """Unified interface for multiple LLM providers"""
        
        def __init__(self):
            """Initialize unified interface"""
            self.adapters: Dict[ModelProvider, LLMAdapter] = {}
            self.default_provider: Optional[ModelProvider] = None
        
        def register_adapter(self, provider: ModelProvider, adapter: LLMAdapter):
            """Register a provider adapter"""
            self.adapters[provider] = adapter
            if self.default_provider is None:
                self.default_provider = provider
            print(f"✓ Registered adapter for {provider.value}")
        
        def set_default_provider(self, provider: ModelProvider):
            """Set default provider"""
            if provider not in self.adapters:
                raise ValueError(f"Provider {provider.value} not registered")
            self.default_provider = provider
            print(f"✓ Set default provider to {provider.value}")
        
        def complete(self, messages: List[Message], 
                    provider: Optional[ModelProvider] = None, 
                    **kwargs) -> CompletionResponse:
            """Generate completion using specified or default provider"""
            provider = provider or self.default_provider
            
            if provider not in self.adapters:
                raise ValueError(f"Provider {provider.value} not registered")
            
            adapter = self.adapters[provider]
            return adapter.complete(messages, **kwargs)
        
        def stream(self, messages: List[Message], 
                  provider: Optional[ModelProvider] = None, 
                  **kwargs):
            """Stream completion using specified or default provider"""
            provider = provider or self.default_provider
            
            if provider not in self.adapters:
                raise ValueError(f"Provider {provider.value} not registered")
            
            adapter = self.adapters[provider]
            return adapter.stream(messages, **kwargs)
        
        def complete_with_fallback(self, messages: List[Message], 
                                  providers: Optional[List[ModelProvider]] = None,
                                  **kwargs) -> CompletionResponse:
            """Try completion with fallback providers"""
            if providers is None:
                providers = [self.default_provider] + [
                    p for p in self.adapters.keys() if p != self.default_provider
                ]
            
            last_error = None
            for provider in providers:
                try:
                    print(f"Trying provider: {provider.value}")
                    return self.complete(messages, provider=provider, **kwargs)
                except Exception as e:
                    print(f"  Failed: {str(e)}")
                    last_error = e
                    continue
            
            raise RuntimeError(f"All providers failed. Last error: {last_error}")
        
        def get_all_model_info(self) -> Dict[str, Dict[str, Any]]:
            """Get information about all registered models"""
            return {
                provider.value: adapter.get_model_info()
                for provider, adapter in self.adapters.items()
            }
    
    
    def demonstrate_adapter_pattern():
        """Demonstrate adapter/wrapper pattern"""
        print("=" * 80)
        print("ADAPTER/WRAPPER PATTERN DEMONSTRATION")
        print("=" * 80)
        
        # Initialize unified interface
        interface = UnifiedLLMInterface()
        
        # Register adapters for different providers
        print("\n" + "="*80)
        print("SETUP: Registering Provider Adapters")
        print("="*80)
        
        openai_config = ModelConfig(
            provider=ModelProvider.OPENAI,
            model_name="gpt-3.5-turbo",
            temperature=0.7
        )
        interface.register_adapter(
            ModelProvider.OPENAI,
            OpenAIAdapter(openai_config)
        )
        
        anthropic_config = ModelConfig(
            provider=ModelProvider.ANTHROPIC,
            model_name="claude-3-sonnet",
            temperature=0.7
        )
        interface.register_adapter(
            ModelProvider.ANTHROPIC,
            AnthropicAdapter(anthropic_config)
        )
        
        google_config = ModelConfig(
            provider=ModelProvider.GOOGLE,
            model_name="gemini-pro",
            temperature=0.7
        )
        interface.register_adapter(
            ModelProvider.GOOGLE,
            GoogleAdapter(google_config)
        )
        
        # Example 1: Complete with default provider
        print("\n" + "="*80)
        print("EXAMPLE 1: Completion with Default Provider")
        print("="*80)
        messages = [
            Message(role="system", content="You are a helpful assistant."),
            Message(role="user", content="What is the capital of France?")
        ]
        response = interface.complete(messages)
        print(f"Provider: {response.provider.value}")
        print(f"Model: {response.model}")
        print(f"Response: {response.content[:200]}...")
        
        # Example 2: Complete with specific provider
        print("\n" + "="*80)
        print("EXAMPLE 2: Completion with Specific Provider (Anthropic)")
        print("="*80)
        response = interface.complete(messages, provider=ModelProvider.ANTHROPIC)
        print(f"Provider: {response.provider.value}")
        print(f"Model: {response.model}")
        print(f"Response: {response.content[:200]}...")
        
        # Example 3: Switch between providers
        print("\n" + "="*80)
        print("EXAMPLE 3: Switching Between Providers")
        print("="*80)
        question = Message(role="user", content="Explain quantum computing briefly.")
        
        for provider in [ModelProvider.OPENAI, ModelProvider.GOOGLE, ModelProvider.ANTHROPIC]:
            response = interface.complete([question], provider=provider)
            print(f"\n{provider.value}:")
            print(f"  {response.content[:150]}...")
        
        # Example 4: Streaming response
        print("\n" + "="*80)
        print("EXAMPLE 4: Streaming Response")
        print("="*80)
        stream_messages = [
            Message(role="user", content="Count from 1 to 5.")
        ]
        print("Streaming: ", end="", flush=True)
        for chunk in interface.stream(stream_messages):
            print(chunk, end="", flush=True)
        print()
        
        # Example 5: Fallback mechanism
        print("\n" + "="*80)
        print("EXAMPLE 5: Fallback Mechanism")
        print("="*80)
        messages = [
            Message(role="user", content="What is machine learning?")
        ]
        response = interface.complete_with_fallback(messages)
        print(f"Succeeded with provider: {response.provider.value}")
        print(f"Response: {response.content[:200]}...")
        
        # Example 6: Model information
        print("\n" + "="*80)
        print("EXAMPLE 6: Model Information")
        print("="*80)
        all_info = interface.get_all_model_info()
        for provider, info in all_info.items():
            print(f"\n{provider}:")
            for key, value in info.items():
                print(f"  {key}: {value}")
        
        # Example 7: Conversation across providers
        print("\n" + "="*80)
        print("EXAMPLE 7: Multi-Turn Conversation")
        print("="*80)
        conversation = [
            Message(role="user", content="What is Python?")
        ]
        
        # First turn with OpenAI
        response1 = interface.complete(conversation, provider=ModelProvider.OPENAI)
        print(f"Turn 1 ({response1.provider.value}): {response1.content[:100]}...")
        
        # Add response and continue
        conversation.append(Message(role="assistant", content=response1.content))
        conversation.append(Message(role="user", content="What are its main features?"))
        
        # Second turn with different provider
        response2 = interface.complete(conversation, provider=ModelProvider.GOOGLE)
        print(f"Turn 2 ({response2.provider.value}): {response2.content[:100]}...")
        
        # Summary
        print("\n" + "="*80)
        print("SUMMARY: Adapter/Wrapper Pattern Best Practices")
        print("="*80)
        print("""
    1. INTERFACE DESIGN:
       - Define clear, provider-agnostic interface
       - Standardize message formats
       - Normalize response structures
       - Handle provider-specific features gracefully
    
    2. ADAPTER IMPLEMENTATION:
       - Encapsulate provider-specific logic
       - Handle format conversions
       - Map error codes consistently
       - Maintain provider-specific optimizations
    
    3. RESPONSE NORMALIZATION:
       - Standardize response format
       - Map finish reasons consistently
       - Include provider metadata
       - Handle streaming uniformly
    
    4. ERROR HANDLING:
       - Translate provider-specific errors
       - Provide meaningful error messages
       - Implement fallback mechanisms
       - Log provider-specific details
    
    5. CONFIGURATION:
       - Centralize configuration management
       - Support provider-specific parameters
       - Enable easy provider switching
       - Validate configurations
    
    6. TESTING:
       - Test each adapter independently
       - Verify format conversions
       - Test fallback mechanisms
       - Mock external API calls
    
    Benefits:
    ✓ Provider independence and flexibility
    ✓ Easy model switching
    ✓ Simplified application code
    ✓ Consistent interface across providers
    ✓ Facilitates testing and mocking
    ✓ Vendor lock-in prevention
    
    Challenges:
    - Abstraction may limit provider features
    - Mapping complexity for edge cases
    - Performance overhead
    - Maintaining feature parity
    - Handling provider-specific capabilities
        """)
    
    
    if __name__ == "__main__":
        demonstrate_adapter_pattern()
    
        

    Agentic Design Pattern: Adapter Wrapper

    Pattern 157: Adapter Wrapper – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Adapter Wrapper pattern implementation.


    Pattern Description

    Pattern 157: Adapter/Wrapper Pattern

    Description: Provides unified interfaces to different LLM providers and models, abstracting away provider-specific details. Enables easy switching between models without changing application code.

    Components: – Base interface definition – Provider-specific adapters – Response normalization – Error handling translation – Configuration management

    Use Cases: – Multi-provider support – Model switching – Vendor independence – Testing with different models – Cost optimization

    Benefits: – Provider independence – Simplified integration – Easy model switching – Consistent interface – Testability

    Trade-offs: – Abstraction overhead – May limit provider-specific features – Mapping complexity – Performance overhead

    LangChain Implementation: Uses abstract base classes, provider-specific implementations, and normalization layers for consistent responses


    File Statistics

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

    Module Documentation

    """
    Pattern 157: Adapter/Wrapper Pattern
    
    Description:
        Provides unified interfaces to different LLM providers and models, abstracting
        away provider-specific details. Enables easy switching between models without
        changing application code.
    
    Components:
        - Base interface definition
        - Provider-specific adapters
        - Response normalization
        - Error handling translation
        - Configuration management
    
    Use Cases:
        - Multi-provider support
        - Model switching
        - Vendor independence
        - Testing with different models
        - Cost optimization
    
    Benefits:
        - Provider independence
        - Simplified integration
        - Easy model switching
        - Consistent interface
        - Testability
    
    Trade-offs:
        - Abstraction overhead
        - May limit provider-specific features
        - Mapping complexity
        - Performance overhead
    
    LangChain Implementation:
        Uses abstract base classes, provider-specific implementations,
        and normalization layers for consistent responses
    """
    

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


    Import Statements

    Standard Library Imports

    import dataclasses.dataclass
    import enum.Enum
    import langchain_core.messages.AIMessage
    import langchain_core.messages.BaseMessage
    import langchain_core.messages.HumanMessage
    import langchain_core.messages.SystemMessage
    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.Protocol
    

    Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – enum.Enum: Enumeration support – langchain_core.messages.AIMessage: Regular expression operations – langchain_core.messages.BaseMessage: 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 – 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.Protocol: 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 abc.ABC
    from abc.abstractmethod
    from dotenv.load_dotenv
    

    Purpose: Additional dependencies for specialized functionality.


    Classes

    Class: ModelProvider

    Lines 56-61

    Inherits from: Enum

    Purpose:

    Supported model providers
    

    Class: ModelConfig

    Lines 65-71

    Purpose:

    Model configuration
    

    Class: Message

    Lines 75-83

    Purpose:

    Standardized message format
    

    Methods:

    __post_init__(self)

    Lines 81-83

    Purpose: Post Init operation.

    Parameters:


    Class: CompletionResponse

    Lines 87-98

    Purpose:

    Standardized completion response
    

    Methods:

    __post_init__(self)

    Lines 96-98

    Purpose: Post Init operation.

    Parameters:


    Class: LLMAdapter

    Lines 101-117

    Inherits from: ABC

    Purpose:

    Abstract base class for LLM adapters
    

    Methods:

    complete(self, messages)

    Lines 105-107

    Decorators: abstractmethod

    Purpose:

    Generate completion from messages
    

    Parameters: – messages: Messages

    Returns: CompletionResponse

    stream(self, messages)

    Lines 110-112

    Decorators: abstractmethod

    Purpose:

    Stream completion
    

    Parameters: – messages: Messages

    get_model_info(self)

    Lines 115-117

    Decorators: abstractmethod

    Purpose:

    Get model information
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D75BC50>


    Class: OpenAIAdapter

    Lines 120-183

    Inherits from: LLMAdapter

    Purpose:

    Adapter for OpenAI models
    

    Methods:

    __init__(self, config)

    Lines 123-131

    Purpose:

    Initialize OpenAI adapter
    

    Parameters: – config: Config

    complete(self, messages)

    Lines 133-152

    Purpose:

    Generate completion
    

    Parameters: – messages: Messages

    Returns: CompletionResponse

    stream(self, messages)

    Lines 154-162

    Purpose:

    Stream completion
    

    Parameters: – messages: Messages

    get_model_info(self)

    Lines 164-171

    Purpose:

    Get model information
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D753690>

    _convert_messages(self, messages)

    Lines 173-183

    Purpose:

    Convert standard messages to LangChain messages
    

    Parameters: – messages: Messages

    Returns: <ast.Subscript object at 0x000001A06D7CB290>


    Class: AnthropicAdapter

    Lines 186-250

    Inherits from: LLMAdapter

    Purpose:

    Adapter for Anthropic models (simulated)
    

    Methods:

    __init__(self, config)

    Lines 189-197

    Purpose:

    Initialize Anthropic adapter
    

    Parameters: – config: Config

    complete(self, messages)

    Lines 199-217

    Purpose:

    Generate completion
    

    Parameters: – messages: Messages

    Returns: CompletionResponse

    stream(self, messages)

    Lines 219-227

    Purpose:

    Stream completion
    

    Parameters: – messages: Messages

    get_model_info(self)

    Lines 229-236

    Purpose:

    Get model information
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D7AB310>

    _format_anthropic_prompt(self, messages)

    Lines 238-250

    Purpose:

    Format messages for Anthropic API
    

    Parameters: – messages: Messages

    Returns: str


    Class: GoogleAdapter

    Lines 253-310

    Inherits from: LLMAdapter

    Purpose:

    Adapter for Google models (simulated)
    

    Methods:

    __init__(self, config)

    Lines 256-263

    Purpose:

    Initialize Google adapter
    

    Parameters: – config: Config

    complete(self, messages)

    Lines 265-288

    Purpose:

    Generate completion
    

    Parameters: – messages: Messages

    Returns: CompletionResponse

    stream(self, messages)

    Lines 290-301

    Purpose:

    Stream completion
    

    Parameters: – messages: Messages

    get_model_info(self)

    Lines 303-310

    Purpose:

    Get model information
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D80FD50>


    Class: UnifiedLLMInterface

    Lines 313-385

    Purpose:

    Unified interface for multiple LLM providers
    

    Methods:

    __init__(self)

    Lines 316-319

    Purpose:

    Initialize unified interface
    

    Parameters:

    register_adapter(self, provider, adapter)

    Lines 321-326

    Purpose:

    Register a provider adapter
    

    Parameters: – provider: Provider – adapter: Adapter

    set_default_provider(self, provider)

    Lines 328-333

    Purpose:

    Set default provider
    

    Parameters: – provider: Provider

    complete(self, messages, provider)

    Lines 335-345

    Purpose:

    Generate completion using specified or default provider
    

    Parameters: – messages: Messages – provider: Provider

    Returns: CompletionResponse

    stream(self, messages, provider)

    Lines 347-357

    Purpose:

    Stream completion using specified or default provider
    

    Parameters: – messages: Messages – provider: Provider

    complete_with_fallback(self, messages, providers)

    Lines 359-378

    Purpose:

    Try completion with fallback providers
    

    Parameters: – messages: Messages – providers: Providers

    Returns: CompletionResponse

    get_all_model_info(self)

    Lines 380-385

    Purpose:

    Get information about all registered models
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D622590>


    Functions

    demonstrate_adapter_pattern()

    Lines 388-573

    Purpose:

    Demonstrate adapter/wrapper pattern
    

    Key Concepts Summary

    Pattern Overview

    The Adapter Wrapper 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 156
    • Pattern 158

    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 Adapter Wrapper 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: 157_adapter_wrapper.py
    Total Lines: 578
    Pattern Category: AI Agent Design Pattern #157


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Api Gateway

    
        """
    Pattern 156: API Gateway Pattern
    
    Description:
        Centralized gateway for managing API requests to multiple agents/services.
        Provides routing, authentication, rate limiting, monitoring, and unified
        interface for diverse backend services.
    
    Components:
        - Request router
        - Authentication & authorization
        - Rate limiting
        - Load balancing
        - Monitoring & logging
        - Response transformation
        - Error handling
    
    Use Cases:
        - Multi-agent systems
        - Microservices architecture
        - API versioning
        - Security enforcement
        - Traffic management
    
    Benefits:
        - Centralized control
        - Security enforcement
        - Traffic management
        - Simplified client interface
        - Monitoring visibility
    
    Trade-offs:
        - Single point of failure
        - Added latency
        - Complexity overhead
        - Gateway bottleneck potential
    
    LangChain Implementation:
        Uses routing logic, rate limiting, authentication checks,
        and multiple LLM backends with unified interface
    """
    
    import os
    import time
    from typing import Dict, Any, Optional, Callable, List
    from dataclasses import dataclass, field
    from datetime import datetime, timedelta
    from enum import Enum
    from collections import defaultdict
    from dotenv import load_dotenv
    
    from langchain_openai import ChatOpenAI
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser
    
    load_dotenv()
    
    
    class ServiceType(Enum):
        """Types of backend services"""
        CHAT = "chat"
        COMPLETION = "completion"
        ANALYSIS = "analysis"
        TRANSLATION = "translation"
        SUMMARIZATION = "summarization"
    
    
    class AuthLevel(Enum):
        """Authentication levels"""
        PUBLIC = "public"
        AUTHENTICATED = "authenticated"
        PREMIUM = "premium"
        ADMIN = "admin"
    
    
    @dataclass
    class Route:
        """API route configuration"""
        path: str
        service_type: ServiceType
        handler: Callable
        auth_level: AuthLevel = AuthLevel.PUBLIC
        rate_limit: int = 60  # requests per minute
        timeout: float = 30.0  # seconds
    
    
    @dataclass
    class ServiceConfig:
        """Backend service configuration"""
        name: str
        service_type: ServiceType
        model_name: str
        enabled: bool = True
        weight: int = 1  # For load balancing
        max_concurrent: int = 10
    
    
    @dataclass
    class Request:
        """Gateway request"""
        path: str
        method: str
        payload: Dict[str, Any]
        headers: Dict[str, str] = field(default_factory=dict)
        user_id: Optional[str] = None
        api_key: Optional[str] = None
        timestamp: datetime = field(default_factory=datetime.now)
    
    
    @dataclass
    class Response:
        """Gateway response"""
        status_code: int
        data: Any
        headers: Dict[str, str] = field(default_factory=dict)
        execution_time: float = 0.0
        service_name: Optional[str] = None
        error: Optional[str] = None
    
    
    @dataclass
    class RateLimitInfo:
        """Rate limit tracking"""
        requests: List[datetime] = field(default_factory=list)
        limit: int = 60
        window: int = 60  # seconds
    
    
    class APIGateway:
        """Centralized API gateway for agent services"""
        
        def __init__(self):
            """Initialize API gateway"""
            self.routes: Dict[str, Route] = {}
            self.services: Dict[str, ServiceConfig] = {}
            self.rate_limits: Dict[str, RateLimitInfo] = defaultdict(
                lambda: RateLimitInfo()
            )
            self.api_keys: Dict[str, AuthLevel] = {}
            self.request_log: List[Dict[str, Any]] = []
            
            # Initialize services
            self._initialize_services()
            
            # Set up routes
            self._setup_routes()
        
        def _initialize_services(self):
            """Initialize backend services"""
            self.services = {
                "chat_service": ServiceConfig(
                    name="chat_service",
                    service_type=ServiceType.CHAT,
                    model_name="gpt-3.5-turbo"
                ),
                "analysis_service": ServiceConfig(
                    name="analysis_service",
                    service_type=ServiceType.ANALYSIS,
                    model_name="gpt-3.5-turbo"
                ),
                "translation_service": ServiceConfig(
                    name="translation_service",
                    service_type=ServiceType.TRANSLATION,
                    model_name="gpt-3.5-turbo"
                ),
            }
            
            # Initialize LLMs for services
            self.llms = {
                name: ChatOpenAI(model=config.model_name, temperature=0.7)
                for name, config in self.services.items()
            }
        
        def _setup_routes(self):
            """Set up API routes"""
            self.routes = {
                "/chat": Route(
                    path="/chat",
                    service_type=ServiceType.CHAT,
                    handler=self._handle_chat,
                    auth_level=AuthLevel.AUTHENTICATED,
                    rate_limit=30
                ),
                "/analyze": Route(
                    path="/analyze",
                    service_type=ServiceType.ANALYSIS,
                    handler=self._handle_analysis,
                    auth_level=AuthLevel.PREMIUM,
                    rate_limit=10
                ),
                "/translate": Route(
                    path="/translate",
                    service_type=ServiceType.TRANSLATION,
                    handler=self._handle_translation,
                    auth_level=AuthLevel.AUTHENTICATED,
                    rate_limit=20
                ),
                "/summarize": Route(
                    path="/summarize",
                    service_type=ServiceType.SUMMARIZATION,
                    handler=self._handle_summarization,
                    auth_level=AuthLevel.AUTHENTICATED,
                    rate_limit=15
                ),
            }
        
        def register_api_key(self, api_key: str, auth_level: AuthLevel):
            """Register an API key with authentication level"""
            self.api_keys[api_key] = auth_level
        
        def authenticate(self, request: Request) -> tuple[bool, Optional[str]]:
            """Authenticate request"""
            # Get route
            route = self.routes.get(request.path)
            if not route:
                return False, "Route not found"
            
            # Check authentication level
            if route.auth_level == AuthLevel.PUBLIC:
                return True, None
            
            # Check API key
            if not request.api_key:
                return False, "API key required"
            
            user_level = self.api_keys.get(request.api_key)
            if not user_level:
                return False, "Invalid API key"
            
            # Check authorization level
            required_levels = {
                AuthLevel.AUTHENTICATED: [AuthLevel.AUTHENTICATED, AuthLevel.PREMIUM, AuthLevel.ADMIN],
                AuthLevel.PREMIUM: [AuthLevel.PREMIUM, AuthLevel.ADMIN],
                AuthLevel.ADMIN: [AuthLevel.ADMIN]
            }
            
            if user_level not in required_levels.get(route.auth_level, []):
                return False, f"Insufficient permissions. Requires {route.auth_level.value}"
            
            return True, None
        
        def check_rate_limit(self, request: Request) -> tuple[bool, Optional[str]]:
            """Check if request exceeds rate limit"""
            route = self.routes.get(request.path)
            if not route:
                return True, None
            
            # Get rate limit info for user
            key = request.api_key or request.user_id or "anonymous"
            rate_info = self.rate_limits[key]
            
            # Update limit if different from current
            if rate_info.limit != route.rate_limit:
                rate_info.limit = route.rate_limit
            
            # Clean old requests outside window
            now = datetime.now()
            cutoff = now - timedelta(seconds=rate_info.window)
            rate_info.requests = [
                req_time for req_time in rate_info.requests
                if req_time > cutoff
            ]
            
            # Check limit
            if len(rate_info.requests) >= rate_info.limit:
                return False, f"Rate limit exceeded. Max {rate_info.limit} requests per minute"
            
            # Add current request
            rate_info.requests.append(now)
            return True, None
        
        def route_request(self, request: Request) -> Response:
            """Route and process request through gateway"""
            start_time = time.time()
            
            # Log request
            self._log_request(request)
            
            # Authenticate
            auth_ok, auth_error = self.authenticate(request)
            if not auth_ok:
                return Response(
                    status_code=401,
                    data=None,
                    error=auth_error,
                    execution_time=time.time() - start_time
                )
            
            # Check rate limit
            rate_ok, rate_error = self.check_rate_limit(request)
            if not rate_ok:
                return Response(
                    status_code=429,
                    data=None,
                    error=rate_error,
                    execution_time=time.time() - start_time
                )
            
            # Get route
            route = self.routes.get(request.path)
            if not route:
                return Response(
                    status_code=404,
                    data=None,
                    error=f"Route not found: {request.path}",
                    execution_time=time.time() - start_time
                )
            
            # Select service
            service = self._select_service(route.service_type)
            if not service:
                return Response(
                    status_code=503,
                    data=None,
                    error="No available service",
                    execution_time=time.time() - start_time
                )
            
            # Execute request
            try:
                result = route.handler(request, service)
                execution_time = time.time() - start_time
                
                return Response(
                    status_code=200,
                    data=result,
                    execution_time=execution_time,
                    service_name=service.name
                )
                
            except Exception as e:
                execution_time = time.time() - start_time
                return Response(
                    status_code=500,
                    data=None,
                    error=str(e),
                    execution_time=execution_time,
                    service_name=service.name
                )
        
        def _select_service(self, service_type: ServiceType) -> Optional[ServiceConfig]:
            """Select service based on type and load balancing"""
            # Find services matching type
            matching = [
                s for s in self.services.values()
                if s.service_type == service_type and s.enabled
            ]
            
            if not matching:
                return None
            
            # Simple weighted random selection
            import random
            weights = [s.weight for s in matching]
            return random.choices(matching, weights=weights)[0]
        
        def _handle_chat(self, request: Request, service: ServiceConfig) -> str:
            """Handle chat request"""
            llm = self.llms[service.name]
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are a helpful AI assistant."),
                ("user", "{message}")
            ])
            
            chain = prompt | llm | StrOutputParser()
            return chain.invoke({"message": request.payload.get("message", "")})
        
        def _handle_analysis(self, request: Request, service: ServiceConfig) -> str:
            """Handle analysis request"""
            llm = self.llms[service.name]
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are an expert analyst. Provide detailed analysis."),
                ("user", "Analyze the following:\n\n{text}")
            ])
            
            chain = prompt | llm | StrOutputParser()
            return chain.invoke({"text": request.payload.get("text", "")})
        
        def _handle_translation(self, request: Request, service: ServiceConfig) -> str:
            """Handle translation request"""
            llm = self.llms[service.name]
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are a professional translator."),
                ("user", "Translate the following from {source} to {target}:\n\n{text}")
            ])
            
            chain = prompt | llm | StrOutputParser()
            return chain.invoke({
                "source": request.payload.get("source_lang", "auto"),
                "target": request.payload.get("target_lang", "English"),
                "text": request.payload.get("text", "")
            })
        
        def _handle_summarization(self, request: Request, service: ServiceConfig) -> str:
            """Handle summarization request"""
            llm = self.llms[service.name]
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are an expert at creating concise summaries."),
                ("user", "Summarize the following in {max_words} words:\n\n{text}")
            ])
            
            chain = prompt | llm | StrOutputParser()
            return chain.invoke({
                "max_words": request.payload.get("max_words", 100),
                "text": request.payload.get("text", "")
            })
        
        def _log_request(self, request: Request):
            """Log request for monitoring"""
            self.request_log.append({
                "timestamp": request.timestamp.isoformat(),
                "path": request.path,
                "method": request.method,
                "user_id": request.user_id,
                "api_key_hash": hash(request.api_key) if request.api_key else None
            })
        
        def get_metrics(self) -> Dict[str, Any]:
            """Get gateway metrics"""
            now = datetime.now()
            recent = now - timedelta(minutes=5)
            
            recent_requests = [
                log for log in self.request_log
                if datetime.fromisoformat(log["timestamp"]) > recent
            ]
            
            # Count by path
            by_path = defaultdict(int)
            for log in recent_requests:
                by_path[log["path"]] += 1
            
            # Rate limit status
            rate_status = {}
            for key, info in self.rate_limits.items():
                rate_status[key] = {
                    "current": len(info.requests),
                    "limit": info.limit,
                    "usage_pct": len(info.requests) / info.limit * 100
                }
            
            return {
                "total_requests_5min": len(recent_requests),
                "requests_by_path": dict(by_path),
                "rate_limit_status": rate_status,
                "active_services": len([s for s in self.services.values() if s.enabled]),
                "total_services": len(self.services)
            }
    
    
    def demonstrate_api_gateway():
        """Demonstrate API gateway pattern"""
        print("=" * 80)
        print("API GATEWAY PATTERN DEMONSTRATION")
        print("=" * 80)
        
        # Initialize gateway
        gateway = APIGateway()
        
        # Register API keys
        gateway.register_api_key("free_user_key", AuthLevel.AUTHENTICATED)
        gateway.register_api_key("premium_user_key", AuthLevel.PREMIUM)
        gateway.register_api_key("admin_key", AuthLevel.ADMIN)
        
        # Example 1: Successful authenticated request
        print("\n" + "="*80)
        print("EXAMPLE 1: Successful Chat Request (Authenticated)")
        print("="*80)
        request = Request(
            path="/chat",
            method="POST",
            payload={"message": "What is machine learning?"},
            api_key="free_user_key",
            user_id="user123"
        )
        response = gateway.route_request(request)
        print(f"Status: {response.status_code}")
        print(f"Service: {response.service_name}")
        print(f"Time: {response.execution_time:.2f}s")
        print(f"Response: {response.data[:200] if response.data else response.error}...")
        
        # Example 2: Premium service request
        print("\n" + "="*80)
        print("EXAMPLE 2: Analysis Request (Premium)")
        print("="*80)
        request = Request(
            path="/analyze",
            method="POST",
            payload={"text": "The market showed strong growth in Q1 with a 15% increase in sales."},
            api_key="premium_user_key"
        )
        response = gateway.route_request(request)
        print(f"Status: {response.status_code}")
        print(f"Service: {response.service_name}")
        print(f"Response: {response.data[:200] if response.data else response.error}...")
        
        # Example 3: Authentication failure
        print("\n" + "="*80)
        print("EXAMPLE 3: Authentication Failure")
        print("="*80)
        request = Request(
            path="/analyze",
            method="POST",
            payload={"text": "Test"},
            api_key="free_user_key"  # Not premium
        )
        response = gateway.route_request(request)
        print(f"Status: {response.status_code}")
        print(f"Error: {response.error}")
        
        # Example 4: Rate limiting
        print("\n" + "="*80)
        print("EXAMPLE 4: Rate Limiting")
        print("="*80)
        # Make multiple requests quickly
        for i in range(3):
            request = Request(
                path="/chat",
                method="POST",
                payload={"message": f"Request {i+1}"},
                api_key="free_user_key"
            )
            response = gateway.route_request(request)
            print(f"Request {i+1}: Status {response.status_code} - {response.error if response.error else 'Success'}")
        
        # Example 5: Translation service
        print("\n" + "="*80)
        print("EXAMPLE 5: Translation Service")
        print("="*80)
        request = Request(
            path="/translate",
            method="POST",
            payload={
                "text": "Hello, how are you?",
                "source_lang": "English",
                "target_lang": "Spanish"
            },
            api_key="free_user_key"
        )
        response = gateway.route_request(request)
        print(f"Status: {response.status_code}")
        print(f"Translation: {response.data if response.data else response.error}")
        
        # Example 6: Gateway metrics
        print("\n" + "="*80)
        print("EXAMPLE 6: Gateway Metrics")
        print("="*80)
        metrics = gateway.get_metrics()
        print("Gateway Metrics:")
        print(f"  Total requests (5 min): {metrics['total_requests_5min']}")
        print(f"  Requests by path: {metrics['requests_by_path']}")
        print(f"  Active services: {metrics['active_services']}/{metrics['total_services']}")
        
        # Example 7: Route not found
        print("\n" + "="*80)
        print("EXAMPLE 7: Route Not Found")
        print("="*80)
        request = Request(
            path="/nonexistent",
            method="POST",
            payload={},
            api_key="free_user_key"
        )
        response = gateway.route_request(request)
        print(f"Status: {response.status_code}")
        print(f"Error: {response.error}")
        
        # Summary
        print("\n" + "="*80)
        print("SUMMARY: API Gateway Pattern Best Practices")
        print("="*80)
        print("""
    1. ROUTING:
       - Centralized request routing
       - Service discovery and selection
       - Load balancing across backends
       - Flexible routing rules
    
    2. AUTHENTICATION & AUTHORIZATION:
       - API key management
       - Multiple authorization levels
       - Fine-grained access control
       - Secure credential handling
    
    3. RATE LIMITING:
       - Per-user rate limits
       - Per-route limits
       - Sliding window implementation
       - Graceful limit enforcement
    
    4. MONITORING:
       - Request logging
       - Performance metrics
       - Service health tracking
       - Real-time visibility
    
    5. ERROR HANDLING:
       - Standardized error responses
       - Proper HTTP status codes
       - Detailed error messages
       - Graceful degradation
    
    6. SERVICE MANAGEMENT:
       - Service registration
       - Health checks
       - Dynamic service discovery
       - Version management
    
    Benefits:
    ✓ Centralized control and management
    ✓ Security enforcement at gateway level
    ✓ Simplified client integration
    ✓ Traffic management and optimization
    ✓ Monitoring and observability
    ✓ Service abstraction
    
    Challenges:
    - Single point of failure risk
    - Added network latency
    - Gateway complexity
    - Potential bottleneck
    - Requires high availability
        """)
    
    
    if __name__ == "__main__":
        demonstrate_api_gateway()
    
        

    Agentic Design Pattern: Api Gateway

    Pattern 156: Api Gateway – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Api Gateway pattern implementation.


    Pattern Description

    Pattern 156: API Gateway Pattern

    Description: Centralized gateway for managing API requests to multiple agents/services. Provides routing, authentication, rate limiting, monitoring, and unified interface for diverse backend services.

    Components: – Request router – Authentication & authorization – Rate limiting – Load balancing – Monitoring & logging – Response transformation – Error handling

    Use Cases: – Multi-agent systems – Microservices architecture – API versioning – Security enforcement – Traffic management

    Benefits: – Centralized control – Security enforcement – Traffic management – Simplified client interface – Monitoring visibility

    Trade-offs: – Single point of failure – Added latency – Complexity overhead – Gateway bottleneck potential

    LangChain Implementation: Uses routing logic, rate limiting, authentication checks, and multiple LLM backends with unified interface


    File Statistics

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

    Module Documentation

    """
    Pattern 156: API Gateway Pattern
    
    Description:
        Centralized gateway for managing API requests to multiple agents/services.
        Provides routing, authentication, rate limiting, monitoring, and unified
        interface for diverse backend services.
    
    Components:
        - Request router
        - Authentication & authorization
        - Rate limiting
        - Load balancing
        - Monitoring & logging
        - Response transformation
        - Error handling
    
    Use Cases:
        - Multi-agent systems
        - Microservices architecture
        - API versioning
        - Security enforcement
        - Traffic management
    
    Benefits:
        - Centralized control
        - Security enforcement
        - Traffic management
        - Simplified client interface
        - Monitoring visibility
    
    Trade-offs:
        - Single point of failure
        - Added latency
        - Complexity overhead
        - Gateway bottleneck potential
    
    LangChain Implementation:
        Uses routing logic, rate limiting, authentication checks,
        and multiple LLM backends with unified interface
    """
    

    Purpose: This module implements the Api Gateway 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 langchain_core.output_parsers.StrOutputParser
    import langchain_core.prompts.ChatPromptTemplate
    import os
    import random
    import time
    import typing.Any
    import typing.Callable
    import typing.Dict
    import typing.List
    import typing.Optional
    

    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 – 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 – time: Time-related functions – 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: ServiceType

    Lines 59-65

    Inherits from: Enum

    Purpose:

    Types of backend services
    

    Class: AuthLevel

    Lines 68-73

    Inherits from: Enum

    Purpose:

    Authentication levels
    

    Class: Route

    Lines 77-84

    Purpose:

    API route configuration
    

    Class: ServiceConfig

    Lines 88-95

    Purpose:

    Backend service configuration
    

    Class: Request

    Lines 99-107

    Purpose:

    Gateway request
    

    Class: Response

    Lines 111-118

    Purpose:

    Gateway response
    

    Class: RateLimitInfo

    Lines 122-126

    Purpose:

    Rate limit tracking
    

    Class: APIGateway

    Lines 129-452

    Purpose:

    Centralized API gateway for agent services
    

    Methods:

    __init__(self)

    Lines 132-146

    Purpose:

    Initialize API gateway
    

    Parameters:

    _initialize_services(self)

    Lines 148-172

    Purpose:

    Initialize backend services
    

    Parameters:

    _setup_routes(self)

    Lines 174-205

    Purpose:

    Set up API routes
    

    Parameters:

    register_api_key(self, api_key, auth_level)

    Lines 207-209

    Purpose:

    Register an API key with authentication level
    

    Parameters: – api_key: Api key – auth_level: Auth level

    authenticate(self, request)

    Lines 211-240

    Purpose:

    Authenticate request
    

    Parameters: – request: Request

    Returns: <ast.Subscript object at 0x000001A06D65AE50>

    check_rate_limit(self, request)

    Lines 242-270

    Purpose:

    Check if request exceeds rate limit
    

    Parameters: – request: Request

    Returns: <ast.Subscript object at 0x000001A06D7CD210>

    route_request(self, request)

    Lines 272-339

    Purpose:

    Route and process request through gateway
    

    Parameters: – request: Request

    Returns: Response

    _select_service(self, service_type)

    Lines 341-355

    Purpose:

    Select service based on type and load balancing
    

    Parameters: – service_type: Service type

    Returns: <ast.Subscript object at 0x000001A06D6FDB90>

    _handle_chat(self, request, service)

    Lines 357-367

    Purpose:

    Handle chat request
    

    Parameters: – request: Request – service: Service

    Returns: str

    _handle_analysis(self, request, service)

    Lines 369-379

    Purpose:

    Handle analysis request
    

    Parameters: – request: Request – service: Service

    Returns: str

    _handle_translation(self, request, service)

    Lines 381-395

    Purpose:

    Handle translation request
    

    Parameters: – request: Request – service: Service

    Returns: str

    _handle_summarization(self, request, service)

    Lines 397-410

    Purpose:

    Handle summarization request
    

    Parameters: – request: Request – service: Service

    Returns: str

    _log_request(self, request)

    Lines 412-420

    Purpose:

    Log request for monitoring
    

    Parameters: – request: Request

    get_metrics(self)

    Lines 422-452

    Purpose:

    Get gateway metrics
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D60D450>


    Functions

    demonstrate_api_gateway()

    Lines 455-627

    Purpose:

    Demonstrate API gateway pattern
    

    Key Concepts Summary

    Pattern Overview

    The Api Gateway 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 155
    • Pattern 157

    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 Api Gateway 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: 156_api_gateway.py
    Total Lines: 632
    Pattern Category: AI Agent Design Pattern #156


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Regression Testing

    
        """
    Pattern 155: Regression Testing
    
    Description:
        Automated regression testing detects unintended changes in agent behavior
        when code, models, or prompts are updated. Maintains baselines, compares
        outputs, and flags regressions automatically.
    
    Components:
        - Baseline storage and versioning
        - Test case management
        - Output comparison
        - Regression detection
        - Change analysis
        - Historical tracking
    
    Use Cases:
        - Model version updates
        - Prompt engineering iterations
        - Code refactoring validation
        - CI/CD integration
        - Quality assurance
    
    Benefits:
        - Early regression detection
        - Automated quality checks
        - Historical performance tracking
        - Confidence in updates
    
    Trade-offs:
        - Baseline maintenance overhead
        - Storage requirements
        - Comparison complexity for creative tasks
        - False positive management
    
    LangChain Implementation:
        Uses ChatOpenAI for testing, JSON storage for baselines,
        similarity metrics for comparison, and automated detection
    """
    
    import os
    import json
    from typing import List, Dict, Any, Optional
    from dataclasses import dataclass, field
    from datetime import datetime
    from enum import Enum
    from difflib import SequenceMatcher
    from dotenv import load_dotenv
    
    from langchain_openai import ChatOpenAI
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser
    
    load_dotenv()
    
    
    class ChangeType(Enum):
        """Types of changes detected"""
        NO_CHANGE = "no_change"
        IMPROVEMENT = "improvement"
        REGRESSION = "regression"
        NEUTRAL = "neutral"
        ERROR = "error"
    
    
    class Severity(Enum):
        """Severity levels for regressions"""
        CRITICAL = "critical"
        HIGH = "high"
        MEDIUM = "medium"
        LOW = "low"
        INFO = "info"
    
    
    @dataclass
    class TestCase:
        """Single test case"""
        id: str
        input: str
        category: str
        expected_output: Optional[str] = None
        metadata: Dict[str, Any] = field(default_factory=dict)
    
    
    @dataclass
    class TestResult:
        """Result of a single test execution"""
        test_id: str
        input: str
        output: str
        timestamp: str
        execution_time: float
        error: Optional[str] = None
        metadata: Dict[str, Any] = field(default_factory=dict)
    
    
    @dataclass
    class Comparison:
        """Comparison between baseline and current result"""
        test_id: str
        baseline_output: str
        current_output: str
        similarity_score: float
        change_type: ChangeType
        severity: Severity
        details: str
        timestamp: str
    
    
    @dataclass
    class RegressionReport:
        """Comprehensive regression test report"""
        version: str
        timestamp: str
        total_tests: int
        passed: int
        regressions: int
        improvements: int
        errors: int
        comparisons: List[Comparison]
        summary: str
    
    
    class RegressionTester:
        """Manages regression testing for agent outputs"""
        
        def __init__(self, baseline_file: str = "baselines.json", 
                     model_name: str = "gpt-3.5-turbo"):
            """
            Initialize regression tester
            
            Args:
                baseline_file: Path to baseline storage
                model_name: LLM model to use for testing
            """
            self.llm = ChatOpenAI(model=model_name, temperature=0)
            self.baseline_file = baseline_file
            self.test_cases: List[TestCase] = []
            self.baselines: Dict[str, TestResult] = {}
            self.load_baselines()
            
            # Comparison LLM for semantic analysis
            self.comparison_llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
            
        def load_baselines(self):
            """Load baseline results from file"""
            if os.path.exists(self.baseline_file):
                try:
                    with open(self.baseline_file, 'r') as f:
                        data = json.load(f)
                        self.baselines = {
                            k: TestResult(**v) for k, v in data.items()
                        }
                    print(f"✓ Loaded {len(self.baselines)} baselines from {self.baseline_file}")
                except Exception as e:
                    print(f"Error loading baselines: {e}")
                    self.baselines = {}
            else:
                print(f"No baseline file found at {self.baseline_file}")
                self.baselines = {}
        
        def save_baselines(self):
            """Save baseline results to file"""
            try:
                data = {
                    k: {
                        'test_id': v.test_id,
                        'input': v.input,
                        'output': v.output,
                        'timestamp': v.timestamp,
                        'execution_time': v.execution_time,
                        'error': v.error,
                        'metadata': v.metadata
                    }
                    for k, v in self.baselines.items()
                }
                with open(self.baseline_file, 'w') as f:
                    json.dump(data, f, indent=2)
                print(f"✓ Saved {len(self.baselines)} baselines to {self.baseline_file}")
            except Exception as e:
                print(f"Error saving baselines: {e}")
        
        def add_test_case(self, test_case: TestCase):
            """Add a test case"""
            self.test_cases.append(test_case)
        
        def run_test(self, test_case: TestCase) -> TestResult:
            """Execute a single test case"""
            import time
            
            start_time = time.time()
            error = None
            output = ""
            
            try:
                # Create prompt for the test
                prompt = ChatPromptTemplate.from_messages([
                    ("system", "You are a helpful AI assistant. Provide clear, accurate responses."),
                    ("user", "{input}")
                ])
                
                chain = prompt | self.llm | StrOutputParser()
                output = chain.invoke({"input": test_case.input})
                
            except Exception as e:
                error = str(e)
                output = f"ERROR: {error}"
            
            execution_time = time.time() - start_time
            
            return TestResult(
                test_id=test_case.id,
                input=test_case.input,
                output=output,
                timestamp=datetime.now().isoformat(),
                execution_time=execution_time,
                error=error,
                metadata=test_case.metadata.copy()
            )
        
        def calculate_similarity(self, text1: str, text2: str) -> float:
            """Calculate similarity between two texts"""
            # Use SequenceMatcher for basic similarity
            return SequenceMatcher(None, text1.lower(), text2.lower()).ratio()
        
        def analyze_change(self, baseline: TestResult, current: TestResult) -> Comparison:
            """Analyze changes between baseline and current result"""
            similarity = self.calculate_similarity(baseline.output, current.output)
            
            # Determine change type
            if current.error:
                change_type = ChangeType.ERROR
                severity = Severity.CRITICAL
                details = f"Error occurred: {current.error}"
            elif similarity >= 0.95:
                change_type = ChangeType.NO_CHANGE
                severity = Severity.INFO
                details = "Output is virtually identical to baseline"
            else:
                # Use LLM to classify the change
                change_type, severity, details = self._llm_classify_change(
                    baseline.output, current.output, similarity
                )
            
            return Comparison(
                test_id=current.test_id,
                baseline_output=baseline.output,
                current_output=current.output,
                similarity_score=similarity,
                change_type=change_type,
                severity=severity,
                details=details,
                timestamp=datetime.now().isoformat()
            )
        
        def _llm_classify_change(self, baseline: str, current: str, 
                               similarity: float) -> tuple[ChangeType, Severity, str]:
            """Use LLM to classify the nature of the change"""
            prompt = ChatPromptTemplate.from_messages([
                ("system", """You are an expert at analyzing changes in AI outputs.
    Compare the baseline and current outputs and determine:
    1. Change type: improvement, regression, or neutral
    2. Severity: critical, high, medium, low, info
    3. Brief explanation
    
    Respond in JSON format: {{"change_type": "...", "severity": "...", "details": "..."}}"""),
                ("user", """Baseline output: {baseline}
    
    Current output: {current}
    
    Similarity score: {similarity:.2f}
    
    Analyze the change:""")
            ])
            
            try:
                chain = prompt | self.comparison_llm | StrOutputParser()
                response = chain.invoke({
                    "baseline": baseline,
                    "current": current,
                    "similarity": similarity
                })
                
                # Parse JSON response
                result = json.loads(response)
                change_type = ChangeType(result.get('change_type', 'neutral'))
                severity = Severity(result.get('severity', 'info'))
                details = result.get('details', 'No details provided')
                
            except Exception as e:
                # Fallback classification
                if similarity < 0.5:
                    change_type = ChangeType.REGRESSION
                    severity = Severity.HIGH
                elif similarity  RegressionReport:
            """Run all tests and compare against baseline"""
            print(f"\n=== Running Regression Tests (Version {version}) ===\n")
            
            if not self.baselines:
                print("No baseline found. Please establish baseline first.")
                return None
            
            comparisons = []
            passed = 0
            regressions = 0
            improvements = 0
            errors = 0
            
            for test_case in self.test_cases:
                print(f"Testing: {test_case.id}")
                
                # Run current test
                current = self.run_test(test_case)
                
                # Compare with baseline if exists
                if test_case.id in self.baselines:
                    baseline = self.baselines[test_case.id]
                    comparison = self.analyze_change(baseline, current)
                    comparisons.append(comparison)
                    
                    # Update counters
                    if comparison.change_type == ChangeType.NO_CHANGE:
                        passed += 1
                        print(f"  ✓ PASS (no change)")
                    elif comparison.change_type == ChangeType.REGRESSION:
                        regressions += 1
                        print(f"  ✗ REGRESSION ({comparison.severity.value}): {comparison.details}")
                    elif comparison.change_type == ChangeType.IMPROVEMENT:
                        improvements += 1
                        print(f"  ↑ IMPROVEMENT: {comparison.details}")
                    elif comparison.change_type == ChangeType.ERROR:
                        errors += 1
                        print(f"  ⚠ ERROR: {comparison.details}")
                    else:
                        passed += 1
                        print(f"  ~ NEUTRAL: {comparison.details}")
                else:
                    print(f"  ⚠ No baseline found for this test")
            
            # Generate summary
            total = len(self.test_cases)
            summary = self._generate_summary(passed, regressions, improvements, errors, total)
            
            report = RegressionReport(
                version=version,
                timestamp=datetime.now().isoformat(),
                total_tests=total,
                passed=passed,
                regressions=regressions,
                improvements=improvements,
                errors=errors,
                comparisons=comparisons,
                summary=summary
            )
            
            print(f"\n{summary}")
            return report
        
        def _generate_summary(self, passed: int, regressions: int, 
                             improvements: int, errors: int, total: int) -> str:
            """Generate test summary"""
            lines = [
                "=== Regression Test Summary ===",
                f"Total Tests: {total}",
                f"Passed: {passed} ({passed/total*100:.1f}%)",
                f"Regressions: {regressions} ({regressions/total*100:.1f}%)",
                f"Improvements: {improvements} ({improvements/total*100:.1f}%)",
                f"Errors: {errors} ({errors/total*100:.1f}%)"
            ]
            
            if regressions == 0 and errors == 0:
                lines.append("\n✓ All tests passed! No regressions detected.")
            elif regressions > 0:
                lines.append(f"\n⚠ WARNING: {regressions} regression(s) detected!")
            
            return "\n".join(lines)
        
        def get_regression_details(self, report: RegressionReport) -> str:
            """Get detailed regression information"""
            regressions = [c for c in report.comparisons 
                          if c.change_type == ChangeType.REGRESSION]
            
            if not regressions:
                return "No regressions found."
            
            details = ["\n=== Regression Details ===\n"]
            for i, reg in enumerate(regressions, 1):
                details.append(f"{i}. Test: {reg.test_id}")
                details.append(f"   Severity: {reg.severity.value}")
                details.append(f"   Similarity: {reg.similarity_score:.2%}")
                details.append(f"   Details: {reg.details}")
                details.append(f"   Baseline: {reg.baseline_output[:100]}...")
                details.append(f"   Current:  {reg.current_output[:100]}...")
                details.append("")
            
            return "\n".join(details)
    
    
    def demonstrate_regression_testing():
        """Demonstrate regression testing pattern"""
        print("=" * 80)
        print("REGRESSION TESTING PATTERN DEMONSTRATION")
        print("=" * 80)
        
        # Initialize tester
        tester = RegressionTester(baseline_file="test_baselines.json")
        
        # Define test cases
        test_cases = [
            TestCase(
                id="test_001",
                input="What is the capital of France?",
                category="factual"
            ),
            TestCase(
                id="test_002",
                input="Explain photosynthesis in simple terms.",
                category="explanation"
            ),
            TestCase(
                id="test_003",
                input="Write a haiku about coding.",
                category="creative"
            ),
            TestCase(
                id="test_004",
                input="Calculate 15% of 200.",
                category="calculation"
            ),
            TestCase(
                id="test_005",
                input="List three benefits of exercise.",
                category="informational"
            ),
        ]
        
        # Add test cases
        for tc in test_cases:
            tester.add_test_case(tc)
        
        # Example 1: Establish baseline
        print("\n" + "="*80)
        print("EXAMPLE 1: Establishing Baseline")
        print("="*80)
        tester.establish_baseline(version="1.0.0")
        
        # Example 2: Run regression tests (should pass)
        print("\n" + "="*80)
        print("EXAMPLE 2: Running Regression Tests (Same Model)")
        print("="*80)
        report = tester.run_regression_tests(version="1.0.1")
        
        # Example 3: Simulate model change and test
        print("\n" + "="*80)
        print("EXAMPLE 3: Testing with Different Model")
        print("="*80)
        tester_v2 = RegressionTester(baseline_file="test_baselines.json", 
                                      model_name="gpt-3.5-turbo")
        tester_v2.test_cases = test_cases
        tester_v2.load_baselines()
        report_v2 = tester_v2.run_regression_tests(version="2.0.0")
        
        # Example 4: Show regression details
        if report_v2 and report_v2.regressions > 0:
            print("\n" + "="*80)
            print("EXAMPLE 4: Regression Details")
            print("="*80)
            print(tester_v2.get_regression_details(report_v2))
        
        # Example 5: Compare specific test
        print("\n" + "="*80)
        print("EXAMPLE 5: Detailed Comparison for One Test")
        print("="*80)
        test_id = "test_001"
        if report and any(c.test_id == test_id for c in report.comparisons):
            comparison = next(c for c in report.comparisons if c.test_id == test_id)
            print(f"Test ID: {comparison.test_id}")
            print(f"Change Type: {comparison.change_type.value}")
            print(f"Severity: {comparison.severity.value}")
            print(f"Similarity: {comparison.similarity_score:.2%}")
            print(f"Details: {comparison.details}")
            print(f"\nBaseline Output:\n{comparison.baseline_output}")
            print(f"\nCurrent Output:\n{comparison.current_output}")
        
        # Summary
        print("\n" + "="*80)
        print("SUMMARY: Regression Testing Best Practices")
        print("="*80)
        print("""
    1. BASELINE MANAGEMENT:
       - Establish baselines for stable versions
       - Version baselines appropriately
       - Update baselines when intentional changes are made
       - Store baselines in version control
    
    2. TEST CASE DESIGN:
       - Cover diverse input types (factual, creative, calculation)
       - Include edge cases and error scenarios
       - Categorize tests for targeted analysis
       - Keep test cases stable and representative
    
    3. COMPARISON STRATEGIES:
       - Use similarity metrics for objective comparison
       - Employ LLM-based semantic analysis
       - Define severity thresholds
       - Track both regressions and improvements
    
    4. AUTOMATION:
       - Integrate into CI/CD pipeline
       - Run on every code/model change
       - Generate automated reports
       - Alert on critical regressions
    
    5. CONTINUOUS IMPROVEMENT:
       - Review flagged regressions manually
       - Adjust baselines for intentional improvements
       - Expand test coverage over time
       - Track regression trends
    
    Benefits:
    ✓ Early detection of unintended changes
    ✓ Confidence in updates and refactoring
    ✓ Historical performance tracking
    ✓ Automated quality assurance
    ✓ Reduced manual testing overhead
    ✓ Data-driven decision making
    
    Challenges:
    - Baseline maintenance for creative outputs
    - False positives from legitimate variations
    - Storage and comparison overhead
    - Defining acceptable change thresholds
        """)
    
    
    if __name__ == "__main__":
        demonstrate_regression_testing()
    
        

    Agentic Design Pattern: Regression Testing

    Pattern 155: Regression Testing – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Regression Testing pattern implementation.


    Pattern Description

    Pattern 155: Regression Testing

    Description: Automated regression testing detects unintended changes in agent behavior when code, models, or prompts are updated. Maintains baselines, compares outputs, and flags regressions automatically.

    Components: – Baseline storage and versioning – Test case management – Output comparison – Regression detection – Change analysis – Historical tracking

    Use Cases: – Model version updates – Prompt engineering iterations – Code refactoring validation – CI/CD integration – Quality assurance

    Benefits: – Early regression detection – Automated quality checks – Historical performance tracking – Confidence in updates

    Trade-offs: – Baseline maintenance overhead – Storage requirements – Comparison complexity for creative tasks – False positive management

    LangChain Implementation: Uses ChatOpenAI for testing, JSON storage for baselines, similarity metrics for comparison, and automated detection


    File Statistics

    • Total Lines: 564
    • Classes: 7
    • Functions: 1
    • Constants: 0

    Module Documentation

    """
    Pattern 155: Regression Testing
    
    Description:
        Automated regression testing detects unintended changes in agent behavior
        when code, models, or prompts are updated. Maintains baselines, compares
        outputs, and flags regressions automatically.
    
    Components:
        - Baseline storage and versioning
        - Test case management
        - Output comparison
        - Regression detection
        - Change analysis
        - Historical tracking
    
    Use Cases:
        - Model version updates
        - Prompt engineering iterations
        - Code refactoring validation
        - CI/CD integration
        - Quality assurance
    
    Benefits:
        - Early regression detection
        - Automated quality checks
        - Historical performance tracking
        - Confidence in updates
    
    Trade-offs:
        - Baseline maintenance overhead
        - Storage requirements
        - Comparison complexity for creative tasks
        - False positive management
    
    LangChain Implementation:
        Uses ChatOpenAI for testing, JSON storage for baselines,
        similarity metrics for comparison, and automated detection
    """
    

    Purpose: This module implements the Regression Testing 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 time
    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 – 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 – time: Time-related functions – 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 difflib.SequenceMatcher
    from dotenv.load_dotenv
    

    Purpose: Additional dependencies for specialized functionality.


    Classes

    Class: ChangeType

    Lines 57-63

    Inherits from: Enum

    Purpose:

    Types of changes detected
    

    Class: Severity

    Lines 66-72

    Inherits from: Enum

    Purpose:

    Severity levels for regressions
    

    Class: TestCase

    Lines 76-82

    Purpose:

    Single test case
    

    Class: TestResult

    Lines 86-94

    Purpose:

    Result of a single test execution
    

    Class: Comparison

    Lines 98-107

    Purpose:

    Comparison between baseline and current result
    

    Class: RegressionReport

    Lines 111-121

    Purpose:

    Comprehensive regression test report
    

    Class: RegressionTester

    Lines 124-422

    Purpose:

    Manages regression testing for agent outputs
    

    Methods:

    __init__(self, baseline_file, model_name)

    Lines 127-143

    Purpose:

    Initialize regression tester
    
    Args:
        baseline_file: Path to baseline storage
        model_name: LLM model to use for testing
    

    Parameters: – baseline_file: Baseline file – model_name: Language model instance

    load_baselines(self)

    Lines 145-160

    Purpose:

    Load baseline results from file
    

    Parameters:

    save_baselines(self)

    Lines 162-181

    Purpose:

    Save baseline results to file
    

    Parameters:

    add_test_case(self, test_case)

    Lines 183-185

    Purpose:

    Add a test case
    

    Parameters: – test_case: Test case

    run_test(self, test_case)

    Lines 187-219

    Purpose:

    Execute a single test case
    

    Parameters: – test_case: Test case

    Returns: TestResult

    calculate_similarity(self, text1, text2)

    Lines 221-224

    Purpose:

    Calculate similarity between two texts
    

    Parameters: – text1: Text1 – text2: Text2

    Returns: float

    analyze_change(self, baseline, current)

    Lines 226-254

    Purpose:

    Analyze changes between baseline and current result
    

    Parameters: – baseline: Baseline – current: Current

    Returns: Comparison

    _llm_classify_change(self, baseline, current, similarity)

    Lines 256-303

    Purpose:

    Use LLM to classify the nature of the change
    

    Parameters: – baseline: Baseline – current: Current – similarity: Similarity

    Returns: <ast.Subscript object at 0x000001A06D7016D0>

    establish_baseline(self, version)

    Lines 305-319

    Purpose:

    Run tests and establish baseline
    

    Parameters: – version: Version

    run_regression_tests(self, version)

    Lines 321-383

    Purpose:

    Run all tests and compare against baseline
    

    Parameters: – version: Version

    Returns: RegressionReport

    _generate_summary(self, passed, regressions, improvements, errors, total)

    Lines 385-402

    Purpose:

    Generate test summary
    

    Parameters: – passed: Passed – regressions: Regressions – improvements: Improvements – errors: Errors – total: Total

    Returns: str

    get_regression_details(self, report)

    Lines 404-422

    Purpose:

    Get detailed regression information
    

    Parameters: – report: Report

    Returns: str


    Functions

    demonstrate_regression_testing()

    Lines 425-559

    Purpose:

    Demonstrate regression testing pattern
    

    Key Concepts Summary

    Pattern Overview

    The Regression Testing 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 154
    • Pattern 156

    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 Regression Testing 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: 155_regression_testing.py
    Total Lines: 564
    Pattern Category: AI Agent Design Pattern #155


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Canary Deployment

    
        """
    Pattern 154: Canary Deployment
    
    Description:
        Implements gradual rollout of new agent versions by routing a small
        percentage of traffic to the canary version, monitoring metrics, and
        automatically rolling back if issues are detected.
    
    Components:
        - Traffic Splitter: Routes requests between versions
        - Health Monitor: Tracks canary metrics
        - Rollback Controller: Automatic rollback on issues
        - Metrics Collector: Performance and error tracking
        - Gradual Ramp: Increases traffic percentage over time
    
    Use Cases:
        - Safe production deployments
        - Gradual feature rollout
        - Risk mitigation for changes
        - A/B testing with safety
        - Automated deployment validation
    
    Benefits:
        - Minimal blast radius
        - Automatic rollback
        - Real user validation
        - Gradual confidence building
        - Production safety
    
    Trade-offs:
        - Deployment complexity
        - Monitoring overhead
        - Slower full rollout
        - Dual version maintenance
        - Metrics interpretation
    
    LangChain Implementation:
        Uses traffic routing with health monitoring and automatic rollback.
    """
    
    import os
    import time
    import random
    from typing import List, Dict, Any, Optional
    from dataclasses import dataclass, field
    from enum import Enum
    from datetime import datetime
    from collections import defaultdict
    from dotenv import load_dotenv
    
    from langchain_openai import ChatOpenAI
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser
    
    load_dotenv()
    
    
    class DeploymentStatus(Enum):
        """Status of canary deployment"""
        STARTING = "starting"
        HEALTHY = "healthy"
        WARNING = "warning"
        ROLLING_BACK = "rolling_back"
        ROLLED_BACK = "rolled_back"
        PROMOTED = "promoted"
    
    
    @dataclass
    class CanaryMetrics:
        """Metrics for canary and production versions"""
        # Request counts
        production_requests: int = 0
        canary_requests: int = 0
        
        # Success rates
        production_successes: int = 0
        canary_successes: int = 0
        production_errors: int = 0
        canary_errors: int = 0
        
        # Latency
        production_latency_sum: float = 0.0
        canary_latency_sum: float = 0.0
        
        def get_production_error_rate(self) -> float:
            """Calculate production error rate"""
            if self.production_requests == 0:
                return 0.0
            return self.production_errors / self.production_requests
        
        def get_canary_error_rate(self) -> float:
            """Calculate canary error rate"""
            if self.canary_requests == 0:
                return 0.0
            return self.canary_errors / self.canary_requests
        
        def get_production_avg_latency(self) -> float:
            """Calculate average production latency"""
            if self.production_requests == 0:
                return 0.0
            return self.production_latency_sum / self.production_requests
        
        def get_canary_avg_latency(self) -> float:
            """Calculate average canary latency"""
            if self.canary_requests == 0:
                return 0.0
            return self.canary_latency_sum / self.canary_requests
    
    
    @dataclass
    class CanaryConfig:
        """Configuration for canary deployment"""
        initial_traffic_percent: float = 5.0  # Start with 5%
        max_traffic_percent: float = 50.0  # Max before promotion
        ramp_step: float = 5.0  # Increase by 5% each step
        ramp_interval_seconds: float = 60.0  # Wait 60s between steps
        
        # Health thresholds
        max_error_rate_increase: float = 0.05  # Max 5% error rate increase
        max_latency_increase: float = 0.20  # Max 20% latency increase
        min_requests_for_decision: int = 10  # Min requests before decisions
        
        # Rollback settings
        auto_rollback: bool = True
        rollback_on_error_spike: bool = True
    
    
    class CanaryDeployment:
        """
        Manages canary deployment with gradual traffic routing and
        automatic rollback on issues.
        """
        
        def __init__(self, config: Optional[CanaryConfig] = None):
            """Initialize canary deployment"""
            self.config = config or CanaryConfig()
            
            # Production version (stable)
            self.production_llm = ChatOpenAI(
                model="gpt-3.5-turbo",
                temperature=0.7
            )
            
            # Canary version (new version being tested)
            self.canary_llm = ChatOpenAI(
                model="gpt-3.5-turbo",
                temperature=0.8  # Different config for canary
            )
            
            # State
            self.current_traffic_percent = self.config.initial_traffic_percent
            self.status = DeploymentStatus.STARTING
            self.metrics = CanaryMetrics()
            self.deployment_start_time = datetime.now()
            self.last_ramp_time = time.time()
            
            # Prompts
            self.prompt = ChatPromptTemplate.from_messages([
                ("system", "You are a helpful AI assistant."),
                ("user", "{query}")
            ])
        
        def route_request(self, query: str) -> Dict[str, Any]:
            """
            Route request to either production or canary based on traffic split
            
            Args:
                query: User query
                
            Returns:
                Response with version and metrics
            """
            # Decide routing
            use_canary = random.random() * 100 < self.current_traffic_percent
            
            start_time = time.time()
            
            try:
                if use_canary:
                    # Route to canary
                    chain = self.prompt | self.canary_llm | StrOutputParser()
                    result = chain.invoke({"query": query})
                    version = "canary"
                    
                    # Update canary metrics
                    self.metrics.canary_requests += 1
                    self.metrics.canary_successes += 1
                    latency = time.time() - start_time
                    self.metrics.canary_latency_sum += latency
                    
                else:
                    # Route to production
                    chain = self.prompt | self.production_llm | StrOutputParser()
                    result = chain.invoke({"query": query})
                    version = "production"
                    
                    # Update production metrics
                    self.metrics.production_requests += 1
                    self.metrics.production_successes += 1
                    latency = time.time() - start_time
                    self.metrics.production_latency_sum += latency
                
                # Check health after each request
                self._check_health()
                
                # Consider ramping up traffic
                self._consider_ramp_up()
                
                return {
                    "success": True,
                    "result": result,
                    "version": version,
                    "latency": latency,
                    "canary_traffic_percent": self.current_traffic_percent,
                    "status": self.status.value
                }
                
            except Exception as e:
                # Track error
                latency = time.time() - start_time
                
                if use_canary:
                    self.metrics.canary_requests += 1
                    self.metrics.canary_errors += 1
                    self.metrics.canary_latency_sum += latency
                    version = "canary"
                else:
                    self.metrics.production_requests += 1
                    self.metrics.production_errors += 1
                    self.metrics.production_latency_sum += latency
                    version = "production"
                
                # Check if we should rollback
                self._check_health()
                
                return {
                    "success": False,
                    "error": str(e),
                    "version": version,
                    "latency": latency,
                    "canary_traffic_percent": self.current_traffic_percent,
                    "status": self.status.value
                }
        
        def _check_health(self):
            """Check canary health and trigger rollback if needed"""
            if self.status == DeploymentStatus.ROLLED_BACK:
                return  # Already rolled back
            
            if self.metrics.canary_requests  0:
                error_rate_increase = (canary_error_rate - prod_error_rate) / prod_error_rate
            else:
                error_rate_increase = canary_error_rate
            
            # Check latency
            if prod_latency > 0:
                latency_increase = (canary_latency - prod_latency) / prod_latency
            else:
                latency_increase = 0.0
            
            # Determine health
            issues = []
            
            if error_rate_increase > self.config.max_error_rate_increase:
                issues.append(f"Error rate {error_rate_increase*100:.1f}% higher")
            
            if latency_increase > self.config.max_latency_increase:
                issues.append(f"Latency {latency_increase*100:.1f}% higher")
            
            if issues:
                if self.config.auto_rollback:
                    print(f"\n🚨 ROLLBACK TRIGGERED: {', '.join(issues)}")
                    self._rollback()
                else:
                    self.status = DeploymentStatus.WARNING
                    print(f"\n⚠️  WARNING: {', '.join(issues)}")
            elif self.status != DeploymentStatus.PROMOTED:
                self.status = DeploymentStatus.HEALTHY
        
        def _consider_ramp_up(self):
            """Consider increasing traffic to canary"""
            if self.status != DeploymentStatus.HEALTHY:
                return  # Only ramp if healthy
            
            if self.current_traffic_percent >= self.config.max_traffic_percent:
                return  # Already at max
            
            # Check if enough time has passed
            time_since_last_ramp = time.time() - self.last_ramp_time
            if time_since_last_ramp = self.config.max_traffic_percent:
                print(f"\n✅ CANARY READY FOR PROMOTION (reached {self.config.max_traffic_percent}% traffic)")
        
        def _rollback(self):
            """Rollback to production version"""
            self.status = DeploymentStatus.ROLLING_BACK
            self.current_traffic_percent = 0.0
            print(f"\n⏪ Rolling back to production version...")
            self.status = DeploymentStatus.ROLLED_BACK
            print(f"✅ Rollback complete - 100% traffic to production")
        
        def promote_canary(self):
            """Promote canary to production"""
            if self.status == DeploymentStatus.ROLLED_BACK:
                print("❌ Cannot promote - deployment was rolled back")
                return False
            
            print(f"\n🎉 PROMOTING CANARY TO PRODUCTION")
            self.current_traffic_percent = 100.0
            self.status = DeploymentStatus.PROMOTED
            return True
        
        def get_status_report(self) -> str:
            """Generate status report"""
            m = self.metrics
            
            report = [
                "\n" + "="*70,
                "CANARY DEPLOYMENT STATUS",
                "="*70,
                f"\nStatus: {self.status.value.upper()}",
                f"Canary Traffic: {self.current_traffic_percent:.1f}%",
                f"Deployment Duration: {(datetime.now() - self.deployment_start_time).total_seconds():.0f}s",
                "\nProduction Metrics:",
                f"  Requests: {m.production_requests}",
                f"  Success Rate: {(m.production_successes/m.production_requests*100 if m.production_requests > 0 else 0):.1f}%",
                f"  Error Rate: {m.get_production_error_rate()*100:.1f}%",
                f"  Avg Latency: {m.get_production_avg_latency():.3f}s",
                "\nCanary Metrics:",
                f"  Requests: {m.canary_requests}",
                f"  Success Rate: {(m.canary_successes/m.canary_requests*100 if m.canary_requests > 0 else 0):.1f}%",
                f"  Error Rate: {m.get_canary_error_rate()*100:.1f}%",
                f"  Avg Latency: {m.get_canary_avg_latency():.3f}s",
            ]
            
            # Comparison
            if m.canary_requests >= self.config.min_requests_for_decision:
                error_diff = (m.get_canary_error_rate() - m.get_production_error_rate()) * 100
                latency_diff_pct = ((m.get_canary_avg_latency() - m.get_production_avg_latency()) 
                                   / m.get_production_avg_latency() * 100) if m.get_production_avg_latency() > 0 else 0
                
                report.extend([
                    "\nComparison (Canary vs Production):",
                    f"  Error Rate Diff: {error_diff:+.1f}%",
                    f"  Latency Diff: {latency_diff_pct:+.1f}%"
                ])
            
            report.append("="*70)
            return "\n".join(report)
    
    
    def demonstrate_canary_deployment():
        """Demonstrate canary deployment pattern"""
        print("="*70)
        print("Pattern 154: Canary Deployment")
        print("="*70)
        
        # Configure canary deployment
        config = CanaryConfig(
            initial_traffic_percent=10.0,
            max_traffic_percent=50.0,
            ramp_step=10.0,
            ramp_interval_seconds=5.0,  # Short for demo
            max_error_rate_increase=0.10,
            max_latency_increase=0.30,
            min_requests_for_decision=5,
            auto_rollback=True
        )
        
        deployment = CanaryDeployment(config)
        
        print(f"\n🚀 Starting canary deployment")
        print(f"   Initial traffic: {config.initial_traffic_percent}%")
        print(f"   Ramp step: {config.ramp_step}%")
        print(f"   Target: {config.max_traffic_percent}%")
        
        # Simulate requests
        test_queries = [
            "What is Python?",
            "Explain machine learning",
            "How does AI work?",
            "What is cloud computing?",
            "Tell me about databases",
        ] * 10  # 50 total requests
        
        print(f"\n{'='*70}")
        print("SIMULATING PRODUCTION TRAFFIC")
        print(f"{'='*70}")
        
        results = []
        for i, query in enumerate(test_queries, 1):
            result = deployment.route_request(query)
            results.append(result)
            
            if i % 10 == 0:
                print(f"\n📊 After {i} requests:")
                print(deployment.get_status_report())
                
                # Small delay to allow ramping
                time.sleep(1)
        
        # Final status
        print(f"\n{'='*70}")
        print("FINAL STATUS")
        print(f"{'='*70}")
        print(deployment.get_status_report())
        
        # Show routing distribution
        production_count = sum(1 for r in results if r.get('version') == 'production')
        canary_count = sum(1 for r in results if r.get('version') == 'canary')
        
        print(f"\n📊 Traffic Distribution:")
        print(f"   Production: {production_count} requests ({production_count/len(results)*100:.1f}%)")
        print(f"   Canary: {canary_count} requests ({canary_count/len(results)*100:.1f}%)")
        
        # Promotion decision
        if deployment.status == DeploymentStatus.HEALTHY:
            print(f"\n✅ Canary is healthy and ready for promotion!")
            deployment.promote_canary()
        elif deployment.status == DeploymentStatus.ROLLED_BACK:
            print(f"\n❌ Canary was rolled back due to issues")
        
        # Summary
        print("\n" + "="*70)
        print("SUMMARY")
        print("="*70)
        print("""
    Canary Deployment Pattern provides:
    
    1. Gradual Rollout:
       - Start with small traffic percentage
       - Incrementally increase over time
       - Monitor at each step
       - Automatic promotion when ready
    
    2. Safety Features:
       - Continuous health monitoring
       - Automatic rollback on issues
       - Configurable thresholds
       - Minimal blast radius
    
    3. Metrics Tracking:
       - Error rates comparison
       - Latency comparison
       - Success rates
       - Traffic distribution
    
    4. Deployment Control:
       - Traffic percentage control
       - Ramp-up automation
       - Manual promotion option
       - Emergency rollback
    
    5. Production Safety:
       - Real user validation
       - Low-risk testing
       - Quick issue detection
       - Automatic recovery
    
    This pattern is essential for safely deploying changes to
    production AI systems with minimal risk and automatic safeguards.
        """)
    
    
    if __name__ == "__main__":
        demonstrate_canary_deployment()
    
        

    Agentic Design Pattern: Canary Deployment

    Pattern 154: Canary Deployment – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Canary Deployment pattern implementation.


    Pattern Description

    Pattern 154: Canary Deployment

    Description: Implements gradual rollout of new agent versions by routing a small percentage of traffic to the canary version, monitoring metrics, and automatically rolling back if issues are detected.

    Components: – Traffic Splitter: Routes requests between versions – Health Monitor: Tracks canary metrics – Rollback Controller: Automatic rollback on issues – Metrics Collector: Performance and error tracking – Gradual Ramp: Increases traffic percentage over time

    Use Cases: – Safe production deployments – Gradual feature rollout – Risk mitigation for changes – A/B testing with safety – Automated deployment validation

    Benefits: – Minimal blast radius – Automatic rollback – Real user validation – Gradual confidence building – Production safety

    Trade-offs: – Deployment complexity – Monitoring overhead – Slower full rollout – Dual version maintenance – Metrics interpretation

    LangChain Implementation: Uses traffic routing with health monitoring and automatic rollback.


    File Statistics

    • Total Lines: 491
    • Classes: 4
    • Functions: 1
    • Constants: 0

    Module Documentation

    """
    Pattern 154: Canary Deployment
    
    Description:
        Implements gradual rollout of new agent versions by routing a small
        percentage of traffic to the canary version, monitoring metrics, and
        automatically rolling back if issues are detected.
    
    Components:
        - Traffic Splitter: Routes requests between versions
        - Health Monitor: Tracks canary metrics
        - Rollback Controller: Automatic rollback on issues
        - Metrics Collector: Performance and error tracking
        - Gradual Ramp: Increases traffic percentage over time
    
    Use Cases:
        - Safe production deployments
        - Gradual feature rollout
        - Risk mitigation for changes
        - A/B testing with safety
        - Automated deployment validation
    
    Benefits:
        - Minimal blast radius
        - Automatic rollback
        - Real user validation
        - Gradual confidence building
        - Production safety
    
    Trade-offs:
        - Deployment complexity
        - Monitoring overhead
        - Slower full rollout
        - Dual version maintenance
        - Metrics interpretation
    
    LangChain Implementation:
        Uses traffic routing with health monitoring and automatic rollback.
    """
    

    Purpose: This module implements the Canary Deployment 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 time
    import typing.Any
    import typing.Dict
    import typing.List
    import typing.Optional
    

    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 – time: Time-related functions – 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: DeploymentStatus

    Lines 58-65

    Inherits from: Enum

    Purpose:

    Status of canary deployment
    

    Class: CanaryMetrics

    Lines 69-107

    Purpose:

    Metrics for canary and production versions
    

    Methods:

    get_production_error_rate(self)

    Lines 85-89

    Purpose:

    Calculate production error rate
    

    Parameters:

    Returns: float

    get_canary_error_rate(self)

    Lines 91-95

    Purpose:

    Calculate canary error rate
    

    Parameters:

    Returns: float

    get_production_avg_latency(self)

    Lines 97-101

    Purpose:

    Calculate average production latency
    

    Parameters:

    Returns: float

    get_canary_avg_latency(self)

    Lines 103-107

    Purpose:

    Calculate average canary latency
    

    Parameters:

    Returns: float


    Class: CanaryConfig

    Lines 111-125

    Purpose:

    Configuration for canary deployment
    

    Class: CanaryDeployment

    Lines 128-373

    Purpose:

    Manages canary deployment with gradual traffic routing and
    automatic rollback on issues.
    

    Methods:

    __init__(self, config)

    Lines 134-161

    Purpose:

    Initialize canary deployment
    

    Parameters: – config: Config

    route_request(self, query)

    Lines 163-243

    Purpose:

    Route request to either production or canary based on traffic split
    
    Args:
        query: User query
    
    Returns:
        Response with version and metrics
    

    Parameters: – query: Input query or question

    Returns: <ast.Subscript object at 0x000001A06D60EAD0>

    _check_health(self)

    Lines 245-289

    Purpose:

    Check canary health and trigger rollback if needed
    

    Parameters:

    _consider_ramp_up(self)

    Lines 291-316

    Purpose:

    Consider increasing traffic to canary
    

    Parameters:

    _rollback(self)

    Lines 318-324

    Purpose:

    Rollback to production version
    

    Parameters:

    promote_canary(self)

    Lines 326-335

    Purpose:

    Promote canary to production
    

    Parameters:

    get_status_report(self)

    Lines 337-373

    Purpose:

    Generate status report
    

    Parameters:

    Returns: str


    Functions

    demonstrate_canary_deployment()

    Lines 376-486

    Purpose:

    Demonstrate canary deployment pattern
    

    Key Concepts Summary

    Pattern Overview

    The Canary Deployment 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 153
    • Pattern 155

    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 Canary Deployment 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: 154_canary_deployment.py
    Total Lines: 491
    Pattern Category: AI Agent Design Pattern #154


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Shadow Mode

    
        """
    Pattern 153: Shadow Mode Testing
    
    Description:
        Runs new agent versions alongside production without affecting users.
        Captures both production and shadow responses, compares results, and
        identifies differences for validation before full deployment.
    
    Components:
        - Shadow Executor: Runs shadow version in parallel
        - Response Comparator: Compares production vs shadow outputs
        - Difference Analyzer: Identifies and categorizes differences
        - Metrics Collector: Tracks shadow performance
        - Rollback Monitor: Detects degradation signals
    
    Use Cases:
        - Testing new model versions safely
        - Validating prompt changes
        - A/B testing without user impact
        - Gradual feature rollout
        - Performance regression detection
    
    Benefits:
        - Zero user impact during testing
        - Real production traffic testing
        - Early problem detection
        - Confidence before full deployment
        - Comprehensive comparison data
    
    Trade-offs:
        - Doubled computational cost
        - Infrastructure complexity
        - Storage for dual responses
        - Latency considerations
        - Analysis overhead
    
    LangChain Implementation:
        Uses parallel execution with ChatOpenAI, response comparison,
        and comprehensive metrics tracking.
    """
    
    import os
    import time
    import asyncio
    from typing import List, Dict, Any, Optional, Tuple
    from dataclasses import dataclass, field
    from enum import Enum
    from datetime import datetime
    from collections import defaultdict
    from concurrent.futures import ThreadPoolExecutor, as_completed
    from dotenv import load_dotenv
    
    from langchain_openai import ChatOpenAI
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser
    
    load_dotenv()
    
    
    class DifferenceType(Enum):
        """Types of differences between responses"""
        IDENTICAL = "identical"
        MINOR = "minor"  # Small wording differences
        MODERATE = "moderate"  # Different approach, similar meaning
        MAJOR = "major"  # Significantly different
        ERROR = "error"  # One succeeded, one failed
    
    
    class HealthStatus(Enum):
        """Health status of shadow version"""
        HEALTHY = "healthy"
        WARNING = "warning"
        DEGRADED = "degraded"
        FAILING = "failing"
    
    
    @dataclass
    class ShadowResponse:
        """Response from shadow execution"""
        production_response: str
        shadow_response: str
        production_time: float
        shadow_time: float
        production_error: Optional[str] = None
        shadow_error: Optional[str] = None
        difference_type: Optional[DifferenceType] = None
        similarity_score: float = 0.0
        timestamp: datetime = field(default_factory=datetime.now)
    
    
    @dataclass
    class ShadowMetrics:
        """Metrics for shadow testing"""
        total_requests: int = 0
        identical_responses: int = 0
        minor_differences: int = 0
        moderate_differences: int = 0
        major_differences: int = 0
        production_errors: int = 0
        shadow_errors: int = 0
        avg_production_latency: float = 0.0
        avg_shadow_latency: float = 0.0
        shadow_slower_by: float = 0.0  # Percentage
        
        def get_difference_rate(self, diff_type: DifferenceType) -> float:
            """Get rate of specific difference type"""
            if self.total_requests == 0:
                return 0.0
            
            count_map = {
                DifferenceType.IDENTICAL: self.identical_responses,
                DifferenceType.MINOR: self.minor_differences,
                DifferenceType.MODERATE: self.moderate_differences,
                DifferenceType.MAJOR: self.major_differences
            }
            
            return count_map.get(diff_type, 0) / self.total_requests
        
        def get_health_status(self) -> HealthStatus:
            """Determine overall health status"""
            if self.total_requests  0.3 or major_rate > 0.5:
                return HealthStatus.FAILING
            elif error_rate > 0.1 or major_rate > 0.3:
                return HealthStatus.DEGRADED
            elif error_rate > 0.05 or major_rate > 0.15:
                return HealthStatus.WARNING
            else:
                return HealthStatus.HEALTHY
    
    
    class ShadowModeTester:
        """
        Agent that runs shadow testing by executing production and shadow
        versions in parallel, comparing results without affecting users.
        """
        
        def __init__(self, production_model: str = "gpt-3.5-turbo",
                     shadow_model: str = "gpt-3.5-turbo",
                     production_temp: float = 0.7,
                     shadow_temp: float = 0.7):
            """Initialize shadow mode tester"""
            # Production agent
            self.production_llm = ChatOpenAI(
                model=production_model,
                temperature=production_temp
            )
            
            # Shadow agent (different version or configuration)
            self.shadow_llm = ChatOpenAI(
                model=shadow_model,
                temperature=shadow_temp
            )
            
            # Metrics tracking
            self.metrics = ShadowMetrics()
            self.responses: List[ShadowResponse] = []
            
            # Difference categorization LLM
            self.analyzer_llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.0)
            
            # Prompts
            self.production_prompt = ChatPromptTemplate.from_messages([
                ("system", "You are a helpful AI assistant. Provide accurate, clear responses."),
                ("user", "{query}")
            ])
            
            self.shadow_prompt = ChatPromptTemplate.from_messages([
                ("system", "You are a helpful AI assistant. Provide accurate, clear responses."),
                ("user", "{query}")
            ])
            
            self.comparison_prompt = ChatPromptTemplate.from_messages([
                ("system", """Compare two AI responses and categorize the difference.
                
    Categories:
    - identical: Same or nearly identical responses
    - minor: Small wording differences, same meaning
    - moderate: Different approach but similar quality
    - major: Significantly different responses or quality
    
    Response 1 (Production):
    {response1}
    
    Response 2 (Shadow):
    {response2}
    
    Return only the category name."""),
                ("user", "Categorize the difference between these responses.")
            ])
        
        def execute_shadow_request(self, query: str) -> ShadowResponse:
            """
            Execute query on both production and shadow versions
            
            Args:
                query: User query to execute
                
            Returns:
                ShadowResponse with both results and comparison
            """
            # Execute in parallel using threads
            with ThreadPoolExecutor(max_workers=2) as executor:
                # Submit both requests
                prod_future = executor.submit(self._execute_production, query)
                shadow_future = executor.submit(self._execute_shadow, query)
                
                # Wait for both to complete
                prod_result, prod_time, prod_error = prod_future.result()
                shadow_result, shadow_time, shadow_error = shadow_future.result()
            
            # Compare responses
            if prod_error is None and shadow_error is None:
                diff_type, similarity = self._compare_responses(prod_result, shadow_result)
            else:
                diff_type = DifferenceType.ERROR
                similarity = 0.0
            
            # Create response object
            response = ShadowResponse(
                production_response=prod_result,
                shadow_response=shadow_result,
                production_time=prod_time,
                shadow_time=shadow_time,
                production_error=prod_error,
                shadow_error=shadow_error,
                difference_type=diff_type,
                similarity_score=similarity
            )
            
            # Update metrics
            self._update_metrics(response)
            
            # Store response
            self.responses.append(response)
            
            # Return production result to user (shadow is hidden)
            return response
        
        def _execute_production(self, query: str) -> Tuple[str, float, Optional[str]]:
            """Execute on production system"""
            start_time = time.time()
            
            try:
                chain = self.production_prompt | self.production_llm | StrOutputParser()
                result = chain.invoke({"query": query})
                execution_time = time.time() - start_time
                return result, execution_time, None
                
            except Exception as e:
                execution_time = time.time() - start_time
                return "", execution_time, str(e)
        
        def _execute_shadow(self, query: str) -> Tuple[str, float, Optional[str]]:
            """Execute on shadow system"""
            start_time = time.time()
            
            try:
                chain = self.shadow_prompt | self.shadow_llm | StrOutputParser()
                result = chain.invoke({"query": query})
                execution_time = time.time() - start_time
                return result, execution_time, None
                
            except Exception as e:
                execution_time = time.time() - start_time
                return "", execution_time, str(e)
        
        def _compare_responses(self, prod_response: str, 
                              shadow_response: str) -> Tuple[DifferenceType, float]:
            """Compare two responses and categorize difference"""
            # Quick check for identical
            if prod_response.strip() == shadow_response.strip():
                return DifferenceType.IDENTICAL, 1.0
            
            # Calculate basic similarity
            similarity = self._calculate_similarity(prod_response, shadow_response)
            
            # Use LLM for categorization
            try:
                chain = self.comparison_prompt | self.analyzer_llm | StrOutputParser()
                category_str = chain.invoke({
                    "response1": prod_response[:500],  # Truncate for efficiency
                    "response2": shadow_response[:500]
                }).strip().lower()
                
                # Map to enum
                category_map = {
                    "identical": DifferenceType.IDENTICAL,
                    "minor": DifferenceType.MINOR,
                    "moderate": DifferenceType.MODERATE,
                    "major": DifferenceType.MAJOR
                }
                
                diff_type = category_map.get(category_str, DifferenceType.MODERATE)
                
            except Exception as e:
                # Fallback to similarity-based categorization
                if similarity > 0.9:
                    diff_type = DifferenceType.MINOR
                elif similarity > 0.7:
                    diff_type = DifferenceType.MODERATE
                else:
                    diff_type = DifferenceType.MAJOR
            
            return diff_type, similarity
        
        def _calculate_similarity(self, text1: str, text2: str) -> float:
            """Calculate simple word-overlap similarity"""
            words1 = set(text1.lower().split())
            words2 = set(text2.lower().split())
            
            if not words1 or not words2:
                return 0.0
            
            intersection = words1.intersection(words2)
            union = words1.union(words2)
            
            return len(intersection) / len(union) if union else 0.0
        
        def _update_metrics(self, response: ShadowResponse):
            """Update metrics based on response"""
            self.metrics.total_requests += 1
            
            # Count difference types
            if response.difference_type == DifferenceType.IDENTICAL:
                self.metrics.identical_responses += 1
            elif response.difference_type == DifferenceType.MINOR:
                self.metrics.minor_differences += 1
            elif response.difference_type == DifferenceType.MODERATE:
                self.metrics.moderate_differences += 1
            elif response.difference_type == DifferenceType.MAJOR:
                self.metrics.major_differences += 1
            
            # Count errors
            if response.production_error:
                self.metrics.production_errors += 1
            if response.shadow_error:
                self.metrics.shadow_errors += 1
            
            # Update latency metrics (running average)
            n = self.metrics.total_requests
            self.metrics.avg_production_latency = (
                (self.metrics.avg_production_latency * (n - 1) + response.production_time) / n
            )
            self.metrics.avg_shadow_latency = (
                (self.metrics.avg_shadow_latency * (n - 1) + response.shadow_time) / n
            )
            
            # Calculate relative performance
            if self.metrics.avg_production_latency > 0:
                self.metrics.shadow_slower_by = (
                    (self.metrics.avg_shadow_latency - self.metrics.avg_production_latency) 
                    / self.metrics.avg_production_latency * 100
                )
        
        def batch_test(self, queries: List[str], 
                       show_progress: bool = True) -> List[ShadowResponse]:
            """
            Test multiple queries in shadow mode
            
            Args:
                queries: List of queries to test
                show_progress: Whether to show progress
                
            Returns:
                List of shadow responses
            """
            print(f"\n🔄 Starting shadow mode testing for {len(queries)} queries...")
            
            results = []
            for i, query in enumerate(queries, 1):
                if show_progress:
                    print(f"   Processing {i}/{len(queries)}: {query[:50]}...")
                
                response = self.execute_shadow_request(query)
                results.append(response)
                
                # Show inline status
                if show_progress:
                    status = "✓" if response.difference_type in [DifferenceType.IDENTICAL, DifferenceType.MINOR] else "⚠"
                    print(f"   {status} Difference: {response.difference_type.value}, Similarity: {response.similarity_score:.2f}")
            
            return results
        
        def get_health_status(self) -> HealthStatus:
            """Get current health status of shadow version"""
            return self.metrics.get_health_status()
        
        def get_divergent_examples(self, min_difference: DifferenceType = DifferenceType.MODERATE,
                                   limit: int = 10) -> List[ShadowResponse]:
            """Get examples where responses diverged significantly"""
            divergent = [
                r for r in self.responses
                if r.difference_type in [DifferenceType.MODERATE, DifferenceType.MAJOR]
            ]
            
            # Sort by similarity (lowest first = most divergent)
            divergent.sort(key=lambda r: r.similarity_score)
            
            return divergent[:limit]
        
        def generate_report(self) -> str:
            """Generate comprehensive shadow testing report"""
            metrics = self.metrics
            health = self.get_health_status()
            
            report = [
                "\n" + "="*70,
                "SHADOW MODE TESTING REPORT",
                "="*70,
                f"\nTotal Requests: {metrics.total_requests}",
                f"Health Status: {health.value.upper()} {'🟢' if health == HealthStatus.HEALTHY else '🟡' if health == HealthStatus.WARNING else '🟠' if health == HealthStatus.DEGRADED else '🔴'}",
                "\nResponse Comparison:",
                f"  Identical: {metrics.identical_responses} ({metrics.get_difference_rate(DifferenceType.IDENTICAL)*100:.1f}%)",
                f"  Minor Differences: {metrics.minor_differences} ({metrics.get_difference_rate(DifferenceType.MINOR)*100:.1f}%)",
                f"  Moderate Differences: {metrics.moderate_differences} ({metrics.get_difference_rate(DifferenceType.MODERATE)*100:.1f}%)",
                f"  Major Differences: {metrics.major_differences} ({metrics.get_difference_rate(DifferenceType.MAJOR)*100:.1f}%)",
                "\nError Rates:",
                f"  Production Errors: {metrics.production_errors} ({metrics.production_errors/metrics.total_requests*100:.1f}%)",
                f"  Shadow Errors: {metrics.shadow_errors} ({metrics.shadow_errors/metrics.total_requests*100:.1f}%)",
                "\nPerformance:",
                f"  Avg Production Latency: {metrics.avg_production_latency:.3f}s",
                f"  Avg Shadow Latency: {metrics.avg_shadow_latency:.3f}s",
                f"  Shadow Performance: {metrics.shadow_slower_by:+.1f}% {'(slower)' if metrics.shadow_slower_by > 0 else '(faster)'}",
            ]
            
            # Add recommendation
            report.append("\nRecommendation:")
            if health == HealthStatus.HEALTHY:
                report.append("  ✅ Shadow version looks good for promotion to production")
            elif health == HealthStatus.WARNING:
                report.append("  ⚠️  Monitor shadow version closely before promotion")
            elif health == HealthStatus.DEGRADED:
                report.append("  ⚠️  Shadow version shows degradation - investigate before promotion")
            else:
                report.append("  ❌ Shadow version failing - DO NOT promote to production")
            
            report.append("="*70)
            return "\n".join(report)
        
        def show_divergent_examples(self, limit: int = 5):
            """Show examples where responses diverged"""
            divergent = self.get_divergent_examples(limit=limit)
            
            if not divergent:
                print("\n✅ No significant divergent examples found")
                return
            
            print(f"\n{'='*70}")
            print(f"DIVERGENT EXAMPLES (Top {len(divergent)})")
            print(f"{'='*70}")
            
            for i, response in enumerate(divergent, 1):
                print(f"\nExample {i}:")
                print(f"Difference Type: {response.difference_type.value}")
                print(f"Similarity: {response.similarity_score:.2f}")
                print(f"\nProduction Response:")
                print(f"  {response.production_response[:200]}...")
                print(f"\nShadow Response:")
                print(f"  {response.shadow_response[:200]}...")
                print("-" * 70)
    
    
    def demonstrate_shadow_mode():
        """Demonstrate shadow mode testing"""
        print("="*70)
        print("Pattern 153: Shadow Mode Testing")
        print("="*70)
        
        # Create tester with different configurations
        # In real scenario, these would be different model versions or prompts
        tester = ShadowModeTester(
            production_model="gpt-3.5-turbo",
            shadow_model="gpt-3.5-turbo",
            production_temp=0.7,
            shadow_temp=0.9  # Different temperature for shadow
        )
        
        # Test queries
        test_queries = [
            "What is Python?",
            "Explain machine learning in simple terms",
            "How do I learn programming?",
            "What's the difference between AI and ML?",
            "Write a haiku about coding",
            "What are best practices for API design?",
            "Explain recursion with an example",
            "How does garbage collection work?",
            "What is the difference between REST and GraphQL?",
            "Explain async/await in JavaScript"
        ]
        
        print("\n" + "="*70)
        print("BATCH SHADOW TESTING")
        print("="*70)
        print(f"Production: GPT-3.5-Turbo (temp=0.7)")
        print(f"Shadow: GPT-3.5-Turbo (temp=0.9)")
        
        # Run batch test
        results = tester.batch_test(test_queries, show_progress=True)
        
        # Generate report
        print(tester.generate_report())
        
        # Show divergent examples
        tester.show_divergent_examples(limit=3)
        
        # Test individual request
        print("\n" + "="*70)
        print("INDIVIDUAL REQUEST EXAMPLE")
        print("="*70)
        
        query = "Explain quantum computing"
        print(f"\nQuery: {query}")
        
        response = tester.execute_shadow_request(query)
        
        print(f"\n📊 Results:")
        print(f"  Difference Type: {response.difference_type.value}")
        print(f"  Similarity Score: {response.similarity_score:.2f}")
        print(f"  Production Time: {response.production_time:.3f}s")
        print(f"  Shadow Time: {response.shadow_time:.3f}s")
        
        print(f"\n🏭 Production Response:")
        print(f"  {response.production_response[:150]}...")
        
        print(f"\n🔬 Shadow Response:")
        print(f"  {response.shadow_response[:150]}...")
        
        # Health check
        health = tester.get_health_status()
        print(f"\n🏥 Shadow Health Status: {health.value.upper()}")
        
        # Summary
        print("\n" + "="*70)
        print("SUMMARY")
        print("="*70)
        print("""
    Shadow Mode Testing Pattern provides:
    
    1. Zero-Impact Testing:
       - Tests run parallel to production
       - Users only see production responses
       - Shadow results collected for analysis
       - No service disruption
    
    2. Comprehensive Comparison:
       - Response similarity analysis
       - Difference categorization
       - Performance comparison
       - Error rate tracking
    
    3. Health Monitoring:
       - Automatic health status assessment
       - Divergence detection
       - Performance regression identification
       - Error rate monitoring
    
    4. Deployment Confidence:
       - Real production traffic testing
       - Early problem detection
       - Data-driven promotion decisions
       - Risk mitigation
    
    5. Use Cases:
       - New model version validation
       - Prompt engineering testing
       - Configuration changes
       - Performance optimization
       - Feature rollout validation
    
    This pattern is essential for safely testing changes to production
    AI systems before full deployment.
        """)
    
    
    if __name__ == "__main__":
        demonstrate_shadow_mode()
    
        

    Agentic Design Pattern: Shadow Mode

    Pattern 153: Shadow Mode – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Shadow Mode pattern implementation.


    Pattern Description

    Pattern 153: Shadow Mode Testing

    Description: Runs new agent versions alongside production without affecting users. Captures both production and shadow responses, compares results, and identifies differences for validation before full deployment.

    Components: – Shadow Executor: Runs shadow version in parallel – Response Comparator: Compares production vs shadow outputs – Difference Analyzer: Identifies and categorizes differences – Metrics Collector: Tracks shadow performance – Rollback Monitor: Detects degradation signals

    Use Cases: – Testing new model versions safely – Validating prompt changes – A/B testing without user impact – Gradual feature rollout – Performance regression detection

    Benefits: – Zero user impact during testing – Real production traffic testing – Early problem detection – Confidence before full deployment – Comprehensive comparison data

    Trade-offs: – Doubled computational cost – Infrastructure complexity – Storage for dual responses – Latency considerations – Analysis overhead

    LangChain Implementation: Uses parallel execution with ChatOpenAI, response comparison, and comprehensive metrics tracking.


    File Statistics

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

    Module Documentation

    """
    Pattern 153: Shadow Mode Testing
    
    Description:
        Runs new agent versions alongside production without affecting users.
        Captures both production and shadow responses, compares results, and
        identifies differences for validation before full deployment.
    
    Components:
        - Shadow Executor: Runs shadow version in parallel
        - Response Comparator: Compares production vs shadow outputs
        - Difference Analyzer: Identifies and categorizes differences
        - Metrics Collector: Tracks shadow performance
        - Rollback Monitor: Detects degradation signals
    
    Use Cases:
        - Testing new model versions safely
        - Validating prompt changes
        - A/B testing without user impact
        - Gradual feature rollout
        - Performance regression detection
    
    Benefits:
        - Zero user impact during testing
        - Real production traffic testing
        - Early problem detection
        - Confidence before full deployment
        - Comprehensive comparison data
    
    Trade-offs:
        - Doubled computational cost
        - Infrastructure complexity
        - Storage for dual responses
        - Latency considerations
        - Analysis overhead
    
    LangChain Implementation:
        Uses parallel execution with ChatOpenAI, response comparison,
        and comprehensive metrics tracking.
    """
    

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


    Import Statements

    Standard Library Imports

    import asyncio
    import collections.defaultdict
    import concurrent.futures.ThreadPoolExecutor
    import concurrent.futures.as_completed
    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 time
    import typing.Any
    import typing.Dict
    import typing.List
    import typing.Optional
    import typing.Tuple
    

    Purpose: Core Python functionality: – asyncio: Asynchronous I/O operations – collections.defaultdict: Specialized container datatypes – concurrent.futures.ThreadPoolExecutor: Regular expression operations – concurrent.futures.as_completed: Regular expression operations – 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 – time: Time-related functions – 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: DifferenceType

    Lines 60-66

    Inherits from: Enum

    Purpose:

    Types of differences between responses
    

    Class: HealthStatus

    Lines 69-74

    Inherits from: Enum

    Purpose:

    Health status of shadow version
    

    Class: ShadowResponse

    Lines 78-88

    Purpose:

    Response from shadow execution
    

    Class: ShadowMetrics

    Lines 92-134

    Purpose:

    Metrics for shadow testing
    

    Methods:

    get_difference_rate(self, diff_type)

    Lines 105-117

    Purpose:

    Get rate of specific difference type
    

    Parameters: – diff_type: Diff type

    Returns: float

    get_health_status(self)

    Lines 119-134

    Purpose:

    Determine overall health status
    

    Parameters:

    Returns: HealthStatus


    Class: ShadowModeTester

    Lines 137-466

    Purpose:

    Agent that runs shadow testing by executing production and shadow
    versions in parallel, comparing results without affecting users.
    

    Methods:

    __init__(self, production_model, shadow_model, production_temp, shadow_temp)

    Lines 143-195

    Purpose:

    Initialize shadow mode tester
    

    Parameters: – production_model: Language model instance – shadow_model: Language model instance – production_temp: Production temp – shadow_temp: Shadow temp

    execute_shadow_request(self, query)

    Lines 197-243

    Purpose:

    Execute query on both production and shadow versions
    
    Args:
        query: User query to execute
    
    Returns:
        ShadowResponse with both results and comparison
    

    Parameters: – query: Input query or question

    Returns: ShadowResponse

    _execute_production(self, query)

    Lines 245-257

    Purpose:

    Execute on production system
    

    Parameters: – query: Input query or question

    Returns: <ast.Subscript object at 0x000001A06D7C9910>

    _execute_shadow(self, query)

    Lines 259-271

    Purpose:

    Execute on shadow system
    

    Parameters: – query: Input query or question

    Returns: <ast.Subscript object at 0x000001A06D6380D0>

    _compare_responses(self, prod_response, shadow_response)

    Lines 273-310

    Purpose:

    Compare two responses and categorize difference
    

    Parameters: – prod_response: Prod response – shadow_response: Shadow response

    Returns: <ast.Subscript object at 0x000001A06D85A710>

    _calculate_similarity(self, text1, text2)

    Lines 312-323

    Purpose:

    Calculate simple word-overlap similarity
    

    Parameters: – text1: Text1 – text2: Text2

    Returns: float

    _update_metrics(self, response)

    Lines 325-359

    Purpose:

    Update metrics based on response
    

    Parameters: – response: Response

    batch_test(self, queries, show_progress)

    Lines 361-388

    Purpose:

    Test multiple queries in shadow mode
    
    Args:
        queries: List of queries to test
        show_progress: Whether to show progress
    
    Returns:
        List of shadow responses
    

    Parameters: – queries: Queries – show_progress: Show progress

    Returns: <ast.Subscript object at 0x000001A06D17C050>

    get_health_status(self)

    Lines 390-392

    Purpose:

    Get current health status of shadow version
    

    Parameters:

    Returns: HealthStatus

    get_divergent_examples(self, min_difference, limit)

    Lines 394-405

    Purpose:

    Get examples where responses diverged significantly
    

    Parameters: – min_difference: Min difference – limit: Limit

    Returns: <ast.Subscript object at 0x000001A06D702490>

    generate_report(self)

    Lines 407-444

    Purpose:

    Generate comprehensive shadow testing report
    

    Parameters:

    Returns: str

    show_divergent_examples(self, limit)

    Lines 446-466

    Purpose:

    Show examples where responses diverged
    

    Parameters: – limit: Limit


    Functions

    demonstrate_shadow_mode()

    Lines 469-579

    Purpose:

    Demonstrate shadow mode testing
    

    Key Concepts Summary

    Pattern Overview

    The Shadow Mode 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 152
    • Pattern 154

    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 Shadow Mode 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: 153_shadow_mode.py
    Total Lines: 584
    Pattern Category: AI Agent Design Pattern #153


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Property Based Testing

    
        """
    Pattern 152: Property-Based Testing
    
    Description:
        Implements property-based testing for AI agents by defining invariants
        and properties that should hold across all inputs, then generating test
        cases to verify these properties.
    
    Components:
        - Property Definitions: Invariants that should always hold
        - Test Case Generators: Create diverse inputs
        - Property Verifiers: Check if properties hold
        - Hypothesis Engine: Generate test hypotheses
        - Counterexample Finder: Identify violations
    
    Use Cases:
        - Testing agent robustness across input space
        - Finding edge cases and failures
        - Verifying consistency properties
        - Ensuring output format compliance
        - Testing deterministic vs non-deterministic behaviors
    
    Benefits:
        - Discovers unexpected edge cases
        - More thorough than example-based testing
        - Finds bugs traditional testing misses
        - Documents expected behaviors as properties
        - Scales testing coverage automatically
    
    Trade-offs:
        - Requires careful property definition
        - Can be slower than unit tests
        - May generate false positives
        - Harder to debug failures
        - Non-determinism complicates verification
    
    LangChain Implementation:
        Uses ChatOpenAI with property checking chains and systematic
        test case generation.
    """
    
    import os
    import re
    import time
    from typing import List, Dict, Any, Optional, Callable, Tuple
    from dataclasses import dataclass, field
    from enum import Enum
    from datetime import datetime
    from collections import defaultdict
    from dotenv import load_dotenv
    
    from langchain_openai import ChatOpenAI
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser
    
    load_dotenv()
    
    
    class PropertyType(Enum):
        """Types of properties to test"""
        INVARIANT = "invariant"  # Must always hold
        POSTCONDITION = "postcondition"  # Output constraints
        CONSISTENCY = "consistency"  # Same input → same output
        MONOTONICITY = "monotonicity"  # Ordering preservation
        IDEMPOTENCE = "idempotence"  # f(f(x)) = f(x)
        COMMUTATIVE = "commutative"  # Order independence
        FORMAT = "format"  # Output format compliance
    
    
    @dataclass
    class Property:
        """Defines a testable property"""
        name: str
        description: str
        property_type: PropertyType
        checker: Callable[[Any, Any], bool]  # (input, output) -> bool
        examples: List[Tuple[Any, bool]] = field(default_factory=list)  # (input, expected_result)
        
        def check(self, input_val: Any, output_val: Any) -> bool:
            """Check if property holds"""
            return self.checker(input_val, output_val)
    
    
    @dataclass
    class TestResult:
        """Result of a property test"""
        property_name: str
        passed: bool
        test_case: Any
        output: Any
        error: Optional[str] = None
        execution_time: float = 0.0
        timestamp: datetime = field(default_factory=datetime.now)
    
    
    @dataclass
    class PropertyTestReport:
        """Comprehensive test report"""
        property_name: str
        total_tests: int
        passed: int
        failed: int
        counterexamples: List[Dict[str, Any]]
        execution_time: float
        
        @property
        def pass_rate(self) -> float:
            return self.passed / self.total_tests if self.total_tests > 0 else 0.0
    
    
    class PropertyBasedTester:
        """
        Agent that performs property-based testing on AI systems.
        Generates test cases and verifies properties hold.
        """
        
        def __init__(self, temperature: float = 0.7):
            """Initialize the property-based tester"""
            self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=temperature)
            self.consistent_llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.0)
            
            # Property registry
            self.properties: Dict[str, Property] = {}
            
            # Test case generator prompts
            self.test_gen_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a test case generator. Generate diverse test cases for testing AI systems.
    Property to test: {property_description}
    Constraints: {constraints}
    
    Generate test cases that:
    1. Cover normal cases
    2. Include edge cases
    3. Test boundary conditions
    4. Challenge the property
    
    Return as a list, one test case per line."""),
                ("user", "Generate {count} test cases.")
            ])
            
            self.agent_prompt = ChatPromptTemplate.from_messages([
                ("system", "You are a helpful AI assistant. Provide clear, accurate responses."),
                ("user", "{query}")
            ])
            
            # Register default properties
            self._register_default_properties()
        
        def _register_default_properties(self):
            """Register commonly used properties"""
            
            # 1. Non-empty response
            self.register_property(Property(
                name="non_empty_response",
                description="Agent must return non-empty response",
                property_type=PropertyType.POSTCONDITION,
                checker=lambda inp, out: out is not None and len(str(out).strip()) > 0
            ))
            
            # 2. Response length reasonable
            self.register_property(Property(
                name="reasonable_length",
                description="Response length should be reasonable (not too short or long)",
                property_type=PropertyType.POSTCONDITION,
                checker=lambda inp, out: 10 < len(str(out))  len(str(inp)) * 0.5
            ))
            
            # 7. No hallucinated facts (basic check)
            self.register_property(Property(
                name="no_obvious_hallucination",
                description="Response should not contain obviously false statements",
                property_type=PropertyType.INVARIANT,
                checker=lambda inp, out: not self._has_obvious_hallucination(str(out))
            ))
        
        def register_property(self, property_def: Property):
            """Register a new property to test"""
            self.properties[property_def.name] = property_def
            print(f"✅ Registered property: {property_def.name}")
        
        def test_property(self, property_name: str, test_count: int = 20,
                         custom_test_cases: Optional[List[Any]] = None) -> PropertyTestReport:
            """
            Test a specific property with generated or provided test cases
            
            Args:
                property_name: Name of property to test
                test_count: Number of test cases to generate
                custom_test_cases: Optional list of specific test cases
                
            Returns:
                Comprehensive test report
            """
            if property_name not in self.properties:
                raise ValueError(f"Property '{property_name}' not registered")
            
            property_def = self.properties[property_name]
            start_time = time.time()
            
            print(f"\n{'='*70}")
            print(f"Testing Property: {property_name}")
            print(f"Type: {property_def.property_type.value}")
            print(f"Description: {property_def.description}")
            print(f"{'='*70}")
            
            # Generate or use provided test cases
            if custom_test_cases:
                test_cases = custom_test_cases
            else:
                test_cases = self._generate_test_cases(property_def, test_count)
            
            print(f"\n🔄 Running {len(test_cases)} test cases...")
            
            results = []
            for i, test_case in enumerate(test_cases, 1):
                result = self._run_test_case(property_def, test_case, i)
                results.append(result)
            
            # Analyze results
            passed = sum(1 for r in results if r.passed)
            failed = len(results) - passed
            
            counterexamples = []
            for r in results:
                if not r.passed:
                    counterexamples.append({
                        "input": r.test_case,
                        "output": r.output,
                        "error": r.error
                    })
            
            execution_time = time.time() - start_time
            
            report = PropertyTestReport(
                property_name=property_name,
                total_tests=len(results),
                passed=passed,
                failed=failed,
                counterexamples=counterexamples,
                execution_time=execution_time
            )
            
            self._print_test_results(report)
            
            return report
        
        def test_all_properties(self, test_count: int = 10) -> Dict[str, PropertyTestReport]:
            """Test all registered properties"""
            print(f"\n{'='*70}")
            print("TESTING ALL PROPERTIES")
            print(f"{'='*70}")
            print(f"Properties to test: {len(self.properties)}")
            print(f"Test cases per property: {test_count}")
            
            reports = {}
            for property_name in self.properties:
                report = self.test_property(property_name, test_count)
                reports[property_name] = report
            
            # Overall summary
            self._print_overall_summary(reports)
            
            return reports
        
        def _generate_test_cases(self, property_def: Property, count: int) -> List[str]:
            """Generate test cases for a property"""
            print(f"   🔄 Generating test cases...")
            
            # Use examples as seed
            constraints = "Include diverse scenarios, edge cases, and boundary conditions"
            
            try:
                chain = self.test_gen_prompt | self.llm | StrOutputParser()
                result = chain.invoke({
                    "property_description": property_def.description,
                    "constraints": constraints,
                    "count": count
                })
                
                # Parse test cases (one per line or numbered)
                test_cases = []
                lines = result.strip().split('\n')
                for line in lines:
                    # Remove numbering/bullets
                    cleaned = re.sub(r'^\d+[\.\)]\s*', '', line.strip())
                    cleaned = re.sub(r'^[-•]\s*', '', cleaned)
                    if cleaned and len(cleaned) > 3:
                        test_cases.append(cleaned)
                
                # Fill up to count if needed
                while len(test_cases)  TestResult:
            """Run a single test case"""
            start_time = time.time()
            
            try:
                # Execute agent with test case
                output = self._execute_agent(test_case)
                
                # Check property
                if property_def.property_type == PropertyType.CONSISTENCY:
                    passed = self._check_consistency(test_case)
                else:
                    passed = property_def.check(test_case, output)
                
                result = TestResult(
                    property_name=property_def.name,
                    passed=passed,
                    test_case=test_case,
                    output=output,
                    execution_time=time.time() - start_time
                )
                
                status = "✓" if passed else "✗"
                print(f"   {status} Test {case_number}: {str(test_case)[:60]}...")
                
                return result
                
            except Exception as e:
                print(f"   ✗ Test {case_number}: ERROR - {str(e)}")
                return TestResult(
                    property_name=property_def.name,
                    passed=False,
                    test_case=test_case,
                    output=None,
                    error=str(e),
                    execution_time=time.time() - start_time
                )
        
        def _execute_agent(self, query: str) -> str:
            """Execute the agent being tested"""
            chain = self.agent_prompt | self.llm | StrOutputParser()
            return chain.invoke({"query": query})
        
        def _check_consistency(self, test_case: Any) -> bool:
            """Check consistency property by running multiple times"""
            outputs = []
            
            # Run 3 times with temperature 0 for consistency
            chain = self.agent_prompt | self.consistent_llm | StrOutputParser()
            for _ in range(3):
                output = chain.invoke({"query": test_case})
                outputs.append(output)
            
            # Check if all outputs are similar (basic check)
            # For production, use more sophisticated similarity metrics
            return len(set(outputs))  bool:
            """Check if text contains offensive content (basic)"""
            offensive_words = ["offensive", "inappropriate", "harmful"]  # Simplified
            return any(word in text.lower() for word in offensive_words)
        
        def _is_valid_json_if_requested(self, input_text: str, output_text: str) -> bool:
            """Check if output is valid JSON when requested"""
            if "json" not in input_text.lower():
                return True  # Not requesting JSON, so property holds
            
            try:
                import json
                json.loads(output_text)
                return True
            except:
                # Check if JSON is embedded in response
                if "{" in output_text and "}" in output_text:
                    start = output_text.find("{")
                    end = output_text.rfind("}") + 1
                    try:
                        json.loads(output_text[start:end])
                        return True
                    except:
                        return False
                return False
        
        def _has_obvious_hallucination(self, text: str) -> bool:
            """Check for obvious hallucinations (basic)"""
            # Check for contradictions or obviously false statements
            false_patterns = [
                r"the sun is cold",
                r"water is dry",
                r"earth is flat",
                r"humans have \d+ legs" # where number != 2
            ]
            
            text_lower = text.lower()
            for pattern in false_patterns:
                if re.search(pattern, text_lower):
                    return True
            
            return False
        
        def _print_test_results(self, report: PropertyTestReport):
            """Print test results"""
            print(f"\n{'='*70}")
            print(f"RESULTS: {report.property_name}")
            print(f"{'='*70}")
            print(f"Total Tests: {report.total_tests}")
            print(f"Passed: {report.passed} ({report.pass_rate*100:.1f}%)")
            print(f"Failed: {report.failed}")
            print(f"Execution Time: {report.execution_time:.2f}s")
            
            if report.counterexamples:
                print(f"\n❌ Counterexamples Found ({len(report.counterexamples)}):")
                for i, ce in enumerate(report.counterexamples[:5], 1):  # Show first 5
                    print(f"\n{i}. Input: {str(ce['input'])[:80]}...")
                    if ce['output']:
                        print(f"   Output: {str(ce['output'])[:80]}...")
                    if ce['error']:
                        print(f"   Error: {ce['error']}")
            
            print("="*70)
        
        def _print_overall_summary(self, reports: Dict[str, PropertyTestReport]):
            """Print overall summary of all tests"""
            print(f"\n{'='*70}")
            print("OVERALL TEST SUMMARY")
            print(f"{'='*70}")
            
            total_tests = sum(r.total_tests for r in reports.values())
            total_passed = sum(r.passed for r in reports.values())
            overall_pass_rate = total_passed / total_tests if total_tests > 0 else 0
            
            print(f"\nTotal Tests Run: {total_tests}")
            print(f"Overall Pass Rate: {overall_pass_rate*100:.1f}%")
            print(f"\nProperty Results:")
            
            for name, report in reports.items():
                status = "✓" if report.pass_rate >= 0.9 else "✗" if report.pass_rate  Any:
            """
            Attempt to find minimal counterexample (shrinking)
            
            Args:
                property_name: Property that failed
                counterexample: Original counterexample
                
            Returns:
                Minimal counterexample
            """
            property_def = self.properties[property_name]
            current = str(counterexample)
            
            print(f"\n🔍 Shrinking counterexample for {property_name}...")
            print(f"Original: {current[:100]}...")
            
            # Try progressively smaller versions
            for length in range(len(current) // 2, 10, -10):
                candidate = current[:length]
                
                try:
                    output = self._execute_agent(candidate)
                    if not property_def.check(candidate, output):
                        current = candidate
                        print(f"   Shrunk to: {len(current)} characters")
                except:
                    continue
            
            print(f"✓ Minimal counterexample: {current}")
            return current
    
    
    def demonstrate_property_based_testing():
        """Demonstrate property-based testing"""
        print("="*70)
        print("Pattern 152: Property-Based Testing")
        print("="*70)
        
        tester = PropertyBasedTester()
        
        # Test 1: Non-empty response property
        print("\n" + "="*70)
        print("TEST 1: Non-Empty Response Property")
        print("="*70)
        
        test_cases = [
            "What is Python?",
            "Hello",
            "",  # Edge case
            "Explain quantum physics",
            "?"  # Edge case
        ]
        
        report1 = tester.test_property("non_empty_response", 
                                       custom_test_cases=test_cases)
        
        # Test 2: Reasonable length property
        print("\n" + "="*70)
        print("TEST 2: Reasonable Length Property")
        print("="*70)
        
        report2 = tester.test_property("reasonable_length", test_count=8)
        
        # Test 3: JSON format compliance
        print("\n" + "="*70)
        print("TEST 3: JSON Format Compliance")
        print("="*70)
        
        json_test_cases = [
            "Return the result as JSON: {name: 'test'}",
            "Give me JSON with user info",
            "What is 2+2?",  # Should pass (not requesting JSON)
            "Provide a JSON object with three fields"
        ]
        
        report3 = tester.test_property("json_format", 
                                       custom_test_cases=json_test_cases)
        
        # Test 4: Custom property
        print("\n" + "="*70)
        print("TEST 4: Custom Property - No Numbers in Text Explanation")
        print("="*70)
        
        def no_numbers_checker(inp: Any, out: Any) -> bool:
            """Custom checker: explanations should avoid raw numbers"""
            if "explain" in str(inp).lower():
                # Allow some numbers but not excessive
                return str(out).count("1") + str(out).count("2") < 10
            return True
        
        custom_prop = Property(
            name="minimal_numbers_in_explanation",
            description="Explanations should minimize raw numbers",
            property_type=PropertyType.POSTCONDITION,
            checker=no_numbers_checker
        )
        
        tester.register_property(custom_prop)
        
        explain_cases = [
            "Explain how computers work",
            "What is machine learning?",
            "Tell me about AI"
        ]
        
        report4 = tester.test_property("minimal_numbers_in_explanation",
                                       custom_test_cases=explain_cases)
        
        # Test 5: Consistency property
        print("\n" + "="*70)
        print("TEST 5: Consistency Property")
        print("="*70)
        
        consistent_cases = [
            "What is 2+2?",
            "Name the capital of France"
        ]
        
        report5 = tester.test_property("consistency", 
                                       custom_test_cases=consistent_cases)
        
        # Summary
        print("\n" + "="*70)
        print("SUMMARY")
        print("="*70)
        print(f"""
    Property-Based Testing Pattern provides:
    
    1. Property Definitions:
       - Invariants (must always hold)
       - Postconditions (output constraints)
       - Consistency checks
       - Format compliance
       - Custom properties
    
    2. Test Generation:
       - Automated test case creation
       - Edge case discovery
       - Boundary condition testing
       - Coverage maximization
    
    3. Property Verification:
       - Systematic property checking
       - Counterexample identification
       - Shrinking to minimal cases
       - Detailed reporting
    
    4. Benefits:
       - Finds bugs traditional testing misses
       - Documents expected behaviors
       - Scales testing automatically
       - More thorough coverage
    
    5. Results Summary:
       - Non-empty responses: {report1.pass_rate*100:.0f}% pass
       - Reasonable length: {report2.pass_rate*100:.0f}% pass
       - JSON format: {report3.pass_rate*100:.0f}% pass
       - Custom property: {report4.pass_rate*100:.0f}% pass
       - Consistency: {report5.pass_rate*100:.0f}% pass
    
    This pattern is essential for robust AI agent testing and
    quality assurance in production systems.
        """)
    
    
    if __name__ == "__main__":
        demonstrate_property_based_testing()
    
        

    Agentic Design Pattern: Property Based Testing

    Pattern 152: Property Based Testing – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Property Based Testing pattern implementation.


    Pattern Description

    Pattern 152: Property-Based Testing

    Description: Implements property-based testing for AI agents by defining invariants and properties that should hold across all inputs, then generating test cases to verify these properties.

    Components: – Property Definitions: Invariants that should always hold – Test Case Generators: Create diverse inputs – Property Verifiers: Check if properties hold – Hypothesis Engine: Generate test hypotheses – Counterexample Finder: Identify violations

    Use Cases: – Testing agent robustness across input space – Finding edge cases and failures – Verifying consistency properties – Ensuring output format compliance – Testing deterministic vs non-deterministic behaviors

    Benefits: – Discovers unexpected edge cases – More thorough than example-based testing – Finds bugs traditional testing misses – Documents expected behaviors as properties – Scales testing coverage automatically

    Trade-offs: – Requires careful property definition – Can be slower than unit tests – May generate false positives – Harder to debug failures – Non-determinism complicates verification

    LangChain Implementation: Uses ChatOpenAI with property checking chains and systematic test case generation.


    File Statistics

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

    Module Documentation

    """
    Pattern 152: Property-Based Testing
    
    Description:
        Implements property-based testing for AI agents by defining invariants
        and properties that should hold across all inputs, then generating test
        cases to verify these properties.
    
    Components:
        - Property Definitions: Invariants that should always hold
        - Test Case Generators: Create diverse inputs
        - Property Verifiers: Check if properties hold
        - Hypothesis Engine: Generate test hypotheses
        - Counterexample Finder: Identify violations
    
    Use Cases:
        - Testing agent robustness across input space
        - Finding edge cases and failures
        - Verifying consistency properties
        - Ensuring output format compliance
        - Testing deterministic vs non-deterministic behaviors
    
    Benefits:
        - Discovers unexpected edge cases
        - More thorough than example-based testing
        - Finds bugs traditional testing misses
        - Documents expected behaviors as properties
        - Scales testing coverage automatically
    
    Trade-offs:
        - Requires careful property definition
        - Can be slower than unit tests
        - May generate false positives
        - Harder to debug failures
        - Non-determinism complicates verification
    
    LangChain Implementation:
        Uses ChatOpenAI with property checking chains and systematic
        test case generation.
    """
    

    Purpose: This module implements the Property Based Testing 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 re
    import time
    import typing.Any
    import typing.Callable
    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 – 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 – re: Regular expression operations – time: Time-related functions – 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.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: PropertyType

    Lines 59-67

    Inherits from: Enum

    Purpose:

    Types of properties to test
    

    Class: Property

    Lines 71-81

    Purpose:

    Defines a testable property
    

    Methods:

    check(self, input_val, output_val)

    Lines 79-81

    Purpose:

    Check if property holds
    

    Parameters: – input_val: Input data – output_val: Output val

    Returns: bool


    Class: TestResult

    Lines 85-93

    Purpose:

    Result of a property test
    

    Class: PropertyTestReport

    Lines 97-108

    Purpose:

    Comprehensive test report
    

    Methods:

    pass_rate(self)

    Lines 107-108

    Decorators: property

    Purpose: Pass Rate operation.

    Parameters:

    Returns: float


    Class: PropertyBasedTester

    Lines 111-506

    Purpose:

    Agent that performs property-based testing on AI systems.
    Generates test cases and verifies properties hold.
    

    Methods:

    __init__(self, temperature)

    Lines 117-147

    Purpose:

    Initialize the property-based tester
    

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

    _register_default_properties(self)

    Lines 149-206

    Purpose:

    Register commonly used properties
    

    Parameters:

    register_property(self, property_def)

    Lines 208-211

    Purpose:

    Register a new property to test
    

    Parameters: – property_def: Property def

    test_property(self, property_name, test_count, custom_test_cases)

    Lines 213-277

    Purpose:

    Test a specific property with generated or provided test cases
    
    Args:
        property_name: Name of property to test
        test_count: Number of test cases to generate
        custom_test_cases: Optional list of specific test cases
    
    Returns:
        Comprehensive test report
    

    Parameters: – property_name: Property name – test_count: Test count – custom_test_cases: Custom test cases

    Returns: PropertyTestReport

    test_all_properties(self, test_count)

    Lines 279-295

    Purpose:

    Test all registered properties
    

    Parameters: – test_count: Test count

    Returns: <ast.Subscript object at 0x000001A06D617AD0>

    _generate_test_cases(self, property_def, count)

    Lines 297-331

    Purpose:

    Generate test cases for a property
    

    Parameters: – property_def: Property def – count: Count

    Returns: <ast.Subscript object at 0x000001A06D7CFC90>

    _run_test_case(self, property_def, test_case, case_number)

    Lines 333-370

    Purpose:

    Run a single test case
    

    Parameters: – property_def: Property def – test_case: Test case – case_number: Case number

    Returns: TestResult

    _execute_agent(self, query)

    Lines 372-375

    Purpose:

    Execute the agent being tested
    

    Parameters: – query: Input query or question

    Returns: str

    _check_consistency(self, test_case)

    Lines 377-389

    Purpose:

    Check consistency property by running multiple times
    

    Parameters: – test_case: Test case

    Returns: bool

    _contains_offensive(self, text)

    Lines 391-394

    Purpose:

    Check if text contains offensive content (basic)
    

    Parameters: – text: Text

    Returns: bool

    _is_valid_json_if_requested(self, input_text, output_text)

    Lines 396-415

    Purpose:

    Check if output is valid JSON when requested
    

    Parameters: – input_text: Input data – output_text: Output text

    Returns: bool

    _has_obvious_hallucination(self, text)

    Lines 417-432

    Purpose:

    Check for obvious hallucinations (basic)
    

    Parameters: – text: Text

    Returns: bool

    _print_test_results(self, report)

    Lines 434-453

    Purpose:

    Print test results
    

    Parameters: – report: Report

    _print_overall_summary(self, reports)

    Lines 455-474

    Purpose:

    Print overall summary of all tests
    

    Parameters: – reports: Reports

    shrink_counterexample(self, property_name, counterexample)

    Lines 476-506

    Purpose:

    Attempt to find minimal counterexample (shrinking)
    
    Args:
        property_name: Property that failed
        counterexample: Original counterexample
    
    Returns:
        Minimal counterexample
    

    Parameters: – property_name: Property name – counterexample: Counterexample

    Returns: Any


    Functions

    demonstrate_property_based_testing()

    Lines 509-639

    Purpose:

    Demonstrate property-based testing
    

    Key Concepts Summary

    Pattern Overview

    The Property Based Testing 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 151
    • Pattern 153

    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 Property Based Testing 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: 152_property_based_testing.py
    Total Lines: 644
    Pattern Category: AI Agent Design Pattern #152


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Synthetic Data

    
        """
    Pattern 151: Synthetic Data Generation
    
    Description:
        Generates synthetic training and test data for agent systems using LLMs.
        Creates diverse, high-quality examples with schema validation, quality
        metrics, and coverage analysis.
    
    Components:
        - Data Schema Definitions: Structured format specifications
        - Generation Strategies: Various approaches for data creation
        - Quality Validators: Check data quality and validity
        - Coverage Analyzers: Ensure diverse data distribution
        - Augmentation Engine: Enhance existing datasets
    
    Use Cases:
        - Training data creation for fine-tuning
        - Test case generation for QA
        - Data augmentation for small datasets
        - Edge case generation for robustness testing
        - Privacy-preserving data synthesis
    
    Benefits:
        - Reduces manual data labeling effort
        - Generates diverse test scenarios
        - Creates privacy-safe datasets
        - Enables systematic coverage
        - Scales data creation efficiently
    
    Trade-offs:
        - Generated data may lack real-world complexity
        - Quality depends on LLM capabilities
        - May introduce biases from LLM
        - Validation overhead
        - Cost of generation at scale
    
    LangChain Implementation:
        Uses ChatOpenAI for generation with structured output parsing,
        Pydantic models for validation, and custom metrics.
    """
    
    import os
    import json
    import time
    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
    from dotenv import load_dotenv
    
    from langchain_openai import ChatOpenAI
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
    from pydantic import BaseModel, Field, validator
    
    load_dotenv()
    
    
    class DataType(Enum):
        """Types of synthetic data"""
        QA_PAIR = "qa_pair"
        CLASSIFICATION = "classification"
        CONVERSATION = "conversation"
        INSTRUCTION = "instruction"
        SENTIMENT = "sentiment"
        ENTITY_EXTRACTION = "entity_extraction"
        SUMMARIZATION = "summarization"
    
    
    class QualityMetric(Enum):
        """Quality metrics for generated data"""
        COHERENCE = "coherence"
        RELEVANCE = "relevance"
        DIVERSITY = "diversity"
        COMPLEXITY = "complexity"
        VALIDITY = "validity"
    
    
    # Pydantic Models for Structured Generation
    
    class QAPair(BaseModel):
        """Question-Answer pair"""
        question: str = Field(description="The question")
        answer: str = Field(description="The answer")
        context: Optional[str] = Field(default=None, description="Optional context")
        difficulty: str = Field(default="medium", description="easy, medium, or hard")
        category: Optional[str] = Field(default=None, description="Question category")
    
    
    class ClassificationExample(BaseModel):
        """Text classification example"""
        text: str = Field(description="The text to classify")
        label: str = Field(description="The classification label")
        confidence: float = Field(default=1.0, description="Confidence score 0-1")
        explanation: Optional[str] = Field(default=None, description="Explanation for label")
    
    
    class ConversationTurn(BaseModel):
        """Single conversation turn"""
        speaker: str = Field(description="Speaker identifier")
        utterance: str = Field(description="What was said")
        intent: Optional[str] = Field(default=None, description="Speaker intent")
    
    
    class Conversation(BaseModel):
        """Multi-turn conversation"""
        turns: List[ConversationTurn] = Field(description="Conversation turns")
        topic: Optional[str] = Field(default=None, description="Conversation topic")
        outcome: Optional[str] = Field(default=None, description="Conversation outcome")
    
    
    @dataclass
    class GenerationConfig:
        """Configuration for data generation"""
        data_type: DataType
        count: int = 10
        domain: Optional[str] = None
        difficulty: str = "mixed"  # easy, medium, hard, mixed
        diversity_target: float = 0.7  # Target diversity score
        quality_threshold: float = 0.6  # Minimum quality score
        include_edge_cases: bool = True
        seed_examples: List[Dict[str, Any]] = field(default_factory=list)
    
    
    @dataclass
    class QualityScores:
        """Quality scores for generated data"""
        coherence: float = 0.0
        relevance: float = 0.0
        diversity: float = 0.0
        complexity: float = 0.0
        validity: float = 0.0
        
        def overall_score(self) -> float:
            """Calculate overall quality score"""
            return (self.coherence + self.relevance + self.diversity + 
                    self.complexity + self.validity) / 5
        
        def to_dict(self) -> Dict[str, float]:
            """Convert to dictionary"""
            return {
                "coherence": self.coherence,
                "relevance": self.relevance,
                "diversity": self.diversity,
                "complexity": self.complexity,
                "validity": self.validity,
                "overall": self.overall_score()
            }
    
    
    class SyntheticDataGenerator:
        """
        Agent that generates synthetic training and test data.
        Produces high-quality, diverse datasets with validation.
        """
        
        def __init__(self, temperature: float = 0.9):
            """Initialize the synthetic data generator"""
            self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=temperature)
            self.validator_llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.3)
            
            # Track generated data for diversity
            self.generated_texts: Set[str] = set()
            self.category_counts: Dict[str, int] = defaultdict(int)
            
            # Generation prompts
            self.qa_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a data generation expert. Generate high-quality question-answer pairs.
    Domain: {domain}
    Difficulty: {difficulty}
    Requirements: {requirements}
    
    Generate diverse, realistic questions with accurate answers.
    Return as JSON with fields: question, answer, context, difficulty, category"""),
                ("user", "Generate {count} question-answer pairs.")
            ])
            
            self.classification_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a data generation expert. Generate text classification examples.
    Domain: {domain}
    Labels: {labels}
    Requirements: {requirements}
    
    Generate realistic text examples with correct labels.
    Return as JSON with fields: text, label, confidence, explanation"""),
                ("user", "Generate {count} classification examples.")
            ])
            
            self.conversation_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a conversation generation expert. Create realistic multi-turn conversations.
    Domain: {domain}
    Number of turns: {num_turns}
    Requirements: {requirements}
    
    Generate natural, coherent conversations.
    Return as JSON with structure: turns (list of speaker/utterance), topic, outcome"""),
                ("user", "Generate {count} conversations.")
            ])
            
            self.quality_eval_prompt = ChatPromptTemplate.from_messages([
                ("system", """Evaluate the quality of generated data on multiple dimensions.
    Rate each dimension from 0.0 to 1.0:
    - Coherence: Logical consistency and clarity
    - Relevance: Relevance to the domain/task
    - Diversity: Uniqueness and variety
    - Complexity: Appropriate complexity level
    - Validity: Correctness and accuracy
    
    Return JSON with scores for each dimension."""),
                ("user", "Evaluate this data:\n{data}")
            ])
        
        def generate_dataset(self, config: GenerationConfig) -> Dict[str, Any]:
            """
            Generate synthetic dataset according to configuration
            
            Args:
                config: Generation configuration
                
            Returns:
                Dictionary with generated data and metrics
            """
            start_time = time.time()
            
            print(f"\n🔄 Generating {config.count} {config.data_type.value} examples...")
            if config.domain:
                print(f"   Domain: {config.domain}")
            print(f"   Difficulty: {config.difficulty}")
            
            # Generate data based on type
            if config.data_type == DataType.QA_PAIR:
                examples = self._generate_qa_pairs(config)
            elif config.data_type == DataType.CLASSIFICATION:
                examples = self._generate_classifications(config)
            elif config.data_type == DataType.CONVERSATION:
                examples = self._generate_conversations(config)
            else:
                examples = self._generate_generic(config)
            
            # Validate and score quality
            quality_scores = self._evaluate_quality(examples, config)
            
            # Analyze diversity
            diversity_metrics = self._analyze_diversity(examples)
            
            # Filter by quality threshold
            filtered_examples = self._filter_by_quality(examples, quality_scores, 
                                                        config.quality_threshold)
            
            generation_time = time.time() - start_time
            
            return {
                "examples": filtered_examples,
                "count": len(filtered_examples),
                "quality_scores": quality_scores,
                "diversity_metrics": diversity_metrics,
                "generation_time": generation_time,
                "config": {
                    "data_type": config.data_type.value,
                    "target_count": config.count,
                    "domain": config.domain,
                    "difficulty": config.difficulty
                }
            }
        
        def _generate_qa_pairs(self, config: GenerationConfig) -> List[Dict[str, Any]]:
            """Generate question-answer pairs"""
            examples = []
            
            # Generate in batches for diversity
            batch_size = min(5, config.count)
            num_batches = (config.count + batch_size - 1) // batch_size
            
            for batch in range(num_batches):
                batch_count = min(batch_size, config.count - len(examples))
                
                requirements = self._get_generation_requirements(config, examples)
                
                try:
                    chain = self.qa_prompt | self.llm | StrOutputParser()
                    result = chain.invoke({
                        "domain": config.domain or "general knowledge",
                        "difficulty": config.difficulty,
                        "requirements": requirements,
                        "count": batch_count
                    })
                    
                    # Parse JSON response
                    parsed = self._parse_json_response(result)
                    if isinstance(parsed, list):
                        examples.extend(parsed)
                    elif isinstance(parsed, dict):
                        examples.append(parsed)
                        
                except Exception as e:
                    print(f"   ⚠️  Batch {batch + 1} generation error: {str(e)}")
            
            return examples[:config.count]
        
        def _generate_classifications(self, config: GenerationConfig) -> List[Dict[str, Any]]:
            """Generate classification examples"""
            examples = []
            
            # Default labels if not specified
            labels = ["positive", "negative", "neutral"]
            if config.domain:
                labels = self._infer_labels_from_domain(config.domain)
            
            batch_size = min(5, config.count)
            num_batches = (config.count + batch_size - 1) // batch_size
            
            for batch in range(num_batches):
                batch_count = min(batch_size, config.count - len(examples))
                
                requirements = self._get_generation_requirements(config, examples)
                
                try:
                    chain = self.classification_prompt | self.llm | StrOutputParser()
                    result = chain.invoke({
                        "domain": config.domain or "general text classification",
                        "labels": ", ".join(labels),
                        "requirements": requirements,
                        "count": batch_count
                    })
                    
                    parsed = self._parse_json_response(result)
                    if isinstance(parsed, list):
                        examples.extend(parsed)
                    elif isinstance(parsed, dict):
                        examples.append(parsed)
                        
                except Exception as e:
                    print(f"   ⚠️  Batch {batch + 1} generation error: {str(e)}")
            
            return examples[:config.count]
        
        def _generate_conversations(self, config: GenerationConfig) -> List[Dict[str, Any]]:
            """Generate multi-turn conversations"""
            examples = []
            
            num_turns = 5  # Default conversation length
            
            for i in range(config.count):
                requirements = self._get_generation_requirements(config, examples)
                
                try:
                    chain = self.conversation_prompt | self.llm | StrOutputParser()
                    result = chain.invoke({
                        "domain": config.domain or "general conversation",
                        "num_turns": num_turns,
                        "requirements": requirements,
                        "count": 1
                    })
                    
                    parsed = self._parse_json_response(result)
                    if isinstance(parsed, list) and parsed:
                        examples.append(parsed[0])
                    elif isinstance(parsed, dict):
                        examples.append(parsed)
                        
                except Exception as e:
                    print(f"   ⚠️  Example {i + 1} generation error: {str(e)}")
            
            return examples
        
        def _generate_generic(self, config: GenerationConfig) -> List[Dict[str, Any]]:
            """Generate generic examples"""
            generic_prompt = ChatPromptTemplate.from_messages([
                ("system", "Generate {count} diverse examples of {data_type} in {domain} domain. Return as JSON."),
                ("user", "Generate the examples.")
            ])
            
            chain = generic_prompt | self.llm | StrOutputParser()
            result = chain.invoke({
                "count": config.count,
                "data_type": config.data_type.value,
                "domain": config.domain or "general"
            })
            
            parsed = self._parse_json_response(result)
            return parsed if isinstance(parsed, list) else [parsed]
        
        def _get_generation_requirements(self, config: GenerationConfig, 
                                       existing: List[Dict[str, Any]]) -> str:
            """Generate requirements for next batch"""
            requirements = []
            
            if config.diversity_target > 0.5:
                requirements.append("Ensure high diversity from previous examples")
            
            if config.include_edge_cases and len(existing) > 5:
                requirements.append("Include edge cases and unusual scenarios")
            
            # Category balance
            if self.category_counts:
                underrep_categories = [cat for cat, count in self.category_counts.items() 
                                      if count  List[str]:
            """Infer appropriate labels from domain"""
            domain_lower = domain.lower()
            
            if "sentiment" in domain_lower:
                return ["positive", "negative", "neutral"]
            elif "emotion" in domain_lower:
                return ["joy", "sadness", "anger", "fear", "surprise"]
            elif "intent" in domain_lower:
                return ["question", "command", "statement", "request"]
            elif "urgency" in domain_lower:
                return ["urgent", "normal", "low_priority"]
            else:
                return ["category_a", "category_b", "category_c"]
        
        def _parse_json_response(self, response: str) -> Any:
            """Parse JSON from LLM response"""
            try:
                # Try direct JSON parse
                return json.loads(response)
            except json.JSONDecodeError:
                # Extract JSON from markdown code blocks
                if "```json" in response:
                    start = response.find("```json") + 7
                    end = response.find("```", start)
                    json_str = response[start:end].strip()
                    return json.loads(json_str)
                elif "```" in response:
                    start = response.find("```") + 3
                    end = response.find("```", start)
                    json_str = response[start:end].strip()
                    return json.loads(json_str)
                else:
                    # Try to find JSON array or object
                    for char in ['{', '[']:
                        if char in response:
                            start = response.find(char)
                            return json.loads(response[start:])
                    
                    return []
        
        def _evaluate_quality(self, examples: List[Dict[str, Any]], 
                             config: GenerationConfig) -> List[QualityScores]:
            """Evaluate quality of generated examples"""
            quality_scores = []
            
            # Sample for evaluation (evaluate all if small dataset)
            sample_size = min(len(examples), 10)
            sample_indices = list(range(0, len(examples), max(1, len(examples) // sample_size)))
            
            for idx in sample_indices[:sample_size]:
                example = examples[idx]
                
                try:
                    chain = self.quality_eval_prompt | self.validator_llm | StrOutputParser()
                    result = chain.invoke({"data": json.dumps(example, indent=2)})
                    
                    scores_dict = self._parse_json_response(result)
                    
                    scores = QualityScores(
                        coherence=scores_dict.get("coherence", 0.7),
                        relevance=scores_dict.get("relevance", 0.7),
                        diversity=scores_dict.get("diversity", 0.7),
                        complexity=scores_dict.get("complexity", 0.7),
                        validity=scores_dict.get("validity", 0.7)
                    )
                    quality_scores.append(scores)
                    
                except Exception as e:
                    # Default scores on error
                    quality_scores.append(QualityScores(
                        coherence=0.7, relevance=0.7, diversity=0.7,
                        complexity=0.7, validity=0.7
                    ))
            
            # Extend scores to all examples (reuse evaluated scores)
            while len(quality_scores)  Dict[str, Any]:
            """Analyze diversity of generated examples"""
            if not examples:
                return {"diversity_score": 0.0}
            
            # Extract texts for similarity analysis
            texts = []
            for ex in examples:
                if "question" in ex:
                    texts.append(ex["question"])
                elif "text" in ex:
                    texts.append(ex["text"])
                elif "turns" in ex:
                    texts.append(" ".join(t.get("utterance", "") for t in ex["turns"]))
            
            # Calculate pairwise uniqueness
            unique_ratio = len(set(texts)) / len(texts) if texts else 0
            
            # Category distribution
            categories = [ex.get("category", "unknown") for ex in examples]
            category_dist = {cat: categories.count(cat) / len(categories) 
                            for cat in set(categories)}
            
            # Length variance
            lengths = [len(t) for t in texts]
            avg_length = sum(lengths) / len(lengths) if lengths else 0
            length_variance = sum((l - avg_length) ** 2 for l in lengths) / len(lengths) if lengths else 0
            
            return {
                "diversity_score": unique_ratio,
                "unique_examples": len(set(texts)),
                "total_examples": len(texts),
                "category_distribution": category_dist,
                "avg_length": avg_length,
                "length_variance": length_variance
            }
        
        def _filter_by_quality(self, examples: List[Dict[str, Any]], 
                              quality_scores: List[QualityScores],
                              threshold: float) -> List[Dict[str, Any]]:
            """Filter examples by quality threshold"""
            filtered = []
            
            for example, scores in zip(examples, quality_scores):
                if scores.overall_score() >= threshold:
                    filtered.append(example)
            
            if len(filtered)  List[Dict[str, Any]]:
            """
            Augment existing dataset with variations
            
            Args:
                existing_data: Existing examples to augment
                augmentation_factor: How many variations per example
                
            Returns:
                Augmented dataset
            """
            print(f"\n🔄 Augmenting {len(existing_data)} examples (factor: {augmentation_factor})...")
            
            augmented = existing_data.copy()
            
            augment_prompt = ChatPromptTemplate.from_messages([
                ("system", """Create {count} variations of the following example while maintaining its core meaning and structure.
    Make the variations diverse but semantically similar.
    Return as JSON array."""),
                ("user", "Original example:\n{example}")
            ])
            
            for example in existing_data:
                try:
                    chain = augment_prompt | self.llm | StrOutputParser()
                    result = chain.invoke({
                        "count": augmentation_factor,
                        "example": json.dumps(example, indent=2)
                    })
                    
                    variations = self._parse_json_response(result)
                    if isinstance(variations, list):
                        augmented.extend(variations[:augmentation_factor])
                        
                except Exception as e:
                    print(f"   ⚠️  Augmentation error: {str(e)}")
            
            print(f"   ✅ Augmented to {len(augmented)} examples")
            return augmented
        
        def generate_report(self, results: Dict[str, Any]) -> str:
            """Generate comprehensive generation report"""
            report = [
                "\n" + "="*70,
                "SYNTHETIC DATA GENERATION REPORT",
                "="*70,
                f"\nConfiguration:",
                f"  Data Type: {results['config']['data_type']}",
                f"  Target Count: {results['config']['target_count']}",
                f"  Domain: {results['config']['domain'] or 'General'}",
                f"  Difficulty: {results['config']['difficulty']}",
                f"\nGeneration Results:",
                f"  Generated: {results['count']} examples",
                f"  Generation Time: {results['generation_time']:.2f}s",
                f"  Rate: {results['count']/results['generation_time']:.1f} examples/sec"
            ]
            
            # Quality metrics
            if results['quality_scores']:
                avg_scores = {
                    "coherence": sum(s.coherence for s in results['quality_scores']) / len(results['quality_scores']),
                    "relevance": sum(s.relevance for s in results['quality_scores']) / len(results['quality_scores']),
                    "diversity": sum(s.diversity for s in results['quality_scores']) / len(results['quality_scores']),
                    "complexity": sum(s.complexity for s in results['quality_scores']) / len(results['quality_scores']),
                    "validity": sum(s.validity for s in results['quality_scores']) / len(results['quality_scores'])
                }
                
                report.extend([
                    "\nQuality Scores (0.0-1.0):",
                    f"  Coherence: {avg_scores['coherence']:.3f}",
                    f"  Relevance: {avg_scores['relevance']:.3f}",
                    f"  Diversity: {avg_scores['diversity']:.3f}",
                    f"  Complexity: {avg_scores['complexity']:.3f}",
                    f"  Validity: {avg_scores['validity']:.3f}",
                    f"  Overall: {sum(avg_scores.values())/len(avg_scores):.3f}"
                ])
            
            # Diversity metrics
            div = results['diversity_metrics']
            report.extend([
                "\nDiversity Metrics:",
                f"  Diversity Score: {div['diversity_score']:.3f}",
                f"  Unique Examples: {div['unique_examples']}/{div['total_examples']}",
                f"  Average Length: {div['avg_length']:.0f} characters"
            ])
            
            if 'category_distribution' in div:
                report.append("  Category Distribution:")
                for cat, ratio in sorted(div['category_distribution'].items(), 
                                        key=lambda x: x[1], reverse=True):
                    report.append(f"    - {cat}: {ratio*100:.1f}%")
            
            report.append("="*70)
            return "\n".join(report)
    
    
    def demonstrate_synthetic_data_generation():
        """Demonstrate synthetic data generation"""
        print("="*70)
        print("Pattern 151: Synthetic Data Generation")
        print("="*70)
        
        generator = SyntheticDataGenerator()
        
        # Test 1: QA Pairs
        print("\n" + "="*70)
        print("TEST 1: Generate Question-Answer Pairs")
        print("="*70)
        
        qa_config = GenerationConfig(
            data_type=DataType.QA_PAIR,
            count=10,
            domain="Python programming",
            difficulty="mixed",
            quality_threshold=0.6
        )
        
        qa_results = generator.generate_dataset(qa_config)
        
        print(f"\n✅ Generated {qa_results['count']} QA pairs")
        print("\nSample Examples:")
        for i, example in enumerate(qa_results['examples'][:3], 1):
            print(f"\n{i}. Q: {example.get('question', 'N/A')}")
            print(f"   A: {example.get('answer', 'N/A')[:100]}...")
            if 'difficulty' in example:
                print(f"   Difficulty: {example['difficulty']}")
        
        print(generator.generate_report(qa_results))
        
        # Test 2: Classification Examples
        print("\n" + "="*70)
        print("TEST 2: Generate Classification Examples")
        print("="*70)
        
        class_config = GenerationConfig(
            data_type=DataType.CLASSIFICATION,
            count=8,
            domain="sentiment analysis",
            difficulty="medium",
            quality_threshold=0.6
        )
        
        class_results = generator.generate_dataset(class_config)
        
        print(f"\n✅ Generated {class_results['count']} classification examples")
        print("\nSample Examples:")
        for i, example in enumerate(class_results['examples'][:3], 1):
            print(f"\n{i}. Text: {example.get('text', 'N/A')[:80]}...")
            print(f"   Label: {example.get('label', 'N/A')}")
            if 'confidence' in example:
                print(f"   Confidence: {example['confidence']:.2f}")
        
        # Test 3: Conversations
        print("\n" + "="*70)
        print("TEST 3: Generate Conversations")
        print("="*70)
        
        conv_config = GenerationConfig(
            data_type=DataType.CONVERSATION,
            count=5,
            domain="customer support",
            quality_threshold=0.6
        )
        
        conv_results = generator.generate_dataset(conv_config)
        
        print(f"\n✅ Generated {conv_results['count']} conversations")
        if conv_results['examples']:
            print("\nSample Conversation:")
            example = conv_results['examples'][0]
            if 'turns' in example:
                for turn in example['turns'][:4]:
                    speaker = turn.get('speaker', 'Unknown')
                    utterance = turn.get('utterance', '')
                    print(f"   {speaker}: {utterance}")
            if 'topic' in example:
                print(f"   Topic: {example['topic']}")
        
        # Test 4: Data Augmentation
        print("\n" + "="*70)
        print("TEST 4: Data Augmentation")
        print("="*70)
        
        seed_data = [
            {"question": "What is Python?", "answer": "A programming language", "category": "basics"},
            {"question": "How to define a function?", "answer": "Use the def keyword", "category": "syntax"}
        ]
        
        augmented = generator.augment_dataset(seed_data, augmentation_factor=2)
        print(f"\n✅ Augmented {len(seed_data)} examples to {len(augmented)} examples")
        
        # Summary
        print("\n" + "="*70)
        print("SUMMARY")
        print("="*70)
        print("""
    Synthetic Data Generation Pattern provides:
    
    1. Multiple Data Types:
       - Question-Answer pairs
       - Classification examples
       - Multi-turn conversations
       - Custom formats
    
    2. Quality Control:
       - Automated quality evaluation
       - Coherence, relevance, validity checks
       - Configurable quality thresholds
       - Filtering low-quality examples
    
    3. Diversity Assurance:
       - Uniqueness tracking
       - Category balance
       - Length variation
       - Edge case inclusion
    
    4. Augmentation Capabilities:
       - Variation generation
       - Dataset expansion
       - Semantic preservation
    
    5. Production Features:
       - Batch generation
       - Domain specialization
       - Difficulty levels
       - Comprehensive reporting
    
    This pattern is essential for creating training data, test cases,
    and expanding datasets with privacy-safe synthetic examples.
        """)
    
    
    if __name__ == "__main__":
        demonstrate_synthetic_data_generation()
    
        

    Agentic Design Pattern: Synthetic Data

    Pattern 151: Synthetic Data – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Synthetic Data pattern implementation.


    Pattern Description

    Pattern 151: Synthetic Data Generation

    Description: Generates synthetic training and test data for agent systems using LLMs. Creates diverse, high-quality examples with schema validation, quality metrics, and coverage analysis.

    Components: – Data Schema Definitions: Structured format specifications – Generation Strategies: Various approaches for data creation – Quality Validators: Check data quality and validity – Coverage Analyzers: Ensure diverse data distribution – Augmentation Engine: Enhance existing datasets

    Use Cases: – Training data creation for fine-tuning – Test case generation for QA – Data augmentation for small datasets – Edge case generation for robustness testing – Privacy-preserving data synthesis

    Benefits: – Reduces manual data labeling effort – Generates diverse test scenarios – Creates privacy-safe datasets – Enables systematic coverage – Scales data creation efficiently

    Trade-offs: – Generated data may lack real-world complexity – Quality depends on LLM capabilities – May introduce biases from LLM – Validation overhead – Cost of generation at scale

    LangChain Implementation: Uses ChatOpenAI for generation with structured output parsing, Pydantic models for validation, and custom metrics.


    File Statistics

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

    Module Documentation

    """
    Pattern 151: Synthetic Data Generation
    
    Description:
        Generates synthetic training and test data for agent systems using LLMs.
        Creates diverse, high-quality examples with schema validation, quality
        metrics, and coverage analysis.
    
    Components:
        - Data Schema Definitions: Structured format specifications
        - Generation Strategies: Various approaches for data creation
        - Quality Validators: Check data quality and validity
        - Coverage Analyzers: Ensure diverse data distribution
        - Augmentation Engine: Enhance existing datasets
    
    Use Cases:
        - Training data creation for fine-tuning
        - Test case generation for QA
        - Data augmentation for small datasets
        - Edge case generation for robustness testing
        - Privacy-preserving data synthesis
    
    Benefits:
        - Reduces manual data labeling effort
        - Generates diverse test scenarios
        - Creates privacy-safe datasets
        - Enables systematic coverage
        - Scales data creation efficiently
    
    Trade-offs:
        - Generated data may lack real-world complexity
        - Quality depends on LLM capabilities
        - May introduce biases from LLM
        - Validation overhead
        - Cost of generation at scale
    
    LangChain Implementation:
        Uses ChatOpenAI for generation with structured output parsing,
        Pydantic models for validation, and custom metrics.
    """
    

    Purpose: This module implements the Synthetic Data 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.JsonOutputParser
    import langchain_core.output_parsers.StrOutputParser
    import langchain_core.prompts.ChatPromptTemplate
    import os
    import time
    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.JsonOutputParser: Regular expression operations – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – time: Time-related functions – 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
    from pydantic.BaseModel
    from pydantic.Field
    from pydantic.validator
    

    Purpose: Additional dependencies for specialized functionality.


    Classes

    Class: DataType

    Lines 60-68

    Inherits from: Enum

    Purpose:

    Types of synthetic data
    

    Class: QualityMetric

    Lines 71-77

    Inherits from: Enum

    Purpose:

    Quality metrics for generated data
    

    Class: QAPair

    Lines 82-88

    Inherits from: BaseModel

    Purpose:

    Question-Answer pair
    

    Class: ClassificationExample

    Lines 91-96

    Inherits from: BaseModel

    Purpose:

    Text classification example
    

    Class: ConversationTurn

    Lines 99-103

    Inherits from: BaseModel

    Purpose:

    Single conversation turn
    

    Class: Conversation

    Lines 106-110

    Inherits from: BaseModel

    Purpose:

    Multi-turn conversation
    

    Class: GenerationConfig

    Lines 114-123

    Purpose:

    Configuration for data generation
    

    Class: QualityScores

    Lines 127-149

    Purpose:

    Quality scores for generated data
    

    Methods:

    overall_score(self)

    Lines 135-138

    Purpose:

    Calculate overall quality score
    

    Parameters:

    Returns: float

    to_dict(self)

    Lines 140-149

    Purpose:

    Convert to dictionary
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D852450>


    Class: SyntheticDataGenerator

    Lines 152-631

    Purpose:

    Agent that generates synthetic training and test data.
    Produces high-quality, diverse datasets with validation.
    

    Methods:

    __init__(self, temperature)

    Lines 158-212

    Purpose:

    Initialize the synthetic data generator
    

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

    generate_dataset(self, config)

    Lines 214-265

    Purpose:

    Generate synthetic dataset according to configuration
    
    Args:
        config: Generation configuration
    
    Returns:
        Dictionary with generated data and metrics
    

    Parameters: – config: Config

    Returns: <ast.Subscript object at 0x000001A06E0A8910>

    _generate_qa_pairs(self, config)

    Lines 267-299

    Purpose:

    Generate question-answer pairs
    

    Parameters: – config: Config

    Returns: <ast.Subscript object at 0x000001A06D682490>

    _generate_classifications(self, config)

    Lines 301-336

    Purpose:

    Generate classification examples
    

    Parameters: – config: Config

    Returns: <ast.Subscript object at 0x000001A06D80D9D0>

    _generate_conversations(self, config)

    Lines 338-365

    Purpose:

    Generate multi-turn conversations
    

    Parameters: – config: Config

    Returns: <ast.Subscript object at 0x000001A06D1572D0>

    _generate_generic(self, config)

    Lines 367-382

    Purpose:

    Generate generic examples
    

    Parameters: – config: Config

    Returns: <ast.Subscript object at 0x000001A06D66A490>

    _get_generation_requirements(self, config, existing)

    Lines 384-402

    Purpose:

    Generate requirements for next batch
    

    Parameters: – config: Config – existing: Existing

    Returns: str

    _infer_labels_from_domain(self, domain)

    Lines 404-417

    Purpose:

    Infer appropriate labels from domain
    

    Parameters: – domain: Domain

    Returns: <ast.Subscript object at 0x000001A06D7A1890>

    _parse_json_response(self, response)

    Lines 419-443

    Purpose:

    Parse JSON from LLM response
    

    Parameters: – response: Response

    Returns: Any

    _evaluate_quality(self, examples, config)

    Lines 445-483

    Purpose:

    Evaluate quality of generated examples
    

    Parameters: – examples: Examples – config: Config

    Returns: <ast.Subscript object at 0x000001A06D63FB50>

    _analyze_diversity(self, examples)

    Lines 485-520

    Purpose:

    Analyze diversity of generated examples
    

    Parameters: – examples: Examples

    Returns: <ast.Subscript object at 0x000001A06D7C9590>

    _filter_by_quality(self, examples, quality_scores, threshold)

    Lines 522-535

    Purpose:

    Filter examples by quality threshold
    

    Parameters: – examples: Examples – quality_scores: Quality scores – threshold: Threshold

    Returns: <ast.Subscript object at 0x000001A06D620910>

    augment_dataset(self, existing_data, augmentation_factor)

    Lines 537-576

    Purpose:

    Augment existing dataset with variations
    
    Args:
        existing_data: Existing examples to augment
        augmentation_factor: How many variations per example
    
    Returns:
        Augmented dataset
    

    Parameters: – existing_data: Existing data – augmentation_factor: Augmentation factor

    Returns: <ast.Subscript object at 0x000001A06D16EBD0>

    generate_report(self, results)

    Lines 578-631

    Purpose:

    Generate comprehensive generation report
    

    Parameters: – results: Results

    Returns: str


    Functions

    demonstrate_synthetic_data_generation()

    Lines 634-767

    Purpose:

    Demonstrate synthetic data generation
    

    Key Concepts Summary

    Pattern Overview

    The Synthetic Data 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 150
    • Pattern 152

    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 Synthetic Data 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: 151_synthetic_data.py
    Total Lines: 772
    Pattern Category: AI Agent Design Pattern #151


    Generated from actual implementation code
    For educational and reference purposes