aishastr.live

Agentic Architecture Pattern: Adapter

Written by

admin

in


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

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *

More posts