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:37-42):
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_loop_engine.py:205 / 240 / 332)
│
每一轮:
┌─────────────────────────────────────────────────────────┐
│ ① 把工具清单 + 对话历史丢给 LLM,流式拿回 chunk │
│ run.llm.astream_chat_with_tools(...) (:371) │
│ ② 边流边解析:文字?思考?还是「我要调工具」? │
│ _handle_stream_chunk (:485) │
│ ③ 一旦某个 tool 块流完 → 立刻起一个 task 去执行它 │
│ _close_active_tool_block → _handle_tool_use (:651) │
│ ④ 本轮流完,等所有 tool task,把结果作为 tool 消息续进对话│
│ (:446) │
│ ⑤ 模型这轮没调工具? → stop_reason=end_turn,结束 │
│ (:427) │
└─────────────────────────────────────────────────────────┘
部件一句话职责
| 部件 | 干什么 | 在哪个文件 |
|---|---|---|
SemanticSearchToolBuilder.build_tool | 把检索 workflow 闭包成一个 async 函数,产出 ToolSpec | components/tools/builders/semantic_search_builder.py:213 |
SemanticSearchProcessor.intercept | 在请求里认出 semantic_search 占位符,替换成真工具 | components/tools/processors/semantic_search_processor.py:27 |
ToolPipeline | 依次跑所有 processor,把内部工具「实体化」 | components/tools/tool_pipeline.py:37 |
SemanticSearchToolBuilderFactory | DI 装配:把各组件注入 builder | components/tools/tool_factories.py:42 |
GENERAL_INTERNAL_TOOLS | 内部检索类工具名单(semantic_search / summarize / web_* / database_query …) | components/tools/tool_names.py:28 |
ChatLoopEngine | agent 主循环:流式解析 LLM、起 tool task、回灌结果 | components/engines/chat_loop/chat_loop_engine.py:180 |
execute_tool_call | 真正 await tool.acall(),把输出转成一条 tool 消息 | components/engines/chat_loop/utils/tool_utils.py:43 |
调用方是 ChatService(server/chat/chat_service.py:177 的 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:291-308):
# 真实源码 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:322-329):
# 真实源码 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:235-268):同一组 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:200-202的build_fn_schema)。改函数签名就等于改工具契约。
3.2 它不是唯一的检索工具:GENERAL_INTERNAL_TOOLS 一览
semantic_search 只是一类「内部工具」里的一个。名单在 tool_names.py:28-35:
| 工具名(常量) | 干什么 | 占位描述出处 |
|---|---|---|
semantic_search | 语义检索知识库文件 | tool_placeholders.py:37 |
summarize | 对知识库文件做高层摘要 | tool_placeholders.py:56 |
tabular_analysis | 对表格文件跑 Pandas 查询 | tool_placeholders.py:44 |
database_query | 用自然语言查连接的数据库 | tool_placeholders.py:51 |
web_search | 搜网,返回链接 + 摘要 | tool_placeholders.py:71 |
web_fetch | 抓取某个 URL 的正文 | tool_placeholders.py:63 |
它们共享同一套「占位符 → processor → builder → ToolSpec」的实体化流程,只是各有各的 processor 和 builder。所以「模型自主调检索」这套设计,天然扩展到「模型自主决定用哪种检索」——是搜本地文件、是摘要、还是上网,由模型按问题挑。
3.3 装配:占位符如何被换成真工具
要解决的小问题: 请求进来时,工具清单里的 semantic_search 只是个占位符——它没有绑定这次请求要搜哪个知识库(哪些 artifact)。得在运行前把它「实体化」成绑好上下文的真工具。
思路: 一条 processor 流水线,每个 processor 认领一种工具、把占位符替换成实体。
ToolPipeline.contextualize_internal_tools(tool_pipeline.py:68-74)反复跑,直到没有 processor 再改动请求:
# 真实源码 tool_pipeline.py:68
async def contextualize_internal_tools(self, request):
request_copy = request.model_copy(deep=True)
while await self._intercept_once(request_copy): # 有 processor 改了就 再来一轮
pass
return request_copy
SemanticSearchProcessor.intercept(semantic_search_processor.py:27-57)做的事:
- 在
request.tool_config.tools里找未解析的semantic_search(_tool_matches+_is_unresolved_tool)。 - 从工具上下文里取出
IngestedArtifact——即这次要搜的知识库;没有就报错「需要 ingested artifact 上下文」。 - 调
self._builder.build_tool(...),把context_filter、generate_citations、token_limit等绑进去,产出一个真ToolSpec。 _replace_tool(request, tool, [resolved])把占位符换掉。
组件怎么被塞进 builder,靠 SemanticSearchToolBuilderFactory(tool_factories.py:42-79)的依赖注入:它 @inject 收下 vector_store / node_store / embedding / ingest / parse / prompt_builder 等一堆组件,create() 时组装成 SemanticSearchToolBuilder。这样 builder 才有能力在 build_tool 里真正跑起检索。
3.4 agent 主循环:边流边起工具
这是本章的心脏。ChatLoopEngine 跑一个「让 LLM 说话 → 它要调工具就执行 → 结果回灌 → 再让它说话」的循环,最多 40 轮(chat_loop_engine.py:188 的 max_iterations=40,由 chat_service.py:173 传入)。
主循环骨架(_run_loop_core → while → _run_iteration,chat_loop_engine.py:240-332):
# 真实源码 chat_loop_engine.py:252,循环条件
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=...)(:371-376)把工具清单和整段对话历史交给 LLM,拿回一个 chunk 流。工具清单由 _build_tools(:819)从 context stack 取出、经 select_tool_names 过滤(见 3.7)后转成 llama-index 工具。
(b) 边流边解析成 Anthropic 风格事件。 每个 chunk 进 _handle_stream_chunk(:485),它判断这段增量是文字、思考(thinking)、还是工具调用,分别发出对应事件:
| LLM 输出的增量 | 发出的事件 | 事件类型出处 |
|---|---|---|
| 正文文字 | RawContentBlockStartEvent + TextDelta | events/models |
| 思考过程 | thinking 块 + ThinkingDelta | chat_loop_engine.py:599 |
| 工具调用参数 | ToolUseBlock + InputJSONDelta(逐字拼 JSON) | chat_loop_engine.py:563-596 |
工具调用的参数是一段一段流回来的 JSON。引擎给每个工具调用维护一个 active_tool_block,把增量 JSON 累积进 last_serialized[raw_id](:589-590),边拼边发 InputJSONDelta 让前端能实时看到「模型正在填 query 参数」。
(c) 一个工具块流完,立刻起 task。 当流里出现「切到另一个工具」或整段流结束,_close_active_tool_block(:651-702)会:把最终 JSON 解析成参数对象 → 发 RawContentBlockStopEvent 封块 → 立刻 asyncio.create_task(self._handle_tool_use(...)) 起一个执行任务,塞进 pending_tasks。
这是关键的流水线优化:不等模型把整轮话说完才执行工具,而是边流边执行——模型还在生成后续内容时,前面已确定的工具调用已经在后台跑了。
3.5 串行还是并行:一个信号量说了算
要解决的小 问题: 模型一轮可能要调好几个工具(比如同时搜两个主题)。这些工具该并行跑还是排队?
思路: 用一个 asyncio.Semaphore(1) 当串行开关。请求里 allow_parallel_tool_calls 为假时,建一个容量 1 的信号量;为真时不建(=不限并发)。
# 真实源码 chat_loop_engine.py:361
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,:897)用这个信号量把 execute_tool_call 包起来:
# 真实源码 chat_loop_engine.py:897
async with semaphore if semaphore is not None else contextlib.nullcontext():
result, tool_message = await execute_tool_call(...)
信号量在 → 同一时刻只有一个工具真正执行(串行);信号量为 None → nullcontext() 空上下文,全部并发。注意:块的解析和 task 的起动始终是流式并发的,信号量只卡住「真正执行」这一步。
3.6 工具执行与结果回灌:让模型再思考
要解决的小问题: 工具跑出结果后,怎么让模型「看到」它、并据此继续?
思路: 把工具输出包成一条 role="tool" 的消息,追加进对话历史。下一轮 LLM 调用时,这条消息就是上下文的一部分——模型自然「读到」了检索结果。
execute_tool_call(tool_utils.py:43-122)做两件事:
- 执行:
await tool.acall(**tool_kwargs)(:52-56)。若工具声明requires_context,则额外把循环状态ctx=state_ctx传进去。异常被兜住,转成is_error=True的ToolOutput(:57-65)——工具报错不会炸掉整个循环,而是作为一条「错误结果」回灌,模型可以据此换个查询重试。 - 转成 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_loop_engine.py:906-911),同时对外发一个 ToolResultBlock 事件(:919-932)让前端渲染。
回灌闭环——一轮完整的检索回合:
模型这轮流出「我要 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_loop_engine.py:446)
▼
下一轮 _run_iteration:astream_chat_with_tools(chat_history=已含 tool 消息)
│
▼
模型「读到」检索结果 → 觉得够了就带引用作答 (stop_reason=end_turn)
└─ 觉得不够 → 再发一个 semantic_search,回到上面
3.7 循环怎么停:三种 stop_reason
模型自主调工具,那循环靠什么判断「该收尾了」?靠每轮末尾检查有没有工具调用,以及三个终止信号:
| stop_reason | 何时触发 | 源码 |
|---|---|---|
end_turn | 模型这轮没调任何工具 → 它是在正式作答,循环结束 | chat_loop_engine.py:427-443 |
tool_use | 出现外部工具(需交给调用方/客户端执行的工具)→ 暂停循环、把它挂进 pending_external_tool_calls | chat_loop_engine.py:461-474 |
max_tokens | 跑满 40 轮还没停 → 强制收尾 | chat_loop_engine.py:258-270 |
end_turn 的判定极简洁——本轮流完后 if not tool_calls:(:427),没有工具调用就置 run.stopped = True。这正是 agentic 循环的收敛条件:模型不再需要工具时,自然停下作答。
tool_use 那条揭示了一个精巧设计:server 工具 vs client 工具。在 _handle_tool_use(:887-895),若工具 metadata.return_direct 为真,就不执行、直接返回 NOT_EXECUTED,挂进外部待办。而这个标志的来源是 ToolSpec.to_function_tool 里的一行(chat_config_models.py:266):
# 真实源码 chat_config_models.py:266,注释说逻辑是反的
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:291),模型就能像调普通函数一样自主检索、多次检索。工具描述里直接写「一次搜一个主题、多搜几次」(tool_placeholders.py:37)来引导拆解式检索。 - 边流边执行,不等整轮说完。
_close_active_tool_block在单个工具块一流完就create_task起执行(chat_loop_engine.py:691),把「模型继续生成」和「工具已开始跑」重叠起来,压低端到端延迟。 - 一个反转布尔值统一 server/client 工具。
return_direct = runtime != "server"(chat_config_models.py:266)让同一条循环既能就地执行内部工具、又能把外部工具交回调用方,分支逻辑收进_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:235-268),多次检索复用同一 workflow。 - schema 从函数签名反射。 改
semantic_search的签名即改工具契约(chat_config_models.py:200-202),无需手写 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_loop_engine.py:188),跑满即以max_tokens收尾——一个查询若诱发过多次检索,会被截断。 - 必须是支持函数调用的模型。
initialize_run检查isinstance(llm, FunctionCallingLLM),否则报「Configured model does not support function calling」(chat_loop_engine.py:755-756)。非 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:54-66 列出全部 processor)。差异只在各自的 processor/builder 内部——对循环而言,它们都只是「一个 runtime="server" 的可调用函数」。这正是「工具化」抽象的价值:新增一种检索能力,不动循环一行代码。
7. 代码地图(导航索引)
| 主题 | 文件路径 | 符号名 |
|---|---|---|
| 把检索服务闭包成工具函数 | private_gpt/components/tools/builders/semantic_search_builder.py:291 | semantic_search(内层 async fn) |
| 产出工具规格 | private_gpt/components/tools/builders/semantic_search_builder.py:213 | SemanticSearchToolBuilder.build_tool |
| 结果格式化成上下文 prompt | private_gpt/components/tools/builders/semantic_search_builder.py:270 | _sync_format_result |
| 占位符 → 真工具的替换 | private_gpt/components/tools/processors/semantic_search_processor.py:27 | SemanticSearchProcessor.intercept |
| processor 流水线 | private_gpt/components/tools/tool_pipeline.py:68 | ToolPipeline.contextualize_internal_tools |
| builder 的依赖注入装配 | private_gpt/components/tools/tool_factories.py:42 | SemanticSearchToolBuilderFactory |
| 内部检索类工具名单 | private_gpt/components/tools/tool_names.py:28 | GENERAL_INTERNAL_TOOLS |
| 工具描述文案 | private_gpt/components/tools/tool_placeholders.py:37 | SEMANTIC_SEARCH_TOOL_FN |
| server/client 工具的反转布尔 | private_gpt/components/chat/models/chat_config_models.py:251 | ToolSpec.to_function_tool(return_direct) |
| agent 主循环入口 | private_gpt/components/engines/chat_loop/chat_loop_engine.py:205 | ChatLoopEngine.run |
| 循环核心 / 迭代条件 | private_gpt/components/engines/chat_loop/chat_loop_engine.py:240 | _run_loop_core |
| 单轮迭代:请模型、起工具、回灌 | private_gpt/components/engines/chat_loop/chat_loop_engine.py:332 | _run_iteration |
| 流式解析 chunk 成事件 | private_gpt/components/engines/chat_loop/chat_loop_engine.py:485 | _handle_stream_chunk |
| 工具块流完 → 起执行 task | private_gpt/components/engines/chat_loop/chat_loop_engine.py:651 | _close_active_tool_block |
| 执行单个工具调用 | private_gpt/components/engines/chat_loop/chat_loop_engine.py:834 | _handle_tool_use |
| 并行/串行开关 | private_gpt/components/engines/chat_loop/chat_loop_engine.py:361 | tool_semaphore / allow_parallel_tool_calls |
| 真正 acall + 转 tool 消息 | private_gpt/components/engines/chat_loop/utils/tool_utils.py:43 | execute_tool_call |
| 工具选择过滤 | private_gpt/components/engines/chat_loop/utils/tool_utils.py:32 | select_tool_names |
| 调用方(不展开 HTTP/SSE) | private_gpt/server/chat/chat_service.py:177 | ChatService.stream_chat |