对话越聊越蠢AI Agent 长对话记忆管理的工程化方案一、Token 账单与上下文遗忘长对话 Agent 的双重困境大模型 Agent 落地到真实业务后第一个撞上的墙不是模型能力不够而是对话管理出了问题。具体表现有两类一类是越聊越贵——每轮对话把完整历史塞进 PromptToken 消耗随轮次线性增长10 轮对话的 Token 成本是第 1 轮的 5 倍以上另一类是越聊越蠢——上下文窗口被早期无关内容占满模型对近期关键信息的注意力被稀释回复质量明显下降。以一个客服 Agent 为例用户前 3 轮咨询退货政策第 4 轮开始讨论换货流程到第 8 轮时模型已经忘了用户在第 2 轮提到的订单号。这不是模型能力问题而是记忆管理策略缺失导致的工程问题。长对话记忆管理的本质是在有限的上下文窗口内用最低的 Token 成本保留对当前任务最有价值的信息。这不是一个算法问题而是一个工程架构问题。二、记忆分层架构从工作记忆到长期存储人类大脑的记忆系统分为工作记忆、短期记忆和长期记忆三层。Agent 的记忆架构可以类比设计但需要适配 LLM 的技术约束。graph TB subgraph 记忆分层架构 WM[工作记忆 Working Memorybr/当前对话上下文窗口] SM[短期记忆 Short-term Memorybr/近期对话摘要 关键实体] LM[长期记忆 Long-term Memorybr/向量数据库 结构化知识] end User[用户输入] -- WM WM --|窗口溢出时压缩| SM SM --|关键信息持久化| LM LM --|检索召回| WM WM -- LLM[大模型推理] LLM -- Response[生成回复] Response -- WM style WM fill:#e1f5fe style SM fill:#fff3e0 style LM fill:#e8f5e9工作记忆Working Memory直接填充在 LLM Prompt 中的上下文受模型窗口大小限制。这是模型此刻能看到的全部信息。管理策略的核心是保留什么、丢弃什么、如何压缩。短期记忆Short-term Memory最近 N 轮对话的摘要和关键实体提取。当工作记忆窗口即将溢出时将早期对话压缩为摘要存入短期记忆释放 Token 空间。摘要不是简单的截断而是保留语义关键信息的压缩。长期记忆Long-term Memory持久化存储在向量数据库中的历史知识。当用户提到上次讨论的那个方案时通过语义检索从长期记忆中召回相关片段注入工作记忆。长期记忆的召回精度决定了 Agent 的记忆力。三、生产级记忆管理器实现3.1 核心数据结构from dataclasses import dataclass, field from typing import Optional import time import hashlib import json dataclass class Message: 单条对话消息 role: str # user / assistant / system content: str timestamp: float field(default_factorytime.time) token_count: int 0 msg_id: str def __post_init__(self): if not self.msg_id: # 用内容哈希 时间戳生成唯一 ID避免重复 raw f{self.role}:{self.content}:{self.timestamp} self.msg_id hashlib.md5(raw.encode()).hexdigest()[:12] dataclass class SummaryBlock: 对话摘要块多轮对话压缩后的语义保留 summary: str # 压缩后的摘要文本 original_msg_ids: list # 被压缩的原始消息 ID 列表 key_entities: list # 提取的关键实体订单号、人名等 token_count: int 0 created_at: float field(default_factorytime.time) dataclass class Entity: 关键实体需要在上下文中持续保留的信息 name: str # 实体名称 value: str # 实体值 source_msg_id: str # 来源消息 ID updated_at: float field(default_factorytime.time)3.2 记忆管理器核心逻辑class AgentMemoryManager: Agent 长对话记忆管理器 def __init__( self, max_working_tokens: int 6000, # 工作记忆 Token 上限 summary_threshold: float 0.8, # 触发压缩的阈值比例 max_recent_messages: int 6, # 始终保留的最近消息数 ): self.max_working_tokens max_working_tokens self.summary_threshold summary_threshold self.max_recent_messages max_recent_messages # 三层记忆 self.working_memory: list[Message] [] # 工作记忆 self.short_term_memory: list[SummaryBlock] [] # 短期记忆 self.entities: dict[str, Entity] {} # 关键实体表 def add_message(self, role: str, content: str, token_count: int) - None: 添加消息到工作记忆并检查是否需要压缩 msg Message(rolerole, contentcontent, token_counttoken_count) self.working_memory.append(msg) # 提取关键实体订单号、金额等结构化信息 self._extract_entities(msg) # 检查工作记忆是否接近溢出 current_tokens sum(m.token_count for m in self.working_memory) if current_tokens self.max_working_tokens * self.summary_threshold: self._compress_working_memory() def get_context_for_llm(self) - list[dict]: 组装发送给 LLM 的完整上下文 context [] # 1. 注入关键实体始终保留在上下文头部 if self.entities: entity_text 关键信息\n \n.join( f- {e.name}: {e.value} for e in self.entities.values() ) context.append({role: system, content: entity_text}) # 2. 注入短期记忆摘要 if self.short_term_memory: summary_parts [] for block in self.short_term_memory[-3:]: # 最近 3 个摘要块 summary_parts.append(block.summary) if summary_parts: summary_text 历史对话摘要\n \n.join(summary_parts) context.append({role: system, content: summary_text}) # 3. 注入工作记忆中的近期对话 for msg in self.working_memory: context.append({role: msg.role, content: msg.content}) return context def _compress_working_memory(self) - None: 压缩工作记忆将早期对话转为摘要保留最近消息 if len(self.working_memory) self.max_recent_messages: return # 消息数不够不需要压缩 # 分割待压缩部分 保留部分 split_idx len(self.working_memory) - self.max_recent_messages to_compress self.working_memory[:split_idx] to_keep self.working_memory[split_idx:] # 调用 LLM 生成摘要这里用简化逻辑示意 summary_text self._generate_summary(to_compress) # 收集被压缩消息中的关键实体 compressed_ids [m.msg_id for m in to_compress] related_entities [ e.name for e in self.entities.values() if e.source_msg_id in compressed_ids ] summary_block SummaryBlock( summarysummary_text, original_msg_idscompressed_ids, key_entitiesrelated_entities, token_countlen(summary_text) // 2, # 粗估 Token 数 ) self.short_term_memory.append(summary_block) self.working_memory to_keep def _generate_summary(self, messages: list[Message]) - str: 调用 LLM 对历史对话生成摘要 # 生产环境应调用 LLM API这里用简化逻辑 conversation \n.join( f{m.role}: {m.content[:100]} for m in messages ) # 实际调用llm_client.chat(请用200字以内总结以下对话的关键信息, conversation) return f[摘要] 对话涉及 {len(messages)} 轮交互关键实体已提取 def _extract_entities(self, msg: Message) - None: 从消息中提取关键实体订单号、金额等 import re # 订单号模式ORD-XXXX 或纯数字订单号 order_patterns [ rORD-\w, r订单号[:]\s*(\d), r订单\s*(\d{10,}), ] for pattern in order_patterns: matches re.findall(pattern, msg.content) for match in matches: entity_key forder_{match} self.entities[entity_key] Entity( name订单号, valuematch, source_msg_idmsg.msg_id, )3.3 长期记忆向量检索召回import numpy as np class LongTermMemory: 基于向量数据库的长期记忆存储 def __init__(self, embedding_dim: int 1536): self.embedding_dim embedding_dim # 生产环境使用 Milvus / Qdrant / Pinecone self.vectors: list[np.ndarray] [] self.documents: list[dict] [] def store(self, doc: dict, embedding: np.ndarray) - None: 存储文档及其向量表示 self.vectors.append(embedding) self.documents.append(doc) def recall(self, query_embedding: np.ndarray, top_k: int 3) - list[dict]: 根据查询向量召回最相关的文档 if not self.vectors: return [] # 余弦相似度计算 query_norm query_embedding / np.linalg.norm(query_embedding) similarities [] for vec in self.vectors: vec_norm vec / np.linalg.norm(vec) sim np.dot(query_norm, vec_norm) similarities.append(sim) # 取 Top-K top_indices np.argsort(similarities)[-top_k:][::-1] return [self.documents[i] for i in top_indices if similarities[i] 0.7] # 0.7 是召回阈值低于此值的不注入上下文避免噪声四、记忆管理的代价延迟、一致性与召回噪声记忆分层架构解决了 Token 成本和上下文遗忘问题但引入了新的工程权衡。摘要压缩的信息损失。LLM 生成的摘要必然丢失细节。当用户追问我之前说的那个具体金额是多少时摘要中可能已经没有这个数字。解决方案是关键实体金额、订单号、日期不进入摘要流程而是单独提取到实体表中始终保留在工作记忆头部。但这增加了实体提取的准确率依赖——提取遗漏的信息就真的丢了。长期记忆的召回噪声。向量检索基于语义相似度但相似不等于相关。用户说我要退货可能召回三个月前关于退货政策的对话而实际上用户关心的是当前的退货流程。解决方案是给长期记忆加时间衰减因子近期存储的文档权重更高。多轮压缩的累积误差。每次压缩都是一次有损操作多次压缩后摘要可能偏离原始语义。生产环境中建议限制压缩次数当短期记忆摘要块超过一定数量时将最早的摘要块合并为更高层级的摘要。延迟开销。每次对话需要执行实体提取 压缩判断 向量检索 上下文组装。这些步骤的延迟叠加后首字响应时间TTFT可能增加 200-500ms。对于实时性要求高的场景需要将实体提取和向量检索做成异步流水线。适用边界短对话5 轮以内不需要记忆管理直接全量上下文即可中等对话5-30 轮需要工作记忆压缩 实体提取长对话30 轮以上需要完整三层架构。五、总结Agent 长对话记忆管理的核心思路是分层工作记忆负责当前推理短期记忆负责近期摘要长期记忆负责历史召回。三层之间的流转规则——何时压缩、保留什么、如何召回——决定了 Agent 的记忆质量。落地路线建议第一步实现工作记忆的 Token 计数和溢出检测这是最基础的成本控制第二步实现关键实体提取确保结构化信息不因压缩而丢失第三步引入 LLM 摘要压缩将早期对话转为语义摘要第四步接入向量数据库实现长期记忆召回加时间衰减因子降低噪声第五步将实体提取和向量检索异步化控制 TTFT 延迟增量在 300ms 以内。记忆管理不是可选优化而是 Agent 从 Demo 走向生产的必经之路。没有记忆管理的 Agent就像一个没有笔记本的客服——短期还能应付时间一长必然出错。