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

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_factorylambda: 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_textsummary_text, key_entitiesentities, decisionsdecisions, pending_itemspending, compressed_atdatetime.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### 阶段{i1}) 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_rounds5, compress_every10) # 模拟长对话 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 为窗口大小。