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

    """
Pattern 19: Event-Driven Pattern

Agents react to events in the system asynchronously.
Uses event bus for decoupled event publishing and subscription.
"""

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


@dataclass
class Event:
    """Event in the system."""
    event_type: str
    source: str
    data: Any
    timestamp: datetime = field(default_factory=datetime.now)
    priority: int = 1


class EventDrivenAgent:
    """Agent that reacts to events."""
    
    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.event_handlers: Dict[str, Callable] = {}
        self.event_bus: Optional['EventBus'] = None
    
    def subscribe(self, event_type: str, handler: Optional[Callable] = None):
        """Subscribe to event type."""
        if handler:
            self.event_handlers[event_type] = handler
        else:
            # Use default LLM-based handler
            self.event_handlers[event_type] = self._default_handler
        
        if self.event_bus:
            self.event_bus.subscribe(event_type, self)
    
    def _default_handler(self, event: Event) -> str:
        """Default event handler using LLM."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, a {self.role}. "
                      f"React to this event appropriately."),
            ("human", f"Event: {event.event_type}\n"
                     f"Source: {event.source}\n"
                     f"Data: {event.data}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        return str(response.content)
    
    def handle_event(self, event: Event) -> str:
        """Handle received event."""
        if event.event_type in self.event_handlers:
            return self.event_handlers[event.event_type](event)
        return f"{self.name}: Unhandled event {event.event_type}"
    
    def publish(self, event_type: str, data: Any, priority: int = 1):
        """Publish event to event bus."""
        if self.event_bus:
            event = Event(
                event_type=event_type,
                source=self.name,
                data=data,
                priority=priority
            )
            self.event_bus.publish(event)


class EventBus:
    """Central event bus for event-driven communication."""
    
    def __init__(self):
        self.subscribers: Dict[str, List[EventDrivenAgent]] = defaultdict(list)
        self.event_log: List[Event] = []
    
    def subscribe(self, event_type: str, agent: EventDrivenAgent):
        """Subscribe agent to event type."""
        if agent not in self.subscribers[event_type]:
            self.subscribers[event_type].append(agent)
            agent.event_bus = self
    
    def unsubscribe(self, event_type: str, agent: EventDrivenAgent):
        """Unsubscribe agent from event type."""
        if agent in self.subscribers[event_type]:
            self.subscribers[event_type].remove(agent)
    
    def publish(self, event: Event):
        """Publish event to all subscribers."""
        self.event_log.append(event)
        
        # Notify all subscribers
        if event.event_type in self.subscribers:
            for agent in self.subscribers[event.event_type]:
                response = agent.handle_event(event)
                # Optionally create follow-up event from response
                if response and "event:" in response.lower():
                    # Agent can trigger new events in response
                    pass
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get event bus statistics."""
        return {
            "total_events": len(self.event_log),
            "event_types": len(self.subscribers),
            "total_subscriptions": sum(len(agents) for agents in self.subscribers.values()),
            "events_by_type": {
                event_type: sum(1 for e in self.event_log if e.event_type == event_type)
                for event_type in self.subscribers.keys()
            }
        }


class EventDrivenPattern:
    """
    Event-Driven Pattern - Asynchronous event-based communication.
    
    Agents publish events to a central bus and subscribe to events
    they're interested in, enabling loose coupling and reactivity.
    """
    
    def __init__(self):
        self.event_bus = EventBus()
    
    def register_agent(self, agent: EventDrivenAgent):
        """Register agent with event bus."""
        agent.event_bus = self.event_bus
    
    def run(self, scenario: str) -> Dict[str, Any]:
        """Execute event-driven scenario."""
        stats = self.event_bus.get_statistics()
        return {
            "pattern": "EventDriven",
            "scenario": scenario,
            **stats
        }


# Example usage
if __name__ == "__main__":
    print("=== Event-Driven Pattern Demo ===\n")
    
    # Create pattern
    pattern = EventDrivenPattern()
    event_bus = pattern.event_bus
    
    # Create agents
    monitor = EventDrivenAgent("Monitor", "system monitor")
    logger = EventDrivenAgent("Logger", "event logger")
    alerter = EventDrivenAgent("Alerter", "alert manager")
    
    # Register and subscribe
    print("1. Setting up agents and subscriptions...")
    pattern.register_agent(monitor)
    pattern.register_agent(logger)
    pattern.register_agent(alerter)
    
    # Custom handler for critical events
    def critical_handler(event: Event) -> str:
        return f"ALERT: Critical event from {event.source}: {event.data}"
    
    monitor.subscribe("system_event")
    monitor.subscribe("user_action")
    logger.subscribe("system_event")
    logger.subscribe("user_action")
    logger.subscribe("error")
    alerter.subscribe("error", critical_handler)
    
    # Publish events
    print("\n2. Publishing events...")
    event_bus.publish(Event("system_event", "System", "Server started"))
    event_bus.publish(Event("user_action", "User1", "Login successful"))
    event_bus.publish(Event("error", "System", "Database connection failed", priority=5))
    
    # Statistics
    stats = event_bus.get_statistics()
    print(f"\n3. Event Bus Statistics:")
    print(f"   Total Events: {stats['total_events']}")
    print(f"   Event Types: {stats['event_types']}")
    print(f"   Subscriptions: {stats['total_subscriptions']}")
    print(f"   By Type: {stats['events_by_type']}")
    
    result = pattern.run("System monitoring")
    print(f"\n✓ Event-driven pattern enabled reactive agent communication!")

    

Agentic Architecture Pattern: Event Driven

Pattern 19: 19 Event Driven – Line-by-Line Explanation

Overview

Pattern 19: Event-Driven Pattern

Agents react to events in the system asynchronously. Uses event bus for decoupled event publishing and subscription.

Code Structure Analysis

File Information

• Pattern Number: 19
• Pattern Name: 19 Event Driven
• File Path: patterns\communication\19_event_driven.py
• Total Lines: 194

Detailed Code Explanation

Module Docstring

"""
Pattern 19: Event-Driven Pattern

Agents react to events in the system asynchronously.
Uses event bus for decoupled event publishing and subscription.
"""

Explanation: This module docstring describes the purpose and key features of the 19 Event Driven pattern.

Imports Section

from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
from collections import defaultdict
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 langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Classes

Event (Line 17)

Description: Event in the system.

EventDrivenAgent (Line 26)

Description: Agent that reacts to events.

Methods:

__init__ (Line 29)

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”)

subscribe (Line 36)

def subscribe(self, event_type, handler):
    """Subscribe to event type."""

Purpose: Subscribe to event type.

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

_default_handler (Line 47)

def _default_handler(self, event):
    """Default event handler using LLM."""

Purpose: Default event handler using LLM.

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

handle_event (Line 61)

def handle_event(self, event):
    """Handle received event."""

Purpose: Handle received event.

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

publish (Line 67)

def publish(self, event_type, data, priority):
    """Publish event to event bus."""

Purpose: Publish event to event bus.

Parameters: – self: Instance reference – event_type: Parameter for event_type – data: Input data to process – priority: Parameter for priority

EventBus (Line 79)

Description: Central event bus for event-driven communication.

Methods:

__init__ (Line 82)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

subscribe (Line 86)

def subscribe(self, event_type, agent):
    """Subscribe agent to event type."""

Purpose: Subscribe agent to event type.

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

unsubscribe (Line 92)

def unsubscribe(self, event_type, agent):
    """Unsubscribe agent from event type."""

Purpose: Unsubscribe agent from event type.

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

publish (Line 97)

def publish(self, event):
    """Publish event to all subscribers."""

Purpose: Publish event to all subscribers.

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

get_statistics (Line 110)

def get_statistics(self):
    """Get event bus statistics."""

Purpose: Get event bus statistics.

Parameters: – self: Instance reference

EventDrivenPattern (Line 123)

Description: Event-Driven Pattern – Asynchronous event-based communication.

Agents publish events to a central bus and subscribe to events they’re interested in, enabling loose coupling and reactivity.

Methods:

__init__ (Line 131)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

register_agent (Line 134)

def register_agent(self, agent):
    """Register agent with event bus."""

Purpose: Register agent with event bus.

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

run (Line 138)

def run(self, scenario):
    """Execute event-driven scenario."""

Purpose: Execute event-driven 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 = Event(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

    """
Pattern 18: Tuple Space

Purpose: Enhanced implementation with full LangChain/LangGraph integration
"""

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


@dataclass
class TupleSpaceConfig:
    """Configuration for Tuple Space."""
    model_name: str = "gpt-4"
    temperature: float = 0.7
    max_iterations: int = 10
    enable_logging: bool = True


@dataclass
class TupleSpaceState:
    """State information for Tuple Space."""
    current_step: int = 0
    data: Dict[str, Any] = field(default_factory=dict)
    results: List[Any] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=datetime.now)


class TupleSpaceAgent:
    """Agent implementing Tuple Space."""
    
    def __init__(self, name: str, config: TupleSpaceConfig = None):
        self.name = name
        self.config = config or TupleSpaceConfig()
        self.llm = ChatOpenAI(
            model=self.config.model_name,
            temperature=self.config.temperature
        )
        self.state = TupleSpaceState()
        self.execution_history: List[Dict[str, Any]] = []
    
    def process(self, input_data: Any) -> Dict[str, Any]:
        """Process input according to Tuple Space."""
        self.state.current_step += 1
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, implementing Tuple Space. "
                      f"Provide clear, structured responses."),
            ("human", "Process this input: {input_data}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"input_data": str(input_data)})
        
        result = {
            "agent": self.name,
            "step": self.state.current_step,
            "input": input_data,
            "output": str(response.content),
            "timestamp": datetime.now()
        }
        
        self.execution_history.append(result)
        self.state.results.append(result)
        
        return result
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get agent statistics."""
        return {
            "name": self.name,
            "total_executions": len(self.execution_history),
            "current_step": self.state.current_step,
            "results_count": len(self.state.results)
        }


class TupleSpacePattern:
    """
    Tuple Space - Enhanced implementation
    
    This pattern provides comprehensive functionality for tuple space
    with full LangChain/LangGraph integration.
    
    Features:
    - Multi-agent coordination
    - State management
    - Execution tracking
    - LangGraph workflow integration
    """
    
    def __init__(self, config: TupleSpaceConfig = None):
        self.config = config or TupleSpaceConfig()
        self.agents: Dict[str, TupleSpaceAgent] = {}
        self.workflow = self._build_workflow()
        self.execution_log: List[Dict[str, Any]] = []
    
    def _build_workflow(self) -> StateGraph:
        """Build LangGraph workflow for pattern execution."""
        
        class WorkflowState(TypedDict):
            input: str
            processed: str
            output: str
            metadata: Dict[str, Any]
        
        def process_input(state: WorkflowState) -> WorkflowState:
            """Process input node."""
            return {
                **state,
                "processed": f"Processed: {state['input']}",
                "metadata": {"step": "process_input"}
            }
        
        def generate_output(state: WorkflowState) -> WorkflowState:
            """Generate output node."""
            llm = ChatOpenAI(model=self.config.model_name, temperature=self.config.temperature)
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are implementing Tuple Space. Generate comprehensive output."),
                ("human", "{input}")
            ])
            
            chain = prompt | llm
            response = chain.invoke({"input": state["processed"]})
            
            return {
                **state,
                "output": str(response.content),
                "metadata": {**state.get("metadata", {}), "step": "generate_output"}
            }
        
        # Build workflow graph
        workflow = StateGraph(WorkflowState)
        workflow.add_node("process", process_input)
        workflow.add_node("generate", generate_output)
        workflow.set_entry_point("process")
        workflow.add_edge("process", "generate")
        workflow.add_edge("generate", END)
        
        return workflow.compile()
    
    def register_agent(self, agent: TupleSpaceAgent):
        """Register an agent with the pattern."""
        self.agents[agent.name] = agent
        if self.config.enable_logging:
            print(f"Registered agent: {agent.name}")
    
    def execute(self, input_data: Any, agent_name: str = None) -> Dict[str, Any]:
        """Execute the pattern with given input."""
        
        # Use workflow if no specific agent requested
        if agent_name is None:
            result = self.workflow.invoke({
                "input": str(input_data),
                "processed": "",
                "output": "",
                "metadata": {}
            })
            
            execution_record = {
                "pattern": "Tuple Space",
                "input": input_data,
                "output": result.get("output", ""),
                "metadata": result.get("metadata", {}),
                "timestamp": datetime.now()
            }
            
            self.execution_log.append(execution_record)
            return execution_record
        
        # Execute with specific agent
        agent = self.agents.get(agent_name)
        if not agent:
            raise ValueError(f"Agent {agent_name} not found")
        
        result = agent.process(input_data)
        self.execution_log.append(result)
        return result
    
    def run(self, input_data: Any) -> Dict[str, Any]:
        """Main execution method."""
        return self.execute(input_data)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get pattern statistics."""
        return {
            "pattern": "Tuple Space",
            "total_executions": len(self.execution_log),
            "registered_agents": len(self.agents),
            "agents": [agent.get_statistics() for agent in self.agents.values()]
        }


# Example usage
if __name__ == "__main__":
    print("=== Tuple Space Demo ===\n")
    
    # Create configuration
    config = TupleSpaceConfig(
        model_name="gpt-4",
        temperature=0.7,
        enable_logging=True
    )
    
    # Create pattern
    pattern = TupleSpacePattern(config)
    
    # Create and register agents
    agent1 = TupleSpaceAgent("Agent1", config)
    agent2 = TupleSpaceAgent("Agent2", config)
    pattern.register_agent(agent1)
    pattern.register_agent(agent2)
    
    print("\nExecuting pattern workflow...")
    
    # Execute with workflow
    result1 = pattern.run("Demonstrate tuple space functionality")
    print(f"\nWorkflow Result:")
    print(f"  Output: {result1['output'][:150]}...")
    
    # Execute with specific agents
    result2 = pattern.execute("Specific task for Agent1", agent_name="Agent1")
    print(f"\nAgent1 Result:")
    print(f"  Output: {result2['output'][:150]}...")
    
    result3 = pattern.execute("Specific task for Agent2", agent_name="Agent2")
    print(f"\nAgent2 Result:")
    print(f"  Output: {result3['output'][:150]}...")
    
    # Show statistics
    print("\n=== Statistics ===")
    stats = pattern.get_statistics()
    print(f"Pattern: {stats['pattern']}")
    print(f"Total Executions: {stats['total_executions']}")
    print(f"Registered Agents: {stats['registered_agents']}")
    print(f"\nAgent Statistics:")
    for agent_stats in stats['agents']:
        print(f"  {agent_stats['name']}: {agent_stats['total_executions']} executions")

    

Agentic Architecture Pattern: Tuple Space

Pattern 18: 18 Tuple Space – Line-by-Line Explanation

Overview

Pattern 18: Tuple Space

Purpose: Enhanced implementation with full LangChain/LangGraph integration

Code Structure Analysis

File Information

• Pattern Number: 18
• Pattern Name: 18 Tuple Space
• File Path: patterns\communication\18_tuple_space.py
• Total Lines: 247

Detailed Code Explanation

Module Docstring

"""
Pattern 18: Tuple Space

Purpose: Enhanced implementation with full LangChain/LangGraph integration
"""

Explanation: This module docstring describes the purpose and key features of the 18 Tuple Space 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

Classes

TupleSpaceConfig (Line 18)

Description: Configuration for Tuple Space.

TupleSpaceState (Line 27)

Description: State information for Tuple Space.

TupleSpaceAgent (Line 36)

Description: Agent implementing Tuple Space.

Methods:

__init__ (Line 39)

def __init__(self, name, config):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – config: Configuration settings

process (Line 49)

def process(self, input_data):
    """Process input according to Tuple Space."""

Purpose: Process input according to Tuple Space.

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

get_statistics (Line 75)

def get_statistics(self):
    """Get agent statistics."""

Purpose: Get agent statistics.

Parameters: – self: Instance reference

TupleSpacePattern (Line 85)

Description: Tuple Space – Enhanced implementation

This pattern provides comprehensive functionality for tuple space with full LangChain/LangGraph integration.

Features: – Multi-agent coordination – State management – Execution tracking – LangGraph workflow integration

Methods:

__init__ (Line 99)

def __init__(self, config):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – config: Configuration settings

_build_workflow (Line 105)

def _build_workflow(self):
    """Build LangGraph workflow for pattern execution."""

Purpose: Build LangGraph workflow for pattern execution.

Parameters: – self: Instance reference

register_agent (Line 150)

def register_agent(self, agent):
    """Register an agent with the pattern."""

Purpose: Register an agent with the pattern.

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

execute (Line 156)

def execute(self, input_data, agent_name):
    """Execute the pattern with given input."""

Purpose: Execute the pattern with given input.

Parameters: – self: Instance reference – input_data: Parameter for input_data – agent_name: Parameter for agent_name

run (Line 188)

def run(self, input_data):
    """Main execution method."""

Purpose: Main execution method.

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

get_statistics (Line 192)

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

Purpose: Get pattern statistics.

Parameters: – self: Instance reference

WorkflowState (Line 108)

Description: No class docstring.

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

    """
Pattern 17: Shared Memory

Purpose: Enhanced implementation with full LangChain/LangGraph integration
"""

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


@dataclass
class SharedMemoryConfig:
    """Configuration for Shared Memory."""
    model_name: str = "gpt-4"
    temperature: float = 0.7
    max_iterations: int = 10
    enable_logging: bool = True


@dataclass
class SharedMemoryState:
    """State information for Shared Memory."""
    current_step: int = 0
    data: Dict[str, Any] = field(default_factory=dict)
    results: List[Any] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=datetime.now)


class SharedMemoryAgent:
    """Agent implementing Shared Memory."""
    
    def __init__(self, name: str, config: SharedMemoryConfig = None):
        self.name = name
        self.config = config or SharedMemoryConfig()
        self.llm = ChatOpenAI(
            model=self.config.model_name,
            temperature=self.config.temperature
        )
        self.state = SharedMemoryState()
        self.execution_history: List[Dict[str, Any]] = []
    
    def process(self, input_data: Any) -> Dict[str, Any]:
        """Process input according to Shared Memory."""
        self.state.current_step += 1
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, implementing Shared Memory. "
                      f"Provide clear, structured responses."),
            ("human", "Process this input: {input_data}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"input_data": str(input_data)})
        
        result = {
            "agent": self.name,
            "step": self.state.current_step,
            "input": input_data,
            "output": str(response.content),
            "timestamp": datetime.now()
        }
        
        self.execution_history.append(result)
        self.state.results.append(result)
        
        return result
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get agent statistics."""
        return {
            "name": self.name,
            "total_executions": len(self.execution_history),
            "current_step": self.state.current_step,
            "results_count": len(self.state.results)
        }


class SharedMemoryPattern:
    """
    Shared Memory - Enhanced implementation
    
    This pattern provides comprehensive functionality for shared memory
    with full LangChain/LangGraph integration.
    
    Features:
    - Multi-agent coordination
    - State management
    - Execution tracking
    - LangGraph workflow integration
    """
    
    def __init__(self, config: SharedMemoryConfig = None):
        self.config = config or SharedMemoryConfig()
        self.agents: Dict[str, SharedMemoryAgent] = {}
        self.workflow = self._build_workflow()
        self.execution_log: List[Dict[str, Any]] = []
    
    def _build_workflow(self) -> StateGraph:
        """Build LangGraph workflow for pattern execution."""
        
        class WorkflowState(TypedDict):
            input: str
            processed: str
            output: str
            metadata: Dict[str, Any]
        
        def process_input(state: WorkflowState) -> WorkflowState:
            """Process input node."""
            return {
                **state,
                "processed": f"Processed: {state['input']}",
                "metadata": {"step": "process_input"}
            }
        
        def generate_output(state: WorkflowState) -> WorkflowState:
            """Generate output node."""
            llm = ChatOpenAI(model=self.config.model_name, temperature=self.config.temperature)
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are implementing Shared Memory. Generate comprehensive output."),
                ("human", "{input}")
            ])
            
            chain = prompt | llm
            response = chain.invoke({"input": state["processed"]})
            
            return {
                **state,
                "output": str(response.content),
                "metadata": {**state.get("metadata", {}), "step": "generate_output"}
            }
        
        # Build workflow graph
        workflow = StateGraph(WorkflowState)
        workflow.add_node("process", process_input)
        workflow.add_node("generate", generate_output)
        workflow.set_entry_point("process")
        workflow.add_edge("process", "generate")
        workflow.add_edge("generate", END)
        
        return workflow.compile()
    
    def register_agent(self, agent: SharedMemoryAgent):
        """Register an agent with the pattern."""
        self.agents[agent.name] = agent
        if self.config.enable_logging:
            print(f"Registered agent: {agent.name}")
    
    def execute(self, input_data: Any, agent_name: str = None) -> Dict[str, Any]:
        """Execute the pattern with given input."""
        
        # Use workflow if no specific agent requested
        if agent_name is None:
            result = self.workflow.invoke({
                "input": str(input_data),
                "processed": "",
                "output": "",
                "metadata": {}
            })
            
            execution_record = {
                "pattern": "Shared Memory",
                "input": input_data,
                "output": result.get("output", ""),
                "metadata": result.get("metadata", {}),
                "timestamp": datetime.now()
            }
            
            self.execution_log.append(execution_record)
            return execution_record
        
        # Execute with specific agent
        agent = self.agents.get(agent_name)
        if not agent:
            raise ValueError(f"Agent {agent_name} not found")
        
        result = agent.process(input_data)
        self.execution_log.append(result)
        return result
    
    def run(self, input_data: Any) -> Dict[str, Any]:
        """Main execution method."""
        return self.execute(input_data)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get pattern statistics."""
        return {
            "pattern": "Shared Memory",
            "total_executions": len(self.execution_log),
            "registered_agents": len(self.agents),
            "agents": [agent.get_statistics() for agent in self.agents.values()]
        }


# Example usage
if __name__ == "__main__":
    print("=== Shared Memory Demo ===\n")
    
    # Create configuration
    config = SharedMemoryConfig(
        model_name="gpt-4",
        temperature=0.7,
        enable_logging=True
    )
    
    # Create pattern
    pattern = SharedMemoryPattern(config)
    
    # Create and register agents
    agent1 = SharedMemoryAgent("Agent1", config)
    agent2 = SharedMemoryAgent("Agent2", config)
    pattern.register_agent(agent1)
    pattern.register_agent(agent2)
    
    print("\nExecuting pattern workflow...")
    
    # Execute with workflow
    result1 = pattern.run("Demonstrate shared memory functionality")
    print(f"\nWorkflow Result:")
    print(f"  Output: {result1['output'][:150]}...")
    
    # Execute with specific agents
    result2 = pattern.execute("Specific task for Agent1", agent_name="Agent1")
    print(f"\nAgent1 Result:")
    print(f"  Output: {result2['output'][:150]}...")
    
    result3 = pattern.execute("Specific task for Agent2", agent_name="Agent2")
    print(f"\nAgent2 Result:")
    print(f"  Output: {result3['output'][:150]}...")
    
    # Show statistics
    print("\n=== Statistics ===")
    stats = pattern.get_statistics()
    print(f"Pattern: {stats['pattern']}")
    print(f"Total Executions: {stats['total_executions']}")
    print(f"Registered Agents: {stats['registered_agents']}")
    print(f"\nAgent Statistics:")
    for agent_stats in stats['agents']:
        print(f"  {agent_stats['name']}: {agent_stats['total_executions']} executions")

    

Agentic Architecture Pattern: Shared Memory

Pattern 17: 17 Shared Memory – Line-by-Line Explanation

Overview

Pattern 17: Shared Memory

Purpose: Enhanced implementation with full LangChain/LangGraph integration

Code Structure Analysis

File Information

• Pattern Number: 17
• Pattern Name: 17 Shared Memory
• File Path: patterns\communication\17_shared_memory.py
• Total Lines: 247

Detailed Code Explanation

Module Docstring

"""
Pattern 17: Shared Memory

Purpose: Enhanced implementation with full LangChain/LangGraph integration
"""

Explanation: This module docstring describes the purpose and key features of the 17 Shared Memory 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

Classes

SharedMemoryConfig (Line 18)

Description: Configuration for Shared Memory.

SharedMemoryState (Line 27)

Description: State information for Shared Memory.

SharedMemoryAgent (Line 36)

Description: Agent implementing Shared Memory.

Methods:

__init__ (Line 39)

def __init__(self, name, config):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – config: Configuration settings

process (Line 49)

def process(self, input_data):
    """Process input according to Shared Memory."""

Purpose: Process input according to Shared Memory.

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

get_statistics (Line 75)

def get_statistics(self):
    """Get agent statistics."""

Purpose: Get agent statistics.

Parameters: – self: Instance reference

SharedMemoryPattern (Line 85)

Description: Shared Memory – Enhanced implementation

This pattern provides comprehensive functionality for shared memory with full LangChain/LangGraph integration.

Features: – Multi-agent coordination – State management – Execution tracking – LangGraph workflow integration

Methods:

__init__ (Line 99)

def __init__(self, config):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – config: Configuration settings

_build_workflow (Line 105)

def _build_workflow(self):
    """Build LangGraph workflow for pattern execution."""

Purpose: Build LangGraph workflow for pattern execution.

Parameters: – self: Instance reference

register_agent (Line 150)

def register_agent(self, agent):
    """Register an agent with the pattern."""

Purpose: Register an agent with the pattern.

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

execute (Line 156)

def execute(self, input_data, agent_name):
    """Execute the pattern with given input."""

Purpose: Execute the pattern with given input.

Parameters: – self: Instance reference – input_data: Parameter for input_data – agent_name: Parameter for agent_name

run (Line 188)

def run(self, input_data):
    """Main execution method."""

Purpose: Main execution method.

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

get_statistics (Line 192)

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

Purpose: Get pattern statistics.

Parameters: – self: Instance reference

WorkflowState (Line 108)

Description: No class docstring.

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

    """
Pattern 16: Message Passing Pattern

Direct point-to-point communication between agents through message channels.
Agents send and receive messages asynchronously via mailboxes/queues.
"""

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from collections import deque
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI


@dataclass
class Message:
    """Message passed between agents."""
    sender: str
    receiver: str
    content: Any
    message_type: str = "default"
    timestamp: datetime = field(default_factory=datetime.now)
    reply_to: Optional[str] = None


class MessageQueue:
    """Message queue (mailbox) for an agent."""
    
    def __init__(self, max_size: int = 100):
        self.queue: deque = deque(maxlen=max_size)
        self.processed: List[Message] = []
    
    def enqueue(self, message: Message):
        """Add message to queue."""
        self.queue.append(message)
    
    def dequeue(self) -> Optional[Message]:
        """Get next message from queue."""
        if self.queue:
            msg = self.queue.popleft()
            self.processed.append(msg)
            return msg
        return None
    
    def peek(self) -> Optional[Message]:
        """Peek at next message without removing."""
        return self.queue[0] if self.queue else None
    
    def size(self) -> int:
        """Get queue size."""
        return len(self.queue)


class MessagePassingAgent:
    """Agent with message passing 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.mailbox = MessageQueue()
        self.system: Optional['MessagePassingPattern'] = None
    
    def send(self, receiver: str, content: Any, message_type: str = "default"):
        """Send message to another agent."""
        if not self.system:
            raise ValueError("Agent not connected to message passing system")
        
        message = Message(
            sender=self.name,
            receiver=receiver,
            content=content,
            message_type=message_type
        )
        self.system.route_message(message)
    
    def receive(self) -> Optional[Message]:
        """Receive next message from mailbox."""
        return self.mailbox.dequeue()
    
    def process_message(self, message: Message) -> Optional[str]:
        """Process received message using LLM."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, a {self.role}. "
                      f"Process this message and respond if needed."),
            ("human", f"From: {message.sender}\n"
                     f"Type: {message.type}\n"
                     f"Content: {message.content}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        return str(response.content)
    
    def check_messages(self) -> int:
        """Check number of pending messages."""
        return self.mailbox.size()


class MessagePassingPattern:
    """
    Message Passing Pattern - Asynchronous agent communication.
    
    Agents communicate by sending messages that are queued in mailboxes.
    Each agent has its own message queue for asynchronous processing.
    """
    
    def __init__(self):
        self.agents: Dict[str, MessagePassingAgent] = {}
        self.message_log: List[Message] = []
    
    def register_agent(self, agent: MessagePassingAgent):
        """Register agent with message passing system."""
        self.agents[agent.name] = agent
        agent.system = self
    
    def route_message(self, message: Message):
        """Route message to recipient's mailbox."""
        self.message_log.append(message)
        
        if message.receiver in self.agents:
            recipient = self.agents[message.receiver]
            recipient.mailbox.enqueue(message)
        else:
            print(f"Warning: Recipient {message.receiver} not found")
    
    def process_all_messages(self):
        """Process all pending messages for all agents."""
        for agent in self.agents.values():
            while agent.mailbox.size() > 0:
                message = agent.receive()
                if message:
                    response = agent.process_message(message)
                    
                    # Send reply if needed
                    if response and "reply" in response.lower():
                        agent.send(message.sender, response, "reply")
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get messaging statistics."""
        return {
            agent_name: {
                "pending": agent.mailbox.size(),
                "processed": len(agent.mailbox.processed)
            }
            for agent_name, agent in self.agents.items()
        }
    
    def run(self, scenario: str) -> Dict[str, Any]:
        """Execute message passing scenario."""
        return {
            "pattern": "MessagePassing",
            "total_agents": len(self.agents),
            "total_messages": len(self.message_log),
            "agents": list(self.agents.keys()),
            "scenario": scenario
        }


# Example usage
if __name__ == "__main__":
    print("=== Message Passing Pattern Demo ===\n")
    
    # Create system
    system = MessagePassingPattern()
    
    # Create agents
    coordinator = MessagePassingAgent("Coordinator", "project coordinator")
    dev1 = MessagePassingAgent("Dev1", "backend developer")
    dev2 = MessagePassingAgent("Dev2", "frontend developer")
    tester = MessagePassingAgent("QA", "tester")
    
    # Register agents
    for agent in [coordinator, dev1, dev2, tester]:
        system.register_agent(agent)
    
    print("1. Agents sending messages...")
    coordinator.send("Dev1", "Implement user authentication API")
    coordinator.send("Dev2", "Create login UI components")
    coordinator.send("QA", "Prepare test cases for auth flow")
    
    dev1.send("Dev2", "API endpoints ready at /api/auth")
    dev2.send("QA", "UI components completed, ready for testing")
    
    print(f"\n2. Message queue status:")
    stats = system.get_statistics()
    for agent_name, stat in stats.items():
        print(f"   {agent_name}: {stat['pending']} pending, {stat['processed']} processed")
    
    print("\n3. Processing all messages...")
    system.process_all_messages()
    
    print(f"\n4. After processing:")
    stats = system.get_statistics()
    for agent_name, stat in stats.items():
        print(f"   {agent_name}: {stat['pending']} pending, {stat['processed']} processed")
    
    # Summary
    result = system.run("Team communication")
    print(f"\n5. Summary:")
    print(f"   Agents: {result['agents']}")
    print(f"   Total messages: {result['total_messages']}")
    
    print("\n✓ Message passing pattern enabled asynchronous agent communication!")

    

Agentic Architecture Pattern: Message Passing

Pattern 16: 16 Message Passing – Line-by-Line Explanation

Overview

Pattern 16: Message Passing Pattern

Direct point-to-point communication between agents through message channels. Agents send and receive messages asynchronously via mailboxes/queues.

Code Structure Analysis

File Information

• Pattern Number: 16
• Pattern Name: 16 Message Passing
• File Path: patterns\communication\16_message_passing.py
• Total Lines: 206

Detailed Code Explanation

Module Docstring

"""
Pattern 16: Message Passing Pattern

Direct point-to-point communication between agents through message channels.
Agents send and receive messages asynchronously via mailboxes/queues.
"""

Explanation: This module docstring describes the purpose and key features of the 16 Message Passing pattern.

Imports Section

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from collections import deque
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: 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

Classes

Message (Line 17)

Description: Message passed between agents.

MessageQueue (Line 27)

Description: Message queue (mailbox) for an agent.

Methods:

__init__ (Line 30)

def __init__(self, max_size):
    """No method docstring."""

Purpose: No method docstring.

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

enqueue (Line 34)

def enqueue(self, message):
    """Add message to queue."""

Purpose: Add message to queue.

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

dequeue (Line 38)

def dequeue(self):
    """Get next message from queue."""

Purpose: Get next message from queue.

Parameters: – self: Instance reference

peek (Line 46)

def peek(self):
    """Peek at next message without removing."""

Purpose: Peek at next message without removing.

Parameters: – self: Instance reference

size (Line 50)

def size(self):
    """Get queue size."""

Purpose: Get queue size.

Parameters: – self: Instance reference

MessagePassingAgent (Line 55)

Description: Agent with message passing capabilities.

Methods:

__init__ (Line 58)

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”)

send (Line 65)

def send(self, receiver, content, message_type):
    """Send message to another agent."""

Purpose: Send message to another agent.

Parameters: – self: Instance reference – receiver: Parameter for receiver – content: Parameter for content – message_type: Parameter for message_type

receive (Line 78)

def receive(self):
    """Receive next message from mailbox."""

Purpose: Receive next message from mailbox.

Parameters: – self: Instance reference

process_message (Line 82)

def process_message(self, message):
    """Process received message using LLM."""

Purpose: Process received message using LLM.

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

check_messages (Line 96)

def check_messages(self):
    """Check number of pending messages."""

Purpose: Check number of pending messages.

Parameters: – self: Instance reference

MessagePassingPattern (Line 101)

Description: Message Passing Pattern – Asynchronous agent communication.

Agents communicate by sending messages that are queued in mailboxes. Each agent has its own message queue for asynchronous processing.

Methods:

__init__ (Line 109)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

register_agent (Line 113)

def register_agent(self, agent):
    """Register agent with message passing system."""

Purpose: Register agent with message passing system.

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

route_message (Line 118)

def route_message(self, message):
    """Route message to recipient's mailbox."""

Purpose: Route message to recipient’s mailbox.

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

process_all_messages (Line 128)

def process_all_messages(self):
    """Process all pending messages for all agents."""

Purpose: Process all pending messages for all agents.

Parameters: – self: Instance reference

get_statistics (Line 140)

def get_statistics(self):
    """Get messaging statistics."""

Purpose: Get messaging statistics.

Parameters: – self: Instance reference

run (Line 150)

def run(self, scenario):
    """Execute message passing scenario."""

Purpose: Execute message passing 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 = Message(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

    """
Pattern 15: Publisher Subscriber

Purpose: Enhanced implementation with full LangChain/LangGraph integration
"""

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


@dataclass
class PublisherSubscriberConfig:
    """Configuration for Publisher Subscriber."""
    model_name: str = "gpt-4"
    temperature: float = 0.7
    max_iterations: int = 10
    enable_logging: bool = True


@dataclass
class PublisherSubscriberState:
    """State information for Publisher Subscriber."""
    current_step: int = 0
    data: Dict[str, Any] = field(default_factory=dict)
    results: List[Any] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=datetime.now)


class PublisherSubscriberAgent:
    """Agent implementing Publisher Subscriber."""
    
    def __init__(self, name: str, config: PublisherSubscriberConfig = None):
        self.name = name
        self.config = config or PublisherSubscriberConfig()
        self.llm = ChatOpenAI(
            model=self.config.model_name,
            temperature=self.config.temperature
        )
        self.state = PublisherSubscriberState()
        self.execution_history: List[Dict[str, Any]] = []
    
    def process(self, input_data: Any) -> Dict[str, Any]:
        """Process input according to Publisher Subscriber."""
        self.state.current_step += 1
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, implementing Publisher Subscriber. "
                      f"Provide clear, structured responses."),
            ("human", "Process this input: {input_data}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"input_data": str(input_data)})
        
        result = {
            "agent": self.name,
            "step": self.state.current_step,
            "input": input_data,
            "output": str(response.content),
            "timestamp": datetime.now()
        }
        
        self.execution_history.append(result)
        self.state.results.append(result)
        
        return result
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get agent statistics."""
        return {
            "name": self.name,
            "total_executions": len(self.execution_history),
            "current_step": self.state.current_step,
            "results_count": len(self.state.results)
        }


class PublisherSubscriberPattern:
    """
    Publisher Subscriber - Enhanced implementation
    
    This pattern provides comprehensive functionality for publisher subscriber
    with full LangChain/LangGraph integration.
    
    Features:
    - Multi-agent coordination
    - State management
    - Execution tracking
    - LangGraph workflow integration
    """
    
    def __init__(self, config: PublisherSubscriberConfig = None):
        self.config = config or PublisherSubscriberConfig()
        self.agents: Dict[str, PublisherSubscriberAgent] = {}
        self.workflow = self._build_workflow()
        self.execution_log: List[Dict[str, Any]] = []
    
    def _build_workflow(self) -> StateGraph:
        """Build LangGraph workflow for pattern execution."""
        
        class WorkflowState(TypedDict):
            input: str
            processed: str
            output: str
            metadata: Dict[str, Any]
        
        def process_input(state: WorkflowState) -> WorkflowState:
            """Process input node."""
            return {
                **state,
                "processed": f"Processed: {state['input']}",
                "metadata": {"step": "process_input"}
            }
        
        def generate_output(state: WorkflowState) -> WorkflowState:
            """Generate output node."""
            llm = ChatOpenAI(model=self.config.model_name, temperature=self.config.temperature)
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are implementing Publisher Subscriber. Generate comprehensive output."),
                ("human", "{input}")
            ])
            
            chain = prompt | llm
            response = chain.invoke({"input": state["processed"]})
            
            return {
                **state,
                "output": str(response.content),
                "metadata": {**state.get("metadata", {}), "step": "generate_output"}
            }
        
        # Build workflow graph
        workflow = StateGraph(WorkflowState)
        workflow.add_node("process", process_input)
        workflow.add_node("generate", generate_output)
        workflow.set_entry_point("process")
        workflow.add_edge("process", "generate")
        workflow.add_edge("generate", END)
        
        return workflow.compile()
    
    def register_agent(self, agent: PublisherSubscriberAgent):
        """Register an agent with the pattern."""
        self.agents[agent.name] = agent
        if self.config.enable_logging:
            print(f"Registered agent: {agent.name}")
    
    def execute(self, input_data: Any, agent_name: str = None) -> Dict[str, Any]:
        """Execute the pattern with given input."""
        
        # Use workflow if no specific agent requested
        if agent_name is None:
            result = self.workflow.invoke({
                "input": str(input_data),
                "processed": "",
                "output": "",
                "metadata": {}
            })
            
            execution_record = {
                "pattern": "Publisher Subscriber",
                "input": input_data,
                "output": result.get("output", ""),
                "metadata": result.get("metadata", {}),
                "timestamp": datetime.now()
            }
            
            self.execution_log.append(execution_record)
            return execution_record
        
        # Execute with specific agent
        agent = self.agents.get(agent_name)
        if not agent:
            raise ValueError(f"Agent {agent_name} not found")
        
        result = agent.process(input_data)
        self.execution_log.append(result)
        return result
    
    def run(self, input_data: Any) -> Dict[str, Any]:
        """Main execution method."""
        return self.execute(input_data)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get pattern statistics."""
        return {
            "pattern": "Publisher Subscriber",
            "total_executions": len(self.execution_log),
            "registered_agents": len(self.agents),
            "agents": [agent.get_statistics() for agent in self.agents.values()]
        }


# Example usage
if __name__ == "__main__":
    print("=== Publisher Subscriber Demo ===\n")
    
    # Create configuration
    config = PublisherSubscriberConfig(
        model_name="gpt-4",
        temperature=0.7,
        enable_logging=True
    )
    
    # Create pattern
    pattern = PublisherSubscriberPattern(config)
    
    # Create and register agents
    agent1 = PublisherSubscriberAgent("Agent1", config)
    agent2 = PublisherSubscriberAgent("Agent2", config)
    pattern.register_agent(agent1)
    pattern.register_agent(agent2)
    
    print("\nExecuting pattern workflow...")
    
    # Execute with workflow
    result1 = pattern.run("Demonstrate publisher subscriber functionality")
    print(f"\nWorkflow Result:")
    print(f"  Output: {result1['output'][:150]}...")
    
    # Execute with specific agents
    result2 = pattern.execute("Specific task for Agent1", agent_name="Agent1")
    print(f"\nAgent1 Result:")
    print(f"  Output: {result2['output'][:150]}...")
    
    result3 = pattern.execute("Specific task for Agent2", agent_name="Agent2")
    print(f"\nAgent2 Result:")
    print(f"  Output: {result3['output'][:150]}...")
    
    # Show statistics
    print("\n=== Statistics ===")
    stats = pattern.get_statistics()
    print(f"Pattern: {stats['pattern']}")
    print(f"Total Executions: {stats['total_executions']}")
    print(f"Registered Agents: {stats['registered_agents']}")
    print(f"\nAgent Statistics:")
    for agent_stats in stats['agents']:
        print(f"  {agent_stats['name']}: {agent_stats['total_executions']} executions")

    

Agentic Architecture Pattern: Publisher Subscriber

Pattern 15: 15 Publisher Subscriber – Line-by-Line Explanation

Overview

Pattern 15: Publisher Subscriber

Purpose: Enhanced implementation with full LangChain/LangGraph integration

Code Structure Analysis

File Information

• Pattern Number: 15
• Pattern Name: 15 Publisher Subscriber
• File Path: patterns\communication\15_publisher_subscriber.py
• Total Lines: 247

Detailed Code Explanation

Module Docstring

"""
Pattern 15: Publisher Subscriber

Purpose: Enhanced implementation with full LangChain/LangGraph integration
"""

Explanation: This module docstring describes the purpose and key features of the 15 Publisher Subscriber 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

Classes

PublisherSubscriberConfig (Line 18)

Description: Configuration for Publisher Subscriber.

PublisherSubscriberState (Line 27)

Description: State information for Publisher Subscriber.

PublisherSubscriberAgent (Line 36)

Description: Agent implementing Publisher Subscriber.

Methods:

__init__ (Line 39)

def __init__(self, name, config):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – config: Configuration settings

process (Line 49)

def process(self, input_data):
    """Process input according to Publisher Subscriber."""

Purpose: Process input according to Publisher Subscriber.

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

get_statistics (Line 75)

def get_statistics(self):
    """Get agent statistics."""

Purpose: Get agent statistics.

Parameters: – self: Instance reference

PublisherSubscriberPattern (Line 85)

Description: Publisher Subscriber – Enhanced implementation

This pattern provides comprehensive functionality for publisher subscriber with full LangChain/LangGraph integration.

Features: – Multi-agent coordination – State management – Execution tracking – LangGraph workflow integration

Methods:

__init__ (Line 99)

def __init__(self, config):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – config: Configuration settings

_build_workflow (Line 105)

def _build_workflow(self):
    """Build LangGraph workflow for pattern execution."""

Purpose: Build LangGraph workflow for pattern execution.

Parameters: – self: Instance reference

register_agent (Line 150)

def register_agent(self, agent):
    """Register an agent with the pattern."""

Purpose: Register an agent with the pattern.

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

execute (Line 156)

def execute(self, input_data, agent_name):
    """Execute the pattern with given input."""

Purpose: Execute the pattern with given input.

Parameters: – self: Instance reference – input_data: Parameter for input_data – agent_name: Parameter for agent_name

run (Line 188)

def run(self, input_data):
    """Main execution method."""

Purpose: Main execution method.

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

get_statistics (Line 192)

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

Purpose: Get pattern statistics.

Parameters: – self: Instance reference

WorkflowState (Line 108)

Description: No class docstring.

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

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

    """
Pattern 13: Mediator Pattern

Centralizes complex communications and control between related agents.
Mediator promotes loose coupling by keeping agents from referring to each other explicitly.
"""

from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
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 AgentMessage:
    """Message exchanged between agents."""
    sender: str
    recipient: Optional[str]  # None = broadcast
    content: str
    timestamp: datetime = field(default_factory=datetime.now)
    metadata: Dict[str, Any] = field(default_factory=dict)


class MediatedAgent:
    """Agent that communicates through a mediator."""
    
    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.mediator: Optional['MediatorPattern'] = None
    
    def set_mediator(self, mediator: 'MediatorPattern'):
        """Register with mediator."""
        self.mediator = mediator
    
    def send(self, content: str, recipient: Optional[str] = None):
        """Send message via mediator."""
        if not self.mediator:
            raise ValueError(f"{self.name}: Mediator not set")
        
        message = AgentMessage(
            sender=self.name,
            recipient=recipient,
            content=content
        )
        self.mediator.relay_message(message)
    
    def receive(self, message: AgentMessage) -> str:
        """Process received message."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, a {self.role}. "
                      f"Process this message and respond appropriately."),
            ("human", f"From {message.sender}: {message.content}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        return str(response.content)


class MediatorState(TypedDict):
    """State for mediator graph."""
    messages: List[AgentMessage]
    active_agents: List[str]
    current_message: Optional[AgentMessage]


class MediatorPattern:
    """
    Mediator Pattern - Centralized agent communication coordinator.
    
    Uses LangGraph to manage message routing and agent interactions.
    All agent-to-agent communication flows through the mediator.
    """
    
    def __init__(self):
        self.agents: Dict[str, MediatedAgent] = {}
        self.message_log: List[AgentMessage] = []
        self.graph = self._build_graph()
    
    def register_agent(self, agent: MediatedAgent):
        """Register agent with mediator."""
        self.agents[agent.name] = agent
        agent.set_mediator(self)
    
    def relay_message(self, message: AgentMessage):
        """Relay message to appropriate recipient(s)."""
        self.message_log.append(message)
        
        # Determine recipients
        if message.recipient:
            # Direct message
            if message.recipient in self.agents:
                response = self.agents[message.recipient].receive(message)
                # Log response
                if response:
                    response_msg = AgentMessage(
                        sender=message.recipient,
                        recipient=message.sender,
                        content=response
                    )
                    self.message_log.append(response_msg)
        else:
            # Broadcast to all except sender
            for name, agent in self.agents.items():
                if name != message.sender:
                    response = agent.receive(message)
                    if response:
                        response_msg = AgentMessage(
                            sender=name,
                            recipient=message.sender,
                            content=response
                        )
                        self.message_log.append(response_msg)
    
    def _build_graph(self) -> StateGraph:
        """Build LangGraph for mediator workflow."""
        workflow = StateGraph(MediatorState)
        
        def route_message(state: MediatorState) -> MediatorState:
            """Route messages through mediator."""
            if state["current_message"]:
                self.relay_message(state["current_message"])
            return state
        
        workflow.add_node("route", route_message)
        workflow.set_entry_point("route")
        workflow.add_edge("route", END)
        
        return workflow.compile()
    
    def get_conversation(self, agent1: str, agent2: str) -> List[AgentMessage]:
        """Get conversation between two agents."""
        return [
            msg for msg in self.message_log
            if (msg.sender == agent1 and msg.recipient == agent2) or
               (msg.sender == agent2 and msg.recipient == agent1)
        ]
    
    def run(self, scenario: str) -> Dict[str, Any]:
        """Execute mediator pattern."""
        return {
            "pattern": "Mediator",
            "total_agents": len(self.agents),
            "total_messages": len(self.message_log),
            "agents": list(self.agents.keys()),
            "scenario": scenario
        }


# Example usage
if __name__ == "__main__":
    print("=== Mediator Pattern Demo ===\n")
    
    # Create mediator
    mediator = MediatorPattern()
    
    # Create agents
    project_manager = MediatedAgent("PM", "project manager")
    developer = MediatedAgent("Dev", "developer")
    designer = MediatedAgent("Designer", "UI designer")
    tester = MediatedAgent("QA", "quality assurance")
    
    # Register agents
    for agent in [project_manager, developer, designer, tester]:
        mediator.register_agent(agent)
    
    print("1. PM broadcasts project kickoff...")
    project_manager.send("Starting new e-commerce project. Sprint 1 begins Monday.")
    
    print("\n2. Dev sends direct message to Designer...")
    developer.send("Need mockups for product listing page", recipient="Designer")
    
    print("\n3. QA reports bug to Dev...")
    tester.send("Found critical bug in checkout flow", recipient="Dev")
    
    # Get results
    result = mediator.run("Team collaboration")
    print(f"\n4. Summary:")
    print(f"   Agents: {result['agents']}")
    print(f"   Messages: {result['total_messages']}")
    print(f"   Pattern: {result['pattern']}")
    
    print("\n✓ Mediator pattern coordinated all agent communication!")

    

Agentic Architecture Pattern: Mediator

Pattern 13: 13 Mediator – Line-by-Line Explanation

Overview

Pattern 13: Mediator Pattern

Centralizes complex communications and control between related agents. Mediator promotes loose coupling by keeping agents from referring to each other explicitly.

Code Structure Analysis

File Information

• Pattern Number: 13
• Pattern Name: 13 Mediator
• File Path: patterns\communication\13_mediator.py
• Total Lines: 189

Detailed Code Explanation

Module Docstring

"""
Pattern 13: Mediator Pattern

Centralizes complex communications and control between related agents.
Mediator promotes loose coupling by keeping agents from referring to each other explicitly.
"""

Explanation: This module docstring describes the purpose and key features of the 13 Mediator pattern.

Imports Section

from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
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, 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 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

AgentMessage (Line 18)

Description: Message exchanged between agents.

MediatedAgent (Line 27)

Description: Agent that communicates through a mediator.

Methods:

__init__ (Line 30)

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”)

set_mediator (Line 36)

def set_mediator(self, mediator):
    """Register with mediator."""

Purpose: Register with mediator.

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

send (Line 40)

def send(self, content, recipient):
    """Send message via mediator."""

Purpose: Send message via mediator.

Parameters: – self: Instance reference – content: Parameter for content – recipient: Parameter for recipient

receive (Line 52)

def receive(self, message):
    """Process received message."""

Purpose: Process received message.

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

MediatorState (Line 65)

Description: State for mediator graph.

MediatorPattern (Line 72)

Description: Mediator Pattern – Centralized agent communication coordinator.

Uses LangGraph to manage message routing and agent interactions. All agent-to-agent communication flows through the mediator.

Methods:

__init__ (Line 80)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

register_agent (Line 85)

def register_agent(self, agent):
    """Register agent with mediator."""

Purpose: Register agent with mediator.

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

relay_message (Line 90)

def relay_message(self, message):
    """Relay message to appropriate recipient(s)."""

Purpose: Relay message to appropriate recipient(s).

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

_build_graph (Line 120)

def _build_graph(self):
    """Build LangGraph for mediator workflow."""

Purpose: Build LangGraph for mediator workflow.

Parameters: – self: Instance reference

get_conversation (Line 136)

def get_conversation(self, agent1, agent2):
    """Get conversation between two agents."""

Purpose: Get conversation between two agents.

Parameters: – self: Instance reference – agent1: Parameter for agent1 – agent2: Parameter for agent2

run (Line 144)

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

Purpose: Execute mediator 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 = AgentMessage(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

    """
Pattern 12: Contract Net Protocol
A negotiation protocol where agents bid for tasks.
"""

from typing import Dict, Any, List, Optional
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from enum import Enum


class BidStatus(str, Enum):
    PENDING = "pending"
    ACCEPTED = "accepted"
    REJECTED = "rejected"


class Bid(BaseModel):
    """Represents a bid from an agent."""
    agent_id: str
    task_id: str
    cost: float
    time_estimate: float
    quality_score: float
    proposal: str
    status: BidStatus = BidStatus.PENDING


class Task(BaseModel):
    """Represents a task to be contracted."""
    id: str
    description: str
    requirements: List[str]
    deadline: Optional[float] = None
    budget: Optional[float] = None


class ContractNetPattern:
    """
    Contract Net Protocol - Task allocation through bidding.
    Manager announces tasks, contractors bid, manager awards contracts.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.tasks: Dict[str, Task] = {}
        self.bids: Dict[str, List[Bid]] = {}
        self.contractors: Dict[str, 'ContractorAgent'] = {}
    
    def announce_task(self, task: Task) -> str:
        """Manager announces a task."""
        self.tasks[task.id] = task
        self.bids[task.id] = []
        return task.id
    
    def register_contractor(self, agent_id: str, capabilities: List[str]) -> 'ContractorAgent':
        """Register a contractor agent."""
        contractor = ContractorAgent(agent_id, capabilities, self, self.llm)
        self.contractors[agent_id] = contractor
        return contractor
    
    def receive_bid(self, bid: Bid):
        """Receive a bid from a contractor."""
        if bid.task_id in self.bids:
            self.bids[bid.task_id].append(bid)
    
    def evaluate_bids(self, task_id: str) -> Optional[Bid]:
        """Evaluate bids and select the best one."""
        if task_id not in self.bids or not self.bids[task_id]:
            return None
        
        task = self.tasks[task_id]
        bids = self.bids[task_id]
        
        # Simple evaluation: select lowest cost with acceptable quality
        acceptable_bids = [b for b in bids if b.quality_score >= 7.0]
        if acceptable_bids:
            best_bid = min(acceptable_bids, key=lambda b: b.cost)
        else:
            best_bid = min(bids, key=lambda b: b.cost)
        
        best_bid.status = BidStatus.ACCEPTED
        return best_bid
    
    def award_contract(self, task_id: str) -> Dict[str, Any]:
        """Award contract to the best bidder."""
        winning_bid = self.evaluate_bids(task_id)
        
        if not winning_bid:
            return {"success": False, "message": "No bids received"}
        
        # Mark other bids as rejected
        for bid in self.bids[task_id]:
            if bid != winning_bid:
                bid.status = BidStatus.REJECTED
        
        return {
            "success": True,
            "task_id": task_id,
            "winner": winning_bid.agent_id,
            "cost": winning_bid.cost,
            "time_estimate": winning_bid.time_estimate,
            "all_bids": len(self.bids[task_id])
        }


class ContractorAgent:
    """Agent that bids on tasks."""
    
    def __init__(self, agent_id: str, capabilities: List[str], 
                 contract_net: ContractNetPattern, llm):
        self.agent_id = agent_id
        self.capabilities = capabilities
        self.contract_net = contract_net
        self.llm = llm
    
    def submit_bid(self, task: Task) -> Bid:
        """Submit a bid for a task."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", """You are contractor {agent_id} with capabilities: {capabilities}
            
            Create a bid for this task. Provide cost (dollars), time (hours), quality score (1-10).
            Task: {task_description}
            Requirements: {requirements}"""),
            ("human", "Provide: Cost: X, Time: Y, Quality: Z, Proposal: [text]")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({
            "agent_id": self.agent_id,
            "capabilities": ", ".join(self.capabilities),
            "task_description": task.description,
            "requirements": ", ".join(task.requirements)
        })
        
        # Parse response (simplified)
        import random
        bid = Bid(
            agent_id=self.agent_id,
            task_id=task.id,
            cost=random.uniform(500, 5000),
            time_estimate=random.uniform(10, 50),
            quality_score=random.uniform(6, 10),
            proposal=response.content[:200]
        )
        
        self.contract_net.receive_bid(bid)
        return bid


# Example usage
if __name__ == "__main__":
    cn = ContractNetPattern()
    
    # Register contractors
    c1 = cn.register_contractor("Agent_A", ["web_dev", "backend"])
    c2 = cn.register_contractor("Agent_B", ["web_dev", "frontend"])
    c3 = cn.register_contractor("Agent_C", ["fullstack", "devops"])
    
    # Announce task
    task = Task(
        id="task_001",
        description="Build REST API",
        requirements=["Python", "FastAPI"],
        budget=5000.0
    )
    
    cn.announce_task(task)
    
    # Submit bids
    c1.submit_bid(task)
    c2.submit_bid(task)
    c3.submit_bid(task)
    
    # Award
    result = cn.award_contract(task.id)
    print(f"Winner: {result['winner']}, Cost: ${result['cost']:.2f}")

    

Agentic Architecture Pattern: Contract Net

Pattern 12: 12 Contract Net – Line-by-Line Explanation

Overview

Pattern 12: Contract Net Protocol A negotiation protocol where agents bid for tasks.

Code Structure Analysis

File Information

• Pattern Number: 12
• Pattern Name: 12 Contract Net
• File Path: patterns\communication\12_contract_net.py
• Total Lines: 179

Detailed Code Explanation

Module Docstring

"""
Pattern 12: Contract Net Protocol
A negotiation protocol where agents bid for tasks.
"""

Explanation: This module docstring describes the purpose and key features of the 12 Contract Net pattern.

Imports Section

from typing import Dict, Any, List, Optional
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from enum import Enum
        import random  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional: Type hints for better code documentation – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration – from enum import Enum: Enumeration support for fixed sets of values

Enumerations

BidStatus (Line 13)

class BidStatus(Enum):
    PENDING
    ACCEPTED
    REJECTED

Purpose: No enum docstring.

Values: – PENDING: Initial state, not yet processed – ACCEPTED: Represents accepted – REJECTED: Represents rejected

Classes

BidStatus (Line 13)

Description: No class docstring.

Bid (Line 19)

Description: Represents a bid from an agent.

Task (Line 30)

Description: Represents a task to be contracted.

ContractNetPattern (Line 39)

Description: Contract Net Protocol – Task allocation through bidding. Manager announces tasks, contractors bid, manager awards contracts.

Methods:

__init__ (Line 45)

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”)

announce_task (Line 51)

def announce_task(self, task):
    """Manager announces a task."""

Purpose: Manager announces a task.

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

register_contractor (Line 57)

def register_contractor(self, agent_id, capabilities):
    """Register a contractor agent."""

Purpose: Register a contractor agent.

Parameters: – self: Instance reference – agent_id: Parameter for agent_id – capabilities: Parameter for capabilities

receive_bid (Line 63)

def receive_bid(self, bid):
    """Receive a bid from a contractor."""

Purpose: Receive a bid from a contractor.

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

evaluate_bids (Line 68)

def evaluate_bids(self, task_id):
    """Evaluate bids and select the best one."""

Purpose: Evaluate bids and select the best one.

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

award_contract (Line 86)

def award_contract(self, task_id):
    """Award contract to the best bidder."""

Purpose: Award contract to the best bidder.

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

ContractorAgent (Line 108)

Description: Agent that bids on tasks.

Methods:

__init__ (Line 111)

def __init__(self, agent_id, capabilities, contract_net, llm):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – agent_id: Parameter for agent_id – capabilities: Parameter for capabilities – contract_net: Parameter for contract_net – llm: Parameter for llm

submit_bid (Line 118)

def submit_bid(self, task):
    """Submit a bid for a task."""

Purpose: Submit a bid for a task.

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

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

    """
Pattern 11: Blackboard

Purpose: Enhanced implementation with full LangChain/LangGraph integration
"""

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


@dataclass
class BlackboardConfig:
    """Configuration for Blackboard."""
    model_name: str = "gpt-4"
    temperature: float = 0.7
    max_iterations: int = 10
    enable_logging: bool = True


@dataclass
class BlackboardState:
    """State information for Blackboard."""
    current_step: int = 0
    data: Dict[str, Any] = field(default_factory=dict)
    results: List[Any] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=datetime.now)


class BlackboardAgent:
    """Agent implementing Blackboard."""
    
    def __init__(self, name: str, config: BlackboardConfig = None):
        self.name = name
        self.config = config or BlackboardConfig()
        self.llm = ChatOpenAI(
            model=self.config.model_name,
            temperature=self.config.temperature
        )
        self.state = BlackboardState()
        self.execution_history: List[Dict[str, Any]] = []
    
    def process(self, input_data: Any) -> Dict[str, Any]:
        """Process input according to Blackboard."""
        self.state.current_step += 1
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, implementing Blackboard. "
                      f"Provide clear, structured responses."),
            ("human", "Process this input: {input_data}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"input_data": str(input_data)})
        
        result = {
            "agent": self.name,
            "step": self.state.current_step,
            "input": input_data,
            "output": str(response.content),
            "timestamp": datetime.now()
        }
        
        self.execution_history.append(result)
        self.state.results.append(result)
        
        return result
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get agent statistics."""
        return {
            "name": self.name,
            "total_executions": len(self.execution_history),
            "current_step": self.state.current_step,
            "results_count": len(self.state.results)
        }


class BlackboardPattern:
    """
    Blackboard - Enhanced implementation
    
    This pattern provides comprehensive functionality for blackboard
    with full LangChain/LangGraph integration.
    
    Features:
    - Multi-agent coordination
    - State management
    - Execution tracking
    - LangGraph workflow integration
    """
    
    def __init__(self, config: BlackboardConfig = None):
        self.config = config or BlackboardConfig()
        self.agents: Dict[str, BlackboardAgent] = {}
        self.workflow = self._build_workflow()
        self.execution_log: List[Dict[str, Any]] = []
    
    def _build_workflow(self) -> StateGraph:
        """Build LangGraph workflow for pattern execution."""
        
        class WorkflowState(TypedDict):
            input: str
            processed: str
            output: str
            metadata: Dict[str, Any]
        
        def process_input(state: WorkflowState) -> WorkflowState:
            """Process input node."""
            return {
                **state,
                "processed": f"Processed: {state['input']}",
                "metadata": {"step": "process_input"}
            }
        
        def generate_output(state: WorkflowState) -> WorkflowState:
            """Generate output node."""
            llm = ChatOpenAI(model=self.config.model_name, temperature=self.config.temperature)
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are implementing Blackboard. Generate comprehensive output."),
                ("human", "{input}")
            ])
            
            chain = prompt | llm
            response = chain.invoke({"input": state["processed"]})
            
            return {
                **state,
                "output": str(response.content),
                "metadata": {**state.get("metadata", {}), "step": "generate_output"}
            }
        
        # Build workflow graph
        workflow = StateGraph(WorkflowState)
        workflow.add_node("process", process_input)
        workflow.add_node("generate", generate_output)
        workflow.set_entry_point("process")
        workflow.add_edge("process", "generate")
        workflow.add_edge("generate", END)
        
        return workflow.compile()
    
    def register_agent(self, agent: BlackboardAgent):
        """Register an agent with the pattern."""
        self.agents[agent.name] = agent
        if self.config.enable_logging:
            print(f"Registered agent: {agent.name}")
    
    def execute(self, input_data: Any, agent_name: str = None) -> Dict[str, Any]:
        """Execute the pattern with given input."""
        
        # Use workflow if no specific agent requested
        if agent_name is None:
            result = self.workflow.invoke({
                "input": str(input_data),
                "processed": "",
                "output": "",
                "metadata": {}
            })
            
            execution_record = {
                "pattern": "Blackboard",
                "input": input_data,
                "output": result.get("output", ""),
                "metadata": result.get("metadata", {}),
                "timestamp": datetime.now()
            }
            
            self.execution_log.append(execution_record)
            return execution_record
        
        # Execute with specific agent
        agent = self.agents.get(agent_name)
        if not agent:
            raise ValueError(f"Agent {agent_name} not found")
        
        result = agent.process(input_data)
        self.execution_log.append(result)
        return result
    
    def run(self, input_data: Any) -> Dict[str, Any]:
        """Main execution method."""
        return self.execute(input_data)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get pattern statistics."""
        return {
            "pattern": "Blackboard",
            "total_executions": len(self.execution_log),
            "registered_agents": len(self.agents),
            "agents": [agent.get_statistics() for agent in self.agents.values()]
        }


# Example usage
if __name__ == "__main__":
    print("=== Blackboard Demo ===\n")
    
    # Create configuration
    config = BlackboardConfig(
        model_name="gpt-4",
        temperature=0.7,
        enable_logging=True
    )
    
    # Create pattern
    pattern = BlackboardPattern(config)
    
    # Create and register agents
    agent1 = BlackboardAgent("Agent1", config)
    agent2 = BlackboardAgent("Agent2", config)
    pattern.register_agent(agent1)
    pattern.register_agent(agent2)
    
    print("\nExecuting pattern workflow...")
    
    # Execute with workflow
    result1 = pattern.run("Demonstrate blackboard functionality")
    print(f"\nWorkflow Result:")
    print(f"  Output: {result1['output'][:150]}...")
    
    # Execute with specific agents
    result2 = pattern.execute("Specific task for Agent1", agent_name="Agent1")
    print(f"\nAgent1 Result:")
    print(f"  Output: {result2['output'][:150]}...")
    
    result3 = pattern.execute("Specific task for Agent2", agent_name="Agent2")
    print(f"\nAgent2 Result:")
    print(f"  Output: {result3['output'][:150]}...")
    
    # Show statistics
    print("\n=== Statistics ===")
    stats = pattern.get_statistics()
    print(f"Pattern: {stats['pattern']}")
    print(f"Total Executions: {stats['total_executions']}")
    print(f"Registered Agents: {stats['registered_agents']}")
    print(f"\nAgent Statistics:")
    for agent_stats in stats['agents']:
        print(f"  {agent_stats['name']}: {agent_stats['total_executions']} executions")

    

Agentic Architecture Pattern: Blackboard

Pattern 11: 11 Blackboard – Line-by-Line Explanation

Overview

Pattern 11: Blackboard

Purpose: Enhanced implementation with full LangChain/LangGraph integration

Code Structure Analysis

File Information

• Pattern Number: 11
• Pattern Name: 11 Blackboard
• File Path: patterns\communication\11_blackboard.py
• Total Lines: 247

Detailed Code Explanation

Module Docstring

"""
Pattern 11: Blackboard

Purpose: Enhanced implementation with full LangChain/LangGraph integration
"""

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

Classes

BlackboardConfig (Line 18)

Description: Configuration for Blackboard.

BlackboardState (Line 27)

Description: State information for Blackboard.

BlackboardAgent (Line 36)

Description: Agent implementing Blackboard.

Methods:

__init__ (Line 39)

def __init__(self, name, config):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – config: Configuration settings

process (Line 49)

def process(self, input_data):
    """Process input according to Blackboard."""

Purpose: Process input according to Blackboard.

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

get_statistics (Line 75)

def get_statistics(self):
    """Get agent statistics."""

Purpose: Get agent statistics.

Parameters: – self: Instance reference

BlackboardPattern (Line 85)

Description: Blackboard – Enhanced implementation

This pattern provides comprehensive functionality for blackboard with full LangChain/LangGraph integration.

Features: – Multi-agent coordination – State management – Execution tracking – LangGraph workflow integration

Methods:

__init__ (Line 99)

def __init__(self, config):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – config: Configuration settings

_build_workflow (Line 105)

def _build_workflow(self):
    """Build LangGraph workflow for pattern execution."""

Purpose: Build LangGraph workflow for pattern execution.

Parameters: – self: Instance reference

register_agent (Line 150)

def register_agent(self, agent):
    """Register an agent with the pattern."""

Purpose: Register an agent with the pattern.

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

execute (Line 156)

def execute(self, input_data, agent_name):
    """Execute the pattern with given input."""

Purpose: Execute the pattern with given input.

Parameters: – self: Instance reference – input_data: Parameter for input_data – agent_name: Parameter for agent_name

run (Line 188)

def run(self, input_data):
    """Main execution method."""

Purpose: Main execution method.

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

get_statistics (Line 192)

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

Purpose: Get pattern statistics.

Parameters: – self: Instance reference

WorkflowState (Line 108)

Description: No class docstring.

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