当前位置：首页 > news >正文

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

news 2026/7/7 18:21:05

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_factory=lambda: datetime.now(timezone.utc).timestamp()) metadata: dict = field(default_factory=dict) 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_factory=lambda: datetime.now(timezone.utc).timestamp()) last_accessed_at: float = field(default_factory=lambda: datetime.now(timezone.utc).timestamp()) access_count: int = 0 # 原始对话的引用，便于追溯 source_conversation_id: Optional[str] = None class MemoryManager: """分层记忆管理器。 协调短期记忆（会话消息）、工作记忆（任务状态） 和长期记忆（向量数据库）的读写。 使用示例： manager = MemoryManager(user_id="user_123", max_short_term_tokens=8000) 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_ascii=False, indent=2 ) 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(role="user", content=user_msg, timestamp=now)) self.short_term.append(Message(role="assistant", content=assistant_msg, timestamp=now)) # 检查是否需要压缩 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( id=f"mem_{conversation_id}_1", user_id=self.user_id, content="用户偏好使用表格形式展示数据", category="preference", importance=0.8, source_conversation_id=conversation_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计数（tiktoken）vs 估算计数：tiktoken精确但每次调用有0.5至2ms开销。在每秒数百次调用的场景下，累积开销显著。折中方案是对消息内容的Token计数用估算，只对最终裁剪后的上下文用tiktoken精确验证。

对话中提取记忆 vs 对话后提取记忆：本方案选择对话后提取，优点是避免了每次LLM调用的延迟增加。缺点是用户可能在对话中途退出，导致记忆未被提取。对于关键场景，应在每次工具调用成功后立即提取相关记忆。

全文检索 vs 向量检索：长期记忆的检索可以用全文检索（Elasticsearch）或向量检索（Milvus/Pinecone）。向量检索的语义匹配能力更强（如"我喜欢简洁"和"不要太多细节"能匹配上），但全文检索在精确关键词匹配（如用户名、ID）上有优势。最佳实践是两者结合：精确匹配用全文检索，语义匹配用向量检索。