AI Agent记忆系统实战:从分层架构到LangGraph+MCP落地
2026/9/13 1:48:33 网站建设 项目流程

我之前写过两篇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_k=5) 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, k=top_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, k=top_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记忆系统从头到尾搭完,我最大的体会是:记忆系统的难点不在算法,而在“什么时候该记、什么时候该取”的判断上。你不需要掌握多高深的机器学习知识,但需要对业务场景有足够深的理解。我见过不少团队一上来就上大模型+向量库的豪华配置,结果记了一堆没用的东西,关键信息反而检索不出来。另一个值得说的经验是,不要一上来就做全套,先用最简单的方式把“对话摘要存库+关键词检索”跑通,再逐步升级到结构化抽取和向量检索。你会在迭代过程中清楚看到每一步优化带来的实际收益,而不是对着一个黑盒系统瞎调参数。做记忆系统,最怕的就是“做完了但不知道它有没有用”,给自己设定一些可观察的指标,比如记忆命中率、用户二次提问率,比什么都重要。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询