LangGraph Multi-Agent Systems

Patterns for building multi-agent systems with LangGraph.

Architecture Overview

Supervisor Pattern

Central supervisor routes tasks to specialized agents.

Using langgraph-supervisor

pip install langgraph-supervisor

from langgraph_supervisor import create_supervisor
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")

# Create specialized agents
research_agent = create_react_agent(
    llm,
    tools=[search_tool, wiki_tool],
    name="researcher",
    prompt="You are a research specialist."
)

math_agent = create_react_agent(
    llm,
    tools=[calculator_tool],
    name="mathematician",
    prompt="You are a math expert."
)

# Create supervisor
workflow = create_supervisor(
    agents=[research_agent, math_agent],
    model=llm,
    prompt="Route tasks to the appropriate specialist."
)

app = workflow.compile()
result = app.invoke({"messages": [("user", "What is 25 * 17?")]})

Manual Supervisor Implementation

from langgraph.graph import StateGraph, START, END
from typing import Literal

class SupervisorState(TypedDict):
    messages: Annotated[list, add_messages]
    next_agent: str

def supervisor(state: SupervisorState) -> dict:
    # LLM decides which agent to call
    response = llm.invoke([
        SystemMessage("Route to 'researcher', 'coder', or 'FINISH'"),
        *state["messages"]
    ])
    return {"next_agent": response.content}

def route(state: SupervisorState) -> Literal["researcher", "coder", "__end__"]:
    if state["next_agent"] == "FINISH":
        return "__end__"
    return state["next_agent"]

graph = StateGraph(SupervisorState)
graph.add_node("supervisor", supervisor)
graph.add_node("researcher", research_agent)
graph.add_node("coder", coder_agent)

graph.add_edge(START, "supervisor")
graph.add_conditional_edges("supervisor", route)
graph.add_edge("researcher", "supervisor")
graph.add_edge("coder", "supervisor")

Swarm Pattern

Agents hand off control directly to peers.

Using langgraph-swarm

pip install langgraph-swarm

from langgraph_swarm import create_swarm, create_handoff_tool
from langgraph.prebuilt import create_react_agent

# Create handoff tools
handoff_to_sales = create_handoff_tool(
    agent_name="sales_agent",
    description="Transfer to sales for pricing questions"
)
handoff_to_support = create_handoff_tool(
    agent_name="support_agent",
    description="Transfer to support for technical issues"
)

# Create agents with handoff tools
sales_agent = create_react_agent(
    llm,
    tools=[pricing_tool, handoff_to_support],
    name="sales_agent",
    prompt="You handle sales and pricing."
)

support_agent = create_react_agent(
    llm,
    tools=[docs_tool, handoff_to_sales],
    name="support_agent",
    prompt="You handle technical support."
)

# Create swarm
workflow = create_swarm(
    agents=[sales_agent, support_agent],
    default_active_agent="sales_agent"
)

app = workflow.compile(checkpointer=checkpointer)

Custom Handoff Tool

from langchain.tools import tool
from langgraph.types import Command
from langgraph.prebuilt import InjectedState

@tool
def transfer_to_specialist(
    reason: str,
    state: Annotated[dict, InjectedState]
) -> Command:
    """Transfer conversation to specialist agent."""
    return Command(
        update={
            "messages": state["messages"],
            "transfer_reason": reason
        },
        goto="specialist_agent"
    )

Agent-as-Tool Pattern

Wrap subagents as tools for the main agent:

from langchain.tools import tool

# Create subagent
researcher = create_react_agent(llm, tools=[search_tool])

@tool
def call_researcher(query: str) -> str:
    """Delegate research tasks to the research specialist."""
    result = researcher.invoke({
        "messages": [("user", query)]
    })
    return result["messages"][-1].content

# Main agent uses subagent as tool
main_agent = create_react_agent(
    llm,
    tools=[call_researcher, other_tools],
    prompt="You are a coordinator. Delegate research to the researcher."
)

With State Propagation

from langgraph.prebuilt import InjectedToolCallId
from langchain.messages import ToolMessage
from langgraph.types import Command

@tool
def call_subagent(
    query: str,
    tool_call_id: Annotated[str, InjectedToolCallId]
) -> Command:
    """Call subagent and return result."""
    result = subagent.invoke({"messages": [("user", query)]})
    return Command(update={
        "messages": [ToolMessage(
            content=result["messages"][-1].content,
            tool_call_id=tool_call_id
        )]
    })

Context Engineering

Control what information each agent sees.

Message Filtering

def filter_messages_for_agent(messages: list, agent_name: str) -> list:
    """Filter messages relevant to specific agent."""
    return [
        m for m in messages
        if not hasattr(m, 'name') or m.name in [agent_name, 'user']
    ]

def specialist_node(state: State) -> dict:
    filtered = filter_messages_for_agent(state["messages"], "specialist")
    result = specialist_agent.invoke({"messages": filtered})
    return {"messages": result["messages"]}

Shared vs Private State

class MultiAgentState(TypedDict):
    # Shared across all agents
    messages: Annotated[list, add_messages]
    
    # Agent-specific (private)
    researcher_scratchpad: str
    coder_scratchpad: str

def researcher_node(state: MultiAgentState) -> dict:
    # Only reads/writes own scratchpad
    return {
        "messages": [...],
        "researcher_scratchpad": "notes..."
    }

Summary Handoff

Pass summaries instead of full history:

def create_summary_handoff(from_agent: str, to_agent: str):
    @tool
    def handoff(context_summary: str) -> Command:
        """Hand off with summary instead of full history."""
        return Command(
            update={
                "messages": [HumanMessage(
                    content=f"Context from {from_agent}: {context_summary}"
                )],
                "active_agent": to_agent
            },
            goto=to_agent
        )
    return handoff

Memory and Persistence

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore

checkpointer = InMemorySaver()  # Short-term (conversation)
store = InMemoryStore()          # Long-term (cross-session)

app = workflow.compile(
    checkpointer=checkpointer,
    store=store
)

# Thread per conversation
config = {"configurable": {"thread_id": f"user_{user_id}"}}
result = app.invoke(inputs, config)

Choosing Architecture

Use Supervisor when:

• Clear hierarchy needed
• Consistent routing logic required
• Working with third-party agents

Use Swarm when:

• Latency is critical
• Agents are peers
• Dynamic, organic routing

Key Points

• Supervisor: central control, higher latency, simpler logic
• Swarm: peer-to-peer, lower latency, explicit handoffs
• Agent-as-tool: simple delegation, isolated execution
• Filter context to avoid overwhelming agents
• Use checkpointer for multi-turn conversations
• Consider token usage when designing message flow