☰
Agent上下文压缩与Checkpoint持久化:原理、实现与实战避坑
2026/10/20 20:06:19 网站建设 项目流程

1. 从一次上下文溢出说起:为什么Agent必须学会“忘掉”一些东西

如果你正在用大模型搭建Agent,大概率遇到过这样的场景:一个跑了十几轮对话的Agent,突然开始胡言乱语,或者干脆报错说上下文超限。你检查代码,逻辑没问题;你检查提示词,也没问题。问题出在一个很容易被忽略的地方——上下文窗口被塞满了。

大模型的上下文窗口是有限的。不管是32K、128K还是更长的窗口,只要你让Agent持续运行,它迟早会撞上这堵墙。更麻烦的是,Agent和普通聊天机器人不一样。普通聊天机器人一轮对话就是一轮,用户问一句答一句,上下文增长是线性的、可控的。但Agent会自己调用工具、自己产生中间结果、自己迭代推理,一轮任务下来可能产生几十条消息,上下文膨胀速度远超预期。

这就是上下文压缩(Compaction)要解决的问题。而Checkpoint持久化解决的是另一个问题:当Agent运行到一半,进程挂了、服务重启了、或者用户主动中断了,怎么让它从断点恢复,而不是从头再来。

这两个机制放在一起看,其实是一件事的两面:Compaction管的是“空间”,Checkpoint管的是“时间”。空间不够就压缩,时间断了就存档。我在实际项目里踩过不少坑,下面把这两个机制拆开讲透。

2. 上下文压缩的核心逻辑:不是删消息,而是做信息蒸馏

2.1 上下文膨胀的真实来源拆解

很多人以为上下文膨胀就是“对话轮数太多”。这只说对了一小部分。在一个典型的Agent执行链路里,上下文里塞的东西远比你想的复杂:

  • 系统提示词:通常几百到几千token,固定开销,一般不动它。
  • 用户输入:每轮对话的原始输入,累积起来不可忽视。
  • Agent的推理文本:模型每次输出的思考过程,尤其是带思维链的模型,这部分增长极快。
  • 工具调用请求:Agent决定调用哪个工具、传什么参数,这些结构化数据会留在上下文里。
  • 工具返回结果:这是最大的膨胀源。一次网页抓取可能返回上万token的HTML,一次数据库查询可能返回几百行JSON。
  • 历史消息:以上所有内容按轮次累积。

我实测过一个简单的任务:让Agent查三个网页、总结成一份报告。三轮工具调用下来,上下文从最初的2K token直接飙到45K token。其中工具返回结果占了将近80%。所以压缩的重点,首先是工具返回结果,其次才是历史对话。

2.2 三种压缩策略的取舍与适用边界

上下文压缩不是简单地“截断”或“删除”。粗暴截断会让Agent丢失关键信息,导致后续推理断裂。我总结下来,实际可用的策略有三类:

第一类:滑动窗口截断。保留最近N轮消息,超出的直接丢弃。这是最简单的做法,但问题很明显——早期的重要信息(比如用户最初的需求描述)会被丢掉。适合那种“每轮独立、不需要长期记忆”的场景,比如简单的问答Agent。

第二类:摘要压缩。把旧消息交给模型,让它生成一段摘要,用摘要替换原始消息。这样能保留语义信息,同时大幅减少token。但摘要本身有信息损失,而且摘要生成需要额外调用一次模型,有延迟和成本。

第三类:结构化蒸馏。这是我在生产环境里最推荐的做法。不是让模型自由发挥写摘要,而是按照预定义的结构提取关键信息。比如工具返回结果,只保留状态码、关键字段、错误信息,把冗余的格式、重复内容、无关字段全部剔除。

提示:摘要压缩和结构化蒸馏的区别在于,前者是“让模型决定保留什么”,后者是“你告诉模型必须保留什么”。生产环境里,后者更可控。

2.3 压缩触发时机的判断:阈值、轮次还是语义信号

什么时候触发压缩?常见做法是设一个token阈值,比如用到窗口的70%就触发。但这个做法有个问题:阈值设高了,压缩时已经很挤,压缩效果不好;阈值设低了,频繁压缩,增加延迟和成本。

我现在的做法是组合判断:

  • 硬阈值:上下文占用超过窗口的75%,必须压缩。
  • 轮次阈值:每完成5轮工具调用,主动检查一次。
  • 语义信号:当Agent开始重复之前已经做过的操作,或者工具返回结果明显与当前任务无关时,触发压缩。

第三点比较微妙,需要你在Agent循环里加一些启发式判断。比如检测Agent是否在重复调用同一个工具、传相似的参数,这往往意味着上下文里的信息已经混乱了。

3. 压缩的具体实现:从消息结构到蒸馏模板

3.1 消息分类与优先级标记

要实现结构化蒸馏,第一步是给上下文里的每条消息打标签。我通常分成四个优先级:

优先级消息类型压缩策略
P0系统提示词、用户原始需求永不压缩
P1最近3轮对话不压缩
P2工具调用请求与关键返回字段蒸馏保留
P3中间推理文本、冗余工具返回可摘要或丢弃

这个分类不是固定的,你可以根据任务类型调整。比如做代码生成任务时,代码片段属于P1,不能压缩;做信息检索任务时,检索结果的关键字段属于P2。

3.2 工具返回结果的蒸馏模板设计

工具返回结果是压缩的主战场。我一般会为每类工具定义一个蒸馏模板。以网页抓取为例,原始返回可能是完整的HTML,蒸馏后只保留:

def distill_web_result(raw_html, max_length=500): # 提取标题 title = extract_title(raw_html) # 提取正文纯文本 text = extract_text(raw_html) # 截断到指定长度 if len(text) > max_length: text = text[:max_length] + "...[truncated]" return { "title": title, "content": text, "status": "success" }

这样一次网页抓取从上万token压缩到几百token,信息密度大幅提升。关键是,Agent后续推理需要的是内容本身,而不是HTML标签。

对于数据库查询结果,蒸馏逻辑类似:只保留字段名和值,去掉重复的行、空值、格式化字符。如果结果行数太多,只保留前N行加一个统计摘要。

3.3 摘要生成中的信息保真技巧

有些内容没法用模板蒸馏,比如Agent的推理过程、多轮对话的语义。这时候需要摘要。但直接让模型“总结一下”效果往往不好,容易丢关键信息。

我的做法是给摘要模型一个明确的检查清单:

  • 用户的核心目标是什么?
  • 已经完成了哪些步骤?
  • 当前处于哪个阶段?
  • 有哪些未解决的问题?
  • 有哪些关键数据需要保留?

让模型按这个结构输出摘要,而不是自由发挥。实测下来,结构化摘要的信息保真度比自由摘要高很多。另外,摘要里要保留具体的数值、名称、ID,这些是后续推理的锚点,一旦丢失很难恢复。

4. Checkpoint持久化:让Agent拥有“存档读档”能力

4.1 什么该存、什么不该存

Checkpoint的核心是决定存什么。存多了,恢复慢、存储成本高;存少了,恢复后Agent状态不对。

我的经验是存这几样:

  • 消息历史:完整的对话和工具调用记录,这是恢复的基础。
  • Agent状态:当前处于哪个步骤、下一步该做什么。
  • 工具调用结果缓存:已经执行过的工具调用及其结果,避免恢复后重复调用。
  • 压缩后的上下文快照:如果做过压缩,存压缩后的版本,而不是原始版本。
  • 元数据:时间戳、版本号、任务ID。

不该存的东西:模型的临时推理状态(这个没法存)、外部系统的实时状态(恢复时需要重新查询)。

4.2 存储介质选型:内存、文件还是数据库

三种选择,各有适用场景:

内存存储:最快,但进程重启就没了。适合短生命周期、不需要跨进程恢复的场景。实现最简单,一个字典就够了。

文件存储:把Checkpoint序列化成JSON或Pickle写到本地文件。适合单机部署、任务量不大的场景。恢复时读文件反序列化即可。缺点是并发写入需要加锁,多进程场景下容易出问题。

数据库存储:适合生产环境。可以用关系型数据库存结构化字段,用对象存储存大块的上下文快照。支持并发、支持查询、支持过期清理。我一般用一张表存Checkpoint元数据,上下文快照单独存。

选型的判断标准很简单:你的Agent会不会跨进程恢复?会不会多实例部署?如果会,直接上数据库,别犹豫。

4.3 序列化格式的坑:JSON、Pickle与自定义协议

序列化格式的选择直接影响恢复的成功率。

JSON的好处是通用、可读、跨语言。但问题是它不支持复杂对象,比如Python的datetime、自定义类实例,都需要手动转换。而且JSON序列化大对象时性能一般。

Pickle是Python原生的,支持几乎所有对象,序列化反序列化都很快。但它有安全风险——反序列化不可信的Pickle数据可能执行任意代码。另外Pickle跨Python版本兼容性不好,升级Python版本后老Checkpoint可能读不出来。

我现在的做法是混合使用:元数据用JSON存,方便查询和调试;大的上下文快照用压缩后的JSON或MessagePack,兼顾性能和兼容性。自定义类统一转成字典再序列化,恢复时再转回去。

注意:不管用什么格式,一定要加版本号。Agent的逻辑会迭代,Checkpoint的格式也会变。没有版本号,恢复时遇到格式不匹配就是一场灾难。

4.4 恢复流程的幂等性设计

Checkpoint恢复最大的坑是重复执行。假设Agent执行到第5步时存了Checkpoint,然后进程挂了。恢复后从第5步继续,但如果第5步的工具调用已经产生了副作用(比如发了一封邮件、写了一条数据库记录),重复执行就会出问题。

解决办法是幂等性设计:

  • 每个工具调用带一个唯一ID,执行前先检查这个ID是否已经执行过。
  • 有副作用的操作,执行前先查状态,确认没做过再执行。
  • Checkpoint里记录已执行的操作ID列表,恢复时跳过这些操作。

这个设计需要在工具层面配合,不是Agent框架单独能解决的。但如果你要做生产级的Agent,这一步绕不过去。

5. 压缩与Checkpoint的协同:两个机制怎么配合

5.1 压缩后立即存档的策略

压缩和Checkpoint不是独立的。我的做法是:每次压缩完成后,立即存一次Checkpoint。原因很简单——压缩是有成本的(调用模型生成摘要),如果压缩完还没存档就挂了,下次恢复还得重新压缩。而且压缩后的上下文更小,存档也更快。

具体流程是:触发压缩条件 → 执行压缩 → 生成压缩后的上下文 → 写Checkpoint → 继续执行。这样即使后续步骤失败,恢复时直接加载压缩后的版本,不用重做压缩。

5.2 恢复时的上下文重建顺序

恢复时不是简单地把Checkpoint读出来就完事。上下文重建有顺序要求:

  1. 先加载系统提示词和用户原始需求(P0内容)。
  2. 加载压缩后的历史摘要。
  3. 加载最近几轮的完整消息(P1内容)。
  4. 加载工具调用缓存,用于幂等性检查。
  5. 恢复Agent状态,确定下一步动作。

这个顺序不能乱。如果先加载了最近消息再加载摘要,可能会导致消息顺序错乱,模型理解出问题。

5.3 版本兼容与迁移处理

Agent逻辑迭代时,Checkpoint格式可能变化。比如你给消息结构加了一个新字段,老Checkpoint里没有这个字段,恢复时就会报错。

处理办法有两个:一是向后兼容,新代码读取老Checkpoint时给缺失字段填默认值;二是版本迁移,写一个迁移函数,把老格式转成新格式。我一般两者结合:小改动用默认值兼容,大改动写迁移脚本。

另外,Checkpoint要设过期时间。太老的Checkpoint恢复后可能已经没意义了,定期清理能省存储空间,也能避免误用。

6. 实战中踩过的坑与性能调优经验

6.1 压缩过度导致Agent“失忆”的案例

有一次我为了省token,把压缩阈值设得很激进,上下文用到50%就触发压缩,而且摘要写得很简略。结果Agent在执行一个多步任务时,做到第4步突然问用户“你刚才说的需求是什么”。原因是压缩把用户最初的需求摘要得太狠,只剩一句话,丢失了关键约束条件。

教训是:P0内容永远不要压缩。用户原始需求、系统提示词、关键约束,这些必须原样保留。压缩只能动P2和P3的内容。另外摘要要有最低信息量保证,不能为了短而短。

6.2 Checkpoint写入频率与性能的平衡

Checkpoint写太频繁,I/O成为瓶颈;写太少,恢复时丢失的进度太多。我试过每步都写,结果Agent执行速度下降30%以上。后来改成关键节点写入:每完成一个工具调用写一次,压缩后写一次,任务阶段切换时写一次。这样既保证了恢复粒度,又不会拖慢执行。

如果用的是数据库,还可以用异步写入——Checkpoint先写内存队列,后台线程批量落库。这样对主流程几乎无影响。但要注意进程崩溃时内存队列里的数据会丢,所以关键Checkpoint还是要同步写。

6.3 大上下文序列化的耗时优化

上下文大了之后,序列化本身就很耗时。我测过一个200K token的上下文,JSON序列化要几百毫秒。优化手段有:

  • 用更快的序列化库,比如orjson、MessagePack。
  • 只序列化变化的部分,做增量Checkpoint。
  • 压缩后再序列化,减少数据量。
  • 大块内容单独存对象存储,Checkpoint里只存引用。

增量Checkpoint是个好思路,但实现复杂,需要跟踪哪些消息是新增的。如果任务不长,全量写也能接受,就别过度设计。

6.4 恢复后工具缓存失效的处理

工具缓存失效是个隐蔽的坑。比如Agent查了一个网页,缓存了结果。恢复后,如果直接复用缓存,可能拿到的是过期数据。但如果不用缓存重新查,又失去了Checkpoint的意义。

我的处理方式是给缓存加TTL(生存时间)。恢复时检查缓存是否过期,没过期就用,过期了就重新调用。TTL设多长取决于工具的性质:查天气可能几分钟就过期,查文档可能几小时都没问题。

另外,对于有副作用的工具调用,缓存不是用来复用的,而是用来做幂等性检查的。这类缓存不能设TTL,必须永久保留到任务结束。

7. 一个可复用的压缩与Checkpoint实现骨架

7.1 核心数据结构定义

下面是我常用的数据结构,用Python伪代码表示:

class Message: def __init__(self, role, content, priority, msg_id): self.role = role # system/user/assistant/tool self.content = content # 消息内容 self.priority = priority # P0-P3 self.msg_id = msg_id # 唯一ID self.timestamp = time.time() class Checkpoint: def __init__(self, task_id, version): self.task_id = task_id self.version = version self.messages = [] # 消息列表 self.agent_state = {} # Agent状态 self.tool_cache = {} # 工具调用缓存 self.executed_ops = set() # 已执行操作ID self.created_at = time.time()

这个结构足够简单,也足够灵活。priority字段是压缩的依据,executed_ops是幂等性的依据。

7.2 压缩器的接口设计

压缩器我设计成一个可插拔的组件:

class Compactor: def should_compact(self, context) -> bool: # 判断是否需要压缩 pass def compact(self, context) -> context: # 执行压缩,返回压缩后的上下文 pass

不同的压缩策略实现不同的Compactor。滑动窗口是一个,摘要是一个,结构化蒸馏是一个。Agent循环里只调接口,不关心具体实现。这样切换策略很方便。

7.3 Checkpoint管理器的读写流程

class CheckpointManager: def save(self, checkpoint): # 序列化并写入存储 pass def load(self, task_id) -> Checkpoint: # 从存储读取并反序列化 pass def should_save(self, step_count, event_type) -> bool: # 判断是否该存Checkpoint pass

should_save里放写入频率的控制逻辑。event_type可以是"tool_call_done"、"compaction_done"、"phase_change"等,不同事件有不同的写入策略。

7.4 与Agent主循环的集成点

集成到Agent主循环里大概是这样:

while not task_done: # 1. 检查是否需要压缩 if compactor.should_compact(context): context = compactor.compact(context) checkpoint_manager.save(build_checkpoint(context)) # 2. 执行一步Agent逻辑 action = agent.step(context) # 3. 执行工具调用(带幂等性检查) if action.type == "tool_call": if action.id in checkpoint.executed_ops: result = checkpoint.tool_cache[action.id] else: result = execute_tool(action) checkpoint.executed_ops.add(action.id) checkpoint.tool_cache[action.id] = result # 4. 关键节点存Checkpoint if checkpoint_manager.should_save(step_count, "tool_call_done"): checkpoint_manager.save(build_checkpoint(context))

这个骨架不复杂,但覆盖了压缩和Checkpoint的核心逻辑。实际项目里需要根据具体框架调整,比如LangChain、AutoGPT这些都有自己的消息结构,需要做适配。

8. 几个容易被忽略的边界情况

8.1 压缩时正在进行的工具调用怎么处理

如果Agent正在等一个工具返回结果,这时候触发压缩,要小心不要把这次工具调用的请求消息压掉了。否则工具返回结果回来时,找不到对应的请求,上下文就乱了。

我的做法是压缩时跳过最近一轮未完成的工具调用,只压缩已完成的部分。等工具返回、这一轮结束后,再纳入压缩范围。

8.2 多Agent场景下的Checkpoint隔离

多个Agent协作时,每个Agent有自己的上下文和状态。Checkpoint要按Agent ID隔离,不能混在一起。恢复时也要分别恢复,然后重建Agent间的通信状态。

如果Agent之间有共享状态(比如共享内存、共享任务队列),这部分状态不在Checkpoint里,恢复时需要额外处理。我一般会在Checkpoint里记录共享状态的引用或版本号,恢复时检查共享状态是否一致。

8.3 长任务中Checkpoint的清理策略

长任务跑几天,Checkpoint会积累很多。不能无限存着,需要清理策略:

  • 保留最近的N个Checkpoint。
  • 保留关键节点的Checkpoint(如阶段完成时)。
  • 任务完成后,延迟一段时间删除所有Checkpoint。
  • 设置最大存储配额,超了自动清理最老的。

清理策略要可配置,不同任务对恢复粒度的要求不一样。

8.4 压缩与Checkpoint的失败重试

压缩可能失败(模型调用超时、返回格式不对),Checkpoint写入也可能失败(磁盘满、数据库连接断)。这些失败不能导致Agent崩溃,要有降级方案:

  • 压缩失败:跳过本次压缩,继续执行,下次再试。但要记录失败次数,连续失败多次就告警。
  • Checkpoint写入失败:重试几次,还失败就写本地临时文件,后台再同步。最坏情况是放弃本次Checkpoint,但Agent继续跑。

关键是不能让辅助机制拖垮主流程。压缩和Checkpoint是为了让Agent更可靠,如果它们本身成了故障源,就本末倒置了。

9. 我在实际项目中的几点体会

做了几个Agent项目之后,我对压缩和Checkpoint的理解也在变化。一开始我觉得这是两个独立的功能,后来发现它们其实是一套“状态管理”体系的两个面。压缩管的是状态的空间效率,Checkpoint管的是状态的时间可靠性。两者共享同一套消息结构和优先级定义,配合好了才能让Agent真正跑得久、跑得稳。

另一个体会是,不要追求一步到位。我见过有人一上来就设计复杂的增量Checkpoint、多级压缩策略,结果调试成本极高,还不如先用最简单的全量Checkpoint加滑动窗口压缩跑起来,遇到问题再优化。Agent本身已经够复杂了,状态管理这块能简单就简单。

最后说一个具体的技巧:在开发阶段,把每次压缩前后的上下文都打日志存下来。这样当Agent行为异常时,你可以回放看是哪次压缩丢了关键信息。这个习惯帮我定位过好几次“Agent突然变傻”的问题,比盲目调提示词有效得多。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询