外观
在子智能体架构中,一个居中的主 智能体(通常称为监督者)通过将子智能体作为 工具 调用来协调它们。主智能体决定调用哪个子智能体、提供什么输入,以及如何组合结果。子智能体是无状态的——它们不记住过去的交互,所有对话记忆都由主智能体维护。这提供了 上下文 隔离:每次子智能体调用都在干净的上下文窗口中运行,防止主对话中出现上下文膨胀。
如需内置的子智能体支持,请参阅 Deep Agents。
关键特征
- 集中化控制:所有路由都经由主智能体
- 不直接与用户交互:子智能体将结果返回给主智能体,而不是用户(不过你可以在子智能体内使用 中断 来允许用户交互)
- 通过工具调用子智能体:子智能体通过工具被调用
- 并行执行:主智能体可以在单轮中调用多个子智能体
INFO
监督者 vs. 路由器:监督智能体(本模式)与 路由器 不同。监督者是完整的智能体,它维护对话上下文,并跨多轮动态决定调用哪些子智能体。路由器通常是一个分类步骤,它分派到智能体,但不维护持续的对话状态。
何时使用
当你拥有多个不同的领域(如日历、电子邮件、CRM、数据库)、子智能体不需要直接与用户对话,或者你想要集中化的工作流控制时,请使用子智能体模式。对于只有少量 工具 的更简单情况,请使用 单个智能体。
TIP
需要在子智能体内进行用户交互? 虽然子智能体通常将结果返回给主智能体而不是直接与用户对话,但你可以在子智能体内使用 中断 来暂停执行并收集用户输入。当子智能体在继续之前需要澄清或批准时,这很有用。主智能体仍然是编排者,但子智能体可以在任务中途从用户那里收集信息。
基本实现
核心机制是将子智能体包装为主智能体可以调用的工具:
python
from langchain.tools import tool
from langchain.agents import create_agent
# 创建子智能体
subagent = create_agent(model="google_genai:gemini-3.6-flash", 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="google_genai:gemini-3.6-flash", tools=[call_research_agent])typescript
import { createAgent, tool } from "langchain";
import { z } from "zod";
// 创建子智能体
const subagent = createAgent({ model: "google_genai:gemini-3.6-flash", tools: [...] });
// 将其包装为工具
const callResearchAgent = tool(
async ({ query }) => {
const result = await subagent.invoke({
messages: [{ role: "user", content: query }]
});
return result.messages.at(-1)?.content;
},
{
name: "research",
description: "Research a topic and return findings",
schema: z.object({ query: z.string() })
}
);
// 主智能体将子智能体作为工具
const mainAgent = createAgent({ model: "google_genai:gemini-3.6-flash", tools: [callResearchAgent] });- 教程:构建带子智能体的个人助手 — 了解如何使用子智能体模式构建个人助手,其中居中的主智能体(监督者)协调专门的 worker 智能体。
设计决策
实现子智能体模式时,你需要做出几个关键的设计选择。此表格总结了各选项——每一项都在下面的小节中详细说明。
| 决策 | 选项 |
|---|---|
| 同步 vs. 异步 | 同步(阻塞)vs. 异步(后台) |
| 工具模式 | 每智能体一个工具 vs. 单一分派工具 |
| 子智能体规格 | 系统提示词 vs. 枚举约束 vs. 基于工具的发现(仅限单一分派工具) |
| 子智能体输入 | 仅查询 vs. 完整上下文 |
| 子智能体输出 | 子智能体结果 vs. 完整对话历史 |
同步 vs. 异步
子智能体执行可以是同步(阻塞)或异步(后台)的。你的选择取决于主智能体是否需要该结果才能继续。
| 模式 | 主智能体行为 | 最适用 | 权衡 |
|---|---|---|---|
| 同步 | 等待子智能体完成 | 主智能体需要结果才能继续 | 简单,但会阻塞对话 |
| 异步 | 在子智能体于后台运行时继续 | 独立任务,用户不应等待 | 响应迅速,但更复杂 |
TIP
不要与 Python 的 async/await 混淆。这里的"异步"指的是主智能体启动一个后台作业(通常是在单独的进程或服务中),然后不阻塞地继续运行。
同步(默认)
默认情况下,子智能体调用是同步的:主智能体等待每个子智能体完成后再继续。当主智能体的下一步动作依赖于子智能体的结果时,请使用同步。
何时使用同步:
- 主智能体需要子智能体的结果来构建其响应
- 任务有顺序依赖(如获取数据 → 分析 → 响应)
- 子智能体失败应阻塞主智能体的响应
权衡:
- 实现简单——只需调用并等待
- 在所有子智能体完成之前,用户看不到任何响应
- 长时间运行的任务会冻结对话
异步
当子智能体的工作独立时,使用异步执行——主智能体不需要结果即可继续与用户对话。主智能体启动一个后台作业并保持响应。
何时使用异步:
- 子智能体的工作独立于主对话流程
- 用户应该能够在工作运行的同时继续聊天
- 你想要并行运行多个独立任务
三工具模式:
- 启动作业:启动后台任务,返回作业 ID
- 检查状态:返回当前状态(pending、running、completed、failed)
- 获取结果:检索已完成的结果
处理作业完成: 当作业完成时,你的应用需要通知用户。一种方法:显示一个通知,点击后发送一条 HumanMessage,如"Check job_123 and summarize the results."
工具模式
将子智能体暴露为工具有两种主要方式:
| 模式 | 最适用 | 权衡 |
|---|---|---|
| 每智能体一个工具 | 对每个子智能体的输入/输出进行细粒度控制 | 设置更多,但可定制性更强 |
| 单一分派工具 | 智能体众多、分布式团队、约定优于配置 | 组合更简单,每个智能体的定制化更少 |
每智能体一个工具
关键思想是将子智能体包装为主智能体可以调用的工具:
python
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]) typescript
import { createAgent, tool } from "langchain";
import * as z from "zod";
// 创建子智能体
const subagent = createAgent({...});
// 将其包装为工具
const callSubagent = tool(
async ({ query }) => {
const result = await subagent.invoke({
messages: [{ role: "user", content: query }]
});
return result.messages.at(-1)?.text;
},
{
name: "subagent_name",
description: "subagent_description",
schema: z.object({
query: z.string().describe("The query to send to subagent")
})
}
);
// 主智能体将子智能体作为工具
const mainAgent = createAgent({ model, tools: [callSubagent] }); 当主智能体判断任务与子智能体的描述匹配时,它会调用子智能体工具,接收结果,并继续编排。有关细粒度控制,请参阅 上下文工程。
单一分派工具
另一种方法使用单一的参数化工具来为独立任务调用临时子智能体。与 每智能体一个工具 方法(每个子智能体都被包装为单独的工具)不同,这种方法使用基于约定的单一 task 工具:任务描述作为人类消息传递给子智能体,子智能体的最后一条消息作为工具结果返回。
当你希望将智能体开发分布到多个团队、需要将复杂任务隔离到单独的上下文窗口中、需要一种无需修改协调器即可添加新智能体的可扩展方式,或者更倾向于约定优于定制时,请使用这种方法。这种方法以上下文工程的灵活性换取了智能体组合的简单性和强大的上下文隔离。
关键特征:
- 单一任务工具:一个参数化工具,可以按名称调用任何已注册的子智能体
- 基于约定的调用:按名称选择智能体,任务作为人类消息传递,最后一条消息作为工具结果返回
- 团队分工:不同团队可以独立开发和部署智能体
- 智能体发现:子智能体可以通过系统提示词(列出可用智能体)或通过 渐进式披露(通过工具按需加载智能体信息)被发现
TIP
这种方法的一个有趣方面是,子智能体可能具有与主智能体完全相同的能力。在这种情况下,调用子智能体本质上主要是为了上下文隔离——允许复杂、多步骤的任务在隔离的上下文窗口中运行,而不会使主智能体的对话历史膨胀。子智能体自主完成其工作,只返回简明的摘要,使主线程保持聚焦和高效。
带任务分派器的智能体注册表
python
from langchain.tools import tool
from langchain.agents import create_agent
# 由不同团队开发的子智能体
research_agent = create_agent(
model="gpt-5.5",
prompt="You are a research specialist..."
)
writer_agent = create_agent(
model="gpt-5.5",
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-5.5",
tools=[task],
system_prompt=(
"You coordinate specialized sub-agents. "
"Available: research (fact-finding), "
"writer (content creation). "
"Use the task tool to delegate work."
),
)typescript
import { tool, createAgent } from "langchain";
import * as z from "zod";
// 由不同团队开发的子智能体
const researchAgent = createAgent({
model: "gpt-5.5",
prompt: "You are a research specialist...",
});
const writerAgent = createAgent({
model: "gpt-5.5",
prompt: "You are a writing specialist...",
});
// 可用子智能体的注册表
const SUBAGENTS = {
research: researchAgent,
writer: writerAgent,
};
const task = tool(
async ({ agentName, description }) => {
const agent = SUBAGENTS[agentName];
const result = await agent.invoke({
messages: [
{ role: "user", content: description }
],
});
return result.messages.at(-1)?.content;
},
{
name: "task",
description: `Launch an ephemeral subagent.
Available agents:
- research: Research and fact-finding
- writer: Content creation and editing`,
schema: z.object({
agentName: z
.string()
.describe("Name of agent to invoke"),
description: z
.string()
.describe("Task description"),
}),
}
);
// 主协调智能体
const mainAgent = createAgent({
model: "gpt-5.5",
tools: [task],
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 个)
- 智能体注册表很少变化
- 你想要最简单的实现
示例:
python
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 个)
- 你想要类型安全和显式的智能体名称
- 你更倾向于基于模式的验证,而不是基于提示词的指导
示例:
python
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 个)或不断增长的注册表
- 智能体注册表频繁变化或是动态的
- 你想要减少提示词大小和 token 用量
- 不同团队独立管理不同的智能体
示例:
python
@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."
)子智能体输入
自定义子智能体为执行其任务所接收的上下文。通过从智能体的状态中提取,添加在静态提示词中难以捕捉的输入——完整的消息历史、先前的结果或任务元数据。
python
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,
# 你还可以根据需要在此处传递其他状态键。
# 请确保在主智能体和子智能体的状态
# schema 中都定义了这些键。
"example_state_key": runtime.state["example_state_key"]
})
return result["messages"][-1].contenttypescript
import { createAgent, tool, AgentState, ToolMessage } from "langchain";
import { Command } from "@langchain/langgraph";
import * as z from "zod";
// 示例:通过状态将完整的对话历史传递给子智能体。
const callSubagent1 = tool(
async ({query}) => {
const state = getCurrentTaskInput<AgentState>();
// 应用所需逻辑,将消息转换为合适的输入
const subAgentInput = someLogic(query, state.messages);
const result = await subagent1.invoke({
messages: subAgentInput,
// 你还可以根据需要在此处传递其他状态键。
// 请确保在主智能体和子智能体的状态
// schema 中都定义了这些键。
exampleStateKey: state.exampleStateKey
});
return result.messages.at(-1)?.content;
},
{
name: "subagent1_name",
description: "subagent1_description",
}
);子智能体输出
自定义主智能体接收到的内容,以便它做出良好的决策。两种策略:
- 提示子智能体:明确指定应返回的内容。一个常见的失败模式是子智能体执行了工具调用或推理,但未将结果包含在其最后一条消息中——提醒它,监督者只会看到最终输出。
- 在代码中格式化:在返回响应之前调整或丰富它。例如,使用
Command除了返回最终文本外,还传回特定的状态键。
python
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
)
]
})typescript
import { tool, ToolMessage } from "langchain";
import { Command } from "@langchain/langgraph";
import * as z from "zod";
const callSubagent1 = tool(
async ({ query }, config) => {
const result = await subagent1.invoke({
messages: [{ role: "user", content: query }]
});
// 返回 Command 以更新多个状态键
return new Command({
update: {
// 传回子智能体的额外状态
exampleStateKey: result.exampleStateKey,
messages: [
new ToolMessage({
content: result.messages.at(-1)?.text,
tool_call_id: config.toolCall?.id!
})
]
}
});
},
{
name: "subagent1_name",
description: "subagent1_description",
schema: z.object({
query: z.string().describe("The query to send to subagent1")
})
}
);检查点持久化与状态检查
默认情况下,子智能体使用继承的检查点器模式——每次调用都从全新状态开始,支持 中断,并可安全地并行运行。如果你需要子智能体跨调用维护自己的持久对话历史,请使用 checkpointer=True(continuations 模式)编译它。有关各模式的完整对比,请参阅 子图持久化。
由于子智能体在工具函数内部被调用,LangGraph 无法 静态发现 它们。这意味着带 subgraphs 的 get_state 不会返回子智能体状态。如果你需要读取嵌套的图状态(例如在 中断 期间),请改为在自定义图的 节点函数 中调用子智能体。有关每种模式如何影响状态可见性的详细信息,请参阅 子图持久化。
- 从 langgraph-supervisor 迁移 — langgraph-supervisor 包已不再积极维护。了解如何从 create_supervisor 迁移到子智能体模式,包括带外部 API 回调的中断与恢复流程。