ARTICLE DETAIL

资讯详情

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

从上下文窗口到长期记忆:企业级Agent记忆系统构建指南

从上下文窗口到长期记忆:企业级Agent记忆系统构建指南 当你的 Agent 在一次长对话中突然抛出codex ran out of room in the models context window. Start a new thread or compact the conversation.或者api error: 400 this models maximum context length is 1048576 tokens. However, your messages resulted in 1200000 tokens.这类错误时很多人的第一反应是把模型上下文窗口买大一点或者提醒用户“开个新会话”。如果只是写一个演示用的 ChatBot这个办法确实够了。但如果你正在做的是需要长期服务某个用户的智能客服、私人助理或多步骤任务调度系统你会很快发现问题的本质不在窗口大小而在于你的 Agent 根本没有一套记忆系统。Agent 能不能像人一样记住“你是谁”“你上次聊到哪一单”“你偏好简短的答案还是带表格的答案”以及过去三个月里业务规则发生过哪些变化这已经不纯粹是模型能力问题而是系统架构与工程治理问题。本文要和大家聊的正是如何从 Context 走向 Long-term Me构建一套企业级 Agent 记忆系统并解释 LangChain、LangGraph 与 DeepAgent 这类框架在记忆工程里分别扮演什么角色。读完你会得到一个可落地到生产环境的记忆分层思路、一组基于 LangGraph 的持久化代码示例以及一份排查记忆类事故的检查清单。1. 这篇文章真正要解决的问题先说结论企业级 Agent 与 Demo 级 Agent 的最大分水岭就是记忆系统是否可设计、可治理、可恢复。单纯依赖模型上下文窗口就像一个人每次见面都重新自我介绍不仅浪费 token更无法建立持续的信任关系。我在不同团队里见过三类典型困境你应该也遇到过其中一种。第一类是智能客服场景。用户第一轮问“我的订单为什么还没发货”Agent 调用了订单接口给出答复。第二轮用户说“那帮我申请退款”如果 Agent 没有记住上一轮刚查过的订单号就要重新问一遍体验非常割裂。更麻烦的是用户可能三天后又回来问“上次那个退款到账了吗”如果系统没有跨会话记忆整个对话要从零开始。第二类是任务型 Agent 场景。一个多步骤工作流比如“调研竞品 → 生成报告 → 发送邮件”中间任意一步失败重试时已经完成的上游结果能不能复用任务的状态、中间产物、重试次数这些都属于记忆。没有状态持久化的 Agent一旦进程重启或请求超时整个任务就像断电的服务器一样丢失现场。第三类是长期画像场景。企业内部的知识库助手、销售助手、面试官助手如果每次都把用户的姓名、偏好、历史决策当成新信息来理解不仅浪费 token还会让用户觉得这个系统“很蠢”。真正生产级的 Agent需要长期记住“这个人是谁、他关心什么、我们之间发生过什么”这个持续存在的用户模型就是标题里说的 Long-term Me。这三类困境有一个共同本质Agent 的上下文窗口只是“一次会话的工作台”而记忆系统要解决的是把工作台延伸到“整个客户生命周期”。因此本文不会教你如何把全部历史一股脑塞进 Prompt而是从架构层面给出分层记忆设计并展示如何在 LangGraph 中实现状态持久化和记忆检索最后讨论企业落地时绕不开的安全、权限与审计问题。2. Agent 记忆的基础概念与三层模型要讲清楚记忆系统先得把“记忆”这个词从文学比喻还原成工程概念。人类记忆分为感觉记忆、工作记忆和长期记忆Agent 的记忆也可以做类似的映射。很多初学者把 Agent 记忆等同于“把聊天记录存到数据库”这是典型的误区。聊天记录只是最原始的日志真正的记忆系统要解决的是如何从日志中提取、组织、检索和遗忘信息。在企业级 Agent 架构里我更推荐把记忆拆成三层来设计。第一层是即时上下文Context也就是一次请求内送入模型的所有内容包括系统 Prompt、用户最新输入、工具返回结果、相关检索片段。这一层的特点是“请求结束即消失”它对应模型的上下文窗口也是你每天在报错日志里看到的 token 超限问题发生的地方。第二层是工作记忆Working State也就是一个多步骤任务在执行过程中的中间状态。比如 Agent 正在执行“查天气 → 订餐厅 → 发邀请”的流程天气已经查完餐厅正在选择这些中间结果必须保存在可恢复的存储中否则一次网络抖动就会让整个任务崩溃。在 LangGraph 里工作记忆对应 StateGraph 的 State它可以被 Checkpointer 持久化到数据库。第三层是长期记忆Long-term Memory这是真正体现“Long-term Me”价值的一层。它需要跨会话、跨任务地保存用户画像、历史偏好、业务规则和企业知识。长期记忆通常不是原始聊天记录而是经过提取、向量化、摘要化之后的结构化信息。比如系统从过去十次对话中提取出“用户偏好 Markdown 格式”“用户公司是某制造企业”“用户拒绝过 A 方案的报价”等事实这些事实才是长期记忆的核心。三层之间的关系是逐级向上汇聚的即时上下文是瞬间状态工作记忆是任务生命周期内的状态长期记忆是跨任务、跨会话积累下来的稳定知识。没有第三层Agent 只能在新会话里重新认识用户没有第二层Agent 连一个复杂任务都跑不完没有第一层Agent 无法完成任何一次具体推理。理解了这三层你再看那些“上下文窗口不够用”的报错就会明白把窗口从 10 万 token 换到 100 万 token只解决了第一层的容量问题并没有解决第二层和第三层的结构化问题。真正合理的方案是在进入模型之前先完成记忆检索和上下文压缩只把最相关的信息送进窗口。3. 框架选型LangChain、LangGraph、DeepAgent 的分工与演进聊完记忆模型再看工具链。很多开发者对 LangChain、LangGraph 和 DeepAgent 的边界感到困惑尤其是“LangChain 是不是过时了”“LangGraph 和 LangChain 到底什么关系”这类问题几乎每周都能在社区里看到。我的判断是它们不是同一个层级的替代品而是分别解决编排、状态和平台治理三个不同问题。LangChain 是最早被广泛接受的 Agent 开发框架它提供了一套统一的抽象把 Prompt 组装、模型调用、工具调用、输出解析封装成链式调用。它的价值在于降低了 Agent 开发门槛让一个新手可以在几小时内写出一个能调用搜索和计算器的 Agent。但它的局限也很明显传统的 Chain 设计是线性的难以表达循环、分支、人工介入和复杂状态流转。你可以在 LangChain 里硬写出一个多步 Agent但当任务需要回退、并行、重试时代码会迅速变成一团乱麻。用一句话概括LangChain 适合“从零开始快速搭 Demo”但不适合做复杂状态机。LangGraph 正是为状态流控而生的。它把 Agent 的执行过程建模成一个图结构节点Node是处理逻辑边Edge是流转控制State 是全局共享状态。和 LangChain 线性 Chain 最大的区别是LangGraph 天然支持循环、分支、条件路由而且每一个 State 都可以通过 Checkpointer 持久化到外部存储。这意味着工作记忆可以真正落地任务中断后重启还能从最后保存的状态继续执行。LangGraph 不是 LangChain 的替代品而是 LangChain 生态中面向复杂 Agent 工作流的那部分。事实上LangChain 官方也推荐在做多步骤、可恢复的 Agent 时使用 LangGraph“LangChain 过时了吗”的说法并不准确更准确的说法是“简单链式调用仍在用 LangChain复杂状态机应该用 LangGraph”。DeepAgent 这类企业级 Agent 平台则是在模型、流程之上再加一层治理能力。从公开材料看DeepAgent 解决的问题更偏工程化如何统一管理 Agent 的配置、记忆、工具权限、上下文策略和审计日志。如果说 LangGraph 给了你构建状态机的能力那么 DeepAgent 这类平台负责把这些能力变成企业里多个团队可以共用的标准化底座。它和 LangGraph 并不冲突更多是“框架层”与“平台层”的关系。对于技术选型我的建议是如果你的项目还在验证阶段直接用 LangGraph 向量数据库把记忆逻辑写清楚如果公司已经有多条 Agent 业务线需要统一记忆模型和权限体系可以评估 DeepAgent 或同类平台。下面的表格可以帮助你快速理解边界。框架/平台核心抽象主要解决典型使用阶段LangChainChain、Tool、Prompt Template快速编排 LLM 调用原型验证、简单问答LangGraphStateGraph、State、Node、Edge复杂状态流转、任务持久化生产级工作流 AgentDeepAgentAgent 平台、记忆治理、权限审计多业务线统一治理企业级平台建设4. 架构设计从 Context 到 Long-term Me 的分层记忆方案有了三层的抽象和框架选型现在把它们组合成一套完整的企业级记忆架构。这套架构的核心设计原则是任何进入模型的内容都不是直接来自原始日志而是经过一个记忆管道加工后的产物。这个管道由提取、存储、检索、压缩四个环节组成。先讲提取。原始对话记录进入系统后不能只做简单的拼接存储而是要通过 LLM 或规则抽取出有意义的结构化信息。比如从用户的提问中提取姓名、公司、偏好、未完成事项从业务对话中提取订单号、售后原因、处理结果。这些提取结果以键值对或 JSON 结构写入长期记忆库。没有提取这一步长期记忆库很快就会被重复、噪声信息淹没。再讲存储。短期工作记忆建议使用支持事务的 KV 数据库或关系数据库因为任务状态需要强一致性和快速读取长期记忆库则适合使用向量数据库加结构化数据库的混合方案向量库负责语义检索结构化库负责事实查询和权限控制。用户画像、企业知识图谱这类“稳定事实”更适合结构化存储而“用户上次提到的一个模糊想法”更适合向量化存储。然后是检索。在每一轮对话进入模型之前系统会基于当前用户和当前问题从长期记忆库中召回 Top-K 条相关记忆再和即时上下文拼接。这里要注意不要试图把“用户的全部历史”都塞进 Prompt而只召回当前任务确实需要的部分。加入一个简单的打分环节比如同时计算向量相似度和时间衰减因子可以显著提升召回质量。最后是压缩。即使做了检索召回多轮对话的累积历史仍然可能触达窗口上限。通常的做法是超出阈值后把早期历史用 LLM 生成摘要替换成“摘要 最近 N 轮完整消息”的结构。这一步直接对应热词里那条context is too large and auto-compaction could not recover this turn的报错——如果压缩策略设计得好就不会走到 auto-compaction 失败的绝境。从数据流角度看架构可以描述为用户请求 → LangGraph 节点编排 → 记忆检索节点读取长期记忆→ 工作状态节点读当前任务→ Prompt 组装节点合并压缩→ 模型调用 → 结果返回 → 记忆写入节点异步更新状态和长期画像。每一步都需要可观测方便定位“为什么 Agent 没有记住某件事”。5. 代码实现基于 LangGraph 的状态设计与记忆持久化下面我们进入可落地部分。以一个“带记忆的售后服务 Agent”为例演示如何在 LangGraph 中实现工作记忆持久化和长期记忆读取。环境要求是 Python 3.10安装 langgraph 和相关依赖版本请以实际项目为准本文代码重点演示通用思路。首先定义状态。在 LangGraph 中State 是一个 TypedDict所有节点读写同一个状态对象。这里需要注意列表类型的字段如果直接赋值会被覆盖需要使用Annotated[list, operator.add]做 Reducer让每次节点返回的新消息追加到原有列表。# 文件路径agent_memory/state.py from typing import Annotated, TypedDict from operator import add class AgentState(TypedDict): # 当前会话消息多个节点可以追加 messages: Annotated[list, add] # 用户唯一标识用于记忆隔离 user_id: str # 长期记忆从记忆库加载 long_term_memory: dict # 工作记忆里的任务状态 task_status: str retry_count: int接着实现长期记忆检索节点。这里假设memory_store是一个封装了向量数据库和关系数据库的客户端query_user_profile负责按 user_id 读取结构化画像query_similar_memories负责按用户问题做向量召回。实际项目中这些 API 可以换成 Redis、MongoDB、Milvus 或 pgvector 的实现。# 文件路径agent_memory/nodes/memory_nodes.py from agent_memory.state import AgentState from agent_memory.store.memory_store import memory_store def load_long_term_memory(state: AgentState) - dict: user_id state.get(user_id, unknown) latest_question state[messages][-1][content] if state[messages] else # 读取用户稳定画像 profile memory_store.query_user_profile(user_id) # 根据当前问题做语义检索 related memory_store.query_similar_memories( user_iduser_id, querylatest_question, top_k5, ) return { long_term_memory: { profile: profile, related: related, } } def load_working_state(state: AgentState) - dict: user_id state.get(user_id, unknown) current_task memory_store.query_task_state(user_id) if current_task: return { task_status: current_task.get(task_status, new), retry_count: current_task.get(retry_count, 0), } return { task_status: new, retry_count: 0, }接下来是业务节点和路由逻辑。业务节点假装调用一个售后 API并返回结果路由逻辑根据task_status决定流程是继续还是结束。# 文件路径agent_memory/nodes/handle_order_node.py from agent_memory.state import AgentState def handle_order(state: AgentState) - dict: user_id state[user_id] profile state[long_term_memory].get(profile, {}) # 如果有用户历史订单偏好可以拼入分析逻辑 preferred_reason profile.get(preferred_refund_reason, 未填写) # 此处替换为真实订单查询接口 order_result { order_id: ORD-2025-001, status: 已发货, reason: preferred_reason, } return { messages: [{role: assistant, content: f查询到订单 {order_result[order_id]} 当前状态为 {order_result[status]}}], task_status: finished, }现在把这些节点组装成图。我们用StateGraph创建图添加记忆加载节点、售后处理节点并通过add_conditional_edges控制路由。关键地方是compile时传入 Checkpointer这样每一次节点执行后的 State 都会自动持久化。下面代码使用InMemorySaver作为内存版 Checkpointer适合本地调试生产环境建议替换为SqliteSaver、PostgresSaver或 Redis 等外部实现。# 文件路径agent_memory/graph.py from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import InMemorySaver from agent_memory.state import AgentState from agent_memory.nodes.memory_nodes import load_long_term_memory, load_working_state from agent_memory.nodes.handle_order_node import handle_order def should_continue(state: AgentState) - str: if state.get(task_status) finished: return end return handle_order def build_graph(): graph StateGraph(AgentState) # 注册节点 graph.add_node(load_long_term_memory, load_long_term_memory) graph.add_node(load_working_state, load_working_state) graph.add_node(handle_order, handle_order) # 定义入口和顺序 graph.set_entry_point(load_long_term_memory) graph.add_edge(load_long_term_memory, load_working_state) graph.add_conditional_edges( load_working_state, should_continue, { handle_order: handle_order, end: END, }, ) graph.add_edge(handle_order, END) # 持久化内存版 Checkpointer生产替换为外部存储 checkpointer InMemorySaver() app graph.compile(checkpointercheckpointer) return app最后是运行入口。每次调用需要传入一个thread_id它就是工作记忆的隔离维度。同一个用户的不同任务可以使用不同 thread_id但长期记忆始终按user_id隔离。运行后你会在输出里看到 Agent 成功加载了用户画像并在查询订单时使用了画像中的偏好字段。# 文件路径run_service_agent.py from agent_memory.graph import build_graph app build_graph() config {configurable: {thread_id: task-001, user_id: user-12345}} result app.invoke( { user_id: user-12345, messages: [{role: user, content: 帮我查一下最近订单的状态}], }, configconfig, ) print(result[messages][-1][content])如果你愿意可以把InMemorySaver()换成SqliteSaver然后重启 Python 进程再跑一次你会发现任务状态可以从上次中断点恢复。这就是工作记忆持久化真正的价值Agent 的“记忆”不再存放在进程内存里而是存放在可恢复的存储介质中。6. 记忆检索、上下文压缩与多轮会话管理有了状态持久化Agent 已经不会在任务中途“断电失忆”。但长期运行的多轮会话还会碰到另一个问题历史对话太长就算有向量检索也不能把几十轮原文全部保留在 State 里无限膨胀。所以记忆管道里还需要两个关键环节检索优化和上下文压缩。检索优化的核心是让“最该被想起的记忆”出现在 Prompt 里。一个常见的低效实现是把用户近 50 轮消息全部向量化后按相似度排序取 Top 5。这样做的问题是忽略了时间衰减和事实稳定性。比如用户三个月前说过“我喜欢用表格”这个偏好今天仍然有效但向量相似度可能很低而用户上周说的“这个接口报错”对当前对话可能已经没有价值。更合理的打分公式会同时考虑语义相似度、时间衰减、记忆类型权重。下面是一个简化示意def memory_score(similarity: float, age_days: int, memory_type: str) - float: if memory_type user_preference: # 用户偏好长期有效 time_penalty 0.95 else: # 短期事件衰减更快 time_penalty 0.9 ** age_days return similarity * time_penalty上下文压缩解决的是“窗口溢出”问题。最简单可靠的做法是维护一个message budget比如 8000 token。当历史消息超过预算时把最早的消息交给一个摘要模型生成一个“此前的对话摘要”然后用“摘要 最近 N 轮完整消息”替代原始全量历史。注意摘要本身也要设置大小上限否则摘要会越滚越大。伪代码如下def compact_messages(messages, max_tokens8000, summary_modelllm): budget max_tokens recent [] total_tokens 0 # 倒序遍历从最近的消息开始保留完整内容 for msg in reversed(messages): msg_tokens count_tokens(msg) if total_tokens msg_tokens budget * 0.6: break recent.append(msg) total_tokens msg_tokens budget - msg_tokens recent.reverse() # 剩余早期消息压成摘要 early_messages messages[:len(messages) - len(recent)] if early_messages: summary summary_model.invoke( 请把以下对话压缩成 200 字以内的摘要保留关键事实 \n.join(m[content] for m in early_messages) ) return [{role: system, content: f历史摘要{summary.content}}] recent return recent多轮会话管理还有一个容易忽略的点消息写入异步化。用户请求进入模型前不需要等待长期记忆写入完成但状态持久化必须在任务执行过程中同步完成否则崩溃会丢状态。一个实用的分层策略是工作记忆同步持久化长期记忆异步更新。这样既保证了任务恢复的可靠性又不增加用户请求的延迟。7. 常见问题与排查思路记忆系统上线后最常见的故障往往不是模型能力不足而是记忆管道哪里断了。本节整理了几类高频问题建议收藏备用。问题现象可能原因排查方式解决方案请求报错context length is maximum历史消息未经压缩直接全量入 Prompt查看请求日志中的 token 统计启用上下文压缩将早期消息转摘要Agent 回答不记得用户历史长期记忆检索失败或未生效检查记忆库查询日志手工执行同款查询验证向量库连接和 top_k 阈值同一个用户出现 A/B 串号记忆隔离维度用错检查 thread_id 与 user_id 的映射长期记忆一律按 user_id 隔离工作记忆按 thread_id任务中断后重试从头开始Checkpointer 未配置或不可用查看 LangGraph 状态持久化配置使用外部持久化 Checkpointer 替代内存版Agent 输出包含敏感历史信息检索召回范围过宽检查召回记忆的权限标签记忆存储增加可见性字段按角色过滤报错SSL communication error代理或证书问题检查网络代理和 CA 证书为 HTTP 客户端配置正确的 SSL 上下文报错execution provider did not respond in time模型调用超时查看调用链耗时和 Provider 状态增加超时重试或改用更快的模型报错auto-compaction could not recover this turn上下文已严重超限压缩动作晚了一步检查对话轮次长度和 token 估算逻辑在压力到达阈值之前主动压缩预留安全余量排查这类问题有个通用方法给记忆管道的每一段加可观测性。比如记录“读取了哪几段记忆、它们来自哪个存储、拼接后总 token 数、压缩后总 token 数”。只要这些关键节点都有日志上述大部分问题都能在五分钟内定位。8. 企业级记忆治理安全、权限与一致性架构与代码能解决“能不能记住”的问题但企业环境还关心“谁允许记住什么”和“记错了怎么办”。记忆治理是 Agent 工程里最容易在早期被忽视、后期最难补的环节。首先是数据隔离。长期记忆里保存的是用户的真实信息一旦串号就是严重事故。最基础的要求是所有记忆读写接口必须强制校验user_id存储 key 也必须是user_id 命名空间的组合。更进一步可以按业务线划分命名空间比如客服助手和销售助手可以共享部分用户偏好但不应共享内部报价等敏感信息。其次是权限模型。不是所有 Agent 都能读取用户的全部记忆。一个常规做法是给记忆打上可见性标签例如public、internal、secret并在检索时根据当前 Agent 的身份过滤。这一点和 RAG 里的权限控制是同一套思路检索层过滤比 Prompt 层过滤更可靠因为最后一步的 Prompt 很容易被越权指令绕过。第三是一致性与版本管理。长期记忆会不断被更新如果更新逻辑没有约束昨天“用户不喜欢电话沟通”的画像可能今天就被一条错误提取覆盖成“用户喜欢电话沟通”。生产实践上记忆写入应当是“新增事实”而不是“覆盖事实”并为每个事实保留来源会话 ID 和时间戳。这样即使出错也能回溯到是哪一轮对话、哪个 Agent 写了这条记忆。最后是审计与遗忘。企业在做 Agent 时必须尊重用户的“被遗忘权”。设计记忆系统时要预留删除接口不仅删除向量数据库里的 embedding还要删除结构化画像、关联日志和摘要缓存。审计日志则需要记录每一次记忆写入和读取至少做到“什么时候、谁、基于什么会话、写了或读了哪条记忆”。这不是锦上添花而是 Agent 能够进入严肃业务场景的前提。如果你在 DeepAgent 这类企业级平台上建设记忆系统上述治理能力通常已经具备一部分。但无论平台提供多少能力架构师都需要理解底层机制才能在平台能力不足时自己补齐。9. 总结与后续学习方向再回到开头那个报错。真正让 Agent 不“失忆”的关键不是盲目买更大的上下文窗口而是把记忆当成一个独立的、可治理的子系统来设计。本文给出的三层记忆架构、LangGraph 状态持久化示例、上下文压缩策略和治理清单足够支撑一个 Demo 级 Agent 走向生产环境的第一版。如果你正准备在项目里实践建议按照这四步走第一先盘点自己的业务属于哪类记忆困境是任务状态丢失、用户画像缺失还是上下文超限第二用 LangGraph 写一个最小闭环把工作记忆持久化跑通第三接入向量库把长期记忆检索和上下文压缩做出来第四加上审计、权限和可观测性再谈规模化。后续可以继续深入的方向包括LangGraph 里更复杂的状态 Reducer 设计、基于图的记忆组织知识图谱 向量检索混合方案、记忆遗忘策略的算法设计以及多 Agent 共享记忆时的冲突消解。这些话题每一个都能单独写一篇长文。希望这篇文章能帮你在 Agent 记忆这条路上少踩几个真正的坑。
返回列表