代码读到第四篇，终于碰到所有长会话系统的命门：上下文窗口不够用。Pi-Agent这个智能体项目，在连续跑完几十轮对话、大量工具调用和记忆写入之后，如果不对历史上下文做压缩处理，模型面对的就不是对话，而是一堆互相矛盾的指令和几乎占满预算的废料。Compaction模块要做的事情，是用尽量小的Token代价，把已经发生过的会话重新组织成能被继续引用的有效记忆。这篇文章我用源码解析的视角，把触发机制、数据结构、分块策略、摘要链和验证方法完整拆开，适合正在做智能体长期记忆、多轮工具调度，或者准备优化Token成本的后端工程师。
1. 为什么必须有Compaction：从内存窗口和对话保留的矛盾切入
1.1 不压缩时，长会话会怎么退化
先把账算清楚。假设一个Agent的系统提示写得比较全，占1200 token；工具定义五六个，加起来1800 token；用户每条消息平均200 token，工具返回结果平均1200 token。那么一轮"用户提问+工具调用+最终回答"大概消耗200+1200+600=2000 token。对于8K上下文窗口，扣掉系统提示和工具定义占用的3K，剩5K能用，最多容纳两三轮完整交互。到第四轮，要么直接截断，要么模型开始表现出"选择性地遗忘中间步骤"。
这种遗忘不是模型变笨，而是信息被挤出了有效注意力范围。更麻烦的是，你不知道哪一步会撞上截断边界。模型可能已经通过工具A拿到结果，接下来准备调用工具B时，工具A的上下文恰好被挤掉了，表现就是"有中间结果但模型已经看不见"，产生一种非常难排查的隐性bug。
成本问题也同样突出。不压缩的长会话，每一轮新请求都要把全部历史重发一遍，输入Token随轮次线性增长。10轮对话，不压缩和历史一起重发约30K token；而每隔几轮做一次Compaction，可以把发送量压到8K到10K左右。这在按Token计费的生产环境里不是小钱，所以Compaction不只是工程优化，它是长会话Agent能不能跑下去的基础设施。
| 模式 | 10轮累积发送输入Token | 费用趋势 |
|---|---|---|
| 不压缩 | 30K以上，每轮重复全部历史 | 随轮次线性增长 |
| 定期Compaction | 8K-10K，历史被摘要化 | 增速明显放缓 |
1.2 Compaction、Truncate、Summarization到底有什么不同
很多刚接触源码的开发者，会把Compaction直接理解成"把历史消息总结一下"。实际上在Pi-Agent这类项目的代码里，这三个动作是严格区分的。
Truncate是删除，直接从消息列表里丢掉一部分，最常见的是丢最老的、丢寒暄、丢工具中间日志。它的特点是速度快、不产生额外Token，但信息损失不可控。
Summarization是改写，用LLM把一段原始对话变成一段摘要。它能把几十条消息压成三五句话，但摘要的质量高度依赖上下文结构和提示词，而且执行时要额外消耗LLM调用。
Compaction是结构化重组。它通常会同时调用以上两个动作：删除该删的、摘要该概括的、并额外把关键实体、任务状态、未完成的工具调用单独抽出来存放。这才是完整意义上的"压缩"。
| 动作 | 操作方式 | 信息损失 | 典型源码方法 | 使用时机 |
|---|---|---|---|---|
| Truncate | 删除旧消息 | 不可控 | truncate_messages | 硬性水位溢出 |
| Summarization | LLM改写为摘要 | 可控但偏高 | summarize_history | 需要保留叙事线索 |
| Compaction | 删除+摘要+实体抽取+状态保留 | 可控且可审计 | compact_context | 长会话到达水位阈值 |
所以读源码时，不要只搜"compact"一个词，还要同时看"truncate_messages""summarize_history""extract_entities"几个入口。Compaction往往是外层调度，内部会把这套动作串成一个管线。这也是我解析这一篇时觉得最有价值的地方：不要把压缩当成"总结一下历史"，它本质上是一个决策流程，决定哪些消息直接删，哪些消息改写，哪些关键信息必须原样保留。
2. 源码里Compaction的触发与调度逻辑：水位线、时机与并发切换
2.1 Token水位线：软阈值、硬阈值与动态预算
源码里最直接的入口往往是一个Scheduler类，职责不是生成摘要，而是回答一个问题：现在该不该压缩。这类实现通常围绕软阈值和硬阈值展开。
class CompactionScheduler: def __init__(self, soft_ratio=0.7, hard_ratio=0.9, model_window=8192): self.soft_budget = int(model_window * soft_ratio) self.hard_budget = int(model_window * hard_ratio) def should_compact(self, current_tokens): if current_tokens >= self.hard_budget: return TriggerMode.IMMEDIATE if current_tokens >= self.soft_budget: return TriggerMode.BACKGROUND return TriggerMode.NONE硬阈值0.9意味着上下文占用达到窗口的九成时必须马上处理，否则下一个请求很可能直接溢出。软阈值0.7则是一个预警点，允许系统在空闲时进行后台压缩，而不是打断正在进行的对话。两个阈值之间这20%的余量，就是为了容纳用户正在输入的下一句话和即将产生的工具输出。
这里有一个很容易踩的坑：直接用模型最大上下文窗口去算，多半不准。生产环境里还需要预留输出Token，而且长文本存在注意力衰减，很多模型在接近最大窗口时效果明显下降。更稳的做法是引入一个可用窗口系数，把它从最大窗口里打折扣掉。
usable_window = int(model_window * 0.85) - reserved_output_tokensPi-Agent这类项目里一般会把reserved_output_tokens按模型组单独配置，比如预留1024或2048。读源码时看到这类参数，不要觉得是拍脑袋定的，它对应的是"我实际能安全塞进模型的内容量"。
2.2 执行时机：请求前、请求中还是请求后
触发条件定好之后，更麻烦的是什么时候执行压缩。在线Agent服务通常不能接受"每次请求前先压缩一遍"的做法，因为压缩长历史可能耗时几百毫秒甚至数秒，会直接拖慢首Token响应。
所以很多项目采用三段式："低水位标脏、空闲后压缩、下次请求前切换"。打个比方，不是每次饭前都把冰箱里的旧菜重炒一遍，而是看到快空了列个清单，晚上统一整理，第二天做饭时直接用整理好的食材。
Compaction放在请求后还有一个重要原因：必须等一个完整的交互回合结束。如果某个工具调用链还没收敛，比如Agent刚发出一个工具请求，正在等待结果，这时候去压缩历史，很容易把"等待结果"这个中间状态弄丢。源码里通常会有一个会话状态字段，比如is_closed或者last_activity_timestamp，只有会话进入完整回合的间隙，压缩线程才会真正启动。
请求前、请求中、请求后三种方案对比如下：
| 执行时机 | 优势 | 劣势 | 适用场景 |
|---|---|---|---|
| 请求前同步压缩 | 逻辑最简单 | 阻塞首Token响应 | 离线分析或短会话 |
| 请求中异步标记 | 不阻塞用户请求 | 并发切换复杂 | 在线低延迟Agent服务 |
| 请求后后台压缩 | 对在线服务最友好 | 需要双Buffer和版本管理 | 长会话长期运行 |
2.3 并发下的双Buffer与上下文版本
读源码时如果看到两个MessageBuffer，不用奇怪，那是为了不让压缩线程和推理线程打架。设计思路很直接：一个active buffer给正在处理推理请求的线程读，一个archive buffer给压缩线程写，压缩完成后原子替换指针。
class ContextStore: def __init__(self): self._lock = threading.RLock() self._active = MessageBuffer() self._archive = MessageBuffer() def replace_active_with_compact(self, messages, next_version): with self._lock: self._archive = MessageBuffer(messages, version=next_version) self._active, self._archive = self._archive, self._active为什么不能原地修改消息列表？因为在并发场景下，正在跑的推理请求可能已经迭代到一半，你突然把它的历史消息数组换了，轻则上下文错位，重则直接抛数组越界。双Buffer的语义是：旧请求可以继续用旧版本跑到结束，新请求落到新版本上。源码里通常还会有一个递增的context_version字段，排查问题的时候，对比两个请求的版本号是否一致，就能判断是否发生了"半路切换"。
这个设计在拆解源码时容易被当成无关细节，但实际上它是整个并发安全的基石。如果你在自研项目里做Compaction，我建议把这个双Buffer加身份版本的小结构直接抄走，它能帮你省掉大量偶发性的线上bug。
3. 核心数据结构与压缩策略拆解：从分块到摘要链
3.1 消息记录与压缩块：保证压缩后可回溯
Compaction之后，历史已经不是原始的逐条消息，而是若干压缩块。为了让后续检索和问题定位有据可依，消息记录和压缩块通常会分开建模，并通过ID关联。
@dataclass class MessageRecord: id: str role: str content: str token_count: int timestamp: float block_id: str | None = None meta: dict | None = None @dataclass class CompressedBlock: block_id: str summary: str entities: list token_usage: int original_msg_ids: list created_at: float这种结构最核心的一点是保留original_msg_ids。摘要只是给人看和给模型快速引用的，真正要回溯细节时，可以靠block_id映射回原始消息。这对线上问题排查极其重要：压缩之后用户投诉"模型答错了"，你得能查清楚某段结论到底是从哪几条原始消息里提炼出来的。
一个常见误区是压缩完成后把原始消息直接清空。这么做会省存储，但从可审计角度来说风险很大。稳妥做法是压缩块进热内存，原始消息落冷存储或者日志，至少保留一段时间。这样才能做到既能压Token，又能审计。
3.2 怎样分块才不会切断工具调用链
分块是Compaction里看起来最不起眼、实际上最影响质量的一步。分块太小，每块都需要一次LLM摘要，Token开销高；分块太大，单块摘要会丢失细节。实践经验里，单块原始内容在2048 token左右比较合适，压缩后的摘要控制在原始块的四分之一左右。
分块不能只看Token数，还要看语义边界。一个完整的工具调用链需要尽量落在同一个块内。工具调用、工具结果、模型基于结果的决策是一段因果链，如果被切到两个块里，第二块的摘要就可能缺失"之前拿到了什么结果"，后续推理断裂。
先按Token数滑动切分，再做一次事件边界补齐。典型计算方式如下：
def sliding_chunk(messages, chunk_token_size=2048, overlap=0.15): chunks = [] current = [] used_tokens = 0 for msg in messages: current.append(msg) used_tokens += msg.token_count if used_tokens >= chunk_token_size: # 事件边界修正：如果最后一条是工具结果，说明一条链路还没走完 while current and current[-1].role == "tool_result": next_msg = next_message(messages, current[-1]) if next_msg is None or current[-1].role in ("assistant", "user"): break current.append(next_msg) chunks.append(current) keep = max(1, int(len(current) * overlap)) current = current[-keep:] used_tokens = sum(m.token_count for m in current) if current: chunks.append(current) return chunksoverlap保留15%左右的尾部队列，表面上是多占了几个Token，实际是为了避免摘要边界处出现信息断带。这个参数不宜超过0.2，否则会出现下一块的目标人物、城市、任务编号反复被重复摘要，反而干扰模型。
事件边界修正这段逻辑，在源码里经常表现为一个is_event_complete(msg)的辅助函数。读源码时看到这类函数名，基本就能确认这个项目的分块策略做过语义层面的打磨，而不只是简单的数量切片。
3.3 摘要链：增量压缩与再摘要的区别
分块完成后，下一步是生成摘要。这里有两种流派：增量摘要和全局重摘要。
增量摘要的做法是把上一块的摘要作为输入，连同新块的原始内容一起交给LLM，产出新的摘要。这样每次只处理一个块，成本低、速度快，但如果旧摘要里有错，错误会一路传下去，越滚越偏。
全局重摘要则不依赖中间产物，每次都从原始消息列表重新生成。准确度最高，但每次都要重新读取大量数据，成本和延迟都不可接受。适合做离线归档，不适合做在线实时压缩。
生产级方案通常是混合模式：平时用增量摘要，每压缩固定次数做一次全量重算校正。比如每8次增量，强制一次全局重摘要，纠正累积失真。
| 策略 | 准确性 | 成本 | 延迟 | 适用场景 |
|---|---|---|---|---|
| 增量摘要 | 可能累积失真 | 低 | 低 | 在线实时Compaction |
| 全局重摘要 | 高 | 高 | 高 | 离线归档或周期校正 |
| 混合模式 | 高 | 中 | 中 | 长会话生产环境 |
解析源码时，搜索summary_mode这个开关，能很快判断这个项目属于哪一派。如果看到对old_summary的引用和拼接，那基本就是增量模式；如果看到每次都从原始MessageStore加载全部消息，那就是重算模式。这两种模式对后续实体的保留方式也不同，增量模式更需要靠实体抽屉兜底。
3.4 除了摘要还有实体抽屉与任务状态
Compaction做到后面，会暴露出一个本质问题：摘要存的是"发生了什么"，但模型在某些场景下只需要"某个事实是什么"或"事情进展到哪一步"。把这些精确信息全部埋进散文式摘要里，检索效率太低了。
所以成熟的实现里会有独立于摘要的结构化信息，我习惯叫它KeyInfo抽屉。
@dataclass class KeyInfo: entities: dict pending_tool_calls: list user_goal: str version: int举个例子。用户几天后回来问"上次说的那个城市是哪里？"如果这个城市名只存在于某一段摘要散文里，模型要在压缩后的长文本里自己翻；但如果封装时把它放进了entities字典，压缩后的Agent可以直接命中这个字面量，给用户答复。
这个设计也解释了为什么Compaction不是单一摘要，而是"叙事摘要+事实索引+状态机"三者并存。查看源码时，注意压缩器里是否同时出现compress_messages和compress_entities两个方法。如果只有前者，那么这个项目的Compaction还停留在"总结历史"阶段，后面大概率会碰到记忆丢失的问题。
4. 实战：手写一个迷你Compaction模块并跑通验证
4.1 需求定义：输入、输出和边界
理论讲完，落到能跑的代码。我拿一个迷你版Compaction模块做演示。它不代表Pi-Agent的真实源码，也不是逐行对应实际仓库，而是把源码里提炼出的核心思路重构成一个可独立运行的骨架。先定义接口：
def compact_context(messages: list[dict], budget_tokens: int) -> dict: # messages: 按时间升序的原始消息，含 role/content/token_count # budget_tokens: 压缩后允许占用的最大Token数 # 返回: {"messages": [...], "summary": "...", "entities": {...}, "stats": {...}}先明确边界条件：
- messages为空，或者原始总Token数已经小于预算，直接返回，不压缩。
- 最近一个完整回合必须原样保留，不允许压缩。因为那是对话正在发生的部分，是用户的实时语境。
- 如果摘要后仍然超预算，此时对摘要再做一次二级压缩，而不是继续去压原始消息。
- 压缩后的总Token必须小于压缩前，否则视为无效压缩，宁可保留原列表。
4.2 完整实现：分层压缩与实体保留
核心代码大概可以写成下面这样。我故意把函数拆成清晰段落，方便你对齐源码里的调度、分块、摘要、实体抽取四个环节。
def compact_context(messages: list[dict], budget_tokens: int) -> dict: total_before = sum(m.get("token_count", 0) for m in messages) if total_before <= budget_tokens: return {"messages": messages, "summary": "", "entities": {}, "stats": {"compact": False, "tokens_before": total_before, "tokens_after": total_before}} protect_round = 2 # 保护最近两个完整回合 compressible, protected = messages[:-protect_round], messages[-protect_round:] if not compressible: return {"messages": messages, "summary": "", "entities": {}, "stats": {"compact": False}} chunks = sliding_chunk(compressible, chunk_token_size=1024, overlap=0.15) summaries = [] entities = {} for chunk in chunks: summaries.append(summarize(chunk)) chunk_entities = extract_entities(chunk) entities = merge_entities(entities, chunk_entities) if estimate_tokens(summaries) + estimate_tokens(protected) > budget_tokens: summaries = [summarize(summaries)] # 二级再摘要 summary_messages = [{ "role": "summary", "content": " ".join(summaries), "token_count": estimate_tokens(summaries) }] new_messages = protected + summary_messages return { "messages": new_messages, "summary": summary_messages[0]["content"], "entities": entities, "stats": { "compact": True, "tokens_before": total_before, "tokens_after": estimate_tokens(new_messages), "ratio": round(estimate_tokens(new_messages) / total_before, 4) } }这里面summarize和extract_entities是大模型调用的抽象，你可以在实际项目里替换成自己的LLM封装。有一点要注意：这个迷你实现里实体抽取后没有真正把entities写回new_messages，真实集成时必须把它们注入系统提示词，否则entities就只是日志里的一堆好看数据，模型感知不到。
代码里的protect_round体现了前面说过的核心原则：用户正在感知的最近对话轮次，是这条对话的“当下”，压缩它等于让模型当场失忆。所以哪怕上下文再紧张，也要先保住最近一个完整回合。
4.3 压缩后用户感觉模型失忆的三个原因
这部分是我调试各种Agent后积累下来的高概率问题，按出现频率排序。
第一个原因是窗口划分把"刚刚用户说的话"也一起压缩了。比如用户问完"帮我查一下今天北京天气"，Agent还没来得及完整答复，压缩线程就把这轮对话塞进了历史块。等用户追问"那明天呢"，模型手里只有一条"用户问过天气"的摘要，不知道城市是北京。修复思路就是刚才代码里的protect_round，把最近完整回合设为不可压缩区。
第二个原因是工具调用链被摘要成了单条短语。原始上下文里能看清"weather_api(location=北京) -> 北京晴25℃ -> 模型回答今天晴"，压缩后变成"用户今天问过北京天气，已答复"。看起来没错，但模型后续需要知道调用过哪个工具、返回过什么字段。这就是KeyInfo抽屉里的pending_tool_calls应该保存的内容。工具的参数和最终结果只要不敏感，尽量结构化保留，不要扔进摘要散文。
第三个原因是关键实体没有重新注入系统提示。就算你在entities里存了city="北京"，如果压缩后的上下文里没有一行"用户近期关注城市：北京"，模型在续写时依然看不到。所以每次压缩完成后，要有一个环节把实体列表以纯文本方式拼到系统提示底部。这一步在源码里往往体现为build_system_prompt_with_entities。
这三个原因都指向同一个教训：Compaction之后，模型要能继续工作的所有前提条件，都必须显式存在于新的上下文里。隐式地"指望它从摘要中自己推断"，基本都会失败。
4.4 验证压缩效果：四个指标一个也不能少
压缩做得对不对，不能只看Token省了多少。我在项目里长期维护四个验证指标，缺一个都容易出问题。
| 指标 | 定义 | 理想范围 | 测试方法 |
|---|---|---|---|
| CompressionRatio | 压缩后Token / 压缩前Token | 0.3-0.6 | 从stats里直接读 |
| EntityRecall | 关键实体在压缩后的保留率 | 100% | 对比实体快照 |
| QuestionHitRate | 黄金问题集的答案正确率 | 与原上下文接近 | 压缩前后各问一遍 |
| ToolCallSurvivalRate | 工具调用链完整保留比例 | 100% | 检查pending_tool_calls |
开发环境里建议准备20个黄金问题，覆盖四种类型：跨多块的长期问题、实体点查询、工具链持续调用、上下文边界问题。把原上下文和压缩后的上下文分别输入模型，逐题对比答案一致性。这套东西只需要搭一次，后面每次修改压缩参数，跑一遍就能快速发现问题。
尤其要覆盖"跨多块"的长问题，因为摘要链的bug大多发生在信息跨越多个压缩块边界的时候。单块内的摘要通常问题不大，跨块之后，A块的结论和B块里的新事实被拼在一起，才容易出错。
5. 常见问题排查：压缩后失忆、耗时高与Token不减反增
5.1 压缩后模型反复追问已经给过的信息
表现是用户说"刚才不是给过你了吗"，模型还在问同样的问题。优先检查最近完整回合是否被正确保护，再检查实体抽屉里是否存了城市、时间、编号等信息，最后看这些实体是否真的被注入系统提示。
| 症状 | 可能原因 | 检查点 | 修复建议 |
|---|---|---|---|
| 追问已有实体 | 最近回合被压缩 | protect_round是否生效 | 保留最近完整回合 |
| 追问工具结果 | 工具返回没进实体抽屉 | pending_tool_calls字段 | 工具结果结构化保留 |
| 追问用户目标 | 用户目标未保存 | user_goal字段 | 对话开始时写入KeyInfo |
5.2 日志里Compaction耗时夸张
一次压缩耗时几百毫秒很常见，但如果经常超过好几秒，问题往往不在LLM摘要本身，而在锁等待。很多实现会用一个全局锁保护整个ContextStore，压缩线程长时间拿着锁，推理线程只能排队。
排查时可以给压缩过程打三段耗时：prepare_time、lock_wait_time、llm_time。如果llm_time不高但总耗时长，基本就是锁粒度太大。修复方向是把全局锁替换成按Block粒度的细锁，或者直接采用前面讲的双Buffer方案，让压缩线程根本不和推理线程抢同一个数据结构。
5.3 摘要重复与Token不减反增
压缩后Token不降反升，是代码集成阶段很容易出现的经典症状。原因通常是源码在压缩时把summary作为新消息追加到了列表尾部，却没有删除已经被摘要覆盖的原始块，导致消息数组里同时存在新旧两套内容。
排查方式很简单：检查压缩前后的消息数量。如果压缩后消息条数比压缩前还多，那就不是Compaction，是Append。正确做法是整体替换消息数组，而不是在尾部追加。压缩块和原始消息之间应该通过original_msg_ids引用关联，而不是同时存在于活跃上下文里。
摘要重复则是另一个相关症状，多出现在overlap设置过大，或者摘要提示词没有明确禁止复制上一块内容。把重叠度降到0.2以下，并在摘要提示词里写一句"不要重复上一块已出现的结论"，能明显减少这类问题。
我啃完这一段源码之后，最大的感受是：Compaction不是summary，它本质上是一个决策模块。它要决定哪些内容必须一字不动地活下来，哪些可以概括成几条摘要，哪些要被彻底忘掉。如果你准备在自己的智能体里做长期记忆，我的建议是不要一上来就上向量库。先把Compaction做好，把触发水位、分块粒度、实体抽屉和最近回合保护都跑通，很多记忆问题已经解决了一大半。
还有一个我实测很管用的小技巧：在开发环境里把每次压缩的输入、输出、实体列表全部打点存档。连续跑几次之后你会看到，绝大多数"失忆"都不是模型不行，而是压缩阶段把该留的细节丢了。排查问题时直接翻压缩记录，比反复对模型调提示词高效得多。这篇先聊到这里，后续有空我再把Compaction和向量检索、记忆回放结合的实现单独整理一篇出来。
