Building Multi-Agent Systems: Coordination and Communication 🤖

TL;DR – Learn to build sophisticated multi-agent AI systems with proper coordination, communication protocols, and distributed decision-making using modern frameworks and best practices.

Introduction

Multi-agent systems (MAS) represent one of the most exciting frontiers in AI development. Unlike single-agent systems, MAS involves multiple autonomous agents working together to solve complex problems that would be difficult or impossible for a single agent to handle alone.

In this comprehensive guide, we'll explore how to build robust multi-agent systems that can:

• Coordinate actions across multiple agents
• Communicate effectively using standardized protocols
• Make distributed decisions
• Handle conflicts and resource allocation
• Scale to hundreds or thousands of agents

Understanding Multi-Agent Systems

What Are Multi-Agent Systems?

A multi-agent system consists of multiple interacting intelligent agents within an environment. Each agent:

• Has its own goals and objectives
• Can perceive its environment
• Can take actions to modify the environment
• Can communicate with other agents
• Makes decisions autonomously

Key Components

1. Agents: Autonomous entities with specific capabilities
2. Environment: The shared space where agents operate
3. Communication Infrastructure: Protocols for agent interaction
4. Coordination Mechanisms: Methods for organizing agent behavior
5. Conflict Resolution: Systems for handling disagreements

Architecture Patterns

1. Hierarchical Architecture

1# agents/hierarchical.py
2from abc import ABC, abstractmethod
3from typing import List, Dict, Any
4import asyncio
5
6class Agent(ABC):
7    def __init__(self, agent_id: str, role: str):
8        self.agent_id = agent_id
9        self.role = role
10        self.subordinates: List[Agent] = []
11        self.supervisor: Agent = None
12        
13    @abstractmethod
14    async def execute_task(self, task: Dict[str, Any]) -> Dict[str, Any]:
15        pass
16    
17    async def delegate_task(self, task: Dict[str, Any], subordinate: Agent):
18        return await subordinate.execute_task(task)
19    
20    async def report_to_supervisor(self, result: Dict[str, Any]):
21        if self.supervisor:
22            await self.supervisor.receive_report(self.agent_id, result)
23
24class ManagerAgent(Agent):
25    def __init__(self, agent_id: str):
26        super().__init__(agent_id, "manager")
27        self.task_queue = asyncio.Queue()
28        
29    async def execute_task(self, task: Dict[str, Any]) -> Dict[str, Any]:
30        # Decompose task and delegate to subordinates
31        subtasks = self.decompose_task(task)
32        results = []
33        
34        for subtask in subtasks:
35            best_agent = self.select_best_agent(subtask)
36            result = await self.delegate_task(subtask, best_agent)
37            results.append(result)
38            
39        return self.combine_results(results)
40    
41    def decompose_task(self, task: Dict[str, Any]) -> List[Dict[str, Any]]:
42        # Task decomposition logic
43        return [
44            {"type": "data_collection", "params": task.get("data_params", {})},
45            {"type": "analysis", "params": task.get("analysis_params", {})},
46            {"type": "reporting", "params": task.get("report_params", {})}
47        ]
48    
49    def select_best_agent(self, subtask: Dict[str, Any]) -> Agent:
50        # Agent selection based on capabilities and availability
51        for agent in self.subordinates:
52            if agent.can_handle_task(subtask):
53                return agent
54        return self.subordinates[0]  # Fallback
55    
56    def combine_results(self, results: List[Dict[str, Any]]) -> Dict[str, Any]:
57        return {
58            "status": "completed",
59            "results": results,
60            "timestamp": asyncio.get_event_loop().time()
61        }
62
63class WorkerAgent(Agent):
64    def __init__(self, agent_id: str, capabilities: List[str]):
65        super().__init__(agent_id, "worker")
66        self.capabilities = capabilities
67        
68    async def execute_task(self, task: Dict[str, Any]) -> Dict[str, Any]:
69        task_type = task.get("type")
70        
71        if task_type == "data_collection":
72            return await self.collect_data(task["params"])
73        elif task_type == "analysis":
74            return await self.analyze_data(task["params"])
75        elif task_type == "reporting":
76            return await self.generate_report(task["params"])
77        else:
78            return {"error": f"Unknown task type: {task_type}"}
79    
80    def can_handle_task(self, task: Dict[str, Any]) -> bool:
81        return task.get("type") in self.capabilities
82    
83    async def collect_data(self, params: Dict[str, Any]) -> Dict[str, Any]:
84        # Simulate data collection
85        await asyncio.sleep(1)
86        return {"data": f"collected_data_{self.agent_id}", "source": params.get("source")}
87    
88    async def analyze_data(self, params: Dict[str, Any]) -> Dict[str, Any]:
89        # Simulate data analysis
90        await asyncio.sleep(2)
91        return {"analysis": f"analysis_result_{self.agent_id}", "confidence": 0.95}
92    
93    async def generate_report(self, params: Dict[str, Any]) -> Dict[str, Any]:
94        # Simulate report generation
95        await asyncio.sleep(1)
96        return {"report": f"report_{self.agent_id}", "format": params.get("format", "pdf")}

2. Peer-to-Peer Architecture

1# agents/p2p.py
2import asyncio
3from typing import Set, Dict, Any
4import json
5
6class P2PAgent:
7    def __init__(self, agent_id: str, capabilities: List[str]):
8        self.agent_id = agent_id
9        self.capabilities = capabilities
10        self.peers: Set[str] = set()
11        self.message_queue = asyncio.Queue()
12        self.knowledge_base = {}
13        
14    async def connect_to_peer(self, peer_id: str):
15        self.peers.add(peer_id)
16        await self.send_message(peer_id, {
17            "type": "connection_request",
18            "sender": self.agent_id,
19            "capabilities": self.capabilities
20        })
21    
22    async def send_message(self, recipient: str, message: Dict[str, Any]):
23        # In a real implementation, this would use network protocols
24        message["sender"] = self.agent_id
25        message["timestamp"] = asyncio.get_event_loop().time()
26        # Simulate message delivery
27        await asyncio.sleep(0.1)
28        
29    async def broadcast_message(self, message: Dict[str, Any]):
30        tasks = []
31        for peer in self.peers:
32            tasks.append(self.send_message(peer, message))
33        await asyncio.gather(*tasks)
34    
35    async def handle_messages(self):
36        while True:
37            try:
38                message = await asyncio.wait_for(self.message_queue.get(), timeout=1.0)
39                await self.process_message(message)
40            except asyncio.TimeoutError:
41                continue
42    
43    async def process_message(self, message: Dict[str, Any]):
44        msg_type = message.get("type")
45        
46        if msg_type == "connection_request":
47            await self.handle_connection_request(message)
48        elif msg_type == "task_request":
49            await self.handle_task_request(message)
50        elif msg_type == "knowledge_share":
51            await self.handle_knowledge_share(message)
52        elif msg_type == "coordination":
53            await self.handle_coordination(message)
54    
55    async def handle_connection_request(self, message: Dict[str, Any]):
56        sender = message["sender"]
57        self.peers.add(sender)
58        
59        # Send acknowledgment
60        await self.send_message(sender, {
61            "type": "connection_ack",
62            "capabilities": self.capabilities
63        })
64    
65    async def handle_task_request(self, message: Dict[str, Any]):
66        task = message.get("task", {})
67        
68        if self.can_handle_task(task):
69            result = await self.execute_task(task)
70            await self.send_message(message["sender"], {
71                "type": "task_result",
72                "task_id": task.get("id"),
73                "result": result
74            })
75        else:
76            # Forward to capable peer
77            capable_peer = self.find_capable_peer(task)
78            if capable_peer:
79                await self.send_message(capable_peer, message)
80    
81    def can_handle_task(self, task: Dict[str, Any]) -> bool:
82        required_capability = task.get("required_capability")
83        return required_capability in self.capabilities
84    
85    def find_capable_peer(self, task: Dict[str, Any]) -> str:
86        # In a real implementation, maintain peer capability registry
87        return list(self.peers)[0] if self.peers else None
88    
89    async def execute_task(self, task: Dict[str, Any]) -> Dict[str, Any]:
90        # Task execution logic
91        await asyncio.sleep(1)
92        return {"status": "completed", "agent": self.agent_id}

Communication Protocols

1. Message Passing

1# communication/message_passing.py
2from enum import Enum
3from dataclasses import dataclass
4from typing import Any, Optional
5import asyncio
6import json
7
8class MessageType(Enum):
9    REQUEST = "request"
10    RESPONSE = "response"
11    BROADCAST = "broadcast"
12    COORDINATION = "coordination"
13    HEARTBEAT = "heartbeat"
14
15@dataclass
16class Message:
17    sender: str
18    recipient: str
19    message_type: MessageType
20    content: Dict[str, Any]
21    timestamp: float
22    message_id: str
23    reply_to: Optional[str] = None
24
25class MessageBus:
26    def __init__(self):
27        self.subscribers = {}
28        self.message_history = []
29        
30    def subscribe(self, agent_id: str, callback):
31        if agent_id not in self.subscribers:
32            self.subscribers[agent_id] = []
33        self.subscribers[agent_id].append(callback)
34    
35    async def publish(self, message: Message):
36        self.message_history.append(message)
37        
38        # Deliver to specific recipient
39        if message.recipient in self.subscribers:
40            for callback in self.subscribers[message.recipient]:
41                await callback(message)
42        
43        # Handle broadcasts
44        if message.message_type == MessageType.BROADCAST:
45            for agent_id, callbacks in self.subscribers.items():
46                if agent_id != message.sender:
47                    for callback in callbacks:
48                        await callback(message)
49
50class CommunicationProtocol:
51    def __init__(self, agent_id: str, message_bus: MessageBus):
52        self.agent_id = agent_id
53        self.message_bus = message_bus
54        self.pending_requests = {}
55        
56        # Subscribe to messages
57        self.message_bus.subscribe(agent_id, self.handle_message)
58    
59    async def send_request(self, recipient: str, content: Dict[str, Any]) -> Dict[str, Any]:
60        message_id = f"{self.agent_id}_{asyncio.get_event_loop().time()}"
61        
62        message = Message(
63            sender=self.agent_id,
64            recipient=recipient,
65            message_type=MessageType.REQUEST,
66            content=content,
67            timestamp=asyncio.get_event_loop().time(),
68            message_id=message_id
69        )
70        
71        # Create future for response
72        future = asyncio.Future()
73        self.pending_requests[message_id] = future
74        
75        await self.message_bus.publish(message)
76        
77        # Wait for response with timeout
78        try:
79            response = await asyncio.wait_for(future, timeout=30.0)
80            return response
81        except asyncio.TimeoutError:
82            del self.pending_requests[message_id]
83            raise TimeoutError(f"Request to {recipient} timed out")
84    
85    async def send_response(self, original_message: Message, content: Dict[str, Any]):
86        response = Message(
87            sender=self.agent_id,
88            recipient=original_message.sender,
89            message_type=MessageType.RESPONSE,
90            content=content,
91            timestamp=asyncio.get_event_loop().time(),
92            message_id=f"resp_{original_message.message_id}",
93            reply_to=original_message.message_id
94        )
95        
96        await self.message_bus.publish(response)
97    
98    async def broadcast(self, content: Dict[str, Any]):
99        message = Message(
100            sender=self.agent_id,
101            recipient="*",
102            message_type=MessageType.BROADCAST,
103            content=content,
104            timestamp=asyncio.get_event_loop().time(),
105            message_id=f"broadcast_{self.agent_id}_{asyncio.get_event_loop().time()}"
106        )
107        
108        await self.message_bus.publish(message)
109    
110    async def handle_message(self, message: Message):
111        if message.message_type == MessageType.RESPONSE:
112            # Handle response to our request
113            if message.reply_to in self.pending_requests:
114                future = self.pending_requests.pop(message.reply_to)
115                future.set_result(message.content)
116        elif message.message_type == MessageType.REQUEST:
117            # Handle incoming request
118            await self.handle_request(message)
119        elif message.message_type == MessageType.BROADCAST:
120            # Handle broadcast message
121            await self.handle_broadcast(message)
122    
123    async def handle_request(self, message: Message):
124        # Override in subclasses
125        await self.send_response(message, {"status": "not_implemented"})
126    
127    async def handle_broadcast(self, message: Message):
128        # Override in subclasses
129        pass

2. Consensus Mechanisms

1# coordination/consensus.py
2from typing import Dict, List, Any, Optional
3import asyncio
4import random
5
6class RaftConsensus:
7    def __init__(self, agent_id: str, peers: List[str]):
8        self.agent_id = agent_id
9        self.peers = peers
10        self.state = "follower"  # follower, candidate, leader
11        self.current_term = 0
12        self.voted_for = None
13        self.log = []
14        self.commit_index = 0
15        self.last_applied = 0
16        
17        # Leader state
18        self.next_index = {}
19        self.match_index = {}
20        
21        # Election timeout
22        self.election_timeout = random.uniform(5, 10)
23        self.last_heartbeat = asyncio.get_event_loop().time()
24    
25    async def start(self):
26        asyncio.create_task(self.election_timer())
27        if self.state == "leader":
28            asyncio.create_task(self.send_heartbeats())
29    
30    async def election_timer(self):
31        while True:
32            await asyncio.sleep(0.1)
33            
34            if self.state != "leader":
35                current_time = asyncio.get_event_loop().time()
36                if current_time - self.last_heartbeat > self.election_timeout:
37                    await self.start_election()
38    
39    async def start_election(self):
40        self.state = "candidate"
41        self.current_term += 1
42        self.voted_for = self.agent_id
43        self.last_heartbeat = asyncio.get_event_loop().time()
44        
45        votes = 1  # Vote for self
46        
47        # Request votes from peers
48        for peer in self.peers:
49            vote_granted = await self.request_vote(peer)
50            if vote_granted:
51                votes += 1
52        
53        # Check if won election
54        if votes > len(self.peers) // 2:
55            self.become_leader()
56        else:
57            self.state = "follower"
58    
59    async def request_vote(self, peer: str) -> bool:
60        # Send vote request to peer
61        request = {
62            "type": "vote_request",
63            "term": self.current_term,
64            "candidate_id": self.agent_id,
65            "last_log_index": len(self.log) - 1,
66            "last_log_term": self.log[-1]["term"] if self.log else 0
67        }
68        
69        # Simulate network call
70        await asyncio.sleep(random.uniform(0.1, 0.5))
71        
72        # Simulate response (in real implementation, this would be actual network communication)
73        return random.choice([True, False])
74    
75    def become_leader(self):
76        self.state = "leader"
77        
78        # Initialize leader state
79        for peer in self.peers:
80            self.next_index[peer] = len(self.log)
81            self.match_index[peer] = 0
82        
83        asyncio.create_task(self.send_heartbeats())
84    
85    async def send_heartbeats(self):
86        while self.state == "leader":
87            for peer in self.peers:
88                await self.send_append_entries(peer)
89            await asyncio.sleep(1)  # Heartbeat interval
90    
91    async def send_append_entries(self, peer: str):
92        prev_log_index = self.next_index[peer] - 1
93        prev_log_term = self.log[prev_log_index]["term"] if prev_log_index >= 0 else 0
94        
95        entries = self.log[self.next_index[peer]:]
96        
97        request = {
98            "type": "append_entries",
99            "term": self.current_term,
100            "leader_id": self.agent_id,
101            "prev_log_index": prev_log_index,
102            "prev_log_term": prev_log_term,
103            "entries": entries,
104            "leader_commit": self.commit_index
105        }
106        
107        # Simulate network call
108        await asyncio.sleep(random.uniform(0.05, 0.2))
109        
110        # In real implementation, handle response
111        success = random.choice([True, False])
112        if success:
113            self.match_index[peer] = prev_log_index + len(entries)
114            self.next_index[peer] = self.match_index[peer] + 1
115    
116    async def append_log_entry(self, entry: Dict[str, Any]):
117        if self.state == "leader":
118            entry["term"] = self.current_term
119            self.log.append(entry)
120            
121            # Replicate to followers
122            for peer in self.peers:
123                await self.send_append_entries(peer)

Coordination Strategies

1. Task Allocation

1# coordination/task_allocation.py
2from typing import List, Dict, Any, Tuple
3import heapq
4from dataclasses import dataclass
5
6@dataclass
7class Task:
8    id: str
9    priority: int
10    required_capabilities: List[str]
11    estimated_duration: float
12    deadline: float
13    dependencies: List[str]
14
15@dataclass
16class AgentCapability:
17    agent_id: str
18    capabilities: List[str]
19    current_load: float
20    max_capacity: float
21    efficiency_ratings: Dict[str, float]
22
23class TaskAllocator:
24    def __init__(self):
25        self.agents: Dict[str, AgentCapability] = {}
26        self.tasks: Dict[str, Task] = {}
27        self.allocations: Dict[str, str] = {}  # task_id -> agent_id
28    
29    def register_agent(self, agent: AgentCapability):
30        self.agents[agent.agent_id] = agent
31    
32    def add_task(self, task: Task):
33        self.tasks[task.id] = task
34    
35    def allocate_tasks(self) -> Dict[str, str]:
36        # Sort tasks by priority and deadline
37        sorted_tasks = sorted(
38            self.tasks.values(),
39            key=lambda t: (t.priority, t.deadline)
40        )
41        
42        allocations = {}
43        
44        for task in sorted_tasks:
45            best_agent = self.find_best_agent(task)
46            if best_agent:
47                allocations[task.id] = best_agent.agent_id
48                # Update agent load
49                best_agent.current_load += task.estimated_duration
50        
51        return allocations
52    
53    def find_best_agent(self, task: Task) -> AgentCapability:
54        eligible_agents = []
55        
56        for agent in self.agents.values():
57            if self.can_handle_task(agent, task):
58                score = self.calculate_agent_score(agent, task)
59                eligible_agents.append((score, agent))
60        
61        if eligible_agents:
62            # Return agent with highest score
63            return max(eligible_agents, key=lambda x: x[0])[1]
64        
65        return None
66    
67    def can_handle_task(self, agent: AgentCapability, task: Task) -> bool:
68        # Check if agent has required capabilities
69        if not all(cap in agent.capabilities for cap in task.required_capabilities):
70            return False
71        
72        # Check if agent has capacity
73        if agent.current_load + task.estimated_duration > agent.max_capacity:
74            return False
75        
76        return True
77    
78    def calculate_agent_score(self, agent: AgentCapability, task: Task) -> float:
79        # Calculate efficiency score
80        efficiency_scores = [
81            agent.efficiency_ratings.get(cap, 0.5)
82            for cap in task.required_capabilities
83        ]
84        avg_efficiency = sum(efficiency_scores) / len(efficiency_scores)
85        
86        # Calculate load factor (prefer less loaded agents)
87        load_factor = 1 - (agent.current_load / agent.max_capacity)
88        
89        # Combine scores
90        return avg_efficiency * 0.7 + load_factor * 0.3
91
92class AuctionBasedAllocator:
93    def __init__(self):
94        self.agents: Dict[str, AgentCapability] = {}
95        
96    async def allocate_task_by_auction(self, task: Task) -> str:
97        # Send task announcement to all agents
98        bids = {}
99        
100        for agent_id, agent in self.agents.items():
101            if self.can_handle_task(agent, task):
102                bid = await self.request_bid(agent_id, task)
103                if bid:
104                    bids[agent_id] = bid
105        
106        if bids:
107            # Select winner (lowest cost bid)
108            winner = min(bids.items(), key=lambda x: x[1]["cost"])
109            return winner[0]
110        
111        return None
112    
113    async def request_bid(self, agent_id: str, task: Task) -> Dict[str, Any]:
114        # Simulate bid request
115        await asyncio.sleep(0.1)
116        
117        agent = self.agents[agent_id]
118        
119        # Calculate bid based on current load and efficiency
120        base_cost = task.estimated_duration
121        load_multiplier = 1 + (agent.current_load / agent.max_capacity)
122        
123        efficiency = min([
124            agent.efficiency_ratings.get(cap, 0.5)
125            for cap in task.required_capabilities
126        ])
127        
128        final_cost = base_cost * load_multiplier / efficiency
129        
130        return {
131            "cost": final_cost,
132            "estimated_completion": asyncio.get_event_loop().time() + final_cost,
133            "confidence": efficiency
134        }

2. Conflict Resolution

1# coordination/conflict_resolution.py
2from typing import List, Dict, Any, Optional
3from enum import Enum
4
5class ConflictType(Enum):
6    RESOURCE_CONFLICT = "resource_conflict"
7    GOAL_CONFLICT = "goal_conflict"
8    PRIORITY_CONFLICT = "priority_conflict"
9    COORDINATION_CONFLICT = "coordination_conflict"
10
11@dataclass
12class Conflict:
13    id: str
14    type: ConflictType
15    involved_agents: List[str]
16    resources: List[str]
17    description: str
18    severity: int  # 1-10
19    timestamp: float
20
21class ConflictResolver:
22    def __init__(self):
23        self.active_conflicts: Dict[str, Conflict] = {}
24        self.resolution_strategies = {
25            ConflictType.RESOURCE_CONFLICT: self.resolve_resource_conflict,
26            ConflictType.GOAL_CONFLICT: self.resolve_goal_conflict,
27            ConflictType.PRIORITY_CONFLICT: self.resolve_priority_conflict,
28            ConflictType.COORDINATION_CONFLICT: self.resolve_coordination_conflict
29        }
30    
31    async def detect_conflict(self, agents: List[str], resources: List[str]) -> Optional[Conflict]:
32        # Resource conflict detection
33        resource_usage = {}
34        for agent in agents:
35            agent_resources = await self.get_agent_resources(agent)
36            for resource in agent_resources:
37                if resource in resource_usage:
38                    # Conflict detected
39                    conflict = Conflict(
40                        id=f"conflict_{asyncio.get_event_loop().time()}",
41                        type=ConflictType.RESOURCE_CONFLICT,
42                        involved_agents=[resource_usage[resource], agent],
43                        resources=[resource],
44                        description=f"Resource {resource} requested by multiple agents",
45                        severity=5,
46                        timestamp=asyncio.get_event_loop().time()
47                    )
48                    return conflict
49                resource_usage[resource] = agent
50        
51        return None
52    
53    async def resolve_conflict(self, conflict: Conflict) -> Dict[str, Any]:
54        strategy = self.resolution_strategies.get(conflict.type)
55        if strategy:
56            return await strategy(conflict)
57        else:
58            return {"status": "unresolved", "reason": "no_strategy"}
59    
60    async def resolve_resource_conflict(self, conflict: Conflict) -> Dict[str, Any]:
61        # Priority-based resolution
62        agent_priorities = {}
63        for agent in conflict.involved_agents:
64            priority = await self.get_agent_priority(agent)
65            agent_priorities[agent] = priority
66        
67        # Assign resource to highest priority agent
68        winner = max(agent_priorities.items(), key=lambda x: x[1])
69        
70        # Notify agents
71        for agent in conflict.involved_agents:
72            if agent == winner[0]:
73                await self.notify_agent(agent, {
74                    "type": "resource_granted",
75                    "resources": conflict.resources
76                })
77            else:
78                await self.notify_agent(agent, {
79                    "type": "resource_denied",
80                    "resources": conflict.resources,
81                    "reason": "lower_priority"
82                })
83        
84        return {
85            "status": "resolved",
86            "winner": winner[0],
87            "method": "priority_based"
88        }
89    
90    async def resolve_goal_conflict(self, conflict: Conflict) -> Dict[str, Any]:
91        # Negotiation-based resolution
92        proposals = {}
93        
94        for agent in conflict.involved_agents:
95            proposal = await self.request_proposal(agent, conflict)
96            proposals[agent] = proposal
97        
98        # Find compromise solution
99        compromise = self.find_compromise(proposals)
100        
101        # Notify agents of compromise
102        for agent in conflict.involved_agents:
103            await self.notify_agent(agent, {
104                "type": "compromise_solution",
105                "solution": compromise
106            })
107        
108        return {
109            "status": "resolved",
110            "solution": compromise,
111            "method": "negotiation"
112        }
113    
114    async def get_agent_priority(self, agent_id: str) -> int:
115        # Get agent priority (implementation specific)
116        return 5  # Default priority
117    
118    async def get_agent_resources(self, agent_id: str) -> List[str]:
119        # Get resources requested by agent
120        return []  # Implementation specific
121    
122    async def notify_agent(self, agent_id: str, message: Dict[str, Any]):
123        # Send notification to agent
124        pass
125    
126    async def request_proposal(self, agent_id: str, conflict: Conflict) -> Dict[str, Any]:
127        # Request proposal from agent for conflict resolution
128        return {"proposal": "default"}
129    
130    def find_compromise(self, proposals: Dict[str, Dict[str, Any]]) -> Dict[str, Any]:
131        # Find compromise between proposals
132        return {"type": "compromise", "details": "balanced_solution"}

Implementation Example

Let's put it all together in a complete multi-agent system:

1# main.py
2import asyncio
3from agents.hierarchical import ManagerAgent, WorkerAgent
4from communication.message_passing import MessageBus, CommunicationProtocol
5from coordination.task_allocation import TaskAllocator, Task, AgentCapability
6
7async def main():
8    # Create message bus
9    message_bus = MessageBus()
10    
11    # Create agents
12    manager = ManagerAgent("manager_001")
13    
14    workers = [
15        WorkerAgent("worker_001", ["data_collection", "analysis"]),
16        WorkerAgent("worker_002", ["analysis", "reporting"]),
17        WorkerAgent("worker_003", ["data_collection", "reporting"])
18    ]
19    
20    # Set up hierarchy
21    for worker in workers:
22        manager.subordinates.append(worker)
23        worker.supervisor = manager
24    
25    # Create communication protocols
26    protocols = {}
27    for agent in [manager] + workers:
28        protocols[agent.agent_id] = CommunicationProtocol(agent.agent_id, message_bus)
29    
30    # Set up task allocator
31    allocator = TaskAllocator()
32    
33    # Register agents
34    for worker in workers:
35        capability = AgentCapability(
36            agent_id=worker.agent_id,
37            capabilities=worker.capabilities,
38            current_load=0.0,
39            max_capacity=10.0,
40            efficiency_ratings={cap: 0.8 for cap in worker.capabilities}
41        )
42        allocator.register_agent(capability)
43    
44    # Create and allocate tasks
45    tasks = [
46        Task(
47            id="task_001",
48            priority=1,
49            required_capabilities=["data_collection"],
50            estimated_duration=2.0,
51            deadline=asyncio.get_event_loop().time() + 10,
52            dependencies=[]
53        ),
54        Task(
55            id="task_002",
56            priority=2,
57            required_capabilities=["analysis"],
58            estimated_duration=3.0,
59            deadline=asyncio.get_event_loop().time() + 15,
60            dependencies=["task_001"]
61        )
62    ]
63    
64    for task in tasks:
65        allocator.add_task(task)
66    
67    allocations = allocator.allocate_tasks()
68    print("Task allocations:", allocations)
69    
70    # Execute tasks
71    main_task = {
72        "type": "complex_analysis",
73        "data_params": {"source": "database"},
74        "analysis_params": {"method": "ml"},
75        "report_params": {"format": "pdf"}
76    }
77    
78    result = await manager.execute_task(main_task)
79    print("Task result:", result)
80
81if __name__ == "__main__":
82    asyncio.run(main())

Testing and Monitoring

1# testing/mas_test.py
2import unittest
3import asyncio
4from unittest.mock import Mock, patch
5
6class TestMultiAgentSystem(unittest.TestCase):
7    def setUp(self):
8        self.loop = asyncio.new_event_loop()
9        asyncio.set_event_loop(self.loop)
10    
11    def tearDown(self):
12        self.loop.close()
13    
14    def test_agent_communication(self):
15        async def test():
16            message_bus = MessageBus()
17            
18            # Create mock agents
19            agent1 = Mock()
20            agent2 = Mock()
21            
22            # Test message passing
23            protocol1 = CommunicationProtocol("agent1", message_bus)
24            protocol2 = CommunicationProtocol("agent2", message_bus)
25            
26            # Send message
27            response = await protocol1.send_request("agent2", {"test": "data"})
28            
29            # Verify message was received
30            self.assertIsNotNone(response)
31        
32        self.loop.run_until_complete(test())
33    
34    def test_task_allocation(self):
35        allocator = TaskAllocator()
36        
37        # Add test agents
38        agent = AgentCapability(
39            agent_id="test_agent",
40            capabilities=["test_capability"],
41            current_load=0.0,
42            max_capacity=10.0,
43            efficiency_ratings={"test_capability": 0.9}
44        )
45        allocator.register_agent(agent)
46        
47        # Add test task
48        task = Task(
49            id="test_task",
50            priority=1,
51            required_capabilities=["test_capability"],
52            estimated_duration=2.0,
53            deadline=100.0,
54            dependencies=[]
55        )
56        allocator.add_task(task)
57        
58        # Test allocation
59        allocations = allocator.allocate_tasks()
60        self.assertEqual(allocations["test_task"], "test_agent")
61
62# monitoring/metrics.py
63class MASMetrics:
64    def __init__(self):
65        self.message_count = 0
66        self.task_completion_times = []
67        self.agent_utilization = {}
68        self.conflict_count = 0
69    
70    def record_message(self):
71        self.message_count += 1
72    
73    def record_task_completion(self, duration: float):
74        self.task_completion_times.append(duration)
75    
76    def record_agent_utilization(self, agent_id: str, utilization: float):
77        self.agent_utilization[agent_id] = utilization
78    
79    def record_conflict(self):
80        self.conflict_count += 1
81    
82    def get_metrics(self) -> Dict[str, Any]:
83        avg_completion_time = (
84            sum(self.task_completion_times) / len(self.task_completion_times)
85            if self.task_completion_times else 0
86        )
87        
88        return {
89            "total_messages": self.message_count,
90            "average_task_completion_time": avg_completion_time,
91            "agent_utilization": self.agent_utilization,
92            "total_conflicts": self.conflict_count,
93            "system_efficiency": self.calculate_efficiency()
94        }
95    
96    def calculate_efficiency(self) -> float:
97        if not self.agent_utilization:
98            return 0.0
99        
100        total_utilization = sum(self.agent_utilization.values())
101        avg_utilization = total_utilization / len(self.agent_utilization)
102        
103        # Factor in conflict rate
104        conflict_penalty = min(self.conflict_count * 0.1, 0.5)
105        
106        return max(0.0, avg_utilization - conflict_penalty)

Best Practices and Optimization

1. Scalability Considerations

• Hierarchical Organization: Use hierarchical structures for large systems
• Load Balancing: Distribute tasks evenly across agents
• Caching: Cache frequently accessed information
• Asynchronous Communication: Use async patterns for better performance

2. Fault Tolerance

• Redundancy: Have backup agents for critical functions
• Health Monitoring: Regularly check agent status
• Graceful Degradation: System should continue operating with reduced capacity
• Recovery Mechanisms: Automatic restart and state recovery

3. Security

• Authentication: Verify agent identities
• Authorization: Control access to resources and capabilities
• Encryption: Secure inter-agent communication
• Audit Logging: Track all agent actions

Conclusion

Building effective multi-agent systems requires careful consideration of:

• Architecture patterns that fit your use case
• Communication protocols for reliable interaction
• Coordination mechanisms for organized behavior
• Conflict resolution strategies for handling disputes
• Monitoring and testing for system reliability

The examples provided offer a solid foundation for building sophisticated multi-agent systems that can scale and adapt to complex real-world scenarios.

Next Steps

1. Implement machine learning for adaptive agent behavior
2. Add blockchain integration for decentralized coordination
3. Develop specialized agents for domain-specific tasks
4. Create visualization tools for system monitoring
5. Integrate with external APIs for enhanced capabilities

Multi-agent systems represent the future of distributed AI, enabling solutions that are more robust, scalable, and intelligent than single-agent approaches.