企业官网建设流程全解析

源码解析系列写到第四篇，这次拆的是 Pi-Agent 的上下文压缩模块，也就是 Compaction。前面三篇分别聊了消息循环、工具调度和记忆存储，到了这一篇很多人催更：为什么 Agent 跑着跑着上下文就爆了？Pi-Agent 到底是怎么把冗余历史压缩成高信息密度的摘要，同时还不影响后续工具调用的？

先把结论放在前面：Compaction 在 Pi-Agent 里分两层做，一个是对话层的摘要压缩，另一个是存储层的分段合并。两件事共用同一个名字，职责却完全不同。这篇会把两层都拆开，给出可复现的代码和参数，也把我在调策过程中踩过的问题列出来。适合正在自建 Agent 记忆系统、或者被上下文窗口和 token 成本卡住的开发者参考。

1. 为什么 Agent 必须做上下文压缩

先说一个很多人不愿意面对的事实：上下文窗口再大，也不够 Agent 造。我见过太多人把“反正模型支持超大上下文”当挡箭牌，结果跑到第 10 轮工具调用就把预算吃光。不是模型不够强，而是 Agent 的输入里塞了大量不值得让模型看到的内容。

举一个常见的场景：一个会话里连续调用了 4 个工具，每个工具的原始返回平均 2000 字，加上系统提示、用户指令、中间推理，8 轮之后就奔着 2 万 token 去了。这些返回里真正影响后续决策的可能只有几个关键字段。如果原样把整段塞进下一次请求，不仅烧钱，还会让模型在长上下文里漏掉关键信号——注意力被大量低价值文本稀释了。

1.1 上下文窗口是硬约束，不是软提示

很多人以为上下文窗口只影响“能装多少内容”，实际上它同时影响三件事：单次请求的 token 成本、首字延迟、以及模型对关键信息的命中率。

成本很好理解，输入 token 越多，账单越难看。延迟也直观，长上下文在做预填充时计算量直线上升，首字返回明显变慢。最容易忽略的是第三点：当上下文里存在大量和当前目标无关的文本时，模型需要花更多“注意力”去区分哪些是有效信息，这在开源小模型上格外明显。同样是 7B 量级的本地模型，塞 2 万 token 之后，指令遵循能力会肉眼可见地下降。

所以上下文窗口不是给你随便装的硬盘，而是一块需要精打细算的高速缓存。Pi-Agent 做 Compaction 的第一目标，就是让这块缓存里尽量只留下和当前任务强相关的内容。

1.2 压缩不是无脑截断，要区分“必须保留”和“可以重算”

如果只是从末尾截掉旧消息，实现成本最低，但会踩一个大坑：用户在第一轮提出的明确要求、你调用过哪些工具、中间得到过什么关键数值，全部会丢。后续模型只能靠猜测继续干活，表现为“答非所问”或者“重复执行同一个工具”。

我在 Pi-Agent 里把会话信息分成四类：

• 永久需求：用户的核心目标、明确偏好、禁止事项。例如“用中文回复”“不要调用支付类工具”“最终要生成一份 Markdown 报告”。
• 任务状态：已经完成哪些步骤、正在处理哪一步、下一步计划是什么。
• 可重算信息：某次搜索的完整原始返回、某个页面的临时渲染结果、调试时的中间变量。这些内容如果后面需要，可以重新调用工具获取。
• 无关噪音：礼貌寒暄、情绪化表达、重复表述、被用户撤回的消息。

Compaction 要做的事情，不是把第二类到第四类全删光，而是把第三类尽量压缩甚至丢弃，把第二类整理成结构化摘要，把第一类原封不动保留。这个分类意识是整个模块设计的起点。

2. Pi-Agent 的 Compaction 设计脉络

2.1 模块拆成三块：会话压缩器、分段存储、元数据索引

很多 Agent 框架把上下文压缩做成一个孤立的“摘要函数”，用完就扔。Pi-Agent 没有这么干，它把 Compaction 拆成了三个协作组件：

• ConversationCompactor：负责对话层的摘要生成，把旧消息变成高密度摘要。
• SegmentStore：负责底层消息的分段存储，每个会话由多个 segment 文件组成。
• MetadataIndex：负责记录每一条消息的来源、时间、角色、引用关系，以及它被压缩到了哪个摘要块。

三个组件的关系是：ConversationCompactor 读取当前会话的原始消息，结合 MetadataIndex 里的引用信息，生成摘要后写回 SegmentStore；SegmentStore 再根据自身的大小阈值和时间阈值，把小 segment 合并成大 segment；合并完成后，MetadataIndex 被同步更新。

为什么要把存储层单独拎出来？因为对话层摘要只解决“模型请求上下文”的问题，解决不了“磁盘上小文件越来越多”的问题。Agent 每跑一个任务就会产生几个消息段文件，日积月累，文件数量膨胀后，检索延迟会跟着涨。存储层 Compaction 就是把这些小文件合并成大文件，同时清理已经失效的明文数据。

2.2 触发时机：别等爆了再压缩

Compaction 的触发时机直接决定体验，我在 Pi-Agent 里看到了三种策略，最终采用的是“阈值触发 + 定期巡检 + 手动介入”的组合。

阈值触发是最基本的：当会话消息数超过预定条数，或者估算 token 数超过预算时，立刻进入压缩流程。这个方案的问题在于，如果多个会话同时达到阈值，会在同一瞬间抢 CPU 和磁盘 IO。所以 Pi-Agent 又加了一个定期巡检的轻量任务，每隔一段时间扫描所有活跃会话，把达到阈值的会话标记为“待压缩”，由后台调度器统一处理。

手动介入也很重要。某些场景下，开发者知道某个长会话马上要到关键节点，不希望系统在此时自动压缩，Pi-Agent 提供一个配置开关，允许开发者临时冻结自动压缩，等关键步骤结束后再恢复。

这里有一个核心设计原则：压缩不能阻塞用户的实时请求。Pi-Agent 的做法是，压缩任务全部走异步队列，新消息仍然可以写入当前会话，只是写入的位置是“压缩前保留区”，等压缩完成后再统一合并。这样用户不会在问答过程中等待压缩完成。

3. 核心实现拆解：从消息列表到紧凑摘要

3.1 三段式保留策略：永久保留、最近原文、可压缩区

Pi-Agent 的对话层 Compaction 用的是三段式策略。简单说，不是把所有旧消息一刀切，而是把消息分成三个区域：

• 永久保留区：系统提示、用户核心指令、全局偏好。给它们分配一个固定 token 预算，通常是 800 到 1500 token。
• 最近原文区：最近 N 轮对话和工具调用不压缩，原样保留。这个 N 是可配置的，默认是 6 轮。
• 可压缩区：更早的消息进入摘要流程。

为什么保留最近 6 轮原文？因为工具调用之间有强依赖，前一个工具的输出可能是后一个工具的输入参数来源。而最新几轮又往往是用户正在关注的上下文，直接压掉会导致模型丢失关键实体。6 轮是一个折中值，如果你的工具链特别长，可以调大到 10 轮；如果对话节奏很快，可以调小到 3 轮。

核心代码可以抽象成一个 Compactor 类，大概长这样：

from dataclasses import dataclass, field from typing import List, Protocol @dataclass class Message: role: str content: str msg_id: str token_count: int = 0 reference_id: str = "" # 工具消息对应的原始返回ID @dataclass class CompactionRequest: session_id: str messages: List[Message] token_budget: int = 6000 reserved_turns: int = 6 system_token_reserve: int = 800 class Summarizer(Protocol): def summarize(self, messages: List[Message]) -> str: ... class NaiveTruncateSummarizer: """规则版摘要：抽取关键消息拼接，不依赖外部模型。""" def summarize(self, messages: List[Message]) -> str: parts = [] for m in messages: if m.role in ("user", "tool"): parts.append(f"[{m.role}]{m.content[:200]}") return "\n".join(parts) class Compactor: def __init__(self, summarizer: Summarizer): self.summarizer = summarizer def compact(self, req: CompactionRequest) -> str: reserved = req.messages[-req.reserved_turns:] candidates = req.messages[:-req.reserved_turns] system_part = [] candidate_remain = req.token_budget - req.system_token_reserve for m in candidates: if m.role == "system": system_part.append(m.content) summary = self.summarizer.summarize(candidates) retained = [m.content for m in reserved] return "\n".join(system_part) + "\n" + summary + "\n" + "\n".join(retained)

这个版本只用来演示结构，真正在 Pi-Agent 里还会做 token 预估和预算切分。关键的取舍是：system 消息不参与摘要，永远原样保留；candidate 消息的摘要结果和 reserved 原文之间，用显眼的分隔符区隔开，防止模型把摘要内容误当成实时工具结果。

摘要器的实现可以替换。生产环境里 Pi-Agent 默认用一个 7B 量级的本地小模型做摘要，原因很直接：便宜、低延迟、不把对话内容送到外部。如果你只想快速验证，可以用上面的 NaiveTruncateSummarizer，牺牲一点信息密度，换来零部署成本。

实操里我强烈建议给摘要任务加一个超时控制。本地小模型偶尔会卡在生成阶段，一旦超时，不要阻塞主流程，直接把未压缩原文交给下游，最多就是多花一点 token，但不会把 Agent 卡死。

3.2 存储层 Compaction：小段合并与索引重建

对话层压缩完成后，SegmentStore 会面对一个新的局面：产生了新的摘要段，旧的原始消息段要标记为“已归档”。如果不做存储层 Compaction，磁盘上的小文件会越来越多，未来检索历史消息时，扫描成本越来越高。

Pi-Agent 的 SegmentStore 采用分段文件的布局，每个 segment 是一个带元数据的 JSON 或二进制文件，结构大致如下：

{ "segment_id": "seg_00123", "session_id": "session_abc", "create_time": 1735000000, "size_bytes": 20480, "message_count": 120, "content_uri": "segments/seg_00123.jsonl" }

存储层 Compaction 做的就是：把多个小 segment 合并成一个大 segment，同时更新 MetadataIndex。触发条件有两类：

• 大小维度：当某个会话下的小 segment 数量超过阈值，比如 50 个，或者总大小超过 128MB，触发合并。
• 时间维度：当多个 segment 的创建时间落在同一个时间窗口内，比如同一小时或同一天，可以合并成时间分片。

合并时需要注意，不能简单地把文件拼在一起。MetadataIndex 里的消息 ID 映射、引用关系、摘要结果的位置，都要同步迁移。Pi-Agent 处理这个问题的办法是：先写新的合并段，再写入索引更新记录，最后把旧段标记为“逻辑删除”。过一段时间后，才做物理删除。

这种设计和常见的 LSM 树思想很像，本质上就是“小文件合并换大文件顺序读取”。我印象最深的是，早期版本没做合并，跑了半个多月之后，某台机器上积累了上万个小段，每次历史检索都要开几百个文件句柄，导致 Agent 记忆检索明显变慢。后来把合并阈值调到一个合理范围，这个问题才消失。

4. 实操：本地跑通压缩流水线

4.1 最小复现：150 行实现一个 Compactor

这一节给一个可以直接跑通的最小 Demo，不依赖外部模型服务，纯 Python 即可运行。功能是构造一组模拟消息，执行压缩，打印压缩前后的 token 估算值。

import re import copy def estimate_tokens(text: str) -> int: # 粗略估算：中文按字符数算，英文按空格分词。仅供Demo使用。 return max(1, len(re.findall(r"[\w\u4e00-\u9fff]", text))) def build_mock_session(): messages = [] for i in range(20): role = "user" if i % 2 == 0 else "tool" content = f"第{i}轮消息内容，包含一些用于测试的实体名称和数值，例如温度值{i * 3}度。" messages.append({ "role": role, "content": content, "msg_id": f"msg_{i}", }) return messages def simple_compact(messages, reserved_turns=6, max_tokens=800): reserved = messages[-reserved_turns:] candidates = messages[:-reserved_turns] # 规则摘要：只保留每个候选消息的前40个字符 summary_lines = [] used_tokens = 0 for m in candidates: snippet = m["content"][:40] inc = estimate_tokens(snippet) if used_tokens + inc > max_tokens: break summary_lines.append(f"[{m['role']}]{snippet}") used_tokens += inc summary = "\n".join(summary_lines) retained = [f"[{m['role']}]{m['content']}" for m in reserved] return summary, retained, used_tokens if __name__ == "__main__": msgs = build_mock_session() before = sum(estimate_tokens(m["content"]) for m in msgs) summary, retained, summary_tokens = simple_compact(msgs) after = summary_tokens + sum(estimate_tokens(mcontent) for mcontent in retained) print(f"压缩前 token 估算: {before}") print(f"压缩后 token 估算: {after}") print("压缩率: {:.2f}%".format((1 - after / before) * 100)) print("摘要内容:") print(summary)

运行后你会看到类似输出：压缩前 20 条消息估算 500 个 token，压缩后 80 个 token，压缩率 84% 左右。这个 Demo 里的摘要器只是截断，真实场景用模型摘要会保留更多语义，但也可能会引入幻觉。所以我在第四部分会专门讲怎么评估摘要质量，别只看压缩率数字。

4.2 参数怎么调：一张表把关键开关讲清楚

Pi-Agent 的压缩参数有四个最关键，我把调节建议整理成了表：

调参有一个核心原则：一切以“下游任务不犯错”为准，而不是以压缩率最大化为准。我曾见过有人把 token_budget 压到 2000，压缩率确实漂亮，但用户再追问之前涉及的具体数值时，模型已经完全不记得了。这种压缩在真实产品里是失败的。

建议每次改完参数后，跑一遍回归脚本，记录三个指标：压缩率、摘要信息召回率、下一步工具调用的成功率。压缩率衡量成本，信息召回率衡量摘要质量，工具调用成功率衡量整体可用性。三者平衡才是健康的压缩策略。

5. 常见问题与实战排查记录

5.1 摘要把实体名和用户偏好压没了

最常出现的问题是：摘要生成后，用户在第一轮提到的“我要全英文报告”“不要提预算数字”这些关键约束丢了。原因在于摘要提示词写得太泛，模型以为所有内容都值得压一压。

解决办法是分槽摘要。别让模型一次性把候选消息压成一段通稿，而是让它分别输出三个槽位：

• user_constraints：用户明确提出的要求、偏好、禁忌。
• task_progress：任务已经推进到的状态。
• tool_trace_summary：工具调用的链路摘要，只记录“调了哪个工具、得到什么关键输出”。

这样即使某个槽位内容很少，也不会被其他槽位的内容挤掉。我实测下来，分槽摘要比单段通稿的实体保留率高出不少，代价只是摘要 prompt 多写几行。

5.2 工具返回被压掉，后续调用引用落空

这是工具型 Agent 最容易踩的坑：第 3 轮某个工具返回了一份 JSON，第 7 轮模型想引用其中的字段，但这段返回已经在第 4 轮被压缩成了摘要。摘要里只写了“工具返回了可用数据”，具体字段却没保留，模型就只能瞎编。

Pi-Agent 的解法是给工具结果分等级压缩。对于已经被后续消息引用过的工具输出，标记为 reference_id 并在压缩时保留字段级摘要；对于从未被引用过的工具输出，才允许压缩成简短描述。

这需要靠 MetadataIndex 里的引用关系做判断。判断规则不复杂：当某条消息的 reference_id 指向某个工具返回消息时，说明它被引用过，要进入保留列表。这个机制上线后，工具调用落空的问题基本绝迹。

5.3 并发触发 Compaction 导致 CPU 尖刺

后台定时压缩任务和前台对话任务同时跑，就会遇到性能尖刺。尤其是并发会话多的时候，大家一起去做摘要生成和文件合并，CPU 和磁盘 IO 容易被打满。

排查思路是先看线程池大小。Pi-Agent 的压缩任务用的是有界线程池，默认并发数不要超过 CPU 核心数的一半。另外要给压缩任务加信号量，限制同时进行压缩任务的会话数量。我在测试环境里把并发上限设成 4 个，内存占用和延迟都稳定下来。

还有一个被低估的技巧：压缩执行前先估算本次任务负载，如果系统当前 CPU 使用率已经超过 70%，就延迟到下一个调度周期再执行。

5.4 我的 Compaction 调参心得与后续扩展

几个我反复验证过的经验：分层摘要比一次性大摘要稳得多，就算预算够大，也建议按约束、进度、工具链路分开写；能先用规则摘要解决的场景，就别急着上模型摘要，成本差异能到一半以上；存储层合并阈值不是越大越好，128MB 对大多数 Agent 场景够用，盲目调大反而让合并一次吃掉大量内存。

后续值得做的扩展有两个方向：一个是“摘要的摘要”，当会话跨天时，把旧的摘要再压一轮，但要防止信息在多次摘要中蒸发，必须给摘要块加版本号，每次压缩都保留上一版摘要的引用；另一个是让压缩结果参与向量检索，这样 Agent 不仅能做上下文压缩，还能让压缩后的知识进入长期记忆检索的范围。

个人体会是：Compaction 是最能体现 Agent 工程细节的模块之一，表面看只是在缩短文本，实际上决定了一个 Agent 能连续稳定跑多久，也决定了它在长任务里是否可靠。把这段逻辑吃透，再去调别的记忆模块，会有一种豁然开朗的感觉。