基于LangChain和LangGraph构建长时记忆对话Agent

1. 项目概述构建带长时记忆的对话Agent在传统对话系统中上下文丢失一直是个令人头疼的问题。想象一下你告诉智能助手我喜欢吃辣但五分钟后再问推荐个餐厅它却完全忘了你的口味偏好。这种割裂的交互体验正是我们要解决的核心痛点。基于LangChain和LangGraph技术栈我们可以构建一个具备长时记忆能力的智能Agent。这个Agent不仅能记住单次对话中的上下文还能将关键信息持久化存储在后续对话中智能调用。就像一位贴心的私人助理它能记住你的饮食习惯、居住城市、兴趣爱好并在不同对话线程中保持连贯的个性化服务。2. 技术架构解析2.1 核心组件设计这个长时记忆Agent系统主要由三大模块构成记忆存储层采用向量数据库存储记忆片段支持语义检索记忆处理层包含记忆的提取、存储和检索工具对话引擎层基于LangGraph的状态机管理对话流程# 架构伪代码示意 class LongMemoryAgent: def __init__(self): self.vector_store VectorStore() # 记忆仓库 self.tools [save_memory, search_memory] # 记忆工具 self.graph StateGraph() # 对话状态机2.2 关键技术选型选择LangGraph而非原生LangChain Agent主要考虑以下因素持久化状态管理LangGraph内置的checkpoint机制能更好地处理长时记忆灵活的工作流可以自定义记忆加载节点和工具调用逻辑多线程支持不同对话线程可以共享同一用户的记忆池3. 实现细节拆解3.1 记忆存储实现记忆以两种形式存储文本片段如用户喜欢披萨知识三元组(subject, predicate, object)结构化数据tool def save_recall_memory(memory: str, config: RunnableConfig) - str: 保存记忆到向量数据库 doc Document( page_contentmemory, metadata{user_id: config[user_id]} ) vector_store.add_documents([doc]) return memory3.2 记忆检索机制检索时采用相似度搜索用户过滤的双重机制从当前对话提取关键词在向量库搜索相似记忆过滤出同一用户的记忆片段def search_memories(query: str, user_id: str) - List[str]: docs vector_store.similarity_search(query, k3) return [d.page_content for d in docs if d.metadata[user_id] user_id]3.3 对话图构建LangGraph的状态机设计是关键记忆加载节点在对话开始前注入相关记忆Agent节点处理用户输入并决定是否调用工具工具节点执行记忆存储/检索等操作builder StateGraph(State) builder.add_node(load_memories, load_memories) builder.add_node(agent, agent_node) builder.add_node(tools, tool_node) builder.add_edge(START, load_memories) builder.add_edge(load_memories, agent) builder.add_conditional_edges(agent, route_tools) builder.add_edge(tools, agent)4. 完整实现流程4.1 环境准备首先安装必要依赖pip install langgraph langchain-openai tiktoken4.2 核心代码实现初始化向量存储和LLMfrom langchain_openai import ChatOpenAI, OpenAIEmbeddings vector_store InMemoryVectorStore(OpenAIEmbeddings()) llm ChatOpenAI(modelgpt-4)定义记忆工具tool def save_memory(content: str, config: RunnableConfig): 保存用户记忆 user_id config[configurable][user_id] doc Document(content, metadata{user_id: user_id}) vector_store.add_documents([doc]) return 记忆保存成功 tool def recall_memory(query: str, config: RunnableConfig): 检索相关记忆 user_id config[configurable][user_id] docs vector_store.similarity_search(query, filter{user_id: user_id}) return [d.page_content for d in docs]构建对话图from langgraph.graph import StateGraph class AgentState(TypedDict): messages: List[BaseMessage] memories: List[str] graph StateGraph(AgentState) graph.add_node(agent, call_model) graph.add_node(tools, call_tools) graph.add_conditional_edges(agent, should_call_tools) graph.add_edge(tools, agent)4.3 运行示例# 第一次对话 response graph.invoke( {messages: [HumanMessage(我喜欢吃辣)]}, config{user_id: 123} ) # 后续对话 response graph.invoke( {messages: [HumanMessage(推荐个餐厅)]}, config{user_id: 123} ) # 此时Agent会记得用户口味偏好5. 实战技巧与避坑指南5.1 记忆优化策略记忆分块过长的记忆文本会影响检索效果建议控制在50-100字元数据标注为记忆添加时间戳、重要性等元数据定期清理实现记忆衰减机制淘汰过时信息# 带元数据的记忆存储示例 doc Document( content用户喜欢川菜, metadata{ user_id: 123, timestamp: datetime.now(), importance: 0.8 } )5.2 常见问题解决问题1记忆检索不准确解决方案调整相似度阈值添加关键词提取预处理from langchain.text_splitter import CharacterTextSplitter text_splitter CharacterTextSplitter( chunk_size100, chunk_overlap20 )问题2多用户记忆混淆解决方案严格用户隔离检索时添加user_id过滤def filter_by_user(doc: Document) - bool: return doc.metadata.get(user_id) current_user问题3LLM不主动调用记忆工具解决方案在系统提示中强调记忆使用规范你是一个有长期记忆的助手必须遵守以下规则 1. 遇到个人信息时主动调用save_memory 2. 回答前先检索相关记忆 3. 记忆内容要简洁明确6. 进阶扩展方向6.1 记忆结构化升级将文本记忆升级为知识图谱class KnowledgeNode(TypedDict): subject: str relation: str object: str def save_to_knowledge_graph(fact: KnowledgeNode): neo4j_graph.run( MERGE (s:Entity {name: $subject}) MERGE (o:Entity {name: $object}) MERGE (s)-[r:RELATION {type: $relation}]-(o), **fact )6.2 多模态记忆支持图片、语音等记忆形式tool def save_image_memory(image: bytes, description: str): # 存储到向量数据库 img_embedding clip_model.encode_image(image) text_embedding clip_model.encode_text(description) combined_embedding np.concatenate([img_embedding, text_embedding]) vector_store.add_embeddings([combined_embedding])6.3 记忆安全控制添加记忆访问权限管理class MemoryAccessPolicy: def __init__(self): self.policies defaultdict(list) def add_policy(self, user_id, memory_type, access_level): self.policies[user_id].append((memory_type, access_level)) def check_access(self, user_id, memory): return any( memory.metadata[type] mem_type and level ACCESS_READ for mem_type, level in self.policies[user_id] )构建带长时记忆的Agent时最关键的是要在记忆的丰富性和检索效率之间找到平衡。经过多个项目实践我发现采用分层记忆架构效果最好高频信息存于内存中频存向量数据库低频存关系型数据库。同时建议为每个记忆添加置信度评分在检索时优先返回高置信度内容。