老周虾扯：AI 记忆机制工程实践

前言上一篇我们理论讲完了现在动手实践实践。毕竟“纸上得来终觉浅绝知此事要躬行。”本文聚焦工程落地从最简单的 Buffer Memory 到生产级的向量记忆、图记忆一步步带你把 AI 记忆真正跑起来。所有代码均可直接运行。一、最简单的起点ConversationBufferMemory1.1 原理把完整的对话历史直接塞进 Prompt 的{history}占位符。fromlangchain_openaiimportOpenAIfromlangchain.chainsimportConversationChainfromlangchain.memoryimportConversationBufferMemory llmOpenAI(temperature0,openai_api_keyYOUR_KEY)conversationConversationChain(llmllm,memoryConversationBufferMemory())# 第一轮conversation.predict(input我叫张三我是一名 Python 工程师。)# 第二轮conversation.predict(input我叫什么名字)# 输出你叫张三。✅1.2 问题Token 爆炸随着对话增长历史记录越来越长Token 消耗线性增长第1轮~50 tokens 第10轮~500 tokens 第100轮~5000 tokens ← 开始超出 Context Window解决方案引入记忆压缩策略。二、记忆压缩策略2.1 ConversationBufferWindowMemory滑动窗口只保留最近 K 轮对话fromlangchain.memoryimportConversationBufferWindowMemory memoryConversationBufferWindowMemory(k5)# 只保留最近5轮conversationConversationChain(llmllm,memorymemory)优点Token 消耗可控缺点早期重要信息会丢失2.2 ConversationSummaryMemory摘要压缩用 LLM 对历史对话进行摘要用摘要替代原始历史fromlangchain.memoryimportConversationSummaryMemory memoryConversationSummaryMemory(llmllm)conversationConversationChain(llmllm,memorymemory)# 内部会自动调用 LLM 生成摘要# 用户张三是 Python 工程师对 LLM 集成感兴趣...优点长对话也能保持低 Token 消耗缺点摘要过程有额外 LLM 调用成本细节可能丢失2.3 ConversationSummaryBufferMemory混合策略结合滑动窗口和摘要近期对话保留原文远期对话压缩为摘要。fromlangchain.memoryimportConversationSummaryBufferMemory memoryConversationSummaryBufferMemory(llmllm,max_token_limit2000# 超过2000 token 时触发摘要)三、向量记忆语义检索长期记忆3.1 架构设计用户输入 ↓ [向量化] → 查询向量数据库 → 检索相关记忆 ↓ 注入 Prompt → LLM 生成回复 ↓ [异步] 提取新记忆 → 向量化 → 存入向量数据库3.2 使用 Chroma 实现本地向量记忆fromlangchain_openaiimportOpenAIEmbeddings,ChatOpenAIfromlangchain_chromaimportChromafromlangchain.memoryimportVectorStoreRetrieverMemoryfromlangchain.chainsimportConversationChainfromlangchain.promptsimportPromptTemplate# 初始化向量数据库embeddingsOpenAIEmbeddings()vectorstoreChroma(collection_nameagent_memory,embedding_functionembeddings,persist_directory./chroma_db)# 创建记忆检索器retrievervectorstore.as_retriever(search_kwargs{k:3})memoryVectorStoreRetrieverMemory(retrieverretriever)# 预存一些记忆memory.save_context({input:我最喜欢的编程语言是 Python},{output:好的我记住了})memory.save_context({input:我在做一个 RAG 项目},{output:了解RAG 是检索增强生成})# 使用记忆promptPromptTemplate(input_variables[history,input],template 以下是与用户的相关历史记忆 {history} 当前对话 用户{input} AI)chainConversationChain(llmChatOpenAI(),memorymemory,promptprompt)# 测试语义检索resultchain.predict(input推荐一些适合我项目的工具)# AI 会检索到RAG 项目相关记忆给出针对性建议3.3 使用 Pinecone 实现云端向量记忆frompineconeimportPineconefromlangchain_pineconeimportPineconeVectorStore pcPinecone(api_keyYOUR_PINECONE_KEY)indexpc.Index(agent-memory)vectorstorePineconeVectorStore(indexindex,embeddingOpenAIEmbeddings(),namespaceuser_123# 按用户隔离)# 存储记忆vectorstore.add_texts(texts[用户偏好 Python正在做 RAG 项目],metadatas[{user_id:123,timestamp:2026-03-16,type:preference}])# 检索记忆resultsvectorstore.similarity_search(用户的技术栈是什么,k3,filter{user_id:123})四、LangGraph 长期记忆生产级方案4.1 LangGraph Memory StoreLangGraph 提供了内置的 Memory Store支持跨会话持久化fromlanggraph.store.memoryimportInMemoryStorefromlanggraph.graphimportStateGraph,MessagesStatefromlangchain_openaiimportChatOpenAIfromlangchain_core.messagesimportSystemMessageimportuuid# 初始化存储生产环境用 PostgresStore 或 RedisStorestoreInMemoryStore()modelChatOpenAI(modelgpt-4o)defcall_model(state:MessagesState,config,*,store):user_idconfig[configurable][user_id]namespace(memories,user_id)# 检索用户记忆memoriesstore.search(namespace)info\n.join([m.value[data]forminmemories])system_msgf你是一个有记忆的 AI 助手。 关于用户的已知信息{info}根据以上信息个性化你的回复。responsemodel.invoke([SystemMessage(contentsystem_msg)]state[messages])# 提取并保存新记忆new_memoryextract_memory(state[messages][-1].content)ifnew_memory:store.put(namespace,str(uuid.uuid4()),{data:new_memory})return{messages:response}defextract_memory(text:str)-str:用 LLM 提取值得记忆的信息extraction_promptf 从以下用户消息中提取值得长期记忆的关键信息偏好、事实、背景。 如果没有值得记忆的信息返回空字符串。 用户消息{text}提取的记忆简洁格式resultmodel.invoke(extraction_prompt)returnresult.content.strip()# 构建 GraphbuilderStateGraph(MessagesState)builder.add_node(model,call_model)builder.set_entry_point(model)builder.set_finish_point(model)graphbuilder.compile(storestore)# 使用config{configurable:{user_id:user_123,thread_id:thread_1}}# 第一次对话graph.invoke({messages:[{role:user,content:我叫李四我是一名架构师}]},configconfig)# 新会话记忆依然存在config2{configurable:{user_id:user_123,thread_id:thread_2}}resultgraph.invoke({messages:[{role:user,content:你还记得我是做什么的吗}]},configconfig2)# 输出你是一名架构师叫李四。✅五、Mem0专业记忆引擎5.1 快速上手pipinstallmem0aifrommem0importMemoryimportos os.environ[OPENAI_API_KEY]YOUR_KEYos.environ[MEM0_API_KEY]YOUR_MEM0_KEY# 使用托管版memoryMemory()# 添加记忆memory.add(messages[{role:user,content:我是一名后端工程师专注于 Java 和 Spring Boot},{role:assistant,content:了解我会记住你的技术背景}],user_iduser_001)# 搜索记忆resultsmemory.search(query用户的技术栈,user_iduser_001)forrinresults:print(r[memory])# 输出用户是后端工程师专注于 Java 和 Spring Boot5.2 多层记忆管理# 会话级记忆短期memory.add(messages[{role:user,content:今天我在调试一个 OOM 问题}],user_iduser_001,session_idsession_20260316# 会话结束后可清除)# 用户级记忆长期memory.add(messages[{role:user,content:我最近在学习 Rust}],user_iduser_001# 无 session_id 长期保存)# 获取所有记忆all_memoriesmemory.get_all(user_iduser_001)forminall_memories:print(f[{m[created_at]}]{m[memory]})5.3 与 LangChain 集成frommem0importMemoryClientfromlangchain_openaiimportChatOpenAIfromlangchain_core.messagesimportSystemMessage,HumanMessage clientMemoryClient(api_keyYOUR_MEM0_KEY)llmChatOpenAI(modelgpt-4o)defchat_with_memory(user_id:str,user_message:str)-str:# 1. 检索相关记忆memoriesclient.search(user_message,user_iduser_id)memory_context\n.join([m[memory]forminmemories])# 2. 构建带记忆的 Promptsystem_promptf你是一个有记忆的 AI 助手。 用户相关记忆{memory_contextifmemory_contextelse暂无历史记忆}请根据以上记忆个性化回复。# 3. 调用 LLMresponsellm.invoke([SystemMessage(contentsystem_prompt),HumanMessage(contentuser_message)])# 4. 异步保存新记忆client.add(messages[{role:user,content:user_message},{role:assistant,content:response.content}],user_iduser_id)returnresponse.content# 测试print(chat_with_memory(user_001,我叫王五我在做微服务架构设计))print(chat_with_memory(user_001,给我推荐一些适合我工作的工具))六、Zep图记忆与上下文工程6.1 Zep 的核心优势Zep 不只是存储记忆而是构建时序上下文图Temporal Context Graph自动提取实体和关系当事实变化时自动失效旧记忆支持 Graph RAG 多跳推理200ms 检索延迟6.2 快速集成pipinstallzep-cloudfromzep_cloud.clientimportAsyncZepfromzep_cloud.typesimportMessageimportasyncio clientAsyncZep(api_keyYOUR_ZEP_KEY)asyncdefmain():# 创建用户awaitclient.user.add(user_iduser_001)# 创建会话awaitclient.memory.add_session(session_idsession_001,user_iduser_001)# 添加对话记忆awaitclient.memory.add(session_idsession_001,messages[Message(roleuser,role_typeuser,content我是一名 DevOps 工程师负责公司的 K8s 集群),Message(roleassistant,role_typeassistant,content了解你负责 Kubernetes 集群运维)])# 检索上下文Zep 自动组装最相关的上下文memoryawaitclient.memory.get(session_idsession_001)print(memory.context)# 输出用户是 DevOps 工程师管理 K8s 集群...# 图查询查找与用户相关的所有实体factsawaitclient.graph.search(user_iduser_001,query用户的技术职责)forfactinfacts.edges:print(f{fact.source}→{fact.relation}→{fact.target})asyncio.run(main())七、生产环境最佳实践7.1 记忆分层架构┌─────────────────────────────────────────────────────┐ │ 生产级记忆架构 │ ├─────────────────────────────────────────────────────┤ │ L1: 会话缓存Redis │ │ - 当前对话历史 │ │ - TTL: 24小时 │ ├─────────────────────────────────────────────────────┤ │ L2: 向量记忆Pinecone/Weaviate │ │ - 语义化长期记忆 │ │ - 按用户 namespace 隔离 │ ├─────────────────────────────────────────────────────┤ │ L3: 图记忆Zep/Neo4j │ │ - 实体关系网络 │ │ - 支持复杂推理查询 │ ├─────────────────────────────────────────────────────┤ │ L4: 结构化存储PostgreSQL │ │ - 用户档案、偏好设置 │ │ - 精确查询场景 │ └─────────────────────────────────────────────────────┘7.2 记忆写入流水线importasynciofromdatetimeimportdatetimeclassMemoryPipeline:def__init__(self,vector_store,graph_store,kv_store):self.vector_storevector_store self.graph_storegraph_store self.kv_storekv_storeasyncdefprocess_conversation(self,user_id:str,messages:list):异步处理对话提取并存储记忆# 并行执行多个记忆提取任务awaitasyncio.gather(self._extract_and_store_facts(user_id,messages),self._update_session_cache(user_id,messages),self._update_graph(user_id,messages))asyncdef_extract_and_store_facts(self,user_id,messages):提取事实性记忆存入向量库factsawaitself._llm_extract_facts(messages)iffacts:awaitself.vector_store.add_texts(textsfacts,metadatas[{user_id:user_id,timestamp:datetime.now().isoformat(),type:fact}]*len(facts))asyncdef_update_session_cache(self,user_id,messages):更新 Redis 会话缓存keyfsession:{user_id}:historyawaitself.kv_store.lpush(key,*[str(m)forminmessages[-5:]])awaitself.kv_store.expire(key,86400)# 24小时过期asyncdef_update_graph(self,user_id,messages):更新知识图谱entitiesawaitself._extract_entities(messages)awaitself.graph_store.upsert_entities(user_id,entities)7.3 隐私保护importhashlibimportredefsanitize_memory(text:str)-str:脱敏处理移除手机号、邮箱、身份证等敏感信息# 手机号脱敏textre.sub(r1[3-9]\d{9},[手机号已脱敏],text)# 邮箱脱敏textre.sub(r\b[A-Za-z0-9._%-][A-Za-z0-9.-]\.[A-Z|a-z]{2,}\b,[邮箱已脱敏],text)# 身份证脱敏textre.sub(r\d{17}[\dXx],[身份证已脱敏],text)returntextdefhash_user_id(user_id:str)-str:对用户 ID 进行哈希避免直接存储明文 IDreturnhashlib.sha256(user_id.encode()).hexdigest()[:16]八、常见踩坑与解决方案问题原因解决方案记忆检索不准确向量化质量差使用更好的 Embedding 模型text-embedding-3-largeToken 超限注入记忆过多限制检索数量top_k3使用 Reranker 过滤记忆冲突用户信息更新添加时间戳优先使用最新记忆跨用户记忆泄露namespace 未隔离严格按 user_id 隔离存储和检索记忆写入延迟同步写入阻塞改为异步后台写入冷启动无记忆新用户首次使用设计引导流程主动收集用户信息九、总结记忆实现路径选择 简单场景 → ConversationBufferMemory / SummaryMemory 中等场景 → VectorStoreRetrieverMemory Chroma/Pinecone 复杂场景 → LangGraph Memory Store Mem0 企业级 → Zep图记忆 上下文工程从 Buffer Memory 到图记忆每一层都有其适用场景。不要过度设计——从最简单的方案开始随着业务复杂度增长逐步升级。