aishastr.live

Agentic Architecture Pattern: Gateway Agent

Written by

admin

in


    """
Pattern 87: Gateway Agent Pattern

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

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


class Protocol(Enum):
    """Communication protocols."""
    HTTP = "http"
    GRPC = "grpc"
    MQTT = "mqtt"
    WEBSOCKET = "websocket"
    REST = "rest"
    SOAP = "soap"
    CUSTOM = "custom"


@dataclass
class Message:
    """Message with protocol information."""
    content: Any
    protocol: Protocol
    headers: Dict[str, str] = field(default_factory=dict)
    metadata: Dict[str, Any] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class Translation:
    """Translation record."""
    original_message: Message
    translated_message: Message
    from_protocol: Protocol
    to_protocol: Protocol
    timestamp: datetime = field(default_factory=datetime.now)


class ProtocolTranslator:
    """Translates between protocols."""
    
    def __init__(self, llm: ChatOpenAI):
        self.llm = llm
        self.translations: List[Translation] = []
    
    def translate(self, message: Message, target_protocol: Protocol) -> Message:
        """Translate message to target protocol."""
        if message.protocol == target_protocol:
            return message
        
        # Use LLM for intelligent translation
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a protocol translator. Convert messages between protocols "
                      "while preserving semantic meaning."),
            ("human", """Translate this message:
From Protocol: {from_protocol}
To Protocol: {to_protocol}
Content: {content}
Headers: {headers}

Provide the translated content and appropriate headers for the target protocol.""")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({
            "from_protocol": message.protocol.value,
            "to_protocol": target_protocol.value,
            "content": str(message.content),
            "headers": str(message.headers)
        })
        
        translated = Message(
            content=response.content,
            protocol=target_protocol,
            headers={"translated": "true"},
            metadata={"original_protocol": message.protocol.value}
        )
        
        # Record translation
        self.translations.append(Translation(
            original_message=message,
            translated_message=translated,
            from_protocol=message.protocol,
            to_protocol=target_protocol
        ))
        
        return translated


class GatewayAgent:
    """Gateway agent for protocol translation and routing."""
    
    def __init__(self, name: str, model_name: str = "gpt-4"):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        self.translator = ProtocolTranslator(self.llm)
        self.supported_protocols: List[Protocol] = []
        self.routes: Dict[Protocol, Callable[[Message], None]] = {}
        self.messages_processed = 0
    
    def add_protocol_support(self, protocol: Protocol):
        """Add support for a protocol."""
        if protocol not in self.supported_protocols:
            self.supported_protocols.append(protocol)
            print(f"   {self.name}: Added support for {protocol.value}")
    
    def register_route(self, protocol: Protocol, handler: Callable[[Message], None]):
        """Register a message handler for a protocol."""
        self.routes[protocol] = handler
    
    def process_message(self, message: Message, target_protocol: Optional[Protocol] = None) -> Message:
        """Process and translate message."""
        self.messages_processed += 1
        
        print(f"\n   {self.name}: Processing message")
        print(f"   Source protocol: {message.protocol.value}")
        
        # Translate if needed
        if target_protocol and target_protocol != message.protocol:
            if target_protocol not in self.supported_protocols:
                raise ValueError(f"Protocol {target_protocol.value} not supported")
            
            print(f"   Translating to: {target_protocol.value}")
            translated = self.translator.translate(message, target_protocol)
            
            # Route to handler
            if target_protocol in self.routes:
                self.routes[target_protocol](translated)
            
            return translated
        
        return message
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get gateway statistics."""
        translation_counts = {}
        for trans in self.translator.translations:
            key = f"{trans.from_protocol.value} -> {trans.to_protocol.value}"
            translation_counts[key] = translation_counts.get(key, 0) + 1
        
        return {
            "messages_processed": self.messages_processed,
            "total_translations": len(self.translator.translations),
            "supported_protocols": [p.value for p in self.supported_protocols],
            "translation_routes": translation_counts
        }


class GatewayAgentPattern:
    """
    Gateway Agent Pattern - Protocol translation gateway.
    
    Gateway agents translate between different communication protocols,
    enabling interoperability between heterogeneous systems.
    """
    
    def __init__(self):
        self.agents: Dict[str, GatewayAgent] = {}
    
    def register_agent(self, agent: GatewayAgent):
        """Register gateway agent."""
        self.agents[agent.name] = agent
    
    def run(self, scenarios: List[Dict[str, Any]]) -> Dict[str, Any]:
        """Execute gateway scenarios."""
        print("=== Gateway Agent Pattern Demo ===\n")
        
        for scenario in scenarios:
            print(f"\nScenario: {scenario.get('description', 'Unknown')}")
            
            agent_name = scenario.get("agent")
            if agent_name not in self.agents:
                continue
            
            agent = self.agents[agent_name]
            message = Message(
                content=scenario.get("content"),
                protocol=Protocol(scenario.get("source_protocol")),
                headers=scenario.get("headers", {})
            )
            
            target = Protocol(scenario.get("target_protocol")) if scenario.get("target_protocol") else None
            result = agent.process_message(message, target)
            
            print(f"   Result protocol: {result.protocol.value}")
        
        # Statistics
        total_translations = sum(
            len(agent.translator.translations)
            for agent in self.agents.values()
        )
        
        return {
            "pattern": "GatewayAgent",
            "total_agents": len(self.agents),
            "total_translations": total_translations,
            "scenarios": len(scenarios)
        }


# Example usage
if __name__ == "__main__":
    print("=== Gateway Agent Pattern Demo ===\n")
    
    # Create pattern
    pattern = GatewayAgentPattern()
    
    # Create gateway agent
    print("1. Creating gateway agent...")
    gateway = GatewayAgent("APIGateway")
    
    # Add protocol support
    print("\n2. Adding protocol support...")
    gateway.add_protocol_support(Protocol.HTTP)
    gateway.add_protocol_support(Protocol.GRPC)
    gateway.add_protocol_support(Protocol.MQTT)
    gateway.add_protocol_support(Protocol.WEBSOCKET)
    
    # Register handlers
    def http_handler(msg: Message):
        print(f"     HTTP handler received: {str(msg.content)[:50]}...")
    
    def grpc_handler(msg: Message):
        print(f"     gRPC handler received: {str(msg.content)[:50]}...")
    
    gateway.register_route(Protocol.HTTP, http_handler)
    gateway.register_route(Protocol.GRPC, grpc_handler)
    
    pattern.register_agent(gateway)
    
    # Test scenarios
    print("\n3. Processing messages...")
    
    scenarios = [
        {
            "description": "Translate REST to gRPC",
            "agent": "APIGateway",
            "content": {"method": "GET", "path": "/api/users", "params": {"id": 123}},
            "source_protocol": "rest",
            "target_protocol": "grpc",
            "headers": {"Content-Type": "application/json"}
        },
        {
            "description": "Translate MQTT to WebSocket",
            "agent": "APIGateway",
            "content": {"topic": "sensors/temperature", "value": 23.5},
            "source_protocol": "mqtt",
            "target_protocol": "websocket",
            "headers": {"qos": "1"}
        },
        {
            "description": "Translate HTTP to MQTT",
            "agent": "APIGateway",
            "content": {"event": "user.login", "data": {"user_id": 456}},
            "source_protocol": "http",
            "target_protocol": "mqtt",
            "headers": {"Authorization": "Bearer token123"}
        }
    ]
    
    summary = pattern.run(scenarios)
    
    print(f"\n4. Summary:")
    print(f"   Total Translations: {summary['total_translations']}")
    print(f"   Scenarios Processed: {summary['scenarios']}")
    
    print("\n5. Gateway Statistics:")
    stats = gateway.get_statistics()
    print(f"   Messages Processed: {stats['messages_processed']}")
    print(f"   Supported Protocols: {', '.join(stats['supported_protocols'])}")
    print(f"   Translation Routes:")
    for route, count in stats['translation_routes'].items():
        print(f"     {route}: {count}")
    
    print("\nāœ“ Gateway agent successfully translated between protocols!")

Agentic Architecture Pattern: Gateway Agent

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

Overview

Pattern 87: Gateway Agent Pattern

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

Code Structure Analysis

File Information

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

Detailed Code Explanation

Module Docstring

"""
Pattern 87: Gateway Agent Pattern

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

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

Imports Section

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

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

Enumerations

Protocol (Line 16)

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

Purpose: Communication protocols.

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

Classes

Protocol (Line 16)

Description: Communication protocols.

Message (Line 28)

Description: Message with protocol information.

Translation (Line 38)

Description: Translation record.

ProtocolTranslator (Line 47)

Description: Translates between protocols.

Methods:

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

Purpose: No method docstring.

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

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

Purpose: Translate message to target protocol.

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

GatewayAgent (Line 98)

Description: Gateway agent for protocol translation and routing.

Methods:

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

Purpose: No method docstring.

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

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

Purpose: Add support for a protocol.

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

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

Purpose: Register a message handler for a protocol.

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

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

Purpose: Process and translate message.

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

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

Purpose: Get gateway statistics.

Parameters: – self: Instance reference

GatewayAgentPattern (Line 157)

Description: Gateway Agent Pattern – Protocol translation gateway.

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

Methods:

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

Purpose: No method docstring.

Parameters: – self: Instance reference

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

Purpose: Register gateway agent.

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

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

Purpose: Execute gateway scenarios.

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

Design Patterns Used

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

Usage Example

# Initialize the pattern
pattern = Protocol(name="example", model_name="gpt-4")

# Configure and execute
result = pattern.execute(data={"key": "value"})

# Check results
print(result)

Key Concepts

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

Extension Points

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

Comments

Leave a Reply

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

More posts