基于LangGraph构建智能Agentic RAG系统：从原理到实践

如果你正在构建基于大语言模型的智能应用可能会遇到这样的困境传统RAG系统总是机械地检索文档无论用户问题是简单问候还是复杂查询都要走一遍完整的检索流程。这不仅浪费计算资源还降低了响应速度。更糟糕的是当检索到的文档与问题不相关时系统仍然基于这些无关内容生成答案导致回答质量下降。这正是Agentic RAG要解决的核心问题——让LLM能够智能决策何时检索、何时直接回答以及如何评估检索结果的相关性。而LangGraph作为构建复杂AI工作流的框架为实现这种智能决策提供了强大的工具链。本文将带你从零构建一个完整的Agentic RAG系统重点不是简单调用API而是深入理解每个决策环节的设计原理和实现细节。你会发现真正的智能不在于工具本身而在于如何设计让工具协同工作的流程。1. 环境准备与依赖安装在开始构建之前我们需要准备开发环境。LangGraph生态包含多个组件每个都有特定的用途。# 安装核心依赖包 pip install -U langgraph langchain-anthropic langchain-text-splitters bs4 requests # 如果需要使用OpenAI模型还需要安装 pip install langchain-openai环境配置的关键是正确设置API密钥。不同的模型提供商需要不同的密钥import getpass import os def setup_environment(): 设置必要的环境变量 # OpenAI API密钥 if OPENAI_API_KEY not in os.environ: os.environ[OPENAI_API_KEY] getpass.getpass(OpenAI API Key: ) # 可选Anthropic Claude API密钥 if ANTHROPIC_API_KEY not in os.environ: os.environ[ANTHROPIC_API_KEY] getpass.getpass(Anthropic API Key: ) setup_environment()重要提醒在生产环境中建议使用环境变量或安全的密钥管理服务避免在代码中硬编码密钥。本文为演示目的使用getpass实际项目中请根据安全要求调整。2. LangGraph核心概念解析在深入代码之前需要理解LangGraph的几个核心概念这些概念构成了我们构建智能工作流的基础。2.1 状态管理State ManagementLangGraph使用状态对象在节点间传递数据。对于聊天应用最常用的是MessagesStatefrom langgraph.graph import MessagesState from typing import TypedDict, List from langchain_core.messages import BaseMessage # MessagesState的简化定义 class ChatState(TypedDict): messages: List[BaseMessage] # 存储对话消息历史状态对象确保每个节点都能访问到完整的对话上下文这是实现多轮对话的基础。2.2 节点Nodes与边Edges节点是工作流中的处理单元边定义了节点间的流转逻辑from langgraph.graph import StateGraph, END # 创建状态图 workflow StateGraph(MessagesState) # 添加节点 workflow.add_node(process_input, process_input_function) workflow.add_node(generate_response, generate_response_function) # 添加边定义执行顺序 workflow.add_edge(process_input, generate_response) workflow.add_edge(generate_response, END)2.3 条件边Conditional Edges条件边让工作流具备决策能力这是实现智能RAG的关键def should_retrieve(state: MessagesState) - str: 根据当前状态决定下一步行动 last_message state[messages][-1] # 如果最后一条消息包含工具调用需要检索 if hasattr(last_message, tool_calls) and last_message.tool_calls: return retrieve # 否则直接生成回答 return generate_response # 添加条件边 workflow.add_conditional_edges( process_input, should_retrieve, { retrieve: retrieve_documents, generate_response: generate_response } )理解这些概念后我们就能开始构建具体的RAG组件了。3. 文档预处理与向量化智能RAG系统的质量很大程度上取决于检索基础。我们先从文档处理开始。3.1 文档获取与解析import bs4 import requests from langchain_core.documents import Document from langchain_text_splitters import RecursiveCharacterTextSplitter def load_web_page(url: str) - List[Document]: 从网页加载文档内容 try: response requests.get(url, timeout30) response.raise_for_status() soup bs4.BeautifulSoup(response.text, html.parser) # 移除脚本和样式标签 for script in soup([script, style]): script.decompose() text_content soup.get_text() # 清理多余的空白字符 lines (line.strip() for line in text_content.splitlines()) chunks (phrase.strip() for line in lines for phrase in line.split( )) text_content .join(chunk for chunk in chunks if chunk) return [Document(page_contenttext_content, metadata{source: url})] except Exception as e: print(fError loading {url}: {e}) return [] # 示例加载技术博客文章 urls [ https://lilianweng.github.io/posts/2024-11-28-reward-hacking/, https://lilianweng.github.io/posts/2024-07-07-hallucination/, https://lilianweng.github.io/posts/2024-04-12-diffusion-video/, ] # 并行加载多个文档 docs [] for url in urls: doc_chunks load_web_page(url) docs.extend(doc_chunks) print(fLoaded {len(docs)} documents)3.2 文本分割与向量化文档分割策略直接影响检索质量。过大的块会包含无关信息过小的块可能丢失上下文。def prepare_documents(documents: List[Document]) - List[Document]: 预处理文档分割和清理 # 创建文本分割器 text_splitter RecursiveCharacterTextSplitter( chunk_size500, # 每个块约500字符 chunk_overlap50, # 块间重叠50字符 length_functionlen, separators[\n\n, \n, . , ! , ? , , ] ) # 分割文档 all_splits [] for doc in documents: splits text_splitter.split_documents([doc]) # 为每个分割块添加源信息 for split in splits: split.metadata[source] doc.metadata.get(source, unknown) split.metadata[chunk_id] len(all_splits) all_splits.extend(splits) return all_splits # 处理文档 doc_splits prepare_documents(docs) print(fCreated {len(doc_splits)} document chunks)3.3 创建向量检索器from langchain_core.vectorstores import InMemoryVectorStore from langchain_openai import OpenAIEmbeddings from functools import lru_cache lru_cache(maxsize1) def create_retriever(): 创建向量检索器使用缓存避免重复创建 # 初始化嵌入模型 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) # 创建内存向量存储 vectorstore InMemoryVectorStore.from_documents( documentsdoc_splits, embeddingembeddings, ) # 配置检索器返回最相关的3个文档 retriever vectorstore.as_retriever( search_typesimilarity, search_kwargs{k: 3} ) return retriever # 测试检索器 retriever create_retriever() test_results retriever.invoke(什么是reward hacking) print(f检索到 {len(test_results)} 个相关文档)4. 构建智能检索工具现在创建核心的检索工具让LLM能够按需调用。from langchain.tools import tool from typing import Dict, Any tool def retrieve_blog_posts(query: str) - str: 根据查询检索相关的博客文章内容。 Args: query: 搜索查询字符串 Returns: 检索到的相关文档内容用换行符分隔 try: retriever create_retriever() retrieved_docs retriever.invoke(query) # 格式化返回结果 results [] for i, doc in enumerate(retrieved_docs): source doc.metadata.get(source, 未知来源) content_preview doc.page_content[:200] ... if len(doc.page_content) 200 else doc.page_content results.append(f[文档 {i1} - {source}]: {content_preview}) return \n\n.join(results) except Exception as e: return f检索过程中出现错误: {str(e)} # 测试工具 retriever_tool retrieve_blog_posts test_query reward hacking的类型 result retriever_tool.invoke({query: test_query}) print(检索工具测试结果:) print(result)5. 实现智能决策节点这是Agentic RAG的核心部分我们创建多个决策节点来实现智能工作流。5.1 查询生成与响应决策节点from langchain.chat_models import init_chat_model from langgraph.graph import MessagesState # 初始化响应模型 response_model init_chat_model(openai:gpt-4o-mini, temperature0) def generate_query_or_respond(state: MessagesState): 决策节点根据用户输入决定是检索文档还是直接响应 这个节点让LLM自主判断 - 如果问题需要外部知识调用检索工具 - 如果是一般对话直接响应 # 绑定工具让模型知道可以调用检索功能 model_with_tools response_model.bind_tools([retriever_tool]) # 基于当前对话状态生成响应 response model_with_tools.invoke(state[messages]) return {messages: [response]} # 测试决策逻辑 test_cases [ {role: user, content: 你好}, # 应该直接响应 {role: user, content: Lilian Weng关于reward hacking的观点是什么} # 应该调用检索 ] for i, test_case in enumerate(test_cases): state {messages: [test_case]} result generate_query_or_respond(state) last_msg result[messages][-1] print(f测试用例 {i1}: {test_case[content]}) if hasattr(last_msg, tool_calls) and last_msg.tool_calls: print(决策需要检索文档) print(f检索查询: {last_msg.tool_calls[0][args][query]}) else: print(决策直接响应) print(---)5.2 文档相关性评估节点检索到文档后需要评估其与问题的相关性避免基于无关内容生成答案。from pydantic import BaseModel, Field from typing import Literal class GradeDocuments(BaseModel): 文档相关性评估结果 binary_score: str Field( description相关性评分yes表示相关no表示不相关, pattern^(yes|no)$ ) # 相关性评估提示词 GRADE_PROMPT 你是一个评估检索文档与用户问题相关性的评分器。 检索到的文档内容 context {context} /context 用户问题{question} 请严格评估文档内容是否与用户问题相关。如果文档包含与问题相关的关键词或语义含义评为相关(yes)否则评为不相关(no)。 只返回二进制评分不要解释。 def grade_documents(state: MessagesState) - Literal[generate_answer, rewrite_question]: 评估检索到的文档与用户问题的相关性 Returns: generate_answer: 文档相关直接生成答案 rewrite_question: 文档不相关需要重写问题 # 初始化评估模型 grader_model init_chat_model(openai:gpt-4o-mini, temperature0) # 提取问题和上下文 question state[messages][0].content context state[messages][-1].content # 构建评估提示 prompt GRADE_PROMPT.format(questionquestion, contextcontext) # 获取结构化评估结果 response grader_model.with_structured_output(GradeDocuments).invoke( [{role: user, content: prompt}] ) # 根据评估结果路由到不同节点 if response.binary_score yes: return generate_answer return rewrite_question # 测试评估逻辑 test_contexts [ reward hacking可以分为环境错误规范和奖励篡改两种类型, # 相关 深度学习模型在图像生成中的应用, # 不相关 ] for context in test_contexts: test_state { messages: [ {role: user, content: reward hacking有哪些类型}, {role: assistant, content: , tool_calls: [{id: 1, name: retrieve_blog_posts, args: {query: reward hacking类型}}]}, {role: tool, content: context, tool_call_id: 1} ] } decision grade_documents(test_state) relevance 相关 if decision generate_answer else 不相关 print(f文档内容: {context[:50]}... → 评估结果: {relevance})5.3 问题重写节点当检索结果不相关时可能是原始问题表述不够清晰需要重写问题。from langchain.messages import HumanMessage REWRITE_PROMPT 请分析用户的原始问题并推理其背后的语义意图。 原始问题 ------- {question} ------- 当前检索到的文档与问题不相关这可能是因为 1. 问题表述不够具体 2. 使用了不准确的关键词 3. 问题范围太宽泛或太狭窄 请重新表述一个更可能检索到相关信息的问题 def rewrite_question(state: MessagesState): 重写用户问题以提高检索效果 question state[messages][0].content prompt REWRITE_PROMPT.format(questionquestion) response response_model.invoke([{role: user, content: prompt}]) # 创建重写后的问题消息 rewritten_question HumanMessage(contentresponse.content) return {messages: [rewritten_question]} # 测试问题重写 test_question 关于AI安全的一些想法 rewritten rewrite_question({messages: [{role: user, content: test_question}]}) print(f原始问题: {test_question}) print(f重写后: {rewritten[messages][-1].content})5.4 答案生成节点当文档相关时基于检索内容生成最终答案。GENERATE_PROMPT 你是一个问答助手请基于检索到的上下文回答问题。 要求 - 严格基于提供的上下文内容回答 - 如果上下文不包含答案明确说明不知道 - 保持回答简洁最多3句话 - 不要编造上下文之外的信息 问题{question} 上下文 context {context} /context 请基于以上上下文回答问题 def generate_answer(state: MessagesState): 基于相关文档生成答案 question state[messages][0].content context state[messages][-1].content prompt GENERATE_PROMPT.format(questionquestion, contextcontext) response response_model.invoke([{role: user, content: prompt}]) return {messages: [response]} # 测试答案生成 test_qa_state { messages: [ {role: user, content: reward hacking有哪些类型}, {role: assistant, content: , tool_calls: [{id: 1, name: retrieve_blog_posts, args: {query: reward hacking类型}}]}, {role: tool, content: reward hacking主要包括两种类型环境错误规范和奖励篡改。环境错误规范指智能体利用环境漏洞获得奖励奖励篡改指直接修改奖励函数。, tool_call_id: 1} ] } answer_result generate_answer(test_qa_state) print(生成的答案:) print(answer_result[messages][-1].content)6. 组装完整的工作流图现在我们将所有节点组装成完整的智能RAG工作流。from langgraph.graph import END, START, StateGraph from langgraph.prebuilt import ToolNode def create_agentic_rag_graph(): 创建完整的Agentic RAG工作流图 workflow StateGraph(MessagesState) # 添加所有节点 workflow.add_node(generate_query_or_respond, generate_query_or_respond) workflow.add_node(retrieve, ToolNode([retriever_tool])) workflow.add_node(rewrite_question, rewrite_question) workflow.add_node(generate_answer, generate_answer) # 设置起始节点 workflow.add_edge(START, generate_query_or_respond) def route_on_tool_calls(state: MessagesState): 根据工具调用情况路由 last_message state[messages][-1] if hasattr(last_message, tool_calls) and last_message.tool_calls: return retrieve return END # 决策是否需要检索 workflow.add_conditional_edges( generate_query_or_respond, route_on_tool_calls, { retrieve: retrieve, END: END } ) # 决策文档是否相关 workflow.add_conditional_edges( retrieve, grade_documents, { generate_answer: generate_answer, rewrite_question: rewrite_question } ) # 设置结束边 workflow.add_edge(generate_answer, END) # 重写问题后重新决策 workflow.add_edge(rewrite_question, generate_query_or_respond) # 编译图 graph workflow.compile() return graph # 创建并测试完整工作流 rag_graph create_agentic_rag_graph() print(Agentic RAG工作流创建成功) print(工作流节点:, list(rag_graph.nodes))7. 运行与测试智能RAG系统现在让我们测试完整的系统观察其智能决策过程。def test_agentic_rag_system(): 全面测试Agentic RAG系统 test_cases [ { name: 简单问候, input: 你好请介绍下你自己, expected: 直接响应 # 应该不需要检索 }, { name: 具体技术问题, input: Lilian Weng如何分类reward hacking的类型, expected: 检索后回答 # 应该检索相关文档 }, { name: 模糊问题, input: AI安全相关的内容, expected: 可能重写问题 # 初始检索可能不相关 } ] for i, test_case in enumerate(test_cases): print(f\n 测试用例 {i1}: {test_case[name]} ) print(f输入: {test_case[input]}) # 运行工作流 result rag_graph.invoke({ messages: [{role: user, content: test_case[input]}] }) # 分析执行路径 final_message result[messages][-1] has_tool_calls hasattr(final_message, tool_calls) and final_message.tool_calls if has_tool_calls: print(决策路径: 需要检索文档) print(f检索查询: {final_message.tool_calls[0][args][query]}) else: print(决策路径: 直接生成回答) print(f回答: {final_message.content}) print(f预期: {test_case[expected]}) print( * 50) # 运行测试 test_agentic_rag_system()7.1 流式输出实现对于实际应用流式输出能提供更好的用户体验。def stream_agentic_rag_response(question: str): 流式输出Agentic RAG的响应过程 print(f用户问题: {question}) print(处理过程:) print(- * 30) events rag_graph.stream( {messages: [{role: user, content: question}]}, config{recursion_limit: 50} ) for event in events: if generate_query_or_respond in event: print(✓ 决策节点: 分析问题意图) elif retrieve in event: print(✓ 检索节点: 搜索相关文档) elif grade_documents in event: print(✓ 评估节点: 检查文档相关性) elif rewrite_question in event: print(✓ 重写节点: 优化问题表述) elif generate_answer in event: print(✓ 生成节点: 创建最终答案) print(- * 30) final_result rag_graph.invoke( {messages: [{role: user, content: question}]} ) print(最终回答:, final_result[messages][-1].content) # 测试流式输出 stream_agentic_rag_response(请解释reward hacking的概念和类型)8. 高级功能与优化8.1 添加对话记忆让系统能够处理多轮对话记住之前的上下文。from langgraph.checkpoint.memory import MemorySaver def create_rag_with_memory(): 创建带记忆功能的RAG系统 # 添加记忆存储 memory MemorySaver() # 创建带记忆的工作流 graph_with_memory create_agentic_rag_graph() graph_with_memory graph_with_memory.compile(checkpointermemory) return graph_with_memory # 测试多轮对话 memory_graph create_rag_with_memory() def test_multi_turn_conversation(): 测试多轮对话能力 config {configurable: {thread_id: test_conversation_1}} # 第一轮 print(用户: 什么是reward hacking) result1 memory_graph.invoke( {messages: [{role: user, content: 什么是reward hacking}]}, configconfig ) print(助手:, result1[messages][-1].content) # 第二轮基于上文 print(\n用户: 能详细说说它的类型吗) result2 memory_graph.invoke( {messages: [{role: user, content: 能详细说说它的类型吗}]}, configconfig ) print(助手:, result2[messages][-1].content) test_multi_turn_conversation()8.2 性能优化建议在实际项目中这些优化能显著提升系统性能import time from functools import wraps def timing_decorator(func): 性能计时装饰器 wraps(func) def wrapper(*args, **kwargs): start_time time.time() result func(*args, **kwargs) end_time time.time() print(f{func.__name__} 执行时间: {end_time - start_time:.2f}秒) return result return wrapper # 为关键函数添加性能监控 timing_decorator def optimized_retrieve(query: str) - str: 优化版本的检索函数 # 添加缓存逻辑 # 添加批量处理 # 添加超时控制 return retrieve_blog_posts(query) # 性能优化配置建议 optimization_tips Agentic RAG性能优化建议 1. 检索优化 - 使用更高效的向量数据库Chroma、Pinecone - 实现检索结果缓存 - 调整chunk大小和重叠比例 2. 模型优化 - 根据场景选择合适大小的模型 - 实现响应流式输出 - 设置合理的超时时间 3. 系统优化 - 实现异步处理 - 添加速率限制 - 监控关键指标延迟、准确率 print(optimization_tips)9. 常见问题与解决方案在实际部署中你可能会遇到这些问题9.1 检索相关性问题问题检索到的文档与问题不相关解决方案def improve_retrieval_quality(): 提升检索质量的实用技巧 improvements 1. 优化文本分割策略 - 尝试不同的chunk_size300-1000字符 - 调整chunk_overlap10%-20% - 基于语义分割而非固定长度 2. 改进检索参数 - 调整返回文档数量k值 - 尝试混合检索语义关键词 - 使用重排序re-ranking技术 3. 增强查询理解 - 实现查询扩展 - 添加同义词处理 - 考虑多语言支持 return improvements print(improve_retrieval_quality())9.2 错误处理与容错def robust_rag_invocation(question: str, max_retries: int 3): 带错误处理的RAG调用 for attempt in range(max_retries): try: result rag_graph.invoke( {messages: [{role: user, content: question}]} ) return result except Exception as e: print(f第{attempt 1}次尝试失败: {str(e)}) if attempt max_retries - 1: # 最后一次尝试失败返回降级响应 return {messages: [{ role: assistant, content: 抱歉系统暂时无法处理您的请求。请稍后再试或简化您的问题。 }]} time.sleep(1) # 重试前等待 # 测试容错能力 result robust_rag_invocation(test question) print(容错测试结果:, result[messages][-1].content)10. 生产环境部署建议当准备将Agentic RAG系统部署到生产环境时考虑以下关键因素10.1 安全考虑def security_considerations(): 生产环境安全注意事项 security_checklist 安全清单 1. API密钥管理 - 使用环境变量或密钥管理服务 - 定期轮换密钥 - 设置最小权限原则 2. 输入验证 - 验证用户输入长度和内容 - 防止提示词注入攻击 - 实现速率限制 3. 输出过滤 - 检查模型输出是否包含敏感信息 - 实现内容审核机制 - 添加滥用检测 4. 数据隐私 - 匿名化用户数据 - 遵守数据保护法规 - 定期清理日志 return security_checklist print(security_considerations())10.2 监控与日志def setup_monitoring(): 设置系统监控 monitoring_config 监控配置建议 1. 关键指标监控 - 请求延迟P50、P95、P99 - 错误率和成功率 - Token使用量统计 2. 业务指标 - 检索相关性评分 - 用户满意度反馈 - 对话完成率 3. 日志策略 - 结构化日志记录 - 敏感信息脱敏 - 日志保留策略 return monitoring_config print(setup_monitoring())通过本文的完整实现你已经构建了一个真正智能的Agentic RAG系统。关键在于理解每个决策节点的作用以及如何将它们组合成有效的工作流。这种架构不仅提升了回答质量还显著优化了资源使用效率。实际项目中建议从简单版本开始逐步添加复杂功能并持续基于用户反馈进行优化。智能RAG系统的价值在于它能随着使用不断改进真正理解用户需求并提供精准的知识服务。