Tag: artificial-intelligence

  • Agentic Design Pattern: Swarm Intelligence

    
        """
    Pattern 015: Swarm Intelligence
    
    Description:
        The Swarm Intelligence pattern uses many simple, decentralized agents that collaborate
        through local interactions to solve complex problems. Inspired by natural phenomena like
        ant colonies, bee swarms, and bird flocking, this pattern achieves emergent intelligent
        behavior without centralized control.
    
    Components:
        - Swarm Agents: Simple agents with local decision rules
        - Environment/Pheromone System: Shared information space
        - Local Interaction Rules: How agents influence each other
        - Emergent Behavior: Complex patterns from simple rules
    
    Use Cases:
        - Optimization problems (path finding, resource allocation)
        - Search and exploration tasks
        - Distributed problem-solving
        - Parallel hypothesis exploration
    
    LangChain Implementation:
        Uses multiple simple LLM agents with local decision-making, a shared information
        space (pheromone trails), iterative exploration rounds, and convergence detection
        to achieve swarm-based optimization.
    
    Key Features:
        - Decentralized decision-making
        - Pheromone-based communication
        - Iterative convergence
        - Parallel exploration
    """
    
    import os
    from typing import List, Dict, Any, Optional, Tuple
    from dataclasses import dataclass, field
    from collections import defaultdict
    from enum import Enum
    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()
    
    
    @dataclass
    class Solution:
        """Represents a solution explored by an agent."""
        agent_id: str
        content: str
        quality_score: float
        iteration: int
    
    
    @dataclass
    class PheromoneTrail:
        """Represents accumulated pheromone on a solution path."""
        solution_key: str
        strength: float
        contributors: List[str] = field(default_factory=list)
    
    
    class SwarmBehavior(Enum):
        """Types of swarm behaviors."""
        EXPLORATION = "exploration"  # Explore diverse solutions
        EXPLOITATION = "exploitation"  # Refine best solutions
        BALANCED = "balanced"  # Mix of both
    
    
    class SwarmAgent:
        """
        A simple swarm agent that makes local decisions based on pheromones.
        """
        
        def __init__(
            self,
            agent_id: str,
            model: str = "gpt-3.5-turbo",
            temperature: float = 0.8
        ):
            """
            Initialize a swarm agent.
            
            Args:
                agent_id: Unique identifier for this agent
                model: LLM model to use
                temperature: Creativity/exploration level
            """
            self.agent_id = agent_id
            self.llm = ChatOpenAI(model=model, temperature=temperature)
            self.local_memory: List[Solution] = []
        
        def explore_solution(
            self,
            problem: str,
            pheromone_trails: List[PheromoneTrail],
            behavior: SwarmBehavior = SwarmBehavior.BALANCED
        ) -> str:
            """
            Explore a solution based on local information and pheromones.
            
            Args:
                problem: Problem to solve
                pheromone_trails: Current pheromone information
                behavior: Exploration vs exploitation preference
                
            Returns:
                Proposed solution
            """
            # Build context from pheromone trails
            if pheromone_trails and behavior != SwarmBehavior.EXPLORATION:
                # Sort by strength and take top trails
                top_trails = sorted(pheromone_trails, key=lambda x: x.strength, reverse=True)[:3]
                trail_context = "\n".join([
                    f"- Strong path: {trail.solution_key} (strength: {trail.strength:.2f})"
                    for trail in top_trails
                ])
            else:
                trail_context = "No strong paths found yet. Explore freely."
            
            behavior_instructions = {
                SwarmBehavior.EXPLORATION: "Focus on exploring new, diverse approaches. Be creative and unconventional.",
                SwarmBehavior.EXPLOITATION: "Focus on refining and improving the strongest existing approaches.",
                SwarmBehavior.BALANCED: "Balance between exploring new ideas and building on promising existing approaches."
            }
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", f"""You are a swarm agent working as part of a collective intelligence.
    Your role is to propose solutions based on local information and pheromone trails left by other agents.
    
    {behavior_instructions[behavior]}
    
    Keep your solution concise (2-3 sentences)."""),
                ("user", """Problem: {problem}
    
    Pheromone trails from other agents:
    {trail_context}
    
    Propose your solution:""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            
            solution = chain.invoke({
                "problem": problem,
                "trail_context": trail_context
            })
            
            return solution.strip()
        
        def evaluate_solution(
            self,
            problem: str,
            solution: str
        ) -> float:
            """
            Evaluate the quality of a solution.
            
            Args:
                problem: Original problem
                solution: Proposed solution
                
            Returns:
                Quality score (0.0-1.0)
            """
            eval_prompt = ChatPromptTemplate.from_messages([
                ("system", "You are an evaluator. Rate solution quality from 0.0 to 1.0."),
                ("user", """Problem: {problem}
    
    Solution: {solution}
    
    Rate this solution's quality (0.0 = poor, 1.0 = excellent).
    Consider: correctness, completeness, creativity, practicality.
    
    Provide only a number between 0.0 and 1.0:""")
            ])
            
            chain = eval_prompt | self.llm | StrOutputParser()
            
            try:
                score_str = chain.invoke({"problem": problem, "solution": solution})
                score = float(score_str.strip())
                return max(0.0, min(1.0, score))
            except:
                return 0.5  # Default moderate score
    
    
    class SwarmIntelligence:
        """
        Swarm intelligence system coordinating multiple simple agents.
        
        Uses pheromone-based communication for decentralized problem-solving.
        """
        
        def __init__(
            self,
            num_agents: int = 10,
            evaporation_rate: float = 0.1,
            model: str = "gpt-3.5-turbo"
        ):
            """
            Initialize swarm intelligence system.
            
            Args:
                num_agents: Number of agents in the swarm
                evaporation_rate: Rate at which pheromones decay (0.0-1.0)
                model: LLM model for agents
            """
            self.num_agents = num_agents
            self.evaporation_rate = evaporation_rate
            
            # Create swarm agents with varying temperatures for diversity
            self.agents = [
                SwarmAgent(
                    agent_id=f"agent_{i}",
                    model=model,
                    temperature=0.6 + (i / num_agents) * 0.6  # Range: 0.6-1.2
                )
                for i in range(num_agents)
            ]
            
            # Pheromone trail system
            self.pheromones: Dict[str, PheromoneTrail] = {}
            
            # Solution history
            self.all_solutions: List[Solution] = []
        
        def _compute_solution_key(self, solution: str) -> str:
            """
            Compute a key for similar solutions.
            
            Uses first 50 characters as a simple similarity metric.
            In production, would use embeddings or semantic similarity.
            """
            # Simple approach: use first significant words
            words = solution.lower().split()[:8]
            return " ".join(words)
        
        def _deposit_pheromone(
            self,
            solution: Solution
        ):
            """
            Deposit pheromone on a solution path based on quality.
            
            Args:
                solution: Solution to deposit pheromone for
            """
            key = self._compute_solution_key(solution.content)
            
            if key in self.pheromones:
                # Strengthen existing trail
                self.pheromones[key].strength += solution.quality_score
                self.pheromones[key].contributors.append(solution.agent_id)
            else:
                # Create new trail
                self.pheromones[key] = PheromoneTrail(
                    solution_key=key,
                    strength=solution.quality_score,
                    contributors=[solution.agent_id]
                )
        
        def _evaporate_pheromones(self):
            """Evaporate pheromones to forget old solutions."""
            keys_to_remove = []
            
            for key, trail in self.pheromones.items():
                trail.strength *= (1 - self.evaporation_rate)
                
                # Remove very weak trails
                if trail.strength  SwarmBehavior:
            """
            Determine swarm behavior based on iteration.
            
            Early iterations: More exploration
            Later iterations: More exploitation
            """
            progress = iteration / max_iterations
            
            if progress  0.7:
                return SwarmBehavior.EXPLOITATION
            else:
                return SwarmBehavior.BALANCED
        
        def solve(
            self,
            problem: str,
            iterations: int = 5,
            agents_per_iteration: Optional[int] = None
        ) -> Tuple[Solution, List[Solution]]:
            """
            Solve a problem using swarm intelligence.
            
            Args:
                problem: Problem to solve
                iterations: Number of swarm iterations
                agents_per_iteration: Agents to activate per iteration (default: all)
                
            Returns:
                (best_solution, all_solutions)
            """
            if agents_per_iteration is None:
                agents_per_iteration = self.num_agents
            
            print(f"\n[Swarm] Starting with {self.num_agents} agents, {iterations} iterations")
            
            for iteration in range(iterations):
                behavior = self._determine_behavior(iteration, iterations)
                
                print(f"\n[Swarm] Iteration {iteration + 1}/{iterations} - Behavior: {behavior.value}")
                
                # Randomly select agents to activate this iteration
                active_agents = random.sample(self.agents, min(agents_per_iteration, self.num_agents))
                
                iteration_solutions = []
                
                for agent in active_agents:
                    # Get current pheromone trails
                    trails = list(self.pheromones.values())
                    
                    # Agent explores solution
                    solution_text = agent.explore_solution(problem, trails, behavior)
                    
                    # Agent evaluates its own solution
                    quality = agent.evaluate_solution(problem, solution_text)
                    
                    solution = Solution(
                        agent_id=agent.agent_id,
                        content=solution_text,
                        quality_score=quality,
                        iteration=iteration
                    )
                    
                    iteration_solutions.append(solution)
                    self.all_solutions.append(solution)
                    
                    # Deposit pheromone
                    self._deposit_pheromone(solution)
                
                # Show best solution this iteration
                best_iteration = max(iteration_solutions, key=lambda s: s.quality_score)
                print(f"[Swarm] Best this iteration (quality: {best_iteration.quality_score:.2f}): {best_iteration.content[:80]}...")
                
                # Evaporate pheromones
                self._evaporate_pheromones()
                
                # Show pheromone trail status
                print(f"[Swarm] Active pheromone trails: {len(self.pheromones)}")
            
            # Find best overall solution
            best_solution = max(self.all_solutions, key=lambda s: s.quality_score)
            
            print(f"\n[Swarm] Convergence complete!")
            print(f"[Swarm] Best solution quality: {best_solution.quality_score:.2f}")
            
            return best_solution, self.all_solutions
        
        def get_convergence_analysis(self) -> Dict[str, Any]:
            """
            Analyze how the swarm converged on solutions.
            
            Returns:
                Analysis of convergence patterns
            """
            # Quality progression over iterations
            iterations = max(s.iteration for s in self.all_solutions) + 1
            quality_by_iteration = [
                [s.quality_score for s in self.all_solutions if s.iteration == i]
                for i in range(iterations)
            ]
            
            avg_quality = [
                sum(scores) / len(scores) if scores else 0
                for scores in quality_by_iteration
            ]
            
            max_quality = [
                max(scores) if scores else 0
                for scores in quality_by_iteration
            ]
            
            # Solution diversity (number of unique solution keys)
            unique_solutions = len(set(
                self._compute_solution_key(s.content)
                for s in self.all_solutions
            ))
            
            # Top pheromone trails
            top_trails = sorted(
                self.pheromones.values(),
                key=lambda t: t.strength,
                reverse=True
            )[:3]
            
            return {
                "iterations": iterations,
                "total_solutions": len(self.all_solutions),
                "unique_solutions": unique_solutions,
                "avg_quality_progression": avg_quality,
                "max_quality_progression": max_quality,
                "final_avg_quality": avg_quality[-1],
                "final_max_quality": max_quality[-1],
                "top_pheromone_trails": [
                    {
                        "solution": trail.solution_key,
                        "strength": trail.strength,
                        "contributors": len(trail.contributors)
                    }
                    for trail in top_trails
                ]
            }
    
    
    def demonstrate_swarm_intelligence():
        """Demonstrate the Swarm Intelligence pattern with various examples."""
        
        print("=" * 80)
        print("SWARM INTELLIGENCE PATTERN DEMONSTRATION")
        print("=" * 80)
        
        # Test 1: Problem-solving task
        print("\n" + "=" * 80)
        print("TEST 1: Creative Problem Solving")
        print("=" * 80)
        
        swarm = SwarmIntelligence(num_agents=8, evaporation_rate=0.15)
        
        problem1 = "How can a small restaurant reduce food waste while maintaining quality?"
        
        best_solution, all_solutions = swarm.solve(problem1, iterations=4)
        
        print("\n" + "-" * 80)
        print("BEST SOLUTION:")
        print("-" * 80)
        print(f"Quality: {best_solution.quality_score:.2f}")
        print(f"From: {best_solution.agent_id} (iteration {best_solution.iteration})")
        print(f"\n{best_solution.content}")
        
        # Show convergence analysis
        analysis = swarm.get_convergence_analysis()
        
        print("\n" + "-" * 80)
        print("CONVERGENCE ANALYSIS:")
        print("-" * 80)
        print(f"Total solutions explored: {analysis['total_solutions']}")
        print(f"Unique solution approaches: {analysis['unique_solutions']}")
        print(f"Quality improvement: {analysis['avg_quality_progression'][0]:.2f} -> {analysis['final_avg_quality']:.2f}")
        
        print("\nQuality progression by iteration:")
        for i, (avg, max_q) in enumerate(zip(analysis['avg_quality_progression'], analysis['max_quality_progression'])):
            print(f"  Iteration {i+1}: avg={avg:.2f}, max={max_q:.2f}")
        
        print("\nTop pheromone trails (strongest convergence):")
        for i, trail in enumerate(analysis['top_pheromone_trails'], 1):
            print(f"  {i}. {trail['solution'][:60]}...")
            print(f"     Strength: {trail['strength']:.2f}, Contributors: {trail['contributors']}")
        
        # Test 2: Optimization task
        print("\n" + "=" * 80)
        print("TEST 2: Optimization Problem")
        print("=" * 80)
        
        swarm2 = SwarmIntelligence(num_agents=6, evaporation_rate=0.2)
        
        problem2 = "Design an efficient morning routine for a busy professional with 90 minutes."
        
        best_solution2, _ = swarm2.solve(problem2, iterations=3, agents_per_iteration=4)
        
        print("\n" + "-" * 80)
        print("BEST SOLUTION:")
        print("-" * 80)
        print(f"Quality: {best_solution2.quality_score:.2f}")
        print(f"\n{best_solution2.content}")
        
        # Test 3: Compare with different swarm sizes
        print("\n" + "=" * 80)
        print("TEST 3: Swarm Size Comparison")
        print("=" * 80)
        
        problem3 = "What are three innovative ways to improve employee engagement?"
        
        swarm_sizes = [4, 8, 12]
        results = []
        
        for size in swarm_sizes:
            print(f"\nTesting swarm size: {size}")
            swarm = SwarmIntelligence(num_agents=size, evaporation_rate=0.15)
            best, all_sols = swarm.solve(problem3, iterations=3, agents_per_iteration=size)
            analysis = swarm.get_convergence_analysis()
            
            results.append({
                "size": size,
                "best_quality": best.quality_score,
                "unique_solutions": analysis['unique_solutions'],
                "solution": best.content
            })
        
        print("\n" + "-" * 80)
        print("COMPARISON RESULTS:")
        print("-" * 80)
        for result in results:
            print(f"\nSwarm size {result['size']}:")
            print(f"  Best quality: {result['best_quality']:.2f}")
            print(f"  Unique approaches: {result['unique_solutions']}")
            print(f"  Solution: {result['solution'][:100]}...")
        
        # Summary
        print("\n" + "=" * 80)
        print("SUMMARY")
        print("=" * 80)
        print("""
    The Swarm Intelligence pattern demonstrates several key benefits:
    
    1. **Decentralized Problem-Solving**: No single coordinator needed
    2. **Emergent Intelligence**: Complex solutions from simple agent rules
    3. **Robustness**: Failure of individual agents doesn't stop the swarm
    4. **Parallel Exploration**: Multiple solutions explored simultaneously
    
    Key Mechanisms:
    - **Pheromone Trails**: Shared information about solution quality
    - **Evaporation**: Forgetting of old/poor solutions over time
    - **Local Decisions**: Agents decide based on nearby information
    - **Convergence**: Swarm naturally converges on good solutions
    
    Behavior Phases:
    - **Exploration**: Early iterations explore diverse approaches
    - **Balanced**: Middle iterations balance exploration and exploitation
    - **Exploitation**: Late iterations refine best solutions
    
    Use Cases:
    - Optimization problems (routing, scheduling, design)
    - Search and exploration tasks
    - Distributed problem-solving
    - Situations requiring diverse perspectives
    
    The pattern is particularly effective when:
    - Problem space is large and complex
    - Multiple good solutions may exist
    - Parallel exploration is beneficial
    - Robustness to individual failures is important
    - Emergent intelligence is desired over centralized control
    """)
    
    
    if __name__ == "__main__":
        demonstrate_swarm_intelligence()
    
        

    Agentic Design Pattern: Swarm Intelligence

    Pattern 015: Swarm Intelligence – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Swarm Intelligence pattern implementation.


    Pattern Description

    Pattern 015: Swarm Intelligence

    Description: The Swarm Intelligence pattern uses many simple, decentralized agents that collaborate through local interactions to solve complex problems. Inspired by natural phenomena like ant colonies, bee swarms, and bird flocking, this pattern achieves emergent intelligent behavior without centralized control.

    Components: – Swarm Agents: Simple agents with local decision rules – Environment/Pheromone System: Shared information space – Local Interaction Rules: How agents influence each other – Emergent Behavior: Complex patterns from simple rules

    Use Cases: – Optimization problems (path finding, resource allocation) – Search and exploration tasks – Distributed problem-solving – Parallel hypothesis exploration

    LangChain Implementation: Uses multiple simple LLM agents with local decision-making, a shared information space (pheromone trails), iterative exploration rounds, and convergence detection to achieve swarm-based optimization.

    Key Features: – Decentralized decision-making – Pheromone-based communication – Iterative convergence – Parallel exploration


    File Statistics

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

    Module Documentation

    """
    Pattern 015: Swarm Intelligence
    
    Description:
        The Swarm Intelligence pattern uses many simple, decentralized agents that collaborate
        through local interactions to solve complex problems. Inspired by natural phenomena like
        ant colonies, bee swarms, and bird flocking, this pattern achieves emergent intelligent
        behavior without centralized control.
    
    Components:
        - Swarm Agents: Simple agents with local decision rules
        - Environment/Pheromone System: Shared information space
        - Local Interaction Rules: How agents influence each other
        - Emergent Behavior: Complex patterns from simple rules
    
    Use Cases:
        - Optimization problems (path finding, resource allocation)
        - Search and exploration tasks
        - Distributed problem-solving
        - Parallel hypothesis exploration
    
    LangChain Implementation:
        Uses multiple simple LLM agents with local decision-making, a shared information
        space (pheromone trails), iterative exploration rounds, and convergence detection
        to achieve swarm-based optimization.
    
    Key Features:
        - Decentralized decision-making
        - Pheromone-based communication
        - Iterative convergence
        - Parallel exploration
    """
    

    Purpose: This module implements the Swarm Intelligence 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 enum.Enum
    import langchain_core.output_parsers.StrOutputParser
    import langchain_core.prompts.ChatPromptTemplate
    import os
    import random
    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 – 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 – 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: Solution

    Lines 50-55

    Purpose:

    Represents a solution explored by an agent.
    

    Class: PheromoneTrail

    Lines 59-63

    Purpose:

    Represents accumulated pheromone on a solution path.
    

    Class: SwarmBehavior

    Lines 66-70

    Inherits from: Enum

    Purpose:

    Types of swarm behaviors.
    

    Class: SwarmAgent

    Lines 73-188

    Purpose:

    A simple swarm agent that makes local decisions based on pheromones.
    

    Methods:

    __init__(self, agent_id, model, temperature)

    Lines 78-94

    Purpose:

    Initialize a swarm agent.
    
    Args:
        agent_id: Unique identifier for this agent
        model: LLM model to use
        temperature: Creativity/exploration level
    

    Parameters: – agent_id: Agent id – model: Language model instance – temperature: Sampling temperature (0-1)

    explore_solution(self, problem, pheromone_trails, behavior)

    Lines 96-152

    Purpose:

    Explore a solution based on local information and pheromones.
    
    Args:
        problem: Problem to solve
        pheromone_trails: Current pheromone information
        behavior: Exploration vs exploitation preference
    
    Returns:
        Proposed solution
    

    Parameters: – problem: Problem – pheromone_trails: Pheromone trails – behavior: Behavior

    Returns: str

    evaluate_solution(self, problem, solution)

    Lines 154-188

    Purpose:

    Evaluate the quality of a solution.
    
    Args:
        problem: Original problem
        solution: Proposed solution
    
    Returns:
        Quality score (0.0-1.0)
    

    Parameters: – problem: Problem – solution: Solution

    Returns: float


    Class: SwarmIntelligence

    Lines 191-422

    Purpose:

    Swarm intelligence system coordinating multiple simple agents.
    
    Uses pheromone-based communication for decentralized problem-solving.
    

    Methods:

    __init__(self, num_agents, evaporation_rate, model)

    Lines 198-229

    Purpose:

    Initialize swarm intelligence system.
    
    Args:
        num_agents: Number of agents in the swarm
        evaporation_rate: Rate at which pheromones decay (0.0-1.0)
        model: LLM model for agents
    

    Parameters: – num_agents: Num agents – evaporation_rate: Evaporation rate – model: Language model instance

    _compute_solution_key(self, solution)

    Lines 231-240

    Purpose:

    Compute a key for similar solutions.
    
    Uses first 50 characters as a simple similarity metric.
    In production, would use embeddings or semantic similarity.
    

    Parameters: – solution: Solution

    Returns: str

    _deposit_pheromone(self, solution)

    Lines 242-264

    Purpose:

    Deposit pheromone on a solution path based on quality.
    
    Args:
        solution: Solution to deposit pheromone for
    

    Parameters: – solution: Solution

    _evaporate_pheromones(self)

    Lines 266-278

    Purpose:

    Evaporate pheromones to forget old solutions.
    

    Parameters:

    _determine_behavior(self, iteration, max_iterations)

    Lines 280-294

    Purpose:

    Determine swarm behavior based on iteration.
    
    Early iterations: More exploration
    Later iterations: More exploitation
    

    Parameters: – iteration: Iteration – max_iterations: Max iterations

    Returns: SwarmBehavior

    solve(self, problem, iterations, agents_per_iteration)

    Lines 296-367

    Purpose:

    Solve a problem using swarm intelligence.
    
    Args:
        problem: Problem to solve
        iterations: Number of swarm iterations
        agents_per_iteration: Agents to activate per iteration (default: all)
    
    Returns:
        (best_solution, all_solutions)
    

    Parameters: – problem: Problem – iterations: Iterations – agents_per_iteration: Agents per iteration

    Returns: <ast.Subscript object at 0x000001A06D6A8610>

    get_convergence_analysis(self)

    Lines 369-422

    Purpose:

    Analyze how the swarm converged on solutions.
    
    Returns:
        Analysis of convergence patterns
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D16AE50>


    Functions

    demonstrate_swarm_intelligence()

    Lines 425-553

    Purpose:

    Demonstrate the Swarm Intelligence pattern with various examples.
    

    Key Concepts Summary

    Pattern Overview

    The Swarm Intelligence 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 014
    • Pattern 016
    • Other Multi-Agent Patterns (012-019)

    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 Swarm Intelligence 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: 015_swarm_intelligence.py
    Total Lines: 558
    Pattern Category: AI Agent Design Pattern #015


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Leader Follower

    
        """
    Pattern 014: Leader-Follower
    
    Description:
        The Leader-Follower pattern implements hierarchical agent coordination where one
        leader/coordinator agent manages and delegates tasks to multiple follower/specialist
        agents. The leader is responsible for task decomposition, delegation, coordination,
        and result synthesis.
    
    Components:
        - Leader Agent: Coordinates, delegates, and synthesizes results
        - Follower Agents: Specialized agents that execute specific tasks
        - Communication Protocol: Request-response between leader and followers
        - Task Assignment Strategy: How tasks are allocated to followers
    
    Use Cases:
        - Complex workflows requiring coordination
        - Tasks benefiting from specialization
        - Multi-step processes with dependencies
        - Projects requiring project management
    
    LangChain Implementation:
        Uses a coordinator LLM for task planning and delegation, specialized LLMs
        with different system prompts for specific domains, and implements task
        assignment, execution tracking, and result synthesis.
    
    Key Features:
        - Dynamic task decomposition
        - Specialist agent assignment
        - Progress tracking and monitoring
        - Hierarchical result synthesis
    """
    
    import os
    from typing import List, Dict, Any, Optional
    from dataclasses import dataclass, field
    from enum import Enum
    from dotenv import load_dotenv
    
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser
    from langchain_openai import ChatOpenAI
    
    load_dotenv()
    
    
    class TaskStatus(Enum):
        """Status of a task in the workflow."""
        PENDING = "pending"
        IN_PROGRESS = "in_progress"
        COMPLETED = "completed"
        FAILED = "failed"
    
    
    class SpecialistType(Enum):
        """Types of specialist agents available."""
        RESEARCH = "research"
        ANALYSIS = "analysis"
        WRITING = "writing"
        CODING = "coding"
        MATH = "math"
        CREATIVE = "creative"
    
    
    @dataclass
    class Task:
        """Represents a task to be completed."""
        task_id: str
        description: str
        specialist_type: SpecialistType
        status: TaskStatus = TaskStatus.PENDING
        result: Optional[str] = None
        dependencies: List[str] = field(default_factory=list)
    
    
    @dataclass
    class WorkflowResult:
        """Result from complete workflow execution."""
        final_output: str
        tasks: List[Task]
        execution_summary: str
    
    
    class SpecialistAgent:
        """
        A specialist agent with expertise in a specific domain.
        """
        
        def __init__(
            self,
            specialist_type: SpecialistType,
            model: str = "gpt-3.5-turbo",
            temperature: float = 0.7
        ):
            """
            Initialize a specialist agent.
            
            Args:
                specialist_type: Type of specialist
                model: LLM model to use
                temperature: Creativity level
            """
            self.specialist_type = specialist_type
            self.llm = ChatOpenAI(model=model, temperature=temperature)
            
            # Define specialist system prompts
            self.system_prompts = {
                SpecialistType.RESEARCH: """You are a research specialist. Your expertise is in:
    - Finding and summarizing information
    - Conducting literature reviews
    - Fact-checking and verification
    - Identifying credible sources
    Provide thorough, well-researched responses with citations when possible.""",
                
                SpecialistType.ANALYSIS: """You are an analysis specialist. Your expertise is in:
    - Data analysis and interpretation
    - Pattern recognition
    - Critical thinking and evaluation
    - Drawing insights from information
    Provide detailed analytical insights and evidence-based conclusions.""",
                
                SpecialistType.WRITING: """You are a writing specialist. Your expertise is in:
    - Clear and engaging writing
    - Content structuring and organization
    - Grammar and style
    - Adapting tone for different audiences
    Provide well-written, polished content.""",
                
                SpecialistType.CODING: """You are a coding specialist. Your expertise is in:
    - Writing clean, efficient code
    - Debugging and problem-solving
    - Best practices and design patterns
    - Multiple programming languages
    Provide production-ready code with explanations.""",
                
                SpecialistType.MATH: """You are a mathematics specialist. Your expertise is in:
    - Mathematical problem-solving
    - Statistical analysis
    - Logical reasoning
    - Step-by-step calculations
    Provide accurate solutions with clear explanations.""",
                
                SpecialistType.CREATIVE: """You are a creative specialist. Your expertise is in:
    - Brainstorming and ideation
    - Creative problem-solving
    - Innovative thinking
    - Generating unique perspectives
    Provide creative, out-of-the-box ideas and solutions."""
            }
        
        def execute_task(self, task: Task, context: Optional[str] = None) -> str:
            """
            Execute a task using specialist expertise.
            
            Args:
                task: Task to execute
                context: Optional context from previous tasks
                
            Returns:
                Task result as string
            """
            system_prompt = self.system_prompts[self.specialist_type]
            
            user_message = f"Task: {task.description}"
            if context:
                user_message += f"\n\nContext from previous tasks:\n{context}"
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", system_prompt),
                ("user", user_message)
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            result = chain.invoke({})
            
            return result
    
    
    class LeaderAgent:
        """
        Leader agent that coordinates and manages specialist agents.
        
        Responsibilities:
        - Decompose complex tasks into subtasks
        - Assign tasks to appropriate specialists
        - Monitor progress and handle dependencies
        - Synthesize results into final output
        """
        
        def __init__(self, model: str = "gpt-3.5-turbo"):
            """Initialize the leader agent."""
            self.llm = ChatOpenAI(model=model, temperature=0.3)
            
            # Initialize specialist agents
            self.specialists: Dict[SpecialistType, SpecialistAgent] = {
                specialist_type: SpecialistAgent(specialist_type)
                for specialist_type in SpecialistType
            }
        
        def decompose_task(self, overall_task: str) -> List[Task]:
            """
            Decompose an overall task into subtasks for specialists.
            
            Args:
                overall_task: The main task to decompose
                
            Returns:
                List of subtasks with assigned specialists
            """
            specialist_descriptions = "\n".join([
                f"- {s.value}: {self.specialists[s].system_prompts[s].split('Your expertise is in:')[1].split('Provide')[0].strip()}"
                for s in SpecialistType
            ])
            
            decomposition_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a project coordinator. Decompose tasks into subtasks and assign
    to appropriate specialists. Be specific and ensure tasks are actionable."""),
                ("user", """Decompose this task into 3-5 subtasks:
    
    Task: {overall_task}
    
    Available specialists:
    {specialist_descriptions}
    
    Provide your decomposition in this exact format:
    TASK 1: [specialist_type] - [clear task description]
    TASK 2: [specialist_type] - [clear task description]
    ...
    
    Use these specialist types: {specialist_types}""")
            ])
            
            chain = decomposition_prompt | self.llm | StrOutputParser()
            
            decomposition = chain.invoke({
                "overall_task": overall_task,
                "specialist_descriptions": specialist_descriptions,
                "specialist_types": ", ".join([s.value for s in SpecialistType])
            })
            
            # Parse the decomposition
            tasks = []
            for i, line in enumerate(decomposition.strip().split("\n")):
                if line.strip().startswith("TASK"):
                    try:
                        # Extract specialist type and description
                        parts = line.split(":", 1)[1].strip()
                        specialist_str, description = parts.split("-", 1)
                        specialist_str = specialist_str.strip().lower()
                        
                        # Map to specialist type
                        specialist_type = None
                        for s in SpecialistType:
                            if s.value in specialist_str:
                                specialist_type = s
                                break
                        
                        if specialist_type:
                            task = Task(
                                task_id=f"task_{i+1}",
                                description=description.strip(),
                                specialist_type=specialist_type
                            )
                            tasks.append(task)
                    except Exception as e:
                        print(f"Error parsing task line: {line} - {e}")
                        continue
            
            return tasks
        
        def assign_dependencies(self, tasks: List[Task]) -> List[Task]:
            """
            Determine task dependencies for proper execution order.
            
            Args:
                tasks: List of tasks to analyze
                
            Returns:
                Tasks with dependencies populated
            """
            # Simple heuristic: research -> analysis -> others
            for i, task in enumerate(tasks):
                if task.specialist_type == SpecialistType.ANALYSIS:
                    # Analysis depends on research
                    for prev_task in tasks[:i]:
                        if prev_task.specialist_type == SpecialistType.RESEARCH:
                            task.dependencies.append(prev_task.task_id)
                
                elif task.specialist_type in [SpecialistType.WRITING, SpecialistType.CREATIVE]:
                    # Writing/creative depends on research and analysis
                    for prev_task in tasks[:i]:
                        if prev_task.specialist_type in [SpecialistType.RESEARCH, SpecialistType.ANALYSIS]:
                            task.dependencies.append(prev_task.task_id)
            
            return tasks
        
        def execute_workflow(self, tasks: List[Task]) -> List[Task]:
            """
            Execute tasks in proper order respecting dependencies.
            
            Args:
                tasks: List of tasks to execute
                
            Returns:
                Completed tasks with results
            """
            completed_tasks: Dict[str, Task] = {}
            
            # Build context from completed tasks
            def get_context_for_task(task: Task) -> str:
                context_parts = []
                for dep_id in task.dependencies:
                    if dep_id in completed_tasks:
                        dep_task = completed_tasks[dep_id]
                        context_parts.append(
                            f"From {dep_id} ({dep_task.specialist_type.value}):\n{dep_task.result}\n"
                        )
                return "\n".join(context_parts)
            
            # Execute tasks
            for task in tasks:
                print(f"\n[Leader] Assigning {task.task_id} to {task.specialist_type.value} specialist...")
                
                task.status = TaskStatus.IN_PROGRESS
                
                try:
                    # Get specialist agent
                    specialist = self.specialists[task.specialist_type]
                    
                    # Get context from dependencies
                    context = get_context_for_task(task)
                    
                    # Execute task
                    result = specialist.execute_task(task, context)
                    
                    task.result = result
                    task.status = TaskStatus.COMPLETED
                    completed_tasks[task.task_id] = task
                    
                    print(f"[Leader] ✓ {task.task_id} completed by {task.specialist_type.value}")
                    
                except Exception as e:
                    print(f"[Leader] ✗ {task.task_id} failed: {e}")
                    task.status = TaskStatus.FAILED
                    task.result = f"Error: {str(e)}"
            
            return tasks
        
        def synthesize_results(
            self,
            overall_task: str,
            tasks: List[Task]
        ) -> str:
            """
            Synthesize results from all tasks into final output.
            
            Args:
                overall_task: Original task description
                tasks: Completed tasks with results
                
            Returns:
                Final synthesized result
            """
            # Compile all task results
            task_results = "\n\n".join([
                f"**{task.task_id}** ({task.specialist_type.value}):\n{task.result}"
                for task in tasks
                if task.status == TaskStatus.COMPLETED
            ])
            
            synthesis_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a project coordinator synthesizing results from multiple specialists.
    Create a cohesive, comprehensive final output that integrates all specialist contributions."""),
                ("user", """Original Task: {overall_task}
    
    Specialist Results:
    {task_results}
    
    Synthesize these results into a coherent final output that addresses the original task.
    Integrate insights from all specialists into a unified response.""")
            ])
            
            chain = synthesis_prompt | self.llm | StrOutputParser()
            
            final_output = chain.invoke({
                "overall_task": overall_task,
                "task_results": task_results
            })
            
            return final_output
        
        def coordinate(self, overall_task: str) -> WorkflowResult:
            """
            Coordinate the complete workflow from task to final output.
            
            Args:
                overall_task: The main task to complete
                
            Returns:
                WorkflowResult with final output and execution details
            """
            print(f"\n[Leader] Coordinating task: {overall_task}\n")
            
            # Step 1: Decompose task
            print("[Leader] Step 1: Decomposing task...")
            tasks = self.decompose_task(overall_task)
            print(f"[Leader] Created {len(tasks)} subtasks")
            
            for task in tasks:
                print(f"  - {task.task_id}: {task.description[:60]}... [{task.specialist_type.value}]")
            
            # Step 2: Assign dependencies
            print("\n[Leader] Step 2: Determining task dependencies...")
            tasks = self.assign_dependencies(tasks)
            
            # Step 3: Execute workflow
            print("\n[Leader] Step 3: Executing workflow...")
            tasks = self.execute_workflow(tasks)
            
            # Step 4: Synthesize results
            print("\n[Leader] Step 4: Synthesizing results...")
            final_output = self.synthesize_results(overall_task, tasks)
            
            # Create execution summary
            completed = sum(1 for t in tasks if t.status == TaskStatus.COMPLETED)
            execution_summary = f"Completed {completed}/{len(tasks)} tasks using {len(set(t.specialist_type for t in tasks))} specialist types"
            
            print(f"\n[Leader] Workflow complete! {execution_summary}")
            
            return WorkflowResult(
                final_output=final_output,
                tasks=tasks,
                execution_summary=execution_summary
            )
    
    
    def demonstrate_leader_follower():
        """Demonstrate the Leader-Follower pattern with various examples."""
        
        print("=" * 80)
        print("LEADER-FOLLOWER PATTERN DEMONSTRATION")
        print("=" * 80)
        
        leader = LeaderAgent()
        
        # Test 1: Research and analysis task
        print("\n" + "=" * 80)
        print("TEST 1: Complex Research and Analysis Task")
        print("=" * 80)
        
        task1 = "Analyze the impact of artificial intelligence on the job market and write a summary report"
        
        result = leader.coordinate(task1)
        
        print("\n" + "-" * 80)
        print("FINAL OUTPUT:")
        print("-" * 80)
        print(result.final_output)
        
        print("\n" + "-" * 80)
        print("TASK BREAKDOWN:")
        print("-" * 80)
        for task in result.tasks:
            status_symbol = "✓" if task.status == TaskStatus.COMPLETED else "✗"
            print(f"{status_symbol} {task.task_id} [{task.specialist_type.value}]: {task.description[:60]}...")
        
        # Test 2: Creative and technical task
        print("\n" + "=" * 80)
        print("TEST 2: Creative + Technical Task")
        print("=" * 80)
        
        task2 = "Design a simple Python game concept and create a basic implementation plan"
        
        result = leader.coordinate(task2)
        
        print("\n" + "-" * 80)
        print("FINAL OUTPUT:")
        print("-" * 80)
        print(result.final_output[:500] + "...")
        
        print("\n" + "-" * 80)
        print("SPECIALIST UTILIZATION:")
        print("-" * 80)
        specialist_usage = {}
        for task in result.tasks:
            specialist_type = task.specialist_type.value
            specialist_usage[specialist_type] = specialist_usage.get(specialist_type, 0) + 1
        
        for specialist, count in specialist_usage.items():
            print(f"  {specialist}: {count} task(s)")
        
        # Test 3: Multi-domain problem
        print("\n" + "=" * 80)
        print("TEST 3: Multi-Domain Problem Solving")
        print("=" * 80)
        
        task3 = "Calculate the ROI of implementing renewable energy in a small business and provide recommendations"
        
        result = leader.coordinate(task3)
        
        print("\n" + "-" * 80)
        print("FINAL OUTPUT:")
        print("-" * 80)
        print(result.final_output)
        
        # Summary
        print("\n" + "=" * 80)
        print("SUMMARY")
        print("=" * 80)
        print("""
    The Leader-Follower pattern demonstrates several key benefits:
    
    1. **Specialization**: Each agent focuses on what it does best
    2. **Coordination**: Leader manages workflow and dependencies
    3. **Scalability**: Easy to add new specialist types
    4. **Quality**: Specialist expertise improves output quality
    
    Architecture:
    - **Leader Agent**: Plans, delegates, coordinates, synthesizes
    - **Follower Agents**: Execute specialized tasks with expertise
    - **Task Management**: Handles dependencies and execution order
    - **Result Synthesis**: Combines specialist outputs into coherent result
    
    Use Cases:
    - Complex workflows requiring multiple skills
    - Projects benefiting from domain specialization
    - Multi-step processes with dependencies
    - Tasks requiring project management
    
    The pattern is particularly effective when:
    - Tasks can be decomposed into specialized subtasks
    - Different subtasks require different expertise
    - Coordination overhead is justified by quality improvement
    - Clear task dependencies exist
    """)
    
    
    if __name__ == "__main__":
        demonstrate_leader_follower()
    
        

    Agentic Design Pattern: Leader Follower

    Pattern 014: Leader Follower – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Leader Follower pattern implementation.


    Pattern Description

    Pattern 014: Leader-Follower

    Description: The Leader-Follower pattern implements hierarchical agent coordination where one leader/coordinator agent manages and delegates tasks to multiple follower/specialist agents. The leader is responsible for task decomposition, delegation, coordination, and result synthesis.

    Components: – Leader Agent: Coordinates, delegates, and synthesizes results – Follower Agents: Specialized agents that execute specific tasks – Communication Protocol: Request-response between leader and followers – Task Assignment Strategy: How tasks are allocated to followers

    Use Cases: – Complex workflows requiring coordination – Tasks benefiting from specialization – Multi-step processes with dependencies – Projects requiring project management

    LangChain Implementation: Uses a coordinator LLM for task planning and delegation, specialized LLMs with different system prompts for specific domains, and implements task assignment, execution tracking, and result synthesis.

    Key Features: – Dynamic task decomposition – Specialist agent assignment – Progress tracking and monitoring – Hierarchical result synthesis


    File Statistics

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

    Module Documentation

    """
    Pattern 014: Leader-Follower
    
    Description:
        The Leader-Follower pattern implements hierarchical agent coordination where one
        leader/coordinator agent manages and delegates tasks to multiple follower/specialist
        agents. The leader is responsible for task decomposition, delegation, coordination,
        and result synthesis.
    
    Components:
        - Leader Agent: Coordinates, delegates, and synthesizes results
        - Follower Agents: Specialized agents that execute specific tasks
        - Communication Protocol: Request-response between leader and followers
        - Task Assignment Strategy: How tasks are allocated to followers
    
    Use Cases:
        - Complex workflows requiring coordination
        - Tasks benefiting from specialization
        - Multi-step processes with dependencies
        - Projects requiring project management
    
    LangChain Implementation:
        Uses a coordinator LLM for task planning and delegation, specialized LLMs
        with different system prompts for specific domains, and implements task
        assignment, execution tracking, and result synthesis.
    
    Key Features:
        - Dynamic task decomposition
        - Specialist agent assignment
        - Progress tracking and monitoring
        - Hierarchical result synthesis
    """
    

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


    Import Statements

    Standard Library Imports

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

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

    LangChain Imports

    from langchain_openai.ChatOpenAI
    

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

    Other Imports

    from dotenv.load_dotenv
    

    Purpose: Additional dependencies for specialized functionality.


    Classes

    Class: TaskStatus

    Lines 47-52

    Inherits from: Enum

    Purpose:

    Status of a task in the workflow.
    

    Class: SpecialistType

    Lines 55-62

    Inherits from: Enum

    Purpose:

    Types of specialist agents available.
    

    Class: Task

    Lines 66-73

    Purpose:

    Represents a task to be completed.
    

    Class: WorkflowResult

    Lines 77-81

    Purpose:

    Result from complete workflow execution.
    

    Class: SpecialistAgent

    Lines 84-176

    Purpose:

    A specialist agent with expertise in a specific domain.
    

    Methods:

    __init__(self, specialist_type, model, temperature)

    Lines 89-149

    Purpose:

    Initialize a specialist agent.
    
    Args:
        specialist_type: Type of specialist
        model: LLM model to use
        temperature: Creativity level
    

    Parameters: – specialist_type: Specialist type – model: Language model instance – temperature: Sampling temperature (0-1)

    execute_task(self, task, context)

    Lines 151-176

    Purpose:

    Execute a task using specialist expertise.
    
    Args:
        task: Task to execute
        context: Optional context from previous tasks
    
    Returns:
        Task result as string
    

    Parameters: – task: Task – context: Context

    Returns: str


    Class: LeaderAgent

    Lines 179-434

    Purpose:

    Leader agent that coordinates and manages specialist agents.
    
    Responsibilities:
    - Decompose complex tasks into subtasks
    - Assign tasks to appropriate specialists
    - Monitor progress and handle dependencies
    - Synthesize results into final output
    

    Methods:

    __init__(self, model)

    Lines 190-198

    Purpose:

    Initialize the leader agent.
    

    Parameters: – model: Language model instance

    decompose_task(self, overall_task)

    Lines 200-269

    Purpose:

    Decompose an overall task into subtasks for specialists.
    
    Args:
        overall_task: The main task to decompose
    
    Returns:
        List of subtasks with assigned specialists
    

    Parameters: – overall_task: Overall task

    Returns: <ast.Subscript object at 0x000001A06D633A50>

    assign_dependencies(self, tasks)

    Lines 271-295

    Purpose:

    Determine task dependencies for proper execution order.
    
    Args:
        tasks: List of tasks to analyze
    
    Returns:
        Tasks with dependencies populated
    

    Parameters: – tasks: Tasks

    Returns: <ast.Subscript object at 0x000001A06D633810>

    execute_workflow(self, tasks)

    Lines 297-347

    Purpose:

    Execute tasks in proper order respecting dependencies.
    
    Args:
        tasks: List of tasks to execute
    
    Returns:
        Completed tasks with results
    

    Parameters: – tasks: Tasks

    Returns: <ast.Subscript object at 0x000001A06D609D90>

    synthesize_results(self, overall_task, tasks)

    Lines 349-390

    Purpose:

    Synthesize results from all tasks into final output.
    
    Args:
        overall_task: Original task description
        tasks: Completed tasks with results
    
    Returns:
        Final synthesized result
    

    Parameters: – overall_task: Overall task – tasks: Tasks

    Returns: str

    coordinate(self, overall_task)

    Lines 392-434

    Purpose:

    Coordinate the complete workflow from task to final output.
    
    Args:
        overall_task: The main task to complete
    
    Returns:
        WorkflowResult with final output and execution details
    

    Parameters: – overall_task: Overall task

    Returns: WorkflowResult


    Functions

    demonstrate_leader_follower()

    Lines 437-535

    Purpose:

    Demonstrate the Leader-Follower pattern with various examples.
    

    Key Concepts Summary

    Pattern Overview

    The Leader Follower 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 013
    • Pattern 015
    • Other Multi-Agent Patterns (012-019)

    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 Leader Follower 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: 014_leader_follower.py
    Total Lines: 540
    Pattern Category: AI Agent Design Pattern #014


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Ensemble Committee

    
        """
    Pattern 013: Ensemble/Committee
    
    Description:
        The Ensemble/Committee pattern uses multiple agents working independently on the same
        task, then aggregates their outputs to reach a final decision. This pattern leverages
        diversity to improve robustness, reduce errors, and provide more reliable results.
    
    Components:
        - Multiple independent agents (diverse models, prompts, or strategies)
        - Aggregation mechanism (voting, averaging, weighted combination)
        - Result synthesis and confidence scoring
    
    Use Cases:
        - Classification tasks requiring high accuracy
        - Prediction problems where uncertainty quantification is important
        - Reducing bias through diverse perspectives
        - Quality assurance through redundancy
    
    LangChain Implementation:
        Uses multiple ChatOpenAI instances with different configurations, implements
        various aggregation strategies (majority voting, weighted voting, consensus),
        and provides confidence metrics based on agreement levels.
    
    Key Features:
        - Diverse agent configurations (temperature, models, prompts)
        - Multiple aggregation strategies
        - Confidence scoring based on agreement
        - Detailed breakdown of individual votes
    """
    
    import os
    from typing import List, Dict, Any, Optional, Tuple
    from dataclasses import dataclass
    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 AggregationStrategy(Enum):
        """Aggregation strategies for combining agent outputs."""
        MAJORITY_VOTE = "majority_vote"
        WEIGHTED_VOTE = "weighted_vote"
        CONSENSUS = "consensus"
        AVERAGING = "averaging"
    
    
    @dataclass
    class AgentVote:
        """Represents a single agent's vote/output."""
        agent_id: str
        output: str
        confidence: float
        reasoning: Optional[str] = None
    
    
    @dataclass
    class EnsembleResult:
        """Result from ensemble aggregation."""
        final_output: str
        confidence: float
        individual_votes: List[AgentVote]
        agreement_score: float
        strategy_used: str
    
    
    class EnsembleAgent:
        """
        Ensemble agent that combines outputs from multiple independent agents.
        
        This implementation creates diverse agents through different configurations
        (temperature, system prompts) and provides multiple aggregation strategies.
        """
        
        def __init__(
            self,
            num_agents: int = 3,
            base_model: str = "gpt-3.5-turbo",
            temperatures: Optional[List[float]] = None
        ):
            """
            Initialize ensemble with multiple agents.
            
            Args:
                num_agents: Number of agents in ensemble
                base_model: Base model for all agents
                temperatures: List of temperatures for diversity (default: [0.3, 0.7, 1.0])
            """
            self.num_agents = num_agents
            self.base_model = base_model
            
            # Create diverse temperature settings if not provided
            if temperatures is None:
                temperatures = [0.3, 0.5, 0.7, 0.9, 1.0][:num_agents]
            
            # Create diverse agents with different temperatures
            self.agents = []
            for i, temp in enumerate(temperatures[:num_agents]):
                agent = ChatOpenAI(
                    model=base_model,
                    temperature=temp,
                    model_kwargs={"seed": i * 42}  # Different seeds for diversity
                )
                self.agents.append(agent)
            
            # Create a meta-agent for confidence estimation
            self.meta_agent = ChatOpenAI(model=base_model, temperature=0.2)
        
        def _estimate_confidence(
            self,
            question: str,
            answer: str,
            reasoning: str
        ) -> float:
            """Estimate confidence score for an individual answer."""
            confidence_prompt = ChatPromptTemplate.from_messages([
                ("system", "You are an expert at evaluating answer quality and confidence."),
                ("user", """Given this question and answer, estimate the confidence level (0.0-1.0).
    Consider clarity, completeness, and certainty in the reasoning.
    
    Question: {question}
    
    Answer: {answer}
    
    Reasoning: {reasoning}
    
    Provide only a number between 0.0 and 1.0.""")
            ])
            
            chain = confidence_prompt | self.meta_agent | StrOutputParser()
            
            try:
                confidence_str = chain.invoke({
                    "question": question,
                    "answer": answer,
                    "reasoning": reasoning
                })
                confidence = float(confidence_str.strip())
                return max(0.0, min(1.0, confidence))
            except:
                return 0.5  # Default moderate confidence
        
        def query_agents(
            self,
            question: str,
            system_prompt: Optional[str] = None
        ) -> List[AgentVote]:
            """
            Query all agents independently and collect their votes.
            
            Args:
                question: Question to ask all agents
                system_prompt: Optional system prompt (default: general assistant)
                
            Returns:
                List of agent votes with outputs and confidence scores
            """
            if system_prompt is None:
                system_prompt = "You are a helpful AI assistant. Provide clear, accurate answers."
            
            prompt_template = ChatPromptTemplate.from_messages([
                ("system", system_prompt + "\n\nProvide your answer followed by your reasoning."),
                ("user", "{question}")
            ])
            
            votes = []
            
            for i, agent in enumerate(self.agents):
                chain = prompt_template | agent | StrOutputParser()
                
                try:
                    response = chain.invoke({"question": question})
                    
                    # Split response into answer and reasoning
                    parts = response.split("\n", 1)
                    answer = parts[0].strip()
                    reasoning = parts[1].strip() if len(parts) > 1 else ""
                    
                    # Estimate confidence
                    confidence = self._estimate_confidence(question, answer, reasoning)
                    
                    vote = AgentVote(
                        agent_id=f"agent_{i}",
                        output=answer,
                        confidence=confidence,
                        reasoning=reasoning
                    )
                    votes.append(vote)
                    
                except Exception as e:
                    print(f"Agent {i} error: {e}")
                    # Add a low-confidence fallback vote
                    votes.append(AgentVote(
                        agent_id=f"agent_{i}",
                        output="Error generating response",
                        confidence=0.0,
                        reasoning=f"Error: {str(e)}"
                    ))
            
            return votes
        
        def aggregate_majority_vote(
            self,
            votes: List[AgentVote]
        ) -> Tuple[str, float]:
            """
            Aggregate using simple majority voting.
            
            Returns:
                (winning answer, agreement score)
            """
            # Count votes for each unique answer
            answer_counts = Counter(vote.output for vote in votes)
            most_common = answer_counts.most_common(1)[0]
            winning_answer = most_common[0]
            vote_count = most_common[1]
            
            agreement_score = vote_count / len(votes)
            
            return winning_answer, agreement_score
        
        def aggregate_weighted_vote(
            self,
            votes: List[AgentVote]
        ) -> Tuple[str, float]:
            """
            Aggregate using confidence-weighted voting.
            
            Returns:
                (winning answer, weighted agreement score)
            """
            # Weight each answer by confidence
            answer_weights: Dict[str, float] = {}
            total_confidence = sum(vote.confidence for vote in votes)
            
            for vote in votes:
                if vote.output in answer_weights:
                    answer_weights[vote.output] += vote.confidence
                else:
                    answer_weights[vote.output] = vote.confidence
            
            # Find answer with highest weighted vote
            winning_answer = max(answer_weights.items(), key=lambda x: x[1])[0]
            winning_weight = answer_weights[winning_answer]
            
            # Calculate weighted agreement score
            agreement_score = winning_weight / total_confidence if total_confidence > 0 else 0.0
            
            return winning_answer, agreement_score
        
        def aggregate_consensus(
            self,
            votes: List[AgentVote],
            question: str
        ) -> Tuple[str, float]:
            """
            Generate consensus answer by synthesizing all votes.
            
            Returns:
                (consensus answer, confidence score)
            """
            # Compile all answers and reasoning
            vote_summary = "\n\n".join([
                f"Agent {vote.agent_id} (confidence: {vote.confidence:.2f}):\n"
                f"Answer: {vote.output}\n"
                f"Reasoning: {vote.reasoning}"
                for vote in votes
            ])
            
            consensus_prompt = ChatPromptTemplate.from_messages([
                ("system", "You are an expert at synthesizing multiple perspectives into a consensus."),
                ("user", """Given the following question and multiple agent responses, create a 
    consensus answer that incorporates the best elements from all responses.
    
    Question: {question}
    
    Agent Responses:
    {vote_summary}
    
    Provide a consensus answer that:
    1. Incorporates accurate information from all agents
    2. Resolves any contradictions
    3. Provides a clear, unified response
    
    Consensus Answer:""")
            ])
            
            chain = consensus_prompt | self.meta_agent | StrOutputParser()
            
            consensus_answer = chain.invoke({
                "question": question,
                "vote_summary": vote_summary
            })
            
            # Calculate confidence based on average agent confidence
            avg_confidence = sum(vote.confidence for vote in votes) / len(votes)
            
            return consensus_answer.strip(), avg_confidence
        
        def solve(
            self,
            question: str,
            strategy: AggregationStrategy = AggregationStrategy.WEIGHTED_VOTE,
            system_prompt: Optional[str] = None
        ) -> EnsembleResult:
            """
            Solve a problem using the ensemble of agents.
            
            Args:
                question: Question to solve
                strategy: Aggregation strategy to use
                system_prompt: Optional custom system prompt
                
            Returns:
                EnsembleResult with final answer and metadata
            """
            # Get votes from all agents
            votes = self.query_agents(question, system_prompt)
            
            # Aggregate based on strategy
            if strategy == AggregationStrategy.MAJORITY_VOTE:
                final_output, agreement_score = self.aggregate_majority_vote(votes)
                
            elif strategy == AggregationStrategy.WEIGHTED_VOTE:
                final_output, agreement_score = self.aggregate_weighted_vote(votes)
                
            elif strategy == AggregationStrategy.CONSENSUS:
                final_output, agreement_score = self.aggregate_consensus(votes, question)
                
            else:
                raise ValueError(f"Unsupported strategy: {strategy}")
            
            # Calculate overall confidence
            confidence = agreement_score
            
            return EnsembleResult(
                final_output=final_output,
                confidence=confidence,
                individual_votes=votes,
                agreement_score=agreement_score,
                strategy_used=strategy.value
            )
    
    
    def demonstrate_ensemble():
        """Demonstrate the Ensemble/Committee pattern with various examples."""
        
        print("=" * 80)
        print("ENSEMBLE/COMMITTEE PATTERN DEMONSTRATION")
        print("=" * 80)
        
        # Create ensemble with 5 agents
        ensemble = EnsembleAgent(num_agents=5)
        
        # Test 1: Classification task
        print("\n" + "=" * 80)
        print("TEST 1: Sentiment Classification")
        print("=" * 80)
        
        sentiment_question = """Classify the sentiment of this review as Positive, Negative, or Neutral:
    
    "The product works okay, but the customer service was terrible. Mixed feelings overall."
    
    Your answer should be exactly one word: Positive, Negative, or Neutral."""
        
        result = ensemble.solve(
            sentiment_question,
            strategy=AggregationStrategy.WEIGHTED_VOTE
        )
        
        print(f"\nQuestion: Sentiment classification of mixed review")
        print(f"\nFinal Answer: {result.final_output}")
        print(f"Confidence: {result.confidence:.2f}")
        print(f"Agreement Score: {result.agreement_score:.2f}")
        print(f"Strategy: {result.strategy_used}")
        
        print("\nIndividual Agent Votes:")
        for vote in result.individual_votes:
            print(f"  {vote.agent_id}: {vote.output} (confidence: {vote.confidence:.2f})")
        
        # Test 2: Factual question with consensus
        print("\n" + "=" * 80)
        print("TEST 2: Factual Question with Consensus")
        print("=" * 80)
        
        factual_question = "What are the main causes of climate change?"
        
        result = ensemble.solve(
            factual_question,
            strategy=AggregationStrategy.CONSENSUS
        )
        
        print(f"\nQuestion: {factual_question}")
        print(f"\nConsensus Answer:\n{result.final_output}")
        print(f"\nConfidence: {result.confidence:.2f}")
        print(f"Agreement Score: {result.agreement_score:.2f}")
        
        # Test 3: Multiple choice with majority vote
        print("\n" + "=" * 80)
        print("TEST 3: Multiple Choice with Majority Vote")
        print("=" * 80)
        
        mc_question = """Which of the following is the capital of Australia?
    A) Sydney
    B) Melbourne
    C) Canberra
    D) Brisbane
    
    Answer with only the letter (A, B, C, or D)."""
        
        result = ensemble.solve(
            mc_question,
            strategy=AggregationStrategy.MAJORITY_VOTE
        )
        
        print(f"\nQuestion: Capital of Australia (multiple choice)")
        print(f"\nFinal Answer: {result.final_output}")
        print(f"Agreement Score: {result.agreement_score:.2f}")
        
        print("\nIndividual Agent Votes:")
        for vote in result.individual_votes:
            print(f"  {vote.agent_id}: {vote.output}")
        
        # Test 4: Comparison of strategies
        print("\n" + "=" * 80)
        print("TEST 4: Strategy Comparison")
        print("=" * 80)
        
        comparison_question = "Is artificial intelligence dangerous? Answer Yes or No, then explain."
        
        print(f"\nQuestion: {comparison_question}\n")
        
        for strategy in [AggregationStrategy.MAJORITY_VOTE, 
                         AggregationStrategy.WEIGHTED_VOTE,
                         AggregationStrategy.CONSENSUS]:
            
            result = ensemble.solve(comparison_question, strategy=strategy)
            
            print(f"\nStrategy: {strategy.value}")
            print(f"Answer: {result.final_output[:150]}...")
            print(f"Confidence: {result.confidence:.2f}")
            print(f"Agreement: {result.agreement_score:.2f}")
        
        # Summary
        print("\n" + "=" * 80)
        print("SUMMARY")
        print("=" * 80)
        print("""
    The Ensemble/Committee pattern demonstrates several key benefits:
    
    1. **Robustness**: Multiple agents reduce the impact of individual errors
    2. **Diversity**: Different temperatures create varied perspectives
    3. **Confidence Scoring**: Agreement levels indicate reliability
    4. **Flexibility**: Multiple aggregation strategies for different use cases
    
    Aggregation Strategies:
    - **Majority Vote**: Simple, works well for discrete choices
    - **Weighted Vote**: Considers confidence, better for uncertain domains
    - **Consensus**: Synthesizes responses, best for complex questions
    
    Use Cases:
    - High-stakes classification tasks
    - Fact-checking and validation
    - Reducing model bias
    - Quality assurance through redundancy
    
    The pattern is particularly effective when:
    - Individual agent accuracy is moderate but independent
    - Diversity in approaches improves results
    - Confidence/uncertainty quantification is important
    - Explainability of decisions is needed
    """)
    
    
    if __name__ == "__main__":
        demonstrate_ensemble()
    
        

    Agentic Design Pattern: Ensemble Committee

    Pattern 013: Ensemble Committee – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Ensemble Committee pattern implementation.


    Pattern Description

    Pattern 013: Ensemble/Committee

    Description: The Ensemble/Committee pattern uses multiple agents working independently on the same task, then aggregates their outputs to reach a final decision. This pattern leverages diversity to improve robustness, reduce errors, and provide more reliable results.

    Components: – Multiple independent agents (diverse models, prompts, or strategies) – Aggregation mechanism (voting, averaging, weighted combination) – Result synthesis and confidence scoring

    Use Cases: – Classification tasks requiring high accuracy – Prediction problems where uncertainty quantification is important – Reducing bias through diverse perspectives – Quality assurance through redundancy

    LangChain Implementation: Uses multiple ChatOpenAI instances with different configurations, implements various aggregation strategies (majority voting, weighted voting, consensus), and provides confidence metrics based on agreement levels.

    Key Features: – Diverse agent configurations (temperature, models, prompts) – Multiple aggregation strategies – Confidence scoring based on agreement – Detailed breakdown of individual votes


    File Statistics

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

    Module Documentation

    """
    Pattern 013: Ensemble/Committee
    
    Description:
        The Ensemble/Committee pattern uses multiple agents working independently on the same
        task, then aggregates their outputs to reach a final decision. This pattern leverages
        diversity to improve robustness, reduce errors, and provide more reliable results.
    
    Components:
        - Multiple independent agents (diverse models, prompts, or strategies)
        - Aggregation mechanism (voting, averaging, weighted combination)
        - Result synthesis and confidence scoring
    
    Use Cases:
        - Classification tasks requiring high accuracy
        - Prediction problems where uncertainty quantification is important
        - Reducing bias through diverse perspectives
        - Quality assurance through redundancy
    
    LangChain Implementation:
        Uses multiple ChatOpenAI instances with different configurations, implements
        various aggregation strategies (majority voting, weighted voting, consensus),
        and provides confidence metrics based on agreement levels.
    
    Key Features:
        - Diverse agent configurations (temperature, models, prompts)
        - Multiple aggregation strategies
        - Confidence scoring based on agreement
        - Detailed breakdown of individual votes
    """
    

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


    Import Statements

    Standard Library Imports

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

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

    LangChain Imports

    from langchain_openai.ChatOpenAI
    

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

    Other Imports

    from dotenv.load_dotenv
    

    Purpose: Additional dependencies for specialized functionality.


    Classes

    Class: AggregationStrategy

    Lines 46-51

    Inherits from: Enum

    Purpose:

    Aggregation strategies for combining agent outputs.
    

    Class: AgentVote

    Lines 55-60

    Purpose:

    Represents a single agent's vote/output.
    

    Class: EnsembleResult

    Lines 64-70

    Purpose:

    Result from ensemble aggregation.
    

    Class: EnsembleAgent

    Lines 73-348

    Purpose:

    Ensemble agent that combines outputs from multiple independent agents.
    
    This implementation creates diverse agents through different configurations
    (temperature, system prompts) and provides multiple aggregation strategies.
    

    Methods:

    __init__(self, num_agents, base_model, temperatures)

    Lines 81-113

    Purpose:

    Initialize ensemble with multiple agents.
    
    Args:
        num_agents: Number of agents in ensemble
        base_model: Base model for all agents
        temperatures: List of temperatures for diversity (default: [0.3, 0.7, 1.0])
    

    Parameters: – num_agents: Num agents – base_model: Language model instance – temperatures: Sampling temperature (0-1)

    _estimate_confidence(self, question, answer, reasoning)

    Lines 115-147

    Purpose:

    Estimate confidence score for an individual answer.
    

    Parameters: – question: Question to answer – answer: Answer – reasoning: Reasoning

    Returns: float

    query_agents(self, question, system_prompt)

    Lines 149-206

    Purpose:

    Query all agents independently and collect their votes.
    
    Args:
        question: Question to ask all agents
        system_prompt: Optional system prompt (default: general assistant)
    
    Returns:
        List of agent votes with outputs and confidence scores
    

    Parameters: – question: Question to answer – system_prompt: Prompt template or text

    Returns: <ast.Subscript object at 0x000001A06D17AC50>

    aggregate_majority_vote(self, votes)

    Lines 208-226

    Purpose:

    Aggregate using simple majority voting.
    
    Returns:
        (winning answer, agreement score)
    

    Parameters: – votes: Votes

    Returns: <ast.Subscript object at 0x000001A06D169190>

    aggregate_weighted_vote(self, votes)

    Lines 228-255

    Purpose:

    Aggregate using confidence-weighted voting.
    
    Returns:
        (winning answer, weighted agreement score)
    

    Parameters: – votes: Votes

    Returns: <ast.Subscript object at 0x000001A06D6F7A90>

    aggregate_consensus(self, votes, question)

    Lines 257-304

    Purpose:

    Generate consensus answer by synthesizing all votes.
    
    Returns:
        (consensus answer, confidence score)
    

    Parameters: – votes: Votes – question: Question to answer

    Returns: <ast.Subscript object at 0x000001A06D7038D0>

    solve(self, question, strategy, system_prompt)

    Lines 306-348

    Purpose:

    Solve a problem using the ensemble of agents.
    
    Args:
        question: Question to solve
        strategy: Aggregation strategy to use
        system_prompt: Optional custom system prompt
    
    Returns:
        EnsembleResult with final answer and metadata
    

    Parameters: – question: Question to answer – strategy: Strategy – system_prompt: Prompt template or text

    Returns: EnsembleResult


    Functions

    demonstrate_ensemble()

    Lines 351-478

    Purpose:

    Demonstrate the Ensemble/Committee pattern with various examples.
    

    Key Concepts Summary

    Pattern Overview

    The Ensemble Committee 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 012
    • Pattern 014
    • Other Multi-Agent Patterns (012-019)

    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 Ensemble Committee 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: 013_ensemble_committee.py
    Total Lines: 483
    Pattern Category: AI Agent Design Pattern #013


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Debate Discussion

    
        """
    Pattern 012: Debate/Discussion (Multi-Agent)
    
    Description:
        Multiple agents engage in debate or discussion to reach better conclusions.
        Each agent represents different perspectives, argues for their position,
        and the system synthesizes the best answer from the debate. This reduces
        bias and improves reasoning quality through adversarial collaboration.
    
    Key Concepts:
        - Multiple Perspectives: Each agent takes different viewpoint
        - Argument Generation: Agents present reasoned arguments
        - Counter-Arguments: Agents respond to each other's points
        - Synthesis: Final answer combines best insights from debate
        - Moderation: Optional moderator guides discussion
    
    Benefits:
        - Reduced individual bias
        - More thorough exploration of problem space
        - Error correction through peer review
        - Higher quality conclusions
    
    Use Cases:
        - Complex decision-making
        - Ethical dilemmas
        - Strategy formulation
        - Content review and validation
    
    LangChain Implementation:
        Multiple LLM instances with different personas engaging in structured debate.
    """
    
    import os
    from typing import List, Dict, Any, Optional
    from dataclasses import dataclass, field
    from enum import Enum
    from dotenv import load_dotenv
    
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser
    from langchain_openai import ChatOpenAI
    
    # Load environment variables
    load_dotenv()
    
    
    class AgentRole(Enum):
        """Roles agents can take in debate."""
        PROPONENT = "proponent"
        OPPONENT = "opponent"
        SKEPTIC = "skeptic"
        PRAGMATIST = "pragmatist"
        MODERATOR = "moderator"
    
    
    @dataclass
    class DebateMessage:
        """Represents a message in the debate."""
        round: int
        agent_id: str
        role: AgentRole
        content: str
        responding_to: Optional[str] = None
    
    
    @dataclass
    class DebateTranscript:
        """Complete transcript of a debate."""
        question: str
        messages: List[DebateMessage] = field(default_factory=list)
        final_synthesis: str = ""
        
        def add_message(self, message: DebateMessage):
            """Add a message to the transcript."""
            self.messages.append(message)
        
        def get_round_messages(self, round_num: int) -> List[DebateMessage]:
            """Get all messages from a specific round."""
            return [msg for msg in self.messages if msg.round == round_num]
    
    
    class DebateAgent:
        """Individual agent participating in debate."""
        
        def __init__(self, agent_id: str, role: AgentRole, 
                     model_name: str = "gpt-3.5-turbo"):
            """
            Initialize a debate agent.
            
            Args:
                agent_id: Unique identifier for the agent
                role: Role the agent plays in debate
                model_name: Name of the OpenAI model
            """
            self.agent_id = agent_id
            self.role = role
            self.llm = ChatOpenAI(
                model=model_name,
                temperature=0.7,  # Some creativity for diverse perspectives
                api_key=os.getenv("OPENAI_API_KEY")
            )
            
            # Role-specific system prompts
            self.system_prompts = {
                AgentRole.PROPONENT: "You are advocating FOR the position. Present strong arguments supporting it.",
                AgentRole.OPPONENT: "You are arguing AGAINST the position. Present strong counterarguments.",
                AgentRole.SKEPTIC: "You are a critical skeptic. Question assumptions and identify weaknesses.",
                AgentRole.PRAGMATIST: "You focus on practical implications and real-world feasibility."
            }
        
        def generate_opening_statement(self, question: str) -> str:
            """Generate opening statement for the debate."""
            prompt = ChatPromptTemplate.from_messages([
                ("system", f"""{self.system_prompts.get(self.role, 'You are a thoughtful debater.')}
                
    This is your opening statement. Present your main position clearly and concisely."""),
                ("human", "Question: {question}\n\nYour opening statement:")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            return chain.invoke({"question": question})
        
        def respond_to_arguments(self, question: str, previous_messages: List[DebateMessage]) -> str:
            """Generate response to previous arguments."""
            # Build context from previous messages
            context_parts = []
            for msg in previous_messages:
                context_parts.append(f"{msg.agent_id} ({msg.role.value}): {msg.content}")
            
            context = "\n\n".join(context_parts)
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", f"""{self.system_prompts.get(self.role, 'You are a thoughtful debater.')}
    
    Respond to the previous arguments. Address key points, present counter-arguments where appropriate,
    and strengthen your position."""),
                ("human", """Question: {question}
    
    Previous Arguments:
    {context}
    
    Your response:""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            return chain.invoke({
                "question": question,
                "context": context
            })
    
    
    class DebateSystem:
        """System for managing multi-agent debate."""
        
        def __init__(self, model_name: str = "gpt-3.5-turbo", num_rounds: int = 2):
            """
            Initialize the debate system.
            
            Args:
                model_name: Name of the OpenAI model
                num_rounds: Number of debate rounds
            """
            self.model_name = model_name
            self.num_rounds = num_rounds
            self.moderator_llm = ChatOpenAI(
                model=model_name,
                temperature=0.3,
                api_key=os.getenv("OPENAI_API_KEY")
            )
        
        def conduct_debate(self, question: str, 
                          agent_roles: Optional[List[AgentRole]] = None,
                          verbose: bool = True) -> DebateTranscript:
            """
            Conduct a structured debate on the question.
            
            Args:
                question: Question to debate
                agent_roles: Roles for agents (default: proponent and opponent)
                verbose: Whether to print debate progress
                
            Returns:
                Complete debate transcript with synthesis
            """
            if agent_roles is None:
                agent_roles = [AgentRole.PROPONENT, AgentRole.OPPONENT]
            
            transcript = DebateTranscript(question=question)
            
            if verbose:
                print(f"\nQuestion: {question}\n")
                print(f"{'='*60}")
                print(f"MULTI-AGENT DEBATE")
                print(f"{'='*60}")
                print(f"Participants: {len(agent_roles)}")
                print(f"Rounds: {self.num_rounds}\n")
            
            # Create agents
            agents = []
            for i, role in enumerate(agent_roles):
                agent = DebateAgent(
                    agent_id=f"Agent_{i+1}",
                    role=role,
                    model_name=self.model_name
                )
                agents.append(agent)
                if verbose:
                    print(f"  {agent.agent_id}: {role.value}")
            
            print()
            
            # Round 1: Opening statements
            if verbose:
                print(f"\n{'='*60}")
                print(f"ROUND 1: Opening Statements")
                print('='*60)
            
            for agent in agents:
                statement = agent.generate_opening_statement(question)
                
                message = DebateMessage(
                    round=1,
                    agent_id=agent.agent_id,
                    role=agent.role,
                    content=statement
                )
                transcript.add_message(message)
                
                if verbose:
                    print(f"\n{agent.agent_id} ({agent.role.value}):")
                    print(f"{statement}")
            
            # Subsequent rounds: Responses and rebuttals
            for round_num in range(2, self.num_rounds + 1):
                if verbose:
                    print(f"\n{'='*60}")
                    print(f"ROUND {round_num}: Discussion")
                    print('='*60)
                
                # Get previous round messages for context
                prev_messages = transcript.get_round_messages(round_num - 1)
                
                for agent in agents:
                    # Agent responds to previous arguments
                    response = agent.respond_to_arguments(question, prev_messages)
                    
                    message = DebateMessage(
                        round=round_num,
                        agent_id=agent.agent_id,
                        role=agent.role,
                        content=response
                    )
                    transcript.add_message(message)
                    
                    if verbose:
                        print(f"\n{agent.agent_id} ({agent.role.value}):")
                        print(f"{response}")
            
            # Synthesis: Combine insights from debate
            if verbose:
                print(f"\n{'='*60}")
                print("SYNTHESIS")
                print('='*60)
            
            synthesis = self._synthesize_debate(question, transcript)
            transcript.final_synthesis = synthesis
            
            if verbose:
                print(f"\n{synthesis}")
            
            return transcript
        
        def _synthesize_debate(self, question: str, transcript: DebateTranscript) -> str:
            """
            Synthesize the debate into a final answer.
            
            Args:
                question: Original question
                transcript: Complete debate transcript
                
            Returns:
                Synthesized answer
            """
            # Build debate summary
            debate_text = []
            for msg in transcript.messages:
                debate_text.append(f"[Round {msg.round}] {msg.agent_id} ({msg.role.value}):\n{msg.content}")
            
            debate_summary = "\n\n".join(debate_text)
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a moderator synthesizing a debate. 
                
    Review all arguments presented and create a balanced, thoughtful conclusion that:
    1. Acknowledges strong points from all sides
    2. Identifies areas of consensus
    3. Notes remaining disagreements
    4. Provides a nuanced final answer
    
    Be objective and fair to all perspectives."""),
                ("human", """Question: {question}
    
    Debate Transcript:
    {debate}
    
    Synthesis:""")
            ])
            
            chain = prompt | self.moderator_llm | StrOutputParser()
            synthesis = chain.invoke({
                "question": question,
                "debate": debate_summary
            })
            
            return synthesis
    
    
    def demonstrate_debate_pattern():
        """Demonstrates the Debate/Discussion pattern."""
        
        print("=" * 80)
        print("PATTERN 012: Debate/Discussion (Multi-Agent)")
        print("=" * 80)
        print()
        print("Multi-Agent Debate improves reasoning through:")
        print("1. Multiple Perspectives: Different viewpoints represented")
        print("2. Adversarial Collaboration: Agents challenge each other")
        print("3. Iterative Refinement: Arguments evolve through rounds")
        print("4. Synthesis: Best insights combined into final answer")
        print()
        
        # Create debate system
        system = DebateSystem(num_rounds=2)
        
        # Test questions
        questions = [
            "Should artificial intelligence development be regulated by governments?",
            "Is remote work more productive than office work for software development?"
        ]
        
        for idx, question in enumerate(questions, 1):
            print(f"\n{'='*80}")
            print(f"Debate {idx}")
            print('='*80)
            
            try:
                # Conduct debate with different roles
                roles = [AgentRole.PROPONENT, AgentRole.OPPONENT, AgentRole.SKEPTIC]
                
                transcript = system.conduct_debate(
                    question=question,
                    agent_roles=roles,
                    verbose=True
                )
                
                print(f"\n\n{'='*80}")
                print("DEBATE SUMMARY")
                print('='*80)
                print(f"\nQuestion: {transcript.question}")
                print(f"Total Messages: {len(transcript.messages)}")
                print(f"Rounds: {system.num_rounds}")
                print(f"\nFinal Synthesis:\n{transcript.final_synthesis}")
                
            except Exception as e:
                print(f"\n✗ Error: {str(e)}")
                import traceback
                traceback.print_exc()
        
        # Summary
        print("\n\n" + "=" * 80)
        print("DEBATE/DISCUSSION PATTERN DEMONSTRATION COMPLETE")
        print("=" * 80)
        print()
        print("Key Features Demonstrated:")
        print("1. Multiple Agent Roles: Proponent, Opponent, Skeptic")
        print("2. Structured Rounds: Opening statements followed by discussion")
        print("3. Contextual Responses: Agents respond to previous arguments")
        print("4. Moderator Synthesis: Balanced conclusion from all perspectives")
        print("5. Transcript Management: Complete debate history maintained")
        print()
        print("Advantages:")
        print("- Reduces individual bias")
        print("- More thorough problem exploration")
        print("- Error correction through peer review")
        print("- Higher quality conclusions")
        print()
        print("When to use Debate Pattern:")
        print("- Complex decision-making")
        print("- Ethical or controversial questions")
        print("- Strategy formulation")
        print("- Content validation and review")
        print()
        print("LangChain Components Used:")
        print("- Multiple ChatOpenAI instances: Different agents")
        print("- Role-based system prompts: Agent personas")
        print("- Structured debate management: Rounds and synthesis")
        print("- Context building: Previous arguments inform responses")
        print()
    
    
    if __name__ == "__main__":
        # Check for API key
        if not os.getenv("OPENAI_API_KEY"):
            print("Error: OPENAI_API_KEY not found in environment variables.")
            print("Please set it in your .env file or environment.")
            exit(1)
        
        demonstrate_debate_pattern()
    
        

    Agentic Design Pattern: Debate Discussion

    Pattern 012: Debate Discussion – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Debate Discussion pattern implementation.


    Pattern Description

    Pattern 012: Debate/Discussion (Multi-Agent)

    Description: Multiple agents engage in debate or discussion to reach better conclusions. Each agent represents different perspectives, argues for their position, and the system synthesizes the best answer from the debate. This reduces bias and improves reasoning quality through adversarial collaboration.

    Key Concepts: – Multiple Perspectives: Each agent takes different viewpoint – Argument Generation: Agents present reasoned arguments – Counter-Arguments: Agents respond to each other’s points – Synthesis: Final answer combines best insights from debate – Moderation: Optional moderator guides discussion

    Benefits: – Reduced individual bias – More thorough exploration of problem space – Error correction through peer review – Higher quality conclusions

    Use Cases: – Complex decision-making – Ethical dilemmas – Strategy formulation – Content review and validation

    LangChain Implementation: Multiple LLM instances with different personas engaging in structured debate.


    File Statistics

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

    Module Documentation

    """
    Pattern 012: Debate/Discussion (Multi-Agent)
    
    Description:
        Multiple agents engage in debate or discussion to reach better conclusions.
        Each agent represents different perspectives, argues for their position,
        and the system synthesizes the best answer from the debate. This reduces
        bias and improves reasoning quality through adversarial collaboration.
    
    Key Concepts:
        - Multiple Perspectives: Each agent takes different viewpoint
        - Argument Generation: Agents present reasoned arguments
        - Counter-Arguments: Agents respond to each other's points
        - Synthesis: Final answer combines best insights from debate
        - Moderation: Optional moderator guides discussion
    
    Benefits:
        - Reduced individual bias
        - More thorough exploration of problem space
        - Error correction through peer review
        - Higher quality conclusions
    
    Use Cases:
        - Complex decision-making
        - Ethical dilemmas
        - Strategy formulation
        - Content review and validation
    
    LangChain Implementation:
        Multiple LLM instances with different personas engaging in structured debate.
    """
    

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


    Import Statements

    Standard Library Imports

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

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

    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 traceback
    

    Purpose: Additional dependencies for specialized functionality.


    Classes

    Class: AgentRole

    Lines 47-53

    Inherits from: Enum

    Purpose:

    Roles agents can take in debate.
    

    Class: DebateMessage

    Lines 57-63

    Purpose:

    Represents a message in the debate.
    

    Class: DebateTranscript

    Lines 67-79

    Purpose:

    Complete transcript of a debate.
    

    Methods:

    add_message(self, message)

    Lines 73-75

    Purpose:

    Add a message to the transcript.
    

    Parameters: – message: Message

    get_round_messages(self, round_num)

    Lines 77-79

    Purpose:

    Get all messages from a specific round.
    

    Parameters: – round_num: Round num

    Returns: <ast.Subscript object at 0x000001A06D60F4D0>


    Class: DebateAgent

    Lines 82-149

    Purpose:

    Individual agent participating in debate.
    

    Methods:

    __init__(self, agent_id, role, model_name)

    Lines 85-109

    Purpose:

    Initialize a debate agent.
    
    Args:
        agent_id: Unique identifier for the agent
        role: Role the agent plays in debate
        model_name: Name of the OpenAI model
    

    Parameters: – agent_id: Agent id – role: Role – model_name: Language model instance

    generate_opening_statement(self, question)

    Lines 111-121

    Purpose:

    Generate opening statement for the debate.
    

    Parameters: – question: Question to answer

    Returns: str

    respond_to_arguments(self, question, previous_messages)

    Lines 123-149

    Purpose:

    Generate response to previous arguments.
    

    Parameters: – question: Question to answer – previous_messages: Previous messages

    Returns: str


    Class: DebateSystem

    Lines 152-315

    Purpose:

    System for managing multi-agent debate.
    

    Methods:

    __init__(self, model_name, num_rounds)

    Lines 155-169

    Purpose:

    Initialize the debate system.
    
    Args:
        model_name: Name of the OpenAI model
        num_rounds: Number of debate rounds
    

    Parameters: – model_name: Language model instance – num_rounds: Num rounds

    conduct_debate(self, question, agent_roles, verbose)

    Lines 171-271

    Purpose:

    Conduct a structured debate on the question.
    
    Args:
        question: Question to debate
        agent_roles: Roles for agents (default: proponent and opponent)
        verbose: Whether to print debate progress
    
    Returns:
        Complete debate transcript with synthesis
    

    Parameters: – question: Question to answer – agent_roles: Agent roles – verbose: Enable detailed logging

    Returns: DebateTranscript

    _synthesize_debate(self, question, transcript)

    Lines 273-315

    Purpose:

    Synthesize the debate into a final answer.
    
    Args:
        question: Original question
        transcript: Complete debate transcript
    
    Returns:
        Synthesized answer
    

    Parameters: – question: Question to answer – transcript: Transcript

    Returns: str


    Functions

    demonstrate_debate_pattern()

    Lines 318-398

    Purpose:

    Demonstrates the Debate/Discussion pattern.
    

    Key Concepts Summary

    Pattern Overview

    The Debate Discussion 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 011
    • Pattern 013
    • Other Multi-Agent Patterns (012-019)

    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 Debate Discussion 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: 012_debate_discussion.py
    Total Lines: 409
    Pattern Category: AI Agent Design Pattern #012


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Metacognitive Monitoring

    
        """
    Pattern 011: Metacognitive Monitoring
    
    Description:
        Agent monitors its own thinking process, confidence levels, and decision-making.
        Provides self-awareness about reasoning quality, uncertainty, and when to seek
        help or alternative strategies. Essential for safe and reliable AI systems.
    
    Key Concepts:
        - Confidence Estimation: Quantify certainty in outputs
        - Uncertainty Quantification: Identify areas of low confidence
        - Self-Monitoring: Track reasoning quality during execution
        - Strategy Selection: Choose appropriate approaches based on confidence
        - Error Detection: Identify potential mistakes proactively
    
    Use Cases:
        - Safety-critical applications
        - Decision validation and verification
        - Knowing when to escalate to humans
        - Quality control and error detection
    
    LangChain Implementation:
        Confidence scoring chains with self-monitoring and adaptive strategies.
    """
    
    import os
    from typing import List, Dict, Any, Optional, Tuple
    from dataclasses import dataclass, field
    from enum import Enum
    from dotenv import load_dotenv
    
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser
    from langchain_openai import ChatOpenAI
    
    # Load environment variables
    load_dotenv()
    
    
    class ConfidenceLevel(Enum):
        """Confidence levels for reasoning."""
        VERY_LOW = "very_low"
        LOW = "low"
        MEDIUM = "medium"
        HIGH = "high"
        VERY_HIGH = "very_high"
    
    
    @dataclass
    class MetacognitiveState:
        """Represents the agent's metacognitive state."""
        task: str
        current_step: str
        confidence: float = 0.5  # 0.0 to 1.0
        uncertainty_sources: List[str] = field(default_factory=list)
        alternative_strategies: List[str] = field(default_factory=list)
        should_seek_help: bool = False
        reasoning_quality: float = 0.5
        
        def get_confidence_level(self) -> ConfidenceLevel:
            """Convert numeric confidence to level."""
            if self.confidence >= 0.9:
                return ConfidenceLevel.VERY_HIGH
            elif self.confidence >= 0.7:
                return ConfidenceLevel.HIGH
            elif self.confidence >= 0.5:
                return ConfidenceLevel.MEDIUM
            elif self.confidence >= 0.3:
                return ConfidenceLevel.LOW
            else:
                return ConfidenceLevel.VERY_LOW
    
    
    class MetacognitiveAgent:
        """Agent with metacognitive monitoring capabilities."""
        
        def __init__(self, model_name: str = "gpt-3.5-turbo",
                     confidence_threshold: float = 0.6,
                     help_threshold: float = 0.4):
            """
            Initialize the Metacognitive Agent.
            
            Args:
                model_name: Name of the OpenAI model
                confidence_threshold: Minimum acceptable confidence
                help_threshold: Threshold below which to seek help
            """
            self.llm = ChatOpenAI(
                model=model_name,
                temperature=0.3,
                api_key=os.getenv("OPENAI_API_KEY")
            )
            
            self.confidence_threshold = confidence_threshold
            self.help_threshold = help_threshold
            self.metacognitive_log: List[MetacognitiveState] = []
        
        def estimate_confidence(self, task: str, reasoning: str, answer: str) -> Tuple[float, List[str]]:
            """
            Estimate confidence in the answer and identify uncertainty sources.
            
            Args:
                task: The task being solved
                reasoning: The reasoning process
                answer: The proposed answer
                
            Returns:
                Tuple of (confidence_score, uncertainty_sources)
            """
            prompt = ChatPromptTemplate.from_messages([
                ("system", """Evaluate your confidence in this answer. Consider:
    1. Completeness of reasoning
    2. Availability of relevant information
    3. Complexity of the task
    4. Potential ambiguities or uncertainties
    
    Provide:
    CONFIDENCE: [0.0-1.0]
    UNCERTAINTY_SOURCES:
    - [source 1]
    - [source 2]
    ...
    
    Be honest about limitations and uncertainties."""),
                ("human", """Task: {task}
    
    Reasoning:
    {reasoning}
    
    Answer: {answer}
    
    Confidence Evaluation:""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            response = chain.invoke({
                "task": task,
                "reasoning": reasoning,
                "answer": answer
            })
            
            # Parse confidence and uncertainty sources
            confidence = 0.5
            uncertainty_sources = []
            
            for line in response.split('\n'):
                line = line.strip()
                if line.upper().startswith('CONFIDENCE:'):
                    try:
                        conf_str = line.split(':', 1)[1].strip()
                        confidence = float(conf_str)
                    except:
                        pass
                elif line.startswith('-') and 'UNCERTAINTY_SOURCES' in response.upper():
                    source = line.lstrip('- ').strip()
                    if source:
                        uncertainty_sources.append(source)
            
            return confidence, uncertainty_sources
        
        def identify_alternative_strategies(self, task: str, current_approach: str,
                                           issues: List[str]) -> List[str]:
            """
            Identify alternative strategies when current approach has low confidence.
            
            Args:
                task: The task being solved
                current_approach: Current approach being used
                issues: Identified issues with current approach
                
            Returns:
                List of alternative strategies
            """
            issues_text = "\n".join([f"- {issue}" for issue in issues])
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", """Suggest alternative strategies for solving this task given the identified issues.
    Provide 2-3 concrete alternative approaches.
    
    Format:
    1. [Strategy 1]
    2. [Strategy 2]
    3. [Strategy 3]"""),
                ("human", """Task: {task}
    
    Current Approach: {current_approach}
    
    Issues:
    {issues}
    
    Alternative Strategies:""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            response = chain.invoke({
                "task": task,
                "current_approach": current_approach,
                "issues": issues_text
            })
            
            # Parse strategies
            strategies = []
            for line in response.strip().split('\n'):
                line = line.strip()
                if line and (line[0].isdigit() or line.startswith('-')):
                    strategy = line.split('.', 1)[-1].strip().lstrip('- ').strip()
                    if strategy:
                        strategies.append(strategy)
            
            return strategies[:3]
        
        def evaluate_reasoning_quality(self, reasoning: str) -> float:
            """
            Evaluate the quality of the reasoning process.
            
            Args:
                reasoning: The reasoning to evaluate
                
            Returns:
                Quality score 0.0-1.0
            """
            prompt = ChatPromptTemplate.from_messages([
                ("system", """Evaluate the quality of this reasoning. Consider:
    - Logical consistency
    - Completeness
    - Clarity
    - Rigor
    
    Provide a quality score from 0.0 (poor) to 1.0 (excellent).
    Respond with just the number."""),
                ("human", """Reasoning:
    {reasoning}
    
    Quality Score (0.0-1.0):""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            
            try:
                response = chain.invoke({"reasoning": reasoning})
                score = float(response.strip())
                return max(0.0, min(1.0, score))
            except:
                return 0.5
        
        def solve_with_monitoring(self, task: str, verbose: bool = True) -> Dict[str, Any]:
            """
            Solve a task with metacognitive monitoring.
            
            Args:
                task: The task to solve
                verbose: Whether to print monitoring information
                
            Returns:
                Dictionary with solution and metacognitive information
            """
            if verbose:
                print(f"\nTask: {task}\n")
                print("="*60)
                print("METACOGNITIVE MONITORING ACTIVE")
                print("="*60)
            
            # Step 1: Initial reasoning
            if verbose:
                print("\nStep 1: Initial Reasoning")
            
            reasoning_prompt = ChatPromptTemplate.from_messages([
                ("system", "Solve this task with clear step-by-step reasoning."),
                ("human", "{task}")
            ])
            
            chain = reasoning_prompt | self.llm | StrOutputParser()
            reasoning = chain.invoke({"task": task})
            
            # Extract answer
            answer_lines = reasoning.strip().split('\n')
            answer = answer_lines[-1] if answer_lines else reasoning[:100]
            
            if verbose:
                print(f"Reasoning: {reasoning[:200]}...")
                print(f"Answer: {answer}")
            
            # Step 2: Estimate confidence
            if verbose:
                print("\nStep 2: Confidence Estimation")
            
            confidence, uncertainty_sources = self.estimate_confidence(task, reasoning, answer)
            
            # Step 3: Evaluate reasoning quality
            reasoning_quality = self.evaluate_reasoning_quality(reasoning)
            
            if verbose:
                print(f"Confidence: {confidence:.2f} ({ConfidenceLevel(confidence >= 0.9 and 'very_high' or confidence >= 0.7 and 'high' or confidence >= 0.5 and 'medium' or confidence >= 0.3 and 'low' or 'very_low').value})")
                print(f"Reasoning Quality: {reasoning_quality:.2f}")
                if uncertainty_sources:
                    print(f"Uncertainty Sources: {len(uncertainty_sources)}")
                    for source in uncertainty_sources:
                        print(f"  - {source}")
            
            # Step 4: Determine if help needed
            should_seek_help = confidence < self.help_threshold
            
            # Step 5: Identify alternative strategies if confidence low
            alternative_strategies = []
            if confidence < self.confidence_threshold:
                if verbose:
                    print(f"\n⚠️ Low confidence detected (threshold: {self.confidence_threshold})")
                    print("Step 3: Identifying Alternative Strategies")
                
                alternative_strategies = self.identify_alternative_strategies(
                    task, "Initial direct reasoning", uncertainty_sources
                )
                
                if verbose:
                    print("Alternative Strategies:")
                    for i, strategy in enumerate(alternative_strategies, 1):
                        print(f"  {i}. {strategy}")
            
            # Step 6: Decide on action
            if should_seek_help:
                action = "ESCALATE_TO_HUMAN"
                recommendation = "Task complexity or uncertainty requires human oversight"
            elif confidence < self.confidence_threshold:
                action = "TRY_ALTERNATIVE_STRATEGY"
                recommendation = f"Recommend trying: {alternative_strategies[0] if alternative_strategies else 'alternative approach'}"
            else:
                action = "PROCEED_WITH_ANSWER"
                recommendation = "Confidence sufficient to proceed"
            
            if verbose:
                print(f"\nStep 4: Decision")
                print(f"Action: {action}")
                print(f"Recommendation: {recommendation}")
            
            # Create metacognitive state
            state = MetacognitiveState(
                task=task,
                current_step="Solution",
                confidence=confidence,
                uncertainty_sources=uncertainty_sources,
                alternative_strategies=alternative_strategies,
                should_seek_help=should_seek_help,
                reasoning_quality=reasoning_quality
            )
            self.metacognitive_log.append(state)
            
            return {
                "task": task,
                "reasoning": reasoning,
                "answer": answer,
                "confidence": confidence,
                "confidence_level": state.get_confidence_level().value,
                "reasoning_quality": reasoning_quality,
                "uncertainty_sources": uncertainty_sources,
                "alternative_strategies": alternative_strategies,
                "should_seek_help": should_seek_help,
                "recommended_action": action,
                "recommendation": recommendation
            }
    
    
    def demonstrate_metacognitive_monitoring():
        """Demonstrates the Metacognitive Monitoring pattern."""
        
        print("=" * 80)
        print("PATTERN 011: Metacognitive Monitoring")
        print("=" * 80)
        print()
        print("Metacognitive Monitoring enables self-awareness through:")
        print("1. Confidence Estimation: Quantify certainty in outputs")
        print("2. Uncertainty Detection: Identify sources of uncertainty")
        print("3. Quality Assessment: Evaluate reasoning quality")
        print("4. Strategy Selection: Choose alternative approaches when needed")
        print("5. Help-Seeking: Know when to escalate to humans")
        print()
        
        # Create agent
        agent = MetacognitiveAgent(
            confidence_threshold=0.6,
            help_threshold=0.4
        )
        
        # Test tasks with varying complexity
        tasks = [
            "What is 15 + 27?",  # High confidence expected
            "Explain the implications of quantum entanglement for future communication technologies",  # Medium confidence
            "Should Company X acquire Company Y given uncertain market conditions?"  # Low confidence
        ]
        
        for idx, task in enumerate(tasks, 1):
            print(f"\n{'='*80}")
            print(f"Example {idx}")
            print('='*80)
            
            try:
                result = agent.solve_with_monitoring(task, verbose=True)
                
                print(f"\n\n{'='*80}")
                print("METACOGNITIVE SUMMARY")
                print('='*80)
                print(f"\nTask: {result['task']}")
                print(f"Answer: {result['answer']}")
                print(f"\nConfidence: {result['confidence']:.2f} ({result['confidence_level']})")
                print(f"Reasoning Quality: {result['reasoning_quality']:.2f}")
                print(f"Should Seek Help: {'Yes' if result['should_seek_help'] else 'No'}")
                print(f"\nRecommended Action: {result['recommended_action']}")
                print(f"Recommendation: {result['recommendation']}")
                
            except Exception as e:
                print(f"\n✗ Error: {str(e)}")
                import traceback
                traceback.print_exc()
        
        # Summary
        print("\n\n" + "=" * 80)
        print("METACOGNITIVE MONITORING PATTERN DEMONSTRATION COMPLETE")
        print("=" * 80)
        print()
        print("Key Features Demonstrated:")
        print("1. Confidence Estimation: Quantified certainty levels")
        print("2. Uncertainty Sources: Identified specific areas of uncertainty")
        print("3. Quality Assessment: Evaluated reasoning quality")
        print("4. Alternative Strategies: Suggested alternatives for low confidence")
        print("5. Decision Making: Determined appropriate actions based on confidence")
        print()
        print("Advantages:")
        print("- Self-awareness of limitations")
        print("- Proactive error detection")
        print("- Intelligent help-seeking")
        print("- Adaptive strategy selection")
        print()
        print("When to use Metacognitive Monitoring:")
        print("- Safety-critical applications")
        print("- High-stakes decision-making")
        print("- Uncertain or ambiguous tasks")
        print("- When transparency is required")
        print()
        print("LangChain Components Used:")
        print("- ChatPromptTemplate: Confidence estimation and evaluation")
        print("- StrOutputParser: Parse monitoring outputs")
        print("- Threshold-based decision making")
        print("- Structured metacognitive state tracking")
        print()
    
    
    if __name__ == "__main__":
        # Check for API key
        if not os.getenv("OPENAI_API_KEY"):
            print("Error: OPENAI_API_KEY not found in environment variables.")
            print("Please set it in your .env file or environment.")
            exit(1)
        
        demonstrate_metacognitive_monitoring()
    
        

    Agentic Design Pattern: Metacognitive Monitoring

    Pattern 011: Metacognitive Monitoring – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Metacognitive Monitoring pattern implementation.


    Pattern Description

    Pattern 011: Metacognitive Monitoring

    Description: Agent monitors its own thinking process, confidence levels, and decision-making. Provides self-awareness about reasoning quality, uncertainty, and when to seek help or alternative strategies. Essential for safe and reliable AI systems.

    Key Concepts: – Confidence Estimation: Quantify certainty in outputs – Uncertainty Quantification: Identify areas of low confidence – Self-Monitoring: Track reasoning quality during execution – Strategy Selection: Choose appropriate approaches based on confidence – Error Detection: Identify potential mistakes proactively

    Use Cases: – Safety-critical applications – Decision validation and verification – Knowing when to escalate to humans – Quality control and error detection

    LangChain Implementation: Confidence scoring chains with self-monitoring and adaptive strategies.


    File Statistics

    • Total Lines: 454
    • Classes: 3
    • Functions: 1
    • Constants: 0

    Module Documentation

    """
    Pattern 011: Metacognitive Monitoring
    
    Description:
        Agent monitors its own thinking process, confidence levels, and decision-making.
        Provides self-awareness about reasoning quality, uncertainty, and when to seek
        help or alternative strategies. Essential for safe and reliable AI systems.
    
    Key Concepts:
        - Confidence Estimation: Quantify certainty in outputs
        - Uncertainty Quantification: Identify areas of low confidence
        - Self-Monitoring: Track reasoning quality during execution
        - Strategy Selection: Choose appropriate approaches based on confidence
        - Error Detection: Identify potential mistakes proactively
    
    Use Cases:
        - Safety-critical applications
        - Decision validation and verification
        - Knowing when to escalate to humans
        - Quality control and error detection
    
    LangChain Implementation:
        Confidence scoring chains with self-monitoring and adaptive strategies.
    """
    

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


    Import Statements

    Standard Library Imports

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

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

    LangChain Imports

    from langchain_openai.ChatOpenAI
    

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

    Other Imports

    from dotenv.load_dotenv
    from traceback
    

    Purpose: Additional dependencies for specialized functionality.


    Classes

    Class: ConfidenceLevel

    Lines 40-46

    Inherits from: Enum

    Purpose:

    Confidence levels for reasoning.
    

    Class: MetacognitiveState

    Lines 50-71

    Purpose:

    Represents the agent's metacognitive state.
    

    Methods:

    get_confidence_level(self)

    Lines 60-71

    Purpose:

    Convert numeric confidence to level.
    

    Parameters:

    Returns: ConfidenceLevel


    Class: MetacognitiveAgent

    Lines 74-359

    Purpose:

    Agent with metacognitive monitoring capabilities.
    

    Methods:

    __init__(self, model_name, confidence_threshold, help_threshold)

    Lines 77-96

    Purpose:

    Initialize the Metacognitive Agent.
    
    Args:
        model_name: Name of the OpenAI model
        confidence_threshold: Minimum acceptable confidence
        help_threshold: Threshold below which to seek help
    

    Parameters: – model_name: Language model instance – confidence_threshold: Confidence threshold – help_threshold: Help threshold

    estimate_confidence(self, task, reasoning, answer)

    Lines 98-159

    Purpose:

    Estimate confidence in the answer and identify uncertainty sources.
    
    Args:
        task: The task being solved
        reasoning: The reasoning process
        answer: The proposed answer
    
    Returns:
        Tuple of (confidence_score, uncertainty_sources)
    

    Parameters: – task: Task – reasoning: Reasoning – answer: Answer

    Returns: <ast.Subscript object at 0x000001A06D1592D0>

    identify_alternative_strategies(self, task, current_approach, issues)

    Lines 161-210

    Purpose:

    Identify alternative strategies when current approach has low confidence.
    
    Args:
        task: The task being solved
        current_approach: Current approach being used
        issues: Identified issues with current approach
    
    Returns:
        List of alternative strategies
    

    Parameters: – task: Task – current_approach: Current approach – issues: Issues

    Returns: <ast.Subscript object at 0x000001A06D700F50>

    evaluate_reasoning_quality(self, reasoning)

    Lines 212-244

    Purpose:

    Evaluate the quality of the reasoning process.
    
    Args:
        reasoning: The reasoning to evaluate
    
    Returns:
        Quality score 0.0-1.0
    

    Parameters: – reasoning: Reasoning

    Returns: float

    solve_with_monitoring(self, task, verbose)

    Lines 246-359

    Purpose:

    Solve a task with metacognitive monitoring.
    
    Args:
        task: The task to solve
        verbose: Whether to print monitoring information
    
    Returns:
        Dictionary with solution and metacognitive information
    

    Parameters: – task: Task – verbose: Enable detailed logging

    Returns: <ast.Subscript object at 0x000001A06D156C50>


    Functions

    demonstrate_metacognitive_monitoring()

    Lines 362-443

    Purpose:

    Demonstrates the Metacognitive Monitoring pattern.
    

    Key Concepts Summary

    Pattern Overview

    The Metacognitive Monitoring 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 010
    • Pattern 012
    • Other Reasoning Patterns (001-011)

    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 Metacognitive Monitoring 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: 011_metacognitive_monitoring.py
    Total Lines: 454
    Pattern Category: AI Agent Design Pattern #011


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Constitutional Ai

    
        """
    Pattern 010: Constitutional AI
    
    Description:
        Ensures agent behavior aligns with explicit principles or "constitution."
        The agent generates responses, evaluates them against constitutional principles,
        and revises if violations are found. This creates safer, more aligned AI systems.
    
    Key Concepts:
        - Constitution: Set of principles/rules the agent must follow
        - Self-Critique: Agent evaluates its own outputs against principles
        - Revision: Generate improved response that adheres to constitution
        - Iterative Alignment: Multiple rounds of critique and revision
        - Transparency: Explicit principles make behavior interpretable
    
    Constitutional Principles Examples:
        - Harmlessness: Avoid harmful, dangerous, or offensive content
        - Helpfulness: Provide useful, relevant information
        - Honesty: Be truthful, acknowledge uncertainty
        - Privacy: Respect privacy and confidentiality
        - Fairness: Avoid bias and discrimination
    
    Use Cases:
        - Production AI systems requiring safety
        - Content moderation and filtering
        - Ethical decision-making systems
        - Regulated domains (healthcare, finance, legal)
    
    LangChain Implementation:
        Constitutional chain with critique and revision loops based on principles.
    """
    
    import os
    from typing import List, Dict, Any, Optional
    from dataclasses import dataclass, field
    from enum import Enum
    from dotenv import load_dotenv
    
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser
    from langchain_openai import ChatOpenAI
    
    # Load environment variables
    load_dotenv()
    
    
    @dataclass
    class Principle:
        """Represents a constitutional principle."""
        name: str
        description: str
        critique_prompt: str
        weight: float = 1.0  # Importance weight
    
    
    @dataclass
    class ViolationReport:
        """Report of constitutional violations."""
        principle: str
        violated: bool
        explanation: str
        severity: float = 0.0  # 0.0 (no violation) to 1.0 (severe)
    
    
    @dataclass
    class ConstitutionalReview:
        """Complete review against constitution."""
        original_output: str
        violations: List[ViolationReport] = field(default_factory=list)
        total_severity: float = 0.0
        passes: bool = True
        
        def calculate_severity(self):
            """Calculate total severity of violations."""
            self.total_severity = sum(v.severity for v in self.violations if v.violated)
            self.passes = self.total_severity == 0.0
    
    
    class Constitution:
        """Set of principles for constitutional AI."""
        
        def __init__(self, name: str = "Default Constitution"):
            """Initialize constitution with default or custom principles."""
            self.name = name
            self.principles: List[Principle] = []
        
        def add_principle(self, principle: Principle):
            """Add a principle to the constitution."""
            self.principles.append(principle)
        
        def get_principles_text(self) -> str:
            """Get formatted text of all principles."""
            lines = [f"Constitution: {self.name}\n"]
            for i, p in enumerate(self.principles, 1):
                lines.append(f"{i}. {p.name}: {p.description}")
            return "\n".join(lines)
        
        @staticmethod
        def get_default() -> 'Constitution':
            """Get default constitution with common principles."""
            constitution = Constitution("Default AI Constitution")
            
            # Harmlessness
            constitution.add_principle(Principle(
                name="Harmlessness",
                description="Responses must not include harmful, dangerous, illegal, or offensive content",
                critique_prompt="""Does the response contain any harmful, dangerous, illegal, or offensive content?
    Consider: violence, illegal activities, self-harm, harassment, discrimination.
    Answer YES if violations exist, NO if the response is harmless."""
            ))
            
            # Helpfulness
            constitution.add_principle(Principle(
                name="Helpfulness",
                description="Responses should be helpful, relevant, and answer the user's question",
                critique_prompt="""Is the response helpful and relevant to the user's query?
    Does it provide useful information or assistance?
    Answer YES if unhelpful, NO if the response is helpful."""
            ))
            
            # Honesty
            constitution.add_principle(Principle(
                name="Honesty",
                description="Responses should be truthful and acknowledge uncertainty when appropriate",
                critique_prompt="""Does the response contain false information or make claims without acknowledging uncertainty?
    Answer YES if dishonest or overconfident, NO if honest and appropriately uncertain."""
            ))
            
            # Privacy
            constitution.add_principle(Principle(
                name="Privacy",
                description="Responses must respect privacy and not request or share personal information inappropriately",
                critique_prompt="""Does the response inappropriately request, share, or handle personal information?
    Answer YES if privacy is violated, NO if privacy is respected."""
            ))
            
            # Fairness
            constitution.add_principle(Principle(
                name="Fairness",
                description="Responses should be fair and avoid bias or discrimination",
                critique_prompt="""Does the response show bias or discrimination based on race, gender, religion, or other protected characteristics?
    Answer YES if biased or discriminatory, NO if fair and balanced."""
            ))
            
            return constitution
    
    
    class ConstitutionalAI:
        """Agent that follows constitutional principles."""
        
        def __init__(self, constitution: Constitution,
                     model_name: str = "gpt-3.5-turbo",
                     max_revisions: int = 3,
                     severity_threshold: float = 0.3):
            """
            Initialize Constitutional AI agent.
            
            Args:
                constitution: The constitution to follow
                model_name: Name of the OpenAI model
                max_revisions: Maximum number of revision attempts
                severity_threshold: Severity threshold for triggering revision
            """
            self.constitution = constitution
            self.max_revisions = max_revisions
            self.severity_threshold = severity_threshold
            
            self.llm = ChatOpenAI(
                model=model_name,
                temperature=0.7,
                api_key=os.getenv("OPENAI_API_KEY")
            )
            
            self.critic_llm = ChatOpenAI(
                model=model_name,
                temperature=0.2,  # Lower temperature for consistent critique
                api_key=os.getenv("OPENAI_API_KEY")
            )
        
        def generate_initial(self, query: str, context: str = "") -> str:
            """
            Generate initial response to query.
            
            Args:
                query: User query
                context: Additional context
                
            Returns:
                Initial response
            """
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are a helpful AI assistant. Respond to the user's query."),
                ("human", """{context}
    
    Query: {query}
    
    Response:""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            response = chain.invoke({
                "query": query,
                "context": context if context else ""
            })
            
            return response.strip()
        
        def critique_against_principle(self, output: str, query: str,
                                       principle: Principle) -> ViolationReport:
            """
            Critique output against a single principle.
            
            Args:
                output: Output to critique
                query: Original query
                principle: Principle to check against
                
            Returns:
                ViolationReport indicating if principle is violated
            """
            prompt = ChatPromptTemplate.from_messages([
                ("system", f"""You are evaluating whether a response violates the principle: "{principle.name}"
    
    Principle: {principle.description}
    
    {principle.critique_prompt}
    
    If YES (violation), also rate severity from 0.1 (minor) to 1.0 (severe).
    
    Format your response as:
    VIOLATION: YES/NO
    SEVERITY: [0.0-1.0]
    EXPLANATION: [Brief explanation]"""),
                ("human", """Query: {query}
    
    Response to evaluate:
    {output}
    
    Evaluation:""")
            ])
            
            chain = prompt | self.critic_llm | StrOutputParser()
            critique_text = chain.invoke({
                "query": query,
                "output": output
            })
            
            # Parse critique
            violated = False
            severity = 0.0
            explanation = ""
            
            for line in critique_text.split('\n'):
                line = line.strip()
                if line.upper().startswith('VIOLATION:'):
                    violated = 'YES' in line.upper()
                elif line.upper().startswith('SEVERITY:'):
                    try:
                        severity_str = line.split(':', 1)[1].strip()
                        severity = float(severity_str)
                    except:
                        severity = 0.5 if violated else 0.0
                elif line.upper().startswith('EXPLANATION:'):
                    explanation = line.split(':', 1)[1].strip()
            
            return ViolationReport(
                principle=principle.name,
                violated=violated,
                explanation=explanation,
                severity=severity if violated else 0.0
            )
        
        def review(self, output: str, query: str) -> ConstitutionalReview:
            """
            Review output against all constitutional principles.
            
            Args:
                output: Output to review
                query: Original query
                
            Returns:
                ConstitutionalReview with all violations
            """
            review = ConstitutionalReview(original_output=output)
            
            for principle in self.constitution.principles:
                violation = self.critique_against_principle(output, query, principle)
                review.violations.append(violation)
            
            review.calculate_severity()
            
            return review
        
        def revise(self, query: str, output: str, review: ConstitutionalReview) -> str:
            """
            Revise output to address constitutional violations.
            
            Args:
                query: Original query
                output: Current output with violations
                review: Review identifying violations
                
            Returns:
                Revised output addressing violations
            """
            # Build violation summary
            violations_text = []
            for v in review.violations:
                if v.violated:
                    violations_text.append(f"- {v.principle}: {v.explanation} (severity: {v.severity:.2f})")
            
            violations_summary = "\n".join(violations_text) if violations_text else "No violations"
            
            # Get principles text
            principles_text = self.constitution.get_principles_text()
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", f"""You are revising a response to address constitutional violations.
    
    {principles_text}
    
    Revise the response to:
    1. Address all identified violations
    2. Maintain helpfulness and relevance
    3. Fully comply with all principles
    
    Generate a new, improved response that adheres to the constitution."""),
                ("human", """Original Query: {query}
    
    Current Response:
    {output}
    
    Violations Found:
    {violations}
    
    Revised Response:""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            revised = chain.invoke({
                "query": query,
                "output": output,
                "violations": violations_summary
            })
            
            return revised.strip()
        
        def generate_with_constitution(self, query: str, context: str = "",
                                       verbose: bool = True) -> Dict[str, Any]:
            """
            Generate response following constitutional principles.
            
            Args:
                query: User query
                context: Additional context
                verbose: Whether to print progress
                
            Returns:
                Dictionary with final response and review history
            """
            if verbose:
                print(f"\nQuery: {query}\n")
                print(f"Constitution: {self.constitution.name}")
                print(f"Principles: {len(self.constitution.principles)}\n")
            
            # Generate initial response
            if verbose:
                print("="*60)
                print("INITIAL GENERATION")
                print("="*60)
            
            current_output = self.generate_initial(query, context)
            
            if verbose:
                print(f"\nGenerated:\n{current_output[:200]}..." if len(current_output) > 200 else f"\nGenerated:\n{current_output}")
            
            reviews = []
            revision_count = 0
            
            # Review and revise loop
            for iteration in range(self.max_revisions + 1):
                if verbose:
                    print(f"\n{'='*60}")
                    print(f"CONSTITUTIONAL REVIEW {iteration + 1}")
                    print("="*60)
                
                # Review against constitution
                review = self.review(current_output, query)
                reviews.append(review)
                
                if verbose:
                    print(f"\nViolations Found: {sum(1 for v in review.violations if v.violated)}/{len(review.violations)}")
                    print(f"Total Severity: {review.total_severity:.2f}")
                    
                    for v in review.violations:
                        if v.violated:
                            print(f"  ⚠ {v.principle}: {v.explanation} (severity: {v.severity:.2f})")
                        else:
                            print(f"  ✓ {v.principle}: OK")
                
                # Check if passes constitutional review
                if review.passes or review.total_severity = self.max_revisions:
                    if verbose:
                        print(f"\n⚠ Max revisions ({self.max_revisions}) reached")
                    break
                
                # Revise
                if verbose:
                    print(f"\n{'='*60}")
                    print(f"REVISION {iteration + 1}")
                    print("="*60)
                
                current_output = self.revise(query, current_output, review)
                revision_count += 1
                
                if verbose:
                    print(f"\nRevised:\n{current_output[:200]}..." if len(current_output) > 200 else f"\nRevised:\n{current_output}")
            
            return {
                "query": query,
                "final_response": current_output,
                "revisions": revision_count,
                "reviews": reviews,
                "final_severity": reviews[-1].total_severity,
                "constitutional_compliance": reviews[-1].passes
            }
    
    
    def demonstrate_constitutional_ai():
        """Demonstrates the Constitutional AI pattern."""
        
        print("=" * 80)
        print("PATTERN 010: Constitutional AI")
        print("=" * 80)
        print()
        print("Constitutional AI ensures alignment with principles through:")
        print("1. Constitution: Explicit set of principles")
        print("2. Self-Critique: Evaluate outputs against principles")
        print("3. Revision: Modify responses to comply")
        print("4. Iteration: Repeat until constitutional compliance")
        print()
        
        # Create constitution
        constitution = Constitution.get_default()
        
        print(f"{constitution.get_principles_text()}\n")
        
        # Create agent
        agent = ConstitutionalAI(
            constitution=constitution,
            max_revisions=2,
            severity_threshold=0.2
        )
        
        # Test queries (some designed to trigger violations)
        queries = [
            "How can I improve my programming skills?",  # Should pass
            "Tell me how to hack into someone's email account"  # Should trigger violations
        ]
        
        for idx, query in enumerate(queries, 1):
            print(f"\n{'='*80}")
            print(f"Example {idx}")
            print('='*80)
            
            try:
                result = agent.generate_with_constitution(query, verbose=True)
                
                print(f"\n\n{'='*80}")
                print("FINAL RESULT")
                print('='*80)
                print(f"\nFinal Response:\n{result['final_response']}")
                print(f"\nRevisions Made: {result['revisions']}")
                print(f"Final Severity: {result['final_severity']:.2f}")
                print(f"Constitutional Compliance: {'✓ PASS' if result['constitutional_compliance'] else '✗ FAIL'}")
                
            except Exception as e:
                print(f"\n✗ Error: {str(e)}")
                import traceback
                traceback.print_exc()
        
        # Summary
        print("\n\n" + "=" * 80)
        print("CONSTITUTIONAL AI PATTERN DEMONSTRATION COMPLETE")
        print("=" * 80)
        print()
        print("Key Features Demonstrated:")
        print("1. Explicit Principles: Clear constitutional guidelines")
        print("2. Automated Review: Self-critique against principles")
        print("3. Iterative Revision: Multiple refinement rounds")
        print("4. Severity Tracking: Quantify violation severity")
        print("5. Compliance Verification: Ensure constitutional adherence")
        print()
        print("Advantages:")
        print("- Safer AI systems through explicit principles")
        print("- Interpretable behavior (clear rules)")
        print("- Automated alignment checking")
        print("- Flexible - can customize constitution")
        print()
        print("When to use Constitutional AI:")
        print("- Production systems requiring safety")
        print("- Regulated domains (healthcare, finance, legal)")
        print("- Content moderation and filtering")
        print("- Any application requiring ethical guidelines")
        print()
        print("LangChain Components Used:")
        print("- ChatPromptTemplate: Generation, critique, and revision")
        print("- Multiple LLM instances: Different roles and temperatures")
        print("- Structured principles: Modular constitution system")
        print("- Iterative loops: Review and revision cycles")
        print()
    
    
    if __name__ == "__main__":
        # Check for API key
        if not os.getenv("OPENAI_API_KEY"):
            print("Error: OPENAI_API_KEY not found in environment variables.")
            print("Please set it in your .env file or environment.")
            exit(1)
        
        demonstrate_constitutional_ai()
    
        

    Agentic Design Pattern: Constitutional Ai

    Pattern 010: Constitutional Ai – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Constitutional Ai pattern implementation.


    Pattern Description

    Pattern 010: Constitutional AI

    Description: Ensures agent behavior aligns with explicit principles or “constitution.” The agent generates responses, evaluates them against constitutional principles, and revises if violations are found. This creates safer, more aligned AI systems.

    Key Concepts: – Constitution: Set of principles/rules the agent must follow – Self-Critique: Agent evaluates its own outputs against principles – Revision: Generate improved response that adheres to constitution – Iterative Alignment: Multiple rounds of critique and revision – Transparency: Explicit principles make behavior interpretable

    Constitutional Principles Examples: – Harmlessness: Avoid harmful, dangerous, or offensive content – Helpfulness: Provide useful, relevant information – Honesty: Be truthful, acknowledge uncertainty – Privacy: Respect privacy and confidentiality – Fairness: Avoid bias and discrimination

    Use Cases: – Production AI systems requiring safety – Content moderation and filtering – Ethical decision-making systems – Regulated domains (healthcare, finance, legal)

    LangChain Implementation: Constitutional chain with critique and revision loops based on principles.


    File Statistics

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

    Module Documentation

    """
    Pattern 010: Constitutional AI
    
    Description:
        Ensures agent behavior aligns with explicit principles or "constitution."
        The agent generates responses, evaluates them against constitutional principles,
        and revises if violations are found. This creates safer, more aligned AI systems.
    
    Key Concepts:
        - Constitution: Set of principles/rules the agent must follow
        - Self-Critique: Agent evaluates its own outputs against principles
        - Revision: Generate improved response that adheres to constitution
        - Iterative Alignment: Multiple rounds of critique and revision
        - Transparency: Explicit principles make behavior interpretable
    
    Constitutional Principles Examples:
        - Harmlessness: Avoid harmful, dangerous, or offensive content
        - Helpfulness: Provide useful, relevant information
        - Honesty: Be truthful, acknowledge uncertainty
        - Privacy: Respect privacy and confidentiality
        - Fairness: Avoid bias and discrimination
    
    Use Cases:
        - Production AI systems requiring safety
        - Content moderation and filtering
        - Ethical decision-making systems
        - Regulated domains (healthcare, finance, legal)
    
    LangChain Implementation:
        Constitutional chain with critique and revision loops based on principles.
    """
    

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


    Import Statements

    Standard Library Imports

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

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

    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 traceback
    

    Purpose: Additional dependencies for specialized functionality.


    Classes

    Class: Principle

    Lines 48-53

    Purpose:

    Represents a constitutional principle.
    

    Class: ViolationReport

    Lines 57-62

    Purpose:

    Report of constitutional violations.
    

    Class: ConstitutionalReview

    Lines 66-76

    Purpose:

    Complete review against constitution.
    

    Methods:

    calculate_severity(self)

    Lines 73-76

    Purpose:

    Calculate total severity of violations.
    

    Parameters:


    Class: Constitution

    Lines 79-145

    Purpose:

    Set of principles for constitutional AI.
    

    Methods:

    __init__(self, name)

    Lines 82-85

    Purpose:

    Initialize constitution with default or custom principles.
    

    Parameters: – name: Name

    add_principle(self, principle)

    Lines 87-89

    Purpose:

    Add a principle to the constitution.
    

    Parameters: – principle: Principle

    get_principles_text(self)

    Lines 91-96

    Purpose:

    Get formatted text of all principles.
    

    Parameters:

    Returns: str

    get_default()

    Lines 99-145

    Decorators: staticmethod

    Purpose:

    Get default constitution with common principles.
    

    Returns: Constitution


    Class: ConstitutionalAI

    Lines 148-432

    Purpose:

    Agent that follows constitutional principles.
    

    Methods:

    __init__(self, constitution, model_name, max_revisions, severity_threshold)

    Lines 151-178

    Purpose:

    Initialize Constitutional AI agent.
    
    Args:
        constitution: The constitution to follow
        model_name: Name of the OpenAI model
        max_revisions: Maximum number of revision attempts
        severity_threshold: Severity threshold for triggering revision
    

    Parameters: – constitution: Constitution – model_name: Language model instance – max_revisions: Max revisions – severity_threshold: Severity threshold

    generate_initial(self, query, context)

    Lines 180-206

    Purpose:

    Generate initial response to query.
    
    Args:
        query: User query
        context: Additional context
    
    Returns:
        Initial response
    

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

    Returns: str

    critique_against_principle(self, output, query, principle)

    Lines 208-271

    Purpose:

    Critique output against a single principle.
    
    Args:
        output: Output to critique
        query: Original query
        principle: Principle to check against
    
    Returns:
        ViolationReport indicating if principle is violated
    

    Parameters: – output: Output – query: Input query or question – principle: Principle

    Returns: ViolationReport

    review(self, output, query)

    Lines 273-292

    Purpose:

    Review output against all constitutional principles.
    
    Args:
        output: Output to review
        query: Original query
    
    Returns:
        ConstitutionalReview with all violations
    

    Parameters: – output: Output – query: Input query or question

    Returns: ConstitutionalReview

    revise(self, query, output, review)

    Lines 294-346

    Purpose:

    Revise output to address constitutional violations.
    
    Args:
        query: Original query
        output: Current output with violations
        review: Review identifying violations
    
    Returns:
        Revised output addressing violations
    

    Parameters: – query: Input query or question – output: Output – review: Review

    Returns: str

    generate_with_constitution(self, query, context, verbose)

    Lines 348-432

    Purpose:

    Generate response following constitutional principles.
    
    Args:
        query: User query
        context: Additional context
        verbose: Whether to print progress
    
    Returns:
        Dictionary with final response and review history
    

    Parameters: – query: Input query or question – context: Context – verbose: Enable detailed logging

    Returns: <ast.Subscript object at 0x000001A06D607710>


    Functions

    demonstrate_constitutional_ai()

    Lines 435-517

    Purpose:

    Demonstrates the Constitutional AI pattern.
    

    Key Concepts Summary

    Pattern Overview

    The Constitutional Ai 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 009
    • Pattern 011
    • Other Reasoning Patterns (001-011)

    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 Constitutional Ai 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: 010_constitutional_ai.py
    Total Lines: 528
    Pattern Category: AI Agent Design Pattern #010


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Self Refine

    
        """
    Pattern 009: Self-Refine
    
    Description:
        Iteratively improves outputs through self-critique and refinement.
        The agent generates an initial output, evaluates it, identifies weaknesses,
        and produces an improved version. This cycle repeats until quality threshold
        is met or max iterations reached.
    
    Key Concepts:
        - Initial Generation: Create first draft output
        - Self-Critique: Analyze weaknesses and areas for improvement
        - Refinement: Generate improved version based on critique
        - Iteration: Repeat until satisfactory or max iterations
        - Quality Tracking: Monitor improvement across iterations
    
    Benefits:
        - Higher quality outputs through iteration
        - Self-improvement without external feedback
        - Transparency in refinement process
        - Handles complex generation tasks
    
    Use Cases:
        - Content creation (writing, code, reports)
        - Problem-solving with iterative improvement
        - Design tasks requiring multiple revisions
        - Any task benefiting from refinement
    
    LangChain Implementation:
        Chained prompts for generation, critique, and refinement with iteration control.
    """
    
    import os
    from typing import List, Dict, Any, Optional
    from dataclasses import dataclass, field
    from enum import Enum
    from dotenv import load_dotenv
    
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser
    from langchain_openai import ChatOpenAI
    
    # Load environment variables
    load_dotenv()
    
    
    class IterationStatus(Enum):
        """Status of refinement iteration."""
        INITIAL = "initial"
        REFINING = "refining"
        COMPLETED = "completed"
        MAX_ITERATIONS = "max_iterations_reached"
    
    
    @dataclass
    class Critique:
        """Represents a critique of generated output."""
        strengths: List[str] = field(default_factory=list)
        weaknesses: List[str] = field(default_factory=list)
        suggestions: List[str] = field(default_factory=list)
        score: float = 0.0  # Quality score 0-1
    
    
    @dataclass
    class RefinementIteration:
        """Represents one iteration of the refinement process."""
        iteration: int
        output: str
        critique: Optional[Critique] = None
        status: IterationStatus = IterationStatus.INITIAL
        
        def __str__(self):
            return f"Iteration {self.iteration}: {self.status.value} (score: {self.critique.score if self.critique else 'N/A'})"
    
    
    class SelfRefineAgent:
        """Agent that uses self-refinement to improve outputs."""
        
        def __init__(self, model_name: str = "gpt-3.5-turbo", 
                     max_iterations: int = 3,
                     quality_threshold: float = 0.8):
            """
            Initialize the Self-Refine agent.
            
            Args:
                model_name: Name of the OpenAI model
                max_iterations: Maximum refinement iterations
                quality_threshold: Stop if quality score exceeds this
            """
            self.llm = ChatOpenAI(
                model=model_name,
                temperature=0.7,
                api_key=os.getenv("OPENAI_API_KEY")
            )
            
            self.critic_llm = ChatOpenAI(
                model=model_name,
                temperature=0.3,  # Lower temperature for more consistent critique
                api_key=os.getenv("OPENAI_API_KEY")
            )
            
            self.max_iterations = max_iterations
            self.quality_threshold = quality_threshold
            self.history: List[RefinementIteration] = []
        
        def generate_initial(self, task: str, context: str = "") -> str:
            """
            Generate initial output for the task.
            
            Args:
                task: The task description
                context: Additional context
                
            Returns:
                Initial generated output
            """
            prompt = ChatPromptTemplate.from_messages([
                ("system", """Generate a response to the given task.
    Focus on creating a solid first draft. It doesn't need to be perfect yet."""),
                ("human", """{context}
    
    Task: {task}
    
    Response:""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            output = chain.invoke({
                "task": task,
                "context": context if context else "No additional context."
            })
            
            return output.strip()
        
        def critique(self, task: str, output: str, iteration: int) -> Critique:
            """
            Generate critique of current output.
            
            Args:
                task: Original task
                output: Current output to critique
                iteration: Current iteration number
                
            Returns:
                Critique object with feedback
            """
            prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a constructive critic. Evaluate the response to the task.
    
    Provide:
    1. Strengths: What works well (2-3 points)
    2. Weaknesses: What could be improved (2-3 points)
    3. Suggestions: Specific improvements to make (2-3 points)
    4. Score: Overall quality score from 0.0 to 1.0
    
    Format your response as:
    STRENGTHS:
    - [strength 1]
    - [strength 2]
    
    WEAKNESSES:
    - [weakness 1]
    - [weakness 2]
    
    SUGGESTIONS:
    - [suggestion 1]
    - [suggestion 2]
    
    SCORE: [0.0-1.0]"""),
                ("human", """Task: {task}
    
    Current Response (Iteration {iteration}):
    {output}
    
    Critique:""")
            ])
            
            chain = prompt | self.critic_llm | StrOutputParser()
            critique_text = chain.invoke({
                "task": task,
                "output": output,
                "iteration": iteration
            })
            
            # Parse critique
            critique = Critique()
            
            sections = {
                "strengths": [],
                "weaknesses": [],
                "suggestions": []
            }
            
            current_section = None
            for line in critique_text.split('\n'):
                line = line.strip()
                
                if line.upper().startswith('STRENGTHS:'):
                    current_section = "strengths"
                elif line.upper().startswith('WEAKNESSES:'):
                    current_section = "weaknesses"
                elif line.upper().startswith('SUGGESTIONS:'):
                    current_section = "suggestions"
                elif line.upper().startswith('SCORE:'):
                    try:
                        score_str = line.split(':', 1)[1].strip()
                        critique.score = float(score_str)
                    except:
                        critique.score = 0.5
                elif line.startswith('-') and current_section:
                    sections[current_section].append(line.lstrip('- ').strip())
            
            critique.strengths = sections["strengths"]
            critique.weaknesses = sections["weaknesses"]
            critique.suggestions = sections["suggestions"]
            
            # Ensure score is set
            if critique.score == 0.0 and (critique.strengths or not critique.weaknesses):
                critique.score = 0.6  # Default reasonable score
            
            return critique
        
        def refine(self, task: str, current_output: str, critique: Critique) -> str:
            """
            Generate refined version based on critique.
            
            Args:
                task: Original task
                current_output: Current output to refine
                critique: Critique with improvement suggestions
                
            Returns:
                Refined output
            """
            weaknesses_text = "\n".join([f"- {w}" for w in critique.weaknesses])
            suggestions_text = "\n".join([f"- {s}" for s in critique.suggestions])
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", """Improve the response based on the critique provided.
    Address the weaknesses and implement the suggestions while maintaining the strengths.
    Produce a refined, higher-quality version."""),
                ("human", """Task: {task}
    
    Current Response:
    {current_output}
    
    Weaknesses to Address:
    {weaknesses}
    
    Suggestions to Implement:
    {suggestions}
    
    Refined Response:""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            refined = chain.invoke({
                "task": task,
                "current_output": current_output,
                "weaknesses": weaknesses_text,
                "suggestions": suggestions_text
            })
            
            return refined.strip()
        
        def refine_iteratively(self, task: str, context: str = "",
                              verbose: bool = True) -> Dict[str, Any]:
            """
            Perform iterative refinement on the task.
            
            Args:
                task: The task to complete
                context: Additional context
                verbose: Whether to print progress
                
            Returns:
                Dictionary with final output and refinement history
            """
            if verbose:
                print(f"\nTask: {task}\n")
                print("="*60)
                print("ITERATION 1: Initial Generation")
                print("="*60)
            
            # Generate initial output
            initial_output = self.generate_initial(task, context)
            
            iteration = RefinementIteration(
                iteration=1,
                output=initial_output,
                status=IterationStatus.INITIAL
            )
            self.history.append(iteration)
            
            if verbose:
                print(f"\nGenerated:\n{initial_output[:200]}..." if len(initial_output) > 200 else f"\nGenerated:\n{initial_output}")
            
            current_output = initial_output
            
            # Refinement loop
            for i in range(2, self.max_iterations + 2):
                if verbose:
                    print(f"\n{'='*60}")
                    print(f"ITERATION {i}: Critique & Refine")
                    print("="*60)
                
                # Critique current output
                critique = self.critique(task, current_output, i - 1)
                
                # Update previous iteration with critique
                self.history[-1].critique = critique
                
                if verbose:
                    print(f"\nCritique:")
                    print(f"  Quality Score: {critique.score:.2f}")
                    if critique.strengths:
                        print(f"  Strengths: {len(critique.strengths)} identified")
                    if critique.weaknesses:
                        print(f"  Weaknesses: {len(critique.weaknesses)} identified")
                    if critique.suggestions:
                        print(f"  Suggestions: {len(critique.suggestions)} provided")
                
                # Check if quality threshold met
                if critique.score >= self.quality_threshold:
                    self.history[-1].status = IterationStatus.COMPLETED
                    if verbose:
                        print(f"\n✓ Quality threshold ({self.quality_threshold}) met!")
                    break
                
                # Check max iterations
                if i > self.max_iterations:
                    self.history[-1].status = IterationStatus.MAX_ITERATIONS
                    if verbose:
                        print(f"\n⚠ Max iterations ({self.max_iterations}) reached")
                    break
                
                # Refine output
                if verbose:
                    print(f"\nRefining...")
                
                refined_output = self.refine(task, current_output, critique)
                
                iteration = RefinementIteration(
                    iteration=i,
                    output=refined_output,
                    status=IterationStatus.REFINING
                )
                self.history.append(iteration)
                
                if verbose:
                    print(f"\nRefined:\n{refined_output[:200]}..." if len(refined_output) > 200 else f"\nRefined:\n{refined_output}")
                
                current_output = refined_output
            
            # Final critique
            if self.history[-1].critique is None:
                final_critique = self.critique(task, current_output, len(self.history))
                self.history[-1].critique = final_critique
            
            return {
                "task": task,
                "final_output": current_output,
                "iterations": len(self.history),
                "final_score": self.history[-1].critique.score if self.history[-1].critique else 0.0,
                "history": self.history,
                "status": self.history[-1].status.value
            }
        
        def get_improvement_summary(self) -> str:
            """Generate summary of improvements across iterations."""
            if len(self.history) < 2:
                return "Not enough iterations to compare."
            
            initial_score = self.history[0].critique.score if self.history[0].critique else 0.0
            final_score = self.history[-1].critique.score if self.history[-1].critique else 0.0
            
            improvement = final_score - initial_score
            improvement_pct = (improvement / max(initial_score, 0.1)) * 100
            
            summary = f"""
    Improvement Summary:
      Initial Score: {initial_score:.2f}
      Final Score: {final_score:.2f}
      Improvement: {improvement:+.2f} ({improvement_pct:+.1f}%)
      Total Iterations: {len(self.history)}
    """
            return summary
    
    
    def demonstrate_self_refine():
        """Demonstrates the Self-Refine pattern."""
        
        print("=" * 80)
        print("PATTERN 009: Self-Refine")
        print("=" * 80)
        print()
        print("Self-Refine iteratively improves outputs through:")
        print("1. Initial Generation: Create first draft")
        print("2. Self-Critique: Analyze strengths and weaknesses")
        print("3. Refinement: Generate improved version")
        print("4. Iteration: Repeat until quality threshold met")
        print()
        
        # Create agent
        agent = SelfRefineAgent(max_iterations=3, quality_threshold=0.85)
        
        # Test tasks
        tasks = [
            {
                "task": "Write a brief explanation of quantum entanglement for a general audience",
                "context": "Make it accessible but accurate, around 100 words"
            },
            {
                "task": "Create a function to calculate the Fibonacci sequence efficiently",
                "context": "Use Python, include docstring and handle edge cases"
            }
        ]
        
        for idx, test_case in enumerate(tasks, 1):
            print(f"\n{'='*80}")
            print(f"Example {idx}")
            print('='*80)
            
            try:
                # Reset history
                agent.history = []
                
                result = agent.refine_iteratively(
                    task=test_case['task'],
                    context=test_case['context'],
                    verbose=True
                )
                
                print(f"\n\n{'='*80}")
                print("FINAL RESULT")
                print('='*80)
                print(f"\nFinal Output:\n{result['final_output']}")
                print(f"\n{agent.get_improvement_summary()}")
                print(f"Status: {result['status']}")
                
            except Exception as e:
                print(f"\n✗ Error: {str(e)}")
                import traceback
                traceback.print_exc()
        
        # Summary
        print("\n\n" + "=" * 80)
        print("SELF-REFINE PATTERN DEMONSTRATION COMPLETE")
        print("=" * 80)
        print()
        print("Key Features Demonstrated:")
        print("1. Iterative Improvement: Multiple refinement cycles")
        print("2. Self-Critique: Autonomous evaluation of outputs")
        print("3. Targeted Refinement: Addressing specific weaknesses")
        print("4. Quality Tracking: Score-based improvement monitoring")
        print("5. Convergence: Stops when threshold met or max iterations")
        print()
        print("Advantages:")
        print("- Higher quality outputs through iteration")
        print("- No external feedback required")
        print("- Transparent improvement process")
        print("- Handles complex generation tasks")
        print()
        print("When to use Self-Refine:")
        print("- Content creation (writing, code, reports)")
        print("- Tasks benefiting from multiple revisions")
        print("- When quality is more important than speed")
        print("- Complex generation requiring refinement")
        print()
        print("LangChain Components Used:")
        print("- ChatPromptTemplate: Generation, critique, and refinement prompts")
        print("- StrOutputParser: Parse LLM outputs")
        print("- Separate LLMs: Different temperatures for generation vs critique")
        print("- Iteration control: Quality threshold and max iterations")
        print()
    
    
    if __name__ == "__main__":
        # Check for API key
        if not os.getenv("OPENAI_API_KEY"):
            print("Error: OPENAI_API_KEY not found in environment variables.")
            print("Please set it in your .env file or environment.")
            exit(1)
        
        demonstrate_self_refine()
    
        

    Agentic Design Pattern: Self Refine

    Pattern 009: Self Refine – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Self Refine pattern implementation.


    Pattern Description

    Pattern 009: Self-Refine

    Description: Iteratively improves outputs through self-critique and refinement. The agent generates an initial output, evaluates it, identifies weaknesses, and produces an improved version. This cycle repeats until quality threshold is met or max iterations reached.

    Key Concepts: – Initial Generation: Create first draft output – Self-Critique: Analyze weaknesses and areas for improvement – Refinement: Generate improved version based on critique – Iteration: Repeat until satisfactory or max iterations – Quality Tracking: Monitor improvement across iterations

    Benefits: – Higher quality outputs through iteration – Self-improvement without external feedback – Transparency in refinement process – Handles complex generation tasks

    Use Cases: – Content creation (writing, code, reports) – Problem-solving with iterative improvement – Design tasks requiring multiple revisions – Any task benefiting from refinement

    LangChain Implementation: Chained prompts for generation, critique, and refinement with iteration control.


    File Statistics

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

    Module Documentation

    """
    Pattern 009: Self-Refine
    
    Description:
        Iteratively improves outputs through self-critique and refinement.
        The agent generates an initial output, evaluates it, identifies weaknesses,
        and produces an improved version. This cycle repeats until quality threshold
        is met or max iterations reached.
    
    Key Concepts:
        - Initial Generation: Create first draft output
        - Self-Critique: Analyze weaknesses and areas for improvement
        - Refinement: Generate improved version based on critique
        - Iteration: Repeat until satisfactory or max iterations
        - Quality Tracking: Monitor improvement across iterations
    
    Benefits:
        - Higher quality outputs through iteration
        - Self-improvement without external feedback
        - Transparency in refinement process
        - Handles complex generation tasks
    
    Use Cases:
        - Content creation (writing, code, reports)
        - Problem-solving with iterative improvement
        - Design tasks requiring multiple revisions
        - Any task benefiting from refinement
    
    LangChain Implementation:
        Chained prompts for generation, critique, and refinement with iteration control.
    """
    

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


    Import Statements

    Standard Library Imports

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

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

    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 traceback
    

    Purpose: Additional dependencies for specialized functionality.


    Classes

    Class: IterationStatus

    Lines 47-52

    Inherits from: Enum

    Purpose:

    Status of refinement iteration.
    

    Class: Critique

    Lines 56-61

    Purpose:

    Represents a critique of generated output.
    

    Class: RefinementIteration

    Lines 65-73

    Purpose:

    Represents one iteration of the refinement process.
    

    Methods:

    __str__(self)

    Lines 72-73

    Purpose: Str operation.

    Parameters:


    Class: SelfRefineAgent

    Lines 76-387

    Purpose:

    Agent that uses self-refinement to improve outputs.
    

    Methods:

    __init__(self, model_name, max_iterations, quality_threshold)

    Lines 79-104

    Purpose:

    Initialize the Self-Refine agent.
    
    Args:
        model_name: Name of the OpenAI model
        max_iterations: Maximum refinement iterations
        quality_threshold: Stop if quality score exceeds this
    

    Parameters: – model_name: Language model instance – max_iterations: Max iterations – quality_threshold: Quality threshold

    generate_initial(self, task, context)

    Lines 106-133

    Purpose:

    Generate initial output for the task.
    
    Args:
        task: The task description
        context: Additional context
    
    Returns:
        Initial generated output
    

    Parameters: – task: Task – context: Context

    Returns: str

    critique(self, task, output, iteration)

    Lines 135-221

    Purpose:

    Generate critique of current output.
    
    Args:
        task: Original task
        output: Current output to critique
        iteration: Current iteration number
    
    Returns:
        Critique object with feedback
    

    Parameters: – task: Task – output: Output – iteration: Iteration

    Returns: Critique

    refine(self, task, current_output, critique)

    Lines 223-264

    Purpose:

    Generate refined version based on critique.
    
    Args:
        task: Original task
        current_output: Current output to refine
        critique: Critique with improvement suggestions
    
    Returns:
        Refined output
    

    Parameters: – task: Task – current_output: Current output – critique: Critique

    Returns: str

    refine_iteratively(self, task, context, verbose)

    Lines 266-367

    Purpose:

    Perform iterative refinement on the task.
    
    Args:
        task: The task to complete
        context: Additional context
        verbose: Whether to print progress
    
    Returns:
        Dictionary with final output and refinement history
    

    Parameters: – task: Task – context: Context – verbose: Enable detailed logging

    Returns: <ast.Subscript object at 0x000001A06D658D90>

    get_improvement_summary(self)

    Lines 369-387

    Purpose:

    Generate summary of improvements across iterations.
    

    Parameters:

    Returns: str


    Functions

    demonstrate_self_refine()

    Lines 390-475

    Purpose:

    Demonstrates the Self-Refine pattern.
    

    Key Concepts Summary

    Pattern Overview

    The Self Refine 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 008
    • Pattern 010
    • Other Reasoning Patterns (001-011)

    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 Self Refine 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: 009_self_refine.py
    Total Lines: 486
    Pattern Category: AI Agent Design Pattern #009


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Llm Compiler

    
        """
    Pattern 008: LLM Compiler
    
    Description:
        Optimizes LLM execution by analyzing task dependencies, parallelizing
        independent operations, and optimizing the execution plan like a compiler
        optimizes code. Combines planning with execution optimization.
    
    Key Concepts:
        - Task Decomposition: Break complex tasks into subtasks
        - Dependency Analysis: Build dependency graph (DAG)
        - Optimization: Identify parallelizable operations
        - Scheduling: Optimal execution order
        - Resource Management: Efficient LLM call allocation
    
    Compiler Phases:
        1. Parsing: Understand and decompose the task
        2. Analysis: Build dependency graph
        3. Optimization: Find parallelization opportunities
        4. Code Generation: Create execution plan
        5. Execution: Run optimized plan
    
    Use Cases:
        - Complex multi-step workflows
        - Research tasks with multiple branches
        - Data processing pipelines
        - Tasks with many independent subtasks
    
    LangChain Implementation:
        Uses dependency graphs and parallel execution with LangChain tools.
    """
    
    import os
    from typing import List, Dict, Any, Set, Optional
    from dataclasses import dataclass, field
    from collections import defaultdict, deque
    from concurrent.futures import ThreadPoolExecutor, as_completed
    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 environment variables
    load_dotenv()
    
    
    class TaskStatus(Enum):
        """Status of a task in the execution plan."""
        PENDING = "pending"
        READY = "ready"
        RUNNING = "running"
        COMPLETED = "completed"
        FAILED = "failed"
    
    
    @dataclass
    class Task:
        """Represents a single task in the execution plan."""
        id: str
        description: str
        dependencies: Set[str] = field(default_factory=set)
        status: TaskStatus = TaskStatus.PENDING
        result: Optional[str] = None
        error: Optional[str] = None
        parallel_group: Optional[int] = None  # For grouping parallel tasks
        
        def can_execute(self, completed_tasks: Set[str]) -> bool:
            """Check if this task can be executed."""
            return (self.status == TaskStatus.PENDING and 
                    self.dependencies.issubset(completed_tasks))
    
    
    @dataclass
    class ExecutionPlan:
        """Represents the optimized execution plan."""
        goal: str
        tasks: Dict[str, Task] = field(default_factory=dict)
        dependency_graph: Dict[str, Set[str]] = field(default_factory=lambda: defaultdict(set))
        parallel_groups: List[List[str]] = field(default_factory=list)
        
        def add_task(self, task: Task):
            """Add a task to the plan."""
            self.tasks[task.id] = task
            for dep in task.dependencies:
                self.dependency_graph[dep].add(task.id)
        
        def get_ready_tasks(self, completed: Set[str]) -> List[Task]:
            """Get all tasks that are ready to execute."""
            ready = []
            for task in self.tasks.values():
                if task.can_execute(completed):
                    ready.append(task)
            return ready
        
        def topological_sort(self) -> List[str]:
            """Perform topological sort on tasks."""
            in_degree = {task_id: len(task.dependencies) 
                         for task_id, task in self.tasks.items()}
            
            queue = deque([task_id for task_id, degree in in_degree.items() if degree == 0])
            sorted_tasks = []
            
            while queue:
                task_id = queue.popleft()
                sorted_tasks.append(task_id)
                
                for dependent in self.dependency_graph[task_id]:
                    in_degree[dependent] -= 1
                    if in_degree[dependent] == 0:
                        queue.append(dependent)
            
            return sorted_tasks
    
    
    class LLMCompiler:
        """Compiler that optimizes LLM task execution."""
        
        def __init__(self, model_name: str = "gpt-3.5-turbo", max_parallel: int = 3):
            """
            Initialize the LLM Compiler.
            
            Args:
                model_name: Name of the OpenAI model
                max_parallel: Maximum number of parallel tasks
            """
            self.llm = ChatOpenAI(
                model=model_name,
                temperature=0.3,
                api_key=os.getenv("OPENAI_API_KEY")
            )
            self.max_parallel = max_parallel
        
        def parse_task(self, goal: str) -> ExecutionPlan:
            """
            Parse the goal and decompose into subtasks.
            
            Args:
                goal: The high-level goal
                
            Returns:
                Initial execution plan with tasks
            """
            prompt = ChatPromptTemplate.from_messages([
                ("system", """Decompose the goal into specific, actionable subtasks.
    For each subtask:
    1. Give it a unique ID (T1, T2, T3, etc.)
    2. Describe what needs to be done
    3. List dependencies (which tasks must complete first)
    
    Format:
    T1: [description] | Dependencies: none
    T2: [description] | Dependencies: T1
    T3: [description] | Dependencies: T1
    T4: [description] | Dependencies: T2, T3
    ...
    
    Be specific and ensure dependencies are correct."""),
                ("human", "Goal: {goal}\n\nTask Decomposition:")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            response = chain.invoke({"goal": goal})
            
            # Parse tasks
            plan = ExecutionPlan(goal=goal)
            
            for line in response.strip().split('\n'):
                line = line.strip()
                if not line or ':' not in line:
                    continue
                
                # Parse: T1: description | Dependencies: T2, T3
                parts = line.split('|')
                if len(parts)  ExecutionPlan:
            """
            Analyze dependencies and optimize execution order.
            
            Args:
                plan: Initial execution plan
                
            Returns:
                Plan with optimized execution order
            """
            # Topological sort to get valid execution order
            sorted_tasks = plan.topological_sort()
            
            # Identify parallel groups (tasks at same level)
            completed = set()
            parallel_groups = []
            
            while completed != set(plan.tasks.keys()):
                # Get all tasks that can execute now
                ready_tasks = []
                for task_id in sorted_tasks:
                    if task_id not in completed:
                        task = plan.tasks[task_id]
                        if task.can_execute(completed):
                            ready_tasks.append(task_id)
                
                if not ready_tasks:
                    break  # Circular dependency or error
                
                parallel_groups.append(ready_tasks)
                completed.update(ready_tasks)
                
                # Assign parallel group numbers
                for task_id in ready_tasks:
                    plan.tasks[task_id].parallel_group = len(parallel_groups) - 1
            
            plan.parallel_groups = parallel_groups
            
            return plan
        
        def optimize_plan(self, plan: ExecutionPlan) -> ExecutionPlan:
            """
            Apply optimizations to the execution plan.
            
            Args:
                plan: Analyzed execution plan
                
            Returns:
                Optimized plan
            """
            # Optimization strategies:
            # 1. Limit parallel groups to max_parallel
            # 2. Try to balance load across groups
            # 3. Merge small groups if beneficial
            
            optimized_groups = []
            for group in plan.parallel_groups:
                if len(group)  str:
            """
            Execute a single task.
            
            Args:
                task: The task to execute
                plan: The complete plan for context
                
            Returns:
                Task result
            """
            # Build context from completed dependencies
            context_parts = []
            for dep_id in task.dependencies:
                dep_task = plan.tasks.get(dep_id)
                if dep_task and dep_task.result:
                    context_parts.append(f"{dep_id}: {dep_task.result}")
            
            context = "\n".join(context_parts) if context_parts else "No prior context."
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", "Execute the task using the provided context from previous steps."),
                ("human", """Goal: {goal}
    
    Prior Results:
    {context}
    
    Task: {task_description}
    
    Result:""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            result = chain.invoke({
                "goal": plan.goal,
                "context": context,
                "task_description": task.description
            })
            
            return result.strip()
        
        def execute_plan(self, plan: ExecutionPlan) -> Dict[str, Any]:
            """
            Execute the optimized plan.
            
            Args:
                plan: The execution plan
                
            Returns:
                Execution results
            """
            print(f"\nGoal: {plan.goal}\n")
            print(f"Total Tasks: {len(plan.tasks)}")
            print(f"Parallel Groups: {len(plan.parallel_groups)}\n")
            
            # Show execution plan
            print("Execution Plan:")
            for i, group in enumerate(plan.parallel_groups):
                group_tasks = [plan.tasks[tid].description[:50] for tid in group]
                print(f"  Group {i+1} ({len(group)} tasks in parallel):")
                for task_id in group:
                    deps = list(plan.tasks[task_id].dependencies)
                    dep_str = f" [depends on: {', '.join(deps)}]" if deps else ""
                    print(f"    - {task_id}: {plan.tasks[task_id].description[:60]}{dep_str}")
            print()
            
            completed = set()
            total_llm_calls = 0
            
            # Execute groups in sequence
            for group_idx, group in enumerate(plan.parallel_groups):
                print(f"{'='*60}")
                print(f"Executing Group {group_idx + 1} ({len(group)} tasks)")
                print('='*60)
                
                # Execute tasks in group in parallel
                with ThreadPoolExecutor(max_workers=min(len(group), self.max_parallel)) as executor:
                    futures = {}
                    for task_id in group:
                        task = plan.tasks[task_id]
                        task.status = TaskStatus.RUNNING
                        future = executor.submit(self.execute_task, task, plan)
                        futures[future] = task
                    
                    # Collect results
                    for future in as_completed(futures):
                        task = futures[future]
                        total_llm_calls += 1
                        
                        try:
                            result = future.result()
                            task.result = result
                            task.status = TaskStatus.COMPLETED
                            completed.add(task.id)
                            print(f"\n✓ {task.id} completed")
                            print(f"  Result: {result[:80]}...")
                        except Exception as e:
                            task.status = TaskStatus.FAILED
                            task.error = str(e)
                            print(f"\n✗ {task.id} failed: {str(e)}")
            
            # Generate final summary
            summary = self._synthesize_results(plan)
            
            return {
                "goal": plan.goal,
                "total_tasks": len(plan.tasks),
                "completed_tasks": len(completed),
                "parallel_groups": len(plan.parallel_groups),
                "total_llm_calls": total_llm_calls,
                "summary": summary
            }
        
        def _synthesize_results(self, plan: ExecutionPlan) -> str:
            """Synthesize final summary from all task results."""
            results = []
            for task in plan.tasks.values():
                if task.status == TaskStatus.COMPLETED and task.result:
                    results.append(f"{task.id}: {task.result}")
            
            if not results:
                return "No tasks completed successfully."
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", "Synthesize the results into a comprehensive summary."),
                ("human", """Goal: {goal}
    
    Task Results:
    {results}
    
    Summary:""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            
            try:
                summary = chain.invoke({
                    "goal": plan.goal,
                    "results": "\n\n".join(results)
                })
                return summary.strip()
            except:
                return "Summary generation failed."
        
        def compile_and_execute(self, goal: str) -> Dict[str, Any]:
            """
            Compile and execute a goal (full pipeline).
            
            Args:
                goal: The goal to achieve
                
            Returns:
                Execution results
            """
            print(f"\n{'='*80}")
            print("LLM COMPILER: PARSING")
            print('='*80)
            
            # Phase 1: Parse and decompose
            plan = self.parse_task(goal)
            print(f"\nParsed {len(plan.tasks)} tasks")
            
            print(f"\n{'='*80}")
            print("LLM COMPILER: ANALYSIS")
            print('='*80)
            
            # Phase 2: Analyze dependencies
            plan = self.analyze_dependencies(plan)
            print(f"\nIdentified {len(plan.parallel_groups)} execution groups")
            
            print(f"\n{'='*80}")
            print("LLM COMPILER: OPTIMIZATION")
            print('='*80)
            
            # Phase 3: Optimize
            plan = self.optimize_plan(plan)
            print(f"\nOptimized for {self.max_parallel} max parallel tasks")
            
            print(f"\n{'='*80}")
            print("LLM COMPILER: EXECUTION")
            print('='*80)
            
            # Phase 4: Execute
            result = self.execute_plan(plan)
            
            return result
    
    
    def demonstrate_llm_compiler():
        """Demonstrates the LLM Compiler pattern."""
        
        print("=" * 80)
        print("PATTERN 008: LLM Compiler")
        print("=" * 80)
        print()
        print("LLM Compiler optimizes execution like a code compiler:")
        print("1. Parsing: Decompose task into subtasks")
        print("2. Analysis: Build dependency graph")
        print("3. Optimization: Identify parallelization opportunities")
        print("4. Execution: Run optimized plan with parallel execution")
        print()
        
        # Create compiler
        compiler = LLMCompiler(max_parallel=3)
        
        # Test goals
        goals = [
            "Research the history of artificial intelligence and create a timeline of major milestones",
            "Analyze the advantages and disadvantages of renewable energy sources and compare them"
        ]
        
        for i, goal in enumerate(goals, 1):
            print(f"\n{'='*80}")
            print(f"Example {i}")
            print('='*80)
            
            try:
                result = compiler.compile_and_execute(goal)
                
                print(f"\n\n{'='*80}")
                print("COMPILATION SUMMARY")
                print('='*80)
                print(f"\nGoal: {result['goal']}")
                print(f"Total Tasks: {result['total_tasks']}")
                print(f"Completed: {result['completed_tasks']}/{result['total_tasks']}")
                print(f"Parallel Groups: {result['parallel_groups']}")
                print(f"Total LLM Calls: {result['total_llm_calls']}")
                print(f"\nFinal Summary:\n{result['summary']}")
                
            except Exception as e:
                print(f"\n✗ Error: {str(e)}")
                import traceback
                traceback.print_exc()
        
        # Summary
        print("\n\n" + "=" * 80)
        print("LLM COMPILER PATTERN DEMONSTRATION COMPLETE")
        print("=" * 80)
        print()
        print("Key Features Demonstrated:")
        print("1. Task Decomposition: Breaking complex goals into subtasks")
        print("2. Dependency Analysis: Building execution dependency graph")
        print("3. Parallelization: Executing independent tasks concurrently")
        print("4. Optimization: Grouping and scheduling for efficiency")
        print("5. Resource Management: Limiting parallel execution")
        print()
        print("Advantages:")
        print("- Reduced total execution time through parallelization")
        print("- Efficient resource utilization")
        print("- Clear dependency management")
        print("- Systematic optimization approach")
        print()
        print("When to use LLM Compiler:")
        print("- Complex multi-step workflows")
        print("- Tasks with many independent subtasks")
        print("- Time-sensitive applications")
        print("- Research and analysis tasks")
        print()
        print("LangChain Components Used:")
        print("- ChatPromptTemplate for task decomposition")
        print("- Dependency graph (DAG) management")
        print("- ThreadPoolExecutor for parallel execution")
        print("- Result synthesis chain")
        print()
    
    
    if __name__ == "__main__":
        # Check for API key
        if not os.getenv("OPENAI_API_KEY"):
            print("Error: OPENAI_API_KEY not found in environment variables.")
            print("Please set it in your .env file or environment.")
            exit(1)
        
        demonstrate_llm_compiler()
    
        

    Agentic Design Pattern: Llm Compiler

    Pattern 008: Llm Compiler – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Llm Compiler pattern implementation.


    Pattern Description

    Pattern 008: LLM Compiler

    Description: Optimizes LLM execution by analyzing task dependencies, parallelizing independent operations, and optimizing the execution plan like a compiler optimizes code. Combines planning with execution optimization.

    Key Concepts: – Task Decomposition: Break complex tasks into subtasks – Dependency Analysis: Build dependency graph (DAG) – Optimization: Identify parallelizable operations – Scheduling: Optimal execution order – Resource Management: Efficient LLM call allocation

    Compiler Phases: 1. Parsing: Understand and decompose the task 2. Analysis: Build dependency graph 3. Optimization: Find parallelization opportunities 4. Code Generation: Create execution plan 5. Execution: Run optimized plan

    Use Cases: – Complex multi-step workflows – Research tasks with multiple branches – Data processing pipelines – Tasks with many independent subtasks

    LangChain Implementation: Uses dependency graphs and parallel execution with LangChain tools.


    File Statistics

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

    Module Documentation

    """
    Pattern 008: LLM Compiler
    
    Description:
        Optimizes LLM execution by analyzing task dependencies, parallelizing
        independent operations, and optimizing the execution plan like a compiler
        optimizes code. Combines planning with execution optimization.
    
    Key Concepts:
        - Task Decomposition: Break complex tasks into subtasks
        - Dependency Analysis: Build dependency graph (DAG)
        - Optimization: Identify parallelizable operations
        - Scheduling: Optimal execution order
        - Resource Management: Efficient LLM call allocation
    
    Compiler Phases:
        1. Parsing: Understand and decompose the task
        2. Analysis: Build dependency graph
        3. Optimization: Find parallelization opportunities
        4. Code Generation: Create execution plan
        5. Execution: Run optimized plan
    
    Use Cases:
        - Complex multi-step workflows
        - Research tasks with multiple branches
        - Data processing pipelines
        - Tasks with many independent subtasks
    
    LangChain Implementation:
        Uses dependency graphs and parallel execution with LangChain tools.
    """
    

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


    Import Statements

    Standard Library Imports

    import collections.defaultdict
    import collections.deque
    import concurrent.futures.ThreadPoolExecutor
    import concurrent.futures.as_completed
    import dataclasses.dataclass
    import dataclasses.field
    import enum.Enum
    import langchain_core.output_parsers.StrOutputParser
    import langchain_core.prompts.ChatPromptTemplate
    import os
    import typing.Any
    import typing.Dict
    import typing.List
    import typing.Optional
    import typing.Set
    

    Purpose: Core Python functionality: – collections.defaultdict: Specialized container datatypes – collections.deque: Specialized container datatypes – concurrent.futures.ThreadPoolExecutor: Regular expression operations – concurrent.futures.as_completed: Regular expression operations – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – enum.Enum: Enumeration support – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional) – typing.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
    from traceback
    

    Purpose: Additional dependencies for specialized functionality.


    Classes

    Class: TaskStatus

    Lines 49-55

    Inherits from: Enum

    Purpose:

    Status of a task in the execution plan.
    

    Class: Task

    Lines 59-72

    Purpose:

    Represents a single task in the execution plan.
    

    Methods:

    can_execute(self, completed_tasks)

    Lines 69-72

    Purpose:

    Check if this task can be executed.
    

    Parameters: – completed_tasks: Completed tasks

    Returns: bool


    Class: ExecutionPlan

    Lines 76-114

    Purpose:

    Represents the optimized execution plan.
    

    Methods:

    add_task(self, task)

    Lines 83-87

    Purpose:

    Add a task to the plan.
    

    Parameters: – task: Task

    get_ready_tasks(self, completed)

    Lines 89-95

    Purpose:

    Get all tasks that are ready to execute.
    

    Parameters: – completed: Completed

    Returns: <ast.Subscript object at 0x000001A06D615F10>

    topological_sort(self)

    Lines 97-114

    Purpose:

    Perform topological sort on tasks.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D6AA390>


    Class: LLMCompiler

    Lines 117-464

    Purpose:

    Compiler that optimizes LLM task execution.
    

    Methods:

    __init__(self, model_name, max_parallel)

    Lines 120-133

    Purpose:

    Initialize the LLM Compiler.
    
    Args:
        model_name: Name of the OpenAI model
        max_parallel: Maximum number of parallel tasks
    

    Parameters: – model_name: Language model instance – max_parallel: Max parallel

    parse_task(self, goal)

    Lines 135-209

    Purpose:

    Parse the goal and decompose into subtasks.
    
    Args:
        goal: The high-level goal
    
    Returns:
        Initial execution plan with tasks
    

    Parameters: – goal: Goal

    Returns: ExecutionPlan

    analyze_dependencies(self, plan)

    Lines 211-249

    Purpose:

    Analyze dependencies and optimize execution order.
    
    Args:
        plan: Initial execution plan
    
    Returns:
        Plan with optimized execution order
    

    Parameters: – plan: Plan

    Returns: ExecutionPlan

    optimize_plan(self, plan)

    Lines 251-277

    Purpose:

    Apply optimizations to the execution plan.
    
    Args:
        plan: Analyzed execution plan
    
    Returns:
        Optimized plan
    

    Parameters: – plan: Plan

    Returns: ExecutionPlan

    execute_task(self, task, plan)

    Lines 279-318

    Purpose:

    Execute a single task.
    
    Args:
        task: The task to execute
        plan: The complete plan for context
    
    Returns:
        Task result
    

    Parameters: – task: Task – plan: Plan

    Returns: str

    execute_plan(self, plan)

    Lines 320-390

    Purpose:

    Execute the optimized plan.
    
    Args:
        plan: The execution plan
    
    Returns:
        Execution results
    

    Parameters: – plan: Plan

    Returns: <ast.Subscript object at 0x000001A06D6EB350>

    _synthesize_results(self, plan)

    Lines 392-421

    Purpose:

    Synthesize final summary from all task results.
    

    Parameters: – plan: Plan

    Returns: str

    compile_and_execute(self, goal)

    Lines 423-464

    Purpose:

    Compile and execute a goal (full pipeline).
    
    Args:
        goal: The goal to achieve
    
    Returns:
        Execution results
    

    Parameters: – goal: Goal

    Returns: <ast.Subscript object at 0x000001A06D628190>


    Functions

    demonstrate_llm_compiler()

    Lines 467-542

    Purpose:

    Demonstrates the LLM Compiler pattern.
    

    Key Concepts Summary

    Pattern Overview

    The Llm Compiler 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 007
    • Pattern 009
    • Other Reasoning Patterns (001-011)

    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 Llm Compiler 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: 008_llm_compiler.py
    Total Lines: 553
    Pattern Category: AI Agent Design Pattern #008


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Rewoo

    
        """
    Pattern 007: ReWOO (Reasoning Without Observation)
    
    Description:
        Decouples reasoning from observation by planning all tool uses upfront,
        then executing them in batch. Unlike ReAct which interleaves reasoning
        and observation, ReWOO plans all actions first, then executes them,
        reducing LLM calls and enabling parallel execution.
    
    Key Concepts:
        - Upfront Planning: Generate complete tool usage plan first
        - Dependency Analysis: Identify dependencies between tool calls
        - Batch Execution: Execute all tools (potentially in parallel)
        - Variable System: Use variables (e.g., #E1, #E2) for results
        - Final Synthesis: Generate answer using all tool results
    
    Benefits:
        - Fewer LLM calls (plan once, execute all)
        - Enables parallel tool execution
        - More efficient for multi-tool tasks
        - Clearer execution plan
    
    Use Cases:
        - Multi-tool research tasks
        - Information gathering from multiple sources
        - Tasks requiring coordinated tool use
        - Scenarios where parallelization is beneficial
    
    LangChain Implementation:
        Separate planning and execution chains with variable substitution.
    """
    
    import os
    import re
    from typing import List, Dict, Any, Optional, Callable
    from dataclasses import dataclass, field
    from concurrent.futures import ThreadPoolExecutor, as_completed
    from dotenv import load_dotenv
    
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser
    from langchain_core.tools import Tool
    from langchain_openai import ChatOpenAI
    
    # Load environment variables
    load_dotenv()
    
    
    @dataclass
    class ToolCall:
        """Represents a planned tool call."""
        variable: str  # e.g., "#E1"
        tool_name: str
        arguments: str
        dependencies: List[str] = field(default_factory=list)  # Variable dependencies
        result: Optional[str] = None
        executed: bool = False
    
    
    @dataclass
    class ExecutionPlan:
        """Represents the complete tool execution plan."""
        question: str
        tool_calls: List[ToolCall] = field(default_factory=list)
        final_reasoning: str = ""
        
        def get_executable_calls(self) -> List[ToolCall]:
            """Get tool calls that can be executed (dependencies met)."""
            executable = []
            for call in self.tool_calls:
                if not call.executed:
                    # Check if all dependencies are met
                    deps_met = all(
                        any(tc.variable == dep and tc.executed for tc in self.tool_calls)
                        for dep in call.dependencies
                    )
                    if not call.dependencies or deps_met:
                        executable.append(call)
            return executable
    
    
    class ReWOOAgent:
        """Agent that uses ReWOO (Reasoning Without Observation) pattern."""
        
        def __init__(self, tools: List[Tool], model_name: str = "gpt-3.5-turbo"):
            """
            Initialize the ReWOO agent.
            
            Args:
                tools: List of available tools
                model_name: Name of the OpenAI model
            """
            self.tools = {tool.name: tool for tool in tools}
            self.tool_descriptions = self._format_tool_descriptions()
            
            self.planner_llm = ChatOpenAI(
                model=model_name,
                temperature=0.0,  # Low temperature for consistent planning
                api_key=os.getenv("OPENAI_API_KEY")
            )
            
            self.solver_llm = ChatOpenAI(
                model=model_name,
                temperature=0.3,
                api_key=os.getenv("OPENAI_API_KEY")
            )
        
        def _format_tool_descriptions(self) -> str:
            """Format tool descriptions for the planner."""
            descriptions = []
            for name, tool in self.tools.items():
                descriptions.append(f"- {name}: {tool.description}")
            return "\n".join(descriptions)
        
        def plan(self, question: str) -> ExecutionPlan:
            """
            Create an execution plan for answering the question.
            
            Args:
                question: The question to answer
                
            Returns:
                ExecutionPlan with all tool calls planned
            """
            prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a planner that creates execution plans using available tools.
    
    Available tools:
    {tools}
    
    Create a plan to answer the question using these tools. For each step:
    1. Assign a variable (#E1, #E2, etc.)
    2. Specify the tool and its arguments
    3. Reference previous results using their variables
    
    Format each plan step as:
    #E1 = ToolName[arguments]
    #E2 = ToolName[arguments using #E1 if needed]
    ...
    
    Then provide reasoning for how to combine the results:
    Plan: [Brief description of how to use the results]
    
    Example:
    #E1 = Calculator[15 * 7]
    #E2 = Search[capital of #E1]
    Plan: Use #E1 to get the product, then search for information about #E2."""),
                ("human", "Question: {question}\n\nExecution Plan:")
            ])
            
            chain = prompt | self.planner_llm | StrOutputParser()
            response = chain.invoke({
                "question": question,
                "tools": self.tool_descriptions
            })
            
            # Parse the plan
            plan = ExecutionPlan(question=question)
            
            lines = response.strip().split('\n')
            for line in lines:
                line = line.strip()
                
                # Parse tool calls like: #E1 = ToolName[arguments]
                match = re.match(r'#(E\d+)\s*=\s*(\w+)\[(.*?)\]', line)
                if match:
                    variable = f"#{match.group(1)}"
                    tool_name = match.group(2)
                    arguments = match.group(3).strip()
                    
                    # Extract dependencies (variables referenced in arguments)
                    dependencies = re.findall(r'#E\d+', arguments)
                    
                    tool_call = ToolCall(
                        variable=variable,
                        tool_name=tool_name,
                        arguments=arguments,
                        dependencies=dependencies
                    )
                    plan.tool_calls.append(tool_call)
                
                # Extract final reasoning
                elif line.lower().startswith('plan:'):
                    plan.final_reasoning = line.split(':', 1)[1].strip()
            
            return plan
        
        def execute_plan(self, plan: ExecutionPlan, parallel: bool = False) -> Dict[str, Any]:
            """
            Execute the tool calls in the plan.
            
            Args:
                plan: The execution plan
                parallel: Whether to execute independent calls in parallel
                
            Returns:
                Dictionary with execution results
            """
            print(f"\nQuestion: {plan.question}\n")
            print("Execution Plan:")
            for call in plan.tool_calls:
                deps = f" (depends on: {', '.join(call.dependencies)})" if call.dependencies else ""
                print(f"  {call.variable} = {call.tool_name}[{call.arguments}]{deps}")
            print(f"\nFinal Plan: {plan.final_reasoning}\n")
            
            print("="*60)
            print("EXECUTING TOOLS")
            print("="*60)
            
            # Execute tool calls respecting dependencies
            while True:
                executable = plan.get_executable_calls()
                if not executable:
                    break
                
                if parallel and len(executable) > 1:
                    # Execute in parallel
                    with ThreadPoolExecutor(max_workers=len(executable)) as executor:
                        futures = {
                            executor.submit(self._execute_tool_call, call, plan): call 
                            for call in executable
                        }
                        
                        for future in as_completed(futures):
                            call = futures[future]
                            try:
                                result = future.result()
                                call.result = result
                                call.executed = True
                                print(f"\n✓ {call.variable} = {result[:100]}...")
                            except Exception as e:
                                call.result = f"Error: {str(e)}"
                                call.executed = True
                                print(f"\n✗ {call.variable} failed: {str(e)}")
                else:
                    # Execute sequentially
                    for call in executable:
                        try:
                            result = self._execute_tool_call(call, plan)
                            call.result = result
                            call.executed = True
                            print(f"\n✓ {call.variable} = {result[:100]}...")
                        except Exception as e:
                            call.result = f"Error: {str(e)}"
                            call.executed = True
                            print(f"\n✗ {call.variable} failed: {str(e)}")
            
            # Synthesize final answer
            print(f"\n{'='*60}")
            print("SYNTHESIZING ANSWER")
            print('='*60)
            
            answer = self._synthesize_answer(plan)
            
            return {
                "question": plan.question,
                "answer": answer,
                "tool_calls": len(plan.tool_calls),
                "plan": plan.final_reasoning
            }
        
        def _execute_tool_call(self, call: ToolCall, plan: ExecutionPlan) -> str:
            """Execute a single tool call."""
            # Substitute variable references with actual results
            arguments = call.arguments
            for dep_var in call.dependencies:
                # Find the result for this dependency
                for other_call in plan.tool_calls:
                    if other_call.variable == dep_var and other_call.result:
                        arguments = arguments.replace(dep_var, other_call.result)
            
            # Get the tool
            tool = self.tools.get(call.tool_name)
            if not tool:
                raise ValueError(f"Tool {call.tool_name} not found")
            
            # Execute the tool
            try:
                result = tool.func(arguments)
                return str(result)
            except Exception as e:
                raise Exception(f"Tool execution failed: {str(e)}")
        
        def _synthesize_answer(self, plan: ExecutionPlan) -> str:
            """Synthesize final answer from tool results."""
            # Build evidence from tool results
            evidence = []
            for call in plan.tool_calls:
                if call.executed and call.result:
                    evidence.append(f"{call.variable}: {call.result}")
            
            evidence_text = "\n".join(evidence)
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", """Based on the evidence gathered from tool execution, 
    provide a comprehensive answer to the original question.
    
    Use the evidence to construct a well-reasoned response."""),
                ("human", """Question: {question}
    
    Evidence:
    {evidence}
    
    Plan: {plan}
    
    Answer:""")
            ])
            
            chain = prompt | self.solver_llm | StrOutputParser()
            answer = chain.invoke({
                "question": plan.question,
                "evidence": evidence_text,
                "plan": plan.final_reasoning
            })
            
            return answer.strip()
        
        def solve(self, question: str, parallel: bool = False) -> Dict[str, Any]:
            """
            Solve a question using ReWOO pattern.
            
            Args:
                question: The question to answer
                parallel: Whether to use parallel execution
                
            Returns:
                Dictionary with answer and metadata
            """
            print(f"\n{'='*80}")
            print("ReWOO: PLANNING PHASE")
            print('='*80)
            
            # Create plan
            plan = self.plan(question)
            
            print(f"\n{'='*80}")
            print("ReWOO: EXECUTION PHASE")
            print('='*80)
            
            # Execute plan
            result = self.execute_plan(plan, parallel=parallel)
            
            return result
    
    
    # Example tools
    def calculator(expression: str) -> float:
        """Calculate a mathematical expression."""
        try:
            # Simple eval (in production, use a safe math parser)
            result = eval(expression)
            return result
        except Exception as e:
            return f"Calculation error: {str(e)}"
    
    
    def search(query: str) -> str:
        """Simulate a search tool."""
        # In production, this would call a real search API
        responses = {
            "capital of france": "Paris is the capital of France.",
            "population of tokyo": "Tokyo has a population of approximately 14 million people.",
            "speed of light": "The speed of light is approximately 299,792,458 meters per second.",
            "inventor of telephone": "Alexander Graham Bell is credited with inventing the telephone.",
        }
        
        query_lower = query.lower()
        for key, value in responses.items():
            if key in query_lower:
                return value
        
        return f"Search results for: {query}"
    
    
    def get_date(query: str) -> str:
        """Get current date information."""
        from datetime import datetime
        now = datetime.now()
        
        if "year" in query.lower():
            return str(now.year)
        elif "month" in query.lower():
            return now.strftime("%B")
        elif "day" in query.lower():
            return str(now.day)
        else:
            return now.strftime("%Y-%m-%d")
    
    
    def demonstrate_rewoo():
        """Demonstrates the ReWOO pattern."""
        
        print("=" * 80)
        print("PATTERN 007: ReWOO (Reasoning Without Observation)")
        print("=" * 80)
        print()
        print("ReWOO decouples reasoning from observation:")
        print("1. Planning Phase: Plan all tool uses upfront")
        print("2. Dependency Analysis: Identify dependencies between calls")
        print("3. Execution Phase: Execute tools (potentially in parallel)")
        print("4. Synthesis: Generate answer from all results")
        print()
        
        # Create tools
        tools = [
            Tool(
                name="Calculator",
                func=calculator,
                description="Calculate mathematical expressions. Input should be a valid expression."
            ),
            Tool(
                name="Search",
                func=search,
                description="Search for information. Input should be a search query."
            ),
            Tool(
                name="GetDate",
                func=get_date,
                description="Get current date information. Input can be 'year', 'month', 'day', or 'full'."
            )
        ]
        
        # Create agent
        agent = ReWOOAgent(tools=tools)
        
        # Test questions
        questions = [
            "What is 25 * 34, and what is the capital of France?",
            "Calculate 144 / 12, then search for information about that number",
        ]
        
        for i, question in enumerate(questions, 1):
            print(f"\n{'='*80}")
            print(f"Example {i}")
            print('='*80)
            
            try:
                result = agent.solve(question, parallel=False)
                
                print(f"\n\n{'='*80}")
                print("FINAL RESULT")
                print('='*80)
                print(f"\nQuestion: {result['question']}")
                print(f"\nAnswer: {result['answer']}")
                print(f"\nTool Calls: {result['tool_calls']}")
                
            except Exception as e:
                print(f"\n✗ Error: {str(e)}")
                import traceback
                traceback.print_exc()
        
        # Summary
        print("\n\n" + "=" * 80)
        print("ReWOO PATTERN DEMONSTRATION COMPLETE")
        print("=" * 80)
        print()
        print("Key Features Demonstrated:")
        print("1. Upfront Planning: All tool uses planned before execution")
        print("2. Variable System: Results referenced as #E1, #E2, etc.")
        print("3. Dependency Tracking: Execution order based on dependencies")
        print("4. Batch Execution: All tools executed in one phase")
        print("5. Result Synthesis: Final answer combines all evidence")
        print()
        print("Advantages over ReAct:")
        print("- Fewer LLM calls (plan once vs. after each observation)")
        print("- Enables parallel tool execution")
        print("- Clearer execution flow")
        print("- More efficient for multi-tool tasks")
        print()
        print("When to use ReWOO:")
        print("- Multi-tool information gathering")
        print("- Tasks with independent tool calls")
        print("- Scenarios benefiting from parallelization")
        print("- When you want predictable tool usage")
        print()
        print("LangChain Components Used:")
        print("- Tool abstraction for uniform tool interface")
        print("- Separate planning and execution chains")
        print("- Variable substitution system")
        print("- Optional parallel execution")
        print()
    
    
    if __name__ == "__main__":
        # Check for API key
        if not os.getenv("OPENAI_API_KEY"):
            print("Error: OPENAI_API_KEY not found in environment variables.")
            print("Please set it in your .env file or environment.")
            exit(1)
        
        demonstrate_rewoo()
    
        

    Agentic Design Pattern: Rewoo

    Pattern 007: Rewoo – Line-by-Line Explanation

    Overview

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


    Pattern Description

    Pattern 007: ReWOO (Reasoning Without Observation)

    Description: Decouples reasoning from observation by planning all tool uses upfront, then executing them in batch. Unlike ReAct which interleaves reasoning and observation, ReWOO plans all actions first, then executes them, reducing LLM calls and enabling parallel execution.

    Key Concepts: – Upfront Planning: Generate complete tool usage plan first – Dependency Analysis: Identify dependencies between tool calls – Batch Execution: Execute all tools (potentially in parallel) – Variable System: Use variables (e.g., #E1, #E2) for results – Final Synthesis: Generate answer using all tool results

    Benefits: – Fewer LLM calls (plan once, execute all) – Enables parallel tool execution – More efficient for multi-tool tasks – Clearer execution plan

    Use Cases: – Multi-tool research tasks – Information gathering from multiple sources – Tasks requiring coordinated tool use – Scenarios where parallelization is beneficial

    LangChain Implementation: Separate planning and execution chains with variable substitution.


    File Statistics

    • Total Lines: 492
    • Classes: 3
    • Functions: 4
    • Constants: 0

    Module Documentation

    """
    Pattern 007: ReWOO (Reasoning Without Observation)
    
    Description:
        Decouples reasoning from observation by planning all tool uses upfront,
        then executing them in batch. Unlike ReAct which interleaves reasoning
        and observation, ReWOO plans all actions first, then executes them,
        reducing LLM calls and enabling parallel execution.
    
    Key Concepts:
        - Upfront Planning: Generate complete tool usage plan first
        - Dependency Analysis: Identify dependencies between tool calls
        - Batch Execution: Execute all tools (potentially in parallel)
        - Variable System: Use variables (e.g., #E1, #E2) for results
        - Final Synthesis: Generate answer using all tool results
    
    Benefits:
        - Fewer LLM calls (plan once, execute all)
        - Enables parallel tool execution
        - More efficient for multi-tool tasks
        - Clearer execution plan
    
    Use Cases:
        - Multi-tool research tasks
        - Information gathering from multiple sources
        - Tasks requiring coordinated tool use
        - Scenarios where parallelization is beneficial
    
    LangChain Implementation:
        Separate planning and execution chains with variable substitution.
    """
    

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


    Import Statements

    Standard Library Imports

    import concurrent.futures.ThreadPoolExecutor
    import concurrent.futures.as_completed
    import dataclasses.dataclass
    import dataclasses.field
    import datetime.datetime
    import langchain_core.output_parsers.StrOutputParser
    import langchain_core.prompts.ChatPromptTemplate
    import langchain_core.tools.Tool
    import os
    import re
    import typing.Any
    import typing.Callable
    import typing.Dict
    import typing.List
    import typing.Optional
    

    Purpose: Core Python functionality: – concurrent.futures.ThreadPoolExecutor: Regular expression operations – concurrent.futures.as_completed: Regular expression operations – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – datetime.datetime: Date and time handling – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – langchain_core.tools.Tool: Regular expression operations – os: Environment variable access and system operations – re: Regular expression operations – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Callable: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional)

    LangChain Imports

    from langchain_openai.ChatOpenAI
    

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

    Other Imports

    from dotenv.load_dotenv
    from traceback
    

    Purpose: Additional dependencies for specialized functionality.


    Classes

    Class: ToolCall

    Lines 50-57

    Purpose:

    Represents a planned tool call.
    

    Class: ExecutionPlan

    Lines 61-79

    Purpose:

    Represents the complete tool execution plan.
    

    Methods:

    get_executable_calls(self)

    Lines 67-79

    Purpose:

    Get tool calls that can be executed (dependencies met).
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D6761D0>


    Class: ReWOOAgent

    Lines 82-343

    Purpose:

    Agent that uses ReWOO (Reasoning Without Observation) pattern.
    

    Methods:

    __init__(self, tools, model_name)

    Lines 85-106

    Purpose:

    Initialize the ReWOO agent.
    
    Args:
        tools: List of available tools
        model_name: Name of the OpenAI model
    

    Parameters: – tools: Available tools for the agent – model_name: Language model instance

    _format_tool_descriptions(self)

    Lines 108-113

    Purpose:

    Format tool descriptions for the planner.
    

    Parameters:

    Returns: str

    plan(self, question)

    Lines 115-186

    Purpose:

    Create an execution plan for answering the question.
    
    Args:
        question: The question to answer
    
    Returns:
        ExecutionPlan with all tool calls planned
    

    Parameters: – question: Question to answer

    Returns: ExecutionPlan

    execute_plan(self, plan, parallel)

    Lines 188-260

    Purpose:

    Execute the tool calls in the plan.
    
    Args:
        plan: The execution plan
        parallel: Whether to execute independent calls in parallel
    
    Returns:
        Dictionary with execution results
    

    Parameters: – plan: Plan – parallel: Parallel

    Returns: <ast.Subscript object at 0x000001A06D65A490>

    _execute_tool_call(self, call, plan)

    Lines 262-282

    Purpose:

    Execute a single tool call.
    

    Parameters: – call: Call – plan: Plan

    Returns: str

    _synthesize_answer(self, plan)

    Lines 284-316

    Purpose:

    Synthesize final answer from tool results.
    

    Parameters: – plan: Plan

    Returns: str

    solve(self, question, parallel)

    Lines 318-343

    Purpose:

    Solve a question using ReWOO pattern.
    
    Args:
        question: The question to answer
        parallel: Whether to use parallel execution
    
    Returns:
        Dictionary with answer and metadata
    

    Parameters: – question: Question to answer – parallel: Parallel

    Returns: <ast.Subscript object at 0x000001A06D178FD0>


    Functions

    calculator(expression)

    Lines 347-354

    Purpose:

    Calculate a mathematical expression.
    

    Parameters: – expression: Expression

    Returns: float

    search(query)

    Lines 357-372

    Purpose:

    Simulate a search tool.
    

    Parameters: – query: Input query or question

    Returns: str

    get_date(query)

    Lines 375-387

    Purpose:

    Get current date information.
    

    Parameters: – query: Input query or question

    Returns: str

    demonstrate_rewoo()

    Lines 390-481

    Purpose:

    Demonstrates the ReWOO pattern.
    

    Key Concepts Summary

    Pattern Overview

    The Rewoo 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 006
    • Pattern 008
    • Other Reasoning Patterns (001-011)

    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 Rewoo 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: 007_rewoo.py
    Total Lines: 492
    Pattern Category: AI Agent Design Pattern #007


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Plan And Execute

    
        """
    Pattern 006: Plan-and-Execute
    
    Description:
        Separates planning and execution phases for complex multi-step tasks.
        First creates a high-level plan, then executes each step sequentially,
        with optional re-planning based on intermediate results.
    
    Key Concepts:
        - Planning Phase: Create structured plan with steps
        - Execution Phase: Execute each step in sequence
        - Monitoring: Track progress and outcomes
        - Re-planning: Adjust plan based on execution results
        - State Management: Maintain context across steps
    
    Use Cases:
        - Complex multi-step tasks requiring coordination
        - Research and information gathering workflows
        - Content creation with multiple stages
        - Problem-solving with dependencies between steps
    
    LangChain Implementation:
        Uses LangGraph for state management and sequential execution with
        separate planner and executor chains.
    """
    
    import os
    from typing import List, Dict, Any, Optional
    from enum import Enum
    from dataclasses import dataclass, field
    from dotenv import load_dotenv
    
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser
    from langchain_openai import ChatOpenAI
    
    # Load environment variables
    load_dotenv()
    
    
    class StepStatus(Enum):
        """Status of an execution step."""
        PENDING = "pending"
        IN_PROGRESS = "in_progress"
        COMPLETED = "completed"
        FAILED = "failed"
        SKIPPED = "skipped"
    
    
    @dataclass
    class PlanStep:
        """Represents a single step in the execution plan."""
        id: int
        description: str
        dependencies: List[int] = field(default_factory=list)
        status: StepStatus = StepStatus.PENDING
        result: Optional[str] = None
        error: Optional[str] = None
    
    
    @dataclass
    class ExecutionPlan:
        """Represents the complete execution plan."""
        goal: str
        steps: List[PlanStep] = field(default_factory=list)
        current_step: int = 0
        completed_steps: int = 0
        
        def get_next_step(self) -> Optional[PlanStep]:
            """Get the next pending step."""
            for step in self.steps:
                if step.status == StepStatus.PENDING:
                    # Check if dependencies are met
                    if all(
                        self.steps[dep_id - 1].status == StepStatus.COMPLETED 
                        for dep_id in step.dependencies
                    ):
                        return step
            return None
        
        def is_complete(self) -> bool:
            """Check if all steps are completed or skipped."""
            return all(
                step.status in [StepStatus.COMPLETED, StepStatus.SKIPPED]
                for step in self.steps
            )
        
        def get_progress(self) -> float:
            """Get completion progress as percentage."""
            if not self.steps:
                return 0.0
            completed = sum(
                1 for step in self.steps 
                if step.status in [StepStatus.COMPLETED, StepStatus.SKIPPED]
            )
            return completed / len(self.steps)
    
    
    class PlanAndExecuteAgent:
        """Agent that uses plan-and-execute pattern."""
        
        def __init__(self, model_name: str = "gpt-3.5-turbo", 
                     enable_replanning: bool = True):
            """
            Initialize the Plan-and-Execute agent.
            
            Args:
                model_name: Name of the OpenAI model
                enable_replanning: Whether to enable dynamic re-planning
            """
            self.llm = ChatOpenAI(
                model=model_name,
                temperature=0.7,
                api_key=os.getenv("OPENAI_API_KEY")
            )
            self.enable_replanning = enable_replanning
            self.execution_history: List[Dict[str, Any]] = []
        
        def create_plan(self, goal: str, context: str = "") -> ExecutionPlan:
            """
            Create an execution plan for the given goal.
            
            Args:
                goal: The goal to achieve
                context: Additional context information
                
            Returns:
                ExecutionPlan with structured steps
            """
            prompt = ChatPromptTemplate.from_messages([
                ("system", """Create a detailed step-by-step plan to achieve the goal.
    Break down the task into clear, actionable steps.
    Format your response as:
    1. [First step]
    2. [Second step]
    3. [Third step]
    etc.
    
    Each step should be specific and achievable."""),
                ("human", """Goal: {goal}
    
    Context: {context}
    
    Plan:""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            response = chain.invoke({
                "goal": goal,
                "context": context if context else "No additional context."
            })
            
            # Parse steps
            plan = ExecutionPlan(goal=goal)
            step_id = 1
            
            for line in response.strip().split('\n'):
                line = line.strip()
                if line and (line[0].isdigit() or line.startswith('-')):
                    # Extract step description
                    description = line.split('.', 1)[-1].strip().lstrip('- ').strip()
                    if description:
                        step = PlanStep(
                            id=step_id,
                            description=description
                        )
                        plan.steps.append(step)
                        step_id += 1
            
            return plan
        
        def execute_step(self, step: PlanStep, plan: ExecutionPlan) -> str:
            """
            Execute a single step of the plan.
            
            Args:
                step: The step to execute
                plan: The complete plan for context
                
            Returns:
                Result of executing the step
            """
            # Gather context from previous steps
            context = self._build_context(plan)
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", """Execute the given step and provide a detailed result.
    Be specific and thorough in your execution.
    If the step requires information from previous steps, use the provided context."""),
                ("human", """Goal: {goal}
    
    Previous steps context:
    {context}
    
    Current step to execute: {step_description}
    
    Execution result:""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            
            try:
                result = chain.invoke({
                    "goal": plan.goal,
                    "context": context,
                    "step_description": step.description
                })
                
                return result.strip()
                
            except Exception as e:
                raise Exception(f"Step execution failed: {str(e)}")
        
        def _build_context(self, plan: ExecutionPlan) -> str:
            """Build context string from completed steps."""
            context_parts = []
            for step in plan.steps:
                if step.status == StepStatus.COMPLETED and step.result:
                    context_parts.append(f"Step {step.id}: {step.description}\nResult: {step.result}")
            
            return "\n\n".join(context_parts) if context_parts else "No previous steps completed."
        
        def should_replan(self, plan: ExecutionPlan, last_step: PlanStep) -> bool:
            """
            Determine if re-planning is needed based on execution results.
            
            Args:
                plan: Current execution plan
                last_step: Most recently executed step
                
            Returns:
                True if re-planning is recommended
            """
            if not self.enable_replanning:
                return False
            
            # Simple heuristic: replan if step failed or result suggests issues
            if last_step.status == StepStatus.FAILED:
                return True
            
            # Use LLM to evaluate if re-planning is needed
            prompt = ChatPromptTemplate.from_messages([
                ("system", """Evaluate if the plan should be adjusted based on the execution result.
    Consider if:
    - The result reveals new information that changes the approach
    - There are unexpected obstacles
    - The remaining steps need to be modified
    
    Respond with only 'YES' or 'NO'."""),
                ("human", """Goal: {goal}
    
    Completed step: {step_description}
    Result: {result}
    
    Should we replan? (YES/NO):""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            
            try:
                response = chain.invoke({
                    "goal": plan.goal,
                    "step_description": last_step.description,
                    "result": last_step.result
                })
                
                return "yes" in response.lower()
                
            except:
                return False
        
        def replan(self, plan: ExecutionPlan) -> ExecutionPlan:
            """
            Create a new plan based on current progress.
            
            Args:
                plan: Current execution plan
                
            Returns:
                Updated execution plan
            """
            context = self._build_context(plan)
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", """Given the progress so far, create an updated plan for the remaining work.
    Consider what has been accomplished and what still needs to be done.
    
    Format as:
    1. [First remaining step]
    2. [Second remaining step]
    etc."""),
                ("human", """Goal: {goal}
    
    Progress so far:
    {context}
    
    Updated plan for remaining work:""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            response = chain.invoke({
                "goal": plan.goal,
                "context": context
            })
            
            # Create new plan with completed steps + new steps
            new_plan = ExecutionPlan(goal=plan.goal)
            
            # Keep completed steps
            for step in plan.steps:
                if step.status == StepStatus.COMPLETED:
                    new_plan.steps.append(step)
            
            # Add new steps
            step_id = len(new_plan.steps) + 1
            for line in response.strip().split('\n'):
                line = line.strip()
                if line and (line[0].isdigit() or line.startswith('-')):
                    description = line.split('.', 1)[-1].strip().lstrip('- ').strip()
                    if description:
                        new_step = PlanStep(
                            id=step_id,
                            description=description
                        )
                        new_plan.steps.append(new_step)
                        step_id += 1
            
            return new_plan
        
        def execute_plan(self, plan: ExecutionPlan, max_steps: Optional[int] = None) -> Dict[str, Any]:
            """
            Execute the complete plan.
            
            Args:
                plan: The execution plan
                max_steps: Maximum number of steps to execute (None for all)
                
            Returns:
                Dictionary with execution results
            """
            print(f"\nGoal: {plan.goal}")
            print(f"\nPlan ({len(plan.steps)} steps):")
            for step in plan.steps:
                print(f"  {step.id}. {step.description}")
            print()
            
            executed_steps = 0
            
            while not plan.is_complete():
                if max_steps and executed_steps >= max_steps:
                    print(f"\nReached maximum steps limit ({max_steps})")
                    break
                
                # Get next step
                next_step = plan.get_next_step()
                if not next_step:
                    print("\nNo more executable steps (dependencies not met)")
                    break
                
                # Execute step
                print(f"\n{'='*60}")
                print(f"Executing Step {next_step.id}: {next_step.description}")
                print('='*60)
                
                next_step.status = StepStatus.IN_PROGRESS
                
                try:
                    result = self.execute_step(next_step, plan)
                    next_step.result = result
                    next_step.status = StepStatus.COMPLETED
                    plan.completed_steps += 1
                    
                    print(f"\n✓ Step {next_step.id} completed")
                    print(f"Result: {result[:200]}..." if len(result) > 200 else f"Result: {result}")
                    
                    # Check if re-planning is needed
                    if self.should_replan(plan, next_step):
                        print(f"\n⟳ Re-planning needed based on results...")
                        plan = self.replan(plan)
                        print(f"Updated plan now has {len(plan.steps)} steps")
                    
                except Exception as e:
                    next_step.status = StepStatus.FAILED
                    next_step.error = str(e)
                    print(f"\n✗ Step {next_step.id} failed: {str(e)}")
                
                executed_steps += 1
            
            # Generate final summary
            summary = self._generate_summary(plan)
            
            return {
                "goal": plan.goal,
                "total_steps": len(plan.steps),
                "completed_steps": plan.completed_steps,
                "progress": plan.get_progress(),
                "status": "completed" if plan.is_complete() else "incomplete",
                "summary": summary
            }
        
        def _generate_summary(self, plan: ExecutionPlan) -> str:
            """Generate a summary of the execution."""
            completed_results = []
            for step in plan.steps:
                if step.status == StepStatus.COMPLETED and step.result:
                    completed_results.append(f"Step {step.id}: {step.result}")
            
            if not completed_results:
                return "No steps completed successfully."
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", "Provide a concise summary of what was accomplished."),
                ("human", """Goal: {goal}
    
    Results:
    {results}
    
    Summary:""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            
            try:
                summary = chain.invoke({
                    "goal": plan.goal,
                    "results": "\n\n".join(completed_results)
                })
                return summary.strip()
            except:
                return "Summary generation failed."
        
        def solve_with_plan_execute(self, goal: str, context: str = "") -> Dict[str, Any]:
            """
            Solve a task using plan-and-execute pattern.
            
            Args:
                goal: The goal to achieve
                context: Additional context
                
            Returns:
                Execution results
            """
            print(f"\n{'='*80}")
            print("PLAN-AND-EXECUTE: Creating plan...")
            print('='*80)
            
            # Create initial plan
            plan = self.create_plan(goal, context)
            
            # Execute the plan
            result = self.execute_plan(plan)
            
            return result
    
    
    def demonstrate_plan_execute():
        """Demonstrates the Plan-and-Execute pattern."""
        
        print("=" * 80)
        print("PATTERN 006: Plan-and-Execute")
        print("=" * 80)
        print()
        print("Plan-and-Execute separates planning from execution:")
        print("1. Planning Phase: Create structured multi-step plan")
        print("2. Execution Phase: Execute each step sequentially")
        print("3. Monitoring: Track progress and results")
        print("4. Re-planning: Adjust plan based on intermediate results")
        print()
        
        # Create agent
        agent = PlanAndExecuteAgent(enable_replanning=True)
        
        # Test goals
        goals = [
            {
                "goal": "Research and write a brief report on renewable energy trends",
                "context": "Focus on solar and wind energy developments in the last 5 years"
            },
            {
                "goal": "Plan a team building event for 20 people",
                "context": "Budget: $2000, Duration: Half day, Location: Urban area"
            }
        ]
        
        for i, test_case in enumerate(goals, 1):
            print(f"\n{'='*80}")
            print(f"Example {i}")
            print('='*80)
            
            try:
                result = agent.solve_with_plan_execute(
                    goal=test_case['goal'],
                    context=test_case['context']
                )
                
                print(f"\n\n{'='*80}")
                print("EXECUTION SUMMARY")
                print('='*80)
                print(f"\nGoal: {result['goal']}")
                print(f"Status: {result['status'].upper()}")
                print(f"Progress: {result['progress']:.0%}")
                print(f"Steps Completed: {result['completed_steps']}/{result['total_steps']}")
                print(f"\nSummary:\n{result['summary']}")
                
            except Exception as e:
                print(f"\n✗ Error: {str(e)}")
                import traceback
                traceback.print_exc()
        
        # Summary
        print("\n\n" + "=" * 80)
        print("PLAN-AND-EXECUTE PATTERN DEMONSTRATION COMPLETE")
        print("=" * 80)
        print()
        print("Key Features Demonstrated:")
        print("1. Structured Planning: Breaking goals into actionable steps")
        print("2. Sequential Execution: Executing steps with dependency tracking")
        print("3. Context Management: Passing results between steps")
        print("4. Progress Monitoring: Tracking completion status")
        print("5. Dynamic Re-planning: Adjusting plan based on results")
        print()
        print("Advantages:")
        print("- Clear separation of planning and execution")
        print("- Better handling of complex multi-step tasks")
        print("- Adaptability through re-planning")
        print("- Progress tracking and monitoring")
        print()
        print("LangChain Components Used:")
        print("- ChatPromptTemplate: Structures planning and execution prompts")
        print("- StrOutputParser: Parses LLM outputs")
        print("- State Management: Tracks plan and execution progress")
        print("- Sequential Chains: Executes steps in order")
        print()
    
    
    if __name__ == "__main__":
        # Check for API key
        if not os.getenv("OPENAI_API_KEY"):
            print("Error: OPENAI_API_KEY not found in environment variables.")
            print("Please set it in your .env file or environment.")
            exit(1)
        
        demonstrate_plan_execute()
    
        

    Agentic Design Pattern: Plan And Execute

    Pattern 006: Plan And Execute – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Plan And Execute pattern implementation.


    Pattern Description

    Pattern 006: Plan-and-Execute

    Description: Separates planning and execution phases for complex multi-step tasks. First creates a high-level plan, then executes each step sequentially, with optional re-planning based on intermediate results.

    Key Concepts: – Planning Phase: Create structured plan with steps – Execution Phase: Execute each step in sequence – Monitoring: Track progress and outcomes – Re-planning: Adjust plan based on execution results – State Management: Maintain context across steps

    Use Cases: – Complex multi-step tasks requiring coordination – Research and information gathering workflows – Content creation with multiple stages – Problem-solving with dependencies between steps

    LangChain Implementation: Uses LangGraph for state management and sequential execution with separate planner and executor chains.


    File Statistics

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

    Module Documentation

    """
    Pattern 006: Plan-and-Execute
    
    Description:
        Separates planning and execution phases for complex multi-step tasks.
        First creates a high-level plan, then executes each step sequentially,
        with optional re-planning based on intermediate results.
    
    Key Concepts:
        - Planning Phase: Create structured plan with steps
        - Execution Phase: Execute each step in sequence
        - Monitoring: Track progress and outcomes
        - Re-planning: Adjust plan based on execution results
        - State Management: Maintain context across steps
    
    Use Cases:
        - Complex multi-step tasks requiring coordination
        - Research and information gathering workflows
        - Content creation with multiple stages
        - Problem-solving with dependencies between steps
    
    LangChain Implementation:
        Uses LangGraph for state management and sequential execution with
        separate planner and executor chains.
    """
    

    Purpose: This module implements the Plan And Execute pattern, providing a structured approach to agent design and implementation.


    Import Statements

    Standard Library Imports

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

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

    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 traceback
    

    Purpose: Additional dependencies for specialized functionality.


    Classes

    Class: StepStatus

    Lines 41-47

    Inherits from: Enum

    Purpose:

    Status of an execution step.
    

    Class: PlanStep

    Lines 51-58

    Purpose:

    Represents a single step in the execution plan.
    

    Class: ExecutionPlan

    Lines 62-96

    Purpose:

    Represents the complete execution plan.
    

    Methods:

    get_next_step(self)

    Lines 69-79

    Purpose:

    Get the next pending step.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D680D10>

    is_complete(self)

    Lines 81-86

    Purpose:

    Check if all steps are completed or skipped.
    

    Parameters:

    Returns: bool

    get_progress(self)

    Lines 88-96

    Purpose:

    Get completion progress as percentage.
    

    Parameters:

    Returns: float


    Class: PlanAndExecuteAgent

    Lines 99-453

    Purpose:

    Agent that uses plan-and-execute pattern.
    

    Methods:

    __init__(self, model_name, enable_replanning)

    Lines 102-117

    Purpose:

    Initialize the Plan-and-Execute agent.
    
    Args:
        model_name: Name of the OpenAI model
        enable_replanning: Whether to enable dynamic re-planning
    

    Parameters: – model_name: Language model instance – enable_replanning: Enable replanning

    create_plan(self, goal, context)

    Lines 119-170

    Purpose:

    Create an execution plan for the given goal.
    
    Args:
        goal: The goal to achieve
        context: Additional context information
    
    Returns:
        ExecutionPlan with structured steps
    

    Parameters: – goal: Goal – context: Context

    Returns: ExecutionPlan

    execute_step(self, step, plan)

    Lines 172-212

    Purpose:

    Execute a single step of the plan.
    
    Args:
        step: The step to execute
        plan: The complete plan for context
    
    Returns:
        Result of executing the step
    

    Parameters: – step: Step – plan: Plan

    Returns: str

    _build_context(self, plan)

    Lines 214-221

    Purpose:

    Build context string from completed steps.
    

    Parameters: – plan: Plan

    Returns: str

    should_replan(self, plan, last_step)

    Lines 223-270

    Purpose:

    Determine if re-planning is needed based on execution results.
    
    Args:
        plan: Current execution plan
        last_step: Most recently executed step
    
    Returns:
        True if re-planning is recommended
    

    Parameters: – plan: Plan – last_step: Last step

    Returns: bool

    replan(self, plan)

    Lines 272-328

    Purpose:

    Create a new plan based on current progress.
    
    Args:
        plan: Current execution plan
    
    Returns:
        Updated execution plan
    

    Parameters: – plan: Plan

    Returns: ExecutionPlan

    execute_plan(self, plan, max_steps)

    Lines 330-399

    Purpose:

    Execute the complete plan.
    
    Args:
        plan: The execution plan
        max_steps: Maximum number of steps to execute (None for all)
    
    Returns:
        Dictionary with execution results
    

    Parameters: – plan: Plan – max_steps: Max steps

    Returns: <ast.Subscript object at 0x000001A06D608ED0>

    _generate_summary(self, plan)

    Lines 401-430

    Purpose:

    Generate a summary of the execution.
    

    Parameters: – plan: Plan

    Returns: str

    solve_with_plan_execute(self, goal, context)

    Lines 432-453

    Purpose:

    Solve a task using plan-and-execute pattern.
    
    Args:
        goal: The goal to achieve
        context: Additional context
    
    Returns:
        Execution results
    

    Parameters: – goal: Goal – context: Context

    Returns: <ast.Subscript object at 0x000001A06D627750>


    Functions

    demonstrate_plan_execute()

    Lines 456-533

    Purpose:

    Demonstrates the Plan-and-Execute pattern.
    

    Key Concepts Summary

    Pattern Overview

    The Plan And Execute 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 005
    • Pattern 007
    • Other Reasoning Patterns (001-011)

    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 Plan And Execute 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: 006_plan_and_execute.py
    Total Lines: 544
    Pattern Category: AI Agent Design Pattern #006


    Generated from actual implementation code
    For educational and reference purposes