subagents 架构中,中心主代理(通常称为 supervisor)会通过将子代理作为工具调用来协调它们。主代理决定调用哪个子代理、提供什么输入,以及如何合并结果。子代理是无状态的,它们不会记住过去的交互,所有对话记忆都由主代理维护。这提供了上下文隔离:每次子代理调用都在干净的上下文窗口中工作,防止主对话中的上下文膨胀。 如需内置子代理支持,请参阅 Deep Agents

关键特征

  • 集中控制:所有路由都经过主代理
  • 无直接用户交互:子代理将结果返回给主代理,而不是用户(不过你可以在子代理中使用 interrupts 来允许用户交互)
  • 通过工具调用子代理:子代理通过工具被调用
  • 并行执行:主代理可以在单轮中调用多个子代理
Supervisor 与 Router:supervisor 代理(本模式)不同于 router。supervisor 是完整代理,会维护对话上下文,并跨多轮动态决定调用哪些子代理。router 通常是单个分类步骤,它将工作分派给代理,但不维护持续的对话状态。

何时使用

当你有多个不同领域(例如日历、电子邮件、CRM、数据库)、子代理不需要直接与用户对话,或你希望集中控制工作流时,请使用 subagents 模式。对于只有少量工具的简单场景,请使用单个代理
需要在子代理内进行用户交互? 虽然子代理通常会将结果返回给主代理,而不是直接与用户对话,但你可以在子代理中使用 interrupts 暂停执行并收集用户输入。当子代理在继续前需要澄清或批准时,这很有用。主代理仍然是编排者,但子代理可以在任务中途向用户收集信息。

基本实现

核心机制是将子代理封装为主代理可以调用的工具:
from langchain.tools import tool
from langchain.agents import create_agent

# Create a subagent
subagent = create_agent(model="google_genai:gemini-3.5-flash", tools=[...])

# Wrap it as a tool
@tool("research", description="Research a topic and return findings")
def call_research_agent(query: str):
    result = subagent.invoke({"messages": [{"role": "user", "content": query}]})
    return result["messages"][-1].content

# Main agent with subagent as a tool
main_agent = create_agent(model="google_genai:gemini-3.5-flash", tools=[call_research_agent])

教程:使用子代理构建个人助手

学习如何使用 subagents 模式构建个人助手,其中中心主代理(supervisor)会协调专门工作代理。

设计决策

实现 subagents 模式时,你需要做出几个关键设计选择。下表总结了这些选项,后续小节会详细介绍每一项。
决策选项
同步与异步同步(阻塞)与异步(后台)
工具模式每个代理一个工具与单个分派工具
子代理规格系统提示词、枚举约束、基于工具的发现(仅适用于单个分派工具)
子代理输入仅查询与完整上下文
子代理输出子代理结果与完整对话历史

同步与异步

子代理执行可以是同步(阻塞)或异步(后台)。你的选择取决于主代理是否需要结果才能继续。
模式主代理行为最适合权衡
同步等待子代理完成主代理需要结果才能继续简单,但会阻塞对话
异步子代理在后台运行时继续执行独立任务,用户不应等待响应更快,但更复杂
不要与 Python 的 async/await 混淆。这里的“异步”表示主代理启动一个后台作业(通常在单独进程或服务中),然后在不阻塞的情况下继续。

同步(默认)

默认情况下,子代理调用是同步的:主代理会等待每个子代理完成后再继续。当主代理的下一步操作依赖子代理结果时,请使用同步。 何时使用同步:
  • 主代理需要子代理结果来组织响应
  • 任务存在顺序依赖(例如获取数据 → 分析 → 响应)
  • 子代理失败应阻止主代理响应
权衡:
  • 实现简单,只需调用并等待
  • 在所有子代理完成之前,用户看不到响应
  • 长时间运行的任务会冻结对话

异步

当子代理工作是独立的,即主代理不需要其结果也能继续与用户对话时,请使用异步执行。主代理会启动后台作业,并保持响应。 何时使用异步:
  • 子代理工作独立于主对话流程
  • 用户应能在工作进行时继续聊天
  • 你希望并行运行多个独立任务
三工具模式:
  1. 启动作业:启动后台任务并返回作业 ID
  2. 检查状态:返回当前状态(pending、running、completed、failed)
  3. 获取结果:检索已完成结果
处理作业完成: 作业完成后,应用需要通知用户。一种做法是显示通知,点击后发送类似 “Check job_123 and summarize the results.” 的 HumanMessage

工具模式

将子代理暴露为工具主要有两种方式:
模式最适合权衡
每个代理一个工具细粒度控制每个子代理的输入和输出设置更多,但自定义能力更强
单个分派工具代理很多、团队分布式开发、约定优于配置组合更简单,但每个代理的自定义较少

每个代理一个工具

关键思路是将子代理封装为主代理可以调用的工具:
from langchain.tools import tool
from langchain.agents import create_agent

# Create a sub-agent
subagent = create_agent(model="...", tools=[...])

# Wrap it as a tool  #
@tool("subagent_name", description="subagent_description")
def call_subagent(query: str):
    result = subagent.invoke({"messages": [{"role": "user", "content": query}]})
    return result["messages"][-1].content

# Main agent with subagent as a tool  #
main_agent = create_agent(model="...", tools=[call_subagent])
当主代理判断任务与子代理描述匹配时,会调用子代理工具、接收结果并继续编排。如需细粒度控制,请参阅上下文工程

单个分派工具

另一种方法是使用单个参数化工具,为独立任务调用临时子代理。不同于每个代理一个工具的方法(每个子代理都被封装为单独工具),这种方法使用基于约定的单个 task 工具:任务描述作为 human message 传给子代理,子代理的最终消息作为工具结果返回。 当你希望将代理开发分散到多个团队、需要将复杂任务隔离到单独上下文窗口、需要一种可扩展方式在不修改协调器的情况下添加新代理,或偏好约定优于自定义时,请使用此方法。该方法以牺牲上下文工程灵活性为代价,换来代理组合的简单性和强上下文隔离。 关键特征:
  • 单个任务工具:一个参数化工具,可以按名称调用任何已注册子代理
  • 基于约定的调用:按名称选择代理,将任务作为 human message 传入,并将最终消息作为工具结果返回
  • 团队分工:不同团队可以独立开发和部署代理
  • 代理发现:可以通过系统提示词(列出可用代理)或渐进式披露(通过工具按需加载代理信息)发现子代理
这种方法的一个有趣之处是,子代理可能拥有与主代理完全相同的能力。在这种情况下,调用子代理的主要原因其实是上下文隔离,让复杂的多步骤任务在隔离的上下文窗口中运行,而不膨胀主代理的对话历史。子代理自主完成工作,并仅返回简洁摘要,从而保持主线程聚焦且高效。
from langchain.tools import tool
from langchain.agents import create_agent

# Sub-agents developed by different teams
research_agent = create_agent(
    model="gpt-5.4",
    prompt="You are a research specialist..."
)

writer_agent = create_agent(
    model="gpt-5.4",
    prompt="You are a writing specialist..."
)

# Registry of available sub-agents
SUBAGENTS = {
    "research": research_agent,
    "writer": writer_agent,
}

@tool
def task(
    agent_name: str,
    description: str
) -> str:
    """Launch an ephemeral subagent for a task.

    Available agents:
    - research: Research and fact-finding
    - writer: Content creation and editing
    """
    agent = SUBAGENTS[agent_name]
    result = agent.invoke({
        "messages": [
            {"role": "user", "content": description}
        ]
    })
    return result["messages"][-1].content

# Main coordinator agent
main_agent = create_agent(
    model="gpt-5.4",
    tools=[task],
    system_prompt=(
        "You coordinate specialized sub-agents. "
        "Available: research (fact-finding), "
        "writer (content creation). "
        "Use the task tool to delegate work."
    ),
)

上下文工程

控制上下文如何在主代理和子代理之间流动:
类别目的影响
子代理规格确保子代理在应被调用时被调用主代理路由决策
子代理输入确保子代理能在优化后的上下文中良好执行子代理性能
子代理输出确保 supervisor 能基于子代理结果采取行动主代理性能
另请参阅面向代理的 context engineering 综合指南。

子代理规格

与子代理关联的名称描述是主代理了解应调用哪些子代理的主要方式。这些都是提示词控制手段,请谨慎选择。
  • 名称:主代理引用子代理的方式。保持清晰且面向动作(例如 research_agentcode_reviewer)。
  • 描述:主代理了解的子代理能力。请具体说明它处理哪些任务,以及何时使用它。
对于单个分派工具设计,还必须向主代理提供它可以调用的子代理信息。 你可以根据代理数量以及注册表是静态还是动态,用不同方式提供这些信息:
方法最适合权衡
系统提示词枚举小型、静态代理列表(少于 10 个代理)简单,但代理变化时需要更新提示词
枚举约束小型、静态代理列表(少于 10 个代理)类型安全且显式,但代理变化时需要修改代码
基于工具的发现大型或动态代理注册表灵活且可扩展,但会增加复杂性

系统提示词枚举

在主代理的系统提示词中直接列出可用代理。主代理会将代理列表及其描述作为指令的一部分看到。 何时使用:
  • 代理集合较小且固定(少于 10 个)
  • 代理注册表很少变化
  • 你想要最简单的实现
示例:
main_agent = create_agent(
    model="...",
    tools=[task],
    system_prompt=(
        "You coordinate specialized sub-agents. "
        "Available agents:\n"
        "- research: Research and fact-finding\n"
        "- writer: Content creation and editing\n"
        "- reviewer: Code and document review\n"
        "Use the task tool to delegate work."
    ),
)

分派工具上的枚举约束

向分派工具的 agent_name 参数添加枚举约束。这会提供类型安全,并在工具 schema 中显式列出可用代理。 何时使用:
  • 代理集合较小且固定(少于 10 个)
  • 你想要类型安全和显式代理名称
  • 你更偏好基于 schema 的验证,而不是基于提示词的引导
示例:
from enum import Enum

class AgentName(str, Enum):
    RESEARCH = "research"
    WRITER = "writer"
    REVIEWER = "reviewer"

@tool
def task(
    agent_name: AgentName,  # Enum constraint
    description: str
) -> str:
    """Launch an ephemeral subagent for a task."""
    # ...

基于工具的发现

提供一个主代理可调用的独立工具(例如 list_agentssearch_agents),用于按需发现可用代理。这支持渐进式披露,并支持动态注册表。 何时使用:
  • 代理很多(超过 10 个)或注册表持续增长
  • 代理注册表频繁变化或是动态的
  • 你希望减少提示词大小和 token 使用量
  • 不同团队独立管理不同代理
示例:
@tool
def list_agents(query: str = "") -> str:
    """List available subagents, optionally filtered by query."""
    agents = search_agent_registry(query)
    return format_agent_list(agents)

@tool
def task(agent_name: str, description: str) -> str:
    """Launch an ephemeral subagent for a task."""
    # ...

main_agent = create_agent(
    model="...",
    tools=[task, list_agents],
    system_prompt="Use list_agents to discover available subagents, then use task to invoke them."
)

子代理输入

自定义子代理执行任务时接收的上下文。可以从代理状态中提取那些不适合放入静态提示词的输入,例如完整消息历史、先前结果或任务元数据。
Subagent inputs example
from langchain.agents import AgentState
from langchain.tools import tool, ToolRuntime

class CustomState(AgentState):
    example_state_key: str

@tool(
    "subagent1_name",
    description="subagent1_description"
)
def call_subagent1(query: str, runtime: ToolRuntime[None, CustomState]):
    # Apply any logic needed to transform the messages into a suitable input
    subagent_input = some_logic(query, runtime.state["messages"])
    result = subagent1.invoke({
        "messages": subagent_input,
        # You could also pass other state keys here as needed.
        # Make sure to define these in both the main and subagent's
        # state schemas.
        "example_state_key": runtime.state["example_state_key"]
    })
    return result["messages"][-1].content

子代理输出

自定义主代理收到的返回内容,以便它做出良好决策。两种策略:
  1. 提示子代理:明确指定应返回什么。常见失败模式是子代理执行了工具调用或推理,但没有在最终消息中包含结果,请提醒它 supervisor 只能看到最终输出。
  2. 在代码中格式化:返回前调整或丰富响应。例如,使用 Command 在最终文本之外传回特定状态键。
Subagent outputs example
from typing import Annotated
from langchain.agents import AgentState
from langchain.tools import InjectedToolCallId
from langgraph.types import Command


@tool(
    "subagent1_name",
    description="subagent1_description"
)
def call_subagent1(
    query: str,
    tool_call_id: Annotated[str, InjectedToolCallId],
) -> Command:
    result = subagent1.invoke({
        "messages": [{"role": "user", "content": query}]
    })
    return Command(update={
        # Pass back additional state from the subagent
        "example_state_key": result["example_state_key"],
        "messages": [
            ToolMessage(
                content=result["messages"][-1].content,
                tool_call_id=tool_call_id
            )
        ]
    })

Checkpointing 和状态检查

默认情况下,子代理使用 inherited checkpointer 模式,每次调用都会从全新状态开始,支持 interrupts,并且可以安全并行运行。如果需要子代理跨调用维护自己的持久对话历史,请使用 checkpointer=True(continuations 模式)进行编译。如需完整模式对比,请参阅子图持久化 由于子代理是在工具函数内部被调用的,LangGraph 无法静态发现它们。这意味着带 subgraphsget_state 不会返回子代理状态。如果需要读取嵌套图状态(例如在 interrupt 期间),请改为在自定义图中通过节点函数调用子代理。关于每种模式如何影响状态可见性,请参阅子图持久化