multi-agent-orchestration

Multi-Agent Orchestration

Patterns for coordinating multiple AI agents to solve complex tasks. Covers orchestrator, pipeline, consensus, delegation, supervisor, and swarm architectures.

When to Use

• User is building a system with multiple cooperating agents
• User needs task delegation or agent supervision patterns
• User wants consensus-based decision making across agents
• User is designing pipeline processing with agent stages
• User asks about swarm intelligence or emergent agent behavior

Core Patterns

Orchestrator Pattern

A central orchestrator decomposes tasks and delegates to specialized worker agents.

import anthropic

client = anthropic.Anthropic()

def orchestrator(task: str) -> str:
    # Step 1: Plan and decompose
    plan = client.messages.create(
        model="claude-sonnet-4-6-20250514",
        max_tokens=2048,
        system="""You are a task orchestrator. Break the task into subtasks.
Return a JSON array of subtasks, each with "id", "agent", "instruction", and "depends_on" (list of ids).
Available agents: researcher, coder, reviewer.""",
        messages=[{"role": "user", "content": task}]
    )

    subtasks = json.loads(plan.content[0].text)

    # Step 2: Execute subtasks respecting dependencies
    results = {}
    for subtask in topological_sort(subtasks):
        dep_context = "\n".join(
            f"Result of {d}: {results[d]}" for d in subtask["depends_on"]
        )
        result = run_worker(
            agent=subtask["agent"],
            instruction=subtask["instruction"],
            context=dep_context
        )
        results[subtask["id"]] = result

    # Step 3: Synthesize final result
    synthesis = client.messages.create(
        model="claude-sonnet-4-6-20250514",
        max_tokens=4096,
        system="Synthesize the worker results into a coherent final response.",
        messages=[{"role": "user", "content": json.dumps(results)}]
    )
    return synthesis.content[0].text

def run_worker(agent: str, instruction: str, context: str) -> str:
    system_prompts = {
        "researcher": "You are a research agent. Find and summarize relevant information.",
        "coder": "You are a coding agent. Write clean, tested code.",
        "reviewer": "You are a review agent. Find bugs, security issues, and improvements."
    }
    response = client.messages.create(
        model="claude-haiku-4-5-20251001",  # Workers use faster model
        max_tokens=2048,
        system=system_prompts[agent],
        messages=[{"role": "user", "content": f"{instruction}\n\nContext:\n{context}"}]
    )
    return response.content[0].text