Agent 记忆架构：滑动窗口、Token 预算裁剪与分层摘要压缩

本节目标

为什么「上下文窗口变大了」依然需要精细的记忆管理？

尽管现代大模型的最大上下文窗口已扩展到 128K 甚至 1M Token，但在 Agent 工程中，粗暴地塞满上下文会引发三大问题：

  1. 成本与首字延迟（TTFT）随轮次呈二次方增长：假设每步工具返回 2,000 Token，跑 15 步下来累计输入 Token 超过 $2000 imes (1 + 2 + \dots + 15) = 240,000$ Token。
  2. 注意力稀释（Attention Dilution）：当上下文塞满大量过期的工具原始 JSON 输出时，模型对 System Prompt 核心约束和最新指令的遵循度显著下降。
  3. 裁剪破坏协议完整性：如果你简单地用 messages.slice(-10) 截取最近 10 条消息，一旦第 1 条恰好是一条孤立的 role: 'tool' 消息（其对应的 assistant.tool_calls 被切掉了），大模型 API 会直接报 400 错误拒收。
❌ 错误裁剪（切断了原子消息组，触发 API 400 报错）：
  ...被裁掉: [assistant (tool_calls: call_1)]
  保留窗口: [tool (tool_call_id: call_1)] ──► 孤儿 tool 消息！协议非法！

✅ 正确裁剪（按原子对话轮 Turn Group 裁剪 + 滚动摘要）：
  [systemPrompt] + [system: 历史滚动摘要] + [完整的近期原子消息组...]

手写协议安全的分层记忆管理器

下面的代码将消息流划分为不可拆分的原子单元（Atomic Turn Groups），并在超出 Token 预算时自动将旧单元提炼为增量摘要：

export class ConversationMemory {
  constructor({ systemPrompt, maxBudgetTokens = 6000, summarizer }) {
    this.systemPrompt = systemPrompt
    this.maxBudgetTokens = maxBudgetTokens
    this.summarizer = summarizer // async (oldSummary, droppedMessages) => newSummary
    this.rollingSummary = ''
    this.history = []
  }

  // 轻量级中英混合 Token 估算器（中文约 1.5 字/Token，英文/代码约 4 字符/Token）
  estimateTokens(text = '') {
    const str = typeof text === 'string' ? text : JSON.stringify(text)
    let cjk = 0
    let ascii = 0
    for (const ch of str) {
      if (ch.charCodeAt(0) > 0x7f) cjk++
      else ascii++
    }
    return Math.ceil(cjk / 1.5 + ascii / 4) + 4
  }

  // 将扁平的 messages 聚合成不可从中间劈开的「原子消息组」
  // 规则：一条含 tool_calls 的 assistant 消息必须与紧随其后的所有 role:'tool' 消息绑在同一组
  #groupIntoAtomicUnits(messages) {
    const groups = []
    let i = 0
    while (i < messages.length) {
      const msg = messages[i]
      if (msg.role === 'assistant' && Array.isArray(msg.tool_calls) && msg.tool_calls.length > 0) {
        const group = [msg]
        i++
        while (i < messages.length && messages[i].role === 'tool') {
          group.push(messages[i])
          i++
        }
        groups.push(group)
      } else {
        groups.push([msg])
        i++
      }
    }
    return groups
  }

  async appendAndCompact(newMessages) {
    this.history.push(...newMessages)

    const groups = this.#groupIntoAtomicUnits(this.history)
    const sysTokens = this.estimateTokens(this.systemPrompt) + this.estimateTokens(this.rollingSummary)
    let currentTokens = sysTokens + groups.reduce((sum, g) => sum + this.estimateTokens(g), 0)

    if (currentTokens <= this.maxBudgetTokens || groups.length <= 2) {
      return this.getMessages()
    }

    // 从最旧的原子组开始弹出，但至少保留最后 2 个最新交互组
    const evictedMessages = []
    while (currentTokens > this.maxBudgetTokens && groups.length > 2) {
      const oldestGroup = groups.shift()
      evictedMessages.push(...oldestGroup)
      currentTokens -= this.estimateTokens(oldestGroup)
    }

    this.history = groups.flat()

    // 将被淘汰的旧对话增量融合进 rollingSummary，保留关键事实与已确认结论
    if (evictedMessages.length > 0 && this.summarizer) {
      this.rollingSummary = await this.summarizer(this.rollingSummary, evictedMessages)
    }

    return this.getMessages()
  }

  getMessages() {
    const out = [{ role: 'system', content: this.systemPrompt }]
    if (this.rollingSummary) {
      out.push({
        role: 'system',
        content: `【前序对话关键事实与工具结论摘要】\n${this.rollingSummary}`
      })
    }
    out.push(...this.history)
    return out
  }
}

课后练习

练习 1：为什么在 #groupIntoAtomicUnits 中，必须把含有 tool_calls 的 assistant 消息和紧随其后的所有 role: 'tool' 消息打包成同一个不可分割的数组单元？

答案：大模型对话接口强制要求 assistant.tool_calls 与对应的 tool 响应消息成对出现。如果滑动窗口刚好从它们中间截断，剩下的孤立 tool 消息（或缺少 tool 响应的 assistant.tool_calls 消息）会导致后续所有 API 请求直接报 400 协议错误。

练习 2：在压缩超长工具返回结果（如 500 行的文件搜索输出）时，为什么「滚动摘要（Rolling Summary）」比直接丢弃旧消息更好？

答案：直接丢弃旧消息会让 Agent 患上「顺行性遗忘」——它忘了自己在第 2 步已经确认过某个配置项的值，从而在第 8 步又去重复读取同一个文件；滚动摘要用不到 5% 的 Token 保留了「已查过什么、得出了什么关键结论」，兼顾了上下文预算与长期一致性。