LangGraph 状态图：从入门到实践

一、代码概述与核心思想

这段代码展示了如何使用 LangGraph 构建一个有状态的工作流（stateful workflow），通过节点（Node）和边（Edge）的组合创建可执行的流程图。核心特点是状态传递和条件分支。

关键概念：

  • 状态（State）：在整个图执行过程中传递的数据容器
  • 节点（Node）：执行具体操作的函数单元
  • 边（Edge）：定义节点之间的流转规则
  • 条件边（Conditional Edge）：根据当前状态动态选择下一个节点

二、代码逐层解析

1. 状态定义（State Definition）

1
2
3
4
from typing_extensions import TypedDict

class State(TypedDict):
    graph_state: str  # 图状态字段，类型为字符串
  • 使用 TypedDict 定义强类型的状态结构
  • graph_state 是状态中唯一的字段，存储字符串类型数据
  • 这确保了整个工作流中状态的一致性和类型安全

2. 节点函数（Node Functions）

节点1：基础处理节点

1
2
3
def node_1(state: State) -> dict:
    print("---Node 1---")
    return {"graph_state": state['graph_state'] + " I am"}
  • 接收当前状态，添加 “ I am” 后返回更新状态
  • 所有节点函数都必须返回与 State 结构兼容的字典

节点2和节点3：分支处理节点

1
2
3
4
5
def node_2(state: State) -> dict:
    return {"graph_state": state['graph_state'] + " happy!"}

def node_3(state: State) -> dict:
    return {"graph_state": state['graph_state'] + " sad!"}
  • 分别处理”开心”和”悲伤”两种路径
  • 结构相同但输出不同，展示分支的可能性

3. 条件路由函数（Conditional Routing）

1
2
3
4
def decide_mood(state: State) -> Literal["node_2", "node_3"]:
    if random.random() < 0.5:
        return "node_2"
    return "node_3"
  • 关键函数：决定从 node_1 到哪个节点
  • 返回值为字符串字面量，精确匹配节点名称
  • 使用随机数模拟50/50的概率选择
  • 实际应用中可以是基于状态内容的复杂逻辑判断

4. 图构建流程（Graph Construction）

步骤1：初始化

1
builder = StateGraph(State)
  • 创建图构建器，指定状态类型

步骤2：添加节点

1
2
3
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)
  • 注册三个节点，将函数与名称绑定
  • 节点名称是图中的唯一标识符

步骤3：添加边（定义流转逻辑）

1
2
3
4
5
6
7
8
9
# 固定起点
builder.add_edge(START, "node_1")

# 条件分支
builder.add_conditional_edges("node_1", decide_mood)

# 固定终点
builder.add_edge("node_2", END)
builder.add_edge("node_3", END)
  • 普通边：固定路径，如 START → node_1
  • 条件边：动态路径，根据函数返回值选择下一节点
  • 结束边：将节点连接到 END，终止流程

步骤4：编译与执行

1
2
graph = builder.compile()
result = graph.invoke({"graph_state": "Hi, this is FLY."})
  • compile()：检查图结构并创建可执行对象
  • invoke()：传入初始状态，启动图执行

三、执行流程可视化

流程图

两种可能的执行路径：

路径A（开心路径）：

1
2
3
4
初始状态: "Hi, this is FLY."
node_1后: "Hi, this is FLY. I am"
node_2后: "Hi, this is FLY. I am happy!"
最终结果: "Hi, this is FLY. I am happy!"

路径B（悲伤路径）：

1
2
3
4
初始状态: "Hi, this is FLY."
node_1后: "Hi, this is FLY. I am"
node_3后: "Hi, this is FLY. I am sad!"
最终结果: "Hi, this is FLY. I am sad!"

四、学习要点与记忆技巧

1. 四大核心组件（记忆口诀：NESC）

  • Nodes（节点）：执行具体任务
  • Edges（边）：定义流转路径
  • State（状态）：传递数据
  • Conditionals（条件）：实现分支逻辑

2. 构建流程四步法：

  1. 定义状态：明确数据结构和类型
  2. 创建节点：编写处理函数
  3. 连接节点：用边定义执行顺序
  4. 编译执行：检查并运行图

3. 条件边的关键特性：

  • 连接函数必须返回下一个节点的名称
  • 返回值类型要精确匹配（使用 Literal）
  • 条件决策可以基于：状态内容、外部输入、随机数等

4. 调试技巧：

  • 在每个节点中添加 print 语句追踪执行
  • 使用 print("---Node X---") 分隔输出
  • 检查最终状态是否与预期一致

五、实际应用扩展

这个简单示例展示了 LangGraph 的核心模式，在实际应用中可以进行以下扩展：

1. 复杂状态管理

1
2
3
4
5
class ComplexState(TypedDict):
    messages: List[str]      # 对话历史
    user_info: Dict[str, Any] # 用户信息
    context: str             # 上下文
    metadata: Dict           # 元数据

2. 多条件分支

1
2
3
4
5
6
7
8
9
def complex_router(state: State) -> Literal["node_a", "node_b", "node_c", "node_d"]:
    if condition1:
        return "node_a"
    elif condition2:
        return "node_b"
    elif condition3:
        return "node_c"
    else:
        return "node_d"

3. 循环与迭代

1
2
3
# 添加循环边，实现迭代处理
builder.add_edge("process_node", "check_node")
builder.add_conditional_edges("check_node", should_continue)

六、常见问题与解决方案

Q1: 状态更新不生效？

  • 确保节点函数返回完整的更新字典
  • 检查字段名称是否与 State 定义一致

Q2: 条件边不工作？

  • 确认路由函数返回值与节点名称完全一致（大小写敏感）
  • 检查返回值类型是否正确使用 Literal

Q3: 图编译失败？

  • 确保所有引用的节点都已添加
  • 检查没有孤立节点（未被任何边连接）

七、总结与记忆练习

核心要点回顾：

  1. LangGraph 通过状态传递连接各个节点
  2. 条件边实现了动态工作流分支
  3. 图构建遵循：定义→添加→连接→编译的标准流程
  4. 类型提示（TypedDict, Literal）提高了代码可靠性

动手练习建议：

  1. 修改 decide_mood 函数，基于输入内容（而非随机）做决策
  2. 添加第四个节点，创建更复杂的分支结构
  3. 扩展 State 类，添加更多字段并在节点间传递
  4. 实现一个循环流程，直到满足特定条件才结束

通过这个简单的”心情选择器”示例，我们掌握了 LangGraph 的核心概念。记住：状态是血液，节点是器官，边是血管，条件分支是神经——共同构成了智能工作流的生命体。