以编程方式使用文档

langgraph-supervisor 包不再积极维护。请改用 子代理 模式:主代理通过将专门的 worker 作为 工具.

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

变更摘要

langgraph-supervisor推荐替代方案
create_supervisor 使用 worker 代理作为图节点create_agent 使用包装为 @tool 函数
output_mode 用于消息历史在工具包装器中格式化子代理输出(见 子代理输出)
create_handoff_tool 用于自定义路由自定义 @tool 调用 subagent.invoke(...)
嵌套 supervisor(create_supervisor 个 supervisor)作为 @tool 包装的调用其他子代理的子代理

基础迁移

使用 langgraph-supervisor,worker 代理是图节点,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)

通过将每个 worker 包装为工具来迁移到子代理模式:

有关完整演练,请参阅 使用子代理构建个人助手.

迁移中断和恢复流程

常见的 langgraph-supervisor 设置使用 interrupt 在 worker 代理的工具中暂停执行,直到外部服务完成:

# 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 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 编译最外层的图。** 让子代理没有 checkpointer=... 这样它们使用 每次调用的持久化 并在运行时继承父级的 checkpointer。
  2. **传递 thread_id in configurable.** 外部 invoke() or stream_events() 调用必须包含 thread_id 以便图可以 checkpoint 并恢复。

这些规则适用于任意嵌套的设置。例如,一个自定义 StateGraph 外层、一个中间 create_agent supervisor,以及一个内部 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 层并作为 __interrupt__ 在外层 StateGraph 的调用结果上。您现有的 Command(resume=result) 回调路径继续工作。

有关中断如何通过子图传播的更多信息,请参阅 子图持久化:中断检查点与状态检查.

何时使用自定义 StateGraph

当需要混合确定性步骤与代理步骤时,使用自定义 StateGraph。例如,固定路由、验证或外部 API 调用与 create_agent 节点配合使用。

迁移嵌套监督器

langgraph-supervisor 通过编译监督器并将其传递给另一个 create_supervisor 调用来支持多层级层次结构。使用子代理模式时,有两个选项:

  1. **展平为单个监督器** 每个叶代理一个工具。当每个工作单元独立时,这是最简单的方法。
  2. **嵌套工具调用** 当你需要中间协调时。将一个中层代理(本身是一个 create_agent,拥有自己的子代理工具)包装为顶级监督器上的一个工具。

如果需要静态子图发现、每层的检查点命名空间或跨层级共享状态键,请使用带有StateGraph的自定义 @ [子图节点 instead.

迁移消息历史选项

create_supervisor 暴露 output_mode 来控制工作单元消息在对话历史中的显示方式:

  • - full_history:包含工作单元代理的所有消息。
  • - last_message:仅包含工作单元的最终响应。

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

另请参阅