====== How to Build a Multi-Agent System ======

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.(([[https://gurusup.com/blog/best-multi-agent-frameworks-2026|Best Multi-Agent Frameworks 2026 - GuruSup]]))(([[https://www.adopt.ai/blog/multi-agent-frameworks|Multi-Agent Frameworks - Adopt AI]]))(([[https://towardsagenticai.com/building-multi-agent-systems-with-langgraph-a-practical-guide/|Building Multi-Agent Systems with LangGraph - Towards Agentic AI]]))(([[https://pub.towardsai.net/7-multi-agent-patterns-every-developer-needs-in-2026-and-how-to-pick-the-right-one-e8edcd99c96a|7 Multi-Agent Patterns for 2026 - Towards AI]]))

===== Architecture Overview =====

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.

<mermaid>
graph TD
    A[User Request] --> B[Orchestrator Agent]
    B --> C{Task Router}
    C -->|Research needed| D[Research Agent]
    C -->|Code needed| E[Coding Agent]
    C -->|Analysis needed| F[Analysis Agent]
    C -->|Review needed| G[Review Agent]
    D --> H[Shared State / Message Bus]
    E --> H
    F --> H
    G --> H
    H --> I[Result Aggregator]
    I --> B
    B --> J{All tasks complete?}
    J -->|No| C
    J -->|Yes| K[Final Response]
</mermaid>

===== Core Patterns =====

==== 1. Orchestrator-Worker Pattern ====

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).

==== 2. Handoff Pattern ====

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.(([[https://agentincome.io/blog/openai-agents-sdk-tutorial-2026/|OpenAI Agents SDK Tutorial - AgentIncome]]))

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

==== 3. Debate/Critique Pattern ====

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.

==== 4. Pipeline Pattern ====

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.

==== 5. Parallel Fan-Out/Fan-In ====

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

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

===== Approach 1: CrewAI (Hierarchical Crews) =====

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

<code python>
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)
</code>

===== Approach 2: LangGraph (Stateful Graph Orchestration) =====

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

<code python>
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"])
</code>

===== Approach 3: OpenAI Agents SDK (Handoffs) =====

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

<code python>
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)
</code>

===== Comparison: CrewAI vs LangGraph vs Agents SDK =====

^ Criteria ^ CrewAI ^ LangGraph ^ OpenAI Agents SDK ^
| Abstraction level | High (roles, goals) | Medium (graph nodes) | Low (agents + handoffs) |
| Learning curve | Low | Medium | Low |
| Orchestration | Built-in sequential/hierarchical | Custom graph edges | Handoff-based |
| State management | Crew memory | Typed state + checkpoints | Conversation context |
| Error recovery | Auto-retry on tasks | Custom via graph edges | Manual in instructions |
| Observability | Verbose logging | Graph visualization + tracing | Built-in tracing |
| LLM flexibility | Multi-provider | Multi-provider | OpenAI only |
| Best for | Rapid prototyping | Complex workflows | OpenAI-native apps |

===== Communication Protocols =====

==== Model Context Protocol (MCP) ====

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

==== Agent-to-Agent Protocol (A2A) ====

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.

==== Shared State vs Message Passing ====

^ Pattern ^ Shared State ^ Message Passing ^
| Coupling | Tight (agents share memory) | Loose (agents are independent) |
| Consistency | Easier to maintain | Eventual consistency |
| Scalability | Limited by shared store | Horizontally scalable |
| Debugging | Inspect shared state | Trace message flows |
| Best for | Small teams, fast iteration | Large systems, microservices |

===== Best Practices =====

  * **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

===== See Also =====

  * [[how_to_build_a_coding_agent|How to Build a Coding Agent]]
  * [[how_to_build_a_research_agent|How to Build a Research Agent]]
  * [[how_to_build_a_data_analysis_agent|How to Build a Data Analysis Agent]]

===== References =====