AI 新兴范式：Agentic RAG 技术详解

目录一、检索增强生成RAG简介二、为什么使用 RAG三、传统 RAG 的局限性四、Agentic RAG更强大的方法什么是 Agentic RAGAgentic RAG 的主要优势五、Agentic RAG 完整实现环境安装步骤 1预处理文档步骤 2创建检索工具步骤 3生成查询节点步骤 4对检索结果进行评分步骤 5问题重写节点步骤 6生成最终答案步骤 7构建状态图步骤 8运行 Agentic RAG六、总结本文深入介绍 Agentic RAG 的核心概念、优势与完整实现方案涵盖从文档预处理到状态图构建的全流程代码。一、检索增强生成RAG简介检索增强生成Retrieval-Augmented GenerationRAG将大型语言模型的强大功能与外部知识检索相结合从而生成更准确、更贴近事实、更符合语境的响应。RAG 的核心理念使用 LLM 回答用户查询但答案基于从知识库检索到的信息。二、为什么使用 RAG与直接使用原始或微调的 LLM 相比RAG 具有以下显著优势优势说明事实基础将响应锚定在检索到的事实中减少幻觉领域专业化无需模型再训练即可提供特定领域知识知识新近度允许访问超出模型训练截止时间的最新信息透明度允许引用生成内容的信息来源可控性对模型可访问的信息进行细粒度控制三、传统 RAG 的局限性尽管传统 RAG 优势明显但仍面临以下挑战单次检索步骤初始检索结果质量差时最终生成结果也会受到影响查询-文档不匹配用户问题通常是疑问句与包含答案的文档通常是陈述句语义上不够匹配推理能力有限简单的 RAG 流程无法进行多步推理或查询细化上下文窗口约束检索到的文档内容必须适合模型的上下文窗口大小四、Agentic RAG更强大的方法什么是 Agentic RAGAgentic RAG是一种新兴的 AI 范式其中大型语言模型LLM自主规划下一步行动同时从外部资源提取信息。与静态的先检索后阅读模式不同Agentic RAG 的核心特征对 LLM 进行迭代调用穿插使用工具或函数调用并输出结构化结果系统评估结果、优化查询必要时调用更多工具持续循环直到获得满意答案采用迭代的制造者-检查者模式提升正确性处理格式错误的查询确保结果质量Agentic 系统能够主动重写失败的查询、选择不同的检索方式、整合多种工具向量搜索、SQL 数据库、自定义 API 等最终再给出答案。Agentic RAG 的主要优势能力说明制定优化查询将用户问题转换为易于检索的查询语句执行多次检索根据需要迭代检索信息多源推理从多个来源分析、综合并得出结论自我批评与改进评估检索结果并动态调整检索策略这种方法天然实现了以下先进 RAG 技术Hypothetical Document Embedding (HyDE)代理不直接使用用户查询而是制定检索优化查询Self-Query Refinement代理分析初始检索结果并使用细化查询执行后续检索五、Agentic RAG 完整实现下面以 LangGraph 通义千问 为例展示 Agentic RAG 系统的完整构建过程。环境安装pip install -U --quiet langgraph langchain[openai] langchain-community langchain-text-splitters提示在 Jupyter Notebook 中建议使用%pip install魔法命令而非!pip install或直接写pip install以确保安装到当前 Notebook 所在的 Python 环境。步骤 1预处理文档从网页加载文档并使用RecursiveCharacterTextSplitter进行分块import os from langchain_community.document_loaders import WebBaseLoader from langchain_text_splitters import RecursiveCharacterTextSplitter ​ urls [ https://baike.baidu.com/item/大语言模型/62884793, https://baike.baidu.com/item/深度学习/3729729, https://baike.baidu.com/item/人工智能/9180, https://sports.sina.com.cn/ ] ​ # 加载文档 docs [WebBaseLoader(url).load() for url in urls] docs_list [item for sublist in docs for item in sublist] ​ # 分块处理 text_splitter RecursiveCharacterTextSplitter.from_tiktoken_encoder( chunk_size100, chunk_overlap50 ) doc_splits text_splitter.split_documents(docs_list)步骤 2创建检索工具使用向量存储构建检索器并将其封装为 LangChain 工具from langchain_core.vectorstores import InMemoryVectorStore from langchain_community.embeddings import DashScopeEmbeddings from langchain.tools.retriever import create_retriever_tool ​ # 构建向量存储与检索器 vectorstore InMemoryVectorStore.from_documents( documentsdoc_splits, embeddingDashScopeEmbeddings(modeltext-embedding-v1) ) retriever vectorstore.as_retriever() ​ # 封装为工具 retriever_tool create_retriever_tool( retriever, 百度百科, # 工具名称 搜索百度或新浪体育上的信息. # 工具描述 )步骤 3生成查询节点模型决策节点根据用户输入决定是直接回答还是调用检索工具from langgraph.graph import MessagesState from langchain_community.chat_models import ChatTongyi ​ response_model ChatTongyi( modelqwen-plus, temperature0, api_keyos.getenv(DASHESCOPE_API_KEY) ) ​ def generate_query_or_respond(state: MessagesState): 根据当前状态生成响应直接回答用户或调用检索工具获取信息。 response ( response_model.bind_tools([retriever_tool]).invoke(state[messages]) ) return {messages: [response]}步骤 4对检索结果进行评分使用结构化输出模型对检索文档与问题的相关性进行二元评分yes/nofrom pydantic import BaseModel, Field from typing import Literal ​ GRADE_PROMPT ( You are a grader assessing relevance of a retrieved document to a user question. \n Here is the retrieved document: \n\n {context} \n\n Here is the user question: {question} \n If the document contains keyword(s) or semantic meaning related to the user question, grade it as relevant. \n Give a binary score yes or no score to indicate whether the document is relevant. ) ​ class GradeDocuments(BaseModel): 文档相关性评分二元 binary_score: str Field( descriptionRelevance score: yes if relevant, or no if not relevant ) ​ grader_model ChatTongyi(modelqwen-plus, temperature0, api_keyos.getenv(DASHESCOPE_API_KEY)) ​ def grade_documents(state: MessagesState) - Literal[generate_answer, rewrite_question]: 判断检索到的文档是否与问题相关决定下一步走向。 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}] ) return generate_answer if response.binary_score yes else rewrite_question步骤 5问题重写节点当检索结果不相关时对原始问题进行语义重写优化后重新检索REWRITE_PROMPT ( Look at the input and try to reason about the underlying semantic intent / meaning.\n Here is the initial question:\n ------- \n{question}\n ------- \n Formulate an improved question: ) ​ def rewrite_question(state: MessagesState): 重写原始用户问题提升检索质量。 question state[messages][0].content prompt REWRITE_PROMPT.format(questionquestion) response response_model.invoke([{role: user, content: prompt}]) return {messages: [{role: user, content: response.content}]}步骤 6生成最终答案基于原始问题与检索到的上下文生成简洁准确的最终答案GENERATE_PROMPT ( You are an assistant for question-answering tasks. Use the following pieces of retrieved context to answer the question. If you dont know the answer, just say that you dont know. Use three sentences maximum and keep the answer concise.\n Question: {question} \n 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]}步骤 7构建状态图使用 LangGraph 的StateGraph将所有节点与边组合成完整的 Agentic RAG 工作流from langgraph.graph import StateGraph, START, END from langgraph.prebuilt import ToolNode, tools_condition ​ workflow StateGraph(MessagesState) ​ # 注册节点 workflow.add_node(generate_query_or_respond) workflow.add_node(retrieve, ToolNode([retriever_tool])) workflow.add_node(rewrite_question) workflow.add_node(generate_answer) ​ # 入口从 generate_query_or_respond 开始 workflow.add_edge(START, generate_query_or_respond) ​ # 条件边决定是检索还是直接回答 workflow.add_conditional_edges( generate_query_or_respond, tools_condition, {tools: 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) ​ # 编译 graph workflow.compile()整体工作流示意如下START └─► generate_query_or_respond ├─► END无需检索直接回答 └─► retrieve调用检索工具 ├─► generate_answer ──► END文档相关 └─► rewrite_question ──► generate_query_or_respond文档不相关重写后重试步骤 8运行 Agentic RAGfor chunk in graph.stream( { messages: [ {role: user, content: 国足刚刚赢比赛的比分是多少?} ] } ): for node, update in chunk.items(): print(Update from node, node) update[messages][-1].pretty_print() print(\n\n)系统会根据问题自动决策若知识库中有相关内容则直接检索回答若检索结果不相关则重写查询后重试若完全无需检索则直接响应。六、总结Agentic RAG 通过引入自主决策、迭代检索、结果评估等能力将传统 RAG 的单次检索流程升级为智能推理驱动的动态循环。对比维度传统 RAGAgentic RAG检索次数单次多轮迭代查询优化无自动重写结果评估无相关性评分推理能力弱多步推理工具集成单一多工具协同随着 LangGraph 等框架的成熟构建生产级 Agentic RAG 系统正变得越来越便捷。这一范式将成为企业 AI 应用落地的重要基础架构。本文基于 LangGraph 通义千问Qwen DashScope Embeddings 实现适用于中文场景下的知识库问答系统构建。