ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

企业级Agent记忆系统设计:从Context管理到LangGraph长期记忆

企业级Agent记忆系统设计:从Context管理到LangGraph长期记忆 当你的 Agent 在生产环境跑了一个月用户开始问出这样的问题“你上次不是说帮我处理过工单吗”“我不记得你是老客户了”这时候你才会意识到Agent 不是模型不够聪明而是没有记忆。很多团队把 Agent 做成了“一次性的问答机器”每次请求都把用户的历史对话打包塞进模型上下文Token 预算飙到离谱还频繁撞上context length exceeded。你换更大的窗口买更高的额度问题依然存在——因为长期记忆的架构从来不是靠扩窗口解决的。本文想聊清楚一件事企业级 Agent 的记忆系统应该怎么设计从最基础的 Context 管理到跨会话的 Long-term Me以及 LangChain、LangGraph 和 DeepAgent 这类企业级框架在其中的角色。读完你会知道为什么模型窗口再大也会“失忆”记忆系统应该分层LangGraph 的图状态如何支撑记忆写入与检索以及生产环境里记忆治理真正该管什么。1. 这篇文章要解决的问题为什么 Agent 会“失忆”先看几个真实的生产现象。现象一用户在一个客服 Agent 里反复咨询同一个售后问题。第一次对话很顺利但用户隔天再来Agent 完全不记得他昨天提供过订单号和故障描述于是用户又从头讲一遍。体验断崖式下跌。现象二开发团队觉得“模型上下文窗口不够大”于是换了一款支持百万 Token 的模型。结果运行一段时间后日志里依然出现maximum context length exceeded或context is too large and auto-compaction could not recover。窗口再大也扛不住无限累积的会话历史。现象三LLM 在对话过程中会产生大量中间推理、工具调用结果和临时数据。开发团队为了省事把这些全部塞进 messages 列表导致每次请求的延迟和成本都在持续上升。用户每次追问模型都要把过去所有内容重新推理一遍。这些问题的共同根源是同一个把“记忆”等同于“把更多历史文本放进上下文窗口”。其实模型的 Context 窗口更适合类比为人类的“工作记忆”——它的容量有限作用是在当前任务中临时加工信息。而真正的长期记忆是另一套机制它需要存储、索引、检索、更新和遗忘。一个正常的企业级 Agent必须同时具备这两种记忆机制并且让它们协同工作。企业级场景比个人 Demo 更复杂多租户隔离、数据敏感等级不同、记忆写入可能引入噪声、历史事实可能被用户更正、合规上必须支持删除。这些都不是单纯调 Prompt 能解决的而是需要一套工程治理方案。这篇文章不是讲某个框架的 API 手册而是讲清楚“Agent 记忆系统”这个问题的架构演进路径。如果你是正在做 Agent 应用的工程师、技术负责人或者准备从 LangChain 过渡到 LangGraph 的开发者这篇文章应该能帮你少走几个月的弯路。2. 基础概念Context、会话记忆、长期记忆与 Long-term Me在进入代码之前先把几个经常被混用的概念分清楚。2.1 Context模型的“工作记忆”Context 是模型在一次请求中能看到的所有信息包括系统提示词、用户输入、历史消息、工具返回结果等。模型基于 Context 生成回复但 Context 是有长度上限、有成本、有延迟代价的资源。在 LangChain 和 LangGraph 的术语里messages数组通常就是这个 Context 的载体。每多一条历史消息模型输入 Token 就多一份开销。如果用工具调用工具返回的 JSON 也会占用 Context。所以Context 管理的第一原则是只把当前任务真正需要的信息放进窗口而不是把用户从注册到现在的所有行为都堆进去。2.2 会话记忆让同一轮对话“不迷路”会话记忆是 Agent 在一段连续对话中维持的状态。比如用户说了“帮我查一下昨天的订单”模型需要知道“昨天”是相对于哪一天、哪个账号、哪个订单。在实现层会话记忆通常由框架提供LangChain 里有各种 Memory 类例如ConversationBufferMemory、ConversationSummaryMemory、VectorStoreRetrieverMemory。LangGraph 则通过State和checkpointer把会话状态持久化每次请求从上一次的状态往下继续。会话记忆是最容易实现的一层但它解决不了“用户隔天再来”的问题。要解决跨会话的连续性问题需要真正意义上的长期记忆。2.3 长期记忆跨会话的事实与偏好长期记忆的核心是Agent 能从历史交互中沉淀出关于用户、业务和任务的持久知识。举个例子。用户第一次对话时说“我在上海用的是企业版套餐”。这句话如果被写进长期记忆下次用户无论什么时候来Agent 都能在生成回复时自动调用这条事实不需要用户重复说明。长期记忆的存储形态主要有几种记忆类型存储内容检索方式典型场景用户画像 / 属性记忆姓名、地区、套餐、偏好结构化查询SQL、键值个性化推荐、客服身份识别情景记忆过去某次对话、工单、操作记录向量相似度检索用户说“我之前投诉过”语义记忆业务知识、产品规则、领域概念向量检索 全文检索产品名别称、常见问题程序记忆工作流偏好、常用工具选择路由规则、脚本配置Agent 自动执行用户常用流程“Long-term Me”这个说法实质是强调长期记忆应该构建成关于“用户/实体”的个体化档案而不是一堆无法归属的杂散文本。每个用户、每个企业租户都应该有一个持续演进、可查询、可审计的长期记忆主体。2.4 为什么不能把记忆全部塞进上下文很多团队的第一个方案是“那我直接用一个大窗口把所有历史都扔进去”。这个方案在 Demo 阶段看似能用到生产环境会暴露几个问题。成本失控每次请求都携带全部历史Token 消耗随对话轮数线性增长甚至更糟。延迟失控长上下文会让模型首字返回时间明显变慢用户体验下降。语义稀释当窗口里充斥着大量无关历史时模型对当前关键信息的注意力会被稀释反而更容易答错。运行错误即使窗口足够宽中间步骤、工具返回和其他系统注入内容仍然可能超限。类似maximum context length和auto-compaction could not recover的错误往往就发生在这个阶段。长期记忆系统的意义不是“塞更多”而是“更精准地检索当前任务需要的那一小部分”然后把这一小部分注入到 Context 里。一句话总结记忆在外部Context 是记忆的临时投影。3. LangChain、LangGraph 与 DeepAgent 的定位与选型很多初学者会问LangChain 和 LangGraph 到底有什么区别是不是选一个就行再加上 DeepAgent 这类企业级框架三者关系是什么用一个比喻来理解LangChain 是“工具箱”LangGraph 是“流水线”DeepAgent 这类企业级框架是“工厂管理制度”。3.1 LangChain组件生态与快速原型LangChain 的核心价值是提供了一套统一的接口把模型调用、Prompt 管理、工具调用、向量存储、Agent 行为等组件串联起来。它适合快速搭建原型验证一个概念是否可行。在记忆场景中LangChain 提供了多种 Memory 实现。但对复杂生产系统来说直接使用高层 Memory 类往往不够灵活因为记忆的写入、更新、删除策略需要和业务逻辑深度绑定。3.2 LangGraph图状态与流程控制LangGraph 是建立在 LangChain 之上的一层图式编排框架。它的核心思想是Agent 的每一次运行不是一个黑盒调用而是一个有明确节点和边的状态图。LangGraph 的几个关键能力正好是记忆系统需要的State 管理每个节点都可以读取和修改全局状态。记忆可以在节点之间传递而不是塞在 messages 里。checkpointer 持久化把图的中间状态保存下来支持断点续跑、人工干预、会话恢复。条件路由根据状态内容决定下一步走向。例如判断“这次对话是否需要更新长期记忆”可以走不同的边。子图与并行分支可以把记忆写入、向量化、异步存储设计成独立的子图或并行执行不影响主流程响应速度。流式执行节点逐个执行可观察、可调试生产环境更容易定位问题。所以从记忆系统角度看LangGraph 是比 LangChain 更合适的底座它把“记忆读、记忆写、记忆更新”这些操作变成了流程图中可以精细控制的节点。3.3 DeepAgent企业级工程治理视角在标题中出现的 DeepAgent这里并不是指某一个具体 API 系列而是代表企业级 Agent 框架这一类方案。这类框架通常不仅提供基础编排能力还内置了权限、审计、记忆管理、多租户隔离、可观测性等生产级能力。从工程治理的角度看DeepAgent 这类框架解决的是“图能跑但跑不长久”的问题谁来写记忆记忆的访问权限怎么控跨租户的数据怎么隔离Agent 执行超时怎么处理这些问题不在 LangGraph 的职责范围内但却是企业落地时必须回答的。3.4 三者选型建议维度LangChainLangGraphDeepAgent 类企业框架定位组件工具箱图式编排框架企业级 Agent 治理框架记忆支持Memory 组件State checkpointer 节点编排记忆生命周期、权限、审计适合阶段原型验证、学习生产级 Agent 流程开发大规模多团队治理学习成本较低中等较高取决于团队规模典型问题灵活性不足需要自己实现治理绑定特定平台或规范实际项目的稳妥路线是用 LangChain 理解组件用 LangGraph 落地流程用 DeepAgent 类框架处理治理。如果团队规模不大、场景单一直接用 LangGraph 加自己的治理模块也是完全可行的。4. 环境准备与前置条件为了保证下面的代码能跑通需要先准备一套最小环境。版本信息请以各项目官方文档为准这里重点演示通用思路。4.1 基础环境Python 3.10 或 3.11多数 Agent 框架在此版本上兼容性最好pip 或 poetry 作为依赖管理工具可以访问 OpenAI 兼容接口或本地模型的 API Key一个向量数据库用于长期记忆存储4.2 安装依赖下面的依赖覆盖了 LangChain、LangGraph 和常见向量存储接口pip install --upgrade langchain langchain-openai langchain-core langgraph如果需要向量存储可以选一个轻量方案先跑通流程pip install langchain-community chromadb如果生产环境使用 PostgreSQL 做向量检索可以后续安装pgvector相关驱动。先不建议把存储绑定得太死先用内存或本地文件方案验证架构。4.3 环境变量在.env文件或系统环境中配置模型服务信息OPENAI_API_KEYyour-api-key OPENAI_API_BASEhttps://your-endpoint.example.com/v1如果你使用的是国内模型服务或企业内部网关OPENAI_API_BASE指向对应兼容地址即可。不要在代码中硬编码密钥。4.4 验证安装写一个最小脚本验证环境# app/check_env.py import os from langchain_openai import ChatOpenAI llm ChatOpenAI( modelgpt-4o-mini, api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_API_BASE), ) response llm.invoke(你好请回复环境正常) print(response.content)如果能正常输出说明模型调用链路已经打通。如果卡在这里优先检查 API Key、网络代理设置和模型服务地址是否可达。5. 先从短期记忆开始用 LangGraph 搭建一个最少可用的 Agent很多人一上来就做长期记忆结果代码复杂到无法维护。更稳妥的顺序是从短期记忆开始把会话状态管理跑通再逐步扩展。5.1 定义状态在 LangGraph 中State 是整个图的“公共上下文”。这里用messages字段保存对话历史add_messages是 LangGraph 提供的 reducer用于把新消息合并到已有消息列表。# app/agent_basic.py from typing import Annotated from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages]5.2 实现对话节点节点函数接收一个 State处理后返回更新后的 State 片段。这里让模型直接基于整个 messages 列表生成回复from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, temperature0.2) def chat_node(state: AgentState): response llm.invoke(state[messages]) return {messages: [response]}5.3 构建图并加入持久化短期记忆的关键是每次调用之间能保存 State。LangGraph 的checkpointer就是干这个的。先用内存版跑通生产环境再换 SQLite 或 Postgres。from langgraph.checkpoint.memory import InMemorySaver builder StateGraph(AgentState) builder.add_node(chat, chat_node) builder.add_edge(START, chat) builder.add_edge(chat, END) checkpointer InMemorySaver() graph builder.compile(checkpointercheckpointer) config {configurable: {thread_id: user-zhangsan-001}} # 第一次对话 result1 graph.invoke( {messages: [(user, 你好我是张三我的订单号是 T-2024-001)]}, config, ) print(result1[messages][-1].content) # 第二次对话 result2 graph.invoke( {messages: [(user, 我刚刚提到我的订单号是多少)]}, config, ) print(result2[messages][-1].content)运行第二次提问时模型应该能从messages历史中找到订单号。这里的thread_id相当于“会话编号”同一个会话 ID 的多次请求共享同一份状态。这就是短期记忆的最小实现。它解决了“同一会话内不迷路”的问题但还没有解决跨会话记忆。6. 从短期记忆到长期记忆分层架构设计要让 Agent 拥有 Long-term Me不是再加一个 Memory 组件那么简单而是要重新设计记忆的读写流程。6.1 记忆分层模型一套可落地的记忆架构建议分成三层层级存储介质生命周期典型数据工作记忆Context模型输入窗口单次请求当前对话、临时工具结果会话记忆Session Memory图 State checkpointer一次会话本轮对话历史、中间状态长期记忆Long-term Memory向量库 结构化表跨会话、长期用户偏好、事实、历史工单摘要长期记忆不是把所有对话都原样存下来而是在合适的时机从对话中提取关键事实按实体和主题组织再写入外部存储。使用时通过检索把相关记忆重新注入到 Context 中。6.2 记忆读写流程一个包含长期记忆的 Agent 运行周期通常会经过以下环节解析输入接收用户消息识别用户 ID、会话 ID。检索记忆根据用户输入和用户 ID从长期记忆中检索相关事实注入系统 Prompt。生成回复LLM 基于“系统 Prompt 检索到的记忆 当前消息”生成回复。沉淀记忆在对话结束后或异步判断这段对话中有没有值得长期保存的事实。如果有提取、去重、写入向量库或结构化表。这里最容易犯的错误是“全量写入”把整段对话直接丢进向量库。这样做的问题在于大量无关内容被存进去后续检索噪声高用户隐私数据被无限期保存合规风险大向量库会迅速膨胀检索速度下降。更合理的做法是让 LLM 以结构化形式抽取记忆例如返回一个 JSON包含实体、属性、关系和置信度再由程序判断是否写入、以什么方式写入。6.3 记忆的更新与遗忘长期记忆不是“写一次就永远不改”。用户可能更正自己的电话号码可能修改套餐可能对之前的事实进行补充。所以记忆系统还需要记忆更新当同一实体出现了新的、冲突的信息根据时间戳和置信度决定是否覆盖旧值。记忆遗忘企业要遵守数据保留政策用户也可以主动要求删除自己的数据。记忆系统必须提供按用户、按实体、按时间段删除的能力。记忆审计记录什么时间、哪个 Agent、基于哪次对话写入了哪条记忆。这在合规审计中很重要。这些能力正好是 LangGraph 节点化的优势所在记忆检索是节点记忆写入是节点记忆更新和遗忘也可以是节点甚至可以是独立的子图。7. 完整示例用 LangGraph 实现一个带长期记忆的 Agent下面用一个完整但不复杂的示例演示“检索记忆 - 生成回复 - 异步沉淀记忆”的核心流程。7.1 定义长期记忆存储接口先用最简单的向量存储实现一个可替换的存储层。生产环境可以替换为 Chroma、FAISS、pgvector 或 Milvus。# app/memory_store.py import uuid from langchain_openai import OpenAIEmbeddings from langchain_core.documents import Document from langchain_community.vectorstores import Chroma embeddings OpenAIEmbeddings() vector_store Chroma( collection_nameagent_long_term_memory, embedding_functionembeddings, persist_directory./memory_db, ) def save_memory(user_id: str, content: str, metadata: dict | None None): doc Document( page_contentcontent, metadata{ user_id: user_id, memory_id: str(uuid.uuid4()), **(metadata or {}), }, ) vector_store.add_documents([doc]) def search_memory(query: str, user_id: str, k: int 3): docs vector_store.similarity_search( query, kk, filter{user_id: user_id}, ) return docs注意这里的filter参数用于按用户隔离记忆。生产环境必须做租户级隔离避免用户 A 检索到用户 B 的记忆。7.2 定义带长期记忆的 State相比基础版 State这里增加了user_id、memory_hits和memory_updated字段# app/agent_memory.py from typing import Annotated from typing_extensions import TypedDict from langgraph.graph.message import add_messages class MemoryState(TypedDict): messages: Annotated[list, add_messages] user_id: str memory_hits: list memory_updated: bool7.3 实现记忆节点三个核心节点分别负责检索、生成、写入from langchain_openai import ChatOpenAI from langchain_core.messages import SystemMessage llm ChatOpenAI(modelgpt-4o-mini, temperature0.2) def retrieve_memory_node(state: MemoryState): last_message state[messages][-1].content hits search_memory(last_message, state[user_id], k3) return {memory_hits: hits} def generate_node(state: MemoryState): memory_text \n.join( [f- {doc.page_content} for doc in state[memory_hits]] ) system_prompt ( 你是一个企业客服 Agent。 以下是该用户的长期记忆回答时请合理使用\n f{memory_text}\n 如果长期记忆与当前对话冲突以当前对话为准。 ) messages [SystemMessage(contentsystem_prompt)] state[messages] response llm.invoke(messages) return {messages: [response]} def write_memory_node(state: MemoryState): # 生产环境建议用 LLM 抽取结构化记忆而不是全量写入 last_user_message state[messages][-2].content if len(state[messages]) 2 else last_assistant_message state[messages][-1].content combined f用户说{last_user_message}\nAgent 说{last_assistant_message} save_memory( user_idstate[user_id], contentcombined[:500], metadata{source: conversation}, ) return {memory_updated: True}write_memory_node里做了简化处理每次调用都会写入最后一条对话。真实项目中不建议这样做应该在写入前增加“是否值得记忆”的判断逻辑避免噪声入库。7.4 构建图并加入条件路由在基础流程上加一个条件判断如果用户明确表达了“这是关键信息”或者对话中有可能存在重要事实才进入写入分支。这里用一个非常简单粗暴的规则做演示from langgraph.graph import StateGraph, START, END def should_write_memory(state: MemoryState) - str: last_user_message state[messages][-1].content keywords [订单号, 我是, 电话, 地址, 偏好, 投诉, 工单] if any(kw in last_user_message for kw in keywords): return write return skip builder StateGraph(MemoryState) builder.add_node(retrieve_memory, retrieve_memory_node) builder.add_node(generate, generate_node) builder.add_node(write_memory, write_memory_node) builder.add_edge(START, retrieve_memory) builder.add_edge(retrieve_memory, generate) builder.add_conditional_edges( generate, should_write_memory, {write: write_memory, skip: END}, ) builder.add_edge(write_memory, END) graph builder.compile(checkpointerInMemorySaver())这里用add_conditional_edges实现了条件路由模型生成回复后根据关键词判断是否需要写入记忆。这只是演示真实项目可以用更复杂的规则或由 LLM 输出结构化判断结果。7.5 运行与验证# app/run_demo.py config {configurable: {thread_id: session-zhangsan-001}} # 第一次对话告诉 Agent 自己的订单号 graph.invoke( { messages: [(user, 你好我是张三我的订单号是 T-2024-001)], user_id: zhangsan, }, config, ) # 新会话验证长期记忆是否生效 config2 {configurable: {thread_id: session-zhangsan-002}} result graph.invoke( { messages: [(user, 你还记得我的订单号吗)], user_id: zhangsan, }, config2, ) print(result[messages][-1].content)第二次对话使用了新的thread_id但如果之前写入的记忆已经进入向量库retrieve_memory_node应该能检索到包含订单号的记忆并注入生成节点的系统 Prompt。这就是“跨会话记忆”的最小区块链验证。如果你的模型返回了正确的订单号说明长期记忆链路已经打通。8. 生产环境中的记忆治理权限、更新与审计代码跑通只是第一步。企业级 Agent 记忆系统真正复杂的地方在于“治理”。8.1 记忆访问的权限边界长期记忆存储了大量用户敏感信息。生产环境中必须遵循最小权限原则检索记忆前校验用户身份和租户身份向量检索必须带租户过滤条件从存储层隔离数据不同角色的 Agent客服、运营、管理员应有不同的记忆访问范围内部工具调用记忆接口时需要有独立的服务账号和访问审计。如果记忆数据跨系统共享建议通过统一 API 网关访问而不是让每个 Agent 直接连接向量库。8.2 记忆写入的质量控制“写入垃圾记忆”比“丢失记忆”更可怕。因为脏记忆一旦进入长期存储就会持续影响后续所有对话。建议引入记忆写入的审核机制由 LLM 抽取结构化候选记忆返回 JSON 字段设置置信度阈值置信度过低的不写入对用户明确更正的旧记忆执行更新而非追加对敏感信息身份证、银行卡做脱敏或加密存储定期抽样检查记忆库内容清理明显错误或过时数据。8.3 遗忘与删除合规视角下记忆系统必须支持“被遗忘权”。这意味着用户注销后能删除其在记忆库中的所有记录支持按user_id批量删除并且删除操作要有审计留痕向量库中的删除不能只删索引要同步清理底层数据如果记忆被复制到缓存或备份需要设计过期机制。8.4 可观测性与监控记忆系统需要监控的指标包括指标含义建议监控方式记忆检索命中率有效检索次数 / 总请求次数命中率过低说明记忆写入策略有问题记忆写入成功率成功写入条数 / 尝试写入条数过高可能说明写入缺少筛选检索延迟记忆检索接口的 P95 延迟向量库索引和过滤器效率记忆删除延迟删除请求的处理时间合规请求需要 SLA上下文 Token 消耗每次请求注入的记忆长度长期监控是否出现无效膨胀在生产环境建议为记忆读写日志单独建索引方便排查“为什么 Agent 回答得不对”很多时候是因为检索到了错误记忆或旧记忆。9. 常见问题与排查思路下面汇总几个 Agent 记忆系统开发中常见的问题以及排查方向。问题现象可能原因排查方式解决方案出现maximum context length exceeded会话历史或记忆注入过多检查请求日志中的 Token 统计限制 messages 数量改用摘要记忆降低记忆注入条数context is too large and auto-compaction could not recover历史压缩失败上下文超出恢复范围查看框架的自动压缩日志手动清理历史调整压缩策略必要时重新启动会话Agent 在跨会话后“不记得”用户信息长期记忆没有写入成功查记忆库中是否存在该用户记录检查写入节点是否执行、向量筛选条件是否正确检索到了其他用户的记忆向量检索过滤条件缺失或错误审查检索接口的 filter强制在存储层按租户 ID 过滤不能只在应用层过滤记忆库不断膨胀检索越来越慢全量写入策略不合理查看记忆条数和平均 Token 大小引入结构化抽取、去重、定期清理策略error running context: ssl communication网络代理、证书链或服务地址配置问题检查网络代理、CA 证书、服务端点更新证书、确认代理配置确保模型服务地址可达Agent 执行超时返回 provider did not respond in time模型服务响应慢或记忆检索环节阻塞查看调用链路各环节耗时为 LLM 调用和记忆检索设置超时将记忆写入改为异步排查这类问题不要一上来就怀疑框架。正确顺序是先看日志确认是上下文超限、存储查询失败还是模型响应慢再对症处理。10. 总结与后续学习方向回到开头那个场景用户问“你上次不是帮我处理过吗”如果 Agent 能立刻从自己的长期记忆中调出上次的工单、结论和承诺这个体验就会完全不同。而做到这一点靠的并不是一个更大的模型窗口而是一套完整的记忆架构。本文的核心结论可以概括为三句话Context 是工作记忆长期记忆应该在外部存储通过检索注入而不是全量塞入。LangGraph 的 State、checkpointer、条件路由和子图是搭建记忆读写流程的合适底座LangChain 负责组件连接DeepAgent 类框架负责企业级治理。长期记忆系统的难点不在“写入”而在“治理”权限隔离、更新策略、遗忘删除、审计监控这些才是决定生产系统能否长期稳定运行的关键。建议你从本文第 5 节的最小 Agent 开始先跑通会话记忆然后按第 7 节加入向量检索和记忆写入验证跨会话记忆最后再参考第 8 节补齐权限、审计和删除能力。每一步都不复杂但组合起来就是一个可以交付到生产环境的记忆系统。如果继续深入可以关注这几个方向结构化记忆抽取把对话转为实体关系图谱、记忆冲突消解同一事实被用户多次变更时的处理策略、记忆融合与共享同一企业在多个 Agent 之间共享记忆但保留权限边界以及 Agent 执行链路的可观测性治理。把这些做扎实Agent 的“长期记忆”才真正配得上 Long-term Me 这个名字。
返回列表