第三部分 · 持久化编排

第 7 章:上下文压缩与分支摘要

压缩不是删除历史,而是往会话树里插入一个摘要节点——让有限的窗口装得下无限的任务

有限的窗口,无限的任务

每个模型都有一个上下文窗口——一次能"看到"的 token 上限。但真实的编码任务往往很长:读几十个文件、跑几十次命令、来回讨论几十轮。转录会膨胀,迟早撞上那堵墙。撞上之后怎么办?

综合体路线的回答是"四层压缩"——snip、microcompact、collapse、autocompact,从轻到重逐级触发,外加 circuit breaker 防止无限重试。那是一个潜艇式的复杂系统。

Pi Agent 的回答简单得多,却有一个独特的 twists:压缩不删除历史。 回忆第 5 章:转录是一棵不可变的树。压缩不是从树里删节点,而是往树里插入一个摘要节点(CompactionEntry)。原始历史完好无损地留在树上——压缩只改变"发给模型的上下文视图"。这让压缩成为一个可逆的视图变换:用户可以回到压缩点之前,重新展开那段历史。

本章拆解这个机制:何时压缩、在哪里切、摘要怎么写、以及它如何与那棵树咬合。


何时压缩:阈值与估算

压缩的触发判断是一个简单的不等式。shouldCompactpackages/agent/src/harness/compaction/compaction.ts):

export function shouldCompact(contextTokens: number, contextWindow: number, settings: CompactionSettings): boolean {
	if (!settings.enabled) return false;
	return contextTokens > contextWindow - settings.reserveTokens;
}

当估算的上下文 token 数超过"窗口大小减去预留空间"时,就该压缩了。默认设置(DEFAULT_COMPACTION_SETTINGS):

export const DEFAULT_COMPACTION_SETTINGS: CompactionSettings = {
	enabled: true,
	reserveTokens: 16384,   // 为摘要提示词和输出预留
	keepRecentTokens: 20000, // 压缩后大约保留多少近期 token
};
  • reserveTokens(16,384):在窗口顶部预留的空间,给摘要生成那次模型调用用,也作为安全边际。
  • keepRecentTokens(20,000):压缩后希望保留的近期上下文量——切点选择围绕这个预算展开。

token 估算:以权威用量为锚

判断"现在有多少 token"本身是个难题——你不可能每次都把整个转录发给模型去精确计数。estimateContextTokens 用了一个聪明的混合策略:

export function estimateContextTokens(messages: AgentMessage[]): ContextUsageEstimate {
	const usageInfo = getLastAssistantUsageInfo(messages);
	if (!usageInfo) {
		// 没有任何权威用量:全部用启发式估算
		let estimated = 0;
		for (const message of messages) estimated += estimateTokens(message);
		return { tokens: estimated, usageTokens: 0, trailingTokens: estimated, lastUsageIndex: null };
	}
	// 有权威用量:用它做锚,只估算锚点之后的新增消息
	const usageTokens = calculateContextTokens(usageInfo.usage);
	let trailingTokens = 0;
	for (let i = usageInfo.index + 1; i < messages.length; i++) {
		trailingTokens += estimateTokens(messages[i]);
	}
	return { tokens: usageTokens + trailingTokens, usageTokens, trailingTokens, lastUsageIndex: usageInfo.index };
}

关键洞见:最近一次助手响应的 usage 字段是权威的——它是提供商真实报告的、包含那条消息为止整个上下文的 token 数。于是估算变成"权威锚点 + 锚点之后的启发式增量"。误差被限制在"最近一轮新增的内容",而不是累积整个对话。这比从头估算整个转录精确得多。

锚点不存在时(比如会话刚开始),退回到纯启发式 estimateTokens——一个保守的"字符数 / 4"估算,图片按 ESTIMATED_IMAGE_CHARS = 4800 字符计:

export function estimateTokens(message: AgentMessage): number {
	let chars = 0;
	switch (message.role) {
		case "user":
			chars = estimateTextAndImageContentChars(message.content);
			return Math.ceil(chars / 4);
		case "assistant":
			// 累加 text、thinking、toolCall(名字 + 参数 JSON)的字符
			return Math.ceil(chars / 4);
		// ... toolResult, bashExecution, branchSummary, compactionSummary
	}
	return 0;
}

calculateContextTokens(usage) 则从权威用量里取总数:usage.totalTokens || usage.input + usage.output + usage.cacheRead + usage.cacheWrite


在哪里切:切点选择

决定要压缩之后,下一个问题是:从哪里切开?切点之前进摘要,切点之后保留原文。findCutPoint 做这个决策,目标是保留大约 keepRecentTokens 的近期上下文

合法的切点

不是任何位置都能切。findValidCutPoints 筛选出合法的切点:

function findValidCutPoints(entries: SessionTreeEntry[], startIndex: number, endIndex: number): number[] {
	const cutPoints: number[] = [];
	for (let i = startIndex; i < endIndex; i++) {
		const entry = entries[i];
		switch (entry.type) {
			case "message": {
				const role = entry.message.role;
				switch (role) {
					case "bashExecution":
					case "custom":
					case "branchSummary":
					case "compactionSummary":
					case "user":
					case "assistant":
						cutPoints.push(i); // 这些消息边界可以切
						break;
					case "toolResult":
						break; // 工具结果不能作为切点
				}
				break;
			}
			// 配置变更条目(model_change 等)不是切点
		}
		if (entry.type === "branch_summary" || entry.type === "custom_message") cutPoints.push(i);
	}
	return cutPoints;
}

注意 toolResult 不能作为切点。为什么?因为协议要求每个 tool_use 后面必须跟着对应的 tool_result(第 3 章的"孤立工具结果"问题)。如果在工具调用和它的结果之间切开,保留的后半段就会以一个孤立的 toolResult 开头,提供商会拒绝。把切点限制在 user/assistant 等消息边界,保证了切开后的两半各自都是合法的对话片段。

从尾部倒推预算

findCutPoint 从尾部往前累积 token,直到达到 keepRecentTokens 预算,然后选那个位置之后最近的合法切点:

export function findCutPoint(entries, startIndex, endIndex, keepRecentTokens): CutPointResult {
	const cutPoints = findValidCutPoints(entries, startIndex, endIndex);
	if (cutPoints.length === 0) {
		return { firstKeptEntryIndex: startIndex, turnStartIndex: -1, isSplitTurn: false };
	}
	let accumulatedTokens = 0;
	let cutIndex = cutPoints[0];

	// 从尾部往前累积,直到达到 keepRecentTokens
	for (let i = endIndex - 1; i >= startIndex; i--) {
		const entry = entries[i];
		if (entry.type !== "message") continue;
		accumulatedTokens += estimateTokens(entry.message);
		if (accumulatedTokens >= keepRecentTokens) {
			// 找到这个位置之后最近的合法切点
			for (let c = 0; c < cutPoints.length; c++) {
				if (cutPoints[c] >= i) { cutIndex = cutPoints[c]; break; }
			}
			break;
		}
	}
	// 往前微调,避免切在 compaction 或 message 之前的"空档"
	while (cutIndex > startIndex) {
		const prevEntry = entries[cutIndex - 1];
		if (prevEntry.type === "compaction" || prevEntry.type === "message") break;
		cutIndex--;
	}
	const cutEntry = entries[cutIndex];
	const isUserMessage = cutEntry.type === "message" && cutEntry.message.role === "user";
	const turnStartIndex = isUserMessage ? -1 : findTurnStartIndex(entries, cutIndex, startIndex);
	return {
		firstKeptEntryIndex: cutIndex,
		turnStartIndex,
		isSplitTurn: !isUserMessage && turnStartIndex !== -1,
	};
}

逻辑是:从最新的消息往前数,数到累计约 20,000 token 为止,然后对齐到最近的合法切点。这样保留的尾部大约是 keepRecentTokens 大小,切点之前的进摘要。

split-turn:切在一轮中间

有一个微妙的情况:切点可能落在一轮对话的中间——比如一个用户请求触发了 10 次工具调用,切点恰好切在这 10 次调用的第 5 次之后。这时切点之后的"保留尾部"会以一个没有上下文的工具结果开头,模型会困惑。

findCutPoint 通过 isSplitTurn 标志检测这种情况:如果切点不是 user 消息,它会用 findTurnStartIndex 往前找到这一轮的起点(发起这轮的 user 消息)。isSplitTurn: true 意味着"切点把一轮切成了两半"。

这种情况需要特殊处理——保留的尾部(一轮的后半)需要一段"前半发生了什么"的额外说明。我们马上会看到 compact 如何用一个独立的"turn prefix 摘要"来解决它。


准备:prepareCompaction

prepareCompaction 把切点决策转化成一个 CompactionPreparation——压缩所需的一切输入:

export function prepareCompaction(pathEntries, settings): Result<CompactionPreparation | undefined, CompactionError> {
	// 如果路径为空,或最后一个条目已经是 compaction,无需压缩
	if (pathEntries.length === 0 || pathEntries[pathEntries.length - 1].type === "compaction") {
		return ok(undefined);
	}

	// 找到上一个 compaction(用于迭代更新摘要)
	let prevCompactionIndex = -1;
	for (let i = pathEntries.length - 1; i >= 0; i--) {
		if (pathEntries[i].type === "compaction") { prevCompactionIndex = i; break; }
	}
	let previousSummary: string | undefined;
	let boundaryStart = 0;
	if (prevCompactionIndex >= 0) {
		previousSummary = (pathEntries[prevCompactionIndex] as CompactionEntry).summary;
		// 边界从上次压缩的 firstKeptEntry 开始
		// ...
	}

	const tokensBefore = estimateContextTokens(buildSessionContext(pathEntries).messages).tokens;
	const cutPoint = findCutPoint(pathEntries, boundaryStart, pathEntries.length, settings.keepRecentTokens);
	const firstKeptEntryId = pathEntries[cutPoint.firstKeptEntryIndex].id;

	// 三段消息:
	const historyEnd = cutPoint.isSplitTurn ? cutPoint.turnStartIndex : cutPoint.firstKeptEntryIndex;
	const messagesToSummarize = []; // 边界 → historyEnd:进主摘要
	const turnPrefixMessages = [];  // turnStart → 切点:split-turn 的前半,单独摘要
	const retainedTail = [];        // 切点 → 末尾:保留原文

	const fileOps = extractFileOperations(messagesToSummarize, pathEntries, prevCompactionIndex);

	return ok({ firstKeptEntryId, messagesToSummarize, turnPrefixMessages, retainedTail, isSplitTurn, tokensBefore, previousSummary, fileOps, settings });
}

它把路径切成三段:

图 1:准备压缩时把路径切成三段——主摘要、split-turn 的前缀摘要与保留原文的尾部。
  • messagesToSummarize:从边界到切点(或 split-turn 的轮次起点)的消息,将被压缩进主摘要。
  • turnPrefixMessages:仅 split-turn 时有——被切开的那一轮的前半段,需要单独摘要。
  • retainedTail:切点之后的近期消息,保留原文,直接存在压缩条目上。

注意 previousSummary:如果之前已经压缩过,这次压缩不是从零开始,而是在上次摘要的基础上增量更新——这是一个重要的优化,我们下面会看到。

fileOps 则提取这段历史里读过的、改过的文件(extractFileOperations),存进压缩条目的 details。于是压缩后的上下文依然"记得"动过哪些文件——CompactionDetails { readFiles, modifiedFiles }


摘要怎么写:结构化提示词

摘要的质量决定了压缩后 agent 还能不能接着干活。Pi Agent 用一个高度结构化的提示词来约束摘要格式。系统提示词(SUMMARIZATION_SYSTEM_PROMPT)先把模型钉死在"只输出摘要"的角色上:

You are a context summarization assistant. Your task is to read a conversation between a user and an AI assistant, then produce a structured summary following the exact format specified.

Do NOT continue the conversation. Do NOT respond to any questions in the conversation. ONLY output the structured summary.

然后用户提示词(SUMMARIZATION_PROMPT)规定了精确的输出格式:

## Goal
[What is the user trying to accomplish?]

## Constraints & Preferences
- [Any constraints, preferences, or requirements mentioned by user]

## Progress
### Done
- [x] [Completed tasks/changes]
### In Progress
- [ ] [Current work]
### Blocked
- [Issues preventing progress, if any]

## Key Decisions
- **[Decision]**: [Brief rationale]

## Next Steps
1. [Ordered list of what should happen next]

## Critical Context
- [Any data, examples, or references needed to continue]

Keep each section concise. Preserve exact file paths, function names, and error messages.

这个格式的设计很讲究。它不是"把对话缩写成一段话",而是把对话重构成一份交接文档——目标、约束、进度(已完成/进行中/受阻)、关键决策、下一步、关键上下文。最后一句"Preserve exact file paths, function names, and error messages"尤其关键:摘要可以丢失措辞,但不能丢失精确的标识符,因为接下来的工作要靠它们。

迭代更新:UPDATE_SUMMARIZATION_PROMPT

如果已有 previousSummary(之前压缩过),就用 UPDATE_SUMMARIZATION_PROMPT 代替。它把旧摘要放在 <previous-summary> 标签里,要求模型增量更新而非重写:

Update the existing structured summary with new information. RULES:

- PRESERVE all existing information from the previous summary

- ADD new progress, decisions, and context from the new messages

- UPDATE the Progress section: move items from "In Progress" to "Done" when completed

- PRESERVE exact file paths, function names, and error messages

这避免了"每次压缩都把整个历史重新摘要一遍"的浪费。第二次压缩时,模型只需要读"上次摘要 + 上次之后新增的消息",而不是"全部原始历史"。压缩的成本因此不会随会话长度线性增长。

生成:generateSummaryWithUsage

实际的生成调用(generateSummaryWithUsage)有几个细节:

const maxTokens = Math.min(
	Math.floor(0.8 * reserveTokens),
	model.maxTokens > 0 ? model.maxTokens : Number.POSITIVE_INFINITY,
);
// ...
const response = await completeSimpleWithRetries(models, model, {
	systemPrompt: SUMMARIZATION_SYSTEM_PROMPT,
	messages: summarizationMessages,
}, completionOptions, retry, callbacks);
  • 输出上限maxTokens0.8 * reserveTokens 和模型上限的较小值——摘要输出被约束在预留空间内,不会自己撑爆窗口。
  • 隔离的请求completeSimpleWithRetries 给摘要调用一个全新的 sessionIdcacheRetention: "none"
export async function completeSimpleWithRetries(models, model, context, options, retry?, callbacks?) {
	// 摘要是独立请求:隔离路由,避免写入无法复用的缓存
	const requestOptions = { ...options, cacheRetention: "none", sessionId: uuidv7() };
	return retryAssistantCall(() => models.completeSimple(model, context, requestOptions), retry, requestOptions.signal, callbacks);
}

为什么隔离?因为摘要请求的内容(序列化的历史 + 摘要提示词)跟正常对话的 prompt 前缀完全不同,为它写 prompt cache 是浪费——那个缓存永远不会被复用。用 fresh sessionId + no cache,避免污染主对话的缓存统计。它还套了 retryAssistantCall(第 2 章的助手级重试),让摘要生成对临时失败有韧性。


落地:compact 与 split-turn

compact 把准备和生成串起来,产出最终的 CompactionResult

export async function compact(preparation, models, model, customInstructions?, signal?, thinkingLevel?, retry?, callbacks?): Promise<Result<CompactionResult, CompactionError>> {
	const { firstKeptEntryId, messagesToSummarize, turnPrefixMessages, retainedTail, isSplitTurn, tokensBefore, previousSummary, fileOps, settings } = preparation;

	let summary: string;
	let summaryUsage: Usage;

	if (isSplitTurn && turnPrefixMessages.length > 0) {
		// split-turn:先摘要历史,再单独摘要这一轮的前半
		let historyText = "No prior history.";
		if (messagesToSummarize.length > 0) {
			const historyResult = await generateSummaryWithUsage(messagesToSummarize, ...);
			if (!historyResult.ok) return err(historyResult.error);
			historyText = historyResult.value.text;
		}
		const turnPrefixResult = await generateTurnPrefixSummary(turnPrefixMessages, ...);
		if (!turnPrefixResult.ok) return err(turnPrefixResult.error);
		summary = `${historyText}\n\n---\n\n**Turn Context (split turn):**\n\n${turnPrefixResult.value.text}`;
		summaryUsage = combineUsage(...);
	} else {
		// 普通情况:一次摘要
		const summaryResult = await generateSummaryWithUsage(messagesToSummarize, ...);
		if (!summaryResult.ok) return err(summaryResult.error);
		summary = summaryResult.value.text;
		summaryUsage = summaryResult.value.usage;
	}
	// 返回 { summary, firstKeptEntryId, tokensBefore, usage, retainedTail, details: fileOps }
}

split-turn 的处理是亮点:当切点把一轮切成两半,compact 会做两次摘要——一次是常规的历史摘要(SUMMARIZATION_PROMPT),一次是专门的"轮次前缀摘要"(TURN_PREFIX_SUMMARIZATION_PROMPT),后者用另一个格式:

This is the PREFIX of a turn that was too large to keep. The SUFFIX (recent work) is retained.

## Original Request / ## Early Progress / ## Context for Suffix

然后把两段拼起来。这样保留的尾部(一轮的后半)就有了一段"前半发生了什么"的说明,模型不会看到一个没头没尾的工具结果。

CompactionResult 携带 summaryfirstKeptEntryId(保留历史从哪里开始)、tokensBeforeusageretainedTail(保留的尾部消息)、details(文件操作)。


与树咬合:压缩如何生效

compact 产出的结果,最终被 AgentHarness(或产品层)写成一个 CompactionEntry 追加进会话树。回忆第 5 章的 defaultContextEntryTransform:之后每次构建上下文时,如果路径里有这个 compaction 条目,上下文就变成"[compaction] + 保留的尾部"。

于是整个压缩流程闭环了:

图 2:压缩闭环。从估算、选切点到追加 CompactionEntry 进树,之后构建上下文用"摘要 + 尾部"替代压缩点之前。

而原始历史——messagesToSummarize 对应的那些条目——从未被删除。它们还在树上。用户 moveTo 回压缩点之前,defaultContextEntryTransform 就看不到那个 compaction 条目(因为它在路径之外),原始历史重新展开。压缩是可逆的。

这是 Pi Agent 与"删除式压缩"的根本区别。删除式压缩(从数组里 splice 掉旧消息)简单但不可逆,且丢失了"回看历史"的能力。Pi Agent 用树结构把"存储"和"视图"分离,代价是多一层抽象,收益是压缩、分支、回溯全都自然且可逆。


分支摘要:换分支时的交接

除了压缩,还有一种摘要:分支摘要(branch summarization,packages/agent/src/harness/compaction/branch-summarization.ts)。

回忆第 5 章:用户可以在会话树上 moveTo 到另一个分支。当你从分支 A 切到分支 B 时,分支 A 上发生的事对分支 B 的后续工作可能是有用的上下文("我刚才在另一个分支试过 X 方案,不行")。generateBranchSummary 在你离开一个分支时,为它生成一段摘要(BranchSummaryEntry),这样切回来或切过去时,那段探索的精华不会丢失。

collectEntriesForBranchSummary 收集要摘要的分支条目,generateBranchSummary 用类似的结构化提示词生成摘要。它和压缩摘要共享"把对话重构成结构化交接文档"的理念,只是触发时机不同:压缩是"上下文太满了",分支摘要是"我要离开这个分支了"。

AgentHarness.navigateTree(第 6 章)在分支导航时会触发它,phase 切到 branch_summary,并触发 session_before_tree/session_tree hook。


实践应用

Pi Agent 的压缩为"如何管理 Agent 的有限上下文"提供了四条可迁移的模式。

以权威用量为锚,只估算增量。 用最近一次提供商报告的真实 token 数做锚点,只对其后新增的消息做启发式估算。它解决的问题是:对整个转录做启发式估算会累积误差,而精确计数又太贵。锚点法把误差限制在"最近一轮",在成本与精度之间取得平衡。

切点对齐到协议合法的边界。 切点不能落在工具调用和它的结果之间,否则保留的后半段会以孤立结果开头。它解决的问题是:天真的"按 token 数切"会切出协议非法的片段。把切点约束在合法边界(user/assistant),保证切开后的两半各自合法。

把摘要写成结构化交接文档,而非缩写。 用固定格式(目标/约束/进度/决策/下一步/关键上下文)约束摘要,并强制保留精确的标识符。它解决的问题是:自由形式的摘要会丢失继续工作所需的关键细节(文件路径、函数名、错误信息)。结构化格式让摘要成为一份可用的交接文档,而不是一段模糊的概括。

压缩是视图变换,不是数据删除。 把"存储完整历史"和"发给模型什么"分离——压缩插入一个摘要节点,原始历史保留,上下文构建时按需替代。它解决的问题是:删除式压缩不可逆,丢失了回看和回溯的能力。当压缩可逆,用户就能在"省 token"和"看历史"之间自由切换。代价是需要树状的存储模型——又一次,第 5 章的树结构在这里兑现了它的价值。


总结

Pi Agent 的压缩是一个围绕"树状转录"设计的机制:shouldCompact 用"权威锚点 + 增量估算"判断时机,findCutPoint 在协议合法的边界上保留约 keepRecentTokens 的近期上下文(并检测 split-turn),prepareCompaction 把路径切成三段,generateSummary 用结构化提示词把历史重构成一份交接文档(支持迭代更新与 split-turn 的双段摘要),最后 compact 产出的 CompactionEntry 被追加进树——原始历史永不删除,压缩只是上下文视图的可逆替换。分支摘要则在切换分支时保存探索的精华。

它比综合体路线的"四层压缩"简单一个数量级,却没有牺牲核心能力——因为复杂性被那棵树吸收了。当你理解了第 5 章的树,这里的压缩几乎是水到渠成的。这正是好的底层抽象的价值:它让上层的功能变得简单。

下一章,我们离开 pi-agent-core,进入产品层 pi-coding-agent——看 pi 命令如何从一行 cli.ts 启动,一路装配出完整的 Agent。