1. 项目概述为Agent构建记忆宫殿最近在折腾AI Agent开发的朋友估计都绕不开一个核心问题如何让Agent记住东西你肯定遇到过这样的场景你跟一个Agent聊了十分钟详细规划了一个项目结果你问它“我们刚才说的第一步是什么”它一脸茫然地告诉你“根据我的知识库……”完全忘了你们刚才的对话。这感觉就像跟一个金鱼聊天记忆只有七秒。这背后的症结就是Agent的记忆系统设计。我们今天要聊的就是如何给你的Agent打造一个“最强大脑”核心在于短期记忆与长期记忆的分层设计。这不仅仅是技术实现更是一种架构哲学。一个只有短期记忆的Agent就像在沙滩上写字潮水新的对话一来就抹平了而一个只有笨重长期记忆的Agent则像背着一座图书馆走路反应迟钝找不到重点。分层设计的精髓在于让Agent像人一样既有手边正在处理的“工作记忆”短期也有随时可以调取的“经验知识库”长期两者协同才能做出连贯、智能的决策。这个设计适用于所有需要持续交互、状态保持的Agent场景无论是智能客服、游戏NPC、个人助理还是自动化工作流。如果你正在用LangChain、LangGraph、AutoGen或是自研框架构建Agent理解并实现这套记忆分层将是你的Agent从“玩具”迈向“工具”的关键一步。接下来我会结合Python实战拆解这套系统的设计思路、核心模块与避坑指南。2. 记忆系统的核心架构与设计哲学2.1 为什么需要分层记忆在深入代码之前我们必须先想清楚“为什么”。Agent的记忆不是简单的聊天记录堆积。从信息处理的角度看记忆需要满足几个核心需求上下文关联性当前对话或任务的前因后果必须清晰。例如用户说“把刚才提到的那个文档总结一下”Agent必须知道“刚才提到的”具体指哪个文档。信息时效性与优先级刚发生的信息如用户最新指令通常比一小时前的闲聊更重要需要更快被检索到。存储与检索的成本平衡将每次交互的所有细节都存入一个向量数据库进行语义搜索即作为长期记忆在频繁交互下会产生巨大的计算和存储开销且可能引入大量噪声。状态保持与会话管理在长时间的、可能中断的交互中如一个持续数天的项目规划Agent需要保持任务状态并在重新连接时快速恢复。分层设计正是为了平衡这些矛盾。短期记忆Short-Term Memory, STM专注于高优先级、高时效性的当前会话信息通常存储在内存中读写极快结构简单。长期记忆Long-Term Memory, LTM则负责存储经过提炼的、重要的、可复用的知识通常使用外部数据库如向量数据库、图数据库、关系型数据库支持复杂的语义检索但访问成本较高。一个常见的类比是人的认知系统STM就像你的“工作记忆”正在心算一道数学题LTM就像你的“知识库”存储着乘法口诀和数学公式。当你解题时你会从LTM中调取口诀长期知识放入STM中进行组合计算短期处理。2.2 分层记忆的典型架构设计一个实用的分层记忆架构通常包含以下三层我将其称为“记忆金字塔”会话缓存层超短期记忆定位存储当前轮次或最近几轮对话的原始信息。生命周期极短主要用于保证单次请求内的连贯性。实现简单的内存数据结构如Python的deque双端队列或定长列表。内容原始的(role, content)对话记录。淘汰策略FIFO先进先出或基于固定长度。短期记忆/工作记忆层定位核心的“思考黑板”。存储当前任务相关的所有上下文包括用户目标、已执行步骤、中间结果、临时变量等。它是Agent进行推理和决策的直接依据。实现内存中的结构化对象。通常与Agent的“状态”State概念紧密绑定。在LangGraph中这对应着StateGraph的state在自研Agent中这可能是一个Pydantic模型或字典。内容结构化的键值对。例如{“user_goal”: “写一份周报”, “completed_steps”: [“收集数据”, “撰写初稿”], “current_step”: “润色修改”, “draft_content”: “...”}。特点随着任务推进不断更新任务完成后通常被清空或归档。长期记忆层定位Agent的“经验库”和“知识库”。存储需要长期保留、跨会话复用的信息。实现外部存储系统。根据信息类型进一步细分语义记忆存储非结构化的经验、事实、对话摘要。通常使用向量数据库如Chroma Pinecone Weaviate实现。信息被编码成向量Embedding通过相似度搜索进行检索。适合回答“以前遇到过类似问题吗”。程序记忆存储Agent学会的技能、工作流模板、API调用模式。可能用代码库、配置文件或关系型数据库存储。适合“上次是怎么解决这个问题的”。情景记忆存储重要的历史会话摘要或关键决策点。可以用时序数据库或带时间戳的文档数据库如MongoDB存储。适合“上周我和用户讨论过这个项目达成了什么共识”。读写策略并非所有信息都写入LTM。通常需要一个“提炼”过程例如当STM中的一个任务成功完成时将其关键信息和结果总结成一段文本再存入向量数据库。数据流向用户输入和工具调用结果首先进入会话缓存然后被提炼、结构化后更新短期记忆状态。Agent基于短期记忆进行思考决策。在任务里程碑或会话结束时系统将短期记忆中的关键信息进行摘要、去重然后存入长期记忆。当Agent遇到新任务时会先从长期记忆中检索相关经验作为上下文注入到短期记忆中开启新一轮循环。3. 核心模块实现与Python实战理论讲完了我们动手搭一个。下面我将用一个简化的Python示例展示如何构建一个具备分层记忆的Agent核心系统。我们将使用langchain用于大模型交互和基础组件和chromadb作为向量数据库实现长期语义记忆。3.1 环境准备与依赖安装首先确保你的Python环境建议3.9并安装必要库。我们将使用LangChain的社区版本和OpenAI的Embedding也可用开源的sentence-transformers。pip install langchain langchain-openai chromadb tiktoken接下来进行一些基础配置。你需要准备一个OpenAI的API Key或其它兼容API的Key。import os from typing import List, Dict, Any, Optional from datetime import datetime from collections import deque import json # 假设你的API Key已设置在环境变量中 os.environ[OPENAI_API_KEY] your-api-key-here from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain.schema import HumanMessage, AIMessage, SystemMessage from langchain.vectorstores import Chroma from langchain.schema import Document3.2 实现短期记忆状态管理短期记忆的本质是Agent的当前状态。我们设计一个AgentState类来管理它。class AgentState: Agent的短期记忆/状态容器。 存储当前任务相关的所有上下文信息。 def __init__(self, session_id: str, max_history_len: int 10): self.session_id session_id self.created_at datetime.now() # 会话缓存使用deque限制长度实现FIFO self.conversation_buffer deque(maxlenmax_history_len) # 工作记忆结构化的任务状态字典 self.working_memory { goal: None, # 当前用户目标 plan: [], # 执行计划 completed_actions: [], # 已完成动作 results: {}, # 动作结果缓存 context: {}, # 其他临时上下文 } # 当前正在思考的“念头” self.current_thought None def add_conversation_turn(self, role: str, content: str): 向会话缓存添加一轮对话。 self.conversation_buffer.append({role: role, content: content, timestamp: datetime.now()}) def update_working_memory(self, key: str, value: Any): 更新工作记忆中的特定字段。 if key in self.working_memory: if isinstance(self.working_memory[key], list) and isinstance(value, list): self.working_memory[key].extend(value) elif isinstance(self.working_memory[key], dict) and isinstance(value, dict): self.working_memory[key].update(value) else: self.working_memory[key] value else: self.working_memory[key] value def get_recent_conversation(self, n: int 5) - str: 获取最近N轮对话的文本用于构造LLM提示词。 recent list(self.conversation_buffer)[-n:] return \n.join([f{turn[role]}: {turn[content]} for turn in recent]) def summarize_for_long_term(self) - str: 将当前短期记忆提炼成一个文本摘要准备存入长期记忆。 这是一个关键函数决定了什么信息值得长期保留。 summary_parts [] if self.working_memory[goal]: summary_parts.append(f任务目标: {self.working_memory[goal]}) if self.working_memory[completed_actions]: summary_parts.append(f已完成步骤: {, .join(self.working_memory[completed_actions][-3:])}) # 只存最后几个关键步骤 if self.working_memory.get(results): # 只提炼关键结果避免存储过多细节 key_result str(list(self.working_memory[results].values())[-1])[:200] if self.working_memory[results] else 无 summary_parts.append(f关键结果摘要: {key_result}) summary_parts.append(f会话时间: {self.created_at.strftime(%Y-%m-%d %H:%M)}) return | .join(summary_parts)关键设计点conversation_buffer是原始的对话记录用于直接构造LLM的上下文。限制长度防止上下文爆炸。working_memory是结构化的状态是Agent“大脑”中的信息。这里的设计可以根据你的任务类型扩展比如加入constraints约束条件、preferences用户偏好等。summarize_for_long_term函数是短期记忆到长期记忆的“桥梁”。切忌把整个working_memory都存进去那会让长期记忆充满噪音。应该提取目标、关键决策、最终结果等精华信息。3.3 实现长期记忆向量数据库集成长期记忆我们使用ChromaDB实现一个简单的语义记忆模块。class LongTermMemory: 长期记忆管理基于向量数据库实现语义记忆。 def __init__(self, persist_directory: str ./chroma_db): self.embedding_function OpenAIEmbeddings(modeltext-embedding-3-small) # 使用小模型以节约成本 self.vector_store Chroma( collection_nameagent_experiences, embedding_functionself.embedding_function, persist_directorypersist_directory ) self.persist_directory persist_directory def store_experience(self, session_id: str, summary: str, metadata: Optional[Dict] None): 存储一条经验总结到长期记忆。 summary: 从短期记忆提炼的摘要文本。 metadata: 可附加的元数据如session_id, 任务类型时间戳等。 if metadata is None: metadata {} metadata.update({session_id: session_id, stored_at: datetime.now().isoformat()}) doc Document(page_contentsummary, metadatametadata) self.vector_store.add_documents([doc]) self.vector_store.persist() # 持久化到磁盘 print(f[LTM] 经验已存储: {summary[:50]}...) def retrieve_relevant_experiences(self, query: str, k: int 3) - List[str]: 根据当前查询如用户问题或任务描述从长期记忆中检索最相关的K条经验。 返回的是经验的文本内容。 docs self.vector_store.similarity_search(query, kk) experiences [doc.page_content for doc in docs] if experiences: print(f[LTM] 检索到 {len(experiences)} 条相关经验。) return experiences def clear_memory(self): 清空记忆谨慎使用。 # Chroma的删除操作稍复杂这里示意性重置。生产环境需要更安全的操作。 import shutil if os.path.exists(self.persist_directory): shutil.rmtree(self.persist_directory) self.vector_store Chroma( collection_nameagent_experiences, embedding_functionself.embedding_function, persist_directoryself.persist_directory )关键设计点存储的不是原始对话而是经过summarize_for_long_term处理后的摘要。这大大降低了存储量和检索噪声。元数据metadata至关重要。除了向量搜索你还可以通过元数据过滤例如只检索某个session_id或特定task_type的经验实现更精确的查找。retrieve_relevant_experiences返回的是文本这些文本会被注入到Agent的思考上下文中。检索的数量k需要平衡太少可能参考不足太多会挤占宝贵的上下文窗口并可能引入不相关信息。3.4 构建具备分层记忆的Agent核心现在我们将短期记忆状态和长期记忆组合起来形成一个有“记忆”的Agent核心逻辑。class AgentWithMemory: 具备分层记忆的Agent核心类。 def __init__(self, llm_model: str gpt-3.5-turbo): self.llm ChatOpenAI(modelllm_model, temperature0.7) self.long_term_memory LongTermMemory() # 活跃会话的状态字典 {session_id: AgentState} self.active_sessions: Dict[str, AgentState] {} def get_or_create_state(self, session_id: str) - AgentState: 获取或创建一个会话状态短期记忆。 if session_id not in self.active_sessions: print(f[Agent] 创建新会话状态: {session_id}) self.active_sessions[session_id] AgentState(session_id) return self.active_sessions[session_id] def process_query(self, session_id: str, user_input: str) - str: 处理用户输入的核心流程。 1. 获取/创建状态 2. 从长期记忆中检索相关经验 3. 更新短期记忆对话记录 4. 构造包含记忆的提示词调用LLM 5. 解析LLM响应更新状态可能触发长期记忆存储 state self.get_or_create_state(session_id) # 步骤1 2: 检索长期记忆获取相关经验 relevant_experiences self.long_term_memory.retrieve_relevant_experiences(user_input) experience_context if relevant_experiences: experience_context \n--- 相关历史经验 ---\n \n.join([f- {exp} for exp in relevant_experiences]) \n-------------------\n # 步骤3: 更新短期记忆中的对话缓存 state.add_conversation_turn(用户, user_input) # 步骤4: 构造增强的提示词 system_prompt f你是一个有帮助的AI助手并且拥有记忆能力。 以下是与你当前任务可能相关的历史经验仅供参考 {experience_context} 请结合以上经验和当前对话历史回应用户。 recent_chat state.get_recent_conversation(6) # 获取最近6轮对话 messages [ SystemMessage(contentsystem_prompt), *[HumanMessage(contentrecent_chat)], # 这里简化了消息构造实际需按role拆分 ] # 调用LLM print(f[Agent] 调用LLM进行思考...) response self.llm.invoke(messages) ai_response response.content # 更新短期记忆 state.add_conversation_turn(助手, ai_response) # 这里可以添加更复杂的逻辑解析AI响应更新working_memory中的plan, results等。 # 例如如果AI响应中包含“步骤已完成”则更新completed_actions。 if 目标设定为 in user_input: state.update_working_memory(goal, user_input) if 步骤 in ai_response and 完成 in ai_response: # 简单示例提取步骤名 state.update_working_memory(completed_actions, [某步骤]) # 步骤5: 判断是否需要将本次交互存入长期记忆 # 一个简单的启发式规则如果用户表达了感谢或任务明显结束则进行存储。 if any(word in user_input.lower() for word in [谢谢, 完成, 结束, 好了]): summary state.summarize_for_long_term() if summary: self.long_term_memory.store_experience(session_id, summary, {trigger: task_completion}) print(f[Agent] 任务结束经验已归档。) return ai_response def end_session(self, session_id: str): 结束一个会话可选地强制保存记忆。 if session_id in self.active_sessions: state self.active_sessions[session_id] # 会话结束时无论如何都尝试保存一个最终摘要 summary state.summarize_for_long_term() if summary: self.long_term_memory.store_experience(session_id, summary, {trigger: session_end}) del self.active_sessions[session_id] print(f[Agent] 会话 {session_id} 已结束并归档。)4. 实战演练与效果分析让我们用一个模拟的对话流程来测试这个Agent。# 初始化Agent agent AgentWithMemory(llm_modelgpt-3.5-turbo) # 模拟第一次会话学习如何泡茶 session_id user_123_tea print( 第一次会话学习泡茶 ) response1 agent.process_query(session_id, 我想学习如何泡一杯绿茶。) print(f助手: {response1}\n) response2 agent.process_query(session_id, 需要哪些步骤) print(f助手: {response2}\n) response3 agent.process_query(session_id, 水温多少度合适) print(f助手: {response3}\n) # 模拟用户结束本次学习 response4 agent.process_query(session_id, 明白了谢谢指导) print(f助手: {response4}\n) agent.end_session(session_id) # 显式结束会话触发最终归档 print(\n 第二次会话几天后用户再次询问 ) # 几天后同一用户相同session_id再次提问 response5 agent.process_query(session_id, 我忘了泡绿茶的水温了你能再告诉我吗) print(f助手: {response5}\n) # 观察此次助手的回复理论上它应该能从长期记忆中检索到上次关于“水温”的经验回答更精准。预期效果分析 在第一次会话中Agent会按部就班地回答。当用户说“谢谢”时process_query中的启发式规则被触发会将本次关于“泡绿茶步骤和水温”的会话摘要存入长期记忆向量数据库。 在第二次会话中用户提问“忘了泡绿茶的水温”。此时process_query函数会首先调用long_term_memory.retrieve_relevant_experiences(“我忘了泡绿茶的水温了”)。向量数据库会进行语义搜索很可能找到第一次会话存储的摘要其中包含“水温”关键词。这条经验会被作为上下文注入系统提示词从而让LLM能够回答“根据我们之前的经验泡绿茶的水温通常在80-85摄氏度...”而不是仅仅依靠其原始知识库给出一个通用答案。这就实现了跨会话的记忆Agent“记得”之前和这个用户讨论过什么。5. 高级技巧、常见问题与避坑指南实现基础分层记忆只是第一步要让它在生产环境中稳定可靠还需要考虑很多细节。5.1 记忆的提炼、存储与检索优化摘要Summarization的质量决定长期记忆的效用。直接存储原始对话是下策。更好的方法是在任务里程碑处总结不要每轮对话都存而是在一个子任务完成、用户明确确认或会话超时时进行总结。使用LLM进行摘要让LLM帮你提炼对话的核心目标、关键决策、最终结果和重要事实。这比我们上面写的简单规则summarize_for_long_term要强大得多。你可以设计一个提示词“请将以下对话总结成一段可供未来参考的经验突出用户的目标、解决的关键问题和最终方案。”结构化存储除了向量存储的文本摘要可以考虑将关键信息如goal,steps,result用JSON格式存入关系数据库便于精确查询。检索策略的混合使用语义检索向量搜索解决“类似问题”的查找。元数据过滤解决“谁在什么时候做了什么”的查找。例如session_idxxx或task_typedata_analysis。时间衰减给更近期的记忆更高的检索权重。可以在检索后对结果按时间戳进行重排序。重排序Reranking先用向量数据库召回一批比如10条相关记忆再用一个更精细的交叉编码器Cross-Encoder模型对它们进行精排选出最相关的3条。这能显著提升精度。5.2 状态管理与会话隔离会话Session的生命周期管理session_id是关键。它可以是用户ID、聊天窗口ID、线程ID。必须设计清晰的会话创建、销毁和过期策略。对于Web应用通常一个浏览器标签页对应一个会话。长时间不活动的会话其短期记忆AgentState应该被清除以释放内存但长期记忆已持久化不受影响。状态State的序列化与持久化上面的例子中AgentState存在于内存。如果服务重启所有活跃会话的短期记忆会丢失。对于关键应用需要将AgentState定期或实时序列化如用Pickle或JSON保存到Redis或数据库中并在恢复时反序列化。上下文长度限制这是LLM应用的硬约束。你的conversation_buffer原始对话和检索到的relevant_experiences长期记忆加起来不能超过LLM的上下文窗口。必须设计截断策略例如优先保留最近对话和相关性最高的记忆。5.3 典型问题排查与解决方案问题Agent的回答变得前后矛盾或混乱。排查检查conversation_buffer是否过长导致早期重要指令被挤出上下文。检查检索到的relevant_experiences是否过多或包含不相关/冲突的信息。解决缩短conversation_buffer的maxlen或实现一个更智能的摘要机制将长对话历史总结成一段固定长度的背景描述。减少检索数量k并加强元数据过滤。问题长期记忆检索不到相关内容或者检索到的内容没用。排查首先检查记忆是否成功存储。查看向量数据库的存储记录。其次检查摘要文本的质量。过于简略或模糊的摘要如“用户问了问题助手回答了”会导致向量搜索失效。解决优化摘要生成提示词确保摘要包含具体的关键实体如“绿茶”、“水温80°C”、“三步法”。尝试不同的Embedding模型如text-embedding-3-large效果更好但更贵。调整检索时的相似度阈值。问题Agent性能下降响应变慢。排查瓶颈可能出现在a) Embedding生成存储和检索时b) 向量数据库相似度搜索c) LLM上下文因记忆注入而变长。解决a) 对存储的摘要进行批处理Embedding而非实时处理。b) 为向量数据库建立索引并确保硬件资源充足。c) 严格限制注入上下文的记忆条数和总长度。问题记忆“污染”——错误或无效信息被存入长期记忆。排查存储触发条件是否太宽松比如用户说“不对你错了”这句话也可能触发“谢谢”规则而被错误存储。解决设计更可靠的存储触发逻辑。例如结合情感分析用户表达高度满意、任务状态机标记任务为“成功完成”或让LLM判断“本次交互是否值得作为经验保存”。同时可以考虑为长期记忆设计一个“遗忘”或“降权”机制或者允许人工审核和清理。5.4 安全与成本考量隐私与数据安全长期记忆存储了用户交互的摘要。必须明确告知用户并遵循数据隐私法规。考虑对存储的摘要进行去标识化处理或提供用户清除自己记忆的接口。成本控制每一次存储生成Embedding和检索向量搜索都有成本OpenAI API调用或计算开销。避免高频的、无差别的存储操作。可以设置一个“经验价值”阈值只有达到一定信息密度或确认度的对话才进行存储。给Agent打造“最强大脑”是一个持续迭代的过程。分层记忆设计提供了一个清晰、可扩展的框架。从简单的会话缓存和状态字典开始逐步引入向量数据库实现长期语义记忆再根据业务需求混合元数据过滤、重排序等高级策略。记住记忆系统的目标不是记住一切而是记住对的东西并在对的时间想起来。这其中的“对”就需要你在开发过程中结合具体的Agent任务场景不断地调试和优化你的提炼策略与检索逻辑。
1. 项目概述为Agent构建记忆宫殿最近在折腾AI Agent开发的朋友估计都绕不开一个核心问题如何让Agent记住东西你肯定遇到过这样的场景你跟一个Agent聊了十分钟详细规划了一个项目结果你问它“我们刚才说的第一步是什么”它一脸茫然地告诉你“根据我的知识库……”完全忘了你们刚才的对话。这感觉就像跟一个金鱼聊天记忆只有七秒。这背后的症结就是Agent的记忆系统设计。我们今天要聊的就是如何给你的Agent打造一个“最强大脑”核心在于短期记忆与长期记忆的分层设计。这不仅仅是技术实现更是一种架构哲学。一个只有短期记忆的Agent就像在沙滩上写字潮水新的对话一来就抹平了而一个只有笨重长期记忆的Agent则像背着一座图书馆走路反应迟钝找不到重点。分层设计的精髓在于让Agent像人一样既有手边正在处理的“工作记忆”短期也有随时可以调取的“经验知识库”长期两者协同才能做出连贯、智能的决策。这个设计适用于所有需要持续交互、状态保持的Agent场景无论是智能客服、游戏NPC、个人助理还是自动化工作流。如果你正在用LangChain、LangGraph、AutoGen或是自研框架构建Agent理解并实现这套记忆分层将是你的Agent从“玩具”迈向“工具”的关键一步。接下来我会结合Python实战拆解这套系统的设计思路、核心模块与避坑指南。2. 记忆系统的核心架构与设计哲学2.1 为什么需要分层记忆在深入代码之前我们必须先想清楚“为什么”。Agent的记忆不是简单的聊天记录堆积。从信息处理的角度看记忆需要满足几个核心需求上下文关联性当前对话或任务的前因后果必须清晰。例如用户说“把刚才提到的那个文档总结一下”Agent必须知道“刚才提到的”具体指哪个文档。信息时效性与优先级刚发生的信息如用户最新指令通常比一小时前的闲聊更重要需要更快被检索到。存储与检索的成本平衡将每次交互的所有细节都存入一个向量数据库进行语义搜索即作为长期记忆在频繁交互下会产生巨大的计算和存储开销且可能引入大量噪声。状态保持与会话管理在长时间的、可能中断的交互中如一个持续数天的项目规划Agent需要保持任务状态并在重新连接时快速恢复。分层设计正是为了平衡这些矛盾。短期记忆Short-Term Memory, STM专注于高优先级、高时效性的当前会话信息通常存储在内存中读写极快结构简单。长期记忆Long-Term Memory, LTM则负责存储经过提炼的、重要的、可复用的知识通常使用外部数据库如向量数据库、图数据库、关系型数据库支持复杂的语义检索但访问成本较高。一个常见的类比是人的认知系统STM就像你的“工作记忆”正在心算一道数学题LTM就像你的“知识库”存储着乘法口诀和数学公式。当你解题时你会从LTM中调取口诀长期知识放入STM中进行组合计算短期处理。2.2 分层记忆的典型架构设计一个实用的分层记忆架构通常包含以下三层我将其称为“记忆金字塔”会话缓存层超短期记忆定位存储当前轮次或最近几轮对话的原始信息。生命周期极短主要用于保证单次请求内的连贯性。实现简单的内存数据结构如Python的deque双端队列或定长列表。内容原始的(role, content)对话记录。淘汰策略FIFO先进先出或基于固定长度。短期记忆/工作记忆层定位核心的“思考黑板”。存储当前任务相关的所有上下文包括用户目标、已执行步骤、中间结果、临时变量等。它是Agent进行推理和决策的直接依据。实现内存中的结构化对象。通常与Agent的“状态”State概念紧密绑定。在LangGraph中这对应着StateGraph的state在自研Agent中这可能是一个Pydantic模型或字典。内容结构化的键值对。例如{“user_goal”: “写一份周报”, “completed_steps”: [“收集数据”, “撰写初稿”], “current_step”: “润色修改”, “draft_content”: “...”}。特点随着任务推进不断更新任务完成后通常被清空或归档。长期记忆层定位Agent的“经验库”和“知识库”。存储需要长期保留、跨会话复用的信息。实现外部存储系统。根据信息类型进一步细分语义记忆存储非结构化的经验、事实、对话摘要。通常使用向量数据库如Chroma Pinecone Weaviate实现。信息被编码成向量Embedding通过相似度搜索进行检索。适合回答“以前遇到过类似问题吗”。程序记忆存储Agent学会的技能、工作流模板、API调用模式。可能用代码库、配置文件或关系型数据库存储。适合“上次是怎么解决这个问题的”。情景记忆存储重要的历史会话摘要或关键决策点。可以用时序数据库或带时间戳的文档数据库如MongoDB存储。适合“上周我和用户讨论过这个项目达成了什么共识”。读写策略并非所有信息都写入LTM。通常需要一个“提炼”过程例如当STM中的一个任务成功完成时将其关键信息和结果总结成一段文本再存入向量数据库。数据流向用户输入和工具调用结果首先进入会话缓存然后被提炼、结构化后更新短期记忆状态。Agent基于短期记忆进行思考决策。在任务里程碑或会话结束时系统将短期记忆中的关键信息进行摘要、去重然后存入长期记忆。当Agent遇到新任务时会先从长期记忆中检索相关经验作为上下文注入到短期记忆中开启新一轮循环。3. 核心模块实现与Python实战理论讲完了我们动手搭一个。下面我将用一个简化的Python示例展示如何构建一个具备分层记忆的Agent核心系统。我们将使用langchain用于大模型交互和基础组件和chromadb作为向量数据库实现长期语义记忆。3.1 环境准备与依赖安装首先确保你的Python环境建议3.9并安装必要库。我们将使用LangChain的社区版本和OpenAI的Embedding也可用开源的sentence-transformers。pip install langchain langchain-openai chromadb tiktoken接下来进行一些基础配置。你需要准备一个OpenAI的API Key或其它兼容API的Key。import os from typing import List, Dict, Any, Optional from datetime import datetime from collections import deque import json # 假设你的API Key已设置在环境变量中 os.environ[OPENAI_API_KEY] your-api-key-here from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain.schema import HumanMessage, AIMessage, SystemMessage from langchain.vectorstores import Chroma from langchain.schema import Document3.2 实现短期记忆状态管理短期记忆的本质是Agent的当前状态。我们设计一个AgentState类来管理它。class AgentState: Agent的短期记忆/状态容器。 存储当前任务相关的所有上下文信息。 def __init__(self, session_id: str, max_history_len: int 10): self.session_id session_id self.created_at datetime.now() # 会话缓存使用deque限制长度实现FIFO self.conversation_buffer deque(maxlenmax_history_len) # 工作记忆结构化的任务状态字典 self.working_memory { goal: None, # 当前用户目标 plan: [], # 执行计划 completed_actions: [], # 已完成动作 results: {}, # 动作结果缓存 context: {}, # 其他临时上下文 } # 当前正在思考的“念头” self.current_thought None def add_conversation_turn(self, role: str, content: str): 向会话缓存添加一轮对话。 self.conversation_buffer.append({role: role, content: content, timestamp: datetime.now()}) def update_working_memory(self, key: str, value: Any): 更新工作记忆中的特定字段。 if key in self.working_memory: if isinstance(self.working_memory[key], list) and isinstance(value, list): self.working_memory[key].extend(value) elif isinstance(self.working_memory[key], dict) and isinstance(value, dict): self.working_memory[key].update(value) else: self.working_memory[key] value else: self.working_memory[key] value def get_recent_conversation(self, n: int 5) - str: 获取最近N轮对话的文本用于构造LLM提示词。 recent list(self.conversation_buffer)[-n:] return \n.join([f{turn[role]}: {turn[content]} for turn in recent]) def summarize_for_long_term(self) - str: 将当前短期记忆提炼成一个文本摘要准备存入长期记忆。 这是一个关键函数决定了什么信息值得长期保留。 summary_parts [] if self.working_memory[goal]: summary_parts.append(f任务目标: {self.working_memory[goal]}) if self.working_memory[completed_actions]: summary_parts.append(f已完成步骤: {, .join(self.working_memory[completed_actions][-3:])}) # 只存最后几个关键步骤 if self.working_memory.get(results): # 只提炼关键结果避免存储过多细节 key_result str(list(self.working_memory[results].values())[-1])[:200] if self.working_memory[results] else 无 summary_parts.append(f关键结果摘要: {key_result}) summary_parts.append(f会话时间: {self.created_at.strftime(%Y-%m-%d %H:%M)}) return | .join(summary_parts)关键设计点conversation_buffer是原始的对话记录用于直接构造LLM的上下文。限制长度防止上下文爆炸。working_memory是结构化的状态是Agent“大脑”中的信息。这里的设计可以根据你的任务类型扩展比如加入constraints约束条件、preferences用户偏好等。summarize_for_long_term函数是短期记忆到长期记忆的“桥梁”。切忌把整个working_memory都存进去那会让长期记忆充满噪音。应该提取目标、关键决策、最终结果等精华信息。3.3 实现长期记忆向量数据库集成长期记忆我们使用ChromaDB实现一个简单的语义记忆模块。class LongTermMemory: 长期记忆管理基于向量数据库实现语义记忆。 def __init__(self, persist_directory: str ./chroma_db): self.embedding_function OpenAIEmbeddings(modeltext-embedding-3-small) # 使用小模型以节约成本 self.vector_store Chroma( collection_nameagent_experiences, embedding_functionself.embedding_function, persist_directorypersist_directory ) self.persist_directory persist_directory def store_experience(self, session_id: str, summary: str, metadata: Optional[Dict] None): 存储一条经验总结到长期记忆。 summary: 从短期记忆提炼的摘要文本。 metadata: 可附加的元数据如session_id, 任务类型时间戳等。 if metadata is None: metadata {} metadata.update({session_id: session_id, stored_at: datetime.now().isoformat()}) doc Document(page_contentsummary, metadatametadata) self.vector_store.add_documents([doc]) self.vector_store.persist() # 持久化到磁盘 print(f[LTM] 经验已存储: {summary[:50]}...) def retrieve_relevant_experiences(self, query: str, k: int 3) - List[str]: 根据当前查询如用户问题或任务描述从长期记忆中检索最相关的K条经验。 返回的是经验的文本内容。 docs self.vector_store.similarity_search(query, kk) experiences [doc.page_content for doc in docs] if experiences: print(f[LTM] 检索到 {len(experiences)} 条相关经验。) return experiences def clear_memory(self): 清空记忆谨慎使用。 # Chroma的删除操作稍复杂这里示意性重置。生产环境需要更安全的操作。 import shutil if os.path.exists(self.persist_directory): shutil.rmtree(self.persist_directory) self.vector_store Chroma( collection_nameagent_experiences, embedding_functionself.embedding_function, persist_directoryself.persist_directory )关键设计点存储的不是原始对话而是经过summarize_for_long_term处理后的摘要。这大大降低了存储量和检索噪声。元数据metadata至关重要。除了向量搜索你还可以通过元数据过滤例如只检索某个session_id或特定task_type的经验实现更精确的查找。retrieve_relevant_experiences返回的是文本这些文本会被注入到Agent的思考上下文中。检索的数量k需要平衡太少可能参考不足太多会挤占宝贵的上下文窗口并可能引入不相关信息。3.4 构建具备分层记忆的Agent核心现在我们将短期记忆状态和长期记忆组合起来形成一个有“记忆”的Agent核心逻辑。class AgentWithMemory: 具备分层记忆的Agent核心类。 def __init__(self, llm_model: str gpt-3.5-turbo): self.llm ChatOpenAI(modelllm_model, temperature0.7) self.long_term_memory LongTermMemory() # 活跃会话的状态字典 {session_id: AgentState} self.active_sessions: Dict[str, AgentState] {} def get_or_create_state(self, session_id: str) - AgentState: 获取或创建一个会话状态短期记忆。 if session_id not in self.active_sessions: print(f[Agent] 创建新会话状态: {session_id}) self.active_sessions[session_id] AgentState(session_id) return self.active_sessions[session_id] def process_query(self, session_id: str, user_input: str) - str: 处理用户输入的核心流程。 1. 获取/创建状态 2. 从长期记忆中检索相关经验 3. 更新短期记忆对话记录 4. 构造包含记忆的提示词调用LLM 5. 解析LLM响应更新状态可能触发长期记忆存储 state self.get_or_create_state(session_id) # 步骤1 2: 检索长期记忆获取相关经验 relevant_experiences self.long_term_memory.retrieve_relevant_experiences(user_input) experience_context if relevant_experiences: experience_context \n--- 相关历史经验 ---\n \n.join([f- {exp} for exp in relevant_experiences]) \n-------------------\n # 步骤3: 更新短期记忆中的对话缓存 state.add_conversation_turn(用户, user_input) # 步骤4: 构造增强的提示词 system_prompt f你是一个有帮助的AI助手并且拥有记忆能力。 以下是与你当前任务可能相关的历史经验仅供参考 {experience_context} 请结合以上经验和当前对话历史回应用户。 recent_chat state.get_recent_conversation(6) # 获取最近6轮对话 messages [ SystemMessage(contentsystem_prompt), *[HumanMessage(contentrecent_chat)], # 这里简化了消息构造实际需按role拆分 ] # 调用LLM print(f[Agent] 调用LLM进行思考...) response self.llm.invoke(messages) ai_response response.content # 更新短期记忆 state.add_conversation_turn(助手, ai_response) # 这里可以添加更复杂的逻辑解析AI响应更新working_memory中的plan, results等。 # 例如如果AI响应中包含“步骤已完成”则更新completed_actions。 if 目标设定为 in user_input: state.update_working_memory(goal, user_input) if 步骤 in ai_response and 完成 in ai_response: # 简单示例提取步骤名 state.update_working_memory(completed_actions, [某步骤]) # 步骤5: 判断是否需要将本次交互存入长期记忆 # 一个简单的启发式规则如果用户表达了感谢或任务明显结束则进行存储。 if any(word in user_input.lower() for word in [谢谢, 完成, 结束, 好了]): summary state.summarize_for_long_term() if summary: self.long_term_memory.store_experience(session_id, summary, {trigger: task_completion}) print(f[Agent] 任务结束经验已归档。) return ai_response def end_session(self, session_id: str): 结束一个会话可选地强制保存记忆。 if session_id in self.active_sessions: state self.active_sessions[session_id] # 会话结束时无论如何都尝试保存一个最终摘要 summary state.summarize_for_long_term() if summary: self.long_term_memory.store_experience(session_id, summary, {trigger: session_end}) del self.active_sessions[session_id] print(f[Agent] 会话 {session_id} 已结束并归档。)4. 实战演练与效果分析让我们用一个模拟的对话流程来测试这个Agent。# 初始化Agent agent AgentWithMemory(llm_modelgpt-3.5-turbo) # 模拟第一次会话学习如何泡茶 session_id user_123_tea print( 第一次会话学习泡茶 ) response1 agent.process_query(session_id, 我想学习如何泡一杯绿茶。) print(f助手: {response1}\n) response2 agent.process_query(session_id, 需要哪些步骤) print(f助手: {response2}\n) response3 agent.process_query(session_id, 水温多少度合适) print(f助手: {response3}\n) # 模拟用户结束本次学习 response4 agent.process_query(session_id, 明白了谢谢指导) print(f助手: {response4}\n) agent.end_session(session_id) # 显式结束会话触发最终归档 print(\n 第二次会话几天后用户再次询问 ) # 几天后同一用户相同session_id再次提问 response5 agent.process_query(session_id, 我忘了泡绿茶的水温了你能再告诉我吗) print(f助手: {response5}\n) # 观察此次助手的回复理论上它应该能从长期记忆中检索到上次关于“水温”的经验回答更精准。预期效果分析 在第一次会话中Agent会按部就班地回答。当用户说“谢谢”时process_query中的启发式规则被触发会将本次关于“泡绿茶步骤和水温”的会话摘要存入长期记忆向量数据库。 在第二次会话中用户提问“忘了泡绿茶的水温”。此时process_query函数会首先调用long_term_memory.retrieve_relevant_experiences(“我忘了泡绿茶的水温了”)。向量数据库会进行语义搜索很可能找到第一次会话存储的摘要其中包含“水温”关键词。这条经验会被作为上下文注入系统提示词从而让LLM能够回答“根据我们之前的经验泡绿茶的水温通常在80-85摄氏度...”而不是仅仅依靠其原始知识库给出一个通用答案。这就实现了跨会话的记忆Agent“记得”之前和这个用户讨论过什么。5. 高级技巧、常见问题与避坑指南实现基础分层记忆只是第一步要让它在生产环境中稳定可靠还需要考虑很多细节。5.1 记忆的提炼、存储与检索优化摘要Summarization的质量决定长期记忆的效用。直接存储原始对话是下策。更好的方法是在任务里程碑处总结不要每轮对话都存而是在一个子任务完成、用户明确确认或会话超时时进行总结。使用LLM进行摘要让LLM帮你提炼对话的核心目标、关键决策、最终结果和重要事实。这比我们上面写的简单规则summarize_for_long_term要强大得多。你可以设计一个提示词“请将以下对话总结成一段可供未来参考的经验突出用户的目标、解决的关键问题和最终方案。”结构化存储除了向量存储的文本摘要可以考虑将关键信息如goal,steps,result用JSON格式存入关系数据库便于精确查询。检索策略的混合使用语义检索向量搜索解决“类似问题”的查找。元数据过滤解决“谁在什么时候做了什么”的查找。例如session_idxxx或task_typedata_analysis。时间衰减给更近期的记忆更高的检索权重。可以在检索后对结果按时间戳进行重排序。重排序Reranking先用向量数据库召回一批比如10条相关记忆再用一个更精细的交叉编码器Cross-Encoder模型对它们进行精排选出最相关的3条。这能显著提升精度。5.2 状态管理与会话隔离会话Session的生命周期管理session_id是关键。它可以是用户ID、聊天窗口ID、线程ID。必须设计清晰的会话创建、销毁和过期策略。对于Web应用通常一个浏览器标签页对应一个会话。长时间不活动的会话其短期记忆AgentState应该被清除以释放内存但长期记忆已持久化不受影响。状态State的序列化与持久化上面的例子中AgentState存在于内存。如果服务重启所有活跃会话的短期记忆会丢失。对于关键应用需要将AgentState定期或实时序列化如用Pickle或JSON保存到Redis或数据库中并在恢复时反序列化。上下文长度限制这是LLM应用的硬约束。你的conversation_buffer原始对话和检索到的relevant_experiences长期记忆加起来不能超过LLM的上下文窗口。必须设计截断策略例如优先保留最近对话和相关性最高的记忆。5.3 典型问题排查与解决方案问题Agent的回答变得前后矛盾或混乱。排查检查conversation_buffer是否过长导致早期重要指令被挤出上下文。检查检索到的relevant_experiences是否过多或包含不相关/冲突的信息。解决缩短conversation_buffer的maxlen或实现一个更智能的摘要机制将长对话历史总结成一段固定长度的背景描述。减少检索数量k并加强元数据过滤。问题长期记忆检索不到相关内容或者检索到的内容没用。排查首先检查记忆是否成功存储。查看向量数据库的存储记录。其次检查摘要文本的质量。过于简略或模糊的摘要如“用户问了问题助手回答了”会导致向量搜索失效。解决优化摘要生成提示词确保摘要包含具体的关键实体如“绿茶”、“水温80°C”、“三步法”。尝试不同的Embedding模型如text-embedding-3-large效果更好但更贵。调整检索时的相似度阈值。问题Agent性能下降响应变慢。排查瓶颈可能出现在a) Embedding生成存储和检索时b) 向量数据库相似度搜索c) LLM上下文因记忆注入而变长。解决a) 对存储的摘要进行批处理Embedding而非实时处理。b) 为向量数据库建立索引并确保硬件资源充足。c) 严格限制注入上下文的记忆条数和总长度。问题记忆“污染”——错误或无效信息被存入长期记忆。排查存储触发条件是否太宽松比如用户说“不对你错了”这句话也可能触发“谢谢”规则而被错误存储。解决设计更可靠的存储触发逻辑。例如结合情感分析用户表达高度满意、任务状态机标记任务为“成功完成”或让LLM判断“本次交互是否值得作为经验保存”。同时可以考虑为长期记忆设计一个“遗忘”或“降权”机制或者允许人工审核和清理。5.4 安全与成本考量隐私与数据安全长期记忆存储了用户交互的摘要。必须明确告知用户并遵循数据隐私法规。考虑对存储的摘要进行去标识化处理或提供用户清除自己记忆的接口。成本控制每一次存储生成Embedding和检索向量搜索都有成本OpenAI API调用或计算开销。避免高频的、无差别的存储操作。可以设置一个“经验价值”阈值只有达到一定信息密度或确认度的对话才进行存储。给Agent打造“最强大脑”是一个持续迭代的过程。分层记忆设计提供了一个清晰、可扩展的框架。从简单的会话缓存和状态字典开始逐步引入向量数据库实现长期语义记忆再根据业务需求混合元数据过滤、重排序等高级策略。记住记忆系统的目标不是记住一切而是记住对的东西并在对的时间想起来。这其中的“对”就需要你在开发过程中结合具体的Agent任务场景不断地调试和优化你的提炼策略与检索逻辑。
