生命不息,折腾不止。前五课你的 Agent 已经会画图、会用工具、会打包上线、会带队派活、会并行干活。收官这一课,教你用 Command 让一个 Agent 干到一半、握着现场直接换岗——不用老板在中间传话,交接一句不丢。

一、Supervisor 的交接,输在哪

回顾第四课的 Supervisor:老板(调度 Agent)手里拿着一堆「转移工具」,哪个专家该上场,得老板先调用 transfer_to_xxx 工具、把要传的话塞进工具参数里,再路由过去。

这套「工具式交接」有两个天花板:

  1. 交接内容受限于工具参数:能传的只有你事先在工具签名里定义好的那几样,想多塞点上下文、塞点结构化结论,就得把工具改来改去;
  2. 绕不开中间人:每次交接都要老板转发,专家之间没法直接对接,多转一道就多一层传话损耗。

Command 就是来破这个天花板的。它让你从任意一个节点(或者 Agent 内部的工具)里,直接「宣告」下一步去哪、顺便把现场状态写回去。交接不再靠老板传话,而是专家自己带着状态找下家——这就是「徒手交接」。

二、Command 到底是个啥:三个字段说透

Command 是 LangGraph 里一个很轻的对象(from langgraph.types import Command),官方签名里最关键的就三个字段:

字段 作用 取值
update 往图的状态里写东西(现场状态、交接说明都放这) 一个 dict,key 必须和状态 schema 对上
goto 下一步跳去哪个节点 节点名字符串 / 一串节点名 / Send 对象
graph 这个 goto哪张图里找节点 默认 None(当前图);Command.PARENT 表示「回最近的父图」

update 解决「带现场」,goto 解决「去哪」,graph 解决「在哪张图里找」。三个合起来就是一句完整的话:把这份状态写进去,然后去(父图的)某某节点继续。

三、最小可跑:节点直接 return Command

3.1 一条 goto,徒手接管控制权

先看一个不依赖任何工具的裸例子——研究节点干完,直接带着结论跳去写手节点:

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
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.types import Command
from langchain_core.messages import AIMessage

def researcher(state: MessagesState) -> Command:
# 调研干完了,把结论和交接说明写进状态,然后直接跳去 writer
return Command(
update={"messages": [AIMessage(content="调研结论:目标用户是中小开发者,预算敏感。")]},
goto="writer",
)

def writer(state: MessagesState) -> dict:
conclusion = state["messages"][-1].content
return {"messages": [AIMessage(content=f"开写:{conclusion}")]}

builder = StateGraph(MessagesState)
builder.add_node("researcher", researcher)
builder.add_node("writer", writer)
builder.add_edge(START, "researcher")
# researcher 靠返回值里的 Command 决定下一跳,这里不再给它加静态边
builder.add_edge("writer", END)
graph = builder.compile()

result = graph.invoke({"messages": []})
print(result["messages"][-1].content)

关键点:researcher 这个节点没有 add_edge("researcher", "writer") 这条静态边,它是靠返回值里的 Command(goto="writer") 决定下一跳的。这就是「接管控制权」——路由逻辑从「图的结构」下沉到了「节点自己的返回值」。

3.2 goto 还能是一串节点

goto 不只能跳一个节点,也支持一次跳一串(依次执行):

1
2
3
4
return Command(
update={"messages": [AIMessage(content="A")]},
goto=["step_b", "step_c"], # 先 step_b,再 step_c,按顺序走
)

四、手把手:多 Agent 徒手交接(官方 handoffs 模式)

第四课靠工具转接,这一课我们换成在 @tool 里直接 return Command(... + graph=Command.PARENT),让「客服 / 销售」两个 Agent 一节一节把现场传下去。下面是官方多智能体 handoff 的完整骨架(create_agent 需要较新的 langchain 版本,以官方文档为准):

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
from langchain.agents import AgentState, create_agent
from langchain.messages import AIMessage, ToolMessage
from langchain.tools import tool, ToolRuntime
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
from typing_extensions import NotRequired

# 1. 状态里加一个 active_agent,记录「现在谁在岗」
class MultiAgentState(AgentState):
active_agent: NotRequired[str]

# 2. 交接工具:直接返回 Command,把控制权交给另一个 Agent
@tool
def transfer_to_sales(runtime: ToolRuntime) -> Command:
"""转接给销售 Agent。"""
last_ai = next(m for m in reversed(runtime.state["messages"]) if isinstance(m, AIMessage))
transfer_msg = ToolMessage(content="已转接给销售 Agent", tool_call_id=runtime.tool_call_id)
return Command(
goto="sales_agent", # 去父图的 sales_agent 节点
update={
"active_agent": "sales_agent", # 更新「在岗人」
"messages": [last_ai, transfer_msg], # 把上下文连锅端带过去
},
graph=Command.PARENT, # 关键:跳回父图,不是当前子图
)

@tool
def transfer_to_support(runtime: ToolRuntime) -> Command:
"""转接给客服 Agent。"""
last_ai = next(m for m in reversed(runtime.state["messages"]) if isinstance(m, AIMessage))
transfer_msg = ToolMessage(content="已转接给客服 Agent", tool_call_id=runtime.tool_call_id)
return Command(
goto="support_agent",
update={"active_agent": "support_agent", "messages": [last_ai, transfer_msg]},
graph=Command.PARENT,
)

看到没,核心变化就三个:goto 指定下家、updateactive_agent 和上下文一起写回去、graph=Command.PARENT 把这一跳「发射」回父图。交接信息不再挤在工具参数里,而是能带着整段对话历史一起走。

接着把两个 Agent 各包成一个节点,再用条件边在它们之间路由:

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
# 3. 造两个带交接工具的 Agent(model 换成任意 OpenAI 兼容接口都行,
# 比如接中转站 ai.aklibk.com:国内直连、免绑卡、人民币按量付费)
sales_agent = create_agent(
model="gpt-4o-mini", tools=[transfer_to_support],
system_prompt="你是销售。遇到技术/售后问题就转接客服。",
)
support_agent = create_agent(
model="gpt-4o-mini", tools=[transfer_to_sales],
system_prompt="你是客服。遇到报价/购买问题就转接销售。",
)

def call_sales_agent(state): return sales_agent.invoke(state)
def call_support_agent(state): return support_agent.invoke(state)

# 4. 路由:agent 干完没调工具 → 收尾;调了交接工具 → 跳到 active_agent 指的那个
def route_after_agent(state):
last = state.get("messages", [])[-1]
if isinstance(last, AIMessage) and not last.tool_calls:
return END
return state.get("active_agent", "sales_agent")

# 5. 搭图
builder = StateGraph(MultiAgentState)
builder.add_node("sales_agent", call_sales_agent)
builder.add_node("support_agent", call_support_agent)
builder.add_conditional_edges(
START, lambda s: s.get("active_agent", "sales_agent"), ["sales_agent", "support_agent"])
builder.add_conditional_edges("sales_agent", route_after_agent, ["sales_agent", "support_agent", END])
builder.add_conditional_edges("support_agent", route_after_agent, ["sales_agent", "support_agent", END])
graph = builder.compile()

result = graph.invoke({"messages": [{"role": "user", "content": "我登录不上账号,帮帮我。"}]})
for m in result["messages"]:
m.pretty_print()

route_after_agent 就是那个「自动门」:agent 干完如果没再调交接工具(最后一条 AI 消息不带 tool_calls),说明活干完了,收尾;如果调了工具,就按 active_agent 新值跳到下家。整段跑下来,登录问题先进客服,客服一看是账号问题自己就能处理,就直接答;要是用户接着问「你们套餐多少钱」,客服调 transfer_to_sales,上下文原封不动甩给销售。

五、顺带一提:resume 做人审

Command 其实还有第四个字段 resume(我在第二节故意留到这说),它配合 interrupt() 做人审:

1
2
3
4
5
6
7
8
from langgraph.types import Command, interrupt

def approval_node(state):
ok = interrupt("是否批准上架?") # 图在这卡住,把控制权交还人类
return {"approved": ok}

graph.invoke(input, config=cfg) # 跑到 interrupt 处停下
graph.invoke(Command(resume={"approved": True}), config=cfg) # 人审通过后从断点续跑

interrupt() 让图停下来、控制权还给人类;Command(resume=...) 拿到人审结果后,让图从断点继续。这俩配合就是 LangGraph 的 Human-in-the-Loop 骨架。人审不是今天重点,你有兴趣拿这段自己延伸即可。

六、避坑清单(收官,把坑一次填平)

  1. graph=Command.PARENT 只在子图里才有意义:顶层图没有父图,硬用会报错。判断标准——你的节点是不是作为子图 / 子 Agent 被父图调进来的;
  2. gotograph 别搞混goto 是「去哪」,graph 是「在哪张图里找那个节点」。回父图时,goto 里填的必须是父图的节点名,不是子图的;
  3. update 的 key 必须和状态 schema 对上:写了个 schema 里没有的字段,直接 KeyError;messages 这种要想清楚是追加还是覆盖语义;
  4. 交接时把 AI 消息和 ToolMessage 一起塞进 update:只留 ToolMessage 会丢掉上一轮的问答上下文,下家接过来一脸懵;
  5. 别把 Command 当普通数据塞进 stateCommand 只该从节点 / 工具的返回值里发出去,存进状态里没有意义(跟第五课 Send 的坑一个道理);
  6. goto 一串节点时按顺序走goto=["a", "b"] 是「先 a 后 b」,不是并行;要并行用第五课的 Send

七、六课收官,一张图带走

从第一课到现在,这套 LangGraph 技能树已经长齐了:

  • 画图:State / Node / Edge(第一课);
  • 装记忆:checkpointer 长期记忆、断点续跑、人审(第二课);
  • 上线:把 Agent 打包成 API 对外服务(第三课);
  • 带队:Supervisor 派活(第四课);
  • 并行:Send 扇出 + reducer 汇总 + 子图复用(第五课);
  • 交接:Command 徒手换岗 + graph=Command.PARENT(第六课)。

多智能体的三件套——派活(Supervisor)、并行(Send)、交接(Command)——你到这一课全配齐了。把它们串起来,就是一套能上生产、能规模化、能把活拆明白也把责权划清楚的 Agent 系统。

生命不息,折腾不止。LangGraph 实战系列到这就收官了。下一篇开新坑,聊一个更「接地气」的方向:RAG 实战——从零搭一个能读懂你私有文档库的问答 Agent,第一课先搞明白向量化、Embedding、相似度检索这三块地基,不急着上框架,先手动跑通一条「文档 → 向量 → 检索 → 喂给大模型」的最短链路。折腾起来。