仿照 Gemini CLI 的上下文压缩：智能对话历史管理

引言

在构建基于大语言模型（LLM）的智能体时，一个常见的挑战是如何处理不断增长的对话历史。随着交互轮次增加，上下文长度可能迅速接近模型的 token 限制，导致后续响应质量下降甚至请求失败。传统的截断方法会丢失重要信息，而简单的摘要又可能遗漏关键细节。

Google 的 Gemini CLI 在这方面提供了一个优雅的解决方案：聊天历史压缩。当对话接近 token 限制时，系统会自动将历史压缩为一个结构化的摘要，既保留了核心信息，又大幅减少了 token 占用。Infini-Agent-Framework 借鉴了这一设计思想，实现了自己的上下文压缩模块，本文将深入解析其设计与实现。

压缩模块概览

Infini-Agent-Framework 的压缩模块位于 infini_agent_framework.langgraph.compression，提供了以下核心能力：

  • 阈值触发：当对话历史 token 数超过预设阈值（默认 70% 的模型最大长度）时自动触发压缩。
  • 结构化摘要：使用 LLM 将历史转换为格式化的 XML 快照，包含目标、知识、文件状态、行动和计划等关键维度。
  • 智能分割：保留最近的历史（默认 30%）以维持连贯性，仅压缩较早部分。
  • 验证机制：确保压缩后的 token 数确实减少，否则回退到原始历史。

核心设计

抽象基类：Compressor

模块定义了一个抽象基类 Compressor，规定了所有压缩器必须实现的接口：

1
2
3
4
5
6
7
8
9
class Compressor(ABC):
    def __init__(self, config: Settings):
        self.config: Settings = config

    @abstractmethod
    async def compress(
        self, messages: list[BaseMessage], config: RunnableConfig
    ) -> tuple[list[BaseMessage], bool]:
        raise NotImplementedError

这种设计允许未来扩展不同的压缩策略，而当前实现是 GoogleStyleCompressor。

GoogleStyleCompressor：生产级压缩器

GoogleStyleCompressor 是框架中的默认压缩器，其工作流程如下：

  1. 消息筛选：过滤掉无效或空消息，保留完整的用户-AI-工具调用序列。
  2. 阈值检查：计算当前 token 数，若未超过阈值则直接返回。
  3. 寻找分割点：在用户消息边界处划分“待压缩部分”和“保留部分”。
  4. 生成摘要：使用专门的 LLM 提示词将待压缩历史转换为结构化 XML。
  5. 重建上下文：用摘要替换原始历史，并保留最近消息。
  6. 验证效果：确认压缩后的 token 数确实减少。

压缩触发机制

压缩的触发依赖于两个关键配置参数：

  • compress_threshold：压缩触发阈值，默认 0.7（即达到模型最大 token 数的 70% 时触发）。
  • preserve_threshold：保留比例，默认 0.3（即最近 30% 的历史保持不变）。

例如，对于最大 token 数为 2000 的模型：

  • 触发阈值 = 2000 × 0.7 = 1400 tokens
  • 当历史超过 1400 tokens 时，系统会压缩前 70% 的历史，保留后 30%。

这种设计确保了压缩只在必要时发生，且始终保留最近的上下文以维持对话连贯性。

结构化 XML 快照

压缩的核心在于将非结构化的对话历史转换为高度结构化的 XML 快照。以下是使用的提示词模板（简化版）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
<state_snapshot>
    <overall_goal>
        <!-- 用户的高级目标 -->
    </overall_goal>
    <key_knowledge>
        <!-- 关键事实、约定和约束 -->
    </key_knowledge>
    <file_system_state>
        <!-- 文件创建、读取、修改、删除状态 -->
    </file_system_state>
    <recent_actions>
        <!-- 最近的重要行动及其结果 -->
    </recent_actions>
    <current_plan>
        <!-- 当前步骤计划，标记已完成/进行中/待办 -->
    </current_plan>
</state_snapshot>

这种结构化的表示有多个优势：

  • 信息密度高：去除了对话中的冗余和无关内容。
  • 易于解析：后续处理可以精确提取特定字段。
  • 保持语义完整性：所有关键信息都被保留。

压缩执行机制与分割点算法

压缩动作如何执行？

压缩的核心动作是通过提示词让 LLM 完成的。当历史超过阈值时，系统会：

  1. 构造提示上下文：将待压缩的历史消息（经过筛选）与系统提示词 COMPRESSION_PROMPT 组合，最后追加一条用户指令 COMPRESSION_USER_INSTRUCTION。
  2. 调用 LLM：使用专用的压缩模型（默认 kimi-k2-instruct）生成结构化 XML 快照。
  3. 替换历史：将生成的快照封装为一个 HumanMessage，并添加一个 AIMessage 作为确认，然后与保留的最近消息拼接。

整个过程中，LLM 扮演了“摘要生成器”的角色，但压缩的触发、分割、验证等逻辑均由框架控制，确保压缩的可靠性和效率。

分割点如何确定？

分割点的确定是压缩算法的关键，目标是将历史划分为“待压缩部分”和“保留部分”。具体算法如下：

  1. 计算目标 token 数：根据配置的 preserve_threshold（保留比例），计算出需要压缩的 token 比例 fraction = 1 - preserve_threshold。例如，若保留 30%，则压缩前 70% 的 token。
  2. 遍历消息：从最早的消息开始，累计每条消息的 token 数（使用 count_tokens_approximately）。
  3. 寻找用户消息边界：当累计 token 数达到目标 token 数时，算法会继续向前扫描，直到遇到一条用户消息，并将该用户消息的索引作为分割点。这样确保分割点总是落在用户消息的起始处，保持对话的完整性。
  4. 特殊情况处理：
    • 如果遍历完所有消息仍未找到合适的用户消息，则根据最后一条消息的类型决定：
      • 若最后一条消息不是工具消息或用户消息（即 AI 消息），可以安全地压缩全部历史。
      • 否则，返回最后一个可能的分割点（即最近一个非用户消息的索引）。

这种设计保证了压缩后的历史仍然以完整的用户‑AI 对话轮次为单位，避免截断中间的工具调用序列，从而维持了对话的连贯性。

消息筛选与验证

并非所有消息都适合压缩。模块实现了精细的筛选逻辑：

  • 用户消息：始终保留，因为它们是对话的驱动因素。
  • AI 消息：仅当内容非空时才考虑。
  • 工具消息：需要有效的 tool_call_id。
  • 系统消息：在压缩过程中单独处理，通常保留在上下文开头。

此外，压缩器会验证生成的摘要是否真的减少了 token 数。如果压缩后 token 数反而增加（可能由于 LLM 生成过多内容），系统会回退到原始消息，确保不会意外扩大上下文。

完整压缩流程

让我们通过一个代码示例看看压缩的实际使用：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from infini_agent_framework.langgraph.compression import get_compressor
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

# 获取压缩器实例
compressor = get_compressor(config)

# 模拟一个长对话历史
messages = [
    SystemMessage(content="You are a helpful assistant."),
    HumanMessage(content="Build a web application for task management."),
    AIMessage(content="I'll start by creating the project structure..."),
    # ... 数十条后续消息
]

# 触发压缩
compressed_messages, was_compressed = await compressor.compress(messages, config)

if was_compressed:
    print(f"压缩成功！Token 数从 {original_tokens} 减少到 {new_tokens}")
    # 压缩后的消息列表包含：
    # 1. 原始系统消息（如果存在）
    # 2. HumanMessage 包含 XML 状态快照
    # 3. AIMessage 确认收到快照
    # 4. 保留的最近消息

实际应用场景

场景一：长对话会话管理

在客服机器人场景中，用户可能进行多轮复杂咨询。使用上下文压缩可以在不丢失关键信息的前提下，将长达数百条的消息压缩为几个结构化消息，使对话能够持续数小时而不触及 token 限制。

场景二：代码生成与迭代

当智能体协助开发时，用户会不断提出修改要求、查看代码、运行测试。压缩模块可以保持对项目目标、当前计划和文件状态的清晰跟踪，即使经过数十次代码修改也不会混淆。

场景三：多步骤任务执行

对于需要多个工具调用的复杂任务（如数据分析、报告生成），压缩可以定期“快照”进度，确保智能体始终记住最终目标，同时丢弃中间步骤的细节。