在构建 Agent 项目的过程中很多人会遇到同一个坎单轮对话里 Agent 表现得非常聪明工具调用、代码生成、知识库问答都能完成但只要进入多轮对话或跨会话场景Agent 就像“失忆”一样忘记用户刚说过的话、偏好甚至业务上下文。最近我在一个企业级电商客服项目中反复调试 Agent 记忆模块踩了不少坑也整理出一套完整的实现思路。本文将基于 Langchain、Langgraph 和 DeepAgents 三种技术栈从短期记忆、长期记忆到电商场景实战完整拆解一套企业级 Agent 记忆系统的设计方法新手可以先看概念有经验的开发者可以直接跳到代码和排错部分。1. 为什么要给 Agent 装上记忆系统1.1 Agent 没有记忆时的真实状况假设我们要做一个电商导购 Agent用户第一轮说“我想给女朋友买一瓶适合干皮的香水预算 500 以内”第二轮说“上次说的那个有没有玫瑰味的”。如果没有记忆机制Agent 在第二轮根本不知道“上次说的那个”指的是什么它必须重新向用户确认预算、肤质、使用场景甚至可能重新推荐完全不符合要求的商品。这种体验在问答机器人时代还能忍受但在需要连续决策的 Agent 系统里基本等于不可用。不只是多轮对话跨会话身份识别同样重要。用户上周浏览过某款商品但没有下单今天再次打开对话时Agent 最好能记得用户的历史兴趣、历史订单和沟通偏好。这就是记忆系统要解决的核心问题让 Agent 在时间维度上具备连续性。1.2 短期记忆与长期记忆的边界很多人对“记忆系统”的理解是“把更多上下文塞进 Prompt”这是最常见的误区。我们先建立两个基础概念。短期记忆Short-term Memory / Working Memory指的是当前会话内、短时间内需要保持的信息。例如对话历史、用户刚输入的内容、正在处理的中间状态。它一般随着会话结束或超时被清理特点是读写速度快、容量小、不需要持久化。长期记忆Long-term Memory则是指跨会话、跨时间窗口仍然需要保留的信息。例如用户画像、偏好标签、历史订单、售后记录、业务规则的个人化适配。长期记忆需要持久化存储并且要进行结构化建模否则时间一长就会变成一堆无法召回的信息垃圾。在企业级系统中短期记忆和长期记忆并不是两条独立的链路而是像一个“两级缓存”短期记忆承接当前任务长期记忆为每次新会话提供初始化上下文。1.3 记忆系统在业务中的价值从业务角度看记忆系统至少有四个直接价值减少用户重复输入提升交互效率。让 Agent 的推荐和决策更有个性而不是千人一面。在多轮工具调用和任务拆解中保持状态一致避免中间过程丢失。为企业沉淀可复用的用户画像知识支持后续的数据分析和精准运营。这也是为什么当前主流 Agent 框架都在重点解决“记忆”问题。Langchain 提供的是组件化能力Langgraph 提供的是状态管理能力而 DeepAgents 这类深度 Agent 框架则在“规划—执行—回顾”的循环中反复使用记忆三者配合起来恰好能组成一套完整的企业级记忆方案。2. 技术选型Langchain、Langgraph、DeepAgents 如何分工2.1 Langchain组件生态与工具集成Langchain 是目前使用最广泛的 LLM 应用开发框架之一。它对大模型调用、Prompt 模板、输出解析、工具调用、向量库接入等做了大量封装让开发者可以快速组合出一条 AI 应用链路。在记忆系统里Langchain 主要负责两件事提供模型调用、向量化、嵌入等基础组件提供记忆相关的工具类比如基于字符串的对话历史、基于摘要的对话历史以及多种向量存储的封装。但 Langchain 本身是一个“链式”思维它更适合线性流程。当我们的 Agent 流程开始出现分支、循环、条件判断甚至需要人在回路时链式结构会变得很难维护。2.2 Langgraph有状态图编排Langgraph 是 Langchain 团队推出的图状态编排框架。它的核心模型是“状态图”每一个节点就是一个处理步骤节点之间通过边连接整体状态由一个全局 State 对象维护。Langgraph 和 Langchain 的核心区别在于维度LangchainLanggraph流程模型线性 Chain 为主图状态编排支持分支、循环状态管理链路间通过参数传递全局 State天然适合持久化记忆能力依赖外部 Memory 类状态快照 Checkpoint更接近系统级使用场景简单问答、单轮工具调用多轮 Agent、复杂工作流对于记忆系统来说Langgraph 最有价值的能力是状态持久化。它内置了 Checkpoint 机制可以在每个节点执行后保存一份状态快照。这意味着即使进程重启只要把 Checkpoint 文件或数据库持久化下来Agent 就能恢复到之前的状态。2.3 DeepAgents深度执行与自主规划DeepAgents 是“深度 Agent”方向的框架。它和普通 Agent 的差异在于普通 Agent 往往是一个模型 一组工具按一次调用直接返回结果而 DeepAgents 更强调任务拆解、多级规划、工具协同、结果反思和记忆复用。为什么记忆系统要引入 DeepAgents因为企业级场景中的任务往往是长周期、多步骤的。比如“帮用户对比 3 款手机按性价比排序并生成推荐报告”这个过程需要多次调用搜索工具、多次读取用户历史偏好、然后汇总结果。DeepAgents 可以把这类任务拆成多个子节点执行并在每个阶段读取短期记忆中的中间结果同时把结果中值得沉淀的信息写回长期记忆。这里需要特别提醒DeepAgents 目前仍属于演进比较快的方向不同版本、不同官方实现的 API 可能有差异。本文在代码演示中会把 DeepAgents 抽象成接口层方便你替换成自己实际使用的实现。2.4 三者的协同关系我的建议是Langchain 负责“底座”提供大模型、向量数据库、工具等基础设施Langgraph 负责“编排”用 StateGraph 管理 Agent 的状态流和 CheckpointDeepAgents 负责“深度执行”在 Langgraph 的状态框架内承担规划、反思等复杂 Agent 行为。简单画一个逻辑关系Langchain 是材料库Langgraph 是施工图纸DeepAgents 是执行团队。材料由图纸调度团队按图纸施工而记忆系统则是贯穿期间的“项目档案室”。3. 环境准备与项目结构3.1 运行环境与依赖本文的示例代码使用 Python 实现。文章中使用的库建议安装如下pip install langchain pip install langchain-core pip install langchain-openai pip install langgraph pip install langgraph-checkpoint pip install chromadb pip install openai版本需要根据你的项目实际情况调整。Langchain 和 Langgraph 的版本迭代比较快安装时建议锁定一个稳定版本避免因为 API 变动造成示例不兼容。本文重点演示设计思路具体 API 请以你本地的版本为准。如果你使用的是国内的大模型平台或其他国外模型请将模型调用部分替换成对应的 Langchain 模型封装类整体架构并不受影响。3.2 项目目录结构为了保证代码可读我们按模块拆分项目agent_memory_system/ ├── config.py # 全局配置模型、向量库、存储路径 ├── models.py # 数据模型订单、偏好、记忆单元 ├── memory/ │ ├── __init__.py │ ├── short_term.py # 短期记忆会话状态维护 │ ├── long_term.py # 长期记忆向量库 用户画像 │ └── recall.py # 记忆召回检索 解析 ├── agents/ │ ├── __init__.py │ ├── planner.py # DeepAgents 规划器抽象 │ ├── executor.py # 执行器调用工具/模型 │ └── reflector.py # 反思器抽取可沉淀记忆 ├── graph.py # Langgraph 状态图定义 └── main.py # 入口启动 Agent 对话3.3 基础配置创建一个config.py存放模型和向量库相关配置# config.py import os MODEL_NAME os.getenv(MODEL_NAME, gpt-4o-mini) EMBEDDING_MODEL os.getenv(EMBEDDING_MODEL, text-embedding-3-small) # 向量数据库持久化目录 VECTOR_STORE_PATH os.getenv(VECTOR_STORE_PATH, ./data/vector_store) # 用户画像存储位置演示用 JSON 文件生产环境建议换数据库 PROFILE_STORE_PATH os.getenv(PROFILE_STORE_PATH, ./data/profiles) # 短期记忆 Checkpoint 存储 CHECKPOINT_STORE_PATH os.getenv(CHECKPOINT_STORE_PATH, ./data/checkpoints)生产环境中建议使用 Redis 存储短期记忆状态使用 PostgreSQL 或专业的向量数据库存储长期记忆。这里用本地路径和 JSON 文件只是为了便于演示。4. 核心原理从“更多检索”到“学会回忆”4.1 三种常见的记忆实现误区在动手写代码之前先聊聊记忆系统设计的三个常见误区。第一个误区是“把全部历史塞进 Prompt”。这种做法在对话轮次少的时候没问题但随着对话变长Token 消耗会快速膨胀而且模型会被大量无关信息干扰回答质量反而下降。第二个误区是“只做向量检索”。很多人觉得长期记忆 把对话文本切成块存进向量库然后每次语义检索。这个方案能解决一部分“信息召回”问题但记忆不只是“找相似文本”。用户偏好这种强结构化信息靠向量检索很容易丢失。第三个误区是“不区分记忆的时效性”。有些信息是长期有效的比如用户的肤质类型有些信息是临时的比如用户本次想买礼物的对象。如果不加区分临时信息也会污染长期画像。4.2 记忆分层模型一套合理的 Agent 记忆系统我习惯分成四个层级。工作记忆当前轮次的数据比如用户刚刚输入的消息、当前工具返回的结果。短期记忆当前会话内的上下文摘要由 Langgraph 的 State 和 Checkpoint 维护。长期记忆跨会话的持久化知识包括用户画像、偏好标签、历史关键事件。业务记忆与具体业务逻辑绑定的数据比如订单状态、售后单进度。其中业务记忆一般由业务系统提供Agent 记忆系统要做的是“读取”和“引用”而不是“重复存储”。比如用户问订单到哪里了Agent 只需要把用户 ID 传给订单查询工具由工具实时获取不需要在模型上下文里保存整个订单详情。4.3 RippleMem 带来的启发在记忆系统领域现在有一个思路值得关注记忆系统不只是把更多东西检索出来而是让 Agent 学会“回忆”。RippleMem 的核心思想是“记忆涟漪传播”可以把每个记忆看作一个节点节点之间通过关联关系连成网络。当一个记忆被激活时它会像水面上的涟漪一样逐层扩散到相邻的、弱关联的记忆节点从而唤起一段“上下文场景”而不仅仅是返回一段相似文本。这个思路给我的最大启发是长期记忆的召回不是单次向量搜索就能完成的而应该是一次“记忆激活”。先找到核心锚点比如“干皮”再以锚点为中心扩散到相关标签比如“护肤品”“玫瑰味”“500 元以内”最后组装成对当前任务有完整帮助的记忆片段。在我们的代码实现中会用“标签 向量检索 规则过滤”的组合方式来模拟这种记忆激活。5. 实战一基于 Langchain Langgraph 实现短期记忆5.1 设计思路短期记忆的目标是让 Agent 在一次会话内保持状态。Langgraph 的状态图本身具备 State我们只需要把对话消息列表和中间状态都放在 State 中并通过 Checkpoint 持久化即可。这里我们用一个常见的MessageGraph思路整个图只有一个“对话节点”每次调用时接收用户输入读取当前 State 中的历史消息调用大模型然后把新的问答追加进 State。5.2 核心代码先定义短期记忆模块# memory/short_term.py from typing import Annotated, Sequence from langchain_core.messages import BaseMessage from langgraph.graph.message import add_messages from typing_extensions import TypedDict class ShortTermMemoryState(TypedDict): messages: Annotated[Sequence[BaseMessage], add_messages] user_id: str session_id: str这里的关键是add_messages。Langgraph 的add_messages是一个合并函数当新消息产生时它会自动追加到已有消息列表中而不会覆盖旧消息。这样我们就不用手动维护 message 列表。接着定义对话图# graph.py from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver from langchain_core.messages import SystemMessage from config import MODEL_NAME from memory.short_term import ShortTermMemoryState # 初始化模型 llm ChatOpenAI(modelMODEL_NAME, temperature0.3) # 对话节点根据当前状态生成回复 async def chat_node(state: ShortTermMemoryState): messages state[messages] system_prompt SystemMessage(content你是一个企业级智能客服助手请结合上下文回答用户问题。) response await llm.ainvoke([system_prompt] messages) return {messages: [response]} # 构建状态图 def build_chat_graph(): graph StateGraph(ShortTermMemoryState) graph.add_node(chat, chat_node) graph.add_edge(START, chat) graph.add_edge(chat, END) return graph # 使用内存 Checkpoint重启会丢失生产建议使用持久化 Checkpoint checkpointer MemorySaver() app build_chat_graph().compile(checkpointercheckpointer)5.3 运行与验证在main.py中编写运行逻辑# main.py import asyncio from langchain_core.messages import HumanMessage from graph import app async def main(): # 模拟第一次对话 result1 await app.ainvoke( {messages: [HumanMessage(content我想给女朋友买一瓶适合干皮的香水预算500以内)]}, config{configurable: {thread_id: session_001}} ) print(Agent:, result1[messages][-1].content) # 模拟第二轮对话测试短期记忆 result2 await app.ainvoke( {messages: [HumanMessage(content上次说的那个有没有玫瑰味的)]}, config{configurable: {thread_id: session_001}} ) print(Agent:, result2[messages][-1].content) asyncio.run(main())注意这里的thread_id是 Langgraph 的关键参数它相当于会话的“记忆分区标识”。同一个thread_id下的所有 State 会被 Checkpoint 串联起来所以第二轮对话时Agent 能看到第一轮的历史消息。预期效果Agent 在第二轮不会再问“你说的是哪个”而是能基于第一轮的香水推荐进一步筛选玫瑰味。5.4 为什么短期记忆不只需要消息列表很多场景下短期记忆还需要保存中间状态比如“已经查询过的商品列表”“当前正在比价的商品 ID”“是否已经向用户确认过预算”。这些不属于对话消息但对决策很重要。因此建议在 ShortTermMemoryState 中增加业务字段class ShortTermMemoryState(TypedDict): messages: Annotated[Sequence[BaseMessage], add_messages] user_id: str session_id: str # 业务中间状态 candidate_products: list current_intent: str pending_question: str这些字段会在节点之间传递也能被 Checkpoint 持久化。6. 实战二基于 Langgraph DeepAgents 实现长期记忆6.1 长期记忆的整体链路长期记忆的完整链路可以概括为 5 步提取、概要化、结构化、存储、召回。提取从多轮对话中抽取出值得记住的信息例如用户肤质、预算区间、品牌偏好。概要化把一段长对话压缩为精简的记忆描述。结构化将记忆整理成标签、键值或三元组便于后续精确匹配。存储分别写入向量库和结构化画像存储。召回在新会话开始时根据用户 ID 和当前任务目标召回相关记忆。6.2 记忆提取与结构化下面我们用 Pydantic 定义一个记忆单元模型# models.py from pydantic import BaseModel, Field from typing import List, Optional class MemoryUnit(BaseModel): user_id: str memory_type: str Field(description记忆类型preference/order/event/fact) content: str Field(description记忆内容描述) tags: List[str] Field(default_factorylist, description相关标签) importance: float Field(default0.5, description重要性权重 0-1) expire_at: Optional[str] Field(defaultNone, description过期时间None表示长期有效)通过memory_type区分记忆类型tags用于快速匹配importance用于记忆衰减和清理策略。这个模型可以存储到 Redis、MySQL 或向量库中按业务需要选择。6.3 实现 DeepAgents 规划器抽象这里我们需要谨慎说明DeepAgents 是一个方向性的 Agent 框架名具体 API 以官方文档为准。为了演示架构我们定义一个抽象接口生产环境中可以替换为真正的 DeepAgents 实现。# agents/planner.py from langchain_core.messages import BaseMessage class DeepAgentPlanner: DeepAgents 规划器抽象。 实际使用时可以把内部实现替换为官方 DeepAgents 的 Agent 实例。 def __init__(self, llm): self.llm llm async def plan(self, task: str, context: dict) - list[str]: 将复杂任务拆解为多个子步骤。 返回一个字符串列表每个字符串是一个子任务描述。 prompt f 你是一个深度任务规划器。请将以下任务拆解为最多5个子步骤 任务{task} 可用上下文{context} 只输出步骤列表不要解释。 resp await self.llm.ainvoke(prompt) steps [line.strip().lstrip(0123456789.- ) for line in resp.content.split(\n) if line.strip()] return steps这个规划器对应 DeepAgents 的“深度拆解”能力。在实际 DeepAgents 框架中每个子步骤还可能对应一个独立的 Agent并拥有自己的工具集和记忆上下文。6.4 长期记忆的写入与召回下面编写长期记忆管理模块# memory/long_term.py import json import os from typing import List from langchain_openai import OpenAIEmbeddings from langchain_chroma import Chroma from models import MemoryUnit from config import VECTOR_STORE_PATH, PROFILE_STORE_PATH class LongTermMemory: def __init__(self): self.embeddings OpenAIEmbeddings(modeltext-embedding-3-small) os.makedirs(VECTOR_STORE_PATH, exist_okTrue) os.makedirs(PROFILE_STORE_PATH, exist_okTrue) self.vector_store Chroma( collection_namelong_term_memory, embedding_functionself.embeddings, persist_directoryVECTOR_STORE_PATH, ) async def save_memory(self, memory: MemoryUnit): 写入长期记忆同时写向量库和画像文件 # 向量库保存语义信息 await self.vector_store.aadd_texts( texts[memory.content], metadatas[{ user_id: memory.user_id, memory_type: memory.memory_type, importance: memory.importance, expire_at: memory.expire_at or , }], ids[f{memory.user_id}_{memory.tags[0]}_{memory.content[:20]}] ) # 画像文件保存结构化标签 profile_path os.path.join(PROFILE_STORE_PATH, f{memory.user_id}.json) profile {} if os.path.exists(profile_path): with open(profile_path, r, encodingutf-8) as f: profile json.load(f) for tag in memory.tags: profile[tag] memory.content with open(profile_path, w, encodingutf-8) as f: json.dump(profile, f, ensure_asciiFalse, indent2) async def recall_memory(self, user_id: str, query: str, top_k: int 5) - List[MemoryUnit]: 召回记忆结合向量检索 用户画像 docs await self.vector_store.asimilarity_search( query, ktop_k, filter{user_id: user_id} ) units [] for doc in docs: meta doc.metadata units.append(MemoryUnit( user_iduser_id, memory_typemeta.get(memory_type, fact), contentdoc.page_content, tags[], importancefloat(meta.get(importance, 0.5)), )) return units这里展示了两条召回路径向量库用于语义相似召回画像文件用于精确标签匹配。在真正的生产系统中建议把画像放到图数据库或关系型数据库中方便处理复杂的标签组合查询。6.5 在 Langgraph 状态图中接入长期记忆我们在状态图中新增两个节点一个是“回忆节点”在新会话开始时加载长期记忆另一个是“沉淀节点”在当前对话结束后抽取值得沉淀的记忆。# graph.py 追加 from agents.planner import DeepAgentPlanner from memory.long_term import LongTermMemory from langchain_core.messages import SystemMessage, HumanMessage long_term_memory LongTermMemory() planner DeepAgentPlanner(llmllm) async def recall_node(state: ShortTermMemoryState): 对话开始前加载用户长期记忆 user_memories await long_term_memory.recall_memory(state[user_id], query用户偏好) if user_memories: memory_text \n.join([m.content for m in user_memories]) state[messages].insert(0, SystemMessage( contentf以下是对该用户的长期记忆信息请参考但不局限于这些信息\n{memory_text} )) return {messages: state[messages]} async def reflect_node(state: ShortTermMemoryState): 对话结束后抽取并沉淀长期记忆 conversation state[messages] # 在大模型深层抽取之前我们先做基础的关键信息抽取 extract_prompt 从以下对话中抽取出用户的长期偏好、个人特征、重要事件输出结构化JSON # 实际项目中这里会调用 LLM 处理示例从简 return {}注意insert(0, ...)这种操作在 Langgraph 中建议在新的返回值中完成避免直接修改原 State。上面的代码只是演示记忆加载逻辑实际编写时建议构建一个新的消息列表返回。沉淀节点是长期记忆的重要一环。如果只写短期记忆不沉淀那么每次会话结束后所有有价值的用户信息都会丢失长期记忆就没有意义了。沉淀节点的执行策略一般有两种对话结束时对整个会话做一次记忆抽取。对话过程中当检测到强偏好、关键事件时实时抽取。第一种实现简单第二种实时性更好。企业级系统通常会两者结合。7. 实战三电商场景下的完整案例7.1 需求分析现在我们把前面所有模块组合起来实现一个电商导购 Agent。需求如下用户首次访问时明确自己的肤质是“干皮”预算 500 元以内用户退出后再次访问时Agent 能自动看到“干皮”标签不再重复询问当用户提到“上次推荐的那款香水”时Agent 能关联短期记忆中的商品候选列表Agent 可以调用商品检索工具返回符合条件的结果。7.2 数据结构我们在记忆单元基础上增加商品检索工具所需的数据# models.py 追加 from typing import List class Product(BaseModel): product_id: str name: str price: float tags: List[str] skin_type: str7.3 商品工具# agents/executor.py from models import Product # 模拟商品库 PRODUCTS [ Product(product_idP001, name玫瑰之水香水, price480, tags[玫瑰, 清新], skin_type干皮), Product(product_idP002, name木质调淡香水, price520, tags[木质, 沉稳], skin_type中性), Product(product_idP003, name保湿花香香水, price450, tags[花香, 甜润], skin_type干皮), ] def search_products(skin_type: str, max_price: float) - list[Product]: return [p for p in PRODUCTS if p.skin_type skin_type and p.price max_price]这里只是最简单的规则过滤。真实电商场景中应该替换为搜索服务或商品 API 调用。7.4 完整对话流程我们扩展状态图加入工具调用节点# graph.py 追加工具节点 from langchain_core.tools import tool from agents.executor import search_products tool def search_product_tool(skin_type: str, max_price: float) - str: 根据肤质和价格上限搜索商品 products search_products(skin_type, max_price) if not products: return 没有找到匹配的商品 return \n.join([f{p.name}{p.price}元适合{p.skin_type} for p in products])为了让工具被大模型自动调用需要把工具绑定到 LLM 上llm_with_tools llm.bind_tools([search_product_tool])完整状态图# graph.py from langgraph.prebuilt import ToolNode, tools_condition def build_ecommerce_graph(): graph StateGraph(ShortTermMemoryState) # 节点划分 graph.add_node(recall, recall_node) # 加载长期记忆 graph.add_node(chat, chat_with_tools) # 对话 工具决策 graph.add_node(tools, ToolNode([search_product_tool])) # 执行工具 graph.add_node(reflect, reflect_node) # 沉淀长期记忆 # 边关系 graph.add_edge(START, recall) graph.add_edge(recall, chat) graph.add_conditional_edges(chat, tools_condition, {tools: tools, END: END}) graph.add_edge(tools, chat) graph.add_edge(chat, reflect) graph.add_edge(reflect, END) return graph其中chat_with_tools节点与之前chat_node类似区别在于调用llm_with_tools让模型能够自主决定是否调用商品搜索工具。7.5 运行效果演示# main.py 电商案例演示 async def ecommerce_demo(): # 第一次会话用户说明偏好 result await app.ainvoke( { messages: [HumanMessage(content我是干皮想买一瓶500以内的香水推荐一下)], user_id: user_9527, session_id: session_100, }, config{configurable: {thread_id: session_100}} ) print(Agent:, result[messages][-1].content) # 第二次会话模拟用户隔天再次访问 result2 await app.ainvoke( { messages: [HumanMessage(content上次推荐的那款玫瑰味香水还有吗)], user_id: user_9527, session_id: session_200, }, config{configurable: {thread_id: session_200}} ) print(Agent:, result2[messages][-1].content)预期行为分析第一轮系统通过recall_node查询长期记忆。由于是新用户没有历史信息所以 Agent 会主动询问或根据用户输入直接推荐商品。当用户明确说了“干皮”“500 以内”沉淀节点会将这两个关键信息写入长期记忆。第二轮虽然session_id变了thread_id从session_100变为session_200短期记忆已经被隔离但recall_node会根据user_id从长期记忆中拉取“干皮”标签再结合第一轮推荐的玫瑰香水信息Agent 就能准确知道用户说的“上次推荐”是哪一款。7.6 结果说明这个案例虽然简单但它覆盖了记忆系统的三个关键动作短期记忆保持会话内消息、长期记忆沉淀跨会话标签、业务数据读取商品工具调用。在真实项目中无论你接入的是订单系统、会员系统还是商品中台都可以沿用这套架构。8. 常见问题与排查思路在企业级开发里记忆系统出现问题的概率很高。下面整理一些高频问题和排查思路问题现象常见原因解决思路第二轮对话 Agent 不记得上文thread_id未保持一致检查每次请求的configurable.thread_id长期记忆不生效用户 ID 不一致或未传user_id在上游统一登录认证中提取用户 ID向量召回结果不准确记忆文本过短、检索 Query 不清晰为记忆单元增加标签结合结构化匹配历史消息过多导致 Token 超限短期记忆没有裁剪 / 摘要化设置窗口长度超出部分自动摘要需要持久化的记忆写失败向量库连接或磁盘权限问题检查向量库连接和存储目录权限Agent 死循环工具调用工具返回结果缺少退出条件设置最大工具调用次数增加异常分支补充几个小经验thread_id是短期记忆的隔离空间建议使用会话 ID而不是用户 ID。同一个用户可以开多个会话每个会话的短期记忆应该相互隔离。user_id最好从统一的用户体系中获取后端服务只在可信的内网环境中传递不要在公网请求里盲目信任前端传入的 user_id。长期记忆的写入要有去重逻辑否则同样的偏好每次对话都会重复写入。9. 最佳实践与工程建议9.1 记忆裁剪与摘要策略短期记忆不能无限增长。建议设置消息窗口比如保留最近 20 条完整消息更早的消息压缩为一段摘要。Langgraph 中可以在节点内检查len(state[messages])超过阈值时调用 LLM 摘要。也可以使用 Langchain 自带的SummarizationMemory思路但注意状态图环境下需要自己实现消息合并。9.2 记忆衰减与清理长期记忆必须有“生命周期”。有些信息是临时性的比如“用户本周计划购买生日礼物”一周后这条记忆可能就过期了。建议在 MemoryUnit 中设置expire_at定期清理过期记忆对importance较低的记忆进行衰减比如长时间未被命中就降低权重最终自动删除。9.3 隐私与数据安全边界记忆系统存储的是真实用户的个人信息必须重视安全合规敏感数据手机号、地址、身份证号不应明文写入记忆库可以使用脱敏或外部加密存储。用户应有权查看、导出、删除自己的记忆数据。建议提供“记忆管理”接口。记忆召回不要无限制开放给所有 Agent建议按业务域隔离。9.4 可观测性与调试记忆系统是个“黑盒”问题的重灾区。建议每次记忆写入和召回都记录日志包含触发的用户 ID、会话 ID写入的记忆内容与标签召回时的查询语句目标记忆来源。这样当模型使用了过期记忆或错误记忆时才能快速定位是哪一步写入造成的。9.5 架构层面的演进方向如果要从演示版本演进到生产级建议做以下改造将 Chroma 替换为独立的向量数据库服务支持水平扩展。将 JSON 用户画像替换为 Redis 或 PostgreSQL并增加唯一约束。引入消息队列异步执行记忆沉淀避免影响主对话链路延迟。为记忆系统增加版本管理记忆结构升级时方便做数据迁移。10. 总结与学习路线本文围绕 Agent 记忆系统从概念到代码完成了三件事先用 Langgraph 的实现搞定了短期记忆用 Langchain 向量库 用户画像的方式实现了长期记忆再引入 DeepAgents 层的规划器抽象把记忆系统接入到电商导购场景中。核心思路是短期记忆靠状态图 Checkpoint长期记忆靠“提取 结构化 向量召回”而不是简单地把更多文本塞进 Prompt。接下来你可以继续深入研究几个方向一是 Langgraph 的 Checkpoint 持久化尝试把状态保存到 Redis 或数据库中二是 RippleMem 这类“记忆涟漪”思想用图网络改造记忆召回逻辑三是 DeepAgents 的完整官方文档把规划器替换成真正的深度 Agent 执行链路四是结合业务系统把订单、客服工单等实时业务数据接入 Agent 记忆。如果你在实际搭建过程中遇到奇怪的报错或者对记忆分层有不同的理解欢迎在评论区留言讨论。动手跑通一遍比读十篇概念文章都管用。 Agent记忆系统实战：Langchain+Langgraph+DeepAgents实现短期与长期记忆
在构建 Agent 项目的过程中很多人会遇到同一个坎单轮对话里 Agent 表现得非常聪明工具调用、代码生成、知识库问答都能完成但只要进入多轮对话或跨会话场景Agent 就像“失忆”一样忘记用户刚说过的话、偏好甚至业务上下文。最近我在一个企业级电商客服项目中反复调试 Agent 记忆模块踩了不少坑也整理出一套完整的实现思路。本文将基于 Langchain、Langgraph 和 DeepAgents 三种技术栈从短期记忆、长期记忆到电商场景实战完整拆解一套企业级 Agent 记忆系统的设计方法新手可以先看概念有经验的开发者可以直接跳到代码和排错部分。1. 为什么要给 Agent 装上记忆系统1.1 Agent 没有记忆时的真实状况假设我们要做一个电商导购 Agent用户第一轮说“我想给女朋友买一瓶适合干皮的香水预算 500 以内”第二轮说“上次说的那个有没有玫瑰味的”。如果没有记忆机制Agent 在第二轮根本不知道“上次说的那个”指的是什么它必须重新向用户确认预算、肤质、使用场景甚至可能重新推荐完全不符合要求的商品。这种体验在问答机器人时代还能忍受但在需要连续决策的 Agent 系统里基本等于不可用。不只是多轮对话跨会话身份识别同样重要。用户上周浏览过某款商品但没有下单今天再次打开对话时Agent 最好能记得用户的历史兴趣、历史订单和沟通偏好。这就是记忆系统要解决的核心问题让 Agent 在时间维度上具备连续性。1.2 短期记忆与长期记忆的边界很多人对“记忆系统”的理解是“把更多上下文塞进 Prompt”这是最常见的误区。我们先建立两个基础概念。短期记忆Short-term Memory / Working Memory指的是当前会话内、短时间内需要保持的信息。例如对话历史、用户刚输入的内容、正在处理的中间状态。它一般随着会话结束或超时被清理特点是读写速度快、容量小、不需要持久化。长期记忆Long-term Memory则是指跨会话、跨时间窗口仍然需要保留的信息。例如用户画像、偏好标签、历史订单、售后记录、业务规则的个人化适配。长期记忆需要持久化存储并且要进行结构化建模否则时间一长就会变成一堆无法召回的信息垃圾。在企业级系统中短期记忆和长期记忆并不是两条独立的链路而是像一个“两级缓存”短期记忆承接当前任务长期记忆为每次新会话提供初始化上下文。1.3 记忆系统在业务中的价值从业务角度看记忆系统至少有四个直接价值减少用户重复输入提升交互效率。让 Agent 的推荐和决策更有个性而不是千人一面。在多轮工具调用和任务拆解中保持状态一致避免中间过程丢失。为企业沉淀可复用的用户画像知识支持后续的数据分析和精准运营。这也是为什么当前主流 Agent 框架都在重点解决“记忆”问题。Langchain 提供的是组件化能力Langgraph 提供的是状态管理能力而 DeepAgents 这类深度 Agent 框架则在“规划—执行—回顾”的循环中反复使用记忆三者配合起来恰好能组成一套完整的企业级记忆方案。2. 技术选型Langchain、Langgraph、DeepAgents 如何分工2.1 Langchain组件生态与工具集成Langchain 是目前使用最广泛的 LLM 应用开发框架之一。它对大模型调用、Prompt 模板、输出解析、工具调用、向量库接入等做了大量封装让开发者可以快速组合出一条 AI 应用链路。在记忆系统里Langchain 主要负责两件事提供模型调用、向量化、嵌入等基础组件提供记忆相关的工具类比如基于字符串的对话历史、基于摘要的对话历史以及多种向量存储的封装。但 Langchain 本身是一个“链式”思维它更适合线性流程。当我们的 Agent 流程开始出现分支、循环、条件判断甚至需要人在回路时链式结构会变得很难维护。2.2 Langgraph有状态图编排Langgraph 是 Langchain 团队推出的图状态编排框架。它的核心模型是“状态图”每一个节点就是一个处理步骤节点之间通过边连接整体状态由一个全局 State 对象维护。Langgraph 和 Langchain 的核心区别在于维度LangchainLanggraph流程模型线性 Chain 为主图状态编排支持分支、循环状态管理链路间通过参数传递全局 State天然适合持久化记忆能力依赖外部 Memory 类状态快照 Checkpoint更接近系统级使用场景简单问答、单轮工具调用多轮 Agent、复杂工作流对于记忆系统来说Langgraph 最有价值的能力是状态持久化。它内置了 Checkpoint 机制可以在每个节点执行后保存一份状态快照。这意味着即使进程重启只要把 Checkpoint 文件或数据库持久化下来Agent 就能恢复到之前的状态。2.3 DeepAgents深度执行与自主规划DeepAgents 是“深度 Agent”方向的框架。它和普通 Agent 的差异在于普通 Agent 往往是一个模型 一组工具按一次调用直接返回结果而 DeepAgents 更强调任务拆解、多级规划、工具协同、结果反思和记忆复用。为什么记忆系统要引入 DeepAgents因为企业级场景中的任务往往是长周期、多步骤的。比如“帮用户对比 3 款手机按性价比排序并生成推荐报告”这个过程需要多次调用搜索工具、多次读取用户历史偏好、然后汇总结果。DeepAgents 可以把这类任务拆成多个子节点执行并在每个阶段读取短期记忆中的中间结果同时把结果中值得沉淀的信息写回长期记忆。这里需要特别提醒DeepAgents 目前仍属于演进比较快的方向不同版本、不同官方实现的 API 可能有差异。本文在代码演示中会把 DeepAgents 抽象成接口层方便你替换成自己实际使用的实现。2.4 三者的协同关系我的建议是Langchain 负责“底座”提供大模型、向量数据库、工具等基础设施Langgraph 负责“编排”用 StateGraph 管理 Agent 的状态流和 CheckpointDeepAgents 负责“深度执行”在 Langgraph 的状态框架内承担规划、反思等复杂 Agent 行为。简单画一个逻辑关系Langchain 是材料库Langgraph 是施工图纸DeepAgents 是执行团队。材料由图纸调度团队按图纸施工而记忆系统则是贯穿期间的“项目档案室”。3. 环境准备与项目结构3.1 运行环境与依赖本文的示例代码使用 Python 实现。文章中使用的库建议安装如下pip install langchain pip install langchain-core pip install langchain-openai pip install langgraph pip install langgraph-checkpoint pip install chromadb pip install openai版本需要根据你的项目实际情况调整。Langchain 和 Langgraph 的版本迭代比较快安装时建议锁定一个稳定版本避免因为 API 变动造成示例不兼容。本文重点演示设计思路具体 API 请以你本地的版本为准。如果你使用的是国内的大模型平台或其他国外模型请将模型调用部分替换成对应的 Langchain 模型封装类整体架构并不受影响。3.2 项目目录结构为了保证代码可读我们按模块拆分项目agent_memory_system/ ├── config.py # 全局配置模型、向量库、存储路径 ├── models.py # 数据模型订单、偏好、记忆单元 ├── memory/ │ ├── __init__.py │ ├── short_term.py # 短期记忆会话状态维护 │ ├── long_term.py # 长期记忆向量库 用户画像 │ └── recall.py # 记忆召回检索 解析 ├── agents/ │ ├── __init__.py │ ├── planner.py # DeepAgents 规划器抽象 │ ├── executor.py # 执行器调用工具/模型 │ └── reflector.py # 反思器抽取可沉淀记忆 ├── graph.py # Langgraph 状态图定义 └── main.py # 入口启动 Agent 对话3.3 基础配置创建一个config.py存放模型和向量库相关配置# config.py import os MODEL_NAME os.getenv(MODEL_NAME, gpt-4o-mini) EMBEDDING_MODEL os.getenv(EMBEDDING_MODEL, text-embedding-3-small) # 向量数据库持久化目录 VECTOR_STORE_PATH os.getenv(VECTOR_STORE_PATH, ./data/vector_store) # 用户画像存储位置演示用 JSON 文件生产环境建议换数据库 PROFILE_STORE_PATH os.getenv(PROFILE_STORE_PATH, ./data/profiles) # 短期记忆 Checkpoint 存储 CHECKPOINT_STORE_PATH os.getenv(CHECKPOINT_STORE_PATH, ./data/checkpoints)生产环境中建议使用 Redis 存储短期记忆状态使用 PostgreSQL 或专业的向量数据库存储长期记忆。这里用本地路径和 JSON 文件只是为了便于演示。4. 核心原理从“更多检索”到“学会回忆”4.1 三种常见的记忆实现误区在动手写代码之前先聊聊记忆系统设计的三个常见误区。第一个误区是“把全部历史塞进 Prompt”。这种做法在对话轮次少的时候没问题但随着对话变长Token 消耗会快速膨胀而且模型会被大量无关信息干扰回答质量反而下降。第二个误区是“只做向量检索”。很多人觉得长期记忆 把对话文本切成块存进向量库然后每次语义检索。这个方案能解决一部分“信息召回”问题但记忆不只是“找相似文本”。用户偏好这种强结构化信息靠向量检索很容易丢失。第三个误区是“不区分记忆的时效性”。有些信息是长期有效的比如用户的肤质类型有些信息是临时的比如用户本次想买礼物的对象。如果不加区分临时信息也会污染长期画像。4.2 记忆分层模型一套合理的 Agent 记忆系统我习惯分成四个层级。工作记忆当前轮次的数据比如用户刚刚输入的消息、当前工具返回的结果。短期记忆当前会话内的上下文摘要由 Langgraph 的 State 和 Checkpoint 维护。长期记忆跨会话的持久化知识包括用户画像、偏好标签、历史关键事件。业务记忆与具体业务逻辑绑定的数据比如订单状态、售后单进度。其中业务记忆一般由业务系统提供Agent 记忆系统要做的是“读取”和“引用”而不是“重复存储”。比如用户问订单到哪里了Agent 只需要把用户 ID 传给订单查询工具由工具实时获取不需要在模型上下文里保存整个订单详情。4.3 RippleMem 带来的启发在记忆系统领域现在有一个思路值得关注记忆系统不只是把更多东西检索出来而是让 Agent 学会“回忆”。RippleMem 的核心思想是“记忆涟漪传播”可以把每个记忆看作一个节点节点之间通过关联关系连成网络。当一个记忆被激活时它会像水面上的涟漪一样逐层扩散到相邻的、弱关联的记忆节点从而唤起一段“上下文场景”而不仅仅是返回一段相似文本。这个思路给我的最大启发是长期记忆的召回不是单次向量搜索就能完成的而应该是一次“记忆激活”。先找到核心锚点比如“干皮”再以锚点为中心扩散到相关标签比如“护肤品”“玫瑰味”“500 元以内”最后组装成对当前任务有完整帮助的记忆片段。在我们的代码实现中会用“标签 向量检索 规则过滤”的组合方式来模拟这种记忆激活。5. 实战一基于 Langchain Langgraph 实现短期记忆5.1 设计思路短期记忆的目标是让 Agent 在一次会话内保持状态。Langgraph 的状态图本身具备 State我们只需要把对话消息列表和中间状态都放在 State 中并通过 Checkpoint 持久化即可。这里我们用一个常见的MessageGraph思路整个图只有一个“对话节点”每次调用时接收用户输入读取当前 State 中的历史消息调用大模型然后把新的问答追加进 State。5.2 核心代码先定义短期记忆模块# memory/short_term.py from typing import Annotated, Sequence from langchain_core.messages import BaseMessage from langgraph.graph.message import add_messages from typing_extensions import TypedDict class ShortTermMemoryState(TypedDict): messages: Annotated[Sequence[BaseMessage], add_messages] user_id: str session_id: str这里的关键是add_messages。Langgraph 的add_messages是一个合并函数当新消息产生时它会自动追加到已有消息列表中而不会覆盖旧消息。这样我们就不用手动维护 message 列表。接着定义对话图# graph.py from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver from langchain_core.messages import SystemMessage from config import MODEL_NAME from memory.short_term import ShortTermMemoryState # 初始化模型 llm ChatOpenAI(modelMODEL_NAME, temperature0.3) # 对话节点根据当前状态生成回复 async def chat_node(state: ShortTermMemoryState): messages state[messages] system_prompt SystemMessage(content你是一个企业级智能客服助手请结合上下文回答用户问题。) response await llm.ainvoke([system_prompt] messages) return {messages: [response]} # 构建状态图 def build_chat_graph(): graph StateGraph(ShortTermMemoryState) graph.add_node(chat, chat_node) graph.add_edge(START, chat) graph.add_edge(chat, END) return graph # 使用内存 Checkpoint重启会丢失生产建议使用持久化 Checkpoint checkpointer MemorySaver() app build_chat_graph().compile(checkpointercheckpointer)5.3 运行与验证在main.py中编写运行逻辑# main.py import asyncio from langchain_core.messages import HumanMessage from graph import app async def main(): # 模拟第一次对话 result1 await app.ainvoke( {messages: [HumanMessage(content我想给女朋友买一瓶适合干皮的香水预算500以内)]}, config{configurable: {thread_id: session_001}} ) print(Agent:, result1[messages][-1].content) # 模拟第二轮对话测试短期记忆 result2 await app.ainvoke( {messages: [HumanMessage(content上次说的那个有没有玫瑰味的)]}, config{configurable: {thread_id: session_001}} ) print(Agent:, result2[messages][-1].content) asyncio.run(main())注意这里的thread_id是 Langgraph 的关键参数它相当于会话的“记忆分区标识”。同一个thread_id下的所有 State 会被 Checkpoint 串联起来所以第二轮对话时Agent 能看到第一轮的历史消息。预期效果Agent 在第二轮不会再问“你说的是哪个”而是能基于第一轮的香水推荐进一步筛选玫瑰味。5.4 为什么短期记忆不只需要消息列表很多场景下短期记忆还需要保存中间状态比如“已经查询过的商品列表”“当前正在比价的商品 ID”“是否已经向用户确认过预算”。这些不属于对话消息但对决策很重要。因此建议在 ShortTermMemoryState 中增加业务字段class ShortTermMemoryState(TypedDict): messages: Annotated[Sequence[BaseMessage], add_messages] user_id: str session_id: str # 业务中间状态 candidate_products: list current_intent: str pending_question: str这些字段会在节点之间传递也能被 Checkpoint 持久化。6. 实战二基于 Langgraph DeepAgents 实现长期记忆6.1 长期记忆的整体链路长期记忆的完整链路可以概括为 5 步提取、概要化、结构化、存储、召回。提取从多轮对话中抽取出值得记住的信息例如用户肤质、预算区间、品牌偏好。概要化把一段长对话压缩为精简的记忆描述。结构化将记忆整理成标签、键值或三元组便于后续精确匹配。存储分别写入向量库和结构化画像存储。召回在新会话开始时根据用户 ID 和当前任务目标召回相关记忆。6.2 记忆提取与结构化下面我们用 Pydantic 定义一个记忆单元模型# models.py from pydantic import BaseModel, Field from typing import List, Optional class MemoryUnit(BaseModel): user_id: str memory_type: str Field(description记忆类型preference/order/event/fact) content: str Field(description记忆内容描述) tags: List[str] Field(default_factorylist, description相关标签) importance: float Field(default0.5, description重要性权重 0-1) expire_at: Optional[str] Field(defaultNone, description过期时间None表示长期有效)通过memory_type区分记忆类型tags用于快速匹配importance用于记忆衰减和清理策略。这个模型可以存储到 Redis、MySQL 或向量库中按业务需要选择。6.3 实现 DeepAgents 规划器抽象这里我们需要谨慎说明DeepAgents 是一个方向性的 Agent 框架名具体 API 以官方文档为准。为了演示架构我们定义一个抽象接口生产环境中可以替换为真正的 DeepAgents 实现。# agents/planner.py from langchain_core.messages import BaseMessage class DeepAgentPlanner: DeepAgents 规划器抽象。 实际使用时可以把内部实现替换为官方 DeepAgents 的 Agent 实例。 def __init__(self, llm): self.llm llm async def plan(self, task: str, context: dict) - list[str]: 将复杂任务拆解为多个子步骤。 返回一个字符串列表每个字符串是一个子任务描述。 prompt f 你是一个深度任务规划器。请将以下任务拆解为最多5个子步骤 任务{task} 可用上下文{context} 只输出步骤列表不要解释。 resp await self.llm.ainvoke(prompt) steps [line.strip().lstrip(0123456789.- ) for line in resp.content.split(\n) if line.strip()] return steps这个规划器对应 DeepAgents 的“深度拆解”能力。在实际 DeepAgents 框架中每个子步骤还可能对应一个独立的 Agent并拥有自己的工具集和记忆上下文。6.4 长期记忆的写入与召回下面编写长期记忆管理模块# memory/long_term.py import json import os from typing import List from langchain_openai import OpenAIEmbeddings from langchain_chroma import Chroma from models import MemoryUnit from config import VECTOR_STORE_PATH, PROFILE_STORE_PATH class LongTermMemory: def __init__(self): self.embeddings OpenAIEmbeddings(modeltext-embedding-3-small) os.makedirs(VECTOR_STORE_PATH, exist_okTrue) os.makedirs(PROFILE_STORE_PATH, exist_okTrue) self.vector_store Chroma( collection_namelong_term_memory, embedding_functionself.embeddings, persist_directoryVECTOR_STORE_PATH, ) async def save_memory(self, memory: MemoryUnit): 写入长期记忆同时写向量库和画像文件 # 向量库保存语义信息 await self.vector_store.aadd_texts( texts[memory.content], metadatas[{ user_id: memory.user_id, memory_type: memory.memory_type, importance: memory.importance, expire_at: memory.expire_at or , }], ids[f{memory.user_id}_{memory.tags[0]}_{memory.content[:20]}] ) # 画像文件保存结构化标签 profile_path os.path.join(PROFILE_STORE_PATH, f{memory.user_id}.json) profile {} if os.path.exists(profile_path): with open(profile_path, r, encodingutf-8) as f: profile json.load(f) for tag in memory.tags: profile[tag] memory.content with open(profile_path, w, encodingutf-8) as f: json.dump(profile, f, ensure_asciiFalse, indent2) async def recall_memory(self, user_id: str, query: str, top_k: int 5) - List[MemoryUnit]: 召回记忆结合向量检索 用户画像 docs await self.vector_store.asimilarity_search( query, ktop_k, filter{user_id: user_id} ) units [] for doc in docs: meta doc.metadata units.append(MemoryUnit( user_iduser_id, memory_typemeta.get(memory_type, fact), contentdoc.page_content, tags[], importancefloat(meta.get(importance, 0.5)), )) return units这里展示了两条召回路径向量库用于语义相似召回画像文件用于精确标签匹配。在真正的生产系统中建议把画像放到图数据库或关系型数据库中方便处理复杂的标签组合查询。6.5 在 Langgraph 状态图中接入长期记忆我们在状态图中新增两个节点一个是“回忆节点”在新会话开始时加载长期记忆另一个是“沉淀节点”在当前对话结束后抽取值得沉淀的记忆。# graph.py 追加 from agents.planner import DeepAgentPlanner from memory.long_term import LongTermMemory from langchain_core.messages import SystemMessage, HumanMessage long_term_memory LongTermMemory() planner DeepAgentPlanner(llmllm) async def recall_node(state: ShortTermMemoryState): 对话开始前加载用户长期记忆 user_memories await long_term_memory.recall_memory(state[user_id], query用户偏好) if user_memories: memory_text \n.join([m.content for m in user_memories]) state[messages].insert(0, SystemMessage( contentf以下是对该用户的长期记忆信息请参考但不局限于这些信息\n{memory_text} )) return {messages: state[messages]} async def reflect_node(state: ShortTermMemoryState): 对话结束后抽取并沉淀长期记忆 conversation state[messages] # 在大模型深层抽取之前我们先做基础的关键信息抽取 extract_prompt 从以下对话中抽取出用户的长期偏好、个人特征、重要事件输出结构化JSON # 实际项目中这里会调用 LLM 处理示例从简 return {}注意insert(0, ...)这种操作在 Langgraph 中建议在新的返回值中完成避免直接修改原 State。上面的代码只是演示记忆加载逻辑实际编写时建议构建一个新的消息列表返回。沉淀节点是长期记忆的重要一环。如果只写短期记忆不沉淀那么每次会话结束后所有有价值的用户信息都会丢失长期记忆就没有意义了。沉淀节点的执行策略一般有两种对话结束时对整个会话做一次记忆抽取。对话过程中当检测到强偏好、关键事件时实时抽取。第一种实现简单第二种实时性更好。企业级系统通常会两者结合。7. 实战三电商场景下的完整案例7.1 需求分析现在我们把前面所有模块组合起来实现一个电商导购 Agent。需求如下用户首次访问时明确自己的肤质是“干皮”预算 500 元以内用户退出后再次访问时Agent 能自动看到“干皮”标签不再重复询问当用户提到“上次推荐的那款香水”时Agent 能关联短期记忆中的商品候选列表Agent 可以调用商品检索工具返回符合条件的结果。7.2 数据结构我们在记忆单元基础上增加商品检索工具所需的数据# models.py 追加 from typing import List class Product(BaseModel): product_id: str name: str price: float tags: List[str] skin_type: str7.3 商品工具# agents/executor.py from models import Product # 模拟商品库 PRODUCTS [ Product(product_idP001, name玫瑰之水香水, price480, tags[玫瑰, 清新], skin_type干皮), Product(product_idP002, name木质调淡香水, price520, tags[木质, 沉稳], skin_type中性), Product(product_idP003, name保湿花香香水, price450, tags[花香, 甜润], skin_type干皮), ] def search_products(skin_type: str, max_price: float) - list[Product]: return [p for p in PRODUCTS if p.skin_type skin_type and p.price max_price]这里只是最简单的规则过滤。真实电商场景中应该替换为搜索服务或商品 API 调用。7.4 完整对话流程我们扩展状态图加入工具调用节点# graph.py 追加工具节点 from langchain_core.tools import tool from agents.executor import search_products tool def search_product_tool(skin_type: str, max_price: float) - str: 根据肤质和价格上限搜索商品 products search_products(skin_type, max_price) if not products: return 没有找到匹配的商品 return \n.join([f{p.name}{p.price}元适合{p.skin_type} for p in products])为了让工具被大模型自动调用需要把工具绑定到 LLM 上llm_with_tools llm.bind_tools([search_product_tool])完整状态图# graph.py from langgraph.prebuilt import ToolNode, tools_condition def build_ecommerce_graph(): graph StateGraph(ShortTermMemoryState) # 节点划分 graph.add_node(recall, recall_node) # 加载长期记忆 graph.add_node(chat, chat_with_tools) # 对话 工具决策 graph.add_node(tools, ToolNode([search_product_tool])) # 执行工具 graph.add_node(reflect, reflect_node) # 沉淀长期记忆 # 边关系 graph.add_edge(START, recall) graph.add_edge(recall, chat) graph.add_conditional_edges(chat, tools_condition, {tools: tools, END: END}) graph.add_edge(tools, chat) graph.add_edge(chat, reflect) graph.add_edge(reflect, END) return graph其中chat_with_tools节点与之前chat_node类似区别在于调用llm_with_tools让模型能够自主决定是否调用商品搜索工具。7.5 运行效果演示# main.py 电商案例演示 async def ecommerce_demo(): # 第一次会话用户说明偏好 result await app.ainvoke( { messages: [HumanMessage(content我是干皮想买一瓶500以内的香水推荐一下)], user_id: user_9527, session_id: session_100, }, config{configurable: {thread_id: session_100}} ) print(Agent:, result[messages][-1].content) # 第二次会话模拟用户隔天再次访问 result2 await app.ainvoke( { messages: [HumanMessage(content上次推荐的那款玫瑰味香水还有吗)], user_id: user_9527, session_id: session_200, }, config{configurable: {thread_id: session_200}} ) print(Agent:, result2[messages][-1].content)预期行为分析第一轮系统通过recall_node查询长期记忆。由于是新用户没有历史信息所以 Agent 会主动询问或根据用户输入直接推荐商品。当用户明确说了“干皮”“500 以内”沉淀节点会将这两个关键信息写入长期记忆。第二轮虽然session_id变了thread_id从session_100变为session_200短期记忆已经被隔离但recall_node会根据user_id从长期记忆中拉取“干皮”标签再结合第一轮推荐的玫瑰香水信息Agent 就能准确知道用户说的“上次推荐”是哪一款。7.6 结果说明这个案例虽然简单但它覆盖了记忆系统的三个关键动作短期记忆保持会话内消息、长期记忆沉淀跨会话标签、业务数据读取商品工具调用。在真实项目中无论你接入的是订单系统、会员系统还是商品中台都可以沿用这套架构。8. 常见问题与排查思路在企业级开发里记忆系统出现问题的概率很高。下面整理一些高频问题和排查思路问题现象常见原因解决思路第二轮对话 Agent 不记得上文thread_id未保持一致检查每次请求的configurable.thread_id长期记忆不生效用户 ID 不一致或未传user_id在上游统一登录认证中提取用户 ID向量召回结果不准确记忆文本过短、检索 Query 不清晰为记忆单元增加标签结合结构化匹配历史消息过多导致 Token 超限短期记忆没有裁剪 / 摘要化设置窗口长度超出部分自动摘要需要持久化的记忆写失败向量库连接或磁盘权限问题检查向量库连接和存储目录权限Agent 死循环工具调用工具返回结果缺少退出条件设置最大工具调用次数增加异常分支补充几个小经验thread_id是短期记忆的隔离空间建议使用会话 ID而不是用户 ID。同一个用户可以开多个会话每个会话的短期记忆应该相互隔离。user_id最好从统一的用户体系中获取后端服务只在可信的内网环境中传递不要在公网请求里盲目信任前端传入的 user_id。长期记忆的写入要有去重逻辑否则同样的偏好每次对话都会重复写入。9. 最佳实践与工程建议9.1 记忆裁剪与摘要策略短期记忆不能无限增长。建议设置消息窗口比如保留最近 20 条完整消息更早的消息压缩为一段摘要。Langgraph 中可以在节点内检查len(state[messages])超过阈值时调用 LLM 摘要。也可以使用 Langchain 自带的SummarizationMemory思路但注意状态图环境下需要自己实现消息合并。9.2 记忆衰减与清理长期记忆必须有“生命周期”。有些信息是临时性的比如“用户本周计划购买生日礼物”一周后这条记忆可能就过期了。建议在 MemoryUnit 中设置expire_at定期清理过期记忆对importance较低的记忆进行衰减比如长时间未被命中就降低权重最终自动删除。9.3 隐私与数据安全边界记忆系统存储的是真实用户的个人信息必须重视安全合规敏感数据手机号、地址、身份证号不应明文写入记忆库可以使用脱敏或外部加密存储。用户应有权查看、导出、删除自己的记忆数据。建议提供“记忆管理”接口。记忆召回不要无限制开放给所有 Agent建议按业务域隔离。9.4 可观测性与调试记忆系统是个“黑盒”问题的重灾区。建议每次记忆写入和召回都记录日志包含触发的用户 ID、会话 ID写入的记忆内容与标签召回时的查询语句目标记忆来源。这样当模型使用了过期记忆或错误记忆时才能快速定位是哪一步写入造成的。9.5 架构层面的演进方向如果要从演示版本演进到生产级建议做以下改造将 Chroma 替换为独立的向量数据库服务支持水平扩展。将 JSON 用户画像替换为 Redis 或 PostgreSQL并增加唯一约束。引入消息队列异步执行记忆沉淀避免影响主对话链路延迟。为记忆系统增加版本管理记忆结构升级时方便做数据迁移。10. 总结与学习路线本文围绕 Agent 记忆系统从概念到代码完成了三件事先用 Langgraph 的实现搞定了短期记忆用 Langchain 向量库 用户画像的方式实现了长期记忆再引入 DeepAgents 层的规划器抽象把记忆系统接入到电商导购场景中。核心思路是短期记忆靠状态图 Checkpoint长期记忆靠“提取 结构化 向量召回”而不是简单地把更多文本塞进 Prompt。接下来你可以继续深入研究几个方向一是 Langgraph 的 Checkpoint 持久化尝试把状态保存到 Redis 或数据库中二是 RippleMem 这类“记忆涟漪”思想用图网络改造记忆召回逻辑三是 DeepAgents 的完整官方文档把规划器替换成真正的深度 Agent 执行链路四是结合业务系统把订单、客服工单等实时业务数据接入 Agent 记忆。如果你在实际搭建过程中遇到奇怪的报错或者对记忆分层有不同的理解欢迎在评论区留言讨论。动手跑通一遍比读十篇概念文章都管用。