1. 项目概述为什么LLM智能体需要一个“双进程”记忆系统如果你最近在折腾LLM智能体比如让它们帮你自动处理邮件、分析数据或者管理项目大概率会遇到一个让人头疼的问题这“家伙”记性太差了。你让它总结上周的会议纪要它可能把上个月无关的讨论也混进来你希望它基于长期的客户反馈优化方案它却总是被最新的几条无关评论带偏方向。这背后的核心瓶颈就是智能体的记忆系统。传统的LLM智能体记忆方案无论是简单的滑动窗口只记住最近N轮对话还是基于向量数据库的检索增强生成RAG都像是一个“单线程”的大脑。它们要么健忘要么容易“胡思乱想”无法同时满足对近期细节的精准把握和对长期主题的抽象归纳。这正是“D-Mem: A Dual-Process Memory System for LLM Agents”这个项目要解决的核心痛点。D-Mem即双进程记忆系统其灵感来源于认知心理学中关于人类记忆“双过程”的理论旨在为LLM智能体构建一个更接近人类工作方式的记忆架构。简单来说你可以把D-Mem想象成智能体大脑里的两个协同工作的“部门”一个叫“快速反应部”专门负责处理当下任务相关的、具体的、细节性的信息追求快速和精准另一个叫“战略规划部”负责从长期互动中提炼模式、归纳主题、形成高层次的“经验”或“知识”用于指导未来的宏观决策。这套系统不是为了取代现有的RAG技术而是对其进行的架构级增强通过明确的分工与协同让智能体既拥有“摄影师”般的对焦能力也具备“战略家”般的远见。对于开发者、AI应用架构师以及对智能体可靠性有高要求的用户来说理解并应用D-Mem这样的系统至关重要。它直接决定了智能体在复杂、多轮次交互中的表现是否稳定、连贯且富有洞察力。接下来我将深入拆解D-Mem的设计思路、核心组件、实现细节并分享在构建此类系统时那些文档里不会写的“坑”与技巧。2. 核心设计思路拆解“双过程”理论的技术实现D-Mem的设计哲学根植于一个核心观察LLM智能体在不同时间尺度上对信息的需求和利用方式是截然不同的。我们将这两种需求映射为两个并行的记忆进程。2.1 进程一工作记忆Working Memory—— 聚焦当下的“便签本”工作记忆是智能体的“意识焦点”。它的设计目标是低延迟、高精度、强相关性。功能定位专门服务于当前正在执行的任务或对话轮次。例如在一个多步骤的代码调试任务中工作记忆会牢牢记住用户刚刚报出的错误信息、你已尝试过的修复方法、以及当前文件的上下文。技术实现这通常不是一个独立的数据库而是一个精心设计的信息管理策略。滑动窗口增强不仅仅是保留最近N条消息。D-Mem的工作记忆会动态调整窗口并赋予不同消息不同的权重。例如用户的直接指令权重最高智能体自身的工具调用结果次之而一些确认性的对话如“好的”、“明白了”权重可能很低甚至被过滤。相关性即时过滤在将信息存入工作记忆前会用一个轻量级的模型如小型BERT或专门训练的文本分类器进行实时判断确保只保留与当前任务高度相关的片段。这避免了无关历史对话对当前思考的“噪声”干扰。数据结构通常采用类似缓存的数据结构如LRU最近最少使用队列保证记忆的快速存取和更新。注意工作记忆的“容量”是有限的这是有意为之的设计。它模拟了人类认知的注意力瓶颈迫使系统必须做出取舍只保留最关键的信息从而提升处理当前任务的效率和准确性。2.2 进程二长期记忆Long-Term Memory—— 沉淀经验的“知识库”长期记忆是智能体的“经验宝库”。它的设计目标是大容量、结构化、可归纳。功能定位存储智能体与用户或环境所有有意义的交互历史。它的目的不是记住每一句话而是从中提炼出模式、偏好、事实和主题。例如长期记忆会知道用户“通常在每周一上午要求生成周报”并且“偏好用图表而非纯文字展示数据”。技术实现这里通常结合了向量数据库和更高级的信息聚合技术。分层存储原始交互日志被清洗和分割后存入向量数据库如Chroma, Weaviate, Pinecone用于基于语义的相似性检索。这是基础层。主题提炼与摘要这是D-Mem的进阶能力。系统会定期例如每24小时或每完成一个大型任务对近期存入长期记忆的“原始记忆片段”进行批处理。使用LLM对这些片段进行聚类分析、主题提取和自动摘要生成更高阶的“记忆摘要”或“用户画像片段”。这些摘要同样被向量化存储但它们代表了更抽象、更浓缩的知识。结构化标签系统为记忆片段打上多维度的标签如任务类型:数据分析、情感倾向:积极、关键实体:项目Alpha。这为后续的检索提供了除语义相似性外的另一条精准路径。2.3 双进程的协同机制检索与融合两个记忆进程并非孤立工作它们的价值体现在协同检索与信息融合上。当智能体需要响应一个新查询或执行新任务时D-Mem的“调度中心”会同时向两个记忆进程发起检索请求。并行检索向工作记忆请求检索与当前对话上下文最直接相关的近期信息。向长期记忆请求基于当前查询的语义检索相关的历史交互片段以及更高阶的主题摘要和用户偏好。相关性重排序与去重将两个来源的检索结果合并用一个重排序模型如Cross-Encoder对所有候选记忆片段进行统一的相关性打分。这个过程会自动过滤掉来自长期记忆中可能已经过时或与当前上下文冲突的信息并去重高度相似的片段。上下文构建与注入将重排序后的Top-K个记忆片段按照相关性、时间顺序或逻辑顺序进行组织格式化后作为上下文提示Prompt的一部分输入给LLM核心。这里的关键技巧是需要在提示词中明确告知LLM哪些是“近期具体细节”哪些是“长期总结的经验”帮助LLM更好地理解和利用这些信息。这种双路检索、统一融合的机制确保了智能体在决策时既能“看清脚下的路”工作记忆也能“望见远方的山”长期记忆。3. 核心组件深度解析与实操要点理解了设计思路我们来看看构建D-Mem系统需要哪些核心组件以及每个组件在实现时的关键决策点和实操细节。3.1 记忆编码器与向量化策略记忆的存储始于编码。选择什么样的模型将文本转化为向量嵌入直接影响后续检索的质量。模型选型通用vs专用对于大多数应用使用开源的通用文本嵌入模型如text-embedding-3-small,BGE-M3,Snowflake Arctic Embed是性价比很高的起点。它们的通用语义理解能力已经很强。微调以获得领域优势如果你的智能体专注于法律、医疗或金融等专业领域可以考虑用领域内的数据对通用嵌入模型进行轻量级微调例如使用LoRA技术。这能显著提升专业术语和领域概念的表示质量。工作记忆的轻量化编码对于工作记忆由于对延迟极其敏感可以考虑使用更小、更快的模型或者对嵌入进行降维如PCA到128维在精度和速度间取得平衡。分块Chunking策略长期记忆的分块切忌简单按固定字数分割。应采用基于语义的分块例如使用滑动窗口结合句子边界检测确保每个“块”是一个相对完整的语义单元如一个事件描述、一个问答对、一个方法步骤。对于代码、结构化数据需要特殊的处理规则。工作记忆的“块”工作记忆的“块”可能就是单轮的对话消息但需要附加丰富的元数据如role用户/助手、timestamp、parent_message_id用于构建对话树、priority手动或自动赋予的优先级。实操心得分块大小没有黄金标准。从256-512个token的块开始实验是个好主意。太小会导致信息碎片化太大会降低检索精度。一个有效的技巧是“重叠分块”即让相邻的块有少量文字重叠如50个token这可以防止一个完整的语义单元被恰好切分在两个块的边界而丢失。3.2 记忆存储与索引架构存储层需要同时支持工作记忆的快速键值访问和长期记忆的稠密向量检索。工作记忆存储推荐使用内存数据库或高性能键值存储如Redis。将当前会话ID作为主键存储一个结构化的消息列表。Redis的过期TTL功能可以自动清理过期会话的记忆非常方便。消息结构应包含原始文本、嵌入向量可选、以及丰富的元数据。长期记忆存储向量数据库是核心Pinecone、Weaviate、Qdrant、Milvus是主流选择。它们专为大规模向量相似性搜索优化。选型考量评估维度包括云服务/自托管、分布式支持、过滤查询性能、多向量支持如为同一文本存储不同模型的嵌入、成本。对于初创项目Pinecone的易用性很吸引人对于需要深度控制和定制的大型应用自托管Weaviate或Qdrant可能更合适。混合搜索现代向量数据库都支持“混合搜索”即结合向量相似性得分和基于元数据如时间、标签、来源的过滤/加权分数。这是实现精准检索的关键。例如你可以设置“最近三个月内的记忆权重加倍”。元数据设计这是长期记忆的“导航系统”。必须精心设计。常见的元数据字段包括session_id: 所属对话会话。timestamp: 精确时间戳。memory_type:raw_interaction原始交互或summary摘要或user_profile用户画像。tags: 数组字段存储多个标签如[“bug_fix”, “python”, “high_priority”]。source: 信息来源。confidence: 对于自动生成的摘要可以附加一个置信度分数。3.3 记忆检索、重排序与融合引擎这是D-Mem系统的“大脑皮层”负责协调两个记忆进程。检索Retrieval工作记忆检索通常是直接拉取当前会话的完整消息链或根据内部的消息ID和引用关系进行查询。复杂度低但要求数据结构设计合理。长期记忆检索这是性能关键路径。使用查询文本的嵌入向量在向量数据库中进行相似性搜索如余弦相似度。务必开启元数据过滤例如只检索memory_type为summary的高阶记忆或者只检索某个特定项目相关的记忆。这能大幅提升召回结果的相关性。重排序Re-Ranking为什么需要向量检索召回阶段追求的是“全”可能会返回一些语义相关但实际无用的结果。重排序阶段追求的是“精”。如何实现使用一个交叉编码器Cross-Encoder模型如BGE-reranker。它将查询文本和每一个候选记忆文本成对地输入模型直接输出一个相关性分数。这个计算比向量点积更精细但代价是计算量更大因为需要为每个候选都算一次。因此通常的做法是先用向量检索快速召回100-200个候选再用重排序模型对这100-200个结果进行精排选出Top-5或Top-10。训练自定义重排序器如果你的领域非常特殊可以使用任务相关的数据正例查询和相关文档负例查询和不相关文档来微调一个重排序模型这能带来质的提升。融合Fusion与上下文构建将来自工作记忆和长期记忆的、经过重排序的最终候选列表合并。去重基于嵌入向量相似度或文本指纹去除内容高度重复的片段。上下文窗口分配LLM的上下文长度是有限的。你需要一个策略来决定给工作记忆和长期记忆各分配多少“额度”。一个动态策略是始终优先保证工作记忆的完整性例如最近10轮对话剩余的空间再按比例分配给长期记忆中相关性最高的片段。提示词工程这是将记忆“喂”给LLM的最后一步也是至关重要的一步。不要简单地把记忆文本拼接起来。应该用清晰的结构化格式例如以下是与你当前任务相关的背景信息 【近期对话上下文工作记忆】 1. [消息1] 2. [消息2] ... 【相关历史经验与总结长期记忆】 * 用户偏好... * 项目历史... * 相关方法... 请基于以上信息回答用户的问题。明确区分记忆来源能极大帮助LLM理解并运用这些信息。4. 系统实现流程与核心环节让我们以一个“智能研发助手”为例串联起D-Mem的实现流程。4.1 环境搭建与初始化假设我们使用Python作为主要开发语言。基础设施部署部署一个Redis实例用于工作记忆。可以使用Docker快速启动docker run -p 6379:6379 redis。部署或连接一个向量数据库。以Qdrant为例使用Dockerdocker run -p 6333:6333 qdrant/qdrant。你也可以直接使用Pinecone的云服务无需管理服务器。核心库安装pip install openai langchain qdrant-client redis sentence-transformersopenai/anthropic用于调用LLM API。langchain虽然我们可能不完全遵循其框架但其丰富的集成和工具类很有参考价值。qdrant-clientQdrant客户端。redisRedis客户端。sentence-transformers用于本地运行嵌入模型和重排序模型。配置管理创建配置文件如config.yaml集中管理API密钥、数据库连接地址、模型名称、分块大小、检索数量等参数。这便于后续调整和实验。4.2 记忆写入流程当智能体完成一轮交互用户输入智能体响应触发记忆写入。import uuid from datetime import datetime from sentence_transformers import SentenceTransformer # 初始化模型和客户端 embed_model SentenceTransformer(BAAI/bge-small-zh-v1.5) # 中文嵌入模型 rerank_model SentenceTransformer(BAAI/bge-reranker-base) # 重排序模型 redis_client ... qdrant_client ... def save_to_memory(session_id: str, user_input: str, agent_response: str, task_type: str): 保存单轮交互到记忆系统 memory_id str(uuid.uuid4()) timestamp datetime.utcnow().isoformat() # 1. 构建原始记忆文本 raw_text fUser: {user_input}\nAssistant: {agent_response} # 2. 生成嵌入向量 embedding embed_model.encode(raw_text).tolist() # 3. 提取/生成标签此处简化实际可用LLM分析生成 tags extract_tags_with_llm(raw_text, task_type) # 假设的函数 # 4. 存入工作记忆 (Redis) wm_key fwm:{session_id} memory_item { id: memory_id, text: raw_text, timestamp: timestamp, tags: tags, embedding: embedding # 可选存储用于快速计算 } redis_client.lpush(wm_key, json.dumps(memory_item)) # 左插入最新在最前 redis_client.ltrim(wm_key, 0, 49) # 只保留最近50条 # 5. 存入长期记忆 (Qdrant) qdrant_client.upsert( collection_namelong_term_memory, points[ { id: memory_id, vector: embedding, payload: { text: raw_text, session_id: session_id, timestamp: timestamp, memory_type: raw_interaction, tags: tags, task_type: task_type } } ] ) # 6. 异步触发长期记忆的摘要生成 # 可以定期如每24小时或在特定事件后对某个session或特定tag的raw memory进行批处理摘要。 # schedule_summarization_job(session_id, tags)4.3 记忆检索与响应生成流程当新的用户查询到达时。def retrieve_and_respond(session_id: str, query: str): 核心检索与响应流程 # 1. 从工作记忆中检索 wm_key fwm:{session_id} raw_wm_items redis_client.lrange(wm_key, 0, 9) # 取最近10条 working_memories [json.loads(item) for item in raw_wm_items] # 2. 从长期记忆中检索 query_embedding embed_model.encode(query).tolist() # 第一步向量相似性检索召回 search_results qdrant_client.search( collection_namelong_term_memory, query_vectorquery_embedding, query_filterFilter( # 示例过滤条件提升相关性 must[ FieldCondition(keymemory_type, matchMatchValue(valuesummary)), # 优先检索摘要 FieldCondition(keytask_type, matchMatchValue(valuecoding)), # 假设当前是编程任务 ] ), limit100 # 召回100个候选 ) # 准备重排序的候选列表 candidates [] # 加入工作记忆候选可以赋予较高初始分因为相关性最高 for wm in working_memories: candidates.append({ text: wm[text], source: working_memory, original_score: 1.0 # 最高初始分 }) # 加入长期记忆候选 for hit in search_results: candidates.append({ text: hit.payload[text], source: long_term_memory, original_score: hit.score }) # 3. 重排序 # 构建 (query, candidate_text) 对 pairs [(query, cand[text]) for cand in candidates] # 使用交叉编码器计算精细相关性分数 rerank_scores rerank_model.encode(pairs, convert_to_tensorTrue, normalize_embeddingsTrue) # 假设rerank_scores返回的是相似度矩阵的对角线 for i, cand in enumerate(candidates): cand[rerank_score] rerank_scores[i][i].item() if rerank_scores.dim() 1 else rerank_scores[i].item() # 4. 排序、去重、选择Top-K candidates.sort(keylambda x: x[rerank_score], reverseTrue) # 简单基于文本哈希的去重 seen set() final_memories [] for cand in candidates: text_hash hash(cand[text][:500]) # 取前500字符哈希 if text_hash not in seen: seen.add(text_hash) final_memories.append(cand) if len(final_memories) 15: # 最终选择15条记忆 break # 5. 构建上下文提示 context_prompt build_context_prompt(final_memories, query) # 自定义函数格式化记忆 # 6. 调用LLM生成最终响应 llm_response call_llm_api(context_prompt) # 调用OpenAI/Claude等 # 7. 将本轮新的交互query, llm_response保存回记忆调用save_to_memory save_to_memory(session_id, query, llm_response, task_typecoding) return llm_response4.4 长期记忆的摘要与提炼后台进程这是一个独立的后台任务定期运行。def summarize_long_term_memory(session_id: str, time_range: tuple): 为特定会话在特定时间范围内的原始记忆生成摘要 # 1. 从向量数据库检索出该时间段内的所有原始交互 raw_memories qdrant_client.scroll( collection_namelong_term_memory, scroll_filterFilter( must[ FieldCondition(keysession_id, matchMatchValue(valuesession_id)), FieldCondition(keytimestamp, rangeRange(...)), # 时间范围 FieldCondition(keymemory_type, matchMatchValue(valueraw_interaction)), ] ) ) # 2. 将原始文本聚合 aggregated_text \n---\n.join([hit.payload[text] for hit in raw_memories]) # 3. 调用LLM进行摘要生成 summary_prompt f 你是一个智能助手。请根据以下与用户的历史交互记录提炼出 1. 用户的核心需求和偏好。 2. 我们共同完成的主要任务和关键决策。 3. 用户反馈中提到的任何重要问题或赞赏。 请用简洁、结构化的要点形式输出。 历史交互记录 {aggregated_text} summary call_llm_api(summary_prompt, modelgpt-4) # 使用能力更强的模型进行摘要 # 4. 将摘要作为新的高阶记忆存入长期记忆 summary_embedding embed_model.encode(summary).tolist() summary_id str(uuid.uuid4()) qdrant_client.upsert( collection_namelong_term_memory, points[ { id: summary_id, vector: summary_embedding, payload: { text: summary, session_id: session_id, timestamp: datetime.utcnow().isoformat(), memory_type: summary, source_memory_ids: [hit.id for hit in raw_memories], # 关联原始记忆 tags: [auto_summary, user_profile] } } ] ) # 可选5. 可以标记原始记忆已被摘要或将其移至归档集合避免未来被重复检索。5. 常见问题、排查技巧与性能优化在实际部署D-Mem系统时你会遇到各种预料之外的问题。以下是一些典型问题及其解决思路。5.1 检索结果不相关或噪声大症状LLM的回复明显被无关的历史信息带偏。排查与解决检查嵌入模型你的嵌入模型是否与任务领域匹配尝试用一些典型的查询-相关文档对计算其嵌入相似度看分数是否合理。考虑更换或微调模型。强化元数据过滤这是最有效的改进手段之一。确保你的记忆在写入时打上了足够丰富和准确的标签task_type,project,sentiment等。在检索时充分利用这些元数据进行前置过滤。例如在代码调试场景下只检索tags包含“error”和“python”的记忆。调整重排序模型默认的重排序模型可能不适合你的数据。尝试不同的模型或者用你自己构造的查询相关/不相关文档数据对进行微调哪怕只有几百个样本效果也可能有显著提升。审视分块策略记忆块是否太大包含了多个不相关的主题尝试减小分块大小或采用更智能的语义分割如用LLM判断段落边界。5.2 系统延迟过高症状用户查询响应时间很长体验差。排查与解决性能剖析使用 profiling 工具如Python的cProfile定位瓶颈。通常瓶颈在嵌入生成、向量数据库检索、重排序、LLM API调用。嵌入缓存对常见的、不变的查询文本如“总结一下”、“帮我写个邮件模板”的嵌入结果进行缓存避免重复计算。异步与非阻塞将记忆的写入操作尤其是存入向量数据库改为异步。用户无需等待记忆持久化完成即可收到响应。确保你的消息队列或异步任务框架可靠。限制检索规模严格控制向量检索的候选数量limit参数和重排序的候选数量。从50开始测试逐步增加观察精度和延迟的平衡点。工作记忆优化确保工作记忆的读写是纯内存操作避免任何不必要的序列化/反序列化开销。5.3 记忆冲突与信息过时症状长期记忆中的旧信息如过时的项目状态、已被纠正的用户偏好干扰了当前决策。排查与解决实施记忆衰减或版本管理为记忆片段引入“置信度”或“新鲜度”分数该分数随时间推移或根据后续用户的明确否定而下降。在检索结果融合时将这个分数作为权重因子。显式记忆更新与废止当用户明确说“我之前说的X不对应该是Y”时系统应能触发一个流程检索出包含X的记忆要么将其标记为废止要么用新的记忆覆盖它在向量库中新增一条并建立关联。在提示词中强调时效性在构建给LLM的上下文时明确标注每条记忆的时间戳并加入指令“请优先依据最新的信息进行判断旧信息可能已过时。”5.4 长期记忆摘要质量差症状自动生成的摘要空洞、错误或丢失关键信息。排查与解决优化摘要提示词给你的摘要LLM更具体、更结构化的指令。要求它按特定维度如事实、决策、偏好、问题进行总结并给出例子Few-shot。分阶段摘要不要一次性总结大量文本。可以先按时间或主题对原始记忆进行聚类对每个簇分别生成摘要然后再对这些摘要进行二次归纳。人工审核与迭代在初期将自动摘要提供给人工审核纠正错误。这些纠正后的数据可以作为微调摘要模型或优化提示词的宝贵素材。控制摘要粒度为不同“级别”的记忆生成不同粒度的摘要。例如“会话级摘要”单次对话、“任务级摘要”一个完整功能开发、“用户级摘要”整体画像。5.5 成本控制挑战嵌入模型调用、LLM API调用尤其是摘要生成、向量数据库存储与查询都可能产生可观成本。优化策略分层存储与冷热分离将很少被访问的旧记忆从高性能向量数据库迁移到更便宜的对象存储如S3并只保留其元数据和关键嵌入在向量库中用于粗略检索。当需要细节时再从对象存储中加载。摘要替代原始文本一旦生成了高质量的摘要在大多数检索场景下可以优先检索摘要而非海量的原始交互文本。这大大减少了需要处理的文本量和注入LLM上下文的长度。选择性摘要并非所有对话都值得摘要。可以设定规则只对那些交互轮次多、涉及重要操作如工具调用成功/失败、或包含明确反馈用户说“很好”或“不对”的会话进行摘要。使用更经济的模型对于嵌入生成可以评估更小尺寸的模型。对于摘要生成可以尝试使用gpt-3.5-turbo而非gpt-4并通过更好的提示词来弥补模型能力的差距。构建一个健壮的D-Mem系统是一个持续迭代的过程。从最简单的双路检索开始逐步引入重排序、动态摘要、记忆衰减等高级特性。关键是在每一步都建立评估机制用真实的用户查询和任务成功率来衡量每一次架构改进的实际效果。记住记忆系统的终极目标不是存储最多而是让智能体在需要时能想起最该想起的事情。