"""
Pattern 14: Broker Pattern

Decouples clients from service providers by acting as an intermediary.
Handles service discovery, load balancing, and request routing.
"""

from typing import Dict, Any, List, Optional
from dataclasses import dataclass
from datetime import datetime
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict


@dataclass
class ServiceProvider:
    """Registered service provider."""
    name: str
    service_type: str
    capabilities: List[str]
    load: int = 0
    max_capacity: int = 10


@dataclass
class ServiceRequest:
    """Client service request."""
    request_id: str
    client: str
    service_type: str
    capability: str
    data: Any
    
    
class BrokerState(TypedDict):
    """State for broker workflow."""
    request: Optional[ServiceRequest]
    provider: Optional[str]
    result: Optional[Any]
    error: Optional[str]


class BrokerPattern:
    """
    Broker Pattern - Service request router and load balancer.
    
    Uses LangGraph to manage service discovery and request routing.
    Automatically selects best available provider based on load.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.providers: Dict[str, ServiceProvider] = {}
        self.request_log: List[ServiceRequest] = []
        self.graph = self._build_graph()
    
    def register_provider(self, provider: ServiceProvider):
        """Register service provider with broker."""
        self.providers[provider.name] = provider
    
    def _find_provider(self, service_type: str, capability: str) -> Optional[ServiceProvider]:
        """Find available provider for service."""
        candidates = [
            p for p in self.providers.values()
            if p.service_type == service_type
            and capability in p.capabilities
            and p.load 
 StateGraph:
        """Build broker routing workflow."""
        workflow = StateGraph(BrokerState)
        
        def discover_service(state: BrokerState) -> BrokerState:
            """Find appropriate service provider."""
            request = state["request"]
            if not request:
                state["error"] = "No request provided"
                return state
            
            provider = self._find_provider(request.service_type, request.capability)
            if provider:
                state["provider"] = provider.name
                provider.load += 1
            else:
                state["error"] = f"No provider found for {request.service_type}/{request.capability}"
            
            return state
        
        def execute_service(state: BrokerState) -> BrokerState:
            """Execute service through provider."""
            if state.get("error"):
                return state
            
            provider_name = state["provider"]
            request = state["request"]
            
            if provider_name and request:
                provider = self.providers[provider_name]
                
                # Execute service using LLM
                prompt = ChatPromptTemplate.from_messages([
                    ("system", f"You are {provider.name}, a {provider.service_type} provider.\n"
                              f"Capabilities: {', '.join(provider.capabilities)}\n"
                              f"Execute the requested service."),
                    ("human", f"Service: {request.capability}\nData: {request.data}")
                ])
                
                chain = prompt | self.llm
                response = chain.invoke({})
                state["result"] = str(response.content)
                
                provider.load -= 1
            
            return state
        
        workflow.add_node("discover", discover_service)
        workflow.add_node("execute", execute_service)
        
        workflow.set_entry_point("discover")
        workflow.add_edge("discover", "execute")
        workflow.add_edge("execute", END)
        
        return workflow.compile()
    
    def handle_request(self, request: ServiceRequest) -> Dict[str, Any]:
        """Handle service request through broker."""
        self.request_log.append(request)
        
        initial_state: BrokerState = {
            "request": request,
            "provider": None,
            "result": None,
            "error": None
        }
        
        final_state = self.graph.invoke(initial_state)
        
        return {
            "request_id": request.request_id,
            "provider": final_state.get("provider"),
            "result": final_state.get("result"),
            "error": final_state.get("error"),
            "success": final_state.get("error") is None
        }
    
    def run(self, scenario: str) -> Dict[str, Any]:
        """Execute broker pattern."""
        return {
            "pattern": "Broker",
            "total_providers": len(self.providers),
            "available_providers": sum(1 for p in self.providers.values() if p.load < p.max_capacity),
            "total_requests": len(self.request_log),
            "scenario": scenario
        }


# Example usage
if __name__ == "__main__":
    print("=== Broker Pattern Demo ===\n")
    
    # Create broker
    broker = BrokerPattern()
    
    # Register providers
    print("1. Registering service providers...")
    broker.register_provider(ServiceProvider(
        name="DataAnalyzer1",
        service_type="analytics",
        capabilities=["analysis", "visualization"],
        max_capacity=5
    ))
    
    broker.register_provider(ServiceProvider(
        name="DataAnalyzer2",
        service_type="analytics",
        capabilities=["analysis", "reporting"],
        max_capacity=3
    ))
    
    broker.register_provider(ServiceProvider(
        name="TextProcessor1",
        service_type="nlp",
        capabilities=["summarization", "translation"],
        max_capacity=4
    ))
    
    # Submit requests
    print("\n2. Submitting service requests...")
    
    requests = [
        ServiceRequest(
            request_id="req1",
            client="ClientA",
            service_type="analytics",
            capability="analysis",
            data={"dataset": "sales_2024"}
        ),
        ServiceRequest(
            request_id="req2",
            client="ClientB",
            service_type="nlp",
            capability="summarization",
            data={"text": "Long document content..."}
        )
    ]
    
    for req in requests:
        print(f"\n   Processing {req.request_id}...")
        result = broker.handle_request(req)
        if result["success"]:
            print(f"   ✓ Routed to {result['provider']}")
        else:
            print(f"   ✗ Error: {result['error']}")
    
    # Summary
    summary = broker.run("Service routing")
    print(f"\n3. Summary:")
    print(f"   Providers: {summary['total_providers']}")
    print(f"   Available: {summary['available_providers']}")
    print(f"   Requests: {summary['total_requests']}")
    
    print("\n✓ Broker pattern successfully routed service requests!")

    

Agentic Architecture Pattern: Broker

Pattern 14: 14 Broker – Line-by-Line Explanation

Overview

Pattern 14: Broker Pattern

Decouples clients from service providers by acting as an intermediary. Handles service discovery, load balancing, and request routing.

Code Structure Analysis

File Information

• Pattern Number: 14
• Pattern Name: 14 Broker
• File Path: patterns\communication\14_broker.py
• Total Lines: 232

Detailed Code Explanation

Module Docstring

"""
Pattern 14: Broker Pattern

Decouples clients from service providers by acting as an intermediary.
Handles service discovery, load balancing, and request routing.
"""

Explanation: This module docstring describes the purpose and key features of the 14 Broker pattern.

Imports Section

from typing import Dict, Any, List, Optional
from dataclasses import dataclass
from datetime import datetime
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: 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 langgraph.graph import StateGraph, END: LangGraph for workflow orchestration – from typing_extensions import TypedDict: Type hints for better code documentation

Classes

ServiceProvider (Line 18)

Description: Registered service provider.

ServiceRequest (Line 28)

Description: Client service request.

BrokerState (Line 37)

Description: State for broker workflow.

BrokerPattern (Line 45)

Description: Broker Pattern – Service request router and load balancer.

Uses LangGraph to manage service discovery and request routing. Automatically selects best available provider based on load.

Methods:

__init__ (Line 53)

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_provider (Line 59)

def register_provider(self, provider):
    """Register service provider with broker."""

Purpose: Register service provider with broker.

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

_find_provider (Line 63)

def _find_provider(self, service_type, capability):
    """Find available provider for service."""

Purpose: Find available provider for service.

Parameters: – self: Instance reference – service_type: Parameter for service_type – capability: Parameter for capability

_build_graph (Line 78)

def _build_graph(self):
    """Build broker routing workflow."""

Purpose: Build broker routing workflow.

Parameters: – self: Instance reference

handle_request (Line 134)

def handle_request(self, request):
    """Handle service request through broker."""

Purpose: Handle service request through broker.

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

run (Line 155)

def run(self, scenario):
    """Execute broker pattern."""

Purpose: Execute broker pattern.

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