Tag: ai

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

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

When to Use

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

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

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

LangChain Components Used

ChatOpenAI: OpenAI model interface
Chains: Sequential component composition
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

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

Related Patterns

Pattern 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

November 2, 2025

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

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

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

When to Use

❌ 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

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

Related Patterns

Pattern 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

November 2, 2025

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

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

When to Use

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

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

LangChain Components Used

ChatOpenAI: OpenAI model interface
Chains: Sequential component composition

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

Execution Flow

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

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

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

Related Patterns

Pattern 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

November 2, 2025

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

LangChain Imports

from langchain.memory.ConversationBufferMemory
from langchain_openai.ChatOpenAI

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

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

When to Use

❌ 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

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

Related Patterns

Pattern 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

November 2, 2025

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

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

When to Use

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

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

LangChain Components Used

ChatOpenAI: OpenAI model interface
Chains: Sequential component composition

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

Execution Flow

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

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

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

Related Patterns

Pattern 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

November 2, 2025

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

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

When to Use

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

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

LangChain Components Used

ChatOpenAI: OpenAI model interface
Chains: Sequential component composition

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

Execution Flow

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

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

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

Related Patterns

Pattern 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

November 2, 2025

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

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

When to Use

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

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

LangChain Components Used

ChatOpenAI: OpenAI model interface
Chains: Sequential component composition

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

Execution Flow

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

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

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

Related Patterns

Pattern 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

November 2, 2025

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

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

When to Use

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

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

LangChain Components Used

ChatOpenAI: OpenAI model interface
Chains: Sequential component composition

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

Execution Flow

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

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

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

Related Patterns

Pattern 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

November 2, 2025

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

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

When to Use

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

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

LangChain Components Used

ChatOpenAI: OpenAI model interface
Chains: Sequential component composition

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

Execution Flow

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

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

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

Related Patterns

Pattern 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

November 2, 2025

Agentic Design Pattern: Hierarchical Planning


    """
Pattern 105: Hierarchical Planning

Description:
    The Hierarchical Planning pattern breaks down complex goals into a hierarchy
    of sub-goals and actions at multiple levels of abstraction. This pattern
    enables agents to handle long-horizon tasks by decomposing them into manageable
    pieces, planning at different levels of granularity, and progressively refining
    high-level strategies into concrete actions.
    
    Hierarchical planning is inspired by hierarchical task networks (HTN) and
    goal decomposition techniques. It allows agents to reason about problems at
    multiple scales: strategic (what to achieve), tactical (how to achieve it),
    and operational (specific actions). Each level focuses on appropriate
    abstractions, making planning more tractable and interpretable.
    
    This pattern is particularly effective for complex, multi-step tasks where
    flat planning would result in intractably large search spaces. By organizing
    goals hierarchically, the agent can plan efficiently, adapt at different
    levels, and explain its reasoning at appropriate levels of detail.

Key Components:
    1. Goal Hierarchy: Tree structure of goals and sub-goals
    2. Abstraction Levels: Strategic, tactical, operational layers
    3. Decomposition Rules: How to break goals into sub-goals
    4. Level Planner: Plans at each abstraction level
    5. Refinement Engine: Converts abstract plans to concrete actions
    6. Execution Monitor: Tracks progress at all levels
    7. Re-planning Trigger: Identifies when to replan at each level

Abstraction Levels:
    1. Strategic Level: High-level goals and objectives
       - What needs to be achieved
       - Overall strategy and approach
       - Long-term planning horizon
    
    2. Tactical Level: Sub-goals and procedures
       - How to achieve strategic goals
       - Intermediate milestones
       - Medium-term planning
    
    3. Operational Level: Concrete actions
       - Specific actions to execute
       - Immediate next steps
       - Short-term execution

Planning Phases:
    1. Goal Decomposition: Break high-level goal into sub-goals
    2. Level Planning: Plan at each abstraction level
    3. Plan Refinement: Refine abstract plans into detailed plans
    4. Execution: Execute lowest-level actions
    5. Monitoring: Track progress at all levels
    6. Re-planning: Adapt when needed at appropriate level

Use Cases:
    - Complex project planning
    - Multi-step task execution
    - Robotic mission planning
    - Software development workflows
    - Travel itinerary planning
    - Research project organization
    - Business process automation

Advantages:
    - Manages complexity through abstraction
    - More efficient than flat planning
    - Allows reasoning at appropriate granularity
    - Easier to explain and understand
    - Facilitates partial replanning
    - Modular and reusable sub-plans
    - Scales to long-horizon tasks

Challenges:
    - Defining appropriate abstraction levels
    - Goal decomposition complexity
    - Ensuring consistency across levels
    - Handling dependencies between sub-goals
    - Determining when to replan at each level
    - Balancing planning depth vs. efficiency

LangChain Implementation:
    This implementation uses LangChain for:
    - LLM-based goal decomposition
    - Multi-level plan generation
    - Plan refinement and detailing
    - Natural language planning
    
Production Considerations:
    - Define clear abstraction levels for domain
    - Implement efficient goal decomposition
    - Track dependencies between sub-goals
    - Monitor execution at all levels
    - Enable replanning at appropriate level
    - Cache and reuse common sub-plans
    - Provide visualization of plan hierarchy
    - Handle failures at correct abstraction level
"""

import os
from typing import List, Dict, Any, Optional, Tuple
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 PlanLevel(Enum):
    """Planning abstraction levels."""
    STRATEGIC = "strategic"  # High-level goals
    TACTICAL = "tactical"    # Sub-goals and procedures
    OPERATIONAL = "operational"  # Concrete actions


class GoalStatus(Enum):
    """Status of a goal."""
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"
    BLOCKED = "blocked"


@dataclass
class Goal:
    """
    Represents a goal at any level of the hierarchy.
    
    Attributes:
        goal_id: Unique identifier
        description: Goal description
        level: Abstraction level
        parent_id: Parent goal ID (None for root)
        children_ids: List of child goal IDs
        status: Current status
        priority: Priority (1-10)
        dependencies: IDs of goals that must complete first
        actions: Concrete actions (for operational level)
        metadata: Additional information
    """
    goal_id: str
    description: str
    level: PlanLevel
    parent_id: Optional[str] = None
    children_ids: List[str] = field(default_factory=list)
    status: GoalStatus = GoalStatus.PENDING
    priority: int = 5
    dependencies: List[str] = field(default_factory=list)
    actions: List[str] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    created_at: datetime = field(default_factory=datetime.now)
    completed_at: Optional[datetime] = None


class HierarchicalPlanner:
    """
    Hierarchical planning system that plans at multiple abstraction levels.
    
    This planner decomposes high-level goals into hierarchies of sub-goals,
    plans at each level, and progressively refines abstract plans into
    concrete actions.
    """
    
    def __init__(self, temperature: float = 0.3):
        """
        Initialize hierarchical planner.
        
        Args:
            temperature: LLM temperature for planning
        """
        self.llm = ChatOpenAI(temperature=temperature, model="gpt-3.5-turbo")
        self.goals: Dict[str, Goal] = {}
        self.goal_counter = 0
        self.execution_log: List[Dict[str, Any]] = []
    
    def create_goal(
        self,
        description: str,
        level: PlanLevel,
        parent_id: Optional[str] = None,
        priority: int = 5,
        dependencies: Optional[List[str]] = None
    ) -> Goal:
        """
        Create a new goal.
        
        Args:
            description: Goal description
            level: Abstraction level
            parent_id: Parent goal ID
            priority: Priority (1-10)
            dependencies: Goal dependencies
            
        Returns:
            Created goal
        """
        self.goal_counter += 1
        goal = Goal(
            goal_id=f"goal_{self.goal_counter}",
            description=description,
            level=level,
            parent_id=parent_id,
            priority=priority,
            dependencies=dependencies or []
        )
        
        self.goals[goal.goal_id] = goal
        
        # Update parent's children
        if parent_id and parent_id in self.goals:
            self.goals[parent_id].children_ids.append(goal.goal_id)
        
        return goal
    
    def decompose_goal(
        self,
        goal: Goal,
        target_level: PlanLevel
    ) -> List[Goal]:
        """
        Decompose a goal into sub-goals at lower abstraction level.
        
        Args:
            goal: Goal to decompose
            target_level: Target abstraction level for sub-goals
            
        Returns:
            List of sub-goals
        """
        # Use LLM to decompose goal
        prompt = ChatPromptTemplate.from_template(
            "You are a hierarchical planning assistant.\n\n"
            "Goal: {goal_description}\n"
            "Current Level: {current_level}\n"
            "Target Level: {target_level}\n\n"
            "Decompose this goal into 3-5 sub-goals at the {target_level} level.\n"
            "List each sub-goal on a new line, numbered.\n"
            "Sub-goals should be concrete, achievable, and collectively accomplish the parent goal.\n\n"
            "Sub-goals:"
        )
        
        chain = prompt | self.llm | StrOutputParser()
        result = chain.invoke({
            "goal_description": goal.description,
            "current_level": goal.level.value,
            "target_level": target_level.value
        })
        
        # Parse sub-goals from LLM output
        lines = [line.strip() for line in result.strip().split('\n') if line.strip()]
        sub_goals = []
        
        for line in lines:
            # Remove numbering (e.g., "1. ", "1) ", etc.)
            description = line
            for prefix in ["1.", "2.", "3.", "4.", "5.", "1)", "2)", "3)", "4)", "5)", "-", "*"]:
                if description.startswith(prefix):
                    description = description[len(prefix):].strip()
                    break
            
            if description:
                sub_goal = self.create_goal(
                    description=description,
                    level=target_level,
                    parent_id=goal.goal_id,
                    priority=goal.priority
                )
                sub_goals.append(sub_goal)
        
        return sub_goals
    
    def plan_strategic(self, objective: str) -> Goal:
        """
        Create strategic-level plan.
        
        Args:
            objective: High-level objective
            
        Returns:
            Strategic goal
        """
        # Create strategic goal
        strategic_goal = self.create_goal(
            description=objective,
            level=PlanLevel.STRATEGIC,
            priority=10
        )
        
        return strategic_goal
    
    def plan_tactical(self, strategic_goal: Goal) -> List[Goal]:
        """
        Create tactical-level plans from strategic goal.
        
        Args:
            strategic_goal: Strategic goal to decompose
            
        Returns:
            List of tactical goals
        """
        tactical_goals = self.decompose_goal(strategic_goal, PlanLevel.TACTICAL)
        return tactical_goals
    
    def plan_operational(self, tactical_goal: Goal) -> List[Goal]:
        """
        Create operational-level plans from tactical goal.
        
        Args:
            tactical_goal: Tactical goal to decompose
            
        Returns:
            List of operational goals
        """
        operational_goals = self.decompose_goal(tactical_goal, PlanLevel.OPERATIONAL)
        
        # Generate concrete actions for operational goals
        for op_goal in operational_goals:
            op_goal.actions = self.generate_actions(op_goal)
        
        return operational_goals
    
    def generate_actions(self, operational_goal: Goal) -> List[str]:
        """
        Generate concrete actions for an operational goal.
        
        Args:
            operational_goal: Operational goal
            
        Returns:
            List of concrete actions
        """
        prompt = ChatPromptTemplate.from_template(
            "Generate 2-4 specific, concrete actions to accomplish this goal:\n\n"
            "Goal: {goal}\n\n"
            "List each action on a new line, numbered.\n"
            "Actions should be specific and executable.\n\n"
            "Actions:"
        )
        
        chain = prompt | self.llm | StrOutputParser()
        result = chain.invoke({"goal": operational_goal.description})
        
        # Parse actions
        lines = [line.strip() for line in result.strip().split('\n') if line.strip()]
        actions = []
        
        for line in lines:
            # Remove numbering
            action = line
            for prefix in ["1.", "2.", "3.", "4.", "1)", "2)", "3)", "4)", "-", "*"]:
                if action.startswith(prefix):
                    action = action[len(prefix):].strip()
                    break
            if action:
                actions.append(action)
        
        return actions
    
    def create_full_hierarchy(self, objective: str) -> Goal:
        """
        Create complete hierarchical plan from objective.
        
        Args:
            objective: High-level objective
            
        Returns:
            Root strategic goal with full hierarchy
        """
        # Strategic level
        strategic_goal = self.plan_strategic(objective)
        
        # Tactical level
        tactical_goals = self.plan_tactical(strategic_goal)
        
        # Operational level
        for tactical_goal in tactical_goals:
            self.plan_operational(tactical_goal)
        
        return strategic_goal
    
    def get_next_action(self) -> Optional[Tuple[Goal, str]]:
        """
        Get next executable action from the plan.
        
        Returns:
            Tuple of (goal, action) or None if no actions available
        """
        # Find operational goals that are ready to execute
        operational_goals = [
            g for g in self.goals.values()
            if g.level == PlanLevel.OPERATIONAL and g.status == GoalStatus.PENDING
        ]
        
        # Sort by priority
        operational_goals.sort(key=lambda g: g.priority, reverse=True)
        
        for goal in operational_goals:
            # Check dependencies
            deps_satisfied = all(
                self.goals[dep_id].status == GoalStatus.COMPLETED
                for dep_id in goal.dependencies
                if dep_id in self.goals
            )
            
            if deps_satisfied and goal.actions:
                return (goal, goal.actions[0])
        
        return None
    
    def execute_action(self, goal: Goal, action: str) -> bool:
        """
        Execute an action (simulated).
        
        Args:
            goal: Goal being executed
            action: Action to execute
            
        Returns:
            Success status
        """
        # Mark goal as in progress
        if goal.status == GoalStatus.PENDING:
            goal.status = GoalStatus.IN_PROGRESS
        
        # Log execution
        self.execution_log.append({
            "timestamp": datetime.now(),
            "goal_id": goal.goal_id,
            "action": action,
            "level": goal.level.value
        })
        
        # Remove completed action
        if action in goal.actions:
            goal.actions.remove(action)
        
        # Mark goal complete if no more actions
        if not goal.actions:
            goal.status = GoalStatus.COMPLETED
            goal.completed_at = datetime.now()
            self.propagate_completion(goal)
        
        return True
    
    def propagate_completion(self, goal: Goal):
        """
        Propagate goal completion up the hierarchy.
        
        Args:
            goal: Completed goal
        """
        if not goal.parent_id or goal.parent_id not in self.goals:
            return
        
        parent = self.goals[goal.parent_id]
        
        # Check if all children are complete
        all_children_complete = all(
            self.goals[child_id].status == GoalStatus.COMPLETED
            for child_id in parent.children_ids
            if child_id in self.goals
        )
        
        if all_children_complete:
            parent.status = GoalStatus.COMPLETED
            parent.completed_at = datetime.now()
            self.propagate_completion(parent)
    
    def get_hierarchy_summary(self, goal_id: Optional[str] = None) -> str:
        """
        Get textual summary of goal hierarchy.
        
        Args:
            goal_id: Root goal ID (None for all roots)
            
        Returns:
            Formatted hierarchy string
        """
        def format_goal(goal: Goal, indent: int = 0) -> str:
            status_symbol = {
                GoalStatus.PENDING: "⏳",
                GoalStatus.IN_PROGRESS: "🔄",
                GoalStatus.COMPLETED: "✅",
                GoalStatus.FAILED: "❌",
                GoalStatus.BLOCKED: "🚫"
            }
            
            symbol = status_symbol.get(goal.status, "❓")
            prefix = "  " * indent
            
            result = f"{prefix}{symbol} [{goal.level.value}] {goal.description}\n"
            
            # Add actions for operational level
            if goal.level == PlanLevel.OPERATIONAL and goal.actions:
                for action in goal.actions:
                    result += f"{prefix}  → {action}\n"
            
            # Recursively add children
            for child_id in goal.children_ids:
                if child_id in self.goals:
                    result += format_goal(self.goals[child_id], indent + 1)
            
            return result
        
        if goal_id:
            if goal_id in self.goals:
                return format_goal(self.goals[goal_id])
            return "Goal not found"
        
        # Format all root goals
        root_goals = [g for g in self.goals.values() if g.parent_id is None]
        return "\n".join(format_goal(g) for g in root_goals)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get planning statistics."""
        goals_by_level = {level: 0 for level in PlanLevel}
        goals_by_status = {status: 0 for status in GoalStatus}
        
        for goal in self.goals.values():
            goals_by_level[goal.level] += 1
            goals_by_status[goal.status] += 1
        
        return {
            "total_goals": len(self.goals),
            "by_level": {level.value: count for level, count in goals_by_level.items()},
            "by_status": {status.value: count for status, count in goals_by_status.items()},
            "total_actions_executed": len(self.execution_log)
        }


def demonstrate_hierarchical_planning():
    """Demonstrate hierarchical planning pattern."""
    
    print("=" * 80)
    print("HIERARCHICAL PLANNING PATTERN DEMONSTRATION")
    print("=" * 80)
    
    # Example 1: Basic three-level hierarchy
    print("\n" + "=" * 80)
    print("Example 1: Three-Level Planning Hierarchy")
    print("=" * 80)
    
    planner = HierarchicalPlanner()
    
    print("\nCreating hierarchical plan for: 'Build a web application'")
    
    # Strategic level
    strategic = planner.plan_strategic("Build a web application")
    print(f"\n[STRATEGIC] {strategic.description}")
    
    # Tactical level
    tactical = planner.plan_tactical(strategic)
    print(f"\n[TACTICAL] Decomposed into {len(tactical)} sub-goals:")
    for i, tac_goal in enumerate(tactical, 1):
        print(f"  {i}. {tac_goal.description}")
    
    # Operational level (for first tactical goal)
    if tactical:
        operational = planner.plan_operational(tactical[0])
        print(f"\n[OPERATIONAL] '{tactical[0].description}' decomposed into {len(operational)} goals:")
        for i, op_goal in enumerate(operational, 1):
            print(f"  {i}. {op_goal.description}")
            if op_goal.actions:
                for action in op_goal.actions:
                    print(f"     → {action}")
    
    # Example 2: Complete hierarchy creation
    print("\n" + "=" * 80)
    print("Example 2: Complete Hierarchical Plan")
    print("=" * 80)
    
    planner2 = HierarchicalPlanner()
    
    objective = "Organize a team offsite event"
    print(f"\nObjective: {objective}\n")
    
    root = planner2.create_full_hierarchy(objective)
    
    print("Complete Hierarchy:")
    print(planner2.get_hierarchy_summary())
    
    # Example 3: Plan execution
    print("\n" + "=" * 80)
    print("Example 3: Executing Hierarchical Plan")
    print("=" * 80)
    
    planner3 = HierarchicalPlanner()
    
    # Create simpler plan for execution demo
    strategic = planner3.create_goal(
        "Complete project report",
        PlanLevel.STRATEGIC
    )
    
    tactical1 = planner3.create_goal(
        "Gather data and research",
        PlanLevel.TACTICAL,
        parent_id=strategic.goal_id
    )
    
    tactical2 = planner3.create_goal(
        "Write and format report",
        PlanLevel.TACTICAL,
        parent_id=strategic.goal_id,
        dependencies=[tactical1.goal_id]
    )
    
    # Operational goals for tactical1
    op1 = planner3.create_goal(
        "Collect data from databases",
        PlanLevel.OPERATIONAL,
        parent_id=tactical1.goal_id
    )
    op1.actions = ["Connect to database", "Run queries", "Export results"]
    
    op2 = planner3.create_goal(
        "Review literature",
        PlanLevel.OPERATIONAL,
        parent_id=tactical1.goal_id
    )
    op2.actions = ["Search papers", "Read abstracts"]
    
    print("Initial Plan:")
    print(planner3.get_hierarchy_summary())
    
    print("\nExecuting actions...")
    
    # Execute actions
    for i in range(5):
        next_item = planner3.get_next_action()
        if next_item:
            goal, action = next_item
            print(f"  {i+1}. Executing: {action} (for '{goal.description}')")
            planner3.execute_action(goal, action)
        else:
            break
    
    print("\nUpdated Plan:")
    print(planner3.get_hierarchy_summary())
    
    # Example 4: Priority-based planning
    print("\n" + "=" * 80)
    print("Example 4: Priority-Based Goal Selection")
    print("=" * 80)
    
    planner4 = HierarchicalPlanner()
    
    # Create goals with different priorities
    strategic = planner4.create_goal(
        "Launch product",
        PlanLevel.STRATEGIC,
        priority=10
    )
    
    high_priority = planner4.create_goal(
        "Fix critical bug",
        PlanLevel.OPERATIONAL,
        parent_id=strategic.goal_id,
        priority=9
    )
    high_priority.actions = ["Identify bug", "Write fix", "Test fix"]
    
    low_priority = planner4.create_goal(
        "Update documentation",
        PlanLevel.OPERATIONAL,
        parent_id=strategic.goal_id,
        priority=3
    )
    low_priority.actions = ["Review docs", "Make updates"]
    
    print("Goals by priority:")
    for goal in sorted(planner4.goals.values(), key=lambda g: g.priority, reverse=True):
        if goal.level == PlanLevel.OPERATIONAL:
            print(f"  Priority {goal.priority}: {goal.description}")
    
    print("\nExecution order (priority-based):")
    for i in range(5):
        next_item = planner4.get_next_action()
        if next_item:
            goal, action = next_item
            print(f"  {i+1}. [P{goal.priority}] {action}")
            planner4.execute_action(goal, action)
        else:
            break
    
    # Example 5: Dependency handling
    print("\n" + "=" * 80)
    print("Example 5: Goal Dependencies")
    print("=" * 80)
    
    planner5 = HierarchicalPlanner()
    
    strategic = planner5.create_goal("Deploy application", PlanLevel.STRATEGIC)
    
    goal_a = planner5.create_goal(
        "Run tests",
        PlanLevel.OPERATIONAL,
        parent_id=strategic.goal_id
    )
    goal_a.actions = ["Execute test suite"]
    
    goal_b = planner5.create_goal(
        "Build for production",
        PlanLevel.OPERATIONAL,
        parent_id=strategic.goal_id,
        dependencies=[goal_a.goal_id]
    )
    goal_b.actions = ["Compile code"]
    
    goal_c = planner5.create_goal(
        "Deploy to server",
        PlanLevel.OPERATIONAL,
        parent_id=strategic.goal_id,
        dependencies=[goal_b.goal_id]
    )
    goal_c.actions = ["Upload to server"]
    
    print("Dependency chain: Tests → Build → Deploy\n")
    print("Initial state:")
    for goal in [goal_a, goal_b, goal_c]:
        deps = ", ".join(goal.dependencies) if goal.dependencies else "none"
        print(f"  {goal.description}: dependencies=[{deps}], status={goal.status.value}")
    
    print("\nExecution respects dependencies:")
    for i in range(3):
        next_item = planner5.get_next_action()
        if next_item:
            goal, action = next_item
            print(f"  {i+1}. {action} (from '{goal.description}')")
            planner5.execute_action(goal, action)
    
    # Example 6: Progress monitoring
    print("\n" + "=" * 80)
    print("Example 6: Multi-Level Progress Monitoring")
    print("=" * 80)
    
    planner6 = HierarchicalPlanner()
    
    root = planner6.create_full_hierarchy("Write research paper")
    
    print("Initial Statistics:")
    stats = planner6.get_statistics()
    print(f"  Total Goals: {stats['total_goals']}")
    print(f"  By Level: {stats['by_level']}")
    print(f"  By Status: {stats['by_status']}")
    
    # Execute some actions
    print("\nExecuting 10 actions...")
    for _ in range(10):
        next_item = planner6.get_next_action()
        if next_item:
            goal, action = next_item
            planner6.execute_action(goal, action)
    
    print("\nUpdated Statistics:")
    stats = planner6.get_statistics()
    print(f"  Total Goals: {stats['total_goals']}")
    print(f"  By Status: {stats['by_status']}")
    print(f"  Actions Executed: {stats['total_actions_executed']}")
    
    # Example 7: Plan visualization
    print("\n" + "=" * 80)
    print("Example 7: Hierarchical Plan Visualization")
    print("=" * 80)
    
    planner7 = HierarchicalPlanner()
    
    # Create multi-level plan
    strategic = planner7.create_goal("Learn Python", PlanLevel.STRATEGIC)
    
    tac1 = planner7.create_goal("Master basics", PlanLevel.TACTICAL, parent_id=strategic.goal_id)
    tac2 = planner7.create_goal("Build projects", PlanLevel.TACTICAL, parent_id=strategic.goal_id)
    
    op1 = planner7.create_goal("Learn syntax", PlanLevel.OPERATIONAL, parent_id=tac1.goal_id)
    op1.actions = ["Study variables", "Practice loops"]
    op1.status = GoalStatus.COMPLETED  # Mark as done
    
    op2 = planner7.create_goal("Learn data structures", PlanLevel.OPERATIONAL, parent_id=tac1.goal_id)
    op2.actions = ["Study lists", "Study dictionaries"]
    op2.status = GoalStatus.IN_PROGRESS
    
    op3 = planner7.create_goal("Build web app", PlanLevel.OPERATIONAL, parent_id=tac2.goal_id)
    op3.actions = ["Choose framework", "Create app"]
    
    print("Hierarchical Plan with Status:")
    print(planner7.get_hierarchy_summary())
    
    # Example 8: Replanning scenario
    print("\n" + "=" * 80)
    print("Example 8: Adaptive Replanning")
    print("=" * 80)
    
    planner8 = HierarchicalPlanner()
    
    print("Original Plan:")
    strategic = planner8.create_goal("Travel to conference", PlanLevel.STRATEGIC)
    
    tac1 = planner8.create_goal("Book transportation", PlanLevel.TACTICAL, parent_id=strategic.goal_id)
    op1 = planner8.create_goal("Book flight", PlanLevel.OPERATIONAL, parent_id=tac1.goal_id)
    op1.actions = ["Search flights", "Purchase ticket"]
    
    print(planner8.get_hierarchy_summary())
    
    print("\nSimulating failure (flight cancelled)...")
    op1.status = GoalStatus.FAILED
    
    # Create alternative plan
    print("\nReplanning at tactical level:")
    op2 = planner8.create_goal("Book train", PlanLevel.OPERATIONAL, parent_id=tac1.goal_id)
    op2.actions = ["Search trains", "Purchase ticket"]
    
    print(planner8.get_hierarchy_summary())
    
    # Summary
    print("\n" + "=" * 80)
    print("SUMMARY: Hierarchical Planning Pattern")
    print("=" * 80)
    
    summary = """
    The Hierarchical Planning pattern demonstrated:
    
    1. THREE-LEVEL HIERARCHY (Example 1):
       - Strategic: High-level objectives
       - Tactical: Sub-goals and procedures
       - Operational: Concrete actions
       - Progressive decomposition
       - Abstraction management
    
    2. COMPLETE HIERARCHY (Example 2):
       - Full plan generation
       - Multi-level decomposition
       - Parent-child relationships
       - Structured goal tree
       - Hierarchical visualization
    
    3. PLAN EXECUTION (Example 3):
       - Sequential action execution
       - Status tracking at all levels
       - Progress propagation
       - Completion detection
       - Hierarchical updates
    
    4. PRIORITY HANDLING (Example 4):
       - Priority-based selection
       - Urgent task handling
       - Importance ordering
       - Resource allocation
       - Efficient scheduling
    
    5. DEPENDENCY MANAGEMENT (Example 5):
       - Goal dependencies
       - Execution ordering
       - Prerequisite checking
       - Constraint satisfaction
       - Sequential execution
    
    6. PROGRESS MONITORING (Example 6):
       - Multi-level tracking
       - Status aggregation
       - Statistics collection
       - Performance metrics
       - Completion monitoring
    
    7. PLAN VISUALIZATION (Example 7):
       - Hierarchical display
       - Status indicators
       - Tree structure
       - Action details
       - Clear representation
    
    8. ADAPTIVE REPLANNING (Example 8):
       - Failure handling
       - Alternative generation
       - Level-appropriate replanning
       - Plan modification
       - Resilience
    
    KEY BENEFITS:
    ✓ Manages complexity through abstraction
    ✓ Efficient planning for long horizons
    ✓ Reasoning at appropriate granularity
    ✓ Easy to understand and explain
    ✓ Facilitates partial replanning
    ✓ Modular and reusable
    ✓ Scales to complex tasks
    ✓ Clear progress tracking
    
    USE CASES:
    • Complex project planning
    • Multi-step task execution
    • Robotic mission planning
    • Software development workflows
    • Travel itinerary planning
    • Research project organization
    • Business process automation
    • Strategic decision-making
    
    ABSTRACTION LEVELS:
    → Strategic: What to achieve (objectives)
    → Tactical: How to achieve (procedures)
    → Operational: Specific actions (execution)
    
    BEST PRACTICES:
    1. Define clear abstraction levels for domain
    2. Decompose goals systematically
    3. Track dependencies between goals
    4. Monitor execution at all levels
    5. Enable replanning at appropriate level
    6. Cache and reuse common sub-plans
    7. Visualize plan hierarchy
    8. Handle failures at correct level
    
    TRADE-OFFS:
    • Planning overhead vs. execution efficiency
    • Abstraction depth vs. simplicity
    • Flexibility vs. structure
    • Decomposition detail vs. overview
    
    PRODUCTION CONSIDERATIONS:
    → Implement domain-specific decomposition rules
    → Use caching for common sub-plans
    → Monitor performance at each level
    → Enable interactive plan adjustment
    → Provide clear visualization
    → Track dependencies explicitly
    → Implement efficient replanning
    → Handle partial plan execution
    → Support concurrent goal execution
    → Log decisions at all levels
    
    This pattern enables agents to tackle complex, long-horizon tasks
    by organizing planning at multiple levels of abstraction, from
    high-level strategies to concrete actions.
    """
    
    print(summary)


if __name__ == "__main__":
    demonstrate_hierarchical_planning()

Agentic Design Pattern: Hierarchical Planning

Pattern 105: Hierarchical Planning – Line-by-Line Explanation

Overview

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

Pattern Description

Pattern 105: Hierarchical Planning

Description: The Hierarchical Planning pattern breaks down complex goals into a hierarchy of sub-goals and actions at multiple levels of abstraction. This pattern enables agents to handle long-horizon tasks by decomposing them into manageable pieces, planning at different levels of granularity, and progressively refining high-level strategies into concrete actions.

Hierarchical planning is inspired by hierarchical task networks (HTN) and
goal decomposition techniques. It allows agents to reason about problems at
multiple scales: strategic (what to achieve), tactical (how to achieve it),
and operational (specific actions). Each level focuses on appropriate
abstractions, making planning more tractable and interpretable.

This pattern is particularly effective for complex, multi-step tasks where
flat planning would result in intractably large search spaces. By organizing
goals hierarchically, the agent can plan efficiently, adapt at different
levels, and explain its reasoning at appropriate levels of detail.

Key Components: 1. Goal Hierarchy: Tree structure of goals and sub-goals 2. Abstraction Levels: Strategic, tactical, operational layers 3. Decomposition Rules: How to break goals into sub-goals 4. Level Planner: Plans at each abstraction level 5. Refinement Engine: Converts abstract plans to concrete actions 6. Execution Monitor: Tracks progress at all levels 7. Re-planning Trigger: Identifies when to replan at each level

Abstraction Levels: 1. Strategic Level: High-level goals and objectives – What needs to be achieved – Overall strategy and approach – Long-term planning horizon

2. Tactical Level: Sub-goals and procedures
   - How to achieve strategic goals
   - Intermediate milestones
   - Medium-term planning

3. Operational Level: Concrete actions
   - Specific actions to execute
   - Immediate next steps
   - Short-term execution

Planning Phases: 1. Goal Decomposition: Break high-level goal into sub-goals 2. Level Planning: Plan at each abstraction level 3. Plan Refinement: Refine abstract plans into detailed plans 4. Execution: Execute lowest-level actions 5. Monitoring: Track progress at all levels 6. Re-planning: Adapt when needed at appropriate level

Use Cases: – Complex project planning – Multi-step task execution – Robotic mission planning – Software development workflows – Travel itinerary planning – Research project organization – Business process automation

Advantages: – Manages complexity through abstraction – More efficient than flat planning – Allows reasoning at appropriate granularity – Easier to explain and understand – Facilitates partial replanning – Modular and reusable sub-plans – Scales to long-horizon tasks

Challenges: – Defining appropriate abstraction levels – Goal decomposition complexity – Ensuring consistency across levels – Handling dependencies between sub-goals – Determining when to replan at each level – Balancing planning depth vs. efficiency

LangChain Implementation: This implementation uses LangChain for: – LLM-based goal decomposition – Multi-level plan generation – Plan refinement and detailing – Natural language planning

Production Considerations: – Define clear abstraction levels for domain – Implement efficient goal decomposition – Track dependencies between sub-goals – Monitor execution at all levels – Enable replanning at appropriate level – Cache and reuse common sub-plans – Provide visualization of plan hierarchy – Handle failures at correct abstraction level

File Statistics

Total Lines: 946
Classes: 4
Functions: 1
Constants: 0

Module Documentation

"""
Pattern 105: Hierarchical Planning

Description:
    The Hierarchical Planning pattern breaks down complex goals into a hierarchy
    of sub-goals and actions at multiple levels of abstraction. This pattern
    enables agents to handle long-horizon tasks by decomposing them into manageable
    pieces, planning at different levels of granularity, and progressively refining
    high-level strategies into concrete actions.

    Hierarchical planning is inspired by hierarchical task networks (HTN) and
    goal decomposition techniques. It allows agents to reason about problems at
    multiple scales: strategic (what to achieve), tactical (how to achieve it),
    and operational (specific actions). Each level focuses on appropriate
    abstractions, making planning more tractable and interpretable.

    This pattern is particularly effective for complex, multi-step tasks where
    flat planning would result in intractably large search spaces. By organizing
    goals hierarchically, the agent can plan efficiently, adapt at different
    levels, and explain its reasoning at appropriate levels of detail.

Key Components:
    1. Goal Hierarchy: Tree structure of goals and sub-goals
    2. Abstraction Levels: Strategic, tactical, operational layers
    3. Decomposition Rules: How to break goals into sub-goals
    4. Level Planner: Plans at each abstraction level
    5. Refinement Engine: Converts abstract plans to concrete actions
    6. Execution Monitor: Tracks progress at all levels
    7. Re-planning Trigger: Identifies when to replan at each level

Abstraction Levels:
    1. Strategic Level: High-level goals and objectives
       - What needs to be achieved
       - Overall strategy and approach
       - Long-term planning horizon

    2. Tactical Level: Sub-goals and procedures
       - How to achieve strategic goals
       - Intermediate milestones
       - Medium-term planning

    3. Operational Level: Concrete actions
       - Specific actions to execute
       - Immediate next steps
       - Short-term execution

Planning Phases:
    1. Goal Decomposition: Break high-level goal into sub-goals
    2. Level Planning: Plan at each abstraction level
    3. Plan Refinement: Refine abstract plans into detailed plans
    4. Execution: Execute lowest-level actions
    5. Monitoring: Track progress at all levels
    6. Re-planning: Adapt when needed at appropriate level

Use Cases:
    - Complex project planning
    - Multi-step task execution
    - Robotic mission planning
    - Software development workflows
    - Travel itinerary planning
    - Research project organization
    - Business process automation

Advantages:
    - Manages complexity through abstraction
    - More efficient than flat planning
    - Allows reasoning at appropriate granularity
    - Easier to explain and understand
    - Facilitates partial replanning
    - Modular and reusable sub-plans
    - Scales to long-horizon tasks

Challenges:
    - Defining appropriate abstraction levels
    - Goal decomposition complexity
    - Ensuring consistency across levels
    - Handling dependencies between sub-goals
    - Determining when to replan at each level
    - Balancing planning depth vs. efficiency

LangChain Implementation:
    This implementation uses LangChain for:
    - LLM-based goal decomposition
    - Multi-level plan generation
    - Plan refinement and detailing
    - Natural language planning

Production Considerations:
    - Define clear abstraction levels for domain
    - Implement efficient goal decomposition
    - Track dependencies between sub-goals
    - Monitor execution at all levels
    - Enable replanning at appropriate level
    - Cache and reuse common sub-plans
    - Provide visualization of plan hierarchy
    - Handle failures at correct abstraction level
"""

Purpose: This module implements the Hierarchical 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.Tuple

Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – datetime.datetime: Date and time handling – enum.Enum: Enumeration support – 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: `PlanLevel`

Lines 113-117

Inherits from: Enum

Purpose:

Planning abstraction levels.

Class: `GoalStatus`

Lines 120-126

Inherits from: Enum

Purpose:

Status of a goal.

Class: `Goal`

Lines 130-157

Purpose:

Represents a goal at any level of the hierarchy.

Attributes:
    goal_id: Unique identifier
    description: Goal description
    level: Abstraction level
    parent_id: Parent goal ID (None for root)
    children_ids: List of child goal IDs
    status: Current status
    priority: Priority (1-10)
    dependencies: IDs of goals that must complete first
    actions: Concrete actions (for operational level)
    metadata: Additional information

Class: `HierarchicalPlanner`

Lines 160-533

Purpose:

Hierarchical planning system that plans at multiple abstraction levels.

This planner decomposes high-level goals into hierarchies of sub-goals,
plans at each level, and progressively refines abstract plans into
concrete actions.

Methods:

`init(self, temperature)`

Lines 169-179

Purpose:

Initialize hierarchical planner.

Args:
    temperature: LLM temperature for planning

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

`create_goal(self, description, level, parent_id, priority, dependencies)`

Lines 181-218

Purpose:

Create a new goal.

Args:
    description: Goal description
    level: Abstraction level
    parent_id: Parent goal ID
    priority: Priority (1-10)
    dependencies: Goal dependencies

Returns:
    Created goal

Parameters: – description: Description – level: Level – parent_id: Parent id – priority: Priority – dependencies: Dependencies

Returns: Goal

`decompose_goal(self, goal, target_level)`

Lines 220-275

Purpose:

Decompose a goal into sub-goals at lower abstraction level.

Args:
    goal: Goal to decompose
    target_level: Target abstraction level for sub-goals

Returns:
    List of sub-goals

Parameters: – goal: Goal – target_level: Target level

Returns: <ast.Subscript object at 0x000001A06D6F6A50>

`plan_strategic(self, objective)`

Lines 277-294

Purpose:

Create strategic-level plan.

Args:
    objective: High-level objective

Returns:
    Strategic goal

Parameters: – objective: Objective

Returns: Goal

`plan_tactical(self, strategic_goal)`

Lines 296-307

Purpose:

Create tactical-level plans from strategic goal.

Args:
    strategic_goal: Strategic goal to decompose

Returns:
    List of tactical goals

Parameters: – strategic_goal: Strategic goal

Returns: <ast.Subscript object at 0x000001A06D605650>

`plan_operational(self, tactical_goal)`

Lines 309-325

Purpose:

Create operational-level plans from tactical goal.

Args:
    tactical_goal: Tactical goal to decompose

Returns:
    List of operational goals

Parameters: – tactical_goal: Tactical goal

Returns: <ast.Subscript object at 0x000001A06D718510>

`generate_actions(self, operational_goal)`

Lines 327-362

Purpose:

Generate concrete actions for an operational goal.

Args:
    operational_goal: Operational goal

Returns:
    List of concrete actions

Parameters: – operational_goal: Operational goal

Returns: <ast.Subscript object at 0x000001A06D866310>

`create_full_hierarchy(self, objective)`

Lines 364-384

Purpose:

Create complete hierarchical plan from objective.

Args:
    objective: High-level objective

Returns:
    Root strategic goal with full hierarchy

Parameters: – objective: Objective

Returns: Goal

`get_next_action(self)`

Lines 386-413

Purpose:

Get next executable action from the plan.

Returns:
    Tuple of (goal, action) or None if no actions available

Parameters:

Returns: <ast.Subscript object at 0x000001A06D72C9D0>

`execute_action(self, goal, action)`

Lines 415-448

Purpose:

Execute an action (simulated).

Args:
    goal: Goal being executed
    action: Action to execute

Returns:
    Success status

Parameters: – goal: Goal – action: Action

Returns: bool

`propagate_completion(self, goal)`

Lines 450-472

Purpose:

Propagate goal completion up the hierarchy.

Args:
    goal: Completed goal

Parameters: – goal: Goal

`get_hierarchy_summary(self, goal_id)`

Lines 474-517

Purpose:

Get textual summary of goal hierarchy.

Args:
    goal_id: Root goal ID (None for all roots)

Returns:
    Formatted hierarchy string

Parameters: – goal_id: Goal id

Returns: str

`get_statistics(self)`

Lines 519-533

Purpose:

Get planning statistics.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D853BD0>

Functions

`demonstrate_hierarchical_planning()`

Lines 536-941

Purpose:

Demonstrate hierarchical planning pattern.

Key Concepts Summary

Pattern Overview

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

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

When to Use

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

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

LangChain Components Used

ChatOpenAI: OpenAI model interface
Chains: Sequential component composition

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

Execution Flow

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

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

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

Related Patterns

Pattern 104
Pattern 106

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 Hierarchical 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: 105_hierarchical_planning.py
Total Lines: 946
Pattern Category: AI Agent Design Pattern #105

Generated from actual implementation code
For educational and reference purposes

November 2, 2025

Tag: ai

Agentic Design Pattern: Online Learning

Agentic Design Pattern: Online Learning

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

Overview

Pattern Description

File Statistics

Module Documentation

Import Statements

Standard Library Imports

LangChain Imports

Other Imports

Classes

Class: ExampleBuffer

__init__(self, max_size)

add_example(self, input_text, output_text, feedback_score)

get_recent_examples(self, n)

get_best_examples(self, n)

detect_drift(self, window_size)

get_performance_trend(self)

Class: OnlineLearningAgent

__init__(self, model_name, domain)

create_dynamic_prompt(self, use_best)

respond(self, user_input)

learn_from_feedback(self, user_input, response, feedback_score)

interact_and_learn(self, user_input, feedback_score)

get_learning_stats(self)

Functions

demonstrate_online_learning()

Key Concepts Summary

Pattern Overview

Core Components

When to Use

Best Practices Demonstrated

1. Environment Configuration

2. Error Handling

3. Prompt Engineering

4. Code Organization

5. Performance Optimization

LangChain Components Used

Execution Flow

Performance Considerations

Computational Cost

Optimization Strategies

Cost Estimates (Approximate)

Common Pitfalls to Avoid

1. Prompt Design Issues

2. Error Handling

3. Token Management

4. Security

5. Testing

6. Cost Management

Further Improvements

Production Enhancements

Feature Extensions

Code Quality

Related Patterns

Summary

Agentic Design Pattern: Contextual Adaptation

Agentic Design Pattern: Contextual Adaptation

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

Overview

Pattern Description

File Statistics

Module Documentation

Import Statements

Standard Library Imports

LangChain Imports

Other Imports

Classes

Class: UserExperience

Class: Formality

Class: TaskUrgency

Class: Context

__init__(self)

to_dict(self)

summarize(self)

Class: ContextualAdaptationAgent

__init__(self, model_name)

detect_context(self, query)

Class: `ExampleBuffer`

`init(self, max_size)`

`add_example(self, input_text, output_text, feedback_score)`

`get_recent_examples(self, n)`

`get_best_examples(self, n)`

`detect_drift(self, window_size)`

`get_performance_trend(self)`

Class: `OnlineLearningAgent`

`init(self, model_name, domain)`

`create_dynamic_prompt(self, use_best)`

`respond(self, user_input)`

`learn_from_feedback(self, user_input, response, feedback_score)`

`interact_and_learn(self, user_input, feedback_score)`

`get_learning_stats(self)`

`demonstrate_online_learning()`

Class: `UserExperience`

Class: `Formality`

Class: `TaskUrgency`

Class: `Context`

`init(self)`

`to_dict(self)`

`summarize(self)`

Class: `ContextualAdaptationAgent`

`init(self, model_name)`

`detect_context(self, query)`

`update_context(self, updates)`

`respond(self, query, auto_detect)`

`demonstrate_adaptation(self, query, contexts)`

`demonstrate_contextual_adaptation()`

Class: `CommonSenseKnowledgeBase`

`init(self)`

`query(self, concept)`

`get_category(self, category)`

`get_relevant_facts(self, context)`

Class: `CommonSenseReasoner`

`init(self, model_name)`

`reason_with_common_sense(self, question)`

`check_plausibility(self, conclusion, context)`

`generate_assumptions(self, scenario)`

`explain_with_common_sense(self, observation)`

`demonstrate_common_sense_reasoning()`