"""
Pattern 02: Multi-Agent System (MAS)
Multiple agents working together, coordinating and collaborating to achieve goals.
"""

from typing import Dict, Any, List, Optional
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 Message:
    """Message exchanged between agents."""
    sender: str
    receiver: str
    content: str
    timestamp: datetime = field(default_factory=datetime.now)


class AgentState(TypedDict):
    """State shared among agents."""
    messages: List[Message]
    current_agent: str
    task: str
    results: Dict[str, Any]
    iteration: int


class Agent:
    """Individual agent in the multi-agent system."""
    
    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.memory: List[Message] = []
    
    def process(self, task: str, context: Dict[str, Any]) -> str:
        """Process a task with given context."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"""You are {self.name}, a {self.role} agent.
            Your role is to {self.role}.
            Previous context: {context}"""),
            ("human", "{task}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"task": task})
        return str(response.content)
    
    def send_message(self, receiver: str, content: str) -> Message:
        """Send a message to another agent."""
        msg = Message(sender=self.name, receiver=receiver, content=content)
        self.memory.append(msg)
        return msg
    
    def receive_message(self, message: Message):
        """Receive a message from another agent."""
        self.memory.append(message)


class MultiAgentSystemPattern:
    """
    Multi-Agent System implementation using LangGraph.
    
    Features:
    - Multiple specialized agents
    - Agent communication and coordination
    - Shared state management
    - Sequential and parallel execution
    """
    
    def __init__(self):
        self.agents: Dict[str, Agent] = {}
        self.workflow = None
        self.message_history: List[Message] = []
    
    def add_agent(self, agent: Agent):
        """Add an agent to the system."""
        self.agents[agent.name] = agent
    
    def _agent_node(self, state: AgentState, agent_name: str) -> AgentState:
        """Node function for an agent in the workflow."""
        agent = self.agents[agent_name]
        
        # Get relevant messages for this agent
        relevant_msgs = [m for m in state["messages"] if m.receiver == agent_name or m.receiver == "all"]
        context = {
            "messages": [f"{m.sender}: {m.content}" for m in relevant_msgs],
            "results": state["results"]
        }
        
        # Process task
        result = agent.process(state["task"], context)
        
        # Update state
        state["results"][agent_name] = result
        state["current_agent"] = agent_name
        state["iteration"] += 1
        
        return state
    
    def build_workflow(self, agent_sequence: List[str]):
        """Build a sequential workflow through agents."""
        workflow = StateGraph(AgentState)
        
        # Add nodes for each agent
        for agent_name in agent_sequence:
            workflow.add_node(
                agent_name,
                lambda state, name=agent_name: self._agent_node(state, name)
            )
        
        # Add edges to connect agents sequentially
        for i in range(len(agent_sequence) - 1):
            workflow.add_edge(agent_sequence[i], agent_sequence[i + 1])
        
        # Set entry and exit points
        workflow.set_entry_point(agent_sequence[0])
        workflow.add_edge(agent_sequence[-1], END)
        
        self.workflow = workflow.compile()
    
    def broadcast(self, sender: str, content: str):
        """Broadcast a message to all agents."""
        msg = Message(sender=sender, receiver="all", content=content)
        self.message_history.append(msg)
        for agent in self.agents.values():
            if agent.name != sender:
                agent.receive_message(msg)
    
    def run(self, task: str, agent_sequence: Optional[List[str]] = None) -> Dict[str, Any]:
        """
        Execute the multi-agent system on a task.
        
        Args:
            task: The task for agents to work on
            agent_sequence: Optional sequence of agents to execute
            
        Returns:
            Dictionary with results from all agents
        """
        if agent_sequence is None:
            agent_sequence = list(self.agents.keys())
        
        if not self.workflow:
            self.build_workflow(agent_sequence)
        
        # Initialize state
        initial_state: AgentState = {
            "messages": [],
            "current_agent": "",
            "task": task,
            "results": {},
            "iteration": 0
        }
        
        # Execute workflow
        final_state = self.workflow.invoke(initial_state)
        
        return {
            "task": task,
            "agents": agent_sequence,
            "results": final_state["results"],
            "iterations": final_state["iteration"]
        }


# Example usage
if __name__ == "__main__":
    print("\n=== Multi-Agent System Example ===")
    
    # Create multi-agent system
    mas = MultiAgentSystemPattern()
    
    # Create specialized agents
    researcher = Agent("Researcher", "research and gather information")
    analyzer = Agent("Analyzer", "analyze data and identify patterns")
    writer = Agent("Writer", "write clear and comprehensive reports")
    
    # Add agents to system
    mas.add_agent(researcher)
    mas.add_agent(analyzer)
    mas.add_agent(writer)
    
    # Run a collaborative task
    task = "Create a report about the benefits of exercise"
    result = mas.run(task, ["Researcher", "Analyzer", "Writer"])
    
    print(f"\nTask: {result['task']}")
    print(f"\nAgent Sequence: {' -> '.join(result['agents'])}")
    print(f"\nResults from each agent:")
    for agent_name, output in result['results'].items():
        print(f"\n{agent_name}:")
        print(f"  {output[:200]}..." if len(output) > 200 else f"  {output}")
    
    print(f"\nTotal iterations: {result['iterations']}")

    

Agentic Architecture Pattern: Multi Agent System

Pattern 02: 02 Multi Agent System – Line-by-Line Explanation

Overview

Pattern 02: Multi-Agent System (MAS) Multiple agents working together, coordinating and collaborating to achieve goals.

Code Structure Analysis

File Information

• Pattern Number: 02
• Pattern Name: 02 Multi Agent System
• File Path: patterns\core_agent\02_multi_agent_system.py
• Total Lines: 202

Detailed Code Explanation

Module Docstring

"""
Pattern 02: Multi-Agent System (MAS)
Multiple agents working together, coordinating and collaborating to achieve goals.
"""

Explanation: This module docstring describes the purpose and key features of the 02 Multi Agent System pattern.

Imports Section

from typing import Dict, Any, List, Optional
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: 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

Message (Line 16)

Description: Message exchanged between agents.

AgentState (Line 24)

Description: State shared among agents.

Agent (Line 33)

Description: Individual agent in the multi-agent system.

Methods:

__init__ (Line 36)

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

process (Line 42)

def process(self, task, context):
    """Process a task with given context."""

Purpose: Process a task with given context.

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

send_message (Line 55)

def send_message(self, receiver, content):
    """Send a message to another agent."""

Purpose: Send a message to another agent.

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

receive_message (Line 61)

def receive_message(self, message):
    """Receive a message from another agent."""

Purpose: Receive a message from another agent.

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

MultiAgentSystemPattern (Line 66)

Description: Multi-Agent System implementation using LangGraph.

Features: – Multiple specialized agents – Agent communication and coordination – Shared state management – Sequential and parallel execution

Methods:

__init__ (Line 77)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

add_agent (Line 82)

def add_agent(self, agent):
    """Add an agent to the system."""

Purpose: Add an agent to the system.

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

_agent_node (Line 86)

def _agent_node(self, state, agent_name):
    """Node function for an agent in the workflow."""

Purpose: Node function for an agent in the workflow.

Parameters: – self: Instance reference – state: Current workflow state – agent_name: Parameter for agent_name

build_workflow (Line 107)

def build_workflow(self, agent_sequence):
    """Build a sequential workflow through agents."""

Purpose: Build a sequential workflow through agents.

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

broadcast (Line 128)

def broadcast(self, sender, content):
    """Broadcast a message to all agents."""

Purpose: Broadcast a message to all agents.

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

run (Line 136)

def run(self, task, agent_sequence):
    """Execute the multi-agent system on a task.

Args:
    task: The task for agents to work on
    agent_sequence: Optional sequence of agents to execute

Returns:
    Dictionary with results from all agents"""

Purpose: Execute the multi-agent system on a task.

Args: task: The task for agents to work on agent_sequence: Optional sequence of agents to execute

Returns: Dictionary with results from all agents

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

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