Tag: technology

  • Agentic Architecture Pattern: Function Calling

    
        """
    Pattern 93: Function Calling Pattern
    
    Agents directly call Python functions based on natural language requests.
    Uses OpenAI's function calling API to map queries to function invocations.
    """
    
    from typing import Dict, Any, List, Optional, Callable
    from dataclasses import dataclass, field
    from datetime import datetime
    import json
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_openai import ChatOpenAI
    from langchain_core.utils.function_calling import convert_to_openai_function
    
    
    @dataclass
    class FunctionDefinition:
        """Definition of callable function."""
        name: str
        description: str
        function: Callable
        parameters: Dict[str, Any]
        returns: str
    
    
    @dataclass
    class FunctionCall:
        """Record of function call."""
        call_id: str
        function_name: str
        arguments: Dict[str, Any]
        result: Any
        success: bool
        timestamp: datetime = field(default_factory=datetime.now)
        error: Optional[str] = None
    
    
    # Example functions that agents can call
    def calculate_discount(price: float, discount_percent: float) -> float:
        """Calculate final price after discount.
        
        Args:
            price: Original price
            discount_percent: Discount percentage (0-100)
        
        Returns:
            Final price after discount
        """
        return price * (1 - discount_percent / 100)
    
    
    def schedule_meeting(date: str, time: str, attendees: List[str], topic: str) -> str:
        """Schedule a meeting.
        
        Args:
            date: Meeting date (YYYY-MM-DD)
            time: Meeting time (HH:MM)
            attendees: List of attendee names
            topic: Meeting topic
        
        Returns:
            Confirmation message
        """
        return f"Meeting scheduled for {date} at {time} with {len(attendees)} attendees. Topic: {topic}"
    
    
    def send_email(to: str, subject: str, body: str) -> str:
        """Send an email.
        
        Args:
            to: Recipient email address
            subject: Email subject
            body: Email body
        
        Returns:
            Confirmation message
        """
        return f"Email sent to {to} with subject '{subject}'"
    
    
    def query_database(table: str, filters: Dict[str, Any]) -> List[Dict]:
        """Query database table.
        
        Args:
            table: Table name
            filters: Dictionary of filter conditions
        
        Returns:
            List of matching records
        """
        # Simulated database query
        return [
            {"id": 1, "name": "Record 1", "value": 100},
            {"id": 2, "name": "Record 2", "value": 200}
        ]
    
    
    def create_report(report_type: str, data_source: str, format: str = "pdf") -> str:
        """Generate a report.
        
        Args:
            report_type: Type of report (sales, analytics, etc.)
            data_source: Data source name
            format: Output format (pdf, excel, html)
        
        Returns:
            Report generation confirmation
        """
        return f"Generated {report_type} report from {data_source} in {format} format"
    
    
    class FunctionCallingAgent:
        """Agent that can call functions based on natural language."""
        
        def __init__(self, name: str, model_name: str = "gpt-4"):
            self.name = name
            self.llm = ChatOpenAI(model=model_name, temperature=0)
            self.functions: Dict[str, FunctionDefinition] = {}
            self.call_history: List[FunctionCall] = []
        
        def register_function(self, func_def: FunctionDefinition):
            """Register a function that can be called."""
            self.functions[func_def.name] = func_def
        
        def execute_query(self, query: str) -> Dict[str, Any]:
            """Execute natural language query by calling appropriate functions."""
            
            # Prepare function descriptions for LLM
            functions_for_llm = []
            for func_def in self.functions.values():
                functions_for_llm.append({
                    "name": func_def.name,
                    "description": func_def.description,
                    "parameters": func_def.parameters
                })
            
            # Create LLM with function calling
            llm_with_functions = self.llm.bind(functions=functions_for_llm)
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", f"You are {self.name}, an AI assistant that can call functions to help users. "
                          f"Analyze the user's request and call the appropriate function."),
                ("human", "{query}")
            ])
            
            chain = prompt | llm_with_functions
            
            try:
                response = chain.invoke({"query": query})
                
                # Check if function was called
                if hasattr(response, 'additional_kwargs') and 'function_call' in response.additional_kwargs:
                    function_call = response.additional_kwargs['function_call']
                    function_name = function_call['name']
                    arguments = json.loads(function_call['arguments'])
                    
                    # Execute the function
                    if function_name in self.functions:
                        func_def = self.functions[function_name]
                        result = func_def.function(**arguments)
                        
                        # Record call
                        call = FunctionCall(
                            call_id=f"call_{len(self.call_history)}",
                            function_name=function_name,
                            arguments=arguments,
                            result=result,
                            success=True
                        )
                        self.call_history.append(call)
                        
                        return {
                            "success": True,
                            "function_called": function_name,
                            "arguments": arguments,
                            "result": result,
                            "agent": self.name
                        }
                    else:
                        return {
                            "success": False,
                            "error": f"Function {function_name} not registered"
                        }
                else:
                    # No function called, just return text response
                    return {
                        "success": True,
                        "response": str(response.content),
                        "function_called": None
                    }
            
            except Exception as e:
                call = FunctionCall(
                    call_id=f"call_{len(self.call_history)}",
                    function_name="unknown",
                    arguments={},
                    result=None,
                    success=False,
                    error=str(e)
                )
                self.call_history.append(call)
                
                return {
                    "success": False,
                    "error": str(e)
                }
    
    
    class FunctionCallingPattern:
        """
        Function Calling Pattern - Natural language to function execution.
        
        Agents interpret natural language requests and automatically call
        appropriate Python functions with correct parameters.
        """
        
        def __init__(self):
            self.agents: Dict[str, FunctionCallingAgent] = {}
            self.available_functions: Dict[str, FunctionDefinition] = {}
        
        def register_function(self, func: Callable, description: str):
            """Register a function for agents to call."""
            # Extract function signature
            import inspect
            sig = inspect.signature(func)
            
            # Build parameter schema
            parameters = {
                "type": "object",
                "properties": {},
                "required": []
            }
            
            for param_name, param in sig.parameters.items():
                if param_name != 'self':
                    param_type = "string"  # Simplified
                    if param.annotation in [int, float]:
                        param_type = "number"
                    elif param.annotation == bool:
                        param_type = "boolean"
                    elif param.annotation == list:
                        param_type = "array"
                    
                    parameters["properties"][param_name] = {
                        "type": param_type,
                        "description": f"Parameter {param_name}"
                    }
                    
                    if param.default == inspect.Parameter.empty:
                        parameters["required"].append(param_name)
            
            func_def = FunctionDefinition(
                name=func.__name__,
                description=description,
                function=func,
                parameters=parameters,
                returns="Function result"
            )
            
            self.available_functions[func.__name__] = func_def
        
        def create_agent(self, name: str, function_names: Optional[List[str]] = None) -> FunctionCallingAgent:
            """Create agent with access to specified functions."""
            agent = FunctionCallingAgent(name)
            
            # Register functions to agent
            if function_names:
                for func_name in function_names:
                    if func_name in self.available_functions:
                        agent.register_function(self.available_functions[func_name])
            else:
                # Register all functions
                for func_def in self.available_functions.values():
                    agent.register_function(func_def)
            
            self.agents[name] = agent
            return agent
        
        def execute_query(self, agent_name: str, query: str) -> Dict[str, Any]:
            """Execute query with specified agent."""
            if agent_name not in self.agents:
                return {"success": False, "error": "Agent not found"}
            
            agent = self.agents[agent_name]
            result = agent.execute_query(query)
            
            if result.get("success"):
                if result.get("function_called"):
                    print(f"\n   ✓ {agent_name} called {result['function_called']}")
                    print(f"     Arguments: {result['arguments']}")
                    print(f"     Result: {result['result']}")
                else:
                    print(f"\n   ℹ️ {agent_name}: {result.get('response', 'No response')[:80]}")
            else:
                print(f"\n   ✗ {agent_name}: {result.get('error', 'Unknown error')}")
            
            return result
        
        def run(self, scenario: str) -> Dict[str, Any]:
            """Execute function calling pattern."""
            total_calls = sum(len(agent.call_history) for agent in self.agents.values())
            successful_calls = sum(
                sum(1 for call in agent.call_history if call.success)
                for agent in self.agents.values()
            )
            
            return {
                "pattern": "FunctionCalling",
                "total_agents": len(self.agents),
                "available_functions": len(self.available_functions),
                "total_calls": total_calls,
                "successful_calls": successful_calls,
                "scenario": scenario
            }
    
    
    # Example usage
    if __name__ == "__main__":
        print("=== Function Calling Pattern Demo ===\n")
        
        # Create pattern
        pattern = FunctionCallingPattern()
        
        # Register functions
        print("1. Registering callable functions...")
        pattern.register_function(calculate_discount, "Calculate final price after applying discount")
        pattern.register_function(schedule_meeting, "Schedule a meeting with date, time, and attendees")
        pattern.register_function(send_email, "Send an email to a recipient")
        pattern.register_function(query_database, "Query database table with filters")
        pattern.register_function(create_report, "Generate a report from data source")
        
        print(f"   Registered {len(pattern.available_functions)} functions")
        
        # Create agents
        print("\n2. Creating function-calling agents...")
        assistant = pattern.create_agent("VirtualAssistant")
        print(f"   VirtualAssistant: {len(assistant.functions)} functions available")
        
        # Execute natural language queries
        print("\n3. Executing natural language queries...")
        
        queries = [
            "Calculate the final price for an item that costs $100 with a 20% discount",
            "Schedule a meeting for tomorrow at 2pm with Alice, Bob, and Carol to discuss Q4 planning",
            "Send an email to john@example.com with subject 'Project Update' and body 'The project is on track'",
            "Query the customers table and filter by status active",
            "Create a sales report from the Q4 data in PDF format"
        ]
        
        for query in queries:
            print(f"\n   Query: {query}")
            pattern.execute_query("VirtualAssistant", query)
        
        # Summary
        print("\n4. Execution Summary:")
        summary = pattern.run("Virtual assistant with function calling")
        print(f"   Total Functions: {summary['available_functions']}")
        print(f"   Function Calls: {summary['total_calls']}")
        print(f"   Successful: {summary['successful_calls']}")
        
        print("\n5. Call History:")
        for call in assistant.call_history:
            status = "✓" if call.success else "✗"
            print(f"   {status} {call.function_name}: {call.arguments}")
        
        print("\n✓ Function calling pattern executed natural language to function calls!")
    
        

    Agentic Architecture Pattern: Function Calling

    Pattern 93: 93 Function Calling – Line-by-Line Explanation

    Overview

    Pattern 93: Function Calling Pattern

    Agents directly call Python functions based on natural language requests. Uses OpenAI’s function calling API to map queries to function invocations.


    Code Structure Analysis

    File Information

    • Pattern Number: 93
    • Pattern Name: 93 Function Calling
    • File Path: patterns\tool_resource\93_function_calling.py
    • Total Lines: 368

    Detailed Code Explanation

    Module Docstring

    """
    Pattern 93: Function Calling Pattern
    
    Agents directly call Python functions based on natural language requests.
    Uses OpenAI's function calling API to map queries to function invocations.
    """
    

    Explanation: This module docstring describes the purpose and key features of the 93 Function Calling pattern.


    Imports Section

    from typing import Dict, Any, List, Optional, Callable
    from dataclasses import dataclass, field
    from datetime import datetime
    import json
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_openai import ChatOpenAI
    from langchain_core.utils.function_calling import convert_to_openai_function
            import inspect  # First 20 imports shown
    

    Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional, Callable: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime: Date and time utilities – import json: JSON parsing and serialization – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration – from langchain_core.utils.function_calling import convert_to_openai_function: LangChain framework integration


    Classes

    FunctionDefinition (Line 18)

    Description: Definition of callable function.

    FunctionCall (Line 28)

    Description: Record of function call.

    FunctionCallingAgent (Line 113)

    Description: Agent that can call functions based on natural language.

    Methods:

    __init__ (Line 116)
    def __init__(self, name, model_name):
        """No method docstring."""
    

    Purpose: No method docstring.

    Parameters: – self: Instance reference – name: Identifier or label – model_name: LLM model to use (e.g., “gpt-4”)

    register_function (Line 122)
    def register_function(self, func_def):
        """Register a function that can be called."""
    

    Purpose: Register a function that can be called.

    Parameters: – self: Instance reference – func_def: Parameter for func_def

    execute_query (Line 126)
    def execute_query(self, query):
        """Execute natural language query by calling appropriate functions."""
    

    Purpose: Execute natural language query by calling appropriate functions.

    Parameters: – self: Instance reference – query: Parameter for query

    FunctionCallingPattern (Line 210)

    Description: Function Calling Pattern – Natural language to function execution.

    Agents interpret natural language requests and automatically call appropriate Python functions with correct parameters.

    Methods:

    __init__ (Line 218)
    def __init__(self):
        """No method docstring."""
    

    Purpose: No method docstring.

    Parameters: – self: Instance reference

    register_function (Line 222)
    def register_function(self, func, description):
        """Register a function for agents to call."""
    

    Purpose: Register a function for agents to call.

    Parameters: – self: Instance reference – func: Parameter for func – description: Parameter for description

    create_agent (Line 263)
    def create_agent(self, name, function_names):
        """Create agent with access to specified functions."""
    

    Purpose: Create agent with access to specified functions.

    Parameters: – self: Instance reference – name: Identifier or label – function_names: Parameter for function_names

    execute_query (Line 280)
    def execute_query(self, agent_name, query):
        """Execute query with specified agent."""
    

    Purpose: Execute query with specified agent.

    Parameters: – self: Instance reference – agent_name: Parameter for agent_name – query: Parameter for query

    run (Line 300)
    def run(self, scenario):
        """Execute function calling pattern."""
    

    Purpose: Execute function calling pattern.

    Parameters: – self: Instance reference – scenario: Parameter for scenario


    Functions

    calculate_discount (Line 40)

    def calculate_discount(price, discount_percent):
        """Calculate final price after discount.
    
    Args:
        price: Original price
        discount_percent: Discount percentage (0-100)
    
    Returns:
        Final price after discount"""
    

    Purpose: Calculate final price after discount.

    Args: price: Original price discount_percent: Discount percentage (0-100)

    Returns: Final price after discount

    schedule_meeting (Line 53)

    def schedule_meeting(date, time, attendees, topic):
        """Schedule a meeting.
    
    Args:
        date: Meeting date (YYYY-MM-DD)
        time: Meeting time (HH:MM)
        attendees: List of attendee names
        topic: Meeting topic
    
    Returns:
        Confirmation message"""
    

    Purpose: Schedule a meeting.

    Args: date: Meeting date (YYYY-MM-DD) time: Meeting time (HH:MM) attendees: List of attendee names topic: Meeting topic

    Returns: Confirmation message

    send_email (Line 68)

    def send_email(to, subject, body):
        """Send an email.
    
    Args:
        to: Recipient email address
        subject: Email subject
        body: Email body
    
    Returns:
        Confirmation message"""
    

    Purpose: Send an email.

    Args: to: Recipient email address subject: Email subject body: Email body

    Returns: Confirmation message

    query_database (Line 82)

    def query_database(table, filters):
        """Query database table.
    
    Args:
        table: Table name
        filters: Dictionary of filter conditions
    
    Returns:
        List of matching records"""
    

    Purpose: Query database table.

    Args: table: Table name filters: Dictionary of filter conditions

    Returns: List of matching records

    create_report (Line 99)

    def create_report(report_type, data_source, format):
        """Generate a report.
    
    Args:
        report_type: Type of report (sales, analytics, etc.)
        data_source: Data source name
        format: Output format (pdf, excel, html)
    
    Returns:
        Report generation confirmation"""
    

    Purpose: Generate a report.

    Args: report_type: Type of report (sales, analytics, etc.) data_source: Data source name format: Output format (pdf, excel, html)

    Returns: Report generation confirmation


    Design Patterns Used

    • Strategy Pattern: Different execution strategies for various scenarios
    • State Pattern: Managing different states in the workflow
    • Chain of Responsibility: Sequential processing through workflow nodes
    • Observer Pattern: Tracking and monitoring pattern execution
    • Factory Pattern: Creating instances with proper configuration

    Usage Example

    # Initialize the pattern
    pattern = FunctionDefinition(name="example", model_name="gpt-4")
    
    # Configure and execute
    result = pattern.execute(data={"key": "value"})
    
    # Check results
    print(result)
    

    Key Concepts

    • LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
    • Workflow Orchestration: LangGraph manages execution flow through nodes
    • State Management: TypedDict defines workflow state structure
    • Error Handling: Comprehensive error handling and fallback mechanisms
    • Extensibility: Designed for easy customization and extension

    Extension Points

    This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities

  • Agentic Architecture Pattern: Tool Augmented

    
        """
    Pattern 92: Tool Augmented
    
    Purpose: Enhanced implementation with full LangChain/LangGraph integration
    """
    
    from typing import Dict, Any, List, Optional
    from dataclasses import dataclass, field
    from datetime import datetime
    from enum import Enum
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_openai import ChatOpenAI
    from langgraph.graph import StateGraph, END
    from typing_extensions import TypedDict
    
    
    @dataclass
    class ToolAugmentedConfig:
        """Configuration for Tool Augmented."""
        model_name: str = "gpt-4"
        temperature: float = 0.7
        max_iterations: int = 10
        enable_logging: bool = True
    
    
    @dataclass
    class ToolAugmentedState:
        """State information for Tool Augmented."""
        current_step: int = 0
        data: Dict[str, Any] = field(default_factory=dict)
        results: List[Any] = field(default_factory=list)
        metadata: Dict[str, Any] = field(default_factory=dict)
        timestamp: datetime = field(default_factory=datetime.now)
    
    
    class ToolAugmentedAgent:
        """Agent implementing Tool Augmented."""
        
        def __init__(self, name: str, config: ToolAugmentedConfig = None):
            self.name = name
            self.config = config or ToolAugmentedConfig()
            self.llm = ChatOpenAI(
                model=self.config.model_name,
                temperature=self.config.temperature
            )
            self.state = ToolAugmentedState()
            self.execution_history: List[Dict[str, Any]] = []
        
        def process(self, input_data: Any) -> Dict[str, Any]:
            """Process input according to Tool Augmented."""
            self.state.current_step += 1
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", f"You are {self.name}, implementing Tool Augmented. "
                          f"Provide clear, structured responses."),
                ("human", "Process this input: {input_data}")
            ])
            
            chain = prompt | self.llm
            response = chain.invoke({"input_data": str(input_data)})
            
            result = {
                "agent": self.name,
                "step": self.state.current_step,
                "input": input_data,
                "output": str(response.content),
                "timestamp": datetime.now()
            }
            
            self.execution_history.append(result)
            self.state.results.append(result)
            
            return result
        
        def get_statistics(self) -> Dict[str, Any]:
            """Get agent statistics."""
            return {
                "name": self.name,
                "total_executions": len(self.execution_history),
                "current_step": self.state.current_step,
                "results_count": len(self.state.results)
            }
    
    
    class ToolAugmentedPattern:
        """
        Tool Augmented - Enhanced implementation
        
        This pattern provides comprehensive functionality for tool augmented
        with full LangChain/LangGraph integration.
        
        Features:
        - Multi-agent coordination
        - State management
        - Execution tracking
        - LangGraph workflow integration
        """
        
        def __init__(self, config: ToolAugmentedConfig = None):
            self.config = config or ToolAugmentedConfig()
            self.agents: Dict[str, ToolAugmentedAgent] = {}
            self.workflow = self._build_workflow()
            self.execution_log: List[Dict[str, Any]] = []
        
        def _build_workflow(self) -> StateGraph:
            """Build LangGraph workflow for pattern execution."""
            
            class WorkflowState(TypedDict):
                input: str
                processed: str
                output: str
                metadata: Dict[str, Any]
            
            def process_input(state: WorkflowState) -> WorkflowState:
                """Process input node."""
                return {
                    **state,
                    "processed": f"Processed: {state['input']}",
                    "metadata": {"step": "process_input"}
                }
            
            def generate_output(state: WorkflowState) -> WorkflowState:
                """Generate output node."""
                llm = ChatOpenAI(model=self.config.model_name, temperature=self.config.temperature)
                
                prompt = ChatPromptTemplate.from_messages([
                    ("system", "You are implementing Tool Augmented. Generate comprehensive output."),
                    ("human", "{input}")
                ])
                
                chain = prompt | llm
                response = chain.invoke({"input": state["processed"]})
                
                return {
                    **state,
                    "output": str(response.content),
                    "metadata": {**state.get("metadata", {}), "step": "generate_output"}
                }
            
            # Build workflow graph
            workflow = StateGraph(WorkflowState)
            workflow.add_node("process", process_input)
            workflow.add_node("generate", generate_output)
            workflow.set_entry_point("process")
            workflow.add_edge("process", "generate")
            workflow.add_edge("generate", END)
            
            return workflow.compile()
        
        def register_agent(self, agent: ToolAugmentedAgent):
            """Register an agent with the pattern."""
            self.agents[agent.name] = agent
            if self.config.enable_logging:
                print(f"Registered agent: {agent.name}")
        
        def execute(self, input_data: Any, agent_name: str = None) -> Dict[str, Any]:
            """Execute the pattern with given input."""
            
            # Use workflow if no specific agent requested
            if agent_name is None:
                result = self.workflow.invoke({
                    "input": str(input_data),
                    "processed": "",
                    "output": "",
                    "metadata": {}
                })
                
                execution_record = {
                    "pattern": "Tool Augmented",
                    "input": input_data,
                    "output": result.get("output", ""),
                    "metadata": result.get("metadata", {}),
                    "timestamp": datetime.now()
                }
                
                self.execution_log.append(execution_record)
                return execution_record
            
            # Execute with specific agent
            agent = self.agents.get(agent_name)
            if not agent:
                raise ValueError(f"Agent {agent_name} not found")
            
            result = agent.process(input_data)
            self.execution_log.append(result)
            return result
        
        def run(self, input_data: Any) -> Dict[str, Any]:
            """Main execution method."""
            return self.execute(input_data)
        
        def get_statistics(self) -> Dict[str, Any]:
            """Get pattern statistics."""
            return {
                "pattern": "Tool Augmented",
                "total_executions": len(self.execution_log),
                "registered_agents": len(self.agents),
                "agents": [agent.get_statistics() for agent in self.agents.values()]
            }
    
    
    # Example usage
    if __name__ == "__main__":
        print("=== Tool Augmented Demo ===\n")
        
        # Create configuration
        config = ToolAugmentedConfig(
            model_name="gpt-4",
            temperature=0.7,
            enable_logging=True
        )
        
        # Create pattern
        pattern = ToolAugmentedPattern(config)
        
        # Create and register agents
        agent1 = ToolAugmentedAgent("Agent1", config)
        agent2 = ToolAugmentedAgent("Agent2", config)
        pattern.register_agent(agent1)
        pattern.register_agent(agent2)
        
        print("\nExecuting pattern workflow...")
        
        # Execute with workflow
        result1 = pattern.run("Demonstrate tool augmented functionality")
        print(f"\nWorkflow Result:")
        print(f"  Output: {result1['output'][:150]}...")
        
        # Execute with specific agents
        result2 = pattern.execute("Specific task for Agent1", agent_name="Agent1")
        print(f"\nAgent1 Result:")
        print(f"  Output: {result2['output'][:150]}...")
        
        result3 = pattern.execute("Specific task for Agent2", agent_name="Agent2")
        print(f"\nAgent2 Result:")
        print(f"  Output: {result3['output'][:150]}...")
        
        # Show statistics
        print("\n=== Statistics ===")
        stats = pattern.get_statistics()
        print(f"Pattern: {stats['pattern']}")
        print(f"Total Executions: {stats['total_executions']}")
        print(f"Registered Agents: {stats['registered_agents']}")
        print(f"\nAgent Statistics:")
        for agent_stats in stats['agents']:
            print(f"  {agent_stats['name']}: {agent_stats['total_executions']} executions")
    
        

    Agentic Architecture Pattern: Tool Augmented

    Pattern 92: 92 Tool Augmented – Line-by-Line Explanation

    Overview

    Pattern 92: Tool Augmented

    Purpose: Enhanced implementation with full LangChain/LangGraph integration


    Code Structure Analysis

    File Information

    • Pattern Number: 92
    • Pattern Name: 92 Tool Augmented
    • File Path: patterns\tool_resource\92_tool_augmented.py
    • Total Lines: 247

    Detailed Code Explanation

    Module Docstring

    """
    Pattern 92: Tool Augmented
    
    Purpose: Enhanced implementation with full LangChain/LangGraph integration
    """
    

    Explanation: This module docstring describes the purpose and key features of the 92 Tool Augmented pattern.


    Imports Section

    from typing import Dict, Any, List, Optional
    from dataclasses import dataclass, field
    from datetime import datetime
    from enum import Enum
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_openai import ChatOpenAI
    from langgraph.graph import StateGraph, END
    from typing_extensions import TypedDict  # First 20 imports shown
    

    Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime: Date and time utilities – from enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration – from langgraph.graph import StateGraph, END: LangGraph for workflow orchestration – from typing_extensions import TypedDict: Type hints for better code documentation


    Classes

    ToolAugmentedConfig (Line 18)

    Description: Configuration for Tool Augmented.

    ToolAugmentedState (Line 27)

    Description: State information for Tool Augmented.

    ToolAugmentedAgent (Line 36)

    Description: Agent implementing Tool Augmented.

    Methods:

    __init__ (Line 39)
    def __init__(self, name, config):
        """No method docstring."""
    

    Purpose: No method docstring.

    Parameters: – self: Instance reference – name: Identifier or label – config: Configuration settings

    process (Line 49)
    def process(self, input_data):
        """Process input according to Tool Augmented."""
    

    Purpose: Process input according to Tool Augmented.

    Parameters: – self: Instance reference – input_data: Parameter for input_data

    get_statistics (Line 75)
    def get_statistics(self):
        """Get agent statistics."""
    

    Purpose: Get agent statistics.

    Parameters: – self: Instance reference

    ToolAugmentedPattern (Line 85)

    Description: Tool Augmented – Enhanced implementation

    This pattern provides comprehensive functionality for tool augmented with full LangChain/LangGraph integration.

    Features: – Multi-agent coordination – State management – Execution tracking – LangGraph workflow integration

    Methods:

    __init__ (Line 99)
    def __init__(self, config):
        """No method docstring."""
    

    Purpose: No method docstring.

    Parameters: – self: Instance reference – config: Configuration settings

    _build_workflow (Line 105)
    def _build_workflow(self):
        """Build LangGraph workflow for pattern execution."""
    

    Purpose: Build LangGraph workflow for pattern execution.

    Parameters: – self: Instance reference

    register_agent (Line 150)
    def register_agent(self, agent):
        """Register an agent with the pattern."""
    

    Purpose: Register an agent with the pattern.

    Parameters: – self: Instance reference – agent: Parameter for agent

    execute (Line 156)
    def execute(self, input_data, agent_name):
        """Execute the pattern with given input."""
    

    Purpose: Execute the pattern with given input.

    Parameters: – self: Instance reference – input_data: Parameter for input_data – agent_name: Parameter for agent_name

    run (Line 188)
    def run(self, input_data):
        """Main execution method."""
    

    Purpose: Main execution method.

    Parameters: – self: Instance reference – input_data: Parameter for input_data

    get_statistics (Line 192)
    def get_statistics(self):
        """Get pattern statistics."""
    

    Purpose: Get pattern statistics.

    Parameters: – self: Instance reference

    WorkflowState (Line 108)

    Description: No class docstring.


    Design Patterns Used

    • Strategy Pattern: Different execution strategies for various scenarios
    • State Pattern: Managing different states in the workflow
    • Chain of Responsibility: Sequential processing through workflow nodes
    • Observer Pattern: Tracking and monitoring pattern execution
    • Factory Pattern: Creating instances with proper configuration

    Usage Example

    # Initialize the pattern
    pattern = ToolAugmentedConfig(name="example", model_name="gpt-4")
    
    # Configure and execute
    result = pattern.execute(data={"key": "value"})
    
    # Check results
    print(result)
    

    Key Concepts

    • LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
    • Workflow Orchestration: LangGraph manages execution flow through nodes
    • State Management: TypedDict defines workflow state structure
    • Error Handling: Comprehensive error handling and fallback mechanisms
    • Extensibility: Designed for easy customization and extension

    Extension Points

    This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities

  • Agentic Architecture Pattern: Tool Using Agent

    
        """
    Pattern 91: Tool-Using Agent Pattern
    
    Agents that can use external tools to extend their capabilities.
    Integrates with LangChain's tool system for dynamic tool selection and execution.
    """
    
    from typing import Dict, Any, List, Optional, Callable
    from dataclasses import dataclass, field
    from datetime import datetime
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_openai import ChatOpenAI
    from langchain.agents import AgentExecutor, create_openai_tools_agent
    from langchain_core.tools import Tool, StructuredTool
    from langchain.tools import tool
    
    
    @dataclass
    class ToolExecution:
        """Record of tool execution."""
        tool_name: str
        input_data: Any
        output: Any
        success: bool
        timestamp: datetime = field(default_factory=datetime.now)
        error: Optional[str] = None
    
    
    # Define some example tools
    @tool
    def calculator(expression: str) -> str:
        """Evaluate a mathematical expression. Input should be a valid Python expression."""
        try:
            result = eval(expression)
            return f"Result: {result}"
        except Exception as e:
            return f"Error: {str(e)}"
    
    
    @tool
    def web_search(query: str) -> str:
        """Search the web for information. Returns simulated search results."""
        # In real implementation, would call actual search API
        return f"Search results for '{query}': [Simulated] Found 3 relevant articles about {query}."
    
    
    @tool
    def get_weather(location: str) -> str:
        """Get current weather for a location."""
        # Simulated weather data
        return f"Weather in {location}: Sunny, 72°F (22°C), Light breeze"
    
    
    @tool
    def text_analyzer(text: str) -> str:
        """Analyze text and return statistics."""
        words = text.split()
        chars = len(text)
        return f"Analysis: {len(words)} words, {chars} characters, avg word length: {chars/len(words):.1f}"
    
    
    class ToolUsingAgent:
        """Agent that can use tools to accomplish tasks."""
        
        def __init__(self, name: str, tools: List[Tool], model_name: str = "gpt-4"):
            self.name = name
            self.tools = tools
            self.llm = ChatOpenAI(model=model_name, temperature=0)
            self.execution_history: List[ToolExecution] = []
            
            # Create prompt for tool-using agent
            self.prompt = ChatPromptTemplate.from_messages([
                ("system", f"You are {name}, an AI assistant with access to tools. "
                          f"Use the available tools to help answer questions accurately. "
                          f"Available tools: {', '.join([t.name for t in tools])}"),
                ("human", "{input}"),
                ("placeholder", "{agent_scratchpad}")
            ])
            
            # Create agent
            self.agent = create_openai_tools_agent(self.llm, self.tools, self.prompt)
            self.agent_executor = AgentExecutor(
                agent=self.agent,
                tools=self.tools,
                verbose=False,
                max_iterations=5
            )
        
        def execute_task(self, task: str) -> Dict[str, Any]:
            """Execute task using available tools."""
            try:
                result = self.agent_executor.invoke({"input": task})
                
                # Record execution
                execution = ToolExecution(
                    tool_name="agent_executor",
                    input_data=task,
                    output=result.get("output", ""),
                    success=True
                )
                self.execution_history.append(execution)
                
                return {
                    "success": True,
                    "agent": self.name,
                    "task": task,
                    "result": result.get("output", ""),
                    "tools_used": self._extract_tools_used(result)
                }
            
            except Exception as e:
                execution = ToolExecution(
                    tool_name="agent_executor",
                    input_data=task,
                    output=None,
                    success=False,
                    error=str(e)
                )
                self.execution_history.append(execution)
                
                return {
                    "success": False,
                    "agent": self.name,
                    "task": task,
                    "error": str(e)
                }
        
        def _extract_tools_used(self, result: Dict) -> List[str]:
            """Extract which tools were used from result."""
            # Simplified - in real implementation would parse intermediate steps
            tools_mentioned = []
            output = str(result.get("output", ""))
            for tool in self.tools:
                if tool.name.lower() in output.lower():
                    tools_mentioned.append(tool.name)
            return tools_mentioned
    
    
    class ToolUsingAgentPattern:
        """
        Tool-Using Agent Pattern - Extend agent capabilities with tools.
        
        Agents can dynamically select and use tools based on task requirements.
        Integrates with LangChain's tool system for seamless tool calling.
        """
        
        def __init__(self):
            self.agents: Dict[str, ToolUsingAgent] = {}
            self.available_tools: List[Tool] = []
        
        def register_tool(self, tool: Tool):
            """Register a tool for agents to use."""
            self.available_tools.append(tool)
        
        def create_agent(self, name: str, tool_names: Optional[List[str]] = None) -> ToolUsingAgent:
            """Create agent with specified tools."""
            # Select tools
            if tool_names:
                agent_tools = [t for t in self.available_tools if t.name in tool_names]
            else:
                agent_tools = self.available_tools.copy()
            
            agent = ToolUsingAgent(name, agent_tools)
            self.agents[name] = agent
            
            return agent
        
        def execute_with_agent(self, agent_name: str, task: str) -> Dict[str, Any]:
            """Execute task with specific agent."""
            if agent_name not in self.agents:
                return {"success": False, "error": "Agent not found"}
            
            agent = self.agents[agent_name]
            result = agent.execute_task(task)
            
            if result["success"]:
                print(f"\n   ✓ {agent_name}: {task[:60]}...")
                print(f"     Result: {result['result'][:100]}...")
                if result.get("tools_used"):
                    print(f"     Tools used: {', '.join(result['tools_used'])}")
            else:
                print(f"\n   ✗ {agent_name}: {result.get('error', 'Unknown error')}")
            
            return result
        
        def run(self, scenario: str) -> Dict[str, Any]:
            """Execute tool-using agent pattern."""
            total_executions = sum(len(agent.execution_history) for agent in self.agents.values())
            successful_executions = sum(
                sum(1 for e in agent.execution_history if e.success)
                for agent in self.agents.values()
            )
            
            return {
                "pattern": "ToolUsingAgent",
                "total_agents": len(self.agents),
                "available_tools": len(self.available_tools),
                "total_executions": total_executions,
                "successful_executions": successful_executions,
                "scenario": scenario
            }
    
    
    # Example usage
    if __name__ == "__main__":
        print("=== Tool-Using Agent Pattern Demo ===\n")
        
        # Create pattern
        pattern = ToolUsingAgentPattern()
        
        # Register tools
        print("1. Registering tools...")
        pattern.register_tool(calculator)
        pattern.register_tool(web_search)
        pattern.register_tool(get_weather)
        pattern.register_tool(text_analyzer)
        
        print(f"   Registered {len(pattern.available_tools)} tools")
        
        # Create agents with different tool sets
        print("\n2. Creating tool-using agents...")
        
        # General agent with all tools
        general_agent = pattern.create_agent("GeneralAssistant")
        print(f"   - GeneralAssistant: {len(general_agent.tools)} tools")
        
        # Math specialist with calculator
        math_agent = pattern.create_agent("MathSpecialist", ["calculator"])
        print(f"   - MathSpecialist: {len(math_agent.tools)} tools")
        
        # Research agent with search
        research_agent = pattern.create_agent("ResearchAssistant", ["web_search", "text_analyzer"])
        print(f"   - ResearchAssistant: {len(research_agent.tools)} tools")
        
        # Execute tasks
        print("\n3. Executing tasks with tool-using agents...")
        
        # Math task
        pattern.execute_with_agent(
            "MathSpecialist",
            "Calculate the result of (123 + 456) * 2"
        )
        
        # Weather task
        pattern.execute_with_agent(
            "GeneralAssistant",
            "What's the weather like in San Francisco?"
        )
        
        # Research task
        pattern.execute_with_agent(
            "ResearchAssistant",
            "Search for information about artificial intelligence and analyze the query"
        )
        
        # Complex task requiring multiple tools
        pattern.execute_with_agent(
            "GeneralAssistant",
            "Calculate 15% tip on a $85 dinner bill, then tell me the weather in New York"
        )
        
        # Summary
        print("\n4. Pattern Summary:")
        summary = pattern.run("Multi-agent tool-using system")
        print(f"   Total Agents: {summary['total_agents']}")
        print(f"   Available Tools: {summary['available_tools']}")
        print(f"   Total Executions: {summary['total_executions']}")
        print(f"   Successful: {summary['successful_executions']}")
        
        print("\n✓ Tool-using agents successfully extended capabilities with tools!")
    
        

    Agentic Architecture Pattern: Tool Using Agent

    Pattern 91: 91 Tool Using Agent – Line-by-Line Explanation

    Overview

    Pattern 91: Tool-Using Agent Pattern

    Agents that can use external tools to extend their capabilities. Integrates with LangChain’s tool system for dynamic tool selection and execution.


    Code Structure Analysis

    File Information

    • Pattern Number: 91
    • Pattern Name: 91 Tool Using Agent
    • File Path: patterns\tool_resource\91_tool_using_agent.py
    • Total Lines: 271

    Detailed Code Explanation

    Module Docstring

    """
    Pattern 91: Tool-Using Agent Pattern
    
    Agents that can use external tools to extend their capabilities.
    Integrates with LangChain's tool system for dynamic tool selection and execution.
    """
    

    Explanation: This module docstring describes the purpose and key features of the 91 Tool Using Agent pattern.


    Imports Section

    from typing import Dict, Any, List, Optional, Callable
    from dataclasses import dataclass, field
    from datetime import datetime
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_openai import ChatOpenAI
    from langchain.agents import AgentExecutor, create_openai_tools_agent
    from langchain_core.tools import Tool, StructuredTool
    from langchain.tools import tool  # First 20 imports shown
    

    Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional, Callable: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime: Date and time utilities – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration – from langchain.agents import AgentExecutor, create_openai_tools_agent: LangChain framework integration – from langchain_core.tools import Tool, StructuredTool: LangChain framework integration – from langchain.tools import tool: LangChain framework integration


    Classes

    ToolExecution (Line 19)

    Description: Record of tool execution.

    ToolUsingAgent (Line 62)

    Description: Agent that can use tools to accomplish tasks.

    Methods:

    __init__ (Line 65)
    def __init__(self, name, tools, model_name):
        """No method docstring."""
    

    Purpose: No method docstring.

    Parameters: – self: Instance reference – name: Identifier or label – tools: Parameter for tools – model_name: LLM model to use (e.g., “gpt-4”)

    execute_task (Line 89)
    def execute_task(self, task):
        """Execute task using available tools."""
    

    Purpose: Execute task using available tools.

    Parameters: – self: Instance reference – task: Parameter for task

    _extract_tools_used (Line 128)
    def _extract_tools_used(self, result):
        """Extract which tools were used from result."""
    

    Purpose: Extract which tools were used from result.

    Parameters: – self: Instance reference – result: Parameter for result

    ToolUsingAgentPattern (Line 139)

    Description: Tool-Using Agent Pattern – Extend agent capabilities with tools.

    Agents can dynamically select and use tools based on task requirements. Integrates with LangChain’s tool system for seamless tool calling.

    Methods:

    __init__ (Line 147)
    def __init__(self):
        """No method docstring."""
    

    Purpose: No method docstring.

    Parameters: – self: Instance reference

    register_tool (Line 151)
    def register_tool(self, tool):
        """Register a tool for agents to use."""
    

    Purpose: Register a tool for agents to use.

    Parameters: – self: Instance reference – tool: Parameter for tool

    create_agent (Line 155)
    def create_agent(self, name, tool_names):
        """Create agent with specified tools."""
    

    Purpose: Create agent with specified tools.

    Parameters: – self: Instance reference – name: Identifier or label – tool_names: Parameter for tool_names

    execute_with_agent (Line 168)
    def execute_with_agent(self, agent_name, task):
        """Execute task with specific agent."""
    

    Purpose: Execute task with specific agent.

    Parameters: – self: Instance reference – agent_name: Parameter for agent_name – task: Parameter for task

    run (Line 186)
    def run(self, scenario):
        """Execute tool-using agent pattern."""
    

    Purpose: Execute tool-using agent pattern.

    Parameters: – self: Instance reference – scenario: Parameter for scenario


    Functions

    calculator (Line 31)

    def calculator(expression):
        """Evaluate a mathematical expression. Input should be a valid Python expression."""
    

    Purpose: Evaluate a mathematical expression. Input should be a valid Python expression.

    web_search (Line 41)

    def web_search(query):
        """Search the web for information. Returns simulated search results."""
    

    Purpose: Search the web for information. Returns simulated search results.

    get_weather (Line 48)

    def get_weather(location):
        """Get current weather for a location."""
    

    Purpose: Get current weather for a location.

    text_analyzer (Line 55)

    def text_analyzer(text):
        """Analyze text and return statistics."""
    

    Purpose: Analyze text and return statistics.


    Design Patterns Used

    • Strategy Pattern: Different execution strategies for various scenarios
    • State Pattern: Managing different states in the workflow
    • Chain of Responsibility: Sequential processing through workflow nodes
    • Observer Pattern: Tracking and monitoring pattern execution
    • Factory Pattern: Creating instances with proper configuration

    Usage Example

    # Initialize the pattern
    pattern = ToolExecution(name="example", model_name="gpt-4")
    
    # Configure and execute
    result = pattern.execute(data={"key": "value"})
    
    # Check results
    print(result)
    

    Key Concepts

    • LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
    • Workflow Orchestration: LangGraph manages execution flow through nodes
    • State Management: TypedDict defines workflow state structure
    • Error Handling: Comprehensive error handling and fallback mechanisms
    • Extensibility: Designed for easy customization and extension

    Extension Points

    This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities

  • Agentic Architecture Pattern: Delegate Agent

    
        """
    Pattern 90: Delegate Agent Pattern
    
    Delegates work to specialized worker agents based on task requirements.
    Features task routing, load balancing, and result aggregation.
    """
    
    from typing import Dict, Any, List, Optional, Tuple
    from dataclasses import dataclass, field
    from datetime import datetime
    from enum import Enum
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_openai import ChatOpenAI
    
    
    class TaskPriority(Enum):
        """Task priority levels."""
        LOW = 1
        MEDIUM = 2
        HIGH = 3
        CRITICAL = 4
    
    
    @dataclass
    class Task:
        """Represents a task to be delegated."""
        task_id: str
        description: str
        required_skills: List[str]
        priority: TaskPriority = TaskPriority.MEDIUM
        params: Dict[str, Any] = field(default_factory=dict)
        timestamp: datetime = field(default_factory=datetime.now)
    
    
    @dataclass
    class WorkerAgent:
        """Worker agent that can execute tasks."""
        name: str
        skills: List[str]
        capacity: int = 5
        current_load: int = 0
        tasks_completed: int = 0
        
        def can_accept(self) -> bool:
            """Check if worker can accept more tasks."""
            return self.current_load  bool:
            """Check if worker has required skills."""
            return all(skill in self.skills for skill in required_skills)
        
        def match_score(self, required_skills: List[str]) -> float:
            """Calculate match score (0.0 to 1.0)."""
            if not required_skills:
                return 1.0
            matching = sum(1 for skill in required_skills if skill in self.skills)
            return matching / len(required_skills)
    
    
    @dataclass
    class DelegationRecord:
        """Records a delegation event."""
        task: Task
        worker: str
        start_time: datetime = field(default_factory=datetime.now)
        end_time: Optional[datetime] = None
        result: Optional[str] = None
        success: bool = False
    
    
    class DelegateAgent:
        """Agent that delegates tasks to workers."""
        
        def __init__(self, name: str, model_name: str = "gpt-4"):
            self.name = name
            self.llm = ChatOpenAI(model=model_name, temperature=0.3)
            self.workers: Dict[str, WorkerAgent] = {}
            self.delegation_history: List[DelegationRecord] = []
            self.pending_tasks: List[Task] = []
        
        def register_worker(self, worker: WorkerAgent):
            """Register a worker agent."""
            self.workers[worker.name] = worker
            print(f"   {self.name}: Registered worker '{worker.name}' with skills: {', '.join(worker.skills)}")
        
        def find_best_worker(self, task: Task) -> Optional[WorkerAgent]:
            """Find the best worker for a task."""
            candidates = []
            
            for worker in self.workers.values():
                if worker.can_accept() and worker.has_skills(task.required_skills):
                    score = worker.match_score(task.required_skills)
                    # Consider load in scoring
                    load_factor = 1.0 - (worker.current_load / worker.capacity)
                    final_score = score * 0.7 + load_factor * 0.3
                    candidates.append((worker, final_score))
            
            if not candidates:
                return None
            
            # Sort by score (higher is better)
            candidates.sort(key=lambda x: x[1], reverse=True)
            return candidates[0][0]
        
        def delegate_task(self, task: Task) -> Optional[str]:
            """Delegate task to best available worker."""
            print(f"\n   {self.name}: Delegating task '{task.task_id}'")
            print(f"   Description: {task.description}")
            print(f"   Required skills: {', '.join(task.required_skills)}")
            print(f"   Priority: {task.priority.name}")
            
            # Find best worker
            worker = self.find_best_worker(task)
            
            if not worker:
                print(f"   {self.name}: No suitable worker found")
                self.pending_tasks.append(task)
                return None
            
            print(f"   {self.name}: Assigned to '{worker.name}' (match: {worker.match_score(task.required_skills):.2f})")
            
            # Update worker load
            worker.current_load += 1
            
            # Execute task via worker
            result = self._execute_task_via_worker(task, worker)
            
            # Update worker
            worker.current_load -= 1
            worker.tasks_completed += 1
            
            # Record delegation
            record = DelegationRecord(
                task=task,
                worker=worker.name,
                end_time=datetime.now(),
                result=result,
                success=True
            )
            self.delegation_history.append(record)
            
            print(f"   {self.name}: Task completed by '{worker.name}'")
            return result
        
        def _execute_task_via_worker(self, task: Task, worker: WorkerAgent) -> str:
            """Execute task through worker agent."""
            prompt = ChatPromptTemplate.from_messages([
                ("system", f"You are {worker.name}, a worker agent with skills: {', '.join(worker.skills)}. "
                          f"Execute the delegated task."),
                ("human", "Task: {description}\nParameters: {params}\n\nExecute this task.")
            ])
            
            chain = prompt | self.llm
            response = chain.invoke({
                "description": task.description,
                "params": str(task.params)
            })
            
            return response.content
        
        def process_pending_tasks(self):
            """Try to process pending tasks."""
            if not self.pending_tasks:
                return
            
            print(f"\n   {self.name}: Processing {len(self.pending_tasks)} pending tasks...")
            
            processed = []
            for task in self.pending_tasks:
                result = self.delegate_task(task)
                if result:
                    processed.append(task)
            
            # Remove processed tasks
            for task in processed:
                self.pending_tasks.remove(task)
        
        def get_statistics(self) -> Dict[str, Any]:
            """Get delegation statistics."""
            total_delegations = len(self.delegation_history)
            successful = sum(1 for d in self.delegation_history if d.success)
            
            worker_stats = {
                name: {
                    "tasks_completed": worker.tasks_completed,
                    "current_load": worker.current_load,
                    "utilization": f"{worker.current_load / worker.capacity * 100:.1f}%"
                }
                for name, worker in self.workers.items()
            }
            
            return {
                "total_delegations": total_delegations,
                "successful_delegations": successful,
                "pending_tasks": len(self.pending_tasks),
                "worker_stats": worker_stats
            }
    
    
    class DelegateAgentPattern:
        """
        Delegate Agent Pattern - Task delegation and routing.
        
        Delegate agents route tasks to specialized workers based on
        skills, availability, and load balancing.
        """
        
        def __init__(self):
            self.delegates: Dict[str, DelegateAgent] = {}
        
        def register_delegate(self, delegate: DelegateAgent):
            """Register delegate agent."""
            self.delegates[delegate.name] = delegate
        
        def run(self, tasks: List[Task]) -> Dict[str, Any]:
            """Execute delegation scenarios."""
            print("=== Delegate Agent Pattern Demo ===\n")
            
            for delegate in self.delegates.values():
                for task in tasks:
                    delegate.delegate_task(task)
                
                # Try pending tasks
                delegate.process_pending_tasks()
            
            total_delegations = sum(
                len(d.delegation_history)
                for d in self.delegates.values()
            )
            
            return {
                "pattern": "DelegateAgent",
                "total_delegates": len(self.delegates),
                "total_delegations": total_delegations,
                "tasks_processed": len(tasks)
            }
    
    
    # Example usage
    if __name__ == "__main__":
        print("=== Delegate Agent Pattern Demo ===\n")
        
        # Create pattern
        pattern = DelegateAgentPattern()
        
        # Create delegate agent
        print("1. Creating delegate agent...")
        delegate = DelegateAgent("TaskCoordinator")
        
        # Register workers
        print("\n2. Registering worker agents...")
        
        delegate.register_worker(WorkerAgent(
            name="DataWorker",
            skills=["data_processing", "analysis", "sql"],
            capacity=3
        ))
        
        delegate.register_worker(WorkerAgent(
            name="WebWorker",
            skills=["web_scraping", "api_calls", "http"],
            capacity=5
        ))
        
        delegate.register_worker(WorkerAgent(
            name="MLWorker",
            skills=["machine_learning", "training", "prediction"],
            capacity=2
        ))
        
        delegate.register_worker(WorkerAgent(
            name="GeneralWorker",
            skills=["data_processing", "web_scraping", "text_processing"],
            capacity=4
        ))
        
        pattern.register_delegate(delegate)
        
        # Create tasks
        print("\n3. Creating tasks to delegate...")
        
        tasks = [
            Task(
                task_id="T001",
                description="Analyze sales data and generate report",
                required_skills=["data_processing", "analysis"],
                priority=TaskPriority.HIGH
            ),
            Task(
                task_id="T002",
                description="Scrape product information from website",
                required_skills=["web_scraping"],
                priority=TaskPriority.MEDIUM
            ),
            Task(
                task_id="T003",
                description="Train predictive model on customer data",
                required_skills=["machine_learning", "training"],
                priority=TaskPriority.HIGH
            ),
            Task(
                task_id="T004",
                description="Extract data from SQL database",
                required_skills=["sql", "data_processing"],
                priority=TaskPriority.LOW
            ),
            Task(
                task_id="T005",
                description="Process text documents",
                required_skills=["text_processing"],
                priority=TaskPriority.MEDIUM
            )
        ]
        
        # Execute
        print("\n4. Delegating tasks...")
        summary = pattern.run(tasks)
        
        print(f"\n5. Summary:")
        print(f"   Total Delegations: {summary['total_delegations']}")
        print(f"   Tasks Processed: {summary['tasks_processed']}")
        
        print("\n6. Delegation Statistics:")
        stats = delegate.get_statistics()
        print(f"   Successful Delegations: {stats['successful_delegations']}")
        print(f"   Pending Tasks: {stats['pending_tasks']}")
        
        print("\n7. Worker Performance:")
        for worker_name, worker_stat in stats['worker_stats'].items():
            print(f"   {worker_name}:")
            print(f"     Tasks Completed: {worker_stat['tasks_completed']}")
            print(f"     Current Load: {worker_stat['current_load']}")
            print(f"     Utilization: {worker_stat['utilization']}")
        
        print("\n✓ Delegate agent successfully routed tasks to workers!")
    
        

    Agentic Architecture Pattern: Delegate Agent

    Pattern 90: 90 Delegate Agent – Line-by-Line Explanation

    Overview

    Pattern 90: Delegate Agent Pattern

    Delegates work to specialized worker agents based on task requirements. Features task routing, load balancing, and result aggregation.


    Code Structure Analysis

    File Information

    • Pattern Number: 90
    • Pattern Name: 90 Delegate Agent
    • File Path: patterns\specialized_agent\90_delegate_agent.py
    • Total Lines: 336

    Detailed Code Explanation

    Module Docstring

    """
    Pattern 90: Delegate Agent Pattern
    
    Delegates work to specialized worker agents based on task requirements.
    Features task routing, load balancing, and result aggregation.
    """
    

    Explanation: This module docstring describes the purpose and key features of the 90 Delegate Agent pattern.


    Imports Section

    from typing import Dict, Any, List, Optional, Tuple
    from dataclasses import dataclass, field
    from datetime import datetime
    from enum import Enum
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_openai import ChatOpenAI  # First 20 imports shown
    

    Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional, Tuple: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime: Date and time utilities – from enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration


    Enumerations

    TaskPriority (Line 16)

    class TaskPriority(Enum):
        LOW
        MEDIUM
        HIGH
        CRITICAL
    

    Purpose: Task priority levels.

    Values: – LOW: Represents low – MEDIUM: Represents medium – HIGH: Represents high – CRITICAL: Represents critical


    Classes

    TaskPriority (Line 16)

    Description: Task priority levels.

    Task (Line 25)

    Description: Represents a task to be delegated.

    WorkerAgent (Line 36)

    Description: Worker agent that can execute tasks.

    Methods:

    can_accept (Line 44)
    def can_accept(self):
        """Check if worker can accept more tasks."""
    

    Purpose: Check if worker can accept more tasks.

    Parameters: – self: Instance reference

    has_skills (Line 48)
    def has_skills(self, required_skills):
        """Check if worker has required skills."""
    

    Purpose: Check if worker has required skills.

    Parameters: – self: Instance reference – required_skills: Parameter for required_skills

    match_score (Line 52)
    def match_score(self, required_skills):
        """Calculate match score (0.0 to 1.0)."""
    

    Purpose: Calculate match score (0.0 to 1.0).

    Parameters: – self: Instance reference – required_skills: Parameter for required_skills

    DelegationRecord (Line 61)

    Description: Records a delegation event.

    DelegateAgent (Line 71)

    Description: Agent that delegates tasks to workers.

    Methods:

    __init__ (Line 74)
    def __init__(self, name, model_name):
        """No method docstring."""
    

    Purpose: No method docstring.

    Parameters: – self: Instance reference – name: Identifier or label – model_name: LLM model to use (e.g., “gpt-4”)

    register_worker (Line 81)
    def register_worker(self, worker):
        """Register a worker agent."""
    

    Purpose: Register a worker agent.

    Parameters: – self: Instance reference – worker: Parameter for worker

    find_best_worker (Line 86)
    def find_best_worker(self, task):
        """Find the best worker for a task."""
    

    Purpose: Find the best worker for a task.

    Parameters: – self: Instance reference – task: Parameter for task

    delegate_task (Line 105)
    def delegate_task(self, task):
        """Delegate task to best available worker."""
    

    Purpose: Delegate task to best available worker.

    Parameters: – self: Instance reference – task: Parameter for task

    _execute_task_via_worker (Line 145)
    def _execute_task_via_worker(self, task, worker):
        """Execute task through worker agent."""
    

    Purpose: Execute task through worker agent.

    Parameters: – self: Instance reference – task: Parameter for task – worker: Parameter for worker

    process_pending_tasks (Line 161)
    def process_pending_tasks(self):
        """Try to process pending tasks."""
    

    Purpose: Try to process pending tasks.

    Parameters: – self: Instance reference

    get_statistics (Line 178)
    def get_statistics(self):
        """Get delegation statistics."""
    

    Purpose: Get delegation statistics.

    Parameters: – self: Instance reference

    DelegateAgentPattern (Line 200)

    Description: Delegate Agent Pattern – Task delegation and routing.

    Delegate agents route tasks to specialized workers based on skills, availability, and load balancing.

    Methods:

    __init__ (Line 208)
    def __init__(self):
        """No method docstring."""
    

    Purpose: No method docstring.

    Parameters: – self: Instance reference

    register_delegate (Line 211)
    def register_delegate(self, delegate):
        """Register delegate agent."""
    

    Purpose: Register delegate agent.

    Parameters: – self: Instance reference – delegate: Parameter for delegate

    run (Line 215)
    def run(self, tasks):
        """Execute delegation scenarios."""
    

    Purpose: Execute delegation scenarios.

    Parameters: – self: Instance reference – tasks: Parameter for tasks


    Design Patterns Used

    • Strategy Pattern: Different execution strategies for various scenarios
    • State Pattern: Managing different states in the workflow
    • Chain of Responsibility: Sequential processing through workflow nodes
    • Observer Pattern: Tracking and monitoring pattern execution
    • Factory Pattern: Creating instances with proper configuration

    Usage Example

    # Initialize the pattern
    pattern = TaskPriority(name="example", model_name="gpt-4")
    
    # Configure and execute
    result = pattern.execute(data={"key": "value"})
    
    # Check results
    print(result)
    

    Key Concepts

    • LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
    • Workflow Orchestration: LangGraph manages execution flow through nodes
    • State Management: TypedDict defines workflow state structure
    • Error Handling: Comprehensive error handling and fallback mechanisms
    • Extensibility: Designed for easy customization and extension

    Extension Points

    This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities

  • Agentic Architecture Pattern: Proxy Agent

    
        """
    Pattern 89: Proxy Agent Pattern
    
    Acts as an intermediary controlling access to another agent or service.
    Features access control, caching, logging, and lazy initialization.
    """
    
    from typing import Dict, Any, List, Optional, Callable
    from dataclasses import dataclass, field
    from datetime import datetime, timedelta
    from enum import Enum
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_openai import ChatOpenAI
    
    
    class AccessLevel(Enum):
        """Access control levels."""
        PUBLIC = "public"
        PRIVATE = "private"
        RESTRICTED = "restricted"
        ADMIN = "admin"
    
    
    @dataclass
    class CacheEntry:
        """Cached response entry."""
        key: str
        value: Any
        timestamp: datetime = field(default_factory=datetime.now)
        hits: int = 0
        ttl_seconds: int = 300  # 5 minutes default
        
        def is_expired(self) -> bool:
            """Check if cache entry has expired."""
            return datetime.now() > self.timestamp + timedelta(seconds=self.ttl_seconds)
    
    
    @dataclass
    class AccessRequest:
        """Access request record."""
        user: str
        operation: str
        access_level: AccessLevel
        allowed: bool
        timestamp: datetime = field(default_factory=datetime.now)
    
    
    class RealAgent:
        """The real agent being proxied."""
        
        def __init__(self, name: str, model_name: str = "gpt-4"):
            self.name = name
            self.llm = ChatOpenAI(model=model_name, temperature=0.7)
            self.initialized = False
            self.operation_count = 0
        
        def initialize(self):
            """Expensive initialization (lazy loaded)."""
            if not self.initialized:
                print(f"   {self.name}: Performing expensive initialization...")
                self.initialized = True
        
        def execute(self, operation: str, params: Dict[str, Any]) -> str:
            """Execute real operation."""
            if not self.initialized:
                self.initialize()
            
            self.operation_count += 1
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", f"You are {self.name}. Execute the requested operation."),
                ("human", "Operation: {operation}\nParameters: {params}")
            ])
            
            chain = prompt | self.llm
            response = chain.invoke({"operation": operation, "params": str(params)})
            
            return response.content
    
    
    class ProxyAgent:
        """Proxy agent that controls access to real agent."""
        
        def __init__(self, name: str, real_agent: RealAgent):
            self.name = name
            self.real_agent = real_agent
            self.cache: Dict[str, CacheEntry] = {}
            self.access_log: List[AccessRequest] = []
            self.permissions: Dict[str, List[str]] = {
                "public_ops": ["read", "list", "search"],
                "restricted_ops": ["write", "update"],
                "admin_ops": ["delete", "admin"]
            }
        
        def check_access(self, user: str, operation: str, user_level: AccessLevel) -> bool:
            """Check if user has access to operation."""
            # Public operations
            if operation in self.permissions["public_ops"]:
                allowed = True
            # Restricted operations
            elif operation in self.permissions["restricted_ops"]:
                allowed = user_level in [AccessLevel.RESTRICTED, AccessLevel.ADMIN]
            # Admin operations
            elif operation in self.permissions["admin_ops"]:
                allowed = user_level == AccessLevel.ADMIN
            else:
                allowed = True  # Unknown operations allowed by default
            
            # Log access request
            self.access_log.append(AccessRequest(
                user=user,
                operation=operation,
                access_level=user_level,
                allowed=allowed
            ))
            
            return allowed
        
        def get_from_cache(self, key: str) -> Optional[Any]:
            """Get value from cache if not expired."""
            if key in self.cache:
                entry = self.cache[key]
                if not entry.is_expired():
                    entry.hits += 1
                    print(f"   {self.name}: Cache hit! (hits: {entry.hits})")
                    return entry.value
                else:
                    del self.cache[key]
                    print(f"   {self.name}: Cache expired")
            return None
        
        def put_in_cache(self, key: str, value: Any, ttl_seconds: int = 300):
            """Put value in cache."""
            self.cache[key] = CacheEntry(key=key, value=value, ttl_seconds=ttl_seconds)
            print(f"   {self.name}: Cached result (TTL: {ttl_seconds}s)")
        
        def execute(self, user: str, user_level: AccessLevel, operation: str, 
                    params: Dict[str, Any], use_cache: bool = True) -> Optional[str]:
            """Execute operation through proxy."""
            print(f"\n   {self.name}: Proxy request from {user}")
            print(f"   Operation: {operation}, Access Level: {user_level.value}")
            
            # Check access
            if not self.check_access(user, operation, user_level):
                print(f"   {self.name}: Access DENIED")
                return None
            
            print(f"   {self.name}: Access granted")
            
            # Check cache
            if use_cache and operation == "read":
                cache_key = f"{operation}:{str(params)}"
                cached_result = self.get_from_cache(cache_key)
                if cached_result is not None:
                    return cached_result
            
            # Forward to real agent
            print(f"   {self.name}: Forwarding to real agent...")
            result = self.real_agent.execute(operation, params)
            
            # Cache result for read operations
            if use_cache and operation == "read":
                cache_key = f"{operation}:{str(params)}"
                self.put_in_cache(cache_key, result)
            
            return result
        
        def get_statistics(self) -> Dict[str, Any]:
            """Get proxy statistics."""
            total_requests = len(self.access_log)
            denied_requests = sum(1 for req in self.access_log if not req.allowed)
            cache_entries = len(self.cache)
            total_cache_hits = sum(entry.hits for entry in self.cache.values())
            
            return {
                "total_requests": total_requests,
                "denied_requests": denied_requests,
                "cache_entries": cache_entries,
                "total_cache_hits": total_cache_hits,
                "real_agent_calls": self.real_agent.operation_count
            }
    
    
    class ProxyAgentPattern:
        """
        Proxy Agent Pattern - Controlled access intermediary.
        
        Proxy agents control access to real agents providing caching,
        access control, logging, and lazy initialization.
        """
        
        def __init__(self):
            self.proxies: Dict[str, ProxyAgent] = {}
        
        def register_proxy(self, proxy: ProxyAgent):
            """Register proxy agent."""
            self.proxies[proxy.name] = proxy
        
        def run(self, scenarios: List[Dict[str, Any]]) -> Dict[str, Any]:
            """Execute proxy scenarios."""
            print("=== Proxy Agent Pattern Demo ===\n")
            
            for scenario in scenarios:
                print(f"\nScenario: {scenario.get('description', 'Unknown')}")
                
                proxy_name = scenario.get("proxy")
                if proxy_name not in self.proxies:
                    continue
                
                proxy = self.proxies[proxy_name]
                user = scenario.get("user", "anonymous")
                user_level = AccessLevel(scenario.get("access_level", "public"))
                operation = scenario.get("operation")
                params = scenario.get("params", {})
                use_cache = scenario.get("use_cache", True)
                
                result = proxy.execute(user, user_level, operation, params, use_cache)
                
                if result:
                    print(f"   Result: {result[:80]}...")
            
            return {
                "pattern": "ProxyAgent",
                "total_proxies": len(self.proxies),
                "scenarios": len(scenarios)
            }
    
    
    # Example usage
    if __name__ == "__main__":
        print("=== Proxy Agent Pattern Demo ===\n")
        
        # Create pattern
        pattern = ProxyAgentPattern()
        
        # Create real agent
        print("1. Creating real agent (not initialized yet)...")
        real_agent = RealAgent("DatabaseAgent")
        
        # Create proxy
        print("\n2. Creating proxy agent...")
        proxy = ProxyAgent("DatabaseProxy", real_agent)
        pattern.register_proxy(proxy)
        
        # Test scenarios
        print("\n3. Executing operations through proxy...")
        
        scenarios = [
            {
                "description": "Public read (should be cached)",
                "proxy": "DatabaseProxy",
                "user": "alice",
                "access_level": "public",
                "operation": "read",
                "params": {"id": "user_123"}
            },
            {
                "description": "Same read (cache hit expected)",
                "proxy": "DatabaseProxy",
                "user": "bob",
                "access_level": "public",
                "operation": "read",
                "params": {"id": "user_123"}
            },
            {
                "description": "Write operation (restricted)",
                "proxy": "DatabaseProxy",
                "user": "alice",
                "access_level": "public",
                "operation": "write",
                "params": {"id": "user_456", "data": {"name": "Alice"}}
            },
            {
                "description": "Write with restricted access",
                "proxy": "DatabaseProxy",
                "user": "admin",
                "access_level": "restricted",
                "operation": "write",
                "params": {"id": "user_456", "data": {"name": "Alice"}}
            },
            {
                "description": "Delete operation (admin only)",
                "proxy": "DatabaseProxy",
                "user": "admin",
                "access_level": "admin",
                "operation": "delete",
                "params": {"id": "user_789"}
            }
        ]
        
        summary = pattern.run(scenarios)
        
        print(f"\n4. Summary:")
        print(f"   Total Proxies: {summary['total_proxies']}")
        print(f"   Scenarios Executed: {summary['scenarios']}")
        
        print("\n5. Proxy Statistics:")
        stats = proxy.get_statistics()
        print(f"   Total Requests: {stats['total_requests']}")
        print(f"   Denied Requests: {stats['denied_requests']}")
        print(f"   Cache Entries: {stats['cache_entries']}")
        print(f"   Cache Hits: {stats['total_cache_hits']}")
        print(f"   Real Agent Calls: {stats['real_agent_calls']}")
        
        print("\n6. Access Log:")
        for req in proxy.access_log[-5:]:  # Last 5 requests
            status = "✓" if req.allowed else "✗"
            print(f"   {status} {req.user} ({req.access_level.value}): {req.operation}")
        
        print("\n✓ Proxy agent successfully controlled access and caching!")
    
        

    Agentic Architecture Pattern: Proxy Agent

    Pattern 89: 89 Proxy Agent – Line-by-Line Explanation

    Overview

    Pattern 89: Proxy Agent Pattern

    Acts as an intermediary controlling access to another agent or service. Features access control, caching, logging, and lazy initialization.


    Code Structure Analysis

    File Information

    • Pattern Number: 89
    • Pattern Name: 89 Proxy Agent
    • File Path: patterns\specialized_agent\89_proxy_agent.py
    • Total Lines: 311

    Detailed Code Explanation

    Module Docstring

    """
    Pattern 89: Proxy Agent Pattern
    
    Acts as an intermediary controlling access to another agent or service.
    Features access control, caching, logging, and lazy initialization.
    """
    

    Explanation: This module docstring describes the purpose and key features of the 89 Proxy Agent pattern.


    Imports Section

    from typing import Dict, Any, List, Optional, Callable
    from dataclasses import dataclass, field
    from datetime import datetime, timedelta
    from enum import Enum
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_openai import ChatOpenAI  # First 20 imports shown
    

    Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional, Callable: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime, timedelta: Date and time utilities – from enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration


    Enumerations

    AccessLevel (Line 16)

    class AccessLevel(Enum):
        PUBLIC
        PRIVATE
        RESTRICTED
        ADMIN
    

    Purpose: Access control levels.

    Values: – PUBLIC: Represents public – PRIVATE: Represents private – RESTRICTED: Represents restricted – ADMIN: Represents admin


    Classes

    AccessLevel (Line 16)

    Description: Access control levels.

    CacheEntry (Line 25)

    Description: Cached response entry.

    Methods:

    is_expired (Line 33)
    def is_expired(self):
        """Check if cache entry has expired."""
    

    Purpose: Check if cache entry has expired.

    Parameters: – self: Instance reference

    AccessRequest (Line 39)

    Description: Access request record.

    RealAgent (Line 48)

    Description: The real agent being proxied.

    Methods:

    __init__ (Line 51)
    def __init__(self, name, model_name):
        """No method docstring."""
    

    Purpose: No method docstring.

    Parameters: – self: Instance reference – name: Identifier or label – model_name: LLM model to use (e.g., “gpt-4”)

    initialize (Line 57)
    def initialize(self):
        """Expensive initialization (lazy loaded)."""
    

    Purpose: Expensive initialization (lazy loaded).

    Parameters: – self: Instance reference

    execute (Line 63)
    def execute(self, operation, params):
        """Execute real operation."""
    

    Purpose: Execute real operation.

    Parameters: – self: Instance reference – operation: Parameter for operation – params: Parameter for params

    ProxyAgent (Line 81)

    Description: Proxy agent that controls access to real agent.

    Methods:

    __init__ (Line 84)
    def __init__(self, name, real_agent):
        """No method docstring."""
    

    Purpose: No method docstring.

    Parameters: – self: Instance reference – name: Identifier or label – real_agent: Parameter for real_agent

    check_access (Line 95)
    def check_access(self, user, operation, user_level):
        """Check if user has access to operation."""
    

    Purpose: Check if user has access to operation.

    Parameters: – self: Instance reference – user: Parameter for user – operation: Parameter for operation – user_level: Parameter for user_level

    get_from_cache (Line 119)
    def get_from_cache(self, key):
        """Get value from cache if not expired."""
    

    Purpose: Get value from cache if not expired.

    Parameters: – self: Instance reference – key: Parameter for key

    put_in_cache (Line 132)
    def put_in_cache(self, key, value, ttl_seconds):
        """Put value in cache."""
    

    Purpose: Put value in cache.

    Parameters: – self: Instance reference – key: Parameter for key – value: Parameter for value – ttl_seconds: Parameter for ttl_seconds

    execute (Line 137)
    def execute(self, user, user_level, operation, params, use_cache):
        """Execute operation through proxy."""
    

    Purpose: Execute operation through proxy.

    Parameters: – self: Instance reference – user: Parameter for user – user_level: Parameter for user_level – operation: Parameter for operation – params: Parameter for params – use_cache: Parameter for use_cache

    get_statistics (Line 168)
    def get_statistics(self):
        """Get proxy statistics."""
    

    Purpose: Get proxy statistics.

    Parameters: – self: Instance reference

    ProxyAgentPattern (Line 184)

    Description: Proxy Agent Pattern – Controlled access intermediary.

    Proxy agents control access to real agents providing caching, access control, logging, and lazy initialization.

    Methods:

    __init__ (Line 192)
    def __init__(self):
        """No method docstring."""
    

    Purpose: No method docstring.

    Parameters: – self: Instance reference

    register_proxy (Line 195)
    def register_proxy(self, proxy):
        """Register proxy agent."""
    

    Purpose: Register proxy agent.

    Parameters: – self: Instance reference – proxy: Parameter for proxy

    run (Line 199)
    def run(self, scenarios):
        """Execute proxy scenarios."""
    

    Purpose: Execute proxy scenarios.

    Parameters: – self: Instance reference – scenarios: Parameter for scenarios


    Design Patterns Used

    • Strategy Pattern: Different execution strategies for various scenarios
    • State Pattern: Managing different states in the workflow
    • Chain of Responsibility: Sequential processing through workflow nodes
    • Observer Pattern: Tracking and monitoring pattern execution
    • Factory Pattern: Creating instances with proper configuration

    Usage Example

    # Initialize the pattern
    pattern = AccessLevel(name="example", model_name="gpt-4")
    
    # Configure and execute
    result = pattern.execute(data={"key": "value"})
    
    # Check results
    print(result)
    

    Key Concepts

    • LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
    • Workflow Orchestration: LangGraph manages execution flow through nodes
    • State Management: TypedDict defines workflow state structure
    • Error Handling: Comprehensive error handling and fallback mechanisms
    • Extensibility: Designed for easy customization and extension

    Extension Points

    This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities

  • Agentic Architecture Pattern: Wrapper Agent

    
        """
    Pattern 88: Wrapper Agent Pattern
    
    Wraps legacy systems providing modern interfaces while maintaining backward compatibility.
    Features interface adaptation, protocol translation, and behavior enhancement.
    """
    
    from typing import Dict, Any, List, Optional, Callable
    from dataclasses import dataclass, field
    from datetime import datetime
    from enum import Enum
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_openai import ChatOpenAI
    
    
    class InterfaceType(Enum):
        """Types of interfaces."""
        LEGACY = "legacy"
        MODERN = "modern"
        REST_API = "rest_api"
        CLI = "cli"
        GUI = "gui"
    
    
    @dataclass
    class LegacySystem:
        """Represents a legacy system."""
        name: str
        interface_type: InterfaceType
        operations: List[str]
        limitations: List[str] = field(default_factory=list)
        
        def execute_legacy_operation(self, operation: str, params: Dict[str, Any]) -> str:
            """Simulate legacy system operation."""
            return f"Legacy {operation} executed with {params}"
    
    
    @dataclass
    class WrapperCall:
        """Records a wrapper invocation."""
        operation: str
        modern_params: Dict[str, Any]
        legacy_params: Dict[str, Any]
        result: Any
        timestamp: datetime = field(default_factory=datetime.now)
    
    
    class WrapperAgent:
        """Agent that wraps legacy systems with modern interfaces."""
        
        def __init__(self, name: str, legacy_system: LegacySystem, model_name: str = "gpt-4"):
            self.name = name
            self.legacy_system = legacy_system
            self.llm = ChatOpenAI(model=model_name, temperature=0.3)
            self.call_history: List[WrapperCall] = []
            self.modern_operations: Dict[str, Callable] = {}
            self.parameter_mappings: Dict[str, Dict[str, str]] = {}
        
        def register_modern_operation(self, modern_name: str, legacy_name: str, 
                                       param_mapping: Optional[Dict[str, str]] = None):
            """Register a modern operation that wraps a legacy one."""
            self.parameter_mappings[modern_name] = param_mapping or {}
            
            def wrapper(**modern_params):
                return self.invoke_legacy(modern_name, legacy_name, modern_params)
            
            self.modern_operations[modern_name] = wrapper
            print(f"   {self.name}: Registered '{modern_name}' -> '{legacy_name}'")
        
        def invoke_legacy(self, modern_op: str, legacy_op: str, 
                          modern_params: Dict[str, Any]) -> Any:
            """Invoke legacy operation with parameter translation."""
            print(f"\n   {self.name}: Wrapping call to '{modern_op}'")
            
            # Translate parameters
            legacy_params = self._translate_parameters(modern_op, modern_params)
            print(f"   Translated parameters: {legacy_params}")
            
            # Call legacy system
            result = self.legacy_system.execute_legacy_operation(legacy_op, legacy_params)
            
            # Enhance result with LLM
            enhanced_result = self._enhance_result(result, modern_op)
            
            # Record call
            self.call_history.append(WrapperCall(
                operation=modern_op,
                modern_params=modern_params,
                legacy_params=legacy_params,
                result=enhanced_result
            ))
            
            print(f"   Result: {enhanced_result[:80]}...")
            return enhanced_result
        
        def _translate_parameters(self, operation: str, modern_params: Dict[str, Any]) -> Dict[str, Any]:
            """Translate modern parameters to legacy format."""
            mapping = self.parameter_mappings.get(operation, {})
            legacy_params = {}
            
            for modern_key, value in modern_params.items():
                legacy_key = mapping.get(modern_key, modern_key)
                legacy_params[legacy_key] = value
            
            return legacy_params
        
        def _enhance_result(self, legacy_result: str, operation: str) -> str:
            """Enhance legacy result with modern formatting."""
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are enhancing legacy system output for modern applications. "
                          "Format the result in a more user-friendly way."),
                ("human", "Operation: {operation}\nLegacy Result: {result}\n\n"
                         "Provide an enhanced, user-friendly version.")
            ])
            
            chain = prompt | self.llm
            response = chain.invoke({
                "operation": operation,
                "result": legacy_result
            })
            
            return response.content
        
        def call(self, operation: str, **params) -> Any:
            """Call a modern operation."""
            if operation not in self.modern_operations:
                raise ValueError(f"Operation '{operation}' not found")
            
            return self.modern_operations[operation](**params)
        
        def get_statistics(self) -> Dict[str, Any]:
            """Get wrapper statistics."""
            operation_counts = {}
            for call in self.call_history:
                operation_counts[call.operation] = operation_counts.get(call.operation, 0) + 1
            
            return {
                "total_calls": len(self.call_history),
                "operation_counts": operation_counts,
                "wrapped_system": self.legacy_system.name,
                "modern_operations": list(self.modern_operations.keys())
            }
    
    
    class WrapperAgentPattern:
        """
        Wrapper Agent Pattern - Legacy system integration.
        
        Wrapper agents provide modern interfaces to legacy systems,
        enabling integration without modifying the legacy code.
        """
        
        def __init__(self):
            self.agents: Dict[str, WrapperAgent] = {}
        
        def register_agent(self, agent: WrapperAgent):
            """Register wrapper agent."""
            self.agents[agent.name] = agent
        
        def run(self, scenarios: List[Dict[str, Any]]) -> Dict[str, Any]:
            """Execute wrapper scenarios."""
            print("=== Wrapper Agent Pattern Demo ===\n")
            
            for scenario in scenarios:
                print(f"\nScenario: {scenario.get('description', 'Unknown')}")
                
                agent_name = scenario.get("agent")
                if agent_name not in self.agents:
                    continue
                
                agent = self.agents[agent_name]
                operation = scenario.get("operation")
                params = scenario.get("params", {})
                
                try:
                    result = agent.call(operation, **params)
                except Exception as e:
                    print(f"   Error: {e}")
            
            # Statistics
            total_calls = sum(
                len(agent.call_history)
                for agent in self.agents.values()
            )
            
            return {
                "pattern": "WrapperAgent",
                "total_agents": len(self.agents),
                "total_calls": total_calls,
                "scenarios": len(scenarios)
            }
    
    
    # Example usage
    if __name__ == "__main__":
        print("=== Wrapper Agent Pattern Demo ===\n")
        
        # Create pattern
        pattern = WrapperAgentPattern()
        
        # Create legacy system
        print("1. Defining legacy system...")
        legacy = LegacySystem(
            name="MainframeDB",
            interface_type=InterfaceType.CLI,
            operations=["FETCH_REC", "UPD_REC", "DEL_REC", "INS_REC"],
            limitations=["Fixed-width fields", "No JSON support", "Cryptic error codes"]
        )
        print(f"   Legacy system: {legacy.name}")
        print(f"   Operations: {', '.join(legacy.operations)}")
        
        # Create wrapper agent
        print("\n2. Creating wrapper agent...")
        wrapper = WrapperAgent("ModernDBWrapper", legacy)
        
        # Register modern operations
        print("\n3. Registering modern operations...")
        wrapper.register_modern_operation(
            modern_name="get_user",
            legacy_name="FETCH_REC",
            param_mapping={"user_id": "REC_ID", "fields": "FLD_LIST"}
        )
        
        wrapper.register_modern_operation(
            modern_name="update_user",
            legacy_name="UPD_REC",
            param_mapping={"user_id": "REC_ID", "data": "FLD_DATA"}
        )
        
        wrapper.register_modern_operation(
            modern_name="delete_user",
            legacy_name="DEL_REC",
            param_mapping={"user_id": "REC_ID"}
        )
        
        pattern.register_agent(wrapper)
        
        # Test scenarios
        print("\n4. Executing modern operations...")
        
        scenarios = [
            {
                "description": "Get user by ID",
                "agent": "ModernDBWrapper",
                "operation": "get_user",
                "params": {"user_id": "12345", "fields": ["name", "email", "age"]}
            },
            {
                "description": "Update user data",
                "agent": "ModernDBWrapper",
                "operation": "update_user",
                "params": {"user_id": "12345", "data": {"email": "new@example.com"}}
            },
            {
                "description": "Delete user",
                "agent": "ModernDBWrapper",
                "operation": "delete_user",
                "params": {"user_id": "67890"}
            }
        ]
        
        summary = pattern.run(scenarios)
        
        print(f"\n5. Summary:")
        print(f"   Total Calls: {summary['total_calls']}")
        print(f"   Scenarios Executed: {summary['scenarios']}")
        
        print("\n6. Wrapper Statistics:")
        stats = wrapper.get_statistics()
        print(f"   Wrapped System: {stats['wrapped_system']}")
        print(f"   Modern Operations: {', '.join(stats['modern_operations'])}")
        print(f"   Operation Counts:")
        for op, count in stats['operation_counts'].items():
            print(f"     {op}: {count}")
        
        print("\n✓ Wrapper agent successfully modernized legacy system!")
    
        

    Agentic Architecture Pattern: Wrapper Agent

    Pattern 88: 88 Wrapper Agent – Line-by-Line Explanation

    Overview

    Pattern 88: Wrapper Agent Pattern

    Wraps legacy systems providing modern interfaces while maintaining backward compatibility. Features interface adaptation, protocol translation, and behavior enhancement.


    Code Structure Analysis

    File Information

    • Pattern Number: 88
    • Pattern Name: 88 Wrapper Agent
    • File Path: patterns\specialized_agent\88_wrapper_agent.py
    • Total Lines: 277

    Detailed Code Explanation

    Module Docstring

    """
    Pattern 88: Wrapper Agent Pattern
    
    Wraps legacy systems providing modern interfaces while maintaining backward compatibility.
    Features interface adaptation, protocol translation, and behavior enhancement.
    """
    

    Explanation: This module docstring describes the purpose and key features of the 88 Wrapper Agent pattern.


    Imports Section

    from typing import Dict, Any, List, Optional, Callable
    from dataclasses import dataclass, field
    from datetime import datetime
    from enum import Enum
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_openai import ChatOpenAI  # First 20 imports shown
    

    Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional, Callable: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime: Date and time utilities – from enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration


    Enumerations

    InterfaceType (Line 16)

    class InterfaceType(Enum):
        LEGACY
        MODERN
        REST_API
        CLI
        GUI
    

    Purpose: Types of interfaces.

    Values: – LEGACY: Represents legacy – MODERN: Represents modern – REST_API: Represents rest api – CLI: Represents cli – GUI: Represents gui


    Classes

    InterfaceType (Line 16)

    Description: Types of interfaces.

    LegacySystem (Line 26)

    Description: Represents a legacy system.

    Methods:

    execute_legacy_operation (Line 33)
    def execute_legacy_operation(self, operation, params):
        """Simulate legacy system operation."""
    

    Purpose: Simulate legacy system operation.

    Parameters: – self: Instance reference – operation: Parameter for operation – params: Parameter for params

    WrapperCall (Line 39)

    Description: Records a wrapper invocation.

    WrapperAgent (Line 48)

    Description: Agent that wraps legacy systems with modern interfaces.

    Methods:

    __init__ (Line 51)
    def __init__(self, name, legacy_system, model_name):
        """No method docstring."""
    

    Purpose: No method docstring.

    Parameters: – self: Instance reference – name: Identifier or label – legacy_system: Parameter for legacy_system – model_name: LLM model to use (e.g., “gpt-4”)

    register_modern_operation (Line 59)
    def register_modern_operation(self, modern_name, legacy_name, param_mapping):
        """Register a modern operation that wraps a legacy one."""
    

    Purpose: Register a modern operation that wraps a legacy one.

    Parameters: – self: Instance reference – modern_name: Parameter for modern_name – legacy_name: Parameter for legacy_name – param_mapping: Parameter for param_mapping

    invoke_legacy (Line 70)
    def invoke_legacy(self, modern_op, legacy_op, modern_params):
        """Invoke legacy operation with parameter translation."""
    

    Purpose: Invoke legacy operation with parameter translation.

    Parameters: – self: Instance reference – modern_op: Parameter for modern_op – legacy_op: Parameter for legacy_op – modern_params: Parameter for modern_params

    _translate_parameters (Line 96)
    def _translate_parameters(self, operation, modern_params):
        """Translate modern parameters to legacy format."""
    

    Purpose: Translate modern parameters to legacy format.

    Parameters: – self: Instance reference – operation: Parameter for operation – modern_params: Parameter for modern_params

    _enhance_result (Line 107)
    def _enhance_result(self, legacy_result, operation):
        """Enhance legacy result with modern formatting."""
    

    Purpose: Enhance legacy result with modern formatting.

    Parameters: – self: Instance reference – legacy_result: Parameter for legacy_result – operation: Parameter for operation

    call (Line 124)
    def call(self, operation):
        """Call a modern operation."""
    

    Purpose: Call a modern operation.

    Parameters: – self: Instance reference – operation: Parameter for operation

    get_statistics (Line 131)
    def get_statistics(self):
        """Get wrapper statistics."""
    

    Purpose: Get wrapper statistics.

    Parameters: – self: Instance reference

    WrapperAgentPattern (Line 145)

    Description: Wrapper Agent Pattern – Legacy system integration.

    Wrapper agents provide modern interfaces to legacy systems, enabling integration without modifying the legacy code.

    Methods:

    __init__ (Line 153)
    def __init__(self):
        """No method docstring."""
    

    Purpose: No method docstring.

    Parameters: – self: Instance reference

    register_agent (Line 156)
    def register_agent(self, agent):
        """Register wrapper agent."""
    

    Purpose: Register wrapper agent.

    Parameters: – self: Instance reference – agent: Parameter for agent

    run (Line 160)
    def run(self, scenarios):
        """Execute wrapper scenarios."""
    

    Purpose: Execute wrapper scenarios.

    Parameters: – self: Instance reference – scenarios: Parameter for scenarios


    Design Patterns Used

    • Strategy Pattern: Different execution strategies for various scenarios
    • State Pattern: Managing different states in the workflow
    • Chain of Responsibility: Sequential processing through workflow nodes
    • Observer Pattern: Tracking and monitoring pattern execution
    • Factory Pattern: Creating instances with proper configuration

    Usage Example

    # Initialize the pattern
    pattern = InterfaceType(name="example", model_name="gpt-4")
    
    # Configure and execute
    result = pattern.execute(data={"key": "value"})
    
    # Check results
    print(result)
    

    Key Concepts

    • LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
    • Workflow Orchestration: LangGraph manages execution flow through nodes
    • State Management: TypedDict defines workflow state structure
    • Error Handling: Comprehensive error handling and fallback mechanisms
    • Extensibility: Designed for easy customization and extension

    Extension Points

    This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities

  • Agentic Architecture Pattern: Gateway Agent

    
        """
    Pattern 87: Gateway Agent Pattern
    
    Acts as a protocol translator between different systems or communication protocols.
    Features protocol translation, message routing, and format conversion.
    """
    
    from typing import Dict, Any, List, Optional, Callable
    from dataclasses import dataclass, field
    from datetime import datetime
    from enum import Enum
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_openai import ChatOpenAI
    
    
    class Protocol(Enum):
        """Communication protocols."""
        HTTP = "http"
        GRPC = "grpc"
        MQTT = "mqtt"
        WEBSOCKET = "websocket"
        REST = "rest"
        SOAP = "soap"
        CUSTOM = "custom"
    
    
    @dataclass
    class Message:
        """Message with protocol information."""
        content: Any
        protocol: Protocol
        headers: Dict[str, str] = field(default_factory=dict)
        metadata: Dict[str, Any] = field(default_factory=dict)
        timestamp: datetime = field(default_factory=datetime.now)
    
    
    @dataclass
    class Translation:
        """Translation record."""
        original_message: Message
        translated_message: Message
        from_protocol: Protocol
        to_protocol: Protocol
        timestamp: datetime = field(default_factory=datetime.now)
    
    
    class ProtocolTranslator:
        """Translates between protocols."""
        
        def __init__(self, llm: ChatOpenAI):
            self.llm = llm
            self.translations: List[Translation] = []
        
        def translate(self, message: Message, target_protocol: Protocol) -> Message:
            """Translate message to target protocol."""
            if message.protocol == target_protocol:
                return message
            
            # Use LLM for intelligent translation
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are a protocol translator. Convert messages between protocols "
                          "while preserving semantic meaning."),
                ("human", """Translate this message:
    From Protocol: {from_protocol}
    To Protocol: {to_protocol}
    Content: {content}
    Headers: {headers}
    
    Provide the translated content and appropriate headers for the target protocol.""")
            ])
            
            chain = prompt | self.llm
            response = chain.invoke({
                "from_protocol": message.protocol.value,
                "to_protocol": target_protocol.value,
                "content": str(message.content),
                "headers": str(message.headers)
            })
            
            translated = Message(
                content=response.content,
                protocol=target_protocol,
                headers={"translated": "true"},
                metadata={"original_protocol": message.protocol.value}
            )
            
            # Record translation
            self.translations.append(Translation(
                original_message=message,
                translated_message=translated,
                from_protocol=message.protocol,
                to_protocol=target_protocol
            ))
            
            return translated
    
    
    class GatewayAgent:
        """Gateway agent for protocol translation and routing."""
        
        def __init__(self, name: str, model_name: str = "gpt-4"):
            self.name = name
            self.llm = ChatOpenAI(model=model_name, temperature=0.3)
            self.translator = ProtocolTranslator(self.llm)
            self.supported_protocols: List[Protocol] = []
            self.routes: Dict[Protocol, Callable[[Message], None]] = {}
            self.messages_processed = 0
        
        def add_protocol_support(self, protocol: Protocol):
            """Add support for a protocol."""
            if protocol not in self.supported_protocols:
                self.supported_protocols.append(protocol)
                print(f"   {self.name}: Added support for {protocol.value}")
        
        def register_route(self, protocol: Protocol, handler: Callable[[Message], None]):
            """Register a message handler for a protocol."""
            self.routes[protocol] = handler
        
        def process_message(self, message: Message, target_protocol: Optional[Protocol] = None) -> Message:
            """Process and translate message."""
            self.messages_processed += 1
            
            print(f"\n   {self.name}: Processing message")
            print(f"   Source protocol: {message.protocol.value}")
            
            # Translate if needed
            if target_protocol and target_protocol != message.protocol:
                if target_protocol not in self.supported_protocols:
                    raise ValueError(f"Protocol {target_protocol.value} not supported")
                
                print(f"   Translating to: {target_protocol.value}")
                translated = self.translator.translate(message, target_protocol)
                
                # Route to handler
                if target_protocol in self.routes:
                    self.routes[target_protocol](translated)
                
                return translated
            
            return message
        
        def get_statistics(self) -> Dict[str, Any]:
            """Get gateway statistics."""
            translation_counts = {}
            for trans in self.translator.translations:
                key = f"{trans.from_protocol.value} -> {trans.to_protocol.value}"
                translation_counts[key] = translation_counts.get(key, 0) + 1
            
            return {
                "messages_processed": self.messages_processed,
                "total_translations": len(self.translator.translations),
                "supported_protocols": [p.value for p in self.supported_protocols],
                "translation_routes": translation_counts
            }
    
    
    class GatewayAgentPattern:
        """
        Gateway Agent Pattern - Protocol translation gateway.
        
        Gateway agents translate between different communication protocols,
        enabling interoperability between heterogeneous systems.
        """
        
        def __init__(self):
            self.agents: Dict[str, GatewayAgent] = {}
        
        def register_agent(self, agent: GatewayAgent):
            """Register gateway agent."""
            self.agents[agent.name] = agent
        
        def run(self, scenarios: List[Dict[str, Any]]) -> Dict[str, Any]:
            """Execute gateway scenarios."""
            print("=== Gateway Agent Pattern Demo ===\n")
            
            for scenario in scenarios:
                print(f"\nScenario: {scenario.get('description', 'Unknown')}")
                
                agent_name = scenario.get("agent")
                if agent_name not in self.agents:
                    continue
                
                agent = self.agents[agent_name]
                message = Message(
                    content=scenario.get("content"),
                    protocol=Protocol(scenario.get("source_protocol")),
                    headers=scenario.get("headers", {})
                )
                
                target = Protocol(scenario.get("target_protocol")) if scenario.get("target_protocol") else None
                result = agent.process_message(message, target)
                
                print(f"   Result protocol: {result.protocol.value}")
            
            # Statistics
            total_translations = sum(
                len(agent.translator.translations)
                for agent in self.agents.values()
            )
            
            return {
                "pattern": "GatewayAgent",
                "total_agents": len(self.agents),
                "total_translations": total_translations,
                "scenarios": len(scenarios)
            }
    
    
    # Example usage
    if __name__ == "__main__":
        print("=== Gateway Agent Pattern Demo ===\n")
        
        # Create pattern
        pattern = GatewayAgentPattern()
        
        # Create gateway agent
        print("1. Creating gateway agent...")
        gateway = GatewayAgent("APIGateway")
        
        # Add protocol support
        print("\n2. Adding protocol support...")
        gateway.add_protocol_support(Protocol.HTTP)
        gateway.add_protocol_support(Protocol.GRPC)
        gateway.add_protocol_support(Protocol.MQTT)
        gateway.add_protocol_support(Protocol.WEBSOCKET)
        
        # Register handlers
        def http_handler(msg: Message):
            print(f"     HTTP handler received: {str(msg.content)[:50]}...")
        
        def grpc_handler(msg: Message):
            print(f"     gRPC handler received: {str(msg.content)[:50]}...")
        
        gateway.register_route(Protocol.HTTP, http_handler)
        gateway.register_route(Protocol.GRPC, grpc_handler)
        
        pattern.register_agent(gateway)
        
        # Test scenarios
        print("\n3. Processing messages...")
        
        scenarios = [
            {
                "description": "Translate REST to gRPC",
                "agent": "APIGateway",
                "content": {"method": "GET", "path": "/api/users", "params": {"id": 123}},
                "source_protocol": "rest",
                "target_protocol": "grpc",
                "headers": {"Content-Type": "application/json"}
            },
            {
                "description": "Translate MQTT to WebSocket",
                "agent": "APIGateway",
                "content": {"topic": "sensors/temperature", "value": 23.5},
                "source_protocol": "mqtt",
                "target_protocol": "websocket",
                "headers": {"qos": "1"}
            },
            {
                "description": "Translate HTTP to MQTT",
                "agent": "APIGateway",
                "content": {"event": "user.login", "data": {"user_id": 456}},
                "source_protocol": "http",
                "target_protocol": "mqtt",
                "headers": {"Authorization": "Bearer token123"}
            }
        ]
        
        summary = pattern.run(scenarios)
        
        print(f"\n4. Summary:")
        print(f"   Total Translations: {summary['total_translations']}")
        print(f"   Scenarios Processed: {summary['scenarios']}")
        
        print("\n5. Gateway Statistics:")
        stats = gateway.get_statistics()
        print(f"   Messages Processed: {stats['messages_processed']}")
        print(f"   Supported Protocols: {', '.join(stats['supported_protocols'])}")
        print(f"   Translation Routes:")
        for route, count in stats['translation_routes'].items():
            print(f"     {route}: {count}")
        
        print("\n✓ Gateway agent successfully translated between protocols!")
    
        

    Agentic Architecture Pattern: Gateway Agent

    Pattern 87: 87 Gateway Agent – Line-by-Line Explanation

    Overview

    Pattern 87: Gateway Agent Pattern

    Acts as a protocol translator between different systems or communication protocols. Features protocol translation, message routing, and format conversion.


    Code Structure Analysis

    File Information

    • Pattern Number: 87
    • Pattern Name: 87 Gateway Agent
    • File Path: patterns\specialized_agent\87_gateway_agent.py
    • Total Lines: 284

    Detailed Code Explanation

    Module Docstring

    """
    Pattern 87: Gateway Agent Pattern
    
    Acts as a protocol translator between different systems or communication protocols.
    Features protocol translation, message routing, and format conversion.
    """
    

    Explanation: This module docstring describes the purpose and key features of the 87 Gateway Agent pattern.


    Imports Section

    from typing import Dict, Any, List, Optional, Callable
    from dataclasses import dataclass, field
    from datetime import datetime
    from enum import Enum
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_openai import ChatOpenAI  # First 20 imports shown
    

    Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional, Callable: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime: Date and time utilities – from enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration


    Enumerations

    Protocol (Line 16)

    class Protocol(Enum):
        HTTP
        GRPC
        MQTT
        WEBSOCKET
        REST
        SOAP
        CUSTOM
    

    Purpose: Communication protocols.

    Values: – HTTP: Represents http – GRPC: Represents grpc – MQTT: Represents mqtt – WEBSOCKET: Represents websocket – REST: Represents rest – SOAP: Represents soap – CUSTOM: Represents custom


    Classes

    Protocol (Line 16)

    Description: Communication protocols.

    Message (Line 28)

    Description: Message with protocol information.

    Translation (Line 38)

    Description: Translation record.

    ProtocolTranslator (Line 47)

    Description: Translates between protocols.

    Methods:

    __init__ (Line 50)
    def __init__(self, llm):
        """No method docstring."""
    

    Purpose: No method docstring.

    Parameters: – self: Instance reference – llm: Parameter for llm

    translate (Line 54)
    def translate(self, message, target_protocol):
        """Translate message to target protocol."""
    

    Purpose: Translate message to target protocol.

    Parameters: – self: Instance reference – message: Parameter for message – target_protocol: Parameter for target_protocol

    GatewayAgent (Line 98)

    Description: Gateway agent for protocol translation and routing.

    Methods:

    __init__ (Line 101)
    def __init__(self, name, model_name):
        """No method docstring."""
    

    Purpose: No method docstring.

    Parameters: – self: Instance reference – name: Identifier or label – model_name: LLM model to use (e.g., “gpt-4”)

    add_protocol_support (Line 109)
    def add_protocol_support(self, protocol):
        """Add support for a protocol."""
    

    Purpose: Add support for a protocol.

    Parameters: – self: Instance reference – protocol: Parameter for protocol

    register_route (Line 115)
    def register_route(self, protocol, handler):
        """Register a message handler for a protocol."""
    

    Purpose: Register a message handler for a protocol.

    Parameters: – self: Instance reference – protocol: Parameter for protocol – handler: Parameter for handler

    process_message (Line 119)
    def process_message(self, message, target_protocol):
        """Process and translate message."""
    

    Purpose: Process and translate message.

    Parameters: – self: Instance reference – message: Parameter for message – target_protocol: Parameter for target_protocol

    get_statistics (Line 142)
    def get_statistics(self):
        """Get gateway statistics."""
    

    Purpose: Get gateway statistics.

    Parameters: – self: Instance reference

    GatewayAgentPattern (Line 157)

    Description: Gateway Agent Pattern – Protocol translation gateway.

    Gateway agents translate between different communication protocols, enabling interoperability between heterogeneous systems.

    Methods:

    __init__ (Line 165)
    def __init__(self):
        """No method docstring."""
    

    Purpose: No method docstring.

    Parameters: – self: Instance reference

    register_agent (Line 168)
    def register_agent(self, agent):
        """Register gateway agent."""
    

    Purpose: Register gateway agent.

    Parameters: – self: Instance reference – agent: Parameter for agent

    run (Line 172)
    def run(self, scenarios):
        """Execute gateway scenarios."""
    

    Purpose: Execute gateway scenarios.

    Parameters: – self: Instance reference – scenarios: Parameter for scenarios


    Design Patterns Used

    • Strategy Pattern: Different execution strategies for various scenarios
    • State Pattern: Managing different states in the workflow
    • Chain of Responsibility: Sequential processing through workflow nodes
    • Observer Pattern: Tracking and monitoring pattern execution
    • Factory Pattern: Creating instances with proper configuration

    Usage Example

    # Initialize the pattern
    pattern = Protocol(name="example", model_name="gpt-4")
    
    # Configure and execute
    result = pattern.execute(data={"key": "value"})
    
    # Check results
    print(result)
    

    Key Concepts

    • LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
    • Workflow Orchestration: LangGraph manages execution flow through nodes
    • State Management: TypedDict defines workflow state structure
    • Error Handling: Comprehensive error handling and fallback mechanisms
    • Extensibility: Designed for easy customization and extension

    Extension Points

    This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities

  • Agentic Architecture Pattern: Stationary Agent

    
        """
    Pattern 86: Stationary Agent
    
    Purpose: Enhanced implementation with full LangChain/LangGraph integration
    """
    
    from typing import Dict, Any, List, Optional
    from dataclasses import dataclass, field
    from datetime import datetime
    from enum import Enum
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_openai import ChatOpenAI
    from langgraph.graph import StateGraph, END
    from typing_extensions import TypedDict
    
    
    @dataclass
    class StationaryAgentConfig:
        """Configuration for Stationary Agent."""
        model_name: str = "gpt-4"
        temperature: float = 0.7
        max_iterations: int = 10
        enable_logging: bool = True
    
    
    @dataclass
    class StationaryAgentState:
        """State information for Stationary Agent."""
        current_step: int = 0
        data: Dict[str, Any] = field(default_factory=dict)
        results: List[Any] = field(default_factory=list)
        metadata: Dict[str, Any] = field(default_factory=dict)
        timestamp: datetime = field(default_factory=datetime.now)
    
    
    class StationaryAgentAgent:
        """Agent implementing Stationary Agent."""
        
        def __init__(self, name: str, config: StationaryAgentConfig = None):
            self.name = name
            self.config = config or StationaryAgentConfig()
            self.llm = ChatOpenAI(
                model=self.config.model_name,
                temperature=self.config.temperature
            )
            self.state = StationaryAgentState()
            self.execution_history: List[Dict[str, Any]] = []
        
        def process(self, input_data: Any) -> Dict[str, Any]:
            """Process input according to Stationary Agent."""
            self.state.current_step += 1
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", f"You are {self.name}, implementing Stationary Agent. "
                          f"Provide clear, structured responses."),
                ("human", "Process this input: {input_data}")
            ])
            
            chain = prompt | self.llm
            response = chain.invoke({"input_data": str(input_data)})
            
            result = {
                "agent": self.name,
                "step": self.state.current_step,
                "input": input_data,
                "output": str(response.content),
                "timestamp": datetime.now()
            }
            
            self.execution_history.append(result)
            self.state.results.append(result)
            
            return result
        
        def get_statistics(self) -> Dict[str, Any]:
            """Get agent statistics."""
            return {
                "name": self.name,
                "total_executions": len(self.execution_history),
                "current_step": self.state.current_step,
                "results_count": len(self.state.results)
            }
    
    
    class StationaryAgentPattern:
        """
        Stationary Agent - Enhanced implementation
        
        This pattern provides comprehensive functionality for stationary agent
        with full LangChain/LangGraph integration.
        
        Features:
        - Multi-agent coordination
        - State management
        - Execution tracking
        - LangGraph workflow integration
        """
        
        def __init__(self, config: StationaryAgentConfig = None):
            self.config = config or StationaryAgentConfig()
            self.agents: Dict[str, StationaryAgentAgent] = {}
            self.workflow = self._build_workflow()
            self.execution_log: List[Dict[str, Any]] = []
        
        def _build_workflow(self) -> StateGraph:
            """Build LangGraph workflow for pattern execution."""
            
            class WorkflowState(TypedDict):
                input: str
                processed: str
                output: str
                metadata: Dict[str, Any]
            
            def process_input(state: WorkflowState) -> WorkflowState:
                """Process input node."""
                return {
                    **state,
                    "processed": f"Processed: {state['input']}",
                    "metadata": {"step": "process_input"}
                }
            
            def generate_output(state: WorkflowState) -> WorkflowState:
                """Generate output node."""
                llm = ChatOpenAI(model=self.config.model_name, temperature=self.config.temperature)
                
                prompt = ChatPromptTemplate.from_messages([
                    ("system", "You are implementing Stationary Agent. Generate comprehensive output."),
                    ("human", "{input}")
                ])
                
                chain = prompt | llm
                response = chain.invoke({"input": state["processed"]})
                
                return {
                    **state,
                    "output": str(response.content),
                    "metadata": {**state.get("metadata", {}), "step": "generate_output"}
                }
            
            # Build workflow graph
            workflow = StateGraph(WorkflowState)
            workflow.add_node("process", process_input)
            workflow.add_node("generate", generate_output)
            workflow.set_entry_point("process")
            workflow.add_edge("process", "generate")
            workflow.add_edge("generate", END)
            
            return workflow.compile()
        
        def register_agent(self, agent: StationaryAgentAgent):
            """Register an agent with the pattern."""
            self.agents[agent.name] = agent
            if self.config.enable_logging:
                print(f"Registered agent: {agent.name}")
        
        def execute(self, input_data: Any, agent_name: str = None) -> Dict[str, Any]:
            """Execute the pattern with given input."""
            
            # Use workflow if no specific agent requested
            if agent_name is None:
                result = self.workflow.invoke({
                    "input": str(input_data),
                    "processed": "",
                    "output": "",
                    "metadata": {}
                })
                
                execution_record = {
                    "pattern": "Stationary Agent",
                    "input": input_data,
                    "output": result.get("output", ""),
                    "metadata": result.get("metadata", {}),
                    "timestamp": datetime.now()
                }
                
                self.execution_log.append(execution_record)
                return execution_record
            
            # Execute with specific agent
            agent = self.agents.get(agent_name)
            if not agent:
                raise ValueError(f"Agent {agent_name} not found")
            
            result = agent.process(input_data)
            self.execution_log.append(result)
            return result
        
        def run(self, input_data: Any) -> Dict[str, Any]:
            """Main execution method."""
            return self.execute(input_data)
        
        def get_statistics(self) -> Dict[str, Any]:
            """Get pattern statistics."""
            return {
                "pattern": "Stationary Agent",
                "total_executions": len(self.execution_log),
                "registered_agents": len(self.agents),
                "agents": [agent.get_statistics() for agent in self.agents.values()]
            }
    
    
    # Example usage
    if __name__ == "__main__":
        print("=== Stationary Agent Demo ===\n")
        
        # Create configuration
        config = StationaryAgentConfig(
            model_name="gpt-4",
            temperature=0.7,
            enable_logging=True
        )
        
        # Create pattern
        pattern = StationaryAgentPattern(config)
        
        # Create and register agents
        agent1 = StationaryAgentAgent("Agent1", config)
        agent2 = StationaryAgentAgent("Agent2", config)
        pattern.register_agent(agent1)
        pattern.register_agent(agent2)
        
        print("\nExecuting pattern workflow...")
        
        # Execute with workflow
        result1 = pattern.run("Demonstrate stationary agent functionality")
        print(f"\nWorkflow Result:")
        print(f"  Output: {result1['output'][:150]}...")
        
        # Execute with specific agents
        result2 = pattern.execute("Specific task for Agent1", agent_name="Agent1")
        print(f"\nAgent1 Result:")
        print(f"  Output: {result2['output'][:150]}...")
        
        result3 = pattern.execute("Specific task for Agent2", agent_name="Agent2")
        print(f"\nAgent2 Result:")
        print(f"  Output: {result3['output'][:150]}...")
        
        # Show statistics
        print("\n=== Statistics ===")
        stats = pattern.get_statistics()
        print(f"Pattern: {stats['pattern']}")
        print(f"Total Executions: {stats['total_executions']}")
        print(f"Registered Agents: {stats['registered_agents']}")
        print(f"\nAgent Statistics:")
        for agent_stats in stats['agents']:
            print(f"  {agent_stats['name']}: {agent_stats['total_executions']} executions")
    
        

    Agentic Architecture Pattern: Stationary Agent

    Pattern 86: 86 Stationary Agent – Line-by-Line Explanation

    Overview

    Pattern 86: Stationary Agent

    Purpose: Enhanced implementation with full LangChain/LangGraph integration


    Code Structure Analysis

    File Information

    • Pattern Number: 86
    • Pattern Name: 86 Stationary Agent
    • File Path: patterns\specialized_agent\86_stationary_agent.py
    • Total Lines: 247

    Detailed Code Explanation

    Module Docstring

    """
    Pattern 86: Stationary Agent
    
    Purpose: Enhanced implementation with full LangChain/LangGraph integration
    """
    

    Explanation: This module docstring describes the purpose and key features of the 86 Stationary Agent pattern.


    Imports Section

    from typing import Dict, Any, List, Optional
    from dataclasses import dataclass, field
    from datetime import datetime
    from enum import Enum
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_openai import ChatOpenAI
    from langgraph.graph import StateGraph, END
    from typing_extensions import TypedDict  # First 20 imports shown
    

    Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime: Date and time utilities – from enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration – from langgraph.graph import StateGraph, END: LangGraph for workflow orchestration – from typing_extensions import TypedDict: Type hints for better code documentation


    Classes

    StationaryAgentConfig (Line 18)

    Description: Configuration for Stationary Agent.

    StationaryAgentState (Line 27)

    Description: State information for Stationary Agent.

    StationaryAgentAgent (Line 36)

    Description: Agent implementing Stationary Agent.

    Methods:

    __init__ (Line 39)
    def __init__(self, name, config):
        """No method docstring."""
    

    Purpose: No method docstring.

    Parameters: – self: Instance reference – name: Identifier or label – config: Configuration settings

    process (Line 49)
    def process(self, input_data):
        """Process input according to Stationary Agent."""
    

    Purpose: Process input according to Stationary Agent.

    Parameters: – self: Instance reference – input_data: Parameter for input_data

    get_statistics (Line 75)
    def get_statistics(self):
        """Get agent statistics."""
    

    Purpose: Get agent statistics.

    Parameters: – self: Instance reference

    StationaryAgentPattern (Line 85)

    Description: Stationary Agent – Enhanced implementation

    This pattern provides comprehensive functionality for stationary agent with full LangChain/LangGraph integration.

    Features: – Multi-agent coordination – State management – Execution tracking – LangGraph workflow integration

    Methods:

    __init__ (Line 99)
    def __init__(self, config):
        """No method docstring."""
    

    Purpose: No method docstring.

    Parameters: – self: Instance reference – config: Configuration settings

    _build_workflow (Line 105)
    def _build_workflow(self):
        """Build LangGraph workflow for pattern execution."""
    

    Purpose: Build LangGraph workflow for pattern execution.

    Parameters: – self: Instance reference

    register_agent (Line 150)
    def register_agent(self, agent):
        """Register an agent with the pattern."""
    

    Purpose: Register an agent with the pattern.

    Parameters: – self: Instance reference – agent: Parameter for agent

    execute (Line 156)
    def execute(self, input_data, agent_name):
        """Execute the pattern with given input."""
    

    Purpose: Execute the pattern with given input.

    Parameters: – self: Instance reference – input_data: Parameter for input_data – agent_name: Parameter for agent_name

    run (Line 188)
    def run(self, input_data):
        """Main execution method."""
    

    Purpose: Main execution method.

    Parameters: – self: Instance reference – input_data: Parameter for input_data

    get_statistics (Line 192)
    def get_statistics(self):
        """Get pattern statistics."""
    

    Purpose: Get pattern statistics.

    Parameters: – self: Instance reference

    WorkflowState (Line 108)

    Description: No class docstring.


    Design Patterns Used

    • Strategy Pattern: Different execution strategies for various scenarios
    • State Pattern: Managing different states in the workflow
    • Chain of Responsibility: Sequential processing through workflow nodes
    • Observer Pattern: Tracking and monitoring pattern execution
    • Factory Pattern: Creating instances with proper configuration

    Usage Example

    # Initialize the pattern
    pattern = StationaryAgentConfig(name="example", model_name="gpt-4")
    
    # Configure and execute
    result = pattern.execute(data={"key": "value"})
    
    # Check results
    print(result)
    

    Key Concepts

    • LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
    • Workflow Orchestration: LangGraph manages execution flow through nodes
    • State Management: TypedDict defines workflow state structure
    • Error Handling: Comprehensive error handling and fallback mechanisms
    • Extensibility: Designed for easy customization and extension

    Extension Points

    This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities

  • Agentic Architecture Pattern: Mobile Agent

    
        """
    Pattern 85: Mobile Agent Pattern with LangChain/LangGraph
    
    Purpose: Agent migration between execution environments with intelligent routing.
    Enables agents to move across environments while preserving state and optimizing placement.
    """
    
    from typing import Dict, Any, List, Optional, TypedDict
    from dataclasses import dataclass, field
    from datetime import datetime
    from enum import Enum
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_openai import ChatOpenAI
    from langgraph.graph import StateGraph, END
    import json
    import time
    
    
    class AgentStatus(Enum):
        """Agent execution status."""
        IDLE = "idle"
        EXECUTING = "executing"
        MIGRATING = "migrating"
        SUSPENDED = "suspended"
        TERMINATED = "terminated"
    
    
    class MigrationType(Enum):
        """Types of migration."""
        MANUAL = "manual"
        LOAD_BALANCING = "load_balancing"
        RESOURCE_OPTIMIZATION = "resource_optimization"
        FAULT_TOLERANCE = "fault_tolerance"
        STRATEGIC = "strategic"
    
    
    class EnvironmentCapability(Enum):
        """Environment capabilities."""
        COMPUTE = "compute"
        STORAGE = "storage"
        NETWORK = "network"
        GPU = "gpu"
        MEMORY = "memory"
    
    
    @dataclass
    class ExecutionEnvironment:
        """Execution environment definition."""
        env_id: str
        name: str
        capabilities: List[EnvironmentCapability]
        current_load: float = 0.0
        max_load: float = 100.0
        latency_ms: float = 10.0
        available: bool = True
        metadata: Dict[str, Any] = field(default_factory=dict)
    
    
    @dataclass
    class MigrationRecord:
        """Record of agent migration."""
        migration_id: str
        agent_id: str
        from_env: str
        to_env: str
        migration_type: MigrationType
        reason: str
        timestamp: datetime
        duration_ms: float
        success: bool
        state_size_bytes: int
    
    
    @dataclass
    class AgentTask:
        """Task executed by agent."""
        task_id: str
        description: str
        required_capabilities: List[EnvironmentCapability]
        execution_time: float
        result: Optional[str] = None
    
    
    class MigrationState(TypedDict):
        """State for migration workflow."""
        agent_id: str
        current_env: str
        target_env: Optional[str]
        migration_reason: str
        agent_state: Dict[str, Any]
        task_history: List[Dict[str, Any]]
        environments: Dict[str, Dict[str, Any]]
        migration_score: float
        recommendations: List[str]
        approved: bool
    
    
    class MobileAgent:
        """
        Mobile Agent with state preservation and migration capabilities.
        """
        
        def __init__(self, agent_id: str, name: str, initial_env: str):
            self.agent_id = agent_id
            self.name = name
            self.current_env = initial_env
            self.status = AgentStatus.IDLE
            self.state_data: Dict[str, Any] = {}
            self.task_history: List[AgentTask] = []
            self.migration_history: List[MigrationRecord] = []
            self.created_at = datetime.now()
        
        def serialize_state(self) -> Dict[str, Any]:
            """Serialize agent state for migration."""
            return {
                "agent_id": self.agent_id,
                "name": self.name,
                "current_env": self.current_env,
                "status": self.status.value,
                "state_data": self.state_data,
                "task_history": [
                    {
                        "task_id": t.task_id,
                        "description": t.description,
                        "result": t.result
                    } for t in self.task_history
                ],
                "created_at": self.created_at.isoformat()
            }
        
        def deserialize_state(self, state: Dict[str, Any], new_env: str):
            """Restore agent state after migration."""
            self.current_env = new_env
            self.state_data = state.get("state_data", {})
            # Task history restored from serialized data
        
        def execute_task(self, task: AgentTask) -> str:
            """Execute task in current environment."""
            self.status = AgentStatus.EXECUTING
            
            # Simulate task execution
            time.sleep(0.01)  # Simulate work
            
            result = f"Task {task.task_id} completed in {self.current_env}"
            task.result = result
            self.task_history.append(task)
            
            self.status = AgentStatus.IDLE
            return result
    
    
    class IntelligentMigrationManager:
        """
        Intelligent Migration Manager with LangGraph workflow.
        
        Features:
        - Environment analysis
        - Load balancing
        - Intelligent routing
        - State preservation
        - Migration optimization
        """
        
        def __init__(self, model_name: str = "gpt-4", temperature: float = 0.3):
            self.llm = ChatOpenAI(model=model_name, temperature=temperature)
            self.agents: Dict[str, MobileAgent] = {}
            self.environments: Dict[str, ExecutionEnvironment] = {}
            self.workflow = self._build_workflow()
            
            # Statistics
            self.stats = {
                "total_migrations": 0,
                "migrations_by_type": {t.value: 0 for t in MigrationType},
                "successful_migrations": 0,
                "failed_migrations": 0,
                "total_agents": 0,
                "avg_migration_time_ms": 0.0
            }
        
        def _build_workflow(self) -> StateGraph:
            """Build LangGraph workflow for intelligent migration."""
            workflow = StateGraph(MigrationState)
            
            workflow.add_node("analyze_environments", self._analyze_environments_node)
            workflow.add_node("evaluate_migration", self._evaluate_migration_node)
            workflow.add_node("select_target", self._select_target_node)
            workflow.add_node("validate_migration", self._validate_migration_node)
            workflow.add_node("generate_recommendations", self._generate_recommendations_node)
            
            workflow.set_entry_point("analyze_environments")
            workflow.add_edge("analyze_environments", "evaluate_migration")
            workflow.add_edge("evaluate_migration", "select_target")
            workflow.add_edge("select_target", "validate_migration")
            workflow.add_edge("validate_migration", "generate_recommendations")
            workflow.add_edge("generate_recommendations", END)
            
            return workflow.compile()
        
        def _analyze_environments_node(self, state: MigrationState) -> MigrationState:
            """Analyze available environments."""
            env_analysis = {}
            
            for env_id, env in self.environments.items():
                env_analysis[env_id] = {
                    "available": env.available,
                    "load": env.current_load,
                    "capacity": env.max_load - env.current_load,
                    "latency_ms": env.latency_ms,
                    "capabilities": [c.value for c in env.capabilities]
                }
            
            state["environments"] = env_analysis
            return state
        
        def _evaluate_migration_node(self, state: MigrationState) -> MigrationState:
            """Evaluate need for migration using LLM."""
            current_env = state["current_env"]
            reason = state["migration_reason"]
            environments = state["environments"]
            
            # Use LLM to evaluate migration necessity
            prompt = ChatPromptTemplate.from_template(
                "Evaluate if agent migration is beneficial:\n"
                "Current Environment: {current_env}\n"
                "Reason: {reason}\n"
                "Available Environments: {envs}\n\n"
                "Provide migration score (0.0-1.0) and brief analysis."
            )
            
            chain = prompt | self.llm
            response = chain.invoke({
                "current_env": current_env,
                "reason": reason,
                "envs": str(environments)
            })
            
            # Extract score (simplified - in production, parse LLM response)
            analysis = str(response.content)
            
            # Calculate score based on environment conditions
            if current_env in environments:
                current_load = environments[current_env]["load"]
                capacity = environments[current_env]["capacity"]
                
                if current_load > 80:
                    score = 0.9  # High load, strong migration candidate
                elif capacity  MigrationState:
            """Select optimal target environment."""
            current_env = state["current_env"]
            environments = state["environments"]
            migration_score = state["migration_score"]
            
            if migration_score  best_score:
                    best_score = total_score
                    best_env = env_id
            
            state["target_env"] = best_env
            return state
        
        def _validate_migration_node(self, state: MigrationState) -> MigrationState:
            """Validate migration is safe and beneficial."""
            target_env = state["target_env"]
            
            if target_env is None:
                state["approved"] = False
            else:
                # Check if target environment can accept agent
                target_data = state["environments"].get(target_env, {})
                
                if target_data.get("available") and target_data.get("capacity", 0) > 10:
                    state["approved"] = True
                else:
                    state["approved"] = False
                    state["target_env"] = None
            
            return state
        
        def _generate_recommendations_node(self, state: MigrationState) -> MigrationState:
            """Generate migration recommendations."""
            recommendations = []
            
            if state["approved"]:
                recommendations.append(f"Migrate to {state['target_env']} - optimal environment identified")
                recommendations.append("State preservation verified")
                recommendations.append("Target environment has sufficient capacity")
            else:
                recommendations.append("Migration not recommended at this time")
                
                if state["migration_score"]  Optional[MigrationRecord]:
            """Migrate agent with intelligent routing."""
            
            if agent_id not in self.agents:
                return None
            
            agent = self.agents[agent_id]
            start_time = time.time()
            
            # Prepare workflow state
            initial_state: MigrationState = {
                "agent_id": agent_id,
                "current_env": agent.current_env,
                "target_env": target_env,
                "migration_reason": reason,
                "agent_state": agent.serialize_state(),
                "task_history": [
                    {"task_id": t.task_id, "result": t.result}
                    for t in agent.task_history
                ],
                "environments": {},
                "migration_score": 0.0,
                "recommendations": [],
                "approved": False
            }
            
            # Run workflow
            result_state = self.workflow.invoke(initial_state)
            
            # Check if migration approved
            if not result_state["approved"]:
                return None
            
            final_target = result_state["target_env"]
            if not final_target or final_target not in self.environments:
                return None
            
            # Perform migration
            agent.status = AgentStatus.MIGRATING
            
            # Serialize state
            serialized_state = agent.serialize_state()
            state_size = len(json.dumps(serialized_state))
            
            # Update environment loads
            old_env = self.environments.get(agent.current_env)
            if old_env:
                old_env.current_load = max(0, old_env.current_load - 10)
            
            new_env = self.environments[final_target]
            new_env.current_load = min(new_env.max_load, new_env.current_load + 10)
            
            # Migrate agent
            agent.deserialize_state(serialized_state, final_target)
            agent.status = AgentStatus.IDLE
            
            duration = (time.time() - start_time) * 1000  # ms
            
            # Record migration
            record = MigrationRecord(
                migration_id=f"mig-{len(agent.migration_history) + 1}",
                agent_id=agent_id,
                from_env=result_state["current_env"],
                to_env=final_target,
                migration_type=migration_type,
                reason=reason,
                timestamp=datetime.now(),
                duration_ms=duration,
                success=True,
                state_size_bytes=state_size
            )
            
            agent.migration_history.append(record)
            
            # Update stats
            self.stats["total_migrations"] += 1
            self.stats["migrations_by_type"][migration_type.value] += 1
            self.stats["successful_migrations"] += 1
            
            total_time = self.stats["avg_migration_time_ms"] * (self.stats["total_migrations"] - 1) + duration
            self.stats["avg_migration_time_ms"] = total_time / self.stats["total_migrations"]
            
            return record
        
        def auto_balance(self):
            """Automatically balance agents across environments."""
            # Find overloaded environments
            overloaded = [
                env_id for env_id, env in self.environments.items()
                if env.current_load > 80
            ]
            
            # Migrate agents from overloaded environments
            for agent in self.agents.values():
                if agent.current_env in overloaded:
                    self.migrate_agent(
                        agent.agent_id,
                        reason="load_balancing",
                        migration_type=MigrationType.LOAD_BALANCING
                    )
        
        def get_statistics(self) -> Dict[str, Any]:
            """Get migration statistics."""
            return {
                **self.stats,
                "total_environments": len(self.environments),
                "active_agents": sum(1 for a in self.agents.values() if a.status != AgentStatus.TERMINATED)
            }
    
    
    class MobileAgentPattern:
        """
        Comprehensive Mobile Agent Pattern coordinator with LangChain/LangGraph.
        """
        
        def __init__(self, model_name: str = "gpt-4"):
            self.manager = IntelligentMigrationManager(model_name=model_name)
        
        def create_environment(
            self,
            env_id: str,
            name: str,
            capabilities: List[EnvironmentCapability],
            max_load: float = 100.0
        ) -> ExecutionEnvironment:
            """Create execution environment."""
            env = ExecutionEnvironment(
                env_id=env_id,
                name=name,
                capabilities=capabilities,
                max_load=max_load
            )
            self.manager.register_environment(env)
            return env
        
        def create_agent(
            self,
            agent_id: str,
            name: str,
            initial_env: str
        ) -> MobileAgent:
            """Create mobile agent."""
            agent = MobileAgent(agent_id, name, initial_env)
            self.manager.register_agent(agent)
            return agent
        
        def migrate(
            self,
            agent_id: str,
            target_env: Optional[str] = None,
            reason: str = "manual"
        ) -> Optional[MigrationRecord]:
            """Migrate agent."""
            return self.manager.migrate_agent(agent_id, target_env, reason)
        
        def auto_balance(self):
            """Auto-balance agents."""
            self.manager.auto_balance()
        
        def get_statistics(self) -> Dict[str, Any]:
            """Get statistics."""
            return self.manager.get_statistics()
    
    
    if __name__ == "__main__":
        print("=== Mobile Agent Pattern with LangChain/LangGraph Demo ===\n")
        
        print("Scenario 1: Environment Setup and Agent Creation")
        print("-" * 50)
        pattern = MobileAgentPattern(model_name="gpt-4")
        
        # Create environments
        env1 = pattern.create_environment(
            "env-001",
            "Production Server",
            [EnvironmentCapability.COMPUTE, EnvironmentCapability.STORAGE, EnvironmentCapability.NETWORK],
            max_load=100.0
        )
        print(f"✓ Created environment: {env1.name}")
        
        env2 = pattern.create_environment(
            "env-002",
            "Edge Node",
            [EnvironmentCapability.COMPUTE, EnvironmentCapability.NETWORK],
            max_load=80.0
        )
        print(f"✓ Created environment: {env2.name}")
        
        env3 = pattern.create_environment(
            "env-003",
            "GPU Cluster",
            [EnvironmentCapability.COMPUTE, EnvironmentCapability.GPU, EnvironmentCapability.MEMORY],
            max_load=120.0
        )
        print(f"✓ Created environment: {env3.name}")
        
        # Create agents
        agent1 = pattern.create_agent("agent-001", "DataProcessor", "env-001")
        print(f"\n✓ Created agent: {agent1.name} in {agent1.current_env}")
        
        agent2 = pattern.create_agent("agent-002", "APIWorker", "env-001")
        print(f"✓ Created agent: {agent2.name} in {agent2.current_env}")
        
        agent3 = pattern.create_agent("agent-003", "MLAgent", "env-002")
        print(f"✓ Created agent: {agent3.name} in {agent3.current_env}")
        
        print("\n" + "=" * 50)
        print("\nScenario 2: Task Execution")
        print("-" * 50)
        
        # Execute tasks
        task1 = AgentTask(
            task_id="task-001",
            description="Process data batch",
            required_capabilities=[EnvironmentCapability.COMPUTE],
            execution_time=0.1
        )
        
        result = agent1.execute_task(task1)
        print(f"Agent {agent1.name}: {result}")
        
        task2 = AgentTask(
            task_id="task-002",
            description="Handle API request",
            required_capabilities=[EnvironmentCapability.NETWORK],
            execution_time=0.05
        )
        
        result = agent2.execute_task(task2)
        print(f"Agent {agent2.name}: {result}")
        
        print("\n" + "=" * 50)
        print("\nScenario 3: Intelligent Migration")
        print("-" * 50)
        
        # Simulate high load on env-001
        pattern.manager.environments["env-001"].current_load = 85.0
        print(f"Simulated high load on {env1.name}: {env1.current_load}%")
        
        # Migrate agent1 with intelligent routing
        print(f"\nMigrating {agent1.name} with intelligent routing...")
        record = pattern.migrate(agent1.agent_id, reason="high_load_detected")
        
        if record:
            print(f"✓ Migration successful!")
            print(f"  From: {record.from_env}")
            print(f"  To: {record.to_env}")
            print(f"  Duration: {record.duration_ms:.2f}ms")
            print(f"  State Size: {record.state_size_bytes} bytes")
        else:
            print("✗ Migration not recommended")
        
        print("\n" + "=" * 50)
        print("\nScenario 4: Manual Migration")
        print("-" * 50)
        
        # Manual migration to GPU cluster
        print(f"Migrating {agent3.name} to GPU cluster...")
        record = pattern.migrate(agent3.agent_id, target_env="env-003", reason="gpu_required")
        
        if record:
            print(f"✓ Migration successful!")
            print(f"  From: {record.from_env}")
            print(f"  To: {record.to_env}")
            print(f"  Reason: {record.reason}")
        else:
            print("✗ Migration failed or not approved")
        
        print("\n" + "=" * 50)
        print("\nScenario 5: Auto Load Balancing")
        print("-" * 50)
        
        print("Running auto load balancing...")
        pattern.auto_balance()
        print("✓ Load balancing complete")
        
        print("\n" + "=" * 50)
        print("\nScenario 6: Migration History")
        print("-" * 50)
        
        for agent in pattern.manager.agents.values():
            if agent.migration_history:
                print(f"\n{agent.name} migration history:")
                for i, mig in enumerate(agent.migration_history, 1):
                    print(f"  {i}. {mig.from_env} → {mig.to_env} ({mig.migration_type.value})")
        
        print("\n" + "=" * 50)
        print("\nStatistics")
        print("-" * 50)
        
        stats = pattern.get_statistics()
        print(f"Total Agents: {stats['total_agents']}")
        print(f"Active Agents: {stats['active_agents']}")
        print(f"Total Migrations: {stats['total_migrations']}")
        print(f"Successful Migrations: {stats['successful_migrations']}")
        print(f"Average Migration Time: {stats['avg_migration_time_ms']:.2f}ms")
        print(f"Total Environments: {stats['total_environments']}")
        
        print(f"\nMigrations by Type:")
        for mig_type, count in stats['migrations_by_type'].items():
            if count > 0:
                print(f"  {mig_type}: {count}")
        
        print("\n" + "=" * 50)
        print("Mobile agent pattern demonstration complete!")
        print("=" * 50)
    
        

    Agentic Architecture Pattern: Mobile Agent

    Pattern 85: Mobile Agent Pattern – Line-by-Line Explanation

    Overview

    This pattern implements mobile agents that can migrate between execution environments, carrying their state and logic. It includes intelligent routing, load balancing, state serialization, and LLM-powered migration decisions.


    Code Explanation

    Lines 1-17: Imports

    """
    Pattern 85: Mobile Agent Pattern with LangChain/LangGraph
    
    Purpose: Mobile agents with migration, state transfer, and intelligent routing.
    Provides agent mobility with LLM-powered migration decisions and load balancing.
    """
    

    Explanation: Module docstring – pattern for agents that move between environments.

    from typing import Dict, Any, List, Optional, Callable, TypedDict
    from dataclasses import dataclass, field
    from datetime import datetime
    from enum import Enum
    import pickle
    import base64
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_openai import ChatOpenAI
    from langgraph.graph import StateGraph, END
    import json
    

    Explanation: Includes pickle for object serialization and base64 for encoding serialized state.


    Lines 19-34: Enum Definitions

    class AgentStatus(Enum):
        """Status of mobile agent."""
        IDLE = "idle"
        MIGRATING = "migrating"
        EXECUTING = "executing"
        WAITING = "waiting"
        TERMINATED = "terminated"
    

    Explanation: Five agent lifecycle states.

    class MigrationStrategy(Enum):
        """Strategy for agent migration."""
        LOAD_BALANCED = "load_balanced"  # Migrate to least loaded
        PROXIMITY = "proximity"           # Migrate to closest
        CAPABILITY = "capability"         # Migrate to most capable
        COST = "cost"                     # Migrate to cheapest
        INTELLIGENT = "intelligent"       # LLM decides
    

    Explanation: Five migration strategies: – LOAD_BALANCED: Choose environment with lowest load (CPU, memory, queue size) – PROXIMITY: Choose geographically closest (minimize latency) – CAPABILITY: Choose environment with best resources (GPU, special hardware) – COST: Choose cheapest environment (cloud cost optimization) – INTELLIGENT: Use LLM to analyze requirements and choose best environment


    Lines 36-92: Dataclasses

    @dataclass
    class Environment:
        """Execution environment for mobile agents."""
        id: str
        name: str
        capabilities: List[str]
        load: float = 0.0  # 0.0 to 1.0
        cost_per_hour: float = 0.0
        location: str = "unknown"
        active_agents: List[str] = field(default_factory=list)
    

    Explanation: Represents an execution environment: – capabilities: What this environment can do (e.g., [“gpu”, “high_memory”, “python3.11”]) – load: Current utilization (0.0 = idle, 1.0 = maxed out) – cost_per_hour: For cloud environments (e.g., $0.05/hour) – location: Geographic location (e.g., “us-east-1”, “europe-west”) – active_agents: List of agent IDs currently running

    @dataclass
    class AgentTask:
        """Task for mobile agent."""
        id: str
        code: Callable
        data: Dict[str, Any]
        requirements: List[str] = field(default_factory=list)
    

    Explanation: Task agent will execute: – code: Actual function to run – data: Input data – requirements: Required capabilities (e.g., [“gpu”, “tensorflow”])

    @dataclass
    class MigrationDecision:
        """Decision about agent migration."""
        should_migrate: bool
        target_environment: Optional[str]
        reasoning: str
        strategy: MigrationStrategy
        confidence: float
    

    Explanation: Migration decision made by system or LLM: – should_migrate: Yes/no decision – target_environment: Destination environment ID – reasoning: Why this decision was made – strategy: Which strategy was used – confidence: 0.0-1.0 confidence in decision

    @dataclass
    class MigrationRecord:
        """Record of agent migration."""
        agent_id: str
        from_environment: str
        to_environment: str
        timestamp: datetime = field(default_factory=datetime.now)
        state_size_bytes: int = 0
        migration_time_ms: float = 0.0
        success: bool = True
    

    Explanation: Historical record of migration event. Used for analysis and optimization.


    Lines 94-110: TypedDict State

    class MobileAgentState(TypedDict):
        """State for mobile agent workflow."""
        agent_id: str
        current_environment: str
        task: AgentTask
        agent_state: Dict[str, Any]
        serialized_state: Optional[str]
        migration_decision: Optional[MigrationDecision]
        target_environment: Optional[str]
        status: AgentStatus
        execution_result: Any
        insights: List[str]
    

    Explanation: Workflow state tracking agent identity, location, state, and migration decisions.


    Lines 112-160: MobileAgent Initialization

    class MobileAgent:
        """
        Mobile Agent with LangGraph workflow integration.
    
        Features:
        - Agent migration between environments
        - State serialization and transfer
        - Intelligent routing decisions
        - Load balancing
        - Capability-based placement
        - Cost optimization
        - Proximity awareness
        - Migration history tracking
        """
    

    Explanation: Main mobile agent class with intelligent migration.

    def __init__(
        self,
        agent_id: str,
        initial_environment: str,
        model_name: str = "gpt-4",
        temperature: float = 0.5
    ):
        self.agent_id = agent_id
        self.current_environment = initial_environment
        self.llm = ChatOpenAI(model=model_name, temperature=temperature)
    
        # Agent state
        self.state: Dict[str, Any] = {}
        self.status = AgentStatus.IDLE
    
        # Migration history
        self.migration_history: List[MigrationRecord] = []
    
        # Build workflow
        self.workflow = self._build_workflow()
    
        # Statistics
        self.stats = {
            "total_migrations": 0,
            "successful_migrations": 0,
            "failed_migrations": 0,
            "total_execution_time_ms": 0.0
        }
    

    Explanation: Initialize agent with ID, starting environment, LLM for decisions, state, and tracking.


    Lines 162-180: Workflow Building

    def _build_workflow(self) -> StateGraph:
        """Build LangGraph workflow."""
        workflow = StateGraph(MobileAgentState)
    
        workflow.add_node("analyze_requirements", self._analyze_requirements_node)
        workflow.add_node("decide_migration", self._decide_migration_node)
        workflow.add_node("serialize_state", self._serialize_state_node)
        workflow.add_node("migrate", self._migrate_node)
        workflow.add_node("execute_task", self._execute_task_node)
    

    Explanation: Create 5-node workflow: 1. Analyze requirements: Check what task needs 2. Decide migration: Determine if/where to migrate 3. Serialize state: Package agent state for transfer 4. Migrate: Move agent to new environment 5. Execute task: Run the actual task

        workflow.set_entry_point("analyze_requirements")
        workflow.add_edge("analyze_requirements", "decide_migration")
        workflow.add_edge("decide_migration", "serialize_state")
        workflow.add_edge("serialize_state", "migrate")
        workflow.add_edge("migrate", "execute_task")
        workflow.add_edge("execute_task", END)
    
        return workflow.compile()
    

    Explanation: Linear flow through migration and execution.


    Lines 182-204: Analyze Requirements Node

    def _analyze_requirements_node(self, state: MobileAgentState) -> MobileAgentState:
        """Analyze task requirements."""
        task = state["task"]
    
        state["insights"] = []
        state["insights"].append(f"Analyzing requirements for task {task.id}")
    
        if task.requirements:
            state["insights"].append(f"Task requires: {', '.join(task.requirements)}")
        else:
            state["insights"].append("Task has no specific requirements")
    
        return state
    

    Explanation: Extract and document task requirements. This informs migration decision.


    Lines 206-312: Decide Migration Node (LLM-Powered)

    def _decide_migration_node(
        self,
        state: MobileAgentState,
        environments: Dict[str, Environment],
        strategy: MigrationStrategy = MigrationStrategy.INTELLIGENT
    ) -> MobileAgentState:
        """Decide whether and where to migrate."""
        current_env_id = state["current_environment"]
        task = state["task"]
        current_env = environments.get(current_env_id)
    

    Explanation: Extract current environment and task information.

        if strategy == MigrationStrategy.LOAD_BALANCED:
            # Choose least loaded environment
            candidates = [(env_id, env) for env_id, env in environments.items()]
            candidates.sort(key=lambda x: x[1].load)
    
            if candidates and candidates[0][1].load < current_env.load:
                state["migration_decision"] = MigrationDecision(
                    should_migrate=True,
                    target_environment=candidates[0][0],
                    reasoning=f"Target has lower load ({candidates[0][1].load:.2f} vs {current_env.load:.2f})",
                    strategy=strategy,
                    confidence=0.9
                )
            else:
                state["migration_decision"] = MigrationDecision(
                    should_migrate=False,
                    target_environment=None,
                    reasoning="Current environment has best load",
                    strategy=strategy,
                    confidence=0.9
                )
    

    Explanation: LOAD_BALANCED strategy – choose environment with lowest load. Migrate only if better option exists.

        elif strategy == MigrationStrategy.CAPABILITY:
            # Choose environment with required capabilities
            suitable = []
            for env_id, env in environments.items():
                if all(req in env.capabilities for req in task.requirements):
                    suitable.append((env_id, env))
    
            if suitable:
                # Prefer environment with most capabilities
                suitable.sort(key=lambda x: len(x[1].capabilities), reverse=True)
    
                if suitable[0][0] != current_env_id:
                    state["migration_decision"] = MigrationDecision(
                        should_migrate=True,
                        target_environment=suitable[0][0],
                        reasoning=f"Target has required capabilities: {suitable[0][1].capabilities}",
                        strategy=strategy,
                        confidence=0.95
                    )
                else:
                    state["migration_decision"] = MigrationDecision(
                        should_migrate=False,
                        target_environment=None,
                        reasoning="Current environment has required capabilities",
                        strategy=strategy,
                        confidence=0.95
                    )
            else:
                state["migration_decision"] = MigrationDecision(
                    should_migrate=False,
                    target_environment=None,
                    reasoning="No suitable environments found",
                    strategy=strategy,
                    confidence=0.5
                )
    

    Explanation: CAPABILITY strategy – find environments that meet task requirements. Choose one with most capabilities.

        elif strategy == MigrationStrategy.INTELLIGENT:
            # Use LLM to decide
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are an expert in distributed systems and agent placement optimization."),
                ("user", """Analyze this mobile agent migration scenario:
    
    Agent ID: {agent_id}
    Current Environment: {current_env}
      - Capabilities: {current_capabilities}
      - Load: {current_load:.2f}
      - Cost: ${current_cost:.2f}/hour
    
    Task Requirements: {requirements}
    
    Available Environments:
    {environments}
    
    Provide migration decision as JSON with:
    - should_migrate: true/false
    - target_environment: ID of target (or null)
    - reasoning: Detailed explanation
    - confidence: 0.0-1.0 confidence score
    
    Consider: capabilities, load, cost, and migration overhead.""")
            ])
    
            # Format environments
            env_list = []
            for env_id, env in environments.items():
                if env_id != current_env_id:
                    env_list.append(
                        f"  - {env_id}: {env.name}\n"
                        f"    Capabilities: {env.capabilities}\n"
                        f"    Load: {env.load:.2f}\n"
                        f"    Cost: ${env.cost_per_hour:.2f}/hour"
                    )
    

    Explanation: INTELLIGENT strategy – prepare comprehensive context for LLM.

            chain = prompt | self.llm
            response = chain.invoke({
                "agent_id": state["agent_id"],
                "current_env": current_env.name,
                "current_capabilities": current_env.capabilities,
                "current_load": current_env.load,
                "current_cost": current_env.cost_per_hour,
                "requirements": task.requirements,
                "environments": "\n".join(env_list)
            })
    

    Explanation: Invoke LLM with full context about current and available environments.

            try:
                decision_data = json.loads(response.content)
    
                state["migration_decision"] = MigrationDecision(
                    should_migrate=decision_data.get("should_migrate", False),
                    target_environment=decision_data.get("target_environment"),
                    reasoning=decision_data.get("reasoning", "LLM decision"),
                    strategy=strategy,
                    confidence=decision_data.get("confidence", 0.5)
                )
    
            except Exception as e:
                # Fallback to staying
                state["migration_decision"] = MigrationDecision(
                    should_migrate=False,
                    target_environment=None,
                    reasoning=f"LLM decision failed: {str(e)}",
                    strategy=strategy,
                    confidence=0.0
                )
    

    Explanation: Parse LLM response. Fallback to staying if parsing fails.

        state["target_environment"] = state["migration_decision"].target_environment
        return state
    

    Lines 314-355: Serialize State Node

    def _serialize_state_node(self, state: MobileAgentState) -> MobileAgentState:
        """Serialize agent state for migration."""
        decision = state["migration_decision"]
    
        if not decision or not decision.should_migrate:
            state["serialized_state"] = None
            state["insights"].append("No migration needed, skipping serialization")
            return state
    

    Explanation: Skip serialization if not migrating.

        try:
            # Serialize agent state
            agent_state = state["agent_state"]
            serialized = pickle.dumps(agent_state)
            encoded = base64.b64encode(serialized).decode('utf-8')
    
            state["serialized_state"] = encoded
            state["insights"].append(f"State serialized: {len(encoded)} bytes")
    
        except Exception as e:
            state["insights"].append(f"Serialization failed: {str(e)}")
            state["serialized_state"] = None
            # Cancel migration
            state["migration_decision"].should_migrate = False
    
        return state
    

    Explanation: Serialize state using pickle, encode as base64 string for transfer. If serialization fails, cancel migration.

    Why pickle + base64? – Pickle: Python object serialization (preserves types, nested structures) – Base64: Encode binary as text (safe for JSON, network transfer)


    Lines 357-420: Migrate Node

    def _migrate_node(
        self,
        state: MobileAgentState,
        environments: Dict[str, Environment]
    ) -> MobileAgentState:
        """Migrate agent to new environment."""
        decision = state["migration_decision"]
    
        if not decision or not decision.should_migrate:
            state["insights"].append("Staying in current environment")
            return state
    

    Explanation: Skip migration if decision says stay.

        target_env_id = decision.target_environment
        current_env_id = state["current_environment"]
    
        if target_env_id not in environments:
            state["insights"].append(f"Target environment {target_env_id} not found")
            return state
    

    Explanation: Validate target environment exists.

        try:
            start_time = datetime.now()
    
            # Update environment tracking
            current_env = environments[current_env_id]
            target_env = environments[target_env_id]
    
            # Remove from current
            if state["agent_id"] in current_env.active_agents:
                current_env.active_agents.remove(state["agent_id"])
    
            # Add to target
            target_env.active_agents.append(state["agent_id"])
    
            # Update state
            state["current_environment"] = target_env_id
            state["status"] = AgentStatus.IDLE
    
            # Deserialize state in new environment
            if state["serialized_state"]:
                decoded = base64.b64decode(state["serialized_state"])
                deserialized_state = pickle.loads(decoded)
                state["agent_state"] = deserialized_state
    

    Explanation: Perform migration: 1. Remove agent from current environment’s active list 2. Add agent to target environment’s active list 3. Update agent’s current_environment 4. Deserialize state in new environment

            # Record migration
            end_time = datetime.now()
            migration_time = (end_time - start_time).total_seconds() * 1000
    
            record = MigrationRecord(
                agent_id=state["agent_id"],
                from_environment=current_env_id,
                to_environment=target_env_id,
                state_size_bytes=len(state["serialized_state"]) if state["serialized_state"] else 0,
                migration_time_ms=migration_time,
                success=True
            )
    
            self.migration_history.append(record)
            self.stats["total_migrations"] += 1
            self.stats["successful_migrations"] += 1
    
            state["insights"].append(
                f"Migrated from {current_env.name} to {target_env.name} in {migration_time:.2f}ms"
            )
    
        except Exception as e:
            state["insights"].append(f"Migration failed: {str(e)}")
            self.stats["failed_migrations"] += 1
    
        return state
    

    Explanation: Record migration metrics, update statistics, handle failures.


    Lines 422-457: Execute Task Node

    def _execute_task_node(self, state: MobileAgentState) -> MobileAgentState:
        """Execute agent task in current environment."""
        task = state["task"]
    
        state["status"] = AgentStatus.EXECUTING
    
        try:
            start_time = datetime.now()
    
            # Execute task code
            result = task.code(task.data)
    
            end_time = datetime.now()
            execution_time = (end_time - start_time).total_seconds() * 1000
    
            state["execution_result"] = result
            state["status"] = AgentStatus.IDLE
    
            self.stats["total_execution_time_ms"] += execution_time
    
            state["insights"].append(f"Task executed in {execution_time:.2f}ms")
    
        except Exception as e:
            state["execution_result"] = {"error": str(e)}
            state["status"] = AgentStatus.IDLE
            state["insights"].append(f"Execution failed: {str(e)}")
    
        return state
    

    Explanation: Execute task code with data, measure execution time, handle errors.


    Lines 482-640: Demo Code

    Scenario 1: Setup Environments

    # Create environments
    environments = {
        "env1": Environment(
            id="env1",
            name="Standard Compute",
            capabilities=["python", "basic"],
            load=0.3,
            cost_per_hour=0.05,
            location="us-east-1"
        ),
        "env2": Environment(
            id="env2",
            name="GPU Compute",
            capabilities=["python", "gpu", "tensorflow"],
            load=0.6,
            cost_per_hour=0.50,
            location="us-west-2"
        ),
        "env3": Environment(
            id="env3",
            name="High Memory",
            capabilities=["python", "high_memory"],
            load=0.2,
            cost_per_hour=0.15,
            location="europe-west"
        )
    }
    

    Explanation: Define three diverse environments with different capabilities, loads, costs, and locations.

    Scenario 2: Load Balanced Migration

    agent1 = pattern.create_agent("agent1", "env1")
    
    def simple_task(data: Dict) -> Dict:
        return {"result": data.get("value", 0) * 2}
    
    task1 = AgentTask(
        id="task1",
        code=simple_task,
        data={"value": 10},
        requirements=[]
    )
    
    result1 = pattern.execute_with_migration(
        "agent1",
        task1,
        strategy=MigrationStrategy.LOAD_BALANCED
    )
    

    Explanation: Execute task with load balancing. Agent migrates from env1 (load=0.3) to env3 (load=0.2) since it’s less loaded.

    Scenario 3: Capability-Based Migration

    agent2 = pattern.create_agent("agent2", "env1")
    
    def gpu_task(data: Dict) -> Dict:
        # Simulated GPU computation
        return {"processed": True, "gpu_accelerated": True}
    
    task2 = AgentTask(
        id="task2",
        code=gpu_task,
        data={"matrix": "large"},
        requirements=["gpu", "tensorflow"]
    )
    
    result2 = pattern.execute_with_migration(
        "agent2",
        task2,
        strategy=MigrationStrategy.CAPABILITY
    )
    

    Explanation: Task requires GPU. Agent migrates from env1 (no GPU) to env2 (has GPU + TensorFlow).

    Scenario 4: Intelligent LLM-Based Migration

    agent3 = pattern.create_agent("agent3", "env2")
    
    def memory_intensive_task(data: Dict) -> Dict:
        return {"memory_used": "high", "completed": True}
    
    task3 = AgentTask(
        id="task3",
        code=memory_intensive_task,
        data={"large_dataset": True},
        requirements=["high_memory"]
    )
    
    result3 = pattern.execute_with_migration(
        "agent3",
        task3,
        strategy=MigrationStrategy.INTELLIGENT
    )
    

    Explanation: LLM analyzes: – Current: env2 (GPU, expensive $0.50/hour, high load 0.6) – Task needs: high memory – env3 has high_memory, lower cost ($0.15/hour), lower load (0.2) – Decision: Migrate to env3 (better match, cheaper, less loaded)


    Migration Strategy Comparison

    Strategy Decision Basis Use Case
    LOAD_BALANCED Current load Distribute work evenly
    PROXIMITY Geographic location Minimize network latency
    CAPABILITY Required features Match task to hardware
    COST Price per hour Optimize cloud spending
    INTELLIGENT LLM analysis Complex multi-factor decisions

    Key Design Patterns

    1. Strategy Pattern: Different migration strategies
    2. State Pattern: Agent status management
    3. Memento Pattern: State serialization for migration
    4. Proxy Pattern: Agent acts as proxy in remote environment
    5. Observer Pattern: Migration history tracking

    Performance Considerations

    • Serialization: O(state_size) – pickle is fast but grows with state
    • Migration: O(network_latency + deserialization) – typically 10-100ms
    • Decision (Rule-Based): O(environments) – scan all options
    • Decision (LLM): O(LLM) – ~1-2 seconds
    • Execution: O(task_complexity) – depends on task

    Best Practices: – Minimize agent state size (only essentials) – Use compression for large states (gzip before base64) – Batch migrate multiple agents to same destination – Cache LLM decisions for similar scenarios – Consider migration cost vs execution time savings – Use load balancing for homogeneous tasks – Use capability matching for specialized tasks

    When to Migrate: – Migration time << execution time savings – Task requires unavailable capabilities – Current environment overloaded – Cost savings justify overhead

    Extension Points

    1. Add checkpointing during long-running tasks
    2. Implement secure migration with encryption
    3. Add fault tolerance (retry, rollback)
    4. Support partial migration (leave state, move code)
    5. Add network-aware routing (bandwidth, latency)
    6. Implement agent cloning for parallel execution
    7. Add environment auto-scaling based on agent demand
    8. Support heterogeneous code (not just Python)
  • Agentic Architecture Pattern: Service Agent

    
        """
    Pattern 84: Service Agent Pattern
    Purpose: Agent that provides services to other agents
    """
    
    from typing import Dict, Any, List, Callable
    from dataclasses import dataclass, field
    from datetime import datetime
    from enum import Enum
    from langchain_openai import ChatOpenAI
    from langchain_core.prompts import ChatPromptTemplate
    
    
    class ServiceType(Enum):
        """Types of services."""
        COMPUTATION = "computation"
        INFORMATION = "information"
        TRANSLATION = "translation"
        VALIDATION = "validation"
        TRANSFORMATION = "transformation"
    
    
    @dataclass
    class ServiceRequest:
        """A request for service."""
        id: str
        service_type: ServiceType
        payload: Any
        requester: str
        created_at: datetime = field(default_factory=datetime.now)
    
    
    @dataclass
    class ServiceResponse:
        """A response from service."""
        request_id: str
        result: Any
        success: bool
        error: str = None
        completed_at: datetime = field(default_factory=datetime.now)
    
    
    class ServiceAgent:
        """
        Service Agent - Provides services to other agents
        
        Offers specialized services that other agents can consume.
        Maintains service catalog and handles requests.
        """
        
        def __init__(self, name: str, services: List[ServiceType], model_name: str = "gpt-4"):
            self.name = name
            self.services = services
            self.llm = ChatOpenAI(model=model_name, temperature=0.5)
            self.request_history: List[ServiceRequest] = []
            self.response_history: List[ServiceResponse] = []
            
        def provides_service(self, service_type: ServiceType) -> bool:
            """Check if agent provides this service."""
            return service_type in self.services
        
        def process_request(self, request: ServiceRequest) -> ServiceResponse:
            """Process a service request."""
            self.request_history.append(request)
            
            if not self.provides_service(request.service_type):
                response = ServiceResponse(
                    request_id=request.id,
                    result=None,
                    success=False,
                    error=f"Service {request.service_type.value} not provided"
                )
                self.response_history.append(response)
                return response
            
            try:
                # Process using LLM
                prompt = ChatPromptTemplate.from_messages([
                    ("system", f"You are {self.name}, providing {request.service_type.value} service"),
                    ("human", "Process this request: {payload}")
                ])
                
                chain = prompt | self.llm
                result = chain.invoke({"payload": str(request.payload)})
                
                response = ServiceResponse(
                    request_id=request.id,
                    result=str(result.content),
                    success=True
                )
            except Exception as e:
                response = ServiceResponse(
                    request_id=request.id,
                    result=None,
                    success=False,
                    error=str(e)
                )
            
            self.response_history.append(response)
            return response
        
        def get_service_catalog(self) -> Dict[str, Any]:
            """Get catalog of provided services."""
            return {
                "agent": self.name,
                "services": [s.value for s in self.services],
                "requests_handled": len(self.request_history),
                "success_rate": sum(1 for r in self.response_history if r.success) / len(self.response_history) if self.response_history else 0
            }
    
    
    class ServiceAgentPattern:
        """
        Service Agent Pattern - Manages service agents and routing
        """
        
        def __init__(self):
            self.agents: Dict[str, ServiceAgent] = {}
            self.service_registry: Dict[ServiceType, List[str]] = {}
            
        def register_agent(self, agent: ServiceAgent):
            """Register a service agent."""
            self.agents[agent.name] = agent
            
            # Update service registry
            for service in agent.services:
                if service not in self.service_registry:
                    self.service_registry[service] = []
                self.service_registry[service].append(agent.name)
            
            print(f"✓ Registered: {agent.name} - Services: {[s.value for s in agent.services]}")
        
        def find_provider(self, service_type: ServiceType) -> ServiceAgent:
            """Find agent that provides service."""
            if service_type in self.service_registry:
                agent_name = self.service_registry[service_type][0]
                return self.agents[agent_name]
            return None
        
        def request_service(self, service_type: ServiceType, payload: Any, requester: str = "client") -> ServiceResponse:
            """Request a service."""
            provider = self.find_provider(service_type)
            
            if not provider:
                return ServiceResponse(
                    request_id="no_provider",
                    result=None,
                    success=False,
                    error=f"No provider for {service_type.value}"
                )
            
            request = ServiceRequest(
                id=f"req_{len(provider.request_history) + 1}",
                service_type=service_type,
                payload=payload,
                requester=requester
            )
            
            return provider.process_request(request)
        
        def run(self, requests: List[tuple]) -> Dict[str, Any]:
            """Process multiple service requests."""
            results = []
            
            for service_type, payload in requests:
                response = self.request_service(service_type, payload)
                results.append({
                    "service": service_type.value,
                    "success": response.success,
                    "result": response.result[:100] if response.result else None,
                    "error": response.error
                })
            
            return {
                "pattern": "ServiceAgent",
                "total_requests": len(requests),
                "successful": sum(1 for r in results if r['success']),
                "results": results
            }
    
    
    if __name__ == "__main__":
        print("=== Service Agent Pattern Demo ===\n")
        
        pattern = ServiceAgentPattern()
        
        # Register service providers
        pattern.register_agent(ServiceAgent(
            "ComputeService", 
            [ServiceType.COMPUTATION, ServiceType.VALIDATION]
        ))
        pattern.register_agent(ServiceAgent(
            "DataService",
            [ServiceType.INFORMATION, ServiceType.TRANSFORMATION]
        ))
        pattern.register_agent(ServiceAgent(
            "LinguisticsService",
            [ServiceType.TRANSLATION]
        ))
        
        # Make service requests
        print("\nRequesting services...\n")
        requests = [
            (ServiceType.COMPUTATION, "Calculate fibonacci(10)"),
            (ServiceType.TRANSLATION, "Translate 'hello' to Spanish"),
            (ServiceType.INFORMATION, "Get user data for ID 12345"),
            (ServiceType.VALIDATION, "Validate email: user@example.com")
        ]
        
        summary = pattern.run(requests)
        
        print(f"\nPattern: {summary['pattern']}")
        print(f"Total Requests: {summary['total_requests']}")
        print(f"Successful: {summary['successful']}/{summary['total_requests']}")
        
        print("\nService Results:")
        for result in summary['results']:
            status = "✓" if result['success'] else "✗"
            print(f"  {status} {result['service']}: {result['result'] or result['error']}")
    
        

    Agentic Architecture Pattern: Service Agent

    Pattern 84: 84 Service Agent – Line-by-Line Explanation

    Overview

    Pattern 84: Service Agent Pattern Purpose: Agent that provides services to other agents


    Code Structure Analysis

    File Information

    • Pattern Number: 84
    • Pattern Name: 84 Service Agent
    • File Path: patterns\specialized_agent\84_service_agent.py
    • Total Lines: 220

    Detailed Code Explanation

    Module Docstring

    """
    Pattern 84: Service Agent Pattern
    Purpose: Agent that provides services to other agents
    """
    

    Explanation: This module docstring describes the purpose and key features of the 84 Service Agent pattern.


    Imports Section

    from typing import Dict, Any, List, Callable
    from dataclasses import dataclass, field
    from datetime import datetime
    from enum import Enum
    from langchain_openai import ChatOpenAI
    from langchain_core.prompts import ChatPromptTemplate  # First 20 imports shown
    

    Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Callable: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime: Date and time utilities – from enum import Enum: Enumeration support for fixed sets of values – from langchain_openai import ChatOpenAI: LangChain framework integration – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration


    Enumerations

    ServiceType (Line 14)

    class ServiceType(Enum):
        COMPUTATION
        INFORMATION
        TRANSLATION
        VALIDATION
        TRANSFORMATION
    

    Purpose: Types of services.

    Values: – COMPUTATION: Represents computation – INFORMATION: Represents information – TRANSLATION: Represents translation – VALIDATION: Represents validation – TRANSFORMATION: Represents transformation


    Classes

    ServiceType (Line 14)

    Description: Types of services.

    ServiceRequest (Line 24)

    Description: A request for service.

    ServiceResponse (Line 34)

    Description: A response from service.

    ServiceAgent (Line 43)

    Description: Service Agent – Provides services to other agents

    Offers specialized services that other agents can consume. Maintains service catalog and handles requests.

    Methods:

    __init__ (Line 51)
    def __init__(self, name, services, model_name):
        """No method docstring."""
    

    Purpose: No method docstring.

    Parameters: – self: Instance reference – name: Identifier or label – services: Parameter for services – model_name: LLM model to use (e.g., “gpt-4”)

    provides_service (Line 58)
    def provides_service(self, service_type):
        """Check if agent provides this service."""
    

    Purpose: Check if agent provides this service.

    Parameters: – self: Instance reference – service_type: Parameter for service_type

    process_request (Line 62)
    def process_request(self, request):
        """Process a service request."""
    

    Purpose: Process a service request.

    Parameters: – self: Instance reference – request: Parameter for request

    get_service_catalog (Line 102)
    def get_service_catalog(self):
        """Get catalog of provided services."""
    

    Purpose: Get catalog of provided services.

    Parameters: – self: Instance reference

    ServiceAgentPattern (Line 112)

    Description: Service Agent Pattern – Manages service agents and routing

    Methods:

    __init__ (Line 117)
    def __init__(self):
        """No method docstring."""
    

    Purpose: No method docstring.

    Parameters: – self: Instance reference

    register_agent (Line 121)
    def register_agent(self, agent):
        """Register a service agent."""
    

    Purpose: Register a service agent.

    Parameters: – self: Instance reference – agent: Parameter for agent

    find_provider (Line 133)
    def find_provider(self, service_type):
        """Find agent that provides service."""
    

    Purpose: Find agent that provides service.

    Parameters: – self: Instance reference – service_type: Parameter for service_type

    request_service (Line 140)
    def request_service(self, service_type, payload, requester):
        """Request a service."""
    

    Purpose: Request a service.

    Parameters: – self: Instance reference – service_type: Parameter for service_type – payload: Parameter for payload – requester: Parameter for requester

    run (Line 161)
    def run(self, requests):
        """Process multiple service requests."""
    

    Purpose: Process multiple service requests.

    Parameters: – self: Instance reference – requests: Parameter for requests


    Design Patterns Used

    • Strategy Pattern: Different execution strategies for various scenarios
    • State Pattern: Managing different states in the workflow
    • Chain of Responsibility: Sequential processing through workflow nodes
    • Observer Pattern: Tracking and monitoring pattern execution
    • Factory Pattern: Creating instances with proper configuration

    Usage Example

    # Initialize the pattern
    pattern = ServiceType(name="example", model_name="gpt-4")
    
    # Configure and execute
    result = pattern.execute(data={"key": "value"})
    
    # Check results
    print(result)
    

    Key Concepts

    • LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
    • Workflow Orchestration: LangGraph manages execution flow through nodes
    • State Management: TypedDict defines workflow state structure
    • Error Handling: Comprehensive error handling and fallback mechanisms
    • Extensibility: Designed for easy customization and extension

    Extension Points

    This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities