LangGraph 是一个基于状态的工作流框架，它通过 节点（Nodes） 和 边（Edges） 的组合，构建出复杂的工作流逻辑。这种设计特别适合处理需要动态决策、循环、多步骤交互的场景（比如对话系统、智能代理等）。下面我们通过一个简单的例子，逐步讲解 LangGraph 中的图（Graph）是如何定义的。
1. 什么是 StateGraph？ 
StateGraph 是 LangGraph 的核心类，用于定义和管理基于状态的图结构。它的核心思想是：
- 节点
：表示执行的操作（如调用模型、处理数据）。 - 边
：定义节点之间的执行顺序（支持无条件跳转、条件跳转、循环）。 - 状态（State）
：贯穿整个图的数据结构，存储和传递上下文信息。
举个例子：
想象你正在开发一个聊天机器人，用户输入一条消息后，机器人需要依次完成以下步骤：
解析用户意图（节点A）。 查询数据库（节点B）。 生成回复（节点C）。
通过 StateGraph，你可以将这三个步骤定义为节点，并通过边连接它们，同时用状态对象传递上下文（比如用户的消息、查询结果等）。
2. 如何定义一个图？
定义一个图的核心步骤包括：创建状态、添加节点、设置边，并最终编译图。以下是具体操作：
状态是图中传递的数据结构，通常是一个字典（TypedDict 或自定义类）。例如：
from typing_extensions import TypedDict
class State(TypedDict):
messages: list # 存储对话历史
query: str # 用户当前问题
使用 StateGraph 类初始化一个图：
from langgraph.graph import StateGraph
graph_builder = StateGraph(State)
节点是执行具体任务的函数。例如，添加一个解析用户意图的节点：
def parse_intent(state: State) -> dict:
# 假设解析出用户意图是"查询天气"
return {"query": "查询天气"}
graph_builder.add_node("parse_intent", parse_intent)
边定义了节点的执行顺序。例如，将解析意图节点连接到查询数据库节点：
graph_builder.add_edge("parse_intent", "query_database")
指定图的起点和终点：
graph_builder.set_entry_point("parse_intent") # 起点
graph_builder.set_finish_point("generate_response") # 终点
将图编译为可执行对象：
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, 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 都能提供清晰的结构和强大的状态管理能力。通过合理设计状态和节点，你可以快速实现对话系统、智能代理等应用。
