Agentic_RAG AI学习笔记02

Agentic_RAG AI学习笔记02

从传统 RAG 到 Agentic RAG:一份讲清「是什么 / 为什么 / 怎么实现」的学习笔记

写作主线:以 RAG 的演进逻辑为线索——每一代方案都是为了解决上一代的具体痛点。 对每一种范式都讲三件事:是什么(架构)→ 为什么这么设计(动机)→ 怎么实现(最小可运行代码)。 代码栈统一为 Python + LangChain + LangGraph;配套可逐格运行的 Notebook 见 agentic_rag_demo.ipynb

学习笔记的配套代码: https://github.com/LT-IENG/AI-Agent-Study-Notes


目录


0. 导读:怎么读这份笔记

如果你只记一句话,请记这句:

RAG 的进化史,就是「把越来越多的决策权,从写死的流水线交还给模型本身」的历史。

  • Naive RAG:流程写死,检索一次 → 拼接 → 生成,模型没有任何决策权。
  • Advanced RAG:流程仍然写死,但在检索前/中/后插入更多固定优化环节,提升检索质量。
  • Self-RAG / CRAG:加入「评估—纠错」的判断分支,但分支规则仍是人设计的。
  • GraphRAG:换索引结构(知识图谱 + 社区摘要),解决向量检索答不了「全局性问题」的短板。
  • Agentic RAG:把检索能力变成工具,让模型通过 思考 → 行动 → 观察 的循环自主决定搜不搜、搜什么、搜几次、何时停。
  • RL 驱动的 Agentic RAG(Search-R1):连「怎么决策」都不靠人写提示词,而是用强化学习训练出来。

学习时建议的顺序:先吃透第 3 章 Naive RAG 和它的 5 个缺陷(后面每一章都在对应补其中一个缺陷),再看第 8 章 Agentic RAG 做对照,其余章节按需查阅。


1. 背景:为什么会有 RAG

1.1 是什么

RAG(Retrieval-Augmented Generation,检索增强生成):在大语言模型(LLM)生成回答之前,先从外部知识库检索出相关资料,把资料拼进提示词(Prompt),再让模型「看着资料回答」。

RAG 概念最早来自 Facebook AI(现 Meta AI)的论文 Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks(Lewis et al., NeurIPS 2020),最初是把「检索器 + 生成器」联合训练用于知识密集型任务;ChatGPT 引爆 LLM 之后,RAG 演变成了今天大家熟悉的「向量检索 + LLM 问答」工程形态。

1.2 为什么需要它:纯 LLM 的三个硬伤

硬伤具体表现RAG 如何缓解
知识截止(Knowledge Cutoff)模型只知道训练数据截止前的事,不知道最新信息知识库可随时更新,检索的是最新资料
幻觉(Hallucination)不知道也会「一本正经地编」,且无法核实要求模型基于给定材料作答,答案可溯源
缺乏私有/内部知识模型没读过你公司的文档、你的个人笔记把私有文档做成知识库,无需重新训练模型

1.3 为什么不是「微调模型」而是 RAG

这是面试和工程选型都绕不开的对比:

维度RAG(外挂知识)Fine-tuning(微调)加长上下文(Long Context)
知识更新改知识库即可,分钟级需重新训练,天/周级、成本高每次请求都要塞入,token 成本高
可溯源性强,能给出引用来源弱,知识融进参数无法溯源中,可引用但易「大海捞针」(Lost in the Middle)
适配场景事实性问答、知识频繁变动改变模型风格/语气/固定技能一次性通读少量文档
成本低,无需训练高,需要 GPU 与标注数据推理成本随长度线性甚至超线性增长

结论:RAG、微调、长上下文不是互斥关系。RAG 负责「知识」(随时更新、可引用),微调负责「能力与风格」,长上下文负责「一次性通读」。生产系统里三者经常组合使用。

1.4 RAG 的最小思想

用一句伪代码概括所有 RAG 的共同内核:

docs = retrieve(user_query, knowledge_base, top_k=3)   # 1. 检索:找资料
prompt = f"根据以下资料回答问题:\n{docs}\n问题:{user_query}"  # 2. 增强:拼进 Prompt
answer = llm.generate(prompt)                          # 3. 生成:基于资料回答

后面所有复杂范式,都是在这三行的不同位置「做文章」。


2. RAG 全景与演进谱系

复旦 Gao 等人的综述 Retrieval-Augmented Generation for Large Language Models: A Survey(arXiv:2312.10997)把学术界的 RAG 归纳为 Naive → Advanced → Modular 三个范式;2024 年后随着 Agent 兴起,工业界又发展出 Agentic RAG。把它们放在一张图上:

flowchart TB
    Q[用户问题]

    subgraph N[① Naive RAG 一次写死的流水线]
        direction LR
        N1[检索 Top-K] --> N2[拼接 Prompt] --> N3[LLM 生成]
    end

    subgraph A[② Advanced RAG 固定流程 + 多点优化]
        direction LR
        A1["检索前<br/>查询改写/HyDE/多路查询"] --> A2["检索中<br/>混合检索"] --> A3["检索后<br/>Rerank/压缩"] --> A4[LLM 生成]
    end

    subgraph R[③ 带判断分支的 RAG 规则仍由人设计]
        direction LR
        R1[检索] --> R2{质量评估} -->|相关| R3[精炼后生成]
        R2 -->|不相关| R4[改写/回退 Web 搜索] --> R1
    end

    subgraph G[④ GraphRAG 换索引结构]
        direction LR
        G1[实体关系图谱] --> G2[社区层级摘要] --> G3[Local/Global 查询]
    end

    subgraph AG[⑤ Agentic RAG 模型自主决策循环]
        direction LR
        AG1[思考 Reason] --> AG2[行动 Act: 调用检索工具] --> AG3[观察 Observe] --> AG1
        AG1 -->|证据充分| AG4[生成答案 + 引用]
    end

    subgraph RL[⑥ RL 驱动的 Agentic RAG]
        direction LR
        RL1[策略模型] --> RL2[推理-搜索交织 Rollout] --> RL3[结果奖励] --> RL4[GRPO/PPO 更新] --> RL1
    end

    N --> A --> R --> AG
    A --> G
    AG --> RL

演进的两条主线

  1. 检索质量线(让找到的内容更准):Naive → Advanced(改写/混合/Rerank)→ GraphRAG(换结构)。
  2. 决策自主权线(让模型更主动):Naive(写死)→ Self-RAG/CRAG(固定判断分支)→ Agentic RAG(自主工具循环)→ RL Agentic RAG(决策能力靠训练获得)。

3. Naive RAG(传统 RAG)

别名:Native RAG / Vanilla RAG / Naive RAG,都是同一个东西。

3.1 是什么:离线 + 在线两条链路

Naive RAG 分成**离线入库(Indexing)在线问答(Retrieval & Generation)**两条链路:

flowchart LR
    subgraph OFF[离线链路:一次性建库]
        D[原始文档<br/>PDF/网页/Markdown/JSON] --> L[加载 Load]
        L --> S[切分 Split<br/>chunk_size / overlap]
        S --> E[向量化 Embedding]
        E --> V[(向量数据库<br/>+ 原文库)]
    end

    subgraph ON[在线链路:每次提问执行]
        U[用户 Query] --> SE[Query 向量化]
        SE --> SIM[相似度检索 Top-K]
        SIM --> CTX[拼接 Context]
        CTX --> P[构建最终 Prompt]
        P --> LLM[LLM]
        LLM --> O[回答 Output]
    end

    V -. 提供候选 .-> SIM

3.2 为什么这么设计:必须先理解的 5 个核心概念

概念含义设计要点 / 为什么
Chunk(文本块)把长文档切成的小段,是检索与注入的最小单位块太大 → 噪声多、稀释重点;块太小 → 语义不完整。典型 300–800 字
Chunk Overlap(重叠)相邻块之间保留一段重叠文本避免一句话/一个论点被从中间切断,保证上下文连续,典型取 chunk 的 10%
Embedding(嵌入向量)把文本映射成高维向量,语义相近的文本向量距离近这是「用意思检索而非关键词匹配」的基础
向量数据库存储向量并支持近似最近邻(ANN)检索支持海量向量下的毫秒级相似度搜索,如 Chroma/Milvus/Qdrant
Top-K取相似度最高的 K 个块K 太小可能漏证据,K 太大污染上下文、浪费 token,常取 3–5 起步调优

为什么要切分而不是整篇入库? ① Embedding 模型有长度上限;② 一整篇文档主题混杂,向量会被「平均化」,检索不精准;③ 注入 LLM 的上下文长度有限,小块才能「按需取用」。

3.3 怎么实现(最小可运行代码)

两种实现,原理一致、可互换

  • 配套 Notebook(零依赖、开箱即跑):用纯本地轻量 Embedding(字符 n-gram + 余弦相似度)+ 内存向量库,无需部署数据库、无需在线 Embedding 服务,DeepSeek 等只负责生成;
  • 下面代码块是生产标准写法:在线 Embedding(OpenAI / bge / Qwen 等)+ 持久向量库 Chroma。把 Notebook 里的 LocalHashEmbeddings/LocalVectorStore 换成这里的 OpenAIEmbeddings + Chroma 即上生产。

依赖(生产写法):pip install langchain langchain-openai langchain-text-splitters langchain-chroma chromadb。API 配置从 .env 读取。

① 离线入库:加载 → 切分 → 向量化 → 存储

# offline_index.py —— Naive RAG 离线链路(生产写法)
import os
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter  # 1.x 独立包
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma

# 1. 加载文档(实际项目中可换成 PyPDFLoader / UnstructuredMarkdownLoader 等)
documents = TextLoader("knowledge_base.txt", encoding="utf-8").load()

# 2. 递归字符切分:优先按段落、再按句子边界切,尽量不破坏语义
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,       # 每个文本块目标大小(字符数)
    chunk_overlap=50,     # 相邻块重叠,避免论点被切断
)
splits = text_splitter.split_documents(documents)

# 3. Embedding 模型(OpenAI 兼容接口;换成你所用服务商提供的 embedding 模型名)
#    注意:DeepSeek 官方不提供 embedding 接口,需用 OpenAI / 硅基流动 Qwen / 本地 bge 等
embeddings = OpenAIEmbeddings(
    base_url=os.getenv("EMBED_BASE_URL", "https://api.openai.com/v1"),
    model=os.getenv("EMBED_MODEL", "text-embedding-3-small"),
    api_key=os.getenv("EMBED_API_KEY", os.getenv("LLM_API_KEY")),
)

# 4. 向量化并持久化到本地 Chroma
vectorstore = Chroma.from_documents(
    documents=splits,
    embedding=embeddings,
    persist_directory="./chroma_db",
)
print(f"已将 {len(splits)} 个文本块存入向量数据库")

② 在线问答:检索 → 拼 Prompt → 生成

# online_qa.py —— Naive RAG 在线链路(生产写法)
import os
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_core.prompts import PromptTemplate   # 1.x 走 langchain_core

# 1. 加载已建好的向量库
embeddings = OpenAIEmbeddings(
    base_url=os.getenv("EMBED_BASE_URL", "https://api.openai.com/v1"),
    model=os.getenv("EMBED_MODEL", "text-embedding-3-small"),
    api_key=os.getenv("EMBED_API_KEY", os.getenv("LLM_API_KEY")),
)
vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embeddings)

# 2. 检索 Top-K
query = "什么是 RAG?"
docs = vectorstore.similarity_search(query, k=3)
context = "\n\n".join(doc.page_content for doc in docs)

# 3. 构建 Prompt(注意:明确约束「资料里没有就说不知道」,是抑制幻觉的关键)
prompt = PromptTemplate.from_template("""你是一个专业问答助手。请只根据下面的参考文档回答问题。
如果参考文档中没有相关信息,请直接说“根据现有资料无法回答”,不要编造。

参考文档:
{context}

用户问题:{question}
回答:""").format(context=context, question=query)

# 4. LLM 生成(对话模型与 embedding 可以是不同服务商)
llm = ChatOpenAI(
    model=os.getenv("CHAT_MODEL", "deepseek-v4-flash"),
    temperature=0,                 # 知识问答用 0,降低发散
    base_url=os.getenv("CHAT_BASE_URL", "https://api.deepseek.com/v1"),
    api_key=os.getenv("LLM_API_KEY"),
)
answer = llm.invoke(prompt).content
print(answer)

3.4 Naive RAG 的五个天生缺陷(重点:后面每章都在补这些坑)

  1. 一次性流水线,没有回头路:「检索 → 拼接 → 生成」一步到位。第一次没搜好,不会改写查询、不会补搜,直接带着错误/缺失的证据硬答。
  2. 缺乏任务拆解:复杂问题往往要「先定位文件 → 再选片段 → 再对比综合」,Naive RAG 没有多步规划能力。
  3. 工具单一,只会相似度检索:不会看文件元数据、不会指定读某个块、不会换关键词或换检索方式。
  4. 证据利用浅,Top-K 直接「糊」进上下文:做不到「先粗筛、后精读」(coarse-to-fine),也无法精确引用到具体片段;无关块还会稀释上下文。
  5. 适应性差:面对多跳问题(multi-hop,答案要跨多个文档串联)、术语不匹配(用户说中文、文档是英文术语)、信息不足等场景,不会回溯重试。

记住这 5 条,第 4、5、6、8 章分别对应不同的补救路线。


4. Advanced RAG(进阶 RAG:检索前 / 中 / 后优化)

4.1 是什么 & 为什么这么设计

Advanced RAG 不改变「先检索后生成」的固定流水线结构,而是在三个阶段插入优化模块,专门解决 Naive RAG「检索不准、证据被稀释」的问题(对应缺陷 1、4):

flowchart LR
    Q[原始 Query] --> PRE["【检索前 Pre-Retrieval】<br/>查询改写 / HyDE / 多路查询 / Step-back"]
    PRE --> RET["【检索中 Retrieval】<br/>稠密向量 + 稀疏 BM25 混合检索"]
    RET --> POST["【检索后 Post-Retrieval】<br/>Rerank 重排序 / 上下文压缩 / 去重"]
    POST --> LLM[LLM 生成]

设计哲学:检索质量决定 RAG 上限,生成只是逼近这个上限。 业界经验是,把精力花在「检索准不准」上的 ROI,远高于反复调生成 Prompt。

4.2 检索前(Pre-Retrieval):让 Query 变得「更好搜」

用户的原始提问往往短、口语化、和文档用词不一致,直接拿去做向量检索效果差。检索前优化就是把 Query 改写成更适合检索的形态。

(1)Query Rewrite(查询改写)

让 LLM 把口语问题改写成更规范、更贴近文档表述的检索语句(可纠错、补全指代、扩展同义词)。

(2)Multi-Query(多路查询 / 子查询分解)

让 LLM 从多个角度把同一个问题改写成 N 个不同的查询,分别检索后合并去重——用「广撒网」提高召回率,特别适合问题本身有多个侧面的场景。

(3)HyDE(Hypothetical Document Embeddings,假设文档嵌入)

  • 论文:Gao et al., Precise Zero-Shot Dense Retrieval without Relevance Labels(2022)。
  • 要解决的痛点:Query 很短(几个字),文档很长(一段话),两者在向量空间里天然「不对称」,哪怕语义相关,余弦相似度也不高。
  • 核心思想(非常巧妙):先让 LLM 「假装」生成一段能回答该问题的假设性答案文档(哪怕内容是编的、可能有错也没关系),再用这段「假答案」去做向量检索。因为假答案和真实文档在长度、术语、表述风格上更接近,向量匹配反而更准。
flowchart LR
    Q[短 Query] -->|Naive: 直接编码| E1[(向量空间)]
    Q --> G[LLM 生成假设答案文档<br/>不要求事实正确]
    G -->|HyDE: 编码假文档| E2[(向量空间)]
    E1 -. 距离真实文档远 .-> D[真实文档]
    E2 == 距离真实文档更近 ==> D

(4)Step-back Prompting(后退提问)

遇到很具体的问题时,先让 LLM 提出一个更抽象、更上位的问题(如「某型号电池 2024 年故障率」→ 后退为「该型号电池的可靠性历史」),先检索背景知识,再结合具体问题作答,避免被细节带偏。

检索前优化最小代码(Query 改写 + HyDE + Multi-Query)

# advanced_pre_retrieval.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="deepseek-v4-flash", temperature=0)

# ---------- 1) Query Rewrite ----------
rewrite_prompt = ChatPromptTemplate.from_template(
    "请把下面的用户问题改写为更适合知识库检索的规范查询(补全指代、扩展同义术语),"
    "只输出改写后的查询,不要解释:\n{query}"
)
better_query = (rewrite_prompt | llm).invoke({"query": "它的函数调用咋写?"}).content

# ---------- 2) Multi-Query:一个问题生成多个视角的查询 ----------
multi_prompt = ChatPromptTemplate.from_template("""你是检索查询生成器。请围绕下面的问题,生成 {n} 个不同角度、
用于向量检索的子查询,每行一个,不要编号:
问题:{query}""")
queries = (multi_prompt | llm).invoke({"query": "RAG 和微调怎么选?", "n": 3}).content.splitlines()
all_docs = []
for q in [better_query, *queries]:
    all_docs.extend(vectorstore.similarity_search(q, k=3))
# 按内容去重
unique_docs = {d.page_content: d for d in all_docs}.values()

# ---------- 3) HyDE:先生成假设答案,再用假设答案去检索 ----------
hyde_prompt = ChatPromptTemplate.from_template(
    "请写一段简短的、能够回答以下问题的说明性文字(假设性文档,100字内):\n{query}"
)
hypothetical_doc = (hyde_prompt | llm).invoke({"query": "什么是 Agentic RAG?"}).content
hyde_docs = vectorstore.similarity_search(hypothetical_doc, k=3)  # 注意:编码的是“假答案”
  • 稠密检索(Dense,向量相似度):懂语义、能匹配同义词,但对专有名词、数字、错误拼写、罕见关键词不敏感。
  • 稀疏检索(Sparse,BM25 关键词):精确匹配关键词强(型号、人名、错误码),但不懂同义/语义。
  • 混合检索:两者同时召回,再用 RRF(Reciprocal Rank Fusion,倒数排名融合) 合并排名,兼顾「语义」与「关键词」。这是生产环境的标配。
# hybrid_search.py —— 向量 + BM25 的最小混合检索
from langchain_community.retrievers import BM25Retriever
# 1.x 起 EnsembleRetriever 迁入兼容包 langchain_classic;也可像配套 Notebook 那样
# 用十几行自写 RRF(score = Σ 1/(k+rank)),零额外依赖、原理透明
from langchain_classic.retrievers import EnsembleRetriever

all_splits = text_splitter.split_documents(documents)

vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
bm25_retriever = BM25Retriever.from_documents(all_splits)
bm25_retriever.k = 5

# weights:两路检索的权重;RRF 融合在 EnsembleRetriever 内部完成
hybrid_retriever = EnsembleRetriever(
    retrievers=[vector_retriever, bm25_retriever],
    weights=[0.6, 0.4],
)
docs = hybrid_retriever.invoke("DeepSeek 的 Function Calling 用法")

4.4 检索后(Post-Retrieval):Rerank 与上下文压缩

召回阶段追求「别漏掉」(宁可多召回,如 Top-20),但直接把 20 个块全塞给 LLM 既贵又噪声大。检索后要做「精选」:

  1. Rerank(重排序):用 Cross-Encoder 交叉编码器(如 bge-reranker、Cohere Rerank)让「问题 + 每个候选块」成对打分精排。
    • 为什么比向量相似度更准?向量检索是「双塔」:Query 和文档分别编码成向量再比距离,两者在编码时没见过彼此,速度快但粗;Rerank 是「交叉」:把 Query 和文档拼在一起送进模型做精细相关判断,准但慢。所以形成「双塔粗召回 → Cross-Encoder 精排」的两级架构。
  2. 上下文压缩(Contextual Compression):只抽取每个块里真正和问题相关的句子,删掉无关部分,省 token、降噪声。
  3. 去重 / 多样性(MMR):避免召回内容高度重复。
# post_retrieval.py —— Rerank 两级检索最小示例
# pip install langchain-cohere  (或用本地 bge-reranker,思路一致)
from langchain_cohere import CohereRerank
# 1.x 起 ContextualCompressionRetriever 同样在 langchain_classic.retrievers
from langchain_classic.retrievers import ContextualCompressionRetriever

# 第一级:混合检索粗召回 20 个
base_retriever = hybrid_retriever
# 第二级:Cross-Encoder 精排,只留最相关的 3 个
reranker = CohereRerank(top_n=3, model="rerank-multilingual-v3.0")
compression_retriever = ContextualCompressionRetriever(
    base_compressor=reranker,
    base_retriever=base_retriever,
)
final_docs = compression_retriever.invoke("Agentic RAG 为什么需要多轮检索?")

4.5 Advanced RAG 的局限

它显著提升了「单次检索」的质量,但流程依然是写死的:不管问题简单还是复杂,都固定走「改写→混合检索→Rerank→生成」一遍;检索结果到底够不够、要不要换个说法再来一次,系统并不判断。这正是下一类范式要解决的问题。


5. 会「自我检查」的 RAG:Self-RAG、CRAG、Adaptive RAG

这一类方案的共同思路:在流水线里插入「评估/判断」环节,检索结果不好就纠正。它们是从「写死流程」走向「自主决策」的过渡形态——已经有了循环和分支,但「何时分支、如何纠错」的规则仍是人设计的。

5.1 Self-RAG:让模型学会「检索、生成、并自我批判」

  • 论文:Asai et al., Self-RAG: Learning to Retrieve, Generate, and Critique through Self-Reflections(ICLR 2024, arXiv:2310.11511)。
  • 要解决的痛点:Naive RAG 无论问题是否需要,都强制检索一次;而且无论检索到的内容是否相关、能否支撑答案,都强行使用,照样产生幻觉。
  • 核心机制:通过微调,让模型在正常输出文字之外,内联生成 4 类「反思令牌(Reflection Tokens)」,把「要不要检索、证据可不可信」变成模型自己输出的机器可读控制信号:
反思令牌触发时机取值作用(为什么这么设计)
Retrieve生成每段内容前Yes / No / Continue简单常识题直接答(省检索);需要外部知识才检索
ISREL(相关性)拿到每个检索段落后Relevant / Irrelevant过滤无关段落,防止污染上下文
ISSUP(支撑度)生成每个候选句子后Fully / Partially / No Support判断这句话是否真的由证据支撑,掐断幻觉
ISUSE(有用性)整段答案生成后1–5 分对最终答案整体打分,选最优答案
flowchart TD
    Q[输入问题] --> R{Retrieve?<br/>需要检索吗}
    R -->|No 模型自己知道| G0[直接生成]
    R -->|Yes| RT[检索 Top-K 段落]
    RT --> REL{ISREL<br/>段落相关吗}
    REL -->|Irrelevant| DROP[丢弃该段落]
    REL -->|Relevant| G[逐句生成]
    G --> SUP{ISSUP<br/>证据能支撑吗}
    SUP -->|No Support| RT
    SUP -->|Fully/Partial| U{ISUSE<br/>答案有用性打分}
    U -->|分数最高者| OUT[输出最终答案]

工程上的轻量替代:不必微调模型,用一个普通 LLM 按 Prompt 输出这些判断,即可模拟 Self-RAG 的反思闭环:

# self_rag_lite.py —— 用 Prompt 模拟 Self-RAG 的反思闭环(无需微调)
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="deepseek-v4-flash", temperature=0)

def need_retrieval(question: str) -> bool:
    """对应 Retrieve 令牌:判断是否需要外部检索"""
    resp = llm.invoke(f"回答下面问题是否需要查阅外部资料?只回答 yes 或 no:\n{question}")
    return resp.content.strip().lower().startswith("y")

def is_relevant(question: str, doc: str) -> bool:
    """对应 ISREL:判断检索段落是否相关"""
    resp = llm.invoke(f"问题:{question}\n文档:{doc}\n该文档与问题相关吗?只回答 relevant/irrelevant")
    return "irrelevant" not in resp.content.lower()

def is_supported(question: str, claim: str, doc: str) -> bool:
    """对应 ISSUP:判断生成内容是否被证据支撑(幻觉闸门)"""
    resp = llm.invoke(
        f"文档:{doc}\n论断:{claim}\n该论断是否能由文档支撑?只回答 supported/unsupported"
    )
    return "unsupported" not in resp.content.lower()

def self_rag_answer(question):
    if not need_retrieval(question):
        return llm.invoke(question).content
    candidates = vectorstore.similarity_search(question, k=5)
    good_docs = [d for d in candidates if is_relevant(question, d.page_content)]  # ISREL
    context = "\n\n".join(d.page_content for d in good_docs)
    draft = llm.invoke(f"根据资料回答:\n{context}\n问题:{question}").content
    # ISSUP:不被支撑则带证据重答一次
    if good_docs and not is_supported(question, draft, context):
        draft = llm.invoke(f"上一版回答缺乏证据,请严格只依据资料重答:\n资料:{context}\n问题:{question}").content
    return draft

5.2 CRAG(Corrective RAG,纠错式 RAG)

  • 论文:Yan et al., Corrective Retrieval Augmented Generation(2024, arXiv:2401.15884);LangGraph 官方教程的经典案例。
  • 要解决的痛点:检索器可能整体返回错误/不相关的结果(比如知识库压根没有相关内容),Naive RAG 会拿着错误资料硬答。CRAG 的思路是「先给检索结果打分,错了就纠错」。
  • 核心机制
  1. 轻量检索评估器(Retrieval Evaluator):把 Query 和每个检索文档配对,输出置信度,分成三档:
    • Correct(正确,置信度高):本地知识库够用 → 进入「知识精炼」。
    • Incorrect(错误,置信度低):本地结果不可信 → 丢弃本地结果,回退到 Web 搜索扩大信息源。
    • Ambiguous(模糊,介于之间):本地 + Web 都用,互为补充。
  2. 知识精炼(Knowledge Refinement):用「先分解再重组(decompose-then-recompose)」算法,把文档切成细条、过滤掉无关条、只保留关键知识条,起到去噪作用。
  3. Web 兜底:静态、有限的本地知识库不可能什么都有,大规模 Web 搜索作为「纠错扩展」。
flowchart TD
    Q[用户 Query] --> RT[本地知识库检索]
    RT --> EV{检索评估器<br/>置信度分档}
    EV -->|Correct 正确| KF[知识精炼<br/>分解→过滤→重组]
    EV -->|Ambiguous 模糊| KF
    EV -->|Ambiguous 模糊| WEB[Web 搜索补充]
    EV -->|Incorrect 错误| WEB2[丢弃本地结果<br/>仅用 Web 搜索]
    KF --> MERGE[合并证据]
    WEB --> MERGE
    WEB2 --> MERGE
    MERGE --> GEN[LLM 生成答案]

用 LangGraph 实现 CRAG 的最小骨架(LangGraph 用「状态图」表达分支与循环,非常契合这类范式):

# crag_langgraph.py —— CRAG 最小骨架(省略部分模板代码,完整可运行版见 Notebook)
from typing import List
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class GraphState(TypedDict):
    question: str
    documents: List[str]
    web_results: List[str]
    generation: str

def retrieve(state):
    return {"documents": hybrid_retriever.invoke(state["question"])}

def grade_documents(state):
    """检索评估器:二分类简化版(论文中为三档 + 置信度阈值)"""
    q = state["question"]
    good, bad = [], []
    for d in state["documents"]:
        verdict = llm.invoke(
            f"问题:{q}\n文档:{d}\n文档是否与问题相关?只回答 yes/no"
        ).content.lower()
        (good if "yes" in verdict else bad).append(d)
    # 只要存在不相关文档,就触发 Web 搜索兜底(对应 Ambiguous/Incorrect)
    return {"documents": good, "web_needed": bool(bad) or not good}

def web_search(state):
    if not state.get("web_needed"):
        return {}
    results = tavily_search(state["question"])   # 任意 Web 搜索 API
    return {"web_results": results}

def refine_and_generate(state):
    """知识精炼(简化:LLM 抽取关键句)+ 生成"""
    evidence = "\n".join(state["documents"] + state.get("web_results", []))
    answer = llm.invoke(f"只依据以下证据回答,无法回答就明说:\n{evidence}\n问题:{state['question']}")
    return {"generation": answer.content}

def route_after_grade(state):
    return "web_search" if state.get("web_needed") else "generate"

graph = StateGraph(GraphState)
graph.add_node("retrieve", retrieve)
graph.add_node("grade_documents", grade_documents)
graph.add_node("web_search", web_search)
graph.add_node("generate", refine_and_generate)

graph.add_edge(START, "retrieve")
graph.add_edge("retrieve", "grade_documents")
graph.add_conditional_edges("grade_documents", route_after_grade,
                            {"web_search": "web_search", "generate": "generate"})
graph.add_edge("web_search", "generate")
graph.add_edge("generate", END)

crag = graph.compile()
print(crag.invoke({"question": "公司报销流程的截止日期是哪天?"}))

5.3 Adaptive RAG(自适应 RAG)

  • 论文:Jeong et al., Adaptive-RAG: Learning to Adapt What to Retrieve for Questions Requiring Multi-Hop Reasoning(NAACL 2024)。
  • 思想:不同复杂度的问题没必要走同一套重流程。先用一个轻量分类器/路由判断问题类型,再选择策略:
    • 简单事实/常识问题 → 不检索,直接回答(最省);
    • 单跳问题 → 单次检索
    • 复杂多跳问题 → 多步迭代检索(最重但最稳)。
  • 它和 Self-RAG/CRAG 一起,构成了「按问题动态选择处理路径」的思想,这其实已经非常接近 Agentic RAG——区别是:Adaptive RAG 的可选路径集合是人预先枚举的,而 Agentic RAG 的行动序列是模型临场自主规划的。

6. 图谱型 RAG:Microsoft GraphRAG

6.1 是什么 & 为什么这么设计

  • 论文:Edge et al., From Local to Global: A Graph RAG Approach to Query-Focused Summarization(Microsoft, 2024, arXiv:2404.16130),开源仓库 microsoft/graphrag
  • 它要解决的是向量 RAG 的一个根本盲区——答不了「全局性问题(global / sensemaking questions)」

举个例子:问「这批文档里讨论的几个主题分别是什么、它们之间有什么共性?」。向量 RAG 只能用问题去匹配局部最相似的几个块,再怎么 Top-K 也只是「管中窥豹」——没有任何一个单独的块包含「全局主题」这个信息,它需要通读整个语料库后归纳。这就是论文标题说的 From Local to Global

6.2 核心设计:知识图谱 + 社区层级摘要

GraphRAG 的索引阶段(离线,成本较高,大量调用 LLM):

flowchart TD
    C[文本切块] --> EX["① 实体/关系抽取<br/>LLM 从每个块抽取实体节点与关系边"]
    EX --> KG[("② 构建知识图谱<br/>实体=节点, 关系=边")]
    KG --> CD["③ 社区检测<br/>Leiden 算法把紧密相连的实体聚成社区<br/>并形成多层级社区树"]
    CD --> SM["④ 自底向上生成社区摘要/报告<br/>每个社区一份:关键实体、关系、主题、洞察"]
    SM --> IDX[(图谱 + 多层社区报告索引)]

为什么「社区摘要」是点睛之笔? 它相当于图书馆每个主题区门口贴的「本区域概览海报」。回答全局问题时,不必读完所有原文,只需让 LLM 对各社区摘要做一次 Map-Reduce 式汇总:先让 LLM 分别基于每份社区报告给出局部答案(Map),再归纳成全局答案(Reduce)。全局信息因此被「预先压缩」进了层级摘要里。

6.3 查询阶段:Local / Global / DRIFT 三种模式

模式适用问题工作方式
Local Search关于某个具体实体的问题(「X 项目负责人是谁」)从图谱定位实体,扩展其关系、文本块、社区报告后作答
Global Search全局性/归纳性问题(「整个语料的主要主题/趋势」)遍历社区报告做 Map-Reduce 综合,成本高但全局感强
DRIFT Search兼顾广度与深度先 Global 获得概览方向,再沿方向做 Local 下钻(微软后续提出的混合模式)

6.4 最小实现思路(伪代码 + 关键步骤)

GraphRAG 完整工程较重(需图谱构建、Leiden 聚类、大量 LLM 调用),这里给出理解原理的最小骨架;生产中建议直接用官方 graphrag 包。

# graphrag_mini.py —— 原理级最小骨架(省略 LLM 解析的容错细节)
import networkx as nx
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="deepseek-v4-flash", temperature=0)

# ---------- 索引阶段 ----------
# ① 从每个文本块抽取 (实体, 关系, 实体) 三元组
def extract_triples(chunk: str):
    prompt = f"""从下面文本抽取实体与关系,每行一个三元组,格式:头实体 || 关系 || 尾实体
文本:{chunk}"""
    lines = llm.invoke(prompt).content.splitlines()
    triples = []
    for ln in lines:
        parts = [p.strip() for p in ln.split("||")]
        if len(parts) == 3:
            triples.append(tuple(parts))
    return triples

G = nx.Graph()
for chunk in all_chunks:
    for h, r, t in extract_triples(chunk):
        G.add_edge(h, t, relation=r)

# ③ 社区检测(需要 pip install python-louvain;官方用 Leiden)
import community as community_louvain
partition = community_louvain.best_partition(G)   # {实体: 社区id}

# ④ 为每个社区生成摘要(官方还会自底向上做多层级)
from collections import defaultdict
comm_nodes = defaultdict(list)
for node, cid in partition.items():
    comm_nodes[cid].append(node)

community_reports = {}
for cid, nodes in comm_nodes.items():
    sub = G.subgraph(nodes)
    edges = [(u, G[u][v]["relation"], v) for u, v in sub.edges()]
    report = llm.invoke(
        f"以下是一个主题社区的实体与关系,请归纳其主题、关键实体与核心洞察:\n{edges}"
    ).content
    community_reports[cid] = report

# ---------- 查询阶段:Global Search = 社区报告的 Map-Reduce ----------
def global_search(question):
    # Map:每份社区报告独立回答
    partials = [
        llm.invoke(f"基于社区摘要回答问题,无关就回答“无”:\n摘要:{rep}\n问题:{question}").content
        for rep in community_reports.values()
    ]
    partials = [p for p in partials if p.strip() != "无"]
    # Reduce:汇总所有局部答案为全局答案
    return llm.invoke(f"把以下多份局部结论综合成一个全面回答:\n{partials}\n问题:{question}").content

6.5 代价与适用边界

  • 代价:离线建图要对全语料做大量 LLM 抽取与摘要,索引成本和时间远高于向量 RAG;文档更新后增量建图也更麻烦。
  • 适用:语料相对稳定、需要跨文档归纳/洞察/关系网络分析的场景(舆情分析、情报研判、大型代码库/文献库理解)。
  • 不适用:文档高频变动、只是简单事实问答——那是向量 RAG 的主场。
  • 轻量替代:LightRAG、HippoRAG 等在「图结构带来的全局关联能力」和「成本」之间做折中,可作为延伸阅读。

7. Modular RAG(模块化 RAG,承上启下)

7.1 是什么

Gao 等人综述中提出的第三个学术范式。它把 RAG 拆成可替换、可重组、可循环调用的功能模块(Retriever、Memory、Router、Fusion、Predictor 等),模块之间不再是一条固定直线,而可以根据任务动态编排、甚至多次循环

  • 可重组:比如先路由(Router)判断走哪类索引;检索结果用 Fusion 模块融合多路答案。
  • 可迭代:检索 → 生成 → 发现不够 → 再检索,形成循环(这一点已经和 Agentic RAG 非常像)。
  • 可融合训练:把检索器微调、强化学习等也纳入模块体系。

7.2 为什么它是「承上启下」的一环

flowchart LR
    subgraph Naive
        N1[Retrieve] --> N2[Generate]
    end
    subgraph Advanced
        A1[Pre] --> A2[Retrieve] --> A3[Post] --> A4[Generate]
    end
    subgraph Modular
        M1[Router] --> M2[Retriever A/B/C]
        M2 --> M3[Fusion]
        M3 --> M4[Predictor]
        M4 -. 不足则再次检索 .-> M2
    end
    Naive --> Advanced --> Modular --> Agentic[Agentic RAG<br/>编排权交给 LLM Agent]

Modular RAG 在工程架构上准备好了「模块 + 路由 + 循环」的乐高积木,但「谁来决定怎么搭积木」早期仍是代码编排;Agentic RAG 做的最后一步,就是把这个「编排者」也换成 LLM——由模型在运行时自己决定调用哪个模块、调用几次。理解这条脉络,就理解了 Agentic RAG 不是凭空冒出来的概念。


8. Agentic RAG(重点)

8.1 是什么:一句话祛魅

Agentic RAG = 传统 RAG + 一个具有自主决策能力的 Agent。 只要 RAG 过程中存在模型自主决策(自己决定搜不搜、搜什么、搜几次、读哪些片段、何时停止),而不是走写死的流水线,就可以叫 Agentic RAG。

它的核心不是用了更复杂的模型,而是把模型从「流水线末端的生成器」变成「全程的决策-执行控制器」:先制定策略,再像人查资料一样分步调用工具收集证据,最后基于读到的证据作答并给出引用。Anthropic 把这种能力称为 Agentic Search(智能体式搜索),并认为它是 Agent 应用区别于传统 RAG 的关键。

8.2 本质区别:固定流水线 vs 自主决策循环

flowchart TB
    subgraph TRAD["传统 RAG:开发者写死的一条直线(模型无决策权)"]
        direction LR
        t1[Query] --> t2[固定检索一次] --> t3[固定拼 Top-K] --> t4[生成]
    end

    subgraph AGT["Agentic RAG:模型驱动的循环(每一步都由模型决定)"]
        a1[Query] --> a2[LLM 思考: 我需要什么证据]
        a2 --> a3[行动: 调用某个检索/读取工具]
        a3 --> a4[观察: 工具返回结果]
        a4 --> a5{证据够了吗?}
        a5 -->|不够: 改写查询/换工具/读相邻片段| a2
        a5 -->|够了| a6[基于证据生成 + 标注引用]
    end
对比维度传统/Advanced RAGAgentic RAG
控制流开发者写死的线性流程模型在运行时动态决定行动序列
检索次数固定 1 次(或固定 N 次)按需多轮,直到模型判断证据充分
检索的角色流程中固定的一环被封装成工具(Tool),模型可选择调用与否
失败处理一次没搜好就硬答回溯、改写查询、换工具、补读上下文
证据利用Top-K 整体塞入先粗后细(coarse-to-fine):粗筛 → 看元信息 → 精读片段
可归因性只能给一批来源可精确引用「读了哪个文件的哪个片段」
代价低、延迟可预测token 消耗与延迟更高、流程不完全可预测

8.3 理论基石:ReAct(Reason + Act)

Agentic RAG 的循环直接源自 ReAct 范式。

  • 论文:Yao et al., ReAct: Synergizing Reasoning and Acting in Language Models(Princeton + Google, ICLR 2023, arXiv:2210.03629)。
  • 要解决的矛盾:纯思维链(CoT)只会「想」,一旦内部知识是错的就越想越偏;纯行动(只会调工具)没有计划,容易反复做无效动作。
  • 核心机制:把模型的动作空间从「只能调外部工具」扩展为「外部动作 Action + 内部语言 Thought」,让 Thought → Action → Observation 三元组不断循环:
    • Thought(思考):用自然语言做计划、追踪进度、处理异常,相当于可读的「工作记忆」;
    • Action(行动):调用工具(搜索、读片段、查元数据……);
    • Observation(观察):工具返回结果,拼回上下文,供下一轮 Thought 使用。
sequenceDiagram
    participant U as 用户
    participant L as LLM(大脑)
    participant T as 工具集(检索/读取)
    U->>L: 问题
    loop ReAct 循环,直到证据充分
        L->>L: Thought:分析当前已知与缺口,决定下一步
        L->>T: Action:调用具体工具(参数)
        T-->>L: Observation:返回结果
    end
    L->>U: 基于已读取证据的最终答案 + 引用

一段典型的 ReAct 轨迹长这样(这也是后面 LangGraph create_react_agent 自动帮你维护的东西):

Thought: 我需要先在知识库里搜索“函数调用”相关的文件。
Action: query_knowledge_base("LangChain 函数调用 实现")
Observation: 命中 3 个片段,最高相似度仅 0.65,可能术语不匹配。
Thought: 文档可能使用英文术语 Function Calling,我换英文关键词再搜一次。
Action: query_knowledge_base("LangChain Function Calling")
Observation: 命中 fileId=42 的 chunk8,相似度 0.89,含实现细节。
Thought: chunk8 上下文不完整,我需要读它前后相邻的 chunk7、chunk9。
Action: read_file_chunks([{"fileId":42,"chunkIndex":7},{"fileId":42,"chunkIndex":9}])
Observation: 拿到完整实现上下文。
Thought: 证据已充分,组织答案并标注引用。
Answer: ……(引用:fileId 42, chunk 7/8/9)

8.4 核心设计:Search Ability as a Tool(把检索变成工具)

这是 Agentic RAG 最关键的工程思想,也是它和 MCP(Model Context Protocol,模型上下文协议)天然契合的原因:不再让检索成为必经流程,而是把各种检索/读取能力封装成一个个「自包含、职责单一」的工具,由模型按需调用

参考开源企业级项目 Chatbox(36.8k+ star,较早把知识库问答做成 Agentic 形态的桌面 LLM 客户端)的设计,一套好用的知识库工具集通常是四级、由粗到细的:

工具职责在「先粗后细」中的位置为什么需要它
query_knowledge_base(query)语义搜索,返回候选文件/片段「线索」① 粗筛最基础的入口,快速定位方向
list_files(page, pageSize)列出知识库文件清单① 粗筛(兜底)搜索线索不足时,像翻目录一样浏览
get_files_meta(fileIds)看候选文件的元信息(文件名、大小、chunk 数)② 定位帮模型决定「该读哪几个文件的哪部分」,避免盲读
read_file_chunks([{fileId, chunkIndex}])按「文件 ID + 块序号」精读具体片段③ 精读取证只把真正相关的少量原文读进上下文,降噪、省 token,且能精确引用

为什么要拆成四个,而不是一个「大检索」接口? 这正是「证据利用浅」(Naive 缺陷 4)的解药:

  1. 语义搜索便宜但粗,只用来找线索,不直接把结果当证据;
  2. 元信息几乎不占 token,却能让模型做「读哪个」的明智决策;
  3. 精读接口让模型自己选片段,还能在片段不完整时主动读相邻 chunk 补全上下文(Naive RAG 做不到);
  4. 每个工具职责单一、描述清晰,模型才不容易调错——这也符合 Anthropic 上下文工程中「工具应自包含、互不重叠」的原则。

8.5 两个走查案例:体会「自主决策」到底强在哪

案例 1:术语不匹配 → 自主改写查询(对应 Naive 缺陷 5) 场景:用户问「LangChain 的函数调用怎么实现?」,而文档用的是英文 “Function/Tool Calling”,首次中文检索相似度只有 0.65。

  • Naive RAG:一次检索命中低 → 直接回答「未找到相关信息」,失败。
  • Agentic RAG:
    • 第 1 轮 query_knowledge_base("LangChain 函数调用 实现") → 观察到命中低;
    • Thought:「函数调用」对应英文 Function Calling → 第 2 轮换英文搜,命中 0.89;
    • (可选)第 3 轮补搜 Tool Calling → 证据充分后作答。

案例 2:分块截断 → 自主补全相邻上下文(对应 Naive 缺陷 4) 场景:关键实现被切分策略拆断,命中的 chunk8 内容不完整。

  • Agentic RAG:
    • 第 1 轮搜到 fileId=42 的 chunk8,发现它「讲到一半」;
    • Thought:需要前后文 → 第 2 轮 read_file_chunks(chunk7, chunk9)
    • 拼上 chunk 7/8/9 得到完整实现,再作答并精确引用。

这两个案例里,改写查询、补读相邻块都不是开发者写死的分支,而是模型根据 Observation 临场决定的——这就是「Agentic」的含义。

8.6 怎么实现:LangGraph 最小可运行 Agentic RAG

完整可运行版见配套 Notebook。依赖追加:pip install langgraph。 下面用预置 ReAct Agent(1.x 推荐 from langchain.agents import create_agent;旧名 langgraph.prebuilt.create_react_agent 仍可用但有弃用告警,它内部自动维护 8.3 的 Thought-Action-Observation 循环)。

# agentic_rag.py —— 最小可运行的 Agentic RAG(四级工具 + ReAct 循环)
import os
from typing import List, Dict, Any
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent   # 1.x 推荐;旧名 langgraph.prebuilt.create_react_agent

# ============================================================
# 0. 知识库控制器:真实项目中由你的平台/向量库提供。
#    这里给出接口约定,Notebook 里用本地内存向量库实现了一个零依赖可运行版本。
#    - search(kb_id, query)            语义搜索,返回候选 chunk 线索
#    - list_files_paginated(...)       列出文件清单
#    - get_files_meta(kb_id, file_ids) 取文件元信息
#    - read_file_chunks(kb_id, chunks) 按 fileId+chunkIndex 精读原文
# ============================================================
kb_controller = ...  # 见 Notebook 中的 LocalKBController
KB_ID = 0

# ---------- 1) 把四种检索/读取能力封装成工具 ----------
@tool
def query_knowledge_base(query: str) -> Any:
    """在知识库中做语义搜索,返回相关文件与片段的线索(粗筛)。"""
    return kb_controller.search(KB_ID, query)

@tool
def list_files(page: int = 1, page_size: int = 20) -> Any:
    """列出知识库中的文件清单(id、文件名、chunk 数量),用于浏览兜底。"""
    return kb_controller.list_files_paginated(KB_ID, page, page_size)

@tool
def get_files_meta(file_ids: List[int]) -> Any:
    """根据文件 id 列表查看文件元信息(文件名、大小、chunk 数),辅助决定读哪里。"""
    if not file_ids:
        return "请提供文件 id 列表。"
    return kb_controller.get_files_meta(KB_ID, file_ids)

@tool
def read_file_chunks(chunks: List[Dict[str, int]]) -> Any:
    """精读指定片段,参数形如 [{"file_id": 1, "chunk_index": 3}]。用于最终取证。"""
    if not chunks:
        return "请提供要读取的 [{file_id, chunk_index}] 列表。"
    return kb_controller.read_file_chunks(KB_ID, chunks)

tools = [query_knowledge_base, list_files, get_files_meta, read_file_chunks]

# ---------- 2) 系统提示词 = 行为策略(把“先粗后细 + 必须引用”固化为策略) ----------
SYSTEM_PROMPT = """你是一个 Agentic RAG 研究助手,必须按“先粗后细”的方式取证后再回答:
1. 先用 query_knowledge_base 搜索线索;线索不足时用 list_files 浏览、get_files_meta 看文件信息。
2. 不要只凭搜索线索直接作答,最终必须用 read_file_chunks 精读少量最相关片段。
3. 若片段不完整,主动读取其相邻 chunk;若搜索词命中差,换同义/英文术语重搜。
4. 严禁编造。证据不足就明确说明,并给出下一步建议。
5. 回答末尾用“引用:”列出你实际精读的 file_id 与 chunk_index(或文件名)。"""

# ---------- 3) 模型 + ReAct Agent(模型必须支持 tool calling) ----------
llm = ChatOpenAI(
    model=os.getenv("CHAT_MODEL", "deepseek-v4-flash"),
    temperature=0,
    base_url=os.getenv("CHAT_BASE_URL", "https://api.deepseek.com/v1"),
    api_key=os.getenv("LLM_API_KEY"),
)
# 1.x 用 system_prompt=(旧 create_react_agent 用 prompt=)
agent = create_agent(llm, tools, system_prompt=SYSTEM_PROMPT)

# ---------- 4. 一次调用,内部自动多轮“思考-工具-观察” ----------
result = agent.invoke({"messages": [("user", "LangChain 的函数调用功能是怎么实现的?请给出引用")]})
print(result["messages"][-1].content)

代码设计点解读(为什么这么写)

  • 工具的 docstring 不是注释,是给模型看的「使用说明书」:模型靠它判断何时该调哪个工具,务必写清「做什么、什么时候用、参数格式」。
  • system_prompt 里固化的是「策略」而非「流程」:我们没有规定它必须搜几轮、读哪个块(那是它自主决定的),只规定了「先粗后细、必须精读取证、必须引用」的原则。这正是 Agentic 与写死流程的分界线。
  • temperature=0:工具决策需要稳定、可复现,不宜发散。
  • 模型必须支持 Function/Tool Calling:ReAct 的 Action 要靠模型输出结构化工具调用,DeepSeek-V3、Qwen3、GPT、Claude、GLM 等均支持。

8.7 工业级形态:从 Chatbox 到 Deep Research

  • Chatbox(客户端形态):作为离线桌面应用,对延迟不敏感,因此可以「更激进」地把大量决策交给 LLM——先用判断 Prompt 决定是否需要检索,再让模型在四级文件工具中自主游走;对不支持工具调用的模型,它还退化为一套「Prompt 模拟 search/proceed 动作」的方案(介于 Naive 与 Agentic 之间)。这解释了为什么它在复杂场景下的知识库问答效果优于同类产品。
  • Anthropic 的观点(Context Engineering):Agent 的核心是「在正确的时间把正确的 token 放进上下文」。Agentic Search 让模型通过工具按需、增量地把证据拉进上下文,而不是一次性塞满,从而同时缓解上下文污染与「大海捞针」。
  • OpenAI Deep Research(产品级巅峰形态):基于推理模型(o 系列)的特化版本 + 端到端强化学习训练而成。它能自主澄清意图、制定多步研究计划、执行几十上百次网页浏览与检索、走不通时回退、最后产出带引用的长篇研究报告。这已经是「Agentic RAG + 长程任务规划」的综合体,也是第 9 章 Search-R1 的工业对应物。

8.8 优势与代价(客观看待)

优势:① 适应性强(改写、补搜、回退);② 证据利用深(先粗后细、读到再答);③ 可归因(精确引用);④ 对多跳、模糊、术语不匹配等难题显著更稳。 代价:① 多轮工具调用导致延迟和 token 成本上升,流程不可完全预测;② 强依赖模型的工具调用能力与提示策略,弱模型效果差;③ 需要设计良好的工具粒度与可观测性(trace),否则难以调试;④ 要设最大轮数等「护栏」,防止模型陷入无效循环。


9. 用强化学习训练 Agentic RAG:Search-R1

9.1 为什么还需要强化学习(提示词方案的天花板)

第 8 章的 Agentic RAG 依赖人工写的提示词和工具描述来引导决策,它有三个天花板:

  1. 策略是人「拍脑袋」设计的:何时改写、何时补搜、何时停,全靠 Prompt 约束,难以覆盖复杂多变的真实问题;
  2. 无法从经验中学习:同样的弯路会反复走,不会因为做过一万次而变聪明;
  3. 多轮搜索的长程决策很难用规则写全:每一步都依赖前面所有观察,是一个序列决策问题。

Search-R1(Jin et al., 2025,复现/对齐 DeepResearch 思路的代表性开源工作)给出的答案是:用强化学习(RL)直接训练模型,让它自己学会「何时搜、搜什么、怎么用搜索结果」,而不是靠人教。

9.2 是什么:推理与搜索交织(Interleaved Reasoning & Retrieval)

模型在一条生成轨迹中,可以多次暂停推理、吐出一个搜索查询、取回结果、再继续推理,形成 推理 → 搜索 → 再推理 → 再搜索 … → 最终答案 的循环,直到自己判断信息足够:

flowchart TD
    Q[用户问题] --> T[LLM 推理 think]
    T --> D{需要搜索吗?}
    D -->|是| S1["生成 <search>查询</search>"]
    S1 --> S2[执行检索]
    S2 --> S3["结果包上 <information>...</information> 拼回上下文"]
    S3 --> T
    D -->|否| A["生成 <answer>最终答案</answer>"]

模型用特殊标记控制流程(这是 Search-R1 的关键「协议」):

  • <search>查询词</search>:请求一次检索,训练/推理框架拦截它、执行搜索、注入结果;
  • <information>检索结果</information>:检索结果被包裹后拼回;
  • <answer>答案</answer>:给出最终答案,轨迹结束。

9.3 三个关键设计(为什么这么设计)

  1. 检索令牌掩码(Retrieval Token Masking):在计算 RL 的策略梯度时,把「检索返回的内容」这些并非模型自己生成的 token 屏蔽掉,不让模型为外部塞进来的结果「背梯度」——否则模型会试图去拟合随机的检索文本,训练极不稳定。同时保证生成时这些位置由检索器而非模型填充。
  2. 基于结果的简单奖励(Outcome Reward):只对最终答案与标准答案比对给奖励(正确 1 / 错误 0,或用格式奖励 + EM/F1),不对中间搜索步骤做繁琐奖励。简单但有效,把「怎么搜」的策略空间完全留给模型自己探索。
  3. Rollout 算法(多轮搜索引擎调用):训练采样时,模型一边生成一边被框架检测特殊标记,遇到 <search> 就真的去检索并把结果注入,再继续生成,直到 <answer> 或达到最大动作预算 B。算法主干如下(译自论文 Algorithm 1):
输入:问题 x、策略模型 πθ、搜索引擎 R、最大动作数 B
1: 初始化轨迹 y ← ∅,动作计数 b ← 0
2: while b < B do
3:   让模型逐 token 生成,直到产出 </search>、</answer> 或 <eos>
4:   if 检测到 <search>...</search>:
5:       解析出查询 q,检索 d = R(q),把 <information>d</information> 拼回轨迹
6:   else if 检测到 <answer>...</answer>:
7:       返回最终答案 y
8:   else(格式错误):
9:       追加 “My action is not correct. Let me rethink.” 让它重来
10:  b ← b + 1

9.4 训练流程与最小实现

训练四步:① 用预训练模型(Qwen2.5、Llama3 等)初始化;② 采样出含多轮搜索的轨迹;③ 用最终答案正确性算奖励;④ 用 PPO/GRPO 做策略优化(Search-R1 实际用 GRPO,下面用最易理解的 Policy Gradient 演示原理)。

# search_r1_mini.py —— 原理级最小实现(仅用于理解,非可直接训练的生产代码)
import torch
from torch.optim import Adam
from transformers import AutoModelForCausalLM, AutoTokenizer

model = AutoModelForCausalLM.from_pretrained("Qwen/Qwen2.5-7B")
tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen2.5-7B")

class SearchEngine:
    def search(self, query: str) -> str:
        ...  # 接本地检索器或在线搜索 API,返回文本结果

def generate_trajectory(model, tokenizer, question, search_engine, max_actions=5):
    """采样一条“推理-搜索”交织的轨迹,并记录每步 log-prob"""
    state, log_probs, actions, b = question, [], [], 0
    while b < max_actions:
        inputs = tokenizer(state, return_tensors="pt")
        out = model.generate(**inputs, max_new_tokens=128,
                             output_scores=True, return_dict_in_generate=True)
        text = tokenizer.decode(out.sequences[0][inputs["input_ids"].shape[1]:])
        step_lp = torch.log_softmax(out.scores[-1], dim=-1).max()
        log_probs.append(step_lp)

        if "<search>" in text and "</search>" in text:
            actions.append("search")
            query = text.split("<search>")[1].split("</search>")[0].strip()
            result = search_engine.search(query)
            state = state + text + f"<information>{result}</information>"  # 注入检索结果
            b += 1
        elif "<answer>" in text:
            actions.append("answer")
            break
        else:
            state = state + text + " My action is not correct. Let me rethink."
            b += 1
    return text, actions, torch.stack(log_probs)

def compute_reward(prediction: str, ground_truth: str) -> float:
    """结果奖励:只看最终答案(真实 Search-R1 还会加格式奖励)"""
    return 1.0 if ground_truth.strip().lower() in prediction.strip().lower() else 0.0

def train_rl(model, dataset, search_engine, epochs=1, lr=1e-5):
    opt = Adam(model.parameters(), lr=lr)
    for _ in range(epochs):
        for question, gt in dataset:
            with torch.no_grad():
                final, _, log_probs = generate_trajectory(model, tokenizer, question, search_engine)
            reward = compute_reward(final, gt)
            loss = -(log_probs * reward).mean()   # Policy Gradient:REINFORCE(Search-R1 用 GRPO)
            opt.zero_grad(); loss.backward(); opt.step()
    return model

注意:上面省略了 Search-R1 真正的工程难点——检索 token 掩码的精确实现、GRPO 的组内优势归一化、分布式 Rollout 与检索服务吞吐。想深入请直接读论文与官方仓库(见第 12 章)。普通业务中不需要自己训练,用第 8 章的「提示词 + 工具」方案即可;RL 路线是模型厂商/研究团队打造 DeepResearch 级产品时的选择。

9.5 三代方案的关系

传统 RAG提示词驱动 Agentic RAGRL 驱动 Agentic RAG(Search-R1)
决策机制固定流程人工 Prompt 规则从奖励中学习得到
搜索次数单次多次(规则引导)自适应多次(策略学到)
适应性
实现/训练成本高(需 GPU、训练数据、检索环境)

10. 全景对比

把所有范式放在一张表里横向对比(没有银弹,按问题复杂度和成本选):

范式核心思想决策主体检索次数擅长问题成本/延迟实现复杂度
Naive RAG检索一次拼 Prompt无决策,写死1简单事实问答
Advanced RAG检索前/中/后固定优化开发者预设1(多路召回)术语不匹配、要高召回/高精度低-中
Self-RAG模型用反思令牌自检模型(按训练/提示)按需需要抑制幻觉、判断是否检索中-高(原版需微调)
CRAG检索质量评估 + 纠错/Web 兜底评估器分支1-2知识库覆盖不全、检索易出错
Adaptive RAG按问题复杂度路由不同策略分类器路由按类型问题难易混杂的线上系统
GraphRAG知识图谱 + 社区摘要固定(换索引结构)全局/局部跨文档全局归纳、关系网络索引成本高
Agentic RAG检索即工具,ReAct 自主多轮取证LLM 自主按需多轮多跳、模糊、需精读与引用中-高
RL Agentic(Search-R1)用 RL 训练搜索策略训练得到的策略自适应多轮DeepResearch 级开放研究训练成本极高极高

一条主线再强调一遍:从左到右,「检索什么、检索几次、是否采信」的决策权,逐步从开发者代码转移到模型运行时判断,最终到训练过程中习得


11. 选型决策指南

11.1 决策树

flowchart TD
    START[要做知识问答] --> Q1{问题大多是简单事实查询?}
    Q1 -->|是| N[Naive RAG 起步<br/>够用就别过度设计]
    Q1 -->|否| Q2{检索经常不准/术语不匹配?}
    Q2 -->|是| ADV[加 Advanced 优化<br/>混合检索 + Rerank + 查询改写]
    Q2 -->|否| Q3{知识库经常覆盖不到/会答错?}
    Q3 -->|是| CRAG[上 CRAG:质量评估 + Web 兜底]
    Q3 -->|否| Q4{需要跨大量文档做全局归纳/关系分析?}
    Q4 -->|是| GR[GraphRAG]
    Q4 -->|否| Q5{存在多跳推理/需要精读片段/精确引用?}
    Q5 -->|是| AG[Agentic RAG:工具化 + ReAct 多轮]
    Q5 -->|否| N
    AG --> Q6{提示词策略仍达不到要求<br/>且有训练资源?}
    Q6 -->|是| RL[RL 路线 / Search-R1 思路]

11.2 落地建议(工程经验)

  1. 从 Naive RAG 开始,用评测驱动迭代:先建一个 20–50 条的问答评测集,量化「检索召回率 / 答案正确率 / 引用命中率」,每加一个模块都用数据证明它有效,别一上来堆全套。
  2. 性价比最高的两步:① 混合检索(向量 + BM25);② 加一个 Reranker。多数业务场景做到这两步就够了。
  3. Agentic RAG 的触发信号:当你发现问题需要「跨多个文档串联」「第一次总是搜不到、换个词就行」「需要回答时给出处」时,再上 Agent,不要为了概念而 Agent。
  4. Agentic 必备护栏:设置最大工具轮数、每轮 trace 可观测、工具描述清晰单一、低置信度时允许明说「不知道」。
  5. 成本控制:粗筛工具返回精简线索、只在 read_file_chunks 精读时拉原文;用小模型做改写/打分、大模型做最终综合。
  6. RL 路线是厂商游戏:绝大多数团队不需要训练 Search-R1,理解其思想、用好「提示词 + 工具 + 评测」即可。

12. 参考资料

主线参考

  1. 袁朝发,《RAG 进化之路:传统 RAG 到工具与强化学习双轮驱动的 Agentic RAG》:https://yuanchaofa.com/post/from-native-rag-to-agentic-rag
  2. 配套源码《动手学习大模型(中文版)》第八章 RAG:https://github.com/bbruceyuan/Hands-On-Large-Language-Models-CN/tree/master/chapter08

综述与经典论文

  1. Lewis et al., Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks, NeurIPS 2020:https://arxiv.org/abs/2005.11401
  2. Gao et al., Retrieval-Augmented Generation for Large Language Models: A Survey(Naive/Advanced/Modular 三范式):https://arxiv.org/abs/2312.10997
  3. Gao et al., Precise Zero-Shot Dense Retrieval without Relevance Labels(HyDE):http://boston.lti.cs.cmu.edu/luyug/HyDE/HyDE.pdf
  4. Asai et al., Self-RAG: Learning to Retrieve, Generate, and Critique through Self-Reflections, ICLR 2024:https://arxiv.org/abs/2310.11511 | 项目页 https://selfrag.github.io/
  5. Yan et al., Corrective Retrieval Augmented Generation (CRAG), 2024:https://arxiv.org/abs/2401.15884
  6. Jeong et al., Adaptive-RAG, NAACL 2024:https://arxiv.org/abs/2403.14403
  7. Edge et al., From Local to Global: A Graph RAG Approach to Query-Focused Summarization(Microsoft GraphRAG):https://arxiv.org/abs/2404.16130 | 官网 http://microsoft.github.io/graphrag/
  8. Yao et al., ReAct: Synergizing Reasoning and Acting in Language Models, ICLR 2023:https://arxiv.org/abs/2210.03629 | 项目页 https://react-lm.github.io/
  9. Jin et al., Search-R1: Training LLMs to Reason and Leverage Search Engines with RLhttps://arxiv.org/abs/2503.09516 | 仓库 https://github.com/PeterGriffinJin/Search-R1

工业界官方资料

  1. Anthropic, Effective Context Engineering for AI Agentshttps://www.anthropic.com/engineering/effective-context-engineering-for-ai-agents
  2. Anthropic, Building Effective Agentshttps://www.anthropic.com/engineering/building-effective-agents
  3. OpenAI, Introducing Deep Researchhttps://openai.com/index/introducing-deep-research/
  4. LangChain 官方文档 — Retrieval / Agentic RAG / Self-Correction(CRAG 教程):https://docs.langchain.com/oss/python/langchain/retrieval ;LangGraph CRAG Notebook:https://github.com/langchain-ai/langgraph/tree/main/docs/docs/tutorials/rag
  5. 开源项目 Chatbox:https://github.com/Bin-Huang/chatbox

附:配套动手材料

  • agentic_rag_demo.ipynb:可逐 Cell 运行的演示(环境准备 → Naive RAG → Advanced RAG → Agentic RAG)。
  • .env.example:API Key 与模型配置模板,复制为 .env 后填入你自己的 Key。
  • requirements.txt:全部依赖。