Agentic RAG：用 CrewAI 实现“先查文档、查不到再搜网页”的双工具检索智能体

导读

本文以仓库中的 agentic_rag 子项目为主体，讲解如何用 CrewAI 构建一个 Agentic RAG（智能体式检索增强生成）系统：检索 Agent 优先在用户上传的 PDF 文档内做向量检索，当文档中找不到答案时自动回退到网络搜索，再由响应合成 Agent 整合生成最终回答。读完本文，你将理解“工具回退（fallback）”式 Agentic RAG 的完整架构、DocumentSearchTool 的文档解析—分块—索引—检索实现细节，以及基于 Streamlit + 本地大模型（DeepSeek-R1 7B / Llama 3.2 via Ollama）的完整运行方式。

1. 什么是本项目中的 Agentic RAG

传统 RAG 是固定流水线：查询 → 向量库取 Top-K → 拼提示词 → 生成答案。而 agentic_rag/README.md 描述的方案把“检索”这一步交给了 LLM Agent 自主决策：

• 系统里有两类知识源：本地上传的 PDF 文档、公网搜索结果；
• 检索 Agent（retriever_agent） 同时挂载两个工具——DocumentSearchTool（查文档）和网页搜索工具（搜网络），其目标（goal）中明确写入了回退策略：“always try to use the pdf search tool first. If you are not able to retrieve the information from the pdf search tool then try to use the web search tool.”（见 config/agents.yaml）；
• 响应合成 Agent（response_synthesizer_agent） 只负责把检索结果组织成简洁连贯的回答；若确实检索不到，则按约定回复 “I'm sorry, I couldn't find the information you're looking for.”。

“Agentic” 的关键在于：选择哪个工具、何时判定文档里“查不到”需要回退，都是 LLM 在工具调用循环中动态决定的，而不是代码里写死的 if-else 路由。项目源码中还保留了一段被注释掉的 routing_agent（见 agents.yaml 前 7 行与 crew.py 中的注释块），它代表另一种“显式路由”设计——先由一个 Agent 判断该查 PDF 还是查网页。本项目最终采用的是更简洁的单检索 Agent + 双工具回退方案。

2. 环境安装与运行命令

根据 agentic_rag/README.md：

1. 准备 API Key：需要 FireCrawl API Key 用于网页搜索；
2. Python 版本：README 要求 Python 3.11 或更高；而 pyproject.toml 中 requires-python = ">=3.10,<=3.13"，两者取交集，即建议 Python 3.11–3.13；
3. 安装依赖：

pip install crewai crewai-tools chonkie[semantic] markitdown qdrant-client fastembed

这条命令正好覆盖了 DocumentSearchTool 的全部依赖链：markitdown 负责 PDF 转文本，chonkie[semantic] + fastembed 负责语义分块与本地嵌入，qdrant-client 负责向量存储。crewai 则要求版本 >=0.86.0,<1.0.0（见 pyproject.toml）。

启动命令（README 原文）：

• 使用本地 DeepSeek-R1：streamlit run app_deep_seek.py
• 使用本地 Llama 3.2：streamlit run app_llama3.2.py

两个入口都通过 Ollama 访问本地模型，因此需要先在本地 Ollama（默认地址 http://localhost:11434）中拉取对应模型。

3. 核心工具：DocumentSearchTool 的四步实现

tools/custom_tool.py 是本项目 RAG 能力的核心，继承自 CrewAI 的 BaseTool，整个生命周期分为四步：

3.1 文本抽取：MarkItDown

def _extract_text(self) -> str:
    """Extract raw text from PDF using MarkItDown."""
    md = MarkItDown()
    result = md.convert(self.file_path)
    return result.text_content

使用微软开源的 MarkItDown 将 PDF 直接转换为文本（Markdown 中间格式），避免手写 PDF 解析逻辑。

3.2 语义分块：Chonkie SemanticChunker

def _create_chunks(self, raw_text: str) -> list:
    chunker = SemanticChunker(
        embedding_model="minishlab/potion-base-8M",
        threshold=0.5,
        chunk_size=512,
        min_sentences=1
    )
    return chunker.chunk(raw_text)

参数含义：

参数	取值	作用
embedding_model	minishlab/potion-base-8M	分块边界判定用的本地嵌入模型（8M 参数轻量级，通过 fastembed 本地运行，无需 GPU/API）
threshold	0.5	相邻句子嵌入余弦相似度低于该阈值时认为语义发生“跳跃”，在此处切块
chunk_size	512	单块最大长度（token 级），超过则强制截断
min_sentences	1	每块至少包含 1 个句子

语义分块相比固定长度分块的优势是：块边界尽量落在语义转折处，使每个 chunk 主题内聚，检索命中后上下文更完整。

3.3 内存索引：Qdrant 本地集合

self.client = QdrantClient(":memory:")  # For small experiments
...
self.client.add(
    collection_name="demo_collection",
    documents=docs,
    metadata=metadata,      # [{"source": 文件名}, ...]
    ids=ids
)

源码注释写明 :memory: 是“For small experiments”——向量库随进程存活，适合演示与小规模实验；生产环境应改为持久化 Qdrant 实例。每个 chunk 都附带 source 元数据（文件名），便于回答时标注出处。

3.4 检索：以 ___ 分隔返回

def _run(self, query: str) -> list:
    relevant_chunks = self.client.query(
        collection_name="demo_collection",
        query_text=query
    )
    docs = [chunk.document for chunk in relevant_chunks]
    separator = "\n___\n"
    return separator.join(docs)

工具把命中的多个 chunk 用 \n___\n 拼接成单段文本返回给 LLM——分隔符让检索 Agent 能感知“这是多段独立来源”。

该工具对 LLM 暴露的接口由 Pydantic schema 定义：输入只有一个字段 query: str，工具描述为 “Search the document for the given query.”（custom_tool.py）。

一个需要注意的实现细节：README 宣称网页搜索由 FireCrawl 完成，Streamlit 入口 app_deep_seek.py / app_llama3.2.py 也确实导入了 FireCrawlWebSearchTool；而本目录的 custom_tool.py 只实现了 DocumentSearchTool。从源码结构看，FireCrawlWebSearchTool 需要在运行环境中自行补充（同样继承 BaseTool，内部调用 FireCrawl API），这也是 README 强调“先拿 FireCrawl API Key”的原因。另外，DocumentSearchTool 的构造参数在 custom_tool.py 中定义为 file_path，而 crew.py 中以 pdf= 关键字实例化并使用了作者机器上的绝对路径——复制该项目运行前，需要把 PDF 路径改为你自己的本地文档（本仓库提供了示例文档 knowledge/dspy.pdf）。

4. Crew 定义：Agent 与 Task 如何装配

4.1 Crew 类结构（配置驱动版）

crew.py 使用 CrewAI 的 @CrewBase 项目模板：

@CrewBase
class AgenticRag():
    """AgenticRag crew"""
    agents_config = 'config/agents.yaml'
    tasks_config = 'config/tasks.yaml'

    @agent
    def retriever_agent(self) -> Agent:
        return Agent(
            config=self.agents_config['retriever_agent'],
            verbose=True,
            tools=[pdf_tool, web_search_tool]   # 双工具只挂给检索 Agent
        )

    @agent
    def response_synthesizer_agent(self) -> Agent:
        return Agent(config=self.agents_config['response_synthesizer_agent'], verbose=True)
    ...
    @crew
    def crew(self) -> Crew:
        return Crew(
            agents=self.agents,
            tasks=self.tasks,
            process=Process.sequential,
            verbose=True,
        )

要点：

• 工具只挂载在 retriever_agent 上，合成 Agent 保持“纯净”，职责单一；
• Process.sequential 表示任务按声明顺序串行执行（检索 → 合成）；源码注释同时给出了切换为 Process.hierarchical 的选项。

4.2 提示词配置：agents.yaml 与 tasks.yaml

Agent 的角色设定全部放在 config/agents.yaml，并用 {query} 占位符在运行时插值（CrewAI 的 kickoff inputs 会替换它）：

retriever_agent:
  role: >
    Retrieve relevant information to answer the user query: {query}
  goal: >
    Retrieve the most relevant information from the available sources for the user query: {query},
    always try to use the pdf search tool first. If you are not able to retrieve the information
    from the pdf search tool then try to use the web search tool.
  backstory: >
    You're a meticulous analyst with a keen eye for detail. ...

response_synthesizer_agent:
  role: >
    Response synthesizer agent for the user query: {query}
  goal: >
    Synthesize the retrieved information into a concise and coherent response based on the user query: {query}.
    If you are not ble to retrieve the information then respond with
    "I'm sorry, I couldn't find the information you're looking for."

任务定义在 config/tasks.yaml：

retrieval_task:
  description: >
    Retrieve the most relevant information from the available sources for the user query: {query}
  expected_output: >
    The most relevant information in form of text as retrieved from the sources.
  agent: retriever_agent

response_task:
  description: >
    Synthesize the final response for the user query: {query}
  expected_output: >
    A concise and coherent response based on the retrieved infromation from the right source ...
  agent: response_synthesizer_agent

可以看到“回退策略”实际写在 Agent 的 goal 里，而“查不到时统一回复固定话术”写在合成 Agent 的 goal 和 response_task.expected_output 中——两者互为兜底，约束了模型的输出行为。

4.3 命令行运行方式

main.py 提供了 run / train / replay / test 四个入口，并在 pyproject.toml 中注册为控制台脚本：

[project.scripts]
agentic_rag = "agentic_rag.main:run"
run_crew  = "agentic_rag.main:run"
train     = "agentic_rag.main:train"
replay    = "agentic_rag.main:replay"
test      = "agentic_rag.main:test"

默认 run 使用测试查询 'Who is elon musk?'（一个文档里大概率查不到、必须触发网络回退的问题）；train 可对 Crew 做迭代优化，replay 可针对某个 task_id 回放调试——这是 CrewAI 项目模板自带的调试能力。

5. Streamlit 应用：本地模型 + PDF 上传聊天界面

app_deep_seek.py 与 app_llama3.2.py 结构几乎一致，唯一实质差异是模型配置：

@st.cache_resource
def load_llm():
    llm = LLM(
        model="ollama/deepseek-r1:7b",   # app_llama3.2.py 中为 "ollama/llama3.2"
        base_url="http://localhost:11434"
    )
    return llm

@st.cache_resource 保证 LLM 客户端在整个 Streamlit 会话中只初始化一次。界面与执行流程：

1. 侧边栏上传 PDF：文件写入临时目录后，st.session_state.pdf_tool = DocumentSearchTool(file_path=temp_file_path) 完成“上传即索引”，并在侧边栏用 base64 iframe 内嵌预览原 PDF；
2. Crew 懒构建：首次提问时才调用 create_agents_and_tasks(pdf_tool) 构造 Crew 并缓存到 session_state；工具列表用 [t for t in [pdf_tool, web_search_tool] if t] 过滤，即未上传 PDF 时退化为纯网页搜索模式；
3. 发起检索：result = st.session_state.crew.kickoff(inputs={"query": prompt}).raw——{query} 占位符在此被用户问题替换，Crew 串行执行检索、合成两个任务；
4. 伪流式渲染：由于本地模型不支持流式输出，代码先把完整结果取出，再按行、以 0.15 秒间隔逐行追加渲染（message_placeholder.markdown(full_response + "▌")），模拟打字机效果；
5. 会话管理：聊天记录存于 st.session_state.messages，“Clear Chat” 按钮清空记录并 gc.collect() 释放内存（内存版 Qdrant 与工具实例随之可回收）。

整个链路中没有任何远程 LLM API：向量嵌入用本地 8M 小模型，对话模型走本地 Ollama，唯一的云端依赖是 FireCrawl 网页搜索——这正是“本地私有文档 + 受控外部搜索”的典型 Agentic RAG 部署形态。

6. 小结与适用边界

本项目的 Agentic RAG 方案可以概括为三层：

1. 工具层：DocumentSearchTool（MarkItDown 抽取 → Chonkie 语义分块 → Qdrant 内存索引 → 相似度检索）+ FireCrawl 网页搜索工具，双工具构成“文档优先、网络兜底”的检索策略；
2. 智能体层：retriever_agent（双工具、goal 中编码回退规则）与 response_synthesizer_agent（受约束的合成话术）按 Process.sequential 串行执行；
3. 应用层：Streamlit 聊天界面 + Ollama 本地模型（DeepSeek-R1 7B / Llama 3.2），PDF 上传即索引。

从源码结构看，当前实现面向演示与小规模实验：Qdrant 使用 :memory: 模式、Crew 与工具实例随 Streamlit 会话缓存、PDF 路径需要按本地环境修改。若要走向生产，需要替换为持久化向量库、补齐 FireCrawlWebSearchTool 的具体实现（README 与入口文件的导入均可确认其定位，但 custom_tool.py 中未包含该类源码）、并按需为不同文档建立独立集合。作为教程项目，它完整展示了“把检索决策权交给 Agent、用工具组合实现 RAG 回退”这一核心模式的落地方式。