代码仓库ChainReaction

LangChain 官方文档的 Multi-agent 页面开头就写了一句不太像推销的话:not every complex task requires this approach。一个 agent 配上合适的工具和 prompt,往往就能拿到差不多的结果。

我见过反过来的情况更多。工具还没写到第十个,先搭了一套 supervisor,主代理调子代理,子代理再调工具,出了错要在三层日志里找是谁说错了话。调试成本涨上去,效果没变好。

那什么时候拆是对的。官方给了三个动机,按我实际项目里的出现频率排一下:上下文隔离、职责分离、并行。这三个词听起来都像正确的废话,落到代码上是三件很具体的事。

上下文隔离是说,某个领域有几千 token 的规范和一堆专用工具,塞进主对话会把窗口顶满,而且每轮都要重新付一遍 token 钱。子代理可以只在需要的时候开一个干净窗口,干完活把结论交回来。

职责分离是说,日历、邮件、CRM、数据库四块能力由四个团队维护,各自发布各自测试,主流程只认工具名和描述这个接口。

并行是说,三个互不依赖的子任务同时跑,总延迟按最慢的那个算,而不是按三个加起来算。

反过来,如果你的工具不到十个、领域知识一个 prompt 装得下、任务本身是线性的,拆多代理只会多出一堆胶水代码。这篇文章把官方文档里的四种模式逐个跑一遍,subagents、handoffs、router 都有可运行代码和真实终端输出,custom workflow 我说明它和前三者的关系。脚本放在仓库的 MultiAgent/,正文默认用 DeepSeek;router 的分类器另外在 Gemini 上跑了一组正向对照。

官方自己算的成本账

拆之前先看一张表。官方按三种请求场景统计了各模式的模型调用次数和 token 总量,数字直接来自 Multi-agent 总览。

场景 Subagents Handoffs Skills Router
一次性请求 4 次调用 3 次 3 次 3 次
同一请求再说一遍 4+4 次 3+2 次 3+2 次 3+3 次
多领域并行(三份各 2000 token 的文档) 5 次,约 9K token 7 次以上,14K+ token 3 次,约 15K token 5 次,约 9K token

三行数字各说明一件事。

第一行,subagents 比其他模式多一次调用。子代理的结论要先回到主代理,主代理再组织语言回给用户,多出来的这一步是集中控制的代价。

第二行,handoffs 和 skills 这类有状态的模式在重复请求上能省掉将近一半调用,因为状态还在,不需要重新交接一次。subagents 是无状态的,每次调用子代理都从零开始,成本恒定,换来的是强隔离。

第三行最能说明问题。多领域场景下,skills 只调用三次模型,但每次调用都要重新处理已经加载进上下文的 6K token 文档,总量反而最高。subagents 和 router 靠隔离和并行,token 总量少三成以上。handoffs 在这个场景里最差,因为它必须串行交接,没法把三个语言的研究同时发出去。

官方还有一张按能力打分的表,我把它和前面这张揉在一起,做成四种模式的选择依据。

模式 控制流 上下文边界 最适合 代价
Subagents 主代理用工具调子代理,全部路由经过主代理 子代理默认只拿到任务描述,各自一个干净窗口 多个独立领域、需要并行、需要集中管控 每个请求多一次模型调用,子代理看不到主对话
Handoffs 工具写状态变量,系统读变量换配置或换 agent 消息历史只有一份,变的是 system prompt 和工具集 多轮对话、有先后约束的流程、子代理要直接跟用户说话 串行,不能并行;状态机设计不好会绕圈
Router 一次分类调用,再分发到一个或多个处理方 分类器只拿 query,不带对话历史 输入能明确分类、需要并行查询多个来源再汇总 无状态,多轮对话要自己在外面套一层
Custom workflow 自己用 LangGraph 画节点和边 由你声明的 state schema 决定,每个节点读写哪些字段是显式的 要混确定性逻辑和模型调用、要复杂分支或循环 图的形状得自己想清楚,官方不给现成结构

官方对最后一行还有个补充:router 本身就是 custom workflow 的一个例子,四种模式可以互相嵌套,subagents 调用的工具里可以是一个 custom workflow,workflow 的某个节点里也可以是一个完整的 subagents 系统。

下面逐个看。

Subagents:主代理只拿结论

这是最常被拿来用的一种。核心机制只有一句话:把子代理包成一个工具。

实线是控制流,虚线指向的是各自的上下文。两条虚线互不相交,这是整个模式的关键。

定义与包装

子代理和主代理是同一个 create_agent,只是 prompt 和工具不同。包装成工具的写法就是官方 Basic implementation 里的那几行:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
import os

from langchain.agents import create_agent
from langchain.tools import ToolRuntime, tool
from langchain_openai import ChatOpenAI


def make_model(max_tokens: int = 1000):
"""全篇共用的模型工厂,后面片段里的 make_model(...) 都指这个。"""
return ChatOpenAI(
api_key=os.getenv('DEEPSEEK_API_KEY'),
base_url="https://api.deepseek.com/v1",
model="deepseek-chat",
temperature=0.1,
max_tokens=max_tokens,
)


policy_agent = create_agent(
model=make_model(max_tokens=300),
tools=[lookup_order],
system_prompt=POLICY_DOC, # 359 字符的售后政策全文
)

@tool("policy_expert", description="回答耳机退货、换货、保修的政策问题。输入是一句自然语言问题,需要带上订单号。")
def call_policy_expert(query: str, runtime: ToolRuntime) -> str:
result = policy_agent.invoke({"messages": [{"role": "user", "content": query}]})
return result["messages"][-1].content

@tool 的名字和描述是主代理判断要不要调它的全部依据,也是这个模式里最值得反复改的两行。名字用动词加对象,policy_expert 比 helper 强。描述要写清楚「什么时候用它」,而不只是「它能做什么」。官方在 Subagent specs 一节里把这一点单独列了出来,因为子代理该被调用时没被调用,九成是描述没写清楚。

传工具的方式就这么直白:子代理有它自己的 tools=[…],主代理只拿到 call_policy_expert 一个工具,看不到 lookup_order。这就是上下文边界的物理实现,子代理手里的工具根本不在主代理的工具列表里。

主代理和子代理各自看到什么

我在包装函数里加了一行打印,把主代理调用那一刻的消息数和子代理收到的消息数都记下来,跑两轮对话。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
第 1 轮用户输入: 订单 A1024,我上周买的降噪耳机,包装拆了,用了两次,还能退吗?

--- 主代理上下文里最终留下的 4 条消息 ---
[0] HumanMessage len=34 :: 订单 A1024,我上周买的降噪耳机,包装拆了,用了两次,还能退吗?
[1] AIMessage tool_calls=['policy_expert'] len=37 :: I'll check the return policy for you.
[2] ToolMessage len=44 :: 能退。依据第1、2、5条:签收7天内,序列号未激活、无耳垢未换耳塞,拆封不影响二次销售。
[3] AIMessage len=87 :: 好消息,您这单是可以退的。我这边已经把您的情况和订单号 A1024 转给政策专员确认过了,结论是符合退货条件。
主代理最终回复: 好消息,您这单是可以退的。我这边已经把您的情况和订单号 A1024 转给政策专员确认过了,结论是符合退货条件。

第 2 轮用户输入: 那如果我想换成 Lite 刻字款呢,订单 A2088。

--- 主代理上下文里最终留下的 8 条消息 ---
[0] HumanMessage len=34 :: 订单 A1024,我上周买的降噪耳机,包装拆了,用了两次,还能退吗?
...(中间 4 条省略)
[4] HumanMessage len=27 :: 那如果我想换成 Lite 刻字款呢,订单 A2088。
[5] AIMessage tool_calls=['policy_expert'] len=0 ::
[6] ToolMessage len=52 :: 只能保修/不受理:定制刻字商品不适用七天无理由(第6条),且换货需同第2条,刻字款不适用(第3、6条)。
[7] AIMessage len=85 :: 抱歉,这个换货请求恐怕不行。政策专员确认,刻字属于定制商品,不适用七天无理由,换货也不受理。
主代理最终回复: 抱歉,这个换货请求恐怕不行。政策专员确认,刻字属于定制商品,不适用七天无理由,换货也不受理。

子代理每次被调用时看到的东西
--- 第 1 次调用 ---
主代理此刻的对话轮数: 2
传给子代理的消息条数: 1
子代理收到的第一条消息: 订单 A1024,用户上周购买的降噪耳机,包装已拆封,使用过两次,询问是否还能退货?
子代理内部跑了几条消息: 4
返回给主代理的文本: 能退。依据第1、2、5条:签收7天内,序列号未激活、无耳垢未换耳塞,拆封不影响二次销售。
--- 第 2 次调用 ---
主代理此刻的对话轮数: 6
传给子代理的消息条数: 1
子代理收到的第一条消息: 订单 A2088,用户想把商品换成 Lite 刻字款,询问换货政策是否支持?
子代理内部跑了几条消息: 4
返回给主代理的文本: 只能保修/不受理:定制刻字商品不适用七天无理由(第6条),且换货需同第2条,刻字款不适用(第3、6条)。

对比
主代理最终上下文消息数 : 8
子代理策略原文长度 : 359 字符(从未进入主代理上下文)
子代理被调用次数 : 2
每次调用子代理的起始消息数: [1, 1]

几件事在这段输出里看得见。

主代理的上下文从 2 条涨到 6 条再到 8 条,子代理每次进来永远只有 1 条消息。那 359 字符的政策原文只在子代理的 system prompt 里,从没进过主代理的上下文,主代理只拿到 44 个字的结论。

主代理并没有原样转发用户的话。它把「包装拆了,用了两次」改写成「包装已拆封,使用过两次」再交给子代理。子代理看到什么,完全由你在包装函数里决定,这是一个可以精确控制的接口,不是自动继承。

子代理内部跑了 4 条消息,中间调了一次 lookup_order,这些过程全部留在子代理自己的窗口里,主代理的 [2] 只看到最终那句话。

怎么限制子代理

限制手段有三层,从便宜到贵排。

输出层面最直接:在子代理的 system prompt 里写明返回格式和长度。我的政策专员 prompt 最后一句是「最终回答必须控制在 60 字以内,主代理只会看到你的这段最终文本」。官方在 Subagent outputs 里专门提醒过这个坑:子代理经常调完工具、推理完了,却在最终消息里没把结果说出来,因为主代理只看最后一条。把那句话写进 prompt 能省掉很多来回。

调用层面,给子代理单独设 max_tokens。我的子代理是 300,主代理是 1000。子代理是干活的不需要长篇大论,主代理要组织语言。

输入层面,用 runtime.state[“messages”] 决定传多少历史过去。官方给了两种模式的名字:isolated 只传任务描述,fork 传父代理的完整历史。默认应该是 isolated。只有子代理要接着父代理没干完的活,比如继续审查同一个 PR,才值得 fork,代价是 prompt 变大、隔离性变弱。

Handoffs:状态决定配置,不是消息决定

Handoffs 和 subagents 的区别,用一句话说:subagents 是主代理「叫」一个子代理干活然后等它回来,handoffs 是把控制权整个交出去,交接之后当前这个 agent 的 prompt 和工具集都变了。

它的核心机制是工具返回 Command 改一个状态变量,系统读这个变量换配置。

和 subagents 那张图对比一下就看出来了。这里只有一份 messages,对话历史从头到尾是同一条线。变的是 system prompt 和工具列表,也就是上下文边界划在 prompt 和工具上,而不是划在消息历史上。

状态怎么传

官方给了两种实现。单 agent 加中间件,或者多个 agent 子图。文档里明说优先用前者,后者只在某个 agent 节点本身就是一张复杂图(带反思、带检索)的时候才用。

单 agent 版本里,状态更新全靠工具返回 Command:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
class SupportState(AgentState):
current_step: str = "triage"
warranty_status: str | None = None

@tool
def record_warranty_status(status: str, runtime: ToolRuntime[None, SupportState]) -> Command:
"""记录保修状态,并把流程推进到 specialist 步骤。"""
return Command(update={
"messages": [
ToolMessage(content=f"已记录保修状态:{status}", tool_call_id=runtime.tool_call_id)
],
"warranty_status": status,
"current_step": "specialist",
})

那个 ToolMessage 不能省。模型发起工具调用之后会等一个响应,缺了它消息序列就是坏的,官方文档里专门用了一个 Note 说明这件事。

中间件负责把 current_step 翻译成实际配置:

1
2
3
4
5
6
@wrap_model_call
def apply_step_config(request: ModelRequest, handler):
step = request.state.get("current_step", "triage")
cfg = CONFIGS[step]
request = request.override(system_prompt=cfg["prompt"], tools=cfg["tools"])
return handler(request)

request.override 是官方的写法,一次调用可以同时换 system prompt 和工具。中间件按 current_step 取配置,这一步是纯 Python,没有任何模型参与,所以流程推进是确定的,不是「希望模型自己记得」。

跑起来是这样

三轮对话,每轮结束后我打印状态和中间件看到的配置。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
第 1 轮用户: 我的手机屏幕摔碎了。
本轮模型调用次数: 1
状态: current_step=None warranty_status=None
回复: 您好,很抱歉听到您的手机屏幕摔碎了。为了帮您处理,我需要先确认一下:您的手机目前是否还在保修期内呢?
消息轨迹: ['HumanMessage', 'AIMessage']

第 2 轮用户: 还在保修期内。
本轮模型调用次数: 3
状态: current_step='specialist' warranty_status='in_warranty'
回复: 为您查询到维修方案如下:
……(这段回复原文带 emoji 和 markdown 加粗,共 6 行,此处省略)
消息轨迹: ['HumanMessage', 'AIMessage', 'HumanMessage', 'AIMessage', 'ToolMessage', 'AIMessage', 'ToolMessage', 'AIMessage']

第 3 轮用户: 那具体怎么修?大概要多久?
本轮模型调用次数: 1
状态: current_step='specialist' warranty_status='in_warranty'
回复: 关于您关心的两个问题,为您说明如下:
……(同上,这段回复原文也带 emoji 和加粗,共 7 行,此处省略)
消息轨迹: ['HumanMessage', 'AIMessage', 'HumanMessage', 'AIMessage', 'ToolMessage', 'AIMessage', 'ToolMessage', 'AIMessage', 'HumanMessage', 'AIMessage']

中间件每次拦截到的配置
--- 第 1 次模型调用 ---
current_step : triage
warranty_status: None
可用工具 : ['record_warranty_status']
--- 第 2 次模型调用 ---
current_step : triage
warranty_status: None
可用工具 : ['record_warranty_status']
--- 第 3 次模型调用 ---
current_step : specialist
warranty_status: in_warranty
可用工具 : ['provide_solution', 'escalate_to_human']
--- 第 4 次模型调用 ---
current_step : specialist
warranty_status: in_warranty
可用工具 : ['provide_solution', 'escalate_to_human']

第 1 轮里 current_step 打印出来是 None。这不是 bug,是我在 AgentState 上写的类默认值不会自动写进 state,中间件靠 request.state.get(“current_step”, “triage”) 的兜底才拿到 triage。第一次调用时 provide_solution 根本不在工具列表里,模型就算想跳过收集信息这一步也没有工具可调,这就是官方说的「unlock capabilities only after preconditions are met」。

第 2 轮三次模型调用完成了整个过程:模型先调 record_warranty_status,状态变成 specialist,中间件立刻换了配置,模型在新配置下又调了 provide_solution,然后才组织语言回复。一次用户输入,配置在中途换了。

什么时候该交接而不是调用

判断标准是控制权归谁。

如果主代理拿回结论之后还要继续做别的事,用 subagents。比如客服主管拿到退货结论之后,还要决定要不要发优惠券安抚用户,这个决定得由主代理做。

如果交接之后就是另一个角色在跟用户对话,中间不需要谁来做统筹,用 handoffs。售前转售后,售后接手之后整段对话都是售后在处理,这时候留一个「主管」在中间转发没有意义,还多一次模型调用。

还有一种情况只有 handoffs 能做:给流程加先后约束。上面那个例子里,用户在保修状态没记录之前,物理上就调不到 provide_solution。subagents 做不到这一点,因为工具是挂在主代理身上的,主代理想调就调。

多个 agent 子图那版

官方给的另一种实现,用 Command(goto=…, graph=Command.PARENT) 在图的节点之间跳。这版依赖 LangGraph 的 Command.PARENT 和状态 reducer,我照着文档抄下来跑通了,两个 agent 之间成功交接:

1
2
3
4
5
6
[0] HumanMessage    tool_calls=                         :: Hi, I'm having trouble with my account login. Can you help?
[1] AIMessage tool_calls=transfer_to_support :: I'd be happy to help, but login and account issues are technical suppo
[2] ToolMessage tool_calls= :: Transferred to support_agent
[3] AIMessage tool_calls= :: You've been transferred to our support agent, who will help you with y
active_agent = support_agent
总消息数 = 4

transfer_to_support 这个工具没有返回字符串,它返回了一个 Command,把 active_agent 改成 support_agent,并 goto=”support_agent”。图上一条条件边读到 active_agent,下一轮就进了 support 节点。

要注意的是 update 里的 messages 只有两条:触发交接的那条 AIMessage,和一条人工构造的 ToolMessage。官方特意解释了为什么不把子代理的全部消息历史带过去:接收方会被无关的内部推理干扰,token 也白花。交接时该带什么、不该带什么,得一条条想。

Router:先分类,再分发,最后汇总

Router 的适用场景很窄也很明确:输入能分成几个清楚的类别,每类背后是一套独立的知识。

Router 和 subagents 长得像,区别在谁来路由。subagents 的主代理是一个完整的 agent,它带着对话上下文、跨多轮决定调谁。router 的分类器是一次独立调用,它只看当前这个 query,不知道之前聊过什么。官方文档用一句话划清了界限:有清楚的输入分类、想要确定或轻量的判断,用 router;需要灵活的、感知上下文的编排,用 supervisor。

分类器

分类器就是一次 with_structured_output 调用。先看跑得通的那条路,Gemini(gemini_provider_strategy.py):

1
2
3
4
=== 3. Multi-agent 篇:分类器 with_structured_output(Route) ===
明天上海会下雨吗 -> Route destination='weather' reason='用户查询上海明天的天气情况(是否会下雨),属于天气预报类需求。'
苹果股价多少 -> Route destination='stock' reason='用户在询问苹果公司的股票价格。'
帮我给张三发封邮件 -> Route destination='email' reason='用户请求给张三发送邮件'

Route 是 Pydantic 模型,destination 是 Literal["weather", "stock", "email"],reason 是自由文本。三次都选对了类别,reason 也说得通。这条路径靠的是模型的原生结构化输出,跟 Structured Output 那篇里 ProviderStrategy 跑通的是同一件事,只是这里走 with_structured_output 这一层。

DeepSeek 上就有具体问题了。with_structured_output 传 Pydantic 模型时默认 method 是 json_schema,而 DeepSeek 不认这个 response_format:

1
2
3
=== 1. DeepSeek + with_structured_output(Route) 默认走 json_schema ===
--- deepseek-chat ---
OpenAIInvalidRequestError: Error code: 400 - {'error': {'message': 'This response_format type is unavailable now (request_id: 78fb6d97-...)', 'type': 'invalid_request_error'}}

原生结构化输出这条路我另外找了 Ollama 云端试。这里要分清两条通路:前面用的 https://ollama.com/v1 是 OpenAI 兼容端点,参数收下不生效,模型照旧输出散文,报错推迟到解析那一步:

1
2
3
4
5
6
7
=== 3. Ollama 云端 gpt-oss:120b 的两条路 ===
--- with_structured_output(Route) 默认 json_schema ---
ValidationError: 1 validation error for Route
Invalid JSON: expected value at line 1 column 1 [type=json_invalid, input_value='您好,针对客户反...(输出已截断)', input_type=str]

--- create_agent + ProviderStrategy(Route) ---
StructuredOutputValidationError: Failed to parse structured output for tool 'Route': Native structured output expected valid JSON for Route, but parsing failed: Expecting value: line 1 column 1 (char 0)..

换成 langchain-ollama 的 ChatOllama 接原生端点 https://ollama.com,直接 with_structured_output 能拿回对象,但同一个模型放进 create_agent 的 ProviderStrategy 还是失败(structured_native_ollama.py):

1
2
3
4
5
6
7
=== 3. 直接 with_structured_output,temperature=0,连续 12 次 ===
run04: OK name='John Doe' email='john@example.com' phone='(555) 123-4567'
...(中间省略 11 行)
成功率: 5/12

=== 6. 同一原生通路,create_agent + ProviderStrategy ===
ProviderStrategy(strict=True), temperature=0: 失败 StructuredOutputValidationError: Failed to parse structured output for tool 'ContactInfo': Native structured output expected valid JSON for ContactInfo, but parsing failed: Expecting value: line 1 column 1 (char 0)..

原生端点上的直接调用能成,功劳在 LangChain 的 PydanticOutputParser:它会把散文里那段围栏 JSON 剥出来。Ollama 并没有按 schema 约束住模型。ProviderStrategy 内部只有一句 json.loads,内容以散文开头就直接抛。所以 DeepSeek 和 Ollama 都没跑通,但 Gemini 跑通了,结论跟 Structured Output 那篇一致:provider 的 profile 里 structured_output 为 True 时,原生结构化输出是能用的。「厂商不支持」这个说法同样站不住,同一份 key,换个端点、换个调用层,结果就不一样。成功率也不稳,这次 12 次里 5 次,另一次 6 次的重跑只成了 2 次。

四档摊开是这样。Gemini 的 gemini-3.5-flash 属于正向跑通,profile.structured_output 为 True,三次分类全对。DeepSeek 三个模型(deepseek-chat、deepseek-flash、deepseek-v4-pro)一律 400,属于明确拒绝。Ollama 云端 OpenAI 兼容端点(https://ollama.com/v1)的 gpt-oss:120b、gpt-oss:20b、nemotron-3-super 收了参数不生效,profile 全是 None,LangChain 的能力探测认不出它们,只能退回内置的模型名正则名单,而名单里没有这些名字。换成原生端点,直接 with_structured_output 通了,但 create_agent + ProviderStrategy 仍不通。剩下的 glm-5.2、minimax-m3、mistral-large-3:675b 在免费额度外(402 this model is not included in your free usage),根本没跑起来。

DeepSeek 也不是完全不给 json。同一个模型,response_format 换成 json_object 就能用,前提是 prompt 里出现 json 这个词:

1
2
3
4
=== 2. DeepSeek + with_structured_output(Route, method='json_mode') ===
--- prompt 里带 json 字样 ---
type: <class '__main__.Route'>
repr: Route(destination='账单', reason='客户反馈发票金额不一致,属于账单/财务核对问题,应由账单部门处理。')

所以 DeepSeek 上的分类器有两种写法:留在 with_structured_output 这一层,显式写 method="json_mode" 并在 prompt 里提 json;或者按官方 Structured output 的说明走 ToolStrategy。这里选后者,理由是分类 schema 里有个 list[Literal[...]] 多选字段,ToolStrategy 至少会用 Pydantic 校验一遍,json_mode 拿到什么就是什么,没有第二次机会。换成 Gemini 就没这些事,Pydantic 模型直接传给 with_structured_output 就行。写法:

1
2
3
4
5
6
7
8
9
10
class Route(BaseModel):
categories: list[Literal["tech", "billing", "account"]] = Field(description="命中的领域,可多选")
reason: str = Field(description="一句话说明为什么这么分")

classifier = create_agent(
model=make_model(max_tokens=300),
tools=[],
response_format=ToolStrategy(Route),
system_prompt="你是客服工单分类器。只做分类,不回答问题,不解释业务。",
)

跑出来是这样:

1
2
3
4
5
输入: 我的账单这个月多扣了 30 块,另外 App 登录一直转圈进不去。
-> categories=['billing', 'account'] reason=用户反映账单多扣30元属账单与扣费问题,App登录转圈进不去属账号与登录问题。

输入: 怎么开发票?
-> categories=['billing'] reason=用户咨询如何开具发票,属于账单与扣费相关事项。

第一条被分到两个类别。list[Literal[…]] 这个 schema 决定了它可以多选,这也是 router 能扇出的前提。

分发与汇总

官方给了两个原语:单个目标用 Command(goto=…),多个目标并行用 Send。我用 Send 把同一个 query 扇出到三个处理函数,处理函数是纯 Python 函数,没有模型参与:

1
2
3
def fan_out(state: State):
targets = [c for c in state["categories"] if c in ("tech", "billing", "account")]
return [Send(f"handle_{c}", {"query": state["query"], "category": c}) for c in targets]

每个处理函数返回 {“findings”: […]},State 里 findings 字段带 Annotated[list[str], operator.add],并行写回来会自动拼成一个列表。汇总节点把这几个片段交给模型压成一段话。真实输出:

1
2
3
4
5
6
7
8
9
10
11
[classify] '我的账单这个月多扣了 30 块,另外 App 登录一直转圈进不去。' -> ['billing', 'account']
[handle_billing] 收到 payload={'query': '我的账单这个月多扣了 30 块,另外 App 登录一直转圈进不去。', 'category': 'billing'}
[handle_account] 收到 payload={'query': '我的账单这个月多扣了 30 块,另外 App 登录一直转圈进不去。', 'category': 'account'}
总耗时 1.46s
各处理函数的执行区间(相对起点,秒):
handle_billing [0.62 -> 0.92]
handle_account [0.62 -> 0.92]
findings:
[账单] 账单侧:多扣费先核对 11 月增值包续订记录,确认重复扣费则提交退款工单,3 到 5 个工作日原路退回。
[账号] 账号侧:登录失败先确认手机号是否变更,验证码每日上限 5 次,超限需等次日 0 点。
最终回答: 先核对11月增值包续订记录,若确认重复扣费,提交退款工单,3到5个工作日原路退回。登录失败请确认手机号是否变更,验证码每日最多5次,超限需等次日0点再试。

两个处理函数各 sleep(0.3),执行区间都是 [0.62 -> 0.92],完全重叠。这就是并行扇出的证据,串行的话第二个应该从 0.92 开始。

汇总节点那句 prompt 里我写了一条约束:不要提「专员」「领域」这些词。并行分发之后如果直接把三段拼起来给用户,读起来会像三个客服同时说话,汇总这一步的质量直接决定用户看到的体验。

Router 不带记忆,要记忆就自己在外面套

上面那个 workflow 是无状态的,每次 invoke 都要重新分类。多轮对话怎么办。官方给了个轻量办法:把 router 包成一个工具,交给一个有 checkpointer 的会话代理。

1
2
3
4
5
6
第 1 轮用户: App 登录一直转圈怎么办?
代理回复: App 登录转圈通常是 1.6.3 版本的已知问题,请升级到 1.6.4 版本;如果升级后仍失败,可以清除 App 缓存后重试。
消息轨迹: ['HumanMessage', 'AIMessage', 'ToolMessage', 'AIMessage']
第 2 轮用户: 那我刚才问的是什么问题?
代理回复: 你刚才问的是:App 登录一直转圈怎么办。
会话代理上下文消息数: 6 (router 自己不带任何记忆)

记忆全在会话代理这一层,router 每次都是干净的。官方也提醒过,如果让 router 自己维护历史、还要在多个 agent 之间切换,语气和 prompt 不一致会让用户觉得对话不连贯,那种情况下文档建议直接换 handoffs 或者 subagents。

Custom workflow:前面三种都是它的特例

第四种模式没有固定形状,就是自己用 LangGraph 的 StateGraph 画节点和边。每个节点可以是一个纯函数、一次模型调用,或者一整个 agent。

上下文边界在这张图里是显式的。你声明一个 TypedDict 当 state,每个节点读哪几个字段、写哪几个字段,代码里看得一清二楚,不需要猜。官方文档说得很直接:LangGraph 的 state 就是用来在步骤之间传信息的,每个节点读和写结构化字段。

官方那个 RAG 例子里有节点分类很值得学:改写节点是模型节点,检索节点是确定性节点,最后那个 agent 节点带工具、可以继续调模型。三种节点混在一张图里,确定性逻辑和 agent 行为各就各位。

Router 本身就是一个 custom workflow,subagents 也可以塞进某个节点。所以实际项目里很少只用一种,通常是外面一张图管流程,图里的某个节点是一个 subagents 系统。

四个坑

无限委派。主代理调子代理,子代理又有一个「再委派」的工具,两边描述写得含糊,模型就会来回踢皮球。防的办法很土:子代理不挂任何委派工具,委派只发生在主代理这一层。真需要多层,给调用深度加一个计数器,超过阈值直接抛错。

上下文重复。子代理每次从零开始,它要重新推导主代理已经知道的东西。我的例子里子代理收到的 query 是主代理改写过的,多花了主代理 37 个 token 的思考,省下的是子代理不用读前面那 4 条消息。这个交换在长对话里划算,在只有一轮的请求里就是白花。官方的性能表里,一次性请求 subagents 是 4 次调用、其他模式 3 次,多出来的就是这一笔。

成本失控。四个坑里这个最容易发生。并行扇出看着爽,但 Send 出去三个处理函数,如果每个处理函数内部又是一个 agent,一次用户输入就是三次模型调用起步。上线前拿官方那张表估一下量级:多领域场景 subagents 约 9K token,skills 约 15K,差了六成。router 的分类器每次请求都要跑一次,重复请求场景下这个开销会累积,官方的数字是两轮 6 次调用,而 handoffs 只要 5 次。

调试困难。子代理是在工具函数里被 invoke 的,LangGraph 静态发现不了它。官方的原话是 get_state 配 subgraphs 拿不到子代理的状态。要读嵌套图的状态,得把子代理放到一个 custom workflow 的节点函数里调用,而不是包成工具。我自己的做法是在包装函数里把子代理的输入输出都打到日志里,这也正是前面那段「子代理每次被调用时看到的东西」的来历。真上生产还是接 LangSmith,官方的多代理页面第一段 Tip 就是推荐开 tracing。

小结

拆之前先问一句:是因为上下文装不下、因为要并行、还是因为不同团队要各管一块。三个理由一个都不占,就不该拆。

四种模式的选择可以缩成三条判断。子代理不该跟用户说话、主代理要集中管控,用 subagents。子代理要直接跟用户多轮对话、流程有先后约束,用 handoffs。输入能分类、要并行查多个来源,用 router。三种都不合适,再用 custom workflow 自己画。

上下文边界是设计这个系统时真正要动手的地方。subagents 把它划在消息历史上,handoffs 划在 prompt 和工具上,router 划在「分类器只看 query」这条线上。想清楚每个 agent 该看到什么,比选哪个模式更重要。

子代理默认无状态,这是特性不是缺陷。它让成本可预测、隔离彻底,代价是每次重新推导。要不要把父代理的历史传过去,取决于子代理是在开新活还是在接手旧活。

官方那张性能表值得存下来。一次性请求 3 次还是 4 次调用、多领域场景 9K 还是 15K token,这些数字在评审方案的时候比架构图管用。