LangChain+LangGraph+DeepAgent构建企业级AI Agent长期记忆架构实战

如果你正在构建一个需要“记住”用户历史对话、持续学习并做出长期决策的智能应用，比如一个能陪伴用户数月、提供个性化建议的AI助手，或者一个能处理复杂、多步骤业务流程的自动化系统，那么你很可能已经遇到了一个核心难题：如何让AI Agent拥有真正有效的长期记忆？

这不仅仅是把对话记录存进数据库那么简单。一个健壮的长期记忆系统，需要解决记忆的存储、检索、更新、遗忘、关联以及跨会话一致性等一系列复杂问题。很多开发者尝试用简单的向量数据库+相似度搜索来构建，结果却发现Agent要么“记性不好”（检索不到关键信息），要么“思维混乱”（被无关记忆干扰），要么“人格分裂”（不同会话间行为不一致）。

这正是LangChain + LangGraph + DeepAgent这套技术栈要解决的核心痛点。它不是一个简单的工具介绍，而是一套面向2026年及未来的、企业级Agent长期记忆架构的完整工程化解决方案。本文将彻底解析这套架构的底层原理，并通过实战演示如何从零搭建一个具备长期记忆能力的智能Agent。你将获得的不是一堆零散的概念，而是一个可以直接复用于生产环境的、清晰的架构蓝图和代码实现。

1. 长期记忆：Agent智能进化的分水岭

在深入技术细节之前，我们必须先达成一个共识：长期记忆是区分“玩具级”Agent和“工业级”Agent的关键标志。

一个没有长期记忆的Agent，就像金鱼一样，每次对话都是全新的开始。它无法进行深度的个性化服务，无法执行跨越数天甚至数月的复杂任务，更无法在持续的交互中学习和进化。其价值仅限于单次、简单的问答或操作。

而一个拥有长期记忆的Agent，则具备了“成长”的能力。它可以：

• 实现真正的个性化：记住用户的偏好、习惯和历史问题，提供量身定制的响应。
• 处理复杂工作流：记住多轮对话的上下文和任务状态，驱动一个需要多次交互才能完成的业务流程（如订票、客服工单跟进、项目规划）。
• 保持状态一致性：在分布式或长时间运行的环境中，确保Agent的“人格”和行为逻辑在不同时间点保持一致。
• 实现持续学习：从历史交互中总结经验，优化未来的决策策略。

LangChain、LangGraph、DeepAgent在这个架构中扮演着截然不同但又相辅相成的角色：

• LangChain：是“胶水”和“工具箱”。它提供了连接大模型、处理文本、与各种数据库（尤其是向量数据库）交互的基础能力。它是构建记忆存储和检索层的基石。
• LangGraph：是“大脑”和“调度中心”。它用图（Graph）的概念来建模Agent的工作流，其中节点代表处理步骤（如“思考”、“执行”、“记忆”），边代表步骤间的流转逻辑。长期记忆的读写时机、记忆如何影响决策流程，正是由LangGraph来精确控制的。
• DeepAgent：是“高级认知框架”或“最佳实践封装”。它通常建立在LangChain和LangGraph之上，提供更高层次的抽象，比如预定义的记忆管理策略、更复杂的Agent角色设定、多Agent协作模式等。它让构建一个具备深度记忆和推理能力的Agent变得更加模块化和高效。

理解了这三者的分工，我们就能看清整个架构的脉络：用LangChain处理数据和工具，用LangGraph编排包含记忆操作的复杂工作流，用DeepAgent（或类似框架）来封装和提升整个系统的智能水平。

2. 核心概念与架构原理拆解

2.1 什么是Agent的“记忆”？

在技术实现上，Agent的记忆远不止于聊天记录。我们可以将其分为几个层次：

1. 短期记忆/工作记忆：即当前对话的上下文窗口。这通常由大模型本身的Token长度限制，是临时的、易失的。
2. 长期记忆：需要持久化到外部存储的信息。它又可以分为：
   • 事实性记忆：用户明确提供的个人信息、偏好、历史事件等。
   • 程序性记忆：Agent学会的如何完成某项任务的步骤或规则。
   • 情感/关系记忆：基于交互历史推断出的用户情绪倾向或关系亲密度（用于调整对话风格）。
   • 摘要性记忆：对长对话或复杂事件的浓缩摘要，用于高效检索。

2.2 LangGraph：将记忆编织进工作流

LangGraph的核心思想是用有向图来定义Agent的行为。一个典型的具备记忆功能的Agent工作流可能包含以下节点：

• retrieve_memory(检索记忆)：根据当前查询，从长期记忆库中查找相关历史信息。
• reason(推理)：结合检索到的记忆和当前问题，进行思考并规划行动。
• act(执行)：调用工具（如搜索、计算、API）来执行规划。
• update_memory(更新记忆)：将本次交互中有价值的信息（如新的用户偏好、任务进展）写回长期记忆库。

这些节点通过条件边（conditional edges）连接，形成一个动态的、可循环的决策流程。记忆的检索和更新，成为了工作流中一等公民的步骤，而不再是被动触发的副作用。

2.3 记忆的存储与检索：超越简单的向量搜索

这是长期记忆系统的核心引擎。常见的模式是“向量索引 + 元数据过滤 + 递归检索”。

1. 向量化与嵌入：使用文本嵌入模型（如OpenAI的text-embedding-3-small）将记忆文本转换为高维向量。
2. 向量数据库存储：将向量及其关联的原始文本、元数据（如时间戳、用户ID、记忆类型、重要性分数）存入向量数据库（如Chroma, Pinecone, Weaviate）。
3. 混合检索：
   • 相似性检索：计算当前查询的向量与记忆向量的余弦相似度，返回最相关的几条。
   • 元数据过滤：同时用user_id=‘xxx’和memory_type=‘preference’这样的过滤器来精确缩小范围。
   • 时间衰减：为较旧的记忆赋予较低的权重，让Agent更“健忘”一些无关紧要的过去，专注于近期和重要的信息。
4. 记忆聚合与摘要：对于检索到的多条相关记忆，可能需要用一个LLM来生成一个连贯的摘要，再喂给推理节点，避免输入过于冗杂。

2.4 DeepAgent的角色：提供高层设计模式

DeepAgent这类框架，通常会提供一些开箱即用的记忆管理组件，例如：

• MemoryManager类：统一管理不同记忆类型的读写接口。
• SummarizationMemory：自动将长对话总结成要点并存储。
• EntityMemory：专门用于提取和记忆对话中出现的实体（人名、地点、产品名）及其属性。
• 多Agent共享记忆：定义不同Agent之间如何安全、可控地共享部分记忆。

3. 环境准备与项目初始化

在开始实战之前，请确保你的开发环境已就绪。我们将使用Python作为主要语言。

基础环境要求：

• Python: 3.10 或更高版本（推荐3.11+）
• 包管理工具: pip 或 poetry
• OpenAI API Key: 用于调用GPT模型和嵌入模型（也可替换为其他兼容API的模型，如DeepSeek、通义千问等）

创建项目并安装核心依赖：我们创建一个新的项目目录，并安装必要的库。

# 创建项目目录 mkdir enterprise-agent-memory && cd enterprise-agent-memory # 创建虚拟环境（可选但推荐） python -m venv venv # Windows: venv\Scripts\activate # Mac/Linux: source venv/bin/activate # 安装核心依赖 pip install langchain langgraph langchain-openai chromadb tiktoken # langchain: Agent基础框架 # langgraph: 工作流编排 # langchain-openai: OpenAI模型集成 # chromadb: 轻量级向量数据库（用于本地演示） # tiktoken: Token计数工具

环境变量配置：创建一个.env文件来安全地管理你的API密钥和其他配置。

# .env OPENAI_API_KEY=sk-your-openai-api-key-here # 可选：其他模型的API密钥，如 DASHSCOPE_API_KEY (阿里), DEEPSEEK_API_KEY 等

在代码中，使用python-dotenv加载配置：

pip install python-dotenv

# config.py import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的环境变量 OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") assert OPENAI_API_KEY, "OPENAI_API_KEY 未在环境变量或 .env 文件中设置"

4. 实战：构建一个具有长期记忆的个人学习助手Agent

我们将构建一个“学习助手”Agent，它能记住用户学习过的概念、提出的问题、以及用户自述的掌握程度，并在后续对话中提供个性化的复习建议和知识关联。

4.1 步骤一：设计记忆结构并初始化存储

首先，我们定义记忆的数据结构，并初始化向量数据库。

# memory_schema.py from pydantic import BaseModel, Field from datetime import datetime from enum import Enum from typing import Optional class MemoryType(str, Enum): """定义记忆的类型""" FACT = "fact" # 客观事实，如“用户是Python开发者” CONCEPT = "concept" # 学习的概念，如“什么是闭包” QUESTION = "question" # 用户问过的问题 PREFERENCE = "preference" # 用户偏好，如“喜欢通过例子学习” SUMMARY = "summary" # 对话摘要 class MemoryEntity(BaseModel): """单个记忆条目的数据模型""" id: Optional[str] = None # 由数据库生成 content: str = Field(..., description="记忆的文本内容") memory_type: MemoryType user_id: str = Field(..., description="关联的用户ID") importance: float = Field(default=1.0, ge=0.0, le=10.0, description="重要性权重，1-10") created_at: datetime = Field(default_factory=datetime.now) metadata: dict = Field(default_factory=dict) # 额外元数据，如关联的标签、来源等 class Config: use_enum_values = True

接下来，我们创建一个记忆管理类，封装与向量数据库（Chroma）的交互。

# memory_manager.py import hashlib from typing import List, Dict, Any from langchain_chroma import Chroma from langchain_openai import OpenAIEmbeddings from langchain.schema import Document from memory_schema import MemoryEntity, MemoryType from config import OPENAI_API_KEY class MemoryManager: def __init__(self, user_id: str, persist_directory: str = "./chroma_db"): self.user_id = user_id self.persist_directory = persist_directory # 初始化嵌入模型 self.embeddings = OpenAIEmbeddings( openai_api_key=OPENAI_API_KEY, model="text-embedding-3-small" ) # 初始化Chroma向量数据库，按用户ID分集合(collection) collection_name = f"user_memories_{self._hash_user_id(user_id)}" self.vectorstore = Chroma( collection_name=collection_name, embedding_function=self.embeddings, persist_directory=persist_directory ) def _hash_user_id(self, user_id: str) -> str: """对用户ID进行简单哈希，作为集合名的一部分""" return hashlib.md5(user_id.encode()).hexdigest()[:8] def add_memory(self, memory: MemoryEntity) -> str: """添加一条记忆到向量数据库""" # 为记忆生成唯一ID（这里用内容哈希模拟） doc_id = hashlib.md5(f"{memory.content}_{memory.created_at}".encode()).hexdigest() # 创建LangChain Document对象 doc = Document( page_content=memory.content, metadata={ "memory_type": memory.memory_type, "user_id": memory.user_id, "importance": memory.importance, "created_at": memory.created_at.isoformat(), **memory.metadata }, id=doc_id ) # 添加到向量库 self.vectorstore.add_documents(documents=[doc], ids=[doc_id]) return doc_id def retrieve_related_memories(self, query: str, memory_types: List[MemoryType] = None, k: int = 5) -> List[Dict[str, Any]]: """检索与查询相关的记忆，支持按类型过滤""" # 构建元数据过滤器 filter_dict = {"user_id": self.user_id} if memory_types: filter_dict["memory_type"] = {"$in": [mt.value for mt in memory_types]} # 执行相似性搜索，并应用过滤器 docs_and_scores = self.vectorstore.similarity_search_with_score( query, k=k, filter=filter_dict ) # 格式化返回结果 results = [] for doc, score in docs_and_scores: results.append({ "content": doc.page_content, "memory_type": doc.metadata.get("memory_type"), "importance": doc.metadata.get("importance", 1.0), "relevance_score": float(score), # 距离分数，越小越相关 "metadata": {k: v for k, v in doc.metadata.items() if k not in ["memory_type", "user_id", "importance", "created_at"]} }) return results def get_conversation_summary(self, recent_n: int = 10) -> str: """获取最近N条记忆的摘要（简化版：这里仅连接最近几条CONCEPT记忆）""" # 注意：生产环境应使用LLM生成真正的摘要 filter_dict = { "user_id": self.user_id, "memory_type": MemoryType.CONCEPT.value } # 这里我们简单获取最新的几条概念记忆 # Chroma默认按ID排序，对于时间检索不友好。生产环境应考虑使用支持按时间过滤的数据库或在元数据中存储时间戳用于过滤。 # 此处为演示，我们假设add的顺序即时间顺序。 all_docs = self.vectorstore.get(where=filter_dict) recent_contents = [doc['content'] for doc in all_docs['documents'][-recent_n:]] return " | ".join(recent_contents) if recent_contents else "暂无近期学习概念。"

4.2 步骤二：使用LangGraph定义Agent工作流

现在，我们使用LangGraph来编排一个包含记忆检索和更新的学习助手工作流。

# agent_workflow.py from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage, SystemMessage from memory_manager import MemoryManager, MemoryEntity, MemoryType from memory_schema import MemoryType as MT from config import OPENAI_API_KEY # 1. 定义Graph的状态结构 class AgentState(TypedDict): """图工作流的状态定义""" messages: Annotated[List, add_messages] # 对话消息历史 user_id: str # 当前用户ID query: str # 用户当前查询 retrieved_memories: List[dict] # 检索到的相关记忆 new_memories_to_save: List[MemoryEntity] # 需要保存的新记忆 response: str # Agent的最终响应 # 2. 初始化组件 llm = ChatOpenAI(model="gpt-4o-mini", openai_api_key=OPENAI_API_KEY, temperature=0.2) memory_manager = None # 将在运行时根据user_id初始化 # 3. 定义各个节点函数 def initialize_memory(state: AgentState): """初始化或获取该用户的MemoryManager""" global memory_manager if memory_manager is None or memory_manager.user_id != state["user_id"]: memory_manager = MemoryManager(user_id=state["user_id"]) return state def retrieve_memory(state: AgentState): """从长期记忆中检索与当前查询相关的内容""" query = state["query"] # 优先检索概念(concept)和事实(fact)类记忆 related_memories = memory_manager.retrieve_related_memories( query, memory_types=[MT.CONCEPT, MT.FACT], k=3 ) return {"retrieved_memories": related_memories} def reason_and_generate(state: AgentState): """核心推理节点：结合记忆和当前查询，生成思考过程和响应""" messages = state["messages"] query = state["query"] retrieved = state["retrieved_memories"] # 构建系统提示词，注入检索到的记忆 memory_context = "" if retrieved: memory_context = "## 相关历史记忆（供参考）：\n" for mem in retrieved: memory_context += f"- [{mem['memory_type']}] {mem['content']} (相关性：{1-mem['relevance_score']:.2f})\n" system_prompt = f"""你是一个专业、耐心的个人学习助手。你的目标是帮助用户高效学习和巩固知识。 {memory_context} 请基于以上历史记忆（如果存在）和当前对话，理解用户的需求，并给出清晰、有帮助的回应。 如果用户正在学习一个新概念，请用易于理解的方式解释，并尝试关联他已学过的知识。 如果用户是在提问或寻求复习，请根据记忆提供个性化建议。 """ # 构建对话历史（最后几条） chat_history = messages[-5:] if len(messages) > 5 else messages # 构建给LLM的完整消息列表 prompt_messages = [SystemMessage(content=system_prompt)] + chat_history + [HumanMessage(content=query)] # 调用LLM生成响应 ai_message = llm.invoke(prompt_messages) response_text = ai_message.content # 判断是否需要创建新的长期记忆 new_memories = [] # 规则1：如果用户明确陈述了一个新概念或事实，则创建CONCEPT或FACT记忆 if "什么是" in query or "解释一下" in query or "告诉我" in query: # 这里简化处理，实际应用可以用另一个LLM调用或规则来提取关键概念 # 假设我们提取查询中的核心名词作为记忆内容（简化版） new_memories.append( MemoryEntity( content=query, # 实际应提取更精炼的表述 memory_type=MT.CONCEPT, user_id=state["user_id"], importance=7.0, metadata={"source_query": query} ) ) # 规则2：如果用户表达了偏好，创建PREFERENCE记忆 if "我喜欢" in query or "我讨厌" in query or "偏好" in query: new_memories.append( MemoryEntity( content=query, memory_type=MT.PREFERENCE, user_id=state["user_id"], importance=5.0 ) ) return { "response": response_text, "new_memories_to_save": new_memories, "messages": messages + [HumanMessage(content=query), AIMessage(content=response_text)] } def save_memory(state: AgentState): """将需要保存的新记忆持久化到数据库""" new_memories = state.get("new_memories_to_save", []) saved_ids = [] for memory in new_memories: mem_id = memory_manager.add_memory(memory) saved_ids.append(mem_id) print(f"[Memory Saved] 保存了 {len(saved_ids)} 条新记忆。") return {"new_memories_to_save": []} # 清空待保存列表 # 4. 构建工作流图 workflow = StateGraph(AgentState) # 添加节点 workflow.add_node("init_memory", initialize_memory) workflow.add_node("retrieve", retrieve_memory) workflow.add_node("reason", reason_and_generate) workflow.add_node("save", save_memory) # 设置边（定义执行顺序） workflow.set_entry_point("init_memory") workflow.add_edge("init_memory", "retrieve") workflow.add_edge("retrieve", "reason") # 只有在有新记忆需要保存时，才走向save节点，否则结束 workflow.add_conditional_edges( "reason", # 判断条件：是否有新记忆需要保存？ lambda state: "save" if state.get("new_memories_to_save") else END, {"save": "save", END: END} ) workflow.add_edge("save", END) # 编译图 app = workflow.compile()

4.3 步骤三：创建并运行Agent

现在，我们创建一个主程序来使用这个工作流，模拟多轮对话。

# main.py from agent_workflow import app, AgentState from langchain_core.messages import HumanMessage def run_learning_assistant(user_id: str): """运行学习助手Agent的对话循环""" print(f"\n=== 启动学习助手 (用户: {user_id}) ===") print("输入 'quit' 或 '退出' 结束对话。\n") # 初始化状态 config = {"user_id": user_id} state = AgentState(messages=[], user_id=user_id, query="", retrieved_memories=[], new_memories_to_save=[], response="") while True: try: user_input = input("\n你: ").strip() if user_input.lower() in ['quit', '退出', 'exit']: print("助手: 再见！期待下次与你探讨知识。") break if not user_input: continue # 更新状态中的查询 state["query"] = user_input # 执行工作流 result = app.invoke(state, config=config) # 输出助手响应 print(f"\n助手: {result['response']}") # 更新状态，保留对话历史，为下一轮做准备 state["messages"] = result["messages"] state["retrieved_memories"] = [] state["new_memories_to_save"] = [] state["response"] = "" except KeyboardInterrupt: print("\n\n对话被中断。") break except Exception as e: print(f"\n系统出错: {e}") break if __name__ == "__main__": # 可以模拟不同用户 test_user_id = "student_zhangsan" run_learning_assistant(test_user_id)

5. 运行与效果验证

1. 启动对话：

   python main.py

   你会看到提示符，启动与学习助手的对话。

2. 模拟对话流程：

   • 第一轮（学习新概念）：

     你: 什么是Python中的装饰器？ 助手: （解释装饰器概念...）...简单来说，装饰器允许你在不修改原函数代码的情况下，为函数添加新功能。比如，你可以用`@log`装饰器来自动记录函数执行日志。

     • 后台动作：Agent检索了长期记忆（初始为空），生成了回答，并因为问题符合“解释概念”的规则，创建了一条类型为CONCEPT、内容为“什么是Python中的装饰器？”的记忆，并存入向量数据库。

   • 第二轮（关联记忆）：

     你: 那它和闭包有什么关系吗？ 助手: 问得好！你之前了解过装饰器（什么是Python中的装饰器？）。实际上，装饰器本身就是闭包的一个经典应用场景。闭包指的是一个函数记住了它被定义时的外部作用域变量。而装饰器通常就是一个返回内部函数（闭包）的高阶函数，这个内部函数“记住”了原函数，并对其进行了包装扩展。

     • 后台动作：Agent检索记忆时，找到了上一轮创建的关于“装饰器”的记忆。在生成回答时，系统提示词中包含了这条记忆（“## 相关历史记忆...”），因此LLM能够主动提及“你之前了解过装饰器”，并在此基础上建立新旧知识的关联。

   • 第三轮（表达偏好）：

     你: 我喜欢通过实际的代码例子来学习，不喜欢太理论化的解释。 助手: 明白了！我会在后续的解释中多提供具体的代码示例。比如刚才说的装饰器，一个典型的日志装饰器实现是这样的...（附上代码示例）。

     • 后台动作：Agent识别出用户表达了学习偏好（“我喜欢...”），创建了一条类型为PREFERENCE的记忆。这条记忆会在未来的对话中，通过元数据过滤（memory_type=PREFERENCE）被检索出来，从而持续地个性化助手的回答风格。

3. 验证记忆持久化： 对话结束后，你可以检查./chroma_db目录，里面存储了向量化的记忆数据。重启程序后，用同一个user_id对话，Agent依然能“记得”之前学过的概念和表达的偏好。

6. 常见问题与排查思路

7. 企业级最佳实践与工程建议

将上述Demo升级到企业生产环境，需要考虑以下关键点：

1. 记忆的粒度与结构化：

   • 不要存储原始对话：避免存储冗长的原始对话。应提取关键信息（实体、意图、结论）作为结构化记忆。
   • 使用更丰富的数据模型：扩展MemoryEntity，增加tags（标签）、expires_at（过期时间）、access_count（访问计数）等字段，便于精细化管理。

2. 检索策略优化：

   • 混合检索：结合向量相似性搜索（语义匹配）和关键词搜索（精确匹配），并使用LLM作为重排序器对初步结果进行智能排序。
   • 时间衰减与重要性加权：在检索评分中引入时间衰减因子（recency）和重要性权重（importance），让Agent更“健忘”陈旧琐事，更“牢记”重要信息。
   • 分层记忆：实现短期（会话内）、中期（近几天）、长期（核心知识）的多级记忆池，采用不同的存储和检索策略。

3. 记忆的更新与遗忘：

   • 记忆融合：当新信息与旧记忆冲突或重复时，应触发记忆融合逻辑（如用LLM判断是更新、合并还是忽略）。
   • 主动遗忘：设计记忆衰减或清理机制。例如，长期未被访问且重要性低的记忆可以被归档或删除。

4. 使用生产级向量数据库：

   • 本地演示用Chroma，生产环境推荐：Pinecone（全托管）、Weaviate（开源，功能强大）、Qdrant（高性能，Rust编写）或Milvus（适用于超大规模）。
   • 这些数据库提供了更好的性能、可扩展性、高可用性和高级过滤功能。

5. 工作流的可观测性与调试：

   • 记录完整轨迹：利用LangGraph的日志功能，记录每个节点的输入输出，便于调试复杂的工作流。
   • 可视化：使用graph.get_graph().draw_mermaid()生成流程图，帮助团队理解Agent的决策路径。
   • 监控关键指标：监控记忆检索命中率、响应延迟、Token消耗、记忆增长量等。

6. 安全与隐私：

   • 数据加密：对存储的长期记忆进行加密，尤其是在云端。
   • 用户数据隔离：严格确保不同用户的数据在存储和检索时完全隔离。
   • 记忆审查：提供工具让用户查看、编辑或删除Agent存储的关于自己的记忆，符合数据隐私法规（如GDPR）。

7. 与DeepAgent等高级框架集成：

   • 本文的MemoryManager和AgentState可以封装成DeepAgent框架中的一个MemoryModule。
   • DeepAgent可能提供了更优雅的多Agent通信、技能(Skill)管理和记忆共享的原语，可以在此基础上构建更复杂的协作智能体系统。

构建一个拥有长期记忆的Agent，是从“对话机器人”迈向“数字伴侣”或“自主业务助理”的关键一步。本文通过LangChain（数据处理）、LangGraph（流程编排）、DeepAgent（高层模式）的组合，为你展示了一条从原理到实战的清晰路径。真正的挑战不在于实现单个功能，而在于设计一个稳定、高效、可扩展且符合业务逻辑的记忆系统。建议你从本文的Demo出发，根据实际业务需求，重点优化记忆的质量（存什么）、检索效率（怎么找）和效用（怎么用）这三个核心维度，逐步迭代出属于你自己的企业级智能体记忆架构。