查询分解：把大问题拆成小问题再检索

可视化演示

演示LR分析器对输入id+id*id的移进-归约解析过程：通过栈操作展示shift和reduce动作，最终接受输入并完成语法分析。

问题分解策略

请客吃饭的第一步永远不是冲进厨房，而是坐下来写菜单。主人家会先想清楚客人的口味：有没有人忌辣？有没有小孩？荤素怎么搭配？然后才落笔把“一桌菜”拆成凉菜、热菜、汤、甜品。查询分解（Query Decomposition） 的第一节，做的就是这件事——拿到用户的大问题后，先不急着检索，而是先“写菜单”：拆成哪几个子问题。

从比喻落到技术上，这一步的输入是主查询（Main Query），输出是一组子查询（Sub-Query） 的列表。这组子查询必须满足两个关键性质：覆盖性（Coverage） 和独立性（Independence）。覆盖性指所有子问题合起来，必须完整覆盖主问题的全部信息需求，不能漏掉任何关键维度；独立性指子问题之间尽可能互不重叠，避免重复检索同一段知识。回到菜单的比喻：覆盖性保证“每道菜都有着落”，独立性保证“两道菜不会都跑去抢同一块五花肉”。

用集合语言可以精确描述这两个性质。设主问题为 $\text{Main}$，拆出的 $n$ 个子问题为 $\text{Sub}_1, \text{Sub}_2, \ldots, \text{Sub}_n$：

写菜单时有条经验：好菜单的每道菜都边界清晰、互不抢料。子问题拆分也是如此。拆得好，后续每一步都会顺畅；拆得烂，检索再快也弥补不了结构上的漏洞。

提示词设计

家宴主人决定了菜单后，不会自己闷头干，而是把任务分给帮厨。但真正会做事的主人，一定会写“任务卡”：菜名、要点、时限、交付物。含糊一句“你去弄一下”的结果往往是帮厨端出一盘完全跑偏的菜。提示词（Prompt）就是给大模型的任务卡。

在设计查询分解提示词时，有几样“任务卡必填项”。第一，角色与任务设定（Role Setting）：明确告诉模型“你是一个问题拆解助手，负责将复杂问题拆成可独立检索的原子子问题”。第二，输出格式（Output Format）：要求模型输出结构化的JSON列表，例如 {"sub_questions": ["子问题1", "子问题2"]}，方便下游程序直接解析调度。第三，示例引导（Few-shot）：给一两个“问题→拆解结果”的样例，模型会模仿样例的拆分粒度。第四，边界约束（Constraints）：提醒模型“不得改变原问题语义”“按原子粒度拆分”“拆分数量3-5个为宜”。

拆解质量不是拍脑袋，可以量化打分量化为三个维度——覆盖度、独立度、可检索度。令覆盖度指标为 $\alpha$ 权重、独立度为 $\beta$ 权重、可检索度为 $\gamma$ 权重，那么一次拆分的整体质量分数可以写成：

任务卡写得越好，帮厨越不容易跑偏。提示词里把角色、格式、示例、边界四件事写清楚，模型的拆分质量就会像拿着任务卡做菜的帮厨一样稳定可靠。

并行检索调度

菜单定好、任务卡发出去，真正高效的后厨是什么样的？绝对不会是每个帮厨都排队等同一口锅。有经验的厨房一定是：凉菜师傅在自己案板上切，蒸锅师傅守着灶，炒锅师傅在另一边颠勺——多线并行，互不阻塞。子问题RAG的检索阶段，正是借鉴了这样的“后厨流水线”。

调度机制分三步：批量分发（Dispatch）、并发执行（Execute）、结果收集（Collect）。先把所有子问题打包成一批任务，同时发给多个检索执行单元（可以理解为多个“检索器”或“工人”），每个工人独立完成自己的检索任务后，把结果统一交回调度中心。因为子问题之间具有独立性，所以它们的检索过程互不依赖，天然适合并行。

并行带来的时间收益非常直观。假如串行执行，总耗时是每个子问题检索耗时的累加；并行执行时，总耗时约等于最慢的那个子问题耗时，加上最后的聚合开销。设 $n$ 个子问题的检索耗时分别为 $T_1, T_2, \ldots, T_n$：

后厨的高效来自并行，RAG的高效同样来自并行。拆出的子问题越多，并行检索的省时效果越显著，但也要注意一次拆得太多会引入调度和聚合的额外开销——这正是第五节“递归与终止”要讲的度。

答案聚合与冲突消解

没有哪家家宴是菜炒好直接往桌上一扔就开吃的。主人会先尝一遍：凉菜咸了要调，热菜重复了要撤掉一道，口味相冲的菜调整上菜顺序。答案聚合（Answer Aggregation） 做的就是“尝菜调整”这一步：把各条子问题的检索答案收集起来，去重、消解、校验，最后端出一桌协调的席面。

聚合面对的首要问题是冗余。三个子问题可能都检索到同一篇文档的同一段话，直接拼接会让最终答案里出现三遍重复内容，所以第一步是去重。第二步是冲突消解（Conflict Resolution）。子问题A回答“这家云厂商东南亚节点延迟低”，子问题B回答“这家云厂商东南亚没有可用区”——两个答案矛盾时，需要一套仲裁机制。常用的方法包括多数投票（Majority Voting） 和置信度加权（Confidence Weighting），即看多个子答案中哪个答案的证据更充分、被更多子检索支持，就采信哪个。

设某个候选最终答案 $F$，$n$ 个子答案中有 $k$ 个支持 $F$，可以用投票分数来衡量这个候选答案的可信度：

上菜前的“尝菜”环节决定了这桌饭能不能拿得出手。答案聚合后的最终回答，也要经过去重、仲裁、校验这三道把关，才能交给用户。

递归与终止

家宴菜单不是一成不变的。如果客人是广东人，主人想做一道“佛跳墙”当硬菜——这道菜里又有鲍鱼、又有花胶、又有鸽子蛋，每个食材的处理方式不同，于是任务卡上又拆出三张小任务卡。这就是递归分解（Recursive Decomposition）：子问题如果仍然太复杂，可以继续往下拆，形成一棵“问题树”。

但递归不能无限进行下去。厨房里时间有限、灶台有限，过了火候菜反而做砸了。所以必须有终止条件（Termination Condition） 和深度上限（Depth Limit）。两个常用判据：一是看子问题的粒度是否已足够“原子化”——如果子问题已经具体到可以直接检索命中，就停止拆分；二是看拆分的深度是否超过预设上限——比如规定最多拆3层，更深就不再递归。此外，如果拆到最底层仍然无法检索到有效答案，还需要一个兜底策略（Fallback）：直接返回现有检索结果或提示用户收窄问题。

把递归的收益和成本放在一起看，可以得到终止的量化依据。设分解深度为 $d$，每多拆一层的平均精度收益为 $\Delta A(d)$，而每层的代价为 $c$（时间与计算开销），最优深度出现在边际收益等于边际成本处：

好厨师懂得“见好就收”。递归分解同样讲究“恰到好处”——既要拆到子问题足够清晰可检索，又不能在收益递减后继续浪费算力。

适用场景

不是每次吃饭都得摆一桌家宴。平时一个人在家，下碗面条5分钟搞定就好；只有客人来访、值得庆贺的日子才值得大动干戈准备一桌。查询分解也是如此：先评估收益，再决定要不要拆。 不是所有问题都值得分解，拆得好是效率倍增器，拆得不好是画蛇添足。

最适合查询分解的是三类场景。第一，多条件约束查询（Multi-constraint Query）：例如“找出上海静安区、月租3000以内、允许养猫的一室户”，每个条件是一个独立的检索维度，合在一起检索容易顾此失彼。第二，对比类问题（Comparative Question）：例如“A云和B云谁的按量计费更便宜”——需要分别检索两家厂商的计费规则再对比，天然适合拆成两个子问题。第三，多属性检索（Multi-attribute Retrieval）：例如“找一款屏幕好、续航长、价格低于5000的笔记本”，三个属性各自检索再聚合，比一次性检索更精准。

什么时候拆、什么时候不拆，可以量化成收益评估模型。设 $V$ 为答案的价值（比如用户决策的重要性），$P(\text{miss})$ 为一次直接检索失败的先验概率（问题越复杂这个概率越高），$C_{\text{decomp}}$ 为分解成本，$C_{\text{latency}}$ 为额外延迟成本：

一句话：“值得请客才开家宴，问题够复杂才拆解。”掌握这个判断标准，查询分解就不是一把无脑挥舞的锤子，而是一把只在合适时机出鞘的手术刀。

AI 应用场景

Python 演示

# -*- coding: utf-8 -*-"""query_decomposition_demo.py一个极简的查询分解RAG演示：1. 用规则把大问题拆成子问题2. 每个子问题独立检索模拟文档库3. 聚合子答案并去重输出""" import reimport timefrom concurrent.futures import ThreadPoolExecutor  DOC_STORE = {    "aws": "AWS在东南亚有新加坡和雅加达两个region，按量计费每小时0.10美元起。",    "azure": "Azure东南亚region在新加坡，月度订阅套餐更优惠，需预付1年合约。",    "gcp": "GCP在东南亚有新加坡和雅加达region，网络流量费较低。",    "price": "按量计费适合短期弹性负载，长期稳定业务建议包年包月。",    "latency": "新加坡region对东南亚用户延迟最低，雅加达region覆盖印尼本地更优。",    "team": "中小团队通常选择包年包月以控制成本，起步配置约每月50美元。",}  class QueryDecomposer:    def __init__(self):        self.max_depth = 2        self.gc_count = 0     def decompose(self, question):        """按关键词规则将问题拆成子问题列表"""        sub_list = []        if "云厂商" in question and "东南亚" in question:            for vendor in ["AWS", "Azure", "GCP"]:                sub_list.append(f"{vendor}在东南亚的计费规则是什么")                sub_list.append(f"{vendor}在东南亚的region分布在哪里")        if "适合" in question and "团队" in question:            sub_list.append("中小团队应该选择按量计费还是包年包月")        if not sub_list:            sub_list = [question]        return sub_list[:6]  class SimpleRetriever:    def __init__(self):        self.doc_list = list(DOC_STORE.keys())        self.free_lists = {}     def search(self, sub_query):        """按关键词命中文档，模拟检索"""        self.free_lists["hit"] = []        for key, text in DOC_STORE.items():            for kw in sub_query.split("的"):                kw = kw.strip()                if kw and kw in text:                    self.free_lists["hit"].append(text)                    break        time.sleep(0.2)        return list(set(self.free_lists["hit"]))  class AnswerAggregator:    def __init__(self):        self.gc_count = 0     def aggregate(self, sub_answers):        """合并多个子答案，去重后拼接"""        seen_set = set()        merged = []        for ans in sub_answers:            for piece in ans:                if piece not in seen_set:                    seen_set.add(piece)                    merged.append(piece)        return " ".join(merged)  def main():    question = "对比AWS、Azure、GCP在东南亚的云服务，推荐一个适合中小团队的方案"    print("原始问题:", question)     decomposer = QueryDecomposer()    sub_questions = decomposer.decompose(question)    print("拆解出的子问题:")    for i, sq in enumerate(sub_questions, 1):        print(f"  {i}. {sq}")     retriever = SimpleRetriever()    start = time.time()    with ThreadPoolExecutor(max_workers=4) as pool:        results = list(pool.map(retriever.search, sub_questions))    print(f"并行检索完成，耗时 {time.time() - start:.2f} 秒")     aggregator = AnswerAggregator()    final_answer = aggregator.aggregate(results)    print("聚合后的最终回答:")    print(" ", final_answer)  if __name__ == "__main__":    main()

1. sub_list.append(f"{vendor}在东南亚的计费规则是什么")按供应商维度生成独立子问题，保证覆盖性与独立性
2. if not sub_list: sub_list = [question]兜底策略：如果拆不出子问题，就直接用原问题检索，不强行拆分
3. ThreadPoolExecutor(max_workers=4)创建4个并行线程执行检索调度，对应正文的并行检索阶段
4. pool.map(retriever.search, sub_questions)批量分发所有子问题，并发执行，自动收集结果
5. for kw in sub_query.split("的")用简单关键词匹配模拟向量检索，教学演示用，生产环境替换为真实检索
6. if piece not in seen_set答案聚合时按片段内容去重，避免多个子问题命中同一段文档被重复拼接
7. aggregator.aggregate(results)统一收集所有子答案并合并为最终回答，对应答案聚合与冲突消解阶段

常见误区

面试题