"""
Pattern 061: Diffusion-Based Planning

Description:
    Diffusion-Based Planning uses diffusion models (inspired by diffusion processes
    in physics) for trajectory planning and action sequence generation. This approach
    can handle multimodal distributions and generate diverse, high-quality plans by
    iteratively denoising random noise into coherent action sequences.

Components:
    1. Diffusion Process: Forward (add noise) and reverse (denoise)
    2. Action Sequence Generator: Creates trajectory proposals
    3. Denoising Steps: Iterative refinement of plans
    4. Scoring Function: Evaluates plan quality
    5. Sampling Strategy: Controls exploration-exploitation

Use Cases:
    - Robotics motion planning
    - Game AI pathfinding
    - Multi-step action planning
    - Creative sequence generation
    - Trajectory optimization
    - Scenario planning

LangChain Implementation:
    Implements diffusion-inspired planning using LLM-based iterative refinement,
    noise injection, and progressive denoising for plan generation.
"""

import os
import time
from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import random
import math

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 NoiseLevel(Enum):
    """Noise levels for diffusion process"""
    HIGH = "high"  # Initial high noise
    MEDIUM = "medium"  # Mid-process
    LOW = "low"  # Near convergence
    MINIMAL = "minimal"  # Final refinement


class SamplingStrategy(Enum):
    """Strategies for plan sampling"""
    DETERMINISTIC = "deterministic"  # Single best plan
    STOCHASTIC = "stochastic"  # Sample from distribution
    BEAM_SEARCH = "beam_search"  # Keep top-k candidates
    DIVERSE = "diverse"  # Maximize diversity


@dataclass
class ActionStep:
    """Single action in a sequence"""
    step_id: int
    action: str
    parameters: Dict[str, Any] = field(default_factory=dict)
    confidence: float = 1.0
    
    def add_noise(self, noise_level: float) -> 'ActionStep':
        """Add noise to action for diffusion"""
        # In real implementation, this would modify action parameters
        # Here we adjust confidence as proxy for noise
        noisy_confidence = max(0.1, self.confidence * (1 - noise_level * 0.5))
        return ActionStep(
            step_id=self.step_id,
            action=self.action,
            parameters=self.parameters.copy(),
            confidence=noisy_confidence
        )


@dataclass
class Plan:
    """Complete action plan"""
    plan_id: str
    goal: str
    steps: List[ActionStep]
    noise_level: float = 0.0
    quality_score: float = 0.0
    diversity_score: float = 0.0
    
    @property
    def num_steps(self) -> int:
        return len(self.steps)
    
    def to_text(self) -> str:
        """Convert plan to text representation"""
        text = f"Goal: {self.goal}\n\nSteps:\n"
        for step in self.steps:
            text += f"{step.step_id}. {step.action}\n"
        return text


@dataclass
class DiffusionResult:
    """Result from diffusion planning"""
    goal: str
    final_plans: List[Plan]
    diffusion_steps: int
    best_plan: Plan
    diversity_score: float
    total_time_ms: float
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "goal": self.goal[:80] + "...",
            "num_plans": len(self.final_plans),
            "best_plan_steps": self.best_plan.num_steps,
            "best_quality": f"{self.best_plan.quality_score:.2f}",
            "diversity": f"{self.diversity_score:.2f}",
            "diffusion_steps": self.diffusion_steps,
            "total_time_ms": f"{self.total_time_ms:.1f}"
        }


class DiffusionPlanner:
    """
    Diffusion-based planning agent.
    
    Features:
    1. Iterative plan denoising
    2. Multi-modal plan generation
    3. Quality-guided refinement
    4. Diverse plan sampling
    5. Progressive improvement
    """
    
    def __init__(
        self,
        num_diffusion_steps: int = 5,
        num_candidates: int = 3,
        sampling_strategy: SamplingStrategy = SamplingStrategy.DIVERSE,
        temperature: float = 0.7
    ):
        self.num_diffusion_steps = num_diffusion_steps
        self.num_candidates = num_candidates
        self.sampling_strategy = sampling_strategy
        self.temperature = temperature
        
        # Planner LLM (higher temperature for creativity)
        self.planner = ChatOpenAI(
            model_name="gpt-3.5-turbo",
            temperature=temperature
        )
        
        # Refiner LLM (lower temperature for quality)
        self.refiner = ChatOpenAI(
            model_name="gpt-3.5-turbo",
            temperature=0.3
        )
        
        # Evaluator LLM
        self.evaluator = ChatOpenAI(
            model_name="gpt-3.5-turbo",
            temperature=0.2
        )
    
    def _generate_noisy_plan(
        self,
        goal: str,
        noise_level: float,
        context: Optional[str] = None
    ) -> Plan:
        """Generate initial noisy plan"""
        
        # Higher noise = more randomness = more creative/diverse
        creativity_instruction = ""
        if noise_level > 0.7:
            creativity_instruction = "Be highly creative and explore unusual approaches."
        elif noise_level > 0.4:
            creativity_instruction = "Consider multiple different approaches."
        else:
            creativity_instruction = "Focus on practical, direct solutions."
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"""You are generating an action plan to achieve a goal.

{creativity_instruction}

{context if context else ""}

Generate a step-by-step plan. Format:
1. [action step]
2. [action step]
..."""),
            ("user", "Goal: {goal}\n\nPlan:")
        ])
        
        chain = prompt | self.planner | StrOutputParser()
        plan_text = chain.invoke({"goal": goal})
        
        # Parse steps
        steps = []
        for i, line in enumerate(plan_text.split('\n'), 1):
            line = line.strip()
            if line and (line[0].isdigit() or line.startswith('-')):
                # Remove numbering
                action = line.lstrip('0123456789.-) ').strip()
                if action:
                    steps.append(ActionStep(
                        step_id=i,
                        action=action,
                        confidence=1.0 - noise_level * 0.3
                    ))
        
        return Plan(
            plan_id=f"plan_{random.randint(1000, 9999)}",
            goal=goal,
            steps=steps,
            noise_level=noise_level
        )
    
    def _denoise_plan(
        self,
        noisy_plan: Plan,
        target_noise_level: float,
        iteration: int
    ) -> Plan:
        """Denoise/refine a plan"""
        
        # Calculate denoising strength
        denoise_strength = noisy_plan.noise_level - target_noise_level
        
        if denoise_strength  float:
        """Evaluate plan quality"""
        
        evaluation_prompt = ChatPromptTemplate.from_messages([
            ("system", """Evaluate this action plan on a scale of 0.0 to 1.0.

Consider:
- Completeness: Does it achieve the goal?
- Feasibility: Is it realistic?
- Efficiency: Minimal unnecessary steps?
- Clarity: Clear and actionable?

Respond with just a number between 0.0 and 1.0"""),
            ("user", "{plan}")
        ])
        
        chain = evaluation_prompt | self.evaluator | StrOutputParser()
        
        try:
            score_str = chain.invoke({"plan": plan.to_text()})
            score = float(''.join(c for c in score_str if c.isdigit() or c == '.'))
            return max(0.0, min(1.0, score))
        except:
            # Fallback: score based on step count and confidence
            avg_confidence = sum(s.confidence for s in plan.steps) / len(plan.steps) if plan.steps else 0
            step_score = min(1.0, len(plan.steps) / 5.0)  # Prefer 5-ish steps
            return (avg_confidence + step_score) / 2
    
    def _calculate_diversity(self, plans: List[Plan]) -> float:
        """Calculate diversity among plans"""
        
        if len(plans)  0 else 0.0
    
    def plan(
        self,
        goal: str,
        constraints: Optional[List[str]] = None
    ) -> DiffusionResult:
        """Generate plans using diffusion process"""
        
        start_time = time.time()
        
        print(f"\n🌊 Starting diffusion-based planning...")
        print(f"   Goal: {goal}")
        print(f"   Diffusion steps: {self.num_diffusion_steps}")
        print(f"   Candidates: {self.num_candidates}")
        
        # Build context from constraints
        context = ""
        if constraints:
            context = "Constraints:\n" + "\n".join(f"- {c}" for c in constraints)
        
        # Generate initial noisy plans
        print(f"\n📍 Step 0: Generating {self.num_candidates} initial noisy plans...")
        
        current_plans = []
        for i in range(self.num_candidates):
            # Start with high noise for diversity
            initial_noise = 0.8 + random.random() * 0.2  # 0.8-1.0
            plan = self._generate_noisy_plan(goal, initial_noise, context)
            current_plans.append(plan)
            print(f"   Plan {i+1}: {len(plan.steps)} steps (noise: {initial_noise:.2f})")
        
        # Diffusion process: progressively denoise
        for step in range(1, self.num_diffusion_steps + 1):
            # Calculate target noise level (decreasing)
            target_noise = max(0.0, 1.0 - (step / self.num_diffusion_steps))
            
            print(f"\n📍 Step {step}/{self.num_diffusion_steps}: Denoising to {target_noise:.2f}")
            
            # Denoise all plans
            denoised_plans = []
            for plan in current_plans:
                denoised = self._denoise_plan(plan, target_noise, step)
                denoised_plans.append(denoised)
            
            # Evaluate plans
            for plan in denoised_plans:
                plan.quality_score = self._evaluate_plan(plan)
            
            # Select best plans based on strategy
            if self.sampling_strategy == SamplingStrategy.BEAM_SEARCH:
                # Keep top-k by quality
                denoised_plans.sort(key=lambda p: p.quality_score, reverse=True)
                current_plans = denoised_plans[:self.num_candidates]
                
            elif self.sampling_strategy == SamplingStrategy.DIVERSE:
                # Balance quality and diversity
                # Keep best plan + diverse alternatives
                denoised_plans.sort(key=lambda p: p.quality_score, reverse=True)
                current_plans = [denoised_plans[0]]  # Best plan
                
                # Add diverse plans
                for plan in denoised_plans[1:]:
                    if len(current_plans) >= self.num_candidates:
                        break
                    # Check if sufficiently different
                    is_diverse = True
                    for existing in current_plans:
                        # Simple diversity check
                        common_actions = sum(
                            1 for s1 in plan.steps
                            for s2 in existing.steps
                            if s1.action[:30] == s2.action[:30]
                        )
                        if common_actions > len(plan.steps) * 0.7:
                            is_diverse = False
                            break
                    
                    if is_diverse:
                        current_plans.append(plan)
            
            else:  # DETERMINISTIC or STOCHASTIC
                denoised_plans.sort(key=lambda p: p.quality_score, reverse=True)
                current_plans = denoised_plans[:self.num_candidates]
            
            # Show progress
            for i, plan in enumerate(current_plans, 1):
                print(f"   Plan {i}: Quality {plan.quality_score:.2f}, {len(plan.steps)} steps")
        
        # Final evaluation
        print(f"\n✅ Diffusion complete. Evaluating final plans...")
        
        for plan in current_plans:
            plan.quality_score = self._evaluate_plan(plan)
        
        # Calculate diversity
        diversity_score = self._calculate_diversity(current_plans)
        
        # Select best plan
        best_plan = max(current_plans, key=lambda p: p.quality_score)
        
        total_time_ms = (time.time() - start_time) * 1000
        
        return DiffusionResult(
            goal=goal,
            final_plans=current_plans,
            diffusion_steps=self.num_diffusion_steps,
            best_plan=best_plan,
            diversity_score=diversity_score,
            total_time_ms=total_time_ms
        )


def demonstrate_diffusion_planning():
    """Demonstrate Diffusion-Based Planning pattern"""
    
    print("=" * 80)
    print("PATTERN 061: DIFFUSION-BASED PLANNING DEMONSTRATION")
    print("=" * 80)
    print("\nIterative plan refinement through diffusion process\n")
    
    # Test 1: Simple goal with diverse sampling
    print("\n" + "=" * 80)
    print("TEST 1: Diverse Plan Generation")
    print("=" * 80)
    
    planner1 = DiffusionPlanner(
        num_diffusion_steps=4,
        num_candidates=3,
        sampling_strategy=SamplingStrategy.DIVERSE,
        temperature=0.8
    )
    
    goal1 = "Plan a successful product launch"
    
    result1 = planner1.plan(goal1)
    
    print(f"\n📊 Results:")
    print(f"   Plans Generated: {len(result1.final_plans)}")
    print(f"   Diversity Score: {result1.diversity_score:.2f}")
    print(f"   Total Time: {result1.total_time_ms:.1f}ms")
    
    print(f"\n🏆 Best Plan (Quality: {result1.best_plan.quality_score:.2f}):")
    for step in result1.best_plan.steps:
        print(f"   {step.step_id}. {step.action}")
    
    print(f"\n🔄 Alternative Plans:")
    for i, plan in enumerate(result1.final_plans[1:], 2):
        print(f"\n   Plan {i} (Quality: {plan.quality_score:.2f}):")
        for step in plan.steps[:3]:
            print(f"      {step.step_id}. {step.action}")
        if len(plan.steps) > 3:
            print(f"      ... ({len(plan.steps) - 3} more steps)")
    
    # Test 2: Constrained planning
    print("\n" + "=" * 80)
    print("TEST 2: Constrained Diffusion Planning")
    print("=" * 80)
    
    planner2 = DiffusionPlanner(
        num_diffusion_steps=5,
        num_candidates=2,
        sampling_strategy=SamplingStrategy.BEAM_SEARCH
    )
    
    goal2 = "Organize a team-building event"
    constraints2 = [
        "Budget limited to $1000",
        "Must be completed within 2 weeks",
        "Remote team across different time zones"
    ]
    
    print(f"\n🎯 Goal: {goal2}")
    print(f"📋 Constraints:")
    for constraint in constraints2:
        print(f"   - {constraint}")
    
    result2 = planner2.plan(goal2, constraints=constraints2)
    
    print(f"\n🏆 Optimized Plan (Quality: {result2.best_plan.quality_score:.2f}):")
    for step in result2.best_plan.steps:
        print(f"   {step.step_id}. {step.action}")
    
    # Test 3: Comparing diffusion steps
    print("\n" + "=" * 80)
    print("TEST 3: Impact of Diffusion Steps")
    print("=" * 80)
    
    goal3 = "Improve customer satisfaction"
    
    for num_steps in [2, 4, 6]:
        print(f"\n   Testing with {num_steps} diffusion steps...")
        
        planner = DiffusionPlanner(
            num_diffusion_steps=num_steps,
            num_candidates=2,
            sampling_strategy=SamplingStrategy.DIVERSE
        )
        
        result = planner.plan(goal3)
        
        print(f"      Best Quality: {result.best_plan.quality_score:.2f}")
        print(f"      Diversity: {result.diversity_score:.2f}")
        print(f"      Time: {result.total_time_ms:.1f}ms")
        print(f"      Steps in plan: {result.best_plan.num_steps}")
    
    # Summary
    print("\n" + "=" * 80)
    print("DIFFUSION-BASED PLANNING PATTERN SUMMARY")
    print("=" * 80)
    print("""
Key Benefits:
1. Multi-Modal Plans: Generates diverse alternative plans
2. Progressive Refinement: Iterative quality improvement
3. Exploration-Exploitation: Balances creativity and quality
4. Robustness: Multiple candidate solutions
5. Flexible Control: Adjustable diffusion process

Diffusion Process:
1. Forward Process: Add noise (generate diverse initial plans)
2. Reverse Process: Denoise (iteratively refine plans)
3. Multiple Steps: Progressive quality improvement
4. Sampling: Control exploration vs exploitation

Components:
- Noise Injection: Creates initial diversity
- Denoising: Iterative refinement
- Quality Scoring: Evaluates plan effectiveness
- Diversity Measurement: Ensures variety
- Selection Strategy: Chooses best candidates

Sampling Strategies:
1. Deterministic: Single best path
   - Highest quality guarantee
   - No diversity

2. Stochastic: Sample from distribution
   - Controlled randomness
   - Balanced exploration

3. Beam Search: Keep top-k candidates
   - Quality-focused
   - Limited diversity

4. Diverse: Maximize variety
   - Multiple alternatives
   - Creative solutions

Noise Schedules:
- High Noise (1.0 → 0.7): Initial creativity
- Medium Noise (0.7 → 0.4): Refinement begins
- Low Noise (0.4 → 0.1): Convergence
- Minimal Noise (0.1 → 0.0): Final polish

Use Cases:
- Robotics: Motion planning, trajectory optimization
- Game AI: Pathfinding, strategy generation
- Planning: Project planning, resource allocation
- Creative Tasks: Story generation, design
- Optimization: Multi-objective planning
- Scenario Analysis: Alternative futures

Best Practices:
1. More steps = higher quality (but slower)
2. More candidates = more diversity
3. Appropriate noise schedule
4. Quality-diversity balance
5. Constraint incorporation
6. Iterative evaluation
7. Early stopping when converged

Production Considerations:
- Computational cost (multiple LLM calls)
- Caching intermediate results
- Parallel plan generation
- Quality thresholds
- Timeout handling
- Plan validation
- User preference learning

Comparison with Related Patterns:
- vs. Tree-of-Thoughts: Continuous vs discrete
- vs. Beam Search: Diffusion vs greedy
- vs. Monte Carlo: Systematic vs random
- vs. Evolutionary: Denoising vs mutation

Advanced Techniques:
1. Conditional Diffusion: Constraint-aware
2. Guided Diffusion: Quality-guided sampling
3. Classifier-Free: Balance guidance
4. Latent Diffusion: Work in abstract space
5. Cascaded: Multi-resolution planning

Research Inspiration:
- Denoising Diffusion Probabilistic Models (DDPM)
- Score-based Generative Models
- Diffusion for Planning (Diffuser)
- Planning with Diffusion for Flexible Behavior

The Diffusion-Based Planning pattern provides a powerful framework
for generating diverse, high-quality plans through iterative
refinement inspired by diffusion processes in physics.
""")


if __name__ == "__main__":
    demonstrate_diffusion_planning()

    

Agentic Design Pattern: Diffusion Planning

Pattern 061: Diffusion Planning – Line-by-Line Explanation

Overview

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

---

Pattern Description

Pattern 061: Diffusion-Based Planning

Description: Diffusion-Based Planning uses diffusion models (inspired by diffusion processes in physics) for trajectory planning and action sequence generation. This approach can handle multimodal distributions and generate diverse, high-quality plans by iteratively denoising random noise into coherent action sequences.

Components: 1. Diffusion Process: Forward (add noise) and reverse (denoise) 2. Action Sequence Generator: Creates trajectory proposals 3. Denoising Steps: Iterative refinement of plans 4. Scoring Function: Evaluates plan quality 5. Sampling Strategy: Controls exploration-exploitation

Use Cases: – Robotics motion planning – Game AI pathfinding – Multi-step action planning – Creative sequence generation – Trajectory optimization – Scenario planning

LangChain Implementation: Implements diffusion-inspired planning using LLM-based iterative refinement, noise injection, and progressive denoising for plan generation.

---

File Statistics

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

---

Module Documentation

"""
Pattern 061: Diffusion-Based Planning

Description:
    Diffusion-Based Planning uses diffusion models (inspired by diffusion processes
    in physics) for trajectory planning and action sequence generation. This approach
    can handle multimodal distributions and generate diverse, high-quality plans by
    iteratively denoising random noise into coherent action sequences.

Components:
    1. Diffusion Process: Forward (add noise) and reverse (denoise)
    2. Action Sequence Generator: Creates trajectory proposals
    3. Denoising Steps: Iterative refinement of plans
    4. Scoring Function: Evaluates plan quality
    5. Sampling Strategy: Controls exploration-exploitation

Use Cases:
    - Robotics motion planning
    - Game AI pathfinding
    - Multi-step action planning
    - Creative sequence generation
    - Trajectory optimization
    - Scenario planning

LangChain Implementation:
    Implements diffusion-inspired planning using LLM-based iterative refinement,
    noise injection, and progressive denoising for plan generation.
"""

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

---

Import Statements

Standard Library Imports

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

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

LangChain Imports

from langchain_openai.ChatOpenAI

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

Other Imports

from dotenv.load_dotenv
from math

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: NoiseLevel

Lines 47-52

Inherits from: Enum

Purpose:

Noise levels for diffusion process

---

Class: SamplingStrategy

Lines 55-60

Inherits from: Enum

Purpose:

Strategies for plan sampling

---

Class: ActionStep

Lines 64-81

Purpose:

Single action in a sequence

Methods:

add_noise(self, noise_level)

Lines 71-81

Purpose:

Add noise to action for diffusion

Parameters: – noise_level: Noise level

Returns: ActionStep

---

Class: Plan

Lines 85-103

Purpose:

Complete action plan

Methods:

num_steps(self)

Lines 95-96

Decorators: property

Purpose: Num Steps operation.

Parameters:

Returns: int

to_text(self)

Lines 98-103

Purpose:

Convert plan to text representation

Parameters:

Returns: str

---

Class: DiffusionResult

Lines 107-125

Purpose:

Result from diffusion planning

Methods:

to_dict(self)

Lines 116-125

Purpose: To Dict operation.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D79BDD0>

---

Class: DiffusionPlanner

Lines 128-440

Purpose:

Diffusion-based planning agent.

Features:
1. Iterative plan denoising
2. Multi-modal plan generation
3. Quality-guided refinement
4. Diverse plan sampling
5. Progressive improvement

Methods:

__init__(self, num_diffusion_steps, num_candidates, sampling_strategy, temperature)

Lines 140-168

Purpose: Initialize the class instance with required parameters.

Parameters: – num_diffusion_steps: Num diffusion steps – num_candidates: Num candidates – sampling_strategy: Sampling strategy – temperature: Sampling temperature (0-1)

_generate_noisy_plan(self, goal, noise_level, context)

Lines 170-223

Purpose:

Generate initial noisy plan

Parameters: – goal: Goal – noise_level: Noise level – context: Context

Returns: Plan

_denoise_plan(self, noisy_plan, target_noise_level, iteration)

Lines 225-286

Purpose:

Denoise/refine a plan

Parameters: – noisy_plan: Noisy plan – target_noise_level: Target noise level – iteration: Iteration

Returns: Plan

_evaluate_plan(self, plan)

Lines 288-314

Purpose:

Evaluate plan quality

Parameters: – plan: Plan

Returns: float

_calculate_diversity(self, plans)

Lines 316-330

Purpose:

Calculate diversity among plans

Parameters: – plans: Plans

Returns: float

plan(self, goal, constraints)

Lines 332-440

Purpose:

Generate plans using diffusion process

Parameters: – goal: Goal – constraints: Constraints

Returns: DiffusionResult

---

Functions

demonstrate_diffusion_planning()

Lines 443-632

Purpose:

Demonstrate Diffusion-Based Planning pattern

---

Key Concepts Summary

Pattern Overview

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

Core Components

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

When to Use

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

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

---

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

---

LangChain Components Used

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

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

---

Execution Flow

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

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

---

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

---

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

---

Related Patterns

• Pattern 060
• Pattern 062

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

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

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

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

---

File: 061_diffusion_planning.py
Total Lines: 637
Pattern Category: AI Agent Design Pattern #061

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 060: Curriculum Learning

Description:
    Curriculum Learning structures the learning process by presenting training
    examples in a meaningful order, from simple to complex. Like human education,
    this progressive approach helps agents build foundational skills before
    tackling harder problems, leading to faster learning and better performance.

Components:
    1. Difficulty Scorer: Evaluates task complexity
    2. Curriculum Designer: Orders tasks by difficulty
    3. Pacing Strategy: Determines progression speed
    4. Performance Monitor: Tracks learning progress
    5. Adaptive Scheduler: Adjusts curriculum based on performance

Use Cases:
    - Training complex reasoning agents
    - Gradual skill acquisition
    - Hierarchical task learning
    - Educational tutoring systems
    - Reinforcement learning with sparse rewards
    - Knowledge scaffolding

LangChain Implementation:
    Implements curriculum learning by ordering prompts and examples from
    simple to complex, adapting progression based on agent performance.
"""

import os
import time
from typing import Dict, Any, List, Optional, Tuple, Callable
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import statistics
import random

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 DifficultyLevel(Enum):
    """Difficulty levels for curriculum"""
    TRIVIAL = 1
    EASY = 2
    MEDIUM = 3
    HARD = 4
    EXPERT = 5


class PacingStrategy(Enum):
    """Strategies for curriculum pacing"""
    FIXED = "fixed"  # Predetermined progression
    ADAPTIVE = "adaptive"  # Based on performance
    MIXED = "mixed"  # Combine easy and hard
    SPIRAL = "spiral"  # Revisit with increasing complexity
    COMPETENCY_BASED = "competency_based"  # Advance when mastered


@dataclass
class LearningExample:
    """A single learning example"""
    example_id: str
    task_type: str
    input_text: str
    expected_output: str
    difficulty: DifficultyLevel
    prerequisites: List[str] = field(default_factory=list)
    concepts: List[str] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    
    def is_prerequisite_met(self, mastered_examples: Set[str]) -> bool:
        """Check if prerequisites are satisfied"""
        return all(prereq in mastered_examples for prereq in self.prerequisites)


@dataclass
class CurriculumStage:
    """A stage in the curriculum"""
    stage_id: int
    stage_name: str
    difficulty_range: Tuple[DifficultyLevel, DifficultyLevel]
    examples: List[LearningExample]
    mastery_threshold: float = 0.8
    
    @property
    def num_examples(self) -> int:
        return len(self.examples)


@dataclass
class PerformanceRecord:
    """Record of performance on an example"""
    example_id: str
    attempt_number: int
    correct: bool
    response: str
    timestamp: datetime
    time_taken_ms: float


@dataclass
class CurriculumResult:
    """Result from curriculum learning"""
    stages_completed: int
    total_examples: int
    examples_mastered: int
    overall_accuracy: float
    learning_curve: List[Tuple[int, float]]  # (example_num, accuracy)
    time_per_stage: List[float]
    final_performance: float
    total_time_ms: float
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "stages_completed": self.stages_completed,
            "examples_mastered": f"{self.examples_mastered}/{self.total_examples}",
            "overall_accuracy": f"{self.overall_accuracy:.1%}",
            "final_performance": f"{self.final_performance:.1%}",
            "total_time_ms": f"{self.total_time_ms:.1f}"
        }


class CurriculumLearner:
    """
    Agent that learns through curriculum progression.
    
    Features:
    1. Difficulty-ordered example presentation
    2. Performance-based pacing
    3. Mastery tracking
    4. Adaptive progression
    5. Skill scaffolding
    """
    
    def __init__(
        self,
        pacing_strategy: PacingStrategy = PacingStrategy.ADAPTIVE,
        mastery_threshold: float = 0.8,
        temperature: float = 0.7
    ):
        self.pacing_strategy = pacing_strategy
        self.mastery_threshold = mastery_threshold
        self.temperature = temperature
        
        # Learner LLM
        self.learner = ChatOpenAI(
            model_name="gpt-3.5-turbo",
            temperature=temperature
        )
        
        # Difficulty assessor
        self.assessor = ChatOpenAI(
            model_name="gpt-3.5-turbo",
            temperature=0.2
        )
        
        # Learning state
        self.mastered_examples: Set[str] = set()
        self.performance_history: List[PerformanceRecord] = []
        self.current_stage: int = 0
        
        # Learning statistics
        self.learning_curve: List[Tuple[int, float]] = []
    
    def _assess_difficulty(self, example: LearningExample) -> DifficultyLevel:
        """Assess difficulty of an example"""
        
        # If already labeled, use that
        if example.difficulty != DifficultyLevel.MEDIUM:
            return example.difficulty
        
        # Otherwise, assess automatically
        assessment_prompt = ChatPromptTemplate.from_messages([
            ("system", """Rate the difficulty of this learning task on a scale:
1 - Trivial (very simple, obvious answer)
2 - Easy (straightforward, basic reasoning)
3 - Medium (requires some thought)
4 - Hard (complex, multiple steps)
5 - Expert (very challenging, advanced concepts)

Respond with just the number."""),
            ("user", "Task: {input}\nExpected: {output}\n\nDifficulty:")
        ])
        
        chain = assessment_prompt | self.assessor | StrOutputParser()
        
        try:
            rating = chain.invoke({
                "input": example.input_text,
                "output": example.expected_output
            })
            
            difficulty_num = int(''.join(c for c in rating if c.isdigit()))
            return DifficultyLevel(min(max(difficulty_num, 1), 5))
        except:
            return DifficultyLevel.MEDIUM
    
    def _create_curriculum(
        self,
        examples: List[LearningExample]
    ) -> List[CurriculumStage]:
        """Create curriculum stages from examples"""
        
        # Assess difficulties if needed
        for example in examples:
            if example.difficulty == DifficultyLevel.MEDIUM and not example.metadata.get("assessed"):
                example.difficulty = self._assess_difficulty(example)
                example.metadata["assessed"] = True
        
        # Group by difficulty
        stages_dict: Dict[DifficultyLevel, List[LearningExample]] = {}
        for example in examples:
            if example.difficulty not in stages_dict:
                stages_dict[example.difficulty] = []
            stages_dict[example.difficulty].append(example)
        
        # Create stages
        stages = []
        for difficulty in sorted(stages_dict.keys(), key=lambda x: x.value):
            stage = CurriculumStage(
                stage_id=len(stages),
                stage_name=f"Stage {len(stages)+1}: {difficulty.name}",
                difficulty_range=(difficulty, difficulty),
                examples=stages_dict[difficulty],
                mastery_threshold=self.mastery_threshold
            )
            stages.append(stage)
        
        return stages
    
    def _attempt_example(
        self,
        example: LearningExample,
        previous_examples: List[LearningExample]
    ) -> PerformanceRecord:
        """Attempt to solve an example"""
        
        start_time = time.time()
        
        # Build context from previous examples
        context = ""
        if previous_examples:
            context = "Previous examples:\n" + "\n".join([
                f"Input: {ex.input_text}\nOutput: {ex.expected_output}"
                for ex in previous_examples[-3:]  # Last 3 examples
            ]) + "\n\n"
        
        # Generate response
        prompt = ChatPromptTemplate.from_messages([
            ("system", """You are learning to solve tasks step by step.

{context}Now solve this task:"""),
            ("user", "{input}")
        ])
        
        chain = prompt | self.learner | StrOutputParser()
        response = chain.invoke({
            "context": context,
            "input": example.input_text
        })
        
        time_taken_ms = (time.time() - start_time) * 1000
        
        # Check correctness (simplified)
        response_normalized = response.strip().lower()
        expected_normalized = example.expected_output.strip().lower()
        
        # Flexible matching
        correct = (
            response_normalized == expected_normalized or
            expected_normalized in response_normalized or
            response_normalized in expected_normalized
        )
        
        # Count attempts
        attempt_number = len([
            r for r in self.performance_history
            if r.example_id == example.example_id
        ]) + 1
        
        record = PerformanceRecord(
            example_id=example.example_id,
            attempt_number=attempt_number,
            correct=correct,
            response=response,
            timestamp=datetime.now(),
            time_taken_ms=time_taken_ms
        )
        
        self.performance_history.append(record)
        
        return record
    
    def _calculate_stage_performance(
        self,
        stage: CurriculumStage
    ) -> float:
        """Calculate performance on current stage"""
        
        stage_example_ids = {ex.example_id for ex in stage.examples}
        
        recent_records = [
            r for r in self.performance_history[-20:]  # Last 20 attempts
            if r.example_id in stage_example_ids
        ]
        
        if not recent_records:
            return 0.0
        
        correct_count = sum(1 for r in recent_records if r.correct)
        return correct_count / len(recent_records)
    
    def _should_advance(
        self,
        stage: CurriculumStage,
        attempts_on_stage: int
    ) -> bool:
        """Determine if should advance to next stage"""
        
        if self.pacing_strategy == PacingStrategy.FIXED:
            # Fixed progression after N attempts
            return attempts_on_stage >= len(stage.examples) * 2
        
        elif self.pacing_strategy == PacingStrategy.ADAPTIVE:
            # Advance when performance exceeds threshold
            performance = self._calculate_stage_performance(stage)
            return performance >= stage.mastery_threshold and attempts_on_stage >= len(stage.examples)
        
        elif self.pacing_strategy == PacingStrategy.COMPETENCY_BASED:
            # Advance when all examples mastered
            stage_example_ids = {ex.example_id for ex in stage.examples}
            mastered_in_stage = stage_example_ids & self.mastered_examples
            return len(mastered_in_stage) >= len(stage_example_ids) * stage.mastery_threshold
        
        else:
            # Default: adaptive
            performance = self._calculate_stage_performance(stage)
            return performance >= stage.mastery_threshold
    
    def learn_curriculum(
        self,
        examples: List[LearningExample],
        max_attempts_per_stage: int = 20
    ) -> CurriculumResult:
        """Learn through curriculum progression"""
        
        start_time = time.time()
        
        print(f"\n📚 Starting curriculum learning with {len(examples)} examples...")
        
        # Create curriculum
        stages = self._create_curriculum(examples)
        
        print(f"\n🎯 Curriculum Structure:")
        for stage in stages:
            print(f"   {stage.stage_name}: {stage.num_examples} examples")
        
        # Learn through stages
        time_per_stage = []
        
        for stage_idx, stage in enumerate(stages):
            stage_start = time.time()
            
            print(f"\n📖 Stage {stage_idx + 1}/{len(stages)}: {stage.stage_name}")
            print(f"   Examples: {stage.num_examples}")
            print(f"   Mastery Threshold: {stage.mastery_threshold:.0%}")
            
            attempts_on_stage = 0
            previous_examples = []
            
            # Learn examples in this stage
            while attempts_on_stage  float:
        """Calculate accuracy over recent attempts"""
        
        if not self.performance_history:
            return 0.0
        
        recent = self.performance_history[-window:]
        correct = sum(1 for r in recent if r.correct)
        return correct / len(recent)


def demonstrate_curriculum_learning():
    """Demonstrate Curriculum Learning pattern"""
    
    print("=" * 80)
    print("PATTERN 060: CURRICULUM LEARNING DEMONSTRATION")
    print("=" * 80)
    print("\nProgressive learning from simple to complex\n")
    
    # Create learning examples with varying difficulty
    examples = [
        # Trivial examples
        LearningExample("ex1", "math", "What is 2 + 2?", "4", DifficultyLevel.TRIVIAL),
        LearningExample("ex2", "math", "What is 5 + 3?", "8", DifficultyLevel.TRIVIAL),
        LearningExample("ex3", "math", "What is 10 - 6?", "4", DifficultyLevel.TRIVIAL),
        
        # Easy examples
        LearningExample("ex4", "math", "What is 12 + 15?", "27", DifficultyLevel.EASY, prerequisites=["ex1"]),
        LearningExample("ex5", "math", "What is 25 - 8?", "17", DifficultyLevel.EASY, prerequisites=["ex3"]),
        LearningExample("ex6", "math", "What is 7 times 3?", "21", DifficultyLevel.EASY),
        
        # Medium examples
        LearningExample("ex7", "math", "What is 15 times 4?", "60", DifficultyLevel.MEDIUM, prerequisites=["ex6"]),
        LearningExample("ex8", "math", "What is 100 divided by 4?", "25", DifficultyLevel.MEDIUM),
        LearningExample("ex9", "math", "What is (5 + 3) times 2?", "16", DifficultyLevel.MEDIUM, prerequisites=["ex4", "ex6"]),
        
        # Hard examples
        LearningExample("ex10", "math", "What is 25% of 80?", "20", DifficultyLevel.HARD, prerequisites=["ex8"]),
        LearningExample("ex11", "math", "If x + 5 = 12, what is x?", "7", DifficultyLevel.HARD, prerequisites=["ex4"]),
    ]
    
    # Test 1: Adaptive curriculum
    print("\n" + "=" * 80)
    print("TEST 1: Adaptive Curriculum Learning")
    print("=" * 80)
    
    learner1 = CurriculumLearner(
        pacing_strategy=PacingStrategy.ADAPTIVE,
        mastery_threshold=0.7
    )
    
    result1 = learner1.learn_curriculum(examples, max_attempts_per_stage=15)
    
    print(f"\n📊 Learning Results:")
    print(f"   Stages Completed: {result1.stages_completed}")
    print(f"   Examples Mastered: {result1.examples_mastered}/{result1.total_examples}")
    print(f"   Overall Accuracy: {result1.overall_accuracy:.1%}")
    print(f"   Final Performance: {result1.final_performance:.1%}")
    print(f"   Total Time: {result1.total_time_ms/1000:.1f}s")
    
    print(f"\n📈 Learning Curve (samples):")
    for i in range(0, len(result1.learning_curve), max(1, len(result1.learning_curve)//5)):
        attempt_num, accuracy = result1.learning_curve[i]
        print(f"   Attempt {attempt_num}: {accuracy:.1%} accuracy")
    
    print(f"\n⏱️  Time per Stage:")
    for i, stage_time in enumerate(result1.time_per_stage, 1):
        print(f"   Stage {i}: {stage_time:.1f}s")
    
    # Test 2: Fixed pacing
    print("\n" + "=" * 80)
    print("TEST 2: Fixed Pacing Curriculum")
    print("=" * 80)
    
    learner2 = CurriculumLearner(
        pacing_strategy=PacingStrategy.FIXED,
        mastery_threshold=0.6
    )
    
    # Use subset for faster demo
    subset_examples = examples[:8]
    
    result2 = learner2.learn_curriculum(subset_examples, max_attempts_per_stage=10)
    
    print(f"\n📊 Fixed Pacing Results:")
    print(f"   Stages: {result2.stages_completed}")
    print(f"   Mastered: {result2.examples_mastered}/{result2.total_examples}")
    print(f"   Accuracy: {result2.overall_accuracy:.1%}")
    
    # Test 3: Learning curve comparison
    print("\n" + "=" * 80)
    print("TEST 3: Learning Progress Visualization")
    print("=" * 80)
    
    print(f"\n📈 Performance Improvement:")
    curve = result1.learning_curve
    
    # Show improvement
    if len(curve) >= 3:
        early_perf = statistics.mean([acc for _, acc in curve[:3]])
        mid_perf = statistics.mean([acc for _, acc in curve[len(curve)//2:len(curve)//2+3]])
        late_perf = statistics.mean([acc for _, acc in curve[-3:]])
        
        print(f"   Early (first 3): {early_perf:.1%}")
        print(f"   Middle: {mid_perf:.1%}")
        print(f"   Late (last 3): {late_perf:.1%}")
        print(f"   Improvement: +{(late_perf - early_perf):.1%}")
    
    # Summary
    print("\n" + "=" * 80)
    print("CURRICULUM LEARNING PATTERN SUMMARY")
    print("=" * 80)
    print("""
Key Benefits:
1. Faster Learning: Build on foundations progressively
2. Better Performance: Master basics before advanced topics
3. Reduced Frustration: Appropriate difficulty at each stage
4. Stable Training: Avoid overwhelming with complexity
5. Improved Retention: Scaffold knowledge systematically

Curriculum Design:
1. Difficulty Ordering: Simple → Complex
2. Prerequisite Structure: Build dependencies
3. Concept Progression: Foundational → Advanced
4. Skill Scaffolding: Support skill development
5. Spiral Learning: Revisit concepts with depth

Difficulty Assessment:
- Automatic: LLM-based complexity scoring
- Manual: Expert labeling
- Hybrid: Combine both approaches
- Dynamic: Adjust based on learner performance
- Multi-dimensional: Consider multiple factors

Pacing Strategies:
1. Fixed: Predetermined progression
   - Simple, predictable
   - May not fit all learners

2. Adaptive: Performance-based
   - Personalized pace
   - More complex logic

3. Mixed: Interleave difficulties
   - Maintain engagement
   - Prevent boredom

4. Spiral: Revisit with complexity
   - Reinforce learning
   - Deepen understanding

5. Competency-Based: Master before advance
   - Ensure readiness
   - May be slower

Curriculum Stages:
1. Foundation: Basic concepts
2. Development: Core skills
3. Integration: Combine concepts
4. Application: Real problems
5. Mastery: Advanced challenges

Performance Monitoring:
- Accuracy tracking
- Time per example
- Mastery identification
- Learning curve analysis
- Plateau detection

Advancement Criteria:
- Accuracy threshold (e.g., 80%)
- Minimum attempts per stage
- Mastery of N examples
- Time-based progression
- Adaptive combinations

Use Cases:
- Educational tutoring systems
- Agent skill development
- Reinforcement learning
- Language learning
- Technical training
- Complex task acquisition

Best Practices:
1. Clear difficulty progression
2. Prerequisite tracking
3. Performance monitoring
4. Adaptive pacing
5. Regular assessment
6. Sufficient practice
7. Gradual complexity increase

Production Considerations:
- Curriculum versioning
- Learner state persistence
- Performance analytics
- A/B testing curricula
- Difficulty calibration
- Prerequisite validation
- Scalability

Comparison with Related Patterns:
- vs. Random: Structured vs unordered
- vs. Hard Example Mining: Systematic vs selection
- vs. Meta-Learning: Progression vs strategy learning
- vs. Active Learning: Curriculum vs query strategy

Challenges:
1. Difficulty assessment accuracy
2. Optimal ordering
3. Individual differences
4. Curriculum generalization
5. Pacing calibration

Research Findings:
- 30-50% faster learning vs random
- Better final performance
- Improved generalization
- Reduced training instability
- Context-dependent effectiveness

The Curriculum Learning pattern enables efficient learning through
strategic ordering of examples from simple to complex, mirroring
effective human educational approaches.
""")


if __name__ == "__main__":
    demonstrate_curriculum_learning()

    

Agentic Design Pattern: Curriculum Learning

Pattern 060: Curriculum Learning – Line-by-Line Explanation

Overview

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

---

Pattern Description

Pattern 060: Curriculum Learning

Description: Curriculum Learning structures the learning process by presenting training examples in a meaningful order, from simple to complex. Like human education, this progressive approach helps agents build foundational skills before tackling harder problems, leading to faster learning and better performance.

Components: 1. Difficulty Scorer: Evaluates task complexity 2. Curriculum Designer: Orders tasks by difficulty 3. Pacing Strategy: Determines progression speed 4. Performance Monitor: Tracks learning progress 5. Adaptive Scheduler: Adjusts curriculum based on performance

Use Cases: – Training complex reasoning agents – Gradual skill acquisition – Hierarchical task learning – Educational tutoring systems – Reinforcement learning with sparse rewards – Knowledge scaffolding

LangChain Implementation: Implements curriculum learning by ordering prompts and examples from simple to complex, adapting progression based on agent performance.

---

File Statistics

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

---

Module Documentation

"""
Pattern 060: Curriculum Learning

Description:
    Curriculum Learning structures the learning process by presenting training
    examples in a meaningful order, from simple to complex. Like human education,
    this progressive approach helps agents build foundational skills before
    tackling harder problems, leading to faster learning and better performance.

Components:
    1. Difficulty Scorer: Evaluates task complexity
    2. Curriculum Designer: Orders tasks by difficulty
    3. Pacing Strategy: Determines progression speed
    4. Performance Monitor: Tracks learning progress
    5. Adaptive Scheduler: Adjusts curriculum based on performance

Use Cases:
    - Training complex reasoning agents
    - Gradual skill acquisition
    - Hierarchical task learning
    - Educational tutoring systems
    - Reinforcement learning with sparse rewards
    - Knowledge scaffolding

LangChain Implementation:
    Implements curriculum learning by ordering prompts and examples from
    simple to complex, adapting progression based on agent performance.
"""

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

---

Import Statements

Standard Library Imports

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

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

LangChain Imports

from langchain_openai.ChatOpenAI

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

Other Imports

from dotenv.load_dotenv
from statistics

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: DifficultyLevel

Lines 47-53

Inherits from: Enum

Purpose:

Difficulty levels for curriculum

---

Class: PacingStrategy

Lines 56-62

Inherits from: Enum

Purpose:

Strategies for curriculum pacing

---

Class: LearningExample

Lines 66-79

Purpose:

A single learning example

Methods:

is_prerequisite_met(self, mastered_examples)

Lines 77-79

Purpose:

Check if prerequisites are satisfied

Parameters: – mastered_examples: Mastered examples

Returns: bool

---

Class: CurriculumStage

Lines 83-93

Purpose:

A stage in the curriculum

Methods:

num_examples(self)

Lines 92-93

Decorators: property

Purpose: Num Examples operation.

Parameters:

Returns: int

---

Class: PerformanceRecord

Lines 97-104

Purpose:

Record of performance on an example

---

Class: CurriculumResult

Lines 108-126

Purpose:

Result from curriculum learning

Methods:

to_dict(self)

Lines 119-126

Purpose: To Dict operation.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D700050>

---

Class: CurriculumLearner

Lines 129-444

Purpose:

Agent that learns through curriculum progression.

Features:
1. Difficulty-ordered example presentation
2. Performance-based pacing
3. Mastery tracking
4. Adaptive progression
5. Skill scaffolding

Methods:

__init__(self, pacing_strategy, mastery_threshold, temperature)

Lines 141-169

Purpose: Initialize the class instance with required parameters.

Parameters: – pacing_strategy: Pacing strategy – mastery_threshold: Mastery threshold – temperature: Sampling temperature (0-1)

_assess_difficulty(self, example)

Lines 171-202

Purpose:

Assess difficulty of an example

Parameters: – example: Example

Returns: DifficultyLevel

_create_curriculum(self, examples)

Lines 204-235

Purpose:

Create curriculum stages from examples

Parameters: – examples: Examples

Returns: <ast.Subscript object at 0x000001A06D1683D0>

_attempt_example(self, example, previous_examples)

Lines 237-298

Purpose:

Attempt to solve an example

Parameters: – example: Example – previous_examples: Previous examples

Returns: PerformanceRecord

_calculate_stage_performance(self, stage)

Lines 300-317

Purpose:

Calculate performance on current stage

Parameters: – stage: Stage

Returns: float

_should_advance(self, stage, attempts_on_stage)

Lines 319-344

Purpose:

Determine if should advance to next stage

Parameters: – stage: Stage – attempts_on_stage: Attempts on stage

Returns: bool

learn_curriculum(self, examples, max_attempts_per_stage)

Lines 346-434

Purpose:

Learn through curriculum progression

Parameters: – examples: Examples – max_attempts_per_stage: Max attempts per stage

Returns: CurriculumResult

_calculate_recent_accuracy(self, window)

Lines 436-444

Purpose:

Calculate accuracy over recent attempts

Parameters: – window: Window

Returns: float

---

Functions

demonstrate_curriculum_learning()

Lines 447-661

Purpose:

Demonstrate Curriculum Learning pattern

---

Key Concepts Summary

Pattern Overview

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

Core Components

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

When to Use

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

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

---

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

---

LangChain Components Used

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

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

---

Execution Flow

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

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

---

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

---

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

---

Related Patterns

• Pattern 059
• Pattern 061

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 Curriculum 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: 060_curriculum_learning.py
Total Lines: 666
Pattern Category: AI Agent Design Pattern #060

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 059: Meta-Learning Agent

Description:
    Meta-learning (learning to learn) enables agents to improve their learning
    efficiency over time by learning general strategies that transfer across tasks.
    Rather than learning task-specific solutions, meta-learning agents learn how
    to adapt quickly to new tasks with minimal examples or experience.

Components:
    1. Task Distribution: Collection of related learning tasks
    2. Meta-Learner: Learns general learning strategies
    3. Base Learner: Applies strategies to specific tasks
    4. Adaptation Mechanism: Quick task-specific tuning
    5. Transfer Module: Knowledge transfer across tasks

Use Cases:
    - Few-shot learning scenarios
    - Rapid adaptation to new domains
    - Personalization with limited data
    - Multi-task learning systems
    - Continuous learning environments
    - Cross-domain knowledge transfer

LangChain Implementation:
    Implements meta-learning through prompt optimization, example selection,
    and strategy learning that transfers across tasks.
"""

import os
import time
from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import defaultdict
import statistics

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 MetaStrategy(Enum):
    """Meta-learning strategies"""
    MAML = "maml"  # Model-Agnostic Meta-Learning
    PROMPT_LEARNING = "prompt_learning"  # Learn optimal prompts
    EXAMPLE_SELECTION = "example_selection"  # Learn to select examples
    STRATEGY_LEARNING = "strategy_learning"  # Learn problem-solving strategies
    TRANSFER_LEARNING = "transfer_learning"  # Transfer knowledge across tasks


class TaskType(Enum):
    """Types of learning tasks"""
    CLASSIFICATION = "classification"
    GENERATION = "generation"
    REASONING = "reasoning"
    TRANSLATION = "translation"
    SUMMARIZATION = "summarization"


@dataclass
class Task:
    """A learning task"""
    task_id: str
    task_type: TaskType
    description: str
    examples: List[Tuple[str, str]]  # (input, output)
    test_cases: List[Tuple[str, str]]
    metadata: Dict[str, Any] = field(default_factory=dict)


@dataclass
class LearnedStrategy:
    """A learned problem-solving strategy"""
    strategy_id: str
    applicable_tasks: List[TaskType]
    prompt_template: str
    example_count: int
    success_rate: float
    usage_count: int = 0
    
    def is_applicable(self, task_type: TaskType) -> bool:
        return task_type in self.applicable_tasks


@dataclass
class AdaptationResult:
    """Result from adaptation to new task"""
    task_id: str
    strategy_used: str
    examples_used: int
    performance: float
    adaptation_time_ms: float
    predictions: List[str]


@dataclass
class MetaLearningResult:
    """Result from meta-learning process"""
    tasks_trained: int
    strategies_learned: List[LearnedStrategy]
    avg_adaptation_performance: float
    total_training_time_ms: float
    transfer_effectiveness: float
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "tasks_trained": self.tasks_trained,
            "strategies_learned": len(self.strategies_learned),
            "avg_performance": f"{self.avg_adaptation_performance:.2%}",
            "transfer_effectiveness": f"{self.transfer_effectiveness:.2%}",
            "training_time_ms": f"{self.total_training_time_ms:.1f}"
        }


class MetaLearningAgent:
    """
    Agent that learns to learn across tasks.
    
    Features:
    1. Strategy learning from multiple tasks
    2. Fast adaptation to new tasks
    3. Optimal example selection
    4. Prompt optimization
    5. Knowledge transfer
    """
    
    def __init__(
        self,
        meta_strategy: MetaStrategy = MetaStrategy.STRATEGY_LEARNING,
        temperature: float = 0.7
    ):
        self.meta_strategy = meta_strategy
        self.temperature = temperature
        
        # Meta-learner (learns strategies)
        self.meta_learner = ChatOpenAI(
            model_name="gpt-3.5-turbo",
            temperature=0.3
        )
        
        # Base learner (applies strategies)
        self.base_learner = ChatOpenAI(
            model_name="gpt-3.5-turbo",
            temperature=temperature
        )
        
        # Learned strategies
        self.strategies: List[LearnedStrategy] = []
        
        # Task performance history
        self.performance_history: Dict[str, List[float]] = defaultdict(list)
        
        # Experience buffer
        self.experience: List[Tuple[Task, float]] = []
    
    def _learn_strategy_from_tasks(
        self,
        tasks: List[Task]
    ) -> LearnedStrategy:
        """Learn a general strategy from multiple tasks"""
        
        # Analyze patterns across tasks
        task_descriptions = "\n".join([
            f"Task {i+1} ({task.task_type.value}): {task.description}\n"
            f"Example: {task.examples[0][0]} → {task.examples[0][1]}"
            for i, task in enumerate(tasks[:3])
        ])
        
        strategy_prompt = ChatPromptTemplate.from_messages([
            ("system", """Analyze these learning tasks and identify a general strategy
that would work across all of them.

Provide:
1. Strategy name
2. General approach
3. Key steps
4. When to use

Be concise but specific."""),
            ("user", "Tasks:\n{tasks}\n\nGeneral strategy:")
        ])
        
        chain = strategy_prompt | self.meta_learner | StrOutputParser()
        strategy_description = chain.invoke({"tasks": task_descriptions})
        
        # Create prompt template from strategy
        prompt_template = f"""Strategy: {strategy_description}

Examples:
{{examples}}

Now apply this strategy to: {{query}}

Response:"""
        
        # Determine applicable task types
        task_types = list(set(t.task_type for t in tasks))
        
        strategy = LearnedStrategy(
            strategy_id=f"strategy_{len(self.strategies)}",
            applicable_tasks=task_types,
            prompt_template=prompt_template,
            example_count=3,
            success_rate=0.0  # Will be updated with experience
        )
        
        return strategy
    
    def _select_examples(
        self,
        task: Task,
        k: int = 3
    ) -> List[Tuple[str, str]]:
        """Select most informative examples for task"""
        
        # Simple selection: use first k examples
        # In production: use diversity, difficulty, relevance
        return task.examples[:k]
    
    def _evaluate_performance(
        self,
        predictions: List[str],
        ground_truth: List[str]
    ) -> float:
        """Evaluate prediction performance"""
        
        if not predictions or not ground_truth:
            return 0.0
        
        # Simple accuracy (in production: use task-specific metrics)
        correct = sum(
            1 for pred, truth in zip(predictions, ground_truth)
            if pred.strip().lower() == truth.strip().lower()
        )
        
        return correct / len(ground_truth)
    
    def meta_train(
        self,
        tasks: List[Task],
        episodes_per_task: int = 2
    ) -> MetaLearningResult:
        """Meta-training: Learn strategies from multiple tasks"""
        
        start_time = time.time()
        
        print(f"\n🎓 Meta-Training on {len(tasks)} tasks...")
        
        # Group tasks by type
        tasks_by_type = defaultdict(list)
        for task in tasks:
            tasks_by_type[task.task_type].append(task)
        
        # Learn strategies for each task type
        for task_type, type_tasks in tasks_by_type.items():
            if len(type_tasks) >= 2:  # Need multiple tasks to learn strategy
                print(f"\n  Learning strategy for {task_type.value} tasks...")
                
                strategy = self._learn_strategy_from_tasks(type_tasks)
                
                # Evaluate strategy on held-out examples
                performances = []
                for task in type_tasks:
                    # Use strategy to predict on test cases
                    examples = self._select_examples(task, k=2)
                    examples_text = "\n".join([
                        f"Input: {inp}\nOutput: {out}"
                        for inp, out in examples
                    ])
                    
                    predictions = []
                    for test_input, _ in task.test_cases[:3]:
                        prompt = strategy.prompt_template.format(
                            examples=examples_text,
                            query=test_input
                        )
                        
                        pred = self.base_learner.invoke(prompt).content
                        predictions.append(pred)
                    
                    # Evaluate
                    ground_truth = [out for _, out in task.test_cases[:3]]
                    perf = self._evaluate_performance(predictions, ground_truth)
                    performances.append(perf)
                    
                    # Store experience
                    self.experience.append((task, perf))
                
                # Update strategy success rate
                strategy.success_rate = statistics.mean(performances)
                self.strategies.append(strategy)
                
                print(f"    Strategy learned with {strategy.success_rate:.1%} success rate")
        
        # Calculate metrics
        all_performances = [perf for _, perf in self.experience]
        avg_performance = statistics.mean(all_performances) if all_performances else 0.0
        
        # Transfer effectiveness: improvement over baseline
        # (simplified - in production, compare to zero-shot)
        transfer_effectiveness = min(avg_performance * 1.2, 1.0)
        
        total_time_ms = (time.time() - start_time) * 1000
        
        return MetaLearningResult(
            tasks_trained=len(tasks),
            strategies_learned=self.strategies.copy(),
            avg_adaptation_performance=avg_performance,
            total_training_time_ms=total_time_ms,
            transfer_effectiveness=transfer_effectiveness
        )
    
    def adapt(
        self,
        new_task: Task,
        support_examples: Optional[List[Tuple[str, str]]] = None
    ) -> AdaptationResult:
        """Adapt to new task using learned strategies"""
        
        start_time = time.time()
        
        # Select best strategy for this task type
        applicable_strategies = [
            s for s in self.strategies
            if s.is_applicable(new_task.task_type)
        ]
        
        if applicable_strategies:
            # Use best performing strategy
            strategy = max(applicable_strategies, key=lambda s: s.success_rate)
            strategy.usage_count += 1
        else:
            # Fallback: learn from task examples only
            strategy = LearnedStrategy(
                strategy_id="fallback",
                applicable_tasks=[new_task.task_type],
                prompt_template="Examples:\n{examples}\n\nQuery: {query}\nResponse:",
                example_count=3,
                success_rate=0.5
            )
        
        # Select examples
        if support_examples:
            examples = support_examples
        else:
            examples = self._select_examples(new_task, k=strategy.example_count)
        
        examples_text = "\n".join([
            f"Input: {inp}\nOutput: {out}"
            for inp, out in examples
        ])
        
        # Make predictions on test cases
        predictions = []
        for test_input, _ in new_task.test_cases:
            prompt = strategy.prompt_template.format(
                examples=examples_text,
                query=test_input
            )
            
            pred = self.base_learner.invoke(prompt).content
            predictions.append(pred)
        
        # Evaluate performance
        ground_truth = [out for _, out in new_task.test_cases]
        performance = self._evaluate_performance(predictions, ground_truth)
        
        # Update strategy success rate
        self.performance_history[strategy.strategy_id].append(performance)
        if self.performance_history[strategy.strategy_id]:
            strategy.success_rate = statistics.mean(
                self.performance_history[strategy.strategy_id]
            )
        
        adaptation_time_ms = (time.time() - start_time) * 1000
        
        return AdaptationResult(
            task_id=new_task.task_id,
            strategy_used=strategy.strategy_id,
            examples_used=len(examples),
            performance=performance,
            adaptation_time_ms=adaptation_time_ms,
            predictions=predictions
        )
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get meta-learning statistics"""
        
        return {
            "strategies_learned": len(self.strategies),
            "total_experience": len(self.experience),
            "strategy_details": [
                {
                    "id": s.strategy_id,
                    "task_types": [t.value for t in s.applicable_tasks],
                    "success_rate": f"{s.success_rate:.1%}",
                    "usage_count": s.usage_count
                }
                for s in self.strategies
            ]
        }


def demonstrate_meta_learning():
    """Demonstrate Meta-Learning Agent pattern"""
    
    print("=" * 80)
    print("PATTERN 059: META-LEARNING AGENT DEMONSTRATION")
    print("=" * 80)
    print("\nLearning to learn across tasks\n")
    
    # Create meta-learning agent
    agent = MetaLearningAgent(
        meta_strategy=MetaStrategy.STRATEGY_LEARNING,
        temperature=0.7
    )
    
    # Create training tasks
    training_tasks = [
        Task(
            task_id="sentiment_1",
            task_type=TaskType.CLASSIFICATION,
            description="Classify sentiment as positive or negative",
            examples=[
                ("This movie was amazing!", "positive"),
                ("I hated this book.", "negative"),
                ("Great product, highly recommend.", "positive"),
            ],
            test_cases=[
                ("Loved it!", "positive"),
                ("Terrible experience.", "negative"),
            ]
        ),
        Task(
            task_id="sentiment_2",
            task_type=TaskType.CLASSIFICATION,
            description="Determine if review is positive or negative",
            examples=[
                ("Best purchase ever!", "positive"),
                ("Waste of money.", "negative"),
                ("Exceeded expectations.", "positive"),
            ],
            test_cases=[
                ("Fantastic!", "positive"),
                ("Very disappointed.", "negative"),
            ]
        ),
        Task(
            task_id="translate_1",
            task_type=TaskType.TRANSLATION,
            description="Translate English to simple terms",
            examples=[
                ("utilize", "use"),
                ("commence", "begin"),
                ("terminate", "end"),
            ],
            test_cases=[
                ("facilitate", "help"),
                ("acquire", "get"),
            ]
        ),
        Task(
            task_id="translate_2",
            task_type=TaskType.TRANSLATION,
            description="Convert formal to informal",
            examples=[
                ("I am pleased to inform you", "I'm happy to tell you"),
                ("Please be advised", "Just so you know"),
                ("At your earliest convenience", "When you can"),
            ],
            test_cases=[
                ("It is recommended", "You should"),
                ("We regret to inform", "Sorry to say"),
            ]
        ),
    ]
    
    # Test 1: Meta-training
    print("\n" + "=" * 80)
    print("TEST 1: Meta-Training on Multiple Tasks")
    print("=" * 80)
    
    print(f"\n📚 Training Tasks:")
    for task in training_tasks:
        print(f"   - {task.task_id} ({task.task_type.value}): {task.description}")
    
    meta_result = agent.meta_train(training_tasks, episodes_per_task=1)
    
    print(f"\n📊 Meta-Training Results:")
    print(f"   Tasks Trained: {meta_result.tasks_trained}")
    print(f"   Strategies Learned: {len(meta_result.strategies_learned)}")
    print(f"   Avg Performance: {meta_result.avg_adaptation_performance:.1%}")
    print(f"   Transfer Effectiveness: {meta_result.transfer_effectiveness:.1%}")
    print(f"   Training Time: {meta_result.total_training_time_ms:.1f}ms")
    
    print(f"\n🧠 Learned Strategies:")
    for strategy in meta_result.strategies_learned:
        print(f"   {strategy.strategy_id}:")
        print(f"      Task Types: {[t.value for t in strategy.applicable_tasks]}")
        print(f"      Success Rate: {strategy.success_rate:.1%}")
    
    # Test 2: Adaptation to new task (few-shot)
    print("\n" + "=" * 80)
    print("TEST 2: Fast Adaptation to New Task (Few-Shot)")
    print("=" * 80)
    
    new_task = Task(
        task_id="sentiment_new",
        task_type=TaskType.CLASSIFICATION,
        description="Classify restaurant reviews",
        examples=[
            ("Food was delicious!", "positive"),
            ("Service was terrible.", "negative"),
        ],
        test_cases=[
            ("Best meal I've had!", "positive"),
            ("Worst restaurant ever.", "negative"),
            ("Amazing ambiance.", "positive"),
        ]
    )
    
    print(f"\n🆕 New Task: {new_task.description}")
    print(f"   Task Type: {new_task.task_type.value}")
    print(f"   Training Examples: {len(new_task.examples)}")
    print(f"   Test Cases: {len(new_task.test_cases)}")
    
    adapt_result = agent.adapt(new_task)
    
    print(f"\n⚡ Adaptation Results:")
    print(f"   Strategy Used: {adapt_result.strategy_used}")
    print(f"   Examples Used: {adapt_result.examples_used}")
    print(f"   Performance: {adapt_result.performance:.1%}")
    print(f"   Adaptation Time: {adapt_result.adaptation_time_ms:.1f}ms")
    
    print(f"\n💭 Predictions:")
    for i, (test_input, _) in enumerate(new_task.test_cases):
        print(f"   Input: {test_input}")
        print(f"   Predicted: {adapt_result.predictions[i]}")
    
    # Test 3: Transfer across task types
    print("\n" + "=" * 80)
    print("TEST 3: Transfer Learning to Different Task Type")
    print("=" * 80)
    
    different_task = Task(
        task_id="translate_new",
        task_type=TaskType.TRANSLATION,
        description="Simplify technical jargon",
        examples=[
            ("optimize", "improve"),
            ("implement", "do"),
        ],
        test_cases=[
            ("leverage", "use"),
            ("paradigm", "model"),
        ]
    )
    
    print(f"\n🔄 Transfer Task: {different_task.description}")
    print(f"   Task Type: {different_task.task_type.value}")
    
    transfer_result = agent.adapt(different_task)
    
    print(f"\n📊 Transfer Results:")
    print(f"   Strategy: {transfer_result.strategy_used}")
    print(f"   Performance: {transfer_result.performance:.1%}")
    print(f"   Adaptation Time: {transfer_result.adaptation_time_ms:.1f}ms")
    
    # Show statistics
    print(f"\n📈 Agent Statistics:")
    stats = agent.get_statistics()
    print(f"   Total Strategies: {stats['strategies_learned']}")
    print(f"   Total Experience: {stats['total_experience']}")
    
    print(f"\n   Strategy Usage:")
    for strategy_info in stats['strategy_details']:
        if strategy_info['usage_count'] > 0:
            print(f"      {strategy_info['id']}: {strategy_info['usage_count']} times "
                  f"({strategy_info['success_rate']} success)")
    
    # Summary
    print("\n" + "=" * 80)
    print("META-LEARNING AGENT PATTERN SUMMARY")
    print("=" * 80)
    print("""
Key Benefits:
1. Fast Adaptation: Learn from few examples
2. Transfer Learning: Knowledge across tasks
3. Efficiency: Less data needed per task
4. Generalization: Learn general strategies
5. Continuous Improvement: Better over time

Meta-Learning Strategies:
1. MAML: Model-agnostic meta-learning
   - Learn initialization that adapts quickly
   
2. Prompt Learning: Optimal prompt discovery
   - Learn effective prompt templates
   
3. Example Selection: Choose informative examples
   - Learn which examples help most
   
4. Strategy Learning: General approaches
   - Learn problem-solving strategies
   
5. Transfer Learning: Cross-task knowledge
   - Transfer learned representations

Meta-Training Process:
1. Sample task distribution
2. For each task:
   - Support set: few training examples
   - Query set: test examples
3. Learn strategy from patterns
4. Evaluate on held-out tasks
5. Update meta-parameters

Adaptation Process:
1. Receive new task
2. Select applicable strategy
3. Choose informative examples
4. Apply learned approach
5. Make predictions
6. Update based on feedback

Components:
- Meta-Learner: Learns strategies
- Base Learner: Applies to tasks
- Strategy Library: Learned approaches
- Experience Buffer: Historical performance
- Transfer Module: Cross-task knowledge

Use Cases:
- Few-shot learning
- Personalization (limited user data)
- Multi-domain systems
- Rapid prototyping
- Cross-lingual transfer
- Continuous learning

Key Concepts:
1. Support Set: Training examples
2. Query Set: Test examples
3. Inner Loop: Task-specific learning
4. Outer Loop: Meta-learning update
5. Task Distribution: Set of related tasks

Best Practices:
1. Diverse training tasks
2. Appropriate support set size
3. Strategy evaluation
4. Performance monitoring
5. Adaptive example selection
6. Regular meta-updates
7. Transfer validation

Production Considerations:
- Strategy library management
- Performance tracking per strategy
- Task similarity metrics
- Example caching
- Incremental meta-learning
- Cold start handling
- Computational efficiency

Comparison with Related Patterns:
- vs. Transfer Learning: Meta vs direct transfer
- vs. Few-Shot: General framework vs specific technique
- vs. Fine-tuning: Strategy learning vs parameter tuning
- vs. Ensemble: Meta-learning vs combination

Challenges:
1. Task distribution design
2. Strategy generalization
3. Computational cost
4. Overfitting to task distribution
5. Evaluation methodology

The Meta-Learning Agent pattern enables rapid adaptation to new
tasks with minimal examples by learning general strategies that
transfer effectively across related problems.
""")


if __name__ == "__main__":
    demonstrate_meta_learning()

    

Agentic Design Pattern: Meta Learning

Pattern 059: Meta Learning – Line-by-Line Explanation

Overview

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

---

Pattern Description

Pattern 059: Meta-Learning Agent

Description: Meta-learning (learning to learn) enables agents to improve their learning efficiency over time by learning general strategies that transfer across tasks. Rather than learning task-specific solutions, meta-learning agents learn how to adapt quickly to new tasks with minimal examples or experience.

Components: 1. Task Distribution: Collection of related learning tasks 2. Meta-Learner: Learns general learning strategies 3. Base Learner: Applies strategies to specific tasks 4. Adaptation Mechanism: Quick task-specific tuning 5. Transfer Module: Knowledge transfer across tasks

Use Cases: – Few-shot learning scenarios – Rapid adaptation to new domains – Personalization with limited data – Multi-task learning systems – Continuous learning environments – Cross-domain knowledge transfer

LangChain Implementation: Implements meta-learning through prompt optimization, example selection, and strategy learning that transfers across tasks.

---

File Statistics

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

---

Module Documentation

"""
Pattern 059: Meta-Learning Agent

Description:
    Meta-learning (learning to learn) enables agents to improve their learning
    efficiency over time by learning general strategies that transfer across tasks.
    Rather than learning task-specific solutions, meta-learning agents learn how
    to adapt quickly to new tasks with minimal examples or experience.

Components:
    1. Task Distribution: Collection of related learning tasks
    2. Meta-Learner: Learns general learning strategies
    3. Base Learner: Applies strategies to specific tasks
    4. Adaptation Mechanism: Quick task-specific tuning
    5. Transfer Module: Knowledge transfer across tasks

Use Cases:
    - Few-shot learning scenarios
    - Rapid adaptation to new domains
    - Personalization with limited data
    - Multi-task learning systems
    - Continuous learning environments
    - Cross-domain knowledge transfer

LangChain Implementation:
    Implements meta-learning through prompt optimization, example selection,
    and strategy learning that transfers across tasks.
"""

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

---

Import Statements

Standard Library Imports

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

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

LangChain Imports

from langchain_openai.ChatOpenAI

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

Other Imports

from dotenv.load_dotenv
from statistics

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: MetaStrategy

Lines 47-53

Inherits from: Enum

Purpose:

Meta-learning strategies

---

Class: TaskType

Lines 56-62

Inherits from: Enum

Purpose:

Types of learning tasks

---

Class: Task

Lines 66-73

Purpose:

A learning task

---

Class: LearnedStrategy

Lines 77-87

Purpose:

A learned problem-solving strategy

Methods:

is_applicable(self, task_type)

Lines 86-87

Purpose: Is Applicable operation.

Parameters: – task_type: Task type

Returns: bool

---

Class: AdaptationResult

Lines 91-98

Purpose:

Result from adaptation to new task

---

Class: MetaLearningResult

Lines 102-117

Purpose:

Result from meta-learning process

Methods:

to_dict(self)

Lines 110-117

Purpose: To Dict operation.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D628D50>

---

Class: MetaLearningAgent

Lines 120-406

Purpose:

Agent that learns to learn across tasks.

Features:
1. Strategy learning from multiple tasks
2. Fast adaptation to new tasks
3. Optimal example selection
4. Prompt optimization
5. Knowledge transfer

Methods:

__init__(self, meta_strategy, temperature)

Lines 132-159

Purpose: Initialize the class instance with required parameters.

Parameters: – meta_strategy: Meta strategy – temperature: Sampling temperature (0-1)

_learn_strategy_from_tasks(self, tasks)

Lines 161-212

Purpose:

Learn a general strategy from multiple tasks

Parameters: – tasks: Tasks

Returns: LearnedStrategy

_select_examples(self, task, k)

Lines 214-223

Purpose:

Select most informative examples for task

Parameters: – task: Task – k: K

Returns: <ast.Subscript object at 0x000001A06D683350>

_evaluate_performance(self, predictions, ground_truth)

Lines 225-241

Purpose:

Evaluate prediction performance

Parameters: – predictions: Predictions – ground_truth: Ground truth

Returns: float

meta_train(self, tasks, episodes_per_task)

Lines 243-316

Purpose:

Meta-training: Learn strategies from multiple tasks

Parameters: – tasks: Tasks – episodes_per_task: Episodes per task

Returns: MetaLearningResult

adapt(self, new_task, support_examples)

Lines 318-389

Purpose:

Adapt to new task using learned strategies

Parameters: – new_task: New task – support_examples: Support examples

Returns: AdaptationResult

get_statistics(self)

Lines 391-406

Purpose:

Get meta-learning statistics

Parameters:

Returns: <ast.Subscript object at 0x000001A06D728F50>

---

Functions

demonstrate_meta_learning()

Lines 409-687

Purpose:

Demonstrate Meta-Learning Agent pattern

---

Key Concepts Summary

Pattern Overview

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

Core Components

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

When to Use

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

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

---

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

---

LangChain Components Used

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

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

---

Execution Flow

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

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

---

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

---

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

---

Related Patterns

• Pattern 058
• Pattern 060

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 Meta 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: 059_meta_learning.py
Total Lines: 692
Pattern Category: AI Agent Design Pattern #059

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 058: Neuro-Symbolic Integration

Description:
    Neuro-Symbolic Integration combines neural (sub-symbolic, learned) approaches
    with symbolic (logical, rule-based) reasoning. This hybrid approach leverages
    the pattern recognition and generalization of neural methods with the
    interpretability, logical rigor, and structured reasoning of symbolic systems.

Components:
    1. Neural Module: Pattern recognition, learning from data
    2. Symbolic Module: Logical reasoning, rules, knowledge graphs
    3. Translation Layer: Converts between neural and symbolic representations
    4. Integration Strategy: Combines both approaches
    5. Verification Engine: Validates symbolic constraints

Use Cases:
    - Explainable AI requiring both learning and reasoning
    - Knowledge-grounded generation
    - Constraint-based problem solving with learning
    - Mathematical reasoning with verification
    - Legal/medical systems requiring compliance
    - Scientific reasoning with logical constraints

LangChain Implementation:
    Combines LLM-based neural reasoning with explicit symbolic rules
    and knowledge representation for hybrid problem-solving.
"""

import os
import time
from typing import Dict, Any, List, Optional, Set, Tuple, Union
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import re

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 ReasoningMode(Enum):
    """Integration modes"""
    NEURAL_FIRST = "neural_first"  # Neural then symbolic verification
    SYMBOLIC_FIRST = "symbolic_first"  # Symbolic constraints then neural
    INTERLEAVED = "interleaved"  # Alternate between both
    PARALLEL = "parallel"  # Both simultaneously, then merge
    HYBRID = "hybrid"  # Integrated throughout


class SymbolicConstraintType(Enum):
    """Types of symbolic constraints"""
    LOGICAL = "logical"  # Boolean logic rules
    MATHEMATICAL = "mathematical"  # Numerical constraints
    TEMPORAL = "temporal"  # Time-based rules
    ONTOLOGICAL = "ontological"  # Knowledge hierarchy
    PROCEDURAL = "procedural"  # Step-by-step rules


@dataclass
class SymbolicRule:
    """Symbolic rule or constraint"""
    rule_id: str
    rule_type: SymbolicConstraintType
    description: str
    condition: str  # Logical expression
    action: Optional[str] = None
    priority: int = 1
    
    def evaluate(self, facts: Dict[str, Any]) -> bool:
        """Evaluate rule against facts"""
        try:
            # Simple evaluation (in production, use proper logic engine)
            for key, value in facts.items():
                self.condition = self.condition.replace(key, str(value))
            
            # Basic evaluation
            if "AND" in self.condition:
                parts = self.condition.split("AND")
                return all(self._evaluate_simple(p.strip(), facts) for p in parts)
            elif "OR" in self.condition:
                parts = self.condition.split("OR")
                return any(self._evaluate_simple(p.strip(), facts) for p in parts)
            else:
                return self._evaluate_simple(self.condition, facts)
        except:
            return False
    
    def _evaluate_simple(self, expr: str, facts: Dict[str, Any]) -> bool:
        """Evaluate simple expression"""
        try:
            # Very simple evaluation
            if ">" in expr:
                left, right = expr.split(">")
                return float(left.strip()) > float(right.strip())
            elif "<" in expr:
                left, right = expr.split("<")
                return float(left.strip())  List[Tuple[str, str, str]]:
        """Query knowledge graph"""
        results = []
        for rel in self.relations:
            if all(
                p is None or p == r
                for p, r in zip(pattern, rel)
            ):
                results.append(rel)
        return results
    
    def to_text(self) -> str:
        """Convert to text representation"""
        text = "Knowledge Base:\n"
        for subj, pred, obj in self.relations:
            text += f"  {subj} {pred} {obj}\n"
        return text


@dataclass
class NeuroSymbolicResult:
    """Result from neuro-symbolic reasoning"""
    query: str
    neural_output: str
    symbolic_verification: Dict[str, Any]
    final_output: str
    constraints_satisfied: bool
    reasoning_trace: List[str]
    confidence: float
    execution_time_ms: float
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "query": self.query[:100] + "...",
            "constraints_satisfied": self.constraints_satisfied,
            "confidence": f"{self.confidence:.2f}",
            "reasoning_steps": len(self.reasoning_trace),
            "execution_time_ms": f"{self.execution_time_ms:.1f}"
        }


class NeuroSymbolicAgent:
    """
    Agent integrating neural and symbolic reasoning.
    
    Features:
    1. Neural reasoning via LLM
    2. Symbolic constraint checking
    3. Knowledge graph integration
    4. Logical verification
    5. Hybrid decision making
    """
    
    def __init__(
        self,
        reasoning_mode: ReasoningMode = ReasoningMode.NEURAL_FIRST,
        temperature: float = 0.3
    ):
        self.reasoning_mode = reasoning_mode
        self.temperature = temperature
        
        # Neural component (LLM)
        self.neural_model = ChatOpenAI(
            model_name="gpt-3.5-turbo",
            temperature=temperature
        )
        
        # Symbolic components
        self.rules: List[SymbolicRule] = []
        self.knowledge_graph = KnowledgeGraph()
        
        # Reasoning trace
        self.reasoning_trace: List[str] = []
    
    def add_rule(self, rule: SymbolicRule):
        """Add symbolic rule"""
        self.rules.append(rule)
    
    def add_knowledge(
        self,
        subject: str,
        predicate: str,
        obj: str
    ):
        """Add knowledge to graph"""
        self.knowledge_graph.add_relation(subject, predicate, obj)
    
    def _neural_reasoning(self, query: str, context: Optional[str] = None) -> str:
        """Neural reasoning step"""
        
        self.reasoning_trace.append("Neural: Generating initial response")
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", """You are reasoning about a query.
Provide a thoughtful, logical response.

{context}"""),
            ("user", "{query}")
        ])
        
        context_text = context if context else "Use your knowledge to respond."
        
        chain = prompt | self.neural_model | StrOutputParser()
        response = chain.invoke({
            "query": query,
            "context": context_text
        })
        
        return response
    
    def _symbolic_verification(
        self,
        neural_output: str,
        extracted_facts: Dict[str, Any]
    ) -> Dict[str, Any]:
        """Verify neural output against symbolic constraints"""
        
        self.reasoning_trace.append("Symbolic: Verifying constraints")
        
        verification = {
            "all_satisfied": True,
            "violations": [],
            "satisfied_rules": [],
            "extracted_facts": extracted_facts
        }
        
        # Check each rule
        for rule in self.rules:
            satisfied = rule.evaluate(extracted_facts)
            
            if satisfied:
                verification["satisfied_rules"].append(rule.rule_id)
                self.reasoning_trace.append(f"  ✓ Rule {rule.rule_id}: {rule.description}")
            else:
                verification["all_satisfied"] = False
                verification["violations"].append({
                    "rule_id": rule.rule_id,
                    "description": rule.description,
                    "condition": rule.condition
                })
                self.reasoning_trace.append(f"  ✗ Rule {rule.rule_id}: {rule.description}")
        
        return verification
    
    def _extract_facts(self, text: str) -> Dict[str, Any]:
        """Extract facts from neural output"""
        
        self.reasoning_trace.append("Extracting facts from neural output")
        
        extraction_prompt = ChatPromptTemplate.from_messages([
            ("system", """Extract key facts and values from the text.

Format as key-value pairs:
key1: value1
key2: value2"""),
            ("user", "Text: {text}\n\nExtracted facts:")
        ])
        
        chain = extraction_prompt | self.neural_model | StrOutputParser()
        facts_text = chain.invoke({"text": text})
        
        # Parse facts
        facts = {}
        for line in facts_text.split('\n'):
            if ':' in line:
                key, value = line.split(':', 1)
                key = key.strip().replace(' ', '_').lower()
                value = value.strip()
                
                # Try to convert to number
                try:
                    value = float(value)
                except:
                    pass
                
                facts[key] = value
        
        return facts
    
    def _refine_with_constraints(
        self,
        query: str,
        neural_output: str,
        violations: List[Dict[str, Any]]
    ) -> str:
        """Refine neural output to satisfy constraints"""
        
        self.reasoning_trace.append("Neural: Refining to satisfy constraints")
        
        violations_text = "\n".join([
            f"- {v['description']}: {v['condition']}"
            for v in violations
        ])
        
        refinement_prompt = ChatPromptTemplate.from_messages([
            ("system", """Your previous response violated some constraints.
Revise your response to satisfy these constraints:

{violations}

Maintain the core meaning but adjust to meet all constraints."""),
            ("user", """Original query: {query}

Previous response: {response}

Revised response:""")
        ])
        
        chain = refinement_prompt | self.neural_model | StrOutputParser()
        refined = chain.invoke({
            "query": query,
            "response": neural_output,
            "violations": violations_text
        })
        
        return refined
    
    def _integrate_knowledge(self, query: str) -> str:
        """Integrate knowledge graph into reasoning"""
        
        # Query knowledge graph for relevant facts
        # (simplified - in production, use semantic matching)
        relevant_facts = []
        
        query_lower = query.lower()
        for subject, predicate, obj in self.knowledge_graph.relations:
            if (subject.lower() in query_lower or 
                obj.lower() in query_lower):
                relevant_facts.append(f"{subject} {predicate} {obj}")
        
        if relevant_facts:
            context = "Relevant knowledge:\n" + "\n".join(relevant_facts[:5])
            self.reasoning_trace.append(f"Retrieved {len(relevant_facts)} relevant facts")
            return context
        
        return ""
    
    def reason(
        self,
        query: str,
        max_refinement_iterations: int = 3
    ) -> NeuroSymbolicResult:
        """Perform neuro-symbolic reasoning"""
        
        start_time = time.time()
        self.reasoning_trace = []
        
        if self.reasoning_mode == ReasoningMode.NEURAL_FIRST:
            # Neural reasoning first, then symbolic verification
            
            # 1. Integrate knowledge
            context = self._integrate_knowledge(query)
            
            # 2. Neural reasoning
            neural_output = self._neural_reasoning(query, context)
            
            # 3. Extract facts
            facts = self._extract_facts(neural_output)
            
            # 4. Symbolic verification
            verification = self._symbolic_verification(neural_output, facts)
            
            # 5. Refine if needed
            final_output = neural_output
            iteration = 0
            
            while not verification["all_satisfied"] and iteration  0",
        priority=1
    ))
    
    agent1.add_rule(SymbolicRule(
        rule_id="reasonable_range",
        rule_type=SymbolicConstraintType.MATHEMATICAL,
        description="Result must be less than 1000",
        condition="result < 1000",
        priority=1
    ))
    
    query1 = "What is 25 multiplied by 8?"
    
    print(f"\n💭 Query: {query1}")
    print(f"🔧 Constraints:")
    for rule in agent1.rules:
        print(f"   - {rule.description}")
    
    result1 = agent1.reason(query1)
    
    print(f"\n🧠 Neural Output: {result1.neural_output[:150]}...")
    print(f"\n⚙️  Symbolic Verification:")
    print(f"   Constraints Satisfied: {result1.constraints_satisfied}")
    
    if result1.symbolic_verification["violations"]:
        print(f"   Violations:")
        for v in result1.symbolic_verification["violations"]:
            print(f"      - {v['description']}")
    
    if result1.symbolic_verification["satisfied_rules"]:
        print(f"   Satisfied Rules: {', '.join(result1.symbolic_verification['satisfied_rules'])}")
    
    print(f"\n✅ Final Output: {result1.final_output[:200]}...")
    print(f"   Confidence: {result1.confidence:.2f}")
    
    # Test 2: Knowledge graph integration
    print("\n" + "=" * 80)
    print("TEST 2: Knowledge Graph Integration")
    print("=" * 80)
    
    agent2 = NeuroSymbolicAgent(
        reasoning_mode=ReasoningMode.HYBRID
    )
    
    # Build knowledge graph
    agent2.add_knowledge("Python", "is_a", "programming_language")
    agent2.add_knowledge("Python", "created_by", "Guido_van_Rossum")
    agent2.add_knowledge("Python", "released_in", "1991")
    agent2.add_knowledge("Python", "used_for", "machine_learning")
    agent2.add_knowledge("Python", "used_for", "web_development")
    
    print(f"\n📚 Knowledge Graph:")
    for subj, pred, obj in agent2.knowledge_graph.relations:
        print(f"   {subj} {pred} {obj}")
    
    query2 = "When was Python created and what is it used for?"
    
    print(f"\n💭 Query: {query2}")
    
    result2 = agent2.reason(query2)
    
    print(f"\n🔍 Reasoning Trace:")
    for i, step in enumerate(result2.reasoning_trace[:5], 1):
        print(f"   {i}. {step}")
    
    print(f"\n💬 Response: {result2.final_output[:250]}...")
    
    # Test 3: Constraint refinement
    print("\n" + "=" * 80)
    print("TEST 3: Iterative Constraint Refinement")
    print("=" * 80)
    
    agent3 = NeuroSymbolicAgent(
        reasoning_mode=ReasoningMode.NEURAL_FIRST
    )
    
    # Add logical constraint
    agent3.add_rule(SymbolicRule(
        rule_id="format_check",
        rule_type=SymbolicConstraintType.LOGICAL,
        description="Response must include steps",
        condition="steps",  # Simplified
        priority=2
    ))
    
    query3 = "How do I solve a quadratic equation?"
    
    print(f"\n💭 Query: {query3}")
    
    result3 = agent3.reason(query3, max_refinement_iterations=2)
    
    print(f"\n🔄 Refinement Process:")
    print(f"   Total Steps: {len(result3.reasoning_trace)}")
    print(f"   Constraints Satisfied: {result3.constraints_satisfied}")
    
    print(f"\n📝 Reasoning Trace:")
    for step in result3.reasoning_trace:
        print(f"   - {step}")
    
    print(f"\n✅ Final Answer: {result3.final_output[:200]}...")
    
    # Summary
    print("\n" + "=" * 80)
    print("NEURO-SYMBOLIC INTEGRATION PATTERN SUMMARY")
    print("=" * 80)
    print("""
Key Benefits:
1. Best of Both: Neural flexibility + symbolic rigor
2. Explainability: Symbolic rules provide interpretability
3. Constraint Satisfaction: Guaranteed compliance
4. Knowledge Integration: Use structured knowledge
5. Verification: Validate neural outputs

Integration Modes:
1. Neural-First: Generate then verify
   - Fast, flexible generation
   - Symbolic post-processing

2. Symbolic-First: Constrain then generate
   - Guaranteed satisfaction
   - May limit creativity

3. Interleaved: Alternate steps
   - Balanced approach
   - More iterations

4. Parallel: Both simultaneously
   - Independent processing
   - Merge results

5. Hybrid: Fully integrated
   - Seamless combination
   - Complex implementation

Components:
1. Neural Module: LLM for pattern recognition
2. Symbolic Module: Rules, logic, knowledge graph
3. Translation Layer: Convert representations
4. Verification Engine: Check constraints
5. Refinement Loop: Iterative improvement

Symbolic Constraints:
- Logical: Boolean rules (AND, OR, NOT)
- Mathematical: Numerical constraints
- Temporal: Time-based rules
- Ontological: Knowledge hierarchies
- Procedural: Step sequences

Knowledge Representation:
- Knowledge Graphs: Entity-relation-entity
- Rules: If-then logic
- Ontologies: Hierarchical concepts
- Constraints: Mathematical inequalities
- Facts: Ground truth statements

Use Cases:
- Explainable AI systems
- Compliance-critical applications
- Mathematical reasoning
- Legal/medical systems
- Scientific hypothesis generation
- Code generation with verification

Best Practices:
1. Clear symbolic constraints
2. Efficient fact extraction
3. Iterative refinement limits
4. Knowledge graph maintenance
5. Rule priority ordering
6. Verification performance
7. Fallback strategies

Production Considerations:
- Rule engine scalability
- Knowledge graph indexing
- Refinement iteration limits
- Caching verified results
- Parallel verification
- Constraint complexity
- Performance optimization

Comparison with Related Patterns:
- vs. Pure Neural: Adds verification
- vs. Pure Symbolic: Adds flexibility
- vs. Retrieval: Structured knowledge vs search
- vs. Chain-of-Thought: Formal logic vs reasoning

Challenges:
1. Representation gap: Neural ↔ symbolic translation
2. Scalability: Large rule sets
3. Conflict resolution: Contradictory rules
4. Learning rules: Manual vs automated
5. Integration complexity

The Neuro-Symbolic Integration pattern enables robust AI systems
that combine the pattern recognition power of neural networks
with the interpretability and logical rigor of symbolic reasoning.
""")


if __name__ == "__main__":
    demonstrate_neuro_symbolic()

    

Agentic Design Pattern: Neuro Symbolic

Pattern 058: Neuro Symbolic – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Neuro Symbolic pattern implementation.

---

Pattern Description

Pattern 058: Neuro-Symbolic Integration

Description: Neuro-Symbolic Integration combines neural (sub-symbolic, learned) approaches with symbolic (logical, rule-based) reasoning. This hybrid approach leverages the pattern recognition and generalization of neural methods with the interpretability, logical rigor, and structured reasoning of symbolic systems.

Components: 1. Neural Module: Pattern recognition, learning from data 2. Symbolic Module: Logical reasoning, rules, knowledge graphs 3. Translation Layer: Converts between neural and symbolic representations 4. Integration Strategy: Combines both approaches 5. Verification Engine: Validates symbolic constraints

Use Cases: – Explainable AI requiring both learning and reasoning – Knowledge-grounded generation – Constraint-based problem solving with learning – Mathematical reasoning with verification – Legal/medical systems requiring compliance – Scientific reasoning with logical constraints

LangChain Implementation: Combines LLM-based neural reasoning with explicit symbolic rules and knowledge representation for hybrid problem-solving.

---

File Statistics

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

---

Module Documentation

"""
Pattern 058: Neuro-Symbolic Integration

Description:
    Neuro-Symbolic Integration combines neural (sub-symbolic, learned) approaches
    with symbolic (logical, rule-based) reasoning. This hybrid approach leverages
    the pattern recognition and generalization of neural methods with the
    interpretability, logical rigor, and structured reasoning of symbolic systems.

Components:
    1. Neural Module: Pattern recognition, learning from data
    2. Symbolic Module: Logical reasoning, rules, knowledge graphs
    3. Translation Layer: Converts between neural and symbolic representations
    4. Integration Strategy: Combines both approaches
    5. Verification Engine: Validates symbolic constraints

Use Cases:
    - Explainable AI requiring both learning and reasoning
    - Knowledge-grounded generation
    - Constraint-based problem solving with learning
    - Mathematical reasoning with verification
    - Legal/medical systems requiring compliance
    - Scientific reasoning with logical constraints

LangChain Implementation:
    Combines LLM-based neural reasoning with explicit symbolic rules
    and knowledge representation for hybrid problem-solving.
"""

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

---

Import Statements

Standard Library Imports

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

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

LangChain Imports

from langchain_openai.ChatOpenAI

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

Other Imports

from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: ReasoningMode

Lines 46-52

Inherits from: Enum

Purpose:

Integration modes

---

Class: SymbolicConstraintType

Lines 55-61

Inherits from: Enum

Purpose:

Types of symbolic constraints

---

Class: SymbolicRule

Lines 65-113

Purpose:

Symbolic rule or constraint

Methods:

evaluate(self, facts)

Lines 74-91

Purpose:

Evaluate rule against facts

Parameters: – facts: Facts

Returns: bool

_evaluate_simple(self, expr, facts)

Lines 93-113

Purpose:

Evaluate simple expression

Parameters: – expr: Expr – facts: Facts

Returns: bool

---

Class: KnowledgeGraph

Lines 117-147

Purpose:

Simple knowledge graph

Methods:

add_entity(self, entity)

Lines 123-124

Purpose: Add Entity operation.

Parameters: – entity: Entity

add_relation(self, subject, predicate, obj)

Lines 126-129

Purpose: Add Relation operation.

Parameters: – subject: Subject – predicate: Predicate – obj: Obj

query(self, pattern)

Lines 131-140

Purpose:

Query knowledge graph

Parameters: – pattern: Pattern

Returns: <ast.Subscript object at 0x000001A06D632150>

to_text(self)

Lines 142-147

Purpose:

Convert to text representation

Parameters:

Returns: str

---

Class: NeuroSymbolicResult

Lines 151-169

Purpose:

Result from neuro-symbolic reasoning

Methods:

to_dict(self)

Lines 162-169

Purpose: To Dict operation.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D668610>

---

Class: NeuroSymbolicAgent

Lines 172-462

Purpose:

Agent integrating neural and symbolic reasoning.

Features:
1. Neural reasoning via LLM
2. Symbolic constraint checking
3. Knowledge graph integration
4. Logical verification
5. Hybrid decision making

Methods:

__init__(self, reasoning_mode, temperature)

Lines 184-203

Purpose: Initialize the class instance with required parameters.

Parameters: – reasoning_mode: Reasoning mode – temperature: Sampling temperature (0-1)

add_rule(self, rule)

Lines 205-207

Purpose:

Add symbolic rule

Parameters: – rule: Rule

add_knowledge(self, subject, predicate, obj)

Lines 209-216

Purpose:

Add knowledge to graph

Parameters: – subject: Subject – predicate: Predicate – obj: Obj

_neural_reasoning(self, query, context)

Lines 218-239

Purpose:

Neural reasoning step

Parameters: – query: Input query or question – context: Context

Returns: str

_symbolic_verification(self, neural_output, extracted_facts)

Lines 241-273

Purpose:

Verify neural output against symbolic constraints

Parameters: – neural_output: Neural output – extracted_facts: Extracted facts

Returns: <ast.Subscript object at 0x000001A06D626310>

_extract_facts(self, text)

Lines 275-308

Purpose:

Extract facts from neural output

Parameters: – text: Text

Returns: <ast.Subscript object at 0x000001A06D6B10D0>

_refine_with_constraints(self, query, neural_output, violations)

Lines 310-346

Purpose:

Refine neural output to satisfy constraints

Parameters: – query: Input query or question – neural_output: Neural output – violations: Violations

Returns: str

_integrate_knowledge(self, query)

Lines 348-366

Purpose:

Integrate knowledge graph into reasoning

Parameters: – query: Input query or question

Returns: str

reason(self, query, max_refinement_iterations)

Lines 368-462

Purpose:

Perform neuro-symbolic reasoning

Parameters: – query: Input query or question – max_refinement_iterations: Max refinement iterations

Returns: NeuroSymbolicResult

---

Functions

demonstrate_neuro_symbolic()

Lines 465-686

Purpose:

Demonstrate Neuro-Symbolic Integration pattern

---

Key Concepts Summary

Pattern Overview

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

Core Components

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

When to Use

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

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

---

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

---

LangChain Components Used

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

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

---

Execution Flow

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

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

---

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

---

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

---

Related Patterns

• Pattern 057
• Pattern 059

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 Neuro Symbolic 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: 058_neuro_symbolic.py
Total Lines: 691
Pattern Category: AI Agent Design Pattern #058

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 057: Attention Mechanism Patterns

Description:
    Attention mechanisms enable agents to selectively focus on the most relevant
    parts of large inputs, contexts, or action spaces. Inspired by human attention
    and transformer architectures, these patterns help agents efficiently process
    information by allocating computational resources where they matter most.

Components:
    1. Attention Scorer: Calculates relevance scores
    2. Context Selector: Chooses relevant information
    3. Focus Manager: Manages attention allocation
    4. Multi-Head Attention: Multiple attention perspectives
    5. Self-Attention: Internal context relationships

Use Cases:
    - Long document processing
    - Multi-source information synthesis
    - Large context windows
    - Memory retrieval from extensive history
    - Action selection in large action spaces
    - Feature selection in complex data

LangChain Implementation:
    Implements attention scoring and selection mechanisms using LLMs
    to identify and focus on relevant portions of context.
"""

import os
import time
from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import math

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 AttentionType(Enum):
    """Types of attention mechanisms"""
    SELF_ATTENTION = "self_attention"  # Internal relationships
    CROSS_ATTENTION = "cross_attention"  # Query-context relationships
    MULTI_HEAD = "multi_head"  # Multiple attention perspectives
    SPARSE = "sparse"  # Attend to subset
    HIERARCHICAL = "hierarchical"  # Multi-level attention


class FocusMode(Enum):
    """Modes of attention focus"""
    NARROW = "narrow"  # Focus on very specific parts
    BROAD = "broad"  # Consider wider context
    ADAPTIVE = "adaptive"  # Adjust based on task
    DISTRIBUTED = "distributed"  # Spread across multiple items


@dataclass
class AttentionScore:
    """Attention score for an item"""
    item_id: str
    content: str
    score: float  # 0.0-1.0
    reasoning: Optional[str] = None
    
    def __repr__(self) -> str:
        return f"AttentionScore(id={self.item_id}, score={self.score:.3f})"


@dataclass
class AttentionHead:
    """Single attention head with specific perspective"""
    head_id: str
    perspective: str
    scores: List[AttentionScore]
    
    @property
    def top_items(self, k: int = 3) -> List[AttentionScore]:
        return sorted(self.scores, key=lambda x: x.score, reverse=True)[:k]


@dataclass
class AttentionResult:
    """Result from attention mechanism"""
    query: str
    attended_items: List[AttentionScore]
    attention_type: AttentionType
    focus_mode: FocusMode
    total_items: int
    items_attended: int
    attention_distribution: Dict[str, float]
    execution_time_ms: float
    
    @property
    def coverage(self) -> float:
        """Percentage of items attended"""
        return self.items_attended / self.total_items if self.total_items > 0 else 0.0
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "query": self.query[:80] + "...",
            "attention_type": self.attention_type.value,
            "focus_mode": self.focus_mode.value,
            "coverage": f"{self.coverage:.1%}",
            "top_attended": [
                f"{item.item_id} ({item.score:.2f})"
                for item in self.attended_items[:3]
            ],
            "execution_time_ms": f"{self.execution_time_ms:.1f}"
        }


class AttentionMechanism:
    """
    Implements attention mechanisms for selective focus.
    
    Features:
    1. Query-based attention scoring
    2. Multi-head attention for diverse perspectives
    3. Sparse attention for efficiency
    4. Hierarchical attention for structured data
    5. Adaptive focus modes
    """
    
    def __init__(
        self,
        attention_type: AttentionType = AttentionType.CROSS_ATTENTION,
        focus_mode: FocusMode = FocusMode.ADAPTIVE,
        num_heads: int = 3,
        temperature: float = 0.3
    ):
        self.attention_type = attention_type
        self.focus_mode = focus_mode
        self.num_heads = num_heads
        self.temperature = temperature
        
        # Attention scorer LLM
        self.scorer = ChatOpenAI(
            model_name="gpt-3.5-turbo",
            temperature=temperature
        )
        
        # Synthesizer for multi-head combination
        self.synthesizer = ChatOpenAI(
            model_name="gpt-3.5-turbo",
            temperature=0.2
        )
    
    def _score_relevance(
        self,
        query: str,
        item: str,
        item_id: str,
        perspective: str = "general"
    ) -> AttentionScore:
        """Score relevance of item to query"""
        
        scoring_prompt = ChatPromptTemplate.from_messages([
            ("system", f"""You are scoring relevance from a {perspective} perspective.

Rate how relevant the item is to the query on a scale of 0.0 to 1.0.

Respond with just the number (e.g., 0.85)"""),
            ("user", "Query: {query}\n\nItem: {item}\n\nRelevance score:")
        ])
        
        chain = scoring_prompt | self.scorer | StrOutputParser()
        
        try:
            score_str = chain.invoke({"query": query, "item": item}).strip()
            # Extract number from response
            score = float(''.join(c for c in score_str if c.isdigit() or c == '.'))
            score = max(0.0, min(1.0, score))
        except:
            # Fallback: simple keyword matching
            query_words = set(query.lower().split())
            item_words = set(item.lower().split())
            overlap = len(query_words & item_words)
            score = min(1.0, overlap / max(len(query_words), 1) * 2)
        
        return AttentionScore(
            item_id=item_id,
            content=item,
            score=score,
            reasoning=f"{perspective} relevance"
        )
    
    def _cross_attention(
        self,
        query: str,
        context_items: List[Tuple[str, str]],  # (id, content)
        top_k: int = 5
    ) -> List[AttentionScore]:
        """Cross-attention: query attending to context"""
        
        scores = []
        
        for item_id, content in context_items:
            score = self._score_relevance(query, content, item_id)
            scores.append(score)
        
        # Apply softmax-like normalization
        total = sum(s.score for s in scores)
        if total > 0:
            for score in scores:
                score.score = score.score / total
        
        # Return top-k
        scores.sort(key=lambda x: x.score, reverse=True)
        return scores[:top_k]
    
    def _self_attention(
        self,
        items: List[Tuple[str, str]],  # (id, content)
        top_k: int = 5
    ) -> List[AttentionScore]:
        """Self-attention: items attending to each other"""
        
        # For each item, calculate its relationship to all others
        attention_matrix = []
        
        for i, (id1, content1) in enumerate(items):
            row_scores = []
            for j, (id2, content2) in enumerate(items):
                if i != j:
                    # Simple similarity (could use embeddings)
                    words1 = set(content1.lower().split())
                    words2 = set(content2.lower().split())
                    similarity = len(words1 & words2) / max(len(words1 | words2), 1)
                    row_scores.append((j, similarity))
            
            # Sum of attention from this item to others
            total_attention = sum(s for _, s in row_scores)
            attention_matrix.append((i, total_attention))
        
        # Sort by total attention received
        attention_matrix.sort(key=lambda x: x[1], reverse=True)
        
        # Return top-k most attended items
        result = []
        for idx, attention in attention_matrix[:top_k]:
            item_id, content = items[idx]
            result.append(AttentionScore(
                item_id=item_id,
                content=content,
                score=attention,
                reasoning="Self-attention importance"
            ))
        
        # Normalize scores
        total = sum(s.score for s in result)
        if total > 0:
            for score in result:
                score.score = score.score / total
        
        return result
    
    def _multi_head_attention(
        self,
        query: str,
        context_items: List[Tuple[str, str]],
        perspectives: List[str],
        top_k: int = 5
    ) -> Tuple[List[AttentionScore], List[AttentionHead]]:
        """Multi-head attention: multiple perspectives"""
        
        heads = []
        
        # Each head has different perspective
        for i, perspective in enumerate(perspectives[:self.num_heads]):
            head_scores = []
            
            for item_id, content in context_items:
                score = self._score_relevance(
                    query, content, item_id, perspective=perspective
                )
                head_scores.append(score)
            
            head = AttentionHead(
                head_id=f"head_{i}",
                perspective=perspective,
                scores=head_scores
            )
            heads.append(head)
        
        # Combine heads (average scores)
        combined_scores = {}
        for head in heads:
            for score in head.scores:
                if score.item_id not in combined_scores:
                    combined_scores[score.item_id] = {
                        "content": score.content,
                        "scores": []
                    }
                combined_scores[score.item_id]["scores"].append(score.score)
        
        # Average across heads
        final_scores = []
        for item_id, data in combined_scores.items():
            avg_score = sum(data["scores"]) / len(data["scores"])
            final_scores.append(AttentionScore(
                item_id=item_id,
                content=data["content"],
                score=avg_score,
                reasoning="Multi-head average"
            ))
        
        # Normalize and return top-k
        total = sum(s.score for s in final_scores)
        if total > 0:
            for score in final_scores:
                score.score = score.score / total
        
        final_scores.sort(key=lambda x: x.score, reverse=True)
        
        return final_scores[:top_k], heads
    
    def _sparse_attention(
        self,
        query: str,
        context_items: List[Tuple[str, str]],
        top_k: int = 5
    ) -> List[AttentionScore]:
        """Sparse attention: only attend to subset"""
        
        # Quick filtering using keywords
        query_keywords = set(query.lower().split())
        
        # Score only items with keyword overlap
        candidate_items = []
        for item_id, content in context_items:
            content_words = set(content.lower().split())
            if query_keywords & content_words:  # Has overlap
                candidate_items.append((item_id, content))
        
        # If too few candidates, expand
        if len(candidate_items)  AttentionResult:
        """Apply attention mechanism"""
        
        start_time = time.time()
        
        total_items = len(context_items)
        
        # Apply appropriate attention type
        if self.attention_type == AttentionType.CROSS_ATTENTION:
            attended = self._cross_attention(query, context_items, top_k)
            
        elif self.attention_type == AttentionType.SELF_ATTENTION:
            attended = self._self_attention(context_items, top_k)
            
        elif self.attention_type == AttentionType.MULTI_HEAD:
            perspectives = ["factual", "emotional", "practical"]
            attended, heads = self._multi_head_attention(
                query, context_items, perspectives, top_k
            )
            
        elif self.attention_type == AttentionType.SPARSE:
            attended = self._sparse_attention(query, context_items, top_k)
            
        else:  # Default to cross-attention
            attended = self._cross_attention(query, context_items, top_k)
        
        # Calculate attention distribution
        distribution = {
            item.item_id: item.score
            for item in attended
        }
        
        execution_time_ms = (time.time() - start_time) * 1000
        
        return AttentionResult(
            query=query,
            attended_items=attended,
            attention_type=self.attention_type,
            focus_mode=self.focus_mode,
            total_items=total_items,
            items_attended=len(attended),
            attention_distribution=distribution,
            execution_time_ms=execution_time_ms
        )
    
    def focused_response(
        self,
        query: str,
        context_items: List[Tuple[str, str]],
        top_k: int = 3
    ) -> Tuple[str, AttentionResult]:
        """Generate response using attended context"""
        
        # Apply attention
        attention = self.attend(query, context_items, top_k)
        
        # Build focused context
        focused_context = "\n\n".join([
            f"[Relevance: {item.score:.2f}] {item.content}"
            for item in attention.attended_items
        ])
        
        # Generate response with focused context
        response_prompt = ChatPromptTemplate.from_messages([
            ("system", """You have access to focused, relevant context.
Use this context to provide an accurate, helpful response.

Relevant Context:
{context}"""),
            ("user", "{query}")
        ])
        
        chain = response_prompt | self.synthesizer | StrOutputParser()
        response = chain.invoke({
            "query": query,
            "context": focused_context
        })
        
        return response, attention


def demonstrate_attention_mechanisms():
    """Demonstrate Attention Mechanism patterns"""
    
    print("=" * 80)
    print("PATTERN 057: ATTENTION MECHANISM PATTERNS DEMONSTRATION")
    print("=" * 80)
    print("\nSelective focus on relevant information\n")
    
    # Create sample context
    documents = [
        ("doc1", "Python is a high-level programming language known for its simplicity and readability."),
        ("doc2", "Machine learning models require large amounts of training data to perform well."),
        ("doc3", "The transformer architecture uses self-attention mechanisms for sequence processing."),
        ("doc4", "Python's simplicity makes it popular for machine learning and data science applications."),
        ("doc5", "Deep learning frameworks like TensorFlow and PyTorch are commonly used in Python."),
        ("doc6", "Attention mechanisms allow models to focus on relevant parts of the input."),
        ("doc7", "Natural language processing has benefited greatly from attention-based models."),
        ("doc8", "Python's extensive libraries make it suitable for various AI tasks."),
    ]
    
    # Test 1: Cross-attention
    print("\n" + "=" * 80)
    print("TEST 1: Cross-Attention (Query attending to Context)")
    print("=" * 80)
    
    attn1 = AttentionMechanism(
        attention_type=AttentionType.CROSS_ATTENTION,
        focus_mode=FocusMode.NARROW
    )
    
    query1 = "What is Python used for in machine learning?"
    
    print(f"\n💭 Query: {query1}")
    print(f"📚 Total Documents: {len(documents)}")
    
    result1 = attn1.attend(query1, documents, top_k=3)
    
    print(f"\n🎯 Attention Results:")
    print(f"   Coverage: {result1.coverage:.1%} ({result1.items_attended}/{result1.total_items})")
    print(f"   Execution Time: {result1.execution_time_ms:.1f}ms")
    
    print(f"\n📊 Top Attended Items:")
    for i, item in enumerate(result1.attended_items, 1):
        print(f"   {i}. {item.item_id} (score: {item.score:.3f})")
        print(f"      {item.content[:80]}...")
    
    # Test 2: Self-attention
    print("\n" + "=" * 80)
    print("TEST 2: Self-Attention (Items attending to each other)")
    print("=" * 80)
    
    attn2 = AttentionMechanism(
        attention_type=AttentionType.SELF_ATTENTION,
        focus_mode=FocusMode.ADAPTIVE
    )
    
    result2 = attn2.attend("", documents, top_k=4)  # No query for self-attention
    
    print(f"\n🔗 Most Interconnected Documents:")
    for i, item in enumerate(result2.attended_items, 1):
        print(f"   {i}. {item.item_id} (attention: {item.score:.3f})")
        print(f"      {item.content[:80]}...")
    
    # Test 3: Multi-head attention
    print("\n" + "=" * 80)
    print("TEST 3: Multi-Head Attention (Multiple Perspectives)")
    print("=" * 80)
    
    attn3 = AttentionMechanism(
        attention_type=AttentionType.MULTI_HEAD,
        num_heads=3
    )
    
    query3 = "Explain attention mechanisms"
    
    print(f"\n💭 Query: {query3}")
    print(f"👁️  Perspectives: factual, emotional, practical")
    
    result3 = attn3.attend(query3, documents, top_k=3)
    
    print(f"\n📊 Combined Multi-Head Results:")
    for i, item in enumerate(result3.attended_items, 1):
        print(f"   {i}. {item.item_id} (combined score: {item.score:.3f})")
        print(f"      {item.content[:80]}...")
    
    # Test 4: Focused response generation
    print("\n" + "=" * 80)
    print("TEST 4: Focused Response Generation")
    print("=" * 80)
    
    attn4 = AttentionMechanism(
        attention_type=AttentionType.CROSS_ATTENTION,
        focus_mode=FocusMode.NARROW
    )
    
    query4 = "Why is Python popular for AI development?"
    
    print(f"\n💭 Query: {query4}")
    
    response, attention = attn4.focused_response(query4, documents, top_k=3)
    
    print(f"\n🎯 Attention Focus:")
    for item in attention.attended_items:
        print(f"   - {item.item_id}: {item.score:.3f}")
    
    print(f"\n💬 Focused Response:")
    print(f"   {response}")
    
    # Test 5: Sparse attention
    print("\n" + "=" * 80)
    print("TEST 5: Sparse Attention (Efficient Processing)")
    print("=" * 80)
    
    # Create larger context
    large_docs = documents + [
        (f"doc{i}", f"This is an irrelevant document number {i}.")
        for i in range(9, 20)
    ]
    
    attn5 = AttentionMechanism(
        attention_type=AttentionType.SPARSE,
        focus_mode=FocusMode.ADAPTIVE
    )
    
    query5 = "transformer architecture attention"
    
    print(f"\n💭 Query: {query5}")
    print(f"📚 Total Documents: {len(large_docs)}")
    
    result5 = attn5.attend(query5, large_docs, top_k=3)
    
    print(f"\n⚡ Sparse Attention Results:")
    print(f"   Only scored relevant subset")
    print(f"   Execution Time: {result5.execution_time_ms:.1f}ms")
    
    print(f"\n🎯 Top Results:")
    for i, item in enumerate(result5.attended_items, 1):
        print(f"   {i}. {item.item_id}: {item.score:.3f}")
    
    # Summary
    print("\n" + "=" * 80)
    print("ATTENTION MECHANISM PATTERNS SUMMARY")
    print("=" * 80)
    print("""
Key Benefits:
1. Selective Focus: Process only relevant information
2. Scalability: Handle large contexts efficiently
3. Interpretability: Attention scores show reasoning
4. Quality: Focus improves response accuracy
5. Efficiency: Reduce computational cost

Attention Types:
1. Cross-Attention: Query attending to context
   - Use: Question answering, retrieval
   - Pattern: Query → Context scoring

2. Self-Attention: Internal relationships
   - Use: Document clustering, summarization
   - Pattern: Items → Mutual relevance

3. Multi-Head Attention: Multiple perspectives
   - Use: Complex analysis, diverse views
   - Pattern: Parallel attention heads → Combine

4. Sparse Attention: Subset focus
   - Use: Large-scale processing
   - Pattern: Filter → Attend to subset

5. Hierarchical Attention: Multi-level focus
   - Use: Structured documents
   - Pattern: Coarse → Fine attention

Focus Modes:
- Narrow: Highly selective (top-1 or top-3)
- Broad: Consider wider context (top-10+)
- Adaptive: Adjust based on task
- Distributed: Spread attention evenly

Scoring Methods:
1. LLM-based: Relevance scoring by LLM
2. Embedding-based: Cosine similarity
3. Keyword: Simple word overlap
4. Learned: Trained attention weights

Applications:
- Long document Q&A
- Multi-document synthesis
- Memory retrieval
- Context selection
- Feature importance
- Action space filtering

Implementation Patterns:
1. Score → Normalize → Select top-k
2. Multiple heads → Average/Weighted combine
3. Hierarchical: Sentence → Word attention
4. Causal: Attend only to past
5. Bidirectional: Full context attention

Best Practices:
1. Choose appropriate top-k
2. Normalize attention scores (softmax)
3. Multi-head for complex tasks
4. Sparse for large contexts
5. Cache attention scores
6. Visualize attention patterns
7. Combine with retrieval

Production Considerations:
- Caching scored items
- Parallel scoring
- Incremental attention updates
- Attention visualization
- Performance profiling
- Memory management
- Score normalization methods

Comparison with Related Patterns:
- vs. Retrieval: Attention vs search
- vs. Filtering: Soft vs hard selection
- vs. Ranking: Continuous scores vs discrete order
- vs. Memory: Attention for retrieval

Attention mechanisms are fundamental for efficient processing
of large contexts, enabling agents to focus computational
resources where they matter most for the task at hand.
""")


if __name__ == "__main__":
    demonstrate_attention_mechanisms()

    

Agentic Design Pattern: Attention Mechanisms

Pattern 057: Attention Mechanisms – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Attention Mechanisms pattern implementation.

---

Pattern Description

Pattern 057: Attention Mechanism Patterns

Description: Attention mechanisms enable agents to selectively focus on the most relevant parts of large inputs, contexts, or action spaces. Inspired by human attention and transformer architectures, these patterns help agents efficiently process information by allocating computational resources where they matter most.

Components: 1. Attention Scorer: Calculates relevance scores 2. Context Selector: Chooses relevant information 3. Focus Manager: Manages attention allocation 4. Multi-Head Attention: Multiple attention perspectives 5. Self-Attention: Internal context relationships

Use Cases: – Long document processing – Multi-source information synthesis – Large context windows – Memory retrieval from extensive history – Action selection in large action spaces – Feature selection in complex data

LangChain Implementation: Implements attention scoring and selection mechanisms using LLMs to identify and focus on relevant portions of context.

---

File Statistics

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

---

Module Documentation

"""
Pattern 057: Attention Mechanism Patterns

Description:
    Attention mechanisms enable agents to selectively focus on the most relevant
    parts of large inputs, contexts, or action spaces. Inspired by human attention
    and transformer architectures, these patterns help agents efficiently process
    information by allocating computational resources where they matter most.

Components:
    1. Attention Scorer: Calculates relevance scores
    2. Context Selector: Chooses relevant information
    3. Focus Manager: Manages attention allocation
    4. Multi-Head Attention: Multiple attention perspectives
    5. Self-Attention: Internal context relationships

Use Cases:
    - Long document processing
    - Multi-source information synthesis
    - Large context windows
    - Memory retrieval from extensive history
    - Action selection in large action spaces
    - Feature selection in complex data

LangChain Implementation:
    Implements attention scoring and selection mechanisms using LLMs
    to identify and focus on relevant portions of context.
"""

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

---

Import Statements

Standard Library Imports

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

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

LangChain Imports

from langchain_openai.ChatOpenAI

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

Other Imports

from dotenv.load_dotenv
from math

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: AttentionType

Lines 46-52

Inherits from: Enum

Purpose:

Types of attention mechanisms

---

Class: FocusMode

Lines 55-60

Inherits from: Enum

Purpose:

Modes of attention focus

---

Class: AttentionScore

Lines 64-72

Purpose:

Attention score for an item

Methods:

__repr__(self)

Lines 71-72

Purpose: Repr operation.

Parameters:

Returns: str

---

Class: AttentionHead

Lines 76-84

Purpose:

Single attention head with specific perspective

Methods:

top_items(self, k)

Lines 83-84

Decorators: property

Purpose: Top Items operation.

Parameters: – k: K

Returns: <ast.Subscript object at 0x000001A06D70C4D0>

---

Class: AttentionResult

Lines 88-115

Purpose:

Result from attention mechanism

Methods:

coverage(self)

Lines 100-102

Decorators: property

Purpose:

Percentage of items attended

Parameters:

Returns: float

to_dict(self)

Lines 104-115

Purpose: To Dict operation.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D15A150>

---

Class: AttentionMechanism

Lines 118-431

Purpose:

Implements attention mechanisms for selective focus.

Features:
1. Query-based attention scoring
2. Multi-head attention for diverse perspectives
3. Sparse attention for efficiency
4. Hierarchical attention for structured data
5. Adaptive focus modes

Methods:

__init__(self, attention_type, focus_mode, num_heads, temperature)

Lines 130-152

Purpose: Initialize the class instance with required parameters.

Parameters: – attention_type: Attention type – focus_mode: Focus mode – num_heads: Num heads – temperature: Sampling temperature (0-1)

_score_relevance(self, query, item, item_id, perspective)

Lines 154-191

Purpose:

Score relevance of item to query

Parameters: – query: Input query or question – item: Item – item_id: Item id – perspective: Perspective

Returns: AttentionScore

_cross_attention(self, query, context_items, top_k)

Lines 193-215

Purpose:

Cross-attention: query attending to context

Parameters: – query: Input query or question – context_items: Context items – top_k: Top k

Returns: <ast.Subscript object at 0x000001A06D15D2D0>

_self_attention(self, items, top_k)

Lines 217-261

Purpose:

Self-attention: items attending to each other

Parameters: – items: Items – top_k: Top k

Returns: <ast.Subscript object at 0x000001A06D7F7F90>

_multi_head_attention(self, query, context_items, perspectives, top_k)

Lines 263-321

Purpose:

Multi-head attention: multiple perspectives

Parameters: – query: Input query or question – context_items: Context items – perspectives: Perspectives – top_k: Top k

Returns: <ast.Subscript object at 0x000001A06D75A050>

_sparse_attention(self, query, context_items, top_k)

Lines 323-346

Purpose:

Sparse attention: only attend to subset

Parameters: – query: Input query or question – context_items: Context items – top_k: Top k

Returns: <ast.Subscript object at 0x000001A06D65AD50>

attend(self, query, context_items, top_k)

Lines 348-396

Purpose:

Apply attention mechanism

Parameters: – query: Input query or question – context_items: Context items – top_k: Top k

Returns: AttentionResult

focused_response(self, query, context_items, top_k)

Lines 398-431

Purpose:

Generate response using attended context

Parameters: – query: Input query or question – context_items: Context items – top_k: Top k

Returns: <ast.Subscript object at 0x000001A06D62BC10>

---

Functions

demonstrate_attention_mechanisms()

Lines 434-660

Purpose:

Demonstrate Attention Mechanism patterns

---

Key Concepts Summary

Pattern Overview

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

Core Components

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

When to Use

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

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

---

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

---

LangChain Components Used

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

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

---

Execution Flow

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

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

---

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

---

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

---

Related Patterns

• Pattern 056
• Pattern 058

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 Attention Mechanisms 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: 057_attention_mechanisms.py
Total Lines: 665
Pattern Category: AI Agent Design Pattern #057

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 056: Blackboard System

Description:
    The Blackboard System pattern implements a shared workspace (blackboard) where
    multiple independent knowledge sources (agents) collaboratively solve complex
    problems. Each knowledge source monitors the blackboard and contributes when
    it has relevant expertise, enabling opportunistic problem-solving through
    emergent collaboration.

Components:
    1. Blackboard: Shared data structure storing problem state
    2. Knowledge Sources (KS): Independent expert agents
    3. Control Module: Coordinates knowledge source activation
    4. Problem Space: Hierarchical representation of problem
    5. Solution Assembly: Combines partial solutions

Use Cases:
    - Complex multi-disciplinary problems
    - Speech and image recognition systems
    - Medical diagnosis with multiple specialists
    - Software architecture design
    - Scientific research synthesis
    - Strategic planning and decision-making

LangChain Implementation:
    Uses a shared state dictionary as blackboard, with multiple LLM-based
    knowledge sources that opportunistically contribute based on current state.
"""

import os
import time
from typing import Dict, Any, List, Optional, Callable, Set
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import json

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 KSPriority(Enum):
    """Priority levels for knowledge sources"""
    CRITICAL = 5
    HIGH = 4
    MEDIUM = 3
    LOW = 2
    MINIMAL = 1


class BlackboardState(Enum):
    """States of blackboard processing"""
    INITIALIZED = "initialized"
    GATHERING_INFO = "gathering_info"
    ANALYZING = "analyzing"
    GENERATING_SOLUTIONS = "generating_solutions"
    REFINING = "refining"
    COMPLETE = "complete"


@dataclass
class BlackboardEntry:
    """Entry on the blackboard"""
    entry_id: str
    layer: str  # hypothesis, evidence, solution, etc.
    content: str
    contributor: str  # which KS added this
    confidence: float  # 0.0-1.0
    timestamp: datetime
    dependencies: List[str] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "id": self.entry_id,
            "layer": self.layer,
            "content": self.content[:100] + "..." if len(self.content) > 100 else self.content,
            "contributor": self.contributor,
            "confidence": f"{self.confidence:.2f}",
            "timestamp": self.timestamp.isoformat()
        }


@dataclass
class KSActivation:
    """Knowledge source activation record"""
    ks_name: str
    trigger_condition: str
    contribution: str
    confidence: float
    timestamp: datetime


@dataclass
class Blackboard:
    """Shared workspace for collaborative problem-solving"""
    problem: str
    state: BlackboardState
    entries: List[BlackboardEntry] = field(default_factory=list)
    activations: List[KSActivation] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    
    def add_entry(self, entry: BlackboardEntry):
        """Add entry to blackboard"""
        self.entries.append(entry)
    
    def get_entries_by_layer(self, layer: str) -> List[BlackboardEntry]:
        """Get all entries in a specific layer"""
        return [e for e in self.entries if e.layer == layer]
    
    def get_latest_entries(self, n: int = 5) -> List[BlackboardEntry]:
        """Get most recent entries"""
        return sorted(self.entries, key=lambda x: x.timestamp, reverse=True)[:n]
    
    def record_activation(self, activation: KSActivation):
        """Record knowledge source activation"""
        self.activations.append(activation)
    
    def get_state_summary(self) -> str:
        """Get summary of current blackboard state"""
        layer_counts = {}
        for entry in self.entries:
            layer_counts[entry.layer] = layer_counts.get(entry.layer, 0) + 1
        
        summary = f"Problem: {self.problem}\n"
        summary += f"State: {self.state.value}\n"
        summary += f"Total Entries: {len(self.entries)}\n"
        summary += "Entries by Layer:\n"
        for layer, count in layer_counts.items():
            summary += f"  - {layer}: {count}\n"
        
        return summary


class KnowledgeSource:
    """Independent expert agent that contributes to blackboard"""
    
    def __init__(
        self,
        name: str,
        expertise: str,
        trigger_conditions: List[str],
        priority: KSPriority,
        llm: ChatOpenAI,
        system_prompt: str
    ):
        self.name = name
        self.expertise = expertise
        self.trigger_conditions = trigger_conditions
        self.priority = priority
        self.llm = llm
        self.system_prompt = system_prompt
        self.activation_count = 0
    
    def should_activate(self, blackboard: Blackboard) -> tuple[bool, str]:
        """Determine if this KS should contribute"""
        
        # Check if any trigger conditions are met
        bb_summary = blackboard.get_state_summary()
        latest_entries = blackboard.get_latest_entries(3)
        
        for condition in self.trigger_conditions:
            # Simple keyword matching (could be more sophisticated)
            if condition.lower() in bb_summary.lower():
                return True, f"Triggered by: {condition}"
            
            # Check recent entries
            for entry in latest_entries:
                if condition.lower() in entry.content.lower():
                    return True, f"Triggered by entry: {entry.entry_id}"
        
        return False, ""
    
    def contribute(self, blackboard: Blackboard) -> Optional[BlackboardEntry]:
        """Generate contribution to blackboard"""
        
        # Get context from blackboard
        recent_entries = blackboard.get_latest_entries(5)
        context = "\n".join([
            f"[{e.layer}] {e.content[:100]}"
            for e in recent_entries
        ])
        
        # Generate contribution
        prompt = ChatPromptTemplate.from_messages([
            ("system", self.system_prompt),
            ("user", """Problem: {problem}

Current Blackboard State:
{context}

Based on your expertise in {expertise}, provide your contribution.
Be specific and build on existing information.

Your contribution:""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        
        try:
            contribution = chain.invoke({
                "problem": blackboard.problem,
                "context": context or "Empty blackboard",
                "expertise": self.expertise
            })
            
            # Determine layer for contribution
            layer = "analysis"  # Default
            if "hypothesis" in contribution.lower() or "theory" in contribution.lower():
                layer = "hypothesis"
            elif "evidence" in contribution.lower() or "data" in contribution.lower():
                layer = "evidence"
            elif "solution" in contribution.lower() or "approach" in contribution.lower():
                layer = "solution"
            elif "recommend" in contribution.lower():
                layer = "recommendation"
            
            # Create entry
            entry = BlackboardEntry(
                entry_id=f"{self.name}_{len(blackboard.entries)}",
                layer=layer,
                content=contribution,
                contributor=self.name,
                confidence=0.7 + (self.priority.value * 0.05),
                timestamp=datetime.now()
            )
            
            self.activation_count += 1
            
            return entry
            
        except Exception as e:
            print(f"Error in {self.name} contribution: {str(e)}")
            return None


class BlackboardControl:
    """Controls knowledge source activation and coordination"""
    
    def __init__(self, problem: str):
        self.blackboard = Blackboard(
            problem=problem,
            state=BlackboardState.INITIALIZED
        )
        self.knowledge_sources: List[KnowledgeSource] = []
        self.max_iterations = 10
        self.current_iteration = 0
    
    def register_knowledge_source(self, ks: KnowledgeSource):
        """Register a knowledge source"""
        self.knowledge_sources.append(ks)
    
    def _select_next_ks(self) -> Optional[KnowledgeSource]:
        """Select next knowledge source to activate"""
        
        # Find all KS that can contribute
        candidates = []
        
        for ks in self.knowledge_sources:
            should_activate, trigger = ks.should_activate(self.blackboard)
            if should_activate:
                candidates.append((ks, trigger))
        
        if not candidates:
            return None
        
        # Sort by priority and activation count (prefer high priority, low activation)
        candidates.sort(
            key=lambda x: (x[0].priority.value, -x[0].activation_count),
            reverse=True
        )
        
        return candidates[0][0]
    
    def _update_state(self):
        """Update blackboard state based on progress"""
        
        entry_count = len(self.blackboard.entries)
        
        if entry_count == 0:
            self.blackboard.state = BlackboardState.INITIALIZED
        elif entry_count < 3:
            self.blackboard.state = BlackboardState.GATHERING_INFO
        elif entry_count < 6:
            self.blackboard.state = BlackboardState.ANALYZING
        elif entry_count  Blackboard:
        """Execute blackboard problem-solving cycle"""
        
        if max_iterations:
            self.max_iterations = max_iterations
        
        print(f"\n🎯 Starting blackboard problem-solving for: {self.blackboard.problem}")
        
        while self.current_iteration = 2 and len(recommendations) >= 1:
                print("\nSufficient solutions generated")
                self.blackboard.state = BlackboardState.COMPLETE
                break
        
        return self.blackboard


def create_default_knowledge_sources() -> List[KnowledgeSource]:
    """Create default set of knowledge sources"""
    
    knowledge_sources = []
    
    # Problem Analyzer
    ks1 = KnowledgeSource(
        name="ProblemAnalyzer",
        expertise="problem decomposition and analysis",
        trigger_conditions=["initialized", "problem"],
        priority=KSPriority.CRITICAL,
        llm=ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.3),
        system_prompt="""You are a problem analysis expert. Break down complex problems
into components, identify key challenges, and frame the problem clearly."""
    )
    knowledge_sources.append(ks1)
    
    # Hypothesis Generator
    ks2 = KnowledgeSource(
        name="HypothesisGenerator",
        expertise="generating hypotheses and theories",
        trigger_conditions=["analysis", "decomp", "component"],
        priority=KSPriority.HIGH,
        llm=ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.7),
        system_prompt="""You are a creative thinker who generates hypotheses and theories.
Propose possible explanations and approaches based on the analysis."""
    )
    knowledge_sources.append(ks2)
    
    # Evidence Collector
    ks3 = KnowledgeSource(
        name="EvidenceCollector",
        expertise="gathering and evaluating evidence",
        trigger_conditions=["hypothesis", "theory", "claim"],
        priority=KSPriority.MEDIUM,
        llm=ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.2),
        system_prompt="""You are an evidence expert who evaluates claims and hypotheses.
Identify what evidence would support or refute the hypotheses."""
    )
    knowledge_sources.append(ks3)
    
    # Solution Designer
    ks4 = KnowledgeSource(
        name="SolutionDesigner",
        expertise="designing practical solutions",
        trigger_conditions=["evidence", "hypothesis", "analyzing"],
        priority=KSPriority.HIGH,
        llm=ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.5),
        system_prompt="""You are a solution architect who designs practical approaches.
Create concrete, actionable solutions based on the analysis and evidence."""
    )
    knowledge_sources.append(ks4)
    
    # Critic
    ks5 = KnowledgeSource(
        name="Critic",
        expertise="critical evaluation and refinement",
        trigger_conditions=["solution", "approach", "recommend"],
        priority=KSPriority.MEDIUM,
        llm=ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.4),
        system_prompt="""You are a critical evaluator who identifies weaknesses and risks.
Critique proposed solutions and suggest improvements."""
    )
    knowledge_sources.append(ks5)
    
    # Synthesizer
    ks6 = KnowledgeSource(
        name="Synthesizer",
        expertise="synthesis and final recommendations",
        trigger_conditions=["solution", "refining"],
        priority=KSPriority.CRITICAL,
        llm=ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.3),
        system_prompt="""You are a synthesizer who combines multiple perspectives.
Provide final recommendations that integrate all contributions."""
    )
    knowledge_sources.append(ks6)
    
    return knowledge_sources


def demonstrate_blackboard_system():
    """Demonstrate Blackboard System pattern"""
    
    print("=" * 80)
    print("PATTERN 056: BLACKBOARD SYSTEM DEMONSTRATION")
    print("=" * 80)
    print("\nCollaborative problem-solving through shared workspace\n")
    
    # Test 1: Technical problem
    print("\n" + "=" * 80)
    print("TEST 1: Technical Problem Solving")
    print("=" * 80)
    
    problem1 = "How can we improve the performance of a slow web application?"
    
    control1 = BlackboardControl(problem1)
    
    # Register knowledge sources
    for ks in create_default_knowledge_sources():
        control1.register_knowledge_source(ks)
    
    print(f"\nRegistered {len(control1.knowledge_sources)} knowledge sources:")
    for ks in control1.knowledge_sources:
        print(f"  - {ks.name}: {ks.expertise}")
    
    # Solve problem
    result1 = control1.solve(max_iterations=8)
    
    print(f"\n📊 Final Blackboard State:")
    print(f"   Total Entries: {len(result1.entries)}")
    print(f"   Total Activations: {len(result1.activations)}")
    print(f"   Final State: {result1.state.value}")
    
    # Show contributions by layer
    print(f"\n📚 Entries by Layer:")
    layers = {}
    for entry in result1.entries:
        if entry.layer not in layers:
            layers[entry.layer] = []
        layers[entry.layer].append(entry)
    
    for layer, entries in layers.items():
        print(f"\n   {layer.upper()} ({len(entries)} entries):")
        for entry in entries[:2]:  # Show first 2 per layer
            print(f"      [{entry.contributor}] {entry.content[:100]}...")
    
    # Test 2: Strategic planning problem
    print("\n" + "=" * 80)
    print("TEST 2: Strategic Planning Problem")
    print("=" * 80)
    
    problem2 = "What strategy should a startup use to compete with established companies?"
    
    control2 = BlackboardControl(problem2)
    for ks in create_default_knowledge_sources():
        control2.register_knowledge_source(ks)
    
    result2 = control2.solve(max_iterations=6)
    
    print(f"\n🎯 Problem Solving Summary:")
    print(f"   Problem: {problem2}")
    print(f"   Iterations: {control2.current_iteration}")
    print(f"   Knowledge Sources Activated:")
    
    # Count activations per KS
    ks_counts = {}
    for activation in result2.activations:
        ks_counts[activation.ks_name] = ks_counts.get(activation.ks_name, 0) + 1
    
    for ks_name, count in sorted(ks_counts.items(), key=lambda x: x[1], reverse=True):
        print(f"      {ks_name}: {count} times")
    
    # Show final recommendation
    recommendations = result2.get_entries_by_layer("recommendation")
    if recommendations:
        print(f"\n💡 Final Recommendation:")
        print(f"   {recommendations[-1].content[:300]}...")
    
    # Test 3: Show collaboration pattern
    print("\n" + "=" * 80)
    print("TEST 3: Collaboration Pattern Analysis")
    print("=" * 80)
    
    print(f"\n🔄 Knowledge Source Activation Sequence:")
    for i, activation in enumerate(result2.activations, 1):
        print(f"   {i}. {activation.ks_name} (confidence: {activation.confidence:.2f})")
    
    print(f"\n🏗️  Blackboard Evolution:")
    for i in range(0, len(result2.entries), 2):
        entry = result2.entries[i]
        print(f"   Entry {i+1}: [{entry.layer}] by {entry.contributor}")
    
    # Summary
    print("\n" + "=" * 80)
    print("BLACKBOARD SYSTEM PATTERN SUMMARY")
    print("=" * 80)
    print("""
Key Benefits:
1. Opportunistic Problem-Solving: KS contribute when relevant
2. Emergent Collaboration: No predefined workflow
3. Modularity: Independent knowledge sources
4. Incremental Progress: Builds solution step-by-step
5. Transparency: All contributions visible on blackboard

System Components:
1. Blackboard: Shared workspace with layered structure
2. Knowledge Sources: Independent expert agents
3. Control Module: Coordinates KS activation
4. Activation Conditions: Triggers for KS contributions
5. Priority System: Manages contribution order

Blackboard Layers:
- Hypothesis: Theories and possible explanations
- Evidence: Supporting or refuting data
- Analysis: Problem decomposition and insights
- Solution: Concrete approaches and methods
- Recommendation: Final synthesized advice

Knowledge Source Types:
1. Problem Analyzer: Breaks down problems
2. Hypothesis Generator: Creates theories
3. Evidence Collector: Gathers supporting data
4. Solution Designer: Designs approaches
5. Critic: Evaluates and refines
6. Synthesizer: Combines perspectives

Control Strategies:
1. Priority-Based: High-priority KS activate first
2. Trigger-Based: KS activate when conditions met
3. Load-Balanced: Distribute activations evenly
4. State-Driven: Blackboard state guides activation

Activation Patterns:
- Initial Analysis: Problem decomposition
- Hypothesis Generation: Multiple theories
- Evidence Collection: Supporting data
- Solution Design: Practical approaches
- Critical Evaluation: Refinement
- Synthesis: Final recommendations

Use Cases:
- Complex multi-disciplinary problems
- Medical diagnosis systems
- Scientific research synthesis
- Software architecture design
- Strategic planning
- Signal processing and recognition

Best Practices:
1. Clear KS expertise boundaries
2. Well-defined trigger conditions
3. Appropriate priority levels
4. Layered blackboard structure
5. Limit iteration count
6. Record all contributions
7. Final synthesis step

Production Considerations:
- Blackboard persistence
- Concurrent KS activation
- Conflict resolution
- Termination criteria
- Performance optimization
- Scalability with many KS
- Debugging and tracing

Comparison with Related Patterns:
- vs. Multi-Agent: Shared state vs communication
- vs. Pipeline: Opportunistic vs sequential
- vs. Mixture of Agents: Collaboration vs aggregation
- vs. Ensemble: Emergent vs predetermined combination

The Blackboard System pattern excels at complex problems requiring
multiple perspectives and emergent, opportunistic collaboration where
the solution path is not predetermined.
""")


if __name__ == "__main__":
    demonstrate_blackboard_system()

    

Agentic Design Pattern: Blackboard System

Pattern 056: Blackboard System – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Blackboard System pattern implementation.

---

Pattern Description

Pattern 056: Blackboard System

Description: The Blackboard System pattern implements a shared workspace (blackboard) where multiple independent knowledge sources (agents) collaboratively solve complex problems. Each knowledge source monitors the blackboard and contributes when it has relevant expertise, enabling opportunistic problem-solving through emergent collaboration.

Components: 1. Blackboard: Shared data structure storing problem state 2. Knowledge Sources (KS): Independent expert agents 3. Control Module: Coordinates knowledge source activation 4. Problem Space: Hierarchical representation of problem 5. Solution Assembly: Combines partial solutions

Use Cases: – Complex multi-disciplinary problems – Speech and image recognition systems – Medical diagnosis with multiple specialists – Software architecture design – Scientific research synthesis – Strategic planning and decision-making

LangChain Implementation: Uses a shared state dictionary as blackboard, with multiple LLM-based knowledge sources that opportunistically contribute based on current state.

---

File Statistics

• Total Lines: 611
• Classes: 7
• Functions: 2
• Constants: 0

---

Module Documentation

"""
Pattern 056: Blackboard System

Description:
    The Blackboard System pattern implements a shared workspace (blackboard) where
    multiple independent knowledge sources (agents) collaboratively solve complex
    problems. Each knowledge source monitors the blackboard and contributes when
    it has relevant expertise, enabling opportunistic problem-solving through
    emergent collaboration.

Components:
    1. Blackboard: Shared data structure storing problem state
    2. Knowledge Sources (KS): Independent expert agents
    3. Control Module: Coordinates knowledge source activation
    4. Problem Space: Hierarchical representation of problem
    5. Solution Assembly: Combines partial solutions

Use Cases:
    - Complex multi-disciplinary problems
    - Speech and image recognition systems
    - Medical diagnosis with multiple specialists
    - Software architecture design
    - Scientific research synthesis
    - Strategic planning and decision-making

LangChain Implementation:
    Uses a shared state dictionary as blackboard, with multiple LLM-based
    knowledge sources that opportunistically contribute based on current state.
"""

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

---

Import Statements

Standard Library Imports

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

Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – datetime.datetime: Date and time handling – enum.Enum: Enumeration support – json: JSON serialization/deserialization – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – time: Time-related functions – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.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) – 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: KSPriority

Lines 47-53

Inherits from: Enum

Purpose:

Priority levels for knowledge sources

---

Class: BlackboardState

Lines 56-63

Inherits from: Enum

Purpose:

States of blackboard processing

---

Class: BlackboardEntry

Lines 67-86

Purpose:

Entry on the blackboard

Methods:

to_dict(self)

Lines 78-86

Purpose: To Dict operation.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D7664D0>

---

Class: KSActivation

Lines 90-96

Purpose:

Knowledge source activation record

---

Class: Blackboard

Lines 100-137

Purpose:

Shared workspace for collaborative problem-solving

Methods:

add_entry(self, entry)

Lines 108-110

Purpose:

Add entry to blackboard

Parameters: – entry: Entry

get_entries_by_layer(self, layer)

Lines 112-114

Purpose:

Get all entries in a specific layer

Parameters: – layer: Layer

Returns: <ast.Subscript object at 0x000001A06D6F48D0>

get_latest_entries(self, n)

Lines 116-118

Purpose:

Get most recent entries

Parameters: – n: N

Returns: <ast.Subscript object at 0x000001A06D6F5F90>

record_activation(self, activation)

Lines 120-122

Purpose:

Record knowledge source activation

Parameters: – activation: Activation

get_state_summary(self)

Lines 124-137

Purpose:

Get summary of current blackboard state

Parameters:

Returns: str

---

Class: KnowledgeSource

Lines 140-239

Purpose:

Independent expert agent that contributes to blackboard

Methods:

__init__(self, name, expertise, trigger_conditions, priority, llm, system_prompt)

Lines 143-158

Purpose: Initialize the class instance with required parameters.

Parameters: – name: Name – expertise: Expertise – trigger_conditions: Trigger conditions – priority: Priority – llm: Language model – system_prompt: Prompt template or text

should_activate(self, blackboard)

Lines 160-177

Purpose:

Determine if this KS should contribute

Parameters: – blackboard: Blackboard

Returns: <ast.Subscript object at 0x000001A06D16D310>

contribute(self, blackboard)

Lines 179-239

Purpose:

Generate contribution to blackboard

Parameters: – blackboard: Blackboard

Returns: <ast.Subscript object at 0x000001A06D626810>

---

Class: BlackboardControl

Lines 242-348

Purpose:

Controls knowledge source activation and coordination

Methods:

__init__(self, problem)

Lines 245-252

Purpose: Initialize the class instance with required parameters.

Parameters: – problem: Problem

register_knowledge_source(self, ks)

Lines 254-256

Purpose:

Register a knowledge source

Parameters: – ks: Ks

_select_next_ks(self)

Lines 258-278

Purpose:

Select next knowledge source to activate

Parameters:

Returns: <ast.Subscript object at 0x000001A06D60B210>

_update_state(self)

Lines 280-294

Purpose:

Update blackboard state based on progress

Parameters:

solve(self, max_iterations)

Lines 296-348

Purpose:

Execute blackboard problem-solving cycle

Parameters: – max_iterations: Max iterations

Returns: Blackboard

---

Functions

create_default_knowledge_sources()

Lines 351-428

Purpose:

Create default set of knowledge sources

Returns: <ast.Subscript object at 0x000001A06D60CC90>

demonstrate_blackboard_system()

Lines 431-606

Purpose:

Demonstrate Blackboard System pattern

---

Key Concepts Summary

Pattern Overview

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

Core Components

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

When to Use

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

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

---

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

---

LangChain Components Used

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

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

---

Execution Flow

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

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

---

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

---

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

---

Related Patterns

• Pattern 055
• Pattern 057

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 Blackboard System 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: 056_blackboard_system.py
Total Lines: 611
Pattern Category: AI Agent Design Pattern #056

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 055: Cognitive Architecture

Description:
    The Cognitive Architecture pattern models human-like cognitive processes in AI agents,
    including perception, attention, working memory, reasoning, planning, and action.
    This creates agents that process information in stages similar to human cognition,
    enabling more sophisticated and contextual decision-making.

Components:
    1. Perception Module: Processes and interprets inputs
    2. Attention Mechanism: Focuses on relevant information
    3. Working Memory: Maintains current context
    4. Long-Term Memory: Stores knowledge and experiences
    5. Reasoning Engine: Logical inference and problem-solving
    6. Planning Module: Goal-directed action planning
    7. Action Selection: Chooses and executes actions

Use Cases:
    - Complex decision-making systems
    - Autonomous agents with human-like reasoning
    - Educational tutoring systems
    - Game AI with realistic behavior
    - Personal assistants with context awareness
    - Research and analysis agents

LangChain Implementation:
    Implements a multi-stage cognitive pipeline using LangChain components
    to simulate perception, memory, reasoning, and action selection.
"""

import os
import time
from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import deque

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 CognitiveState(Enum):
    """States in cognitive processing cycle"""
    PERCEIVING = "perceiving"  # Processing input
    ATTENDING = "attending"  # Focusing attention
    ENCODING = "encoding"  # Storing in memory
    REASONING = "reasoning"  # Logical processing
    PLANNING = "planning"  # Creating action plan
    ACTING = "acting"  # Executing action
    REFLECTING = "reflecting"  # Evaluating outcome


class AttentionFocus(Enum):
    """What the agent is focusing on"""
    GOAL = "goal"  # Primary objective
    PROBLEM = "problem"  # Current problem
    CONTEXT = "context"  # Surrounding context
    MEMORY = "memory"  # Past experiences
    ACTION = "action"  # Potential actions


@dataclass
class PerceptualInput:
    """Input perceived by the agent"""
    raw_input: str
    modality: str  # "text", "visual", "audio", etc.
    timestamp: datetime
    features: Dict[str, Any] = field(default_factory=dict)
    
    def to_text(self) -> str:
        return f"[{self.modality}] {self.raw_input}"


@dataclass
class WorkingMemoryItem:
    """Item in working memory"""
    content: str
    relevance: float  # 0.0-1.0
    timestamp: datetime
    source: str  # perception, reasoning, memory
    
    def __str__(self) -> str:
        return f"{self.content} (rel: {self.relevance:.2f})"


@dataclass
class LongTermMemory:
    """Long-term memory store"""
    facts: List[str] = field(default_factory=list)
    experiences: List[str] = field(default_factory=list)
    skills: List[str] = field(default_factory=list)
    goals: List[str] = field(default_factory=list)
    
    def add_fact(self, fact: str):
        if fact not in self.facts:
            self.facts.append(fact)
    
    def add_experience(self, experience: str):
        self.experiences.append(experience)
        # Keep recent experiences
        if len(self.experiences) > 50:
            self.experiences = self.experiences[-50:]
    
    def recall(self, query: str, k: int = 3) -> List[str]:
        """Recall relevant memories"""
        # Simple keyword matching (could use embeddings in production)
        query_lower = query.lower()
        
        relevant = []
        for fact in self.facts:
            if any(word in fact.lower() for word in query_lower.split()):
                relevant.append(("fact", fact))
        
        for exp in self.experiences[-10:]:  # Recent experiences
            if any(word in exp.lower() for word in query_lower.split()):
                relevant.append(("experience", exp))
        
        return relevant[:k]


@dataclass
class ReasoningResult:
    """Result from reasoning process"""
    conclusion: str
    reasoning_steps: List[str]
    confidence: float
    used_memories: List[str]


@dataclass
class ActionPlan:
    """Plan for achieving goal"""
    goal: str
    steps: List[str]
    expected_outcome: str
    confidence: float


@dataclass
class CognitiveResult:
    """Complete result from cognitive processing"""
    input: str
    perception: str
    attention_focus: AttentionFocus
    reasoning: ReasoningResult
    plan: ActionPlan
    action: str
    reflection: str
    cognitive_trace: List[Tuple[CognitiveState, str]]
    execution_time_ms: float
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "input": self.input[:100] + "..." if len(self.input) > 100 else self.input,
            "attention": self.attention_focus.value,
            "conclusion": self.reasoning.conclusion[:100] + "...",
            "plan_steps": len(self.plan.steps),
            "action": self.action[:100] + "..." if len(self.action) > 100 else self.action,
            "execution_time_ms": f"{self.execution_time_ms:.1f}"
        }


class CognitiveAgent:
    """
    Agent with human-like cognitive architecture.
    
    Implements:
    1. Perception: Input processing and feature extraction
    2. Attention: Selective focus on relevant information
    3. Working Memory: Temporary context storage (limited capacity)
    4. Long-Term Memory: Persistent knowledge storage
    5. Reasoning: Logical inference and problem-solving
    6. Planning: Goal-directed action planning
    7. Action: Selection and execution
    8. Reflection: Learning from outcomes
    """
    
    def __init__(
        self,
        working_memory_capacity: int = 7,  # Miller's 7±2
        temperature: float = 0.7
    ):
        self.working_memory_capacity = working_memory_capacity
        self.temperature = temperature
        
        # Cognitive components
        self.llm = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=temperature)
        
        # Memory systems
        self.working_memory: deque = deque(maxlen=working_memory_capacity)
        self.long_term_memory = LongTermMemory()
        
        # Current state
        self.current_goal: Optional[str] = None
        self.attention_focus: AttentionFocus = AttentionFocus.CONTEXT
        
        # Cognitive trace for transparency
        self.cognitive_trace: List[Tuple[CognitiveState, str]] = []
    
    def _perceive(self, input_text: str) -> Tuple[str, PerceptualInput]:
        """Perception: Process and interpret input"""
        
        self.cognitive_trace.append((CognitiveState.PERCEIVING, "Processing input"))
        
        # Create perceptual input
        perceptual_input = PerceptualInput(
            raw_input=input_text,
            modality="text",
            timestamp=datetime.now()
        )
        
        # Extract features and intent
        perception_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are processing sensory input. Extract:
1. Main topic/subject
2. Intent (question, request, statement, etc.)
3. Key entities mentioned
4. Emotional tone

Format: Topic: X | Intent: Y | Entities: Z | Tone: W"""),
            ("user", "{input}")
        ])
        
        chain = perception_prompt | self.llm | StrOutputParser()
        perception_result = chain.invoke({"input": input_text})
        
        self.cognitive_trace.append((CognitiveState.PERCEIVING, perception_result))
        
        return perception_result, perceptual_input
    
    def _attend(self, perception: str, input_text: str) -> AttentionFocus:
        """Attention: Focus on most relevant aspect"""
        
        self.cognitive_trace.append((CognitiveState.ATTENDING, "Directing attention"))
        
        # Determine what to focus on
        attention_prompt = ChatPromptTemplate.from_messages([
            ("system", """Determine attentional focus for this input.
Choose: goal, problem, context, memory, or action

Respond with just the focus word."""),
            ("user", "Input: {input}\nPerception: {perception}\nFocus:")
        ])
        
        chain = attention_prompt | self.llm | StrOutputParser()
        focus_str = chain.invoke({
            "input": input_text,
            "perception": perception
        }).strip().lower()
        
        # Map to enum
        focus_map = {
            "goal": AttentionFocus.GOAL,
            "problem": AttentionFocus.PROBLEM,
            "context": AttentionFocus.CONTEXT,
            "memory": AttentionFocus.MEMORY,
            "action": AttentionFocus.ACTION
        }
        
        self.attention_focus = focus_map.get(focus_str, AttentionFocus.CONTEXT)
        self.cognitive_trace.append((CognitiveState.ATTENDING, f"Focus: {self.attention_focus.value}"))
        
        return self.attention_focus
    
    def _encode_working_memory(self, content: str, relevance: float, source: str):
        """Encode information into working memory"""
        
        item = WorkingMemoryItem(
            content=content,
            relevance=relevance,
            timestamp=datetime.now(),
            source=source
        )
        
        self.working_memory.append(item)
        self.cognitive_trace.append((CognitiveState.ENCODING, f"Stored: {content[:50]}"))
    
    def _reason(self, input_text: str, perception: str) -> ReasoningResult:
        """Reasoning: Logical inference and problem-solving"""
        
        self.cognitive_trace.append((CognitiveState.REASONING, "Applying reasoning"))
        
        # Recall relevant long-term memories
        recalled_memories = self.long_term_memory.recall(input_text)
        
        # Get working memory context
        wm_context = "\n".join([
            f"- {item.content}"
            for item in sorted(self.working_memory, key=lambda x: x.relevance, reverse=True)
        ])
        
        # Build reasoning prompt
        memory_context = "\n".join([
            f"[{mem_type}] {mem_content}"
            for mem_type, mem_content in recalled_memories
        ]) if recalled_memories else "No relevant memories"
        
        reasoning_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are reasoning about a situation step-by-step.

Working Memory Context:
{working_memory}

Long-Term Memory:
{long_term_memory}

Provide:
1. Step-by-step reasoning
2. Logical conclusion
3. Confidence level (0.0-1.0)

Format:
Steps: [numbered list]
Conclusion: [statement]
Confidence: [0.0-1.0]"""),
            ("user", "Situation: {input}\nPerception: {perception}\n\nReason:")
        ])
        
        chain = reasoning_prompt | self.llm | StrOutputParser()
        reasoning_output = chain.invoke({
            "input": input_text,
            "perception": perception,
            "working_memory": wm_context or "Empty",
            "long_term_memory": memory_context
        })
        
        # Parse reasoning output (simplified)
        lines = reasoning_output.split('\n')
        steps = []
        conclusion = ""
        confidence = 0.7
        
        for line in lines:
            if line.strip().startswith(('1.', '2.', '3.', '4.', '5.', '-')):
                steps.append(line.strip())
            elif 'conclusion:' in line.lower():
                conclusion = line.split(':', 1)[1].strip()
            elif 'confidence:' in line.lower():
                try:
                    conf_str = line.split(':', 1)[1].strip()
                    confidence = float(conf_str)
                except:
                    pass
        
        if not conclusion:
            conclusion = reasoning_output.split('\n')[0]
        
        result = ReasoningResult(
            conclusion=conclusion,
            reasoning_steps=steps,
            confidence=confidence,
            used_memories=[m[1] for m in recalled_memories]
        )
        
        self.cognitive_trace.append((CognitiveState.REASONING, f"Conclusion: {conclusion[:50]}"))
        
        # Store conclusion in working memory
        self._encode_working_memory(conclusion, relevance=0.9, source="reasoning")
        
        return result
    
    def _plan(self, reasoning: ReasoningResult, input_text: str) -> ActionPlan:
        """Planning: Create action plan to achieve goal"""
        
        self.cognitive_trace.append((CognitiveState.PLANNING, "Creating action plan"))
        
        planning_prompt = ChatPromptTemplate.from_messages([
            ("system", """Create an action plan based on the reasoning.

Provide:
1. Clear goal
2. Ordered steps to achieve it
3. Expected outcome

Format:
Goal: [goal]
Steps:
1. [step]
2. [step]
...
Expected: [outcome]"""),
            ("user", "Situation: {input}\nConclusion: {conclusion}\n\nPlan:")
        ])
        
        chain = planning_prompt | self.llm | StrOutputParser()
        plan_output = chain.invoke({
            "input": input_text,
            "conclusion": reasoning.conclusion
        })
        
        # Parse plan
        lines = plan_output.split('\n')
        goal = ""
        steps = []
        expected = ""
        
        for line in lines:
            if 'goal:' in line.lower():
                goal = line.split(':', 1)[1].strip()
            elif line.strip().startswith(('1.', '2.', '3.', '4.', '5.', '-')):
                steps.append(line.strip())
            elif 'expected:' in line.lower():
                expected = line.split(':', 1)[1].strip()
        
        if not goal:
            goal = input_text
        if not expected:
            expected = reasoning.conclusion
        
        plan = ActionPlan(
            goal=goal,
            steps=steps if steps else ["Execute reasoning conclusion"],
            expected_outcome=expected,
            confidence=reasoning.confidence
        )
        
        self.current_goal = goal
        self.cognitive_trace.append((CognitiveState.PLANNING, f"Goal: {goal[:50]}"))
        
        return plan
    
    def _act(self, plan: ActionPlan, reasoning: ReasoningResult) -> str:
        """Action: Select and execute action"""
        
        self.cognitive_trace.append((CognitiveState.ACTING, "Executing action"))
        
        # Generate response based on plan and reasoning
        action_prompt = ChatPromptTemplate.from_messages([
            ("system", """Based on the plan and reasoning, provide a helpful response.

Be clear, actionable, and directly address the user's need."""),
            ("user", """Goal: {goal}
Conclusion: {conclusion}
Plan: {plan}

Provide response:""")
        ])
        
        chain = action_prompt | self.llm | StrOutputParser()
        action = chain.invoke({
            "goal": plan.goal,
            "conclusion": reasoning.conclusion,
            "plan": "\n".join(plan.steps)
        })
        
        self.cognitive_trace.append((CognitiveState.ACTING, f"Action: {action[:50]}"))
        
        # Store action in long-term memory as experience
        self.long_term_memory.add_experience(
            f"Goal: {plan.goal} | Action: {action[:100]}"
        )
        
        return action
    
    def _reflect(self, action: str, reasoning: ReasoningResult) -> str:
        """Reflection: Evaluate outcome and learn"""
        
        self.cognitive_trace.append((CognitiveState.REFLECTING, "Reflecting on action"))
        
        reflection_prompt = ChatPromptTemplate.from_messages([
            ("system", """Reflect on the action taken.

Consider:
1. Was it appropriate?
2. What was learned?
3. How to improve?

Be brief (2-3 sentences)."""),
            ("user", "Action: {action}\nConfidence: {confidence}\n\nReflection:")
        ])
        
        chain = reflection_prompt | self.llm | StrOutputParser()
        reflection = chain.invoke({
            "action": action,
            "confidence": reasoning.confidence
        })
        
        self.cognitive_trace.append((CognitiveState.REFLECTING, reflection))
        
        # Store reflection as learning
        self.long_term_memory.add_experience(f"Learned: {reflection}")
        
        return reflection
    
    def process(self, input_text: str) -> CognitiveResult:
        """Process input through complete cognitive cycle"""
        
        start_time = time.time()
        self.cognitive_trace = []
        
        # 1. Perception
        perception, perceptual_input = self._perceive(input_text)
        
        # 2. Attention
        attention_focus = self._attend(perception, input_text)
        
        # 3. Encoding (store perception in working memory)
        self._encode_working_memory(perception, relevance=0.8, source="perception")
        
        # 4. Reasoning
        reasoning = self._reason(input_text, perception)
        
        # 5. Planning
        plan = self._plan(reasoning, input_text)
        
        # 6. Action
        action = self._act(plan, reasoning)
        
        # 7. Reflection
        reflection = self._reflect(action, reasoning)
        
        execution_time_ms = (time.time() - start_time) * 1000
        
        return CognitiveResult(
            input=input_text,
            perception=perception,
            attention_focus=attention_focus,
            reasoning=reasoning,
            plan=plan,
            action=action,
            reflection=reflection,
            cognitive_trace=self.cognitive_trace,
            execution_time_ms=execution_time_ms
        )


def demonstrate_cognitive_architecture():
    """Demonstrate Cognitive Architecture pattern"""
    
    print("=" * 80)
    print("PATTERN 055: COGNITIVE ARCHITECTURE DEMONSTRATION")
    print("=" * 80)
    print("\nHuman-like cognitive processing with perception, reasoning, and action\n")
    
    # Create cognitive agent
    agent = CognitiveAgent(working_memory_capacity=7, temperature=0.7)
    
    # Add some initial knowledge
    agent.long_term_memory.add_fact("Python is a programming language")
    agent.long_term_memory.add_fact("Lists in Python are mutable")
    agent.long_term_memory.add_skill("Able to explain programming concepts")
    agent.long_term_memory.add_goal("Help users learn effectively")
    
    # Test 1: Simple query
    print("\n" + "=" * 80)
    print("TEST 1: Simple Query Processing")
    print("=" * 80)
    
    query1 = "How do I reverse a list in Python?"
    
    print(f"\n💭 Input: {query1}")
    result1 = agent.process(query1)
    
    print(f"\n🔍 Cognitive Processing Trace:")
    for state, description in result1.cognitive_trace[:5]:  # Show first 5
        print(f"   [{state.value}] {description[:80]}")
    
    print(f"\n👁️  Perception: {result1.perception}")
    print(f"🎯 Attention Focus: {result1.attention_focus.value}")
    print(f"🧠 Reasoning Conclusion: {result1.reasoning.conclusion}")
    print(f"📋 Plan ({len(result1.plan.steps)} steps): {result1.plan.goal}")
    print(f"⚡ Action: {result1.action[:200]}...")
    print(f"🤔 Reflection: {result1.reflection}")
    print(f"⏱️  Execution Time: {result1.execution_time_ms:.1f}ms")
    
    # Test 2: Complex reasoning task
    print("\n" + "=" * 80)
    print("TEST 2: Complex Reasoning Task")
    print("=" * 80)
    
    query2 = "I need to build a web app but don't know where to start. What should I do?"
    
    print(f"\n💭 Input: {query2}")
    result2 = agent.process(query2)
    
    print(f"\n🔍 Full Cognitive Cycle:")
    print(f"   Perception: {result2.perception[:80]}...")
    print(f"   Attention: {result2.attention_focus.value}")
    print(f"   Reasoning Steps:")
    for i, step in enumerate(result2.reasoning.steps[:3], 1):
        print(f"      {i}. {step}")
    print(f"   Conclusion: {result2.reasoning.conclusion}")
    print(f"   Plan Goal: {result2.plan.goal}")
    print(f"   Action Plan:")
    for i, step in enumerate(result2.plan.steps[:4], 1):
        print(f"      {i}. {step}")
    
    print(f"\n💬 Final Response: {result2.action[:250]}...")
    
    # Test 3: Memory integration
    print("\n" + "=" * 80)
    print("TEST 3: Memory Integration")
    print("=" * 80)
    
    # First query establishes context
    query3a = "What are decorators in Python?"
    print(f"\n💭 Query 1: {query3a}")
    result3a = agent.process(query3a)
    print(f"   Response: {result3a.action[:150]}...")
    
    # Second query builds on previous
    query3b = "Can you give me a practical example?"
    print(f"\n💭 Query 2: {query3b}")
    result3b = agent.process(query3b)
    
    print(f"\n📚 Working Memory State ({len(agent.working_memory)} items):")
    for item in list(agent.working_memory)[-3:]:
        print(f"   - {str(item)[:60]}...")
    
    print(f"\n💬 Contextual Response: {result3b.action[:200]}...")
    
    # Show memory usage
    if result3b.reasoning.used_memories:
        print(f"\n🧠 Long-Term Memories Recalled:")
        for mem in result3b.reasoning.used_memories[:2]:
            print(f"   - {mem[:70]}...")
    
    # Summary
    print("\n" + "=" * 80)
    print("COGNITIVE ARCHITECTURE PATTERN SUMMARY")
    print("=" * 80)
    print("""
Key Benefits:
1. Human-Like Processing: Mirrors human cognitive stages
2. Transparency: Clear cognitive trace for debugging
3. Context Awareness: Working and long-term memory integration
4. Adaptive Reasoning: Adjusts to different input types
5. Learning: Reflection enables continuous improvement

Cognitive Components:
1. Perception: Input processing and interpretation
2. Attention: Selective focus on relevant information
3. Working Memory: Limited-capacity temporary storage (7±2 items)
4. Long-Term Memory: Persistent knowledge storage
5. Reasoning: Logical inference and problem-solving
6. Planning: Goal-directed action sequencing
7. Action: Response generation and execution
8. Reflection: Outcome evaluation and learning

Memory Systems:
- Working Memory: Temporary, limited capacity, high relevance
- Long-Term Memory: Permanent, unlimited, organized by type
  * Facts: Declarative knowledge
  * Experiences: Episodic memories
  * Skills: Procedural knowledge
  * Goals: Long-term objectives

Attention Mechanism:
- Goal-focused: Primary objective attention
- Problem-focused: Current challenge attention
- Context-focused: Environmental awareness
- Memory-focused: Past experience recall
- Action-focused: Next step planning

Processing Stages:
1. Perceive → Extract features and intent
2. Attend → Focus on relevant aspects
3. Encode → Store in working memory
4. Reason → Logical inference
5. Plan → Create action sequence
6. Act → Execute plan
7. Reflect → Learn from outcome

Use Cases:
- Intelligent tutoring systems
- Personal assistant agents
- Complex decision support
- Game AI characters
- Research and analysis agents
- Autonomous problem-solving

Best Practices:
1. Limit working memory capacity (realistic)
2. Prioritize by relevance scores
3. Regular reflection for learning
4. Clear cognitive traces for transparency
5. Balance depth vs speed
6. Maintain long-term memory
7. Adaptive attention mechanisms

Production Considerations:
- Memory persistence (database)
- Working memory cleanup
- Long-term memory search (embeddings)
- Cognitive trace logging
- Performance optimization
- State management
- Error recovery

Comparison with Related Patterns:
- vs. ReAct: Full cognitive cycle vs action-observation
- vs. Chain-of-Thought: Multi-stage vs single reasoning
- vs. Memory Pattern: Integrated architecture vs storage
- vs. Agent: Cognitive model vs tool use

The Cognitive Architecture pattern provides sophisticated, human-like
information processing that enables contextual understanding, adaptive
reasoning, and continuous learning through reflection.
""")


if __name__ == "__main__":
    demonstrate_cognitive_architecture()

    

Agentic Design Pattern: Cognitive Architecture

Pattern 055: Cognitive Architecture – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Cognitive Architecture pattern implementation.

---

Pattern Description

Pattern 055: Cognitive Architecture

Description: The Cognitive Architecture pattern models human-like cognitive processes in AI agents, including perception, attention, working memory, reasoning, planning, and action. This creates agents that process information in stages similar to human cognition, enabling more sophisticated and contextual decision-making.

Components: 1. Perception Module: Processes and interprets inputs 2. Attention Mechanism: Focuses on relevant information 3. Working Memory: Maintains current context 4. Long-Term Memory: Stores knowledge and experiences 5. Reasoning Engine: Logical inference and problem-solving 6. Planning Module: Goal-directed action planning 7. Action Selection: Chooses and executes actions

Use Cases: – Complex decision-making systems – Autonomous agents with human-like reasoning – Educational tutoring systems – Game AI with realistic behavior – Personal assistants with context awareness – Research and analysis agents

LangChain Implementation: Implements a multi-stage cognitive pipeline using LangChain components to simulate perception, memory, reasoning, and action selection.

---

File Statistics

• Total Lines: 710
• Classes: 9
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 055: Cognitive Architecture

Description:
    The Cognitive Architecture pattern models human-like cognitive processes in AI agents,
    including perception, attention, working memory, reasoning, planning, and action.
    This creates agents that process information in stages similar to human cognition,
    enabling more sophisticated and contextual decision-making.

Components:
    1. Perception Module: Processes and interprets inputs
    2. Attention Mechanism: Focuses on relevant information
    3. Working Memory: Maintains current context
    4. Long-Term Memory: Stores knowledge and experiences
    5. Reasoning Engine: Logical inference and problem-solving
    6. Planning Module: Goal-directed action planning
    7. Action Selection: Chooses and executes actions

Use Cases:
    - Complex decision-making systems
    - Autonomous agents with human-like reasoning
    - Educational tutoring systems
    - Game AI with realistic behavior
    - Personal assistants with context awareness
    - Research and analysis agents

LangChain Implementation:
    Implements a multi-stage cognitive pipeline using LangChain components
    to simulate perception, memory, reasoning, and action selection.
"""

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

---

Import Statements

Standard Library Imports

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

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

LangChain Imports

from langchain_openai.ChatOpenAI

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

Lines 48-56

Inherits from: Enum

Purpose:

States in cognitive processing cycle

---

Class: AttentionFocus

Lines 59-65

Inherits from: Enum

Purpose:

What the agent is focusing on

---

Class: PerceptualInput

Lines 69-77

Purpose:

Input perceived by the agent

Methods:

to_text(self)

Lines 76-77

Purpose: To Text operation.

Parameters:

Returns: str

---

Class: WorkingMemoryItem

Lines 81-89

Purpose:

Item in working memory

Methods:

__str__(self)

Lines 88-89

Purpose: Str operation.

Parameters:

Returns: str

---

Class: LongTermMemory

Lines 93-124

Purpose:

Long-term memory store

Methods:

add_fact(self, fact)

Lines 100-102

Purpose: Add Fact operation.

Parameters: – fact: Fact

add_experience(self, experience)

Lines 104-108

Purpose: Add Experience operation.

Parameters: – experience: Experience

recall(self, query, k)

Lines 110-124

Purpose:

Recall relevant memories

Parameters: – query: Input query or question – k: K

Returns: <ast.Subscript object at 0x000001A06D681350>

---

Class: ReasoningResult

Lines 128-133

Purpose:

Result from reasoning process

---

Class: ActionPlan

Lines 137-142

Purpose:

Plan for achieving goal

---

Class: CognitiveResult

Lines 146-166

Purpose:

Complete result from cognitive processing

Methods:

to_dict(self)

Lines 158-166

Purpose: To Dict operation.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D686850>

---

Class: CognitiveAgent

Lines 169-530

Purpose:

Agent with human-like cognitive architecture.

Implements:
1. Perception: Input processing and feature extraction
2. Attention: Selective focus on relevant information
3. Working Memory: Temporary context storage (limited capacity)
4. Long-Term Memory: Persistent knowledge storage
5. Reasoning: Logical inference and problem-solving
6. Planning: Goal-directed action planning
7. Action: Selection and execution
8. Reflection: Learning from outcomes

Methods:

__init__(self, working_memory_capacity, temperature)

Lines 184-204

Purpose: Initialize the class instance with required parameters.

Parameters: – working_memory_capacity: Memory store for conversation history – temperature: Sampling temperature (0-1)

_perceive(self, input_text)

Lines 206-235

Purpose:

Perception: Process and interpret input

Parameters: – input_text: Input data

Returns: <ast.Subscript object at 0x000001A06D607890>

_attend(self, perception, input_text)

Lines 237-269

Purpose:

Attention: Focus on most relevant aspect

Parameters: – perception: Perception – input_text: Input data

Returns: AttentionFocus

_encode_working_memory(self, content, relevance, source)

Lines 271-282

Purpose:

Encode information into working memory

Parameters: – content: Content – relevance: Relevance – source: Source

_reason(self, input_text, perception)

Lines 284-366

Purpose:

Reasoning: Logical inference and problem-solving

Parameters: – input_text: Input data – perception: Perception

Returns: ReasoningResult

_plan(self, reasoning, input_text)

Lines 368-426

Purpose:

Planning: Create action plan to achieve goal

Parameters: – reasoning: Reasoning – input_text: Input data

Returns: ActionPlan

_act(self, plan, reasoning)

Lines 428-459

Purpose:

Action: Select and execute action

Parameters: – plan: Plan – reasoning: Reasoning

Returns: str

_reflect(self, action, reasoning)

Lines 461-489

Purpose:

Reflection: Evaluate outcome and learn

Parameters: – action: Action – reasoning: Reasoning

Returns: str

process(self, input_text)

Lines 491-530

Purpose:

Process input through complete cognitive cycle

Parameters: – input_text: Input data

Returns: CognitiveResult

---

Functions

demonstrate_cognitive_architecture()

Lines 533-705

Purpose:

Demonstrate Cognitive Architecture pattern

---

Key Concepts Summary

Pattern Overview

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

Core Components

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

When to Use

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

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

---

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

---

LangChain Components Used

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

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

---

Execution Flow

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

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

---

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

---

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

---

Related Patterns

• Pattern 054
• Pattern 056

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 Cognitive Architecture 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: 055_cognitive_architecture.py
Total Lines: 710
Pattern Category: AI Agent Design Pattern #055

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 054: Agent Specialization & Routing

Description:
    The Agent Specialization & Routing pattern creates a system of domain-specific
    expert agents with intelligent routing logic. A meta-agent classifies incoming
    tasks and routes them to the most appropriate specialist, enabling efficient
    handling of diverse tasks by leveraging specialized expertise.

Components:
    1. Specialist Agents: Domain-specific expert agents
    2. Router Agent: Classifies and routes tasks
    3. Task Classifier: Determines task domain/type
    4. Capability Registry: Tracks agent specializations
    5. Load Balancer: Distributes work efficiently

Use Cases:
    - Multi-domain customer support
    - Technical help desk routing
    - Content moderation by category
    - Code review by language
    - Medical diagnosis routing by specialty
    - Legal document analysis by practice area

LangChain Implementation:
    Uses a classifier LLM to route queries to specialized LLM instances
    configured for specific domains, with fallback to generalist agents.
"""

import os
import time
from typing import Dict, Any, List, Optional, Set, Callable
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import defaultdict

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 Domain(Enum):
    """Domains for specialized agents"""
    TECHNICAL = "technical"  # Programming, IT
    MEDICAL = "medical"  # Healthcare, diagnosis
    LEGAL = "legal"  # Law, contracts
    FINANCIAL = "financial"  # Finance, accounting
    CREATIVE = "creative"  # Writing, design
    SCIENTIFIC = "scientific"  # Research, analysis
    CUSTOMER_SERVICE = "customer_service"  # Support, relations
    GENERAL = "general"  # Default fallback


class RoutingStrategy(Enum):
    """Strategies for routing tasks"""
    BEST_MATCH = "best_match"  # Route to single best specialist
    MULTI_ROUTE = "multi_route"  # Route to multiple relevant specialists
    LOAD_BALANCED = "load_balanced"  # Balance across similar specialists
    CONFIDENCE_BASED = "confidence_based"  # Route based on confidence threshold


@dataclass
class AgentCapability:
    """Describes what an agent can do"""
    domain: Domain
    specializations: List[str]
    expertise_level: float  # 0.0-1.0
    keywords: Set[str]
    examples: List[str]
    
    def matches_query(self, query: str) -> float:
        """Calculate match score for query"""
        query_lower = query.lower()
        
        # Keyword matching
        keyword_score = sum(
            1 for keyword in self.keywords
            if keyword.lower() in query_lower
        ) / max(len(self.keywords), 1)
        
        # Specialization matching
        spec_score = sum(
            1 for spec in self.specializations
            if spec.lower() in query_lower
        ) / max(len(self.specializations), 1)
        
        # Combine scores
        match_score = (keyword_score * 0.6 + spec_score * 0.4) * self.expertise_level
        
        return min(match_score, 1.0)


@dataclass
class SpecialistAgent:
    """A specialized agent with specific capabilities"""
    agent_id: str
    domain: Domain
    capability: AgentCapability
    llm: ChatOpenAI
    system_prompt: str
    request_count: int = 0
    total_time_ms: float = 0.0
    
    @property
    def avg_time_ms(self) -> float:
        return self.total_time_ms / self.request_count if self.request_count > 0 else 0.0


@dataclass
class RoutingDecision:
    """Decision made by router"""
    query: str
    selected_agents: List[str]
    domain: Domain
    confidence: float
    reasoning: str
    alternatives: List[tuple[str, float]]  # (agent_id, score)


@dataclass
class RoutingResult:
    """Result from specialized agent routing"""
    query: str
    routing_decision: RoutingDecision
    specialist_response: str
    agent_id: str
    domain: Domain
    execution_time_ms: float
    routing_time_ms: float
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "query": self.query[:100] + "..." if len(self.query) > 100 else self.query,
            "routed_to": self.agent_id,
            "domain": self.domain.value,
            "confidence": f"{self.routing_decision.confidence:.2f}",
            "execution_time_ms": f"{self.execution_time_ms:.1f}",
            "routing_time_ms": f"{self.routing_time_ms:.1f}"
        }


class AgentRouter:
    """
    Intelligent routing system for specialized agents.
    
    Features:
    1. Multiple domain-specific specialist agents
    2. Intelligent task classification and routing
    3. Confidence-based decision making
    4. Load balancing capabilities
    5. Fallback to generalist agent
    """
    
    def __init__(
        self,
        routing_strategy: RoutingStrategy = RoutingStrategy.BEST_MATCH,
        confidence_threshold: float = 0.6
    ):
        self.routing_strategy = routing_strategy
        self.confidence_threshold = confidence_threshold
        
        # Classifier for routing
        self.classifier = ChatOpenAI(
            model_name="gpt-3.5-turbo",
            temperature=0.1  # Low temperature for consistent classification
        )
        
        # Registry of specialist agents
        self.specialists: Dict[str, SpecialistAgent] = {}
        
        # Generalist fallback
        self.generalist = ChatOpenAI(
            model_name="gpt-3.5-turbo",
            temperature=0.7
        )
        
        # Initialize specialists
        self._initialize_specialists()
    
    def _initialize_specialists(self):
        """Create specialized agents for different domains"""
        
        # Technical specialist
        tech_capability = AgentCapability(
            domain=Domain.TECHNICAL,
            specializations=["programming", "debugging", "algorithms", "systems"],
            expertise_level=0.9,
            keywords={"code", "program", "bug", "error", "function", "algorithm", 
                     "database", "API", "server", "debug"},
            examples=["How do I debug this code?", "Explain quicksort algorithm"]
        )
        
        self.specialists["tech_expert"] = SpecialistAgent(
            agent_id="tech_expert",
            domain=Domain.TECHNICAL,
            capability=tech_capability,
            llm=ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.3),
            system_prompt="""You are a senior software engineer and technical expert.
You specialize in programming, algorithms, debugging, and system design.
Provide clear, accurate technical answers with code examples when appropriate."""
        )
        
        # Medical specialist
        medical_capability = AgentCapability(
            domain=Domain.MEDICAL,
            specializations=["diagnosis", "treatment", "health", "medicine"],
            expertise_level=0.85,
            keywords={"health", "medical", "doctor", "symptom", "disease", "treatment",
                     "medicine", "diagnosis", "patient", "clinical"},
            examples=["What causes headaches?", "Explain diabetes management"]
        )
        
        self.specialists["medical_expert"] = SpecialistAgent(
            agent_id="medical_expert",
            domain=Domain.MEDICAL,
            capability=medical_capability,
            llm=ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.2),
            system_prompt="""You are a medical information specialist.
Provide accurate health and medical information. Always include disclaimer
that this is educational information and not medical advice."""
        )
        
        # Financial specialist
        financial_capability = AgentCapability(
            domain=Domain.FINANCIAL,
            specializations=["investing", "accounting", "economics", "trading"],
            expertise_level=0.85,
            keywords={"money", "investment", "stock", "finance", "accounting", "tax",
                     "budget", "economy", "market", "portfolio"},
            examples=["How to start investing?", "Explain compound interest"]
        )
        
        self.specialists["finance_expert"] = SpecialistAgent(
            agent_id="finance_expert",
            domain=Domain.FINANCIAL,
            capability=financial_capability,
            llm=ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.2),
            system_prompt="""You are a financial advisor and economics expert.
Provide clear explanations of financial concepts. Include disclaimer that
this is educational information and not financial advice."""
        )
        
        # Creative specialist
        creative_capability = AgentCapability(
            domain=Domain.CREATIVE,
            specializations=["writing", "storytelling", "design", "creative"],
            expertise_level=0.9,
            keywords={"story", "write", "creative", "design", "art", "narrative",
                     "plot", "character", "style", "aesthetic"},
            examples=["Help me write a story", "Design a logo concept"]
        )
        
        self.specialists["creative_expert"] = SpecialistAgent(
            agent_id="creative_expert",
            domain=Domain.CREATIVE,
            capability=creative_capability,
            llm=ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.9),
            system_prompt="""You are a creative writing and design expert.
Provide imaginative, original ideas and guidance. Be encouraging and
help users explore their creativity."""
        )
        
        # Scientific specialist
        science_capability = AgentCapability(
            domain=Domain.SCIENTIFIC,
            specializations=["research", "physics", "chemistry", "biology", "data"],
            expertise_level=0.85,
            keywords={"research", "science", "experiment", "hypothesis", "data",
                     "analysis", "physics", "chemistry", "biology", "scientific"},
            examples=["Explain quantum mechanics", "How to design an experiment"]
        )
        
        self.specialists["science_expert"] = SpecialistAgent(
            agent_id="science_expert",
            domain=Domain.SCIENTIFIC,
            capability=science_capability,
            llm=ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.3),
            system_prompt="""You are a research scientist with expertise across disciplines.
Provide accurate scientific explanations. Cite principles and methodologies.
Be precise and evidence-based."""
        )
    
    def _classify_query(self, query: str) -> RoutingDecision:
        """Classify query and determine routing"""
        
        start_time = time.time()
        
        # Calculate match scores for all specialists
        scores = []
        for agent_id, specialist in self.specialists.items():
            match_score = specialist.capability.matches_query(query)
            
            # Factor in load balancing
            if self.routing_strategy == RoutingStrategy.LOAD_BALANCED:
                # Penalize busy agents slightly
                load_penalty = specialist.request_count * 0.01
                match_score = max(0, match_score - load_penalty)
            
            scores.append((agent_id, match_score, specialist.domain))
        
        # Sort by score
        scores.sort(key=lambda x: x[1], reverse=True)
        
        # Best match
        best_agent_id, best_score, best_domain = scores[0]
        
        # Use LLM for low-confidence cases
        if best_score < self.confidence_threshold:
            # Ask LLM to classify
            classifier_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a query classifier. Classify queries into domains:
- technical: Programming, IT, debugging
- medical: Health, medicine, diagnosis
- financial: Money, investing, economics
- creative: Writing, art, design
- scientific: Research, physics, experiments
- general: Everything else

Respond with just the domain name."""),
                ("user", "Classify: {query}")
            ])
            
            chain = classifier_prompt | self.classifier | StrOutputParser()
            llm_domain = chain.invoke({"query": query}).strip().lower()
            
            # Find agent for this domain
            for agent_id, specialist in self.specialists.items():
                if specialist.domain.value == llm_domain:
                    best_agent_id = agent_id
                    best_domain = specialist.domain
                    best_score = 0.7  # LLM classification gets moderate confidence
                    break
        
        # Determine routing
        if self.routing_strategy == RoutingStrategy.MULTI_ROUTE:
            # Route to top 2 specialists
            selected_agents = [scores[0][0], scores[1][0]]
        else:
            selected_agents = [best_agent_id]
        
        reasoning = f"Matched domain: {best_domain.value} (score: {best_score:.2f})"
        if best_score  str:
        """Execute query with specialist agent"""
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", specialist.system_prompt),
            ("user", "{query}")
        ])
        
        chain = prompt | specialist.llm | StrOutputParser()
        
        start_time = time.time()
        response = chain.invoke({"query": query})
        execution_time = (time.time() - start_time) * 1000
        
        # Update statistics
        specialist.request_count += 1
        specialist.total_time_ms += execution_time
        
        return response
    
    def route_and_execute(self, query: str) -> RoutingResult:
        """Route query to specialist and execute"""
        
        total_start = time.time()
        
        # Classify and route
        routing_start = time.time()
        routing_decision = self._classify_query(query)
        routing_time_ms = (time.time() - routing_start) * 1000
        
        # Execute with primary specialist
        primary_agent_id = routing_decision.selected_agents[0]
        
        if primary_agent_id in self.specialists:
            specialist = self.specialists[primary_agent_id]
            response = self._execute_with_specialist(specialist, query)
        else:
            # Fallback to generalist
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are a helpful assistant providing accurate information."),
                ("user", "{query}")
            ])
            chain = prompt | self.generalist | StrOutputParser()
            response = chain.invoke({"query": query})
            primary_agent_id = "generalist"
            routing_decision.domain = Domain.GENERAL
        
        total_time_ms = (time.time() - total_start) * 1000
        execution_time_ms = total_time_ms - routing_time_ms
        
        return RoutingResult(
            query=query,
            routing_decision=routing_decision,
            specialist_response=response,
            agent_id=primary_agent_id,
            domain=routing_decision.domain,
            execution_time_ms=execution_time_ms,
            routing_time_ms=routing_time_ms
        )
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get routing statistics"""
        
        stats = {
            "total_specialists": len(self.specialists),
            "specialists": {}
        }
        
        for agent_id, specialist in self.specialists.items():
            stats["specialists"][agent_id] = {
                "domain": specialist.domain.value,
                "request_count": specialist.request_count,
                "avg_time_ms": f"{specialist.avg_time_ms:.1f}",
                "expertise": f"{specialist.capability.expertise_level:.2f}"
            }
        
        return stats


def demonstrate_agent_routing():
    """Demonstrate Agent Specialization & Routing pattern"""
    
    print("=" * 80)
    print("PATTERN 054: AGENT SPECIALIZATION & ROUTING DEMONSTRATION")
    print("=" * 80)
    print("\nIntelligent routing to domain-specific expert agents\n")
    
    # Create router
    router = AgentRouter(
        routing_strategy=RoutingStrategy.BEST_MATCH,
        confidence_threshold=0.6
    )
    
    # Test queries for different domains
    test_queries = [
        ("How do I fix a memory leak in Python?", Domain.TECHNICAL),
        ("What causes high blood pressure?", Domain.MEDICAL),
        ("Should I invest in index funds or individual stocks?", Domain.FINANCIAL),
        ("Help me write an opening for a mystery novel", Domain.CREATIVE),
        ("Explain the double-slit experiment", Domain.SCIENTIFIC),
        ("What's the weather like today?", Domain.GENERAL)  # Should route to general
    ]
    
    # Test 1: Basic routing
    print("\n" + "=" * 80)
    print("TEST 1: Routing to Specialized Agents")
    print("=" * 80)
    
    for query, expected_domain in test_queries[:3]:
        print(f"\n📝 Query: {query}")
        print(f"   Expected Domain: {expected_domain.value}")
        
        result = router.route_and_execute(query)
        
        print(f"\n   ✅ Routing Decision:")
        print(f"      Routed to: {result.agent_id}")
        print(f"      Domain: {result.domain.value}")
        print(f"      Confidence: {result.routing_decision.confidence:.2f}")
        print(f"      Reasoning: {result.routing_decision.reasoning}")
        
        if result.routing_decision.alternatives:
            print(f"\n   📊 Alternative Matches:")
            for alt_agent, alt_score in result.routing_decision.alternatives:
                print(f"      - {alt_agent}: {alt_score:.2f}")
        
        print(f"\n   💬 Response: {result.specialist_response[:150]}...")
        print(f"\n   ⏱️  Timing:")
        print(f"      Routing: {result.routing_time_ms:.1f}ms")
        print(f"      Execution: {result.execution_time_ms:.1f}ms")
    
    # Test 2: Creative vs Technical routing
    print("\n" + "=" * 80)
    print("TEST 2: Domain Contrast - Creative vs Technical")
    print("=" * 80)
    
    creative_query = "Write a poem about artificial intelligence"
    technical_query = "Implement a binary search algorithm"
    
    for query in [creative_query, technical_query]:
        print(f"\n📝 Query: {query}")
        result = router.route_and_execute(query)
        
        print(f"   Routed to: {result.agent_id} ({result.domain.value})")
        print(f"   Response: {result.specialist_response[:200]}...")
    
    # Test 3: Load balancing simulation
    print("\n" + "=" * 80)
    print("TEST 3: Load Balancing")
    print("=" * 80)
    
    # Create load-balanced router
    lb_router = AgentRouter(
        routing_strategy=RoutingStrategy.LOAD_BALANCED,
        confidence_threshold=0.5
    )
    
    # Simulate multiple requests
    tech_queries = [
        "What is recursion?",
        "Explain REST APIs",
        "How does Git work?"
    ]
    
    print("\nSimulating multiple technical queries with load balancing...")
    
    for i, query in enumerate(tech_queries, 1):
        result = lb_router.route_and_execute(query)
        print(f"\n   Request {i}: {query[:40]}...")
        print(f"      Routed to: {result.agent_id}")
        print(f"      Execution time: {result.execution_time_ms:.1f}ms")
    
    # Show statistics
    print("\n   📊 Load Distribution:")
    stats = lb_router.get_statistics()
    for agent_id, agent_stats in stats["specialists"].items():
        if agent_stats["request_count"] > 0:
            print(f"      {agent_id}: {agent_stats['request_count']} requests, "
                  f"avg {agent_stats['avg_time_ms']}ms")
    
    # Test 4: Ambiguous query routing
    print("\n" + "=" * 80)
    print("TEST 4: Handling Ambiguous Queries")
    print("=" * 80)
    
    ambiguous_queries = [
        "Tell me about neural networks",  # Could be technical or scientific
        "How to manage stress",  # Could be medical or general
    ]
    
    for query in ambiguous_queries:
        print(f"\n📝 Query: {query}")
        result = router.route_and_execute(query)
        
        print(f"   Decision:")
        print(f"      Selected: {result.agent_id} ({result.domain.value})")
        print(f"      Confidence: {result.routing_decision.confidence:.2f}")
        print(f"      Reasoning: {result.routing_decision.reasoning}")
        
        if result.routing_decision.alternatives:
            print(f"   Alternatives considered:")
            for alt_agent, alt_score in result.routing_decision.alternatives:
                specialist = router.specialists.get(alt_agent)
                domain = specialist.domain.value if specialist else "unknown"
                print(f"      - {alt_agent} ({domain}): {alt_score:.2f}")
    
    # Summary
    print("\n" + "=" * 80)
    print("AGENT SPECIALIZATION & ROUTING PATTERN SUMMARY")
    print("=" * 80)
    print("""
Key Benefits:
1. Expert Quality: Specialized agents provide domain-specific expertise
2. Efficiency: Route to appropriate agent immediately
3. Scalability: Easy to add new specialists
4. Load Distribution: Balance work across agents
5. Maintainability: Domain-specific logic in separate agents

Implementation Features:
1. Capability Registry: Tracks agent specializations
2. Intelligent Classification: Keyword + LLM-based routing
3. Confidence Thresholding: Fallback for uncertain cases
4. Load Balancing: Distribute requests evenly
5. Statistics Tracking: Monitor agent performance

Routing Strategies:
1. Best Match: Route to single best specialist
2. Multi-Route: Send to multiple relevant specialists
3. Load Balanced: Distribute across similar specialists
4. Confidence-Based: Threshold-based routing decisions

Agent Specializations:
- Technical: Programming, debugging, algorithms
- Medical: Health, diagnosis, treatment
- Financial: Investing, accounting, economics
- Creative: Writing, design, storytelling
- Scientific: Research, experiments, analysis
- General: Fallback for unclassified queries

Classification Methods:
1. Keyword Matching: Fast, rule-based
2. LLM Classification: Accurate for ambiguous cases
3. Hybrid: Combine both approaches
4. Confidence Scoring: Quantify match quality

Use Cases:
- Customer support ticket routing
- Technical helpdesk by expertise
- Content moderation by category
- Medical triage systems
- Legal document routing
- Educational tutoring systems

Best Practices:
1. Clear domain boundaries
2. Comprehensive keyword sets
3. LLM fallback for edge cases
4. Monitor routing accuracy
5. Track specialist performance
6. Regular capability updates
7. Maintain generalist fallback

Production Considerations:
- Caching classification results
- A/B testing routing strategies
- Monitoring routing accuracy
- Specialist availability checks
- Timeout handling per agent
- Cost optimization by routing
- User feedback on routing quality

Comparison with Related Patterns:
- vs. Mixture of Agents: Routing vs aggregation
- vs. Multi-Agent: Specialized vs collaborative
- vs. Chain: Direct routing vs sequential
- vs. Hierarchical: Flat routing vs tree structure

The Agent Specialization & Routing pattern enables efficient, scalable
systems that leverage domain expertise while maintaining simplicity
through intelligent task distribution.
""")


if __name__ == "__main__":
    demonstrate_agent_routing()

    

Agentic Design Pattern: Agent Specialization Routing

Pattern 054: Agent Specialization Routing – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Agent Specialization Routing pattern implementation.

---

Pattern Description

Pattern 054: Agent Specialization & Routing

Description: The Agent Specialization & Routing pattern creates a system of domain-specific expert agents with intelligent routing logic. A meta-agent classifies incoming tasks and routes them to the most appropriate specialist, enabling efficient handling of diverse tasks by leveraging specialized expertise.

Components: 1. Specialist Agents: Domain-specific expert agents 2. Router Agent: Classifies and routes tasks 3. Task Classifier: Determines task domain/type 4. Capability Registry: Tracks agent specializations 5. Load Balancer: Distributes work efficiently

Use Cases: – Multi-domain customer support – Technical help desk routing – Content moderation by category – Code review by language – Medical diagnosis routing by specialty – Legal document analysis by practice area

LangChain Implementation: Uses a classifier LLM to route queries to specialized LLM instances configured for specific domains, with fallback to generalist agents.

---

File Statistics

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

---

Module Documentation

"""
Pattern 054: Agent Specialization & Routing

Description:
    The Agent Specialization & Routing pattern creates a system of domain-specific
    expert agents with intelligent routing logic. A meta-agent classifies incoming
    tasks and routes them to the most appropriate specialist, enabling efficient
    handling of diverse tasks by leveraging specialized expertise.

Components:
    1. Specialist Agents: Domain-specific expert agents
    2. Router Agent: Classifies and routes tasks
    3. Task Classifier: Determines task domain/type
    4. Capability Registry: Tracks agent specializations
    5. Load Balancer: Distributes work efficiently

Use Cases:
    - Multi-domain customer support
    - Technical help desk routing
    - Content moderation by category
    - Code review by language
    - Medical diagnosis routing by specialty
    - Legal document analysis by practice area

LangChain Implementation:
    Uses a classifier LLM to route queries to specialized LLM instances
    configured for specific domains, with fallback to generalist agents.
"""

Purpose: This module implements the Agent Specialization Routing pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

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

Purpose: Core Python functionality: – collections.defaultdict: Specialized container datatypes – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – datetime.datetime: Date and time handling – enum.Enum: Enumeration support – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – time: Time-related functions – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Callable: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional) – 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: Domain

Lines 46-55

Inherits from: Enum

Purpose:

Domains for specialized agents

---

Class: RoutingStrategy

Lines 58-63

Inherits from: Enum

Purpose:

Strategies for routing tasks

---

Class: AgentCapability

Lines 67-94

Purpose:

Describes what an agent can do

Methods:

matches_query(self, query)

Lines 75-94

Purpose:

Calculate match score for query

Parameters: – query: Input query or question

Returns: float

---

Class: SpecialistAgent

Lines 98-110

Purpose:

A specialized agent with specific capabilities

Methods:

avg_time_ms(self)

Lines 109-110

Decorators: property

Purpose: Avg Time Ms operation.

Parameters:

Returns: float

---

Class: RoutingDecision

Lines 114-121

Purpose:

Decision made by router

---

Class: RoutingResult

Lines 125-143

Purpose:

Result from specialized agent routing

Methods:

to_dict(self)

Lines 135-143

Purpose: To Dict operation.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D609750>

---

Class: AgentRouter

Lines 146-438

Purpose:

Intelligent routing system for specialized agents.

Features:
1. Multiple domain-specific specialist agents
2. Intelligent task classification and routing
3. Confidence-based decision making
4. Load balancing capabilities
5. Fallback to generalist agent

Methods:

__init__(self, routing_strategy, confidence_threshold)

Lines 158-182

Purpose: Initialize the class instance with required parameters.

Parameters: – routing_strategy: Routing strategy – confidence_threshold: Confidence threshold

_initialize_specialists(self)

Lines 184-285

Purpose:

Create specialized agents for different domains

Parameters:

_classify_query(self, query)

Lines 287-356

Purpose:

Classify query and determine routing

Parameters: – query: Input query or question

Returns: RoutingDecision

_execute_with_specialist(self, specialist, query)

Lines 358-380

Purpose:

Execute query with specialist agent

Parameters: – specialist: Specialist – query: Input query or question

Returns: str

route_and_execute(self, query)

Lines 382-420

Purpose:

Route query to specialist and execute

Parameters: – query: Input query or question

Returns: RoutingResult

get_statistics(self)

Lines 422-438

Purpose:

Get routing statistics

Parameters:

Returns: <ast.Subscript object at 0x000001A06D6F7610>

---

Functions

demonstrate_agent_routing()

Lines 441-641

Purpose:

Demonstrate Agent Specialization & Routing pattern

---

Key Concepts Summary

Pattern Overview

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

Core Components

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

When to Use

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

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

---

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

---

LangChain Components Used

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

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

---

Execution Flow

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

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

---

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

---

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

---

Related Patterns

• Pattern 053
• Pattern 055

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 Agent Specialization Routing 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: 054_agent_specialization_routing.py
Total Lines: 646
Pattern Category: AI Agent Design Pattern #054

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 053: Mixture of Agents (MoA)

Description:
    The Mixture of Agents pattern combines outputs from multiple specialized agents
    using learned or rule-based aggregation strategies. Each agent brings specific
    expertise, and their outputs are intelligently combined to produce superior
    results that leverage the strengths of each individual agent.

Components:
    1. Specialized Agents: Multiple agents with different capabilities/models
    2. Proposer Agents: Generate initial candidate responses
    3. Aggregator Agent: Combines and synthesizes multiple responses
    4. Confidence Scoring: Weights agent contributions
    5. Quality Evaluation: Assesses individual and combined outputs

Use Cases:
    - Complex tasks requiring diverse expertise
    - Multi-domain question answering
    - Code generation with multiple approaches
    - Creative content requiring varied perspectives
    - Research synthesis from multiple sources
    - Decision-making with multiple viewpoints

LangChain Implementation:
    Uses multiple LLM instances with different configurations or models,
    then aggregates their outputs using a meta-agent that synthesizes
    the best aspects of each response.
"""

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

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

load_dotenv()


class AgentRole(Enum):
    """Roles for agents in MoA"""
    PROPOSER = "proposer"  # Generates candidate responses
    AGGREGATOR = "aggregator"  # Combines multiple responses
    CRITIC = "critic"  # Evaluates quality
    SPECIALIST = "specialist"  # Domain expert


class AggregationStrategy(Enum):
    """Strategies for combining agent outputs"""
    SYNTHESIS = "synthesis"  # Synthesize best parts
    VOTING = "voting"  # Majority vote or ranking
    WEIGHTED = "weighted"  # Weighted by confidence
    SEQUENTIAL = "sequential"  # Sequential refinement
    BEST_OF = "best_of"  # Select single best


@dataclass
class AgentOutput:
    """Output from a single agent"""
    agent_id: str
    agent_role: AgentRole
    response: str
    confidence: float  # 0.0-1.0
    reasoning: Optional[str] = None
    metadata: Dict[str, Any] = field(default_factory=dict)
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "agent_id": self.agent_id,
            "role": self.agent_role.value,
            "response": self.response[:150] + "..." if len(self.response) > 150 else self.response,
            "confidence": f"{self.confidence:.2f}",
            "reasoning": self.reasoning[:100] + "..." if self.reasoning and len(self.reasoning) > 100 else self.reasoning
        }


@dataclass
class MoAResult:
    """Result from Mixture of Agents execution"""
    query: str
    final_response: str
    aggregation_strategy: AggregationStrategy
    agent_outputs: List[AgentOutput]
    aggregator_reasoning: str
    confidence_score: float
    execution_time_ms: float
    layer_count: int
    
    @property
    def num_agents(self) -> int:
        return len(self.agent_outputs)
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "query": self.query[:100] + "..." if len(self.query) > 100 else self.query,
            "final_response": self.final_response[:200] + "..." if len(self.final_response) > 200 else self.final_response,
            "strategy": self.aggregation_strategy.value,
            "num_agents": self.num_agents,
            "layer_count": self.layer_count,
            "confidence": f"{self.confidence_score:.2f}",
            "execution_time_ms": f"{self.execution_time_ms:.1f}"
        }


class MixtureOfAgents:
    """
    Mixture of Agents implementation with multiple aggregation strategies.
    
    This implementation provides:
    1. Multiple proposer agents generating diverse responses
    2. Aggregator agent synthesizing best combined response
    3. Layered architecture for iterative refinement
    4. Confidence-based weighting
    5. Multiple aggregation strategies
    """
    
    def __init__(
        self,
        num_proposers: int = 3,
        aggregation_strategy: AggregationStrategy = AggregationStrategy.SYNTHESIS,
        temperature: float = 0.7,
        use_layers: bool = True,
        num_layers: int = 2
    ):
        self.num_proposers = num_proposers
        self.aggregation_strategy = aggregation_strategy
        self.temperature = temperature
        self.use_layers = use_layers
        self.num_layers = num_layers if use_layers else 1
        
        # Create proposer agents with varied temperatures for diversity
        self.proposers = [
            ChatOpenAI(
                model_name="gpt-3.5-turbo",
                temperature=temperature + (i * 0.1)  # Vary temperature
            )
            for i in range(num_proposers)
        ]
        
        # Aggregator agent (lower temperature for focused synthesis)
        self.aggregator = ChatOpenAI(
            model_name="gpt-3.5-turbo",
            temperature=0.3
        )
    
    def _generate_proposer_response(
        self,
        agent_id: str,
        llm: ChatOpenAI,
        query: str,
        context: Optional[str] = None
    ) -> AgentOutput:
        """Generate response from a proposer agent"""
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", """You are an expert assistant providing thoughtful, comprehensive answers.
Provide your best response to the user's question."""),
            ("user", "{query}")
        ])
        
        if context:
            prompt = ChatPromptTemplate.from_messages([
                ("system", """You are an expert assistant. You have access to previous responses from other agents.
Consider their insights but provide your own unique perspective and improvements.

Previous responses:
{context}"""),
                ("user", "Now provide your response to: {query}")
            ])
        
        try:
            chain = prompt | llm | StrOutputParser()
            
            if context:
                response = chain.invoke({"query": query, "context": context})
            else:
                response = chain.invoke({"query": query})
            
            # Estimate confidence based on response length and coherence
            confidence = min(0.9, 0.5 + (len(response) / 1000))
            
            return AgentOutput(
                agent_id=agent_id,
                agent_role=AgentRole.PROPOSER,
                response=response,
                confidence=confidence
            )
            
        except Exception as e:
            return AgentOutput(
                agent_id=agent_id,
                agent_role=AgentRole.PROPOSER,
                response=f"Error: {str(e)}",
                confidence=0.0,
                metadata={"error": True}
            )
    
    def _aggregate_synthesis(
        self,
        query: str,
        agent_outputs: List[AgentOutput]
    ) -> Tuple[str, str, float]:
        """Synthesize multiple responses into one cohesive answer"""
        
        # Prepare responses for aggregator
        responses_text = "\n\n".join([
            f"Agent {i+1} (Confidence: {output.confidence:.2f}):\n{output.response}"
            for i, output in enumerate(agent_outputs)
            if not output.metadata.get("error")
        ])
        
        synthesis_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are an expert aggregator. You receive multiple responses to the same question.

Your task:
1. Identify the best insights from each response
2. Synthesize them into a single, comprehensive answer
3. Resolve any contradictions
4. Provide a cohesive, well-structured response

Be concise but thorough. Leverage the strengths of each response."""),
            ("user", """Question: {query}

Multiple Agent Responses:
{responses}

Synthesize these into the best possible answer:""")
        ])
        
        chain = synthesis_prompt | self.aggregator | StrOutputParser()
        
        try:
            final_response = chain.invoke({
                "query": query,
                "responses": responses_text
            })
            
            # Calculate confidence as weighted average
            valid_outputs = [o for o in agent_outputs if not o.metadata.get("error")]
            avg_confidence = sum(o.confidence for o in valid_outputs) / len(valid_outputs) if valid_outputs else 0.0
            
            reasoning = "Synthesized best insights from all agents"
            
            return final_response, reasoning, avg_confidence
            
        except Exception as e:
            # Fallback to best individual response
            best = max(agent_outputs, key=lambda x: x.confidence)
            return best.response, f"Synthesis failed: {str(e)}", best.confidence
    
    def _aggregate_voting(
        self,
        query: str,
        agent_outputs: List[AgentOutput]
    ) -> Tuple[str, str, float]:
        """Use voting to select best response"""
        
        # For voting, we'll use the aggregator to rank responses
        responses_text = "\n\n".join([
            f"Response {i+1}:\n{output.response}"
            for i, output in enumerate(agent_outputs)
            if not output.metadata.get("error")
        ])
        
        voting_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are evaluating multiple responses to select the best one.

Rank the responses and select the winner. Explain your choice briefly.

Format:
Winner: [Response number]
Reason: [Brief explanation]"""),
            ("user", """Question: {query}

Responses to evaluate:
{responses}

Select the best response:""")
        ])
        
        chain = voting_prompt | self.aggregator | StrOutputParser()
        
        try:
            evaluation = chain.invoke({
                "query": query,
                "responses": responses_text
            })
            
            # Parse winner (simplified)
            winner_idx = 0
            for i in range(len(agent_outputs)):
                if f"Response {i+1}" in evaluation or f"#{i+1}" in evaluation:
                    winner_idx = i
                    break
            
            winner = agent_outputs[winner_idx]
            
            return winner.response, evaluation, winner.confidence
            
        except Exception as e:
            best = max(agent_outputs, key=lambda x: x.confidence)
            return best.response, f"Voting failed: {str(e)}", best.confidence
    
    def _aggregate_weighted(
        self,
        query: str,
        agent_outputs: List[AgentOutput]
    ) -> Tuple[str, str, float]:
        """Weighted combination based on confidence scores"""
        
        # Similar to synthesis but explicitly weights by confidence
        valid_outputs = [o for o in agent_outputs if not o.metadata.get("error")]
        
        if not valid_outputs:
            return "No valid responses", "All agents failed", 0.0
        
        # Sort by confidence
        sorted_outputs = sorted(valid_outputs, key=lambda x: x.confidence, reverse=True)
        
        # Give more weight to high-confidence responses
        weighted_text = "\n\n".join([
            f"Agent {i+1} (High Confidence - Weight: {output.confidence:.2f}):\n{output.response}"
            if output.confidence > 0.7 else
            f"Agent {i+1} (Moderate Confidence - Weight: {output.confidence:.2f}):\n{output.response}"
            for i, output in enumerate(sorted_outputs)
        ])
        
        weighted_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are combining multiple responses weighted by their confidence scores.

Give more importance to high-confidence responses but don't ignore lower-confidence ones
if they contain unique valuable insights.

Produce a balanced synthesis."""),
            ("user", """Question: {query}

Weighted Responses:
{responses}

Produce weighted synthesis:""")
        ])
        
        chain = weighted_prompt | self.aggregator | StrOutputParser()
        
        final_response = chain.invoke({
            "query": query,
            "responses": weighted_text
        })
        
        # Confidence is weighted average favoring high-confidence responses
        total_weight = sum(o.confidence ** 2 for o in sorted_outputs)
        weighted_confidence = sum(
            (o.confidence ** 2) * o.confidence for o in sorted_outputs
        ) / total_weight if total_weight > 0 else 0.5
        
        reasoning = "Weighted synthesis favoring high-confidence responses"
        
        return final_response, reasoning, weighted_confidence
    
    def _aggregate_sequential(
        self,
        query: str,
        agent_outputs: List[AgentOutput]
    ) -> Tuple[str, str, float]:
        """Sequential refinement - each response builds on previous"""
        
        # Start with first response
        current_response = agent_outputs[0].response
        
        for i, output in enumerate(agent_outputs[1:], 1):
            if output.metadata.get("error"):
                continue
            
            refine_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are refining an answer by incorporating new insights.

Current answer:
{current}

New perspective:
{new}

Produce an improved, refined answer that combines both:"""),
                ("user", "Refined answer:")
            ])
            
            chain = refine_prompt | self.aggregator | StrOutputParser()
            
            current_response = chain.invoke({
                "current": current_response,
                "new": output.response
            })
        
        avg_confidence = sum(o.confidence for o in agent_outputs) / len(agent_outputs)
        reasoning = f"Sequential refinement through {len(agent_outputs)} agents"
        
        return current_response, reasoning, avg_confidence
    
    def execute(self, query: str) -> MoAResult:
        """Execute Mixture of Agents with multiple layers"""
        
        start_time = time.time()
        
        all_agent_outputs = []
        current_context = None
        
        # Execute layers
        for layer in range(self.num_layers):
            layer_outputs = []
            
            # Generate responses from all proposers
            for i, proposer in enumerate(self.proposers):
                agent_id = f"layer{layer+1}_agent{i+1}"
                
                output = self._generate_proposer_response(
                    agent_id,
                    proposer,
                    query,
                    context=current_context
                )
                
                layer_outputs.append(output)
                all_agent_outputs.append(output)
            
            # Prepare context for next layer
            if layer < self.num_layers - 1:
                valid_responses = [
                    o.response for o in layer_outputs
                    if not o.metadata.get("error")
                ]
                current_context = "\n\n".join(valid_responses[:2])  # Top 2 responses
        
        # Aggregate final responses
        final_layer_outputs = all_agent_outputs[-self.num_proposers:]
        
        if self.aggregation_strategy == AggregationStrategy.SYNTHESIS:
            final_response, reasoning, confidence = self._aggregate_synthesis(
                query, final_layer_outputs
            )
        elif self.aggregation_strategy == AggregationStrategy.VOTING:
            final_response, reasoning, confidence = self._aggregate_voting(
                query, final_layer_outputs
            )
        elif self.aggregation_strategy == AggregationStrategy.WEIGHTED:
            final_response, reasoning, confidence = self._aggregate_weighted(
                query, final_layer_outputs
            )
        elif self.aggregation_strategy == AggregationStrategy.SEQUENTIAL:
            final_response, reasoning, confidence = self._aggregate_sequential(
                query, final_layer_outputs
            )
        else:  # BEST_OF
            best = max(final_layer_outputs, key=lambda x: x.confidence)
            final_response = best.response
            reasoning = f"Selected best individual response (confidence: {best.confidence:.2f})"
            confidence = best.confidence
        
        execution_time_ms = (time.time() - start_time) * 1000
        
        return MoAResult(
            query=query,
            final_response=final_response,
            aggregation_strategy=self.aggregation_strategy,
            agent_outputs=all_agent_outputs,
            aggregator_reasoning=reasoning,
            confidence_score=confidence,
            execution_time_ms=execution_time_ms,
            layer_count=self.num_layers
        )


def demonstrate_mixture_of_agents():
    """Demonstrate Mixture of Agents pattern"""
    
    print("=" * 80)
    print("PATTERN 053: MIXTURE OF AGENTS (MoA) DEMONSTRATION")
    print("=" * 80)
    print("\nCombining multiple specialized agents for superior results\n")
    
    # Test 1: Basic MoA with synthesis
    print("\n" + "=" * 80)
    print("TEST 1: Basic MoA with Synthesis Aggregation")
    print("=" * 80)
    
    moa1 = MixtureOfAgents(
        num_proposers=3,
        aggregation_strategy=AggregationStrategy.SYNTHESIS,
        use_layers=False
    )
    
    query1 = "What are the pros and cons of remote work?"
    
    print(f"\nQuery: {query1}")
    print(f"Number of Proposer Agents: {moa1.num_proposers}")
    print(f"Aggregation Strategy: {moa1.aggregation_strategy.value}")
    
    result1 = moa1.execute(query1)
    
    print(f"\n📊 Individual Agent Responses:")
    for i, output in enumerate(result1.agent_outputs, 1):
        print(f"\n  Agent {i} (Confidence: {output.confidence:.2f}):")
        print(f"    {output.response[:120]}...")
    
    print(f"\n✨ Final Synthesized Response:")
    print(f"  {result1.final_response[:300]}...")
    print(f"\n  Aggregator Reasoning: {result1.aggregator_reasoning}")
    print(f"  Final Confidence: {result1.confidence_score:.2f}")
    print(f"  Execution Time: {result1.execution_time_ms:.1f}ms")
    
    # Test 2: Layered MoA
    print("\n" + "=" * 80)
    print("TEST 2: Layered MoA (Iterative Refinement)")
    print("=" * 80)
    
    moa2 = MixtureOfAgents(
        num_proposers=3,
        aggregation_strategy=AggregationStrategy.SYNTHESIS,
        use_layers=True,
        num_layers=2
    )
    
    query2 = "Explain quantum computing in simple terms"
    
    print(f"\nQuery: {query2}")
    print(f"Architecture: {moa2.num_layers} layers with {moa2.num_proposers} agents per layer")
    
    result2 = moa2.execute(query2)
    
    print(f"\n📊 Layer-by-Layer Generation:")
    layer1_agents = result2.agent_outputs[:moa2.num_proposers]
    layer2_agents = result2.agent_outputs[moa2.num_proposers:]
    
    print(f"\n  Layer 1 (Initial Responses):")
    for i, output in enumerate(layer1_agents, 1):
        print(f"    Agent {i}: {output.response[:80]}...")
    
    if layer2_agents:
        print(f"\n  Layer 2 (Refined with Context):")
        for i, output in enumerate(layer2_agents, 1):
            print(f"    Agent {i}: {output.response[:80]}...")
    
    print(f"\n✨ Final Layered Response:")
    print(f"  {result2.final_response[:300]}...")
    print(f"  Confidence: {result2.confidence_score:.2f}")
    
    # Test 3: Different aggregation strategies
    print("\n" + "=" * 80)
    print("TEST 3: Comparing Aggregation Strategies")
    print("=" * 80)
    
    query3 = "What's the best way to learn programming?"
    
    strategies = [
        AggregationStrategy.SYNTHESIS,
        AggregationStrategy.VOTING,
        AggregationStrategy.WEIGHTED
    ]
    
    print(f"\nQuery: {query3}")
    print(f"\nTesting different aggregation strategies...")
    
    results = []
    
    for strategy in strategies:
        moa = MixtureOfAgents(
            num_proposers=3,
            aggregation_strategy=strategy,
            use_layers=False
        )
        
        result = moa.execute(query3)
        results.append((strategy, result))
        
        print(f"\n  {strategy.value.upper()}:")
        print(f"    Response: {result.final_response[:100]}...")
        print(f"    Confidence: {result.confidence_score:.2f}")
        print(f"    Reasoning: {result.aggregator_reasoning[:80]}...")
    
    # Summary
    print("\n" + "=" * 80)
    print("MIXTURE OF AGENTS PATTERN SUMMARY")
    print("=" * 80)
    print("""
Key Benefits:
1. Diverse Perspectives: Multiple agents provide varied viewpoints
2. Error Reduction: Aggregation filters out individual mistakes
3. Robustness: Less sensitive to single agent failures
4. Quality Improvement: Combined output better than any individual
5. Expertise Combination: Leverage multiple specializations

Implementation Features:
1. Multiple proposer agents with diverse configurations
2. Aggregator agent for intelligent synthesis
3. Layered architecture for iterative refinement
4. Confidence-based weighting
5. Multiple aggregation strategies

Aggregation Strategies:
1. Synthesis: Combines best insights from all responses
2. Voting: Democratic selection of best response
3. Weighted: Confidence-based combination
4. Sequential: Iterative refinement chain
5. Best-of: Simple selection of highest quality

Layered Architecture:
- Layer 1: Initial independent responses
- Layer 2+: Responses informed by previous layer
- Each layer refines and improves outputs
- Context flows from layer to layer

Use Cases:
- Complex questions requiring multiple perspectives
- Creative tasks needing diverse ideas
- Technical problems with multiple solutions
- Research synthesis from various sources
- Decision-making with high stakes
- Content generation requiring variety

Best Practices:
1. Use 3-5 proposer agents (balance diversity vs cost)
2. Vary agent configurations for diversity
3. Match aggregation strategy to use case
4. Use layers for complex reasoning tasks
5. Monitor confidence scores for quality
6. A/B test different strategies
7. Cache common query results

Production Considerations:
- Parallel execution for latency (async)
- Cost vs quality tradeoffs
- Model diversity (different providers)
- Caching for repeated queries
- Load balancing across agents
- Timeout handling
- Fallback strategies

Comparison with Related Patterns:
- vs. Ensemble: Synthesis vs simple voting
- vs. Redundancy: Quality improvement vs fault tolerance
- vs. Multi-Agent Debate: Aggregation vs deliberation
- vs. Leader-Follower: Parallel vs hierarchical

The Mixture of Agents pattern is powerful for tasks requiring high quality
outputs that benefit from multiple perspectives and expert combination.
""")


if __name__ == "__main__":
    demonstrate_mixture_of_agents()

    

Agentic Design Pattern: Mixture Of Agents

Pattern 053: Mixture Of Agents – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Mixture Of Agents pattern implementation.

---

Pattern Description

Pattern 053: Mixture of Agents (MoA)

Description: The Mixture of Agents pattern combines outputs from multiple specialized agents using learned or rule-based aggregation strategies. Each agent brings specific expertise, and their outputs are intelligently combined to produce superior results that leverage the strengths of each individual agent.

Components: 1. Specialized Agents: Multiple agents with different capabilities/models 2. Proposer Agents: Generate initial candidate responses 3. Aggregator Agent: Combines and synthesizes multiple responses 4. Confidence Scoring: Weights agent contributions 5. Quality Evaluation: Assesses individual and combined outputs

Use Cases: – Complex tasks requiring diverse expertise – Multi-domain question answering – Code generation with multiple approaches – Creative content requiring varied perspectives – Research synthesis from multiple sources – Decision-making with multiple viewpoints

LangChain Implementation: Uses multiple LLM instances with different configurations or models, then aggregates their outputs using a meta-agent that synthesizes the best aspects of each response.

---

File Statistics

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

---

Module Documentation

"""
Pattern 053: Mixture of Agents (MoA)

Description:
    The Mixture of Agents pattern combines outputs from multiple specialized agents
    using learned or rule-based aggregation strategies. Each agent brings specific
    expertise, and their outputs are intelligently combined to produce superior
    results that leverage the strengths of each individual agent.

Components:
    1. Specialized Agents: Multiple agents with different capabilities/models
    2. Proposer Agents: Generate initial candidate responses
    3. Aggregator Agent: Combines and synthesizes multiple responses
    4. Confidence Scoring: Weights agent contributions
    5. Quality Evaluation: Assesses individual and combined outputs

Use Cases:
    - Complex tasks requiring diverse expertise
    - Multi-domain question answering
    - Code generation with multiple approaches
    - Creative content requiring varied perspectives
    - Research synthesis from multiple sources
    - Decision-making with multiple viewpoints

LangChain Implementation:
    Uses multiple LLM instances with different configurations or models,
    then aggregates their outputs using a meta-agent that synthesizes
    the best aspects of each response.
"""

Purpose: This module implements the Mixture Of Agents pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

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

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

LangChain Imports

from langchain_openai.ChatOpenAI

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

Lines 46-51

Inherits from: Enum

Purpose:

Roles for agents in MoA

---

Class: AggregationStrategy

Lines 54-60

Inherits from: Enum

Purpose:

Strategies for combining agent outputs

---

Class: AgentOutput

Lines 64-80

Purpose:

Output from a single agent

Methods:

to_dict(self)

Lines 73-80

Purpose: To Dict operation.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D68EF90>

---

Class: MoAResult

Lines 84-108

Purpose:

Result from Mixture of Agents execution

Methods:

num_agents(self)

Lines 96-97

Decorators: property

Purpose: Num Agents operation.

Parameters:

Returns: int

to_dict(self)

Lines 99-108

Purpose: To Dict operation.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D17C750>

---

Class: MixtureOfAgents

Lines 111-475

Purpose:

Mixture of Agents implementation with multiple aggregation strategies.

This implementation provides:
1. Multiple proposer agents generating diverse responses
2. Aggregator agent synthesizing best combined response
3. Layered architecture for iterative refinement
4. Confidence-based weighting
5. Multiple aggregation strategies

Methods:

__init__(self, num_proposers, aggregation_strategy, temperature, use_layers, num_layers)

Lines 123-150

Purpose: Initialize the class instance with required parameters.

Parameters: – num_proposers: Num proposers – aggregation_strategy: Aggregation strategy – temperature: Sampling temperature (0-1) – use_layers: Use layers – num_layers: Num layers

_generate_proposer_response(self, agent_id, llm, query, context)

Lines 152-202

Purpose:

Generate response from a proposer agent

Parameters: – agent_id: Agent id – llm: Language model – query: Input query or question – context: Context

Returns: AgentOutput

_aggregate_synthesis(self, query, agent_outputs)

Lines 204-255

Purpose:

Synthesize multiple responses into one cohesive answer

Parameters: – query: Input query or question – agent_outputs: Agent outputs

Returns: <ast.Subscript object at 0x000001A06D614950>

_aggregate_voting(self, query, agent_outputs)

Lines 257-308

Purpose:

Use voting to select best response

Parameters: – query: Input query or question – agent_outputs: Agent outputs

Returns: <ast.Subscript object at 0x000001A06D674A50>

_aggregate_weighted(self, query, agent_outputs)

Lines 310-364

Purpose:

Weighted combination based on confidence scores

Parameters: – query: Input query or question – agent_outputs: Agent outputs

Returns: <ast.Subscript object at 0x000001A06D73B0D0>

_aggregate_sequential(self, query, agent_outputs)

Lines 366-403

Purpose:

Sequential refinement - each response builds on previous

Parameters: – query: Input query or question – agent_outputs: Agent outputs

Returns: <ast.Subscript object at 0x000001A06D622590>

execute(self, query)

Lines 405-475

Purpose:

Execute Mixture of Agents with multiple layers

Parameters: – query: Input query or question

Returns: MoAResult

---

Functions

demonstrate_mixture_of_agents()

Lines 478-651

Purpose:

Demonstrate Mixture of Agents pattern

---

Key Concepts Summary

Pattern Overview

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

Core Components

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

When to Use

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

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

---

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

---

LangChain Components Used

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

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

---

Execution Flow

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

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

---

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

---

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

---

Related Patterns

• Pattern 052
• Pattern 054

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 Mixture Of Agents 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: 053_mixture_of_agents.py
Total Lines: 656
Pattern Category: AI Agent Design Pattern #053

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 052: Redundancy & Consensus

Description:
    The Redundancy & Consensus pattern executes the same operation across multiple
    independent agents or models, then uses voting mechanisms to reach consensus.
    This pattern provides fault tolerance, error reduction, and improved reliability
    through N-version programming and majority voting.

Components:
    1. Redundant Executors: Multiple independent agents/models
    2. Voting Mechanism: Majority vote, weighted vote, ranked choice
    3. Consensus Builder: Aggregates results to reach agreement
    4. Disagreement Resolver: Handles cases with no clear consensus
    5. Quality Scorer: Evaluates individual responses

Use Cases:
    - Critical decisions requiring high reliability
    - Error-prone tasks needing validation
    - Safety-critical applications
    - Medical diagnosis support
    - Financial predictions
    - Content moderation at scale

LangChain Implementation:
    Uses multiple LLM instances or models to generate responses independently,
    then applies voting algorithms to reach consensus on the final answer.
"""

import os
from typing import Dict, Any, List, Optional, Tuple, Callable
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import Counter

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 VotingStrategy(Enum):
    """Voting strategies for consensus"""
    MAJORITY = "majority"  # Simple majority wins
    UNANIMOUS = "unanimous"  # All must agree
    WEIGHTED = "weighted"  # Weighted by confidence/quality
    RANKED_CHOICE = "ranked_choice"  # Ranked voting
    THRESHOLD = "threshold"  # Must meet minimum agreement threshold


class AgreementLevel(Enum):
    """Level of agreement among voters"""
    UNANIMOUS = "unanimous"  # 100% agreement
    STRONG_CONSENSUS = "strong_consensus"  # 80%+
    CONSENSUS = "consensus"  # 60-79%
    MAJORITY = "majority"  # 51-59%
    SPLIT = "split"  # No clear majority
    DISAGREEMENT = "disagreement"  # High conflict


@dataclass
class AgentResponse:
    """Response from a single agent"""
    agent_id: str
    response: str
    confidence: float  # 0.0-1.0
    execution_time_ms: float
    model_name: str
    metadata: Dict[str, Any] = field(default_factory=dict)
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "agent_id": self.agent_id,
            "response": self.response[:100] + "..." if len(self.response) > 100 else self.response,
            "confidence": f"{self.confidence:.2f}",
            "execution_time_ms": f"{self.execution_time_ms:.1f}",
            "model": self.model_name
        }


@dataclass
class VoteResult:
    """Result of voting process"""
    winner: str
    vote_count: int
    total_votes: int
    agreement_level: AgreementLevel
    confidence: float
    all_responses: List[str]
    vote_distribution: Dict[str, int]
    
    @property
    def agreement_percentage(self) -> float:
        return (self.vote_count / max(1, self.total_votes)) * 100
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "winner": self.winner[:100] + "..." if len(self.winner) > 100 else self.winner,
            "votes": f"{self.vote_count}/{self.total_votes}",
            "agreement": f"{self.agreement_percentage:.1f}%",
            "agreement_level": self.agreement_level.value,
            "confidence": f"{self.confidence:.2f}",
            "unique_responses": len(set(self.all_responses)),
            "vote_distribution": self.vote_distribution
        }


@dataclass
class ConsensusResult:
    """Result from consensus-based execution"""
    query: str
    final_answer: str
    voting_strategy: VotingStrategy
    vote_result: VoteResult
    individual_responses: List[AgentResponse]
    execution_time_ms: float
    redundancy_factor: int
    disagreements: List[str]
    
    @property
    def reliability_score(self) -> float:
        """Calculate reliability based on agreement and confidence"""
        agreement_factor = self.vote_result.agreement_percentage / 100
        confidence_factor = self.vote_result.confidence
        redundancy_factor = min(1.0, self.redundancy_factor / 5)  # Cap at 5
        
        return (agreement_factor * 0.5 + confidence_factor * 0.3 + redundancy_factor * 0.2)
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "query": self.query[:80] + "..." if len(self.query) > 80 else self.query,
            "final_answer": self.final_answer[:150] + "..." if len(self.final_answer) > 150 else self.final_answer,
            "voting_strategy": self.voting_strategy.value,
            "vote_result": self.vote_result.to_dict(),
            "execution_time_ms": f"{self.execution_time_ms:.1f}",
            "redundancy_factor": self.redundancy_factor,
            "reliability_score": f"{self.reliability_score:.2f}",
            "had_disagreements": len(self.disagreements) > 0
        }


class ConsensusAgent:
    """
    Agent that uses redundancy and voting for reliable decisions.
    
    This implementation provides:
    1. Multiple independent executions (N-version programming)
    2. Various voting strategies (majority, weighted, unanimous)
    3. Consensus building with conflict resolution
    4. Reliability scoring based on agreement
    5. Disagreement detection and handling
    """
    
    def __init__(
        self,
        model_name: str = "gpt-3.5-turbo",
        redundancy_factor: int = 3,
        voting_strategy: VotingStrategy = VotingStrategy.MAJORITY,
        temperature: float = 0.7,
        min_confidence_threshold: float = 0.6
    ):
        self.model_name = model_name
        self.redundancy_factor = redundancy_factor
        self.voting_strategy = voting_strategy
        self.temperature = temperature
        self.min_confidence_threshold = min_confidence_threshold
        
        # Create multiple independent agents
        self.agents = [
            ChatOpenAI(model_name=model_name, temperature=temperature)
            for _ in range(redundancy_factor)
        ]
    
    def _execute_single_agent(
        self,
        agent_id: str,
        llm: ChatOpenAI,
        query: str
    ) -> AgentResponse:
        """Execute query on single agent"""
        import time
        
        start_time = time.time()
        
        try:
            response = llm.invoke(query)
            execution_time_ms = (time.time() - start_time) * 1000
            
            # Estimate confidence (simplified - production would use actual model confidence)
            confidence = 0.7 + (0.3 * (len(response.content) / max(1, len(query))))
            confidence = min(1.0, confidence)
            
            return AgentResponse(
                agent_id=agent_id,
                response=response.content,
                confidence=confidence,
                execution_time_ms=execution_time_ms,
                model_name=self.model_name
            )
            
        except Exception as e:
            execution_time_ms = (time.time() - start_time) * 1000
            
            return AgentResponse(
                agent_id=agent_id,
                response=f"ERROR: {str(e)}",
                confidence=0.0,
                execution_time_ms=execution_time_ms,
                model_name=self.model_name,
                metadata={"error": True}
            )
    
    def _normalize_response(self, response: str) -> str:
        """Normalize response for comparison"""
        # Remove extra whitespace, lowercase, strip punctuation
        normalized = response.lower().strip()
        normalized = " ".join(normalized.split())
        return normalized
    
    def _calculate_similarity(self, resp1: str, resp2: str) -> float:
        """Calculate similarity between two responses"""
        # Simple word overlap similarity
        words1 = set(self._normalize_response(resp1).split())
        words2 = set(self._normalize_response(resp2).split())
        
        if not words1 or not words2:
            return 0.0
        
        intersection = words1.intersection(words2)
        union = words1.union(words2)
        
        return len(intersection) / len(union) if union else 0.0
    
    def _majority_vote(
        self,
        responses: List[AgentResponse],
        similarity_threshold: float = 0.7
    ) -> VoteResult:
        """Majority voting with similarity grouping"""
        # Group similar responses
        groups: List[List[AgentResponse]] = []
        
        for response in responses:
            added_to_group = False
            
            for group in groups:
                # Check similarity with group representative
                similarity = self._calculate_similarity(
                    response.response,
                    group[0].response
                )
                
                if similarity >= similarity_threshold:
                    group.append(response)
                    added_to_group = True
                    break
            
            if not added_to_group:
                groups.append([response])
        
        # Find largest group
        largest_group = max(groups, key=len)
        vote_count = len(largest_group)
        total_votes = len(responses)
        
        # Select best response from winning group (highest confidence)
        winner = max(largest_group, key=lambda r: r.confidence)
        
        # Calculate average confidence of winning group
        avg_confidence = sum(r.confidence for r in largest_group) / len(largest_group)
        
        # Determine agreement level
        agreement_pct = (vote_count / total_votes) * 100
        
        if agreement_pct == 100:
            agreement_level = AgreementLevel.UNANIMOUS
        elif agreement_pct >= 80:
            agreement_level = AgreementLevel.STRONG_CONSENSUS
        elif agreement_pct >= 60:
            agreement_level = AgreementLevel.CONSENSUS
        elif agreement_pct >= 51:
            agreement_level = AgreementLevel.MAJORITY
        elif len(groups) == len(responses):
            agreement_level = AgreementLevel.DISAGREEMENT
        else:
            agreement_level = AgreementLevel.SPLIT
        
        # Create vote distribution
        vote_distribution = {}
        for i, group in enumerate(groups, 1):
            key = f"response_group_{i}"
            vote_distribution[key] = len(group)
        
        return VoteResult(
            winner=winner.response,
            vote_count=vote_count,
            total_votes=total_votes,
            agreement_level=agreement_level,
            confidence=avg_confidence,
            all_responses=[r.response for r in responses],
            vote_distribution=vote_distribution
        )
    
    def _weighted_vote(self, responses: List[AgentResponse]) -> VoteResult:
        """Weighted voting based on confidence scores"""
        # Weight responses by confidence
        weighted_responses = []
        
        for response in responses:
            # Weight = confidence score
            weight = response.confidence
            weighted_responses.append((response, weight))
        
        # Find response with highest total weight (considering similarity)
        best_response = None
        best_weight = 0.0
        
        for response, base_weight in weighted_responses:
            total_weight = base_weight
            
            # Add weights from similar responses
            for other_response, other_weight in weighted_responses:
                if response.agent_id != other_response.agent_id:
                    similarity = self._calculate_similarity(
                        response.response,
                        other_response.response
                    )
                    if similarity >= 0.7:
                        total_weight += other_weight * similarity
            
            if total_weight > best_weight:
                best_weight = total_weight
                best_response = response
        
        # Calculate agreement level based on weight distribution
        total_weight = sum(w for _, w in weighted_responses)
        agreement_pct = (best_weight / total_weight) * 100
        
        if agreement_pct >= 90:
            agreement_level = AgreementLevel.STRONG_CONSENSUS
        elif agreement_pct >= 70:
            agreement_level = AgreementLevel.CONSENSUS
        elif agreement_pct >= 50:
            agreement_level = AgreementLevel.MAJORITY
        else:
            agreement_level = AgreementLevel.SPLIT
        
        return VoteResult(
            winner=best_response.response,
            vote_count=1,  # Weighted doesn't use simple counts
            total_votes=len(responses),
            agreement_level=agreement_level,
            confidence=best_response.confidence,
            all_responses=[r.response for r in responses],
            vote_distribution={"weighted_winner": 1}
        )
    
    def _unanimous_vote(self, responses: List[AgentResponse]) -> VoteResult:
        """Unanimous voting - all must agree"""
        # Check if all responses are similar
        if not responses:
            raise ValueError("No responses to vote on")
        
        reference = responses[0].response
        all_agree = all(
            self._calculate_similarity(r.response, reference) >= 0.9
            for r in responses[1:]
        )
        
        if all_agree:
            avg_confidence = sum(r.confidence for r in responses) / len(responses)
            
            return VoteResult(
                winner=reference,
                vote_count=len(responses),
                total_votes=len(responses),
                agreement_level=AgreementLevel.UNANIMOUS,
                confidence=avg_confidence,
                all_responses=[r.response for r in responses],
                vote_distribution={"unanimous": len(responses)}
            )
        else:
            # No consensus - return best individual response
            best = max(responses, key=lambda r: r.confidence)
            
            return VoteResult(
                winner=best.response,
                vote_count=1,
                total_votes=len(responses),
                agreement_level=AgreementLevel.DISAGREEMENT,
                confidence=best.confidence,
                all_responses=[r.response for r in responses],
                vote_distribution={"no_consensus": len(responses)}
            )
    
    def _apply_voting_strategy(
        self,
        responses: List[AgentResponse]
    ) -> VoteResult:
        """Apply configured voting strategy"""
        if self.voting_strategy == VotingStrategy.MAJORITY:
            return self._majority_vote(responses)
        elif self.voting_strategy == VotingStrategy.WEIGHTED:
            return self._weighted_vote(responses)
        elif self.voting_strategy == VotingStrategy.UNANIMOUS:
            return self._unanimous_vote(responses)
        else:
            # Default to majority
            return self._majority_vote(responses)
    
    def execute_with_consensus(self, query: str) -> ConsensusResult:
        """Execute query with redundancy and consensus"""
        import time
        
        start_time = time.time()
        
        # Execute across all redundant agents
        responses = []
        for i, agent in enumerate(self.agents):
            agent_id = f"agent_{i+1}"
            response = self._execute_single_agent(agent_id, agent, query)
            responses.append(response)
        
        # Filter out errors
        valid_responses = [r for r in responses if not r.metadata.get("error")]
        
        if not valid_responses:
            # All failed - return error consensus
            execution_time_ms = (time.time() - start_time) * 1000
            
            return ConsensusResult(
                query=query,
                final_answer="ERROR: All agents failed to respond",
                voting_strategy=self.voting_strategy,
                vote_result=VoteResult(
                    winner="ERROR",
                    vote_count=0,
                    total_votes=len(responses),
                    agreement_level=AgreementLevel.DISAGREEMENT,
                    confidence=0.0,
                    all_responses=[r.response for r in responses],
                    vote_distribution={"errors": len(responses)}
                ),
                individual_responses=responses,
                execution_time_ms=execution_time_ms,
                redundancy_factor=self.redundancy_factor,
                disagreements=[]
            )
        
        # Apply voting strategy
        vote_result = self._apply_voting_strategy(valid_responses)
        
        # Identify disagreements
        disagreements = []
        winner_normalized = self._normalize_response(vote_result.winner)
        
        for response in valid_responses:
            similarity = self._calculate_similarity(response.response, vote_result.winner)
            if similarity < 0.5:  # Significant disagreement
                disagreements.append(response.response[:100])
        
        execution_time_ms = (time.time() - start_time) * 1000
        
        return ConsensusResult(
            query=query,
            final_answer=vote_result.winner,
            voting_strategy=self.voting_strategy,
            vote_result=vote_result,
            individual_responses=responses,
            execution_time_ms=execution_time_ms,
            redundancy_factor=self.redundancy_factor,
            disagreements=disagreements
        )


def demonstrate_redundancy_consensus():
    """Demonstrate redundancy and consensus pattern"""
    
    print("=" * 80)
    print("PATTERN 052: REDUNDANCY & CONSENSUS DEMONSTRATION")
    print("=" * 80)
    print("\nDemonstrating fault-tolerant decision making through redundancy\n")
    
    # Test 1: Majority voting
    print("\n" + "=" * 80)
    print("TEST 1: Majority Voting with Redundancy")
    print("=" * 80)
    
    agent1 = ConsensusAgent(
        redundancy_factor=5,
        voting_strategy=VotingStrategy.MAJORITY,
        temperature=0.3  # Low temperature for more consistent responses
    )
    
    query1 = "What is the capital of France?"
    
    print(f"\nQuery: {query1}")
    print(f"Redundancy: {agent1.redundancy_factor} agents")
    print(f"Strategy: {agent1.voting_strategy.value}")
    
    result1 = agent1.execute_with_consensus(query1)
    
    print(f"\n✓ Consensus reached:")
    print(f"  Final Answer: {result1.final_answer}")
    print(f"  Agreement: {result1.vote_result.agreement_percentage:.1f}% ({result1.vote_result.agreement_level.value})")
    print(f"  Confidence: {result1.vote_result.confidence:.2f}")
    print(f"  Reliability Score: {result1.reliability_score:.2f}")
    print(f"  Execution Time: {result1.execution_time_ms:.1f}ms")
    
    print(f"\n  Individual Responses:")
    for i, resp in enumerate(result1.individual_responses, 1):
        print(f"    Agent {i}: {resp.response[:60]}... (confidence: {resp.confidence:.2f})")
    
    # Test 2: Weighted voting
    print("\n" + "=" * 80)
    print("TEST 2: Weighted Voting by Confidence")
    print("=" * 80)
    
    agent2 = ConsensusAgent(
        redundancy_factor=3,
        voting_strategy=VotingStrategy.WEIGHTED,
        temperature=0.5
    )
    
    query2 = "Is water wet?"
    
    print(f"\nQuery: {query2}")
    print(f"Strategy: {agent2.voting_strategy.value}")
    
    result2 = agent2.execute_with_consensus(query2)
    
    print(f"\n✓ Weighted consensus:")
    print(f"  Final Answer: {result2.final_answer[:120]}...")
    print(f"  Agreement Level: {result2.vote_result.agreement_level.value}")
    print(f"  Confidence: {result2.vote_result.confidence:.2f}")
    
    # Test 3: Unanimous voting (strict)
    print("\n" + "=" * 80)
    print("TEST 3: Unanimous Voting (Strict Consensus)")
    print("=" * 80)
    
    agent3 = ConsensusAgent(
        redundancy_factor=3,
        voting_strategy=VotingStrategy.UNANIMOUS,
        temperature=0.1  # Very low for consistency
    )
    
    query3 = "What is 2+2?"
    
    print(f"\nQuery: {query3}")
    print(f"Strategy: {agent3.voting_strategy.value}")
    
    result3 = agent3.execute_with_consensus(query3)
    
    print(f"\n✓ Unanimous check:")
    print(f"  Final Answer: {result3.final_answer}")
    print(f"  Agreement: {result3.vote_result.agreement_level.value}")
    print(f"  All Agreed: {result3.vote_result.agreement_level == AgreementLevel.UNANIMOUS}")
    
    # Test 4: Disagreement handling
    print("\n" + "=" * 80)
    print("TEST 4: Handling Disagreements")
    print("=" * 80)
    
    agent4 = ConsensusAgent(
        redundancy_factor=5,
        voting_strategy=VotingStrategy.MAJORITY,
        temperature=0.9  # High temperature for diverse responses
    )
    
    query4 = "What's the best programming language?"
    
    print(f"\nQuery: {query4}")
    print("(Intentionally subjective to generate disagreements)")
    
    result4 = agent4.execute_with_consensus(query4)
    
    print(f"\n📊 Consensus Analysis:")
    print(f"  Final Answer: {result4.final_answer[:100]}...")
    print(f"  Agreement: {result4.vote_result.agreement_percentage:.1f}%")
    print(f"  Agreement Level: {result4.vote_result.agreement_level.value}")
    print(f"  Unique Responses: {len(set(result4.vote_result.all_responses))}")
    print(f"  Disagreements Found: {len(result4.disagreements)}")
    
    if result4.disagreements:
        print(f"\n  Minority Opinions:")
        for i, disagreement in enumerate(result4.disagreements[:2], 1):
            print(f"    {i}. {disagreement}...")
    
    print(f"\n  Vote Distribution:")
    for group, count in result4.vote_result.vote_distribution.items():
        print(f"    {group}: {count} votes")
    
    # Test 5: Reliability comparison
    print("\n" + "=" * 80)
    print("TEST 5: Reliability Score Comparison")
    print("=" * 80)
    
    test_queries = [
        "What is the speed of light?",
        "Who wrote Hamlet?",
        "What's the meaning of life?"
    ]
    
    print(f"\nComparing reliability across different query types...")
    
    reliability_scores = []
    
    for query in test_queries:
        result = agent1.execute_with_consensus(query)
        reliability_scores.append((query, result.reliability_score, result.vote_result.agreement_level))
        
        print(f"\nQuery: {query}")
        print(f"  Reliability: {result.reliability_score:.2f}")
        print(f"  Agreement: {result.vote_result.agreement_level.value}")
        print(f"  Confidence: {result.vote_result.confidence:.2f}")
    
    # Show ranking
    print(f"\n📊 Reliability Ranking:")
    sorted_scores = sorted(reliability_scores, key=lambda x: x[1], reverse=True)
    
    for i, (query, score, agreement) in enumerate(sorted_scores, 1):
        print(f"  {i}. {query}")
        print(f"     Score: {score:.2f}, Agreement: {agreement.value}")
    
    # Summary
    print("\n" + "=" * 80)
    print("REDUNDANCY & CONSENSUS PATTERN SUMMARY")
    print("=" * 80)
    print("""
Key Benefits:
1. Fault Tolerance: Multiple agents provide backup if one fails
2. Error Reduction: Voting filters out individual mistakes
3. Reliability: Higher confidence through consensus
4. Robustness: Less sensitive to individual model quirks
5. Quality Assurance: Cross-validation of responses

Implementation Features:
1. N-version programming with multiple agents
2. Multiple voting strategies (majority, weighted, unanimous)
3. Similarity-based response grouping
4. Disagreement detection and tracking
5. Reliability scoring based on agreement and confidence
6. Configurable redundancy factor

Voting Strategies:
1. Majority Vote: Most common response wins
2. Weighted Vote: Responses weighted by confidence
3. Unanimous: All agents must agree (strictest)
4. Ranked Choice: Ordered preference voting
5. Threshold: Minimum agreement percentage required

Agreement Levels:
- Unanimous (100%): Perfect agreement
- Strong Consensus (80%+): Very high agreement
- Consensus (60-79%): Good agreement
- Majority (51-59%): Simple majority
- Split: No clear majority
- Disagreement: High conflict

Use Cases:
- Medical diagnosis support systems
- Financial predictions and recommendations
- Safety-critical decision making
- Content moderation at scale
- Fraud detection systems
- Legal document analysis
- Scientific research validation

Best Practices:
1. Use redundancy factor 3-7 (balance cost vs reliability)
2. Match voting strategy to use case criticality
3. Monitor disagreement patterns for insights
4. Consider computational cost vs reliability gain
5. Use lower temperature for factual queries
6. Track reliability scores over time
7. Escalate high-disagreement cases to humans
8. Document when consensus failed

Production Considerations:
- Parallel execution for latency (async processing)
- Cost vs reliability tradeoffs
- Caching for repeated queries
- Timeout handling for slow agents
- Load balancing across models
- Model diversity (different providers)
- Fallback strategies for no consensus
- A/B testing of voting strategies

Comparison with Related Patterns:
- vs. Ensemble: Voting vs averaging predictions
- vs. Self-Consistency: Multiple executions vs one with sampling
- vs. Multi-Agent Debate: Consensus vs deliberation
- vs. Fallback: Redundancy vs sequential alternatives

The Redundancy & Consensus pattern is essential for high-reliability AI systems
where decisions must be trustworthy and errors minimized through cross-validation
and democratic decision-making among multiple independent agents.
""")


if __name__ == "__main__":
    demonstrate_redundancy_consensus()

    

Agentic Design Pattern: Redundancy Consensus

Pattern 052: Redundancy Consensus – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Redundancy Consensus pattern implementation.

---

Pattern Description

Pattern 052: Redundancy & Consensus

Description: The Redundancy & Consensus pattern executes the same operation across multiple independent agents or models, then uses voting mechanisms to reach consensus. This pattern provides fault tolerance, error reduction, and improved reliability through N-version programming and majority voting.

Components: 1. Redundant Executors: Multiple independent agents/models 2. Voting Mechanism: Majority vote, weighted vote, ranked choice 3. Consensus Builder: Aggregates results to reach agreement 4. Disagreement Resolver: Handles cases with no clear consensus 5. Quality Scorer: Evaluates individual responses

Use Cases: – Critical decisions requiring high reliability – Error-prone tasks needing validation – Safety-critical applications – Medical diagnosis support – Financial predictions – Content moderation at scale

LangChain Implementation: Uses multiple LLM instances or models to generate responses independently, then applies voting algorithms to reach consensus on the final answer.

---

File Statistics

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

---

Module Documentation

"""
Pattern 052: Redundancy & Consensus

Description:
    The Redundancy & Consensus pattern executes the same operation across multiple
    independent agents or models, then uses voting mechanisms to reach consensus.
    This pattern provides fault tolerance, error reduction, and improved reliability
    through N-version programming and majority voting.

Components:
    1. Redundant Executors: Multiple independent agents/models
    2. Voting Mechanism: Majority vote, weighted vote, ranked choice
    3. Consensus Builder: Aggregates results to reach agreement
    4. Disagreement Resolver: Handles cases with no clear consensus
    5. Quality Scorer: Evaluates individual responses

Use Cases:
    - Critical decisions requiring high reliability
    - Error-prone tasks needing validation
    - Safety-critical applications
    - Medical diagnosis support
    - Financial predictions
    - Content moderation at scale

LangChain Implementation:
    Uses multiple LLM instances or models to generate responses independently,
    then applies voting algorithms to reach consensus on the final answer.
"""

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

---

Import Statements

Standard Library Imports

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

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

LangChain Imports

from langchain_openai.ChatOpenAI

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

Other Imports

from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: VotingStrategy

Lines 45-51

Inherits from: Enum

Purpose:

Voting strategies for consensus

---

Class: AgreementLevel

Lines 54-61

Inherits from: Enum

Purpose:

Level of agreement among voters

---

Class: AgentResponse

Lines 65-81

Purpose:

Response from a single agent

Methods:

to_dict(self)

Lines 74-81

Purpose: To Dict operation.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D68E510>

---

Class: VoteResult

Lines 85-108

Purpose:

Result of voting process

Methods:

agreement_percentage(self)

Lines 96-97

Decorators: property

Purpose: Agreement Percentage operation.

Parameters:

Returns: float

to_dict(self)

Lines 99-108

Purpose: To Dict operation.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D77CF10>

---

Class: ConsensusResult

Lines 112-142

Purpose:

Result from consensus-based execution

Methods:

reliability_score(self)

Lines 124-130

Decorators: property

Purpose:

Calculate reliability based on agreement and confidence

Parameters:

Returns: float

to_dict(self)

Lines 132-142

Purpose: To Dict operation.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D63AB10>

---

Class: ConsensusAgent

Lines 145-476

Purpose:

Agent that uses redundancy and voting for reliable decisions.

This implementation provides:
1. Multiple independent executions (N-version programming)
2. Various voting strategies (majority, weighted, unanimous)
3. Consensus building with conflict resolution
4. Reliability scoring based on agreement
5. Disagreement detection and handling

Methods:

__init__(self, model_name, redundancy_factor, voting_strategy, temperature, min_confidence_threshold)

Lines 157-175

Purpose: Initialize the class instance with required parameters.

Parameters: – model_name: Language model instance – redundancy_factor: Redundancy factor – voting_strategy: Voting strategy – temperature: Sampling temperature (0-1) – min_confidence_threshold: Min confidence threshold

_execute_single_agent(self, agent_id, llm, query)

Lines 177-214

Purpose:

Execute query on single agent

Parameters: – agent_id: Agent id – llm: Language model – query: Input query or question

Returns: AgentResponse

_normalize_response(self, response)

Lines 216-221

Purpose:

Normalize response for comparison

Parameters: – response: Response

Returns: str

_calculate_similarity(self, resp1, resp2)

Lines 223-235

Purpose:

Calculate similarity between two responses

Parameters: – resp1: Resp1 – resp2: Resp2

Returns: float

_majority_vote(self, responses, similarity_threshold)

Lines 237-305

Purpose:

Majority voting with similarity grouping

Parameters: – responses: Responses – similarity_threshold: Similarity threshold

Returns: VoteResult

_weighted_vote(self, responses)

Lines 307-359

Purpose:

Weighted voting based on confidence scores

Parameters: – responses: Responses

Returns: VoteResult

_unanimous_vote(self, responses)

Lines 361-397

Purpose:

Unanimous voting - all must agree

Parameters: – responses: Responses

Returns: VoteResult

_apply_voting_strategy(self, responses)

Lines 399-412

Purpose:

Apply configured voting strategy

Parameters: – responses: Responses

Returns: VoteResult

execute_with_consensus(self, query)

Lines 414-476

Purpose:

Execute query with redundancy and consensus

Parameters: – query: Input query or question

Returns: ConsensusResult

---

Functions

demonstrate_redundancy_consensus()

Lines 479-702

Purpose:

Demonstrate redundancy and consensus pattern

---

Key Concepts Summary

Pattern Overview

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

Core Components

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

When to Use

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

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

---

Best Practices Demonstrated

1. Environment Configuration

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

2. Error Handling

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

3. Prompt Engineering

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

4. Code Organization

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

5. Performance Optimization

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

---

LangChain Components Used

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

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

---

Execution Flow

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

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

---

Performance Considerations

Computational Cost

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

Optimization Strategies

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

Cost Estimates (Approximate)

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

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

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

2. Error Handling

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

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

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

6. Cost Management

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

---

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

---

Related Patterns

• Pattern 051
• Pattern 053

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 Redundancy Consensus 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: 052_redundancy_consensus.py
Total Lines: 707
Pattern Category: AI Agent Design Pattern #052

---

Generated from actual implementation code
For educational and reference purposes