当你的 Agent 在一次长对话中突然抛出codex ran out of room in the model's context window. Start a new thread or compact the conversation.或者api error: 400 this model's 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 或同类平台。下面的表格可以帮助你快速理解边界。
| 框架/平台 | 核心抽象 | 主要解决 | 典型使用阶段 |
|---|---|---|---|
| LangChain | Chain、Tool、Prompt Template | 快速编排 LLM 调用 | 原型验证、简单问答 |
| LangGraph | StateGraph、State、Node、Edge | 复杂状态流转、任务持久化 | 生产级工作流 Agent |
| DeepAgent | Agent 平台、记忆治理、权限审计 | 多业务线统一治理 | 企业级平台建设 |
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_id=user_id, query=latest_question, top_k=5, ) 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(checkpointer=checkpointer) 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": "帮我查一下最近订单的状态"}], }, config=config, ) 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_tokens=8000, summary_model=llm): 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 记忆这条路上,少踩几个真正的坑。