Multi-Agent Orchestration: Building Parallel Coordination Systems
Complete Python implementation: Event-driven coordination, shared state management, and conflict resolution for multi-agent systems that work in parallel without blocking.
The Problem: Agent Coordination Hell
You've built individual AI agents that work great in isolation. Your infrastructure agent can deploy AWS resources, your content agent can migrate blog posts, and your security agent can configure SSL. But when you try to run them together, everything breaks.
The infrastructure agent creates an S3 bucket while the content agent is already trying to upload files. The security agent configures CloudFront before the infrastructure agent finishes creating the distribution. Agents overwrite each other's state changes, timeout waiting for dependencies, and fail in unpredictable ways.
This is the coordination problem: specialized agents that can't work together. You end up with either sequential execution (slow and inefficient) or parallel chaos (fast and broken).
The Solution: Event-Driven Multi-Agent Orchestration
We built a coordination system that lets specialized agents work in parallel without stepping on each other. Here's how it works:
- Event-driven communication: Agents communicate through immutable events, not shared state
- Dependency-aware execution: The system automatically orders agents based on their dependencies
- Conflict detection and resolution: When agents clash, the system resolves conflicts systematically
- Resource locking: Agents can request exclusive access to shared resources
- Complete observability: Every coordination decision is logged and auditable
The result: Infrastructure deployment that used to take 15 minutes sequentially now runs in 4 minutes with parallel coordination—without any coordination failures.
What You'll Learn
This guide provides a complete Python implementation for building production-ready multi-agent coordination systems. By the end, you'll have:
Core Coordination System:
- Event-driven communication bus for agent coordination
- Dependency-aware execution engine with automatic ordering
- Shared state management with conflict detection and resolution
- Resource locking system for exclusive access control
Agent Framework:
- Standardized agent interface for reliable coordination
- Specialized agent implementations (Infrastructure, Content, Security, QA)
- Error handling and graceful failure recovery
- Complete lifecycle management with cleanup procedures
Advanced Features:
- Real-time orchestration monitoring and metrics
- Conflict resolution strategies (timestamp-based, priority-based)
- Performance optimization for concurrent agent execution
- Comprehensive testing framework for coordination validation
- Integration with Prometheus for production monitoring
Production Deployment:
- Kubernetes deployment with Redis backend for coordination
- ConfigMap-based policy management without code changes
- Network policies and security configurations
- Health checks and rolling update strategies
- Auto-scaling and resource management
Key Components:
EventBus- Central event coordination systemCoordinationEngine- Dependency resolution and parallel executionSharedStateManager- Conflict-free state updatesResourceManager- Exclusive resource access control
Prerequisites: Python 3.8+, AsyncIO knowledge, Redis, Kubernetes, understanding of distributed systems concepts
Estimated Setup Time: 4-5 hours for local development, 8-10 hours for production deployment
Architecture Overview
Here's the technical architecture that makes reliable multi-agent coordination possible:
Core Architecture Components
100%
Rendering diagram...
The system creates a central coordination engine that manages agent communication, dependency resolution, and conflict detection. Agents never communicate directly—everything flows through the event bus for complete observability.
Event-Driven Coordination Framework
The foundation of multi-agent coordination is treating all communication as immutable events. No shared state, no direct agent communication—just events that create an auditable coordination trail.
Core Event System
PYTHON
1# orchestration/event_system.py
2import asyncio
3import json
4import uuid
5from datetime import datetime
6from typing import Dict, List, Any, Callable, Optional
7from dataclasses import dataclass, field
8from enum import Enum
9import logging
10
11class EventType(Enum):
12 AGENT_STARTED = "agent_started"
13 AGENT_COMPLETED = "agent_completed"
14 AGENT_FAILED = "agent_failed"
15 STATE_UPDATED = "state_updated"
16 RESOURCE_REQUESTED = "resource_requested"
17 RESOURCE_ACQUIRED = "resource_acquired"
18 RESOURCE_RELEASED = "resource_released"
19 CONFLICT_DETECTED = "conflict_detected"
20 COORDINATION_REQUIRED = "coordination_required"
21
22@dataclass
23class Event:
24 """Immutable event for agent coordination"""
25 event_id: str = field(default_factory=lambda: str(uuid.uuid4()))
26 event_type: EventType = None
27 agent_id: str = None
28 timestamp: datetime = field(default_factory=datetime.utcnow)
29 payload: Dict[str, Any] = field(default_factory=dict)
30 metadata: Dict[str, Any] = field(default_factory=dict)
31
32 def to_dict(self) -> Dict[str, Any]:
33 return {
34 "event_id": self.event_id,
35 "event_type": self.event_type.value if self.event_type else None,
36 "agent_id": self.agent_id,
37 "timestamp": self.timestamp.isoformat(),
38 "payload": self.payload,
39 "metadata": self.metadata
40 }
41
42class EventBus:
43 """Central event bus for multi-agent coordination"""
44
45 def __init__(self):
46 self.subscribers: Dict[EventType, List[Callable]] = {}
47 self.event_history: List[Event] = []
48 self.active_subscriptions = set()
49 self._lock = asyncio.Lock()
50
51 async def publish(self, event: Event):
52 """Publish event to all subscribers"""
53 async with self._lock:
54 self.event_history.append(event)
55
56 # Log event
57 logging.info(f"Event published: {event.event_type.value} from {event.agent_id}")
58
59 # Notify subscribers
60 if event.event_type in self.subscribers:
61 tasks = []
62 for handler in self.subscribers[event.event_type]:
63 task = asyncio.create_task(handler(event))
64 tasks.append(task)
65
66 # Wait for all handlers to complete
67 if tasks:
68 await asyncio.gather(*tasks, return_exceptions=True)
69
70 async def subscribe(self, event_type: EventType, handler: Callable[[Event], None]) -> str:
71 """Subscribe to specific event type"""
72 subscription_id = str(uuid.uuid4())
73
74 if event_type not in self.subscribers:
75 self.subscribers[event_type] = []
76
77 self.subscribers[event_type].append(handler)
78 self.active_subscriptions.add(subscription_id)
79
80 logging.info(f"New subscription {subscription_id} for {event_type.value}")
81 return subscription_id
82
83 async def get_events_since(self, timestamp: datetime) -> List[Event]:
84 """Get all events since specified timestamp"""
85 return [e for e in self.event_history if e.timestamp >= timestamp]
86
87 async def get_events_by_agent(self, agent_id: str) -> List[Event]:
88 """Get all events from specific agent"""
89 return [e for e in self.event_history if e.agent_id == agent_id]
90
91# Shared State Management
92class SharedStateManager:
93 """Immutable state management for multi-agent coordination"""
94
95 def __init__(self, event_bus: EventBus):
96 self.event_bus = event_bus
97 self.state_snapshots: List[Dict[str, Any]] = []
98 self.current_state: Dict[str, Any] = {}
99 self.state_lock = asyncio.Lock()
100
101 async def update_state(self, agent_id: str, state_updates: Dict[str, Any],
102 merge_strategy: str = "merge") -> Dict[str, Any]:
103 """Update shared state with conflict detection"""
104
105 async with self.state_lock:
106 # Create new state snapshot
107 previous_state = self.current_state.copy()
108
109 if merge_strategy == "merge":
110 new_state = self._merge_states(previous_state, state_updates)
111 elif merge_strategy == "replace":
112 new_state = state_updates.copy()
113 elif merge_strategy == "append":
114 new_state = self._append_to_state(previous_state, state_updates)
115 else:
116 raise ValueError(f"Unknown merge strategy: {merge_strategy}")
117
118 # Check for conflicts
119 conflicts = self._detect_conflicts(previous_state, new_state, agent_id)
120
121 if conflicts:
122 # Publish conflict event
123 conflict_event = Event(
124 event_type=EventType.CONFLICT_DETECTED,
125 agent_id=agent_id,
126 payload={
127 "conflicts": conflicts,
128 "attempted_updates": state_updates,
129 "previous_state": previous_state
130 }
131 )
132 await self.event_bus.publish(conflict_event)
133
134 # Return previous state (don't apply conflicting update)
135 return previous_state
136
137 # Apply state update
138 self.current_state = new_state
139 self.state_snapshots.append({
140 "timestamp": datetime.utcnow().isoformat(),
141 "agent_id": agent_id,
142 "state": new_state.copy(),
143 "updates_applied": state_updates
144 })
145
146 # Publish state update event
147 update_event = Event(
148 event_type=EventType.STATE_UPDATED,
149 agent_id=agent_id,
150 payload={
151 "state_updates": state_updates,
152 "new_state_size": len(new_state),
153 "merge_strategy": merge_strategy
154 }
155 )
156 await self.event_bus.publish(update_event)
157
158 return new_state
159
160 def _detect_conflicts(self, previous_state: Dict[str, Any],
161 new_state: Dict[str, Any], agent_id: str) -> List[Dict[str, Any]]:
162 """Detect state conflicts between agents"""
163
164 conflicts = []
165
166 # Check for concurrent modifications
167 recent_updates = self._get_recent_state_updates(minutes=1)
168
169 for update in recent_updates:
170 if update["agent_id"] != agent_id:
171 # Check if both agents modified same keys
172 their_keys = set(update["updates_applied"].keys())
173 our_keys = set(new_state.keys()) - set(previous_state.keys())
174
175 overlapping_keys = their_keys.intersection(our_keys)
176
177 if overlapping_keys:
178 conflicts.append({
179 "type": "concurrent_modification",
180 "conflicting_agent": update["agent_id"],
181 "conflicting_keys": list(overlapping_keys),
182 "their_timestamp": update["timestamp"],
183 "our_agent": agent_id
184 })
185
186 # Check for resource conflicts
187 resource_conflicts = self._check_resource_conflicts(new_state)
188 conflicts.extend(resource_conflicts)
189
190 return conflicts
191
192 def _merge_states(self, base_state: Dict[str, Any], updates: Dict[str, Any]) -> Dict[str, Any]:
193 """Intelligently merge state updates"""
194
195 merged = base_state.copy()
196
197 for key, value in updates.items():
198 if key in merged:
199 # Handle different merge strategies for different data types
200 if isinstance(merged[key], dict) and isinstance(value, dict):
201 merged[key] = {**merged[key], **value}
202 elif isinstance(merged[key], list) and isinstance(value, list):
203 merged[key] = merged[key] + value
204 else:
205 # Simple replacement for primitive types
206 merged[key] = value
207 else:
208 merged[key] = value
209
210 return merged
211
212 def _get_recent_state_updates(self, minutes: int) -> List[Dict[str, Any]]:
213 """Get state updates from recent time window"""
214
215 cutoff_time = datetime.utcnow().replace(minute=datetime.utcnow().minute - minutes)
216 cutoff_iso = cutoff_time.isoformat()
217
218 return [
219 snapshot for snapshot in self.state_snapshots
220 if snapshot["timestamp"] >= cutoff_iso
221 ]The EventBus handles all agent communication through immutable events. When the infrastructure agent completes AWS resource creation, it publishes an AGENT_COMPLETED event. The content agent subscribes to this event and starts uploading files only after infrastructure is ready.
The SharedStateManager provides conflict-free state updates through versioning and conflict detection. If two agents try to modify the same state simultaneously, the system detects the conflict and triggers resolution—no silent data corruption.
Agent Base Classes and Interfaces
Every agent in the system follows a standardized interface that enables reliable coordination while maintaining specialized functionality.
Specialized Agent Framework
PYTHON
1# agents/base_agent.py
2import asyncio
3from abc import ABC, abstractmethod
4from typing import Dict, Any, List, Optional
5from datetime import datetime
6import logging
7
8class AgentInterface(ABC):
9 """Base interface for all coordinated agents"""
10
11 def __init__(self, agent_id: str, event_bus: EventBus, state_manager: SharedStateManager):
12 self.agent_id = agent_id
13 self.event_bus = event_bus
14 self.state_manager = state_manager
15 self.status = "initialized"
16 self.capabilities = []
17 self.dependencies = []
18 self.start_time = None
19 self.completion_time = None
20
21 @abstractmethod
22 async def execute(self, task_context: Dict[str, Any]) -> Dict[str, Any]:
23 """Main agent execution method"""
24 pass
25
26 @abstractmethod
27 async def validate_preconditions(self, context: Dict[str, Any]) -> bool:
28 """Check if agent can execute given current context"""
29 pass
30
31 @abstractmethod
32 async def cleanup(self):
33 """Cleanup resources when agent completes"""
34 pass
35
36 async def start_execution(self, task_context: Dict[str, Any]) -> Dict[str, Any]:
37 """Coordinated agent execution with event publishing"""
38
39 self.start_time = datetime.utcnow()
40 self.status = "running"
41
42 # Publish start event
43 start_event = Event(
44 event_type=EventType.AGENT_STARTED,
45 agent_id=self.agent_id,
46 payload={
47 "capabilities": self.capabilities,
48 "dependencies": self.dependencies,
49 "task_context": task_context
50 }
51 )
52 await self.event_bus.publish(start_event)
53
54 try:
55 # Validate preconditions
56 if not await self.validate_preconditions(task_context):
57 raise ValueError(f"Preconditions not met for agent {self.agent_id}")
58
59 # Execute agent logic
60 result = await self.execute(task_context)
61
62 # Update shared state with results
63 state_updates = self._create_state_updates(result)
64 if state_updates:
65 await self.state_manager.update_state(self.agent_id, state_updates)
66
67 self.status = "completed"
68 self.completion_time = datetime.utcnow()
69
70 # Publish completion event
71 completion_event = Event(
72 event_type=EventType.AGENT_COMPLETED,
73 agent_id=self.agent_id,
74 payload={
75 "result": result,
76 "execution_time_seconds": (self.completion_time - self.start_time).total_seconds(),
77 "state_updates": state_updates
78 }
79 )
80 await self.event_bus.publish(completion_event)
81
82 return result
83
84 except Exception as e:
85 self.status = "failed"
86
87 # Publish failure event
88 failure_event = Event(
89 event_type=EventType.AGENT_FAILED,
90 agent_id=self.agent_id,
91 payload={
92 "error": str(e),
93 "error_type": type(e).__name__,
94 "task_context": task_context
95 },
96 metadata={"traceback": str(e)}
97 )
98 await self.event_bus.publish(failure_event)
99
100 raise e
101
102 finally:
103 await self.cleanup()
104
105 def _create_state_updates(self, result: Dict[str, Any]) -> Dict[str, Any]:
106 """Convert agent result to state updates"""
107
108 # Default implementation - agents can override
109 return {
110 f"{self.agent_id}_result": result,
111 f"{self.agent_id}_completed_at": datetime.utcnow().isoformat(),
112 f"{self.agent_id}_status": self.status
113 }
114
115# Concrete Agent Implementations
116class InfrastructureAgent(AgentInterface):
117 """Agent specialized in infrastructure deployment"""
118
119 def __init__(self, agent_id: str, event_bus: EventBus, state_manager: SharedStateManager):
120 super().__init__(agent_id, event_bus, state_manager)
121 self.capabilities = [
122 "aws_resource_creation",
123 "cloudfront_deployment",
124 "route53_configuration",
125 "ssl_certificate_management"
126 ]
127 self.dependencies = [] # No dependencies - can start immediately
128
129 async def execute(self, task_context: Dict[str, Any]) -> Dict[str, Any]:
130 """Deploy infrastructure components"""
131
132 domain_name = task_context.get("domain_name")
133 if not domain_name:
134 raise ValueError("domain_name required for infrastructure deployment")
135
136 # Parallel infrastructure tasks
137 tasks = []
138
139 # CloudFront distribution
140 tasks.append(self._create_cloudfront_distribution(domain_name))
141
142 # SSL certificate (can happen in parallel)
143 tasks.append(self._request_ssl_certificate(domain_name))
144
145 # S3 bucket
146 tasks.append(self._create_s3_bucket(domain_name))
147
148 # Wait for all infrastructure components
149 cloudfront_result, ssl_result, s3_result = await asyncio.gather(*tasks)
150
151 # Configure Route53 (depends on CloudFront being ready)
152 route53_result = await self._configure_route53(domain_name, cloudfront_result["domain_name"])
153
154 return {
155 "cloudfront": cloudfront_result,
156 "ssl_certificate": ssl_result,
157 "s3_bucket": s3_result,
158 "route53": route53_result,
159 "infrastructure_ready": True
160 }
161
162 async def validate_preconditions(self, context: Dict[str, Any]) -> bool:
163 """Validate infrastructure deployment preconditions"""
164
165 # Check AWS credentials
166 if not self._check_aws_credentials():
167 return False
168
169 # Verify domain ownership
170 domain_name = context.get("domain_name")
171 if not domain_name or not self._verify_domain_ownership(domain_name):
172 return False
173
174 return True
175
176 async def _create_cloudfront_distribution(self, domain_name: str) -> Dict[str, Any]:
177 """Create CloudFront distribution"""
178
179 # Simulated CloudFront creation
180 await asyncio.sleep(2) # Simulate API call delay
181
182 return {
183 "distribution_id": f"E{uuid.uuid4().hex[:12].upper()}",
184 "domain_name": f"{domain_name}.cloudfront.net",
185 "status": "deployed",
186 "created_at": datetime.utcnow().isoformat()
187 }
188
189 async def _request_ssl_certificate(self, domain_name: str) -> Dict[str, Any]:
190 """Request SSL certificate from ACM"""
191
192 await asyncio.sleep(1) # Simulate certificate request
193
194 return {
195 "certificate_arn": f"arn:aws:acm:us-east-1:123456789:certificate/{uuid.uuid4()}",
196 "domain_name": domain_name,
197 "status": "issued",
198 "validation_method": "dns"
199 }
200
201class ContentAgent(AgentInterface):
202 """Agent specialized in content management and migration"""
203
204 def __init__(self, agent_id: str, event_bus: EventBus, state_manager: SharedStateManager):
205 super().__init__(agent_id, event_bus, state_manager)
206 self.capabilities = [
207 "content_migration",
208 "image_optimization",
209 "metadata_extraction",
210 "content_validation"
211 ]
212 self.dependencies = ["infrastructure_agent"] # Needs S3 bucket
213
214 async def execute(self, task_context: Dict[str, Any]) -> Dict[str, Any]:
215 """Migrate and optimize content"""
216
217 content_sources = task_context.get("content_sources", [])
218 target_bucket = await self._wait_for_infrastructure()
219
220 # Process content in parallel
221 migration_tasks = []
222 for source in content_sources:
223 task = self._migrate_content_source(source, target_bucket)
224 migration_tasks.append(task)
225
226 # Wait for all content migration to complete
227 migration_results = await asyncio.gather(*migration_tasks)
228
229 # Generate content index
230 content_index = self._generate_content_index(migration_results)
231
232 return {
233 "migrated_files": len(migration_results),
234 "content_index": content_index,
235 "optimization_summary": self._summarize_optimizations(migration_results),
236 "content_ready": True
237 }
238
239 async def validate_preconditions(self, context: Dict[str, Any]) -> bool:
240 """Validate content migration preconditions"""
241
242 content_sources = context.get("content_sources", [])
243 if not content_sources:
244 return False
245
246 # Check if content sources are accessible
247 for source in content_sources:
248 if not await self._verify_content_source_access(source):
249 return False
250
251 return True
252
253 async def _wait_for_infrastructure(self) -> str:
254 """Wait for infrastructure agent to complete S3 bucket creation"""
255
256 # Subscribe to infrastructure completion events
257 infrastructure_ready = False
258 s3_bucket = None
259
260 async def infrastructure_handler(event: Event):
261 nonlocal infrastructure_ready, s3_bucket
262 if event.agent_id == "infrastructure_agent" and event.event_type == EventType.AGENT_COMPLETED:
263 s3_bucket = event.payload["result"]["s3_bucket"]["bucket_name"]
264 infrastructure_ready = True
265
266 await self.event_bus.subscribe(EventType.AGENT_COMPLETED, infrastructure_handler)
267
268 # Wait for infrastructure to be ready
269 while not infrastructure_ready:
270 await asyncio.sleep(0.1)
271
272 return s3_bucket
273
274class QualityAssuranceAgent(AgentInterface):
275 """Agent specialized in quality validation and testing"""
276
277 def __init__(self, agent_id: str, event_bus: EventBus, state_manager: SharedStateManager):
278 super().__init__(agent_id, event_bus, state_manager)
279 self.capabilities = [
280 "endpoint_testing",
281 "content_validation",
282 "security_scanning",
283 "performance_testing"
284 ]
285 self.dependencies = ["infrastructure_agent", "content_agent"]
286
287 async def execute(self, task_context: Dict[str, Any]) -> Dict[str, Any]:
288 """Validate system quality and functionality"""
289
290 # Wait for dependencies
291 await self._wait_for_dependencies()
292
293 # Get current system state
294 current_state = self.state_manager.current_state
295
296 # Run quality checks in parallel
297 check_tasks = [
298 self._validate_infrastructure(current_state),
299 self._validate_content_integrity(current_state),
300 self._run_security_scan(current_state),
301 self._test_performance(current_state)
302 ]
303
304 check_results = await asyncio.gather(*check_tasks, return_exceptions=True)
305
306 # Aggregate quality results
307 quality_report = self._generate_quality_report(check_results)
308
309 return {
310 "quality_score": quality_report["overall_score"],
311 "checks_passed": quality_report["passed_checks"],
312 "issues_found": quality_report["issues"],
313 "validation_complete": True
314 }
315
316 async def validate_preconditions(self, context: Dict[str, Any]) -> bool:
317 """QA agent can always run - validates system state"""
318 return True
319
320 async def _wait_for_dependencies(self):
321 """Wait for infrastructure and content agents to complete"""
322
323 completed_agents = set()
324
325 async def completion_handler(event: Event):
326 if event.agent_id in self.dependencies and event.event_type == EventType.AGENT_COMPLETED:
327 completed_agents.add(event.agent_id)
328
329 await self.event_bus.subscribe(EventType.AGENT_COMPLETED, completion_handler)
330
331 # Wait for all dependencies
332 while len(completed_agents) < len(self.dependencies):
333 await asyncio.sleep(0.1)The AgentInterface base class handles all the coordination complexity—event publishing, state management, error handling, and cleanup. Individual agents just implement their core functionality.
Notice how the InfrastructureAgent runs CloudFront, SSL, and S3 creation in parallel using asyncio.gather(). The ContentAgent waits for infrastructure completion by subscribing to events, not by polling state. The QualityAssuranceAgent depends on both infrastructure and content, so it automatically waits for both.
This design pattern lets you build complex agent workflows without writing coordination logic in every agent.
Orchestration Engine
The coordination engine is where dependency resolution, conflict management, and parallel execution come together into a cohesive system.
Coordination and Conflict Resolution
PYTHON
1# orchestration/coordination_engine.py
2import asyncio
3from typing import Dict, List, Any, Set
4from datetime import datetime, timedelta
5import logging
6
7class CoordinationEngine:
8 """Central coordination engine for multi-agent systems"""
9
10 def __init__(self, event_bus: EventBus, state_manager: SharedStateManager):
11 self.event_bus = event_bus
12 self.state_manager = state_manager
13 self.active_agents: Dict[str, AgentInterface] = {}
14 self.dependency_graph = {}
15 self.coordination_policies = {}
16 self.resource_locks = {}
17
18 async def orchestrate_agents(self, agents: List[AgentInterface],
19 task_context: Dict[str, Any]) -> Dict[str, Any]:
20 """Orchestrate multiple agents with dependency management"""
21
22 # Build dependency graph
23 self.dependency_graph = self._build_dependency_graph(agents)
24
25 # Register agents
26 for agent in agents:
27 self.active_agents[agent.agent_id] = agent
28
29 # Subscribe to coordination events
30 await self._setup_coordination_handlers()
31
32 # Determine execution order based on dependencies
33 execution_groups = self._group_agents_by_dependencies(agents)
34
35 # Execute agents group by group
36 all_results = {}
37
38 for group_index, agent_group in enumerate(execution_groups):
39 logging.info(f"Starting execution group {group_index + 1}: {[a.agent_id for a in agent_group]}")
40
41 # Execute agents in group concurrently
42 group_tasks = []
43 for agent in agent_group:
44 task = agent.start_execution(task_context)
45 group_tasks.append((agent.agent_id, task))
46
47 # Wait for group completion
48 group_results = await self._execute_agent_group(group_tasks)
49 all_results.update(group_results)
50
51 # Update task context with results for next group
52 task_context["previous_results"] = all_results
53
54 return all_results
55
56 def _build_dependency_graph(self, agents: List[AgentInterface]) -> Dict[str, Set[str]]:
57 """Build dependency graph from agent dependencies"""
58
59 graph = {}
60
61 for agent in agents:
62 graph[agent.agent_id] = set()
63
64 for dependency in agent.dependencies:
65 # Find dependency by agent ID or capability
66 for other_agent in agents:
67 if (other_agent.agent_id == dependency or
68 dependency in other_agent.capabilities):
69 graph[agent.agent_id].add(other_agent.agent_id)
70
71 return graph
72
73 def _group_agents_by_dependencies(self, agents: List[AgentInterface]) -> List[List[AgentInterface]]:
74 """Group agents into execution order based on dependencies"""
75
76 agent_map = {agent.agent_id: agent for agent in agents}
77 execution_groups = []
78 remaining_agents = set(agent.agent_id for agent in agents)
79 completed_agents = set()
80
81 while remaining_agents:
82 # Find agents with no unfulfilled dependencies
83 ready_agents = []
84
85 for agent_id in remaining_agents:
86 dependencies = self.dependency_graph.get(agent_id, set())
87 unfulfilled = dependencies - completed_agents
88
89 if not unfulfilled:
90 ready_agents.append(agent_map[agent_id])
91
92 if not ready_agents:
93 raise ValueError("Circular dependency detected in agent graph")
94
95 # Add ready agents to current execution group
96 execution_groups.append(ready_agents)
97
98 # Mark agents as completed and remove from remaining
99 for agent in ready_agents:
100 completed_agents.add(agent.agent_id)
101 remaining_agents.remove(agent.agent_id)
102
103 return execution_groups
104
105 async def _execute_agent_group(self, group_tasks: List[tuple]) -> Dict[str, Any]:
106 """Execute a group of agents concurrently with coordination"""
107
108 results = {}
109
110 # Start all agents in group
111 running_tasks = {}
112 for agent_id, task in group_tasks:
113 running_tasks[agent_id] = asyncio.create_task(task)
114
115 # Monitor completion and handle coordination
116 while running_tasks:
117 done, pending = await asyncio.wait(
118 running_tasks.values(),
119 return_when=asyncio.FIRST_COMPLETED
120 )
121
122 for completed_task in done:
123 # Find which agent completed
124 completed_agent_id = None
125 for agent_id, task in running_tasks.items():
126 if task == completed_task:
127 completed_agent_id = agent_id
128 break
129
130 if completed_agent_id:
131 try:
132 result = await completed_task
133 results[completed_agent_id] = result
134 logging.info(f"Agent {completed_agent_id} completed successfully")
135
136 except Exception as e:
137 logging.error(f"Agent {completed_agent_id} failed: {e}")
138 results[completed_agent_id] = {"error": str(e)}
139
140 # Remove completed task
141 del running_tasks[completed_agent_id]
142
143 return results
144
145 async def _setup_coordination_handlers(self):
146 """Setup event handlers for agent coordination"""
147
148 # Conflict resolution handler
149 async def conflict_handler(event: Event):
150 await self._resolve_conflict(event)
151
152 # Resource coordination handler
153 async def resource_handler(event: Event):
154 if event.event_type == EventType.RESOURCE_REQUESTED:
155 await self._handle_resource_request(event)
156
157 # Subscribe to coordination events
158 await self.event_bus.subscribe(EventType.CONFLICT_DETECTED, conflict_handler)
159 await self.event_bus.subscribe(EventType.RESOURCE_REQUESTED, resource_handler)
160
161 async def _resolve_conflict(self, conflict_event: Event):
162 """Resolve conflicts between agents"""
163
164 conflicts = conflict_event.payload["conflicts"]
165 conflicting_agent = conflict_event.agent_id
166
167 for conflict in conflicts:
168 if conflict["type"] == "concurrent_modification":
169 # Use timestamp-based resolution
170 await self._resolve_concurrent_modification(conflict, conflicting_agent)
171
172 elif conflict["type"] == "resource_conflict":
173 # Use priority-based resolution
174 await self._resolve_resource_conflict(conflict, conflicting_agent)
175
176 async def _resolve_concurrent_modification(self, conflict: Dict[str, Any], agent_id: str):
177 """Resolve concurrent state modifications"""
178
179 their_agent = conflict["conflicting_agent"]
180 their_timestamp = datetime.fromisoformat(conflict["their_timestamp"])
181 our_timestamp = datetime.utcnow()
182
183 # First-writer-wins policy
184 if their_timestamp < our_timestamp:
185 # Their change wins - notify our agent to retry
186 logging.info(f"Resolving conflict: {their_agent} wins over {agent_id} (timestamp)")
187
188 retry_event = Event(
189 event_type=EventType.COORDINATION_REQUIRED,
190 agent_id=agent_id,
191 payload={
192 "action": "retry_state_update",
193 "reason": "concurrent_modification_conflict",
194 "winner": their_agent
195 }
196 )
197 await self.event_bus.publish(retry_event)
198
199 async def _handle_resource_request(self, event: Event):
200 """Handle resource acquisition requests"""
201
202 resource_name = event.payload["resource_name"]
203 requesting_agent = event.agent_id
204
205 # Check if resource is already locked
206 if resource_name in self.resource_locks:
207 current_owner = self.resource_locks[resource_name]["owner"]
208
209 # Queue the request
210 if "queue" not in self.resource_locks[resource_name]:
211 self.resource_locks[resource_name]["queue"] = []
212
213 self.resource_locks[resource_name]["queue"].append({
214 "agent_id": requesting_agent,
215 "requested_at": datetime.utcnow()
216 })
217
218 logging.info(f"Resource {resource_name} queued for {requesting_agent} (owned by {current_owner})")
219
220 else:
221 # Grant resource immediately
222 self.resource_locks[resource_name] = {
223 "owner": requesting_agent,
224 "acquired_at": datetime.utcnow(),
225 "queue": []
226 }
227
228 # Notify agent of resource acquisition
229 acquired_event = Event(
230 event_type=EventType.RESOURCE_ACQUIRED,
231 agent_id=requesting_agent,
232 payload={
233 "resource_name": resource_name,
234 "acquired_at": datetime.utcnow().isoformat()
235 }
236 )
237 await self.event_bus.publish(acquired_event)
238
239# Resource Management
240class ResourceManager:
241 """Manage shared resources between agents"""
242
243 def __init__(self, coordination_engine: CoordinationEngine):
244 self.coordination_engine = coordination_engine
245 self.resources = {}
246
247 async def request_resource(self, agent_id: str, resource_name: str,
248 resource_type: str = "exclusive") -> bool:
249 """Request exclusive or shared access to resource"""
250
251 request_event = Event(
252 event_type=EventType.RESOURCE_REQUESTED,
253 agent_id=agent_id,
254 payload={
255 "resource_name": resource_name,
256 "resource_type": resource_type,
257 "requested_at": datetime.utcnow().isoformat()
258 }
259 )
260
261 await self.coordination_engine.event_bus.publish(request_event)
262
263 # Wait for resource acquisition
264 acquired = False
265
266 async def acquisition_handler(event: Event):
267 nonlocal acquired
268 if (event.agent_id == agent_id and
269 event.event_type == EventType.RESOURCE_ACQUIRED and
270 event.payload["resource_name"] == resource_name):
271 acquired = True
272
273 await self.coordination_engine.event_bus.subscribe(EventType.RESOURCE_ACQUIRED, acquisition_handler)
274
275 # Wait for acquisition (with timeout)
276 timeout_seconds = 30
277 wait_start = datetime.utcnow()
278
279 while not acquired:
280 if (datetime.utcnow() - wait_start).total_seconds() > timeout_seconds:
281 raise TimeoutError(f"Resource {resource_name} acquisition timeout for {agent_id}")
282
283 await asyncio.sleep(0.1)
284
285 return acquired
286
287 async def release_resource(self, agent_id: str, resource_name: str):
288 """Release resource and handle queue"""
289
290 if resource_name in self.coordination_engine.resource_locks:
291 lock_info = self.coordination_engine.resource_locks[resource_name]
292
293 if lock_info["owner"] == agent_id:
294 # Release resource
295 del self.coordination_engine.resource_locks[resource_name]
296
297 # Process queue
298 if "queue" in lock_info and lock_info["queue"]:
299 next_request = lock_info["queue"].pop(0)
300 next_agent = next_request["agent_id"]
301
302 # Grant to next agent
303 self.coordination_engine.resource_locks[resource_name] = {
304 "owner": next_agent,
305 "acquired_at": datetime.utcnow(),
306 "queue": lock_info["queue"]
307 }
308
309 # Notify next agent
310 acquired_event = Event(
311 event_type=EventType.RESOURCE_ACQUIRED,
312 agent_id=next_agent,
313 payload={
314 "resource_name": resource_name,
315 "acquired_at": datetime.utcnow().isoformat()
316 }
317 )
318 await self.coordination_engine.event_bus.publish(acquired_event)
319
320 # Publish release event
321 release_event = Event(
322 event_type=EventType.RESOURCE_RELEASED,
323 agent_id=agent_id,
324 payload={
325 "resource_name": resource_name,
326 "released_at": datetime.utcnow().isoformat()
327 }
328 )
329 await self.coordination_engine.event_bus.publish(release_event)The CoordinationEngine is the brain of the system. It analyzes agent dependencies, groups them into parallel execution batches, and handles conflicts when they arise.
The key insight is the _group_agents_by_dependencies() method. Instead of running agents randomly or sequentially, it creates execution groups where:
- Group 1: Agents with no dependencies (can run immediately)
- Group 2: Agents that depend only on Group 1 agents
- Group 3: Agents that depend on Groups 1 or 2
- And so on...
This gives you maximum parallelism while respecting dependencies. The ResourceManager provides additional coordination through exclusive resource locking—ensuring two agents can't modify the same AWS resource simultaneously.
Complete Orchestration Example
Here's how all the pieces come together in a real-world deployment scenario.
Real-World Multi-Agent Deployment
PYTHON
1# example/blog_deployment_orchestration.py
2import asyncio
3from typing import Dict, Any
4import logging
5
6# Configure logging
7logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
8
9class BlogDeploymentOrchestrator:
10 """Complete orchestration example for blog deployment"""
11
12 def __init__(self):
13 self.event_bus = EventBus()
14 self.state_manager = SharedStateManager(self.event_bus)
15 self.coordination_engine = CoordinationEngine(self.event_bus, self.state_manager)
16
17 async def deploy_blog_infrastructure(self, deployment_config: Dict[str, Any]) -> Dict[str, Any]:
18 """Deploy complete blog infrastructure with multi-agent coordination"""
19
20 # Create specialized agents
21 agents = [
22 InfrastructureAgent("infrastructure_agent", self.event_bus, self.state_manager),
23 ContentAgent("content_agent", self.event_bus, self.state_manager),
24 SecurityAgent("security_agent", self.event_bus, self.state_manager),
25 QualityAssuranceAgent("qa_agent", self.event_bus, self.state_manager)
26 ]
27
28 # Create task context from deployment config
29 task_context = {
30 "domain_name": deployment_config["domain"],
31 "content_sources": deployment_config["content_sources"],
32 "security_requirements": deployment_config.get("security_requirements", []),
33 "quality_thresholds": deployment_config.get("quality_thresholds", {}),
34 "deployment_id": f"deployment_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}"
35 }
36
37 logging.info(f"Starting blog deployment orchestration for {deployment_config['domain']}")
38
39 try:
40 # Orchestrate all agents
41 results = await self.coordination_engine.orchestrate_agents(agents, task_context)
42
43 # Generate deployment summary
44 deployment_summary = self._generate_deployment_summary(results, task_context)
45
46 return deployment_summary
47
48 except Exception as e:
49 logging.error(f"Deployment orchestration failed: {e}")
50
51 # Generate failure report
52 failure_report = await self._generate_failure_report(e, task_context)
53 raise Exception(f"Deployment failed: {failure_report}")
54
55 def _generate_deployment_summary(self, results: Dict[str, Any],
56 context: Dict[str, Any]) -> Dict[str, Any]:
57 """Generate comprehensive deployment summary"""
58
59 successful_agents = [agent_id for agent_id, result in results.items()
60 if not isinstance(result, dict) or "error" not in result]
61
62 failed_agents = [agent_id for agent_id, result in results.items()
63 if isinstance(result, dict) and "error" in result]
64
65 # Calculate deployment metrics
66 total_time = None
67 if "infrastructure_agent" in results:
68 infra_result = results["infrastructure_agent"]
69 if "execution_time_seconds" in infra_result:
70 total_time = max(
71 result.get("execution_time_seconds", 0)
72 for result in results.values()
73 if isinstance(result, dict) and "execution_time_seconds" in result
74 )
75
76 return {
77 "deployment_id": context["deployment_id"],
78 "domain": context["domain_name"],
79 "status": "success" if not failed_agents else "partial_failure",
80 "successful_agents": successful_agents,
81 "failed_agents": failed_agents,
82 "total_execution_time_seconds": total_time,
83 "infrastructure_deployed": "infrastructure_agent" in successful_agents,
84 "content_migrated": "content_agent" in successful_agents,
85 "security_configured": "security_agent" in successful_agents,
86 "quality_validated": "qa_agent" in successful_agents,
87 "agent_results": results,
88 "deployment_timestamp": datetime.utcnow().isoformat()
89 }
90
91# Security Agent Implementation
92class SecurityAgent(AgentInterface):
93 """Agent specialized in security configuration"""
94
95 def __init__(self, agent_id: str, event_bus: EventBus, state_manager: SharedStateManager):
96 super().__init__(agent_id, event_bus, state_manager)
97 self.capabilities = [
98 "ssl_configuration",
99 "access_control",
100 "security_headers",
101 "vulnerability_scanning"
102 ]
103 self.dependencies = ["infrastructure_agent"]
104
105 async def execute(self, task_context: Dict[str, Any]) -> Dict[str, Any]:
106 """Configure security settings"""
107
108 # Wait for infrastructure
109 await self._wait_for_infrastructure_completion()
110
111 # Security configuration tasks
112 security_tasks = [
113 self._configure_ssl_redirect(),
114 self._setup_security_headers(),
115 self._configure_access_controls(),
116 self._run_vulnerability_scan()
117 ]
118
119 # Execute security configurations
120 security_results = await asyncio.gather(*security_tasks)
121
122 return {
123 "ssl_redirect_configured": security_results[0],
124 "security_headers_set": security_results[1],
125 "access_controls_enabled": security_results[2],
126 "vulnerability_scan_passed": security_results[3],
127 "security_score": self._calculate_security_score(security_results),
128 "security_ready": True
129 }
130
131 async def validate_preconditions(self, context: Dict[str, Any]) -> bool:
132 """Validate security configuration preconditions"""
133 return True # Security can always be configured
134
135 async def _wait_for_infrastructure_completion(self):
136 """Wait for infrastructure agent to complete"""
137 infrastructure_ready = False
138
139 async def infra_handler(event: Event):
140 nonlocal infrastructure_ready
141 if event.agent_id == "infrastructure_agent" and event.event_type == EventType.AGENT_COMPLETED:
142 infrastructure_ready = True
143
144 await self.event_bus.subscribe(EventType.AGENT_COMPLETED, infra_handler)
145
146 while not infrastructure_ready:
147 await asyncio.sleep(0.1)
148
149 async def _configure_ssl_redirect(self) -> bool:
150 """Configure SSL redirect rules"""
151 await asyncio.sleep(0.5) # Simulate configuration
152 return True
153
154 async def _setup_security_headers(self) -> bool:
155 """Setup security headers"""
156 await asyncio.sleep(0.3) # Simulate header configuration
157 return True
158
159 async def _configure_access_controls(self) -> bool:
160 """Configure access control policies"""
161 await asyncio.sleep(0.4) # Simulate access control setup
162 return True
163
164 async def _run_vulnerability_scan(self) -> bool:
165 """Run basic vulnerability scan"""
166 await asyncio.sleep(1.0) # Simulate security scan
167 return True
168
169 def _calculate_security_score(self, results: List[bool]) -> float:
170 """Calculate overall security score"""
171 return sum(results) / len(results)
172
173# Usage Example
174async def main():
175 """Example usage of multi-agent orchestration"""
176
177 orchestrator = BlogDeploymentOrchestrator()
178
179 deployment_config = {
180 "domain": "blogs.briefcasebrain.com",
181 "content_sources": [
182 "/content/blog/post1.md",
183 "/content/blog/post2.md",
184 "/images/blog/"
185 ],
186 "security_requirements": [
187 "ssl_redirect",
188 "security_headers",
189 "access_control"
190 ],
191 "quality_thresholds": {
192 "min_security_score": 0.9,
193 "max_load_time_seconds": 2.0
194 }
195 }
196
197 try:
198 result = await orchestrator.deploy_blog_infrastructure(deployment_config)
199 print("Deployment Summary:")
200 print(json.dumps(result, indent=2))
201
202 except Exception as e:
203 print(f"Deployment failed: {e}")
204
205if __name__ == "__main__":
206 asyncio.run(main())The BlogDeploymentOrchestrator ties everything together. It creates specialized agents (infrastructure, content, security, QA), defines the task context, and lets the coordination engine handle the complexity.
The execution flow:
- Infrastructure agent runs first (no dependencies)
- Content and Security agents run in parallel after infrastructure completes
- QA agent runs last, validating the entire deployment
The deployment summary provides complete visibility into execution timing, success rates, and any failures. This observability is crucial for debugging coordination issues in production.
Monitoring and Observability
Multi-agent coordination requires comprehensive monitoring to debug issues and optimize performance.
Real-Time Orchestration Monitoring
PYTHON
1# monitoring/orchestration_monitor.py
2from typing import Dict, List, Any
3from datetime import datetime, timedelta
4import json
5
6class OrchestrationMonitor:
7 """Monitor multi-agent orchestration performance and health"""
8
9 def __init__(self, event_bus: EventBus):
10 self.event_bus = event_bus
11 self.metrics = {}
12 self.active_orchestrations = {}
13
14 # Subscribe to all events for monitoring
15 asyncio.create_task(self._setup_monitoring())
16
17 async def _setup_monitoring(self):
18 """Setup comprehensive monitoring"""
19
20 # Monitor all event types
21 for event_type in EventType:
22 await self.event_bus.subscribe(event_type, self._record_event)
23
24 async def _record_event(self, event: Event):
25 """Record event for monitoring purposes"""
26
27 event_key = f"{event.event_type.value}_{event.agent_id}"
28
29 if event_key not in self.metrics:
30 self.metrics[event_key] = {
31 "count": 0,
32 "last_seen": None,
33 "average_interval": None
34 }
35
36 # Update metrics
37 current_time = datetime.utcnow()
38 self.metrics[event_key]["count"] += 1
39
40 if self.metrics[event_key]["last_seen"]:
41 interval = (current_time - self.metrics[event_key]["last_seen"]).total_seconds()
42
43 if self.metrics[event_key]["average_interval"]:
44 # Exponential moving average
45 self.metrics[event_key]["average_interval"] = (
46 0.7 * self.metrics[event_key]["average_interval"] + 0.3 * interval
47 )
48 else:
49 self.metrics[event_key]["average_interval"] = interval
50
51 self.metrics[event_key]["last_seen"] = current_time
52
53 # Track orchestration progress
54 if event.event_type == EventType.AGENT_STARTED:
55 orchestration_id = event.metadata.get("orchestration_id", "default")
56 if orchestration_id not in self.active_orchestrations:
57 self.active_orchestrations[orchestration_id] = {
58 "started_at": current_time,
59 "agents": {},
60 "status": "running"
61 }
62
63 self.active_orchestrations[orchestration_id]["agents"][event.agent_id] = {
64 "status": "running",
65 "started_at": current_time
66 }
67
68 elif event.event_type == EventType.AGENT_COMPLETED:
69 orchestration_id = event.metadata.get("orchestration_id", "default")
70 if orchestration_id in self.active_orchestrations:
71 if event.agent_id in self.active_orchestrations[orchestration_id]["agents"]:
72 agent_info = self.active_orchestrations[orchestration_id]["agents"][event.agent_id]
73 agent_info["status"] = "completed"
74 agent_info["completed_at"] = current_time
75 agent_info["execution_time"] = (current_time - agent_info["started_at"]).total_seconds()
76
77 def get_orchestration_metrics(self) -> Dict[str, Any]:
78 """Get current orchestration performance metrics"""
79
80 current_time = datetime.utcnow()
81
82 # Active orchestrations
83 active_count = len([
84 o for o in self.active_orchestrations.values()
85 if o["status"] == "running"
86 ])
87
88 # Agent performance
89 agent_performance = {}
90 for orchestration in self.active_orchestrations.values():
91 for agent_id, agent_info in orchestration["agents"].items():
92 if agent_id not in agent_performance:
93 agent_performance[agent_id] = {
94 "total_executions": 0,
95 "successful_executions": 0,
96 "average_execution_time": 0,
97 "failure_rate": 0
98 }
99
100 agent_performance[agent_id]["total_executions"] += 1
101
102 if agent_info["status"] == "completed":
103 agent_performance[agent_id]["successful_executions"] += 1
104
105 if "execution_time" in agent_info:
106 current_avg = agent_performance[agent_id]["average_execution_time"]
107 new_time = agent_info["execution_time"]
108 total = agent_performance[agent_id]["total_executions"]
109
110 # Running average
111 agent_performance[agent_id]["average_execution_time"] = (
112 (current_avg * (total - 1) + new_time) / total
113 )
114
115 # Calculate failure rates
116 for agent_id, perf in agent_performance.items():
117 if perf["total_executions"] > 0:
118 perf["failure_rate"] = 1 - (perf["successful_executions"] / perf["total_executions"])
119
120 return {
121 "timestamp": current_time.isoformat(),
122 "active_orchestrations": active_count,
123 "total_orchestrations": len(self.active_orchestrations),
124 "agent_performance": agent_performance,
125 "event_metrics": self.metrics,
126 "system_health": self._calculate_system_health()
127 }
128
129 def _calculate_system_health(self) -> str:
130 """Calculate overall system health"""
131
132 recent_failures = 0
133 total_recent_events = 0
134
135 cutoff_time = datetime.utcnow() - timedelta(minutes=5)
136
137 for event in self.event_bus.event_history:
138 if event.timestamp >= cutoff_time:
139 total_recent_events += 1
140 if event.event_type == EventType.AGENT_FAILED:
141 recent_failures += 1
142
143 if total_recent_events == 0:
144 return "idle"
145
146 failure_rate = recent_failures / total_recent_events
147
148 if failure_rate == 0:
149 return "healthy"
150 elif failure_rate < 0.1:
151 return "warning"
152 else:
153 return "critical"
154
155# Prometheus metrics export
156def export_orchestration_metrics(monitor: OrchestrationMonitor) -> str:
157 """Export orchestration metrics in Prometheus format"""
158
159 metrics = monitor.get_orchestration_metrics()
160
161 prometheus_metrics = [
162 "# HELP orchestration_active_count Number of active orchestrations",
163 "# TYPE orchestration_active_count gauge",
164 f"orchestration_active_count {metrics['active_orchestrations']}",
165 "",
166 "# HELP orchestration_agent_execution_time Agent execution time in seconds",
167 "# TYPE orchestration_agent_execution_time histogram"
168 ]
169
170 for agent_id, perf in metrics["agent_performance"].items():
171 prometheus_metrics.extend([
172 f"orchestration_agent_execution_time_sum{{agent_id=\"{agent_id}\"}} {perf['average_execution_time'] * perf['total_executions']}",
173 f"orchestration_agent_execution_time_count{{agent_id=\"{agent_id}\"}} {perf['total_executions']}"
174 ])
175
176 prometheus_metrics.extend([
177 "",
178 "# HELP orchestration_agent_failure_rate Agent failure rate",
179 "# TYPE orchestration_agent_failure_rate gauge"
180 ])
181
182 for agent_id, perf in metrics["agent_performance"].items():
183 prometheus_metrics.append(f"orchestration_agent_failure_rate{{agent_id=\"{agent_id}\"}} {perf['failure_rate']}")
184
185 return "\n".join(prometheus_metrics)The OrchestrationMonitor tracks every event and agent performance metric. Key insights:
- Agent execution times: How long each agent type takes on average
- Failure rates: Which agents fail most often and why
- Coordination efficiency: How much time is spent waiting vs. executing
- Resource contention: Which resources cause bottlenecks
The Prometheus metrics export enables integration with existing monitoring infrastructure. Alerts trigger when agent failure rates spike or coordination times exceed thresholds.
This monitoring is essential because multi-agent coordination failures can be subtle—an agent might succeed but take 10x longer due to resource contention.
Testing Framework
You can't deploy multi-agent coordination without comprehensive tests that validate parallel execution, dependency ordering, and failure handling.
Integration Tests for Multi-Agent Systems
PYTHON
1# tests/test_multi_agent_orchestration.py
2import pytest
3import asyncio
4from datetime import datetime
5
6class TestMultiAgentOrchestration:
7
8 @pytest.fixture
9 async def setup_orchestration(self):
10 """Setup test orchestration environment"""
11 event_bus = EventBus()
12 state_manager = SharedStateManager(event_bus)
13 coordination_engine = CoordinationEngine(event_bus, state_manager)
14
15 yield event_bus, state_manager, coordination_engine
16
17 @pytest.mark.asyncio
18 async def test_parallel_agent_execution(self, setup_orchestration):
19 """Test agents execute in parallel when no dependencies exist"""
20 event_bus, state_manager, coordination_engine = setup_orchestration
21
22 # Create independent agents (no dependencies)
23 agents = [
24 TestAgent("agent1", event_bus, state_manager, execution_time=1.0),
25 TestAgent("agent2", event_bus, state_manager, execution_time=1.0),
26 TestAgent("agent3", event_bus, state_manager, execution_time=1.0)
27 ]
28
29 start_time = datetime.utcnow()
30
31 # Execute agents
32 results = await coordination_engine.orchestrate_agents(agents, {})
33
34 end_time = datetime.utcnow()
35 total_time = (end_time - start_time).total_seconds()
36
37 # Should complete in ~1 second (parallel), not ~3 seconds (sequential)
38 assert total_time < 2.0
39 assert len(results) == 3
40 assert all(agent_id in results for agent_id in ["agent1", "agent2", "agent3"])
41
42 @pytest.mark.asyncio
43 async def test_dependency_ordering(self, setup_orchestration):
44 """Test agents execute in correct dependency order"""
45 event_bus, state_manager, coordination_engine = setup_orchestration
46
47 # Create agents with dependencies
48 agent1 = TestAgent("agent1", event_bus, state_manager) # No dependencies
49 agent2 = TestAgent("agent2", event_bus, state_manager) # No dependencies
50 agent3 = TestAgent("agent3", event_bus, state_manager, dependencies=["agent1", "agent2"])
51
52 agents = [agent3, agent1, agent2] # Deliberately wrong order
53
54 execution_order = []
55
56 async def track_execution(event: Event):
57 if event.event_type == EventType.AGENT_STARTED:
58 execution_order.append(event.agent_id)
59
60 await event_bus.subscribe(EventType.AGENT_STARTED, track_execution)
61
62 # Execute agents
63 results = await coordination_engine.orchestrate_agents(agents, {})
64
65 # Verify dependency order
66 assert len(execution_order) == 3
67 assert execution_order.index("agent3") > execution_order.index("agent1")
68 assert execution_order.index("agent3") > execution_order.index("agent2")
69
70 @pytest.mark.asyncio
71 async def test_conflict_resolution(self, setup_orchestration):
72 """Test state conflict detection and resolution"""
73 event_bus, state_manager, coordination_engine = setup_orchestration
74
75 # Create agents that will conflict
76 agent1 = ConflictingAgent("agent1", event_bus, state_manager,
77 state_key="shared_resource", state_value="value1")
78 agent2 = ConflictingAgent("agent2", event_bus, state_manager,
79 state_key="shared_resource", state_value="value2")
80
81 agents = [agent1, agent2]
82
83 conflicts_detected = []
84
85 async def conflict_handler(event: Event):
86 conflicts_detected.append(event.payload["conflicts"])
87
88 await event_bus.subscribe(EventType.CONFLICT_DETECTED, conflict_handler)
89
90 # Execute conflicting agents
91 results = await coordination_engine.orchestrate_agents(agents, {})
92
93 # Verify conflict was detected
94 assert len(conflicts_detected) > 0
95
96 @pytest.mark.asyncio
97 async def test_agent_failure_handling(self, setup_orchestration):
98 """Test system handles agent failures gracefully"""
99 event_bus, state_manager, coordination_engine = setup_orchestration
100
101 # Create agents with one that will fail
102 agent1 = TestAgent("agent1", event_bus, state_manager)
103 agent2 = FailingAgent("agent2", event_bus, state_manager)
104 agent3 = TestAgent("agent3", event_bus, state_manager, dependencies=["agent1"])
105
106 agents = [agent1, agent2, agent3]
107
108 failures_detected = []
109
110 async def failure_handler(event: Event):
111 failures_detected.append(event.agent_id)
112
113 await event_bus.subscribe(EventType.AGENT_FAILED, failure_handler)
114
115 # Execute agents (should handle failure gracefully)
116 results = await coordination_engine.orchestrate_agents(agents, {})
117
118 # Verify failure was captured
119 assert "agent2" in failures_detected
120 assert "agent1" in results # Should succeed
121 assert "agent3" in results # Should succeed (doesn't depend on failed agent)
122
123# Test helper classes
124class TestAgent(AgentInterface):
125 """Simple test agent for orchestration testing"""
126
127 def __init__(self, agent_id: str, event_bus: EventBus, state_manager: SharedStateManager,
128 execution_time: float = 0.1, dependencies: List[str] = None):
129 super().__init__(agent_id, event_bus, state_manager)
130 self.execution_time = execution_time
131 self.dependencies = dependencies or []
132
133 async def execute(self, task_context: Dict[str, Any]) -> Dict[str, Any]:
134 await asyncio.sleep(self.execution_time)
135 return {"status": "completed", "agent_id": self.agent_id}
136
137 async def validate_preconditions(self, context: Dict[str, Any]) -> bool:
138 return True
139
140 async def cleanup(self):
141 pass
142
143class ConflictingAgent(AgentInterface):
144 """Test agent that creates state conflicts"""
145
146 def __init__(self, agent_id: str, event_bus: EventBus, state_manager: SharedStateManager,
147 state_key: str, state_value: Any):
148 super().__init__(agent_id, event_bus, state_manager)
149 self.state_key = state_key
150 self.state_value = state_value
151
152 async def execute(self, task_context: Dict[str, Any]) -> Dict[str, Any]:
153 # Try to update shared state (will conflict)
154 await self.state_manager.update_state(
155 self.agent_id,
156 {self.state_key: self.state_value}
157 )
158 return {"status": "completed"}
159
160 async def validate_preconditions(self, context: Dict[str, Any]) -> bool:
161 return True
162
163 async def cleanup(self):
164 pass
165
166class FailingAgent(AgentInterface):
167 """Test agent that always fails"""
168
169 def __init__(self, agent_id: str, event_bus: EventBus, state_manager: SharedStateManager):
170 super().__init__(agent_id, event_bus, state_manager)
171
172 async def execute(self, task_context: Dict[str, Any]) -> Dict[str, Any]:
173 raise Exception("Intentional test failure")
174
175 async def validate_preconditions(self, context: Dict[str, Any]) -> bool:
176 return True
177
178 async def cleanup(self):
179 passThe test suite covers the four critical scenarios for multi-agent systems:
- Parallel execution: Independent agents should run simultaneously, not sequentially
- Dependency ordering: Agents with dependencies should wait for their requirements
- Conflict resolution: The system should detect and resolve state conflicts
- Failure handling: One agent's failure shouldn't crash the entire coordination
The key insight in the parallel execution test: we verify that 3 agents each taking 1 second complete in ~1 second total, not ~3 seconds. This proves true parallel execution.
The dependency ordering test ensures agents execute in the right order even when created in the wrong order—the system automatically figures out the correct sequence.
Deployment Configuration
Running multi-agent coordination in production requires careful resource management and configuration.
Production Kubernetes Deployment
YAML
1# k8s/multi-agent-orchestration.yaml
2apiVersion: apps/v1
3kind: Deployment
4metadata:
5 name: multi-agent-orchestrator
6 namespace: agent-coordination
7spec:
8 replicas: 2
9 strategy:
10 type: RollingUpdate
11 rollingUpdate:
12 maxUnavailable: 1
13 maxSurge: 1
14 selector:
15 matchLabels:
16 app: multi-agent-orchestrator
17 template:
18 metadata:
19 labels:
20 app: multi-agent-orchestrator
21 spec:
22 containers:
23 - name: orchestrator
24 image: briefcase-ai/multi-agent-orchestrator:latest
25 ports:
26 - containerPort: 8080
27 name: http
28 - containerPort: 9090
29 name: metrics
30 env:
31 - name: REDIS_URL
32 valueFrom:
33 secretKeyRef:
34 name: redis-credentials
35 key: url
36 - name: EVENT_STORE_BACKEND
37 value: "redis"
38 - name: MAX_CONCURRENT_AGENTS
39 value: "10"
40 - name: COORDINATION_TIMEOUT_SECONDS
41 value: "300"
42 resources:
43 requests:
44 memory: "1Gi"
45 cpu: "500m"
46 limits:
47 memory: "2Gi"
48 cpu: "1000m"
49 livenessProbe:
50 httpGet:
51 path: /health
52 port: 8080
53 initialDelaySeconds: 30
54 periodSeconds: 10
55 readinessProbe:
56 httpGet:
57 path: /ready
58 port: 8080
59 initialDelaySeconds: 5
60 periodSeconds: 5
61 volumeMounts:
62 - name: agent-config
63 mountPath: /app/config
64 readOnly: true
65 volumes:
66 - name: agent-config
67 configMap:
68 name: agent-configuration
69
70---
71apiVersion: v1
72kind: ConfigMap
73metadata:
74 name: agent-configuration
75 namespace: agent-coordination
76data:
77 agent_policies.yaml: |
78 coordination_policies:
79 default:
80 max_execution_time_seconds: 300
81 conflict_resolution_strategy: "timestamp_based"
82 resource_timeout_seconds: 30
83
84 high_priority:
85 max_execution_time_seconds: 600
86 conflict_resolution_strategy: "priority_based"
87 resource_timeout_seconds: 60
88
89 agent_capabilities:
90 infrastructure_agent:
91 - aws_resource_creation
92 - cloudfront_deployment
93 - route53_configuration
94
95 content_agent:
96 - content_migration
97 - image_optimization
98 - metadata_extraction
99
100 security_agent:
101 - ssl_configuration
102 - access_control
103 - vulnerability_scanning
104
105---
106apiVersion: v1
107kind: Service
108metadata:
109 name: orchestrator-service
110 namespace: agent-coordination
111spec:
112 selector:
113 app: multi-agent-orchestrator
114 ports:
115 - name: http
116 protocol: TCP
117 port: 80
118 targetPort: 8080
119 - name: metrics
120 protocol: TCP
121 port: 9090
122 targetPort: 9090
123 type: ClusterIPThe Kubernetes deployment includes several critical configuration decisions:
- Redis backend: For event persistence and cross-replica coordination
- Resource limits: Prevent runaway agents from consuming all cluster resources
- Health checks: Ensure coordination engine is responsive and not deadlocked
- ConfigMap policies: Define coordination behavior without code changes
The key insight is treating the orchestration engine like critical infrastructure—multiple replicas, proper monitoring, and graceful degradation when individual replicas fail.
What We Learned
Building production multi-agent orchestration taught us several crucial lessons:
1. Events Beat Shared State Every Time
Direct agent communication leads to race conditions, timeouts, and debugging hell. Event-driven coordination provides complete observability and eliminates most coordination bugs.
2. Dependencies Must Be Explicit and Enforced
Implicit agent dependencies (like assuming infrastructure is ready) cause intermittent failures. Making dependencies explicit in code enables automatic ordering and clear error messages.
3. Resource Locking Prevents Silent Failures
Two agents modifying the same AWS resource simultaneously often succeeds initially but creates inconsistent state. Explicit resource locking prevents these subtle bugs.
4. Monitoring Is Essential for Coordination
Multi-agent systems fail in complex ways. Comprehensive monitoring of agent performance, coordination efficiency, and resource contention is crucial for production reliability.
5. Testing Must Validate Parallelism
Unit tests for individual agents don't catch coordination issues. Integration tests must verify that agents actually run in parallel and handle dependencies correctly.
Next Steps
This orchestration system provides the foundation for reliable multi-agent coordination. From here, you can extend it with:
- Priority-based scheduling for high-priority agents
- Circuit breakers to prevent cascading failures
- Dynamic agent registration for runtime agent discovery
- Cross-cluster coordination for distributed agent systems
The key is maintaining the event-driven architecture while adding sophistication. Every coordination decision should remain observable and auditable.
Related Reading
- Documentation That Writes Itself From Your Existing Work — Executive-focused multi-agent coordination results
- Infrastructure That Sets Up Itself While You Focus on Product — Business outcomes from parallel agent coordination
- Secure Site Deployment Without The Coordination Hell — Real-world deployment case study
- Building Git-Style Legal Infrastructure: LakeFS Implementation Guide — Version control infrastructure for agent coordination
Want fewer escalations? See a live trace.
See Briefcase on your stack
Reduce escalations: Catch issues before they hit production with comprehensive observability
Auditability & replay: Complete trace capture for debugging and compliance