Blog

Agentic Design Pattern: Workflow Orchestration


    """
Pattern 120: Workflow Orchestration

Description:
    Manages complex multi-step workflows with dependencies, branching,
    error handling, and monitoring.

Components:
    - Workflow engine
    - Step dependencies
    - Error handling
    - Progress monitoring

Use Cases:
    - Business processes
    - Data pipelines
    - Complex automation

LangChain Implementation:
    Uses LangGraph for state machine-based workflow orchestration.
"""

import os
from typing import List, Dict, Any, Callable, Optional
from enum import Enum
from dataclasses import dataclass
from datetime import datetime
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 workflow step."""
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    SKIPPED = "skipped"


@dataclass
class WorkflowStep:
    """Represents a step in the workflow."""
    name: str
    description: str
    dependencies: List[str]
    action: Callable
    status: StepStatus = StepStatus.PENDING
    result: Any = None
    error: str = None
    retry_count: int = 0
    max_retries: int = 3


class WorkflowOrchestrator:
    """Orchestrates complex multi-step workflows."""
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        self.steps: Dict[str, WorkflowStep] = {}
        self.execution_log: List[Dict[str, Any]] = []
        self.context: Dict[str, Any] = {}
        
    def add_step(self, step: WorkflowStep):
        """Add a step to the workflow."""
        self.steps[step.name] = step
        print(f"✓ Added step: {step.name}")
    
    def validate_workflow(self) -> bool:
        """Validate workflow structure and dependencies."""
        print("\n=== Validating Workflow ===")
        
        # Check for circular dependencies
        for step_name, step in self.steps.items():
            if self._has_circular_dependency(step_name, set()):
                print(f"✗ Circular dependency detected for step: {step_name}")
                return False
        
        # Check all dependencies exist
        for step_name, step in self.steps.items():
            for dep in step.dependencies:
                if dep not in self.steps:
                    print(f"✗ Missing dependency: {dep} for step {step_name}")
                    return False
        
        print("✓ Workflow validation passed")
        return True
    
    def _has_circular_dependency(self, step_name: str, visited: set) -> bool:
        """Check for circular dependencies."""
        if step_name in visited:
            return True
        
        visited.add(step_name)
        step = self.steps.get(step_name)
        
        if step:
            for dep in step.dependencies:
                if self._has_circular_dependency(dep, visited.copy()):
                    return True
        
        return False
    
    def get_ready_steps(self) -> List[WorkflowStep]:
        """Get steps that are ready to execute."""
        ready = []
        
        for step in self.steps.values():
            if step.status != StepStatus.PENDING:
                continue
            
            # Check if all dependencies are completed
            deps_completed = all(
                self.steps[dep].status == StepStatus.COMPLETED
                for dep in step.dependencies
            )
            
            if deps_completed:
                ready.append(step)
        
        return ready
    
    def execute_step(self, step: WorkflowStep) -> bool:
        """Execute a single workflow step."""
        print(f"\n▶ Executing: {step.name}")
        print(f"  Description: {step.description}")
        
        step.status = StepStatus.RUNNING
        start_time = datetime.now()
        
        try:
            # Execute the step's action
            step.result = step.action(self.context)
            step.status = StepStatus.COMPLETED
            
            # Log execution
            log_entry = {
                "step": step.name,
                "status": "completed",
                "duration": (datetime.now() - start_time).total_seconds(),
                "timestamp": datetime.now().isoformat()
            }
            self.execution_log.append(log_entry)
            
            print(f"  ✓ Completed")
            return True
            
        except Exception as e:
            step.error = str(e)
            step.retry_count += 1
            
            if step.retry_count < step.max_retries:
                print(f"  ⚠ Failed (attempt {step.retry_count}/{step.max_retries}): {e}")
                step.status = StepStatus.PENDING  # Retry
            else:
                step.status = StepStatus.FAILED
                print(f"  ✗ Failed after {step.retry_count} attempts: {e}")
            
            log_entry = {
                "step": step.name,
                "status": "failed",
                "error": str(e),
                "retry_count": step.retry_count,
                "timestamp": datetime.now().isoformat()
            }
            self.execution_log.append(log_entry)
            
            return False
    
    def execute_workflow(self):
        """Execute the entire workflow."""
        print("\n=== Executing Workflow ===")
        
        if not self.validate_workflow():
            print("✗ Workflow validation failed")
            return False
        
        total_steps = len(self.steps)
        completed = 0
        
        while completed  0:
                    break
                else:
                    continue
            
            # Execute ready steps (can be parallelized)
            for step in ready_steps:
                self.execute_step(step)
            
            completed = sum(1 for s in self.steps.values() if s.status == StepStatus.COMPLETED)
        
        # Generate execution summary
        self.generate_summary()
        
        return all(s.status == StepStatus.COMPLETED for s in self.steps.values())
    
    def generate_summary(self):
        """Generate workflow execution summary."""
        print("\n=== Workflow Summary ===")
        
        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)
        total = len(self.steps)
        
        print(f"Total Steps: {total}")
        print(f"Completed: {completed}")
        print(f"Failed: {failed}")
        print(f"Success Rate: {completed/total*100:.1f}%")
        
        if failed > 0:
            print("\nFailed Steps:")
            for step in self.steps.values():
                if step.status == StepStatus.FAILED:
                    print(f"  - {step.name}: {step.error}")
        
        # LLM-based analysis
        if self.execution_log:
            analysis_prompt = ChatPromptTemplate.from_messages([
                ("system", "Analyze this workflow execution and provide insights."),
                ("user", """Execution Log:
{log}

Provide:
1. Performance insights
2. Bottlenecks identified
3. Optimization suggestions""")
            ])
            
            log_str = "\n".join([
                f"{entry['step']}: {entry['status']} ({entry.get('duration', 'N/A')}s)"
                for entry in self.execution_log
            ])
            
            chain = analysis_prompt | self.llm | StrOutputParser()
            analysis = chain.invoke({"log": log_str})
            
            print("\nWorkflow Analysis:")
            print(analysis)


def demonstrate_workflow_orchestration():
    """Demonstrate workflow orchestration pattern."""
    print("=== Workflow Orchestration Pattern ===\n")
    
    orchestrator = WorkflowOrchestrator()
    
    # Define workflow steps for data processing pipeline
    print("1. Defining Workflow Steps")
    print("-" * 50)
    
    # Step 1: Data Extraction
    def extract_data(context):
        print("    Extracting data from source...")
        context['raw_data'] = [1, 2, 3, 4, 5]
        return "Data extracted"
    
    orchestrator.add_step(WorkflowStep(
        name="extract_data",
        description="Extract data from source system",
        dependencies=[],
        action=extract_data
    ))
    
    # Step 2: Data Validation
    def validate_data(context):
        print("    Validating data...")
        data = context.get('raw_data', [])
        if not data:
            raise ValueError("No data to validate")
        context['validated_data'] = data
        return "Data validated"
    
    orchestrator.add_step(WorkflowStep(
        name="validate_data",
        description="Validate extracted data",
        dependencies=["extract_data"],
        action=validate_data
    ))
    
    # Step 3: Data Transformation
    def transform_data(context):
        print("    Transforming data...")
        data = context.get('validated_data', [])
        context['transformed_data'] = [x * 2 for x in data]
        return "Data transformed"
    
    orchestrator.add_step(WorkflowStep(
        name="transform_data",
        description="Transform data format",
        dependencies=["validate_data"],
        action=transform_data
    ))
    
    # Step 4: Data Enrichment
    def enrich_data(context):
        print("    Enriching data...")
        data = context.get('transformed_data', [])
        context['enriched_data'] = [{"value": x, "category": "processed"} for x in data]
        return "Data enriched"
    
    orchestrator.add_step(WorkflowStep(
        name="enrich_data",
        description="Enrich data with additional information",
        dependencies=["transform_data"],
        action=enrich_data
    ))
    
    # Step 5: Load to Database
    def load_to_db(context):
        print("    Loading to database...")
        data = context.get('enriched_data', [])
        context['db_records'] = len(data)
        return f"Loaded {len(data)} records"
    
    orchestrator.add_step(WorkflowStep(
        name="load_to_db",
        description="Load data to database",
        dependencies=["enrich_data"],
        action=load_to_db
    ))
    
    # Step 6: Generate Report
    def generate_report(context):
        print("    Generating report...")
        records = context.get('db_records', 0)
        context['report'] = f"Successfully processed {records} records"
        return "Report generated"
    
    orchestrator.add_step(WorkflowStep(
        name="generate_report",
        description="Generate processing report",
        dependencies=["load_to_db"],
        action=generate_report
    ))
    
    print()
    
    # Execute workflow
    print("2. Executing Workflow")
    print("-" * 50)
    success = orchestrator.execute_workflow()
    
    print("\n=== Results ===")
    print(f"Workflow {'succeeded' if success else 'failed'}")
    print(f"Final context: {orchestrator.context}")
    
    print("\n=== Summary ===")
    print("Workflow orchestration demonstrated with:")
    print("- Multi-step pipeline")
    print("- Dependency management")
    print("- Error handling and retries")
    print("- Progress tracking")
    print("- Execution logging")
    print("- Performance analysis")


if __name__ == "__main__":
    demonstrate_workflow_orchestration()

Agentic Design Pattern: Workflow Orchestration

Pattern 120: Workflow Orchestration – Line-by-Line Explanation

Overview

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

Pattern Description

Pattern 120: Workflow Orchestration

Description: Manages complex multi-step workflows with dependencies, branching, error handling, and monitoring.

Components: – Workflow engine – Step dependencies – Error handling – Progress monitoring

Use Cases: – Business processes – Data pipelines – Complex automation

LangChain Implementation: Uses LangGraph for state machine-based workflow orchestration.

File Statistics

Total Lines: 372
Classes: 3
Functions: 1
Constants: 0

Module Documentation

"""
Pattern 120: Workflow Orchestration

Description:
    Manages complex multi-step workflows with dependencies, branching,
    error handling, and monitoring.

Components:
    - Workflow engine
    - Step dependencies
    - Error handling
    - Progress monitoring

Use Cases:
    - Business processes
    - Data pipelines
    - Complex automation

LangChain Implementation:
    Uses LangGraph for state machine-based workflow orchestration.
"""

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

Import Statements

Standard Library Imports

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

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

LangChain Imports

from langchain_openai.ChatOpenAI

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

Other Imports

from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

Classes

Class: `StepStatus`

Lines 37-43

Inherits from: Enum

Purpose:

Status of workflow step.

Class: `WorkflowStep`

Lines 47-57

Purpose:

Represents a step in the workflow.

Class: `WorkflowOrchestrator`

Lines 60-251

Purpose:

Orchestrates complex multi-step workflows.

Methods:

`init(self, model_name)`

Lines 63-67

Purpose: Initialize the class instance with required parameters.

Parameters: – model_name: Language model instance

`add_step(self, step)`

Lines 69-72

Purpose:

Add a step to the workflow.

Parameters: – step: Step

`validate_workflow(self)`

Lines 74-92

Purpose:

Validate workflow structure and dependencies.

Parameters:

Returns: bool

`_has_circular_dependency(self, step_name, visited)`

Lines 94-107

Purpose:

Check for circular dependencies.

Parameters: – step_name: Step name – visited: Visited

Returns: bool

`get_ready_steps(self)`

Lines 109-126

Purpose:

Get steps that are ready to execute.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D889CD0>

`execute_step(self, step)`

Lines 128-173

Purpose:

Execute a single workflow step.

Parameters: – step: Step

Returns: bool

`execute_workflow(self)`

Lines 175-208

Purpose:

Execute the entire workflow.

Parameters:

`generate_summary(self)`

Lines 210-251

Purpose:

Generate workflow execution summary.

Parameters:

Functions

`demonstrate_workflow_orchestration()`

Lines 254-367

Purpose:

Demonstrate workflow orchestration pattern.

Key Concepts Summary

Pattern Overview

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

Core Components

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

When to Use

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

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

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

LangChain Components Used

ChatOpenAI: OpenAI model interface
Chains: Sequential component composition

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

Execution Flow

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

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

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

Related Patterns

Pattern 119
Pattern 121

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

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

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

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

File: 120_workflow_orchestration.py
Total Lines: 372
Pattern Category: AI Agent Design Pattern #120

Generated from actual implementation code
For educational and reference purposes

November 2, 2025

Agentic Design Pattern: Task Allocation Scheduling


    """
Pattern 119: Task Allocation & Scheduling

Description:
    Assigns tasks to agents based on capabilities, load, and constraints
    using auction-based or optimal assignment algorithms.

Components:
    - Task queue
    - Agent capabilities
    - Allocation algorithm
    - Load balancing

Use Cases:
    - Multi-agent systems
    - Distributed computing
    - Resource optimization

LangChain Implementation:
    Uses LLM-based task analysis and agent matching with LangGraph for orchestration.
"""

import os
from typing import List, Dict, Any
from dataclasses import dataclass
from datetime import datetime
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()


@dataclass
class Task:
    """Represents a task to be allocated."""
    id: str
    description: str
    required_skills: List[str]
    priority: int
    estimated_duration: int
    deadline: str = None


@dataclass
class Agent:
    """Represents an agent with capabilities."""
    id: str
    name: str
    skills: List[str]
    current_load: int = 0
    max_capacity: int = 100
    performance_history: List[float] = None
    
    def __post_init__(self):
        if self.performance_history is None:
            self.performance_history = []


class TaskAllocationScheduler:
    """Schedules and allocates tasks to agents optimally."""
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        self.agents: List[Agent] = []
        self.task_queue: List[Task] = []
        self.assignments: Dict[str, str] = {}  # task_id -> agent_id
        
    def register_agent(self, agent: Agent):
        """Register an agent in the system."""
        self.agents.append(agent)
        print(f"✓ Registered agent: {agent.name} with skills {agent.skills}")
    
    def add_task(self, task: Task):
        """Add task to queue."""
        self.task_queue.append(task)
        print(f"✓ Added task: {task.description} (Priority: {task.priority})")
    
    def analyze_task_requirements(self, task: Task) -> Dict[str, Any]:
        """Analyze task requirements using LLM."""
        analysis_prompt = ChatPromptTemplate.from_messages([
            ("system", "Analyze this task and provide detailed requirements."),
            ("user", """Task: {description}
Required Skills: {skills}
Priority: {priority}

Provide:
1. Complexity level (1-10)
2. Required expertise level
3. Any special considerations""")
        ])
        
        chain = analysis_prompt | self.llm | StrOutputParser()
        analysis = chain.invoke({
            "description": task.description,
            "skills": ", ".join(task.required_skills),
            "priority": task.priority
        })
        
        return {"analysis": analysis}
    
    def calculate_agent_score(self, agent: Agent, task: Task) -> float:
        """Calculate how well an agent matches a task."""
        scoring_prompt = ChatPromptTemplate.from_messages([
            ("system", "Score how well this agent matches the task requirements (0-100)."),
            ("user", """Agent: {agent_name}
Agent Skills: {agent_skills}
Current Load: {current_load}/{max_capacity}
Performance History: {performance}

Task: {task_description}
Required Skills: {required_skills}
Priority: {priority}

Return only a numerical score (0-100):""")
        ])
        
        chain = scoring_prompt | self.llm | StrOutputParser()
        response = chain.invoke({
            "agent_name": agent.name,
            "agent_skills": ", ".join(agent.skills),
            "current_load": agent.current_load,
            "max_capacity": agent.max_capacity,
            "performance": sum(agent.performance_history) / len(agent.performance_history) if agent.performance_history else "No history",
            "task_description": task.description,
            "required_skills": ", ".join(task.required_skills),
            "priority": task.priority
        })
        
        try:
            score = float(response.strip().split()[0])
            
            # Adjust for current load
            load_factor = (agent.max_capacity - agent.current_load) / agent.max_capacity
            adjusted_score = score * load_factor
            
            return max(0.0, min(100.0, adjusted_score))
        except:
            return 0.0
    
    def allocate_tasks(self):
        """Allocate tasks to agents using optimal matching."""
        print("\n=== Task Allocation Process ===\n")
        
        # Sort tasks by priority
        sorted_tasks = sorted(self.task_queue, key=lambda t: t.priority, reverse=True)
        
        for task in sorted_tasks:
            print(f"Allocating: {task.description}")
            
            # Analyze task
            analysis = self.analyze_task_requirements(task)
            print(f"  Task Analysis: {analysis['analysis'][:100]}...")
            
            # Score all agents
            agent_scores = []
            for agent in self.agents:
                score = self.calculate_agent_score(agent, task)
                agent_scores.append((agent, score))
                print(f"  {agent.name}: {score:.1f}")
            
            # Select best agent
            if agent_scores:
                best_agent, best_score = max(agent_scores, key=lambda x: x[1])
                
                if best_score > 0:
                    self.assignments[task.id] = best_agent.id
                    best_agent.current_load += task.estimated_duration
                    print(f"  ✓ Assigned to: {best_agent.name} (Score: {best_score:.1f})")
                else:
                    print(f"  ✗ No suitable agent found")
            print()
        
        self.task_queue.clear()
    
    def get_schedule_summary(self) -> str:
        """Generate schedule summary."""
        summary_prompt = ChatPromptTemplate.from_messages([
            ("system", "Summarize this task allocation and schedule."),
            ("user", """Assignments:
{assignments}

Agent Loads:
{agent_loads}

Provide:
1. Overall load distribution
2. Potential bottlenecks
3. Optimization suggestions""")
        ])
        
        assignments_str = "\n".join([
            f"Task {task_id} -> Agent {agent_id}"
            for task_id, agent_id in self.assignments.items()
        ])
        
        agent_loads_str = "\n".join([
            f"{agent.name}: {agent.current_load}/{agent.max_capacity} ({agent.current_load/agent.max_capacity*100:.1f}%)"
            for agent in self.agents
        ])
        
        chain = summary_prompt | self.llm | StrOutputParser()
        return chain.invoke({
            "assignments": assignments_str,
            "agent_loads": agent_loads_str
        })
    
    def rebalance_load(self):
        """Rebalance load across agents if needed."""
        avg_load = sum(a.current_load for a in self.agents) / len(self.agents)
        overloaded = [a for a in self.agents if a.current_load > avg_load * 1.5]
        underloaded = [a for a in self.agents if a.current_load < avg_load * 0.5]
        
        if overloaded and underloaded:
            print("\n=== Load Rebalancing ===")
            print(f"Overloaded agents: {[a.name for a in overloaded]}")
            print(f"Underloaded agents: {[a.name for a in underloaded]}")
            print("Recommending task redistribution...")


def demonstrate_task_allocation():
    """Demonstrate task allocation and scheduling pattern."""
    print("=== Task Allocation & Scheduling Pattern ===\n")
    
    scheduler = TaskAllocationScheduler()
    
    # Register agents
    print("1. Registering Agents")
    print("-" * 50)
    
    scheduler.register_agent(Agent(
        id="agent1",
        name="Alice",
        skills=["python", "data_analysis", "machine_learning"],
        max_capacity=100,
        performance_history=[0.85, 0.90, 0.88]
    ))
    
    scheduler.register_agent(Agent(
        id="agent2",
        name="Bob",
        skills=["javascript", "frontend", "ui_design"],
        max_capacity=100,
        performance_history=[0.92, 0.89, 0.91]
    ))
    
    scheduler.register_agent(Agent(
        id="agent3",
        name="Charlie",
        skills=["devops", "cloud", "kubernetes"],
        max_capacity=100,
        performance_history=[0.88, 0.87, 0.90]
    ))
    
    scheduler.register_agent(Agent(
        id="agent4",
        name="Diana",
        skills=["python", "backend", "database"],
        max_capacity=100,
        performance_history=[0.90, 0.93, 0.91]
    ))
    
    print()
    
    # Add tasks
    print("2. Adding Tasks to Queue")
    print("-" * 50)
    
    scheduler.add_task(Task(
        id="task1",
        description="Build machine learning model for customer churn prediction",
        required_skills=["python", "machine_learning", "data_analysis"],
        priority=9,
        estimated_duration=40
    ))
    
    scheduler.add_task(Task(
        id="task2",
        description="Design and implement responsive dashboard UI",
        required_skills=["javascript", "frontend", "ui_design"],
        priority=8,
        estimated_duration=30
    ))
    
    scheduler.add_task(Task(
        id="task3",
        description="Set up CI/CD pipeline with Kubernetes deployment",
        required_skills=["devops", "cloud", "kubernetes"],
        priority=7,
        estimated_duration=35
    ))
    
    scheduler.add_task(Task(
        id="task4",
        description="Optimize database queries and add indexes",
        required_skills=["database", "backend", "python"],
        priority=8,
        estimated_duration=25
    ))
    
    scheduler.add_task(Task(
        id="task5",
        description="Implement user authentication system",
        required_skills=["backend", "python", "security"],
        priority=9,
        estimated_duration=45
    ))
    
    print()
    
    # Allocate tasks
    print("3. Allocating Tasks to Agents")
    print("-" * 50)
    scheduler.allocate_tasks()
    
    # Schedule summary
    print("4. Schedule Summary")
    print("-" * 50)
    summary = scheduler.get_schedule_summary()
    print(summary)
    print()
    
    # Check for rebalancing needs
    print("5. Load Balancing Check")
    print("-" * 50)
    scheduler.rebalance_load()
    
    print("\n=== Summary ===")
    print(f"Total assignments: {len(scheduler.assignments)}")
    print("Task allocation demonstrated with:")
    print("- Skill-based matching")
    print("- Load-aware assignment")
    print("- Priority-based scheduling")
    print("- Performance history consideration")
    print("- Load balancing recommendations")


if __name__ == "__main__":
    demonstrate_task_allocation()

Agentic Design Pattern: Task Allocation Scheduling

Pattern 119: Task Allocation Scheduling – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Task Allocation Scheduling pattern implementation.

Pattern Description

Pattern 119: Task Allocation & Scheduling

Description: Assigns tasks to agents based on capabilities, load, and constraints using auction-based or optimal assignment algorithms.

Components: – Task queue – Agent capabilities – Allocation algorithm – Load balancing

Use Cases: – Multi-agent systems – Distributed computing – Resource optimization

LangChain Implementation: Uses LLM-based task analysis and agent matching with LangGraph for orchestration.

File Statistics

Total Lines: 342
Classes: 3
Functions: 1
Constants: 0

Module Documentation

"""
Pattern 119: Task Allocation & Scheduling

Description:
    Assigns tasks to agents based on capabilities, load, and constraints
    using auction-based or optimal assignment algorithms.

Components:
    - Task queue
    - Agent capabilities
    - Allocation algorithm
    - Load balancing

Use Cases:
    - Multi-agent systems
    - Distributed computing
    - Resource optimization

LangChain Implementation:
    Uses LLM-based task analysis and agent matching with LangGraph for orchestration.
"""

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

Import Statements

Standard Library Imports

import dataclasses.dataclass
import datetime.datetime
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: – dataclasses.dataclass: Simplified class definitions with @dataclass – datetime.datetime: Date and time handling – 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: `Task`

Lines 37-44

Purpose:

Represents a task to be allocated.

Class: `Agent`

Lines 48-59

Purpose:

Represents an agent with capabilities.

Methods:

`__post_init__(self)`

Lines 57-59

Purpose: Post Init operation.

Parameters:

Class: `TaskAllocationScheduler`

Lines 62-220

Purpose:

Schedules and allocates tasks to agents optimally.

Methods:

`init(self, model_name)`

Lines 65-69

Purpose: Initialize the class instance with required parameters.

Parameters: – model_name: Language model instance

`register_agent(self, agent)`

Lines 71-74

Purpose:

Register an agent in the system.

Parameters: – agent: Agent

`add_task(self, task)`

Lines 76-79

Purpose:

Add task to queue.

Parameters: – task: Task

`analyze_task_requirements(self, task)`

Lines 81-102

Purpose:

Analyze task requirements using LLM.

Parameters: – task: Task

Returns: <ast.Subscript object at 0x000001A06D765F50>

`calculate_agent_score(self, agent, task)`

Lines 104-141

Purpose:

Calculate how well an agent matches a task.

Parameters: – agent: Agent – task: Task

Returns: float

`allocate_tasks(self)`

Lines 143-176

Purpose:

Allocate tasks to agents using optimal matching.

Parameters:

`get_schedule_summary(self)`

Lines 178-208

Purpose:

Generate schedule summary.

Parameters:

Returns: str

`rebalance_load(self)`

Lines 210-220

Purpose:

Rebalance load across agents if needed.

Parameters:

Functions

`demonstrate_task_allocation()`

Lines 223-337

Purpose:

Demonstrate task allocation and scheduling pattern.

Key Concepts Summary

Pattern Overview

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

Core Components

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

When to Use

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

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

LangChain Components Used

ChatOpenAI: OpenAI model interface
Chains: Sequential component composition

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

Execution Flow

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

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

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

Related Patterns

Pattern 118
Pattern 120

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 Task Allocation Scheduling 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: 119_task_allocation_scheduling.py
Total Lines: 342
Pattern Category: AI Agent Design Pattern #119

Generated from actual implementation code
For educational and reference purposes

November 2, 2025

Agentic Design Pattern: Curiosity Driven Exploration


    """
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

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

When to Use

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

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

LangChain Components Used

ChatOpenAI: OpenAI model interface
Chains: Sequential component composition

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

Execution Flow

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

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

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

Related Patterns

Pattern 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

November 2, 2025

Agentic Design Pattern: Imitation Learning


    """
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

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

When to Use

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

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

LangChain Components Used

ChatOpenAI: OpenAI model interface
Chains: Sequential component composition

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

Execution Flow

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

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

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

Related Patterns

Pattern 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

November 2, 2025

Agentic Design Pattern: Multi Task Learning


    """
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

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

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

When to Use

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

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

LangChain Components Used

ChatOpenAI: OpenAI model interface
Chains: Sequential component composition

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

Execution Flow

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

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

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

Related Patterns

Pattern 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

November 2, 2025

Agentic Design Pattern: Transfer Learning


    """
Pattern 115: Transfer Learning Agent

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

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

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

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

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

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

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

load_dotenv()


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


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

Target task: {target_task}

Available source domains:
{domains_desc}

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

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

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

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

Transferable patterns:
{patterns_str}

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

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

Transferable patterns:
{patterns_str}

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


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

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

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

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

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

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


if __name__ == "__main__":
    demonstrate_transfer_learning()

Agentic Design Pattern: Transfer Learning

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

Overview

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

Pattern Description

Pattern 115: Transfer Learning Agent

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

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

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

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

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

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

File Statistics

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

Module Documentation

"""
Pattern 115: Transfer Learning Agent

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

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

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

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

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

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

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

Import Statements

Standard Library Imports

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

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

LangChain Imports

from langchain_openai.ChatOpenAI

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

Other Imports

from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

Classes

Class: `DomainKnowledge`

Lines 52-76

Purpose:

Represents knowledge from a source domain

Methods:

`init(self, domain_name)`

Lines 55-59

Purpose: Initialize the class instance with required parameters.

Parameters: – domain_name: Domain name

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

Lines 61-67

Purpose:

Add an example from this domain

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

`add_pattern(self, pattern)`

Lines 69-72

Purpose:

Add a learned pattern

Parameters: – pattern: Pattern

`add_concept(self, concept, description)`

Lines 74-76

Purpose:

Add a domain concept

Parameters: – concept: Concept – description: Description

Class: `TransferLearningAgent`

Lines 79-277

Purpose:

Agent that transfers knowledge between domains

Methods:

`init(self, model_name)`

Lines 82-85

Purpose: Initialize the class instance with required parameters.

Parameters: – model_name: Language model instance

`add_source_domain(self, domain)`

Lines 87-93

Purpose:

Add a source domain with learned knowledge

Parameters: – domain: Domain

`identify_relevant_source(self, target_task)`

Lines 95-124

Purpose:

Identify most relevant source domain for transfer

Parameters: – target_task: Target task

Returns: <ast.Subscript object at 0x000001A06D7FEF50>

`adapt_examples(self, source_domain, target_domain, n)`

Lines 126-164

Purpose:

Adapt source examples to target domain

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

Returns: <ast.Subscript object at 0x000001A06D72EDD0>

`transfer_and_respond(self, target_task, target_domain, query)`

Lines 166-249

Purpose:

Transfer knowledge and respond to query

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

Returns: <ast.Subscript object at 0x000001A06D79B310>

`compare_with_without_transfer(self, target_task, target_domain, query)`

Lines 251-277

Purpose:

Compare performance with and without transfer learning

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

Returns: <ast.Subscript object at 0x000001A06D7E43D0>

Functions

`demonstrate_transfer_learning()`

Lines 280-398

Purpose:

Demonstrate transfer learning pattern

Key Concepts Summary

Pattern Overview

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

Core Components

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

When to Use

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

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

LangChain Components Used

ChatOpenAI: OpenAI model interface
Chains: Sequential component composition

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

Execution Flow

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

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

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

Related Patterns

Pattern 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

November 2, 2025

Agentic Design Pattern: Online Learning


    """
Pattern 114: Online Learning

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

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

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

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

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

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

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

load_dotenv()


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

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

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


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

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

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

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

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

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


if __name__ == "__main__":
    demonstrate_online_learning()

Agentic Design Pattern: Online Learning

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

Overview

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

Pattern Description

Pattern 114: Online Learning

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

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

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

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

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

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

File Statistics

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

Module Documentation

"""
Pattern 114: Online Learning

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

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

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

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

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

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

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

Import Statements

Standard Library Imports

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

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

LangChain Imports

from langchain.memory.ConversationBufferMemory
from langchain_openai.ChatOpenAI

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

Other Imports

from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

Classes

Class: `ExampleBuffer`

Lines 56-135

Purpose:

Manages a buffer of learning examples

Methods:

`init(self, max_size)`

Lines 59-62

Purpose: Initialize the class instance with required parameters.

Parameters: – max_size: Max size

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

Lines 64-76

Purpose:

Add a new example to the buffer

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

`get_recent_examples(self, n)`

Lines 78-81

Purpose:

Get n most recent examples

Parameters: – n: N

Returns: <ast.Subscript object at 0x000001A06D88BE90>

`get_best_examples(self, n)`

Lines 83-97

Purpose:

Get n best-rated examples

Parameters: – n: N

Returns: <ast.Subscript object at 0x000001A06D765F50>

`detect_drift(self, window_size)`

Lines 99-113

Purpose:

Detect if performance is degrading (concept drift)

Parameters: – window_size: Window size

Returns: bool

`get_performance_trend(self)`

Lines 115-135

Purpose:

Get description of performance trend

Parameters:

Returns: str

Class: `OnlineLearningAgent`

Lines 138-254

Purpose:

Agent that learns continuously from interactions

Methods:

`init(self, model_name, domain)`

Lines 141-156

Purpose: Initialize the class instance with required parameters.

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

`create_dynamic_prompt(self, use_best)`

Lines 158-192

Purpose:

Create prompt with dynamic examples from learning buffer

Parameters: – use_best: Use best

Returns: ChatPromptTemplate

`respond(self, user_input)`

Lines 194-204

Purpose:

Generate response using current learned knowledge

Parameters: – user_input: Input data

Returns: str

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

Lines 206-221

Purpose:

Learn from user feedback on a response

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

`interact_and_learn(self, user_input, feedback_score)`

Lines 223-244

Purpose:

Complete interaction cycle with learning

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

Returns: <ast.Subscript object at 0x000001A06D16DBD0>

`get_learning_stats(self)`

Lines 246-254

Purpose:

Get statistics about learning progress

Parameters:

Returns: <ast.Subscript object at 0x000001A06E094F10>

Functions

`demonstrate_online_learning()`

Lines 257-380

Purpose:

Demonstrate online learning pattern

Key Concepts Summary

Pattern Overview

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

Core Components

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

When to Use

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

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

LangChain Components Used

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

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

Execution Flow

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

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

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

Related Patterns

Pattern 113
Pattern 115

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

Summary

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

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

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

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

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

Generated from actual implementation code
For educational and reference purposes

November 2, 2025

Agentic Design Pattern: Contextual Adaptation


    """
Pattern 113: Contextual Adaptation

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

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

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

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

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

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

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

load_dotenv()


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


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


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


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


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

{context_summary}

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

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

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

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


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

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

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

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

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


if __name__ == "__main__":
    demonstrate_contextual_adaptation()

Agentic Design Pattern: Contextual Adaptation

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

Overview

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

Pattern Description

Pattern 113: Contextual Adaptation

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

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

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

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

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

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

File Statistics

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

Module Documentation

"""
Pattern 113: Contextual Adaptation

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

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

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

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

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

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

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

Import Statements

Standard Library Imports

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

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

LangChain Imports

from langchain.memory.ConversationBufferMemory
from langchain_openai.ChatOpenAI

Other Imports

from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

Classes

Class: `UserExperience`

Lines 56-61

Inherits from: Enum

Purpose:

User experience levels

Class: `Formality`

Lines 64-69

Inherits from: Enum

Purpose:

Communication formality levels

Class: `TaskUrgency`

Lines 72-77

Inherits from: Enum

Purpose:

Task urgency levels

Class: `Context`

Lines 80-153

Purpose:

Represents current context

Methods:

`init(self)`

Lines 83-109

Purpose: Initialize the class instance with required parameters.

Parameters:

`to_dict(self)`

Lines 111-141

Purpose:

Convert context to dictionary

Parameters:

Returns: <ast.Subscript object at 0x000001A06D798F90>

`summarize(self)`

Lines 143-153

Purpose:

Generate human-readable context summary

Parameters:

Returns: str

Class: `ContextualAdaptationAgent`

Lines 156-308

Purpose:

Agent that adapts behavior based on context

Methods:

`init(self, model_name)`

Lines 159-197

Purpose: Initialize the class instance with required parameters.

Parameters: – model_name: Language model instance

`detect_context(self, query)`

Lines 199-247

Purpose:

Detect context from user query

Parameters: – query: Input query or question

Returns: <ast.Subscript object at 0x000001A06D125F50>

`update_context(self, updates)`

Lines 249-253

Purpose:

Manually update context

Parameters: – updates: Updates

`respond(self, query, auto_detect)`

Lines 255-283

Purpose:

Generate context-adapted response

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

Returns: <ast.Subscript object at 0x000001A06D7FF950>

`demonstrate_adaptation(self, query, contexts)`

Lines 285-308

Purpose:

Demonstrate how response changes with different contexts

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

Functions

`demonstrate_contextual_adaptation()`

Lines 311-427

Purpose:

Demonstrate contextual adaptation pattern

Key Concepts Summary

Pattern Overview

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

Core Components

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

When to Use

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

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

LangChain Components Used

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

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

Execution Flow

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

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

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

Related Patterns

Pattern 112
Pattern 114

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

Summary

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

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

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

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

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

Generated from actual implementation code
For educational and reference purposes

November 2, 2025

Agentic Design Pattern: Common Sense Reasoning


    """
Pattern 112: Common Sense Reasoning

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

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

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

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

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

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

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

load_dotenv()


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


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

Common sense facts:
{facts}

Question: {question}

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

Proposed conclusion: {conclusion}

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

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

Observation: {observation}

Common sense causal facts:
{facts_str}

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


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

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

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

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


if __name__ == "__main__":
    demonstrate_common_sense_reasoning()

Agentic Design Pattern: Common Sense Reasoning

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

Overview

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

Pattern Description

Pattern 112: Common Sense Reasoning

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

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

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

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

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

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

File Statistics

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

Module Documentation

"""
Pattern 112: Common Sense Reasoning

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

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

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

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

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

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

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

Import Statements

Standard Library Imports

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

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

LangChain Imports

from langchain.chains.LLMChain
from langchain_openai.ChatOpenAI

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

Other Imports

from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

Classes

Class: `CommonSenseKnowledgeBase`

Lines 54-121

Purpose:

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

Methods:

`init(self)`

Lines 57-102

Purpose: Initialize the class instance with required parameters.

Parameters:

`query(self, concept)`

Lines 104-106

Purpose:

Query knowledge base for common sense fact

Parameters: – concept: Concept

Returns: <ast.Subscript object at 0x000001A06D6156D0>

`get_category(self, category)`

Lines 108-111

Purpose:

Get all facts in a category

Parameters: – category: Category

Returns: <ast.Subscript object at 0x000001A06D6168D0>

`get_relevant_facts(self, context)`

Lines 113-121

Purpose:

Get facts relevant to context

Parameters: – context: Context

Returns: <ast.Subscript object at 0x000001A06D73A2D0>

Class: `CommonSenseReasoner`

Lines 124-285

Purpose:

Agent that applies common sense reasoning

Methods:

`init(self, model_name)`

Lines 127-172

Purpose: Initialize the class instance with required parameters.

Parameters: – model_name: Language model instance

`reason_with_common_sense(self, question)`

Lines 174-205

Purpose:

Apply common sense reasoning to answer a question

Parameters: – question: Question to answer

Returns: <ast.Subscript object at 0x000001A06D16E510>

`check_plausibility(self, conclusion, context)`

Lines 207-234

Purpose:

Check if a conclusion is plausible

Parameters: – conclusion: Conclusion – context: Context

Returns: <ast.Subscript object at 0x000001A06D824AD0>

`generate_assumptions(self, scenario)`

Lines 236-254

Purpose:

Generate reasonable assumptions for a scenario

Parameters: – scenario: Scenario

Returns: <ast.Subscript object at 0x000001A06D68C590>

`explain_with_common_sense(self, observation)`

Lines 256-285

Purpose:

Explain an observation using common sense

Parameters: – observation: Observation

Returns: <ast.Subscript object at 0x000001A06D888FD0>

Functions

`demonstrate_common_sense_reasoning()`

Lines 288-360

Purpose:

Demonstrate common sense reasoning patterns

Key Concepts Summary

Pattern Overview

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

Core Components

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

When to Use

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

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

LangChain Components Used

ChatOpenAI: OpenAI model interface
Chains: Sequential component composition

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

Execution Flow

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

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

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

Related Patterns

Pattern 111
Pattern 113

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

Summary

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

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

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

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

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

Generated from actual implementation code
For educational and reference purposes

November 2, 2025

Agentic Design Pattern: Situational Awareness


    """
Pattern 111: Situational Awareness

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

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

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

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

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

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

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

load_dotenv()


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


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

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

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

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

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

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

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

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

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


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


if __name__ == "__main__":
    demonstrate_situational_awareness()

Agentic Design Pattern: Situational Awareness

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

Overview

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

Pattern Description

Pattern 111: Situational Awareness

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

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

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

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

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

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

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

File Statistics

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

Module Documentation

"""
Pattern 111: Situational Awareness

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

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

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

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

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

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

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

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

Import Statements

Standard Library Imports

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

LangChain Imports

from langchain.memory.ConversationBufferMemory
from langchain_openai.ChatOpenAI

Other Imports

from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

Classes

Class: `SituationStatus`

Lines 63-69

Inherits from: Enum

Purpose:

Enumeration of situation statuses.

Class: `SituationalAwarenessAgent`

Lines 72-345

Purpose:

Agent that maintains comprehensive awareness of current situation.

Methods:

`init(self, agent_role)`

Lines 77-91

Purpose:

Initialize with role and context.

Parameters: – agent_role: Agent role

`perceive_environment(self, observations)`

Lines 93-152

Purpose:

Process environmental observations to update awareness.

Args:
    observations: Dictionary of observed facts

Returns:
    Updated situational model

Parameters: – observations: Observations

Returns: <ast.Subscript object at 0x000001A06D7FDFD0>

`assess_situation(self)`

Lines 154-223

Purpose:

Assess current situation for threats and opportunities.

Returns:
    Situation assessment with status and recommendations

Parameters:

Returns: <ast.Subscript object at 0x000001A06D72EE50>

`predict_changes(self, time_horizon)`

Lines 225-278

Purpose:

Predict likely changes to the situation.

Args:
    time_horizon: Time frame for prediction

Returns:
    Predictions about situation evolution

Parameters: – time_horizon: Time horizon

Returns: <ast.Subscript object at 0x000001A06D604250>

`context_appropriate_response(self, query)`

Lines 280-335

Purpose:

Generate response appropriate to current situational context.

Args:
    query: Query or request

Returns:
    Context-aware response

Parameters: – query: Input query or question

Returns: <ast.Subscript object at 0x000001A06D17C810>

`get_situation_summary(self)`

Lines 337-345

Purpose:

Get comprehensive summary of current situation.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D6F5AD0>

Functions

`demonstrate_situational_awareness()`

Lines 348-428

Purpose:

Demonstrate situational awareness capabilities.

Key Concepts Summary

Pattern Overview

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

Core Components

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

When to Use

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

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

LangChain Components Used

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

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

Execution Flow

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

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

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

Related Patterns

Pattern 110
Pattern 112

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

Summary

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

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

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

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

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

Generated from actual implementation code
For educational and reference purposes

November 2, 2025

Blog

Agentic Design Pattern: Workflow Orchestration

Agentic Design Pattern: Workflow Orchestration

Pattern 120: Workflow Orchestration – Line-by-Line Explanation

Overview

Pattern Description

File Statistics

Module Documentation

Import Statements

Standard Library Imports

LangChain Imports

Other Imports

Classes

Class: StepStatus

Class: WorkflowStep

Class: WorkflowOrchestrator

__init__(self, model_name)

add_step(self, step)

validate_workflow(self)

_has_circular_dependency(self, step_name, visited)

get_ready_steps(self)

execute_step(self, step)

execute_workflow(self)

generate_summary(self)

Functions

demonstrate_workflow_orchestration()

Key Concepts Summary

Pattern Overview

Core Components

When to Use

Best Practices Demonstrated

1. Environment Configuration

2. Error Handling

3. Prompt Engineering

4. Code Organization

5. Performance Optimization

LangChain Components Used

Execution Flow

Performance Considerations

Computational Cost

Optimization Strategies

Cost Estimates (Approximate)

Common Pitfalls to Avoid

1. Prompt Design Issues

2. Error Handling

3. Token Management

4. Security

5. Testing

6. Cost Management

Further Improvements

Production Enhancements

Feature Extensions

Code Quality

Related Patterns

Summary

Agentic Design Pattern: Task Allocation Scheduling

Agentic Design Pattern: Task Allocation Scheduling

Pattern 119: Task Allocation Scheduling – Line-by-Line Explanation

Overview

Pattern Description

File Statistics

Module Documentation

Import Statements

Standard Library Imports

LangChain Imports

Other Imports

Classes

Class: Task

Class: Agent

__post_init__(self)

Class: TaskAllocationScheduler

__init__(self, model_name)

register_agent(self, agent)

add_task(self, task)

analyze_task_requirements(self, task)

calculate_agent_score(self, agent, task)

allocate_tasks(self)

get_schedule_summary(self)

rebalance_load(self)

Functions

Class: `StepStatus`

Class: `WorkflowStep`

Class: `WorkflowOrchestrator`

`init(self, model_name)`

`add_step(self, step)`

`validate_workflow(self)`

`_has_circular_dependency(self, step_name, visited)`

`get_ready_steps(self)`

`execute_step(self, step)`

`execute_workflow(self)`

`generate_summary(self)`

`demonstrate_workflow_orchestration()`

Class: `Task`

Class: `Agent`

`__post_init__(self)`

Class: `TaskAllocationScheduler`

`init(self, model_name)`

`register_agent(self, agent)`

`add_task(self, task)`

`analyze_task_requirements(self, task)`

`calculate_agent_score(self, agent, task)`

`allocate_tasks(self)`

`get_schedule_summary(self)`

`rebalance_load(self)`

`demonstrate_task_allocation()`

Class: `CuriosityDrivenAgent`

`init(self, model_name)`

`assess_novelty(self, topic)`

`generate_curious_questions(self, context)`

`calculate_information_gain(self, potential_exploration)`

`explore(self, topic)`

`choose_next_exploration(self, candidates)`

`get_exploration_summary(self)`

`demonstrate_curiosity_driven_exploration()`