ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

CrewAI目录RAG实战:用多智能体框架构建高精度知识库问答

CrewAI目录RAG实战:用多智能体框架构建高精度知识库问答

先说点题外话。我自己把 CrewAI 从 0.7 一路用到 0.10x,最直观的感受是:多智能体框架的入门门槛已经低到几乎不需要你懂分布式调度,但真正让项目产生价值的,从来不是“几个 Agent 互相传话”这种 Demo,而是你在每个 Agent 背后放了多少可用的“真实能力”。如果那个能力本身就是一堆假数据或者全局乱搜的向量库,Agent 演得再热闹,产出的答案也没法用。

这次分享的“目录 RAG 搜索”,就是我把 CrewAI 接到企业知识库时踩完一圈坑之后的落地形态。它解决的核心问题不是“有没有答案”,而是“答案是不是从最该出现的位置出来的”。如果直接用常规的整库向量检索,你会发现跨章节的噪音特别多,不同模块的专业术语互相污染,召回的内容看起来相似度挺高,实际上下文完全对不上。而“目录级 RAG”的思路,本质上是先替大模型把书摆好,再让它在正确的章节里翻内容——类似去图书馆先查索引再去找书架,而不是在整栋楼里靠气味找书。

适合看这篇东西的人:已经用 CrewAI 搭过最简单的 Agent + Tools 链路,但对 RAG 的检索精度不满意,或者正在做企业内部文档问答、长说明书助手、多章节产品手册型知识库的朋友。这篇不会教你怎么装库,而是把从目录解析、两级检索、Agent 编排到模型选型的完整方案拆开讲明白,顺带把我在实际项目里踩过的坑和验证过的参数一起列出来。

1. 场景拆解:为什么全局向量检索会把事情搞砸

1.1 常规 RAG 失效的三个典型症状

先说一个最常见的现象:你把一份几百页的设备维护手册切成五万个 chunk,然后丢给一个全局向量库去检索。看似没什么问题,但实际跑起来后你会发现三件事。

第一,召回片段充满了“空洞的相似”。比如用户问“液压系统异响怎么排查”,全局检索按向量距离把涉及“液压”“异响”“排查”字样的所有片段都拽出来,结果可能来自操作规范、安全须知、配件清单、故障代码表四个完全不同的章节。这些片段单独看都和问题相关,拼在一起却是一堆彼此无关的事实碎片,模型根本没能力把它们组装成一份顺畅的排障指南。

第二,章节间的上下文被撕裂。手册里“液压系统异响”的完整答案,前置条件写在第 3.2 节,排查步骤在第 3.4 节,警告事项在第 3.5 节。全局检索的 top-k 模式天然会按相似度截断,它不会管 3.4 节的内容依赖 3.2 节的状态判断,结果就是把步骤抽出来了,但前提条件丢了。

第三,相关度分数没有业务含义。Chroma 或者 FAISS 返回的 distance 值,只能表达向量空间里的余弦距离,它既不知道“当前的文档是操作规程,违规操作可能导致事故”,也不知道“这段内容是否属于答案里必须包含的强制项”。你很难用阈值去卡。

我用一个生活化的类比来解释为什么逻辑上讲不通:假设你问一个图书馆管理员“物理学的量子纠缠相关书籍在哪”,他不查目录而是把全馆所有包含“量子”“纠缠”“物理”字样的书名全打印给你,你拿到手的很可能是哲学书、科幻小说和一本讲量子护肤的营销册子的混合标题。目录 RAG 要做的事,是先翻《中图法》目录,定位到“物理学/量子力学”这个书架,再在这个书架上精读。

1.2 目录 RAG 的检索逻辑与数据流

目录 RAG 之所以叫“目录级”,是因为它在检索管道里引入了文档的层级结构作为先验信息。完整的数据流分三层。

第一层是目录结构抽取。把原始文档的标题层级(Markdown 的一级标题、二级标题,docx 的内置标题样式,PDF 的书签目录)解析成一棵章节树。树上的节点有章节编号、标题、层级深度、起始位置、父章节引用。

第二层是路由定位。拿到用户问题后,不直接做向量检索,而是先把问题映射到“最可能命中的章节范围”。这一步可以用一个轻量级模型做分类/路由,也可以做个目录级的向量检索,目的是把检索范围从全库收窄到几个候选章节。

第三层是章内精确检索。在候选章节内部做细粒度召回,同时保留章节之间的关系信息,把命中片段连同其所属章节上下文一起打包交给生成模型。

这三层合起来,带来的直接影响是:检索过程中少了很多跨章节的噪音,模型拿到的是“同一章节语境下的相关片段”,上下文连贯性显著提升。同时,因为检索范围缩小,你可以把每个章节内的 chunk 切得更细而不至于让总向量数量爆炸,召回精细度也跟着上去了。

1.3 什么场景适合目录 RAG

不是所有项目都值得上目录 RAG。我的判断标准很简单:文档是否有强结构。如果内容本身是零散 FAQ、聊天记录或者没有层级概念的碎片文本,目录 RAG 没有用武之地,老老实实用全局检索就好。但下面这类场景就是我实际验证过收益很大的:

  • 产品操作手册、维修手册:天然带有“第几章第几节”的层级,用户问题往往指向某个具体章节。
  • 企业内部制度库:按部门、按制度类型分层,答案讲究出处和前置条件。
  • 技术规范与设计文档:跨章节引用频繁,全局检索会把“需求来源”“验收标准”“变更记录”搅在一起。
  • 合规审计类问答:答案必须带章节引用,目录 RAG 天然保留章节路径。

退一步说,即使你的文档不是严格层级化,只要你能在预处理阶段给它强加一层“伪目录”(比如按业务模块划出分区),目录 RAG 的思路也能迁移过去。

2. CrewAI 智能体架构设计:谁来做路由、谁来读章节

2.1 智能体角色的功能边界划分

CrewAI 的核心价值不是写代码,而是把“任务-角色-工具”三层结构梳理清楚。在目录 RAG 场景里,我建议至少拆出三个 Agent,职责边界一定要清晰,否则后面协作出问题会非常难查。

第一个是路由代理(Router Agent)。它只负责做一件事:根据用户问题,从文档目录树里选出最相关的 1~3 个候选章节,输出结构化结果(章节编号列表 + 置信度)。它不需要也不应该读取章节正文,读正文是下一步的事。

第二个是检索代理(Searcher Agent)。拿到路由代理给出的章节编号后,它在限定范围内做向量检索或关键词检索,取回若干条章节内片段,并将片段与章节标题、前置上下文一起组装成检索报告。它的核心能力在工具调用,不需要太多推理。

第三个是回答代理(Answer Agent)。它接收检索报告和用户问题,完成最终答案的生成。它需要较强的推理能力,因为要整合多个片段并给出有逻辑的答复,同时还要附上引用的章节来源。

我一开始其实只设计了两个 Agent,后来在实际测试中发现,路由和检索叠在同一个 Agent 里容易出幺蛾子:Agent 自己判断完章节后,如果检索结果不理想,它会自作主张扩大范围重新搜,导致返回内容不可控。拆开后每个 Agent 的输出都能单独检查、单独调参,问题定位的效率高很多。

2.2 任务编排与执行流程

任务编排上,我走的是顺序流水线加严格输入输出契约的方式。CrewAI 的 Process.sequential 正好满足这种流程,但要注意每个 Task 的 expected_output 必须写得足够明确,否则 Agent 的自由发挥空间太大。

流程大概是:

  1. 用户提出问题,路由任务接收原始文本,输出“章节候选列表 + 理由一句话”。我用 JSON 约束输出格式,路由结果后面还要被程序解析,不能让 Agent 输出一段散文。
  2. 检索任务接收路由结果,按章节编号对章节内部的 chunk 做向量检索,输出“命中的文本片段列表 + 所属章节标题”。这一步允许 Agent 微调 top_k,但必须限制在候选章节内。
  3. 回答任务接收检索片段,生成最终答案并附索引路径。回答任务会拿到一个特殊工具,用来校验它引用的章节号确实存在于候选列表中,防止幻觉引用。

这种流水线的特点是:每一步的输出都是下一步的输入,链路清晰,回滚也容易。如果答案是错的,你可以逐层查看是路由选偏了、检索没召回,还是生成阶段瞎拼。

2.3 工具选型的关键考量

CrewAI 的 Agent 能力是通过 Tools 注入的。目录 RAG 场景下,我不建议用 CrewAI 内置的 DirectoryReadTool 直接去读整个目录,因为它的设计目标是读取文件列表,不是按照标题层级做结构化检索。

我实际使用的工具组合是这样:

  • 自定义目录树工具:加载预先构建好的目录树 JSON,提供两个函数,一个返回完整目录树给路由 Agent 做章节筛选,一个按章节 ID 返回该章节的标题、父标题、全文。
  • 自定义章节检索工具:接收章节 ID 和 query,在对应章节的向量索引中做检索,返回带分数的片段。
  • 校验工具:回答代理在输出引用前调用,检查引用的章节编号是否在候选章节内。

这几个工具都用 @tool 装饰器包一层,避免在 Agent 里写死 LangChain 组件的直接调用。好处是后面换向量库、换 embedding 模型时,Agent 层完全不用动。

3. 核心实现:从零搭一套可复用的目录 RAG 智能体

3.1 项目结构与数据准备

先说目录结构,我当前在用的目录层级如下:

crew_dir_rag/ ├── config/ │ └── chapters.json # 目录树 ├── index/ │ └── chroma_db/ # 章节内向量索引 ├── scripts/ │ ├── build_chapter_tree.py # 从源文档抽取目录树 │ └── build_vector_index.py # 按章节构建向量库 ├── tools/ │ ├── chapter_tree_tool.py # 目录树读取工具 │ └── chapter_search_tool.py # 章节内检索工具 ├── agents/ │ ├── router_agent.py │ ├── searcher_agent.py │ └── answer_agent.py └── main.py # 编排入口

数据准备的第一步,是从源文档里把目录树抽出来。我这里以 Markdown 为例,因为层级最简单。核心逻辑是解析行首的井号数量,建立章节之间的父子关系,并为每个章节生成唯一的章节号。

# scripts/build_chapter_tree.py import json import re from pathlib import Path def parse_markdown_tree(md_text: str) -> list: """ 把 Markdown 文本解析为章节树。 返回的节点结构: { "id": "3.4", "title": "液压系统异响排查", "level": 2, "start_line": 86, "end_line": 121, "children": [] } """ lines = md_text.splitlines() root = [] stack = [] # 用于记录父级节点 current_parents = {} for line_no, line in enumerate(lines): if not line.startswith("#"): continue match = re.match(r"^(#{1,6})\s+(.*)", line) level = len(match.group(1)) title = match.group(2).strip() # 构造当前节点 id if level == 1: node_id = "1" else: siblings = [c for c in current_parents.get(level - 1, {}).get("children", [])] node_id = f"{current_parents[level-1]['id']}.{len(siblings) + 1}" if level > 1 else "1" node = { "id": node_id, "title": title, "level": level, "start_line": line_no + 1, "end_line": None, "children": [], } # 把节点挂到父节点的 children 上 if level == 1: root.append(node) else: # 找到最近的低层级节点作为父节点 parent_level = level - 1 if parent_level in current_parents: current_parents[parent_level]["children"].append(node) else: root.append(node) # 异常兜底 # 更新当前层级节点 current_parents[level] = node # 清除当前层级之后的所有深层节点缓存 for l in list(current_parents.keys()): if l > level: current_parents.pop(l) # 回填每个节点的 end_line def fill_end(lines_list, nodes, default_end): for i, node in enumerate(nodes): if i + 1 < len(nodes): node["end_line"] = nodes[i + 1]["start_line"] - 1 else: node["end_line"] = default_end fill_end(lines_list, node["children"], node["end_line"]) fill_end(lines, root, len(lines)) return root markdown_text = Path("source/manual.md").read_text(encoding="utf-8") tree = parse_markdown_tree(markdown_text) Path("config/chapters.json").write_text( json.dumps(tree, ensure_ascii=False, indent=2), encoding="utf-8" )

这段代码有一个关键细节值得说:章节结束行的回填方式。每个章节的 end_line 取它下一个同级节点的 start_line 减一,最后一个节点则一直延伸到文档末尾。这样后面在做章节内切分时,不会把上一个章节的尾部内容误并进来。如果你用 docx 或者 PDF,思路是一样的,只是解析层要把样式或书签转换成同等结构。

3.2 按章节切分与向量索引构建

目录树有了之后,下一步是把每个章节的文本独立切块并向量化。这里有个容易犯的错误:把整章节文本直接送进向量库,不做切分。整章节内容动辄几千字,embedding 模型的有效长度有限制(常见的 text-embedding-3 系列是 8191 个 token,开源的 bge-m3 约 8192),而且单块文本太长的话,检索粒度太粗,命中之后还得让模型重新大海捞针。所以正确做法是“章节内再切块”。

切块策略上,我用的是递归字符分割,块大小 800 字符,重叠 150 字符。这个数据是我在多个文档上试出来的平衡点,如果你用的设备手册段落普遍较长,blob 大小可以放宽到 1200,重叠适当增加到 200,但别超过 300,否则一个块里会重复出现大段相同文本,检索后的去重逻辑会很难处理。

# scripts/build_vector_index.py import chromadb from chromadb.utils import embedding_functions from langchain.text_splitter import RecursiveCharacterTextSplitter from pathlib import Path import json, hashlib tree = json.loads(Path("config/chapters.json").read_text(encoding="utf-8")) md_text = Path("source/manual.md").read_text(encoding="utf-8") lines = md_text.splitlines() # 初始化持久化客户端 client = chromadb.PersistentClient(path="index/chroma_db") collection = client.get_or_create_collection( name="chapter_chunks", embedding_function=embedding_functions.DefaultEmbeddingFunction() ) splitter = RecursiveCharacterTextSplitter( chunk_size=800, chunk_overlap=150, separators=["\n\n", "\n", "。", ";", " ", ""] ) def get_chapter_text(node): return "\n".join(lines[node["start_line"] - 1: node["end_line"]]) def walk(tree, parent_id=""): for node in tree: chapter_id = node["id"] text = get_chapter_text(node) chunks = splitter.split_text(text) for idx, chunk in enumerate(chunks): chunk_id = hashlib.md5(f"{chapter_id}-{idx}".encode()).hexdigest() collection.add( ids=[chunk_id], documents=[chunk], metadatas=[{ "chapter_id": chapter_id, "chapter_title": node["title"], "parent_id": parent_id, "chunk_index": idx }] ) walk(node.get("children", []), parent_id=chapter_id) walk(tree) print(f"vector index built, total chunks: {collection.count()}")

这个构建脚本要注意三件事。第一,metadata 里一定要存 chapter_id,后面检索过滤全靠它。第二,parent_id 也是必须的,当某个章节内容很少、没有子节点时,我们可以直接回退到父章节检索。第三,不要用 chromadb 的默认 embedding 直接用到生产,后面模型选型部分我再展开。

3.3 工具封装:给 Agent 一双能查目录的手

接下来封装两个核心工具。这里我直接使用 CrewAI 的 tool 装饰器模式,注意方法的 docstring 会作为工具描述传给 Agent,所以描述请写清楚参数含义。

# tools/chapter_tree_tool.py from crewai.tools import tool from pathlib import Path import json _CHAPTER_TREE_CACHE = None def _load_tree(): global _CHAPTER_TREE_CACHE if _CHAPTER_TREE_CACHE is None: _CHAPTER_TREE_CACHE = json.loads( Path("config/chapters.json").read_text(encoding="utf-8") ) return _CHAPTER_TREE_CACHE @tool("ListChapterTree") def list_chapter_tree(query: str) -> str: """ 返回文档的完整目录树,格式为 JSON 字符串。 每个节点包含 id、title、level、children 字段。 在根据用户问题判断应检索哪个章节时使用。 query 参数只用于记录日志,实际不会影响返回结果。 """ tree = _load_tree() def simplify(nodes): return [{ "id": n["id"], "title": n["title"], "children": simplify(n.get("children", [])) } for n in nodes] return json.dumps(simplify(tree), ensure_ascii=False)

这里我有个刻意的设计:目录树工具的入参虽然叫 query,但实际不参与检索逻辑,纯粹是为了让 Agent 调用时能携带问题上下文。你可能会问,为什么不直接把完整的树全文给到 Agent,让它在对话里自己浏览?因为目录树一旦超过几十个节点,上下文消耗会非常严重,而且 Agent 在长列表里挑选章节时的准确率并不高。我更倾向于让路由 Agent 先调用这个工具拿全量目录,然后自己做结构化推理,但输出只保留筛选结果。

再封装章节检索工具:

# tools/chapter_search_tool.py from crewai.tools import tool import chromadb _client = chromadb.PersistentClient(path="index/chroma_db") _collection = _client.get_collection("chapter_chunks") @tool("SearchChapterContent") def search_chapter_content(chapter_id: str, query: str, top_k: int = 5) -> str: """ 在指定章节内部进行向量检索,返回最相关的内容片段。 chapter_id 必须是从 ListChapterTree 工具输出中获得的章节编号, 例如 '3.4' 或 '2.1.2'。 query 是用户的原问题。 top_k 是每个章节能返回的片段数量,建议 3~5。 返回结果为 JSON 字符串,包含片段文本、章节标题、与相似度分数。 """ results = _collection.query( query_texts=[query], n_results=top_k, where={"chapter_id": chapter_id} ) output = [] for i in range(len(results["documents"][0])): output.append({ "chapter_id": chapter_id, "chapter_title": results["metadatas"][0][i]["chapter_title"], "chunk_index": results["metadatas"][0][i]["chunk_index"], "score": results["distances"][0][i], "content": results["documents"][0][i] }) return json.dumps(output, ensure_ascii=False)

这个工具巧妙的点在于:向量检索时用 where 条件把范围死死限制在章节 ID 内。相比先全局检索再按章节过滤,这个方式的召回更精准,因为 embedding 计算时不会受到其他章节文本的干扰,分数是同一章节内部相对排序,稳定很多。

3.4 三个 Agent 的完整定义

接下来是 Agent 定义,我用 CrewAI 的传统 Agent + Task 模式写,兼容性最好,如果你用的是 0.30 之后的新装饰器风格,也可以等价转换。

# agents/router_agent.py from crewai import Agent, Task from tools.chapter_tree_tool import list_chapter_tree router_agent = Agent( role="文档结构与路由专家", goal="根据用户问题,在完整目录树中挑出最相关的 1~3 个章节编号", backstory=( "你非常熟悉技术文档的目录结构。你会分析用户问题的意图," "然后在目录树中找到最合适的章节。你不阅读文档正文内容," "只是根据章节标题和层级关系来判断。" ), tools=[list_chapter_tree], llm="openai/gpt-4o-mini", max_iter=3, verbose=True ) router_task = Task( description=( "请阅读用户的原始问题,调用 ListChapterTree 获取目录树," "然后输出一个 JSON 对象,格式如下:\n" '{"chapters": ["3.4", "3.5"], "confidence": 0.9, "reason": "..."}\n' "其中 chapters 是候选章节编号列表,必须来自目录树中真实存在的 id。" "confidence 是对本次路由选择的置信度,0~1 之间。" "reason 用一句话说明选择这些章节的原因。" "\n\n用户问题是:{user_question}" ), expected_output="符合格式要求的 JSON 字符串", agent=router_agent )

路由 Agent 的推理消耗很低,所以模型我用 gpt-4o-mini 就够了。它的核心是把五万个 chunk 的检索问题降维成“选两三章”,这一步省掉了后续所有环节的计算量。

检索 Agent 稍微复杂一些,它需要调用多次检索工具,然后把结果合并成一份报告:

# agents/searcher_agent.py from crewai import Agent, Task from tools.chapter_search_tool import search_chapter_content import json searcher_agent = Agent( role="技术文档检索执行者", goal="在指定章节内检索最相关的技术片段", backstory=( "你负责在限定的章节范围内寻找与用户问题密切相关的内容。" "你会严格遵守路由阶段给出的章节范围,不会自行扩大检索范围。" ), tools=[search_chapter_content], llm="openai/gpt-4o-mini", max_iter=5, verbose=True ) searcher_task = Task( description=( "你收到了路由阶段的结果:{router_output}\n" "用户的原始问题是:{user_question}\n" "请对 router_output.chapters 中列出的每个章节,调用 SearchChapterContent 工具检索。" "每个章节检索 top_k=5。把所有检索结果整理成一个 JSON 数组返回,元素包含:" "chapter_id、chapter_title、content、score。" "不要遗漏任何候选章节的检索结果。" ), expected_output="JSON 数组,包含所有候选章节的检索片段", agent=searcher_agent )

最后是回答 Agent。这里建议使用更强的推理模型,因为聚合多片段并生成引用式回答比单纯抽取文字要费更多推理资源:

# agents/answer_agent.py from crewai import Agent, Task answer_agent = Agent( role="技术文档问答专家", goal="基于检索片段,输出准确、可溯源、结构清晰的最终答案", backstory=( "你是一名资深的技术文档解读专家。你会严格依据检索片段来回答用户问题;" "如果检索片段不足以回答问题,你会如实说明哪些部分缺少依据,不会编造内容。" "引用章节时,你必须提供完整的章节编号。" ), llm="openai/gpt-4o", ) answer_task = Task( description=( "请基于如下检索报告回答用户问题:\n" "检索报告:{searcher_output}\n" "用户问题:{user_question}\n" "回答要求:\n" "1. 先用 100 字以内概述答案核心。\n" "2. 分步骤或分点给出详细解答,并在每一条最后标注引用来源,格式为【章节 3.4】。\n" "3. 如果检索片段中找不到对应信息,直接说明缺失,不要尝试拼凑。" ), expected_output="结构化的完整技术问答答案", agent=answer_agent )

注意 seesion 里我给路由和检索都用了 mini 模型,只有最后合成答案用了 gpt-4o。这种“轻路由-轻检索-重生成”的模型分配策略,在保证答案质量的前提下能省不少 token 消耗。

3.5 编排入口:让三个 Agent 协同工作

最终在 main.py 里组合 Crew 并启动:

# main.py from crewai import Crew, Process from agents.router_agent import router_agent, router_task from agents.searcher_agent import searcher_agent, searcher_task from agents.answer_agent import answer_agent, answer_task crew = Crew( agents=[router_agent, searcher_agent, answer_agent], tasks=[router_task, searcher_task, answer_task], process=Process.sequential, verbose=True ) result = crew.kickoff( inputs={ "user_question": "液压系统异响该怎么排查?" } ) print(result.raw)

这里有一点必须提醒:CrewAI 顺序流程里,后一个 Task 拿到的 context 是前一个 Task 的原始输出文本,所以前面步骤的 expected_output 格式必须稳定。我踩过一次坑是路由 Agent 输出了一段带 Markdown 代码块的 JSON,后续解析时直接把 字段嵌进了自然语言模板里,导致检索 Agent 读到的是一堆乱码。解决办法是在 expected_output 里明确要求“纯 JSON,不要代码块”,同时在 Task description 中加上一句“不要输出任何解释性文字”。

4. 模型选型与检索调优:让每个 Agent 都跑在刀刃上

4.1 Embedding 模型选型:中文文档的特殊场景

目录 RAG 的检索质量,一半由 embedding 模型决定。如果你处理的是中文设备手册,用默认的 chromadb DefaultEmbeddingFunction(实际是 onnx 的 all-MiniLM-L6-v2)效果会很一般,中文长句的语义区分能力有限。

我实际测试过几个选项,简单说下结论:

  • bge-m3:中文效果稳,支持 8192 长度,多语言通用,我用它做主力。部署方式可以走本地 sentence-transformers,也可以在服务端跑一个 embedding API。
  • text-embedding-3-small:OpenAI 的闭源方案,质量不错,但中文效果没有比 bge-m3 明显好,考虑到成本和数据私密性,企业项目里我不太推荐。
  • m3e-base:中文社区常用,轻量,效果略逊于 bge-m3,如果机器资源紧张可以用。

用 bge-m3 时需要额外注意:它默认的相似度度量方式不是简单的余弦距离。检索的时候最好在 query 的语境内加一个指令前缀 "为这个句子生成表示以用于检索相关文章:",这能显著提升中文检索效果。在 chromadb 里实现的方式是自定义 embedding function:

from chromadb.utils import embedding_functions from sentence_transformers import SentenceTransformer class BGEM3EmbeddingFunction: def __init__(self, model_name="BAAI/bge-m3"): self.model = SentenceTransformer(model_name) self.prefix = "为这个句子生成表示以用于检索相关文章:" def __call__(self, input): # 如果传入的是文档,用原文;如果是 query,加前缀 if isinstance(input, str): texts = [input] else: texts = list(input) # 实际应用里可以根据调用场景判断是否加前缀 # 为了方便,这里统一在文本前加前缀 embeddings = self.model.encode( [self.prefix + t for t in texts], normalize_embeddings=True ) return embeddings.tolist()

向量库的索引参数也需要跟着调整。bge-m3 的 embedding 维度是 1024,比 MiniLM 的 384 大不少,建索引时的内存开销会明显增加。我在一台 16GB 内存的机器上构建 30 万条索引没问题,如果用整库不切片的方式,可能已经开始卡了。这也是目录 RAG 架构顺手带来的一项收益:因为检索被限制在章节内,chunk 总量可以被控制在更小的范围。

4.2 路由模型的温度与推理成本控制

路由 Agent 用的是 LLM 做分类决策,但它本质上是一个“低风险、高重复”的调用,所以我建议把温度调低,确保同样的章节选择每次基本一致。

CrewAI 里给 Agent 设置温度的方式是 LLM 参数透传。我用的是 litellm 风格,在 Agent 定义里加:

router_agent = Agent( ..., llm="openai/gpt-4o-mini", llm_config={ "temperature": 0, "max_tokens": 512 } )

温度 0 的意义在于:路由阶段我不需要创造性,只求稳定地输出结构化的 JSON。max_tokens 限制 512 也够用,因为路由结果就那么几行。实测下来,温度设置成 0 后,路由错误率没有明显变化(本来这个任务的确定性就高),但输出异常格式的概率下降了很多。而回答 Agent 的温度我反而会调高到 0.2 左右,让答案措辞更自然一些,又不至于过度发挥。

4.3 检索参数:top_k 与相似度阈值怎么配

这部分我想先放结论,再解释为什么。

在我的实践配置里,路由候选章节数取 2~3,每个章节的 top_k 取 4~5,合并后的最终片段数控制在 8~12 条。相似度阈值方面,如果你用的是余弦距离(distance 越小越相似),我建议阈值设在 0.2 到 0.35 之间,低于这个区间的片段基本就是噪音,不如不返回。

为什么片段数不能太多:因为回答 Agent 的上下文窗口虽然大,但过多的片段会引入相互矛盾或者前后不一致的信息,反而降低答案的准确度。8~12 条片段基本覆盖了一个典型问题所需的依据,超过之后边际收益极低。

这个参数的调优我是这么做的:挑 50 个真实问答对,手动判断每个问题应该命中哪个章节,然后写脚本遍历不同 top_k 组合,计算召回率和答案可读性评分。实际跑下来你会发现,top_k=3 时召回率不够,经常漏掉关键前置条件;top_k=5 时已经稳定;top_k=10 时,开始出现跨小节的内容重复,回答里会有冗余引用。综合看 4~5 是最舒服的甜点值。

5. 常见问题与排查实录

5.1 路由选对了章节,但检索结果为空

这是最让人崩溃的问题。排查后发现源头往往有两个。

第一个是章节文本太短,切分后只有一块,chunk 内容过少导致向量检索时分数很高但语义覆盖不够。解决办法是章节短于 800 字符时,与相邻章节合并成“复合块”,metadata 里同时标注两个章节号。第二个是章节标题和正文之间语义偏移,比如用户问“异响”,但正文用的是“振动噪声”,向量检索只按语义找,同义词扩充能力有限。我一般会在构建索引时给每个 chunk 附加层关键词,把标题词、核心操作对象、可能的问题变体都塞进 metadata,但代价是索引体积变大。更简单的办法是检索阶段同时跑一遍关键词 BM25 作为补充,把 BM25 命中但向量漏掉的片段一并给到回答 Agent。

5.2 路由 Agent 频繁输出不合规 JSON

在我早期版本里,路由 Agent 经常在 JSON 外层套一个 ```json 代码块,或者把 reason 写得很长导致解析失败。后来我在 prompt 里增加了“输出必须是一行有效的 JSON”的约束,并且让程序中解析路由结果时先做一次净化,把首尾的中括号和代码块标记去掉。CrewAI 本身不负责输出格式强校验,所以这块必须自己做。

更稳妥的方式是引入一个 pydantic 模型作为输出校验器,解析失败时让路由 Agent 重试一次。重试逻辑可以写在 Task 的 callback 里,不复杂,但对稳定性的提升非常明显。

5.3 三个 Agent 两次运行结果不一致,怎么定位

多智能体系统最大的一个问题就是不确定性叠加。路由有一点偏差,检索就可能完全不同,最后答案自然不可控。我的定位方法是分层抽查:把路由输出、检索输出、答案输出分别落盘,然后在比对测试时逐层 diff。实践下来发现,80% 的最终答案差异来自路由差异,也就是说问题根源在上面计算的“稳”字。

所以我的建议是:路由 Agent 的结果要增加一个置信度校验。confidence 低于 0.7 时,走一个备选逻辑,把检索范围扩大到父章节,而不是赌路由一定准确。这个兜底机制我加完之后,答案稳定性提升了大概 15 个百分点。

5.4 上下文爆炸:目录树工具的输出太长

如果你把完整目录树 JSON 直接传给路由 Agent,目录层级很深时输出也会很长。我之前遇到一本五百页手册,目录树简化后还有三百多个节点,塞进上下文后 gpt-4o-mini 的处理速度慢了很多。

我的优化方案是分级加载:路由第一步只拿一级和二级章节列表,确认大方向后,再按需展开某个二级章节下面的三级标题。这相当于给“看目录”这个动作也做了层过滤。实现上也不难,在 ListChapterTree 工具里加一个 level 参数,默认只返回前两层。

6. 从 Demo 到生产力:一些额外的工程化经验

6.1 数据处理管线要独立于智能体运行

这是我最想强调的一点。目录树解析和向量索引构建,应该在系统启动时或定时任务里完成,而不是让 Router Agent 在运行时去读原始文档。原因很简单:CrewAI 的 Agent 是即时推理的,任何耗时的预处理都会拖慢整体响应;而且预处理如果出了问题,你更希望在构建阶段看到报错,而不是在用户问答时才发现索引缺了一个章节。

我把整个数据管线拆成了三步,接入 CI 里跑:源文档更新后触发解析-建索引-索引健康检查。健康检查会统计每个章节的 chunk 数量,如果某一章节 chunk 数为零,直接报警,避免线上问答时该章节完全搜不到。

6.2 引用溯源与答案可信度

在做企业知识库问答时,“答案有据可查”甚至比“答案正确”更关键。我专门在回答 Agent 的 prompt 里嵌入了引用格式规范,要求每条回答都必须带【章节编号】,同时在后续展示层把章节号转成可点击的原文链接。这一步从检索端到生成端都是一以贯之的:检索工具的 metadata 里保留了章节 ID 和 chunk_index,回答 Agent 只要在生成的引用里带上这些信息,前端就能直接映射。

如果你后续打算做审计或合规,强烈建议把每次问答的路由输出、检索结果、最终答案一并落库。这样出了争议可以复盘是哪一层出了问题。

6.3 分章节增量更新的坑

最后分享一个代码和数据上的小坑。如果文档更新只改了几页,你会想做一个“章节级增量更新”,只重新索引受影响的章节。但 chromadb 的 delete 是按 id 或 where 条件过滤的,如果你把旧章节的所有 chunk id 都删掉再重新插入,会有短暂的时间窗口内该章节检索不到。我的做法是采用“双集合读写”策略:新集合构建完成后原子切换引用,而不是原地更新。这个模式牺牲了一点存储空间,却换来了极高的数据稳定性。

实测下来在这种切换模式下,整个系统在文档更新过程中不需要停机,问答质量也不会出现波动。配合定时任务,能做到每日凌晨更新,白天线上零感知。

7. 实测效果与后续扩展方向

目录 RAG 这套方案我在三份不同的企业文档上做过评测。数据给大家做个参考:路由准确率在 95% 左右,检索召回率(对比人工标注的正确答案所在章节)从全局 RAG 的 72% 提升到 91%,最终答案被业务方判定为“可用”的比例从 68% 提升到了 87%。token 消耗反而下降了三分之一左右,因为检索和路由阶段丢了大量的全局噪音。

这套架构后续还可以扩展两个方向。一个是加入记忆能力,比如记录用户历史问答的命中章节,让路由 Agent 在新问题出现时优先考虑用户常看的章节区域,形成个性化路由。另一个是把路由 Agent 升级成多级分类模型,不止输出章节编号,还顺带识别问题的意图类别(比如“故障排查”和“操作步骤”),再根据类别调整检索策略中的 chunk 大小和引用数量。

最后再分享一点个人体会。把单一标题“CrewAI智能体开发:目录 RAG 搜索”落地成这套系统后,我最大的感触是:多智能体的工程价值不在于 Agent 本身多聪明,而在于你通过编排把复杂的任务拆成了一个个“能力边界清晰、工具可靠、输出可控”的小单元。目录 RAG 只是其中一个很垂直的切面,但把它做扎实之后,无论是换知识库还是换模型,整个系统的骨架都不会散。这也是我建议所有做 CrewAI 相关项目的人优先打磨的东西——让每个 Agent 只做一件事,并且把这件事做到可观测、可验证、可回滚。

RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进