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

python-LangGraph框架(3-29-LangGraph 「覆盖式状态」的原理与实践)

一、什么是覆盖式状态?

覆盖式状态是 LangGraph 默认提供的浅合并状态模式,也是最简单、最直观的状态管理方式。

它的核心逻辑可以用一句话概括:节点只需要返回需要更新的字段,未返回的字段会自动保留原有值;返回的字段会直接覆盖状态中的旧值。

简单理解:状态就像一个全局的「数据账本」,每个执行节点只需要修改自己关心的那几项数据,不用关心其他数据,账本会自动帮你保留所有历史数据,只更新你指定的部分。

二、覆盖式状态的核心特点

1. 自动保留未修改字段

这是覆盖式状态最大的优势。在工作流的每个节点中,你不需要返回所有状态字段,只需要返回需要更新的字段即可。比如状态中包含用户查询、结果、步骤三个字段,某个节点只需要更新结果,那么用户查询和步骤会自动保留,无需手动传递。

2. 极简的更新逻辑

更新方式为直接覆盖,返回什么新值,状态中对应的字段就变成什么值,没有复杂的合并规则,逻辑清晰易懂。

3. 无侵入式开发

节点函数只关注自身的业务逻辑,不需要感知全局状态的完整结构,代码更简洁、维护成本更低。

4. 轻量无依赖

开箱即用,无需额外配置状态合并规则,LangGraph 底层自动处理,适合快速开发。

三、覆盖式状态的适用场景

这种模式虽然简单,但足以应对绝大多数基础工作流,典型使用场景:

  • 单路径线性执行的工作流(无分支、无循环)
  • 简单的多步骤任务处理
  • 数据逐步加工、结果逐层更新
  • 轻量级智能体、问答流程、数据解析流程

只要你的工作流不需要复杂的状态合并(如列表追加、深度合并),覆盖式状态都是最优选择。

四、覆盖式状态的实现核心逻辑

基于 LangGraph 实现覆盖式状态,只需要抓住四个核心步骤,全程无需自定义合并规则:

1. 定义结构化状态

使用 TypedDict 定义固定结构的状态,明确所有需要存储的字段(如用户输入、中间结果、执行步骤等),作为全流程的数据规范。

2. 编写流程节点

每个节点函数接收当前完整状态作为参数,只返回需要更新的字段字典。未返回的字段,LangGraph 会自动保留;返回的字段,自动覆盖旧值。

3. 构建状态图

使用 StateGraph 绑定定义好的状态结构,添加所有节点,设置节点之间的执行边,确定线性执行路径。

4. 执行工作流

传入初始状态,启动图执行,全程状态自动流转、自动合并,最终拿到完整的最终状态。

整个实现过程,完全依托 LangGraph 的默认机制,代码简洁、可读性极高,新手也能快速上手。

五、覆盖式状态的核心优势总结

开发极简:不用手动传递所有字段,减少冗余代码✅逻辑清晰:直接覆盖更新,无隐藏逻辑,易于调试✅自动保留:未修改数据全程留存,不丢失上下文✅默认支持:开箱即用,无需额外配置✅场景通用:满足大部分基础工作流需求

六、写在最后

覆盖式状态是 LangGraph 状态管理的入门基石,也是构建线性工作流的首选方案。它用最简单的「自动保留 + 字段覆盖」机制,解决了工作流中数据传递和存储的核心问题。

当你的工作流是单路径、简单步骤、只需要基础数据更新时,优先使用覆盖式状态;后续遇到复杂场景(如列表追加、深度状态合并),再进阶学习自定义合并状态即可。

掌握这一模式,你就已经能快速搭建出稳定、简洁的 LangGraph 工作流啦!

代码实现:

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

"""

📌 模式1: 简单覆盖式状态

核心机制: 默认浅合并(只更新返回字段,其他字段自动保留)

适用场景: 简单工作流、单路径执行

"""

importasyncio

fromtypingimportTypedDict

fromlanggraph.graphimportStateGraph, START, END

# ===== 1. 状态定义 =====

classSimpleState(TypedDict):

"""

简单状态结构:

- query: 用户原始查询(全程保留)

- result: 中间/最终结果(逐步更新)

- step: 当前执行步骤(递增)

"""

query:str

result:str

step:int

# ===== 2. 节点函数 =====

defstep_1_node(state: SimpleState)->dict:

"""

第一步: 更新 step 和部分 result

⚠️ 注意: 未返回 'query' 字段,但会被自动保留!

"""

print(f"[Step1] 输入状态: {state}")

return{

"step": state["step"]+1,# 更新 step

"result":"初步分析完成"# 更新 result

# query 字段未返回 → 自动保留原始值

}

defstep_2_node(state: SimpleState)->dict:

"""

第二步: 只更新 result,step 保持不变

"""

print(f"[Step2] 输入状态: {state}")

return{

"result": f"{state['result']} → 最终结论生成"# 仅更新 result

# step 和 query 自动保留

}

# ===== 3. 构建图 =====

defbuild_simple_graph():

builder=StateGraph(SimpleState)

builder.add_node("step_1", step_1_node)

builder.add_node("step_2", step_2_node)

builder.add_edge(START,"step_1")

builder.add_edge("step_1","step_2")

builder.add_edge("step_2", END)

returnbuilder.compile()

# ===== 4. 执行演示 =====

asyncdefmain():

print("="*60)

print("🧠 模式1: 简单覆盖式状态(默认浅合并)")

print("="*60)

graph=build_simple_graph()

# 画图

print(graph.get_graph().draw_ascii())

# 初始状态(包含所有字段)

initial_state={

"query":"北京天气如何?",

"result":"初始状态",

"step":0

}

print("\n【初始状态】")

print(f" {initial_state}")

# 执行图

final_state=await graph.ainvoke(initial_state)

print("\n【最终状态】")

print(f" {final_state}")

if__name__=="__main__":

asyncio.run(main())

结果输出:

============================================================
🧠 模式1: 简单覆盖式状态(默认浅合并)
============================================================
+-----------+
| __start__ |
+-----------+
*
*
*
+--------+
| step_1 |
+--------+
*
*
*
+--------+
| step_2 |
+--------+
*
*
*
+---------+
| __end__ |
+---------+

【初始状态】
{'query': '北京天气如何?', 'result': '初始状态', 'step': 0}
[Step1] 输入状态: {'query': '北京天气如何?', 'result': '初始状态', 'step': 0}
[Step2] 输入状态: {'query': '北京天气如何?', 'result': '初步分析完成', 'step': 1}

http://www.jsqmd.com/news/1139128/

相关文章:

  • HappyHorse AI 视频提示词完全指南:4步引导 + 五要素公式 + 常见坑解法
  • 【Java课程设计/毕业设计】基于 SpringBoot 的家政服务分类管理系统的设计与实现 基于前后端分离的智能化便民家政运维系统【附源码、数据库、万字文档】
  • 代码复杂度的代价远比你想象的大!
  • 字典列表导出为 Excel
  • FreeLLMAPI安装(WSL版)
  • 【HackMyVM】latestWasALie
  • 如何一键获取国家中小学智慧教育平台电子课本?这个免费工具帮你解决备课难题
  • 为什么“双触发器”不够用?
  • 国际eSIM物联卡如何远程写号切换运营商?深度适配多国物流场景解析
  • 常德本地环保装修哪家技术强
  • 2026最新实操:小白也能用游戏护航小程序,快速搭建校园电竞服务SaaS系统
  • 终极GTA5安全防护菜单:YimMenu完全使用指南
  • 一站式可观测新选择:OpenObserve 深度对标 ELK / Grafana+Prometheus / Netdata,附Docker一键部署实战
  • Python 获取 A 股 K 线、实时行情和五档盘口:QuantDash API 示例
  • ubuntu使用很久之后网络突然连不上解决方法
  • WebAssembly AI 插件热加载:运行时替换模块而不丢会话状态
  • 有什么针对新闻媒体行业的安全解决方案
  • 蔡司三坐标能否检测齿轮?完整检测方案详解
  • Copilot上下文感知提醒:基于Graph API的工作流驱动机制
  • 使用vivado开发FPGA完整流程
  • Rust 智能指针选型:Box、Rc、Arc 什么时候该用哪一个
  • OpenClaw智能体工作流中枢:轻量级AI技能编排引擎实战指南
  • 潮汛网快讯:英国GEO公司geoSurge完成1200万美元种子轮融资
  • Lifecycle中出现的监听器
  • ClaudeAgent】并发-后台任务
  • 科辉荣盛小程序定制开发,多重保障安心数字化
  • 佳木斯装修交叉施工多?轻质抹灰石膏减少墙面破损返工
  • 双轴追踪电站 vs 固定式光伏——从三项强制国标看电站支架选型的技术参数逻辑
  • 每日安全情报报告 · 2026-07-06
  • Fable 5降本新招:pxpipe将上下文转图片,token输入成本最多省70%!