"""
Pattern 97: Adapter Pattern

Adapt incompatible interfaces between agents or systems, allowing them to work together.
Translates data formats, protocols, and communication patterns.
"""

from typing import Dict, Any, List, Optional, Protocol
from dataclasses import dataclass, field
from datetime import datetime
from abc import ABC, abstractmethod
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
import json


class AgentInterface(ABC):
    """Standard agent interface."""
    
    @abstractmethod
    def process(self, data: Dict[str, Any]) -> Dict[str, Any]:
        """Process data in standard format."""
        pass


class LegacySystemA:
    """Legacy system with incompatible interface."""
    
    def execute_command(self, cmd: str, params: List[str]) -> str:
        """Legacy method with string-based interface."""
        return f"LegacyA executed: {cmd} with {params}"


class LegacySystemB:
    """Another legacy system with different interface."""
    
    def run_task(self, task_id: int, config: str) -> tuple:
        """Legacy method returning tuple."""
        return (task_id, f"Task {task_id} completed", config)


@dataclass
class AdaptedRequest:
    """Standardized request format."""
    action: str
    parameters: Dict[str, Any]
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class AdaptedResponse:
    """Standardized response format."""
    success: bool
    data: Any
    message: str
    timestamp: datetime = field(default_factory=datetime.now)


class LegacySystemAAdapter(AgentInterface):
    """Adapter for LegacySystemA."""
    
    def __init__(self):
        self.system = LegacySystemA()
    
    def process(self, data: Dict[str, Any]) -> Dict[str, Any]:
        """Adapt standard format to LegacySystemA format."""
        # Convert standard format to legacy format
        cmd = data.get("action", "default")
        params = data.get("parameters", {})
        params_list = [f"{k}={v}" for k, v in params.items()]
        
        # Call legacy method
        result = self.system.execute_command(cmd, params_list)
        
        # Convert result back to standard format
        return {
            "success": True,
            "data": result,
            "message": "LegacySystemA adapted successfully"
        }


class LegacySystemBAdapter(AgentInterface):
    """Adapter for LegacySystemB."""
    
    def __init__(self):
        self.system = LegacySystemB()
    
    def process(self, data: Dict[str, Any]) -> Dict[str, Any]:
        """Adapt standard format to LegacySystemB format."""
        # Convert standard format to legacy format
        task_id = hash(data.get("action", "")) % 10000
        config = json.dumps(data.get("parameters", {}))
        
        # Call legacy method
        result_tuple = self.system.run_task(task_id, config)
        
        # Convert tuple result back to standard format
        return {
            "success": True,
            "data": {
                "task_id": result_tuple[0],
                "message": result_tuple[1],
                "config": result_tuple[2]
            },
            "message": "LegacySystemB adapted successfully"
        }


class ModernAgent(AgentInterface):
    """Modern agent with standard interface."""
    
    def __init__(self, name: str, model_name: str = "gpt-4"):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=0.5)
    
    def process(self, data: Dict[str, Any]) -> Dict[str, Any]:
        """Process with LLM."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}. Process the request and provide a response."),
            ("human", "Action: {action}\nParameters: {parameters}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({
            "action": data.get("action", ""),
            "parameters": str(data.get("parameters", {}))
        })
        
        return {
            "success": True,
            "data": response.content,
            "message": f"Processed by {self.name}"
        }


class AdapterPattern:
    """
    Adapter Pattern - Manage adapters for different systems.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.model_name = model_name
        self.adapters: Dict[str, AgentInterface] = {}
        self.call_history: List[Dict[str, Any]] = []
    
    def register_adapter(self, name: str, adapter: AgentInterface):
        """Register an adapter."""
        self.adapters[name] = adapter
        print(f"Adapter registered: {name}")
    
    def process_request(self, adapter_name: str, request: AdaptedRequest) -> AdaptedResponse:
        """Process request through specified adapter."""
        if adapter_name not in self.adapters:
            return AdaptedResponse(
                success=False,
                data=None,
                message=f"Adapter {adapter_name} not found"
            )
        
        adapter = self.adapters[adapter_name]
        
        # Convert request to dict
        request_data = {
            "action": request.action,
            "parameters": request.parameters
        }
        
        # Process through adapter
        result = adapter.process(request_data)
        
        # Track call
        self.call_history.append({
            "adapter": adapter_name,
            "request": request_data,
            "response": result,
            "timestamp": datetime.now()
        })
        
        # Convert to standard response
        return AdaptedResponse(
            success=result.get("success", False),
            data=result.get("data"),
            message=result.get("message", "")
        )
    
    def run(self, adapter_name: str, action: str, 
            parameters: Dict[str, Any]) -> Dict[str, Any]:
        """Execute through adapter."""
        request = AdaptedRequest(action=action, parameters=parameters)
        response = self.process_request(adapter_name, request)
        
        return {
            "adapter": adapter_name,
            "request": {
                "action": action,
                "parameters": parameters
            },
            "response": {
                "success": response.success,
                "data": response.data,
                "message": response.message
            }
        }
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get adapter usage statistics."""
        return {
            "total_adapters": len(self.adapters),
            "total_calls": len(self.call_history),
            "calls_by_adapter": {
                name: sum(1 for call in self.call_history if call["adapter"] == name)
                for name in self.adapters.keys()
            }
        }


# Example usage
if __name__ == "__main__":
    # Create pattern
    pattern = AdapterPattern()
    
    # Register adapters
    pattern.register_adapter("legacy_a", LegacySystemAAdapter())
    pattern.register_adapter("legacy_b", LegacySystemBAdapter())
    pattern.register_adapter("modern", ModernAgent("ModernProcessor"))
    
    print("\n=== Adapter Pattern Demo ===\n")
    
    # Test LegacySystemA adapter
    print("1. Testing LegacySystemA Adapter:")
    result1 = pattern.run(
        "legacy_a",
        action="deploy",
        parameters={"env": "production", "version": "2.0"}
    )
    print(f"   Action: {result1['request']['action']}")
    print(f"   Response: {result1['response']['data']}")
    print(f"   Message: {result1['response']['message']}\n")
    
    # Test LegacySystemB adapter
    print("2. Testing LegacySystemB Adapter:")
    result2 = pattern.run(
        "legacy_b",
        action="backup",
        parameters={"database": "users", "incremental": True}
    )
    print(f"   Action: {result2['request']['action']}")
    print(f"   Response Data: {result2['response']['data']}")
    print(f"   Message: {result2['response']['message']}\n")
    
    # Test Modern agent
    print("3. Testing Modern Agent:")
    result3 = pattern.run(
        "modern",
        action="analyze",
        parameters={"dataset": "sales_q4", "metric": "revenue"}
    )
    print(f"   Action: {result3['request']['action']}")
    print(f"   Response: {result3['response']['data'][:100]}...")
    print(f"   Message: {result3['response']['message']}\n")
    
    # Show statistics
    print("=== Statistics ===")
    stats = pattern.get_statistics()
    print(f"Total adapters: {stats['total_adapters']}")
    print(f"Total calls: {stats['total_calls']}")
    print(f"Calls by adapter: {stats['calls_by_adapter']}")

    

Agentic Architecture Pattern: Adapter

Pattern 97: 97 Adapter – Line-by-Line Explanation

Overview

Pattern 97: Adapter Pattern

Adapt incompatible interfaces between agents or systems, allowing them to work together. Translates data formats, protocols, and communication patterns.

Code Structure Analysis

File Information

• Pattern Number: 97
• Pattern Name: 97 Adapter
• File Path: patterns\tool_resource\97_adapter.py
• Total Lines: 269

Detailed Code Explanation

Module Docstring

"""
Pattern 97: Adapter Pattern

Adapt incompatible interfaces between agents or systems, allowing them to work together.
Translates data formats, protocols, and communication patterns.
"""

Explanation: This module docstring describes the purpose and key features of the 97 Adapter pattern.

Imports Section

from typing import Dict, Any, List, Optional, Protocol
from dataclasses import dataclass, field
from datetime import datetime
from abc import ABC, abstractmethod
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
import json  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional, Protocol: 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 – import json: JSON parsing and serialization

Classes

AgentInterface (Line 17)

Description: Standard agent interface.

Methods:

process (Line 21)

def process(self, data):
    """Process data in standard format."""

Purpose: Process data in standard format.

Parameters: – self: Instance reference – data: Input data to process

LegacySystemA (Line 26)

Description: Legacy system with incompatible interface.

Methods:

execute_command (Line 29)

def execute_command(self, cmd, params):
    """Legacy method with string-based interface."""

Purpose: Legacy method with string-based interface.

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

LegacySystemB (Line 34)

Description: Another legacy system with different interface.

Methods:

run_task (Line 37)

def run_task(self, task_id, config):
    """Legacy method returning tuple."""

Purpose: Legacy method returning tuple.

Parameters: – self: Instance reference – task_id: Parameter for task_id – config: Configuration settings

AdaptedRequest (Line 43)

Description: Standardized request format.

AdaptedResponse (Line 51)

Description: Standardized response format.

LegacySystemAAdapter (Line 59)

Description: Adapter for LegacySystemA.

Methods:

__init__ (Line 62)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

process (Line 65)

def process(self, data):
    """Adapt standard format to LegacySystemA format."""

Purpose: Adapt standard format to LegacySystemA format.

Parameters: – self: Instance reference – data: Input data to process

LegacySystemBAdapter (Line 83)

Description: Adapter for LegacySystemB.

Methods:

__init__ (Line 86)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

process (Line 89)

def process(self, data):
    """Adapt standard format to LegacySystemB format."""

Purpose: Adapt standard format to LegacySystemB format.

Parameters: – self: Instance reference – data: Input data to process

ModernAgent (Line 110)

Description: Modern agent with standard interface.

Methods:

__init__ (Line 113)

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”)

process (Line 117)

def process(self, data):
    """Process with LLM."""

Purpose: Process with LLM.

Parameters: – self: Instance reference – data: Input data to process

AdapterPattern (Line 137)

Description: Adapter Pattern – Manage adapters for different systems.

Methods:

__init__ (Line 142)

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

Purpose: No method docstring.

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

register_adapter (Line 147)

def register_adapter(self, name, adapter):
    """Register an adapter."""

Purpose: Register an adapter.

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

process_request (Line 152)

def process_request(self, adapter_name, request):
    """Process request through specified adapter."""

Purpose: Process request through specified adapter.

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

run (Line 187)

def run(self, adapter_name, action, parameters):
    """Execute through adapter."""

Purpose: Execute through adapter.

Parameters: – self: Instance reference – adapter_name: Parameter for adapter_name – action: Parameter for action – parameters: Parameter for parameters

get_statistics (Line 206)

def get_statistics(self):
    """Get adapter usage statistics."""

Purpose: Get adapter usage statistics.

Parameters: – self: Instance reference

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 = AgentInterface(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

    """
Pattern 96: Extension Pattern

Extend agent behavior by adding new capabilities without modifying existing code.
Extensions inherit from base agent and add specific features.
"""

from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
from abc import ABC, abstractmethod
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI


class BaseAgent(ABC):
    """Base agent that can be extended."""
    
    def __init__(self, name: str, model_name: str = "gpt-4"):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
    
    @abstractmethod
    def core_capability(self, input_data: str) -> str:
        """Core capability of the agent."""
        pass


@dataclass
class Extension:
    """Extension that adds capabilities."""
    name: str
    capability: Callable
    description: str
    priority: int = 0
    enabled: bool = True
    call_count: int = 0


class ExtensibleAgent(BaseAgent):
    """Agent that supports extensions."""
    
    def __init__(self, name: str, model_name: str = "gpt-4"):
        super().__init__(name, model_name)
        self.extensions: Dict[str, Extension] = {}
    
    def core_capability(self, input_data: str) -> str:
        """Core capability - basic text processing."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, a base AI agent."),
            ("human", "{input}")
        ])
        chain = prompt | self.llm
        response = chain.invoke({"input": input_data})
        return str(response.content)
    
    def add_extension(self, name: str, capability: Callable, 
                     description: str, priority: int = 0):
        """Add an extension."""
        extension = Extension(
            name=name,
            capability=capability,
            description=description,
            priority=priority
        )
        self.extensions[name] = extension
        print(f"Extension added: {name} (priority: {priority})")
    
    def remove_extension(self, name: str):
        """Remove an extension."""
        if name in self.extensions:
            del self.extensions[name]
            print(f"Extension removed: {name}")
    
    def enable_extension(self, name: str):
        """Enable an extension."""
        if name in self.extensions:
            self.extensions[name].enabled = True
    
    def disable_extension(self, name: str):
        """Disable an extension."""
        if name in self.extensions:
            self.extensions[name].enabled = False
    
    def execute_with_extensions(self, input_data: str) -> Dict[str, Any]:
        """Execute with all enabled extensions."""
        results = {
            "core_result": self.core_capability(input_data),
            "extension_results": {}
        }
        
        # Sort extensions by priority (higher priority first)
        sorted_extensions = sorted(
            self.extensions.values(),
            key=lambda e: e.priority,
            reverse=True
        )
        
        # Execute enabled extensions
        for ext in sorted_extensions:
            if ext.enabled:
                try:
                    result = ext.capability(input_data)
                    results["extension_results"][ext.name] = result
                    ext.call_count += 1
                except Exception as e:
                    results["extension_results"][ext.name] = f"Error: {str(e)}"
        
        return results
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get extension statistics."""
        return {
            "total_extensions": len(self.extensions),
            "enabled_extensions": sum(1 for e in self.extensions.values() if e.enabled),
            "extension_calls": {
                name: ext.call_count
                for name, ext in self.extensions.items()
            }
        }


class ExtensionPattern:
    """
    Extension Pattern - Manage extensible agents.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.model_name = model_name
        self.agents: Dict[str, ExtensibleAgent] = {}
    
    def create_agent(self, name: str) -> ExtensibleAgent:
        """Create a new extensible agent."""
        agent = ExtensibleAgent(name, self.model_name)
        self.agents[name] = agent
        return agent
    
    def run(self, agent_name: str, input_data: str) -> Dict[str, Any]:
        """Run agent with extensions."""
        if agent_name not in self.agents:
            return {"error": f"Agent {agent_name} not found"}
        
        agent = self.agents[agent_name]
        result = agent.execute_with_extensions(input_data)
        
        return {
            "agent": agent_name,
            "input": input_data,
            "results": result,
            "statistics": agent.get_statistics()
        }


# Example usage
if __name__ == "__main__":
    # Create pattern
    pattern = ExtensionPattern()
    
    # Create agent
    agent = pattern.create_agent("SmartAssistant")
    
    # Add extensions
    agent.add_extension(
        name="sentiment_analyzer",
        capability=lambda text: "positive" if "good" in text.lower() or "great" in text.lower() else "neutral",
        description="Analyzes sentiment of text",
        priority=10
    )
    
    agent.add_extension(
        name="word_counter",
        capability=lambda text: f"{len(text.split())} words",
        description="Counts words in text",
        priority=5
    )
    
    agent.add_extension(
        name="text_summarizer",
        capability=lambda text: text[:50] + "..." if len(text) > 50 else text,
        description="Summarizes long text",
        priority=3
    )
    
    # Test
    print("=== Extension Pattern Demo ===\n")
    
    # Run with extensions
    result = pattern.run("SmartAssistant", "This is a great day to learn about AI extensions!")
    
    print(f"Agent: {result['agent']}")
    print(f"\nInput: {result['input']}")
    print(f"\nCore Result: {result['results']['core_result'][:100]}...")
    print(f"\nExtension Results:")
    for ext_name, ext_result in result['results']['extension_results'].items():
        print(f"  - {ext_name}: {ext_result}")
    
    print(f"\nStatistics:")
    stats = result['statistics']
    print(f"  Total extensions: {stats['total_extensions']}")
    print(f"  Enabled extensions: {stats['enabled_extensions']}")
    print(f"  Extension calls: {stats['extension_calls']}")
    
    # Test with disabled extension
    print("\n\n=== Disabling sentiment_analyzer ===")
    agent.disable_extension("sentiment_analyzer")
    
    result2 = pattern.run("SmartAssistant", "Another test message")
    print(f"\nExtension Results:")
    for ext_name, ext_result in result2['results']['extension_results'].items():
        print(f"  - {ext_name}: {ext_result}")

    

Agentic Architecture Pattern: Extension

Pattern 96: 96 Extension – Line-by-Line Explanation

Overview

Pattern 96: Extension Pattern

Extend agent behavior by adding new capabilities without modifying existing code. Extensions inherit from base agent and add specific features.

Code Structure Analysis

File Information

• Pattern Number: 96
• Pattern Name: 96 Extension
• File Path: patterns\tool_resource\96_extension.py
• Total Lines: 211

Detailed Code Explanation

Module Docstring

"""
Pattern 96: Extension Pattern

Extend agent behavior by adding new capabilities without modifying existing code.
Extensions inherit from base agent and add specific features.
"""

Explanation: This module docstring describes the purpose and key features of the 96 Extension pattern.

Imports Section

from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
from abc import ABC, abstractmethod
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 langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Classes

BaseAgent (Line 16)

Description: Base agent that can be extended.

Methods:

__init__ (Line 19)

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”)

core_capability (Line 24)

def core_capability(self, input_data):
    """Core capability of the agent."""

Purpose: Core capability of the agent.

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

Extension (Line 30)

Description: Extension that adds capabilities.

ExtensibleAgent (Line 40)

Description: Agent that supports extensions.

Methods:

__init__ (Line 43)

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”)

core_capability (Line 47)

def core_capability(self, input_data):
    """Core capability - basic text processing."""

Purpose: Core capability – basic text processing.

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

add_extension (Line 57)

def add_extension(self, name, capability, description, priority):
    """Add an extension."""

Purpose: Add an extension.

Parameters: – self: Instance reference – name: Identifier or label – capability: Parameter for capability – description: Parameter for description – priority: Parameter for priority

remove_extension (Line 69)

def remove_extension(self, name):
    """Remove an extension."""

Purpose: Remove an extension.

Parameters: – self: Instance reference – name: Identifier or label

enable_extension (Line 75)

def enable_extension(self, name):
    """Enable an extension."""

Purpose: Enable an extension.

Parameters: – self: Instance reference – name: Identifier or label

disable_extension (Line 80)

def disable_extension(self, name):
    """Disable an extension."""

Purpose: Disable an extension.

Parameters: – self: Instance reference – name: Identifier or label

execute_with_extensions (Line 85)

def execute_with_extensions(self, input_data):
    """Execute with all enabled extensions."""

Purpose: Execute with all enabled extensions.

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

get_statistics (Line 111)

def get_statistics(self):
    """Get extension statistics."""

Purpose: Get extension statistics.

Parameters: – self: Instance reference

ExtensionPattern (Line 123)

Description: Extension Pattern – Manage extensible agents.

Methods:

__init__ (Line 128)

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

Purpose: No method docstring.

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

create_agent (Line 132)

def create_agent(self, name):
    """Create a new extensible agent."""

Purpose: Create a new extensible agent.

Parameters: – self: Instance reference – name: Identifier or label

run (Line 138)

def run(self, agent_name, input_data):
    """Run agent with extensions."""

Purpose: Run agent with extensions.

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

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 = BaseAgent(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

    """
Pattern 95: Plugin Pattern

Dynamically extend agent capabilities by loading plugins at runtime.
Plugins implement a common interface and can be added/removed without
modifying the core agent.
"""

from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
from abc import ABC, abstractmethod
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI


class Plugin(ABC):
    """Base class for plugins."""
    
    @abstractmethod
    def get_name(self) -> str:
        """Return plugin name."""
        pass
    
    @abstractmethod
    def get_version(self) -> str:
        """Return plugin version."""
        pass
    
    @abstractmethod
    def execute(self, context: Dict[str, Any], input_data: Any) -> Any:
        """Execute plugin functionality."""
        pass
    
    def initialize(self):
        """Optional initialization."""
        pass
    
    def cleanup(self):
        """Optional cleanup."""
        pass


@dataclass
class PluginMetadata:
    """Metadata about an installed plugin."""
    name: str
    version: str
    enabled: bool = True
    load_time: datetime = field(default_factory=datetime.now)
    execution_count: int = 0


class PluginPattern:
    """
    Plugin Pattern - Extensible agent through plugins.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.5)
        self.plugins: Dict[str, Plugin] = {}
        self.metadata: Dict[str, PluginMetadata] = {}
    
    def register_plugin(self, plugin: Plugin):
        """Register a plugin."""
        plugin.initialize()
        name = plugin.get_name()
        self.plugins[name] = plugin
        self.metadata[name] = PluginMetadata(
            name=name,
            version=plugin.get_version()
        )
        print(f"Plugin registered: {name} v{plugin.get_version()}")
    
    def unregister_plugin(self, plugin_name: str):
        """Unregister a plugin."""
        if plugin_name in self.plugins:
            self.plugins[plugin_name].cleanup()
            del self.plugins[plugin_name]
            del self.metadata[plugin_name]
            print(f"Plugin unregistered: {plugin_name}")
    
    def enable_plugin(self, plugin_name: str):
        """Enable a plugin."""
        if plugin_name in self.metadata:
            self.metadata[plugin_name].enabled = True
    
    def disable_plugin(self, plugin_name: str):
        """Disable a plugin."""
        if plugin_name in self.metadata:
            self.metadata[plugin_name].enabled = False
    
    def execute_plugin(self, plugin_name: str, input_data: Any, context: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
        """Execute a specific plugin."""
        if plugin_name not in self.plugins:
            return {"error": f"Plugin {plugin_name} not found"}
        
        if not self.metadata[plugin_name].enabled:
            return {"error": f"Plugin {plugin_name} is disabled"}
        
        context = context or {}
        
        try:
            result = self.plugins[plugin_name].execute(context, input_data)
            self.metadata[plugin_name].execution_count += 1
            return {"success": True, "result": result}
        except Exception as e:
            return {"success": False, "error": str(e)}
    
    def run(self, task: str) -> Dict[str, Any]:
        """Run task using available plugins."""
        # Use LLM to determine which plugins to use
        plugin_list = [f"{name} ({meta.version})" for name, meta in self.metadata.items() if meta.enabled]
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"Available plugins: {', '.join(plugin_list)}. Determine which plugins to use for the task."),
            ("human", "{task}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"task": task})
        
        return {
            "task": task,
            "available_plugins": len(plugin_list),
            "recommendation": response.content
        }
    
    def list_plugins(self) -> List[Dict[str, Any]]:
        """List all registered plugins."""
        return [
            {
                "name": meta.name,
                "version": meta.version,
                "enabled": meta.enabled,
                "executions": meta.execution_count
            }
            for meta in self.metadata.values()
        ]


# Example plugins
class TextAnalysisPlugin(Plugin):
    """Plugin for text analysis."""
    
    def get_name(self) -> str:
        return "TextAnalysis"
    
    def get_version(self) -> str:
        return "1.0.0"
    
    def execute(self, context: Dict[str, Any], input_data: Any) -> Any:
        text = str(input_data)
        return {
            "length": len(text),
            "words": len(text.split()),
            "uppercase": text.upper(),
            "lowercase": text.lower()
        }


class MathPlugin(Plugin):
    """Plugin for math operations."""
    
    def get_name(self) -> str:
        return "Math"
    
    def get_version(self) -> str:
        return "1.0.0"
    
    def execute(self, context: Dict[str, Any], input_data: Any) -> Any:
        if isinstance(input_data, (list, tuple)):
            return {
                "sum": sum(input_data),
                "average": sum(input_data) / len(input_data) if input_data else 0,
                "min": min(input_data) if input_data else None,
                "max": max(input_data) if input_data else None
            }
        return {"error": "Input must be a list or tuple"}


class LoggerPlugin(Plugin):
    """Plugin for logging."""
    
    def __init__(self):
        self.logs = []
    
    def get_name(self) -> str:
        return "Logger"
    
    def get_version(self) -> str:
        return "1.0.0"
    
    def execute(self, context: Dict[str, Any], input_data: Any) -> Any:
        log_entry = f"[{datetime.now()}] {input_data}"
        self.logs.append(log_entry)
        return {"logged": True, "entry": log_entry}


# Example usage
if __name__ == "__main__":
    print("=== Plugin Pattern Demo ===\n")
    
    pattern = PluginPattern()
    
    # Register plugins
    pattern.register_plugin(TextAnalysisPlugin())
    pattern.register_plugin(MathPlugin())
    pattern.register_plugin(LoggerPlugin())
    
    print(f"\nRegistered plugins: {len(pattern.list_plugins())}\n")
    
    # Execute plugins
    result1 = pattern.execute_plugin("TextAnalysis", "Hello World!")
    print(f"TextAnalysis result: {result1['result']}\n")
    
    result2 = pattern.execute_plugin("Math", [1, 2, 3, 4, 5])
    print(f"Math result: {result2['result']}\n")
    
    result3 = pattern.execute_plugin("Logger", "Task completed successfully")
    print(f"Logger result: {result3['result']}\n")
    
    # List all plugins
    print("=== Plugin Status ===")
    for plugin in pattern.list_plugins():
        print(f"- {plugin['name']} v{plugin['version']}: "
              f"{'enabled' if plugin['enabled'] else 'disabled'}, "
              f"{plugin['executions']} executions")
    
    # Disable and test
    print("\n=== Testing Plugin Disable ===")
    pattern.disable_plugin("Math")
    result4 = pattern.execute_plugin("Math", [1, 2, 3])
    print(f"Attempt to use disabled plugin: {result4}")

    

Agentic Architecture Pattern: Plugin

Pattern 95: 95 Plugin – Line-by-Line Explanation

Overview

Pattern 95: Plugin Pattern

Dynamically extend agent capabilities by loading plugins at runtime. Plugins implement a common interface and can be added/removed without modifying the core agent.

Code Structure Analysis

File Information

• Pattern Number: 95
• Pattern Name: 95 Plugin
• File Path: patterns\tool_resource\95_plugin.py
• Total Lines: 235

Detailed Code Explanation

Module Docstring

"""
Pattern 95: Plugin Pattern

Dynamically extend agent capabilities by loading plugins at runtime.
Plugins implement a common interface and can be added/removed without
modifying the core agent.
"""

Explanation: This module docstring describes the purpose and key features of the 95 Plugin pattern.

Imports Section

from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
from abc import ABC, abstractmethod
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 langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Classes

Plugin (Line 17)

Description: Base class for plugins.

Methods:

get_name (Line 21)

def get_name(self):
    """Return plugin name."""

Purpose: Return plugin name.

Parameters: – self: Instance reference

get_version (Line 26)

def get_version(self):
    """Return plugin version."""

Purpose: Return plugin version.

Parameters: – self: Instance reference

execute (Line 31)

def execute(self, context, input_data):
    """Execute plugin functionality."""

Purpose: Execute plugin functionality.

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

initialize (Line 35)

def initialize(self):
    """Optional initialization."""

Purpose: Optional initialization.

Parameters: – self: Instance reference

cleanup (Line 39)

def cleanup(self):
    """Optional cleanup."""

Purpose: Optional cleanup.

Parameters: – self: Instance reference

PluginMetadata (Line 45)

Description: Metadata about an installed plugin.

PluginPattern (Line 54)

Description: Plugin Pattern – Extensible agent through plugins.

Methods:

__init__ (Line 59)

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

Purpose: No method docstring.

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

register_plugin (Line 64)

def register_plugin(self, plugin):
    """Register a plugin."""

Purpose: Register a plugin.

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

unregister_plugin (Line 75)

def unregister_plugin(self, plugin_name):
    """Unregister a plugin."""

Purpose: Unregister a plugin.

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

enable_plugin (Line 83)

def enable_plugin(self, plugin_name):
    """Enable a plugin."""

Purpose: Enable a plugin.

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

disable_plugin (Line 88)

def disable_plugin(self, plugin_name):
    """Disable a plugin."""

Purpose: Disable a plugin.

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

execute_plugin (Line 93)

def execute_plugin(self, plugin_name, input_data, context):
    """Execute a specific plugin."""

Purpose: Execute a specific plugin.

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

run (Line 110)

def run(self, task):
    """Run task using available plugins."""

Purpose: Run task using available plugins.

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

list_plugins (Line 129)

def list_plugins(self):
    """List all registered plugins."""

Purpose: List all registered plugins.

Parameters: – self: Instance reference

TextAnalysisPlugin (Line 143)

Description: Plugin for text analysis.

Methods:

get_name (Line 146)

def get_name(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

get_version (Line 149)

def get_version(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

execute (Line 152)

def execute(self, context, input_data):
    """No method docstring."""

Purpose: No method docstring.

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

MathPlugin (Line 162)

Description: Plugin for math operations.

Methods:

get_name (Line 165)

def get_name(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

get_version (Line 168)

def get_version(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

execute (Line 171)

def execute(self, context, input_data):
    """No method docstring."""

Purpose: No method docstring.

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

LoggerPlugin (Line 182)

Description: Plugin for logging.

Methods:

__init__ (Line 185)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

get_name (Line 188)

def get_name(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

get_version (Line 191)

def get_version(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

execute (Line 194)

def execute(self, context, input_data):
    """No method docstring."""

Purpose: No method docstring.

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

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 = Plugin(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

    """
Pattern 94: API Integration Pattern

Agents that integrate with external APIs (REST, GraphQL, etc.) to fetch data,
trigger actions, or interact with external systems.
"""

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
import json


class HTTPMethod(Enum):
    """HTTP methods."""
    GET = "GET"
    POST = "POST"
    PUT = "PUT"
    DELETE = "DELETE"
    PATCH = "PATCH"


@dataclass
class APIEndpoint:
    """API endpoint configuration."""
    name: str
    base_url: str
    path: str
    method: HTTPMethod
    headers: Dict[str, str] = field(default_factory=dict)
    auth_required: bool = False
    description: str = ""


@dataclass
class APICall:
    """Track API calls."""
    endpoint: str
    method: str
    params: Dict[str, Any]
    response: Any
    status_code: int
    timestamp: datetime = field(default_factory=datetime.now)
    execution_time: float = 0.0


class APIIntegrationPattern:
    """
    API Integration Pattern - Agent with API integration capabilities.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        self.endpoints: Dict[str, APIEndpoint] = {}
        self.call_history: List[APICall] = []
    
    def register_endpoint(self, endpoint: APIEndpoint):
        """Register an API endpoint."""
        self.endpoints[endpoint.name] = endpoint
    
    def call_api(self, endpoint_name: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
        """
        Simulate API call (in real implementation, would use requests library).
        """
        if endpoint_name not in self.endpoints:
            return {"error": f"Endpoint {endpoint_name} not found"}
        
        endpoint = self.endpoints[endpoint_name]
        params = params or {}
        
        # Simulate API response
        start = datetime.now()
        
        # Mock response based on endpoint
        mock_response = self._generate_mock_response(endpoint, params)
        
        execution_time = (datetime.now() - start).total_seconds()
        
        # Track call
        call = APICall(
            endpoint=endpoint_name,
            method=endpoint.method.value,
            params=params,
            response=mock_response,
            status_code=200,
            execution_time=execution_time
        )
        self.call_history.append(call)
        
        return {
            "endpoint": endpoint_name,
            "status": "success",
            "data": mock_response,
            "execution_time": execution_time
        }
    
    def _generate_mock_response(self, endpoint: APIEndpoint, params: Dict[str, Any]) -> Any:
        """Generate mock API response."""
        # Simulate different responses based on endpoint
        if "user" in endpoint.name.lower():
            return {"id": 1, "name": "John Doe", "email": "john@example.com"}
        elif "weather" in endpoint.name.lower():
            return {"temperature": 72, "condition": "sunny", "humidity": 45}
        elif "data" in endpoint.name.lower():
            return {"items": [{"id": i, "value": f"item_{i}"} for i in range(5)]}
        else:
            return {"message": "Success", "data": params}
    
    def run(self, task: str) -> Dict[str, Any]:
        """Execute task using API integration."""
        # Use LLM to decide which API to call
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You have access to these APIs: {list(self.endpoints.keys())}. Decide which to call based on the task."),
            ("human", "{task}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"task": task})
        
        # Simple parsing (in production, would use function calling)
        response_text = str(response.content)
        
        # Call APIs mentioned in response
        results = []
        for endpoint_name in self.endpoints.keys():
            if endpoint_name.lower() in response_text.lower():
                result = self.call_api(endpoint_name)
                results.append(result)
        
        return {
            "task": task,
            "endpoints_called": len(results),
            "results": results,
            "analysis": response.content
        }
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get API call statistics."""
        return {
            "total_calls": len(self.call_history),
            "endpoints_registered": len(self.endpoints),
            "calls_by_endpoint": {
                name: sum(1 for call in self.call_history if call.endpoint == name)
                for name in self.endpoints.keys()
            }
        }


# Example usage
if __name__ == "__main__":
    print("=== API Integration Pattern Demo ===\n")
    
    pattern = APIIntegrationPattern()
    
    # Register endpoints
    pattern.register_endpoint(APIEndpoint(
        name="GetUser",
        base_url="https://api.example.com",
        path="/users/{id}",
        method=HTTPMethod.GET,
        description="Fetch user information"
    ))
    
    pattern.register_endpoint(APIEndpoint(
        name="GetWeather",
        base_url="https://api.weather.com",
        path="/current",
        method=HTTPMethod.GET,
        description="Get current weather"
    ))
    
    pattern.register_endpoint(APIEndpoint(
        name="GetData",
        base_url="https://api.data.com",
        path="/items",
        method=HTTPMethod.GET,
        description="Fetch data items"
    ))
    
    # Test task
    result = pattern.run("Get user information and current weather")
    print(f"Task: {result['task']}")
    print(f"Endpoints called: {result['endpoints_called']}")
    print(f"\nResults:")
    for r in result['results']:
        print(f"  - {r['endpoint']}: {r['data']}")
    
    # Statistics
    print(f"\nStatistics: {pattern.get_statistics()}")

    

Agentic Architecture Pattern: Api Integration

Pattern 94: 94 Api Integration – Line-by-Line Explanation

Overview

Pattern 94: API Integration Pattern

Agents that integrate with external APIs (REST, GraphQL, etc.) to fetch data, trigger actions, or interact with external systems.

Code Structure Analysis

File Information

• Pattern Number: 94
• Pattern Name: 94 Api Integration
• File Path: patterns\tool_resource\94_api_integration.py
• Total Lines: 193

Detailed Code Explanation

Module Docstring

"""
Pattern 94: API Integration Pattern

Agents that integrate with external APIs (REST, GraphQL, etc.) to fetch data,
trigger actions, or interact with external systems.
"""

Explanation: This module docstring describes the purpose and key features of the 94 Api Integration 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
import json  # 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 – import json: JSON parsing and serialization

Enumerations

HTTPMethod (Line 17)

class HTTPMethod(Enum):
    GET
    POST
    PUT
    DELETE
    PATCH

Purpose: HTTP methods.

Values: – GET: Represents get – POST: Represents post – PUT: Represents put – DELETE: Represents delete – PATCH: Represents patch

Classes

HTTPMethod (Line 17)

Description: HTTP methods.

APIEndpoint (Line 27)

Description: API endpoint configuration.

APICall (Line 39)

Description: Track API calls.

APIIntegrationPattern (Line 50)

Description: API Integration Pattern – Agent with API integration capabilities.

Methods:

__init__ (Line 55)

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

Purpose: No method docstring.

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

register_endpoint (Line 60)

def register_endpoint(self, endpoint):
    """Register an API endpoint."""

Purpose: Register an API endpoint.

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

call_api (Line 64)

def call_api(self, endpoint_name, params):
    """Simulate API call (in real implementation, would use requests library)."""

Purpose: Simulate API call (in real implementation, would use requests library).

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

_generate_mock_response (Line 100)

def _generate_mock_response(self, endpoint, params):
    """Generate mock API response."""

Purpose: Generate mock API response.

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

run (Line 112)

def run(self, task):
    """Execute task using API integration."""

Purpose: Execute task using API integration.

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

get_statistics (Line 140)

def get_statistics(self):
    """Get API call statistics."""

Purpose: Get API call statistics.

Parameters: – self: Instance reference

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 = HTTPMethod(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

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

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

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

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

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

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