当前位置：首页 > news >正文

从Naive RAG到Agentic RAG：智能检索增强生成的演进与实战

news 2026/8/14 12:36:36

1. 从“查字典”到“找专家”：RAG技术的演进脉络

如果你最近在折腾大语言模型应用，那“RAG”这个词肯定已经听得耳朵起茧了。它就像AI应用开发里的“水电煤”，成了构建靠谱智能系统的标配。但很多人对RAG的理解，可能还停留在“把文档切碎、变成向量、存起来、然后搜一下”这个层面。这没错，这是最基础的玩法，业内通常称之为Naive RAG，或者叫“朴素RAG”。它就像一个老实巴交的图书管理员，你问一个问题，它就去庞大的向量书库里，找到最“像”你问题的几段文字，然后原封不动地递给大模型，让模型基于这些片段生成答案。

这个模式解决了大模型“胡说八道”（幻觉）和知识陈旧的核心痛点，立下了汗马功劳。但用久了你会发现，这位“朴素”的图书管理员有点轴。比如，你问“公司最新的差旅报销政策有什么变化？”，它可能会返回给你三份文档的片段：一份是两年前的旧政策全文，一份是今年新政策的第三章，还有一份是财务部发的关于发票类型的通知。模型拿到这些信息后，需要自己费力地去拼凑、对比、理解“变化”在哪里，效果很不稳定。整个过程是单向、静态的，检索和生成是割裂的两步。

于是，大家开始思考，能不能让这个系统变得更“智能”一点？能不能让检索过程本身也具备思考、判断和决策的能力？这就是Agentic RAG（智能体驱动的RAG）诞生的背景。它不再是那个被动的图书管理员，而更像一个拥有专业经验的“领域专家”或“侦探”。你提出一个问题，这位专家会主动分析你的意图，拆解任务，决定去哪里找资料、用什么方式找、找到后如何验证和整合，甚至会在信息不足时，主动向你追问细节。Agentic RAG的核心，是将大型语言模型本身的推理和规划能力，注入到检索增强生成的每一个环节，让“检索”成为一个动态、迭代、可决策的智能过程。

从Naive RAG到Agentic RAG，不是简单的版本升级，而是一次设计范式的跃迁。前者关注的是“信息匹配”的精度和效率，后者追求的是“任务达成”的可靠性和智能化。接下来，我们就深入这个智能检索系统的内部，看看它是如何一步步“进化”的。

2. Naive RAG：经典架构的基石与局限性

在深入Agentic的复杂世界之前，我们必须先扎实理解Naive RAG。它是所有高级RAG系统的地基，其设计中的优缺点直接催生了后续的进化方向。一个典型的Naive RAG系统，可以清晰地分为三个核心阶段：索引、检索、生成。

2.1 核心流程的三段论

索引阶段的目标是把非结构化的原始知识（如PDF、Word、网页、数据库），加工成便于检索的格式。这个过程远不止是“切碎”那么简单。

文档加载与解析：使用像Unstructured、PyPDF2、Docling这样的库，把各种格式的文件转换成纯文本。这里第一个坑就来了：格式丢失。PDF里的精美表格、图片里的文字、扫描件，处理不好就会变成乱码或直接缺失。我的经验是，对于复杂文档，最好采用“组合拳”，比如用pdfplumber提取表格，用pytesseract做OCR识别图片，再自己写规则做信息缝合。
文本分割：这是影响效果最关键的步骤之一。你不能简单按固定字符数（比如512个token）来切。想象一下，一个完整的操作步骤被拦腰切断，或者一个定义句被分在两段，检索效果会大打折扣。常用的策略有：
- 递归字符分割：按段落、句子、换行符等自然分隔符进行递归分割，尽量保证语义完整性。
- 滑动窗口：在固定大小的块上使用重叠窗口，避免信息在边界丢失。比如块大小500token，重叠100token。
- 基于语义的分割：用更小的模型（如sentence-transformers）计算句子间的相似度，在语义变化大的地方进行分割。这更智能，但计算成本也高。
- 我的实操心得：没有银弹。我通常会对文档类型做分析。技术手册适合按章节/子标题分割；会议纪要适合按议题分割；长篇文章可以尝试滑动窗口。一个黄金法则是：分割后的块，应该能够独立回答一个潜在的问题。你可以用一些假想问题来检验。
向量化与存储：将文本块通过嵌入模型转化为向量，存入向量数据库。选型是关键：
- 嵌入模型：别只盯着OpenAI的text-embedding-ada-002。开源模型如BGE-M3、Snowflake Arctic Embed、Nomic Embed在MTEB等基准测试上表现非常亮眼，且数据隐私可控。选择时一定要用你自己领域的数据做测试，看模型在你关心的任务（如相似性搜索、检索）上的表现。
- 向量数据库：Pinecone、Weaviate、Qdrant、Milvus是主流选择。对于快速原型，ChromaDB轻量易用；对于生产级海量数据和高吞吐，Milvus或Qdrant更合适。需要关注其是否支持过滤（metadata filtering）、多向量搜索等高级功能。

检索阶段是查询发生时的工作。系统将用户问题同样向量化，然后在向量库中进行相似性搜索（通常是余弦相似度或点积），返回Top-K个最相关的文本块。

注意：这里最大的误区是认为“相似度最高=最相关”。语义相似度模型可能无法理解“反义词”或“否定”关系。例如，问题“哪些产品不支持7天无理由退货？”可能与描述“支持7天无理由退货”的文本有高相似度，因为关键词高度重叠，但这会返回完全相反的答案。

生成阶段是最简单的部分。将检索到的Top-K个文本块作为上下文，连同用户问题，一起构造成提示词（Prompt），提交给大语言模型，让其生成最终答案。典型的Prompt模板是：“请基于以下上下文信息回答问题。如果上下文不包含答案，请说‘根据已知信息无法回答’。上下文：{context} \n 问题：{question}”。

2.2 朴素架构的“阿喀琉斯之踵”

尽管流程清晰，但Naive RAG在实战中暴露出一系列棘手问题，我称之为“七宗罪”：

检索精度不足（“找不到”）：仅靠语义相似度，无法处理多义词（“苹果”是水果还是公司？）、指代消解（“他说的那个方法”指哪个？）、复杂逻辑查询（“找出A和B都满足，但C不满足的条件”）。
上下文窗口浪费（“塞垃圾”）：Top-K个块可能包含大量重复信息或无关细节，挤占了宝贵的上下文窗口，导致模型无法聚焦关键信息，甚至因无关信息干扰而生成错误答案。
缺失全局信息（“见木不见林”）：检索到的都是局部片段，模型可能无法理解文档的整体结构、核心论点或叙事脉络。比如，要总结一篇论文的贡献，只给几个方法细节的片段是远远不够的。
静态检索，缺乏交互（“一锤子买卖”）：一次检索定生死。如果第一次没找到，系统就“放弃”了，不会根据初步结果调整搜索策略。
无法处理多跳问题（“只跑一趟腿”）：对于需要串联多个信息源才能回答的问题（例如，“张三的经理的老板是谁？”），Naive RAG通常无能为力。它一次性检索，缺乏分步推理和迭代检索的能力。
对噪声敏感（“脆弱”）：检索结果中只要混入一个高度相关但事实错误的片段（比如社区里有人写错了答案），就可能导致模型产生“幻觉”，因为它倾向于相信提供的上下文。
缺乏验证与溯源（“说不清”）：虽然提供了引用片段，但模型生成答案时，是否真的严格依据了这些片段？有没有偷偷掺杂自己的知识或臆测？这个过程缺乏透明度和可控性。

正是这些痛点，推动着RAG技术向更智能、更动态的方向演进。而解决这些问题的钥匙，就在于引入“智能体”的思维。

3. Agentic RAG：引入“智能”与“决策”的新范式

Agentic RAG不是一个具体的工具或框架，而是一种设计范式。其核心思想是：将大型语言模型作为一个具有推理和规划能力的“大脑”（智能体），来动态地指挥和控制整个RAG流程。检索不再是一个孤立的预处理步骤，而是成为了一个由LLM驱动的、可迭代、可决策的子任务。

3.1 智能体核心能力解构

在这个范式下，LLM智能体需要具备以下几种关键能力：

任务规划与分解：面对一个复杂查询，智能体首先不是直接去搜，而是像人类专家一样，先“理解”和“拆解”问题。例如，用户问“如何为公司的新产品‘星海’设计一个整合营销方案？”。智能体可能会将其分解为：“1. 查找公司‘星海’产品的核心卖点与技术文档。2. 检索过往类似产品的成功营销案例。3. 查找当前目标市场的用户画像分析报告。4. 汇总公司现有的营销渠道和预算规范。” 每一步都是一个更具体、更易检索的子任务。
工具调用：智能体需要知道“用什么工具”去完成每个子任务。工具不仅包括向量数据库检索，还可以是：
- 关键词搜索：在传统全文检索引擎（如Elasticsearch）中查找。
- 数据库查询：用SQL查询结构化数据。
- API调用：获取实时信息，如天气、股价、新闻。
- 计算器：进行数值计算。
- 代码解释器：执行数据分析或生成图表。
- 甚至调用另一个RAG系统：针对特定知识库进行检索。
反思与迭代：这是Agentic RAG超越Naive RAG的灵魂。智能体拿到初步检索结果后，会进行自我评估：“这些信息足够回答子问题了吗？信息之间有没有矛盾？我是否需要换个关键词或换个数据源再查一次？” 基于反思，它可以调整查询词、切换检索工具、或者决定进行多轮检索，直到收集到满意、一致的信息。
信息合成与验证：收集齐所有子任务的信息后，智能体不是简单拼接，而是进行交叉验证、去重、梳理逻辑关系，最后合成一个连贯、准确、完整的最终答案。它需要注明每部分信息的来源，实现可追溯。

3.2 主流实现模式剖析

目前，社区和实践中主要涌现出几种典型的Agentic RAG实现模式：

1. 自适应检索型这是最直接的一种增强。系统不是固定返回Top-K个结果，而是让LLM参与决定“怎么搜”和“搜多少”。

查询重写/扩展：LLM分析原始问题，将其重写为更利于向量检索的形式，或生成多个相关的查询词进行多查询检索。例如，将“怎么养好一只小奶猫？”重写为“幼猫喂养指南”、“新生小猫护理注意事项”、“0-6个月猫咪饮食健康”。
自适应检索深度：LLM根据问题的复杂性，动态决定检索多少个片段（K值）。简单问题可能K=3就够了，复杂综述性问题可能需要K=10。甚至可以引入递归检索，先检索少量片段，如果LLM判断信息不足，则触发新一轮检索。
混合检索路由：系统拥有多种检索器（向量检索、关键词检索、图数据库检索等）。LLM作为“路由器”，分析问题类型，决定调用哪个或哪几个检索器。例如，事实性问答用向量检索，精确术语匹配用关键词检索，关系查询用图检索。

2. 多智能体协作型这是更复杂的架构，模拟了一个专家团队。针对一个复杂任务，系统会虚拟出多个具有不同角色的智能体。

角色示例：
- 调度智能体：负责接收用户原始任务，进行任务分解和规划。
- 检索专家智能体：专门负责执行各类检索任务，精通不同数据源和检索语法。
- 分析智能体：负责对检索回来的信息进行批判性分析、对比和验证。
- 合成智能体：负责将分析后的信息整合成最终答案，并确保文风一致。
工作流程：调度智能体将子任务分发给检索专家，检索专家将结果交给分析智能体，分析智能体评估后可能要求检索专家重新查找，最后将净化后的信息交给合成智能体。整个过程中，智能体们通过“工作空间”或消息队列进行通信和协作。LangGraph、CrewAI等框架非常适合构建此类系统。

3. 迭代式问答与主动追问型这种模式将RAG变成了一个交互式、对话式的探索过程。

过程：用户提出初始问题。系统进行检索并生成一个初步答案，但同时会评估答案的置信度或信息的完整性。如果发现信息有缺口、存在多种可能、或问题本身模糊，系统会主动向用户提出澄清性问题。例如，“您问的‘高性能’具体是指计算速度还是内存带宽？”，“关于XX事件的报告，您是需要2023年的还是2024年的？”。根据用户的反馈，系统再次进行更精准的检索和生成。如此循环，直至产出令人满意的答案。这极大地提升了复杂问题解决的准确性和用户体验。

4. 构建实战：从零搭建一个Agentic RAG系统原型

理论说了这么多，我们来点实际的。我将带你用LangChain和LangGraph框架，搭建一个具备自适应检索和多跳推理能力的Agentic RAG系统原型。这个系统将能回答类似“公司里负责‘星海’项目的团队经理，他的直属上级是谁？”这样的问题。

4.1 环境准备与工具选型

首先，明确我们的技术栈：

框架：LangChain+LangGraph。LangChain提供了丰富的RAG组件和智能体工具，LangGraph则擅长描述和控制有状态的、循环的智能体工作流。
大模型：选择Qwen2.5-7B-Instruct的API版本或本地部署。它推理能力强，对中文支持好，且指令遵循能力出色，非常适合做智能体的“大脑”。（注：此处仅为示例，实际可根据资源选择GPT-4o、DeepSeek或GLM-4等）。
嵌入模型：选用BGE-M3。它支持多语言，在检索和重排序任务上表现均衡，并且可以输出密集向量、稀疏向量和ColBERT向量，为未来扩展混合检索留有余地。
向量数据库：使用ChromaDB，因其轻量、易用，适合原型开发。生产环境可替换为Qdrant。
知识源：我们模拟一个公司内部知识库，包含员工信息表（employees.csv）、部门结构表（departments.csv）和项目文档（若干PDF）。

# 环境安装（假设已安装Python 3.10+） pip install langchain langchain-community langgraph chromadb pypdf sentence-transformers # 如果需要使用Qwen API pip install dashscope # 如果使用本地Qwen模型 # pip install transformers torch

4.2 构建智能体工作流

我们的目标是构建一个能处理“多跳查询”的智能体。工作流设计如下：

接收问题：用户输入问题。
规划与分解：由LLM判断是否需要多步检索。如果需要，分解出第一个子问题。
工具调用与检索：根据子问题类型，选择合适工具（检索员工表？检索项目文档？）进行查询。
反思与迭代：LLM分析检索结果，判断是否已回答当前子问题，以及是否需要提出新的子问题。
合成答案：当所有必要子问题都被解答后，LLM综合所有信息，生成最终答案。

我们用LangGraph来实现这个有状态的工作流。

from typing import TypedDict, List, Annotated import operator from langchain_core.messages import HumanMessage, SystemMessage from langchain_qwen import ChatQwen # 假设使用Qwen API from langchain_community.vectorstores import Chroma from langchain_community.embeddings import HuggingFaceEmbeddings from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langchain_community.document_loaders import CSVLoader, PyPDFLoader from langchain_text_splitters import RecursiveCharacterTextSplitter # 1. 定义状态结构 class AgentState(TypedDict): messages: Annotated[List, add_messages] # 对话消息历史 original_question: str # 原始问题 current_subquestion: str # 当前正在处理的子问题 retrieved_context: List[str] # 累积检索到的上下文 final_answer: str # 最终答案 step: int # 当前步骤 # 2. 初始化核心组件 llm = ChatQwen(model="qwen2.5-7b-instruct", api_key="your-api-key") # 加载并索引知识库（此处简化，实际需分别处理CSV和PDF） embedding_model = HuggingFaceEmbeddings(model_name="BAAI/bge-m3") # ... 假设我们已经有了一个填充好的向量库 `vectorstore` # vectorstore = Chroma.from_documents(documents, embedding_model, persist_directory="./chroma_db") vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embedding_model) # 3. 定义工具函数 def retrieve_from_vectorstore(query: str, k: int = 5) -> str: """从向量库检索相关文档片段""" docs = vectorstore.similarity_search(query, k=k) return "\n\n".join([doc.page_content for doc in docs]) def query_employee_db(query: str) -> str: """模拟查询员工数据库（这里简化为一个函数）""" # 实际应连接数据库执行SQL，这里返回模拟数据 if "张三" in query: return "员工姓名：张三， 职位：高级工程师， 所属部门：产品研发部， 经理：李四" elif "李四" in query: return "员工姓名：李四， 职位：研发经理， 所属部门：产品研发部， 上级：王五" elif "王五" in query: return "员工姓名：王五， 职位：技术总监， 所属部门：技术中心" else: return "未找到匹配的员工信息。" # 4. 定义智能体节点函数 def planner_node(state: AgentState) -> AgentState: """规划节点：分析问题，决定下一步""" messages = state['messages'] # 系统提示词，引导LLM进行任务分解 system_prompt = """你是一个智能信息助理。你的任务是分析用户问题，判断是否需要通过多步检索来回答。 如果需要，请生成第一个需要回答的子问题。请只输出子问题本身，不要输出其他解释。 如果原始问题可以直接回答，请输出“FINAL”。""" planner_prompt = [ SystemMessage(content=system_prompt), HumanMessage(content=f"原始问题：{state['original_question']}\n当前已收集的信息：{state['retrieved_context']}") ] response = llm.invoke(planner_prompt) decision = response.content.strip() if decision.upper() == "FINAL": state['current_subquestion'] = "FINAL" else: state['current_subquestion'] = decision state['step'] += 1 return state def retrieval_node(state: AgentState) -> AgentState: """检索节点：根据子问题调用工具检索""" sub_q = state['current_subquestion'] if sub_q == "FINAL": return state # 简单的路由逻辑：如果问题包含“员工”“谁”“经理”等，查员工库；否则查向量库 if any(keyword in sub_q for keyword in ["员工", "谁", "经理", "上级"]): context = query_employee_db(sub_q) else: context = retrieve_from_vectorstore(sub_q) state['retrieved_context'].append(f"【子问题】{sub_q}\n【相关信息】{context}") return state def reflector_node(state: AgentState) -> AgentState: """反思节点：评估信息是否足够，决定继续还是结束""" if state['current_subquestion'] == "FINAL": state['current_subquestion'] = "READY_TO_ANSWER" return state # 让LLM判断是否还需要继续追问 reflection_prompt = f""" 基于当前收集到的信息： {chr(10).join(state['retrieved_context'])} 针对原始问题“{state['original_question']}”，我们是否已经获得了足够的信息来合成最终答案？ 如果已经足够，请回复“SUFFICIENT”。 如果还需要更多信息，请提出下一个需要解决的关键子问题。 """ response = llm.invoke([HumanMessage(content=reflection_prompt)]) decision = response.content.strip() if decision.upper() == "SUFFICIENT": state['current_subquestion'] = "READY_TO_ANSWER" else: # 将LLM提出的新子问题作为下一轮的开始 state['current_subquestion'] = decision state['step'] += 1 return state def answer_node(state: AgentState) -> AgentState: """答案合成节点：生成最终答案""" synthesis_prompt = f""" 请严格根据以下提供的信息，回答原始问题：“{state['original_question']}”。 如果信息不足，请明确说明哪些部分无法确定。 信息： {chr(10).join(state['retrieved_context'])} 请给出完整、准确的答案： """ response = llm.invoke([HumanMessage(content=synthesis_prompt)]) state['final_answer'] = response.content state['messages'].append(HumanMessage(content=state['original_question'])) state['messages'].append(AIMessage(content=state['final_answer'])) return state # 5. 构建并编译工作流图 workflow = StateGraph(AgentState) # 添加节点 workflow.add_node("planner", planner_node) workflow.add_node("retriever", retrieval_node) workflow.add_node("reflector", reflector_node) workflow.add_node("answer", answer_node) # 设置边和条件流转 workflow.set_entry_point("planner") workflow.add_edge("planner", "retriever") # 关键：条件边。从反思节点出来，根据状态决定是继续检索还是去生成答案 def decide_after_reflection(state: AgentState): if state['current_subquestion'] == "READY_TO_ANSWER": return "answer" else: return "planner" # 返回规划节点，开始新一轮“规划-检索-反思” workflow.add_conditional_edges( "reflector", decide_after_reflection, { "answer": "answer", "planner": "planner" } ) workflow.add_edge("retriever", "reflector") workflow.add_edge("answer", END) # 编译图 app = workflow.compile() # 6. 运行测试 initial_state = { "messages": [], "original_question": "负责‘星海’项目的团队经理，他的直属上级是谁？", "current_subquestion": "", "retrieved_context": [], "final_answer": "", "step": 0 } # 假设我们的知识库中，项目文档提到“星海”项目由“张三”负责，而员工表中有张三及其经理的信息。 result = app.invoke(initial_state, config={"recursion_limit": 10}) # 限制递归深度 print("最终答案：", result['final_answer']) print("检索历史：", result['retrieved_context'])

这个原型展示了Agentic RAG的核心循环：规划 -> 执行（检索）-> 反思 -> 再规划。当面对“星海项目经理的上级”这个问题时，智能体可能会先规划出子问题1：“‘星海’项目的负责人是谁？”，从项目文档中检索出“张三”；然后反思，发现需要知道张三的上级，于是规划出子问题2：“员工张三的经理是谁？”，从员工数据库中检索出“李四”；最后合成答案：“星海项目的团队经理是李四”。

4.3 关键配置与调优心得

提示词工程是灵魂：planner_node和reflector_node的系统提示词（System Prompt）直接决定了智能体的“思考方式”。你需要精心设计，明确告诉它扮演什么角色、输出格式是什么、决策逻辑是什么。多迭代几次提示词，效果天差地别。
控制循环与超时：必须设置递归深度限制（recursion_limit）或超时机制，防止智能体陷入“思考死循环”。例如，同一个问题反复检索却无法推进时，应强制跳出并提示用户。
工具描述的准确性：在更复杂的系统中，你需要用自然语言清晰地向LLM描述每个工具的功能、输入和输出格式。LangChain的tool decorator和bind_tools方法能很好地完成这一点。
状态管理：AgentState的设计至关重要。它需要包含所有必要的历史信息，以供后续节点决策。确保状态简洁且包含所有上下文。
评估与监控：Agentic系统比Naive RAG更难评估。除了答案准确性，还要关注其推理步骤的合理性、工具调用的正确率和循环次数。建立一套评估体系至关重要。