1. 从“健忘”到“有记忆”：为什么智能体需要记忆系统

在智能体开发的前几篇实战中，我们搭建了感知、决策和行动的基础框架。一个能听会说、能调用工具的智能体已经初具雏形。但如果你和它多聊几句，或者让它处理一个稍长的任务链，很快就会发现一个致命问题：它像个“金鱼”，只有七秒记忆。你刚告诉它“我叫张三，喜欢喝冰美式”，下一句问“给我推荐一杯咖啡”，它可能就会推荐拿铁或卡布奇诺，完全忘了你上一秒的偏好。这种“健忘”让智能体无法进行连贯的对话，也无法执行需要上下文依赖的复杂任务。记忆系统，就是解决这个问题的核心组件，它让智能体从“单次对话模型”进化为“持续交互的智能体”。

简单来说，记忆系统就是智能体的“大脑皮层”，负责信息的存储、索引和提取。没有它，每次交互都是孤立的；有了它，智能体才能认识你、理解任务背景、积累经验，甚至形成独特的“个性”。在智能体开发中，记忆系统绝非一个可选的“加分项”，而是实现实用化、人格化智能体的基石。本次实战，我们将深入一个智能体记忆系统的设计与实现，从零开始构建一个具备短期、长期和工作记忆的完整记忆模块，并解决其中最棘手的几个问题：记忆如何高效存储？如何在海量记忆中快速找到相关的？以及如何防止记忆被无关信息污染？

2. 记忆系统的三层架构设计：短期、长期与工作记忆

人的记忆是分层的，智能体的记忆系统也应如此。直接用一个列表或数据库存储所有对话历史是最简陋的做法，这会导致检索效率低下和记忆干扰。一个健壮的记忆系统通常借鉴认知心理学，设计为三层结构：短期记忆、长期记忆和工作记忆。

2.1 短期记忆：对话的即时上下文

短期记忆，或称对话记忆，保存当前对话轮次中的信息。它的容量有限，生命周期短，通常只持续一次对话会话。在技术实现上，这往往就是传递给大语言模型（LLM）的上下文窗口（Context Window）内的内容。

核心实现要点：

1. 数据结构：通常是一个固定长度的队列（FIFO）。当新的对话轮次（用户输入+智能体回复）加入时，如果队列已满，则自动移除最旧的轮次。
2. 容量管理：容量不是指“条数”，而是“Token数”。你需要根据所用LLM的上下文长度来设定。例如，如果使用上下文窗口为128K的模型，你可能需要为短期记忆预留4K-8K Token，剩下的空间留给系统指令、工具描述和本次查询。
3. 内容格式化：每条记忆不应是原始文本，而应被结构化为一个对象，至少包含role（user/assistant）、content和timestamp。更高级的可以包含type（陈述、问题、指令等）。

from collections import deque import tiktoken # 用于计算Token的库 class ShortTermMemory: def __init__(self, max_token_limit=4000): self.memory = deque() self.max_token_limit = max_token_limit self.current_tokens = 0 self.encoder = tiktoken.get_encoding("cl100k_base") # 适配GPT-4等模型 def add_interaction(self, role, content): """添加一次交互到短期记忆""" interaction = {"role": role, "content": content, "timestamp": time.time()} token_count = len(self.encoder.encode(content)) # 检查并确保加入后不超过限制 while self.current_tokens + token_count > self.max_token_limit and self.memory: removed = self.memory.popleft() self.current_tokens -= len(self.encoder.encode(removed["content"])) self.memory.append(interaction) self.current_tokens += token_count def get_context(self): """获取格式化后的上下文，用于拼接到LLM提示词中""" return [{"role": m["role"], "content": m["content"]} for m in self.memory]

注意：tiktoken是OpenAI官方的Token计算库，但如果你使用非OpenAI系列模型，需要找到对应模型的Tokenizer。短期记忆的溢出管理策略也可以是“总结压缩”，即当快满时，让LLM自动对最早期的几条记忆进行摘要，然后用摘要替换原始内容，但这会引入额外的计算开销。

2.2 长期记忆：智能体的知识库与经验库

长期记忆是智能体的持久化存储，用于保存跨越对话会话的重要信息，例如用户画像、关键事实、执行任务的经验教训等。它的特点是容量大、保存时间长，但检索是关键挑战。

设计核心：向量数据库（Vector Database）纯文本匹配检索在长期记忆中效率低下。例如，用户说“我上次说的那家意大利餐厅”，关键词匹配可能失效。向量数据库通过将文本转换为高维向量（嵌入），并计算向量间的相似度（如余弦相似度）来检索，可以实现语义级别的模糊匹配。

实现步骤：

1. 记忆生成：当判断一条信息需要长期保存时（例如，用户明确说“记住这个”或智能体自行判断为关键信息），将其转换为文本片段。
2. 向量化：使用嵌入模型（如text-embedding-3-small）将该文本转换为一个固定长度的向量。
3. 存储：将向量、原始文本以及元数据（如来源对话ID、时间戳、信息类型标签）一起存入向量数据库。
4. 检索：当需要从长期记忆中回忆时，将当前查询或上下文也转换为向量，然后在向量数据库中搜索最相似的K个记忆片段。

# 示例使用ChromaDB（轻量级向量数据库） import chromadb from chromadb.config import Settings from openai import OpenAI # 假设使用OpenAI Embedding模型 class LongTermMemory: def __init__(self, persist_directory="./chroma_db"): self.client = chromadb.PersistentClient(path=persist_directory) self.collection = self.client.get_or_create_collection(name="agent_memories") self.embed_client = OpenAI() # 初始化Embedding客户端 def _get_embedding(self, text): response = self.embed_client.embeddings.create(model="text-embedding-3-small", input=text) return response.data[0].embedding def store_memory(self, text, metadata={}): """存储一条长期记忆""" embedding = self._get_embedding(text) # ChromaDB会自动生成ID，这里我们用时间戳+哈希生成唯一ID memory_id = f"mem_{int(time.time())}_{hash(text) % 10000:04d}" self.collection.add( embeddings=[embedding], documents=[text], metadatas=[metadata], ids=[memory_id] ) def retrieve_memories(self, query_text, n_results=3): """根据查询检索相关记忆""" query_embedding = self._get_embedding(query_text) results = self.collection.query( query_embeddings=[query_embedding], n_results=n_results ) # results 包含 `documents`, `metadatas`, `distances` 等 return results['documents'][0] if results['documents'] else []

实操心得：元数据（metadata）的设计至关重要。除了时间戳，建议加上entity（实体，如“用户偏好”、“餐厅信息”）、importance（重要性评分，由LLM生成）等字段。这样在检索时，可以结合向量相似度和元数据过滤，得到更精准的结果。例如，可以优先检索与当前对话实体相关且重要性高的记忆。

2.3 工作记忆：任务执行的“草稿纸”

工作记忆是智能体在执行当前任务时的临时信息暂存区。它不同于短期记忆的对话历史，而是专注于一个任务目标的上下文。例如，在完成“订机票-订酒店-租车”的复杂任务时，工作记忆会暂存已查询到的航班信息、酒店选项等，供后续步骤决策使用。

实现模式：工作记忆通常是一个结构化的、任务粒度的字典或对象。它在任务开始时被初始化，在任务执行过程中被不断更新，在任务完成后被清理或归档到长期记忆。

class WorkingMemory: def __init__(self, task_id): self.task_id = task_id self.data = { "goal": "", # 任务目标 "steps": [], # 已完成的步骤 "current_step": "", # 当前步骤 "context": {}, # 任务相关上下文，如收集到的参数 "intermediate_results": {} # 中间结果，如查询到的航班列表 } def update(self, key, value): """更新工作记忆中的特定字段""" if key in self.data: self.data[key] = value else: self.data['context'][key] = value def get(self, key): """获取工作记忆中的值""" return self.data.get(key, self.data['context'].get(key))

三层记忆各司其职：短期记忆维持对话流畅，长期记忆形成持久认知，工作记忆保障复杂任务执行。它们之间需要协同工作，这引出了下一个核心问题：记忆的流动与调度。

3. 记忆的流动、检索与更新策略

记忆不是静态存储的，需要在三层之间智能地流动。同时，如何从海量长期记忆中精准检索，以及如何更新或遗忘记忆，是工程上的难点。

3.1 记忆的写入：什么该记，记在哪里？

并非所有信息都值得记忆。无差别存储会导致记忆库充满噪音。我们需要一个“记忆写入控制器”。

策略建议：

1. 规则触发：定义明确规则。例如，用户说“记住，我喜欢喝茶”或“我的生日是7月10日”，这类包含明确指令或关键个人事实的信息，直接触发写入长期记忆。
2. LLM判断：更灵活的方式是让LLM充当“记忆过滤器”。在每轮对话后，将对话片段交给一个轻量级LLM（或使用主模型的一个特定提示），让其判断：
   • 是否值得存入长期记忆？（提取关键事实、用户偏好、承诺等）
   • 信息类型是什么？（用户属性、事实知识、待办事项等）
   • 重要性分数是多少？（1-10分） 基于LLM的输出，决定是否存储及如何打标签。

def should_save_to_long_term(conversation_snippet, llm_client): prompt = f""" 请分析以下对话片段，判断是否包含值得智能体长期记住的重要信息。 对话：{conversation_snippet} 请按JSON格式输出：{{"save": true/false, "type": "preference/fact/todo", "key_info": "提取的关键信息", "importance": 分数}}。 """ response = llm_client.chat.completions.create( model="gpt-4o-mini", # 使用小模型降低成本 messages=[{"role": "user", "content": prompt}], response_format={"type": "json_object"} ) decision = json.loads(response.choices[0].message.content) return decision

3.2 记忆的检索：在正确的时间想起正确的事

检索是记忆系统的灵魂。核心是相关性和时效性。

混合检索策略：

1. 向量检索（语义相似性）：如上文所述，是基础。它能找到语义上相关的内容。
2. 元数据过滤：结合向量检索的结果，用元数据（如entity、recency）进行二次筛选。例如，当对话主题是“咖啡”时，可以过滤entity包含“咖啡”或“饮品”的记忆。
3. 时间衰减加权：更近的记忆通常更相关。可以在计算最终相关性分数时，加入一个时间衰减因子，让近期记忆的排名提升。
4. LLM重排序（Rerank）：对于初步检索到的Top N条记忆，可以再用LLM根据当前对话的完整上下文，对其相关性进行精细排序和去重。这是效果最好但成本也最高的方法。

def retrieve_relevant_memories(current_query, lt_memory, st_memory, n=5): """综合检索相关记忆""" # 1. 从长期记忆进行向量检索 raw_memories = lt_memory.retrieve_memories(current_query, n_results=n*2) # 2. 构建包含短期记忆的完整上下文 full_context = "\n".join([f"{m['role']}: {m['content']}" for m in st_memory.get_context()[-5:]]) # 取最近5轮 # 3. (可选)LLM重排序与摘要 if raw_memories: prompt = f""" 当前对话上下文： {full_context} 用户最新查询：{current_query} 以下是从知识库中找到的一些可能相关的记忆片段： {chr(10).join([f'{i+1}. {mem}' for i, mem in enumerate(raw_memories)])} 请根据当前对话，选出最相关的1-3条记忆，并简要说明理由。如果都不相关，请回答“无”。 """ # 调用LLM进行判断... # 根据LLM输出选择最终的记忆 return filtered_memories

3.3 记忆的更新与遗忘：保持记忆的鲜活与准确

信息会过时，用户偏好会改变。一个只会增加不会修改的记忆系统最终会充满矛盾。

更新机制：

1. 冲突检测：当要存入的新记忆与已有记忆在语义上高度相似但内容矛盾时（例如，旧记忆“用户喜欢狗”，新记忆“用户对狗毛过敏”），触发更新流程。
2. LLM仲裁：将冲突的记忆对和当前上下文交给LLM，判断哪个信息更可能正确、更新，或是可以合并。
3. 执行操作：根据LLM的判断，执行替换、合并或添加时间戳注释（例如，“截至2023年喜欢狗，2024年起因过敏不再养狗”）。

遗忘（压缩）机制：

1. 基于重要性的清理：定期扫描长期记忆，对重要性评分极低且很久未被访问的记忆进行归档或删除。
2. 总结性压缩：对于关于同一主题的多次琐碎记忆（例如，多次提到“喜欢某家店的拿铁”），可以定期让LLM生成一个总结性的记忆条目，并删除原始琐碎条目。

踩坑实录：在早期版本中，我们只实现了记忆的添加，导致当用户说“我其实不喜欢吃辣了”之后，智能体仍然会推荐川菜，因为旧的“喜欢吃辣”记忆还在且被检索到。实现更新逻辑后，我们让LLM判断“不喜欢吃辣了”是对“喜欢吃辣”的否定更新，从而用新记忆替换了旧记忆的content字段，并在元数据中记录了更新时间，彻底解决了这个问题。

4. 将记忆整合进智能体推理循环

记忆系统不是孤立的模块，它必须深度嵌入智能体的核心推理循环（感知→思考→行动）中。

4.1 在感知阶段注入记忆

在将用户输入（Query）提交给LLM进行核心思考（规划、工具调用判断）之前，先进行记忆检索。

class AgentWithMemory: def __init__(self): self.st_memory = ShortTermMemory() self.lt_memory = LongTermMemory() self.working_memory = None def perceive(self, user_input): """感知阶段：整合记忆""" # 1. 检索相关长期记忆 relevant_memories = retrieve_relevant_memories(user_input, self.lt_memory, self.st_memory) # 2. 获取短期记忆上下文 short_term_context = self.st_memory.get_context() # 3. 获取工作记忆（如果存在活跃任务） working_context = self.working_memory.data if self.working_memory else {} # 4. 构建增强的提示词 enhanced_prompt = self._construct_prompt(user_input, short_term_context, relevant_memories, working_context) return enhanced_prompt def _construct_prompt(self, query, short_term, long_term, working): """构建包含所有上下文的系统提示词""" system_message = f"""你是一个有帮助的智能体，拥有以下记忆和上下文： 关于用户的长期记忆： {chr(10).join(long_term) if long_term else '暂无相关长期记忆。'} 当前任务的工作记忆： {json.dumps(working, indent=2, ensure_ascii=False) if working else '无活跃任务。'} 最近的对话历史（短期记忆）： {chr(10).join([f"{m['role']}: {m['content']}" for m in short_term])} 请基于以上信息，回应用户的最新输入：{query} """ return system_message

4.2 在思考与行动后更新记忆

在LLM生成回复或执行工具调用后，需要根据本轮交互的结果，更新各个记忆层。

def cycle(self, user_input): """智能体的完整交互循环""" # 感知 prompt = self.perceive(user_input) # 思考与行动（调用LLM，可能涉及工具调用） llm_response, used_tools_result = self.think_and_act(prompt) # 记忆更新阶段 # 1. 更新短期记忆 self.st_memory.add_interaction("user", user_input) self.st_memory.add_interaction("assistant", llm_response) # 2. 判断是否存入长期记忆 conversation_snippet = f"User: {user_input}\nAssistant: {llm_response}" save_decision = should_save_to_long_term(conversation_snippet, llm_client) if save_decision.get("save"): self.lt_memory.store_memory( text=save_decision["key_info"], metadata={"type": save_decision["type"], "importance": save_decision["importance"]} ) # 3. 更新工作记忆（如果任务有进展） if self.working_memory and used_tools_result: self.working_memory.update("intermediate_results", used_tools_result) return llm_response

4.3 工作记忆驱动任务分解

对于复杂任务，工作记忆是任务分解（Chain-of-Thought）和执行的协调中心。LLM根据任务目标（存储在working_memory[“goal”]）和已有上下文（working_memory[“context”]），规划下一步，更新工作记忆，然后执行。

5. 实战中的性能优化与常见陷阱

实现一个可用的记忆系统是一回事，实现一个高效、稳定的记忆系统是另一回事。以下是几个关键的优化点和避坑指南。

5.1 向量检索的精度与效率平衡

问题：直接使用用户当前查询语句进行向量检索，可能不够准确，特别是查询很短或很模糊时。解决方案：查询扩展。在检索前，先用LLM对当前查询和对话上下文进行简要分析，生成一个更丰富、包含潜在语义的搜索查询语句。

def expand_query_for_retrieval(user_query, conversation_context): prompt = f"""基于以下对话历史和最新问题，生成一个更全面、用于在知识库中搜索相关信息的查询语句。 历史： {conversation_context} 最新问题：{user_query} 生成的搜索查询：""" expanded_query = llm_client.chat.completions.create(...) # 调用小模型 return expanded_query # 然后用 expanded_query 去检索长期记忆

5.2 记忆token消耗与成本控制

问题：记忆内容（尤其是长期记忆检索结果）会占用大量LLM上下文Token，显著增加API调用成本。解决方案：

1. 记忆摘要：在将长期记忆插入上下文前，如果记忆文本过长，让LLM先对其进行摘要。
2. 选择性注入：不是所有检索到的记忆都放入上下文。可以设定一个相关性分数阈值，或只选择Top 1-2条最相关的记忆。
3. 分层提示：将最核心的指令和最近几轮对话放在提示词最前面，将检索到的记忆放在靠后位置。有些模型对靠前的内容关注度更高。

5.3 记忆幻觉与冲突解决

问题：LLM可能会混淆记忆来源，甚至“脑补”出不存在于记忆库中的细节（幻觉）。或者，记忆库中存在多条相互矛盾的信息。解决方案：

1. 明确引用来源：在向LLM提供记忆时，明确标注来源，例如“[记忆-2024-01-01]：用户说喜欢蓝色”。这能一定程度上减少幻觉。
2. 设置置信度：为每条记忆附加一个置信度分数（基于来源可靠性、确认次数等）。在出现冲突时，优先采用置信度高的记忆。
3. 主动询问：当冲突无法通过规则解决时，最稳妥的方式是让智能体向用户主动确认。“我记得您之前提过喜欢狗，但刚才又提到对狗毛过敏，请问您目前的偏好是怎样的？”

5.4 长期记忆的初始化与冷启动

问题：新智能体长期记忆为空，无法进行个性化服务，影响初期体验。解决方案：

1. 预制知识：根据智能体的定位，预置一些通用知识或领域常识。例如，一个咖啡推荐机器人，可以预置常见咖啡种类、口味特点等记忆。
2. 主动引导：在初次交互时，智能体可以主动询问一些关键信息来填充记忆，例如“为了更好的为您服务，可以告诉我您喜欢的咖啡口味吗？（例如：偏苦/偏酸，加奶/不加奶）”。
3. 从交互中快速学习：设计机制，让智能体在头几次交互中，以更高的频率和更低的阈值将信息存入长期记忆，快速构建用户画像。

构建记忆系统的过程，是让智能体真正“活”起来的关键一步。它从本质上改变了智能体的交互模式，从一次性的问答，变成了持续的、有历史感的陪伴与合作。这套系统目前运行在我们的多个智能体项目中，显著提升了任务完成率和用户满意度。当然，记忆系统本身也是一个持续学习和优化的过程，下一步我们正在探索如何让记忆之间产生更复杂的关联，形成真正的“知识图谱”，但那将是另一个更深入的话题了。