Agent 上下文管理实战：三层渐进式压缩

Context Window 是有限资源，如何让它持续服务长时间对话？

---

为什么需要上下文管理

问题

问题	后果
Token 溢出	超过模型限制，API 报错
注意力退化	Lost-in-Middle，中间信息召回率低 10-40%
成本暴增	每个 Token 都要付费
响应变慢	更多 Token = 更长推理时间

解决方案：三层渐进式压缩

┌─────────────────────────────────────────┐
│          上下文管理层次                   │
├─────────────────────────────────────────┤
│ Level 1: Prune    → 丢弃旧工具输出        │
│ Level 2: Compact  → 生成总结性消息        │
│ Level 3: Summarize → 会话级摘要           │
└─────────────────────────────────────────┘

---

Level 1：修剪 (Prune)

丢弃什么

工具输出：工具执行结果通常很长，但价值随时间递减。

// 修剪策略
function prune(messages: Message[]): Message[] {
  return messages.map(msg => {
    if (msg.role === 'tool') {
      // 保留前 500 字符，丢弃其余
      return {
        ...msg,
        content: msg.content.slice(0, 500) + '\n...[trimmed]'
      }
    }
    return msg
  })
}

保留什么

• 用户原始请求（不能丢）
• 助手的决策和推理（保留）
• 最近的工具输出（保留）
• 旧的、已处理的工具输出（可修剪）

---

Level 2：压缩 (Compact)

触发时机

// 检测是否需要压缩
async function isOverflow(tokens: TokenUsage, model: Model): boolean {
  const context = model.contextWindow    // 比如 200K
  const output = Math.min(model.maxOutput, 20000)  // 预留输出
  const usable = context - output

  const used = tokens.input + tokens.cached + tokens.output
  return used > usable
}

压缩策略

将多条消息压缩成一条总结性消息：

// 压缩前：10 条消息，5000 tokens
const before = [
  { role: 'user', content: '帮我分析这个 bug' },
  { role: 'assistant', content: '好的，我先...' },
  { role: 'tool', content: '很长很长的输出...' },
  // ... 更多消息
]

// 压缩后：1 条消息，500 tokens
const after = [
  {
    role: 'system',
    content: `[历史摘要] 用户请求分析 bug，助手检查了代码，
              发现问题在 async 函数的错误处理。当前状态：已定位问题。`
  }
]

Token 估算

// 简单估算：4 字符 ≈ 1 token
function estimateToken(text: string): number {
  return Math.ceil(text.length / 4)
}

---

Level 3：总结 (Summarize)

会话级摘要

当压缩也不够时，生成整个会话的摘要：

interface SessionSummary {
  mainTask: string        // 主要任务
  completed: string[]     // 已完成
  pending: string[]       // 待处理
  decisions: string[]     // 关键决策
  context: string         // 当前上下文
}

摘要生成

async function summarize(messages: Message[]): Promise<string> {
  const prompt = `
请总结以下对话的关键信息：
1. 主要任务是什么
2. 已完成什么
3. 当前状态
4. 待处理事项

对话历史：
${messages.map(m => `${m.role}: ${m.content}`).join('\n')}
`
  return await llm.complete(prompt)
}

---

实战：实现一个简单的上下文管理器

class ContextManager {
  private maxTokens: number
  private reserveTokens: number

  constructor(maxTokens: number = 200000, reserveTokens: number = 20000) {
    this.maxTokens = maxTokens
    this.reserveTokens = reserveTokens
  }

  // 估算 token 数量
  estimateTokens(text: string): number {
    return Math.ceil(text.length / 4)
  }

  // 检查是否需要压缩
  needsCompaction(messages: Message[]): boolean {
    const total = messages.reduce((sum, msg) =>
      sum + this.estimateTokens(msg.content), 0
    )
    return total > this.maxTokens - this.reserveTokens
  }

  // Level 1: 修剪工具输出
  prune(messages: Message[], maxLength: number = 500): Message[] {
    return messages.map(msg => {
      if (msg.role === 'tool' && msg.content.length > maxLength) {
        return {
          ...msg,
          content: msg.content.slice(0, maxLength) + '\n...[trimmed]'
        }
      }
      return msg
    })
  }

  // Level 2: 压缩历史
  async compact(messages: Message[], llm: LLM): Promise<Message[]> {
    if (!this.needsCompaction(messages)) {
      return messages
    }

    // 先修剪
    let processed = this.prune(messages)

    // 如果还是太长，生成摘要
    if (this.needsCompaction(processed)) {
      const summary = await this.generateSummary(processed, llm)
      processed = [
        { role: 'system', content: `[历史摘要]\n${summary}` },
        ...processed.slice(-3)  // 保留最近 3 条
      ]
    }

    return processed
  }

  // 生成摘要
  private async generateSummary(messages: Message[], llm: LLM): Promise<string> {
    const history = messages
      .map(m => `${m.role}: ${m.content.slice(0, 200)}`)
      .join('\n')

    const prompt = `总结以下对话的关键信息（200字以内）：

${history}

摘要：`

    return await llm.complete(prompt)
  }
}

使用示例

const manager = new ContextManager(200000, 20000)
const messages: Message[] = []

// 添加消息
messages.push({ role: 'user', content: '...' })
messages.push({ role: 'assistant', content: '...' })
messages.push({ role: 'tool', content: '...' })

// 压缩前检查
if (manager.needsCompaction(messages)) {
  const compacted = await manager.compact(messages, llm)
  // 使用压缩后的消息继续对话
}

---

最佳实践

1. 预留输出 Token

// ❌ 错误：用满整个 context window
const usable = model.contextWindow

// ✅ 正确：预留输出空间
const usable = model.contextWindow - model.maxOutput

2. 优先级保留

用户请求 > 助手决策 > 最近工具输出 > 旧工具输出

3. 分层压缩

Prune → Compact → Summarize

先用轻量方法，不够再用重方法

4. 监控 Token 使用

// 每次 API 调用后记录
function trackUsage(usage: TokenUsage) {
  console.log(`Input: ${usage.input}, Output: ${usage.output}`)
  console.log(`Usage: ${usage.input + usage.output} / ${model.contextWindow}`)
}

---

常见问题

Q: 压缩会丢失信息吗？

会。但：

• 旧信息的价值随时间递减
• 摘要保留了关键信息
• 用户可以重新提问

Q: 什么时候触发压缩？

通常在：

• Token 使用超过 80%
• 对话超过 50 轮
• 模型返回 context overflow 错误

Q: 如何评估压缩效果？

指标：

• Token 减少量
• 对话连贯性（主观）
• 任务完成率

---

参考资料

• OpenCode 源码 - 上下文管理实现
• Lost in the Middle - 注意力退化研究
• Claude Context Windows - 官方文档
• OpenCode Context Management - compaction.ts 源码