本地AI Agent长会话内存优化实战：分片存储与缓存回收方案

1. 项目概述：先说说我为什么要折腾这个

1.1 长会话内存问题的真实场景

如果你也跑过本地AI Agent，尤其是那种整天挂在后台、隔几分钟就要对话一次的常驻型Agent，大概率会遇到同一个问题：刚启动的时候内存占用还挺正常，跑个半天一天之后，内存占用肉眼可见地在涨，再过几天系统就开始卡顿，最后只能杀掉进程重来。

我这次踩的就是这个坑。本地跑的是一个基于主流开源大模型的Agent服务，承载的功能不算复杂：挂了一个内部知识库，接了好几个工具调用，最主要的是支持多轮会话，用户可以在一个会话里连续提问、让Agent反复推理、调用工具、修正结果。看起来没什么特别的，但问题恰恰出在“多轮会话”这四个字上。

先说一下我当时的运行环境：

• 硬件：一台32GB内存的工作站，显卡8GB显存，大模型跑在CPU+部分GPU混合推理模式下
• 框架：Python写的Agent调度层，底层通过Ollama调用本地模型，会话历史用列表保存在内存里
• 会话特征：单个会话最长可以做到200+轮，多用户并发，最多同时在线20个会话左右
• 现象：运行8小时后，进程内存从启动时的1.2GB涨到7.8GB，16小时后逼近12GB，最后OOM被系统杀掉

当时我还以为是模型推理有内存泄漏，后来排查下来发现：模型本身的内存占用基本稳定，真正失控的是会话上下文相关的缓存和服务端的消息历史。这就要说到长会话场景下一个非常典型的架构问题了——你不敢丢上下文，但你又没地方放那么多上下文。

1.2 为什么不能简单粗暴地“少存点东西”

可能有人会说，你不存历史不就行了？每次只带最近几轮对话去调模型，内存自然就降下来了。

这个思路在简单对话机器人上是成立的，但在Agent场景下完全走不通。因为Agent的行为和普通聊天不一样：它在多轮交互中会依赖早期的用户意图、工具返回结果、中间推理结论。比如用户在前几轮说了一个需求约束，后续的工具调用结果要持续遵守这个约束；又比如用户在第五轮让Agent生成了一段代码，然后到第三十轮要求基于那段代码做修改。如果没有完整历史，Agent就是“失忆”状态，回答质量会断崖式下跌。

所以核心矛盾非常清晰：既要保证长会话的连续性，又要控制内存不能无限膨胀。粗暴截断历史会牺牲Agent的能力，全量保留历史会拖垮内存。

我前前后后试了大概两周，最终落地了一套“会话分片存储 + 分级缓存 + 主动回收”的方案。说实话，这套方案在大型系统里不新鲜，但很多人在本地小规模部署时容易忽略它，或者不知道如何用轻量级手段实现。这篇文章就把我实际踩过的坑和最终跑通的方案完整记录下来，给同样在折腾本地Agent的朋友一个参考。

2. 方案设计思路：把“无限膨胀的历史”变成“可分层管理的数据”

2.1 长会话内存为什么会失控：先算一笔账

在给出具体方案之前，我们先搞清楚内存到底消耗在哪里。

我用Python的内置工具tracemalloc和objgraph跑了一轮分析，发现内存大头由三部分组成：

• 消息历史对象本身：每轮对话包含用户消息、助手消息、工具调用记录、工具返回结果，这些字段多且杂。以一条带工具调用的消息为例，包含消息ID、角色、内容、时间戳、工具名、工具参数、工具结果、Token数等多个字段，光一个Python字典加字符串就轻松到几十KB。
• 上下文窗口的重复拼接：每次向模型发起推理请求时，需要把历史消息拼成一份完整的上下文发送给推理引擎。为了不破坏原始历史，我最初的做法是每次copy.deepcopy一份完整历史再拼接，这意味着200轮会话的历史在每次请求时都会产生一份完整的临时复制。
• 推理框架自身的KV Cache：长上下文的KV Cache是推理内存的大头。虽然KV Cache不由我的Python层控制，但它会直接影响系统总内存，而且会话越长PV越大。

这里有一个关键点：很多人以为把历史消息列表从Python内存里挪走就完事了，但实际上推理框架的KV Cache才是真正的内存大户。如果你用的推理框架不支持自动化管理历史KV Cache，就只能靠控制每次送入的上下文长度来控制它。

为此我做了一个简单的压测统计（单会话、连续对话）：

从表中可以明显感受到：消息历史本身的内存增长是线性的，但KV Cache在超过模型负载能力后会指数恶化，甚至直接爆显存或爆内存。

所以我的核心思路是：Python层尽量少持有原始消息对象，只在真正需要构造上下文的那一刻才把消息拼起来，用完立刻释放；同时控制单次进入推理框架的上下文长度，超出限制的部分用摘要或关键信息替代。这样既能维持Agent的连续性，又能把内存保持在一个比较健康的水位。

2.2 方案选型：本地部署能做的最小改动方案

在选型时我对比过几种思路：

• 引入Redis或外部数据库存储会话历史— 能解决内存问题，但对本地单机部署来说太重了，还要额外维护一个服务。而且即使存到外部，推理时依然要把内容读回来构造上下文，治标不治本。
• 引入向量数据库做长期记忆— 这是很多RAG方案的常规做法，但实现成本较高，还要处理嵌入模型、相似度检索等额外环节。对普通长会话来说有点杀鸡用牛刀。
• 消息分片 + 磁盘冷存储 + 分段召回— 结构轻量、逻辑直观、不需要额外服务，只需要一套文件存储策略和缓存管理策略。
• 直接用LangChain等框架内置的对话记忆组件— 确实省事，但自定义控制不够，尤其是工具调用类的历史格式，框架默认的处理方式经常会丢关键字段。

最后选了第三种，同时和LangChain的memory机制做了对比参考，在此基础上自己写了一套轻量的实现。这套方案的特点概括成一句话就是：历史不常驻内存，推理前按需组装，组装结果有TTL回收，超长内容自动降级压缩。

整套方案的组件划分大致是这样的：

1. 会话消息分片存储模块：把对话历史按轮数和Token数切成固定大小的分片，热分片放内存，冷分片落磁盘
2. 推理上下文构造器：每次请求时从分片里挑出需要的消息，组装成上下文，构造器只服务于当前这一次请求
3. 缓存管理器：给组装好的上下文加TTL和引用计数，空闲一段时间后自动回收
4. 上下文压缩器：当历史太长时，把早期消息总结成结构化摘要，用摘要替代原始消息参与拼接
5. 监控与统计：记录每轮会话的内存变化、缓存命中率、上下文长度，方便定位异常

我实际落地后，跑了48小时压力测试，进程内存稳定在2.5GB以内，对比之前8小时就涨到7.8GB的情况，改善非常明显。下面把每个组件的具体实现和踩坑点逐一展开。

3. 分片存储实现：从全量常驻内存到按需加载

3.1 分片的基本单位：怎么切才是合理的

分片的第一步是确定切分单位。直接按“每N轮一切”是最容易想到的，但我实测后发现问题很大——因为每一轮的Token数差异非常大。连续普通的Q&A可能一轮只有100个Token，但一旦涉及工具调用，工具返回的JSON可能一轮就有3000到5000个Token。按轮数切分会出现分片大小严重不均匀的情况，后续做冷热判断和上下文截断时都很难统一处理。

所以我最终采用“按轮数为主、Token数兜底”的双重切分规则：

• 主规则：每10轮对话为一个分片
• 兜底规则：当单个分片的累计Token数超过8000时，强制切一个新分片（以角色信息完整为前提）
• 分片边界记录：每个分片记录起始轮号、结束轮号、消息数量、Token总数、最后活跃时间

这样做的好处是每个分片的大小基本可控，后续加载时可以根据目标Token上限，精确地决定需要加载哪些分片。举个例子，如果当前需要对上下文做12000 Token以内的请求，先加载最新分片，然后向前逐片加载，直到总Token数接近上限为止。

在实现上，我定义了一个简单的数据模型：

class SessionShard: def __init__(self, session_id: str, shard_id: int, start_round: int, end_round: int): self.session_id = session_id self.shard_id = shard_id self.start_round = start_round self.end_round = end_round self.messages = [] # 消息对象列表 self.total_tokens = 0 self.last_active_ts = time.time() self.loaded_from_disk = False # 标记是否从磁盘加载 def add_message(self, message: dict, tokens: int): self.messages.append(message) self.total_tokens += tokens self.end_round += 1 self.last_active_ts = time.time() def estimate_tokens(self): """估算当前分片的Token数，用于分片切分判断""" return self.total_tokens

你可能注意到我这里直接用了total_tokens字段而不是现场重新数Token，这是有原因的——数Token非常昂贵。我试过在每轮消息到达时调用模型的Tokenizer去数一遍，每次耗时几十毫秒，高频对话场景下累计开销非常惊人。所以我只在消息写入分片时用len(...) // 4这种粗略估算方法，或者在第一次真正进入推理上下文时再精算一次并更新到字段里。

3.2 冷热分离：内存只留“最近用得到”的分片

切好分片之后，下一步是决定哪些分片留在内存、哪些落盘。我的策略如下：

• 每个会话最多保留3个最新分片在内存中（覆盖最近30轮对话左右）
• 超过3个分片时，最旧的分片被序列化到本地磁盘
• 分片被访问时，如果它不在内存中，从磁盘反序列化并放回内存，然后把当前内存中最旧的分片淘汰掉
• 淘汰策略用LRU（Least Recent Used），但在LRU之外增加一项保护机制——正在被推理请求引用的分片不会被淘汰

这个策略的取舍逻辑是：Agent的推理往往依据的是最近几轮的对话状态，特别是工具调用链集中在会话尾部。把最近的内容留在内存里，可以保证高频请求不需要读磁盘，而早期的历史虽然存在，但只有在需要时才会被捞回来。

磁盘文件的组织方式上，我用的是每个分片一个JSON文件，目录结构如下：

storage/ sessions/ {session_id}/ shard_00000.json shard_00001.json shard_00002.json

文件命名中的序号就是分片ID，天然有序，方便加载时的顺序读取。每次落盘时把消息列表转换成JSON字符串写入临时文件再重命名，避免写到一半进程崩溃产生损坏文件。

这里插一个实操重点：JSON序列化大文本时的内存开销不可小觑。一个几十MB的分片在序列化时会python内再产生一份几十MB的字符串对象，如果同时多个分片落盘，容易出现瞬时内存尖峰。我的处理方法是先json.dumps到临时变量后再写文件，这个临时变量用完后马上del，并调用gc.collect()帮助回收。如果你不想手动调GC，也可以直接用json.dump直接写入文件对象，这个方法不会在内存中生成完整的字符串，只是写入过程中内存占用比较平稳。实测推荐用json.dump，处理300KB以上的分片效果明显。

3.3 消息对象的瘦身：从源头减少内存消耗

在分片方案基础上，我还做了一步非常重要的优化——给消息对象“瘦身”。

最初我设计的消息结构是直接把模型API返回的完整对象塞进历史列表里，这导致每个消息对象都保留了大量冗余字段。例如一次工具调用会保存完整的请求参数、响应体、异常堆栈、耗时统计、甚至中间日志。这些东西对调试有帮助，但对后续上下文恢复没有任何用处，白白增加了内存占用。

我重新设计了一个轻量级消息结构：

class LightMessage: __slots__ = ("role", "content", "tool_name", "tool_args", "tool_result", "timestamp", "token_count") def __init__(self, role, content, tool_name=None, tool_args=None, tool_result=None, timestamp=None, token_count=None): self.role = role self.content = content self.tool_name = tool_name self.tool_args = tool_args self.tool_result = tool_result self.timestamp = timestamp self.token_count = token_count

用__slots__替代普通类的__dict__后，每个消息实例的内存占用大约下降了一半。在Python中每个实例默认会维护一个属性字典，这个字典本身就有不小的开销，而__slots__直接让每个实例不再需要属性字典，改成更紧凑的槽位存储。实测10万条消息的情况下，瘦身后的总内存比原来少了约40%。

同样的思想也应用到工具结果上。工具返回结果通常是一大段JSON文本，如果只是为后续拼接上下文，完全没必要保留格式化后的Python对象，保留原始字符串即可。等真正需要把它嵌入提示词时，直接以字符串形式拼接，省去了反复序列化/反序列化的开销。

3.4 分片加载与冷启动优化

在分片从磁盘加载时，有一个细节很容易被忽略：加载操作如果发生在推理请求的路径上，会造成一次较高的延迟毛刺。所以我在实现时先把加载结果放进一个临时缓存，待当前请求完成后，再在后台线程里把分片真正载入内存。这样用户看到的首包延迟不会因为磁盘IO而明显增加。

后台加载的实现逻辑比较简单：

def background_load_shard(session_id, shard_id): def _load(): shard = load_shard_from_disk(session_id, shard_id) shard_cache.set(session_id, shard_id, shard) Thread(target=_load, daemon=True).start()

一次推理请求可能需要加载多个分片，如果每个分片都开一个线程，线程数量会失控。所以我做了一个简单的队列，限制同时加载的分片数不超过2个。实测下来，普通机械硬盘上一次反序列化一个5MB分片耗时约200ms到300ms，SSD上基本50ms以内，后台加载的体验完全可以接受。

4. 缓存回收与内存治理：让内存回到“用多少取多少”的健康模式

4.1 分级缓存的等级划分：永远只计算最贵的部分一次

分片存储解决的是“历史不常驻”的问题，但还有一个隐藏的内存消耗来源——每次推理请求都需要把分片里的消息拼装成一份模型可用的上下文。如果每次都从头拼接，不仅CPU开销大，而且拼接过程中产生的临时字符串对象会大量堆积，最终让旧的内存迟迟不被释放。

我的做法是引入三级缓存：

• L1 热点缓存：最近5分钟内被使用过的上下文拼接结果，直接存在内存中，key是(session_id, 分片范围, 模型名)。命中时直接返回。
• L2 温缓存：5到30分钟内被使用过的上下文，但为了节约内存，不会保存完整的拼接结果，只保存分片加载状态和消息级hash，下次请求时只需做增量拼接而非全量拼接。
• L3 冷存储：超过30分钟未被访问的上下文不做任何缓存，每次请求从分片级别重新构造。

缓存失效策略是TTL加容量双限制，缓存最大占用内存200MB，超过时优先淘汰最久未使用的条目。我选择200MB作为阈值，是因为这个数量级对本地32GB内存的机器压力不大，同时又能覆盖到高峰时段的并发请求。

这里有一个容易忽视的问题：你缓存的到底是什么？如果你直接把拼接好的提示词字符串缓存，缓存会非常大，因为200轮会话的完整提示词可能达到数万Token，也就是几十KB到几百KB。一条还好，几百条缓存轻松上几百MB。我的方案是只缓存消息列表的轻量索引结构，真正的消息对象仍然指向分片内部的messages列表。拼接动作只有在模型请求发出前那一瞬间执行，拼接出的完整提示词结束后立即释放。

4.2 缓存回收机制：从引用计数到主动GC

JVM世界里有一句老话叫“没有GC解决不了的内存问题，如果有，就调一下GC参数”。Python的GC机制和JVM不一样，但思路可以借鉴。我用的是一个轻量级的引用计数辅助主动GC的方案：

• 每个上下文对象被引用时计数加1，释放时减1，引用数归零的对象立即清除引用
• 全局计时器每30秒扫描一次缓存表，清理过期条目和未被引用的分片对象
• 定期调用gc.collect()处理循环引用，但这只在必要时做，因为强制GC会带来短暂停顿

初版实现里我没有主动触发GC，导致一个非常奇怪的现象：明明代码里已经删除了对象引用，内存却还是只增不减。排查后发现是Python的内存分配器从操作系统申请的内存不会自动“归还”，即使不再使用，内存块仍然留在进程的arena里。要真正让操作系统回收内存，最直接的办法就是通过gc.collect()配合分段释放，或者更彻底一点，把大内存工作放到子进程中去做，完成后直接杀掉子进程，让操作系统完整收回它的内存空间。

我这个场景不涉及子进程，所以采用的策略是：

import gc import threading class CacheReclaimer: def __init__(self, interval=30): self.interval = interval self._stop = False def start(self): Thread(target=self._run, daemon=True).start() def _run(self): while not self._stop: time.sleep(self.interval) self._reclaim() def _reclaim(self): # 1. 清理过期缓存 expired_keys = [k for k, v in ctx_cache.items() if time.time() - v.last_access_ts > 1800] for k in expired_keys: ctx_cache.pop(k, None) # 2. 强制触发一次 GC gc.collect()

实测这个回收器每次执行时大约会带来20到50ms的停顿，对于偶尔一次的清理任务来说可以接受。如果你对延迟特别敏感，可以把gc.collect()改成gc.collect(0)只回收最新一代的对象，停顿会小很多，但回收效果会打折。

4.3 堆外内存的理解误区：本地Python服务也要留意底层堆

热词里有一个“堆外内存”，这里顺带提一嘴。JVM里的堆外内存是指不在Java堆管理范围内的内存，很多中间件和高性能服务都在用它来绕过GC停顿。我们Python本地服务虽然没有“堆外”的概念，但如果你在Agent底层用的是JVM系的应用服务器（比如某些模型网关），那堆外内存的管理同样值得注意。

典型的JVM堆外内存增长来自Netty的Direct Memory和本地线程栈，它们不受-Xmx管控，出了问题表现为Java进程的RSS持续上涨，而GC日志显示堆内使用率正常。排查时可以用NMT（Native Memory Tracking）打开本地内存追踪：-XX:NativeMemoryTracking=summary，之后用jcmd <pid> VM.native_memory summary查看各部分内存占用。如果确认是Direct Memory问题，通过-XX:MaxDirectMemorySize设置上限；如果是线程栈，则需要查是否有线程泄漏。

我虽然主服务是Python，但在一次排查中意外发现模型网关侧堆外内存也在涨。这提醒我们：复杂链路的本地AI服务通常不是单一技术栈，排查内存问题一定要沿请求链路逐层确认，不要只盯自己最熟悉的那一层。

4.4 手动触发内存整理的时机

GC只是帮Python回收了不可达对象的空间，但由于Python的malloc行为，很多时候即使回收了空间，进程占用的内存也不会立刻下降。这就要用到malloc_trim或者干脆把大内存任务放子进程执行。

在纯Python层面，一个常用但很多人不知道的方法是：

import ctypes def release_memory(): libc = ctypes.CDLL("libc.so.6") libc.malloc_trim(0)

这个函数会向glibc的内存分配器请求把空闲的堆内存归还给操作系统。实测在缓存清理后调用，内存占用能下降约10%到20%。Windows和macOS上没有对应的malloc_trim，所以这段代码要加平台判断，不能无脑执行。

更稳妥的方案：如果你真的需要伺候一个长期运行的内存敏感型服务，建议把容易膨胀的模块隔离到独立进程中，比如把“会话记忆管理”独立成一个memory进程，主进程通过IPC与它通信。这样即使记忆管理进程内存失控，杀掉重启它也不会影响主进程。这也是很多生产级Agent框架底层的设计模式。

5. 会话上下文的组装与压缩：如何在不丢失能力的前提下限制上下文长度

5.1 上下文窗口的动态组装逻辑

分片存储和缓存回收解决的是“历史数据的存储成本”，但真正送进模型推理的上下文长度也需要控制。每个模型的上下文长度是不同的，即便同一个模型，可用的推理长度也可能因为配置、量化方式而变化。

我用一个配置文件来管理模型上下文上限：

model_config: default_context_limit: 16000 # token max_output_tokens: 2048 reserved_buffer_tokens: 512

每次推理时，组装器可用的输入Token上限是context_limit - max_output_tokens - reserved_buffer_tokens，即约13300个Token。这个预留的Buffer非常关键，如果你把上下文塞到刚好等于上限，模型一生成输出就超限了，轻则报错，重则截断输出。

动态组装器的流程是这样的：

1. 获取当前会话的最新分片列表 2. 从最新分片开始向前加载分片，Token数累加 3. 当累加Token数超过可用上限时，停止加载更早的分片 4. 对最早的那个分片，从消息尾部向前丢弃部分消息，直到刚好满足上限 5. 将选中的消息序列按时间顺序排列，生成完整提示词

这个过程看似简单，但有一个重要的取舍问题：如果执行到第4步时过早地把某轮对话截断了一半，可能会导致Agent看到不完整的工具调用记录，误以为工具没有返回结果，或者上下文逻辑错乱。我后来调整策略：截断以“完整轮次”为单位，一旦某轮进入选择集，它所属的用户消息、助手消息、工具调用链必须全部包含，不能只保留部分消息。必要的话宁可让总Token数超过上限一点点，也不能丢弃同一轮内关联消息。

5.2 早期历史的摘要化：把“逐字记忆”变成“段落总结”

只靠截断无法满足所有场景，有些超长会话中，用户在很早之前做出的决策会在后续反复引用。直接丢弃早期历史会让Agent彻底“失忆”。我的方案是维护一个“会话记忆摘要”，在分片超过一定数量后自动触发摘要生成。

摘要的内容不是简单的“用户问了很多问题”，而是需要保留以下关键信息：

• 用户的核心目标与约束
• 已经确认的产品或方案决策
• 已完成的工具调用及关键结果
• 仍待办或尚未解决的问题
• 用户偏好与重要实体信息

这个摘要生成任务本身也要消耗Token和时间。如果每次都调用主模型生成摘要，开销不小。我的做法是用一个更小的快速模型来生成摘要，这样既能省资源，又不占用主模型的并发推理能力。

摘要的生成时机选在消息写入分片且该分片被判定为“不活跃”（比如超过10分钟没有新消息）时后台执行。生成后把摘要存到会话元信息中，之后当需要加载更早期历史时，优先用摘要替代原始消息参与上下文组装。只要模型的system prompt中明确告诉它“以下是对话早期的摘要，替代了原始对话记录”，Agent通常能保持良好的行为一致性。

5.3 上下文压缩参数的调优

摘要压缩中最容易踩的坑是压缩阈值设置不当。阈值太小会导致摘要频繁生成，每次生成都要消耗Token，压缩的收益完全被生成成本覆盖；阈值太大会导致摘要内容过多，甚至和原始历史的长度差不多。

我的初始参数是上下文超过1.2万Token就开始压缩，结果发现60%的会话在1万到1.5万Token之间反复横跳，压缩任务频繁触发，模型并发被摘要请求占满，正常对话变慢。后来我把触发阈值上调到2万Token，并且加了一个“压缩后大小必须小于压缩前50%才执行”的判断条件，问题才得到缓解。

另一个重要参数是摘要的Token预算。摘要内容本身不能太长，否则塞进提示词后留给实际历史的额度就少了。我设置的摘要上限是1200Token，通过一个独立的摘要模型调用生成，实测99%的会话摘要能控制在800Token以内。如果摘要生成器输出超过1200Token，会做一次截断和二次压缩，确保最终不超过预算。

5.4 组装过程中的临时对象释放

这一点是纯性能优化，但非常重要。组装完整提示词时会产生大量的字符串拼接和列表复制。如果用+号拼接大量字符串，时间复杂度是O(n^2)的，内存更是灾难，每次拼接都会产生一个新的大字符串，而旧的字符串会滞留在内存里等GC。我的做法是用List[str]收集所有消息片段，最后一次性用"".join(...)生成完整提示词。这个改动看起来细小，但在我这边降低了组装阶段约35%的内存峰值。

这个技巧在JAVA里面对应的是使用StringBuilder而非字符串直接拼接，在Go里面对应的是strings.Builder，思路完全一致。写Python的朋友尤其要注意，"a" + "b" + "c"在循环里写一百次，产生的临时字符串对象是上百个，GC压力非常大。

6. 实操过程：从监控到压测，把方案稳定到可上线水平

6.1 监控体系：先能看见内存曲线，才能谈优化

方案跑通之前，必须先有一套能够观察内存变化的工具。我这边用了三层监控：

第一层是系统级的，直接用psutil取样进程的RSS、VMS和CPU占用率，每5秒记录一次。这个维度能看到整体趋势，判断当前服务是否处于危险水位。

第二层是Python对象级的，用tracemalloc对关键的几类对象做采样统计。注意tracemalloc有一定性能开销，所以我没有全程开着，而是在压测期间分段开启，每段跑10分钟拿到快照，分析完毕后关闭。

第三层是业务层的，记录每个会话的消息数量、分片数量、缓存条目数、缓存内存估算值，方便把内存问题定位到具体会话。

在压测期间我写了一个简单的统计日志，每条长这样：

[metrics] total_rss=2450MB python_heap=680MB ctx_cache_entries=142 shard_mem=118MB shard_disk=2.3GB active_sessions=18 gc_cycles=3421

通过这个日志，我可以很直观地看到内存构成。如果某一天shard_mem异常增长，那一定是分片冷热淘汰逻辑出了问题，优先排查LRU的部分；如果python_heap增长而其他指标不变，那多半是消息对象本身在堆积，需要查消息生命周期管理。

6.2 压测设计与结果对比

压测方式我选择了模拟真实对话场景的脚本：模拟20个并发用户，每个用户持续与Agent对话，每轮间隔10到30秒随机，对话内容包含普通问答、信息查询和工具调用三种类型。跑48小时，观察内存增长曲线。

改造前的数据我先放出来做个基准：

• 启动：RSS 1.2GB
• 8小时：RSS 7.8GB
• 16小时：RSS 接近12GB，出现明显卡顿
• 24小时：进程被OOM Killer杀死

改造后的数据：

• 启动：RSS 0.9GB（因为消息结构瘦身了）
• 8小时：RSS 2.1GB
• 16小时：RSS 2.3GB
• 24小时：RSS 2.4GB
• 48小时：RSS 2.5GB，基本进入平台期

48小时内内存波动幅度约600MB，平台期稳定在2.5GB左右，过程中没有出现请求超时、上下文丢失、Agent“失忆”等问题。这个结果可以接受，说明方案达到了预期效果。

更让我高兴的是缓存命中率数据——L1缓存的命中率在稳定运行8小时后达到约70%，意味着70%的重复性请求不需要重新组装完整上下文，系统响应速度也明显变好了。

6.3 实操过程中的现场记录：遇到过的三个“没想到”

第一个“没想到”是：Python的列表删除元素不释放内存。我在实现分片LRU淘汰时，直接对内存中的列表执行pop(0)，结果内存纹丝不动。查阅后确认，Python列表在弹出头部元素时，只是把后面的元素向前移动，列表底层的数组容量并不会缩小，除非你重新创建一个新列表。所以我的淘汰逻辑改成了：把内存中剩余分片复制到新列表，然后整体替换原列表。这样旧列表的底层数组会被释放，新列表从头开始按需扩容。

第二个“没想到”是：同一个会话的并发写入会导致分片错乱。Agent场景下，如果用户在多个终端同时操作同一个会话，或者后台摘要任务与主对话同时写入历史，不加锁就会出现两个线程同时往同一个分片追加消息的情况，要么丢消息，要么分片边界错乱。解决方案是在会话级加一个写锁，同一时刻只允许一个线程写入该会话历史。读操作不需要加锁，因为读取时可接受一定程度的不一致。

第三个“没想到”是：JSON文件反复读写会带来磁盘碎片。24小时压测结束后我去查看存储目录，发现几千个JSON文件把目录搞得非常碎，单个文件大小也不均匀。这个不影响功能，但会影响之后清理和备份的速度。我给文件管理器加了统一的大小限制，每个分片文件超过10MB时进行二次切片，同时定期把过期的冷分片归档到压缩包中。

6.4 Python垃圾回收调优：不要默认值用到老

Python的GC虽然是自动的，但如果你的服务是长时间运行且会创建大量对象，手动调参非常必要。主要看三个参数：

• gc.set_threshold(700, 10, 10)：默认阈值对高对象创建率场景偏低，会导致GC过于频繁，大量时间花在年轻代清扫上。适当调高可以减少GC频率，但会让老年代回收周期变长。我实测在Agent场景下，700/10/10的组合比默认的700/10/10稍好用，但区别不大，真正有用的是下一项。
• gc.set_debug(gc.DEBUG_LEAK)：这个选项可以在开发期打印泄露对象的GC跟踪信息，看到哪些对象无法被收集。上线时记得关闭，否则日志会非常膨胀。
• gc.freeze()：这是Python 3.7+新增的功能，可以把启动阶段创建的对象冻结起来，之后GC不再遍历这些对象。启动加载模块时产生的对象通常不会再被引用，冻结后GC扫描量会明显下降。实测开启后GC耗时降低了约30%。

需要强调的是，GC参数的调整必须基于对象生命周期分析，不要盲调。我的建议是先开启tracemalloc跑一段时间，看清楚对象的分配和释放规律，再决定是否需要调节阈值和代际比例。

6.5 真实流量下长期运行的稳定性观察

48小时压测之后，我又把服务跑了一周，重点观察两件事：一是内存是否真的稳定，二是Agent的对话质量是否下降。

内存方面整体平稳。在高峰期同时有25个活跃会话、10个会话处于长会话状态（超过100轮）时，RSS峰值到过3.8GB，但压测结束后2小时内回落到2.3GB，说明缓存回收机制和分片淘汰规则正常工作。

对话质量方面，摘要压缩的效果需要重点关注。我抽查了20个超过150轮的会话，其中15个在早期历史被摘要化后仍能正确引用早期约束，另外5个出现了不同程度的“遗忘”现象。深入分析后发现，遗忘场景高度集中在“用户在第5轮提到过一个非常具体的数字/名称，在第120轮要求基于该数字做计算”这类需要精确记忆的场景。摘要模型在提炼时把精确数字摘要成了大致范围，导致后续Agent计算错误。

针对这个问题，我优化了摘要提示词，要求摘要生成器必须原样保留所有具体数字、日期、专有名词，不允许概括这些信息。优化后抽查的准确率提升到了95%以上。这说明摘要生成的质量很大程度上取决于提示词的设计，而不是模型的能力。

7. 通用工具函数与关键代码片段

7.1 Token计算与分片切分工具

前面反复提到Token估算，这里给一个我在用的轻量工具函数，不依赖额外库：

def estimate_tokens(text: str) -> int: """粗略估算Token数，中英文混合场景较为可靠""" if not text: return 0 # 中文/日文/韩文字符每个约为1个Token cjk_chars = sum(1 for ch in text if '\u4e00' <= ch <= '\u9fff' or '\u3040' <= ch <= '\u30ff' or '\xac00' <= ch <= '\xd7a3') # 其他字符每4个算一个Token other_chars = len(text) - cjk_chars return cjk_chars + other_chars // 4 + 1

这个方法不是完美的，和模型真正的Tokenizer会有偏差。实测在中文内容为主的情况下偏差在15%以内，在英文内容为主时偏差约20%。如果需要精确的Token数，最好的办法还是调用对应模型的分词器，但那就需要加载Token模型到内存里，对轻量级场景来说不划算。对于上下文截断这种场景，15%的偏差完全可以接受，大不了多预留一点Buffer。

分片切分函数的伪代码如下：

class ShardManager: def __init__(self, shard_max_rounds=10, shard_max_tokens=8000): self.shard_max_rounds = shard_max_rounds self.shard_max_tokens = shard_max_tokens self._shards = {} # session_id -> list[SessionShard] def add_message(self, session_id, message): shards = self._get_or_create_session_shards(session_id) current = shards[-1] tokens = estimate_tokens(message.get("content", "") or message.get("tool_result", "")) # 条件满足时切新分片 if (current.end_round - current.start_round >= self.shard_max_rounds or current.total_tokens + tokens > self.shard_max_tokens): current = self._new_shard(session_id, len(shards)) shards.append(current) current.add_message(message, tokens)

这段逻辑看起来平平无奇，但其实隐含了一个容易被忽略的点：判断切分时一方面看轮数，另一方面看Token数，是“或”关系不是“与”关系。如果你用“与”关系，即轮数和Token数都达到阈值才切分，那么对话轮数很多但每轮很短时会一直不切分；反之如果每轮很长但轮数不多，也可能长时间不切分。用“或”关系才能保证分片在任何维度上都不会超出预期。

7.2 缓存回收的线程实现要点

回收线程涉及共享内存操作，线程安全问题绕不开。Python的dict操作本身在CPython下因为有GIL是线程安全的，但两个线程并发做“先检查后删除”的操作仍然可能产生竞态。

安全实践是给回收器加一个独立的锁：

cache_lock = threading.Lock() def safe_reclaim(): with cache_lock: expired_keys = [k for k, v in list(ctx_cache.items()) if time.time() - v.last_access_ts > ttl] for k in expired_keys: ctx_cache.pop(k, None)

为什么回收操作要加锁？因为正常请求路径上也可能在读缓存、写缓存，如果回收线程在遍历 dict 的同时另一个线程在往dict里添加条目，CPython下虽然不会crash，但可能出现“遍历时修改dict”导致漏掉一些条目，或者触发RuntimeError: dictionary changed size during iteration。加上锁之后，回收和写入互斥，不会有这个问题。

不过GC操作本身不应该放在持锁状态下执行，否则当一个长请求正在写缓存时，GC会被无限期阻塞等待锁释放。正确顺序是：先持锁清理缓存条目，释放锁，再执行gc.collect()。在代码上就是两个独立的代码块：

with cache_lock: # 清理缓存条目的逻辑 gc.collect()

这个顺序问题是我在调试一次诡异卡顿时发现的。当时回收线程持着锁执行GC，而GC触发了堆扫描，需要遍历所有对象，结果和正在写入缓存的主线程产生竞争，整个主线程停顿了将近3秒钟。把GC挪到锁外之后问题彻底消失。

7.3 从磁盘恢复分片时的校验逻辑

最后再说一个磁盘持久化场景容易被忽视的点——数据完整性。进程崩溃时如果正好有分片在写磁盘，会留下半截JSON文件。虽然本地使用场景下很少崩溃，但不等于永远不会发生，比如突然断电或者OOM后系统强杀进程。

我的分片加载函数里加入了简单的容错：

def load_shard_from_disk(session_id, shard_id): path = get_shard_path(session_id, shard_id) try: with open(path, "r", encoding="utf-8") as f: return json.load(f) except (json.JSONDecodeError, FileNotFoundError): # 文件损坏，尝试加载上一个有效备份 backup_path = path + ".bak" if os.path.exists(backup_path): with open(backup_path, "r", encoding="utf-8") as f: return json.load(f) return None

对应的写入逻辑：先写主文件，再复制一份为.bak文件，两个文件都写完才认为本次落盘成功。这样即使主文件写了一半崩溃，还可以用备份文件恢复。由于备份文件总是在主文件成功写入后才更新，所以备份文件要么是完整的旧版本，要么是完整的新版本，不存在半截状态。

需要注意，“先写主文件再复制为bak”其实存在一个小窗口：如果主文件写完后还没来得及复制bak就崩溃，此时bak还是上一个版本，但主文件是最新且完整的。这种情况下从主文件恢复没问题，因为主文件已经是完整的。如果主文件写到一半崩溃，那它是不完整的，但这个情况下备份还没被覆盖，仍然是上次完整的旧版本，依然能恢复。所以这个顺序配合两文件方案是可靠的。

8. 常见问题与排查技巧

8.1 内存还在涨？按照这个顺序做现场定位

如果你也遇到了“内存只增不减”的怪问题，我建议按照下面这个顺序做现场定位，比漫无目的地抓瞎高效：

1. 先确认是Python对象内存还是非Python对象内存。用psutil.Process().memory_full_info()看uss和rss的差异，差得越大说明非Python内存占比越高，这时要检查是否有C扩展、子进程或推理框架在占用。
2. 如果是Python对象内存，用tracemalloc抓对象分配快照。连续抓几次，看哪些类型的对象在持续增加，定位到具体的分配栈。
3. 如果是某个特定列表或缓存持续增长，查它是否有删除路径。写了缓存但不清理、加了历史但从不淘汰，这是最常见的两个原因。
4. 如果一切正常内存还是涨，关注消息队列或日志堆积。我就遇到过因为日志处理队列消费慢于生产，导致消息对象在队列里堆积到几GB的情况。

排查过程中，建议用objgraph去查看某个类型对象的引用链，快速判断是谁在引用这些对象导致无法回收。命令是objgraph.show_backrefs([obj], filename="backrefs.png")，生成的图能直观看到引用关系，非常有用。

8.2 长会话上下文为什么“感觉变短了”

有朋友跑类似的方案后反馈：Agent会话长了之后，模型好像是忘记了早期内容。一开始他们以为是上下文压缩丢了历史，后来查看发现是分片加载时Token预算不足，早期分片根本没被加载进来。

这个问题的本质是：模型上下文长度是有限的，你的“摘要”和“最近历史”只是分配方式不同，并没有增加总预算。如果你的会话总内容超过模型上下文上限，那么在物理上就不可能把所有原始消息都送进模型。这时候的方案只能是在“摘要的质量”上下功夫，或者换用上下文更长的模型。

我给一条实操建议：如果预算紧张，优先保留最近10轮的完整历史，这覆盖了Agent当前工作状态；再往前的内容用时间衰减策略，比如5轮前到50轮前的消息每2轮保留1轮，50轮以外的全部压缩成摘要。这种折中在大部分Agent场景下的效果都不错。

8.3 缓存命中率为什么上不去

缓存命中率上不去，和会话内容高度动态有关。Agent的工具调用结果每次都可能不同，比如查天气、查库存这种实时性强的信息根本没法缓存。我一开始把整个上下文拼接结果作为缓存key，命中率只有30%左右，效果很差。

后来我把缓存粒度从“完整上下文”降为“分片消息列表”，只在分片没有被修改时复用，分片被修改后只做局部更新。命中率提高了不少。更进一步，对于包含动态工具结果的分片，我在缓存中单独标记“不可缓存”，这样该分片每次请求都走完整加载路径，不会因缓存了旧数据而给Agent喂过期信息。

8.4 排查工具速查

8.5 最具迷惑性的“假泄漏”：内存池机制

最后提醒一个迷惑性极高的情况。我在压测时看到Python进程的内存从2GB涨到4GB之后不再回落，通过对象分析也没发现明显的泄漏点。后来查了很久，才发现是Python的小对象分配器（pymalloc）从操作系统申请了大块内存后，虽然释放了一部分小对象，但是原先申请的大块内存没有被归还给操作系统。这个状态下内存“涨上去”了但并没有真正的“泄漏”，因为后续需要分配对象时可以直接复用这块缓存区。

判断是否属于这种情况的方法很简单：观察CPU。如果内存涨到高位后CPU占用没有继续异常增长，说明服务运行是健康的，只是单纯的内存池保留问题。这时不需要紧张，如果实在在意空闲内存的数字，定时调用malloc_trim(0)即可。但如果内存增长的同时伴随CPU占用持续走高，那大概率是真正的业务对象泄漏，需要深入排查代码逻辑。

9. 总结与个人实操感受

9.1 方案适用边界与未来可能的方向

这套分片+缓存回收方案适合本地部署、单机运行、并发量中低等的Agent服务。如果你的场景是云原生的多实例高并发，其实直接用现成的Redis或对象存储来管理会话历史更合理，没必要自己在每个实例里维护一套分片逻辑。但本地个人项目、内网小规模部署、开发测试环境，这套方案可以说是性价比最高的选择了。

后续我还在考虑几个优化方向：

• 把分片存储升级为SQLite，替代散文件，事务性和查询方便性都能上一层楼。
• 摘要生成尝试用流式方案，避免大段摘要一次生成占住模型推理资源。
• 在缓存层加入持久化能力，让Agent重启后还能快速恢复最近一段时间的会话状态。

9.2 给同样在踩坑路上的朋友几句实在话

这次折腾下来，我最大的感受是：本地AI Agent的内存问题，很多时候不是模型推理框架的问题，而是我们外围代码把不值得留在内存里的东西都留在了内存里。模型推理的KV Cache是固定的，但历史消息、缓存条目、临时拼接对象这些都是可以管理掉的。把心思花在数据生命周期管理上，内存问题大概率就迎刃而解。

如果想把内存控制做好，建议从最简单的监控开始。先把“内存增长的来源”搞清楚，再谈优化。不要在一开始就背着“一定要用什么高级缓存框架”的包袱。我整篇文章讲的这些方法和函数，任何一个熟悉Python的朋友都能自己实现，核心是思路——数据分层、按需加载、缓存有界、超限回收。把这四个词刻在脑子里，比任何框架都靠谱。

如果你现在正被长会话内存问题困扰，不妨从今天开始，把你会话历史里哪些是热点、哪些是冷数据先分级列出来，看看有多少内容是真的需要常驻内存的。大概率你会发现，真相是：你以为需要常驻的，其实绝大多数都是可以按需再加载的。想明白这一点，你的内存问题就已经解决了一大半。