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是9，preserve消息的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、你业务最常见的对话长度分布。
以我常用的配置为例，模型上下文窗口设定为32000，max_new_tokens设定为4096，那么预警阈值大约是多少呢？计算逻辑为：当current_tokens + max_new_tokens超过32000的85%，也就是27200时，就应该强制触发压缩。这个30%的余量不是浪费，而是为模型生成时的中间推理过程留出缓冲空间。如果卡得太紧，压缩后摘要本身又占据几百token，再加上本轮生成内容，有可能再次触碰上限；如果太松，模型可能在大量早期垃圾消息上浪费注意力。
兼容不同业务形态的参数组合，我整理了一张表可以参考：
| 业务场景 | context_window | max_new_tokens | 触发阈值 | 压缩频率 |
|---|---|---|---|---|
| 轻量问答（短历史） | 8192 | 1024 | 6800 | 不频繁 / 按轮次兜底 |
| 代码生成与执行（多工具调用） | 16000 | 2048 | 13600 | 每5~8轮检查一次 |
| 数据清洗长流程 | 32000 | 4096 | 27200 | 每10轮检查一次 |
| 多Agent协作 | 64000 | 8192 | 54400 | 按token压力触发为主 |
这里要强调一句：表里的“压缩频率”不是越频繁越好。压缩本身也是一次模型调用，消耗时间和token。如果一个任务总共只有15轮，那其实完全不需要定期压缩，只靠轮次上限兜底就够了。压缩在长流程里才会体现价值，短对话频繁压缩反而得不偿失。
3.2 从配置到代码接入
在Pi-Agent里开启Compaction非常轻量化。项目默认不开启，需要通过配置对象定义主动打开开关，类似这样：
from pi_agent import AgentConfig config = AgentConfig( enable_compaction=True, context_limit=32000, max_new_tokens=4096, compaction_freq=5, round_limitation=80, # 可选：打开时同时采用缓存复用机制 reuse_cache=True, ) agent = AgentFactory.create(config)只要配置了enable_compaction=True，框架内部会在每次执行步骤前自动调用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字段）的消息，系统写入时自动设置preserve=True。与其指望模型摘要不丢细节，不如在源头保住细节。
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方向的同行们一点参考，尤其是正在被“历史越跑越乱”困扰的人，能顺着它的设计思路找回一点清晰感。
