从 langgraph-supervisor 迁移

从 langgraph-supervisor 包迁移到使用 create_agent 和工具封装的子智能体的子智能体模式。

langgraph-supervisor 包不再积极维护。相反，请使用 子智能体 模式：主智能体通过作为 工具 调用它们来协调专业工作者。

本指南涵盖了从 create_supervisor 迁移到 @[create_agent] 的方法，包括使用 @[interrupt] 和外部 API 回调的设置。

从 langgraph-supervisor 迁移到 create_agent 子智能体

CopiedCopy prompt

变更摘要

langgraph-supervisor	推荐的替换
使用工作智能体作为图节点的 create_supervisor	@[create_agent] 使用作为 @[@tool] 函数包装的子智能体
用于消息历史记录的 output_mode	在工具包装器中格式化子智能体输出（参见 子智能体输出）
用于自定义路由的 create_handoff_tool	调用 subagent.invoke(...) 的自定义 @[@tool]
嵌套监督器（create_supervisor 个监督器）	作为 @[@tool] 包装的子智能体，调用其他子智能体

基本迁移

使用 langgraph-supervisor 时，工作智能体是图节点，监督器使用手交工具在它们之间路由：

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)

通过将每个工作者包装为主智能体上的工具，迁移到子智能体模式：

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(),
)

有关完整演练，请参阅 使用子智能体构建个人助理。

迁移中断和恢复流程

常见的 langgraph-supervisor 设置使用 @[interrupt] 在工作智能体的工具内部，以暂停执行，直到外部服务完成：

# 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

使用子智能体模式，相同的流程可以工作。@[interrupt] 在子智能体工具内部通过工具包装的 @[create_agent] 层级向上传播到最外层图。您的外部回调仍然可以使用 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)

中断传播的要求

为了使 @[interrupt] 能够通过嵌套的 @[create_agent] 层级冒泡，请遵循以下规则：

1. 仅编译最外层图并使用检查点器。 保留子智能体不使用 checkpointer=...，以便它们使用 每次调用持久化 并在运行时继承父智能体的检查点器。
2. 在 configurable 中传递 thread_id。 外部 invoke() 或 stream_events() 调用必须包含一个 thread_id，以便图可以检查点和恢复。

这些规则适用于任意嵌套设置。例如，一个自定义 @[StateGraph] 外层，一个中间 @[create_agent] 监督器，和一个内部 @[create_agent] 子智能体都遵循相同的机制：

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

当 preview_tool 调用 @[interrupt] 时，异常会通过两个 @[create_agent] 层级冒泡，并在最外层 @[StateGraph] 的调用结果中作为 __interrupt__ 显示。您现有的 Command(resume=result) 回调路径仍然可以正常工作。

有关中断如何通过子图传播的更多信息，请参阅 子图持久化：中断 和 检查点和状态检查。

何时使用自定义 StateGraph

当您需要混合确定性步骤和智能体步骤时，使用自定义 @[StateGraph]。例如，固定路由、验证或与 @[create_agent] 节点一起的外部 API 调用。

迁移嵌套监督器

langgraph-supervisor 通过编译监督器并将它们传递给另一个 create_supervisor 调用来支持多级层次结构。使用子智能体模式时，您有两个选项：

1. 将所有智能体扁平化为单个监督者，每个叶节点智能体对应一个工具。当每个工作智能体独立时，这是最简单的方法。
2. 在需要中间协调时嵌套工具调用。将一个中间层智能体（它本身是一个 @[create_agent]，拥有自己的子节点工具）作为顶层监督者的一个工具进行包装。

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(),
)

如果需要静态子图发现、按层级检查点命名空间或跨层级的共享状态键，请使用自定义 @[StateGraph] 和 子图节点。

迁移消息历史选项

create_supervisor 暴露 output_mode 来控制工作智能体的消息如何在对话历史中显示：

• full_history：包含工作智能体的所有消息。
• last_message：仅包含工作智能体的最终响应。

使用子节点模式时，在工具包装器中控制此行为。对于 last_message 行为，仅返回最终消息；对于 full_history 行为，返回格式化的完整对话摘要。参见 子节点输出 了解将额外状态传递回监督者的模式。

参见

• 子节点：模式概述和设计决策
• 使用子节点构建个人助理：分步监督者教程
• 使用子图：子图持久化、中断和状态检查
• 中断：暂停和恢复图执行
• LangGraph v1 迁移指南：从 create_react_agent 迁移到 create_agent

---

官方原文：src/oss/python/migrate/langgraph-supervisor.mdx · 非官方翻译，译文由 AI 生成，以英文原文为准。