RAG 查询意图分类：在检索之前先把用户问题分类以选择最优检索策略
一、深度引言与场景痛点
做过 RAG 系统的人都有这种体验：明明向量库里有答案，但检索就是找不出来。你把 embedding 模型从 text-embedding-ada-002 换到 bge-large-zh，Top-K 从 5 调到 20，召回率涨了几个点，但关键问题依然没解决——同一个检索策略在应对不同类型问题时效果天差地别。
举个例子。用户问"公司第三季度的营收是多少？"这是个事实查询，最优策略是精确匹配关键字段 + 高相似度阈值。用户问"和去年同期相比增长了多少？"这是比较查询，需要同时检索两个时间段的数据，还得做数值对比。用户问"哪个部门的增长速度最快？"这是聚合查询，需要对多个文档做统计排序。用户问"如果保持这个增速，明年的营收大概多少？"这是推理查询，需要检索历史趋势数据并做外推。
如果你对所有问题都用同样的 Top-K、同样的相似度阈值、同样的重排序策略，那每种问题都只能得到一个"折中但不够好"的结果。本质上，RAG 系统的质量瓶颈不在检索算法本身，而在于检索策略和问题类型的匹配度。
二、底层机制与原理深度剖析
意图分类不是在给系统增加复杂度，而是在把隐式的策略选择变成显式的。核心思路是：在检索之前加一个轻量级的意图分类器，根据分类结果选择不同的检索策略（索引选择、Top-K、相似度阈值、重排序权重）。
这个架构的关键在于意图分类器必须快。如果分类本身就耗掉 500ms，那后面检索策略优化省下的时间全白搭了。实践中用两级分类：第一级是规则匹配（关键词 + 正则），覆盖 60%-70% 的简单意图，延迟接近零；第二级才调 LLM，处理规则覆盖不到的复杂语义。LLM 分类不走大模型，用轻量级的分类专用模型就行。
策略路由不是简单的 switch-case。事实查询对应高精度策略：少量文档、高相似度阈值、优先匹配结构化字段。比较查询需要更大范围的检索，然后由后处理做数据抽取和对比。聚合查询需要 Low-K 的大范围检索，然后通过 map-reduce 做统计汇总。推理查询则需要在检索结果上做链式推理：先检索事实数据，再从数据中推导趋势。
三、生产级代码实现
import asyncio
import re
from dataclasses import dataclass, field
from enum import Enum
from typing import Any
import redis.asyncio as aioredis
class QueryIntent(str, Enum):
FACTUAL = "factual" # 事实查询：精确、单一答案
COMPARATIVE = "comparative" # 比较查询：对比多个维度
AGGREGATE = "aggregate" # 聚合查询：统计排序
REASONING = "reasoning" # 推理查询：需要推导
CHITCHAT = "chitchat" # 闲聊：不需要检索
@dataclass
class RetrievalStrategy:
top_k: int = 5
similarity_threshold: float = 0.75
index_name: str = "default"
enable_rerank: bool = True
post_process: str = "none" # none / compare / aggregate / chain
# 意图到检索策略的映射
STRATEGY_MAP: dict[QueryIntent, RetrievalStrategy] = {
QueryIntent.FACTUAL: RetrievalStrategy(
top_k=3, similarity_threshold=0.80, index_name="documents", enable_rerank=True,
),
QueryIntent.COMPARATIVE: RetrievalStrategy(
top_k=10, similarity_threshold=0.65, index_name="documents",
post_process="compare",
),
QueryIntent.AGGREGATE: RetrievalStrategy(
top_k=20, similarity_threshold=0.55, index_name="documents",
post_process="aggregate",
),
QueryIntent.REASONING: RetrievalStrategy(
top_k=15, similarity_threshold=0.60, index_name="documents",
post_process="chain",
),
QueryIntent.CHITCHAT: RetrievalStrategy(
top_k=0, similarity_threshold=0.0, index_name="none", enable_rerank=False,
),
}
class IntentClassifier:
"""意图分类器：规则优先，LLM 兜底"""
# 规则模式：正则 + 关键词
PATTERNS: dict[QueryIntent, list[str]] = {
QueryIntent.FACTUAL: [
r"(多少|几[个位]|什么(是|叫)|谁|哪[一个]|何时|在哪里)",
r"^(查询|查看|显示).*(数据|信息|记录)$",
],
QueryIntent.COMPARATIVE: [
r"(和|与|跟|同).*(相比|对比|比较|区别|差异|不同)",
r"(更高|更低|更多|更少|更好|更差|超过|低于)",
r"(同比|环比|增长率|变化趋势)",
],
QueryIntent.AGGREGATE: [
r"(排名|排行|TOP|前[几多少]|最[高低多大长短])",
r"(汇总|统计|合计|平均|占比|分布)",
r"(哪个|哪些).*(最多|最少|最高|最低)",
],
QueryIntent.REASONING: [
r"(为什么|原因|导致|造成|影响|后果)",
r"(如果|假如|假设|预测|预计|趋势|将来|未来)",
r"(建议|方案|措施|对策|如何解决|怎么处理)",
],
}
def __init__(self, cache_ttl: int = 300) -> None:
self._cache_ttl = cache_ttl
self._redis: aioredis.Redis | None = None
# 编译正则
self._compiled_patterns: dict[QueryIntent, list[re.Pattern[str]]] = {}
for intent, patterns in self.PATTERNS.items():
self._compiled_patterns[intent] = [re.compile(p) for p in patterns]
async def classify(self, query: str) -> tuple[QueryIntent, float]:
"""分类查询意图，返回意图和置信度"""
# 1. 检查缓存
if self._redis:
cached = await self._redis.get(f"intent:{hash(query)}")
if cached:
intent_str, confidence = cached.decode().split(":")
return QueryIntent(intent_str), float(confidence)
# 2. 规则匹配（快速路径）
intent, confidence = self._rule_classify(query)
if confidence >= 0.8:
await self._cache_result(query, intent, confidence)
return intent, confidence
# 3. LLM 分类（慢速路径，仅在规则不确定时触发）
intent, confidence = await self._llm_classify(query)
await self._cache_result(query, intent, confidence)
return intent, confidence
def _rule_classify(self, query: str) -> tuple[QueryIntent, float]:
"""基于规则的快速分类"""
scores: dict[QueryIntent, float] = {}
for intent, patterns in self._compiled_patterns.items():
score = sum(
1.0 for p in patterns if p.search(query)
) / len(patterns) if patterns else 0
scores[intent] = score
if not scores:
return QueryIntent.CHITCHAT, 0.0
best = max(scores, key=lambda k: scores[k])
return best, min(scores[best], 0.95)
async def _llm_classify(self, query: str) -> tuple[QueryIntent, float]:
"""LLM 兜底分类（生产环境替换为真实 LLM 调用）"""
# 这里用简单的启发式模拟 LLM 分类
# 实际部署替换为 OpenAI/本地模型的分类调用
await asyncio.sleep(0) # 模拟 async 行为
return QueryIntent.CHITCHAT, 0.5
async def _cache_result(
self, query: str, intent: QueryIntent, confidence: float
) -> None:
"""缓存分类结果，避免重复分类"""
if self._redis:
try:
key = f"intent:{hash(query)}"
value = f"{intent.value}:{confidence}"
await self._redis.setex(key, self._cache_ttl, value)
except (aioredis.ConnectionError, asyncio.TimeoutError):
pass # 缓存失败不影响主流程
class RetrievalRouter:
"""检索路由器：根据意图分发到不同检索策略"""
def __init__(self, classifier: IntentClassifier) -> None:
self.classifier = classifier
async def route_and_retrieve(
self, query: str
) -> tuple[RetrievalStrategy, list[dict[str, Any]], QueryIntent]:
"""路由 + 检索的一体化接口"""
intent, confidence = await self.classifier.classify(query)
strategy = STRATEGY_MAP[intent]
# 闲聊意图直接跳过检索
if intent == QueryIntent.CHITCHAT:
return strategy, [], intent
# 执行检索（生产环境替换为真实的向量检索调用）
docs = await self._execute_retrieval(query, strategy)
return strategy, docs, intent
async def _execute_retrieval(
self, query: str, strategy: RetrievalStrategy
) -> list[dict[str, Any]]:
"""执行向量检索（生产环境接入 Milvus/Pinecone）"""
# 模拟异步检索
await asyncio.sleep(0.05)
return [
{
"id": f"doc_{i}",
"content": f"检索结果 {i}：相关文档片段…",
"score": 1.0 – i * 0.05,
}
for i in range(min(strategy.top_k, 5))
]
async def main() -> None:
classifier = IntentClassifier()
router = RetrievalRouter(classifier)
test_queries = [
"第三季度的销售额是多少？",
"今年的数据和去年相比有哪些变化？",
"哪个产品线的毛利率最高？",
"如果保持现在的增速，明年能到多少？",
"今天天气真不错啊",
]
for query in test_queries:
strategy, docs, intent = await router.route_and_retrieve(query)
print(f"\\n查询: {query}")
print(f" 意图: {intent.value}")
print(f" 策略: top_k={strategy.top_k}, threshold={strategy.similarity_threshold}")
print(f" 结果: {len(docs)} 条文档")
if __name__ == "__main__":
asyncio.run(main())
代码中的性能优化思路：意图分类结果用 Redis 缓存，相同查询不会重复走分类流程。规则匹配先于 LLM——正则的速度是微秒级的，能覆盖大部分常见查询模板。_cache_result 中的 Redis 异常被静默吞下，因为缓存失败不应该阻塞主检索流程。策略映射用字典查表而非 if-else 链，O(1) 的路由开销。
四、边界分析与架构权衡
意图分类器的核心矛盾是准确率和延迟。规则匹配快但覆盖不全，"和去年比有什么变化"能被正则命中，但"看起来没达预期，到底差在哪"这种隐性比较查询规则就抓不住。LLM 分类准但慢，一个分类调用增加 200-500ms，对于要求端到端延迟在 1 秒内的应用来说太奢侈。
折中方案是分层分类 + 缓存。高频查询模式的分类结果被缓存后，命中率能做到 30%-50%。对于长尾查询，规则提供低置信度的猜测，LLM 提供最终确认。如果业务场景里查询模式相对固定（比如客服系统的常见问题），规则 + 缓存就能覆盖 80% 以上的流量。
另一个边界是意图类别的粒度。4 个类别够用吗？实际项目中可能会需要更细的分类（比如"技术问题"vs"业务问题"，"写代码"vs"写文档"）。但分类越细，策略映射越复杂，维护成本越高。建议从粗粒度开始，当某一类的召回率持续偏低时再拆分。
五、总结
RAG 查询意图分类的核心洞察是：不同的查询需要不同的检索策略，一刀切是召回率的天花板。实现上分两层——规则匹配处理高频模式，LLM 兜底处理复杂语义。分类结果加缓存减少重复开销。策略路由用查表方式 O(1) 完成，不影响检索链路的尾延迟。关键是别让分类器变成新的性能瓶颈：该用缓存用缓存，该上规则上规则，LLM 只做它擅长的事。
