最近在逛技术社区和各类 Agent 项目仓库时发现一个很有意思的现象很多开发者做 Agent 时前期跑通对话、工具调用都非常顺利一旦进入“让 Agent 记住用户偏好”“在多轮对话里保持状态一致”“跨会话恢复上下文”这类需求时就开始抓瞎了。不是把聊天记录丢给大模型就是记忆也不是在 Prompt 里写一句“请记住用户的喜好”就能生效。真正的企业级 Agent 记忆系统需要解决的是三个层面的问题短期记忆怎么管理上下文、长期记忆怎么在需要时被准确唤醒、多个 Agent 之间怎么共享和隔离记忆。这篇文章基于 LangChain LangGraph DeepAgents 这套组合完整拆解如何从零搭建一个企业级的 Agent 记忆系统并给出一个电商场景下的实战案例。如果你正在做 Agent 开发、RAG 应用或者正在调研 Agent 记忆方案这篇文章值得收藏。1. 为什么 Agent 记忆系统这么难做又为什么非做不可先说一个核心判断记忆系统真正难的地方不是“把更多东西存下来”而是让 Agent 学会“回忆”。很多团队做 Agent 记忆时第一反应是“上向量数据库”把用户历史对话全部切块、 embedding、存入向量库然后每次请求都检索 Top-K 相关内容塞进 Prompt。这套路看起来没毛病但实际跑起来会发现几个问题检索噪声大。用户聊了 50 轮你检索出 5 段最相似的文本结果 4 段都是无关紧要的寒暄真正重要的偏好信息反而没被召回。上下文窗口浪费严重。每次把大量历史记录塞进 Prompt既增加 Token 成本又稀释了模型对当前问题的注意力。短期记忆和长期记忆混为一谈。对话轮次内的状态、用户跨会话的长期偏好、系统运行过程中的业务数据这三类信息的时效性、存储方式、更新策略完全不同不能用一个方案一把梭。多 Agent 场景下记忆隔离与共享没有设计。企业级系统不是一个 Agent 打天下而是多个 Agent 协作比如客服 Agent、推荐 Agent、订单 Agent它们之间哪些记忆该共享、哪些该隔离必须有清晰的架构设计。为什么这套问题值得现在认真研究因为 Agent 从“能用”到“好用”记忆系统是绕不开的一道坎。目前的 Agent 框架比如 LangChain、LangGraph本质上解决的是“Agent 如何思考、如何调用工具、如何组织流程”DeepAgents 解决的是“多个 Agent 如何分工协作”而记忆系统解决的是“Agent 如何基于历史经验做出更个性化的反应”。三者叠加才是企业级落地的完整形态。记忆系统 状态管理短期 知识沉淀长期 个性化上下文用户维度2. LangChain、LangGraph、DeepAgents 三者的分工与关系在实际动手之前先把这套技术栈中三个核心组件的关系理清楚。很多初学者容易把 LangChain 和 LangGraph 搞混看到 DeepAgents 又不知道它属于哪个层次。2.1 LangChain组件生态与工具链LangChain 是 Agent 开发中最基础的框架它提供的是“积木”模型封装统一调用 OpenAI、Claude、国产大模型等。Prompt 管理模板化、变量化。工具调用Function Calling、HTTP 工具、代码解释器等。RAG 组件文档加载、切分、向量化、检索。在企业级记忆系统中LangChain 主要负责跟模型和向量数据库打交道。它本身不关心“对话状态怎么流转”“记忆怎么写入”这些是上游编排层的事。2.2 LangGraph状态化工作流编排LangGraph 是 LangChain 官方推出的图编排框架核心能力是把 Agent 的推理过程建模成一张有向图每个节点是一个处理步骤调用模型、调用工具、写记忆、读记忆节点之间通过状态State传递数据。相比传统的 Chain 串行调用LangGraph 的核心优势有三个支持循环Agent 可以反复“思考 → 调用工具 → 观察结果 → 再思考”直到完成任务。支持条件分支不同情况下走不同路径而不是一条路走到黑。内置 Checkpoint 机制可以把每一步的状态持久化下来这恰恰是短期记忆的基础设施。用一句话总结LangGraph 管的是“Agent 中枢神经系统”——什么时候调用模型、什么时候检索记忆、什么时候停下。2.3 DeepAgents多 Agent 协同架构DeepAgents 是面向复杂任务的多 Agent 协作框架。它解决的是“单 Agent 搞不定”的问题任务拆解成子任务。不同 Agent 各司其职。通过编排机制汇总结果。在记忆系统里DeepAgents 的价值在于不同角色的 Agent 可以共享同一个记忆底座但各自只读取与自己相关的记忆子集。2.4 三者的关系图用户请求 ↓ LangChain模型调用 Prompt 管理 工具封装 ↓ LangGraph编排流程推理 → 决策 → 记忆读写 → 生成回复 ↓ DeepAgents多 Agent 协作任务拆解 → 子 Agent 执行 → 结果汇总 ↓ 记忆系统底座 ├── 短期记忆LangGraph Checkpoint Thread 状态 ├── 长期记忆向量数据库 实体记忆 摘要记忆 └── 业务记忆业务系统数据库/缓存三者不是替代关系而是分层协作DeepAgents 负责“谁来做”LangGraph 负责“怎么做”LangChain 负责“用什么做”。3. 记忆系统架构设计别再把所有东西塞进一个向量库企业级记忆系统的架构设计核心原则是分类存储、按需召回、分层管理。3.1 记忆的分类根据时效性把记忆分为三类。记忆类型时效性存储介质典型内容更新方式短期记忆工作记忆单次会话内内存 / Checkpoint当前对话上下文、临时状态每轮对话实时更新长期记忆知识记忆跨会话向量数据库 / 图数据库用户偏好、历史订单、业务知识异步/定期写入会话摘要记忆跨会话精简数据库字段 / 向量库上次对话的摘要压缩版历史会话结束时生成很多方案把长期记忆和会话摘要混在一起这是不对的。会话摘要是“压缩后的对话过程”长期记忆是“从对话中提炼出来的事实与偏好”前者用于“想起来上次聊了啥”后者用于“知道这个用户喜欢什么”。3.2 记忆的写入策略记忆不是每轮对话都写也不是等会话结束才写而是事件驱动 异步写入。事件驱动的意思是当 Agent 检测到用户的某个信息具有“持久价值”时才触发写入。比如用户在对话中明确说了“我喜欢白色”“我常用顺丰”“我的预算是一万以内”。Agent 通过意图识别判断用户在表达偏好。Agent 完成了一次订单查询产生了有价值的业务结果。异步写入的意思是写入操作不要阻塞主对话流程。如果用户在等回复时你还在这里做 embedding、写向量库体验会非常差。合理做法是先回复用户再后台异步完成记忆写入。3.3 记忆的读取策略读取记忆时最忌讳的是“全量召回”。企业级系统推荐采用三级召回最近上下文直接读取当前会话最近 N 轮对话短期记忆。相关历史根据当前用户问题做向量检索召回 Top-K 相关长期记忆。全局画像读取用户的基础属性画像Region、偏好、消费能力等这部分数据很小但很关键。三级召回的结果合并后经过一个“去重 重排”的步骤再拼接到 Prompt 里。4. 企业级 Agent 记忆系统环境准备开始写代码之前先把环境准备好。4.1 运行环境操作系统Linux / macOS / Windows建议 Linux 或 macOS部署到生产环境也更友好。Python3.10 及以上。包管理pip 或 poetry。4.2 安装依赖pip install langchain langgraph langchain-openai chromadb deepagents这里说明一下langgraph现在已经作为独立包发布不再依赖langchain的旧版本。如果你用的是 LangChain 0.1 以上版本两者可以共存。deepagents是本文要演示的多 Agent 框架具体包名以官方文档为准。4.3 环境变量配置在项目根目录创建.env文件OPENAI_API_KEYsk-xxxx OPENAI_BASE_URLhttps://api.openai.com/v1如果你的环境需要走代理网关把OPENAI_BASE_URL换成实际的网关地址即可。注意不要在生产代码里硬编码密钥。4.4 目录结构规划一个干净的项目结构能让后续扩展省很多事agent_memory/ ├── .env ├── requirements.txt ├── app/ │ ├── __init__.py │ ├── main.py # 入口 │ ├── memory/ │ │ ├── __init__.py │ │ ├── short_term.py # 短期记忆模块 │ │ ├── long_term.py # 长期记忆模块 │ │ └── summary.py # 会话摘要模块 │ ├── agents/ │ │ ├── __init__.py │ │ ├── customer_service.py # 客服 Agent │ │ ├── recommender.py # 推荐 Agent │ │ └── order.py # 订单 Agent │ └── utils/ │ ├── __init__.py │ └── logger.py5. 核心代码实现从短期记忆到长期记忆理论说完直接上代码。这一部分会给出一个可以直接跑通的完整示例。5.1 短期记忆基于 LangGraph 的 Checkpoint 机制短期记忆的本质是“对话状态保持”。LangGraph 已经内置了 Checkpoint 机制可以把每一步的状态持久化到 SQLite 或 Postgres。# 文件路径app/memory/short_term.py from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.sqlite import SqliteSaver from typing import TypedDict, Annotated import sqlite3 # 定义对话状态结构 class AgentState(TypedDict): messages: Annotated[list, 对话消息列表] user_id: str session_id: str from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage # 初始化模型 llm ChatOpenAI(modelgpt-4o-mini, temperature0.7) # 定义节点函数处理用户输入并生成回复 def chat_node(state: AgentState): messages state[messages] response llm.invoke(messages) # 将新的 AI 回复追加到消息列表 new_messages messages [response] return {messages: new_messages} # 构建状态图 graph_builder StateGraph(AgentState) graph_builder.add_node(chat, chat_node) graph_builder.add_edge(START, chat) graph_builder.add_edge(chat, END) # 使用 SQLite 保存检查点 conn sqlite3.connect(memory.db, check_same_threadFalse) memory SqliteSaver(conn) # 编译图 graph graph_builder.compile(checkpointermemory) # 模拟用户会话 config {configurable: {thread_id: session-001}} user_input HumanMessage(content你好我喜欢白色和简约风格的家具) result graph.invoke( {messages: [user_input], user_id: U1001, session_id: session-001}, configconfig ) print(result[messages][-1].content)关键逻辑说明AgentState定义了对话中需要传递的状态结构messages是核心。SqliteSaver是 LangGraph 内置的检查点持久化实现会把每一轮的状态写入 SQLite。thread_id是短期记忆的关键同一个thread_id的会话共享上下文不同thread_id之间完全隔离。这个机制对应到企业场景就是用户打开了客服聊天窗口我们给他分配一个thread_id他说的每一句话、Agent 的每一次回复、调用的每一个工具结果都会自动保存。即使用户刷新浏览器重新发起请求只要thread_id不变Agent 就能“记得”之前聊了什么。5.2 长期记忆向量存储 实体记忆长期记忆解决的是“跨会话记住用户偏好”。这里采用“向量库 实体记忆”双轨方案向量库存储对话片段用于语义检索。实体记忆存储结构化偏好用于精确查询。# 文件路径app/memory/long_term.py from langchain_openai import OpenAIEmbeddings from langchain_chroma import Chroma from langchain_core.documents import Document import uuid # 初始化向量数据库 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) vector_store Chroma( collection_nameuser_memory, embedding_functionembeddings, persist_directory./chroma_db ) def write_memory(user_id: str, content: str, metadata: dict None): 写入一条长期记忆 doc Document( page_contentcontent, metadata{ user_id: user_id, timestamp: datetime.utcnow().isoformat(), **(metadata or {}) } ) vector_store.add_documents([doc], ids[str(uuid.uuid4())]) def search_memory(user_id: str, query: str, top_k: int 5): 从长期记忆中召回相关内容 results vector_store.similarity_search_with_score( query, ktop_k, filter{user_id: user_id} ) return results这段代码做的事情很简单write_memory把用户的重要信息写入向量库search_memory按用户过滤后做相似度检索。这里必须注意filter{user_id: user_id}不加这个过滤会导致所有用户的记忆互相污染这是做记忆系统最容易犯的错误之一。5.3 会话摘要记忆压缩历史保留精华长会话跑几十分钟后上下文窗口迟早会爆。业界通用方案是当对话轮次达到阈值时把已有对话做一次摘要然后把摘要作为新的上下文起点。# 文件路径app/memory/summary.py from langchain_openai import ChatOpenAI from langchain_core.messages import SystemMessage, HumanMessage SUMMARY_PROMPT 请将以下对话内容压缩为一段简洁的摘要保留 1. 用户提到的关键偏好和事实信息 2. 当前任务的进展状态 3. 需要后续继续处理的事项 不要保留寒暄、重复表达和无意义内容。 def summarize_messages(messages: list) - str: llm ChatOpenAI(modelgpt-4o-mini, temperature0.3) serialized \n.join( f{用户 if msg.type human else 助手}: {msg.content} for msg in messages ) response llm.invoke([ SystemMessage(contentSUMMARY_PROMPT), HumanMessage(contentserialized) ]) return response.content这个摘要的价值是当对话超过 20 轮时使用摘要 最近 5 轮完整消息作为上下文既保留了对整体任务的理解又控制住了 Token 成本。5.4 DeepAgents 多 Agent 集成上面实现了单 Agent 的记忆能力而 DeepAgents 让我们可以在多 Agent 场景下复用这套记忆系统。# 文件路径app/agents/order.py from langchain_core.tools import tool tool def get_user_orders(user_id: str) - str: 查询用户的历史订单信息。当用户询问我买了什么我的订单时使用。 # 假设这里是一个真实的订单服务 API return f用户 {user_id} 最近有 3 笔订单最新一笔是白色简约风格沙发金额 2399 元。 # 文件路径app/agents/customer_service.py from langchain_core.tools import tool tool def get_user_preference(user_id: str) - str: 读取用户长期记忆中的偏好信息。当需要个性化推荐或回答偏好问题时使用。 from app.memory.long_term import search_memory results search_memory(user_id, 用户偏好, top_k3) if not results: return 暂无偏好记录 return \n.join([doc.page_content for doc, _ in results])在 DeepAgents 中每个子 Agent 拥有独立的 Tool 权限。客服 Agent 可以调用get_user_preference但不需要查订单的 Tool订单 Agent 可以查订单但不需要读偏好。这样既实现了记忆共享又做到了权限隔离。6. 电商场景完整实战让 Agent 真正“记住”用户技术框架搭好了现在用一个电商售前客服 推荐场景串联全部记忆能力。6.1 场景设定用户小 A 是个对家居品质有要求的人第一次访问网站时和客服 Agent 聊了 10 分钟明确表达了自己的偏好喜欢白色、简约风格。预算每件家具 2000-3000 元之间。正在装修客厅。特别讨厌推销电话喜欢第一种方案网站浏览时不弹窗打扰。第二次访问时小 A 直接问“上次说的那款白色沙发现在还有优惠吗”如果 Agent 没有记忆系统它会一脸茫然。现在看我们如何一步步处理。6.2 第一轮会话写入长期记忆用户第一次聊天结束时触发记忆写入# 在 LangGraph 的结束节点中调用 from app.memory.long_term import write_memory def end_node(state: AgentState): # 从对话历史中提取用户偏好 preference 用户喜欢白色简约风格家具预算每件2000-3000元正在装修客厅 write_memory( user_idstate[user_id], contentpreference, metadata{type: preference, source: chat} ) return state6.3 第二轮会话短期记忆 长期记忆联合召回用户第二次访问请求进入 LangGraph 后先做记忆召回def retrieve_memory_node(state: AgentState): # 短期记忆自动从 Checkpoint 恢复 # 长期记忆根据当前问题检索 query state[messages][-1].content long_term_memories search_memory(state[user_id], query, top_k3) # 将记忆拼接到系统提示中 memory_context \n.join([doc.page_content for doc, _ in long_term_memories]) system_prompt f你是家居卖场的智能客服。以下是该用户的历史记忆信息 {memory_context} 请基于记忆信息结合用户当前问题给出个性化回复。 state[system_prompt] system_prompt return state最终 Agent 能准确回答“您上次看中的那款白色简约沙发目前参与满 3000 减 300 的活动叠加您会员等级优惠后到手 2399 元。”6.4 多 Agent 协作推荐 Agent 联动用户接着问“我还想看看配套的茶几有推荐吗”这里不需要用户重新描述偏好因为推荐 Agent 直接从记忆底座读取# 推荐 Agent 内 from app.memory.long_term import search_memory preferences search_memory(user_idU1001, query家具风格和预算, top_k5) # 召回结果白色、简约、预算2000-3000、客厅 # 推荐 Agent 基于这些约束过滤商品库返回符合条件的茶几这就是企业级记忆系统的威力一次输入偏好全链路生效。7. 运行结果与效果验证以上代码跑通后可以按下面的方式验证记忆是否正确生效。7.1 验证短期记忆# 第一次对话 python main.py --session session-001 --user U1001 --message 你好 # 第二次对话同一 session python main.py --session session-001 --user U1001 --message 我们刚才聊到哪里了预期结果第二次对话时 Agent 能准确回忆起上一轮的内容。如果返回“我们第一次见面吧”说明 Checkpoint 没有生效检查thread_id是否一致。7.2 验证长期记忆# 第一次会话写入偏好 python main.py --session session-001 --user U1001 --message 我喜欢白色简约风格预算3000以内 # 新会话测试召回 python main.py --session session-002 --user U1001 --message 给我推荐一款合适的家具预期结果新会话中 Agent 应该提到“根据您喜欢的白色简约风格”之类的个性化内容。如果 Agent 像失忆一样去检查filter{user_id: user_id}是否生效、向量库是否有数据写入。7.3 预期输出示例[用户] 上次说的那款白色沙发现在还有优惠吗 [助手] 有的呢您之前看过的那款白色简约风格沙发目前参加满3000减300的活动 结合您的会员优惠后到手价2399元比您上次咨询时便宜了100元。 需要帮您锁定这个优惠价格吗看到这个回答说明短期记忆知道上次聊过沙发、长期记忆白色简约风格、业务数据优惠价格三层记忆全部生效了。8. 常见问题与排查方法在实际开发过程中以下 5 个问题出现频率最高。问题现象可能原因排查方式解决方案同一 Session 两次对话Agent 不记得上下文thread_id不一致或 Checkpoint 未配置打印请求中的thread_id检查 SQLite 是否有数据统一使用用户会话 ID 作为thread_id长期记忆检索结果和当前问题无关向量化模型效果差或没有做相似度阈值过滤打印检索结果和 score观察是否存在低分噪声增加 score 阈值比如只保留 0.7 以上的结果多个用户的记忆互相串线查询时没有按user_id过滤检查检索代码中的 filter 条件所有读取操作强制加入user_id过滤Token 成本过高每次把全部历史都塞进 Prompt查看请求日志中 messages 的长度引入会话摘要机制压缩早期冗余对话Agent 回复时出现“记忆幻觉”把检索到的记忆当成事实未做校验对比记忆条目的元数据和实际业务数据重要记忆写入时增加置信度字段读取时优先采信高置信度记忆8.1 记忆写入失败的隐蔽坑还有一个容易被忽略的坑当你向向量库写入记忆时如果同时有多个 Agent 在写会出现并发冲突。Chroma 这类本地向量库需要加锁或者改用支持并发写入的数据库如 PostgreSQL pgvector / Milvus。# 生产环境推荐使用支持并发的向量库 from langchain_postgres import PGVector from sqlalchemy import create_engine engine create_engine(postgresql://user:passwordlocalhost/agent_memory) vector_store PGVector( embeddingsembeddings, connectionengine, collection_nameuser_memory )9. 生产环境最佳实践与工程建议9.1 记忆的写入要克制不是所有对话内容都值得写入长期记忆。建议定义“记忆价值判定规则”包含明确偏好“我喜欢”“我习惯”“我通常”。包含事实信息订单号、地址、联系方式。与当前业务目标强相关装修中、备孕、在找工作。其余内容让短期记忆自然滚动淘汰即可。记忆库的噪声越低召回精度越高。9.2 用户隐私与数据安全企业级系统必须考虑隐私合规记忆数据加密存储敏感字段脱敏。用户有权查看和删除自己的记忆数据需要提供接口。记忆写入和读取都需要权限控制防止越权。# 删除用户记忆的接口示例 def delete_user_memory(user_id: str): # 从向量库中删除该 user_id 的所有记录 vector_store.delete(where{user_id: user_id})9.3 记忆质量的监控与评估上线后要持续监控记忆系统的效果建议关注三个指标记忆召回率在测试集中模拟用户多轮提问统计 Agent 是否能正确用到历史信息。记忆误用率Agent 是否在错误的场景下使用了历史记忆比如用户只是随口说说的偏好被当成长期事实。记忆写入延迟异步写入的平均耗时是否影响主流程性能。9.4 版本兼容性提示LangChain、LangGraph 目前版本迭代较快API 变更是家常便饭。本文代码基于主流的langchain0.1和langgraph0.0.60版本编写。如果你遇到的 API 和文中不一致优先查看官方迁移文档。生产环境建议锁定版本pip freeze | grep langchain requirements.txt pip freeze | grep langgraph requirements.txt pip freeze | grep deepagents requirements.txt10. 总结与后续学习方向这篇文章的核心内容可以归纳为四句话第一Agent 记忆系统不是“聊天记录 向量检索”那么简单它需要分层设计短期记忆管状态长期记忆管偏好摘要记忆管压缩。第二LangGraph 的 Checkpoint 机制是短期记忆的最佳基础设施之一thread_id的设计直接决定了多会话隔离与共享的能力边界。第三长期记忆的工程难点不在存储而在写入策略和召回精度。写入要克制召回要过滤目标是让 Agent 在需要时“回忆起”正确的事情而不是“检索出”一堆相似文本。第四DeepAgents 多 Agent 架构下记忆系统要作为独立底座存在各 Agent 通过工具按需读取做好权限隔离。如果你准备继续深入建议按这个路线往下走本地向量库换成熟的向量数据库Milvus、Qdrant、pgvector性能差距很大。给长期记忆增加遗忘曲线机制不常访问的记忆自动降级保持记忆库“鲜活”。尝试引入知识图谱存储实体关系记忆尤其在电商场景中“用户-商品-偏好”的关系很适合图结构。结合 RAG 让记忆系统能访问企业知识库实现从“记住用户”到“记住企业业务”的升级。我把整套代码拆成了可以快速改造成自己项目的最小结构建议你拿到代码后先把单会话跑通再加上长期记忆最后再上多 Agent。每一步验证完再进下一步比一口气写完再 Debug 要高效得多。