aishastr.live

Agentic Architecture Pattern: Request Response

Written by

admin

in


    """
Pattern 20: Request-Response Pattern

Synchronous communication where requestor waits for response from responder.
Implements classic client-server request-response with timeout handling.
"""

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


class RequestStatus(Enum):
    """Status of request."""
    PENDING = "pending"
    PROCESSING = "processing"
    COMPLETED = "completed"
    FAILED = "failed"
    TIMEOUT = "timeout"


@dataclass
class Request:
    """Request from client to server."""
    request_id: str
    sender: str
    receiver: str
    operation: str
    payload: Any
    timestamp: datetime = field(default_factory=datetime.now)
    timeout: float = 30.0


@dataclass
class Response:
    """Response from server to client."""
    request_id: str
    sender: str
    receiver: str
    result: Any
    status: RequestStatus
    error: Optional[str] = None
    timestamp: datetime = field(default_factory=datetime.now)


class RequestResponseState(TypedDict):
    """State for request-response workflow."""
    request: Optional[Request]
    response: Optional[Response]
    status: RequestStatus


class RequestResponseAgent:
    """Agent with request-response capabilities."""
    
    def __init__(self, name: str, role: str, model_name: str = "gpt-4"):
        self.name = name
        self.role = role
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.pattern: Optional['RequestResponsePattern'] = None
    
    def request(self, receiver: str, operation: str, payload: Any, timeout: float = 30.0) -> Response:
        """Send synchronous request and wait for response."""
        if not self.pattern:
            raise ValueError("Agent not connected to pattern")
        
        request_id = f"{self.name}_{datetime.now().timestamp()}"
        request = Request(
            request_id=request_id,
            sender=self.name,
            receiver=receiver,
            operation=operation,
            payload=payload,
            timeout=timeout
        )
        
        return self.pattern.process_request(request)
    
    def handle_request(self, request: Request) -> Response:
        """Handle incoming request and generate response."""
        try:
            # Process request using LLM
            prompt = ChatPromptTemplate.from_messages([
                ("system", f"You are {self.name}, a {self.role}. "
                          f"Process this request and provide a response."),
                ("human", f"Operation: {request.operation}\n"
                         f"From: {request.sender}\n"
                         f"Payload: {request.payload}")
            ])
            
            chain = prompt | self.llm
            result = chain.invoke({})
            
            return Response(
                request_id=request.request_id,
                sender=self.name,
                receiver=request.sender,
                result=str(result.content),
                status=RequestStatus.COMPLETED
            )
            
        except Exception as e:
            return Response(
                request_id=request.request_id,
                sender=self.name,
                receiver=request.sender,
                result=None,
                status=RequestStatus.FAILED,
                error=str(e)
            )


class RequestResponsePattern:
    """
    Request-Response Pattern - Synchronous agent communication.
    
    Uses LangGraph to model the request-response workflow with
    timeout handling and error recovery.
    """
    
    def __init__(self):
        self.agents: Dict[str, RequestResponseAgent] = {}
        self.request_log: List[Request] = []
        self.response_log: List[Response] = []
        self.graph = self._build_graph()
    
    def register_agent(self, agent: RequestResponseAgent):
        """Register agent with pattern."""
        self.agents[agent.name] = agent
        agent.pattern = self
    
    def _build_graph(self) -> StateGraph:
        """Build request-response workflow graph."""
        workflow = StateGraph(RequestResponseState)
        
        def route_request(state: RequestResponseState) -> RequestResponseState:
            """Route request to target agent."""
            request = state["request"]
            if not request:
                state["status"] = RequestStatus.FAILED
                return state
            
            state["status"] = RequestStatus.PROCESSING
            
            if request.receiver in self.agents:
                receiver = self.agents[request.receiver]
                response = receiver.handle_request(request)
                state["response"] = response
                state["status"] = response.status
            else:
                state["response"] = Response(
                    request_id=request.request_id,
                    sender="system",
                    receiver=request.sender,
                    result=None,
                    status=RequestStatus.FAILED,
                    error=f"Receiver {request.receiver} not found"
                )
                state["status"] = RequestStatus.FAILED
            
            return state
        
        workflow.add_node("process", route_request)
        workflow.set_entry_point("process")
        workflow.add_edge("process", END)
        
        return workflow.compile()
    
    def process_request(self, request: Request) -> Response:
        """Process request through pattern."""
        self.request_log.append(request)
        
        initial_state: RequestResponseState = {
            "request": request,
            "response": None,
            "status": RequestStatus.PENDING
        }
        
        # Execute workflow
        try:
            final_state = self.graph.invoke(initial_state)
            response = final_state["response"]
            
            if response:
                self.response_log.append(response)
                return response
            else:
                # Timeout or error
                timeout_response = Response(
                    request_id=request.request_id,
                    sender="system",
                    receiver=request.sender,
                    result=None,
                    status=RequestStatus.TIMEOUT,
                    error="Request timed out"
                )
                self.response_log.append(timeout_response)
                return timeout_response
                
        except Exception as e:
            error_response = Response(
                request_id=request.request_id,
                sender="system",
                receiver=request.sender,
                result=None,
                status=RequestStatus.FAILED,
                error=str(e)
            )
            self.response_log.append(error_response)
            return error_response
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get pattern statistics."""
        return {
            "total_requests": len(self.request_log),
            "total_responses": len(self.response_log),
            "completed": sum(1 for r in self.response_log if r.status == RequestStatus.COMPLETED),
            "failed": sum(1 for r in self.response_log if r.status == RequestStatus.FAILED),
            "timeout": sum(1 for r in self.response_log if r.status == RequestStatus.TIMEOUT)
        }
    
    def run(self, scenario: str) -> Dict[str, Any]:
        """Execute request-response scenario."""
        return {
            "pattern": "RequestResponse",
            "scenario": scenario,
            **self.get_statistics()
        }


# Example usage
if __name__ == "__main__":
    print("=== Request-Response Pattern Demo ===\n")
    
    # Create pattern
    pattern = RequestResponsePattern()
    
    # Create agents
    client = RequestResponseAgent("Client", "service client")
    server = RequestResponseAgent("Server", "data server")
    auth_service = RequestResponseAgent("AuthService", "authentication service")
    
    # Register agents
    pattern.register_agent(client)
    pattern.register_agent(server)
    pattern.register_agent(auth_service)
    
    print("1. Client making requests...")
    
    # Request 1: Authentication
    print("\n   Request 1: Authentication")
    response1 = client.request("AuthService", "authenticate", {"user": "john", "password": "***"})
    print(f"   Status: {response1.status.value}")
    if response1.status == RequestStatus.COMPLETED:
        print(f"   Result: {response1.result[:50]}...")
    
    # Request 2: Data query
    print("\n   Request 2: Data query")
    response2 = client.request("Server", "get_data", {"query": "SELECT * FROM users"})
    print(f"   Status: {response2.status.value}")
    if response2.status == RequestStatus.COMPLETED:
        print(f"   Result: {response2.result[:50]}...")
    
    # Request 3: Non-existent service (will fail)
    print("\n   Request 3: Non-existent service")
    response3 = client.request("InvalidService", "operation", {})
    print(f"   Status: {response3.status.value}")
    if response3.error:
        print(f"   Error: {response3.error}")
    
    # Statistics
    stats = pattern.get_statistics()
    print(f"\n2. Pattern Statistics:")
    print(f"   Total Requests: {stats['total_requests']}")
    print(f"   Completed: {stats['completed']}")
    print(f"   Failed: {stats['failed']}")
    print(f"   Timeout: {stats['timeout']}")
    
    result = pattern.run("Service communication")
    print(f"\nāœ“ Request-response pattern enabled synchronous agent communication!")

Agentic Architecture Pattern: Request Response

Pattern 20: 20 Request Response – Line-by-Line Explanation

Overview

Pattern 20: Request-Response Pattern

Synchronous communication where requestor waits for response from responder. Implements classic client-server request-response with timeout handling.

Code Structure Analysis

File Information

  • Pattern Number: 20
  • Pattern Name: 20 Request Response
  • File Path: patterns\communication\20_request_response.py
  • Total Lines: 286

Detailed Code Explanation

Module Docstring

"""
Pattern 20: Request-Response Pattern

Synchronous communication where requestor waits for response from responder.
Implements classic client-server request-response with timeout handling.
"""

Explanation: This module docstring describes the purpose and key features of the 20 Request Response 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

Enumerations

RequestStatus (Line 18)

class RequestStatus(Enum):
    PENDING
    PROCESSING
    COMPLETED
    FAILED
    TIMEOUT

Purpose: Status of request.

Values: – PENDING: Initial state, not yet processed – PROCESSING: Represents processing – COMPLETED: Successfully finished – FAILED: Execution failed – TIMEOUT: Represents timeout

Classes

RequestStatus (Line 18)

Description: Status of request.

Request (Line 28)

Description: Request from client to server.

Response (Line 40)

Description: Response from server to client.

RequestResponseState (Line 51)

Description: State for request-response workflow.

RequestResponseAgent (Line 58)

Description: Agent with request-response capabilities.

Methods:

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

Purpose: No method docstring.

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

request (Line 67)
def request(self, receiver, operation, payload, timeout):
    """Send synchronous request and wait for response."""

Purpose: Send synchronous request and wait for response.

Parameters: – self: Instance reference – receiver: Parameter for receiver – operation: Parameter for operation – payload: Parameter for payload – timeout: Parameter for timeout

handle_request (Line 84)
def handle_request(self, request):
    """Handle incoming request and generate response."""

Purpose: Handle incoming request and generate response.

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

RequestResponsePattern (Line 118)

Description: Request-Response Pattern – Synchronous agent communication.

Uses LangGraph to model the request-response workflow with timeout handling and error recovery.

Methods:

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

Purpose: No method docstring.

Parameters: – self: Instance reference

register_agent (Line 132)
def register_agent(self, agent):
    """Register agent with pattern."""

Purpose: Register agent with pattern.

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

_build_graph (Line 137)
def _build_graph(self):
    """Build request-response workflow graph."""

Purpose: Build request-response workflow graph.

Parameters: – self: Instance reference

process_request (Line 174)
def process_request(self, request):
    """Process request through pattern."""

Purpose: Process request through pattern.

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

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

Purpose: Get pattern statistics.

Parameters: – self: Instance reference

run (Line 227)
def run(self, scenario):
    """Execute request-response scenario."""

Purpose: Execute request-response scenario.

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

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