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