★ FEATURED ARTICLE 如果你搭过 RAG 知识库大概率见过这种场面用户问了一个很正经的问题系统召回一大堆不相干的片段最后大模型一本正经地拼出一个看似合理、实则跑偏的答案。我第一次做 RAG 时也以为是向量检索精度不够花了好几个晚上换 embedding 模型、调 top_k、加重排序效果始终上不去。直到把整条链路拆开一条条看才反应过来真正的问题在更靠前的位置——链路里少了一个“判断”的环节。这个项目的核心思路就是给 RAG 加两个东西一个叫分诊台Triage一个叫拆题术Query Decomposition。用大白话说先判断该走哪条路再规划怎么拆着走别一上来就闷头检索。这也是 RAG 实战里最常被忽略的一环。本文适合两类人一类是搭过 RAG 知识库但总被效果卡住的人另一类是准备从简单 RAG 向 RAG 智能体方向过渡的开发者。我会把我自己实现时踩过的坑、用过的提示词和代码骨架都写出来你照着改就能用。1. 先搞清楚RAG 的瓶颈在哪为什么要加一道“分诊台”1.1 让我印象深刻的 RAG 翻车现场先讲一个我实际遇到的案例。当时给一个企业做客服问答知识库有一万多份文档涵盖产品说明、常见故障、物流售后。系统上线后第一个星期用户问“退货流程需要多久”系统从某个产品说明里召回了一段关于“纯棉材质”的描述然后一本正经地回答“退货流程具体时长取决于材质成分”。这个答案放在测试集里会被直接判死刑但它确实发生了。问题出在哪一眼看过去是召回错了本质上是整个链路没有判断力。传统 RAG 的流程很机械用户 query 做向量化去向量库里找 top_k 个相似片段把片段拼进 prompt再让大模型生成答案。这里每一步都“看似合理”但注意它把所有问题都当成同一种问题处理。问“退货流程多久”和“这款路由器的 lan 口速率是多少”在链路里走的路一模一样都是同一个 embedding 模型同一个向量库同一个 top_k 参数。这就必然出现两个结果。第一高频简单问题也被迫走一遍完整检索链路延迟高、成本没省下来第二长 query 或复杂 query 的语义太密很容易跟知识库里的多个片段都“沾边但不精准”于是召回一堆噪音。噪音进入上下文之后生成模型只能在里面挑顺眼的答案自然开始飘。这就是我理解的 RAG 瓶颈不是检索算法不够先进而是链路缺少一个入口层的把关机制。1.2 链路里缺的不是检索精度是“判断”我后来反思了很久发现 RAG 这个缩写里其实藏着一个隐含假设——所有问题都该被检索增强。但现实世界的问题并非如此。有些问题跟知识库完全无关比如“早上好”和“今天几号”有些问题属于某个单一知识库比如“你们家路由器支持 WiFi 6 吗”明显对应产品文档有些问题需要跨两个知识库综合回答还有些问题根本不需要检索现成文档比如“帮我算一下 15 个订单的平均金额”或者“查一下当前天气”。如果链路里没有分诊层那么所有问题都会被一股脑塞进同一个管道。后果就是该直接回答的走了检索白白增加延迟该检索一个库的跑到所有库里捞了一遍捞回来一堆置信度差不多的片段该跨库的又因为每个库都只取 top_k把关键信息挤出了窗口。这些问题不是靠优化 embedding 能解决的因为它们的根子在“路由策略”和“查询粒度”上。这里我习惯用一个类比医院分诊台。医院不会让所有病人都直接去做 CT而是先由分诊护士问几句判断去内科、外科还是直接回家休息。RAG 的分诊台做的是同一件事它先看用户问题属于哪一类决定后续路由。判断做对了后面每一步才有意义判断错了再好的检索模型也只是把一个错误的方向执行得更充分而已。1.3 分诊台分级从“检不检”到“拆不拆”在我实现的版本里分诊台没有做成“一个问题对应一个动作”的粗暴分类而是做成了多级判断每一级都在解决上一级处理不了的情况。第一级最基础判断“要不要检索”。如果问题不需要外部知识比如闲聊、常识、与业务无关的日常问题就直接让大模型回答不查任何知识库。第二级判断“去哪个知识库查”。前提是问题确实需要检索而且指向明确比如“产品文档”或“故障手册”。第三级处理的是跨库问题比如“产品保修卡丢了怎么补办”既涉及产品政策又涉及售后流程这时需要路由到多个知识库分别取片段。第四级是今天说的拆题判断“这个问题是否需要拆开处理”。典型代表是多跳问题比如“A 公司的创始人的毕业院校是哪所”你得先查出创始人是谁再去查这个人的教育背景。分级不需要一步到位。你完全可以先只做“要不要检索”和“去哪个库检索”这两层等系统稳定了再加“拆题”。把分级做成分层的另一个好处是便于排查——问题出在哪一级直接看哪一级的日志就行。我项目里最后用的是一份 JSON 结构action 表示动作kb 数组表示要去的知识库need_decompose 表示是否拆题reason 字段记录判断理由方便回溯。2. 分诊台与拆题术的整体设计先判断再规划2.1 分诊层到底在分什么分诊层看起来是给问题分个类但真正设计时你会发现它其实是在替整个 RAG 系统做“资源调度”。传统链路里一次问答的成本基本固定向量检索一次生成一次。加入分诊层之后成本模型变成了动态的简单问题走轻量路径复杂问题走重型路径。从效率角度看这是一笔很划算的账。实现分诊层有三种主流方案。第一种是纯规则用关键词和正则做意图匹配优点是快、稳、可解释缺点是覆盖不了开放域问题企业知识库里的业务问法千奇百怪规则很容易漏。第二种是小模型分类器比如用 BERT、fasttext 微调一个文本分类模型效果不错但冷启动需要标注数据业务一变模型就得重训。第三种是让大模型充当 Router直接输入用户问题和知识库描述要求输出一个结构化 JSON。我最终选的是第三种原因很实际上线快零样本就能跑而且在新领域里可以先用大模型路由收集一批数据以后真要换小模型这些数据正好当成训练集。大模型当 Router 的关键不是提示词多复杂而是信息要给够。我踩过一次坑最初我只把用户问题发给大模型让它判断“是否检索”结果它经常瞎猜因为根本不知道知识库里有什么。后来我在提示词里加了一个知识库清单每个库配两到三句话描述例如“product_docs产品功能说明、版本更新、使用教程troubleshooting故障排查手册、报错原因、解决办法”。它才知道自己手里有哪些“科室”判断准确率立刻上了一个台阶。2.2 拆题术把多跳问题拆成单跳检索拆题术要解决的是 RAG 在复杂问题面前的失效问题。你可能试过拿一个很长的问题直接去检索结果召回片段东一段西一段谁都没答到点子上。原因不复杂embedding 相似度检索擅长的是“一段文本和另一段文本的语义接近”但多跳问题的信息太密了query 里同时出现 A 公司、创始人、毕业院校三个实体向量空间里它跟任何一段只包含部分内容的文档都只有部分相似检索分数被摊薄真正的关键片段反而排不到前面。拆题的本质就是把一个需要多步推理的问题拆成若干个可以独立检索的单跳问题。比如上面的问题可以拆成三个子问题A 公司是谁创立的这个创始人是谁他的毕业院校是哪所。前两个是中间步骤最后一个才是终点。每个子问题都是单跳检索候选片段只需要回答“一个事实点”命中率自然高得多。这里有个设计要点子问题之间可能有依赖关系。有些可以并行拆比如“A 公司的主要业务和总部城市”两个子问题互不依赖可以同时检索有些必须顺序拆比如必须先知道创始人的名字才能查他的毕业院校。我在实现时没有一上来就做复杂的 DAG 调度而是做了一个简单约定让拆题提示词尽量拆成“可并行检索”的原子子问题如果确实存在依赖就在子问题里显式带上中间信息比如拆题结果直接写成“A 公司的创始人是张三张三的毕业院校在哪”这样虽然第二个子问题依赖第一个的结果但拆题时已经替它算好了前置变量。2.3 一套可落地的 Pipeline 编排分诊和拆题不是独立组件而是要编排进原有的 RAG 流程里。我最终跑的流程是这样的用户问题进来之后先过 LLM Router输出结构化结果。如果 action 是 direct_answer直接把用户问题和“你是一个客服助手”之类的系统提示交给生成模型不再碰向量库。如果 action 是 retrieve就根据 kb 数组确定去哪些库检索。如果 need_decompose 为 true先走拆题把子问题列表拿去并行检索每个子问题各自取少量片段再生成一个子答案。最后把所有子答案或者普通检索的片段交给最终生成环节输出答案。这套编排我们可以用一个简表来看用户问题类型分诊结果处理链路闲聊/问候direct_answer直接生成不检索单一知识库的简单事实retrieve单个kb检索该库 → 生成跨知识库的综合问题retrieve多个kb多库分别检索 → 合并 → 生成多跳/对比类复杂问题retrieveneed_decomposetrue拆题 → 并行子检索 → 子答案 → 汇总生成实时数据/计算任务tool_call调用外部 API/计算器 → 生成我这里没有把 tool_call 展开太多因为项目重心在于分诊和拆题但分诊层顺手把它预留了。经验是分诊层把“需要工具”的动作先暴露出来后续接入 Agent 工具调用会平滑很多。这也符合你从 RAG 框架往 RAG 智能体过渡的路径——先有判断再有规划最后才有工具。3. 逐步实现让 RAG 既会分诊又会拆题3.1 分诊提示词的写法与结构化输出我先把分诊提示词写出来你直接复制就能改。这里的关键是输出必须是结构化 JSON而且要有 reason 字段否则出问题你连它为什么这么判断都不知道。你是知识问答系统的分诊员。系统可用的知识库如下 - product_docs产品功能说明、版本更新、使用教程覆盖型号参数、功能操作。 - troubleshooting故障排查手册、报错原因、解决办法覆盖异常现象和维修指引。 - order_after_sales订单状态、退换货、物流查询覆盖购买后的所有流程。 请判断用户问题应该走哪条处理路径并输出JSON {action: direct_answer | retrieve | tool_call, kb: [product_docs], need_decompose: true/false, reason: 判断理由} 约束 1. direct_answer 表示无需检索知识库直接回答即可。仅用于与业务无关的闲聊、问候、常识问题。 2. retrieve 表示需要检索知识库kb 数组给出要检索的库名可以多个。 3. tool_call 表示需要调用外部工具比如实时信息、计算器不进入知识库检索。 4. need_decompose 为 true 时表示该问题需要拆分成多个子问题才能回答例如多跳问题、跨实体关联问题。配合这段提示词我建议 temperature 设为 0让输出尽量稳定。解析时别直接 json.loads因为本地部署的模型经常在 JSON 前后夹带解释性文字。我的做法是用正则把 response 里第一个“{”到最后一个“}”之间的内容截出来再交给 json.loads。解析失败时不要静默兜底要打一条 warn 日志方便后面发现问题。下面是一段解析代码很短但我在生产里实际用了很久import json import re def parse_triage_result(text: str) - dict: match re.search(r\{.*\}, text, re.DOTALL) if not match: return {action: retrieve, kb: [], need_decompose: False, reason: parse_failed} try: data json.loads(match.group(0)) return data except json.JSONDecodeError: return {action: retrieve, kb: [], need_decompose: False, reason: parse_failed}解析失败时我故意让它默认走 retrieve 而不是 direct_answer这是我从教训里学到的漏检比多检的危害大得多。宁愿多检索一次浪费一点资源也不要因为解析失败把一个需要知识库的问题直接跳过检索。3.2 拆题提示词与子问题检索细节拆题提示词也要结构化但约束更细。我会在提示词里要求模型在指定的业务领域里拆解并把子问题数量限制在四五个以内。我常用的拆题提示词长这样你是问题拆解助手。请把用户问题拆分成可以独立检索的子问题每个子问题必须简短、单跳、可单独检索。 要求 1. 子问题数量不超过4个能少则少。 2. 每个子问题只能询问一个事实点禁止出现“既要又要”的复合问法。 3. 子问题不要重复含义接近的只保留一个。 4. 输出JSON{sub_questions: [子问题1, 子问题2]}拆题结果的质量直接决定检索效果。我测下来一个需要控制的地方是子问题的“粒度”。如果拆得太粗比如“介绍一下这家公司”它还是包含公司背景、业务、地址一堆信息检索时精度依然上不去如果拆得太细比如把“创始人是谁”再拆成“这个创始人叫什么”和“这个创始人是做什么的”就纯属浪费检索名额。我的经验是靠提示词约束加强力去重而不是靠模型自觉。子问题检索的代码不难但有一个参数要特别注意每个子问题的 top_k 不要设太大。整段 query 检索可能需要 top_k10 才能覆盖信息点但子问题已经是原子问题top_k 设 3 到 5 就够了。因为拆题之后目标更聚焦少量精准片段比一堆相关片段更有用。多余的片段塞进上下文只会增加噪声。如果有多个子问题强烈建议并行检索。这里有个隐藏坑如果你用的是同一个向量库客户端要确认它是否线程安全。我之前图省事直接用线程池同时查同一个 chroma 客户端结果出现过连接冲突。后来改成每个线程单独初始化一个只读客户端问题就消失了。并行之后拆题链路的总延迟基本等于最慢那个子问题的检索耗时体感好了很多。3.3 多知识库路由与兜底策略多知识库路由是分诊层里最容易被低估的环节。很多人觉得只要把数据库描述写清楚大模型自然能选对。实际上在业务库里不同知识库的主题常常重叠比如“product_docs”和“troubleshooting”都可能会提到“开机失败”这时候靠描述选库就容易翻车。我的方案是双保险大模型分诊为主向量路由做校验。也就是在分诊输出的 kb 数组之外再用 query 的 embedding 分别跟每个知识库的代表性标题或摘要向量算相似度看哪两个库最接近。如果大模型选了库 A但向量路由强烈指向库 B我会按“同时检索两个库”处理而不是让它们二选一。多召回一两个片段后面反正有重排序兜底。兜底策略里另一个关键是路由到多库时的 top_k 分配。假设上下文窗口允许 8 个片段路由到两个库时我不是每个库取 8 个而是每个库取 4 个然后合并在一起过一个 reranker保留与问题最相关的 5 到 6 个。这个做法的好处是避免单个库霸占上下文让跨库信息都有机会参与生成。3.4 完整串接dispatch_and_plan 骨架代码把上面这些串起来你最终得到的核心函数大概长这样。我把它压成一个简单的骨架去掉了具体向量库细节你看了就能理解执行顺序import json from concurrent.futures import ThreadPoolExecutor def load_triage_prompt(): # 包含知识库描述和分诊JSON要求这里省略实际文本 pass def load_decompose_prompt(): # 包含子问题拆解的JSON要求这里省略实际文本 pass def call_llm(prompt: str, temperature: float 0.0) - str: # 调用本地模型或API返回原始文本 pass def triage(query: str) - dict: prompt load_triage_prompt() f\n用户问题{query}\n请输出JSON text call_llm(prompt, temperature0.0) result parse_triage_result(text) # 兜底解析失败时默认检索 if result[action] direct_answer: return result return result def decompose(query: str) - list[str]: prompt load_decompose_prompt() f\n用户问题{query}\n请输出JSON text call_llm(prompt, temperature0.0) match re.search(r\{.*\}, text, re.DOTALL) if not match: return [query] data json.loads(match.group(0)) return data.get(sub_questions, [query])[:4] def search_single(sub_query: str, kb_names: list[str], top_k: int 3): # 在指定知识库中检索返回片段列表 pass def parallel_search(sub_questions: list[str], kb_names: list[str]): with ThreadPoolExecutor(max_workerslen(sub_questions)) as executor: futures [ executor.submit(search_single, q, kb_names, 3) for q in sub_questions ] return [f.result() for f in futures] def generate_answer(context, query: str) - str: # 将上下文与query交给最终生成模型 pass def dispatch_and_plan(query: str) - str: triage_result triage(query) if triage_result[action] direct_answer: return generate_answer([], query) kb_names triage_result.get(kb, []) if not kb_names: kb_names [default] if triage_result.get(need_decompose): sub_questions decompose(query) fragments parallel_search(sub_questions, kb_names) # 先为每个子问题生成简短子答案再汇总 sub_answers [ generate_answer(one_group, sub_questions[i]) for i, one_group in enumerate(fragments) ] return generate_answer(sub_answers, query) fragments search_single(query, kb_names, top_k8) return generate_answer(fragments, query)这里面有一个值得注意的细节拆题分支里我没有直接把所有子问题的检索片段合并成一个 context 丢给最终模型而是先为每个子问题生成一个“子答案”再把子答案列表交给最终模型。直接拼片段的问题在于上下文太长之后模型容易丢失关键证据而且不同片段可能互相矛盾。子答案相当于做了一次中间摘要把每个子问题的结论提取出来最终生成阶段只需要综合这些结论负担小很多。有人会问拆题之后先生成子答案会不会丢失信息我的经验是丢失的通常是噪音而不是关键信息。关键证据一旦被子问题定向检索到几乎都会出现在子答案里。相比一次性拼接原始片段这种“检索-子答-汇总”的流程更可控也更容易在每个阶段加日志查问题。4. 问题排查与效果评估实录4.1 常见问题速查表项目跑起来之后至少有半年我都在跟各种奇怪的故障搏斗。很多问题表面上看是“模型不够聪明”实际上都是链路设计和解析兜底的问题。我整理了一个速查表你可以直接当排查手册用现象可能原因排查与解决简单问题也被送去检索分诊提示词缺少 direct_answer 的示例在提示词末尾加两个“直接回答”的 few-shot 例子解析 JSON 频繁失败本地小模型输出格式不稳定用正则截取 JSON 片段解析失败时默认走检索并打日志拆出来的子问题太多或太相似拆题提示词没有限制数量和粒度强制“最多4个”并加去重指令子答案汇总后答非所问最终生成阶段没带上原始 query把原始用户问题放回汇总生成 prompt路由经常选错知识库kb 描述太短或太抽象为每个库写两三句话说明包含什么、不包含什么跨库检索后上下文超长每个库独立 top_k 叠加过多多库路由时每库减小 top_k合并后统一 rerank链路延迟突然变高子问题串行检索用线程池并行检索并确认向量库客户端线程安全这些问题的共性都是“链路设计不清晰”而不是“模型能力不足”。分诊和拆题两个环节给你加上了判断但判断本身也需要调试。我的习惯是给每条调用都打 structured log把分诊 result、拆题子问题列表、每路子问题检索的片段 ID 都记录下来。出了 case 再回看日志定位到具体环节解决问题就只是改提示词或调参数的事。4.2 三个印象深刻的踩坑教训第一个教训是分诊误判“直接回答”导致漏检。上线初期我把答疑库里的高频问题都当成“常识问题”处理让分诊层直接回答结果发现模型经常把业务规则答错。后来我改了一条策略只有分诊结果里 reason 明确显示“与知识库无关”时才允许走 direct_answer一旦模型对是否检索犹豫不决就默认转 retrieve。别怕多检索检索一次的成本远低于答错一次。第二个教训是拆题后的子答案互相矛盾。场景是这样的用户问“两个型号的保修期分别是多久”子问题分别检索了两个型号的说明文档但文档版本不同一份说过保一年一份说过保两年。最终汇总时模型选择了“各打五十大板”的输出说“保修期可能是一年或两年”。这个答案没有错但没有价值。改进办法是让最终生成阶段必须引用子答案的原始片段编号同时要求它“如果子答案之间存在冲突需要明确指出冲突并给出倾向性结论”。实测下来明确要求引用原文之后模型糊弄的概率降低了不少。第三个教训是路由到多个库之后片段总量超窗。一开始我在多库路由时每个库都取 top_k8三个库就是 24 个片段很快就把上下文撑爆了生成质量反而下降。后来我把多库场景下的每个库 top_k 改成 3 到 4然后统一过一个 rerank只保留 6 个以内片段效果立刻恢复。多库场景跟单库完全不同它讲究的是“平衡”而不是“越多越好”。4.3 评估指标与本地工具选型建议评估分诊和拆题不能只看端到端最终答案对不对要分阶段看。分诊阶段的评估我通常准备一份带标注的 query 集每个 query 标注了“该走 direct_answer / 单库 / 多库 / 拆题”然后看 action 命中率。拆题阶段的评估我主要看两点子问题是否覆盖原始问题的信息需求以及子问题是否都能独立检索。这两个指标不需要太复杂抽几十条评估样本人工看一遍就知道问题在哪。端到端评估则是老办法检索命中率黄金答案是否出现在召回片段里加最终答案准确率有条件就用 LLM-as-judge没条件就人工抽检。工具选型上如果你的目标场景是本地部署我建议优先用 Ollama 跑一个参数量至少 70 亿的模型做分诊和拆题再配合自定义提示词和解析逻辑。别一上来就引入重框架LangChain 和 LlamaIndex 固然提供了 Router 和 SubQuestionQueryEngine 之类的组件但核心还是提示词和结构化输出。我见过太多人把框架搭起来之后真正跑起来还是靠一堆 prompt protocol 和回调钩子累得要死。先用最简单的 Python 函数把链路跑通等业务上确实需要扩展再去接框架。Java 侧的话可以关注 LangChain4j 的 Easy RAG 思路不过原理大同小异。另外提一句做拆题之前先把知识库切好。这个听着像废话但我见过太多人花大力气优化拆题和分诊结果原始文档切得乱七八糟标题被切断、表格被拆碎拆题拆得再好也检索不到正确答案。文本拆解是地基本地 RAG 文本拆解工具不一定要多高级先把段落和标题结构吃透再谈 AI 拆题。分享了这么多我个人的核心体会是给 RAG 加分诊台和拆题术之后整个系统真正提升的不只是准确率更是确定性。以前一次问答全链路匀速运转什么问题都一个待遇现在系统知道什么时候该快什么时候该慢什么时候该调工具链路变得清清楚楚。最后再分享一个小技巧做这套东西不要一上来就接复杂的 Agent 调度先把分诊和拆题在两个固定业务问题上跑通把结构化输出和兜底策略写稳再慢慢加工具调用。判断和规划这条路走稳了后面接什么能力都顺。
阅读完成 · 觉得有帮助？
