Multi Agent Orchestration

Multi-Agent Orchestration

Coordinate multiple specialized agents for complex tasks.

Fan-Out/Fan-In Pattern

async def multi_agent_analysis(content: str) -> dict:
    """Fan-out to specialists, fan-in to synthesize."""
    agents = [
        ("security", security_agent),
        ("performance", performance_agent),
        ("code_quality", quality_agent),
        ("architecture", architecture_agent),
    ]

    # Fan-out: Run all agents in parallel
    tasks = [agent(content) for _, agent in agents]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    # Filter successful results
    findings = [
        {"agent": name, "result": result}
        for (name, _), result in zip(agents, results)
        if not isinstance(result, Exception)
    ]

    # Fan-in: Synthesize findings
    return await synthesize_findings(findings)

Supervisor Pattern

class Supervisor:
    """Central coordinator that routes to specialists."""

    def __init__(self, agents: dict):
        self.agents = agents  # {"security": agent, "performance": agent}
        self.completed = []

    async def run(self, task: str) -> dict:
        """Route task through appropriate agents."""
        # 1. Determine which agents to use
        plan = await self.plan_routing(task)

        # 2. Execute in dependency order
        results = {}
        for agent_name in plan.execution_order:
            if plan.can_parallelize(agent_name):
                # Run parallel batch
                batch = plan.get_parallel_batch(agent_name)
                batch_results = await asyncio.gather(*[
                    self.agents[name](task, context=results)
                    for name in batch
                ])
                results.update(dict(zip(batch, batch_results)))
            else:
                # Run sequential
                results[agent_name] = await self.agents[agent_name](
                    task, context=results
                )

        return results

    async def plan_routing(self, task: str) -> RoutingPlan:
        """Use LLM to determine agent routing."""
        response = await llm.chat([{
            "role": "user",
            "content": f"""Task: {task}

Available agents: {list(self.agents.keys())}

Which agents should handle this task?
What order? Can any run in parallel?"""
        }])
        return parse_routing_plan(response.content)

Conflict Resolution

async def resolve_conflicts(findings: list[dict]) -> list[dict]:
    """When agents disagree, resolve by confidence or LLM."""
    conflicts = detect_conflicts(findings)

    if not conflicts:
        return findings

    for conflict in conflicts:
        # Option 1: Higher confidence wins
        winner = max(conflict.agents, key=lambda a: a.confidence)

        # Option 2: LLM arbitration
        resolution = await llm.chat([{
            "role": "user",
            "content": f"""Two agents disagree:

Agent A ({conflict.agent_a.name}): {conflict.agent_a.finding}
Agent B ({conflict.agent_b.name}): {conflict.agent_b.finding}

Which is more likely correct and why?"""
        }])

        # Record resolution
        conflict.resolution = parse_resolution(resolution.content)

    return apply_resolutions(findings, conflicts)

Synthesis Pattern

async def synthesize_findings(findings: list[dict]) -> dict:
    """Combine multiple agent outputs into coherent result."""
    # Group by category
    by_category = {}
    for f in findings:
        cat = f.get("category", "general")
        by_category.setdefault(cat, []).append(f)

    # Synthesize each category
    synthesis = await llm.chat([{
        "role": "user",
        "content": f"""Synthesize these agent findings into a coherent summary:

{json.dumps(by_category, indent=2)}

Output format:
- Executive summary (2-3 sentences)
- Key findings by category
- Recommendations
- Confidence score (0-1)"""
    }])

    return parse_synthesis(synthesis.content)

Agent Communication Bus

class AgentBus:
    """Message passing between agents."""

    def __init__(self):
        self.messages = []
        self.subscribers = {}

    def publish(self, from_agent: str, message: dict):
        """Broadcast message to all agents."""
        msg = {"from": from_agent, "data": message, "ts": time.time()}
        self.messages.append(msg)

        for callback in self.subscribers.values():
            callback(msg)

    def subscribe(self, agent_id: str, callback):
        """Register agent to receive messages."""
        self.subscribers[agent_id] = callback

    def get_history(self, agent_id: str = None) -> list:
        """Get message history, optionally filtered."""
        if agent_id:
            return [m for m in self.messages if m["from"] == agent_id]
        return self.messages

CC Agent Teams (CC 2.1.33+)

CC 2.1.33 introduces native Agent Teams — teammates with peer-to-peer messaging, shared task lists, and mesh topology.

Star vs Mesh Topology

Star (Task tool):              Mesh (Agent Teams):
      Lead                           Lead (delegate)
     /||\                          /  |  \
    / || \                        /   |   \
   A  B  C  D                   A ←→ B ←→ C
   (no cross-talk)              (peer messaging)

Dual-Mode Decision Tree

Complexity Assessment:
├── Score < 3.0  → Task tool subagents (cheaper, simpler)
├── Score 3.0-3.5 → User choice (recommend Teams for cross-cutting)
└── Score > 3.5  → Agent Teams (if CLAUDE_CODE_EXPERIMENTAL_AGENT_TEAMS=1)

Override: ORCHESTKIT_PREFER_TEAMS=1 forces Agent Teams
Fallback: Teams disabled → always use Task tool

Team Formation

# 1. Create team with shared task list
TeamCreate(team_name="feature-auth", description="User auth implementation")

# 2. Create tasks in shared list
TaskCreate(subject="Design API schema", description="...")
TaskCreate(subject="Build React components", description="...", addBlockedBy=["1"])
TaskCreate(subject="Write integration tests", description="...", addBlockedBy=["1","2"])

# 3. Spawn teammates (each is a full CC session)
Task(prompt="You are the backend architect...",
     team_name="feature-auth", name="backend-dev",
     subagent_type="backend-system-architect")

Task(prompt="You are the frontend developer...",
     team_name="feature-auth", name="frontend-dev",
     subagent_type="frontend-ui-developer")

# 4. Teammates self-claim tasks from shared list
# 5. Teammates message each other directly
# 6. Lead monitors via idle notifications

Peer Messaging Patterns

# Direct message (default — use this)
SendMessage(type="message", recipient="frontend-dev",
  content="API contract: GET /users/:id → {id, name, email}",
  summary="API contract ready")

# Broadcast (expensive — use sparingly)
SendMessage(type="broadcast",
  content="Auth header format changed to Bearer",
  summary="Breaking auth change")

# Shutdown when done
SendMessage(type="shutdown_request", recipient="frontend-dev",
  content="All tasks complete")

Cost Comparison

Teams cost more because each teammate is a full CC session. Worth it when cross-agent communication prevents rework.

Key Decisions

Common Mistakes

• No timeout per agent (one slow agent blocks all)
• No error isolation (one failure crashes workflow)
• Over-coordination (too much overhead)
• Missing synthesis (raw agent outputs not useful)
• Using Agent Teams for simple sequential work (use Task tool)
• Broadcasting when a direct message suffices (wastes tokens)

Related Skills

• langgraph-supervisor - LangGraph supervisor pattern
• langgraph-parallel - Fan-out/fan-in with LangGraph
• agent-loops - Single agent patterns
• task-dependency-patterns - Task management with Agent Teams workflow

Capability Details

agent-communication

Keywords: agent communication, message passing, agent protocol, inter-agent Solves:

• Establish communication between agents
• Implement message passing patterns
• Handle async agent communication

task-delegation

Keywords: delegate, task routing, work distribution, agent dispatch Solves:

• Route tasks to specialized agents
• Implement work distribution strategies
• Handle agent capability matching

result-aggregation

Keywords: aggregate, combine results, merge outputs, synthesis Solves:

• Combine outputs from multiple agents
• Implement result synthesis patterns
• Handle conflicting agent outputs

error-coordination

Keywords: error handling, retry, fallback agent, failure recovery Solves:

• Handle agent failures gracefully
• Implement retry and fallback patterns
• Coordinate error recovery

agent-lifecycle

Keywords: lifecycle, spawn agent, terminate, agent pool Solves:

• Manage agent creation and termination
• Implement agent pooling
• Handle agent health checks

Multi-Agent Orchestration Checklist

Architecture

• [ ] Define agent responsibilities
• [ ] Plan communication patterns
• [ ] Set coordination strategy
• [ ] Design failure handling

Agent Design

• [ ] Single responsibility per agent
• [ ] Clear input/output contracts
• [ ] Independent operation
• [ ] Stateless when possible

Communication

• [ ] Message format definition
• [ ] Async message passing
• [ ] Result aggregation
• [ ] Error propagation

Coordination

• [ ] Central orchestrator
• [ ] Task queue management
• [ ] Priority handling
• [ ] Deadlock prevention

Monitoring

• [ ] Agent health checks
• [ ] Task completion tracking
• [ ] Performance metrics
• [ ] Error rates

Agent Coordination Patterns

Patterns for coordinating multiple specialized agents in complex workflows.

Supervisor-Worker Pattern

from typing import Protocol, Any
import asyncio

class Agent(Protocol):
    async def run(self, task: str, context: dict) -> dict: ...

class SupervisorCoordinator:
    """Central supervisor that routes tasks to worker agents."""

    def __init__(self, workers: dict[str, Agent]):
        self.workers = workers
        self.execution_log: list[dict] = []

    async def route_and_execute(
        self,
        task: str,
        required_agents: list[str],
        parallel: bool = True
    ) -> dict[str, Any]:
        """Route task to specified agents."""
        context = {"task": task, "results": {}}

        if parallel:
            tasks = [
                self._run_worker(name, task, context)
                for name in required_agents
            ]
            results = await asyncio.gather(*tasks, return_exceptions=True)
            return dict(zip(required_agents, results))
        else:
            for name in required_agents:
                context["results"][name] = await self._run_worker(
                    name, task, context
                )
            return context["results"]

    async def _run_worker(
        self, name: str, task: str, context: dict
    ) -> dict:
        """Execute single worker with timeout."""
        try:
            result = await asyncio.wait_for(
                self.workers[name].run(task, context),
                timeout=30.0
            )
            self.execution_log.append({
                "agent": name, "status": "success", "result": result
            })
            return result
        except asyncio.TimeoutError:
            return {"error": f"{name} timed out"}

Conflict Resolution

async def resolve_agent_conflicts(
    findings: list[dict],
    llm: Any
) -> dict:
    """Resolve conflicts between agent outputs."""
    conflicts = []
    for i, f1 in enumerate(findings):
        for f2 in findings[i+1:]:
            if f1.get("recommendation") != f2.get("recommendation"):
                conflicts.append((f1, f2))

    if not conflicts:
        return {"status": "no_conflicts", "findings": findings}

    # LLM arbitration
    resolution = await llm.ainvoke(f"""
        Agents disagree. Determine best recommendation:
        Agent 1: {conflicts[0][0]}
        Agent 2: {conflicts[0][1]}
        Provide: winner, reasoning, confidence (0-1)
    """)
    return {"status": "resolved", "resolution": resolution}

Configuration

• Worker timeout: 30s default
• Max parallel agents: 8
• Retry failed agents: 1 attempt
• Log all executions for debugging

Cost Optimization

• Batch similar tasks to reduce overhead
• Cache agent results by task hash
• Use cheaper models for simple agents
• Parallelize independent agents always