"""
Pattern 118: Curiosity-Driven Exploration

Description:
    Agent explores environment based on intrinsic motivation driven by
    novelty, information gain, and prediction error.

Components:
    - Novelty detection
    - Information gain calculation
    - Exploration strategy
    - Intrinsic reward

Use Cases:
    - Open-ended learning
    - Exploration tasks
    - Discovery systems

LangChain Implementation:
    Uses LLM to generate curious questions and exploration strategies based on
    knowledge gaps and novelty.
"""

import os
from typing import List, Dict, Any, Set
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 CuriosityDrivenAgent:
    """Agent that explores based on intrinsic curiosity."""
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.8)
        self.knowledge_base = set()
        self.explored_topics = []
        self.curiosity_log = []
        
    def assess_novelty(self, topic: str) -> float:
        """Assess how novel a topic is based on existing knowledge."""
        novelty_prompt = ChatPromptTemplate.from_messages([
            ("system", "Rate the novelty of this topic given existing knowledge. Return a score from 0.0 (completely known) to 1.0 (completely novel)."),
            ("user", """Topic: {topic}

Known Knowledge:
{knowledge}

Novelty score (0.0-1.0):""")
        ])
        
        knowledge_str = "\n".join([f"- {k}" for k in list(self.knowledge_base)[:20]]) or "Empty"
        
        chain = novelty_prompt | self.llm | StrOutputParser()
        response = chain.invoke({"topic": topic, "knowledge": knowledge_str})
        
        try:
            score = float(response.strip())
            return max(0.0, min(1.0, score))
        except:
            return 0.5
    
    def generate_curious_questions(self, context: str) -> List[str]:
        """Generate questions driven by curiosity about knowledge gaps."""
        curiosity_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a curious agent exploring knowledge.
Generate 3-5 interesting questions that would expand understanding in unexplored directions.
Focus on:
- Novel connections
- Surprising implications
- Unexplored aspects
- Deep mechanisms"""),
            ("user", """Current Context: {context}

Known Knowledge:
{knowledge}

Generate curious questions:""")
        ])
        
        knowledge_str = "\n".join([f"- {k}" for k in list(self.knowledge_base)[-10:]]) or "Empty"
        
        chain = curiosity_prompt | self.llm | StrOutputParser()
        response = chain.invoke({"context": context, "knowledge": knowledge_str})
        
        # Parse questions
        questions = [line.strip('- ').strip() for line in response.split('\n') if line.strip()]
        return [q for q in questions if '?' in q][:5]
    
    def calculate_information_gain(self, potential_exploration: str) -> float:
        """Calculate expected information gain from exploring a topic."""
        gain_prompt = ChatPromptTemplate.from_messages([
            ("system", "Estimate information gain (0.0-1.0) from exploring this topic given current knowledge."),
            ("user", """Potential Exploration: {exploration}

Current Knowledge:
{knowledge}

Information gain score (0.0-1.0):""")
        ])
        
        knowledge_str = "\n".join([f"- {k}" for k in list(self.knowledge_base)[-15:]]) or "Empty"
        
        chain = gain_prompt | self.llm | StrOutputParser()
        response = chain.invoke({"exploration": potential_exploration, "knowledge": knowledge_str})
        
        try:
            score = float(response.strip())
            return max(0.0, min(1.0, score))
        except:
            return 0.5
    
    def explore(self, topic: str) -> Dict[str, Any]:
        """Explore a topic and gain knowledge."""
        print(f"\n🔍 Exploring: {topic}")
        
        # Assess novelty
        novelty = self.assess_novelty(topic)
        print(f"  Novelty: {novelty:.2f}")
        
        # Calculate information gain
        info_gain = self.calculate_information_gain(topic)
        print(f"  Information Gain: {info_gain:.2f}")
        
        # Intrinsic reward
        intrinsic_reward = (novelty + info_gain) / 2
        print(f"  Intrinsic Reward: {intrinsic_reward:.2f}")
        
        # Explore the topic
        exploration_prompt = ChatPromptTemplate.from_messages([
            ("system", "Explore this topic deeply and discover interesting insights."),
            ("user", "Topic: {topic}\n\nProvide interesting discoveries:")
        ])
        
        chain = exploration_prompt | self.llm | StrOutputParser()
        discoveries = chain.invoke({"topic": topic})
        
        # Update knowledge base
        self.knowledge_base.add(topic)
        self.explored_topics.append(topic)
        
        result = {
            "topic": topic,
            "novelty": novelty,
            "information_gain": info_gain,
            "intrinsic_reward": intrinsic_reward,
            "discoveries": discoveries
        }
        
        self.curiosity_log.append(result)
        
        print(f"\n  Discoveries:")
        for line in discoveries.split('\n')[:3]:
            if line.strip():
                print(f"    • {line.strip()}")
        
        return result
    
    def choose_next_exploration(self, candidates: List[str]) -> str:
        """Choose next exploration target based on curiosity."""
        print("\n📊 Evaluating exploration candidates...")
        
        scores = []
        for candidate in candidates:
            novelty = self.assess_novelty(candidate)
            info_gain = self.calculate_information_gain(candidate)
            curiosity_score = (novelty + info_gain) / 2
            scores.append((candidate, curiosity_score))
            print(f"  {candidate}: {curiosity_score:.2f}")
        
        # Choose highest curiosity score
        best_candidate = max(scores, key=lambda x: x[1])
        print(f"\n✨ Selected: {best_candidate[0]} (score: {best_candidate[1]:.2f})")
        
        return best_candidate[0]
    
    def get_exploration_summary(self) -> str:
        """Summarize exploration journey."""
        summary_prompt = ChatPromptTemplate.from_messages([
            ("system", "Summarize this curiosity-driven exploration journey."),
            ("user", """Explored Topics:
{topics}

Curiosity Log:
{log}

Provide a summary of:
1. Key discoveries
2. Exploration patterns
3. Knowledge growth
4. Most rewarding explorations""")
        ])
        
        topics = "\n".join([f"{i+1}. {t}" for i, t in enumerate(self.explored_topics)])
        log = "\n".join([
            f"Topic: {entry['topic']}, Reward: {entry['intrinsic_reward']:.2f}"
            for entry in self.curiosity_log
        ])
        
        chain = summary_prompt | self.llm | StrOutputParser()
        return chain.invoke({"topics": topics, "log": log})


def demonstrate_curiosity_driven_exploration():
    """Demonstrate curiosity-driven exploration pattern."""
    print("=== Curiosity-Driven Exploration Pattern ===\n")
    
    agent = CuriosityDrivenAgent()
    
    # Initial topic
    print("1. Starting Exploration from Initial Topic")
    print("-" * 50)
    agent.explore("Quantum Computing")
    
    # Generate curious questions
    print("\n2. Generating Curious Questions")
    print("-" * 50)
    questions = agent.generate_curious_questions("Quantum Computing")
    print("Curious questions generated:")
    for i, q in enumerate(questions, 1):
        print(f"{i}. {q}")
    
    # Explore based on curiosity
    print("\n3. Curiosity-Driven Exploration")
    print("-" * 50)
    
    # Candidate topics
    candidates = [
        "Quantum Entanglement Applications",
        "Quantum Error Correction",
        "Superconducting Qubits",
        "Quantum Algorithms for Chemistry"
    ]
    
    # Choose and explore
    for i in range(3):
        print(f"\n--- Exploration Round {i+1} ---")
        next_topic = agent.choose_next_exploration(candidates)
        agent.explore(next_topic)
        
        # Remove explored topic and add related ones
        candidates.remove(next_topic)
        
        # Generate new candidates based on discoveries
        if i < 2:
            new_candidates_prompt = ChatPromptTemplate.from_messages([
                ("system", "Suggest 2 related unexplored topics based on recent discovery."),
                ("user", "Recent exploration: {topic}\n\nSuggest new topics:")
            ])
            chain = new_candidates_prompt | agent.llm | StrOutputParser()
            new_suggestions = chain.invoke({"topic": next_topic})
            new_topics = [line.strip('- ').strip() for line in new_suggestions.split('\n') if line.strip()][:2]
            candidates.extend(new_topics)
    
    # Summary
    print("\n4. Exploration Journey Summary")
    print("-" * 50)
    summary = agent.get_exploration_summary()
    print(summary)
    
    print("\n=== Summary ===")
    print(f"Total explorations: {len(agent.explored_topics)}")
    print(f"Knowledge base size: {len(agent.knowledge_base)}")
    print(f"Average intrinsic reward: {sum(e['intrinsic_reward'] for e in agent.curiosity_log) / len(agent.curiosity_log):.2f}")
    print("\nCuriosity-driven exploration demonstrated with:")
    print("- Novelty assessment")
    print("- Information gain calculation")
    print("- Intrinsic reward-based decisions")
    print("- Autonomous exploration strategy")
    print("- Continuous knowledge expansion")


if __name__ == "__main__":
    demonstrate_curiosity_driven_exploration()

    

Agentic Design Pattern: Curiosity Driven Exploration

Pattern 118: Curiosity Driven Exploration – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Curiosity Driven Exploration pattern implementation.

---

Pattern Description

Pattern 118: Curiosity-Driven Exploration

Description: Agent explores environment based on intrinsic motivation driven by novelty, information gain, and prediction error.

Components: – Novelty detection – Information gain calculation – Exploration strategy – Intrinsic reward

Use Cases: – Open-ended learning – Exploration tasks – Discovery systems

LangChain Implementation: Uses LLM to generate curious questions and exploration strategies based on knowledge gaps and novelty.

---

File Statistics

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

---

Module Documentation

"""
Pattern 118: Curiosity-Driven Exploration

Description:
    Agent explores environment based on intrinsic motivation driven by
    novelty, information gain, and prediction error.

Components:
    - Novelty detection
    - Information gain calculation
    - Exploration strategy
    - Intrinsic reward

Use Cases:
    - Open-ended learning
    - Exploration tasks
    - Discovery systems

LangChain Implementation:
    Uses LLM to generate curious questions and exploration strategies based on
    knowledge gaps and novelty.
"""

Purpose: This module implements the Curiosity Driven Exploration pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import typing.Any
import typing.Dict
import typing.List
import typing.Set

Purpose: Core Python functionality: – 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.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: CuriosityDrivenAgent

Lines 35-205

Purpose:

Agent that explores based on intrinsic curiosity.

Methods:

__init__(self, model_name)

Lines 38-42

Purpose: Initialize the class instance with required parameters.

Parameters: – model_name: Language model instance

assess_novelty(self, topic)

Lines 44-65

Purpose:

Assess how novel a topic is based on existing knowledge.

Parameters: – topic: Topic

Returns: float

generate_curious_questions(self, context)

Lines 67-92

Purpose:

Generate questions driven by curiosity about knowledge gaps.

Parameters: – context: Context

Returns: <ast.Subscript object at 0x000001A06D7CC0D0>

calculate_information_gain(self, potential_exploration)

Lines 94-115

Purpose:

Calculate expected information gain from exploring a topic.

Parameters: – potential_exploration: Potential exploration

Returns: float

explore(self, topic)

Lines 117-161

Purpose:

Explore a topic and gain knowledge.

Parameters: – topic: Topic

Returns: <ast.Subscript object at 0x000001A06D88B210>

choose_next_exploration(self, candidates)

Lines 163-179

Purpose:

Choose next exploration target based on curiosity.

Parameters: – candidates: Candidates

Returns: str

get_exploration_summary(self)

Lines 181-205

Purpose:

Summarize exploration journey.

Parameters:

Returns: str

---

Functions

demonstrate_curiosity_driven_exploration()

Lines 208-274

Purpose:

Demonstrate curiosity-driven exploration pattern.

---

Key Concepts Summary

Pattern Overview

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

Core Components

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

When to Use

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

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

---

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

---

LangChain Components Used

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

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

---

Execution Flow

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

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

---

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

---

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

---

Related Patterns

• Pattern 117
• Pattern 119

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 Curiosity Driven Exploration 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: 118_curiosity_driven_exploration.py
Total Lines: 279
Pattern Category: AI Agent Design Pattern #118

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 117: Imitation Learning

Description:
    Agent learns by observing and imitating expert behavior through
    behavioral cloning and inverse reinforcement learning approaches.

Components:
    - Expert demonstrations
    - Behavioral cloning
    - Policy learning

Use Cases:
    - Robotics
    - Complex skill acquisition
    - Learning from examples

LangChain Implementation:
    Uses few-shot learning and demonstration-based prompting to imitate expert behavior.
"""

import os
from typing import List, Dict, Any
from dotenv import load_dotenv

from langchain_core.prompts import ChatPromptTemplate, FewShotChatMessagePromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

load_dotenv()


class ImitationLearningAgent:
    """Agent that learns by observing and imitating expert demonstrations."""
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        self.expert_demonstrations = []
        self.learned_policy = None
        
    def observe_expert(self, situation: str, expert_action: str, reasoning: str):
        """Observe and record expert demonstration."""
        demonstration = {
            "situation": situation,
            "action": expert_action,
            "reasoning": reasoning
        }
        self.expert_demonstrations.append(demonstration)
        print(f"Observed demonstration {len(self.expert_demonstrations)}:")
        print(f"  Situation: {situation}")
        print(f"  Action: {expert_action}")
        print(f"  Reasoning: {reasoning}")
        print()
        
    def learn_from_demonstrations(self):
        """Learn policy from all observed demonstrations."""
        learning_prompt = ChatPromptTemplate.from_messages([
            ("system", "Analyze these expert demonstrations and extract the underlying policy, patterns, and decision-making strategy."),
            ("user", """Expert Demonstrations:
{demonstrations}

Extract:
1. Common patterns in decision-making
2. Key factors considered
3. General policy rules
4. Decision strategy""")
        ])
        
        demos_text = "\n\n".join([
            f"Demo {i+1}:\nSituation: {d['situation']}\nAction: {d['action']}\nReasoning: {d['reasoning']}"
            for i, d in enumerate(self.expert_demonstrations)
        ])
        
        chain = learning_prompt | self.llm | StrOutputParser()
        self.learned_policy = chain.invoke({"demonstrations": demos_text})
        
        print("=== Learned Policy ===")
        print(self.learned_policy)
        print()
        
    def imitate_expert(self, new_situation: str) -> Dict[str, str]:
        """Apply learned policy to a new situation."""
        if not self.learned_policy:
            return {"error": "No policy learned yet. Need more demonstrations."}
        
        # Create few-shot prompt with demonstrations
        example_template = """Situation: {situation}
Expert Action: {action}
Reasoning: {reasoning}"""
        
        few_shot_prompt = FewShotChatMessagePromptTemplate(
            example_prompt=ChatPromptTemplate.from_messages([
                ("user", "Situation: {situation}"),
                ("assistant", "Action: {action}\nReasoning: {reasoning}")
            ]),
            examples=self.expert_demonstrations,
        )
        
        full_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are imitating an expert's decision-making style.

Learned Policy:
{learned_policy}

Use the learned policy and demonstrated patterns to decide actions for new situations."""),
            few_shot_prompt,
            ("user", "Situation: {new_situation}\n\nWhat action should be taken? Provide action and reasoning.")
        ])
        
        chain = full_prompt | self.llm | StrOutputParser()
        response = chain.invoke({
            "learned_policy": self.learned_policy,
            "new_situation": new_situation
        })
        
        # Parse response
        lines = response.split('\n')
        action = ""
        reasoning = ""
        
        for line in lines:
            if line.startswith("Action:"):
                action = line.replace("Action:", "").strip()
            elif line.startswith("Reasoning:"):
                reasoning = line.replace("Reasoning:", "").strip()
            elif action and not reasoning:
                reasoning = line.strip()
        
        return {
            "situation": new_situation,
            "action": action or response,
            "reasoning": reasoning or "Based on learned policy"
        }
    
    def evaluate_imitation(self, test_situation: str, expected_action: str) -> bool:
        """Evaluate how well the agent imitates expert behavior."""
        result = self.imitate_expert(test_situation)
        
        print(f"Test Situation: {test_situation}")
        print(f"Expected Action: {expected_action}")
        print(f"Agent Action: {result['action']}")
        print(f"Reasoning: {result['reasoning']}")
        
        # Simple similarity check
        similarity = expected_action.lower() in result['action'].lower()
        print(f"Match: {'✓' if similarity else '✗'}")
        print()
        
        return similarity


def demonstrate_imitation_learning():
    """Demonstrate imitation learning pattern."""
    print("=== Imitation Learning Pattern ===\n")
    
    agent = ImitationLearningAgent()
    
    # Scenario: Learning customer service expert behavior
    print("1. Observing Expert Demonstrations")
    print("-" * 50)
    
    # Demonstration 1: Angry customer
    agent.observe_expert(
        situation="Customer is angry about a delayed delivery",
        expert_action="Apologize sincerely, explain the situation, offer compensation (discount or free shipping on next order)",
        reasoning="Acknowledge emotions first, show empathy, provide solution to restore trust"
    )
    
    # Demonstration 2: Confused customer
    agent.observe_expert(
        situation="Customer is confused about how to use a product feature",
        expert_action="Ask clarifying questions, provide step-by-step guidance, offer visual aids or video tutorial",
        reasoning="Understand specific confusion point, break down into simple steps, provide multiple learning formats"
    )
    
    # Demonstration 3: Billing inquiry
    agent.observe_expert(
        situation="Customer questions an unexpected charge on their bill",
        expert_action="Review the charge details, explain clearly what it's for, provide itemized breakdown if needed",
        reasoning="Be transparent about charges, ensure customer understands, address concerns directly"
    )
    
    # Demonstration 4: Feature request
    agent.observe_expert(
        situation="Customer requests a feature that doesn't exist",
        expert_action="Thank them for feedback, explain current capabilities, document request for product team, suggest workaround if possible",
        reasoning="Value customer input, manage expectations, show commitment to improvement, help immediately if possible"
    )
    
    print("✓ Observed 4 expert demonstrations\n")
    
    # Learn from demonstrations
    print("2. Learning Policy from Demonstrations")
    print("-" * 50)
    agent.learn_from_demonstrations()
    
    # Test imitation on new situations
    print("3. Testing Imitation on New Situations")
    print("-" * 50)
    
    # Test 1: Product defect
    print("Test 1: Product Defect Situation")
    result = agent.imitate_expert("Customer reports a defective product that stopped working after one week")
    print(f"Situation: {result['situation']}")
    print(f"Action: {result['action']}")
    print(f"Reasoning: {result['reasoning']}")
    print()
    
    # Test 2: Refund request
    print("Test 2: Refund Request Situation")
    result = agent.imitate_expert("Customer wants a refund for a product they bought 2 months ago and barely used")
    print(f"Situation: {result['situation']}")
    print(f"Action: {result['action']}")
    print(f"Reasoning: {result['reasoning']}")
    print()
    
    # Test 3: Complex technical issue
    print("Test 3: Technical Issue Situation")
    result = agent.imitate_expert("Customer experiencing intermittent connectivity issues with the app")
    print(f"Situation: {result['situation']}")
    print(f"Action: {result['action']}")
    print(f"Reasoning: {result['reasoning']}")
    print()
    
    # Evaluation
    print("4. Evaluating Imitation Quality")
    print("-" * 50)
    agent.evaluate_imitation(
        "Customer complaint about rude staff member",
        "Apologize for experience, escalate to manager, ensure follow-up"
    )
    
    print("=== Summary ===")
    print("Imitation learning demonstrated with:")
    print("- Expert demonstration observation")
    print("- Policy learning from demonstrations")
    print("- Behavioral cloning for new situations")
    print("- Few-shot learning approach")
    print("- Pattern extraction from expert behavior")


if __name__ == "__main__":
    demonstrate_imitation_learning()

    

Agentic Design Pattern: Imitation Learning

Pattern 117: Imitation Learning – Line-by-Line Explanation

Overview

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

---

Pattern Description

Pattern 117: Imitation Learning

Description: Agent learns by observing and imitating expert behavior through behavioral cloning and inverse reinforcement learning approaches.

Components: – Expert demonstrations – Behavioral cloning – Policy learning

Use Cases: – Robotics – Complex skill acquisition – Learning from examples

LangChain Implementation: Uses few-shot learning and demonstration-based prompting to imitate expert behavior.

---

File Statistics

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

---

Module Documentation

"""
Pattern 117: Imitation Learning

Description:
    Agent learns by observing and imitating expert behavior through
    behavioral cloning and inverse reinforcement learning approaches.

Components:
    - Expert demonstrations
    - Behavioral cloning
    - Policy learning

Use Cases:
    - Robotics
    - Complex skill acquisition
    - Learning from examples

LangChain Implementation:
    Uses few-shot learning and demonstration-based prompting to imitate expert behavior.
"""

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

---

Import Statements

Standard Library Imports

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

Purpose: Core Python functionality: – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – langchain_core.prompts.FewShotChatMessagePromptTemplate: Regular expression operations – os: Environment variable access and system operations – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional)

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

Lines 33-149

Purpose:

Agent that learns by observing and imitating expert demonstrations.

Methods:

__init__(self, model_name)

Lines 36-39

Purpose: Initialize the class instance with required parameters.

Parameters: – model_name: Language model instance

observe_expert(self, situation, expert_action, reasoning)

Lines 41-53

Purpose:

Observe and record expert demonstration.

Parameters: – situation: Situation – expert_action: Expert action – reasoning: Reasoning

learn_from_demonstrations(self)

Lines 55-79

Purpose:

Learn policy from all observed demonstrations.

Parameters:

imitate_expert(self, new_situation)

Lines 81-133

Purpose:

Apply learned policy to a new situation.

Parameters: – new_situation: New situation

Returns: <ast.Subscript object at 0x000001A06D7F4090>

evaluate_imitation(self, test_situation, expected_action)

Lines 135-149

Purpose:

Evaluate how well the agent imitates expert behavior.

Parameters: – test_situation: Test situation – expected_action: Expected action

Returns: bool

---

Functions

demonstrate_imitation_learning()

Lines 152-239

Purpose:

Demonstrate imitation learning pattern.

---

Key Concepts Summary

Pattern Overview

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

Core Components

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

When to Use

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

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

---

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

---

LangChain Components Used

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

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

---

Execution Flow

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

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

---

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

---

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

---

Related Patterns

• Pattern 116
• Pattern 118

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 Imitation 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: 117_imitation_learning.py
Total Lines: 244
Pattern Category: AI Agent Design Pattern #117

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 116: Multi-Task Learning

Description:
    Agent learns multiple related tasks simultaneously using shared representations
    with task-specific heads for specialization.

Components:
    - Shared representations
    - Task-specific heads
    - Multi-task optimization

Use Cases:
    - Related task sets
    - Efficient training
    - Knowledge sharing

LangChain Implementation:
    Uses multi-task prompt templates and shared context for learning across tasks.
"""

import os
from typing import List, Dict, Any
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 MultiTaskLearningAgent:
    """Agent that learns multiple related tasks simultaneously."""
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.tasks = {}
        self.shared_knowledge = {}
        self.task_performance = {}
        
    def register_task(self, task_name: str, task_description: str, examples: List[Dict[str, str]]):
        """Register a new task for multi-task learning."""
        self.tasks[task_name] = {
            "description": task_description,
            "examples": examples
        }
        self.task_performance[task_name] = []
        
    def learn_from_shared_representations(self):
        """Extract shared knowledge from all registered tasks."""
        shared_prompt = ChatPromptTemplate.from_messages([
            ("system", "Analyze these tasks and identify shared patterns, concepts, and skills that apply across them."),
            ("user", "Tasks:\n{tasks}\n\nIdentify shared knowledge:")
        ])
        
        tasks_description = "\n\n".join([
            f"{name}: {info['description']}\nExamples: {info['examples']}"
            for name, info in self.tasks.items()
        ])
        
        chain = shared_prompt | self.llm | StrOutputParser()
        self.shared_knowledge["patterns"] = chain.invoke({"tasks": tasks_description})
        
        print("Shared Knowledge Extracted:")
        print(self.shared_knowledge["patterns"])
        print()
        
    def execute_task(self, task_name: str, input_data: str) -> str:
        """Execute a specific task using shared knowledge and task-specific head."""
        if task_name not in self.tasks:
            return f"Task '{task_name}' not registered"
        
        task_info = self.tasks[task_name]
        
        # Create task-specific prompt with shared knowledge
        task_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a multi-task learning agent.

Shared Knowledge (applies to all tasks):
{shared_knowledge}

Current Task: {task_name}
Description: {task_description}

Examples:
{examples}

Use both shared knowledge and task-specific patterns."""),
            ("user", "{input}")
        ])
        
        examples_str = "\n".join([
            f"Input: {ex['input']}\nOutput: {ex['output']}"
            for ex in task_info['examples']
        ])
        
        chain = task_prompt | self.llm | StrOutputParser()
        
        result = chain.invoke({
            "shared_knowledge": self.shared_knowledge.get("patterns", "None extracted yet"),
            "task_name": task_name,
            "task_description": task_info['description'],
            "examples": examples_str,
            "input": input_data
        })
        
        return result
    
    def evaluate_task(self, task_name: str, test_cases: List[Dict[str, str]]):
        """Evaluate performance on a specific task."""
        if task_name not in self.tasks:
            print(f"Task '{task_name}' not registered")
            return
        
        print(f"\n=== Evaluating Task: {task_name} ===")
        correct = 0
        total = len(test_cases)
        
        for case in test_cases:
            result = self.execute_task(task_name, case['input'])
            is_correct = case['expected'].lower() in result.lower()
            if is_correct:
                correct += 1
            print(f"Input: {case['input']}")
            print(f"Expected: {case['expected']}")
            print(f"Got: {result}")
            print(f"✓" if is_correct else "✗")
            print()
        
        accuracy = correct / total if total > 0 else 0
        self.task_performance[task_name].append(accuracy)
        print(f"Accuracy: {accuracy:.2%}\n")
        
    def get_task_correlations(self) -> str:
        """Analyze how tasks benefit from shared learning."""
        correlation_prompt = ChatPromptTemplate.from_messages([
            ("system", "Analyze how these tasks benefit from shared multi-task learning."),
            ("user", """Tasks and Performance:
{task_info}

Shared Knowledge:
{shared_knowledge}

Explain how tasks benefit from each other and which skills transfer across tasks.""")
        ])
        
        task_info = "\n".join([
            f"{name}: {self.tasks[name]['description']} - Performance: {self.task_performance[name]}"
            for name in self.tasks
        ])
        
        chain = correlation_prompt | self.llm | StrOutputParser()
        return chain.invoke({
            "task_info": task_info,
            "shared_knowledge": self.shared_knowledge.get("patterns", "Not extracted")
        })


def demonstrate_multi_task_learning():
    """Demonstrate multi-task learning pattern."""
    print("=== Multi-Task Learning Pattern ===\n")
    
    agent = MultiTaskLearningAgent()
    
    # Register related NLP tasks
    print("1. Registering Multiple Related Tasks")
    print("-" * 50)
    
    # Task 1: Sentiment Analysis
    agent.register_task(
        "sentiment_analysis",
        "Classify the sentiment of text as positive, negative, or neutral",
        [
            {"input": "I love this product!", "output": "positive"},
            {"input": "This is terrible.", "output": "negative"},
            {"input": "It's okay, nothing special.", "output": "neutral"}
        ]
    )
    
    # Task 2: Text Classification
    agent.register_task(
        "text_classification",
        "Classify text into categories: technology, sports, politics, entertainment",
        [
            {"input": "New AI breakthrough announced", "output": "technology"},
            {"input": "Championship game tonight", "output": "sports"},
            {"input": "Election results are in", "output": "politics"}
        ]
    )
    
    # Task 3: Intent Detection
    agent.register_task(
        "intent_detection",
        "Detect user intent: question, command, statement, request",
        [
            {"input": "What time is it?", "output": "question"},
            {"input": "Please send the report.", "output": "command"},
            {"input": "I think it will rain.", "output": "statement"}
        ]
    )
    
    print("✓ Registered 3 related NLP tasks\n")
    
    # Extract shared knowledge
    print("2. Learning Shared Representations")
    print("-" * 50)
    agent.learn_from_shared_representations()
    
    # Test on sentiment analysis
    print("3. Testing on Sentiment Analysis Task")
    print("-" * 50)
    test_cases = [
        {"input": "This is amazing and wonderful!", "expected": "positive"},
        {"input": "Worst experience ever", "expected": "negative"},
    ]
    agent.evaluate_task("sentiment_analysis", test_cases)
    
    # Test on text classification
    print("4. Testing on Text Classification Task")
    print("-" * 50)
    test_cases = [
        {"input": "Latest smartphone release announced", "expected": "technology"},
        {"input": "Team wins the finals", "expected": "sports"},
    ]
    agent.evaluate_task("text_classification", test_cases)
    
    # Analyze task correlations
    print("5. Analyzing Task Correlations and Knowledge Transfer")
    print("-" * 50)
    correlations = agent.get_task_correlations()
    print(correlations)
    print()
    
    print("=== Summary ===")
    print("Multi-task learning demonstrated with:")
    print("- Multiple related NLP tasks")
    print("- Shared knowledge extraction")
    print("- Task-specific execution with shared representations")
    print("- Knowledge transfer analysis")


if __name__ == "__main__":
    demonstrate_multi_task_learning()

    

Agentic Design Pattern: Multi Task Learning

Pattern 116: Multi Task Learning – Line-by-Line Explanation

Overview

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

---

Pattern Description

Pattern 116: Multi-Task Learning

Description: Agent learns multiple related tasks simultaneously using shared representations with task-specific heads for specialization.

Components: – Shared representations – Task-specific heads – Multi-task optimization

Use Cases: – Related task sets – Efficient training – Knowledge sharing

LangChain Implementation: Uses multi-task prompt templates and shared context for learning across tasks.

---

File Statistics

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

---

Module Documentation

"""
Pattern 116: Multi-Task Learning

Description:
    Agent learns multiple related tasks simultaneously using shared representations
    with task-specific heads for specialization.

Components:
    - Shared representations
    - Task-specific heads
    - Multi-task optimization

Use Cases:
    - Related task sets
    - Efficient training
    - Knowledge sharing

LangChain Implementation:
    Uses multi-task prompt templates and shared context for learning across tasks.
"""

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

---

Import Statements

Standard Library Imports

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

Purpose: Core Python functionality: – 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)

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

Lines 33-157

Purpose:

Agent that learns multiple related tasks simultaneously.

Methods:

__init__(self, model_name)

Lines 36-40

Purpose: Initialize the class instance with required parameters.

Parameters: – model_name: Language model instance

register_task(self, task_name, task_description, examples)

Lines 42-48

Purpose:

Register a new task for multi-task learning.

Parameters: – task_name: Task name – task_description: Task description – examples: Examples

learn_from_shared_representations(self)

Lines 50-67

Purpose:

Extract shared knowledge from all registered tasks.

Parameters:

execute_task(self, task_name, input_data)

Lines 69-108

Purpose:

Execute a specific task using shared knowledge and task-specific head.

Parameters: – task_name: Task name – input_data: Input data

Returns: str

evaluate_task(self, task_name, test_cases)

Lines 110-133

Purpose:

Evaluate performance on a specific task.

Parameters: – task_name: Task name – test_cases: Test cases

get_task_correlations(self)

Lines 135-157

Purpose:

Analyze how tasks benefit from shared learning.

Parameters:

Returns: str

---

Functions

demonstrate_multi_task_learning()

Lines 160-240

Purpose:

Demonstrate multi-task learning pattern.

---

Key Concepts Summary

Pattern Overview

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

Core Components

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

When to Use

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

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

---

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

---

LangChain Components Used

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

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

---

Execution Flow

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

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

---

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

---

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

---

Related Patterns

• Pattern 115
• Pattern 117

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 Task 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: 116_multi_task_learning.py
Total Lines: 245
Pattern Category: AI Agent Design Pattern #116

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 115: Transfer Learning Agent

Description:
    Transfer Learning enables agents to leverage knowledge learned from one
    task or domain to improve performance on related tasks. This pattern
    identifies transferable knowledge, adapts it to new contexts, and combines
    it with task-specific learning for faster adaptation and better performance.
    
    Essential for agents that need to work across multiple related domains or
    tasks, transfer learning reduces the data and time needed to achieve good
    performance in new areas.

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

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

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

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

import os
from typing import List, Dict, Any, Optional
from dotenv import load_dotenv
from datetime import datetime

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, FewShotChatMessagePromptTemplate
from langchain_core.output_parsers import StrOutputParser

load_dotenv()


class DomainKnowledge:
    """Represents knowledge from a source domain"""
    
    def __init__(self, domain_name: str):
        self.domain_name = domain_name
        self.examples: List[Dict[str, str]] = []
        self.patterns: List[str] = []
        self.concepts: Dict[str, str] = {}
    
    def add_example(self, input_text: str, output_text: str, pattern: str = ""):
        """Add an example from this domain"""
        self.examples.append({
            "input": input_text,
            "output": output_text,
            "pattern": pattern
        })
    
    def add_pattern(self, pattern: str):
        """Add a learned pattern"""
        if pattern not in self.patterns:
            self.patterns.append(pattern)
    
    def add_concept(self, concept: str, description: str):
        """Add a domain concept"""
        self.concepts[concept] = description


class TransferLearningAgent:
    """Agent that transfers knowledge between domains"""
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.source_domains: Dict[str, DomainKnowledge] = {}
        self.current_domain: Optional[str] = None
    
    def add_source_domain(self, domain: DomainKnowledge):
        """Add a source domain with learned knowledge"""
        self.source_domains[domain.domain_name] = domain
        print(f"Added source domain: {domain.domain_name}")
        print(f"  - {len(domain.examples)} examples")
        print(f"  - {len(domain.patterns)} patterns")
        print(f"  - {len(domain.concepts)} concepts")
    
    def identify_relevant_source(self, target_task: str) -> Optional[DomainKnowledge]:
        """Identify most relevant source domain for transfer"""
        if not self.source_domains:
            return None
        
        # Use LLM to identify most relevant source
        domains_desc = "\n".join([
            f"- {name}: {len(d.examples)} examples, patterns: {', '.join(d.patterns[:3])}"
            for name, d in self.source_domains.items()
        ])
        
        selection_prompt = f"""Given a target task and available source domains, identify which source domain would provide the most useful transferable knowledge.

Target task: {target_task}

Available source domains:
{domains_desc}

Which source domain is most relevant? Answer with just the domain name."""
        
        response = self.llm.invoke(selection_prompt).content.strip()
        
        # Find matching domain
        for domain_name in self.source_domains:
            if domain_name.lower() in response.lower():
                print(f"\nSelected source domain: {domain_name}")
                return self.source_domains[domain_name]
        
        # Default to first domain
        return list(self.source_domains.values())[0]
    
    def adapt_examples(self, source_domain: DomainKnowledge, 
                      target_domain: str, n: int = 3) -> List[Dict[str, str]]:
        """Adapt source examples to target domain"""
        if not source_domain.examples:
            return []
        
        # Take first n examples
        source_examples = source_domain.examples[:n]
        
        # Create adaptation prompt
        examples_str = "\n\n".join([
            f"Input: {ex['input']}\nOutput: {ex['output']}"
            for ex in source_examples
        ])
        
        adaptation_prompt = f"""Adapt these examples from {source_domain.domain_name} domain to {target_domain} domain.
Keep the same pattern and structure, but change the content to fit the target domain.

Source examples ({source_domain.domain_name}):
{examples_str}

Generate {n} adapted examples for {target_domain} domain in the same format."""
        
        response = self.llm.invoke(adaptation_prompt).content
        
        # Parse adapted examples (simplified parsing)
        adapted = []
        lines = response.split('\n')
        current_input = None
        
        for line in lines:
            if line.startswith('Input:'):
                current_input = line.replace('Input:', '').strip()
            elif line.startswith('Output:') and current_input:
                output = line.replace('Output:', '').strip()
                adapted.append({"input": current_input, "output": output})
                current_input = None
        
        return adapted[:n] if adapted else source_examples[:n]
    
    def transfer_and_respond(self, target_task: str, target_domain: str,
                            query: str) -> Dict[str, Any]:
        """Transfer knowledge and respond to query"""
        print(f"\n{'='*60}")
        print(f"Target Domain: {target_domain}")
        print(f"Target Task: {target_task}")
        print(f"Query: {query}")
        print(f"{'='*60}")
        
        # Identify relevant source domain
        source_domain = self.identify_relevant_source(target_task)
        
        if source_domain:
            print(f"\nTransferring knowledge from: {source_domain.domain_name}")
            
            # Adapt examples to target domain
            adapted_examples = self.adapt_examples(source_domain, target_domain, n=3)
            print(f"Adapted {len(adapted_examples)} examples to target domain")
            
            # Extract transferable patterns
            patterns_str = "\n".join([f"- {p}" for p in source_domain.patterns])
            
            # Create transfer-enhanced prompt
            if adapted_examples:
                example_prompt = ChatPromptTemplate.from_messages([
                    ("human", "{input}"),
                    ("ai", "{output}")
                ])
                
                few_shot_prompt = FewShotChatMessagePromptTemplate(
                    example_prompt=example_prompt,
                    examples=adapted_examples
                )
                
                prompt = ChatPromptTemplate.from_messages([
                    ("system", f"""You are an AI assistant for {target_domain}.

You are leveraging transferred knowledge from {source_domain.domain_name}.

Transferable patterns:
{patterns_str}

Apply these learned patterns to the current domain:"""),
                    few_shot_prompt,
                    ("human", "{input}")
                ])
            else:
                prompt = ChatPromptTemplate.from_messages([
                    ("system", f"""You are an AI assistant for {target_domain}.

You are leveraging transferred knowledge from {source_domain.domain_name}.

Transferable patterns:
{patterns_str}

Apply these learned patterns to the current task."""),
                    ("human", "{input}")
                ])
            
            chain = prompt | self.llm | StrOutputParser()
            response = chain.invoke({"input": query})
        else:
            # No source domain available, use base prompt
            print("\nNo source domain available, using base knowledge")
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", f"You are an AI assistant for {target_domain}."),
                ("human", "{input}")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            response = chain.invoke({"input": query})
        
        print(f"\nResponse:")
        print(response)
        
        return {
            "target_domain": target_domain,
            "target_task": target_task,
            "query": query,
            "source_domain": source_domain.domain_name if source_domain else None,
            "response": response,
            "timestamp": datetime.now().isoformat()
        }
    
    def compare_with_without_transfer(self, target_task: str, target_domain: str,
                                     query: str) -> Dict[str, Any]:
        """Compare performance with and without transfer learning"""
        print(f"\n{'='*70}")
        print(f"COMPARISON: With vs Without Transfer Learning")
        print(f"{'='*70}")
        
        # Response with transfer
        print("\n--- WITH TRANSFER LEARNING ---")
        with_transfer = self.transfer_and_respond(target_task, target_domain, query)
        
        # Response without transfer (base model)
        print("\n--- WITHOUT TRANSFER LEARNING (Baseline) ---")
        base_prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are an AI assistant for {target_domain}."),
            ("human", "{input}")
        ])
        base_chain = base_prompt | self.llm | StrOutputParser()
        without_transfer = base_chain.invoke({"input": query})
        
        print(f"\nBaseline Response:")
        print(without_transfer)
        
        return {
            "with_transfer": with_transfer,
            "without_transfer": without_transfer
        }


def demonstrate_transfer_learning():
    """Demonstrate transfer learning pattern"""
    print("\n" + "="*70)
    print("TRANSFER LEARNING AGENT DEMONSTRATION")
    print("="*70)
    
    agent = TransferLearningAgent()
    
    # Create source domain: Customer Support
    print("\n" + "="*70)
    print("Setting Up Source Domain Knowledge")
    print("="*70)
    
    customer_support = DomainKnowledge("customer_support")
    
    # Add examples
    customer_support.add_example(
        "How do I reset my password?",
        "To reset your password: 1) Click 'Forgot Password' on login page, "
        "2) Enter your email, 3) Check your email for reset link, 4) Create new password.",
        "step_by_step_instructions"
    )
    customer_support.add_example(
        "Where is my order?",
        "To track your order: 1) Log into your account, 2) Go to 'Orders', "
        "3) Find your order number, 4) Click 'Track' to see current status.",
        "step_by_step_instructions"
    )
    customer_support.add_example(
        "How do I cancel my subscription?",
        "To cancel subscription: 1) Go to Account Settings, 2) Select 'Subscriptions', "
        "3) Choose subscription to cancel, 4) Confirm cancellation.",
        "step_by_step_instructions"
    )
    
    # Add patterns
    customer_support.add_pattern("Provide step-by-step instructions")
    customer_support.add_pattern("Be clear and concise")
    customer_support.add_pattern("Offer alternative solutions")
    customer_support.add_pattern("Include troubleshooting tips")
    
    # Add concepts
    customer_support.add_concept("account_management", "User account operations")
    customer_support.add_concept("order_processing", "Order lifecycle and tracking")
    
    agent.add_source_domain(customer_support)
    
    # Example 1: Transfer to Technical Support
    print("\n" + "="*70)
    print("Example 1: Transfer to Technical Support Domain")
    print("="*70)
    
    result1 = agent.transfer_and_respond(
        target_task="Help users troubleshoot technical issues",
        target_domain="technical_support",
        query="How do I fix a blue screen error?"
    )
    
    # Example 2: Transfer to Medical Assistant
    print("\n" + "="*70)
    print("Example 2: Transfer to Medical Assistant Domain")
    print("="*70)
    
    result2 = agent.transfer_and_respond(
        target_task="Provide medical information and guidance",
        target_domain="medical_assistant",
        query="How do I check my blood pressure at home?"
    )
    
    # Example 3: Compare with and without transfer
    print("\n" + "="*70)
    print("Example 3: Comparing With/Without Transfer")
    print("="*70)
    
    comparison = agent.compare_with_without_transfer(
        target_task="Help users with cooking recipes",
        target_domain="cooking_assistant",
        query="How do I make homemade pasta?"
    )
    
    # Summary
    print("\n" + "="*70)
    print("SUMMARY")
    print("="*70)
    print("""
Transfer Learning Pattern demonstrated:

Key Features:
1. Source Domain Selection - Identify most relevant source knowledge
2. Knowledge Adaptation - Adapt examples and patterns to new domain
3. Pattern Transfer - Transfer learned patterns across domains
4. Few-Shot Enhancement - Use adapted examples for better performance
5. Comparative Evaluation - Compare with/without transfer

Transfer Mechanisms:
- Example adaptation across domains
- Pattern and structure transfer
- Concept mapping between domains
- Meta-learning of general strategies
- Domain-specific fine-tuning

Applications:
- Multi-domain chatbots (customer → technical support)
- Cross-lingual NLP (English → Spanish)
- Domain adaptation (medical → legal documentation)
- Few-shot learning in new domains
- Skill transfer in robotics

Benefits:
- Faster adaptation to new domains
- Better performance with limited data
- Leverages existing knowledge
- Reduces training time and data needs
- Improves generalization

The transfer learning pattern enables agents to leverage knowledge from
related domains, significantly improving performance on new tasks with
limited domain-specific training.
    """)


if __name__ == "__main__":
    demonstrate_transfer_learning()

    

Agentic Design Pattern: Transfer Learning

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

Overview

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

---

Pattern Description

Pattern 115: Transfer Learning Agent

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

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

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

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

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

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

---

File Statistics

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

---

Module Documentation

"""
Pattern 115: Transfer Learning Agent

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

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

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

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

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

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

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

---

Import Statements

Standard Library Imports

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

Purpose: Core Python functionality: – datetime.datetime: Date and time handling – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – langchain_core.prompts.FewShotChatMessagePromptTemplate: Regular expression operations – os: Environment variable access and system operations – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI

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

Other Imports

from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: DomainKnowledge

Lines 52-76

Purpose:

Represents knowledge from a source domain

Methods:

__init__(self, domain_name)

Lines 55-59

Purpose: Initialize the class instance with required parameters.

Parameters: – domain_name: Domain name

add_example(self, input_text, output_text, pattern)

Lines 61-67

Purpose:

Add an example from this domain

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

add_pattern(self, pattern)

Lines 69-72

Purpose:

Add a learned pattern

Parameters: – pattern: Pattern

add_concept(self, concept, description)

Lines 74-76

Purpose:

Add a domain concept

Parameters: – concept: Concept – description: Description

---

Class: TransferLearningAgent

Lines 79-277

Purpose:

Agent that transfers knowledge between domains

Methods:

__init__(self, model_name)

Lines 82-85

Purpose: Initialize the class instance with required parameters.

Parameters: – model_name: Language model instance

add_source_domain(self, domain)

Lines 87-93

Purpose:

Add a source domain with learned knowledge

Parameters: – domain: Domain

identify_relevant_source(self, target_task)

Lines 95-124

Purpose:

Identify most relevant source domain for transfer

Parameters: – target_task: Target task

Returns: <ast.Subscript object at 0x000001A06D7FEF50>

adapt_examples(self, source_domain, target_domain, n)

Lines 126-164

Purpose:

Adapt source examples to target domain

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

Returns: <ast.Subscript object at 0x000001A06D72EDD0>

transfer_and_respond(self, target_task, target_domain, query)

Lines 166-249

Purpose:

Transfer knowledge and respond to query

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

Returns: <ast.Subscript object at 0x000001A06D79B310>

compare_with_without_transfer(self, target_task, target_domain, query)

Lines 251-277

Purpose:

Compare performance with and without transfer learning

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

Returns: <ast.Subscript object at 0x000001A06D7E43D0>

---

Functions

demonstrate_transfer_learning()

Lines 280-398

Purpose:

Demonstrate transfer learning pattern

---

Key Concepts Summary

Pattern Overview

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

Core Components

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

When to Use

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

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

---

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

---

LangChain Components Used

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

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

---

Execution Flow

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

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

---

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

---

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

---

Related Patterns

• Pattern 114
• Pattern 116

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The Transfer Learning pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

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

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

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

---

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

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 114: Online Learning

Description:
    Online Learning enables agents to continuously learn and adapt from incoming
    data streams in real-time. Unlike batch learning, online learning updates
    the agent's knowledge incrementally as new data arrives, allowing adaptation
    to changing environments and concept drift.
    
    This pattern is crucial for agents operating in dynamic environments where
    patterns, user preferences, or optimal strategies change over time. It
    combines incremental updates, drift detection, and continuous adaptation.

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

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

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

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

import os
from typing import List, Dict, Any, Optional, Deque
from collections import deque
from dotenv import load_dotenv
from datetime import datetime
import json

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, FewShotChatMessagePromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain.memory import ConversationBufferMemory

load_dotenv()


class ExampleBuffer:
    """Manages a buffer of learning examples"""
    
    def __init__(self, max_size: int = 100):
        self.max_size = max_size
        self.examples: Deque[Dict[str, Any]] = deque(maxlen=max_size)
        self.performance_history: List[float] = []
    
    def add_example(self, input_text: str, output_text: str, 
                    feedback_score: Optional[float] = None):
        """Add a new example to the buffer"""
        example = {
            "input": input_text,
            "output": output_text,
            "feedback": feedback_score,
            "timestamp": datetime.now().isoformat()
        }
        self.examples.append(example)
        
        if feedback_score is not None:
            self.performance_history.append(feedback_score)
    
    def get_recent_examples(self, n: int = 5) -> List[Dict[str, str]]:
        """Get n most recent examples"""
        recent = list(self.examples)[-n:]
        return [{"input": ex["input"], "output": ex["output"]} for ex in recent]
    
    def get_best_examples(self, n: int = 5) -> List[Dict[str, str]]:
        """Get n best-rated examples"""
        if not self.examples:
            return []
        
        # Filter examples with feedback
        rated = [ex for ex in self.examples if ex["feedback"] is not None]
        if not rated:
            return self.get_recent_examples(n)
        
        # Sort by feedback score
        sorted_examples = sorted(rated, key=lambda x: x["feedback"], reverse=True)
        top_n = sorted_examples[:n]
        
        return [{"input": ex["input"], "output": ex["output"]} for ex in top_n]
    
    def detect_drift(self, window_size: int = 20) -> bool:
        """Detect if performance is degrading (concept drift)"""
        if len(self.performance_history)  drift_threshold
    
    def get_performance_trend(self) -> str:
        """Get description of performance trend"""
        if len(self.performance_history) = 20 else []
        
        recent_avg = sum(recent) / len(recent)
        
        if not older:
            return f"recent_avg: {recent_avg:.2f}"
        
        older_avg = sum(older) / len(older)
        
        if recent_avg > older_avg + 0.1:
            return f"improving (from {older_avg:.2f} to {recent_avg:.2f})"
        elif recent_avg  ChatPromptTemplate:
        """Create prompt with dynamic examples from learning buffer"""
        # Get examples from buffer
        if use_best:
            examples = self.example_buffer.get_best_examples(n=3)
        else:
            examples = self.example_buffer.get_recent_examples(n=3)
        
        if not examples:
            # No examples yet, use base prompt
            return ChatPromptTemplate.from_messages([
                ("system", self.base_system_prompt),
                ("human", "{input}")
            ])
        
        # Create few-shot prompt with learned examples
        few_shot_prompt = FewShotChatMessagePromptTemplate(
            example_prompt=self.example_prompt,
            examples=examples
        )
        
        # Get performance trend
        trend = self.example_buffer.get_performance_trend()
        system_msg = f"""{self.base_system_prompt}

Learning Status: {len(self.example_buffer.examples)} examples learned
Performance Trend: {trend}

Learn from these example interactions:"""
        
        return ChatPromptTemplate.from_messages([
            ("system", system_msg),
            few_shot_prompt,
            ("human", "{input}")
        ])
    
    def respond(self, user_input: str) -> str:
        """Generate response using current learned knowledge"""
        # Create dynamic prompt with learned examples
        prompt = self.create_dynamic_prompt(use_best=True)
        
        # Generate response
        chain = prompt | self.llm | StrOutputParser()
        response = chain.invoke({"input": user_input})
        
        self.total_interactions += 1
        return response
    
    def learn_from_feedback(self, user_input: str, response: str, 
                           feedback_score: float):
        """Learn from user feedback on a response"""
        print(f"\nLearning from feedback (score: {feedback_score})...")
        
        # Add to example buffer
        self.example_buffer.add_example(user_input, response, feedback_score)
        
        # Check for drift
        if self.example_buffer.detect_drift():
            print("⚠️  Concept drift detected! Performance declining.")
            print("Consider adjusting learning strategy or model.")
        
        # Show learning progress
        trend = self.example_buffer.get_performance_trend()
        print(f"Performance trend: {trend}")
    
    def interact_and_learn(self, user_input: str, feedback_score: Optional[float] = None) -> Dict[str, Any]:
        """Complete interaction cycle with learning"""
        print(f"\n{'='*60}")
        print(f"User: {user_input}")
        print(f"{'='*60}")
        
        # Generate response
        response = self.respond(user_input)
        print(f"\nAssistant: {response}")
        
        # Learn from feedback if provided
        if feedback_score is not None:
            self.learn_from_feedback(user_input, response, feedback_score)
        
        return {
            "input": user_input,
            "response": response,
            "feedback": feedback_score,
            "total_interactions": self.total_interactions,
            "examples_learned": len(self.example_buffer.examples),
            "performance_trend": self.example_buffer.get_performance_trend()
        }
    
    def get_learning_stats(self) -> Dict[str, Any]:
        """Get statistics about learning progress"""
        return {
            "total_interactions": self.total_interactions,
            "examples_in_buffer": len(self.example_buffer.examples),
            "performance_history_size": len(self.example_buffer.performance_history),
            "performance_trend": self.example_buffer.get_performance_trend(),
            "drift_detected": self.example_buffer.detect_drift()
        }


def demonstrate_online_learning():
    """Demonstrate online learning pattern"""
    print("\n" + "="*70)
    print("ONLINE LEARNING DEMONSTRATION")
    print("="*70)
    
    # Create agent for customer support domain
    agent = OnlineLearningAgent(domain="customer support")
    
    # Example 1: Initial interactions (building knowledge)
    print("\n" + "="*70)
    print("Example 1: Initial Learning Phase")
    print("="*70)
    
    interactions = [
        ("How do I reset my password?", 0.9),
        ("Where can I find my order history?", 0.85),
        ("How do I contact support?", 0.95),
        ("What is your return policy?", 0.8),
        ("How long does shipping take?", 0.9),
    ]
    
    for user_input, feedback in interactions:
        agent.interact_and_learn(user_input, feedback)
    
    # Show learning stats
    print("\n" + "-"*60)
    print("Learning Statistics After Initial Phase:")
    stats = agent.get_learning_stats()
    print(json.dumps(stats, indent=2))
    
    # Example 2: Applying learned knowledge
    print("\n" + "="*70)
    print("Example 2: Applying Learned Knowledge")
    print("="*70)
    
    new_queries = [
        "I forgot my password, what should I do?",  # Similar to learned
        "How can I see my past orders?",  # Similar to learned
    ]
    
    for query in new_queries:
        result = agent.interact_and_learn(query, feedback_score=0.9)
    
    # Example 3: Continuous learning with feedback
    print("\n" + "="*70)
    print("Example 3: Continuous Learning with Mixed Feedback")
    print("="*70)
    
    mixed_interactions = [
        ("Can I change my delivery address?", 0.7),  # Medium feedback
        ("Do you offer gift wrapping?", 0.85),  # Good feedback
        ("How do I cancel my subscription?", 0.6),  # Lower feedback
        ("What payment methods do you accept?", 0.9),  # High feedback
    ]
    
    for user_input, feedback in mixed_interactions:
        agent.interact_and_learn(user_input, feedback)
    
    # Example 4: Detecting concept drift
    print("\n" + "="*70)
    print("Example 4: Simulating Performance Degradation (Drift)")
    print("="*70)
    
    # Simulate declining performance
    drift_interactions = [
        ("Question about new feature X", 0.4),  # New topics, poor performance
        ("Tell me about feature Y", 0.3),
        ("How does feature Z work?", 0.35),
        ("Feature A questions", 0.4),
        ("Feature B support", 0.3),
    ]
    
    for user_input, feedback in drift_interactions:
        agent.interact_and_learn(user_input, feedback)
    
    # Final statistics
    print("\n" + "="*70)
    print("Final Learning Statistics:")
    print("="*70)
    final_stats = agent.get_learning_stats()
    print(json.dumps(final_stats, indent=2))
    
    # Summary
    print("\n" + "="*70)
    print("SUMMARY")
    print("="*70)
    print("""
Online Learning Pattern demonstrated:

Key Features:
1. Incremental Learning - Updates knowledge with each interaction
2. Example Buffer - Maintains recent and best examples
3. Performance Tracking - Monitors learning effectiveness
4. Drift Detection - Identifies when patterns change
5. Dynamic Adaptation - Adjusts prompts based on learned examples

Learning Mechanisms:
- Few-shot learning with best examples
- Continuous feedback integration
- Performance trend analysis
- Concept drift detection
- Adaptive example selection

Applications:
- Customer support bots improving from interactions
- Personalized assistants learning user preferences
- Recommendation systems adapting to trends
- Fraud detection with evolving patterns
- Adaptive tutoring systems

Benefits:
- Continuous improvement without retraining
- Adaptation to changing patterns
- Personalization based on interaction history
- Early warning for performance degradation
- Efficient learning from limited examples

Limitations:
- Requires consistent feedback
- May need periodic full retraining
- Can accumulate biases from feedback
- Limited by context window size
    """)


if __name__ == "__main__":
    demonstrate_online_learning()

    

Agentic Design Pattern: Online Learning

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

Overview

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

---

Pattern Description

Pattern 114: Online Learning

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

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

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

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

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

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

---

File Statistics

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

---

Module Documentation

"""
Pattern 114: Online Learning

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

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

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

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

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

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

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

---

Import Statements

Standard Library Imports

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

Purpose: Core Python functionality: – collections.deque: Specialized container datatypes – datetime.datetime: Date and time handling – json: JSON serialization/deserialization – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – langchain_core.prompts.FewShotChatMessagePromptTemplate: Regular expression operations – os: Environment variable access and system operations – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Deque: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain.memory.ConversationBufferMemory
from langchain_openai.ChatOpenAI

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

Other Imports

from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: ExampleBuffer

Lines 56-135

Purpose:

Manages a buffer of learning examples

Methods:

__init__(self, max_size)

Lines 59-62

Purpose: Initialize the class instance with required parameters.

Parameters: – max_size: Max size

add_example(self, input_text, output_text, feedback_score)

Lines 64-76

Purpose:

Add a new example to the buffer

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

get_recent_examples(self, n)

Lines 78-81

Purpose:

Get n most recent examples

Parameters: – n: N

Returns: <ast.Subscript object at 0x000001A06D88BE90>

get_best_examples(self, n)

Lines 83-97

Purpose:

Get n best-rated examples

Parameters: – n: N

Returns: <ast.Subscript object at 0x000001A06D765F50>

detect_drift(self, window_size)

Lines 99-113

Purpose:

Detect if performance is degrading (concept drift)

Parameters: – window_size: Window size

Returns: bool

get_performance_trend(self)

Lines 115-135

Purpose:

Get description of performance trend

Parameters:

Returns: str

---

Class: OnlineLearningAgent

Lines 138-254

Purpose:

Agent that learns continuously from interactions

Methods:

__init__(self, model_name, domain)

Lines 141-156

Purpose: Initialize the class instance with required parameters.

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

create_dynamic_prompt(self, use_best)

Lines 158-192

Purpose:

Create prompt with dynamic examples from learning buffer

Parameters: – use_best: Use best

Returns: ChatPromptTemplate

respond(self, user_input)

Lines 194-204

Purpose:

Generate response using current learned knowledge

Parameters: – user_input: Input data

Returns: str

learn_from_feedback(self, user_input, response, feedback_score)

Lines 206-221

Purpose:

Learn from user feedback on a response

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

interact_and_learn(self, user_input, feedback_score)

Lines 223-244

Purpose:

Complete interaction cycle with learning

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

Returns: <ast.Subscript object at 0x000001A06D16DBD0>

get_learning_stats(self)

Lines 246-254

Purpose:

Get statistics about learning progress

Parameters:

Returns: <ast.Subscript object at 0x000001A06E094F10>

---

Functions

demonstrate_online_learning()

Lines 257-380

Purpose:

Demonstrate online learning pattern

---

Key Concepts Summary

Pattern Overview

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

Core Components

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

When to Use

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

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

---

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

---

LangChain Components Used

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

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

---

Execution Flow

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

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

---

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

---

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

---

Related Patterns

• Pattern 113
• Pattern 115

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The Online Learning pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

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

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

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

---

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

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 113: Contextual Adaptation

Description:
    Contextual Adaptation enables agents to dynamically adjust their behavior,
    communication style, and strategies based on the detected context. The
    pattern monitors multiple context dimensions (user preferences, environment,
    task type, constraints) and adapts agent behavior accordingly.
    
    This pattern is essential for creating flexible, user-friendly agents that
    can operate effectively across diverse situations and user needs. It combines
    context detection, preference learning, and dynamic behavior adjustment.

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

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

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

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

import os
from typing import List, Dict, Any, Optional
from dotenv import load_dotenv
from datetime import datetime
from enum import Enum

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain.memory import ConversationBufferMemory

load_dotenv()


class UserExperience(Enum):
    """User experience levels"""
    BEGINNER = "beginner"
    INTERMEDIATE = "intermediate"
    ADVANCED = "advanced"
    EXPERT = "expert"


class Formality(Enum):
    """Communication formality levels"""
    CASUAL = "casual"
    NEUTRAL = "neutral"
    FORMAL = "formal"
    VERY_FORMAL = "very_formal"


class TaskUrgency(Enum):
    """Task urgency levels"""
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"


class Context:
    """Represents current context"""
    
    def __init__(self):
        # User context
        self.user_experience = UserExperience.INTERMEDIATE
        self.user_preferences = {}
        self.user_goals = []
        
        # Environment context
        self.time_of_day = "morning"
        self.location = "office"
        self.available_resources = []
        
        # Task context
        self.task_type = "general"
        self.task_complexity = "medium"
        self.task_urgency = TaskUrgency.MEDIUM
        
        # Social context
        self.formality = Formality.NEUTRAL
        self.relationship = "professional"
        
        # Technical context
        self.device = "desktop"
        self.bandwidth = "high"
        
        # Domain context
        self.domain = "general"
        self.domain_conventions = {}
    
    def to_dict(self) -> Dict[str, Any]:
        """Convert context to dictionary"""
        return {
            "user": {
                "experience": self.user_experience.value,
                "preferences": self.user_preferences,
                "goals": self.user_goals
            },
            "environment": {
                "time": self.time_of_day,
                "location": self.location,
                "resources": self.available_resources
            },
            "task": {
                "type": self.task_type,
                "complexity": self.task_complexity,
                "urgency": self.task_urgency.value
            },
            "social": {
                "formality": self.formality.value,
                "relationship": self.relationship
            },
            "technical": {
                "device": self.device,
                "bandwidth": self.bandwidth
            },
            "domain": {
                "name": self.domain,
                "conventions": self.domain_conventions
            }
        }
    
    def summarize(self) -> str:
        """Generate human-readable context summary"""
        return f"""
Current Context:
- User: {self.user_experience.value} level, wants {', '.join(self.user_goals) if self.user_goals else 'assistance'}
- Environment: {self.time_of_day} at {self.location}
- Task: {self.task_type} ({self.task_complexity} complexity, {self.task_urgency.value} urgency)
- Communication: {self.formality.value} style, {self.relationship} relationship
- Technical: {self.device} with {self.bandwidth} bandwidth
- Domain: {self.domain}
"""


class ContextualAdaptationAgent:
    """Agent that adapts behavior based on context"""
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.context = Context()
        self.memory = ConversationBufferMemory(return_messages=True)
        
        # Adaptive response generation
        self.adaptive_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are an adaptive AI assistant that adjusts your behavior based on context.

{context_summary}

Adaptation Guidelines:
- For beginners: Use simple language, explain concepts, provide examples
- For experts: Be concise, use technical terms, assume background knowledge
- For casual contexts: Be friendly and conversational
- For formal contexts: Be professional and structured
- For urgent tasks: Be direct and action-oriented
- For complex tasks: Break down steps and provide detailed guidance

Adapt your response to match the context appropriately."""),
            ("human", "{query}")
        ])
        
        # Context detection prompt
        self.context_detector = ChatPromptTemplate.from_messages([
            ("system", """Analyze the user's query and extract context information.

Determine:
1. User experience level (beginner/intermediate/advanced/expert)
2. Formality expected (casual/neutral/formal/very_formal)
3. Task urgency (low/medium/high/critical)
4. Task type and complexity

Return JSON with: experience, formality, urgency, task_type, complexity"""),
            ("human", "{query}")
        ])
        
        self.response_chain = self.adaptive_prompt | self.llm | StrOutputParser()
        self.detector_chain = self.context_detector | self.llm | JsonOutputParser()
    
    def detect_context(self, query: str) -> Dict[str, Any]:
        """Detect context from user query"""
        try:
            detected = self.detector_chain.invoke({"query": query})
            
            # Update context based on detection
            if "experience" in detected:
                exp_map = {
                    "beginner": UserExperience.BEGINNER,
                    "intermediate": UserExperience.INTERMEDIATE,
                    "advanced": UserExperience.ADVANCED,
                    "expert": UserExperience.EXPERT
                }
                self.context.user_experience = exp_map.get(
                    detected["experience"], UserExperience.INTERMEDIATE
                )
            
            if "formality" in detected:
                form_map = {
                    "casual": Formality.CASUAL,
                    "neutral": Formality.NEUTRAL,
                    "formal": Formality.FORMAL,
                    "very_formal": Formality.VERY_FORMAL
                }
                self.context.formality = form_map.get(
                    detected["formality"], Formality.NEUTRAL
                )
            
            if "urgency" in detected:
                urg_map = {
                    "low": TaskUrgency.LOW,
                    "medium": TaskUrgency.MEDIUM,
                    "high": TaskUrgency.HIGH,
                    "critical": TaskUrgency.CRITICAL
                }
                self.context.task_urgency = urg_map.get(
                    detected["urgency"], TaskUrgency.MEDIUM
                )
            
            if "task_type" in detected:
                self.context.task_type = detected["task_type"]
            
            if "complexity" in detected:
                self.context.task_complexity = detected["complexity"]
            
            return detected
        except Exception as e:
            print(f"Context detection error: {e}")
            return {}
    
    def update_context(self, updates: Dict[str, Any]):
        """Manually update context"""
        for key, value in updates.items():
            if hasattr(self.context, key):
                setattr(self.context, key, value)
    
    def respond(self, query: str, auto_detect: bool = True) -> Dict[str, Any]:
        """Generate context-adapted response"""
        print(f"\n{'='*60}")
        print(f"Query: {query}")
        print(f"{'='*60}")
        
        # Auto-detect context if enabled
        if auto_detect:
            detected = self.detect_context(query)
            print(f"\nDetected context: {detected}")
        
        # Show current context
        print(self.context.summarize())
        
        # Generate adapted response
        response = self.response_chain.invoke({
            "context_summary": self.context.summarize(),
            "query": query
        })
        
        print(f"\nAdapted Response:")
        print(response)
        
        return {
            "query": query,
            "context": self.context.to_dict(),
            "response": response,
            "timestamp": datetime.now().isoformat()
        }
    
    def demonstrate_adaptation(self, query: str, contexts: List[Dict[str, Any]]):
        """Demonstrate how response changes with different contexts"""
        print(f"\n{'='*70}")
        print(f"Demonstrating Context Adaptation")
        print(f"Query: {query}")
        print(f"{'='*70}")
        
        responses = []
        for i, context_updates in enumerate(contexts, 1):
            print(f"\n--- Context {i}: {context_updates.get('description', '')} ---")
            
            # Reset context
            self.context = Context()
            
            # Apply updates
            for key, value in context_updates.items():
                if key != "description" and hasattr(self.context, key):
                    setattr(self.context, key, value)
            
            # Generate response
            result = self.respond(query, auto_detect=False)
            responses.append(result)
        
        return responses


def demonstrate_contextual_adaptation():
    """Demonstrate contextual adaptation pattern"""
    print("\n" + "="*70)
    print("CONTEXTUAL ADAPTATION DEMONSTRATION")
    print("="*70)
    
    agent = ContextualAdaptationAgent()
    
    # Example 1: Auto-detection of context
    print("\n" + "="*70)
    print("Example 1: Auto-Detection of Context")
    print("="*70)
    
    query1 = "I'm new to Python. Can you explain what a variable is?"
    result1 = agent.respond(query1, auto_detect=True)
    
    query2 = "Explain the time complexity of quicksort's partitioning scheme"
    result2 = agent.respond(query2, auto_detect=True)
    
    # Example 2: Adaptation to different user levels
    print("\n" + "="*70)
    print("Example 2: Adapting to User Experience Level")
    print("="*70)
    
    query = "How does machine learning work?"
    
    contexts = [
        {
            "description": "Beginner",
            "user_experience": UserExperience.BEGINNER,
            "formality": Formality.CASUAL
        },
        {
            "description": "Expert",
            "user_experience": UserExperience.EXPERT,
            "formality": Formality.NEUTRAL
        }
    ]
    
    responses = agent.demonstrate_adaptation(query, contexts)
    
    # Example 3: Adaptation to urgency
    print("\n" + "="*70)
    print("Example 3: Adapting to Task Urgency")
    print("="*70)
    
    query_urgent = "My server is down! How do I restart it?"
    
    contexts_urgent = [
        {
            "description": "Critical Urgency",
            "task_urgency": TaskUrgency.CRITICAL,
            "task_type": "troubleshooting"
        },
        {
            "description": "Low Urgency",
            "task_urgency": TaskUrgency.LOW,
            "task_type": "learning"
        }
    ]
    
    responses_urgent = agent.demonstrate_adaptation(query_urgent, contexts_urgent)
    
    # Example 4: Adaptation to formality
    print("\n" + "="*70)
    print("Example 4: Adapting to Formality Level")
    print("="*70)
    
    query_formal = "What are your capabilities?"
    
    contexts_formal = [
        {
            "description": "Casual",
            "formality": Formality.CASUAL,
            "relationship": "friendly"
        },
        {
            "description": "Very Formal",
            "formality": Formality.VERY_FORMAL,
            "relationship": "professional"
        }
    ]
    
    responses_formal = agent.demonstrate_adaptation(query_formal, contexts_formal)
    
    # Summary
    print("\n" + "="*70)
    print("SUMMARY")
    print("="*70)
    print("""
Contextual Adaptation Pattern demonstrated:

Key Features:
1. Context Detection - Automatically detect context from input
2. Multi-Dimensional Adaptation - Adapt across multiple context dimensions
3. Dynamic Behavior - Adjust communication style and strategy
4. User Personalization - Adapt to user experience and preferences
5. Situational Awareness - Respond appropriately to urgency and formality

Context Dimensions Handled:
- User experience level (beginner to expert)
- Communication formality (casual to very formal)
- Task urgency (low to critical)
- Task complexity and type
- Environment and technical constraints
- Domain-specific conventions

Applications:
- Personalized assistants adapting to user expertise
- Customer service bots adjusting tone
- Educational systems matching student level
- Multi-domain chatbots switching contexts
- International applications respecting cultural norms

Contextual adaptation makes agents more natural, effective, and user-friendly
by tailoring their behavior to the specific situation and user needs.
    """)


if __name__ == "__main__":
    demonstrate_contextual_adaptation()

    

Agentic Design Pattern: Contextual Adaptation

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

Overview

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

---

Pattern Description

Pattern 113: Contextual Adaptation

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

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

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

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

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

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

---

File Statistics

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

---

Module Documentation

"""
Pattern 113: Contextual Adaptation

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

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

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

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

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

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

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

---

Import Statements

Standard Library Imports

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

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

LangChain Imports

from langchain.memory.ConversationBufferMemory
from langchain_openai.ChatOpenAI

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

Other Imports

from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: UserExperience

Lines 56-61

Inherits from: Enum

Purpose:

User experience levels

---

Class: Formality

Lines 64-69

Inherits from: Enum

Purpose:

Communication formality levels

---

Class: TaskUrgency

Lines 72-77

Inherits from: Enum

Purpose:

Task urgency levels

---

Class: Context

Lines 80-153

Purpose:

Represents current context

Methods:

__init__(self)

Lines 83-109

Purpose: Initialize the class instance with required parameters.

Parameters:

to_dict(self)

Lines 111-141

Purpose:

Convert context to dictionary

Parameters:

Returns: <ast.Subscript object at 0x000001A06D798F90>

summarize(self)

Lines 143-153

Purpose:

Generate human-readable context summary

Parameters:

Returns: str

---

Class: ContextualAdaptationAgent

Lines 156-308

Purpose:

Agent that adapts behavior based on context

Methods:

__init__(self, model_name)

Lines 159-197

Purpose: Initialize the class instance with required parameters.

Parameters: – model_name: Language model instance

detect_context(self, query)

Lines 199-247

Purpose:

Detect context from user query

Parameters: – query: Input query or question

Returns: <ast.Subscript object at 0x000001A06D125F50>

update_context(self, updates)

Lines 249-253

Purpose:

Manually update context

Parameters: – updates: Updates

respond(self, query, auto_detect)

Lines 255-283

Purpose:

Generate context-adapted response

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

Returns: <ast.Subscript object at 0x000001A06D7FF950>

demonstrate_adaptation(self, query, contexts)

Lines 285-308

Purpose:

Demonstrate how response changes with different contexts

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

---

Functions

demonstrate_contextual_adaptation()

Lines 311-427

Purpose:

Demonstrate contextual adaptation pattern

---

Key Concepts Summary

Pattern Overview

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

Core Components

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

When to Use

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

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

---

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

---

LangChain Components Used

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

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

---

Execution Flow

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

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

---

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

---

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

---

Related Patterns

• Pattern 112
• Pattern 114

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The Contextual Adaptation pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

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

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

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

---

File: 113_contextual_adaptation.py
Total Lines: 432
Pattern Category: AI Agent Design Pattern #113

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 112: Common Sense Reasoning

Description:
    Common Sense Reasoning enables agents to apply implicit, everyday knowledge
    that humans take for granted. This pattern integrates common sense knowledge
    bases (like ConceptNet) with LLM reasoning to make inferences about the
    physical world, social norms, causality, and typical behaviors.
    
    The pattern helps agents understand context, make reasonable assumptions,
    avoid absurd conclusions, and reason about unstated but obvious facts.
    It bridges the gap between explicit knowledge and implicit understanding.

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

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

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

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

import os
from typing import List, Dict, Any, Optional, Tuple
from dotenv import load_dotenv
from datetime import datetime

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain.chains import LLMChain

load_dotenv()


class CommonSenseKnowledgeBase:
    """Simulated common sense knowledge base (in practice, use ConceptNet API)"""
    
    def __init__(self):
        self.knowledge = {
            # Physical laws
            "gravity": "Objects fall downward when dropped",
            "water_wet": "Water makes things wet",
            "fire_hot": "Fire is hot and can burn",
            "ice_cold": "Ice is cold and can freeze",
            "glass_fragile": "Glass is fragile and can break",
            
            # Social norms
            "greeting_polite": "Greeting people is considered polite",
            "stealing_wrong": "Taking things without permission is wrong",
            "helping_good": "Helping others is generally good",
            "lying_negative": "Lying usually has negative consequences",
            
            # Causal relationships
            "rain_wet": "Rain makes outdoor surfaces wet",
            "eating_hunger": "Eating satisfies hunger",
            "sleep_rest": "Sleep provides rest and recovery",
            "exercise_fitness": "Exercise improves fitness",
            
            # Temporal facts
            "time_forward": "Time moves forward, not backward",
            "aging_continuous": "Living things age continuously",
            "seasons_cycle": "Seasons cycle in a predictable pattern",
            
            # Spatial facts
            "one_place": "Physical objects can't be in two places at once",
            "inside_outside": "Something inside a closed container is not outside",
            "solid_through": "Solid objects can't pass through solid walls",
            
            # Functional knowledge
            "key_lock": "Keys are used to open locks",
            "knife_cut": "Knives are used for cutting",
            "chair_sit": "Chairs are for sitting on",
            "umbrella_rain": "Umbrellas protect from rain",
        }
        
        self.categories = {
            "physical": ["gravity", "water_wet", "fire_hot", "ice_cold", "glass_fragile"],
            "social": ["greeting_polite", "stealing_wrong", "helping_good", "lying_negative"],
            "causal": ["rain_wet", "eating_hunger", "sleep_rest", "exercise_fitness"],
            "temporal": ["time_forward", "aging_continuous", "seasons_cycle"],
            "spatial": ["one_place", "inside_outside", "solid_through"],
            "functional": ["key_lock", "knife_cut", "chair_sit", "umbrella_rain"],
        }
    
    def query(self, concept: str) -> Optional[str]:
        """Query knowledge base for common sense fact"""
        return self.knowledge.get(concept)
    
    def get_category(self, category: str) -> List[str]:
        """Get all facts in a category"""
        concept_ids = self.categories.get(category, [])
        return [self.knowledge[cid] for cid in concept_ids if cid in self.knowledge]
    
    def get_relevant_facts(self, context: str) -> List[str]:
        """Get facts relevant to context"""
        # Simple keyword matching (in practice, use embeddings)
        relevant = []
        context_lower = context.lower()
        for key, fact in self.knowledge.items():
            if any(word in context_lower for word in key.split("_")):
                relevant.append(fact)
        return relevant


class CommonSenseReasoner:
    """Agent that applies common sense reasoning"""
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.kb = CommonSenseKnowledgeBase()
        
        # Reasoning with common sense
        self.reasoning_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a reasoning agent that applies common sense knowledge.
            
Given a question or situation, use the provided common sense facts to reason
about the answer. Make reasonable assumptions based on everyday knowledge.

Common sense facts:
{facts}

Question: {question}

Provide your reasoning and answer:"""),
            ("human", "{question}")
        ])
        
        # Plausibility checking
        self.plausibility_prompt = ChatPromptTemplate.from_messages([
            ("system", """Evaluate if a conclusion is plausible based on common sense.
            
Common sense facts:
{facts}

Proposed conclusion: {conclusion}

Rate plausibility from 0-10 and explain why:"""),
            ("human", "{conclusion}")
        ])
        
        # Assumption generation
        self.assumption_prompt = ChatPromptTemplate.from_messages([
            ("system", """Given a scenario, generate reasonable assumptions based on common sense.
            
Scenario: {scenario}

List 5 reasonable assumptions that most people would make:"""),
            ("human", "{scenario}")
        ])
        
        self.reasoning_chain = self.reasoning_prompt | self.llm | StrOutputParser()
        self.plausibility_chain = self.plausibility_prompt | self.llm | StrOutputParser()
        self.assumption_chain = self.assumption_prompt | self.llm | StrOutputParser()
    
    def reason_with_common_sense(self, question: str) -> Dict[str, Any]:
        """Apply common sense reasoning to answer a question"""
        print(f"\n{'='*60}")
        print(f"Question: {question}")
        print(f"{'='*60}")
        
        # Get relevant common sense facts
        relevant_facts = self.kb.get_relevant_facts(question)
        if not relevant_facts:
            relevant_facts = list(self.kb.knowledge.values())[:10]
        
        facts_str = "\n".join(f"- {fact}" for fact in relevant_facts)
        
        print(f"\nRelevant common sense facts:")
        for fact in relevant_facts:
            print(f"  - {fact}")
        
        # Reason with facts
        reasoning = self.reasoning_chain.invoke({
            "question": question,
            "facts": facts_str
        })
        
        print(f"\nReasoning:")
        print(reasoning)
        
        return {
            "question": question,
            "facts": relevant_facts,
            "reasoning": reasoning,
            "timestamp": datetime.now().isoformat()
        }
    
    def check_plausibility(self, conclusion: str, context: str = "") -> Dict[str, Any]:
        """Check if a conclusion is plausible"""
        print(f"\n{'='*60}")
        print(f"Checking plausibility: {conclusion}")
        print(f"{'='*60}")
        
        # Get relevant facts
        relevant_facts = self.kb.get_relevant_facts(conclusion + " " + context)
        if not relevant_facts:
            relevant_facts = list(self.kb.knowledge.values())[:10]
        
        facts_str = "\n".join(f"- {fact}" for fact in relevant_facts)
        
        # Check plausibility
        evaluation = self.plausibility_chain.invoke({
            "conclusion": conclusion,
            "facts": facts_str
        })
        
        print(f"\nEvaluation:")
        print(evaluation)
        
        return {
            "conclusion": conclusion,
            "facts": relevant_facts,
            "evaluation": evaluation,
            "timestamp": datetime.now().isoformat()
        }
    
    def generate_assumptions(self, scenario: str) -> Dict[str, Any]:
        """Generate reasonable assumptions for a scenario"""
        print(f"\n{'='*60}")
        print(f"Scenario: {scenario}")
        print(f"{'='*60}")
        
        # Generate assumptions
        assumptions = self.assumption_chain.invoke({
            "scenario": scenario
        })
        
        print(f"\nGenerated assumptions:")
        print(assumptions)
        
        return {
            "scenario": scenario,
            "assumptions": assumptions,
            "timestamp": datetime.now().isoformat()
        }
    
    def explain_with_common_sense(self, observation: str) -> Dict[str, Any]:
        """Explain an observation using common sense"""
        print(f"\n{'='*60}")
        print(f"Observation: {observation}")
        print(f"{'='*60}")
        
        # Get causal facts
        causal_facts = self.kb.get_category("causal")
        facts_str = "\n".join(f"- {fact}" for fact in causal_facts)
        
        explanation_prompt = f"""Using common sense causal knowledge, explain this observation:

Observation: {observation}

Common sense causal facts:
{facts_str}

Provide a simple, common sense explanation:"""
        
        explanation = self.llm.invoke(explanation_prompt).content
        
        print(f"\nExplanation:")
        print(explanation)
        
        return {
            "observation": observation,
            "causal_facts": causal_facts,
            "explanation": explanation,
            "timestamp": datetime.now().isoformat()
        }


def demonstrate_common_sense_reasoning():
    """Demonstrate common sense reasoning patterns"""
    print("\n" + "="*70)
    print("COMMON SENSE REASONING DEMONSTRATION")
    print("="*70)
    
    reasoner = CommonSenseReasoner()
    
    # Example 1: Question answering with common sense
    print("\n" + "="*70)
    print("Example 1: Question Answering with Common Sense")
    print("="*70)
    
    question1 = "If I drop a glass cup on the floor, what will happen?"
    result1 = reasoner.reason_with_common_sense(question1)
    
    question2 = "Should I use an umbrella when it's raining?"
    result2 = reasoner.reason_with_common_sense(question2)
    
    # Example 2: Plausibility checking
    print("\n" + "="*70)
    print("Example 2: Plausibility Checking")
    print("="*70)
    
    conclusion1 = "I can walk through a solid brick wall"
    plausibility1 = reasoner.check_plausibility(conclusion1)
    
    conclusion2 = "If I leave ice cream outside in the sun, it will melt"
    plausibility2 = reasoner.check_plausibility(conclusion2)
    
    # Example 3: Assumption generation
    print("\n" + "="*70)
    print("Example 3: Assumption Generation")
    print("="*70)
    
    scenario = "A person enters a restaurant and sits at a table"
    assumptions = reasoner.generate_assumptions(scenario)
    
    # Example 4: Explanation with common sense
    print("\n" + "="*70)
    print("Example 4: Explaining Observations")
    print("="*70)
    
    observation1 = "The sidewalk is wet even though no one watered it"
    explanation1 = reasoner.explain_with_common_sense(observation1)
    
    observation2 = "The person looks tired and is yawning"
    explanation2 = reasoner.explain_with_common_sense(observation2)
    
    # Summary
    print("\n" + "="*70)
    print("SUMMARY")
    print("="*70)
    print("""
Common Sense Reasoning Pattern demonstrated:

Key Features:
1. Knowledge Base Integration - Access to common sense facts
2. Contextual Reasoning - Apply relevant facts to situations
3. Plausibility Checking - Validate conclusions against common sense
4. Assumption Generation - Make reasonable implicit assumptions
5. Causal Explanation - Explain observations with common sense

Applications:
- Natural language understanding
- Question answering systems
- Story comprehension
- Situation interpretation
- Robotics and embodied AI

Common sense reasoning helps agents bridge the gap between explicit
knowledge and implicit understanding that humans naturally possess.
    """)


if __name__ == "__main__":
    demonstrate_common_sense_reasoning()

    

Agentic Design Pattern: Common Sense Reasoning

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

Overview

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

---

Pattern Description

Pattern 112: Common Sense Reasoning

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

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

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

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

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

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

---

File Statistics

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

---

Module Documentation

"""
Pattern 112: Common Sense Reasoning

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

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

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

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

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

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

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

---

Import Statements

Standard Library Imports

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

Purpose: Core Python functionality: – datetime.datetime: Date and time handling – langchain_core.output_parsers.JsonOutputParser: Regular expression operations – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Tuple: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain.chains.LLMChain
from langchain_openai.ChatOpenAI

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

Other Imports

from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: CommonSenseKnowledgeBase

Lines 54-121

Purpose:

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

Methods:

__init__(self)

Lines 57-102

Purpose: Initialize the class instance with required parameters.

Parameters:

query(self, concept)

Lines 104-106

Purpose:

Query knowledge base for common sense fact

Parameters: – concept: Concept

Returns: <ast.Subscript object at 0x000001A06D6156D0>

get_category(self, category)

Lines 108-111

Purpose:

Get all facts in a category

Parameters: – category: Category

Returns: <ast.Subscript object at 0x000001A06D6168D0>

get_relevant_facts(self, context)

Lines 113-121

Purpose:

Get facts relevant to context

Parameters: – context: Context

Returns: <ast.Subscript object at 0x000001A06D73A2D0>

---

Class: CommonSenseReasoner

Lines 124-285

Purpose:

Agent that applies common sense reasoning

Methods:

__init__(self, model_name)

Lines 127-172

Purpose: Initialize the class instance with required parameters.

Parameters: – model_name: Language model instance

reason_with_common_sense(self, question)

Lines 174-205

Purpose:

Apply common sense reasoning to answer a question

Parameters: – question: Question to answer

Returns: <ast.Subscript object at 0x000001A06D16E510>

check_plausibility(self, conclusion, context)

Lines 207-234

Purpose:

Check if a conclusion is plausible

Parameters: – conclusion: Conclusion – context: Context

Returns: <ast.Subscript object at 0x000001A06D824AD0>

generate_assumptions(self, scenario)

Lines 236-254

Purpose:

Generate reasonable assumptions for a scenario

Parameters: – scenario: Scenario

Returns: <ast.Subscript object at 0x000001A06D68C590>

explain_with_common_sense(self, observation)

Lines 256-285

Purpose:

Explain an observation using common sense

Parameters: – observation: Observation

Returns: <ast.Subscript object at 0x000001A06D888FD0>

---

Functions

demonstrate_common_sense_reasoning()

Lines 288-360

Purpose:

Demonstrate common sense reasoning patterns

---

Key Concepts Summary

Pattern Overview

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

Core Components

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

When to Use

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

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

---

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

---

LangChain Components Used

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

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

---

Execution Flow

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

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

---

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

---

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

---

Related Patterns

• Pattern 111
• Pattern 113

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The Common Sense Reasoning pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

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

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

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

---

File: 112_common_sense_reasoning.py
Total Lines: 365
Pattern Category: AI Agent Design Pattern #112

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 111: Situational Awareness

Description:
    Situational Awareness enables agents to maintain comprehensive understanding
    of their current context, environment, and state. The pattern continuously
    monitors and updates a model of the situation, including relevant entities,
    relationships, constraints, opportunities, and threats. This awareness informs
    decision-making and enables context-appropriate behavior.
    
    The pattern is crucial for adaptive agents operating in dynamic environments
    where context changes frequently. It combines perception, state estimation,
    context modeling, and predictive awareness to maintain an accurate and
    up-to-date understanding of the situation.
    
    Situational awareness includes environmental monitoring, state tracking,
    context interpretation, threat detection, opportunity identification, and
    predictive situation assessment.

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

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

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

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

import os
from typing import List, Dict, Any, Optional
from datetime import datetime
from dotenv import load_dotenv
from enum import Enum

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain.memory import ConversationBufferMemory

load_dotenv()


class SituationStatus(Enum):
    """Enumeration of situation statuses."""
    NORMAL = "normal"
    ALERT = "alert"
    WARNING = "warning"
    CRITICAL = "critical"
    OPPORTUNITY = "opportunity"


class SituationalAwarenessAgent:
    """
    Agent that maintains comprehensive awareness of current situation.
    """
    
    def __init__(self, agent_role: str = "general_agent"):
        """Initialize with role and context."""
        self.llm = ChatOpenAI(temperature=0.3, model="gpt-4")
        self.role = agent_role
        self.situation_memory = ConversationBufferMemory(
            memory_key="situation_history",
            return_messages=True
        )
        
        # Situational state
        self.current_context = {}
        self.active_entities = {}
        self.threats = []
        self.opportunities = []
        self.status = SituationStatus.NORMAL
    
    def perceive_environment(
        self,
        observations: Dict[str, Any]
    ) -> Dict[str, Any]:
        """
        Process environmental observations to update awareness.
        
        Args:
            observations: Dictionary of observed facts
            
        Returns:
            Updated situational model
        """
        print(f"\n{'='*60}")
        print("PERCEPTION & STATE ESTIMATION")
        print(f"{'='*60}")
        print(f"Processing {len(observations)} observations...")
        
        # Update context with observations
        self.current_context.update(observations)
        self.current_context['last_update'] = datetime.now().isoformat()
        
        # Analyze observations for entities
        prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a situational awareness system analyzing observations.
Extract key entities, their states, and relationships from the observations.

Provide a JSON response with:
- entities: List of identified entities with properties
- relationships: List of relationships between entities
- key_facts: Important facts about the situation"""),
            ("user", "Observations: {observations}")
        ])
        
        chain = prompt | self.llm | JsonOutputParser()
        
        try:
            analysis = chain.invoke({"observations": str(observations)})
            
            # Update entities
            if 'entities' in analysis:
                for entity in analysis['entities']:
                    entity_id = entity.get('id', entity.get('name', 'unknown'))
                    self.active_entities[entity_id] = entity
            
            print(f"Identified {len(self.active_entities)} entities")
            print(f"Key facts: {analysis.get('key_facts', [])}")
            
            return {
                "perception": analysis,
                "context_updated": True,
                "timestamp": self.current_context['last_update']
            }
        except:
            # Fallback if JSON parsing fails
            return {
                "perception": observations,
                "context_updated": True,
                "timestamp": self.current_context['last_update']
            }
    
    def assess_situation(self) -> Dict[str, Any]:
        """
        Assess current situation for threats and opportunities.
        
        Returns:
            Situation assessment with status and recommendations
        """
        print(f"\n{'='*60}")
        print("SITUATION ASSESSMENT")
        print(f"{'='*60}")
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a situational assessment expert. Analyze the current
situation and provide:
1. Overall situation status (normal/alert/warning/critical/opportunity)
2. Identified threats or risks
3. Identified opportunities
4. Recommended actions
5. Confidence in assessment (0-1)

Format as: STATUS | Threats: [...] | Opportunities: [...] | Actions: [...] | Confidence: X.X"""),
            ("user", """Current Context: {context}
Active Entities: {entities}
Situation History: {history}

Assess this situation comprehensively.""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        
        assessment_text = chain.invoke({
            "context": str(self.current_context),
            "entities": str(self.active_entities),
            "history": self.situation_memory.load_memory_variables({}).get('situation_history', [])
        })
        
        # Parse assessment
        parts = assessment_text.split('|')
        status_text = parts[0].strip().lower()
        
        # Update status
        if 'critical' in status_text:
            self.status = SituationStatus.CRITICAL
        elif 'warning' in status_text:
            self.status = SituationStatus.WARNING
        elif 'alert' in status_text:
            self.status = SituationStatus.ALERT
        elif 'opportunity' in status_text:
            self.status = SituationStatus.OPPORTUNITY
        else:
            self.status = SituationStatus.NORMAL
        
        result = {
            "status": self.status.value,
            "assessment": assessment_text,
            "context_size": len(self.current_context),
            "entities_tracked": len(self.active_entities),
            "timestamp": datetime.now().isoformat()
        }
        
        print(f"Status: {self.status.value.upper()}")
        print(f"Assessment:\n{assessment_text}")
        
        # Store in memory
        self.situation_memory.save_context(
            {"input": "situation_assessment"},
            {"output": assessment_text}
        )
        
        return result
    
    def predict_changes(
        self,
        time_horizon: str = "near future"
    ) -> Dict[str, Any]:
        """
        Predict likely changes to the situation.
        
        Args:
            time_horizon: Time frame for prediction
            
        Returns:
            Predictions about situation evolution
        """
        print(f"\n{'='*60}")
        print("PREDICTIVE AWARENESS")
        print(f"{'='*60}")
        print(f"Predicting changes for: {time_horizon}")
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a predictive analyst. Based on current situation,
predict likely changes and developments.

Provide:
1. Most likely changes
2. Probability estimates
3. Potential triggers
4. Recommended preparations"""),
            ("user", """Current Situation:
Context: {context}
Entities: {entities}
Current Status: {status}

Predict changes for: {time_horizon}""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        
        predictions = chain.invoke({
            "context": str(self.current_context),
            "entities": str(self.active_entities),
            "status": self.status.value,
            "time_horizon": time_horizon
        })
        
        result = {
            "time_horizon": time_horizon,
            "predictions": predictions,
            "based_on_entities": len(self.active_entities),
            "confidence": "medium"  # Could be calculated
        }
        
        print(f"Predictions:\n{predictions}")
        
        return result
    
    def context_appropriate_response(
        self,
        query: str
    ) -> Dict[str, Any]:
        """
        Generate response appropriate to current situational context.
        
        Args:
            query: Query or request
            
        Returns:
            Context-aware response
        """
        print(f"\n{'='*60}")
        print("CONTEXT-AWARE RESPONSE")
        print(f"{'='*60}")
        print(f"Query: {query}")
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", """You are an agent with full situational awareness. 
Respond to queries considering the complete situational context.

Your response should:
1. Acknowledge relevant context
2. Address the query appropriately
3. Consider current status and constraints
4. Provide context-aware recommendations"""),
            ("user", """Query: {query}

Current Situation:
- Status: {status}
- Context: {context}
- Active Entities: {entities}

Provide a situationally appropriate response.""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        
        response = chain.invoke({
            "query": query,
            "status": self.status.value,
            "context": str(self.current_context)[:500],  # Limit size
            "entities": str(list(self.active_entities.keys()))
        })
        
        result = {
            "query": query,
            "response": response,
            "context_considered": True,
            "status_at_response": self.status.value
        }
        
        print(f"Response:\n{response}")
        
        return result
    
    def get_situation_summary(self) -> Dict[str, Any]:
        """Get comprehensive summary of current situation."""
        return {
            "status": self.status.value,
            "context": self.current_context,
            "entities": self.active_entities,
            "role": self.role,
            "last_update": self.current_context.get('last_update', 'never')
        }


def demonstrate_situational_awareness():
    """Demonstrate situational awareness capabilities."""
    print("=" * 60)
    print("SITUATIONAL AWARENESS AGENT DEMONSTRATION")
    print("=" * 60)
    
    agent = SituationalAwarenessAgent(agent_role="security_monitor")
    
    # Example 1: Initial Perception
    print("\n" + "=" * 60)
    print("Example 1: Environmental Perception")
    print("=" * 60)
    
    observations1 = {
        "time": "14:30",
        "location": "Building A, Floor 3",
        "people_count": 15,
        "noise_level": "normal",
        "access_points": ["door_301", "door_302"],
        "cameras_active": True
    }
    
    result1 = agent.perceive_environment(observations1)
    
    # Example 2: Situation Assessment
    print("\n" + "=" * 60)
    print("Example 2: Situation Assessment")
    print("=" * 60)
    
    result2 = agent.assess_situation()
    
    # Example 3: Changed Situation
    print("\n" + "=" * 60)
    print("Example 3: Detecting Changes")
    print("=" * 60)
    
    observations2 = {
        "time": "14:35",
        "people_count": 25,  # Increased
        "noise_level": "elevated",
        "unauthorized_access_attempt": "door_304",
        "alarm_triggered": True
    }
    
    result3 = agent.perceive_environment(observations2)
    result4 = agent.assess_situation()
    
    # Example 4: Predictive Awareness
    print("\n" + "=" * 60)
    print("Example 4: Predicting Changes")
    print("=" * 60)
    
    result5 = agent.predict_changes("next 15 minutes")
    
    # Example 5: Context-Aware Response
    print("\n" + "=" * 60)
    print("Example 5: Context-Appropriate Response")
    print("=" * 60)
    
    result6 = agent.context_appropriate_response(
        "Should I allow the scheduled maintenance crew to enter?"
    )
    
    # Summary
    print("\n" + "=" * 60)
    print("SITUATION SUMMARY")
    print("=" * 60)
    summary = agent.get_situation_summary()
    print(f"Current Status: {summary['status'].upper()}")
    print(f"Entities Tracked: {len(summary['entities'])}")
    print(f"Context Elements: {len(summary['context'])}")
    print(f"Agent Role: {summary['role']}")
    
    print("\n" + "=" * 60)
    print("DEMONSTRATION COMPLETE")
    print("=" * 60)
    print("✓ Processed environmental observations")
    print("✓ Maintained situational context")
    print("✓ Detected and assessed changes")
    print("✓ Predicted future developments")
    print("✓ Generated context-aware responses")


if __name__ == "__main__":
    demonstrate_situational_awareness()

    

Agentic Design Pattern: Situational Awareness

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

Overview

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

---

Pattern Description

Pattern 111: Situational Awareness

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

The pattern is crucial for adaptive agents operating in dynamic environments
where context changes frequently. It combines perception, state estimation,
context modeling, and predictive awareness to maintain an accurate and
up-to-date understanding of the situation.

Situational awareness includes environmental monitoring, state tracking,
context interpretation, threat detection, opportunity identification, and
predictive situation assessment.

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

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

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

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

---

File Statistics

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

---

Module Documentation

"""
Pattern 111: Situational Awareness

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

    The pattern is crucial for adaptive agents operating in dynamic environments
    where context changes frequently. It combines perception, state estimation,
    context modeling, and predictive awareness to maintain an accurate and
    up-to-date understanding of the situation.

    Situational awareness includes environmental monitoring, state tracking,
    context interpretation, threat detection, opportunity identification, and
    predictive situation assessment.

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

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

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

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

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

---

Import Statements

Standard Library Imports

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

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

LangChain Imports

from langchain.memory.ConversationBufferMemory
from langchain_openai.ChatOpenAI

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

Other Imports

from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: SituationStatus

Lines 63-69

Inherits from: Enum

Purpose:

Enumeration of situation statuses.

---

Class: SituationalAwarenessAgent

Lines 72-345

Purpose:

Agent that maintains comprehensive awareness of current situation.

Methods:

__init__(self, agent_role)

Lines 77-91

Purpose:

Initialize with role and context.

Parameters: – agent_role: Agent role

perceive_environment(self, observations)

Lines 93-152

Purpose:

Process environmental observations to update awareness.

Args:
    observations: Dictionary of observed facts

Returns:
    Updated situational model

Parameters: – observations: Observations

Returns: <ast.Subscript object at 0x000001A06D7FDFD0>

assess_situation(self)

Lines 154-223

Purpose:

Assess current situation for threats and opportunities.

Returns:
    Situation assessment with status and recommendations

Parameters:

Returns: <ast.Subscript object at 0x000001A06D72EE50>

predict_changes(self, time_horizon)

Lines 225-278

Purpose:

Predict likely changes to the situation.

Args:
    time_horizon: Time frame for prediction

Returns:
    Predictions about situation evolution

Parameters: – time_horizon: Time horizon

Returns: <ast.Subscript object at 0x000001A06D604250>

context_appropriate_response(self, query)

Lines 280-335

Purpose:

Generate response appropriate to current situational context.

Args:
    query: Query or request

Returns:
    Context-aware response

Parameters: – query: Input query or question

Returns: <ast.Subscript object at 0x000001A06D17C810>

get_situation_summary(self)

Lines 337-345

Purpose:

Get comprehensive summary of current situation.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D6F5AD0>

---

Functions

demonstrate_situational_awareness()

Lines 348-428

Purpose:

Demonstrate situational awareness capabilities.

---

Key Concepts Summary

Pattern Overview

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

Core Components

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

When to Use

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

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

---

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

---

LangChain Components Used

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

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

---

Execution Flow

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

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

---

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

---

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

---

Related Patterns

• Pattern 110
• Pattern 112

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The Situational Awareness pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

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

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

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

---

File: 111_situational_awareness.py
Total Lines: 433
Pattern Category: AI Agent Design Pattern #111

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 110: Multi-Modal Grounding

Description:
    Multi-Modal Grounding enables agents to understand and reason across multiple
    modalities simultaneously (text, images, audio, video, etc.). The pattern grounds
    abstract concepts and language in concrete perceptual experiences across different
    sensory modalities, enabling richer understanding and more contextually appropriate
    responses.
    
    This pattern is essential for embodied AI, robotics, and applications requiring
    comprehensive understanding of the real world. It integrates information from
    different modalities, resolves cross-modal references, and maintains coherent
    multimodal representations.
    
    The pattern includes vision-language integration, audio processing, cross-modal
    attention mechanisms, unified embeddings, and multimodal fusion strategies.

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

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

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

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

import os
from typing import List, Dict, Any, Optional, Union
from dotenv import load_dotenv
import base64
from pathlib import Path

from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

load_dotenv()


class MultiModalGroundingAgent:
    """
    Agent that processes and reasons across multiple modalities.
    """
    
    def __init__(self, model_name: str = "gpt-4o"):
        """Initialize with multimodal LLM."""
        # Use multimodal model (GPT-4 with vision)
        self.llm = ChatOpenAI(model=model_name, temperature=0.7, max_tokens=1000)
        self.parser = StrOutputParser()
    
    def encode_image(self, image_path: str) -> str:
        """Encode image to base64."""
        with open(image_path, "rb") as image_file:
            return base64.b64encode(image_file.read()).decode('utf-8')
    
    def process_image_text(
        self,
        text_query: str,
        image_path: Optional[str] = None,
        image_url: Optional[str] = None
    ) -> Dict[str, Any]:
        """
        Process text query grounded in visual context.
        
        Args:
            text_query: Text question or instruction
            image_path: Local path to image file
            image_url: URL to image
            
        Returns:
            Dictionary with response and reasoning
        """
        print(f"\n{'='*60}")
        print("VISION-LANGUAGE GROUNDING")
        print(f"{'='*60}")
        print(f"Query: {text_query}")
        
        # Construct message with image
        content = []
        content.append({"type": "text", "text": text_query})
        
        if image_url:
            content.append({"type": "image_url", "image_url": {"url": image_url}})
            print(f"Image: {image_url}")
        elif image_path:
            # For local images, would need base64 encoding
            # Simplified for demonstration
            print(f"Image: {image_path}")
            # In real implementation: encode and add to content
        
        message = HumanMessage(content=content if image_url or image_path else text_query)
        
        # Process with multimodal LLM
        response = self.llm.invoke([message])
        
        result = {
            "query": text_query,
            "response": response.content,
            "modalities": ["text", "vision"] if (image_url or image_path) else ["text"],
            "grounded": bool(image_url or image_path)
        }
        
        print(f"\nResponse: {result['response']}")
        print(f"Modalities Used: {result['modalities']}")
        
        return result
    
    def compare_images(
        self,
        image_url_1: str,
        image_url_2: str,
        comparison_aspect: str = "general"
    ) -> Dict[str, Any]:
        """
        Compare two images and identify differences/similarities.
        
        Args:
            image_url_1: First image URL
            image_url_2: Second image URL
            comparison_aspect: What to compare (general, specific aspect)
            
        Returns:
            Comparison results
        """
        print(f"\n{'='*60}")
        print("CROSS-MODAL COMPARISON")
        print(f"{'='*60}")
        
        prompt = f"""Compare these two images focusing on {comparison_aspect}.
        
Provide:
1. Key similarities
2. Key differences
3. Notable features in each
4. Overall assessment"""
        
        content = [
            {"type": "text", "text": prompt},
            {"type": "image_url", "image_url": {"url": image_url_1}},
            {"type": "image_url", "image_url": {"url": image_url_2}}
        ]
        
        message = HumanMessage(content=content)
        response = self.llm.invoke([message])
        
        result = {
            "aspect": comparison_aspect,
            "comparison": response.content,
            "images_compared": 2
        }
        
        print(f"Comparison Result:\n{result['comparison']}")
        
        return result
    
    def multimodal_reasoning(
        self,
        text_context: str,
        image_url: Optional[str] = None,
        structured_data: Optional[Dict] = None,
        reasoning_task: str = "analyze and synthesize"
    ) -> Dict[str, Any]:
        """
        Perform reasoning across multiple modalities.
        
        Args:
            text_context: Textual information
            image_url: Visual information
            structured_data: Structured data (tables, graphs)
            reasoning_task: Type of reasoning required
            
        Returns:
            Reasoning results integrating all modalities
        """
        print(f"\n{'='*60}")
        print("MULTIMODAL REASONING")
        print(f"{'='*60}")
        print(f"Task: {reasoning_task}")
        
        # Build multimodal context
        modalities_used = ["text"]
        content = [{"type": "text", "text": f"Context: {text_context}\n\nTask: {reasoning_task}"}]
        
        if structured_data:
            modalities_used.append("structured_data")
            data_str = "\n".join([f"{k}: {v}" for k, v in structured_data.items()])
            content[0]["text"] += f"\n\nStructured Data:\n{data_str}"
        
        if image_url:
            modalities_used.append("vision")
            content.append({"type": "image_url", "image_url": {"url": image_url}})
        
        content[0]["text"] += "\n\nProvide comprehensive analysis integrating all available information."
        
        message = HumanMessage(content=content)
        response = self.llm.invoke([message])
        
        result = {
            "task": reasoning_task,
            "modalities_used": modalities_used,
            "reasoning": response.content,
            "integrated": len(modalities_used) > 1
        }
        
        print(f"Modalities: {modalities_used}")
        print(f"Reasoning:\n{result['reasoning']}")
        
        return result
    
    def spatial_grounding(
        self,
        image_url: str,
        spatial_query: str
    ) -> Dict[str, Any]:
        """
        Ground spatial language in visual context.
        
        Args:
            image_url: Image to analyze
            spatial_query: Query about spatial relationships
            
        Returns:
            Spatial grounding results
        """
        print(f"\n{'='*60}")
        print("SPATIAL GROUNDING")
        print(f"{'='*60}")
        
        prompt = f"""Analyze the spatial relationships in this image.

Query: {spatial_query}

Provide:
1. Identified objects and their locations
2. Spatial relationships between objects
3. Answer to the spatial query
4. Confidence in your spatial understanding"""
        
        content = [
            {"type": "text", "text": prompt},
            {"type": "image_url", "image_url": {"url": image_url}}
        ]
        
        message = HumanMessage(content=content)
        response = self.llm.invoke([message])
        
        result = {
            "query": spatial_query,
            "spatial_analysis": response.content,
            "grounding_type": "spatial"
        }
        
        print(f"Query: {spatial_query}")
        print(f"Analysis:\n{result['spatial_analysis']}")
        
        return result


def demonstrate_multimodal_grounding():
    """Demonstrate multi-modal grounding capabilities."""
    print("=" * 60)
    print("MULTI-MODAL GROUNDING AGENT DEMONSTRATION")
    print("=" * 60)
    
    agent = MultiModalGroundingAgent()
    
    # Example 1: Vision-Language Query
    print("\n" + "=" * 60)
    print("Example 1: Vision-Language Understanding")
    print("=" * 60)
    
    # Using a sample image URL (placeholder)
    sample_image_url = "https://upload.wikimedia.org/wikipedia/commons/thumb/d/dd/Gfp-wisconsin-madison-the-nature-boardwalk.jpg/2560px-Gfp-wisconsin-madison-the-nature-boardwalk.jpg"
    
    result1 = agent.process_image_text(
        text_query="What is in this image? Describe the scene in detail.",
        image_url=sample_image_url
    )
    
    # Example 2: Multimodal Reasoning
    print("\n" + "=" * 60)
    print("Example 2: Multimodal Reasoning Integration")
    print("=" * 60)
    
    result2 = agent.multimodal_reasoning(
        text_context="This is a natural landscape photograph taken during daylight hours.",
        image_url=sample_image_url,
        structured_data={
            "location": "Wisconsin",
            "type": "Nature boardwalk",
            "season": "Summer"
        },
        reasoning_task="Analyze the environmental conditions and recommend best time to visit"
    )
    
    # Example 3: Spatial Grounding
    print("\n" + "=" * 60)
    print("Example 3: Spatial Grounding")
    print("=" * 60)
    
    result3 = agent.spatial_grounding(
        image_url=sample_image_url,
        spatial_query="What objects are in the foreground vs background? Describe the depth perspective."
    )
    
    # Summary
    print("\n" + "=" * 60)
    print("SUMMARY")
    print("=" * 60)
    print(f"✓ Processed {3} multimodal examples")
    print("✓ Demonstrated vision-language understanding")
    print("✓ Integrated structured data with visual information")
    print("✓ Performed spatial grounding and reasoning")
    print("\nKey Capabilities:")
    print("  - Cross-modal attention and integration")
    print("  - Unified multimodal representations")
    print("  - Spatial and visual grounding")
    print("  - Context-aware multimodal reasoning")


if __name__ == "__main__":
    demonstrate_multimodal_grounding()

    

Agentic Design Pattern: Multi Modal Grounding

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

Overview

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

---

Pattern Description

Pattern 110: Multi-Modal Grounding

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

This pattern is essential for embodied AI, robotics, and applications requiring
comprehensive understanding of the real world. It integrates information from
different modalities, resolves cross-modal references, and maintains coherent
multimodal representations.

The pattern includes vision-language integration, audio processing, cross-modal
attention mechanisms, unified embeddings, and multimodal fusion strategies.

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

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

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

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

---

File Statistics

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

---

Module Documentation

"""
Pattern 110: Multi-Modal Grounding

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

    This pattern is essential for embodied AI, robotics, and applications requiring
    comprehensive understanding of the real world. It integrates information from
    different modalities, resolves cross-modal references, and maintains coherent
    multimodal representations.

    The pattern includes vision-language integration, audio processing, cross-modal
    attention mechanisms, unified embeddings, and multimodal fusion strategies.

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

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

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

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

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

---

Import Statements

Standard Library Imports

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

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

LangChain Imports

from langchain_openai.ChatOpenAI

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

Other Imports

from base64
from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: MultiModalGroundingAgent

Lines 60-276

Purpose:

Agent that processes and reasons across multiple modalities.

Methods:

__init__(self, model_name)

Lines 65-69

Purpose:

Initialize with multimodal LLM.

Parameters: – model_name: Language model instance

encode_image(self, image_path)

Lines 71-74

Purpose:

Encode image to base64.

Parameters: – image_path: Image path

Returns: str

process_image_text(self, text_query, image_path, image_url)

Lines 76-126

Purpose:

Process text query grounded in visual context.

Args:
    text_query: Text question or instruction
    image_path: Local path to image file
    image_url: URL to image

Returns:
    Dictionary with response and reasoning

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

Returns: <ast.Subscript object at 0x000001A06D88BF50>

compare_images(self, image_url_1, image_url_2, comparison_aspect)

Lines 128-174

Purpose:

Compare two images and identify differences/similarities.

Args:
    image_url_1: First image URL
    image_url_2: Second image URL
    comparison_aspect: What to compare (general, specific aspect)

Returns:
    Comparison results

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

Returns: <ast.Subscript object at 0x000001A06D71B3D0>

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

Lines 176-228

Purpose:

Perform reasoning across multiple modalities.

Args:
    text_context: Textual information
    image_url: Visual information
    structured_data: Structured data (tables, graphs)
    reasoning_task: Type of reasoning required

Returns:
    Reasoning results integrating all modalities

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

Returns: <ast.Subscript object at 0x000001A06D824250>

spatial_grounding(self, image_url, spatial_query)

Lines 230-276

Purpose:

Ground spatial language in visual context.

Args:
    image_url: Image to analyze
    spatial_query: Query about spatial relationships

Returns:
    Spatial grounding results

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

Returns: <ast.Subscript object at 0x000001A06D16C350>

---

Functions

demonstrate_multimodal_grounding()

Lines 279-338

Purpose:

Demonstrate multi-modal grounding capabilities.

---

Key Concepts Summary

Pattern Overview

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

Core Components

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

When to Use

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

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

---

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

---

LangChain Components Used

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

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

---

Execution Flow

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

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

---

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

---

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

---

Related Patterns

• Pattern 109
• Pattern 111

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The Multi Modal Grounding pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

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

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

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

---

File: 110_multi_modal_grounding.py
Total Lines: 343
Pattern Category: AI Agent Design Pattern #110

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 109: Adaptive Planning

Description:
    The Adaptive Planning pattern enables agents to dynamically adjust their plans
    in response to execution feedback, environmental changes, and unexpected events.
    Rather than rigidly following a predetermined plan, adaptive planners monitor
    execution, detect deviations from expected outcomes, diagnose problems, and
    modify plans on the fly to maintain progress toward goals.
    
    Adaptive planning is essential in dynamic, uncertain environments where initial
    assumptions may become invalid, resources may change, or unexpected opportunities
    and obstacles may arise. The pattern combines continuous monitoring, deviation
    detection, root cause analysis, and replanning strategies to ensure resilience
    and goal achievement despite changing conditions.
    
    This pattern includes mechanisms for execution monitoring, deviation detection,
    impact assessment, root cause diagnosis, plan repair strategies, replanning
    triggers, and learning from outcomes to improve future plans.

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

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

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

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

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

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

LangChain Implementation:
    This implementation uses LangChain for:
    - LLM-based deviation analysis
    - Root cause diagnosis
    - Plan repair suggestions
    - Natural language explanations
    
Production Considerations:
    - Implement efficient monitoring
    - Use event-driven detection
    - Cache replan results
    - Set replanning thresholds
    - Limit replanning frequency
    - Preserve partial progress
    - Log all adaptations
    - Enable replay for debugging
"""

import os
from typing import List, Dict, Any, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from dotenv import load_dotenv

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

load_dotenv()


class StepStatus(Enum):
    """Status of a plan step."""
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"
    SKIPPED = "skipped"


class DeviationType(Enum):
    """Types of plan deviations."""
    DELAY = "delay"
    RESOURCE_SHORTAGE = "resource_shortage"
    QUALITY_ISSUE = "quality_issue"
    COST_OVERRUN = "cost_overrun"
    DEPENDENCY_FAILURE = "dependency_failure"
    ENVIRONMENTAL_CHANGE = "environmental_change"


class AdaptationAction(Enum):
    """Types of adaptation actions."""
    CONTINUE = "continue"
    REPAIR = "repair"
    REVISE = "revise"
    REPLAN = "replan"
    ABORT = "abort"


@dataclass
class PlanStep:
    """
    A step in an adaptive plan.
    
    Attributes:
        step_id: Unique identifier
        description: Step description
        expected_duration: Expected time to complete
        expected_cost: Expected cost
        dependencies: Prerequisite steps
        status: Current status
        started_at: When started
        completed_at: When completed
        actual_duration: Actual time taken
        actual_cost: Actual cost
    """
    step_id: str
    description: str
    expected_duration: float = 1.0
    expected_cost: float = 1.0
    dependencies: List[str] = field(default_factory=list)
    status: StepStatus = StepStatus.PENDING
    started_at: Optional[datetime] = None
    completed_at: Optional[datetime] = None
    actual_duration: Optional[float] = None
    actual_cost: Optional[float] = None


@dataclass
class Deviation:
    """
    Detected deviation from plan.
    
    Attributes:
        deviation_id: Unique identifier
        deviation_type: Type of deviation
        affected_step: Step with deviation
        severity: Severity score (0-10)
        description: Human-readable description
        detected_at: When detected
        resolved: Whether resolved
    """
    deviation_id: str
    deviation_type: DeviationType
    affected_step: str
    severity: float
    description: str
    detected_at: datetime = field(default_factory=datetime.now)
    resolved: bool = False


@dataclass
class Adaptation:
    """
    Adaptation made to plan.
    
    Attributes:
        adaptation_id: Unique identifier
        action: Type of adaptation
        reason: Why adaptation was made
        changes: Description of changes
        timestamp: When adaptation occurred
        success: Whether adaptation succeeded
    """
    adaptation_id: str
    action: AdaptationAction
    reason: str
    changes: str
    timestamp: datetime = field(default_factory=datetime.now)
    success: Optional[bool] = None


class AdaptivePlanner:
    """
    Adaptive planning system with execution monitoring and replanning.
    
    This planner monitors execution, detects deviations, and adapts
    plans dynamically to maintain progress toward goals.
    """
    
    def __init__(self, temperature: float = 0.3):
        """
        Initialize adaptive planner.
        
        Args:
            temperature: LLM temperature
        """
        self.llm = ChatOpenAI(temperature=temperature, model="gpt-3.5-turbo")
        self.steps: Dict[str, PlanStep] = {}
        self.deviations: List[Deviation] = []
        self.adaptations: List[Adaptation] = []
        self.step_counter = 0
        self.deviation_counter = 0
        self.adaptation_counter = 0
        self.replanning_threshold = 5.0  # Severity threshold for replanning
    
    def create_plan(self, goal: str, num_steps: int = 5) -> List[PlanStep]:
        """
        Create initial plan using LLM.
        
        Args:
            goal: Goal description
            num_steps: Number of steps
            
        Returns:
            List of plan steps
        """
        prompt = ChatPromptTemplate.from_template(
            "Create a {num_steps}-step plan to achieve this goal:\n\n"
            "Goal: {goal}\n\n"
            "For each step, provide:\n"
            "- Description\n"
            "- Expected duration (hours)\n"
            "- Expected cost (units)\n\n"
            "Format as:\n"
            "Step N: [description]\n"
            "Duration: [hours]\n"
            "Cost: [units]\n"
            "---\n"
        )
        
        chain = prompt | self.llm | StrOutputParser()
        result = chain.invoke({"goal": goal, "num_steps": num_steps})
        
        # Parse steps
        steps = []
        step_blocks = result.split("---")
        
        for block in step_blocks:
            if not block.strip():
                continue
            
            lines = [line.strip() for line in block.strip().split('\n')]
            description = ""
            duration = 1.0
            cost = 1.0
            
            for line in lines:
                if line.startswith("Step"):
                    description = line.split(":", 1)[1].strip() if ":" in line else line
                elif line.startswith("Duration:"):
                    try:
                        duration = float(line.split(":")[1].strip().split()[0])
                    except:
                        duration = 1.0
                elif line.startswith("Cost:"):
                    try:
                        cost = float(line.split(":")[1].strip().split()[0])
                    except:
                        cost = 1.0
            
            if description:
                self.step_counter += 1
                step = PlanStep(
                    step_id=f"step_{self.step_counter}",
                    description=description,
                    expected_duration=duration,
                    expected_cost=cost
                )
                self.steps[step.step_id] = step
                steps.append(step)
        
        return steps
    
    def execute_step(
        self,
        step_id: str,
        actual_duration: Optional[float] = None,
        actual_cost: Optional[float] = None,
        success: bool = True
    ) -> bool:
        """
        Execute a plan step.
        
        Args:
            step_id: Step to execute
            actual_duration: Actual time taken
            actual_cost: Actual cost
            success: Whether execution succeeded
            
        Returns:
            Success status
        """
        if step_id not in self.steps:
            return False
        
        step = self.steps[step_id]
        
        # Start execution
        if step.status == StepStatus.PENDING:
            step.status = StepStatus.IN_PROGRESS
            step.started_at = datetime.now()
        
        # Complete execution
        if step.status == StepStatus.IN_PROGRESS:
            if success:
                step.status = StepStatus.COMPLETED
                step.completed_at = datetime.now()
                step.actual_duration = actual_duration or step.expected_duration
                step.actual_cost = actual_cost or step.expected_cost
            else:
                step.status = StepStatus.FAILED
        
        return success
    
    def monitor_execution(self) -> List[Deviation]:
        """
        Monitor plan execution and detect deviations.
        
        Returns:
            List of new deviations
        """
        new_deviations = []
        
        for step in self.steps.values():
            if step.status == StepStatus.COMPLETED:
                # Check for delays
                if step.actual_duration and step.actual_duration > step.expected_duration * 1.2:
                    self.deviation_counter += 1
                    deviation = Deviation(
                        deviation_id=f"dev_{self.deviation_counter}",
                        deviation_type=DeviationType.DELAY,
                        affected_step=step.step_id,
                        severity=min(10.0, (step.actual_duration / step.expected_duration - 1) * 10),
                        description=f"Step took {step.actual_duration:.1f}h vs expected {step.expected_duration:.1f}h"
                    )
                    self.deviations.append(deviation)
                    new_deviations.append(deviation)
                
                # Check for cost overruns
                if step.actual_cost and step.actual_cost > step.expected_cost * 1.2:
                    self.deviation_counter += 1
                    deviation = Deviation(
                        deviation_id=f"dev_{self.deviation_counter}",
                        deviation_type=DeviationType.COST_OVERRUN,
                        affected_step=step.step_id,
                        severity=min(10.0, (step.actual_cost / step.expected_cost - 1) * 10),
                        description=f"Step cost {step.actual_cost:.1f} vs expected {step.expected_cost:.1f}"
                    )
                    self.deviations.append(deviation)
                    new_deviations.append(deviation)
            
            elif step.status == StepStatus.FAILED:
                # Failure is high severity
                self.deviation_counter += 1
                deviation = Deviation(
                    deviation_id=f"dev_{self.deviation_counter}",
                    deviation_type=DeviationType.DEPENDENCY_FAILURE,
                    affected_step=step.step_id,
                    severity=9.0,
                    description=f"Step {step.step_id} failed"
                )
                self.deviations.append(deviation)
                new_deviations.append(deviation)
        
        return new_deviations
    
    def diagnose_deviation(self, deviation: Deviation) -> str:
        """
        Diagnose root cause of deviation using LLM.
        
        Args:
            deviation: Deviation to diagnose
            
        Returns:
            Diagnosis text
        """
        step = self.steps[deviation.affected_step]
        
        prompt = ChatPromptTemplate.from_template(
            "Analyze this plan deviation and suggest root causes:\n\n"
            "Step: {step_description}\n"
            "Deviation: {deviation_type}\n"
            "Details: {deviation_description}\n"
            "Severity: {severity}/10\n\n"
            "Provide:\n"
            "1. Likely root cause\n"
            "2. Impact on remaining plan\n"
            "3. Recommended action (continue/repair/replan)\n"
        )
        
        chain = prompt | self.llm | StrOutputParser()
        diagnosis = chain.invoke({
            "step_description": step.description,
            "deviation_type": deviation.deviation_type.value,
            "deviation_description": deviation.description,
            "severity": deviation.severity
        })
        
        return diagnosis
    
    def decide_adaptation(self, deviation: Deviation) -> AdaptationAction:
        """
        Decide what adaptation action to take.
        
        Args:
            deviation: Deviation triggering adaptation
            
        Returns:
            Recommended action
        """
        # Simple severity-based decision
        if deviation.severity < 3.0:
            return AdaptationAction.CONTINUE
        elif deviation.severity < 5.0:
            return AdaptationAction.REPAIR
        elif deviation.severity  Adaptation:
        """
        Repair plan by adjusting expectations.
        
        Args:
            deviation: Deviation to repair
            
        Returns:
            Adaptation record
        """
        step = self.steps[deviation.affected_step]
        
        # Adjust expectations for remaining steps
        remaining_steps = [
            s for s in self.steps.values()
            if s.status == StepStatus.PENDING
        ]
        
        if deviation.deviation_type == DeviationType.DELAY:
            # Increase duration estimates
            for s in remaining_steps:
                s.expected_duration *= 1.2
            changes = f"Increased duration estimates by 20% for {len(remaining_steps)} remaining steps"
        
        elif deviation.deviation_type == DeviationType.COST_OVERRUN:
            # Increase cost estimates
            for s in remaining_steps:
                s.expected_cost *= 1.2
            changes = f"Increased cost estimates by 20% for {len(remaining_steps)} remaining steps"
        
        else:
            changes = "No specific repair applied"
        
        self.adaptation_counter += 1
        adaptation = Adaptation(
            adaptation_id=f"adapt_{self.adaptation_counter}",
            action=AdaptationAction.REPAIR,
            reason=deviation.description,
            changes=changes,
            success=True
        )
        self.adaptations.append(adaptation)
        
        return adaptation
    
    def replan_from_current(self, goal: str) -> List[PlanStep]:
        """
        Create new plan from current state.
        
        Args:
            goal: Original goal
            
        Returns:
            New plan steps
        """
        # Preserve completed steps
        completed = [s for s in self.steps.values() if s.status == StepStatus.COMPLETED]
        
        # Create new steps for remaining work
        prompt = ChatPromptTemplate.from_template(
            "The original plan encountered issues. Create a revised plan.\n\n"
            "Goal: {goal}\n"
            "Completed steps: {completed_count}\n"
            "Previous issues: {issues}\n\n"
            "Create 3-5 steps to complete the remaining work.\n"
            "Consider lessons learned from previous issues.\n\n"
            "Format as:\n"
            "Step N: [description]\n"
            "Duration: [hours]\n"
            "Cost: [units]\n"
            "---\n"
        )
        
        issues = ", ".join([d.description for d in self.deviations[-3:]])
        
        chain = prompt | self.llm | StrOutputParser()
        result = chain.invoke({
            "goal": goal,
            "completed_count": len(completed),
            "issues": issues or "None"
        })
        
        # Parse new steps
        new_steps = []
        step_blocks = result.split("---")
        
        for block in step_blocks:
            if not block.strip():
                continue
            
            lines = [line.strip() for line in block.strip().split('\n')]
            description = ""
            duration = 1.0
            cost = 1.0
            
            for line in lines:
                if line.startswith("Step"):
                    description = line.split(":", 1)[1].strip() if ":" in line else line
                elif line.startswith("Duration:"):
                    try:
                        duration = float(line.split(":")[1].strip().split()[0])
                    except:
                        duration = 1.0
                elif line.startswith("Cost:"):
                    try:
                        cost = float(line.split(":")[1].strip().split()[0])
                    except:
                        cost = 1.0
            
            if description:
                self.step_counter += 1
                step = PlanStep(
                    step_id=f"step_{self.step_counter}",
                    description=description,
                    expected_duration=duration,
                    expected_cost=cost
                )
                self.steps[step.step_id] = step
                new_steps.append(step)
        
        # Record adaptation
        self.adaptation_counter += 1
        adaptation = Adaptation(
            adaptation_id=f"adapt_{self.adaptation_counter}",
            action=AdaptationAction.REPLAN,
            reason="Significant deviations detected",
            changes=f"Created {len(new_steps)} new steps",
            success=True
        )
        self.adaptations.append(adaptation)
        
        return new_steps
    
    def adapt(self, goal: str) -> Optional[Adaptation]:
        """
        Monitor and adapt plan as needed.
        
        Args:
            goal: Original goal
            
        Returns:
            Adaptation if made, None otherwise
        """
        # Monitor for deviations
        new_deviations = self.monitor_execution()
        
        if not new_deviations:
            return None
        
        # Check most severe deviation
        most_severe = max(new_deviations, key=lambda d: d.severity)
        
        # Decide action
        action = self.decide_adaptation(most_severe)
        
        # Execute adaptation
        if action == AdaptationAction.CONTINUE:
            return None
        elif action == AdaptationAction.REPAIR:
            return self.repair_plan(most_severe)
        elif action == AdaptationAction.REPLAN:
            self.replan_from_current(goal)
            return self.adaptations[-1]
        
        return None
    
    def get_progress_summary(self) -> Dict[str, Any]:
        """Get plan progress summary."""
        total = len(self.steps)
        completed = sum(1 for s in self.steps.values() if s.status == StepStatus.COMPLETED)
        failed = sum(1 for s in self.steps.values() if s.status == StepStatus.FAILED)
        in_progress = sum(1 for s in self.steps.values() if s.status == StepStatus.IN_PROGRESS)
        
        total_expected_duration = sum(s.expected_duration for s in self.steps.values())
        total_actual_duration = sum(
            s.actual_duration for s in self.steps.values()
            if s.actual_duration is not None
        )
        
        return {
            "total_steps": total,
            "completed": completed,
            "failed": failed,
            "in_progress": in_progress,
            "pending": total - completed - failed - in_progress,
            "completion_rate": completed / total if total > 0 else 0,
            "total_deviations": len(self.deviations),
            "total_adaptations": len(self.adaptations),
            "expected_duration": total_expected_duration,
            "actual_duration": total_actual_duration
        }


def demonstrate_adaptive_planning():
    """Demonstrate adaptive planning pattern."""
    
    print("=" * 80)
    print("ADAPTIVE PLANNING PATTERN DEMONSTRATION")
    print("=" * 80)
    
    # Example 1: Basic adaptive planning
    print("\n" + "=" * 80)
    print("Example 1: Plan Creation and Execution")
    print("=" * 80)
    
    planner = AdaptivePlanner()
    
    print("\nCreating initial plan...")
    steps = planner.create_plan("Develop a mobile app", num_steps=5)
    
    print(f"\nCreated {len(steps)} steps:")
    for step in steps:
        print(f"  {step.step_id}: {step.description}")
        print(f"    Expected: {step.expected_duration}h, ${step.expected_cost}")
    
    print("\nExecuting steps...")
    for i, step in enumerate(steps[:2]):
        planner.execute_step(step.step_id)
        print(f"  ✓ Completed: {step.description}")
    
    progress = planner.get_progress_summary()
    print(f"\nProgress: {progress['completed']}/{progress['total_steps']} steps completed")
    
    # Example 2: Deviation detection
    print("\n" + "=" * 80)
    print("Example 2: Deviation Detection")
    print("=" * 80)
    
    planner2 = AdaptivePlanner()
    planner2.create_plan("Build website", num_steps=4)
    
    # Execute with delays
    steps2 = list(planner2.steps.values())
    planner2.execute_step(steps2[0].step_id, actual_duration=3.0, actual_cost=1.0)
    steps2[0].expected_duration = 2.0  # Set expected for comparison
    
    print("\nExecuted step with delay:")
    print(f"  Expected: {steps2[0].expected_duration}h")
    print(f"  Actual: {steps2[0].actual_duration}h")
    print(f"  Delay: {steps2[0].actual_duration - steps2[0].expected_duration:.1f}h")
    
    print("\nMonitoring for deviations...")
    deviations = planner2.monitor_execution()
    
    print(f"\nDetected {len(deviations)} deviation(s):")
    for dev in deviations:
        print(f"  • Type: {dev.deviation_type.value}")
        print(f"    Severity: {dev.severity:.1f}/10")
        print(f"    Description: {dev.description}")
    
    # Example 3: Root cause diagnosis
    print("\n" + "=" * 80)
    print("Example 3: Root Cause Diagnosis")
    print("=" * 80)
    
    if deviations:
        dev = deviations[0]
        print(f"\nDiagnosing deviation: {dev.description}")
        print("\nLLM Analysis:")
        diagnosis = planner2.diagnose_deviation(dev)
        print(diagnosis)
    
    # Example 4: Plan repair
    print("\n" + "=" * 80)
    print("Example 4: Plan Repair")
    print("=" * 80)
    
    planner3 = AdaptivePlanner()
    planner3.create_plan("Complete project", num_steps=5)
    
    # Simulate deviation
    steps3 = list(planner3.steps.values())
    planner3.execute_step(steps3[0].step_id, actual_duration=5.0)
    steps3[0].expected_duration = 2.0
    
    deviations3 = planner3.monitor_execution()
    
    if deviations3:
        print("\nDeviation detected:")
        print(f"  Severity: {deviations3[0].severity:.1f}/10")
        
        print("\nDeciding adaptation action...")
        action = planner3.decide_adaptation(deviations3[0])
        print(f"  Recommended: {action.value}")
        
        print("\nRepairing plan...")
        adaptation = planner3.repair_plan(deviations3[0])
        print(f"  Changes: {adaptation.changes}")
    
    # Example 5: Full replanning
    print("\n" + "=" * 80)
    print("Example 5: Full Replanning")
    print("=" * 80)
    
    planner4 = AdaptivePlanner()
    original_steps = planner4.create_plan("Launch product", num_steps=4)
    
    print(f"\nOriginal plan: {len(original_steps)} steps")
    
    # Simulate major failure
    steps4 = list(planner4.steps.values())
    planner4.execute_step(steps4[0].step_id)
    planner4.execute_step(steps4[1].step_id, success=False)
    steps4[1].status = StepStatus.FAILED
    
    print("\nMajor failure detected!")
    deviations4 = planner4.monitor_execution()
    
    if deviations4:
        severe_dev = max(deviations4, key=lambda d: d.severity)
        print(f"  Severity: {severe_dev.severity:.1f}/10")
        
        action = planner4.decide_adaptation(severe_dev)
        print(f"  Action: {action.value}")
        
        if action == AdaptationAction.REPLAN:
            print("\nReplanning from current state...")
            new_steps = planner4.replan_from_current("Launch product")
            print(f"  Created {len(new_steps)} new steps")
    
    # Example 6: Adaptation decision tree
    print("\n" + "=" * 80)
    print("Example 6: Adaptation Decision Making")
    print("=" * 80)
    
    planner5 = AdaptivePlanner()
    
    # Test different severity levels
    test_cases = [
        (2.0, "Minor delay"),
        (4.5, "Moderate cost overrun"),
        (7.0, "Significant quality issue"),
        (9.5, "Critical failure")
    ]
    
    print("\nAdaptation decisions by severity:")
    for severity, description in test_cases:
        planner5.step_counter += 1
        test_step = PlanStep(
            step_id=f"step_{planner5.step_counter}",
            description="Test step"
        )
        planner5.steps[test_step.step_id] = test_step
        
        planner5.deviation_counter += 1
        test_dev = Deviation(
            deviation_id=f"dev_{planner5.deviation_counter}",
            deviation_type=DeviationType.DELAY,
            affected_step=test_step.step_id,
            severity=severity,
            description=description
        )
        
        action = planner5.decide_adaptation(test_dev)
        print(f"  Severity {severity:.1f} ({description}) → {action.value}")
    
    # Example 7: Automatic adaptation loop
    print("\n" + "=" * 80)
    print("Example 7: Automatic Adaptation Loop")
    print("=" * 80)
    
    planner6 = AdaptivePlanner()
    planner6.create_plan("Deploy service", num_steps=4)
    
    print("\nExecuting with automatic adaptation...")
    
    steps6 = list(planner6.steps.values())
    
    # Step 1: Normal
    planner6.execute_step(steps6[0].step_id, actual_duration=1.0)
    adaptation = planner6.adapt("Deploy service")
    print(f"  Step 1: Completed normally - {adaptation or 'No adaptation needed'}")
    
    # Step 2: Minor delay
    planner6.execute_step(steps6[1].step_id, actual_duration=3.0)
    steps6[1].expected_duration = 2.0
    adaptation = planner6.adapt("Deploy service")
    print(f"  Step 2: Minor delay - {adaptation.action.value if adaptation else 'Continue'}")
    
    # Step 3: Major issue
    planner6.execute_step(steps6[2].step_id, actual_duration=10.0)
    steps6[2].expected_duration = 2.0
    adaptation = planner6.adapt("Deploy service")
    print(f"  Step 3: Major issue - {adaptation.action.value if adaptation else 'Continue'}")
    
    print(f"\nTotal adaptations: {len(planner6.adaptations)}")
    
    # Example 8: Learning from experience
    print("\n" + "=" * 80)
    print("Example 8: Progress Tracking and Summary")
    print("=" * 80)
    
    planner7 = AdaptivePlanner()
    planner7.create_plan("Complete migration", num_steps=6)
    
    # Execute with various outcomes
    steps7 = list(planner7.steps.values())
    planner7.execute_step(steps7[0].step_id, actual_duration=1.5, actual_cost=1.0)
    steps7[0].expected_duration = 1.0
    planner7.execute_step(steps7[1].step_id, actual_duration=3.0, actual_cost=2.0)
    steps7[1].expected_duration = 2.0
    steps7[1].expected_cost = 1.5
    planner7.execute_step(steps7[2].step_id, actual_duration=1.0, actual_cost=1.0)
    
    # Monitor and adapt
    planner7.adapt("Complete migration")
    
    print("\nFINAL SUMMARY:")
    print("=" * 60)
    
    progress = planner7.get_progress_summary()
    for key, value in progress.items():
        if isinstance(value, float):
            print(f"  {key}: {value:.2f}")
        else:
            print(f"  {key}: {value}")
    
    print("\nDeviations:")
    for dev in planner7.deviations:
        print(f"  • {dev.deviation_type.value} (severity: {dev.severity:.1f})")
    
    print("\nAdaptations:")
    for adapt in planner7.adaptations:
        print(f"  • {adapt.action.value}: {adapt.changes}")
    
    # Summary
    print("\n" + "=" * 80)
    print("SUMMARY: Adaptive Planning Pattern")
    print("=" * 80)
    
    summary = """
    The Adaptive Planning pattern demonstrated:
    
    1. PLAN CREATION (Example 1):
       - LLM-based plan generation
       - Step-by-step structure
       - Duration and cost estimation
       - Progress tracking
       - Execution monitoring
    
    2. DEVIATION DETECTION (Example 2):
       - Automatic monitoring
       - Delay detection (20% threshold)
       - Cost overrun detection
       - Severity scoring
       - Real-time alerts
    
    3. ROOT CAUSE DIAGNOSIS (Example 3):
       - LLM-based analysis
       - Context-aware diagnosis
       - Impact assessment
       - Actionable recommendations
       - Natural language explanations
    
    4. PLAN REPAIR (Example 4):
       - Local plan adjustments
       - Expectation updates
       - Propagation to remaining steps
       - Minimal disruption
       - Quick fixes
    
    5. FULL REPLANNING (Example 5):
       - New plan generation
       - Progress preservation
       - Lesson incorporation
       - Context-aware replanning
       - Fresh start when needed
    
    6. DECISION MAKING (Example 6):
       - Severity-based decisions
       - Action selection (continue/repair/revise/replan)
       - Graduated response
       - Threshold-based triggers
       - Balanced approach
    
    7. AUTOMATIC ADAPTATION (Example 7):
       - Continuous monitoring
       - Automatic response
       - Multi-step adaptation
       - Event-driven updates
       - Hands-free operation
    
    8. PROGRESS TRACKING (Example 8):
       - Comprehensive metrics
       - Deviation history
       - Adaptation log
       - Performance summary
       - Complete audit trail
    
    KEY BENEFITS:
    ✓ Robust to uncertainty and changes
    ✓ Maintains progress despite issues
    ✓ Exploits new opportunities
    ✓ Reduces plan failure rate
    ✓ Learns from experience
    ✓ Handles unexpected events
    ✓ Flexible goal achievement
    ✓ Automatic problem response
    
    USE CASES:
    • Robot navigation (dynamic environments)
    • Project management (changing requirements)
    • Manufacturing (equipment failures)
    • Supply chain (disruptions)
    • Military operations (uncertain terrain)
    • Autonomous vehicles (route planning)
    • Cloud resource allocation
    • Emergency response coordination
    
    ADAPTATION STRATEGIES:
    → Plan Repair: Minor modifications
    → Plan Reuse: Apply similar past plan
    → Plan Revision: Significant restructuring
    → Replanning: Create entirely new plan
    → Reactive: Immediate adjustments
    → Proactive: Anticipate changes
    
    MONITORING ASPECTS:
    • Progress: Completion vs. expected
    • Resources: Availability and consumption
    • Time: Schedule adherence
    • Quality: Output measures
    • Cost: Budget tracking
    • Risks: Emerging threats
    
    BEST PRACTICES:
    1. Implement efficient monitoring
    2. Use event-driven detection
    3. Cache replan results
    4. Set replanning thresholds
    5. Limit replanning frequency
    6. Preserve partial progress
    7. Log all adaptations
    8. Enable replay for debugging
    
    TRADE-OFFS:
    • Monitoring overhead vs. responsiveness
    • Plan stability vs. adaptability
    • Replanning cost vs. plan quality
    • Automation vs. human control
    
    PRODUCTION CONSIDERATIONS:
    → Monitor continuously without blocking
    → Use asynchronous event processing
    → Implement replanning rate limits
    → Cache similar plan patterns
    → Log all deviations and adaptations
    → Support manual override
    → Enable plan playback
    → Track adaptation success rate
    → Alert on excessive replanning
    → Preserve completed work
    
    This pattern enables plans to evolve dynamically in response to
    execution feedback and environmental changes, maintaining progress
    toward goals despite uncertainty and unexpected events.
    """
    
    print(summary)


if __name__ == "__main__":
    demonstrate_adaptive_planning()

    

Agentic Design Pattern: Adaptive Planning

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

Overview

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

---

Pattern Description

Pattern 109: Adaptive Planning

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

Adaptive planning is essential in dynamic, uncertain environments where initial
assumptions may become invalid, resources may change, or unexpected opportunities
and obstacles may arise. The pattern combines continuous monitoring, deviation
detection, root cause analysis, and replanning strategies to ensure resilience
and goal achievement despite changing conditions.

This pattern includes mechanisms for execution monitoring, deviation detection,
impact assessment, root cause diagnosis, plan repair strategies, replanning
triggers, and learning from outcomes to improve future plans.

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

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

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

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

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

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

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

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

---

File Statistics

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

---

Module Documentation

"""
Pattern 109: Adaptive Planning

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

    Adaptive planning is essential in dynamic, uncertain environments where initial
    assumptions may become invalid, resources may change, or unexpected opportunities
    and obstacles may arise. The pattern combines continuous monitoring, deviation
    detection, root cause analysis, and replanning strategies to ensure resilience
    and goal achievement despite changing conditions.

    This pattern includes mechanisms for execution monitoring, deviation detection,
    impact assessment, root cause diagnosis, plan repair strategies, replanning
    triggers, and learning from outcomes to improve future plans.

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

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

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

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

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

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

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

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

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

---

Import Statements

Standard Library Imports

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

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

LangChain Imports

from langchain_openai.ChatOpenAI

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

Other Imports

from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: StepStatus

Lines 107-113

Inherits from: Enum

Purpose:

Status of a plan step.

---

Class: DeviationType

Lines 116-123

Inherits from: Enum

Purpose:

Types of plan deviations.

---

Class: AdaptationAction

Lines 126-132

Inherits from: Enum

Purpose:

Types of adaptation actions.

---

Class: PlanStep

Lines 136-161

Purpose:

A step in an adaptive plan.

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

---

Class: Deviation

Lines 165-184

Purpose:

Detected deviation from plan.

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

---

Class: Adaptation

Lines 188-205

Purpose:

Adaptation made to plan.

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

---

Class: AdaptivePlanner

Lines 208-637

Purpose:

Adaptive planning system with execution monitoring and replanning.

This planner monitors execution, detects deviations, and adapts
plans dynamically to maintain progress toward goals.

Methods:

__init__(self, temperature)

Lines 216-230

Purpose:

Initialize adaptive planner.

Args:
    temperature: LLM temperature

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

create_plan(self, goal, num_steps)

Lines 232-298

Purpose:

Create initial plan using LLM.

Args:
    goal: Goal description
    num_steps: Number of steps

Returns:
    List of plan steps

Parameters: – goal: Goal – num_steps: Num steps

Returns: <ast.Subscript object at 0x000001A06D6F5050>

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

Lines 300-339

Purpose:

Execute a plan step.

Args:
    step_id: Step to execute
    actual_duration: Actual time taken
    actual_cost: Actual cost
    success: Whether execution succeeded

Returns:
    Success status

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

Returns: bool

monitor_execution(self)

Lines 341-391

Purpose:

Monitor plan execution and detect deviations.

Returns:
    List of new deviations

Parameters:

Returns: <ast.Subscript object at 0x000001A06D72FC50>

diagnose_deviation(self, deviation)

Lines 393-425

Purpose:

Diagnose root cause of deviation using LLM.

Args:
    deviation: Deviation to diagnose

Returns:
    Diagnosis text

Parameters: – deviation: Deviation

Returns: str

decide_adaptation(self, deviation)

Lines 427-445

Purpose:

Decide what adaptation action to take.

Args:
    deviation: Deviation triggering adaptation

Returns:
    Recommended action

Parameters: – deviation: Deviation

Returns: AdaptationAction

repair_plan(self, deviation)

Lines 447-490

Purpose:

Repair plan by adjusting expectations.

Args:
    deviation: Deviation to repair

Returns:
    Adaptation record

Parameters: – deviation: Deviation

Returns: Adaptation

replan_from_current(self, goal)

Lines 492-578

Purpose:

Create new plan from current state.

Args:
    goal: Original goal

Returns:
    New plan steps

Parameters: – goal: Goal

Returns: <ast.Subscript object at 0x000001A06D686710>

adapt(self, goal)

Lines 580-611

Purpose:

Monitor and adapt plan as needed.

Args:
    goal: Original goal

Returns:
    Adaptation if made, None otherwise

Parameters: – goal: Goal

Returns: <ast.Subscript object at 0x000001A06E0951D0>

get_progress_summary(self)

Lines 613-637

Purpose:

Get plan progress summary.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D812450>

---

Functions

demonstrate_adaptive_planning()

Lines 640-1005

Purpose:

Demonstrate adaptive planning pattern.

---

Key Concepts Summary

Pattern Overview

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

Core Components

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

When to Use

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

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

---

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

---

LangChain Components Used

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

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

---

Execution Flow

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

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

---

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

---

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

---

Related Patterns

• Pattern 108
• Pattern 110

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The Adaptive Planning pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

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

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

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

---

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

---

Generated from actual implementation code
For educational and reference purposes