为Claude Agent构建Mnemara内存层：解决大模型多轮对话失忆问题

在实际开发基于 Claude 或类似大语言模型的智能体Agent应用时一个核心的工程挑战是如何让 Agent 在多次对话或执行过程中“记住”上下文。很多开发者会发现一个看似聪明的 Agent 在回答完一个问题后面对后续的提问时可能表现得像第一次对话一样完全忘记了之前的交互历史、用户偏好或任务状态。这种“失忆”问题严重制约了 Agent 在复杂、多轮任务中的实用性。Mnemara 正是为了解决这一问题而提出的概念——一个专为保持 Claude Agent 连续性而设计的内存层。本文将深入探讨如何为 Claude Agent 构建一个有效的内存层。我们将从理解 Agent 连续性问题的根源开始逐步设计一个内存层的核心组件包括短期记忆、长期记忆和记忆检索机制。然后我们会通过一个具体的代码示例演示如何利用 Claude 的 API 和简单的后端服务来实现 Mnemara 的核心功能。最后我们将讨论在生产环境中部署此类内存层时需要考虑的性能、安全性和扩展性问题。无论你是正在构建客服机器人、个人助理还是复杂的任务自动化流程理解并实现一个可靠的内存层都是提升 Agent 智能水平的关键一步。1. 理解 Agent 的“失忆症”与内存层的必要性1.1 为什么原生对话模型会“失忆”以 Claude 为代表的大语言模型在单次 API 调用中其“记忆”完全依赖于我们提供的上下文Context Window。这个上下文通常以消息列表如[system, user, assistant, user...]的形式传入。模型会根据这个列表生成下一个回复。一旦这次调用结束模型内部并不会主动保存这次对话的状态。下一次调用时如果你不把历史消息再次放入上下文模型就无从得知之前的对话内容。这种设计带来了几个关键限制上下文长度限制所有主流模型都有固定的最大上下文长度如 128K tokens。长对话会迅速耗尽这个额度导致最早的历史被“挤出”上下文。成本与延迟每次调用都携带冗长的历史消息会显著增加输入的 token 数量从而提升 API 调用成本和请求延迟。状态丢失在复杂的多步骤任务中例如“帮我订一张机票然后选靠窗的座位最后预订机场接送”Agent 需要维护任务状态如航班号、座位偏好。仅靠原始对话历史来维护状态既低效又不可靠。1.2 内存层Memory Layer的核心职责一个专门的内存层如 Mnemara旨在外部化地管理 Agent 的状态和记忆使其突破单次调用的限制。它的核心职责包括记忆持久化将重要的对话片段、用户信息、任务状态等结构化地存储到外部数据库或缓存中。记忆检索在需要时根据当前对话的查询从海量记忆中快速、准确地找到最相关的信息并注入到本次调用的上下文中。记忆抽象与压缩不是简单存储每一句对话而是能够总结、提炼关键信息例如将一段关于用户喜好的长对话压缩成“用户偏好靠窗座位、素食、下午航班”这样的结构化标签。会话与状态管理区分不同用户、不同会话并维护每个会话的独立状态机。1.3 Mnemara 的架构设想Mnemara 不应是一个单一的库而是一个位于 Agent 逻辑与 Claude API 之间的服务层。其简化架构如下[用户/系统] - [Agent 逻辑] - [Mnemara 内存层] - [Claude API] |--- [记忆存储数据库/向量库]Agent 逻辑在调用 Claude API 前会先咨询 Mnemara“关于当前用户和这个话题有什么相关的记忆” Mnemara 从存储中检索并返回相关记忆片段。Agent 将这些片段作为系统提示System Prompt或上下文的一部分发给 Claude。Claude 回复后Agent 再决定将哪些新信息交由 Mnemara 存储起来。2. 构建 Mnemara 内存层的核心组件我们将使用 Python 和一个简单的键值数据库如 Redis和向量数据库如 Chroma来演示核心组件的实现。生产环境可能需要更 robust 的方案。2.1 环境准备与依赖配置首先确保你的开发环境已就绪。我们将需要以下工具和库Python 3.9Anthropic Claude API Key用于调用 Claude 模型。Redis用于存储键值型记忆如会话状态、用户属性。Chroma DB一个轻量级向量数据库用于存储和检索基于语义的记忆片段。必要的 Python 包创建一个新的项目目录并初始化虚拟环境mkdir mnemara-agent cd mnemara-agent python -m venv venv # Windows: venv\Scripts\activate # macOS/Linux: source venv/bin/activate然后安装依赖。创建一个requirements.txt文件anthropic0.25.0 redis5.0.0 chromadb0.4.22 openai1.12.0 # 用于 embeddings也可以用其他库 pydantic2.0.0 python-dotenv1.0.0执行安装pip install -r requirements.txt同时确保 Redis 服务正在本地运行默认端口 6379。对于 Chroma它通常以客户端库形式运行数据可持久化到磁盘。2.2 定义记忆数据结构记忆不是简单的字符串。我们需要结构化的数据来表示一条记忆。在项目根目录创建memory_models.pyfrom pydantic import BaseModel, Field from datetime import datetime from typing import Optional, Dict, Any from enum import Enum class MemoryType(str, Enum): 记忆类型枚举 FACT fact # 事实性信息如“用户叫张三” PREFERENCE preference # 用户偏好如“喜欢靠窗座位” TASK_STATE task_state # 任务状态如“机票预订进行中” CONVERSATION_SUMMARY conversation_summary # 对话摘要 SYSTEM system # 系统级记忆 class MemoryEntity(BaseModel): 记忆实体核心数据结构 id: str Field(default_factorylambda: str(uuid.uuid4())) user_id: str # 关联的用户ID session_id: Optional[str] None # 关联的会话IDNone表示跨会话记忆 content: str # 记忆内容文本 memory_type: MemoryType # 记忆类型 embedding: Optional[List[float]] None # 内容的向量嵌入用于语义检索 metadata: Dict[str, Any] Field(default_factorydict) # 扩展元数据如来源、置信度 created_at: datetime Field(default_factorydatetime.utcnow) last_accessed_at: Optional[datetime] None access_count: int 0 class Config: arbitrary_types_allowed True这个MemoryEntity模型定义了记忆的完整信息它属于谁user_id、在什么场景下session_id、是什么内容content、属于哪一类memory_type以及用于检索的向量embedding。metadata字段提供了灵活性可以存储自定义信息。2.3 实现记忆存储后端接下来我们实现两个存储后端RedisStore用于存储键值类和状态信息VectorMemoryStore用于存储和检索基于语义的记忆片段。创建storage.pyimport json import redis import chromadb from chromadb.config import Settings from typing import List, Optional from memory_models import MemoryEntity, MemoryType class RedisStore: 处理会话状态、用户属性和快速键值查询 def __init__(self, hostlocalhost, port6379, db0): self.client redis.Redis(hosthost, portport, dbdb, decode_responsesTrue) def save_session_state(self, session_id: str, state: dict): key fsession:{session_id}:state self.client.set(key, json.dumps(state)) def get_session_state(self, session_id: str) - dict: key fsession:{session_id}:state data self.client.get(key) return json.loads(data) if data else {} def save_user_attribute(self, user_id: str, attribute: str, value: str): key fuser:{user_id}:attr:{attribute} self.client.set(key, value) def get_user_attribute(self, user_id: str, attribute: str) - Optional[str]: key fuser:{user_id}:attr:{attribute} return self.client.get(key) class VectorMemoryStore: 处理基于语义的记忆存储与检索 def __init__(self, persist_directory: str ./chroma_db): # Chroma 客户端数据持久化到本地目录 self.client chromadb.PersistentClient(pathpersist_directory) # 获取或创建一个集合Collection类似于数据库的表 self.collection self.client.get_or_create_collection( nameagent_memories, metadata{hnsw:space: cosine} # 使用余弦相似度进行检索 ) # 在实际项目中你需要一个 Embedding 模型来生成向量。 # 这里我们假设有一个 get_embedding 函数后续实现。 from embedding_utils import get_embedding self.embedding_fn get_embedding def save_memory(self, memory: MemoryEntity): 保存一条记忆到向量数据库 # 生成内容向量 embedding self.embedding_fn(memory.content) memory.embedding embedding # 准备存储到 Chroma 的格式 self.collection.add( documents[memory.content], embeddings[embedding], metadatas[{ user_id: memory.user_id, session_id: memory.session_id or , memory_type: memory.memory_type.value, created_at: memory.created_at.isoformat(), id: memory.id }], ids[memory.id] ) def search_memories(self, query: str, user_id: str, limit: int 5, memory_types: Optional[List[MemoryType]] None) - List[MemoryEntity]: 根据查询文本检索对应用户最相关的记忆 # 生成查询文本的向量 query_embedding self.embedding_fn(query) # 构建过滤条件 where_filter {user_id: user_id} if memory_types: where_filter[memory_type] {$in: [mt.value for mt in memory_types]} # 执行相似性搜索 results self.collection.query( query_embeddings[query_embedding], n_resultslimit, wherewhere_filter, include[documents, metadatas, distances] ) # 将结果转换回 MemoryEntity 对象列表 memories [] if results[ids][0]: for doc, meta, dist in zip(results[documents][0], results[metadatas][0], results[distances][0]): memory MemoryEntity( idmeta[id], user_idmeta[user_id], session_idmeta[session_id] if meta[session_id] else None, contentdoc, memory_typeMemoryType(meta[memory_type]), metadata{similarity_distance: dist} # 将相似度距离存入元数据 ) memories.append(memory) return memories这里我们创建了两个存储类。RedisStore用于快速存取结构简单的状态数据。VectorMemoryStore利用 Chroma DB 实现语义搜索它根据记忆内容的向量表示来查找相似信息这是实现“相关性检索”的核心。2.4 实现 Embedding 生成函数向量检索的核心是将文本转换为数值向量Embedding。我们需要一个 Embedding 模型。创建embedding_utils.pyimport os from openai import OpenAI from typing import List # 初始化 OpenAI 客户端用于 Embedding API也可以用其他开源模型如 sentence-transformers # 注意虽然我们使用 Claude 做对话但 Embedding 可以使用其他性价比更高的模型。 client OpenAI(api_keyos.getenv(OPENAI_API_KEY)) # 需要设置 OPENAI_API_KEY EMBEDDING_MODEL text-embedding-3-small def get_embedding(text: str) - List[float]: 获取单段文本的 embedding 向量 if not text.strip(): return [0.0] * 1536 # 默认维度根据模型调整 text text.replace(\n, ) response client.embeddings.create(input[text], modelEMBEDDING_MODEL) return response.data[0].embedding def get_embeddings(texts: List[str]) - List[List[float]]: 批量获取 embedding 向量效率更高 texts [t.replace(\n, ) for t in texts if t.strip()] if not texts: return [] response client.embeddings.create(inputtexts, modelEMBEDDING_MODEL) return [item.embedding for item in response.data]注意Embedding 调用会产生额外成本。在生产环境中可以考虑缓存常见文本的 Embedding或使用本地部署的开源 Embedding 模型如all-MiniLM-L6-v2来降低成本和控制延迟。3. 实现 Mnemara 内存层服务与 Agent 集成现在我们将存储组件组合成一个完整的记忆层服务并演示如何将其集成到一个简单的 Claude Agent 中。3.1 创建 Mnemara 核心服务创建mnemara_core.pyimport uuid from typing import List, Optional from memory_models import MemoryEntity, MemoryType from storage import RedisStore, VectorMemoryStore class Mnemara: 内存层核心服务 def __init__(self): self.redis_store RedisStore() self.vector_store VectorMemoryStore() def create_session(self, user_id: str) - str: 为用户创建一个新的会话返回会话ID session_id fsess_{uuid.uuid4().hex[:8]} initial_state {user_id: user_id, created_at: datetime.utcnow().isoformat()} self.redis_store.save_session_state(session_id, initial_state) return session_id def add_memory(self, memory: MemoryEntity): 添加一条记忆 # 首先保存到向量存储用于语义检索 self.vector_store.save_memory(memory) # 如果是特定类型的记忆也可能在 Redis 中存一份快照以便快速访问 if memory.memory_type MemoryType.TASK_STATE: self.redis_store.save_session_state( memory.session_id, {task_state: memory.content, **memory.metadata} ) def retrieve_relevant_memories(self, query: str, user_id: str, session_id: Optional[str] None, limit: int 5) - List[MemoryEntity]: 检索与当前查询相关的记忆 # 1. 从向量存储进行语义检索 semantic_memories self.vector_store.search_memories(query, user_id, limitlimit) # 2. 从 Redis 获取当前会话的即时状态如果有 contextual_memories [] if session_id: state self.redis_store.get_session_state(session_id) if state.get(task_state): contextual_memories.append( MemoryEntity( user_iduser_id, session_idsession_id, contentf当前任务状态: {state[task_state]}, memory_typeMemoryType.TASK_STATE, metadata{source: redis_state} ) ) # 合并并去重简单基于内容去重 all_memories semantic_memories contextual_memories seen set() unique_memories [] for mem in all_memories: if mem.content not in seen: seen.add(mem.content) unique_memories.append(mem) return unique_memories[:limit] # 确保不超过限制 def summarize_and_compress(self, conversation_history: List[dict], user_id: str, session_id: str): 总结并压缩一段对话历史形成长期记忆。 这是一个高级功能可以调用 Claude 的 API 来生成摘要。 此处仅展示框架。 # 将对话历史拼接成文本 history_text \n.join([f{msg[role]}: {msg[content]} for msg in conversation_history[-10:]]) # 取最后10轮 # 构建提示词让 Claude 进行摘要 summary_prompt f 请将以下对话历史总结成几条关键的用户事实、偏好或决策。 输出格式为简短的要点列表。 对话历史 {history_text} 关键要点 # 这里需要调用 Claude API 获取摘要代码略 # summary claude_client.complete(summary_prompt) # 然后将摘要作为一条 CONVERSATION_SUMMARY 类型的记忆保存 # self.add_memory(MemoryEntity(...)) pass这个Mnemara类封装了记忆层的核心操作创建会话、添加记忆、检索记忆以及高级的摘要功能。3.2 构建一个具有记忆能力的 Claude Agent现在我们创建一个使用 Mnemara 的 Agent。创建claude_agent.pyimport os from anthropic import Anthropic from mnemara_core import Mnemara, MemoryEntity, MemoryType from datetime import datetime class ClaudeAgentWithMemory: def __init__(self, api_key: str): self.claude Anthropic(api_keyapi_key) self.mnemara Mnemara() # 简单的内存用于维护当前对话的临时上下文 self.conversation_context {} def chat(self, user_id: str, session_id: str, user_input: str) - str: 核心聊天方法处理用户输入并返回 Agent 回复 # 1. 检索相关记忆 relevant_memories self.mnemara.retrieve_relevant_memories( queryuser_input, user_iduser_id, session_idsession_id, limit3 # 限制记忆数量避免上下文过长 ) # 2. 构建系统提示注入相关记忆 memory_context if relevant_memories: memory_context 以下是你已知的关于用户和当前对话的相关信息\n for i, mem in enumerate(relevant_memories, 1): memory_context f{i}. [{mem.memory_type.value}] {mem.content}\n memory_context \n请参考以上信息进行回复。\n system_prompt f你是一个有帮助的助手。{memory_context} 请保持对话的连贯性并利用已知信息提供更精准的帮助。 # 3. 获取或初始化当前会话的对话历史简易版生产环境需持久化 context_key f{user_id}:{session_id} if context_key not in self.conversation_context: self.conversation_context[context_key] [] message_history self.conversation_context[context_key] # 4. 构建发送给 Claude 的消息列表 messages [] # 添加历史消息最后若干轮避免超出上下文 for msg in message_history[-6:]: # 保留最近3轮对话6条消息 messages.append(msg) # 添加当前用户输入 messages.append({role: user, content: user_input}) # 5. 调用 Claude API try: response self.claude.messages.create( modelclaude-3-haiku-20240307, # 可根据需要选择模型 max_tokens1000, systemsystem_prompt, messagesmessages ) assistant_reply response.content[0].text except Exception as e: assistant_reply f抱歉处理您的请求时出现错误{str(e)} # 6. 更新对话历史 message_history.append({role: user, content: user_input}) message_history.append({role: assistant, content: assistant_reply}) # 7. 决定是否将本轮交互中的重要信息存入长期记忆 self._evaluate_and_store_memory(user_id, session_id, user_input, assistant_reply) return assistant_reply def _evaluate_and_store_memory(self, user_id: str, session_id: str, user_input: str, assistant_reply: str): 一个简单的规则如果用户明确表达了偏好或事实则存储它 # 这是一个非常简单的启发式规则。生产环境应该更复杂可能使用另一个 LLM 调用来判断。 memory_triggers [我喜欢, 我讨厌, 我习惯, 我的名字是, 我住在, 我想要一个] for trigger in memory_triggers: if trigger in user_input: # 判断为偏好或事实类信息 memory MemoryEntity( user_iduser_id, session_idsession_id, contentf用户提到{user_input}, memory_typeMemoryType.PREFERENCE if 喜欢 in trigger or 习惯 in trigger else MemoryType.FACT ) self.mnemara.add_memory(memory) print(f[Mnemara] 已存储记忆{memory.content[:50]}...) break # 如果检测到任务状态变化例如用户确认了某个步骤也可以存储 TASK_STATE # 这里省略具体逻辑这个 Agent 在每次对话时会先通过Mnemara检索与当前用户输入相关的记忆并将这些记忆作为系统提示的一部分发送给 Claude。这样Claude 在生成回复时就能“记得”之前的重要信息。同时Agent 会在对话后尝试提取关键信息通过简单规则保存到记忆层中。3.3 运行一个完整的示例创建一个main.py来演示整个流程import os from dotenv import load_dotenv from claude_agent import ClaudeAgentWithMemory # 加载环境变量其中应有 ANTHROPIC_API_KEY 和 OPENAI_API_KEY load_dotenv() def main(): api_key os.getenv(ANTHROPIC_API_KEY) if not api_key: print(错误请设置 ANTHROPIC_API_KEY 环境变量。) return # 初始化 Agent agent ClaudeAgentWithMemory(api_keyapi_key) # 模拟用户和会话 user_id user_123 session_id agent.mnemara.create_session(user_id) print(f新会话已创建: {session_id}) # 模拟多轮对话 conversations [ 你好我的名字叫李雷。, 我喜欢吃披萨尤其是海鲜口味的。, 我讨厌等待希望事情都能高效完成。, 你还记得我喜欢吃什么吗, 我的名字是什么 ] for user_input in conversations: print(f\n[用户] {user_input}) reply agent.chat(user_id, session_id, user_input) print(f[助手] {reply}) if __name__ __main__: main()运行这个脚本前请确保在项目根目录创建.env文件并填入你的 API 密钥ANTHROPIC_API_KEY你的_Claude_API_Key OPENAI_API_KEY你的_OpenAI_API_Key # 用于 Embedding然后运行python main.py预期你会看到类似以下的输出新会话已创建: sess_a1b2c3d4 [用户] 你好我的名字叫李雷。 [助手] 你好李雷很高兴认识你。 [Mnemara] 已存储记忆用户提到你好我的名字叫李雷。... [用户] 我喜欢吃披萨尤其是海鲜口味的。 [助手] 海鲜披萨是个不错的选择很多地方都有卖。 [Mnemara] 已存储记忆用户提到我喜欢吃披萨尤其是海鲜口味的。... [用户] 我讨厌等待希望事情都能高效完成。 [助手] 理解效率很重要。我会尽力快速准确地回应你。 [Mnemara] 已存储记忆用户提到我讨厌等待希望事情都能高效完成。... [用户] 你还记得我喜欢吃什么吗 [助手] 是的根据我们的对话记录你提到过你喜欢吃披萨尤其是海鲜口味的。 [用户] 我的名字是什么 [助手] 你之前告诉过我你的名字是李雷。可以看到在最后两轮即使对话历史中没有直接包含前几轮的全部内容Agent 依然能正确回答因为它从 Mnemara 内存层检索到了相关的记忆“喜欢吃海鲜披萨”和“名字叫李雷”并注入到了系统提示中。4. 生产环境考量、常见问题与最佳实践上述示例是一个简化版本。要将 Mnemara 这样的内存层投入生产需要考虑更多因素。4.1 性能与可扩展性组件潜在瓶颈优化建议向量检索记忆条目过多时检索延迟高。1. 对记忆进行分片按用户、时间。2. 使用更高效的向量索引如 HNSW。3. 考虑使用专业的向量数据库如 Pinecone, Weaviate, Qdrant。Embedding 生成调用外部 API 产生延迟和成本。1. 缓存已生成的 Embedding。2. 使用本地轻量级 Embedding 模型如all-MiniLM-L6-v2。3. 批量处理文本生成 Embedding。记忆存储非结构化记忆增长过快。1. 定期清理低访问频率、低重要性的记忆。2. 实现记忆的“摘要-细节”两级存储长期只保留摘要。Agent 逻辑每次对话都检索记忆增加延迟。1. 实现记忆缓存短期内相同查询直接返回缓存结果。2. 异步执行记忆存储操作不阻塞主回复流程。4.2 记忆的隐私、安全与合规数据隔离确保记忆严格按user_id隔离。向量检索的过滤条件必须准确防止串户。敏感信息处理在存储前可能需要对用户输入进行敏感信息检测和脱敏如手机号、邮箱。记忆遗忘权提供 API 让用户查询、修改或删除自己的记忆以满足 GDPR 等法规要求。存储加密敏感的记忆内容在数据库层应进行加密存储。4.3 高级记忆策略记忆重要性评分不是所有对话都值得记忆。可以训练一个分类器或设计规则为每条潜在记忆打分只存储高分项。打分依据可以包括信息明确度、情感强度、是否涉及用户属性等。记忆衰减与合并旧的、长时间未访问的记忆应逐渐“衰减”降低检索优先级或最终删除。相似的新记忆可以合并到旧记忆中而不是重复存储。主动记忆触发除了被动存储Agent 可以在检测到信息矛盾或模糊时主动询问用户以澄清从而获得高质量记忆。例如“您刚才说喜欢咖啡但之前提过喜欢茶请问您的偏好是否有变化”4.4 常见问题排查问题现象可能原因检查与解决步骤Agent 似乎“想不起”刚说过的事。1. 记忆未成功存储。2. 检索时user_id或session_id不匹配。3. Embedding 模型不一致或质量差。1. 检查add_memory方法是否被调用且无异常。2. 核对存储和检索时使用的 ID。3. 检查向量数据库中对应记忆的元数据是否正确。4. 测试 Embedding 相似度确保相关文本能被检索到。响应速度变慢。1. 记忆条目过多检索慢。2. Embedding API 调用延迟高。3. 上下文过长导致 Claude API 响应慢。1. 限制单次检索数量limit。2. 为向量数据库建立优化索引。3. 考虑使用本地 Embedding 模型。4. 压缩或摘要长记忆后再注入上下文。存储了无关或错误的记忆。记忆提取规则_evaluate_and_store_memory过于简单或错误。1. 优化记忆触发规则加入更精确的模式匹配或意图识别。2. 引入基于 LLM 的记忆重要性判断步骤在存储前进行过滤。不同用户记忆混淆。向量检索的过滤条件where未正确应用或失效。1. 在search_memories方法中打印或日志记录生成的过滤条件。2. 直接查询向量数据库验证返回结果是否包含其他用户的数据。4.5 部署与监控建议服务化将 Mnemara 封装成独立的 gRPC 或 RESTful 服务供多个 Agent 实例调用。监控指标记忆存储/检索延迟P95 P99。记忆存储成功率。向量数据库索引大小和内存使用情况。记忆命中率检索到的记忆中有多少被实际用于生成回复。回滚机制记忆层的逻辑错误可能导致 Agent 行为异常。确保有快照和回滚能力并能清除特定时间段内写入的“坏记忆”。构建一个像 Mnemara 这样的内存层是开发现实世界智能体的关键一步。它超越了简单的对话历史拼接通过结构化的存储和语义检索为 Agent 提供了真正意义上的持续性和上下文感知能力。从简单的键值对存储到复杂的向量检索从基于规则的记忆提取到基于模型的记忆重要性评估内存层的设计是一个可以根据应用复杂度逐步深化的领域。开始实践时可以从本文提供的简化版本出发优先解决 Agent 的“短期失忆”问题然后随着业务需求的增长逐步引入更高级的记忆管理策略和基础设施优化。