企业级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章的生产治理建议中。
| 维度 | LangChain | LangGraph | DeepAgent |
|---|---|---|---|
| 核心定位 | 组件库与工具链 | 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中的State
LangGraph的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实现带长期记忆的Agent
4.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_path="agent_memory.db"): self.conn = sqlite3.connect(db_path, check_same_thread=False) 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_type="fact", importance=0.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, keyword=None, top_k=5): 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, limit=50): 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_num=200): 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记忆Agent:graph.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( model=os.getenv("LLM_MODEL", "gpt-4o-mini"), temperature=0.2, api_key=os.getenv("OPENAI_API_KEY"), base_url=os.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, config=None): uid = config["configurable"]["thread_id"] memories = store.search_memory(uid, top_k=5) return {"memories": [m["content"] for m in memories]} def agent_node(state: AgentState, config=None): memory_text = "\n".join(state.get("memories", [])) or "暂无长期记忆" system = SystemMessage(content=SYSTEM_TEMPLATE.format(memory_text=memory_text)) # 只取最近6条消息,控制Context长度 history = list(state["messages"])[-6:] response = llm.invoke([system] + history) return {"messages": [AIMessage(content=response.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_text=conversation_text) resp = llm.invoke([HumanMessage(content=prompt)]) facts = parse_json_array(resp.content) return {"new_facts": facts} def save_memory_node(state: AgentState, config=None): uid = config["configurable"]["thread_id"] for fact in state.get("new_facts", []): store.add_memory(uid=uid, content=fact, memory_type="fact", importance=0.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(checkpointer=checkpointer) 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(content=user_input)]}, config=config, ) 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(model=os.getenv("EMBEDDING_MODEL", "text-embedding-3-small")) vectorstore = Chroma( collection_name="agent_memory", embedding_function=embeddings, 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_k=5): docs = vectorstore.similarity_search_with_score( query, k=top_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项目有帮助。