ARTICLE DETAIL

资讯详情

深耕网站视觉设计与运营推广的一线实战洞察。

AI Agent记忆分层设计:从上下文窗口到长期记忆的工程落地

AI Agent记忆分层设计:从上下文窗口到长期记忆的工程落地 最近在梳理智能体框架的技术架构时发现很多团队在开发 Agent 应用时都会遇到同一个现象模型单轮能力很强但一旦进入多轮对话、跨会话协同、长期用户画像这类场景效果就明显下降。根因往往不在模型本身而在于 Agent 没有一套结构化的记忆系统。本文会以 Prime Agent 这类具备复杂任务编排能力的框架为背景系统拆解记忆层级机制的设计思路包括每一层级的定位、存取流程、数据结构、存储选型以及工程落地时常见的坑点。适合正在做 AI 应用开发、Agent 框架设计、RAG 系统优化的开发者阅读。1. 记忆层级机制为什么 Agent 不能没有“分层记忆”1.1 从“上下文窗口”到“记忆系统”大模型应用的常规做法是把用户问题、系统提示词、历史消息全部塞进上下文窗口然后让模型一次性生成结果。这种方式在单轮问答中完全够用但放到真实业务里会迅速暴露问题。首先是上下文窗口容量有限。即使目前主流模型的上下文窗口已经做到几十万 token真实业务中多轮对话、文档资料、用户画像叠加在一起很快就会接近上限。而且上下文越长推理耗时和调用成本也会同步上升不是所有场景都适合“全量塞入”。其次是信息遗忘。如果 Agent 只依赖当前会话上下文那么用户昨天表达过的偏好、上个任务里确定的约束条件在当前会话中就无法被感知。核心矛盾在于LLM 本身是“无状态”的而业务场景需要“有状态”的智能体。解决这个矛盾的通用方案就是为 Agent 设计一套独立的记忆系统让模型在需要的时候能够以低成本的方式获取最相关的历史信息。1.2 记忆层级机制是什么记忆层级机制是智能体系统设计中一套结构化的数据管理方案。它的核心思想是不要把记忆当成单一数据库而是按照数据的生命周期、访问频率、抽象程度和持久化要求把记忆划分为多个层级并设计相应的读写、沉淀、召回和淘汰策略。这里可以类比人类记忆的工作方式。人类的记忆并不是一个整体而是由感觉记忆、工作记忆、短期记忆、长期记忆等多个子系统协同工作的。工作记忆负责处理当前正在思考的信息短期记忆负责保存最近一段时间发生的事情长期记忆则负责存放那些需要长期保留的知识、经验和情感。Agent 的记忆层级机制本质上就是把这套认知模型工程化。专业一点说记忆层级机制需要回答以下问题当前这次推理哪些信息必须进入上下文会话过程中产生的信息哪些值得保存跨会话的信息如何从历史数据中提炼出来长期存储的数据如何被快速、准确地召回信息什么时候过期什么时候应该被遗忘1.3 记忆层级与 RAG 的关系很多开发者会把记忆机制与 RAG检索增强生成混为一谈这里需要做一个区分。RAG 解决的是“外部知识获取”问题。它的核心流程是把文档切分、向量化、存入库中然后在用户提问时检索相关片段注入上下文。RAG 面向的是静态知识库例如企业知识库、产品文档、法律法规。记忆系统解决的是“交互历史与个性化信息”问题。它保存的是用户与 Agent 之间的对话历史、用户偏好、任务状态、执行结果等动态数据。两者在实践中通常共存。比如一个客服 Agent既需要通过 RAG 获取产品知识库内容也需要通过记忆系统记住用户之前的工单信息、投诉记录和沟通偏好。架构设计上两者可以共用向量检索基础设施但数据模型、写入链路、更新策略完全不同。了解了这些差异之后再来看 Prime Agent 这类复杂智能体框架中的记忆层级设计就会清晰很多。2. 记忆层级机制的整体架构2.1 四级分层模型综合目前主流 Agent 框架的实践记忆层级机制通常可以划分为四个核心层级。记忆层级定位存储时效典型存储形式访问频率工作记忆上下文层当前推理所需的信息集合单次请求内存中的上下文窗口最高短期记忆会话层当前会话的对话历史与状态单次会话Redis、内存、消息队列高长期记忆持久化层跨会话的用户画像、领域事实、历史事件数天到长期向量数据库、关系型数据库中语义记忆抽象层从历史中提炼的偏好、规则、技能模式长期向量库 结构化存储低命名上不同框架会有差异比如有的框架把长期记忆进一步拆成“情景记忆”和“语义记忆”有的框架会在长期记忆之上增加“程序性记忆”来保存工具调用经验。但整体分层思路是一致的越靠近模型推理的层级访问越快、容量越小、生命周期越短越靠近持久化存储的层级抽象程度越高、容量越大、构建成本也越高。2.2 层级之间的数据流方向记忆层级不是彼此孤立的它们之间存在清晰的数据流转链路。可以用下面这个简化的流程来描述用户输入 - 工作记忆组织当前上下文包含系统提示、近期消息、召回结果 - 短期记忆会话历史落地滚动窗口 / 摘要压缩 - 长期记忆跨会话信息沉淀向量化 元数据存储 - 语义记忆离线或异步聚合提炼偏好与画像读取方向则相反通常是先有当前请求然后触发记忆召回从长期记忆中检索候选信息注入到工作记忆中最终参与模型推理。整个链路中写入是层层沉淀的读取是逐级召回的。2.3 记忆层级设计的三条核心原则在实际设计记忆层级机制时我建议优先把握三条原则它们能避免很多后期返工。第一按访问速度分层。高频访问的数据必须放在低延迟的存储中低频数据则可以放在成本更低的存储中。不要为了架构统一而把所有记忆都塞进一个数据库那样会导致上下文构建变慢也会让存储成本失控。第二按重要程度决定沉淀策略。并非所有交互历史都值得进入长期记忆。工作记忆和短期记忆可以“全量记录”但进入长期记忆之前必须经过重要性筛选、去重和聚合否则长期记忆库会迅速被噪声填满检索质量也会下降。第三遗忘与过期必须显式设计。很多初版记忆系统只设计了“写入”和“读取”忘了设计“遗忘”。这在数据量小的时候没问题但一旦运行几个月过期的用户画像、废弃的任务状态就会持续干扰模型判断甚至引发隐私合规风险。3. 核心层级拆解3.1 工作记忆上下文窗口的动态管理工作记忆是模型在单次推理时能够感知到的全部信息。它既要包含用户当前的问题也要包含系统提示词、工具返回结果、短期记忆摘要、以及从长期记忆中召回的相关信息。工作记忆的设计核心是“上下文预算管理”。由于上下文窗口是有限的预算分配就需要有优先级。我的常用分配策略如下系统提示词和角色设定固定占用一部分预算。当前用户输入是最高优先级必须完整保留。实时工具返回结果次之它与当前任务直接相关。短期记忆摘要、长期记忆召回片段按重要度动态分配剩余预算。工作记忆的代码层面通常表现为一个上下文组装器负责在每次请求前把多个来源的信息合并成最终的 prompt。下面是一个简化的上下文组装示例展示如何管理工作记忆中的不同信息块# 文件路径memory/layers/working_memory.py from dataclasses import dataclass, field from typing import List, Optional dataclass class ContextBlock: 上下文信息块 block_type: str # system / user / memory / tool_result content: str priority: int 0 # 优先级数值越大越重要 token_estimate: int 0 dataclass class WorkingMemory: 工作记忆层负责将多源信息组装成上下文。 这里仅展示核心设计思路实际工程中需要叠加 token 计算和裁剪策略。 max_tokens: int 8000 blocks: List[ContextBlock] field(default_factorylist) def add_block(self, block: ContextBlock) - None: self.blocks.append(block) def build_context(self) - str: 按优先级排序组装最终上下文 self.blocks.sort(keylambda b: b.priority, reverseTrue) used_tokens 0 selected [] for block in self.blocks: if used_tokens block.token_estimate self.max_tokens: continue selected.append(block.content) used_tokens block.token_estimate return \n\n.join(selected)这段代码的核心思路是维护一个带优先级的上下文块列表在组装时按优先级从高到低裁剪直到达到预算上限。实际项目里还需要接入 tokenizer 做精确计算并根据模型类型动态调整 max_tokens。3.2 短期记忆会话历史的滚动与摘要短期记忆解决的是当前会话中的连续性。最朴素的做法是把所有历史消息都保存下来每轮都带上前面的完整对话。但在长对话中这种方案会快速占满上下文窗口。常见优化方案有两种可以组合使用。第一种是滚动窗口Sliding Window。只保留最近 N 轮对话更早的消息直接丢弃。实现简单但会丢失长对话早期的关键信息。第二种是摘要压缩Summary Compression。当对话超过一定轮数时把早期对话交给模型生成一个结构化摘要后续只携带摘要 最近几轮完整消息。这种方式能保留更多信息代价是需要额外的模型调用。下面是一个摘要压缩管理的参考实现# 文件路径memory/layers/short_term_memory.py class ShortTermMemory: 短期记忆层管理当前会话的对话历史。 当历史消息数量超过阈值时触发摘要压缩。 def __init__(self, max_messages: int 20): self.messages [] self.summary self.max_messages max_messages def add_message(self, role: str, content: str) - None: self.messages.append({role: role, content: content}) if len(self.messages) self.max_messages: self._compress() def _compress(self) - None: 实际工程中这里调用 LLM 对 self.messages 进行摘要提取 并将摘要合并到 self.summary 中。 old_messages self.messages[: self.max_messages // 2] self.summary self._summary_with_llm(old_messages) self.messages self.messages[self.max_messages // 2:] def _summary_with_llm(self, messages: list) - str: # 示意代码实际需替换为真实的 LLM 调用 return .join(m[content] for m in messages)[:200] def get_context(self) - list: 返回用于拼接到 prompt 的上下文 if self.summary: return [{role: system, content: f对话摘要{self.summary}}] self.messages return self.messages这里需要特别注意的是摘要的分级更新策略。如果每次压缩都重写全部摘要模型调用成本会很高。更优的做法是保留“摘要链”每次只对新增部分做增量摘要再与旧摘要合并。3.3 长期记忆向量化存储与跨会话召回长期记忆是整个记忆层级机制中最核心、也最复杂的一层。它的目标是把跨会话的有价值信息持久化保存并在需要的时候做精准召回。进入长期记忆的信息通常是几种用户明确表达过的偏好和身份信息任务执行过程中沉淀下的事实记录多轮对话中提炼出的结论和决策。长期记忆的数据结构不建议直接存储原始对话文本。更推荐的做法是设计一种“记忆条目”的结构化格式。一个经过实践验证的字段设计如下{ memory_id: mem_20250101_001, user_id: user_123, session_id: session_456, memory_type: preference, content: 用户偏好使用 Python 进行数据处理对 pandas 和 polars 比较熟悉, importance: 0.85, created_at: 2025-01-01T10:00:00Z, updated_at: 2025-01-01T10:00:00Z, expire_at: null, source: session_456 }字段说明memory_id记忆条目的唯一标识。user_id / session_id用于归属和多租户隔离。memory_type记忆类型比如偏好、事实、事件、任务状态。content记忆的核心内容需要语义完整、自包含。importance重要度评分用于召回排序和淘汰决策。created_at / updated_at记录创建和更新时间。expire_at过期时间用于自动遗忘。source记录来源方便回溯和审计。这种结构化设计的好处是召回时可以做多维过滤不只是向量相似度匹配还能叠加时间、类型、用户等结构化条件。3.4 情景记忆与语义记忆的沉淀长期记忆可以进一步细分为情景记忆和语义记忆。情景记忆记录的是具体发生过的事件。例如“用户在 2025 年 1 月 1 日询问过如何搭建 FastAPI 项目”。这类记忆的特点是具体、有明确时间点、包含上下文细节。语义记忆则是从大量情景记忆中抽象出来的通用知识。例如“用户偏好 FastAPI 作为 Python Web 框架”“用户在部署时倾向于使用 Docker 容器”。语义记忆是长期记忆的高级形态通常需要异步任务对多个情景记忆做聚合提炼。沉淀策略上可以设计一个定时任务定期扫描一段时间内的情景记忆调用模型生成语义记忆候选再通过规则或模型判断是否入库。这个过程有点像离线数仓里的 ETL只是处理对象从结构化数据变成了文本记忆。3.5 关于程序性记忆的补充部分 Agent 框架还会引入程序性记忆层用来保存工具调用、工作流执行的成功经验。例如“当用户要求批量处理文件时优先使用 Python 脚本而不是逐条回复”。这类记忆对提升 Agent 效率很有帮助但风险也更高。程序性记忆一旦出现偏差会导致 Agent 在相似任务中反复采用错误的工具调用方式。如果需要引入建议额外增加置信度字段只有多次成功执行后才允许写入。4. 记忆读写流程与生命周期管理4.1 记忆写入流程记忆写入不是简单地把对话内容丢进数据库。一个规范的写入链路通常包括四个步骤。第一步记忆候选提取。从当前对话中提取可能值得保存的信息。这一步可以基于规则也可以调用模型结合“是否包含用户偏好、是否包含关键事实、是否为重要决策”等维度来判断。第二步重要度评分。对每个候选记忆打一个 0 到 1 之间的重要度分数。重要度影响后续的召回排序和存储周期。第三步去重与合并。在写入前检索是否已有相同或相似的记忆条目如果存在则做内容合并和 updated_at 刷新而不是新插入一条。第四步向量化与存储。将记忆内容通过 Embedding 模型转为向量连同元数据一起写入存储。参考写入函数如下# 文件路径memory/service/memory_writer.py def write_memory(memory_payload: dict, vector_store, embedder) - str: 写入一条记忆 1. 生成 embedding 2. 检查相似记忆决定新增或更新 3. 写入向量库 content memory_payload[content] embedding embedder.embed(content) # 相似度检查阈值需根据业务调整 similar vector_store.search( embeddingembedding, top_k3, filters{user_id: memory_payload[user_id]} ) for item in similar: if item.score 0.92 and item.memory_type memory_payload.get(memory_type): # 更新已有记忆 memory_payload[memory_id] item.memory_id memory_payload[updated_at] now() vector_store.update(memory_payload) return item.memory_id # 新增记忆 memory_payload[memory_id] generate_memory_id() memory_payload[created_at] now() vector_store.insert(memory_payload) return memory_payload[memory_id]需要强调这里对相似度阈值的设定要谨慎。阈值过高会导致重复记忆大量存在阈值过低会把不同含义的句子错误合并。建议在真实业务数据上做一次阈值试验观察 precision/recall 的变化。4.2 记忆读取与召回流程记忆读取发生在每次模型推理之前它的质量直接影响生成效果。标准流程如下触发判断。基于当前用户输入判断是否需要检索长期记忆。例如用户只是打招呼就没有检索必要。这个判断可以用简单的分类模型或规则完成能大幅降低检索调用量。生成查询向量。将当前用户问题的核心部分向量化。多路召回。除了向量相似度检索还要叠加结构化过滤条件比如 user_id 必须匹配、created_at 在指定时间范围内、memory_type 符合当前场景。重排过滤。对召回的候选记忆按重要度、时间衰减、相似度做综合打分。注入上下文。把最终选中的记忆格式化后注入上下文并标注来源帮助模型判断信息可信度。一个简化版的多路召回实现思路# 文件路径memory/service/memory_retriever.py def retrieve_memories(query: str, user_id: str, top_k: int 5) - list: # 1. 向量召回 query_embedding embedder.embed(query) vector_results vector_store.search( embeddingquery_embedding, top_ktop_k * 2, filters{user_id: user_id, is_deleted: False} ) # 2. 结构化过滤去掉过期记忆 filtered [r for r in vector_results if not is_expired(r)] # 3. 综合打分相似度 重要度 时间衰减 scored [] for item in filtered: score ( item.score * 0.6 item.importance * 0.3 time_decay(item.updated_at) * 0.1 ) scored.append((item, score)) # 4. 取 Top-K scored.sort(keylambda x: x[1], reverseTrue) return [item for item, _ in scored[:top_k]]实际生产中重排环节往往比单纯的向量检索更影响效果。因为向量相似度只能衡量“语义接近”不能衡量“对当前任务是否有用”。重要度、时效性和置信度这些维度必须纳入综合打分。4.3 遗忘与淘汰机制很多团队在构建记忆系统时会把全部精力放在写入和检索上遗忘机制放在最后处理甚至完全忽略。这是一个隐患。遗忘机制的核心意义有两个。第一个是保证记忆新鲜度。用户偏好会变化半年前的画像可能已经不再适用。第二个是控制存储成本。长期运行后重复、低质的记忆条目会越积越多导致检索效率下降。常见的遗忘策略有TTL 过期。为记忆设置 expire_at 字段到期后自动转为不可检索状态。重要度阈值淘汰。定期清理低于重要度阈值的记忆。手动删除。当用户明确要求删除某条记忆或全部记忆时必须支持彻底删除。事件驱动淘汰。当用户主动修改偏好时旧偏好条目应立即弱化或删除。需要特别注意的是AI 应用中“删除记忆”不只是关闭可见性而是要有真正的删除链路包括向量库、结构化备份、日志中的敏感字段。这一点在下文的最佳实践中会进一步展开。5. 工程落地设计要点5.1 记忆数据模型与字段版本控制记忆条目的字段不是一成不变的。随着业务迭代可能会出现需要新增字段、调整 memory_type 枚举值的情况。因此在设计数据模型时建议预留一个 version 字段并对 memory_type 做枚举管理避免代码里散落魔法字符串。如果使用向量数据库 关系型数据库组合存储可以采取如下分工向量库负责语义检索关系型数据库负责元数据管理和业务查询。两个存储通过 memory_id 关联写入时采用事务或消息队列保证最终一致。5.2 存储选型建议存储类型典型代表优点局限向量数据库Milvus、Qdrant、pgvector、Chroma语义检索能力强精确定位和聚合查询较弱键值存储Redis低延迟适合短期会话状态不支持复杂检索文档数据库MongoDB灵活 schema适合存记忆条目语义检索需要额外集成关系型数据库PostgreSQLpgvector结构化查询和向量检索兼具数据量大时性能需调优轻量项目可以先用 PostgreSQL pgvector减少引入额外的组件。数据规模上来后再迁移到专用向量数据库。5.3 多租户与权限隔离记忆数据高度敏感多租户隔离是必须做好的安全底线。隔离的基本原则是所有写入和检索操作都必须携带 user_id / tenant_id并在存储查询层强制附加过滤条件不能依赖上层业务代码自觉遵守。此外还需要分层设计权限模型。普通用户只能访问自己的记忆管理员可以查看系统级统计数据但不应该能看到用户原始记忆内容模型调试环境应使用脱敏数据。5.4 可观测性与记忆命中评估记忆系统的上线只是开始持续评估和调优才是关键。建议至少记录以下指标记忆召回率当前请求中有多少比例实际触发了记忆检索。记忆注入率检索后有多少结果真正被注入上下文。记忆有用性人工抽检或基于模型评分判断注入的记忆对答案是否有正向帮助。检索延迟从发起检索到结果返回的耗时需要控制在百毫秒级别。调试时还需要支持“查看某次回答使用了哪些记忆”。可以在 prompt 注入时为每条记忆添加类似引用标记的做法例如[memory_ref: mem_20250101_001] 用户偏好使用 Python 进行数据处理。这样即使生成结果不符合预期也能快速定位是哪条记忆产生了影响。6. 常见问题与排查思路问题现象常见原因解决思路用户 A 的回答中出现用户 B 的信息检索时未强制 user_id 过滤检查检索层过滤条件在存储层强制隔离召回结果与当前问题无关纯向量检索未叠加重要度和时间衰减引入多路召回 重排策略上下文越来越大响应变慢短期记忆未做滚动窗口压缩增加摘要压缩机制控制注入 token 数长期记忆中存在大量重复条目写入前缺少去重与合并增加相似度检测和记忆合并逻辑用户要求删除记忆但对话中依然出现旧信息仅删除向量库缓存或日志中仍有残留建立完整删除链路覆盖向量库、缓存、日志脱敏向量检索结果不稳定同一问题不同结果阈值设置不合理或 embedding 对短文本不敏感调整阈值优化查询文本增加候选集大小后重排这里挑两个高频问题详细说明排查思路。第一个是“跨用户记忆串扰”。如果发现用户 A 的回答里出现了用户 B 的偏好优先检查两个位置一是检索函数是否显式传入了当前 user_id 并作为过滤条件二是底层存储的查询语句是否遗漏了过滤。很多问题都出在“代码里过滤了但历史数据没有 user_id 字段”或“部分接口复用了公共检索函数”。第二个是“记忆检索不到但数据库里明明有”。这种情况往往是因为向量相似度检索的 top_k 设置过小或者过滤条件过严。建议先把 top_k 调大再逐步缩小过滤条件对比召回结果差异。也可以直接查看 query 向量与目标记忆向量的余弦相似度确认阈值设置是否合理。7. 最佳实践与工程建议7.1 最小化记忆单元让信息自包含写入长期记忆的内容要尽量做到自包含。不要存储“用户很喜欢”而要存储“用户偏好使用 Python 完成数据处理任务熟悉 pandas 和 polars”。自包含的记忆条目在检索时可以独立发挥作用也能避免与上下文产生歧义。7.2 写入前先判断“是否值得记住”不是所有对话内容都值得进入长期记忆。建议建立一套明确的写入门槛例如包含用户主动表达的偏好。包含任务执行结果中的关键事实。包含影响后续决策的约束条件。用户明确要求“记住”的信息。满足其中至少一条才进入记忆候选池。7.3 检索结果必须带上来源标记注入上下文的记忆信息要附带 memory_id 和来源标签。一方面便于模型判断信息是否可信另一方面可以提升后期排查效率。当回答出现错误时可以快速回溯是哪一条记忆导致的。7.4 记忆系统同样需要灰度发布修改记忆提取逻辑、摘要策略或重排算法都可能对线上效果产生显著影响。建议设计对照组在小流量用户范围内验证效果后再全量发布。评估指标可以是任务完成率、用户满意度或人工评测分数。7.5 建立“忘记”机制尊重用户隐私长期记忆能力越强隐私风险就越大。产品层面需要提供用户可以查看、删除、导出自身记忆的入口。系统层面要支持单条记忆删除和全量记忆清除并保证删除操作真正生效。7.6 从轻量方案开始不要过度设计记忆层级机制很容易陷入过度设计的陷阱。如果项目还处于 MVP 阶段可以先用 Redis 保存短期会话、用 pgvector 保存长期记忆两三个服务就能跑起来。等到数据规模变大、检索质量出现瓶颈再逐步引入语义记忆聚合、程序性记忆等高级层级。8. 总结与下一步学习路线本文围绕 Prime Agent 的记忆层级机制从概念、架构、层级拆解、读写流程到工程落地做了系统梳理。核心要点可以概括为四句话记忆要分层写入要筛选召回要混合遗忘要显式。如果接下来想继续深入建议按下面的路径逐步实践第一步先管理好工作记忆。学会做上下文预算控制理解 token 成本与效果之间的平衡。第二步实现短期记忆的摘要压缩。用滚动窗口和摘要链解决长对话衰减问题。第三步搭建长期记忆的向量检索链路。先用 pgvector 做最小闭环从一条记忆写入到检索命中全流程跑通。第四步补充遗忘机制和隐私删除能力再考虑语义记忆聚合。第五步也是最高级的阶段引入评估体系用数据驱动的方式持续优化记忆召回质量。记忆系统没有一劳永逸的完美方案它更像一个需要持续迭代的独立模块。建议先让最小链路跑起来再根据真实场景的问题逐步演进。如果你正在做 Agent 应用不妨把这套记忆层级机制当成一个检查清单对照自己的项目看看哪些层级已经具备哪些还是空白。
返回列表