Agentic RAG 与 Deep Research:把检索变成可迭代工具

30 秒导读: 前四章把"怎么检索"讲透了(三路检索 → 检索策略 → 图谱聚合 → 带引用的生成)。 但那都是一问一答:一次检索、一次生成。本章是最深的一层——把那整套检索封装成一个个工具, 交给一个会多轮推理的 agent:它自己决定"要不要搜、搜什么、搜完再想、想完再搜",循环往复, 直到能回答。RAG 模式是"会自己调检索工具的问答";Deep Research 模式在此之上再加推理 / 批判 / Python 执行工具,做多步研究。

本章聚焦编排与工具化,不重复检索内部实现——检索本身请看 01 底层三路检索、02 检索策略与两级 RRF、 03 知识图谱检索与多源聚合、04 RAG 生成与 short-id 引用。

---

1. 这是什么(零基础也能懂)

一句话定义: 把"检索"做成 LLM 能像调函数一样反复调用的工具,再套一个"想一步、调一次工具、 看结果、再想"的循环,让模型自己驱动一次或多次检索来完成任务。

它和普通 RAG 有什么不一样? 一张对比说清:

	普通 RAG(第 04 章)	Agentic RAG(本章)
检索时机	固定:先检索一次,再生成一次	由 LLM 决定:想搜就发一个 tool_call
检索次数	一次	0 次到多次,可迭代
谁在决策	代码写死的流水线	模型自己(推理 → 调工具 → 再推理)
能力边界	答"文档里写了什么"	能拆解问题、多轮查证、跨源综合

给谁用 / 解决什么问题: 当一个问题一次检索答不了——需要先查 A、根据 A 的结果再查 B、 还要算一算、再回头校验——普通 RAG 就卡住了。Agentic RAG 让模型像研究员一样自己安排检索步骤。

用起来什么样: 对外就是一个 agent() 调用,给一条用户消息和一个 mode:

# 示意,非源码:调用方视角
resp = await retrieval_service.agent(
    message=Message(role="user", content="比较 A 公司和 B 公司近三年的营收趋势"),
    rag_generation_config=GenerationConfig(model="anthropic/claude-...", stream=True),
    mode="research",              # "rag" = 单层搜答;"research" = 多步研究
    research_tools=["rag", "reasoning", "critique", "python_executor"],
)
# 内部:模型会自己多次调 rag 工具搜两家公司,调 python_executor 算增长率,调 critique 自查

一句话直觉: 前四章造好了一台"检索机器",本章给这台机器装了个方向盘,然后把方向盘 交到 LLM 手里,让它自己边开边看路。

---

2. 顶层全景(它大概怎么转)

怎么读这张图: 从上到下是一次 agent() 调用的装配与运行。左边是装配期(选出具体 agent), 右边是运行期(agent 的迭代循环)。

                          一次 agent() 调用
                                 │
   ┌─────────────────────────────┼──────────────────────────────┐
   │  装配期(retrieval_service.py)                               │
   │                                                              │
   │  agent()  ── 按 mode 选生成配置/系统提示/工具清单             │
   │     │        rag→quality_llm   research→planning_llm         │
   │     ▼                                                        │
   │  AgentFactory.create_agent(mode × streaming × xml)           │
   │     │  挑一个具体类:R2RRAGAgent / R2RStreamingRAGAgent /     │
   │     ▼                R2RResearchAgent / …(共 8 个变体)      │
   │  具体 Agent 实例(已注册好工具)                              │
   └─────────────────────────────┬──────────────────────────────┘
                                 │ agent.arun(messages, system_instruction)
                                 ▼
   ┌──────────────────────────────────────────────────────────┐
   │  运行期:arun 迭代循环(agent/base.py)                     │
   │                                                            │
   │   while 未完成 and 迭代数 < max_iterations:                │
   │      ① 把整段对话发给 LLM(aget_completion)               │
   │      ② process_llm_response 看回复:                       │
   │           - finish_reason=="stop" → 收尾,完成             │
   │           - 有 tool_calls → 逐个执行                       │
   │      ③ handle_function_or_tool_call:                      │
   │           找到工具 → execute() → 结果塞回对话 → 回到 ①     │
   └───────────────────────┬──────────────────────────────────┘
                           │  每个工具执行时
                           ▼
   ┌──────────────────────────────────────────────────────────┐
   │  工具 = 前四章的检索(agent/rag.py, tools/built_in/*)      │
   │   search_file_knowledge → self.search()(01–03 章)        │
   │   get_file_content       → self.get_context()             │
   │   web_search / tavily / web_scrape → 外部 API             │
   │  结果 → AggregateSearchResult → 截断格式化 → 回喂 LLM      │
   └──────────────────────────────────────────────────────────┘

部件一句话职责:

部件	干什么	在哪(相对 py/core)
RetrievalService.agent	对外入口:验参、装对话、按 mode 定配置与系统提示	main/services/retrieval_service.py:1268
AgentFactory.create_agent	按 mode × streaming × xml 装配出具体 agent 类	main/services/retrieval_service.py:67
R2RAgent.arun	核心迭代循环:反复"问 LLM → 处理回复 → 执行工具"	agent/base.py:101
RAGAgentMixin	把检索方法注册成工具、并做上下文截断	agent/rag.py:39
ResearchAgentMixin	在 RAG 工具之外再挂 reasoning/critique/python 等研究工具	agent/research.py:35
ToolRegistry	自动发现 built_in/ 下的工具类,按名实例化	base/agent/tools/registry.py:14
built_in/*.py	每个内置工具的实现(检索/网页)	base/agent/tools/built_in/

主线走一遍(高层): 用户消息进 agent() → 按 mode 选好模型、系统提示和工具清单 → AgentFactory 挑出具体 agent → arun 进入循环:模型看着对话决定"直接答"还是"发个 tool_call" → 若调工具,就跑对应检索、把结果 拼回对话、再问一遍模型 → 直到模型给出 stop(或撞上 max_iterations 触发兜底总结)。

---

3. 核心原理(逐个机制,由浅入深)

3.1 入口装配:一个 agent(),按 mode 分流

要解决的小问题: 同一个对外接口,既要能做"轻量搜答"(RAG),又要能做"重型研究"(Research), 还要支持流式/非流式。怎么用一处入口把这些差异收敛掉?

思路: agent() 只做决策与准备,真正"选哪个类"交给工厂。它按 mode 做几件事:

• 挑模型: RAG 用 quality_llm,Research 用 planning_llm(未显式指定 model 时)—— 见 retrieval_service.py:1369-1377。
• 挑工具清单: RAG 用 rag_tools,Research 用 research_tools,都可被调用方覆盖, 否则回落到 self.config.agent 的默认值(retrieval_service.py:1479-1488)。
• 挑系统提示: 按 mode 走 _build_aware_system_instruction(:1466)。

真实实现: 工厂本身是一张"三层开关表"——mode → is_streaming → use_xml_format, 共 8 个叶子,每个叶子 return 一个具体类:

# 真实源码节选 retrieval_service.py:135-162(AgentFactory.create_agent)
if mode == "rag":
    if is_streaming:
        if use_xml_format:
            return R2RXMLToolsStreamingRAGAgent(...)
        else:
            return R2RStreamingRAGAgent(...)     # ← 最常见的 RAG 路径
    else:
        ...  # R2RXMLToolsRAGAgent / R2RRAGAgent
else:
    ...      # research 分支:四个 Research 变体一一对应

关键细节:

• use_xml_format 目前被硬编码为 False(:126-129 那段判断被注释掉了)——原本想给 deepseek/gemini 走 XML 工具协议,现在四个 XML 变体在工厂里其实走不到,是"留了口子暂未启用"。
• 流式与否来自 generation_config.stream(:132);Research 与 RAG 各自的 streaming 变体 只是基类不同(见 §3.4),工具装配逻辑是共享的。

3.2 迭代循环:arun 怎么把"检索"变成"可反复调用"

要解决的小问题: 普通 RAG 是"检索一次就生成"。要让模型自己决定检索几次,就得有个循环: 问模型 → 如果它想调工具就去调 → 把结果给它 → 再问,直到它说"我答完了"。

思路(非流式 R2RAgent.arun): 一个 while 循环,上限 max_iterations。每轮把整段对话 (含上一轮工具结果)重新发给 LLM,靠 finish_reason 判断是继续还是收尾。

# 真实源码节选 agent/base.py:115-127(R2RAgent.arun 的循环主体)
while not self._completed and iterations_count < self.config.max_iterations:
    iterations_count += 1
    messages_list = await self.conversation.get_messages()
    generation_config = self.get_generation_config(messages_list[-1])
    response = await self.llm_provider.aget_completion(messages_list, generation_config)
    await self.process_llm_response(response, *args, **kwargs)

process_llm_response 决定"停 or 调工具"(agent/base.py:153):

• finish_reason == "stop" → 置 self._completed = True,循环退出。
• 有 message.tool_calls → 把这条 assistant 消息(带 tool_calls)入库,然后 顺序遍历每个 tool_call,逐个交给 handle_function_or_tool_call:

# 真实源码节选 agent/base.py:178-186(OpenAI 分支,多工具顺序执行)
# If there are multiple tool_calls, call them sequentially here
for tool_call in message.tool_calls:
    await self.handle_function_or_tool_call(
        tool_call.function.name,
        tool_call.function.arguments,
        tool_id=tool_call.id,
    )

要点 —— 顺序 vs 并行:

Agent 类型	一轮里多个 tool_calls 怎么跑	出处
非流式 R2RAgent	for 逐个 await,顺序	agent/base.py:179-186、:340-348
流式 R2RStreamingAgent	asyncio.gather(...),并行	agent/base.py:561-570

非流式对 Anthropic 与 OpenAI 走两套分支(:169 与 :193),差别只在怎么把 thinking / tool_use 块组织进对话消息,最后都落到同一句 handle_function_or_tool_call(:340)。

撞上限的兜底: 若循环耗尽 max_iterations 仍未 stop,_generate_llm_summary(:48)会让 模型产出一段以 ⚠️ **Maximum iterations exceeded** 开头的总结,而不是空手而归。

3.3 一次工具调用怎么落地:handle_function_or_tool_call

要解决的小问题: 模型只是吐出"工具名 + 一串 JSON 参数字符串"。怎么把它变成真的一次检索, 再把结果以模型能读的形式塞回对话?

思路: 按名找到工具 → 解析参数 → execute() → 用工具自带的 llm_format_function 把原始结果 格式化成文本 → 作为一条 role="tool" 消息入库,下一轮模型就能看到。

# 真实源码节选 base/agent/agent.py:186-216(handle_function_or_tool_call 主体)
if tool := next((t for t in self.tools if t.name == function_name), None):
    function_args = json.loads(function_arguments)
    merged_kwargs = {**kwargs, **function_args}
    raw_result = await tool.execute(*args, **merged_kwargs)
    llm_formatted_result = tool.llm_format_function(raw_result)
    tool_result = ToolResult(
        raw_result=raw_result,
        llm_formatted_result=llm_formatted_result,
    )

关键细节:

• 双份结果: ToolResult 同时留 raw_result(结构化,AggregateSearchResult,供引用系统按 short-id 反查,见第 04 章)和 llm_formatted_result(纯文本,回喂给模型)。
• Anthropic 的坑: 开启 extended thinking + Anthropic 时,工具结果后会硬塞一条 role="user", content="Continue..."(base/agent/agent.py:237-246)——这是 Claude thinking+tool 的消息续接要求,对别的模型反而有害,所以按模型名开关。
• 工具怎么拿到检索能力: 工具类本身不含检索逻辑,靠 self.context(即 agent)反查。 例如 SearchFileKnowledgeTool.execute 调的是 context.knowledge_search_method,而这个方法 就是 RetrievalService.search(装配时传入,retrieval_service.py:1505)——第 01–03 章那套 三路检索,就是在这里被当成工具跑起来的。

3.4 工具从哪来:ToolRegistry 自动发现 + Mixin 注册

要解决的小问题: 内置工具有七八个,还允许用户放自定义工具。怎么不用手写一长串 import, 就能"按名字要一个工具实例"?

思路两段式:

1. 发现(ToolRegistry): 启动时扫 built_in/ 目录,importlib + inspect 把每个 Tool 子类按 tool_instance.name 存进表(registry.py:47-96);还会扫 R2R_USER_TOOLS_PATH 下的用户工具(:98-143)。要用时 create_tool_instance(name, format_fn, context) 造实例、 注入格式化函数与上下文(registry.py:169)。

2. 注册(RAGAgentMixin): agent 初始化时按 config.rag_tools 逐名向 registry 要实例, 塞进 self._tools:

# 真实源码节选 agent/rag.py:88-98(RAGAgentMixin._register_tools)
for tool_name in set(self.config.rag_tools):
    if tool_instance := self.tool_registry.create_tool_instance(
        tool_name, format_function, context=self
    ):
        self._tools.append(tool_instance)
    else:
        logger.warning(f"Unknown tool requested: {tool_name}")

上下文截断(一个容易忽略的精华): 检索结果可能很长,直接回喂会撑爆上下文。所有 RAG 工具的 llm_format_function 都走 RAGAgentMixin.format_search_results_for_llm,按 max_tool_context_length 做按比例截断:

# 真实源码节选 agent/rag.py:102-112(format_search_results_for_llm)
context = format_search_results_for_llm(results)        # 第 04 章的格式化(带 short-id)
context_tokens = num_tokens(context) + 1
frac_to_return = self.max_tool_context_length / (context_tokens)
if frac_to_return > 1:
    return context                                      # 没超,原样返回
else:
    return context[: int(frac_to_return * len(context))]  # 超了,按比例砍尾

max_tool_context_length 各处默认值不同(RAGAgentMixin.__init__ 默认 10_000, agent/rag.py:55;RAG agent 具体子类给到 20_000;agent() 入口默认 32_768)。

3.5 内置 RAG 工具集

它们是什么: 一组把"检索/取内容/上网"暴露给模型的工具。默认只开前三个本地检索工具, 网页类默认关闭(RAGAgentConfig.rag_tools,base/agent/agent.py:266-276)。

工具名	干什么	底层接到哪	文件
search_file_knowledge	语义/混合检索本地知识库(chunk + 图谱)	context.knowledge_search_method(= RetrievalService.search,第 01–03 章)	built_in/search_file_knowledge.py
search_file_descriptions	按文档级摘要/描述检索	文档级搜索方法	built_in/search_file_descriptions.py
get_file_content	取整篇文档/整块 chunk 的原文	context.content_method(= get_context)	built_in/get_file_content.py
web_search	Google 网页搜索(Serper)	SerperClient	built_in/web_search.py
tavily_search	Tavily 网页搜索	Tavily API	built_in/tavily_search.py
tavily_extract	用 Tavily 抽取网页正文	Tavily API	built_in/tavily_extract.py
web_scrape	抓取指定 URL 正文	抓取器	built_in/web_scrape.py

每个工具都很薄:定义 name/description/parameters(给模型看的 JSON schema),execute 里 从 self.context 取出真正的检索方法调用之。工具是壳,检索是核。

3.6 Deep Research:在检索之上叠推理

要解决的小问题: 有些任务光靠"搜 + 答"不够,还要算、要自我批判、要分步推理。 Research 模式在 RAG 工具之外再挂四个工具。

思路: ResearchAgentMixin 继承 RAGAgentMixin(所以照样有全部 RAG 能力),再在 _register_research_tools 里按 config.research_tools 追加研究工具(agent/research.py:74-95):

研究工具	干什么	底层	出处
rag	把"整个 RAG agent"当成一个工具:内部新起一个 R2RRAGAgent 跑完一整轮搜答,把答案(含引用)回传	嵌套 agent	agent/research.py:97、_rag:208
reasoning	把问题转给一个专用推理模型(reasoning_llm,高 reasoning_effort、低温),无对话记忆	独立 LLM 调用	agent/research.py:121、_reason:281
critique	把当前对话历史喂给推理模型,找逻辑漏洞/偏见/替代方案	复用 _reason	agent/research.py:150、_critique:308
python_executor	在子进程 + 超时里跑 Python(numpy/pandas/sympy 等),做计算	subprocess.run	agent/research.py:180、_execute_python_with_process_timeout:358

最能体现"检索是可迭代工具"的一处 —— rag 工具嵌套 agent:

# 真实源码节选 agent/research.py:239-253(_rag:研究 agent 内部再造一个 RAG agent)
rag_agent = R2RRAGAgent(
    database_provider=self.database_provider,
    llm_provider=self.llm_provider,
    config=config_copy,                       # max_iterations 调成 10,自带 web 工具
    search_settings=self.search_settings,
    rag_generation_config=generation_config,  # 用 quality_llm
    knowledge_search_method=self.knowledge_search_method,
    content_method=self.content_method,
    file_search_method=self.file_search_method,
    max_tool_context_length=self.max_tool_context_length,
)
user_message = Message(role="user", content=query)
response = await rag_agent.arun(messages=[user_message])

于是形成两层 agent 套娃:外层 Research agent 每调一次 rag 工具,内层就跑一整个 RAG 迭代循环(它自己又会多次调 search_file_knowledge 等)。_rag 结束时还会把内层 search_results_collector 里被引用到的结果过户给外层,保证 short-id 引用在最终答案里可反查 (agent/research.py:263-278)。

坑: python_executor 用 subprocess + 10 秒超时做弱隔离(agent/research.py:358-420), 不是强沙箱——它只是"换个进程 + 限时",不做资源/网络隔离。默认 research_tools 里 critique、python_executor 也是开着的(base/agent/agent.py:279-285),与代码注释里写的 "DISABLED by default" 并不完全一致,以实际列表为准。

---

4. 类继承全景(8 个变体怎么来的)

怎么读: 纵轴是"基础 agent 能力"(是否流式、是否 XML 协议),横轴是"挂哪套工具" (RAG 工具 / Research 工具)。每个具体类都是"一个基类 + 一个 Mixin"多继承拼出来的。

   基础 agent(agent/base.py)          + 工具 Mixin(rag.py / research.py)
   ─────────────────────────           ──────────────────────────────────
   R2RAgent(非流式)          ┐
   R2RStreamingAgent(流式)   ├─ + RAGAgentMixin ──────► 4 个 R2R…RAGAgent
   R2RXMLToolsAgent          │       (search/content/web 工具)
   R2RXMLStreamingAgent      ┘
                                   + ResearchAgentMixin ─► 4 个 R2R…ResearchAgent
                                       (rag/reasoning/critique/python)

• Mixin 负责"挂什么工具",基类负责"怎么迭代 / 怎么解析工具调用协议"。
• ResearchAgentMixin 继承自 RAGAgentMixin,所以 Research agent 天然带全部 RAG 工具, 再叠研究工具(agent/research.py:35)。
• XML 系列(R2RXMLTools*)用 <Action><ToolCall> 文本协议解析工具调用(agent/base.py:788 起),给不支持原生 function-calling 的模型用;如 §3.1 所述,工厂当前默认不走它们。

---

5. 巧妙之处(可借鉴的技术)

• 检索即工具,一套复用两处。 同一份 RetrievalService.search(前四章)既服务普通 RAG, 又作为 search_file_knowledge 工具被 agent 反复调用——没有为 agent 另写一套检索。 黏合点就是装配时把方法名传进工具上下文(retrieval_service.py:1505-1507)。
• 工具结果双形态。 ToolResult 同时保留结构化 raw_result 和文本 llm_formatted_result (base/agent/agent.py:213-216):前者喂给引用系统按 short-id 反查,后者喂给模型—— 一次执行,两种消费。
• 按比例截断而非硬砍。 format_search_results_for_llm 用 token 预算 / 实际 token 的比例 去截字符(agent/rag.py:107-112),简单但有效地防止工具结果撑爆上下文。
• agent 套 agent。 Research 的 rag 工具直接 new 一个 R2RRAGAgent 跑完整循环 (agent/research.py:239),把"一次完整的搜答"抽象成"一次工具调用",天然支持分层研究。
• 撞上限也有交代。 迭代耗尽不静默失败,而是让 LLM 产出带 ⚠️ 前缀的进度总结 (agent/base.py:48-94)。

---

6. 边界与局限(诚实)

• XML 工具协议目前走不到。 工厂里 use_xml_format 被写死 False (retrieval_service.py:126-129),四个 R2RXMLTools* 变体实际不会被 create_agent 选中。
• Python 执行是弱隔离。 仅 subprocess + 超时(agent/research.py:358),无资源/网络沙箱, 不适合跑不可信代码。
• 迭代上限即能力上限。 超过 max_iterations 就兜底总结收场(RAG 默认 10,Research 默认 15, 见 agent/research.py:505),复杂任务可能"话没说完"。
• 工具执行的顺序性不一致。 非流式一轮多工具是顺序 await,流式是并行 gather(见 §3.2 表), 行为差异来自实现而非配置。
• 默认工具清单与注释不符。 research_tools 默认把 critique/python_executor 也打开了, 但源码注释标注为 "DISABLED"(base/agent/agent.py:279-285)——以实际列表为准。
• 代码里可见的 FIXME。 search_file_knowledge.execute 内多处 FIXME (built_in/search_file_knowledge.py),提示 settings/embedding 传参历史上踩过坑。

---

7. 横向对比(同组其它章)

关切	本章(第 05 章)	相关章
检索三路怎么实现	只把它当工具调用	01 底层三路检索
basic/RAG-Fusion/HyDE 策略	agent 通过 search_settings 间接用到	02 检索策略与两级 RRF
多源结果如何聚合	工具返回的就是 AggregateSearchResult	03 知识图谱检索与多源聚合
short-id 方括号引用	agent 靠 search_results_collector 反查、跨 agent 过户引用	04 RAG 生成与 short-id 引用
R2R 全景与阅读地图	—	index

一句话: 前四章造"检索机器",本章造"会开这台机器的司机";Research 模式则是"司机手里 还多了纸笔和计算器"。

---

8. 代码地图(导航索引)

主题	文件路径(相对 py/core)	关键符号
对外入口:装配对话与配置	main/services/retrieval_service.py:1268	RetrievalService.agent
工厂:mode×streaming×xml 装配	main/services/retrieval_service.py:67	AgentFactory.create_agent
迭代循环(非流式)	agent/base.py:101	R2RAgent.arun
解析回复 / 决定停或调工具	agent/base.py:153	R2RAgent.process_llm_response
一次工具调用落地	base/agent/agent.py:174	Agent.handle_function_or_tool_call
生成配置装工具 schema	base/agent/agent.py:133	Agent.get_generation_config
流式循环 / 并行执行工具	agent/base.py:369、:561	R2RStreamingAgent.arun
RAG 工具注册	agent/rag.py:72	RAGAgentMixin._register_tools
工具结果截断格式化	agent/rag.py:102	RAGAgentMixin.format_search_results_for_llm
研究工具注册	agent/research.py:74	ResearchAgentMixin._register_research_tools
嵌套 RAG agent(rag 工具)	agent/research.py:208	ResearchAgentMixin._rag
推理 / 批判 / Python 工具	agent/research.py:121、:150、:180	reasoning_tool / critique_tool / python_execution_tool
工具自动发现与实例化	base/agent/tools/registry.py:14、:169	ToolRegistry / create_tool_instance
本地知识检索工具	base/agent/tools/built_in/search_file_knowledge.py	SearchFileKnowledgeTool
默认工具清单	base/agent/agent.py:258	RAGAgentConfig.rag_tools / research_tools
Tool / ToolResult 抽象	py/shared/abstractions/tool.py	Tool / ToolResult