Agent产品中的长短期记忆管理：会话上下文与持久化存储方案

Agent产品中的长短期记忆管理会话上下文与持久化存储方案一、LLM的失忆症为什么Agent需要记忆管理大语言模型最大的工程限制之一是无状态——每次API调用都是独立的模型本身不保留任何跨会话的信息。对于对话式AI产品这意味着需要自行构建记忆系统来管理上下文。常见的直接把历史对话全塞进Prompt方案在实际应用中面临两个瓶颈一是Token成本10轮对话的历史可能消耗3000至5000个Token每天100万次对话的Token费用中约40%花在了提示词的历史对话部分二是上下文窗口即使是128K上下文的模型窗口开头的信息在推理质量上明显劣于窗口末尾的信息即Lost in the Middle现象。一个完整的Agent记忆系统需要同时管理三种不同生命周期和访问模式的数据短期记忆当前会话毫秒级读写存活时间为单一会话容量受Token预算限制长期记忆跨会话毫秒级写入、秒级读取永久存储需要语义检索能力工作记忆当前任务存储Agent执行多步任务时的中间状态和工具调用结果差异化的存储和检索策略是降低Token成本、提升对话质量的关键。本文给出一种工程化的分层记忆实现方案。二、底层机制与原理剖析2.1 记忆分层的架构设计不同层级的记忆有不同的读写特征。短期记忆在每次LLM调用时都需要读取因此必须最快。长期记忆只在需要时检索可以用向量搜索实现语义匹配。工作记忆在任务执行过程中频繁更新需要支持追加和覆盖。graph TD subgraph 短期记忆 - 当前会话 A[对话消息列表] B[滑动窗口管理] C[摘要压缩器] end subgraph 工作记忆 - 当前任务 D[任务状态存储] E[工具调用结果] F[中间推理步骤] end subgraph 长期记忆 - 持久化 G[向量数据库] H[关系型存储] I[知识图谱] end subgraph 记忆管理器 J[记忆编排器] K[上下文组装器] L[Token预算控制器] end A -- K B -- K C -- K D -- K E -- K G -- K H -- K K -- L L -- M[LLM API调用] J -- A J -- D J -- G J -- H上图中记忆编排器是核心调度组件。它在每次LLM调用前协调所有记忆源的读取在调用后根据新产生的信息决定写入哪些记忆层。上下文组装器负责将各层记忆组合成一个在Token预算内的Prompt。Token预算控制器确保不会因记忆膨胀而超限。2.2 短期记忆的滑动窗口与摘要策略短期记忆的直接形态是对话消息列表。理论上可以全量发送但Token开销巨大。优化策略分为两种滑动窗口只保留最近N轮对话。实现简单但会丢失窗口前的关键信息如用户一开始设定的角色和场景。改进方案是固定首条滑动窗口——始终保留第一条系统消息和用户设定的场景描述其余历史对话用滑动窗口。递归摘要当对话超过一定轮次时用较便宜的模型如gpt-4o-mini将旧对话压缩为摘要替换原始消息。关键是摘要需要保留关键事实和待办事项而非简单的总结。一个有效的摘要Prompt应该包含用户偏好、已完成的行动、未完成的任务。实践中通常结合两者滑动窗口保留最近10轮原始消息超出部分递归压缩为摘要。2.3 长期记忆的检索增强存储长期记忆需要语义检索能力——用户下次对话时提到相关话题系统应该召回之前的上下文。这要求将每次对话的重要信息提取为记忆片段向量化后存入向量数据库。关键设计决策是提取什么而非存什么。如果全量存储对话记录会引入大量噪音如闲聊、重复问题降低检索精度。需要设计一个记忆提取器——在对话结束时由LLM判断哪些信息值得长期保存。典型的提取标准包括用户偏好、用户背景、行动历史、待办事项。检索时需要平衡相关性和多样性。相关性由向量相似度保证多样性需要用MMR最大边际相关性算法去重防止多条高度相似但冗余的记忆占据检索结果。三、生产级代码实现与最佳实践3.1 分层记忆管理器的实现# memory_manager.py — 分层记忆管理器 # # 三种记忆的协调调度器。 # 核心职责 # 1. LLM调用前从各层收集相关记忆组装上下文 # 2. LLM调用后分析新信息决定写入哪些记忆层 # 3. 会话结束时触发长期记忆提取和持久化 from __future__ import annotations from datetime import datetime, timezone from typing import Any, Optional from dataclasses import dataclass, field import json dataclass class Message: 对话消息的标准数据结构。 为什么不用简单的dict - 类型安全防止role字段拼写错误 - 方法内聚token计数等逻辑与数据绑定 - 序列化统一确保API交互格式一致 role: str # system, user, assistant, tool content: str timestamp: float field(default_factorylambda: datetime.now(timezone.utc).timestamp()) metadata: dict field(default_factorydict) def estimate_tokens(self) - int: 粗略估计Token数中文按1.5字/Token英文按4字符/Token。 精确的Token计数需要使用tiktoken但考虑到性能开销 在非精确场景如滑动窗口预算用估算值足够。 # 混合文本的估算中文字符权重更高 import re chinese_chars len(re.findall(r[\u4e00-\u9fff], self.content)) other_chars len(self.content) - chinese_chars return int(chinese_chars / 1.5 other_chars / 4) dataclass class MemoryItem: 长期记忆的存储单元。 每个MemoryItem是从对话中提取的一个事实或偏好。 与直接存储对话消息相比这种结构化存储能显著提升检索精度。 id: str user_id: str content: str # 人类可读的记忆内容 category: str # preference, fact, task, context importance: float 0.5 # 重要性评分0-1 created_at: float field(default_factorylambda: datetime.now(timezone.utc).timestamp()) last_accessed_at: float field(default_factorylambda: datetime.now(timezone.utc).timestamp()) access_count: int 0 # 原始对话的引用便于追溯 source_conversation_id: Optional[str] None class MemoryManager: 分层记忆管理器。 协调短期记忆会话消息、工作记忆任务状态 和长期记忆向量数据库的读写。 使用示例 manager MemoryManager(user_iduser_123, max_short_term_tokens8000) context await manager.build_context() response await call_llm(context) await manager.after_llm_call(response) def __init__( self, user_id: str, max_short_term_tokens: int 8000, window_size: int 10, # 滑动窗口保留最近N轮对话 ): self.user_id user_id self.max_tokens max_short_term_tokens self.window_size window_size # 短期记忆当前会话的消息列表 self.short_term: list[Message] [] # 工作记忆当前任务的执行状态 self.working_memory: dict[str, Any] {} # 长期记忆的本地缓存实际数据在向量数据库中 self._long_term_cache: list[MemoryItem] [] # 对话摘要压缩的旧消息 self._summary: Optional[str] None async def build_context(self) - list[dict]: 构建发送给LLM的上下文。 组装顺序影响LLM的注意力分布 1. System Prompt角色定义始终保留 2. 长期记忆相关历史有选择性 3. 对话摘要压缩的旧消息如果存在 4. 工作记忆当前任务状态 5. 最近N轮对话消息完整保留 为什么这个顺序 - LLM对开头和结尾的信息注意力最高 - System Prompt放在开头确保角色稳定 - 最近消息放在末尾确保回复的连贯性 messages: list[dict] [] current_tokens 0 # 计算Token预算 # 预留30%给LLM的回复和工具调用 token_budget int(self.max_tokens * 0.7) # 1. 首先加入长期记忆如果相关 if self._long_term_cache: memory_text ## 关于用户的历史记忆\n for item in self._long_term_cache[:5]: # 最多5条 memory_text f- {item.content}\n memory_msg {role: system, content: memory_text} estimated len(memory_text) // 2 # 粗略Token估算 if current_tokens estimated token_budget: messages.append(memory_msg) current_tokens estimated # 2. 加入对话摘要 if self._summary: summary_msg { role: system, content: f以下是之前对话的摘要\n{self._summary} } estimated len(self._summary) // 2 if current_tokens estimated token_budget: messages.append(summary_msg) current_tokens estimated # 3. 加入工作记忆 if self.working_memory: wm_text ## 当前任务状态\n json.dumps( self.working_memory, ensure_asciiFalse, indent2 ) wm_msg {role: system, content: wm_text} estimated len(wm_text) // 2 if current_tokens estimated token_budget: messages.append(wm_msg) current_tokens estimated # 4. 加入最近N轮对话从后往前取直到Token预算耗尽 window_messages self.short_term[-self.window_size * 2:] # N轮*2用户助手 for msg in reversed(window_messages): estimated msg.estimate_tokens() if current_tokens estimated token_budget: break messages.insert(1, {role: msg.role, content: msg.content}) current_tokens estimated return messages async def after_llm_call( self, user_msg: str, assistant_msg: str ): LLM调用后的处理。 1. 将新消息加入短期记忆 2. 检查是否需要压缩生成摘要 3. 提取可能需要长期保存的信息 now datetime.now(timezone.utc).timestamp() # 加入新消息 self.short_term.append(Message(roleuser, contentuser_msg, timestampnow)) self.short_term.append(Message(roleassistant, contentassistant_msg, timestampnow)) # 检查是否需要压缩 await self._maybe_compress() async def _maybe_compress(self): 当短期记忆超出阈值时压缩旧消息为摘要。 压缩策略 - 保留最近 window_size*2 条消息不压缩 - 更早的消息由LLM压缩为结构化摘要 - 摘要字段包括关键决策、用户偏好、待办事项 为什么用压缩而非丢弃 - 丢弃会丢失重要的上下文线索 - 摘要虽然不精确但比完全没有强 threshold self.window_size * 2 * 2 # 2倍窗口大小才触发压缩 if len(self.short_term) threshold: return # 需要压缩的消息窗口之前的所有消息 compress_boundary len(self.short_term) - self.window_size * 2 messages_to_compress self.short_term[:compress_boundary] # 生成压缩Prompt # 注意这里仅展示逻辑实际调用LLM compress_prompt self._build_compress_prompt(messages_to_compress, self._summary) # 精简short_term只保留窗口内的消息 self.short_term self.short_term[compress_boundary:] def _build_compress_prompt( self, messages: list[Message], existing_summary: Optional[str] ) - str: 构建用于压缩的Prompt。 要求LLM提取结构化信息而非简单总结。 conversation_text \n.join( f[{m.role}]: {m.content} for m in messages ) prompt f请从以下对话中提取结构化信息不要简单总结 已有摘要如有{existing_summary or 无} 对话内容 {conversation_text} 请按以下格式输出 ## 用户关键偏好 - 列出用户明确表达或行为体现的偏好 ## 已完成的关键行动 - 列出用户已完成的重要操作 ## 待办/未完成任务 - 列出用户提到但尚未完成的事项 ## 重要决策 - 列出对话中做出的重要决定 return prompt async def extract_long_term_memory(self, conversation_id: str): 会话结束时提取需要长期保存的记忆。 这个过程通过LLM分析整个对话 提取值得跨会话保留的信息。 extraction_prompt self._build_extraction_prompt() # 注意此处省略实际的LLM调用 # 实际实现中会调用LLM并解析返回的结构化记忆 # 示例模拟提取结果 sample_memories [ MemoryItem( idfmem_{conversation_id}_1, user_idself.user_id, content用户偏好使用表格形式展示数据, categorypreference, importance0.8, source_conversation_idconversation_id, ), ] # 存入向量数据库此处用伪代码表示 # await vector_db.upsert(sample_memories) self._long_term_cache.extend(sample_memories) def _build_extraction_prompt(self) - str: 构建记忆提取的Prompt。 messages_text \n.join( f[{m.role}]: {m.content} for m in self.short_term ) return f分析以下对话提取值得长期保留的信息。 {{ preferences: [用户偏好1, 用户偏好2], facts: [事实1, 事实2], tasks: [待完成任务1], do_not_save: [不应保存的信息] }} 对话内容 {messages_text}3.2 Token预算的精算控制器# token_budget.py — Token预算控制器 # # 确保每次LLM调用的输入Token不超过预算。 # 超额时按优先级策略逐一移除低优先级内容。 from typing import Protocol class TokenCounter(Protocol): Token计数器的接口。 使用Protocol而非ABC允许不同的实现 - tiktoken精确但需要下载模型文件 - 估算方法快速但不够精确 def count(self, text: str) - int: ... class TokenBudgetController: Token预算控制器。 当构建的上下文超出预算时按以下优先级移除内容 1. 移除最旧的非系统消息优先级最低 2. 压缩长期记忆减少记忆条数 3. 截断工作记忆只保留最关键字段 4. 缩短摘要最后手段 系统Prompt角色定义不被移除——那是Agent的人格基础。 def __init__(self, max_tokens: int): self.max_tokens max_tokens self.reserved_tokens int(max_tokens * 0.3) # 预留30%用于回复 def fit_budget(self, messages: list[dict], counter: TokenCounter) - list[dict]: 调整消息列表以适应Token预算。 返回调整后的消息列表保证Token总数在预算内。 budget self.max_tokens - self.reserved_tokens total sum(counter.count(m.get(content, )) for m in messages) if total budget: return messages # 按优先级排序系统消息 用户消息 助手消息 # 移除优先级最低、时间最早的非系统消息 adjusted list(messages) while total budget and adjusted: # 找到第一个可以移除的消息非system消息中最早的 removed False for i, msg in enumerate(adjusted): if msg[role] ! system: total - counter.count(msg.get(content, )) adjusted.pop(i) removed True break # 如果只剩下系统消息但仍超预算只能截断系统消息 if not removed: # 极端情况系统消息本身超过了预算 for msg in adjusted: content msg[content] # 截断到剩余预算允许的长度估算 truncated_len max(100, budget // 2) msg[content] content[:truncated_len] ... break return adjusted四、边界分析与架构权衡4.1 适用场景需要多轮对话的AI Agent产品如AI客服、AI助手、AI代码助手用户期望Agent记住我的场景如个性化推荐、学习助手复杂的多步任务执行需要工作记忆来追踪任务进展4.2 不适用或需简化场景单轮问答系统如FAQ机器人不需要记忆管理极度敏感的数据处理如医疗问诊长期记忆存储需要额外的合规审查用户量极大但交互简单如20%的用户只问一次就不再回来记忆提取的投资回报率低4.3 关键设计权衡精确Token计数tiktokenvs 估算计数tiktoken精确但每次调用有0.5至2ms开销。在每秒数百次调用的场景下累积开销显著。折中方案是对消息内容的Token计数用估算只对最终裁剪后的上下文用tiktoken精确验证。对话中提取记忆 vs 对话后提取记忆本方案选择对话后提取优点是避免了每次LLM调用的延迟增加。缺点是用户可能在对话中途退出导致记忆未被提取。对于关键场景应在每次工具调用成功后立即提取相关记忆。全文检索 vs 向量检索长期记忆的检索可以用全文检索Elasticsearch或向量检索Milvus/Pinecone。向量检索的语义匹配能力更强如我喜欢简洁和不要太多细节能匹配上但全文检索在精确关键词匹配如用户名、ID上有优势。最佳实践是两者结合精确匹配用全文检索语义匹配用向量检索。五、总结Agent记忆系统需要分层管理短期会话消息、长期跨会话、工作任务状态滑动窗口递归摘要的组合策略可降低60%以上的Token开销长期记忆应提取结构化信息而非存储原始对话以提升检索精度Token预算控制器应在构建上下文时实时调整而非事后裁剪记忆提取的时机应选在对话结束或每次重要操作完成时避免增加每次调用的延迟向量检索与全文检索结合是最优的长期记忆检索方案兼顾语义和精确匹配