从零构建Agent记忆系统：短期上下文与长期向量化存储实战

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），而且一旦关闭对话窗口（会话结束），这些“工作状态”就清零了。

它的核心职责包括：

1. 承载本轮对话的完整历史：用户最新的Query、Agent的思考过程（Chain-of-Thought）、工具调用（Function Call）的请求和执行结果，以及系统指令（System Prompt）。这里必须划重点：Function Call的“执行结果”必须立刻、马上、无条件地塞进短期上下文！这是血的教训。我见过不止一个项目，工具执行的结果（比如查询到的航班信息、计算出的数据）被直接返回给调用方，却没有在第一时间追加到发给大模型的上下文里。导致下一轮Agent的推理基于的是不完整的信息，轻则答非所问，重则逻辑崩盘，“本轮对话会当场死机”。所以，短期上下文管理器的第一个铁律就是：任何行动的输出，都是后续思考的输入，必须无缝衔接。
2. 维持超短期的状态追踪：比如在一个多步任务中，当前执行到第几步，上一步的输出是什么，下一步的输入依赖是什么。这通常需要一些轻量的状态管理。
3. 提供最快速的上下文检索：当大模型生成需要参考之前的某句话时，短期上下文应该能立刻提供。这通常就是简单的数组或列表，按时间顺序排列。

设计要点与避坑指南：

• 容量管理是头等大事：大模型的上下文窗口是宝贵且有限的资源。你不能让对话历史无限膨胀。必须实现一个摘要（Summarization）或滑动窗口（Sliding Window）策略。例如，当token数接近阈值（如120K）时，自动将最早、最不重要的几轮对话压缩成一个摘要，腾出空间。很多开源框架（如LangChain）的ConversationSummaryBufferMemory就是干这个的。
• 结构化管理优于纯文本：不要把整个对话历史当成一个字符串扔进去。最好将其结构化为消息列表（List[Dict]），每条消息明确角色（user,assistant,system,function）。这样不仅清晰，也便于后续的检索和过滤。例如，你可能只想在特定步骤中让模型参考“所有工具执行的结果”，结构化数据让这变得很容易。
• 系统指令（System Prompt）要“钉”住：确保你的系统指令（定义了Agent的角色、核心能力、约束条件）始终存在于短期上下文的开头，并且不会被摘要或滑动窗口机制挤掉。这是Agent的“人设”根基，不能丢。

2.2 长期外部记忆：Agent的“私人知识库”

如果说短期记忆是工作台，那长期记忆就是你的书架、硬盘、云笔记。它存储那些需要跨会话持久化、在未来可能被反复使用的信息。容量理论上可以无限（取决于你的存储介质），但检索需要时间（I/O操作），并且存在“遗忘”（检索不到）的风险。

它的核心职责包括：

1. 存储用户画像与偏好：用户说过他喜欢靠窗的座位、偏好下午的航班、是某航空公司的金卡会员。这些信息不应该每次对话都让用户重复，而应该存入长期记忆，并在相关场景下自动激活、提供给短期上下文。
2. 积累任务经验与知识：Agent成功解决过一个复杂问题（比如配置某个特定服务器），这个解决方案的步骤、关键参数、遇到的坑和解决办法，可以结构化后存入长期记忆。当下次遇到类似问题时，可以直接检索参考，实现“经验复用”。
3. 保存会话摘要：当一个长会话结束时，可以将本次会话的核心议题、关键决策、最终结果生成一个摘要，存入长期记忆。这有助于未来进行更高层次的回顾和关联。

设计要点与避坑指南：

• 存储不是目的，高效检索才是：往数据库里扔一堆文本很简单，难的是如何在需要的时候精准地找回来。这引出了记忆系统的核心挑战：索引与检索（Indexing & Retrieval）。你不能用数据库的LIKE语句去搜，效率太低。主流做法是使用向量数据库（Vector Database）。
• 向量化检索的工作流程：
  1. 编码（Embedding）：当一段信息（如用户偏好“我喜欢靠窗座位”）需要存入长期记忆时，用一个嵌入模型（Embedding Model，如text-embedding-3-small）将其转换为一个高维向量（一串数字）。
  2. 存储：将这个向量和对应的原始文本（或结构化数据）一起存入向量数据库（如Chroma, Pinecone, Weaviate）。
  3. 检索：当新对话发生时，将当前的用户问题或对话上下文也编码成向量，然后在向量数据库中进行相似度搜索（Similarity Search），找出最相关的几条历史记忆。
  4. 注入：将检索到的相关记忆文本，作为背景信息插入到本轮对话的短期上下文（通常是系统指令之后，用户问题之前），供大模型参考。
• 记忆的“冷启动”问题：新用户、新领域一开始长期记忆是空的，检索不到东西怎么办？设计上要有降级策略。例如，可以设置一个置信度阈值，如果检索到的记忆相似度低于某个值，则认为不相关，不注入上下文，避免引入噪声。同时，系统初期可以更依赖短期上下文和强大的系统指令来完成任务。
• 记忆的更新与冲突：用户的偏好可能会变（以前喜欢靠窗，现在喜欢过道）。长期记忆需要支持更新机制。简单的做法是新增一条记录，并在检索时给予时间戳更高的记录更大权重。更复杂的可能需要一个记忆合并或版本管理机制。

2.3 双记忆系统的协作流程

理解了各自职责，我们来看它们如何配合完成一次典型的Agent交互：

1. 用户发起请求：“帮我查一下明天北京飞深圳的机票，我要靠窗的。”
2. 长期记忆检索：记忆服务（Memory Service）将用户query编码，在长期记忆库中搜索“用户偏好”。成功检索到“用户喜欢靠窗座位”。
3. 组装短期上下文：
   • 固定系统指令（“你是一个机票助手…”）。
   • 注入检索到的长期记忆（“用户历史偏好：靠窗座位”）。
   • 附上本轮及近几轮的对话历史（目前为空）。
   • 加入用户当前query。
4. 大模型推理与行动：大模型基于丰富的上下文，理解任务，并决定调用“搜索航班”工具。它生成一个结构化的Function Call请求。
5. 执行工具并强制写入短期记忆：执行“搜索航班”工具，获得结果列表。关键一步：必须立即将“工具执行结果：找到XX航班…”作为一个function角色的消息，追加到短期上下文消息列表的末尾。
6. 第二轮模型调用：将更新后的（包含了工具执行结果的）短期上下文，再次发送给大模型。
7. 模型生成最终回复：大模型基于航班结果和用户靠窗偏好，筛选并推荐最合适的航班，生成回复给用户。
8. 选择性写入长期记忆：判断本次交互中是否有值得长期保存的信息（例如，用户最终选择了南航的航班，可能暗示他对南航有偏好）。如果有，将其编码并存入向量数据库。

这个流程清晰地展示了短期上下文作为“实时工作流”的载体，而长期记忆作为“背景知识库”在关键时刻提供支持。

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 = f"Function {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()

关键实现细节与心得：

1. add_function_result方法是生命线：我把它单独拎出来，就是为了强调。确保所有工具回调函数里，都必须调用这个方法把结果写回去。
2. Token计算要准确：不同模型（GPT-3.5, GPT-4, Claude等）的编码方式不同，tiktoken是OpenAI系的官方方案，比较准。算不准会导致实际API调用时报错。
3. 容量管理策略需要测试调优：滑动窗口移除哪条消息？摘要的触发阈值和摘要范围怎么定？这需要根据你的任务类型进行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(model="text-embedding-3-small") # 初始化Chroma客户端，持久化存储 self.client = chromadb.PersistentClient(path=persist_directory, settings=Settings(anonymized_telemetry=False)) # 获取或创建集合（类似数据库的表） self.collection = self.client.get_or_create_collection(name="agent_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_results=n_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_results=2) 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)

关键实现细节与心得：

1. 嵌入模型的选择至关重要：text-embedding-3-small在成本和效果间取得了很好的平衡。对于中文场景，可能需要考虑bge-large-zh等开源模型。嵌入模型的质量直接决定了检索的准确性。
2. 元数据（Metadata）是富矿：存储时除了文本向量，尽量把结构化信息放进metadata，比如：memory_type（user_preference/task_solution）、timestamp、session_id、importance_score。这样后续你可以进行混合检索（Hybrid Search），即结合向量相似度和元数据过滤（例如，只检索memory_type为user_preference的记忆），精度更高。
3. 检索query的构造有技巧：直接拿用户当前一句话去搜，可能效果不好。更好的做法是把最近的几轮对话（比如最后3轮）拼接起来作为检索query，这样能提供更丰富的上下文线索。这就是上面retrieve_relevant_context方法里conversation_history参数的用意。
4. 相似度阈值需要校准：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. 实战中的典型问题与排查清单

即使架构清晰，代码写完，在实际跑起来的时候还是会遇到各种妖魔鬼怪。下面是我踩过坑之后总结的常见问题清单和排查思路。

问题1：Agent突然“失忆”，不记得刚刚自己做的事情或用户说过的话。

• 排查点1：短期上下文是否被意外清空？检查你的ShortTermMemory实例是否在每次请求时被重新创建了。它应该在整个会话周期内保持单例。
• 排查点2：Function 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项目从玩具走向实用的关键一跃。