在子代理架构中,一个中心主代理(通常称为监督器)通过将子代理作为工具调用来协调它们。主代理决定调用哪个子代理、提供什么输入以及如何组合结果。子代理是无状态的——它们不记得过去的交互,所有对话记忆都由主代理维护。这提供了上下文隔离:每次子代理调用都在干净的上下文窗口中工作,防止主对话中的上下文膨胀。
关键特征
- 集中控制:所有路由都通过主代理
- 无直接用户交互:子代理将结果返回给主代理,而不是用户(尽管您可以在子代理中使用中断来允许用户交互)
- 通过工具调用子代理:子代理通过工具被调用
- 并行执行:主代理可以在单轮中调用多个子代理
监督器 vs. 路由器:监督器代理(此模式)与路由器不同。监督器是一个维护对话上下文并动态决定在多轮中调用哪些子代理的完整代理。路由器通常是一个单一的分类步骤,在不维护持续对话状态的情况下将请求分派给代理。
何时使用
当您有多个不同领域(例如日历、电子邮件、CRM、数据库)、子代理不需要直接与用户对话,或者您想要集中式工作流程控制时,使用子代理模式。对于只有几个工具的更简单情况,请使用单个代理。
需要在子代理中进行用户交互吗? 虽然子代理通常将结果返回给主代理而不是直接与用户对话,但您可以在子代理中使用中断来暂停执行并收集用户输入。当子代理在继续之前需要澄清或批准时,这很有用。主代理保持编排器的角色,但子代理可以在任务中间从用户那里收集信息。
基本实现
核心机制是将子代理包装为可以由主代理调用的工具:
from langchain.tools import tool
from langchain.agents import create_agent
# 创建一个子代理
subagent = create_agent(model="anthropic:claude-sonnet-4-20250514", tools=[...])
# 将其包装为工具
@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 = create_agent(model="anthropic:claude-sonnet-4-20250514", tools=[call_research_agent])
教程:使用子代理构建个人助手
了解如何使用子代理模式构建个人助手,其中中心主代理(监督器)协调专门的工作代理。
设计决策
在实现子代理模式时,您需要做出几个关键的设计选择。此表总结了一选项——每个选项在下面的部分中有详细介绍。
| 决策 | 选项 |
|---|
| 同步 vs. 异步 | 同步(阻塞)vs. 异步(后台) |
| 工具模式 | 每个代理一个工具 vs. 单一分派工具 |
| 子代理规范 | 系统提示 vs. 枚举约束 vs. 基于工具的发现(仅限单一分派工具) |
| 子代理输入 | 仅查询 vs. 完整上下文 |
| 子代理输出 | 子代理结果 vs. 完整对话历史 |
同步 vs. 异步
子代理执行可以是同步的(阻塞)或异步的(后台)。您的选择取决于主代理是否需要结果才能继续。
| 模式 | 主代理行为 | 最佳场景 | 权衡 |
|---|
| 同步 | 等待子代理完成 | 主代理需要结果才能继续 | 简单,但阻塞对话 |
| 异步 | 在后台运行子代理时继续 | 独立任务,用户不应等待 | 响应性好,但更复杂 |
不要与 Python 的 async/await 混淆。这里,“async”意味着主代理启动一个后台作业(通常在单独的进程或服务中)并继续而不阻塞。
同步(默认)
默认情况下,子代理调用是同步的:主代理在继续之前等待每个子代理完成。当主代理的下一步行动依赖于子代理的结果时,使用同步。
何时使用同步:
- 主代理需要子代理的结果来制定响应
- 任务有顺序依赖(例如,获取数据 → 分析 → 响应)
- 子代理失败应该阻止主代理的响应
权衡:
- 简单实现——只需调用并等待
- 在所有子代理完成之前用户看不到响应
- 长时间运行的任务会冻结对话
当子代理的工作是独立的——主代理不需要结果就能继续与用户对话时,使用异步执行。主代理启动后台作业并保持响应性。
何时使用异步:
- 子代理工作独立于主对话流程
- 用户应该在工作进行时能够继续聊天
- 您想并行运行多个独立任务
三工具模式:
- 启动作业:启动后台任务,返回作业 ID
- 检查状态:返回当前状态(待处理、运行中、已完成、失败)
- 获取结果:检索完成的结果
处理作业完成: 当作业完成时,您的应用程序需要通知用户。一种方法是显示一个通知,点击后发送类似”检查 job_123 并总结结果”的 HumanMessage。
工具模式
有两种主要方式将子代理作为工具暴露:
| 模式 | 最佳场景 | 权衡 |
|---|
| 每个代理一个工具 | 对每个子代理的输入/输出进行细粒度控制 | 更多设置,但更多定制 |
| 单一分派工具 | 多个代理、分布式团队、约定优于配置 | 更简单的组合,每个代理的定制更少 |
每个代理一个工具
关键思想是将子代理包装为可以由主代理调用的工具:
from langchain.tools import tool
from langchain.agents import create_agent
# 创建一个子代理
subagent = create_agent(model="...", tools=[...])
# 将其包装为工具 #
@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 = create_agent(model="...", tools=[call_subagent])
主代理在决定任务匹配子代理描述时调用子代理工具,接收结果,并继续编排。请参阅上下文工程以进行细粒度控制。
单一分派工具
另一种方法使用单个参数化工具来为独立任务调用临时子代理。与每个代理一个工具方法不同,每个子代理被包装为单独的工具,这使用基于约定的方法和单个 task 工具:任务描述作为人类消息传递给子代理,子代理的最终消息作为工具结果返回。
在以下情况下使用此方法:您想在多个团队之间分布式开发代理、需要将复杂任务隔离到单独的上下文窗口、需要一种可扩展的方式来添加新代理而不修改协调器,或更喜欢约定优于定制。这种方法以代理组合的简单性和强大的上下文隔离换取上下文工程的灵活性。
关键特征:
- 单任务工具:一个参数化工具,可以通过名称调用任何已注册的子代理
- 基于约定的调用:按名称选择代理,任务作为人类消息传递,最终消息作为工具结果返回
- 团队分布:不同团队可以独立开发和部署代理
- 代理发现:子代理可以通过系统提示(列出可用代理)或通过渐进式披露(通过工具按需加载代理信息)发现
这种方法的一个有趣方面是,子代理可能与主代理具有完全相同的能力。在这种情况下,调用子代理实际上是关于上下文隔离作为主要原因——允许复杂的多步任务在隔离的上下文窗口中运行,而不会使主代理的对话历史膨胀。子代理自主完成其工作并只返回简洁的摘要,保持主线程专注和高效。
from langchain.tools import tool
from langchain.agents import create_agent
# 由不同团队开发的子代理
research_agent = create_agent(
model="gpt-4.1",
prompt="You are a research specialist..."
)
writer_agent = create_agent(
model="gpt-4.1",
prompt="You are a writing specialist..."
)
# 可用子代理的注册表
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_agent = create_agent(
model="gpt-4.1",
tools=[task],
system_prompt=(
"You coordinate specialized sub-agents. "
"Available: research (fact-finding), "
"writer (content creation). "
"Use the task tool to delegate work."
),
)
上下文工程
控制上下文如何在主代理及其子代理之间流动:
| 类别 | 目的 | 影响 |
|---|
| 子代理规范 | 确保在应该调用子代理时调用它们 | 主代理路由决策 |
| 子代理输入 | 确保子代理能够使用优化的上下文良好执行 | 子代理性能 |
| 子代理输出 | 确保监督器能够对子代理结果采取行动 | 主代理性能 |
另请参阅我们关于代理上下文工程的综合指南。
子代理规范
与子代理关联的名称和描述是主代理知道要调用哪些子代理的主要方式。这些是提示杠杆——仔细选择它们。
- 名称:主代理如何引用子代理。保持清晰且以行动为导向(例如,
research_agent、code_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 参数上添加枚举约束。这提供了类型安全性,并在工具模式中使可用代理明确。
何时使用:
- 您有少量固定的代理集(< 10个)
- 您想要类型安全和明确的代理名称
- 您更喜欢基于模式的验证而不是基于提示的指导
示例:
from enum import Enum
class AgentName(str, Enum):
RESEARCH = "research"
WRITER = "writer"
REVIEWER = "reviewer"
@tool
def task(
agent_name: AgentName, # 枚举约束
description: str
) -> str:
"""Launch an ephemeral subagent for a task."""
# ...
基于工具的发现
提供一个单独的工具(例如,list_agents 或 search_agents),主代理可以调用它来按需发现可用代理。这支持渐进式披露并支持动态注册表。
何时使用:
- 您有很多代理(> 10个)或不断增长的注册表
- 代理注册表经常更改或动态
- 您想减少提示大小和令牌使用
- 不同团队独立管理不同代理
示例:
@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."
)
子代理输入
自定义子代理接收什么上下文来执行其任务。通过从代理状态中提取,添加在静态提示中不切实际捕获的输入——完整的消息历史、先前结果或任务元数据。
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]):
# 应用任何需要的逻辑来将消息转换为合适的输入
subagent_input = some_logic(query, runtime.state["messages"])
result = subagent1.invoke({
"messages": subagent_input,
# 您也可以根据需要在此处传递其他状态键。
# 确保在主代理和子代理的状态模式中都定义这些。
# state schemas.
"example_state_key": runtime.state["example_state_key"]
})
return result["messages"][-1].content
子代理输出
自定义主代理接收什么内容,以便它能够做出好的决策。有两种策略:
- 提示子代理:精确指定应返回什么。一种常见的失败模式是子代理执行工具调用或推理,但不将结果包含在其最终消息中——提醒它监督器只能看到最终输出。
- 在代码中格式化:在返回之前调整或丰富响应。例如,使用
Command 除了最终文本外还传递特定状态键。
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={
# 从子代理传回其他状态
"example_state_key": result["example_state_key"],
"messages": [
ToolMessage(
content=result["messages"][-1].content,
tool_call_id=tool_call_id
)
]
})
检查点保存和状态检查
默认情况下,子代理使用继承的检查点保存器模式——每次调用以新鲜状态开始,支持中断,并安全地并行运行。如果您需要子代理在调用之间维护自己的持久对话历史,请使用 checkpointer=True(延续模式)编译它。请参阅子图持久化以获取模式完整比较。
因为子代理在工具函数内部被调用,LangGraph 无法静态发现它们。这意味着 get_state 与 subgraphs 将不会返回子代理状态。如果您需要读取嵌套图形状态(例如,在中断期间),请改为在自定义图形中的节点函数中调用子代理。请参阅子图持久化以获取有关每种模式如何影响状态可见性的详细信息。
通过 MCP 将这些文档连接到 Claude、VSCode 等,获取实时答案。