一文搞定LangGraph Command：让多智能体协作像流水线一样顺畅

一、为什么你的AI工作流需要Command？

在构建多智能体系统时，我们常常面临一个困境：一个智能体完成工作后，既要更新系统状态，又要告诉系统下一步该去哪里。传统做法是将这两件事分开处理——先更新状态，再单独写一个条件判断函数来决定路由。这就像让一个快递员送完包裹后，先写报告，再打电话问下一站去哪里，效率低且容易出错。

LangGraph的Command对象正是为解决这个问题而生。它允许你在同一个节点中，同时完成状态更新和流程控制。本文将通过三个由浅入深的实战案例，带你彻底掌握Command的使用方法。

二、Command基础：实现智能任务路由

2.1 场景描述

假设我们需要构建一个AI助手，能够根据用户输入自动判断任务类型，并分配给专门的智能体处理。用户说“帮我算数学”则交给数学智能体，说“帮我翻译”则交给翻译智能体，其他内容则结束流程。每个专业智能体处理完毕后，需要返回决策智能体，等待下一个任务。

2.2 核心概念解释

在深入代码之前，先理解三个关键概念：

状态（State）：相当于一个所有智能体共享的笔记本。每个智能体都可以在上面读写信息。在这个例子中，笔记本包含三页：对话记录（自动追加新消息）、当前正在工作的智能体名称、任务是否完成。

Command对象：它是一个特殊的返回值，包含两个部分。update字段指定要写入笔记本的新内容；goto字段指定下一步要执行哪个节点。当节点返回Command时，系统会自动应用状态更新，然后跳转到目标节点。

Reducer函数：当多个智能体向同一个状态字段写入内容时，我们需要定义合并规则。Annotated[list, lambda x, y: x + y]表示采用追加模式，新内容添加到列表末尾，而不是覆盖原有内容。这对于保存对话历史至关重要。

2.3 完整代码实现

""" LangGraph Command 基础演示 功能：根据用户输入自动路由到数学智能体或翻译智能体 每个节点通过Command同时完成状态更新和流程控制 """ from typing import Annotated from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END from langgraph.types import Command # 定义共享状态结构 class AgentState(TypedDict): """ 所有智能体共享的状态 - messages: 对话历史，使用lambda实现追加合并 - current_agent: 当前处理的智能体名称 - task_completed: 任务是否完成 """ messages: Annotated[list, lambda x, y: x + y] current_agent: str task_completed: bool # 决策智能体节点 def decision_agent(state: AgentState) -> Command[AgentState]: """分析用户意图，决定下一步调用哪个智能体""" print("[决策智能体] 开始分析...") # 获取用户最新消息 last_message = state["messages"][-1] if state["messages"] else "" print(f"用户消息: {last_message}") # 根据消息内容路由 if "数学" in last_message: print("-> 识别为数学任务，路由到数学智能体") return Command( update={ "messages": [("system", "路由到数学智能体")], "current_agent": "math_agent" }, goto="math_agent" ) elif "翻译" in last_message: print("-> 识别为翻译任务，路由到翻译智能体") return Command( update={ "messages": [("system", "路由到翻译智能体")], "current_agent": "translation_agent" }, goto="translation_agent" ) else: print("-> 无法识别任务，结束流程") return Command( update={ "messages": [("system", "任务完成，流程结束")], "task_completed": True }, goto=END ) # 数学智能体节点 def math_agent(state: AgentState) -> Command[AgentState]: """执行数学计算，完成后返回决策智能体""" print("[数学智能体] 开始计算...") result = "2 + 2 = 4" print(f"计算结果: {result}") # 返回决策智能体，并携带计算结果 return Command( update={ "messages": [("assistant", f"数学计算结果：{result}")], "current_agent": "decision_agent" }, goto="decision_agent" ) # 翻译智能体节点 def translation_agent(state: AgentState) -> Command[AgentState]: """执行翻译任务，完成后返回决策智能体""" print("[翻译智能体] 开始翻译...") translation = "Hello -> 你好" print(f"翻译结果: {translation}") return Command( update={ "messages": [("assistant", f"翻译结果：{translation}")], "current_agent": "decision_agent" }, goto="decision_agent" ) # 构建并运行图 def main(): print("=== Command基础演示 ===\n") # 创建状态图构建器 builder = StateGraph(AgentState) # 添加节点 builder.add_node("decision_agent", decision_agent) builder.add_node("math_agent", math_agent) builder.add_node("translation_agent", translation_agent) # 设置入口 builder.add_edge(START, "decision_agent") # 编译图 graph = builder.compile() # 测试1：数学任务 print("测试1: 数学任务") result = graph.invoke({ "messages": [("user", "我需要计算数学题")], "current_agent": "user", "task_completed": False }) print("最终状态:", result) print("\n" + "="*50 + "\n") # 测试2：翻译任务 print("测试2: 翻译任务") result = graph.invoke({ "messages": [("user", "我需要翻译文本")], "current_agent": "user", "task_completed": False }) print("最终状态:", result) print("\n" + "="*50 + "\n") # 测试3：无法识别的任务 print("测试3: 无法识别的任务") result = graph.invoke({ "messages": [("user", "你好")], "current_agent": "user", "task_completed": False }) print("最终状态:", result) if __name__ == "__main__": main()

2.4 代码运行结果解读

运行上述代码后，你会看到每个节点的执行顺序。以数学任务为例：

1. 决策智能体检测到“数学”关键词，返回Command，其中update记录了路由动作，goto="math_agent"。
2. 系统自动跳转到数学智能体，执行计算。
3. 数学智能体返回Command，goto="decision_agent"，系统再次跳回决策智能体。
4. 决策智能体发现最后一条消息不是“数学”或“翻译”，返回goto=END，流程结束。

三、Command与条件边：如何选择？

很多初学者会困惑：什么时候用Command，什么时候用条件边？下面用一个表格清晰说明：

选择原则：如果你在决定下一步去向的同时，还需要往状态里写入任何新内容（比如记录日志、切换智能体标识、保存API返回结果），就用Command。如果只是读取状态中的某个字段，根据字段值走不同的分支，不需要修改任何内容，用条件边更简洁。

四、进阶技巧：子图如何跳转回父图

4.1 问题场景

在实际的多智能体系统中，我们经常将独立的功能模块封装为子图。例如，一个数据处理子图专门负责复杂计算，计算完成后需要通知父图中的某个节点进行收尾。问题在于：子图内部的节点如何直接跳转到父图节点？

LangGraph提供了graph=Command.PARENT参数来解决这个问题。当子图节点返回Command时，设置此参数，系统会在父图范围内查找目标节点。

4.2 代码示例

""" LangGraph Command 父图导航演示 功能：子图节点完成后，直接跳转到父图的指定节点 """ from typing import Annotated from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END from langgraph.types import Command # 父图状态 class ParentState(TypedDict): messages: Annotated[list, lambda x, y: x + y] task_status: str subtask_result: str # 子图状态（继承父图字段，并添加子图专属字段） class ChildState(TypedDict): messages: Annotated[list, lambda x, y: x + y] task_status: str subtask_result: str child_data: str # 子图独有数据 # 父图节点：主控制器 def main_controller(state: ParentState) -> Command[ParentState]: print("[父图-主控制器] 启动子任务") return Command( update={ "messages": [("system", "启动子任务")], "task_status": "subtask_started" }, goto="subgraph_node" ) # 父图节点：任务结束器 def task_finisher(state: ParentState) -> dict: print("[父图-任务结束器] 执行收尾工作") print(f"子任务结果: {state['subtask_result']}") return { "messages": [("system", "任务完成")], "task_status": "completed" } # 子图节点：数据处理器 def data_processor(state: ChildState) -> Command[ParentState]: print("[子图-数据处理器] 开始处理数据") processed_data = "经过子图处理的结果" print(f"处理结果: {processed_data}") # 关键：通过 graph=Command.PARENT 跳转到父图节点 return Command( update={ "messages": [("subtask", f"子任务完成: {processed_data}")], "subtask_result": processed_data, "task_status": "subtask_completed" }, goto="task_finisher", graph=Command.PARENT ) def create_subgraph(): """创建并编译子图""" subgraph_builder = StateGraph(ChildState) subgraph_builder.add_node("data_processor", data_processor) subgraph_builder.add_edge(START, "data_processor") subgraph_builder.add_edge("data_processor", END) return subgraph_builder.compile() def main(): print("=== Command父图导航演示 ===\n") # 构建父图 parent_builder = StateGraph(ParentState) parent_builder.add_node("main_controller", main_controller) parent_builder.add_node("task_finisher", task_finisher) parent_builder.add_node("subgraph_node", create_subgraph()) parent_builder.add_edge(START, "main_controller") parent_builder.add_edge("main_controller", "subgraph_node") # 注意：subgraph_node 到 task_finisher 的边由子图内部的Command实现 graph = parent_builder.compile() initial_state = { "messages": [("user", "开始处理任务")], "task_status": "init", "subtask_result": "" } result = graph.invoke(initial_state) print("最终状态:", result) if __name__ == "__main__": main()

4.3 关键点解析

• graph=Command.PARENT告诉系统：不要在当前子图中查找goto指定的节点，而是向上查找父图。
• 子图状态必须包含父图状态的所有字段，否则跨图更新时会因缺少字段而报错。推荐使用继承方式定义子图状态。
• 这种模式非常适合实现“子任务完成后向上汇报”的场景，比如数据处理子图完成后通知主控节点。

五、实战案例：在工具函数中使用Command更新状态

5.1 业务场景

以客户支持系统为例：用户提供客户ID，系统需要调用查询工具获取客户信息，然后根据会员等级提供个性化服务，最后模拟问题解决流程。这个场景的难点在于：查询工具执行完毕后，需要同时保存查询结果并决定下一步转交客服代理。

5.2 实现思路

我们将查询工具封装为一个节点，内部调用模拟数据库查询函数。查询完成后，返回Command对象，update中写入客户信息，goto指向客服代理节点。这样就避免了额外编写条件边来处理工具返回结果。

5.3 完整代码

""" LangGraph Command 工具内部状态更新演示 场景：客户支持系统，根据客户ID查询信息并提供个性化服务 """ import time import random from typing import Annotated from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END from langgraph.types import Command # 定义状态 class SupportState(TypedDict): customer_id: str customer_info: dict messages: Annotated[list, lambda x, y: x + y] issue_resolved: bool # 模拟客户数据库 CUSTOMER_DATABASE = { "CUST001": { "name": "张三", "email": "zhangsan@example.com", "membership_level": "金牌会员", "account_status": "正常" }, "CUST002": { "name": "李四", "email": "lisi@example.com", "membership_level": "银牌会员", "account_status": "正常" }, "CUST003": { "name": "王五", "email": "wangwu@example.com", "membership_level": "普通会员", "account_status": "欠费" } } def lookup_customer_info(customer_id: str) -> dict: """模拟查询客户信息的工具函数""" print(f"正在查询客户 {customer_id} 的信息...") time.sleep(0.5) # 模拟延迟 customer_info = CUSTOMER_DATABASE.get(customer_id, {}) if customer_info: print(f"找到客户: {customer_info['name']}") else: print("未找到客户信息") customer_info = {"error": "客户未找到"} return customer_info # 节点：客户信息查询工具 def customer_lookup_tool(state: SupportState) -> Command[SupportState]: print("[查询工具] 执行中...") customer_id = state["customer_id"] customer_info = lookup_customer_info(customer_id) # 使用Command同时更新状态并决定下一步 return Command( update={ "customer_info": customer_info, "messages": [("system", f"已查询客户 {customer_id} 的信息")] }, goto="support_agent" ) # 节点：客服代理 def support_agent(state: SupportState) -> Command[SupportState]: print("[客服代理] 处理中...") customer_info = state["customer_info"] # 未找到客户的情况 if "error" in customer_info: response = "抱歉，未找到您的客户信息，请确认客户ID是否正确。" print(f"回复: {response}") return Command( update={"messages": [("assistant", response)]}, goto=END ) # 根据会员等级提供个性化回复 name = customer_info["name"] level = customer_info["membership_level"] if level == "金牌会员": response = f"尊敬的金牌会员{name}，我们将优先为您处理。" elif level == "银牌会员": response = f"{name}您好，我们会尽快为您解决问题。" else: response = f"{name}您好，感谢您的咨询。" response += "\n请问您需要什么帮助？" print(f"回复: {response}") return Command( update={"messages": [("assistant", response)]}, goto="issue_resolver" ) # 节点：问题解决器 def issue_resolver(state: SupportState) -> Command[SupportState]: print("[问题解决器] 处理中...") time.sleep(0.5) # 随机模拟解决结果 resolved = random.choice([True, False]) if resolved: response = "您的问题已成功解决。" issue_status = True print("问题已解决") else: response = "问题需要进一步处理，专员会联系您。" issue_status = False print("问题需要升级处理") return Command( update={ "messages": [("system", response)], "issue_resolved": issue_status }, goto=END ) def main(): print("=== Command工具内部状态更新演示 ===\n") # 构建图 builder = StateGraph(SupportState) builder.add_node("customer_lookup_tool", customer_lookup_tool) builder.add_node("support_agent", support_agent) builder.add_node("issue_resolver", issue_resolver) builder.add_edge(START, "customer_lookup_tool") graph = builder.compile() # 测试1：金牌会员 print("测试1: 金牌会员客户 CUST001") result = graph.invoke({ "customer_id": "CUST001", "customer_info": {}, "messages": [("user", "我需要查询账户信息")], "issue_resolved": False }) print(f"最终状态: customer_info={result['customer_info']}, issue_resolved={result['issue_resolved']}") print("\n" + "="*50 + "\n") # 测试2：不存在的客户 print("测试2: 不存在的客户 CUST999") result = graph.invoke({ "customer_id": "CUST999", "customer_info": {}, "messages": [("user", "查询账户信息")], "issue_resolved": False }) print(f"最终状态: {result['customer_info']}") print("\n" + "="*50 + "\n") # 测试3：普通会员 print("测试3: 普通会员客户 CUST003") result = graph.invoke({ "customer_id": "CUST003", "customer_info": {}, "messages": [("user", "账户有问题需要处理")], "issue_resolved": False }) print(f"最终状态: customer_info={result['customer_info']}, issue_resolved={result['issue_resolved']}") if __name__ == "__main__": main()

5.4 运行结果分析

从三个测试用例可以看出：

• 测试1（金牌会员）：查询工具成功获取客户信息，客服代理识别到“金牌会员”层级提供了优先处理话术，问题解决器随机决定解决状态。
• 测试2（不存在的客户）：查询工具返回错误信息，客服代理检测到错误后直接结束流程，没有进入问题解决器，避免了无效处理。
• 测试3（普通会员）：正常查询到客户信息，提供标准回复，问题解决器进行后续处理。

这个案例展示了Command在工具节点中的典型应用：工具执行后，立刻用Command返回结果并指定下一步，无需额外的条件边。

六、总结

6.1 核心要点回顾

1. Command的本质：一个同时包含update（状态更新）和goto（流程控制）的返回值，让节点能够一站式完成“处理业务 + 决定去向”。
2. 适用场景：
3. 多智能体系统中需要在切换智能体时更新标识和传递信息
4. 工具或API调用后需要立即保存结果并路由到下一个节点
5. 子图任务完成后需要向上跳转到父图节点
6. 与条件边的区别：条件边只做路由判断不修改状态，Command两者兼具。需要修改状态时选择Command，否则用条件边更轻量。
7. 父图导航：通过graph=Command.PARENT实现从子图到父图的逆向跳转，是多层图结构中任务交接的关键技术。

6.2 最佳实践建议

• 状态字段中需要累加的数据（如消息历史）务必使用Annotated配合reducer函数，避免意外覆盖。
• 每个Command都应明确指定goto目标，除非确实需要结束流程（goto=END）。
• 子图状态建议继承父图状态，确保字段兼容性，避免跨图更新时报错。
• 在开发阶段，每个节点入口添加打印语句，有助于理解执行顺序和调试。

6.3 结语

LangGraph Command提供了一种优雅的方式来整合状态管理和流程控制，使得构建复杂的多智能体系统变得更加直观。掌握Command，你就能够在每个节点中同时完成“做事”和“指路”，让智能体协作如同流水线般顺畅。希望本文的三个案例能够帮助你在实际项目中灵活运用这一强大工具。