🗜️ Context Compaction —— 智能对话历史压缩的艺术

"记忆不是关于记住一切，而是关于记住重要的东西。"

一、问题的本质：Token 限制与无限对话的矛盾

在 LLM 时代，我们面临一个根本性的矛盾：

用户的期望： 希望 Agent 记住整个对话历史，越详细越好

LLM 的现实： Token 有限制（4K, 8K, 100K, 200K...但终究有限）

想象一下这样的场景：

用户正在和 Agent 一起开发一个大型项目...

第 1 轮：讨论架构设计（500 tokens）
第 2 轮：分析代码库（2000 tokens）
第 3 轮：编写核心模块（3000 tokens）
第 4 轮：调试和修复（2500 tokens）
第 5 轮：添加测试（2000 tokens）
...
第 50 轮：累计 100,000+ tokens

【危机时刻】
Agent: "抱歉，上下文已满，我无法继续。
        请开始一个新的会话。"

用户: "但是...我需要之前所有的讨论历史！"

这不是科幻场景，这是每个使用 LLM 的人都会遇到的问题。

传统的解决方案：

1. 粗暴截断 —— 只保留最近 N 轮对话，丢失早期信息
2. 人工总结 —— 让用户自己整理要点，体验极差
3. 分会话处理 —— 割裂的上下文，丢失连贯性

我们需要一种智能的机制，自动压缩对话历史，保留关键信息。

这就是 Context Compaction。

---

二、Context Compaction 是什么？

Context Compaction（上下文压缩） 是 Kimi Code CLI 中的智能对话管理技术。当对话历史接近 Token 限制时，它会自动：

1. 分析 整个对话历史
2. 提取 关键信息（任务状态、错误、设计决策等）
3. 压缩 冗余内容（中间尝试、重复解释等）
4. 重构 为简洁的摘要

核心目标

---

三、Compaction 的触发机制

触发条件

async def _agent_loop(self):
    while True:
        # 检查是否需要压缩
        reserved = self._loop_control.reserved_context_size
        if self._context.token_count + reserved >= self._runtime.llm.max_context_size:
            logger.info("Context too long, compacting...")
            await self.compact_context()  # ← 触发压缩！

阈值计算：

当前 Token 数 + 预留空间 >= LLM 最大容量
        ↓
     触发压缩

预留空间（reserved）是为了给下一轮对话留出余地。

压缩过程

async def compact_context(self):
    # 1. 发送压缩开始事件
    wire_send(CompactionBegin())
    
    # 2. 调用 Compaction 策略
    compacted_messages = await self._compaction.compact(
        self._context.history, 
        self._runtime.llm
    )
    
    # 3. 清空并重建上下文
    await self._context.clear()
    await self._checkpoint()
    await self._context.append_message(compacted_messages)
    
    # 4. 发送压缩结束事件
    wire_send(CompactionEnd())

---

四、SimpleCompaction：保留最近，压缩过去

Kimi Code CLI 目前实现了 SimpleCompaction 策略。虽然名字里有 "Simple"，但它的设计非常精巧。

核心思想

┌─────────────────────────────────────────────────────────┐
│                    原始对话历史                          │
├─────────────────────────────────────────────────────────┤
│ [0] 用户: 开始任务                                      │
│ [1] Agent: 收到，让我分析...                             │
│ [2] 用户: 发现了问题 A                                   │
│ [3] Agent: 解决 A 的方案是...                            │
│ ...（很多轮对话）...                                     │
│ [46] Agent: 尝试方案 X（失败）                           │
│ [47] Agent: 尝试方案 Y（失败）                           │
│ [48] Agent: 尝试方案 Z（成功！）                         │
│ [49] 用户: 很好，继续下一步                              │
└─────────────────────────────────────────────────────────┘
                          │
                          ▼ 压缩
┌─────────────────────────────────────────────────────────┐
│                    压缩后的上下文                        │
├─────────────────────────────────────────────────────────┤
│ [摘要] 任务：XXX；关键问题：A、B、C；                    │
│        解决方案：Z；当前状态：进行中                     │
│ [48] Agent: 尝试方案 Z（成功！）  ← 保留（最近2条）     │
│ [49] 用户: 很好，继续下一步        ← 保留（最近2条）    │
└─────────────────────────────────────────────────────────┘

实现代码

class SimpleCompaction:
    def __init__(self, max_preserved_messages: int = 2):
        # 保留最近 N 条消息不压缩
        self.max_preserved_messages = max_preserved_messages

    async def compact(self, messages: Sequence[Message], llm: LLM):
        compact_message, to_preserve = self.prepare(messages)
        
        if compact_message is None:
            return to_preserve
        
        # 调用 LLM 生成摘要
        result = await kosong.step(
            chat_provider=llm.chat_provider,
            system_prompt="You are a helpful assistant that compacts conversation context.",
            toolset=EmptyToolset(),  # 压缩时禁用工具
            history=[compact_message],
        )
        
        # 构建压缩后的消息列表
        content = [
            system("Previous context has been compacted. Here is the compaction output:")
        ]
        # 去除思考部分
        content.extend(part for part in result.message.content 
                      if not isinstance(part, ThinkPart))
        
        return [
            Message(role="user", content=content),
            *to_preserve  # 保留的最近消息
        ]

准备阶段：分离保留和压缩

def prepare(self, messages: Sequence[Message]):
    history = list(messages)
    
    # 从后往前找，保留最近的用户/助手消息
    preserve_start_index = len(history)
    n_preserved = 0
    for index in range(len(history) - 1, -1, -1):
        if history[index].role in {"user", "assistant"}:
            n_preserved += 1
            if n_preserved == self.max_preserved_messages:
                preserve_start_index = index
                break
    
    to_compact = history[:preserve_start_index]   # 需要压缩的部分
    to_preserve = history[preserve_start_index:]  # 保留的部分
    
    # 构建压缩输入
    compact_message = Message(role="user", content=[])
    for i, msg in enumerate(to_compact):
        compact_message.content.append(
            TextPart(text=f"## Message {i + 1}\\nRole: {msg.role}\\nContent:\\n")
        )
        compact_message.content.extend(
            part for part in msg.content if not isinstance(part, ThinkPart)
        )
    compact_message.content.append(TextPart(text="\\n" + prompts.COMPACT))
    
    return compact_message, to_preserve

---

五、压缩 Prompt：如何指导 LLM 摘要

压缩的质量完全取决于 Prompt。这是 Kimi Code CLI 使用的 COMPACT prompt：

---

The above is a list of messages in an agent conversation. 
You are now given a task to compact this conversation context 
according to specific priorities and rules.

**Compression Priorities (in order):**
1. **Current Task State**: What is being worked on RIGHT NOW
2. **Errors & Solutions**: All encountered errors and their resolutions
3. **Code Evolution**: Final working versions only (remove intermediate attempts)
4. **System Context**: Project structure, dependencies, environment setup
5. **Design Decisions**: Architectural choices and their rationale
6. **TODO Items**: Unfinished tasks and known issues

**Compression Rules:**
- MUST KEEP: Error messages, stack traces, working solutions, current task
- MERGE: Similar discussions into single summary points
- REMOVE: Redundant explanations, failed attempts (keep lessons learned), verbose comments
- CONDENSE: Long code blocks → keep signatures + key logic only

**Special Handling:**
- For code: Keep full version if < 20 lines, otherwise keep signature + key logic
- For errors: Keep full error message + final solution
- For discussions: Extract decisions and action items only

**Required Output Structure:**

<current_focus>
[What we're working on now]
</current_focus>

<environment>
- [Key setup/config points]
</environment>

<completed_tasks>
- [Task]: [Brief outcome]
</completed_tasks>

<active_issues>
- [Issue]: [Status/Next steps]
</active_issues>

<code_state>
<file>
[filename]
**Summary:** [What this code file does]
**Key elements:** [Important functions/classes]
**Latest version:** [Critical code snippets]
</file>
</code_state>

<important_context>
- [Any crucial information not covered above]
</important_context>

Prompt 设计分析

1. 优先级排序
最重要的信息排在前面，LLM 会优先保留：

• 当前任务状态（绝对不能丢）
• 错误和解决方案（避免重复踩坑）
• 代码最终版本（中间尝试没用）

2. 明确规则
用 MUST KEEP / MERGE / REMOVE / CONDENSE 明确告诉 LLM 如何处理不同类型的内容。

3. 特殊处理
代码、错误、讨论各自有不同的处理策略。

4. 结构化输出
强制使用 XML 标签，便于后续解析（虽然当前只是展示给用户）。

---

六、压缩效果示例

示例：Web 开发会话

原始对话（10 轮，约 5000 tokens）：

[1] 用户: 帮我创建一个 Flask 博客应用
[2] Agent: 好的，我需要先了解项目结构...
[3] Agent: （读取文件，分析）
[4] Agent: 我建议这样的架构：...
[5] 用户: 可以，但要用 SQLAlchemy
[6] Agent: 好的，使用 SQLAlchemy。首先安装依赖...
[7] Agent: pip install flask sqlalchemy
[8] Agent: 创建 app.py...
[9] Agent: （编写 50 行代码）
[10] 用户: 运行报错了
[11] Agent: 让我看看... 是导入错误
[12] Agent: 修复：把 from models import db 改为 ...
[13] Agent: （修复代码）
[14] 用户: 可以运行了，现在添加文章发布功能
[15] Agent: 好的，添加路由和模板...
[16] Agent: （编写 80 行代码）
[17] 用户: 模板报错了
[18] Agent: Jinja2 语法错误，缺少 endif
[19] Agent: （修复模板）
[20] 用户: 很好，继续添加评论功能

压缩后（约 800 tokens）：

<current_focus>
正在开发 Flask 博客应用，已完成基础架构和文章发布功能，
正在添加评论功能。
</current_focus>

<environment>
- 框架：Flask + SQLAlchemy
- 数据库：SQLite（默认）
- 模板引擎：Jinja2
</environment>

<completed_tasks>
- 项目初始化：创建 app.py，配置 SQLAlchemy
- 文章发布：/post 路由 + 表单 + 模板（修复了 Jinja2 语法错误）
</completed_tasks>

<active_issues>
- 正在添加：评论功能
</active_issues>

<code_state>
<file>
app.py
**Summary:** Flask 应用主文件
**Key elements:**
- create_app(): 应用工厂函数
- Post 模型: 文章数据模型
- /post 路由: 文章发布
**Latest version:**
（关键代码片段，约 20 行）
</file>
</code_state>

<important_context>
- 注意 Jinja2 模板语法要闭合（已修复一次错误）
- 使用 SQLAlchemy 的 db.Model 作为基类
</important_context>

压缩率：5000 → 800 tokens（约 84% 的压缩率）

---

七、Compaction 的时机与频率

理想情况

理论上，Compaction 应该在：

• 接近 Token 限制时触发（当前实现）
• 任务阶段转换时触发（更智能）
• 用户主动请求时触发（可控）

当前策略的局限

SimpleCompaction 保留最近 2 条消息 的设计有一些问题：

假设对话：
[45] 用户: 帮我优化这个函数
[46] Agent: （很长的分析，1000 tokens）
[47] 用户: 好的按你说的做
[48] Agent: （很长的实现，1500 tokens）

触发压缩时：
- 保留 [47], [48]（最近2条）
- 压缩 [0] 到 [46]

问题：
- [47] 的 "好的按你说的做" 没有实际信息量
- [48] 的具体实现代码会被保留，占用大量 Token
- 但 [46] 的分析过程却被压缩了

可能的改进

class SmartCompaction:
    def prepare(self, messages):
        # 根据信息量而非时间选择保留的消息
        # 使用嵌入向量评估每条消息的重要性
        # 保留"信息密度"最高的 N 条
        
        importance_scores = [
            (msg, self.calculate_importance(msg))
            for msg in messages
        ]
        importance_scores.sort(key=lambda x: x[1], reverse=True)
        
        to_preserve = [msg for msg, _ in importance_scores[:self.n_preserve]]
        to_compact = [msg for msg, _ in importance_scores[self.n_preserve:]]
        
        return to_compact, to_preserve

---

八、Compaction 与用户体验

UI 反馈

当 Compaction 发生时，用户会收到视觉反馈：

wire_send(CompactionBegin())
# ... 执行压缩 ...
wire_send(CompactionEnd())

Shell UI 可以显示：

🔄 正在压缩对话历史...
✅ 压缩完成（5000 → 800 tokens）

透明度与可控性

当前设计：

• 自动触发，用户无感知
• 压缩内容对用户不可见（除非查看日志）

可能的改进：

# 选项 1：显示压缩摘要
CompactionEnd(summary="保留：任务状态、代码最终版本、错误记录")

# 选项 2：用户确认
wire_send(CompactionRequest())
# 等待用户确认后再压缩

# 选项 3：手动压缩
用户输入：/compact
Agent：显示压缩预览，等待确认

---

九、与其他系统的协作

与 Checkpoint 的协作

[对话历史]
    │
    ▼ 触发 Compaction
[压缩后的上下文]
    │
    ▼ 创建 Checkpoint
[Checkpoint 指向压缩后的状态]

如果之后发生 D-Mail 回滚，会回滚到压缩后的状态，而不是原始状态。

与 Wire 协议的协作

class CompactionBegin(BaseModel):
    """压缩开始事件"""
    pass

class CompactionEnd(BaseModel):
    """压缩结束事件"""
    pass

UI 可以监听这些事件，显示进度或提供交互。

与 Token 计算的协作

async def update_token_count(self, token_count: int):
    self._token_count = token_count
    # 写入持久化存储
    await f.write(json.dumps({"role": "_usage", "token_count": token_count}) + "\\n")

准确的 Token 计数是触发 Compaction 的前提。

---

十、设计哲学

1. 遗忘的艺术

"完美的记忆不是记住一切，而是记住值得记住的。"

Compaction 本质上是一种有选择性的遗忘。它要求 LLM 扮演一个经验丰富的助手，知道什么信息是关键，什么可以舍弃。

2. 渐进式抽象

随着对话的进行，细节被逐步抽象：

具体实现 → 代码摘要 → 功能描述 → 任务状态

这是人类处理复杂任务的天然方式 —— 从细节到概览。

3. 信任 LLM 的理解能力

Compaction 假设 LLM 能够从摘要中恢复上下文。这是对 LLM 能力的信任，也是对其能力的有效利用。

---

十一、局限与未来

当前局限

1. 固定保留数量 —— 总是保留最近 2 条，不考虑信息量
2. 单次压缩 —— 不支持分层压缩（压缩后的内容再次被压缩）
3. 无用户控制 —— 全自动，用户无法干预
4. 英文 Prompt —— 对于中文对话可能不是最优

可能的增强

# 分层压缩
class HierarchicalCompaction:
    async def compact(self, messages, llm):
        # 第 1 层：压缩单个消息（代码摘要）
        # 第 2 层：压缩轮次（合并相似讨论）
        # 第 3 层：压缩主题（提取里程碑）
        pass

# 语义压缩
class SemanticCompaction:
    def calculate_importance(self, message):
        # 使用嵌入向量计算语义重要性
        embedding = self.embed(message.content)
        # 与当前任务的相关度
        return cosine_similarity(embedding, task_embedding)

# 交互式压缩
class InteractiveCompaction:
    async def compact(self, messages, llm):
        summary = await self.generate_summary(messages)
        # 询问用户是否确认
        if await self.confirm_with_user(summary):
            return summary
        else:
            # 用户提供指导
            guidance = await self.get_user_guidance()
            return await self.regenerate(summary, guidance)

---

十二、结语

Context Compaction 是 Kimi Code CLI 应对 LLM Token 限制的智能解决方案。它让 Agent 能够：

• 持续工作 —— 不受 Token 限制约束
• 保持连贯 —— 压缩后仍能理解上下文
• 聚焦重点 —— 自动提取关键信息

更重要的是，Compaction 代表了一种人机协作的新范式 —— 不是人类手动整理笔记，而是 AI 自动管理记忆。

正如一位哲学家所说：

"智慧不在于记住多少，而在于知道什么值得记住。"

Context Compaction 正是这种智慧的工程化实现。

---

文章完成时间：2026-02-23
作者：爪爪