BytesRouter

主题

LangChain 文档版本 跟随上游 源快照 de5fc235 更新于 2026-10-08

从 langgraph-supervisor 迁移

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

langgraph-supervisor 包已不再积极维护。请改用子智能体模式：主智能体通过将专门的工作器作为工具调用来协调它们。

本指南介绍如何从 create_supervisor 迁移到 create_agent，包括使用 interrupt 和外部 API 回调的设置。

将此代码库从 langgraph-supervisor 包迁移到使用 create_agent 和以工具封装的工作器的 LangChain 子智能体模式。

第 1 步：阅读指南

获取并遵循 https://www.bytesrouter.com/langchain/oss/python/migrate/langgraph-supervisor 作为前后对比模式、中断传播规则、嵌套主管和消息历史选项的权威来源。对于目标模式，如需还可获取 https://www.bytesrouter.com/langchain/oss/python/langchain/multi-agent/subagents 及关联页面。如果代码库仍在使用 create_react_agent，还应获取 https://www.bytesrouter.com/langchain/oss/python/migrate/langgraph-v1。

步骤 2：更新依赖项

使用此项目中已使用的包管理器（uv 或 pip）移除 langgraph-supervisor。确保项目中的 LangChain 和 LangGraph 版本支持指南中所示的 create_agent、@tool 和 interrupt。当项目已经固定版本时，优先固定当前稳定版本。

步骤 3：应用迁移

找出所有从 langgraph_supervisor 导入或使用 create_supervisor、create_handoff_tool 或主管嵌套的调用点，然后应用指南中的映射，包括：

  • 将 create_supervisor 替换为一个主 create_agent，其工具包装每个工作子智能体。
  • 将交接路由（create_handoff_tool）替换为调用 subagent.invoke(...) 的自定义 @tool 函数。
  • 在每个工具包装器内重建 output_mode 行为（full_history 与 last_message），而不是将其作为主管构造函数参数。
  • 通过扁平化为一个主管、每个叶子智能体一个工具，或在需要中间协调时将工具包装的中间层智能体嵌套起来，来迁移嵌套的主管。

步骤 4：保留中断和恢复流程

如果工作智能体使用 interrupt 或通过 Command(resume=...) 进行外部回调恢复，请保留该行为，并遵循指南的传播规则：仅使用检查点器编译最外层的图，让子智能体不带自己的检查点器，并在外层 invoke 或 stream_events 调用的 configurable 中传入 thread_id。

步骤 5：显式处理缺口

如果代码库需要静态子图发现、每层级的检查点命名空间、层级之间共享的状态键，或指南说明应放在自定义 StateGraph 中的确定性与智能体步骤的混合，请停下来询问用户如何继续。除非指南涵盖，否则不要自行发明自定义图布局。

规则

  • 将范围限定在此次迁移内。不要重写无关的智能体代码。
  • 优先使用指南中的子智能体模式，而不是保留 langgraph-supervisor。
  • 当某调用点或模式未在指南中涵盖时，请询问而不是猜测。

变更摘要

langgraph-supervisor推荐的替代方案
以工作智能体作为图节点的 create_supervisor将子智能体包装为 @tool 函数的 create_agent
用于消息历史的 output_mode在工具包装器中格式化子智能体输出（参见 子智能体输出）
用于自定义路由的 create_handoff_tool调用 subagent.invoke(...) 的自定义 @tool
嵌套的主管（主管的 create_supervisor）一个包装为 @tool 并调用其他子智能体的子智能体

基本迁移

使用 langgraph-supervisor 时，工作智能体是图节点，主管使用交接工具在它们之间路由：

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

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

python
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 来暂停执行，直到外部服务完成：

python
# 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) 恢复。

python
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"}}
python
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 子智能体，都遵循相同的机制：

txt
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 层，并作为 __interrupt__ 出现在外层 StateGraph 的 invoke 结果上。你现有的 Command(resume=result) 回调路径会继续工作。

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

何时改用自定义 StateGraph

当你需要将确定性步骤与智能体步骤混合时，请使用自定义 StateGraph。例如，在 create_agent 节点之外搭配固定路由、校验或外部 API 调用。

迁移嵌套的主管

langgraph-supervisor 通过编译主管并将其传递给另一个 create_supervisor 调用来支持多层级层次结构。在子智能体模式下，你有两个选择：

  1. 扁平化为单个主管，每个叶子智能体一个工具。当每个工作智能体相互独立时，这是最简单的方法。
  2. 嵌套工具调用，当你需要中间协调时。将一个中间层智能体（它本身是一个带有自己子智能体工具的 create_agent）包装为顶层主管上的工具。
python
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 行为，返回完整对话的格式化摘要。关于将额外状态传回主管的模式，请参阅 子智能体输出。

另请参阅

友情链接

  • Agents.md面向 AI 编码助手的仓库说明文件规范
  • LangChainLangChain 是一个开放的Agent平台
  • LangSmithAgent 工程平台：可观测与评估、部署、Fleet、Engine 等
  • EmbabelEmbabel 是一个JVM平台Agent框架
  • SpringAISpringAI 是一个基于 Spring 的 AI 框架
  • MCPModel Context Protocol，连接 AI 应用与外部数据/工具的开放标准
  • LangChain4jJava 生态的 LLM 应用开发框架
  • LlamaIndex把数据接入 LLM 的框架与文档平台
  • LlamaParse智能体式 OCR 文档解析：把 PDF、扫描件变成 LLM 可用的文本与结构化数据
  • LiteParse本地优先、轻量的文档解析库，不需要 API key
  • CrewAI多智能体编排框架：以 Crew/Flow 组织自主协作的 AI Agent
  • MAFMicrosoft Agent Framework，企业级 .NET/Python 智能体框架

BytesRouter 是社区翻译站点，所有译文仅供参考，一切以官方英文原文为准。

© 2026 BytesRouter · 译文采用 CC BY 4.0 许可