1. 项目概述为什么记忆是Agent的命门最近在社区里跟几个做Agent的朋友聊天发现一个挺有意思的现象很多团队初期都把精力花在“思考”和“行动”模块上比如怎么让大模型更好地调用工具Function Calling怎么设计更复杂的任务拆解逻辑。但项目一跑起来尤其是涉及到多轮、长周期的任务时最先出问题、也最让人头疼的往往不是“脑子”不够用而是“记性”太差。一个典型的翻车现场是你让Agent帮你订一张下周从北京飞上海的机票它前几步都执行得很好查航班、比价格但就在最后一步填写乘客信息时它突然问你“您要订哪天的票来着”——得前面白忙活了。这就是典型的“记忆丢失”Agent忘了本轮对话甚至前几轮对话的关键上下文。所以当我们谈《从零实现Agent系统》的“记忆系统”时我们谈的绝不是一个锦上添花的附属功能而是Agent能否真正“持续工作”、具备“个性化”和“上下文连贯性”的基石。没有记忆的Agent就像金鱼只有7秒的“短期上下文”每次交互都是全新的开始无法积累经验无法执行复杂任务。而一个健壮的记忆系统需要清晰地处理两种核心记忆短期上下文Short-term Context和长期外部记忆Long-term External Memory。前者决定了Agent单次交互的“工作内存”容量和效率后者则关乎Agent的“人生经验”库能否持续增长并有效检索。搞不清这两者的边界和协作方式你的Agent很容易陷入“当场死机”或者“历史健忘”的窘境。接下来我们就深入这个核心模块看看如何从零搭建一个既可靠又高效的内存系统。2. 记忆系统核心设计短期与长期的职责边界在动手写代码之前我们必须从设计层面厘清短期记忆和长期记忆各自该管什么、怎么管以及它们之间如何握手。这是一个架构问题搞错了后期会非常痛苦。2.1 短期上下文本轮对话的“工作台”你可以把短期上下文想象成程序员电脑上开着的IDE界面、浏览器标签页和终端窗口。所有你当前正在专注处理的信息都放在这里访问速度极快内存级但容量有限受限于大模型上下文窗口比如128K而且一旦关闭对话窗口会话结束这些“工作状态”就清零了。它的核心职责包括承载本轮对话的完整历史用户最新的Query、Agent的思考过程Chain-of-Thought、工具调用Function Call的请求和执行结果以及系统指令System Prompt。这里必须划重点Function Call的“执行结果”必须立刻、马上、无条件地塞进短期上下文这是血的教训。我见过不止一个项目工具执行的结果比如查询到的航班信息、计算出的数据被直接返回给调用方却没有在第一时间追加到发给大模型的上下文里。导致下一轮Agent的推理基于的是不完整的信息轻则答非所问重则逻辑崩盘“本轮对话会当场死机”。所以短期上下文管理器的第一个铁律就是任何行动的输出都是后续思考的输入必须无缝衔接。维持超短期的状态追踪比如在一个多步任务中当前执行到第几步上一步的输出是什么下一步的输入依赖是什么。这通常需要一些轻量的状态管理。提供最快速的上下文检索当大模型生成需要参考之前的某句话时短期上下文应该能立刻提供。这通常就是简单的数组或列表按时间顺序排列。设计要点与避坑指南容量管理是头等大事大模型的上下文窗口是宝贵且有限的资源。你不能让对话历史无限膨胀。必须实现一个摘要Summarization或滑动窗口Sliding Window策略。例如当token数接近阈值如120K时自动将最早、最不重要的几轮对话压缩成一个摘要腾出空间。很多开源框架如LangChain的ConversationSummaryBufferMemory就是干这个的。结构化管理优于纯文本不要把整个对话历史当成一个字符串扔进去。最好将其结构化为消息列表List[Dict]每条消息明确角色user,assistant,system,function。这样不仅清晰也便于后续的检索和过滤。例如你可能只想在特定步骤中让模型参考“所有工具执行的结果”结构化数据让这变得很容易。系统指令System Prompt要“钉”住确保你的系统指令定义了Agent的角色、核心能力、约束条件始终存在于短期上下文的开头并且不会被摘要或滑动窗口机制挤掉。这是Agent的“人设”根基不能丢。2.2 长期外部记忆Agent的“私人知识库”如果说短期记忆是工作台那长期记忆就是你的书架、硬盘、云笔记。它存储那些需要跨会话持久化、在未来可能被反复使用的信息。容量理论上可以无限取决于你的存储介质但检索需要时间I/O操作并且存在“遗忘”检索不到的风险。它的核心职责包括存储用户画像与偏好用户说过他喜欢靠窗的座位、偏好下午的航班、是某航空公司的金卡会员。这些信息不应该每次对话都让用户重复而应该存入长期记忆并在相关场景下自动激活、提供给短期上下文。积累任务经验与知识Agent成功解决过一个复杂问题比如配置某个特定服务器这个解决方案的步骤、关键参数、遇到的坑和解决办法可以结构化后存入长期记忆。当下次遇到类似问题时可以直接检索参考实现“经验复用”。保存会话摘要当一个长会话结束时可以将本次会话的核心议题、关键决策、最终结果生成一个摘要存入长期记忆。这有助于未来进行更高层次的回顾和关联。设计要点与避坑指南存储不是目的高效检索才是往数据库里扔一堆文本很简单难的是如何在需要的时候精准地找回来。这引出了记忆系统的核心挑战索引与检索Indexing Retrieval。你不能用数据库的LIKE语句去搜效率太低。主流做法是使用向量数据库Vector Database。向量化检索的工作流程编码Embedding当一段信息如用户偏好“我喜欢靠窗座位”需要存入长期记忆时用一个嵌入模型Embedding Model如text-embedding-3-small将其转换为一个高维向量一串数字。存储将这个向量和对应的原始文本或结构化数据一起存入向量数据库如Chroma, Pinecone, Weaviate。检索当新对话发生时将当前的用户问题或对话上下文也编码成向量然后在向量数据库中进行相似度搜索Similarity Search找出最相关的几条历史记忆。注入将检索到的相关记忆文本作为背景信息插入到本轮对话的短期上下文通常是系统指令之后用户问题之前供大模型参考。记忆的“冷启动”问题新用户、新领域一开始长期记忆是空的检索不到东西怎么办设计上要有降级策略。例如可以设置一个置信度阈值如果检索到的记忆相似度低于某个值则认为不相关不注入上下文避免引入噪声。同时系统初期可以更依赖短期上下文和强大的系统指令来完成任务。记忆的更新与冲突用户的偏好可能会变以前喜欢靠窗现在喜欢过道。长期记忆需要支持更新机制。简单的做法是新增一条记录并在检索时给予时间戳更高的记录更大权重。更复杂的可能需要一个记忆合并或版本管理机制。2.3 双记忆系统的协作流程理解了各自职责我们来看它们如何配合完成一次典型的Agent交互用户发起请求“帮我查一下明天北京飞深圳的机票我要靠窗的。”长期记忆检索记忆服务Memory Service将用户query编码在长期记忆库中搜索“用户偏好”。成功检索到“用户喜欢靠窗座位”。组装短期上下文固定系统指令“你是一个机票助手…”。注入检索到的长期记忆“用户历史偏好靠窗座位”。附上本轮及近几轮的对话历史目前为空。加入用户当前query。大模型推理与行动大模型基于丰富的上下文理解任务并决定调用“搜索航班”工具。它生成一个结构化的Function Call请求。执行工具并强制写入短期记忆执行“搜索航班”工具获得结果列表。关键一步必须立即将“工具执行结果找到XX航班…”作为一个function角色的消息追加到短期上下文消息列表的末尾。第二轮模型调用将更新后的包含了工具执行结果的短期上下文再次发送给大模型。模型生成最终回复大模型基于航班结果和用户靠窗偏好筛选并推荐最合适的航班生成回复给用户。选择性写入长期记忆判断本次交互中是否有值得长期保存的信息例如用户最终选择了南航的航班可能暗示他对南航有偏好。如果有将其编码并存入向量数据库。这个流程清晰地展示了短期上下文作为“实时工作流”的载体而长期记忆作为“背景知识库”在关键时刻提供支持。3. 核心模块实现与代码实战理论讲完了我们上干货。这里我以一个Python实现的简易Memory Service为例拆解核心模块。我们假设使用openai库的大模型、chromadb作为向量数据库、langchain的OpenAIEmbeddings做编码。3.1 短期上下文管理器的实现短期上下文管理器的核心是维护一个消息列表并处理token限制。import tiktoken # 用于计算token from typing import List, Dict, Any class ShortTermMemory: def __init__(self, system_prompt: str, model: str gpt-4, max_tokens: int 128000): self.system_prompt system_prompt self.model model self.max_tokens max_tokens # 消息结构[{role: system/user/assistant/function, content: ...}, ...] self.messages: List[Dict[str, str]] [{role: system, content: system_prompt}] self.encoder tiktoken.encoding_for_model(model) def add_message(self, role: str, content: str): 添加一条消息并强制进行容量管理 self.messages.append({role: role, content: content}) self._manage_capacity() def add_function_result(self, function_name: str, result: Any): 专门添加工具执行结果这是防止对话死机的关键 # 将结果转换为字符串复杂对象可以json.dumps content fFunction {function_name} returned: {result} self.add_message(function, content) def _manage_capacity(self): 管理上下文容量采用滑动窗口摘要策略 current_tokens self._count_tokens() if current_tokens self.max_tokens * 0.9: # 留10%缓冲 return # 策略1: 优先移除最早的非系统、非关键对话 # 这里简化处理移除最早的一条用户/助手对话 for i, msg in enumerate(self.messages): if msg[role] in [user, assistant]: removed_msg self.messages.pop(i) print(f[Memory] 移除早期消息以控制长度: {removed_msg[role][:10]}...) break # 如果移除一条后仍然超限触发策略2摘要压缩 if self._count_tokens() self.max_tokens * 0.9: self._summarize_early_conversation() def _count_tokens(self) - int: 计算当前messages列表的总token数 total 0 for msg in self.messages: total len(self.encoder.encode(msg[content])) return total def _summarize_early_conversation(self): 将早期对话压缩成摘要此处为示意实际需调用大模型 # 简化示例这里只是模拟实际项目中你需要调用大模型生成摘要 # 例如将前3轮非系统消息替换为一条摘要消息 print([Memory] 上下文过长触发摘要生成此处为模拟...) # 实际实现略涉及调用LLM和消息列表的重组 def get_context(self) - List[Dict[str, str]]: 获取当前完整的上下文消息列表 return self.messages.copy()关键实现细节与心得add_function_result方法是生命线我把它单独拎出来就是为了强调。确保所有工具回调函数里都必须调用这个方法把结果写回去。Token计算要准确不同模型GPT-3.5, GPT-4, Claude等的编码方式不同tiktoken是OpenAI系的官方方案比较准。算不准会导致实际API调用时报错。容量管理策略需要测试调优滑动窗口移除哪条消息摘要的触发阈值和摘要范围怎么定这需要根据你的任务类型进行AB测试。对于任务型Agent早期的问题定义可能比中间的某个工具结果更重要移除策略需要更智能。3.2 长期记忆服务Memory Service的实现长期记忆服务的核心是向量化的存与取。import chromadb from chromadb.config import Settings from langchain.embeddings import OpenAIEmbeddings # 或其他Embedding模型 import uuid from typing import List, Optional class LongTermMemoryService: def __init__(self, persist_directory: str ./chroma_db): # 初始化嵌入模型 self.embedding_function OpenAIEmbeddings(modeltext-embedding-3-small) # 初始化Chroma客户端持久化存储 self.client chromadb.PersistentClient(pathpersist_directory, settingsSettings(anonymized_telemetryFalse)) # 获取或创建集合类似数据库的表 self.collection self.client.get_or_create_collection(nameagent_memories) def store_memory(self, text: str, metadata: dict None): 存储一段记忆到向量数据库 # 生成唯一ID memory_id str(uuid.uuid4()) # 使用嵌入模型将文本转换为向量 # 注意这里简化了实际中embedding_function.embed_documents是批量处理的 embedding self.embedding_function.embed_documents([text])[0] # 准备元数据 meta metadata or {} meta[text] text # 把原始文本也存一份在元数据里方便查看 # 存入Chroma self.collection.add( ids[memory_id], embeddings[embedding], metadatas[meta], documents[text] # Chroma也可以帮你存原文 ) print(f[LongTermMemory] 已存储记忆: {text[:50]}...) def search_similar_memories(self, query: str, n_results: int 3, threshold: float 0.7) - List[dict]: 根据查询文本搜索最相关的记忆 # 将查询文本向量化 query_embedding self.embedding_function.embed_documents([query])[0] # 在集合中搜索 results self.collection.query( query_embeddings[query_embedding], n_resultsn_results ) # 解析结果results包含 ids, distances, metadatas, documents memories [] if results[ids][0]: # 确保有结果 for i, distance in enumerate(results[distances][0]): # distance是余弦距离越小越相似。可以设置阈值过滤 if distance (1 - threshold): # 近似处理实际根据相似度度量方式调整 memory { text: results[documents][0][i], metadata: results[metadatas][0][i], score: 1 - distance # 转换为相似度分数 } memories.append(memory) return memories def retrieve_relevant_context(self, current_query: str, conversation_history: str ) - str: 检索长期记忆并格式化为可注入上下文的文本 # 可以将当前查询和部分对话历史拼接起来作为检索query提高相关性 search_query f{current_query} {conversation_history[-500:]} if conversation_history else current_query similar_memories self.search_similar_memories(search_query, n_results2) if not similar_memories: return # 将检索到的记忆格式化为文本 context_lines [以下是相关历史信息长期记忆] for mem in similar_memories: context_lines.append(f- {mem[text]} (相关性: {mem[score]:.2f})) return \n.join(context_lines)关键实现细节与心得嵌入模型的选择至关重要text-embedding-3-small在成本和效果间取得了很好的平衡。对于中文场景可能需要考虑bge-large-zh等开源模型。嵌入模型的质量直接决定了检索的准确性。元数据Metadata是富矿存储时除了文本向量尽量把结构化信息放进metadata比如memory_typeuser_preference/task_solution、timestamp、session_id、importance_score。这样后续你可以进行混合检索Hybrid Search即结合向量相似度和元数据过滤例如只检索memory_type为user_preference的记忆精度更高。检索query的构造有技巧直接拿用户当前一句话去搜可能效果不好。更好的做法是把最近的几轮对话比如最后3轮拼接起来作为检索query这样能提供更丰富的上下文线索。这就是上面retrieve_relevant_context方法里conversation_history参数的用意。相似度阈值需要校准threshold参数不是固定的。可以通过人工评估观察在不同阈值下检索到的记忆是否真的相关。不相关的记忆噪声注入上下文反而会干扰大模型判断。3.3 双记忆系统的整合与调度最后我们需要一个AgentMemoryManager来统筹短期和长期记忆。class AgentMemoryManager: def __init__(self, system_prompt: str): self.short_term ShortTermMemory(system_prompt) self.long_term LongTermMemoryService() def process_user_input(self, user_input: str) - List[Dict[str, str]]: 处理用户输入检索长期记忆更新短期上下文返回组装好的完整上下文 # 1. 从长期记忆中检索相关信息 # 可以传入部分短期历史作为检索的上下文 recent_history self._get_recent_history_for_retrieval() long_term_context self.long_term.retrieve_relevant_context(user_input, recent_history) # 2. 如果有长期记忆需要将其作为“系统提示”的补充或独立消息加入短期上下文 # 常见做法在系统消息后用户消息前插入一条角色为system的长期记忆消息。 # 但为了避免混淆我更喜欢用一个独立的角色比如 memory。 full_context_messages [] # 先加入原始系统指令 full_context_messages.extend(self.short_term.get_context()) # 如果有长期记忆插入一条 if long_term_context: # 注意这里插入到系统消息之后历史对话之前 # 我们需要找到系统消息的位置插在它后面 sys_msg_index next(i for i, m in enumerate(full_context_messages) if m[role] system) full_context_messages.insert(sys_msg_index 1, {role: memory, content: long_term_context}) # 3. 将用户本次输入加入短期记忆这也会触发容量管理 self.short_term.add_message(user, user_input) # 4. 将用户输入也追加到本次要返回的上下文列表中 full_context_messages.append({role: user, content: user_input}) return full_context_messages def _get_recent_history_for_retrieval(self) - str: 从短期记忆中提取最近几轮对话文本用于辅助长期记忆检索 # 获取最近N条非系统、非记忆的消息拼接成文本 history_texts [] for msg in self.short_term.messages[-6:]: # 取最近6条 if msg[role] in [user, assistant, function]: # 简单拼接角色和内容 history_texts.append(f{msg[role]}: {msg[content]}) return \n.join(history_texts) def on_function_executed(self, function_name: str, result: Any): 工具执行后的回调结果必须写入短期记忆 self.short_term.add_function_result(function_name, result) def on_conversation_turn_end(self, assistant_reply: str): 一轮对话结束后的回调可选地将本轮关键信息提炼存入长期记忆 self.short_term.add_message(assistant, assistant_reply) # 这里可以添加逻辑判断本轮对话是否产生了值得长期存储的信息 # 例如检测到用户表达了明确的偏好或任务成功完成。 # if self._should_save_to_long_term(user_input, assistant_reply): # memory_text self._summarize_for_long_term(...) # self.long_term.store_memory(memory_text, metadata{...}) def _should_save_to_long_term(self, user_input: str, assistant_reply: str) - bool: 启发式规则判断是否需要存入长期记忆 # 示例规则如果助手回复中包含确认用户偏好的语句 import re preference_patterns [r(记住|下次).*(喜欢|偏好|想要), r您的偏好.*已保存] combined_text user_input assistant_reply for pattern in preference_patterns: if re.search(pattern, combined_text, re.IGNORECASE): return True return False这个管理器扮演了“交通枢纽”的角色它规范了记忆流动的秩序用户输入先触发长期记忆检索检索结果和用户输入一起构成完整的短期上下文工具执行的结果被强制写回短期记忆一轮结束后再评估是否有价值沉淀到长期记忆。4. 实战中的典型问题与排查清单即使架构清晰代码写完在实际跑起来的时候还是会遇到各种妖魔鬼怪。下面是我踩过坑之后总结的常见问题清单和排查思路。问题1Agent突然“失忆”不记得刚刚自己做的事情或用户说过的话。排查点1短期上下文是否被意外清空检查你的ShortTermMemory实例是否在每次请求时被重新创建了。它应该在整个会话周期内保持单例。排查点2Function Call结果是否成功写入在工具调用的回调函数里打日志确认add_function_result被调用且执行成功。检查写入后的self.messages列表里是否确实多了一条role为function的消息。排查点3上下文长度管理是否过于激进检查你的_manage_capacity策略。是不是过早或过多地移除了历史消息尝试调高token阈值或者优化摘要策略确保关键信息如任务目标、约束条件不被移除。问题2长期记忆检索出来的东西不相关甚至干扰模型判断。排查点1嵌入模型是否匹配检查你存储和检索时使用的嵌入模型是否是同一个。不同模型生成的向量空间不同无法直接比较相似度。排查点2检索Query构造是否合理尝试优化你的search_query。单独用用户最后一句话检索和结合最近3轮对话历史一起检索效果可能天差地别。可以尝试不同的拼接方式。排查点3相似度阈值是否合适调低threshold观察检索结果。如果还是很多不相关的可能是嵌入模型对你的领域文本效果不好考虑更换或微调嵌入模型。排查点4记忆“污染”。早期测试时存入了一些无意义或错误的记忆。需要为长期记忆库设计一个管理后台支持查看、编辑、删除记忆条目。生产环境要考虑记忆的“衰减”或“冷存储”机制定期清理低质量、低使用频率的记忆。问题3随着对话轮数增加API调用速度变慢成本飙升。排查点1短期上下文是否无限膨胀这是最常见的原因。确保你的容量管理策略真正生效了。使用tiktoken准确计算token并在日志中输出每轮对话前后的token数监控其增长情况。排查点2长期记忆检索时注入了过多文本检查n_results参数不要一次性注入太多条记忆。通常2-3条最相关的足矣。每条记忆文本也可以进行摘要压缩后再存储减少注入的token量。排查点3系统提示词是否过于冗长精炼你的系统指令去掉不必要的描述。每个token都在花钱。问题4多用户场景下记忆串台了。用户A的信息被检索给了用户B。排查点1向量存储是否隔离最根本的解决方案是在元数据metadata中增加user_id字段。在检索时除了向量相似度必须加上过滤器where{user_id: current_user_id}。这样ChromaDB只会在这个用户的记忆空间里搜索。排查点2短期上下文实例是否共享确保每个用户会话或每个对话线程拥有自己独立的AgentMemoryManager和ShortTermMemory实例。问题5想实现更复杂的记忆比如“记忆图谱”Memory Graph关联不同记忆点感觉现有向量检索不够用。进阶方向你说得对简单的向量检索是“扁平化”的。对于需要深层次推理、记忆间有关联的场景比如“用户喜欢咖啡”和“用户上周买了咖啡机”这两条记忆应该关联可以考虑在上层构建一个记忆图谱。每条记忆作为一个节点通过关系边连接。向量数据库负责基于内容的相似性检索而图数据库如Neo4j负责管理记忆间的逻辑关系。检索时可以先通过向量检索找到种子记忆再通过图谱找到关联记忆。这属于高级玩法对于大多数任务型Agent向量检索丰富元数据过滤已经足够强大。记忆系统的构建是一个从简到繁、持续迭代的过程。我的建议是先从确保“短期上下文不丢、工具结果必回”这个底线开始实现一个可用的版本。然后引入长期记忆解决“用户偏好记忆”这类高价值、易评估的场景。最后再根据业务复杂度逐步考虑摘要、记忆更新、混合检索等高级特性。记住一个稳定可靠的记忆系统是你Agent项目从玩具走向实用的关键一跃。