从零开始学LangGraph（12）：状态更新与流程控制一体化！如何用Command赋予节点决策权

关注我~第一时间学习如何更好地使用AI。

重要的不是我们是否会被AI替代，

而是我们要比被替代的人更懂AI。

前期导览：

从零开始学LangGraph（1）：Chat Model --- 如何通过代码与模型对话（上）

从零开始学LangGraph（2）：Message和Template —— 如何通过代码与模型对话（下）

从零开始学LangGraph（3）：Tools --- 如何实现大模型与外部系统对接

从零开始学LangGraph（4）：用Excel表格来理解LangGraph的基本工作原理

从零开始学LangGraph（5）：手把手教你搭建简易Graph

从零开始学LangGraph（6）：轻松玩转Conditional Edges

从零开始学LangGraph（7）：手把手教你手搓带Memory的Chat Bot

从零开始学LangGraph（8）：Tools + Subgraph实战，手把手教你构建ReAct Agent

从零开始学LangGraph（9）：详解State的定义、更新与管理

番外：Deep Agent入门，60行代码实现网络检索小助手

从零开始学LangGraph（10）：利用Runtime自定义实现Model和System Prompt的灵活切换

从零开始学LangGraph（11）：动态控制流与批量并行处理！如何用Send实现Map-Reduce

大家好，上一期我们学习了Send ，它让我们能够实现在Graph运行的过程中动态地创建边，传递自定义的状态，并实现Map-Reduce 这种高效的数据处理模式。今天，我们要学习另一个非常强大且实用的功能：Command 。

Command是LangGraph中一个特殊的返回类型，它允许节点在执行过程中同时更新Graph的状态并控制流程走向 。这个功能看似简单，但在实际应用中却能解决很多复杂场景下的问题。

简单来说，Command就像是给节点赋予了"决策权"——它不仅能够处理数据、更新状态，还能直接决定下一步要执行哪个节点，而不需要依赖外部的边或路由函数。这种"一体化"的设计让我们的代码更加集中、清晰。

注意：文章的最后，我用chat model、subgraph、command给大家搭建了一个简易的能够对用户问题进行分类并针对性回复的智能客服系统，一定不要错过~

什么是Command

在理解Command之前，我们先回顾一下之前学过的知识：

• • Node（节点） ：负责处理业务逻辑，通常返回一个字典来更新State
• • Edge（边） ：负责连接节点，决定数据流向
• • Conditional Edge（条件边） ：根据条件决定走哪条边

传统的LangGraph工作流中，节点和边的职责是分离的：节点负责处理逻辑和更新状态，边负责控制流程。但在某些场景下，我们需要在节点内部同时完成状态更新和流程控制 ，这时候Command就派上用场了。

Command本质上是一个特殊的对象 ，节点函数可以返回它来同时执行两个操作：

更新Graph的状态 （通过 update 参数）

指定下一个执行的节点 （通过 goto 参数）

为什么需要Command

传统方式的局限性

在引入Command之前，我们先回顾一下如何通过Conditional Edge 来实现 "根据节点执行结果动态决定下一个节点" 的功能，这里主要包括两个环节。

首先，我们需要一个节点函数来负责state的更新：

  
def my\_node(state: State) -> State:  
    # 处理逻辑  
    if some\_condition:  
        state["status"] = "success"  
    else:  
        state["status"] = "failure"  
    return state

然后，我们需要编写路由函数，并添加条件边，来实现控制流：

  
def routing\_function(state: State) -> str:  
    if state["status"] == "success":  
        return "success\_node"  
    else:  
        return "failure\_node"  
  
graph.add\_node("my\_node", my\_node)  
graph.add\_conditional\_edges("my\_node", routing\_function, {  
    "success\_node": "success\_node",  
    "failure\_node": "failure\_node"  
})

这种方式的特点在于：节点和路由逻辑是分离的 。我们需要先更新状态，然后在另一个函数中读取状态来决定路由。这样，当我们的逻辑变得非常复杂时，代码就会变得分散，难以维护。

Command的优势

使用Command后，同样的功能只需要一个节点函数 就能完成：

  
from langgraph.types import Command  
from typing importLiteral  
  
defmy\_node(state: State) -> Command[Literal["success\_node", "failure\_node"]]:  
    # 在一个函数中同时完成状态更新和路由决策  
    if some\_condition:  
        return Command(  
            update={"status": "success"},  # 更新状态  
            goto="success\_node"              # 决定路由  
        )  
    else:  
        return Command(  
            update={"status": "failure"},    # 更新状态  
            goto="failure\_node"              # 决定路由  
        )  
  
# 只需要添加节点，不需要额外的路由函数  
graph.add\_node("my\_node", my\_node)

如上述代码所示，使用Command后，在一个节点函数内部就能同时完成状态更新和路由决策，这样逻辑集中在一个地方，代码更清晰、简洁、易于维护。

Command的参数详解

Command类的使用关键，主要在于对四个参数的运用。我们先从最常用的两个开始说起。

1. update 参数

update参数用来更新Graph的状态，用法很简单，就是传入一个字典，字典里的键值对会更新到State中。比如：

  
def my\_node(state: State) -> Command:  
    return Command(  
        update={  
            "user\_info": {"name": "张三", "age": 25},  
            "status": "processed"  
        },  
        goto="next\_node"  
    )

这里，update参数会更新State中的user\_info和status字段。

2. goto 参数

goto参数用来指定下一个要执行的节点，这是Command最核心的功能之一。它支持多种形式，我们可以传入节点名称（字符串）、节点序列、Send对象 等。

最简单的用法就是传入一个节点名称 ：

  
from langgraph.types import Command  
from typing import Literal  
  
def my\_node(state: State) -> Command[Literal["next\_node"]]:  
    return Command(  
        update={"status": "done"},  
        goto="next\_node"  # 跳转到next\_node节点  
    )

如果需要按顺序执行多个节点，可以传入一个节点序列 ：

  
def my\_node(state: State) -> Command[Literal["node\_a", "node\_b", "node\_c"]]:  
    return Command(  
        update={"status": "done"},  
        goto=["node\_a", "node\_b", "node\_c"]  # 按顺序执行这三个节点  
    )

goto参数还可以接受Send对象 ，这样就能在Command中实现更复杂的动态路由。比如我们可以传入Send对象列表来实现Map-Reduce模式：

  
from langgraph.graph import Send  
  
def distribute\_tasks(state: State) -> Command:  
    """将任务分发到多个处理节点"""  
    tasks = state["tasks"]  
    return Command(  
        update={"status": "distributed"},  
        goto=[Send("process\_task", {"task": task}) for task in tasks]  # 使用Send列表实现并行处理  
    )

需要注意的是，当goto是节点序列时，这些节点会按顺序执行；如果传入的是Send对象列表，LangGraph会尝试并行执行这些Send指向的节点。

3. graph 参数

graph参数用来指定要发送命令的目标Graph，默认值是None，表示当前Graph。这个参数主要用于子图场景。

需要说明的是，graph参数和goto参数需要配合理解，即goto参数指定的节点名称，必须是graph参数指定的Graph内部的节点。也就是说：

• • 如果 graph=None （默认值）， goto 指向的是当前Graph的节点
• • 如果 graph=Command.PARENT ， goto 指向的是父图的节点（必须是父图中存在的节点）

换言之，当我们在子图中执行节点时，如果需要跳转到父图的某个节点，就需要使用graph=Command.PARENT，并且goto参数的值必须是父图中存在的节点名称。比如：

  
from langgraph.types import Command  
from typing import Literal  
  
def subgraph\_node(state: State) -> Command[Literal["parent\_node"]]:  
    # 在子图中执行某些逻辑后，跳转到父图的节点  
    return Command(  
        update={"result": "completed"},  
        goto="parent\_node",  # 这个节点必须是父图中存在的节点  
        graph=Command.PARENT  # 告诉LangGraph要跳转到父图，而不是当前子图  
    )

这个功能在多智能体交接的场景中特别有用，可以实现跨层级的流程控制。

4. resume 参数

resume参数通常与interrupt()函数配合使用，这个功能相对高级，主要用于需要中断Graph执行、等待外部输入或异步操作完成的场景。

比如，当Graph执行到某个节点时，我们可以使用interrupt()来暂停执行，等待用户输入或其他异步操作完成，然后通过resume参数来恢复执行：

  
def my\_node(state: State) -> Command:  
    return Command(  
        update={"status": "waiting"},  
        resume={"interrupt\_id\_1": "resume\_value"},  # 恢复指定的中断  
        goto="next\_node"  
    )

这个功能在实现人机交互（human-in-the-loop）的工作流中非常有用，但展开讲会比较复杂，我们会在后续文章中详细介绍。

Command的典型使用场景

在实际业务中，Command主要有四个典型的使用场景。下面我们通过具体的业务例子来看看每个场景的应用。

场景一：动态控制流

在电商订单处理系统中，不同金额的订单需要走不同的审核流程。小额订单可以自动通过，大额订单需要人工审核。使用Command可以在一个节点中同时完成状态更新和流程控制。

  
def check\_order(state: State) -> Command[Literal["auto\_approve", "manual\_review"]]:  
    """根据订单金额决定审核流程"""  
    order\_amount = state["order\_amount"]  
      
    if order\_amount < 1000:  
        # 小额订单：更新状态并路由到自动审核节点  
        return Command(  
            update={"review\_status": "auto\_approved"},  
            goto="auto\_approve"  
        )  
    else:  
        # 大额订单：更新状态并路由到人工审核节点  
        return Command(  
            update={"review\_status": "pending\_review"},  
            goto="manual\_review"  
        )

这样，系统就能根据订单金额自动选择不同的处理流程，同时更新订单的审核状态。

场景二：工具中的状态更新

在智能客服系统中，当用户咨询时，我们需要先查询用户的基本信息（如VIP等级、历史订单等），以便提供更个性化的服务。工具执行后可以直接更新Graph状态，后续节点就能直接使用这些信息。

  
from langchain\_core.tools import tool  
from langgraph.types import Command  
  
@tool  
deflookup\_customer\_info(user\_id: str) -> Command:  
    """查询用户信息并更新Graph状态"""  
    # 模拟查询用户信息  
    customer\_info = {  
        "name": "张三",  
        "vip\_level": "gold",  
        "order\_count": 15  
    }  
      
    # 从工具返回Command，直接更新Graph状态  
    return Command(  
        update={  
            "customer\_info": customer\_info  
        }  
    )

在这个示例中，lookup\_customer\_info工具函数接收user\_id参数，然后查询用户信息（示例中使用模拟数据，实际场景中应该从数据库或API查询）。查询到的customer\_info，再通过Command的update参数直接更新到Graph状态中。这样，后续的客服节点就可以直接使用state["customer\_info"]来获取用户信息，而不需要从消息中解析。

场景三：子图中的导航

在订单处理系统中，订单验证是一个复杂的流程，通常包含多个步骤：检查商品库存、验证商品价格是否变动、检查用户是否有购买权限、验证收货地址是否有效等。这些验证步骤逻辑相对独立，但又需要作为一个整体来执行。为了代码的模块化和可维护性，我们可以将订单验证流程封装成一个子图（Subgraph）。

而当子图完成所有验证步骤后，需要跳转回主流程的某个节点（比如支付节点）继续执行。这时候就需要使用graph=Command.PARENT来告诉LangGraph，我们要跳转到的是父图（主流程）中的节点，而不是当前子图中的节点。

  
def validate\_order\_complete(state: State) -> Command[Literal["payment"]]:  
    """订单验证子图完成，跳转到主流程的支付节点"""  
    # 验证完成后的处理  
    validation\_result = {  
        "is\_valid": True,  
        "validation\_time": "2024-01-01 10:00:00"  
    }  
      
    return Command(  
        update={"validation\_result": validation\_result},  
        goto="payment",  # 跳转到主流程的支付节点  
        graph=Command.PARENT  # 告诉LangGraph这是父图的节点  
    )

这样，订单验证子图完成后，就能直接跳转到主流程的支付节点，实现跨层级的流程控制。

场景四：多智能体交接

在智能客服系统中，当普通客服无法解决用户的问题时，需要将问题转交给专家客服。转交时需要传递问题的上下文信息，让专家客服能够快速了解情况。

  
def regular\_service(state: State) -> Command[Literal["expert\_service", "end"]]:  
    """普通客服处理用户问题"""  
    # 普通客服尝试解决问题  
    tried\_solutions = ["重启应用", "清除缓存", "检查网络连接"]  
      
    # 判断是否需要转交给专家  
    if state["problem\_complexity"] == "high":  
        # 准备转交信息，跳转到专家客服  
        return Command(  
            update={  
                "tried\_solutions": tried\_solutions,  
                "current\_handler": "expert",  
                "transfer\_reason": "问题复杂，需要专业技术支持"  
            },  
            goto="expert\_service"# 转交给专家客服节点  
        )  
    else:  
        # 普通客服可以解决，流程结束  
        return Command(  
            update={"status": "resolved"},  
            goto="end"  
        )  
  
defexpert\_service(state: State) -> State:  
    """专家客服接收转交并处理"""  
    # 专家客服从State中获取转交信息  
    print(f"接收转交：{state['transfer\_reason']}")  
    print(f"已尝试方案：{state['tried\_solutions']}")  
    # 专家客服提供专业解决方案  
    return {"status": "expert\_resolved", "solution": "专业技术方案"}

在这个示例中，regular\_service节点代表普通客服，它会先尝试解决用户问题。当判断问题过于复杂（problem\_complexity == "high"）时，普通客服会通过Command将已尝试的解决方案和转交原因更新到State中，并跳转到expert\_service节点。专家客服节点接收到转交后，可以从State中获取tried\_solutions和transfer\_reason等信息，了解问题的背景和已尝试的方案，从而提供更专业、更有针对性的解决方案。这样就实现了两个智能体之间的无缝交接。

从这四个场景可以看出，Command让我们能够在节点中同时完成状态更新和流程控制，代码更加集中、清晰，特别适合需要动态决定流程走向的业务场景。

使用Command的注意事项

在使用Command时，有几个重要的注意事项需要了解：

1. 类型提示的重要性

使用类型提示非常重要！对于返回Command的节点函数，必须 使用Command[Literal[...]]类型来指定goto参数的可能值。这不仅可以帮助IDE和类型检查工具更好地理解代码，更重要的是：

• • Graph渲染 ：LangGraph需要这个类型提示来正确渲染Graph结构
• • 节点识别 ：告诉LangGraph这个节点可以导航到哪些节点

如果不加类型提示，Graph可能无法正确识别所有可能的路径。

  
from langgraph.types import Command  
from typing import Literal  
  
def my\_node(state: State) -> Command[Literal["node\_a", "node\_b"]]:  
    # 这样类型检查器就知道goto只能是"node\_a"或"node\_b"  
    # 同时，这个类型提示对Graph的渲染也很重要，告诉LangGraph这个节点可以导航到哪些节点  
    if condition:  
        return Command(update={}, goto="node\_a")  
    else:  
        return Command(update={}, goto="node\_b")

2. update参数的合并规则

Command的update参数会按照State的Reducer规则进行合并。如果State中某个字段没有指定Reducer，默认行为是覆盖 （后写入的值会覆盖先写入的值）。

  
# State定义  
class AppState(TypedDict):  
    count: int  # 没有Reducer，默认覆盖  
    items: Annotated[list[str], operator.add]  # 有Reducer，会合并  
  
# Command更新  
Command(  
    update={  
        "count": 10,  # 会覆盖之前的值  
        "items": ["new\_item"]  # 会通过operator.add合并到列表中  
    }  
)

3. 子图中使用Command.PARENT的编译注意事项

当子图中的节点使用Command(graph=Command.PARENT)跳转到父图的节点时，需要注意：

• • 编译时验证 ：由于目标节点在父图中，子图编译时无法验证该节点是否存在，LangGraph会要求子图必须有明确的出口（边）
• • 临时END边 ：我们需要添加一个临时的END边来通过编译验证，但实际运行时会被Command覆盖，不会真正执行到END

  
# 子图中的节点使用Command跳转到父图  
defsubgraph\_node(state: State) -> Command:  
    return Command(  
        update={"status": "done"},  
        goto="parent\_node",  
        graph=Command.PARENT  
    )  
  
# 构建子图  
subgraph = StateGraph(State)  
subgraph.add\_node("subgraph\_node", subgraph\_node)  
subgraph.add\_edge(START, "subgraph\_node")  
# 注意：需要添加临时的END边来通过编译验证  
# 实际运行时会被Command覆盖，不会真正执行到END  
subgraph.add\_edge("subgraph\_node", END)  
subgraph\_app = subgraph.compile()

这样设计的好处是：子图在编译时结构完整，运行时通过Command动态跳转，既保证了代码的正确性，又实现了灵活的流程控制。

实战示例：构建一个智能客服系统

下面我们通过一个完整的示例来演示Command的使用。这个示例模拟了一个智能客服系统，使用Chat Model 来分析用户问题，通过Subgraph 封装问题分类逻辑，并使用Command 实现子图到主图的流程控制。

1. 导入必要的模块

  
from typing import TypedDict, Annotated, Sequence, Literal  
from langchain\_core.messages import AnyMessage, HumanMessage, SystemMessage  
from langgraph.graph.message import add\_messages  
from langgraph.graph import StateGraph, START, END  
from langgraph.types import Command  
from dotenv import load\_dotenv  
from langchain.chat\_models import init\_chat\_model

2. 初始化Chat Model

  
# 初始化LLM  
load\_dotenv()  
  
llm = init\_chat\_model("deepseek-chat", model\_provider="deepseek")

3. 定义State

  
class CustomerServiceState(TypedDict):  
    messages: Annotated[Sequence[AnyMessage], add\_messages]  
    question\_type: str  
    status: str  
    result: str

4. 构建分类子图

子图负责使用Chat Model分析用户问题，并将分类结果传递给主图。子图完成后，使用Command跳转到主图的route\_to\_support路由节点，由该节点根据分类结果决定路由到哪个处理节点。这种两层路由设计使流程更加清晰，也便于处理LLM输出的各种格式。

  
def analyze\_with\_llm(state: CustomerServiceState) -> CustomerServiceState:  
    """使用Chat Model分析用户问题"""  
    last\_message = state["messages"][-1]  
      
    system\_prompt = SystemMessage(  
        content="""你是一个智能客服问题分类助手。你的任务是对用户问题进行精确分类。  
  
## 分类标准：  
  
**技术问题**：涉及系统故障、功能异常、操作错误、无法使用、报错、bug、登录失败、页面打不开、功能无法使用等问题。  
- 关键词：无法、错误、故障、打不开、登录不了、不能用、报错、bug、异常、失败等  
- 示例："我的账户无法登录了"、"页面打不开"、"功能报错"  
  
**账单问题**：涉及费用、付款、退款、账单查询、价格、扣费、充值、余额、发票等问题。  
- 关键词：账单、付款、退款、价格、费用、扣费、充值、余额、发票、明细等  
- 示例："我想查询账单"、"如何退款"、"费用是多少"  
  
**一般咨询**：其他所有不涉及技术故障和账单的问题，如营业时间、服务介绍、政策说明、使用指导等。  
- 示例："营业时间是什么时候"、"你们有什么服务"、"如何使用"  
  
## 输出要求（非常重要）：  
1. 仔细分析用户问题的关键词和核心诉求  
2. 必须严格按照以上三类进行分类  
3. **你的回答格式必须严格按照以下格式，不要添加任何其他文字、解释或推理过程：**  
     
   技术问题  
     
   或者  
     
   账单问题  
     
   或者  
     
   一般咨询  
  
4. **只输出以上三种之一，不要输出推理过程或其他内容**  
  
现在请对用户问题进行分类，严格按照以上三种之一输出："""  
    )  
      
    response = llm.invoke([system\_prompt, last\_message])  
    analysis\_result = response.content.strip()  
        
    return {  
        "question\_type": analysis\_result,  
        "status": "analyzed"  
    }  
  
defclassify\_and\_route\_in\_subgraph(state: CustomerServiceState):  
    """子图中的路由节点：根据分析结果，使用Command跳转到主图的路由节点"""  
    # 子图完成分析后，使用 Command 跳转到主图的 route\_to\_support 节点  
    # 该节点会根据 question\_type 决定跳转到哪个支持节点  
    # 重要：必须保留 question\_type，否则主图无法正确路由  
    question\_type = state.get("question\_type", "一般咨询")  
    return Command(  
        update={"status": "classified", "question\_type": question\_type},  
        goto="route\_to\_support",  
        graph=Command.PARENT  
    )  
  
# 构建分类子图（子图执行完后使用Command跳转回主图）  
classification\_subgraph = StateGraph(CustomerServiceState)  
classification\_subgraph.add\_node("analyze\_with\_llm", analyze\_with\_llm)  
classification\_subgraph.add\_node("classify\_and\_route", classify\_and\_route\_in\_subgraph)  
classification\_subgraph.add\_edge(START, "analyze\_with\_llm")  
classification\_subgraph.add\_edge("analyze\_with\_llm", "classify\_and\_route")  
# 注意：classify\_and\_route 使用 Command(graph=Command.PARENT) 跳转到主图  
# 由于目标节点在主图中，子图编译时无法验证，但运行时可以正常工作  
# 我们需要添加一个临时的END边来通过编译验证，但实际运行时会被Command覆盖  
classification\_subgraph.add\_edge("classify\_and\_route", END)  
classification\_app = classification\_subgraph.compile()

注意 ：在classify\_and\_route\_in\_subgraph函数中，update参数必须包含question\_type字段。这是因为子图执行完成后，状态会传递到主图的route\_to\_support节点，该节点需要根据question\_type来决定路由到哪个支持节点。如果不在update中显式保留question\_type，主图节点可能无法获取到正确的分类结果，导致路由失败。这是使用Command进行跨层级跳转时的一个关键点：必须确保子图产生的关键状态数据通过update参数传递到父图 。

5. 定义主图节点

  
def route\_to\_support(state: CustomerServiceState) -> Command[  
    Literal["technical\_support", "billing\_support", "general\_support"]  
]:  
    """主图中的路由节点：根据子图分析的结果，使用Command跳转到相应的支持节点"""  
    question\_type = state.get("question\_type", "一般咨询")  
      
    # 更健壮的匹配逻辑，处理各种可能的输出格式  
    question\_type\_lower = question\_type.lower()  
      
    if"技术"in question\_type or"technical"in question\_type\_lower:  
        next\_node = "technical\_support"  
    elif"账单"in question\_type or"billing"in question\_type\_lower:  
        next\_node = "billing\_support"  
    else:  
        # 默认为一般咨询  
        next\_node = "general\_support"  
      
    return Command(goto=next\_node)  
  
deftechnical\_support(state: CustomerServiceState) -> CustomerServiceState:  
    """处理技术支持问题"""  
    result = "已为您创建技术支持工单，工单号：TS-2024-001"  
    print(f"\n[技术支持] {result}\n")  
    return {"status": "processed", "result": result}  
  
defbilling\_support(state: CustomerServiceState) -> CustomerServiceState:  
    """处理账单相关问题"""  
    result = "已为您查询账单信息，如有疑问请联系财务部门"  
    print(f"\n[账单支持] {result}\n")  
    return {"status": "processed", "result": result}  
  
defgeneral\_support(state: CustomerServiceState) -> CustomerServiceState:  
    """处理一般咨询问题"""  
    result = "已为您提供相关信息，如需进一步帮助请随时联系"  
    print(f"\n[一般支持] {result}\n")  
    return {"status": "processed", "result": result}

6. 构建主图

  
# 构建主图  
main\_graph = StateGraph(CustomerServiceState)  
main\_graph.add\_node("classification\_subgraph", classification\_app)  
main\_graph.add\_node("route\_to\_support", route\_to\_support)  
main\_graph.add\_node("technical\_support", technical\_support)  
main\_graph.add\_node("billing\_support", billing\_support)  
main\_graph.add\_node("general\_support", general\_support)  
  
main\_graph.add\_edge(START, "classification\_subgraph")  
# 注意：子图中的 classify\_and\_route 节点使用 Command(graph=Command.PARENT)   
# 会跳转到主图的 route\_to\_support 节点  
# route\_to\_support 节点再根据 question\_type 使用 Command 跳转到相应的支持节点  
# 因此不需要从 classification\_subgraph 或 route\_to\_support 添加静态边  
main\_graph.add\_edge("technical\_support", END)  
main\_graph.add\_edge("billing\_support", END)  
main\_graph.add\_edge("general\_support", END)  
  
app = main\_graph.compile()

7. 使用Graph

  
# 测试不同类型的问题  
test\_cases = [  
    "我的账户无法登录了，一直提示密码错误",  # 应该路由到technical\_support  
    "我想查询一下这个月的账单明细",  # 应该路由到billing\_support  
    "你们公司的营业时间是什么时候？"# 应该路由到general\_support  
]  
  
for question in test\_cases:  
    print(f"\n{'='*50}")  
    print(f"用户问题：{question}")  
    print("-" * 50)  
      
    result = app.invoke({  
        "messages": [HumanMessage(content=question)],  
        "question\_type": "",  
        "status": "",  
        "result": ""  
    })  
      
    print(f"\n问题类型：{result['question\_type']}")  
    print(f"最终状态：{result['status']}")  
    print(f"处理结果：{result['result']}")  
    print("="*50)

运行结果会显示，系统使用Chat Model分析用户问题，通过子图进行分类，然后使用Command跳转到主图的相应处理节点。

这个示例展示了Command的核心优势：在子图中完成状态更新和流程控制，并能够跳转回父图的节点 ，实现了模块化设计和清晰的代码结构。同时，通过在主图中增加路由节点，我们可以更灵活地处理LLM输出的各种格式，提高系统的健壮性。

完整代码

为了方便大家理解和复现，我把完整的代码整理如下：

  
from typing import TypedDict, Annotated, Sequence, Literal  
from langchain\_core.messages import AnyMessage, HumanMessage, SystemMessage  
from langgraph.graph.message import add\_messages  
from langgraph.graph import StateGraph, START, END  
from langgraph.types import Command  
from dotenv import load\_dotenv  
load\_dotenv()  
  
from langchain.chat\_models import init\_chat\_model  
  
# 初始化LLM  
llm = init\_chat\_model("deepseek-chat", model\_provider="deepseek")  
  
# 定义State  
classCustomerServiceState(TypedDict):  
    messages: Annotated[Sequence[AnyMessage], add\_messages]  
    question\_type: str  
    status: str  
    result: str  
  
# 子图节点函数  
defanalyze\_with\_llm(state: CustomerServiceState) -> CustomerServiceState:  
    """使用Chat Model分析用户问题"""  
    last\_message = state["messages"][-1]  
      
    system\_prompt = SystemMessage(  
        content="""你是一个智能客服问题分类助手。你的任务是对用户问题进行精确分类。  
  
## 分类标准：  
  
**技术问题**：涉及系统故障、功能异常、操作错误、无法使用、报错、bug、登录失败、页面打不开、功能无法使用等问题。  
- 关键词：无法、错误、故障、打不开、登录不了、不能用、报错、bug、异常、失败等  
- 示例："我的账户无法登录了"、"页面打不开"、"功能报错"  
  
**账单问题**：涉及费用、付款、退款、账单查询、价格、扣费、充值、余额、发票等问题。  
- 关键词：账单、付款、退款、价格、费用、扣费、充值、余额、发票、明细等  
- 示例："我想查询账单"、"如何退款"、"费用是多少"  
  
**一般咨询**：其他所有不涉及技术故障和账单的问题，如营业时间、服务介绍、政策说明、使用指导等。  
- 示例："营业时间是什么时候"、"你们有什么服务"、"如何使用"  
  
## 输出要求（非常重要）：  
1. 仔细分析用户问题的关键词和核心诉求  
2. 必须严格按照以上三类进行分类  
3. **你的回答格式必须严格按照以下格式，不要添加任何其他文字、解释或推理过程：**  
     
   技术问题  
     
   或者  
     
   账单问题  
     
   或者  
     
   一般咨询  
  
4. **只输出以上三种之一，不要输出推理过程或其他内容**  
  
现在请对用户问题进行分类，严格按照以上三种之一输出："""  
    )  
      
    response = llm.invoke([system\_prompt, last\_message])  
    analysis\_result = response.content.strip()  
        
    return {  
        "question\_type": analysis\_result,  
        "status": "analyzed"  
    }  
  
defclassify\_and\_route\_in\_subgraph(state: CustomerServiceState):  
    """子图中的路由节点：根据分析结果，使用Command跳转到主图的路由节点"""  
    # 子图完成分析后，使用 Command 跳转到主图的 route\_to\_support 节点  
    # 该节点会根据 question\_type 决定跳转到哪个支持节点  
    # 重要：必须保留 question\_type，否则主图无法正确路由  
    question\_type = state.get("question\_type", "一般咨询")  
    return Command(  
        update={"status": "classified", "question\_type": question\_type},  
        goto="route\_to\_support",  
        graph=Command.PARENT  
    )  
  
# 构建分类子图（子图执行完后使用Command跳转回主图）  
classification\_subgraph = StateGraph(CustomerServiceState)  
classification\_subgraph.add\_node("analyze\_with\_llm", analyze\_with\_llm)  
classification\_subgraph.add\_node("classify\_and\_route", classify\_and\_route\_in\_subgraph)  
classification\_subgraph.add\_edge(START, "analyze\_with\_llm")  
classification\_subgraph.add\_edge("analyze\_with\_llm", "classify\_and\_route")  
# 注意：classify\_and\_route 使用 Command(graph=Command.PARENT) 跳转到主图  
# 由于目标节点在主图中，子图编译时无法验证，但运行时可以正常工作  
# 我们需要添加一个临时的END边来通过编译验证，但实际运行时会被Command覆盖  
classification\_subgraph.add\_edge("classify\_and\_route", END)  
classification\_app = classification\_subgraph.compile()  
  
# 主图节点函数  
defroute\_to\_support(state: CustomerServiceState) -> Command[  
    Literal["technical\_support", "billing\_support", "general\_support"]  
]:  
    """主图中的路由节点：根据子图分析的结果，使用Command跳转到相应的支持节点"""  
    question\_type = state.get("question\_type", "一般咨询")  
      
    # 更健壮的匹配逻辑，处理各种可能的输出格式  
    question\_type\_lower = question\_type.lower()  
      
    if"技术"in question\_type or"technical"in question\_type\_lower:  
        next\_node = "technical\_support"  
    elif"账单"in question\_type or"billing"in question\_type\_lower:  
        next\_node = "billing\_support"  
    else:  
        # 默认为一般咨询  
        next\_node = "general\_support"  
      
    return Command(goto=next\_node)  
  
deftechnical\_support(state: CustomerServiceState) -> CustomerServiceState:  
    """处理技术支持问题"""  
    result = "已为您创建技术支持工单，工单号：TS-2024-001"  
    print(f"\n[技术支持] {result}\n")  
    return {"status": "processed", "result": result}  
  
defbilling\_support(state: CustomerServiceState) -> CustomerServiceState:  
    """处理账单相关问题"""  
    result = "已为您查询账单信息，如有疑问请联系财务部门"  
    print(f"\n[账单支持] {result}\n")  
    return {"status": "processed", "result": result}  
  
defgeneral\_support(state: CustomerServiceState) -> CustomerServiceState:  
    """处理一般咨询问题"""  
    result = "已为您提供相关信息，如需进一步帮助请随时联系"  
    print(f"\n[一般支持] {result}\n")  
    return {"status": "processed", "result": result}  
  
# 构建主图  
main\_graph = StateGraph(CustomerServiceState)  
main\_graph.add\_node("classification\_subgraph", classification\_app)  
main\_graph.add\_node("route\_to\_support", route\_to\_support)  
main\_graph.add\_node("technical\_support", technical\_support)  
main\_graph.add\_node("billing\_support", billing\_support)  
main\_graph.add\_node("general\_support", general\_support)  
  
main\_graph.add\_edge(START, "classification\_subgraph")  
# 注意：子图中的 classify\_and\_route 节点使用 Command(graph=Command.PARENT)   
# 会跳转到主图的 route\_to\_support 节点  
# route\_to\_support 节点再根据 question\_type 使用 Command 跳转到相应的支持节点  
# 因此不需要从 classification\_subgraph 或 route\_to\_support 添加静态边  
main\_graph.add\_edge("technical\_support", END)  
main\_graph.add\_edge("billing\_support", END)  
main\_graph.add\_edge("general\_support", END)  
  
app = main\_graph.compile()  
  
# 使用Graph  
test\_cases = [  
    "我的账户无法登录了，一直提示密码错误",  
    "我想查询一下这个月的账单明细",  
    "你们公司的营业时间是什么时候？"  
]  
  
for question in test\_cases:  
    print(f"\n{'='*50}")  
    print(f"用户问题：{question}")  
    print("-" * 50)  
      
    result = app.invoke({  
        "messages": [HumanMessage(content=question)],  
        "question\_type": "",  
        "status": "",  
        "result": ""  
    })  
      
    print(f"\n问题类型：{result['question\_type']}")  
    print(f"最终状态：{result['status']}")  
    print(f"处理结果：{result['result']}")  
    print("="*50)

总结

通过今天的学习，我们掌握了LangGraph中Command的核心概念和使用方法。

Command是LangGraph中一个特殊的返回类型，它允许节点在执行过程中同时更新Graph的状态并控制流程走向 。简单来说，Command就像是给节点赋予了"决策权"——它不仅能够处理数据、更新状态，还能直接决定下一步要执行哪个节点，而不需要依赖外部的边或路由函数。

Command的典型使用场景 包括：

动态控制流 ：根据业务逻辑动态决定下一步流程，比如订单审核系统根据订单金额选择不同的审核流程

工具中的状态更新 ：工具执行后直接更新Graph状态，比如客服系统查询用户信息后更新状态

子图中的导航 ：子图完成后跳转到主流程的某个节点，实现跨层级的流程控制

多智能体交接 ：智能体之间传递控制权和信息，比如客服转交给专家并传递上下文

好了，以上就是本期关于Command的全部内容，如果大家觉得对自己有帮助的，还请多多点赞、收藏、转发、关注！祝大家玩的开心，我们下期再见！

—— END——

往期精华：

1.COZE教程

零基础搞定！萌新的 Coze 开源版保姆级本地部署指南

AI工作流编排手把手指南之一：Coze智能体的创建与基本设置

AI工作流编排手把手指南之二：Coze智能体的插件添加与调用

AI工作流编排手把手指南之三：Coze智能体的工作流

Agent | 工作流编排指南4：萌新友好的Coze选择器节点原理及配置教程

Agent | 工作流编排指南5：长文扩写自由 — Coze循环节点用法详解

Coze工作流编排指南6：聊天陪伴类智能体基本工作流详解-快来和玛奇玛小姐姐谈心吧~

PPT自由！Coze工作流 X iSlide插件-小白也能看懂的节点参数配置原理详解

2.MCP探索

Excel-MCP应用 | 自动提取图片数据到Excel的极简工作流手把手教程

markitdown-mcp联动Obsidian-mcp | 一个极简知识管理工作流

【15合1神器】不会代码也能做高级图表！这个MCP工具让我工作效率翻了不止三倍！

【效率翻倍】Obsidian自动待办清单实现：MCP联动Prompt保姆级教程（萌新3分钟上手）

萌新靠MCP实现RPA、爬虫自由？playwright-mcp实操案例分享！

高德、彩云MCP全体验：让Cherry Studio化身私人小助理的喂饭版指南！

3.Prompt设计

干货分享 | Prompt设计心法 - 如何3步做到清晰表达需求？

打工人看了流泪的Prompt设计原理，如何用老板思维让AI一次听懂需求？

不会Prompt还敢说自己会用DeepSeek？别怕！10分钟让你成为提示大神！