Orchestrate Complex Multi-Agent Systems

AI Agent Orchestrator is a skill for designing multi-agent systems where an orchestrator agent delegates tasks to specialized, utility, and monitoring agents. It covers architecture patterns for coordinating those agents (hub-and-spoke, pipeline), task delegation (intelligent routing by capability, dynamic task decomposition with dependencies), inter-agent communication (a typed message protocol, shared-state synchronization with locking), and failure handling (circuit breaker, graceful degradation across primary, fallback, and emergency model tiers).

Use it when building a system where multiple agents need to coordinate on a task - routing a subtask to the right specialized agent, running agents in a pipeline or in parallel with a concurrency limit, or keeping a task working if a preferred agent or model tier fails. It's not for a single-agent workflow: the patterns here (orchestrator/specialized/utility/monitoring roles, message buses, shared state) only pay off once more than one agent needs to communicate and hand off work.

Input is a complex task plus the pool of available specialized agents and their capabilities. Output is orchestration code in Python: an OrchestratorAgent that decomposes a task into subtasks and routes each to a capable agent before synthesizing results; a TaskRouter that matches required skills against agent capabilities and load-balances across capable agents; a TaskDecomposer that breaks a task into dependency-ordered subtasks; an AgentMessage/MessageBus protocol with typed message types, correlation IDs, priority, and TTL; a SharedState class with per-key locking and subscriber notification; a WorkflowOrchestrator for conditional and branching workflows; a ParallelOrchestrator using an asyncio semaphore to cap concurrent agents; an AgentCircuitBreaker with closed, open, and half-open states; a ResilientOrchestrator that falls back through primary, fallback, and emergency model tiers; and an AgentPool for reusing initialized agent instances.

Built on Python's asyncio and concurrent.futures.ThreadPoolExecutor for concurrency, dataclasses and Enum for the message schema, and a message-queue-backed bus for decoupled, asynchronous agent communication:

class AgentCircuitBreaker:
    def __init__(self, failure_threshold=5, timeout=60):
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.last_failure_time = None
        self.state = 'CLOSED'  # CLOSED, OPEN, HALF_OPEN

Engineers building multi-agent systems who need the coordination scaffolding, not just individual agents - plus the operational side: correlation-ID logging and distributed tracing, tracking per-agent latency/success-rate/resource-usage with alerts on failures and performance degradation, inter-agent authentication and input sanitization over encrypted channels, stateless and horizontally-scaled agent pools, and testing strategies that mock agent dependencies and simulate failures and network partitions with realistic load testing.