尧图网站建设 
如果你正在开发基于 Claude 的 AI Agent是否遇到过这样的场景你精心设计的 Agent 在完成一次对话后所有关于用户偏好、任务上下文、历史决策的记忆瞬间清零下一次交互它又变回了一张白纸你需要重新解释一遍所有背景。这种“失忆症”不仅让 Agent 显得笨拙更严重制约了其在复杂、长周期任务中的应用价值。这正是当前 AI Agent 开发中的一个核心痛点状态连续性的缺失。一个没有记忆的 Agent就像一台每次开机都恢复出厂设置的电脑无法积累经验更谈不上个性化服务。而Mnemara的出现正是为了解决这个问题。它不是一个全新的 Agent 框架而是一个专为 Claude 等模型设计的“记忆层”。它的核心价值在于让 Agent 在不同会话、甚至不同运行实例之间能够持久化、检索和利用历史信息从而实现真正的“连续智能体”。本文将深入解析 Mnemara 的设计理念、核心原理并通过一个完整的实战示例手把手教你如何为你的 Claude Agent 装上“记忆大脑”。你将了解到Mnemara 如何将抽象的“记忆”转化为可存储、可查询的数据结构。如何通过简单的 SDK 调用实现记忆的存储与读取。在真实开发中如何设计记忆的存储策略、检索策略以及隐私与成本权衡。1. 这篇文章真正要解决的问题为什么 Agent 需要独立的记忆层在深入代码之前我们必须先理解问题本身。很多人认为给 Agent 加记忆不就是把对话历史塞进下一次的 Prompt 里吗这种简单拼接的方式在对话轮次稍多、上下文窗口Context Window有限时会立刻遇到瓶颈成本飙升、响应变慢并且无关的历史信息会干扰当前决策。Mnemara 要解决的是更深层次的三个工程问题1. 上下文长度与成本的矛盾直接将所有历史对话放入 Prompt会迅速耗尽 Claude 模型的上下文令牌Token。这不仅意味着更高的 API 调用成本也可能触及模型的最大上下文限制导致最早的、可能关键的记忆被“挤出”窗口。2. 记忆的精准检索问题并非所有历史信息都对当前任务有用。当用户问“我上次提到的那个项目进展如何”时Agent 需要从海量历史中快速、准确地找到关于“那个项目”的记忆而不是返回所有聊天记录。这需要一套检索机制。3. 记忆的持久化与结构化Agent 进程重启后内存中的对话历史会消失。我们需要一个外部存储系统并且最好能以结构化的方式例如区分“用户偏好”、“任务事实”、“决策逻辑”等保存记忆方便后续的查询和管理。Mnemara 的定位就是一个介于你的应用业务逻辑与大语言模型LLM之间的专用中间件。它接管了记忆的“写”存储与索引、“存”持久化、“读”检索与召回全流程让你的业务代码只需关注“何时记忆”和“如何使用记忆”。2. Mnemara 核心概念与工作原理理解 Mnemara需要掌握几个关键概念记忆Memory 这是 Mnemara 处理的基本单元。它不仅仅是一段文本而是一个结构化的对象。通常包含content: 记忆的文本内容例如“用户喜欢喝黑咖啡不加糖”。metadata: 描述这段记忆的元数据例如来源会话ID、时间戳、记忆类型user_preference,fact,plan、关联实体如project_name: “A项目”等。元数据是后续高效检索的关键。记忆存储Memory Store 负责持久化记忆的组件。Mnemara 设计上支持多种后端例如向量数据库如 Pinecone, Weaviate 将记忆内容转换为向量Embedding存储支持基于语义相似度的检索。这是实现“模糊查找”和“语义关联”记忆的核心。传统数据库如 PostgreSQL, SQLite 基于元数据如时间、类型、标签进行精确过滤和查询适合结构化程度高的记忆。混合检索 结合两者优势先通过元数据过滤范围再用向量检索筛选最相关的内容。检索器Retriever 根据当前查询Query从 Memory Store 中找出最相关的若干条记忆。检索策略决定了记忆的“相关性”。SDK/API Mnemara 提供给开发者的编程接口。通过它你可以执行save_memory(),search_memories()等操作。其工作流程可以概括为以下几步记忆生成 在你的 Agent 运行过程中在关键节点如用户表达明确偏好、任务达成里程碑、做出重要决策调用 Mnemara SDK 的save方法将当前信息连同丰富的元数据保存为一条记忆。记忆索引 Mnemara 内部会将记忆内容转换为向量如果使用向量库并与元数据一同存入配置的 Memory Store。记忆检索 当 Agent 需要历史信息时例如每次处理用户请求前构造一个查询可能是当前用户问题或提炼出的关键词调用search方法。Mnemara 利用检索器从存储中找到最相关的 N 条记忆。记忆注入 将检索到的记忆以一种清晰、有条理的方式例如“根据我们之前的对话1. ... 2. ...”格式化并插入到发给 Claude 模型的 Prompt 上下文中。这样Claude 就能在“知情”的情况下进行回复。这个过程将庞大的、线性的对话历史变成了一个可随时按需查询的、结构化的“知识库”从而实现了 Agent 的连续性。3. 环境准备与项目初始化接下来我们通过一个实战项目来演示。假设我们要开发一个“个人项目助手”Agent它能记住用户提到的各个项目细节、截止日期和个人偏好。3.1 前置条件Python 环境 3.8 或更高版本。本文使用 Python 3.10。Claude API 密钥 你需要一个可用的 Anthropic Claude API 密钥。可以从 Anthropic 控制台获取。向量数据库可选但推荐 为了演示语义检索我们使用一个轻量级、无需服务器的向量库ChromaDB。你也可以选择 Pinecone云服务或 Qdrant自托管。3.2 安装依赖创建一个新的项目目录并初始化虚拟环境。# 创建项目目录并进入 mkdir claude-agent-with-memory cd claude-agent-with-memory # 创建并激活虚拟环境 (Linux/macOS) python3 -m venv venv source venv/bin/activate # 创建并激活虚拟环境 (Windows) # python -m venv venv # venv\Scripts\activate安装核心依赖。由于 Mnemara 可能不是一个广泛安装的库请根据实际网络搜索情况调整这里我们假设其可通过 pip 安装或我们模拟其核心逻辑我们将使用langchain社区中处理记忆的通用模式并结合chromadb来实现一个简化版的“记忆层”。这更能体现其原理。# 安装 Claude SDK、向量数据库和必要工具 pip install anthropic chromadb langchain langchain-chroma # 安装用于生成文本向量的模型这里使用开源的 sentence-transformers # 你也可以使用 OpenAI 的 Embeddings API但需要额外密钥和费用。 pip install sentence-transformers3.3 项目结构创建以下文件结构claude-agent-with-memory/ ├── venv/ # Python 虚拟环境 ├── .env # 环境变量文件用于存储API密钥 ├── memory_layer.py # 核心记忆层实现 ├── agent_core.py # Agent 核心逻辑 └── main.py # 主程序入口4. 实现核心记忆层我们将首先构建一个简化的MemoryLayer类它封装了记忆的存储和检索逻辑。这模拟了 Mnemara SDK 的核心功能。创建memory_layer.py# memory_layer.py import json from datetime import datetime from typing import List, Dict, Any, Optional import chromadb from chromadb.config import Settings from sentence_transformers import SentenceTransformer import numpy as np class MemoryLayer: 一个简化的记忆层实现模拟 Mnemara 的核心功能。 使用 ChromaDB 作为向量存储sentence-transformers 生成嵌入向量。 def __init__(self, persist_directory: str ./chroma_db): 初始化记忆层。 :param persist_directory: ChromaDB 数据持久化目录 # 初始化嵌入模型用于将文本转换为向量 # 使用一个轻量且效果不错的模型 self.embedding_model SentenceTransformer(all-MiniLM-L6-v2) # 初始化 ChromaDB 客户端并创建一个集合collection来存储记忆 self.client chromadb.PersistentClient(pathpersist_directory) # 集合名称 self.collection_name agent_memories # 获取或创建集合 self.collection self.client.get_or_create_collection( nameself.collection_name, metadata{description: Storage for AI agents long-term memories} ) print(fMemoryLayer initialized. Persisting data to: {persist_directory}) def _generate_embedding(self, text: str) - List[float]: 为文本生成嵌入向量。 # sentence-transformers 模型直接返回 numpy array转换为 list embedding self.embedding_model.encode(text) return embedding.tolist() def save_memory(self, content: str, metadata: Optional[Dict[str, Any]] None) - str: 保存一条记忆。 :param content: 记忆的文本内容 :param metadata: 关联的元数据例如 {type: user_preference, project: ProjectX} :return: 该记忆的唯一ID if metadata is None: metadata {} # 确保有基本的时间戳 if timestamp not in metadata: metadata[timestamp] datetime.now().isoformat() # 为内容生成向量 embedding self._generate_embedding(content) # 生成一个唯一ID在实际生产中可能使用UUID memory_id fmem_{datetime.now().strftime(%Y%m%d_%H%M%S_%f)} # 存储到 ChromaDB self.collection.add( embeddings[embedding], documents[content], metadatas[metadata], ids[memory_id] ) print(fMemory saved. ID: {memory_id}, Type: {metadata.get(type, N/A)}) return memory_id def search_memories(self, query: str, filter_conditions: Optional[Dict[str, Any]] None, n_results: int 5) - List[Dict[str, Any]]: 搜索相关记忆。 :param query: 查询文本 :param filter_conditions: 对元数据进行过滤的条件例如 {type: user_preference} :param n_results: 返回最相关的记忆条数 :return: 包含记忆内容、元数据和相关度的字典列表 # 为查询文本生成向量 query_embedding self._generate_embedding(query) # 执行查询 # ChromaDB 的 query 方法可以同时接受 embedding 和 where 过滤器 results self.collection.query( query_embeddings[query_embedding], n_resultsn_results, wherefilter_conditions, # 元数据过滤 include[documents, metadatas, distances] ) # 格式化返回结果 memories [] if results[documents]: for i in range(len(results[documents][0])): memory { content: results[documents][0][i], metadata: results[metadatas][0][i], similarity_score: 1 - results[distances][0][i], # 将距离转换为相似度分数余弦距离 id: results[ids][0][i] } memories.append(memory) # 按相似度降序排序 memories.sort(keylambda x: x[similarity_score], reverseTrue) return memories def get_all_memories(self, limit: int 50) - List[Dict[str, Any]]: 获取所有记忆按时间倒序用于调试。 # ChromaDB 的 get 方法可以获取所有数据 all_data self.collection.get() memories [] for i in range(len(all_data[ids])): memory { id: all_data[ids][i], content: all_data[documents][i], metadata: all_data[metadatas][i] } memories.append(memory) # 按时间戳倒序 memories.sort(keylambda x: x[metadata].get(timestamp, ), reverseTrue) return memories[:limit] # 提供一个全局实例单例模式简单演示 memory_layer_instance MemoryLayer()关键代码解释初始化 类初始化时加载嵌入模型并连接 ChromaDB。all-MiniLM-L6-v2是一个轻量级的句子转换模型适合本地运行。save_memory 接收记忆内容和元数据生成向量并存储到 ChromaDB 集合中。每条记忆都有一个唯一 ID。search_memories 这是核心。它接收一个查询字符串如用户当前问题将其转换为向量然后在向量空间中找到最相似的记忆。filter_conditions参数允许我们根据元数据如type进行筛选实现混合检索。get_all_memories 辅助方法用于查看所有记忆方便调试。这个类已经具备了 Mnemara 记忆层最核心的“存储”和“语义检索”能力。5. 构建具有记忆的 Claude Agent接下来我们创建 Agent 的核心逻辑它将利用上面的记忆层。创建agent_core.py# agent_core.py import os from typing import List, Dict, Any from anthropic import Anthropic from dotenv import load_dotenv from memory_layer import memory_layer_instance # 加载环境变量.env文件中的CLAUDE_API_KEY load_dotenv() class ClaudeAgentWithMemory: def __init__(self): # 初始化 Claude 客户端 self.api_key os.getenv(CLAUDE_API_KEY) if not self.api_key: raise ValueError(请设置环境变量 CLAUDE_API_KEY。请在 .env 文件中添加CLAUDE_API_KEY你的密钥) self.client Anthropic(api_keyself.api_key) # 使用我们创建的记忆层实例 self.memory_layer memory_layer_instance # 系统提示词定义了 Agent 的角色和能力并预留了插入记忆的位置 self.base_system_prompt 你是一个高效的个人项目助手。你拥有与用户互动的完整记忆。 以下是与你当前任务相关的过往记忆按相关性排序 {formatted_memories} 请基于这些记忆和当前对话为用户提供准确、连贯的帮助。 如果记忆中有相关信息请直接利用。如果用户的问题涉及新信息请将其更新到你的知识中我会在后台保存。 回答应简洁、专业、有帮助。 def _format_memories_for_prompt(self, memories: List[Dict[str, Any]]) - str: 将检索到的记忆列表格式化为适合放入 Prompt 的文本。 if not memories: return 暂无相关记忆 formatted [] for i, mem in enumerate(memories, 1): mem_content mem[content] mem_type mem[metadata].get(type, general) # 可选加入相关性分数让模型知道信息的可信度 # score mem.get(similarity_score, 0) # formatted.append(f{i}. [{mem_type}] {mem_content} (相关性: {score:.2f})) formatted.append(f{i}. [{mem_type}] {mem_content}) return \n.join(formatted) def _extract_metadata_from_conversation(self, user_input: str, assistant_response: str) - Dict[str, Any]: 一个简单的元数据提取函数。 在实际应用中这里可以用更复杂的逻辑或另一个LLM调用来分析对话提取实体、类型等。 此处我们做一个简单的规则模拟。 metadata {source: conversation} # 简单关键词匹配来确定记忆类型实际项目应更智能 type_keywords { user_preference: [喜欢, 偏好, 讨厌, 希望, 想要, 觉得], project_detail: [项目, 任务, deadline, 截止, 进度, 需求], fact: [是, 有, 在, 包括], # 很宽泛 } input_lower user_input.lower() for mem_type, keywords in type_keywords.items(): if any(keyword in input_lower for keyword in keywords): metadata[type] mem_type break else: metadata[type] general # 可以尝试提取项目名简单演示 if project in metadata.get(type, ): # 非常简单的提取实际应用需要NLP words user_input.split() for i, word in enumerate(words): if word.lower() in [project, 项目, 任务] and i1 len(words): metadata[project_name] words[i1].strip(.,!?\) break return metadata def chat_round(self, user_input: str) - str: 处理一轮用户输入并返回助手的回复。 这是核心交互循环。 # 步骤1在回复前先根据用户输入检索相关记忆 relevant_memories self.memory_layer.search_memories( queryuser_input, n_results3 # 每次携带3条最相关的记忆避免上下文过长 ) # 步骤2格式化记忆并构建完整的系统提示词 formatted_mems self._format_memories_for_prompt(relevant_memories) system_prompt self.base_system_prompt.format(formatted_memoriesformatted_mems) # 步骤3调用 Claude API 获取回复 try: message self.client.messages.create( modelclaude-3-haiku-20240307, # 使用成本较低的 Haiku 模型进行演示 max_tokens500, systemsystem_prompt, messages[ {role: user, content: user_input} ] ) assistant_response message.content[0].text except Exception as e: assistant_response f调用 Claude API 时出错{e} return assistant_response # 步骤4判断当前对话是否值得作为长期记忆保存 # 这里是一个简单的启发式规则如果用户输入包含明确的事实陈述、偏好或任务信息则保存。 should_save any(keyword in user_input.lower() for keyword in [ 记住, 我喜欢, 我讨厌, 项目需要, 截止日期是, 目标是 ]) if should_save: # 提取元数据并保存记忆 metadata self._extract_metadata_from_conversation(user_input, assistant_response) # 我们将用户输入作为记忆内容保存。更复杂的场景可以保存整个对话轮次或提炼后的要点。 self.memory_layer.save_memory(contentuser_input, metadatametadata) print(f[系统] 已将本次对话的关键信息存入长期记忆。) return assistant_response def run_conversation(self): 运行一个简单的对话循环。 print( * 50) print(个人项目助手已启用长期记忆已启动。) print(输入 退出 或 quit 结束对话。) print(输入 查看记忆 可以查看所有存储的记忆。) print( * 50) while True: try: user_input input(\n你: ).strip() except (EOFError, KeyboardInterrupt): print(\n再见) break if user_input.lower() in [退出, quit, exit]: print(助手: 再见期待下次继续为您管理项目。) break if user_input.lower() 查看记忆: memories self.memory_layer.get_all_memories(limit10) print(\n 存储的所有记忆最近10条) for mem in memories: print(f- ID: {mem[id]}) print(f 内容: {mem[content]}) print(f 元数据: {mem[metadata]}) print() continue # 正常对话处理 response self.chat_round(user_input) print(f助手: {response})关键逻辑解析__init__ 初始化 Claude 客户端和记忆层。base_system_prompt包含一个占位符{formatted_memories}用于在每次调用前插入检索到的相关记忆。chat_round 单轮对话的核心方法。检索 首先用用户输入作为查询去记忆层搜索相关记忆。构建上下文 将记忆格式化后填入系统提示词形成 Claude 本次调用的完整“背景”。调用模型 使用包含记忆的上下文调用 Claude API。选择性保存 根据简单规则可扩展为更复杂的逻辑或另一个 LLM 判断决定是否将本轮对话保存为新记忆。这模拟了 Agent 的“学习”过程。run_conversation 提供一个简单的命令行交互界面。6. 运行与效果验证现在让我们将整个项目运行起来看看记忆层如何工作。6.1 配置环境变量在项目根目录创建.env文件填入你的 Claude API 密钥。# .env 文件内容 CLAUDE_API_KEY你的-claude-api-key-here重要 请勿将.env文件提交到版本控制系统如 Git。确保它在.gitignore中。6.2 创建主程序入口创建main.py# main.py from agent_core import ClaudeAgentWithMemory def main(): agent ClaudeAgentWithMemory() agent.run_conversation() if __name__ __main__: main()6.3 启动对话在终端运行python main.py6.4 交互示例与验证让我们模拟一个跨越多次对话的连续任务场景个人项目助手已启用长期记忆已启动。 输入 退出 或 quit 结束对话。 输入 查看记忆 可以查看所有存储的记忆。 你: 你好请帮我记住我正在进行的项目叫“星辰计划”下周一是原型设计的截止日期。 助手: 好的我已经记下了。您的项目“星辰计划”的原型设计截止日期是下周一。我会在相关时间提醒您。 你: 另外关于这个项目我比较喜欢用蓝色的主题色。 助手: 明白已将“星辰计划”的主题色偏好蓝色记录下来。 你: 查看记忆 存储的所有记忆最近10条 - ID: mem_20231027_143022_123456 内容: 你好请帮我记住我正在进行的项目叫“星辰计划”下周一是原型设计的截止日期。 元数据: {source: conversation, type: project_detail, project_name: 星辰计划, timestamp: 2023-10-27T14:30:22.123456} - ID: mem_20231027_143045_654321 内容: 另外关于这个项目我比较喜欢用蓝色的主题色。 元数据: {source: conversation, type: user_preference, project_name: 星辰计划, timestamp: 2023-10-27T14:30:45.654321} 你: “星辰计划”的截止日是什么时候我用的主题色是什么 助手: 根据我们的对话记录 1. [project_detail] 你好请帮我记住我正在进行的项目叫“星辰计划”下周一是原型设计的截止日期。 2. [user_preference] 另外关于这个项目我比较喜欢用蓝色的主题色。 因此“星辰计划”的原型设计截止日期是下周一您为该项目选择的主题色是蓝色。效果验证点记忆保存 当用户提供明确信息项目名、截止日、偏好时Agent 识别并保存了记忆控制台有提示且可通过“查看记忆”命令确认。记忆检索 当用户后续询问“星辰计划”的细节时Agent 在回复中直接引用了之前保存的两条具体记忆而不是泛泛而谈。这证明记忆层成功检索到了相关信息。连续性体现 即使在同一个会话中如果没有记忆层后一个问题也需要在 Prompt 中包含整个历史对话。而我们的实现仅通过检索到的2条关键记忆就实现了精准回复大大节省了上下文令牌。7. 常见问题与排查思路在实际集成 Mnemara 或自建记忆层时你可能会遇到以下问题问题现象可能原因排查方式解决方案调用save_memory失败连接错误ChromaDB 持久化目录权限问题嵌入模型下载失败。1. 检查persist_directory路径是否存在且可写。2. 查看控制台初始化日志检查sentence-transformers模型是否下载成功。1. 确保程序对目标目录有读写权限。2. 对于模型下载可以尝试设置代理或使用离线模型文件。搜索记忆时返回结果不相关1. 查询文本与记忆内容语义差异大。2. 嵌入模型不适合当前领域。3. 元数据过滤条件太严格。1. 打印出search_memories返回的原始结果和相似度分数。2. 检查记忆保存时的内容是否清晰、完整。3. 尝试不使用过滤器进行搜索。1. 优化记忆的保存内容使其更具概括性。2. 尝试更换更适合的嵌入模型如all-mpnet-base-v2效果更好但更慢。3. 调整检索策略结合关键词匹配进行混合检索。Claude 回复未利用记忆1. 记忆检索失败或为空。2. 系统提示词中记忆的格式化方式不佳模型“看不到”或“不理解”。3. 检索到的记忆过多淹没了当前指令。1. 在chat_round中打印relevant_memories和formatted_mems。2. 检查最终发送给 Claude 的完整 Prompt可临时打印。1. 确保记忆检索逻辑正确查询文本有代表性。2. 优化_format_memories_for_prompt函数让记忆的呈现更清晰如编号、加标签。3. 减少n_results只保留最相关的1-3条记忆。记忆保存过于频繁存储膨胀保存记忆的触发条件should_save太宽松。查看存储的记忆列表判断哪些是不必要保存的。设计更精细的记忆保存策略。例如1. 只保存用户明确指令“请记住”的内容。2. 使用另一个轻量级LLM或规则判断信息是否具有长期价值。3. 定期清理过时或低访问频率的记忆。程序重启后Agent“失忆”记忆层未正确配置持久化。检查 ChromaDB 的PersistentClient路径是否正确以及程序重启后是否加载了同一个集合。确保MemoryLayer初始化时使用的persist_directory路径一致。ChromaDB 会从该目录加载已有数据。8. 最佳实践与工程建议将记忆层投入生产环境需要考虑更多工程细节记忆的粒度与结构避免保存原始对话 不要简单保存“用户说X助手说Y”。应该提炼出核心事实、决策或偏好。可以考虑用一个小型LLM如 Claude Haiku对对话进行摘要再将摘要保存为记忆。设计元数据模式 提前规划好metadata的字段如type(fact, preference, goal, action),entity(项目名、人名),priority,expiry_date等。良好的元数据是高效检索的基础。检索策略优化混合检索 结合向量检索语义相似和基于元数据的过滤精确匹配。例如先过滤type“user_preference” AND entity“ProjectA”再在结果中进行向量检索。查询重写 在检索前可以用LLM将用户的自然语言问题重写为更利于检索的关键词或陈述句。记忆评分与衰减 为记忆设计一个“重要性”或“新鲜度”分数。频繁被检索到的记忆得分更高久未使用的记忆得分衰减。检索时按综合分排序。成本与性能权衡控制上下文长度 严格限制注入到 Prompt 中的记忆条数和总令牌数。这是控制成本最直接的手段。异步记忆处理 保存记忆生成嵌入向量和更新索引可能是耗时操作应将其放入后台任务队列避免阻塞主对话流程。缓存热点记忆 对于高频使用的记忆如用户姓名、基础偏好可以缓存在应用内存中避免每次检索都查询向量库。隐私与安全数据隔离 确保不同用户、不同租户的记忆数据完全隔离。可以在 ChromaDB 中为每个用户/会话创建独立的集合或在元数据中加入user_id并严格过滤。记忆审查与删除 提供用户查看和删除其个人记忆的接口满足数据合规要求如 GDPR。敏感信息过滤 在保存记忆前可以增加一个过滤层检测并脱敏密码、密钥、个人身份信息等敏感内容。与现有 Agent 框架集成LangChain LangChain 有成熟的内存模块ConversationBufferMemory,ConversationSummaryMemory,VectorStoreRetrieverMemory。我们的MemoryLayer可以视为一个自定义的BaseMemory实现或与VectorStoreRetriever结合。LlamaIndex LlamaIndex 本身就是处理索引和检索的专家。你可以用 LlamaIndex 来构建和管理记忆的索引然后将其查询接口封装到你的记忆层中。通过本文的实战你已经掌握了为 Claude Agent 构建记忆层的核心逻辑。从简单的向量存储检索到复杂的记忆生命周期管理这其中的每一步都决定了你的 Agent 是“金鱼”还是“顾问”。真正的挑战不在于实现存储而在于设计一套让记忆产生最大价值的策略——记住什么、何时记住、如何想起。 为Claude AI Agent构建记忆层：解决状态连续性缺失的实战指南
如果你正在开发基于 Claude 的 AI Agent是否遇到过这样的场景你精心设计的 Agent 在完成一次对话后所有关于用户偏好、任务上下文、历史决策的记忆瞬间清零下一次交互它又变回了一张白纸你需要重新解释一遍所有背景。这种“失忆症”不仅让 Agent 显得笨拙更严重制约了其在复杂、长周期任务中的应用价值。这正是当前 AI Agent 开发中的一个核心痛点状态连续性的缺失。一个没有记忆的 Agent就像一台每次开机都恢复出厂设置的电脑无法积累经验更谈不上个性化服务。而Mnemara的出现正是为了解决这个问题。它不是一个全新的 Agent 框架而是一个专为 Claude 等模型设计的“记忆层”。它的核心价值在于让 Agent 在不同会话、甚至不同运行实例之间能够持久化、检索和利用历史信息从而实现真正的“连续智能体”。本文将深入解析 Mnemara 的设计理念、核心原理并通过一个完整的实战示例手把手教你如何为你的 Claude Agent 装上“记忆大脑”。你将了解到Mnemara 如何将抽象的“记忆”转化为可存储、可查询的数据结构。如何通过简单的 SDK 调用实现记忆的存储与读取。在真实开发中如何设计记忆的存储策略、检索策略以及隐私与成本权衡。1. 这篇文章真正要解决的问题为什么 Agent 需要独立的记忆层在深入代码之前我们必须先理解问题本身。很多人认为给 Agent 加记忆不就是把对话历史塞进下一次的 Prompt 里吗这种简单拼接的方式在对话轮次稍多、上下文窗口Context Window有限时会立刻遇到瓶颈成本飙升、响应变慢并且无关的历史信息会干扰当前决策。Mnemara 要解决的是更深层次的三个工程问题1. 上下文长度与成本的矛盾直接将所有历史对话放入 Prompt会迅速耗尽 Claude 模型的上下文令牌Token。这不仅意味着更高的 API 调用成本也可能触及模型的最大上下文限制导致最早的、可能关键的记忆被“挤出”窗口。2. 记忆的精准检索问题并非所有历史信息都对当前任务有用。当用户问“我上次提到的那个项目进展如何”时Agent 需要从海量历史中快速、准确地找到关于“那个项目”的记忆而不是返回所有聊天记录。这需要一套检索机制。3. 记忆的持久化与结构化Agent 进程重启后内存中的对话历史会消失。我们需要一个外部存储系统并且最好能以结构化的方式例如区分“用户偏好”、“任务事实”、“决策逻辑”等保存记忆方便后续的查询和管理。Mnemara 的定位就是一个介于你的应用业务逻辑与大语言模型LLM之间的专用中间件。它接管了记忆的“写”存储与索引、“存”持久化、“读”检索与召回全流程让你的业务代码只需关注“何时记忆”和“如何使用记忆”。2. Mnemara 核心概念与工作原理理解 Mnemara需要掌握几个关键概念记忆Memory 这是 Mnemara 处理的基本单元。它不仅仅是一段文本而是一个结构化的对象。通常包含content: 记忆的文本内容例如“用户喜欢喝黑咖啡不加糖”。metadata: 描述这段记忆的元数据例如来源会话ID、时间戳、记忆类型user_preference,fact,plan、关联实体如project_name: “A项目”等。元数据是后续高效检索的关键。记忆存储Memory Store 负责持久化记忆的组件。Mnemara 设计上支持多种后端例如向量数据库如 Pinecone, Weaviate 将记忆内容转换为向量Embedding存储支持基于语义相似度的检索。这是实现“模糊查找”和“语义关联”记忆的核心。传统数据库如 PostgreSQL, SQLite 基于元数据如时间、类型、标签进行精确过滤和查询适合结构化程度高的记忆。混合检索 结合两者优势先通过元数据过滤范围再用向量检索筛选最相关的内容。检索器Retriever 根据当前查询Query从 Memory Store 中找出最相关的若干条记忆。检索策略决定了记忆的“相关性”。SDK/API Mnemara 提供给开发者的编程接口。通过它你可以执行save_memory(),search_memories()等操作。其工作流程可以概括为以下几步记忆生成 在你的 Agent 运行过程中在关键节点如用户表达明确偏好、任务达成里程碑、做出重要决策调用 Mnemara SDK 的save方法将当前信息连同丰富的元数据保存为一条记忆。记忆索引 Mnemara 内部会将记忆内容转换为向量如果使用向量库并与元数据一同存入配置的 Memory Store。记忆检索 当 Agent 需要历史信息时例如每次处理用户请求前构造一个查询可能是当前用户问题或提炼出的关键词调用search方法。Mnemara 利用检索器从存储中找到最相关的 N 条记忆。记忆注入 将检索到的记忆以一种清晰、有条理的方式例如“根据我们之前的对话1. ... 2. ...”格式化并插入到发给 Claude 模型的 Prompt 上下文中。这样Claude 就能在“知情”的情况下进行回复。这个过程将庞大的、线性的对话历史变成了一个可随时按需查询的、结构化的“知识库”从而实现了 Agent 的连续性。3. 环境准备与项目初始化接下来我们通过一个实战项目来演示。假设我们要开发一个“个人项目助手”Agent它能记住用户提到的各个项目细节、截止日期和个人偏好。3.1 前置条件Python 环境 3.8 或更高版本。本文使用 Python 3.10。Claude API 密钥 你需要一个可用的 Anthropic Claude API 密钥。可以从 Anthropic 控制台获取。向量数据库可选但推荐 为了演示语义检索我们使用一个轻量级、无需服务器的向量库ChromaDB。你也可以选择 Pinecone云服务或 Qdrant自托管。3.2 安装依赖创建一个新的项目目录并初始化虚拟环境。# 创建项目目录并进入 mkdir claude-agent-with-memory cd claude-agent-with-memory # 创建并激活虚拟环境 (Linux/macOS) python3 -m venv venv source venv/bin/activate # 创建并激活虚拟环境 (Windows) # python -m venv venv # venv\Scripts\activate安装核心依赖。由于 Mnemara 可能不是一个广泛安装的库请根据实际网络搜索情况调整这里我们假设其可通过 pip 安装或我们模拟其核心逻辑我们将使用langchain社区中处理记忆的通用模式并结合chromadb来实现一个简化版的“记忆层”。这更能体现其原理。# 安装 Claude SDK、向量数据库和必要工具 pip install anthropic chromadb langchain langchain-chroma # 安装用于生成文本向量的模型这里使用开源的 sentence-transformers # 你也可以使用 OpenAI 的 Embeddings API但需要额外密钥和费用。 pip install sentence-transformers3.3 项目结构创建以下文件结构claude-agent-with-memory/ ├── venv/ # Python 虚拟环境 ├── .env # 环境变量文件用于存储API密钥 ├── memory_layer.py # 核心记忆层实现 ├── agent_core.py # Agent 核心逻辑 └── main.py # 主程序入口4. 实现核心记忆层我们将首先构建一个简化的MemoryLayer类它封装了记忆的存储和检索逻辑。这模拟了 Mnemara SDK 的核心功能。创建memory_layer.py# memory_layer.py import json from datetime import datetime from typing import List, Dict, Any, Optional import chromadb from chromadb.config import Settings from sentence_transformers import SentenceTransformer import numpy as np class MemoryLayer: 一个简化的记忆层实现模拟 Mnemara 的核心功能。 使用 ChromaDB 作为向量存储sentence-transformers 生成嵌入向量。 def __init__(self, persist_directory: str ./chroma_db): 初始化记忆层。 :param persist_directory: ChromaDB 数据持久化目录 # 初始化嵌入模型用于将文本转换为向量 # 使用一个轻量且效果不错的模型 self.embedding_model SentenceTransformer(all-MiniLM-L6-v2) # 初始化 ChromaDB 客户端并创建一个集合collection来存储记忆 self.client chromadb.PersistentClient(pathpersist_directory) # 集合名称 self.collection_name agent_memories # 获取或创建集合 self.collection self.client.get_or_create_collection( nameself.collection_name, metadata{description: Storage for AI agents long-term memories} ) print(fMemoryLayer initialized. Persisting data to: {persist_directory}) def _generate_embedding(self, text: str) - List[float]: 为文本生成嵌入向量。 # sentence-transformers 模型直接返回 numpy array转换为 list embedding self.embedding_model.encode(text) return embedding.tolist() def save_memory(self, content: str, metadata: Optional[Dict[str, Any]] None) - str: 保存一条记忆。 :param content: 记忆的文本内容 :param metadata: 关联的元数据例如 {type: user_preference, project: ProjectX} :return: 该记忆的唯一ID if metadata is None: metadata {} # 确保有基本的时间戳 if timestamp not in metadata: metadata[timestamp] datetime.now().isoformat() # 为内容生成向量 embedding self._generate_embedding(content) # 生成一个唯一ID在实际生产中可能使用UUID memory_id fmem_{datetime.now().strftime(%Y%m%d_%H%M%S_%f)} # 存储到 ChromaDB self.collection.add( embeddings[embedding], documents[content], metadatas[metadata], ids[memory_id] ) print(fMemory saved. ID: {memory_id}, Type: {metadata.get(type, N/A)}) return memory_id def search_memories(self, query: str, filter_conditions: Optional[Dict[str, Any]] None, n_results: int 5) - List[Dict[str, Any]]: 搜索相关记忆。 :param query: 查询文本 :param filter_conditions: 对元数据进行过滤的条件例如 {type: user_preference} :param n_results: 返回最相关的记忆条数 :return: 包含记忆内容、元数据和相关度的字典列表 # 为查询文本生成向量 query_embedding self._generate_embedding(query) # 执行查询 # ChromaDB 的 query 方法可以同时接受 embedding 和 where 过滤器 results self.collection.query( query_embeddings[query_embedding], n_resultsn_results, wherefilter_conditions, # 元数据过滤 include[documents, metadatas, distances] ) # 格式化返回结果 memories [] if results[documents]: for i in range(len(results[documents][0])): memory { content: results[documents][0][i], metadata: results[metadatas][0][i], similarity_score: 1 - results[distances][0][i], # 将距离转换为相似度分数余弦距离 id: results[ids][0][i] } memories.append(memory) # 按相似度降序排序 memories.sort(keylambda x: x[similarity_score], reverseTrue) return memories def get_all_memories(self, limit: int 50) - List[Dict[str, Any]]: 获取所有记忆按时间倒序用于调试。 # ChromaDB 的 get 方法可以获取所有数据 all_data self.collection.get() memories [] for i in range(len(all_data[ids])): memory { id: all_data[ids][i], content: all_data[documents][i], metadata: all_data[metadatas][i] } memories.append(memory) # 按时间戳倒序 memories.sort(keylambda x: x[metadata].get(timestamp, ), reverseTrue) return memories[:limit] # 提供一个全局实例单例模式简单演示 memory_layer_instance MemoryLayer()关键代码解释初始化 类初始化时加载嵌入模型并连接 ChromaDB。all-MiniLM-L6-v2是一个轻量级的句子转换模型适合本地运行。save_memory 接收记忆内容和元数据生成向量并存储到 ChromaDB 集合中。每条记忆都有一个唯一 ID。search_memories 这是核心。它接收一个查询字符串如用户当前问题将其转换为向量然后在向量空间中找到最相似的记忆。filter_conditions参数允许我们根据元数据如type进行筛选实现混合检索。get_all_memories 辅助方法用于查看所有记忆方便调试。这个类已经具备了 Mnemara 记忆层最核心的“存储”和“语义检索”能力。5. 构建具有记忆的 Claude Agent接下来我们创建 Agent 的核心逻辑它将利用上面的记忆层。创建agent_core.py# agent_core.py import os from typing import List, Dict, Any from anthropic import Anthropic from dotenv import load_dotenv from memory_layer import memory_layer_instance # 加载环境变量.env文件中的CLAUDE_API_KEY load_dotenv() class ClaudeAgentWithMemory: def __init__(self): # 初始化 Claude 客户端 self.api_key os.getenv(CLAUDE_API_KEY) if not self.api_key: raise ValueError(请设置环境变量 CLAUDE_API_KEY。请在 .env 文件中添加CLAUDE_API_KEY你的密钥) self.client Anthropic(api_keyself.api_key) # 使用我们创建的记忆层实例 self.memory_layer memory_layer_instance # 系统提示词定义了 Agent 的角色和能力并预留了插入记忆的位置 self.base_system_prompt 你是一个高效的个人项目助手。你拥有与用户互动的完整记忆。 以下是与你当前任务相关的过往记忆按相关性排序 {formatted_memories} 请基于这些记忆和当前对话为用户提供准确、连贯的帮助。 如果记忆中有相关信息请直接利用。如果用户的问题涉及新信息请将其更新到你的知识中我会在后台保存。 回答应简洁、专业、有帮助。 def _format_memories_for_prompt(self, memories: List[Dict[str, Any]]) - str: 将检索到的记忆列表格式化为适合放入 Prompt 的文本。 if not memories: return 暂无相关记忆 formatted [] for i, mem in enumerate(memories, 1): mem_content mem[content] mem_type mem[metadata].get(type, general) # 可选加入相关性分数让模型知道信息的可信度 # score mem.get(similarity_score, 0) # formatted.append(f{i}. [{mem_type}] {mem_content} (相关性: {score:.2f})) formatted.append(f{i}. [{mem_type}] {mem_content}) return \n.join(formatted) def _extract_metadata_from_conversation(self, user_input: str, assistant_response: str) - Dict[str, Any]: 一个简单的元数据提取函数。 在实际应用中这里可以用更复杂的逻辑或另一个LLM调用来分析对话提取实体、类型等。 此处我们做一个简单的规则模拟。 metadata {source: conversation} # 简单关键词匹配来确定记忆类型实际项目应更智能 type_keywords { user_preference: [喜欢, 偏好, 讨厌, 希望, 想要, 觉得], project_detail: [项目, 任务, deadline, 截止, 进度, 需求], fact: [是, 有, 在, 包括], # 很宽泛 } input_lower user_input.lower() for mem_type, keywords in type_keywords.items(): if any(keyword in input_lower for keyword in keywords): metadata[type] mem_type break else: metadata[type] general # 可以尝试提取项目名简单演示 if project in metadata.get(type, ): # 非常简单的提取实际应用需要NLP words user_input.split() for i, word in enumerate(words): if word.lower() in [project, 项目, 任务] and i1 len(words): metadata[project_name] words[i1].strip(.,!?\) break return metadata def chat_round(self, user_input: str) - str: 处理一轮用户输入并返回助手的回复。 这是核心交互循环。 # 步骤1在回复前先根据用户输入检索相关记忆 relevant_memories self.memory_layer.search_memories( queryuser_input, n_results3 # 每次携带3条最相关的记忆避免上下文过长 ) # 步骤2格式化记忆并构建完整的系统提示词 formatted_mems self._format_memories_for_prompt(relevant_memories) system_prompt self.base_system_prompt.format(formatted_memoriesformatted_mems) # 步骤3调用 Claude API 获取回复 try: message self.client.messages.create( modelclaude-3-haiku-20240307, # 使用成本较低的 Haiku 模型进行演示 max_tokens500, systemsystem_prompt, messages[ {role: user, content: user_input} ] ) assistant_response message.content[0].text except Exception as e: assistant_response f调用 Claude API 时出错{e} return assistant_response # 步骤4判断当前对话是否值得作为长期记忆保存 # 这里是一个简单的启发式规则如果用户输入包含明确的事实陈述、偏好或任务信息则保存。 should_save any(keyword in user_input.lower() for keyword in [ 记住, 我喜欢, 我讨厌, 项目需要, 截止日期是, 目标是 ]) if should_save: # 提取元数据并保存记忆 metadata self._extract_metadata_from_conversation(user_input, assistant_response) # 我们将用户输入作为记忆内容保存。更复杂的场景可以保存整个对话轮次或提炼后的要点。 self.memory_layer.save_memory(contentuser_input, metadatametadata) print(f[系统] 已将本次对话的关键信息存入长期记忆。) return assistant_response def run_conversation(self): 运行一个简单的对话循环。 print( * 50) print(个人项目助手已启用长期记忆已启动。) print(输入 退出 或 quit 结束对话。) print(输入 查看记忆 可以查看所有存储的记忆。) print( * 50) while True: try: user_input input(\n你: ).strip() except (EOFError, KeyboardInterrupt): print(\n再见) break if user_input.lower() in [退出, quit, exit]: print(助手: 再见期待下次继续为您管理项目。) break if user_input.lower() 查看记忆: memories self.memory_layer.get_all_memories(limit10) print(\n 存储的所有记忆最近10条) for mem in memories: print(f- ID: {mem[id]}) print(f 内容: {mem[content]}) print(f 元数据: {mem[metadata]}) print() continue # 正常对话处理 response self.chat_round(user_input) print(f助手: {response})关键逻辑解析__init__ 初始化 Claude 客户端和记忆层。base_system_prompt包含一个占位符{formatted_memories}用于在每次调用前插入检索到的相关记忆。chat_round 单轮对话的核心方法。检索 首先用用户输入作为查询去记忆层搜索相关记忆。构建上下文 将记忆格式化后填入系统提示词形成 Claude 本次调用的完整“背景”。调用模型 使用包含记忆的上下文调用 Claude API。选择性保存 根据简单规则可扩展为更复杂的逻辑或另一个 LLM 判断决定是否将本轮对话保存为新记忆。这模拟了 Agent 的“学习”过程。run_conversation 提供一个简单的命令行交互界面。6. 运行与效果验证现在让我们将整个项目运行起来看看记忆层如何工作。6.1 配置环境变量在项目根目录创建.env文件填入你的 Claude API 密钥。# .env 文件内容 CLAUDE_API_KEY你的-claude-api-key-here重要 请勿将.env文件提交到版本控制系统如 Git。确保它在.gitignore中。6.2 创建主程序入口创建main.py# main.py from agent_core import ClaudeAgentWithMemory def main(): agent ClaudeAgentWithMemory() agent.run_conversation() if __name__ __main__: main()6.3 启动对话在终端运行python main.py6.4 交互示例与验证让我们模拟一个跨越多次对话的连续任务场景个人项目助手已启用长期记忆已启动。 输入 退出 或 quit 结束对话。 输入 查看记忆 可以查看所有存储的记忆。 你: 你好请帮我记住我正在进行的项目叫“星辰计划”下周一是原型设计的截止日期。 助手: 好的我已经记下了。您的项目“星辰计划”的原型设计截止日期是下周一。我会在相关时间提醒您。 你: 另外关于这个项目我比较喜欢用蓝色的主题色。 助手: 明白已将“星辰计划”的主题色偏好蓝色记录下来。 你: 查看记忆 存储的所有记忆最近10条 - ID: mem_20231027_143022_123456 内容: 你好请帮我记住我正在进行的项目叫“星辰计划”下周一是原型设计的截止日期。 元数据: {source: conversation, type: project_detail, project_name: 星辰计划, timestamp: 2023-10-27T14:30:22.123456} - ID: mem_20231027_143045_654321 内容: 另外关于这个项目我比较喜欢用蓝色的主题色。 元数据: {source: conversation, type: user_preference, project_name: 星辰计划, timestamp: 2023-10-27T14:30:45.654321} 你: “星辰计划”的截止日是什么时候我用的主题色是什么 助手: 根据我们的对话记录 1. [project_detail] 你好请帮我记住我正在进行的项目叫“星辰计划”下周一是原型设计的截止日期。 2. [user_preference] 另外关于这个项目我比较喜欢用蓝色的主题色。 因此“星辰计划”的原型设计截止日期是下周一您为该项目选择的主题色是蓝色。效果验证点记忆保存 当用户提供明确信息项目名、截止日、偏好时Agent 识别并保存了记忆控制台有提示且可通过“查看记忆”命令确认。记忆检索 当用户后续询问“星辰计划”的细节时Agent 在回复中直接引用了之前保存的两条具体记忆而不是泛泛而谈。这证明记忆层成功检索到了相关信息。连续性体现 即使在同一个会话中如果没有记忆层后一个问题也需要在 Prompt 中包含整个历史对话。而我们的实现仅通过检索到的2条关键记忆就实现了精准回复大大节省了上下文令牌。7. 常见问题与排查思路在实际集成 Mnemara 或自建记忆层时你可能会遇到以下问题问题现象可能原因排查方式解决方案调用save_memory失败连接错误ChromaDB 持久化目录权限问题嵌入模型下载失败。1. 检查persist_directory路径是否存在且可写。2. 查看控制台初始化日志检查sentence-transformers模型是否下载成功。1. 确保程序对目标目录有读写权限。2. 对于模型下载可以尝试设置代理或使用离线模型文件。搜索记忆时返回结果不相关1. 查询文本与记忆内容语义差异大。2. 嵌入模型不适合当前领域。3. 元数据过滤条件太严格。1. 打印出search_memories返回的原始结果和相似度分数。2. 检查记忆保存时的内容是否清晰、完整。3. 尝试不使用过滤器进行搜索。1. 优化记忆的保存内容使其更具概括性。2. 尝试更换更适合的嵌入模型如all-mpnet-base-v2效果更好但更慢。3. 调整检索策略结合关键词匹配进行混合检索。Claude 回复未利用记忆1. 记忆检索失败或为空。2. 系统提示词中记忆的格式化方式不佳模型“看不到”或“不理解”。3. 检索到的记忆过多淹没了当前指令。1. 在chat_round中打印relevant_memories和formatted_mems。2. 检查最终发送给 Claude 的完整 Prompt可临时打印。1. 确保记忆检索逻辑正确查询文本有代表性。2. 优化_format_memories_for_prompt函数让记忆的呈现更清晰如编号、加标签。3. 减少n_results只保留最相关的1-3条记忆。记忆保存过于频繁存储膨胀保存记忆的触发条件should_save太宽松。查看存储的记忆列表判断哪些是不必要保存的。设计更精细的记忆保存策略。例如1. 只保存用户明确指令“请记住”的内容。2. 使用另一个轻量级LLM或规则判断信息是否具有长期价值。3. 定期清理过时或低访问频率的记忆。程序重启后Agent“失忆”记忆层未正确配置持久化。检查 ChromaDB 的PersistentClient路径是否正确以及程序重启后是否加载了同一个集合。确保MemoryLayer初始化时使用的persist_directory路径一致。ChromaDB 会从该目录加载已有数据。8. 最佳实践与工程建议将记忆层投入生产环境需要考虑更多工程细节记忆的粒度与结构避免保存原始对话 不要简单保存“用户说X助手说Y”。应该提炼出核心事实、决策或偏好。可以考虑用一个小型LLM如 Claude Haiku对对话进行摘要再将摘要保存为记忆。设计元数据模式 提前规划好metadata的字段如type(fact, preference, goal, action),entity(项目名、人名),priority,expiry_date等。良好的元数据是高效检索的基础。检索策略优化混合检索 结合向量检索语义相似和基于元数据的过滤精确匹配。例如先过滤type“user_preference” AND entity“ProjectA”再在结果中进行向量检索。查询重写 在检索前可以用LLM将用户的自然语言问题重写为更利于检索的关键词或陈述句。记忆评分与衰减 为记忆设计一个“重要性”或“新鲜度”分数。频繁被检索到的记忆得分更高久未使用的记忆得分衰减。检索时按综合分排序。成本与性能权衡控制上下文长度 严格限制注入到 Prompt 中的记忆条数和总令牌数。这是控制成本最直接的手段。异步记忆处理 保存记忆生成嵌入向量和更新索引可能是耗时操作应将其放入后台任务队列避免阻塞主对话流程。缓存热点记忆 对于高频使用的记忆如用户姓名、基础偏好可以缓存在应用内存中避免每次检索都查询向量库。隐私与安全数据隔离 确保不同用户、不同租户的记忆数据完全隔离。可以在 ChromaDB 中为每个用户/会话创建独立的集合或在元数据中加入user_id并严格过滤。记忆审查与删除 提供用户查看和删除其个人记忆的接口满足数据合规要求如 GDPR。敏感信息过滤 在保存记忆前可以增加一个过滤层检测并脱敏密码、密钥、个人身份信息等敏感内容。与现有 Agent 框架集成LangChain LangChain 有成熟的内存模块ConversationBufferMemory,ConversationSummaryMemory,VectorStoreRetrieverMemory。我们的MemoryLayer可以视为一个自定义的BaseMemory实现或与VectorStoreRetriever结合。LlamaIndex LlamaIndex 本身就是处理索引和检索的专家。你可以用 LlamaIndex 来构建和管理记忆的索引然后将其查询接口封装到你的记忆层中。通过本文的实战你已经掌握了为 Claude Agent 构建记忆层的核心逻辑。从简单的向量存储检索到复杂的记忆生命周期管理这其中的每一步都决定了你的 Agent 是“金鱼”还是“顾问”。真正的挑战不在于实现存储而在于设计一套让记忆产生最大价值的策略——记住什么、何时记住、如何想起。