详解RAG系统如何根据问题类型(简单查询/多跳推理/无需检索)动态选择检索模式,避免固定策略带来的成本与质量损失。
你的 RAG 系统在每一个查询上都犯着同样的错误。
不是那种糟糕的错误,而是一种固定不变的错误。
它对每个问题都采用相同数量的文档检索策略——无论这个问题是简单的事实查询、复杂的多跳推理任务,还是你的模型已经足够了解、无需任何检索就能回答的问题。
这种固定策略方式是当前生产 RAG 系统中可避免成本和质量损失的最大来源。而且这完全是架构问题,模型不是问题所在,流水线才是。
自适应 RAG 通过在检索开始前回答一个问题来解决这个问题:这是什么类型的问题,它实际上需要什么检索策略?
这是设计运行时正确做出这一决策的检索流水线的完整端到端指南。
一个生产 RAG 系统接收的查询在需求上差异巨大。
"公司是什么时候成立的?"不需要检索。模型从训练数据中就知道这个答案,检索关于公司历史的文档只会增加延迟、成本和上下文噪声,而不会改善答案。
"我们当前的退款政策是什么?"需要单步检索。针对政策文档语料库进行一次精准的检索就能返回相关内容,迭代式多跳检索只会增加不必要的开销。
"哪些供应商受到了 Q3 物流中断的影响,这与报告给企业客户的延迟发货有何关联?"需要在多个数据源之间进行多跳检索,每个检索轮次之间有中间推理步骤。
固定策略 RAG 流水线对所有三种情况都采用同一种方法。它要么对简单查询过度检索——增加延迟和成本而没有任何质量提升——要么对复杂查询检索不足——返回不完整的上下文,导致幻觉或不完整的答案。
这方面的研究是明确的。Adaptive-RAG 由 Jeong 等人在 NAACL 2024 上发表,证明了将查询路由到最便宜且足够的检索策略——不检索、单步或多步——能够在大幅降低成本的同时达到始终昂贵多跳基线的水平。大多数已部署系统对每个查询都应用同一种范式。将每个查询路由到最便宜且足够的范式可以将消耗的 token 数量减少几个数量级,同时不会损失准确性。
来自 arXiv:2605.31176(2026 年 5 月发表)的 Retriever Portfolios 论文以前所未有的精确度阐述了这个问题:没有单一检索器对所有查询都是最优的,而固定单一检索策略会在多样化的信息需求面前损失大量性能。实践者社区已经从"哪种检索策略最好?"转向"对于这个特定查询,此刻哪种检索策略最好?"
自适应 RAG 是一种检索架构,其中检索策略——检索的方法、深度和数量——是在运行时根据传入查询的特征确定的,而不是在系统设计时固定的。
它不是一种算法。它是一种包含三个设计决策的架构模式:
决策 1:何时检索。 这个查询是否应该触发检索,还是模型可以从参数知识中回答?
决策 2:使用什么策略。 如果需要检索,应该是密集向量搜索、稀疏关键词搜索、混合搜索、图遍历,还是迭代多跳?
决策 3:检索多少。 应该返回多少文档或块?对于聚焦的事实查询和综合任务的正确答案差异巨大。
自适应 RAG 在运行时回答所有这三个问题。该架构由两个阶段组成,在标准检索流水线之前运行:一个复杂度分类器对查询进行特征描述,一个策略路由器将分类映射到检索配置。
这个路由阶段的输出不是一组检索到的文档。它是对检索子系统的指令:执行这个特定策略,使用这些参数,针对这些数据源。
任何自适应 RAG 实现的第一步都是定义系统必须区分的复杂度类别。研究社区已经收敛到一个足够实用可以实现、又足够精确可以驱动有意义的路由决策的分类法。
A 类:无需检索。 可以从模型的参数知识中回答的查询,无需外部依据。简单的事实问题众所周知的定义、在训练数据中稳定且有良好表示的历史事件。将这些路由到完整 RAG 流水线会浪费计算资源,并引入来自检索文档的上下文噪声——而这些文档并没有为模型现有知识增加任何内容。
B 类:单步检索。 需要一次有针对性检索的查询。策略查询、产品规格、程序文档、特定语料库内的定义。答案存在于一个或少数几个文档中,一次精准的检索就足以将其浮出水面。
C 类:多步检索。 需要顺序检索的查询,其中中间结果决定后续检索查询。"哪些客户受到了由基础设施变更导致的服务中断影响?"需要先找到基础设施变更,然后找到它造成的服务中断,然后找到受影响的客户。每个步骤的输出都为下一个查询提供信息。
D 类:聚合查询。 需要跨多个文档综合而没有明确检索链的查询——"本季度客户反馈中的 recurring themes 是什么?"答案不在任何单个文档中;它是从整个语料库的分析中涌现出来的。
E 类:混合查询。 需要多种检索模态的查询——结构化数据和非结构化文本、图连接实体和向量相似内容。策略必须跨模态展开并综合结果。
为你的特定领域正确制定这个分类法比实现任何特定的路由算法都更重要。基于通用研究假设的五类分类法将不如根据你实际查询分布校准的三类分类法表现好。
路由器是自适应 RAG 的核心。它接收传入的查询并产生路由决策——这个查询属于哪个复杂度类别,应该执行哪种检索策略。
三种路由方法已在研究中得到验证:
训练分类器路由。 一个轻量级分类器——原始 Adaptive-RAG 论文中的 T5-Large,2026 年 4 月发表的 RAGRouter-Bench 研究中更小的模型——被训练来从查询文本预测查询复杂度类别。训练需要带有正确复杂度类别标注的查询标注示例。分类器很小、很快、很便宜——在流水线中增加不到 100 毫秒,同时做出路由决策,节省了不必要的检索工作。
RAGRouter-Bench 研究发现,更轻量的分类器——句子嵌入上的逻辑回归、小型 Transformer 分类器——在路由准确性上与更大的 T5 分类器表现相当,同时部署更快、更便宜。路由决策本身不需要大模型。
免训练自适应门控。 TARG——Retrieval as a Decision,arXiv:2511.09803,2026 年 4 月更新——引入了一种免训练方法,使用模型自身的置信度信号来决定是否需要检索。如果模型在查询上的输出概率分布是置信的——少数 token 具有高概率——模型可能从参数知识中知道答案,不需要检索。如果分布是分散的,则触发检索。
TARG 的关键发现:在涵盖简短答案、多跳和长格式任务的五个 QA 基准上,它始终匹配或超过始终检索方法的精确匹配和 F1 分数,同时显著减少检索频率。无需训练数据。无需标注的复杂度类别。只有模型自身的置信度作为路由信号。
基于嵌入的相似度路由。对于拥有成熟查询历史的系统,路由可以基于与先前已分类查询的相似度来驱动。新查询被嵌入后,与具有已知复杂度类别的查询库进行比较。如果库中存在足够相似的已分类查询,新查询继承该分类。这种方法随时间累积价值——随着查询库的增长,路由准确性也随之提升。
路由器的实现
from enum import Enum
from dataclasses import dataclass
from sentence_transformers import SentenceTransformer
import numpy as np
class ComplexityClass(Enum):
NO_RETRIEVAL = "no_retrieval"
SINGLE_STEP = "single_step"
MULTI_STEP = "multi_step"
AGGREGATION = "aggregation"
HYBRID = "hybrid"
@dataclass
class RoutingDecision:
complexity_class: ComplexityClass
retrieval_strategy: str
k_documents: int
data_sources: list[str]
confidence: float
class AdaptiveRouter:
def __init__(self, classifier_model: str, threshold: float = 0.75):
self.encoder = SentenceTransformer(classifier_model)
self.threshold = threshold
def route(self, query: str) -> RoutingDecision:
complexity = self._classify_complexity(query)
return self._map_to_strategy(query, complexity)
def _classify_complexity(self, query: str) -> ComplexityClass:
multi_hop_signals = [
"which", "how does", "why did", "what caused",
"relationship between", "impact of", "correlation"
]
aggregation_signals = [
"themes", "patterns", "summarize all", "across all",
"common", "recurring", "overall"
]
simple_signals = [
"what is", "define", "when was", "who is"
]
query_lower = query.lower()
if any(s in query_lower for s in aggregation_signals):
return ComplexityClass.AGGREGATION
if any(s in query_lower for s in multi_hop_signals):
return ComplexityClass.MULTI_STEP
if any(s in query_lower for s in simple_signals):
if self._model_likely_knows(query):
return ComplexityClass.NO_RETRIEVAL
return ComplexityClass.SINGLE_STEP
return ComplexityClass.SINGLE_STEP
def _model_likely_knows(self, query: str) -> bool:
# In production: call model with low max_tokens,
# measure output entropy as confidence signal (TARG approach)
return False
def _map_to_strategy(
self, query: str, complexity: ComplexityClass
) -> RoutingDecision:
strategy_map = {
ComplexityClass.NO_RETRIEVAL: RoutingDecision(
complexity_class=complexity,
retrieval_strategy="parametric",
k_documents=0,
data_sources=[],
confidence=0.9
),
ComplexityClass.SINGLE_STEP: RoutingDecision(
complexity_class=complexity,
retrieval_strategy="hybrid_search",
k_documents=5,
data_sources=["primary_vector_store"],
confidence=0.85
),
ComplexityClass.MULTI_STEP: RoutingDecision(
complexity_class=complexity,
retrieval_strategy="iterative_multihop",
k_documents=3,
data_sources=["primary_vector_store", "graph_db"],
confidence=0.80
),
ComplexityClass.AGGREGATION: RoutingDecision(
complexity_class=complexity,
retrieval_strategy="global_search",
k_documents=20,
data_sources=["primary_vector_store"],
confidence=0.75
),
}
return strategy_map.get(complexity, strategy_map[ComplexityClass.SINGLE_STEP])
一旦路由器生成了路由决策,检索子系统就会执行相应的策略。六种模式覆盖了企业检索需求的完整空间。
模式 1:参数化(无检索)。查询直接路由到 LLM,不进行任何检索增强。仅用于 A 类查询,此时模型的参数知识足够且可靠。成本:完全消除嵌入和检索成本。
模式 2:单步稠密检索。针对主向量存储进行一次向量相似性搜索。标准的 RAG 管道。适用于 B 类查询,语义内容清晰。成本:一次嵌入调用,一次 ANN 搜索。
模式 3:单步混合检索。结合稠密向量搜索与稀疏 BM25 关键词搜索的一次检索,通过倒数排名融合(Reciprocal Rank Fusion)合并。在典型企业语料上比纯稠密检索提升 15% 到 30% 的召回率。适用于同时包含语义意图和特定术语的查询。
模式 4:迭代多跳检索。多次检索 passes,每个 pass 由前一个 pass 的结果驱动。查询被分解为子查询。每个子查询执行一次检索 pass。结果告知下一个子查询。持续直到检索链满足或达到最大迭代限制。适用于 C 类查询,需要对文档链进行顺序推理。
模式 5:全局聚合搜索。在整个语料库中广泛检索以支持综合任务。可能使用 GraphRAG 社区摘要、文档聚类,或高 k 向量检索配合激进的重排序。适用于 D 类聚合查询,答案从语料库范围的模式分析中浮现,而非来自特定文档检索。
模式 6:联邦多源检索。同时查询多种不同类型的数据源——向量存储、知识图谱、SQL 数据库、文档仓库——并通过合并步骤综合结果。适用于 E 类混合查询,需要来自多种模态类型的信息。
即使在单一检索策略内,检索的文档数量——k——也应该适应查询而非保持固定。
Sun 等人 2025 年提出的 DynamicRAG,自适应地为每个查询确定排名和检索文档数量。其核心组件是使用强化学习训练的动态重排序器,以 LLM 生成响应的质量作为奖励信号。重排序器通过观察哪些 k 值产生最佳下游答案来学习为每个查询选择最优 k。
基于聚类的自适应检索——CAR(arXiv:2511.14769,2025 年 10 月发表)——采用了一种互补方法。CAR 不训练重排序器,而是分析查询-文档相似度距离的聚类模式来确定相似度分布中的自然断点。断点以上的文档被包含;以下的被排除。k 由相似度分布的结构决定,而非固定参数。
CAR 背后的直觉正确且重要:对于聚焦的、具体的查询,相似度分布在少量高度相关文档之后有明显的陡然下降。对于宽泛的、模糊的查询,分布在整个文档集中逐渐衰减。分布的形状告诉你查询需要多少文档。
Taguchi 等人的 Adaptive-k 论文实现了一个更简单的版本:检索大量候选集,然后在相似度分数从最高分数下降超过定义阈值的点处截断。这种基于阈值的截断无需训练即可实现,并提供了学习型 adaptive-k 选择的大部分收益。
def adaptive_k_retrieval(
query_embedding: list[float],
vector_store,
max_candidates: int = 50,
similarity_drop_threshold: float = 0.15
) -> list[dict]:
candidates = vector_store.similarity_search_with_score(
query_embedding, k=max_candidates
)
if not candidates:
return []
top_score = candidates[0][1]
cutoff_score = top_score - similarity_drop_threshold
selected = [
doc for doc, score in candidates
if score >= cutoff_score
]
return selected
当多个检索策略在同一个查询上运行时——稠密向量搜索、稀疏 BM25、图遍历——它们的结果必须融合成单一排名列表。
标准倒数排名融合使用固定权重。每种检索方法对融合排名的贡献相等,无论哪种方法对当前查询最合适。这在平均情况下表现尚可,但对于某种检索方法明显优于其他方法的情况,会在性能上留下相当大的提升空间。
MoRE-RAG — Mixture-of-Retrieval-Experts RAG — 发表在 Lecture Notes in Business Information Processing 2026,将贝叶斯决策理论引入融合机制。它根据每个检索专家在过去相似查询上的可靠程度,推导出每个专家的最优权重。看起来应该偏向稠密检索的查询,会在稠密检索结果上获得较大权重。包含特定技术术语的查询,会在稀疏检索结果上获得较大权重。
核心发现:MoRE-Ensemble 在四个 BEIR 基准数据集和工业维护语料库上,相比标准 RRF 在 NDCG@10 上实现了 18.82% 的平均提升。关键在于,只需要 50 到 200 个带标签的查询-文档对就能学习到稳定的融合权重——这使得在标注预算有限的工业部署场景中变得实用。
贝叶斯融合方法:
import numpy as np
from scipy.special import softmax
class BayesianRetrieverFusion:
def __init__(self, n_experts: int):
self.n_experts = n_experts
self.expert_weights = np.ones(n_experts) / n_experts
self.query_expert_history = []
def update_weights(
self,
query_embedding: list[float],
expert_scores: list[float],
ground_truth_score: float
):
weight_updates = np.array([
ground_truth_score * score
for score in expert_scores
])
self.expert_weights = softmax(
self.expert_weights + 0.01 * weight_updates
)
def fuse(
self,
expert_rankings: list[list[tuple]],
query_embedding: list[float]
) -> list[tuple]:
doc_scores = {}
for expert_idx, ranking in enumerate(expert_rankings):
weight = self.expert_weights[expert_idx]
for rank, (doc_id, score) in enumerate(ranking):
rrf_score = weight / (60 + rank + 1)
doc_scores[doc_id] = doc_scores.get(doc_id, 0) + rrf_score
return sorted(doc_scores.items(), key=lambda x: x[1], reverse=True)
TARG 值得单独介绍,因为它解决了自适应 RAG 问题中最昂贵的部分——决定何时不检索——而无需任何训练数据。
其核心思想优雅而简洁。当模型从参数化知识中知道问题的答案时,其输出 token 概率分布是自信的:少数 token 具有高概率,分布呈峰值状。当模型不知道、且能从检索中受益时,分布则是分散的——许多 token 具有相似的概率,模型真正处于不确定状态。
TARG 将这种置信度信号用作检索门控。模型用非常短的最大 token 预算处理查询——仅需判断它生成的内容是自信的还是不确定的。如果自信,则跳过检索。如果不确定,则运行完整检索流程。
在五个 QA 基准测试(涵盖 NQ-Open、TriviaQA、PopQA、MuSiQue 和 ASQA)上,TARG 在精确匹配和 F1 分数上始终达到或超越基准水平,同时相比总是检索的方法显著降低了检索频率。
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch
class TARGGate:
def __init__(self, model_name: str, confidence_threshold: float = 0.7):
self.tokenizer = AutoTokenizer.from_pretrained(model_name)
self.model = AutoModelForCausalLM.from_pretrained(model_name)
self.threshold = confidence_threshold
def should_retrieve(self, query: str) -> bool:
inputs = self.tokenizer(query, return_tensors="pt")
with torch.no_grad():
outputs = self.model(**inputs)
logits = outputs.logits[:, -1, :]
probs = torch.softmax(logits, dim=-1)
top_prob = probs.max().item()
entropy = -(probs * torch.log(probs + 1e-10)).sum().item()
is_confident = top_prob > self.threshold and entropy < 2.0
return not is_confident
def gate(self, query: str) -> str:
if self.should_retrieve(query):
return "retrieve"
return "parametric"
Retriever Portfolios 论文(arXiv:2605.31176)提供了迄今为止理论上最严谨的自适应 RAG 框架。它将策略选择问题表述为投资组合优化:给定一组具有已知性能画像的可用检索策略,选择能最大化每个查询预期检索质量的组合配置。
投资组合类比非常精确。在金融投资组合理论中,你不会把所有资金投入单一资产。你会根据预期回报以及当前市场状况下哪种资产最合适来进行资产配置。在检索组合中,你也不会只依赖一种检索策略。你维护一组策略,并根据查询特征和预期策略表现,为每个查询选择最优配置。
相比此前自适应 RAG 工作的关键贡献在于:不再从一个小的固定策略菜单——如 Adaptive-RAG 的三个选项——中进行选择,而是转向在更大策略空间中进行原则性选择的方法。该投资组合框架不手工设计固定策略集,而是允许将任何检索配置添加到组合中,并从生产数据中学习哪种配置在哪种查询类型上表现最佳。
该框架使自适应 RAG 成为随时间推移而非静态改进的系统。随着生产数据积累——关于哪些策略在哪些查询类型上表现最佳——组合权重会更新,路由决策也会改进。
整合所有组件的完整端到端自适应 RAG 流程:
from dataclasses import dataclass
from typing import Optional
@dataclass
class AdaptiveRAGResult:
response: str
strategy_used: str
k_retrieved: int
routing_confidence: float
retrieved_documents: list[dict]
latency_ms: float
cost_estimate_usd: float
class AdaptiveRAGPipeline:
def __init__(
self,
router: AdaptiveRouter,
targ_g