在大模型应用真正走入业务之后很多人会发现一个尴尬的现实单轮问答已经足够惊艳但只要连续对话几次、或者隔天再问模型就“失忆”了。用户昨天刚说过的偏好、刚浏览过的商品、刚表达过的要求第二天全都归零。这种现象在电商导购、企业客服、个人助理类Agent应用中非常致命。本文将围绕Langchain、Langgraph、DeepAgents这套技术组合完整拆解如何为企业级Agent打造一套记忆系统覆盖短期记忆、长期记忆、向量检索、用户画像沉淀并以电商推荐助手为例跑通一条可落地的闭环方案。无论你是刚接触Langchain的新手还是已经在做Agent工程化的开发者都可以从本文中找到可直接复用的代码与设计思路。1. 背景与核心概念1.1 为什么Agent需要记忆系统先思考一个最简单的场景用户对电商助手说“帮我找一款适合油性皮肤的洗面奶”助手推荐了几款用户说“第二款不错”。如果Agent没有记忆下一次对话时它不会记得“第二款”是哪款更不会记得用户的肤质是油性。用户只能重复诉求体验会变得异常糟糕。Agent记忆系统要解决的就是这个核心问题让Agent在多次对话之间保留上下文并把有意义的信息沉淀下来形成对用户的持续理解。它不只是“把聊天记录存下来”而是要区分哪些信息需要短期记住、哪些信息值得长期保留以及在不同对话场景中如何召回。1.2 短期记忆与长期记忆记忆系统通常分为两层短期记忆指当前会话或最近几轮对话中的上下文信息比如用户刚说的“第二款不错”或者用户在当前会话里切换过几次意向。短期记忆一般存放在会话状态或消息列表中使用成本低、读取快但过了会话有效期后通常会被清理。长期记忆指跨会话持久化的用户信息比如用户肤质、消费习惯、价格偏好、历史咨询记录等。长期记忆需要写到外部存储通常结合向量数据库做相似度检索或者将用户画像整理成结构化键值对保存。在设计Agent时二者必须配合使用短期记忆保障对话连续性长期记忆保障个性化服务质量。1.3 Langchain、Langgraph、DeepAgents三者的关系很多初学者会把Langchain、Langgraph、DeepAgents混为一谈这里先做一个简单区分。Langchain是一套用于构建大模型应用的开发框架它提供了模型封装、Prompt模板、工具调用、输出解析、向量存储集成、RAG流程等基础组件。可以说Langchain解决的是“大模型应用开发中的通用零件问题”。Langgraph是Langchain生态中的图编排框架。它把Agent的运行流程建模为一张有向图图中的每个节点可以是模型调用、工具执行、记忆读取、条件判断边则定义了节点之间的流转关系。Langgraph非常适合实现可控、可维护、可观测的Agent流程尤其适合需要精细控制状态流转的场景。DeepAgents是面向深度Agent任务的高层开发框架它在Langchain之上进一步封装了Agent的核心运行循环比如“思考-调用工具-观察结果-再思考”这类harness机制。开发者只需要构建Agent并传入工具列表DeepAgents会负责管理执行循环和上下文。简单来说Langchain提供能力组件Langgraph负责流程编排DeepAgents提供开箱即用的高级Agent运行框架。记忆系统可以在Langgraph层做精细控制也可以借助DeepAgents的上下文管理机制快速集成。2. 技术选型与环境准备2.1 技术栈说明本文示例采用以下技术组合Python 3.10 及以上版本。OpenAI 兼容的Chat模型通过网络请求调用需配置环境变量保存API Key。Langchain与Langgraph库版本需要根据你的项目实际情况调整。本文重点演示设计思路不同版本间API可能略有差异。Chroma作为本地向量数据库用于长期记忆的相似度检索。它轻量、无需额外部署适合示例场景生产环境建议替换为Milvus、Qdrant、PGVector等。DeepAgents用于构建高层Agent运行循环。版本建议以官方最新稳定版为准不要盲目锁定我写到的版本号。配置时保持langchain、langgraph、deepagents的版本兼容性这是最常见的坑之一。2.2 安装依赖建议使用虚拟环境安装python -m venv venv source venv/bin/activate # Windows 下使用 venv\Scripts\activate pip install --upgrade langchain langchain-openai langgraph chromadb deepagents其中langchain-openai提供OpenAI兼容模型的调用封装。chromadb是长期记忆的向量存储。deepagents用于高层Agent封装。如果你的网络环境无法直接安装可以配置国内镜像源pip install --upgrade langchain langchain-openai langgraph chromadb deepagents -i https://pypi.tuna.tsinghua.edu.cn/simple2.3 项目结构为了便于维护示例采用一个小型工程目录agent_memory_demo/ ├── requirements.txt ├── .env ├── config.py ├── memory/ │ ├── __init__.py │ ├── short_term.py │ └── long_term.py ├── agent/ │ ├── __init__.py │ ├── graph.py │ └── deep_agent.py └── ecommerce/ ├── __init__.py ├── tools.py └── main.py这个目录结构适合把记忆能力封装成独立模块后续接入不同业务时只需要替换工具层。2.4 环境变量配置在项目根目录创建.env文件OPENAI_API_KEYsk-xxxxx OPENAI_BASE_URLhttps://api.openai.com/v1如果使用国内大模型服务商提供的OpenAI兼容接口把OPENAI_BASE_URL改成对应地址即可。注意不要把Key硬编码在代码中。3. 记忆系统核心原理拆解3.1 Langgraph状态机制Langgraph的核心设计是状态管理。每一个节点函数接收当前状态处理后返回状态的部分更新图引擎负责把这些更新合并到全局状态中。以聊天型Agent为例我们会定义一个消息状态from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] user_id: str这里的messages字段使用add_messages注解意味着每次节点返回新的消息时Langgraph会把消息追加到原来的列表而不是整体覆盖。这是短期记忆最底层的实现机制。3.2 短期记忆实现方式短期记忆最朴素的做法是把对话历史拼接进Prompt。但随着对话变长Token成本会上升模型对早期信息的关注度也会下降。因此工程上通常采用滑动窗口只保留最近N条消息更早的历史归档到长期存储。在Langgraph中可以通过节点函数实现窗口裁剪def trim_messages(state: AgentState) - AgentState: if len(state[messages]) 20: return {messages: state[messages][-20:]} return {messages: state[messages]}这里返回的是一个新的消息列表Langgraph会通过add_messages注解合并。需要注意add_messages的合并行为是追加如果要用裁剪后的列表覆盖原有列表不能直接通过add_messages实现需要改用自定义reducer或直接处理原始状态。更稳妥的做法是在调用模型之前把裁剪后的消息列表传给Prompt模板而不是修改全局消息状态。3.3 长期记忆实现方式长期记忆通常有两种形态向量记忆把用户说过的话、行为记录、偏好描述切块向量化后存入向量数据库。需要召回时用当前问题做相似度检索取出最相关的历史片段作为上下文。结构化记忆通过大模型把用户信息抽取成结构化的键值对或JSON比如“肤质油性”、“价格偏好200-400元”。结构化记忆适合精确查询但抽取效果依赖模型能力。两种形态不是互斥的。优秀的企业级记忆系统通常结合两者向量记忆负责模糊召回结构化画像负责精确匹配。3.4 让Agent学会“回忆”这里有一个值得借鉴的设计理念真正的记忆系统不是把更多东西检索出来而是让Agent学会在合适的时机主动回忆起关键信息。也就是说不要把当前用户问题的每个词都拿去做向量检索然后盲目拼接检索结果。更合理的做法分两步判断当前问题是否需要回忆。如果用户只是闲聊“你好”不需要拉取历史画像。如果需要回忆再确定回忆什么。比如用户问“有什么适合我的护肤品”此时需要从长期记忆中检索肤质、品牌偏好、价格区间。这个“回忆”动作可以建模成Langgraph中的一个节点。在调用大模型之前Agent先调用记忆查询工具把检索结果注入上下文再让模型作答。与RAG检索增强生成类似但记忆系统的检索目标不是外部文档而是用户自身的历史信息。4. 基于Langgraph的记忆Agent实战这一节我们会实现一个通用记忆Agent它具备短期记忆和长期记忆能力。代码会分文件展示方便直接复制。4.1 定义状态与配置首先创建config.py负责加载环境变量和模型import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI load_dotenv() def create_llm(): return ChatOpenAI( modelgpt-4o-mini, temperature0.3, api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL) )model名称要根据你实际使用的模型调整。对于国内模型服务商通常传服务商提供的模型标识。然后定义图状态# agent/state.py from typing import TypedDict, Annotated class AgentState(TypedDict): user_id: str query: str response: str short_memory: list long_memory: list这里没有直接用add_messages自动管理消息而是显式维护short_memory和long_memory两个字段这样逻辑更清晰short_memory当前窗口内的最近几轮问答。long_memory从长期记忆库中检索到的用户画像片段。4.2 实现长期记忆存储模块创建memory/long_term.py使用Chroma向量数据库保存用户历史信息片段import chromadb from chromadb.utils import embedding_functions class LongTermMemory: def __init__(self, user_id: str): self.user_id user_id self.client chromadb.PersistentClient(path./chroma_db) self.collection self.client.get_or_create_collection( nameuser_memory, embedding_functionembedding_functions.DefaultEmbeddingFunction() ) def add_memory(self, text: str, memory_type: str general): self.collection.add( documents[text], ids[f{self.user_id}_{uuid.uuid4().hex}], metadatas[{user_id: self.user_id, memory_type: memory_type}] ) def search(self, query: str, top_k: int 3): results self.collection.query( query_texts[query], n_resultstop_k, where{user_id: self.user_id} ) return results[documents][0] if results[documents] else []需要补充import uuid。这个模块的语义是每个用户拥有独立的记忆片段集合通过user_id过滤避免跨用户串记忆。4.3 实现记忆管理节点创建agent/nodes.py包含三个核心节点函数。第一个节点负责短期记忆管理。它把历史消息保留最近6轮更早的内容不再传给模型from state import AgentState def manage_short_term_memory(state: AgentState) - AgentState: # 假设 state[short_memory] 保存的是 [{role: user/assistant, content: ...}] recent state.get(short_memory, [])[-6:] return {short_memory: recent}第二个节点是“回忆”节点根据当前用户问题从长期记忆中检索相关内容from memory.long_term import LongTermMemory def recall_memory(state: AgentState) - AgentState: user_id state[user_id] query state[query] memory_store LongTermMemory(user_id) recalled memory_store.search(query, top_k3) return {long_memory: recalled}第三个节点是“存储”节点在Agent回答完成后把对用户有价值的陈述写入长期记忆def store_memory(state: AgentState) - AgentState: user_id state[user_id] history state.get(short_memory, []) if len(history) 2: # 取最后一条用户消息作为沉淀来源 last_user_msg history[-1][content] memory_store LongTermMemory(user_id) memory_store.add_memory(last_user_msg, memory_typeuser_statement) return {}真实项目中不建议把用户所有发言都存入长期记忆这样会造成记忆污染。更合理的做法是让大模型先抽取值得记忆的信息再写入存储。4.4 实现模型调用节点模型调用节点要把短期记忆和长期记忆组合成Promptfrom langchain_core.prompts import ChatPromptTemplate from config import create_llm SYSTEM_TEMPLATE 你是一个有记忆能力的智能助手。 以下是关于用户的长期记忆信息可能包含用户偏好、历史关注点 long_memory {long_memory} /long_memory 请结合这些记忆信息以及本次会话内容回答用户问题。 def call_model(state: AgentState) - AgentState: llm create_llm() prompt ChatPromptTemplate.from_template(SYSTEM_TEMPLATE) long_memory_text \n.join(state.get(long_memory, [])) or 暂无已知记忆 messages [ (system, prompt.format(long_memorylong_memory_text)), ] for item in state.get(short_memory, []): role user if item[role] user else assistant messages.append((role, item[content])) response llm.invoke(messages) return {response: response.content}这段代码的核心在于长期记忆被注入System Prompt短期记忆按角色组成多轮消息列表。模型在生成回答前已经知道自己“记得”哪些信息。4.5 构建Graph并运行创建agent/graph.py把节点连接成一张图from langgraph.graph import Graph from agent.nodes import manage_short_term_memory, recall_memory, call_model, store_memory def build_memory_agent(): workflow Graph() workflow.add_node(manage_short_term_memory, manage_short_term_memory) workflow.add_node(recall_memory, recall_memory) workflow.add_node(call_model, call_model) workflow.add_node(store_memory, store_memory) workflow.set_entry_point(manage_short_term_memory) workflow.add_edge(manage_short_term_memory, recall_memory) workflow.add_edge(recall_memory, call_model) workflow.add_edge(call_model, store_memory) workflow.set_finish_point(store_memory) return workflow.compile()运行一次对话from agent.graph import build_memory_agent agent build_memory_agent() def chat(user_id: str, user_message: str, history: list): result agent.invoke({ user_id: user_id, query: user_message, short_memory: history, long_memory: [] }) return result history [] response chat(user_001, 我是油性皮肤想找控油洗面奶, history) print(Agent:, response[response]) history.append({role: user, content: 我是油性皮肤想找控油洗面奶}) history.append({role: assistant, content: response[response]}) response chat(user_001, 有什么品牌推荐, history) print(Agent:, response[response])第二次提问虽然表面上没提肤质但Agent通过长期记忆已经检索到“用户是油性皮肤”这一关键信息推荐结果会更精准。5. 基于DeepAgents的进阶方案5.1 DeepAgents为记忆管理带来了什么DeepAgents的核心价值在于它实现了完整的Agent运行循环也就是“Agent Harness”。在这个循环里Agent会反复思考、调用工具、观察结果直到生成最终答案。它底层与Langchain深度集成因此我们可以把上一节实现的长期记忆能力作为工具提供给DeepAgents。相比自己手写Langgraph节点DeepAgents能减少很多控制流代码。但代价是控制粒度不如Langgraph精细所以它的定位更适合标准Agent任务而不太适合需要复杂状态机编排的业务。5.2 创建带记忆能力的DeepAgent把长期记忆模块封装成工具然后构建DeepAgentfrom deepagents import DeepAgent from langchain_core.tools import tool from config import create_llm from memory.long_term import LongTermMemory tool def search_user_memory(query: str, user_id: str) - str: 从用户长期记忆中检索相关信息参数为查询内容和用户ID。 memory_store LongTermMemory(user_id) results memory_store.search(query, top_k3) return \n.join(results) if results else 暂无相关记忆注意这里的工具参数格式需要根据DeegAgents的版本调整有些版本支持自动从上下文推断user_id。更稳妥的做法是在创建Agent时通过system prompt注入用户信息def create_memory_deep_agent(user_id: str): system_prompt f 你是用户的私人助手当前用户ID是{user_id}。 在回答用户问题之前如果问题涉及到用户偏好或历史信息 请先调用search_user_memory工具进行回忆。 agent DeepAgent.build( llmcreate_llm(), tools[search_user_memory], system_promptsystem_prompt ) return agent构建之后就可以直接调用agent create_memory_deep_agent(user_001) response agent.run(我的肤质是什么来着我适合用什么产品) print(response)DeepAgents会自动完成“调用工具→获取记忆→组织回答”的循环你不需要手动管理中间步骤。5.3 DeepAgents与Langgraph如何选择实际项目中两者可以组合使用。Langgraph适合做外层业务流程控制比如订单处理、多步审批、复杂条件分支DeepAgents适合做单任务内的深度推理比如需要不断调用工具才能完成的咨询问答。在记忆系统层面我的建议是如果你需要精细控制记忆读取时机和上下文拼接方式优先使用Langgraph如果你想快速上线一个具备基本记忆能力的AgentDeepAgents能把工作量减少一半。6. 电商实战案例带记忆的商品推荐助手6.1 场景与需求假设我们要做一个电商导购Agent核心需求是用户第一次咨询时Agent记录用户肤质、价格偏好、品牌偏好。用户后续咨询时Agent自动回忆起这些信息结合当前商品库做个性化推荐。推荐结果要能调用商品搜索工具而不是只靠模型编造商品。这个案例能同时体现短期记忆、长期记忆、工具调用三者的配合。6.2 用户画像存储设计长期记忆模块沿用上一节代码。我们在写入时增加一个类型字段比如“skin_type”、“price_preference”、“brand_preference”。这样后续筛选时更精确。同时可以加一个summarize_and_store函数用大模型抽取用户信息def extract_user_profile(llm, user_message: str) - list: prompt ChatPromptTemplate.from_template( 从用户信息中提取值得长期记忆的结构化内容。 输出JSON数组每个元素格式为 {{type: skin_type, content: 油性皮肤}}。 无法提取时返回空数组。 用户信息{message} ) chain prompt | llm result chain.invoke({message: user_message}) # 这里需要解析JSON建议使用PydanticOutputParser return extract_json_array(result.content)生产环境中请务必加JSON解析校验防止模型输出格式不合法导致程序崩溃。6.3 商品搜索工具创建ecommerce/tools.py模拟一个商品搜索工具。实际项目中可替换为数据库查询或调用电商搜索服务PRODUCT_DB [ {id: 1, name: 控油洁面乳, tags: [油性, 清洁, 控油], price: 159}, {id: 2, name: 保湿爽肤水, tags: [干性, 保湿, 温和], price: 199}, {id: 3, name: 淡斑精华, tags: [中性, 美白, 淡斑], price: 359}, {id: 4, name: 防晒乳, tags: [所有肤质, 防晒, 日常], price: 129}, ] def search_product(tags: list[str], max_price: int 500): results [] for product in PRODUCT_DB: if any(tag in product[tags] for tag in tags) and product[price] max_price: results.append(product) return results6.4 电商助手核心流程在Langgraph中增加工具调用节点让模型决定是否需要搜索商品。由于完整实现需要引入ToolNode相关机制这里给出一个简化但可运行的结构from langgraph.graph import Graph from agent.nodes import recall_memory, store_memory from ecommerce.tools import search_product def recommend_node(state: AgentState) - AgentState: llm create_llm() long_memory_text \n.join(state.get(long_memory, [])) or 暂无已知记忆 prompt f 你是电商导购助手。用户信息如下 {long_memory_text} 请从以下商品中选择合适推荐并说明理由。 商品库 {PRODUCT_DB} response llm.invoke(prompt) return {response: response.content} def ecommerce_agent(): workflow Graph() workflow.add_node(recall_memory, recall_memory) workflow.add_node(recommend, recommend_node) workflow.add_node(store_memory, store_memory) workflow.set_entry_point(recall_memory) workflow.add_edge(recall_memory, recommend) workflow.add_edge(recommend, store_memory) workflow.set_finish_point(store_memory) return workflow.compile()6.5 运行效果说明第一次对话result agent.invoke({ user_id: user_001, query: 我皮肤很油预算300以内有什么推荐的, short_memory: [], long_memory: [] })Agent的回答会偏向控油类商品。回答结束后要点是检查Chroma中是否写入了用户偏好。第二次对话result agent.invoke({ user_id: user_001, query: 给我推荐一款日常用的防晒, short_memory: [], long_memory: [] })由于长期记忆中有“油性皮肤”的记录推荐结果会自动优先考虑质地清爽的防晒产品而不是无差别推荐所有防晒。这个闭环的价值在于用户不需要重复描述自己的肤质和预算体验更接近真人导购。7. 常见问题与排查思路7.1 高频报错与解决问题现象常见原因解决思路启动时提示langchain版本冲突各库要求不同版本统一升级到最新稳定版避免混用新旧APIAgent执行时报“Agent execution terminated due to error.”工具内部异常未捕获为每个工具增加try-except返回错误描述而不是直接抛出长期记忆检索不到内容向量库中还没有写入数据或写入未落盘检查Chroma持久化目录确认调用过add_memory检索结果与当前问题不相关嵌入模型语义能力不足或chunk过大更换更强的Embedding模型按语义块控制存储粒度对话历史越来越长Token消耗过高没有做窗口裁剪在调用模型前裁剪最近N轮消息模型回答时忽略长期记忆System Prompt中记忆信息排布靠后或过于冗长精简记忆内容把最关键的画像信息放在Prompt显眼位置Langgraph中send(node_name, state)用法不清晰对动态并行图理解不到位放到7.2节单独说明7.2 Langgraph中的send到底怎么理解send是Langgraph中用于动态生成并行任务的API。它允许你在一个节点内为多个数据源并行触发同一个后续节点。举个例子假设你拿到了10个商品希望针对每个商品并行调用一个大模型节点生成评论可以在中间节点里from langgraph.types import send def distribute_tasks(state): products state[products] return [send(analyze_product, {product: p}) for p in products]这里的send(node_name, state)表示为每个product生成一条任务发送到analyze_product节点去处理各个任务互相独立Langgraph会并行调度。如果你还没搞懂send可以把它理解为“循环触发同一个下游节点每个元素传一份独立状态”。7.3 排查清单如果Agent记忆功能表现异常可以用以下顺序排查确认短期记忆传入模型前的顺序是否正确通常是System前置历史对话按时间顺序排列。确认长期记忆有数据写入可以直接查询Chroma集合的count()。确认检索时使用了正确的user_id过滤条件避免串记忆。确认模型Prompt中是否真的拼接了长期记忆内容必要时打印最终Prompt调试。8. 工程化最佳实践8.1 记忆内容治理记忆系统的核心风险不是“记不住”而是“什么都记”。如果用户随口说的“我不喜欢蓝色”也被永久保存长期积累下来会形成大量低质量和互相矛盾的信息。建议在写入长期记忆前增加一道“记忆筛选”步骤让大模型判断这段话是否值得记忆、属于什么类型、是否与已有记忆冲突。如果与已有记忆冲突优先采用最新描述。8.2 用户身份隔离与隐私安全企业级系统必须做到用户维度隔离。向量集合中增加user_id元数据并在查询时强制过滤防止A用户检索到B用户的私有信息。在数据库层面建议按租户分collection或增加租户ID过滤条件。涉及用户敏感信息时要遵循最小化原则只记忆为用户服务所必需的信息并在用户注销或明确要求时提供删除接口。生产环境操作涉及用户数据删除时应先备份、走审批流程并在测试环境验证后再执行。8.3 性能与成本控制长期记忆的检索请求不宜每次都触发可以在节点中加入条件判断当用户问题长度过短或属于寒暄类时跳过检索。同时为用户画像增加缓存减少重复的向量查询。写入侧也要控制频率。只有用户表达了新的有效信息时才写入记忆不要每轮对话都调用Embedding和写入操作。这能显著降低成本。8.4 可观测性设计记忆系统是隐性的用户不会直接看到Agent“回忆”的过程但工程师必须能观测到。建议为每个环节增加日志记忆写入日志记录写入时间、用户ID、记忆类型、内容摘要。记忆检索日志记录查询文本、召回结果、相似度得分。Prompt组装日志记录最终传给模型的System Prompt内容。有了这些日志当用户反馈回答异常时可以快速定位是记忆没写入、检索没召回还是Prompt拼接问题。9. 总结本文从Agent记忆的痛点出发完整介绍了短期记忆和长期记忆的设计方案并用Langgraph实现了可控的记忆流短期记忆裁剪、长期记忆检索、模型调用、记忆沉淀。之后引入了DeepAgents作为更轻量的高级Agent方案最后通过电商推荐助手案例串起了完整闭环。如果你想在真实项目中落地建议先跑通第4节的通用记忆Agent再根据业务场景替换模型、向量库和工具层。在动手过程中你可以重点观察三个问题哪些用户信息真正影响了回答质量、检索召回是否有噪音、记忆写入是否过度。这三个问题的答案往往决定了记忆系统的ROI。希望这篇长文能帮你把“大模型有记忆”从演示项目推进到可维护的工程系统。 基于Langchain、Langgraph与DeepAgents构建企业级Agent记忆系统
在大模型应用真正走入业务之后很多人会发现一个尴尬的现实单轮问答已经足够惊艳但只要连续对话几次、或者隔天再问模型就“失忆”了。用户昨天刚说过的偏好、刚浏览过的商品、刚表达过的要求第二天全都归零。这种现象在电商导购、企业客服、个人助理类Agent应用中非常致命。本文将围绕Langchain、Langgraph、DeepAgents这套技术组合完整拆解如何为企业级Agent打造一套记忆系统覆盖短期记忆、长期记忆、向量检索、用户画像沉淀并以电商推荐助手为例跑通一条可落地的闭环方案。无论你是刚接触Langchain的新手还是已经在做Agent工程化的开发者都可以从本文中找到可直接复用的代码与设计思路。1. 背景与核心概念1.1 为什么Agent需要记忆系统先思考一个最简单的场景用户对电商助手说“帮我找一款适合油性皮肤的洗面奶”助手推荐了几款用户说“第二款不错”。如果Agent没有记忆下一次对话时它不会记得“第二款”是哪款更不会记得用户的肤质是油性。用户只能重复诉求体验会变得异常糟糕。Agent记忆系统要解决的就是这个核心问题让Agent在多次对话之间保留上下文并把有意义的信息沉淀下来形成对用户的持续理解。它不只是“把聊天记录存下来”而是要区分哪些信息需要短期记住、哪些信息值得长期保留以及在不同对话场景中如何召回。1.2 短期记忆与长期记忆记忆系统通常分为两层短期记忆指当前会话或最近几轮对话中的上下文信息比如用户刚说的“第二款不错”或者用户在当前会话里切换过几次意向。短期记忆一般存放在会话状态或消息列表中使用成本低、读取快但过了会话有效期后通常会被清理。长期记忆指跨会话持久化的用户信息比如用户肤质、消费习惯、价格偏好、历史咨询记录等。长期记忆需要写到外部存储通常结合向量数据库做相似度检索或者将用户画像整理成结构化键值对保存。在设计Agent时二者必须配合使用短期记忆保障对话连续性长期记忆保障个性化服务质量。1.3 Langchain、Langgraph、DeepAgents三者的关系很多初学者会把Langchain、Langgraph、DeepAgents混为一谈这里先做一个简单区分。Langchain是一套用于构建大模型应用的开发框架它提供了模型封装、Prompt模板、工具调用、输出解析、向量存储集成、RAG流程等基础组件。可以说Langchain解决的是“大模型应用开发中的通用零件问题”。Langgraph是Langchain生态中的图编排框架。它把Agent的运行流程建模为一张有向图图中的每个节点可以是模型调用、工具执行、记忆读取、条件判断边则定义了节点之间的流转关系。Langgraph非常适合实现可控、可维护、可观测的Agent流程尤其适合需要精细控制状态流转的场景。DeepAgents是面向深度Agent任务的高层开发框架它在Langchain之上进一步封装了Agent的核心运行循环比如“思考-调用工具-观察结果-再思考”这类harness机制。开发者只需要构建Agent并传入工具列表DeepAgents会负责管理执行循环和上下文。简单来说Langchain提供能力组件Langgraph负责流程编排DeepAgents提供开箱即用的高级Agent运行框架。记忆系统可以在Langgraph层做精细控制也可以借助DeepAgents的上下文管理机制快速集成。2. 技术选型与环境准备2.1 技术栈说明本文示例采用以下技术组合Python 3.10 及以上版本。OpenAI 兼容的Chat模型通过网络请求调用需配置环境变量保存API Key。Langchain与Langgraph库版本需要根据你的项目实际情况调整。本文重点演示设计思路不同版本间API可能略有差异。Chroma作为本地向量数据库用于长期记忆的相似度检索。它轻量、无需额外部署适合示例场景生产环境建议替换为Milvus、Qdrant、PGVector等。DeepAgents用于构建高层Agent运行循环。版本建议以官方最新稳定版为准不要盲目锁定我写到的版本号。配置时保持langchain、langgraph、deepagents的版本兼容性这是最常见的坑之一。2.2 安装依赖建议使用虚拟环境安装python -m venv venv source venv/bin/activate # Windows 下使用 venv\Scripts\activate pip install --upgrade langchain langchain-openai langgraph chromadb deepagents其中langchain-openai提供OpenAI兼容模型的调用封装。chromadb是长期记忆的向量存储。deepagents用于高层Agent封装。如果你的网络环境无法直接安装可以配置国内镜像源pip install --upgrade langchain langchain-openai langgraph chromadb deepagents -i https://pypi.tuna.tsinghua.edu.cn/simple2.3 项目结构为了便于维护示例采用一个小型工程目录agent_memory_demo/ ├── requirements.txt ├── .env ├── config.py ├── memory/ │ ├── __init__.py │ ├── short_term.py │ └── long_term.py ├── agent/ │ ├── __init__.py │ ├── graph.py │ └── deep_agent.py └── ecommerce/ ├── __init__.py ├── tools.py └── main.py这个目录结构适合把记忆能力封装成独立模块后续接入不同业务时只需要替换工具层。2.4 环境变量配置在项目根目录创建.env文件OPENAI_API_KEYsk-xxxxx OPENAI_BASE_URLhttps://api.openai.com/v1如果使用国内大模型服务商提供的OpenAI兼容接口把OPENAI_BASE_URL改成对应地址即可。注意不要把Key硬编码在代码中。3. 记忆系统核心原理拆解3.1 Langgraph状态机制Langgraph的核心设计是状态管理。每一个节点函数接收当前状态处理后返回状态的部分更新图引擎负责把这些更新合并到全局状态中。以聊天型Agent为例我们会定义一个消息状态from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] user_id: str这里的messages字段使用add_messages注解意味着每次节点返回新的消息时Langgraph会把消息追加到原来的列表而不是整体覆盖。这是短期记忆最底层的实现机制。3.2 短期记忆实现方式短期记忆最朴素的做法是把对话历史拼接进Prompt。但随着对话变长Token成本会上升模型对早期信息的关注度也会下降。因此工程上通常采用滑动窗口只保留最近N条消息更早的历史归档到长期存储。在Langgraph中可以通过节点函数实现窗口裁剪def trim_messages(state: AgentState) - AgentState: if len(state[messages]) 20: return {messages: state[messages][-20:]} return {messages: state[messages]}这里返回的是一个新的消息列表Langgraph会通过add_messages注解合并。需要注意add_messages的合并行为是追加如果要用裁剪后的列表覆盖原有列表不能直接通过add_messages实现需要改用自定义reducer或直接处理原始状态。更稳妥的做法是在调用模型之前把裁剪后的消息列表传给Prompt模板而不是修改全局消息状态。3.3 长期记忆实现方式长期记忆通常有两种形态向量记忆把用户说过的话、行为记录、偏好描述切块向量化后存入向量数据库。需要召回时用当前问题做相似度检索取出最相关的历史片段作为上下文。结构化记忆通过大模型把用户信息抽取成结构化的键值对或JSON比如“肤质油性”、“价格偏好200-400元”。结构化记忆适合精确查询但抽取效果依赖模型能力。两种形态不是互斥的。优秀的企业级记忆系统通常结合两者向量记忆负责模糊召回结构化画像负责精确匹配。3.4 让Agent学会“回忆”这里有一个值得借鉴的设计理念真正的记忆系统不是把更多东西检索出来而是让Agent学会在合适的时机主动回忆起关键信息。也就是说不要把当前用户问题的每个词都拿去做向量检索然后盲目拼接检索结果。更合理的做法分两步判断当前问题是否需要回忆。如果用户只是闲聊“你好”不需要拉取历史画像。如果需要回忆再确定回忆什么。比如用户问“有什么适合我的护肤品”此时需要从长期记忆中检索肤质、品牌偏好、价格区间。这个“回忆”动作可以建模成Langgraph中的一个节点。在调用大模型之前Agent先调用记忆查询工具把检索结果注入上下文再让模型作答。与RAG检索增强生成类似但记忆系统的检索目标不是外部文档而是用户自身的历史信息。4. 基于Langgraph的记忆Agent实战这一节我们会实现一个通用记忆Agent它具备短期记忆和长期记忆能力。代码会分文件展示方便直接复制。4.1 定义状态与配置首先创建config.py负责加载环境变量和模型import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI load_dotenv() def create_llm(): return ChatOpenAI( modelgpt-4o-mini, temperature0.3, api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL) )model名称要根据你实际使用的模型调整。对于国内模型服务商通常传服务商提供的模型标识。然后定义图状态# agent/state.py from typing import TypedDict, Annotated class AgentState(TypedDict): user_id: str query: str response: str short_memory: list long_memory: list这里没有直接用add_messages自动管理消息而是显式维护short_memory和long_memory两个字段这样逻辑更清晰short_memory当前窗口内的最近几轮问答。long_memory从长期记忆库中检索到的用户画像片段。4.2 实现长期记忆存储模块创建memory/long_term.py使用Chroma向量数据库保存用户历史信息片段import chromadb from chromadb.utils import embedding_functions class LongTermMemory: def __init__(self, user_id: str): self.user_id user_id self.client chromadb.PersistentClient(path./chroma_db) self.collection self.client.get_or_create_collection( nameuser_memory, embedding_functionembedding_functions.DefaultEmbeddingFunction() ) def add_memory(self, text: str, memory_type: str general): self.collection.add( documents[text], ids[f{self.user_id}_{uuid.uuid4().hex}], metadatas[{user_id: self.user_id, memory_type: memory_type}] ) def search(self, query: str, top_k: int 3): results self.collection.query( query_texts[query], n_resultstop_k, where{user_id: self.user_id} ) return results[documents][0] if results[documents] else []需要补充import uuid。这个模块的语义是每个用户拥有独立的记忆片段集合通过user_id过滤避免跨用户串记忆。4.3 实现记忆管理节点创建agent/nodes.py包含三个核心节点函数。第一个节点负责短期记忆管理。它把历史消息保留最近6轮更早的内容不再传给模型from state import AgentState def manage_short_term_memory(state: AgentState) - AgentState: # 假设 state[short_memory] 保存的是 [{role: user/assistant, content: ...}] recent state.get(short_memory, [])[-6:] return {short_memory: recent}第二个节点是“回忆”节点根据当前用户问题从长期记忆中检索相关内容from memory.long_term import LongTermMemory def recall_memory(state: AgentState) - AgentState: user_id state[user_id] query state[query] memory_store LongTermMemory(user_id) recalled memory_store.search(query, top_k3) return {long_memory: recalled}第三个节点是“存储”节点在Agent回答完成后把对用户有价值的陈述写入长期记忆def store_memory(state: AgentState) - AgentState: user_id state[user_id] history state.get(short_memory, []) if len(history) 2: # 取最后一条用户消息作为沉淀来源 last_user_msg history[-1][content] memory_store LongTermMemory(user_id) memory_store.add_memory(last_user_msg, memory_typeuser_statement) return {}真实项目中不建议把用户所有发言都存入长期记忆这样会造成记忆污染。更合理的做法是让大模型先抽取值得记忆的信息再写入存储。4.4 实现模型调用节点模型调用节点要把短期记忆和长期记忆组合成Promptfrom langchain_core.prompts import ChatPromptTemplate from config import create_llm SYSTEM_TEMPLATE 你是一个有记忆能力的智能助手。 以下是关于用户的长期记忆信息可能包含用户偏好、历史关注点 long_memory {long_memory} /long_memory 请结合这些记忆信息以及本次会话内容回答用户问题。 def call_model(state: AgentState) - AgentState: llm create_llm() prompt ChatPromptTemplate.from_template(SYSTEM_TEMPLATE) long_memory_text \n.join(state.get(long_memory, [])) or 暂无已知记忆 messages [ (system, prompt.format(long_memorylong_memory_text)), ] for item in state.get(short_memory, []): role user if item[role] user else assistant messages.append((role, item[content])) response llm.invoke(messages) return {response: response.content}这段代码的核心在于长期记忆被注入System Prompt短期记忆按角色组成多轮消息列表。模型在生成回答前已经知道自己“记得”哪些信息。4.5 构建Graph并运行创建agent/graph.py把节点连接成一张图from langgraph.graph import Graph from agent.nodes import manage_short_term_memory, recall_memory, call_model, store_memory def build_memory_agent(): workflow Graph() workflow.add_node(manage_short_term_memory, manage_short_term_memory) workflow.add_node(recall_memory, recall_memory) workflow.add_node(call_model, call_model) workflow.add_node(store_memory, store_memory) workflow.set_entry_point(manage_short_term_memory) workflow.add_edge(manage_short_term_memory, recall_memory) workflow.add_edge(recall_memory, call_model) workflow.add_edge(call_model, store_memory) workflow.set_finish_point(store_memory) return workflow.compile()运行一次对话from agent.graph import build_memory_agent agent build_memory_agent() def chat(user_id: str, user_message: str, history: list): result agent.invoke({ user_id: user_id, query: user_message, short_memory: history, long_memory: [] }) return result history [] response chat(user_001, 我是油性皮肤想找控油洗面奶, history) print(Agent:, response[response]) history.append({role: user, content: 我是油性皮肤想找控油洗面奶}) history.append({role: assistant, content: response[response]}) response chat(user_001, 有什么品牌推荐, history) print(Agent:, response[response])第二次提问虽然表面上没提肤质但Agent通过长期记忆已经检索到“用户是油性皮肤”这一关键信息推荐结果会更精准。5. 基于DeepAgents的进阶方案5.1 DeepAgents为记忆管理带来了什么DeepAgents的核心价值在于它实现了完整的Agent运行循环也就是“Agent Harness”。在这个循环里Agent会反复思考、调用工具、观察结果直到生成最终答案。它底层与Langchain深度集成因此我们可以把上一节实现的长期记忆能力作为工具提供给DeepAgents。相比自己手写Langgraph节点DeepAgents能减少很多控制流代码。但代价是控制粒度不如Langgraph精细所以它的定位更适合标准Agent任务而不太适合需要复杂状态机编排的业务。5.2 创建带记忆能力的DeepAgent把长期记忆模块封装成工具然后构建DeepAgentfrom deepagents import DeepAgent from langchain_core.tools import tool from config import create_llm from memory.long_term import LongTermMemory tool def search_user_memory(query: str, user_id: str) - str: 从用户长期记忆中检索相关信息参数为查询内容和用户ID。 memory_store LongTermMemory(user_id) results memory_store.search(query, top_k3) return \n.join(results) if results else 暂无相关记忆注意这里的工具参数格式需要根据DeegAgents的版本调整有些版本支持自动从上下文推断user_id。更稳妥的做法是在创建Agent时通过system prompt注入用户信息def create_memory_deep_agent(user_id: str): system_prompt f 你是用户的私人助手当前用户ID是{user_id}。 在回答用户问题之前如果问题涉及到用户偏好或历史信息 请先调用search_user_memory工具进行回忆。 agent DeepAgent.build( llmcreate_llm(), tools[search_user_memory], system_promptsystem_prompt ) return agent构建之后就可以直接调用agent create_memory_deep_agent(user_001) response agent.run(我的肤质是什么来着我适合用什么产品) print(response)DeepAgents会自动完成“调用工具→获取记忆→组织回答”的循环你不需要手动管理中间步骤。5.3 DeepAgents与Langgraph如何选择实际项目中两者可以组合使用。Langgraph适合做外层业务流程控制比如订单处理、多步审批、复杂条件分支DeepAgents适合做单任务内的深度推理比如需要不断调用工具才能完成的咨询问答。在记忆系统层面我的建议是如果你需要精细控制记忆读取时机和上下文拼接方式优先使用Langgraph如果你想快速上线一个具备基本记忆能力的AgentDeepAgents能把工作量减少一半。6. 电商实战案例带记忆的商品推荐助手6.1 场景与需求假设我们要做一个电商导购Agent核心需求是用户第一次咨询时Agent记录用户肤质、价格偏好、品牌偏好。用户后续咨询时Agent自动回忆起这些信息结合当前商品库做个性化推荐。推荐结果要能调用商品搜索工具而不是只靠模型编造商品。这个案例能同时体现短期记忆、长期记忆、工具调用三者的配合。6.2 用户画像存储设计长期记忆模块沿用上一节代码。我们在写入时增加一个类型字段比如“skin_type”、“price_preference”、“brand_preference”。这样后续筛选时更精确。同时可以加一个summarize_and_store函数用大模型抽取用户信息def extract_user_profile(llm, user_message: str) - list: prompt ChatPromptTemplate.from_template( 从用户信息中提取值得长期记忆的结构化内容。 输出JSON数组每个元素格式为 {{type: skin_type, content: 油性皮肤}}。 无法提取时返回空数组。 用户信息{message} ) chain prompt | llm result chain.invoke({message: user_message}) # 这里需要解析JSON建议使用PydanticOutputParser return extract_json_array(result.content)生产环境中请务必加JSON解析校验防止模型输出格式不合法导致程序崩溃。6.3 商品搜索工具创建ecommerce/tools.py模拟一个商品搜索工具。实际项目中可替换为数据库查询或调用电商搜索服务PRODUCT_DB [ {id: 1, name: 控油洁面乳, tags: [油性, 清洁, 控油], price: 159}, {id: 2, name: 保湿爽肤水, tags: [干性, 保湿, 温和], price: 199}, {id: 3, name: 淡斑精华, tags: [中性, 美白, 淡斑], price: 359}, {id: 4, name: 防晒乳, tags: [所有肤质, 防晒, 日常], price: 129}, ] def search_product(tags: list[str], max_price: int 500): results [] for product in PRODUCT_DB: if any(tag in product[tags] for tag in tags) and product[price] max_price: results.append(product) return results6.4 电商助手核心流程在Langgraph中增加工具调用节点让模型决定是否需要搜索商品。由于完整实现需要引入ToolNode相关机制这里给出一个简化但可运行的结构from langgraph.graph import Graph from agent.nodes import recall_memory, store_memory from ecommerce.tools import search_product def recommend_node(state: AgentState) - AgentState: llm create_llm() long_memory_text \n.join(state.get(long_memory, [])) or 暂无已知记忆 prompt f 你是电商导购助手。用户信息如下 {long_memory_text} 请从以下商品中选择合适推荐并说明理由。 商品库 {PRODUCT_DB} response llm.invoke(prompt) return {response: response.content} def ecommerce_agent(): workflow Graph() workflow.add_node(recall_memory, recall_memory) workflow.add_node(recommend, recommend_node) workflow.add_node(store_memory, store_memory) workflow.set_entry_point(recall_memory) workflow.add_edge(recall_memory, recommend) workflow.add_edge(recommend, store_memory) workflow.set_finish_point(store_memory) return workflow.compile()6.5 运行效果说明第一次对话result agent.invoke({ user_id: user_001, query: 我皮肤很油预算300以内有什么推荐的, short_memory: [], long_memory: [] })Agent的回答会偏向控油类商品。回答结束后要点是检查Chroma中是否写入了用户偏好。第二次对话result agent.invoke({ user_id: user_001, query: 给我推荐一款日常用的防晒, short_memory: [], long_memory: [] })由于长期记忆中有“油性皮肤”的记录推荐结果会自动优先考虑质地清爽的防晒产品而不是无差别推荐所有防晒。这个闭环的价值在于用户不需要重复描述自己的肤质和预算体验更接近真人导购。7. 常见问题与排查思路7.1 高频报错与解决问题现象常见原因解决思路启动时提示langchain版本冲突各库要求不同版本统一升级到最新稳定版避免混用新旧APIAgent执行时报“Agent execution terminated due to error.”工具内部异常未捕获为每个工具增加try-except返回错误描述而不是直接抛出长期记忆检索不到内容向量库中还没有写入数据或写入未落盘检查Chroma持久化目录确认调用过add_memory检索结果与当前问题不相关嵌入模型语义能力不足或chunk过大更换更强的Embedding模型按语义块控制存储粒度对话历史越来越长Token消耗过高没有做窗口裁剪在调用模型前裁剪最近N轮消息模型回答时忽略长期记忆System Prompt中记忆信息排布靠后或过于冗长精简记忆内容把最关键的画像信息放在Prompt显眼位置Langgraph中send(node_name, state)用法不清晰对动态并行图理解不到位放到7.2节单独说明7.2 Langgraph中的send到底怎么理解send是Langgraph中用于动态生成并行任务的API。它允许你在一个节点内为多个数据源并行触发同一个后续节点。举个例子假设你拿到了10个商品希望针对每个商品并行调用一个大模型节点生成评论可以在中间节点里from langgraph.types import send def distribute_tasks(state): products state[products] return [send(analyze_product, {product: p}) for p in products]这里的send(node_name, state)表示为每个product生成一条任务发送到analyze_product节点去处理各个任务互相独立Langgraph会并行调度。如果你还没搞懂send可以把它理解为“循环触发同一个下游节点每个元素传一份独立状态”。7.3 排查清单如果Agent记忆功能表现异常可以用以下顺序排查确认短期记忆传入模型前的顺序是否正确通常是System前置历史对话按时间顺序排列。确认长期记忆有数据写入可以直接查询Chroma集合的count()。确认检索时使用了正确的user_id过滤条件避免串记忆。确认模型Prompt中是否真的拼接了长期记忆内容必要时打印最终Prompt调试。8. 工程化最佳实践8.1 记忆内容治理记忆系统的核心风险不是“记不住”而是“什么都记”。如果用户随口说的“我不喜欢蓝色”也被永久保存长期积累下来会形成大量低质量和互相矛盾的信息。建议在写入长期记忆前增加一道“记忆筛选”步骤让大模型判断这段话是否值得记忆、属于什么类型、是否与已有记忆冲突。如果与已有记忆冲突优先采用最新描述。8.2 用户身份隔离与隐私安全企业级系统必须做到用户维度隔离。向量集合中增加user_id元数据并在查询时强制过滤防止A用户检索到B用户的私有信息。在数据库层面建议按租户分collection或增加租户ID过滤条件。涉及用户敏感信息时要遵循最小化原则只记忆为用户服务所必需的信息并在用户注销或明确要求时提供删除接口。生产环境操作涉及用户数据删除时应先备份、走审批流程并在测试环境验证后再执行。8.3 性能与成本控制长期记忆的检索请求不宜每次都触发可以在节点中加入条件判断当用户问题长度过短或属于寒暄类时跳过检索。同时为用户画像增加缓存减少重复的向量查询。写入侧也要控制频率。只有用户表达了新的有效信息时才写入记忆不要每轮对话都调用Embedding和写入操作。这能显著降低成本。8.4 可观测性设计记忆系统是隐性的用户不会直接看到Agent“回忆”的过程但工程师必须能观测到。建议为每个环节增加日志记忆写入日志记录写入时间、用户ID、记忆类型、内容摘要。记忆检索日志记录查询文本、召回结果、相似度得分。Prompt组装日志记录最终传给模型的System Prompt内容。有了这些日志当用户反馈回答异常时可以快速定位是记忆没写入、检索没召回还是Prompt拼接问题。9. 总结本文从Agent记忆的痛点出发完整介绍了短期记忆和长期记忆的设计方案并用Langgraph实现了可控的记忆流短期记忆裁剪、长期记忆检索、模型调用、记忆沉淀。之后引入了DeepAgents作为更轻量的高级Agent方案最后通过电商推荐助手案例串起了完整闭环。如果你想在真实项目中落地建议先跑通第4节的通用记忆Agent再根据业务场景替换模型、向量库和工具层。在动手过程中你可以重点观察三个问题哪些用户信息真正影响了回答质量、检索召回是否有噪音、记忆写入是否过度。这三个问题的答案往往决定了记忆系统的ROI。希望这篇长文能帮你把“大模型有记忆”从演示项目推进到可维护的工程系统。