RAG 跨文档推理当答案需要综合多篇文档信息时怎么组织检索流程一、深度引言与场景痛点大部分 RAG 系统的典型工作流是用户问一个问题先检索 Top-K 最相关的文档块然后把文档块和问题一起丢给 LLM 生成答案。这套流程在一句话就能回答的问题上表现不错但遇到跨文档推理就露馅了。举个例子用户问A 公司和 B 公司的营销策略有什么区别A 公司的策略在三篇文档里B 公司的策略在另外两篇标准 RAG 捞回来的是混合的 Top-K 块LLM 需要自己从这一锅粥里分离 A 和 B 的信息再做对比分析。如果 Top-K 里某个公司的信息不完整对比就显然有偏。更棘手的场景是答案的每一部分都在不同文档里但这些文档之间语义相似度不高。比如本季度的销售额下降原因在供应链报告里影响在客户投诉分析里改进措施在管理会议纪要里。这三类文档的语义空间离得很远标准检索很难把它们一起召回。二、底层机制与原理深度剖析flowchart TD A[用户提问] -- B[问题分解器] B -- C1[子问题1: 提取实体A信息] B -- C2[子问题2: 提取实体B信息] B -- C3[子问题3: 提取对比维度] C1 -- D1[第一次检索: 实体A相关] C2 -- D2[第一次检索: 实体B相关] C3 -- D3[第一次检索: 对比框架] D1 -- E1[中间结果A] D2 -- E2[中间结果B] D3 -- E3[对比维度列表] E1 -- F[交叉增强检索] E2 -- F E3 -- F F -- G[第二轮检索: 补全缺失信息] G -- H[信息完整性检查] H -- I{所有维度完整?} I --|否| J[识别缺失维度, 生成补充查询] J -- G I --|是| K[结构化聚合] K -- L[对比分析推理] L -- M[引用标注与溯源] M -- N[输出: 结构化答案]这套流程的核心是把检索拆成两轮。第一轮针对每个子实体独立检索保证 A 的信息从 A 的文档里来B 的信息从 B 的文档里来。第二轮做交叉增强——用第一轮得到的关键词去对方文档里补全缺失维度。比如第一轮发现 A 公司有营销预算数据但 B 公司没有。第二轮就用B 公司 营销预算去全文检索一次补全这个缺失。这样两轮之后对比所需的信息才可能完整。信息完整性检查是一个容易省略但很关键的步骤。它的逻辑很简单检查对比分析所需的每个维度预算、渠道、转化率、ROI 等在 A 和 B 两边是否都有数据。缺少任何一边自动生成补充查询。三、生产级代码实现from __future__ import annotations import asyncio from dataclasses import dataclass, field from typing import Any, Callable, Awaitable dataclass class SubQuery: id: str query_text: str target_entity: str # 要检索的实体 retrieved_chunks: list[str] field(default_factorylist) dataclass class ComparisonDimension: name: str entity_a_value: str entity_b_value: str class MultiDocReasoner: 多文档推理引擎分解 → 多路检索 → 交叉补全 → 聚合推理 def __init__( self, search_fn: Callable[[str, int], Awaitable[list[str]]], llm_fn: Callable[[str], Awaitable[str]], ): self._search search_fn self._llm llm_fn async def cross_document_reason(self, question: str, top_k: int 5) - dict: # 第一步问题分解 sub_queries await self._decompose_question(question) # 第二步并行检索 retrievals [self._search(sq.query_text, top_k) for sq in sub_queries] results await asyncio.gather(*retrievals, return_exceptionsTrue) for sq, result in zip(sub_queries, results): if isinstance(result, Exception): sq.retrieved_chunks [] else: sq.retrieved_chunks result # 第三步交叉增强 await self._cross_augment(sub_queries, top_k) # 第四步完整性检查与补充 missing await self._check_completeness(sub_queries, question) if missing: extra_queries [m[query] for m in missing] extra_results await asyncio.gather( *[self._search(q, top_k) for q in extra_queries], return_exceptionsTrue, ) for i, m in enumerate(missing): if not isinstance(extra_results[i], Exception): m[sub_query].retrieved_chunks.extend(extra_results[i]) # 第五步结构化聚合 answer await self._aggregate_and_reason(sub_queries, question) return {answer: answer, sub_queries: sub_queries, sources: sub_queries} async def _decompose_question(self, question: str) - list[SubQuery]: prompt ( f将以下问题分解为多个独立检索子问题每个子问题对应一个检索对象: \n{question}\n JSON 格式输出: [{\id\: \...\, \query\: \...\, \entity\: \...\}] ) raw await self._llm(prompt) import json try: items json.loads(raw) except json.JSONDecodeError: items [{id: q0, query: question, entity: all}] return [ SubQuery(iditem[id], query_textitem[query], target_entityitem[entity]) for item in items ] async def _cross_augment(self, sub_queries: list[SubQuery], top_k: int) - None: # 用每个子查询的结果作为关键词去其他实体检索 all_keywords: set[str] set() for sq in sub_queries: for chunk in sq.retrieved_chunks: # 提取关键词简化取前 5 个词 all_keywords.update(chunk.split()[:5]) for sq in sub_queries: if sq.retrieved_chunks: continue # 已有结果的跳过 augmented_query f{sq.query_text} { .join(list(all_keywords)[:10])} try: sq.retrieved_chunks await self._search(augmented_query, top_k) except Exception: pass async def _check_completeness( self, sub_queries: list[SubQuery], question: str ) - list[dict]: incomplete [] for sq in sub_queries: if not sq.retrieved_chunks: incomplete.append({ sub_query: sq, query: f关于 {sq.target_entity} 的补充信息: {question}, }) return incomplete async def _aggregate_and_reason( self, sub_queries: list[SubQuery], question: str ) - str: context_parts [] for sq in sub_queries: chunks_text \n.join(sq.retrieved_chunks[:3]) context_parts.append(f【{sq.target_entity}】\n{chunks_text}) context \n\n.join(context_parts) prompt ( f基于以下来自不同文档的信息回答用户问题。\n f如果某些信息缺失或矛盾请明确指出。\n\n f问题: {question}\n\n背景信息:\n{context} ) return await self._llm(prompt)_decompose_question把A 和 B 有什么区别拆成查 A 的信息和查 B 的信息两个子查询。拆分的质量决定了整个流程的上限所以 prompt 里要明确要求 JSON 输出并且做了json.JSONDecodeError的降级处理。_cross_augment是整个流程的差异化武器。它从已有结果中提取关键词拼到空结果的子查询上去重新检索。这个方法比盲目的扩大 Top-K更精准因为关键词来自已确认相关的文档不会引入噪声。_check_completeness和_aggregate_and_reason保证了最终答案的诚实度。如果有子查询完全没命中任何文档聚合阶段会明确告诉用户这部分信息缺失而不是用幻觉填补空白。四、边界分析与架构权衡多轮检索意味着额外的 embedding 计算和 LLM 调用。一轮检索 一轮交叉增强成本大约是标准 RAG 的 1.5~2 倍。如果问题本身不需要跨文档推理这个开销就是浪费。所以建议在入口处加一个复杂度判断问题包含对比区别综合等关键词或涉及多个命名字实体时才走跨文档流程。另一个权衡是交叉增强的方向。如果你用 A 的关键词去 B 的文档里搜可能搜回来一堆不相关的。这里需要做相关度过滤最简单的思路是设一个相似度阈值低于阈值的增强结果直接丢弃。否则给 LLM 喂了一堆噪声推理质量反而下降。在多语言场景下同一个实体在不同语言文档里可能叫法不同。交叉增强要支持同义词扩展比如营销和Marketing在检索时需要映射到同一个概念。这可以用一个简单的同义词配置表解决不需要引入额外的 NER 模型。五、总结跨文档推理的核心挑战不是找不到文档而是找回来的文档信息碎片化、维度不对齐。解决路径很明确先做实体级的问题分解独立检索各实体的信息再用交叉增强补全缺失维度最后做结构化聚合推理。落地时控制好成本只有复杂对比类问题才触发多轮检索流程简单问题走标准单轮 RAG。交叉增强的结果要做阈值过滤别喂噪声给 LLM。最终答案里诚实标注哪些信息缺失比强行拼凑完整答案更有价值。