如果你正在学习大模型应用开发可能会发现一个奇怪的现象很多教程都在教你怎么用 LangChain 搭建 RAG 系统但实际工作中遇到的却是更复杂的需求——用户的问题可能根本不需要检索或者检索到的内容质量参差不齐甚至需要多轮对话才能理清真实意图。这正是传统 RAG 系统的局限性所在。而 LangGraph 的出现让构建能够自主决策的智能体Agent成为可能。与简单调用检索工具不同基于 LangGraph 的 Agentic RAG 能够判断何时检索、何时直接回答、何时重新提问真正实现了智能化的问答流程。本文将带你从零搭建一个企业级的 AgentRAGLangGraph 项目重点解决三个核心问题如何让系统自主决定检索时机如何评估检索内容的相关性以及如何通过多轮交互优化问答质量。相比传统方案这种架构能够减少不必要的检索开销提升回答准确率更适合实际业务场景。1. 这篇文章真正要解决的问题在企业级大模型应用开发中开发者最常遇到的痛点不是技术实现而是系统智能化程度的不足。传统的 RAG 系统往往存在以下问题无差别检索无论用户问题是否需要外部知识都会执行检索操作增加延迟和成本相关性判断缺失检索到内容后直接生成答案缺乏对内容相关性的评估机制问题理解固化无法根据检索结果调整对用户问题的理解缺乏迭代优化能力LangGraph 通过图结构的工作流设计让开发者能够构建具有条件判断和循环逻辑的智能体系统。与 LangChain 相比LangGraph 更专注于复杂工作流的编排特别适合需要多步骤决策的场景。本文将基于最新的 2026 年技术栈演示如何构建一个真正实用的 Agentic RAG 系统重点解决上述三个核心痛点。2. 基础概念与核心原理2.1 LangChain 与 LangGraph 的关系很多初学者容易混淆 LangChain 和 LangGraph 的定位这里需要明确LangChain是一个完整的 LLM 应用开发框架提供了丰富的组件库和工具链包括文档加载、向量存储、链式调用等LangGraph是 LangChain 生态系统中的工作流编排工具专注于构建有状态的多步骤应用程序简单来说LangChain 提供了建筑材料而 LangGraph 提供了建筑设计图。在实际项目中两者通常结合使用。2.2 Agentic RAG 的核心思想传统 RAG 是线性的用户问题 → 检索 → 生成答案。而 Agentic RAG 引入了决策逻辑决策层LLM 判断是否需要检索还是直接回答评估层对检索结果进行相关性评估优化层根据评估结果决定是生成答案还是重新表述问题这种架构的优势在于能够根据上下文动态调整策略避免不必要的操作提升系统效率。2.3 图工作流的基本概念LangGraph 的核心是图结构包含以下几个关键元素节点Node执行特定功能的函数如生成查询、检索文档、评估相关性等边Edge定义节点之间的流转关系条件边Conditional Edge根据当前状态决定下一步执行哪个节点状态State在整个工作流中传递的共享数据3. 环境准备与前置条件在开始实战之前需要确保开发环境准备就绪。以下是基于 2026 年最新技术栈的推荐配置3.1 系统要求Python 3.9 或更高版本8GB 以上内存用于运行向量数据库和大型语言模型稳定的网络连接用于访问 API 服务3.2 安装依赖包# 核心 LangGraph 相关包 pip install -U langgraph langchain-anthropic langchain-text-splitters # 网页内容处理 pip install bs4 requests # 向量存储和嵌入根据需求选择 pip install langchain-openai chromadb # 开发工具 pip install jupyter ipython3.3 API 密钥配置在实际项目中建议使用环境变量管理敏感信息import getpass import os def setup_environment(): 设置必要的环境变量 required_keys { OPENAI_API_KEY: OpenAI API密钥, ANTHROPIC_API_KEY: Claude API密钥可选 } for key, description in required_keys.items(): if key not in os.environ: print(f请输入{description}) os.environ[key] getpass.getpass(f{key}: ) # 执行环境设置 setup_environment()3.4 开发环境验证创建一个简单的测试脚本来验证环境配置# test_environment.py from langchain.chat_models import init_chat_model def test_basic_functionality(): 测试基础功能是否正常 try: # 测试模型初始化 model init_chat_model(openai:gpt-4o-mini, temperature0) response model.invoke(Hello, world!) print(✅ 环境配置成功) print(f测试响应: {response.content}) return True except Exception as e: print(f❌ 环境配置失败: {e}) return False if __name__ __main__: test_basic_functionality()4. 核心流程拆解构建 Agentic RAG 系统的核心在于设计合理的工作流程。我们将整个系统拆解为 8 个关键步骤每个步骤都有明确的职责和输入输出。4.1 文档预处理流程文档预处理是 RAG 系统的基础质量直接影响到后续检索效果import bs4 import requests from langchain_core.documents import Document from langchain_text_splitters import RecursiveCharacterTextSplitter def load_and_process_documents(urls): 加载并处理网页文档 def load_web_page(url: str) - list[Document]: 加载单个网页内容 response requests.get(url, timeout20) response.raise_for_status() soup bs4.BeautifulSoup(response.text, html.parser) return [Document(page_contentsoup.get_text(), metadata{source: url})] # 并行加载所有文档实际项目中建议添加错误处理 docs [load_web_page(url) for url in urls] docs_list [item for sublist in docs for item in sublist] # 文本分割优化策略 text_splitter RecursiveCharacterTextSplitter.from_tiktoken_encoder( chunk_size500, # 根据实际内容调整 chunk_overlap100, # 重叠部分保证上下文连贯性 separators[\n\n, \n, 。, , , , , ] # 中文友好分割 ) return text_splitter.split_documents(docs_list)4.2 检索工具创建检索工具是 Agent 与知识库交互的桥梁from langchain_core.vectorstores import InMemoryVectorStore from langchain_openai import OpenAIEmbeddings from langchain.tools import tool from functools import lru_cache class KnowledgeRetriever: 知识检索器封装类 def __init__(self, documents): self.documents documents self._vectorstore None self._retriever None def initialize(self): 初始化向量存储和检索器 self._vectorstore InMemoryVectorStore.from_documents( documentsself.documents, embeddingOpenAIEmbeddings() ) self._retriever self._vectorstore.as_retriever( search_typesimilarity, search_kwargs{k: 3} # 返回最相关的3个文档 ) property def retriever(self): if self._retriever is None: self.initialize() return self._retriever # 创建检索工具 tool def retrieve_knowledge(query: str) - str: 从知识库中检索相关信息。 Args: query: 检索查询语句 Returns: str: 检索到的相关内容多个文档用换行符分隔 # 在实际项目中这里应该使用单例模式或依赖注入 retriever get_global_retriever() retrieved_docs retriever.invoke(query) # 格式化返回结果包含来源信息 results [] for i, doc in enumerate(retrieved_docs): content doc.page_content[:500] # 限制长度 source doc.metadata.get(source, 未知来源) results.append(f[文档{i1} - {source}]: {content}) return \n\n.join(results) # 全局检索器实例 _global_retriever None def get_global_retriever(): 获取全局检索器实例 global _global_retriever if _global_retriever is None: # 这里需要先初始化文档 urls [ https://lilianweng.github.io/posts/2024-11-28-reward-hacking/, https://lilianweng.github.io/posts/2024-07-07-hallucination/, ] documents load_and_process_documents(urls) _global_retriever KnowledgeRetriever(documents) _global_retriever.initialize() return _global_retriever.retriever5. 完整示例与代码实现现在我们来构建完整的 Agentic RAG 系统。这个系统包含决策、检索、评估、优化四个核心模块。5.1 状态定义与模型初始化首先定义工作流的状态和使用的语言模型from langgraph.graph import MessagesState from langchain.chat_models import init_chat_model from langchain_core.messages import HumanMessage, AIMessage, ToolMessage # 初始化响应模型决策核心 response_model init_chat_model(openai:gpt-4o-mini, temperature0) # 初始化评估模型专门用于相关性判断 grader_model init_chat_model(openai:gpt-4o-mini, temperature0)5.2 决策节点实现决策节点负责判断是否需要检索def generate_query_or_respond(state: MessagesState): 决策节点根据当前对话状态决定是否需要检索 这个节点是工作流的入口它会让LLM判断 1. 直接回答用户问题如果问题简单或不需要外部知识 2. 调用检索工具获取相关信息 # 绑定检索工具让模型知道可以调用这个工具 model_with_tools response_model.bind_tools([retrieve_knowledge]) # 基于当前对话历史生成响应 response model_with_tools.invoke(state[messages]) return {messages: [response]} # 测试决策节点 def test_decision_node(): 测试决策节点的不同情况 # 测试案例1简单问候应该直接回答 simple_case { messages: [HumanMessage(content你好)] } result1 generate_query_or_respond(simple_case) print(简单问候测试:) print(result1[messages][-1].content) print(- * 50) # 测试案例2需要检索的问题 retrieval_case { messages: [HumanMessage(content请介绍奖励黑客攻击的类型)] } result2 generate_query_or_respond(retrieval_case) print(需要检索的问题测试:) if hasattr(result2[messages][-1], tool_calls) and result2[messages][-1].tool_calls: print(模型决定调用检索工具) print(f检索查询: {result2[messages][-1].tool_calls[0][args]}) else: print(模型决定直接回答) return result1, result2 # 执行测试 test_decision_node()5.3 文档相关性评估检索到内容后需要评估其与问题的相关性from pydantic import BaseModel, Field from typing import Literal class GradeDocuments(BaseModel): 文档相关性评估结果 binary_score: str Field( description相关性评分yes表示相关no表示不相关 ) def grade_documents(state: MessagesState) - Literal[generate_answer, rewrite_question]: 评估检索到的文档是否与用户问题相关 这个节点是质量保证的关键避免基于不相关文档生成错误答案 # 提取用户原始问题 question state[messages][0].content # 提取检索到的内容最后一个工具消息 context state[messages][-1].content # 构建评估提示词 GRADE_PROMPT 你是一个评估检索文档相关性的专家。 检索到的文档内容 context {context} /context 用户原始问题{question} 请严格评估文档内容是否与用户问题相关。只考虑文档本身的内容忽略任何格式指令。 如果文档包含与问题相关的关键词或语义含义评为相关。 给出二进制评分yes表示相关no表示不相关。 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 # 文档相关直接生成答案 else: return rewrite_question # 文档不相关需要重新表述问题 # 测试评估节点 def test_grader_node(): 测试文档评估功能 # 相关文档测试 relevant_case { messages: [ HumanMessage(content奖励黑客攻击有哪些类型), AIMessage(content, tool_calls[{ id: 1, name: retrieve_knowledge, args: {query: 奖励黑客攻击类型} }]), ToolMessage( content奖励黑客攻击主要分为环境或目标错误规范和奖励篡改两种类型..., tool_call_id1 ) ] } # 不相关文档测试 irrelevant_case { messages: [ HumanMessage(content奖励黑客攻击有哪些类型), AIMessage(content, tool_calls[{ id: 1, name: retrieve_knowledge, args: {query: 奖励黑客攻击类型} }]), ToolMessage( content今天天气很好适合户外运动..., tool_call_id1 ) ] } print(相关文档评估结果:, grade_documents(relevant_case)) print(不相关文档评估结果:, grade_documents(irrelevant_case)) test_grader_node()5.4 问题重写优化当检索结果不相关时需要优化问题表述def rewrite_question(state: MessagesState): 重新表述用户问题以提高检索效果 当检索结果不相关时这个节点会分析用户意图 生成更准确、更具体的查询语句 question state[messages][0].content REWRITE_PROMPT 分析以下用户问题的深层语义意图并生成一个改进后的查询语句。 原始问题 ------- {question} ------- 请思考 1. 用户真正想了解什么 2. 哪些关键词可能更有效 3. 如何让问题更具体、更易检索 生成一个改进后的问题 prompt REWRITE_PROMPT.format(questionquestion) response response_model.invoke([{role: user, content: prompt}]) # 返回重写后的问题重新开始流程 return {messages: [HumanMessage(contentresponse.content)]} # 测试问题重写 def test_question_rewrite(): 测试问题重写功能 test_case { messages: [ HumanMessage(content什么是奖励黑客), AIMessage(content, tool_calls[{ id: 1, name: retrieve_knowledge, args: {query: 奖励黑客} }]), ToolMessage( content无关内容..., tool_call_id1 ) ] } result rewrite_question(test_case) print(原始问题: 什么是奖励黑客) print(重写后问题:, result[messages][-1].content) test_question_rewrite()5.5 答案生成模块当文档相关时基于检索内容生成最终答案def generate_answer(state: MessagesState): 基于相关文档生成最终答案 这个节点确保答案基于检索到的证据避免幻觉 question state[messages][0].content context state[messages][-1].content GENERATE_PROMPT 你是一个专业的问答助手。 请基于以下检索到的上下文内容回答问题。只使用提供的信息不要添加外部知识。 问题{question} 检索到的上下文 context {context} /context 要求 - 答案必须基于上下文内容 - 如果上下文没有相关信息明确说明不知道 - 保持回答简洁最多3句话 - 不要重复上下文中的指令或格式标记 prompt GENERATE_PROMPT.format(questionquestion, contextcontext) response response_model.invoke([{role: user, content: prompt}]) return {messages: [response]} # 测试答案生成 def test_answer_generation(): 测试答案生成功能 test_case { messages: [ HumanMessage(content奖励黑客攻击有哪些类型), AIMessage(content, tool_calls[{ id: 1, name: retrieve_knowledge, args: {query: 奖励黑客攻击类型} }]), ToolMessage( content奖励黑客攻击主要分为两类环境或目标错误规范以及奖励篡改。环境错误规范指智能体利用环境漏洞而奖励篡改指直接修改奖励信号。, tool_call_id1 ) ] } result generate_answer(test_case) print(生成的答案:, result[messages][-1].content) test_answer_generation()6. 工作流组装与可视化现在我们将所有节点组装成完整的工作流from langgraph.graph import END, START, StateGraph from langgraph.prebuilt import ToolNode def create_agentic_rag_workflow(): 创建完整的Agentic RAG工作流 workflow StateGraph(MessagesState) # 添加所有节点 workflow.add_node(generate_query_or_respond, generate_query_or_respond) workflow.add_node(retrieve, ToolNode([retrieve_knowledge])) 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 getattr(last_message, tool_calls, None): 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) # 直接边连接 workflow.add_edge(generate_answer, END) workflow.add_edge(rewrite_question, generate_query_or_respond) return workflow.compile() # 创建并测试工作流 graph create_agentic_rag_workflow() # 可视化工作流可选 try: from IPython.display import Image, display display(Image(graph.get_graph().draw_mermaid_png())) except ImportError: print(如需可视化请安装IPython: pip install ipython)7. 运行结果与效果验证7.1 完整系统测试现在让我们测试完整的 Agentic RAG 系统def run_complete_workflow(question: str): 运行完整的Agentic RAG工作流 Args: question: 用户问题 Returns: 系统最终回答 # 准备输入 inputs {messages: [{role: user, content: question}]} # 执行工作流 final_state graph.invoke(inputs) # 提取最终回答 final_message final_state[messages][-1] print( * 60) print(f用户问题: {question}) print( * 60) # 显示工作流执行过程 print(工作流执行过程:) for i, msg in enumerate(final_state[messages]): role msg.__class__.__name__ if hasattr(msg, tool_calls) and msg.tool_calls: print(f{i}. {role}: 调用工具 {msg.tool_calls[0][name]}) elif hasattr(msg, content): content_preview msg.content[:100] ... if len(msg.content) 100 else msg.content print(f{i}. {role}: {content_preview}) print( * 60) print(最终答案:, final_message.content) return final_message.content # 测试不同场景 def comprehensive_test(): 综合测试不同场景 test_cases [ 你好, # 简单问候应该直接回答 奖励黑客攻击有哪些类型, # 需要检索的专业问题 请详细解释环境错误规范的具体例子, # 可能需要进一步澄清的问题 ] for i, question in enumerate(test_cases, 1): print(f\n测试案例 {i}:) run_complete_workflow(question) print(\n *80 \n) # 执行测试 comprehensive_test()7.2 性能评估指标在实际项目中还需要建立评估体系class RAGEvaluator: RAG系统评估器 staticmethod def evaluate_response(question: str, answer: str, expected_keywords: list None): 评估回答质量 Args: question: 原始问题 answer: 系统回答 expected_keywords: 期望包含的关键词列表 evaluation { question: question, answer_length: len(answer), has_uncertainty: 不知道 in answer or 不确定 in answer, is_concise: len(answer) 200, # 是否简洁 } if expected_keywords: evaluation[keyword_coverage] sum(1 for keyword in expected_keywords if keyword in answer) / len(expected_keywords) return evaluation # 示例评估 def sample_evaluation(): 示例评估 question 奖励黑客攻击有哪些类型 answer 奖励黑客攻击主要分为环境或目标错误规范和奖励篡改两种类型。 evaluator RAGEvaluator() result evaluator.evaluate_response( question, answer, expected_keywords[环境, 目标错误规范, 奖励篡改] ) print(评估结果:) for key, value in result.items(): print(f{key}: {value}) sample_evaluation()8. 常见问题与排查思路在实际部署过程中可能会遇到各种问题。以下是常见问题及解决方案问题现象可能原因排查方式解决方案工作流卡在决策节点模型无法正确判断是否需要检索检查决策节点的输入输出日志优化提示词明确检索条件检索结果始终不相关文档预处理质量差或查询表述不佳检查文档分割质量和检索查询优化文本分割参数添加问题重写机制评估节点误判率高评估标准不明确或提示词有问题分析评估节点的判断依据细化评估标准增加示例训练系统响应速度慢过多的检索操作或模型调用分析工作流执行路径添加缓存机制优化节点执行顺序答案质量不稳定生成提示词约束不足检查生成节点的提示词设计加强答案约束添加后处理校验8.1 调试技巧def debug_workflow(question: str): 工作流调试工具 print(开始调试工作流...) inputs {messages: [{role: user, content: question}]} # 使用流式执行观察每个步骤 for step, output in graph.stream(inputs): print(f\n步骤: {step}) last_message output[messages][-1] if hasattr(last_message, tool_calls) and last_message.tool_calls: print(f工具调用: {last_message.tool_calls[0][name]}) print(f参数: {last_message.tool_calls[0][args]}) else: content_preview last_message.content[:150] ... if len(last_message.content) 150 else last_message.content print(f内容: {content_preview}) # 示例调试 debug_workflow(什么是奖励黑客攻击)9. 最佳实践与工程建议基于实际项目经验以下是构建生产级 Agentic RAG 系统的最佳实践9.1 性能优化策略class PerformanceOptimizer: 性能优化工具类 staticmethod def add_caching_to_retriever(retriever, max_size: int 1000): 为检索器添加缓存机制 from functools import lru_cache original_invoke retriever.invoke lru_cache(maxsizemax_size) def cached_invoke(query: str): return original_invoke(query) retriever.invoke cached_invoke return retriever staticmethod def optimize_prompt_length(prompt: str, max_tokens: int 2000) - str: 优化提示词长度 if len(prompt) max_tokens * 4: # 粗略估算 # 截断策略保留开头和关键部分 segments prompt.split(\n\n) if len(segments) 3: optimized \n\n.join(segments[:2] [...] segments[-1:]) return optimized return prompt9.2 错误处理与容错机制import logging from typing import Dict, Any class ErrorHandler: 错误处理管理器 def __init__(self): self.logger logging.getLogger(__name__) def safe_node_execution(self, node_func, state: Dict[str, Any], max_retries: int 3): 安全执行节点函数包含重试机制 for attempt in range(max_retries): try: result node_func(state) return result except Exception as e: self.logger.warning(f节点执行失败 (尝试 {attempt 1}/{max_retries}): {e}) if attempt max_retries - 1: # 最后一次尝试失败返回降级响应 return self._get_fallback_response(state) return self._get_fallback_response(state) def _get_fallback_response(self, state: Dict[str, Any]): 获取降级响应 from langchain_core.messages import AIMessage return { messages: [ AIMessage(content抱歉系统暂时无法处理您的请求请稍后重试。) ] } # 使用错误处理包装节点函数 error_handler ErrorHandler() safe_generate_query_or_respond lambda state: error_handler.safe_node_execution(generate_query_or_respond, state)9.3 监控与日志记录import json from datetime import datetime class WorkflowMonitor: 工作流监控器 def __init__(self, log_file: str workflow_monitor.log): self.log_file log_file self.logger logging.getLogger(workflow_monitor) self.setup_logging() def setup_logging(self): 设置日志记录 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(self.log_file), logging.StreamHandler() ] ) def log_execution(self, question: str, workflow_path: list, final_answer: str): 记录工作流执行过程 execution_record { timestamp: datetime.now().isoformat(), question: question, workflow_path: workflow_path, final_answer: final_answer, answer_length: len(final_answer) } self.logger.info(f工作流执行记录: {json.dumps(execution_record, ensure_asciiFalse)}) return execution_record # 使用监控器 monitor WorkflowMonitor()本文构建的 Agentic RAG 系统相比传统方案具有明显的智能化优势能够自主决策检索时机、评估内容质量、优化问题表述。在实际业务场景中这种架构可以显著提升问答准确率和用户体验。建议读者根据具体业务需求调整工作流节点和评估标准特别是在文档预处理、检索策略和答案生成等关键环节进行定制化优化。对于高并发场景还需要考虑缓存、异步处理和负载均衡等工程化问题。