aishastr.live

Agentic Architecture Pattern: Wrapper Agent

Written by

admin

in


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

Comments

Leave a Reply

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

More posts