数据截至 2026-08-14(上游 commit 4a030776a31a)

Agentic RAG:把检索封成工具,挂进流式 agent 循环

30 秒导读: 传统 RAG 是「一次性取 top-k 片段,拼进 prompt,让模型照着答」——检索发生在模型开口之前,模型没得选。PrivateGPT 反过来:它把语义检索包成一个名叫 semantic_search 的工具,交给模型;模型在一个流式 agent 循环里自己决定要不要检索、检索几次、每次用什么查询。检索结果作为一条 tool 消息回灌进对话,模型看过之后可以再检索、也可以带着引用直接作答。这一章讲的就是「检索何时、如何被调用」,不重复 02 章 里检索内部怎么算。

---

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

先看两种 RAG 的区别

RAG(Retrieval-Augmented Generation,检索增强生成)的核心是「让模型答题前先查资料」。但「什么时候查、查几次」有两种截然不同的做法。

	传统「取 top-k 塞 prompt」RAG	Agentic RAG(本章)
谁决定检索	系统(代码写死)	模型自己
检索发生在	模型开口之前,一次	循环中任意轮,可多次
查询词是什么	就是用户原话	模型改写/拆解后的查询
模型的角色	被动照着给定片段答	主动「我需要更多信息 → 再查一次」

一句话直觉:把检索从「喂饭」变成「给你一个搜索框,自己去搜」。 模型像一个拿到内部知识库搜索权限的助手——它读了你的问题,觉得需要资料就调 semantic_search,看了结果觉得不够就换个词再搜,够了才开口。

用起来什么样

从模型视角,它收到的工具清单里有这么一个(描述取自源码 tool_placeholders.py:40-45):

semantic_search —— "Perform semantic search over the files in the knowledge base.
Searches should only be done for a single topic at a time, so this tool should be
used multiple times to get better results."

注意最后一句:明确鼓励模型「一次搜一个主题、多搜几次」。这就是 agentic 的味道——工具描述本身在引导模型把一个复杂问题拆成多次检索。

这一章不讲什么

• 不讲检索内部怎么算(向量相似度、树扩展)——那是 02 章。
• 不讲引用怎么锚回原文——那是 03 章。
• 不讲检索结果、指令、工具怎么分层进 prompt——那是 05 章。
• 本章只讲:检索被包成工具、被挂进循环、被模型调用、结果被回灌。

---

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

整个机制有两大块:装配期(把检索服务包成工具、挂进请求)和运行期(agent 循环里模型调用工具)。

装配期:检索服务 → 可调用的工具

ChatService.stream_chat            请求进来
  │                                (chat_service.py:177)
  ▼
ToolPipeline.contextualize_        逐个 processor 把「占位工具」换成「真工具」
  internal_tools                   (tool_pipeline.py:68)
  │
  ▼
SemanticSearchProcessor.intercept  认出 semantic_search 占位符,取出 artifact 上下文
  │                                (semantic_search_processor.py:27)
  ▼
SemanticSearchToolBuilder.         把 SemanticSearchService/Workflow 闭包进一个
  build_tool                       async 函数,产出一个 ToolSpec
  │                                (semantic_search_builder.py:213)
  ▼
ToolSpec(name="semantic_search",   一个 LLM 可调用的工具规格,runtime="server"
         async_fn=semantic_search)

运行期:agent 循环

ChatLoopEngine.run  → _run_loop → 循环 _run_iteration (最多 40 轮)
  │                  (chat_engine.py:233 / 284 / 377)
  │
  每一轮:
  ┌─────────────────────────────────────────────────────────┐
  │ ① 把工具清单 + 对话历史丢给 LLM,流式拿回 chunk           │
  │    run.llm.astream_chat_with_tools(...)  (:420)          │
  │ ② 边流边解析:文字?思考?还是「我要调工具」?           │
  │    _handle_stream_chunk  (:555)                          │
  │ ③ 一旦某个 tool 块流完 → 立刻起一个 task 去执行它         │
  │    _close_active_tool_block → _handle_tool_use  (:761)   │
  │ ④ 本轮流完,等所有 tool task,把结果作为 tool 消息续进对话│
  │    (:499)                                                 │
  │ ⑤ 模型这轮没调工具? → stop_reason=end_turn,结束        │
  │    (:479)                                                 │
  └─────────────────────────────────────────────────────────┘

部件一句话职责

部件	干什么	在哪个文件
SemanticSearchToolBuilder.build_tool	把检索 workflow 闭包成一个 async 函数,产出 ToolSpec	components/tools/builders/semantic_search_builder.py:216
SemanticSearchProcessor.intercept	在请求里认出 semantic_search 占位符,替换成真工具	components/tools/processors/semantic_search_processor.py:27
ToolPipeline	依次跑所有 processor,把内部工具「实体化」	components/tools/tool_pipeline.py:40
SemanticSearchToolBuilderFactory	DI 装配:把各组件注入 builder	components/tools/tool_factories.py:43
GENERAL_INTERNAL_TOOLS	内部检索类工具名单(semantic_search / summarize / web_* / database_query …)	components/tools/tool_names.py:31
ChatLoopEngine	agent 主循环:流式解析 LLM、起 tool task、回灌结果	components/engines/chat/chat_engine.py:204
execute_tool_call	真正 await tool.acall(),把输出转成一条 tool 消息	components/engines/chat/utils/tool_utils.py:43

调用方是 ChatService(server/chat/chat_service.py:222 的 stream_chat)——它 build_loop_engine() 造一个引擎、loop_engine.run(request) 拿事件流。本章不展开它上面的 HTTP/SSE 传输细节。

---

3. 核心原理

3.1 工具化检索:把 SemanticSearchService 包成一个函数

要解决的小问题: LLM 的函数调用接口只认「一个有名字、有描述、有参数 schema 的函数」。但检索是一整套服务(向量库、embedding、树扩展 postprocessor、prompt 组装)。怎么把这一坨,变成模型眼里「一个简单的 semantic_search(query)」?

思路: 用闭包。builder 在装配期把所有重组件(retriever、llm、tokenizer、workflow)绑进一个 async 函数的作用域,对外只暴露 query 和可选 artifacts 两个参数。模型只看得到这两个。

真实实现里,这个对外函数就叫 semantic_search(semantic_search_builder.py:294-310):

# 真实源码 semantic_search_builder.py:291,只保留骨架
async def semantic_search(
    query: str,
    artifacts: list[str] | None = None,
) -> list[ResultContentBlockType]:
    w = await _ensure_workflow(artifacts=artifacts)   # 懒加载检索 workflow(带缓存)
    result = await w.run(                               # 跑 02 章那套检索
        start_event=SemanticSearchInputEvent(
            query=query, use_condense=True, ...
        )
    )
    return await asyncio.to_thread(                     # 把节点格式化成内容块 + 上下文 prompt
        _sync_format_result, result.retrieval.nodes,
    )

这个函数被 ToolSpec.from_defaults 打包成工具规格(semantic_search_builder.py:324-352):

# 真实源码 semantic_search_builder.py:322
return ToolSpec.from_defaults(
    name=name,                       # "semantic_search"
    type=type,                       # "semantic_search_v1"
    runtime="server",                # 关键:server 工具 → 循环里就地执行(见 3.5)
    description=description,          # 就是那句「一次搜一个主题、多搜几次」
    async_fn=semantic_search,        # 模型调用它时真正跑的函数
    async_callback=format_result,    # 把结果块里的文本抽出来喂回模型
)

关键细节:

• _ensure_workflow 带 asyncio.Lock + dict 缓存(semantic_search_builder.py:238-271):同一组 artifacts 的 workflow 只建一次,多次检索复用。这是「模型可能连搜好几次」这一前提下的必要优化。
• 结果不是裸节点。_sync_format_result(:270-289)会调 prompt_builder_service.create_context_prompt,把检索到的 Document 组装成一段带(可选)引用标记的上下文文本——这一步为 03 章 citations 和 05 章上下文装配 埋了线。
• input_schema 不用手写:ToolSpec.from_defaults 会从 semantic_search 的函数签名反射出 JSON schema(chat_config_models.py:317-319 的 build_fn_schema)。改函数签名就等于改工具契约。

3.2 它不是唯一的检索工具:GENERAL_INTERNAL_TOOLS 一览

semantic_search 只是一类「内部工具」里的一个。名单在 tool_names.py:31-38:

工具名(常量)	干什么	占位描述出处
semantic_search	语义检索知识库文件	tool_placeholders.py:40
summarize	对知识库文件做高层摘要	tool_placeholders.py:59
tabular_analysis	对表格文件跑 Pandas 查询	tool_placeholders.py:47
database_query	用自然语言查连接的数据库	tool_placeholders.py:54
web_search	搜网,返回链接 + 摘要	tool_placeholders.py:74
web_fetch	抓取某个 URL 的正文	tool_placeholders.py:66

它们共享同一套「占位符 → processor → builder → ToolSpec」的实体化流程,只是各有各的 processor 和 builder。所以「模型自主调检索」这套设计,天然扩展到「模型自主决定用哪种检索」——是搜本地文件、是摘要、还是上网,由模型按问题挑。

3.3 装配:占位符如何被换成真工具

要解决的小问题: 请求进来时,工具清单里的 semantic_search 只是个占位符——它没有绑定这次请求要搜哪个知识库(哪些 artifact)。得在运行前把它「实体化」成绑好上下文的真工具。

思路: 一条 processor 流水线,每个 processor 认领一种工具、把占位符替换成实体。

ToolPipeline.contextualize_internal_tools(tool_pipeline.py:73-80)把请求深拷贝后依次过一遍全部 processor(上游重构后从「反复跑到稳定」改成了单趟顺序执行):

# 真实源码 tool_pipeline.py:73
async def contextualize_internal_tools(self, request):
    request_copy = request.model_copy(deep=True)
    for processor in self._processors:          # 依次跑所有 processor
        await processor.intercept(request_copy)
    return request_copy

SemanticSearchProcessor.intercept(semantic_search_processor.py:27-57)做的事:

1. 在 request.tool_config.tools 里找未解析的 semantic_search(_tool_matches + _is_unresolved_tool)。
2. 从工具上下文里取出 IngestedArtifact——即这次要搜的知识库;没有就报错「需要 ingested artifact 上下文」。
3. 调 self._builder.build_tool(...),把 context_filter、generate_citations、token_limit 等绑进去,产出一个真 ToolSpec。
4. _replace_tool(request, tool, [resolved]) 把占位符换掉。

组件怎么被塞进 builder,靠 SemanticSearchToolBuilderFactory(tool_factories.py:43-80)的依赖注入:它 @inject 收下 vector_store / node_store / embedding / ingest / parse / prompt_builder 等一堆组件,create() 时组装成 SemanticSearchToolBuilder。这样 builder 才有能力在 build_tool 里真正跑起检索。

3.4 agent 主循环:边流边起工具

这是本章的心脏。ChatLoopEngine 跑一个「让 LLM 说话 → 它要调工具就执行 → 结果回灌 → 再让它说话」的循环,最多 40 轮(chat_engine.py:212 的 max_iterations=40,由 chat_service.py:194 传入)。

主循环骨架(_run_loop_core → while → _run_iteration,chat_engine.py:284-377):

# 真实源码 chat_engine.py:296,循环条件
while (
    not run.stopped
    and run.state.runtime.iteration < run.state.runtime.max_iterations
):
    await self._run_intercepted_iteration(run, handler)

一轮 _run_iteration 里发生什么:

(a) 请模型说话,流式拿回。 run.llm.astream_chat_with_tools(llm_tools, chat_history=..., allow_parallel_tool_calls=...)(:420-425)把工具清单和整段对话历史交给 LLM,拿回一个 chunk 流。工具清单由 _build_tools(:939)从 context stack 取出、经 select_tool_names 过滤(见 3.7)后转成 llama-index 工具。

(b) 边流边解析成 Anthropic 风格事件。 每个 chunk 进 _handle_stream_chunk(:555),它判断这段增量是文字、思考(thinking)、还是工具调用,分别发出对应事件:

LLM 输出的增量	发出的事件	事件类型出处
正文文字	RawContentBlockStartEvent + TextDelta	events/models
思考过程	thinking 块 + ThinkingDelta	chat_engine.py:704
工具调用参数	ToolUseBlock + InputJSONDelta(逐字拼 JSON)	chat_engine.py:664-701

工具调用的参数是一段一段流回来的 JSON。引擎给每个工具调用维护一个 active_tool_block,把增量 JSON 累积进 last_serialized[raw_id](:694-695),边拼边发 InputJSONDelta 让前端能实时看到「模型正在填 query 参数」。

(c) 一个工具块流完,立刻起 task。 当流里出现「切到另一个工具」或整段流结束,_close_active_tool_block(:761-814)会:把最终 JSON 解析成参数对象 → 发 RawContentBlockStopEvent 封块 → 立刻 asyncio.create_task(self._handle_tool_use(...)) 起一个执行任务,塞进 pending_tasks。

这是关键的流水线优化:不等模型把整轮话说完才执行工具,而是边流边执行——模型还在生成后续内容时,前面已确定的工具调用已经在后台跑了。

3.5 串行还是并行:一个信号量说了算

要解决的小问题: 模型一轮可能要调好几个工具(比如同时搜两个主题)。这些工具该并行跑还是排队?

思路: 用一个 asyncio.Semaphore(1) 当串行开关。请求里 allow_parallel_tool_calls 为假时,建一个容量 1 的信号量;为真时不建(=不限并发)。

# 真实源码 chat_engine.py:410
if not run.state.input.request.tool_config.allow_parallel_tool_calls:
    stream_delta_state.tool_state.tool_semaphore = asyncio.Semaphore(1)

执行时(_handle_tool_use,chat_engine.py:1040)用这个信号量把工具执行包起来(上游重构后,引擎不再直接调 execute_tool_call,而是经 _tool_scheduler.execute(ToolExecutionRequest(...)) 调度,最终仍落到 3.6 的 execute_tool_call):

# 真实源码 chat/chat_engine.py:1040
async with semaphore if semaphore is not None else contextlib.nullcontext():
    response = await self._tool_scheduler.execute(ToolExecutionRequest(...))

信号量在 → 同一时刻只有一个工具真正执行(串行);信号量为 None → nullcontext() 空上下文,全部并发。注意:块的解析和 task 的起动始终是流式并发的,信号量只卡住「真正执行」这一步。

3.6 工具执行与结果回灌:让模型再思考

要解决的小问题: 工具跑出结果后,怎么让模型「看到」它、并据此继续?

思路: 把工具输出包成一条 role="tool" 的消息,追加进对话历史。下一轮 LLM 调用时,这条消息就是上下文的一部分——模型自然「读到」了检索结果。

execute_tool_call(tool_utils.py:43-122)做两件事:

1. 执行:await tool.acall(**tool_kwargs)(:52-56)。若工具声明 requires_context,则额外把循环状态 ctx=state_ctx 传进去。异常被兜住,转成 is_error=True 的 ToolOutput(:57-65)——工具报错不会炸掉整个循环,而是作为一条「错误结果」回灌,模型可以据此换个查询重试。
2. 转成 tool 消息:把输出装进 ChatMessage(role="tool", content=..., additional_kwargs={...})(:103-113),带上 tool_call_id(把结果对回是哪次调用)、tool_call_name、tool_call_args、raw_output。

回到 _handle_tool_use,这条 tool 消息被追加进对话(chat_engine.py:1054-1060),同时对外发一个 ToolResultBlock 事件(:1061-1073)让前端渲染。

回灌闭环——一轮完整的检索回合:

模型这轮流出「我要 semantic_search(query='X')」
        │
        ▼
_close_active_tool_block 起 task ──► _handle_tool_use ──► execute_tool_call
        │                                                       │
        │                                          await tool.acall(query='X')
        │                                                       │  跑 02 章检索
        │                                                       ▼
        │                                          ToolOutput(检索到的片段+上下文)
        │                                                       │
        │                                          包成 ChatMessage(role="tool",
        │                                              tool_call_id=..., ...)
        │                                                       │
        ▼                                                       ▼
本轮流完 → await 所有 pending_tasks ──► 对话历史末尾多了这条 tool 消息
        │                                (chat_engine.py:499)
        ▼
下一轮 _run_iteration:astream_chat_with_tools(chat_history=已含 tool 消息)
        │
        ▼
模型「读到」检索结果 → 觉得够了就带引用作答 (stop_reason=end_turn)
              └─ 觉得不够 → 再发一个 semantic_search,回到上面

3.7 循环怎么停:三种 stop_reason

模型自主调工具,那循环靠什么判断「该收尾了」?靠每轮末尾检查有没有工具调用,以及三个终止信号:

stop_reason	何时触发	源码
end_turn	模型这轮没调任何工具 → 它是在正式作答,循环结束	chat_engine.py:479-496
tool_use	出现外部工具(需交给调用方/客户端执行的工具)→ 暂停循环、把它挂进 pending_external_tool_calls	chat_engine.py:530-544
max_tokens	跑满 40 轮还没停 → 强制收尾	chat_engine.py:302-315

end_turn 的判定极简洁——本轮流完后 if not tool_calls:(:479),没有工具调用就置 run.stopped = True。这正是 agentic 循环的收敛条件:模型不再需要工具时,自然停下作答。

tool_use 那条揭示了一个精巧设计:server 工具 vs client 工具。在 _handle_tool_use(chat_engine.py:1006-1014),若工具 runtime != "server",就不执行、直接返回 NOT_EXECUTED,挂进外部待办。而这个标志的来源是 ToolSpec.to_function_tool 里的一行(chat_config_models.py:388):

# 真实源码 chat_config_models.py:386,注释说逻辑是反的
return_direct=self.runtime != "server",

• semantic_search 的 runtime="server" → return_direct=False → 在循环里就地执行、结果回灌(3.1 里那行 runtime="server" 的意义就在这)。
• 用户自带的客户端工具 runtime="client" → return_direct=True → 不执行,把控制权交回调用方。

所以同一个循环,既能就地跑内部检索、也能把外部工具「吐」给调用方——靠的就是这一个反转的布尔值。

---

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

• 检索即工具,而非前置步骤。 把 SemanticSearchService 用闭包封成两参数的 semantic_search(query, artifacts)(semantic_search_builder.py:294),模型就能像调普通函数一样自主检索、多次检索。工具描述里直接写「一次搜一个主题、多搜几次」(tool_placeholders.py:40)来引导拆解式检索。
• 边流边执行,不等整轮说完。 _close_active_tool_block 在单个工具块一流完就 create_task 起执行(chat_engine.py:803),把「模型继续生成」和「工具已开始跑」重叠起来,压低端到端延迟。
• 一个反转布尔值统一 server/client 工具。 return_direct = runtime != "server"(chat_config_models.py:388)让同一条循环既能就地执行内部工具、又能把外部工具交回调用方,分支逻辑收进 _handle_tool_use 一处。
• 工具报错也回灌,而非崩溃。 execute_tool_call 把异常兜成 is_error=True 的 ToolOutput(tool_utils.py:57-65),模型能读到「这次搜失败了」并自行重试。
• workflow 懒加载 + 缓存。 _ensure_workflow 用 frozenset(artifacts) 当 key、双检锁缓存(semantic_search_builder.py:238-271),多次检索复用同一 workflow。
• schema 从函数签名反射。 改 semantic_search 的签名即改工具契约(chat_config_models.py:317-319),无需手写 JSON schema。

---

5. 边界与局限

• semantic_search 只支持单个 ingested 上下文。 SemanticSearchProcessor.intercept 里,超过一个 IngestedArtifact 直接报错「Only one ingested context is supported」(semantic_search_processor.py:44-45);没有 artifact 也报错。
• 循环硬上限 40 轮。 max_iterations=40(chat_engine.py:212),跑满即以 max_tokens 收尾——一个查询若诱发过多次检索,会被截断。
• 必须是支持函数调用的模型。 initialize_run 检查 isinstance(llm, FunctionCallingLLM),否则报「Configured model does not support function calling」(chat_engine.py:868-869)。非 function-calling 模型走不了这套 agentic 检索。
• 本章不覆盖的: 检索内部算法见 02;上下文/工具/指令如何分层进 prompt 见 05;answer 如何锚回原文见 03。

---

6. 横向对比

同库其它检索类工具(summarize、web_search、web_fetch、database_query、tabular_analysis)与 semantic_search 共用同一套 「占位符 → processor → builder → ToolSpec → 挂进同一个 ChatLoopEngine」流程(tool_pipeline.py:58-71 列出全部 processor)。差异只在各自的 processor/builder 内部——对循环而言,它们都只是「一个 runtime="server" 的可调用函数」。这正是「工具化」抽象的价值:新增一种检索能力,不动循环一行代码。

---

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

主题	文件路径	符号名
把检索服务闭包成工具函数	private_gpt/components/tools/builders/semantic_search_builder.py:294	semantic_search(内层 async fn)
产出工具规格	private_gpt/components/tools/builders/semantic_search_builder.py:216	SemanticSearchToolBuilder.build_tool
结果格式化成上下文 prompt	private_gpt/components/tools/builders/semantic_search_builder.py:273	_sync_format_result
占位符 → 真工具的替换	private_gpt/components/tools/processors/semantic_search_processor.py:27	SemanticSearchProcessor.intercept
processor 流水线	private_gpt/components/tools/tool_pipeline.py:73	ToolPipeline.contextualize_internal_tools
builder 的依赖注入装配	private_gpt/components/tools/tool_factories.py:43	SemanticSearchToolBuilderFactory
内部检索类工具名单	private_gpt/components/tools/tool_names.py:31	GENERAL_INTERNAL_TOOLS
工具描述文案	private_gpt/components/tools/tool_placeholders.py:40	SEMANTIC_SEARCH_TOOL_FN
server/client 工具的反转布尔	private_gpt/components/chat/models/chat_config_models.py:373	ToolSpec.to_function_tool(return_direct)
agent 主循环入口	private_gpt/components/engines/chat/chat_engine.py:233	ChatLoopEngine.run
循环核心 / 迭代条件	private_gpt/components/engines/chat/chat_engine.py:284	_run_loop_core
单轮迭代:请模型、起工具、回灌	private_gpt/components/engines/chat/chat_engine.py:377	_run_iteration
流式解析 chunk 成事件	private_gpt/components/engines/chat/chat_engine.py:555	_handle_stream_chunk
工具块流完 → 起执行 task	private_gpt/components/engines/chat/chat_engine.py:761	_close_active_tool_block
执行单个工具调用	private_gpt/components/engines/chat/chat_engine.py:954	_handle_tool_use
并行/串行开关	private_gpt/components/engines/chat/chat_engine.py:410	tool_semaphore / allow_parallel_tool_calls
真正 acall + 转 tool 消息	private_gpt/components/engines/chat/utils/tool_utils.py:43	execute_tool_call
工具选择过滤	private_gpt/components/engines/chat/utils/tool_utils.py:32	select_tool_names
调用方(不展开 HTTP/SSE)	private_gpt/server/chat/chat_service.py:222	ChatService.stream_chat