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_rounds6): 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_k3, score_threshold0.6): profile memory_db.get_user_profile(user_id) candidates vector_store.search(query, top_ktop_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_k3相似度阈值根据实际效果调整。除了相似度还要加时间过滤。用户去年对某个品类感兴趣不代表现在仍然感兴趣。比如用户去年天天看键盘今年转向了无线鼠标那么“喜欢键盘”这条记忆就不应该主导推荐结果。所以在记忆数据表里要记录写入时间、更新时间、使用次数、失效时间。检索时优先使用最近 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 单用户的记忆链路跑稳再逐步扩展。先把记忆写对再考虑批量先把检索调准再加并发。这套思路比到处搜教程、抄代码要实用得多。