How to Coordinate Multiple AI Agents: 6 Patterns (2026)

// Fan-out / Fan-in with explicit aggregation
import Anthropic from "@anthropic-ai/sdk";

const client = new Anthropic();

interface AgentTask {
  name: string;
  prompt: string;
}

async function fanOutFanIn(tasks: AgentTask[], mergePrompt: string) {
  // Fan out: all agents run in parallel
  const results = await Promise.all(
    tasks.map(async (task) => {
      const response = await client.messages.create({
        model: "claude-sonnet-5-5",
        max_tokens: 4096,
        system: `You are a specialized ${task.name} agent. Be thorough and precise.`,
        messages: [{ role: "user", content: task.prompt }],
      });
      return {
        agent: task.name,
        output: response.content[0].type === "text" ? response.content[0].text : "",
      };
    })
  );

  // Fan in: aggregator merges all outputs
  const mergeInput = results
    .map((r) => `## ${r.agent}\n${r.output}`)
    .join("\n\n---\n\n");

  const merged = await client.messages.create({
    model: "claude-sonnet-5-5",
    max_tokens: 8192,
    system: "You are a synthesis agent. Merge the following agent outputs into a single coherent result. Resolve contradictions. Remove duplicates. Preserve all unique insights.",
    messages: [{ role: "user", content: `${mergePrompt}\n\n${mergeInput}` }],
  });

  return merged.content[0].type === "text" ? merged.content[0].text : "";
}

// Usage
const result = await fanOutFanIn(
  [
    { name: "docs-researcher", prompt: "Research the latest Next.js 16 App Router changes" },
    { name: "migration-analyst", prompt: "Find breaking changes between Next.js 15 and 16" },
    { name: "community-scanner", prompt: "Find common migration issues reported on GitHub" },
  ],
  "Create a comprehensive Next.js 16 migration guide from these research outputs."
);

# Pipeline with inter-stage validation
from anthropic import Anthropic

client = Anthropic()

def run_pipeline(task: str, stages: list[dict]) -> str:
    current_input = task

    for i, stage in enumerate(stages):
        response = client.messages.create(
            model="claude-sonnet-5-5",
            max_tokens=stage.get("max_tokens", 4096),
            system=stage["system_prompt"],
            messages=[{"role": "user", "content": current_input}],
        )
        output = response.content[0].text

        # Validate output before passing to next stage
        if "validator" in stage:
            is_valid, error = stage["validator"](output)
            if not is_valid:
                # Retry with error context
                retry_response = client.messages.create(
                    model="claude-sonnet-5-5",
                    max_tokens=stage.get("max_tokens", 4096),
                    system=stage["system_prompt"],
                    messages=[
                        {"role": "user", "content": current_input},
                        {"role": "assistant", "content": output},
                        {"role": "user", "content": f"Validation failed: {error}. Fix and retry."},
                    ],
                )
                output = retry_response.content[0].text

        current_input = output

    return current_input

# Usage: plan -> implement -> review -> document
result = run_pipeline(
    task="Add rate limiting to the /api/generate endpoint",
    stages=[
        {
            "system_prompt": "You are an architect. Break this into implementation steps with file paths and code changes needed.",
            "validator": lambda x: (True, None) if "##" in x else (False, "Output must contain markdown headers for each step"),
        },
        {
            "system_prompt": "You are a senior developer. Implement each step from the plan. Output complete, working code.",
            "max_tokens": 8192,
        },
        {
            "system_prompt": "You are a code reviewer. Review for bugs, security issues, and edge cases. Output the corrected code with inline comments explaining changes.",
            "max_tokens": 8192,
        },
        {
            "system_prompt": "You are a technical writer. Write clear documentation for this feature: what it does, configuration options, and usage examples.",
        },
    ],
)

// Hierarchical delegation with dynamic task assignment
interface SubAgent {
  name: string;
  capabilities: string[];
  systemPrompt: string;
}

interface Task {
  id: string;
  description: string;
  requiredCapabilities: string[];
  dependencies: string[];
  status: "pending" | "running" | "complete" | "failed";
  result?: string;
}

class Supervisor {
  private agents: SubAgent[];
  private tasks: Map<string, Task> = new Map();
  private results: Map<string, string> = new Map();

  constructor(agents: SubAgent[]) {
    this.agents = agents;
  }

  async decompose(goal: string): Promise<Task[]> {
    const response = await client.messages.create({
      model: "claude-sonnet-5-5",
      max_tokens: 4096,
      system: `You are a project manager. Decompose goals into tasks.
        Available specialists: ${this.agents.map((a) => `${a.name} (${a.capabilities.join(", ")})`).join("; ")}
        Output JSON: { "tasks": [{ "id": "t1", "description": "...", "requiredCapabilities": ["..."], "dependencies": [] }] }`,
      messages: [{ role: "user", content: goal }],
    });

    const parsed = JSON.parse(response.content[0].type === "text" ? response.content[0].text : "{}");
    return parsed.tasks;
  }

  findBestAgent(task: Task): SubAgent | undefined {
    return this.agents.find((agent) =>
      task.requiredCapabilities.every((cap) => agent.capabilities.includes(cap))
    );
  }

  async execute(goal: string): Promise<string> {
    const tasks = await this.decompose(goal);
    tasks.forEach((t) => this.tasks.set(t.id, { ...t, status: "pending" }));

    while ([...this.tasks.values()].some((t) => t.status === "pending")) {
      // Find tasks whose dependencies are all complete
      const ready = [...this.tasks.values()].filter(
        (t) =>
          t.status === "pending" &&
          t.dependencies.every((dep) => this.tasks.get(dep)?.status === "complete")
      );

      // Execute ready tasks in parallel
      await Promise.all(
        ready.map(async (task) => {
          const agent = this.findBestAgent(task);
          if (!agent) {
            task.status = "failed";
            return;
          }
          task.status = "running";

          // Include dependency results as context
          const context = task.dependencies
            .map((dep) => `Result of ${dep}: ${this.results.get(dep)}`)
            .join("\n");

          const response = await client.messages.create({
            model: "claude-sonnet-5-5",
            max_tokens: 4096,
            system: agent.systemPrompt,
            messages: [
              {
                role: "user",
                content: `${task.description}\n\nContext from previous tasks:\n${context}`,
              },
            ],
          });

          const result = response.content[0].type === "text" ? response.content[0].text : "";
          this.results.set(task.id, result);
          task.status = "complete";
        })
      );
    }

    return [...this.results.values()].join("\n\n---\n\n");
  }
}

// Blackboard pattern with change-triggered agents
interface BlackboardState {
  [key: string]: {
    value: any;
    lastUpdatedBy: string;
    version: number;
  };
}

type AgentTrigger = {
  agent: SubAgent;
  watchKeys: string[];
  handler: (state: BlackboardState, changedKey: string) => Promise<Partial<BlackboardState>>;
};

class Blackboard {
  private state: BlackboardState = {};
  private triggers: AgentTrigger[] = [];
  private maxIterations: number;

  constructor(maxIterations = 10) {
    this.maxIterations = maxIterations;
  }

  register(trigger: AgentTrigger) {
    this.triggers.push(trigger);
  }

  async write(key: string, value: any, author: string) {
    const current = this.state[key];
    this.state[key] = {
      value,
      lastUpdatedBy: author,
      version: (current?.version ?? 0) + 1,
    };

    // Fire triggers for agents watching this key
    const watchers = this.triggers.filter(
      (t) => t.watchKeys.includes(key) && t.agent.name !== author
    );

    for (const watcher of watchers) {
      const updates = await watcher.handler(this.state, key);
      for (const [k, v] of Object.entries(updates)) {
        await this.write(k, v, watcher.agent.name);
      }
    }
  }

  getState(): BlackboardState {
    return structuredClone(this.state);
  }
}

// Usage: code review cycle
const board = new Blackboard(5);

board.register({
  agent: { name: "implementer", capabilities: ["code"], systemPrompt: "..." },
  watchKeys: ["review_feedback"],
  handler: async (state, changedKey) => {
    // Read feedback, produce revised code
    const feedback = state["review_feedback"].value;
    const currentCode = state["code"].value;
    // ... call LLM to revise code based on feedback
    return { code: revisedCode };
  },
});

board.register({
  agent: { name: "reviewer", capabilities: ["review"], systemPrompt: "..." },
  watchKeys: ["code"],
  handler: async (state, changedKey) => {
    const code = state["code"].value;
    // ... call LLM to review code
    return { review_feedback: feedback };
  },
});

# Handoff pattern using OpenAI Agents SDK
from agents import Agent, Runner

# Define specialists
frontend_agent = Agent(
    name="Frontend Specialist",
    instructions="You handle React, CSS, and browser-side issues. Hand off to the backend specialist for API or database problems.",
)

backend_agent = Agent(
    name="Backend Specialist",
    instructions="You handle API routes, database queries, and server logic. Hand off to the DevOps specialist for deployment or infrastructure problems.",
)

devops_agent = Agent(
    name="DevOps Specialist",
    instructions="You handle deployment, CI/CD, Docker, and infrastructure. Hand off to the frontend specialist if the issue is client-side.",
)

# handoffs takes Agent (or Handoff) objects, not names, so wire the cycle after all three exist
frontend_agent.handoffs = [backend_agent]
backend_agent.handoffs = [devops_agent]
devops_agent.handoffs = [frontend_agent]

# Triage agent decides the first specialist
triage_agent = Agent(
    name="Triage",
    instructions="Analyze the issue and hand off to the most appropriate specialist.",
    handoffs=[frontend_agent, backend_agent, devops_agent],
)

# Run - the SDK handles handoff routing automatically
result = await Runner.run(triage_agent, "The /api/users endpoint returns 500 but only in production")

// Consensus pattern: generate, evaluate, select
async function consensus(
  task: string,
  numCandidates: number = 3,
  evaluationCriteria: string
): Promise<{ winner: string; reasoning: string }> {
  // Generate N independent solutions
  const candidates = await Promise.all(
    Array.from({ length: numCandidates }, (_, i) =>
      client.messages.create({
        model: "claude-sonnet-5-5",
        max_tokens: 4096,
        system: `You are solution generator #${i + 1}. Solve the task independently. Do not hedge - commit to a specific approach.`,
        messages: [{ role: "user", content: task }],
      })
    )
  );

  const solutions = candidates.map((c, i) => ({
    id: i + 1,
    content: c.content[0].type === "text" ? c.content[0].text : "",
  }));

  // Judge evaluates all solutions
  const judgeInput = solutions
    .map((s) => `## Solution ${s.id}\n${s.content}`)
    .join("\n\n---\n\n");

  const judgment = await client.messages.create({
    model: "claude-sonnet-5-5",
    max_tokens: 4096,
    system: `You are an expert evaluator. Compare the solutions against these criteria: ${evaluationCriteria}. Select the best one or synthesize the strongest elements from multiple solutions. Output JSON: { "winner": "solution content", "reasoning": "why this is best" }`,
    messages: [{ role: "user", content: judgeInput }],
  });

  return JSON.parse(judgment.content[0].type === "text" ? judgment.content[0].text : "{}");
}

from crewai import Agent, Task, Crew, Process
from crewai.flow.flow import Flow, listen, start

# Define agents with roles
researcher = Agent(
    role="Senior Research Analyst",
    goal="Find comprehensive technical information about the given topic",
    backstory="You are a veteran technical researcher who values accuracy over speed.",
    tools=[docs_lookup],  # any CrewAI tool you have configured
    verbose=True,
)

writer = Agent(
    role="Technical Writer",
    goal="Transform research into clear, actionable documentation",
    backstory="You write for practitioners who want to build, not theorize.",
    verbose=True,
)

reviewer = Agent(
    role="Technical Editor",
    goal="Ensure accuracy, completeness, and clarity",
    backstory="You have reviewed thousands of technical documents and have zero tolerance for hand-waving.",
    verbose=True,
)

# Define tasks with dependencies
research_task = Task(
    description="Research {topic} comprehensively. Include version numbers, code examples, and known limitations.",
    expected_output="A structured research report with sections, code blocks, and source citations.",
    agent=researcher,
)

writing_task = Task(
    description="Write a technical guide based on the research. Target audience: senior developers.",
    expected_output="A 2000+ word guide with introduction, sections, code examples, and conclusion.",
    agent=writer,
    context=[research_task],  # Receives research output as context
)

review_task = Task(
    description="Review the guide for technical accuracy, completeness, and readability.",
    expected_output="Reviewed guide with corrections applied and editor notes.",
    agent=reviewer,
    context=[writing_task],
)

# Assemble and run
crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[research_task, writing_task, review_task],
    process=Process.sequential,  # or Process.hierarchical with a manager
    memory=True,  # Enable shared memory across agents
    planning=True,  # Enable planning agent for step-by-step execution
)

result = crew.kickoff(inputs={"topic": "WebSocket authentication patterns"})

class ContentPipeline(Flow):
    @start()
    def research_phase(self):
        research_crew = Crew(agents=[researcher], tasks=[research_task])
        self.state["research"] = research_crew.kickoff()

    @listen(research_phase)
    def writing_phase(self):
        if len(self.state["research"].raw) < 500:
            # Not enough research - send back for more
            return self.research_phase()
        writing_crew = Crew(agents=[writer], tasks=[writing_task])
        self.state["draft"] = writing_crew.kickoff()

    @listen(writing_phase)
    def review_phase(self):
        review_crew = Crew(agents=[reviewer], tasks=[review_task])
        self.state["final"] = review_crew.kickoff()

pipeline = ContentPipeline()
result = pipeline.kickoff()

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
from operator import add

class AgentState(TypedDict):
    task: str
    research: Annotated[list[str], add]
    code: str
    review: str
    final_output: str

# research_agent, code_agent and review_agent are any runnables that take {"messages": [...]}
def research_node(state: AgentState) -> dict:
    result = research_agent.invoke({"messages": [{"role": "user", "content": state["task"]}]})
    return {"research": [result["messages"][-1].content]}

def code_node(state: AgentState) -> dict:
    context = "\n".join(state["research"])
    result = code_agent.invoke({
        "messages": [{"role": "user", "content": f"Task: {state['task']}\nResearch: {context}"}]
    })
    return {"code": result["messages"][-1].content}

def review_node(state: AgentState) -> dict:
    result = review_agent.invoke({
        "messages": [{"role": "user", "content": f"Review this code:\n{state['code']}"}]
    })
    return {"review": result["messages"][-1].content}

def should_revise(state: AgentState) -> str:
    if "APPROVED" in state["review"]:
        return "finalize"
    return "code"  # Loop back for revision

# Build the graph
graph = StateGraph(AgentState)
graph.add_node("research", research_node)
graph.add_node("code", code_node)
graph.add_node("review", review_node)
graph.add_node("finalize", lambda s: {"final_output": s["code"]})

graph.add_edge(START, "research")
graph.add_edge("research", "code")
graph.add_edge("code", "review")
graph.add_conditional_edges("review", should_revise, {"finalize": "finalize", "code": "code"})
graph.add_edge("finalize", END)

app = graph.compile()
result = app.invoke({"task": "Build a rate limiter middleware for Express"})

import os
from autogen import ConversableAgent, GroupChat, GroupChatManager, LLMConfig

# Claude models need api_type "anthropic"; without it AG2 treats the config as OpenAI
llm_config = LLMConfig({
    "model": "claude-sonnet-5-5",
    "api_key": os.getenv("ANTHROPIC_API_KEY"),
    "api_type": "anthropic",
})

# Define conversational agents
planner = ConversableAgent(
    name="Planner",
    system_message="You break down complex tasks into actionable steps. Output numbered lists.",
    llm_config=llm_config,
)

coder = ConversableAgent(
    name="Coder",
    system_message="You write production-quality TypeScript. Always include error handling and types.",
    llm_config=llm_config,
)

critic = ConversableAgent(
    name="Critic",
    system_message="You review code for bugs, performance, and security. Be specific about issues.",
    llm_config=llm_config,
)

# Group chat with automatic speaker selection
group_chat = GroupChat(
    agents=[planner, coder, critic],
    messages=[],
    max_round=12,
    speaker_selection_method="auto",  # LLM decides who speaks next
)

manager = GroupChatManager(groupchat=group_chat, llm_config=llm_config)

# Kick off the conversation
planner.initiate_chat(
    manager,
    message="We need to add WebSocket support to our Express API with JWT authentication.",
)

from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

# Leaf agents - specialists
research_agent = Agent(
    name="researcher",
    model="gemini-3.8-flash",
    description="Researches a topic and returns structured findings.",
    instruction="Research the given topic thoroughly. Return structured findings.",
)

code_agent = Agent(
    name="coder",
    model="gemini-3.8-flash",
    description="Writes clean, tested code from a specification.",
    instruction="Write clean, tested code based on specifications.",
)

# Parent agent - coordinator. ADK picks a sub-agent from its description.
coordinator = Agent(
    name="coordinator",
    model="gemini-3.8-flash",
    instruction="""You coordinate a development team.
    Transfer research tasks to researcher.
    Transfer coding tasks to coder.
    Synthesize results into a final deliverable.""",
    sub_agents=[research_agent, code_agent],
)

# Run: the session must exist before the runner uses it, and new_message is a Content object
async def main():
    session_service = InMemorySessionService()
    await session_service.create_session(app_name="dev-team", user_id="dev", session_id="s1")
    runner = Runner(agent=coordinator, app_name="dev-team", session_service=session_service)
    message = types.Content(role="user", parts=[types.Part(text="Build a CLI tool for transcoding video files")])
    async for event in runner.run_async(user_id="dev", session_id="s1", new_message=message):
        if event.is_final_response():
            print(event.content.parts[0].text)

<!-- .claude/agents/researcher.md -->
---
name: researcher
description: Researches technical topics from documentation and the codebase
tools: WebFetch, Read, Grep, Glob
---

You are a technical research specialist. When given a topic:
1. Read the latest documentation and release notes
2. Find working code examples
3. Identify common pitfalls and known issues
4. Return structured findings with source URLs

<!-- .claude/agents/implementer.md -->
---
name: implementer
description: Writes production code based on specifications
tools: Read, Edit, Write, Bash
---

You are a senior developer. Write clean, typed, tested code.
Follow the project's existing patterns. Check CLAUDE.md for conventions.

User: "Add WebSocket support to the API with JWT auth"

Claude Code (orchestrator):
  -> Agent 1 (researcher): "Find current best practices for WebSocket + JWT in Express"
  -> Agent 2 (researcher): "Check our existing auth middleware implementation"
  -> Agent 3 (implementer): "Scaffold the WebSocket server module" (after agents 1-2)
  -> Agent 4 (implementer): "Write integration tests" (after agent 3)

// Production error handling pattern
async function resilientAgentCall(
  agent: SubAgent,
  input: string,
  maxRetries: number = 3
): Promise<string> {
  let lastError = "";

  for (let attempt = 0; attempt < maxRetries; attempt++) {
    try {
      const prompt = lastError
        ? `Previous attempt failed: ${lastError}\n\nOriginal task: ${input}`
        : input;

      const response = await client.messages.create({
        model: "claude-sonnet-5-5",
        max_tokens: 4096,
        system: agent.systemPrompt,
        messages: [{ role: "user", content: prompt }],
      });

      const output = response.content[0].type === "text" ? response.content[0].text : "";

      // Validate output structure
      if (agent.outputSchema) {
        agent.outputSchema.parse(JSON.parse(output));
      }

      return output;
    } catch (error) {
      lastError = error instanceof Error ? error.message : String(error);
    }
  }

  throw new Error(`Agent ${agent.name} failed after ${maxRetries} attempts: ${lastError}`);
}