我之前写过两篇Agent相关的内容一篇聊Agent的基本盘怎么搭一篇聊Multi-Agent。这次我们聊一个更贴近日常使用体验的点——记忆。不管你是做客服机器人、个人知识助手还是企业内部智能体只要Agent脱离“一句话问答”往“长期陪伴”方向走记忆就是绕不开的那道坎。AI Agent说了这么多真正拉开体验差距的其实就是记忆。同样一个助手有记忆的版本记得你讨厌香菜、知道你上次聊到哪、记得你项目的技术栈没记忆的版本每次从零开始像极了每天上班都要重新做自我介绍的新同事。这篇文章我不讲那些虚的框架概念直接拆一个可落地的记忆系统怎么分层、怎么存、怎么取、怎么避免“记错”和“乱记”最后用LangGraph和一个开放协议把代码跑通。这篇文章适合正在做Agent应用开发的工程师或者准备系统学习AI Agent的读者。如果你刚接触Agent前面几段能帮你建立整体认知如果你已经在写业务代码后面实现部分可以直接抄作业。1. 为什么Agent必须“记住你”——先拆解核心需求1.1 Agent的记忆到底在记什么我见过不少团队做Agent第一版就是“大模型提示词知识库”用户问什么答什么。跑起来看着没问题但用两个月就发现用户流失严重。原因很简单用户觉得这个机器人“没感情”昨天刚说过的事情今天又问你一遍谁受得了。如果把Agent比作一个店员短期记忆就是他在当次对话里记住你点了什么菜工作记忆是他手头正在处理的订单长期记忆则是他记得你是老顾客、口味偏辣、上次来投诉过上菜慢。对照到系统设计上就是三件事短期记忆当前会话的上下文一般就是messages数组直接塞进模型上下文窗口。工作记忆任务执行过程中的中间状态比如你让它做一个调研它查到第几步、收集了哪些素材。长期记忆用户画像、历史偏好、跨会话的事实这是持久化存储也是最值得做文章的部分。大部分团队短期记忆都处理得好长期记忆才是真正的分水岭。原因在于短期记忆架构简单上下文窗口装得下就行而长期记忆牵扯到你怎么抽取、怎么存、怎么在合适的时机取回来整个一套链路下来工程复杂度完全不在一个量级。1.2 为什么“把聊天记录全塞进上下文”不靠谱那你可能会说既然大模型上下文窗口越来越大我把所有历史聊天记录都丢给它不就行了我刚开始也这么干过后来被现实教育了。第一是成本。所有主流大模型都按Token计费历史记录越堆越长每一轮对话都在重复烧钱。一个用户聊了100轮你每次调模型都要把100轮的历史重新发给大模型这成本不是线性增长是复利增长。第二是效果。上下文太长之后模型对中间部分的注意力会明显下降这就是业内常说的Lost in the middle——你塞进去的旧信息不仅帮不上忙反而把当前真正的意图给淹没掉。第三是噪音。用户早期说的“我想买个手机”和他现在问的“手机到了怎么退货”完全不是一回事你把“想买手机”的语境一直挂在系统提示词里不干扰才怪。所以长期记忆的正确打开方式是“按需取用”不把所有历史背在身上而是先把值得记的东西抽出来存好等用户再次问到相关内容时再把对应记忆片段取出来注入到当前对话里。这就是检索增强生成的基本思路也是几乎所有生产级Agent记忆系统的共同底座。2. 记忆系统怎么设计——三层记忆架构与选型思路2.1 记忆分层短期、中期、长期怎么切分我在实际项目里习惯把记忆分成三层来设计不搞花里胡哨就按生命周期和用途来切。短期记忆最简单每个会话独立维护一个上下文消息列表会话结束就清理。中期记忆稍微复杂一点它要跨越同一用户的多次会话但不需要永久保存比如用户正在进行的购物流程、还没提交的表单数据、聊天中临时说到的“下周去出差”。这些信息在一两周内有价值过期就没意义了。长期记忆则是用户的稳定画像和长期事实比如职业、常驻城市、技术栈偏好、对某些话题的态度这些几乎不会变变了一次之后也要长期生效。这三层对应到存储方案上也不一样。短期记忆直接放内存或者Redis中期记忆可以放KV存储或者轻量数据库带个过期时间长期记忆才需要进向量数据库走语义检索。很多新手一上来就把所有记忆全部向量化存进向量库其实没必要——很多临时状态你压根不需要花成本做向量化直接一个键值对存JSON就搞定了。记忆层级生命周期存储方案读取方式典型内容短期记忆单次会话内存/Redis直接拼接对话历史、临时状态中期记忆数天到数周Redis/KV存储键值查询表单草稿、临时任务长期记忆持久保存向量数据库语义检索用户画像、历史偏好2.2 向量数据库 RAG的组合逻辑长期记忆为什么必须用向量检索而不是用传统的关键词匹配我举一个特别常见的例子。用户第一次说“我平时写Python比较多不太会Java。”你把这句存进记忆库。过了两周用户问“帮我推荐一个适合后端开发的学习路线。”他完全没有提Python或者Java但一个合格的Agent应该记得他的技术栈。关键词匹配在这里直接失效因为查询语句和记忆片段之间没有任何共同的分词词项。而向量化之后存储的句子和当前的查询都会被映射到同一个语义空间里它们的向量距离会很近检索系统就能把那条记忆捞出来。整个RAG记忆链路就是四步用户输入进来之后先向量化成嵌入向量然后到向量库里去检索最相似的Top-K个记忆片段接着把检索结果按照时间、相关性、重要程度排序最后拼装成一个记忆块注入到系统提示词里。这里有一个细节很多人会忽略注入的记忆块必须明确标注是哪一轮对话提取的、关于哪个实体的这样模型才能区分“用户正在说的”和“用户曾经说过的”。2.3 嵌入模型与向量库选型嵌入模型的选择取决于你的数据语言和业务场景。中文为主的场景我建议优先试bge-m3或者m3e这类对中文支持更好的模型英文场景用OpenAI的text-embedding-3-small就足够。不是越大的模型越好嵌入模型要跟你的向量库维度、存储成本、推理延迟一起考虑。比如text-embedding-3-small输出1536维bge-m3输出1024维维度越高通常代表信息量越大但存储和计算开销也大。小项目用量低没什么感觉量大了之后维度直接影响检索延迟。向量库这一层的选择我的经验是可以按规模分档。个人项目或者短期内数据量不超过10万条直接用Chroma就行轻量、本地跑、Python接口友好。生产环境要扛并发和规模再考虑Qdrant、Milvus或者云上的Pinecone。还有一条路是直接用你已有的PostgreSQL装上pgvector插件也一样能用对团队来说少引入一个组件省了不少运维成本。我见过不少团队为了一个还没跑起来的临时项目专门部署一套Milvus集群结果运维成本比Agent本身还高没必要。3. 实操给Agent装上可落地的记忆系统3.1 用LangGraph搭建记忆工作流理论讲再多不如直接看代码。下面我用LangGraph演示一个带记忆的Agent工作流。为什么选LangGraph因为它把Agent推理过程拆成了节点和边每个节点只干一件事状态在节点之间流动这种结构特别适合记忆这种横跨多个环节的组件。先定义一个状态结构。这里面既要有当前对话的messages也要有从记忆库检索出来的memory_results以及待写入的新记忆队列。from typing import TypedDict, List, Dict, Any from langchain_core.messages import BaseMessage class AgentState(TypedDict): messages: List[BaseMessage] memory_results: List[Dict[str, Any]] memory_to_store: List[Dict[str, Any]]然后构建Agent流程我按三个节点来分retrieve_memory负责在用户提问进来时先去记忆库检索assistant_node负责让模型根据当前消息和历史记忆生成回复store_memory负责在每轮对话结束之后判断有没有值得写入长期记忆的新信息。from langgraph.graph import StateGraph, START, END def retrieve_memory(state: AgentState): # 把当前用户最新一条消息向量化去记忆库检索 query state[messages][-1].content results memory_store.search(query, top_k5) return {memory_results: results} def assistant_node(state: AgentState): # 将检索到的记忆注入系统提示词 memory_block format_memory_block(state[memory_results]) messages build_prompt(state[messages], memory_block) response chat_model.invoke(messages) return {messages: [response]} def store_memory(state: AgentState): # 抽取本轮值得记忆的新信息 new_facts extract_memories(state[messages]) return {memory_to_store: new_facts} graph_builder StateGraph(AgentState) graph_builder.add_node(retrieve, retrieve_memory) graph_builder.add_node(assistant, assistant_node) graph_builder.add_node(store, store_memory) graph_builder.add_edge(START, retrieve) graph_builder.add_edge(retrieve, assistant) graph_builder.add_edge(assistant, store) graph_builder.add_edge(store, END) agent_graph graph_builder.compile()这个流程看起来简单但涵盖了记忆系统最核心的“先读后写”逻辑。调用Agent之前先读记忆回复之后再把新信息写回去。这两步顺序不能反否则Agent第一轮对话就是“失忆”状态。3.2 记忆写入与提取的完整实现记忆写入是整个系统里最容易糊弄也最容易出错的地方。很多人会把整段对话历史一股脑全存进向量库这样做有两个问题一是存了大量无意义的内容检索时噪音太大二是没有做结构化取出来之后模型不好用。我的做法是先用一个小模型对每一轮对话做信息抽取抽出来的是结构化的事实列表再写入记忆库。比如用户说“我在上海工作平时喜欢喝美式咖啡”抽取出来就是两条事实location上海、coffee_preference美式。这样存下来后续检索到的就是干净、明确的事实而不是一堆口水话。from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import Chroma from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser extract_prompt ChatPromptTemplate.from_messages([ (system, 你是记忆抽取器。从对话中抽取关于用户的可记忆事实只输出JSON数组。 事实类型包括偏好(preference)、身份信息(identity)、长期目标(goal)、禁忌(avoidance)。 如果一句话没有任何可记忆信息输出空数组。 规则只抽取确定的、长期有效的信息不抽取临时情绪或随口闲聊。), (human, {input}) ]) def extract_memories(messages): # 只取最近一轮用户消息做抽取 latest_user_msg [m for m in messages if m.type human][-1].content chain extract_prompt | chat_model | StrOutputParser() raw chain.invoke({input: latest_user_msg}) # 解析JSON返回事实列表 return parse_fact_list(raw)记忆写入之后检索时也有一些技巧。除了基础的相似度搜索我还会在存储时给每条记忆加上时间戳和实体标签。检索时优先返回近期记忆和与当前对话实体相关的记忆再按相似度排序。这样一个简单的加权策略能明显提高记忆使用的准确率。def recall_memory(query: str, top_k: int 5): query_embedding embed_model.embed_query(query) results vector_store.similarity_search_by_vector( query_embedding, ktop_k ) # 按时间衰减和相关性过滤 filtered [ r for r in results if r.metadata.get(score, 1.0) 0.75 ] return filtered这个相似度阈值0.75是我在多轮测试里调出来的平衡点。设置太低会把大量不相关信息当成记忆注入进去设置太高又可能漏掉真正有用的记忆。不同嵌入模型的输出范围不一样这个值需要实际测几轮再定我建议你上线前用一批真实对话样本跑一遍看看检出来的记忆到底准不准。3.3 用MCP把记忆做成可复用服务如果你只是做一个单机小Agent上面那套已经够用。但一旦你要做Multi-Agent——比如客服Agent、推荐Agent、运营Agent同时服务同一个用户——那记忆就必须独立出来做成一个所有Agent共享的服务。这时候MCP协议就能派上用场。MCP的全称是Model Context Protocol它本质上规定了Agent怎么调用外部工具、怎么读取外部数据。把记忆封装成一个MCP Server之后不管前端接的是Claude还是其他任何支持MCP的客户端都能通过一套标准协议来读写记忆Agent和各人不再各自维护一套记忆库数据也不打架。我以一个轻量级记忆服务为例把核心代码贴出来。它的职责只有两件事存一段记忆、查相关记忆。from mcp.server.fastmcp import FastMCP mcp FastMCP(MemoryServer) mcp.tool() def store_memory(user_id: str, content: str, metadata: dict None): 存储一条关于用户的事实记忆 doc_id f{user_id}_{timestamp()} vector_store.add_texts( texts[content], metadatas[{user_id: user_id, time: timestamp(), **metadata}], ids[doc_id] ) return {status: ok, memory_id: doc_id} mcp.tool() def query_memory(user_id: str, query: str, top_k: int 5): 查询某个用户的相关历史记忆 results vector_store.similarity_search( query, ktop_k, filter{user_id: user_id} ) return [{content: r.page_content, metadata: r.metadata} for r in results] mcp.run()封装完之后Agent只需要通过MCP客户端去调用这两个工具就能实现记忆的读写。好处是你后续换记忆实现——从Chroma换成Qdrant或者从本地存储换成云服务——Agent那端的代码一行都不用动。这就是协议标准化的价值。4. 踩坑记录记忆系统常见问题与排查4.1 记忆污染旧记忆干扰新对话记忆系统上线之后我最先遇到的坑是“记忆污染”。典型场景是用户曾经说过“我最近在学Python准备转行做开发”但过了半年他早就不学Python了结果每次对话Agent都在提“你上次学Python学得怎么样”用户体验极其糟糕。这个问题的根源在于写入时没有区分“临时状态”和“长期事实”检索时也没有做时间衰减。我的解决方案是双管齐下。写入侧抽取事实时要求模型只抽稳定的、不会有短期变化的信息凡是带时间限定词的内容——比如“最近”“现在”“这周”——一律不写入长期记忆。读取侧给每条记忆打上时间戳检索结果在排序时对超过30天的记忆做时间衰减加权让近期记忆靠前。4.2 存储膨胀与检索失效记忆系统跑久了第二个问题是存储膨胀。用户聊得越多记忆片段越多向量库里攒了几万条最后检索结果全部乱套——不同时期的记忆内容互相矛盾相似度分数全部挤在高位区间区分度下降召回的东西越来越不准。这时候需要做记忆整理。我目前维护的Agent每天晚上会跑一个离线任务把同一个用户的记忆按实体聚类相近的合并、矛盾的去重、超过180天没被命中的冷记忆归档。这相当于定期给你的记忆库“断舍离”虽然会消耗一些算力但换来的检索质量提升非常明显。4.3 多Agent场景下的记忆一致性最后说一个Multi-Agent场景下特有的问题。多个Agent共同读写同一个记忆库很容易出现数据不一致。比如推荐Agent记下了用户“喜欢喝美式咖啡”但客服Agent在处理用户投诉时又记下“用户说这家店咖啡太苦了”。两条记忆同时存在后续Agent读取的时候不知道该信哪条。我的方案是给每条记忆增加一个版本号写入时带上来源Agent标识和时间戳读取时如果有冲突先比较时间后比较来源——客服对话中产生的用户情绪反馈优先于日常偏好。同时重要用户的记忆变更会进入一个人工审核的中间态确保高价值的记忆不被低质量写入污染。这套机制不复杂但能省掉后续大量排查的时间。写在后面的一点体会把Agent记忆系统从头到尾搭完我最大的体会是记忆系统的难点不在算法而在“什么时候该记、什么时候该取”的判断上。你不需要掌握多高深的机器学习知识但需要对业务场景有足够深的理解。我见过不少团队一上来就上大模型向量库的豪华配置结果记了一堆没用的东西关键信息反而检索不出来。另一个值得说的经验是不要一上来就做全套先用最简单的方式把“对话摘要存库关键词检索”跑通再逐步升级到结构化抽取和向量检索。你会在迭代过程中清楚看到每一步优化带来的实际收益而不是对着一个黑盒系统瞎调参数。做记忆系统最怕的就是“做完了但不知道它有没有用”给自己设定一些可观察的指标比如记忆命中率、用户二次提问率比什么都重要。