`print(f" {msg.timestamp.strftime('%H:%M:%S')} | {msg.sender} → {msg.recipient} | {msg.type.value}") finally: # Stop the bus bus.stop() await bus_task if __name__ == "__main__": asyncio.run(run_content_creation_workflow())`

### Advanced Coordination Patterns

`# coordination_patterns.py from typing import Dict, List, Any, Optional, Callable import asyncio from enum import Enum from dataclasses import dataclass import time class CoordinationPattern(Enum): SEQUENTIAL = "sequential" PARALLEL = "parallel" PIPELINE = "pipeline" CONSENSUS = "consensus" HIERARCHICAL = "hierarchical" class AgentOrchestrator: """Advanced orchestration system for multi-agent coordination""" def __init__(self): self.agents: Dict[str, BaseAgent] = {} self.workflows: Dict[str, 'Workflow'] = {} self.execution_history: List[Dict[str, Any]] = [] async def execute_sequential( self, agents: List[str], initial_input: Any ) -> Any: """Execute agents in sequence, passing output to next""" result = initial_input execution_trace = [] for agent_id in agents: if agent_id not in self.agents: raise ValueError(f"Agent {agent_id} not found") agent = self.agents[agent_id] start_time = time.time() # Execute agent task result = await agent.process(result) execution_trace.append({ 'agent': agent_id, 'duration': time.time() - start_time, 'output_size': len(str(result)) }) self.execution_history.append({ 'pattern': 'sequential', 'agents': agents, 'trace': execution_trace, 'total_time': sum(t['duration'] for t in execution_trace) }) return result async def execute_parallel( self, agents: List[str], input_data: Any, aggregation_fn: Optional[Callable] = None ) -> Any: """Execute agents in parallel and aggregate results""" tasks = [] for agent_id in agents: if agent_id not in self.agents: raise ValueError(f"Agent {agent_id} not found") agent = self.agents[agent_id] task = asyncio.create_task(agent.process(input_data)) tasks.append((agent_id, task)) # Wait for all tasks to complete results = {} for agent_id, task in tasks: try: results[agent_id] = await task except Exception as e: results[agent_id] = {'error': str(e)} # Aggregate results if aggregation_fn: return aggregation_fn(results) return results async def execute_pipeline( self, stages: List[Dict[str, Any]], initial_input: Any ) -> Any: """Execute a multi-stage pipeline with parallel stages""" result = initial_input for stage in stages: if stage['type'] == 'sequential': result = await self.execute_sequential( stage['agents'], result ) elif stage['type'] == 'parallel': result = await self.execute_parallel( stage['agents'], result, stage.get('aggregation_fn') ) elif stage['type'] == 'conditional': condition_result = await self.agents[stage['condition_agent']].process(result) if condition_result.get('condition_met'): result = await self.execute_sequential( stage['true_branch'], result ) else: result = await self.execute_sequential( stage['false_branch'], result ) return result async def execute_consensus( self, agents: List[str], input_data: Any, consensus_threshold: float = 0.6 ) -> Any: """Execute agents and reach consensus on result""" # Get all agent responses responses = await self.execute_parallel(agents, input_data) # Analyze responses for consensus consensus_analyzer = ConsensusAnalyzer(consensus_threshold) consensus_result = consensus_analyzer.analyze(responses) if consensus_result['has_consensus']: return consensus_result['consensus_value'] else: # No consensus - run conflict resolution return await self.resolve_conflict(responses, input_data) async def resolve_conflict( self, responses: Dict[str, Any], original_input: Any ) -> Any: """Resolve conflicts when agents don't reach consensus""" # Use a specialized arbiter agent or voting mechanism if 'arbiter_agent' in self.agents: return await self.agents['arbiter_agent'].process({ 'conflict': responses, 'original_input': original_input }) # Fallback: return most common response from collections import Counter response_values = [str(v) for v in responses.values()] most_common = Counter(response_values).most_common(1)[0][0] return most_common class ConsensusAnalyzer: """Analyzes agent responses for consensus""" def __init__(self, threshold: float = 0.6): self.threshold = threshold def analyze(self, responses: Dict[str, Any]) -> Dict[str, Any]: """Analyze responses for consensus""" from collections import Counter # Convert responses to comparable format response_values = [] for agent_id, response in responses.items(): if isinstance(response, dict): response_str = json.dumps(response, sort_keys=True) else: response_str = str(response) response_values.append(response_str) # Count occurrences counter = Counter(response_values) total = len(response_values) # Check for consensus if total == 0: return {'has_consensus': False, 'consensus_value': None} most_common_value, count = counter.most_common(1)[0] consensus_ratio = count / total return { 'has_consensus': consensus_ratio >= self.threshold, 'consensus_value': json.loads(most_common_value) if most_common_value.startswith('{') else most_common_value, 'consensus_ratio': consensus_ratio, 'distribution': dict(counter) } @dataclass class Workflow: """Defines a reusable workflow pattern""" name: str description: str pattern: CoordinationPattern stages: List[Dict[str, Any]] validators: List[Callable] = None error_handlers: List[Callable] = None async def execute(self, orchestrator: AgentOrchestrator, input_data: Any) -> Any: """Execute the workflow""" try: # Validate input if self.validators: for validator in self.validators: if not validator(input_data): raise ValueError(f"Input validation failed for workflow {self.name}") # Execute based on pattern if self.pattern == CoordinationPattern.SEQUENTIAL: result = await orchestrator.execute_sequential( [s['agent'] for s in self.stages], input_data ) elif self.pattern == CoordinationPattern.PARALLEL: result = await orchestrator.execute_parallel( [s['agent'] for s in self.stages], input_data ) elif self.pattern == CoordinationPattern.PIPELINE: result = await orchestrator.execute_pipeline( self.stages, input_data ) elif self.pattern == CoordinationPattern.CONSENSUS: result = await orchestrator.execute_consensus( [s['agent'] for s in self.stages], input_data ) else: raise ValueError(f"Unsupported pattern: {self.pattern}") return result except Exception as e: if self.error_handlers: for handler in self.error_handlers: handled = await handler(e, input_data) if handled: return handled raise # Example: Book Writing Workflow with Advanced Coordination class BookWritingOrchestrator(AgentOrchestrator): """Specialized orchestrator for book writing""" def __init__(self): super().__init__() self.setup_book_workflow() def setup_book_workflow(self): """Define the book writing workflow""" book_workflow = Workflow( name="complete_book_creation", description="End-to-end book creation from idea to publication", pattern=CoordinationPattern.PIPELINE, stages=[ { 'name': 'ideation', 'type': 'parallel', 'agents': ['creative_agent', 'market_analyst', 'genre_specialist'], 'aggregation_fn': self.aggregate_ideas }, { 'name': 'research', 'type': 'parallel', 'agents': ['research_agent', 'fact_checker', 'source_gatherer'], 'aggregation_fn': self.aggregate_research }, { 'name': 'outlining', 'type': 'sequential', 'agents': ['structure_agent', 'chapter_planner'] }, { 'name': 'writing', 'type': 'parallel', 'agents': ['chapter_writer_1', 'chapter_writer_2', 'chapter_writer_3'], 'aggregation_fn': self.aggregate_chapters }, { 'name': 'editing', 'type': 'sequential', 'agents': ['content_editor', 'copy_editor', 'proofreader'] }, { 'name': 'quality_check', 'type': 'consensus', 'agents': ['quality_agent_1', 'quality_agent_2', 'quality_agent_3'] }, { 'name': 'finalization', 'type': 'sequential', 'agents': ['formatter', 'metadata_generator', 'publisher'] } ], validators=[self.validate_book_input], error_handlers=[self.handle_book_error] ) self.workflows['book_creation'] = book_workflow def aggregate_ideas(self, results: Dict[str, Any]) -> Dict[str, Any]: """Aggregate ideation results from multiple agents""" aggregated = { 'concepts': [], 'themes': [], 'target_audience': [], 'unique_angles': [] } for agent_id, result in results.items(): if 'error' not in result: aggregated['concepts'].extend(result.get('concepts', [])) aggregated['themes'].extend(result.get('themes', [])) aggregated['target_audience'].extend(result.get('audience', [])) aggregated['unique_angles'].extend(result.get('angles', [])) # Deduplicate and rank aggregated['concepts'] = list(set(aggregated['concepts']))[:5] aggregated['themes'] = list(set(aggregated['themes']))[:10] return aggregated def aggregate_research(self, results: Dict[str, Any]) -> Dict[str, Any]: """Aggregate research from multiple sources""" aggregated = { 'facts': [], 'sources': [], 'citations': [], 'key_insights': [] } for agent_id, result in results.items(): if 'error' not in result: aggregated['facts'].extend(result.get('facts', [])) aggregated['sources'].extend(result.get('sources', [])) aggregated['citations'].extend(result.get('citations', [])) aggregated['key_insights'].extend(result.get('insights', [])) return aggregated def aggregate_chapters(self, results: Dict[str, Any]) -> Dict[str, Any]: """Combine chapters from parallel writers""" chapters = [] for agent_id, result in results.items(): if 'error' not in result and 'chapters' in result: chapters.extend(result['chapters']) # Sort chapters by number chapters.sort(key=lambda x: x.get('number', 0)) return {'chapters': chapters} def validate_book_input(self, input_data: Any) -> bool: """Validate input for book creation""" required_fields = ['topic', 'genre', 'target_length'] if not isinstance(input_data, dict): return False for field in required_fields: if field not in input_data: return False return True async def handle_book_error(self, error: Exception, input_data: Any) -> Any: """Handle errors in book creation workflow""" print(f"Error in book creation: {error}") # Attempt recovery if "timeout" in str(error).lower(): # Retry with extended timeout return await self.retry_with_timeout(input_data, timeout=120) # Log error and return partial result return { 'status': 'partial', 'error': str(error), 'completed_stages': self.get_completed_stages() } def get_completed_stages(self) -> List[str]: """Get list of successfully completed stages""" if not self.execution_history: return [] completed = [] for entry in self.execution_history: if 'stage' in entry and entry.get('status') == 'completed': completed.append(entry['stage']) return completed`

## Multi-Agent Systems

Now let's build sophisticated multi-agent systems that can handle complex, real-world tasks through intelligent coordination.

### System Architecture

Task Input
→
Task Decomposition
→
Agent Assignment
→
Parallel Execution
→
Result Synthesis

### Complete Multi-Agent System Implementation

`# multi_agent_system.py import asyncio from typing import Dict, List, Any, Optional, Tuple from dataclasses import dataclass, field from datetime import datetime import json import uuid from enum import Enum from abc import ABC, abstractmethod class TaskPriority(Enum): LOW = 1 MEDIUM = 2 HIGH = 3 CRITICAL = 4 class TaskStatus(Enum): PENDING = "pending" ASSIGNED = "assigned" IN_PROGRESS = "in_progress" COMPLETED = "completed" FAILED = "failed" @dataclass class Task: """Represents a task in the multi-agent system""" id: str type: str description: str input_data: Any priority: TaskPriority status: TaskStatus = TaskStatus.PENDING assigned_agent: Optional[str] = None result: Optional[Any] = None error: Optional[str] = None created_at: datetime = field(default_factory=datetime.now) started_at: Optional[datetime] = None completed_at: Optional[datetime] = None dependencies: List[str] = field(default_factory=list) def to_dict(self) -> Dict[str, Any]: return { 'id': self.id, 'type': self.type, 'description': self.description, 'priority': self.priority.value, 'status': self.status.value, 'assigned_agent': self.assigned_agent, 'dependencies': self.dependencies } class AgentCapability: """Defines what an agent can do""" def __init__(self, task_types: List[str], max_concurrent: int = 3): self.task_types = task_types self.max_concurrent = max_concurrent self.current_load = 0 def can_handle(self, task_type: str) -> bool: return task_type in self.task_types def has_capacity(self) -> bool: return self.current_load < self.max_concurrent class SpecializedAgent(ABC): """Abstract base class for specialized agents""" def __init__(self, agent_id: str, name: str, capabilities: AgentCapability): self.id = agent_id self.name = name self.capabilities = capabilities self.task_queue: asyncio.Queue = asyncio.Queue() self.completed_tasks: List[Task] = [] self.performance_metrics = { 'tasks_completed': 0, 'tasks_failed': 0, 'average_time': 0, 'success_rate': 0 } @abstractmethod async def process_task(self, task: Task) -> Any: """Process a specific task - to be implemented by subclasses""" pass async def run(self): """Main agent loop""" while True: try: # Get next task from queue task = await self.task_queue.get() # Update task status task.status = TaskStatus.IN_PROGRESS task.started_at = datetime.now() self.capabilities.current_load += 1 # Process the task try: result = await self.process_task(task) task.result = result task.status = TaskStatus.COMPLETED self.performance_metrics['tasks_completed'] += 1 except Exception as e: task.error = str(e) task.status = TaskStatus.FAILED self.performance_metrics['tasks_failed'] += 1 # Update metrics task.completed_at = datetime.now() self.completed_tasks.append(task) self.capabilities.current_load -= 1 self.update_metrics(task) except asyncio.CancelledError: break except Exception as e: print(f"Agent {self.id} error: {e}") def update_metrics(self, task: Task): """Update performance metrics""" if task.started_at and task.completed_at: duration = (task.completed_at - task.started_at).total_seconds() # Update average time total_tasks = self.performance_metrics['tasks_completed'] + self.performance_metrics['tasks_failed'] if total_tasks > 0: current_avg = self.performance_metrics['average_time'] self.performance_metrics['average_time'] = ( (current_avg * (total_tasks - 1) + duration) / total_tasks ) # Update success rate self.performance_metrics['success_rate'] = ( self.performance_metrics['tasks_completed'] / total_tasks ) # Concrete Agent Implementations class ResearchSpecialist(SpecializedAgent): """Agent specialized in research tasks""" def __init__(self): capabilities = AgentCapability( task_types=['research', 'fact_check', 'source_gathering'], max_concurrent=5 ) super().__init__("research_specialist", "Research Specialist", capabilities) async def process_task(self, task: Task) -> Any: """Process research tasks""" if task.type == 'research': return await self.conduct_research(task.input_data) elif task.type == 'fact_check': return await self.fact_check(task.input_data) elif task.type == 'source_gathering': return await self.gather_sources(task.input_data) else: raise ValueError(f"Unknown task type: {task.type}") async def conduct_research(self, data: Dict[str, Any]) -> Dict[str, Any]: """Conduct comprehensive research""" await asyncio.sleep(2) # Simulate research time return { 'topic': data.get('topic'), 'findings': [ f"Key finding 1 about {data.get('topic')}", f"Key finding 2 about {data.get('topic')}", f"Key finding 3 about {data.get('topic')}" ], 'sources': ['Academic Paper A', 'Industry Report B', 'Expert Interview C'], 'confidence': 0.85, 'timestamp': datetime.now().isoformat() } async def fact_check(self, data: Dict[str, Any]) -> Dict[str, Any]: """Verify facts and claims""" await asyncio.sleep(1) return { 'claim': data.get('claim'), 'verified': True, 'confidence': 0.92, 'supporting_evidence': ['Evidence 1', 'Evidence 2'] } async def gather_sources(self, data: Dict[str, Any]) -> List[Dict[str, Any]]: """Gather relevant sources""" await asyncio.sleep(1.5) return [ {'title': 'Source 1', 'url': 'http://example.com/1', 'relevance': 0.9}, {'title': 'Source 2', 'url': 'http://example.com/2', 'relevance': 0.85}, {'title': 'Source 3', 'url': 'http://example.com/3', 'relevance': 0.8} ] class ContentWriter(SpecializedAgent): """Agent specialized in content writing""" def __init__(self): capabilities = AgentCapability( task_types=['write_chapter', 'write_section', 'write_summary'], max_concurrent=3 ) super().__init__("content_writer", "Content Writer", capabilities) async def process_task(self, task: Task) -> Any: """Process writing tasks""" if task.type == 'write_chapter': return await self.write_chapter(task.input_data) elif task.type == 'write_section': return await self.write_section(task.input_data) elif task.type == 'write_summary': return await self.write_summary(task.input_data) else: raise ValueError(f"Unknown task type: {task.type}") async def write_chapter(self, data: Dict[str, Any]) -> Dict[str, Any]: """Write a complete chapter""" await asyncio.sleep(3) # Simulate writing time chapter_num = data.get('chapter_number', 1) outline = data.get('outline', {}) content = f"# Chapter {chapter_num}: {outline.get('title', 'Untitled')}\n\n" content += f"{outline.get('introduction', 'Introduction text...')}\n\n" for section in outline.get('sections', []): content += f"## {section.get('title')}\n\n" content += f"{section.get('content', 'Section content...')}\n\n" return { 'chapter_number': chapter_num, 'content': content, 'word_count': len(content.split()), 'status': 'complete' } async def write_section(self, data: Dict[str, Any]) -> Dict[str, Any]: """Write a section of content""" await asyncio.sleep(1.5) return { 'section': data.get('section_title'), 'content': f"Content for section: {data.get('section_title')}", 'word_count': 250 } async def write_summary(self, data: Dict[str, Any]) -> str: """Write a summary""" await asyncio.sleep(1) return f"Summary of {data.get('topic')}: {data.get('content', '')[:200]}..." class QualityController(SpecializedAgent): """Agent specialized in quality control""" def __init__(self): capabilities = AgentCapability( task_types=['review', 'edit', 'proofread'], max_concurrent=4 ) super().__init__("quality_controller", "Quality Controller", capabilities) async def process_task(self, task: Task) -> Any: """Process quality control tasks""" if task.type == 'review': return await self.review_content(task.input_data) elif task.type == 'edit': return await self.edit_content(task.input_data) elif task.type == 'proofread': return await self.proofread_content(task.input_data) else: raise ValueError(f"Unknown task type: {task.type}") async def review_content(self, data: Dict[str, Any]) -> Dict[str, Any]: """Review content for quality""" await asyncio.sleep(2) return { 'quality_score': 0.82, 'issues_found': [ 'Inconsistent terminology in section 2', 'Missing citation in paragraph 5', 'Weak conclusion' ], 'recommendations': [ 'Strengthen the introduction', 'Add more examples', 'Improve transitions between sections' ] } async def edit_content(self, data: Dict[str, Any]) -> Dict[str, Any]: """Edit content for improvement""" await asyncio.sleep(1.5) original = data.get('content', '') edited = original.replace(' ', ' ').replace('\n\n\n', '\n\n') return { 'original': original, 'edited': edited, 'changes_made': 5, 'improvement_score': 0.15 } async def proofread_content(self, data: Dict[str, Any]) -> Dict[str, Any]: """Proofread for errors""" await asyncio.sleep(1) return { 'errors_found': 3, 'corrections': [ {'type': 'spelling', 'location': 'para 2', 'correction': 'correct spelling'}, {'type': 'grammar', 'location': 'para 5', 'correction': 'fix grammar'}, {'type': 'punctuation', 'location': 'para 8', 'correction': 'add comma'} ] } class MultiAgentSystem: """Main multi-agent system coordinator""" def __init__(self): self.agents: Dict[str, SpecializedAgent] = {} self.task_queue: asyncio.Queue = asyncio.Queue() self.pending_tasks: Dict[str, Task] = {} self.completed_tasks: Dict[str, Task] = {} self.task_graph: Dict[str, List[str]] = {} # Dependencies self.agent_tasks: List[asyncio.Task] = [] def register_agent(self, agent: SpecializedAgent): """Register an agent in the system""" self.agents[agent.id] = agent print(f"✅ Registered agent: {agent.name}") async def start_agents(self): """Start all registered agents""" for agent in self.agents.values(): task = asyncio.create_task(agent.run()) self.agent_tasks.append(task) print(f"🚀 Started {len(self.agents)} agents") async def submit_task( self, task_type: str, description: str, input_data: Any, priority: TaskPriority = TaskPriority.MEDIUM, dependencies: List[str] = None ) -> str: """Submit a task to the system""" task = Task( id=str(uuid.uuid4()), type=task_type, description=description, input_data=input_data, priority=priority, dependencies=dependencies or [] ) self.pending_tasks[task.id] = task # Check if dependencies are met if not dependencies or all( dep in self.completed_tasks for dep in dependencies ): await self.assign_task(task) else: # Add to dependency graph self.task_graph[task.id] = dependencies return task.id async def assign_task(self, task: Task): """Assign task to appropriate agent""" # Find capable agents capable_agents = [ agent for agent in self.agents.values() if agent.capabilities.can_handle(task.type) and agent.capabilities.has_capacity() ] if not capable_agents: # Wait for an agent to become available await self.task_queue.put(task) return # Select best agent (lowest load) best_agent = min(capable_agents, key=lambda a: a.capabilities.current_load) # Assign task task.assigned_agent = best_agent.id task.status = TaskStatus.ASSIGNED await best_agent.task_queue.put(task) print(f"📋 Assigned task {task.id} to {best_agent.name}") async def wait_for_task(self, task_id: str, timeout: float = 60) -> Optional[Task]: """Wait for a task to complete""" start_time = asyncio.get_event_loop().time() while asyncio.get_event_loop().time() - start_time < timeout: if task_id in self.completed_tasks: return self.completed_tasks[task_id] # Check if task is completed by any agent for agent in self.agents.values(): for completed_task in agent.completed_tasks: if completed_task.id == task_id: self.completed_tasks[task_id] = completed_task self.check_dependencies(task_id) return completed_task await asyncio.sleep(0.5) return None # Timeout def check_dependencies(self, completed_task_id: str): """Check if any pending tasks can now be executed""" tasks_to_assign = [] for task_id, deps in list(self.task_graph.items()): if completed_task_id in deps: deps.remove(completed_task_id) if not deps: # All dependencies met if task_id in self.pending_tasks: tasks_to_assign.append(self.pending_tasks[task_id]) del self.task_graph[task_id] # Assign tasks that are now ready for task in tasks_to_assign: asyncio.create_task(self.assign_task(task)) def get_system_status(self) -> Dict[str, Any]: """Get current system status""" status = { 'agents': {}, 'tasks': { 'pending': len(self.pending_tasks), 'completed': len(self.completed_tasks), 'in_progress': 0 }, 'performance': {} } for agent_id, agent in self.agents.items(): status['agents'][agent_id] = { 'name': agent.name, 'current_load': agent.capabilities.current_load, 'max_capacity': agent.capabilities.max_concurrent, 'tasks_completed': agent.performance_metrics['tasks_completed'], 'success_rate': agent.performance_metrics['success_rate'] } # Count in-progress tasks status['tasks']['in_progress'] += agent.capabilities.current_load return status async def shutdown(self): """Gracefully shutdown the system""" print("🛑 Shutting down multi-agent system...") # Cancel all agent tasks for task in self.agent_tasks: task.cancel() # Wait for cancellation await asyncio.gather(*self.agent_tasks, return_exceptions=True) print("✅ System shutdown complete") # Example: Book Creation Pipeline async def create_book_with_multi_agent_system(): """Complete example of creating a book using multi-agent system""" # Initialize system system = MultiAgentSystem() # Register specialized agents system.register_agent(ResearchSpecialist()) system.register_agent(ContentWriter()) system.register_agent(QualityController()) # Start all agents await system.start_agents() try: print("\n📚 Starting Book Creation Pipeline\n") # Phase 1: Research print("Phase 1: Research") research_task_id = await system.submit_task( task_type='research', description='Research the topic of AI in Healthcare', input_data={'topic': 'AI in Healthcare'}, priority=TaskPriority.HIGH ) research_result = await system.wait_for_task(research_task_id) print(f"✅ Research completed: {research_result.result['findings'][:2]}...") # Phase 2: Fact Checking print("\nPhase 2: Fact Checking") fact_check_task_id = await system.submit_task( task_type='fact_check', description='Verify research findings', input_data={ 'claim': research_result.result['findings'][0] }, priority=TaskPriority.HIGH, dependencies=[research_task_id] ) fact_check_result = await system.wait_for_task(fact_check_task_id) print(f"✅ Fact check completed: Verified = {fact_check_result.result['verified']}") # Phase 3: Writing Chapters (Parallel) print("\nPhase 3: Writing Chapters") chapter_tasks = [] for i in range(1, 4): task_id = await system.submit_task( task_type='write_chapter', description=f'Write chapter {i}', input_data={ 'chapter_number': i, 'outline': { 'title': f'Chapter {i}: Aspect {i} of AI in Healthcare', 'introduction': f'Introduction to aspect {i}', 'sections': [ {'title': f'Section {i}.1', 'content': 'Content...'}, {'title': f'Section {i}.2', 'content': 'Content...'} ] } }, priority=TaskPriority.MEDIUM, dependencies=[research_task_id] ) chapter_tasks.append(task_id) # Wait for all chapters chapters = [] for task_id in chapter_tasks: result = await system.wait_for_task(task_id) chapters.append(result.result) print(f"✅ Chapter {result.result['chapter_number']} written: {result.result['word_count']} words") # Phase 4: Quality Review print("\nPhase 4: Quality Review") review_task_id = await system.submit_task( task_type='review', description='Review all chapters', input_data={ 'content': '\n\n'.join([ch['content'] for ch in chapters]) }, priority=TaskPriority.HIGH, dependencies=chapter_tasks ) review_result = await system.wait_for_task(review_task_id) print(f"✅ Review completed: Quality Score = {review_result.result['quality_score']:.2f}") print(f" Issues found: {review_result.result['issues_found']}") # Phase 5: Editing print("\nPhase 5: Editing") edit_task_id = await system.submit_task( task_type='edit', description='Edit based on review feedback', input_data={ 'content': '\n\n'.join([ch['content'] for ch in chapters]), 'feedback': review_result.result }, priority=TaskPriority.MEDIUM, dependencies=[review_task_id] ) edit_result = await system.wait_for_task(edit_task_id) print(f"✅ Editing completed: {edit_result.result['changes_made']} changes made") # Phase 6: Final Proofreading print("\nPhase 6: Proofreading") proofread_task_id = await system.submit_task( task_type='proofread', description='Final proofreading', input_data={ 'content': edit_result.result['edited'] }, priority=TaskPriority.LOW, dependencies=[edit_task_id] ) proofread_result = await system.wait_for_task(proofread_task_id) print(f"✅ Proofreading completed: {proofread_result.result['errors_found']} errors corrected") # Get final system status print("\n📊 System Status:") status = system.get_system_status() print(json.dumps(status, indent=2)) print("\n🎉 Book creation pipeline completed successfully!") finally: # Shutdown system await system.shutdown() if __name__ == "__main__": asyncio.run(create_book_with_multi_agent_system())`

#### Multi-Agent System Simulator

Configure and run a multi-agent system: