1. 项目概述为什么Agent需要“双层记忆”最近在折腾AI Agent开发的朋友估计都遇到过类似的头疼事你跟Agent聊得好好的让它帮你分析一份文档、写段代码它当时表现得头头是道。可等你过几天再打开一个新会话想让它基于上次的分析结果继续深入时它却一脸茫然仿佛得了“健忘症”一切又得从头开始。或者在同一个长会话里聊到几百条消息之后Agent开始前言不搭后语把用户A的需求和用户B的指令搞混出现记忆“乱窜”。这些问题的核心都指向了Agent系统里一个关键但常被忽视的组件——记忆模块。我们通常说的“记忆”在AI Agent语境下远不止是记住对话历史那么简单。一个功能完备的记忆系统至少要解决两个层面的问题会话连续性和长期知识沉淀。会话连续性保证在一次交互中Agent能理解上下文做出连贯的回应而长期知识沉淀则能让Agent在跨越不同时间、不同会话的多次交互中积累经验变得越来越“聪明”和个性化。这就像人的记忆有短期工作记忆和长期记忆一样Agent也需要这样的“双层记忆”架构。单纯依赖大语言模型LLM自带的上文窗口比如128K、200K来实现记忆是远远不够且低效的。一方面上下文长度有限长对话必然面临信息丢失或模型性能下降这就是为什么你会看到“你和kimi聊得太长啦发起一个新会话试试吧”的提示。另一方面把所有历史对话都塞进上下文会带来极高的计算Token成本和响应延迟。更重要的是未经处理的原始对话记录是杂乱无章的无法被高效地检索和利用形成真正的“知识”。因此为Agent设计一个独立于模型上下文之外的、结构化的外部记忆系统就成了Agent走向实用和强大的必经之路。这个系统需要能可靠地存储信息更需要能智能地检索、更新和关联信息。今天要讨论的“双层记忆”架构以及结合PostgresSaver这类工具的实现思路就是针对这一痛点的一次工程实践。无论你是想开发一个能记住用户偏好的个人助手还是一个能在多次调试中学习最佳实践的编码Agent这套思路都能提供直接的参考。2. 记忆系统架构设计从理论到蓝图在动手写代码之前我们必须先把架构想清楚。一个健壮的Agent记忆系统不能是简单的聊天记录保存而应该是一个有清晰层次、职责分明的数据处理管道。2.1 三层记忆模型解析参考学术界和工业界的实践一个完整的Agent记忆模型可以抽象为三个层次我们称之为“三层记忆架构”感官记忆/即时记忆这是最原始的一层对应Agent对单次用户输入包括文本、图像、工具调用结果等的瞬时感知和解析。它存在时间极短主要用于初步理解用户的意图和提取关键信息片段。在工程上这一层通常被融合在输入处理管道中。工作记忆/短期记忆这是实现会话连续性的核心。它负责维护当前会话的上下文通常有一个容量限制例如最近10轮对话或最近N个Token。它的目标是保证Agent在本次对话中的回应是连贯、一致的。很多框架的“上下文管理”或“对话历史管理”模块本质上就是在做工作记忆。当对话超过限制时需要有一套策略如摘要、选择性遗忘来压缩或移出旧信息。长期记忆这是实现知识沉淀的关键。它用于存储跨越多个会话的、被认为有价值的信息。这些信息不是简单的对话日志而是经过提炼、结构化或向量化后的“知识”。例如用户的个人偏好、项目特定的规则、从成功经验中总结的模式、从失败中吸取的教训等。长期记忆的容量理论上可以非常大其挑战在于如何高效地存储和检索。我们项目标题中的“双层记忆”主要聚焦于工作记忆和长期记忆这两层因为它们是构建智能体“人格”和“能力”的支柱。感官记忆则更多是预处理环节。2.2 关键组件与数据流设计基于三层模型我们可以设计出系统的核心组件和数据流记忆编码器负责将原始的、非结构化的对话信息转化为适合存储和检索的格式。这可能包括文本分块与清洗将长文本分割成有意义的片段。元数据提取自动或手动为记忆片段打上标签如会话ID、时间戳、用户ID、主题、实体如涉及的人名、项目名、动作类型如“查询”、“指令”、“错误”等。向量化使用嵌入模型如text-embedding-3-small将文本转换为向量这是实现语义检索的基础。记忆存储库这是记忆的物理载体。一个混合存储方案通常是更优解向量数据库用于存储向量和关联的元数据支持高效的相似性搜索语义检索。这是从长期记忆中召回相关知识的核心。关系型数据库用于存储结构化的会话元数据、用户信息、以及那些不适合或不需要向量检索的记忆如精确的键值对配置。PostgreSQL因其可靠性、扩展性以及对JSON字段的良好支持成为许多项目的首选。这也是PostgresSaver这个名字的由来——它很可能是一个将记忆保存到PostgreSQL的工具或模块。缓存如Redis用于存储活跃会话的工作记忆提供极快的读写速度保证对话的实时性。记忆检索器在Agent需要生成回复时根据当前查询用户问题最近上下文从记忆存储库中召回最相关的信息。检索策略至关重要基于时间的检索优先召回最近发生的记忆。基于语义的检索使用当前查询的向量在向量库中搜索最相似的记忆片段。基于元数据的过滤例如只检索属于当前用户或当前项目的记忆。混合检索综合运用以上多种策略并对结果进行重排序Rerank得到最相关的记忆子集。记忆更新与遗忘机制记忆不是只进不出的。系统需要有能力更新当新信息与旧记忆冲突或补充时更新原有记忆例如用户更新了手机号。合并将多个相关的、细碎的记忆片段合并成一个更完整、更简洁的知识点。遗忘/归档定期清理过期、无用或低价值的记忆或将它们移至冷存储以控制存储成本和保持记忆库的“健康度”。注意这里的数据流是一个简化的理想模型。在实际开发中特别是初期你可能不需要实现所有组件。一个常见的起步方案是用PostgreSQL同时存储结构化元数据和向量利用其pgvector扩展用简单的最近邻检索ORDER BY embedding query_embedding来实现语义搜索。这能大大降低系统的复杂度。2.3 工具与框架选型考量看到热搜词里有LangGraph、Hermes Agent、PostgresSaver等这些都是当前生态中相关的工具或框架。选型时需要权衡LangGraph它是一个用于构建有状态、多智能体工作流的框架。其“状态”管理天然适合维护工作记忆。你可以将整个对话历史或记忆指针作为状态的一部分在图中传递。对于构建复杂的、涉及多个步骤和记忆访问的Agent流程LangGraph是一个强有力的候选。Hermes Agent或其他开源Agent框架这些框架通常已经内置了一定程度的记忆管理模块。使用它们可以快速起步但可能需要根据你的“双层记忆”需求进行定制或扩展。需要仔细阅读其文档看其记忆模块是否支持外部数据库存储、向量检索和长期记忆隔离。PostgresSaver从名字推断这很可能是一个专门用于将记忆可能是特定框架如LangChain或LlamaIndex的记忆对象持久化到PostgreSQL的类或工具。如果你的技术栈匹配直接使用它可以节省大量底层数据库操作的开发时间。向量数据库除了PostgreSQL的pgvector还有Chroma轻量、易用、Qdrant高性能、云原生、Weaviate功能丰富等选项。选择时考虑部署复杂度、性能、社区支持和与现有框架的集成度。实操心得对于个人项目或小团队我强烈建议从“PostgreSQL pgvector”开始。它避免了维护多个数据库的运维负担利用SQL能轻松处理复杂的元数据查询且pgvector的性能对于中小规模应用完全足够。这正符合PostgresSaver所倡导的“一体化存储”思路。3. 核心实现构建双层记忆系统理论说得再多不如一行代码。接下来我们以Python为例勾勒出实现双层记忆系统的核心代码骨架。我们将假设使用LangChain作为基础框架因其生态丰富并整合pgvector进行向量存储。3.1 环境准备与依赖安装首先确保你的环境已经就绪。# 安装核心库 pip install langchain langchain-community langchain-openai # 安装PostgreSQL驱动和向量扩展支持 pip install psycopg2-binary pgvector # 安装文本嵌入模型这里以OpenAI为例也可用sentence-transformers等开源模型 pip install openai # 可选安装LangChain对PostgreSQL向量存储的集成如果langchain-community中已有则无需单独安装 # pip install langchain-postgres在PostgreSQL中你需要先启用pgvector扩展并创建存储记忆的表。-- 在目标数据库中执行 CREATE EXTENSION IF NOT EXISTS vector; -- 创建一个存储记忆的表 CREATE TABLE agent_memories ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), session_id VARCHAR(255) NOT NULL, -- 会话标识用于工作记忆隔离和长期记忆关联 user_id VARCHAR(255), -- 用户标识用于长期记忆的用户隔离 memory_type VARCHAR(50) CHECK (memory_type IN (working, long_term)), -- 记忆类型 content TEXT NOT NULL, -- 记忆的原始文本内容 embedding vector(1536), -- 假设使用text-embedding-3-small维度为1536 metadata JSONB DEFAULT {}::jsonb, -- 存储标签、实体、时间戳等元数据 created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP ); -- 为常用查询创建索引 CREATE INDEX idx_memories_session ON agent_memories(session_id); CREATE INDEX idx_memories_user ON agent_memories(user_id); CREATE INDEX idx_memories_type ON agent_memories(memory_type); -- 为向量相似性搜索创建索引根据数据量选择索引类型如ivfflat或hnsw CREATE INDEX idx_memories_embedding ON agent_memories USING ivfflat (embedding vector_cosine_ops) WITH (lists 100);3.2 记忆管理器的实现我们将创建一个MemoryManager类它封装了工作记忆和长期记忆的读写逻辑。import json from datetime import datetime, timedelta from typing import List, Dict, Any, Optional, Tuple import psycopg2 from psycopg2.extras import Json from langchain.embeddings import OpenAIEmbeddings from langchain.schema import Document from langchain.vectorstores import PGVector from langchain.vectorstores.pgvector import DistanceStrategy class MemoryManager: def __init__(self, connection_string: str, embedding_model: OpenAIEmbeddings, working_memory_limit: int 10): 初始化记忆管理器。 :param connection_string: PostgreSQL连接字符串 :param embedding_model: 嵌入模型实例 :param working_memory_limit: 工作记忆在缓存中保留的最近对话轮数 self.conn_string connection_string self.embedder embedding_model self.working_memory_limit working_memory_limit # 初始化PGVector连接用于长期记忆的语义检索 # 注意PGVector会使用自己的表这里仅为演示。实际可统一使用上面自建的表。 # 为了简化我们下面直接使用psycopg2操作自建表来模拟。 self.conn psycopg2.connect(connection_string) self.cur self.conn.cursor() # 内存中的工作记忆缓存 {session_id: [messages]} self.working_memory_cache {} def _get_embedding(self, text: str) - List[float]: 生成文本的向量嵌入。 return self.embedder.embed_query(text) def add_memory(self, session_id: str, user_id: str, content: str, memory_type: str working, metadata: Optional[Dict] None): 添加一条记忆到存储。 if metadata is None: metadata {} # 添加系统元数据 metadata.update({ timestamp: datetime.utcnow().isoformat(), source: dialogue }) embedding self._get_embedding(content) sql INSERT INTO agent_memories (session_id, user_id, memory_type, content, embedding, metadata) VALUES (%s, %s, %s, %s, %s::vector, %s) self.cur.execute(sql, (session_id, user_id, memory_type, content, embedding, Json(metadata))) self.conn.commit() # 如果是工作记忆同时更新缓存 if memory_type working: if session_id not in self.working_memory_cache: self.working_memory_cache[session_id] [] self.working_memory_cache[session_id].append({content: content, metadata: metadata}) # 保持工作记忆缓存不超过限制 if len(self.working_memory_cache[session_id]) self.working_memory_limit: self.working_memory_cache[session_id].pop(0) def retrieve_working_memory(self, session_id: str, limit: int 5) - List[str]: 检索当前会话的工作记忆优先从缓存缓存没有则查库。 if session_id in self.working_memory_cache and self.working_memory_cache[session_id]: # 返回最近几条的内容 memories self.working_memory_cache[session_id][-limit:] return [m[content] for m in memories] else: # 从数据库查询 sql SELECT content FROM agent_memories WHERE session_id %s AND memory_type working ORDER BY created_at DESC LIMIT %s self.cur.execute(sql, (session_id, limit)) results self.cur.fetchall() return [r[0] for r in results] def retrieve_long_term_memory(self, query: str, user_id: str, top_k: int 3, filter_sessions: Optional[List[str]] None) - List[Tuple[str, float]]: 基于语义检索用户的长期记忆。 query_embedding self._get_embedding(query) # 构建查询条件 conditions [memory_type long_term, user_id %s] params [user_id] if filter_sessions: # 例如排除当前会话避免信息重复 placeholders , .join([%s] * len(filter_sessions)) conditions.append(fsession_id NOT IN ({placeholders})) params.extend(filter_sessions) where_clause AND .join(conditions) # 使用pgvector的余弦相似度操作符 sql f SELECT content, 1 - (embedding %s::vector) as similarity FROM agent_memories WHERE {where_clause} ORDER BY embedding %s::vector LIMIT %s params params [query_embedding, query_embedding, top_k] self.cur.execute(sql, params) results self.cur.fetchall() # 返回(内容相似度得分)的列表 return [(r[0], r[1]) for r in results] def promote_to_long_term(self, memory_ids: List[str]): 将指定的工作记忆提升为长期记忆。 # 这是一个关键操作意味着某条记忆被判定为有价值需要永久保存。 # 在实际中可能还需要对内容进行总结、提炼后再存储。 placeholders , .join([%s] * len(memory_ids)) sql f UPDATE agent_memories SET memory_type long_term, updated_at CURRENT_TIMESTAMP WHERE id IN ({placeholders}) self.cur.execute(sql, tuple(memory_ids)) self.conn.commit() def close(self): 关闭数据库连接。 self.cur.close() self.conn.close() # 初始化示例 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) memory_manager MemoryManager( connection_stringpostgresql://user:passwordlocalhost:5432/agent_db, embedding_modelembeddings, working_memory_limit15 )3.3 与Agent的集成策略有了记忆管理器下一步就是将其嵌入到Agent的推理循环中。一个典型的集成点在Agent生成回复之前我们需要为它组装上下文。from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain.tools import Tool class AgentWithMemory: def __init__(self, memory_manager: MemoryManager, llm: ChatOpenAI): self.memory memory_manager self.llm llm # 定义一个“搜索长期记忆”的工具供Agent在需要时主动调用 search_memory_tool Tool( namesearch_long_term_memory, funcself._search_memory_func, description当用户的问题涉及过去的知识、偏好或历史信息时使用此工具搜索长期记忆。输入应为清晰的搜索查询语句。 ) # 构建Agent这里是一个简单的OpenAI Tools Agent示例 prompt ChatPromptTemplate.from_messages([ (system, 你是一个有帮助的AI助手拥有与用户交互的记忆。 当前会话的工作记忆会自动提供给你。 如果你认为需要查询用户更早的历史信息请使用search_long_term_memory工具。 请基于所有可用信息给出准确、连贯的回答。), MessagesPlaceholder(variable_namechat_history), # 这里将填入工作记忆 (human, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), ]) self.agent create_openai_tools_agent(llm, [search_memory_tool], prompt) self.agent_executor AgentExecutor(agentself.agent, tools[search_memory_tool], verboseTrue) def _search_memory_func(self, query: str): 工具函数搜索长期记忆。在实际中这里需要获取当前的user_id和session_id。 # 假设我们能从上下文中获取当前用户和会话ID current_user_id user_123 current_session_id session_abc results self.memory.retrieve_long_term_memory( queryquery, user_idcurrent_user_id, filter_sessions[current_session_id] # 排除当前会话避免重复 ) if results: return \n.join([f[相关记忆 {i1}, 相关性{score:.2f}]: {content} for i, (content, score) in enumerate(results)]) else: return 在长期记忆中未找到相关信息。 def run(self, session_id: str, user_id: str, user_input: str) - str: 处理用户输入的核心方法。 # 1. 检索工作记忆作为对话历史 working_memories self.memory.retrieve_working_memory(session_id, limit10) # 将工作记忆转换为LangChain的Message格式这里简化为字符串连接 chat_history_str \n.join(working_memories) if working_memories else 无历史对话。 # 2. 调用Agent执行传入工作记忆作为历史 # 注意这里需要将chat_history_str适配成正确的Message格式以下为简化示例 inputs { input: user_input, chat_history: chat_history_str, # 实际应用中需转换为List[BaseMessage] } response self.agent_executor.invoke(inputs) # 3. 将本轮交互存入记忆用户输入和AI回复 self.memory.add_memory(session_id, user_id, fUser: {user_input}, memory_typeworking) self.memory.add_memory(session_id, user_id, fAssistant: {response[output]}, memory_typeworking) # 4. 可选判断是否将某些重要信息提升为长期记忆 # 这是一个高级功能可以通过规则或另一个LLM调用来判断。 # if self._should_promote_to_long_term(user_input, response[output]): # memory_id ... # 获取最近记忆的ID # self.memory.promote_to_long_term([memory_id]) return response[output] def _should_promote_to_long_term(self, user_input: str, assistant_output: str) - bool: 启发式规则或LLM调用判断对话是否包含值得长期保存的知识。 # 示例规则如果用户表达了明确的偏好或提供了关键事实 promotion_keywords [我喜欢, 我讨厌, 我的偏好是, 记住一下, 重要, 以后都用这个] for keyword in promotion_keywords: if keyword in user_input: return True # 更复杂的实现可以用一个轻量级LLM来判断 return False # 使用示例 llm ChatOpenAI(modelgpt-4-turbo-preview) agent_with_memory AgentWithMemory(memory_manager, llm) # 模拟一次对话 response agent_with_memory.run( session_idsession_abc, user_iduser_123, user_input我更喜欢用Python而不是Java来写后端服务。 ) print(fAssistant: {response}) # 在另一个会话或未来某次对话中 response2 agent_with_memory.run( session_idsession_def, # 新的会话 user_iduser_123, # 同一个用户 user_input我之前说过我喜欢用什么语言写后端 ) # 理想情况下Agent应能通过检索长期记忆回答“您之前表示过更喜欢用Python而不是Java来写后端服务。”关键点解析工作记忆的自动注入在每次调用run时自动从MemoryManager中取出当前会话的近期对话作为上下文chat_history提供给Agent。这保证了会话的连续性。长期记忆的按需检索我们将其封装成一个Tool。Agent在推理过程中如果认为需要查询更早的、跨会话的信息可以主动调用这个工具。这是一种“检索增强生成RAG”模式在记忆系统中的应用。记忆的写入时机在Agent回复后立即将用户输入和AI回复作为一对“工作记忆”保存。这保证了记忆的实时性。记忆晋升机制promote_to_long_term方法和_should_promote_to_long_term函数展示了如何将重要的临时对话转化为长期知识。这是实现“知识沉淀”的核心步骤可以基于规则、关键词或另一个LLM分类器来实现。4. 高级议题与实战避坑指南实现基础的双层记忆系统后我们会面临更多工程上的挑战。以下是几个关键的高级议题和从实战中总结的避坑经验。4.1 记忆的隔离、安全与隐私记忆“乱窜”是热搜词中提到的一个具体问题。这涉及到记忆的隔离。隔离维度会话隔离这是最基本的。session_id是确保不同对话之间工作记忆不混淆的关键。每次新的聊天窗口或连接都应生成唯一的session_id。用户隔离长期记忆必须严格按user_id进行隔离。用户A绝对不能看到用户B的记忆。在检索长期记忆时user_id是必须的过滤条件。租户/项目隔离在更复杂的系统如企业级应用中可能还需要tenant_id或project_id来实现更细粒度的隔离。安全与隐私敏感信息过滤在记忆存入数据库前应有管道对内容进行扫描过滤或脱敏手机号、邮箱、身份证号等个人敏感信息PII。记忆访问控制除了存储隔离在检索端也要有严格的权限校验。确保每次检索请求都带有合法的、经过认证的用户身份信息。遗忘权提供API让用户可以查看、编辑或删除自己的特定记忆这是合规性如GDPR的要求。实操心得在数据库设计时为所有记忆表加上user_id和session_id字段并以此建立复合索引。在每一次数据库查询中都显式地带上这些过滤条件形成“防御性编程”习惯从根本上避免数据泄露。4.2 记忆的压缩、摘要与更新记忆不能无限膨胀低质量、冗余的记忆会污染检索结果。工作记忆的压缩当对话轮数超过限制如working_memory_limit时简单的“先进先出”丢弃可能丢失重要早期信息。更好的策略是进行摘要。例如每5轮对话后用LLM对之前的对话内容生成一个简短的摘要然后用这个摘要替代那5轮原始对话作为一条新的“压缩记忆”放入工作记忆序列。这能在有限容量内保留更多信息。长期记忆的去重与合并定期运行后台任务对长期记忆进行聚类分析。将语义高度相似向量距离很近的记忆片段找出来用LLM将它们合并成一条更全面、更简洁的记忆并删除冗余的旧片段。这能显著提升记忆库的质量和检索效率。记忆的更新与修正如果用户说“我之前说的手机号错了应该是138xxxxxxx”系统需要能定位到之前那条错误的记忆通过语义检索或元数据标签并将其内容更新或标记为过期。这需要记忆有良好的版本管理或关联机制。4.3 性能优化与规模化当记忆量增长到数十万、百万条时性能问题会凸显。向量索引优化pgvector的ivfflat索引需要在使用一定数据量后REINDEX或者考虑使用hnsw索引以获得更好的查询性能和召回率。调整lists对于ivfflat或m/ef_construction对于hnsw参数以适应你的数据分布和查询负载。分级存储将很少访问的“冷记忆”转移到更便宜的存储如对象存储并在元数据中记录其位置。需要时再异步加载。这能降低主数据库的存储压力和成本。缓存策略工作记忆缓存如我们示例中所做在应用内存中使用字典缓存活跃会话的工作记忆避免对数据库的频繁读取。热点长期记忆缓存对于被频繁检索的长期记忆例如用户的常用地址可以将其内容或向量缓存在Redis中加速检索。异步写入记忆的写入尤其是向量生成和存储可以是异步操作不必阻塞Agent的响应流。可以将记忆写入任务放入消息队列如RabbitMQ、Kafka由后台Worker处理保证Agent的响应速度。4.4 评估记忆系统的有效性如何判断你的双层记忆系统是否真的提升了Agent的智能定性评估进行人工测试设计跨会话的问答对。例如Q1会话A“我的生日是7月10日。”Q2会话B“我的生日是什么时候” 检查Agent是否能正确回答。设计更复杂的场景如偏好记忆、事实记忆、任务步骤记忆等。定量评估检索相关性对于一批测试查询人工标注其应从长期记忆中召回的记忆片段计算系统实际召回结果的准确率Precision和召回率Recall。端到端任务成功率设计一系列依赖记忆的多步骤任务例如“根据我们上周讨论的架构图帮我生成一个部署清单”统计在有记忆系统和无记忆系统或仅有会话记忆下Agent独立完成任务的百分比。延迟与吞吐量监控记忆检索和存储操作的平均延迟、P99延迟以及系统能支撑的每秒记忆操作数OPS确保不影响用户体验。5. 典型问题排查与调试技巧在实际开发和运维中你肯定会遇到各种问题。下面是一个快速排查指南。问题现象可能原因排查步骤与解决方案Agent完全“忘记”了之前同一会话里的内容。1.session_id没有正确传递或生成。2. 工作记忆缓存失效或未命中且数据库查询失败。3. 记忆根本没有被成功写入数据库。1.检查日志确认每次请求的session_id是否稳定且唯一。检查add_memory和retrieve_working_memory的调用日志。2.检查数据库直接连接数据库查询对应session_id和memory_typeworking的记录是否存在。3.检查网络与连接确认应用与PostgreSQL的连接是否正常是否有连接池超时等问题。长期记忆检索不到任何内容或召回的内容不相关。1. 记忆晋升逻辑未触发长期记忆库为空。2. 向量索引未建立或性能差。3. 检索时user_id过滤条件错误或查询向量生成有问题。4. 嵌入模型不匹配存储和检索用的不是同一个模型。1.检查长期记忆表SELECT COUNT(*) FROM agent_memories WHERE memory_typelong_term。2.检查索引\d agent_memories查看索引是否建立。对于大量数据尝试REINDEX。3.调试检索查询打印出实际执行的SQL和参数手动在数据库客户端运行看结果。4.验证向量确保存储和检索时使用的是完全相同的嵌入模型和参数。计算一个已知文本的向量看存储和检索时是否一致。记忆出现“乱窜”用户A看到了用户B的信息。1. 数据库查询缺少user_id过滤条件或条件被错误覆盖。2. 缓存如Redis键设计有误导致不同用户的数据互相覆盖。1.代码审查仔细检查所有涉及记忆检索的SQL语句和缓存键生成逻辑确保user_id被强制使用。2.进行安全测试模拟两个用户同时操作检查日志和数据库查询记录看是否有越权查询。3.实施数据隔离测试作为上线前必检项。Agent响应速度变慢尤其是长对话后期。1. 工作记忆检索时未限制条数导致返回的历史上下文过长拖慢LLM处理速度。2. 长期记忆检索的top_k值设置过大或向量搜索未走索引。3. 记忆管理器的逻辑阻塞了主线程。1.优化工作记忆长度限制retrieve_working_memory的limit参数或引入摘要压缩机制。2.分析查询计划在数据库中对长期记忆检索的SQL执行EXPLAIN ANALYZE确认是否使用了向量索引。3.异步化将记忆的写入和复杂的检索操作改为异步非阻塞模式。存储空间增长过快。1. 所有对话都被无差别地永久存储了。2. 缺乏记忆压缩、合并和清理机制。1.实施TTL生存时间为工作记忆设置自动清理策略例如超过30天的会话记忆自动删除或归档。2.启动记忆维护任务定期运行脚本对长期记忆进行去重、合并和重要性评分删除低分记忆。调试技巧记录详细的记忆日志在关键节点记忆写入、晋升、检索打印结构化日志包含session_id,user_id,memory_id,操作类型和内容摘要。这比查看原始数据库记录直观得多。构建一个记忆查看器开发一个简单的内部管理界面可以按用户、会话、时间查询和浏览记忆内容。这在调试“记忆去哪了”的问题时无比有用。对记忆系统进行单元测试和集成测试模拟跨会话、多用户的复杂交互场景验证记忆的存储、检索、隔离是否正确。将测试用例纳入CI/CD流程。为Agent构建双层记忆系统是一个从“玩具”走向“工具”的关键步骤。它要求我们不仅关注AI模型本身更要重视围绕模型的数据工程和状态管理。从简单的会话历史保存到基于向量的语义检索再到复杂的记忆生命周期管理每一步都充满了工程上的权衡与挑战。但当你看到你的Agent能真正记住用户的喜好能在多次交互中积累经验并越用越聪明时这一切的努力都是值得的。 AI Agent双层记忆架构：基于PostgreSQL与向量检索的工程实践
1. 项目概述为什么Agent需要“双层记忆”最近在折腾AI Agent开发的朋友估计都遇到过类似的头疼事你跟Agent聊得好好的让它帮你分析一份文档、写段代码它当时表现得头头是道。可等你过几天再打开一个新会话想让它基于上次的分析结果继续深入时它却一脸茫然仿佛得了“健忘症”一切又得从头开始。或者在同一个长会话里聊到几百条消息之后Agent开始前言不搭后语把用户A的需求和用户B的指令搞混出现记忆“乱窜”。这些问题的核心都指向了Agent系统里一个关键但常被忽视的组件——记忆模块。我们通常说的“记忆”在AI Agent语境下远不止是记住对话历史那么简单。一个功能完备的记忆系统至少要解决两个层面的问题会话连续性和长期知识沉淀。会话连续性保证在一次交互中Agent能理解上下文做出连贯的回应而长期知识沉淀则能让Agent在跨越不同时间、不同会话的多次交互中积累经验变得越来越“聪明”和个性化。这就像人的记忆有短期工作记忆和长期记忆一样Agent也需要这样的“双层记忆”架构。单纯依赖大语言模型LLM自带的上文窗口比如128K、200K来实现记忆是远远不够且低效的。一方面上下文长度有限长对话必然面临信息丢失或模型性能下降这就是为什么你会看到“你和kimi聊得太长啦发起一个新会话试试吧”的提示。另一方面把所有历史对话都塞进上下文会带来极高的计算Token成本和响应延迟。更重要的是未经处理的原始对话记录是杂乱无章的无法被高效地检索和利用形成真正的“知识”。因此为Agent设计一个独立于模型上下文之外的、结构化的外部记忆系统就成了Agent走向实用和强大的必经之路。这个系统需要能可靠地存储信息更需要能智能地检索、更新和关联信息。今天要讨论的“双层记忆”架构以及结合PostgresSaver这类工具的实现思路就是针对这一痛点的一次工程实践。无论你是想开发一个能记住用户偏好的个人助手还是一个能在多次调试中学习最佳实践的编码Agent这套思路都能提供直接的参考。2. 记忆系统架构设计从理论到蓝图在动手写代码之前我们必须先把架构想清楚。一个健壮的Agent记忆系统不能是简单的聊天记录保存而应该是一个有清晰层次、职责分明的数据处理管道。2.1 三层记忆模型解析参考学术界和工业界的实践一个完整的Agent记忆模型可以抽象为三个层次我们称之为“三层记忆架构”感官记忆/即时记忆这是最原始的一层对应Agent对单次用户输入包括文本、图像、工具调用结果等的瞬时感知和解析。它存在时间极短主要用于初步理解用户的意图和提取关键信息片段。在工程上这一层通常被融合在输入处理管道中。工作记忆/短期记忆这是实现会话连续性的核心。它负责维护当前会话的上下文通常有一个容量限制例如最近10轮对话或最近N个Token。它的目标是保证Agent在本次对话中的回应是连贯、一致的。很多框架的“上下文管理”或“对话历史管理”模块本质上就是在做工作记忆。当对话超过限制时需要有一套策略如摘要、选择性遗忘来压缩或移出旧信息。长期记忆这是实现知识沉淀的关键。它用于存储跨越多个会话的、被认为有价值的信息。这些信息不是简单的对话日志而是经过提炼、结构化或向量化后的“知识”。例如用户的个人偏好、项目特定的规则、从成功经验中总结的模式、从失败中吸取的教训等。长期记忆的容量理论上可以非常大其挑战在于如何高效地存储和检索。我们项目标题中的“双层记忆”主要聚焦于工作记忆和长期记忆这两层因为它们是构建智能体“人格”和“能力”的支柱。感官记忆则更多是预处理环节。2.2 关键组件与数据流设计基于三层模型我们可以设计出系统的核心组件和数据流记忆编码器负责将原始的、非结构化的对话信息转化为适合存储和检索的格式。这可能包括文本分块与清洗将长文本分割成有意义的片段。元数据提取自动或手动为记忆片段打上标签如会话ID、时间戳、用户ID、主题、实体如涉及的人名、项目名、动作类型如“查询”、“指令”、“错误”等。向量化使用嵌入模型如text-embedding-3-small将文本转换为向量这是实现语义检索的基础。记忆存储库这是记忆的物理载体。一个混合存储方案通常是更优解向量数据库用于存储向量和关联的元数据支持高效的相似性搜索语义检索。这是从长期记忆中召回相关知识的核心。关系型数据库用于存储结构化的会话元数据、用户信息、以及那些不适合或不需要向量检索的记忆如精确的键值对配置。PostgreSQL因其可靠性、扩展性以及对JSON字段的良好支持成为许多项目的首选。这也是PostgresSaver这个名字的由来——它很可能是一个将记忆保存到PostgreSQL的工具或模块。缓存如Redis用于存储活跃会话的工作记忆提供极快的读写速度保证对话的实时性。记忆检索器在Agent需要生成回复时根据当前查询用户问题最近上下文从记忆存储库中召回最相关的信息。检索策略至关重要基于时间的检索优先召回最近发生的记忆。基于语义的检索使用当前查询的向量在向量库中搜索最相似的记忆片段。基于元数据的过滤例如只检索属于当前用户或当前项目的记忆。混合检索综合运用以上多种策略并对结果进行重排序Rerank得到最相关的记忆子集。记忆更新与遗忘机制记忆不是只进不出的。系统需要有能力更新当新信息与旧记忆冲突或补充时更新原有记忆例如用户更新了手机号。合并将多个相关的、细碎的记忆片段合并成一个更完整、更简洁的知识点。遗忘/归档定期清理过期、无用或低价值的记忆或将它们移至冷存储以控制存储成本和保持记忆库的“健康度”。注意这里的数据流是一个简化的理想模型。在实际开发中特别是初期你可能不需要实现所有组件。一个常见的起步方案是用PostgreSQL同时存储结构化元数据和向量利用其pgvector扩展用简单的最近邻检索ORDER BY embedding query_embedding来实现语义搜索。这能大大降低系统的复杂度。2.3 工具与框架选型考量看到热搜词里有LangGraph、Hermes Agent、PostgresSaver等这些都是当前生态中相关的工具或框架。选型时需要权衡LangGraph它是一个用于构建有状态、多智能体工作流的框架。其“状态”管理天然适合维护工作记忆。你可以将整个对话历史或记忆指针作为状态的一部分在图中传递。对于构建复杂的、涉及多个步骤和记忆访问的Agent流程LangGraph是一个强有力的候选。Hermes Agent或其他开源Agent框架这些框架通常已经内置了一定程度的记忆管理模块。使用它们可以快速起步但可能需要根据你的“双层记忆”需求进行定制或扩展。需要仔细阅读其文档看其记忆模块是否支持外部数据库存储、向量检索和长期记忆隔离。PostgresSaver从名字推断这很可能是一个专门用于将记忆可能是特定框架如LangChain或LlamaIndex的记忆对象持久化到PostgreSQL的类或工具。如果你的技术栈匹配直接使用它可以节省大量底层数据库操作的开发时间。向量数据库除了PostgreSQL的pgvector还有Chroma轻量、易用、Qdrant高性能、云原生、Weaviate功能丰富等选项。选择时考虑部署复杂度、性能、社区支持和与现有框架的集成度。实操心得对于个人项目或小团队我强烈建议从“PostgreSQL pgvector”开始。它避免了维护多个数据库的运维负担利用SQL能轻松处理复杂的元数据查询且pgvector的性能对于中小规模应用完全足够。这正符合PostgresSaver所倡导的“一体化存储”思路。3. 核心实现构建双层记忆系统理论说得再多不如一行代码。接下来我们以Python为例勾勒出实现双层记忆系统的核心代码骨架。我们将假设使用LangChain作为基础框架因其生态丰富并整合pgvector进行向量存储。3.1 环境准备与依赖安装首先确保你的环境已经就绪。# 安装核心库 pip install langchain langchain-community langchain-openai # 安装PostgreSQL驱动和向量扩展支持 pip install psycopg2-binary pgvector # 安装文本嵌入模型这里以OpenAI为例也可用sentence-transformers等开源模型 pip install openai # 可选安装LangChain对PostgreSQL向量存储的集成如果langchain-community中已有则无需单独安装 # pip install langchain-postgres在PostgreSQL中你需要先启用pgvector扩展并创建存储记忆的表。-- 在目标数据库中执行 CREATE EXTENSION IF NOT EXISTS vector; -- 创建一个存储记忆的表 CREATE TABLE agent_memories ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), session_id VARCHAR(255) NOT NULL, -- 会话标识用于工作记忆隔离和长期记忆关联 user_id VARCHAR(255), -- 用户标识用于长期记忆的用户隔离 memory_type VARCHAR(50) CHECK (memory_type IN (working, long_term)), -- 记忆类型 content TEXT NOT NULL, -- 记忆的原始文本内容 embedding vector(1536), -- 假设使用text-embedding-3-small维度为1536 metadata JSONB DEFAULT {}::jsonb, -- 存储标签、实体、时间戳等元数据 created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP ); -- 为常用查询创建索引 CREATE INDEX idx_memories_session ON agent_memories(session_id); CREATE INDEX idx_memories_user ON agent_memories(user_id); CREATE INDEX idx_memories_type ON agent_memories(memory_type); -- 为向量相似性搜索创建索引根据数据量选择索引类型如ivfflat或hnsw CREATE INDEX idx_memories_embedding ON agent_memories USING ivfflat (embedding vector_cosine_ops) WITH (lists 100);3.2 记忆管理器的实现我们将创建一个MemoryManager类它封装了工作记忆和长期记忆的读写逻辑。import json from datetime import datetime, timedelta from typing import List, Dict, Any, Optional, Tuple import psycopg2 from psycopg2.extras import Json from langchain.embeddings import OpenAIEmbeddings from langchain.schema import Document from langchain.vectorstores import PGVector from langchain.vectorstores.pgvector import DistanceStrategy class MemoryManager: def __init__(self, connection_string: str, embedding_model: OpenAIEmbeddings, working_memory_limit: int 10): 初始化记忆管理器。 :param connection_string: PostgreSQL连接字符串 :param embedding_model: 嵌入模型实例 :param working_memory_limit: 工作记忆在缓存中保留的最近对话轮数 self.conn_string connection_string self.embedder embedding_model self.working_memory_limit working_memory_limit # 初始化PGVector连接用于长期记忆的语义检索 # 注意PGVector会使用自己的表这里仅为演示。实际可统一使用上面自建的表。 # 为了简化我们下面直接使用psycopg2操作自建表来模拟。 self.conn psycopg2.connect(connection_string) self.cur self.conn.cursor() # 内存中的工作记忆缓存 {session_id: [messages]} self.working_memory_cache {} def _get_embedding(self, text: str) - List[float]: 生成文本的向量嵌入。 return self.embedder.embed_query(text) def add_memory(self, session_id: str, user_id: str, content: str, memory_type: str working, metadata: Optional[Dict] None): 添加一条记忆到存储。 if metadata is None: metadata {} # 添加系统元数据 metadata.update({ timestamp: datetime.utcnow().isoformat(), source: dialogue }) embedding self._get_embedding(content) sql INSERT INTO agent_memories (session_id, user_id, memory_type, content, embedding, metadata) VALUES (%s, %s, %s, %s, %s::vector, %s) self.cur.execute(sql, (session_id, user_id, memory_type, content, embedding, Json(metadata))) self.conn.commit() # 如果是工作记忆同时更新缓存 if memory_type working: if session_id not in self.working_memory_cache: self.working_memory_cache[session_id] [] self.working_memory_cache[session_id].append({content: content, metadata: metadata}) # 保持工作记忆缓存不超过限制 if len(self.working_memory_cache[session_id]) self.working_memory_limit: self.working_memory_cache[session_id].pop(0) def retrieve_working_memory(self, session_id: str, limit: int 5) - List[str]: 检索当前会话的工作记忆优先从缓存缓存没有则查库。 if session_id in self.working_memory_cache and self.working_memory_cache[session_id]: # 返回最近几条的内容 memories self.working_memory_cache[session_id][-limit:] return [m[content] for m in memories] else: # 从数据库查询 sql SELECT content FROM agent_memories WHERE session_id %s AND memory_type working ORDER BY created_at DESC LIMIT %s self.cur.execute(sql, (session_id, limit)) results self.cur.fetchall() return [r[0] for r in results] def retrieve_long_term_memory(self, query: str, user_id: str, top_k: int 3, filter_sessions: Optional[List[str]] None) - List[Tuple[str, float]]: 基于语义检索用户的长期记忆。 query_embedding self._get_embedding(query) # 构建查询条件 conditions [memory_type long_term, user_id %s] params [user_id] if filter_sessions: # 例如排除当前会话避免信息重复 placeholders , .join([%s] * len(filter_sessions)) conditions.append(fsession_id NOT IN ({placeholders})) params.extend(filter_sessions) where_clause AND .join(conditions) # 使用pgvector的余弦相似度操作符 sql f SELECT content, 1 - (embedding %s::vector) as similarity FROM agent_memories WHERE {where_clause} ORDER BY embedding %s::vector LIMIT %s params params [query_embedding, query_embedding, top_k] self.cur.execute(sql, params) results self.cur.fetchall() # 返回(内容相似度得分)的列表 return [(r[0], r[1]) for r in results] def promote_to_long_term(self, memory_ids: List[str]): 将指定的工作记忆提升为长期记忆。 # 这是一个关键操作意味着某条记忆被判定为有价值需要永久保存。 # 在实际中可能还需要对内容进行总结、提炼后再存储。 placeholders , .join([%s] * len(memory_ids)) sql f UPDATE agent_memories SET memory_type long_term, updated_at CURRENT_TIMESTAMP WHERE id IN ({placeholders}) self.cur.execute(sql, tuple(memory_ids)) self.conn.commit() def close(self): 关闭数据库连接。 self.cur.close() self.conn.close() # 初始化示例 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) memory_manager MemoryManager( connection_stringpostgresql://user:passwordlocalhost:5432/agent_db, embedding_modelembeddings, working_memory_limit15 )3.3 与Agent的集成策略有了记忆管理器下一步就是将其嵌入到Agent的推理循环中。一个典型的集成点在Agent生成回复之前我们需要为它组装上下文。from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain.tools import Tool class AgentWithMemory: def __init__(self, memory_manager: MemoryManager, llm: ChatOpenAI): self.memory memory_manager self.llm llm # 定义一个“搜索长期记忆”的工具供Agent在需要时主动调用 search_memory_tool Tool( namesearch_long_term_memory, funcself._search_memory_func, description当用户的问题涉及过去的知识、偏好或历史信息时使用此工具搜索长期记忆。输入应为清晰的搜索查询语句。 ) # 构建Agent这里是一个简单的OpenAI Tools Agent示例 prompt ChatPromptTemplate.from_messages([ (system, 你是一个有帮助的AI助手拥有与用户交互的记忆。 当前会话的工作记忆会自动提供给你。 如果你认为需要查询用户更早的历史信息请使用search_long_term_memory工具。 请基于所有可用信息给出准确、连贯的回答。), MessagesPlaceholder(variable_namechat_history), # 这里将填入工作记忆 (human, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), ]) self.agent create_openai_tools_agent(llm, [search_memory_tool], prompt) self.agent_executor AgentExecutor(agentself.agent, tools[search_memory_tool], verboseTrue) def _search_memory_func(self, query: str): 工具函数搜索长期记忆。在实际中这里需要获取当前的user_id和session_id。 # 假设我们能从上下文中获取当前用户和会话ID current_user_id user_123 current_session_id session_abc results self.memory.retrieve_long_term_memory( queryquery, user_idcurrent_user_id, filter_sessions[current_session_id] # 排除当前会话避免重复 ) if results: return \n.join([f[相关记忆 {i1}, 相关性{score:.2f}]: {content} for i, (content, score) in enumerate(results)]) else: return 在长期记忆中未找到相关信息。 def run(self, session_id: str, user_id: str, user_input: str) - str: 处理用户输入的核心方法。 # 1. 检索工作记忆作为对话历史 working_memories self.memory.retrieve_working_memory(session_id, limit10) # 将工作记忆转换为LangChain的Message格式这里简化为字符串连接 chat_history_str \n.join(working_memories) if working_memories else 无历史对话。 # 2. 调用Agent执行传入工作记忆作为历史 # 注意这里需要将chat_history_str适配成正确的Message格式以下为简化示例 inputs { input: user_input, chat_history: chat_history_str, # 实际应用中需转换为List[BaseMessage] } response self.agent_executor.invoke(inputs) # 3. 将本轮交互存入记忆用户输入和AI回复 self.memory.add_memory(session_id, user_id, fUser: {user_input}, memory_typeworking) self.memory.add_memory(session_id, user_id, fAssistant: {response[output]}, memory_typeworking) # 4. 可选判断是否将某些重要信息提升为长期记忆 # 这是一个高级功能可以通过规则或另一个LLM调用来判断。 # if self._should_promote_to_long_term(user_input, response[output]): # memory_id ... # 获取最近记忆的ID # self.memory.promote_to_long_term([memory_id]) return response[output] def _should_promote_to_long_term(self, user_input: str, assistant_output: str) - bool: 启发式规则或LLM调用判断对话是否包含值得长期保存的知识。 # 示例规则如果用户表达了明确的偏好或提供了关键事实 promotion_keywords [我喜欢, 我讨厌, 我的偏好是, 记住一下, 重要, 以后都用这个] for keyword in promotion_keywords: if keyword in user_input: return True # 更复杂的实现可以用一个轻量级LLM来判断 return False # 使用示例 llm ChatOpenAI(modelgpt-4-turbo-preview) agent_with_memory AgentWithMemory(memory_manager, llm) # 模拟一次对话 response agent_with_memory.run( session_idsession_abc, user_iduser_123, user_input我更喜欢用Python而不是Java来写后端服务。 ) print(fAssistant: {response}) # 在另一个会话或未来某次对话中 response2 agent_with_memory.run( session_idsession_def, # 新的会话 user_iduser_123, # 同一个用户 user_input我之前说过我喜欢用什么语言写后端 ) # 理想情况下Agent应能通过检索长期记忆回答“您之前表示过更喜欢用Python而不是Java来写后端服务。”关键点解析工作记忆的自动注入在每次调用run时自动从MemoryManager中取出当前会话的近期对话作为上下文chat_history提供给Agent。这保证了会话的连续性。长期记忆的按需检索我们将其封装成一个Tool。Agent在推理过程中如果认为需要查询更早的、跨会话的信息可以主动调用这个工具。这是一种“检索增强生成RAG”模式在记忆系统中的应用。记忆的写入时机在Agent回复后立即将用户输入和AI回复作为一对“工作记忆”保存。这保证了记忆的实时性。记忆晋升机制promote_to_long_term方法和_should_promote_to_long_term函数展示了如何将重要的临时对话转化为长期知识。这是实现“知识沉淀”的核心步骤可以基于规则、关键词或另一个LLM分类器来实现。4. 高级议题与实战避坑指南实现基础的双层记忆系统后我们会面临更多工程上的挑战。以下是几个关键的高级议题和从实战中总结的避坑经验。4.1 记忆的隔离、安全与隐私记忆“乱窜”是热搜词中提到的一个具体问题。这涉及到记忆的隔离。隔离维度会话隔离这是最基本的。session_id是确保不同对话之间工作记忆不混淆的关键。每次新的聊天窗口或连接都应生成唯一的session_id。用户隔离长期记忆必须严格按user_id进行隔离。用户A绝对不能看到用户B的记忆。在检索长期记忆时user_id是必须的过滤条件。租户/项目隔离在更复杂的系统如企业级应用中可能还需要tenant_id或project_id来实现更细粒度的隔离。安全与隐私敏感信息过滤在记忆存入数据库前应有管道对内容进行扫描过滤或脱敏手机号、邮箱、身份证号等个人敏感信息PII。记忆访问控制除了存储隔离在检索端也要有严格的权限校验。确保每次检索请求都带有合法的、经过认证的用户身份信息。遗忘权提供API让用户可以查看、编辑或删除自己的特定记忆这是合规性如GDPR的要求。实操心得在数据库设计时为所有记忆表加上user_id和session_id字段并以此建立复合索引。在每一次数据库查询中都显式地带上这些过滤条件形成“防御性编程”习惯从根本上避免数据泄露。4.2 记忆的压缩、摘要与更新记忆不能无限膨胀低质量、冗余的记忆会污染检索结果。工作记忆的压缩当对话轮数超过限制如working_memory_limit时简单的“先进先出”丢弃可能丢失重要早期信息。更好的策略是进行摘要。例如每5轮对话后用LLM对之前的对话内容生成一个简短的摘要然后用这个摘要替代那5轮原始对话作为一条新的“压缩记忆”放入工作记忆序列。这能在有限容量内保留更多信息。长期记忆的去重与合并定期运行后台任务对长期记忆进行聚类分析。将语义高度相似向量距离很近的记忆片段找出来用LLM将它们合并成一条更全面、更简洁的记忆并删除冗余的旧片段。这能显著提升记忆库的质量和检索效率。记忆的更新与修正如果用户说“我之前说的手机号错了应该是138xxxxxxx”系统需要能定位到之前那条错误的记忆通过语义检索或元数据标签并将其内容更新或标记为过期。这需要记忆有良好的版本管理或关联机制。4.3 性能优化与规模化当记忆量增长到数十万、百万条时性能问题会凸显。向量索引优化pgvector的ivfflat索引需要在使用一定数据量后REINDEX或者考虑使用hnsw索引以获得更好的查询性能和召回率。调整lists对于ivfflat或m/ef_construction对于hnsw参数以适应你的数据分布和查询负载。分级存储将很少访问的“冷记忆”转移到更便宜的存储如对象存储并在元数据中记录其位置。需要时再异步加载。这能降低主数据库的存储压力和成本。缓存策略工作记忆缓存如我们示例中所做在应用内存中使用字典缓存活跃会话的工作记忆避免对数据库的频繁读取。热点长期记忆缓存对于被频繁检索的长期记忆例如用户的常用地址可以将其内容或向量缓存在Redis中加速检索。异步写入记忆的写入尤其是向量生成和存储可以是异步操作不必阻塞Agent的响应流。可以将记忆写入任务放入消息队列如RabbitMQ、Kafka由后台Worker处理保证Agent的响应速度。4.4 评估记忆系统的有效性如何判断你的双层记忆系统是否真的提升了Agent的智能定性评估进行人工测试设计跨会话的问答对。例如Q1会话A“我的生日是7月10日。”Q2会话B“我的生日是什么时候” 检查Agent是否能正确回答。设计更复杂的场景如偏好记忆、事实记忆、任务步骤记忆等。定量评估检索相关性对于一批测试查询人工标注其应从长期记忆中召回的记忆片段计算系统实际召回结果的准确率Precision和召回率Recall。端到端任务成功率设计一系列依赖记忆的多步骤任务例如“根据我们上周讨论的架构图帮我生成一个部署清单”统计在有记忆系统和无记忆系统或仅有会话记忆下Agent独立完成任务的百分比。延迟与吞吐量监控记忆检索和存储操作的平均延迟、P99延迟以及系统能支撑的每秒记忆操作数OPS确保不影响用户体验。5. 典型问题排查与调试技巧在实际开发和运维中你肯定会遇到各种问题。下面是一个快速排查指南。问题现象可能原因排查步骤与解决方案Agent完全“忘记”了之前同一会话里的内容。1.session_id没有正确传递或生成。2. 工作记忆缓存失效或未命中且数据库查询失败。3. 记忆根本没有被成功写入数据库。1.检查日志确认每次请求的session_id是否稳定且唯一。检查add_memory和retrieve_working_memory的调用日志。2.检查数据库直接连接数据库查询对应session_id和memory_typeworking的记录是否存在。3.检查网络与连接确认应用与PostgreSQL的连接是否正常是否有连接池超时等问题。长期记忆检索不到任何内容或召回的内容不相关。1. 记忆晋升逻辑未触发长期记忆库为空。2. 向量索引未建立或性能差。3. 检索时user_id过滤条件错误或查询向量生成有问题。4. 嵌入模型不匹配存储和检索用的不是同一个模型。1.检查长期记忆表SELECT COUNT(*) FROM agent_memories WHERE memory_typelong_term。2.检查索引\d agent_memories查看索引是否建立。对于大量数据尝试REINDEX。3.调试检索查询打印出实际执行的SQL和参数手动在数据库客户端运行看结果。4.验证向量确保存储和检索时使用的是完全相同的嵌入模型和参数。计算一个已知文本的向量看存储和检索时是否一致。记忆出现“乱窜”用户A看到了用户B的信息。1. 数据库查询缺少user_id过滤条件或条件被错误覆盖。2. 缓存如Redis键设计有误导致不同用户的数据互相覆盖。1.代码审查仔细检查所有涉及记忆检索的SQL语句和缓存键生成逻辑确保user_id被强制使用。2.进行安全测试模拟两个用户同时操作检查日志和数据库查询记录看是否有越权查询。3.实施数据隔离测试作为上线前必检项。Agent响应速度变慢尤其是长对话后期。1. 工作记忆检索时未限制条数导致返回的历史上下文过长拖慢LLM处理速度。2. 长期记忆检索的top_k值设置过大或向量搜索未走索引。3. 记忆管理器的逻辑阻塞了主线程。1.优化工作记忆长度限制retrieve_working_memory的limit参数或引入摘要压缩机制。2.分析查询计划在数据库中对长期记忆检索的SQL执行EXPLAIN ANALYZE确认是否使用了向量索引。3.异步化将记忆的写入和复杂的检索操作改为异步非阻塞模式。存储空间增长过快。1. 所有对话都被无差别地永久存储了。2. 缺乏记忆压缩、合并和清理机制。1.实施TTL生存时间为工作记忆设置自动清理策略例如超过30天的会话记忆自动删除或归档。2.启动记忆维护任务定期运行脚本对长期记忆进行去重、合并和重要性评分删除低分记忆。调试技巧记录详细的记忆日志在关键节点记忆写入、晋升、检索打印结构化日志包含session_id,user_id,memory_id,操作类型和内容摘要。这比查看原始数据库记录直观得多。构建一个记忆查看器开发一个简单的内部管理界面可以按用户、会话、时间查询和浏览记忆内容。这在调试“记忆去哪了”的问题时无比有用。对记忆系统进行单元测试和集成测试模拟跨会话、多用户的复杂交互场景验证记忆的存储、检索、隔离是否正确。将测试用例纳入CI/CD流程。为Agent构建双层记忆系统是一个从“玩具”走向“工具”的关键步骤。它要求我们不仅关注AI模型本身更要重视围绕模型的数据工程和状态管理。从简单的会话历史保存到基于向量的语义检索再到复杂的记忆生命周期管理每一步都充满了工程上的权衡与挑战。但当你看到你的Agent能真正记住用户的喜好能在多次交互中积累经验并越用越聪明时这一切的努力都是值得的。