Category: aiml

  • Agentic Design Pattern: Tool Selection

    
        """
    Pattern 020: Tool Selection & Use
    
    Description:
        The Tool Selection & Use pattern enables agents to discover, select, and use external
        tools and APIs to extend their capabilities beyond the LLM's inherent knowledge. The
        agent analyzes the task, identifies appropriate tools, invokes them correctly, and
        processes the results.
    
    Components:
        - Tool Registry: Catalog of available tools with descriptions
        - Selection Logic: Determines which tool(s) to use for a task
        - Parameter Extraction: Extracts required parameters from context
        - Tool Invocation: Executes tools with proper error handling
        - Result Integration: Incorporates tool outputs into responses
    
    Use Cases:
        - Web search and information retrieval
        - Mathematical calculations
        - API integrations (weather, news, databases)
        - File operations and system commands
        - Data transformations
    
    LangChain Implementation:
        Uses LangChain's Tool abstraction with structured tool definitions,
        implements tool selection logic, and integrates results into agent workflow.
    
    Key Features:
        - Dynamic tool discovery
        - Context-aware tool selection
        - Robust error handling
        - Result formatting and integration
    """
    
    import os
    from typing import List, Dict, Any, Optional, Callable
    from dataclasses import dataclass
    from enum import Enum
    import json
    from datetime import datetime
    from dotenv import load_dotenv
    
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser
    from langchain_openai import ChatOpenAI
    
    load_dotenv()
    
    
    class ToolCategory(Enum):
        """Categories of tools available."""
        SEARCH = "search"
        COMPUTATION = "computation"
        DATA = "data"
        COMMUNICATION = "communication"
        UTILITY = "utility"
    
    
    @dataclass
    class ToolDefinition:
        """Definition of a tool available to the agent."""
        name: str
        category: ToolCategory
        description: str
        parameters: List[str]
        function: Callable
        
        def to_dict(self) -> Dict[str, Any]:
            """Convert to dictionary for LLM context."""
            return {
                "name": self.name,
                "category": self.category.value,
                "description": self.description,
                "parameters": self.parameters
            }
    
    
    @dataclass
    class ToolInvocation:
        """Record of a tool invocation."""
        tool_name: str
        parameters: Dict[str, Any]
        result: Any
        success: bool
        error: Optional[str] = None
    
    
    class ToolRegistry:
        """
        Registry of available tools with discovery and selection capabilities.
        """
        
        def __init__(self):
            """Initialize tool registry with built-in tools."""
            self.tools: Dict[str, ToolDefinition] = {}
            self._register_builtin_tools()
        
        def _register_builtin_tools(self):
            """Register built-in tools."""
            # Calculator tool
            self.register_tool(ToolDefinition(
                name="calculator",
                category=ToolCategory.COMPUTATION,
                description="Performs mathematical calculations. Supports +, -, *, /, **, sqrt, etc.",
                parameters=["expression"],
                function=self._calculator
            ))
            
            # Date/Time tool
            self.register_tool(ToolDefinition(
                name="datetime",
                category=ToolCategory.UTILITY,
                description="Gets current date and time, or formats dates",
                parameters=["format"],
                function=self._datetime
            ))
            
            # String manipulation tool
            self.register_tool(ToolDefinition(
                name="string_transform",
                category=ToolCategory.DATA,
                description="Transforms strings: uppercase, lowercase, reverse, length, etc.",
                parameters=["text", "operation"],
                function=self._string_transform
            ))
            
            # Search simulator (mock)
            self.register_tool(ToolDefinition(
                name="web_search",
                category=ToolCategory.SEARCH,
                description="Searches the web for information (simulated)",
                parameters=["query"],
                function=self._web_search_mock
            ))
            
            # Data formatter
            self.register_tool(ToolDefinition(
                name="format_data",
                category=ToolCategory.DATA,
                description="Formats data as JSON, CSV, or table",
                parameters=["data", "format"],
                function=self._format_data
            ))
        
        def register_tool(self, tool: ToolDefinition):
            """Register a new tool."""
            self.tools[tool.name] = tool
        
        def get_tool(self, name: str) -> Optional[ToolDefinition]:
            """Get a tool by name."""
            return self.tools.get(name)
        
        def get_tools_by_category(self, category: ToolCategory) -> List[ToolDefinition]:
            """Get all tools in a category."""
            return [tool for tool in self.tools.values() if tool.category == category]
        
        def search_tools(self, query: str) -> List[ToolDefinition]:
            """Search tools by query in name or description."""
            query_lower = query.lower()
            return [
                tool for tool in self.tools.values()
                if query_lower in tool.name.lower() or query_lower in tool.description.lower()
            ]
        
        def get_all_tools_description(self) -> str:
            """Get formatted description of all tools."""
            descriptions = []
            for tool in self.tools.values():
                params = ", ".join(tool.parameters)
                descriptions.append(
                    f"- {tool.name} ({tool.category.value}): {tool.description}\n"
                    f"  Parameters: {params}"
                )
            return "\n\n".join(descriptions)
        
        # Built-in tool implementations
        
        def _calculator(self, expression: str) -> str:
            """Calculate mathematical expression."""
            try:
                # Safe eval with limited scope
                allowed_names = {
                    "abs": abs, "round": round, "min": min, "max": max,
                    "sum": sum, "pow": pow, "sqrt": lambda x: x ** 0.5
                }
                result = eval(expression, {"__builtins__": {}}, allowed_names)
                return f"Result: {result}"
            except Exception as e:
                return f"Error: {str(e)}"
        
        def _datetime(self, format: str = "%Y-%m-%d %H:%M:%S") -> str:
            """Get current date/time."""
            try:
                return datetime.now().strftime(format)
            except Exception as e:
                return f"Error: {str(e)}"
        
        def _string_transform(self, text: str, operation: str) -> str:
            """Transform string."""
            operations = {
                "uppercase": lambda s: s.upper(),
                "lowercase": lambda s: s.lower(),
                "reverse": lambda s: s[::-1],
                "length": lambda s: str(len(s)),
                "capitalize": lambda s: s.capitalize(),
                "title": lambda s: s.title()
            }
            
            if operation in operations:
                return operations[operation](text)
            else:
                return f"Unknown operation: {operation}"
        
        def _web_search_mock(self, query: str) -> str:
            """Mock web search (simulated)."""
            # In real implementation, this would use actual search API
            mock_results = {
                "weather": "Current weather: Sunny, 72°F",
                "news": "Latest news: Tech stocks rise, new AI breakthrough announced",
                "python": "Python is a high-level programming language known for readability",
                "default": f"Search results for '{query}': Multiple relevant articles found"
            }
            
            for key, result in mock_results.items():
                if key in query.lower():
                    return result
            return mock_results["default"]
        
        def _format_data(self, data: str, format: str) -> str:
            """Format data in specified format."""
            try:
                if format == "json":
                    # Try to parse and pretty-print
                    parsed = json.loads(data)
                    return json.dumps(parsed, indent=2)
                elif format == "uppercase":
                    return data.upper()
                else:
                    return data
            except Exception as e:
                return f"Error formatting: {str(e)}"
    
    
    class ToolUsingAgent:
        """
        Agent that can discover, select, and use tools to accomplish tasks.
        """
        
        def __init__(
            self,
            tool_registry: ToolRegistry,
            model: str = "gpt-3.5-turbo"
        ):
            """
            Initialize tool-using agent.
            
            Args:
                tool_registry: Registry of available tools
                model: LLM model to use
            """
            self.tool_registry = tool_registry
            self.llm = ChatOpenAI(model=model, temperature=0.3)
            self.invocation_history: List[ToolInvocation] = []
        
        def select_tools(self, task: str) -> List[str]:
            """
            Select appropriate tools for a task.
            
            Args:
                task: Task description
                
            Returns:
                List of selected tool names
            """
            tools_description = self.tool_registry.get_all_tools_description()
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a tool selection expert. Given a task and available tools,
    select the most appropriate tool(s) to accomplish the task."""),
                ("user", """Task: {task}
    
    Available Tools:
    {tools_description}
    
    Select the best tool(s) for this task. Respond with ONLY the tool name(s), one per line.
    If no tool is needed, respond with "NONE".""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            
            response = chain.invoke({
                "task": task,
                "tools_description": tools_description
            })
            
            # Parse tool names from response
            lines = response.strip().split("\n")
            tool_names = [
                line.strip()
                for line in lines
                if line.strip() and line.strip() != "NONE"
            ]
            
            # Validate tool names
            valid_tools = [
                name for name in tool_names
                if self.tool_registry.get_tool(name) is not None
            ]
            
            return valid_tools
        
        def extract_parameters(
            self,
            task: str,
            tool: ToolDefinition
        ) -> Dict[str, Any]:
            """
            Extract tool parameters from task description.
            
            Args:
                task: Task description
                tool: Tool definition
                
            Returns:
                Dictionary of parameter values
            """
            prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a parameter extraction expert. Extract the required
    parameters for a tool from the task description."""),
                ("user", """Task: {task}
    
    Tool: {tool_name}
    Description: {tool_description}
    Required Parameters: {parameters}
    
    Extract the parameter values from the task. Respond ONLY with a JSON object
    mapping parameter names to their values. If a parameter is not mentioned,
    use a reasonable default or omit it.
    
    Example: {{"expression": "2+2", "format": "json"}}""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            
            response = chain.invoke({
                "task": task,
                "tool_name": tool.name,
                "tool_description": tool.description,
                "parameters": ", ".join(tool.parameters)
            })
            
            try:
                # Extract JSON from response
                json_start = response.find("{")
                json_end = response.rfind("}") + 1
                if json_start != -1 and json_end > json_start:
                    json_str = response[json_start:json_end]
                    parameters = json.loads(json_str)
                    return parameters
                else:
                    return {}
            except Exception as e:
                print(f"Error parsing parameters: {e}")
                return {}
        
        def invoke_tool(
            self,
            tool_name: str,
            parameters: Dict[str, Any]
        ) -> ToolInvocation:
            """
            Invoke a tool with given parameters.
            
            Args:
                tool_name: Name of tool to invoke
                parameters: Parameters for the tool
                
            Returns:
                ToolInvocation record
            """
            tool = self.tool_registry.get_tool(tool_name)
            
            if tool is None:
                invocation = ToolInvocation(
                    tool_name=tool_name,
                    parameters=parameters,
                    result=None,
                    success=False,
                    error=f"Tool '{tool_name}' not found"
                )
                self.invocation_history.append(invocation)
                return invocation
            
            try:
                # Invoke tool function with parameters
                result = tool.function(**parameters)
                
                invocation = ToolInvocation(
                    tool_name=tool_name,
                    parameters=parameters,
                    result=result,
                    success=True
                )
            except Exception as e:
                invocation = ToolInvocation(
                    tool_name=tool_name,
                    parameters=parameters,
                    result=None,
                    success=False,
                    error=str(e)
                )
            
            self.invocation_history.append(invocation)
            return invocation
        
        def process_task(self, task: str) -> Dict[str, Any]:
            """
            Process a task using tool selection and invocation.
            
            Args:
                task: Task to process
                
            Returns:
                Results including tool invocations and final response
            """
            print(f"\n[Agent] Processing task: {task}\n")
            
            # Step 1: Select tools
            print("[Agent] Step 1: Selecting tools...")
            selected_tools = self.select_tools(task)
            
            if not selected_tools:
                print("[Agent] No tools needed for this task\n")
                return {
                    "task": task,
                    "tools_used": [],
                    "invocations": [],
                    "final_response": "Task does not require tool usage"
                }
            
            print(f"[Agent] Selected tools: {', '.join(selected_tools)}\n")
            
            # Step 2: Extract parameters and invoke tools
            print("[Agent] Step 2: Invoking tools...")
            invocations = []
            
            for tool_name in selected_tools:
                tool = self.tool_registry.get_tool(tool_name)
                
                # Extract parameters
                parameters = self.extract_parameters(task, tool)
                print(f"[Agent] {tool_name} parameters: {parameters}")
                
                # Invoke tool
                invocation = self.invoke_tool(tool_name, parameters)
                invocations.append(invocation)
                
                if invocation.success:
                    print(f"[Agent] {tool_name} result: {invocation.result}\n")
                else:
                    print(f"[Agent] {tool_name} error: {invocation.error}\n")
            
            # Step 3: Integrate results
            print("[Agent] Step 3: Integrating results...")
            final_response = self._integrate_results(task, invocations)
            
            return {
                "task": task,
                "tools_used": selected_tools,
                "invocations": invocations,
                "final_response": final_response
            }
        
        def _integrate_results(
            self,
            task: str,
            invocations: List[ToolInvocation]
        ) -> str:
            """Integrate tool results into final response."""
            results_text = "\n".join([
                f"- {inv.tool_name}: {inv.result if inv.success else f'Error: {inv.error}'}"
                for inv in invocations
            ])
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You integrate tool results into coherent responses."),
                ("user", """Task: {task}
    
    Tool Results:
    {results_text}
    
    Provide a complete answer to the task incorporating these results naturally.""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            
            response = chain.invoke({
                "task": task,
                "results_text": results_text
            })
            
            return response.strip()
    
    
    def demonstrate_tool_selection():
        """Demonstrate the Tool Selection & Use pattern."""
        
        print("=" * 80)
        print("TOOL SELECTION & USE PATTERN DEMONSTRATION")
        print("=" * 80)
        
        # Initialize registry and agent
        registry = ToolRegistry()
        agent = ToolUsingAgent(registry)
        
        # Show available tools
        print("\n" + "=" * 80)
        print("AVAILABLE TOOLS")
        print("=" * 80)
        print(registry.get_all_tools_description())
        
        # Test 1: Calculation task
        print("\n" + "=" * 80)
        print("TEST 1: Mathematical Calculation")
        print("=" * 80)
        
        task1 = "Calculate the result of 15 squared plus 25"
        result1 = agent.process_task(task1)
        
        print("=" * 80)
        print("FINAL RESPONSE:")
        print("=" * 80)
        print(result1["final_response"])
        
        # Test 2: Multiple tools
        print("\n" + "=" * 80)
        print("TEST 2: Multiple Tools")
        print("=" * 80)
        
        task2 = "What is the current date and time, and also calculate 100 divided by 4?"
        result2 = agent.process_task(task2)
        
        print("=" * 80)
        print("FINAL RESPONSE:")
        print("=" * 80)
        print(result2["final_response"])
        
        print("\n" + "-" * 80)
        print("TOOLS USED:")
        print("-" * 80)
        for inv in result2["invocations"]:
            print(f"  - {inv.tool_name}: {'Success' if inv.success else 'Failed'}")
        
        # Test 3: String manipulation
        print("\n" + "=" * 80)
        print("TEST 3: String Transformation")
        print("=" * 80)
        
        task3 = "Convert 'hello world' to uppercase and tell me its length"
        result3 = agent.process_task(task3)
        
        print("=" * 80)
        print("FINAL RESPONSE:")
        print("=" * 80)
        print(result3["final_response"])
        
        # Test 4: Search task
        print("\n" + "=" * 80)
        print("TEST 4: Web Search (Simulated)")
        print("=" * 80)
        
        task4 = "Search for information about Python programming"
        result4 = agent.process_task(task4)
        
        print("=" * 80)
        print("FINAL RESPONSE:")
        print("=" * 80)
        print(result4["final_response"])
        
        # Summary
        print("\n" + "=" * 80)
        print("SUMMARY")
        print("=" * 80)
        print(f"""
    Total tasks processed: 4
    Total tool invocations: {len(agent.invocation_history)}
    Success rate: {sum(1 for inv in agent.invocation_history if inv.success) / len(agent.invocation_history) * 100:.0f}%
    
    The Tool Selection & Use pattern demonstrates:
    
    1. **Tool Discovery**: Agent can identify available tools
    2. **Intelligent Selection**: Chooses appropriate tools for tasks
    3. **Parameter Extraction**: Extracts parameters from natural language
    4. **Robust Invocation**: Handles tool execution with error handling
    5. **Result Integration**: Combines tool outputs into coherent responses
    
    Key Benefits:
    - **Extended Capabilities**: Goes beyond LLM knowledge
    - **Task-Appropriate**: Selects right tool for each job
    - **Error Resilient**: Graceful handling of failures
    - **Composable**: Can use multiple tools together
    
    Use Cases:
    - API integrations (weather, news, databases)
    - Mathematical computations
    - Web searches and data retrieval
    - File operations
    - Data transformations
    
    This pattern is the foundation for agentic AI systems that can interact
    with the real world through tools and APIs.
    """)
    
    
    if __name__ == "__main__":
        demonstrate_tool_selection()
    
        

    Agentic Design Pattern: Tool Selection

    Pattern 020: Tool Selection – Line-by-Line Explanation

    Overview

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


    Pattern Description

    Pattern 020: Tool Selection & Use

    Description: The Tool Selection & Use pattern enables agents to discover, select, and use external tools and APIs to extend their capabilities beyond the LLM’s inherent knowledge. The agent analyzes the task, identifies appropriate tools, invokes them correctly, and processes the results.

    Components: – Tool Registry: Catalog of available tools with descriptions – Selection Logic: Determines which tool(s) to use for a task – Parameter Extraction: Extracts required parameters from context – Tool Invocation: Executes tools with proper error handling – Result Integration: Incorporates tool outputs into responses

    Use Cases: – Web search and information retrieval – Mathematical calculations – API integrations (weather, news, databases) – File operations and system commands – Data transformations

    LangChain Implementation: Uses LangChain’s Tool abstraction with structured tool definitions, implements tool selection logic, and integrates results into agent workflow.

    Key Features: – Dynamic tool discovery – Context-aware tool selection – Robust error handling – Result formatting and integration


    File Statistics

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

    Module Documentation

    """
    Pattern 020: Tool Selection & Use
    
    Description:
        The Tool Selection & Use pattern enables agents to discover, select, and use external
        tools and APIs to extend their capabilities beyond the LLM's inherent knowledge. The
        agent analyzes the task, identifies appropriate tools, invokes them correctly, and
        processes the results.
    
    Components:
        - Tool Registry: Catalog of available tools with descriptions
        - Selection Logic: Determines which tool(s) to use for a task
        - Parameter Extraction: Extracts required parameters from context
        - Tool Invocation: Executes tools with proper error handling
        - Result Integration: Incorporates tool outputs into responses
    
    Use Cases:
        - Web search and information retrieval
        - Mathematical calculations
        - API integrations (weather, news, databases)
        - File operations and system commands
        - Data transformations
    
    LangChain Implementation:
        Uses LangChain's Tool abstraction with structured tool definitions,
        implements tool selection logic, and integrates results into agent workflow.
    
    Key Features:
        - Dynamic tool discovery
        - Context-aware tool selection
        - Robust error handling
        - Result formatting and integration
    """
    

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


    Import Statements

    Standard Library Imports

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

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

    LangChain Imports

    from langchain_openai.ChatOpenAI
    

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

    Other Imports

    from dotenv.load_dotenv
    

    Purpose: Additional dependencies for specialized functionality.


    Classes

    Class: ToolCategory

    Lines 50-56

    Inherits from: Enum

    Purpose:

    Categories of tools available.
    

    Class: ToolDefinition

    Lines 60-75

    Purpose:

    Definition of a tool available to the agent.
    

    Methods:

    to_dict(self)

    Lines 68-75

    Purpose:

    Convert to dictionary for LLM context.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D60E950>


    Class: ToolInvocation

    Lines 79-85

    Purpose:

    Record of a tool invocation.
    

    Class: ToolRegistry

    Lines 88-241

    Purpose:

    Registry of available tools with discovery and selection capabilities.
    

    Methods:

    __init__(self)

    Lines 93-96

    Purpose:

    Initialize tool registry with built-in tools.
    

    Parameters:

    _register_builtin_tools(self)

    Lines 98-143

    Purpose:

    Register built-in tools.
    

    Parameters:

    register_tool(self, tool)

    Lines 145-147

    Purpose:

    Register a new tool.
    

    Parameters: – tool: Tool

    get_tool(self, name)

    Lines 149-151

    Purpose:

    Get a tool by name.
    

    Parameters: – name: Name

    Returns: <ast.Subscript object at 0x000001A06D6A1510>

    get_tools_by_category(self, category)

    Lines 153-155

    Purpose:

    Get all tools in a category.
    

    Parameters: – category: Category

    Returns: <ast.Subscript object at 0x000001A06D6A1E50>

    search_tools(self, query)

    Lines 157-163

    Purpose:

    Search tools by query in name or description.
    

    Parameters: – query: Input query or question

    Returns: <ast.Subscript object at 0x000001A06D168490>

    get_all_tools_description(self)

    Lines 165-174

    Purpose:

    Get formatted description of all tools.
    

    Parameters:

    Returns: str

    _calculator(self, expression)

    Lines 178-189

    Purpose:

    Calculate mathematical expression.
    

    Parameters: – expression: Expression

    Returns: str

    _datetime(self, format)

    Lines 191-196

    Purpose:

    Get current date/time.
    

    Parameters: – format: Format

    Returns: str

    _string_transform(self, text, operation)

    Lines 198-212

    Purpose:

    Transform string.
    

    Parameters: – text: Text – operation: Operation

    Returns: str

    _web_search_mock(self, query)

    Lines 214-227

    Purpose:

    Mock web search (simulated).
    

    Parameters: – query: Input query or question

    Returns: str

    _format_data(self, data, format)

    Lines 229-241

    Purpose:

    Format data in specified format.
    

    Parameters: – data: Data – format: Format

    Returns: str


    Class: ToolUsingAgent

    Lines 244-502

    Purpose:

    Agent that can discover, select, and use tools to accomplish tasks.
    

    Methods:

    __init__(self, tool_registry, model)

    Lines 249-263

    Purpose:

    Initialize tool-using agent.
    
    Args:
        tool_registry: Registry of available tools
        model: LLM model to use
    

    Parameters: – tool_registry: Tool registry – model: Language model instance

    select_tools(self, task)

    Lines 265-310

    Purpose:

    Select appropriate tools for a task.
    
    Args:
        task: Task description
    
    Returns:
        List of selected tool names
    

    Parameters: – task: Task

    Returns: <ast.Subscript object at 0x000001A06D627510>

    extract_parameters(self, task, tool)

    Lines 312-364

    Purpose:

    Extract tool parameters from task description.
    
    Args:
        task: Task description
        tool: Tool definition
    
    Returns:
        Dictionary of parameter values
    

    Parameters: – task: Task – tool: Tool

    Returns: <ast.Subscript object at 0x000001A06D6076D0>

    invoke_tool(self, tool_name, parameters)

    Lines 366-414

    Purpose:

    Invoke a tool with given parameters.
    
    Args:
        tool_name: Name of tool to invoke
        parameters: Parameters for the tool
    
    Returns:
        ToolInvocation record
    

    Parameters: – tool_name: Tool name – parameters: Parameters

    Returns: ToolInvocation

    process_task(self, task)

    Lines 416-472

    Purpose:

    Process a task using tool selection and invocation.
    
    Args:
        task: Task to process
    
    Returns:
        Results including tool invocations and final response
    

    Parameters: – task: Task

    Returns: <ast.Subscript object at 0x000001A06D633290>

    _integrate_results(self, task, invocations)

    Lines 474-502

    Purpose:

    Integrate tool results into final response.
    

    Parameters: – task: Task – invocations: Invocations

    Returns: str


    Functions

    demonstrate_tool_selection()

    Lines 505-612

    Purpose:

    Demonstrate the Tool Selection & Use pattern.
    

    Key Concepts Summary

    Pattern Overview

    The Tool Selection 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 019
    • Pattern 021
    • Other Tool Use Patterns (020-025)

    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 Tool Selection 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: 020_tool_selection.py
    Total Lines: 617
    Pattern Category: AI Agent Design Pattern #020


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Society Of Mind

    
        """
    Pattern 019: Society of Mind
    
    Description:
        The Society of Mind pattern implements a cognitive architecture where multiple
        specialized sub-agents (each handling a specific aspect of cognition) work together
        to form a coherent intelligent system. Inspired by Marvin Minsky's theory, this
        pattern views intelligence as emerging from the interaction of many simple processes.
    
    Components:
        - Perception Agents: Process and interpret inputs
        - Memory Agents: Store and retrieve information
        - Reasoning Agents: Logical thinking and analysis
        - Planning Agents: Goal setting and strategy
        - Execution Agents: Action taking and output generation
        - Meta-Cognitive Agent: Coordinates other agents
    
    Use Cases:
        - Complex cognitive modeling
        - General AI systems
        - Multi-faceted problem-solving
        - Comprehensive decision-making systems
    
    LangChain Implementation:
        Uses specialized LLM agents for different cognitive functions, implements
        inter-agent communication protocols, maintains shared mental state, and
        coordinates through a meta-cognitive controller.
    
    Key Features:
        - Specialized cognitive sub-agents
        - Integrated cognitive architecture
        - Emergent intelligent behavior
        - Meta-cognitive coordination
    """
    
    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 CognitiveFunction(Enum):
        """Types of cognitive functions."""
        PERCEPTION = "perception"
        MEMORY = "memory"
        REASONING = "reasoning"
        PLANNING = "planning"
        EXECUTION = "execution"
        EMOTION = "emotion"
    
    
    @dataclass
    class MentalState:
        """Shared mental state of the society."""
        current_input: Optional[str] = None
        perceived_situation: Optional[str] = None
        relevant_memories: List[str] = field(default_factory=list)
        reasoning_output: Optional[str] = None
        plan: Optional[str] = None
        emotional_state: Optional[str] = None
        final_output: Optional[str] = None
    
    
    @dataclass
    class CognitiveOutput:
        """Output from a cognitive agent."""
        function: CognitiveFunction
        agent_id: str
        output: str
        confidence: float
    
    
    class CognitiveAgent:
        """
        A specialized cognitive agent handling one aspect of cognition.
        """
        
        def __init__(
            self,
            agent_id: str,
            function: CognitiveFunction,
            description: str,
            model: str = "gpt-3.5-turbo"
        ):
            """
            Initialize a cognitive agent.
            
            Args:
                agent_id: Unique identifier
                function: Cognitive function this agent handles
                description: Description of agent's role
                model: LLM model to use
            """
            self.agent_id = agent_id
            self.function = function
            self.description = description
            self.llm = ChatOpenAI(model=model, temperature=0.6)
        
        def get_system_prompt(self) -> str:
            """Get the system prompt for this cognitive agent."""
            prompts = {
                CognitiveFunction.PERCEPTION: f"""You are the Perception Agent.
    Role: {self.description}
    You interpret inputs, identify key elements, and understand the situation.
    Provide clear, objective perception of what you observe.""",
                
                CognitiveFunction.MEMORY: f"""You are the Memory Agent.
    Role: {self.description}
    You recall relevant past experiences, facts, and patterns.
    Provide relevant memories that can inform current processing.""",
                
                CognitiveFunction.REASONING: f"""You are the Reasoning Agent.
    Role: {self.description}
    You analyze information logically, identify patterns, and draw conclusions.
    Provide sound logical reasoning and analysis.""",
                
                CognitiveFunction.PLANNING: f"""You are the Planning Agent.
    Role: {self.description}
    You create strategies, set goals, and plan actions.
    Provide clear, actionable plans.""",
                
                CognitiveFunction.EXECUTION: f"""You are the Execution Agent.
    Role: {self.description}
    You take plans and create concrete outputs and actions.
    Provide clear, actionable outputs.""",
                
                CognitiveFunction.EMOTION: f"""You are the Emotion Agent.
    Role: {self.description}
    You assess emotional context and implications.
    Provide emotional intelligence and empathy."""
            }
            return prompts[self.function]
        
        def process(
            self,
            mental_state: MentalState,
            context: Optional[str] = None
        ) -> CognitiveOutput:
            """
            Process information according to cognitive function.
            
            Args:
                mental_state: Current mental state
                context: Optional additional context
                
            Returns:
                Cognitive output
            """
            # Build context based on what's available in mental state
            state_context = self._build_context(mental_state)
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", self.get_system_prompt()),
                ("user", """Current Mental State:
    {state_context}
    
    {additional_context}
    
    Perform your cognitive function and provide output (2-3 sentences):""")
            ])
            
            additional = f"Additional Context: {context}" if context else ""
            
            chain = prompt | self.llm | StrOutputParser()
            
            output = chain.invoke({
                "state_context": state_context,
                "additional_context": additional
            })
            
            return CognitiveOutput(
                function=self.function,
                agent_id=self.agent_id,
                output=output.strip(),
                confidence=0.8  # Simplified confidence
            )
        
        def _build_context(self, mental_state: MentalState) -> str:
            """Build context string from mental state."""
            parts = []
            
            if mental_state.current_input:
                parts.append(f"Input: {mental_state.current_input}")
            if mental_state.perceived_situation:
                parts.append(f"Perception: {mental_state.perceived_situation}")
            if mental_state.relevant_memories:
                parts.append(f"Memories: {'; '.join(mental_state.relevant_memories[:3])}")
            if mental_state.reasoning_output:
                parts.append(f"Reasoning: {mental_state.reasoning_output}")
            if mental_state.plan:
                parts.append(f"Plan: {mental_state.plan}")
            if mental_state.emotional_state:
                parts.append(f"Emotional Context: {mental_state.emotional_state}")
            
            return "\n".join(parts) if parts else "No information yet"
    
    
    class MetaCognitiveController:
        """
        Meta-cognitive controller that coordinates cognitive agents.
        """
        
        def __init__(self, model: str = "gpt-3.5-turbo"):
            """Initialize meta-cognitive controller."""
            self.llm = ChatOpenAI(model=model, temperature=0.5)
        
        def decide_next_step(
            self,
            mental_state: MentalState,
            completed_functions: List[CognitiveFunction]
        ) -> Optional[CognitiveFunction]:
            """
            Decide which cognitive function to invoke next.
            
            Args:
                mental_state: Current mental state
                completed_functions: Functions already completed
                
            Returns:
                Next cognitive function to invoke, or None if done
            """
            # Standard cognitive pipeline
            pipeline = [
                CognitiveFunction.PERCEPTION,
                CognitiveFunction.MEMORY,
                CognitiveFunction.EMOTION,
                CognitiveFunction.REASONING,
                CognitiveFunction.PLANNING,
                CognitiveFunction.EXECUTION
            ]
            
            for function in pipeline:
                if function not in completed_functions:
                    return function
            
            return None  # All done
        
        def integrate_outputs(
            self,
            mental_state: MentalState,
            outputs: List[CognitiveOutput]
        ) -> str:
            """
            Integrate outputs from all cognitive agents.
            
            Args:
                mental_state: Final mental state
                outputs: All cognitive outputs
                
            Returns:
                Integrated final response
            """
            outputs_text = "\n\n".join([
                f"{out.function.value.upper()}:\n{out.output}"
                for out in outputs
            ])
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", """You are the Meta-Cognitive Integrator.
    You synthesize outputs from specialized cognitive agents into a coherent response."""),
                ("user", """Original Input: {original_input}
    
    Cognitive Processing Results:
    {outputs_text}
    
    Synthesize these cognitive outputs into a unified, coherent final response that:
    1. Addresses the original input completely
    2. Integrates insights from all cognitive functions
    3. Is clear and actionable
    4. Shows balanced consideration of all aspects""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            
            integration = chain.invoke({
                "original_input": mental_state.current_input,
                "outputs_text": outputs_text
            })
            
            return integration.strip()
    
    
    class SocietyOfMind:
        """
        Implements the Society of Mind cognitive architecture.
        """
        
        def __init__(self, model: str = "gpt-3.5-turbo"):
            """Initialize the society of mind."""
            self.model = model
            self.cognitive_agents: Dict[CognitiveFunction, CognitiveAgent] = {}
            self.meta_controller = MetaCognitiveController(model)
            self.mental_state = MentalState()
            
            # Initialize cognitive agents
            self._initialize_agents()
        
        def _initialize_agents(self):
            """Initialize all cognitive agents."""
            agents_config = {
                CognitiveFunction.PERCEPTION: ("Perceptor", "Interprets and understands inputs"),
                CognitiveFunction.MEMORY: ("Memorist", "Recalls relevant past information"),
                CognitiveFunction.REASONING: ("Reasoner", "Analyzes and draws conclusions"),
                CognitiveFunction.PLANNING: ("Planner", "Creates strategies and plans"),
                CognitiveFunction.EXECUTION: ("Executor", "Generates concrete outputs"),
                CognitiveFunction.EMOTION: ("Empath", "Assesses emotional context")
            }
            
            for function, (agent_id, description) in agents_config.items():
                self.cognitive_agents[function] = CognitiveAgent(
                    agent_id,
                    function,
                    description,
                    self.model
                )
        
        def process(self, input_text: str) -> Dict[str, Any]:
            """
            Process input through the entire society of mind.
            
            Args:
                input_text: Input to process
                
            Returns:
                Processing results with all cognitive outputs
            """
            print(f"\n[Society of Mind] Processing: {input_text}\n")
            
            # Initialize mental state
            self.mental_state = MentalState(current_input=input_text)
            
            completed_functions: List[CognitiveFunction] = []
            cognitive_outputs: List[CognitiveOutput] = []
            
            # Cognitive processing pipeline
            while True:
                # Meta-controller decides next step
                next_function = self.meta_controller.decide_next_step(
                    self.mental_state,
                    completed_functions
                )
                
                if next_function is None:
                    break  # Processing complete
                
                # Invoke cognitive agent
                agent = self.cognitive_agents[next_function]
                
                print(f"[{agent.agent_id}] Processing ({next_function.value})...")
                
                output = agent.process(self.mental_state)
                cognitive_outputs.append(output)
                completed_functions.append(next_function)
                
                print(f"[{agent.agent_id}] Output: {output.output}\n")
                
                # Update mental state
                self._update_mental_state(output)
            
            # Integrate all outputs
            print("[Meta-Controller] Integrating cognitive outputs...\n")
            
            final_output = self.meta_controller.integrate_outputs(
                self.mental_state,
                cognitive_outputs
            )
            
            self.mental_state.final_output = final_output
            
            return {
                "input": input_text,
                "cognitive_outputs": cognitive_outputs,
                "mental_state": self.mental_state,
                "final_output": final_output
            }
        
        def _update_mental_state(self, output: CognitiveOutput):
            """Update mental state based on cognitive output."""
            if output.function == CognitiveFunction.PERCEPTION:
                self.mental_state.perceived_situation = output.output
            elif output.function == CognitiveFunction.MEMORY:
                self.mental_state.relevant_memories = [output.output]
            elif output.function == CognitiveFunction.REASONING:
                self.mental_state.reasoning_output = output.output
            elif output.function == CognitiveFunction.PLANNING:
                self.mental_state.plan = output.output
            elif output.function == CognitiveFunction.EMOTION:
                self.mental_state.emotional_state = output.output
        
        def get_architecture_summary(self) -> Dict[str, Any]:
            """Get summary of the cognitive architecture."""
            return {
                "cognitive_agents": [
                    {
                        "id": agent.agent_id,
                        "function": agent.function.value,
                        "description": agent.description
                    }
                    for agent in self.cognitive_agents.values()
                ],
                "processing_pipeline": [f.value for f in CognitiveFunction]
            }
    
    
    def demonstrate_society_of_mind():
        """Demonstrate the Society of Mind pattern."""
        
        print("=" * 80)
        print("SOCIETY OF MIND PATTERN DEMONSTRATION")
        print("=" * 80)
        
        # Initialize the society
        society = SocietyOfMind()
        
        # Show architecture
        print("\n" + "=" * 80)
        print("COGNITIVE ARCHITECTURE")
        print("=" * 80)
        
        arch = society.get_architecture_summary()
        print("\nCognitive Agents:")
        for agent in arch["cognitive_agents"]:
            print(f"  - {agent['id']} ({agent['function']}): {agent['description']}")
        
        # Test 1: Decision-making scenario
        print("\n" + "=" * 80)
        print("TEST 1: Complex Decision-Making")
        print("=" * 80)
        
        input1 = "I'm considering changing careers from engineering to product management. What should I consider?"
        
        result1 = society.process(input1)
        
        print("=" * 80)
        print("COGNITIVE PROCESSING BREAKDOWN:")
        print("=" * 80)
        for output in result1["cognitive_outputs"]:
            print(f"\n{output.function.value.upper()}:")
            print(f"  {output.output}")
        
        print("\n" + "=" * 80)
        print("FINAL INTEGRATED RESPONSE:")
        print("=" * 80)
        print(result1["final_output"])
        
        # Test 2: Creative problem-solving
        print("\n" + "=" * 80)
        print("TEST 2: Creative Problem-Solving")
        print("=" * 80)
        
        input2 = "How can I make studying more engaging and effective for students who find it boring?"
        
        result2 = society.process(input2)
        
        print("=" * 80)
        print("FINAL INTEGRATED RESPONSE:")
        print("=" * 80)
        print(result2["final_output"])
        
        print("\n" + "-" * 80)
        print("MENTAL STATE SNAPSHOT:")
        print("-" * 80)
        state = result2["mental_state"]
        print(f"Perception: {state.perceived_situation[:80]}...")
        print(f"Emotion: {state.emotional_state[:80]}...")
        print(f"Plan: {state.plan[:80]}...")
        
        # Test 3: Interpersonal situation
        print("\n" + "=" * 80)
        print("TEST 3: Interpersonal Situation")
        print("=" * 80)
        
        input3 = "My teammate is often late to meetings and it's affecting team morale. How should I address this?"
        
        result3 = society.process(input3)
        
        print("=" * 80)
        print("COGNITIVE PROCESSING BREAKDOWN:")
        print("=" * 80)
        for output in result3["cognitive_outputs"]:
            print(f"\n{output.function.value.upper()}:")
            print(f"  {output.output}")
        
        print("\n" + "=" * 80)
        print("FINAL INTEGRATED RESPONSE:")
        print("=" * 80)
        print(result3["final_output"])
        
        # Summary
        print("\n" + "=" * 80)
        print("SUMMARY")
        print("=" * 80)
        print("""
    The Society of Mind pattern demonstrates several key benefits:
    
    1. **Cognitive Specialization**: Each agent handles one aspect of cognition
    2. **Emergent Intelligence**: Complex behavior from simple component interactions
    3. **Comprehensive Processing**: All aspects of problem considered
    4. **Integrated Output**: Coherent response from diverse cognitive functions
    
    Cognitive Architecture:
    - **Perception**: Interprets and understands inputs
    - **Memory**: Recalls relevant past information
    - **Emotion**: Assesses emotional context and empathy
    - **Reasoning**: Analyzes logically and draws conclusions
    - **Planning**: Creates strategies and action plans
    - **Execution**: Generates concrete outputs
    
    Processing Flow:
    1. Input received and stored in mental state
    2. Each cognitive agent processes in sequence
    3. Mental state updated after each agent
    4. Meta-controller integrates all outputs
    5. Final coherent response generated
    
    Use Cases:
    - Complex decision-making systems
    - General AI architectures
    - Multi-faceted problem-solving
    - Cognitive modeling and simulation
    - Comprehensive analysis tasks
    
    The pattern is particularly effective when:
    - Problems require multiple perspectives
    - Cognitive diversity adds value
    - Comprehensive analysis is needed
    - Emotional intelligence matters
    - Integration of multiple factors is important
    
    This architecture models how human intelligence emerges from the interaction
    of many specialized mental processes, creating a more human-like AI system.
    """)
    
    
    if __name__ == "__main__":
        demonstrate_society_of_mind()
    
        

    Agentic Design Pattern: Society Of Mind

    Pattern 019: Society Of Mind – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Society Of Mind pattern implementation.


    Pattern Description

    Pattern 019: Society of Mind

    Description: The Society of Mind pattern implements a cognitive architecture where multiple specialized sub-agents (each handling a specific aspect of cognition) work together to form a coherent intelligent system. Inspired by Marvin Minsky’s theory, this pattern views intelligence as emerging from the interaction of many simple processes.

    Components: – Perception Agents: Process and interpret inputs – Memory Agents: Store and retrieve information – Reasoning Agents: Logical thinking and analysis – Planning Agents: Goal setting and strategy – Execution Agents: Action taking and output generation – Meta-Cognitive Agent: Coordinates other agents

    Use Cases: – Complex cognitive modeling – General AI systems – Multi-faceted problem-solving – Comprehensive decision-making systems

    LangChain Implementation: Uses specialized LLM agents for different cognitive functions, implements inter-agent communication protocols, maintains shared mental state, and coordinates through a meta-cognitive controller.

    Key Features: – Specialized cognitive sub-agents – Integrated cognitive architecture – Emergent intelligent behavior – Meta-cognitive coordination


    File Statistics

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

    Module Documentation

    """
    Pattern 019: Society of Mind
    
    Description:
        The Society of Mind pattern implements a cognitive architecture where multiple
        specialized sub-agents (each handling a specific aspect of cognition) work together
        to form a coherent intelligent system. Inspired by Marvin Minsky's theory, this
        pattern views intelligence as emerging from the interaction of many simple processes.
    
    Components:
        - Perception Agents: Process and interpret inputs
        - Memory Agents: Store and retrieve information
        - Reasoning Agents: Logical thinking and analysis
        - Planning Agents: Goal setting and strategy
        - Execution Agents: Action taking and output generation
        - Meta-Cognitive Agent: Coordinates other agents
    
    Use Cases:
        - Complex cognitive modeling
        - General AI systems
        - Multi-faceted problem-solving
        - Comprehensive decision-making systems
    
    LangChain Implementation:
        Uses specialized LLM agents for different cognitive functions, implements
        inter-agent communication protocols, maintains shared mental state, and
        coordinates through a meta-cognitive controller.
    
    Key Features:
        - Specialized cognitive sub-agents
        - Integrated cognitive architecture
        - Emergent intelligent behavior
        - Meta-cognitive coordination
    """
    

    Purpose: This module implements the Society Of Mind 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: CognitiveFunction

    Lines 49-56

    Inherits from: Enum

    Purpose:

    Types of cognitive functions.
    

    Class: MentalState

    Lines 60-68

    Purpose:

    Shared mental state of the society.
    

    Class: CognitiveOutput

    Lines 72-77

    Purpose:

    Output from a cognitive agent.
    

    Class: CognitiveAgent

    Lines 80-202

    Purpose:

    A specialized cognitive agent handling one aspect of cognition.
    

    Methods:

    __init__(self, agent_id, function, description, model)

    Lines 85-104

    Purpose:

    Initialize a cognitive agent.
    
    Args:
        agent_id: Unique identifier
        function: Cognitive function this agent handles
        description: Description of agent's role
        model: LLM model to use
    

    Parameters: – agent_id: Agent id – function: Function – description: Description – model: Language model instance

    get_system_prompt(self)

    Lines 106-139

    Purpose:

    Get the system prompt for this cognitive agent.
    

    Parameters:

    Returns: str

    process(self, mental_state, context)

    Lines 141-183

    Purpose:

    Process information according to cognitive function.
    
    Args:
        mental_state: Current mental state
        context: Optional additional context
    
    Returns:
        Cognitive output
    

    Parameters: – mental_state: Mental state – context: Context

    Returns: CognitiveOutput

    _build_context(self, mental_state)

    Lines 185-202

    Purpose:

    Build context string from mental state.
    

    Parameters: – mental_state: Mental state

    Returns: str


    Class: MetaCognitiveController

    Lines 205-287

    Purpose:

    Meta-cognitive controller that coordinates cognitive agents.
    

    Methods:

    __init__(self, model)

    Lines 210-212

    Purpose:

    Initialize meta-cognitive controller.
    

    Parameters: – model: Language model instance

    decide_next_step(self, mental_state, completed_functions)

    Lines 214-243

    Purpose:

    Decide which cognitive function to invoke next.
    
    Args:
        mental_state: Current mental state
        completed_functions: Functions already completed
    
    Returns:
        Next cognitive function to invoke, or None if done
    

    Parameters: – mental_state: Mental state – completed_functions: Completed functions

    Returns: <ast.Subscript object at 0x000001A06D71B610>

    integrate_outputs(self, mental_state, outputs)

    Lines 245-287

    Purpose:

    Integrate outputs from all cognitive agents.
    
    Args:
        mental_state: Final mental state
        outputs: All cognitive outputs
    
    Returns:
        Integrated final response
    

    Parameters: – mental_state: Mental state – outputs: Outputs

    Returns: str


    Class: SocietyOfMind

    Lines 290-409

    Purpose:

    Implements the Society of Mind cognitive architecture.
    

    Methods:

    __init__(self, model)

    Lines 295-303

    Purpose:

    Initialize the society of mind.
    

    Parameters: – model: Language model instance

    _initialize_agents(self)

    Lines 305-322

    Purpose:

    Initialize all cognitive agents.
    

    Parameters:

    process(self, input_text)

    Lines 324-382

    Purpose:

    Process input through the entire society of mind.
    
    Args:
        input_text: Input to process
    
    Returns:
        Processing results with all cognitive outputs
    

    Parameters: – input_text: Input data

    Returns: <ast.Subscript object at 0x000001A06D6597D0>

    _update_mental_state(self, output)

    Lines 384-395

    Purpose:

    Update mental state based on cognitive output.
    

    Parameters: – output: Output

    get_architecture_summary(self)

    Lines 397-409

    Purpose:

    Get summary of the cognitive architecture.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D158E90>


    Functions

    demonstrate_society_of_mind()

    Lines 412-539

    Purpose:

    Demonstrate the Society of Mind pattern.
    

    Key Concepts Summary

    Pattern Overview

    The Society Of Mind 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 018
    • Pattern 020
    • 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 Society Of Mind 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: 019_society_of_mind.py
    Total Lines: 544
    Pattern Category: AI Agent Design Pattern #019


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Cooperative Multi Agent

    
        """
    Pattern 018: Cooperative Multi-Agent
    
    Description:
        The Cooperative Multi-Agent pattern features agents working together toward shared
        goals through collaboration, communication, and coordination. Agents share information,
        coordinate actions, and help each other to achieve collective objectives more
        effectively than they could individually.
    
    Components:
        - Cooperative Agents: Agents designed to collaborate
        - Shared Goal System: Common objectives all agents work toward
        - Communication Protocol: Message passing and information sharing
        - Coordination Mechanism: Synchronizing agent actions
        - Resource Sharing: Pooling knowledge and capabilities
    
    Use Cases:
        - Complex collaborative tasks
        - Distributed problem-solving
        - Team-based projects
        - Multi-perspective analysis
    
    LangChain Implementation:
        Uses multiple LLM agents with shared context, implements message passing
        between agents, maintains shared knowledge base, and coordinates actions
        through explicit communication protocols.
    
    Key Features:
        - Shared goal alignment
        - Inter-agent communication
        - Knowledge sharing
        - Coordinated action planning
    """
    
    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 MessageType(Enum):
        """Types of messages agents can exchange."""
        INFORMATION = "information"
        REQUEST = "request"
        OFFER = "offer"
        COORDINATION = "coordination"
        RESULT = "result"
    
    
    @dataclass
    class Message:
        """Message exchanged between agents."""
        from_agent: str
        to_agent: Optional[str]  # None means broadcast to all
        message_type: MessageType
        content: str
        context: Optional[Dict[str, Any]] = None
    
    
    @dataclass
    class SharedKnowledge:
        """Knowledge base shared among cooperative agents."""
        facts: List[str] = field(default_factory=list)
        insights: List[str] = field(default_factory=list)
        decisions: List[str] = field(default_factory=list)
    
    
    @dataclass
    class CooperativeTask:
        """Task for cooperative agents to work on."""
        task_id: str
        description: str
        assigned_agents: List[str]
        result: Optional[str] = None
        completed: bool = False
    
    
    class CooperativeAgent:
        """
        An agent designed for cooperation and collaboration.
        """
        
        def __init__(
            self,
            agent_id: str,
            role: str,
            expertise: str,
            model: str = "gpt-3.5-turbo"
        ):
            """
            Initialize a cooperative agent.
            
            Args:
                agent_id: Unique identifier
                role: Agent's role in the team
                expertise: Agent's area of expertise
                model: LLM model to use
            """
            self.agent_id = agent_id
            self.role = role
            self.expertise = expertise
            self.llm = ChatOpenAI(model=model, temperature=0.6)
            
            # Collaboration state
            self.inbox: List[Message] = []
            self.outbox: List[Message] = []
            self.knowledge: List[str] = []
        
        def get_persona(self) -> str:
            """Get agent's cooperative persona."""
            return f"""You are {self.agent_id}, a cooperative team member.
    Role: {self.role}
    Expertise: {self.expertise}
    
    You are collaborative, supportive, and focused on team success.
    You actively share information, offer help, and coordinate with teammates."""
        
        def analyze_task(
            self,
            task: CooperativeTask,
            shared_knowledge: SharedKnowledge
        ) -> Dict[str, Any]:
            """
            Analyze a task from the agent's perspective.
            
            Args:
                task: Task to analyze
                shared_knowledge: Shared team knowledge
                
            Returns:
                Analysis with contributions and needs
            """
            knowledge_context = "\n".join([
                f"- {fact}" for fact in shared_knowledge.facts[:5]
            ]) if shared_knowledge.facts else "No shared knowledge yet"
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", self.get_persona()),
                ("user", """Task: {task_description}
    
    Team's shared knowledge:
    {knowledge_context}
    
    Analyze this task and provide:
    1. What you can contribute (based on your expertise)
    2. What help you need from teammates
    3. Key considerations
    
    Be specific and collaborative.""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            
            analysis = chain.invoke({
                "task_description": task.description,
                "knowledge_context": knowledge_context
            })
            
            return {"analysis": analysis.strip()}
        
        def contribute_to_task(
            self,
            task: CooperativeTask,
            team_messages: List[Message],
            shared_knowledge: SharedKnowledge
        ) -> str:
            """
            Contribute to solving the task.
            
            Args:
                task: Task to contribute to
                team_messages: Recent messages from team
                shared_knowledge: Shared knowledge base
                
            Returns:
                Agent's contribution
            """
            # Build context from team communication
            recent_messages = [
                f"{msg.from_agent} ({msg.message_type.value}): {msg.content}"
                for msg in team_messages[-5:]  # Last 5 messages
            ]
            message_context = "\n".join(recent_messages) if recent_messages else "No recent messages"
            
            knowledge_context = "\n".join([
                f"- {item}" for item in shared_knowledge.facts[:5]
            ]) if shared_knowledge.facts else "None yet"
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", self.get_persona()),
                ("user", """Task: {task_description}
    
    Recent team communication:
    {message_context}
    
    Shared knowledge:
    {knowledge_context}
    
    Provide your contribution to solving this task.
    Build upon teammates' input and share your expertise.
    Be specific and actionable (2-3 sentences).""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            
            contribution = chain.invoke({
                "task_description": task.description,
                "message_context": message_context,
                "knowledge_context": knowledge_context
            })
            
            return contribution.strip()
        
        def send_message(
            self,
            to_agent: Optional[str],
            message_type: MessageType,
            content: str
        ) -> Message:
            """
            Send a message to another agent or broadcast.
            
            Args:
                to_agent: Recipient agent ID (None for broadcast)
                message_type: Type of message
                content: Message content
                
            Returns:
                Created message
            """
            message = Message(
                from_agent=self.agent_id,
                to_agent=to_agent,
                message_type=message_type,
                content=content
            )
            self.outbox.append(message)
            return message
        
        def receive_message(self, message: Message):
            """Receive a message from another agent."""
            self.inbox.append(message)
    
    
    class CooperativeTeam:
        """
        Manages a team of cooperative agents.
        """
        
        def __init__(self, model: str = "gpt-3.5-turbo"):
            """Initialize cooperative team."""
            self.model = model
            self.agents: Dict[str, CooperativeAgent] = {}
            self.shared_knowledge = SharedKnowledge()
            self.message_log: List[Message] = []
            self.completed_tasks: List[CooperativeTask] = []
        
        def add_agent(
            self,
            agent_id: str,
            role: str,
            expertise: str
        ):
            """Add an agent to the team."""
            agent = CooperativeAgent(agent_id, role, expertise, self.model)
            self.agents[agent_id] = agent
        
        def broadcast_message(self, message: Message):
            """Broadcast a message to all agents."""
            for agent in self.agents.values():
                if agent.agent_id != message.from_agent:
                    agent.receive_message(message)
            self.message_log.append(message)
        
        def direct_message(self, message: Message):
            """Send a message to a specific agent."""
            if message.to_agent and message.to_agent in self.agents:
                self.agents[message.to_agent].receive_message(message)
            self.message_log.append(message)
        
        def share_knowledge(self, knowledge_item: str, category: str = "facts"):
            """Add knowledge to shared knowledge base."""
            if category == "facts":
                self.shared_knowledge.facts.append(knowledge_item)
            elif category == "insights":
                self.shared_knowledge.insights.append(knowledge_item)
            elif category == "decisions":
                self.shared_knowledge.decisions.append(knowledge_item)
        
        def collaborate_on_task(
            self,
            task_description: str,
            assigned_agents: Optional[List[str]] = None
        ) -> Dict[str, Any]:
            """
            Agents collaborate to complete a task.
            
            Args:
                task_description: Description of the task
                assigned_agents: Agents to assign (None = all agents)
                
            Returns:
                Collaboration results
            """
            if assigned_agents is None:
                assigned_agents = list(self.agents.keys())
            
            task = CooperativeTask(
                task_id=f"task_{len(self.completed_tasks) + 1}",
                description=task_description,
                assigned_agents=assigned_agents
            )
            
            print(f"\n[Team] Starting collaborative task: {task_description}")
            print(f"[Team] Assigned: {', '.join(assigned_agents)}\n")
            
            # Phase 1: Individual analysis
            print("=" * 80)
            print("PHASE 1: Individual Analysis")
            print("=" * 80 + "\n")
            
            analyses = {}
            for agent_id in assigned_agents:
                agent = self.agents[agent_id]
                analysis = agent.analyze_task(task, self.shared_knowledge)
                analyses[agent_id] = analysis
                
                print(f"[{agent_id}] Analysis:")
                print(f"  {analysis['analysis'][:150]}...\n")
                
                # Share analysis with team
                msg = agent.send_message(
                    None,  # Broadcast
                    MessageType.INFORMATION,
                    f"My analysis: {analysis['analysis']}"
                )
                self.broadcast_message(msg)
            
            # Phase 2: Coordination and planning
            print("=" * 80)
            print("PHASE 2: Coordination & Planning")
            print("=" * 80 + "\n")
            
            # First agent initiates coordination
            coordinator_id = assigned_agents[0]
            coordinator = self.agents[coordinator_id]
            
            coord_msg = coordinator.send_message(
                None,
                MessageType.COORDINATION,
                f"Let's coordinate our efforts. I'll focus on {coordinator.expertise}. Who can handle other aspects?"
            )
            self.broadcast_message(coord_msg)
            print(f"[{coordinator_id}] Coordination: {coord_msg.content}\n")
            
            # Other agents respond with offers
            for agent_id in assigned_agents[1:]:
                agent = self.agents[agent_id]
                offer_msg = agent.send_message(
                    coordinator_id,
                    MessageType.OFFER,
                    f"I can contribute my {agent.expertise} expertise"
                )
                self.direct_message(offer_msg)
                print(f"[{agent_id}] Offer: {offer_msg.content}\n")
            
            # Phase 3: Collaborative execution
            print("=" * 80)
            print("PHASE 3: Collaborative Execution")
            print("=" * 80 + "\n")
            
            contributions = {}
            for agent_id in assigned_agents:
                agent = self.agents[agent_id]
                
                contribution = agent.contribute_to_task(
                    task,
                    self.message_log,
                    self.shared_knowledge
                )
                contributions[agent_id] = contribution
                
                print(f"[{agent_id}] Contribution:")
                print(f"  {contribution}\n")
                
                # Share contribution
                result_msg = agent.send_message(
                    None,
                    MessageType.RESULT,
                    contribution
                )
                self.broadcast_message(result_msg)
                
                # Add to shared knowledge
                self.share_knowledge(
                    f"{agent_id}: {contribution}",
                    "insights"
                )
            
            # Phase 4: Synthesis
            print("=" * 80)
            print("PHASE 4: Synthesis")
            print("=" * 80 + "\n")
            
            final_result = self._synthesize_contributions(
                task_description,
                contributions
            )
            
            task.result = final_result
            task.completed = True
            self.completed_tasks.append(task)
            
            print(f"[Team] Task completed successfully!\n")
            
            return {
                "task": task,
                "individual_analyses": analyses,
                "contributions": contributions,
                "final_result": final_result,
                "messages_exchanged": len(self.message_log),
                "shared_knowledge_items": len(self.shared_knowledge.facts) + len(self.shared_knowledge.insights)
            }
        
        def _synthesize_contributions(
            self,
            task_description: str,
            contributions: Dict[str, str]
        ) -> str:
            """Synthesize individual contributions into final result."""
            # Use a synthesis agent
            synthesizer = ChatOpenAI(model=self.model, temperature=0.5)
            
            contributions_text = "\n\n".join([
                f"{agent_id} ({self.agents[agent_id].expertise}):\n{contrib}"
                for agent_id, contrib in contributions.items()
            ])
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are synthesizing collaborative team outputs into a cohesive result."),
                ("user", """Task: {task_description}
    
    Team Contributions:
    {contributions_text}
    
    Synthesize these contributions into a unified, coherent solution that:
    1. Integrates all team members' expertise
    2. Addresses the task completely
    3. Highlights collaborative synergies
    4. Provides actionable results""")
            ])
            
            chain = prompt | synthesizer | StrOutputParser()
            
            synthesis = chain.invoke({
                "task_description": task_description,
                "contributions_text": contributions_text
            })
            
            return synthesis.strip()
        
        def get_team_summary(self) -> Dict[str, Any]:
            """Get summary of team composition and performance."""
            return {
                "team_size": len(self.agents),
                "agents": [
                    {
                        "id": agent.agent_id,
                        "role": agent.role,
                        "expertise": agent.expertise
                    }
                    for agent in self.agents.values()
                ],
                "tasks_completed": len(self.completed_tasks),
                "messages_exchanged": len(self.message_log),
                "shared_knowledge_items": len(self.shared_knowledge.facts) + len(self.shared_knowledge.insights)
            }
    
    
    def demonstrate_cooperative_multi_agent():
        """Demonstrate the Cooperative Multi-Agent pattern."""
        
        print("=" * 80)
        print("COOPERATIVE MULTI-AGENT PATTERN DEMONSTRATION")
        print("=" * 80)
        
        # Test 1: Product development team
        print("\n" + "=" * 80)
        print("TEST 1: Product Development Team")
        print("=" * 80)
        
        team1 = CooperativeTeam()
        
        # Build diverse team
        team1.add_agent("Alice", "Product Manager", "Product Strategy")
        team1.add_agent("Bob", "Engineer", "Technical Implementation")
        team1.add_agent("Carol", "Designer", "User Experience")
        team1.add_agent("Dave", "Analyst", "Data Analysis")
        
        print("\nTeam Composition:")
        for agent_id, agent in team1.agents.items():
            print(f"  - {agent_id}: {agent.role} ({agent.expertise})")
        
        task1 = "Design and plan a new feature that helps users track their productivity goals"
        
        result1 = team1.collaborate_on_task(task1)
        
        print("=" * 80)
        print("FINAL RESULT:")
        print("=" * 80)
        print(result1["final_result"])
        
        print("\n" + "-" * 80)
        print("COLLABORATION METRICS:")
        print("-" * 80)
        print(f"  Messages exchanged: {result1['messages_exchanged']}")
        print(f"  Knowledge items shared: {result1['shared_knowledge_items']}")
        print(f"  Agents participated: {len(result1['contributions'])}")
        
        # Test 2: Research team
        print("\n" + "=" * 80)
        print("TEST 2: Research Team Collaboration")
        print("=" * 80)
        
        team2 = CooperativeTeam()
        
        team2.add_agent("ResearcherA", "Data Scientist", "Machine Learning")
        team2.add_agent("ResearcherB", "Domain Expert", "Healthcare")
        team2.add_agent("ResearcherC", "Statistician", "Statistical Analysis")
        
        print("\nTeam Composition:")
        for agent_id, agent in team2.agents.items():
            print(f"  - {agent_id}: {agent.role} ({agent.expertise})")
        
        task2 = "Develop a methodology to predict patient readmission risk using ML"
        
        result2 = team2.collaborate_on_task(task2)
        
        print("=" * 80)
        print("FINAL RESULT:")
        print("=" * 80)
        print(result2["final_result"])
        
        # Summary
        print("\n" + "=" * 80)
        print("SUMMARY")
        print("=" * 80)
        print("""
    The Cooperative Multi-Agent pattern demonstrates several key benefits:
    
    1. **Synergy**: Combined expertise exceeds individual capabilities
    2. **Knowledge Sharing**: Agents build on each other's contributions
    3. **Coordinated Action**: Agents work together toward shared goals
    4. **Communication**: Rich information exchange improves outcomes
    
    Collaboration Phases:
    1. **Individual Analysis**: Each agent analyzes from their perspective
    2. **Coordination**: Agents plan how to work together
    3. **Execution**: Agents contribute their expertise
    4. **Synthesis**: Contributions integrated into final result
    
    Communication Types:
    - **Information**: Sharing facts and knowledge
    - **Request**: Asking for help or information
    - **Offer**: Offering assistance or expertise
    - **Coordination**: Planning joint action
    - **Result**: Sharing completed work
    
    Use Cases:
    - Complex multi-disciplinary projects
    - Team-based problem-solving
    - Distributed collaborative work
    - Multi-perspective analysis
    - Cross-functional initiatives
    
    The pattern is particularly effective when:
    - Tasks require diverse expertise
    - Collaboration adds value over individual work
    - Communication and coordination are feasible
    - Shared goals align team members
    - Synergies between specialties exist
    """)
    
    
    if __name__ == "__main__":
        demonstrate_cooperative_multi_agent()
    
        

    Agentic Design Pattern: Cooperative Multi Agent

    Pattern 018: Cooperative Multi Agent – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Cooperative Multi Agent pattern implementation.


    Pattern Description

    Pattern 018: Cooperative Multi-Agent

    Description: The Cooperative Multi-Agent pattern features agents working together toward shared goals through collaboration, communication, and coordination. Agents share information, coordinate actions, and help each other to achieve collective objectives more effectively than they could individually.

    Components: – Cooperative Agents: Agents designed to collaborate – Shared Goal System: Common objectives all agents work toward – Communication Protocol: Message passing and information sharing – Coordination Mechanism: Synchronizing agent actions – Resource Sharing: Pooling knowledge and capabilities

    Use Cases: – Complex collaborative tasks – Distributed problem-solving – Team-based projects – Multi-perspective analysis

    LangChain Implementation: Uses multiple LLM agents with shared context, implements message passing between agents, maintains shared knowledge base, and coordinates actions through explicit communication protocols.

    Key Features: – Shared goal alignment – Inter-agent communication – Knowledge sharing – Coordinated action planning


    File Statistics

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

    Module Documentation

    """
    Pattern 018: Cooperative Multi-Agent
    
    Description:
        The Cooperative Multi-Agent pattern features agents working together toward shared
        goals through collaboration, communication, and coordination. Agents share information,
        coordinate actions, and help each other to achieve collective objectives more
        effectively than they could individually.
    
    Components:
        - Cooperative Agents: Agents designed to collaborate
        - Shared Goal System: Common objectives all agents work toward
        - Communication Protocol: Message passing and information sharing
        - Coordination Mechanism: Synchronizing agent actions
        - Resource Sharing: Pooling knowledge and capabilities
    
    Use Cases:
        - Complex collaborative tasks
        - Distributed problem-solving
        - Team-based projects
        - Multi-perspective analysis
    
    LangChain Implementation:
        Uses multiple LLM agents with shared context, implements message passing
        between agents, maintains shared knowledge base, and coordinates actions
        through explicit communication protocols.
    
    Key Features:
        - Shared goal alignment
        - Inter-agent communication
        - Knowledge sharing
        - Coordinated action planning
    """
    

    Purpose: This module implements the Cooperative Multi Agent 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: MessageType

    Lines 48-54

    Inherits from: Enum

    Purpose:

    Types of messages agents can exchange.
    

    Class: Message

    Lines 58-64

    Purpose:

    Message exchanged between agents.
    

    Class: SharedKnowledge

    Lines 68-72

    Purpose:

    Knowledge base shared among cooperative agents.
    

    Class: CooperativeTask

    Lines 76-82

    Purpose:

    Task for cooperative agents to work on.
    

    Class: CooperativeAgent

    Lines 85-249

    Purpose:

    An agent designed for cooperation and collaboration.
    

    Methods:

    __init__(self, agent_id, role, expertise, model)

    Lines 90-114

    Purpose:

    Initialize a cooperative agent.
    
    Args:
        agent_id: Unique identifier
        role: Agent's role in the team
        expertise: Agent's area of expertise
        model: LLM model to use
    

    Parameters: – agent_id: Agent id – role: Role – expertise: Expertise – model: Language model instance

    get_persona(self)

    Lines 116-123

    Purpose:

    Get agent's cooperative persona.
    

    Parameters:

    Returns: str

    analyze_task(self, task, shared_knowledge)

    Lines 125-166

    Purpose:

    Analyze a task from the agent's perspective.
    
    Args:
        task: Task to analyze
        shared_knowledge: Shared team knowledge
    
    Returns:
        Analysis with contributions and needs
    

    Parameters: – task: Task – shared_knowledge: Shared knowledge

    Returns: <ast.Subscript object at 0x000001A06D692550>

    contribute_to_task(self, task, team_messages, shared_knowledge)

    Lines 168-219

    Purpose:

    Contribute to solving the task.
    
    Args:
        task: Task to contribute to
        team_messages: Recent messages from team
        shared_knowledge: Shared knowledge base
    
    Returns:
        Agent's contribution
    

    Parameters: – task: Task – team_messages: Team messages – shared_knowledge: Shared knowledge

    Returns: str

    send_message(self, to_agent, message_type, content)

    Lines 221-245

    Purpose:

    Send a message to another agent or broadcast.
    
    Args:
        to_agent: Recipient agent ID (None for broadcast)
        message_type: Type of message
        content: Message content
    
    Returns:
        Created message
    

    Parameters: – to_agent: To agent – message_type: Message type – content: Content

    Returns: Message

    receive_message(self, message)

    Lines 247-249

    Purpose:

    Receive a message from another agent.
    

    Parameters: – message: Message


    Class: CooperativeTeam

    Lines 252-484

    Purpose:

    Manages a team of cooperative agents.
    

    Methods:

    __init__(self, model)

    Lines 257-263

    Purpose:

    Initialize cooperative team.
    

    Parameters: – model: Language model instance

    add_agent(self, agent_id, role, expertise)

    Lines 265-273

    Purpose:

    Add an agent to the team.
    

    Parameters: – agent_id: Agent id – role: Role – expertise: Expertise

    broadcast_message(self, message)

    Lines 275-280

    Purpose:

    Broadcast a message to all agents.
    

    Parameters: – message: Message

    direct_message(self, message)

    Lines 282-286

    Purpose:

    Send a message to a specific agent.
    

    Parameters: – message: Message

    share_knowledge(self, knowledge_item, category)

    Lines 288-295

    Purpose:

    Add knowledge to shared knowledge base.
    

    Parameters: – knowledge_item: Knowledge item – category: Category

    collaborate_on_task(self, task_description, assigned_agents)

    Lines 297-430

    Purpose:

    Agents collaborate to complete a task.
    
    Args:
        task_description: Description of the task
        assigned_agents: Agents to assign (None = all agents)
    
    Returns:
        Collaboration results
    

    Parameters: – task_description: Task description – assigned_agents: Assigned agents

    Returns: <ast.Subscript object at 0x000001A06D658050>

    _synthesize_contributions(self, task_description, contributions)

    Lines 432-467

    Purpose:

    Synthesize individual contributions into final result.
    

    Parameters: – task_description: Task description – contributions: Contributions

    Returns: str

    get_team_summary(self)

    Lines 469-484

    Purpose:

    Get summary of team composition and performance.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D609690>


    Functions

    demonstrate_cooperative_multi_agent()

    Lines 487-589

    Purpose:

    Demonstrate the Cooperative Multi-Agent pattern.
    

    Key Concepts Summary

    Pattern Overview

    The Cooperative Multi Agent 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 017
    • Pattern 019
    • 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 Cooperative Multi Agent 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: 018_cooperative_multi_agent.py
    Total Lines: 594
    Pattern Category: AI Agent Design Pattern #018


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Competitive Multi Agent

    
        """
    Pattern 017: Competitive Multi-Agent
    
    Description:
        The Competitive Multi-Agent pattern uses multiple agents that compete to produce
        the best solution. Agents work independently, their solutions are evaluated against
        each other, and the winning solution is selected. This competition drives quality
        improvement and innovation.
    
    Components:
        - Competitor Agents: Multiple agents with different strategies
        - Evaluation System: Judges solutions across multiple criteria
        - Competition Framework: Manages rounds and determines winners
        - Selection Mechanism: Chooses best solution(s)
    
    Use Cases:
        - Creative tasks (writing, design, ideation)
        - Optimization problems with multiple local optima
        - Generating diverse solution approaches
        - Quality improvement through competition
    
    LangChain Implementation:
        Uses multiple LLM agents with different configurations competing on tasks,
        implements multi-criteria evaluation, tournament-style competition, and
        quality-driven selection mechanisms.
    
    Key Features:
        - Tournament-style competitions
        - Multi-criteria evaluation
        - Diverse agent strategies
        - Winner selection and ranking
    """
    
    import os
    from typing import List, Dict, Any, Optional, Tuple
    from dataclasses import dataclass
    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()
    
    
    class CompetitionMode(Enum):
        """Modes of competition."""
        TOURNAMENT = "tournament"  # Bracket-style elimination
        ROUND_ROBIN = "round_robin"  # All agents compete in every round
        SURVIVAL = "survival"  # Lowest scorers eliminated each round
    
    
    @dataclass
    class Solution:
        """Represents a solution from a competitor."""
        agent_id: str
        content: str
        round_number: int
        scores: Dict[str, float] = None  # Criterion -> Score
        total_score: float = 0.0
        rank: Optional[int] = None
    
    
    @dataclass
    class CompetitionResult:
        """Results from a competition."""
        winner: Solution
        all_solutions: List[Solution]
        rounds: int
        mode: str
    
    
    class CompetitorAgent:
        """
        An agent that competes to produce the best solution.
        """
        
        def __init__(
            self,
            agent_id: str,
            strategy: str,
            model: str = "gpt-3.5-turbo",
            temperature: float = 0.7
        ):
            """
            Initialize a competitor agent.
            
            Args:
                agent_id: Unique identifier
                strategy: Competitive strategy description
                model: LLM model to use
                temperature: Creativity level
            """
            self.agent_id = agent_id
            self.strategy = strategy
            self.llm = ChatOpenAI(model=model, temperature=temperature)
            self.wins = 0
            self.total_score = 0.0
        
        def generate_solution(
            self,
            problem: str,
            round_number: int,
            previous_best: Optional[Solution] = None
        ) -> Solution:
            """
            Generate a solution for the problem.
            
            Args:
                problem: Problem to solve
                round_number: Current round number
                previous_best: Previous best solution (for improvement)
                
            Returns:
                Generated solution
            """
            strategy_context = f"\nYour strategy: {self.strategy}"
            
            if previous_best and round_number > 1:
                improvement_context = f"\n\nPrevious best solution to beat:\n{previous_best.content}\n\nImprove upon this!"
            else:
                improvement_context = ""
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", f"""You are a competitor in a solution competition.{strategy_context}
    
    Generate your BEST solution. Be creative, thorough, and competitive!"""),
                ("user", """Problem: {problem}{improvement_context}
    
    Your solution (aim to win the competition!):""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            
            content = chain.invoke({
                "problem": problem,
                "improvement_context": improvement_context
            })
            
            return Solution(
                agent_id=self.agent_id,
                content=content.strip(),
                round_number=round_number
            )
    
    
    class CompetitionJudge:
        """
        Evaluates and ranks solutions in competitions.
        """
        
        def __init__(
            self,
            criteria: List[str],
            model: str = "gpt-3.5-turbo"
        ):
            """
            Initialize the judge.
            
            Args:
                criteria: List of evaluation criteria
                model: LLM model for judging
            """
            self.criteria = criteria
            self.llm = ChatOpenAI(model=model, temperature=0.2)
        
        def evaluate_solution(
            self,
            problem: str,
            solution: Solution
        ) -> Solution:
            """
            Evaluate a solution across all criteria.
            
            Args:
                problem: Original problem
                solution: Solution to evaluate
                
            Returns:
                Solution with scores populated
            """
            scores = {}
            
            for criterion in self.criteria:
                score = self._score_criterion(problem, solution.content, criterion)
                scores[criterion] = score
            
            solution.scores = scores
            solution.total_score = sum(scores.values()) / len(scores)
            
            return solution
        
        def _score_criterion(
            self,
            problem: str,
            solution: str,
            criterion: str
        ) -> float:
            """Score a solution on a specific criterion."""
            prompt = ChatPromptTemplate.from_messages([
                ("system", f"You are an expert judge evaluating solutions based on: {criterion}"),
                ("user", """Problem: {problem}
    
    Solution: {solution}
    
    Rate this solution on {criterion} from 0.0 (worst) to 10.0 (best).
    Consider only this specific criterion.
    
    Provide only a number between 0.0 and 10.0:""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            
            try:
                score_str = chain.invoke({
                    "problem": problem,
                    "solution": solution,
                    "criterion": criterion
                })
                score = float(score_str.strip())
                return max(0.0, min(10.0, score))
            except:
                return 5.0  # Default moderate score
        
        def rank_solutions(
            self,
            solutions: List[Solution]
        ) -> List[Solution]:
            """
            Rank solutions by total score.
            
            Args:
                solutions: List of solutions to rank
                
            Returns:
                Ranked solutions (highest to lowest)
            """
            sorted_solutions = sorted(
                solutions,
                key=lambda s: s.total_score,
                reverse=True
            )
            
            for rank, solution in enumerate(sorted_solutions, 1):
                solution.rank = rank
            
            return sorted_solutions
    
    
    class CompetitiveSystem:
        """
        Manages competitive multi-agent system.
        """
        
        def __init__(
            self,
            evaluation_criteria: Optional[List[str]] = None,
            model: str = "gpt-3.5-turbo"
        ):
            """
            Initialize competitive system.
            
            Args:
                evaluation_criteria: Criteria for evaluating solutions
                model: LLM model for agents
            """
            if evaluation_criteria is None:
                evaluation_criteria = [
                    "Creativity and Innovation",
                    "Practicality and Feasibility",
                    "Completeness and Detail",
                    "Clarity and Communication"
                ]
            
            self.criteria = evaluation_criteria
            self.judge = CompetitionJudge(evaluation_criteria, model)
            self.agents: List[CompetitorAgent] = []
            self.model = model
        
        def add_competitor(
            self,
            agent_id: str,
            strategy: str,
            temperature: float = 0.7
        ):
            """Add a competitor to the system."""
            agent = CompetitorAgent(agent_id, strategy, self.model, temperature)
            self.agents.append(agent)
        
        def run_competition(
            self,
            problem: str,
            mode: CompetitionMode = CompetitionMode.ROUND_ROBIN,
            rounds: int = 2
        ) -> CompetitionResult:
            """
            Run a competition among agents.
            
            Args:
                problem: Problem for agents to solve
                mode: Competition mode
                rounds: Number of competition rounds
                
            Returns:
                Competition results
            """
            print(f"\n[Competition] Starting {mode.value} with {len(self.agents)} competitors")
            print(f"[Competition] {rounds} rounds, criteria: {', '.join(self.criteria)}\n")
            
            all_solutions: List[Solution] = []
            current_best: Optional[Solution] = None
            
            for round_num in range(1, rounds + 1):
                print(f"{'=' * 80}")
                print(f"ROUND {round_num}/{rounds}")
                print(f"{'=' * 80}\n")
                
                round_solutions = []
                
                # Each agent generates a solution
                for agent in self.agents:
                    print(f"[{agent.agent_id}] Generating solution (strategy: {agent.strategy})...")
                    
                    solution = agent.generate_solution(
                        problem,
                        round_num,
                        current_best
                    )
                    
                    # Evaluate solution
                    solution = self.judge.evaluate_solution(problem, solution)
                    
                    print(f"[{agent.agent_id}] Score: {solution.total_score:.2f}/10.0")
                    print(f"[{agent.agent_id}] Solution: {solution.content[:80]}...\n")
                    
                    round_solutions.append(solution)
                    all_solutions.append(solution)
                
                # Rank this round's solutions
                ranked = self.judge.rank_solutions(round_solutions)
                
                print(f"\n[Judge] Round {round_num} Rankings:")
                for solution in ranked[:3]:  # Top 3
                    print(f"  #{solution.rank} {solution.agent_id}: {solution.total_score:.2f}/10.0")
                
                # Update current best
                if ranked[0].total_score > (current_best.total_score if current_best else 0):
                    current_best = ranked[0]
                    print(f"\n[Judge] New leader: {current_best.agent_id}!")
                
                # Handle elimination for survival mode
                if mode == CompetitionMode.SURVIVAL and round_num  List[Tuple[str, int, float]]:
            """
            Get agent leaderboard.
            
            Returns:
                List of (agent_id, wins, avg_score) tuples
            """
            return sorted(
                [(a.agent_id, a.wins, a.total_score) for a in self.agents],
                key=lambda x: (x[1], x[2]),
                reverse=True
            )
    
    
    def demonstrate_competitive_multi_agent():
        """Demonstrate the Competitive Multi-Agent pattern."""
        
        print("=" * 80)
        print("COMPETITIVE MULTI-AGENT PATTERN DEMONSTRATION")
        print("=" * 80)
        
        # Test 1: Creative competition
        print("\n" + "=" * 80)
        print("TEST 1: Creative Problem-Solving Competition")
        print("=" * 80)
        
        system1 = CompetitiveSystem()
        
        # Add competitors with different strategies
        system1.add_competitor(
            "Innovator",
            "Focus on novel, cutting-edge, unconventional ideas",
            temperature=0.9
        )
        system1.add_competitor(
            "Pragmatist",
            "Focus on practical, proven, implementable solutions",
            temperature=0.5
        )
        system1.add_competitor(
            "Optimizer",
            "Focus on efficiency, cost-effectiveness, and ROI",
            temperature=0.6
        )
        system1.add_competitor(
            "User-Centric",
            "Focus on user experience, accessibility, and satisfaction",
            temperature=0.7
        )
        
        problem1 = "Design a feature to increase user engagement in a productivity app"
        
        result1 = system1.run_competition(
            problem1,
            mode=CompetitionMode.ROUND_ROBIN,
            rounds=2
        )
        
        print(f"\n{'*' * 80}")
        print("WINNER:")
        print(f"{'*' * 80}")
        print(f"Agent: {result1.winner.agent_id}")
        print(f"Score: {result1.winner.total_score:.2f}/10.0")
        print(f"Round: {result1.winner.round_number}")
        print(f"\nWinning Solution:\n{result1.winner.content}")
        
        print(f"\n{'-' * 80}")
        print("SCORE BREAKDOWN:")
        print(f"{'-' * 80}")
        for criterion, score in result1.winner.scores.items():
            print(f"  {criterion}: {score:.2f}/10.0")
        
        # Test 2: Survival mode competition
        print("\n" + "=" * 80)
        print("TEST 2: Survival Mode Competition")
        print("=" * 80)
        
        system2 = CompetitiveSystem(
            evaluation_criteria=[
                "Innovation",
                "Feasibility",
                "Impact"
            ]
        )
        
        system2.add_competitor("Alpha", "Aggressive innovation", 0.9)
        system2.add_competitor("Beta", "Balanced approach", 0.7)
        system2.add_competitor("Gamma", "Conservative and safe", 0.5)
        
        problem2 = "Propose a strategy to reduce customer churn by 30%"
        
        result2 = system2.run_competition(
            problem2,
            mode=CompetitionMode.SURVIVAL,
            rounds=2
        )
        
        print(f"\n{'*' * 80}")
        print("SURVIVOR & WINNER:")
        print(f"{'*' * 80}")
        print(f"Agent: {result2.winner.agent_id}")
        print(f"Score: {result2.winner.total_score:.2f}/10.0")
        print(f"\nWinning Solution:\n{result2.winner.content}")
        
        # Test 3: Tournament with diverse strategies
        print("\n" + "=" * 80)
        print("TEST 3: Multi-Round Tournament")
        print("=" * 80)
        
        system3 = CompetitiveSystem()
        
        system3.add_competitor("Creative", "Maximum creativity and originality", 1.0)
        system3.add_competitor("Analytical", "Data-driven and analytical", 0.4)
        system3.add_competitor("Hybrid", "Balance creativity with analysis", 0.7)
        
        problem3 = "Develop a marketing campaign for an eco-friendly product"
        
        result3 = system3.run_competition(
            problem3,
            mode=CompetitionMode.ROUND_ROBIN,
            rounds=3
        )
        
        print(f"\n{'*' * 80}")
        print("TOURNAMENT WINNER:")
        print(f"{'*' * 80}")
        print(f"Agent: {result3.winner.agent_id}")
        print(f"Average Score: {result3.winner.total_score:.2f}/10.0")
        
        # Show score progression
        print(f"\n{'-' * 80}")
        print("SCORE PROGRESSION:")
        print(f"{'-' * 80}")
        
        for agent_id in ["Creative", "Analytical", "Hybrid"]:
            agent_solutions = [s for s in result3.all_solutions if s.agent_id == agent_id]
            scores = [s.total_score for s in agent_solutions]
            print(f"{agent_id}: {' -> '.join([f'{s:.2f}' for s in scores])}")
        
        # Summary
        print("\n" + "=" * 80)
        print("SUMMARY")
        print("=" * 80)
        print("""
    The Competitive Multi-Agent pattern demonstrates several key benefits:
    
    1. **Quality Through Competition**: Competition drives agents to produce better solutions
    2. **Diverse Approaches**: Different strategies explore solution space differently
    3. **Iterative Improvement**: Agents can improve based on previous best solutions
    4. **Objective Selection**: Multi-criteria evaluation ensures fair judging
    
    Competition Modes:
    - **Round Robin**: All agents compete in every round (fair, comprehensive)
    - **Tournament**: Bracket-style elimination (exciting, efficient)
    - **Survival**: Lowest scorers eliminated each round (competitive pressure)
    
    Evaluation Approach:
    - **Multi-Criteria**: Solutions judged on multiple dimensions
    - **Objective Scoring**: Numerical scores (0-10) for each criterion
    - **Weighted Total**: Final score from criterion average
    - **Ranking**: Clear winner determination
    
    Use Cases:
    - Creative tasks (writing, design, ideation)
    - Optimization with multiple local optima
    - Generating diverse solution approaches
    - Quality improvement through competition
    - A/B testing alternative approaches
    
    The pattern is particularly effective when:
    - Solution quality is critical
    - Multiple valid approaches exist
    - Diversity of perspectives adds value
    - Competition motivates better outputs
    - Objective evaluation is possible
    """)
    
    
    if __name__ == "__main__":
        demonstrate_competitive_multi_agent()
    
        

    Agentic Design Pattern: Competitive Multi Agent

    Pattern 017: Competitive Multi Agent – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Competitive Multi Agent pattern implementation.


    Pattern Description

    Pattern 017: Competitive Multi-Agent

    Description: The Competitive Multi-Agent pattern uses multiple agents that compete to produce the best solution. Agents work independently, their solutions are evaluated against each other, and the winning solution is selected. This competition drives quality improvement and innovation.

    Components: – Competitor Agents: Multiple agents with different strategies – Evaluation System: Judges solutions across multiple criteria – Competition Framework: Manages rounds and determines winners – Selection Mechanism: Chooses best solution(s)

    Use Cases: – Creative tasks (writing, design, ideation) – Optimization problems with multiple local optima – Generating diverse solution approaches – Quality improvement through competition

    LangChain Implementation: Uses multiple LLM agents with different configurations competing on tasks, implements multi-criteria evaluation, tournament-style competition, and quality-driven selection mechanisms.

    Key Features: – Tournament-style competitions – Multi-criteria evaluation – Diverse agent strategies – Winner selection and ranking


    File Statistics

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

    Module Documentation

    """
    Pattern 017: Competitive Multi-Agent
    
    Description:
        The Competitive Multi-Agent pattern uses multiple agents that compete to produce
        the best solution. Agents work independently, their solutions are evaluated against
        each other, and the winning solution is selected. This competition drives quality
        improvement and innovation.
    
    Components:
        - Competitor Agents: Multiple agents with different strategies
        - Evaluation System: Judges solutions across multiple criteria
        - Competition Framework: Manages rounds and determines winners
        - Selection Mechanism: Chooses best solution(s)
    
    Use Cases:
        - Creative tasks (writing, design, ideation)
        - Optimization problems with multiple local optima
        - Generating diverse solution approaches
        - Quality improvement through competition
    
    LangChain Implementation:
        Uses multiple LLM agents with different configurations competing on tasks,
        implements multi-criteria evaluation, tournament-style competition, and
        quality-driven selection mechanisms.
    
    Key Features:
        - Tournament-style competitions
        - Multi-criteria evaluation
        - Diverse agent strategies
        - Winner selection and ranking
    """
    

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


    Import Statements

    Standard Library Imports

    import dataclasses.dataclass
    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: – 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 – 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: CompetitionMode

    Lines 48-52

    Inherits from: Enum

    Purpose:

    Modes of competition.
    

    Class: Solution

    Lines 56-63

    Purpose:

    Represents a solution from a competitor.
    

    Class: CompetitionResult

    Lines 67-72

    Purpose:

    Results from a competition.
    

    Class: CompetitorAgent

    Lines 75-146

    Purpose:

    An agent that competes to produce the best solution.
    

    Methods:

    __init__(self, agent_id, strategy, model, temperature)

    Lines 80-100

    Purpose:

    Initialize a competitor agent.
    
    Args:
        agent_id: Unique identifier
        strategy: Competitive strategy description
        model: LLM model to use
        temperature: Creativity level
    

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

    generate_solution(self, problem, round_number, previous_best)

    Lines 102-146

    Purpose:

    Generate a solution for the problem.
    
    Args:
        problem: Problem to solve
        round_number: Current round number
        previous_best: Previous best solution (for improvement)
    
    Returns:
        Generated solution
    

    Parameters: – problem: Problem – round_number: Round number – previous_best: Previous best

    Returns: Solution


    Class: CompetitionJudge

    Lines 149-249

    Purpose:

    Evaluates and ranks solutions in competitions.
    

    Methods:

    __init__(self, criteria, model)

    Lines 154-167

    Purpose:

    Initialize the judge.
    
    Args:
        criteria: List of evaluation criteria
        model: LLM model for judging
    

    Parameters: – criteria: Criteria – model: Language model instance

    evaluate_solution(self, problem, solution)

    Lines 169-193

    Purpose:

    Evaluate a solution across all criteria.
    
    Args:
        problem: Original problem
        solution: Solution to evaluate
    
    Returns:
        Solution with scores populated
    

    Parameters: – problem: Problem – solution: Solution

    Returns: Solution

    _score_criterion(self, problem, solution, criterion)

    Lines 195-225

    Purpose:

    Score a solution on a specific criterion.
    

    Parameters: – problem: Problem – solution: Solution – criterion: Criterion

    Returns: float

    rank_solutions(self, solutions)

    Lines 227-249

    Purpose:

    Rank solutions by total score.
    
    Args:
        solutions: List of solutions to rank
    
    Returns:
        Ranked solutions (highest to lowest)
    

    Parameters: – solutions: Solutions

    Returns: <ast.Subscript object at 0x000001A06D64EC90>


    Class: CompetitiveSystem

    Lines 252-390

    Purpose:

    Manages competitive multi-agent system.
    

    Methods:

    __init__(self, evaluation_criteria, model)

    Lines 257-280

    Purpose:

    Initialize competitive system.
    
    Args:
        evaluation_criteria: Criteria for evaluating solutions
        model: LLM model for agents
    

    Parameters: – evaluation_criteria: Evaluation criteria – model: Language model instance

    add_competitor(self, agent_id, strategy, temperature)

    Lines 282-290

    Purpose:

    Add a competitor to the system.
    

    Parameters: – agent_id: Agent id – strategy: Strategy – temperature: Sampling temperature (0-1)

    run_competition(self, problem, mode, rounds)

    Lines 292-377

    Purpose:

    Run a competition among agents.
    
    Args:
        problem: Problem for agents to solve
        mode: Competition mode
        rounds: Number of competition rounds
    
    Returns:
        Competition results
    

    Parameters: – problem: Problem – mode: Mode – rounds: Rounds

    Returns: CompetitionResult

    get_leaderboard(self)

    Lines 379-390

    Purpose:

    Get agent leaderboard.
    
    Returns:
        List of (agent_id, wins, avg_score) tuples
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D63CC50>


    Functions

    demonstrate_competitive_multi_agent()

    Lines 393-554

    Purpose:

    Demonstrate the Competitive Multi-Agent pattern.
    

    Key Concepts Summary

    Pattern Overview

    The Competitive Multi Agent 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 016
    • Pattern 018
    • 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 Competitive Multi Agent 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: 017_competitive_multi_agent.py
    Total Lines: 559
    Pattern Category: AI Agent Design Pattern #017


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Hierarchical Multi Agent

    
        """
    Pattern 016: Hierarchical Multi-Agent
    
    Description:
        The Hierarchical Multi-Agent pattern organizes agents in a hierarchical structure with
        multiple levels of management and specialization. Managers coordinate workers, workers
        execute tasks, and specialists provide domain expertise. This creates a scalable
        organizational structure for complex problems.
    
    Components:
        - Manager Agents: High-level coordination and decision-making
        - Worker Agents: Execute assigned tasks
        - Specialist Agents: Provide domain expertise
        - Communication Protocol: Top-down delegation and bottom-up reporting
        - Task Decomposition: Breaking complex tasks into hierarchical subtasks
    
    Use Cases:
        - Large-scale project management
        - Complex multi-domain problems
        - Organizational modeling and simulation
        - Scalable distributed systems
    
    LangChain Implementation:
        Uses multiple LLM instances representing different hierarchy levels,
        implements delegation chains, task assignment strategies, and result
        aggregation across organizational levels.
    
    Key Features:
        - Multi-level hierarchy (Manager -> Worker -> Specialist)
        - Top-down task delegation
        - Bottom-up result reporting
        - Cross-functional coordination
    """
    
    import os
    from typing import List, Dict, Any, Optional, Set
    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 AgentLevel(Enum):
        """Hierarchy levels in the organization."""
        EXECUTIVE = "executive"
        MANAGER = "manager"
        WORKER = "worker"
        SPECIALIST = "specialist"
    
    
    class TaskPriority(Enum):
        """Task priority levels."""
        CRITICAL = "critical"
        HIGH = "high"
        MEDIUM = "medium"
        LOW = "low"
    
    
    @dataclass
    class HierarchicalTask:
        """Represents a task in the hierarchy."""
        task_id: str
        description: str
        priority: TaskPriority
        assigned_to: Optional[str] = None
        assigned_level: Optional[AgentLevel] = None
        parent_task_id: Optional[str] = None
        subtasks: List[str] = field(default_factory=list)
        result: Optional[str] = None
        completed: bool = False
    
    
    @dataclass
    class AgentReport:
        """Report from an agent to their manager."""
        agent_id: str
        agent_level: AgentLevel
        task_id: str
        status: str
        result: str
        issues: List[str] = field(default_factory=list)
        recommendations: List[str] = field(default_factory=list)
    
    
    class HierarchicalAgent:
        """
        Base class for agents in the hierarchy.
        """
        
        def __init__(
            self,
            agent_id: str,
            level: AgentLevel,
            specialty: Optional[str] = None,
            model: str = "gpt-3.5-turbo"
        ):
            """
            Initialize a hierarchical agent.
            
            Args:
                agent_id: Unique identifier
                level: Hierarchy level
                specialty: Optional domain specialty
                model: LLM model to use
            """
            self.agent_id = agent_id
            self.level = level
            self.specialty = specialty
            self.llm = ChatOpenAI(model=model, temperature=0.5)
            
            # Reporting relationships
            self.manager: Optional[str] = None
            self.direct_reports: List[str] = []
            
            # Task management
            self.assigned_tasks: List[HierarchicalTask] = []
            self.completed_tasks: List[HierarchicalTask] = []
        
        def get_role_description(self) -> str:
            """Get the role description for this agent."""
            descriptions = {
                AgentLevel.EXECUTIVE: "You are an executive leader responsible for strategic direction and high-level coordination.",
                AgentLevel.MANAGER: f"You are a manager responsible for coordinating teams and ensuring task completion. Specialty: {self.specialty or 'General'}",
                AgentLevel.WORKER: f"You are a worker who executes assigned tasks efficiently. Specialty: {self.specialty or 'General'}",
                AgentLevel.SPECIALIST: f"You are a specialist with deep expertise in {self.specialty}. You provide expert guidance and solutions."
            }
            return descriptions[self.level]
        
        def delegate_task(
            self,
            task: HierarchicalTask,
            subordinate_id: str
        ) -> str:
            """
            Delegate a task to a subordinate with clear instructions.
            
            Args:
                task: Task to delegate
                subordinate_id: ID of subordinate to delegate to
                
            Returns:
                Delegation message
            """
            prompt = ChatPromptTemplate.from_messages([
                ("system", f"{self.get_role_description()}\n\nYou are delegating a task to {subordinate_id}."),
                ("user", """Task: {task_description}
    Priority: {priority}
    
    Provide clear delegation instructions including:
    1. What needs to be done
    2. Expected outcomes
    3. Any constraints or guidelines
    4. Reporting requirements
    
    Keep it concise (3-4 sentences).""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            
            delegation = chain.invoke({
                "task_description": task.description,
                "priority": task.priority.value
            })
            
            return delegation.strip()
        
        def execute_task(
            self,
            task: HierarchicalTask,
            context: Optional[str] = None
        ) -> str:
            """
            Execute an assigned task.
            
            Args:
                task: Task to execute
                context: Optional context from manager
                
            Returns:
                Task result
            """
            prompt = ChatPromptTemplate.from_messages([
                ("system", self.get_role_description()),
                ("user", """Execute this task:
    
    Task: {task_description}
    Priority: {priority}
    {context_section}
    
    Provide your result:""")
            ])
            
            context_section = f"\nManager's Instructions:\n{context}" if context else ""
            
            chain = prompt | self.llm | StrOutputParser()
            
            result = chain.invoke({
                "task_description": task.description,
                "priority": task.priority.value,
                "context_section": context_section
            })
            
            return result.strip()
        
        def create_report(
            self,
            task: HierarchicalTask,
            result: str
        ) -> AgentReport:
            """
            Create a report for manager.
            
            Args:
                task: Completed task
                result: Task result
                
            Returns:
                Agent report
            """
            prompt = ChatPromptTemplate.from_messages([
                ("system", f"{self.get_role_description()}\n\nCreate a concise status report for your manager."),
                ("user", """Task: {task_description}
    Result: {result}
    
    Provide:
    1. Status (one word: Completed/Blocked/In-Progress)
    2. Any issues encountered (or "None")
    3. Brief recommendations (or "None")
    
    Format:
    STATUS: [status]
    ISSUES: [issues]
    RECOMMENDATIONS: [recommendations]""")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            
            report_text = chain.invoke({
                "task_description": task.description,
                "result": result
            })
            
            # Parse report
            lines = report_text.strip().split("\n")
            status = "Completed"
            issues = []
            recommendations = []
            
            for line in lines:
                if line.startswith("STATUS:"):
                    status = line.split(":", 1)[1].strip()
                elif line.startswith("ISSUES:"):
                    issue_text = line.split(":", 1)[1].strip()
                    if issue_text.lower() != "none":
                        issues = [issue_text]
                elif line.startswith("RECOMMENDATIONS:"):
                    rec_text = line.split(":", 1)[1].strip()
                    if rec_text.lower() != "none":
                        recommendations = [rec_text]
            
            return AgentReport(
                agent_id=self.agent_id,
                agent_level=self.level,
                task_id=task.task_id,
                status=status,
                result=result,
                issues=issues,
                recommendations=recommendations
            )
    
    
    class HierarchicalOrganization:
        """
        Manages a hierarchical multi-agent organization.
        """
        
        def __init__(self, model: str = "gpt-3.5-turbo"):
            """Initialize the hierarchical organization."""
            self.model = model
            self.agents: Dict[str, HierarchicalAgent] = {}
            self.tasks: Dict[str, HierarchicalTask] = {}
            self.reports: List[AgentReport] = []
            
            # Build default organization structure
            self._build_organization()
        
        def _build_organization(self):
            """Build a default organizational structure."""
            # Executive level
            self.add_agent("CEO", AgentLevel.EXECUTIVE, "Strategic Leadership")
            
            # Manager level
            self.add_agent("PM_Tech", AgentLevel.MANAGER, "Technology")
            self.add_agent("PM_Ops", AgentLevel.MANAGER, "Operations")
            self.add_agent("PM_Research", AgentLevel.MANAGER, "Research")
            
            # Worker level
            self.add_agent("Worker_Dev", AgentLevel.WORKER, "Development")
            self.add_agent("Worker_QA", AgentLevel.WORKER, "Quality Assurance")
            self.add_agent("Worker_Data", AgentLevel.WORKER, "Data Processing")
            
            # Specialist level
            self.add_agent("Spec_ML", AgentLevel.SPECIALIST, "Machine Learning")
            self.add_agent("Spec_Security", AgentLevel.SPECIALIST, "Security")
            
            # Set up reporting structure
            self.set_manager("PM_Tech", "CEO")
            self.set_manager("PM_Ops", "CEO")
            self.set_manager("PM_Research", "CEO")
            
            self.set_manager("Worker_Dev", "PM_Tech")
            self.set_manager("Worker_QA", "PM_Tech")
            self.set_manager("Worker_Data", "PM_Ops")
            
            self.set_manager("Spec_ML", "PM_Research")
            self.set_manager("Spec_Security", "PM_Tech")
        
        def add_agent(
            self,
            agent_id: str,
            level: AgentLevel,
            specialty: Optional[str] = None
        ):
            """Add an agent to the organization."""
            agent = HierarchicalAgent(agent_id, level, specialty, self.model)
            self.agents[agent_id] = agent
        
        def set_manager(self, agent_id: str, manager_id: str):
            """Set manager-subordinate relationship."""
            if agent_id in self.agents and manager_id in self.agents:
                self.agents[agent_id].manager = manager_id
                self.agents[manager_id].direct_reports.append(agent_id)
        
        def decompose_task(
            self,
            task_description: str,
            priority: TaskPriority = TaskPriority.MEDIUM
        ) -> List[HierarchicalTask]:
            """
            Decompose a high-level task into hierarchical subtasks.
            
            Args:
                task_description: Main task description
                priority: Task priority
                
            Returns:
                List of hierarchical tasks
            """
            # CEO decomposes into manager-level tasks
            ceo = self.agents["CEO"]
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", ceo.get_role_description()),
                ("user", """Decompose this high-level task into 3-4 manager-level subtasks:
    
    Task: {task_description}
    
    For each subtask, specify:
    1. Which manager should handle it (PM_Tech, PM_Ops, or PM_Research)
    2. Brief description
    
    Format each as:
    [MANAGER_ID]: [task description]""")
            ])
            
            chain = prompt | ceo.llm | StrOutputParser()
            
            decomposition = chain.invoke({"task_description": task_description})
            
            # Parse decomposition
            tasks = []
            parent_task = HierarchicalTask(
                task_id="task_0",
                description=task_description,
                priority=priority,
                assigned_to="CEO",
                assigned_level=AgentLevel.EXECUTIVE
            )
            self.tasks[parent_task.task_id] = parent_task
            tasks.append(parent_task)
            
            for i, line in enumerate(decomposition.strip().split("\n"), 1):
                if ":" in line:
                    try:
                        manager_id, desc = line.split(":", 1)
                        manager_id = manager_id.strip().replace("[", "").replace("]", "")
                        
                        if manager_id in self.agents:
                            subtask = HierarchicalTask(
                                task_id=f"task_{i}",
                                description=desc.strip(),
                                priority=priority,
                                assigned_to=manager_id,
                                assigned_level=AgentLevel.MANAGER,
                                parent_task_id=parent_task.task_id
                            )
                            self.tasks[subtask.task_id] = subtask
                            parent_task.subtasks.append(subtask.task_id)
                            tasks.append(subtask)
                    except:
                        continue
            
            return tasks
        
        def execute_hierarchical_workflow(
            self,
            main_task: str,
            priority: TaskPriority = TaskPriority.MEDIUM
        ) -> Dict[str, Any]:
            """
            Execute a complete hierarchical workflow.
            
            Args:
                main_task: Main task description
                priority: Task priority
                
            Returns:
                Workflow results and summary
            """
            print(f"\n[Organization] Executive assigning task: {main_task}")
            print(f"[Organization] Priority: {priority.value}\n")
            
            # Step 1: Decompose at executive level
            tasks = self.decompose_task(main_task, priority)
            
            print(f"[CEO] Decomposed into {len(tasks)-1} manager-level tasks")
            
            # Step 2: Managers delegate to workers
            for task in tasks[1:]:  # Skip parent task
                if task.assigned_level == AgentLevel.MANAGER:
                    manager = self.agents[task.assigned_to]
                    
                    print(f"\n[{manager.agent_id}] Received task: {task.description[:60]}...")
                    
                    # Manager selects appropriate subordinate
                    subordinates = [self.agents[sid] for sid in manager.direct_reports]
                    if subordinates:
                        # Select first available subordinate (simplified)
                        selected = subordinates[0]
                        
                        # Create worker task
                        worker_task = HierarchicalTask(
                            task_id=f"{task.task_id}_w",
                            description=task.description,
                            priority=priority,
                            assigned_to=selected.agent_id,
                            assigned_level=selected.level,
                            parent_task_id=task.task_id
                        )
                        self.tasks[worker_task.task_id] = worker_task
                        task.subtasks.append(worker_task.task_id)
                        
                        # Delegate with instructions
                        delegation = manager.delegate_task(worker_task, selected.agent_id)
                        print(f"[{manager.agent_id}] Delegating to {selected.agent_id}")
                        print(f"   Instructions: {delegation[:80]}...")
                        
                        # Step 3: Workers execute
                        print(f"\n[{selected.agent_id}] Executing task...")
                        result = selected.execute_task(worker_task, delegation)
                        worker_task.result = result
                        worker_task.completed = True
                        
                        print(f"[{selected.agent_id}] Completed: {result[:80]}...")
                        
                        # Step 4: Workers report to manager
                        report = selected.create_report(worker_task, result)
                        self.reports.append(report)
                        
                        print(f"[{selected.agent_id}] Report status: {report.status}")
                        
                        # Step 5: Manager synthesizes results
                        task.result = result
                        task.completed = True
            
            # Step 6: CEO synthesizes executive summary
            print(f"\n[CEO] Synthesizing final results...")
            
            all_results = "\n\n".join([
                f"Manager {t.assigned_to}: {t.result}"
                for t in tasks[1:]
                if t.result
            ])
            
            exec_summary = self._create_executive_summary(main_task, all_results)
            tasks[0].result = exec_summary
            tasks[0].completed = True
            
            return {
                "main_task": main_task,
                "executive_summary": exec_summary,
                "manager_results": {t.assigned_to: t.result for t in tasks[1:] if t.result},
                "reports": self.reports,
                "total_tasks": len(self.tasks),
                "completed_tasks": sum(1 for t in self.tasks.values() if t.completed)
            }
        
        def _create_executive_summary(
            self,
            main_task: str,
            all_results: str
        ) -> str:
            """Create executive summary from all results."""
            ceo = self.agents["CEO"]
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", ceo.get_role_description()),
                ("user", """Synthesize an executive summary from the results below:
    
    Main Task: {main_task}
    
    Team Results:
    {all_results}
    
    Provide a concise executive summary (3-4 sentences) highlighting:
    1. Overall completion status
    2. Key findings
    3. Strategic recommendations""")
            ])
            
            chain = prompt | ceo.llm | StrOutputParser()
            
            summary = chain.invoke({
                "main_task": main_task,
                "all_results": all_results
            })
            
            return summary.strip()
        
        def get_org_chart(self) -> str:
            """Generate a text representation of the org chart."""
            lines = []
            
            def print_agent(agent_id: str, indent: int = 0):
                agent = self.agents[agent_id]
                prefix = "  " * indent
                lines.append(f"{prefix}- {agent_id} ({agent.level.value}) [{agent.specialty or 'General'}]")
                
                for report_id in agent.direct_reports:
                    print_agent(report_id, indent + 1)
            
            print_agent("CEO")
            return "\n".join(lines)
    
    
    def demonstrate_hierarchical_multi_agent():
        """Demonstrate the Hierarchical Multi-Agent pattern."""
        
        print("=" * 80)
        print("HIERARCHICAL MULTI-AGENT PATTERN DEMONSTRATION")
        print("=" * 80)
        
        # Create organization
        org = HierarchicalOrganization()
        
        # Show org chart
        print("\n" + "=" * 80)
        print("ORGANIZATIONAL STRUCTURE")
        print("=" * 80)
        print(org.get_org_chart())
        
        # Test 1: Complex project
        print("\n" + "=" * 80)
        print("TEST 1: Complex Project Execution")
        print("=" * 80)
        
        task1 = "Launch a new mobile app with AI-powered features"
        
        result1 = org.execute_hierarchical_workflow(task1, TaskPriority.HIGH)
        
        print("\n" + "-" * 80)
        print("EXECUTIVE SUMMARY:")
        print("-" * 80)
        print(result1["executive_summary"])
        
        print("\n" + "-" * 80)
        print("DEPARTMENTAL RESULTS:")
        print("-" * 80)
        for manager, result in result1["manager_results"].items():
            print(f"\n{manager}:")
            print(f"  {result[:150]}...")
        
        print("\n" + "-" * 80)
        print("WORKFLOW METRICS:")
        print("-" * 80)
        print(f"  Total tasks created: {result1['total_tasks']}")
        print(f"  Completed tasks: {result1['completed_tasks']}")
        print(f"  Reports submitted: {len(result1['reports'])}")
        
        # Test 2: Strategic initiative
        print("\n" + "=" * 80)
        print("TEST 2: Strategic Initiative")
        print("=" * 80)
        
        task2 = "Improve system security and implement new compliance measures"
        
        result2 = org.execute_hierarchical_workflow(task2, TaskPriority.CRITICAL)
        
        print("\n" + "-" * 80)
        print("EXECUTIVE SUMMARY:")
        print("-" * 80)
        print(result2["executive_summary"])
        
        # Summary
        print("\n" + "=" * 80)
        print("SUMMARY")
        print("=" * 80)
        print("""
    The Hierarchical Multi-Agent pattern demonstrates several key benefits:
    
    1. **Scalability**: Can handle large, complex tasks through decomposition
    2. **Specialization**: Agents focus on their area of expertise
    3. **Clear Accountability**: Well-defined reporting relationships
    4. **Efficient Coordination**: Structured communication channels
    
    Organizational Structure:
    - **Executive Level**: Strategic direction and overall coordination
    - **Manager Level**: Team coordination and task delegation
    - **Worker Level**: Task execution and implementation
    - **Specialist Level**: Domain expertise and specialized guidance
    
    Communication Flow:
    - **Top-Down**: Task delegation with clear instructions
    - **Bottom-Up**: Status reports and results
    - **Cross-Functional**: Coordination between departments
    
    Use Cases:
    - Large-scale project management
    - Complex multi-domain problems
    - Organizational simulation
    - Scalable distributed systems
    
    The pattern is particularly effective when:
    - Tasks require multiple areas of expertise
    - Clear accountability is important
    - Scalability is needed for large projects
    - Organizational modeling is beneficial
    - Structured communication is preferred
    """)
    
    
    if __name__ == "__main__":
        demonstrate_hierarchical_multi_agent()
    
        

    Agentic Design Pattern: Hierarchical Multi Agent

    Pattern 016: Hierarchical Multi Agent – Line-by-Line Explanation

    Overview

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


    Pattern Description

    Pattern 016: Hierarchical Multi-Agent

    Description: The Hierarchical Multi-Agent pattern organizes agents in a hierarchical structure with multiple levels of management and specialization. Managers coordinate workers, workers execute tasks, and specialists provide domain expertise. This creates a scalable organizational structure for complex problems.

    Components: – Manager Agents: High-level coordination and decision-making – Worker Agents: Execute assigned tasks – Specialist Agents: Provide domain expertise – Communication Protocol: Top-down delegation and bottom-up reporting – Task Decomposition: Breaking complex tasks into hierarchical subtasks

    Use Cases: – Large-scale project management – Complex multi-domain problems – Organizational modeling and simulation – Scalable distributed systems

    LangChain Implementation: Uses multiple LLM instances representing different hierarchy levels, implements delegation chains, task assignment strategies, and result aggregation across organizational levels.

    Key Features: – Multi-level hierarchy (Manager -> Worker -> Specialist) – Top-down task delegation – Bottom-up result reporting – Cross-functional coordination


    File Statistics

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

    Module Documentation

    """
    Pattern 016: Hierarchical Multi-Agent
    
    Description:
        The Hierarchical Multi-Agent pattern organizes agents in a hierarchical structure with
        multiple levels of management and specialization. Managers coordinate workers, workers
        execute tasks, and specialists provide domain expertise. This creates a scalable
        organizational structure for complex problems.
    
    Components:
        - Manager Agents: High-level coordination and decision-making
        - Worker Agents: Execute assigned tasks
        - Specialist Agents: Provide domain expertise
        - Communication Protocol: Top-down delegation and bottom-up reporting
        - Task Decomposition: Breaking complex tasks into hierarchical subtasks
    
    Use Cases:
        - Large-scale project management
        - Complex multi-domain problems
        - Organizational modeling and simulation
        - Scalable distributed systems
    
    LangChain Implementation:
        Uses multiple LLM instances representing different hierarchy levels,
        implements delegation chains, task assignment strategies, and result
        aggregation across organizational levels.
    
    Key Features:
        - Multi-level hierarchy (Manager -> Worker -> Specialist)
        - Top-down task delegation
        - Bottom-up result reporting
        - Cross-functional coordination
    """
    

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

    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.Set: Type hints for better code documentation (List, Dict, Any, Optional)

    LangChain Imports

    from langchain_openai.ChatOpenAI
    

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

    Other Imports

    from dotenv.load_dotenv
    

    Purpose: Additional dependencies for specialized functionality.


    Classes

    Class: AgentLevel

    Lines 48-53

    Inherits from: Enum

    Purpose:

    Hierarchy levels in the organization.
    

    Class: TaskPriority

    Lines 56-61

    Inherits from: Enum

    Purpose:

    Task priority levels.
    

    Class: HierarchicalTask

    Lines 65-75

    Purpose:

    Represents a task in the hierarchy.
    

    Class: AgentReport

    Lines 79-87

    Purpose:

    Report from an agent to their manager.
    

    Class: HierarchicalAgent

    Lines 90-274

    Purpose:

    Base class for agents in the hierarchy.
    

    Methods:

    __init__(self, agent_id, level, specialty, model)

    Lines 95-122

    Purpose:

    Initialize a hierarchical agent.
    
    Args:
        agent_id: Unique identifier
        level: Hierarchy level
        specialty: Optional domain specialty
        model: LLM model to use
    

    Parameters: – agent_id: Agent id – level: Level – specialty: Specialty – model: Language model instance

    get_role_description(self)

    Lines 124-132

    Purpose:

    Get the role description for this agent.
    

    Parameters:

    Returns: str

    delegate_task(self, task, subordinate_id)

    Lines 134-170

    Purpose:

    Delegate a task to a subordinate with clear instructions.
    
    Args:
        task: Task to delegate
        subordinate_id: ID of subordinate to delegate to
    
    Returns:
        Delegation message
    

    Parameters: – task: Task – subordinate_id: Subordinate id

    Returns: str

    execute_task(self, task, context)

    Lines 172-208

    Purpose:

    Execute an assigned task.
    
    Args:
        task: Task to execute
        context: Optional context from manager
    
    Returns:
        Task result
    

    Parameters: – task: Task – context: Context

    Returns: str

    create_report(self, task, result)

    Lines 210-274

    Purpose:

    Create a report for manager.
    
    Args:
        task: Completed task
        result: Task result
    
    Returns:
        Agent report
    

    Parameters: – task: Task – result: Result

    Returns: AgentReport


    Class: HierarchicalOrganization

    Lines 277-548

    Purpose:

    Manages a hierarchical multi-agent organization.
    

    Methods:

    __init__(self, model)

    Lines 282-290

    Purpose:

    Initialize the hierarchical organization.
    

    Parameters: – model: Language model instance

    _build_organization(self)

    Lines 292-321

    Purpose:

    Build a default organizational structure.
    

    Parameters:

    add_agent(self, agent_id, level, specialty)

    Lines 323-331

    Purpose:

    Add an agent to the organization.
    

    Parameters: – agent_id: Agent id – level: Level – specialty: Specialty

    set_manager(self, agent_id, manager_id)

    Lines 333-337

    Purpose:

    Set manager-subordinate relationship.
    

    Parameters: – agent_id: Agent id – manager_id: Manager id

    decompose_task(self, task_description, priority)

    Lines 339-408

    Purpose:

    Decompose a high-level task into hierarchical subtasks.
    
    Args:
        task_description: Main task description
        priority: Task priority
    
    Returns:
        List of hierarchical tasks
    

    Parameters: – task_description: Task description – priority: Priority

    Returns: <ast.Subscript object at 0x000001A06D631250>

    execute_hierarchical_workflow(self, main_task, priority)

    Lines 410-501

    Purpose:

    Execute a complete hierarchical workflow.
    
    Args:
        main_task: Main task description
        priority: Task priority
    
    Returns:
        Workflow results and summary
    

    Parameters: – main_task: Main task – priority: Priority

    Returns: <ast.Subscript object at 0x000001A06D63D850>

    _create_executive_summary(self, main_task, all_results)

    Lines 503-533

    Purpose:

    Create executive summary from all results.
    

    Parameters: – main_task: Main task – all_results: All results

    Returns: str

    get_org_chart(self)

    Lines 535-548

    Purpose:

    Generate a text representation of the org chart.
    

    Parameters:

    Returns: str


    Functions

    demonstrate_hierarchical_multi_agent()

    Lines 551-644

    Purpose:

    Demonstrate the Hierarchical Multi-Agent pattern.
    

    Key Concepts Summary

    Pattern Overview

    The Hierarchical Multi Agent 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 015
    • Pattern 017
    • 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 Hierarchical Multi Agent 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: 016_hierarchical_multi_agent.py
    Total Lines: 649
    Pattern Category: AI Agent Design Pattern #016


    Generated from actual implementation code
    For educational and reference purposes

  • 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