Claude Code 源码解析 (12)：上下文压缩与恢复技术

导读： 这是 Claude Code 20 个功能特性源码解析系列的第 12 篇，深入分析上下文管理系统的架构设计。

---

📋 目录

1. 问题引入：上下文窗口的限制
2. 技术原理：上下文压缩架构
3. 设计思想：为什么这样设计
4. 解决方案：完整实现详解
5. OpenClaw 最佳实践
6. 总结

---

问题引入：上下文窗口的限制

痛点场景

场景 1：长对话丢失上下文

用户："帮我分析这个项目..." (长篇说明)
AI："好的..." (开始分析)
   ↓
对话进行了 100 轮
   ↓
AI："抱歉，我记不清前面说的内容了"

→ 早期重要信息被截断
→ AI 失去上下文连贯性

场景 2：Token 超限

上下文窗口：128K tokens
当前对话：150K tokens

AI：无法处理，超出限制

→ 需要手动清理
→ 或者重新开始

场景 3：重要信息被压缩掉

对话历史：
- 用户偏好：使用 TypeScript
- 项目结构：Monorepo
- 部署目标：Kubernetes
   ↓
自动压缩后：
- 这些信息丢失了

AI：开始用 Python 写代码...
用户："我说过要用 TypeScript！"

核心问题

设计 AI 助手的上下文管理系统时，面临以下挑战：

1. 窗口限制问题

   • 如何突破 Token 限制？
   • 如何保留重要信息？

2. 压缩质量问题

   • 如何智能压缩？
   • 如何不丢失关键信息？

3. 恢复问题

   • 压缩后如何恢复？
   • 如何保持上下文连贯？

4. 性能问题

   • 压缩操作本身消耗 Token
   • 如何平衡压缩成本？

Claude Code 用智能上下文压缩机制解决了这些问题。

---

技术原理：上下文压缩架构

整体架构

┌─────────────────────────────────────────────────────────────┐
│                    原始对话历史                              │
│              (可能超出 Token 限制)                             │
└─────────────────────┬───────────────────────────────────────┘
                      │
                      ▼
┌─────────────────────────────────────────────────────────────┐
│  Context Manager (上下文管理器)                             │
│  ┌─────────────────────────────────────────────────────┐    │
│  │  1. Token 计数器                                     │    │
│  │     - 实时统计                                       │    │
│  │     - 预估新消息                                     │    │
│  │     - 触发压缩                                       │    │
│  └─────────────────────────────────────────────────────┘    │
│  ┌─────────────────────────────────────────────────────┐    │
│  │  2. 压缩策略选择                                     │    │
│  │     - 微压缩 (移除旧消息)                            │    │
│  │     - 智能压缩 (AI 摘要)                              │    │
│  │     - 记忆提取 (存储到记忆系统)                       │    │
│  └─────────────────────────────────────────────────────┘    │
│  ┌─────────────────────────────────────────────────────┐    │
│  │  3. 执行压缩                                         │    │
│  │     - 选择压缩对象                                   │    │
│  │     - 执行压缩算法                                   │    │
│  │     - 验证完整性                                     │    │
│  └─────────────────────────────────────────────────────┘    │
│  ┌─────────────────────────────────────────────────────┐    │
│  │  4. 上下文恢复                                       │    │
│  │     - 按需加载                                       │    │
│  │     - 智能检索                                       │    │
│  │     - 上下文注入                                     │    │
│  └─────────────────────────────────────────────────────┘    │
└─────────────────────────────────────────────────────────────┘
                      │
                      ▼
┌─────────────────────────────────────────────────────────────┐
│  压缩后的上下文                                             │
│  - 保留关键信息                                             │
│  - 移除冗余内容                                             │
│  - Token 使用优化                                            │
└─────────────────────────────────────────────────────────────┘

Token 计数器

class TokenCounter {
  private encoder: Tokenizer;
  private cache: Map<string, number> = new Map();
  
  constructor() {
    // 使用与模型相同的 tokenizer
    this.encoder = new Tokenizer('cl100k_base');
  }
  
  // 计算文本的 Token 数
  count(text: string): number {
    // 检查缓存
    if (this.cache.has(text)) {
      return this.cache.get(text)!;
    }
    
    // 计算 Token
    const tokens = this.encoder.encode(text);
    const count = tokens.length;
    
    // 缓存结果
    this.cache.set(text, count);
    
    return count;
  }
  
  // 计算消息列表的 Token 数
  countMessages(messages: Message[]): number {
    let total = 0;
    
    for (const msg of messages) {
      total += this.count(msg.content);
      
      // 加上消息格式的 overhead
      total += 4;  // role + content 标记
      
      if (msg.toolCalls) {
        for (const call of msg.toolCalls) {
          total += this.count(JSON.stringify(call));
        }
      }
    }
    
    // 加上对话格式的 overhead
    total += 2;  // start + end of conversation
    
    return total;
  }
  
  // 预估添加新消息后的 Token 数
  estimateWithNewMessage(
    messages: Message[],
    newMessage: string
  ): number {
    const current = this.countMessages(messages);
    const newTokens = this.count(newMessage);
    return current + newTokens + 4;  // +4 for message format
  }
  
  // 计算上下文预算
  calculateBudget(
    maxTokens: number,
    systemPrompt: string,
    tools: Tool[]
  ): ContextBudget {
    const systemTokens = this.count(systemPrompt);
    const toolTokens = tools.reduce((sum, tool) => 
      sum + this.count(JSON.stringify(tool.schema)), 0
    );
    
    // 预留 20% 给响应
    const reserved = Math.round(maxTokens * 0.2);
    
    return {
      max: maxTokens,
      system: systemTokens,
      tools: toolTokens,
      conversation: maxTokens - systemTokens - toolTokens - reserved,
      reserved,
    };
  }
}

interface ContextBudget {
  max: number;
  system: number;
  tools: number;
  conversation: number;
  reserved: number;
}

压缩策略

class CompressionStrategy {
  private llm: LLMClient;
  private tokenCounter: TokenCounter;
  
  constructor(llm: LLMClient, tokenCounter: TokenCounter) {
    this.llm = llm;
    this.tokenCounter = tokenCounter;
  }
  
  // 选择压缩策略
  selectStrategy(
    messages: Message[],
    budget: ContextBudget
  ): CompressionPlan {
    const currentTokens = this.tokenCounter.countMessages(messages);
    const excessTokens = currentTokens - budget.conversation;
    
    if (excessTokens <= 0) {
      // 不需要压缩
      return { type: 'none' };
    }
    
    const excessRatio = excessTokens / currentTokens;
    
    if (excessRatio < 0.2) {
      // 超出 20% 以内：微压缩
      return {
        type: 'micro',
        targetReduction: excessTokens,
      };
    }
    
    if (excessRatio < 0.5) {
      // 超出 20-50%：智能压缩
      return {
        type: 'smart',
        targetReduction: excessTokens,
        preserveLastN: 20,  // 保留最近 20 条
      };
    }
    
    // 超出 50% 以上：激进压缩 + 记忆提取
    return {
      type: 'aggressive',
      targetReduction: excessTokens,
      preserveLastN: 10,
      extractMemories: true,
    };
  }
  
  // 执行微压缩 (移除旧消息)
  async microCompact(
    messages: Message[],
    targetReduction: number
  ): Promise<CompactResult> {
    const removed: Message[] = [];
    let reducedTokens = 0;
    
    // 从最旧的消息开始移除
    for (const message of messages) {
      if (reducedTokens >= targetReduction) {
        break;
      }
      
      const tokens = this.tokenCounter.count(message.content);
      removed.push(message);
      reducedTokens += tokens;
    }
    
    const remaining = messages.slice(removed.length);
    
    return {
      removed,
      remaining,
      reducedTokens,
      method: 'micro',
    };
  }
  
  // 执行智能压缩 (AI 摘要)
  async smartCompact(
    messages: Message[],
    preserveLastN: number
  ): Promise<CompactResult> {
    // 保留最近 N 条消息
    const toPreserve = messages.slice(-preserveLastN);
    const toSummarize = messages.slice(0, -preserveLastN);
    
    if (toSummarize.length === 0) {
      return {
        removed: [],
        remaining: messages,
        reducedTokens: 0,
        method: 'smart',
      };
    }
    
    // 生成摘要
    const summary = await this.generateSummary(toSummarize);
    
    // 创建摘要消息
    const summaryMessage: Message = {
      id: generateId(),
      role: 'system',
      content: `之前的对话摘要：\n${summary}`,
      timestamp: new Date(),
      metadata: {
        type: 'summary',
        originalMessageCount: toSummarize.length,
        compressedAt: new Date(),
      },
    };
    
    const originalTokens = this.tokenCounter.countMessages(toSummarize);
    const newTokens = this.tokenCounter.count(summaryMessage.content);
    
    return {
      removed: toSummarize,
      remaining: [summaryMessage, ...toPreserve],
      reducedTokens: originalTokens - newTokens,
      method: 'smart',
      summary: summaryMessage.content,
    };
  }
  
  // 生成对话摘要
  private async generateSummary(messages: Message[]): Promise<string> {
    const conversationText = messages
      .map(m => `${m.role}: ${m.content}`)
      .join('\n');
    
    const prompt = `
请总结以下对话的关键信息：

${conversationText}

提取以下内容：
1. 已完成的任务和结论
2. 重要的上下文信息 (技术栈、项目结构、用户偏好等)
3. 待办事项和未完成的工作
4. 关键决策和原因

用简洁的要点格式输出。
`.trim();
    
    const response = await this.llm.generate(prompt, {
      maxTokens: 1000,
      temperature: 0.3,
    });
    
    return response.content;
  }
  
  // 提取记忆
  async extractMemories(messages: Message[]): Promise<Memory[]> {
    const conversationText = messages
      .map(m => `${m.role}: ${m.content}`)
      .join('\n');
    
    const prompt = `
从以下对话中提取值得长期记住的信息：

${conversationText}

提取为记忆条目，每条记忆包含：
- 内容：具体的事实或信息
- 类型：用户偏好 / 项目上下文 / 技术决策 / 其他
- 重要性：1-5 分

返回 JSON 数组格式：
[
  {
    "content": "记忆内容",
    "type": "user_preference",
    "importance": 4
  }
]
`.trim();
    
    const response = await this.llm.generate(prompt, {
      maxTokens: 2000,
      temperature: 0.3,
    });
    
    const memories = JSON.parse(response.content);
    return memories;
  }
}

interface CompressionPlan {
  type: 'none' | 'micro' | 'smart' | 'aggressive';
  targetReduction?: number;
  preserveLastN?: number;
  extractMemories?: boolean;
}

interface CompactResult {
  removed: Message[];
  remaining: Message[];
  reducedTokens: number;
  method: string;
  summary?: string;
}

上下文恢复

class ContextRestorer {
  private memoryStore: MemoryStore;
  private sessionStore: SessionStore;
  
  constructor(memoryStore: MemoryStore, sessionStore: SessionStore) {
    this.memoryStore = memoryStore;
    this.sessionStore = sessionStore;
  }
  
  // 恢复上下文
  async restore(
    sessionId: string,
    budget: ContextBudget
  ): Promise<Message[]> {
    // 1. 加载会话
    const session = await this.sessionStore.get(sessionId);
    
    if (!session) {
      throw new Error(`Session not found: ${sessionId}`);
    }
    
    // 2. 获取压缩后的消息
    const compressedMessages = session.messages;
    
    // 3. 查找相关记忆
    const relevantMemories = await this.findRelevantMemories(
      compressedMessages,
      budget
    );
    
    // 4. 注入记忆到上下文
    const messagesWithMemories = this.injectMemories(
      compressedMessages,
      relevantMemories
    );
    
    // 5. 验证 Token 预算
    const tokenCount = this.tokenCounter.countMessages(messagesWithMemories);
    
    if (tokenCount > budget.conversation) {
      // 仍然超出，需要进一步压缩
      return await this.furtherCompact(messagesWithMemories, budget);
    }
    
    return messagesWithMemories;
  }
  
  // 查找相关记忆
  private async findRelevantMemories(
    messages: Message[],
    budget: ContextBudget
  ): Promise<Memory[]> {
    // 从最近消息中提取关键词
    const recentMessages = messages.slice(-10);
    const keywords = this.extractKeywords(recentMessages);
    
    // 搜索相关记忆
    const memories = await this.memoryStore.search(keywords);
    
    // 限制记忆占用的 Token
    const memoryBudget = Math.round(budget.conversation * 0.1);  // 10%
    let memoryTokens = 0;
    const selected: Memory[] = [];
    
    for (const memory of memories) {
      const tokens = this.tokenCounter.count(memory.content);
      if (memoryTokens + tokens <= memoryBudget) {
        selected.push(memory);
        memoryTokens += tokens;
      }
    }
    
    return selected;
  }
  
  // 注入记忆到上下文
  private injectMemories(
    messages: Message[],
    memories: Memory[]
  ): Message[]: Message[] {
    if (memories.length === 0) {
      return messages;
    }
    
    const memoryContext = `
<相关记忆>
${memories.map(m => `- [${m.type}] ${m.content}`).join('\n')}
</相关记忆>
`.trim();
    
    const memoryMessage: Message = {
      id: generateId(),
      role: 'system',
      content: memoryContext,
      timestamp: new Date(),
      metadata: { type: 'memory_context' },
    };
    
    // 在第一条消息前插入记忆
    return [memoryMessage, ...messages];
  }
  
  // 提取关键词
  private extractKeywords(messages: Message[]): string[] {
    const text = messages.map(m => m.content).join(' ');
    
    // 简单的关键词提取 (实际可用更复杂的算法)
    const words = text.toLowerCase().match(/\b[a-z]{4,}\b/g) || [];
    
    // 统计词频
    const freq: Record<string, number> = {};
    for (const word of words) {
      freq[word] = (freq[word] || 0) + 1;
    }
    
    // 返回高频词
    return Object.entries(freq)
      .sort((a, b) => b[1] - a[1])
      .slice(0, 20)
      .map(([word]) => word);
  }
}

---

设计思想：为什么这样设计

思想 1：分级压缩

问题： 一刀切的压缩会丢失重要信息。

解决： 根据超出程度分级压缩。

超出程度     压缩策略        Token 回收     信息保留
─────────────────────────────────────────────────────
<20%        微压缩         10-20%        95%+
20-50%      智能压缩       40-60%        80-90%
>50%        激进压缩       60-80%        60-70%

设计智慧：

压缩程度与信息价值成正比。

思想 2：记忆提取

问题： 压缩会丢失长期有价值的信息。

解决： 提取到记忆系统。

// 压缩前
对话历史 (100 条消息)
   ↓
提取记忆
   ↓
记忆系统：
- 用户偏好：TypeScript
- 项目结构：Monorepo
- 部署目标：Kubernetes
   ↓
压缩后的对话 (20 条消息 + 摘要)

记忆类型：

类型	示例	保留时间
用户偏好	“使用 TypeScript”	永久
项目上下文	“Monorepo 结构”	项目周期
技术决策	“选择 PostgreSQL”	长期
临时信息	“今天的待办”	短期

思想 3：按需恢复

问题： 压缩后信息无法访问。

解决： 按需恢复。

// 用户问："我之前说过要用什么语言？"

AI: 检索记忆系统
   → 找到："用户偏好：TypeScript"
   
AI: "您之前提到过使用 TypeScript。"

恢复策略：

1. 关键词触发 - 检测到相关关键词时恢复
2. 语义检索 - 基于语义相似度检索
3. 主动注入 - 在系统提示中注入相关记忆

思想 4：预算感知

问题： 压缩操作本身消耗 Token。

解决： 预算感知压缩。

// 压缩前计算预算
const budget = tokenCounter.calculateBudget(
  maxTokens,
  systemPrompt,
  tools
);

// 压缩目标：回到预算内
const targetTokens = budget.conversation * 0.8;  // 留 20% 余量

// 压缩操作本身的 Token 消耗计入预算
const compressionCost = estimateCompressionCost(messages);
const availableForCompression = budget.conversation - compressionCost;

思想 5：渐进式压缩

问题： 一次性压缩可能过度。

解决： 渐进式压缩。

初始状态：150K tokens (超出 22K)
   ↓
微压缩：移除最旧的 10 条消息
   ↓
检查：135K tokens (仍在预算内)
   ↓
停止压缩

而不是：
一次性压缩到 100K tokens (过度压缩)

---

解决方案：完整实现详解

ContextManager 实现

export class ContextManager {
  private tokenCounter: TokenCounter;
  compressionStrategy: CompressionStrategy;
  private restorer: ContextRestorer;
  
  constructor(config: ContextManagerConfig) {
    this.tokenCounter = new TokenCounter();
    this.compressionStrategy = new CompressionStrategy(
      config.llm,
      this.tokenCounter
    );
    this.restorer = new ContextRestorer(
      config.memoryStore,
      config.sessionStore
    );
  }
  
  // 添加消息并管理上下文
  async addMessage(
    sessionId: string,
    message: Message,
    budget: ContextBudget
  ): Promise<ContextUpdateResult> {
    // 1. 加载当前上下文
    const session = await this.sessionStore.get(sessionId);
    const messages = session?.messages || [];
    
    // 2. 预估添加后的 Token 数
    const estimatedTokens = this.tokenCounter.estimateWithNewMessage(
      messages,
      message.content
    );
    
    // 3. 检查是否需要压缩
    if (estimatedTokens > budget.conversation) {
      // 执行压缩
      const compactResult = await this.compress(messages, budget);
      
      // 添加新消息
      compactResult.remaining.push(message);
      
      // 保存
      await this.sessionStore.update({
        ...session,
        messages: compactResult.remaining,
      });
      
      return {
        compressed: true,
        removedMessages: compactResult.removed.length,
        reducedTokens: compactResult.reducedTokens,
        extractedMemories: compactResult.memories?.length || 0,
      };
    }
    
    // 不需要压缩，直接添加
    messages.push(message);
    await this.sessionStore.update({ ...session, messages });
    
    return {
      compressed: false,
      currentTokens: estimatedTokens,
      remainingBudget: budget.conversation - estimatedTokens,
    };
  }
  
  // 恢复上下文
  async restoreContext(
    sessionId: string,
    budget: ContextBudget
  ): Promise<Message[]> {
    return await this.restorer.restore(sessionId, budget);
  }
  
  // 手动压缩
  async compress(
    messages: Message[],
    budget: ContextBudget
  ): Promise<CompactResult> {
    // 选择压缩策略
    const plan = this.compressionStrategy.selectStrategy(messages, budget);
    
    // 执行压缩
    let result: CompactResult;
    
    switch (plan.type) {
      case 'none':
        return {
          removed: [],
          remaining: messages,
          reducedTokens: 0,
          method: 'none',
        };
      
      case 'micro':
        result = await this.compressionStrategy.microCompact(
          messages,
          plan.targetReduction!
        );
        break;
      
      case 'smart':
        result = await this.compressionStrategy.smartCompact(
          messages,
          plan.preserveLastN!
        );
        break;
      
      case 'aggressive':
        result = await this.compressionStrategy.smartCompact(
          messages,
          plan.preserveLastN!
        );
        
        // 提取记忆
        if (plan.extractMemories) {
          result.memories = await this.compressionStrategy.extractMemories(
            result.removed
          );
        }
        break;
    }
    
    return result;
  }
}

interface ContextUpdateResult {
  compressed: boolean;
  removedMessages?: number;
  reducedTokens?: number;
  extractedMemories?: number;
  currentTokens?: number;
  remainingBudget?: number;
}

配置示例

# ~/.openclaw/config/context.yaml

# Token 预算
budget:
  # 最大 Token 数 (根据模型调整)
  max_tokens: 128000
  
  # 预留比例 (给响应)
  reserved_ratio: 0.2
  
  # 系统提示预算
  system_prompt_budget: 2000
  
  # 工具预算
  tools_budget: 5000

# 压缩配置
compression:
  # 启用自动压缩
  auto_compress: true
  
  # 压缩触发阈值 (使用率超过此值触发)
  trigger_threshold: 0.8
  
  # 压缩目标 (压缩后使用率)
  target_usage: 0.6
  
  # 保留最近消息数
  preserve_recent: 20

# 记忆提取
memory:
  # 启用记忆提取
  extract_on_compress: true
  
  # 记忆重要性阈值
  importance_threshold: 3
  
  # 记忆注入预算 (占上下文预算的比例)
  injection_budget: 0.1

# 恢复配置
restore:
  # 启用记忆注入
  inject_memories: true
  
  # 相关记忆数量限制
  max_memories: 10
  
  # 记忆格式
  format: 'system_message'

---

OpenClaw 最佳实践

实践 1：查看上下文状态

# 查看当前上下文使用
openclaw run context --action status

# 输出：
上下文状态:
─────────────────────────────────────
最大 Token:     128,000
已用 Token:     95,000 (74%)
可用 Token:     33,000

消息数量：85
最后压缩：2 小时前

实践 2：手动压缩

# 压缩上下文
openclaw run context --action compact

# 输出：
压缩完成:
- 移除消息：45 条
- 回收 Token: 25,000
- 提取记忆：5 条
- 当前使用：70,000 (55%)

实践 3：查看提取的记忆

# 列出记忆
openclaw run memory --action list

# 输出：
已提取记忆 (共 15 条):

┌───────────────────────────────────────────┐
│ 类型          │ 内容              │ 重要性 │
├───────────────────────────────────────────┤
│ 用户偏好      │ 使用 TypeScript   │ ⭐⭐⭐⭐⭐│
│ 项目上下文    │ Monorepo 结构     │ ⭐⭐⭐⭐ │
│ 技术决策      │ 选择 PostgreSQL   │ ⭐⭐⭐⭐ │
│ 用户偏好      │ 偏好函数式编程    │ ⭐⭐⭐  │
└───────────────────────────────────────────┘

实践 4：恢复压缩的上下文

# 恢复特定会话的上下文
openclaw run context --action restore --sessionId session-001

# 输出：
上下文已恢复:
- 压缩消息：65 条
- 摘要消息：1 条
- 注入记忆：5 条
- 总 Token: 75,000

实践 5：配置压缩策略

# ~/.openclaw/config/context-custom.yaml

# 保守策略 (保留更多信息)
compression_conservative:
  trigger_threshold: 0.9
  target_usage: 0.75
  preserve_recent: 30

# 激进策略 (更多压缩)
compression_aggressive:
  trigger_threshold: 0.7
  target_usage: 0.5
  preserve_recent: 10

---

总结

核心要点

1. 分级压缩 - 根据超出程度选择策略
2. 记忆提取 - 保留长期有价值的信息
3. 按需恢复 - 压缩后仍可检索
4. 预算感知 - 压缩操作本身计入成本
5. 渐进式压缩 - 避免过度压缩

设计智慧

好的上下文管理让 AI”记住重要的，忘记次要的”。

Claude Code 的上下文管理设计告诉我们：

• Token 限制不是问题，智能压缩是关键
• 记忆提取保证长期连续性
• 按需恢复平衡存储与访问

压缩效果

压缩策略	Token 回收	信息保留	适用场景
微压缩	10-20%	95%+	轻微超出
智能压缩	40-60%	80-90%	中度超出
激进压缩	60-80%	60-70%	严重超出

下一步

• 实现 Token 计数器
• 配置压缩策略
• 启用记忆提取
• 添加上下文监控

---

系列文章：

• [1] Bash 命令执行的安全艺术 (已发布)
• [2] 差异编辑的设计艺术 (已发布)
• [3] 文件搜索的底层原理 (已发布)
• [4] 多 Agent 协作的架构设计 (已发布)
• [5] 技能系统的设计哲学 (已发布)
• [6] MCP 协议集成的完整指南 (已发布)
• [7] 后台任务管理的完整方案 (已发布)
• [8] Web 抓取的 SSRF 防护设计 (已发布)
• [9] 多层权限决策引擎设计 (已发布)
• [10] 插件生命周期的设计智慧 (已发布)
• [11] 会话持久化的架构设计 (已发布)
• [12] 上下文压缩与恢复技术 (本文)
• [13+] 更多高级功能 (继续中…)

---

进度：12/N

上一篇： Claude Code 源码解析 (11)：会话持久化的架构设计

---

关于作者： John，OpenClaw 平台开发者，专注 AI 助手架构设计与实现。