Skip to main content

Migrate from langgraph-supervisor

还没翻译译自英文(docs.langchain.com)本页 45 段都还没译

The langgraph-supervisor package is no longer actively maintained. Instead use the subagents pattern: a main agent coordinates specialized workers by calling them as tools.

This guide covers how to migrate from create_supervisor to create_agent, including setups that use interrupt and external API callbacks.

Migrate from langgraph-supervisor to create_agent subagents

Migrate this codebase from the langgraph-supervisor package to the LangChain subagents pattern with create_agent and tool-wrapped workers.

Step 1: Read the guide

Fetch and follow https://docs.langchain.com/oss/python/migrate/langgraph-supervisor.md as the source of truth for before/after patterns, interrupt propagation rules, nested supervisors, and message history options. For the target pattern, also fetch https://docs.langchain.com/oss/python/langchain/multi-agent/subagents.md and linked pages when needed. If the codebase still uses create_react_agent, also fetch https://docs.langchain.com/oss/python/migrate/langgraph-v1.md.

Step 2: Update dependencies

Remove langgraph-supervisor with the package manager already used in this project (uv or pip). Ensure LangChain and LangGraph versions in the project support create_agent, @tool, and interrupt as shown in the guide. Prefer pinning current stable versions when the project already pins versions.

Step 3: Apply the migration

Identify every call site that imports from langgraph_supervisor or uses create_supervisor, create_handoff_tool, or supervisor nesting, then apply the mappings from the guide, including:

  • Replace create_supervisor with a main create_agent whose tools wrap each worker subagent.
  • Replace handoff routing (create_handoff_tool) with custom @tool functions that call subagent.invoke(...).
  • Recreate output_mode behavior (full_history vs last_message) inside each tool wrapper rather than as a supervisor constructor argument.
  • Migrate nested supervisors by flattening to one supervisor with one tool per leaf agent, or by nesting tool-wrapped middle-tier agents when intermediate coordination is required.

Step 4: Preserve interrupt and resume flows

If workers use interrupt or external callback resume with Command(resume=...), keep that behavior and follow the guide's propagation rules: compile only the outermost graph with a checkpointer, leave subagents without their own checkpointer, and pass thread_id in configurable on outer invoke or stream_events calls.

Step 5: Handle gaps explicitly

If the codebase needs static subgraph discovery, checkpoint namespaces per tier, shared state keys between levels, or a mix of deterministic and agentic steps that the guide says belong in a custom StateGraph, stop and ask the user how to proceed. Do not invent a custom graph layout unless the guide covers it.

Rules

  • Stay scoped to this migration. Do not rewrite unrelated agent code.
  • Prefer the subagents pattern from the guide over keeping langgraph-supervisor.
  • Ask rather than guess when a call site or pattern is not covered by the guide.

Summary of changes

langgraph-supervisorRecommended replacement
create_supervisor with worker agents as graph nodescreate_agent with subagents wrapped as @tool functions
output_mode for message historyFormat subagent output in the tool wrapper (see subagent outputs)
create_handoff_tool for custom routingCustom @tool that calls subagent.invoke(...)
Nested supervisors (create_supervisor of supervisors)A subagent wrapped as a @tool that calls other subagents

Basic migration

With langgraph-supervisor, worker agents were graph nodes and the supervisor routes between them using handoff tools:

from langgraph_supervisor import create_supervisor
from langgraph.prebuilt import create_react_agent

research_agent = create_react_agent(
    model=model,
    tools=[web_search],
    name="research_expert",
    prompt="You are a research expert.",
)

math_agent = create_react_agent(
    model=model,
    tools=[add, multiply],
    name="math_expert",
    prompt="You are a math expert.",
)

workflow = create_supervisor(
    [research_agent, math_agent],
    model=model,
    prompt="Route research questions to research_expert and math to math_expert.",
)
app = workflow.compile(checkpointer=checkpointer)

Migrate to the subagents pattern by wrapping each worker as a tool on a main agent:

from langchain.agents import create_agent
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver

research_agent = create_agent(
    model=model,
    tools=[web_search],
    system_prompt="You are a research expert.",
)

math_agent = create_agent(
    model=model,
    tools=[add, multiply],
    system_prompt="You are a math expert.",
)


@tool("research_expert", description="Research expert for current events and web lookups.")
def call_research_agent(query: str) -> str:
    result = research_agent.invoke({"messages": [{"role": "user", "content": query}]})
    return result["messages"][-1].content


@tool("math_expert", description="Math expert for calculations.")
def call_math_agent(query: str) -> str:
    result = math_agent.invoke({"messages": [{"role": "user", "content": query}]})
    return result["messages"][-1].content


supervisor = create_agent(
    model=model,
    tools=[call_research_agent, call_math_agent],
    system_prompt=(
        "Route research questions to research_expert and math to math_expert."
    ),
    checkpointer=InMemorySaver(),
)

For a full walkthrough, see Build a personal assistant with subagents.

Migrate interrupt and resume flows

A common langgraph-supervisor setup uses interrupt inside a worker agent's tool to pause execution until an external service completes:

# Before: create_supervisor with a subgraph node
#
# Supervisor (create_supervisor)
#   └── ResearchAgent (subgraph node)
#         └── preview_tool
#               ├── fire_external_api()      # kicks off async job
#               ├── result = interrupt(...)  # pauses graph, waits for callback
#               └── render_results(result)   # runs after resume

With the subagents pattern, the same flow works. interrupt inside a subagent tool propagates up through tool-wrapped create_agent layers to the outermost graph. Your external callback can still resume with Command(resume=result).

from langchain.agents import create_agent
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt


@tool
def preview_tool(document_id: str) -> str:
    """Run an async enrichment preview and wait for results."""
    job_id = fire_external_api(document_id)
    result = interrupt({"job_id": job_id, "status": "pending"})
    return render_results(result)


research_agent = create_agent(
    model=model,
    tools=[preview_tool],
    system_prompt="You are a research agent.",
)

@tool("research_agent", description="Research and enrichment tasks.")
def call_research_agent(query: str) -> str:
    result = research_agent.invoke({"messages": [{"role": "user", "content": query}]})
    return result["messages"][-1].content

supervisor = create_agent(
    model=model,
    tools=[call_research_agent],
    system_prompt="Delegate research tasks to research_agent.",
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "1"}}
from langgraph.types import Command

# Invoke — preview_tool calls interrupt() and the graph pauses
response = supervisor.invoke(
    {"messages": [{"role": "user", "content": "Preview enrichment for doc-123"}]},
    config=config,
)
# response contains __interrupt__

# External service completes and calls back into your app
supervisor.invoke(Command(resume=external_result), config=config)

Requirements for interrupt propagation

For interrupt to bubble up through nested create_agent layers, follow these rules:

  1. Compile only the outermost graph with a checkpointer. Leave subagents without checkpointer=... so they use per-invocation persistence and inherit the parent's checkpointer at runtime.
  2. Pass thread_id in configurable. The outer invoke() or stream_events() call must include a thread_id so the graph can checkpoint and resume.

These rules apply to arbitrarily nested setups. For example, a custom StateGraph outer layer, a middle create_agent supervisor, and an inner create_agent subagent all follow the same mechanism:

Custom StateGraph (outer, with checkpointer)
  └── prospecting_agent (create_agent, no checkpointer)
        └── call_powerup_agent tool → powerup_agent.invoke(...)
              └── powerup_agent (create_agent, no checkpointer)
                    └── preview_tool → interrupt(...)

When preview_tool calls interrupt, the exception bubbles through both create_agent layers and surfaces as __interrupt__ on the outer StateGraph's invoke result. Your existing Command(resume=result) callback path keeps working.

For more on how interrupts propagate through subgraphs, see Subgraph persistence: Interrupts and Checkpointing and state inspection.

When to use a custom StateGraph instead

Use a custom StateGraph when you need to mix deterministic steps with agentic ones. For example, fixed routing, validation, or external API calls alongside create_agent nodes.

Migrate nested supervisors

langgraph-supervisor supports multi-level hierarchies by compiling supervisors and passing them to another create_supervisor call. With the subagents pattern, you have two options:

  1. Flatten to a single supervisor with one tool per leaf agent. This is the simplest approach when each worker is independent.
  2. Nest tool calls when you need intermediate coordination. Wrap a middle-tier agent (itself a create_agent with its own subagent tools) as a tool on the top-level supervisor.
from langchain.agents import create_agent
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver

# Middle-tier agent with its own subagents
billing_team = create_agent(
    model=model,
    tools=[call_refunds_agent, call_invoices_agent],
    system_prompt="Coordinate billing specialists.",
)

@tool("billing_team", description="Handle billing, refunds, and invoices.")
def call_billing_team(query: str) -> str:
    result = billing_team.invoke({"messages": [{"role": "user", "content": query}]})
    return result["messages"][-1].content

# Top-level supervisor
top_supervisor = create_agent(
    model=model,
    tools=[call_billing_team, call_support_agent],
    system_prompt="Route billing to billing_team and general support to support_agent.",
    checkpointer=InMemorySaver(),
)

If you need static subgraph discovery, checkpoint namespaces per tier, or shared state keys between levels, use a custom StateGraph with subgraph nodes instead.

Migrate message history options

create_supervisor exposes output_mode to control how worker messages appear in conversation history:

  • full_history: Include all messages from the worker agent.
  • last_message: Include only the worker's final response.

With the subagents pattern, control this in the tool wrapper. Return only the final message for last_message behavior, or return a formatted summary of the full conversation for full_history behavior. See Subagent outputs for patterns that pass additional state back to the supervisor.

See also