当前位置：首页 > news >正文

通俗易懂的LangGraph图定义解析

news 2025/7/3 13:11:45

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

1. 什么是 StateGraph？

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

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

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

解析用户意图（节点A）。
查询数据库（节点B）。
生成回复（节点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 提供了多个方法来构建图，以下是关键方法的简明说明：

方法名	作用
`add_node()`	添加一个节点到图中。可以指定节点名称和执行函数。
`add_edge()`	添加一条有向边，定义两个节点的执行顺序。如果多个起点，所有起点完成后才会执行终点。
`add_conditional_edges()`	添加条件边，根据状态动态选择下一个节点。例如：根据用户意图跳转到不同的处理节点。
`compile()`	将图编译为可执行对象，支持调用、流式处理等功能。

条件边的使用示例

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

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, ENDclass State(TypedDict):user_id: strorder_status: strdef 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': '已发货'}