Tag: artificial-intelligence

  • Agentic Design Pattern: Transfer Learning

    
        """
    Pattern 115: Transfer Learning Agent
    
    Description:
        Transfer Learning enables agents to leverage knowledge learned from one
        task or domain to improve performance on related tasks. This pattern
        identifies transferable knowledge, adapts it to new contexts, and combines
        it with task-specific learning for faster adaptation and better performance.
        
        Essential for agents that need to work across multiple related domains or
        tasks, transfer learning reduces the data and time needed to achieve good
        performance in new areas.
    
    Key Components:
        1. Source Knowledge Extractor: Identifies transferable knowledge
        2. Domain Adapter: Adapts knowledge to new domain
        3. Fine-tuner: Specializes for target task
        4. Transfer Strategy Selector: Chooses transfer approach
        5. Performance Evaluator: Assesses transfer effectiveness
    
    Transfer Strategies:
        - Feature Transfer: Use learned representations
        - Parameter Transfer: Transfer model parameters
        - Instance Transfer: Leverage source examples
        - Relational Transfer: Transfer relationships
        - Meta-transfer: Transfer learning strategies
    
    Use Cases:
        - Multi-domain chatbots
        - Cross-lingual applications
        - Domain adaptation (e.g., medical to legal)
        - Few-shot learning in new domains
        - Skill transfer in robotics
    
    LangChain Implementation:
        Uses prompt engineering, few-shot examples, and domain adaptation
        techniques to transfer knowledge between tasks and domains.
    """
    
    import os
    from typing import List, Dict, Any, Optional
    from dotenv import load_dotenv
    from datetime import datetime
    
    from langchain_openai import ChatOpenAI
    from langchain_core.prompts import ChatPromptTemplate, FewShotChatMessagePromptTemplate
    from langchain_core.output_parsers import StrOutputParser
    
    load_dotenv()
    
    
    class DomainKnowledge:
        """Represents knowledge from a source domain"""
        
        def __init__(self, domain_name: str):
            self.domain_name = domain_name
            self.examples: List[Dict[str, str]] = []
            self.patterns: List[str] = []
            self.concepts: Dict[str, str] = {}
        
        def add_example(self, input_text: str, output_text: str, pattern: str = ""):
            """Add an example from this domain"""
            self.examples.append({
                "input": input_text,
                "output": output_text,
                "pattern": pattern
            })
        
        def add_pattern(self, pattern: str):
            """Add a learned pattern"""
            if pattern not in self.patterns:
                self.patterns.append(pattern)
        
        def add_concept(self, concept: str, description: str):
            """Add a domain concept"""
            self.concepts[concept] = description
    
    
    class TransferLearningAgent:
        """Agent that transfers knowledge between domains"""
        
        def __init__(self, model_name: str = "gpt-4"):
            self.llm = ChatOpenAI(model=model_name, temperature=0.7)
            self.source_domains: Dict[str, DomainKnowledge] = {}
            self.current_domain: Optional[str] = None
        
        def add_source_domain(self, domain: DomainKnowledge):
            """Add a source domain with learned knowledge"""
            self.source_domains[domain.domain_name] = domain
            print(f"Added source domain: {domain.domain_name}")
            print(f"  - {len(domain.examples)} examples")
            print(f"  - {len(domain.patterns)} patterns")
            print(f"  - {len(domain.concepts)} concepts")
        
        def identify_relevant_source(self, target_task: str) -> Optional[DomainKnowledge]:
            """Identify most relevant source domain for transfer"""
            if not self.source_domains:
                return None
            
            # Use LLM to identify most relevant source
            domains_desc = "\n".join([
                f"- {name}: {len(d.examples)} examples, patterns: {', '.join(d.patterns[:3])}"
                for name, d in self.source_domains.items()
            ])
            
            selection_prompt = f"""Given a target task and available source domains, identify which source domain would provide the most useful transferable knowledge.
    
    Target task: {target_task}
    
    Available source domains:
    {domains_desc}
    
    Which source domain is most relevant? Answer with just the domain name."""
            
            response = self.llm.invoke(selection_prompt).content.strip()
            
            # Find matching domain
            for domain_name in self.source_domains:
                if domain_name.lower() in response.lower():
                    print(f"\nSelected source domain: {domain_name}")
                    return self.source_domains[domain_name]
            
            # Default to first domain
            return list(self.source_domains.values())[0]
        
        def adapt_examples(self, source_domain: DomainKnowledge, 
                          target_domain: str, n: int = 3) -> List[Dict[str, str]]:
            """Adapt source examples to target domain"""
            if not source_domain.examples:
                return []
            
            # Take first n examples
            source_examples = source_domain.examples[:n]
            
            # Create adaptation prompt
            examples_str = "\n\n".join([
                f"Input: {ex['input']}\nOutput: {ex['output']}"
                for ex in source_examples
            ])
            
            adaptation_prompt = f"""Adapt these examples from {source_domain.domain_name} domain to {target_domain} domain.
    Keep the same pattern and structure, but change the content to fit the target domain.
    
    Source examples ({source_domain.domain_name}):
    {examples_str}
    
    Generate {n} adapted examples for {target_domain} domain in the same format."""
            
            response = self.llm.invoke(adaptation_prompt).content
            
            # Parse adapted examples (simplified parsing)
            adapted = []
            lines = response.split('\n')
            current_input = None
            
            for line in lines:
                if line.startswith('Input:'):
                    current_input = line.replace('Input:', '').strip()
                elif line.startswith('Output:') and current_input:
                    output = line.replace('Output:', '').strip()
                    adapted.append({"input": current_input, "output": output})
                    current_input = None
            
            return adapted[:n] if adapted else source_examples[:n]
        
        def transfer_and_respond(self, target_task: str, target_domain: str,
                                query: str) -> Dict[str, Any]:
            """Transfer knowledge and respond to query"""
            print(f"\n{'='*60}")
            print(f"Target Domain: {target_domain}")
            print(f"Target Task: {target_task}")
            print(f"Query: {query}")
            print(f"{'='*60}")
            
            # Identify relevant source domain
            source_domain = self.identify_relevant_source(target_task)
            
            if source_domain:
                print(f"\nTransferring knowledge from: {source_domain.domain_name}")
                
                # Adapt examples to target domain
                adapted_examples = self.adapt_examples(source_domain, target_domain, n=3)
                print(f"Adapted {len(adapted_examples)} examples to target domain")
                
                # Extract transferable patterns
                patterns_str = "\n".join([f"- {p}" for p in source_domain.patterns])
                
                # Create transfer-enhanced prompt
                if adapted_examples:
                    example_prompt = ChatPromptTemplate.from_messages([
                        ("human", "{input}"),
                        ("ai", "{output}")
                    ])
                    
                    few_shot_prompt = FewShotChatMessagePromptTemplate(
                        example_prompt=example_prompt,
                        examples=adapted_examples
                    )
                    
                    prompt = ChatPromptTemplate.from_messages([
                        ("system", f"""You are an AI assistant for {target_domain}.
    
    You are leveraging transferred knowledge from {source_domain.domain_name}.
    
    Transferable patterns:
    {patterns_str}
    
    Apply these learned patterns to the current domain:"""),
                        few_shot_prompt,
                        ("human", "{input}")
                    ])
                else:
                    prompt = ChatPromptTemplate.from_messages([
                        ("system", f"""You are an AI assistant for {target_domain}.
    
    You are leveraging transferred knowledge from {source_domain.domain_name}.
    
    Transferable patterns:
    {patterns_str}
    
    Apply these learned patterns to the current task."""),
                        ("human", "{input}")
                    ])
                
                chain = prompt | self.llm | StrOutputParser()
                response = chain.invoke({"input": query})
            else:
                # No source domain available, use base prompt
                print("\nNo source domain available, using base knowledge")
                
                prompt = ChatPromptTemplate.from_messages([
                    ("system", f"You are an AI assistant for {target_domain}."),
                    ("human", "{input}")
                ])
                
                chain = prompt | self.llm | StrOutputParser()
                response = chain.invoke({"input": query})
            
            print(f"\nResponse:")
            print(response)
            
            return {
                "target_domain": target_domain,
                "target_task": target_task,
                "query": query,
                "source_domain": source_domain.domain_name if source_domain else None,
                "response": response,
                "timestamp": datetime.now().isoformat()
            }
        
        def compare_with_without_transfer(self, target_task: str, target_domain: str,
                                         query: str) -> Dict[str, Any]:
            """Compare performance with and without transfer learning"""
            print(f"\n{'='*70}")
            print(f"COMPARISON: With vs Without Transfer Learning")
            print(f"{'='*70}")
            
            # Response with transfer
            print("\n--- WITH TRANSFER LEARNING ---")
            with_transfer = self.transfer_and_respond(target_task, target_domain, query)
            
            # Response without transfer (base model)
            print("\n--- WITHOUT TRANSFER LEARNING (Baseline) ---")
            base_prompt = ChatPromptTemplate.from_messages([
                ("system", f"You are an AI assistant for {target_domain}."),
                ("human", "{input}")
            ])
            base_chain = base_prompt | self.llm | StrOutputParser()
            without_transfer = base_chain.invoke({"input": query})
            
            print(f"\nBaseline Response:")
            print(without_transfer)
            
            return {
                "with_transfer": with_transfer,
                "without_transfer": without_transfer
            }
    
    
    def demonstrate_transfer_learning():
        """Demonstrate transfer learning pattern"""
        print("\n" + "="*70)
        print("TRANSFER LEARNING AGENT DEMONSTRATION")
        print("="*70)
        
        agent = TransferLearningAgent()
        
        # Create source domain: Customer Support
        print("\n" + "="*70)
        print("Setting Up Source Domain Knowledge")
        print("="*70)
        
        customer_support = DomainKnowledge("customer_support")
        
        # Add examples
        customer_support.add_example(
            "How do I reset my password?",
            "To reset your password: 1) Click 'Forgot Password' on login page, "
            "2) Enter your email, 3) Check your email for reset link, 4) Create new password.",
            "step_by_step_instructions"
        )
        customer_support.add_example(
            "Where is my order?",
            "To track your order: 1) Log into your account, 2) Go to 'Orders', "
            "3) Find your order number, 4) Click 'Track' to see current status.",
            "step_by_step_instructions"
        )
        customer_support.add_example(
            "How do I cancel my subscription?",
            "To cancel subscription: 1) Go to Account Settings, 2) Select 'Subscriptions', "
            "3) Choose subscription to cancel, 4) Confirm cancellation.",
            "step_by_step_instructions"
        )
        
        # Add patterns
        customer_support.add_pattern("Provide step-by-step instructions")
        customer_support.add_pattern("Be clear and concise")
        customer_support.add_pattern("Offer alternative solutions")
        customer_support.add_pattern("Include troubleshooting tips")
        
        # Add concepts
        customer_support.add_concept("account_management", "User account operations")
        customer_support.add_concept("order_processing", "Order lifecycle and tracking")
        
        agent.add_source_domain(customer_support)
        
        # Example 1: Transfer to Technical Support
        print("\n" + "="*70)
        print("Example 1: Transfer to Technical Support Domain")
        print("="*70)
        
        result1 = agent.transfer_and_respond(
            target_task="Help users troubleshoot technical issues",
            target_domain="technical_support",
            query="How do I fix a blue screen error?"
        )
        
        # Example 2: Transfer to Medical Assistant
        print("\n" + "="*70)
        print("Example 2: Transfer to Medical Assistant Domain")
        print("="*70)
        
        result2 = agent.transfer_and_respond(
            target_task="Provide medical information and guidance",
            target_domain="medical_assistant",
            query="How do I check my blood pressure at home?"
        )
        
        # Example 3: Compare with and without transfer
        print("\n" + "="*70)
        print("Example 3: Comparing With/Without Transfer")
        print("="*70)
        
        comparison = agent.compare_with_without_transfer(
            target_task="Help users with cooking recipes",
            target_domain="cooking_assistant",
            query="How do I make homemade pasta?"
        )
        
        # Summary
        print("\n" + "="*70)
        print("SUMMARY")
        print("="*70)
        print("""
    Transfer Learning Pattern demonstrated:
    
    Key Features:
    1. Source Domain Selection - Identify most relevant source knowledge
    2. Knowledge Adaptation - Adapt examples and patterns to new domain
    3. Pattern Transfer - Transfer learned patterns across domains
    4. Few-Shot Enhancement - Use adapted examples for better performance
    5. Comparative Evaluation - Compare with/without transfer
    
    Transfer Mechanisms:
    - Example adaptation across domains
    - Pattern and structure transfer
    - Concept mapping between domains
    - Meta-learning of general strategies
    - Domain-specific fine-tuning
    
    Applications:
    - Multi-domain chatbots (customer → technical support)
    - Cross-lingual NLP (English → Spanish)
    - Domain adaptation (medical → legal documentation)
    - Few-shot learning in new domains
    - Skill transfer in robotics
    
    Benefits:
    - Faster adaptation to new domains
    - Better performance with limited data
    - Leverages existing knowledge
    - Reduces training time and data needs
    - Improves generalization
    
    The transfer learning pattern enables agents to leverage knowledge from
    related domains, significantly improving performance on new tasks with
    limited domain-specific training.
        """)
    
    
    if __name__ == "__main__":
        demonstrate_transfer_learning()
    
        

    Agentic Design Pattern: Transfer Learning

    Pattern 115: Transfer Learning – Line-by-Line Explanation

    Overview

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


    Pattern Description

    Pattern 115: Transfer Learning Agent

    Description: Transfer Learning enables agents to leverage knowledge learned from one task or domain to improve performance on related tasks. This pattern identifies transferable knowledge, adapts it to new contexts, and combines it with task-specific learning for faster adaptation and better performance.

    Essential for agents that need to work across multiple related domains or
    tasks, transfer learning reduces the data and time needed to achieve good
    performance in new areas.
    

    Key Components: 1. Source Knowledge Extractor: Identifies transferable knowledge 2. Domain Adapter: Adapts knowledge to new domain 3. Fine-tuner: Specializes for target task 4. Transfer Strategy Selector: Chooses transfer approach 5. Performance Evaluator: Assesses transfer effectiveness

    Transfer Strategies: – Feature Transfer: Use learned representations – Parameter Transfer: Transfer model parameters – Instance Transfer: Leverage source examples – Relational Transfer: Transfer relationships – Meta-transfer: Transfer learning strategies

    Use Cases: – Multi-domain chatbots – Cross-lingual applications – Domain adaptation (e.g., medical to legal) – Few-shot learning in new domains – Skill transfer in robotics

    LangChain Implementation: Uses prompt engineering, few-shot examples, and domain adaptation techniques to transfer knowledge between tasks and domains.


    File Statistics

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

    Module Documentation

    """
    Pattern 115: Transfer Learning Agent
    
    Description:
        Transfer Learning enables agents to leverage knowledge learned from one
        task or domain to improve performance on related tasks. This pattern
        identifies transferable knowledge, adapts it to new contexts, and combines
        it with task-specific learning for faster adaptation and better performance.
    
        Essential for agents that need to work across multiple related domains or
        tasks, transfer learning reduces the data and time needed to achieve good
        performance in new areas.
    
    Key Components:
        1. Source Knowledge Extractor: Identifies transferable knowledge
        2. Domain Adapter: Adapts knowledge to new domain
        3. Fine-tuner: Specializes for target task
        4. Transfer Strategy Selector: Chooses transfer approach
        5. Performance Evaluator: Assesses transfer effectiveness
    
    Transfer Strategies:
        - Feature Transfer: Use learned representations
        - Parameter Transfer: Transfer model parameters
        - Instance Transfer: Leverage source examples
        - Relational Transfer: Transfer relationships
        - Meta-transfer: Transfer learning strategies
    
    Use Cases:
        - Multi-domain chatbots
        - Cross-lingual applications
        - Domain adaptation (e.g., medical to legal)
        - Few-shot learning in new domains
        - Skill transfer in robotics
    
    LangChain Implementation:
        Uses prompt engineering, few-shot examples, and domain adaptation
        techniques to transfer knowledge between tasks and domains.
    """
    

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


    Import Statements

    Standard Library Imports

    import datetime.datetime
    import langchain_core.output_parsers.StrOutputParser
    import langchain_core.prompts.ChatPromptTemplate
    import langchain_core.prompts.FewShotChatMessagePromptTemplate
    import os
    import typing.Any
    import typing.Dict
    import typing.List
    import typing.Optional
    

    Purpose: Core Python functionality: – datetime.datetime: Date and time handling – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – langchain_core.prompts.FewShotChatMessagePromptTemplate: 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: DomainKnowledge

    Lines 52-76

    Purpose:

    Represents knowledge from a source domain
    

    Methods:

    __init__(self, domain_name)

    Lines 55-59

    Purpose: Initialize the class instance with required parameters.

    Parameters: – domain_name: Domain name

    add_example(self, input_text, output_text, pattern)

    Lines 61-67

    Purpose:

    Add an example from this domain
    

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

    add_pattern(self, pattern)

    Lines 69-72

    Purpose:

    Add a learned pattern
    

    Parameters: – pattern: Pattern

    add_concept(self, concept, description)

    Lines 74-76

    Purpose:

    Add a domain concept
    

    Parameters: – concept: Concept – description: Description


    Class: TransferLearningAgent

    Lines 79-277

    Purpose:

    Agent that transfers knowledge between domains
    

    Methods:

    __init__(self, model_name)

    Lines 82-85

    Purpose: Initialize the class instance with required parameters.

    Parameters: – model_name: Language model instance

    add_source_domain(self, domain)

    Lines 87-93

    Purpose:

    Add a source domain with learned knowledge
    

    Parameters: – domain: Domain

    identify_relevant_source(self, target_task)

    Lines 95-124

    Purpose:

    Identify most relevant source domain for transfer
    

    Parameters: – target_task: Target task

    Returns: <ast.Subscript object at 0x000001A06D7FEF50>

    adapt_examples(self, source_domain, target_domain, n)

    Lines 126-164

    Purpose:

    Adapt source examples to target domain
    

    Parameters: – source_domain: Source domain – target_domain: Target domain – n: N

    Returns: <ast.Subscript object at 0x000001A06D72EDD0>

    transfer_and_respond(self, target_task, target_domain, query)

    Lines 166-249

    Purpose:

    Transfer knowledge and respond to query
    

    Parameters: – target_task: Target task – target_domain: Target domain – query: Input query or question

    Returns: <ast.Subscript object at 0x000001A06D79B310>

    compare_with_without_transfer(self, target_task, target_domain, query)

    Lines 251-277

    Purpose:

    Compare performance with and without transfer learning
    

    Parameters: – target_task: Target task – target_domain: Target domain – query: Input query or question

    Returns: <ast.Subscript object at 0x000001A06D7E43D0>


    Functions

    demonstrate_transfer_learning()

    Lines 280-398

    Purpose:

    Demonstrate transfer learning pattern
    

    Key Concepts Summary

    Pattern Overview

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

    Core Components

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

    When to Use

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

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


    Best Practices Demonstrated

    1. Environment Configuration

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

    2. Error Handling

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

    3. Prompt Engineering

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

    4. Code Organization

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

    5. Performance Optimization

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

    LangChain Components Used

    • ChatOpenAI: OpenAI model interface
    • Chains: Sequential component composition

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


    Execution Flow

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

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


    Performance Considerations

    Computational Cost

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

    Optimization Strategies

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

    Cost Estimates (Approximate)

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

    Actual costs vary based on input/output token counts


    Common Pitfalls to Avoid

    1. Prompt Design Issues

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

    2. Error Handling

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

    3. Token Management

    ❌ Exceeding model token limits
    ✅ Truncate inputs, chunk processing

    4. Security

    ❌ Exposing API keys in code
    ✅ Use environment variables

    5. Testing

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

    6. Cost Management

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


    Further Improvements

    Production Enhancements

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

    Feature Extensions

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

    Code Quality

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

    Related Patterns

    • Pattern 114
    • Pattern 116

    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 Transfer Learning pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

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

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

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


    File: 115_transfer_learning.py
    Total Lines: 403
    Pattern Category: AI Agent Design Pattern #115


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Online Learning

    
        """
    Pattern 114: Online Learning
    
    Description:
        Online Learning enables agents to continuously learn and adapt from incoming
        data streams in real-time. Unlike batch learning, online learning updates
        the agent's knowledge incrementally as new data arrives, allowing adaptation
        to changing environments and concept drift.
        
        This pattern is crucial for agents operating in dynamic environments where
        patterns, user preferences, or optimal strategies change over time. It
        combines incremental updates, drift detection, and continuous adaptation.
    
    Key Components:
        1. Stream Processor: Handles incoming data streams
        2. Incremental Updater: Updates knowledge incrementally
        3. Drift Detector: Identifies when patterns change
        4. Performance Monitor: Tracks learning effectiveness
        5. Model Adapter: Adjusts model based on feedback
        6. Forgetting Mechanism: Manages outdated knowledge
    
    Learning Strategies:
        - Incremental Learning: Update with each new example
        - Mini-Batch: Update with small batches
        - Windowed: Learn from recent window
        - Adaptive Rate: Adjust learning rate dynamically
        - Ensemble: Maintain multiple learners
    
    Use Cases:
        - Real-time recommendation systems
        - Adaptive chatbots learning from conversations
        - Fraud detection with evolving patterns
        - Personalization engines
        - Market prediction systems
    
    LangChain Implementation:
        Uses conversation memory, feedback integration, and dynamic prompt
        adjustment to enable continuous learning from interactions.
    """
    
    import os
    from typing import List, Dict, Any, Optional, Deque
    from collections import deque
    from dotenv import load_dotenv
    from datetime import datetime
    import json
    
    from langchain_openai import ChatOpenAI
    from langchain_core.prompts import ChatPromptTemplate, FewShotChatMessagePromptTemplate
    from langchain_core.output_parsers import StrOutputParser
    from langchain.memory import ConversationBufferMemory
    
    load_dotenv()
    
    
    class ExampleBuffer:
        """Manages a buffer of learning examples"""
        
        def __init__(self, max_size: int = 100):
            self.max_size = max_size
            self.examples: Deque[Dict[str, Any]] = deque(maxlen=max_size)
            self.performance_history: List[float] = []
        
        def add_example(self, input_text: str, output_text: str, 
                        feedback_score: Optional[float] = None):
            """Add a new example to the buffer"""
            example = {
                "input": input_text,
                "output": output_text,
                "feedback": feedback_score,
                "timestamp": datetime.now().isoformat()
            }
            self.examples.append(example)
            
            if feedback_score is not None:
                self.performance_history.append(feedback_score)
        
        def get_recent_examples(self, n: int = 5) -> List[Dict[str, str]]:
            """Get n most recent examples"""
            recent = list(self.examples)[-n:]
            return [{"input": ex["input"], "output": ex["output"]} for ex in recent]
        
        def get_best_examples(self, n: int = 5) -> List[Dict[str, str]]:
            """Get n best-rated examples"""
            if not self.examples:
                return []
            
            # Filter examples with feedback
            rated = [ex for ex in self.examples if ex["feedback"] is not None]
            if not rated:
                return self.get_recent_examples(n)
            
            # Sort by feedback score
            sorted_examples = sorted(rated, key=lambda x: x["feedback"], reverse=True)
            top_n = sorted_examples[:n]
            
            return [{"input": ex["input"], "output": ex["output"]} for ex in top_n]
        
        def detect_drift(self, window_size: int = 20) -> bool:
            """Detect if performance is degrading (concept drift)"""
            if len(self.performance_history)  drift_threshold
        
        def get_performance_trend(self) -> str:
            """Get description of performance trend"""
            if len(self.performance_history) = 20 else []
            
            recent_avg = sum(recent) / len(recent)
            
            if not older:
                return f"recent_avg: {recent_avg:.2f}"
            
            older_avg = sum(older) / len(older)
            
            if recent_avg > older_avg + 0.1:
                return f"improving (from {older_avg:.2f} to {recent_avg:.2f})"
            elif recent_avg  ChatPromptTemplate:
            """Create prompt with dynamic examples from learning buffer"""
            # Get examples from buffer
            if use_best:
                examples = self.example_buffer.get_best_examples(n=3)
            else:
                examples = self.example_buffer.get_recent_examples(n=3)
            
            if not examples:
                # No examples yet, use base prompt
                return ChatPromptTemplate.from_messages([
                    ("system", self.base_system_prompt),
                    ("human", "{input}")
                ])
            
            # Create few-shot prompt with learned examples
            few_shot_prompt = FewShotChatMessagePromptTemplate(
                example_prompt=self.example_prompt,
                examples=examples
            )
            
            # Get performance trend
            trend = self.example_buffer.get_performance_trend()
            system_msg = f"""{self.base_system_prompt}
    
    Learning Status: {len(self.example_buffer.examples)} examples learned
    Performance Trend: {trend}
    
    Learn from these example interactions:"""
            
            return ChatPromptTemplate.from_messages([
                ("system", system_msg),
                few_shot_prompt,
                ("human", "{input}")
            ])
        
        def respond(self, user_input: str) -> str:
            """Generate response using current learned knowledge"""
            # Create dynamic prompt with learned examples
            prompt = self.create_dynamic_prompt(use_best=True)
            
            # Generate response
            chain = prompt | self.llm | StrOutputParser()
            response = chain.invoke({"input": user_input})
            
            self.total_interactions += 1
            return response
        
        def learn_from_feedback(self, user_input: str, response: str, 
                               feedback_score: float):
            """Learn from user feedback on a response"""
            print(f"\nLearning from feedback (score: {feedback_score})...")
            
            # Add to example buffer
            self.example_buffer.add_example(user_input, response, feedback_score)
            
            # Check for drift
            if self.example_buffer.detect_drift():
                print("⚠️  Concept drift detected! Performance declining.")
                print("Consider adjusting learning strategy or model.")
            
            # Show learning progress
            trend = self.example_buffer.get_performance_trend()
            print(f"Performance trend: {trend}")
        
        def interact_and_learn(self, user_input: str, feedback_score: Optional[float] = None) -> Dict[str, Any]:
            """Complete interaction cycle with learning"""
            print(f"\n{'='*60}")
            print(f"User: {user_input}")
            print(f"{'='*60}")
            
            # Generate response
            response = self.respond(user_input)
            print(f"\nAssistant: {response}")
            
            # Learn from feedback if provided
            if feedback_score is not None:
                self.learn_from_feedback(user_input, response, feedback_score)
            
            return {
                "input": user_input,
                "response": response,
                "feedback": feedback_score,
                "total_interactions": self.total_interactions,
                "examples_learned": len(self.example_buffer.examples),
                "performance_trend": self.example_buffer.get_performance_trend()
            }
        
        def get_learning_stats(self) -> Dict[str, Any]:
            """Get statistics about learning progress"""
            return {
                "total_interactions": self.total_interactions,
                "examples_in_buffer": len(self.example_buffer.examples),
                "performance_history_size": len(self.example_buffer.performance_history),
                "performance_trend": self.example_buffer.get_performance_trend(),
                "drift_detected": self.example_buffer.detect_drift()
            }
    
    
    def demonstrate_online_learning():
        """Demonstrate online learning pattern"""
        print("\n" + "="*70)
        print("ONLINE LEARNING DEMONSTRATION")
        print("="*70)
        
        # Create agent for customer support domain
        agent = OnlineLearningAgent(domain="customer support")
        
        # Example 1: Initial interactions (building knowledge)
        print("\n" + "="*70)
        print("Example 1: Initial Learning Phase")
        print("="*70)
        
        interactions = [
            ("How do I reset my password?", 0.9),
            ("Where can I find my order history?", 0.85),
            ("How do I contact support?", 0.95),
            ("What is your return policy?", 0.8),
            ("How long does shipping take?", 0.9),
        ]
        
        for user_input, feedback in interactions:
            agent.interact_and_learn(user_input, feedback)
        
        # Show learning stats
        print("\n" + "-"*60)
        print("Learning Statistics After Initial Phase:")
        stats = agent.get_learning_stats()
        print(json.dumps(stats, indent=2))
        
        # Example 2: Applying learned knowledge
        print("\n" + "="*70)
        print("Example 2: Applying Learned Knowledge")
        print("="*70)
        
        new_queries = [
            "I forgot my password, what should I do?",  # Similar to learned
            "How can I see my past orders?",  # Similar to learned
        ]
        
        for query in new_queries:
            result = agent.interact_and_learn(query, feedback_score=0.9)
        
        # Example 3: Continuous learning with feedback
        print("\n" + "="*70)
        print("Example 3: Continuous Learning with Mixed Feedback")
        print("="*70)
        
        mixed_interactions = [
            ("Can I change my delivery address?", 0.7),  # Medium feedback
            ("Do you offer gift wrapping?", 0.85),  # Good feedback
            ("How do I cancel my subscription?", 0.6),  # Lower feedback
            ("What payment methods do you accept?", 0.9),  # High feedback
        ]
        
        for user_input, feedback in mixed_interactions:
            agent.interact_and_learn(user_input, feedback)
        
        # Example 4: Detecting concept drift
        print("\n" + "="*70)
        print("Example 4: Simulating Performance Degradation (Drift)")
        print("="*70)
        
        # Simulate declining performance
        drift_interactions = [
            ("Question about new feature X", 0.4),  # New topics, poor performance
            ("Tell me about feature Y", 0.3),
            ("How does feature Z work?", 0.35),
            ("Feature A questions", 0.4),
            ("Feature B support", 0.3),
        ]
        
        for user_input, feedback in drift_interactions:
            agent.interact_and_learn(user_input, feedback)
        
        # Final statistics
        print("\n" + "="*70)
        print("Final Learning Statistics:")
        print("="*70)
        final_stats = agent.get_learning_stats()
        print(json.dumps(final_stats, indent=2))
        
        # Summary
        print("\n" + "="*70)
        print("SUMMARY")
        print("="*70)
        print("""
    Online Learning Pattern demonstrated:
    
    Key Features:
    1. Incremental Learning - Updates knowledge with each interaction
    2. Example Buffer - Maintains recent and best examples
    3. Performance Tracking - Monitors learning effectiveness
    4. Drift Detection - Identifies when patterns change
    5. Dynamic Adaptation - Adjusts prompts based on learned examples
    
    Learning Mechanisms:
    - Few-shot learning with best examples
    - Continuous feedback integration
    - Performance trend analysis
    - Concept drift detection
    - Adaptive example selection
    
    Applications:
    - Customer support bots improving from interactions
    - Personalized assistants learning user preferences
    - Recommendation systems adapting to trends
    - Fraud detection with evolving patterns
    - Adaptive tutoring systems
    
    Benefits:
    - Continuous improvement without retraining
    - Adaptation to changing patterns
    - Personalization based on interaction history
    - Early warning for performance degradation
    - Efficient learning from limited examples
    
    Limitations:
    - Requires consistent feedback
    - May need periodic full retraining
    - Can accumulate biases from feedback
    - Limited by context window size
        """)
    
    
    if __name__ == "__main__":
        demonstrate_online_learning()
    
        

    Agentic Design Pattern: Online Learning

    Pattern 114: Online Learning – Line-by-Line Explanation

    Overview

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


    Pattern Description

    Pattern 114: Online Learning

    Description: Online Learning enables agents to continuously learn and adapt from incoming data streams in real-time. Unlike batch learning, online learning updates the agent’s knowledge incrementally as new data arrives, allowing adaptation to changing environments and concept drift.

    This pattern is crucial for agents operating in dynamic environments where
    patterns, user preferences, or optimal strategies change over time. It
    combines incremental updates, drift detection, and continuous adaptation.
    

    Key Components: 1. Stream Processor: Handles incoming data streams 2. Incremental Updater: Updates knowledge incrementally 3. Drift Detector: Identifies when patterns change 4. Performance Monitor: Tracks learning effectiveness 5. Model Adapter: Adjusts model based on feedback 6. Forgetting Mechanism: Manages outdated knowledge

    Learning Strategies: – Incremental Learning: Update with each new example – Mini-Batch: Update with small batches – Windowed: Learn from recent window – Adaptive Rate: Adjust learning rate dynamically – Ensemble: Maintain multiple learners

    Use Cases: – Real-time recommendation systems – Adaptive chatbots learning from conversations – Fraud detection with evolving patterns – Personalization engines – Market prediction systems

    LangChain Implementation: Uses conversation memory, feedback integration, and dynamic prompt adjustment to enable continuous learning from interactions.


    File Statistics

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

    Module Documentation

    """
    Pattern 114: Online Learning
    
    Description:
        Online Learning enables agents to continuously learn and adapt from incoming
        data streams in real-time. Unlike batch learning, online learning updates
        the agent's knowledge incrementally as new data arrives, allowing adaptation
        to changing environments and concept drift.
    
        This pattern is crucial for agents operating in dynamic environments where
        patterns, user preferences, or optimal strategies change over time. It
        combines incremental updates, drift detection, and continuous adaptation.
    
    Key Components:
        1. Stream Processor: Handles incoming data streams
        2. Incremental Updater: Updates knowledge incrementally
        3. Drift Detector: Identifies when patterns change
        4. Performance Monitor: Tracks learning effectiveness
        5. Model Adapter: Adjusts model based on feedback
        6. Forgetting Mechanism: Manages outdated knowledge
    
    Learning Strategies:
        - Incremental Learning: Update with each new example
        - Mini-Batch: Update with small batches
        - Windowed: Learn from recent window
        - Adaptive Rate: Adjust learning rate dynamically
        - Ensemble: Maintain multiple learners
    
    Use Cases:
        - Real-time recommendation systems
        - Adaptive chatbots learning from conversations
        - Fraud detection with evolving patterns
        - Personalization engines
        - Market prediction systems
    
    LangChain Implementation:
        Uses conversation memory, feedback integration, and dynamic prompt
        adjustment to enable continuous learning from interactions.
    """
    

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


    Import Statements

    Standard Library Imports

    import collections.deque
    import datetime.datetime
    import json
    import langchain_core.output_parsers.StrOutputParser
    import langchain_core.prompts.ChatPromptTemplate
    import langchain_core.prompts.FewShotChatMessagePromptTemplate
    import os
    import typing.Any
    import typing.Deque
    import typing.Dict
    import typing.List
    import typing.Optional
    

    Purpose: Core Python functionality: – collections.deque: Specialized container datatypes – datetime.datetime: Date and time handling – json: JSON serialization/deserialization – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – langchain_core.prompts.FewShotChatMessagePromptTemplate: Regular expression operations – os: Environment variable access and system operations – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Deque: 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.memory.ConversationBufferMemory
    from langchain_openai.ChatOpenAI
    

    Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows – Memory: State management across interactions

    Other Imports

    from dotenv.load_dotenv
    

    Purpose: Additional dependencies for specialized functionality.


    Classes

    Class: ExampleBuffer

    Lines 56-135

    Purpose:

    Manages a buffer of learning examples
    

    Methods:

    __init__(self, max_size)

    Lines 59-62

    Purpose: Initialize the class instance with required parameters.

    Parameters: – max_size: Max size

    add_example(self, input_text, output_text, feedback_score)

    Lines 64-76

    Purpose:

    Add a new example to the buffer
    

    Parameters: – input_text: Input data – output_text: Output text – feedback_score: Feedback score

    get_recent_examples(self, n)

    Lines 78-81

    Purpose:

    Get n most recent examples
    

    Parameters: – n: N

    Returns: <ast.Subscript object at 0x000001A06D88BE90>

    get_best_examples(self, n)

    Lines 83-97

    Purpose:

    Get n best-rated examples
    

    Parameters: – n: N

    Returns: <ast.Subscript object at 0x000001A06D765F50>

    detect_drift(self, window_size)

    Lines 99-113

    Purpose:

    Detect if performance is degrading (concept drift)
    

    Parameters: – window_size: Window size

    Returns: bool

    get_performance_trend(self)

    Lines 115-135

    Purpose:

    Get description of performance trend
    

    Parameters:

    Returns: str


    Class: OnlineLearningAgent

    Lines 138-254

    Purpose:

    Agent that learns continuously from interactions
    

    Methods:

    __init__(self, model_name, domain)

    Lines 141-156

    Purpose: Initialize the class instance with required parameters.

    Parameters: – model_name: Language model instance – domain: Domain

    create_dynamic_prompt(self, use_best)

    Lines 158-192

    Purpose:

    Create prompt with dynamic examples from learning buffer
    

    Parameters: – use_best: Use best

    Returns: ChatPromptTemplate

    respond(self, user_input)

    Lines 194-204

    Purpose:

    Generate response using current learned knowledge
    

    Parameters: – user_input: Input data

    Returns: str

    learn_from_feedback(self, user_input, response, feedback_score)

    Lines 206-221

    Purpose:

    Learn from user feedback on a response
    

    Parameters: – user_input: Input data – response: Response – feedback_score: Feedback score

    interact_and_learn(self, user_input, feedback_score)

    Lines 223-244

    Purpose:

    Complete interaction cycle with learning
    

    Parameters: – user_input: Input data – feedback_score: Feedback score

    Returns: <ast.Subscript object at 0x000001A06D16DBD0>

    get_learning_stats(self)

    Lines 246-254

    Purpose:

    Get statistics about learning progress
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06E094F10>


    Functions

    demonstrate_online_learning()

    Lines 257-380

    Purpose:

    Demonstrate online learning pattern
    

    Key Concepts Summary

    Pattern Overview

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

    Core Components

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

    When to Use

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

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


    Best Practices Demonstrated

    1. Environment Configuration

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

    2. Error Handling

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

    3. Prompt Engineering

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

    4. Code Organization

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

    5. Performance Optimization

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

    LangChain Components Used

    • ChatOpenAI: OpenAI model interface
    • Chains: Sequential component composition
    • Memory: Conversation state management

    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 113
    • Pattern 115

    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 Online Learning pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

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

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

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


    File: 114_online_learning.py
    Total Lines: 385
    Pattern Category: AI Agent Design Pattern #114


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Contextual Adaptation

    
        """
    Pattern 113: Contextual Adaptation
    
    Description:
        Contextual Adaptation enables agents to dynamically adjust their behavior,
        communication style, and strategies based on the detected context. The
        pattern monitors multiple context dimensions (user preferences, environment,
        task type, constraints) and adapts agent behavior accordingly.
        
        This pattern is essential for creating flexible, user-friendly agents that
        can operate effectively across diverse situations and user needs. It combines
        context detection, preference learning, and dynamic behavior adjustment.
    
    Key Components:
        1. Context Detector: Identifies current context dimensions
        2. Preference Learner: Learns user/situation preferences
        3. Behavior Adapter: Adjusts agent behavior
        4. Style Adapter: Modifies communication style
        5. Strategy Selector: Chooses appropriate strategies
        6. Constraint Manager: Handles situational constraints
    
    Context Dimensions:
        - User: Experience level, preferences, goals
        - Environment: Location, time, available resources
        - Task: Type, complexity, urgency, importance
        - Social: Formality, relationship, cultural context
        - Technical: Device, bandwidth, capabilities
        - Domain: Field-specific conventions and norms
    
    Use Cases:
        - Personalized assistants
        - Adaptive learning systems
        - Multi-domain chatbots
        - Context-aware recommendations
        - International applications
    
    LangChain Implementation:
        Uses dynamic prompting, context tracking, and conditional routing to
        adapt agent behavior based on detected context.
    """
    
    import os
    from typing import List, Dict, Any, Optional
    from dotenv import load_dotenv
    from datetime import datetime
    from enum import Enum
    
    from langchain_openai import ChatOpenAI
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
    from langchain.memory import ConversationBufferMemory
    
    load_dotenv()
    
    
    class UserExperience(Enum):
        """User experience levels"""
        BEGINNER = "beginner"
        INTERMEDIATE = "intermediate"
        ADVANCED = "advanced"
        EXPERT = "expert"
    
    
    class Formality(Enum):
        """Communication formality levels"""
        CASUAL = "casual"
        NEUTRAL = "neutral"
        FORMAL = "formal"
        VERY_FORMAL = "very_formal"
    
    
    class TaskUrgency(Enum):
        """Task urgency levels"""
        LOW = "low"
        MEDIUM = "medium"
        HIGH = "high"
        CRITICAL = "critical"
    
    
    class Context:
        """Represents current context"""
        
        def __init__(self):
            # User context
            self.user_experience = UserExperience.INTERMEDIATE
            self.user_preferences = {}
            self.user_goals = []
            
            # Environment context
            self.time_of_day = "morning"
            self.location = "office"
            self.available_resources = []
            
            # Task context
            self.task_type = "general"
            self.task_complexity = "medium"
            self.task_urgency = TaskUrgency.MEDIUM
            
            # Social context
            self.formality = Formality.NEUTRAL
            self.relationship = "professional"
            
            # Technical context
            self.device = "desktop"
            self.bandwidth = "high"
            
            # Domain context
            self.domain = "general"
            self.domain_conventions = {}
        
        def to_dict(self) -> Dict[str, Any]:
            """Convert context to dictionary"""
            return {
                "user": {
                    "experience": self.user_experience.value,
                    "preferences": self.user_preferences,
                    "goals": self.user_goals
                },
                "environment": {
                    "time": self.time_of_day,
                    "location": self.location,
                    "resources": self.available_resources
                },
                "task": {
                    "type": self.task_type,
                    "complexity": self.task_complexity,
                    "urgency": self.task_urgency.value
                },
                "social": {
                    "formality": self.formality.value,
                    "relationship": self.relationship
                },
                "technical": {
                    "device": self.device,
                    "bandwidth": self.bandwidth
                },
                "domain": {
                    "name": self.domain,
                    "conventions": self.domain_conventions
                }
            }
        
        def summarize(self) -> str:
            """Generate human-readable context summary"""
            return f"""
    Current Context:
    - User: {self.user_experience.value} level, wants {', '.join(self.user_goals) if self.user_goals else 'assistance'}
    - Environment: {self.time_of_day} at {self.location}
    - Task: {self.task_type} ({self.task_complexity} complexity, {self.task_urgency.value} urgency)
    - Communication: {self.formality.value} style, {self.relationship} relationship
    - Technical: {self.device} with {self.bandwidth} bandwidth
    - Domain: {self.domain}
    """
    
    
    class ContextualAdaptationAgent:
        """Agent that adapts behavior based on context"""
        
        def __init__(self, model_name: str = "gpt-4"):
            self.llm = ChatOpenAI(model=model_name, temperature=0.7)
            self.context = Context()
            self.memory = ConversationBufferMemory(return_messages=True)
            
            # Adaptive response generation
            self.adaptive_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are an adaptive AI assistant that adjusts your behavior based on context.
    
    {context_summary}
    
    Adaptation Guidelines:
    - For beginners: Use simple language, explain concepts, provide examples
    - For experts: Be concise, use technical terms, assume background knowledge
    - For casual contexts: Be friendly and conversational
    - For formal contexts: Be professional and structured
    - For urgent tasks: Be direct and action-oriented
    - For complex tasks: Break down steps and provide detailed guidance
    
    Adapt your response to match the context appropriately."""),
                ("human", "{query}")
            ])
            
            # Context detection prompt
            self.context_detector = ChatPromptTemplate.from_messages([
                ("system", """Analyze the user's query and extract context information.
    
    Determine:
    1. User experience level (beginner/intermediate/advanced/expert)
    2. Formality expected (casual/neutral/formal/very_formal)
    3. Task urgency (low/medium/high/critical)
    4. Task type and complexity
    
    Return JSON with: experience, formality, urgency, task_type, complexity"""),
                ("human", "{query}")
            ])
            
            self.response_chain = self.adaptive_prompt | self.llm | StrOutputParser()
            self.detector_chain = self.context_detector | self.llm | JsonOutputParser()
        
        def detect_context(self, query: str) -> Dict[str, Any]:
            """Detect context from user query"""
            try:
                detected = self.detector_chain.invoke({"query": query})
                
                # Update context based on detection
                if "experience" in detected:
                    exp_map = {
                        "beginner": UserExperience.BEGINNER,
                        "intermediate": UserExperience.INTERMEDIATE,
                        "advanced": UserExperience.ADVANCED,
                        "expert": UserExperience.EXPERT
                    }
                    self.context.user_experience = exp_map.get(
                        detected["experience"], UserExperience.INTERMEDIATE
                    )
                
                if "formality" in detected:
                    form_map = {
                        "casual": Formality.CASUAL,
                        "neutral": Formality.NEUTRAL,
                        "formal": Formality.FORMAL,
                        "very_formal": Formality.VERY_FORMAL
                    }
                    self.context.formality = form_map.get(
                        detected["formality"], Formality.NEUTRAL
                    )
                
                if "urgency" in detected:
                    urg_map = {
                        "low": TaskUrgency.LOW,
                        "medium": TaskUrgency.MEDIUM,
                        "high": TaskUrgency.HIGH,
                        "critical": TaskUrgency.CRITICAL
                    }
                    self.context.task_urgency = urg_map.get(
                        detected["urgency"], TaskUrgency.MEDIUM
                    )
                
                if "task_type" in detected:
                    self.context.task_type = detected["task_type"]
                
                if "complexity" in detected:
                    self.context.task_complexity = detected["complexity"]
                
                return detected
            except Exception as e:
                print(f"Context detection error: {e}")
                return {}
        
        def update_context(self, updates: Dict[str, Any]):
            """Manually update context"""
            for key, value in updates.items():
                if hasattr(self.context, key):
                    setattr(self.context, key, value)
        
        def respond(self, query: str, auto_detect: bool = True) -> Dict[str, Any]:
            """Generate context-adapted response"""
            print(f"\n{'='*60}")
            print(f"Query: {query}")
            print(f"{'='*60}")
            
            # Auto-detect context if enabled
            if auto_detect:
                detected = self.detect_context(query)
                print(f"\nDetected context: {detected}")
            
            # Show current context
            print(self.context.summarize())
            
            # Generate adapted response
            response = self.response_chain.invoke({
                "context_summary": self.context.summarize(),
                "query": query
            })
            
            print(f"\nAdapted Response:")
            print(response)
            
            return {
                "query": query,
                "context": self.context.to_dict(),
                "response": response,
                "timestamp": datetime.now().isoformat()
            }
        
        def demonstrate_adaptation(self, query: str, contexts: List[Dict[str, Any]]):
            """Demonstrate how response changes with different contexts"""
            print(f"\n{'='*70}")
            print(f"Demonstrating Context Adaptation")
            print(f"Query: {query}")
            print(f"{'='*70}")
            
            responses = []
            for i, context_updates in enumerate(contexts, 1):
                print(f"\n--- Context {i}: {context_updates.get('description', '')} ---")
                
                # Reset context
                self.context = Context()
                
                # Apply updates
                for key, value in context_updates.items():
                    if key != "description" and hasattr(self.context, key):
                        setattr(self.context, key, value)
                
                # Generate response
                result = self.respond(query, auto_detect=False)
                responses.append(result)
            
            return responses
    
    
    def demonstrate_contextual_adaptation():
        """Demonstrate contextual adaptation pattern"""
        print("\n" + "="*70)
        print("CONTEXTUAL ADAPTATION DEMONSTRATION")
        print("="*70)
        
        agent = ContextualAdaptationAgent()
        
        # Example 1: Auto-detection of context
        print("\n" + "="*70)
        print("Example 1: Auto-Detection of Context")
        print("="*70)
        
        query1 = "I'm new to Python. Can you explain what a variable is?"
        result1 = agent.respond(query1, auto_detect=True)
        
        query2 = "Explain the time complexity of quicksort's partitioning scheme"
        result2 = agent.respond(query2, auto_detect=True)
        
        # Example 2: Adaptation to different user levels
        print("\n" + "="*70)
        print("Example 2: Adapting to User Experience Level")
        print("="*70)
        
        query = "How does machine learning work?"
        
        contexts = [
            {
                "description": "Beginner",
                "user_experience": UserExperience.BEGINNER,
                "formality": Formality.CASUAL
            },
            {
                "description": "Expert",
                "user_experience": UserExperience.EXPERT,
                "formality": Formality.NEUTRAL
            }
        ]
        
        responses = agent.demonstrate_adaptation(query, contexts)
        
        # Example 3: Adaptation to urgency
        print("\n" + "="*70)
        print("Example 3: Adapting to Task Urgency")
        print("="*70)
        
        query_urgent = "My server is down! How do I restart it?"
        
        contexts_urgent = [
            {
                "description": "Critical Urgency",
                "task_urgency": TaskUrgency.CRITICAL,
                "task_type": "troubleshooting"
            },
            {
                "description": "Low Urgency",
                "task_urgency": TaskUrgency.LOW,
                "task_type": "learning"
            }
        ]
        
        responses_urgent = agent.demonstrate_adaptation(query_urgent, contexts_urgent)
        
        # Example 4: Adaptation to formality
        print("\n" + "="*70)
        print("Example 4: Adapting to Formality Level")
        print("="*70)
        
        query_formal = "What are your capabilities?"
        
        contexts_formal = [
            {
                "description": "Casual",
                "formality": Formality.CASUAL,
                "relationship": "friendly"
            },
            {
                "description": "Very Formal",
                "formality": Formality.VERY_FORMAL,
                "relationship": "professional"
            }
        ]
        
        responses_formal = agent.demonstrate_adaptation(query_formal, contexts_formal)
        
        # Summary
        print("\n" + "="*70)
        print("SUMMARY")
        print("="*70)
        print("""
    Contextual Adaptation Pattern demonstrated:
    
    Key Features:
    1. Context Detection - Automatically detect context from input
    2. Multi-Dimensional Adaptation - Adapt across multiple context dimensions
    3. Dynamic Behavior - Adjust communication style and strategy
    4. User Personalization - Adapt to user experience and preferences
    5. Situational Awareness - Respond appropriately to urgency and formality
    
    Context Dimensions Handled:
    - User experience level (beginner to expert)
    - Communication formality (casual to very formal)
    - Task urgency (low to critical)
    - Task complexity and type
    - Environment and technical constraints
    - Domain-specific conventions
    
    Applications:
    - Personalized assistants adapting to user expertise
    - Customer service bots adjusting tone
    - Educational systems matching student level
    - Multi-domain chatbots switching contexts
    - International applications respecting cultural norms
    
    Contextual adaptation makes agents more natural, effective, and user-friendly
    by tailoring their behavior to the specific situation and user needs.
        """)
    
    
    if __name__ == "__main__":
        demonstrate_contextual_adaptation()
    
        

    Agentic Design Pattern: Contextual Adaptation

    Pattern 113: Contextual Adaptation – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Contextual Adaptation pattern implementation.


    Pattern Description

    Pattern 113: Contextual Adaptation

    Description: Contextual Adaptation enables agents to dynamically adjust their behavior, communication style, and strategies based on the detected context. The pattern monitors multiple context dimensions (user preferences, environment, task type, constraints) and adapts agent behavior accordingly.

    This pattern is essential for creating flexible, user-friendly agents that
    can operate effectively across diverse situations and user needs. It combines
    context detection, preference learning, and dynamic behavior adjustment.
    

    Key Components: 1. Context Detector: Identifies current context dimensions 2. Preference Learner: Learns user/situation preferences 3. Behavior Adapter: Adjusts agent behavior 4. Style Adapter: Modifies communication style 5. Strategy Selector: Chooses appropriate strategies 6. Constraint Manager: Handles situational constraints

    Context Dimensions: – User: Experience level, preferences, goals – Environment: Location, time, available resources – Task: Type, complexity, urgency, importance – Social: Formality, relationship, cultural context – Technical: Device, bandwidth, capabilities – Domain: Field-specific conventions and norms

    Use Cases: – Personalized assistants – Adaptive learning systems – Multi-domain chatbots – Context-aware recommendations – International applications

    LangChain Implementation: Uses dynamic prompting, context tracking, and conditional routing to adapt agent behavior based on detected context.


    File Statistics

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

    Module Documentation

    """
    Pattern 113: Contextual Adaptation
    
    Description:
        Contextual Adaptation enables agents to dynamically adjust their behavior,
        communication style, and strategies based on the detected context. The
        pattern monitors multiple context dimensions (user preferences, environment,
        task type, constraints) and adapts agent behavior accordingly.
    
        This pattern is essential for creating flexible, user-friendly agents that
        can operate effectively across diverse situations and user needs. It combines
        context detection, preference learning, and dynamic behavior adjustment.
    
    Key Components:
        1. Context Detector: Identifies current context dimensions
        2. Preference Learner: Learns user/situation preferences
        3. Behavior Adapter: Adjusts agent behavior
        4. Style Adapter: Modifies communication style
        5. Strategy Selector: Chooses appropriate strategies
        6. Constraint Manager: Handles situational constraints
    
    Context Dimensions:
        - User: Experience level, preferences, goals
        - Environment: Location, time, available resources
        - Task: Type, complexity, urgency, importance
        - Social: Formality, relationship, cultural context
        - Technical: Device, bandwidth, capabilities
        - Domain: Field-specific conventions and norms
    
    Use Cases:
        - Personalized assistants
        - Adaptive learning systems
        - Multi-domain chatbots
        - Context-aware recommendations
        - International applications
    
    LangChain Implementation:
        Uses dynamic prompting, context tracking, and conditional routing to
        adapt agent behavior based on detected context.
    """
    

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


    Import Statements

    Standard Library Imports

    import datetime.datetime
    import enum.Enum
    import langchain_core.output_parsers.JsonOutputParser
    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: – datetime.datetime: Date and time handling – enum.Enum: Enumeration support – langchain_core.output_parsers.JsonOutputParser: Regular expression operations – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – 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.memory.ConversationBufferMemory
    from langchain_openai.ChatOpenAI
    

    Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows – Memory: State management across interactions

    Other Imports

    from dotenv.load_dotenv
    

    Purpose: Additional dependencies for specialized functionality.


    Classes

    Class: UserExperience

    Lines 56-61

    Inherits from: Enum

    Purpose:

    User experience levels
    

    Class: Formality

    Lines 64-69

    Inherits from: Enum

    Purpose:

    Communication formality levels
    

    Class: TaskUrgency

    Lines 72-77

    Inherits from: Enum

    Purpose:

    Task urgency levels
    

    Class: Context

    Lines 80-153

    Purpose:

    Represents current context
    

    Methods:

    __init__(self)

    Lines 83-109

    Purpose: Initialize the class instance with required parameters.

    Parameters:

    to_dict(self)

    Lines 111-141

    Purpose:

    Convert context to dictionary
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D798F90>

    summarize(self)

    Lines 143-153

    Purpose:

    Generate human-readable context summary
    

    Parameters:

    Returns: str


    Class: ContextualAdaptationAgent

    Lines 156-308

    Purpose:

    Agent that adapts behavior based on context
    

    Methods:

    __init__(self, model_name)

    Lines 159-197

    Purpose: Initialize the class instance with required parameters.

    Parameters: – model_name: Language model instance

    detect_context(self, query)

    Lines 199-247

    Purpose:

    Detect context from user query
    

    Parameters: – query: Input query or question

    Returns: <ast.Subscript object at 0x000001A06D125F50>

    update_context(self, updates)

    Lines 249-253

    Purpose:

    Manually update context
    

    Parameters: – updates: Updates

    respond(self, query, auto_detect)

    Lines 255-283

    Purpose:

    Generate context-adapted response
    

    Parameters: – query: Input query or question – auto_detect: Auto detect

    Returns: <ast.Subscript object at 0x000001A06D7FF950>

    demonstrate_adaptation(self, query, contexts)

    Lines 285-308

    Purpose:

    Demonstrate how response changes with different contexts
    

    Parameters: – query: Input query or question – contexts: Contexts


    Functions

    demonstrate_contextual_adaptation()

    Lines 311-427

    Purpose:

    Demonstrate contextual adaptation pattern
    

    Key Concepts Summary

    Pattern Overview

    The Contextual Adaptation 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
    • Memory: Conversation state management

    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 112
    • Pattern 114

    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 Contextual Adaptation 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: 113_contextual_adaptation.py
    Total Lines: 432
    Pattern Category: AI Agent Design Pattern #113


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Common Sense Reasoning

    
        """
    Pattern 112: Common Sense Reasoning
    
    Description:
        Common Sense Reasoning enables agents to apply implicit, everyday knowledge
        that humans take for granted. This pattern integrates common sense knowledge
        bases (like ConceptNet) with LLM reasoning to make inferences about the
        physical world, social norms, causality, and typical behaviors.
        
        The pattern helps agents understand context, make reasonable assumptions,
        avoid absurd conclusions, and reason about unstated but obvious facts.
        It bridges the gap between explicit knowledge and implicit understanding.
    
    Key Components:
        1. Knowledge Base: ConceptNet or similar common sense KB
        2. Reasoning Engine: Inference over common sense facts
        3. Context Integrator: Applies common sense to situations
        4. Plausibility Checker: Validates conclusions
        5. Assumption Generator: Makes reasonable assumptions
    
    Common Sense Categories:
        - Physical: Objects fall down, water is wet
        - Social: Greetings are polite, lying is wrong
        - Causal: Rain makes things wet, fire causes heat
        - Temporal: Events have order, time moves forward
        - Spatial: Objects can't be in two places
        - Functional: Tools have purposes, actions have effects
    
    Use Cases:
        - Natural language understanding
        - Story comprehension and generation
        - Situation interpretation
        - Question answering with implicit knowledge
        - Robotics and embodied AI
    
    LangChain Implementation:
        Uses ConceptNet integration, reasoning chains, and validation logic to
        apply common sense knowledge to agent reasoning.
    """
    
    import os
    from typing import List, Dict, Any, Optional, Tuple
    from dotenv import load_dotenv
    from datetime import datetime
    
    from langchain_openai import ChatOpenAI
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
    from langchain.chains import LLMChain
    
    load_dotenv()
    
    
    class CommonSenseKnowledgeBase:
        """Simulated common sense knowledge base (in practice, use ConceptNet API)"""
        
        def __init__(self):
            self.knowledge = {
                # Physical laws
                "gravity": "Objects fall downward when dropped",
                "water_wet": "Water makes things wet",
                "fire_hot": "Fire is hot and can burn",
                "ice_cold": "Ice is cold and can freeze",
                "glass_fragile": "Glass is fragile and can break",
                
                # Social norms
                "greeting_polite": "Greeting people is considered polite",
                "stealing_wrong": "Taking things without permission is wrong",
                "helping_good": "Helping others is generally good",
                "lying_negative": "Lying usually has negative consequences",
                
                # Causal relationships
                "rain_wet": "Rain makes outdoor surfaces wet",
                "eating_hunger": "Eating satisfies hunger",
                "sleep_rest": "Sleep provides rest and recovery",
                "exercise_fitness": "Exercise improves fitness",
                
                # Temporal facts
                "time_forward": "Time moves forward, not backward",
                "aging_continuous": "Living things age continuously",
                "seasons_cycle": "Seasons cycle in a predictable pattern",
                
                # Spatial facts
                "one_place": "Physical objects can't be in two places at once",
                "inside_outside": "Something inside a closed container is not outside",
                "solid_through": "Solid objects can't pass through solid walls",
                
                # Functional knowledge
                "key_lock": "Keys are used to open locks",
                "knife_cut": "Knives are used for cutting",
                "chair_sit": "Chairs are for sitting on",
                "umbrella_rain": "Umbrellas protect from rain",
            }
            
            self.categories = {
                "physical": ["gravity", "water_wet", "fire_hot", "ice_cold", "glass_fragile"],
                "social": ["greeting_polite", "stealing_wrong", "helping_good", "lying_negative"],
                "causal": ["rain_wet", "eating_hunger", "sleep_rest", "exercise_fitness"],
                "temporal": ["time_forward", "aging_continuous", "seasons_cycle"],
                "spatial": ["one_place", "inside_outside", "solid_through"],
                "functional": ["key_lock", "knife_cut", "chair_sit", "umbrella_rain"],
            }
        
        def query(self, concept: str) -> Optional[str]:
            """Query knowledge base for common sense fact"""
            return self.knowledge.get(concept)
        
        def get_category(self, category: str) -> List[str]:
            """Get all facts in a category"""
            concept_ids = self.categories.get(category, [])
            return [self.knowledge[cid] for cid in concept_ids if cid in self.knowledge]
        
        def get_relevant_facts(self, context: str) -> List[str]:
            """Get facts relevant to context"""
            # Simple keyword matching (in practice, use embeddings)
            relevant = []
            context_lower = context.lower()
            for key, fact in self.knowledge.items():
                if any(word in context_lower for word in key.split("_")):
                    relevant.append(fact)
            return relevant
    
    
    class CommonSenseReasoner:
        """Agent that applies common sense reasoning"""
        
        def __init__(self, model_name: str = "gpt-4"):
            self.llm = ChatOpenAI(model=model_name, temperature=0.7)
            self.kb = CommonSenseKnowledgeBase()
            
            # Reasoning with common sense
            self.reasoning_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a reasoning agent that applies common sense knowledge.
                
    Given a question or situation, use the provided common sense facts to reason
    about the answer. Make reasonable assumptions based on everyday knowledge.
    
    Common sense facts:
    {facts}
    
    Question: {question}
    
    Provide your reasoning and answer:"""),
                ("human", "{question}")
            ])
            
            # Plausibility checking
            self.plausibility_prompt = ChatPromptTemplate.from_messages([
                ("system", """Evaluate if a conclusion is plausible based on common sense.
                
    Common sense facts:
    {facts}
    
    Proposed conclusion: {conclusion}
    
    Rate plausibility from 0-10 and explain why:"""),
                ("human", "{conclusion}")
            ])
            
            # Assumption generation
            self.assumption_prompt = ChatPromptTemplate.from_messages([
                ("system", """Given a scenario, generate reasonable assumptions based on common sense.
                
    Scenario: {scenario}
    
    List 5 reasonable assumptions that most people would make:"""),
                ("human", "{scenario}")
            ])
            
            self.reasoning_chain = self.reasoning_prompt | self.llm | StrOutputParser()
            self.plausibility_chain = self.plausibility_prompt | self.llm | StrOutputParser()
            self.assumption_chain = self.assumption_prompt | self.llm | StrOutputParser()
        
        def reason_with_common_sense(self, question: str) -> Dict[str, Any]:
            """Apply common sense reasoning to answer a question"""
            print(f"\n{'='*60}")
            print(f"Question: {question}")
            print(f"{'='*60}")
            
            # Get relevant common sense facts
            relevant_facts = self.kb.get_relevant_facts(question)
            if not relevant_facts:
                relevant_facts = list(self.kb.knowledge.values())[:10]
            
            facts_str = "\n".join(f"- {fact}" for fact in relevant_facts)
            
            print(f"\nRelevant common sense facts:")
            for fact in relevant_facts:
                print(f"  - {fact}")
            
            # Reason with facts
            reasoning = self.reasoning_chain.invoke({
                "question": question,
                "facts": facts_str
            })
            
            print(f"\nReasoning:")
            print(reasoning)
            
            return {
                "question": question,
                "facts": relevant_facts,
                "reasoning": reasoning,
                "timestamp": datetime.now().isoformat()
            }
        
        def check_plausibility(self, conclusion: str, context: str = "") -> Dict[str, Any]:
            """Check if a conclusion is plausible"""
            print(f"\n{'='*60}")
            print(f"Checking plausibility: {conclusion}")
            print(f"{'='*60}")
            
            # Get relevant facts
            relevant_facts = self.kb.get_relevant_facts(conclusion + " " + context)
            if not relevant_facts:
                relevant_facts = list(self.kb.knowledge.values())[:10]
            
            facts_str = "\n".join(f"- {fact}" for fact in relevant_facts)
            
            # Check plausibility
            evaluation = self.plausibility_chain.invoke({
                "conclusion": conclusion,
                "facts": facts_str
            })
            
            print(f"\nEvaluation:")
            print(evaluation)
            
            return {
                "conclusion": conclusion,
                "facts": relevant_facts,
                "evaluation": evaluation,
                "timestamp": datetime.now().isoformat()
            }
        
        def generate_assumptions(self, scenario: str) -> Dict[str, Any]:
            """Generate reasonable assumptions for a scenario"""
            print(f"\n{'='*60}")
            print(f"Scenario: {scenario}")
            print(f"{'='*60}")
            
            # Generate assumptions
            assumptions = self.assumption_chain.invoke({
                "scenario": scenario
            })
            
            print(f"\nGenerated assumptions:")
            print(assumptions)
            
            return {
                "scenario": scenario,
                "assumptions": assumptions,
                "timestamp": datetime.now().isoformat()
            }
        
        def explain_with_common_sense(self, observation: str) -> Dict[str, Any]:
            """Explain an observation using common sense"""
            print(f"\n{'='*60}")
            print(f"Observation: {observation}")
            print(f"{'='*60}")
            
            # Get causal facts
            causal_facts = self.kb.get_category("causal")
            facts_str = "\n".join(f"- {fact}" for fact in causal_facts)
            
            explanation_prompt = f"""Using common sense causal knowledge, explain this observation:
    
    Observation: {observation}
    
    Common sense causal facts:
    {facts_str}
    
    Provide a simple, common sense explanation:"""
            
            explanation = self.llm.invoke(explanation_prompt).content
            
            print(f"\nExplanation:")
            print(explanation)
            
            return {
                "observation": observation,
                "causal_facts": causal_facts,
                "explanation": explanation,
                "timestamp": datetime.now().isoformat()
            }
    
    
    def demonstrate_common_sense_reasoning():
        """Demonstrate common sense reasoning patterns"""
        print("\n" + "="*70)
        print("COMMON SENSE REASONING DEMONSTRATION")
        print("="*70)
        
        reasoner = CommonSenseReasoner()
        
        # Example 1: Question answering with common sense
        print("\n" + "="*70)
        print("Example 1: Question Answering with Common Sense")
        print("="*70)
        
        question1 = "If I drop a glass cup on the floor, what will happen?"
        result1 = reasoner.reason_with_common_sense(question1)
        
        question2 = "Should I use an umbrella when it's raining?"
        result2 = reasoner.reason_with_common_sense(question2)
        
        # Example 2: Plausibility checking
        print("\n" + "="*70)
        print("Example 2: Plausibility Checking")
        print("="*70)
        
        conclusion1 = "I can walk through a solid brick wall"
        plausibility1 = reasoner.check_plausibility(conclusion1)
        
        conclusion2 = "If I leave ice cream outside in the sun, it will melt"
        plausibility2 = reasoner.check_plausibility(conclusion2)
        
        # Example 3: Assumption generation
        print("\n" + "="*70)
        print("Example 3: Assumption Generation")
        print("="*70)
        
        scenario = "A person enters a restaurant and sits at a table"
        assumptions = reasoner.generate_assumptions(scenario)
        
        # Example 4: Explanation with common sense
        print("\n" + "="*70)
        print("Example 4: Explaining Observations")
        print("="*70)
        
        observation1 = "The sidewalk is wet even though no one watered it"
        explanation1 = reasoner.explain_with_common_sense(observation1)
        
        observation2 = "The person looks tired and is yawning"
        explanation2 = reasoner.explain_with_common_sense(observation2)
        
        # Summary
        print("\n" + "="*70)
        print("SUMMARY")
        print("="*70)
        print("""
    Common Sense Reasoning Pattern demonstrated:
    
    Key Features:
    1. Knowledge Base Integration - Access to common sense facts
    2. Contextual Reasoning - Apply relevant facts to situations
    3. Plausibility Checking - Validate conclusions against common sense
    4. Assumption Generation - Make reasonable implicit assumptions
    5. Causal Explanation - Explain observations with common sense
    
    Applications:
    - Natural language understanding
    - Question answering systems
    - Story comprehension
    - Situation interpretation
    - Robotics and embodied AI
    
    Common sense reasoning helps agents bridge the gap between explicit
    knowledge and implicit understanding that humans naturally possess.
        """)
    
    
    if __name__ == "__main__":
        demonstrate_common_sense_reasoning()
    
        

    Agentic Design Pattern: Common Sense Reasoning

    Pattern 112: Common Sense Reasoning – Line-by-Line Explanation

    Overview

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


    Pattern Description

    Pattern 112: Common Sense Reasoning

    Description: Common Sense Reasoning enables agents to apply implicit, everyday knowledge that humans take for granted. This pattern integrates common sense knowledge bases (like ConceptNet) with LLM reasoning to make inferences about the physical world, social norms, causality, and typical behaviors.

    The pattern helps agents understand context, make reasonable assumptions,
    avoid absurd conclusions, and reason about unstated but obvious facts.
    It bridges the gap between explicit knowledge and implicit understanding.
    

    Key Components: 1. Knowledge Base: ConceptNet or similar common sense KB 2. Reasoning Engine: Inference over common sense facts 3. Context Integrator: Applies common sense to situations 4. Plausibility Checker: Validates conclusions 5. Assumption Generator: Makes reasonable assumptions

    Common Sense Categories: – Physical: Objects fall down, water is wet – Social: Greetings are polite, lying is wrong – Causal: Rain makes things wet, fire causes heat – Temporal: Events have order, time moves forward – Spatial: Objects can’t be in two places – Functional: Tools have purposes, actions have effects

    Use Cases: – Natural language understanding – Story comprehension and generation – Situation interpretation – Question answering with implicit knowledge – Robotics and embodied AI

    LangChain Implementation: Uses ConceptNet integration, reasoning chains, and validation logic to apply common sense knowledge to agent reasoning.


    File Statistics

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

    Module Documentation

    """
    Pattern 112: Common Sense Reasoning
    
    Description:
        Common Sense Reasoning enables agents to apply implicit, everyday knowledge
        that humans take for granted. This pattern integrates common sense knowledge
        bases (like ConceptNet) with LLM reasoning to make inferences about the
        physical world, social norms, causality, and typical behaviors.
    
        The pattern helps agents understand context, make reasonable assumptions,
        avoid absurd conclusions, and reason about unstated but obvious facts.
        It bridges the gap between explicit knowledge and implicit understanding.
    
    Key Components:
        1. Knowledge Base: ConceptNet or similar common sense KB
        2. Reasoning Engine: Inference over common sense facts
        3. Context Integrator: Applies common sense to situations
        4. Plausibility Checker: Validates conclusions
        5. Assumption Generator: Makes reasonable assumptions
    
    Common Sense Categories:
        - Physical: Objects fall down, water is wet
        - Social: Greetings are polite, lying is wrong
        - Causal: Rain makes things wet, fire causes heat
        - Temporal: Events have order, time moves forward
        - Spatial: Objects can't be in two places
        - Functional: Tools have purposes, actions have effects
    
    Use Cases:
        - Natural language understanding
        - Story comprehension and generation
        - Situation interpretation
        - Question answering with implicit knowledge
        - Robotics and embodied AI
    
    LangChain Implementation:
        Uses ConceptNet integration, reasoning chains, and validation logic to
        apply common sense knowledge to agent reasoning.
    """
    

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


    Import Statements

    Standard Library Imports

    import datetime.datetime
    import langchain_core.output_parsers.JsonOutputParser
    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: – datetime.datetime: Date and time handling – langchain_core.output_parsers.JsonOutputParser: Regular expression operations – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – 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.chains.LLMChain
    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: CommonSenseKnowledgeBase

    Lines 54-121

    Purpose:

    Simulated common sense knowledge base (in practice, use ConceptNet API)
    

    Methods:

    __init__(self)

    Lines 57-102

    Purpose: Initialize the class instance with required parameters.

    Parameters:

    query(self, concept)

    Lines 104-106

    Purpose:

    Query knowledge base for common sense fact
    

    Parameters: – concept: Concept

    Returns: <ast.Subscript object at 0x000001A06D6156D0>

    get_category(self, category)

    Lines 108-111

    Purpose:

    Get all facts in a category
    

    Parameters: – category: Category

    Returns: <ast.Subscript object at 0x000001A06D6168D0>

    get_relevant_facts(self, context)

    Lines 113-121

    Purpose:

    Get facts relevant to context
    

    Parameters: – context: Context

    Returns: <ast.Subscript object at 0x000001A06D73A2D0>


    Class: CommonSenseReasoner

    Lines 124-285

    Purpose:

    Agent that applies common sense reasoning
    

    Methods:

    __init__(self, model_name)

    Lines 127-172

    Purpose: Initialize the class instance with required parameters.

    Parameters: – model_name: Language model instance

    reason_with_common_sense(self, question)

    Lines 174-205

    Purpose:

    Apply common sense reasoning to answer a question
    

    Parameters: – question: Question to answer

    Returns: <ast.Subscript object at 0x000001A06D16E510>

    check_plausibility(self, conclusion, context)

    Lines 207-234

    Purpose:

    Check if a conclusion is plausible
    

    Parameters: – conclusion: Conclusion – context: Context

    Returns: <ast.Subscript object at 0x000001A06D824AD0>

    generate_assumptions(self, scenario)

    Lines 236-254

    Purpose:

    Generate reasonable assumptions for a scenario
    

    Parameters: – scenario: Scenario

    Returns: <ast.Subscript object at 0x000001A06D68C590>

    explain_with_common_sense(self, observation)

    Lines 256-285

    Purpose:

    Explain an observation using common sense
    

    Parameters: – observation: Observation

    Returns: <ast.Subscript object at 0x000001A06D888FD0>


    Functions

    demonstrate_common_sense_reasoning()

    Lines 288-360

    Purpose:

    Demonstrate common sense reasoning patterns
    

    Key Concepts Summary

    Pattern Overview

    The Common Sense 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 111
    • Pattern 113

    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 Common Sense 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: 112_common_sense_reasoning.py
    Total Lines: 365
    Pattern Category: AI Agent Design Pattern #112


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Situational Awareness

    
        """
    Pattern 111: Situational Awareness
    
    Description:
        Situational Awareness enables agents to maintain comprehensive understanding
        of their current context, environment, and state. The pattern continuously
        monitors and updates a model of the situation, including relevant entities,
        relationships, constraints, opportunities, and threats. This awareness informs
        decision-making and enables context-appropriate behavior.
        
        The pattern is crucial for adaptive agents operating in dynamic environments
        where context changes frequently. It combines perception, state estimation,
        context modeling, and predictive awareness to maintain an accurate and
        up-to-date understanding of the situation.
        
        Situational awareness includes environmental monitoring, state tracking,
        context interpretation, threat detection, opportunity identification, and
        predictive situation assessment.
    
    Key Components:
        1. Perception Module: Gathers situational data
        2. State Estimator: Tracks current state
        3. Context Modeler: Interprets situation meaning
        4. Change Detector: Identifies state changes
        5. Threat Assessor: Evaluates risks
        6. Opportunity Detector: Identifies opportunities
        7. Predictor: Anticipates future states
    
    Awareness Dimensions:
        - Spatial: Where things are
        - Temporal: When things happen
        - Causal: Why things occur
        - Social: Who is involved
        - Functional: What capabilities exist
        - Intentional: What goals are active
    
    Use Cases:
        - Autonomous vehicles understanding traffic
        - Security monitoring systems
        - Personal assistants with context awareness
        - Game AI with environmental awareness
        - Crisis management systems
    
    LangChain Implementation:
        Uses memory systems, context tracking, state estimation chains, and
        predictive models to maintain situational awareness.
    """
    
    import os
    from typing import List, Dict, Any, Optional
    from datetime import datetime
    from dotenv import load_dotenv
    from enum import Enum
    
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_openai import ChatOpenAI
    from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
    from langchain.memory import ConversationBufferMemory
    
    load_dotenv()
    
    
    class SituationStatus(Enum):
        """Enumeration of situation statuses."""
        NORMAL = "normal"
        ALERT = "alert"
        WARNING = "warning"
        CRITICAL = "critical"
        OPPORTUNITY = "opportunity"
    
    
    class SituationalAwarenessAgent:
        """
        Agent that maintains comprehensive awareness of current situation.
        """
        
        def __init__(self, agent_role: str = "general_agent"):
            """Initialize with role and context."""
            self.llm = ChatOpenAI(temperature=0.3, model="gpt-4")
            self.role = agent_role
            self.situation_memory = ConversationBufferMemory(
                memory_key="situation_history",
                return_messages=True
            )
            
            # Situational state
            self.current_context = {}
            self.active_entities = {}
            self.threats = []
            self.opportunities = []
            self.status = SituationStatus.NORMAL
        
        def perceive_environment(
            self,
            observations: Dict[str, Any]
        ) -> Dict[str, Any]:
            """
            Process environmental observations to update awareness.
            
            Args:
                observations: Dictionary of observed facts
                
            Returns:
                Updated situational model
            """
            print(f"\n{'='*60}")
            print("PERCEPTION & STATE ESTIMATION")
            print(f"{'='*60}")
            print(f"Processing {len(observations)} observations...")
            
            # Update context with observations
            self.current_context.update(observations)
            self.current_context['last_update'] = datetime.now().isoformat()
            
            # Analyze observations for entities
            prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a situational awareness system analyzing observations.
    Extract key entities, their states, and relationships from the observations.
    
    Provide a JSON response with:
    - entities: List of identified entities with properties
    - relationships: List of relationships between entities
    - key_facts: Important facts about the situation"""),
                ("user", "Observations: {observations}")
            ])
            
            chain = prompt | self.llm | JsonOutputParser()
            
            try:
                analysis = chain.invoke({"observations": str(observations)})
                
                # Update entities
                if 'entities' in analysis:
                    for entity in analysis['entities']:
                        entity_id = entity.get('id', entity.get('name', 'unknown'))
                        self.active_entities[entity_id] = entity
                
                print(f"Identified {len(self.active_entities)} entities")
                print(f"Key facts: {analysis.get('key_facts', [])}")
                
                return {
                    "perception": analysis,
                    "context_updated": True,
                    "timestamp": self.current_context['last_update']
                }
            except:
                # Fallback if JSON parsing fails
                return {
                    "perception": observations,
                    "context_updated": True,
                    "timestamp": self.current_context['last_update']
                }
        
        def assess_situation(self) -> Dict[str, Any]:
            """
            Assess current situation for threats and opportunities.
            
            Returns:
                Situation assessment with status and recommendations
            """
            print(f"\n{'='*60}")
            print("SITUATION ASSESSMENT")
            print(f"{'='*60}")
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a situational assessment expert. Analyze the current
    situation and provide:
    1. Overall situation status (normal/alert/warning/critical/opportunity)
    2. Identified threats or risks
    3. Identified opportunities
    4. Recommended actions
    5. Confidence in assessment (0-1)
    
    Format as: STATUS | Threats: [...] | Opportunities: [...] | Actions: [...] | Confidence: X.X"""),
                ("user", """Current Context: {context}
    Active Entities: {entities}
    Situation History: {history}
    
    Assess this situation comprehensively.""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            
            assessment_text = chain.invoke({
                "context": str(self.current_context),
                "entities": str(self.active_entities),
                "history": self.situation_memory.load_memory_variables({}).get('situation_history', [])
            })
            
            # Parse assessment
            parts = assessment_text.split('|')
            status_text = parts[0].strip().lower()
            
            # Update status
            if 'critical' in status_text:
                self.status = SituationStatus.CRITICAL
            elif 'warning' in status_text:
                self.status = SituationStatus.WARNING
            elif 'alert' in status_text:
                self.status = SituationStatus.ALERT
            elif 'opportunity' in status_text:
                self.status = SituationStatus.OPPORTUNITY
            else:
                self.status = SituationStatus.NORMAL
            
            result = {
                "status": self.status.value,
                "assessment": assessment_text,
                "context_size": len(self.current_context),
                "entities_tracked": len(self.active_entities),
                "timestamp": datetime.now().isoformat()
            }
            
            print(f"Status: {self.status.value.upper()}")
            print(f"Assessment:\n{assessment_text}")
            
            # Store in memory
            self.situation_memory.save_context(
                {"input": "situation_assessment"},
                {"output": assessment_text}
            )
            
            return result
        
        def predict_changes(
            self,
            time_horizon: str = "near future"
        ) -> Dict[str, Any]:
            """
            Predict likely changes to the situation.
            
            Args:
                time_horizon: Time frame for prediction
                
            Returns:
                Predictions about situation evolution
            """
            print(f"\n{'='*60}")
            print("PREDICTIVE AWARENESS")
            print(f"{'='*60}")
            print(f"Predicting changes for: {time_horizon}")
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a predictive analyst. Based on current situation,
    predict likely changes and developments.
    
    Provide:
    1. Most likely changes
    2. Probability estimates
    3. Potential triggers
    4. Recommended preparations"""),
                ("user", """Current Situation:
    Context: {context}
    Entities: {entities}
    Current Status: {status}
    
    Predict changes for: {time_horizon}""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            
            predictions = chain.invoke({
                "context": str(self.current_context),
                "entities": str(self.active_entities),
                "status": self.status.value,
                "time_horizon": time_horizon
            })
            
            result = {
                "time_horizon": time_horizon,
                "predictions": predictions,
                "based_on_entities": len(self.active_entities),
                "confidence": "medium"  # Could be calculated
            }
            
            print(f"Predictions:\n{predictions}")
            
            return result
        
        def context_appropriate_response(
            self,
            query: str
        ) -> Dict[str, Any]:
            """
            Generate response appropriate to current situational context.
            
            Args:
                query: Query or request
                
            Returns:
                Context-aware response
            """
            print(f"\n{'='*60}")
            print("CONTEXT-AWARE RESPONSE")
            print(f"{'='*60}")
            print(f"Query: {query}")
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", """You are an agent with full situational awareness. 
    Respond to queries considering the complete situational context.
    
    Your response should:
    1. Acknowledge relevant context
    2. Address the query appropriately
    3. Consider current status and constraints
    4. Provide context-aware recommendations"""),
                ("user", """Query: {query}
    
    Current Situation:
    - Status: {status}
    - Context: {context}
    - Active Entities: {entities}
    
    Provide a situationally appropriate response.""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            
            response = chain.invoke({
                "query": query,
                "status": self.status.value,
                "context": str(self.current_context)[:500],  # Limit size
                "entities": str(list(self.active_entities.keys()))
            })
            
            result = {
                "query": query,
                "response": response,
                "context_considered": True,
                "status_at_response": self.status.value
            }
            
            print(f"Response:\n{response}")
            
            return result
        
        def get_situation_summary(self) -> Dict[str, Any]:
            """Get comprehensive summary of current situation."""
            return {
                "status": self.status.value,
                "context": self.current_context,
                "entities": self.active_entities,
                "role": self.role,
                "last_update": self.current_context.get('last_update', 'never')
            }
    
    
    def demonstrate_situational_awareness():
        """Demonstrate situational awareness capabilities."""
        print("=" * 60)
        print("SITUATIONAL AWARENESS AGENT DEMONSTRATION")
        print("=" * 60)
        
        agent = SituationalAwarenessAgent(agent_role="security_monitor")
        
        # Example 1: Initial Perception
        print("\n" + "=" * 60)
        print("Example 1: Environmental Perception")
        print("=" * 60)
        
        observations1 = {
            "time": "14:30",
            "location": "Building A, Floor 3",
            "people_count": 15,
            "noise_level": "normal",
            "access_points": ["door_301", "door_302"],
            "cameras_active": True
        }
        
        result1 = agent.perceive_environment(observations1)
        
        # Example 2: Situation Assessment
        print("\n" + "=" * 60)
        print("Example 2: Situation Assessment")
        print("=" * 60)
        
        result2 = agent.assess_situation()
        
        # Example 3: Changed Situation
        print("\n" + "=" * 60)
        print("Example 3: Detecting Changes")
        print("=" * 60)
        
        observations2 = {
            "time": "14:35",
            "people_count": 25,  # Increased
            "noise_level": "elevated",
            "unauthorized_access_attempt": "door_304",
            "alarm_triggered": True
        }
        
        result3 = agent.perceive_environment(observations2)
        result4 = agent.assess_situation()
        
        # Example 4: Predictive Awareness
        print("\n" + "=" * 60)
        print("Example 4: Predicting Changes")
        print("=" * 60)
        
        result5 = agent.predict_changes("next 15 minutes")
        
        # Example 5: Context-Aware Response
        print("\n" + "=" * 60)
        print("Example 5: Context-Appropriate Response")
        print("=" * 60)
        
        result6 = agent.context_appropriate_response(
            "Should I allow the scheduled maintenance crew to enter?"
        )
        
        # Summary
        print("\n" + "=" * 60)
        print("SITUATION SUMMARY")
        print("=" * 60)
        summary = agent.get_situation_summary()
        print(f"Current Status: {summary['status'].upper()}")
        print(f"Entities Tracked: {len(summary['entities'])}")
        print(f"Context Elements: {len(summary['context'])}")
        print(f"Agent Role: {summary['role']}")
        
        print("\n" + "=" * 60)
        print("DEMONSTRATION COMPLETE")
        print("=" * 60)
        print("✓ Processed environmental observations")
        print("✓ Maintained situational context")
        print("✓ Detected and assessed changes")
        print("✓ Predicted future developments")
        print("✓ Generated context-aware responses")
    
    
    if __name__ == "__main__":
        demonstrate_situational_awareness()
    
        

    Agentic Design Pattern: Situational Awareness

    Pattern 111: Situational Awareness – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Situational Awareness pattern implementation.


    Pattern Description

    Pattern 111: Situational Awareness

    Description: Situational Awareness enables agents to maintain comprehensive understanding of their current context, environment, and state. The pattern continuously monitors and updates a model of the situation, including relevant entities, relationships, constraints, opportunities, and threats. This awareness informs decision-making and enables context-appropriate behavior.

    The pattern is crucial for adaptive agents operating in dynamic environments
    where context changes frequently. It combines perception, state estimation,
    context modeling, and predictive awareness to maintain an accurate and
    up-to-date understanding of the situation.
    
    Situational awareness includes environmental monitoring, state tracking,
    context interpretation, threat detection, opportunity identification, and
    predictive situation assessment.
    

    Key Components: 1. Perception Module: Gathers situational data 2. State Estimator: Tracks current state 3. Context Modeler: Interprets situation meaning 4. Change Detector: Identifies state changes 5. Threat Assessor: Evaluates risks 6. Opportunity Detector: Identifies opportunities 7. Predictor: Anticipates future states

    Awareness Dimensions: – Spatial: Where things are – Temporal: When things happen – Causal: Why things occur – Social: Who is involved – Functional: What capabilities exist – Intentional: What goals are active

    Use Cases: – Autonomous vehicles understanding traffic – Security monitoring systems – Personal assistants with context awareness – Game AI with environmental awareness – Crisis management systems

    LangChain Implementation: Uses memory systems, context tracking, state estimation chains, and predictive models to maintain situational awareness.


    File Statistics

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

    Module Documentation

    """
    Pattern 111: Situational Awareness
    
    Description:
        Situational Awareness enables agents to maintain comprehensive understanding
        of their current context, environment, and state. The pattern continuously
        monitors and updates a model of the situation, including relevant entities,
        relationships, constraints, opportunities, and threats. This awareness informs
        decision-making and enables context-appropriate behavior.
    
        The pattern is crucial for adaptive agents operating in dynamic environments
        where context changes frequently. It combines perception, state estimation,
        context modeling, and predictive awareness to maintain an accurate and
        up-to-date understanding of the situation.
    
        Situational awareness includes environmental monitoring, state tracking,
        context interpretation, threat detection, opportunity identification, and
        predictive situation assessment.
    
    Key Components:
        1. Perception Module: Gathers situational data
        2. State Estimator: Tracks current state
        3. Context Modeler: Interprets situation meaning
        4. Change Detector: Identifies state changes
        5. Threat Assessor: Evaluates risks
        6. Opportunity Detector: Identifies opportunities
        7. Predictor: Anticipates future states
    
    Awareness Dimensions:
        - Spatial: Where things are
        - Temporal: When things happen
        - Causal: Why things occur
        - Social: Who is involved
        - Functional: What capabilities exist
        - Intentional: What goals are active
    
    Use Cases:
        - Autonomous vehicles understanding traffic
        - Security monitoring systems
        - Personal assistants with context awareness
        - Game AI with environmental awareness
        - Crisis management systems
    
    LangChain Implementation:
        Uses memory systems, context tracking, state estimation chains, and
        predictive models to maintain situational awareness.
    """
    

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


    Import Statements

    Standard Library Imports

    import datetime.datetime
    import enum.Enum
    import langchain_core.output_parsers.JsonOutputParser
    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: – datetime.datetime: Date and time handling – enum.Enum: Enumeration support – langchain_core.output_parsers.JsonOutputParser: Regular expression operations – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – 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.memory.ConversationBufferMemory
    from langchain_openai.ChatOpenAI
    

    Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows – Memory: State management across interactions

    Other Imports

    from dotenv.load_dotenv
    

    Purpose: Additional dependencies for specialized functionality.


    Classes

    Class: SituationStatus

    Lines 63-69

    Inherits from: Enum

    Purpose:

    Enumeration of situation statuses.
    

    Class: SituationalAwarenessAgent

    Lines 72-345

    Purpose:

    Agent that maintains comprehensive awareness of current situation.
    

    Methods:

    __init__(self, agent_role)

    Lines 77-91

    Purpose:

    Initialize with role and context.
    

    Parameters: – agent_role: Agent role

    perceive_environment(self, observations)

    Lines 93-152

    Purpose:

    Process environmental observations to update awareness.
    
    Args:
        observations: Dictionary of observed facts
    
    Returns:
        Updated situational model
    

    Parameters: – observations: Observations

    Returns: <ast.Subscript object at 0x000001A06D7FDFD0>

    assess_situation(self)

    Lines 154-223

    Purpose:

    Assess current situation for threats and opportunities.
    
    Returns:
        Situation assessment with status and recommendations
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D72EE50>

    predict_changes(self, time_horizon)

    Lines 225-278

    Purpose:

    Predict likely changes to the situation.
    
    Args:
        time_horizon: Time frame for prediction
    
    Returns:
        Predictions about situation evolution
    

    Parameters: – time_horizon: Time horizon

    Returns: <ast.Subscript object at 0x000001A06D604250>

    context_appropriate_response(self, query)

    Lines 280-335

    Purpose:

    Generate response appropriate to current situational context.
    
    Args:
        query: Query or request
    
    Returns:
        Context-aware response
    

    Parameters: – query: Input query or question

    Returns: <ast.Subscript object at 0x000001A06D17C810>

    get_situation_summary(self)

    Lines 337-345

    Purpose:

    Get comprehensive summary of current situation.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D6F5AD0>


    Functions

    demonstrate_situational_awareness()

    Lines 348-428

    Purpose:

    Demonstrate situational awareness capabilities.
    

    Key Concepts Summary

    Pattern Overview

    The Situational Awareness 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
    • Memory: Conversation state management

    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 110
    • Pattern 112

    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 Situational Awareness 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: 111_situational_awareness.py
    Total Lines: 433
    Pattern Category: AI Agent Design Pattern #111


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Multi Modal Grounding

    
        """
    Pattern 110: Multi-Modal Grounding
    
    Description:
        Multi-Modal Grounding enables agents to understand and reason across multiple
        modalities simultaneously (text, images, audio, video, etc.). The pattern grounds
        abstract concepts and language in concrete perceptual experiences across different
        sensory modalities, enabling richer understanding and more contextually appropriate
        responses.
        
        This pattern is essential for embodied AI, robotics, and applications requiring
        comprehensive understanding of the real world. It integrates information from
        different modalities, resolves cross-modal references, and maintains coherent
        multimodal representations.
        
        The pattern includes vision-language integration, audio processing, cross-modal
        attention mechanisms, unified embeddings, and multimodal fusion strategies.
    
    Key Components:
        1. Modality Encoders: Process different input types
        2. Cross-Modal Attention: Link information across modalities
        3. Fusion Module: Combine multimodal representations
        4. Grounding Engine: Connect language to perception
        5. Unified Representation: Common embedding space
        6. Multimodal Reasoning: Reason across modalities
    
    Modalities Supported:
        - Text: Natural language understanding
        - Vision: Images, video frames
        - Audio: Speech, sounds
        - Structured Data: Tables, graphs
        - Temporal: Time-series data
    
    Use Cases:
        - Vision-language tasks (image captioning, VQA)
        - Robotics with multiple sensors
        - Video understanding and analysis
        - Multimodal document processing
        - Embodied AI agents
    
    LangChain Implementation:
        Uses multimodal LLMs (GPT-4V, Claude 3), custom processors for different
        modalities, and cross-modal reasoning chains.
    """
    
    import os
    from typing import List, Dict, Any, Optional, Union
    from dotenv import load_dotenv
    import base64
    from pathlib import Path
    
    from langchain_core.messages import HumanMessage, SystemMessage
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_openai import ChatOpenAI
    from langchain_core.output_parsers import StrOutputParser
    
    load_dotenv()
    
    
    class MultiModalGroundingAgent:
        """
        Agent that processes and reasons across multiple modalities.
        """
        
        def __init__(self, model_name: str = "gpt-4o"):
            """Initialize with multimodal LLM."""
            # Use multimodal model (GPT-4 with vision)
            self.llm = ChatOpenAI(model=model_name, temperature=0.7, max_tokens=1000)
            self.parser = StrOutputParser()
        
        def encode_image(self, image_path: str) -> str:
            """Encode image to base64."""
            with open(image_path, "rb") as image_file:
                return base64.b64encode(image_file.read()).decode('utf-8')
        
        def process_image_text(
            self,
            text_query: str,
            image_path: Optional[str] = None,
            image_url: Optional[str] = None
        ) -> Dict[str, Any]:
            """
            Process text query grounded in visual context.
            
            Args:
                text_query: Text question or instruction
                image_path: Local path to image file
                image_url: URL to image
                
            Returns:
                Dictionary with response and reasoning
            """
            print(f"\n{'='*60}")
            print("VISION-LANGUAGE GROUNDING")
            print(f"{'='*60}")
            print(f"Query: {text_query}")
            
            # Construct message with image
            content = []
            content.append({"type": "text", "text": text_query})
            
            if image_url:
                content.append({"type": "image_url", "image_url": {"url": image_url}})
                print(f"Image: {image_url}")
            elif image_path:
                # For local images, would need base64 encoding
                # Simplified for demonstration
                print(f"Image: {image_path}")
                # In real implementation: encode and add to content
            
            message = HumanMessage(content=content if image_url or image_path else text_query)
            
            # Process with multimodal LLM
            response = self.llm.invoke([message])
            
            result = {
                "query": text_query,
                "response": response.content,
                "modalities": ["text", "vision"] if (image_url or image_path) else ["text"],
                "grounded": bool(image_url or image_path)
            }
            
            print(f"\nResponse: {result['response']}")
            print(f"Modalities Used: {result['modalities']}")
            
            return result
        
        def compare_images(
            self,
            image_url_1: str,
            image_url_2: str,
            comparison_aspect: str = "general"
        ) -> Dict[str, Any]:
            """
            Compare two images and identify differences/similarities.
            
            Args:
                image_url_1: First image URL
                image_url_2: Second image URL
                comparison_aspect: What to compare (general, specific aspect)
                
            Returns:
                Comparison results
            """
            print(f"\n{'='*60}")
            print("CROSS-MODAL COMPARISON")
            print(f"{'='*60}")
            
            prompt = f"""Compare these two images focusing on {comparison_aspect}.
            
    Provide:
    1. Key similarities
    2. Key differences
    3. Notable features in each
    4. Overall assessment"""
            
            content = [
                {"type": "text", "text": prompt},
                {"type": "image_url", "image_url": {"url": image_url_1}},
                {"type": "image_url", "image_url": {"url": image_url_2}}
            ]
            
            message = HumanMessage(content=content)
            response = self.llm.invoke([message])
            
            result = {
                "aspect": comparison_aspect,
                "comparison": response.content,
                "images_compared": 2
            }
            
            print(f"Comparison Result:\n{result['comparison']}")
            
            return result
        
        def multimodal_reasoning(
            self,
            text_context: str,
            image_url: Optional[str] = None,
            structured_data: Optional[Dict] = None,
            reasoning_task: str = "analyze and synthesize"
        ) -> Dict[str, Any]:
            """
            Perform reasoning across multiple modalities.
            
            Args:
                text_context: Textual information
                image_url: Visual information
                structured_data: Structured data (tables, graphs)
                reasoning_task: Type of reasoning required
                
            Returns:
                Reasoning results integrating all modalities
            """
            print(f"\n{'='*60}")
            print("MULTIMODAL REASONING")
            print(f"{'='*60}")
            print(f"Task: {reasoning_task}")
            
            # Build multimodal context
            modalities_used = ["text"]
            content = [{"type": "text", "text": f"Context: {text_context}\n\nTask: {reasoning_task}"}]
            
            if structured_data:
                modalities_used.append("structured_data")
                data_str = "\n".join([f"{k}: {v}" for k, v in structured_data.items()])
                content[0]["text"] += f"\n\nStructured Data:\n{data_str}"
            
            if image_url:
                modalities_used.append("vision")
                content.append({"type": "image_url", "image_url": {"url": image_url}})
            
            content[0]["text"] += "\n\nProvide comprehensive analysis integrating all available information."
            
            message = HumanMessage(content=content)
            response = self.llm.invoke([message])
            
            result = {
                "task": reasoning_task,
                "modalities_used": modalities_used,
                "reasoning": response.content,
                "integrated": len(modalities_used) > 1
            }
            
            print(f"Modalities: {modalities_used}")
            print(f"Reasoning:\n{result['reasoning']}")
            
            return result
        
        def spatial_grounding(
            self,
            image_url: str,
            spatial_query: str
        ) -> Dict[str, Any]:
            """
            Ground spatial language in visual context.
            
            Args:
                image_url: Image to analyze
                spatial_query: Query about spatial relationships
                
            Returns:
                Spatial grounding results
            """
            print(f"\n{'='*60}")
            print("SPATIAL GROUNDING")
            print(f"{'='*60}")
            
            prompt = f"""Analyze the spatial relationships in this image.
    
    Query: {spatial_query}
    
    Provide:
    1. Identified objects and their locations
    2. Spatial relationships between objects
    3. Answer to the spatial query
    4. Confidence in your spatial understanding"""
            
            content = [
                {"type": "text", "text": prompt},
                {"type": "image_url", "image_url": {"url": image_url}}
            ]
            
            message = HumanMessage(content=content)
            response = self.llm.invoke([message])
            
            result = {
                "query": spatial_query,
                "spatial_analysis": response.content,
                "grounding_type": "spatial"
            }
            
            print(f"Query: {spatial_query}")
            print(f"Analysis:\n{result['spatial_analysis']}")
            
            return result
    
    
    def demonstrate_multimodal_grounding():
        """Demonstrate multi-modal grounding capabilities."""
        print("=" * 60)
        print("MULTI-MODAL GROUNDING AGENT DEMONSTRATION")
        print("=" * 60)
        
        agent = MultiModalGroundingAgent()
        
        # Example 1: Vision-Language Query
        print("\n" + "=" * 60)
        print("Example 1: Vision-Language Understanding")
        print("=" * 60)
        
        # Using a sample image URL (placeholder)
        sample_image_url = "https://upload.wikimedia.org/wikipedia/commons/thumb/d/dd/Gfp-wisconsin-madison-the-nature-boardwalk.jpg/2560px-Gfp-wisconsin-madison-the-nature-boardwalk.jpg"
        
        result1 = agent.process_image_text(
            text_query="What is in this image? Describe the scene in detail.",
            image_url=sample_image_url
        )
        
        # Example 2: Multimodal Reasoning
        print("\n" + "=" * 60)
        print("Example 2: Multimodal Reasoning Integration")
        print("=" * 60)
        
        result2 = agent.multimodal_reasoning(
            text_context="This is a natural landscape photograph taken during daylight hours.",
            image_url=sample_image_url,
            structured_data={
                "location": "Wisconsin",
                "type": "Nature boardwalk",
                "season": "Summer"
            },
            reasoning_task="Analyze the environmental conditions and recommend best time to visit"
        )
        
        # Example 3: Spatial Grounding
        print("\n" + "=" * 60)
        print("Example 3: Spatial Grounding")
        print("=" * 60)
        
        result3 = agent.spatial_grounding(
            image_url=sample_image_url,
            spatial_query="What objects are in the foreground vs background? Describe the depth perspective."
        )
        
        # Summary
        print("\n" + "=" * 60)
        print("SUMMARY")
        print("=" * 60)
        print(f"✓ Processed {3} multimodal examples")
        print("✓ Demonstrated vision-language understanding")
        print("✓ Integrated structured data with visual information")
        print("✓ Performed spatial grounding and reasoning")
        print("\nKey Capabilities:")
        print("  - Cross-modal attention and integration")
        print("  - Unified multimodal representations")
        print("  - Spatial and visual grounding")
        print("  - Context-aware multimodal reasoning")
    
    
    if __name__ == "__main__":
        demonstrate_multimodal_grounding()
    
        

    Agentic Design Pattern: Multi Modal Grounding

    Pattern 110: Multi Modal Grounding – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Multi Modal Grounding pattern implementation.


    Pattern Description

    Pattern 110: Multi-Modal Grounding

    Description: Multi-Modal Grounding enables agents to understand and reason across multiple modalities simultaneously (text, images, audio, video, etc.). The pattern grounds abstract concepts and language in concrete perceptual experiences across different sensory modalities, enabling richer understanding and more contextually appropriate responses.

    This pattern is essential for embodied AI, robotics, and applications requiring
    comprehensive understanding of the real world. It integrates information from
    different modalities, resolves cross-modal references, and maintains coherent
    multimodal representations.
    
    The pattern includes vision-language integration, audio processing, cross-modal
    attention mechanisms, unified embeddings, and multimodal fusion strategies.
    

    Key Components: 1. Modality Encoders: Process different input types 2. Cross-Modal Attention: Link information across modalities 3. Fusion Module: Combine multimodal representations 4. Grounding Engine: Connect language to perception 5. Unified Representation: Common embedding space 6. Multimodal Reasoning: Reason across modalities

    Modalities Supported: – Text: Natural language understanding – Vision: Images, video frames – Audio: Speech, sounds – Structured Data: Tables, graphs – Temporal: Time-series data

    Use Cases: – Vision-language tasks (image captioning, VQA) – Robotics with multiple sensors – Video understanding and analysis – Multimodal document processing – Embodied AI agents

    LangChain Implementation: Uses multimodal LLMs (GPT-4V, Claude 3), custom processors for different modalities, and cross-modal reasoning chains.


    File Statistics

    • Total Lines: 343
    • Classes: 1
    • Functions: 1
    • Constants: 0

    Module Documentation

    """
    Pattern 110: Multi-Modal Grounding
    
    Description:
        Multi-Modal Grounding enables agents to understand and reason across multiple
        modalities simultaneously (text, images, audio, video, etc.). The pattern grounds
        abstract concepts and language in concrete perceptual experiences across different
        sensory modalities, enabling richer understanding and more contextually appropriate
        responses.
    
        This pattern is essential for embodied AI, robotics, and applications requiring
        comprehensive understanding of the real world. It integrates information from
        different modalities, resolves cross-modal references, and maintains coherent
        multimodal representations.
    
        The pattern includes vision-language integration, audio processing, cross-modal
        attention mechanisms, unified embeddings, and multimodal fusion strategies.
    
    Key Components:
        1. Modality Encoders: Process different input types
        2. Cross-Modal Attention: Link information across modalities
        3. Fusion Module: Combine multimodal representations
        4. Grounding Engine: Connect language to perception
        5. Unified Representation: Common embedding space
        6. Multimodal Reasoning: Reason across modalities
    
    Modalities Supported:
        - Text: Natural language understanding
        - Vision: Images, video frames
        - Audio: Speech, sounds
        - Structured Data: Tables, graphs
        - Temporal: Time-series data
    
    Use Cases:
        - Vision-language tasks (image captioning, VQA)
        - Robotics with multiple sensors
        - Video understanding and analysis
        - Multimodal document processing
        - Embodied AI agents
    
    LangChain Implementation:
        Uses multimodal LLMs (GPT-4V, Claude 3), custom processors for different
        modalities, and cross-modal reasoning chains.
    """
    

    Purpose: This module implements the Multi Modal Grounding pattern, providing a structured approach to agent design and implementation.


    Import Statements

    Standard Library Imports

    import langchain_core.messages.HumanMessage
    import langchain_core.messages.SystemMessage
    import langchain_core.output_parsers.StrOutputParser
    import langchain_core.prompts.ChatPromptTemplate
    import os
    import pathlib.Path
    import typing.Any
    import typing.Dict
    import typing.List
    import typing.Optional
    import typing.Union
    

    Purpose: Core Python functionality: – langchain_core.messages.HumanMessage: Regular expression operations – langchain_core.messages.SystemMessage: Regular expression operations – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – pathlib.Path: Object-oriented filesystem paths – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Union: 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 base64
    from dotenv.load_dotenv
    

    Purpose: Additional dependencies for specialized functionality.


    Classes

    Class: MultiModalGroundingAgent

    Lines 60-276

    Purpose:

    Agent that processes and reasons across multiple modalities.
    

    Methods:

    __init__(self, model_name)

    Lines 65-69

    Purpose:

    Initialize with multimodal LLM.
    

    Parameters: – model_name: Language model instance

    encode_image(self, image_path)

    Lines 71-74

    Purpose:

    Encode image to base64.
    

    Parameters: – image_path: Image path

    Returns: str

    process_image_text(self, text_query, image_path, image_url)

    Lines 76-126

    Purpose:

    Process text query grounded in visual context.
    
    Args:
        text_query: Text question or instruction
        image_path: Local path to image file
        image_url: URL to image
    
    Returns:
        Dictionary with response and reasoning
    

    Parameters: – text_query: Input query or question – image_path: Image path – image_url: Image url

    Returns: <ast.Subscript object at 0x000001A06D88BF50>

    compare_images(self, image_url_1, image_url_2, comparison_aspect)

    Lines 128-174

    Purpose:

    Compare two images and identify differences/similarities.
    
    Args:
        image_url_1: First image URL
        image_url_2: Second image URL
        comparison_aspect: What to compare (general, specific aspect)
    
    Returns:
        Comparison results
    

    Parameters: – image_url_1: Image url 1 – image_url_2: Image url 2 – comparison_aspect: Comparison aspect

    Returns: <ast.Subscript object at 0x000001A06D71B3D0>

    multimodal_reasoning(self, text_context, image_url, structured_data, reasoning_task)

    Lines 176-228

    Purpose:

    Perform reasoning across multiple modalities.
    
    Args:
        text_context: Textual information
        image_url: Visual information
        structured_data: Structured data (tables, graphs)
        reasoning_task: Type of reasoning required
    
    Returns:
        Reasoning results integrating all modalities
    

    Parameters: – text_context: Text context – image_url: Image url – structured_data: Structured data – reasoning_task: Reasoning task

    Returns: <ast.Subscript object at 0x000001A06D824250>

    spatial_grounding(self, image_url, spatial_query)

    Lines 230-276

    Purpose:

    Ground spatial language in visual context.
    
    Args:
        image_url: Image to analyze
        spatial_query: Query about spatial relationships
    
    Returns:
        Spatial grounding results
    

    Parameters: – image_url: Image url – spatial_query: Input query or question

    Returns: <ast.Subscript object at 0x000001A06D16C350>


    Functions

    demonstrate_multimodal_grounding()

    Lines 279-338

    Purpose:

    Demonstrate multi-modal grounding capabilities.
    

    Key Concepts Summary

    Pattern Overview

    The Multi Modal Grounding 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 109
    • Pattern 111

    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 Multi Modal Grounding 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: 110_multi_modal_grounding.py
    Total Lines: 343
    Pattern Category: AI Agent Design Pattern #110


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Adaptive Planning

    
        """
    Pattern 109: Adaptive Planning
    
    Description:
        The Adaptive Planning pattern enables agents to dynamically adjust their plans
        in response to execution feedback, environmental changes, and unexpected events.
        Rather than rigidly following a predetermined plan, adaptive planners monitor
        execution, detect deviations from expected outcomes, diagnose problems, and
        modify plans on the fly to maintain progress toward goals.
        
        Adaptive planning is essential in dynamic, uncertain environments where initial
        assumptions may become invalid, resources may change, or unexpected opportunities
        and obstacles may arise. The pattern combines continuous monitoring, deviation
        detection, root cause analysis, and replanning strategies to ensure resilience
        and goal achievement despite changing conditions.
        
        This pattern includes mechanisms for execution monitoring, deviation detection,
        impact assessment, root cause diagnosis, plan repair strategies, replanning
        triggers, and learning from outcomes to improve future plans.
    
    Key Components:
        1. Execution Monitor: Tracks plan execution
        2. Deviation Detector: Identifies plan-reality gaps
        3. Impact Assessor: Evaluates deviation significance
        4. Root Cause Analyzer: Diagnoses problems
        5. Plan Repairer: Fixes plans locally
        6. Replanner: Creates new plans when needed
        7. Learning Module: Improves from experience
    
    Adaptation Strategies:
        1. Plan Repair: Minor modifications to existing plan
        2. Plan Reuse: Apply similar past plan
        3. Plan Revision: Significant restructuring
        4. Replanning: Create entirely new plan
        5. Reactive Adaptation: Immediate adjustments
        6. Proactive Adaptation: Anticipate changes
        7. Learning-based: Use past outcomes
    
    Monitoring Aspects:
        1. Progress: Task completion vs. expected
        2. Resources: Availability and consumption
        3. Time: Schedule adherence
        4. Quality: Output quality measures
        5. Cost: Budget tracking
        6. Risks: Emerging threats
        7. Opportunities: New possibilities
    
    Use Cases:
        - Robot navigation in dynamic environments
        - Project management with changing requirements
        - Manufacturing with equipment failures
        - Supply chain disruptions
        - Military operations in uncertain terrain
        - Autonomous vehicle route planning
        - Resource allocation in cloud systems
    
    Advantages:
        - Robust to uncertainty
        - Maintains progress despite changes
        - Exploits new opportunities
        - Reduces plan failures
        - Learns from experience
        - Handles unexpected events
        - Flexible goal achievement
    
    Challenges:
        - Monitoring overhead
        - Replanning latency
        - Plan stability vs. adaptability
        - Knowing when to replan
        - Avoiding thrashing (excessive replanning)
        - Preserving progress
        - Computational cost
    
    LangChain Implementation:
        This implementation uses LangChain for:
        - LLM-based deviation analysis
        - Root cause diagnosis
        - Plan repair suggestions
        - Natural language explanations
        
    Production Considerations:
        - Implement efficient monitoring
        - Use event-driven detection
        - Cache replan results
        - Set replanning thresholds
        - Limit replanning frequency
        - Preserve partial progress
        - Log all adaptations
        - Enable replay for debugging
    """
    
    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_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser
    from langchain_openai import ChatOpenAI
    
    load_dotenv()
    
    
    class StepStatus(Enum):
        """Status of a plan step."""
        PENDING = "pending"
        IN_PROGRESS = "in_progress"
        COMPLETED = "completed"
        FAILED = "failed"
        SKIPPED = "skipped"
    
    
    class DeviationType(Enum):
        """Types of plan deviations."""
        DELAY = "delay"
        RESOURCE_SHORTAGE = "resource_shortage"
        QUALITY_ISSUE = "quality_issue"
        COST_OVERRUN = "cost_overrun"
        DEPENDENCY_FAILURE = "dependency_failure"
        ENVIRONMENTAL_CHANGE = "environmental_change"
    
    
    class AdaptationAction(Enum):
        """Types of adaptation actions."""
        CONTINUE = "continue"
        REPAIR = "repair"
        REVISE = "revise"
        REPLAN = "replan"
        ABORT = "abort"
    
    
    @dataclass
    class PlanStep:
        """
        A step in an adaptive plan.
        
        Attributes:
            step_id: Unique identifier
            description: Step description
            expected_duration: Expected time to complete
            expected_cost: Expected cost
            dependencies: Prerequisite steps
            status: Current status
            started_at: When started
            completed_at: When completed
            actual_duration: Actual time taken
            actual_cost: Actual cost
        """
        step_id: str
        description: str
        expected_duration: float = 1.0
        expected_cost: float = 1.0
        dependencies: List[str] = field(default_factory=list)
        status: StepStatus = StepStatus.PENDING
        started_at: Optional[datetime] = None
        completed_at: Optional[datetime] = None
        actual_duration: Optional[float] = None
        actual_cost: Optional[float] = None
    
    
    @dataclass
    class Deviation:
        """
        Detected deviation from plan.
        
        Attributes:
            deviation_id: Unique identifier
            deviation_type: Type of deviation
            affected_step: Step with deviation
            severity: Severity score (0-10)
            description: Human-readable description
            detected_at: When detected
            resolved: Whether resolved
        """
        deviation_id: str
        deviation_type: DeviationType
        affected_step: str
        severity: float
        description: str
        detected_at: datetime = field(default_factory=datetime.now)
        resolved: bool = False
    
    
    @dataclass
    class Adaptation:
        """
        Adaptation made to plan.
        
        Attributes:
            adaptation_id: Unique identifier
            action: Type of adaptation
            reason: Why adaptation was made
            changes: Description of changes
            timestamp: When adaptation occurred
            success: Whether adaptation succeeded
        """
        adaptation_id: str
        action: AdaptationAction
        reason: str
        changes: str
        timestamp: datetime = field(default_factory=datetime.now)
        success: Optional[bool] = None
    
    
    class AdaptivePlanner:
        """
        Adaptive planning system with execution monitoring and replanning.
        
        This planner monitors execution, detects deviations, and adapts
        plans dynamically to maintain progress toward goals.
        """
        
        def __init__(self, temperature: float = 0.3):
            """
            Initialize adaptive planner.
            
            Args:
                temperature: LLM temperature
            """
            self.llm = ChatOpenAI(temperature=temperature, model="gpt-3.5-turbo")
            self.steps: Dict[str, PlanStep] = {}
            self.deviations: List[Deviation] = []
            self.adaptations: List[Adaptation] = []
            self.step_counter = 0
            self.deviation_counter = 0
            self.adaptation_counter = 0
            self.replanning_threshold = 5.0  # Severity threshold for replanning
        
        def create_plan(self, goal: str, num_steps: int = 5) -> List[PlanStep]:
            """
            Create initial plan using LLM.
            
            Args:
                goal: Goal description
                num_steps: Number of steps
                
            Returns:
                List of plan steps
            """
            prompt = ChatPromptTemplate.from_template(
                "Create a {num_steps}-step plan to achieve this goal:\n\n"
                "Goal: {goal}\n\n"
                "For each step, provide:\n"
                "- Description\n"
                "- Expected duration (hours)\n"
                "- Expected cost (units)\n\n"
                "Format as:\n"
                "Step N: [description]\n"
                "Duration: [hours]\n"
                "Cost: [units]\n"
                "---\n"
            )
            
            chain = prompt | self.llm | StrOutputParser()
            result = chain.invoke({"goal": goal, "num_steps": num_steps})
            
            # Parse steps
            steps = []
            step_blocks = result.split("---")
            
            for block in step_blocks:
                if not block.strip():
                    continue
                
                lines = [line.strip() for line in block.strip().split('\n')]
                description = ""
                duration = 1.0
                cost = 1.0
                
                for line in lines:
                    if line.startswith("Step"):
                        description = line.split(":", 1)[1].strip() if ":" in line else line
                    elif line.startswith("Duration:"):
                        try:
                            duration = float(line.split(":")[1].strip().split()[0])
                        except:
                            duration = 1.0
                    elif line.startswith("Cost:"):
                        try:
                            cost = float(line.split(":")[1].strip().split()[0])
                        except:
                            cost = 1.0
                
                if description:
                    self.step_counter += 1
                    step = PlanStep(
                        step_id=f"step_{self.step_counter}",
                        description=description,
                        expected_duration=duration,
                        expected_cost=cost
                    )
                    self.steps[step.step_id] = step
                    steps.append(step)
            
            return steps
        
        def execute_step(
            self,
            step_id: str,
            actual_duration: Optional[float] = None,
            actual_cost: Optional[float] = None,
            success: bool = True
        ) -> bool:
            """
            Execute a plan step.
            
            Args:
                step_id: Step to execute
                actual_duration: Actual time taken
                actual_cost: Actual cost
                success: Whether execution succeeded
                
            Returns:
                Success status
            """
            if step_id not in self.steps:
                return False
            
            step = self.steps[step_id]
            
            # Start execution
            if step.status == StepStatus.PENDING:
                step.status = StepStatus.IN_PROGRESS
                step.started_at = datetime.now()
            
            # Complete execution
            if step.status == StepStatus.IN_PROGRESS:
                if success:
                    step.status = StepStatus.COMPLETED
                    step.completed_at = datetime.now()
                    step.actual_duration = actual_duration or step.expected_duration
                    step.actual_cost = actual_cost or step.expected_cost
                else:
                    step.status = StepStatus.FAILED
            
            return success
        
        def monitor_execution(self) -> List[Deviation]:
            """
            Monitor plan execution and detect deviations.
            
            Returns:
                List of new deviations
            """
            new_deviations = []
            
            for step in self.steps.values():
                if step.status == StepStatus.COMPLETED:
                    # Check for delays
                    if step.actual_duration and step.actual_duration > step.expected_duration * 1.2:
                        self.deviation_counter += 1
                        deviation = Deviation(
                            deviation_id=f"dev_{self.deviation_counter}",
                            deviation_type=DeviationType.DELAY,
                            affected_step=step.step_id,
                            severity=min(10.0, (step.actual_duration / step.expected_duration - 1) * 10),
                            description=f"Step took {step.actual_duration:.1f}h vs expected {step.expected_duration:.1f}h"
                        )
                        self.deviations.append(deviation)
                        new_deviations.append(deviation)
                    
                    # Check for cost overruns
                    if step.actual_cost and step.actual_cost > step.expected_cost * 1.2:
                        self.deviation_counter += 1
                        deviation = Deviation(
                            deviation_id=f"dev_{self.deviation_counter}",
                            deviation_type=DeviationType.COST_OVERRUN,
                            affected_step=step.step_id,
                            severity=min(10.0, (step.actual_cost / step.expected_cost - 1) * 10),
                            description=f"Step cost {step.actual_cost:.1f} vs expected {step.expected_cost:.1f}"
                        )
                        self.deviations.append(deviation)
                        new_deviations.append(deviation)
                
                elif step.status == StepStatus.FAILED:
                    # Failure is high severity
                    self.deviation_counter += 1
                    deviation = Deviation(
                        deviation_id=f"dev_{self.deviation_counter}",
                        deviation_type=DeviationType.DEPENDENCY_FAILURE,
                        affected_step=step.step_id,
                        severity=9.0,
                        description=f"Step {step.step_id} failed"
                    )
                    self.deviations.append(deviation)
                    new_deviations.append(deviation)
            
            return new_deviations
        
        def diagnose_deviation(self, deviation: Deviation) -> str:
            """
            Diagnose root cause of deviation using LLM.
            
            Args:
                deviation: Deviation to diagnose
                
            Returns:
                Diagnosis text
            """
            step = self.steps[deviation.affected_step]
            
            prompt = ChatPromptTemplate.from_template(
                "Analyze this plan deviation and suggest root causes:\n\n"
                "Step: {step_description}\n"
                "Deviation: {deviation_type}\n"
                "Details: {deviation_description}\n"
                "Severity: {severity}/10\n\n"
                "Provide:\n"
                "1. Likely root cause\n"
                "2. Impact on remaining plan\n"
                "3. Recommended action (continue/repair/replan)\n"
            )
            
            chain = prompt | self.llm | StrOutputParser()
            diagnosis = chain.invoke({
                "step_description": step.description,
                "deviation_type": deviation.deviation_type.value,
                "deviation_description": deviation.description,
                "severity": deviation.severity
            })
            
            return diagnosis
        
        def decide_adaptation(self, deviation: Deviation) -> AdaptationAction:
            """
            Decide what adaptation action to take.
            
            Args:
                deviation: Deviation triggering adaptation
                
            Returns:
                Recommended action
            """
            # Simple severity-based decision
            if deviation.severity < 3.0:
                return AdaptationAction.CONTINUE
            elif deviation.severity < 5.0:
                return AdaptationAction.REPAIR
            elif deviation.severity  Adaptation:
            """
            Repair plan by adjusting expectations.
            
            Args:
                deviation: Deviation to repair
                
            Returns:
                Adaptation record
            """
            step = self.steps[deviation.affected_step]
            
            # Adjust expectations for remaining steps
            remaining_steps = [
                s for s in self.steps.values()
                if s.status == StepStatus.PENDING
            ]
            
            if deviation.deviation_type == DeviationType.DELAY:
                # Increase duration estimates
                for s in remaining_steps:
                    s.expected_duration *= 1.2
                changes = f"Increased duration estimates by 20% for {len(remaining_steps)} remaining steps"
            
            elif deviation.deviation_type == DeviationType.COST_OVERRUN:
                # Increase cost estimates
                for s in remaining_steps:
                    s.expected_cost *= 1.2
                changes = f"Increased cost estimates by 20% for {len(remaining_steps)} remaining steps"
            
            else:
                changes = "No specific repair applied"
            
            self.adaptation_counter += 1
            adaptation = Adaptation(
                adaptation_id=f"adapt_{self.adaptation_counter}",
                action=AdaptationAction.REPAIR,
                reason=deviation.description,
                changes=changes,
                success=True
            )
            self.adaptations.append(adaptation)
            
            return adaptation
        
        def replan_from_current(self, goal: str) -> List[PlanStep]:
            """
            Create new plan from current state.
            
            Args:
                goal: Original goal
                
            Returns:
                New plan steps
            """
            # Preserve completed steps
            completed = [s for s in self.steps.values() if s.status == StepStatus.COMPLETED]
            
            # Create new steps for remaining work
            prompt = ChatPromptTemplate.from_template(
                "The original plan encountered issues. Create a revised plan.\n\n"
                "Goal: {goal}\n"
                "Completed steps: {completed_count}\n"
                "Previous issues: {issues}\n\n"
                "Create 3-5 steps to complete the remaining work.\n"
                "Consider lessons learned from previous issues.\n\n"
                "Format as:\n"
                "Step N: [description]\n"
                "Duration: [hours]\n"
                "Cost: [units]\n"
                "---\n"
            )
            
            issues = ", ".join([d.description for d in self.deviations[-3:]])
            
            chain = prompt | self.llm | StrOutputParser()
            result = chain.invoke({
                "goal": goal,
                "completed_count": len(completed),
                "issues": issues or "None"
            })
            
            # Parse new steps
            new_steps = []
            step_blocks = result.split("---")
            
            for block in step_blocks:
                if not block.strip():
                    continue
                
                lines = [line.strip() for line in block.strip().split('\n')]
                description = ""
                duration = 1.0
                cost = 1.0
                
                for line in lines:
                    if line.startswith("Step"):
                        description = line.split(":", 1)[1].strip() if ":" in line else line
                    elif line.startswith("Duration:"):
                        try:
                            duration = float(line.split(":")[1].strip().split()[0])
                        except:
                            duration = 1.0
                    elif line.startswith("Cost:"):
                        try:
                            cost = float(line.split(":")[1].strip().split()[0])
                        except:
                            cost = 1.0
                
                if description:
                    self.step_counter += 1
                    step = PlanStep(
                        step_id=f"step_{self.step_counter}",
                        description=description,
                        expected_duration=duration,
                        expected_cost=cost
                    )
                    self.steps[step.step_id] = step
                    new_steps.append(step)
            
            # Record adaptation
            self.adaptation_counter += 1
            adaptation = Adaptation(
                adaptation_id=f"adapt_{self.adaptation_counter}",
                action=AdaptationAction.REPLAN,
                reason="Significant deviations detected",
                changes=f"Created {len(new_steps)} new steps",
                success=True
            )
            self.adaptations.append(adaptation)
            
            return new_steps
        
        def adapt(self, goal: str) -> Optional[Adaptation]:
            """
            Monitor and adapt plan as needed.
            
            Args:
                goal: Original goal
                
            Returns:
                Adaptation if made, None otherwise
            """
            # Monitor for deviations
            new_deviations = self.monitor_execution()
            
            if not new_deviations:
                return None
            
            # Check most severe deviation
            most_severe = max(new_deviations, key=lambda d: d.severity)
            
            # Decide action
            action = self.decide_adaptation(most_severe)
            
            # Execute adaptation
            if action == AdaptationAction.CONTINUE:
                return None
            elif action == AdaptationAction.REPAIR:
                return self.repair_plan(most_severe)
            elif action == AdaptationAction.REPLAN:
                self.replan_from_current(goal)
                return self.adaptations[-1]
            
            return None
        
        def get_progress_summary(self) -> Dict[str, Any]:
            """Get plan progress summary."""
            total = len(self.steps)
            completed = sum(1 for s in self.steps.values() if s.status == StepStatus.COMPLETED)
            failed = sum(1 for s in self.steps.values() if s.status == StepStatus.FAILED)
            in_progress = sum(1 for s in self.steps.values() if s.status == StepStatus.IN_PROGRESS)
            
            total_expected_duration = sum(s.expected_duration for s in self.steps.values())
            total_actual_duration = sum(
                s.actual_duration for s in self.steps.values()
                if s.actual_duration is not None
            )
            
            return {
                "total_steps": total,
                "completed": completed,
                "failed": failed,
                "in_progress": in_progress,
                "pending": total - completed - failed - in_progress,
                "completion_rate": completed / total if total > 0 else 0,
                "total_deviations": len(self.deviations),
                "total_adaptations": len(self.adaptations),
                "expected_duration": total_expected_duration,
                "actual_duration": total_actual_duration
            }
    
    
    def demonstrate_adaptive_planning():
        """Demonstrate adaptive planning pattern."""
        
        print("=" * 80)
        print("ADAPTIVE PLANNING PATTERN DEMONSTRATION")
        print("=" * 80)
        
        # Example 1: Basic adaptive planning
        print("\n" + "=" * 80)
        print("Example 1: Plan Creation and Execution")
        print("=" * 80)
        
        planner = AdaptivePlanner()
        
        print("\nCreating initial plan...")
        steps = planner.create_plan("Develop a mobile app", num_steps=5)
        
        print(f"\nCreated {len(steps)} steps:")
        for step in steps:
            print(f"  {step.step_id}: {step.description}")
            print(f"    Expected: {step.expected_duration}h, ${step.expected_cost}")
        
        print("\nExecuting steps...")
        for i, step in enumerate(steps[:2]):
            planner.execute_step(step.step_id)
            print(f"  ✓ Completed: {step.description}")
        
        progress = planner.get_progress_summary()
        print(f"\nProgress: {progress['completed']}/{progress['total_steps']} steps completed")
        
        # Example 2: Deviation detection
        print("\n" + "=" * 80)
        print("Example 2: Deviation Detection")
        print("=" * 80)
        
        planner2 = AdaptivePlanner()
        planner2.create_plan("Build website", num_steps=4)
        
        # Execute with delays
        steps2 = list(planner2.steps.values())
        planner2.execute_step(steps2[0].step_id, actual_duration=3.0, actual_cost=1.0)
        steps2[0].expected_duration = 2.0  # Set expected for comparison
        
        print("\nExecuted step with delay:")
        print(f"  Expected: {steps2[0].expected_duration}h")
        print(f"  Actual: {steps2[0].actual_duration}h")
        print(f"  Delay: {steps2[0].actual_duration - steps2[0].expected_duration:.1f}h")
        
        print("\nMonitoring for deviations...")
        deviations = planner2.monitor_execution()
        
        print(f"\nDetected {len(deviations)} deviation(s):")
        for dev in deviations:
            print(f"  • Type: {dev.deviation_type.value}")
            print(f"    Severity: {dev.severity:.1f}/10")
            print(f"    Description: {dev.description}")
        
        # Example 3: Root cause diagnosis
        print("\n" + "=" * 80)
        print("Example 3: Root Cause Diagnosis")
        print("=" * 80)
        
        if deviations:
            dev = deviations[0]
            print(f"\nDiagnosing deviation: {dev.description}")
            print("\nLLM Analysis:")
            diagnosis = planner2.diagnose_deviation(dev)
            print(diagnosis)
        
        # Example 4: Plan repair
        print("\n" + "=" * 80)
        print("Example 4: Plan Repair")
        print("=" * 80)
        
        planner3 = AdaptivePlanner()
        planner3.create_plan("Complete project", num_steps=5)
        
        # Simulate deviation
        steps3 = list(planner3.steps.values())
        planner3.execute_step(steps3[0].step_id, actual_duration=5.0)
        steps3[0].expected_duration = 2.0
        
        deviations3 = planner3.monitor_execution()
        
        if deviations3:
            print("\nDeviation detected:")
            print(f"  Severity: {deviations3[0].severity:.1f}/10")
            
            print("\nDeciding adaptation action...")
            action = planner3.decide_adaptation(deviations3[0])
            print(f"  Recommended: {action.value}")
            
            print("\nRepairing plan...")
            adaptation = planner3.repair_plan(deviations3[0])
            print(f"  Changes: {adaptation.changes}")
        
        # Example 5: Full replanning
        print("\n" + "=" * 80)
        print("Example 5: Full Replanning")
        print("=" * 80)
        
        planner4 = AdaptivePlanner()
        original_steps = planner4.create_plan("Launch product", num_steps=4)
        
        print(f"\nOriginal plan: {len(original_steps)} steps")
        
        # Simulate major failure
        steps4 = list(planner4.steps.values())
        planner4.execute_step(steps4[0].step_id)
        planner4.execute_step(steps4[1].step_id, success=False)
        steps4[1].status = StepStatus.FAILED
        
        print("\nMajor failure detected!")
        deviations4 = planner4.monitor_execution()
        
        if deviations4:
            severe_dev = max(deviations4, key=lambda d: d.severity)
            print(f"  Severity: {severe_dev.severity:.1f}/10")
            
            action = planner4.decide_adaptation(severe_dev)
            print(f"  Action: {action.value}")
            
            if action == AdaptationAction.REPLAN:
                print("\nReplanning from current state...")
                new_steps = planner4.replan_from_current("Launch product")
                print(f"  Created {len(new_steps)} new steps")
        
        # Example 6: Adaptation decision tree
        print("\n" + "=" * 80)
        print("Example 6: Adaptation Decision Making")
        print("=" * 80)
        
        planner5 = AdaptivePlanner()
        
        # Test different severity levels
        test_cases = [
            (2.0, "Minor delay"),
            (4.5, "Moderate cost overrun"),
            (7.0, "Significant quality issue"),
            (9.5, "Critical failure")
        ]
        
        print("\nAdaptation decisions by severity:")
        for severity, description in test_cases:
            planner5.step_counter += 1
            test_step = PlanStep(
                step_id=f"step_{planner5.step_counter}",
                description="Test step"
            )
            planner5.steps[test_step.step_id] = test_step
            
            planner5.deviation_counter += 1
            test_dev = Deviation(
                deviation_id=f"dev_{planner5.deviation_counter}",
                deviation_type=DeviationType.DELAY,
                affected_step=test_step.step_id,
                severity=severity,
                description=description
            )
            
            action = planner5.decide_adaptation(test_dev)
            print(f"  Severity {severity:.1f} ({description}) → {action.value}")
        
        # Example 7: Automatic adaptation loop
        print("\n" + "=" * 80)
        print("Example 7: Automatic Adaptation Loop")
        print("=" * 80)
        
        planner6 = AdaptivePlanner()
        planner6.create_plan("Deploy service", num_steps=4)
        
        print("\nExecuting with automatic adaptation...")
        
        steps6 = list(planner6.steps.values())
        
        # Step 1: Normal
        planner6.execute_step(steps6[0].step_id, actual_duration=1.0)
        adaptation = planner6.adapt("Deploy service")
        print(f"  Step 1: Completed normally - {adaptation or 'No adaptation needed'}")
        
        # Step 2: Minor delay
        planner6.execute_step(steps6[1].step_id, actual_duration=3.0)
        steps6[1].expected_duration = 2.0
        adaptation = planner6.adapt("Deploy service")
        print(f"  Step 2: Minor delay - {adaptation.action.value if adaptation else 'Continue'}")
        
        # Step 3: Major issue
        planner6.execute_step(steps6[2].step_id, actual_duration=10.0)
        steps6[2].expected_duration = 2.0
        adaptation = planner6.adapt("Deploy service")
        print(f"  Step 3: Major issue - {adaptation.action.value if adaptation else 'Continue'}")
        
        print(f"\nTotal adaptations: {len(planner6.adaptations)}")
        
        # Example 8: Learning from experience
        print("\n" + "=" * 80)
        print("Example 8: Progress Tracking and Summary")
        print("=" * 80)
        
        planner7 = AdaptivePlanner()
        planner7.create_plan("Complete migration", num_steps=6)
        
        # Execute with various outcomes
        steps7 = list(planner7.steps.values())
        planner7.execute_step(steps7[0].step_id, actual_duration=1.5, actual_cost=1.0)
        steps7[0].expected_duration = 1.0
        planner7.execute_step(steps7[1].step_id, actual_duration=3.0, actual_cost=2.0)
        steps7[1].expected_duration = 2.0
        steps7[1].expected_cost = 1.5
        planner7.execute_step(steps7[2].step_id, actual_duration=1.0, actual_cost=1.0)
        
        # Monitor and adapt
        planner7.adapt("Complete migration")
        
        print("\nFINAL SUMMARY:")
        print("=" * 60)
        
        progress = planner7.get_progress_summary()
        for key, value in progress.items():
            if isinstance(value, float):
                print(f"  {key}: {value:.2f}")
            else:
                print(f"  {key}: {value}")
        
        print("\nDeviations:")
        for dev in planner7.deviations:
            print(f"  • {dev.deviation_type.value} (severity: {dev.severity:.1f})")
        
        print("\nAdaptations:")
        for adapt in planner7.adaptations:
            print(f"  • {adapt.action.value}: {adapt.changes}")
        
        # Summary
        print("\n" + "=" * 80)
        print("SUMMARY: Adaptive Planning Pattern")
        print("=" * 80)
        
        summary = """
        The Adaptive Planning pattern demonstrated:
        
        1. PLAN CREATION (Example 1):
           - LLM-based plan generation
           - Step-by-step structure
           - Duration and cost estimation
           - Progress tracking
           - Execution monitoring
        
        2. DEVIATION DETECTION (Example 2):
           - Automatic monitoring
           - Delay detection (20% threshold)
           - Cost overrun detection
           - Severity scoring
           - Real-time alerts
        
        3. ROOT CAUSE DIAGNOSIS (Example 3):
           - LLM-based analysis
           - Context-aware diagnosis
           - Impact assessment
           - Actionable recommendations
           - Natural language explanations
        
        4. PLAN REPAIR (Example 4):
           - Local plan adjustments
           - Expectation updates
           - Propagation to remaining steps
           - Minimal disruption
           - Quick fixes
        
        5. FULL REPLANNING (Example 5):
           - New plan generation
           - Progress preservation
           - Lesson incorporation
           - Context-aware replanning
           - Fresh start when needed
        
        6. DECISION MAKING (Example 6):
           - Severity-based decisions
           - Action selection (continue/repair/revise/replan)
           - Graduated response
           - Threshold-based triggers
           - Balanced approach
        
        7. AUTOMATIC ADAPTATION (Example 7):
           - Continuous monitoring
           - Automatic response
           - Multi-step adaptation
           - Event-driven updates
           - Hands-free operation
        
        8. PROGRESS TRACKING (Example 8):
           - Comprehensive metrics
           - Deviation history
           - Adaptation log
           - Performance summary
           - Complete audit trail
        
        KEY BENEFITS:
        ✓ Robust to uncertainty and changes
        ✓ Maintains progress despite issues
        ✓ Exploits new opportunities
        ✓ Reduces plan failure rate
        ✓ Learns from experience
        ✓ Handles unexpected events
        ✓ Flexible goal achievement
        ✓ Automatic problem response
        
        USE CASES:
        • Robot navigation (dynamic environments)
        • Project management (changing requirements)
        • Manufacturing (equipment failures)
        • Supply chain (disruptions)
        • Military operations (uncertain terrain)
        • Autonomous vehicles (route planning)
        • Cloud resource allocation
        • Emergency response coordination
        
        ADAPTATION STRATEGIES:
        → Plan Repair: Minor modifications
        → Plan Reuse: Apply similar past plan
        → Plan Revision: Significant restructuring
        → Replanning: Create entirely new plan
        → Reactive: Immediate adjustments
        → Proactive: Anticipate changes
        
        MONITORING ASPECTS:
        • Progress: Completion vs. expected
        • Resources: Availability and consumption
        • Time: Schedule adherence
        • Quality: Output measures
        • Cost: Budget tracking
        • Risks: Emerging threats
        
        BEST PRACTICES:
        1. Implement efficient monitoring
        2. Use event-driven detection
        3. Cache replan results
        4. Set replanning thresholds
        5. Limit replanning frequency
        6. Preserve partial progress
        7. Log all adaptations
        8. Enable replay for debugging
        
        TRADE-OFFS:
        • Monitoring overhead vs. responsiveness
        • Plan stability vs. adaptability
        • Replanning cost vs. plan quality
        • Automation vs. human control
        
        PRODUCTION CONSIDERATIONS:
        → Monitor continuously without blocking
        → Use asynchronous event processing
        → Implement replanning rate limits
        → Cache similar plan patterns
        → Log all deviations and adaptations
        → Support manual override
        → Enable plan playback
        → Track adaptation success rate
        → Alert on excessive replanning
        → Preserve completed work
        
        This pattern enables plans to evolve dynamically in response to
        execution feedback and environmental changes, maintaining progress
        toward goals despite uncertainty and unexpected events.
        """
        
        print(summary)
    
    
    if __name__ == "__main__":
        demonstrate_adaptive_planning()
    
        

    Agentic Design Pattern: Adaptive Planning

    Pattern 109: Adaptive Planning – Line-by-Line Explanation

    Overview

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


    Pattern Description

    Pattern 109: Adaptive Planning

    Description: The Adaptive Planning pattern enables agents to dynamically adjust their plans in response to execution feedback, environmental changes, and unexpected events. Rather than rigidly following a predetermined plan, adaptive planners monitor execution, detect deviations from expected outcomes, diagnose problems, and modify plans on the fly to maintain progress toward goals.

    Adaptive planning is essential in dynamic, uncertain environments where initial
    assumptions may become invalid, resources may change, or unexpected opportunities
    and obstacles may arise. The pattern combines continuous monitoring, deviation
    detection, root cause analysis, and replanning strategies to ensure resilience
    and goal achievement despite changing conditions.
    
    This pattern includes mechanisms for execution monitoring, deviation detection,
    impact assessment, root cause diagnosis, plan repair strategies, replanning
    triggers, and learning from outcomes to improve future plans.
    

    Key Components: 1. Execution Monitor: Tracks plan execution 2. Deviation Detector: Identifies plan-reality gaps 3. Impact Assessor: Evaluates deviation significance 4. Root Cause Analyzer: Diagnoses problems 5. Plan Repairer: Fixes plans locally 6. Replanner: Creates new plans when needed 7. Learning Module: Improves from experience

    Adaptation Strategies: 1. Plan Repair: Minor modifications to existing plan 2. Plan Reuse: Apply similar past plan 3. Plan Revision: Significant restructuring 4. Replanning: Create entirely new plan 5. Reactive Adaptation: Immediate adjustments 6. Proactive Adaptation: Anticipate changes 7. Learning-based: Use past outcomes

    Monitoring Aspects: 1. Progress: Task completion vs. expected 2. Resources: Availability and consumption 3. Time: Schedule adherence 4. Quality: Output quality measures 5. Cost: Budget tracking 6. Risks: Emerging threats 7. Opportunities: New possibilities

    Use Cases: – Robot navigation in dynamic environments – Project management with changing requirements – Manufacturing with equipment failures – Supply chain disruptions – Military operations in uncertain terrain – Autonomous vehicle route planning – Resource allocation in cloud systems

    Advantages: – Robust to uncertainty – Maintains progress despite changes – Exploits new opportunities – Reduces plan failures – Learns from experience – Handles unexpected events – Flexible goal achievement

    Challenges: – Monitoring overhead – Replanning latency – Plan stability vs. adaptability – Knowing when to replan – Avoiding thrashing (excessive replanning) – Preserving progress – Computational cost

    LangChain Implementation: This implementation uses LangChain for: – LLM-based deviation analysis – Root cause diagnosis – Plan repair suggestions – Natural language explanations

    Production Considerations: – Implement efficient monitoring – Use event-driven detection – Cache replan results – Set replanning thresholds – Limit replanning frequency – Preserve partial progress – Log all adaptations – Enable replay for debugging


    File Statistics

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

    Module Documentation

    """
    Pattern 109: Adaptive Planning
    
    Description:
        The Adaptive Planning pattern enables agents to dynamically adjust their plans
        in response to execution feedback, environmental changes, and unexpected events.
        Rather than rigidly following a predetermined plan, adaptive planners monitor
        execution, detect deviations from expected outcomes, diagnose problems, and
        modify plans on the fly to maintain progress toward goals.
    
        Adaptive planning is essential in dynamic, uncertain environments where initial
        assumptions may become invalid, resources may change, or unexpected opportunities
        and obstacles may arise. The pattern combines continuous monitoring, deviation
        detection, root cause analysis, and replanning strategies to ensure resilience
        and goal achievement despite changing conditions.
    
        This pattern includes mechanisms for execution monitoring, deviation detection,
        impact assessment, root cause diagnosis, plan repair strategies, replanning
        triggers, and learning from outcomes to improve future plans.
    
    Key Components:
        1. Execution Monitor: Tracks plan execution
        2. Deviation Detector: Identifies plan-reality gaps
        3. Impact Assessor: Evaluates deviation significance
        4. Root Cause Analyzer: Diagnoses problems
        5. Plan Repairer: Fixes plans locally
        6. Replanner: Creates new plans when needed
        7. Learning Module: Improves from experience
    
    Adaptation Strategies:
        1. Plan Repair: Minor modifications to existing plan
        2. Plan Reuse: Apply similar past plan
        3. Plan Revision: Significant restructuring
        4. Replanning: Create entirely new plan
        5. Reactive Adaptation: Immediate adjustments
        6. Proactive Adaptation: Anticipate changes
        7. Learning-based: Use past outcomes
    
    Monitoring Aspects:
        1. Progress: Task completion vs. expected
        2. Resources: Availability and consumption
        3. Time: Schedule adherence
        4. Quality: Output quality measures
        5. Cost: Budget tracking
        6. Risks: Emerging threats
        7. Opportunities: New possibilities
    
    Use Cases:
        - Robot navigation in dynamic environments
        - Project management with changing requirements
        - Manufacturing with equipment failures
        - Supply chain disruptions
        - Military operations in uncertain terrain
        - Autonomous vehicle route planning
        - Resource allocation in cloud systems
    
    Advantages:
        - Robust to uncertainty
        - Maintains progress despite changes
        - Exploits new opportunities
        - Reduces plan failures
        - Learns from experience
        - Handles unexpected events
        - Flexible goal achievement
    
    Challenges:
        - Monitoring overhead
        - Replanning latency
        - Plan stability vs. adaptability
        - Knowing when to replan
        - Avoiding thrashing (excessive replanning)
        - Preserving progress
        - Computational cost
    
    LangChain Implementation:
        This implementation uses LangChain for:
        - LLM-based deviation analysis
        - Root cause diagnosis
        - Plan repair suggestions
        - Natural language explanations
    
    Production Considerations:
        - Implement efficient monitoring
        - Use event-driven detection
        - Cache replan results
        - Set replanning thresholds
        - Limit replanning frequency
        - Preserve partial progress
        - Log all adaptations
        - Enable replay for debugging
    """
    

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


    Import Statements

    Standard Library Imports

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

    Lines 107-113

    Inherits from: Enum

    Purpose:

    Status of a plan step.
    

    Class: DeviationType

    Lines 116-123

    Inherits from: Enum

    Purpose:

    Types of plan deviations.
    

    Class: AdaptationAction

    Lines 126-132

    Inherits from: Enum

    Purpose:

    Types of adaptation actions.
    

    Class: PlanStep

    Lines 136-161

    Purpose:

    A step in an adaptive plan.
    
    Attributes:
        step_id: Unique identifier
        description: Step description
        expected_duration: Expected time to complete
        expected_cost: Expected cost
        dependencies: Prerequisite steps
        status: Current status
        started_at: When started
        completed_at: When completed
        actual_duration: Actual time taken
        actual_cost: Actual cost
    

    Class: Deviation

    Lines 165-184

    Purpose:

    Detected deviation from plan.
    
    Attributes:
        deviation_id: Unique identifier
        deviation_type: Type of deviation
        affected_step: Step with deviation
        severity: Severity score (0-10)
        description: Human-readable description
        detected_at: When detected
        resolved: Whether resolved
    

    Class: Adaptation

    Lines 188-205

    Purpose:

    Adaptation made to plan.
    
    Attributes:
        adaptation_id: Unique identifier
        action: Type of adaptation
        reason: Why adaptation was made
        changes: Description of changes
        timestamp: When adaptation occurred
        success: Whether adaptation succeeded
    

    Class: AdaptivePlanner

    Lines 208-637

    Purpose:

    Adaptive planning system with execution monitoring and replanning.
    
    This planner monitors execution, detects deviations, and adapts
    plans dynamically to maintain progress toward goals.
    

    Methods:

    __init__(self, temperature)

    Lines 216-230

    Purpose:

    Initialize adaptive planner.
    
    Args:
        temperature: LLM temperature
    

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

    create_plan(self, goal, num_steps)

    Lines 232-298

    Purpose:

    Create initial plan using LLM.
    
    Args:
        goal: Goal description
        num_steps: Number of steps
    
    Returns:
        List of plan steps
    

    Parameters: – goal: Goal – num_steps: Num steps

    Returns: <ast.Subscript object at 0x000001A06D6F5050>

    execute_step(self, step_id, actual_duration, actual_cost, success)

    Lines 300-339

    Purpose:

    Execute a plan step.
    
    Args:
        step_id: Step to execute
        actual_duration: Actual time taken
        actual_cost: Actual cost
        success: Whether execution succeeded
    
    Returns:
        Success status
    

    Parameters: – step_id: Step id – actual_duration: Actual duration – actual_cost: Actual cost – success: Success

    Returns: bool

    monitor_execution(self)

    Lines 341-391

    Purpose:

    Monitor plan execution and detect deviations.
    
    Returns:
        List of new deviations
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D72FC50>

    diagnose_deviation(self, deviation)

    Lines 393-425

    Purpose:

    Diagnose root cause of deviation using LLM.
    
    Args:
        deviation: Deviation to diagnose
    
    Returns:
        Diagnosis text
    

    Parameters: – deviation: Deviation

    Returns: str

    decide_adaptation(self, deviation)

    Lines 427-445

    Purpose:

    Decide what adaptation action to take.
    
    Args:
        deviation: Deviation triggering adaptation
    
    Returns:
        Recommended action
    

    Parameters: – deviation: Deviation

    Returns: AdaptationAction

    repair_plan(self, deviation)

    Lines 447-490

    Purpose:

    Repair plan by adjusting expectations.
    
    Args:
        deviation: Deviation to repair
    
    Returns:
        Adaptation record
    

    Parameters: – deviation: Deviation

    Returns: Adaptation

    replan_from_current(self, goal)

    Lines 492-578

    Purpose:

    Create new plan from current state.
    
    Args:
        goal: Original goal
    
    Returns:
        New plan steps
    

    Parameters: – goal: Goal

    Returns: <ast.Subscript object at 0x000001A06D686710>

    adapt(self, goal)

    Lines 580-611

    Purpose:

    Monitor and adapt plan as needed.
    
    Args:
        goal: Original goal
    
    Returns:
        Adaptation if made, None otherwise
    

    Parameters: – goal: Goal

    Returns: <ast.Subscript object at 0x000001A06E0951D0>

    get_progress_summary(self)

    Lines 613-637

    Purpose:

    Get plan progress summary.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D812450>


    Functions

    demonstrate_adaptive_planning()

    Lines 640-1005

    Purpose:

    Demonstrate adaptive planning pattern.
    

    Key Concepts Summary

    Pattern Overview

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

    Core Components

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

    When to Use

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

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


    Best Practices Demonstrated

    1. Environment Configuration

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

    2. Error Handling

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

    3. Prompt Engineering

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

    4. Code Organization

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

    5. Performance Optimization

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

    LangChain Components Used

    • ChatOpenAI: OpenAI model interface
    • Chains: Sequential component composition

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


    Execution Flow

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

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


    Performance Considerations

    Computational Cost

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

    Optimization Strategies

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

    Cost Estimates (Approximate)

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

    Actual costs vary based on input/output token counts


    Common Pitfalls to Avoid

    1. Prompt Design Issues

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

    2. Error Handling

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

    3. Token Management

    ❌ Exceeding model token limits
    ✅ Truncate inputs, chunk processing

    4. Security

    ❌ Exposing API keys in code
    ✅ Use environment variables

    5. Testing

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

    6. Cost Management

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


    Further Improvements

    Production Enhancements

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

    Feature Extensions

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

    Code Quality

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

    Related Patterns

    • Pattern 108
    • Pattern 110

    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 Adaptive Planning pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

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

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

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


    File: 109_adaptive_planning.py
    Total Lines: 1010
    Pattern Category: AI Agent Design Pattern #109


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Collaborative Planning

    
        """
    Pattern 108: Collaborative Planning
    
    Description:
        The Collaborative Planning pattern enables multiple agents to work together
        in creating and executing plans toward shared goals. Rather than planning in
        isolation, agents coordinate their planning efforts, share information, divide
        tasks, resolve conflicts, and synchronize their actions to achieve objectives
        that would be difficult or impossible for individual agents.
        
        Collaborative planning is essential for multi-agent systems where agents have
        complementary capabilities, shared resources, or interdependent goals. The
        pattern addresses challenges like distributed knowledge, conflicting preferences,
        resource contention, and coordination overhead.
        
        This pattern includes mechanisms for goal sharing, task decomposition and
        allocation, plan merging, conflict resolution, resource negotiation, and
        synchronized execution. Agents may use various collaboration strategies from
        centralized coordination to fully distributed consensus-based planning.
    
    Key Components:
        1. Shared Goal Repository: Common objectives
        2. Task Allocator: Assigns tasks to agents
        3. Plan Merger: Combines individual plans
        4. Conflict Resolver: Handles plan conflicts
        5. Resource Negotiator: Allocates shared resources
        6. Synchronization Manager: Coordinates timing
        7. Communication Protocol: Agent interaction
    
    Collaboration Strategies:
        1. Centralized: Single coordinator assigns tasks
        2. Hierarchical: Tree of coordinators
        3. Distributed: Peer-to-peer negotiation
        4. Auction-based: Agents bid for tasks
        5. Consensus: Democratic decision making
        6. Blackboard: Shared workspace
    
    Planning Phases:
        1. Goal Identification: Define shared objectives
        2. Capability Sharing: Agents advertise skills
        3. Task Decomposition: Break into subtasks
        4. Task Allocation: Assign to appropriate agents
        5. Plan Integration: Merge individual plans
        6. Conflict Resolution: Resolve incompatibilities
        7. Synchronized Execution: Coordinate timing
    
    Use Cases:
        - Multi-robot coordination
        - Distributed project management
        - Supply chain planning
        - Military operations planning
        - Smart city management
        - Collaborative problem solving
        - Team-based game AI
    
    Advantages:
        - Leverages diverse capabilities
        - Handles complex tasks beyond single agent
        - Load balancing across agents
        - Fault tolerance through redundancy
        - Parallel execution speedup
        - Shared knowledge utilization
        - Flexible task distribution
    
    Challenges:
        - Communication overhead
        - Conflict resolution complexity
        - Synchronization requirements
        - Trust and verification
        - Resource contention
        - Plan consistency maintenance
        - Scalability with agent count
    
    LangChain Implementation:
        This implementation uses LangChain for:
        - LLM-based task decomposition
        - Agent capability matching
        - Plan integration strategies
        - Natural language coordination
        
    Production Considerations:
        - Implement efficient communication protocols
        - Use asynchronous messaging
        - Handle agent failures gracefully
        - Monitor collaboration overhead
        - Implement timeout mechanisms
        - Log all coordination decisions
        - Support dynamic agent addition/removal
        - Enable human oversight
    """
    
    import os
    from typing import List, Dict, Any, Optional, Set
    from dataclasses import dataclass, field
    from datetime import datetime
    from enum import Enum
    from dotenv import load_dotenv
    
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser
    from langchain_openai import ChatOpenAI
    
    load_dotenv()
    
    
    class TaskStatus(Enum):
        """Status of a task."""
        UNASSIGNED = "unassigned"
        ASSIGNED = "assigned"
        IN_PROGRESS = "in_progress"
        COMPLETED = "completed"
        FAILED = "failed"
    
    
    class MessageType(Enum):
        """Types of coordination messages."""
        TASK_OFFER = "task_offer"
        TASK_BID = "task_bid"
        TASK_ASSIGNMENT = "task_assignment"
        RESOURCE_REQUEST = "resource_request"
        RESOURCE_GRANT = "resource_grant"
        STATUS_UPDATE = "status_update"
        PLAN_PROPOSAL = "plan_proposal"
    
    
    @dataclass
    class Task:
        """
        Represents a task in collaborative plan.
        
        Attributes:
            task_id: Unique identifier
            description: Task description
            required_capabilities: Capabilities needed
            estimated_effort: Effort estimate
            dependencies: Tasks that must complete first
            assigned_to: Agent ID assigned to
            status: Current status
            priority: Task priority
        """
        task_id: str
        description: str
        required_capabilities: List[str] = field(default_factory=list)
        estimated_effort: float = 1.0
        dependencies: List[str] = field(default_factory=list)
        assigned_to: Optional[str] = None
        status: TaskStatus = TaskStatus.UNASSIGNED
        priority: int = 5
        started_at: Optional[datetime] = None
        completed_at: Optional[datetime] = None
    
    
    @dataclass
    class CollaborativeAgent:
        """
        Agent participating in collaborative planning.
        
        Attributes:
            agent_id: Unique identifier
            name: Agent name
            capabilities: List of capabilities
            capacity: Available work capacity
            current_load: Current workload
            assigned_tasks: Tasks assigned to this agent
        """
        agent_id: str
        name: str
        capabilities: List[str] = field(default_factory=list)
        capacity: float = 10.0
        current_load: float = 0.0
        assigned_tasks: List[str] = field(default_factory=list)
        
        def can_handle(self, task: Task) -> bool:
            """Check if agent can handle a task."""
            # Check capabilities
            has_capabilities = all(
                cap in self.capabilities for cap in task.required_capabilities
            )
            # Check capacity
            has_capacity = (self.current_load + task.estimated_effort)  float:
            """Get remaining capacity."""
            return self.capacity - self.current_load
    
    
    @dataclass
    class Message:
        """
        Communication message between agents.
        
        Attributes:
            sender_id: Sending agent ID
            receiver_id: Receiving agent ID (None for broadcast)
            message_type: Type of message
            content: Message payload
            timestamp: When sent
        """
        sender_id: str
        receiver_id: Optional[str]
        message_type: MessageType
        content: Dict[str, Any]
        timestamp: datetime = field(default_factory=datetime.now)
    
    
    class CollaborativePlanner:
        """
        Collaborative planning system for multi-agent coordination.
        
        This planner enables agents to work together in planning and
        executing tasks toward shared goals.
        """
        
        def __init__(self, temperature: float = 0.4):
            """
            Initialize collaborative planner.
            
            Args:
                temperature: LLM temperature
            """
            self.llm = ChatOpenAI(temperature=temperature, model="gpt-3.5-turbo")
            self.agents: Dict[str, CollaborativeAgent] = {}
            self.tasks: Dict[str, Task] = {}
            self.messages: List[Message] = []
            self.task_counter = 0
        
        def add_agent(
            self,
            agent_id: str,
            name: str,
            capabilities: List[str],
            capacity: float = 10.0
        ) -> CollaborativeAgent:
            """
            Add agent to the planning system.
            
            Args:
                agent_id: Agent identifier
                name: Agent name
                capabilities: Agent capabilities
                capacity: Work capacity
                
            Returns:
                Created agent
            """
            agent = CollaborativeAgent(
                agent_id=agent_id,
                name=name,
                capabilities=capabilities,
                capacity=capacity
            )
            self.agents[agent_id] = agent
            return agent
        
        def decompose_goal(self, goal: str, num_tasks: int = 5) -> List[Task]:
            """
            Decompose goal into tasks using LLM.
            
            Args:
                goal: Goal description
                num_tasks: Number of tasks to create
                
            Returns:
                List of tasks
            """
            prompt = ChatPromptTemplate.from_template(
                "Decompose this goal into {num_tasks} concrete tasks:\n\n"
                "Goal: {goal}\n\n"
                "For each task, provide:\n"
                "- Description\n"
                "- Required capabilities (e.g., analysis, coding, communication)\n"
                "- Estimated effort (1-5)\n\n"
                "Format as:\n"
                "Task N: [description]\n"
                "Capabilities: [cap1, cap2]\n"
                "Effort: [1-5]\n"
                "---\n"
            )
            
            chain = prompt | self.llm | StrOutputParser()
            result = chain.invoke({"goal": goal, "num_tasks": num_tasks})
            
            # Parse tasks
            tasks = []
            task_blocks = result.split("---")
            
            for block in task_blocks:
                if not block.strip():
                    continue
                
                lines = [line.strip() for line in block.strip().split('\n')]
                description = ""
                capabilities = []
                effort = 1.0
                
                for line in lines:
                    if line.startswith("Task"):
                        description = line.split(":", 1)[1].strip() if ":" in line else line
                    elif line.startswith("Capabilities:"):
                        caps_text = line.split(":", 1)[1].strip()
                        capabilities = [c.strip() for c in caps_text.split(",")]
                    elif line.startswith("Effort:"):
                        try:
                            effort = float(line.split(":")[1].strip())
                        except:
                            effort = 1.0
                
                if description:
                    self.task_counter += 1
                    task = Task(
                        task_id=f"task_{self.task_counter}",
                        description=description,
                        required_capabilities=capabilities,
                        estimated_effort=effort
                    )
                    self.tasks[task.task_id] = task
                    tasks.append(task)
            
            return tasks
        
        def allocate_tasks_greedy(self) -> Dict[str, List[Task]]:
            """
            Allocate tasks to agents using greedy algorithm.
            
            Returns:
                Allocation map: agent_id -> tasks
            """
            allocation: Dict[str, List[Task]] = {aid: [] for aid in self.agents.keys()}
            
            # Sort tasks by priority (descending)
            unassigned = [t for t in self.tasks.values() if t.status == TaskStatus.UNASSIGNED]
            unassigned.sort(key=lambda t: t.priority, reverse=True)
            
            for task in unassigned:
                # Find capable agents
                capable_agents = [
                    agent for agent in self.agents.values()
                    if agent.can_handle(task)
                ]
                
                if not capable_agents:
                    continue
                
                # Assign to agent with most available capacity
                capable_agents.sort(key=lambda a: a.get_available_capacity(), reverse=True)
                best_agent = capable_agents[0]
                
                # Assign task
                task.assigned_to = best_agent.agent_id
                task.status = TaskStatus.ASSIGNED
                best_agent.assigned_tasks.append(task.task_id)
                best_agent.current_load += task.estimated_effort
                allocation[best_agent.agent_id].append(task)
            
            return allocation
        
        def allocate_tasks_auction(self) -> Dict[str, List[Task]]:
            """
            Allocate tasks using auction-based mechanism.
            
            Returns:
                Allocation map: agent_id -> tasks
            """
            allocation: Dict[str, List[Task]] = {aid: [] for aid in self.agents.keys()}
            
            unassigned = [t for t in self.tasks.values() if t.status == TaskStatus.UNASSIGNED]
            
            for task in unassigned:
                # Agents bid for task
                bids: List[tuple] = []
                
                for agent in self.agents.values():
                    if agent.can_handle(task):
                        # Bid value: higher if more capacity, lower if already loaded
                        bid_value = agent.get_available_capacity() / agent.capacity
                        bids.append((agent, bid_value))
                
                if bids:
                    # Award to highest bidder
                    bids.sort(key=lambda x: x[1], reverse=True)
                    winner = bids[0][0]
                    
                    task.assigned_to = winner.agent_id
                    task.status = TaskStatus.ASSIGNED
                    winner.assigned_tasks.append(task.task_id)
                    winner.current_load += task.estimated_effort
                    allocation[winner.agent_id].append(task)
                    
                    # Send message
                    self.messages.append(Message(
                        sender_id="coordinator",
                        receiver_id=winner.agent_id,
                        message_type=MessageType.TASK_ASSIGNMENT,
                        content={"task_id": task.task_id}
                    ))
            
            return allocation
        
        def identify_conflicts(self) -> List[Dict[str, Any]]:
            """
            Identify conflicts in task assignments.
            
            Returns:
                List of conflicts
            """
            conflicts = []
            
            # Resource conflicts: agents over capacity
            for agent in self.agents.values():
                if agent.current_load > agent.capacity:
                    conflicts.append({
                        "type": "capacity_exceeded",
                        "agent_id": agent.agent_id,
                        "load": agent.current_load,
                        "capacity": agent.capacity,
                        "excess": agent.current_load - agent.capacity
                    })
            
            # Dependency conflicts: tasks assigned before dependencies
            for task in self.tasks.values():
                if task.status == TaskStatus.ASSIGNED:
                    for dep_id in task.dependencies:
                        if dep_id in self.tasks:
                            dep_task = self.tasks[dep_id]
                            if dep_task.status == TaskStatus.UNASSIGNED:
                                conflicts.append({
                                    "type": "dependency_unassigned",
                                    "task_id": task.task_id,
                                    "dependency_id": dep_id
                                })
            
            return conflicts
        
        def resolve_capacity_conflict(self, agent_id: str):
            """
            Resolve capacity conflict by reassigning tasks.
            
            Args:
                agent_id: Agent with capacity issue
            """
            agent = self.agents[agent_id]
            
            # Sort agent's tasks by priority (keep high priority)
            agent_tasks = [
                self.tasks[tid] for tid in agent.assigned_tasks
                if tid in self.tasks
            ]
            agent_tasks.sort(key=lambda t: t.priority, reverse=True)
            
            # Remove lowest priority tasks until under capacity
            while agent.current_load > agent.capacity and agent_tasks:
                removed_task = agent_tasks.pop()
                removed_task.status = TaskStatus.UNASSIGNED
                removed_task.assigned_to = None
                agent.assigned_tasks.remove(removed_task.task_id)
                agent.current_load -= removed_task.estimated_effort
        
        def execute_tasks(self, agent_id: str, num_tasks: int = 1) -> List[Task]:
            """
            Simulate executing tasks for an agent.
            
            Args:
                agent_id: Agent executing tasks
                num_tasks: Number of tasks to execute
                
            Returns:
                Completed tasks
            """
            agent = self.agents[agent_id]
            completed = []
            
            # Get assigned tasks that are ready (dependencies met)
            ready_tasks = []
            for task_id in agent.assigned_tasks:
                if task_id not in self.tasks:
                    continue
                
                task = self.tasks[task_id]
                if task.status != TaskStatus.ASSIGNED:
                    continue
                
                # Check dependencies
                deps_met = all(
                    self.tasks[dep_id].status == TaskStatus.COMPLETED
                    for dep_id in task.dependencies
                    if dep_id in self.tasks
                )
                
                if deps_met:
                    ready_tasks.append(task)
            
            # Execute up to num_tasks
            for task in ready_tasks[:num_tasks]:
                task.status = TaskStatus.COMPLETED
                task.completed_at = datetime.now()
                agent.current_load -= task.estimated_effort
                completed.append(task)
                
                # Send status update
                self.messages.append(Message(
                    sender_id=agent_id,
                    receiver_id=None,  # Broadcast
                    message_type=MessageType.STATUS_UPDATE,
                    content={
                        "task_id": task.task_id,
                        "status": "completed"
                    }
                ))
            
            return completed
        
        def get_progress_summary(self) -> Dict[str, Any]:
            """Get collaborative planning progress."""
            total_tasks = len(self.tasks)
            completed = sum(1 for t in self.tasks.values() if t.status == TaskStatus.COMPLETED)
            in_progress = sum(1 for t in self.tasks.values() if t.status == TaskStatus.IN_PROGRESS)
            assigned = sum(1 for t in self.tasks.values() if t.status == TaskStatus.ASSIGNED)
            unassigned = sum(1 for t in self.tasks.values() if t.status == TaskStatus.UNASSIGNED)
            
            return {
                "total_tasks": total_tasks,
                "completed": completed,
                "in_progress": in_progress,
                "assigned": assigned,
                "unassigned": unassigned,
                "completion_rate": completed / total_tasks if total_tasks > 0 else 0,
                "total_agents": len(self.agents),
                "messages_sent": len(self.messages)
            }
        
        def get_agent_summary(self) -> str:
            """Get summary of agent workloads."""
            lines = ["AGENT WORKLOADS:"]
            
            for agent in self.agents.values():
                utilization = (agent.current_load / agent.capacity * 100) if agent.capacity > 0 else 0
                lines.append(
                    f"  {agent.name}: {len(agent.assigned_tasks)} tasks, "
                    f"{agent.current_load:.1f}/{agent.capacity:.1f} capacity ({utilization:.0f}%)"
                )
            
            return "\n".join(lines)
    
    
    def demonstrate_collaborative_planning():
        """Demonstrate collaborative planning pattern."""
        
        print("=" * 80)
        print("COLLABORATIVE PLANNING PATTERN DEMONSTRATION")
        print("=" * 80)
        
        # Example 1: Basic collaborative planning
        print("\n" + "=" * 80)
        print("Example 1: Multi-Agent Task Allocation")
        print("=" * 80)
        
        planner = CollaborativePlanner()
        
        # Add agents with different capabilities
        planner.add_agent("agent_1", "Alice", ["analysis", "coding"], capacity=10.0)
        planner.add_agent("agent_2", "Bob", ["coding", "testing"], capacity=10.0)
        planner.add_agent("agent_3", "Charlie", ["design", "communication"], capacity=10.0)
        
        print("\nAgents:")
        for agent in planner.agents.values():
            print(f"  • {agent.name}: {', '.join(agent.capabilities)}")
        
        # Decompose goal into tasks
        print("\nDecomposing goal...")
        tasks = planner.decompose_goal("Build a web application", num_tasks=5)
        
        print(f"\nCreated {len(tasks)} tasks:")
        for task in tasks:
            print(f"  {task.task_id}: {task.description}")
            print(f"    Capabilities: {', '.join(task.required_capabilities)}")
            print(f"    Effort: {task.estimated_effort}")
        
        # Allocate tasks
        print("\nAllocating tasks (greedy)...")
        allocation = planner.allocate_tasks_greedy()
        
        print("\nTask Allocation:")
        for agent_id, tasks in allocation.items():
            agent = planner.agents[agent_id]
            if tasks:
                print(f"  {agent.name}:")
                for task in tasks:
                    print(f"    - {task.description}")
        
        print("\n" + planner.get_agent_summary())
        
        # Example 2: Auction-based allocation
        print("\n" + "=" * 80)
        print("Example 2: Auction-Based Task Allocation")
        print("=" * 80)
        
        planner2 = CollaborativePlanner()
        
        planner2.add_agent("agent_a", "Agent A", ["coding"], capacity=8.0)
        planner2.add_agent("agent_b", "Agent B", ["coding"], capacity=12.0)
        planner2.add_agent("agent_c", "Agent C", ["coding"], capacity=6.0)
        
        # Create tasks manually
        for i in range(4):
            planner2.task_counter += 1
            task = Task(
                task_id=f"task_{planner2.task_counter}",
                description=f"Coding task {i+1}",
                required_capabilities=["coding"],
                estimated_effort=3.0
            )
            planner2.tasks[task.task_id] = task
        
        print("\nRunning auction-based allocation...")
        allocation2 = planner2.allocate_tasks_auction()
        
        print("\nAuction Results:")
        for agent_id, tasks in allocation2.items():
            agent = planner2.agents[agent_id]
            print(f"  {agent.name}: {len(tasks)} tasks "
                  f"(load: {agent.current_load}/{agent.capacity})")
        
        # Example 3: Conflict detection and resolution
        print("\n" + "=" * 80)
        print("Example 3: Conflict Detection and Resolution")
        print("=" * 80)
        
        planner3 = CollaborativePlanner()
        
        agent_d = planner3.add_agent("agent_d", "Agent D", ["task_skill"], capacity=5.0)
        
        # Create tasks that exceed capacity
        for i in range(4):
            planner3.task_counter += 1
            task = Task(
                task_id=f"task_{planner3.task_counter}",
                description=f"Task {i+1}",
                required_capabilities=["task_skill"],
                estimated_effort=2.0,
                status=TaskStatus.ASSIGNED,
                assigned_to="agent_d"
            )
            planner3.tasks[task.task_id] = task
            agent_d.assigned_tasks.append(task.task_id)
            agent_d.current_load += task.estimated_effort
        
        print(f"\nAgent D Load: {agent_d.current_load}/{agent_d.capacity}")
        print("Status: OVERLOADED")
        
        print("\nDetecting conflicts...")
        conflicts = planner3.identify_conflicts()
        
        for conflict in conflicts:
            print(f"  Conflict: {conflict['type']}")
            print(f"    Agent: {conflict['agent_id']}")
            print(f"    Excess load: {conflict['excess']:.1f}")
        
        print("\nResolving capacity conflict...")
        planner3.resolve_capacity_conflict("agent_d")
        
        print(f"\nAfter Resolution:")
        print(f"  Agent D Load: {agent_d.current_load}/{agent_d.capacity}")
        print(f"  Assigned tasks: {len(agent_d.assigned_tasks)}")
        unassigned = sum(1 for t in planner3.tasks.values() if t.status == TaskStatus.UNASSIGNED)
        print(f"  Unassigned tasks: {unassigned}")
        
        # Example 4: Dependency handling
        print("\n" + "=" * 80)
        print("Example 4: Task Dependencies")
        print("=" * 80)
        
        planner4 = CollaborativePlanner()
        
        planner4.add_agent("agent_e", "Agent E", ["any"], capacity=20.0)
        
        # Create tasks with dependencies
        task1 = Task(
            task_id="task_1",
            description="Design system",
            required_capabilities=["any"],
            estimated_effort=2.0,
            status=TaskStatus.ASSIGNED,
            assigned_to="agent_e"
        )
        
        task2 = Task(
            task_id="task_2",
            description="Implement system",
            required_capabilities=["any"],
            estimated_effort=3.0,
            dependencies=["task_1"],
            status=TaskStatus.ASSIGNED,
            assigned_to="agent_e"
        )
        
        task3 = Task(
            task_id="task_3",
            description="Test system",
            required_capabilities=["any"],
            estimated_effort=2.0,
            dependencies=["task_2"],
            status=TaskStatus.ASSIGNED,
            assigned_to="agent_e"
        )
        
        planner4.tasks.update({"task_1": task1, "task_2": task2, "task_3": task3})
        planner4.agents["agent_e"].assigned_tasks = ["task_1", "task_2", "task_3"]
        planner4.agents["agent_e"].current_load = 7.0
        
        print("\nTask Dependencies:")
        print("  task_1 (Design) → task_2 (Implement) → task_3 (Test)")
        
        print("\nExecuting tasks in order:")
        
        # Execute task 1
        completed = planner4.execute_tasks("agent_e", num_tasks=1)
        print(f"  ✓ Completed: {completed[0].description}")
        
        # Execute task 2
        completed = planner4.execute_tasks("agent_e", num_tasks=1)
        print(f"  ✓ Completed: {completed[0].description}")
        
        # Execute task 3
        completed = planner4.execute_tasks("agent_e", num_tasks=1)
        print(f"  ✓ Completed: {completed[0].description}")
        
        # Example 5: Progress tracking
        print("\n" + "=" * 80)
        print("Example 5: Collaborative Progress Tracking")
        print("=" * 80)
        
        planner5 = CollaborativePlanner()
        
        planner5.add_agent("agent_f", "Agent F", ["skill_a"], capacity=10.0)
        planner5.add_agent("agent_g", "Agent G", ["skill_b"], capacity=10.0)
        
        # Create and allocate tasks
        planner5.decompose_goal("Complete project", num_tasks=6)
        planner5.allocate_tasks_greedy()
        
        print("\nInitial Progress:")
        progress = planner5.get_progress_summary()
        print(f"  Total Tasks: {progress['total_tasks']}")
        print(f"  Assigned: {progress['assigned']}")
        print(f"  Unassigned: {progress['unassigned']}")
        print(f"  Completion Rate: {progress['completion_rate']:.0%}")
        
        # Simulate some work
        print("\nSimulating task execution...")
        for agent_id in planner5.agents.keys():
            planner5.execute_tasks(agent_id, num_tasks=2)
        
        print("\nUpdated Progress:")
        progress = planner5.get_progress_summary()
        print(f"  Completed: {progress['completed']}")
        print(f"  Completion Rate: {progress['completion_rate']:.0%}")
        print(f"  Messages Sent: {progress['messages_sent']}")
        
        # Example 6: Load balancing
        print("\n" + "=" * 80)
        print("Example 6: Load Balancing Across Agents")
        print("=" * 80)
        
        planner6 = CollaborativePlanner()
        
        planner6.add_agent("agent_h", "Agent H", ["general"], capacity=10.0)
        planner6.add_agent("agent_i", "Agent I", ["general"], capacity=15.0)
        planner6.add_agent("agent_j", "Agent J", ["general"], capacity=8.0)
        
        # Create uniform tasks
        for i in range(10):
            planner6.task_counter += 1
            task = Task(
                task_id=f"task_{planner6.task_counter}",
                description=f"Task {i+1}",
                required_capabilities=["general"],
                estimated_effort=2.0
            )
            planner6.tasks[task.task_id] = task
        
        print("\nBefore Allocation:")
        print(planner6.get_agent_summary())
        
        planner6.allocate_tasks_greedy()
        
        print("\nAfter Allocation:")
        print(planner6.get_agent_summary())
        
        # Example 7: Message-based coordination
        print("\n" + "=" * 80)
        print("Example 7: Message-Based Coordination")
        print("=" * 80)
        
        planner7 = CollaborativePlanner()
        
        planner7.add_agent("agent_k", "Agent K", ["skill"], capacity=10.0)
        planner7.add_agent("agent_l", "Agent L", ["skill"], capacity=10.0)
        
        # Create tasks and allocate
        for i in range(4):
            planner7.task_counter += 1
            task = Task(
                task_id=f"task_{planner7.task_counter}",
                description=f"Task {i+1}",
                required_capabilities=["skill"],
                estimated_effort=2.0
            )
            planner7.tasks[task.task_id] = task
        
        planner7.allocate_tasks_auction()
        
        print("\nCoordination Messages:")
        for msg in planner7.messages:
            receiver = msg.receiver_id or "ALL"
            print(f"  {msg.sender_id} → {receiver}: {msg.message_type.value}")
            if msg.message_type == MessageType.TASK_ASSIGNMENT:
                print(f"    Task: {msg.content['task_id']}")
        
        # Execute and send status updates
        for agent_id in planner7.agents.keys():
            planner7.execute_tasks(agent_id, num_tasks=1)
        
        print(f"\nTotal Messages: {len(planner7.messages)}")
        
        # Example 8: System summary
        print("\n" + "=" * 80)
        print("Example 8: Collaborative Planning Summary")
        print("=" * 80)
        
        planner8 = CollaborativePlanner()
        
        # Setup system
        planner8.add_agent("agent_m", "Specialist 1", ["analysis"], capacity=10.0)
        planner8.add_agent("agent_n", "Specialist 2", ["coding"], capacity=10.0)
        planner8.add_agent("agent_o", "Generalist", ["analysis", "coding"], capacity=12.0)
        
        planner8.decompose_goal("Build and deploy application", num_tasks=8)
        planner8.allocate_tasks_greedy()
        
        # Execute some tasks
        for agent_id in planner8.agents.keys():
            planner8.execute_tasks(agent_id, num_tasks=1)
        
        print("\nSYSTEM SUMMARY")
        print("=" * 60)
        print(planner8.get_agent_summary())
        
        print("\nPROGRESS:")
        progress = planner8.get_progress_summary()
        for key, value in progress.items():
            if isinstance(value, float):
                print(f"  {key}: {value:.2f}")
            else:
                print(f"  {key}: {value}")
        
        # Summary
        print("\n" + "=" * 80)
        print("SUMMARY: Collaborative Planning Pattern")
        print("=" * 80)
        
        summary = """
        The Collaborative Planning pattern demonstrated:
        
        1. MULTI-AGENT ALLOCATION (Example 1):
           - Goal decomposition
           - Agent capabilities matching
           - Greedy task allocation
           - Workload distribution
           - LLM-based task generation
        
        2. AUCTION MECHANISM (Example 2):
           - Bidding-based allocation
           - Capacity-based bidding
           - Winner selection
           - Fair task distribution
           - Market-based coordination
        
        3. CONFLICT RESOLUTION (Example 3):
           - Capacity conflict detection
           - Over-allocation handling
           - Task reassignment
           - Constraint satisfaction
           - Automatic resolution
        
        4. DEPENDENCY MANAGEMENT (Example 4):
           - Task dependencies
           - Sequential execution
           - Prerequisite checking
           - Ordered completion
           - Dependency-aware scheduling
        
        5. PROGRESS TRACKING (Example 5):
           - Status monitoring
           - Completion rates
           - Message counting
           - System metrics
           - Real-time updates
        
        6. LOAD BALANCING (Example 6):
           - Capacity-aware allocation
           - Even workload distribution
           - Utilization optimization
           - Resource efficiency
           - Balanced assignment
        
        7. MESSAGE COORDINATION (Example 7):
           - Assignment messages
           - Status updates
           - Broadcast communication
           - Coordination protocol
           - Event notification
        
        8. SYSTEM INTEGRATION (Example 8):
           - Comprehensive summary
           - Agent workloads
           - Progress metrics
           - System statistics
           - Complete view
        
        KEY BENEFITS:
        ✓ Leverages diverse agent capabilities
        ✓ Handles complex distributed tasks
        ✓ Load balancing across agents
        ✓ Fault tolerance through redundancy
        ✓ Parallel execution speedup
        ✓ Shared knowledge utilization
        ✓ Flexible task distribution
        ✓ Scalable coordination
        
        USE CASES:
        • Multi-robot coordination
        • Distributed project management
        • Supply chain planning
        • Military operations planning
        • Smart city management
        • Collaborative problem solving
        • Team-based game AI
        • Cloud resource orchestration
        
        COLLABORATION STRATEGIES:
        → Centralized: Coordinator assigns
        → Hierarchical: Tree of coordinators
        → Distributed: Peer negotiation
        → Auction-based: Competitive bidding
        → Consensus: Democratic voting
        
        BEST PRACTICES:
        1. Implement efficient communication protocols
        2. Use asynchronous messaging
        3. Handle agent failures gracefully
        4. Monitor collaboration overhead
        5. Implement timeout mechanisms
        6. Log all coordination decisions
        7. Support dynamic agent join/leave
        8. Enable human oversight
        
        TRADE-OFFS:
        • Communication overhead vs. coordination quality
        • Centralized vs. distributed control
        • Planning time vs. execution efficiency
        • Autonomy vs. coordination
        
        PRODUCTION CONSIDERATIONS:
        → Use message queues for async communication
        → Implement heartbeat monitoring
        → Handle network partitions
        → Cache agent capabilities
        → Support graceful degradation
        → Monitor communication latency
        → Log all agent interactions
        → Implement conflict resolution strategies
        → Support dynamic reconfiguration
        → Enable human intervention points
        
        This pattern enables multiple agents to work together effectively
        by coordinating their planning, sharing tasks based on capabilities,
        and synchronizing execution toward shared goals.
        """
        
        print(summary)
    
    
    if __name__ == "__main__":
        demonstrate_collaborative_planning()
    
        

    Agentic Design Pattern: Collaborative Planning

    Pattern 108: Collaborative Planning – Line-by-Line Explanation

    Overview

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


    Pattern Description

    Pattern 108: Collaborative Planning

    Description: The Collaborative Planning pattern enables multiple agents to work together in creating and executing plans toward shared goals. Rather than planning in isolation, agents coordinate their planning efforts, share information, divide tasks, resolve conflicts, and synchronize their actions to achieve objectives that would be difficult or impossible for individual agents.

    Collaborative planning is essential for multi-agent systems where agents have
    complementary capabilities, shared resources, or interdependent goals. The
    pattern addresses challenges like distributed knowledge, conflicting preferences,
    resource contention, and coordination overhead.
    
    This pattern includes mechanisms for goal sharing, task decomposition and
    allocation, plan merging, conflict resolution, resource negotiation, and
    synchronized execution. Agents may use various collaboration strategies from
    centralized coordination to fully distributed consensus-based planning.
    

    Key Components: 1. Shared Goal Repository: Common objectives 2. Task Allocator: Assigns tasks to agents 3. Plan Merger: Combines individual plans 4. Conflict Resolver: Handles plan conflicts 5. Resource Negotiator: Allocates shared resources 6. Synchronization Manager: Coordinates timing 7. Communication Protocol: Agent interaction

    Collaboration Strategies: 1. Centralized: Single coordinator assigns tasks 2. Hierarchical: Tree of coordinators 3. Distributed: Peer-to-peer negotiation 4. Auction-based: Agents bid for tasks 5. Consensus: Democratic decision making 6. Blackboard: Shared workspace

    Planning Phases: 1. Goal Identification: Define shared objectives 2. Capability Sharing: Agents advertise skills 3. Task Decomposition: Break into subtasks 4. Task Allocation: Assign to appropriate agents 5. Plan Integration: Merge individual plans 6. Conflict Resolution: Resolve incompatibilities 7. Synchronized Execution: Coordinate timing

    Use Cases: – Multi-robot coordination – Distributed project management – Supply chain planning – Military operations planning – Smart city management – Collaborative problem solving – Team-based game AI

    Advantages: – Leverages diverse capabilities – Handles complex tasks beyond single agent – Load balancing across agents – Fault tolerance through redundancy – Parallel execution speedup – Shared knowledge utilization – Flexible task distribution

    Challenges: – Communication overhead – Conflict resolution complexity – Synchronization requirements – Trust and verification – Resource contention – Plan consistency maintenance – Scalability with agent count

    LangChain Implementation: This implementation uses LangChain for: – LLM-based task decomposition – Agent capability matching – Plan integration strategies – Natural language coordination

    Production Considerations: – Implement efficient communication protocols – Use asynchronous messaging – Handle agent failures gracefully – Monitor collaboration overhead – Implement timeout mechanisms – Log all coordination decisions – Support dynamic agent addition/removal – Enable human oversight


    File Statistics

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

    Module Documentation

    """
    Pattern 108: Collaborative Planning
    
    Description:
        The Collaborative Planning pattern enables multiple agents to work together
        in creating and executing plans toward shared goals. Rather than planning in
        isolation, agents coordinate their planning efforts, share information, divide
        tasks, resolve conflicts, and synchronize their actions to achieve objectives
        that would be difficult or impossible for individual agents.
    
        Collaborative planning is essential for multi-agent systems where agents have
        complementary capabilities, shared resources, or interdependent goals. The
        pattern addresses challenges like distributed knowledge, conflicting preferences,
        resource contention, and coordination overhead.
    
        This pattern includes mechanisms for goal sharing, task decomposition and
        allocation, plan merging, conflict resolution, resource negotiation, and
        synchronized execution. Agents may use various collaboration strategies from
        centralized coordination to fully distributed consensus-based planning.
    
    Key Components:
        1. Shared Goal Repository: Common objectives
        2. Task Allocator: Assigns tasks to agents
        3. Plan Merger: Combines individual plans
        4. Conflict Resolver: Handles plan conflicts
        5. Resource Negotiator: Allocates shared resources
        6. Synchronization Manager: Coordinates timing
        7. Communication Protocol: Agent interaction
    
    Collaboration Strategies:
        1. Centralized: Single coordinator assigns tasks
        2. Hierarchical: Tree of coordinators
        3. Distributed: Peer-to-peer negotiation
        4. Auction-based: Agents bid for tasks
        5. Consensus: Democratic decision making
        6. Blackboard: Shared workspace
    
    Planning Phases:
        1. Goal Identification: Define shared objectives
        2. Capability Sharing: Agents advertise skills
        3. Task Decomposition: Break into subtasks
        4. Task Allocation: Assign to appropriate agents
        5. Plan Integration: Merge individual plans
        6. Conflict Resolution: Resolve incompatibilities
        7. Synchronized Execution: Coordinate timing
    
    Use Cases:
        - Multi-robot coordination
        - Distributed project management
        - Supply chain planning
        - Military operations planning
        - Smart city management
        - Collaborative problem solving
        - Team-based game AI
    
    Advantages:
        - Leverages diverse capabilities
        - Handles complex tasks beyond single agent
        - Load balancing across agents
        - Fault tolerance through redundancy
        - Parallel execution speedup
        - Shared knowledge utilization
        - Flexible task distribution
    
    Challenges:
        - Communication overhead
        - Conflict resolution complexity
        - Synchronization requirements
        - Trust and verification
        - Resource contention
        - Plan consistency maintenance
        - Scalability with agent count
    
    LangChain Implementation:
        This implementation uses LangChain for:
        - LLM-based task decomposition
        - Agent capability matching
        - Plan integration strategies
        - Natural language coordination
    
    Production Considerations:
        - Implement efficient communication protocols
        - Use asynchronous messaging
        - Handle agent failures gracefully
        - Monitor collaboration overhead
        - Implement timeout mechanisms
        - Log all coordination decisions
        - Support dynamic agent addition/removal
        - Enable human oversight
    """
    

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


    Import Statements

    Standard Library Imports

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

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

    LangChain Imports

    from langchain_openai.ChatOpenAI
    

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

    Other Imports

    from dotenv.load_dotenv
    

    Purpose: Additional dependencies for specialized functionality.


    Classes

    Class: TaskStatus

    Lines 106-112

    Inherits from: Enum

    Purpose:

    Status of a task.
    

    Class: MessageType

    Lines 115-123

    Inherits from: Enum

    Purpose:

    Types of coordination messages.
    

    Class: Task

    Lines 127-150

    Purpose:

    Represents a task in collaborative plan.
    
    Attributes:
        task_id: Unique identifier
        description: Task description
        required_capabilities: Capabilities needed
        estimated_effort: Effort estimate
        dependencies: Tasks that must complete first
        assigned_to: Agent ID assigned to
        status: Current status
        priority: Task priority
    

    Class: CollaborativeAgent

    Lines 154-185

    Purpose:

    Agent participating in collaborative planning.
    
    Attributes:
        agent_id: Unique identifier
        name: Agent name
        capabilities: List of capabilities
        capacity: Available work capacity
        current_load: Current workload
        assigned_tasks: Tasks assigned to this agent
    

    Methods:

    can_handle(self, task)

    Lines 173-181

    Purpose:

    Check if agent can handle a task.
    

    Parameters: – task: Task

    Returns: bool

    get_available_capacity(self)

    Lines 183-185

    Purpose:

    Get remaining capacity.
    

    Parameters:

    Returns: float


    Class: Message

    Lines 189-204

    Purpose:

    Communication message between agents.
    
    Attributes:
        sender_id: Sending agent ID
        receiver_id: Receiving agent ID (None for broadcast)
        message_type: Type of message
        content: Message payload
        timestamp: When sent
    

    Class: CollaborativePlanner

    Lines 207-543

    Purpose:

    Collaborative planning system for multi-agent coordination.
    
    This planner enables agents to work together in planning and
    executing tasks toward shared goals.
    

    Methods:

    __init__(self, temperature)

    Lines 215-226

    Purpose:

    Initialize collaborative planner.
    
    Args:
        temperature: LLM temperature
    

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

    add_agent(self, agent_id, name, capabilities, capacity)

    Lines 228-254

    Purpose:

    Add agent to the planning system.
    
    Args:
        agent_id: Agent identifier
        name: Agent name
        capabilities: Agent capabilities
        capacity: Work capacity
    
    Returns:
        Created agent
    

    Parameters: – agent_id: Agent id – name: Name – capabilities: Capabilities – capacity: Capacity

    Returns: CollaborativeAgent

    decompose_goal(self, goal, num_tasks)

    Lines 256-320

    Purpose:

    Decompose goal into tasks using LLM.
    
    Args:
        goal: Goal description
        num_tasks: Number of tasks to create
    
    Returns:
        List of tasks
    

    Parameters: – goal: Goal – num_tasks: Num tasks

    Returns: <ast.Subscript object at 0x000001A06D68E690>

    allocate_tasks_greedy(self)

    Lines 322-356

    Purpose:

    Allocate tasks to agents using greedy algorithm.
    
    Returns:
        Allocation map: agent_id -> tasks
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D888BD0>

    allocate_tasks_auction(self)

    Lines 358-398

    Purpose:

    Allocate tasks using auction-based mechanism.
    
    Returns:
        Allocation map: agent_id -> tasks
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D70C610>

    identify_conflicts(self)

    Lines 400-433

    Purpose:

    Identify conflicts in task assignments.
    
    Returns:
        List of conflicts
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D638B90>

    resolve_capacity_conflict(self, agent_id)

    Lines 435-457

    Purpose:

    Resolve capacity conflict by reassigning tasks.
    
    Args:
        agent_id: Agent with capacity issue
    

    Parameters: – agent_id: Agent id

    execute_tasks(self, agent_id, num_tasks)

    Lines 459-511

    Purpose:

    Simulate executing tasks for an agent.
    
    Args:
        agent_id: Agent executing tasks
        num_tasks: Number of tasks to execute
    
    Returns:
        Completed tasks
    

    Parameters: – agent_id: Agent id – num_tasks: Num tasks

    Returns: <ast.Subscript object at 0x000001A06D63D5D0>

    get_progress_summary(self)

    Lines 513-530

    Purpose:

    Get collaborative planning progress.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D7C82D0>

    get_agent_summary(self)

    Lines 532-543

    Purpose:

    Get summary of agent workloads.
    

    Parameters:

    Returns: str


    Functions

    demonstrate_collaborative_planning()

    Lines 546-981

    Purpose:

    Demonstrate collaborative planning pattern.
    

    Key Concepts Summary

    Pattern Overview

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

    Core Components

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

    When to Use

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

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


    Best Practices Demonstrated

    1. Environment Configuration

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

    2. Error Handling

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

    3. Prompt Engineering

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

    4. Code Organization

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

    5. Performance Optimization

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

    LangChain Components Used

    • ChatOpenAI: OpenAI model interface
    • Chains: Sequential component composition

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


    Execution Flow

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

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


    Performance Considerations

    Computational Cost

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

    Optimization Strategies

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

    Cost Estimates (Approximate)

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

    Actual costs vary based on input/output token counts


    Common Pitfalls to Avoid

    1. Prompt Design Issues

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

    2. Error Handling

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

    3. Token Management

    ❌ Exceeding model token limits
    ✅ Truncate inputs, chunk processing

    4. Security

    ❌ Exposing API keys in code
    ✅ Use environment variables

    5. Testing

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

    6. Cost Management

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


    Further Improvements

    Production Enhancements

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

    Feature Extensions

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

    Code Quality

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

    Related Patterns

    • Pattern 107
    • Pattern 109

    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 Collaborative Planning pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

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

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

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


    File: 108_collaborative_planning.py
    Total Lines: 986
    Pattern Category: AI Agent Design Pattern #108


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Multi Objective Planning

    
        """
    Pattern 107: Multi-Objective Planning
    
    Description:
        The Multi-Objective Planning pattern enables agents to plan and make decisions
        when multiple, often conflicting objectives must be balanced. Instead of
        optimizing for a single goal, the agent must consider tradeoffs between
        competing objectives like cost vs. quality, speed vs. accuracy, or risk vs.
        reward.
        
        This pattern is essential for realistic decision-making where perfect solutions
        don't exist and every choice involves tradeoffs. The agent must evaluate plans
        across multiple dimensions, understand Pareto-optimal solutions, and make
        informed decisions about which tradeoffs are acceptable given the context.
        
        Multi-objective planning involves objective weighting, constraint satisfaction,
        Pareto optimization, preference elicitation, and tradeoff analysis. The agent
        can present multiple plan options with different tradeoff profiles, allowing
        humans or automated systems to select based on current priorities.
    
    Key Components:
        1. Objective Set: Multiple goals to optimize
        2. Objective Weights: Relative importance of each objective
        3. Plan Evaluator: Scores plans across all objectives
        4. Pareto Filter: Identifies non-dominated solutions
        5. Tradeoff Analyzer: Analyzes competing objectives
        6. Constraint Checker: Ensures hard constraints met
        7. Preference Engine: Incorporates user preferences
    
    Objective Types:
        1. Maximization: Quality, performance, completeness
        2. Minimization: Cost, time, risk, resource usage
        3. Target: Hit specific value (with tolerance)
        4. Constraint: Must satisfy (hard requirement)
        
    Common Objective Combinations:
        - Cost vs. Quality vs. Speed (iron triangle)
        - Risk vs. Reward
        - Precision vs. Recall
        - Exploration vs. Exploitation
        - Short-term vs. Long-term goals
        - Individual vs. Group benefit
    
    Planning Approaches:
        1. Weighted Sum: Combine objectives with weights
        2. Lexicographic: Prioritize objectives in order
        3. Pareto Optimal: Non-dominated solutions
        4. Satisficing: Good enough on all objectives
        5. Goal Programming: Meet targets with deviations
        6. Multi-Criteria Decision Making (MCDM)
    
    Use Cases:
        - Resource allocation with constraints
        - Project planning with multiple stakeholders
        - Route planning (time vs. cost vs. comfort)
        - Product design (features vs. complexity vs. cost)
        - Investment decisions (risk vs. return)
        - Scheduling (efficiency vs. fairness)
        - Machine learning (accuracy vs. interpretability)
    
    Advantages:
        - Realistic modeling of complex decisions
        - Explicit tradeoff consideration
        - Multiple solution alternatives
        - Stakeholder preference incorporation
        - Better than single-objective optimization
        - Transparent decision criteria
        - Flexible priority adjustment
    
    Challenges:
        - Objective quantification difficulty
        - Weight elicitation from users
        - Computational complexity
        - Visualizing multi-dimensional tradeoffs
        - Handling conflicting constraints
        - Determining Pareto frontier
        - Explaining tradeoff decisions
    
    LangChain Implementation:
        This implementation uses LangChain for:
        - LLM-based objective identification
        - Plan evaluation across dimensions
        - Tradeoff analysis and explanation
        - Natural language objective specification
        
    Production Considerations:
        - Define clear, measurable objectives
        - Implement efficient Pareto filtering
        - Provide tradeoff visualizations
        - Allow dynamic weight adjustment
        - Cache evaluated plans
        - Support constraint hierarchies
        - Enable sensitivity analysis
        - Log decision rationale
    """
    
    import os
    from typing import List, Dict, Any, Optional, Tuple
    from dataclasses import dataclass, field
    from enum import Enum
    from dotenv import load_dotenv
    
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser
    from langchain_openai import ChatOpenAI
    
    load_dotenv()
    
    
    class ObjectiveType(Enum):
        """Type of optimization objective."""
        MAXIMIZE = "maximize"
        MINIMIZE = "minimize"
        TARGET = "target"
        CONSTRAINT = "constraint"
    
    
    @dataclass
    class Objective:
        """
        Represents an optimization objective.
        
        Attributes:
            name: Objective name
            objective_type: Maximization, minimization, target, or constraint
            weight: Relative importance (0-1)
            target_value: Target value (for TARGET type)
            constraint_value: Threshold (for CONSTRAINT type)
            unit: Measurement unit
        """
        name: str
        objective_type: ObjectiveType
        weight: float = 0.5
        target_value: Optional[float] = None
        constraint_value: Optional[float] = None
        unit: str = ""
        
        def evaluate(self, value: float) -> float:
            """
            Evaluate objective given a value.
            
            Args:
                value: Actual value
                
            Returns:
                Normalized score (0-1), higher is better
            """
            if self.objective_type == ObjectiveType.MAXIMIZE:
                # Normalize to 0-1 (assuming reasonable range)
                return min(1.0, max(0.0, value / 100.0))
            
            elif self.objective_type == ObjectiveType.MINIMIZE:
                # Lower is better, invert score
                return max(0.0, 1.0 - (value / 100.0))
            
            elif self.objective_type == ObjectiveType.TARGET:
                # Closer to target is better
                if self.target_value is None:
                    return 0.5
                distance = abs(value - self.target_value)
                return max(0.0, 1.0 - (distance / 50.0))
            
            elif self.objective_type == ObjectiveType.CONSTRAINT:
                # Binary: constraint satisfied or not
                if self.constraint_value is None:
                    return 1.0
                return 1.0 if value  float:
            """Get normalized score for an objective."""
            value = self.objective_values.get(objective.name, 0.0)
            return objective.evaluate(value)
    
    
    class MultiObjectivePlanner:
        """
        Multi-objective planning system.
        
        This planner evaluates and compares plans across multiple objectives,
        identifies Pareto-optimal solutions, and analyzes tradeoffs.
        """
        
        def __init__(self, temperature: float = 0.4):
            """
            Initialize multi-objective planner.
            
            Args:
                temperature: LLM temperature
            """
            self.llm = ChatOpenAI(temperature=temperature, model="gpt-3.5-turbo")
            self.objectives: List[Objective] = []
            self.plans: List[Plan] = []
            self.plan_counter = 0
        
        def add_objective(
            self,
            name: str,
            objective_type: ObjectiveType,
            weight: float = 0.5,
            target_value: Optional[float] = None,
            constraint_value: Optional[float] = None,
            unit: str = ""
        ) -> Objective:
            """
            Add an objective to optimize.
            
            Args:
                name: Objective name
                objective_type: Type of objective
                weight: Relative importance
                target_value: Target value (for TARGET)
                constraint_value: Constraint threshold
                unit: Measurement unit
                
            Returns:
                Created objective
            """
            objective = Objective(
                name=name,
                objective_type=objective_type,
                weight=weight,
                target_value=target_value,
                constraint_value=constraint_value,
                unit=unit
            )
            self.objectives.append(objective)
            return objective
        
        def create_plan(
            self,
            description: str,
            objective_values: Dict[str, float]
        ) -> Plan:
            """
            Create a plan with objective values.
            
            Args:
                description: Plan description
                objective_values: Values for each objective
                
            Returns:
                Created plan
            """
            self.plan_counter += 1
            
            # Check constraint satisfaction
            feasible = True
            for obj in self.objectives:
                if obj.objective_type == ObjectiveType.CONSTRAINT:
                    value = objective_values.get(obj.name, float('inf'))
                    if obj.constraint_value is not None and value > obj.constraint_value:
                        feasible = False
                        break
            
            plan = Plan(
                plan_id=f"plan_{self.plan_counter}",
                description=description,
                objective_values=objective_values,
                feasible=feasible
            )
            
            self.plans.append(plan)
            return plan
        
        def evaluate_plan(self, plan: Plan) -> Dict[str, Any]:
            """
            Evaluate plan across all objectives.
            
            Args:
                plan: Plan to evaluate
                
            Returns:
                Evaluation results
            """
            scores = {}
            weighted_sum = 0.0
            total_weight = 0.0
            
            for obj in self.objectives:
                score = plan.get_score(obj)
                scores[obj.name] = score
                
                if obj.objective_type != ObjectiveType.CONSTRAINT:
                    weighted_sum += score * obj.weight
                    total_weight += obj.weight
            
            overall_score = weighted_sum / total_weight if total_weight > 0 else 0.0
            
            return {
                "plan_id": plan.plan_id,
                "feasible": plan.feasible,
                "scores": scores,
                "overall_score": overall_score,
                "objective_values": plan.objective_values
            }
        
        def find_pareto_optimal(self) -> List[Plan]:
            """
            Find Pareto-optimal (non-dominated) plans.
            
            Returns:
                List of Pareto-optimal plans
            """
            feasible_plans = [p for p in self.plans if p.feasible]
            
            if not feasible_plans:
                return []
            
            # Mark dominated plans
            for plan in feasible_plans:
                plan.dominated = False
            
            for i, plan1 in enumerate(feasible_plans):
                for plan2 in feasible_plans[i+1:]:
                    # Check if plan1 dominates plan2
                    plan1_better = False
                    plan2_better = False
                    
                    for obj in self.objectives:
                        if obj.objective_type == ObjectiveType.CONSTRAINT:
                            continue
                        
                        score1 = plan1.get_score(obj)
                        score2 = plan2.get_score(obj)
                        
                        if score1 > score2:
                            plan1_better = True
                        elif score2 > score1:
                            plan2_better = True
                    
                    # Mark dominated plans
                    if plan1_better and not plan2_better:
                        plan2.dominated = True
                    elif plan2_better and not plan1_better:
                        plan1.dominated = True
            
            return [p for p in feasible_plans if not p.dominated]
        
        def analyze_tradeoffs(
            self,
            plan1: Plan,
            plan2: Plan
        ) -> Dict[str, Any]:
            """
            Analyze tradeoffs between two plans.
            
            Args:
                plan1: First plan
                plan2: Second plan
                
            Returns:
                Tradeoff analysis
            """
            tradeoffs = []
            
            for obj in self.objectives:
                if obj.objective_type == ObjectiveType.CONSTRAINT:
                    continue
                
                score1 = plan1.get_score(obj)
                score2 = plan2.get_score(obj)
                value1 = plan1.objective_values.get(obj.name, 0)
                value2 = plan2.objective_values.get(obj.name, 0)
                
                difference = score2 - score1
                
                if abs(difference) > 0.05:  # Significant difference
                    tradeoffs.append({
                        "objective": obj.name,
                        "plan1_value": value1,
                        "plan2_value": value2,
                        "plan1_score": score1,
                        "plan2_score": score2,
                        "difference": difference,
                        "winner": "plan2" if difference > 0 else "plan1"
                    })
            
            return {
                "plan1": plan1.plan_id,
                "plan2": plan2.plan_id,
                "tradeoffs": tradeoffs,
                "num_tradeoffs": len(tradeoffs)
            }
        
        def generate_plan_with_priorities(
            self,
            task: str,
            priority_objectives: List[str]
        ) -> Plan:
            """
            Generate plan optimizing for priority objectives.
            
            Args:
                task: Task description
                priority_objectives: High-priority objective names
                
            Returns:
                Generated plan
            """
            # Create prompt emphasizing priority objectives
            objectives_text = "\n".join([
                f"- {obj.name} ({obj.objective_type.value})"
                + (f" - PRIORITY" if obj.name in priority_objectives else "")
                for obj in self.objectives
            ])
            
            prompt = ChatPromptTemplate.from_template(
                "Create a plan for this task that optimizes the priority objectives:\n\n"
                "Task: {task}\n\n"
                "Objectives:\n{objectives}\n\n"
                "Provide a brief plan description (one sentence).\n"
                "Plan:"
            )
            
            chain = prompt | self.llm | StrOutputParser()
            description = chain.invoke({
                "task": task,
                "objectives": objectives_text
            }).strip()
            
            # Generate estimated values (in practice, would come from simulation)
            objective_values = {}
            for obj in self.objectives:
                # Favor priority objectives
                if obj.name in priority_objectives:
                    if obj.objective_type == ObjectiveType.MAXIMIZE:
                        objective_values[obj.name] = 80.0
                    elif obj.objective_type == ObjectiveType.MINIMIZE:
                        objective_values[obj.name] = 20.0
                    else:
                        objective_values[obj.name] = 50.0
                else:
                    # Non-priority gets moderate values
                    objective_values[obj.name] = 50.0
            
            return self.create_plan(description, objective_values)
        
        def rank_plans(self) -> List[Tuple[Plan, float]]:
            """
            Rank plans by weighted overall score.
            
            Returns:
                List of (plan, score) tuples, sorted by score
            """
            ranked = []
            
            for plan in self.plans:
                if not plan.feasible:
                    continue
                
                eval_result = self.evaluate_plan(plan)
                ranked.append((plan, eval_result["overall_score"]))
            
            ranked.sort(key=lambda x: x[1], reverse=True)
            return ranked
        
        def get_summary(self) -> str:
            """Get summary of plans and objectives."""
            lines = ["MULTI-OBJECTIVE PLANNING SUMMARY"]
            lines.append("=" * 60)
            
            # Objectives
            lines.append("\nObjectives:")
            for obj in self.objectives:
                lines.append(
                    f"  • {obj.name} ({obj.objective_type.value})"
                    f" - weight: {obj.weight:.2f}"
                )
            
            # Plans
            lines.append(f"\nTotal Plans: {len(self.plans)}")
            lines.append(f"Feasible Plans: {sum(1 for p in self.plans if p.feasible)}")
            
            # Pareto optimal
            pareto = self.find_pareto_optimal()
            lines.append(f"Pareto-Optimal Plans: {len(pareto)}")
            
            return "\n".join(lines)
    
    
    def demonstrate_multi_objective_planning():
        """Demonstrate multi-objective planning pattern."""
        
        print("=" * 80)
        print("MULTI-OBJECTIVE PLANNING PATTERN DEMONSTRATION")
        print("=" * 80)
        
        # Example 1: Basic multi-objective planning
        print("\n" + "=" * 80)
        print("Example 1: Project Planning with Cost, Time, Quality")
        print("=" * 80)
        
        planner = MultiObjectivePlanner()
        
        # Define objectives
        planner.add_objective("cost", ObjectiveType.MINIMIZE, weight=0.4, unit="$")
        planner.add_objective("time", ObjectiveType.MINIMIZE, weight=0.3, unit="days")
        planner.add_objective("quality", ObjectiveType.MAXIMIZE, weight=0.3, unit="score")
        
        print("\nObjectives:")
        for obj in planner.objectives:
            print(f"  • {obj.name}: {obj.objective_type.value} (weight: {obj.weight})")
        
        # Create alternative plans
        plans_data = [
            ("Budget plan: Minimize costs, longer timeline", {"cost": 10000, "time": 60, "quality": 70}),
            ("Fast plan: Minimize time, higher costs", {"cost": 25000, "time": 30, "quality": 75}),
            ("Quality plan: Maximize quality, balanced cost/time", {"cost": 18000, "time": 45, "quality": 90}),
            ("Balanced plan: Moderate on all objectives", {"cost": 15000, "time": 45, "quality": 80}),
        ]
        
        print("\nCreating alternative plans...")
        for desc, values in plans_data:
            plan = planner.create_plan(desc, values)
            print(f"  ✓ {plan.plan_id}: {desc}")
        
        # Evaluate plans
        print("\nPlan Evaluations:")
        for plan in planner.plans:
            eval_result = planner.evaluate_plan(plan)
            print(f"\n  {plan.plan_id}: {plan.description}")
            print(f"    Overall Score: {eval_result['overall_score']:.3f}")
            for obj_name, score in eval_result['scores'].items():
                value = plan.objective_values[obj_name]
                print(f"    {obj_name}: {value} → score: {score:.3f}")
        
        # Example 2: Pareto optimal solutions
        print("\n" + "=" * 80)
        print("Example 2: Finding Pareto-Optimal Solutions")
        print("=" * 80)
        
        planner2 = MultiObjectivePlanner()
        
        planner2.add_objective("performance", ObjectiveType.MAXIMIZE, weight=0.5)
        planner2.add_objective("efficiency", ObjectiveType.MAXIMIZE, weight=0.5)
        
        # Create plans with different tradeoffs
        test_plans = [
            ("High performance, low efficiency", {"performance": 90, "efficiency": 40}),
            ("Low performance, high efficiency", {"performance": 40, "efficiency": 90}),
            ("Balanced", {"performance": 70, "efficiency": 70}),
            ("Dominated plan", {"performance": 50, "efficiency": 50}),
            ("Another balanced", {"performance": 75, "efficiency": 65}),
        ]
        
        for desc, values in test_plans:
            planner2.create_plan(desc, values)
        
        print("\nAll Plans:")
        for plan in planner2.plans:
            print(f"  {plan.plan_id}: {plan.description}")
            print(f"    Performance: {plan.objective_values['performance']}, "
                  f"Efficiency: {plan.objective_values['efficiency']}")
        
        pareto_plans = planner2.find_pareto_optimal()
        
        print(f"\nPareto-Optimal Plans ({len(pareto_plans)} found):")
        for plan in pareto_plans:
            print(f"  ✓ {plan.plan_id}: {plan.description}")
            print(f"    Performance: {plan.objective_values['performance']}, "
                  f"Efficiency: {plan.objective_values['efficiency']}")
        
        # Example 3: Tradeoff analysis
        print("\n" + "=" * 80)
        print("Example 3: Tradeoff Analysis Between Plans")
        print("=" * 80)
        
        planner3 = MultiObjectivePlanner()
        
        planner3.add_objective("speed", ObjectiveType.MAXIMIZE, weight=0.4)
        planner3.add_objective("accuracy", ObjectiveType.MAXIMIZE, weight=0.4)
        planner3.add_objective("cost", ObjectiveType.MINIMIZE, weight=0.2)
        
        plan_a = planner3.create_plan(
            "Fast but less accurate",
            {"speed": 90, "accuracy": 60, "cost": 20}
        )
        
        plan_b = planner3.create_plan(
            "Slow but very accurate",
            {"speed": 50, "accuracy": 95, "cost": 40}
        )
        
        print("\nComparing two plans:")
        print(f"  Plan A: {plan_a.description}")
        print(f"  Plan B: {plan_b.description}")
        
        tradeoff = planner3.analyze_tradeoffs(plan_a, plan_b)
        
        print(f"\nTradeoff Analysis ({tradeoff['num_tradeoffs']} significant differences):")
        for t in tradeoff['tradeoffs']:
            winner = "Plan A" if t['winner'] == 'plan1' else "Plan B"
            print(f"  • {t['objective']}:")
            print(f"    Plan A: {t['plan1_value']} (score: {t['plan1_score']:.3f})")
            print(f"    Plan B: {t['plan2_value']} (score: {t['plan2_score']:.3f})")
            print(f"    Winner: {winner}")
        
        # Example 4: Constraints
        print("\n" + "=" * 80)
        print("Example 4: Hard Constraints")
        print("=" * 80)
        
        planner4 = MultiObjectivePlanner()
        
        planner4.add_objective("profit", ObjectiveType.MAXIMIZE, weight=0.6)
        planner4.add_objective("customer_satisfaction", ObjectiveType.MAXIMIZE, weight=0.4)
        planner4.add_objective("budget", ObjectiveType.CONSTRAINT, constraint_value=50000)
        
        print("\nObjectives:")
        print("  • Maximize profit (weight: 0.6)")
        print("  • Maximize customer satisfaction (weight: 0.4)")
        print("  • Budget constraint: ≤ $50,000")
        
        constraint_plans = [
            ("Within budget", {"profit": 30000, "customer_satisfaction": 80, "budget": 45000}),
            ("Over budget", {"profit": 50000, "customer_satisfaction": 90, "budget": 60000}),
            ("Tight budget", {"profit": 25000, "customer_satisfaction": 75, "budget": 50000}),
        ]
        
        print("\nEvaluating plans:")
        for desc, values in constraint_plans:
            plan = planner4.create_plan(desc, values)
            status = "✓ FEASIBLE" if plan.feasible else "✗ INFEASIBLE"
            print(f"  {status}: {desc}")
            print(f"    Budget: ${values['budget']:,}")
        
        # Example 5: Ranking plans
        print("\n" + "=" * 80)
        print("Example 5: Ranking Plans by Overall Score")
        print("=" * 80)
        
        planner5 = MultiObjectivePlanner()
        
        planner5.add_objective("innovation", ObjectiveType.MAXIMIZE, weight=0.3)
        planner5.add_objective("feasibility", ObjectiveType.MAXIMIZE, weight=0.4)
        planner5.add_objective("risk", ObjectiveType.MINIMIZE, weight=0.3)
        
        ranking_plans = [
            ("Conservative approach", {"innovation": 40, "feasibility": 90, "risk": 10}),
            ("Innovative but risky", {"innovation": 90, "feasibility": 60, "risk": 70}),
            ("Moderate innovation", {"innovation": 70, "feasibility": 80, "risk": 30}),
            ("Safe and feasible", {"innovation": 50, "feasibility": 95, "risk": 15}),
        ]
        
        for desc, values in ranking_plans:
            planner5.create_plan(desc, values)
        
        ranked = planner5.rank_plans()
        
        print("\nPlans Ranked by Weighted Score:")
        for i, (plan, score) in enumerate(ranked, 1):
            print(f"  {i}. {plan.description}")
            print(f"     Overall Score: {score:.3f}")
            print(f"     Innovation: {plan.objective_values['innovation']}, "
                  f"Feasibility: {plan.objective_values['feasibility']}, "
                  f"Risk: {plan.objective_values['risk']}")
        
        # Example 6: Priority-based planning
        print("\n" + "=" * 80)
        print("Example 6: Priority-Based Plan Generation")
        print("=" * 80)
        
        planner6 = MultiObjectivePlanner()
        
        planner6.add_objective("reliability", ObjectiveType.MAXIMIZE, weight=0.3)
        planner6.add_objective("performance", ObjectiveType.MAXIMIZE, weight=0.3)
        planner6.add_objective("maintainability", ObjectiveType.MAXIMIZE, weight=0.2)
        planner6.add_objective("cost", ObjectiveType.MINIMIZE, weight=0.2)
        
        print("\nGenerating plans with different priorities:")
        
        # Priority on reliability
        plan1 = planner6.generate_plan_with_priorities(
            "Build production system",
            ["reliability"]
        )
        print(f"\n  Priority: Reliability")
        print(f"  Plan: {plan1.description}")
        
        # Priority on performance
        plan2 = planner6.generate_plan_with_priorities(
            "Build production system",
            ["performance"]
        )
        print(f"\n  Priority: Performance")
        print(f"  Plan: {plan2.description}")
        
        # Example 7: Weight sensitivity
        print("\n" + "=" * 80)
        print("Example 7: Objective Weight Sensitivity")
        print("=" * 80)
        
        planner7 = MultiObjectivePlanner()
        
        obj1 = planner7.add_objective("feature_count", ObjectiveType.MAXIMIZE)
        obj2 = planner7.add_objective("simplicity", ObjectiveType.MAXIMIZE)
        
        test_plan = planner7.create_plan(
            "Product design",
            {"feature_count": 70, "simplicity": 60}
        )
        
        print("\nTesting different weight combinations:")
        
        weight_combos = [
            (0.8, 0.2, "Feature-focused"),
            (0.5, 0.5, "Balanced"),
            (0.2, 0.8, "Simplicity-focused"),
        ]
        
        for w1, w2, label in weight_combos:
            obj1.weight = w1
            obj2.weight = w2
            
            eval_result = planner7.evaluate_plan(test_plan)
            print(f"\n  {label} (weights: {w1:.1f}, {w2:.1f}):")
            print(f"    Overall Score: {eval_result['overall_score']:.3f}")
        
        # Example 8: System summary
        print("\n" + "=" * 80)
        print("Example 8: Planning System Summary")
        print("=" * 80)
        
        planner8 = MultiObjectivePlanner()
        
        planner8.add_objective("throughput", ObjectiveType.MAXIMIZE, weight=0.35)
        planner8.add_objective("latency", ObjectiveType.MINIMIZE, weight=0.35)
        planner8.add_objective("resource_usage", ObjectiveType.MINIMIZE, weight=0.30)
        
        # Create several plans
        for i in range(5):
            planner8.create_plan(
                f"Design option {i+1}",
                {
                    "throughput": 50 + (i * 10),
                    "latency": 100 - (i * 15),
                    "resource_usage": 40 + (i * 5)
                }
            )
        
        print(planner8.get_summary())
        
        # Summary
        print("\n" + "=" * 80)
        print("SUMMARY: Multi-Objective Planning Pattern")
        print("=" * 80)
        
        summary = """
        The Multi-Objective Planning pattern demonstrated:
        
        1. MULTIPLE OBJECTIVES (Example 1):
           - Cost, time, quality tradeoffs
           - Weighted objective evaluation
           - Overall score calculation
           - Alternative plan comparison
           - Objective value normalization
        
        2. PARETO OPTIMALITY (Example 2):
           - Non-dominated solutions
           - Domination checking
           - Efficient frontier identification
           - Multiple optimal alternatives
           - Tradeoff visualization
        
        3. TRADEOFF ANALYSIS (Example 3):
           - Comparing plan pairs
           - Identifying differences
           - Winner determination
           - Quantifying tradeoffs
           - Decision support
        
        4. HARD CONSTRAINTS (Example 4):
           - Constraint satisfaction
           - Feasibility checking
           - Plan filtering
           - Requirement enforcement
           - Infeasible plan rejection
        
        5. PLAN RANKING (Example 5):
           - Weighted overall scoring
           - Sorted alternatives
           - Best plan identification
           - Multi-criteria comparison
           - Decision making
        
        6. PRIORITY OBJECTIVES (Example 6):
           - LLM-guided plan generation
           - Priority emphasis
           - Context-aware planning
           - Objective focus
           - Adaptive generation
        
        7. WEIGHT SENSITIVITY (Example 7):
           - Dynamic weight adjustment
           - Impact analysis
           - Preference tuning
           - Score variation
           - Stakeholder alignment
        
        8. SYSTEM INTEGRATION (Example 8):
           - Comprehensive summary
           - Statistics tracking
           - Pareto analysis
           - System overview
           - Decision support
        
        KEY BENEFITS:
        ✓ Realistic complex decision modeling
        ✓ Explicit tradeoff consideration
        ✓ Multiple solution alternatives
        ✓ Stakeholder preference integration
        ✓ Better than single-objective
        ✓ Transparent decision criteria
        ✓ Flexible priority adjustment
        ✓ Pareto-optimal solutions
        
        USE CASES:
        • Resource allocation with constraints
        • Project planning (cost/time/quality)
        • Route planning (time/cost/comfort)
        • Product design tradeoffs
        • Investment decisions (risk/return)
        • Scheduling (efficiency/fairness)
        • ML model selection (accuracy/interpretability)
        • System design (performance/cost/reliability)
        
        OBJECTIVE TYPES:
        → Maximize: Quality, performance, revenue
        → Minimize: Cost, time, risk, resource usage
        → Target: Hit specific value
        → Constraint: Must satisfy (hard requirement)
        
        PLANNING APPROACHES:
        → Weighted Sum: Combine with weights
        → Lexicographic: Priority ordering
        → Pareto Optimal: Non-dominated set
        → Satisficing: Good enough
        → Goal Programming: Target achievement
        
        BEST PRACTICES:
        1. Define clear, measurable objectives
        2. Implement efficient Pareto filtering
        3. Provide tradeoff visualizations
        4. Allow dynamic weight adjustment
        5. Cache evaluated plans for reuse
        6. Support constraint hierarchies
        7. Enable sensitivity analysis
        8. Log decision rationale
        
        TRADE-OFFS:
        • Thoroughness vs. computational cost
        • Number of objectives vs. complexity
        • Weight precision vs. simplicity
        • Exact vs. approximate solutions
        
        PRODUCTION CONSIDERATIONS:
        → Normalize objectives to comparable scales
        → Implement efficient Pareto algorithms
        → Provide interactive weight tuning
        → Visualize tradeoff spaces (2D/3D plots)
        → Cache evaluation results
        → Support what-if analysis
        → Enable constraint relaxation
        → Log all evaluation details
        → Provide explanation of tradeoffs
        → Support group decision making
        
        This pattern enables agents to make realistic decisions when
        multiple conflicting objectives must be balanced, providing
        transparency into tradeoffs and multiple solution alternatives.
        """
        
        print(summary)
    
    
    if __name__ == "__main__":
        demonstrate_multi_objective_planning()
    
        

    Agentic Design Pattern: Multi Objective Planning

    Pattern 107: Multi Objective Planning – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Multi Objective Planning pattern implementation.


    Pattern Description

    Pattern 107: Multi-Objective Planning

    Description: The Multi-Objective Planning pattern enables agents to plan and make decisions when multiple, often conflicting objectives must be balanced. Instead of optimizing for a single goal, the agent must consider tradeoffs between competing objectives like cost vs. quality, speed vs. accuracy, or risk vs. reward.

    This pattern is essential for realistic decision-making where perfect solutions
    don't exist and every choice involves tradeoffs. The agent must evaluate plans
    across multiple dimensions, understand Pareto-optimal solutions, and make
    informed decisions about which tradeoffs are acceptable given the context.
    
    Multi-objective planning involves objective weighting, constraint satisfaction,
    Pareto optimization, preference elicitation, and tradeoff analysis. The agent
    can present multiple plan options with different tradeoff profiles, allowing
    humans or automated systems to select based on current priorities.
    

    Key Components: 1. Objective Set: Multiple goals to optimize 2. Objective Weights: Relative importance of each objective 3. Plan Evaluator: Scores plans across all objectives 4. Pareto Filter: Identifies non-dominated solutions 5. Tradeoff Analyzer: Analyzes competing objectives 6. Constraint Checker: Ensures hard constraints met 7. Preference Engine: Incorporates user preferences

    Objective Types: 1. Maximization: Quality, performance, completeness 2. Minimization: Cost, time, risk, resource usage 3. Target: Hit specific value (with tolerance) 4. Constraint: Must satisfy (hard requirement)

    Common Objective Combinations: – Cost vs. Quality vs. Speed (iron triangle) – Risk vs. Reward – Precision vs. Recall – Exploration vs. Exploitation – Short-term vs. Long-term goals – Individual vs. Group benefit

    Planning Approaches: 1. Weighted Sum: Combine objectives with weights 2. Lexicographic: Prioritize objectives in order 3. Pareto Optimal: Non-dominated solutions 4. Satisficing: Good enough on all objectives 5. Goal Programming: Meet targets with deviations 6. Multi-Criteria Decision Making (MCDM)

    Use Cases: – Resource allocation with constraints – Project planning with multiple stakeholders – Route planning (time vs. cost vs. comfort) – Product design (features vs. complexity vs. cost) – Investment decisions (risk vs. return) – Scheduling (efficiency vs. fairness) – Machine learning (accuracy vs. interpretability)

    Advantages: – Realistic modeling of complex decisions – Explicit tradeoff consideration – Multiple solution alternatives – Stakeholder preference incorporation – Better than single-objective optimization – Transparent decision criteria – Flexible priority adjustment

    Challenges: – Objective quantification difficulty – Weight elicitation from users – Computational complexity – Visualizing multi-dimensional tradeoffs – Handling conflicting constraints – Determining Pareto frontier – Explaining tradeoff decisions

    LangChain Implementation: This implementation uses LangChain for: – LLM-based objective identification – Plan evaluation across dimensions – Tradeoff analysis and explanation – Natural language objective specification

    Production Considerations: – Define clear, measurable objectives – Implement efficient Pareto filtering – Provide tradeoff visualizations – Allow dynamic weight adjustment – Cache evaluated plans – Support constraint hierarchies – Enable sensitivity analysis – Log decision rationale


    File Statistics

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

    Module Documentation

    """
    Pattern 107: Multi-Objective Planning
    
    Description:
        The Multi-Objective Planning pattern enables agents to plan and make decisions
        when multiple, often conflicting objectives must be balanced. Instead of
        optimizing for a single goal, the agent must consider tradeoffs between
        competing objectives like cost vs. quality, speed vs. accuracy, or risk vs.
        reward.
    
        This pattern is essential for realistic decision-making where perfect solutions
        don't exist and every choice involves tradeoffs. The agent must evaluate plans
        across multiple dimensions, understand Pareto-optimal solutions, and make
        informed decisions about which tradeoffs are acceptable given the context.
    
        Multi-objective planning involves objective weighting, constraint satisfaction,
        Pareto optimization, preference elicitation, and tradeoff analysis. The agent
        can present multiple plan options with different tradeoff profiles, allowing
        humans or automated systems to select based on current priorities.
    
    Key Components:
        1. Objective Set: Multiple goals to optimize
        2. Objective Weights: Relative importance of each objective
        3. Plan Evaluator: Scores plans across all objectives
        4. Pareto Filter: Identifies non-dominated solutions
        5. Tradeoff Analyzer: Analyzes competing objectives
        6. Constraint Checker: Ensures hard constraints met
        7. Preference Engine: Incorporates user preferences
    
    Objective Types:
        1. Maximization: Quality, performance, completeness
        2. Minimization: Cost, time, risk, resource usage
        3. Target: Hit specific value (with tolerance)
        4. Constraint: Must satisfy (hard requirement)
    
    Common Objective Combinations:
        - Cost vs. Quality vs. Speed (iron triangle)
        - Risk vs. Reward
        - Precision vs. Recall
        - Exploration vs. Exploitation
        - Short-term vs. Long-term goals
        - Individual vs. Group benefit
    
    Planning Approaches:
        1. Weighted Sum: Combine objectives with weights
        2. Lexicographic: Prioritize objectives in order
        3. Pareto Optimal: Non-dominated solutions
        4. Satisficing: Good enough on all objectives
        5. Goal Programming: Meet targets with deviations
        6. Multi-Criteria Decision Making (MCDM)
    
    Use Cases:
        - Resource allocation with constraints
        - Project planning with multiple stakeholders
        - Route planning (time vs. cost vs. comfort)
        - Product design (features vs. complexity vs. cost)
        - Investment decisions (risk vs. return)
        - Scheduling (efficiency vs. fairness)
        - Machine learning (accuracy vs. interpretability)
    
    Advantages:
        - Realistic modeling of complex decisions
        - Explicit tradeoff consideration
        - Multiple solution alternatives
        - Stakeholder preference incorporation
        - Better than single-objective optimization
        - Transparent decision criteria
        - Flexible priority adjustment
    
    Challenges:
        - Objective quantification difficulty
        - Weight elicitation from users
        - Computational complexity
        - Visualizing multi-dimensional tradeoffs
        - Handling conflicting constraints
        - Determining Pareto frontier
        - Explaining tradeoff decisions
    
    LangChain Implementation:
        This implementation uses LangChain for:
        - LLM-based objective identification
        - Plan evaluation across dimensions
        - Tradeoff analysis and explanation
        - Natural language objective specification
    
    Production Considerations:
        - Define clear, measurable objectives
        - Implement efficient Pareto filtering
        - Provide tradeoff visualizations
        - Allow dynamic weight adjustment
        - Cache evaluated plans
        - Support constraint hierarchies
        - Enable sensitivity analysis
        - Log decision rationale
    """
    

    Purpose: This module implements the Multi Objective Planning 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
    import typing.Tuple
    

    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) – 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: ObjectiveType

    Lines 110-115

    Inherits from: Enum

    Purpose:

    Type of optimization objective.
    

    Class: Objective

    Lines 119-169

    Purpose:

    Represents an optimization objective.
    
    Attributes:
        name: Objective name
        objective_type: Maximization, minimization, target, or constraint
        weight: Relative importance (0-1)
        target_value: Target value (for TARGET type)
        constraint_value: Threshold (for CONSTRAINT type)
        unit: Measurement unit
    

    Methods:

    evaluate(self, value)

    Lines 138-169

    Purpose:

    Evaluate objective given a value.
    
    Args:
        value: Actual value
    
    Returns:
        Normalized score (0-1), higher is better
    

    Parameters: – value: Value

    Returns: float


    Class: Plan

    Lines 173-193

    Purpose:

    Represents a plan with multiple objective values.
    
    Attributes:
        plan_id: Unique identifier
        description: Plan description
        objective_values: Values for each objective
        feasible: Whether all constraints are satisfied
        dominated: Whether dominated by another plan
    

    Methods:

    get_score(self, objective)

    Lines 190-193

    Purpose:

    Get normalized score for an objective.
    

    Parameters: – objective: Objective

    Returns: float


    Class: MultiObjectivePlanner

    Lines 196-499

    Purpose:

    Multi-objective planning system.
    
    This planner evaluates and compares plans across multiple objectives,
    identifies Pareto-optimal solutions, and analyzes tradeoffs.
    

    Methods:

    __init__(self, temperature)

    Lines 204-214

    Purpose:

    Initialize multi-objective planner.
    
    Args:
        temperature: LLM temperature
    

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

    add_objective(self, name, objective_type, weight, target_value, constraint_value, unit)

    Lines 216-248

    Purpose:

    Add an objective to optimize.
    
    Args:
        name: Objective name
        objective_type: Type of objective
        weight: Relative importance
        target_value: Target value (for TARGET)
        constraint_value: Constraint threshold
        unit: Measurement unit
    
    Returns:
        Created objective
    

    Parameters: – name: Name – objective_type: Objective type – weight: Weight – target_value: Target value – constraint_value: Constraint value – unit: Unit

    Returns: Objective

    create_plan(self, description, objective_values)

    Lines 250-284

    Purpose:

    Create a plan with objective values.
    
    Args:
        description: Plan description
        objective_values: Values for each objective
    
    Returns:
        Created plan
    

    Parameters: – description: Description – objective_values: Objective values

    Returns: Plan

    evaluate_plan(self, plan)

    Lines 286-316

    Purpose:

    Evaluate plan across all objectives.
    
    Args:
        plan: Plan to evaluate
    
    Returns:
        Evaluation results
    

    Parameters: – plan: Plan

    Returns: <ast.Subscript object at 0x000001A06D8107D0>

    find_pareto_optimal(self)

    Lines 318-358

    Purpose:

    Find Pareto-optimal (non-dominated) plans.
    
    Returns:
        List of Pareto-optimal plans
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06E094690>

    analyze_tradeoffs(self, plan1, plan2)

    Lines 360-404

    Purpose:

    Analyze tradeoffs between two plans.
    
    Args:
        plan1: First plan
        plan2: Second plan
    
    Returns:
        Tradeoff analysis
    

    Parameters: – plan1: Plan1 – plan2: Plan2

    Returns: <ast.Subscript object at 0x000001A06D659890>

    generate_plan_with_priorities(self, task, priority_objectives)

    Lines 406-457

    Purpose:

    Generate plan optimizing for priority objectives.
    
    Args:
        task: Task description
        priority_objectives: High-priority objective names
    
    Returns:
        Generated plan
    

    Parameters: – task: Task – priority_objectives: Priority objectives

    Returns: Plan

    rank_plans(self)

    Lines 459-476

    Purpose:

    Rank plans by weighted overall score.
    
    Returns:
        List of (plan, score) tuples, sorted by score
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D7FF210>

    get_summary(self)

    Lines 478-499

    Purpose:

    Get summary of plans and objectives.
    

    Parameters:

    Returns: str


    Functions

    demonstrate_multi_objective_planning()

    Lines 502-894

    Purpose:

    Demonstrate multi-objective planning pattern.
    

    Key Concepts Summary

    Pattern Overview

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

    Core Components

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

    When to Use

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

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


    Best Practices Demonstrated

    1. Environment Configuration

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

    2. Error Handling

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

    3. Prompt Engineering

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

    4. Code Organization

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

    5. Performance Optimization

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

    LangChain Components Used

    • ChatOpenAI: OpenAI model interface
    • Chains: Sequential component composition

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


    Execution Flow

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

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


    Performance Considerations

    Computational Cost

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

    Optimization Strategies

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

    Cost Estimates (Approximate)

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

    Actual costs vary based on input/output token counts


    Common Pitfalls to Avoid

    1. Prompt Design Issues

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

    2. Error Handling

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

    3. Token Management

    ❌ Exceeding model token limits
    ✅ Truncate inputs, chunk processing

    4. Security

    ❌ Exposing API keys in code
    ✅ Use environment variables

    5. Testing

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

    6. Cost Management

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


    Further Improvements

    Production Enhancements

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

    Feature Extensions

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

    Code Quality

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

    Related Patterns

    • Pattern 106
    • Pattern 108

    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 Multi Objective Planning pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

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

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

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


    File: 107_multi_objective_planning.py
    Total Lines: 899
    Pattern Category: AI Agent Design Pattern #107


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Contingency Planning

    
        """
    Pattern 106: Contingency Planning
    
    Description:
        The Contingency Planning pattern prepares agents to handle unexpected events
        and failures by creating alternative plans before they're needed. This pattern
        involves anticipating potential problems, creating backup plans, and enabling
        quick adaptation when things don't go as expected.
        
        Contingency planning is crucial for reliable agent systems that operate in
        uncertain, dynamic environments. Rather than reacting to failures after they
        occur, agents proactively identify risks, assess their likelihood and impact,
        and prepare alternative strategies. This reduces downtime, improves robustness,
        and enables graceful degradation when primary plans fail.
        
        The pattern includes risk identification, scenario planning, plan B/C/D
        creation, trigger condition specification, and rapid plan switching. It
        balances the cost of planning for contingencies against the benefit of
        being prepared for failures.
    
    Key Components:
        1. Risk Identifier: Detects potential failure points
        2. Scenario Generator: Creates alternative scenarios
        3. Contingency Creator: Develops backup plans
        4. Trigger Monitor: Watches for contingency conditions
        5. Plan Switcher: Activates alternative plans
        6. Impact Assessor: Evaluates risk severity
        7. Recovery Planner: Plans return to primary plan
    
    Contingency Types:
        1. Alternative Action: Different way to achieve same goal
        2. Fallback Goal: Reduced/modified objective
        3. Resource Substitution: Use different resources
        4. Path Alternative: Different sequence of steps
        5. Graceful Degradation: Accept reduced functionality
        6. Abort and Recover: Abandon task, clean up
    
    Risk Assessment:
        1. Likelihood: Probability of occurrence (low/medium/high)
        2. Impact: Severity of consequences (minor/moderate/critical)
        3. Priority: Likelihood × Impact
        4. Mitigation Cost: Resources to prepare contingency
        
    Planning Phases:
        1. Risk Identification: What could go wrong?
        2. Impact Analysis: How bad would it be?
        3. Contingency Development: What's plan B?
        4. Trigger Specification: When to switch plans?
        5. Monitoring: Watch for triggers
        6. Activation: Switch to contingency
        7. Recovery: Return to normal operations
    
    Use Cases:
        - Mission-critical operations
        - Uncertain environments
        - Resource-constrained systems
        - High-stakes decision making
        - Autonomous systems
        - Real-time operations
        - Safety-critical applications
    
    Advantages:
        - Reduces response time to failures
        - Improves system reliability
        - Enables graceful degradation
        - Proactive vs reactive approach
        - Builds resilience
        - Reduces costly downtime
        - Increases confidence
    
    Challenges:
        - Computational cost of planning
        - Predicting all possible failures
        - Balancing thoroughness vs. efficiency
        - Maintaining multiple plans
        - Detecting when to switch
        - Avoiding over-planning
        - Resource allocation for contingencies
    
    LangChain Implementation:
        This implementation uses LangChain for:
        - LLM-based risk identification
        - Alternative plan generation
        - Scenario analysis
        - Natural language contingency description
        
    Production Considerations:
        - Focus on high-impact, high-probability risks
        - Use tiered contingency levels (B, C, D plans)
        - Implement fast plan switching
        - Monitor trigger conditions efficiently
        - Test contingency plans regularly
        - Document all contingencies
        - Balance planning cost vs. benefit
        - Enable human override when needed
    """
    
    import os
    from typing import List, Dict, Any, Optional
    from dataclasses import dataclass, field
    from datetime import datetime
    from enum import Enum
    from dotenv import load_dotenv
    
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser
    from langchain_openai import ChatOpenAI
    
    load_dotenv()
    
    
    class RiskLevel(Enum):
        """Risk likelihood levels."""
        LOW = "low"
        MEDIUM = "medium"
        HIGH = "high"
    
    
    class ImpactLevel(Enum):
        """Impact severity levels."""
        MINOR = "minor"
        MODERATE = "moderate"
        CRITICAL = "critical"
    
    
    class PlanStatus(Enum):
        """Status of a plan."""
        PRIMARY = "primary"
        CONTINGENCY = "contingency"
        ACTIVE = "active"
        FAILED = "failed"
        SUCCEEDED = "succeeded"
    
    
    @dataclass
    class Risk:
        """
        Represents a potential risk or failure point.
        
        Attributes:
            risk_id: Unique identifier
            description: Risk description
            likelihood: Probability of occurrence
            impact: Severity of consequences
            trigger_condition: Condition that indicates risk occurred
            affected_steps: Plan steps affected by this risk
        """
        risk_id: str
        description: str
        likelihood: RiskLevel
        impact: ImpactLevel
        trigger_condition: str
        affected_steps: List[str] = field(default_factory=list)
        
        @property
        def priority(self) -> int:
            """Calculate risk priority."""
            likelihood_scores = {RiskLevel.LOW: 1, RiskLevel.MEDIUM: 2, RiskLevel.HIGH: 3}
            impact_scores = {ImpactLevel.MINOR: 1, ImpactLevel.MODERATE: 2, ImpactLevel.CRITICAL: 3}
            return likelihood_scores[self.likelihood] * impact_scores[self.impact]
    
    
    @dataclass
    class Plan:
        """
        Represents a plan (primary or contingency).
        
        Attributes:
            plan_id: Unique identifier
            description: Plan description
            steps: List of action steps
            status: Plan status
            parent_plan_id: ID of plan this is contingency for
            triggers: Conditions that activate this plan
            cost: Estimated cost/resources
            success_criteria: How to know if plan succeeded
        """
        plan_id: str
        description: str
        steps: List[str]
        status: PlanStatus
        parent_plan_id: Optional[str] = None
        triggers: List[str] = field(default_factory=list)
        cost: float = 1.0
        success_criteria: List[str] = field(default_factory=list)
        activated_at: Optional[datetime] = None
        completed_at: Optional[datetime] = None
    
    
    class ContingencyPlanner:
        """
        Contingency planning system that creates and manages backup plans.
        
        This planner identifies risks, creates alternative plans, monitors
        for trigger conditions, and switches to contingencies when needed.
        """
        
        def __init__(self, temperature: float = 0.4):
            """
            Initialize contingency planner.
            
            Args:
                temperature: LLM temperature
            """
            self.llm = ChatOpenAI(temperature=temperature, model="gpt-3.5-turbo")
            self.plans: Dict[str, Plan] = {}
            self.risks: Dict[str, Risk] = {}
            self.active_plan_id: Optional[str] = None
            self.plan_counter = 0
            self.risk_counter = 0
            self.event_log: List[Dict[str, Any]] = []
        
        def create_primary_plan(
            self,
            description: str,
            steps: List[str]
        ) -> Plan:
            """
            Create primary plan.
            
            Args:
                description: Plan description
                steps: Plan steps
                
            Returns:
                Created plan
            """
            self.plan_counter += 1
            plan = Plan(
                plan_id=f"plan_{self.plan_counter}",
                description=description,
                steps=steps,
                status=PlanStatus.PRIMARY
            )
            
            self.plans[plan.plan_id] = plan
            self.active_plan_id = plan.plan_id
            
            return plan
        
        def identify_risks(self, plan: Plan) -> List[Risk]:
            """
            Identify potential risks in a plan using LLM.
            
            Args:
                plan: Plan to analyze
                
            Returns:
                List of identified risks
            """
            prompt = ChatPromptTemplate.from_template(
                "Analyze this plan and identify 3-5 potential risks or failure points:\n\n"
                "Plan: {description}\n"
                "Steps:\n{steps}\n\n"
                "For each risk, provide:\n"
                "- Description of what could go wrong\n"
                "- Likelihood (low/medium/high)\n"
                "- Impact (minor/moderate/critical)\n"
                "- Trigger condition (how to detect it happened)\n\n"
                "Format each risk as:\n"
                "Risk: [description]\n"
                "Likelihood: [level]\n"
                "Impact: [level]\n"
                "Trigger: [condition]\n"
                "---\n"
            )
            
            steps_text = "\n".join(f"{i+1}. {step}" for i, step in enumerate(plan.steps))
            
            chain = prompt | self.llm | StrOutputParser()
            result = chain.invoke({
                "description": plan.description,
                "steps": steps_text
            })
            
            # Parse risks from LLM output
            risks = []
            risk_blocks = result.split("---")
            
            for block in risk_blocks:
                if not block.strip():
                    continue
                
                lines = [line.strip() for line in block.strip().split('\n')]
                risk_desc = ""
                likelihood = RiskLevel.MEDIUM
                impact = ImpactLevel.MODERATE
                trigger = ""
                
                for line in lines:
                    if line.startswith("Risk:"):
                        risk_desc = line[5:].strip()
                    elif line.startswith("Likelihood:"):
                        level = line[11:].strip().lower()
                        likelihood = RiskLevel(level) if level in [l.value for l in RiskLevel] else RiskLevel.MEDIUM
                    elif line.startswith("Impact:"):
                        level = line[7:].strip().lower()
                        impact = ImpactLevel(level) if level in [i.value for i in ImpactLevel] else ImpactLevel.MODERATE
                    elif line.startswith("Trigger:"):
                        trigger = line[8:].strip()
                
                if risk_desc and trigger:
                    self.risk_counter += 1
                    risk = Risk(
                        risk_id=f"risk_{self.risk_counter}",
                        description=risk_desc,
                        likelihood=likelihood,
                        impact=impact,
                        trigger_condition=trigger
                    )
                    self.risks[risk.risk_id] = risk
                    risks.append(risk)
            
            return risks
        
        def create_contingency_plan(
            self,
            primary_plan: Plan,
            risk: Risk,
            level: str = "B"
        ) -> Plan:
            """
            Create contingency plan for a specific risk.
            
            Args:
                primary_plan: Primary plan
                risk: Risk to mitigate
                level: Contingency level (B, C, D, etc.)
                
            Returns:
                Contingency plan
            """
            prompt = ChatPromptTemplate.from_template(
                "Create a contingency plan (Plan {level}) for this situation:\n\n"
                "Primary Plan: {plan_description}\n"
                "Risk: {risk_description}\n"
                "Trigger: {trigger}\n\n"
                "Design an alternative plan that:\n"
                "1. Achieves the same or similar goal\n"
                "2. Avoids or mitigates the identified risk\n"
                "3. Is practical and feasible\n\n"
                "Provide 3-5 concrete steps for the contingency plan.\n"
                "List each step on a new line, numbered.\n\n"
                "Steps:"
            )
            
            chain = prompt | self.llm | StrOutputParser()
            result = chain.invoke({
                "level": level,
                "plan_description": primary_plan.description,
                "risk_description": risk.description,
                "trigger": risk.trigger_condition
            })
            
            # Parse steps
            lines = [line.strip() for line in result.strip().split('\n') if line.strip()]
            steps = []
            
            for line in lines:
                # Remove numbering
                step = line
                for prefix in ["1.", "2.", "3.", "4.", "5.", "1)", "2)", "3)", "4)", "5)", "-", "*"]:
                    if step.startswith(prefix):
                        step = step[len(prefix):].strip()
                        break
                if step:
                    steps.append(step)
            
            # Create contingency plan
            self.plan_counter += 1
            contingency = Plan(
                plan_id=f"plan_{self.plan_counter}",
                description=f"Plan {level}: {primary_plan.description} (contingency for {risk.description})",
                steps=steps,
                status=PlanStatus.CONTINGENCY,
                parent_plan_id=primary_plan.plan_id,
                triggers=[risk.trigger_condition]
            )
            
            self.plans[contingency.plan_id] = contingency
            
            return contingency
        
        def create_tiered_contingencies(
            self,
            primary_plan: Plan,
            num_tiers: int = 2
        ) -> List[Plan]:
            """
            Create multiple tiers of contingency plans.
            
            Args:
                primary_plan: Primary plan
                num_tiers: Number of contingency tiers
                
            Returns:
                List of contingency plans
            """
            # Identify risks
            risks = self.identify_risks(primary_plan)
            
            # Sort by priority
            risks.sort(key=lambda r: r.priority, reverse=True)
            
            # Create contingencies for top risks
            contingencies = []
            levels = ["B", "C", "D", "E", "F"]
            
            for i in range(min(num_tiers, len(risks), len(levels))):
                contingency = self.create_contingency_plan(
                    primary_plan,
                    risks[i],
                    level=levels[i]
                )
                contingencies.append(contingency)
            
            return contingencies
        
        def check_trigger_conditions(
            self,
            observed_events: List[str]
        ) -> Optional[Plan]:
            """
            Check if any contingency should be activated.
            
            Args:
                observed_events: List of observed events/conditions
                
            Returns:
                Contingency plan to activate, or None
            """
            # Check contingency plans for matching triggers
            contingency_plans = [
                p for p in self.plans.values()
                if p.status == PlanStatus.CONTINGENCY
            ]
            
            for plan in contingency_plans:
                for trigger in plan.triggers:
                    for event in observed_events:
                        if trigger.lower() in event.lower() or event.lower() in trigger.lower():
                            return plan
            
            return None
        
        def activate_contingency(
            self,
            contingency_plan: Plan,
            reason: str
        ):
            """
            Activate a contingency plan.
            
            Args:
                contingency_plan: Contingency to activate
                reason: Reason for activation
            """
            # Deactivate current plan
            if self.active_plan_id and self.active_plan_id in self.plans:
                self.plans[self.active_plan_id].status = PlanStatus.FAILED
            
            # Activate contingency
            contingency_plan.status = PlanStatus.ACTIVE
            contingency_plan.activated_at = datetime.now()
            self.active_plan_id = contingency_plan.plan_id
            
            # Log event
            self.event_log.append({
                "timestamp": datetime.now(),
                "event": "contingency_activated",
                "plan_id": contingency_plan.plan_id,
                "reason": reason
            })
        
        def simulate_execution(
            self,
            plan: Plan,
            simulate_failure: Optional[str] = None
        ) -> Dict[str, Any]:
            """
            Simulate plan execution.
            
            Args:
                plan: Plan to execute
                simulate_failure: Simulate failure at this step
                
            Returns:
                Execution result
            """
            results = []
            
            for i, step in enumerate(plan.steps):
                # Simulate failure
                if simulate_failure and simulate_failure.lower() in step.lower():
                    return {
                        "success": False,
                        "completed_steps": results,
                        "failed_at": step,
                        "reason": f"Simulated failure at: {step}"
                    }
                
                # Successful step
                results.append({
                    "step": step,
                    "status": "completed",
                    "timestamp": datetime.now()
                })
            
            plan.status = PlanStatus.SUCCEEDED
            plan.completed_at = datetime.now()
            
            return {
                "success": True,
                "completed_steps": results,
                "plan_id": plan.plan_id
            }
        
        def get_plan_summary(self) -> str:
            """Get summary of all plans."""
            lines = []
            
            # Primary plans
            primary = [p for p in self.plans.values() if p.status in [PlanStatus.PRIMARY, PlanStatus.ACTIVE]]
            if primary:
                lines.append("PRIMARY PLAN:")
                for plan in primary:
                    active = " (ACTIVE)" if plan.plan_id == self.active_plan_id else ""
                    lines.append(f"  {plan.plan_id}{active}: {plan.description}")
                    for i, step in enumerate(plan.steps, 1):
                        lines.append(f"    {i}. {step}")
            
            # Contingencies
            contingencies = [p for p in self.plans.values() if p.status == PlanStatus.CONTINGENCY]
            if contingencies:
                lines.append("\nCONTINGENCY PLANS:")
                for plan in contingencies:
                    lines.append(f"  {plan.plan_id}: {plan.description}")
                    if plan.triggers:
                        lines.append(f"    Triggers: {', '.join(plan.triggers[:2])}...")
                    for i, step in enumerate(plan.steps, 1):
                        lines.append(f"    {i}. {step}")
            
            return "\n".join(lines)
        
        def get_risk_summary(self) -> str:
            """Get summary of identified risks."""
            lines = ["IDENTIFIED RISKS (by priority):"]
            
            risks = sorted(self.risks.values(), key=lambda r: r.priority, reverse=True)
            
            for risk in risks:
                priority = risk.priority
                lines.append(
                    f"\n  [{priority}] {risk.description}\n"
                    f"      Likelihood: {risk.likelihood.value}, Impact: {risk.impact.value}\n"
                    f"      Trigger: {risk.trigger_condition}"
                )
            
            return "\n".join(lines)
    
    
    def demonstrate_contingency_planning():
        """Demonstrate contingency planning pattern."""
        
        print("=" * 80)
        print("CONTINGENCY PLANNING PATTERN DEMONSTRATION")
        print("=" * 80)
        
        # Example 1: Basic contingency planning
        print("\n" + "=" * 80)
        print("Example 1: Creating Primary and Contingency Plans")
        print("=" * 80)
        
        planner = ContingencyPlanner()
        
        # Create primary plan
        primary = planner.create_primary_plan(
            "Deploy new software version",
            [
                "Run automated tests",
                "Build production package",
                "Deploy to staging",
                "Run smoke tests",
                "Deploy to production"
            ]
        )
        
        print("\nPrimary Plan:")
        print(f"  {primary.description}")
        for i, step in enumerate(primary.steps, 1):
            print(f"  {i}. {step}")
        
        print("\nIdentifying risks...")
        risks = planner.identify_risks(primary)
        
        print(f"\nIdentified {len(risks)} risks:")
        for risk in risks:
            print(f"  • {risk.description}")
            print(f"    Likelihood: {risk.likelihood.value}, Impact: {risk.impact.value}")
            print(f"    Priority: {risk.priority}")
        
        # Create contingency for highest priority risk
        if risks:
            top_risk = max(risks, key=lambda r: r.priority)
            contingency = planner.create_contingency_plan(primary, top_risk, "B")
            
            print(f"\nContingency Plan B (for: {top_risk.description}):")
            for i, step in enumerate(contingency.steps, 1):
                print(f"  {i}. {step}")
        
        # Example 2: Tiered contingencies
        print("\n" + "=" * 80)
        print("Example 2: Multi-Tier Contingency Planning")
        print("=" * 80)
        
        planner2 = ContingencyPlanner()
        
        primary2 = planner2.create_primary_plan(
            "Complete important presentation",
            [
                "Prepare slides",
                "Rehearse presentation",
                "Set up projector",
                "Deliver presentation"
            ]
        )
        
        print("\nCreating tiered contingencies...")
        contingencies = planner2.create_tiered_contingencies(primary2, num_tiers=3)
        
        print(f"\nCreated {len(contingencies)} contingency tiers")
        print(planner2.get_plan_summary())
        
        # Example 3: Trigger detection and activation
        print("\n" + "=" * 80)
        print("Example 3: Contingency Activation")
        print("=" * 80)
        
        planner3 = ContingencyPlanner()
        
        primary3 = planner3.create_primary_plan(
            "Travel to meeting",
            [
                "Drive to meeting location",
                "Park car",
                "Enter building"
            ]
        )
        
        # Create simple contingency
        planner3.risk_counter += 1
        traffic_risk = Risk(
            risk_id=f"risk_{planner3.risk_counter}",
            description="Heavy traffic delays arrival",
            likelihood=RiskLevel.MEDIUM,
            impact=ImpactLevel.MODERATE,
            trigger_condition="traffic jam detected"
        )
        planner3.risks[traffic_risk.risk_id] = traffic_risk
        
        contingency3 = planner3.create_contingency_plan(primary3, traffic_risk, "B")
        
        print("\nPrimary Plan:")
        print(f"  {primary3.description}")
        
        print("\nContingency Plan B:")
        print(f"  {contingency3.description}")
        
        print("\nSimulating execution with problem...")
        observed_events = ["traffic jam detected", "delay expected"]
        
        triggered_plan = planner3.check_trigger_conditions(observed_events)
        
        if triggered_plan:
            print(f"\n⚠️  Trigger detected! Activating: {triggered_plan.plan_id}")
            planner3.activate_contingency(triggered_plan, "Traffic jam encountered")
            print(f"  Now executing: {triggered_plan.description}")
        
        # Example 4: Risk prioritization
        print("\n" + "=" * 80)
        print("Example 4: Risk Assessment and Prioritization")
        print("=" * 80)
        
        planner4 = ContingencyPlanner()
        
        primary4 = planner4.create_primary_plan(
            "Launch product campaign",
            [
                "Finalize marketing materials",
                "Schedule social media posts",
                "Send press releases",
                "Monitor campaign metrics"
            ]
        )
        
        print("\nAnalyzing risks...")
        risks4 = planner4.identify_risks(primary4)
        
        print(planner4.get_risk_summary())
        
        # Example 5: Execution with fallback
        print("\n" + "=" * 80)
        print("Example 5: Execution with Automatic Fallback")
        print("=" * 80)
        
        planner5 = ContingencyPlanner()
        
        primary5 = planner5.create_primary_plan(
            "Process data file",
            [
                "Download file from server",
                "Parse file contents",
                "Validate data",
                "Store in database"
            ]
        )
        
        # Create contingency
        planner5.risk_counter += 1
        download_risk = Risk(
            risk_id=f"risk_{planner5.risk_counter}",
            description="File download fails",
            likelihood=RiskLevel.MEDIUM,
            impact=ImpactLevel.MODERATE,
            trigger_condition="download fails"
        )
        
        contingency5 = Plan(
            plan_id="plan_backup",
            description="Plan B: Use cached file",
            steps=[
                "Retrieve file from cache",
                "Parse file contents",
                "Validate data",
                "Store in database"
            ],
            status=PlanStatus.CONTINGENCY,
            parent_plan_id=primary5.plan_id,
            triggers=["download fails"]
        )
        planner5.plans[contingency5.plan_id] = contingency5
        
        print("\nAttempting primary plan...")
        result = planner5.simulate_execution(primary5, simulate_failure="download")
        
        if not result["success"]:
            print(f"❌ Primary plan failed: {result['reason']}")
            
            # Check for contingency
            events = ["download fails"]
            triggered = planner5.check_trigger_conditions(events)
            
            if triggered:
                print(f"\n✓ Activating contingency: {triggered.plan_id}")
                planner5.activate_contingency(triggered, result["reason"])
                
                print("\nAttempting contingency plan...")
                result2 = planner5.simulate_execution(triggered)
                
                if result2["success"]:
                    print(f"✅ Contingency plan succeeded!")
        
        # Example 6: Graceful degradation
        print("\n" + "=" * 80)
        print("Example 6: Graceful Degradation Strategy")
        print("=" * 80)
        
        planner6 = ContingencyPlanner()
        
        primary6 = planner6.create_primary_plan(
            "Generate comprehensive report",
            [
                "Collect data from all sources",
                "Perform advanced analytics",
                "Create interactive visualizations",
                "Generate PDF report"
            ]
        )
        
        # Degraded plan
        degraded = Plan(
            plan_id="plan_degraded",
            description="Plan B: Generate basic report (degraded)",
            steps=[
                "Collect data from primary source only",
                "Perform basic statistics",
                "Create simple charts",
                "Generate text report"
            ],
            status=PlanStatus.CONTINGENCY,
            parent_plan_id=primary6.plan_id,
            triggers=["data source unavailable", "time constraint"]
        )
        planner6.plans[degraded.plan_id] = degraded
        
        print("\nPrimary Plan (Full Functionality):")
        for i, step in enumerate(primary6.steps, 1):
            print(f"  {i}. {step}")
        
        print("\nDegraded Plan (Reduced Functionality):")
        for i, step in enumerate(degraded.steps, 1):
            print(f"  {i}. {step}")
        
        print("\nStrategy: Accept reduced functionality to meet deadline")
        
        # Example 7: Multiple contingency levels
        print("\n" + "=" * 80)
        print("Example 7: Cascade of Contingencies")
        print("=" * 80)
        
        planner7 = ContingencyPlanner()
        
        primary7 = planner7.create_primary_plan(
            "Connect to service",
            ["Connect to primary server"]
        )
        
        plan_b = Plan(
            plan_id="plan_b",
            description="Plan B: Connect to backup server",
            steps=["Connect to backup server"],
            status=PlanStatus.CONTINGENCY,
            parent_plan_id=primary7.plan_id,
            triggers=["primary server down"]
        )
        
        plan_c = Plan(
            plan_id="plan_c",
            description="Plan C: Use cached data",
            steps=["Load cached data from local storage"],
            status=PlanStatus.CONTINGENCY,
            parent_plan_id=plan_b.plan_id,
            triggers=["backup server down"]
        )
        
        plan_d = Plan(
            plan_id="plan_d",
            description="Plan D: Operate in offline mode",
            steps=["Enable offline mode", "Queue operations"],
            status=PlanStatus.CONTINGENCY,
            parent_plan_id=plan_c.plan_id,
            triggers=["no cached data"]
        )
        
        planner7.plans.update({
            plan_b.plan_id: plan_b,
            plan_c.plan_id: plan_c,
            plan_d.plan_id: plan_d
        })
        
        print("\nContingency Cascade:")
        print("  Primary: Connect to primary server")
        print("  Plan B: Connect to backup server (if primary fails)")
        print("  Plan C: Use cached data (if backup fails)")
        print("  Plan D: Operate offline (if no cache)")
        print("\nStrategy: Progressive degradation with multiple fallback levels")
        
        # Example 8: Contingency metrics
        print("\n" + "=" * 80)
        print("Example 8: Contingency Planning Metrics")
        print("=" * 80)
        
        planner8 = ContingencyPlanner()
        
        primary8 = planner8.create_primary_plan(
            "Complete project milestone",
            [
                "Finish feature implementation",
                "Write documentation",
                "Conduct code review",
                "Merge to main branch"
            ]
        )
        
        # Create multiple contingencies
        planner8.create_tiered_contingencies(primary8, num_tiers=3)
        
        print("\nContingency Planning Statistics:")
        print(f"  Total Plans: {len(planner8.plans)}")
        print(f"  Primary Plans: {sum(1 for p in planner8.plans.values() if p.status == PlanStatus.PRIMARY)}")
        print(f"  Contingency Plans: {sum(1 for p in planner8.plans.values() if p.status == PlanStatus.CONTINGENCY)}")
        print(f"  Identified Risks: {len(planner8.risks)}")
        print(f"  High-Priority Risks: {sum(1 for r in planner8.risks.values() if r.priority >= 6)}")
        
        # Summary
        print("\n" + "=" * 80)
        print("SUMMARY: Contingency Planning Pattern")
        print("=" * 80)
        
        summary = """
        The Contingency Planning pattern demonstrated:
        
        1. BASIC CONTINGENCIES (Example 1):
           - Risk identification using LLM
           - Likelihood and impact assessment
           - Priority calculation
           - Alternative plan creation
           - Trigger condition specification
        
        2. TIERED PLANNING (Example 2):
           - Multiple contingency levels (B, C, D)
           - Risk-based prioritization
           - Comprehensive backup strategies
           - Structured alternatives
        
        3. TRIGGER ACTIVATION (Example 3):
           - Real-time monitoring
           - Condition matching
           - Automatic plan switching
           - Event-driven adaptation
           - Seamless transition
        
        4. RISK PRIORITIZATION (Example 4):
           - Priority scoring (likelihood × impact)
           - Risk ranking
           - Resource allocation guidance
           - Focus on critical risks
        
        5. AUTOMATIC FALLBACK (Example 5):
           - Failure detection
           - Contingency activation
           - Execution retry
           - Success tracking
           - Resilient operation
        
        6. GRACEFUL DEGRADATION (Example 6):
           - Reduced functionality plans
           - Partial goal achievement
           - Quality-time tradeoffs
           - Acceptable alternatives
        
        7. CONTINGENCY CASCADE (Example 7):
           - Multi-level fallbacks
           - Progressive degradation
           - Chain of alternatives
           - Comprehensive coverage
        
        8. METRICS TRACKING (Example 8):
           - Plan statistics
           - Risk counts
           - Priority distribution
           - Coverage assessment
        
        KEY BENEFITS:
        ✓ Reduces response time to failures
        ✓ Improves system reliability
        ✓ Enables graceful degradation
        ✓ Proactive preparation
        ✓ Builds resilience
        ✓ Reduces costly downtime
        ✓ Increases confidence
        ✓ Supports high-stakes operations
        
        USE CASES:
        • Mission-critical operations
        • Uncertain environments
        • Resource-constrained systems
        • High-stakes decision making
        • Autonomous systems
        • Real-time operations
        • Safety-critical applications
        • Production deployments
        
        CONTINGENCY TYPES:
        → Alternative Action: Different approach
        → Fallback Goal: Modified objective
        → Resource Substitution: Different resources
        → Path Alternative: Different sequence
        → Graceful Degradation: Reduced functionality
        → Abort and Recover: Clean exit
        
        BEST PRACTICES:
        1. Focus on high-impact, high-probability risks
        2. Use tiered contingency levels (B, C, D)
        3. Implement fast plan switching
        4. Monitor trigger conditions efficiently
        5. Test contingency plans regularly
        6. Document all contingencies clearly
        7. Balance planning cost vs. benefit
        8. Enable human override capability
        
        TRADE-OFFS:
        • Planning cost vs. preparedness benefit
        • Coverage thoroughness vs. efficiency
        • Multiple plans vs. maintenance burden
        • Proactive vs. reactive approach
        
        PRODUCTION CONSIDERATIONS:
        → Implement efficient trigger monitoring
        → Cache contingency plans for quick access
        → Test contingencies in simulation
        → Monitor plan activation frequency
        → Track contingency success rates
        → Update plans based on experience
        → Provide clear plan switching notifications
        → Enable rollback to primary plan
        → Log all plan transitions
        → Support human decision overrides
        
        This pattern enables agents to handle unexpected failures
        gracefully by preparing alternative plans before they're needed,
        monitoring for trigger conditions, and switching quickly when
        problems occur.
        """
        
        print(summary)
    
    
    if __name__ == "__main__":
        demonstrate_contingency_planning()
    
        

    Agentic Design Pattern: Contingency Planning

    Pattern 106: Contingency Planning – Line-by-Line Explanation

    Overview

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


    Pattern Description

    Pattern 106: Contingency Planning

    Description: The Contingency Planning pattern prepares agents to handle unexpected events and failures by creating alternative plans before they’re needed. This pattern involves anticipating potential problems, creating backup plans, and enabling quick adaptation when things don’t go as expected.

    Contingency planning is crucial for reliable agent systems that operate in
    uncertain, dynamic environments. Rather than reacting to failures after they
    occur, agents proactively identify risks, assess their likelihood and impact,
    and prepare alternative strategies. This reduces downtime, improves robustness,
    and enables graceful degradation when primary plans fail.
    
    The pattern includes risk identification, scenario planning, plan B/C/D
    creation, trigger condition specification, and rapid plan switching. It
    balances the cost of planning for contingencies against the benefit of
    being prepared for failures.
    

    Key Components: 1. Risk Identifier: Detects potential failure points 2. Scenario Generator: Creates alternative scenarios 3. Contingency Creator: Develops backup plans 4. Trigger Monitor: Watches for contingency conditions 5. Plan Switcher: Activates alternative plans 6. Impact Assessor: Evaluates risk severity 7. Recovery Planner: Plans return to primary plan

    Contingency Types: 1. Alternative Action: Different way to achieve same goal 2. Fallback Goal: Reduced/modified objective 3. Resource Substitution: Use different resources 4. Path Alternative: Different sequence of steps 5. Graceful Degradation: Accept reduced functionality 6. Abort and Recover: Abandon task, clean up

    Risk Assessment: 1. Likelihood: Probability of occurrence (low/medium/high) 2. Impact: Severity of consequences (minor/moderate/critical) 3. Priority: Likelihood × Impact 4. Mitigation Cost: Resources to prepare contingency

    Planning Phases: 1. Risk Identification: What could go wrong? 2. Impact Analysis: How bad would it be? 3. Contingency Development: What’s plan B? 4. Trigger Specification: When to switch plans? 5. Monitoring: Watch for triggers 6. Activation: Switch to contingency 7. Recovery: Return to normal operations

    Use Cases: – Mission-critical operations – Uncertain environments – Resource-constrained systems – High-stakes decision making – Autonomous systems – Real-time operations – Safety-critical applications

    Advantages: – Reduces response time to failures – Improves system reliability – Enables graceful degradation – Proactive vs reactive approach – Builds resilience – Reduces costly downtime – Increases confidence

    Challenges: – Computational cost of planning – Predicting all possible failures – Balancing thoroughness vs. efficiency – Maintaining multiple plans – Detecting when to switch – Avoiding over-planning – Resource allocation for contingencies

    LangChain Implementation: This implementation uses LangChain for: – LLM-based risk identification – Alternative plan generation – Scenario analysis – Natural language contingency description

    Production Considerations: – Focus on high-impact, high-probability risks – Use tiered contingency levels (B, C, D plans) – Implement fast plan switching – Monitor trigger conditions efficiently – Test contingency plans regularly – Document all contingencies – Balance planning cost vs. benefit – Enable human override when needed


    File Statistics

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

    Module Documentation

    """
    Pattern 106: Contingency Planning
    
    Description:
        The Contingency Planning pattern prepares agents to handle unexpected events
        and failures by creating alternative plans before they're needed. This pattern
        involves anticipating potential problems, creating backup plans, and enabling
        quick adaptation when things don't go as expected.
    
        Contingency planning is crucial for reliable agent systems that operate in
        uncertain, dynamic environments. Rather than reacting to failures after they
        occur, agents proactively identify risks, assess their likelihood and impact,
        and prepare alternative strategies. This reduces downtime, improves robustness,
        and enables graceful degradation when primary plans fail.
    
        The pattern includes risk identification, scenario planning, plan B/C/D
        creation, trigger condition specification, and rapid plan switching. It
        balances the cost of planning for contingencies against the benefit of
        being prepared for failures.
    
    Key Components:
        1. Risk Identifier: Detects potential failure points
        2. Scenario Generator: Creates alternative scenarios
        3. Contingency Creator: Develops backup plans
        4. Trigger Monitor: Watches for contingency conditions
        5. Plan Switcher: Activates alternative plans
        6. Impact Assessor: Evaluates risk severity
        7. Recovery Planner: Plans return to primary plan
    
    Contingency Types:
        1. Alternative Action: Different way to achieve same goal
        2. Fallback Goal: Reduced/modified objective
        3. Resource Substitution: Use different resources
        4. Path Alternative: Different sequence of steps
        5. Graceful Degradation: Accept reduced functionality
        6. Abort and Recover: Abandon task, clean up
    
    Risk Assessment:
        1. Likelihood: Probability of occurrence (low/medium/high)
        2. Impact: Severity of consequences (minor/moderate/critical)
        3. Priority: Likelihood × Impact
        4. Mitigation Cost: Resources to prepare contingency
    
    Planning Phases:
        1. Risk Identification: What could go wrong?
        2. Impact Analysis: How bad would it be?
        3. Contingency Development: What's plan B?
        4. Trigger Specification: When to switch plans?
        5. Monitoring: Watch for triggers
        6. Activation: Switch to contingency
        7. Recovery: Return to normal operations
    
    Use Cases:
        - Mission-critical operations
        - Uncertain environments
        - Resource-constrained systems
        - High-stakes decision making
        - Autonomous systems
        - Real-time operations
        - Safety-critical applications
    
    Advantages:
        - Reduces response time to failures
        - Improves system reliability
        - Enables graceful degradation
        - Proactive vs reactive approach
        - Builds resilience
        - Reduces costly downtime
        - Increases confidence
    
    Challenges:
        - Computational cost of planning
        - Predicting all possible failures
        - Balancing thoroughness vs. efficiency
        - Maintaining multiple plans
        - Detecting when to switch
        - Avoiding over-planning
        - Resource allocation for contingencies
    
    LangChain Implementation:
        This implementation uses LangChain for:
        - LLM-based risk identification
        - Alternative plan generation
        - Scenario analysis
        - Natural language contingency description
    
    Production Considerations:
        - Focus on high-impact, high-probability risks
        - Use tiered contingency levels (B, C, D plans)
        - Implement fast plan switching
        - Monitor trigger conditions efficiently
        - Test contingency plans regularly
        - Document all contingencies
        - Balance planning cost vs. benefit
        - Enable human override when needed
    """
    

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


    Import Statements

    Standard Library Imports

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

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

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

    Lines 112-116

    Inherits from: Enum

    Purpose:

    Risk likelihood levels.
    

    Class: ImpactLevel

    Lines 119-123

    Inherits from: Enum

    Purpose:

    Impact severity levels.
    

    Class: PlanStatus

    Lines 126-132

    Inherits from: Enum

    Purpose:

    Status of a plan.
    

    Class: Risk

    Lines 136-160

    Purpose:

    Represents a potential risk or failure point.
    
    Attributes:
        risk_id: Unique identifier
        description: Risk description
        likelihood: Probability of occurrence
        impact: Severity of consequences
        trigger_condition: Condition that indicates risk occurred
        affected_steps: Plan steps affected by this risk
    

    Methods:

    priority(self)

    Lines 156-160

    Decorators: property

    Purpose:

    Calculate risk priority.
    

    Parameters:

    Returns: int


    Class: Plan

    Lines 164-187

    Purpose:

    Represents a plan (primary or contingency).
    
    Attributes:
        plan_id: Unique identifier
        description: Plan description
        steps: List of action steps
        status: Plan status
        parent_plan_id: ID of plan this is contingency for
        triggers: Conditions that activate this plan
        cost: Estimated cost/resources
        success_criteria: How to know if plan succeeded
    

    Class: ContingencyPlanner

    Lines 190-559

    Purpose:

    Contingency planning system that creates and manages backup plans.
    
    This planner identifies risks, creates alternative plans, monitors
    for trigger conditions, and switches to contingencies when needed.
    

    Methods:

    __init__(self, temperature)

    Lines 198-211

    Purpose:

    Initialize contingency planner.
    
    Args:
        temperature: LLM temperature
    

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

    create_primary_plan(self, description, steps)

    Lines 213-239

    Purpose:

    Create primary plan.
    
    Args:
        description: Plan description
        steps: Plan steps
    
    Returns:
        Created plan
    

    Parameters: – description: Description – steps: Steps

    Returns: Plan

    identify_risks(self, plan)

    Lines 241-314

    Purpose:

    Identify potential risks in a plan using LLM.
    
    Args:
        plan: Plan to analyze
    
    Returns:
        List of identified risks
    

    Parameters: – plan: Plan

    Returns: <ast.Subscript object at 0x000001A06D6AB310>

    create_contingency_plan(self, primary_plan, risk, level)

    Lines 316-382

    Purpose:

    Create contingency plan for a specific risk.
    
    Args:
        primary_plan: Primary plan
        risk: Risk to mitigate
        level: Contingency level (B, C, D, etc.)
    
    Returns:
        Contingency plan
    

    Parameters: – primary_plan: Primary plan – risk: Risk – level: Level

    Returns: Plan

    create_tiered_contingencies(self, primary_plan, num_tiers)

    Lines 384-417

    Purpose:

    Create multiple tiers of contingency plans.
    
    Args:
        primary_plan: Primary plan
        num_tiers: Number of contingency tiers
    
    Returns:
        List of contingency plans
    

    Parameters: – primary_plan: Primary plan – num_tiers: Num tiers

    Returns: <ast.Subscript object at 0x000001A06D608C50>

    check_trigger_conditions(self, observed_events)

    Lines 419-444

    Purpose:

    Check if any contingency should be activated.
    
    Args:
        observed_events: List of observed events/conditions
    
    Returns:
        Contingency plan to activate, or None
    

    Parameters: – observed_events: Observed events

    Returns: <ast.Subscript object at 0x000001A06D83C910>

    activate_contingency(self, contingency_plan, reason)

    Lines 446-473

    Purpose:

    Activate a contingency plan.
    
    Args:
        contingency_plan: Contingency to activate
        reason: Reason for activation
    

    Parameters: – contingency_plan: Contingency plan – reason: Reason

    simulate_execution(self, plan, simulate_failure)

    Lines 475-516

    Purpose:

    Simulate plan execution.
    
    Args:
        plan: Plan to execute
        simulate_failure: Simulate failure at this step
    
    Returns:
        Execution result
    

    Parameters: – plan: Plan – simulate_failure: Simulate failure

    Returns: <ast.Subscript object at 0x000001A06D156C50>

    get_plan_summary(self)

    Lines 518-543

    Purpose:

    Get summary of all plans.
    

    Parameters:

    Returns: str

    get_risk_summary(self)

    Lines 545-559

    Purpose:

    Get summary of identified risks.
    

    Parameters:

    Returns: str


    Functions

    demonstrate_contingency_planning()

    Lines 562-1006

    Purpose:

    Demonstrate contingency planning pattern.
    

    Key Concepts Summary

    Pattern Overview

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

    Core Components

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

    When to Use

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

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


    Best Practices Demonstrated

    1. Environment Configuration

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

    2. Error Handling

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

    3. Prompt Engineering

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

    4. Code Organization

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

    5. Performance Optimization

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

    LangChain Components Used

    • ChatOpenAI: OpenAI model interface
    • Chains: Sequential component composition

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


    Execution Flow

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

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


    Performance Considerations

    Computational Cost

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

    Optimization Strategies

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

    Cost Estimates (Approximate)

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

    Actual costs vary based on input/output token counts


    Common Pitfalls to Avoid

    1. Prompt Design Issues

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

    2. Error Handling

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

    3. Token Management

    ❌ Exceeding model token limits
    ✅ Truncate inputs, chunk processing

    4. Security

    ❌ Exposing API keys in code
    ✅ Use environment variables

    5. Testing

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

    6. Cost Management

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


    Further Improvements

    Production Enhancements

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

    Feature Extensions

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

    Code Quality

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

    Related Patterns

    • Pattern 105
    • Pattern 107

    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 Contingency Planning pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

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

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

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


    File: 106_contingency_planning.py
    Total Lines: 1011
    Pattern Category: AI Agent Design Pattern #106


    Generated from actual implementation code
    For educational and reference purposes