LangGraph 核心之边（Edge）& 命令（Command）

LangGraph 的核心在于其简洁而强大的图计算模型，这一模型的基石由四个核心原语构成：状态（State）、节点（Node）、边（Edge），以及命令（Command）。
理解这四个原语的概念及其相互作用方式，是掌握 LangGraph 并构建复杂智能体系统的重中之重。可以将这四个原语比作乐高积木最基本的、也是最核心的模块，理解了它们，就如同掌握了乐高搭建的"语言"，后续才能使用更高级的技巧，搭建出各种各样精巧、复杂、功能强大的智能体系统。

本章将深入探讨 LangGraph 核心原语之边（Edge）&命令（Command）

1. 边（Edge）

边在 LangGraph 中负责连接不同的节点，定义智能体系统的执行流程。边决定了在执行完一个节点之后，下一步应该执行哪个节点。

LangGraph 主要支持两种类型的边：

• •普通边：定义节点间固定的、无条件的连接关系
• •条件边：提供基于状态动态路由的能力

示例 1-1：意图识别与技能路由流程

from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict

# 定义状态
class IntentState(TypedDict):
    user_input: str
    user_intent: str
    response: str

# 意图识别节点
def intent_recognition_node(state):
    user_input = state['user_input'].lower()
    
    # 简单的意图识别逻辑
    if '天气' in user_input or 'weather' in user_input:
        intent = "查询天气"
    elif '机票' in user_input or 'flight' in user_input:
        intent = "预订机票"
    elif '投诉' in user_input or 'complaint' in user_input:
        intent = "投诉建议"
    else:
        intent = "未知"
    
    return {"user_intent": intent}

# 技能节点
def weather_query_node(state):
    return {"response": "今天天气晴朗，温度 22-28°C"}

def flight_booking_node(state):
    return {"response": "为您查找合适的机票选项..."}

def complaint_suggestion_node(state):
    return {"response": "感谢您的反馈，我们会认真处理您的建议"}

# 条件路由函数
def route_to_skill(state):
    """条件函数，根据用户意图路由到不同的技能节点"""
    user_intent = state['user_intent']
    if user_intent == "查询天气":
        return "weather_query_node"  # 跳转到查询天气节点
    elif user_intent == "预订机票":
        return "flight_booking_node"  # 跳转到预订机票节点
    elif user_intent == "投诉建议":
        return "complaint_suggestion_node"  # 跳转到投诉建议处理节点
    else:
        return END  # 无法识别意图，结束流程

# 构建图
builder = StateGraph(IntentState)

# 添加节点
builder.add_node("intent_recognition_node", intent_recognition_node)
builder.add_node("weather_query_node", weather_query_node)
builder.add_node("flight_booking_node", flight_booking_node)
builder.add_node("complaint_suggestion_node", complaint_suggestion_node)

# 添加边
builder.add_edge(START, "intent_recognition_node")
builder.add_conditional_edges(
    "intent_recognition_node", 
    route_to_skill,
    ["weather_query_node", "flight_booking_node", "complaint_suggestion_node", END]
)
builder.add_edge("weather_query_node", END)
builder.add_edge("flight_booking_node", END)
builder.add_edge("complaint_suggestion_node", END)

# 编译图
intent_graph = builder.compile()

print("意图识别与技能路由图构建完成")

# 测试不同的用户输入
test_inputs = [
    "今天天气怎么样？",
    "我要预订明天到北京的机票",
    "我要投诉你们的服务",
    "你好"
]

for user_input in test_inputs:
    print(f"\n用户输入: {user_input}")
    result = intent_graph.invoke({"user_input": user_input})
    print(f"识别意图: {result.get('user_intent', '未识别')}")
    print(f"系统回复: {result.get('response', '无回复')}")

意图识别与技能路由图构建完成

用户输入: 今天天气怎么样？
识别意图: 查询天气
系统回复: 今天天气晴朗，温度 22-28°C

用户输入: 我要预订明天到北京的机票
识别意图: 预订机票
系统回复: 为您查找合适的机票选项...

用户输入: 我要投诉你们的服务
识别意图: 投诉建议
系统回复: 感谢您的反馈，我们会认真处理您的建议

用户输入: 你好
识别意图: 未知
系统回复: 无回复

💡 边的类型和应用场景：

• •条件边：add_conditional_edges()实现动态路由，根据运行时状态选择路径
• •普通边：add_edge()创建固定连接，适用于确定的流程序列
• •应用场景：意图识别、工具选择、错误处理、流程分支等

2. 命令（Command）

命令（Command）是 LangGraph 新推出的强大工具，它允许我们将状态更新和流程控制逻辑整合到同一个节点中。Command 打破了节点和边功能上的传统分工，赋予了节点更强大的流程控制能力。

一个Command对象主要包含以下两个部分：

• •update（状态更新）：指定需要更新的状态键值对
• •goto（流程跳转）：指定下一步要执行的节点名称

示例 2-1：使用 Command 的节点函数

from langgraph.types import Command
from typing import Literal
from typing_extensions import TypedDict

# 定义状态
class CommandState(TypedDict):
    input_data: str
    processed_data: str
    decision_result: str

def decide_next_node(state):
    """根据状态决定下一个节点"""
    input_data = state['input_data'].lower()
    if 'process_a' in input_data:
        return 'node_a'
    elif 'process_b' in input_data:
        return 'node_b'
    else:
        return 'node_c'

def decision_node(state) -> Command[Literal["node_a", "node_b", "node_c"]]:
    """
    使用 Command 的节点函数示例
    """
    # 处理输入数据
    input_data = state['input_data']
    processed_result = f"已处理: {input_data}"
    
    # 决定下一个节点
    next_node_name = decide_next_node(state)
    
    return Command(
        update={"processed_data": processed_result, "decision_result": f"决定跳转到: {next_node_name}"},  # 状态更新
        goto=next_node_name  # 流程跳转指令
    )

# 目标节点
def node_a(state):
    return {"processed_data": state['processed_data'] + " -> 经过节点A处理"}

def node_b(state):
    return {"processed_data": state['processed_data'] + " -> 经过节点B处理"}

def node_c(state):
    return {"processed_data": state['processed_data'] + " -> 经过节点C处理"}

# 构建使用 Command 的图
builder = StateGraph(CommandState)

# 添加节点
builder.add_node("decision_node", decision_node)
builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)
builder.add_node("node_c", node_c)

# 添加边
builder.add_edge(START, "decision_node")
builder.add_edge("node_a", END)
builder.add_edge("node_b", END)
builder.add_edge("node_c", END)

# 编译图
command_graph = builder.compile()

print("Command 示例图构建完成")

# 测试不同的输入
test_inputs = [
    "请执行 process_a 操作",
    "需要 process_b 处理",
    "其他类型的操作"
]

for input_data in test_inputs:
    print(f"\n输入数据: {input_data}")
    result = command_graph.invoke({"input_data": input_data})
    print(f"决策结果: {result.get('decision_result', '无')}")
    print(f"处理结果: {result.get('processed_data', '无')}")

Command 示例图构建完成

输入数据: 请执行 process_a 操作
决策结果: 决定跳转到: node_a
处理结果: 已处理: 请执行 process_a 操作 -> 经过节点A处理

输入数据: 需要 process_b 处理
决策结果: 决定跳转到: node_b
处理结果: 已处理: 需要 process_b 处理 -> 经过节点B处理

输入数据: 其他类型的操作
决策结果: 决定跳转到: node_c
处理结果: 已处理: 其他类型的操作 -> 经过节点C处理

💡 Command 的核心价值：

• •内聚性：将状态更新和流程控制逻辑封装在同一个节点中
• •类型安全：通过Literal类型提示确保跳转目标的正确性
• •可视化支持：LangGraph 可以静态分析 Command 的类型提示生成准确的流程图
• •灵活性：特别适合多智能体协作中的"交接"场景

总结

在LangGraph的拓扑结构中，条件边（Conditional Edges）是一个强大的特性，它允许我们根据状态的值动态地选择执行路径。这使得我们可以构建具有决策能力的工作流，根据不同的输入或中间状态执行不同的处理逻辑。Command是一个极具实用性的功能，它能够将控制流（边）和状态更新（节点）巧妙地结合起来。这意味着开发者可以在同一个节点中，既执行状态更新操作，又决定下一个要前往的节点，为工作流的构建带来了极大的灵活性。

参考文献和资料

• • LangGraph文档：https://docs.langchain.com/oss/python/langgraph/overview
• • 《LangGraph实战》

微信公众号（关注获取最新资讯）：