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())
# 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
Now let's build sophisticated multi-agent systems that can handle complex, real-world tasks through intelligent coordination.
# 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())
Configure and run a multi-agent system: