Tag: llm

  • Agentic Design Pattern: Constitutional Chain

    
        """
    Pattern 168: Constitutional Chain
    
    Description:
        The Constitutional Chain pattern ensures AI outputs align with predefined principles,
        values, and rules through iterative self-critique and revision. The agent generates
        an initial response, critiques it against constitutional principles, and refines it
        until it meets all requirements.
    
    Components:
        1. Constitution: Set of principles and rules
        2. Generator: Creates initial responses
        3. Critic: Evaluates responses against constitution
        4. Reviser: Improves responses based on critique
        5. Compliance Checker: Validates final alignment
        6. Iteration Manager: Controls refinement loops
    
    Use Cases:
        - Ethical AI systems
        - Content moderation
        - Policy-compliant generation
        - Bias reduction
        - Safety-critical applications
        - Regulated industries
    
    Benefits:
        - Value alignment
        - Explicit principles
        - Iterative improvement
        - Audit trail
        - Consistent behavior
        - Safety guarantees
    
    Trade-offs:
        - Multiple LLM calls
        - Increased latency
        - May be overly cautious
        - Requires good constitution
        - Complexity overhead
    
    LangChain Implementation:
        Implements iterative critique-revision loops with LangChain. Uses structured
        prompts for critique and revision stages with constitutional principles.
    """
    
    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
    
    load_dotenv()
    
    
    class PrincipleType(Enum):
        """Types of constitutional principles"""
        ETHICAL = "ethical"
        SAFETY = "safety"
        ACCURACY = "accuracy"
        BIAS = "bias"
        TONE = "tone"
        LEGAL = "legal"
        PRIVACY = "privacy"
    
    
    class ComplianceLevel(Enum):
        """Compliance assessment levels"""
        COMPLIANT = "compliant"
        MINOR_ISSUES = "minor_issues"
        MAJOR_VIOLATIONS = "major_violations"
        CRITICAL = "critical"
    
    
    @dataclass
    class ConstitutionalPrinciple:
        """Represents a principle in the constitution"""
        name: str
        description: str
        principle_type: PrincipleType
        priority: int = 5  # 1-10, higher is more important
        examples_good: List[str] = field(default_factory=list)
        examples_bad: List[str] = field(default_factory=list)
    
    
    @dataclass
    class CritiqueResult:
        """Result of critiquing a response"""
        compliance_level: ComplianceLevel
        violations: List[str]
        suggestions: List[str]
        score: float  # 0-1 scale
        critique_text: str
    
    
    @dataclass
    class ConstitutionalResponse:
        """Final response with compliance information"""
        response: str
        iterations: int
        final_compliance: ComplianceLevel
        improvements_made: List[str]
        critique_history: List[CritiqueResult]
    
    
    class Constitution:
        """Defines a set of principles for AI behavior"""
        
        def __init__(self, name: str):
            self.name = name
            self.principles: List[ConstitutionalPrinciple] = []
        
        def add_principle(self, principle: ConstitutionalPrinciple):
            """Add a principle to the constitution"""
            self.principles.append(principle)
        
        def get_principles_text(self) -> str:
            """Get formatted text of all principles"""
            lines = [f"Constitution: {self.name}", "=" * 60, ""]
            
            for i, principle in enumerate(sorted(self.principles, 
                                                key=lambda p: p.priority, 
                                                reverse=True), 1):
                lines.append(f"{i}. {principle.name} (Priority: {principle.priority})")
                lines.append(f"   Type: {principle.principle_type.value}")
                lines.append(f"   {principle.description}")
                if principle.examples_good:
                    lines.append(f"   Good: {principle.examples_good[0]}")
                if principle.examples_bad:
                    lines.append(f"   Bad: {principle.examples_bad[0]}")
                lines.append("")
            
            return "\n".join(lines)
    
    
    class ConstitutionalChain:
        """Implements constitutional AI pattern with critique-revision loops"""
        
        def __init__(self, constitution: Constitution, max_iterations: int = 3):
            """
            Initialize constitutional chain
            
            Args:
                constitution: The constitution to follow
                max_iterations: Maximum number of revision iterations
            """
            self.constitution = constitution
            self.max_iterations = max_iterations
            self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
            
            # Prompts
            self.generation_prompt = ChatPromptTemplate.from_messages([
                ("system", "You are a helpful AI assistant."),
                ("user", "{query}")
            ])
            
            self.critique_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a constitutional AI critic. Evaluate the response 
                against the following principles:
    
    {principles}
    
    Identify any violations and provide specific suggestions for improvement."""),
                ("user", """Query: {query}
    
    Response to evaluate:
    {response}
    
    Provide:
    1. Compliance level: compliant, minor_issues, major_violations, or critical
    2. List of violations (if any)
    3. Specific suggestions for improvement
    4. Overall score (0-1)
    
    Format your response as:
    COMPLIANCE: [level]
    VIOLATIONS: [list violations or "none"]
    SUGGESTIONS: [list suggestions]
    SCORE: [0-1 score]
    CRITIQUE: [detailed explanation]""")
            ])
            
            self.revision_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are an AI assistant that revises responses to align with 
                constitutional principles. Improve the response based on the critique while 
                maintaining helpfulness and relevance."""),
                ("user", """Original query: {query}
    
    Previous response:
    {response}
    
    Critique:
    {critique}
    
    Revise the response to address all issues while staying helpful and relevant.""")
            ])
            
            self.generation_chain = self.generation_prompt | self.llm | StrOutputParser()
            self.critique_chain = self.critique_prompt | self.llm | StrOutputParser()
            self.revision_chain = self.revision_prompt | self.llm | StrOutputParser()
        
        def generate_constitutional_response(self, query: str) -> ConstitutionalResponse:
            """Generate a response that complies with constitutional principles"""
            # Generate initial response
            response = self.generation_chain.invoke({"query": query})
            
            critique_history = []
            improvements = []
            iterations = 0
            
            principles_text = self.constitution.get_principles_text()
            
            for iteration in range(self.max_iterations):
                iterations += 1
                
                # Critique current response
                critique = self._critique_response(query, response, principles_text)
                critique_history.append(critique)
                
                # Check if compliant
                if critique.compliance_level == ComplianceLevel.COMPLIANT:
                    break
                
                if critique.compliance_level == ComplianceLevel.CRITICAL:
                    # Critical issues require immediate attention
                    improvements.append(f"Iteration {iteration + 1}: Fixed critical issues")
                elif critique.compliance_level == ComplianceLevel.MAJOR_VIOLATIONS:
                    improvements.append(f"Iteration {iteration + 1}: Fixed major violations")
                else:
                    improvements.append(f"Iteration {iteration + 1}: Addressed minor issues")
                
                # Revise response
                response = self._revise_response(query, response, critique.critique_text)
            
            # Final critique
            final_critique = critique_history[-1]
            
            return ConstitutionalResponse(
                response=response,
                iterations=iterations,
                final_compliance=final_critique.compliance_level,
                improvements_made=improvements,
                critique_history=critique_history
            )
        
        def _critique_response(self, query: str, response: str, 
                              principles: str) -> CritiqueResult:
            """Critique a response against constitutional principles"""
            critique_text = self.critique_chain.invoke({
                "query": query,
                "response": response,
                "principles": principles
            })
            
            # Parse critique
            compliance_level = ComplianceLevel.COMPLIANT
            violations = []
            suggestions = []
            score = 1.0
            
            for line in critique_text.split('\n'):
                line = line.strip()
                if line.startswith('COMPLIANCE:'):
                    level_str = line.split(':', 1)[1].strip().lower()
                    if 'critical' in level_str:
                        compliance_level = ComplianceLevel.CRITICAL
                    elif 'major' in level_str:
                        compliance_level = ComplianceLevel.MAJOR_VIOLATIONS
                    elif 'minor' in level_str:
                        compliance_level = ComplianceLevel.MINOR_ISSUES
                    else:
                        compliance_level = ComplianceLevel.COMPLIANT
                elif line.startswith('VIOLATIONS:'):
                    viol_text = line.split(':', 1)[1].strip()
                    if viol_text.lower() != 'none':
                        violations = [v.strip() for v in viol_text.split(',')]
                elif line.startswith('SUGGESTIONS:'):
                    sugg_text = line.split(':', 1)[1].strip()
                    suggestions = [s.strip() for s in sugg_text.split(',')]
                elif line.startswith('SCORE:'):
                    try:
                        score = float(line.split(':', 1)[1].strip())
                    except:
                        score = 0.8
            
            return CritiqueResult(
                compliance_level=compliance_level,
                violations=violations,
                suggestions=suggestions,
                score=score,
                critique_text=critique_text
            )
        
        def _revise_response(self, query: str, response: str, critique: str) -> str:
            """Revise response based on critique"""
            revised = self.revision_chain.invoke({
                "query": query,
                "response": response,
                "critique": critique
            })
            return revised
    
    
    def demonstrate_constitutional_chain():
        """Demonstrate constitutional chain pattern"""
        print("=" * 80)
        print("CONSTITUTIONAL CHAIN PATTERN DEMONSTRATION")
        print("=" * 80)
        
        # Example 1: Basic constitutional AI
        print("\n" + "=" * 80)
        print("Example 1: Basic Constitutional AI")
        print("=" * 80)
        
        # Create a simple constitution
        basic_constitution = Constitution("Basic Safety Constitution")
        
        basic_constitution.add_principle(ConstitutionalPrinciple(
            name="Helpful and Harmless",
            description="Responses must be helpful while avoiding harmful content",
            principle_type=PrincipleType.ETHICAL,
            priority=10,
            examples_good=["I can help you learn programming"],
            examples_bad=["I can help you hack systems"]
        ))
        
        basic_constitution.add_principle(ConstitutionalPrinciple(
            name="No Misinformation",
            description="Avoid spreading false or misleading information",
            principle_type=PrincipleType.ACCURACY,
            priority=9,
            examples_good=["According to research..."],
            examples_bad=["I think this is true but not sure..."]
        ))
        
        basic_constitution.add_principle(ConstitutionalPrinciple(
            name="Respectful Tone",
            description="Maintain respectful and professional communication",
            principle_type=PrincipleType.TONE,
            priority=7,
            examples_good=["Let me help you understand this"],
            examples_bad=["That's a stupid question"]
        ))
        
        print("\n" + basic_constitution.get_principles_text())
        
        # Create chain
        chain = ConstitutionalChain(basic_constitution, max_iterations=3)
        
        # Test query
        query = "Tell me how to get rich quick with cryptocurrency"
        
        print("\nQuery:", query)
        print("\nGenerating constitutional response...")
        print("-" * 60)
        
        result = chain.generate_constitutional_response(query)
        
        print(f"\nIterations required: {result.iterations}")
        print(f"Final compliance: {result.final_compliance.value}")
        
        if result.improvements_made:
            print("\nImprovements made:")
            for improvement in result.improvements_made:
                print(f"  - {improvement}")
        
        print(f"\nFinal response:")
        print(result.response)
        
        print("\n" + "-" * 60)
        print("Critique History:")
        for i, critique in enumerate(result.critique_history, 1):
            print(f"\nIteration {i}:")
            print(f"  Compliance: {critique.compliance_level.value}")
            print(f"  Score: {critique.score:.2f}")
            if critique.violations:
                print(f"  Violations: {', '.join(critique.violations[:2])}")
            if critique.suggestions:
                print(f"  Suggestions: {', '.join(critique.suggestions[:2])}")
        
        # Example 2: Content moderation constitution
        print("\n" + "=" * 80)
        print("Example 2: Content Moderation")
        print("=" * 80)
        
        moderation_constitution = Constitution("Content Moderation")
        
        moderation_constitution.add_principle(ConstitutionalPrinciple(
            name="No Hate Speech",
            description="Responses must not contain discriminatory or hateful content",
            principle_type=PrincipleType.ETHICAL,
            priority=10
        ))
        
        moderation_constitution.add_principle(ConstitutionalPrinciple(
            name="Age Appropriate",
            description="Content should be suitable for general audiences",
            principle_type=PrincipleType.SAFETY,
            priority=9
        ))
        
        moderation_constitution.add_principle(ConstitutionalPrinciple(
            name="No Personal Attacks",
            description="Avoid personal attacks or aggressive language",
            principle_type=PrincipleType.TONE,
            priority=8
        ))
        
        mod_chain = ConstitutionalChain(moderation_constitution, max_iterations=2)
        
        queries = [
            "What do you think about people who disagree with me?",
            "How do I respond to criticism online?"
        ]
        
        for query in queries:
            print(f"\nQuery: {query}")
            result = mod_chain.generate_constitutional_response(query)
            print(f"Compliance: {result.final_compliance.value}")
            print(f"Response: {result.response[:120]}...")
        
        # Example 3: Professional communication constitution
        print("\n" + "=" * 80)
        print("Example 3: Professional Communication")
        print("=" * 80)
        
        prof_constitution = Constitution("Professional Communication")
        
        prof_constitution.add_principle(ConstitutionalPrinciple(
            name="Clarity",
            description="Responses must be clear and unambiguous",
            principle_type=PrincipleType.TONE,
            priority=8
        ))
        
        prof_constitution.add_principle(ConstitutionalPrinciple(
            name="Evidence-Based",
            description="Claims should be supported by evidence or clearly marked as opinions",
            principle_type=PrincipleType.ACCURACY,
            priority=9
        ))
        
        prof_constitution.add_principle(ConstitutionalPrinciple(
            name="Conciseness",
            description="Responses should be concise while complete",
            principle_type=PrincipleType.TONE,
            priority=6
        ))
        
        prof_chain = ConstitutionalChain(prof_constitution, max_iterations=3)
        
        query = "Explain quantum computing"
        print(f"\nQuery: {query}")
        print("\nGenerating professional response...")
        
        result = prof_chain.generate_constitutional_response(query)
        
        print(f"\nIterations: {result.iterations}")
        print(f"Final compliance: {result.final_compliance.value}")
        print(f"\nResponse:\n{result.response}")
        
        # Example 4: Comparing with and without constitutional chain
        print("\n" + "=" * 80)
        print("Example 4: Comparison - With vs Without Constitution")
        print("=" * 80)
        
        query = "Should I invest all my money in a single stock?"
        
        # Without constitution
        print(f"\nQuery: {query}")
        print("\n1. Without Constitutional Chain:")
        basic_llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
        basic_chain = ChatPromptTemplate.from_messages([
            ("user", "{query}")
        ]) | basic_llm | StrOutputParser()
        
        basic_response = basic_chain.invoke({"query": query})
        print(f"   {basic_response[:200]}...")
        
        # With constitution
        print("\n2. With Constitutional Chain:")
        safety_constitution = Constitution("Financial Safety")
        safety_constitution.add_principle(ConstitutionalPrinciple(
            name="Risk Disclosure",
            description="Always disclose risks in financial advice",
            principle_type=PrincipleType.ETHICAL,
            priority=10
        ))
        safety_constitution.add_principle(ConstitutionalPrinciple(
            name="No Guarantees",
            description="Avoid making guarantees about financial outcomes",
            principle_type=PrincipleType.ACCURACY,
            priority=9
        ))
        
        safety_chain = ConstitutionalChain(safety_constitution, max_iterations=2)
        result = safety_chain.generate_constitutional_response(query)
        print(f"   {result.response[:200]}...")
        print(f"\n   Iterations: {result.iterations}")
        print(f"   Compliance: {result.final_compliance.value}")
        
        # Summary
        print("\n" + "=" * 80)
        print("SUMMARY")
        print("=" * 80)
        print("""
    The Constitutional Chain pattern enables:
    ✓ Value-aligned AI responses
    ✓ Iterative self-critique and improvement
    ✓ Explicit principle enforcement
    ✓ Audit trail of improvements
    ✓ Consistent ethical behavior
    ✓ Safety and compliance guarantees
    ✓ Transparent reasoning process
    
    This pattern is valuable for:
    - Production AI systems requiring safety
    - Content moderation platforms
    - Regulated industries (finance, healthcare)
    - Ethical AI development
    - Bias mitigation
    - Policy-compliant generation
    - High-stakes decision support
        """)
    
    
    if __name__ == "__main__":
        demonstrate_constitutional_chain()
    
        

    Agentic Design Pattern: Constitutional Chain

    Pattern 168: Constitutional Chain – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Constitutional Chain pattern implementation.


    Pattern Description

    Pattern 168: Constitutional Chain

    Description: The Constitutional Chain pattern ensures AI outputs align with predefined principles, values, and rules through iterative self-critique and revision. The agent generates an initial response, critiques it against constitutional principles, and refines it until it meets all requirements.

    Components: 1. Constitution: Set of principles and rules 2. Generator: Creates initial responses 3. Critic: Evaluates responses against constitution 4. Reviser: Improves responses based on critique 5. Compliance Checker: Validates final alignment 6. Iteration Manager: Controls refinement loops

    Use Cases: – Ethical AI systems – Content moderation – Policy-compliant generation – Bias reduction – Safety-critical applications – Regulated industries

    Benefits: – Value alignment – Explicit principles – Iterative improvement – Audit trail – Consistent behavior – Safety guarantees

    Trade-offs: – Multiple LLM calls – Increased latency – May be overly cautious – Requires good constitution – Complexity overhead

    LangChain Implementation: Implements iterative critique-revision loops with LangChain. Uses structured prompts for critique and revision stages with constitutional principles.


    File Statistics

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

    Module Documentation

    """
    Pattern 168: Constitutional Chain
    
    Description:
        The Constitutional Chain pattern ensures AI outputs align with predefined principles,
        values, and rules through iterative self-critique and revision. The agent generates
        an initial response, critiques it against constitutional principles, and refines it
        until it meets all requirements.
    
    Components:
        1. Constitution: Set of principles and rules
        2. Generator: Creates initial responses
        3. Critic: Evaluates responses against constitution
        4. Reviser: Improves responses based on critique
        5. Compliance Checker: Validates final alignment
        6. Iteration Manager: Controls refinement loops
    
    Use Cases:
        - Ethical AI systems
        - Content moderation
        - Policy-compliant generation
        - Bias reduction
        - Safety-critical applications
        - Regulated industries
    
    Benefits:
        - Value alignment
        - Explicit principles
        - Iterative improvement
        - Audit trail
        - Consistent behavior
        - Safety guarantees
    
    Trade-offs:
        - Multiple LLM calls
        - Increased latency
        - May be overly cautious
        - Requires good constitution
        - Complexity overhead
    
    LangChain Implementation:
        Implements iterative critique-revision loops with LangChain. Uses structured
        prompts for critique and revision stages with constitutional principles.
    """
    

    Purpose: This module implements the Constitutional Chain 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.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 – 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: PrincipleType

    Lines 59-67

    Inherits from: Enum

    Purpose:

    Types of constitutional principles
    

    Class: ComplianceLevel

    Lines 70-75

    Inherits from: Enum

    Purpose:

    Compliance assessment levels
    

    Class: ConstitutionalPrinciple

    Lines 79-86

    Purpose:

    Represents a principle in the constitution
    

    Class: CritiqueResult

    Lines 90-96

    Purpose:

    Result of critiquing a response
    

    Class: ConstitutionalResponse

    Lines 100-106

    Purpose:

    Final response with compliance information
    

    Class: Constitution

    Lines 109-136

    Purpose:

    Defines a set of principles for AI behavior
    

    Methods:

    __init__(self, name)

    Lines 112-114

    Purpose: Initialize the class instance with required parameters.

    Parameters: – name: Name

    add_principle(self, principle)

    Lines 116-118

    Purpose:

    Add a principle to the constitution
    

    Parameters: – principle: Principle

    get_principles_text(self)

    Lines 120-136

    Purpose:

    Get formatted text of all principles
    

    Parameters:

    Returns: str


    Class: ConstitutionalChain

    Lines 139-304

    Purpose:

    Implements constitutional AI pattern with critique-revision loops
    

    Methods:

    __init__(self, constitution, max_iterations)

    Lines 142-203

    Purpose:

    Initialize constitutional chain
    
    Args:
        constitution: The constitution to follow
        max_iterations: Maximum number of revision iterations
    

    Parameters: – constitution: Constitution – max_iterations: Max iterations

    generate_constitutional_response(self, query)

    Lines 205-247

    Purpose:

    Generate a response that complies with constitutional principles
    

    Parameters: – query: Input query or question

    Returns: ConstitutionalResponse

    _critique_response(self, query, response, principles)

    Lines 249-295

    Purpose:

    Critique a response against constitutional principles
    

    Parameters: – query: Input query or question – response: Response – principles: Principles

    Returns: CritiqueResult

    _revise_response(self, query, response, critique)

    Lines 297-304

    Purpose:

    Revise response based on critique
    

    Parameters: – query: Input query or question – response: Response – critique: Critique

    Returns: str


    Functions

    demonstrate_constitutional_chain()

    Lines 307-527

    Purpose:

    Demonstrate constitutional chain pattern
    

    Key Concepts Summary

    Pattern Overview

    The Constitutional Chain 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 167
    • Pattern 169

    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 Constitutional Chain 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: 168_constitutional_chain.py
    Total Lines: 532
    Pattern Category: AI Agent Design Pattern #168


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Agentic Workflows

    
        """
    Pattern 167: Agentic Workflows
    
    Description:
        The Agentic Workflows pattern implements complex, multi-step processes where agents
        autonomously navigate through workflow states, make decisions, handle branching logic,
        and coordinate multiple tasks. It uses state machines and graph-based execution for
        sophisticated task orchestration.
    
    Components:
        1. Workflow Definition: Graph structure with states and transitions
        2. State Manager: Tracks current workflow state
        3. Decision Engine: Makes branching decisions
        4. Task Executor: Executes tasks at each state
        5. Condition Evaluator: Evaluates transition conditions
        6. Error Handler: Manages workflow failures
    
    Use Cases:
        - Multi-step business processes
        - Complex data pipelines
        - Decision trees and workflows
        - Approval workflows
        - Multi-stage content generation
        - Automated testing workflows
    
    Benefits:
        - Complex process automation
        - Clear workflow visualization
        - Conditional branching support
        - State persistence
        - Error recovery
        - Audit trail
    
    Trade-offs:
        - Increased complexity
        - State management overhead
        - Debugging challenges
        - Performance considerations
        - Requires workflow design
    
    LangChain Implementation:
        Uses LangGraph-inspired state machines with LangChain agents for task execution.
        Implements graph-based workflows with conditional branching and state management.
    """
    
    import os
    from typing import List, Dict, Any, Optional, Callable
    from dataclasses import dataclass, field
    from enum import Enum
    from datetime import datetime
    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 WorkflowStatus(Enum):
        """Status of workflow execution"""
        NOT_STARTED = "not_started"
        IN_PROGRESS = "in_progress"
        COMPLETED = "completed"
        FAILED = "failed"
        PAUSED = "paused"
    
    
    class TransitionType(Enum):
        """Types of state transitions"""
        AUTOMATIC = "automatic"
        CONDITIONAL = "conditional"
        MANUAL = "manual"
    
    
    @dataclass
    class WorkflowState:
        """Represents a state in the workflow"""
        name: str
        description: str
        task: Optional[Callable] = None
        is_terminal: bool = False
        metadata: Dict[str, Any] = field(default_factory=dict)
    
    
    @dataclass
    class StateTransition:
        """Represents a transition between states"""
        from_state: str
        to_state: str
        condition: Optional[Callable] = None
        transition_type: TransitionType = TransitionType.AUTOMATIC
        description: str = ""
    
    
    @dataclass
    class WorkflowContext:
        """Context passed through workflow execution"""
        data: Dict[str, Any] = field(default_factory=dict)
        history: List[str] = field(default_factory=list)
        metadata: Dict[str, Any] = field(default_factory=dict)
        
        def add_result(self, state: str, result: Any):
            """Add result from a state execution"""
            self.data[f"{state}_result"] = result
            self.history.append(state)
        
        def get_result(self, state: str) -> Any:
            """Get result from previous state"""
            return self.data.get(f"{state}_result")
    
    
    class AgenticWorkflow:
        """Implements agentic workflow pattern with state machines"""
        
        def __init__(self, name: str):
            """Initialize workflow"""
            self.name = name
            self.states: Dict[str, WorkflowState] = {}
            self.transitions: List[StateTransition] = []
            self.start_state: Optional[str] = None
            self.current_state: Optional[str] = None
            self.status = WorkflowStatus.NOT_STARTED
            self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
            
            # Execution log
            self.execution_log: List[Dict[str, Any]] = []
        
        def add_state(self, state: WorkflowState, is_start: bool = False):
            """Add a state to the workflow"""
            self.states[state.name] = state
            if is_start:
                self.start_state = state.name
        
        def add_transition(self, transition: StateTransition):
            """Add a transition between states"""
            if transition.from_state not in self.states:
                raise ValueError(f"Unknown from_state: {transition.from_state}")
            if transition.to_state not in self.states:
                raise ValueError(f"Unknown to_state: {transition.to_state}")
            self.transitions.append(transition)
        
        def execute(self, initial_context: Optional[WorkflowContext] = None) -> WorkflowContext:
            """Execute the workflow"""
            if not self.start_state:
                raise ValueError("No start state defined")
            
            context = initial_context or WorkflowContext()
            self.current_state = self.start_state
            self.status = WorkflowStatus.IN_PROGRESS
            
            self._log_execution("workflow_started", {
                "start_state": self.start_state,
                "timestamp": datetime.now().isoformat()
            })
            
            try:
                while True:
                    state = self.states[self.current_state]
                    
                    self._log_execution("state_entered", {
                        "state": self.current_state,
                        "timestamp": datetime.now().isoformat()
                    })
                    
                    # Execute state task
                    if state.task:
                        try:
                            result = state.task(context, self.llm)
                            context.add_result(self.current_state, result)
                            
                            self._log_execution("state_executed", {
                                "state": self.current_state,
                                "result": str(result)[:100]
                            })
                        except Exception as e:
                            self._log_execution("state_error", {
                                "state": self.current_state,
                                "error": str(e)
                            })
                            self.status = WorkflowStatus.FAILED
                            raise
                    
                    # Check if terminal state
                    if state.is_terminal:
                        self.status = WorkflowStatus.COMPLETED
                        self._log_execution("workflow_completed", {
                            "final_state": self.current_state,
                            "timestamp": datetime.now().isoformat()
                        })
                        break
                    
                    # Find next state
                    next_state = self._get_next_state(context)
                    
                    if next_state is None:
                        self.status = WorkflowStatus.FAILED
                        raise ValueError(f"No valid transition from state: {self.current_state}")
                    
                    self._log_execution("transition", {
                        "from": self.current_state,
                        "to": next_state,
                        "timestamp": datetime.now().isoformat()
                    })
                    
                    self.current_state = next_state
            
            except Exception as e:
                self.status = WorkflowStatus.FAILED
                self._log_execution("workflow_failed", {
                    "error": str(e),
                    "timestamp": datetime.now().isoformat()
                })
                raise
            
            return context
        
        def _get_next_state(self, context: WorkflowContext) -> Optional[str]:
            """Determine next state based on transitions"""
            # Find applicable transitions from current state
            applicable = [
                t for t in self.transitions 
                if t.from_state == self.current_state
            ]
            
            if not applicable:
                return None
            
            # Evaluate conditions for conditional transitions
            for transition in applicable:
                if transition.transition_type == TransitionType.AUTOMATIC:
                    return transition.to_state
                elif transition.transition_type == TransitionType.CONDITIONAL:
                    if transition.condition and transition.condition(context):
                        return transition.to_state
            
            # If no condition matched, try first automatic transition
            automatic = [t for t in applicable if t.transition_type == TransitionType.AUTOMATIC]
            if automatic:
                return automatic[0].to_state
            
            return None
        
        def _log_execution(self, event_type: str, data: Dict[str, Any]):
            """Log workflow execution event"""
            self.execution_log.append({
                "event": event_type,
                "data": data
            })
        
        def get_execution_summary(self) -> Dict[str, Any]:
            """Get summary of workflow execution"""
            return {
                "workflow_name": self.name,
                "status": self.status.value,
                "states_visited": len(set(e["data"].get("state", "") for e in self.execution_log)),
                "total_events": len(self.execution_log),
                "current_state": self.current_state
            }
        
        def visualize(self) -> str:
            """Create text visualization of workflow"""
            lines = [f"Workflow: {self.name}", "=" * 60, ""]
            
            lines.append("States:")
            for name, state in self.states.items():
                marker = "→" if name == self.start_state else " "
                terminal = " (terminal)" if state.is_terminal else ""
                lines.append(f"  {marker} {name}{terminal}: {state.description}")
            
            lines.append("\nTransitions:")
            for trans in self.transitions:
                cond = f" [conditional]" if trans.transition_type == TransitionType.CONDITIONAL else ""
                lines.append(f"  {trans.from_state} -> {trans.to_state}{cond}")
                if trans.description:
                    lines.append(f"    {trans.description}")
            
            return "\n".join(lines)
    
    
    def demonstrate_agentic_workflows():
        """Demonstrate agentic workflows"""
        print("=" * 80)
        print("AGENTIC WORKFLOWS PATTERN DEMONSTRATION")
        print("=" * 80)
        
        # Example 1: Content generation workflow
        print("\n" + "=" * 80)
        print("Example 1: Content Generation Workflow")
        print("=" * 80)
        
        def research_task(context: WorkflowContext, llm: ChatOpenAI) -> str:
            """Research phase"""
            topic = context.data.get("topic", "AI")
            prompt = f"List 3 key points about {topic} for a blog post"
            chain = ChatPromptTemplate.from_messages([
                ("user", prompt)
            ]) | llm | StrOutputParser()
            return chain.invoke({})
        
        def outline_task(context: WorkflowContext, llm: ChatOpenAI) -> str:
            """Outline creation"""
            research = context.get_result("research")
            prompt = f"Create a blog post outline based on: {research}"
            chain = ChatPromptTemplate.from_messages([
                ("user", prompt)
            ]) | llm | StrOutputParser()
            return chain.invoke({})
        
        def write_task(context: WorkflowContext, llm: ChatOpenAI) -> str:
            """Writing phase"""
            outline = context.get_result("outline")
            prompt = f"Write the first paragraph based on: {outline}"
            chain = ChatPromptTemplate.from_messages([
                ("user", prompt)
            ]) | llm | StrOutputParser()
            return chain.invoke({})
        
        def review_task(context: WorkflowContext, llm: ChatOpenAI) -> str:
            """Review phase"""
            draft = context.get_result("write")
            prompt = f"Review and provide feedback on: {draft}"
            chain = ChatPromptTemplate.from_messages([
                ("user", prompt)
            ]) | llm | StrOutputParser()
            return chain.invoke({})
        
        # Create workflow
        content_workflow = AgenticWorkflow("Content Generation")
        
        # Add states
        content_workflow.add_state(WorkflowState(
            "research",
            "Research topic and gather key points",
            task=research_task
        ), is_start=True)
        
        content_workflow.add_state(WorkflowState(
            "outline",
            "Create content outline",
            task=outline_task
        ))
        
        content_workflow.add_state(WorkflowState(
            "write",
            "Write content draft",
            task=write_task
        ))
        
        content_workflow.add_state(WorkflowState(
            "review",
            "Review and provide feedback",
            task=review_task
        ))
        
        content_workflow.add_state(WorkflowState(
            "complete",
            "Content generation complete",
            is_terminal=True
        ))
        
        # Add transitions
        content_workflow.add_transition(StateTransition(
            "research", "outline", 
            transition_type=TransitionType.AUTOMATIC
        ))
        content_workflow.add_transition(StateTransition(
            "outline", "write",
            transition_type=TransitionType.AUTOMATIC
        ))
        content_workflow.add_transition(StateTransition(
            "write", "review",
            transition_type=TransitionType.AUTOMATIC
        ))
        content_workflow.add_transition(StateTransition(
            "review", "complete",
            transition_type=TransitionType.AUTOMATIC
        ))
        
        # Visualize workflow
        print("\nWorkflow Structure:")
        print(content_workflow.visualize())
        
        # Execute workflow
        print("\n" + "-" * 60)
        print("Executing Workflow...")
        print("-" * 60)
        
        initial_context = WorkflowContext(data={"topic": "Machine Learning"})
        
        try:
            result_context = content_workflow.execute(initial_context)
            
            print("\nWorkflow Execution Summary:")
            summary = content_workflow.get_execution_summary()
            print(f"  Status: {summary['status']}")
            print(f"  States visited: {summary['states_visited']}")
            print(f"  Total events: {summary['total_events']}")
            
            print("\nState Execution Path:")
            for state in result_context.history:
                print(f"  → {state}")
            
            print("\nFinal Results:")
            print(f"  Research: {result_context.get_result('research')[:80]}...")
            print(f"  Outline: {result_context.get_result('outline')[:80]}...")
            print(f"  Draft: {result_context.get_result('write')[:80]}...")
            print(f"  Review: {result_context.get_result('review')[:80]}...")
            
        except Exception as e:
            print(f"Workflow failed: {e}")
        
        # Example 2: Conditional branching workflow
        print("\n" + "=" * 80)
        print("Example 2: Conditional Branching Workflow")
        print("=" * 80)
        
        def analyze_task(context: WorkflowContext, llm: ChatOpenAI) -> Dict[str, Any]:
            """Analyze input"""
            text = context.data.get("text", "")
            prompt = f"Is this text positive or negative sentiment? Answer with just 'positive' or 'negative': {text}"
            chain = ChatPromptTemplate.from_messages([
                ("user", prompt)
            ]) | llm | StrOutputParser()
            sentiment = chain.invoke({}).strip().lower()
            return {"sentiment": sentiment}
        
        def positive_response_task(context: WorkflowContext, llm: ChatOpenAI) -> str:
            """Handle positive sentiment"""
            return "Thank you for the positive feedback!"
        
        def negative_response_task(context: WorkflowContext, llm: ChatOpenAI) -> str:
            """Handle negative sentiment"""
            return "We're sorry to hear that. How can we improve?"
        
        # Create branching workflow
        branching_workflow = AgenticWorkflow("Sentiment Analysis")
        
        # Add states
        branching_workflow.add_state(WorkflowState(
            "analyze",
            "Analyze sentiment",
            task=analyze_task
        ), is_start=True)
        
        branching_workflow.add_state(WorkflowState(
            "positive_response",
            "Generate positive response",
            task=positive_response_task
        ))
        
        branching_workflow.add_state(WorkflowState(
            "negative_response",
            "Generate negative response",
            task=negative_response_task
        ))
        
        branching_workflow.add_state(WorkflowState(
            "end",
            "End state",
            is_terminal=True
        ))
        
        # Add conditional transitions
        branching_workflow.add_transition(StateTransition(
            "analyze", "positive_response",
            condition=lambda ctx: ctx.get_result("analyze").get("sentiment") == "positive",
            transition_type=TransitionType.CONDITIONAL,
            description="If sentiment is positive"
        ))
        
        branching_workflow.add_transition(StateTransition(
            "analyze", "negative_response",
            condition=lambda ctx: ctx.get_result("analyze").get("sentiment") == "negative",
            transition_type=TransitionType.CONDITIONAL,
            description="If sentiment is negative"
        ))
        
        branching_workflow.add_transition(StateTransition(
            "positive_response", "end",
            transition_type=TransitionType.AUTOMATIC
        ))
        
        branching_workflow.add_transition(StateTransition(
            "negative_response", "end",
            transition_type=TransitionType.AUTOMATIC
        ))
        
        # Visualize
        print("\nWorkflow Structure:")
        print(branching_workflow.visualize())
        
        # Execute with positive text
        print("\n" + "-" * 60)
        print("Test 1: Positive Text")
        print("-" * 60)
        
        context1 = WorkflowContext(data={"text": "This product is amazing! I love it."})
        result1 = branching_workflow.execute(context1)
        
        print(f"Path taken: {' → '.join(result1.history)}")
        print(f"Response: {result1.get_result('positive_response')}")
        
        # Execute with negative text
        print("\n" + "-" * 60)
        print("Test 2: Negative Text")
        print("-" * 60)
        
        branching_workflow.status = WorkflowStatus.NOT_STARTED  # Reset
        context2 = WorkflowContext(data={"text": "This is terrible. Very disappointed."})
        result2 = branching_workflow.execute(context2)
        
        print(f"Path taken: {' → '.join(result2.history)}")
        print(f"Response: {result2.get_result('negative_response')}")
        
        # Summary
        print("\n" + "=" * 80)
        print("SUMMARY")
        print("=" * 80)
        print("""
    The Agentic Workflows pattern enables:
    ✓ Complex multi-step process automation
    ✓ Conditional branching based on runtime state
    ✓ State-based execution with context passing
    ✓ Clear workflow visualization
    ✓ Execution logging and audit trails
    ✓ Error handling and recovery
    ✓ Flexible workflow composition
    
    This pattern is valuable for:
    - Business process automation
    - Multi-stage content generation
    - Decision tree execution
    - Approval workflows
    - Complex data pipelines
    - Testing and validation workflows
    - Orchestrating multiple agents
        """)
    
    
    if __name__ == "__main__":
        demonstrate_agentic_workflows()
    
        

    Agentic Design Pattern: Agentic Workflows

    Pattern 167: Agentic Workflows – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Agentic Workflows pattern implementation.


    Pattern Description

    Pattern 167: Agentic Workflows

    Description: The Agentic Workflows pattern implements complex, multi-step processes where agents autonomously navigate through workflow states, make decisions, handle branching logic, and coordinate multiple tasks. It uses state machines and graph-based execution for sophisticated task orchestration.

    Components: 1. Workflow Definition: Graph structure with states and transitions 2. State Manager: Tracks current workflow state 3. Decision Engine: Makes branching decisions 4. Task Executor: Executes tasks at each state 5. Condition Evaluator: Evaluates transition conditions 6. Error Handler: Manages workflow failures

    Use Cases: – Multi-step business processes – Complex data pipelines – Decision trees and workflows – Approval workflows – Multi-stage content generation – Automated testing workflows

    Benefits: – Complex process automation – Clear workflow visualization – Conditional branching support – State persistence – Error recovery – Audit trail

    Trade-offs: – Increased complexity – State management overhead – Debugging challenges – Performance considerations – Requires workflow design

    LangChain Implementation: Uses LangGraph-inspired state machines with LangChain agents for task execution. Implements graph-based workflows with conditional branching and state management.


    File Statistics

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

    Module Documentation

    """
    Pattern 167: Agentic Workflows
    
    Description:
        The Agentic Workflows pattern implements complex, multi-step processes where agents
        autonomously navigate through workflow states, make decisions, handle branching logic,
        and coordinate multiple tasks. It uses state machines and graph-based execution for
        sophisticated task orchestration.
    
    Components:
        1. Workflow Definition: Graph structure with states and transitions
        2. State Manager: Tracks current workflow state
        3. Decision Engine: Makes branching decisions
        4. Task Executor: Executes tasks at each state
        5. Condition Evaluator: Evaluates transition conditions
        6. Error Handler: Manages workflow failures
    
    Use Cases:
        - Multi-step business processes
        - Complex data pipelines
        - Decision trees and workflows
        - Approval workflows
        - Multi-stage content generation
        - Automated testing workflows
    
    Benefits:
        - Complex process automation
        - Clear workflow visualization
        - Conditional branching support
        - State persistence
        - Error recovery
        - Audit trail
    
    Trade-offs:
        - Increased complexity
        - State management overhead
        - Debugging challenges
        - Performance considerations
        - Requires workflow design
    
    LangChain Implementation:
        Uses LangGraph-inspired state machines with LangChain agents for task execution.
        Implements graph-based workflows with conditional branching and state management.
    """
    

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


    Import Statements

    Standard Library Imports

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

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

    LangChain Imports

    from langchain_openai.ChatOpenAI
    

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

    Other Imports

    from dotenv.load_dotenv
    

    Purpose: Additional dependencies for specialized functionality.


    Classes

    Class: WorkflowStatus

    Lines 60-66

    Inherits from: Enum

    Purpose:

    Status of workflow execution
    

    Class: TransitionType

    Lines 69-73

    Inherits from: Enum

    Purpose:

    Types of state transitions
    

    Class: WorkflowState

    Lines 77-83

    Purpose:

    Represents a state in the workflow
    

    Class: StateTransition

    Lines 87-93

    Purpose:

    Represents a transition between states
    

    Class: WorkflowContext

    Lines 97-110

    Purpose:

    Context passed through workflow execution
    

    Methods:

    add_result(self, state, result)

    Lines 103-106

    Purpose:

    Add result from a state execution
    

    Parameters: – state: State – result: Result

    get_result(self, state)

    Lines 108-110

    Purpose:

    Get result from previous state
    

    Parameters: – state: State

    Returns: Any


    Class: AgenticWorkflow

    Lines 113-278

    Purpose:

    Implements agentic workflow pattern with state machines
    

    Methods:

    __init__(self, name)

    Lines 116-127

    Purpose:

    Initialize workflow
    

    Parameters: – name: Name

    add_state(self, state, is_start)

    Lines 129-133

    Purpose:

    Add a state to the workflow
    

    Parameters: – state: State – is_start: Is start

    add_transition(self, transition)

    Lines 135-141

    Purpose:

    Add a transition between states
    

    Parameters: – transition: Transition

    execute(self, initial_context)

    Lines 143-216

    Purpose:

    Execute the workflow
    

    Parameters: – initial_context: Initial context

    Returns: WorkflowContext

    _get_next_state(self, context)

    Lines 218-242

    Purpose:

    Determine next state based on transitions
    

    Parameters: – context: Context

    Returns: <ast.Subscript object at 0x000001A06D5F5B90>

    _log_execution(self, event_type, data)

    Lines 244-249

    Purpose:

    Log workflow execution event
    

    Parameters: – event_type: Event type – data: Data

    get_execution_summary(self)

    Lines 251-259

    Purpose:

    Get summary of workflow execution
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D83F710>

    visualize(self)

    Lines 261-278

    Purpose:

    Create text visualization of workflow
    

    Parameters:

    Returns: str


    Functions

    demonstrate_agentic_workflows()

    Lines 281-538

    Purpose:

    Demonstrate agentic workflows
    

    Key Concepts Summary

    Pattern Overview

    The Agentic Workflows 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 166
    • Pattern 168

    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 Agentic Workflows 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: 167_agentic_workflows.py
    Total Lines: 543
    Pattern Category: AI Agent Design Pattern #167


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Prompt Caching And Reuse

    
        """
    Pattern 166: Prompt Caching & Reuse
    
    Description:
        The Prompt Caching & Reuse pattern stores and retrieves prompts and their results
        to avoid redundant LLM calls, reduce latency, and save costs. It uses semantic
        similarity to match queries with cached results and manages cache invalidation
        and updates.
    
    Components:
        1. Cache Storage: Stores prompts and results
        2. Similarity Matcher: Finds semantically similar cached prompts
        3. Cache Manager: Handles cache lifecycle
        4. Invalidation Strategy: Manages cache freshness
        5. Hit/Miss Tracker: Monitors cache performance
        6. Compression: Optimizes cache storage
    
    Use Cases:
        - Repeated query handling
        - Cost optimization
        - Latency reduction
        - Common question answering
        - Template-based generation
        - Batch processing optimization
    
    Benefits:
        - Reduced API costs
        - Lower latency
        - Better resource utilization
        - Consistent responses
        - Offline capability
    
    Trade-offs:
        - Storage overhead
        - Cache invalidation complexity
        - Stale data risk
        - Memory usage
        - Similarity matching overhead
    
    LangChain Implementation:
        Combines semantic similarity search with LangChain caching mechanisms.
        Uses embeddings for matching similar queries efficiently.
    """
    
    import os
    import json
    import hashlib
    import time
    from typing import List, Dict, Any, Optional, Tuple
    from dataclasses import dataclass, field, asdict
    from datetime import datetime, timedelta
    from collections import defaultdict
    from dotenv import load_dotenv
    
    from langchain_openai import ChatOpenAI, OpenAIEmbeddings
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser
    
    load_dotenv()
    
    
    @dataclass
    class CacheEntry:
        """Represents a cached prompt and response"""
        key: str
        prompt: str
        response: str
        timestamp: datetime
        hit_count: int = 0
        embedding: Optional[List[float]] = None
        metadata: Dict[str, Any] = field(default_factory=dict)
        ttl: Optional[int] = None  # Time-to-live in seconds
        
        def is_expired(self) -> bool:
            """Check if cache entry has expired"""
            if self.ttl is None:
                return False
            age = (datetime.now() - self.timestamp).total_seconds()
            return age > self.ttl
        
        def to_dict(self) -> Dict[str, Any]:
            """Convert to dictionary for serialization"""
            data = asdict(self)
            data['timestamp'] = self.timestamp.isoformat()
            return data
        
        @staticmethod
        def from_dict(data: Dict[str, Any]) -> 'CacheEntry':
            """Create from dictionary"""
            data['timestamp'] = datetime.fromisoformat(data['timestamp'])
            return CacheEntry(**data)
    
    
    class PromptCacheManager:
        """Manages prompt caching and reuse with semantic similarity matching"""
        
        def __init__(self, 
                     similarity_threshold: float = 0.85,
                     max_cache_size: int = 1000,
                     default_ttl: Optional[int] = None):
            """
            Initialize prompt cache manager
            
            Args:
                similarity_threshold: Minimum similarity score for cache hit (0-1)
                max_cache_size: Maximum number of entries to cache
                default_ttl: Default time-to-live in seconds
            """
            self.similarity_threshold = similarity_threshold
            self.max_cache_size = max_cache_size
            self.default_ttl = default_ttl
            
            self.cache: Dict[str, CacheEntry] = {}
            self.embeddings = OpenAIEmbeddings()
            self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
            
            # Statistics
            self.stats = {
                "total_requests": 0,
                "cache_hits": 0,
                "cache_misses": 0,
                "total_cost_saved": 0.0,
                "total_time_saved": 0.0
            }
        
        def get_or_generate(self, prompt: str, 
                           force_refresh: bool = False,
                           ttl: Optional[int] = None) -> Tuple[str, bool]:
            """
            Get cached response or generate new one
            
            Args:
                prompt: The prompt to process
                force_refresh: Skip cache and generate new response
                ttl: Time-to-live for this entry
                
            Returns:
                Tuple of (response, was_cached)
            """
            self.stats["total_requests"] += 1
            
            if not force_refresh:
                # Try to find in cache
                cached_response = self._find_similar_cached(prompt)
                if cached_response:
                    self.stats["cache_hits"] += 1
                    self.stats["total_cost_saved"] += 0.002  # Estimated savings
                    self.stats["total_time_saved"] += 0.5  # Estimated time saved
                    return cached_response, True
            
            # Cache miss - generate new response
            self.stats["cache_misses"] += 1
            response = self._generate_response(prompt)
            
            # Cache the result
            self._add_to_cache(prompt, response, ttl or self.default_ttl)
            
            return response, False
        
        def batch_get_or_generate(self, prompts: List[str]) -> List[Tuple[str, bool]]:
            """Process multiple prompts efficiently"""
            results = []
            for prompt in prompts:
                result, cached = self.get_or_generate(prompt)
                results.append((result, cached))
            return results
        
        def invalidate(self, pattern: Optional[str] = None):
            """Invalidate cache entries matching pattern"""
            if pattern is None:
                # Clear all
                self.cache.clear()
            else:
                # Remove matching entries
                keys_to_remove = [
                    key for key, entry in self.cache.items()
                    if pattern.lower() in entry.prompt.lower()
                ]
                for key in keys_to_remove:
                    del self.cache[key]
        
        def _find_similar_cached(self, prompt: str) -> Optional[str]:
            """Find semantically similar cached prompt"""
            if not self.cache:
                return None
            
            # Clean expired entries
            self._clean_expired()
            
            # Get embedding for query prompt
            query_embedding = self.embeddings.embed_query(prompt)
            
            # Find most similar cached entry
            best_match = None
            best_similarity = 0.0
            
            for entry in self.cache.values():
                if entry.embedding is None:
                    continue
                
                # Calculate cosine similarity
                similarity = self._cosine_similarity(query_embedding, entry.embedding)
                
                if similarity > best_similarity and similarity >= self.similarity_threshold:
                    best_similarity = similarity
                    best_match = entry
            
            if best_match:
                best_match.hit_count += 1
                return best_match.response
            
            return None
        
        def _generate_response(self, prompt: str) -> str:
            """Generate new response using LLM"""
            chain = ChatPromptTemplate.from_messages([
                ("user", "{prompt}")
            ]) | self.llm | StrOutputParser()
            
            response = chain.invoke({"prompt": prompt})
            return response
        
        def _add_to_cache(self, prompt: str, response: str, ttl: Optional[int] = None):
            """Add entry to cache"""
            # Generate cache key
            key = self._generate_key(prompt)
            
            # Get embedding
            embedding = self.embeddings.embed_query(prompt)
            
            # Create entry
            entry = CacheEntry(
                key=key,
                prompt=prompt,
                response=response,
                timestamp=datetime.now(),
                embedding=embedding,
                ttl=ttl
            )
            
            # Check cache size
            if len(self.cache) >= self.max_cache_size:
                self._evict_entry()
            
            # Add to cache
            self.cache[key] = entry
        
        def _generate_key(self, prompt: str) -> str:
            """Generate cache key from prompt"""
            return hashlib.sha256(prompt.encode()).hexdigest()
        
        def _cosine_similarity(self, vec1: List[float], vec2: List[float]) -> float:
            """Calculate cosine similarity between two vectors"""
            dot_product = sum(a * b for a, b in zip(vec1, vec2))
            magnitude1 = sum(a * a for a in vec1) ** 0.5
            magnitude2 = sum(b * b for b in vec2) ** 0.5
            
            if magnitude1 == 0 or magnitude2 == 0:
                return 0.0
            
            return dot_product / (magnitude1 * magnitude2)
        
        def _clean_expired(self):
            """Remove expired entries"""
            keys_to_remove = [
                key for key, entry in self.cache.items()
                if entry.is_expired()
            ]
            for key in keys_to_remove:
                del self.cache[key]
        
        def _evict_entry(self):
            """Evict least recently used entry"""
            if not self.cache:
                return
            
            # Find entry with lowest hit count and oldest timestamp
            lru_key = min(
                self.cache.keys(),
                key=lambda k: (self.cache[k].hit_count, self.cache[k].timestamp)
            )
            del self.cache[lru_key]
        
        def get_stats(self) -> Dict[str, Any]:
            """Get cache statistics"""
            total = self.stats["total_requests"]
            hit_rate = self.stats["cache_hits"] / total if total > 0 else 0
            
            return {
                **self.stats,
                "cache_hit_rate": hit_rate,
                "cache_size": len(self.cache),
                "avg_hit_count": sum(e.hit_count for e in self.cache.values()) / len(self.cache)
                                if self.cache else 0
            }
        
        def get_popular_queries(self, top_n: int = 10) -> List[Dict[str, Any]]:
            """Get most frequently accessed cached queries"""
            sorted_entries = sorted(
                self.cache.values(),
                key=lambda e: e.hit_count,
                reverse=True
            )[:top_n]
            
            return [
                {
                    "prompt": entry.prompt,
                    "hit_count": entry.hit_count,
                    "age_seconds": (datetime.now() - entry.timestamp).total_seconds()
                }
                for entry in sorted_entries
            ]
        
        def save_cache(self, filepath: str):
            """Save cache to file"""
            cache_data = {
                key: entry.to_dict()
                for key, entry in self.cache.items()
            }
            
            with open(filepath, 'w') as f:
                json.dump(cache_data, f, indent=2)
        
        def load_cache(self, filepath: str):
            """Load cache from file"""
            try:
                with open(filepath, 'r') as f:
                    cache_data = json.load(f)
                
                self.cache = {
                    key: CacheEntry.from_dict(data)
                    for key, data in cache_data.items()
                }
            except FileNotFoundError:
                print(f"Cache file {filepath} not found")
            except Exception as e:
                print(f"Error loading cache: {e}")
    
    
    def demonstrate_prompt_caching():
        """Demonstrate prompt caching and reuse"""
        print("=" * 80)
        print("PROMPT CACHING & REUSE PATTERN DEMONSTRATION")
        print("=" * 80)
        
        cache_manager = PromptCacheManager(
            similarity_threshold=0.85,
            max_cache_size=100,
            default_ttl=3600  # 1 hour
        )
        
        # Example 1: Basic caching
        print("\n" + "=" * 80)
        print("Example 1: Basic Caching")
        print("=" * 80)
        
        prompts = [
            "What is the capital of France?",
            "Tell me about the capital of France",  # Similar query
            "What is the capital of Germany?",
        ]
        
        print("\nProcessing prompts with caching:")
        for i, prompt in enumerate(prompts, 1):
            print(f"\n{i}. Query: {prompt}")
            response, was_cached = cache_manager.get_or_generate(prompt)
            status = "CACHE HIT" if was_cached else "CACHE MISS"
            print(f"   Status: {status}")
            print(f"   Response: {response[:100]}...")
        
        # Example 2: Similar query matching
        print("\n" + "=" * 80)
        print("Example 2: Semantic Similarity Matching")
        print("=" * 80)
        
        original = "Explain quantum computing"
        similar = "Can you describe quantum computing?"
        
        print(f"\nOriginal query: {original}")
        response1, cached1 = cache_manager.get_or_generate(original)
        print(f"Status: {'CACHED' if cached1 else 'GENERATED'}")
        print(f"Response: {response1[:100]}...")
        
        print(f"\nSimilar query: {similar}")
        response2, cached2 = cache_manager.get_or_generate(similar)
        print(f"Status: {'CACHED' if cached2 else 'GENERATED'}")
        print(f"Match found: {cached2}")
        if cached2:
            print("✓ Similar query matched with cached result!")
        
        # Example 3: Batch processing
        print("\n" + "=" * 80)
        print("Example 3: Batch Processing with Cache")
        print("=" * 80)
        
        batch_prompts = [
            "What is machine learning?",
            "Define machine learning",  # Similar
            "What is deep learning?",
            "Explain deep learning",  # Similar
            "What is machine learning?",  # Exact duplicate
        ]
        
        print(f"\nProcessing batch of {len(batch_prompts)} prompts:")
        start_time = time.time()
        results = cache_manager.batch_get_or_generate(batch_prompts)
        elapsed = time.time() - start_time
        
        hits = sum(1 for _, cached in results if cached)
        misses = len(results) - hits
        
        print(f"\nResults:")
        print(f"  Total queries: {len(results)}")
        print(f"  Cache hits: {hits}")
        print(f"  Cache misses: {misses}")
        print(f"  Hit rate: {hits/len(results):.1%}")
        print(f"  Processing time: {elapsed:.2f}s")
        
        # Example 4: Cache statistics
        print("\n" + "=" * 80)
        print("Example 4: Cache Statistics")
        print("=" * 80)
        
        stats = cache_manager.get_stats()
        
        print("\nCache Performance:")
        print(f"  Total requests: {stats['total_requests']}")
        print(f"  Cache hits: {stats['cache_hits']}")
        print(f"  Cache misses: {stats['cache_misses']}")
        print(f"  Hit rate: {stats['cache_hit_rate']:.1%}")
        print(f"  Cache size: {stats['cache_size']} entries")
        print(f"  Estimated cost saved: ${stats['total_cost_saved']:.4f}")
        print(f"  Estimated time saved: {stats['total_time_saved']:.2f}s")
        
        # Example 5: Popular queries
        print("\n" + "=" * 80)
        print("Example 5: Popular Queries Analysis")
        print("=" * 80)
        
        popular = cache_manager.get_popular_queries(top_n=5)
        
        print("\nMost frequently accessed queries:")
        for i, query_info in enumerate(popular, 1):
            age_min = query_info['age_seconds'] / 60
            print(f"\n{i}. Hit count: {query_info['hit_count']}")
            print(f"   Age: {age_min:.1f} minutes")
            print(f"   Query: {query_info['prompt'][:60]}...")
        
        # Example 6: TTL and expiration
        print("\n" + "=" * 80)
        print("Example 6: Time-to-Live (TTL) Management")
        print("=" * 80)
        
        # Add entry with short TTL
        short_ttl_prompt = "What's the current time?"
        response, cached = cache_manager.get_or_generate(short_ttl_prompt, ttl=2)
        print(f"\nAdded query with 2 second TTL: {short_ttl_prompt}")
        print(f"Response: {response[:80]}...")
        
        print("\nImmediate retry (should hit cache):")
        response, cached = cache_manager.get_or_generate(short_ttl_prompt)
        print(f"Cached: {cached}")
        
        print("\nWaiting 3 seconds for expiration...")
        time.sleep(3)
        
        print("Retry after expiration (should miss cache):")
        response, cached = cache_manager.get_or_generate(short_ttl_prompt)
        print(f"Cached: {cached}")
        
        # Example 7: Cache invalidation
        print("\n" + "=" * 80)
        print("Example 7: Cache Invalidation")
        print("=" * 80)
        
        print(f"\nCache size before invalidation: {len(cache_manager.cache)}")
        
        # Invalidate specific pattern
        cache_manager.invalidate(pattern="capital")
        print(f"Cache size after invalidating 'capital' queries: {len(cache_manager.cache)}")
        
        # Test that invalidated query is regenerated
        test_prompt = "What is the capital of France?"
        response, cached = cache_manager.get_or_generate(test_prompt)
        print(f"\nRetrying invalidated query: {test_prompt}")
        print(f"Cached: {cached} (should be False)")
        
        # Final statistics
        print("\n" + "=" * 80)
        print("Final Statistics")
        print("=" * 80)
        
        final_stats = cache_manager.get_stats()
        print(f"\nOverall Performance:")
        print(f"  Total requests processed: {final_stats['total_requests']}")
        print(f"  Overall hit rate: {final_stats['cache_hit_rate']:.1%}")
        print(f"  Final cache size: {final_stats['cache_size']} entries")
        print(f"  Total cost saved: ${final_stats['total_cost_saved']:.4f}")
        print(f"  Total time saved: {final_stats['total_time_saved']:.2f}s")
        
        # Summary
        print("\n" + "=" * 80)
        print("SUMMARY")
        print("=" * 80)
        print("""
    The Prompt Caching & Reuse pattern enables:
    ✓ Semantic similarity matching for cache hits
    ✓ Significant cost reduction through reuse
    ✓ Lower latency for repeated queries
    ✓ TTL-based cache expiration
    ✓ Cache invalidation strategies
    ✓ Performance monitoring and analytics
    ✓ Batch processing optimization
    
    This pattern is valuable for:
    - High-traffic applications with repeated queries
    - Cost-sensitive deployments
    - Latency-critical systems
    - Common question answering systems
    - Template-based generation
    - Offline or low-connectivity scenarios
        """)
    
    
    if __name__ == "__main__":
        demonstrate_prompt_caching()
    
        

    Agentic Design Pattern: Prompt Caching And Reuse

    Pattern 166: Prompt Caching & Reuse – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Prompt Caching & Reuse pattern implementation.


    Pattern Description

    Pattern 166: Prompt Caching & Reuse

    Description: The Prompt Caching & Reuse pattern stores and retrieves prompts and their results to avoid redundant LLM calls, reduce latency, and save costs. It uses semantic similarity to match queries with cached results and manages cache invalidation and updates.

    Components: 1. Cache Storage: Stores prompts and results 2. Similarity Matcher: Finds semantically similar cached prompts 3. Cache Manager: Handles cache lifecycle 4. Invalidation Strategy: Manages cache freshness 5. Hit/Miss Tracker: Monitors cache performance 6. Compression: Optimizes cache storage

    Use Cases: – Repeated query handling – Cost optimization – Latency reduction – Common question answering – Template-based generation – Batch processing optimization

    Benefits: – Reduced API costs – Lower latency – Better resource utilization – Consistent responses – Offline capability

    Trade-offs: – Storage overhead – Cache invalidation complexity – Stale data risk – Memory usage – Similarity matching overhead

    LangChain Implementation: Combines semantic similarity search with LangChain caching mechanisms. Uses embeddings for matching similar queries efficiently.


    File Statistics

    • Total Lines: 527
    • Classes: 2
    • Functions: 1
    • Constants: 0

    Module Documentation

    """
    Pattern 166: Prompt Caching & Reuse
    
    Description:
        The Prompt Caching & Reuse pattern stores and retrieves prompts and their results
        to avoid redundant LLM calls, reduce latency, and save costs. It uses semantic
        similarity to match queries with cached results and manages cache invalidation
        and updates.
    
    Components:
        1. Cache Storage: Stores prompts and results
        2. Similarity Matcher: Finds semantically similar cached prompts
        3. Cache Manager: Handles cache lifecycle
        4. Invalidation Strategy: Manages cache freshness
        5. Hit/Miss Tracker: Monitors cache performance
        6. Compression: Optimizes cache storage
    
    Use Cases:
        - Repeated query handling
        - Cost optimization
        - Latency reduction
        - Common question answering
        - Template-based generation
        - Batch processing optimization
    
    Benefits:
        - Reduced API costs
        - Lower latency
        - Better resource utilization
        - Consistent responses
        - Offline capability
    
    Trade-offs:
        - Storage overhead
        - Cache invalidation complexity
        - Stale data risk
        - Memory usage
        - Similarity matching overhead
    
    LangChain Implementation:
        Combines semantic similarity search with LangChain caching mechanisms.
        Uses embeddings for matching similar queries efficiently.
    """
    

    Purpose: This module implements the Prompt Caching & Reuse pattern, providing a structured approach to agent design and implementation.


    Import Statements

    Standard Library Imports

    import collections.defaultdict
    import dataclasses.asdict
    import dataclasses.dataclass
    import dataclasses.field
    import datetime.datetime
    import datetime.timedelta
    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
    import typing.Tuple
    

    Purpose: Core Python functionality: – collections.defaultdict: Specialized container datatypes – dataclasses.asdict: Simplified class definitions with @dataclass – 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 – 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) – typing.Tuple: Type hints for better code documentation (List, Dict, Any, Optional)

    LangChain Imports

    from langchain_openai.ChatOpenAI
    from langchain_openai.OpenAIEmbeddings
    

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

    Other Imports

    from dotenv.load_dotenv
    from hashlib
    

    Purpose: Additional dependencies for specialized functionality.


    Classes

    Class: CacheEntry

    Lines 63-91

    Purpose:

    Represents a cached prompt and response
    

    Methods:

    is_expired(self)

    Lines 74-79

    Purpose:

    Check if cache entry has expired
    

    Parameters:

    Returns: bool

    to_dict(self)

    Lines 81-85

    Purpose:

    Convert to dictionary for serialization
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D6AA950>

    from_dict(data)

    Lines 88-91

    Decorators: staticmethod

    Purpose:

    Create from dictionary
    

    Parameters: – data: Data

    Returns: CacheEntry


    Class: PromptCacheManager

    Lines 94-337

    Purpose:

    Manages prompt caching and reuse with semantic similarity matching
    

    Methods:

    __init__(self, similarity_threshold, max_cache_size, default_ttl)

    Lines 97-124

    Purpose:

    Initialize prompt cache manager
    
    Args:
        similarity_threshold: Minimum similarity score for cache hit (0-1)
        max_cache_size: Maximum number of entries to cache
        default_ttl: Default time-to-live in seconds
    

    Parameters: – similarity_threshold: Similarity threshold – max_cache_size: Max cache size – default_ttl: Default ttl

    get_or_generate(self, prompt, force_refresh, ttl)

    Lines 126-158

    Purpose:

    Get cached response or generate new one
    
    Args:
        prompt: The prompt to process
        force_refresh: Skip cache and generate new response
        ttl: Time-to-live for this entry
    
    Returns:
        Tuple of (response, was_cached)
    

    Parameters: – prompt: Prompt template or text – force_refresh: Force refresh – ttl: Ttl

    Returns: <ast.Subscript object at 0x000001A06D824D50>

    batch_get_or_generate(self, prompts)

    Lines 160-166

    Purpose:

    Process multiple prompts efficiently
    

    Parameters: – prompts: Prompt template or text

    Returns: <ast.Subscript object at 0x000001A06D827310>

    invalidate(self, pattern)

    Lines 168-180

    Purpose:

    Invalidate cache entries matching pattern
    

    Parameters: – pattern: Pattern

    _find_similar_cached(self, prompt)

    Lines 182-212

    Purpose:

    Find semantically similar cached prompt
    

    Parameters: – prompt: Prompt template or text

    Returns: <ast.Subscript object at 0x000001A06D6147D0>

    _generate_response(self, prompt)

    Lines 214-221

    Purpose:

    Generate new response using LLM
    

    Parameters: – prompt: Prompt template or text

    Returns: str

    _add_to_cache(self, prompt, response, ttl)

    Lines 223-246

    Purpose:

    Add entry to cache
    

    Parameters: – prompt: Prompt template or text – response: Response – ttl: Ttl

    _generate_key(self, prompt)

    Lines 248-250

    Purpose:

    Generate cache key from prompt
    

    Parameters: – prompt: Prompt template or text

    Returns: str

    _cosine_similarity(self, vec1, vec2)

    Lines 252-261

    Purpose:

    Calculate cosine similarity between two vectors
    

    Parameters: – vec1: Vec1 – vec2: Vec2

    Returns: float

    _clean_expired(self)

    Lines 263-270

    Purpose:

    Remove expired entries
    

    Parameters:

    _evict_entry(self)

    Lines 272-282

    Purpose:

    Evict least recently used entry
    

    Parameters:

    get_stats(self)

    Lines 284-295

    Purpose:

    Get cache statistics
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D8656D0>

    get_popular_queries(self, top_n)

    Lines 297-312

    Purpose:

    Get most frequently accessed cached queries
    

    Parameters: – top_n: Top n

    Returns: <ast.Subscript object at 0x000001A06D7B8510>

    save_cache(self, filepath)

    Lines 314-322

    Purpose:

    Save cache to file
    

    Parameters: – filepath: Filepath

    load_cache(self, filepath)

    Lines 324-337

    Purpose:

    Load cache from file
    

    Parameters: – filepath: Filepath


    Functions

    demonstrate_prompt_caching()

    Lines 340-522

    Purpose:

    Demonstrate prompt caching and reuse
    

    Key Concepts Summary

    Pattern Overview

    The Prompt Caching & Reuse 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 165
    • Pattern 167

    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 Prompt Caching & Reuse 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: 166_prompt_caching_&_reuse.py
    Total Lines: 527
    Pattern Category: AI Agent Design Pattern #166


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Foundation Model Orchestration

    
        """
    Pattern 165: Foundation Model Orchestration
    
    Description:
        The Foundation Model Orchestration pattern coordinates multiple foundation models
        (LLMs with different capabilities, sizes, and specializations) to solve complex tasks
        optimally. The orchestrator routes tasks to the most appropriate model based on
        requirements, capabilities, cost, and performance trade-offs.
    
    Components:
        1. Model Registry: Catalog of available models and their capabilities
        2. Task Analyzer: Analyzes task requirements and complexity
        3. Model Selector: Chooses optimal model for each task
        4. Result Aggregator: Combines results from multiple models
        5. Performance Monitor: Tracks model performance and costs
        6. Fallback Manager: Handles model failures
    
    Use Cases:
        - Cost-optimized AI pipelines
        - Quality-performance trade-offs
        - Specialized task routing
        - Multi-model ensemble systems
        - Adaptive model selection
        - Resource optimization
    
    Benefits:
        - Optimal cost-performance balance
        - Better quality through specialization
        - Fault tolerance with fallbacks
        - Flexible model upgrades
        - Resource efficiency
    
    Trade-offs:
        - Orchestration complexity
        - Latency overhead
        - Model compatibility issues
        - Cost tracking complexity
        - Requires multiple API keys
    
    LangChain Implementation:
        Uses LangChain's model abstraction to coordinate multiple models. Implements
        intelligent routing logic based on task characteristics and model capabilities.
    """
    
    import os
    import time
    from typing import List, Dict, Any, Optional, Callable
    from dataclasses import dataclass, field
    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 ModelCapability(Enum):
        """Model capability types"""
        REASONING = "reasoning"
        CODING = "coding"
        CREATIVE = "creative"
        ANALYSIS = "analysis"
        SUMMARIZATION = "summarization"
        TRANSLATION = "translation"
        FAST_RESPONSE = "fast_response"
        LONG_CONTEXT = "long_context"
    
    
    class ModelTier(Enum):
        """Model tier for cost/performance trade-off"""
        NANO = "nano"  # Fastest, cheapest, basic
        SMALL = "small"  # Fast, cheap, good quality
        MEDIUM = "medium"  # Balanced
        LARGE = "large"  # High quality, slower, expensive
        PREMIUM = "premium"  # Best quality, slowest, most expensive
    
    
    @dataclass
    class ModelProfile:
        """Profile of a foundation model"""
        name: str
        model_id: str
        tier: ModelTier
        capabilities: List[ModelCapability]
        max_tokens: int
        cost_per_1k_tokens: float
        avg_latency_ms: float
        quality_score: float = 0.8  # 0-1 scale
        available: bool = True
        metadata: Dict[str, Any] = field(default_factory=dict)
    
    
    @dataclass
    class Task:
        """A task to be processed"""
        id: str
        content: str
        required_capabilities: List[ModelCapability]
        priority: str = "medium"  # low, medium, high
        max_cost: Optional[float] = None
        max_latency: Optional[float] = None
        min_quality: float = 0.7
    
    
    @dataclass
    class TaskResult:
        """Result of task execution"""
        task_id: str
        model_used: str
        result: str
        execution_time: float
        estimated_cost: float
        quality_estimate: float
        metadata: Dict[str, Any] = field(default_factory=dict)
    
    
    class FoundationModelOrchestrator:
        """Orchestrates multiple foundation models for optimal task execution"""
        
        def __init__(self):
            """Initialize the orchestrator with model registry"""
            self.models: Dict[str, ModelProfile] = {}
            self.llm_instances: Dict[str, ChatOpenAI] = {}
            self.performance_stats: Dict[str, Dict[str, Any]] = defaultdict(lambda: {
                "tasks_completed": 0,
                "total_cost": 0.0,
                "total_latency": 0.0,
                "failures": 0
            })
            
            # Register models
            self._register_default_models()
            
            # Task routing prompt
            self.routing_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a task analysis expert. Analyze the task and determine
                which capabilities are most important for successful completion."""),
                ("user", """Task: {task}
    
    Analyze this task and identify the key capabilities needed (choose from: reasoning,
    coding, creative, analysis, summarization, translation, fast_response, long_context).
    
    Respond with just the capability names, comma-separated.""")
            ])
        
        def _register_default_models(self):
            """Register default model profiles"""
            # Nano tier - fast and cheap
            self.register_model(ModelProfile(
                name="GPT-3.5 Turbo",
                model_id="gpt-3.5-turbo",
                tier=ModelTier.SMALL,
                capabilities=[
                    ModelCapability.FAST_RESPONSE,
                    ModelCapability.SUMMARIZATION,
                    ModelCapability.ANALYSIS
                ],
                max_tokens=4096,
                cost_per_1k_tokens=0.002,
                avg_latency_ms=800,
                quality_score=0.75
            ))
            
            # Medium tier - balanced
            self.register_model(ModelProfile(
                name="GPT-4 Turbo",
                model_id="gpt-4-turbo-preview",
                tier=ModelTier.LARGE,
                capabilities=[
                    ModelCapability.REASONING,
                    ModelCapability.CODING,
                    ModelCapability.CREATIVE,
                    ModelCapability.ANALYSIS,
                    ModelCapability.LONG_CONTEXT
                ],
                max_tokens=128000,
                cost_per_1k_tokens=0.01,
                avg_latency_ms=2000,
                quality_score=0.95
            ))
            
            # Specialized models (simulated with different configs)
            self.register_model(ModelProfile(
                name="Code Specialist",
                model_id="gpt-3.5-turbo",  # Using same model but different profile
                tier=ModelTier.MEDIUM,
                capabilities=[
                    ModelCapability.CODING,
                    ModelCapability.REASONING
                ],
                max_tokens=4096,
                cost_per_1k_tokens=0.003,
                avg_latency_ms=1000,
                quality_score=0.85
            ))
        
        def register_model(self, profile: ModelProfile):
            """Register a model in the orchestrator"""
            self.models[profile.name] = profile
            # Create LLM instance
            try:
                self.llm_instances[profile.name] = ChatOpenAI(
                    model=profile.model_id,
                    temperature=0.7,
                    max_tokens=1000
                )
            except Exception as e:
                print(f"Warning: Could not create LLM instance for {profile.name}: {e}")
                profile.available = False
        
        def execute_task(self, task: Task, use_fallback: bool = True) -> TaskResult:
            """Execute a task using the most appropriate model"""
            # Select best model for task
            selected_model = self._select_model(task)
            
            if not selected_model:
                raise ValueError("No suitable model found for task")
            
            # Execute task
            start_time = time.time()
            
            try:
                result = self._execute_on_model(task, selected_model)
                execution_time = time.time() - start_time
                
                # Calculate cost
                estimated_tokens = len(task.content.split()) * 1.3  # Rough estimate
                estimated_cost = (estimated_tokens / 1000) * selected_model.cost_per_1k_tokens
                
                # Update stats
                self._update_stats(selected_model.name, execution_time, estimated_cost, success=True)
                
                return TaskResult(
                    task_id=task.id,
                    model_used=selected_model.name,
                    result=result,
                    execution_time=execution_time,
                    estimated_cost=estimated_cost,
                    quality_estimate=selected_model.quality_score
                )
                
            except Exception as e:
                self._update_stats(selected_model.name, 0, 0, success=False)
                
                if use_fallback:
                    # Try fallback model
                    fallback_model = self._select_fallback_model(task, selected_model)
                    if fallback_model:
                        print(f"Primary model failed, trying fallback: {fallback_model.name}")
                        task_copy = Task(
                            id=task.id,
                            content=task.content,
                            required_capabilities=task.required_capabilities,
                            priority=task.priority
                        )
                        return self.execute_task(task_copy, use_fallback=False)
                
                raise RuntimeError(f"Task execution failed: {e}")
        
        def execute_batch(self, tasks: List[Task], parallel: bool = False) -> List[TaskResult]:
            """Execute multiple tasks"""
            results = []
            
            for task in tasks:
                try:
                    result = self.execute_task(task)
                    results.append(result)
                except Exception as e:
                    print(f"Task {task.id} failed: {e}")
            
            return results
        
        def execute_with_ensemble(self, task: Task, num_models: int = 3) -> Dict[str, Any]:
            """Execute task on multiple models and aggregate results"""
            # Select top N models
            candidates = self._rank_models_for_task(task)[:num_models]
            
            results = []
            for model_profile in candidates:
                try:
                    result = self._execute_on_model(task, model_profile)
                    results.append({
                        "model": model_profile.name,
                        "result": result,
                        "quality": model_profile.quality_score
                    })
                except Exception as e:
                    print(f"Model {model_profile.name} failed: {e}")
            
            if not results:
                raise RuntimeError("All models failed")
            
            # Aggregate results (simple voting or quality-weighted selection)
            best_result = max(results, key=lambda x: x["quality"])
            
            return {
                "aggregated_result": best_result["result"],
                "all_results": results,
                "models_used": [r["model"] for r in results]
            }
        
        def _select_model(self, task: Task) -> Optional[ModelProfile]:
            """Select the best model for a task"""
            candidates = self._rank_models_for_task(task)
            
            if not candidates:
                return None
            
            # Filter by constraints
            for model in candidates:
                # Check cost constraint
                if task.max_cost:
                    estimated_tokens = len(task.content.split()) * 1.3
                    estimated_cost = (estimated_tokens / 1000) * model.cost_per_1k_tokens
                    if estimated_cost > task.max_cost:
                        continue
                
                # Check latency constraint
                if task.max_latency and model.avg_latency_ms > task.max_latency:
                    continue
                
                # Check quality constraint
                if model.quality_score  List[ModelProfile]:
            """Rank models by suitability for task"""
            scores = []
            
            for model in self.models.values():
                if not model.available:
                    continue
                
                score = 0.0
                
                # Capability match
                matching_caps = sum(
                    1 for cap in task.required_capabilities 
                    if cap in model.capabilities
                )
                if task.required_capabilities:
                    capability_score = matching_caps / len(task.required_capabilities)
                else:
                    capability_score = 0.5
                
                score += capability_score * 40  # 40% weight
                
                # Quality score
                score += model.quality_score * 30  # 30% weight
                
                # Cost efficiency (lower is better)
                cost_score = 1.0 - min(model.cost_per_1k_tokens / 0.05, 1.0)
                score += cost_score * 15  # 15% weight
                
                # Speed (lower latency is better)
                speed_score = 1.0 - min(model.avg_latency_ms / 5000, 1.0)
                score += speed_score * 15  # 15% weight
                
                # Priority adjustment
                if task.priority == "high":
                    score += model.quality_score * 10
                elif task.priority == "low":
                    score += cost_score * 10
                
                scores.append((score, model))
            
            # Sort by score descending
            scores.sort(key=lambda x: x[0], reverse=True)
            
            return [model for _, model in scores]
        
        def _select_fallback_model(self, task: Task, failed_model: ModelProfile) -> Optional[ModelProfile]:
            """Select a fallback model after primary fails"""
            candidates = self._rank_models_for_task(task)
            
            # Return first available model that's not the failed one
            for model in candidates:
                if model.name != failed_model.name and model.available:
                    return model
            
            return None
        
        def _execute_on_model(self, task: Task, model: ModelProfile) -> str:
            """Execute task on specific model"""
            llm = self.llm_instances.get(model.name)
            
            if not llm:
                raise RuntimeError(f"LLM instance not available for {model.name}")
            
            # Create execution prompt
            prompt = ChatPromptTemplate.from_messages([
                ("system", f"You are processing this task using {model.name}."),
                ("user", "{task_content}")
            ])
            
            chain = prompt | llm | StrOutputParser()
            result = chain.invoke({"task_content": task.content})
            
            return result
        
        def _update_stats(self, model_name: str, execution_time: float, 
                         cost: float, success: bool):
            """Update performance statistics"""
            stats = self.performance_stats[model_name]
            
            if success:
                stats["tasks_completed"] += 1
                stats["total_cost"] += cost
                stats["total_latency"] += execution_time
            else:
                stats["failures"] += 1
        
        def get_performance_report(self) -> Dict[str, Any]:
            """Get performance report for all models"""
            report = {}
            
            for model_name, stats in self.performance_stats.items():
                total_tasks = stats["tasks_completed"] + stats["failures"]
                
                if total_tasks > 0:
                    report[model_name] = {
                        "tasks_completed": stats["tasks_completed"],
                        "failures": stats["failures"],
                        "success_rate": stats["tasks_completed"] / total_tasks,
                        "total_cost": stats["total_cost"],
                        "avg_latency": stats["total_latency"] / stats["tasks_completed"] 
                                      if stats["tasks_completed"] > 0 else 0,
                        "cost_per_task": stats["total_cost"] / stats["tasks_completed"]
                                        if stats["tasks_completed"] > 0 else 0
                    }
            
            return report
    
    
    def demonstrate_foundation_model_orchestration():
        """Demonstrate foundation model orchestration"""
        print("=" * 80)
        print("FOUNDATION MODEL ORCHESTRATION PATTERN DEMONSTRATION")
        print("=" * 80)
        
        orchestrator = FoundationModelOrchestrator()
        
        # Example 1: Simple task routing
        print("\n" + "=" * 80)
        print("Example 1: Simple Task Routing")
        print("=" * 80)
        
        tasks = [
            Task(
                id="task1",
                content="Summarize this article in 2 sentences: AI has transformed modern technology...",
                required_capabilities=[ModelCapability.SUMMARIZATION, ModelCapability.FAST_RESPONSE],
                priority="medium"
            ),
            Task(
                id="task2",
                content="Write a Python function to calculate fibonacci numbers recursively",
                required_capabilities=[ModelCapability.CODING],
                priority="high"
            ),
            Task(
                id="task3",
                content="Write a creative short story about a robot learning to paint",
                required_capabilities=[ModelCapability.CREATIVE],
                priority="low"
            )
        ]
        
        print("\nExecuting tasks with automatic model selection:")
        for task in tasks:
            try:
                result = orchestrator.execute_task(task)
                print(f"\nTask: {task.id}")
                print(f"Required capabilities: {[c.value for c in task.required_capabilities]}")
                print(f"Selected model: {result.model_used}")
                print(f"Execution time: {result.execution_time:.3f}s")
                print(f"Estimated cost: ${result.estimated_cost:.6f}")
                print(f"Quality estimate: {result.quality_estimate:.2f}")
                print(f"Result preview: {result.result[:100]}...")
            except Exception as e:
                print(f"\nTask {task.id} failed: {e}")
        
        # Example 2: Cost-constrained execution
        print("\n" + "=" * 80)
        print("Example 2: Cost-Constrained Execution")
        print("=" * 80)
        
        budget_task = Task(
            id="budget_task",
            content="Analyze the sentiment of this review: The product was okay but overpriced.",
            required_capabilities=[ModelCapability.ANALYSIS],
            priority="medium",
            max_cost=0.001  # Very tight budget
        )
        
        try:
            result = orchestrator.execute_task(budget_task)
            print(f"\nTask with max cost ${budget_task.max_cost:.6f}:")
            print(f"Selected model: {result.model_used}")
            print(f"Actual cost: ${result.estimated_cost:.6f}")
            print(f"Result: {result.result[:150]}...")
        except Exception as e:
            print(f"Task failed: {e}")
        
        # Example 3: Quality-focused execution
        print("\n" + "=" * 80)
        print("Example 3: Quality-Focused Execution")
        print("=" * 80)
        
        quality_task = Task(
            id="quality_task",
            content="Solve this logic puzzle: If all A are B, and some B are C, what can we conclude?",
            required_capabilities=[ModelCapability.REASONING],
            priority="high",
            min_quality=0.9  # High quality requirement
        )
        
        try:
            result = orchestrator.execute_task(quality_task)
            print(f"\nTask with min quality {quality_task.min_quality}:")
            print(f"Selected model: {result.model_used}")
            print(f"Quality estimate: {result.quality_estimate}")
            print(f"Result: {result.result[:200]}...")
        except Exception as e:
            print(f"Task failed: {e}")
        
        # Example 4: Ensemble execution
        print("\n" + "=" * 80)
        print("Example 4: Ensemble Execution")
        print("=" * 80)
        
        ensemble_task = Task(
            id="ensemble_task",
            content="What are the three most important factors in climate change?",
            required_capabilities=[ModelCapability.ANALYSIS, ModelCapability.REASONING],
            priority="high"
        )
        
        try:
            ensemble_result = orchestrator.execute_with_ensemble(ensemble_task, num_models=2)
            print(f"\nTask executed with ensemble approach:")
            print(f"Models used: {', '.join(ensemble_result['models_used'])}")
            print(f"\nAggregated result:")
            print(ensemble_result['aggregated_result'][:200] + "...")
            
            print(f"\nAll model results:")
            for i, r in enumerate(ensemble_result['all_results'], 1):
                print(f"\n  Model {i}: {r['model']} (quality: {r['quality']})")
                print(f"  Result: {r['result'][:100]}...")
        except Exception as e:
            print(f"Ensemble execution failed: {e}")
        
        # Example 5: Batch processing
        print("\n" + "=" * 80)
        print("Example 5: Batch Processing")
        print("=" * 80)
        
        batch_tasks = [
            Task(f"batch_{i}", f"Process item {i}: analyze and summarize", 
                 [ModelCapability.ANALYSIS], "medium")
            for i in range(3)
        ]
        
        print(f"\nProcessing batch of {len(batch_tasks)} tasks:")
        batch_results = orchestrator.execute_batch(batch_tasks)
        
        print(f"Completed: {len(batch_results)}/{len(batch_tasks)} tasks")
        for result in batch_results:
            print(f"  {result.task_id}: {result.model_used} in {result.execution_time:.3f}s")
        
        # Example 6: Performance reporting
        print("\n" + "=" * 80)
        print("Example 6: Performance Reporting")
        print("=" * 80)
        
        report = orchestrator.get_performance_report()
        
        print("\nPerformance Report:")
        print("-" * 60)
        for model_name, stats in report.items():
            print(f"\n{model_name}:")
            print(f"  Tasks completed: {stats['tasks_completed']}")
            print(f"  Failures: {stats['failures']}")
            print(f"  Success rate: {stats['success_rate']:.1%}")
            print(f"  Total cost: ${stats['total_cost']:.6f}")
            print(f"  Avg latency: {stats['avg_latency']:.3f}s")
            print(f"  Cost per task: ${stats['cost_per_task']:.6f}")
        
        # Summary
        print("\n" + "=" * 80)
        print("SUMMARY")
        print("=" * 80)
        print("""
    The Foundation Model Orchestration pattern enables:
    ✓ Intelligent routing of tasks to optimal models
    ✓ Cost-performance trade-off optimization
    ✓ Quality-focused model selection
    ✓ Ensemble execution for critical tasks
    ✓ Fallback handling for reliability
    ✓ Performance monitoring and optimization
    ✓ Batch processing with automatic routing
    
    This pattern is valuable for:
    - Production AI systems with cost constraints
    - Multi-model architectures
    - Quality-performance optimization
    - Resource-efficient AI pipelines
    - Fault-tolerant AI systems
    - Adaptive model selection
        """)
    
    
    if __name__ == "__main__":
        demonstrate_foundation_model_orchestration()
    
        

    Agentic Design Pattern: Foundation Model Orchestration

    Pattern 165: Foundation Model Orchestration – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Foundation Model Orchestration pattern implementation.


    Pattern Description

    Pattern 165: Foundation Model Orchestration

    Description: The Foundation Model Orchestration pattern coordinates multiple foundation models (LLMs with different capabilities, sizes, and specializations) to solve complex tasks optimally. The orchestrator routes tasks to the most appropriate model based on requirements, capabilities, cost, and performance trade-offs.

    Components: 1. Model Registry: Catalog of available models and their capabilities 2. Task Analyzer: Analyzes task requirements and complexity 3. Model Selector: Chooses optimal model for each task 4. Result Aggregator: Combines results from multiple models 5. Performance Monitor: Tracks model performance and costs 6. Fallback Manager: Handles model failures

    Use Cases: – Cost-optimized AI pipelines – Quality-performance trade-offs – Specialized task routing – Multi-model ensemble systems – Adaptive model selection – Resource optimization

    Benefits: – Optimal cost-performance balance – Better quality through specialization – Fault tolerance with fallbacks – Flexible model upgrades – Resource efficiency

    Trade-offs: – Orchestration complexity – Latency overhead – Model compatibility issues – Cost tracking complexity – Requires multiple API keys

    LangChain Implementation: Uses LangChain’s model abstraction to coordinate multiple models. Implements intelligent routing logic based on task characteristics and model capabilities.


    File Statistics

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

    Module Documentation

    """
    Pattern 165: Foundation Model Orchestration
    
    Description:
        The Foundation Model Orchestration pattern coordinates multiple foundation models
        (LLMs with different capabilities, sizes, and specializations) to solve complex tasks
        optimally. The orchestrator routes tasks to the most appropriate model based on
        requirements, capabilities, cost, and performance trade-offs.
    
    Components:
        1. Model Registry: Catalog of available models and their capabilities
        2. Task Analyzer: Analyzes task requirements and complexity
        3. Model Selector: Chooses optimal model for each task
        4. Result Aggregator: Combines results from multiple models
        5. Performance Monitor: Tracks model performance and costs
        6. Fallback Manager: Handles model failures
    
    Use Cases:
        - Cost-optimized AI pipelines
        - Quality-performance trade-offs
        - Specialized task routing
        - Multi-model ensemble systems
        - Adaptive model selection
        - Resource optimization
    
    Benefits:
        - Optimal cost-performance balance
        - Better quality through specialization
        - Fault tolerance with fallbacks
        - Flexible model upgrades
        - Resource efficiency
    
    Trade-offs:
        - Orchestration complexity
        - Latency overhead
        - Model compatibility issues
        - Cost tracking complexity
        - Requires multiple API keys
    
    LangChain Implementation:
        Uses LangChain's model abstraction to coordinate multiple models. Implements
        intelligent routing logic based on task characteristics and model capabilities.
    """
    

    Purpose: This module implements the Foundation Model Orchestration 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 enum.Enum
    import langchain_core.output_parsers.StrOutputParser
    import langchain_core.prompts.ChatPromptTemplate
    import os
    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 – 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.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: ModelCapability

    Lines 60-69

    Inherits from: Enum

    Purpose:

    Model capability types
    

    Class: ModelTier

    Lines 72-78

    Inherits from: Enum

    Purpose:

    Model tier for cost/performance trade-off
    

    Class: ModelProfile

    Lines 82-93

    Purpose:

    Profile of a foundation model
    

    Class: Task

    Lines 97-105

    Purpose:

    A task to be processed
    

    Class: TaskResult

    Lines 109-117

    Purpose:

    Result of task execution
    

    Class: FoundationModelOrchestrator

    Lines 120-444

    Purpose:

    Orchestrates multiple foundation models for optimal task execution
    

    Methods:

    __init__(self)

    Lines 123-147

    Purpose:

    Initialize the orchestrator with model registry
    

    Parameters:

    _register_default_models(self)

    Lines 149-198

    Purpose:

    Register default model profiles
    

    Parameters:

    register_model(self, profile)

    Lines 200-212

    Purpose:

    Register a model in the orchestrator
    

    Parameters: – profile: Profile

    execute_task(self, task, use_fallback)

    Lines 214-261

    Purpose:

    Execute a task using the most appropriate model
    

    Parameters: – task: Task – use_fallback: Use fallback

    Returns: TaskResult

    execute_batch(self, tasks, parallel)

    Lines 263-274

    Purpose:

    Execute multiple tasks
    

    Parameters: – tasks: Tasks – parallel: Parallel

    Returns: <ast.Subscript object at 0x000001A06D6210D0>

    execute_with_ensemble(self, task, num_models)

    Lines 276-303

    Purpose:

    Execute task on multiple models and aggregate results
    

    Parameters: – task: Task – num_models: Language model instance

    Returns: <ast.Subscript object at 0x000001A06E0ABE50>

    _select_model(self, task)

    Lines 305-336

    Purpose:

    Select the best model for a task
    

    Parameters: – task: Task

    Returns: <ast.Subscript object at 0x000001A06D6744D0>

    _rank_models_for_task(self, task)

    Lines 338-382

    Purpose:

    Rank models by suitability for task
    

    Parameters: – task: Task

    Returns: <ast.Subscript object at 0x000001A06D7C90D0>

    _select_fallback_model(self, task, failed_model)

    Lines 384-393

    Purpose:

    Select a fallback model after primary fails
    

    Parameters: – task: Task – failed_model: Language model instance

    Returns: <ast.Subscript object at 0x000001A06D7C9090>

    _execute_on_model(self, task, model)

    Lines 395-411

    Purpose:

    Execute task on specific model
    

    Parameters: – task: Task – model: Language model instance

    Returns: str

    _update_stats(self, model_name, execution_time, cost, success)

    Lines 413-423

    Purpose:

    Update performance statistics
    

    Parameters: – model_name: Language model instance – execution_time: Execution time – cost: Cost – success: Success

    get_performance_report(self)

    Lines 425-444

    Purpose:

    Get performance report for all models
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D889090>


    Functions

    demonstrate_foundation_model_orchestration()

    Lines 447-622

    Purpose:

    Demonstrate foundation model orchestration
    

    Key Concepts Summary

    Pattern Overview

    The Foundation Model Orchestration 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 164
    • Pattern 166

    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 Foundation Model Orchestration 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: 165_foundation_model_orchestration.py
    Total Lines: 627
    Pattern Category: AI Agent Design Pattern #165


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Temporal Reasoning

    
        """
    Pattern 164: Temporal Reasoning
    
    Description:
        The Temporal Reasoning pattern enables agents to understand and reason about time-based
        relationships, sequences, durations, and temporal logic. This pattern helps agents solve
        problems involving timelines, schedules, causality, and temporal constraints.
    
    Components:
        1. Timeline Manager: Maintains ordered sequences of events
        2. Temporal Relation Analyzer: Identifies before/after/during/overlapping relationships
        3. Duration Calculator: Computes time intervals and durations
        4. Temporal Constraint Checker: Validates temporal constraints
        5. Sequence Analyzer: Analyzes event sequences and patterns
        6. Temporal Query Engine: Answers time-based questions
    
    Use Cases:
        - Schedule planning and optimization
        - Event sequence analysis
        - Causal reasoning
        - Historical timeline construction
        - Project timeline management
        - Deadline tracking
        - Temporal pattern detection
    
    Benefits:
        - Enables time-aware reasoning
        - Supports scheduling and planning
        - Detects temporal patterns
        - Validates temporal constraints
        - Better understanding of causality
    
    Trade-offs:
        - Complex time representation
        - Timezone handling challenges
        - Requires precise temporal data
        - May need calendar logic
        - Limited by data accuracy
    
    LangChain Implementation:
        Uses LLM for temporal relationship interpretation with structured datetime
        management and interval calculations. Combines symbolic temporal logic with
        natural language understanding for temporal queries.
    """
    
    import os
    from typing import List, Dict, Any, Optional, Tuple
    from dataclasses import dataclass, field
    from datetime import datetime, timedelta
    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
    
    load_dotenv()
    
    
    class TemporalRelation(Enum):
        """Types of temporal relationships (Allen's interval algebra)"""
        BEFORE = "before"
        AFTER = "after"
        DURING = "during"
        CONTAINS = "contains"
        OVERLAPS = "overlaps"
        OVERLAPPED_BY = "overlapped_by"
        MEETS = "meets"
        MET_BY = "met_by"
        STARTS = "starts"
        STARTED_BY = "started_by"
        FINISHES = "finishes"
        FINISHED_BY = "finished_by"
        EQUALS = "equals"
    
    
    class TimeUnit(Enum):
        """Units of time measurement"""
        SECOND = "second"
        MINUTE = "minute"
        HOUR = "hour"
        DAY = "day"
        WEEK = "week"
        MONTH = "month"
        YEAR = "year"
    
    
    @dataclass
    class TemporalEvent:
        """Represents an event with temporal properties"""
        name: str
        start_time: datetime
        end_time: Optional[datetime] = None
        description: str = ""
        metadata: Dict[str, Any] = field(default_factory=dict)
        
        @property
        def duration(self) -> Optional[timedelta]:
            """Calculate duration of the event"""
            if self.end_time:
                return self.end_time - self.start_time
            return None
        
        @property
        def is_point_event(self) -> bool:
            """Check if this is a point event (no duration)"""
            return self.end_time is None or self.start_time == self.end_time
    
    
    @dataclass
    class Timeline:
        """Represents a timeline with multiple events"""
        name: str
        events: List[TemporalEvent] = field(default_factory=list)
        
        def add_event(self, event: TemporalEvent):
            """Add an event to the timeline"""
            self.events.append(event)
            self.events.sort(key=lambda e: e.start_time)
        
        def get_events_in_range(self, start: datetime, end: datetime) -> List[TemporalEvent]:
            """Get all events within a time range"""
            return [e for e in self.events 
                    if e.start_time = start)]
    
    
    @dataclass
    class TemporalQuery:
        """A temporal reasoning query"""
        question: str
        events: List[TemporalEvent]
        context: str = ""
    
    
    @dataclass
    class TemporalAnalysis:
        """Result of temporal reasoning"""
        answer: str
        relationships: List[str]
        timeline: str
        sequences: List[List[str]]
        reasoning: str
    
    
    class TemporalReasoningAgent:
        """Agent that performs temporal reasoning tasks"""
        
        def __init__(self, model_name: str = "gpt-3.5-turbo"):
            """Initialize the temporal reasoning agent"""
            self.llm = ChatOpenAI(model=model_name, temperature=0)
            
            # Prompts for temporal reasoning
            self.temporal_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a temporal reasoning expert. Analyze events, timelines,
                and temporal relationships. Consider durations, sequences, and causal relationships.
                Be precise about time-based logic."""),
                ("user", """Events and their times:
    {event_descriptions}
    
    Question: {question}
    
    Provide a clear answer based on temporal analysis.""")
            ])
            
            self.sequence_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are an expert at analyzing event sequences and patterns.
                Identify patterns, detect anomalies, and reason about causality."""),
                ("user", """Event sequence:
    {sequence}
    
    Analysis task: {task}
    
    Describe patterns, relationships, and insights.""")
            ])
            
            self.temporal_chain = self.temporal_prompt | self.llm | StrOutputParser()
            self.sequence_chain = self.sequence_prompt | self.llm | StrOutputParser()
        
        def analyze_temporal_query(self, query: TemporalQuery) -> TemporalAnalysis:
            """Analyze a temporal query and return comprehensive analysis"""
            # Identify temporal relationships
            relationships = self._identify_temporal_relationships(query.events)
            
            # Create timeline visualization
            timeline_viz = self._create_timeline_visualization(query.events)
            
            # Identify sequences
            sequences = self._identify_sequences(query.events)
            
            # Create event descriptions
            event_descriptions = self._create_event_descriptions(query.events)
            
            # Get LLM reasoning
            reasoning = self.temporal_chain.invoke({
                "event_descriptions": event_descriptions,
                "question": query.question
            })
            
            return TemporalAnalysis(
                answer=reasoning,
                relationships=relationships,
                timeline=timeline_viz,
                sequences=sequences,
                reasoning=reasoning
            )
        
        def check_temporal_relation(self, event1: TemporalEvent, 
                                   event2: TemporalEvent,
                                   relation: TemporalRelation) -> bool:
            """Check if a specific temporal relationship holds between two events"""
            e1_start = event1.start_time
            e1_end = event1.end_time or event1.start_time
            e2_start = event2.start_time
            e2_end = event2.end_time or event2.start_time
            
            if relation == TemporalRelation.BEFORE:
                return e1_end  e2_end
            elif relation == TemporalRelation.DURING:
                return e1_start > e2_start and e1_end < e2_end
            elif relation == TemporalRelation.CONTAINS:
                return e1_start  e2_end
            elif relation == TemporalRelation.OVERLAPS:
                return e1_start  e2_start and e1_end < e2_end
            elif relation == TemporalRelation.OVERLAPPED_BY:
                return e2_start  e1_start and e2_end < e1_end
            elif relation == TemporalRelation.MEETS:
                return e1_end == e2_start
            elif relation == TemporalRelation.MET_BY:
                return e2_end == e1_start
            elif relation == TemporalRelation.STARTS:
                return e1_start == e2_start and e1_end  e2_end
            elif relation == TemporalRelation.FINISHES:
                return e1_start > e2_start and e1_end == e2_end
            elif relation == TemporalRelation.FINISHED_BY:
                return e1_start  List[Tuple[str, str]]:
            """Find all pairs of overlapping events"""
            overlaps = []
            
            for i, event1 in enumerate(events):
                for event2 in events[i+1:]:
                    if self._events_overlap(event1, event2):
                        overlaps.append((event1.name, event2.name))
            
            return overlaps
        
        def find_gaps(self, events: List[TemporalEvent]) -> List[Tuple[datetime, datetime]]:
            """Find time gaps between consecutive events"""
            if not events:
                return []
            
            sorted_events = sorted(events, key=lambda e: e.start_time)
            gaps = []
            
            for i in range(len(sorted_events) - 1):
                current_end = sorted_events[i].end_time or sorted_events[i].start_time
                next_start = sorted_events[i + 1].start_time
                
                if current_end  timedelta:
            """Calculate total duration of events"""
            if not events:
                return timedelta(0)
            
            if not merge_overlaps:
                # Simple sum of all durations
                total = timedelta(0)
                for event in events:
                    if event.duration:
                        total += event.duration
                return total
            else:
                # Merge overlapping intervals first
                intervals = []
                for event in events:
                    start = event.start_time
                    end = event.end_time or event.start_time
                    intervals.append((start, end))
                
                # Sort and merge
                intervals.sort()
                merged = []
                current_start, current_end = intervals[0]
                
                for start, end in intervals[1:]:
                    if start  Dict[str, Any]:
            """Detect patterns in temporal data"""
            if not events:
                return {"patterns": []}
            
            sorted_events = sorted(events, key=lambda e: e.start_time)
            
            patterns = {
                "total_events": len(events),
                "time_span": {
                    "start": sorted_events[0].start_time,
                    "end": sorted_events[-1].end_time or sorted_events[-1].start_time
                },
                "regular_intervals": self._detect_regular_intervals(sorted_events),
                "clusters": self._detect_time_clusters(sorted_events),
                "gaps": len(self.find_gaps(events)),
                "overlaps": len(self.find_overlapping_events(events))
            }
            
            return patterns
        
        def _events_overlap(self, event1: TemporalEvent, event2: TemporalEvent) -> bool:
            """Check if two events overlap in time"""
            e1_start = event1.start_time
            e1_end = event1.end_time or event1.start_time
            e2_start = event2.start_time
            e2_end = event2.end_time or event2.start_time
            
            return not (e1_end < e2_start or e2_end  List[str]:
            """Identify temporal relationships between events"""
            relationships = []
            
            for i, event1 in enumerate(events):
                for event2 in events[i+1:]:
                    # Check key relationships
                    if self.check_temporal_relation(event1, event2, TemporalRelation.BEFORE):
                        relationships.append(f"{event1.name} before {event2.name}")
                    elif self.check_temporal_relation(event1, event2, TemporalRelation.DURING):
                        relationships.append(f"{event1.name} during {event2.name}")
                    elif self.check_temporal_relation(event1, event2, TemporalRelation.OVERLAPS):
                        relationships.append(f"{event1.name} overlaps {event2.name}")
                    elif self.check_temporal_relation(event1, event2, TemporalRelation.MEETS):
                        relationships.append(f"{event1.name} meets {event2.name}")
            
            return relationships
        
        def _identify_sequences(self, events: List[TemporalEvent]) -> List[List[str]]:
            """Identify sequences of consecutive events"""
            if not events:
                return []
            
            sorted_events = sorted(events, key=lambda e: e.start_time)
            sequences = []
            current_sequence = [sorted_events[0].name]
            
            for i in range(1, len(sorted_events)):
                prev_end = sorted_events[i-1].end_time or sorted_events[i-1].start_time
                curr_start = sorted_events[i].start_time
                
                # Check if events are consecutive (within 1 hour)
                if (curr_start - prev_end)  1:
                        sequences.append(current_sequence)
                    current_sequence = [sorted_events[i].name]
            
            if len(current_sequence) > 1:
                sequences.append(current_sequence)
            
            return sequences
        
        def _detect_regular_intervals(self, sorted_events: List[TemporalEvent]) -> Optional[timedelta]:
            """Detect if events occur at regular intervals"""
            if len(sorted_events) < 3:
                return None
            
            intervals = []
            for i in range(1, len(sorted_events)):
                interval = sorted_events[i].start_time - sorted_events[i-1].start_time
                intervals.append(interval)
            
            # Check if intervals are similar (within 10% variance)
            avg_interval = sum(intervals, timedelta(0)) / len(intervals)
            variance = sum((i - avg_interval).total_seconds()**2 for i in intervals) / len(intervals)
            
            if variance  int:
            """Detect clusters of events happening close together"""
            if len(sorted_events)  timedelta(hours=2):  # New cluster if gap > 2 hours
                    clusters += 1
            
            return clusters
        
        def _create_event_descriptions(self, events: List[TemporalEvent]) -> str:
            """Create textual descriptions of events"""
            descriptions = []
            for event in sorted(events, key=lambda e: e.start_time):
                desc = f"{event.name}: {event.start_time.strftime('%Y-%m-%d %H:%M')}"
                if event.end_time:
                    desc += f" to {event.end_time.strftime('%Y-%m-%d %H:%M')}"
                    desc += f" (duration: {event.duration})"
                if event.description:
                    desc += f" - {event.description}"
                descriptions.append(desc)
            return "\n".join(descriptions)
        
        def _create_timeline_visualization(self, events: List[TemporalEvent]) -> str:
            """Create ASCII timeline visualization"""
            if not events:
                return "No events to visualize"
            
            sorted_events = sorted(events, key=lambda e: e.start_time)
            
            lines = ["Timeline:", "=" * 60]
            
            for event in sorted_events:
                time_str = event.start_time.strftime('%Y-%m-%d %H:%M')
                if event.end_time:
                    duration_str = f" [{event.duration}]"
                else:
                    duration_str = " [instant]"
                
                lines.append(f"  {time_str} | {event.name}{duration_str}")
            
            return "\n".join(lines)
    
    
    def demonstrate_temporal_reasoning():
        """Demonstrate temporal reasoning capabilities"""
        print("=" * 80)
        print("TEMPORAL REASONING PATTERN DEMONSTRATION")
        print("=" * 80)
        
        agent = TemporalReasoningAgent()
        
        # Example 1: Project timeline analysis
        print("\n" + "=" * 80)
        print("Example 1: Project Timeline Analysis")
        print("=" * 80)
        
        base_time = datetime(2024, 1, 1, 9, 0)
        
        project_events = [
            TemporalEvent("Planning", base_time, base_time + timedelta(days=7), "Initial planning phase"),
            TemporalEvent("Design", base_time + timedelta(days=7), base_time + timedelta(days=14), "Design phase"),
            TemporalEvent("Development", base_time + timedelta(days=14), base_time + timedelta(days=45), "Development phase"),
            TemporalEvent("Testing", base_time + timedelta(days=40), base_time + timedelta(days=52), "Testing phase overlaps with dev"),
            TemporalEvent("Deployment", base_time + timedelta(days=52), base_time + timedelta(days=53), "Final deployment")
        ]
        
        query = TemporalQuery(
            question="What is the timeline of the project and are there any overlapping phases?",
            events=project_events,
            context="Software project timeline"
        )
        
        analysis = agent.analyze_temporal_query(query)
        
        print(f"\nQuery: {query.question}")
        print(f"\n{analysis.timeline}")
        print(f"\nTemporal Relationships:")
        for rel in analysis.relationships[:5]:
            print(f"  - {rel}")
        print(f"\nAnalysis:\n{analysis.answer}")
        
        # Example 2: Overlapping events detection
        print("\n" + "=" * 80)
        print("Example 2: Overlapping Events Detection")
        print("=" * 80)
        
        meeting_events = [
            TemporalEvent("Team Meeting", datetime(2024, 1, 15, 10, 0), datetime(2024, 1, 15, 11, 0)),
            TemporalEvent("Client Call", datetime(2024, 1, 15, 10, 30), datetime(2024, 1, 15, 11, 30)),
            TemporalEvent("Lunch Break", datetime(2024, 1, 15, 12, 0), datetime(2024, 1, 15, 13, 0)),
            TemporalEvent("Code Review", datetime(2024, 1, 15, 14, 0), datetime(2024, 1, 15, 15, 0))
        ]
        
        overlaps = agent.find_overlapping_events(meeting_events)
        
        print("\nSchedule for January 15, 2024:")
        for event in sorted(meeting_events, key=lambda e: e.start_time):
            print(f"  {event.start_time.strftime('%H:%M')}-{event.end_time.strftime('%H:%M')}: {event.name}")
        
        if overlaps:
            print(f"\nConflicts detected:")
            for event1, event2 in overlaps:
                print(f"  ⚠ {event1} overlaps with {event2}")
        else:
            print("\n✓ No scheduling conflicts detected")
        
        # Example 3: Temporal relationship checking
        print("\n" + "=" * 80)
        print("Example 3: Temporal Relationship Checking")
        print("=" * 80)
        
        event_a = TemporalEvent("Event A", datetime(2024, 1, 1, 10, 0), datetime(2024, 1, 1, 12, 0))
        event_b = TemporalEvent("Event B", datetime(2024, 1, 1, 11, 0), datetime(2024, 1, 1, 11, 30))
        event_c = TemporalEvent("Event C", datetime(2024, 1, 1, 12, 0), datetime(2024, 1, 1, 13, 0))
        
        relations_to_check = [
            (event_a, event_b, TemporalRelation.CONTAINS),
            (event_b, event_a, TemporalRelation.DURING),
            (event_a, event_c, TemporalRelation.MEETS),
            (event_c, event_a, TemporalRelation.MET_BY),
            (event_a, event_b, TemporalRelation.OVERLAPS)
        ]
        
        print("\nChecking temporal relationships:")
        for evt1, evt2, relation in relations_to_check:
            holds = agent.check_temporal_relation(evt1, evt2, relation)
            symbol = "✓" if holds else "✗"
            print(f"  {symbol} {evt1.name} {relation.value} {evt2.name}: {holds}")
        
        # Example 4: Duration calculations
        print("\n" + "=" * 80)
        print("Example 4: Duration Calculations")
        print("=" * 80)
        
        work_events = [
            TemporalEvent("Task 1", datetime(2024, 1, 1, 9, 0), datetime(2024, 1, 1, 11, 0)),
            TemporalEvent("Task 2", datetime(2024, 1, 1, 10, 30), datetime(2024, 1, 1, 12, 0)),
            TemporalEvent("Task 3", datetime(2024, 1, 1, 13, 0), datetime(2024, 1, 1, 15, 0)),
        ]
        
        total_simple = agent.calculate_total_duration(work_events, merge_overlaps=False)
        total_merged = agent.calculate_total_duration(work_events, merge_overlaps=True)
        
        print("\nWork events:")
        for event in work_events:
            print(f"  {event.name}: {event.duration}")
        
        print(f"\nTotal duration (simple sum): {total_simple}")
        print(f"Total duration (merged overlaps): {total_merged}")
        print(f"Overlap time: {total_simple - total_merged}")
        
        # Example 5: Gap detection
        print("\n" + "=" * 80)
        print("Example 5: Gap Detection")
        print("=" * 80)
        
        sparse_events = [
            TemporalEvent("Morning Session", datetime(2024, 1, 1, 9, 0), datetime(2024, 1, 1, 10, 0)),
            TemporalEvent("Afternoon Session", datetime(2024, 1, 1, 14, 0), datetime(2024, 1, 1, 15, 0)),
            TemporalEvent("Evening Session", datetime(2024, 1, 1, 19, 0), datetime(2024, 1, 1, 20, 0))
        ]
        
        gaps = agent.find_gaps(sparse_events)
        
        print("\nEvents:")
        for event in sparse_events:
            print(f"  {event.start_time.strftime('%H:%M')}-{event.end_time.strftime('%H:%M')}: {event.name}")
        
        print(f"\nGaps detected: {len(gaps)}")
        for start, end in gaps:
            duration = end - start
            print(f"  Gap from {start.strftime('%H:%M')} to {end.strftime('%H:%M')} ({duration})")
        
        # Example 6: Pattern detection
        print("\n" + "=" * 80)
        print("Example 6: Pattern Detection")
        print("=" * 80)
        
        regular_events = [
            TemporalEvent(f"Daily Standup {i}", datetime(2024, 1, i, 9, 0), datetime(2024, 1, i, 9, 15))
            for i in range(1, 6)
        ]
        
        patterns = agent.detect_temporal_patterns(regular_events)
        
        print("\nEvents:")
        for event in regular_events:
            print(f"  {event.start_time.strftime('%Y-%m-%d %H:%M')}: {event.name}")
        
        print(f"\nPattern Analysis:")
        print(f"  Total events: {patterns['total_events']}")
        print(f"  Time span: {patterns['time_span']['start'].strftime('%Y-%m-%d')} to {patterns['time_span']['end'].strftime('%Y-%m-%d')}")
        if patterns['regular_intervals']:
            print(f"  Regular interval detected: {patterns['regular_intervals']}")
        print(f"  Number of clusters: {patterns['clusters']}")
        print(f"  Gaps: {patterns['gaps']}")
        print(f"  Overlaps: {patterns['overlaps']}")
        
        # Summary
        print("\n" + "=" * 80)
        print("SUMMARY")
        print("=" * 80)
        print("""
    The Temporal Reasoning pattern enables agents to:
    ✓ Understand temporal relationships (before, after, during, overlaps)
    ✓ Detect scheduling conflicts and overlaps
    ✓ Calculate durations and time gaps
    ✓ Identify temporal patterns and sequences
    ✓ Validate temporal constraints
    ✓ Reason about causality and event sequences
    ✓ Analyze timelines and schedules
    
    This pattern is valuable for:
    - Project and task scheduling
    - Calendar and meeting management
    - Event sequence analysis
    - Timeline construction and validation
    - Causal reasoning
    - Temporal pattern detection
    - Deadline and milestone tracking
        """)
    
    
    if __name__ == "__main__":
        demonstrate_temporal_reasoning()
    
        

    Agentic Design Pattern: Temporal Reasoning

    Pattern 164: Temporal Reasoning – Line-by-Line Explanation

    Overview

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


    Pattern Description

    Pattern 164: Temporal Reasoning

    Description: The Temporal Reasoning pattern enables agents to understand and reason about time-based relationships, sequences, durations, and temporal logic. This pattern helps agents solve problems involving timelines, schedules, causality, and temporal constraints.

    Components: 1. Timeline Manager: Maintains ordered sequences of events 2. Temporal Relation Analyzer: Identifies before/after/during/overlapping relationships 3. Duration Calculator: Computes time intervals and durations 4. Temporal Constraint Checker: Validates temporal constraints 5. Sequence Analyzer: Analyzes event sequences and patterns 6. Temporal Query Engine: Answers time-based questions

    Use Cases: – Schedule planning and optimization – Event sequence analysis – Causal reasoning – Historical timeline construction – Project timeline management – Deadline tracking – Temporal pattern detection

    Benefits: – Enables time-aware reasoning – Supports scheduling and planning – Detects temporal patterns – Validates temporal constraints – Better understanding of causality

    Trade-offs: – Complex time representation – Timezone handling challenges – Requires precise temporal data – May need calendar logic – Limited by data accuracy

    LangChain Implementation: Uses LLM for temporal relationship interpretation with structured datetime management and interval calculations. Combines symbolic temporal logic with natural language understanding for temporal queries.


    File Statistics

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

    Module Documentation

    """
    Pattern 164: Temporal Reasoning
    
    Description:
        The Temporal Reasoning pattern enables agents to understand and reason about time-based
        relationships, sequences, durations, and temporal logic. This pattern helps agents solve
        problems involving timelines, schedules, causality, and temporal constraints.
    
    Components:
        1. Timeline Manager: Maintains ordered sequences of events
        2. Temporal Relation Analyzer: Identifies before/after/during/overlapping relationships
        3. Duration Calculator: Computes time intervals and durations
        4. Temporal Constraint Checker: Validates temporal constraints
        5. Sequence Analyzer: Analyzes event sequences and patterns
        6. Temporal Query Engine: Answers time-based questions
    
    Use Cases:
        - Schedule planning and optimization
        - Event sequence analysis
        - Causal reasoning
        - Historical timeline construction
        - Project timeline management
        - Deadline tracking
        - Temporal pattern detection
    
    Benefits:
        - Enables time-aware reasoning
        - Supports scheduling and planning
        - Detects temporal patterns
        - Validates temporal constraints
        - Better understanding of causality
    
    Trade-offs:
        - Complex time representation
        - Timezone handling challenges
        - Requires precise temporal data
        - May need calendar logic
        - Limited by data accuracy
    
    LangChain Implementation:
        Uses LLM for temporal relationship interpretation with structured datetime
        management and interval calculations. Combines symbolic temporal logic with
        natural language understanding for temporal queries.
    """
    

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


    Import Statements

    Standard Library Imports

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

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

    LangChain Imports

    from langchain_openai.ChatOpenAI
    

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

    Other Imports

    from dotenv.load_dotenv
    

    Purpose: Additional dependencies for specialized functionality.


    Classes

    Class: TemporalRelation

    Lines 60-74

    Inherits from: Enum

    Purpose:

    Types of temporal relationships (Allen's interval algebra)
    

    Class: TimeUnit

    Lines 77-85

    Inherits from: Enum

    Purpose:

    Units of time measurement
    

    Class: TemporalEvent

    Lines 89-107

    Purpose:

    Represents an event with temporal properties
    

    Methods:

    duration(self)

    Lines 98-102

    Decorators: property

    Purpose:

    Calculate duration of the event
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D7B3590>

    is_point_event(self)

    Lines 105-107

    Decorators: property

    Purpose:

    Check if this is a point event (no duration)
    

    Parameters:

    Returns: bool


    Class: Timeline

    Lines 111-124

    Purpose:

    Represents a timeline with multiple events
    

    Methods:

    add_event(self, event)

    Lines 116-119

    Purpose:

    Add an event to the timeline
    

    Parameters: – event: Event

    get_events_in_range(self, start, end)

    Lines 121-124

    Purpose:

    Get all events within a time range
    

    Parameters: – start: Start – end: End

    Returns: <ast.Subscript object at 0x000001A06D826710>


    Class: TemporalQuery

    Lines 128-132

    Purpose:

    A temporal reasoning query
    

    Class: TemporalAnalysis

    Lines 136-142

    Purpose:

    Result of temporal reasoning
    

    Class: TemporalReasoningAgent

    Lines 145-452

    Purpose:

    Agent that performs temporal reasoning tasks
    

    Methods:

    __init__(self, model_name)

    Lines 148-177

    Purpose:

    Initialize the temporal reasoning agent
    

    Parameters: – model_name: Language model instance

    analyze_temporal_query(self, query)

    Lines 179-205

    Purpose:

    Analyze a temporal query and return comprehensive analysis
    

    Parameters: – query: Input query or question

    Returns: TemporalAnalysis

    check_temporal_relation(self, event1, event2, relation)

    Lines 207-243

    Purpose:

    Check if a specific temporal relationship holds between two events
    

    Parameters: – event1: Event1 – event2: Event2 – relation: Relation

    Returns: bool

    find_overlapping_events(self, events)

    Lines 245-254

    Purpose:

    Find all pairs of overlapping events
    

    Parameters: – events: Events

    Returns: <ast.Subscript object at 0x000001A06D759190>

    find_gaps(self, events)

    Lines 256-271

    Purpose:

    Find time gaps between consecutive events
    

    Parameters: – events: Events

    Returns: <ast.Subscript object at 0x000001A06D7A0E10>

    calculate_total_duration(self, events, merge_overlaps)

    Lines 273-313

    Purpose:

    Calculate total duration of events
    

    Parameters: – events: Events – merge_overlaps: Merge overlaps

    Returns: timedelta

    detect_temporal_patterns(self, events)

    Lines 315-334

    Purpose:

    Detect patterns in temporal data
    

    Parameters: – events: Events

    Returns: <ast.Subscript object at 0x000001A06D624BD0>

    _events_overlap(self, event1, event2)

    Lines 336-343

    Purpose:

    Check if two events overlap in time
    

    Parameters: – event1: Event1 – event2: Event2

    Returns: bool

    _identify_temporal_relationships(self, events)

    Lines 345-361

    Purpose:

    Identify temporal relationships between events
    

    Parameters: – events: Events

    Returns: <ast.Subscript object at 0x000001A06D867290>

    _identify_sequences(self, events)

    Lines 363-387

    Purpose:

    Identify sequences of consecutive events
    

    Parameters: – events: Events

    Returns: <ast.Subscript object at 0x000001A06D77F010>

    _detect_regular_intervals(self, sorted_events)

    Lines 389-406

    Purpose:

    Detect if events occur at regular intervals
    

    Parameters: – sorted_events: Sorted events

    Returns: <ast.Subscript object at 0x000001A06D72AD90>

    _detect_time_clusters(self, sorted_events)

    Lines 408-419

    Purpose:

    Detect clusters of events happening close together
    

    Parameters: – sorted_events: Sorted events

    Returns: int

    _create_event_descriptions(self, events)

    Lines 421-432

    Purpose:

    Create textual descriptions of events
    

    Parameters: – events: Events

    Returns: str

    _create_timeline_visualization(self, events)

    Lines 434-452

    Purpose:

    Create ASCII timeline visualization
    

    Parameters: – events: Events

    Returns: str


    Functions

    demonstrate_temporal_reasoning()

    Lines 455-632

    Purpose:

    Demonstrate temporal reasoning capabilities
    

    Key Concepts Summary

    Pattern Overview

    The Temporal 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 163
    • Pattern 165

    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 Temporal 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: 164_temporal_reasoning.py
    Total Lines: 637
    Pattern Category: AI Agent Design Pattern #164


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Spatial Reasoning

    
        """
    Pattern 163: Spatial Reasoning
    
    Description:
        The Spatial Reasoning pattern enables agents to understand and reason about spatial
        relationships, positions, orientations, and geometric properties. This pattern helps
        agents solve problems involving physical space, layouts, navigation, and spatial
        configurations.
    
    Components:
        1. Coordinate System: Tracks positions and locations
        2. Relationship Analyzer: Identifies spatial relationships (above, below, left, right, etc.)
        3. Distance Calculator: Computes distances and proximity
        4. Orientation Tracker: Manages directions and rotations
        5. Spatial Query Engine: Answers spatial questions
        6. Visualization Helper: Generates spatial descriptions
    
    Use Cases:
        - Navigation and pathfinding
        - Layout design and optimization
        - Object placement and arrangement
        - Map understanding and interpretation
        - Geometry problem solving
        - Spatial puzzle solving
        - Floor plan analysis
    
    Benefits:
        - Enables spatial understanding
        - Supports geometric reasoning
        - Facilitates navigation tasks
        - Improves layout optimization
        - Better scene understanding
    
    Trade-offs:
        - Complex coordinate management
        - Requires clear spatial representation
        - May need visualization capabilities
        - Challenging for abstract spaces
        - Limited by 2D/3D representation
    
    LangChain Implementation:
        Uses LLM for spatial relationship interpretation with structured coordinate
        systems and distance calculations. Combines symbolic representation with
        natural language understanding for spatial queries.
    """
    
    import os
    import math
    from typing import List, Dict, Any, Tuple, Optional
    from dataclasses import dataclass
    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
    
    load_dotenv()
    
    
    class Direction(Enum):
        """Cardinal and ordinal directions"""
        NORTH = "north"
        SOUTH = "south"
        EAST = "east"
        WEST = "west"
        NORTHEAST = "northeast"
        NORTHWEST = "northwest"
        SOUTHEAST = "southeast"
        SOUTHWEST = "southwest"
        UP = "up"
        DOWN = "down"
    
    
    class SpatialRelation(Enum):
        """Types of spatial relationships"""
        ABOVE = "above"
        BELOW = "below"
        LEFT_OF = "left of"
        RIGHT_OF = "right of"
        IN_FRONT_OF = "in front of"
        BEHIND = "behind"
        INSIDE = "inside"
        OUTSIDE = "outside"
        ADJACENT = "adjacent"
        NEAR = "near"
        FAR = "far"
        BETWEEN = "between"
    
    
    @dataclass
    class SpatialObject:
        """Represents an object in space"""
        name: str
        x: float
        y: float
        z: float = 0.0
        width: float = 1.0
        height: float = 1.0
        depth: float = 1.0
        orientation: float = 0.0  # degrees
    
    
    @dataclass
    class SpatialQuery:
        """A spatial reasoning query"""
        question: str
        objects: List[SpatialObject]
        context: str = ""
    
    
    @dataclass
    class SpatialAnalysis:
        """Result of spatial reasoning"""
        answer: str
        relationships: List[str]
        distances: Dict[str, float]
        reasoning: str
        visualization: str
    
    
    class SpatialReasoningAgent:
        """Agent that performs spatial reasoning tasks"""
        
        def __init__(self, model_name: str = "gpt-3.5-turbo"):
            """Initialize the spatial reasoning agent"""
            self.llm = ChatOpenAI(model=model_name, temperature=0)
            
            # Prompts for different spatial reasoning tasks
            self.relation_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a spatial reasoning expert. Analyze the positions 
                of objects and describe their spatial relationships clearly and accurately.
                Consider distances, directions, and relative positions."""),
                ("user", """Objects and their positions:
    {object_descriptions}
    
    Question: {question}
    
    Provide a clear answer based on the spatial relationships.""")
            ])
            
            self.path_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a navigation expert. Given object positions, 
                determine optimal paths and directions."""),
                ("user", """Objects and their positions:
    {object_descriptions}
    
    Navigation task: {task}
    
    Describe the path or navigation instructions.""")
            ])
            
            self.relation_chain = self.relation_prompt | self.llm | StrOutputParser()
            self.path_chain = self.path_prompt | self.llm | StrOutputParser()
        
        def analyze_spatial_query(self, query: SpatialQuery) -> SpatialAnalysis:
            """Analyze a spatial query and return comprehensive analysis"""
            # Calculate all pairwise distances
            distances = self._calculate_distances(query.objects)
            
            # Identify spatial relationships
            relationships = self._identify_relationships(query.objects)
            
            # Create object descriptions
            object_descriptions = self._create_object_descriptions(query.objects)
            
            # Get LLM reasoning
            reasoning = self.relation_chain.invoke({
                "object_descriptions": object_descriptions,
                "question": query.question
            })
            
            # Create visualization
            visualization = self._create_visualization(query.objects)
            
            return SpatialAnalysis(
                answer=reasoning,
                relationships=relationships,
                distances=distances,
                reasoning=reasoning,
                visualization=visualization
            )
        
        def find_path(self, start: SpatialObject, end: SpatialObject, 
                      obstacles: List[SpatialObject]) -> Dict[str, Any]:
            """Find path from start to end avoiding obstacles"""
            # Calculate direct distance
            direct_distance = self._distance(start, end)
            
            # Calculate direction
            direction = self._get_direction(start, end)
            
            # Check for obstacles in path
            obstacles_in_path = self._check_obstacles(start, end, obstacles)
            
            # Get navigation instructions
            object_descriptions = self._create_object_descriptions(
                [start, end] + obstacles
            )
            
            instructions = self.path_chain.invoke({
                "object_descriptions": object_descriptions,
                "task": f"Navigate from {start.name} to {end.name}"
            })
            
            return {
                "direct_distance": direct_distance,
                "direction": direction,
                "obstacles_in_path": obstacles_in_path,
                "instructions": instructions,
                "path_clear": len(obstacles_in_path) == 0
            }
        
        def find_nearest(self, reference: SpatialObject, 
                         candidates: List[SpatialObject]) -> Tuple[SpatialObject, float]:
            """Find the nearest object to reference from candidates"""
            nearest = None
            min_distance = float('inf')
            
            for candidate in candidates:
                distance = self._distance(reference, candidate)
                if distance  bool:
            """Check if a specific spatial relationship holds between two objects"""
            if relation == SpatialRelation.ABOVE:
                return obj1.z > obj2.z
            elif relation == SpatialRelation.BELOW:
                return obj1.z < obj2.z
            elif relation == SpatialRelation.LEFT_OF:
                return obj1.x  obj2.x
            elif relation == SpatialRelation.IN_FRONT_OF:
                return obj1.y > obj2.y
            elif relation == SpatialRelation.BEHIND:
                return obj1.y < obj2.y
            elif relation == SpatialRelation.NEAR:
                distance = self._distance(obj1, obj2)
                return distance  10.0  # Threshold for "far"
            elif relation == SpatialRelation.ADJACENT:
                distance = self._distance(obj1, obj2)
                return distance  List[SpatialObject]:
            """Get all objects within a spherical region"""
            result = []
            cx, cy, cz = center
            
            for obj in objects:
                distance = math.sqrt(
                    (obj.x - cx)**2 + 
                    (obj.y - cy)**2 + 
                    (obj.z - cz)**2
                )
                if distance  float:
            """Calculate Euclidean distance between two objects"""
            return math.sqrt(
                (obj1.x - obj2.x)**2 + 
                (obj1.y - obj2.y)**2 + 
                (obj1.z - obj2.z)**2
            )
        
        def _calculate_distances(self, objects: List[SpatialObject]) -> Dict[str, float]:
            """Calculate all pairwise distances"""
            distances = {}
            for i, obj1 in enumerate(objects):
                for obj2 in objects[i+1:]:
                    key = f"{obj1.name} to {obj2.name}"
                    distances[key] = self._distance(obj1, obj2)
            return distances
        
        def _identify_relationships(self, objects: List[SpatialObject]) -> List[str]:
            """Identify spatial relationships between objects"""
            relationships = []
            
            for i, obj1 in enumerate(objects):
                for obj2 in objects[i+1:]:
                    # Check various relationships
                    if self.check_relationship(obj1, obj2, SpatialRelation.ABOVE):
                        relationships.append(f"{obj1.name} is above {obj2.name}")
                    if self.check_relationship(obj1, obj2, SpatialRelation.BELOW):
                        relationships.append(f"{obj1.name} is below {obj2.name}")
                    if self.check_relationship(obj1, obj2, SpatialRelation.LEFT_OF):
                        relationships.append(f"{obj1.name} is left of {obj2.name}")
                    if self.check_relationship(obj1, obj2, SpatialRelation.RIGHT_OF):
                        relationships.append(f"{obj1.name} is right of {obj2.name}")
                    if self.check_relationship(obj1, obj2, SpatialRelation.NEAR):
                        relationships.append(f"{obj1.name} is near {obj2.name}")
                    if self.check_relationship(obj1, obj2, SpatialRelation.ADJACENT):
                        relationships.append(f"{obj1.name} is adjacent to {obj2.name}")
            
            return relationships
        
        def _get_direction(self, from_obj: SpatialObject, 
                           to_obj: SpatialObject) -> Direction:
            """Determine cardinal direction from one object to another"""
            dx = to_obj.x - from_obj.x
            dy = to_obj.y - from_obj.y
            
            angle = math.atan2(dy, dx) * 180 / math.pi
            
            # Normalize to 0-360
            if angle < 0:
                angle += 360
            
            # Determine direction
            if 22.5 <= angle < 67.5:
                return Direction.NORTHEAST
            elif 67.5 <= angle < 112.5:
                return Direction.NORTH
            elif 112.5 <= angle < 157.5:
                return Direction.NORTHWEST
            elif 157.5 <= angle < 202.5:
                return Direction.WEST
            elif 202.5 <= angle < 247.5:
                return Direction.SOUTHWEST
            elif 247.5 <= angle < 292.5:
                return Direction.SOUTH
            elif 292.5 <= angle  List[str]:
            """Check which obstacles are in the path between start and end"""
            obstacles_in_path = []
            
            for obstacle in obstacles:
                # Simple line intersection check
                if self._is_near_line(start, end, obstacle):
                    obstacles_in_path.append(obstacle.name)
            
            return obstacles_in_path
        
        def _is_near_line(self, start: SpatialObject, end: SpatialObject,
                          point: SpatialObject, threshold: float = 1.0) -> bool:
            """Check if a point is near the line between start and end"""
            # Calculate perpendicular distance from point to line
            x1, y1 = start.x, start.y
            x2, y2 = end.x, end.y
            x0, y0 = point.x, point.y
            
            numerator = abs((y2-y1)*x0 - (x2-x1)*y0 + x2*y1 - y2*x1)
            denominator = math.sqrt((y2-y1)**2 + (x2-x1)**2)
            
            if denominator == 0:
                return False
            
            distance = numerator / denominator
            
            # Also check if point is between start and end
            between = (min(x1, x2) <= x0 <= max(x1, x2) and 
                      min(y1, y2) <= y0 <= max(y1, y2))
            
            return distance  str:
            """Create textual descriptions of object positions"""
            descriptions = []
            for obj in objects:
                desc = f"{obj.name}: position ({obj.x:.1f}, {obj.y:.1f}, {obj.z:.1f})"
                if obj.width != 1.0 or obj.height != 1.0:
                    desc += f", size ({obj.width:.1f}x{obj.height:.1f}x{obj.depth:.1f})"
                descriptions.append(desc)
            return "\n".join(descriptions)
        
        def _create_visualization(self, objects: List[SpatialObject]) -> str:
            """Create ASCII visualization of object positions"""
            if not objects:
                return "No objects to visualize"
            
            # Find bounds
            min_x = min(obj.x for obj in objects) - 2
            max_x = max(obj.x for obj in objects) + 2
            min_y = min(obj.y for obj in objects) - 2
            max_y = max(obj.y for obj in objects) + 2
            
            # Create grid
            width = int(max_x - min_x) + 1
            height = int(max_y - min_y) + 1
            
            grid = [[' ' for _ in range(width)] for _ in range(height)]
            
            # Place objects
            for i, obj in enumerate(objects):
                x = int(obj.x - min_x)
                y = int(obj.y - min_y)
                if 0 <= x < width and 0 <= y < height:
                    grid[height - 1 - y][x] = chr(65 + i)  # A, B, C, ...
            
            # Create visualization string
            vis = "\n" + "\n".join("".join(row) for row in grid)
            vis += "\n\nLegend: " + ", ".join(
                f"{chr(65+i)}={obj.name}" for i, obj in enumerate(objects)
            )
            
            return vis
    
    
    def demonstrate_spatial_reasoning():
        """Demonstrate spatial reasoning capabilities"""
        print("=" * 80)
        print("SPATIAL REASONING PATTERN DEMONSTRATION")
        print("=" * 80)
        
        agent = SpatialReasoningAgent()
        
        # Example 1: Room layout analysis
        print("\n" + "=" * 80)
        print("Example 1: Room Layout Analysis")
        print("=" * 80)
        
        room_objects = [
            SpatialObject("Table", 5, 5, 0, 2, 1, 3),
            SpatialObject("Chair1", 4, 5, 0, 1, 1, 1),
            SpatialObject("Chair2", 6, 5, 0, 1, 1, 1),
            SpatialObject("Lamp", 5, 5, 2, 0.5, 0.5, 1),
            SpatialObject("Door", 0, 5, 0, 1, 2, 0.1),
            SpatialObject("Window", 10, 5, 1, 2, 1, 0.1)
        ]
        
        query = SpatialQuery(
            question="What is the spatial arrangement of furniture in the room?",
            objects=room_objects,
            context="Living room layout"
        )
        
        analysis = agent.analyze_spatial_query(query)
        
        print(f"\nQuery: {query.question}")
        print(f"\nAnswer: {analysis.answer}")
        print(f"\nKey Relationships:")
        for rel in analysis.relationships[:5]:
            print(f"  - {rel}")
        print(f"\nVisualization:{analysis.visualization}")
        
        # Example 2: Navigation pathfinding
        print("\n" + "=" * 80)
        print("Example 2: Navigation Pathfinding")
        print("=" * 80)
        
        start = SpatialObject("Start", 0, 0, 0)
        end = SpatialObject("Goal", 10, 10, 0)
        obstacles = [
            SpatialObject("Obstacle1", 5, 5, 0, 2, 2, 1),
            SpatialObject("Obstacle2", 7, 3, 0, 1, 1, 1)
        ]
        
        path_result = agent.find_path(start, end, obstacles)
        
        print(f"\nNavigation from {start.name} to {end.name}")
        print(f"Direct distance: {path_result['direct_distance']:.2f} units")
        print(f"Direction: {path_result['direction'].value}")
        print(f"Path clear: {path_result['path_clear']}")
        if path_result['obstacles_in_path']:
            print(f"Obstacles in path: {', '.join(path_result['obstacles_in_path'])}")
        print(f"\nInstructions:\n{path_result['instructions']}")
        
        # Example 3: Nearest neighbor search
        print("\n" + "=" * 80)
        print("Example 3: Nearest Neighbor Search")
        print("=" * 80)
        
        reference = SpatialObject("Home", 5, 5, 0)
        stores = [
            SpatialObject("Grocery Store", 3, 4, 0),
            SpatialObject("Pharmacy", 8, 6, 0),
            SpatialObject("Coffee Shop", 5, 7, 0),
            SpatialObject("Bookstore", 10, 10, 0)
        ]
        
        nearest, distance = agent.find_nearest(reference, stores)
        
        print(f"\nFinding nearest store to {reference.name}")
        print(f"Nearest: {nearest.name}")
        print(f"Distance: {distance:.2f} units")
        
        print("\nAll distances:")
        for store in stores:
            dist = agent._distance(reference, store)
            direction = agent._get_direction(reference, store)
            print(f"  {store.name}: {dist:.2f} units {direction.value}")
        
        # Example 4: Spatial relationship checking
        print("\n" + "=" * 80)
        print("Example 4: Spatial Relationship Checking")
        print("=" * 80)
        
        ceiling_light = SpatialObject("Ceiling Light", 5, 5, 8)
        floor_lamp = SpatialObject("Floor Lamp", 5, 5, 0)
        couch = SpatialObject("Couch", 3, 5, 0)
        
        relations_to_check = [
            (ceiling_light, floor_lamp, SpatialRelation.ABOVE),
            (floor_lamp, ceiling_light, SpatialRelation.BELOW),
            (couch, floor_lamp, SpatialRelation.NEAR),
            (couch, floor_lamp, SpatialRelation.ADJACENT)
        ]
        
        print("\nChecking spatial relationships:")
        for obj1, obj2, relation in relations_to_check:
            holds = agent.check_relationship(obj1, obj2, relation)
            print(f"  {obj1.name} is {relation.value} {obj2.name}: {holds}")
        
        # Example 5: Regional search
        print("\n" + "=" * 80)
        print("Example 5: Regional Search")
        print("=" * 80)
        
        all_objects = [
            SpatialObject("Object1", 2, 2, 0),
            SpatialObject("Object2", 5, 5, 0),
            SpatialObject("Object3", 8, 8, 0),
            SpatialObject("Object4", 3, 7, 0),
            SpatialObject("Object5", 10, 2, 0)
        ]
        
        search_center = (5, 5, 0)
        search_radius = 4
        
        objects_in_region = agent.get_objects_in_region(
            all_objects, search_center, search_radius
        )
        
        print(f"\nSearching for objects within {search_radius} units of {search_center}")
        print(f"Found {len(objects_in_region)} objects:")
        for obj in objects_in_region:
            dist = math.sqrt(
                (obj.x - search_center[0])**2 + 
                (obj.y - search_center[1])**2 + 
                (obj.z - search_center[2])**2
            )
            print(f"  {obj.name} at ({obj.x}, {obj.y}, {obj.z}) - {dist:.2f} units away")
        
        # Example 6: Complex spatial query
        print("\n" + "=" * 80)
        print("Example 6: Complex Spatial Query")
        print("=" * 80)
        
        city_objects = [
            SpatialObject("City Hall", 5, 5, 0, 3, 3, 10),
            SpatialObject("Park", 8, 8, 0, 5, 5, 0),
            SpatialObject("School", 2, 8, 0, 2, 2, 5),
            SpatialObject("Hospital", 9, 2, 0, 3, 3, 8),
            SpatialObject("Library", 1, 2, 0, 2, 2, 4)
        ]
        
        complex_query = SpatialQuery(
            question="Which public facilities are located in the northern part of the city and which are in the south?",
            objects=city_objects,
            context="City planning analysis"
        )
        
        complex_analysis = agent.analyze_spatial_query(complex_query)
        
        print(f"\nQuery: {complex_query.question}")
        print(f"\nAnalysis:\n{complex_analysis.answer}")
        
        # Summary
        print("\n" + "=" * 80)
        print("SUMMARY")
        print("=" * 80)
        print("""
    The Spatial Reasoning pattern enables agents to:
    ✓ Understand spatial relationships and positions
    ✓ Calculate distances and directions
    ✓ Find optimal paths and navigation routes
    ✓ Identify nearest neighbors
    ✓ Search within spatial regions
    ✓ Analyze complex spatial arrangements
    ✓ Combine geometric calculations with semantic understanding
    
    This pattern is valuable for:
    - Navigation and routing systems
    - Layout and design optimization
    - Geographic information systems
    - Robotics and autonomous vehicles
    - Game AI and virtual environments
    - Spatial data analysis
    - Map-based applications
        """)
    
    
    if __name__ == "__main__":
        demonstrate_spatial_reasoning()
    
        

    Agentic Design Pattern: Spatial Reasoning

    Pattern 163: Spatial Reasoning – Line-by-Line Explanation

    Overview

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


    Pattern Description

    Pattern 163: Spatial Reasoning

    Description: The Spatial Reasoning pattern enables agents to understand and reason about spatial relationships, positions, orientations, and geometric properties. This pattern helps agents solve problems involving physical space, layouts, navigation, and spatial configurations.

    Components: 1. Coordinate System: Tracks positions and locations 2. Relationship Analyzer: Identifies spatial relationships (above, below, left, right, etc.) 3. Distance Calculator: Computes distances and proximity 4. Orientation Tracker: Manages directions and rotations 5. Spatial Query Engine: Answers spatial questions 6. Visualization Helper: Generates spatial descriptions

    Use Cases: – Navigation and pathfinding – Layout design and optimization – Object placement and arrangement – Map understanding and interpretation – Geometry problem solving – Spatial puzzle solving – Floor plan analysis

    Benefits: – Enables spatial understanding – Supports geometric reasoning – Facilitates navigation tasks – Improves layout optimization – Better scene understanding

    Trade-offs: – Complex coordinate management – Requires clear spatial representation – May need visualization capabilities – Challenging for abstract spaces – Limited by 2D/3D representation

    LangChain Implementation: Uses LLM for spatial relationship interpretation with structured coordinate systems and distance calculations. Combines symbolic representation with natural language understanding for spatial queries.


    File Statistics

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

    Module Documentation

    """
    Pattern 163: Spatial Reasoning
    
    Description:
        The Spatial Reasoning pattern enables agents to understand and reason about spatial
        relationships, positions, orientations, and geometric properties. This pattern helps
        agents solve problems involving physical space, layouts, navigation, and spatial
        configurations.
    
    Components:
        1. Coordinate System: Tracks positions and locations
        2. Relationship Analyzer: Identifies spatial relationships (above, below, left, right, etc.)
        3. Distance Calculator: Computes distances and proximity
        4. Orientation Tracker: Manages directions and rotations
        5. Spatial Query Engine: Answers spatial questions
        6. Visualization Helper: Generates spatial descriptions
    
    Use Cases:
        - Navigation and pathfinding
        - Layout design and optimization
        - Object placement and arrangement
        - Map understanding and interpretation
        - Geometry problem solving
        - Spatial puzzle solving
        - Floor plan analysis
    
    Benefits:
        - Enables spatial understanding
        - Supports geometric reasoning
        - Facilitates navigation tasks
        - Improves layout optimization
        - Better scene understanding
    
    Trade-offs:
        - Complex coordinate management
        - Requires clear spatial representation
        - May need visualization capabilities
        - Challenging for abstract spaces
        - Limited by 2D/3D representation
    
    LangChain Implementation:
        Uses LLM for spatial relationship interpretation with structured coordinate
        systems and distance calculations. Combines symbolic representation with
        natural language understanding for spatial queries.
    """
    

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


    Import Statements

    Standard Library Imports

    import dataclasses.dataclass
    import enum.Enum
    import langchain_core.output_parsers.StrOutputParser
    import langchain_core.prompts.ChatPromptTemplate
    import os
    import typing.Any
    import typing.Dict
    import typing.List
    import typing.Optional
    import typing.Tuple
    

    Purpose: Core Python functionality: – dataclasses.dataclass: 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.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Tuple: Type hints for better code documentation (List, Dict, Any, Optional)

    LangChain Imports

    from langchain_openai.ChatOpenAI
    

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

    Other Imports

    from dotenv.load_dotenv
    from math
    

    Purpose: Additional dependencies for specialized functionality.


    Classes

    Class: Direction

    Lines 61-72

    Inherits from: Enum

    Purpose:

    Cardinal and ordinal directions
    

    Class: SpatialRelation

    Lines 75-88

    Inherits from: Enum

    Purpose:

    Types of spatial relationships
    

    Class: SpatialObject

    Lines 92-101

    Purpose:

    Represents an object in space
    

    Class: SpatialQuery

    Lines 105-109

    Purpose:

    A spatial reasoning query
    

    Class: SpatialAnalysis

    Lines 113-119

    Purpose:

    Result of spatial reasoning
    

    Class: SpatialReasoningAgent

    Lines 122-416

    Purpose:

    Agent that performs spatial reasoning tasks
    

    Methods:

    __init__(self, model_name)

    Lines 125-154

    Purpose:

    Initialize the spatial reasoning agent
    

    Parameters: – model_name: Language model instance

    analyze_spatial_query(self, query)

    Lines 156-182

    Purpose:

    Analyze a spatial query and return comprehensive analysis
    

    Parameters: – query: Input query or question

    Returns: SpatialAnalysis

    find_path(self, start, end, obstacles)

    Lines 184-212

    Purpose:

    Find path from start to end avoiding obstacles
    

    Parameters: – start: Start – end: End – obstacles: Obstacles

    Returns: <ast.Subscript object at 0x000001A06D17A650>

    find_nearest(self, reference, candidates)

    Lines 214-226

    Purpose:

    Find the nearest object to reference from candidates
    

    Parameters: – reference: Reference – candidates: Candidates

    Returns: <ast.Subscript object at 0x000001A06D60A550>

    check_relationship(self, obj1, obj2, relation)

    Lines 228-253

    Purpose:

    Check if a specific spatial relationship holds between two objects
    

    Parameters: – obj1: Obj1 – obj2: Obj2 – relation: Relation

    Returns: bool

    get_objects_in_region(self, objects, center, radius)

    Lines 255-271

    Purpose:

    Get all objects within a spherical region
    

    Parameters: – objects: Objects – center: Center – radius: Radius

    Returns: <ast.Subscript object at 0x000001A06D74A710>

    _distance(self, obj1, obj2)

    Lines 273-279

    Purpose:

    Calculate Euclidean distance between two objects
    

    Parameters: – obj1: Obj1 – obj2: Obj2

    Returns: float

    _calculate_distances(self, objects)

    Lines 281-288

    Purpose:

    Calculate all pairwise distances
    

    Parameters: – objects: Objects

    Returns: <ast.Subscript object at 0x000001A06D669750>

    _identify_relationships(self, objects)

    Lines 290-310

    Purpose:

    Identify spatial relationships between objects
    

    Parameters: – objects: Objects

    Returns: <ast.Subscript object at 0x000001A06D63A550>

    _get_direction(self, from_obj, to_obj)

    Lines 312-340

    Purpose:

    Determine cardinal direction from one object to another
    

    Parameters: – from_obj: From obj – to_obj: To obj

    Returns: Direction

    _check_obstacles(self, start, end, obstacles)

    Lines 342-352

    Purpose:

    Check which obstacles are in the path between start and end
    

    Parameters: – start: Start – end: End – obstacles: Obstacles

    Returns: <ast.Subscript object at 0x000001A06D7EFDD0>

    _is_near_line(self, start, end, point, threshold)

    Lines 354-374

    Purpose:

    Check if a point is near the line between start and end
    

    Parameters: – start: Start – end: End – point: Point – threshold: Threshold

    Returns: bool

    _create_object_descriptions(self, objects)

    Lines 376-384

    Purpose:

    Create textual descriptions of object positions
    

    Parameters: – objects: Objects

    Returns: str

    _create_visualization(self, objects)

    Lines 386-416

    Purpose:

    Create ASCII visualization of object positions
    

    Parameters: – objects: Objects

    Returns: str


    Functions

    demonstrate_spatial_reasoning()

    Lines 419-600

    Purpose:

    Demonstrate spatial reasoning capabilities
    

    Key Concepts Summary

    Pattern Overview

    The Spatial 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 162
    • Pattern 164

    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 Spatial 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: 163_spatial_reasoning.py
    Total Lines: 605
    Pattern Category: AI Agent Design Pattern #163


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Counterfactual Reasoning

    
        """
    Pattern 162: Counterfactual Reasoning
    
    Description:
        Counterfactual reasoning explores "what if" scenarios by considering alternative
        outcomes if certain conditions were different. It helps understand causality,
        evaluate decisions, and plan for contingencies.
    
    Components:
        - Scenario modification
        - Alternative outcome prediction
        - Causal analysis
        - Comparison with actual outcomes
        - Impact assessment
    
    Use Cases:
        - Decision analysis
        - Risk assessment
        - Causal inference
        - Policy evaluation
        - Learning from mistakes
    
    Benefits:
        - Understand causality
        - Evaluate alternatives
        - Learn from hypotheticals
        - Improve decision-making
    
    Trade-offs:
        - Speculative nature
        - Multiple possibilities
        - Requires causal model
        - May be computationally intensive
    
    LangChain Implementation:
        Uses ChatOpenAI for scenario generation, outcome prediction,
        and comparative analysis
    """
    
    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
    
    load_dotenv()
    
    
    @dataclass
    class Scenario:
        """Actual or counterfactual scenario"""
        description: str
        conditions: Dict[str, Any]
        outcome: Optional[str] = None
        is_actual: bool = True
    
    
    @dataclass
    class CounterfactualAnalysis:
        """Analysis of counterfactual scenario"""
        actual_scenario: Scenario
        counterfactual_scenario: Scenario
        predicted_outcome: str
        differences: List[str]
        causal_factors: List[str]
        likelihood: float  # 0-1
        reasoning: str
    
    
    class CounterfactualReasoningAgent:
        """Agent that performs counterfactual reasoning"""
        
        def __init__(self, model_name: str = "gpt-3.5-turbo"):
            self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        
        def analyze(self, actual_scenario: Scenario,
                    modifications: Dict[str, Any]) -> CounterfactualAnalysis:
            """Analyze counterfactual scenario"""
            
            # Create counterfactual scenario
            cf_conditions = actual_scenario.conditions.copy()
            cf_conditions.update(modifications)
            
            cf_scenario = Scenario(
                description=self._generate_cf_description(actual_scenario, modifications),
                conditions=cf_conditions,
                is_actual=False
            )
            
            # Predict counterfactual outcome
            predicted_outcome = self._predict_outcome(cf_scenario, actual_scenario)
            
            # Identify differences
            differences = self._identify_differences(actual_scenario, cf_scenario)
            
            # Analyze causal factors
            causal_factors = self._analyze_causality(actual_scenario, cf_scenario, differences)
            
            # Assess likelihood
            likelihood = self._assess_likelihood(cf_scenario, actual_scenario)
            
            # Generate reasoning
            reasoning = self._generate_reasoning(actual_scenario, cf_scenario, predicted_outcome)
            
            return CounterfactualAnalysis(
                actual_scenario=actual_scenario,
                counterfactual_scenario=cf_scenario,
                predicted_outcome=predicted_outcome,
                differences=differences,
                causal_factors=causal_factors,
                likelihood=likelihood,
                reasoning=reasoning
            )
        
        def _generate_cf_description(self, actual: Scenario, mods: Dict[str, Any]) -> str:
            """Generate counterfactual description"""
            prompt = ChatPromptTemplate.from_messages([
                ("system", "Generate a counterfactual scenario description."),
                ("user", """Actual scenario: {actual}
    
    Modifications: {modifications}
    
    Describe the counterfactual scenario in one sentence:""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            return chain.invoke({
                "actual": actual.description,
                "modifications": str(mods)
            })
        
        def _predict_outcome(self, cf_scenario: Scenario, actual: Scenario) -> str:
            """Predict counterfactual outcome"""
            prompt = ChatPromptTemplate.from_messages([
                ("system", "Predict what would have happened in this counterfactual scenario."),
                ("user", """Actual scenario: {actual}
    Actual outcome: {actual_outcome}
    
    Counterfactual scenario: {counterfactual}
    Conditions: {conditions}
    
    What would be the likely outcome?""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            return chain.invoke({
                "actual": actual.description,
                "actual_outcome": actual.outcome or "Unknown",
                "counterfactual": cf_scenario.description,
                "conditions": str(cf_scenario.conditions)
            })
        
        def _identify_differences(self, actual: Scenario, cf: Scenario) -> List[str]:
            """Identify key differences"""
            differences = []
            for key in set(actual.conditions.keys()) | set(cf.conditions.keys()):
                actual_val = actual.conditions.get(key)
                cf_val = cf.conditions.get(key)
                if actual_val != cf_val:
                    differences.append(f"{key}: {actual_val} → {cf_val}")
            return differences
        
        def _analyze_causality(self, actual: Scenario, cf: Scenario, 
                              differences: List[str]) -> List[str]:
            """Analyze causal factors"""
            prompt = ChatPromptTemplate.from_messages([
                ("system", "Identify the causal factors that would lead to different outcomes."),
                ("user", """Differences: {differences}
    
    What are the key causal factors? List 2-3 main factors.""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            response = chain.invoke({"differences": "\n".join(differences)})
            
            factors = []
            for line in response.split('\n'):
                line = line.strip()
                if line and (line[0].isdigit() or line.startswith('-')):
                    clean = line.lstrip('0123456789.-) ')
                    if clean:
                        factors.append(clean)
            return factors
        
        def _assess_likelihood(self, cf: Scenario, actual: Scenario) -> float:
            """Assess likelihood of counterfactual"""
            # Simplified: count condition changes
            total_conditions = len(actual.conditions)
            changed = sum(1 for k in actual.conditions 
                         if k in cf.conditions and actual.conditions[k] != cf.conditions[k])
            
            if total_conditions == 0:
                return 0.5
            
            # More changes = less likely
            return max(0.1, 1.0 - (changed / total_conditions))
        
        def _generate_reasoning(self, actual: Scenario, cf: Scenario, outcome: str) -> str:
            """Generate reasoning explanation"""
            prompt = ChatPromptTemplate.from_messages([
                ("system", "Explain the counterfactual reasoning."),
                ("user", """Actual: {actual} → {actual_outcome}
    Counterfactual: {cf} → {cf_outcome}
    
    Explain why the outcome would be different:""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            return chain.invoke({
                "actual": actual.description,
                "actual_outcome": actual.outcome or "Unknown",
                "cf": cf.description,
                "cf_outcome": outcome
            })
    
    
    def demonstrate_counterfactual_reasoning():
        """Demonstrate counterfactual reasoning pattern"""
        print("=" * 80)
        print("COUNTERFACTUAL REASONING PATTERN DEMONSTRATION")
        print("=" * 80)
        
        agent = CounterfactualReasoningAgent()
        
        # Example 1: Business decision
        print("\n" + "="*80)
        print("EXAMPLE 1: Business Decision Analysis")
        print("="*80)
        actual = Scenario(
            description="Company launched product in Q4 with minimal marketing",
            conditions={"launch_quarter": "Q4", "marketing_budget": "low", "features": "basic"},
            outcome="Slow initial sales, 1000 units sold",
            is_actual=True
        )
        
        modifications = {"marketing_budget": "high", "launch_quarter": "Q1"}
        
        analysis = agent.analyze(actual, modifications)
        
        print(f"Actual: {actual.description}")
        print(f"Outcome: {actual.outcome}")
        print(f"\nCounterfactual: {analysis.counterfactual_scenario.description}")
        print(f"Predicted outcome: {analysis.predicted_outcome}")
        print(f"\nKey differences:")
        for diff in analysis.differences:
            print(f"  - {diff}")
        print(f"\nLikelihood: {analysis.likelihood:.1%}")
        
        # Example 2: Historical what-if
        print("\n" + "="*80)
        print("EXAMPLE 2: Historical Counterfactual")
        print("="*80)
        actual = Scenario(
            description="Student studied for 2 hours before the exam",
            conditions={"study_time": 2, "sleep": "adequate", "practice_tests": 1},
            outcome="Scored 75% on the exam",
            is_actual=True
        )
        
        modifications = {"study_time": 6, "practice_tests": 3}
        
        analysis = agent.analyze(actual, modifications)
        
        print(f"What actually happened: {actual.description}")
        print(f"Result: {actual.outcome}")
        print(f"\nWhat if: {analysis.counterfactual_scenario.description}")
        print(f"Predicted result: {analysis.predicted_outcome[:200]}...")
        print(f"\nCausal factors:")
        for factor in analysis.causal_factors:
            print(f"  - {factor}")
        
        # Example 3: Risk assessment
        print("\n" + "="*80)
        print("EXAMPLE 3: Risk Assessment")
        print("="*80)
        actual = Scenario(
            description="Project completed on time with original team",
            conditions={"team_size": 5, "timeline": "6 months", "requirements_stable": True},
            outcome="Successful delivery, high quality",
            is_actual=True
        )
        
        modifications = {"team_size": 3, "requirements_stable": False}
        
        analysis = agent.analyze(actual, modifications)
        
        print(f"Actual scenario: {actual.description}")
        print(f"Outcome: {actual.outcome}")
        print(f"\nRisk scenario: {analysis.counterfactual_scenario.description}")
        print(f"Predicted outcome: {analysis.predicted_outcome[:200]}...")
        print(f"\nReasoning: {analysis.reasoning[:300]}...")
        
        # Example 4: Decision evaluation
        print("\n" + "="*80)
        print("EXAMPLE 4: Evaluating Past Decisions")
        print("="*80)
        actual = Scenario(
            description="Chose to work from home during pandemic",
            conditions={"location": "home", "commute_time": 0, "office_interaction": "minimal"},
            outcome="Maintained productivity, saved commute time, missed social interaction",
            is_actual=True
        )
        
        modifications = {"location": "office", "commute_time": 90, "office_interaction": "high"}
        
        analysis = agent.analyze(actual, modifications)
        
        print(f"Decision made: {actual.description}")
        print(f"Result: {actual.outcome}")
        print(f"\nAlternative: {analysis.counterfactual_scenario.description}")
        print(f"Predicted result: {analysis.predicted_outcome[:200]}...")
        print(f"\nLikelihood of alternative: {analysis.likelihood:.1%}")
        
        # Example 5: Multiple counterfactuals
        print("\n" + "="*80)
        print("EXAMPLE 5: Comparing Multiple Counterfactuals")
        print("="*80)
        actual = Scenario(
            description="Invested in stocks during market downturn",
            conditions={"investment_type": "stocks", "timing": "downturn", "amount": 10000},
            outcome="Portfolio recovered after 18 months, 15% gain",
            is_actual=True
        )
        
        alternatives = [
            {"investment_type": "bonds"},
            {"timing": "peak"},
            {"amount": 5000}
        ]
        
        print(f"Actual: {actual.description}")
        print(f"Outcome: {actual.outcome}\n")
        
        for i, mods in enumerate(alternatives, 1):
            analysis = agent.analyze(actual, mods)
            print(f"Alternative {i}: {list(mods.values())[0]}")
            print(f"  Predicted: {analysis.predicted_outcome[:150]}...")
            print()
        
        # Summary
        print("\n" + "="*80)
        print("SUMMARY: Counterfactual Reasoning Best Practices")
        print("="*80)
        print("""
    1. SCENARIO DEFINITION:
       - Clearly define actual scenario
       - Identify key conditions
       - Document actual outcomes
       - Establish baseline
    
    2. COUNTERFACTUAL CONSTRUCTION:
       - Change one or few variables
       - Keep other conditions constant
       - Ensure plausible alternatives
       - Consider realistic modifications
    
    3. OUTCOME PREDICTION:
       - Use causal models
       - Consider downstream effects
       - Account for interactions
       - Assess uncertainty
    
    4. COMPARATIVE ANALYSIS:
       - Identify key differences
       - Analyze causal chains
       - Compare outcomes
       - Evaluate trade-offs
    
    5. LIKELIHOOD ASSESSMENT:
       - Consider feasibility
       - Assess probability
       - Account for constraints
       - Be realistic
    
    6. APPLICATIONS:
       - Decision evaluation
       - Risk assessment
       - Policy analysis
       - Learning from history
       - Strategy planning
    
    Benefits:
    ✓ Understand causality
    ✓ Evaluate decisions
    ✓ Assess risks
    ✓ Learn from alternatives
    ✓ Improve future decisions
    
    Limitations:
    - Speculative nature
    - Multiple valid interpretations
    - Requires good causal model
    - Uncertainty in predictions
    - May be influenced by biases
        """)
    
    
    if __name__ == "__main__":
        demonstrate_counterfactual_reasoning()
    
        

    Agentic Design Pattern: Counterfactual Reasoning

    Pattern 162: Counterfactual Reasoning – Line-by-Line Explanation

    Overview

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


    Pattern Description

    Pattern 162: Counterfactual Reasoning

    Description: Counterfactual reasoning explores “what if” scenarios by considering alternative outcomes if certain conditions were different. It helps understand causality, evaluate decisions, and plan for contingencies.

    Components: – Scenario modification – Alternative outcome prediction – Causal analysis – Comparison with actual outcomes – Impact assessment

    Use Cases: – Decision analysis – Risk assessment – Causal inference – Policy evaluation – Learning from mistakes

    Benefits: – Understand causality – Evaluate alternatives – Learn from hypotheticals – Improve decision-making

    Trade-offs: – Speculative nature – Multiple possibilities – Requires causal model – May be computationally intensive

    LangChain Implementation: Uses ChatOpenAI for scenario generation, outcome prediction, and comparative analysis


    File Statistics

    • Total Lines: 403
    • Classes: 3
    • Functions: 1
    • Constants: 0

    Module Documentation

    """
    Pattern 162: Counterfactual Reasoning
    
    Description:
        Counterfactual reasoning explores "what if" scenarios by considering alternative
        outcomes if certain conditions were different. It helps understand causality,
        evaluate decisions, and plan for contingencies.
    
    Components:
        - Scenario modification
        - Alternative outcome prediction
        - Causal analysis
        - Comparison with actual outcomes
        - Impact assessment
    
    Use Cases:
        - Decision analysis
        - Risk assessment
        - Causal inference
        - Policy evaluation
        - Learning from mistakes
    
    Benefits:
        - Understand causality
        - Evaluate alternatives
        - Learn from hypotheticals
        - Improve decision-making
    
    Trade-offs:
        - Speculative nature
        - Multiple possibilities
        - Requires causal model
        - May be computationally intensive
    
    LangChain Implementation:
        Uses ChatOpenAI for scenario generation, outcome prediction,
        and comparative analysis
    """
    

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


    Import Statements

    Standard Library Imports

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

    Lines 53-58

    Purpose:

    Actual or counterfactual scenario
    

    Class: CounterfactualAnalysis

    Lines 62-70

    Purpose:

    Analysis of counterfactual scenario
    

    Class: CounterfactualReasoningAgent

    Lines 73-217

    Purpose:

    Agent that performs counterfactual reasoning
    

    Methods:

    __init__(self, model_name)

    Lines 76-77

    Purpose: Initialize the class instance with required parameters.

    Parameters: – model_name: Language model instance

    analyze(self, actual_scenario, modifications)

    Lines 79-116

    Purpose:

    Analyze counterfactual scenario
    

    Parameters: – actual_scenario: Actual scenario – modifications: Modifications

    Returns: CounterfactualAnalysis

    _generate_cf_description(self, actual, mods)

    Lines 118-133

    Purpose:

    Generate counterfactual description
    

    Parameters: – actual: Actual – mods: Mods

    Returns: str

    _predict_outcome(self, cf_scenario, actual)

    Lines 135-154

    Purpose:

    Predict counterfactual outcome
    

    Parameters: – cf_scenario: Cf scenario – actual: Actual

    Returns: str

    _identify_differences(self, actual, cf)

    Lines 156-164

    Purpose:

    Identify key differences
    

    Parameters: – actual: Actual – cf: Cf

    Returns: <ast.Subscript object at 0x000001A06D615F10>

    _analyze_causality(self, actual, cf, differences)

    Lines 166-186

    Purpose:

    Analyze causal factors
    

    Parameters: – actual: Actual – cf: Cf – differences: Differences

    Returns: <ast.Subscript object at 0x000001A06D72A110>

    _assess_likelihood(self, cf, actual)

    Lines 188-199

    Purpose:

    Assess likelihood of counterfactual
    

    Parameters: – cf: Cf – actual: Actual

    Returns: float

    _generate_reasoning(self, actual, cf, outcome)

    Lines 201-217

    Purpose:

    Generate reasoning explanation
    

    Parameters: – actual: Actual – cf: Cf – outcome: Outcome

    Returns: str


    Functions

    demonstrate_counterfactual_reasoning()

    Lines 220-398

    Purpose:

    Demonstrate counterfactual reasoning pattern
    

    Key Concepts Summary

    Pattern Overview

    The Counterfactual 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 161
    • Pattern 163

    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 Counterfactual 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: 162_counterfactual_reasoning.py
    Total Lines: 403
    Pattern Category: AI Agent Design Pattern #162


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Deductive Reasoning

    
        """
    Pattern 161: Deductive Reasoning
    
    Description:
        Deductive reasoning derives specific conclusions from general premises using
        logical rules. If the premises are true and the logic is valid, the conclusion
        must be true. This is the foundation of formal logic and mathematical proof.
    
    Components:
        - Premise collection and validation
        - Logical rule application
        - Inference chain construction
        - Conclusion derivation
        - Validity checking
    
    Use Cases:
        - Logical problem solving
        - Mathematical proofs
        - Rule-based systems
        - Legal reasoning
        - Formal verification
    
    Benefits:
        - Guaranteed valid conclusions
        - Transparent reasoning
        - Systematic approach
        - Verifiable logic
    
    Trade-offs:
        - Requires valid premises
        - Limited to formal logic
        - May be rigid
        - Doesn't discover new rules
    
    LangChain Implementation:
        Uses ChatOpenAI for logical inference, premise validation,
        and step-by-step deduction chains
    """
    
    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
    
    load_dotenv()
    
    
    class LogicalOperator(Enum):
        """Logical operators"""
        AND = "and"
        OR = "or"
        NOT = "not"
        IMPLIES = "implies"
        IF_AND_ONLY_IF = "iff"
    
    
    class InferenceRule(Enum):
        """Common inference rules"""
        MODUS_PONENS = "modus_ponens"  # If P→Q and P, then Q
        MODUS_TOLLENS = "modus_tollens"  # If P→Q and ¬Q, then ¬P
        HYPOTHETICAL_SYLLOGISM = "hypothetical_syllogism"  # If P→Q and Q→R, then P→R
        DISJUNCTIVE_SYLLOGISM = "disjunctive_syllogism"  # If P∨Q and ¬P, then Q
        CONTRAPOSITIVE = "contrapositive"  # P→Q equivalent to ¬Q→¬P
    
    
    @dataclass
    class Premise:
        """Logical premise or statement"""
        statement: str
        is_axiom: bool = False  # Assumed true
        confidence: float = 1.0
        source: Optional[str] = None
    
    
    @dataclass
    class InferenceStep:
        """Single step in deductive reasoning"""
        conclusion: str
        rule_applied: str
        premises_used: List[str]
        confidence: float = 1.0
    
    
    @dataclass
    class DeductiveProof:
        """Complete deductive proof"""
        premises: List[Premise]
        steps: List[InferenceStep]
        final_conclusion: str
        is_valid: bool
        reasoning_trace: List[str] = field(default_factory=list)
    
    
    class DeductiveReasoningAgent:
        """Agent that performs deductive reasoning"""
        
        def __init__(self, model_name: str = "gpt-3.5-turbo"):
            """
            Initialize deductive reasoning agent
            
            Args:
                model_name: LLM model to use
            """
            self.llm = ChatOpenAI(model=model_name, temperature=0)
            self.validator_llm = ChatOpenAI(model=model_name, temperature=0)
        
        def prove(self, premises: List[Premise], 
                  goal: str,
                  max_steps: int = 10) -> DeductiveProof:
            """
            Construct deductive proof from premises to goal
            
            Args:
                premises: List of premises
                goal: Goal conclusion to prove
                max_steps: Maximum reasoning steps
                
            Returns:
                DeductiveProof with reasoning chain
            """
            reasoning_trace = []
            steps = []
            
            reasoning_trace.append("Starting deductive proof...")
            reasoning_trace.append(f"Goal: {goal}")
            
            # Format premises
            premises_text = "\n".join([
                f"- {p.statement}" for p in premises
            ])
            reasoning_trace.append(f"Given premises:\n{premises_text}")
            
            # Generate proof steps
            for i in range(max_steps):
                # Check if goal is reached
                if steps and steps[-1].conclusion.lower() in goal.lower():
                    reasoning_trace.append(f"Goal reached in {i} steps!")
                    break
                
                # Generate next inference step
                step = self._generate_next_step(premises, steps, goal)
                if step:
                    steps.append(step)
                    reasoning_trace.append(
                        f"Step {i+1}: {step.conclusion} (via {step.rule_applied})"
                    )
                else:
                    reasoning_trace.append("No more valid inferences possible")
                    break
            
            # Validate proof
            is_valid = self._validate_proof(premises, steps, goal)
            
            final_conclusion = steps[-1].conclusion if steps else "No conclusion reached"
            
            return DeductiveProof(
                premises=premises,
                steps=steps,
                final_conclusion=final_conclusion,
                is_valid=is_valid,
                reasoning_trace=reasoning_trace
            )
        
        def _generate_next_step(self, premises: List[Premise],
                               existing_steps: List[InferenceStep],
                               goal: str) -> Optional[InferenceStep]:
            """Generate next logical inference step"""
            
            # Format context
            premises_text = "\n".join([f"- {p.statement}" for p in premises])
            
            if existing_steps:
                derived_text = "\n".join([
                    f"- {s.conclusion}" for s in existing_steps
                ])
            else:
                derived_text = "None yet"
            
            # Create inference prompt
            prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a logical reasoning expert. Given premises and a goal,
    generate the next logical inference step using valid deductive reasoning rules.
    
    Common rules:
    - Modus Ponens: If P→Q and P, then Q
    - Modus Tollens: If P→Q and ¬Q, then ¬P
    - Hypothetical Syllogism: If P→Q and Q→R, then P→R
    - Disjunctive Syllogism: If P∨Q and ¬P, then Q
    
    Return only the new conclusion and the rule used."""),
                ("user", """Premises:
    {premises}
    
    Already derived:
    {derived}
    
    Goal: {goal}
    
    What is the next logical inference? Format:
    Conclusion: [your conclusion]
    Rule: [rule name]
    Uses: [which premises/conclusions]""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            
            response = chain.invoke({
                "premises": premises_text,
                "derived": derived_text,
                "goal": goal
            })
            
            # Parse response
            return self._parse_inference_step(response)
        
        def _parse_inference_step(self, response: str) -> Optional[InferenceStep]:
            """Parse inference step from LLM response"""
            lines = response.strip().split('\n')
            conclusion = None
            rule = None
            uses = []
            
            for line in lines:
                line = line.strip()
                if line.lower().startswith('conclusion:'):
                    conclusion = line.split(':', 1)[1].strip()
                elif line.lower().startswith('rule:'):
                    rule = line.split(':', 1)[1].strip()
                elif line.lower().startswith('uses:'):
                    uses_text = line.split(':', 1)[1].strip()
                    uses = [u.strip() for u in uses_text.split(',')]
            
            if conclusion and rule:
                return InferenceStep(
                    conclusion=conclusion,
                    rule_applied=rule,
                    premises_used=uses,
                    confidence=1.0
                )
            
            return None
        
        def _validate_proof(self, premises: List[Premise],
                           steps: List[InferenceStep],
                           goal: str) -> bool:
            """Validate that the proof is logically sound"""
            
            if not steps:
                return False
            
            # Check if final conclusion matches goal
            final_matches = goal.lower() in steps[-1].conclusion.lower()
            
            # Use LLM to validate logical soundness
            premises_text = "\n".join([f"- {p.statement}" for p in premises])
            steps_text = "\n".join([
                f"{i+1}. {s.conclusion} (via {s.rule_applied})"
                for i, s in enumerate(steps)
            ])
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are a logic validator. Check if the reasoning steps are logically valid."),
                ("user", """Premises:
    {premises}
    
    Reasoning steps:
    {steps}
    
    Goal: {goal}
    
    Is this proof logically valid? Respond with 'yes' or 'no' and brief explanation.""")
            ])
            
            chain = prompt | self.validator_llm | StrOutputParser()
            
            response = chain.invoke({
                "premises": premises_text,
                "steps": steps_text,
                "goal": goal
            })
            
            is_valid = 'yes' in response.lower() and final_matches
            return is_valid
        
        def apply_rule(self, rule: InferenceRule, 
                       statements: List[str]) -> Optional[str]:
            """Apply specific inference rule to statements"""
            
            rule_descriptions = {
                InferenceRule.MODUS_PONENS: "If P implies Q, and P is true, then Q is true",
                InferenceRule.MODUS_TOLLENS: "If P implies Q, and Q is false, then P is false",
                InferenceRule.HYPOTHETICAL_SYLLOGISM: "If P implies Q, and Q implies R, then P implies R",
                InferenceRule.DISJUNCTIVE_SYLLOGISM: "If P or Q, and not P, then Q",
            }
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", f"Apply the logical rule: {rule_descriptions.get(rule, rule.value)}"),
                ("user", """Given statements:
    {statements}
    
    What conclusion can be derived? Provide just the conclusion.""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            
            statements_text = "\n".join([f"- {s}" for s in statements])
            conclusion = chain.invoke({"statements": statements_text})
            
            return conclusion.strip()
    
    
    def demonstrate_deductive_reasoning():
        """Demonstrate deductive reasoning pattern"""
        print("=" * 80)
        print("DEDUCTIVE REASONING PATTERN DEMONSTRATION")
        print("=" * 80)
        
        agent = DeductiveReasoningAgent()
        
        # Example 1: Classic syllogism
        print("\n" + "="*80)
        print("EXAMPLE 1: Classic Syllogism (Socrates)")
        print("="*80)
        premises = [
            Premise("All humans are mortal", is_axiom=True),
            Premise("Socrates is a human", is_axiom=True)
        ]
        goal = "Socrates is mortal"
        
        proof = agent.prove(premises, goal, max_steps=5)
        
        print("Premises:")
        for p in proof.premises:
            print(f"  - {p.statement}")
        
        print(f"\nGoal: {goal}")
        print(f"\nProof steps:")
        for i, step in enumerate(proof.steps, 1):
            print(f"  {i}. {step.conclusion}")
            print(f"     Rule: {step.rule_applied}")
        
        print(f"\nConclusion: {proof.final_conclusion}")
        print(f"Valid: {proof.is_valid}")
        
        # Example 2: Modus Ponens
        print("\n" + "="*80)
        print("EXAMPLE 2: Modus Ponens")
        print("="*80)
        statements = [
            "If it rains, the ground gets wet",
            "It is raining"
        ]
        conclusion = agent.apply_rule(InferenceRule.MODUS_PONENS, statements)
        print("Given:")
        for s in statements:
            print(f"  - {s}")
        print(f"\nConclusion: {conclusion}")
        
        # Example 3: Mathematical reasoning
        print("\n" + "="*80)
        print("EXAMPLE 3: Mathematical Reasoning")
        print("="*80)
        premises = [
            Premise("If x > 5, then x > 3", is_axiom=True),
            Premise("x = 7", is_axiom=True),
            Premise("7 > 5", is_axiom=True)
        ]
        goal = "x > 3"
        
        proof = agent.prove(premises, goal, max_steps=5)
        
        print("Premises:")
        for p in proof.premises:
            print(f"  - {p.statement}")
        
        print(f"\nReasoning:")
        for trace in proof.reasoning_trace:
            if "Step" in trace or "Goal" in trace:
                print(f"  {trace}")
        
        print(f"\nFinal: {proof.final_conclusion}")
        
        # Example 4: Logical chain
        print("\n" + "="*80)
        print("EXAMPLE 4: Hypothetical Syllogism")
        print("="*80)
        statements = [
            "If it rains, the roads are wet",
            "If the roads are wet, driving is dangerous"
        ]
        conclusion = agent.apply_rule(InferenceRule.HYPOTHETICAL_SYLLOGISM, statements)
        print("Given:")
        for s in statements:
            print(f"  - {s}")
        print(f"\nConclusion: {conclusion}")
        
        # Example 5: Complex proof
        print("\n" + "="*80)
        print("EXAMPLE 5: Multi-Step Proof")
        print("="*80)
        premises = [
            Premise("All birds have feathers", is_axiom=True),
            Premise("All animals with feathers can regulate body temperature", is_axiom=True),
            Premise("A penguin is a bird", is_axiom=True)
        ]
        goal = "A penguin can regulate body temperature"
        
        proof = agent.prove(premises, goal, max_steps=10)
        
        print("Premises:")
        for p in proof.premises:
            print(f"  - {p.statement}")
        
        print(f"\nGoal: {goal}")
        print(f"\nProof trace:")
        for trace in proof.reasoning_trace[-5:]:
            print(f"  {trace}")
        
        # Example 6: Modus Tollens
        print("\n" + "="*80)
        print("EXAMPLE 6: Modus Tollens (Proof by Contrapositive)")
        print("="*80)
        statements = [
            "If the alarm is working, it would have gone off",
            "The alarm did not go off"
        ]
        conclusion = agent.apply_rule(InferenceRule.MODUS_TOLLENS, statements)
        print("Given:")
        for s in statements:
            print(f"  - {s}")
        print(f"\nConclusion: {conclusion}")
        
        # Summary
        print("\n" + "="*80)
        print("SUMMARY: Deductive Reasoning Best Practices")
        print("="*80)
        print("""
    1. PREMISE VALIDATION:
       - Verify premises are true
       - Identify axioms vs derived facts
       - Check premise consistency
       - Document premise sources
    
    2. LOGICAL RULES:
       - Use established inference rules
       - Apply rules systematically
       - Maintain logical validity
       - Document rule application
    
    3. PROOF CONSTRUCTION:
       - Build step-by-step chains
       - Connect premises to conclusion
       - Verify each step
       - Keep proofs as simple as possible
    
    4. COMMON RULES:
       - Modus Ponens: P→Q, P ⊢ Q
       - Modus Tollens: P→Q, ¬Q ⊢ ¬P
       - Hypothetical Syllogism: P→Q, Q→R ⊢ P→R
       - Disjunctive Syllogism: P∨Q, ¬P ⊢ Q
       - Conjunction: P, Q ⊢ P∧Q
    
    5. VALIDATION:
       - Check logical soundness
       - Verify conclusion follows
       - Test for fallacies
       - Review proof structure
    
    6. APPLICATIONS:
       - Mathematical proofs
       - Legal reasoning
       - Rule-based systems
       - Formal verification
       - Logical problem solving
    
    Benefits:
    ✓ Guarantees valid conclusions
    ✓ Transparent reasoning process
    ✓ Systematic and rigorous
    ✓ Verifiable logic
    ✓ Foundation for formal systems
    
    Limitations:
    - Requires true premises
    - Limited to formal logic
    - May be inflexible
    - Doesn't generate new knowledge
    - Can be verbose for complex proofs
        """)
    
    
    if __name__ == "__main__":
        demonstrate_deductive_reasoning()
    
        

    Agentic Design Pattern: Deductive Reasoning

    Pattern 161: Deductive Reasoning – Line-by-Line Explanation

    Overview

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


    Pattern Description

    Pattern 161: Deductive Reasoning

    Description: Deductive reasoning derives specific conclusions from general premises using logical rules. If the premises are true and the logic is valid, the conclusion must be true. This is the foundation of formal logic and mathematical proof.

    Components: – Premise collection and validation – Logical rule application – Inference chain construction – Conclusion derivation – Validity checking

    Use Cases: – Logical problem solving – Mathematical proofs – Rule-based systems – Legal reasoning – Formal verification

    Benefits: – Guaranteed valid conclusions – Transparent reasoning – Systematic approach – Verifiable logic

    Trade-offs: – Requires valid premises – Limited to formal logic – May be rigid – Doesn’t discover new rules

    LangChain Implementation: Uses ChatOpenAI for logical inference, premise validation, and step-by-step deduction chains


    File Statistics

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

    Module Documentation

    """
    Pattern 161: Deductive Reasoning
    
    Description:
        Deductive reasoning derives specific conclusions from general premises using
        logical rules. If the premises are true and the logic is valid, the conclusion
        must be true. This is the foundation of formal logic and mathematical proof.
    
    Components:
        - Premise collection and validation
        - Logical rule application
        - Inference chain construction
        - Conclusion derivation
        - Validity checking
    
    Use Cases:
        - Logical problem solving
        - Mathematical proofs
        - Rule-based systems
        - Legal reasoning
        - Formal verification
    
    Benefits:
        - Guaranteed valid conclusions
        - Transparent reasoning
        - Systematic approach
        - Verifiable logic
    
    Trade-offs:
        - Requires valid premises
        - Limited to formal logic
        - May be rigid
        - Doesn't discover new rules
    
    LangChain Implementation:
        Uses ChatOpenAI for logical inference, premise validation,
        and step-by-step deduction chains
    """
    

    Purpose: This module implements the Deductive 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 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 – 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: LogicalOperator

    Lines 53-59

    Inherits from: Enum

    Purpose:

    Logical operators
    

    Class: InferenceRule

    Lines 62-68

    Inherits from: Enum

    Purpose:

    Common inference rules
    

    Class: Premise

    Lines 72-77

    Purpose:

    Logical premise or statement
    

    Class: InferenceStep

    Lines 81-86

    Purpose:

    Single step in deductive reasoning
    

    Class: DeductiveProof

    Lines 90-96

    Purpose:

    Complete deductive proof
    

    Class: DeductiveReasoningAgent

    Lines 99-314

    Purpose:

    Agent that performs deductive reasoning
    

    Methods:

    __init__(self, model_name)

    Lines 102-110

    Purpose:

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

    Parameters: – model_name: Language model instance

    prove(self, premises, goal, max_steps)

    Lines 112-167

    Purpose:

    Construct deductive proof from premises to goal
    
    Args:
        premises: List of premises
        goal: Goal conclusion to prove
        max_steps: Maximum reasoning steps
    
    Returns:
        DeductiveProof with reasoning chain
    

    Parameters: – premises: Premises – goal: Goal – max_steps: Max steps

    Returns: DeductiveProof

    _generate_next_step(self, premises, existing_steps, goal)

    Lines 169-219

    Purpose:

    Generate next logical inference step
    

    Parameters: – premises: Premises – existing_steps: Existing steps – goal: Goal

    Returns: <ast.Subscript object at 0x000001A06D633F10>

    _parse_inference_step(self, response)

    Lines 221-246

    Purpose:

    Parse inference step from LLM response
    

    Parameters: – response: Response

    Returns: <ast.Subscript object at 0x000001A06E0A9710>

    _validate_proof(self, premises, steps, goal)

    Lines 248-288

    Purpose:

    Validate that the proof is logically sound
    

    Parameters: – premises: Premises – steps: Steps – goal: Goal

    Returns: bool

    apply_rule(self, rule, statements)

    Lines 290-314

    Purpose:

    Apply specific inference rule to statements
    

    Parameters: – rule: Rule – statements: Statements

    Returns: <ast.Subscript object at 0x000001A06D888690>


    Functions

    demonstrate_deductive_reasoning()

    Lines 317-494

    Purpose:

    Demonstrate deductive reasoning pattern
    

    Key Concepts Summary

    Pattern Overview

    The Deductive 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 160
    • Pattern 162

    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 Deductive 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: 161_deductive_reasoning.py
    Total Lines: 499
    Pattern Category: AI Agent Design Pattern #161


    Generated from actual implementation code
    For educational and reference purposes

  • 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