Pi-Agent源码解析四上下文压缩Compaction从第一次启动对话到跑完一条上百轮的工具调用链很多人都会遇到同一个奇怪现象前期模型逻辑清晰、指令执行得干净利落到了后期却开始“失忆”——忘了用户最早提的需求把几轮之前的中间结果当成最新状态甚至在同一件事上反复折腾。我最初以为是上下文窗口不够用但把日志拉出来一看当前轮次的token数明明还有大量富余。问题其实出在对话历史的组织方式上消息越堆越长有效信息被淹没模型每轮都要从几千条历史里“大海捞针”。这一篇要聊的就是Pi-Agent里解决这个问题的核心模块——Compaction也就是上下文压缩。Compaction做的事情并不复杂在不影响任务连续性的前提下对历史对话做有损或无损的压缩。它可以是一个模型生成的摘要也可以是一段代码执行后的结构化提炼甚至是对早期消息的直接裁剪。难的是怎么决定什么该压缩、什么该保留、什么时候触发压缩以及压缩之后如何保证后续推理还能拿到完整的关键信息。这些细节在Pi-Agent的源码里被拆成了清晰的几条链路。这篇文章我会顺着它的设计思路把触发机制、压缩策略、底层数据结构、参数影响范围一个个拆开讲最后再附上我在实际调试中踩过的坑和排查方法。1. 整体设计与思路拆解1.1 Compaction要解决的三个核心问题先看几个真实的数据。某Agent项目里跑了一个多步骤的数据清洗任务每步都会执行Python代码、读取临时文件、再根据结果决定下一步。整个流程跑下来原始消息记录有142条累计token约38000而模型的实际上下文窗口是64000。表面上看没超限但模型在第80轮之后开始频繁出错把上一轮的代码输出当成这一轮的、漏掉用户最初指定的过滤条件甚至有一次把任务目标从“清洗销售额为空的记录”理解成了“删除所有销售额小于100的记录”。这就是典型的“上下文污染”问题。Pi-Agent在设计Compaction时非常明确地盯住了三个目标第一把模型的处理焦点拉回到当前任务。对话历史的真实长度会迅速超过模型的注意力有效范围虽然窗口没满但早期内容对后续推理的贡献已经接近于零。压缩不是简单的删减而是要“提纯”只保留对后续决策有直接影响的信息。第二避免无上限的token累积。一个长期运行的服务端Agent单会话运行数小时是很常见的。如果不压缩内存和token成本都会跟着会话时长线性增长。Compaction对这两个维度做人为控制让每轮请求的token数保持在一个相对平稳的水平而不是一路爬升。第三让压缩结果具备“被继续使用”的能力。摘要不是给人类看的而是给模型看的结构化信息。所以压缩后的内容不能是松散的“故事梗概”必须保持时间线、实体状态、未完成任务、工具调用链路的完整性。过去有很多团队尝试过简单方案——超长就把最旧的消息丢掉。这种方式在短会话里能顶一顶但在任务本身高度依赖早期约束的场景里很容易把关键指令一起丢没。Pi-Agent的做法不是“一刀切”而是给每条消息定义存活期在压缩时区分“必须保留的核心数据”和“可以抽象概括的中间过程”同时额外保留一份专门的执行约束清单。1.2 从源码结构看Compaction在Pi-Agent里的位置Pi-Agent是一个我个人很偏爱的极简Agent框架。它的核心调度逻辑全部围绕一组纯函数式的“任务-执行-响应”循环展开代码风格非常符合阅读源码解析的需求。Compaction模块在整个框架里并不独立存在它与消息存储层、执行引擎、模型调用层紧密耦合这个设计是它最精妙的地方。在源码里Compaction相关代码主要分布在两处一处是负责管理消息列表和缓存对象的消息管理器另一处是独立的任务执行器它负责在合适的时机调用压缩策略并把压缩结果“塞回”上下文。消息管理器内部维护了几个核心对象消息队列message_queue按顺序存放全部历史消息每条消息带有role、content、created_at等基础字段。缓存对象cache_entry用于存储模型调用时使用的临时上下文可以理解为“即将发给模型的token序列”的容器。摘要缓冲区summary_buffer存储最近的摘要结果在压缩发生时回填给消息队列。Compaction真正被触发的位置在执行引擎的每一步迭代之前。这里有一个看似不起眼却至关重要的设计点它检查的不是历史消息的总条数也不是累计token数而是一个经过加权的计算值。这样能同时衡量上下文窗口压力和有效信息密度从而决定是否执行压缩。状态转移非常清晰第一消费完所有输入第二确认上下文已满第三执行摘要或丢弃逻辑第四重组消息队列后继续下一轮。这一套设计最值得学习的部分不是压缩算法本身而是它把“何时压缩”放在了“如何压缩”之上先解决触发时机再考虑压缩效果。1.3 为什么触发条件比压缩策略本身更关键在阅读任何Agent源码时我之前习惯先直接跳到最核心的算法或者策略代码。但读了Pi-Agent之后我强烈意识到触发机制的重要性它直接影响压缩的收益和损耗。触发时机如果太早摘要引入的失真会早早污染整段对话后续压缩会被反复摘要信息丢失像滚雪球一样最后连最开始的任务目标都变形。触发时机太晚历史消息已经膨胀到影响模型注意力的程度就算执行压缩前期“注意力涣散”的行为已经发生错误已经积累压缩能做的只是止血无法挽回已经产生的混乱。Pi-Agent的处理办法是不完全依赖程序设定的固定阈值而是引入一个基于运行步数的动态调整量。让我用源码里的核心计算逻辑来说明。1.4 关键源码Compaction触发判定框架内部通过一个环境变量和辅助类组合完成触发判定。下面是压缩判定相关的方法有简化和注释# 在任务调用链路上位于每一步执行前的检查点 def maybe_compact(self, context, turn_count, exec_steps): should_compact False # 实际项目的判断逻辑会结合多条匹配规则这里展示主要判断规则 if turn_count self._loop_limitation: # 通过“轮次限制”兜底如果超长轮次仍未结束强制触发压缩 should_compact True if exec_steps % self._compaction_freq 0: # 按执行步数定期压缩避免长期不触发 should_compact True if context.current_tokens self._max_new_tokens self._context_limit: # 窗口压力判断当前已用token 本次生成上限 硬性限制 should_compact True if self._manual_hint: # 业务方可主动下发压缩请求 should_compact True if should_compact: self._do_compact(context, context.history) return should_compact这段代码把触发条件拆成了四路人马。第一路是轮次上限这个值通常按业务场景设置例如一个单次任务超过80轮还没跑完说明对话已经失控此时压缩才有必要。第二路是按执行步骤定期触发默认每N步压缩一次类似定时整理房间。第三路是token压力检测这个是最关键的硬保障。第四路是外部指令驱动业务方可以主动要求压缩。其中第三路判断context.current_tokens self._max_new_tokens self._context_limit需要特别说明。它不只是判断“当前上下文快满了”而是把响应预留空间也计算在内。很多同学只看已用token数忽略max_new_tokens结果模型生成过程中直接撞上窗口上限报错或者被强制截断。Pi-Agent在这点上考虑得很细致。源码里还暴露了相关参数给定一个环境变量即可覆盖默认值比如# 实际项目中一般通过环境变量或配置对象传入 MAX_NEW_TOKENS 4096 # 单次生成最长长度 CONTEXT_LIMIT 32000 # 上下文窗口商用上限按模型能力设定 COMPACTION_FREQ 5 # 每隔5步执行一次检查 ROUND_LIMITATION 80 # 轮次上限有一说一这套触发体系胜在“多层兜底”token压力检测保障硬性不越界轮次和步数限制保证即使token远没满也能定期整理对话避免“有效注意力渐渐耗尽”。2. 核心机制拆解Compaction内部如何工作2.1 压缩策略摘要但不止于摘要我对Pi-Agent的Compaction机制印象最深的是它没有把所有压力都压在“生成摘要”这一个策略上。项目里实际生效的压缩策略是组合式的主要分为三种按优先级和上下文情况动态选择。第一种是“完全摘要模式”适用于对话已经严重超过窗口限制或者运行轮次超长的情况。此时框架调用一次模型把除系统消息以外的大部分消息聚合成一段结构化摘要。摘要生成时有一个重要的提示词约束必须保留所有未完成的事项、用户明确的指令约束、以及所有与当前任务相关的关键数值或文件路径。这个提示词写得很具体比如“如果上下文中出现/data/tmp/result.csv这样的路径必须在摘要中完整保留”这能有效避免模型摘要时只提炼感觉丢失机械细节。第二种是“截断模式”适用于部分早期消息已经完全失去价值的情况。典型场景是大量的工具调用结果、临时打印信息、一次性的中间输出。这类消息直接删除不会影响任务但如果反复执行摘要反而可能浪费算力。截断模式的做法是直接裁剪掉指定条数再通过一条固定短语代替类似于[早期工具输出已截断保留最近N轮消息]。这样做省token但会丢失掉早期可能存在的关键信息所以只在高轮次且消息类型高度重复时才启用。第三种是“可选保留模式”用于处理那些虽然旧但绝不能丢的约束信息。实现方式非常巧妙不是让摘要去“猜”哪些该留而是在消息写入时就已经打好标签——凡是被标记为preserve的消息压缩时完全跳过不参与摘要也不被截断。这就像大家平时写文档时把重点句标黄只不过Pi-Agent把“标黄”这个动作自动化了。用户在对话早期指定的硬性规则、配置文件路径、数据表名都会被自动标记preserve。这个“标签先行”的思路直接规避了摘要模型偶尔会遗漏关键数值的问题。纯摘要方案本质上是在让模型做记忆筛选只要模型漏一次后续整条链路都会跑偏。带标签的方案等于多了一道保险——不需要依赖模型判断哪些该留写入时就已经确定了。2.2 缓存复用机制压缩之后怎么保住关键细节压缩一个很容易被忽略的环节是缓存管理。很多同学做上下文压缩只处理消息列表把摘要替换进去就完事。但Pi-Agent的Compaction是和缓存对象一起运作的。在这个框架源码里每条消息在进入缓存对象时会附带一个过期时间戳和优先级。压缩发生时框架不是直接去改写文本而是让旧消息“提前过期”同时把摘要消息作为高优先级内容插入缓存对象的最前端。这样后续N轮生成时摘要会始终待在注意力窗口的有效范围内而不会因为它是“新消息”就被排到队列末尾。底层数据结构借用了PriorityQueue的思路源码里给每条被压缩过的消息加了一个priority字段0-9。摘要消息的priority是9preserve消息的priority是8普通旧消息是1被截断的消息是0。每次构建发送给模型的消息列表时先按priority从高到低排序再按时间顺序排列。这样做有一个很实际的好处当上下文接近上限时可以无差别优先丢掉priority为0的消息再考虑丢弃priority为1的消息摘要和核心约束永远是最后被淘汰的。这套机制在长会话里价值很大——它能保证“压缩”这件事本身不会伤害核心信息。2.3 Compaction链路的关键代码实现光看不写等于白看。源码解析部分我从项目里摘出了最核心的执行流程压缩逻辑的主链路代码如下def _do_compact(self, context, session_history): # 第一步从历史消息里分离“必须保留”和“可压缩”两类 preserve_msgs [m for m in session_history if m.get(preserve, False)] compressible_msgs [m for m in session_history if not m.get(preserve, False)] # 第二步生成摘要只针对可压缩部分 summary_text self._summarize(compressible_msgs) # 第三步重组上下文保留系统消息和preserve消息插入摘要 new_history [] for msg in session_history: if msg.get(role) system: new_history.append(msg) elif msg.get(preserve, False): new_history.append(msg) # 把摘要作为system消息的一部分放在最前保证优先注意 new_history.insert(1, {role: system, content: f[历史消息摘要] {summary_text}}) # 第四步用新历史替换旧历史 context.history new_history context.current_tokens self._count_tokens(new_history) self._trigger_cache_refresh()其中_summarize方法内部是向模型发起一次专门的摘要请求。有一个细节值得注意摘要请求使用的提示词会显式要求模型以“要保留的执行要素”为主线而不是复述对话过程。2.4 代码原理解读多一点_do_compact这段代码有几点设计精巧之处。系统消息在整体替换时不会被摘要覆盖始终保持原样。因为系统消息通常包含模型的角色设定和全局指令一旦被摘要或挪动模型后续行为会出现微妙的不稳定。preserve消息插入到历史和摘要之间既不在摘要生成前被混淆也不在后续检索时被排除。而摘要作为一条新的system消息插到第二位确保了模型一进入对话就能看到压缩后的全貌而不是在历史记录里翻找。实际运行中我验证过这套流程能有效防止token数线性增长。在一次长达200轮的测试里早期无压缩版本总token数上涨到接近上限而加入Compaction之后从第60轮开始每20轮压缩一次总token数始终被控制在限制范围内稳定在窗口大小的70%左右。3. 实操过程与核心参数选择3.1 配置Compaction参数前的现状评估在把Compaction接入项目之前第一步不是改代码而是先做一次“现状体检”。需要搞清楚三个数据你当前模型的实际上下文窗口大小、你单次生成常驻的max_new_tokens、你业务最常见的对话长度分布。以我常用的配置为例模型上下文窗口设定为32000max_new_tokens设定为4096那么预警阈值大约是多少呢计算逻辑为当current_tokens max_new_tokens超过32000的85%也就是27200时就应该强制触发压缩。这个30%的余量不是浪费而是为模型生成时的中间推理过程留出缓冲空间。如果卡得太紧压缩后摘要本身又占据几百token再加上本轮生成内容有可能再次触碰上限如果太松模型可能在大量早期垃圾消息上浪费注意力。兼容不同业务形态的参数组合我整理了一张表可以参考业务场景context_windowmax_new_tokens触发阈值压缩频率轻量问答短历史819210246800不频繁 / 按轮次兜底代码生成与执行多工具调用16000204813600每5~8轮检查一次数据清洗长流程32000409627200每10轮检查一次多Agent协作64000819254400按token压力触发为主这里要强调一句表里的“压缩频率”不是越频繁越好。压缩本身也是一次模型调用消耗时间和token。如果一个任务总共只有15轮那其实完全不需要定期压缩只靠轮次上限兜底就够了。压缩在长流程里才会体现价值短对话频繁压缩反而得不偿失。3.2 从配置到代码接入在Pi-Agent里开启Compaction非常轻量化。项目默认不开启需要通过配置对象定义主动打开开关类似这样from pi_agent import AgentConfig config AgentConfig( enable_compactionTrue, context_limit32000, max_new_tokens4096, compaction_freq5, round_limitation80, # 可选打开时同时采用缓存复用机制 reuse_cacheTrue, ) agent AgentFactory.create(config)只要配置了enable_compactionTrue框架内部会在每次执行步骤前自动调用maybe_compact方法做检查开发者不需要在业务代码里手动处理压缩逻辑。这是框架较友好的地方接入成本低业务代码零侵入逻辑全部内聚在调度层。3.3 把压缩接到长期记忆层如果你希望压缩结果不只是为当前会话服务还想沉淀为某种跨会话记忆需要在压缩后把摘要内容单独保存一份持久化副本。Pi-Agent本身不包含数据库存储但通过回调钩子可以很轻松地实现。在压缩事件触发后把摘要文本写入本地的JSON文件或SQLite表里后续新的会话启动时再把这部分摘要注入系统消息得到的效果非常类似“多轮记忆”。def on_compaction_finished(compacted_summary, meta): # 自定义回调保存摘要到本地存储 storage.append({ summary: compacted_summary, session_id: meta[session_id], timestamp: time.time(), })这个扩展思路适合把Pi-Agent从“单次任务工具”升级成“有长期记忆的辅助助手”。毕竟一旦任务时长超越了会话极限跨会话记忆和压缩其实是同一枚硬币的两面。3.4 压缩流程的完整复现演示为了让你更直观地理解我用一个模拟场景走一遍全流程。假设有一条数据分析任务链用户先指定“只分析2024年之后的订单”然后经过工具调用、数据读取、结果讨论等环节历史消息累计到65条此时缓存对象里已用的token数突破了阈值。第一步执行引擎在进入下一步前调用maybe_compact发现token数超额进入压缩流程。此时历史消息里有一条带preserve标签的约束消息“只分析2024年之后的订单”它不会被压缩。第二步框架对除preserve和最后3轮消息以外的所有历史消息发起摘要请求。摘要指示词里明确要求保留文件路径、条件参数、未完成任务。模型返回的摘要文本压缩了原始约60条消息token从18000降到800左右。第三步框架重建消息队列把摘要作为system消息插入紧跟真实系统消息之后。preserve消息保持原位最近3轮原始消息完整保留。总token数从27300降到约10000。第四步下一轮生成时模型先看到摘要再看到preserve约束和最近对话对用户最早的需求依然清晰。4. 常见问题与排查技巧实录4.1 压缩之后模型反而“听不懂”了怎么办这是Compaction上线时最常踩的坑。现象是压缩前模型能正常理解任务目标压缩后突然开始答非所问、丢三落四。排查思路是检查摘要文本本身是否完整覆盖了所有关键约束。很多情况下是摘要提示词里缺少“保留未完成事项”的强制性要求导致模型生成的摘要只概括了对话过程却没交代当前执行到哪一步。我在实战中把摘要提示词从“请总结以上对话”调整为“请总结以上对话并用列表列出已完成事项、未完成事项、下一步计划、关键参数”。调整之后压缩后的任务连续性明显上了一个台阶。如果你也遇到类似问题先看摘要内容里有没有结构化清单不要急着改压缩频率。4.2 压缩过于频繁导致token不降反升有一种冷启动场景容易被忽略会话刚开始两三轮历史消息很短但配置的compaction_freq很小导致每几轮就触发一次压缩。每次压缩调用模型生成的摘要文本可能比原始消息还长结果总token数不降反升甚至压缩本身成了主要的token消耗源头。这是“为了压缩而压缩”的经典反面教材。解决办法是给压缩加一条前置判断——当历史消息不足预设阈值比如20条或累计token不足5000时直接跳过压缩不管触发条件是否满足。def _should_compact(self, history): if len(history) self._min_history_len: return False return super()._should_compact(history)4.3 压缩时把工具调用的中间状态丢失了工具调用类Agent最严重的问题之一是压缩时把“某个工具已经返回的结果”给弄丢了。比如第10轮已经拿到了某个接口返回的access_token后续请求还在用压缩后这条消息被摘要成“用户获取了一个令牌”而摘要里没写token的实际值后续工具调用直接失效。排查方法是检查消息写入时是否给这类“状态型结果”打了preserve标签。在Pi-Agent里可以设置一个自动标记规则凡是包含特定模式如token、session_id、path字段的消息系统写入时自动设置preserveTrue。与其指望模型摘要不丢细节不如在源头保住细节。PRESERVE_PATTERNS [access_token, session_id, .csv, tmp/] def should_preserve(message_content): for p in PRESERVE_PATTERNS: if p in message_content: return True return False只要自动标记逻辑足够好Compaction就能做到“大块历史做摘要关键细节带标签”两条线互相配合比任何单一策略都稳。4.4 排查速查表再提供一个排查表按现象检索原因能帮你快速定位问题现象可能原因处理方式模型忘记早期指令摘要未包含用户初始约束改为结构化提示词强制保留“用户明确要求”清单工具调用报错变量未定义中间结果未标记preserve增加自动标记规则覆盖token、路径、关键值token消耗不降反升压缩过于频繁或摘要文本过长增加压缩最低历史条数限制精简摘要输出格式压缩后输出风格突变摘要消息插到了system前面导致系统指令错位严格保持system消息在最前摘要插入第二位长会话后期仍出低级错误触发阈值设得太晚已发生污染把阈值从85%下调到75%~80%并配合步数定期压缩完全没有压缩事件配置未生效或字段名拼写错误检查enable_compaction是否置真确认配置对象被正确加载5. 轻量化替代方案与设计启发5.1 不引入Compaction也能做工程取舍如果你暂时不想引入完整的Compaction机制也可以先做一个轻量版给每条消息打上时间戳每隔N轮就把早于M轮的消息全部标记为“可丢弃”在发送前直接过滤。这本质上就是“窗口滑动”的简化实现对代码结构改动最小但效果完全取决于你的业务中早期消息是否真的没用。对于强任务型Agent早期约束通常非常关键所以轻量版更适合闲聊型或检索型应用。另一种思路是把“压缩”交给模型自身不改变消息列表只调整系统提示词让模型“忽略早于某时间点的内容”。这个方法省事但效果不稳定尤其是面对长上下文时模型的注意力掌控能力是公认的弱项不建议在生产环境里依赖。5.2 从Compaction里学到的通用设计原则读Pi-Agent的Compaction实现我最直接的收获其实不是某个方法而是“标签先行”这个原则。任何需要做有损处理的地方都要先在无损阶段把“绝对不能丢的东西”打上标记。这不仅是上下文压缩的核心思想也是很多系统设计的通用解法——先确定“边界条件”再做“常规优化”。无论是缓存淘汰、数据清理、还是配置降级先问一句什么是绝对不能动的把这个答案定下来后面的优化才敢放开手脚。6. 实操总结含经验教训根据我个人的实际操作体验和多次踩坑后的总结对Compaction最终可以归纳出四条经验。第一触发时机是压缩的灵魂。一个设计良好的Agent最重要的不是它的摘要写得多漂亮而是它知道“什么时候不要动手”。不要在短对话里高频压缩不要等到窗口满了才后悔莫及。Pi-Agent反复让人感知到真正专业的系统都是靠多个维度共同约束让主逻辑能长期稳定运行而不是依赖某一次“爆发式救场”。第二摘要和标签要成对使用。纯摘要方案在长对话中本质上是在赌模型每次都能归纳得一丝不苟这本身就是不现实的。标签机制像是一个“安全网”把那些不允许丢失的细节提前捞走。在部署到生产环境时我的建议是宁可多打几个preserve标签也不要少打——多保留几条消息的token成本远远小于丢失关键信息后任务失败带来的重跑成本。第三压缩不是只能单向执行。实际业务里有时候你会在压缩后发现后续需要更早的某条消息但消息已经没了。如果你使用的是支持可逆存储的方案比如把原历史保存到磁盘/数据库可以在压缩后冷存一份需要时再取出做拼接。这个设计可以作为Compaction的自然扩展。第四Compaction的价值不在于让token变少而在于让模型的注意力重新回到关键信息。做了压缩之后不仅要看总token是否降下来了更要盯着任务成功率有没有实质提升。如果你发现压缩后任务成功率没有变化甚至略有下降先别急着优化压缩策略回头检查原对话历史里是不是有太多无用信息影响了模型判断。换句更实在的话说Compaction是Agent工程里最容易被忽视却又最值得投入的一块基础设施。谁先把上下文整理这件事做好谁的长跑型Agent就能比别人稳得多。希望这篇源码解析能给在做Agent方向的同行们一点参考尤其是正在被“历史越跑越乱”困扰的人能顺着它的设计思路找回一点清晰感。