aishastr.live

Agentic Architecture Pattern: Message Passing

Written by

admin

in


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

Comments

Leave a Reply

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

More posts