企业级Agent记忆系统实战：Langgraph与DeepAgents构建短期与长期记忆

大模型做 Agent，聊到最后基本都会卡在同一个问题上：记忆。上下文一长就丢人设，用户隔几天再来就认不出，业务数据没法跨会话复用——这些问题不解决，Agent 就只能在 Demo 里待着。这次来看的方案来自码士集团分享的企业级 Agent 记忆系统，技术栈是 Langchain + Langgraph + DeepAgents，一套代码同时覆盖短期记忆、长期记忆，并且拿电商场景做了完整演示。

先给出我对这套方案的整体判断：它没有把记忆做成一个孤立的“大模型套娃”，而是把记忆拆成两层。短期记忆交给 Langgraph 的线程化状态（Checkpointer），长期记忆交给“记忆抽取 + 向量存储 + 检索注入”，业务编排交给 DeepAgents 的 Supervisor 机制。每一层都可以单独替换、单独测试，这正是企业级系统最看重的模块化能力。

这篇文章会按照“为什么需要记忆 -> 短期记忆怎么落 -> 长期记忆怎么落 -> DeepAgents 怎么编排 -> 电商案例怎么跑通”的顺序展开，最后补上接口封装、批量任务、资源占用和排错清单。适合正在做智能客服、知识库问答、营销助手，或者准备把 Agent 接入生产业务的人，可以收藏备用。

1. 核心能力速览

先把关键信息放在前面。这套 Agent 记忆系统不是单一工具，而是一条完整技术链路，核心能力如下：

这里要提醒一点：如果你只想要一个“给大模型加系统提示词”的简单方案，这套体系对你来说偏重。它解决的是多轮、跨会话、多业务系统联动时的记忆一致性问题，价值出现在场景复杂度足够高的时候。

2. Agent 记忆体系设计思路

2.1 Agent 需要哪几类记忆

业界对 Agent 记忆的分类不完全统一，但按这套方案可以分成三层：

• 短期记忆（Working Memory）：当前会话内的上下文，比如用户本轮问了什么、上一轮回答了什么。这是对话连贯性的基础。
• 长期记忆（Long-term Memory）：跨会话的知识沉淀，比如用户的偏好、身份、历史需求、关注点。这解决“用户隔一天再来，Agent 还记得他”的问题。
• 业务记忆（Business Context）：用户此刻的业务状态，比如最新订单、物流节点、优惠券状态、售后进度。这类信息不适合长期固化管理，更适合按需从业务系统实时查询，然后注入上下文。

这套记忆系统的关键设计，就是把这三类记忆用不同技术承载：短期记忆靠 Langgraph 状态和 Checkpointer，长期记忆靠向量库，业务记忆靠工具调用。三者不混在一起，逻辑清晰，排查也容易。

2.2 Langgraph 在记忆系统中的定位

Langgraph 本质上是一个“可持久化状态”的图执行框架。Agent 的每次运行都被拆成节点，节点之间通过共享状态传递信息。在这个方案里：

• 节点负责执行具体工作，比如“判断用户意图”、“生成回复”、“更新记忆”。
• 状态负责保存对话消息和业务数据。
• Checkpointer 负责把状态持久化到存储后端，而这是短期记忆的技术载体。

因为 Langgraph 天然支持按 thread_id 隔离状态，所以短期记忆不需要额外写一套 Redis 缓存逻辑。只要在编译图时挂一个 Checkpointer，每次调用时带上 thread_id，Langgraph 就会自动把会话历史保存下来，并在下一次调用时恢复。这是整个记忆系统里最省事、最可靠的一层。

2.3 DeepAgents 在记忆系统中的定位

DeepAgents 是 LangChain 生态在多 Agent 编排方向的一个重要实现。它没有另起炉灶，而是基于 Langgraph 的底层机制扩展出 AgentSupervisor 模式。一个 Supervisor 负责理解用户意图、分配任务、汇聚结果，多个 Tool Agents 分别承担不同职责，比如客服对话、订单查询、商品推荐、售后处理。

在记忆系统中，DeepAgents 的价值在于：它让“记忆”不再只是跟在主 Agent 后面的一堆上下文，而是变成所有子 Agent 都可以读取的共享信息层。Supervisor 在派发任务前先决定“这个问题需要读取哪些记忆”；Tool Agents 在执行任务时也可以主动触发记忆更新。记忆不是某个 Agent 独有的，而是整个多 Agent 团队共享的基础设施。

2.4 记忆系统不是“把更多资料检索出来”

现在很多 Agent 的“记忆”其实做成了挂载一堆文档再检索。但真正的长期记忆应该更接近“学会回忆”。这套方案里有几个和单纯 RAG 完全不同的设计点：

• 记忆是结构化的：不是把历史对话塞进上下文，而是抽取出“用户是谁、喜欢什么、最近买了什么、有哪些未解决的问题”，以结构化数据存下来。
• 记忆是选择写入的：不是每句话都入库，而是由抽取模型判断“这段对话里有没有值得长期记住的信息”，有才写入。这样不会产生记忆污染。
• 记忆是主动召回的：回答之前先判断“当前问题需要哪些历史记忆”，再精确检索，而不是把积累的几个月的记忆全部堆给大模型。
• 记忆是允许遗忘和合并的：后续检索到相似记忆时，可以合并冲突记录、删除过期偏好，保证用户画像不被旧数据干扰。

这个思路和社区最近讨论较多的 RippleMem、Agent 记忆反思机制是同一个方向，核心都是让 Agent 具备“回忆能力”而不是“检索能力”。

3. 适用场景与使用边界

3.1 适合谁用

从这套方案的技术选型来看，最适合的是以下四类场景：

• 电商客服：跨会话记住用户的收货偏好、尺码、风格偏好，减少用户反复描述需求。
• 企业知识库助手：记住员工身份、所属部门、近期项目，让同一个知识库对不同人给出差异化回答。
• 销售与营销助手：记录客户跟进状态、关注点、历史报价，减少销售手工翻阅记录的时间。
• 用户运营系统：基于用户历史行为和偏好，让推荐和触达内容更个性化。

3.2 不适合什么场景

• 单轮问答：没有跨会话需求时，长期记忆是多余开销。
• 极低成本场景：记忆抽取、向量检索、状态持久化都会增加 Token 消耗和存储成本。
• 强实时场景：如果要求毫秒级响应，记忆读取和业务工具编排的全链路延迟需要专门优化，普通方案可能不满足。
• 一次性工具类 Agent：比如只做一次代码解释或格式转换的 Agent，根本不存在“认识用户”的需求。

3.3 合规与安全边界

涉及用户记忆，就必须把隐私合规放在第一位。下面是这套方案在落地时必须要处理的边界：

• 明确告知：记录用户偏好和历史行为前，应在产品协议或交互流程中告知用户，并在合理范围内征得同意。
• 脱敏处理：身份证号、手机号、家庭住址、支付信息等敏感字段不建议写入长期记忆，必要时先做脱敏或只存业务侧脱敏 ID。
• 数据可删除：记忆系统必须提供“删除指定用户记忆”的能力，否则用户提出注销请求时无法响应。
• 授权边界：涉及人脸、声音、肖像等生物识别或个性化生成内容时，必须确认已获得明确授权，不能拿未授权素材做画像或跨场景使用。
• 业务数据权限：接入订单、优惠券等领域数据时，要遵循最小权限原则，Agent 只能读取它实际需要的数据。

4. 环境准备与前置条件

4.1 软件环境

这套方案是基于 Python 生态的，建议准备好以下环境：

注意：LangChain、Langgraph、DeepAgents 的版本迭代比较快，安装时建议直接安装最新稳定版。版本差异导致的接口变动，以官方文档为准。

4.2 创建虚拟环境并安装依赖

# 创建虚拟环境 python -m venv .venv source .venv/bin/activate # Windows 下使用 .venv\Scripts\activate # 安装核心依赖，这里给出通用安装命令，版本以官方最新为准 pip install langchain pip install langchain-openai pip install langgraph pip install langgraph-checkpoint-sqlite pip install langchain-deepagents pip install chromadb pip install fastapi uvicorn

如果你的项目需要数据库持久化和 API 服务，可以额外安装：

pip install sqlalchemy pip install pydantic

4.3 模型与存储配置

在项目中新建.env文件，按实际使用的模型服务填写：

OPENAI_API_KEY=your_api_key_here OPENAI_BASE_URL=https://your-model-service.example.com/v1 LLM_MODEL=gpt-4o-mini # 记忆向量库目录 MEMORY_DB_PATH=./memory_db CHECKPOINT_DB_PATH=./checkpoints.db

这里的OPENAI_BASE_URL可以指向任何 OpenAI 兼容服务。如果你用本地模型，可以把地址改成http://127.0.0.1:8000/v1。后面代码统一通过环境变量读取模型配置，方便切换厂商。

5. 短期记忆落地：Langgraph Checkpoint 会话持久化

5.1 为什么短期记忆要用 Checkpointer

在对话应用中，最常见的做法是手动把历史消息拼进 prompt，用户一刷新就丢了。Langgraph 的 Checkpointer 解决了两个核心问题：

• 自动保存：每次图执行结束，状态自动写入存储后端。
• 按会话恢复：带上 thread_id，就能把历史状态完整恢复。

这意味着你不需要自己设计“会话历史”的数据库表，不需要手动 truncate 历史，Langgraph 会按图执行状态来管理。

5.2 带 Checkpointer 的最小客服 Agent

from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver from langgraph.graph import MessagesState from langchain_openai import ChatOpenAI # 初始化模型，参数通过环境变量读取 model = ChatOpenAI(model="gpt-4o-mini") # 定义一个最简单的 Agent 节点 def agent_node(state: MessagesState): response = model.invoke(state["messages"]) return {"messages": [response]} # 构建图 graph = StateGraph(MessagesState) graph.add_node("agent", agent_node) graph.add_edge(START, "agent") graph.add_edge("agent", END) # 注入 Checkpointer，这是短期记忆的关键 checkpointer = MemorySaver() app = graph.compile(checkpointer=checkpointer) # 第一轮对话：用户自我介绍 config = {"configurable": {"thread_id": "user-001"}} app.invoke( {"messages": [("user", "我叫小李，喜欢黑色机械键盘，预算一千以内。")]}, config, ) # 第二轮对话：Agent 应该还能记得用户身份和偏好 result = app.invoke( {"messages": [("user", "帮我看一下合适的键盘品牌")]}, config, ) print(result["messages"][-1].content)

运行逻辑很简单：同一个thread_id下的所有调用共享状态，Agent 直接利用历史消息生成回答。这是短期记忆最小可用的实现。

5.3 从内存 Checkpointer 切换到 SQLite

生产环境不能只用MemorySaver，服务一重启记忆就没了。切换 SQLite 版本：

from langgraph.checkpoint.sqlite import SqliteSaver # DB 文件会持久化到磁盘 with SqliteSaver.from_conn_string("./checkpoints.db") as checkpointer: app = graph.compile(checkpointer=checkpointer) # 后续逻辑与内存版一致

切换后，即使服务重启，只要 thread_id 不变，会话状态还能恢复。

5.4 短期记忆验证清单

• 同一 thread_id 连续两轮对话，Agent 能回答“我是谁”。
• 换一个 thread_id，Agent 不再记得上一轮的内容，确认会话隔离生效。
• 重启服务后，用 SQLite Checkpointer 仍能恢复历史会话。
• 历史消息越长，观察是否出现延迟或上下文超限，设计截断策略。

6. 长期记忆落地：用户画像与偏好存储

短期记忆解决的是“当前会话连贯”，长期记忆解决的是“跨会话认识用户”。长期记忆的核心流程是：抽取 -> 存储 -> 检索 -> 注入。

6.1 定义长期记忆的数据模型

长期记忆不是原样保存对话文本，而是保存结构化的用户画像。可以按下面的模式设计：

{ "user_id": "user-001", "preferences": { "keyboard_style": "mechanical", "color": "black", "budget_range": "1000" }, "facts": [ "用户叫小李", "用户喜欢机械键盘", "用户偏好黑色" ], "last_interaction": "2025-02-15 10:30:00" }

这里的原则是：能结构化的字段尽量结构化，不能结构化的信息以短句形式保存为 facts，再配合向量检索召回。

6.2 从对话中抽取记忆

每次对话结束后，可以调用一个“记忆抽取模型”，判断这段对话里有没有值得长期记住的信息：

from langchain_openai import ChatOpenAI from pydantic import BaseModel, Field class MemoryItem(BaseModel): memory_type: str = Field(description="记忆类型，如 preference / fact / requirement") content: str = Field(description="记忆内容") importance: int = Field(description="重要程度，1-10") class MemoryExtractionOutput(BaseModel): memories: list[MemoryItem] extract_prompt = """用户与客服发生了以下对话。 请从中抽取值得长期记住的用户偏好或事实。 如果没有值得记住的信息，返回空列表。 只输出结构化结果。 对话： {conversation}""" def extract_memories(user_id: str, conversation: str) -> list[MemoryItem]: llm = ChatOpenAI(model="gpt-4o-mini") llm_with_structure = llm.with_structured_output(MemoryExtractionOutput) memories = llm_with_structure.invoke(extract_prompt.format(conversation=conversation)) return memories.memories

抽取模型需要独立调一次 LLM，会产生额外 Token 消耗。优化方向是只在对话长度达到一定阈值、或者用户明确表达了偏好时才触发抽取，而不是每一轮都抽。

6.3 记忆写入与向量检索

抽取出的记忆先做规范化处理，再写入向量库：

import chromadb from langchain_openai import OpenAIEmbeddings # 初始化向量库 client = chromadb.PersistentClient(path="./memory_db") collection = client.get_or_create_collection("user_memories") embeddings = OpenAIEmbeddings(model="text-embedding-3-small") def write_memory(user_id: str, memory: MemoryItem): doc_id = f"{user_id}-{memory.memory_type}-{len(memory.content)}" collection.upsert( ids=[doc_id], documents=[memory.content], metadatas=[ { "user_id": user_id, "memory_type": memory.memory_type, "importance": memory.importance, } ], embeddings=[embeddings.embed_query(memory.content)], )

查询时，先按用户 ID 过滤，再按向量相似度召回：

def recall_memories(user_id: str, query: str, top_k: int = 5) -> list[str]: results = collection.query( query_embeddings=[embeddings.embed_query(query)], where={"user_id": user_id}, n_results=top_k, ) return results["documents"][0]

检索时的query建议使用“当前用户问题 + 用户 ID”，而不是只传原始问题。比如用户说“帮我推荐键盘”，原始问题里没有“黑色”“机械”这些关键词，但向量检索仍能根据语义把相关偏好召回。这就是长期记忆比关键词过滤更实用的原因。

6.4 记忆写入后如何参与生成

召回的记忆不是直接塞进系统提示词，而是经过拼接加入 Agent 的上下文：

def build_context_with_memory(user_id: str, query: str) -> str: memories = recall_memories(user_id, query) if not memories: return "" return "根据用户的历史记忆：\n" + "\n".join(f"- {mem}" for mem in memories)

在实际 Langgraph 中，应该把“记忆读取”设计成一个节点，放在主对话节点之前执行。它从向量库召回记忆，写入当前状态，然后再交给 DeepAgents 的 Supervisor 做意图判断。

6.5 长期记忆的两个常见问题

• 记忆冲突：用户上一周说喜欢黑色，这周说想换白色。写入新记忆的同时，应该对相似旧记忆做合并或失效。
• 记忆膨胀：随着时间推移，用户画像会越来越大。建议定期对同一用户的记忆做一次压缩总结，只保留最新的偏好和重要事实。

7. DeepAgents 多 Agent 编排与记忆共享

7.1 DeepAgents 解决什么

单 Agent 系统里，记忆、工具、模型是绑在一起的。业务复杂后会出现三种问题：工具太多导致提示词过长、不同职责混在一个 Agent 里难以维护、记忆被不同类型任务争夺。DeepAgents 的设计模式是拆成多个专用 Tool Agents，由一个 Agent Supervisor 统一调度。

在这套企业级 Agent 记忆系统中，DeepAgents 不是替代 Langgraph，而是跑在 Langgraph 之上的一层编排策略。

7.2 DeepAgents 的最小接入示例

下面的代码是基于 DeepAgents 通用设计模式的示例。不同版本的导入路径和构造函数可能不同，请以官方文档为准。

from langchain_deepagents import AgentSupervisor, ToolAgent # 1. 定义多个专业 Tool Agent agent_01_tools = [ # 商品查询工具、订单查询工具等 ] agent_02_tools = [ # 售后工具、物流工具等 ] # 2. 创建 Supervisor，管理两个子 Agent supervisor = AgentSupervisor( name="customer_service_supervisor", sub_agents=[ ToolAgent(name="product_browsing", tools=agent_01_tools), ToolAgent(name="after_sales_service", tools=agent_02_tools), ], model=model, # 这里可以传入对话历史，短期记忆会展开成上下文 prompt="你是电商客服中心的主管，负责判断用户请求应该交给哪个 Agent 处理。", )

7.3 记忆在 DeepAgents 中的打通方式

要让记忆和 DeepAgents 配合，最简单的方式是记忆读取节点先行：

用户请求 -> 记忆读取节点（从向量库召回用户偏好） -> Supervisor 编排 -> 商品浏览 Agent（带记忆上下文） -> 订单查询 Agent（调业务工具） -> 结果汇总 -> 记忆更新节点（抽取新记忆并写入）

这样安排的好处是：记忆上下文对所有子 Agent 可见，子 Agent 不需要各自重复查询用户画像；同时每个子 Agent 保持职责单一，后续加新的业务 Agent 不影响已有链路。这也是这套体系能支撑电商多业务线的原因。

8. 电商案例：带记忆的智能客服 Agent 实战

8.1 案例需求

现在把这套记忆系统落地到一个电商客服场景。需求如下：

• 用户首次对话时，能记住用户姓名、商品偏好、预算。
• 用户再次咨询时，不需要重复描述需求，Agent 能直接引用历史偏好推荐。
• 能实时查询订单状态、物流信息。
• 涉及敏感操作（改收货地址、退款）时，调用业务工具前置校验权限。

8.2 图结构设计

用 Langgraph 实现时，图节点可以这样设计：

START -> load_memory（读取用户长期记忆） -> supervisor（DeepAgents 判断意图并调用子 Agent） -> store_memory（抽取并写入新记忆） -> END

8.3 核心代码

下面的代码是一个可以运行的参考骨架，直接把上面的短期记忆和长期记忆代码串起来：

import os from langgraph.graph import StateGraph, START, END, MessagesState from langgraph.checkpoint.memory import MemorySaver from langchain_openai import ChatOpenAI from langgraph.prebuilt import ToolNode class AgentState(MessagesState): user_id: str = "" memory_context: str = "" # 工具 1：查询订单 def query_order(order_id: str) -> str: # 生产环境接入 ERP / 订单中心，这里用演示数据返回 return f"订单 {order_id} 已发货，预计 3 天后送达。" # 工具 2：推荐商品 def recommend_product(query: str) -> str: # 生产环境接入商品中心，这里用演示数据返回 return f"根据需求：{query}，推荐 87 键黑色机械键盘，价格 899 元。" # 记忆读取节点 def load_memory(state: AgentState) -> dict: user_id = state.get("user_id", "") raw_query = state["messages"][-1].content memories = recall_memories(user_id, raw_query) memory_context = "" if memories: memory_context = "用户历史记忆：\n" + "\n".join(f"- {m}" for m in memories) return {"memory_context": memory_context} # 记忆更新节点 def store_memory(state: AgentState) -> dict: user_id = state.get("user_id", "") conversation = "\n".join([f"{m.type}: {m.content}" for m in state["messages"]]) # 为了控制成本，可以只抽取最近两轮 extracted = extract_memories(user_id, conversation[-2000:]) for mem in extracted: if mem.importance >= 6: # 只存重要记忆 write_memory(user_id, mem) return {} # Agent 节点：把记忆和工具一起交给 Supervisor / 模型 def agent_node(state: AgentState): model_with_tools = ChatOpenAI(model="gpt-4o-mini").bind_tools([query_order, recommend_product]) prompt = f"""你是电商客服助手。 {state.get('memory_context', '')} 请基于用户历史记忆和当前问题，使用工具回答。""" response = model_with_tools.invoke( [{"role": "system", "content": prompt}] + state["messages"] ) return {"messages": [response]} # 工具执行节点 tools = ToolNode([query_order, recommend_product]) graph = StateGraph(AgentState) graph.add_node("load_memory", load_memory) graph.add_node("agent", agent_node) graph.add_node("tools", tools) graph.add_node("store_memory", store_memory) graph.add_edge(START, "load_memory") graph.add_edge("load_memory", "agent") graph.add_edge("agent", "tools") graph.add_edge("tools", "agent") graph.add_edge("agent", "store_memory") graph.add_edge("store_memory", END) checkpointer = MemorySaver() app = graph.compile(checkpointer=checkpointer)

实际调用：

config = {"configurable": {"thread_id": "user-001"}} # 第一轮：用户自我介绍并提出偏好 app.invoke( { "messages": [("user", "我叫小李，想买一款黑色机械键盘，预算一千以内。")], "user_id": "user-001", }, config, ) # 第二轮：用户不重复需求，直接问推荐 result = app.invoke( { "messages": [("user", "有什么适合我的键盘推荐吗？")], "user_id": "user-001", }, config, ) print(result["messages"][-1].content)

这个案例跑通后，你能观察到一个关键现象：第一轮对话结束后，store_memory节点已经抽取并写入了“黑色”“机械键盘”“预算一千以内”等记忆；第二轮对话时，load_memory节点把记忆读出来加入上下文，即使第二轮的原始问题没有包含这些偏好词，模型也会基于记忆生成更精准的推荐。

8.4 案例验证清单

• 第一轮说“我喜欢黑色机械键盘”，第二轮问“推荐键盘”，模型能否输出包含黑色、机械特征的答案。
• 修改 thread_id 后再问同样问题，模型是否不再认识用户，证明记忆隔离有效。
• 查看 SQLite / Chroma 是否写入了新记忆。
• 删除对应用户的向量记录后，模型是否回到无记忆状态，确认删除逻辑可用。

9. 服务化部署：FastAPI 接口与批量任务

9.1 为什么需要服务化

电商客服系统通常不是给终端用户直接跑 Python 脚本，而是把 Agent 包装成内部服务，供网页、小程序、客服工作台调用。用 FastAPI 包装是最直接的方式。

9.2 FastAPI 封装示例

from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() class ChatRequest(BaseModel): user_id: str message: str thread_id: str = "" @app.post("/agent/chat") def chat(req: ChatRequest): config = {"configurable": {"thread_id": req.thread_id or req.user_id}} result = app_graph.invoke( { "messages": [("user", req.message)], "user_id": req.user_id, }, config, ) return { "reply": result["messages"][-1].content, "user_id": req.user_id, "thread_id": req.thread_id, }

启动服务：

uvicorn main:app --host 127.0.0.1 --port 8000

9.3 curl 调用示例

服务启动后，可以用 curl 验证接口是否可用：

curl -X POST http://127.0.0.1:8000/agent/chat \ -H "Content-Type: application/json" \ -d '{"user_id": "user-001", "message": "帮我推荐一款键盘", "thread_id": ""}'

接口正常返回后，就可以把http://127.0.0.1:8000/agent/chat接入前端、企业微信机器人、微信客服、钉钉机器人等渠道。

9.4 批量任务设计

记忆系统里有两类典型批量任务：

• 历史会话批量记忆抽取：每天定时把前一天所有客服会话拉出来，逐段抽取记忆并写入向量库。
• 记忆失效与合并：定期扫描过期记忆、冲突记忆，进行合并或删除。

批量任务建议用独立脚本或消息队列触发，避免阻塞在线接口：

# 批量抽取历史会话记忆 python scripts/backfill_memories.py --date 2025-02-14 --batch-size 100

脚本里每个批次处理 100 条会话，处理完成后写日志、记录游标，支持断点续跑。若某批次失败，直接把该批次重新入队即可。

10. 资源占用与性能观察

这套系统的瓶颈通常不在显存，而在 LLM 推理延迟、向量库检索延迟和 Token 消耗。

10.1 内存与磁盘占用

• 短期记忆：使用 SQLite Checkpointer 时，磁盘占用取决于会话消息量。生产环境建议定期清理超过 30 天的旧会话，或转移到冷存储。
• 长期记忆：Chroma 等向量库的磁盘占用取决于记忆条数和 embedding 维度。用户量大的场景建议按用户 ID 分片存储。
• 内存：FastAPI 服务和向量库都会占用内存。可以观察进程 RSS，如果峰值过高，把向量库查询改为异步或加连接池限制。

10.2 Token 消耗如何控制

记忆系统最大的成本点是自动记忆抽取和长期记忆注入。建议这样控制：

• 记忆抽取只在对话包含明显偏好时触发，或者用较低的采样频率。
• 检索回传的记忆条数控制在 3 到 5 条，不要把所有历史都注入上下文。
• 长对话做截断或摘要，避免把整个历史消息都传给模型。
• 对记忆抽取结果做缓存，同一用户一分钟内不重复抽取。

10.3 延迟观察点

调用一次带记忆的 Agent，典型链路是：

FastAPI 收到请求 -> 读取用户记忆（向量检索） -> LLM 思考 -> 调用业务工具 -> LLM 再次生成 -> 返回响应

每一步都会产生延迟。建议在关键节点打印耗时日志，或者在 LangSmith 中配置链路追踪，观察哪一步最慢。工具调用如果超过 3 秒，业务方会明显感受到卡顿，需要考虑缓存或异步预取。

11. 常见问题与排查方法

12. 最佳实践与使用建议

12.1 技术侧建议

• 第一次接入先小规模验证：只用单个用户的 thread_id 跑通“短期记忆 + 长期记忆 + 工具调用”三条链路，再扩展到全量用户。
• 保留一套最小可运行配置，方便复现问题。把环境变量、向量库目录、Checkpointer 路径整理成示例配置，提交到团队知识库。
• 记忆存储分目录管理：模型配置、向量库、会话检查点、批处理日志分别放不同目录，避免互相污染。
• 批量任务的失败重试机制要提前设计：建议使用游标表记录每个用户的处理状态，失败批次可重新入队。
• 接口服务不要直接暴露到公网，内部调用也要加 Access Token 或 IP 白名单。

12.2 业务与合规侧建议

• 用户画像和偏好数据要有明确的保存周期。建议在数据模型中保存created_at和expires_at，由定时任务负责清理过期数据。
• 涉及用户隐私的问答，Agent 应主动拒绝记录和回答敏感信息。
• 接入电商订单、发票等业务数据时，先判断当前用户的会话权限，防止越权查询他人订单。
• 如果未来要把记忆数据用于模型微调或数据分析，需要单独确认数据使用授权。

12.3 维护建议

记忆抽取模型随着业务变化会逐渐出现偏差。建议每个月抽一批真实客服会话，人工检查记忆抽取准确性，及时调整提示词或模型。向量库也是长期运行的服务，建议定期做快照备份，防止数据损坏后用户画像全部丢失。

13. 总结与下一步

这套基于 Langchain + Langgraph + DeepAgents 的企业级 Agent 记忆系统，最值得尝试的是它把短期记忆、长期记忆、业务记忆分成了三个独立层次，真正做到了“该记住的记住、该查询的查询、该遗忘的遗忘”。相比单纯给模型加长上下文，它更接近生产级 Agent 的形态。

如果现在准备动手，建议最先验证一个最小闭环：用同一个 thread_id 跑两轮对话，再插入load_memory和store_memory两个节点，观察从第一轮抽取偏好到第二轮引用偏好的完整链路。最容易踩的坑是向量库的user_id过滤条件和 Langgraph 的thread_id没有对齐，导致记忆读不到或读到别人的记忆。

后续可以沿着三个方向扩展：一是把 Checkpointer 替换为 Postgres，支持多实例部署和会话共享；二是把长期记忆的检索升级为混合检索（向量 + 关键词 + 业务标签）；三是在 DeepAgents 中增加专门的“记忆维护 Agent”，自动完成记忆合并、冲突解决和过期清理。这套体系跑通后，再把客服能力扩展到营销、售后、私域运营，就只是增加子 Agent 的事情了。建议收藏备用，等技术栈更新后回来对照调整。