在 Agent 开发里有一个问题几乎每个团队都会撞上对话一长就乱换一个会话就“失忆”。你问它还记得你昨天说过什么吗它只会礼貌地道歉然后重新问你一遍基本信息。很多人第一反应是“把历史记录都塞进上下文”结果上下文窗口爆炸、Token 成本飙升、回答质量反而下降。更值得思考的判断是记忆系统不是把更多东西检索出来而是让 Agent 学会“回忆”。这也是 RippleMem 这类新方案给大家带来的核心启发。检索只是召回候选集回忆才是理解、筛选、取舍后的结果。这篇文章会围绕 LangChain LangGraph DeepAgents 三件套拆出一套可以落到企业项目的 Agent 记忆系统。先讲清楚三者到底各自负责什么再把短期记忆、长期记忆分别用最小可运行的代码实现出来最后用一个电商导购 Agent 案例把“记得住上一句、也记得住上个月”的记忆链路完整跑通。读完你可以直接照着搭原型再把数据库、向量库、权限体系替换成你们公司已有的基础设施。1. 记忆系统为什么它是 Agent 从“玩具”走向“生产力”的分水岭先说一个很现实的场景。假设你正在做一个电商客服 Agent用户第一次进来说“我身高 175喜欢偏宽松的黑色卫衣”。Agent 当时回答得很好推荐了三件商品。第二天这个用户又来了说“再给我推荐一件卫衣”。如果 Agent 完全忘记昨天的偏好它大概率会重新问一遍身高和风格。用户会觉得这个 Agent 很蠢甚至直接转人工。这正是当前大量 Agent 项目的真实状态每次对话都像第一次见面。模型本身有很强大的推理能力但没有“跨会话记忆”这个基础设施它的能力就只能停留在单轮问答和单次任务上。你可以让它帮你写邮件、总结会议纪要但很难让它成为一个真正懂你的数字员工。要理解记忆系统的价值先要把记忆做一个简单分类记忆类型通俗解释对应 Agent 实现短期记忆当前这轮对话里说了什么上下文窗口内的消息列表情景记忆上次会话中发生过的具体事件历史会话摘要、关键事件记录语义记忆用户的长期偏好、知识、画像向量库、知识库、用户画像表程序记忆怎么调用工具、按什么流程处理问题Agent 编排逻辑、提示词模板大多数团队做 Agent 时模型自带的上下文窗口天然提供了“短期记忆”但缺少的是后面三种。于是他们开始做 RAG把企业文档切碎后放到向量库里。但很快会发现RAG 解决的是“知识检索”不等于“用户记忆”。知识是公开的、静态的而用户偏好是动态的、属于某个具体用户的。这两套机制需要放在一起设计但不能混为一谈。当一个 Agent 的记忆系统真正成型它应该具备四个能力低成本写入、高质量召回、可控遗忘、可观测。写入不是把每句话都存进去而是抽取值得记的信息召回不是把所有历史都翻出来而是找当前最相关的片段遗忘不是删库而是对过期信息做降权和清理。这四个能力就是全文要一步步落到代码里的东西。电商场景只是载体这套设计同样可以迁移到金融客服、HR 助手、销售 Copilot 等任何需要“懂人”的场景。2. 基于 LangChain LangGraph 的 Agent 记忆系统架构分层很多初学者会把 LangChain 和 LangGraph 搞混甚至以为 LangGraph 是 LangChain 的替代品。这里先说结论LangChain 是工具箱LangGraph 是流水线DeepAgents 是流水线上的作业规范。三者解决的不是同一个问题而是三个层次的问题。2.1 LangChain 和 LangGraph 到底有什么区别LangChain 解决的是“用什么能力干活”它封装了大模型接入、工具调用、向量存储、文档加载、输出解析这些组件。比如你想让 Agent 能查数据库LangChain 会帮你统一封装模型调用和工具格式。LangGraph 解决的是“这些能力怎么编排”它把一次 Agent 任务建模成一张有向图每个节点是一个函数节点之间通过状态传递数据并且支持循环、分支、人工介入、持久化。两者关系可以用一个类比理解LangChain 像是厨房里的各种厨具LangGraph 则是后厨的工作流程——洗菜、切菜、炒菜、装盘各是一个步骤步骤之间要传递食材和半成品。没有 LangGraph你也能用 LangChain 做一次调用但一旦流程变复杂你需要自己手写状态管理很容易失控。| 对比维度 | LangChain | LangGraph | | --- | --- | --- | | 定位 | 组件与工具层 | 状态化编排层 | | 核心抽象 | Chain、Tool、Retriever、Memory | StateGraph、Node、State、Checkpointer | | 适合场景 | 单次调用、线性流程、快速验证 | 多分支、循环、人工确认、复杂状态流转 | | 对记忆的支持 | 提供消息历史组件 | 状态本身就是记忆的载体天然支持持久化 | | 学习成本 | 较低 | 需要理解图、状态、归约器这些概念 |实际项目里两者是配合关系。你通过 LangChain 封装模型和工具然后放到 LangGraph 的节点里执行LangGraph 负责把每一步的状态串联起来。很多同学学 LangChain 入门后发现做复杂 Agent 时还是在“写 if-else”核心原因就是没有引入状态编排而 LangGraph 正是解决这个问题的关键一环。2.2 记忆系统的三层架构回到记忆系统这个主题我建议把设计拆成三层每层对应一个技术组件第一层是能力层由 LangChain 提供。负责记忆系统的“读写能力”大模型调用、向量嵌入、向量库检索、消息对象封装。短期记忆在代码层面就是消息列表长期记忆在代码层面就是向量库或用户画像表这些基础操作都由 LangChain 组件完成。第二层是状态层由 LangGraph 提供。负责记忆系统的“流转逻辑”当前会话的消息存放在 State 的哪个字段、长期记忆检索完之后放在哪里、记忆更新在什么时机执行。LangGraph 的最大价值在于State 可以持久化到外部存储所以“跨会话记忆”本质上变成了“恢复上一次的 State”。第三层是行为层由 DeepAgents 这一类 Agent 框架提供。负责记忆系统的“使用策略”什么时候应该检索长期记忆、检索结果如何参与规划、工具调用结束后如何反思、需要把哪些信息写回记忆。DeepAgents 是近期讨论度很高的 Agent 开发框架它把研究、规划、报告这类高层能力封装成了可复用行为官方文档迭代速度很快。在记忆系统里它解决的是“怎么用记忆来思考和决策”这一层的问题而不是替代 LangGraph 的状态管理。这三层在代码里并不是三个独立的包而是环形协作关系。状态层是核心行为层调用状态层读写记忆能力层为两层提供模型和存储能力。如果你团队里已经用了 DeepAgents可以把它的规划结果落到 LangGraph 的 State 中再用 LangChain 的工具去执行如果你们暂时不引入 DeepAgents用 LangGraph 原生节点也能实现同样的记忆闭环。记住一个原则先把状态层设计好行为层只是状态层之上的策略。3. 环境准备与前置条件在动手写代码之前先把环境准备好。这套示例代码不依赖重型服务本地一台开发机就能跑通。3.1 运行环境与 Python 版本建议使用 Python 3.10 及以上版本3.9 也可以运行但部分新版本依赖可能会提示不支持。Windows、macOS、Linux 均可唯一的差异在于向量库安装包。执行以下命令确认 Python 版本python --version如果还没有安装虚拟环境建议先创建一个独立环境避免依赖冲突python -m venv agent-memory-env source agent-memory-env/bin/activate # Windows 使用 agent-memory-env\Scripts\activate3.2 安装依赖核心依赖是 LangChain、LangChain-OpenAI、LangGraph再加上一个向量库。向量库二选一即可FAISS 在 Linux 和 macOS 上安装方便Chroma 是全平台通用Windows 用户如果装 FAISS 失败可以直接用 Chroma。pip install langchain langchain-openai langgraph pip install faiss-cpu # Linux / macOS # 或者 pip install chromadb # 全平台通用版本说明LangChain 和 LangGraph 都处于快速迭代期不同版本的 API 存在细微差异。以下代码基于当前社区最常用的写法如果你的环境中某些接口提示已废弃以官方文档和最新版本迁移说明为准。3.3 模型服务配置本文示例使用 ChatOpenAI 类你可以填入任意兼容 OpenAI 协议的服务商地址也可以换成本地模型。在项目根目录创建.env文件OPENAI_API_KEY你的_API_Key OPENAI_API_BASEhttps://你的模型服务地址然后在代码中加载环境变量import os from dotenv import load_dotenv load_dotenv() from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, temperature0)如果你的模型服务商不叫 OpenAI只需要替换ChatOpenAI为对应的 LangChain 模型类比如ChatZhipuAI、ChatOllama、ChatTongyi。记忆系统的核心逻辑与具体模型无关这一点是 LangChain 做抽象带来的好处。3.4 项目目录结构建议按模块拆分方便后续扩展到真实项目agent-memory-demo/ ├── .env ├── ecommerce_agent.py # 电商案例主文件 ├── memory/ │ ├── __init__.py │ ├── short_term.py # 短期记忆逻辑 │ └── long_term.py # 长期记忆逻辑 └── data/ └── ...如果只是跑通演示可以暂先不拆分把代码集中在单个文件中便于阅读。但一旦进入团队协作按模块拆分是必须的。4. 短期记忆实现让 Agent 记住“刚才说了什么”短期记忆的实现是所有记忆系统的基础。它的本质是在 LangGraph 的 State 中维护一个消息列表每次对话都把新的用户消息和 AI 回复追加进去。LangGraph 的消息归约器会自动处理追加逻辑这也是为什么add_messages这个字段在 LangGraph 中几乎成了一个约定俗成的标准。4.1 用 LangGraph 定义带短期记忆的 Agent先写一个最小可运行的对话图。这个图只有一个节点节点里直接调用大模型State 中通过Annotated[list, add_messages]来声明消息字段# 文件路径demo_short_term.py from typing import Annotated, TypedDict from dotenv import load_dotenv load_dotenv() from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langchain_openai import ChatOpenAI model ChatOpenAI(modelgpt-4o-mini, temperature0) class ChatState(TypedDict): messages: Annotated[list, add_messages] def chat_node(state: ChatState): response model.invoke(state[messages]) return {messages: [response]} graph StateGraph(ChatState) graph.add_node(chat, chat_node) graph.add_edge(START, chat) graph.add_edge(chat, END) app graph.compile() if __name__ __main__: # 模拟两轮对话 config {configurable: {thread_id: demo_user_001}} result app.invoke({messages: [{role: user, content: 你好我叫小明我喜欢黑色}]}, config) print(result[messages][-1].content) result app.invoke({messages: [{role: user, content: 我刚才说了我叫什么}]}, config) print(result[messages][-1].content)这个例子里LangGraph 会自动把所有历史消息追加到state[messages]中。第二次调用的时候模型能看到第一轮的完整上下文所以它能回答出“你刚才说你叫小明”。这也是短期记忆最简单的形态保留当前会话窗口内的所有消息。4.2 限制短期记忆窗口长度直接保留所有历史消息有一个隐患Token 会随着对话轮数线性增长最终撞上模型上下文窗口上限。企业级场景必须给短期记忆设“预算”。两种常见做法是滑动窗口和消息摘要。滑动窗口比较简单保留最近 N 条消息即可# 文件路径demo_short_term.py 中的新增函数 MAX_MESSAGES 10 def trim_messages_window(messages: list, max_messages: int MAX_MESSAGES): 保留最近 max_messages 条消息避免上下文无限增长。 if len(messages) max_messages: return messages return messages[-max_messages:]然后在chat_node中使用它def chat_node_with_trim(state: ChatState): trimmed_messages trim_messages_window(state[messages]) response model.invoke(trimmed_messages) return {messages: [response]}这个方案实现简单缺点是会“硬性遗忘”掉更早的内容。如果用户在第 20 轮突然提起第 2 轮说过的一个关键信息Agent 就再也找不回来了。所以更优雅的方案是用大模型做摘要当消息超过阈值时把旧消息压缩成一段摘要摘要保留在 State 中之后每次对话都携带摘要 最近 N 条完整消息。4.3 短期记忆摘要压缩# 文件路径demo_summary.py from langchain_core.messages import SystemMessage, HumanMessage def summarize_messages(messages: list) - str: 调用模型把较旧的消息压缩成一段摘要。 text_parts [] for msg in messages: text_parts.append(f{msg.type}: {msg.content}) text \n.join(text_parts) summary_prompt ( 你是记忆系统的一部分。请用简洁的中文摘要以下对话中的关键信息 包括用户身份、偏好、待办事项和重要事实。不要遗漏细节。\n\n f对话内容\n{text} ) response model.invoke([SystemMessage(contentsummary_prompt)]) return response.content实际项目中摘要不会每一轮都重新生成而是当消息数量超过阈值时把最早的一部分消息合并成摘要并用摘要替换掉那部分原始消息。LangGraph 官方也提供了RemoveMessage机制来从 State 中删除旧消息from langchain_core.messages import RemoveMessage # 在节点返回中删除旧消息 return { messages: [summary_message, RemoveMessage(idold_message.id)], }短期记忆做到这个程度已经可以支撑大多数客服、导购类场景。核心认知是短期记忆不是把上下文无限撑大而是用预算思维去管理窗口。该截断的截断该摘要的摘要该让位于当前问题的就让位。5. 长期记忆实现让 Agent 学会“回忆”而不是“全量检索”如果说短期记忆是“保留现场”那么长期记忆就是“跨会话重建现场”。它要解决的核心问题是用户已经离开一段时间今天重新回来Agent 怎么快速回忆起这个用户的画像、偏好和过去发生过的事。5.1 长期记忆的工作流程长期记忆不是简单地把所有历史记录塞进向量库然后检索 top-k 丢给模型。一个合格的长期记忆系统至少包含四个步骤写入从对话中抽取值得长期记忆的信息转成向量或结构化数据后存储。召回在当前对话发生时根据用户 ID 和当前问题检索相关记忆片段。评估对召回结果做相关性判断、去重、时效性过滤。更新根据新对话内容更新用户画像和记忆条目。这里特别强调“评估”这一步因为在真实数据中向量检索出来的结果不一定都对当前问题有用。比如用户今天问“帮我看看物流进度”检索系统可能召回了他三个月前对某件商品质量的抱怨。这时如果原样塞给模型反而会带偏回答。所以正确做法是召回结果只是“候选集”最终是否使用、如何使用要交给大模型做判断。这就是“回忆”和“检索”的本质区别。5.2 用向量库存储长期记忆下面的示例用 FAISS 作为向量存储。这里展示的是语义记忆的写入和召回# 文件路径memory/long_term.py from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import FAISS # 实际项目中建议使用统一的向量库配置 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) # 模拟一个用户的历史记忆片段 initial_memories [ 用户小王男身高175cm偏好宽松版型, 用户喜欢黑色和灰色的卫衣, 用户经常在促销活动期间下单, 用户曾在2024年11月购买过一件XL码的黑色卫衣, ] vectorstore FAISS.from_texts(initial_memories, embeddings) def recall_memory(query: str, top_k: int 3) - list: 根据当前问题召回相关记忆。 retriever vectorstore.as_retriever(search_kwargs{k: top_k}) docs retriever.invoke(query) return [doc.page_content for doc in docs] def add_memory(text: str): 写入一条新的长期记忆。 vectorstore.add_texts([text])调用方式# 测试召回 results recall_memory(用户喜欢什么颜色) for r in results: print(-, r)这个示例中“用户喜欢黑色和灰色的卫衣”最有可能被召回。但注意向量检索是“相似度匹配”不是“语义理解”。如果用户问“他平时穿什么尺码”系统应该召回“身高175cm偏好宽松版型”和“购买过XL码”这两条但对当前问题最关键的可能是“XL码”。所以召回之后必须有一个提炼和筛选的过程。5.3 让模型“回忆”而不是“搬运”下面这个函数体现了“回忆”的完整过程。它把检索结果作为素材结合当前用户问题让模型输出一份精简的、与当前问题相关的用户上下文# 文件路径memory/recall_with_llm.py from langchain_core.messages import SystemMessage, HumanMessage def build_llm_recall(llm, user_question: str, retrieved_docs: list) - str: 让模型从检索结果中提炼与当前问题相关的记忆。 doc_text \n.join([f- {d} for d in retrieved_docs]) prompt ( 你是一个记忆整理助手。下面是从用户历史记忆中检索到的片段 请筛选出与当前问题真正相关的内容并整理成一段简洁的用户上下文。\n 如果某些片段与当前问题无关请忽略它们。\n\n f当前用户问题{user_question}\n\n f检索到的记忆片段\n{doc_text}\n\n 请输出整理后的记忆内容不要输出多余的说明。 ) response llm.invoke([ SystemMessage(content你是一个严谨的记忆整理助手), HumanMessage(contentprompt), ]) return response.content这个函数的意义在于不是把 5 条、10 条检索片段全部堆进上下文而是让 LLM 先把它们“过一遍脑子”去掉噪声提取真正和当前问题有关的记忆。这也是为什么前文说“记忆系统不是把更多东西检索出来而是让 Agent 学会回忆”。检索是回忆的起点而不是终点。5.4 记忆的写入从对话中抽取值得记的信息长期记忆的写入不能把所有用户消息都存下来。设计一个抽取函数让模型判断哪些信息值得存入长期记忆# 文件路径memory/extract_memory.py import json def extract_memory_candidates(llm, user_message: str) - dict: 从用户消息中抽取值得长期记忆的信息没有则返回空字典。 prompt ( 从用户消息中抽取值得长期记住的偏好信息或事实。 只抽取稳定、可复用的信息比如用户身份、风格偏好、尺码、职业、家庭情况。 不要抽取一次性提问内容。\n 输出严格JSON格式例如{\偏好\: \喜欢黑色\, \尺码\: \XL\}没有则输出 {}。\n\n f用户消息{user_message} ) try: response llm.invoke(prompt) content response.content.strip() content content.strip(json).strip().strip() return json.loads(content) if content else {} except Exception: return {}在完整案例里会把这个函数接到 LangGraph 的更新节点实现“对话结束后自动沉淀记忆”的闭环。6. 电商案例一个能记住用户偏好和订单的导购 Agent前面把短期记忆和长期记忆分开讲清楚了这一节把它们组合成一个完整的电商导购 Agent。业务背景用户小王在上一轮会话中说了自己喜欢的颜色、尺码和风格今天重新进线问“再推荐一件卫衣”。Agent 需要自动回忆起小王的历史偏好并结合工具查询订单和商品信息给出个性化回答。6.1 记忆数据结构设计在编码之前先把记忆字段定义清楚。短期记忆就是messages列表长期记忆包括用户画像和向量记忆为了演示方便这里用字典数据结构模拟画像向量库仍然负责语义召回。记忆层存储字段示例内容短期记忆messages当前会话的消息历史长期记忆 - 用户画像user_profileuser_id、会员等级、尺码偏好、颜色偏好长期记忆 - 语义记忆vectorstore“小王喜欢宽松版型”“常在促销期下单”等自然语言片段工具结果last_tool_result最近一次订单/商品查询结果6.2 完整代码可运行的电商导购 Agent环境配置和第二、三节保持一致。这个示例中的工具全部使用模拟数据不真实调用外部 API。生产环境替换为真实接口即可。# 文件路径ecommerce_agent.py import json from typing import Annotated, TypedDict from dotenv import load_dotenv load_dotenv() from langchain_openai import ChatOpenAI from langchain_core.messages import AIMessage, HumanMessage, SystemMessage from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages model ChatOpenAI(modelgpt-4o-mini, temperature0) # ---------- 1. 模拟数据库 ---------- orders_db { user_1001: [ {order_id: A10001, product: 黑色宽松卫衣, status: 已签收, price: 299}, {order_id: A10002, product: 灰色运动裤, status: 运输中, price: 199}, ] } products_db [ {name: 黑色宽松卫衣, style: 宽松, color: 黑色, price: 299, stock: 100}, {name: 灰色连帽卫衣, style: 修身, color: 灰色, price: 359, stock: 80}, {name: 白色基础T恤, style: 标准, color: 白色, price: 99, stock: 200}, {name: 深蓝直筒牛仔裤, style: 直筒, color: 深蓝, price: 249, stock: 50}, ] user_profiles { user_1001: { user_id: user_1001, name: 小王, member_level: 黄金会员, height: 175, preferred_style: 宽松, preferred_colors: [黑色, 灰色], preferred_size: XL, } } # ---------- 2. 工具函数模拟 ---------- def query_orders(user_id: str) - list: return orders_db.get(user_id, []) def query_product(style: str None, color: str None) - list: result products_db if style: result [p for p in result if p[style] style] if color: result [p for p in result if p[color] color] return result # ---------- 3. 长期记忆模块 ---------- def recall_user_profile(user_id: str) - dict: 实际项目中应优先从用户画像数据库读取再结合向量检索。 return user_profiles.get(user_id, {}) def update_user_profile(user_id: str, new_info: dict): 把新抽取的偏好写入用户画像和向量记忆。 profile user_profiles.setdefault(user_id, {user_id: user_id}) profile.update(new_info) # 这里可以调用 vectorstore.add_texts() 写入语义记忆 print(f[Memory Updated] {user_id}: {new_info}) def extract_memory_candidates(user_message: str) - dict: 让模型抽取值得长期记忆的信息。 prompt ( 从用户消息中抽取值得长期记住的偏好信息或事实。 只抽取稳定、可复用的信息比如身份、风格偏好、尺码、颜色偏好。 不要抽取一次性提问内容。\n 输出严格JSON格式例如{\偏好\: \喜欢黑色\, \尺码\: \XL\}没有则输出 {}。\n\n f用户消息{user_message} ) try: response model.invoke(prompt) content response.content.strip().strip(json).strip().strip() return json.loads(content) if content else {} except Exception: return {} # ---------- 4. 图状态定义 ---------- class AgentState(TypedDict): messages: Annotated[list, add_messages] user_profile: dict recall_memory: str last_tool_result: str # ---------- 5. 图节点定义 ---------- def recall_node(state: AgentState): 长期记忆召回根据用户ID读取画像。 user_id state[user_profile].get(user_id) profile recall_user_profile(user_id) memory_text ( f姓名{profile.get(name, 未知)} f会员等级{profile.get(member_level, 未知)} f身高{profile.get(height, 未知)} f偏好风格{profile.get(preferred_style, 未知)} f偏好颜色{profile.get(preferred_colors, 未知)} f偏好尺码{profile.get(preferred_size, 未知)} ) return {recall_memory: memory_text} def agent_node(state: AgentState): Agent 决策节点结合当前问题和长期记忆调用工具输出回复。 last_human_message state[messages][-1].content user_id state[user_profile].get(user_id) # 简单的意图识别实际项目可替换为 DeepAgents 的规划逻辑 tool_result if 订单 in last_human_message or 物流 in last_human_message: orders query_orders(user_id) tool_result json.dumps(orders, ensure_asciiFalse) elif 推荐 in last_human_message or 卫衣 in last_human_message or 商品 in last_human_message: profile state[user_profile] style profile.get(preferred_style) color profile.get(preferred_colors, [None])[0] products query_product(stylestyle, colorcolor) tool_result json.dumps(products, ensure_asciiFalse) else: tool_result 未命中工具调用请直接回答。 sys_prompt SystemMessage(content( 你是电商导购助手。请结合用户画像和工具查询结果用简洁、自然的中文回答用户。\n f用户画像{state[recall_memory]}\n f工具查询结果{tool_result}\n )) user_msg HumanMessage(contentlast_human_message) response model.invoke([sys_prompt, user_msg]) return { messages: [AIMessage(contentresponse.content)], last_tool_result: tool_result, } def update_memory_node(state: AgentState): 记忆更新节点从最后一条用户消息中抽取偏好更新用户画像。 # 找到最近一条用户消息 last_user_message None for msg in reversed(state[messages]): if msg.type human: last_user_message msg.content break if last_user_message: candidates extract_memory_candidates(last_user_message) if candidates: update_user_profile(state[user_profile][user_id], candidates) return {user_profile: state[user_profile]} # ---------- 6. 构建图 ---------- graph StateGraph(AgentState) graph.add_node(recall, recall_node) graph.add_node(agent, agent_node) graph.add_node(update_memory, update_memory_node) graph.add_edge(START, recall) graph.add_edge(recall, agent) graph.add_edge(agent, update_memory) graph.add_edge(update_memory, END) app graph.compile() # ---------- 7. 模拟两轮会话 ---------- def run_session(user_message: str, user_id: str): 模拟一次会话自动触发记忆系统。 config {configurable: {thread_id: user_id}} result app.invoke( { messages: [{role: user, content: user_message}], user_profile: {user_id: user_id}, recall_memory: , last_tool_result: , }, config, ) return result[messages][-1].content if __name__ __main__: # 第一轮用户告知偏好 reply1 run_session(我喜欢黑色和灰色的宽松卫衣给我推荐几件, user_1001) print(AI:, reply1) # 第二轮模拟新会话语境直接追问 reply2 run_session(我今天想再看看卫衣, user_1001) print(AI:, reply2)6.3 代码关键逻辑说明这部分代码虽然简化了真实业务的复杂度但已经包含了企业级记忆系统的完整骨架值得逐段理解。recall_node负责长期记忆召回。它在每次会话开始时执行从用户画像中拼接出记忆文本。注意这里用的是硬编码画像读取真实项目中应该把“记忆文本”换成向量检索结果并且经过前文说的模型“回忆”提炼后再传给下游。agent_node是 Agent 的核心决策节点。它先判断用户意图再调用模拟工具获取结果最后把“用户画像 工具结果 当前问题”一起交给大模型生成回复。这里面体现了一个关键设计回忆内容不是模型自己凭空想出来的而是记忆系统在节点链路上显式注入的。update_memory_node负责记忆更新。它在每轮回答结束后运行用大模型从用户最新消息中抽取值得记住的偏好再写入用户画像。这个节点的存在让长期记忆具有了“自我生长”能力——Agent 用得越久对用户越了解而不是永远停留在初始配置。6.4 预期运行效果运行主文件后输出类似如下AI: 小王你好根据你偏好的宽松版型和黑色系我为你推荐这件黑色宽松卫衣价格299元目前库存充足。需要顺便看看搭配的裤子吗 [Memory Updated] user_1001: {偏好: 黑色和灰色的宽松卫衣} AI: 小王你好上次你说喜欢黑色和灰色的宽松卫衣今天我还是优先给你推荐黑色宽松卫衣和灰色连帽卫衣。第一件偏宽松第二件偏修身你更想看哪一件第二轮的回复明显带上了第一轮会话留下的记忆痕迹。如果你把thread_id更换成一个新用户 IDAgent 就不会知道“小王喜欢黑色和宽松版型”。这个对比能直观验证长期记忆是否生效。7. 运行验证与效果评估代码跑通只是第一步。要判断这套记忆系统在企业项目里是否真的可用需要从三个层面验证。7.1 功能验证三个必测场景第一个场景是短期记忆验证。同一个thread_id下连续提问比如先问“我叫小明”再问“我叫什么”Agent 要能正确回答。这验证的是消息列表在 LangGraph State 中的正常流转。第二个场景是长期记忆验证。新开一个thread_id但使用同一个user_id问“你知道我喜欢什么颜色吗”。如果系统读取了用户画像并正确回答出“黑色和灰色”说明长期记忆读取链路没问题。第三个场景是记忆更新验证。在第一轮告诉 Agent“我现在更喜欢蓝色”第二轮问“我的偏好是”看 Agent 是否更新为蓝色而不是停留在旧的黑色。这可以检验update_memory_node的抽取和写入是否生效。7.2 质量评估指标工程化评估时建议关注四个指标指标定义健康标准建议记忆召回率应该被召回的相关记忆有多少被成功召回越高越好低于 60% 需要检查向量库分块和索引记忆准确率召回结果中真正相关、无冲突的比例低于 70% 说明需要加强 LLM“回忆”筛选环节记忆污染率错误记忆或过期记忆被写入的比例必须低于 5%否则会持续误导 Agent记忆写入延迟一轮对话结束后更新记忆的耗时应控制在 1-2 秒内避免影响用户体验记忆污染率是最容易被忽视的指标。很多团队只关心“能不能搜到”不关心“会不会记错”。一旦一条错误记忆被写入它会在后续每次会话中被不断放大导致 Agent 越来越“固执”。所以写入环节的抽取模型必须保守拿不准的信息宁可不记。7.3 可观测性日志里要有什么生产环境必须能回答一个问题Agent 之所以给出某个回答是因为它看到了哪段记忆。建议在关键节点输出结构化日志[recall] user_iduser_1001 memory_typeuser_profile length42 [agent] toolquery_product hits2 cost_tokens1320 [update_memory] extracted{preferred_colors: [黑色, 灰色]}有了这些日志当用户投诉“Agent 为什么乱推荐”时你可以快速定位是检索问题、模型问题还是记忆更新问题。8. 常见问题与排查方法下面这些坑是 Agent 记忆系统在从 demo 走向生产时最容易遇到的。整理成表格方便对照排查问题现象可能原因排查方式解决方案对话轮数一多就报上下文超限短期记忆窗口无限制查看模型请求中的输入 Token 数增加滑动窗口截断或摘要压缩新会话中 Agent 不记得用户长期记忆没有按 user_id 关联检查 recall_node 中的用户 ID 是否从请求头正确传递统一用户 ID 体系所有会话从 Token 或参数中解析检索结果和当前问题完全无关向量库中数据分块不合理打印召回结果的文本内容优化文本切分策略增加元数据过滤模型回 JSON 时格式不稳定指令不够明确或模型能力不足打印模型原始输出使用结构化输出解析器增加失败重试同一记忆被重复写入update_memory_node 每轮都执行且无去重查看用户画像中的数据是否有重复字段写入前做相似度比较或主键去重多个节点并行写入记忆导致冲突LangGraph 的 Send API 并发更新同一字段查看 State 更新日志记忆更新统一收敛到单个节点串行执行Agent 越来越“固执”忽略用户新偏好旧记忆权重过高新记忆无法覆盖检查用户画像中旧值更新逻辑设置记忆版本号同一字段新值覆盖旧值并记录时间向量库体积增长过快每轮对话都写入大量文本检查写入节点触发的频率设置写入置信度阈值抽取出空字典时不写入这里再强调一个容易忽略的点LangGraph 中Send这类并行 API 适合做任务分发但不要用在对同一用户记忆的并发写入上。记忆更新的状态是强一致的一旦两个节点同时修改user_profile后写入的值可能覆盖前一个节点的关键信息而且这种问题非常难复现。设计原则是记忆写入收敛到单一节点串行执行。9. 企业级落地最佳实践与后续学习路线示例代码跑通以后距离真正的企业级系统还有一段距离。下面这几条实践建议是按照优先级排序的可以对照自家项目逐步补齐。9.1 记忆写入要有置信度门槛不是所有对话都值得写入长期记忆。用户在闲聊时说了一句话不应该进入用户画像。建议在extract_memory_candidates的 prompt 中增加强度要求只有信息足够明确、稳定、可复用时才输出并且可以要求模型输出置信度分数低于阈值的直接丢弃。这是降低记忆污染率最有效的手段。9.2 记忆要有过期与遗忘机制用户的偏好会变化会员等级会调整地址电话更会更换。画像表里每个字段都应该带一个updated_at时间戳并根据不同业务字段设置不同的有效期。比如“收货地址”可能一年就变“尺码偏好”相对稳定可以保留更长时间。过期字段在召回时降权甚至不再写入上下文。遗忘不是系统缺陷而是长期可靠性的保证。9.3 权限、隐私与多租户隔离记忆系统存储的是用户敏感信息这一条必须前置考虑。用户画像数据要加密存储模型调用链路中不要输出手机号、地址等敏感字段会话数据的使用要遵循用户授权和公司合规要求。多租户场景下租户 A 的用户记忆绝不能被租户 B 的 Agent 检索到。最简单的实现方式是向量库和画像表中都带上tenant_id和user_id两个维度检索时强制过滤。9.4 成本控制与降级策略每一次长期记忆召回和每次记忆抽取都会产生额外 Token 消耗。建议做一个成本开关当系统负载高或用户为低价值流量时可以关闭update_memory_node只保留短期记忆和读取已有画像的能力。降级后用户体验略降但系统仍能正常工作。另外摘要压缩不要每轮都做只在消息数达到阈值时触发并且可以做成异步任务。9.5 测试与灰度记忆系统要像数据库迁移一样谨慎。上线前准备几组测试用户新用户、老用户、偏好冲突用户、长期不活跃用户。先在小流量灰度重点观察记忆写入的准确率和用户反馈。一旦发现记忆污染问题要提供“一键清除用户画像”的运营工具这在客服场景中几乎是必备的。9.6 后续学习路线如果你刚接触 LangChain建议先把官方入门教程中的基础组件过一遍如果你已经会写简单的 Chain下一个重点是深入 LangGraph 的 State 和 Checkpointer 机制理解消息归约器add_messages和节点返回值之间的协作关系如果你已经能熟练编排 LangGraph再去看 DeepAgents 这类高层框架理解它们如何把规划、反思、工具调用这些行为封装成可复用的 Agent 能力。对于 DeepAgents一个更稳妥的判断是它还在快速迭代期但值得持续关注。不要等框架完全稳定再学先掌握 LangGraph 的状态化编排思想因为无论上层框架怎么变状态、记忆、工具、反馈循环这些核心概念是稳定不变的。等你能够不看示例代码独立画出“召回记忆 → 决策 → 调用工具 → 更新记忆”的状态图时你对 Agent 记忆系统的理解就已经超过大多数停留在 prompt 拼接阶段的开发者了。建议把这篇文章收藏备用动手把电商示例跑通再对照你们自己的业务数据做一次完整的记忆链路改造。
在 Agent 开发里有一个问题几乎每个团队都会撞上对话一长就乱换一个会话就“失忆”。你问它还记得你昨天说过什么吗它只会礼貌地道歉然后重新问你一遍基本信息。很多人第一反应是“把历史记录都塞进上下文”结果上下文窗口爆炸、Token 成本飙升、回答质量反而下降。更值得思考的判断是记忆系统不是把更多东西检索出来而是让 Agent 学会“回忆”。这也是 RippleMem 这类新方案给大家带来的核心启发。检索只是召回候选集回忆才是理解、筛选、取舍后的结果。这篇文章会围绕 LangChain LangGraph DeepAgents 三件套拆出一套可以落到企业项目的 Agent 记忆系统。先讲清楚三者到底各自负责什么再把短期记忆、长期记忆分别用最小可运行的代码实现出来最后用一个电商导购 Agent 案例把“记得住上一句、也记得住上个月”的记忆链路完整跑通。读完你可以直接照着搭原型再把数据库、向量库、权限体系替换成你们公司已有的基础设施。1. 记忆系统为什么它是 Agent 从“玩具”走向“生产力”的分水岭先说一个很现实的场景。假设你正在做一个电商客服 Agent用户第一次进来说“我身高 175喜欢偏宽松的黑色卫衣”。Agent 当时回答得很好推荐了三件商品。第二天这个用户又来了说“再给我推荐一件卫衣”。如果 Agent 完全忘记昨天的偏好它大概率会重新问一遍身高和风格。用户会觉得这个 Agent 很蠢甚至直接转人工。这正是当前大量 Agent 项目的真实状态每次对话都像第一次见面。模型本身有很强大的推理能力但没有“跨会话记忆”这个基础设施它的能力就只能停留在单轮问答和单次任务上。你可以让它帮你写邮件、总结会议纪要但很难让它成为一个真正懂你的数字员工。要理解记忆系统的价值先要把记忆做一个简单分类记忆类型通俗解释对应 Agent 实现短期记忆当前这轮对话里说了什么上下文窗口内的消息列表情景记忆上次会话中发生过的具体事件历史会话摘要、关键事件记录语义记忆用户的长期偏好、知识、画像向量库、知识库、用户画像表程序记忆怎么调用工具、按什么流程处理问题Agent 编排逻辑、提示词模板大多数团队做 Agent 时模型自带的上下文窗口天然提供了“短期记忆”但缺少的是后面三种。于是他们开始做 RAG把企业文档切碎后放到向量库里。但很快会发现RAG 解决的是“知识检索”不等于“用户记忆”。知识是公开的、静态的而用户偏好是动态的、属于某个具体用户的。这两套机制需要放在一起设计但不能混为一谈。当一个 Agent 的记忆系统真正成型它应该具备四个能力低成本写入、高质量召回、可控遗忘、可观测。写入不是把每句话都存进去而是抽取值得记的信息召回不是把所有历史都翻出来而是找当前最相关的片段遗忘不是删库而是对过期信息做降权和清理。这四个能力就是全文要一步步落到代码里的东西。电商场景只是载体这套设计同样可以迁移到金融客服、HR 助手、销售 Copilot 等任何需要“懂人”的场景。2. 基于 LangChain LangGraph 的 Agent 记忆系统架构分层很多初学者会把 LangChain 和 LangGraph 搞混甚至以为 LangGraph 是 LangChain 的替代品。这里先说结论LangChain 是工具箱LangGraph 是流水线DeepAgents 是流水线上的作业规范。三者解决的不是同一个问题而是三个层次的问题。2.1 LangChain 和 LangGraph 到底有什么区别LangChain 解决的是“用什么能力干活”它封装了大模型接入、工具调用、向量存储、文档加载、输出解析这些组件。比如你想让 Agent 能查数据库LangChain 会帮你统一封装模型调用和工具格式。LangGraph 解决的是“这些能力怎么编排”它把一次 Agent 任务建模成一张有向图每个节点是一个函数节点之间通过状态传递数据并且支持循环、分支、人工介入、持久化。两者关系可以用一个类比理解LangChain 像是厨房里的各种厨具LangGraph 则是后厨的工作流程——洗菜、切菜、炒菜、装盘各是一个步骤步骤之间要传递食材和半成品。没有 LangGraph你也能用 LangChain 做一次调用但一旦流程变复杂你需要自己手写状态管理很容易失控。| 对比维度 | LangChain | LangGraph | | --- | --- | --- | | 定位 | 组件与工具层 | 状态化编排层 | | 核心抽象 | Chain、Tool、Retriever、Memory | StateGraph、Node、State、Checkpointer | | 适合场景 | 单次调用、线性流程、快速验证 | 多分支、循环、人工确认、复杂状态流转 | | 对记忆的支持 | 提供消息历史组件 | 状态本身就是记忆的载体天然支持持久化 | | 学习成本 | 较低 | 需要理解图、状态、归约器这些概念 |实际项目里两者是配合关系。你通过 LangChain 封装模型和工具然后放到 LangGraph 的节点里执行LangGraph 负责把每一步的状态串联起来。很多同学学 LangChain 入门后发现做复杂 Agent 时还是在“写 if-else”核心原因就是没有引入状态编排而 LangGraph 正是解决这个问题的关键一环。2.2 记忆系统的三层架构回到记忆系统这个主题我建议把设计拆成三层每层对应一个技术组件第一层是能力层由 LangChain 提供。负责记忆系统的“读写能力”大模型调用、向量嵌入、向量库检索、消息对象封装。短期记忆在代码层面就是消息列表长期记忆在代码层面就是向量库或用户画像表这些基础操作都由 LangChain 组件完成。第二层是状态层由 LangGraph 提供。负责记忆系统的“流转逻辑”当前会话的消息存放在 State 的哪个字段、长期记忆检索完之后放在哪里、记忆更新在什么时机执行。LangGraph 的最大价值在于State 可以持久化到外部存储所以“跨会话记忆”本质上变成了“恢复上一次的 State”。第三层是行为层由 DeepAgents 这一类 Agent 框架提供。负责记忆系统的“使用策略”什么时候应该检索长期记忆、检索结果如何参与规划、工具调用结束后如何反思、需要把哪些信息写回记忆。DeepAgents 是近期讨论度很高的 Agent 开发框架它把研究、规划、报告这类高层能力封装成了可复用行为官方文档迭代速度很快。在记忆系统里它解决的是“怎么用记忆来思考和决策”这一层的问题而不是替代 LangGraph 的状态管理。这三层在代码里并不是三个独立的包而是环形协作关系。状态层是核心行为层调用状态层读写记忆能力层为两层提供模型和存储能力。如果你团队里已经用了 DeepAgents可以把它的规划结果落到 LangGraph 的 State 中再用 LangChain 的工具去执行如果你们暂时不引入 DeepAgents用 LangGraph 原生节点也能实现同样的记忆闭环。记住一个原则先把状态层设计好行为层只是状态层之上的策略。3. 环境准备与前置条件在动手写代码之前先把环境准备好。这套示例代码不依赖重型服务本地一台开发机就能跑通。3.1 运行环境与 Python 版本建议使用 Python 3.10 及以上版本3.9 也可以运行但部分新版本依赖可能会提示不支持。Windows、macOS、Linux 均可唯一的差异在于向量库安装包。执行以下命令确认 Python 版本python --version如果还没有安装虚拟环境建议先创建一个独立环境避免依赖冲突python -m venv agent-memory-env source agent-memory-env/bin/activate # Windows 使用 agent-memory-env\Scripts\activate3.2 安装依赖核心依赖是 LangChain、LangChain-OpenAI、LangGraph再加上一个向量库。向量库二选一即可FAISS 在 Linux 和 macOS 上安装方便Chroma 是全平台通用Windows 用户如果装 FAISS 失败可以直接用 Chroma。pip install langchain langchain-openai langgraph pip install faiss-cpu # Linux / macOS # 或者 pip install chromadb # 全平台通用版本说明LangChain 和 LangGraph 都处于快速迭代期不同版本的 API 存在细微差异。以下代码基于当前社区最常用的写法如果你的环境中某些接口提示已废弃以官方文档和最新版本迁移说明为准。3.3 模型服务配置本文示例使用 ChatOpenAI 类你可以填入任意兼容 OpenAI 协议的服务商地址也可以换成本地模型。在项目根目录创建.env文件OPENAI_API_KEY你的_API_Key OPENAI_API_BASEhttps://你的模型服务地址然后在代码中加载环境变量import os from dotenv import load_dotenv load_dotenv() from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, temperature0)如果你的模型服务商不叫 OpenAI只需要替换ChatOpenAI为对应的 LangChain 模型类比如ChatZhipuAI、ChatOllama、ChatTongyi。记忆系统的核心逻辑与具体模型无关这一点是 LangChain 做抽象带来的好处。3.4 项目目录结构建议按模块拆分方便后续扩展到真实项目agent-memory-demo/ ├── .env ├── ecommerce_agent.py # 电商案例主文件 ├── memory/ │ ├── __init__.py │ ├── short_term.py # 短期记忆逻辑 │ └── long_term.py # 长期记忆逻辑 └── data/ └── ...如果只是跑通演示可以暂先不拆分把代码集中在单个文件中便于阅读。但一旦进入团队协作按模块拆分是必须的。4. 短期记忆实现让 Agent 记住“刚才说了什么”短期记忆的实现是所有记忆系统的基础。它的本质是在 LangGraph 的 State 中维护一个消息列表每次对话都把新的用户消息和 AI 回复追加进去。LangGraph 的消息归约器会自动处理追加逻辑这也是为什么add_messages这个字段在 LangGraph 中几乎成了一个约定俗成的标准。4.1 用 LangGraph 定义带短期记忆的 Agent先写一个最小可运行的对话图。这个图只有一个节点节点里直接调用大模型State 中通过Annotated[list, add_messages]来声明消息字段# 文件路径demo_short_term.py from typing import Annotated, TypedDict from dotenv import load_dotenv load_dotenv() from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langchain_openai import ChatOpenAI model ChatOpenAI(modelgpt-4o-mini, temperature0) class ChatState(TypedDict): messages: Annotated[list, add_messages] def chat_node(state: ChatState): response model.invoke(state[messages]) return {messages: [response]} graph StateGraph(ChatState) graph.add_node(chat, chat_node) graph.add_edge(START, chat) graph.add_edge(chat, END) app graph.compile() if __name__ __main__: # 模拟两轮对话 config {configurable: {thread_id: demo_user_001}} result app.invoke({messages: [{role: user, content: 你好我叫小明我喜欢黑色}]}, config) print(result[messages][-1].content) result app.invoke({messages: [{role: user, content: 我刚才说了我叫什么}]}, config) print(result[messages][-1].content)这个例子里LangGraph 会自动把所有历史消息追加到state[messages]中。第二次调用的时候模型能看到第一轮的完整上下文所以它能回答出“你刚才说你叫小明”。这也是短期记忆最简单的形态保留当前会话窗口内的所有消息。4.2 限制短期记忆窗口长度直接保留所有历史消息有一个隐患Token 会随着对话轮数线性增长最终撞上模型上下文窗口上限。企业级场景必须给短期记忆设“预算”。两种常见做法是滑动窗口和消息摘要。滑动窗口比较简单保留最近 N 条消息即可# 文件路径demo_short_term.py 中的新增函数 MAX_MESSAGES 10 def trim_messages_window(messages: list, max_messages: int MAX_MESSAGES): 保留最近 max_messages 条消息避免上下文无限增长。 if len(messages) max_messages: return messages return messages[-max_messages:]然后在chat_node中使用它def chat_node_with_trim(state: ChatState): trimmed_messages trim_messages_window(state[messages]) response model.invoke(trimmed_messages) return {messages: [response]}这个方案实现简单缺点是会“硬性遗忘”掉更早的内容。如果用户在第 20 轮突然提起第 2 轮说过的一个关键信息Agent 就再也找不回来了。所以更优雅的方案是用大模型做摘要当消息超过阈值时把旧消息压缩成一段摘要摘要保留在 State 中之后每次对话都携带摘要 最近 N 条完整消息。4.3 短期记忆摘要压缩# 文件路径demo_summary.py from langchain_core.messages import SystemMessage, HumanMessage def summarize_messages(messages: list) - str: 调用模型把较旧的消息压缩成一段摘要。 text_parts [] for msg in messages: text_parts.append(f{msg.type}: {msg.content}) text \n.join(text_parts) summary_prompt ( 你是记忆系统的一部分。请用简洁的中文摘要以下对话中的关键信息 包括用户身份、偏好、待办事项和重要事实。不要遗漏细节。\n\n f对话内容\n{text} ) response model.invoke([SystemMessage(contentsummary_prompt)]) return response.content实际项目中摘要不会每一轮都重新生成而是当消息数量超过阈值时把最早的一部分消息合并成摘要并用摘要替换掉那部分原始消息。LangGraph 官方也提供了RemoveMessage机制来从 State 中删除旧消息from langchain_core.messages import RemoveMessage # 在节点返回中删除旧消息 return { messages: [summary_message, RemoveMessage(idold_message.id)], }短期记忆做到这个程度已经可以支撑大多数客服、导购类场景。核心认知是短期记忆不是把上下文无限撑大而是用预算思维去管理窗口。该截断的截断该摘要的摘要该让位于当前问题的就让位。5. 长期记忆实现让 Agent 学会“回忆”而不是“全量检索”如果说短期记忆是“保留现场”那么长期记忆就是“跨会话重建现场”。它要解决的核心问题是用户已经离开一段时间今天重新回来Agent 怎么快速回忆起这个用户的画像、偏好和过去发生过的事。5.1 长期记忆的工作流程长期记忆不是简单地把所有历史记录塞进向量库然后检索 top-k 丢给模型。一个合格的长期记忆系统至少包含四个步骤写入从对话中抽取值得长期记忆的信息转成向量或结构化数据后存储。召回在当前对话发生时根据用户 ID 和当前问题检索相关记忆片段。评估对召回结果做相关性判断、去重、时效性过滤。更新根据新对话内容更新用户画像和记忆条目。这里特别强调“评估”这一步因为在真实数据中向量检索出来的结果不一定都对当前问题有用。比如用户今天问“帮我看看物流进度”检索系统可能召回了他三个月前对某件商品质量的抱怨。这时如果原样塞给模型反而会带偏回答。所以正确做法是召回结果只是“候选集”最终是否使用、如何使用要交给大模型做判断。这就是“回忆”和“检索”的本质区别。5.2 用向量库存储长期记忆下面的示例用 FAISS 作为向量存储。这里展示的是语义记忆的写入和召回# 文件路径memory/long_term.py from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import FAISS # 实际项目中建议使用统一的向量库配置 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) # 模拟一个用户的历史记忆片段 initial_memories [ 用户小王男身高175cm偏好宽松版型, 用户喜欢黑色和灰色的卫衣, 用户经常在促销活动期间下单, 用户曾在2024年11月购买过一件XL码的黑色卫衣, ] vectorstore FAISS.from_texts(initial_memories, embeddings) def recall_memory(query: str, top_k: int 3) - list: 根据当前问题召回相关记忆。 retriever vectorstore.as_retriever(search_kwargs{k: top_k}) docs retriever.invoke(query) return [doc.page_content for doc in docs] def add_memory(text: str): 写入一条新的长期记忆。 vectorstore.add_texts([text])调用方式# 测试召回 results recall_memory(用户喜欢什么颜色) for r in results: print(-, r)这个示例中“用户喜欢黑色和灰色的卫衣”最有可能被召回。但注意向量检索是“相似度匹配”不是“语义理解”。如果用户问“他平时穿什么尺码”系统应该召回“身高175cm偏好宽松版型”和“购买过XL码”这两条但对当前问题最关键的可能是“XL码”。所以召回之后必须有一个提炼和筛选的过程。5.3 让模型“回忆”而不是“搬运”下面这个函数体现了“回忆”的完整过程。它把检索结果作为素材结合当前用户问题让模型输出一份精简的、与当前问题相关的用户上下文# 文件路径memory/recall_with_llm.py from langchain_core.messages import SystemMessage, HumanMessage def build_llm_recall(llm, user_question: str, retrieved_docs: list) - str: 让模型从检索结果中提炼与当前问题相关的记忆。 doc_text \n.join([f- {d} for d in retrieved_docs]) prompt ( 你是一个记忆整理助手。下面是从用户历史记忆中检索到的片段 请筛选出与当前问题真正相关的内容并整理成一段简洁的用户上下文。\n 如果某些片段与当前问题无关请忽略它们。\n\n f当前用户问题{user_question}\n\n f检索到的记忆片段\n{doc_text}\n\n 请输出整理后的记忆内容不要输出多余的说明。 ) response llm.invoke([ SystemMessage(content你是一个严谨的记忆整理助手), HumanMessage(contentprompt), ]) return response.content这个函数的意义在于不是把 5 条、10 条检索片段全部堆进上下文而是让 LLM 先把它们“过一遍脑子”去掉噪声提取真正和当前问题有关的记忆。这也是为什么前文说“记忆系统不是把更多东西检索出来而是让 Agent 学会回忆”。检索是回忆的起点而不是终点。5.4 记忆的写入从对话中抽取值得记的信息长期记忆的写入不能把所有用户消息都存下来。设计一个抽取函数让模型判断哪些信息值得存入长期记忆# 文件路径memory/extract_memory.py import json def extract_memory_candidates(llm, user_message: str) - dict: 从用户消息中抽取值得长期记忆的信息没有则返回空字典。 prompt ( 从用户消息中抽取值得长期记住的偏好信息或事实。 只抽取稳定、可复用的信息比如用户身份、风格偏好、尺码、职业、家庭情况。 不要抽取一次性提问内容。\n 输出严格JSON格式例如{\偏好\: \喜欢黑色\, \尺码\: \XL\}没有则输出 {}。\n\n f用户消息{user_message} ) try: response llm.invoke(prompt) content response.content.strip() content content.strip(json).strip().strip() return json.loads(content) if content else {} except Exception: return {}在完整案例里会把这个函数接到 LangGraph 的更新节点实现“对话结束后自动沉淀记忆”的闭环。6. 电商案例一个能记住用户偏好和订单的导购 Agent前面把短期记忆和长期记忆分开讲清楚了这一节把它们组合成一个完整的电商导购 Agent。业务背景用户小王在上一轮会话中说了自己喜欢的颜色、尺码和风格今天重新进线问“再推荐一件卫衣”。Agent 需要自动回忆起小王的历史偏好并结合工具查询订单和商品信息给出个性化回答。6.1 记忆数据结构设计在编码之前先把记忆字段定义清楚。短期记忆就是messages列表长期记忆包括用户画像和向量记忆为了演示方便这里用字典数据结构模拟画像向量库仍然负责语义召回。记忆层存储字段示例内容短期记忆messages当前会话的消息历史长期记忆 - 用户画像user_profileuser_id、会员等级、尺码偏好、颜色偏好长期记忆 - 语义记忆vectorstore“小王喜欢宽松版型”“常在促销期下单”等自然语言片段工具结果last_tool_result最近一次订单/商品查询结果6.2 完整代码可运行的电商导购 Agent环境配置和第二、三节保持一致。这个示例中的工具全部使用模拟数据不真实调用外部 API。生产环境替换为真实接口即可。# 文件路径ecommerce_agent.py import json from typing import Annotated, TypedDict from dotenv import load_dotenv load_dotenv() from langchain_openai import ChatOpenAI from langchain_core.messages import AIMessage, HumanMessage, SystemMessage from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages model ChatOpenAI(modelgpt-4o-mini, temperature0) # ---------- 1. 模拟数据库 ---------- orders_db { user_1001: [ {order_id: A10001, product: 黑色宽松卫衣, status: 已签收, price: 299}, {order_id: A10002, product: 灰色运动裤, status: 运输中, price: 199}, ] } products_db [ {name: 黑色宽松卫衣, style: 宽松, color: 黑色, price: 299, stock: 100}, {name: 灰色连帽卫衣, style: 修身, color: 灰色, price: 359, stock: 80}, {name: 白色基础T恤, style: 标准, color: 白色, price: 99, stock: 200}, {name: 深蓝直筒牛仔裤, style: 直筒, color: 深蓝, price: 249, stock: 50}, ] user_profiles { user_1001: { user_id: user_1001, name: 小王, member_level: 黄金会员, height: 175, preferred_style: 宽松, preferred_colors: [黑色, 灰色], preferred_size: XL, } } # ---------- 2. 工具函数模拟 ---------- def query_orders(user_id: str) - list: return orders_db.get(user_id, []) def query_product(style: str None, color: str None) - list: result products_db if style: result [p for p in result if p[style] style] if color: result [p for p in result if p[color] color] return result # ---------- 3. 长期记忆模块 ---------- def recall_user_profile(user_id: str) - dict: 实际项目中应优先从用户画像数据库读取再结合向量检索。 return user_profiles.get(user_id, {}) def update_user_profile(user_id: str, new_info: dict): 把新抽取的偏好写入用户画像和向量记忆。 profile user_profiles.setdefault(user_id, {user_id: user_id}) profile.update(new_info) # 这里可以调用 vectorstore.add_texts() 写入语义记忆 print(f[Memory Updated] {user_id}: {new_info}) def extract_memory_candidates(user_message: str) - dict: 让模型抽取值得长期记忆的信息。 prompt ( 从用户消息中抽取值得长期记住的偏好信息或事实。 只抽取稳定、可复用的信息比如身份、风格偏好、尺码、颜色偏好。 不要抽取一次性提问内容。\n 输出严格JSON格式例如{\偏好\: \喜欢黑色\, \尺码\: \XL\}没有则输出 {}。\n\n f用户消息{user_message} ) try: response model.invoke(prompt) content response.content.strip().strip(json).strip().strip() return json.loads(content) if content else {} except Exception: return {} # ---------- 4. 图状态定义 ---------- class AgentState(TypedDict): messages: Annotated[list, add_messages] user_profile: dict recall_memory: str last_tool_result: str # ---------- 5. 图节点定义 ---------- def recall_node(state: AgentState): 长期记忆召回根据用户ID读取画像。 user_id state[user_profile].get(user_id) profile recall_user_profile(user_id) memory_text ( f姓名{profile.get(name, 未知)} f会员等级{profile.get(member_level, 未知)} f身高{profile.get(height, 未知)} f偏好风格{profile.get(preferred_style, 未知)} f偏好颜色{profile.get(preferred_colors, 未知)} f偏好尺码{profile.get(preferred_size, 未知)} ) return {recall_memory: memory_text} def agent_node(state: AgentState): Agent 决策节点结合当前问题和长期记忆调用工具输出回复。 last_human_message state[messages][-1].content user_id state[user_profile].get(user_id) # 简单的意图识别实际项目可替换为 DeepAgents 的规划逻辑 tool_result if 订单 in last_human_message or 物流 in last_human_message: orders query_orders(user_id) tool_result json.dumps(orders, ensure_asciiFalse) elif 推荐 in last_human_message or 卫衣 in last_human_message or 商品 in last_human_message: profile state[user_profile] style profile.get(preferred_style) color profile.get(preferred_colors, [None])[0] products query_product(stylestyle, colorcolor) tool_result json.dumps(products, ensure_asciiFalse) else: tool_result 未命中工具调用请直接回答。 sys_prompt SystemMessage(content( 你是电商导购助手。请结合用户画像和工具查询结果用简洁、自然的中文回答用户。\n f用户画像{state[recall_memory]}\n f工具查询结果{tool_result}\n )) user_msg HumanMessage(contentlast_human_message) response model.invoke([sys_prompt, user_msg]) return { messages: [AIMessage(contentresponse.content)], last_tool_result: tool_result, } def update_memory_node(state: AgentState): 记忆更新节点从最后一条用户消息中抽取偏好更新用户画像。 # 找到最近一条用户消息 last_user_message None for msg in reversed(state[messages]): if msg.type human: last_user_message msg.content break if last_user_message: candidates extract_memory_candidates(last_user_message) if candidates: update_user_profile(state[user_profile][user_id], candidates) return {user_profile: state[user_profile]} # ---------- 6. 构建图 ---------- graph StateGraph(AgentState) graph.add_node(recall, recall_node) graph.add_node(agent, agent_node) graph.add_node(update_memory, update_memory_node) graph.add_edge(START, recall) graph.add_edge(recall, agent) graph.add_edge(agent, update_memory) graph.add_edge(update_memory, END) app graph.compile() # ---------- 7. 模拟两轮会话 ---------- def run_session(user_message: str, user_id: str): 模拟一次会话自动触发记忆系统。 config {configurable: {thread_id: user_id}} result app.invoke( { messages: [{role: user, content: user_message}], user_profile: {user_id: user_id}, recall_memory: , last_tool_result: , }, config, ) return result[messages][-1].content if __name__ __main__: # 第一轮用户告知偏好 reply1 run_session(我喜欢黑色和灰色的宽松卫衣给我推荐几件, user_1001) print(AI:, reply1) # 第二轮模拟新会话语境直接追问 reply2 run_session(我今天想再看看卫衣, user_1001) print(AI:, reply2)6.3 代码关键逻辑说明这部分代码虽然简化了真实业务的复杂度但已经包含了企业级记忆系统的完整骨架值得逐段理解。recall_node负责长期记忆召回。它在每次会话开始时执行从用户画像中拼接出记忆文本。注意这里用的是硬编码画像读取真实项目中应该把“记忆文本”换成向量检索结果并且经过前文说的模型“回忆”提炼后再传给下游。agent_node是 Agent 的核心决策节点。它先判断用户意图再调用模拟工具获取结果最后把“用户画像 工具结果 当前问题”一起交给大模型生成回复。这里面体现了一个关键设计回忆内容不是模型自己凭空想出来的而是记忆系统在节点链路上显式注入的。update_memory_node负责记忆更新。它在每轮回答结束后运行用大模型从用户最新消息中抽取值得记住的偏好再写入用户画像。这个节点的存在让长期记忆具有了“自我生长”能力——Agent 用得越久对用户越了解而不是永远停留在初始配置。6.4 预期运行效果运行主文件后输出类似如下AI: 小王你好根据你偏好的宽松版型和黑色系我为你推荐这件黑色宽松卫衣价格299元目前库存充足。需要顺便看看搭配的裤子吗 [Memory Updated] user_1001: {偏好: 黑色和灰色的宽松卫衣} AI: 小王你好上次你说喜欢黑色和灰色的宽松卫衣今天我还是优先给你推荐黑色宽松卫衣和灰色连帽卫衣。第一件偏宽松第二件偏修身你更想看哪一件第二轮的回复明显带上了第一轮会话留下的记忆痕迹。如果你把thread_id更换成一个新用户 IDAgent 就不会知道“小王喜欢黑色和宽松版型”。这个对比能直观验证长期记忆是否生效。7. 运行验证与效果评估代码跑通只是第一步。要判断这套记忆系统在企业项目里是否真的可用需要从三个层面验证。7.1 功能验证三个必测场景第一个场景是短期记忆验证。同一个thread_id下连续提问比如先问“我叫小明”再问“我叫什么”Agent 要能正确回答。这验证的是消息列表在 LangGraph State 中的正常流转。第二个场景是长期记忆验证。新开一个thread_id但使用同一个user_id问“你知道我喜欢什么颜色吗”。如果系统读取了用户画像并正确回答出“黑色和灰色”说明长期记忆读取链路没问题。第三个场景是记忆更新验证。在第一轮告诉 Agent“我现在更喜欢蓝色”第二轮问“我的偏好是”看 Agent 是否更新为蓝色而不是停留在旧的黑色。这可以检验update_memory_node的抽取和写入是否生效。7.2 质量评估指标工程化评估时建议关注四个指标指标定义健康标准建议记忆召回率应该被召回的相关记忆有多少被成功召回越高越好低于 60% 需要检查向量库分块和索引记忆准确率召回结果中真正相关、无冲突的比例低于 70% 说明需要加强 LLM“回忆”筛选环节记忆污染率错误记忆或过期记忆被写入的比例必须低于 5%否则会持续误导 Agent记忆写入延迟一轮对话结束后更新记忆的耗时应控制在 1-2 秒内避免影响用户体验记忆污染率是最容易被忽视的指标。很多团队只关心“能不能搜到”不关心“会不会记错”。一旦一条错误记忆被写入它会在后续每次会话中被不断放大导致 Agent 越来越“固执”。所以写入环节的抽取模型必须保守拿不准的信息宁可不记。7.3 可观测性日志里要有什么生产环境必须能回答一个问题Agent 之所以给出某个回答是因为它看到了哪段记忆。建议在关键节点输出结构化日志[recall] user_iduser_1001 memory_typeuser_profile length42 [agent] toolquery_product hits2 cost_tokens1320 [update_memory] extracted{preferred_colors: [黑色, 灰色]}有了这些日志当用户投诉“Agent 为什么乱推荐”时你可以快速定位是检索问题、模型问题还是记忆更新问题。8. 常见问题与排查方法下面这些坑是 Agent 记忆系统在从 demo 走向生产时最容易遇到的。整理成表格方便对照排查问题现象可能原因排查方式解决方案对话轮数一多就报上下文超限短期记忆窗口无限制查看模型请求中的输入 Token 数增加滑动窗口截断或摘要压缩新会话中 Agent 不记得用户长期记忆没有按 user_id 关联检查 recall_node 中的用户 ID 是否从请求头正确传递统一用户 ID 体系所有会话从 Token 或参数中解析检索结果和当前问题完全无关向量库中数据分块不合理打印召回结果的文本内容优化文本切分策略增加元数据过滤模型回 JSON 时格式不稳定指令不够明确或模型能力不足打印模型原始输出使用结构化输出解析器增加失败重试同一记忆被重复写入update_memory_node 每轮都执行且无去重查看用户画像中的数据是否有重复字段写入前做相似度比较或主键去重多个节点并行写入记忆导致冲突LangGraph 的 Send API 并发更新同一字段查看 State 更新日志记忆更新统一收敛到单个节点串行执行Agent 越来越“固执”忽略用户新偏好旧记忆权重过高新记忆无法覆盖检查用户画像中旧值更新逻辑设置记忆版本号同一字段新值覆盖旧值并记录时间向量库体积增长过快每轮对话都写入大量文本检查写入节点触发的频率设置写入置信度阈值抽取出空字典时不写入这里再强调一个容易忽略的点LangGraph 中Send这类并行 API 适合做任务分发但不要用在对同一用户记忆的并发写入上。记忆更新的状态是强一致的一旦两个节点同时修改user_profile后写入的值可能覆盖前一个节点的关键信息而且这种问题非常难复现。设计原则是记忆写入收敛到单一节点串行执行。9. 企业级落地最佳实践与后续学习路线示例代码跑通以后距离真正的企业级系统还有一段距离。下面这几条实践建议是按照优先级排序的可以对照自家项目逐步补齐。9.1 记忆写入要有置信度门槛不是所有对话都值得写入长期记忆。用户在闲聊时说了一句话不应该进入用户画像。建议在extract_memory_candidates的 prompt 中增加强度要求只有信息足够明确、稳定、可复用时才输出并且可以要求模型输出置信度分数低于阈值的直接丢弃。这是降低记忆污染率最有效的手段。9.2 记忆要有过期与遗忘机制用户的偏好会变化会员等级会调整地址电话更会更换。画像表里每个字段都应该带一个updated_at时间戳并根据不同业务字段设置不同的有效期。比如“收货地址”可能一年就变“尺码偏好”相对稳定可以保留更长时间。过期字段在召回时降权甚至不再写入上下文。遗忘不是系统缺陷而是长期可靠性的保证。9.3 权限、隐私与多租户隔离记忆系统存储的是用户敏感信息这一条必须前置考虑。用户画像数据要加密存储模型调用链路中不要输出手机号、地址等敏感字段会话数据的使用要遵循用户授权和公司合规要求。多租户场景下租户 A 的用户记忆绝不能被租户 B 的 Agent 检索到。最简单的实现方式是向量库和画像表中都带上tenant_id和user_id两个维度检索时强制过滤。9.4 成本控制与降级策略每一次长期记忆召回和每次记忆抽取都会产生额外 Token 消耗。建议做一个成本开关当系统负载高或用户为低价值流量时可以关闭update_memory_node只保留短期记忆和读取已有画像的能力。降级后用户体验略降但系统仍能正常工作。另外摘要压缩不要每轮都做只在消息数达到阈值时触发并且可以做成异步任务。9.5 测试与灰度记忆系统要像数据库迁移一样谨慎。上线前准备几组测试用户新用户、老用户、偏好冲突用户、长期不活跃用户。先在小流量灰度重点观察记忆写入的准确率和用户反馈。一旦发现记忆污染问题要提供“一键清除用户画像”的运营工具这在客服场景中几乎是必备的。9.6 后续学习路线如果你刚接触 LangChain建议先把官方入门教程中的基础组件过一遍如果你已经会写简单的 Chain下一个重点是深入 LangGraph 的 State 和 Checkpointer 机制理解消息归约器add_messages和节点返回值之间的协作关系如果你已经能熟练编排 LangGraph再去看 DeepAgents 这类高层框架理解它们如何把规划、反思、工具调用这些行为封装成可复用的 Agent 能力。对于 DeepAgents一个更稳妥的判断是它还在快速迭代期但值得持续关注。不要等框架完全稳定再学先掌握 LangGraph 的状态化编排思想因为无论上层框架怎么变状态、记忆、工具、反馈循环这些核心概念是稳定不变的。等你能够不看示例代码独立画出“召回记忆 → 决策 → 调用工具 → 更新记忆”的状态图时你对 Agent 记忆系统的理解就已经超过大多数停留在 prompt 拼接阶段的开发者了。建议把这篇文章收藏备用动手把电商示例跑通再对照你们自己的业务数据做一次完整的记忆链路改造。
