1. 项目概述LangGraph 是一个基于图结构的 AI 开发框架它允许开发者通过连接不同的功能模块来构建复杂的 AI 应用。最近我在一个客户项目中尝试使用 LangGraph 实现了一个具备 RAG检索增强生成和长期记忆能力的 AI Agent效果出乎意料地好。这个 Agent 不仅能回答专业领域的问题还能记住对话历史并在后续交流中主动引用大大提升了用户体验。对于需要处理复杂任务、长期交互的 AI 系统来说RAG 和长期记忆是两大核心能力。RAG 让 AI 能够从外部知识库获取最新信息而长期记忆则使 AI 能够建立上下文感知和个性化交互。本文将详细介绍如何利用 LangGraph 的强大功能实现这两项关键技术。2. 核心架构设计2.1 LangGraph 基础概念LangGraph 的核心思想是将 AI 应用建模为有向图其中节点代表处理单元边代表数据流向。与传统的线性链式处理不同图结构允许更复杂的逻辑流包括条件分支并行处理循环反馈状态保持在 LangGraph 中每个节点都是一个独立的函数或 LangChain 组件可以执行特定任务如调用 LLM执行检索处理数据管理记忆2.2 系统整体架构我们设计的 Agent 架构包含以下核心组件[用户输入] → [输入解析节点] → [记忆检索节点] → [RAG检索节点] → [响应生成节点] → [记忆更新节点] → [输出响应]这个流程看似线性但实际上每个节点都可能触发子图或循环。例如当记忆检索节点发现相关信息不足时可以跳过直接进入 RAG 检索当响应生成节点需要多轮思考时可以形成内部循环。3. RAG 实现细节3.1 知识库构建有效的 RAG 系统始于高质量的知识库构建。我们采用了以下流程数据收集从客户提供的 PDF、网页和内部文档中提取文本分块处理使用递归字符分割器设置 chunk_size1000chunk_overlap200向量化采用 OpenAI 的 text-embedding-3-large 模型生成嵌入索引存储使用 FAISS 实现高效的相似度搜索提示分块大小对检索质量影响很大。经过测试技术文档适合 800-1200 字符的分块而对话记录更适合 400-600 字符的分块。3.2 检索节点实现在 LangGraph 中我们创建了一个专用的检索节点from langchain_community.vectorstores import FAISS from langchain_core.retrievers import BaseRetriever class CustomRetriever(BaseRetriever): def __init__(self, vectorstore): self.vectorstore vectorstore def get_relevant_documents(self, query): # 添加查询重写逻辑 expanded_query self._expand_query(query) return self.vectorstore.similarity_search(expanded_query, k3) def _expand_query(self, query): # 使用小模型生成查询扩展 return query generate_query_expansion(query)这个检索器不仅执行简单的相似度搜索还加入了查询扩展功能显著提高了召回率。3.3 检索结果融合检索到的文档需要与对话上下文融合后才能送入生成环节。我们实现了以下策略相关性过滤只保留相似度分数 0.7 的结果多样性采样确保结果来自不同文档源上下文压缩使用 LLM 提取最相关的片段def format_retrieved_docs(docs, query): context \n\n.join([f来源{doc.metadata[source]}\n内容{doc.page_content} for doc in docs]) compressed llm.invoke(f基于问题{query}从以下文本中提取最相关信息\n{context}) return compressed4. 长期记忆实现4.1 记忆存储设计长期记忆系统需要解决三个核心问题存储什么对话历史、用户偏好、事实知识如何存储向量存储 传统数据库何时更新每次交互后增量更新我们采用了分层存储架构短期记忆保存在对话状态中容量有限但访问快长期记忆存储在外部数据库中容量大但检索慢摘要记忆定期生成的对话摘要平衡细节与效率4.2 记忆检索优化简单的向量检索在记忆系统中效果不佳因为用户可能以不同方式提及相同概念。我们实现了多路检索策略直接向量检索标准的相似度搜索时间加权检索最近记忆获得更高权重关联检索通过知识图谱查找相关概念元数据过滤按记忆类型事实/偏好/事件筛选def retrieve_memories(query, user_id): # 并行执行多种检索 vector_results vector_store.search(query) temporal_results temporal_index.search(query, recency_weight0.3) related_results kg.find_related_concepts(query) # 结果融合与去重 all_results merge_results(vector_results, temporal_results, related_results) return rank_results(all_results)4.3 记忆更新策略记忆系统最容易犯的错误是存储过多无关信息。我们制定了严格的更新规则重要性评估使用 LLM 判断信息是否值得长期记忆去重检查与现有记忆比较避免重复摘要生成定期将多个相关记忆合并为更高层次的摘要def should_remember(text, user_id): response llm.invoke( f评估以下文本是否值得作为长期记忆存储\n{text}\n 仅回答是或否 ) return response.strip() 是5. LangGraph 实现技巧5.1 状态管理LangGraph 的核心是状态对象它在节点间传递。我们设计了专门的状态类class AgentState(TypedDict): user_input: str context: List[dict] memories: List[dict] retrieved_docs: List[Document] response: Optional[str]5.2 条件边与循环通过条件边实现复杂逻辑流from langgraph.graph import Graph workflow Graph() # 定义节点 workflow.add_node(parse_input, parse_input) workflow.add_node(retrieve_memories, retrieve_memories) # ... 其他节点 # 定义边 workflow.add_edge(parse_input, retrieve_memories) workflow.add_conditional_edge( retrieve_memories, decide_next_step, { enough_info: generate_response, need_more: retrieve_docs } )5.3 调试技巧LangGraph 应用调试比较困难我们总结了以下方法可视化图结构使用workflow.get_graph().draw_mermaid()生成流程图状态检查点在每个节点后记录状态快照单元测试节点单独测试每个节点函数追踪执行路径添加日志记录决策过程6. 性能优化6.1 缓存策略重复计算是性能杀手我们实现了多级缓存LLM 响应缓存相同输入直接返回缓存结果检索结果缓存相似查询返回近似结果记忆缓存高频访问记忆保留在内存中from langchain.cache import SQLiteCache from langchain.globals import set_llm_cache set_llm_cache(SQLiteCache(database_path.langchain.db))6.2 异步执行利用 LangGraph 的异步支持提升吞吐量async def aprocess_node(state): # 并行执行独立任务 mem_future retrieve_memories(state[user_input]) docs_future retrieve_docs(state[user_input]) memories, docs await asyncio.gather(mem_future, docs_future) return {**state, memories: memories, retrieved_docs: docs}6.3 负载测试使用 locust 模拟不同负载下的表现from locust import HttpUser, task class AgentUser(HttpUser): task def chat(self): self.client.post(/chat, json{message: 测试消息})测试结果指导我们优化批处理大小和并发设置。7. 实际应用案例7.1 技术支持助手为客户实现的 IT 支持助手表现平均响应时间1.2 秒首次解决率78%用户满意度4.6/5.0关键成功因素整合了产品文档和社区讨论的 RAG 知识库能记住用户的设备配置和历史问题复杂问题自动转人工的平滑交接7.2 教育辅导助手一个数学辅导 Agent 的特殊设计分步解题而非直接给答案记住学生的薄弱知识点根据学习进度调整难度def generate_hint(problem, student_id): mistakes get_common_mistakes(student_id) return llm.invoke( f为这个问题生成提示避免学生常犯的这些错误{mistakes}\n f问题{problem} )8. 常见问题与解决8.1 记忆检索不准确症状Agent 频繁回忆无关信息解决调整向量相似度阈值添加元数据过滤实现记忆衰减机制def decay_memory_weights(user_id): # 每隔一段时间降低旧记忆的权重 for memory in get_memories(user_id): age now() - memory[timestamp] memory[weight] * 0.9 ** (age.days / 30)8.2 RAG 结果不相关症状检索到的文档不能回答问题解决优化分块策略添加查询扩展实现重排序模型8.3 响应速度慢症状简单查询也耗时较长解决实现缓存层优化图结构减少不必要节点设置超时和回退机制9. 进阶技巧9.1 记忆压缩长期积累的记忆会变得臃肿我们定期运行压缩流程聚类相似记忆使用嵌入向量聚类生成摘要用 LLM 合并相关记忆淘汰低频记忆删除长时间未访问的内容9.2 个性化适配通过少量样本学习用户风格few_shot_examples get_user_examples(user_id) prompt PromptTemplate.from_examples( examplesfew_shot_examples, suffix请以类似风格回答用户问题{input} )9.3 安全防护防止记忆系统被恶意利用输入过滤检测并阻止有害内容记忆审查定期扫描存储的记忆权限控制敏感记忆需要额外授权def is_safe_content(text): return not llm.invoke( f判断以下文本是否包含有害内容\n{text}\n 仅回答是或否 ).strip() 是在实际部署中我发现 LangGraph 的最大优势是其灵活性。当客户突然要求添加一个新功能时通常只需要添加一个新节点或调整图结构而不需要重构整个系统。例如当需要增加多模态支持时我们简单地添加了一个图像处理分支节点整个集成过程只用了不到一天时间。 LangGraph框架实现RAG与长期记忆的AI Agent设计
1. 项目概述LangGraph 是一个基于图结构的 AI 开发框架它允许开发者通过连接不同的功能模块来构建复杂的 AI 应用。最近我在一个客户项目中尝试使用 LangGraph 实现了一个具备 RAG检索增强生成和长期记忆能力的 AI Agent效果出乎意料地好。这个 Agent 不仅能回答专业领域的问题还能记住对话历史并在后续交流中主动引用大大提升了用户体验。对于需要处理复杂任务、长期交互的 AI 系统来说RAG 和长期记忆是两大核心能力。RAG 让 AI 能够从外部知识库获取最新信息而长期记忆则使 AI 能够建立上下文感知和个性化交互。本文将详细介绍如何利用 LangGraph 的强大功能实现这两项关键技术。2. 核心架构设计2.1 LangGraph 基础概念LangGraph 的核心思想是将 AI 应用建模为有向图其中节点代表处理单元边代表数据流向。与传统的线性链式处理不同图结构允许更复杂的逻辑流包括条件分支并行处理循环反馈状态保持在 LangGraph 中每个节点都是一个独立的函数或 LangChain 组件可以执行特定任务如调用 LLM执行检索处理数据管理记忆2.2 系统整体架构我们设计的 Agent 架构包含以下核心组件[用户输入] → [输入解析节点] → [记忆检索节点] → [RAG检索节点] → [响应生成节点] → [记忆更新节点] → [输出响应]这个流程看似线性但实际上每个节点都可能触发子图或循环。例如当记忆检索节点发现相关信息不足时可以跳过直接进入 RAG 检索当响应生成节点需要多轮思考时可以形成内部循环。3. RAG 实现细节3.1 知识库构建有效的 RAG 系统始于高质量的知识库构建。我们采用了以下流程数据收集从客户提供的 PDF、网页和内部文档中提取文本分块处理使用递归字符分割器设置 chunk_size1000chunk_overlap200向量化采用 OpenAI 的 text-embedding-3-large 模型生成嵌入索引存储使用 FAISS 实现高效的相似度搜索提示分块大小对检索质量影响很大。经过测试技术文档适合 800-1200 字符的分块而对话记录更适合 400-600 字符的分块。3.2 检索节点实现在 LangGraph 中我们创建了一个专用的检索节点from langchain_community.vectorstores import FAISS from langchain_core.retrievers import BaseRetriever class CustomRetriever(BaseRetriever): def __init__(self, vectorstore): self.vectorstore vectorstore def get_relevant_documents(self, query): # 添加查询重写逻辑 expanded_query self._expand_query(query) return self.vectorstore.similarity_search(expanded_query, k3) def _expand_query(self, query): # 使用小模型生成查询扩展 return query generate_query_expansion(query)这个检索器不仅执行简单的相似度搜索还加入了查询扩展功能显著提高了召回率。3.3 检索结果融合检索到的文档需要与对话上下文融合后才能送入生成环节。我们实现了以下策略相关性过滤只保留相似度分数 0.7 的结果多样性采样确保结果来自不同文档源上下文压缩使用 LLM 提取最相关的片段def format_retrieved_docs(docs, query): context \n\n.join([f来源{doc.metadata[source]}\n内容{doc.page_content} for doc in docs]) compressed llm.invoke(f基于问题{query}从以下文本中提取最相关信息\n{context}) return compressed4. 长期记忆实现4.1 记忆存储设计长期记忆系统需要解决三个核心问题存储什么对话历史、用户偏好、事实知识如何存储向量存储 传统数据库何时更新每次交互后增量更新我们采用了分层存储架构短期记忆保存在对话状态中容量有限但访问快长期记忆存储在外部数据库中容量大但检索慢摘要记忆定期生成的对话摘要平衡细节与效率4.2 记忆检索优化简单的向量检索在记忆系统中效果不佳因为用户可能以不同方式提及相同概念。我们实现了多路检索策略直接向量检索标准的相似度搜索时间加权检索最近记忆获得更高权重关联检索通过知识图谱查找相关概念元数据过滤按记忆类型事实/偏好/事件筛选def retrieve_memories(query, user_id): # 并行执行多种检索 vector_results vector_store.search(query) temporal_results temporal_index.search(query, recency_weight0.3) related_results kg.find_related_concepts(query) # 结果融合与去重 all_results merge_results(vector_results, temporal_results, related_results) return rank_results(all_results)4.3 记忆更新策略记忆系统最容易犯的错误是存储过多无关信息。我们制定了严格的更新规则重要性评估使用 LLM 判断信息是否值得长期记忆去重检查与现有记忆比较避免重复摘要生成定期将多个相关记忆合并为更高层次的摘要def should_remember(text, user_id): response llm.invoke( f评估以下文本是否值得作为长期记忆存储\n{text}\n 仅回答是或否 ) return response.strip() 是5. LangGraph 实现技巧5.1 状态管理LangGraph 的核心是状态对象它在节点间传递。我们设计了专门的状态类class AgentState(TypedDict): user_input: str context: List[dict] memories: List[dict] retrieved_docs: List[Document] response: Optional[str]5.2 条件边与循环通过条件边实现复杂逻辑流from langgraph.graph import Graph workflow Graph() # 定义节点 workflow.add_node(parse_input, parse_input) workflow.add_node(retrieve_memories, retrieve_memories) # ... 其他节点 # 定义边 workflow.add_edge(parse_input, retrieve_memories) workflow.add_conditional_edge( retrieve_memories, decide_next_step, { enough_info: generate_response, need_more: retrieve_docs } )5.3 调试技巧LangGraph 应用调试比较困难我们总结了以下方法可视化图结构使用workflow.get_graph().draw_mermaid()生成流程图状态检查点在每个节点后记录状态快照单元测试节点单独测试每个节点函数追踪执行路径添加日志记录决策过程6. 性能优化6.1 缓存策略重复计算是性能杀手我们实现了多级缓存LLM 响应缓存相同输入直接返回缓存结果检索结果缓存相似查询返回近似结果记忆缓存高频访问记忆保留在内存中from langchain.cache import SQLiteCache from langchain.globals import set_llm_cache set_llm_cache(SQLiteCache(database_path.langchain.db))6.2 异步执行利用 LangGraph 的异步支持提升吞吐量async def aprocess_node(state): # 并行执行独立任务 mem_future retrieve_memories(state[user_input]) docs_future retrieve_docs(state[user_input]) memories, docs await asyncio.gather(mem_future, docs_future) return {**state, memories: memories, retrieved_docs: docs}6.3 负载测试使用 locust 模拟不同负载下的表现from locust import HttpUser, task class AgentUser(HttpUser): task def chat(self): self.client.post(/chat, json{message: 测试消息})测试结果指导我们优化批处理大小和并发设置。7. 实际应用案例7.1 技术支持助手为客户实现的 IT 支持助手表现平均响应时间1.2 秒首次解决率78%用户满意度4.6/5.0关键成功因素整合了产品文档和社区讨论的 RAG 知识库能记住用户的设备配置和历史问题复杂问题自动转人工的平滑交接7.2 教育辅导助手一个数学辅导 Agent 的特殊设计分步解题而非直接给答案记住学生的薄弱知识点根据学习进度调整难度def generate_hint(problem, student_id): mistakes get_common_mistakes(student_id) return llm.invoke( f为这个问题生成提示避免学生常犯的这些错误{mistakes}\n f问题{problem} )8. 常见问题与解决8.1 记忆检索不准确症状Agent 频繁回忆无关信息解决调整向量相似度阈值添加元数据过滤实现记忆衰减机制def decay_memory_weights(user_id): # 每隔一段时间降低旧记忆的权重 for memory in get_memories(user_id): age now() - memory[timestamp] memory[weight] * 0.9 ** (age.days / 30)8.2 RAG 结果不相关症状检索到的文档不能回答问题解决优化分块策略添加查询扩展实现重排序模型8.3 响应速度慢症状简单查询也耗时较长解决实现缓存层优化图结构减少不必要节点设置超时和回退机制9. 进阶技巧9.1 记忆压缩长期积累的记忆会变得臃肿我们定期运行压缩流程聚类相似记忆使用嵌入向量聚类生成摘要用 LLM 合并相关记忆淘汰低频记忆删除长时间未访问的内容9.2 个性化适配通过少量样本学习用户风格few_shot_examples get_user_examples(user_id) prompt PromptTemplate.from_examples( examplesfew_shot_examples, suffix请以类似风格回答用户问题{input} )9.3 安全防护防止记忆系统被恶意利用输入过滤检测并阻止有害内容记忆审查定期扫描存储的记忆权限控制敏感记忆需要额外授权def is_safe_content(text): return not llm.invoke( f判断以下文本是否包含有害内容\n{text}\n 仅回答是或否 ).strip() 是在实际部署中我发现 LangGraph 的最大优势是其灵活性。当客户突然要求添加一个新功能时通常只需要添加一个新节点或调整图结构而不需要重构整个系统。例如当需要增加多模态支持时我们简单地添加了一个图像处理分支节点整个集成过程只用了不到一天时间。