Scalable Multi-Agent Orchestrator

Hard
Agents

Implement scalable multi-agent orchestration:

Auto-Scaling:

  • Scale up when queue_depth / active_agents > threshold
  • Scale down when utilization < threshold
  • Respect min/max bounds

Task Management:

  1. submit_task(task, priority): Add to priority queue
  2. assign_task_to_agent(): Match task to agent
  3. Tasks processed by priority (lower number = higher priority)

Scaling Logic:

  • scale_up_threshold: 0.8 (80% utilization triggers scale up)
  • scale_down_threshold: 0.3 (30% triggers scale down)

Metrics:

  • Queue depth
  • Agent utilization
  • Throughput (tasks/sec)
  • Average latency

Agent States: idle, busy, starting, stopping

Examples

Example 1:
Input: sys = ScalableMultiAgentSystem(); task_id = sys.submit_task('process data', priority=1); isinstance(task_id, str)
Output: True
Explanation: Task submitted, ID returned

Starter Code

class ScalableMultiAgentSystem:
    """
    Scalable multi-agent system with dynamic agent pool management.
    """
    
    def __init__(self):
        self.agent_pool = {}
        self.task_queue = []
        self.max_agents = 100
        self.min_agents = 5
        self.scale_up_threshold = 0.8
        self.scale_down_threshold = 0.3
    
    def submit_task(self, task, priority=5):
        """
        Submit task to queue with priority (1-10, lower = higher priority).
        Returns task_id.
        """
        # Your implementation here
        pass
    
    def scale_agents(self):
        """
        Auto-scale agent pool based on queue depth and utilization.
        """
        # Your implementation here
        pass
    
    def assign_task_to_agent(self):
        """Assign highest priority task to available agent"""
        # Your implementation here
        pass
    
    def get_system_metrics(self):
        """
        Get comprehensive metrics:
        - queue_depth, active_agents, idle_agents
        - throughput, avg_latency, error_rate
        """
        # Your implementation here
        pass
    
    def rebalance_load(self):
        """Rebalance tasks across agents if uneven"""
        # Your implementation here
        pass
Lines: 1Characters: 0
Ready
The AI Interview - Master AI/ML Interviews