各位做 Agent 应用开发的同学应该都有过这样的体验单轮对话效果不错一进入多轮、跨会话、跨场景的复杂任务模型就开始“失忆”。明明用户之前已经提供过偏好、项目背景、历史结论但 Agent 换一个会话之后所有上下文清零只能重新交代。更麻烦的是模型上下文窗口是有限的哪怕是 128K、1M 的窗口也扛不住长期积累的对话历史、工具调用记录、中间推理过程无限写入。本文围绕企业级 Agent 记忆系统的架构设计与工程治理展开讲解怎么从最基础的 Context 管理一步步升级到 Long-term Memory并结合 LangChain、LangGraph 以及 DeepAgent 三种不同抽象层次框架给出可落地的实现思路和代码案例。内容会比较长我会尽量把概念讲清楚代码给完整。适合两类读者正在使用 LangChain、LangGraph 做 Agent 开发想解决上下文丢失、Token 超限、多轮记忆力弱等问题的开发者负责企业级 AI 应用架构设计需要为团队制定 Agent 记忆治理方案的技术负责人。读完本文你会掌握记忆分类模型、Context 窗口管理策略、LangChain 记忆组件用法、LangGraph 状态流持久化方案以及企业级记忆治理的工程规范。1. 背景与核心概念为什么 Agent 必须解决“记忆”问题1.1 没有记忆的 Agent 是什么状态先看一个很常见的业务场景。用户对客服 Agent 说“我上周咨询过企业发票配置问题当时确认了使用电子发票税号也提交过了现在我想改一下接收邮箱。”如果这个 Agent 没有任何跨会话记忆它看到这条消息时对“上周咨询过”“确认了电子发票”“税号提交过”这些信息完全没有概念。它只能把这句话当成一个全新问题重新让用户提供企业名称、纳税人识别号、发票类型、历史处理单号……用户体验非常差。这正是 Agent 记忆系统要解决的核心问题让 Agent 在单次对话、多次对话甚至跨业务场景中保留并使用历史信息。1.2 从 Context 到 Long-term Memory 的本质在聊记忆之前必须先理解 Context 是什么。Context 是模型一次调用中能看到的全部文本内容。它通常由三部分组成系统提示词System Prompt对话历史Conversation History当前用户输入User Query模型的能力上限由训练决定但每轮请求能“记住”多少信息完全取决于 Context 窗口大小和内容组织方式。Long-term Memory 则是一种超越上下文窗口限制的能力。它的核心思路是不把所有信息都塞进一次模型调用而是按需检索、按需注入。Agent 需要记忆时从外部存储中找到最相关的内容只把这一部分加入 Context而不是把历史数据全部堆给模型。1.3 记忆的分类不是所有信息都该用同一种方式保存做 Agent 记忆架构首先要建立分类意识。记忆类型保存内容典型例子实现方式短期工作记忆当前对话中的中间状态用户刚才输入的商品名称、当前正在执行的步骤对话上下文变量、状态对象长期语义记忆用户属性、偏好、业务规则用户所在城市、发票类型偏好、公司规模数据库、键值存储、向量库长期情景记忆过去发生过的具体事件用户上周咨询过发票配置问题向量检索 摘要存储程序记忆Agent 如何完成任务的流程知识发票配置流程、售后处理标准作业程序工作流定义、技能库、提示词模板不同记忆类型对存储介质、检索方式、更新策略的要求完全不同。很多团队一开始就把所有历史消息塞进向量库结果就是检索噪音大、Token 消耗高、关键信息反而找不到。后面的章节会详细展开每种记忆在企业级实现中的具体方案。1.4 三个框架的定位差异LangChain、LangGraph、DeepAgent 三者经常一起出现但它们的抽象层次不同。LangChain 是一个全能型开发框架提供了大量模型接入、Prompt 管理、记忆组件、工具调用、RAG 链路的模块。它的记忆组件比较齐全适合快速搭建但流程控制能力相对弱。LangGraph 主打有状态、可编排的图结构 Agent 流程。核心概念是 State、Node、Edge、Checkpoint特别适合处理需要条件分支、循环、并行、人工介入的复杂任务。它的 Checkpoint 机制天然适合做记忆持久化。DeepAgent 偏向更深层的 Agent 基础能力建设关注 Agent 的技能编排、上下文工程、复杂推理过程治理。在企业级架构中可以将 DeepAgent 视为一个更上层的平台层概念负责把记忆、工具、模型、流程治理统一收口。企业落地时不同团队可能把这个“平台层”命名为 DeepAgent、Agent Gateway 或 Agent Runtime本质都是在 LangGraph 这类图编排之上构建统一的运行治理能力。2. 环境准备与版本说明为了避免后续代码示例出现环境差异问题先说明本文示例的运行环境。依赖建议版本说明Python3.10LangChain 与 LangGraph 均对 3.10 支持良好langchain0.2.x / 0.3.x不同版本 API 有差异按实际安装版本调整langchain-openai0.1.xOpenAI 兼容接口langgraph0.2.xLangGraph 主包langgraph-checkpoint-sqlite0.1.xSQLite 持久化检查点chromadb0.4.x向量存储示例用python-dotenv1.0.x环境变量管理安装依赖pip install langchain langchain-openai langgraph langgraph-checkpoint-sqlite chromadb python-dotenv如果你使用的是国内模型或开源模型只要接口兼容 OpenAI 格式代码基本通用。在初始化模型时替换 base_url、api_key、model 名称即可。需要注意LangChain 和 LangGraph 更新速度很快部分类名和参数名会变化。本文代码以常见版本为示例重点演示设计思路。如果你安装的版本 API 有差异优先查阅官方文档和当前版本的迁移说明。3. 核心原理Context 管理与记忆组织3.1 Context 窗口的硬约束先直观感受一下 Context 窗口限制。假设一个模型的上下文窗口为 128K tokens。看起来很大但实际分配下来非常紧张系统提示词2K tokens工具定义Function Calling Schema5K ~ 10K tokensRAG 检索返回的业务知识5K ~ 20K tokens对话历史很容易膨胀到 50K tokens中间推理步骤每轮工具调用都要新产生几百到几千 tokens当请求内容超过模型上限时就会遇到类似下面这种错误api error: 400 this models maximum context length is 1048576 tokens. however, your messages resulted in 1052000 tokens这类报错说明请求体已经超出模型上下文窗口限制。解决办法不是无限扩容而是做上下文治理。3.2 记忆组织三原则我总结企业级 Agent 记忆治理的三个原则原则一能存外部就不塞提示词。记忆的核心不是把历史数据放进 Context而是把历史数据组织好需要时再检索回来。数据库、向量库、键值存储都比模型上下文窗口性价比高。原则二按需注入而不是全量注入。每次模型调用前要经过一层“记忆筛选”。根据当前用户问题只从记忆库中取最相关的若干条控制注入量。原则三分层存储不同记忆不同待遇。用户基础档案姓名、公司、偏好 → 存在关系型数据库或键值库直接读取。历史对话关键信息 → 使用 LLM 抽取摘要存入摘要存储。历史对话细节 → 使用向量化切块存入向量库按语义检索。当前任务的中间状态 → 存入 LangGraph State 或运行时缓存。3.3 三类常见的上下文压缩策略当对话变长时有三种常用策略可以控制 Context 膨胀策略一滑动窗口Sliding Window只保留最近 N 轮对话更早的内容直接丢弃。适合对历史细节不敏感的场景实现简单但会丢失早期关键信息。策略二摘要压缩Summary Compression用 LLM 对早期对话生成摘要将原始对话替换为摘要保留在上下文中。例如前 20 轮对话压缩成一段 300 字的摘要再连同最近 5 轮原始对话一起提供给模型。策略三记忆检索Memory Retrieval将历史信息存储到外部记忆库对话开始前根据用户输入进行检索只把命中的内容注入上下文。这是 Long-term Memory 的基础形态后面实战部分重点演示。4. LangChain 记忆组件的落地用法LangChain 的记忆生态比较丰富下面介绍几种最常见的记忆组件并说明它们的适用场景和实现思路。4.1 ConversationBufferMemory最简单的短期记忆这个组件会把对话历史全部保存在内存中直接拼接到下一次请求里。# 文件路径examples/buffer_memory_demo.py from langchain.memory import ConversationBufferMemory from langchain_openai import ChatOpenAI # 初始化模型 llm ChatOpenAI( modelgpt-4o-mini, temperature0, ) # 初始化记忆组件 memory ConversationBufferMemory( memory_keychat_history, return_messagesTrue, ) # 模拟两轮对话 memory.chat_memory.add_user_message(我叫李明负责公司的财务系统) memory.chat_memory.add_ai_message(你好李明已记录。请问今天需要处理什么财务问题) memory.chat_memory.add_user_message(我们公司需要配置电子发票) memory.chat_memory.add_ai_message(好的电子发票配置需要提供企业税号请查一下你的纳税人识别号。) # 获取记忆中的完整对话历史 chat_history memory.load_memory_variables({}) print(chat_history[chat_history])适用场景短期会话、原型验证、内部工具。缺点也很明显如果对话轮次很多这种组件的记忆内容会持续膨胀很快触发模型 Context 窗口上限。生产环境不推荐直接使用完整缓存。4.2 ConversationSummaryMemory摘要式记忆为了缓解膨胀问题可以用 LLM 对历史对话生成摘要。这样对话历史再长也能被压缩为有限长度的摘要文本。# 文件路径examples/summary_memory_demo.py from langchain.memory import ConversationSummaryMemory from langchain_openai import ChatOpenAI llm ChatOpenAI( modelgpt-4o-mini, temperature0, ) memory ConversationSummaryMemory( llmllm, memory_keychat_history_summary, return_messagesTrue, ) # 输入多轮对话后memory 内部会生成并更新摘要 memory.chat_memory.add_user_message(我是一家电商公司的运营最近要接入自动退款功能) memory.chat_memory.add_ai_message(自动退款功能需要先配置退款策略包括退款时限和触发条件。) memory.chat_memory.add_user_message(退款时限设成 7 天吧触发条件是用户申请且商品状态为已签收) memory.chat_memory.add_ai_message(已记录退款时限 7 天触发条件为已签收后用户申请。) # 查看摘要 summary memory.load_memory_variables({}) print(summary)注意摘要式记忆的问题在于摘要过程本身会丢失细节而且每轮对话都调用模型生成摘要会带来额外延迟和费用。适合对延迟不敏感、对话轮次适中、历史细节要求不高的场景。4.3 基于向量库的长期记忆当记忆跨会话、跨业务场景时摘要在全局视角下仍然不够。这时需要把历史信息向量化存到向量数据库在需要时按语义检索。# 文件路径examples/vector_memory_demo.py from langchain.memory import VectorStoreRetrieverMemory from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings from langchain.text_splitter import RecursiveCharacterTextSplitter # 模拟历史对话记录 history_texts [ 用户李明是华信科技财务部负责人公司纳税人识别号为 91110108MA01XXXXX。, 2025年3月李明咨询了电子发票配置流程要求使用增值税普通发票。, 李明明确表示发票接收邮箱需要更换为 financehuaxin.com。, ] # 文本切分 splitter RecursiveCharacterTextSplitter( chunk_size200, chunk_overlap20, ) docs splitter.create_documents(history_texts) # 初始化向量库 vectorstore Chroma.from_documents( docs, embeddingOpenAIEmbeddings(), ) # 创建检索器 retriever vectorstore.as_retriever( search_kwargs{k: 2} ) # 构建长期记忆组件 memory VectorStoreRetrieverMemory( retrieverretriever, memory_keylong_term_memory, input_keyinput, ) # 模拟新一轮用户提问 result memory.load_memory_variables( {input: 用户想更换发票接收邮箱他的公司税号是什么} ) print(result[long_term_memory])这个方案的核心思路是历史细节不直接放进 Context而是通过向量检索把相关的记忆找回来再注入模型输入。相比全量缓存Token 消耗显著下降而且可以跨会话使用。在企业级实现中向量库可以换成 Milvus、Elasticsearch、OpenSearch、PGVector 等生产级方案。Chroma 在示例中只是用于本地演示。5. LangGraph 状态流与记忆持久化LangGraph 与 LangChain 最大区别在于LangGraph 把 Agent 定义为一个有状态的图。每个节点的输入输出被记录到 State 中节点之间通过 State 传递数据。这个 State 机制天然就是“短期工作记忆”的载体。5.1 LangGraph 的核心概念先理解几个基础概念State状态整个图执行过程中共享的数据结构可以理解为 Agent 的工作记忆。Node节点一个处理单元接收 State返回对 State 的更新。Edge边定义节点之间的流转关系。Conditional Edge条件边根据 State 内容动态决定下一个节点。Checkpoint检查点每个节点执行后保存 State 快照的机制是持久化记忆的基础。5.2 在 State 中维护短期记忆下面实现一个带记忆的 LangGraph 工作流State 会记住用户与 Agent 之间的多轮对话同时保存用户档案信息。# 文件路径examples/langgraph_state_memory.py from typing import TypedDict, Annotated, List from langgraph.graph import StateGraph, START, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage # 定义 State 结构 class AgentState(TypedDict): # 对话历史 messages: Annotated[List, conversation history] # 用户档案长期记忆示例 user_profile: dict # 当前任务状态 current_task: str # 初始化模型 llm ChatOpenAI( modelgpt-4o-mini, temperature0, ) # 节点提取用户信息 def extract_user_info(state: AgentState): user_profile state.get(user_profile, {}) # 实际项目中这里可以使用 LLM 或规则从 messages 中抽取信息 # 演示模拟解析出用户身份 if 李明 in str(state[messages][-1].content): user_profile[name] 李明 user_profile[company] 华信科技 return {user_profile: user_profile} # 节点更新当前任务 def update_task(state: AgentState): last_message state[messages][-1].content if 发票 in last_message: task 发票配置 elif 退款 in last_message: task 退款处理 else: task 通用咨询 return {current_task: task} # 节点调用模型生成回复 def generate_reply(state: AgentState): messages state[messages] # 注入用户档案到系统提示词 profile state.get(user_profile, {}) system_info f当前用户档案{profile}。请基于对话历史回答问题。 response llm.invoke(messages) return {messages: state[messages] [response]} # 构建图 builder StateGraph(AgentState) builder.add_node(extract_user_info, extract_user_info) builder.add_node(update_task, update_task) builder.add_node(generate_reply, generate_reply) builder.add_edge(START, extract_user_info) builder.add_edge(extract_user_info, update_task) builder.add_edge(update_task, generate_reply) builder.add_edge(generate_reply, END) graph builder.compile() # 运行一段多轮对话 result graph.invoke({ messages: [ HumanMessage(content你好我是华信科技的李明需要咨询发票配置问题) ], user_profile: {}, current_task: , }) print(result[user_profile]) print(result[current_task]) print(result[messages][-1].content)这个示例体现了短期记忆的工程组织方式messages 维护对话轮次信息user_profile 跨轮次保留用户身份current_task 记录当前业务上下文。三个不同粒度的记忆被明确分离。5.3 使用 Checkpoint 实现会话级持久化图的状态默认保存在内存中进程重启后就会丢失。LangGraph 提供了 Checkpoint 机制可以把每个节点执行后的 State 保存到外部存储。# 文件路径examples/langgraph_checkpoint_demo.py from langgraph.checkpoint.sqlite import SqliteSaver # 使用 SQLite 保存状态快照 with SqliteSaver.from_conn_string(checkpoints.db) as checkpointer: graph builder.compile(checkpointercheckpointer) # 指定 thread_id相当于会话标识 config {configurable: {thread_id: session-001}} result graph.invoke( { messages: [ HumanMessage(content我的公司税号是 91110108MA01XXXXX帮我记住) ] }, configconfig, ) # 下一次请求即使是新的进程只要 thread_id 相同也能恢复历史状态 with SqliteSaver.from_conn_string(checkpoints.db) as checkpointer: graph builder.compile(checkpointercheckpointer) config {configurable: {thread_id: session-001}} result2 graph.invoke( { messages: [ HumanMessage(content我刚才说的税号是什么) ] }, configconfig, ) print(result2[messages][-1].content)Checkpoint 机制的核心价值在于LangGraph 把每一次节点执行后的 State 都保存下来即使用户会话中断、服务重启只要传入相同的 thread_idAgent 就能恢复到上次执行的状态继续工作。在生产环境SQLite 适合单机部署或测试环境分布式场景建议使用 Postgres CheckpointSaver、Redis 或其他支持并发的存储方案。5.4 条件路由与长期记忆触发LangGraph 还有一个重要能力条件路由。可以根据 State 的内容决定是否调用记忆检索节点。# 文件路径examples/langgraph_conditional_memory.py from typing import TypedDict from langgraph.graph import StateGraph, START, END from langchain_core.messages import HumanMessage class MemoryState(TypedDict): messages: list needs_memory_retrieval: bool retrieved_memory: str response: str def check_memory_need(state: MemoryState): 判断当前用户输入是否需要检索长期记忆 last_message state[messages][-1].content # 规则示例包含“之前”“上次”“历史”等关键词触发记忆检索 memory_keywords [之前, 上次, 历史, 我记得, 税号, 偏好] needs any(keyword in last_message for keyword in memory_keywords) return {needs_memory_retrieval: needs} def retrieve_memory(state: MemoryState): 从长期记忆库检索内容 # 实际项目中这里会读取向量库、数据库等外部存储 # 这里仅做演示 return {retrieved_memory: 用户李明华信科技税号 91110108MA01XXXXX偏好电子发票} def normal_reply(state: MemoryState): 无需检索时的普通回复 return {response: 这是普通回复不涉及长期记忆} def memory_reply(state: MemoryState): 结合检索结果回复 return {response: f根据历史记忆{state[retrieved_memory]}} builder StateGraph(MemoryState) builder.add_node(check_memory_need, check_memory_need) builder.add_node(retrieve_memory, retrieve_memory) builder.add_node(normal_reply, normal_reply) builder.add_node(memory_reply, memory_reply) builder.add_edge(START, check_memory_need) # 条件路由 builder.add_conditional_edges( check_memory_need, lambda state: retrieve_memory if state[needs_memory_retrieval] else normal_reply, {retrieve_memory: retrieve_memory, normal_reply: normal_reply}, ) builder.add_edge(retrieve_memory, memory_reply) builder.add_edge(normal_reply, END) builder.add_edge(memory_reply, END) graph builder.compile() # 测试 result graph.invoke({ messages: [HumanMessage(content我上次提到的税号是多少)], needs_memory_retrieval: False, retrieved_memory: , response: , }) print(result[response])这里的核心思路是记忆检索不是每轮都执行而是通过条件判断按需触发节省成本也降低噪音干扰。实际企业应用中触发条件可以更丰富关键词匹配、分类模型判断、用户画像标签、业务规则等。6. 完整实战企业级 Long-term Memory 记忆系统前面几节分别演示了 LangChain 的记忆组件和 LangGraph 的状态持久化。现在把它们整合成一个更完整的企业级记忆系统示例。这个系统会包含对话接口接收用户消息短期记忆LangGraph State Checkpoint长期记忆写入从对话中抽取关键信息存入向量库长期记忆检索根据当前问题注入相关历史信息模型回复结合短期与长期记忆生成回答6.1 项目结构agent-memory-system/ ├── app.py # 主程序对话接口 ├── memory/ │ ├── __init__.py │ ├── long_term_store.py # 向量库长期记忆 │ ├── profile_store.py # 用户档案存储 │ └── extractor.py # 关键信息抽取 ├── agent/ │ ├── __init__.py │ ├── graph.py # LangGraph 工作流 │ └── state.py # State 定义 ├── checkpoints/ # Checkpoint 持久化目录 ├── .env # 环境变量 └── requirements.txt # 依赖6.2 定义 State 结构# 文件路径agent/state.py from typing import TypedDict, List from langchain_core.messages import BaseMessage class AgentMemoryState(TypedDict): # 当前会话的短期记忆 messages: List[BaseMessage] # 用户唯一标识 user_id: str # 用户长期档案 user_profile: dict # 检索到的长期记忆 long_term_memory: str # 当前业务任务类型 task_type: str # 最终回复 response: str6.3 实现长期记忆存储# 文件路径memory/long_term_store.py import os from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import Chroma from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.docstore.document import Document class LongTermMemoryStore: 基于向量库的长期记忆存储按用户隔离 def __init__(self, persist_directory./data/vector_store): self.persist_directory persist_directory os.makedirs(persist_directory, exist_okTrue) self.embeddings OpenAIEmbeddings() self.vectorstore Chroma( persist_directorypersist_directory, embedding_functionself.embeddings, collection_nameagent_long_term_memory, ) self.splitter RecursiveCharacterTextSplitter( chunk_size300, chunk_overlap30, ) def add_memory(self, user_id: str, content: str, metadata: dict None): 写入一条长期记忆 Args: user_id: 用户唯一标识 content: 记忆内容文本 metadata: 附加元数据例如时间、业务类型 full_content f[用户{user_id}] {content} docs self.splitter.create_documents( [full_content], metadatas[{user_id: user_id, **(metadata or {})}], ) self.vectorstore.add_documents(docs) def search_memory(self, user_id: str, query: str, k: int 3) - str: 检索用户的历史记忆 Args: user_id: 用户唯一标识 query: 查询文本 k: 返回的记忆条数 Returns: 拼接后的记忆文本 results self.vectorstore.similarity_search( query, kk, filter{user_id: user_id}, ) if not results: return # 拼接检索结果同时保留来源信息 memory_parts [] for i, doc in enumerate(results, 1): source doc.metadata.get(source, 历史对话) memory_parts.append(f[记忆{i}来自{source}] {doc.page_content}) return \n.join(memory_parts)这个存储类做了两件关键事情第一所有记忆按 user_id 进行过滤避免用户 A 的历史记忆被检索出来注入到用户 B 的会话中。这既是数据隔离要求也是隐私安全底线。第二使用元数据过滤 语义相似度检索结合的方式。先按用户过滤再做向量检索既保证了数据范围正确又保证返回内容相关性。6.4 实现关键信息抽取# 文件路径memory/extractor.py from langchain_openai import ChatOpenAI from langchain_core.output_parsers import JsonOutputParser from langchain_core.prompts import PromptTemplate class MemoryExtractor: 从对话中抽取需要长期保存的关键信息 def __init__(self): self.llm ChatOpenAI( modelgpt-4o-mini, temperature0, ) self.parser JsonOutputParser() def extract(self, user_message: str, ai_message: str) - dict: 从一轮对话中抽取关键信息 Returns: { should_save: bool, memory_content: str, task_type: str, metadata: {} } prompt PromptTemplate( template你是一个对话记忆抽取器。请分析下面这轮对话判断是否有值得长期保存的关键信息。 用户消息{user_message} 助手回复{ai_message} 需要抽取的信息类型 1. 用户身份信息姓名、公司、职位、联系方式等 2. 用户偏好业务偏好、联系方式偏好、沟通风格等 3. 业务事实合同信息、配置信息、项目信息等 4. 事件记录用户曾经操作过什么、咨询过什么、解决过什么问题 如果没有任何值得保存的信息should_save 设为 false。 输出 JSON 格式 {format_instructions} , partial_variables{format_instructions: self.parser.get_format_instructions()}, ) chain prompt | self.llm | self.parser result chain.invoke({ user_message: user_message, ai_message: ai_message, }) return result关键信息抽取是企业级记忆系统与玩具级记忆系统的重要分水岭。直接把完整对话历史塞进向量库会产生大量冗余信息通过抽取器提炼出“值得长期记住”的结论性信息再写入长期记忆库准确率和存储效率都会提升。6.5 组装 LangGraph 工作流# 文件路径agent/graph.py from typing import TypedDict from langgraph.graph import StateGraph, START, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage from agent.state import AgentMemoryState from memory.long_term_store import LongTermMemoryStore from memory.extractor import MemoryExtractor class MemoryAgentGraph: 整合了长期记忆与短期记忆的 Agent 图 def __init__(self, llm, vector_store: LongTermMemoryStore): self.llm llm self.vector_store vector_store self.extractor MemoryExtractor() self.graph self._build_graph() def _build_graph(self): builder StateGraph(AgentMemoryState) builder.add_node(check_memory_need, self._check_memory_need) builder.add_node(retrieve_long_term_memory, self._retrieve_long_term_memory) builder.add_node(generate_reply, self._generate_reply) builder.add_node(save_long_term_memory, self._save_long_term_memory) builder.add_edge(START, check_memory_need) builder.add_conditional_edges( check_memory_need, lambda state: retrieve_long_term_memory if state[long_term_memory] and state[task_type] ! general else generate_reply, {retrieve_long_term_memory: retrieve_long_term_memory, generate_reply: generate_reply}, ) builder.add_edge(retrieve_long_term_memory, generate_reply) builder.add_edge(generate_reply, save_long_term_memory) builder.add_edge(save_long_term_memory, END) return builder.compile() def _check_memory_need(self, state: AgentMemoryState): 判断是否需要检索长期记忆 # 第一轮不检索历史 if len(state[messages]) 1: return {task_type: first_turn} last_message state[messages][-1].content # 业务规则包含历史关键词则触发 keywords [之前, 上次, 历史, 我记得, 我的, 税号, 偏好] task_type memory_retrieval if any(k in last_message for k in keywords) else general return {task_type: task_type} def _retrieve_long_term_memory(self, state: AgentMemoryState): 检索长期记忆 user_id state[user_id] last_query state[messages][-1].content memory_text self.vector_store.search_memory(user_id, last_query) return {long_term_memory: memory_text} def _generate_reply(self, state: AgentMemoryState): 生成回复传入短期记忆与长期记忆 messages state[messages] # 构建系统提示词 system_prompt 你是一个企业级 AI 助手请根据对话历史和长期记忆回答用户问题。 if state.get(user_profile): system_prompt f\n用户档案{state[user_profile]} if state.get(long_term_memory): system_prompt f\n检索到的历史记忆\n{state[long_term_memory]} # 组装模型输入 model_messages [{role: system, content: system_prompt}] for msg in messages: if isinstance(msg, HumanMessage): model_messages.append({role: user, content: msg.content}) elif isinstance(msg, AIMessage): model_messages.append({role: assistant, content: msg.content}) response self.llm.invoke(model_messages) return {response: response.content} def _save_long_term_memory(self, state: AgentMemoryState): 从当前对话抽取关键信息保存到长期记忆库 if len(state[messages]) 2: return {} user_msg state[messages][-2].content ai_msg state[messages][-1].content extracted self.extractor.extract(user_msg, ai_msg) if extracted.get(should_save): self.vector_store.add_memory( user_idstate[user_id], contentextracted.get(memory_content, ), metadata{ task_type: extracted.get(task_type, ), source: conversation, }, ) return {} def invoke(self, user_id: str, user_message: str, profile: dict None): 与 Agent 对话 new_message HumanMessage(contentuser_message) # 构造初始 State state { messages: [new_message], user_id: user_id, user_profile: profile or {}, long_term_memory: , task_type: , response: , } result self.graph.invoke(state) return result[response]6.6 运行与验证# 文件路径app.py import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI from agent.graph import MemoryAgentGraph from memory.long_term_store import LongTermMemoryStore load_dotenv() # 初始化组件 llm ChatOpenAI( modelos.getenv(OPENAI_MODEL, gpt-4o-mini), temperature0, ) memory_store LongTermMemoryStore() agent MemoryAgentGraph(llmllm, vector_storememory_store) # 第一轮对话用户提供关键信息 response1 agent.invoke( user_iduser-001, user_message我是华信科技的李明我们公司税号是 91110108MA01XXXXX发票偏好增值税普通发票, profile{name: 李明, company: 华信科技}, ) print(f第一轮回复{response1}) # 第二轮对话跨会话提问检索历史记忆 response2 agent.invoke( user_iduser-001, user_message我之前说过我们公司的发票偏好是什么, ) print(f第二轮回复{response2})预期行为第一轮对话后抽取器从对话中提取“公司税号”“发票偏好”等关键信息写入向量库。第二轮对话触发记忆检索条件从向量库中检索到历史记录注入模型输入。模型基于检索到的记忆回答用户问题。要验证抽取是否生效可以查看向量库持久化目录或者直接打印search_memory的返回值。7. 常见问题与排查思路在 Agent 记忆系统的开发过程中下面几个问题出现频率很高单独列出来说明。问题现象常见原因解决思路上下文超限报错对话历史或工具返回内容无限制膨胀增加摘要压缩、滑动窗口、记忆检索策略长期记忆检索结果不相关向量库写入内容太碎、检索时未按用户过滤优化切块策略抽取关键信息再写入增加 user_id 过滤跨会话恢复了状态但模型回答仍然像“失忆”Checkpoint 恢复的是中间计算状态不是模型输入检查是否把历史 messages 重新组装到模型输入中记忆重复写入每轮对话都执行抽取并保存增加记忆去重机制例如按内容哈希或向量相似度判断用户 A 看到用户 B 的记忆检索时没有按 user_id 过滤所有检索操作必须携带并强制校验 user_id摘要记忆丢失细节摘要压缩丢失具体数值、名称对关键实体税号、手机号、日期单独结构化存储检索延迟高向量库数据量大且没有索引优化使用规模化向量库启用 HNSW 等索引按用户分片再补充两个排查建议第一排查记忆问题时先确认“模型是否真的看到了记忆”。很多时候不是模型记不住而是你压根没有把记忆内容注入到模型输入里。建议在调试阶段把最终发送给模型的 messages 完整打印出来确认长期记忆是否拼接成功。第二日志里要能看到记忆链路。建议对每个请求记录是否触发检索、检索到几条记忆、注入记忆的 token 数、抽取器是否返回 should_save。这样线上问题才能快速定位。8. 最佳实践与工程治理8.1 记忆数据模型设计企业级记忆系统不建议把所有信息混在一个向量库里。我建议至少规划三个存储区域存储区域内容存储方案访问方式用户档案区身份、偏好、标签关系型数据库 / Redis主键精确读取事件记忆区历史操作、对话结论向量库 元数据过滤语义检索工作状态区当前任务进度、中间状态LangGraph Checkpoint按 thread_id 读取用户档案适合精确读取因为用户姓名、税号、偏好这类信息不适合语义检索直接关系型查询更准确。事件记忆适合语义检索因为用户可能用不同的说法表达同一件事。工作状态区则完全由 LangGraph 托管。8.2 记忆更新的安全策略长期记忆一旦写错会影响后续所有对话。所以写入策略要谨慎高置信度信息用户明确说“我的税号是 X”可以自动写入。低置信度信息模型推测的偏好需要二次确认或者设计为“暂存待确认”状态。用户主动要求删除记忆时必须有删除通道并且删除操作要记录审计日志。记忆更新需要保留版本方便回滚。8.3 成本与性能治理记忆系统会带来额外的模型调用和存储开销。工程治理上要注意几点信息抽取环节可以复用原有 LLM 调用结果不要每次都新增一次完整对话调用。用轻量模型做抽取和分类用能力强的模型做最终回复。检索结果要设置数量上限并且对检索相关度设置阈值过滤低置信度结果。记忆摘要可以异步执行避免阻塞主对话链路。8.4 权限与隐私隔离企业级应用必须考虑多租户隔离。即使是同一家公司的不同部门也不能互相读取记忆。实现上至少做到三层数据隔离所有记忆写入时附带 tenant_id 和 user_id。接口隔离检索接口强制从认证上下文获取用户身份禁止客户端传入 user_id 覆盖。审计隔离所有记忆读写操作记录操作者、时间、内容摘要。8.5 观测体系建设带上记忆系统后Agent 的行为链路变长必须建立完整的观测指标指标意义记忆写入成功率判断抽取和存储是否正常记忆检索命中率判断检索策略是否有效注入记忆 token 占比判断上下文预算是否合理记忆导致回复提升率通过 A/B 对比衡量记忆价值记忆误用率模型引用了错误的记忆导致回复错误线上环境建议对每条记忆内容生成来源 trace_id方便回溯是哪一轮对话产生的记忆。9. 总结与后续学习方向Agent 记忆不是单一组件而是一套分层架构。从短期工作记忆到长期语义记忆再到跨会话状态持久化每一层都有不同的技术选型和工程治理要求。我建议按这个顺序继续深入先用 LangChain 的 Memory 组件快速验证记忆基本流程理解 Token 膨胀问题。再用 LangGraph 的 State 和 Checkpoint 管理短期工作记忆掌握会话级别状态恢复。最后构建长期记忆系统把关键信息抽取、向量检索、用户隔离结合起来。本文示例使用的 Chroma 和 SQLite 适合本地演示生产环境需要根据数据规模切换到更成熟的存储方案。想继续深入的话重点研究 LangGraph 的 StateGraph 各节点生命周期管理、Common Retriever 的混合检索策略向量检索 关键词检索 重排序、以及针对超长上下文的自动压缩机制。记忆系统做得好不好对用户体验和 Agent 能力上限影响极大。希望这篇文章能帮你建立一条清晰的落地方案少踩一些重复的坑。如果哪一步卡住了欢迎对照代码逐段运行调试。为了后面查找方便可以先收藏这篇文章备用。
企业级Agent记忆系统架构实战：从Context治理到Long-term Memory落地
各位做 Agent 应用开发的同学应该都有过这样的体验单轮对话效果不错一进入多轮、跨会话、跨场景的复杂任务模型就开始“失忆”。明明用户之前已经提供过偏好、项目背景、历史结论但 Agent 换一个会话之后所有上下文清零只能重新交代。更麻烦的是模型上下文窗口是有限的哪怕是 128K、1M 的窗口也扛不住长期积累的对话历史、工具调用记录、中间推理过程无限写入。本文围绕企业级 Agent 记忆系统的架构设计与工程治理展开讲解怎么从最基础的 Context 管理一步步升级到 Long-term Memory并结合 LangChain、LangGraph 以及 DeepAgent 三种不同抽象层次框架给出可落地的实现思路和代码案例。内容会比较长我会尽量把概念讲清楚代码给完整。适合两类读者正在使用 LangChain、LangGraph 做 Agent 开发想解决上下文丢失、Token 超限、多轮记忆力弱等问题的开发者负责企业级 AI 应用架构设计需要为团队制定 Agent 记忆治理方案的技术负责人。读完本文你会掌握记忆分类模型、Context 窗口管理策略、LangChain 记忆组件用法、LangGraph 状态流持久化方案以及企业级记忆治理的工程规范。1. 背景与核心概念为什么 Agent 必须解决“记忆”问题1.1 没有记忆的 Agent 是什么状态先看一个很常见的业务场景。用户对客服 Agent 说“我上周咨询过企业发票配置问题当时确认了使用电子发票税号也提交过了现在我想改一下接收邮箱。”如果这个 Agent 没有任何跨会话记忆它看到这条消息时对“上周咨询过”“确认了电子发票”“税号提交过”这些信息完全没有概念。它只能把这句话当成一个全新问题重新让用户提供企业名称、纳税人识别号、发票类型、历史处理单号……用户体验非常差。这正是 Agent 记忆系统要解决的核心问题让 Agent 在单次对话、多次对话甚至跨业务场景中保留并使用历史信息。1.2 从 Context 到 Long-term Memory 的本质在聊记忆之前必须先理解 Context 是什么。Context 是模型一次调用中能看到的全部文本内容。它通常由三部分组成系统提示词System Prompt对话历史Conversation History当前用户输入User Query模型的能力上限由训练决定但每轮请求能“记住”多少信息完全取决于 Context 窗口大小和内容组织方式。Long-term Memory 则是一种超越上下文窗口限制的能力。它的核心思路是不把所有信息都塞进一次模型调用而是按需检索、按需注入。Agent 需要记忆时从外部存储中找到最相关的内容只把这一部分加入 Context而不是把历史数据全部堆给模型。1.3 记忆的分类不是所有信息都该用同一种方式保存做 Agent 记忆架构首先要建立分类意识。记忆类型保存内容典型例子实现方式短期工作记忆当前对话中的中间状态用户刚才输入的商品名称、当前正在执行的步骤对话上下文变量、状态对象长期语义记忆用户属性、偏好、业务规则用户所在城市、发票类型偏好、公司规模数据库、键值存储、向量库长期情景记忆过去发生过的具体事件用户上周咨询过发票配置问题向量检索 摘要存储程序记忆Agent 如何完成任务的流程知识发票配置流程、售后处理标准作业程序工作流定义、技能库、提示词模板不同记忆类型对存储介质、检索方式、更新策略的要求完全不同。很多团队一开始就把所有历史消息塞进向量库结果就是检索噪音大、Token 消耗高、关键信息反而找不到。后面的章节会详细展开每种记忆在企业级实现中的具体方案。1.4 三个框架的定位差异LangChain、LangGraph、DeepAgent 三者经常一起出现但它们的抽象层次不同。LangChain 是一个全能型开发框架提供了大量模型接入、Prompt 管理、记忆组件、工具调用、RAG 链路的模块。它的记忆组件比较齐全适合快速搭建但流程控制能力相对弱。LangGraph 主打有状态、可编排的图结构 Agent 流程。核心概念是 State、Node、Edge、Checkpoint特别适合处理需要条件分支、循环、并行、人工介入的复杂任务。它的 Checkpoint 机制天然适合做记忆持久化。DeepAgent 偏向更深层的 Agent 基础能力建设关注 Agent 的技能编排、上下文工程、复杂推理过程治理。在企业级架构中可以将 DeepAgent 视为一个更上层的平台层概念负责把记忆、工具、模型、流程治理统一收口。企业落地时不同团队可能把这个“平台层”命名为 DeepAgent、Agent Gateway 或 Agent Runtime本质都是在 LangGraph 这类图编排之上构建统一的运行治理能力。2. 环境准备与版本说明为了避免后续代码示例出现环境差异问题先说明本文示例的运行环境。依赖建议版本说明Python3.10LangChain 与 LangGraph 均对 3.10 支持良好langchain0.2.x / 0.3.x不同版本 API 有差异按实际安装版本调整langchain-openai0.1.xOpenAI 兼容接口langgraph0.2.xLangGraph 主包langgraph-checkpoint-sqlite0.1.xSQLite 持久化检查点chromadb0.4.x向量存储示例用python-dotenv1.0.x环境变量管理安装依赖pip install langchain langchain-openai langgraph langgraph-checkpoint-sqlite chromadb python-dotenv如果你使用的是国内模型或开源模型只要接口兼容 OpenAI 格式代码基本通用。在初始化模型时替换 base_url、api_key、model 名称即可。需要注意LangChain 和 LangGraph 更新速度很快部分类名和参数名会变化。本文代码以常见版本为示例重点演示设计思路。如果你安装的版本 API 有差异优先查阅官方文档和当前版本的迁移说明。3. 核心原理Context 管理与记忆组织3.1 Context 窗口的硬约束先直观感受一下 Context 窗口限制。假设一个模型的上下文窗口为 128K tokens。看起来很大但实际分配下来非常紧张系统提示词2K tokens工具定义Function Calling Schema5K ~ 10K tokensRAG 检索返回的业务知识5K ~ 20K tokens对话历史很容易膨胀到 50K tokens中间推理步骤每轮工具调用都要新产生几百到几千 tokens当请求内容超过模型上限时就会遇到类似下面这种错误api error: 400 this models maximum context length is 1048576 tokens. however, your messages resulted in 1052000 tokens这类报错说明请求体已经超出模型上下文窗口限制。解决办法不是无限扩容而是做上下文治理。3.2 记忆组织三原则我总结企业级 Agent 记忆治理的三个原则原则一能存外部就不塞提示词。记忆的核心不是把历史数据放进 Context而是把历史数据组织好需要时再检索回来。数据库、向量库、键值存储都比模型上下文窗口性价比高。原则二按需注入而不是全量注入。每次模型调用前要经过一层“记忆筛选”。根据当前用户问题只从记忆库中取最相关的若干条控制注入量。原则三分层存储不同记忆不同待遇。用户基础档案姓名、公司、偏好 → 存在关系型数据库或键值库直接读取。历史对话关键信息 → 使用 LLM 抽取摘要存入摘要存储。历史对话细节 → 使用向量化切块存入向量库按语义检索。当前任务的中间状态 → 存入 LangGraph State 或运行时缓存。3.3 三类常见的上下文压缩策略当对话变长时有三种常用策略可以控制 Context 膨胀策略一滑动窗口Sliding Window只保留最近 N 轮对话更早的内容直接丢弃。适合对历史细节不敏感的场景实现简单但会丢失早期关键信息。策略二摘要压缩Summary Compression用 LLM 对早期对话生成摘要将原始对话替换为摘要保留在上下文中。例如前 20 轮对话压缩成一段 300 字的摘要再连同最近 5 轮原始对话一起提供给模型。策略三记忆检索Memory Retrieval将历史信息存储到外部记忆库对话开始前根据用户输入进行检索只把命中的内容注入上下文。这是 Long-term Memory 的基础形态后面实战部分重点演示。4. LangChain 记忆组件的落地用法LangChain 的记忆生态比较丰富下面介绍几种最常见的记忆组件并说明它们的适用场景和实现思路。4.1 ConversationBufferMemory最简单的短期记忆这个组件会把对话历史全部保存在内存中直接拼接到下一次请求里。# 文件路径examples/buffer_memory_demo.py from langchain.memory import ConversationBufferMemory from langchain_openai import ChatOpenAI # 初始化模型 llm ChatOpenAI( modelgpt-4o-mini, temperature0, ) # 初始化记忆组件 memory ConversationBufferMemory( memory_keychat_history, return_messagesTrue, ) # 模拟两轮对话 memory.chat_memory.add_user_message(我叫李明负责公司的财务系统) memory.chat_memory.add_ai_message(你好李明已记录。请问今天需要处理什么财务问题) memory.chat_memory.add_user_message(我们公司需要配置电子发票) memory.chat_memory.add_ai_message(好的电子发票配置需要提供企业税号请查一下你的纳税人识别号。) # 获取记忆中的完整对话历史 chat_history memory.load_memory_variables({}) print(chat_history[chat_history])适用场景短期会话、原型验证、内部工具。缺点也很明显如果对话轮次很多这种组件的记忆内容会持续膨胀很快触发模型 Context 窗口上限。生产环境不推荐直接使用完整缓存。4.2 ConversationSummaryMemory摘要式记忆为了缓解膨胀问题可以用 LLM 对历史对话生成摘要。这样对话历史再长也能被压缩为有限长度的摘要文本。# 文件路径examples/summary_memory_demo.py from langchain.memory import ConversationSummaryMemory from langchain_openai import ChatOpenAI llm ChatOpenAI( modelgpt-4o-mini, temperature0, ) memory ConversationSummaryMemory( llmllm, memory_keychat_history_summary, return_messagesTrue, ) # 输入多轮对话后memory 内部会生成并更新摘要 memory.chat_memory.add_user_message(我是一家电商公司的运营最近要接入自动退款功能) memory.chat_memory.add_ai_message(自动退款功能需要先配置退款策略包括退款时限和触发条件。) memory.chat_memory.add_user_message(退款时限设成 7 天吧触发条件是用户申请且商品状态为已签收) memory.chat_memory.add_ai_message(已记录退款时限 7 天触发条件为已签收后用户申请。) # 查看摘要 summary memory.load_memory_variables({}) print(summary)注意摘要式记忆的问题在于摘要过程本身会丢失细节而且每轮对话都调用模型生成摘要会带来额外延迟和费用。适合对延迟不敏感、对话轮次适中、历史细节要求不高的场景。4.3 基于向量库的长期记忆当记忆跨会话、跨业务场景时摘要在全局视角下仍然不够。这时需要把历史信息向量化存到向量数据库在需要时按语义检索。# 文件路径examples/vector_memory_demo.py from langchain.memory import VectorStoreRetrieverMemory from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings from langchain.text_splitter import RecursiveCharacterTextSplitter # 模拟历史对话记录 history_texts [ 用户李明是华信科技财务部负责人公司纳税人识别号为 91110108MA01XXXXX。, 2025年3月李明咨询了电子发票配置流程要求使用增值税普通发票。, 李明明确表示发票接收邮箱需要更换为 financehuaxin.com。, ] # 文本切分 splitter RecursiveCharacterTextSplitter( chunk_size200, chunk_overlap20, ) docs splitter.create_documents(history_texts) # 初始化向量库 vectorstore Chroma.from_documents( docs, embeddingOpenAIEmbeddings(), ) # 创建检索器 retriever vectorstore.as_retriever( search_kwargs{k: 2} ) # 构建长期记忆组件 memory VectorStoreRetrieverMemory( retrieverretriever, memory_keylong_term_memory, input_keyinput, ) # 模拟新一轮用户提问 result memory.load_memory_variables( {input: 用户想更换发票接收邮箱他的公司税号是什么} ) print(result[long_term_memory])这个方案的核心思路是历史细节不直接放进 Context而是通过向量检索把相关的记忆找回来再注入模型输入。相比全量缓存Token 消耗显著下降而且可以跨会话使用。在企业级实现中向量库可以换成 Milvus、Elasticsearch、OpenSearch、PGVector 等生产级方案。Chroma 在示例中只是用于本地演示。5. LangGraph 状态流与记忆持久化LangGraph 与 LangChain 最大区别在于LangGraph 把 Agent 定义为一个有状态的图。每个节点的输入输出被记录到 State 中节点之间通过 State 传递数据。这个 State 机制天然就是“短期工作记忆”的载体。5.1 LangGraph 的核心概念先理解几个基础概念State状态整个图执行过程中共享的数据结构可以理解为 Agent 的工作记忆。Node节点一个处理单元接收 State返回对 State 的更新。Edge边定义节点之间的流转关系。Conditional Edge条件边根据 State 内容动态决定下一个节点。Checkpoint检查点每个节点执行后保存 State 快照的机制是持久化记忆的基础。5.2 在 State 中维护短期记忆下面实现一个带记忆的 LangGraph 工作流State 会记住用户与 Agent 之间的多轮对话同时保存用户档案信息。# 文件路径examples/langgraph_state_memory.py from typing import TypedDict, Annotated, List from langgraph.graph import StateGraph, START, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage # 定义 State 结构 class AgentState(TypedDict): # 对话历史 messages: Annotated[List, conversation history] # 用户档案长期记忆示例 user_profile: dict # 当前任务状态 current_task: str # 初始化模型 llm ChatOpenAI( modelgpt-4o-mini, temperature0, ) # 节点提取用户信息 def extract_user_info(state: AgentState): user_profile state.get(user_profile, {}) # 实际项目中这里可以使用 LLM 或规则从 messages 中抽取信息 # 演示模拟解析出用户身份 if 李明 in str(state[messages][-1].content): user_profile[name] 李明 user_profile[company] 华信科技 return {user_profile: user_profile} # 节点更新当前任务 def update_task(state: AgentState): last_message state[messages][-1].content if 发票 in last_message: task 发票配置 elif 退款 in last_message: task 退款处理 else: task 通用咨询 return {current_task: task} # 节点调用模型生成回复 def generate_reply(state: AgentState): messages state[messages] # 注入用户档案到系统提示词 profile state.get(user_profile, {}) system_info f当前用户档案{profile}。请基于对话历史回答问题。 response llm.invoke(messages) return {messages: state[messages] [response]} # 构建图 builder StateGraph(AgentState) builder.add_node(extract_user_info, extract_user_info) builder.add_node(update_task, update_task) builder.add_node(generate_reply, generate_reply) builder.add_edge(START, extract_user_info) builder.add_edge(extract_user_info, update_task) builder.add_edge(update_task, generate_reply) builder.add_edge(generate_reply, END) graph builder.compile() # 运行一段多轮对话 result graph.invoke({ messages: [ HumanMessage(content你好我是华信科技的李明需要咨询发票配置问题) ], user_profile: {}, current_task: , }) print(result[user_profile]) print(result[current_task]) print(result[messages][-1].content)这个示例体现了短期记忆的工程组织方式messages 维护对话轮次信息user_profile 跨轮次保留用户身份current_task 记录当前业务上下文。三个不同粒度的记忆被明确分离。5.3 使用 Checkpoint 实现会话级持久化图的状态默认保存在内存中进程重启后就会丢失。LangGraph 提供了 Checkpoint 机制可以把每个节点执行后的 State 保存到外部存储。# 文件路径examples/langgraph_checkpoint_demo.py from langgraph.checkpoint.sqlite import SqliteSaver # 使用 SQLite 保存状态快照 with SqliteSaver.from_conn_string(checkpoints.db) as checkpointer: graph builder.compile(checkpointercheckpointer) # 指定 thread_id相当于会话标识 config {configurable: {thread_id: session-001}} result graph.invoke( { messages: [ HumanMessage(content我的公司税号是 91110108MA01XXXXX帮我记住) ] }, configconfig, ) # 下一次请求即使是新的进程只要 thread_id 相同也能恢复历史状态 with SqliteSaver.from_conn_string(checkpoints.db) as checkpointer: graph builder.compile(checkpointercheckpointer) config {configurable: {thread_id: session-001}} result2 graph.invoke( { messages: [ HumanMessage(content我刚才说的税号是什么) ] }, configconfig, ) print(result2[messages][-1].content)Checkpoint 机制的核心价值在于LangGraph 把每一次节点执行后的 State 都保存下来即使用户会话中断、服务重启只要传入相同的 thread_idAgent 就能恢复到上次执行的状态继续工作。在生产环境SQLite 适合单机部署或测试环境分布式场景建议使用 Postgres CheckpointSaver、Redis 或其他支持并发的存储方案。5.4 条件路由与长期记忆触发LangGraph 还有一个重要能力条件路由。可以根据 State 的内容决定是否调用记忆检索节点。# 文件路径examples/langgraph_conditional_memory.py from typing import TypedDict from langgraph.graph import StateGraph, START, END from langchain_core.messages import HumanMessage class MemoryState(TypedDict): messages: list needs_memory_retrieval: bool retrieved_memory: str response: str def check_memory_need(state: MemoryState): 判断当前用户输入是否需要检索长期记忆 last_message state[messages][-1].content # 规则示例包含“之前”“上次”“历史”等关键词触发记忆检索 memory_keywords [之前, 上次, 历史, 我记得, 税号, 偏好] needs any(keyword in last_message for keyword in memory_keywords) return {needs_memory_retrieval: needs} def retrieve_memory(state: MemoryState): 从长期记忆库检索内容 # 实际项目中这里会读取向量库、数据库等外部存储 # 这里仅做演示 return {retrieved_memory: 用户李明华信科技税号 91110108MA01XXXXX偏好电子发票} def normal_reply(state: MemoryState): 无需检索时的普通回复 return {response: 这是普通回复不涉及长期记忆} def memory_reply(state: MemoryState): 结合检索结果回复 return {response: f根据历史记忆{state[retrieved_memory]}} builder StateGraph(MemoryState) builder.add_node(check_memory_need, check_memory_need) builder.add_node(retrieve_memory, retrieve_memory) builder.add_node(normal_reply, normal_reply) builder.add_node(memory_reply, memory_reply) builder.add_edge(START, check_memory_need) # 条件路由 builder.add_conditional_edges( check_memory_need, lambda state: retrieve_memory if state[needs_memory_retrieval] else normal_reply, {retrieve_memory: retrieve_memory, normal_reply: normal_reply}, ) builder.add_edge(retrieve_memory, memory_reply) builder.add_edge(normal_reply, END) builder.add_edge(memory_reply, END) graph builder.compile() # 测试 result graph.invoke({ messages: [HumanMessage(content我上次提到的税号是多少)], needs_memory_retrieval: False, retrieved_memory: , response: , }) print(result[response])这里的核心思路是记忆检索不是每轮都执行而是通过条件判断按需触发节省成本也降低噪音干扰。实际企业应用中触发条件可以更丰富关键词匹配、分类模型判断、用户画像标签、业务规则等。6. 完整实战企业级 Long-term Memory 记忆系统前面几节分别演示了 LangChain 的记忆组件和 LangGraph 的状态持久化。现在把它们整合成一个更完整的企业级记忆系统示例。这个系统会包含对话接口接收用户消息短期记忆LangGraph State Checkpoint长期记忆写入从对话中抽取关键信息存入向量库长期记忆检索根据当前问题注入相关历史信息模型回复结合短期与长期记忆生成回答6.1 项目结构agent-memory-system/ ├── app.py # 主程序对话接口 ├── memory/ │ ├── __init__.py │ ├── long_term_store.py # 向量库长期记忆 │ ├── profile_store.py # 用户档案存储 │ └── extractor.py # 关键信息抽取 ├── agent/ │ ├── __init__.py │ ├── graph.py # LangGraph 工作流 │ └── state.py # State 定义 ├── checkpoints/ # Checkpoint 持久化目录 ├── .env # 环境变量 └── requirements.txt # 依赖6.2 定义 State 结构# 文件路径agent/state.py from typing import TypedDict, List from langchain_core.messages import BaseMessage class AgentMemoryState(TypedDict): # 当前会话的短期记忆 messages: List[BaseMessage] # 用户唯一标识 user_id: str # 用户长期档案 user_profile: dict # 检索到的长期记忆 long_term_memory: str # 当前业务任务类型 task_type: str # 最终回复 response: str6.3 实现长期记忆存储# 文件路径memory/long_term_store.py import os from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import Chroma from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.docstore.document import Document class LongTermMemoryStore: 基于向量库的长期记忆存储按用户隔离 def __init__(self, persist_directory./data/vector_store): self.persist_directory persist_directory os.makedirs(persist_directory, exist_okTrue) self.embeddings OpenAIEmbeddings() self.vectorstore Chroma( persist_directorypersist_directory, embedding_functionself.embeddings, collection_nameagent_long_term_memory, ) self.splitter RecursiveCharacterTextSplitter( chunk_size300, chunk_overlap30, ) def add_memory(self, user_id: str, content: str, metadata: dict None): 写入一条长期记忆 Args: user_id: 用户唯一标识 content: 记忆内容文本 metadata: 附加元数据例如时间、业务类型 full_content f[用户{user_id}] {content} docs self.splitter.create_documents( [full_content], metadatas[{user_id: user_id, **(metadata or {})}], ) self.vectorstore.add_documents(docs) def search_memory(self, user_id: str, query: str, k: int 3) - str: 检索用户的历史记忆 Args: user_id: 用户唯一标识 query: 查询文本 k: 返回的记忆条数 Returns: 拼接后的记忆文本 results self.vectorstore.similarity_search( query, kk, filter{user_id: user_id}, ) if not results: return # 拼接检索结果同时保留来源信息 memory_parts [] for i, doc in enumerate(results, 1): source doc.metadata.get(source, 历史对话) memory_parts.append(f[记忆{i}来自{source}] {doc.page_content}) return \n.join(memory_parts)这个存储类做了两件关键事情第一所有记忆按 user_id 进行过滤避免用户 A 的历史记忆被检索出来注入到用户 B 的会话中。这既是数据隔离要求也是隐私安全底线。第二使用元数据过滤 语义相似度检索结合的方式。先按用户过滤再做向量检索既保证了数据范围正确又保证返回内容相关性。6.4 实现关键信息抽取# 文件路径memory/extractor.py from langchain_openai import ChatOpenAI from langchain_core.output_parsers import JsonOutputParser from langchain_core.prompts import PromptTemplate class MemoryExtractor: 从对话中抽取需要长期保存的关键信息 def __init__(self): self.llm ChatOpenAI( modelgpt-4o-mini, temperature0, ) self.parser JsonOutputParser() def extract(self, user_message: str, ai_message: str) - dict: 从一轮对话中抽取关键信息 Returns: { should_save: bool, memory_content: str, task_type: str, metadata: {} } prompt PromptTemplate( template你是一个对话记忆抽取器。请分析下面这轮对话判断是否有值得长期保存的关键信息。 用户消息{user_message} 助手回复{ai_message} 需要抽取的信息类型 1. 用户身份信息姓名、公司、职位、联系方式等 2. 用户偏好业务偏好、联系方式偏好、沟通风格等 3. 业务事实合同信息、配置信息、项目信息等 4. 事件记录用户曾经操作过什么、咨询过什么、解决过什么问题 如果没有任何值得保存的信息should_save 设为 false。 输出 JSON 格式 {format_instructions} , partial_variables{format_instructions: self.parser.get_format_instructions()}, ) chain prompt | self.llm | self.parser result chain.invoke({ user_message: user_message, ai_message: ai_message, }) return result关键信息抽取是企业级记忆系统与玩具级记忆系统的重要分水岭。直接把完整对话历史塞进向量库会产生大量冗余信息通过抽取器提炼出“值得长期记住”的结论性信息再写入长期记忆库准确率和存储效率都会提升。6.5 组装 LangGraph 工作流# 文件路径agent/graph.py from typing import TypedDict from langgraph.graph import StateGraph, START, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage from agent.state import AgentMemoryState from memory.long_term_store import LongTermMemoryStore from memory.extractor import MemoryExtractor class MemoryAgentGraph: 整合了长期记忆与短期记忆的 Agent 图 def __init__(self, llm, vector_store: LongTermMemoryStore): self.llm llm self.vector_store vector_store self.extractor MemoryExtractor() self.graph self._build_graph() def _build_graph(self): builder StateGraph(AgentMemoryState) builder.add_node(check_memory_need, self._check_memory_need) builder.add_node(retrieve_long_term_memory, self._retrieve_long_term_memory) builder.add_node(generate_reply, self._generate_reply) builder.add_node(save_long_term_memory, self._save_long_term_memory) builder.add_edge(START, check_memory_need) builder.add_conditional_edges( check_memory_need, lambda state: retrieve_long_term_memory if state[long_term_memory] and state[task_type] ! general else generate_reply, {retrieve_long_term_memory: retrieve_long_term_memory, generate_reply: generate_reply}, ) builder.add_edge(retrieve_long_term_memory, generate_reply) builder.add_edge(generate_reply, save_long_term_memory) builder.add_edge(save_long_term_memory, END) return builder.compile() def _check_memory_need(self, state: AgentMemoryState): 判断是否需要检索长期记忆 # 第一轮不检索历史 if len(state[messages]) 1: return {task_type: first_turn} last_message state[messages][-1].content # 业务规则包含历史关键词则触发 keywords [之前, 上次, 历史, 我记得, 我的, 税号, 偏好] task_type memory_retrieval if any(k in last_message for k in keywords) else general return {task_type: task_type} def _retrieve_long_term_memory(self, state: AgentMemoryState): 检索长期记忆 user_id state[user_id] last_query state[messages][-1].content memory_text self.vector_store.search_memory(user_id, last_query) return {long_term_memory: memory_text} def _generate_reply(self, state: AgentMemoryState): 生成回复传入短期记忆与长期记忆 messages state[messages] # 构建系统提示词 system_prompt 你是一个企业级 AI 助手请根据对话历史和长期记忆回答用户问题。 if state.get(user_profile): system_prompt f\n用户档案{state[user_profile]} if state.get(long_term_memory): system_prompt f\n检索到的历史记忆\n{state[long_term_memory]} # 组装模型输入 model_messages [{role: system, content: system_prompt}] for msg in messages: if isinstance(msg, HumanMessage): model_messages.append({role: user, content: msg.content}) elif isinstance(msg, AIMessage): model_messages.append({role: assistant, content: msg.content}) response self.llm.invoke(model_messages) return {response: response.content} def _save_long_term_memory(self, state: AgentMemoryState): 从当前对话抽取关键信息保存到长期记忆库 if len(state[messages]) 2: return {} user_msg state[messages][-2].content ai_msg state[messages][-1].content extracted self.extractor.extract(user_msg, ai_msg) if extracted.get(should_save): self.vector_store.add_memory( user_idstate[user_id], contentextracted.get(memory_content, ), metadata{ task_type: extracted.get(task_type, ), source: conversation, }, ) return {} def invoke(self, user_id: str, user_message: str, profile: dict None): 与 Agent 对话 new_message HumanMessage(contentuser_message) # 构造初始 State state { messages: [new_message], user_id: user_id, user_profile: profile or {}, long_term_memory: , task_type: , response: , } result self.graph.invoke(state) return result[response]6.6 运行与验证# 文件路径app.py import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI from agent.graph import MemoryAgentGraph from memory.long_term_store import LongTermMemoryStore load_dotenv() # 初始化组件 llm ChatOpenAI( modelos.getenv(OPENAI_MODEL, gpt-4o-mini), temperature0, ) memory_store LongTermMemoryStore() agent MemoryAgentGraph(llmllm, vector_storememory_store) # 第一轮对话用户提供关键信息 response1 agent.invoke( user_iduser-001, user_message我是华信科技的李明我们公司税号是 91110108MA01XXXXX发票偏好增值税普通发票, profile{name: 李明, company: 华信科技}, ) print(f第一轮回复{response1}) # 第二轮对话跨会话提问检索历史记忆 response2 agent.invoke( user_iduser-001, user_message我之前说过我们公司的发票偏好是什么, ) print(f第二轮回复{response2})预期行为第一轮对话后抽取器从对话中提取“公司税号”“发票偏好”等关键信息写入向量库。第二轮对话触发记忆检索条件从向量库中检索到历史记录注入模型输入。模型基于检索到的记忆回答用户问题。要验证抽取是否生效可以查看向量库持久化目录或者直接打印search_memory的返回值。7. 常见问题与排查思路在 Agent 记忆系统的开发过程中下面几个问题出现频率很高单独列出来说明。问题现象常见原因解决思路上下文超限报错对话历史或工具返回内容无限制膨胀增加摘要压缩、滑动窗口、记忆检索策略长期记忆检索结果不相关向量库写入内容太碎、检索时未按用户过滤优化切块策略抽取关键信息再写入增加 user_id 过滤跨会话恢复了状态但模型回答仍然像“失忆”Checkpoint 恢复的是中间计算状态不是模型输入检查是否把历史 messages 重新组装到模型输入中记忆重复写入每轮对话都执行抽取并保存增加记忆去重机制例如按内容哈希或向量相似度判断用户 A 看到用户 B 的记忆检索时没有按 user_id 过滤所有检索操作必须携带并强制校验 user_id摘要记忆丢失细节摘要压缩丢失具体数值、名称对关键实体税号、手机号、日期单独结构化存储检索延迟高向量库数据量大且没有索引优化使用规模化向量库启用 HNSW 等索引按用户分片再补充两个排查建议第一排查记忆问题时先确认“模型是否真的看到了记忆”。很多时候不是模型记不住而是你压根没有把记忆内容注入到模型输入里。建议在调试阶段把最终发送给模型的 messages 完整打印出来确认长期记忆是否拼接成功。第二日志里要能看到记忆链路。建议对每个请求记录是否触发检索、检索到几条记忆、注入记忆的 token 数、抽取器是否返回 should_save。这样线上问题才能快速定位。8. 最佳实践与工程治理8.1 记忆数据模型设计企业级记忆系统不建议把所有信息混在一个向量库里。我建议至少规划三个存储区域存储区域内容存储方案访问方式用户档案区身份、偏好、标签关系型数据库 / Redis主键精确读取事件记忆区历史操作、对话结论向量库 元数据过滤语义检索工作状态区当前任务进度、中间状态LangGraph Checkpoint按 thread_id 读取用户档案适合精确读取因为用户姓名、税号、偏好这类信息不适合语义检索直接关系型查询更准确。事件记忆适合语义检索因为用户可能用不同的说法表达同一件事。工作状态区则完全由 LangGraph 托管。8.2 记忆更新的安全策略长期记忆一旦写错会影响后续所有对话。所以写入策略要谨慎高置信度信息用户明确说“我的税号是 X”可以自动写入。低置信度信息模型推测的偏好需要二次确认或者设计为“暂存待确认”状态。用户主动要求删除记忆时必须有删除通道并且删除操作要记录审计日志。记忆更新需要保留版本方便回滚。8.3 成本与性能治理记忆系统会带来额外的模型调用和存储开销。工程治理上要注意几点信息抽取环节可以复用原有 LLM 调用结果不要每次都新增一次完整对话调用。用轻量模型做抽取和分类用能力强的模型做最终回复。检索结果要设置数量上限并且对检索相关度设置阈值过滤低置信度结果。记忆摘要可以异步执行避免阻塞主对话链路。8.4 权限与隐私隔离企业级应用必须考虑多租户隔离。即使是同一家公司的不同部门也不能互相读取记忆。实现上至少做到三层数据隔离所有记忆写入时附带 tenant_id 和 user_id。接口隔离检索接口强制从认证上下文获取用户身份禁止客户端传入 user_id 覆盖。审计隔离所有记忆读写操作记录操作者、时间、内容摘要。8.5 观测体系建设带上记忆系统后Agent 的行为链路变长必须建立完整的观测指标指标意义记忆写入成功率判断抽取和存储是否正常记忆检索命中率判断检索策略是否有效注入记忆 token 占比判断上下文预算是否合理记忆导致回复提升率通过 A/B 对比衡量记忆价值记忆误用率模型引用了错误的记忆导致回复错误线上环境建议对每条记忆内容生成来源 trace_id方便回溯是哪一轮对话产生的记忆。9. 总结与后续学习方向Agent 记忆不是单一组件而是一套分层架构。从短期工作记忆到长期语义记忆再到跨会话状态持久化每一层都有不同的技术选型和工程治理要求。我建议按这个顺序继续深入先用 LangChain 的 Memory 组件快速验证记忆基本流程理解 Token 膨胀问题。再用 LangGraph 的 State 和 Checkpoint 管理短期工作记忆掌握会话级别状态恢复。最后构建长期记忆系统把关键信息抽取、向量检索、用户隔离结合起来。本文示例使用的 Chroma 和 SQLite 适合本地演示生产环境需要根据数据规模切换到更成熟的存储方案。想继续深入的话重点研究 LangGraph 的 StateGraph 各节点生命周期管理、Common Retriever 的混合检索策略向量检索 关键词检索 重排序、以及针对超长上下文的自动压缩机制。记忆系统做得好不好对用户体验和 Agent 能力上限影响极大。希望这篇文章能帮你建立一条清晰的落地方案少踩一些重复的坑。如果哪一步卡住了欢迎对照代码逐段运行调试。为了后面查找方便可以先收藏这篇文章备用。
