langgraph-swarm

The Swarm Researcher uses a two-phase collaborative pattern designed for complex research and planning tasks. This pattern separates high-level strategy from low-level execution to improve thoroughness.

1. Planning Phase

The planner agent acts as the initial entry point. Its responsibilities include:

• Analyzing the user request.
• Reading relevant documentation.
• Asking clarifying questions to refine the scope.
• Creating a structured plan with clear objectives.
• Identifying resources for the implementation phase.
• Handing off control to the researcher agent.

2. Research Phase

The researcher agent takes over after the handoff. Its responsibilities include:

• Following the structured plan provided by the planner.
• Reading the recommended documentation sources.
• Implementing the solution to satisfy requirements.
• Requesting additional planning if the scope changes or requirements are unclear.

Use create_agent to define specialized agents with specific tools and system prompts. Use create_handoff_tool to allow agents to transfer control to one another. Finally, use create_swarm to bundle the agents into a workflow and .compile() it with a checkpointer to enable conversation memory.

from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent
from langgraph_swarm import create_handoff_tool, create_swarm

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

def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b

alice = create_agent(
    model,
    tools=[
        add,
        create_handoff_tool(
            agent_name="Bob",
            description="Transfer to Bob",
        ),
    ],
    system_prompt="You are Alice, an addition expert.",
    name="Alice",
)

bob = create_agent(
    model,
    tools=[
        create_handoff_tool(
            agent_name="Alice",
            description="Transfer to Alice, she can help with math",
        ),
    ],
    system_prompt="You are Bob, you speak like a pirate.",
    name="Bob",
)

checkpointer = InMemorySaver()
workflow = create_swarm(
    [alice, bob],
    default_active_agent="Alice"
)
app = workflow.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "1"}}
turn_1 = app.invoke(
    {"messages": [{"role": "user", "content": "i'd like to speak to Bob"}]},
    config,
)
print(turn_1)
turn_2 = app.invoke(
    {"messages": [{"role": "user", "content": "what's 5 + 7?"}]},
    config,
)
print(turn_2)

While create_handoff_tool is the default, you can implement custom handoff tools using langgraph.types.Command. This allows you to:

• Change tool names and descriptions.
• Add extra arguments for the LLM to populate (e.g., a task_description for the next agent).
• Modify the data passed during handoff (e.g., passing specific state keys instead of the full message history).

Requirements for custom tools returning Command:

1. The agent must have a tool-calling node (like ToolNode) that handles Command objects.
2. Both the swarm graph and the target agent graph must share a state schema containing the keys you intend to update in Command.update.

from typing import Annotated

from langchain.tools import tool, BaseTool, InjectedToolCallId
from langchain.messages import ToolMessage
from langgraph.types import Command
from langgraph.prebuilt import InjectedState

def create_custom_handoff_tool(*, agent_name: str, name: str | None, description: str | None) -> BaseTool:

    @tool(name, description=description)
    def handoff_to_agent(
        # you can add additional tool call arguments for the LLM to populate
        # for example, you can ask the LLM to populate a task description for the next agent
        task_description: Annotated[str, "Detailed description of what the next agent should do, including all of the relevant context."],
        # you can inject the state of the agent that is calling the tool
        state: Annotated[dict, InjectedState],
        tool_call_id: Annotated[str, InjectedToolCallId],
    ):
        tool_message = ToolMessage(
            content=f"Successfully transferred to {agent_name}",
            name=name,
            tool_call_id=tool_call_id,
        )
        # you can use a different messages state key here, if your agent uses a different schema
        # e.g., "alice_messages" instead of "messages"
        messages = state["messages"]
        return Command(
            goto=agent_name,
            graph=Command.PARENT,
            # NOTE: this is a state update that will be applied to the swarm multi-agent graph (i.e., the PARENT graph)
            update={
                "messages": messages + [tool_message],
                "active_agent": agent_name,
                # optionally pass the task description to the next agent
                "task_description": task_description,
            },
        )

    return handoff_to_agent

By default, all agents share a single messages key in the swarm state. To prevent agents from seeing each other's full history, you can use custom state schemas (e.g., alice_messages) and manually manage the handoff using add_active_agent_router.

Steps to implement:

1. Define a custom TypedDict for the agent's state.
2. Create a wrapper function that transforms the SwarmState into the agent's specific state and back again.
3. Manually build the StateGraph using add_node and add_active_agent_router instead of create_swarm.

from typing_extensions import TypedDict, Annotated

from langchain.messages import AnyMessage
from langgraph.graph import StateGraph, add_messages
from langgraph_swarm import SwarmState

class AliceState(TypedDict):
    alice_messages: Annotated[list[AnyMessage], add_messages]

# see this guide to learn how you can implement a custom tool-calling agent
# https://langchain-ai.github.io/langgraph/how-tos/react-agent-from-scratch/
alice = (
    StateGraph(AliceState)
    .add_node("model", ...)
    .add_node("tools", ...)
    .add_edge(...)
    ...
    .compile()
)

# wrapper calling the agent
def call_alice(state: SwarmState):
    # you can put any input transformation from parent state -> agent state
    # for example, you can invoke "alice" with "task_description" populated by the LLM
    response = alice.invoke({"alice_messages": state["messages"]})
    # you can put any output transformation from agent state -> parent state
    return {"messages": response["alice_messages"]}

def call_bob(state: SwarmState):
    ...

from langgraph_swarm import add_active_agent_router

workflow = (
    StateGraph(SwarmState)
    .add_node("Alice", call_alice, destinations=("Bob",))
    .add_node("Bob", call_bob, destinations=("Alice",))
)
# this is the router that enables us to keep track of the last active agent
workflow = add_active_agent_router(
    builder=workflow,
    route_to=["Alice", "Bob"],
    default_active_agent="Alice",
)

# compile the workflow
app = workflow.compile()

In a LangGraph swarm, handoffs are implemented as tools that return a Command object.

When an LLM invokes a tool created by create_handoff_tool:

1. State Injection: The tool uses InjectedState to access the current graph state and InjectedToolCallId to track the specific tool call.
2. Command Execution: Instead of returning a simple string, the tool returns a Command(goto=agent_name, graph=Command.PARENT, update=...).
3. Navigation: The goto parameter tells LangGraph which node to move to next. The graph=Command.PARENT parameter is critical for multi-agent architectures where agents might be sub-graphs; it ensures the handoff signal propagates correctly to the parent orchestrator.
4. State Update: The update dictionary modifies the graph state, specifically appending a ToolMessage to the messages list and updating an active_agent key to reflect the new agent in control.