Agent 上下文压缩策略：长对话下的记忆裁剪与摘要

一、对话超过 30 轮后，Agent 开始答非所问

Agent 的上下文窗口不是无限的。
即使 GPT-4 支持 128K Token，长对话仍会出问题。
上下文过长导致三个后果。
一是注意力稀释：模型开始忽略早期关键信息。
二是成本飙升：每轮追加全部历史，Token 消耗指数增长。
三是延迟升高：长上下文让推理时间从 2 秒变成 8 秒。

一个客服 Agent 的真实案例。
用户和 Agent 聊了 40 轮，描述了问题、提供了截图。
但 Agent 在第 41 轮时忘记了用户最初说的产品型号。
因为型号信息在对话开头，已被注意力机制稀释。

上下文压缩的本质是取舍。
保留什么信息？丢弃什么信息？怎么保留？

二、记忆管理的三层架构

Agent 的记忆系统分三个层次。
工作记忆：当前对话窗口，最新的 N 轮。
摘要记忆：对历史对话的结构化压缩。
长期记忆：跨会话的关键信息持久化。

flowchart TB
    A[用户新消息] --> B[当前对话缓冲区]
    B --> C{缓冲区满?}
    C -->|否| D[直接追加上下文]
    C -->|是| E[触发压缩]
    
    E --> F[滑动窗口: 保留最近 K 轮]
    F --> G[旧对话送入压缩器]
    G --> H[生成结构化摘要]
    H --> I[提取关键实体]
    
    I --> J[摘要存储]
    J --> K{关键实体变化?}
    K -->|是| L[更新长期记忆]
    K -->|否| M[保留现有记忆]
    
    L --> D
    M --> D

关键设计决策：
窗口大小：保留最近 10-15 轮完整对话。
摘要粒度：每 10 轮旧对话生成一次摘要。
实体提取：人名、产品名、问题类型等关键信息。

三、上下文压缩的 Python 实现

"""
context_compressor.py - Agent 上下文压缩
"""

import hashlib
import json
import logging
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from datetime import datetime

logger = logging.getLogger(__name__)


@dataclass
class Message:
    """对话消息"""
    role: str          # user / assistant / tool
    content: str
    timestamp: float = field(default_factory=lambda: datetime.now().timestamp())


@dataclass
class ConversationSummary:
    """对话摘要"""
    summary_text: str            # 自然语言摘要
    key_entities: Dict[str, str] # 提取的关键实体
    decisions: List[str]         # 做出的决策
    pending_items: List[str]     # 未完成事项
    compressed_at: float         # 压缩时间戳


class ContextCompressor:
    """Agent 上下文压缩器"""

    def __init__(
        self,
        max_recent_rounds: int = 10,
        compress_every: int = 15,     # 每 15 轮旧对话触发压缩
        max_summary_length: int = 500, # 摘要最大长度（字符）
    ):
        self.max_recent_rounds = max_recent_rounds
        self.compress_every = compress_every
        self.max_summary_length = max_summary_length

        self.buffer: List[Message] = []         # 当前对话缓冲
        self.summaries: List[ConversationSummary] = []  # 历史摘要
        self.long_term: Dict[str, str] = {}      # 长期记忆

    def add_message(self, msg: Message):
        """添加新消息，自动触发压缩"""
        self.buffer.append(msg)

        if len(self.buffer) > self.max_recent_rounds + self.compress_every:
            self._compress()

    def _compress(self):
        """压缩旧消息为摘要"""
        # 保留最近 max_recent_rounds 条
        recent = self.buffer[-self.max_recent_rounds:]
        old = self.buffer[:-self.max_recent_rounds]

        if not old:
            return

        # 生成摘要（生产环境调用 LLM）
        summary = self._generate_summary(old)

        self.summaries.append(summary)
        self.buffer = recent

        # 更新长期记忆
        for key, value in summary.key_entities.items():
            self.long_term[key] = value

        logger.info(
            f"压缩完成：{len(old)} 条消息 → 摘要 "
            f"({len(summary.summary_text)} 字符)"
        )

    def _generate_summary(
        self, messages: List[Message]
    ) -> ConversationSummary:
        """
        生成对话摘要
        生产环境应调用 LLM 生成，此处为简化示例
        """
        # 提取关键信息（简化实现）
        entities = {}
        decisions = []
        pending = []

        conversation_text = ""
        for msg in messages:
            conversation_text += f"[{msg.role}] {msg.content}\n"

        # 简化的实体提取（正则匹配）
        import re
        # 查找产品名、ID等模式
        product_matches = re.findall(
            r'(产品|型号|订单)[：:\s]*([A-Za-z0-9_-]+)',
            conversation_text
        )
        for match in product_matches:
            entities[match[0]] = match[1]

        # 查找决策
        decision_matches = re.findall(
            r'(决定|确认|选择)[：:\s]*(.+)',
            conversation_text
        )
        for match in decision_matches:
            decisions.append(f"{match[0]}：{match[1]}")

        # 生成摘要文本（生产环境用 LLM）
        summary_text = (
            f"历史对话讨论要点："
            f"涉及 {len(messages)} 条消息，"
            f"提取关键实体 {len(entities)} 项，"
            f"做出 {len(decisions)} 项决策。"
        )

        return ConversationSummary(
            summary_text=summary_text,
            key_entities=entities,
            decisions=decisions,
            pending_items=pending,
            compressed_at=datetime.now().timestamp(),
        )

    def build_context(self) -> str:
        """构建发送给 LLM 的完整上下文"""
        parts = []

        # 1. 长期记忆
        if self.long_term:
            parts.append("## 重要信息")
            for key, value in self.long_term.items():
                parts.append(f"- {key}: {value}")
            parts.append("")

        # 2. 历史摘要
        if self.summaries:
            parts.append("## 历史对话摘要")
            for i, summary in enumerate(self.summaries):
                parts.append(f"### 阶段{i+1}")
                parts.append(summary.summary_text[:self.max_summary_length])
            parts.append("")

        # 3. 最近对话
        if self.buffer:
            parts.append("## 最近对话")
            for msg in self.buffer:
                parts.append(f"**{msg.role}**: {msg.content}")
            parts.append("")

        return "\n".join(parts)

    def token_count(self) -> int:
        """估算 Token 数（约 4 字符 = 1 Token）"""
        context = self.build_context()
        return len(context) // 4

    def reset(self):
        """重置上下文"""
        self.buffer = []
        self.summaries = []
        self.long_term = {}


# ---- 使用示例 ----
def demo():
    compressor = ContextCompressor(max_recent_rounds=5, compress_every=10)

    # 模拟长对话
    messages = [
        Message("user", "我的产品型号是 X-200，出现了开机蓝屏问题"),
        Message("assistant", "了解了，X-200 蓝屏通常与驱动有关。请问最近更新了什么?"),
        Message("user", "昨天装了新的显卡驱动"),
        Message("assistant", "建议回滚驱动。先进入安全模式..."),
    ] * 5  # 模拟 20 轮对话

    for msg in messages:
        compressor.add_message(msg)

    print("压缩后的上下文:")
    print(compressor.build_context())
    print(f"\n估算 Token 数: {compressor.token_count()}")

四、压缩策略的边界与局限

摘要质量高度依赖 LLM 能力。
弱模型生成的摘要可能遗漏关键信息。
建议在关键场景使用更强的模型做摘要。

压缩可能丢失上下文完整度。
"上次说的那个方案"这种指代会断裂。
需要在摘要中保留明确的实体引用。

不适合压缩的场景：
短对话（< 10 轮）没必要压缩；
法律合规场景需要完整审计记录；
实时流式输出的聊天场景。

五、总结

长对话下 Agent 需要上下文压缩保持质量。
三层记忆架构：工作记忆、摘要记忆、长期记忆。
滑动窗口保留最近对话，旧对话压缩为结构化摘要。
压缩时提取关键实体和决策，避免信息丢失。
Token 消耗从 O(n) 降为 O(k)，k 为窗口大小。