A multi-agent system (MAS) coordinates multiple specialized AI agents to solve complex tasks that exceed the capabilities of any single agent. Frameworks like CrewAI, AutoGen, LangGraph, and the OpenAI Agents SDK provide different approaches to agent orchestration, message passing, and shared state. Gartner predicts multi-agent architectures will power 33% of enterprise AI applications by 2028. This guide covers the patterns, tradeoffs, and working code for building your own.

Multi-agent systems use an orchestrator to route tasks to specialist agents. The orchestrator manages workflow, handles failures, and merges results. Communication happens through structured message passing or shared state.

A central orchestrator agent plans the workflow and delegates sub-tasks to worker agents. The orchestrator decides task order, handles dependencies, and merges results.

When to use: Complex workflows with clear sub-task decomposition (report generation, code review pipelines).

Agents transfer control to each other based on context. Agent A detects it needs Agent B and “hands off” the conversation with full context. Used by the OpenAI Agents SDK.

When to use: Customer service routing, escalation workflows, specialist consultation.

Multiple agents independently solve the same problem, then a judge agent evaluates and selects or synthesizes the best answer.

When to use: High-stakes decisions, creative tasks, verification workflows.

Agents execute in sequence, each transforming the output of the previous one. Like a Unix pipeline for AI.

When to use: Document processing, ETL workflows, content generation pipelines.

The orchestrator spawns multiple agents simultaneously, then merges their outputs.

When to use: Independent analysis tasks, multi-perspective research, batch processing.

CrewAI provides a high-level abstraction with roles, goals, and tasks. It handles sequencing, context passing, and error retries automatically.

from crewai import Agent, Task, Crew, Process
from crewai.tools import tool
import requests
 
@tool("Web Search")
def web_search(query: str) -> str:
    """Search the web for information."""
    import os
    response = requests.post(
        "https://api.tavily.com/search",
        json={"api_key": os.environ["TAVILY_API_KEY"], "query": query, "max_results": 5},
    )
    results = response.json().get("results", [])
    return "\n".join(f"- {r['title']}: {r['content'][:200]}" for r in results)
 
@tool("Write File")
def write_file(path: str, content: str) -> str:
    """Write content to a file."""
    with open(path, "w") as f:
        f.write(content)
    return f"Wrote {len(content)} chars to {path}"
 
# Define specialist agents
researcher = Agent(
    role="Senior Research Analyst",
    goal="Find comprehensive, accurate information on the given topic",
    backstory="Expert at web research with 10 years of experience in tech analysis.",
    tools=[web_search],
    verbose=True,
)
 
writer = Agent(
    role="Technical Writer",
    goal="Write clear, well-structured technical content based on research",
    backstory="Experienced technical writer who excels at making complex topics accessible.",
    tools=[write_file],
    verbose=True,
)
 
reviewer = Agent(
    role="Quality Reviewer",
    goal="Review content for accuracy, completeness, and clarity",
    backstory="Detail-oriented editor with expertise in technical accuracy.",
    verbose=True,
)
 
# Define tasks with dependencies
research_task = Task(
    description="Research the topic: {topic}. Find key facts, trends, and expert opinions.",
    expected_output="Detailed research notes with sources",
    agent=researcher,
)
 
writing_task = Task(
    description="Write a comprehensive article based on the research. Include code examples where relevant.",
    expected_output="Well-structured article in markdown format",
    agent=writer,
    context=[research_task],  # Depends on research output
)
 
review_task = Task(
    description="Review the article for accuracy and completeness. Suggest improvements.",
    expected_output="Review notes and final approved article",
    agent=reviewer,
    context=[writing_task],
)
 
# Assemble and run the crew
crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[research_task, writing_task, review_task],
    process=Process.sequential,  # Or Process.hierarchical for dynamic routing
    verbose=True,
)
 
result = crew.kickoff(inputs={"topic": "Building AI agents in 2026"})
print(result)

LangGraph models multi-agent workflows as directed graphs with typed state. This gives you explicit control over routing, loops, and error recovery.

import json, operator
from typing import Annotated, Literal, TypedDict
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
 
llm = ChatOpenAI(model="gpt-4o")
 
class TeamState(TypedDict):
    task: str
    plan: list[dict]
    current_step: int
    results: Annotated[list[dict], operator.add]
    final_output: str
    error_count: int
 
# --- Orchestrator: creates and manages the plan ---
def orchestrator(state: TeamState) -> TeamState:
    """Plan the workflow by decomposing the task."""
    response = llm.invoke(
        f"You are a project orchestrator. Decompose this task into steps, "
        f"each assigned to a specialist (researcher, coder, reviewer).\n"
        f"Task: {state['task']}\n"
        f"Return JSON: [{{\"agent\": \"researcher\", \"subtask\": \"...\"}}]"
    )
    plan = json.loads(response.content)
    return {"plan": plan, "current_step": 0, "error_count": 0}
 
# --- Specialist agents ---
def researcher(state: TeamState) -> TeamState:
    step = state["plan"][state["current_step"]]
    response = llm.invoke(
        f"You are a research specialist. Complete this task:\n{step['subtask']}\n"
        f"Previous results: {json.dumps(state['results'][-3:])}"
    )
    return {
        "results": [{"agent": "researcher", "output": response.content}],
        "current_step": state["current_step"] + 1,
    }
 
def coder(state: TeamState) -> TeamState:
    step = state["plan"][state["current_step"]]
    context = "\n".join(r["output"][:500] for r in state["results"][-3:])
    response = llm.invoke(
        f"You are a coding specialist. Complete this task:\n{step['subtask']}\n"
        f"Context from team: {context}"
    )
    return {
        "results": [{"agent": "coder", "output": response.content}],
        "current_step": state["current_step"] + 1,
    }
 
def reviewer(state: TeamState) -> TeamState:
    step = state["plan"][state["current_step"]]
    all_work = "\n\n".join(r["output"][:500] for r in state["results"])
    response = llm.invoke(
        f"You are a quality reviewer. Review the team output:\n{all_work}\n"
        f"Task was: {step['subtask']}\nProvide feedback or approval."
    )
    return {
        "results": [{"agent": "reviewer", "output": response.content}],
        "current_step": state["current_step"] + 1,
    }
 
def synthesizer(state: TeamState) -> TeamState:
    all_results = "\n\n".join(
        f"[{r['agent']}]: {r['output']}" for r in state["results"]
    )
    response = llm.invoke(
        f"Synthesize all team outputs into a final deliverable:\n{all_results}"
    )
    return {"final_output": response.content}
 
# --- Router ---
def route_to_agent(state: TeamState) -> str:
    if state["current_step"] >= len(state["plan"]):
        return "synthesize"
    agent_type = state["plan"][state["current_step"]]["agent"]
    return agent_type
 
# Build the graph
workflow = StateGraph(TeamState)
workflow.add_node("orchestrator", orchestrator)
workflow.add_node("researcher", researcher)
workflow.add_node("coder", coder)
workflow.add_node("reviewer", reviewer)
workflow.add_node("synthesize", synthesizer)
 
workflow.set_entry_point("orchestrator")
workflow.add_conditional_edges("orchestrator", route_to_agent, {
    "researcher": "researcher",
    "coder": "coder",
    "reviewer": "reviewer",
    "synthesize": "synthesize",
})
# After each specialist, route to next step
for node in ["researcher", "coder", "reviewer"]:
    workflow.add_conditional_edges(node, route_to_agent, {
        "researcher": "researcher",
        "coder": "coder",
        "reviewer": "reviewer",
        "synthesize": "synthesize",
    })
workflow.add_edge("synthesize", END)
 
app = workflow.compile()
 
result = app.invoke({
    "task": "Build a REST API for user management with authentication",
    "plan": [],
    "current_step": 0,
    "results": [],
    "final_output": "",
    "error_count": 0,
})
print(result["final_output"])

The OpenAI Agents SDK uses a handoff pattern where agents transfer control to each other.

from agents import Agent, Runner
 
research_agent = Agent(
    name="Researcher",
    instructions="You research topics thoroughly. When done, hand off to the Writer.",
    handoff_description="Handles research tasks",
)
 
writing_agent = Agent(
    name="Writer",
    instructions="You write polished content from research notes. When done, hand off to Reviewer.",
    handoff_description="Handles writing tasks",
)
 
review_agent = Agent(
    name="Reviewer",
    instructions="You review content for accuracy. Provide the final approved version.",
    handoff_description="Handles quality review",
)
 
# Wire up handoffs
research_agent.handoffs = [writing_agent]
writing_agent.handoffs = [review_agent]
 
# Orchestrator delegates to specialists
orchestrator = Agent(
    name="Orchestrator",
    instructions=(
        "You coordinate a team. Analyze the task and hand off to the "
        "appropriate specialist. Start with Researcher for new topics."
    ),
    handoffs=[research_agent, writing_agent, review_agent],
)
 
result = Runner.run_sync(orchestrator, "Write a technical brief on WebAssembly trends in 2026")
print(result.final_output)

MCP standardizes how agents access tools and external services. Any MCP-compatible tool works with any MCP-compatible agent.

Google's A2A protocol (2025) enables cross-vendor agent communication. Agents publish “Agent Cards” describing their capabilities, and other agents can discover and invoke them.

• Start simple: Begin with 2-3 agents before scaling to larger teams
• Single responsibility: Each agent should do one thing well
• Explicit handoffs: Clearly define when and why agents transfer control
• Shared context: Ensure all agents have access to relevant prior work
• Error budgets: Set retry limits per agent and for the overall workflow
• Human-in-the-loop: Add approval gates for high-stakes decisions
• Observability: Log all agent interactions for debugging and optimization
• Cost tracking: Monitor token usage per agent to optimize expensive operations

• Best Multi-Agent Frameworks 2026 - GuruSup
• Multi-Agent Frameworks - Adopt AI
• OpenAI Agents SDK Tutorial - AgentIncome
• Building Multi-Agent Systems with LangGraph - Towards Agentic AI
• 7 Multi-Agent Patterns for 2026 - Towards AI

• How to Build a Coding Agent
• How to Build a Research Agent
• How to Build a Data Analysis Agent