Agent Task Orchestration System: From Design to Production

from typing import List, Dict import asyncio class TaskDecomposer: def __init__(self, llm_service): self.llm = llm_service async def decompose_task(self, task_description: str) -> Dict: """Intelligent task decomposition""" prompt = f""" Task Description: {task_description} Please decompose this task into subtasks, output format: {{ "subtasks": [ {{ "id": "task_1", "name": "subtask name", "description": "detailed description", "dependencies": [], "estimated_time": "estimated duration (minutes)" }} ] }} Requirements: 1. Appropriate subtask granularity 2. Clear task dependencies 3. Suitable for parallel processing """ response = await self.llm.generate(prompt) return self._validate_and_process(response) def _validate_and_process(self, decomposition_result: dict) -> dict: """Validate and process decomposition results""" # Validate task dependency relationships  self._check_circular_dependencies(decomposition_result["subtasks"]) # Build task execution graph  return self._build_execution_graph(decomposition_result["subtasks"]) 

class TaskExecutor: def __init__(self, max_workers: int = 5): self.max_workers = max_workers self.task_queue = asyncio.Queue() self.results = {} self.semaphore = asyncio.Semaphore(max_workers) async def execute_tasks(self, task_graph: Dict): """Execute task graph""" # Create worker pool  workers = [ self._worker(f"worker_{i}") for i in range(self.max_workers) ] # Add executable tasks to queue  ready_tasks = self._get_ready_tasks(task_graph) for task in ready_tasks: await self.task_queue.put(task) # Wait for all tasks to complete  await asyncio.gather(*workers) async def _worker(self, worker_id: str): """Worker coroutine""" while True: try: async with self.semaphore: task = await self.task_queue.get() if task is None: break # Execute task  result = await self._execute_single_task(task) self.results[task["id"]] = result # Check and add new executable tasks  new_ready_tasks = self._get_ready_tasks(task_graph) for task in new_ready_tasks: await self.task_queue.put(task) except Exception as e: logger.error(f"Worker {worker_id} error: {str(e)}") 

class ResourceManager: def __init__(self): self.resource_pool = { 'cpu': ResourcePool(max_units=16), 'memory': ResourcePool(max_units=32), 'gpu': ResourcePool(max_units=4) } async def allocate(self, requirements: Dict[str, int]): """Allocate resources""" allocated = {} try: for resource_type, amount in requirements.items(): allocated[resource_type] = await self.resource_pool[resource_type].acquire(amount) return allocated except InsufficientResourceError: # Rollback allocated resources  await self.release(allocated) raise async def release(self, allocated_resources: Dict): """Release resources""" for resource_type, resource in allocated_resources.items(): await self.resource_pool[resource_type].release(resource) 

class SystemMonitor: def __init__(self): self.metrics = {} self.alerts = AlertManager() async def monitor_task(self, task_id: str): """Monitor single task""" start_time = time.time() try: # Log task start  self.log_task_start(task_id) # Monitor resource usage  resource_usage = await self.track_resource_usage(task_id) # Check performance metrics  if resource_usage['cpu'] > 80: await self.alerts.send_alert( f"High CPU usage for task {task_id}" ) return resource_usage finally: # Log task completion  duration = time.time() - start_time self.log_task_completion(task_id, duration) 

class PerformanceOptimizer: def __init__(self): self.cache = LRUCache(maxsize=1000) self.batch_processor = BatchProcessor() async def optimize_execution(self, tasks: List[Dict]): """Optimize task execution""" # 1. Task grouping  task_groups = self._group_similar_tasks(tasks) # 2. Batch processing optimization  optimized_groups = [] for group in task_groups: if len(group) > 1: # Merge similar tasks  optimized = await self.batch_processor.process(group) else: optimized = group[0] optimized_groups.append(optimized) # 3. Resource pre-allocation  for group in optimized_groups: await self._preallocate_resources(group) return optimized_groups 

class PluginManager: def __init__(self): self.plugins = {} def register_plugin(self, name: str, plugin: Any): """Register plugin""" if not hasattr(plugin, 'execute'): raise InvalidPluginError( "Plugin must implement execute method" ) self.plugins[name] = plugin async def execute_plugin(self, name: str, *args, **kwargs): """Execute plugin""" if name not in self.plugins: raise PluginNotFoundError(f"Plugin {name} not found") try: return await self.plugins[name].execute(*args, **kwargs) except Exception as e: logger.error(f"Plugin {name} execution failed: {str(e)}") raise 

class CustomTaskRegistry: _task_types = {} @classmethod def register(cls, task_type: str): """Register custom task type""" def decorator(task_class): cls._task_types[task_type] = task_class return task_class return decorator @classmethod def create_task(cls, task_type: str, **kwargs): """Create task instance""" if task_type not in cls._task_types: raise UnknownTaskTypeError(f"Unknown task type: {task_type}") return cls._task_types[task_type](**kwargs) @CustomTaskRegistry.register("data_processing") class DataProcessingTask: async def execute(self, data): # Implement data processing logic  pass @CustomTaskRegistry.register("report_generation") class ReportGenerationTask: async def execute(self, data): # Implement report generation logic  pass 

async def generate_market_report(topic: str): # Initialize system components  orchestrator = TaskOrchestrator() optimizer = PerformanceOptimizer() monitor = SystemMonitor() try: # 1. Task planning  task_graph = await orchestrator.plan_tasks({ "topic": topic, "required_sections": [ "market_overview", "competitor_analysis", "trends_analysis", "recommendations" ] }) # 2. Performance optimization  optimized_tasks = await optimizer.optimize_execution( task_graph["tasks"] ) # 3. Execute tasks  with monitor.track_execution(): results = await orchestrator.execute_dag({ "tasks": optimized_tasks }) # 4. Generate report  report = await orchestrator.compile_results(results) return report except Exception as e: logger.error(f"Report generation failed: {str(e)}") # Trigger alert  await monitor.alerts.send_alert( f"Report generation failed for topic: {topic}" ) raise