企业级Agent落地时最容易被低估的技术点不是模型选型而是记忆系统。很多团队在Demo阶段觉得Agent的效果还不错一旦进入生产环境就会发现几个老问题反复出现用户上一轮说过的偏好这一轮就忘同一个用户在多轮任务中提到的关键信息散落在不同会话里无法聚合Agent只能依赖Prompt中的Context上下文一长成本和延迟又快速上升。归根结底是因为模型本身没有记忆机制每次调用都是无状态的所谓的“记住”要么靠上下文拼接要么靠外部存储背后需要一套完整的工程架构去支撑。本文会围绕Context、短期记忆、工作记忆和长期记忆四个层次从原理到LangChain、LangGraph实现再到DeepAgent框架带来的生产治理思路完整梳理企业级Agent记忆系统的设计方法。适合正在做Agent应用开发、想解决“Agent没记性”问题的后端开发者也适合刚入门LangGraph、想理解记忆模块怎么设计的读者。文章会先讲清楚记忆分层的背景和三个框架的定位然后给出一个可以用LangGraph跑通的长期记忆Demo包括SQLite持久化、短期会话记忆、条件路由和记忆写入节点最后补上生产环境必须考虑的安全、审计、召回策略和效果评估。你可以把本文当成一份“从原理到落地”的工程笔记来读。1. 背景与核心概念1.1 为什么Agent会“失忆”大语言模型本身是一个无状态函数你输入一段文本它输出一段文本调用结束后模型内部不会保留任何关于用户的信息。为了让Agent表现得“有记忆”常见做法是把历史对话、用户信息、业务规则全部塞进Prompt的Context窗口里。这种方式在小规模场景下可行但它本质上是在“临时搬运”信息而不是在“沉淀”信息。当应用进入企业级场景后问题会变得非常具体。用户可能在一个星期内多次访问同一个Agent每次访问都带着新的任务但任务之间相互关联用户可能在某个会话里提到“我们公司的审批流程是三级审批”另一个会话里又提到“这个月预算已经用完”Agent需要把这些分散的事实整合成对用户的稳定理解。这些需求已经超出了单纯Context拼接的范畴需要引入一套记忆系统把短期会话信息、任务中间状态和长期用户画像分开管理。从架构上看Agent的记忆系统至少需要回答四个问题记什么、存在哪、怎么取、什么时候忘。记什么决定了我们抽取的是用户偏好、关键事实还是历史结论存在哪决定了用数据库、向量库还是图谱怎么取决定了是关键词检索、向量相似度还是规则命中什么时候忘则决定了记忆的更新、过期和删除策略。这四个问题没有统一答案但有一个稳定的分层模型可以参考。1.2 记忆分层从Context到Long-term Me在工程上我习惯把Agent的记忆分成四层这四层常常被混为一谈但它们的存储方式、更新频率和生命周期完全不同。第一层是模型Context也就是当前Prompt窗口内的内容。它最直接但也最不可靠。一旦超出Token上限最老的信息会被截断即使没有超限把所有历史都堆进Context也会让模型注意力分散、推理变慢。第二层是短期记忆也叫会话记忆通常指一次会话内的多轮对话历史可以通过LangGraph的Checkpointer机制保存。第三层是工作记忆指当前任务执行过程中的临时状态比如一个多步骤任务已经走到哪一步、临时变量是什么这在LangGraph里对应State的设计。第四层是长期记忆它需要跨会话、跨任务保存内容包括用户画像、偏好、历史结论、领域知识是Agent在长期交互中形成的对用户的稳定认知模型也就是标题里说的Long-term Me。这四层之间并不是互斥的而是一条数据流水线。短期记忆和工作记忆负责当次会话与当次任务长期记忆负责把有价值的沉淀抽取出来并在后续会话中通过检索重新注入Context。理解了这个流水线再看LangChain、LangGraph和DeepAgent的定位就会清晰很多。1.3 LangChain、LangGraph与DeepAgent的定位很多开发者刚接触这三个名词时会困惑它们到底是不是同一个东西其实不是。LangChain是一个组件生态提供了模型封装、Prompt模板、工具调用、向量库集成等能力解决的是“和模型对话、调工具”的便利性问题。LangGraph是一个基于状态图的Agent编排框架强调的是节点、边、条件路由、循环和检查点解决的是“复杂Agent流程怎么控制”的问题。DeepAgent则更像一类面向企业级Agent生产化的框架或工程理念它不只是帮你写工作流更关注长期身份、可观测性、安全边界、记忆生命周期这些治理问题。用一个不太严谨的比喻来说LangChain像工具箱LangGraph像流水线控制系统DeepAgent像工厂的管理制度。工具决定了你能干什么流程控制决定了任务怎么走管理制度决定了长期稳定运行需要哪些规范和边界。本文的实战部分主要用LangGraph来实现记忆闭环而DeepAgent的思路会贯穿在第5章的生产治理建议中。维度LangChainLangGraphDeepAgent核心定位组件库与工具链Agent状态图编排企业级Agent生产治理框架关注重点模型调用、Prompt、工具、向量库节点、边、条件路由、循环、检查点长期记忆、身份、安全、审计、生命周期使用场景快速实现LLM应用复杂多步Agent流程生产级Agent平台落地记忆视角提供历史消息封装用Checkpointer和State实现记忆机制治理长期记忆的沉淀、权限与遗忘三者可以配合使用并不冲突。2. 环境准备与版本说明在开始写代码之前先明确环境和项目结构。本文示例以Python 3.10为基础核心依赖是LangChain、LangGraph和一个支持OpenAI接口的大模型服务。由于LangChain和LangGraph迭代速度很快具体的API名称在不同版本中可能有细微差异建议你在实际项目中锁定已验证的版本或者在虚拟环境里先安装最新版本再按官方文档微调。本文演示的重点是记忆系统的工程思路而不是某个特定版本的新特性所以下面的代码你在2024年之后的主流版本中基本都能运行。如果你使用的是更新的大版本遇到个别接口调整优先查看官方迁移文档。python -m venv .venv source .venv/bin/activate # Windows 下使用 .venv\Scripts\activate pip install langchain langchain-openai langgraph项目结构如下。这个结构不复杂但能清晰区分记忆存储、状态定义、图编排和入口调用。agent-memory-demo/ ├── memory_store.py # 长期记忆存储基于 SQLite ├── state.py # 定义 AgentState ├── graph.py # 构建 LangGraph 图 └── main.py # 命令行交互入口LLM服务建议使用OpenAI兼容接口也可以使用国内主流模型的兼容网关只需要设置OPENAI_API_KEY和可选的OPENAI_BASE_URL。模型名称通过环境变量LLM_MODEL控制方便切换。3. 记忆系统的核心工程拆解3.1 Context管理与Token预算Context是模型一次调用能看到的所有内容。很多初学者以为“给模型更多历史效果一定更好”实际上历史消息越多Token成本越高模型对关键信息的敏感度也会下降。企业级Agent必须对Context做预算管理不是把所有历史都塞进去而是选择当前任务最相关的内容。一种常见的做法是把记忆分为“必带信息”和“按需检索信息”。必带信息包括系统提示词、当前用户请求、最近一两轮的对话按需检索信息则从长期记忆库中查找比如用户偏好、历史结论、领域知识只在需要时注入。这样既保证了Agent对当前任务的理解又控制了成本。LangGraph里通过节点的组合可以很容易实现这种设计这一点会在第4章的代码中体现。3.2 短期记忆会话历史与Checkpointer短期记忆的核心是保存多轮对话历史。LangGraph提供了Checkpointer机制会把图执行到某个节点时的State保存下来下次以相同thread_id继续调用时可以自动恢复历史状态。这里说的State就相当于Agent的工作记忆而Checkpointer则实现了会话级记忆的持久化。官方提供的MemorySaver适合本地演示但它是内存级的服务重启后数据会丢失。生产环境建议使用SqliteSaver、PostgresSaver等持久化Checkpointer把会话状态保存到数据库。也就是说短期记忆和工作记忆本质上都是“状态的持久化”只是保存的粒度不同。3.3 工作记忆LangGraph中的StateLangGraph的State是节点之间传递数据的唯一方式。你在StateGraph(AgentState)中定义的TypedDict字段就是所有节点共享的工作记忆。节点函数可以读取state也可以返回一个字典来更新state中的字段。对于需要累积的字段比如对话历史messages需要使用Annotated类型和合并函数否则每次返回都会覆盖旧值。这一点非常容易踩坑。举个例子如果messages字段不加合并逻辑Agent在第二轮回合中可能只看到当前轮消息而丢失了第一轮的上下文。from typing import Annotated, TypedDict from operator import add from langchain_core.messages import BaseMessage class AgentState(TypedDict): messages: Annotated[list[BaseMessage], add] # 对话历史自动累积 memories: list[str] # 当前会话召回的长时记忆 new_facts: list[str] # 本轮需要保存的新事实3.4 长期记忆存储、写入、召回与遗忘长期记忆和短期记忆的最大区别在于生命周期。长期记忆需要跨会话保存并且通常不是原始对话文本而是从对话中抽取出来的结构化事实。你需要设计至少四类能力存储、写入、召回和遗忘。存储层可以用SQLite、PostgreSQL存储结构化事实也可以用向量数据库存储语义向量。写入层负责从对话中抽取值得记住的信息这里的关键是“抽取”而不是“抄录”对话中的寒暄和临时信息不应该进入长期记忆。召回层根据当前用户问题从存储中筛选相关记忆最简单的方式是关键词匹配进阶方式是用Embedding做向量相似度检索。遗忘层则处理记忆的更新与过期防止记忆库无限膨胀。从实现顺序来看我建议你先用SQLite加简单关键词匹配跑通整个闭环再逐步替换成向量检索。这样每次只引入一个变量排查问题会容易得多。4. 基于LangGraph实现带长期记忆的Agent4.1 项目结构与依赖这一节的完整代码会实现一个带长期记忆的对话Agent。它做的事情是每次用户提问时先从SQLite中长期记忆表中检索与该用户相关的记忆拼接到System Prompt里然后调用LLM生成回答接着从最近一轮对话中抽取值得长期保存的事实如果没有新事实流程结束如果有新事实写入记忆库流程结束。4.2 长期记忆存储memory_store.py先实现最底层的存储模块。这里没有引入复杂的向量库先用SQLite实现一个可运行的长期记忆存储并支持按关键词检索。每个记忆记录都包含用户ID、内容、类型、重要度和更新时间。# 文件路径agent-memory-demo/memory_store.py import sqlite3 import uuid from datetime import datetime class MemoryStore: def __init__(self, db_pathagent_memory.db): self.conn sqlite3.connect(db_path, check_same_threadFalse) self._init_table() def _init_table(self): self.conn.execute( CREATE TABLE IF NOT EXISTS memories ( id TEXT PRIMARY KEY, uid TEXT NOT NULL, content TEXT NOT NULL, memory_type TEXT DEFAULT fact, importance REAL DEFAULT 0.5, created_at TEXT, updated_at TEXT ) ) self.conn.execute( CREATE INDEX IF NOT EXISTS idx_memories_uid ON memories(uid) ) self.conn.commit() def add_memory(self, uid, content, memory_typefact, importance0.5): now datetime.now().isoformat() row self.conn.execute( SELECT id FROM memories WHERE uid ? AND content ?, (uid, content), ).fetchone() if row: self.conn.execute( UPDATE memories SET updated_at ?, importance MAX(importance, ?) WHERE id ?, (now, importance, row[0]), ) self.conn.commit() return row[0] mid str(uuid.uuid4()) self.conn.execute( INSERT INTO memories (id, uid, content, memory_type, importance, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?), (mid, uid, content, memory_type, importance, now, now), ) self.conn.commit() return mid def search_memory(self, uid, keywordNone, top_k5): sql SELECT content, memory_type, importance, updated_at FROM memories WHERE uid ? params [uid] if keyword: sql AND content LIKE ? params.append(f%{keyword}%) sql ORDER BY importance DESC, updated_at DESC LIMIT ? params.append(top_k) rows self.conn.execute(sql, params).fetchall() return [ { content: row[0], memory_type: row[1], importance: row[2], updated_at: row[3], } for row in rows ] def get_all_memories(self, uid, limit50): rows self.conn.execute( SELECT content, memory_type, importance, updated_at FROM memories WHERE uid ? ORDER BY updated_at DESC LIMIT ?, (uid, limit), ).fetchall() return [ { content: row[0], memory_type: row[1], importance: row[2], updated_at: row[3], } for row in rows ] def delete_memories(self, uid, keep_num200): ids self.conn.execute( SELECT id FROM memories WHERE uid ? AND id NOT IN ( SELECT id FROM memories WHERE uid ? ORDER BY updated_at DESC LIMIT ? ), (uid, uid, keep_num), ).fetchall() for row in ids: self.conn.execute(DELETE FROM memories WHERE id ?, (row[0],)) self.conn.commit() return len(ids)这段代码的核心在于add_memory会先查重如果同一条事实已经存在就更新时间和重要度而不是重复插入search_memory支持按关键词过滤生产环境可以在这里替换成向量检索或混合检索。delete_memories方法用于清理过期记忆防止数据无限增长。4.3 定义状态state.py状态定义是整个LangGraph流程的“数据结构”。前面已经提到messages字段需要使用add合并器才能在多轮对话中不断累积历史。new_facts字段用来暂存本轮抽取出的新事实供后续节点决定是否写入记忆库。# 文件路径agent-memory-demo/state.py from typing import Annotated, TypedDict from operator import add from langchain_core.messages import BaseMessage class AgentState(TypedDict): messages: Annotated[list[BaseMessage], add] # 对话历史自动累积 memories: list[str] # 当前会话召回的记忆 new_facts: list[str] # 本轮抽取的新事实这里要提醒一下并不是所有字段都需要累积。memories和new_facts只是当次节点执行过程中的临时数据使用普通列表即可。如果给它们也加上add合并器反而可能让上一轮的记忆残留在新会话中造成状态污染。4.4 构建LangGraph记忆Agentgraph.py图的流程设计如下开始节点是recall_memory_node负责从长期记忆中检索当前用户的历史事实写入state[memories]然后进入agent_node把记忆拼接进System Prompt并调用LLM生成回答接着进入extract_memory_node从最近一轮对话中抽取需要长期保存的事实最后通过条件路由判断如果new_facts非空进入save_memory_node写入数据库否则直接结束。# 文件路径agent-memory-demo/graph.py import json import os import re from langchain_core.messages import HumanMessage, SystemMessage, AIMessage from langchain_openai import ChatOpenAI from langgraph.checkpoint.memory import MemorySaver from langgraph.graph import StateGraph, START, END from state import AgentState from memory_store import MemoryStore llm ChatOpenAI( modelos.getenv(LLM_MODEL, gpt-4o-mini), temperature0.2, api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL), ) store MemoryStore() SYSTEM_TEMPLATE 你是企业级智能助手。请结合用户长期记忆和当前对话历史回答。 长期记忆可能存在过期信息请以最新对话为准。 用户长期记忆 {memory_text} EXTRACT_TEMPLATE 请从以下对话中抽取需要长期记住的用户信息。 可以记住的信息包括用户姓名、偏好、禁忌、常用业务规则、项目结论、待办事项。 不要记住寒暄语、临时性问题。 只输出JSON数组格式如[事实1, 事实2] 如果没有值得记住的信息输出[] 对话内容 {conversation_text} def parse_json_array(text: str) - list[str]: try: data json.loads(text) if isinstance(data, list): return [str(item) for item in data] except Exception: pass match re.search(r\[.*\], text, re.S) if match: try: data json.loads(match.group()) if isinstance(data, list): return [str(item) for item in data] except Exception: pass return [] def recall_memory_node(state: AgentState, configNone): uid config[configurable][thread_id] memories store.search_memory(uid, top_k5) return {memories: [m[content] for m in memories]} def agent_node(state: AgentState, configNone): memory_text \n.join(state.get(memories, [])) or 暂无长期记忆 system SystemMessage(contentSYSTEM_TEMPLATE.format(memory_textmemory_text)) # 只取最近6条消息控制Context长度 history list(state[messages])[-6:] response llm.invoke([system] history) return {messages: [AIMessage(contentresponse.content)]} def extract_memory_node(state: AgentState): # 只看最近一条User消息和Agent回复避免重复抽取 recent_messages list(state[messages])[-2:] conversation_text \n.join( f{用户 if isinstance(m, HumanMessage) else 助手}: {m.content} for m in recent_messages ) prompt EXTRACT_TEMPLATE.format(conversation_textconversation_text) resp llm.invoke([HumanMessage(contentprompt)]) facts parse_json_array(resp.content) return {new_facts: facts} def save_memory_node(state: AgentState, configNone): uid config[configurable][thread_id] for fact in state.get(new_facts, []): store.add_memory(uiduid, contentfact, memory_typefact, importance0.5) return {new_facts: []} def should_save_memory(state: AgentState) - str: if state.get(new_facts): return save return end def build_graph(): workflow StateGraph(AgentState) workflow.add_node(recall, recall_memory_node) workflow.add_node(agent, agent_node) workflow.add_node(extract, extract_memory_node) workflow.add_node(save, save_memory_node) workflow.add_edge(START, recall) workflow.add_edge(recall, agent) workflow.add_edge(agent, extract) workflow.add_conditional_edges( extract, should_save_memory, { save: save, end: END, }, ) workflow.add_edge(save, END) checkpointer MemorySaver() app workflow.compile(checkpointercheckpointer) return app app build_graph()这段代码有几个关键点值得展开说明。第一recall_memory_node通过config[configurable][thread_id]获取当前用户ID这个值在调用时传入LangGraph会把运行时配置传给每个节点函数。thread_id既是短期记忆的会话ID也是长期记忆查询的用户维度。第二agent_node并没有把全部历史都塞进Prompt而是只取了最后6条消息。这样做是为了控制Token成本在实际项目中你可以根据业务复杂度调整这个数字或者接入更多召回策略。第三extract_memory_node在每次对话后都会调用一次LLM来抽取事实这会产生额外的模型调用成本。生产环境中可以设置一个抽样的概率阈值或者只在用户明确表达了偏好、禁忌、结论时触发抽取避免每一轮都做无意义的信息抽取。第四should_save_memory就是LangGraph条件路由的体现。它根据new_facts是否为空决定是进入保存节点还是直接结束。这是LangGraph中非常典型的“分支控制”写法。4.5 命令行入口main.py入口程序是一个简单的命令行循环。每次调用app.invoke时传入当前用户输入并保持同一个thread_id这样LangGraph会自动恢复历史消息实现短期记忆的连续性。# 文件路径agent-memory-demo/main.py from langchain_core.messages import HumanMessage from graph import app def main(): uid input(请输入用户ID用于模拟不同用户).strip() or user-001 config {configurable: {thread_id: uid}} print(Agent已启动。输入 exit 或 quit 退出。) while True: user_input input(你).strip() if not user_input: continue if user_input.lower() in {exit, quit}: break result app.invoke( {messages: [HumanMessage(contentuser_input)]}, configconfig, ) answer result[messages][-1].content print(Agent, answer) if __name__ __main__: main()运行方式python main.py启动后你可以连续输入多轮对话。比如第一轮说“我叫张三我喜欢简洁风格的报告”第二轮问“我叫什么名字、偏好什么风格”Agent需要能正确回答。第一轮结束时extract_memory_node会把“用户叫张三”和“喜欢简洁风格报告”写入SQLite第二轮开始时recall_memory_node会检索到这两条事实并注入System Prompt从而让Agent“想起”用户是谁。4.6 从SQLite关键词检索升级到向量检索当记忆条数变多、用户问题表达更加口语化时关键词匹配的召回效果会明显下降。比如用户问“帮我看看上次那个项目的结论”但长期记忆里存的是“XX项目已完成验收结论是供应商评分8.2分”关键词项目能命中但表达稍有变化就可能漏掉。向量检索可以解决这个问题。升级思路是在写入记忆时把每条记忆文本用Embedding模型转成向量存入向量数据库在召回时把用户当前问题也转成向量检索最相似的几条记忆。核心逻辑如下供你参考from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import Chroma embeddings OpenAIEmbeddings(modelos.getenv(EMBEDDING_MODEL, text-embedding-3-small)) vectorstore Chroma( collection_nameagent_memory, embedding_functionembeddings, persist_directory./memory_vectors, ) def add_memory_vector(uid, content): # 实际开发中建议把 uid 也写入 metadata并按 uid 过滤 vectorstore.add_texts( texts[content], metadatas[{uid: uid}], ) def search_memory_vector(uid, query, top_k5): docs vectorstore.similarity_search_with_score( query, ktop_k, filter{uid: uid}, ) return [doc.page_content for doc, score in docs]在实际生产系统中我更推荐混合检索先用向量召回Top30再用关键词或规则过滤最后重排取Top5。只靠向量检索有一个风险是模型对代词和模糊表述的判断不够稳定加入关键词兜底会更可靠。5. DeepAgent视角生产治理与安全边界5.1 记忆治理不是“多存点”很多团队做长期记忆最直接的想法就是把对话记录全部保存下来需要时再查。这种做法在技术上有实现成本但在治理上非常危险。长期记忆存储的不是原始日志而是Agent对用户的“认知模型”。如果不对写入内容做筛选错误的、临时的、敏感的信息都会进入长期记忆后续会被错误地召回造成更难排查的错误输出。从DeepAgent这类企业级框架的治理思路来看记忆应该被当作一等公民来管理。它和代码配置、数据库权限一样需要有明确的负责人、生命周期和审计日志。哪些记忆可以写入取决于置信度阈值哪些记忆可以被召回取决于权限范围哪些记忆必须被删除取决于用户隐私合规要求。这些问题越早设计后续返工成本越低。5.2 敏感记忆的脱敏与权限隔离企业级Agent的长期记忆里不可避免地会出现敏感信息比如用户手机号、部门预算、内部项目代号。对这些信息至少要区分三个级别可明文存储、可加密存储、禁止存储。手机号、身份证号这类高敏信息不应该进入普通记忆表即使需要存储也应该先脱敏后入库并在召回时进行掩码展示。另一个常见问题是多租户下的权限隔离。同一个Agent系统可能服务多个部门或多个客户如果记忆表只按uid区分而uid本身可被伪造就会发生数据越权。生产环境至少需要在记忆表中增加tenant_id字段并在所有查询语句中强制带上租户条件同时在应用层做权限校验而不是只依赖SQL过滤。5.3 记忆动作的可观测性与审计当Agent因为一条错误记忆给出了错误答案时最让人头疼的问题是“不知道这条记忆是哪里来的”。因此长期记忆系统必须能够回答三个问题某条记忆是什么时候写入的写入时对应的原始对话是哪一轮这条记忆在回答哪些问题时被使用过。从这个角度看日志设计比想象中重要。建议在每次记忆写入时记录source_message_id在每次记忆召回时把召回结果打印到业务日志或追踪系统中。这样当线上出现badcase时你能够反推是召回阶段出了问题还是记忆写入阶段产生了错误事实。DeepAgent这类框架强调的可观测性本质就是要让Agent的内部决策过程可回溯。5.4 长期记忆的数据生命周期记忆不是永久有效的。用户的偏好会改变项目状态会变化过期的记忆如果不清理反而会成为噪声。在工程上每个记忆记录都应该有版本、时间戳和状态新增、更新、过期、删除。更新不是简单覆盖而是生成新版本并保留历史版本方便回溯过期则可以通过定期任务扫描实现比如超过180天未更新的偏好可以降权当用户明确表示“我之前说的那个偏好不要了”应该提供删除接口直接从活跃记忆库中移除而不是仅靠重要度降权。6. 常见问题与排查清单下面是社区中LangChain、LangGraph实现记忆系统时最常遇到的一批问题整理成表格供排查参考。问题现象常见原因解决思路LangGraph节点中修改了State但下一个节点读到的还是旧值字段没有定义合适的合并策略节点返回的key拼写错误检查Annotated类型与合并函数统一字段命名多用户之间对话历史串线所有请求共用了同一个thread_id每个用户请求使用独立thread_id生产环境使用UUID长期记忆召回不到内容关键词不匹配记忆写入失败用户ID过滤条件错误检查SQLite中是否有数据换用向量检索打印召回日志模型抽取记忆时输出不是合法JSON模型输出格式不稳定在Prompt中给出示例用正则兜底解析失败时跳过本轮写入Agent执行提供方未在预期时间内返回结果网络超时、LLM服务负载高、单次请求时间过长查看服务端状态增大超时时间增加重试与熔断机制Context中历史消息越来越多成本上升没有做历史窗口截断在agent_node中只取最近N条消息对超长历史做摘要内存级Checkpointer导致重启后短期记忆丢失使用了MemorySaver替换为SqliteSaver或PostgresSaver排查长期记忆问题时有一个很简单但有效的顺序先确认写入成功再确认检索命中最后确认Prompt拼接。按照这个顺序基本能定位80%的“Agent失忆”问题。7. 最佳实践与工程建议7.1 记忆写入要有门槛不要让Agent每次都尝试抽取记忆更不要让模型自由发挥“觉得重要就保存”。建议在抽取节点后增加一个评分环节用规则或模型对候选事实给出置信度和重要度分数。只有置信度高于0.7、重要度高于0.3的记忆才允许写入长期记忆库。这个门槛能在早期就挡住大量噪声。7.2 召回策略要可控召回不是“越多越好”。给LLM的记忆过多可能导致模型被无关信息干扰。一个比较稳妥的做法是先向量召回Top20再用规则过滤最后按重要度排序取Top5。同时把召回结果按照“用户偏好、历史结论、业务规则”分类拼接到Prompt中而不是把所有记忆堆在一起。7.3 成本和延迟控制每次调用extract_memory_node都会增加一次LLM调用成本不可忽视。可以在三个维度控制第一只在用户消息长度大于某个阈值时触发抽取第二在消息流式处理中异步执行抽取不阻塞主回答链路第三对抽取结果做聚合比如5分钟内重复抽取到同一事实时只保留一个版本。7.4 效果评估不能只看“答对了没”Agent记忆系统的效果评估需要单独设计指标。建议至少跟踪以下几项记忆写入准确率抽样检查已保存的记忆是否真实、有用记忆召回命中率看用户问题与检索结果的相关性长期任务完成率比如在一周后的会话中用户是否还需要重复提供之前说过的信息。单独跑几个对话Demo看不出问题只有持续观察线上数据记忆系统才会逐步变好。7.5 配置、安全与生产发布记忆表属于业务数据任何修改都要遵循最小权限原则。开发和测试环境不要直接连接生产数据库执行批量删除或更新前先备份数据所有涉及用户敏感信息的操作都要记录操作者、时间和变更内容。生产发布顺序建议是先上线记忆写入与存储观察数据质量再开放召回能力观察回答效果最后再接入自动遗忘与清理任务。8. 总结与学习路线本文从Agent“失忆”的痛点出发梳理了记忆系统的四层模型Context、短期记忆、工作记忆和长期记忆并对比了LangChain、LangGraph、DeepAgent三者各自的定位。在实战部分我们用LangGraph搭建了一个完整的记忆Agent包括SQLite长期记忆存储、状态定义、记忆召回、LLM生成、事实抽取和条件路由保存代码可以直接复制到本地运行。下一步你可以从三个方向继续深入一是把SQLite检索替换成真正的向量检索并对比召回效果二是研究LangGraph的Checkpointer机制把短期记忆从内存迁移到数据库三是参考DeepAgent这类框架的治理思路完善记忆的权限隔离、审计日志和遗忘策略。做记忆系统的过程中我最大的感受是不要急着把“更多历史”塞给模型先把“什么值得记住”和“如何安全地叫回一段记忆”这两个问题想清楚。希望这篇笔记对你正在做的Agent项目有帮助。 企业级Agent记忆系统设计：从LangGraph实现到生产治理
企业级Agent落地时最容易被低估的技术点不是模型选型而是记忆系统。很多团队在Demo阶段觉得Agent的效果还不错一旦进入生产环境就会发现几个老问题反复出现用户上一轮说过的偏好这一轮就忘同一个用户在多轮任务中提到的关键信息散落在不同会话里无法聚合Agent只能依赖Prompt中的Context上下文一长成本和延迟又快速上升。归根结底是因为模型本身没有记忆机制每次调用都是无状态的所谓的“记住”要么靠上下文拼接要么靠外部存储背后需要一套完整的工程架构去支撑。本文会围绕Context、短期记忆、工作记忆和长期记忆四个层次从原理到LangChain、LangGraph实现再到DeepAgent框架带来的生产治理思路完整梳理企业级Agent记忆系统的设计方法。适合正在做Agent应用开发、想解决“Agent没记性”问题的后端开发者也适合刚入门LangGraph、想理解记忆模块怎么设计的读者。文章会先讲清楚记忆分层的背景和三个框架的定位然后给出一个可以用LangGraph跑通的长期记忆Demo包括SQLite持久化、短期会话记忆、条件路由和记忆写入节点最后补上生产环境必须考虑的安全、审计、召回策略和效果评估。你可以把本文当成一份“从原理到落地”的工程笔记来读。1. 背景与核心概念1.1 为什么Agent会“失忆”大语言模型本身是一个无状态函数你输入一段文本它输出一段文本调用结束后模型内部不会保留任何关于用户的信息。为了让Agent表现得“有记忆”常见做法是把历史对话、用户信息、业务规则全部塞进Prompt的Context窗口里。这种方式在小规模场景下可行但它本质上是在“临时搬运”信息而不是在“沉淀”信息。当应用进入企业级场景后问题会变得非常具体。用户可能在一个星期内多次访问同一个Agent每次访问都带着新的任务但任务之间相互关联用户可能在某个会话里提到“我们公司的审批流程是三级审批”另一个会话里又提到“这个月预算已经用完”Agent需要把这些分散的事实整合成对用户的稳定理解。这些需求已经超出了单纯Context拼接的范畴需要引入一套记忆系统把短期会话信息、任务中间状态和长期用户画像分开管理。从架构上看Agent的记忆系统至少需要回答四个问题记什么、存在哪、怎么取、什么时候忘。记什么决定了我们抽取的是用户偏好、关键事实还是历史结论存在哪决定了用数据库、向量库还是图谱怎么取决定了是关键词检索、向量相似度还是规则命中什么时候忘则决定了记忆的更新、过期和删除策略。这四个问题没有统一答案但有一个稳定的分层模型可以参考。1.2 记忆分层从Context到Long-term Me在工程上我习惯把Agent的记忆分成四层这四层常常被混为一谈但它们的存储方式、更新频率和生命周期完全不同。第一层是模型Context也就是当前Prompt窗口内的内容。它最直接但也最不可靠。一旦超出Token上限最老的信息会被截断即使没有超限把所有历史都堆进Context也会让模型注意力分散、推理变慢。第二层是短期记忆也叫会话记忆通常指一次会话内的多轮对话历史可以通过LangGraph的Checkpointer机制保存。第三层是工作记忆指当前任务执行过程中的临时状态比如一个多步骤任务已经走到哪一步、临时变量是什么这在LangGraph里对应State的设计。第四层是长期记忆它需要跨会话、跨任务保存内容包括用户画像、偏好、历史结论、领域知识是Agent在长期交互中形成的对用户的稳定认知模型也就是标题里说的Long-term Me。这四层之间并不是互斥的而是一条数据流水线。短期记忆和工作记忆负责当次会话与当次任务长期记忆负责把有价值的沉淀抽取出来并在后续会话中通过检索重新注入Context。理解了这个流水线再看LangChain、LangGraph和DeepAgent的定位就会清晰很多。1.3 LangChain、LangGraph与DeepAgent的定位很多开发者刚接触这三个名词时会困惑它们到底是不是同一个东西其实不是。LangChain是一个组件生态提供了模型封装、Prompt模板、工具调用、向量库集成等能力解决的是“和模型对话、调工具”的便利性问题。LangGraph是一个基于状态图的Agent编排框架强调的是节点、边、条件路由、循环和检查点解决的是“复杂Agent流程怎么控制”的问题。DeepAgent则更像一类面向企业级Agent生产化的框架或工程理念它不只是帮你写工作流更关注长期身份、可观测性、安全边界、记忆生命周期这些治理问题。用一个不太严谨的比喻来说LangChain像工具箱LangGraph像流水线控制系统DeepAgent像工厂的管理制度。工具决定了你能干什么流程控制决定了任务怎么走管理制度决定了长期稳定运行需要哪些规范和边界。本文的实战部分主要用LangGraph来实现记忆闭环而DeepAgent的思路会贯穿在第5章的生产治理建议中。维度LangChainLangGraphDeepAgent核心定位组件库与工具链Agent状态图编排企业级Agent生产治理框架关注重点模型调用、Prompt、工具、向量库节点、边、条件路由、循环、检查点长期记忆、身份、安全、审计、生命周期使用场景快速实现LLM应用复杂多步Agent流程生产级Agent平台落地记忆视角提供历史消息封装用Checkpointer和State实现记忆机制治理长期记忆的沉淀、权限与遗忘三者可以配合使用并不冲突。2. 环境准备与版本说明在开始写代码之前先明确环境和项目结构。本文示例以Python 3.10为基础核心依赖是LangChain、LangGraph和一个支持OpenAI接口的大模型服务。由于LangChain和LangGraph迭代速度很快具体的API名称在不同版本中可能有细微差异建议你在实际项目中锁定已验证的版本或者在虚拟环境里先安装最新版本再按官方文档微调。本文演示的重点是记忆系统的工程思路而不是某个特定版本的新特性所以下面的代码你在2024年之后的主流版本中基本都能运行。如果你使用的是更新的大版本遇到个别接口调整优先查看官方迁移文档。python -m venv .venv source .venv/bin/activate # Windows 下使用 .venv\Scripts\activate pip install langchain langchain-openai langgraph项目结构如下。这个结构不复杂但能清晰区分记忆存储、状态定义、图编排和入口调用。agent-memory-demo/ ├── memory_store.py # 长期记忆存储基于 SQLite ├── state.py # 定义 AgentState ├── graph.py # 构建 LangGraph 图 └── main.py # 命令行交互入口LLM服务建议使用OpenAI兼容接口也可以使用国内主流模型的兼容网关只需要设置OPENAI_API_KEY和可选的OPENAI_BASE_URL。模型名称通过环境变量LLM_MODEL控制方便切换。3. 记忆系统的核心工程拆解3.1 Context管理与Token预算Context是模型一次调用能看到的所有内容。很多初学者以为“给模型更多历史效果一定更好”实际上历史消息越多Token成本越高模型对关键信息的敏感度也会下降。企业级Agent必须对Context做预算管理不是把所有历史都塞进去而是选择当前任务最相关的内容。一种常见的做法是把记忆分为“必带信息”和“按需检索信息”。必带信息包括系统提示词、当前用户请求、最近一两轮的对话按需检索信息则从长期记忆库中查找比如用户偏好、历史结论、领域知识只在需要时注入。这样既保证了Agent对当前任务的理解又控制了成本。LangGraph里通过节点的组合可以很容易实现这种设计这一点会在第4章的代码中体现。3.2 短期记忆会话历史与Checkpointer短期记忆的核心是保存多轮对话历史。LangGraph提供了Checkpointer机制会把图执行到某个节点时的State保存下来下次以相同thread_id继续调用时可以自动恢复历史状态。这里说的State就相当于Agent的工作记忆而Checkpointer则实现了会话级记忆的持久化。官方提供的MemorySaver适合本地演示但它是内存级的服务重启后数据会丢失。生产环境建议使用SqliteSaver、PostgresSaver等持久化Checkpointer把会话状态保存到数据库。也就是说短期记忆和工作记忆本质上都是“状态的持久化”只是保存的粒度不同。3.3 工作记忆LangGraph中的StateLangGraph的State是节点之间传递数据的唯一方式。你在StateGraph(AgentState)中定义的TypedDict字段就是所有节点共享的工作记忆。节点函数可以读取state也可以返回一个字典来更新state中的字段。对于需要累积的字段比如对话历史messages需要使用Annotated类型和合并函数否则每次返回都会覆盖旧值。这一点非常容易踩坑。举个例子如果messages字段不加合并逻辑Agent在第二轮回合中可能只看到当前轮消息而丢失了第一轮的上下文。from typing import Annotated, TypedDict from operator import add from langchain_core.messages import BaseMessage class AgentState(TypedDict): messages: Annotated[list[BaseMessage], add] # 对话历史自动累积 memories: list[str] # 当前会话召回的长时记忆 new_facts: list[str] # 本轮需要保存的新事实3.4 长期记忆存储、写入、召回与遗忘长期记忆和短期记忆的最大区别在于生命周期。长期记忆需要跨会话保存并且通常不是原始对话文本而是从对话中抽取出来的结构化事实。你需要设计至少四类能力存储、写入、召回和遗忘。存储层可以用SQLite、PostgreSQL存储结构化事实也可以用向量数据库存储语义向量。写入层负责从对话中抽取值得记住的信息这里的关键是“抽取”而不是“抄录”对话中的寒暄和临时信息不应该进入长期记忆。召回层根据当前用户问题从存储中筛选相关记忆最简单的方式是关键词匹配进阶方式是用Embedding做向量相似度检索。遗忘层则处理记忆的更新与过期防止记忆库无限膨胀。从实现顺序来看我建议你先用SQLite加简单关键词匹配跑通整个闭环再逐步替换成向量检索。这样每次只引入一个变量排查问题会容易得多。4. 基于LangGraph实现带长期记忆的Agent4.1 项目结构与依赖这一节的完整代码会实现一个带长期记忆的对话Agent。它做的事情是每次用户提问时先从SQLite中长期记忆表中检索与该用户相关的记忆拼接到System Prompt里然后调用LLM生成回答接着从最近一轮对话中抽取值得长期保存的事实如果没有新事实流程结束如果有新事实写入记忆库流程结束。4.2 长期记忆存储memory_store.py先实现最底层的存储模块。这里没有引入复杂的向量库先用SQLite实现一个可运行的长期记忆存储并支持按关键词检索。每个记忆记录都包含用户ID、内容、类型、重要度和更新时间。# 文件路径agent-memory-demo/memory_store.py import sqlite3 import uuid from datetime import datetime class MemoryStore: def __init__(self, db_pathagent_memory.db): self.conn sqlite3.connect(db_path, check_same_threadFalse) self._init_table() def _init_table(self): self.conn.execute( CREATE TABLE IF NOT EXISTS memories ( id TEXT PRIMARY KEY, uid TEXT NOT NULL, content TEXT NOT NULL, memory_type TEXT DEFAULT fact, importance REAL DEFAULT 0.5, created_at TEXT, updated_at TEXT ) ) self.conn.execute( CREATE INDEX IF NOT EXISTS idx_memories_uid ON memories(uid) ) self.conn.commit() def add_memory(self, uid, content, memory_typefact, importance0.5): now datetime.now().isoformat() row self.conn.execute( SELECT id FROM memories WHERE uid ? AND content ?, (uid, content), ).fetchone() if row: self.conn.execute( UPDATE memories SET updated_at ?, importance MAX(importance, ?) WHERE id ?, (now, importance, row[0]), ) self.conn.commit() return row[0] mid str(uuid.uuid4()) self.conn.execute( INSERT INTO memories (id, uid, content, memory_type, importance, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?), (mid, uid, content, memory_type, importance, now, now), ) self.conn.commit() return mid def search_memory(self, uid, keywordNone, top_k5): sql SELECT content, memory_type, importance, updated_at FROM memories WHERE uid ? params [uid] if keyword: sql AND content LIKE ? params.append(f%{keyword}%) sql ORDER BY importance DESC, updated_at DESC LIMIT ? params.append(top_k) rows self.conn.execute(sql, params).fetchall() return [ { content: row[0], memory_type: row[1], importance: row[2], updated_at: row[3], } for row in rows ] def get_all_memories(self, uid, limit50): rows self.conn.execute( SELECT content, memory_type, importance, updated_at FROM memories WHERE uid ? ORDER BY updated_at DESC LIMIT ?, (uid, limit), ).fetchall() return [ { content: row[0], memory_type: row[1], importance: row[2], updated_at: row[3], } for row in rows ] def delete_memories(self, uid, keep_num200): ids self.conn.execute( SELECT id FROM memories WHERE uid ? AND id NOT IN ( SELECT id FROM memories WHERE uid ? ORDER BY updated_at DESC LIMIT ? ), (uid, uid, keep_num), ).fetchall() for row in ids: self.conn.execute(DELETE FROM memories WHERE id ?, (row[0],)) self.conn.commit() return len(ids)这段代码的核心在于add_memory会先查重如果同一条事实已经存在就更新时间和重要度而不是重复插入search_memory支持按关键词过滤生产环境可以在这里替换成向量检索或混合检索。delete_memories方法用于清理过期记忆防止数据无限增长。4.3 定义状态state.py状态定义是整个LangGraph流程的“数据结构”。前面已经提到messages字段需要使用add合并器才能在多轮对话中不断累积历史。new_facts字段用来暂存本轮抽取出的新事实供后续节点决定是否写入记忆库。# 文件路径agent-memory-demo/state.py from typing import Annotated, TypedDict from operator import add from langchain_core.messages import BaseMessage class AgentState(TypedDict): messages: Annotated[list[BaseMessage], add] # 对话历史自动累积 memories: list[str] # 当前会话召回的记忆 new_facts: list[str] # 本轮抽取的新事实这里要提醒一下并不是所有字段都需要累积。memories和new_facts只是当次节点执行过程中的临时数据使用普通列表即可。如果给它们也加上add合并器反而可能让上一轮的记忆残留在新会话中造成状态污染。4.4 构建LangGraph记忆Agentgraph.py图的流程设计如下开始节点是recall_memory_node负责从长期记忆中检索当前用户的历史事实写入state[memories]然后进入agent_node把记忆拼接进System Prompt并调用LLM生成回答接着进入extract_memory_node从最近一轮对话中抽取需要长期保存的事实最后通过条件路由判断如果new_facts非空进入save_memory_node写入数据库否则直接结束。# 文件路径agent-memory-demo/graph.py import json import os import re from langchain_core.messages import HumanMessage, SystemMessage, AIMessage from langchain_openai import ChatOpenAI from langgraph.checkpoint.memory import MemorySaver from langgraph.graph import StateGraph, START, END from state import AgentState from memory_store import MemoryStore llm ChatOpenAI( modelos.getenv(LLM_MODEL, gpt-4o-mini), temperature0.2, api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL), ) store MemoryStore() SYSTEM_TEMPLATE 你是企业级智能助手。请结合用户长期记忆和当前对话历史回答。 长期记忆可能存在过期信息请以最新对话为准。 用户长期记忆 {memory_text} EXTRACT_TEMPLATE 请从以下对话中抽取需要长期记住的用户信息。 可以记住的信息包括用户姓名、偏好、禁忌、常用业务规则、项目结论、待办事项。 不要记住寒暄语、临时性问题。 只输出JSON数组格式如[事实1, 事实2] 如果没有值得记住的信息输出[] 对话内容 {conversation_text} def parse_json_array(text: str) - list[str]: try: data json.loads(text) if isinstance(data, list): return [str(item) for item in data] except Exception: pass match re.search(r\[.*\], text, re.S) if match: try: data json.loads(match.group()) if isinstance(data, list): return [str(item) for item in data] except Exception: pass return [] def recall_memory_node(state: AgentState, configNone): uid config[configurable][thread_id] memories store.search_memory(uid, top_k5) return {memories: [m[content] for m in memories]} def agent_node(state: AgentState, configNone): memory_text \n.join(state.get(memories, [])) or 暂无长期记忆 system SystemMessage(contentSYSTEM_TEMPLATE.format(memory_textmemory_text)) # 只取最近6条消息控制Context长度 history list(state[messages])[-6:] response llm.invoke([system] history) return {messages: [AIMessage(contentresponse.content)]} def extract_memory_node(state: AgentState): # 只看最近一条User消息和Agent回复避免重复抽取 recent_messages list(state[messages])[-2:] conversation_text \n.join( f{用户 if isinstance(m, HumanMessage) else 助手}: {m.content} for m in recent_messages ) prompt EXTRACT_TEMPLATE.format(conversation_textconversation_text) resp llm.invoke([HumanMessage(contentprompt)]) facts parse_json_array(resp.content) return {new_facts: facts} def save_memory_node(state: AgentState, configNone): uid config[configurable][thread_id] for fact in state.get(new_facts, []): store.add_memory(uiduid, contentfact, memory_typefact, importance0.5) return {new_facts: []} def should_save_memory(state: AgentState) - str: if state.get(new_facts): return save return end def build_graph(): workflow StateGraph(AgentState) workflow.add_node(recall, recall_memory_node) workflow.add_node(agent, agent_node) workflow.add_node(extract, extract_memory_node) workflow.add_node(save, save_memory_node) workflow.add_edge(START, recall) workflow.add_edge(recall, agent) workflow.add_edge(agent, extract) workflow.add_conditional_edges( extract, should_save_memory, { save: save, end: END, }, ) workflow.add_edge(save, END) checkpointer MemorySaver() app workflow.compile(checkpointercheckpointer) return app app build_graph()这段代码有几个关键点值得展开说明。第一recall_memory_node通过config[configurable][thread_id]获取当前用户ID这个值在调用时传入LangGraph会把运行时配置传给每个节点函数。thread_id既是短期记忆的会话ID也是长期记忆查询的用户维度。第二agent_node并没有把全部历史都塞进Prompt而是只取了最后6条消息。这样做是为了控制Token成本在实际项目中你可以根据业务复杂度调整这个数字或者接入更多召回策略。第三extract_memory_node在每次对话后都会调用一次LLM来抽取事实这会产生额外的模型调用成本。生产环境中可以设置一个抽样的概率阈值或者只在用户明确表达了偏好、禁忌、结论时触发抽取避免每一轮都做无意义的信息抽取。第四should_save_memory就是LangGraph条件路由的体现。它根据new_facts是否为空决定是进入保存节点还是直接结束。这是LangGraph中非常典型的“分支控制”写法。4.5 命令行入口main.py入口程序是一个简单的命令行循环。每次调用app.invoke时传入当前用户输入并保持同一个thread_id这样LangGraph会自动恢复历史消息实现短期记忆的连续性。# 文件路径agent-memory-demo/main.py from langchain_core.messages import HumanMessage from graph import app def main(): uid input(请输入用户ID用于模拟不同用户).strip() or user-001 config {configurable: {thread_id: uid}} print(Agent已启动。输入 exit 或 quit 退出。) while True: user_input input(你).strip() if not user_input: continue if user_input.lower() in {exit, quit}: break result app.invoke( {messages: [HumanMessage(contentuser_input)]}, configconfig, ) answer result[messages][-1].content print(Agent, answer) if __name__ __main__: main()运行方式python main.py启动后你可以连续输入多轮对话。比如第一轮说“我叫张三我喜欢简洁风格的报告”第二轮问“我叫什么名字、偏好什么风格”Agent需要能正确回答。第一轮结束时extract_memory_node会把“用户叫张三”和“喜欢简洁风格报告”写入SQLite第二轮开始时recall_memory_node会检索到这两条事实并注入System Prompt从而让Agent“想起”用户是谁。4.6 从SQLite关键词检索升级到向量检索当记忆条数变多、用户问题表达更加口语化时关键词匹配的召回效果会明显下降。比如用户问“帮我看看上次那个项目的结论”但长期记忆里存的是“XX项目已完成验收结论是供应商评分8.2分”关键词项目能命中但表达稍有变化就可能漏掉。向量检索可以解决这个问题。升级思路是在写入记忆时把每条记忆文本用Embedding模型转成向量存入向量数据库在召回时把用户当前问题也转成向量检索最相似的几条记忆。核心逻辑如下供你参考from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import Chroma embeddings OpenAIEmbeddings(modelos.getenv(EMBEDDING_MODEL, text-embedding-3-small)) vectorstore Chroma( collection_nameagent_memory, embedding_functionembeddings, persist_directory./memory_vectors, ) def add_memory_vector(uid, content): # 实际开发中建议把 uid 也写入 metadata并按 uid 过滤 vectorstore.add_texts( texts[content], metadatas[{uid: uid}], ) def search_memory_vector(uid, query, top_k5): docs vectorstore.similarity_search_with_score( query, ktop_k, filter{uid: uid}, ) return [doc.page_content for doc, score in docs]在实际生产系统中我更推荐混合检索先用向量召回Top30再用关键词或规则过滤最后重排取Top5。只靠向量检索有一个风险是模型对代词和模糊表述的判断不够稳定加入关键词兜底会更可靠。5. DeepAgent视角生产治理与安全边界5.1 记忆治理不是“多存点”很多团队做长期记忆最直接的想法就是把对话记录全部保存下来需要时再查。这种做法在技术上有实现成本但在治理上非常危险。长期记忆存储的不是原始日志而是Agent对用户的“认知模型”。如果不对写入内容做筛选错误的、临时的、敏感的信息都会进入长期记忆后续会被错误地召回造成更难排查的错误输出。从DeepAgent这类企业级框架的治理思路来看记忆应该被当作一等公民来管理。它和代码配置、数据库权限一样需要有明确的负责人、生命周期和审计日志。哪些记忆可以写入取决于置信度阈值哪些记忆可以被召回取决于权限范围哪些记忆必须被删除取决于用户隐私合规要求。这些问题越早设计后续返工成本越低。5.2 敏感记忆的脱敏与权限隔离企业级Agent的长期记忆里不可避免地会出现敏感信息比如用户手机号、部门预算、内部项目代号。对这些信息至少要区分三个级别可明文存储、可加密存储、禁止存储。手机号、身份证号这类高敏信息不应该进入普通记忆表即使需要存储也应该先脱敏后入库并在召回时进行掩码展示。另一个常见问题是多租户下的权限隔离。同一个Agent系统可能服务多个部门或多个客户如果记忆表只按uid区分而uid本身可被伪造就会发生数据越权。生产环境至少需要在记忆表中增加tenant_id字段并在所有查询语句中强制带上租户条件同时在应用层做权限校验而不是只依赖SQL过滤。5.3 记忆动作的可观测性与审计当Agent因为一条错误记忆给出了错误答案时最让人头疼的问题是“不知道这条记忆是哪里来的”。因此长期记忆系统必须能够回答三个问题某条记忆是什么时候写入的写入时对应的原始对话是哪一轮这条记忆在回答哪些问题时被使用过。从这个角度看日志设计比想象中重要。建议在每次记忆写入时记录source_message_id在每次记忆召回时把召回结果打印到业务日志或追踪系统中。这样当线上出现badcase时你能够反推是召回阶段出了问题还是记忆写入阶段产生了错误事实。DeepAgent这类框架强调的可观测性本质就是要让Agent的内部决策过程可回溯。5.4 长期记忆的数据生命周期记忆不是永久有效的。用户的偏好会改变项目状态会变化过期的记忆如果不清理反而会成为噪声。在工程上每个记忆记录都应该有版本、时间戳和状态新增、更新、过期、删除。更新不是简单覆盖而是生成新版本并保留历史版本方便回溯过期则可以通过定期任务扫描实现比如超过180天未更新的偏好可以降权当用户明确表示“我之前说的那个偏好不要了”应该提供删除接口直接从活跃记忆库中移除而不是仅靠重要度降权。6. 常见问题与排查清单下面是社区中LangChain、LangGraph实现记忆系统时最常遇到的一批问题整理成表格供排查参考。问题现象常见原因解决思路LangGraph节点中修改了State但下一个节点读到的还是旧值字段没有定义合适的合并策略节点返回的key拼写错误检查Annotated类型与合并函数统一字段命名多用户之间对话历史串线所有请求共用了同一个thread_id每个用户请求使用独立thread_id生产环境使用UUID长期记忆召回不到内容关键词不匹配记忆写入失败用户ID过滤条件错误检查SQLite中是否有数据换用向量检索打印召回日志模型抽取记忆时输出不是合法JSON模型输出格式不稳定在Prompt中给出示例用正则兜底解析失败时跳过本轮写入Agent执行提供方未在预期时间内返回结果网络超时、LLM服务负载高、单次请求时间过长查看服务端状态增大超时时间增加重试与熔断机制Context中历史消息越来越多成本上升没有做历史窗口截断在agent_node中只取最近N条消息对超长历史做摘要内存级Checkpointer导致重启后短期记忆丢失使用了MemorySaver替换为SqliteSaver或PostgresSaver排查长期记忆问题时有一个很简单但有效的顺序先确认写入成功再确认检索命中最后确认Prompt拼接。按照这个顺序基本能定位80%的“Agent失忆”问题。7. 最佳实践与工程建议7.1 记忆写入要有门槛不要让Agent每次都尝试抽取记忆更不要让模型自由发挥“觉得重要就保存”。建议在抽取节点后增加一个评分环节用规则或模型对候选事实给出置信度和重要度分数。只有置信度高于0.7、重要度高于0.3的记忆才允许写入长期记忆库。这个门槛能在早期就挡住大量噪声。7.2 召回策略要可控召回不是“越多越好”。给LLM的记忆过多可能导致模型被无关信息干扰。一个比较稳妥的做法是先向量召回Top20再用规则过滤最后按重要度排序取Top5。同时把召回结果按照“用户偏好、历史结论、业务规则”分类拼接到Prompt中而不是把所有记忆堆在一起。7.3 成本和延迟控制每次调用extract_memory_node都会增加一次LLM调用成本不可忽视。可以在三个维度控制第一只在用户消息长度大于某个阈值时触发抽取第二在消息流式处理中异步执行抽取不阻塞主回答链路第三对抽取结果做聚合比如5分钟内重复抽取到同一事实时只保留一个版本。7.4 效果评估不能只看“答对了没”Agent记忆系统的效果评估需要单独设计指标。建议至少跟踪以下几项记忆写入准确率抽样检查已保存的记忆是否真实、有用记忆召回命中率看用户问题与检索结果的相关性长期任务完成率比如在一周后的会话中用户是否还需要重复提供之前说过的信息。单独跑几个对话Demo看不出问题只有持续观察线上数据记忆系统才会逐步变好。7.5 配置、安全与生产发布记忆表属于业务数据任何修改都要遵循最小权限原则。开发和测试环境不要直接连接生产数据库执行批量删除或更新前先备份数据所有涉及用户敏感信息的操作都要记录操作者、时间和变更内容。生产发布顺序建议是先上线记忆写入与存储观察数据质量再开放召回能力观察回答效果最后再接入自动遗忘与清理任务。8. 总结与学习路线本文从Agent“失忆”的痛点出发梳理了记忆系统的四层模型Context、短期记忆、工作记忆和长期记忆并对比了LangChain、LangGraph、DeepAgent三者各自的定位。在实战部分我们用LangGraph搭建了一个完整的记忆Agent包括SQLite长期记忆存储、状态定义、记忆召回、LLM生成、事实抽取和条件路由保存代码可以直接复制到本地运行。下一步你可以从三个方向继续深入一是把SQLite检索替换成真正的向量检索并对比召回效果二是研究LangGraph的Checkpointer机制把短期记忆从内存迁移到数据库三是参考DeepAgent这类框架的治理思路完善记忆的权限隔离、审计日志和遗忘策略。做记忆系统的过程中我最大的感受是不要急着把“更多历史”塞给模型先把“什么值得记住”和“如何安全地叫回一段记忆”这两个问题想清楚。希望这篇笔记对你正在做的Agent项目有帮助。