Agent记忆系统最近讨论热度非常高。很多人以为给 Agent 接个数据库、把历史对话存下来就算有记忆了,实际跑起来就会发现:短期记忆容易丢,长期记忆存了但不会用,多个 Agent 各存各的,最后用户问一句“我上次问过什么”,系统就答不上来。这篇文章围绕 LangChain、LangGraph 和 DeepAgents 这套开源组合,讲清楚怎么搭一个企业级的 Agent 记忆系统,覆盖短期记忆、长期记忆和电商场景落地,适合正在做 Agent 应用开发、又不想从零造轮子的同学看。
我会先拆一下记忆系统到底要解决什么问题,再给最小可运行的链路,然后分别讲短期记忆、长期记忆、LangGraph 编排、DeepAgents 多 Agent 协作,最后落到电商客服和推荐场景,并补上生产环境常见的排查思路。整体偏工程实践,不堆概念。
1. 先想清楚:短期记忆、长期记忆和多 Agent 共享记忆分别解决什么问题
记忆系统听起来很简单,但落地时最容易出问题的地方往往不是模型,而是我们根本没分清“短期记忆”“长期记忆”“共享记忆”是三种完全不同的工程问题。
1.1 短期记忆做上下文拼接,长期记忆做知识与画像沉淀
短期记忆解决的是“当前这轮对话里,Agent 还能不能记得前两句说了什么”。典型场景是用户连续提问:“帮我找一款办公用的无线鼠标”“要白色的”“价格在 300 以内”。如果 Agent 没有短期记忆,第三句就会被当成独立问题处理,完全不知道“白色”“300 以内”是对鼠标的限制条件。
长期记忆解决的是“下一次会话,Agent 还能不能记得这个用户是谁”。典型场景是用户昨天问过某款商品,今天回来说“那个鼠标还有货吗”。如果只有短期记忆,今天的新会话就是空白;如果把所有历史都塞给模型,又会因为信息太杂导致回答质量下降。
所以我会把长期记忆拆成三类来设计:用户画像、历史结论、原始对话摘要。用户画像解决“他喜欢什么”,历史结论解决“上次我们商量出了什么”,对话摘要解决“如果用户要追溯,还能找到上下文”。
1.2 记忆系统不是简单加一个数据库
我看到很多早期项目会犯一个错:给 Agent 挂上一个向量数据库,把所有历史对话都写进去,用户提问时就把相似内容检索出来拼到 prompt 里,以为这样就是记忆系统。跑一段时间就会发现两个问题:一是检索结果经常不相关,二是同一个用户的信息散落在几百条记录里,模型根本拼不出完整画像。
这里有一个很关键的观点:记忆系统不是把更多东西检索出来,而是让 Agent 学会“回忆”。你不需要让模型知道用户昨天说过的每一句话,你需要的是在合适的时机,把与当前问题相关的那几条关键信息唤醒出来。类似 RippleMem 这类记忆增强思路强调的也是这一点:记忆要有权重、有关联、要分主次。
所以在设计阶段就要明确:哪些信息进短期记忆窗口,哪些进长期记忆库,哪些可以直接丢弃。不要把所有内容都一股脑写进向量库,更不要每次提问都把 top 20 条历史记录翻出来。
1.3 技术选型:LangChain、LangGraph、DeepAgents 各负责什么
这套方案里三个组件不是竞争关系,而是分工关系。
| 组件 | 在记忆系统中的角色 | 通俗理解 |
|---|---|---|
| LangChain | 负责模型调用、工具调用、提示词组装 | 连接大模型和外部能力的管道 |
| LangGraph | 负责编排有状态的工作流,管理节点和状态流转 | 记忆读写、Agent 决策的流程图 |
| DeepAgents | 负责多 Agent 场景下的任务拆分、委派和结果汇总 | 让多个子 Agent 共享一套记忆协作干活 |
如果你只是做一个带记忆的单 Agent 聊天机器人,LangChain 加 LangGraph 基本够用。如果业务里同时有客服 Agent、推荐 Agent、订单 Agent、售后 Agent,那就需要 DeepAgents 这种多智能体协作思路来统一调度,让它们访问同一份用户记忆,避免各存各的。
2. 环境准备与最小可运行记忆链路
不要一上来就搭复杂的多 Agent 架构。我建议先把一条最简单的“带记忆对话”跑通,再逐步加长短期记忆、加长期记忆库、加多 Agent。
2.1 依赖环境和模型接入准备
开发环境建议用 Python 3.9 或更高版本。需要安装 langchain、langgraph,以及一个向量库客户端。常见组合是 langchain 负责模型调用,langgraph 负责流程编排,向量库用支持本地运行的方案做开发和测试。
pip install langchain langchain-openai langgraph faiss-cpu如果你的长期记忆库要用 PostgreSQL 或 Redis,再根据实际情况安装对应客户端。我建议第一次跑通时不要接太多外部中间件,先用 SQLite 或本地文件存结构化记忆,用 FAISS 存向量,等链路通了再迁到生产环境。
这里要提醒一点:如果你已经装过 langchain 相关依赖,尽量在同一个虚拟环境里操作,不要直接装到系统 Python 里。LangChain 和 LangGraph 版本更新频率高,不同版本之间 API 可能有差异,虚拟环境能减少很多奇怪报错。
2.2 先跑一条带记忆的对话
下面的代码是简化示例,重点不是完全照抄,而是理解记忆链路怎么串起来。我以 LangGraph 的状态流为核心,把用户 ID、会话 ID、消息列表和记忆内容放到全局状态里。
from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END from langchain_openai import ChatOpenAI class AgentState(TypedDict): user_id: str # 用户标识 session_id: str # 会话标识 messages: list # 当前会话的消息列表 long_term_memory: dict # 从长期记忆里读取出来的关键信息 llm = ChatOpenAI(model="你的模型名称") def call_model(state: AgentState): # 把长期记忆拼到系统提示词里 memory_text = "" if state.get("long_term_memory"): for k, v in state["long_term_memory"].items(): memory_text += f"{k}: {v}\n" system_prompt = f"你是企业客服 Agent。\n以下是该用户的已知信息:\n{memory_text}" messages = [{"role": "system", "content": system_prompt}] + state["messages"] response = llm.invoke(messages) return {"messages": state["messages"] + [{"role": "assistant", "content": response.content}]} graph = StateGraph(AgentState) graph.add_node("agent", call_model) graph.add_edge(START, "agent") graph.add_edge("agent", END) app = graph.compile()这段代码的核心点在于:每次调用模型之前,把该用户的长期记忆先拼到系统提示词里。如果用户有“偏好白色”“预算 300 到 500 元”这类记忆,Agent 在首轮回复时就能用上,而不是等用户重复一遍。
2.3 检查记忆是否生效:看状态、看日志、看数据库
很多同学跑完示例发现 Agent 表现和原来一样,就以为记忆没生效。其实问题大多出在“记忆有没有被真正写入”和“记忆有没有被读取到”。
我的验证顺序是:
- 在第一次对话里写入一条明确的用户偏好,例如“用户喜欢白色”。
- 开启一个新会话,不带任何上下文,直接问 Agent:“我喜欢什么颜色?”
- 如果 Agent 能回答“白色”,说明长期记忆读取链路通畅。
- 如果答不上来,先看数据库里有没有这条记录,再看读取函数有没有按 user_id 过滤,最后看拼接 prompt 的代码有没有真的执行。
3. 短期记忆:会话上下文与状态管理
短期记忆的核心是窗口化管理。不要简单地把所有历史消息一直堆到 prompt 里,模型上下文有限,成本也会随长度上涨。
3.1 短期记忆的本质是窗口化的对话状态
短期记忆哪怕实现得再简单,也要包含三样东西:当前用户说的内容、Agent 之前回复的内容、以及当前任务相关的临时变量。
在 LangGraph 里,我一般会用 state 里的 messages 字段保存对话消息,同时维护一个窗口大小。假设窗口是 6 轮,超过之后就把最早的消息移出模型调用范围,而不是真的删除。这样既能控制 context 长度,又能保留最近对话的核心上下文。
def trim_messages(messages, max_rounds=6): if len(messages) <= max_rounds * 2: return messages # 保留最近 max_rounds 轮 return messages[-(max_rounds * 2):]窗口大小不是越大越好。如果你的场景是“用户一次性描述很多条件”,窗口可以放大一点;如果是高频短对话,窗口 4 到 6 轮通常够了。具体值要根据模型上下文长度和业务复杂度调整。
3.2 LangGraph State 和 Checkpoint 在记忆里的作用
LangGraph 的 State 天然适合做短期记忆。因为每个节点都接收同一个状态对象,节点之间不会丢数据。但要注意:状态对象是每次运行结束后就释放的,如果服务重启、进程退出,短期记忆就丢了。
想让短期记忆具备持久性,就要用到 Checkpoint 机制。它的作用是把每个时间点的状态保存下来,比如保存到 SQLite 或 Redis,这样即使 Agent 中断,也能从最近一个状态点继续。
这个机制对长对话特别有用。用户和 Agent 对话到一半,网络断了,服务端重启,重连之后还能从断点继续,不用让用户重新描述一遍。
3.3 上下文超长时如何截断和压缩
假设用户一口气发了很长一段问题,或者对话进行了几十轮,这时候不能只做“截断”,更推荐做“摘要”。
截断适合处理临时溢出,比如把最早的消息丢弃。摘要适合处理重要信息保留,例如每 5 轮结束后,生成一段关于“用户目前的诉求是什么、哪些条件已确认、哪些还没确认”的摘要,放进会话状态里。
def summarize_conversation(messages, llm): text = "\n".join([f"{m['role']}: {m['content']}" for m in messages]) summary_prompt = f"请总结以下对话中的用户需求、约束条件和待办事项:\n{text}" summary = llm.invoke([{"role": "user", "content": summary_prompt}]) return summary.content生成摘要的调用本身有成本,不建议每轮都做。我一般会设定一个轮次阈值,比如每 8 轮触发一次,或者当累计 Token 超过预设值时才触发。
4. 长期记忆:向量库、摘要与用户画像
长期记忆是让 Agent 跨会话记住用户的关键。但长期记忆不能只靠一个向量库,它需要分存储、分流程、分策略。
4.1 长期记忆的三种常见存储形态
| 存储形态 | 适合内容 | 示例 |
|---|---|---|
| 结构化数据库 | 用户画像、订单状态、售后进度 | 用户 ID、偏好品类、收货地址 |
| 向量数据库 | 非结构化语义记忆 | 历史对话摘要、用户表达过的个性化描述 |
| 摘要文本 | 长周期对话核心结论 | “用户上次决定先不买,等双十一再比价” |
实际项目里这三种形态会同时存在。用户画像和订单信息适合用关系型数据库存,因为要支持精确查询;文本型记忆适合向量化,因为用户可能用不同说法表达相同意思;跨多轮对话的结论适合摘要,因为摘要能把十几轮聊天浓缩成两三句话。
4.2 记忆写入和读取的工程流程
写入流程要排在回复流程之后。不要让 Agent 在生成回复的同时同步写记忆,因为模型输出可能包含幻觉内容,直接写入会让记忆库积累错误信息。
更稳妥的做法是:
- 用户提问。
- Agent 生成回复。
- 从用户本次输入中抽取结构化信息,比如偏好、商品、价格区间。
- 将抽取结果写入结构化记忆和向量记忆。
- 如果本轮对话较长,异步生成对话摘要。
读取流程是:
- 用户发起新会话。
- 按 user_id 从结构化库读取用户画像。
- 按 user_id + 当前问题生成向量检索条件,从向量库读取相关记忆片段。
- 筛选时间范围、业务类型、相似度分数。
- 把最终选中的记忆拼进 prompt。
def read_long_term_memory(user_id, query, top_k=3, score_threshold=0.6): profile = memory_db.get_user_profile(user_id) candidates = vector_store.search(query, top_k=top_k) filtered = [ item for item in candidates if item.score >= score_threshold and item.user_id == user_id ] return { "profile": profile, "memory_items": filtered, }4.3 记忆检索时如何避免把无关内容拉进来
向量检索不是召回越多越好。如果你把 top_k 设成 20,模型看到的内容可能有一大半是无关的。我推荐的起点配置是 top_k=3,相似度阈值根据实际效果调整。
除了相似度,还要加时间过滤。用户去年对某个品类感兴趣,不代表现在仍然感兴趣。比如用户去年天天看键盘,今年转向了无线鼠标,那么“喜欢键盘”这条记忆就不应该主导推荐结果。
所以在记忆数据表里要记录写入时间、更新时间、使用次数、失效时间。检索时优先使用最近 90 天内有活跃记录、使用次数较高的记忆。
5. 用 LangGraph 编排记忆读写流程
有了短期记忆、长期记忆和多 Agent,接下来要解决的问题是:这些记忆操作按什么顺序执行、谁先谁后、失败怎么处理。LangGraph 的价值就是把这个流程变成可维护的状态图。
5.1 记忆写入节点和读取节点设计
我会把记忆操作拆成独立节点,而不是塞进模型调用节点里。这样每个节点只做一件事,出了问题也容易排查。
def read_memory_node(state: AgentState): user_id = state["user_id"] user_query = state["messages"][-1]["content"] memory = read_long_term_memory(user_id, user_query) return {"long_term_memory": memory} def write_memory_node(state: AgentState): user_id = state["user_id"] user_query = state["messages"][-1]["content"] facts = extract_memory_facts(user_query) save_memory(user_id, facts) return {}流程上是:先读取记忆,再调用模型,再异步或同步写入记忆。读取一定要在模型调用之前,写入要在模型调用之后。
如果写入失败,不应该阻塞主流程。更合理的做法是:先把 Agent 的回复返回给用户,然后把记忆写入请求放到异步队列里,失败时重试。
5.2 DeepAgents 在多 Agent 场景下如何共享记忆
当系统里有多个 Agent 时,最大的问题是“记忆孤岛”。客服 Agent 记住的信息,推荐 Agent 不知道;订单 Agent 更新了订单状态,客服 Agent 还在用旧状态。解决办法是让所有子 Agent 访问同一个记忆服务,而不是各自维护私有记忆。
用 DeepAgents 这类多智能体方案时,我会这样拆:
| 子 Agent | 负责内容 | 需要的记忆 |
|---|---|---|
| 客服 Agent | 回答售前售后问题 | 用户画像、订单状态、售后进度 |
| 推荐 Agent | 推荐商品 | 用户偏好、历史浏览、价格区间 |
| 订单 Agent | 查询订单、催发货 | 订单状态、物流信息、用户地址 |
它们之间可以通过共享状态对象来传递信息。比如用户问“我上次看的那款鞋有 42 码吗”,客服 Agent 先从长期记忆里找到“上次看的鞋是哪一款”,再调用商品库存工具,最后把结果写回会话状态。
5.3 同一个用户跨会话的上下文如何恢复
跨会话恢复的关键是区分 session_id 和 user_id。session_id 每次会话都可能变,user_id 是用户身份的唯一标识。
新会话开始时的流程应该是:
- 用户登录,拿到 user_id。
- 创建一个新的 session_id。
- 从长期记忆库读取该 user_id 的用户画像。
- 把用户画像作为短期记忆的初始上下文。
这样即使用户换了一台设备、开启了一个全新会话,Agent 也能快速恢复对用户的基本认知。
6. 电商场景实战:客服与推荐共用一个记忆系统
下面用一个电商场景把整套思路串起来。这个场景很典型,因为电商用户的问题往往横跨多个 Agent:先咨询商品,再查订单,然后售后,最后还可能要推荐。每一步都需要用到之前的记忆。
6.1 场景拆解:用户画像、订单状态、售后记录
假设平台有一个老用户,ID 是 u_1024。系统里已经积累了以下记忆:
| 记忆类型 | 内容 | 来源 |
|---|---|---|
| 用户画像 | 偏好白色、办公用品、预算 300 到 500 元 | 历史对话抽取 |
| 历史行为 | 最近 7 天浏览过 3 款无线鼠标 | 行为日志 |
| 订单状态 | 订单 20250118 处于待收货状态 | 订单系统同步 |
| 售后进度 | 上一单因鼠标滚轮异响申请过换货 | 售后工单 |
第一次对话时,用户问:“有没有适合办公室用的无线鼠标?”
客服 Agent 读取长期记忆后知道用户偏好白色、预算 300 到 500,所以回答时就会优先推荐白色、价格在预算范围内的商品。这个推荐不是临时猜的,而是有历史依据的。
6.2 从单会话到跨会话的电商客服记忆
第二天用户又来了,说:“上次你看的那款鼠标,有黑色吗?”
这个问题的难点在于“那款鼠标”并没有在当前会话中出现过。如果没有长期记忆,Agent 只能回答“我不知道你指的是哪款”。有了长期记忆,系统可以通过 user_id 找到上一次会话中的商品 ID,再回复“您上次看的是 A 型号,目前有黑色版本,价格是 399 元”。
这个过程在技术上分成四步:
- 按 user_id 读取最近一次会话摘要。
- 从摘要中提取商品 ID。
- 调用商品查询工具获取该商品的最新库存和颜色。
- 组装回复。
6.3 推荐 Agent 如何复用长期记忆
推荐 Agent 和客服 Agent 不是独立工作的。用户咨询完无线鼠标后,可能接着问“有没有搭配的键盘”。如果推荐 Agent 能读取客服 Agent 刚才写入的“用户正在看无线鼠标”的记忆,就能给出更连贯的搭配推荐。
实现上,推荐 Agent 需要一个入口函数:接收 user_id 和当前问题,返回推荐商品列表。它优先使用用户画像,再结合当前会话中的临时偏好。
def recommend(user_id, current_query): profile = get_user_profile(user_id) recent_context = get_recent_session_summary(user_id) candidates = search_products_by_profile(profile, recent_context) return rank_products(candidates)6.4 记忆数据表设计和 API 返回样例
一个可落地的记忆表至少要有这些字段。
CREATE TABLE user_memory ( id BIGINT PRIMARY KEY AUTO_INCREMENT, tenant_id VARCHAR(64) NOT NULL, user_id VARCHAR(64) NOT NULL, memory_key VARCHAR(128) NOT NULL, memory_value TEXT NOT NULL, memory_type VARCHAR(32) NOT NULL, source_session_id VARCHAR(64), score FLOAT DEFAULT 0, created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL, expire_at DATETIME NULL, KEY idx_user (tenant_id, user_id), KEY idx_type (tenant_id, memory_type) );memory_type 可以用来区分 user_profile、order_status、after_sale、session_summary。检索时必须带 tenant_id 和 user_id,避免租户之间数据串了。
API 返回可以用下面的结构,方便前端理解这次回复依据了哪些记忆:
{ "user_id": "u_1024", "session_id": "s_88321", "memory_used": [ { "type": "user_preference", "content": "偏好白色、预算300-500元", "score": 0.88 } ], "reply": "您上次看的是A型号无线鼠标,目前有黑色版本,价格为399元。" }7. 排查链路与生产化建议
最后这部分是我在实际项目中踩坑最多的,也是很多教程不会写清楚的。建议收藏起来,遇到问题时对照排查。
7.1 记忆不生效时按什么顺序排查
记忆不生效的现象看起来都是“Agent 没记住用户说过的话”,但原因可能完全不同。我一般按这个顺序排查:
- 先确认记忆有没有写入。去数据库按 user_id 查记录,如果记录不存在,检查写入逻辑有没有被触发。
- 再确认记忆读取时用的 user_id 和写入时是否一致。前端只要传错了一个字段,就算逻辑没问题也读不出来。
- 检查检索过滤条件。top_k 是不是设成了 0,相似度阈值是不是太高,时间过滤是不是把最新记录过滤掉了。
- 检查最终给模型的 prompt。把拼装后的 messages 打印出来,看记忆内容到底有没有在里面。
- 最后才怀疑模型本身。有时候模型没按记忆回答,是因为指令不够明确,可以改成“请严格依据以下用户已知信息回答”。
不要一开始就怀疑向量库有问题。大多数情况是数据没写进去、ID 对不上、过滤条件太严格。
7.2 记忆污染和过期数据怎么处理
记忆系统最怕的不是丢数据,而是记住错误数据。模型幻觉、用户临时随口说的一句话、过期很久的画像,都会被当成长期记忆存下来。
我的处理习惯是:
- 结构化记忆写入前要经过规则校验,比如价格必须大于 0、手机号必须是合法格式。
- 文本记忆写入前要用模型或规则判断是否属于“可长期保存的稳定偏好”。
- 给每条记忆设置 TTL,比如“用户最近关注无线鼠标”这种短期兴趣,60 天后自动降级。
- 定期做记忆合并,同一个用户同一类记忆如果超过 5 条,就打包成一条摘要,减少向量检索的噪声。
如果发现已经写入错误记忆,要能手动删除或标记失效。后台管理界面哪怕做得简陋一点,也要能支持按用户 ID 查询和删除记忆,否则线上数据出问题时会非常被动。
7.3 多租户与权限隔离注意事项
企业级系统通常有多个租户,比如不同商家或不同部门。记忆数据必须按租户隔离,否则 A 商家可能会读到 B 商家用户的记忆。
隔离的核心是在所有记忆读写函数的参数里都强制带上 tenant_id,并且在 SQL 查询条件里明确过滤。建议在代码层封装统一的记忆服务,不直接让业务代码访问底层表。
7.4 性能与成本控制经验
记忆系统不是只跑通就行,生产环境要考虑性能和成本。
高频用户的热记忆适合放缓存。比如用户当前会话里刚写入的偏好,短期内重复读取概率很高,不需要每次都查向量库。
批量写入比逐条写入效率高。客服对话结束后,系统可能同时产生用户画像、商品偏好、对话摘要三条记忆,这时候应该合并成一次批量写入。
向量检索也不要所有请求都打向量库。可以先查结构化画像,只有命中不到时才做向量检索。这样能减少向量库的压力,也能降低整体延迟。
最后说一点我的真实感受:企业级 Agent 记忆系统能不能落地,关键不在用了多贵的模型或组件,而在于把记忆的写入时机、读取条件、过滤策略、隔离机制和清理策略设计清楚。刚开始做的时候,先把单 Agent 单用户的记忆链路跑稳,再逐步扩展。先把记忆写对,再考虑批量;先把检索调准,再加并发。这套思路比到处搜教程、抄代码要实用得多。