LangGraph 性能优化:减少图遍历过程中的状态复制开销

1. 引入与连接:从“写代码快了,跑起来卡哭了”的痛场景说起

1.1 一个真实的开发者凌晨3点的吐槽(开场故事)

深夜11:45,字节跳动飞书智能助理「豆包助手Lite」的技术负责人小李敲下了最后一行LangGraph代码——这是他花了3天重构的AI文档分类流水线,把原来的单Agent多prompt模式拆成了「文本预处理Agent」、「多标签初步筛选Agent」、「语义相似度校验Agent」、「标签冲突消解Agent」的四节点状态图,逻辑清晰得让他忍不住拍了拍显示器的边框,还给团队群甩了一张截图:

“重构完成!可维护性×100!单文档测试正确率×15!明天可以上线给运营同学爽爽了!”

凌晨2:17,小李的手机突然疯狂震动——运营同学反馈的第一批测试报告跳了出来,不是正确率,是延迟

“豆包助手分类1000篇500字的会议纪要,原来单Agent只要28分钟,现在居然要2小时47分?!而且后台显示GPU利用率才12%,CPU内存却占满了!这怎么回事啊小李哥?明天就要给CEO汇报新版效率提升的!”

小李瞬间睡意全无,打开本地的测试环境复现——同样的10篇会议纪要,单Agent 32秒,LangGraph四节点 297秒!差了9倍多!他赶紧打开LangChain的Debug日志,翻到最后几页,发现了一行又一行刺眼的红色警告:

“[DEBUG] LangGraph StateManager: Deep copying state dictionary of size 128KB (including 32KB of numpy embeddings array) for transition from ‘preprocessor’ to ‘preliminary_filter’…”

Deep copying?小李猛地拍了一下自己的脑门:重构的时候只关注了逻辑拆分,完全忽略了LangGraph的默认状态管理机制——每次节点间状态传递,都会对整个状态字典做深拷贝!而四节点的图,每次处理一篇文档就要做3次完整深拷贝,1000篇就是3000次,每次128KB就是384MB的临时内存写入+读取,更不用说那些大文档带的几百KB甚至几MB的numpy/pandas/torch张量了!

1.2 你可能也踩过的LangGraph性能坑(与读者已有知识建立连接)

如果你用过LangGraph写过稍微复杂一点的多Agent系统、RAG增强流水线或者复杂决策链,是不是也遇到过类似的场景?

  • ✅ 系统逻辑从“一团糟的条件判断”变成了“清晰的有向无环/有环图”,可维护性、可扩展性提升肉眼可见
  • ❌ 但系统延迟飙升了3-20倍,GPU利用率却一直上不去(大部分时间在等CPU处理内存)
  • ❌ 处理大文档(比如带100个搜索结果、200维numpy embeddings的RAG状态)时,内存占用直接爆炸,服务器频繁OOM重启
  • ❌ 找不到性能瓶颈在哪——只看到整体慢,但不知道是LLM调用慢、工具调用慢、还是状态管理慢

这篇文章,就是为了解决你这些痛点而写的——我们会从LangGraph状态管理的底层原理出发,用“生活化比喻+数学模型+Python代码+实际项目案例”的方式,带你一步步理解“为什么深拷贝会这么慢”、“有哪些替代方案”、“如何根据你的场景选择最优方案”,最后还会给你一套开箱即用的性能优化模板,让你的LangGraph系统逻辑清晰、可维护性强的同时,延迟降低70%-95%,内存占用降低80%-99%

1.3 学习价值与应用场景预览(你能学到什么?)

1.3.1 学习价值(不管你是LangChain新手还是资深用户,都能有所收获)
  • 新手小白:彻底搞懂LangGraph的状态管理机制,从一开始就避免踩“状态深拷贝”的坑
  • 资深用户:掌握5种以上的状态优化高级技巧,构建高性能、可扩展的LangGraph系统
  • 架构师:理解不同优化方案的边界条件和适用范围,设计出符合业务场景的最优状态架构
  • 算法工程师:学会如何把LangGraph状态管理和深度学习张量优化结合起来,最大化GPU利用率
1.3.2 应用场景(这些场景优化后效果最明显!)

我们整理了10个最常见的LangGraph使用场景,其中带「⭐️⭐️⭐️」的是**状态复制开销占比极高(>60%)**的场景,优化后延迟和内存都会有质的提升:

  1. ⭐️⭐️⭐️ 带大向量库RAG结果的智能问答系统
  2. ⭐️⭐️⭐️ 多Agent协作的长文档分析系统(比如合同审查、论文摘要)
  3. ⭐️⭐️⭐️ 批量文本处理流水线(比如分类、标注、翻译)
  4. ⭐️⭐️ 带动态工具调用列表的智能助手
  5. ⭐️⭐️ 有复杂状态历史的决策链/对话链
  6. ⭐️ 带小型图表/JSON数据的数据分析Agent
  7. ⭐️ 单Agent多步骤的简单任务链
  8. 不带复杂状态的单节点工具调用链
  9. 完全无状态的纯LLM调用链
  10. 状态非常小(<1KB)的所有场景

1.4 学习路径概览(我们的知识金字塔)

根据系统提示的「金字塔式知识结构」,我们把这篇文章的学习路径分为4个核心层级:

  1. 基础层(第2-3章):直观理解LangGraph的状态管理机制、深拷贝和浅拷贝的区别,搞懂“为什么状态复制会这么慢”
  2. 连接层(第4章):梳理5种状态优化方案的核心概念、适用范围、边界条件,建立优化方案的对比框架
  3. 深度层(第5-8章):逐一拆解每种优化方案的底层原理、数学模型、算法流程、Python代码、实际项目案例
  4. 整合层(第9-10章):整合所有优化方案,给你一套「场景→方案选型→代码实现→性能测试→最佳实践」的完整模板,最后展望LangGraph状态管理的未来趋势

2. 概念地图:建立整体认知框架

2.1 核心概念与关键术语(先把名词搞懂!)

在开始深入之前,我们先把这篇文章会用到的核心概念关键术语列出来,用「10岁孩童能理解的生活化定义」+「技术上的严谨定义」结合的方式解释:

核心概念/关键术语 10岁孩童能理解的生活化定义 技术上的严谨定义
LangGraph 就像一套“拼积木说明书”——说明书上有不同的“积木块(节点)”、“积木块之间的拼接规则(边)”、“装积木的盒子(状态)”,你可以按照说明书拼出各种不同的玩具(AI应用) LangChain官方推出的基于状态机的有向图构建框架,专门用于构建复杂的多Agent系统、RAG增强流水线、决策链/对话链——核心组件包括「节点(Node)」、「边(Edge)」、「状态(State)」、「状态管理器(StateManager)」
节点(Node) 说明书上的“积木块”——每个积木块都有自己的功能(比如“装轮子的积木”、“装窗户的积木”),拼玩具的时候,每个积木块只做自己的那部分工作 LangGraph中的执行单元——可以是纯Python函数、工具调用封装、LLM调用封装、其他LangGraph子图的封装,每个节点接收当前状态作为输入,返回「新的状态更新」或「状态更新+边跳转指令」
边(Edge) 说明书上的“拼接规则”——告诉你“拼完装轮子的积木之后,下一步拼什么积木” LangGraph中的状态转移规则——可以是「无条件边(从节点A直接跳到节点B)」、「条件边(根据当前状态的某个值,从节点A跳到节点B/C/D)」、「起始边(从空状态/初始状态跳到第一个节点)」、「结束边(从最后一个节点跳到终止状态)」
状态(State) 说明书上的“装积木的盒子”——你拼玩具的所有零件(轮子、窗户、车身、车头)、拼到哪一步了(“刚装完轮子”、“刚装完车头”)、有没有出错(“轮子少了一个”)都放在这个盒子里 LangGraph中的全局共享数据结构——存储了整个图执行过程中的所有中间结果、执行状态、历史记录等——可以是Python字典、Pydantic模型、Dataclass、自定义类
状态管理器(StateManager) 说明书配的“盒子管理员”——每次拼完一个积木块,他都会把盒子里的东西全部拿出来,整理一遍,再放进一个新的盒子里(默认机制),或者直接在原来的盒子里加新东西/改东西(优化机制) LangGraph中负责管理状态的核心组件——核心功能包括「初始化状态」、「合并状态更新」、「状态持久化」、「状态传递」——默认使用CopyOnWriteStateManager,每次状态传递都会对整个状态做深拷贝
深拷贝(Deep Copy) 给你一个“有玩具的旧盒子”,管理员会买一个一模一样的新盒子,再把旧盒子里的每个玩具、每个零件都单独复制一份,放进新盒子里——新旧盒子里的东西完全独立,改新盒子里的东西不会影响旧盒子 Python中完全复制一个对象及其所有嵌套对象的操作——复制后的新对象与原对象完全独立,修改新对象的任何属性(包括嵌套对象的属性)都不会影响原对象——在Python中可以用copy.deepcopy()函数实现
浅拷贝(Shallow Copy) 给你一个“有玩具的旧盒子”,管理员会买一个一模一样的新盒子,但只把旧盒子里的“玩具的标签”复制一份放进新盒子里——新旧盒子里的标签指向的是同一个玩具,改新盒子里的玩具会影响旧盒子,但改标签本身不会 Python中只复制一个对象本身,不复制其嵌套对象的操作——复制后的新对象与原对象共享嵌套对象,修改新对象的嵌套对象属性会影响原对象——在Python中可以用copy.copy()、字典的dict.copy()、列表的list[:]等方式实现
写时复制(Copy-on-Write, CoW) 给你一个“有玩具的旧盒子”,管理员会先只复制旧盒子的“外壳”和“玩具的标签”(也就是浅拷贝),如果你只是看盒子里的东西,就一直用旧盒子和旧玩具;只有当你要改盒子里的某个玩具时,才会把那个玩具单独复制一份放进新盒子里——既节省了空间,又保证了新旧盒子的独立性(只在修改时独立) 一种内存管理优化技术——核心思想是「多个指针/引用指向同一个内存块,只有当其中一个指针/引用要修改内存块时,才会单独复制一份内存块给它」——广泛应用于操作系统的进程管理、文件系统、数据库、编程语言的字符串/列表等数据结构中
状态分片(State Sharding) 给你一个“装了很多很多玩具的大盒子”,管理员会把大盒子分成几个小盒子,每个小盒子只装一类玩具(比如“轮子小盒子”、“窗户小盒子”、“车身小盒子”)——每次拼积木的时候,只拿需要的小盒子,不需要的小盒子就放在仓库里,不用复制 一种数据分片优化技术——核心思想是「把大的全局状态分成几个小的、独立的状态分片,每个状态分片只存储一类相关的数据,每个节点只访问和修改自己需要的状态分片」——广泛应用于分布式系统、数据库、高性能计算中
状态引用(State Reference) 给你一个“装了大零件(比如整个车身)的大盒子”,管理员会不复制车身,只给新盒子里放一张“车身放在仓库哪个位置”的纸条(引用)——每次要用车身的时候,就根据纸条去仓库里拿,不用每次都复制 一种内存优化技术——核心思想是「对于大的、不可变的(或很少修改的)数据,不直接存储在状态字典里,只存储一个指向该数据的引用(比如文件路径、数据库ID、内存地址、对象ID)」——需要时再根据引用加载数据
不可变状态(Immutable State) 给你一个“装了玩具的旧盒子”,管理员会把旧盒子锁起来,任何人都不能改里面的玩具;如果要加新东西或改东西,必须买一个新盒子,把旧盒子里的东西(或部分东西)复制进去,再加新东西或改东西——但因为盒子是锁起来的,管理员可以放心地让多个新盒子共享旧盒子里的玩具(也就是写时复制) 一种状态管理范式——核心思想是「状态一旦创建,就不能被修改;所有的状态更新都会创建一个新的状态对象」——广泛应用于函数式编程(比如Haskell、Scala)、前端框架(比如React、Redux)、区块链中

2.2 概念间的层次与关系(用思维导图直观理解)

为了让你更直观地理解这些概念之间的层次与关系,我们画了一张简化版的思维导图

LangGraph状态
复制优化

基础层

LangGraph核心组件

Node

Edge

State

StateManager

Python内存操作

Deep Copy

Shallow Copy

连接层

优化方案对比

Shallow Copy

CoW State Manager

State Sharding

State Reference

Immutable State

深度层

各方案的

底层原理

数学模型

算法流程

Python代码

实际案例

整合层

场景→方案选型

完整性能优化模板

最佳实践

未来趋势

2.3 概念之间的关系:核心属性维度对比(清晰明了的表格)

接下来,我们从性能(延迟/内存)、可维护性、可扩展性、安全性、适用场景这5个核心属性维度,对比一下5种常见的状态优化方案(默认方案是深拷贝,作为对照组):

优化方案 延迟优化幅度 内存优化幅度 可维护性 可扩展性 安全性 适用场景(优先级从高到低)
默认方案:深拷贝 0%(对照组) 0%(对照组) ⭐️⭐️⭐️⭐️⭐️(完全安全,完全独立,最容易维护) ⭐️⭐️⭐️⭐️⭐️(完全支持所有场景) ⭐️⭐️⭐️⭐️⭐️(新旧状态完全独立,不会有并发冲突) 状态非常小(<1KB)、状态修改非常频繁且复杂、有严格并发要求、新手小白入门
方案1:浅拷贝 70%-90%(只复制状态外壳,不复制嵌套对象) 80%-99%(嵌套对象完全共享) ⭐️⭐️⭐️(需要注意嵌套对象的并发修改问题,有一定学习成本) ⭐️⭐️⭐️(不支持有环图的并行执行,不支持跨进程/跨节点的状态传递) ⭐️⭐️(新旧状态共享嵌套对象,容易出现“意外修改旧状态”的bug) 无环图的串行执行、嵌套对象不可变、单进程/单节点运行、状态中有大的不可变嵌套对象(比如numpy/pandas/torch张量)
方案2:写时复制(CoW)状态管理器 60%-95%(只读时只做浅拷贝,修改时才复制修改的部分) 70%-99%(只读时完全共享嵌套对象,修改时只复制修改的部分) ⭐️⭐️⭐️⭐️(几乎和深拷贝一样容易维护,不需要注意太多并发问题) ⭐️⭐️⭐️⭐️(支持无环图的并行执行,支持部分跨进程/跨节点的状态传递) ⭐️⭐️⭐️⭐️(只读时共享但不能修改,修改时自动复制,安全性接近深拷贝) 无环图的串行/并行执行、嵌套对象大部分时候只读、偶尔修改、单进程/多进程单节点运行、状态中有大的嵌套对象(不管可不可变)
方案3:状态分片 50%-90%(只复制需要的状态分片,不复制整个状态) 60%-95%(不需要的状态分片放在仓库里,不用复制) ⭐️⭐️⭐️(需要合理设计状态分片的边界,有一定架构设计成本) ⭐️⭐️⭐️⭐️⭐️(完全支持分布式执行,状态分片可以放在不同的机器上) ⭐️⭐️⭐️⭐️(每个状态分片独立管理,并发冲突只发生在单个分片内) 分布式LangGraph系统、状态非常大且可以明确分成几类独立的数据、有跨节点状态传递需求、批量文本处理流水线
方案4:状态引用 80%-98%(只复制引用,不复制大对象本身) 90%-99.9%(大对象放在仓库里,只存储引用) ⭐️⭐️(需要自己管理大对象的加载、存储、生命周期,有较高的架构设计成本) ⭐️⭐️⭐️⭐️⭐️(完全支持分布式执行,大对象可以放在不同的存储系统上) ⭐️⭐️⭐️(需要注意大对象的并发修改问题,需要自己实现引用的有效性校验) 状态中有超大对象(>1MB,比如长文档、大向量库RAG结果、大图片/视频)、批量文本处理流水线、分布式LangGraph系统、大对象很少修改或不可变
方案5:不可变状态 40%-85%(和CoW类似,但更强制不可变,编译器/解释器可以做更多优化) 60%-99%(完全共享不可变对象,不需要复制) ⭐️⭐️⭐️(需要习惯函数式编程的思维,有一定学习成本) ⭐️⭐️⭐️⭐️⭐️(完全支持并行执行,完全支持分布式执行,因为状态不可变,没有并发冲突) ⭐️⭐️⭐️⭐️⭐️(状态不可变,完全安全,没有任何并发冲突) 函数式编程爱好者、有严格并发要求的系统、无环图的串行/并行执行、状态大部分时候只读、偶尔修改

2.4 概念联系的ER实体关系图与交互关系图(更专业的架构图)

为了让你更深入地理解这些概念之间的数据关系执行流程交互关系,我们画了两张更专业的mermaid架构图:

2.4.1 ER实体关系图(数据关系)

这张图展示了LangGraph核心组件、优化方案、Python内存操作之间的数据依赖关系

包含1个或多个

包含1个或多个

使用1个

管理1个或多个

使用1个或多个

实现1个

包含0个或多个

读取/修改1个

根据1个跳转

LANGGRAPH

NODE

EDGE

STATE_MANAGER

STATE

MEMORY_OPERATION

string

type

深拷贝/浅拷贝/写时复制

OPTIMIZATION_SCHEME

string

name

5种优化方案之一

float

latency_improvement

延迟优化幅度

float

memory_improvement

内存优化幅度

NESTED_OBJECT

2.4.2 交互关系图(执行流程)

这张图展示了默认深拷贝机制下,LangGraph图执行过程中「节点→状态管理器→边→下一个节点」的完整交互流程

结束边 节点2(preliminary_filter) 条件边 节点1(preprocessor) 状态管理器(默认CoW?不,默认是深拷贝) 起始边 LangGraph图 用户 结束边 节点2(preliminary_filter) 条件边 节点1(preprocessor) 状态管理器(默认CoW?不,默认是深拷贝) 起始边 LangGraph图 用户 默认机制: 1. 对S0做深拷贝 2. 在深拷贝的S0上合并Δ1 3. 生成S1 默认机制: 1. 对S1做深拷贝 2. 在深拷贝的S1上合并Δ2 3. 生成S2 调用Graph.invoke(initial_state) 1 触发起始边 2 初始化状态 3 返回初始状态S0 4 传递初始状态S0 5 执行逻辑,生成状态更新Δ1 6 返回状态更新Δ1 + 边跳转指令(跳条件边) 7 请求合并状态更新Δ1到S0,生成S1 8 返回新状态S1 9 触发条件边,传递S1 10 根据S1的某个值判断跳转到节点2 11 传递状态S1 12 执行逻辑,生成状态更新Δ2 13 返回状态更新Δ2 + 边跳转指令(跳结束边) 14 请求合并状态更新Δ2到S1,生成S2 15 返回新状态S2 16 触发结束边,传递S2 17 返回最终状态S2 18 返回最终状态S2 19

3. 基础理解:为什么状态复制会这么慢?

3.1 深拷贝和浅拷贝的直观演示(用Python代码+生活化比喻搞懂!)

在开始深入LangGraph的状态管理器之前,我们先花一点时间,用Python代码+生活化比喻的方式,彻底搞懂「深拷贝」和「浅拷贝」的区别——这是理解所有状态优化方案的基础!

3.1.1 生活化比喻:装玩具的盒子

假设我们有一个旧盒子,里面装了:

  1. 一张纸条(不可变的字符串),上面写着“乐高积木”
  2. 一个轮子(可变的列表),里面有四个“轮胎”(不可变的字符串
  3. 一个窗户(可变的字典),里面有“颜色”=“蓝色”、“大小”=“10cm”(都是不可变的字符串

现在,我们有两个管理员:

  • 管理员A(深拷贝):给你一个一模一样的新盒子,再把旧盒子里的每个东西都单独复制一份——纸条复制成新的纸条,轮子复制成新的轮子,窗户复制成新的窗户——新旧盒子里的东西完全独立
  • 管理员B(浅拷贝):给你一个一模一样的新盒子,但只把旧盒子里的纸条复制一份,轮子和窗户只复制标签——新旧盒子里的标签指向的是同一个轮子和同一个窗户

现在,我们来做两个实验:

  1. 实验1:修改新盒子里的纸条——新旧盒子里的纸条会不会互相影响?
  2. 实验2:修改新盒子里的轮子(加一个备胎)和窗户(把颜色改成红色)——新旧盒子里的轮子和窗户会不会互相影响?
3.1.2 Python代码演示

我们用Python代码来复现上面的生活化比喻:

import copy

# ------------------------------
# 1. 创建旧盒子(旧状态)
# ------------------------------
old_box = {
    "brand": "乐高积木",  # 不可变的字符串
    "wheels": ["轮胎1", "轮胎2", "轮胎3", "轮胎4"],  # 可变的列表
    "window": {"color": "蓝色", "size": "10cm"}  # 可变的字典
}

print("=== 初始状态(旧盒子) ===")
print(f"旧盒子地址: {id(old_box)}")
print(f"旧盒子 brand 地址: {id(old_box['brand'])}")
print(f"旧盒子 wheels 地址: {id(old_box['wheels'])}")
print(f"旧盒子 window 地址: {id(old_box['window'])}")
print("-" * 80)

# ------------------------------
# 2. 管理员A:深拷贝
# ------------------------------
new_box_deep = copy.deepcopy(old_box)

print("=== 深拷贝后的新盒子 ===")
print(f"新盒子地址: {id(new_box_deep)}")
print(f"新盒子 brand 地址: {id(new_box_deep['brand'])}")
print(f"新盒子 wheels 地址: {id(new_box_deep['wheels'])}")
print(f"新盒子 window 地址: {id(new_box_deep['window'])}")
print("-" * 80)

# ------------------------------
# 3. 管理员B:浅拷贝
# ------------------------------
new_box_shallow = copy.copy(old_box)  # 或者用 new_box_shallow = old_box.copy()

print("=== 浅拷贝后的新盒子 ===")
print(f"新盒子地址: {id(new_box_shallow)}")
print(f"新盒子 brand 地址: {id(new_box_shallow['brand'])}")
print(f"新盒子 wheels 地址: {id(new_box_shallow['wheels'])}")
print(f"新盒子 window 地址: {id(new_box_shallow['window'])}")
print("-" * 80)

# ------------------------------
# 4. 实验1:修改新盒子里的 brand(不可变的字符串)
# ------------------------------
print("=== 实验1:修改深拷贝新盒子的 brand ===")
new_box_deep["brand"] = "小米积木"
print(f"旧盒子 brand: {old_box['brand']}")
print(f"新盒子 brand: {new_box_deep['brand']}")
print(f"旧盒子 brand 地址: {id(old_box['brand'])}")
print(f"新盒子 brand 地址: {id(new_box_deep['brand'])}")
print("-" * 80)

print("=== 实验1:修改浅拷贝新盒子的 brand ===")
new_box_shallow["brand"] = "小米积木"
print(f"旧盒子 brand: {old_box['brand']}")
print(f"新盒子 brand: {new_box_shallow['brand']}")
print(f"旧盒子 brand 地址: {id(old_box['brand'])}")
print(f"新盒子 brand 地址: {id(new_box_shallow['brand'])}")
print("-" * 80)

# ------------------------------
# 5. 实验2:修改新盒子里的 wheels(可变的列表)和 window(可变的字典)
# ------------------------------
print("=== 实验2:修改深拷贝新盒子的 wheels 和 window ===")
new_box_deep["wheels"].append("备胎")
new_box_deep["window"]["color"] = "红色"
print(f"旧盒子 wheels: {old_box['wheels']}")
print(f"新盒子 wheels: {new_box_deep['wheels']}")
print(f"旧盒子 window: {old_box['window']}")
print(f"新盒子 window: {new_box_deep['window']}")
print(f"旧盒子 wheels 地址: {id(old_box['wheels'])}")
print(f"新盒子 wheels 地址: {id(new_box_deep['wheels'])}")
print(f"旧盒子 window 地址: {id(old_box['window'])}")
print(f"新盒子 window 地址: {id(new_box_deep['window'])}")
print("-" * 80)

print("=== 实验2:修改浅拷贝新盒子的 wheels 和 window ===")
# 先把旧盒子的 brand 改回“乐高积木”,方便对比
old_box["brand"] = "乐高积木"
new_box_shallow["wheels"].append("备胎")
new_box_shallow["window"]["color"] = "红色"
print(f"旧盒子 wheels: {old_box['wheels']}")
print(f"新盒子 wheels: {new_box_shallow['wheels']}")
print(f"旧盒子 window: {old_box['window']}")
print(f"新盒子 window: {new_box_shallow['window']}")
print(f"旧盒子 wheels 地址: {id(old_box['wheels'])}")
print(f"新盒子 wheels 地址: {id(new_box_shallow['wheels'])}")
print(f"旧盒子 window 地址: {id(old_box['window'])}")
print(f"新盒子 window 地址: {id(new_box_shallow['window'])}")
print("-" * 80)
3.1.3 代码运行结果分析

你可以自己运行上面的代码,这里我们把关键结果列出来分析:

  1. 深拷贝的所有地址都不一样:旧盒子、深拷贝新盒子的地址不一样,旧盒子里的wheels、window的地址和深拷贝新盒子里的也不一样——完全独立
  2. 浅拷贝的外壳地址不一样,但嵌套可变对象的地址一样:旧盒子、浅拷贝新盒子的地址不一样(外壳复制了),但旧盒子里的wheels、window的地址和浅拷贝新盒子里的完全一样(只复制了标签)
  3. 修改不可变对象(brand),新旧盒子都不会互相影响:不管是深拷贝还是浅拷贝,修改brand都会创建一个新的字符串对象(因为Python中的字符串是不可变的),所以新旧盒子的brand地址不一样,内容也不一样
  4. 修改深拷贝的可变对象(wheels、window),新旧盒子不会互相影响:因为深拷贝复制了整个可变对象,所以新旧盒子的wheels、window地址不一样,内容也不一样
  5. 修改浅拷贝的可变对象(wheels、window),新旧盒子会互相影响:因为浅拷贝只复制了标签,指向的是同一个可变对象,所以修改其中一个,另一个也会变——这就是浅拷贝最大的风险

3.2 深拷贝的开销到底有多大?(用数学模型+Python性能测试搞懂!)

现在,我们已经搞懂了深拷贝和浅拷贝的区别——接下来,我们用数学模型+Python性能测试的方式,搞懂“深拷贝的开销到底有多大”?

3.2.1 深拷贝的数学模型(时间复杂度+空间复杂度)

假设我们有一个状态字典S,它的结构是:

  • S有k个顶层键值对
  • 每个顶层键值对的值要么是不可变对象(比如字符串、整数、浮点数、元组),要么是可变嵌套对象(比如列表、字典、集合、自定义类)
  • i个可变嵌套对象的「大小」(可以理解为内存占用的字节数,或者需要遍历的元素个数)是s_i
  • 所有不可变对象的总大小是s_immutable
  • 所有可变嵌套对象的总大小是s_mutable = sum_{i=1}^{m} s_i,其中m是可变嵌套对象的个数

那么,深拷贝的时间复杂度空间复杂度可以表示为:

  • 时间复杂度O(k+simmutable+smutable)=O(Stotal)O(k + s_immutable + s_mutable) = O(S_{total})O(k+simmutable+smutable)=O(Stotal),其中Stotal=k+simmutable+smutableS_{total} = k + s_immutable + s_mutableStotal=k+simmutable+smutable是整个状态的「总大小」——也就是说,深拷贝需要遍历整个状态的所有元素,不管是可变的还是不可变的
  • 空间复杂度O(Stotal)O(S_{total})O(Stotal)——也就是说,深拷贝需要分配和原状态一样大的内存空间,不管是可变的还是不可变的

浅拷贝的时间复杂度空间复杂度呢?

  • 时间复杂度O(k)O(k)O(k)——也就是说,浅拷贝只需要复制顶层的k个键值对,不需要遍历嵌套对象
  • 空间复杂度O(k+simmutable′)O(k + s_immutable')O(k+simmutable)——其中simmutable′s_immutable'simmutable是顶层不可变对象的大小(如果顶层不可变对象是小的,比如字符串、整数,Python会做「驻留(Interning)」优化,不需要分配新的内存空间,所以simmutable′s_immutable'simmutable可能为0)——也就是说,浅拷贝只需要分配很小的内存空间,用来存储顶层的键值对和可能的小不可变对象

对比一下就知道了:如果状态的总大小StotalS_{total}Stotal很大(比如有一个1MB的numpy数组,smutables_mutablesmutable就是1MB对应的元素个数),那么深拷贝的时间和空间开销会是浅拷贝的几千倍甚至几万倍

3.2.2 Python性能测试(真实数据说话!)

我们用Python代码来测试一下「深拷贝」和「浅拷贝」在不同大小的状态下的延迟内存占用——我们会测试4种常见的状态结构:

  1. 小状态:只有几个顶层键值对,没有嵌套对象(Stotal≈100BS_{total} \approx 100BStotal100B
  2. 中等状态:有几个顶层键值对,一个100KB的列表(Stotal≈100KBS_{total} \approx 100KBStotal100KB
  3. 大状态:有几个顶层键值对,一个1MB的numpy数组(Stotal≈1MBS_{total} \approx 1MBStotal1MB
  4. 超大状态:有几个顶层键值对,一个10MB的numpy数组(Stotal≈10MBS_{total} \approx 10MBStotal10MB

我们会用timeit模块测试延迟,用memory-profiler模块测试内存占用——首先,你需要安装这两个模块:

pip install timeit memory-profiler numpy

然后,我们写测试代码:

import copy
import timeit
import numpy as np
from memory_profiler import memory_usage

# ------------------------------
# 1. 定义4种不同大小的状态
# ------------------------------
def create_small_state():
    return {
        "query": "什么是LangGraph?",
        "temperature": 0.7,
        "max_tokens": 1024
    }

def create_medium_state():
    return {
        "query": "什么是LangGraph?",
        "temperature": 0.7,
        "max_tokens": 1024,
        "search_results": [f"搜索结果{i}" for i in range(10000)]  # 大约100KB
    }

def create_large_state():
    return {
        "query": "什么是LangGraph?",
        "temperature": 0.7,
        "max_tokens": 1024,
        "embeddings": np.random.randn(128, 1024).astype(np.float32)  # 128*1024*4=524,288 B ≈ 512KB?不对,我们改成256*1024=262,144个float32,就是1,048,576 B ≈ 1MB
        "embeddings": np.random.randn(256, 1024).astype(np.float32)
    }

def create_super_large_state():
    return {
        "query": "什么是LangGraph?",
        "temperature": 0.7,
        "max_tokens": 1024,
        "embeddings": np.random.randn(2560, 1024).astype(np.float32)  # 2560*1024*4=10,485,760 B ≈ 10MB
    }

# ------------------------------
# 2. 定义测试函数
# ------------------------------
def test_deep_copy(state):
    return copy.deepcopy(state)

def test_shallow_copy(state):
    return copy.copy(state)

# ------------------------------
# 3. 测试延迟(用timeit)
# ------------------------------
print("=== 测试延迟(timeit,每个测试运行10000次,取平均值) ===")
print("-" * 120)

# 小状态
small_state = create_small_state()
deep_copy_small_latency = timeit.timeit(lambda: test_deep_copy(small_state), number=10000) / 10000 * 1e6  # 转换成微秒
shallow_copy_small_latency = timeit.timeit(lambda: test_shallow_copy(small_state), number=10000) / 10000 * 1e6
print(f"小状态(≈100B):")
print(f"  深拷贝延迟: {deep_copy_small_latency:.2f} μs")
print(f"  浅拷贝延迟: {shallow_copy_small_latency:.2f} μs")
print(f"  优化幅度: {((deep_copy_small_latency - shallow_copy_small_latency) / deep_copy_small_latency * 100):.2f}%")
print("-" * 120)

# 中等状态
medium_state = create_medium_state()
deep_copy_medium_latency = timeit.timeit(lambda: test_deep_copy(medium_state), number=1000) / 1000 * 1e6  # 转换成微秒,只运行1000次,因为太慢了
shallow_copy_medium_latency = timeit.timeit(lambda: test_shallow_copy(medium_state), number=10000) / 10000 * 1e6
print(f"中等状态(≈100KB):")
print(f"  深拷贝延迟: {deep_copy_medium_latency:.2f} μs")
print(f"  浅拷贝延迟: {shallow_copy_medium_latency:.2f} μs")
print(f"  优化幅度: {((deep_copy_medium_latency - shallow_copy_medium_latency) / deep_copy_medium_latency * 100):.2f}%")
print("-" * 120)

# 大状态
large_state = create_large_state()
deep_copy_large_latency = timeit.timeit(lambda: test_deep_copy(large_state), number=100) / 100 * 1e6  # 转换成微秒,只运行100次
shallow_copy_large_latency = timeit.timeit(lambda: test_shallow_copy(large_state), number=10000) / 10000 * 1e6
print(f"大状态(≈1MB):")
print(f"  深拷贝延迟: {deep_copy_large_latency:.2f} μs")
print(f"  浅拷贝延迟: {shallow_copy_large_latency:.2f} μs")
print(f"  优化幅度: {((deep_copy_large_latency - shallow_copy_large_latency) / deep_copy_large_latency * 100):.2f}%")
print("-" * 120)

# 超大状态
super_large_state = create_super_large_state()
deep_copy_super_large_latency = timeit.timeit(lambda: test_deep_copy(super_large_state), number=10) / 10 * 1e6  # 转换成微秒,只运行10次
shallow_copy_super_large_latency = timeit.timeit(lambda: test_shallow_copy(super_large_state), number=10000) / 10000 * 1e6
print(f"超大状态(≈10MB):")
print(f"  深拷贝延迟: {deep_copy_super_large_latency:.2f} μs")
print(f"  浅拷贝延迟: {shallow_copy_super_large_latency:.2f} μs")
print(f"  优化幅度: {((deep_copy_super_large_latency - shallow_copy_super_large_latency) / deep_copy_super_large_latency * 100):.2f}%")
print("-" * 120)

# ------------------------------
# 4. 测试内存占用(用memory-profiler)
# ------------------------------
print("\n=== 测试内存占用(memory-profiler,每个测试运行1次,取峰值内存增量) ===")
print("-" * 120)

# 小状态
small_state = create_small_state()
deep_copy_small_memory = max(memory_usage((test_deep_copy, (small_state,)), interval=0.001)) - memory_usage()[0] * 1024 * 1024  # 转换成字节
shallow_copy_small_memory = max(memory_usage((test_shallow_copy, (small_state,)), interval=0.001)) - memory_usage()[0] * 1024 * 1024
print(f"小状态(≈100B):")
print(f"  深拷贝内存增量: {deep_copy_small_memory:.2f} B")
print(f"  浅拷贝内存增量: {shallow_copy_small_memory:.2f} B")
print(f"  优化幅度: {((deep_copy_small_memory - shallow_copy_small_memory) / deep_copy_small_memory * 100) if deep_copy_small_memory > 0 else 0:.2f}%")
print("-" * 120)

# 中等状态
medium_state = create_medium_state()
deep_copy_medium_memory = max(memory_usage((test_deep_copy, (medium_state,)), interval=0.001)) - memory_usage()[0] * 1024 * 1024
shallow_copy_medium_memory = max(memory_usage((test_shallow_copy, (medium_state,)), interval=0.001)) - memory_usage()[0] * 1024 * 1024
print(f"中等状态(≈100KB):")
print(f"  深拷贝内存增量
Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐