通俗易懂的LangGraph图定义解析

LangGraph 是一个基于状态的工作流框架，它通过 节点（Nodes） 和 边（Edges） 的组合，构建出复杂的工作流逻辑。这种设计特别适合处理需要动态决策、循环、多步骤交互的场景（比如对话系统、智能代理等）。下面我们通过一个简单的例子，逐步讲解 LangGraph 中的图（Graph）是如何定义的。

1. 什么是 StateGraph？

StateGraph 是 LangGraph 的核心类，用于定义和管理基于状态的图结构。它的核心思想是：

• 节点：表示执行的操作（如调用模型、处理数据）。
• 边：定义节点之间的执行顺序（支持无条件跳转、条件跳转、循环）。
• 状态（State）：贯穿整个图的数据结构，存储和传递上下文信息。

举个例子：
想象你正在开发一个聊天机器人，用户输入一条消息后，机器人需要依次完成以下步骤：

1. 解析用户意图（节点A）。
2. 查询数据库（节点B）。
3. 生成回复（节点C）。

通过 StateGraph，你可以将这三个步骤定义为节点，并通过边连接它们，同时用状态对象传递上下文（比如用户的消息、查询结果等）。

2. 如何定义一个图？

定义一个图的核心步骤包括：创建状态、添加节点、设置边，并最终编译图。以下是具体操作：

(1) 定义状态

状态是图中传递的数据结构，通常是一个字典（TypedDict 或自定义类）。例如：

from typing_extensions import TypedDict
class State(TypedDict):
    messages: list  # 存储对话历史
    query: str      # 用户当前问题

(2) 创建图实例

使用 StateGraph 类初始化一个图：

from langgraph.graph import StateGraph
graph_builder = StateGraph(State)

(3) 添加节点

节点是执行具体任务的函数。例如，添加一个解析用户意图的节点：

def parse_intent(state: State) -> dict:
    # 假设解析出用户意图是"查询天气"
    return {"query": "查询天气"}
graph_builder.add_node("parse_intent", parse_intent)

(4) 添加边

边定义了节点的执行顺序。例如，将解析意图节点连接到查询数据库节点：

graph_builder.add_edge("parse_intent", "query_database")

(5) 设置入口和出口

指定图的起点和终点：

graph_builder.set_entry_point("parse_intent")  # 起点
graph_builder.set_finish_point("generate_response")  # 终点

(6) 编译图

将图编译为可执行对象：

compiled_graph = graph_builder.compile()

3. 核心方法详解

LangGraph 提供了多个方法来构建图，以下是关键方法的简明说明：

条件边的使用示例

假设需要根据用户意图选择不同的处理路径：

def decide_next_node(state: State) -> str:
    if state["query"] == "查询天气":
        return "query_weather"
    else:
        return "handle_other"

graph_builder.add_conditional_edges("parse_intent", decide_next_node)

4. 实际应用示例

场景：用户询问订单状态，机器人需要验证身份后查询数据库。

代码实现：

from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    user_id: str
    order_status: str

def verify_user(state: State) -> dict:
    # 模拟验证用户身份
    return {"user_id": "12345"}

def query_order(state: State) -> dict:
    # 模拟查询订单状态
    return {"order_status": "已发货"}

builder = StateGraph(State)
builder.add_node("verify_user", verify_user)
builder.add_node("query_order", query_order)
builder.add_edge(START, "verify_user")
builder.add_edge("verify_user", "query_order")
builder.add_edge("query_order", END)

compiled = builder.compile()
result = compiled.invoke({"user_id": "", "order_status": ""})
print(result)  # 输出: {'user_id': '12345', 'order_status': '已发货'}

5. 状态管理的关键技巧

• 状态更新：每个节点接收当前状态，并返回更新后的状态。例如，节点A修改 state["x"]，节点B可以直接读取新值。
• 状态持久化：通过 checkpointer 参数保存状态，支持断点续传（例如用户中断对话后恢复上下文）。
• 条件逻辑：利用条件边（add_conditional_edges）动态决定下一步操作，比如根据用户输入选择不同分支。

6. 常见问题与注意事项

• 节点命名冲突：确保每个节点名称唯一。
• 状态字段设计：状态字段应包含所有必要的上下文，避免遗漏关键信息。
• 调试技巧：启用 debug=True 编译图，观察节点执行过程和状态变化。

总结

LangGraph 的 StateGraph 通过节点和边的灵活组合，让开发者能够轻松构建复杂的工作流。无论是简单的线性流程，还是带条件分支、循环的复杂逻辑，StateGraph 都能提供清晰的结构和强大的状态管理能力。通过合理设计状态和节点，你可以快速实现对话系统、智能代理等应用。