企业官网建设流程全解析

1. Pi-Agent 从“记不住”到“压缩再上阵”的设计演变

我维护 Pi-Agent 这个项目有段时间了，最早版本里用的是最粗暴的方案：对话历史超出窗口就一把梭，把最早的消息直接丢掉。结果实测下来问题很大——用户早上问过的需求、之前定好的偏好、甚至工具调用的中间结果，一旦被裁剪，Agent 后续行为就开始“失忆”。后来才下定决心把上下文压缩 Compaction 做成正式模块，这篇就当系列第四篇，专门拆这块。

先给没接触过的朋友交代一下背景：Pi-Agent 是一个基于大语言模型的多轮对话 Agent 项目，核心逻辑是维护一个可以持续增长的会话上下文列表，每次请求前把所有历史消息拼进 Prompt。长对话场景下，上下文会迅速逼近模型窗口上限。Compaction 要解决的问题很具体——在窗口还不够用之前，把已有对话压缩成更精炼的表示，腾出空间给新消息，同时尽量保留对后续决策有用的信息。

这篇源码解析的目标读者分两类：一类是想给自家 Agent 加内存管理能力的开发者，另一类是已经在用现成框架，但想知道底层“摘要+裁剪”逻辑的人。我会直接贴 Pi-Agent 项目里的核心类设计，说明每一步在做什么、为什么这么做，以及我在实际调参中踩过的坑。

如果你只是把上下文截断到固定长度，那其实不用看 Compaction。Compaction 的复杂之处在于“压缩后还必须能支撑后续对话”这件事，它既要做信息降维，又不能破坏 Agent 的连贯性。下面从架构边界开始讲。

2. Compaction 模块的边界与触发链路

2.1 模块在 Pi-Agent 消息流水线中的位置

Pi-Agent 的消息处理流程大体上是：用户输入 -> 会话管理器 -> 上下文收集器 -> LLM 调用 -> 工具执行 -> 结果回填。Compaction 被设计成一个可插拔的中间件，挂在上下文收集器和 LLM 调用之间。

# pi_agent/pipeline.py 中的简化调用链 class MessagePipeline: def __init__(self): self.compactor = CompactionManager( max_context_tokens=12000, min_available_tokens=2000, summary_model="compact-summary", ) def process(self, user_message, session): session.add_message("user", user_message) context = self.compactor.build_compact_context(session) response = self.llm_client.chat(context) session.add_message("assistant", response) return response

放在这个位置的用意很明确：每次构建 Prompt 前都要检查当前上下文是否“超标”，而不是由用户主动触发。这样对上层调用方透明，Agent 的编排逻辑不需要关心压缩细节，只需调用同一个build_compact_context方法。

2.2 CompactionManager 的核心状态与职责

CompactionManager内部维护了三块关键状态：原始消息序列、压缩中间产物、压缩后的结构化摘要。它不是一个一次性工具函数，而是一个有生命周期状态的管理器，因为压缩需要对同一会话持续负责。

# pi_agent/compaction/manager.py @dataclass class CompactionState: original_messages: list[Message] # 压缩前的完整消息历史 compacted_records: list[CompactionRecord] # 每次压缩产生的记录 last_compaction_token: int = 0 # 上次压缩时的上下文 token 总量 compressed_at_round: int = 0 # 上次压缩发生在第几轮对话 class CompactionManager: def __init__(self, max_context_tokens, min_available_tokens, summary_model): self.max_context_tokens = max_context_tokens self.min_available_tokens = min_available_tokens self.summary_model = summary_model self._state: dict[str, CompactionState] = {}

我特意让每个 session 独立维护状态，避免不同对话相互污染。后来踩坑时发现，如果不把compacted_records留档，一旦压缩完就丢弃原始消息，出问题时根本没法回溯到底丢了什么。所以源码里第二个版本开始就保留了原始消息引用，只在确认压缩产出合理后才允许 GC 清理，这在后面排查问题时会讲到。

2.3 触发条件：不只看总量，还要看余量

触发压缩最直觉的做法是“总 tokens 超过 N 就压缩”，但这样会有一个边界抖动问题：刚压缩完，下一轮对话又超了，再压缩，如此反复。Pi-Agent 的实际判断逻辑用了两个阈值：max_context_tokens和min_available_tokens。

# pi_agent/compaction/triggers.py def should_compact(self, session_id: str, estimated_tokens: int) -> bool: state = self._state[session_id] available = self.max_context_tokens - estimated_tokens if estimated_tokens >= self.max_context_tokens: return True # 即使总量没超，但剩余空间不足于支撑下一轮可能的大输出，也要提前压缩 if available < self.min_available_tokens: return True # 如果距上次压缩不足 3 轮，强制抑制，避免反复压缩 current_round = self._get_current_round(session_id) if current_round - state.compressed_at_round < 3: return False return False

之所以留min_available_tokens这个余量判断，是因为大模型的输出 token 数往往比输入更不可控。如果上下文已经塞到只剩 200 token，LLM 生成一个小工具调用都可能被截断，所以不能等满了才压缩。强制抑制的逻辑也很关键：压缩本身有成本，摘要模型也要算 token，频繁压缩得不偿失。

2.4 Token 估算的实现细节

触发判断依赖 token 估算，Pi-Agent 没有直接调商用模型的计数接口，而是用了一个轻量级的分词器近似计算。这部分源码看起来简单，但容易写错。

# pi_agent/tokenizer/approx.py def estimate_tokens(messages: list[Message]) -> int: total = 0 for msg in messages: total += 4 # 每条消息的角色/元数据开销 total += len(msg.content) * 0.6 # 英文约 0.6 token/字符，中文约 0.8 if msg.tool_calls: total += sum(len(str(c)) * 0.5 for c in msg.tool_calls) return int(total)

这里用乘系数而非精确分词，是为了在每次用户输入时都能毫秒级完成估算。后来我发现不同模型的分词差异很大，干脆把系数做成了可配置项，不同模型走不同配置。当然，生产环境如果对预算要求严格，定期用真实 tokenizer 校准一下是必要的。

3. Compaction 核心逻辑：摘要不是简单“总结一段话”

3.1 消息分组与字段筛选

一开始我把整个对话历史一股脑丢给摘要模型，让它输出一段总结，效果很差。因为对话里混着用户指令、系统状态、工具调用、Assistant 思考过程，它们的信息密集度和重要程度完全不同。Pi-Agent 的压缩逻辑第一步是把消息按“轮次”分组，一个轮次包含：用户输入、Agent 的思考/回复、可能的工具调用链、工具返回结果。

# pi_agent/compaction/grouping.py def group_by_round(messages: list[Message]) -> list[ConversationRound]: rounds: list[ConversationRound] = [] current_round: list[Message] = [] for msg in messages: if msg.role == "user" and current_round: rounds.append(ConversationRound(messages=current_round)) current_round = [msg] else: current_round.append(msg) if current_round: rounds.append(ConversationRound(messages=current_round)) return rounds

分组后，每个ConversationRound内部还要做字段级别的提取，而不是把完整 Message 塞给摘要模型。像tool_calls里的参数可能很冗长，但真正要保留的是“调了哪个工具、输入的关键参数、返回结果的成功与否”。源码里对每个字段设置了保留策略。

3.2 分级摘要：轮次摘要 + 全局摘要

单层摘要往往会顾此失彼。Pi-Agent 的机制是两层：先对每个轮次做一次短摘要（round summary），然后在这些轮次摘要的基础上，再生成一个全局摘要（global summary）。

# pi_agent/compaction/summarizer.py class HierarchicalSummarizer: def summarize_round(self, round_msgs: list[Message]) -> RoundSummary: prompt = self._build_round_prompt(round_msgs) result = self.summary_model.complete(prompt) return RoundSummary( round_id=round_msgs[0].round_id, user_intent=result["user_intent"], key_facts=result["key_facts"], tool_actions=result["tool_actions"], result_status=result["result_status"], ) def summarize_global(self, round_summaries: list[RoundSummary]) -> GlobalSummary: text = "\n".join([r.to_compact_text() for r in round_summaries]) prompt = f"根据以下对话轮次摘要，生成一条全局摘要，包含用户长期目标和已确认的决策：\n{text}" return GlobalSummary(...)

为什么要分层？因为全局摘要只看局部摘要时，是拿不到原始细节的，所以局部摘要必须保留结构化的字段，而不是一段散文。我在实现时踩过坑：第一版让摘要模型输出“自然语言概述”，结果后续问答中用户问起某个具体数字，模型完全答不上来。后来把 CompactionRecord 改成结构化字段，数字、名字、时间这些关键实体单独列出，存活率高了很多。

3.3 工具调用链的压缩策略

工具调用是 Agent 最容易失忆的部分。比如用户让 Agent 查询天气并订机票，过程中调了天气 API、航班 API、支付接口，如果压缩时只记录“用户订了机票”，后续用户问“上次航班是几点”就废了。Pi-Agent 对工具调用采用“保留签名 + 压缩参数摘要”的方式。

# pi_agent/compaction/record.py @dataclass class ToolActionRecord: tool_name: str input_summary: str # 参数摘要，保留目标值与关键 ID output_status: str # success / failed output_key_data: list[str] # 如航班号、订单号、金额等

具体实现上，tool_calls中的原始参数不直接进摘要，而是经过一层summarize_tool_input，它把参数里的大段文本（比如查询条件文案）压缩成核心实体，但把标识符（比如订单号、ID）原样保留。这个设计后来被证明极其重要——用户回访类问题的命中率，几乎都靠这些标识符撑着。

3.4 CompactionRecord 存储格式

每一次压缩都会产出一条CompactionRecord，它是之后上下文重建的“原材料”。

# pi_agent/compaction/record.py @dataclass class CompactionRecord: version: int = 1 compressed_round: int global_summary: GlobalSummary round_summaries: list[RoundSummary] preserved_messages: list[Message] # 最新且未压缩的原始消息 dropped_message_ids: list[int]

我在 record 里特意加上了dropped_message_ids，不是为了记仇，而是为了可审计性。每次压缩后，如果后续回答质量下降，可以快速定位是哪些原始消息被压掉了，然后调整压缩策略。这比把日志打成一团乱麻再人肉翻找高效得多。

4. 上下文重建：怎么把压缩结果拼回可用的 Prompt

4.1 重建的组装顺序

压缩结果不能直接展示给用户看，而是要“伪装”成正常的对话历史喂给 LLM。Pi-Agent 的build_compact_context方法负责这件事。

# pi_agent/compaction/context_builder.py def build_compact_context(self, session_id: str) -> list[Message]: state = self._state[session_id] records = state.compacted_records preserved = state.preserved_messages system_prompt = self._restore_system_prompt(state) compact_messages = [system_prompt] for record in records: compact_messages.append({ "role": "user", "content": f"[历史会话摘要]\n{record.global_summary.to_text()}", }) compact_messages.append({ "role": "assistant", "content": f"[已根据摘要确认，继续等待用户指令]", }) compact_messages.extend(preserved) return compact_messages

这里的技巧是：不把摘要塞进 system prompt，而是模拟成“用户曾经说过的话 + 助手确认”的对话轮次。这样 LLM 能理解这些是历史信息，而不是当前指令，同时又有足够的表现力去引用摘要里的内容。我试验过直接把摘要塞进 system prompt，模型会倾向于在回答开头复述摘要，反而干扰当前问题。

4.2 系统提示与长期指令保真还原

Pi-Agent 的系统提示会随时间变化，比如用户开启了新技能、或者管理员更新了行为准则。压缩时必须把当前生效的系统提示和压缩时历史版本做区分。

def _restore_system_prompt(self, state): # 从会话元数据中取当前系统提示，覆盖压缩时提示 latest_system_prompt = state.session_meta.latest_system_prompt if state.session_meta.system_prompt_versions: return latest_system_prompt

但只是取最新系统提示还不够——如果用户在早期对话中确认过某个特殊偏好，而该偏好不在系统提示中，就容易被摘要漏掉。所以Restore过程中会扫描全局摘要里是否有user_confirmed_preferences字段，有的话以附加指令的方式拼到系统提示后，确保用户偏好不被覆盖。

4.3 保持时间线与因果顺序

摘要轮次的顺序不能乱。LLM 没有“时间线”概念，如果你把后面的轮次摘要放到前面，它会把因果关系理解反。这里的保持逻辑其实很朴素，就是按round_id升序排列，但有一个隐蔽问题：工具调用的结果可能影响下一轮用户输入的理解。比如用户说“就按刚才那个价格买”，这个“刚才”指的是压缩之前的对话。所以摘要里的key_facts必须包含能支撑指代的实体，重建时再通过preserved_messages里保留最近几轮原文来兜底。

# 重建时，保留最新 N 轮不压缩，防止指代断裂 KEEP_RECENT_ROUNDS = 4

这个KEEP_RECENT_ROUNDS是调参调出来的经验值。保留得太少，指代容易断；保留得太多，压缩效果下降。对于大多数对话型 Agent，4 轮是一个比较平衡的点。

4.4 容易丢失的隐式状态

源码解析到这里，我想特别提一个非显性字段：情绪和态度。Agent 在某些场景下需要感知用户情绪变化，比如用户从平静变得不耐烦。压缩摘要时一般只记录事实，情绪这种隐式状态很容易丢。Pi-Agent 早期的压缩完全不考虑情绪，后来客服场景的用户反馈说 Agent 变得“冷漠”，才开始在 RoundSummary 里加user_sentiment字段，维度包括neutral/positive/urgent/angry。这个字段在重建时只作为辅助信息，不参与决策，但会让后续回复的语气更对味。

5. 实测效果与参数调优经验

5.1 压测数据：压缩比与触发轮次

我在一组模拟客服对话数据上做了压测，每个 session 固定 50 轮，模型上下文窗口设为 16K。不启用压缩时，第 40 轮左右就会超限报错；启用压缩后，全程未超限，平均压缩比大概在 4.2：1，也就是 10000 token 的原始历史可以压到 2300 左右。

这里的“用户意图保持率”是人工评估的：在压缩后的会话中继续提问“我刚才想干什么”，能正确回答的比例。分层结构化摘要明显优于单层摘要，原因就是单层摘要把关键细节揉碎在长文本里，模型检索不到。

5.2 摘要质量损失的观测指标

压缩不是无损的，我们要接受损失，但要能度量损失。Pi-Agent 在系统里埋了两个观测指标：指代召回率和实体准确率。

指代召回率指代的是——在压缩后的上下文里，LLM 能否正确识别“它”“那个订单”等指代对象。实体准确率则是摘要中提取的人名、时间、数字是否正确。我建议所有做 Compaction 的人都把这两项加进日志，否则你根本不知道升级策略后是变好还是变差。

5.3 不同模型对压缩后上下文的敏感度

这里有一个容易忽略的事实：摘要模型和主对话模型不一定是同一个。Pi-Agent 默认允许独立配置summary_model，实测下来，摘要模型能力弱，压缩质量就明显差；但即使摘要模型很强，主模型如果对“历史摘要”这种伪对话格式理解能力差，效果也会打折扣。

我用同一条摘要分别喂给强推理模型和轻量模型做了对比：

• 强推理模型：能主动引用摘要里的事实，并且区分“摘要历史”和“当前对话”。
• 轻量模型：容易把摘要里的内容误认为当前系统指令，回答时带上“根据历史摘要”之类的多余前缀。

所以选型时要考虑主模型的指令遵循能力，不能只盯着摘要模型。如果主模型较弱，[历史会话摘要]的提示词可能需要改成更明确的“以下是你之前与用户的对话记录，仅供回顾”之类的描述。

5.4 三个参数的联动调节

max_context_tokens、min_available_tokens、KEEP_RECENT_ROUNDS是三个要联动调整的参数。

我的经验是：

• 如果模型上下文是 8K，max_context_tokens设置为 6K，min_available_tokens设为 1K，给输出留足余量。
• 如果对话类型是“任务型”的，比如用户一次性提出需求，执行完就结束，那么KEEP_RECENT_ROUNDS可以小一点，比如 2，因为后续指代少。
• 如果对话类型是“顾问型”的，用户会反复追问、修正，那么KEEP_RECENT_ROUNDS最好调到 5~6，避免最近几轮的关键修正信息被摘要抹平。

还有一个隐藏参数是摘要提示词里的“详细程度”。我发现控制这个参数比调整保留轮数更有效。把摘要提示词从“简要总结”改为“保留所有涉及数字、名称、偏好的细节”，压缩比会稍微下降一点（4.2 变 3.5），但实体准确率从 78% 升到 91%，非常值得。

6. Compaction 的坑、排查与进一步扩展

6.1 工具调用序列被压碎导致的“幽灵操作”

这是我调试过程中最头疼的一个 bug：压缩后的 Agent 在用户没有发出新指令时，突然重复执行某个历史工具调用。排查过程一步步走：

• 第一步，看触发记录：发现dropped_message_ids里丢了某次工具调用的“成功标志”。
• 第二步，进入摘要原文：tool_actions里只记了“调用 flight_search”，没有记录参数和结果。
• 第三步，重建后的伪对话里，LLM 误以为那个调用还未完成，于是主动补完成动作。

这个问题的根因在于摘要模型将“工具调用过程”和“工具调用结论”混在一起，模型分不清已完成和未完成。修复方式是在ToolActionRecord中把result_status单独取出，重建时在摘要文本前明确加一句“以下工具调用均已完成”：

status_line = "所有记录的工具调用均已完全结束并返回结果。\n" if all_success else "注意：存在未完成的工具调用：...\n"

这个补充让 LLM 不再把历史摘要里的调用当成待办事项。如果你也遇到 Agent 在压缩后自作主张调用工具，大概率是这里出了问题。

6.2 用户偏好随摘要漂移

另一个常见问题是用户的个性化要求被摘要“平均化”。例如用户三周前说过“回复尽量简短”，但摘要里写的是“用户偏好简洁风格”，重建后模型对“简短”的理解会漂移，渐渐变成正常长度。

解决办法是在摘要中把用户偏好单独划区，并使用原文引用：

preferences = [ {"source_round": 12, "original_quote": "回复不要超过两句话", "deduced_pref": "简洁"}, ]

重建时，将original_quote原样放进上下文，而不是只放推论。这样模型看到的是用户的原始表达，风格还原度高很多。这条经验非常适合做“个性化助手”方向的开发者。

6.3 日志与可观测性建设：压缩也要能回溯

我的建议是：每次压缩都往日志系统写一条compaction_event，至少包含压缩前 token 数、压缩后 token 数、摘要模型输出、被丢弃消息 ID 列表、保留消息 ID 列表。这样当用户反馈“你忘了之前的事”时，能直接查询是不是压缩搞的鬼。

在 Pi-Agent 里，这个事件就是一个普通日志：

{ "event": "compaction", "session_id": "sess_123", "round": 34, "before_tokens": 14500, "after_tokens": 3200, "dropped_ids": [101, 102, 103], "preserved_ids": [104, 105, 106, 107] }

有了这个，排查“压缩导致失忆”就不需要猜了。我做过一个分析脚本，把每次用户抱怨“你忘了”的会话拉出来对比压缩事件，发现 80% 以上都出现在“全局摘要只保留了 3 条关键事实”的会话里，直接把摘要条数下限从 3 改成 5，投诉率降了一半。

6.4 后续方向：增量压缩与语义索引

目前 Pi-Agent 的 Compaction 是“整段重写”式的，每次压缩都把已有摘要和新对话重新合在一起再生成一次全局摘要。这样做的缺点是有信息重复消耗，且频率高了会累积误差。

我正在尝试的改进是增量压缩：每次只摘要新增的对话轮次，然后与旧摘要进行“合并”而不是“重新生成”。合并提示词大概是“检查旧摘要和新摘要，保留旧摘要中所有仍然成立的事实，加入新出现的关键实体”。这个方向对节省 token 很有帮助，但合并后的逻辑一致性还要更多测试。

另一个方向是语义索引：不把所有信息压缩进 Prompt，而是把关键事实存入一个本地检索库，重建上下文时只把与当前问题相关的摘要片段注入 Prompt。这其实就是 RAG（检索增强生成）和 Compaction 的结合。对超长生命周期会话来说，这是比无限压缩更可行的路。

最后分享一个小技巧：无论采用哪种压缩策略，上线前都一定要准备一组“标准遗忘记忆测试集”，里面包含二十个类似“三十分钟前我让你查的那家餐厅的地址是什么”的问题。每次调整压缩逻辑后，先跑一遍测试集，对比回答准确率。没有这个测试集，压缩参数就是盲调，出问题还不知道怪谁。这是我在 Pi-Agent 上做过最值的一次投资。