RAG 查询意图分类在检索之前先把用户问题分类以选择最优检索策略一、深度引言与场景痛点做过 RAG 系统的人都有这种体验明明向量库里有答案但检索就是找不出来。你把 embedding 模型从 text-embedding-ada-002 换到 bge-large-zhTop-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_k3, similarity_threshold0.80, index_namedocuments, enable_rerankTrue, ), QueryIntent.COMPARATIVE: RetrievalStrategy( top_k10, similarity_threshold0.65, index_namedocuments, post_processcompare, ), QueryIntent.AGGREGATE: RetrievalStrategy( top_k20, similarity_threshold0.55, index_namedocuments, post_processaggregate, ), QueryIntent.REASONING: RetrievalStrategy( top_k15, similarity_threshold0.60, index_namedocuments, post_processchain, ), QueryIntent.CHITCHAT: RetrievalStrategy( top_k0, similarity_threshold0.0, index_namenone, enable_rerankFalse, ), } 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(fintent:{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, keylambda 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 fintent:{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: fdoc_{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 只做它擅长的事。
RAG 查询意图分类在检索之前先把用户问题分类以选择最优检索策略一、深度引言与场景痛点做过 RAG 系统的人都有这种体验明明向量库里有答案但检索就是找不出来。你把 embedding 模型从 text-embedding-ada-002 换到 bge-large-zhTop-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_k3, similarity_threshold0.80, index_namedocuments, enable_rerankTrue, ), QueryIntent.COMPARATIVE: RetrievalStrategy( top_k10, similarity_threshold0.65, index_namedocuments, post_processcompare, ), QueryIntent.AGGREGATE: RetrievalStrategy( top_k20, similarity_threshold0.55, index_namedocuments, post_processaggregate, ), QueryIntent.REASONING: RetrievalStrategy( top_k15, similarity_threshold0.60, index_namedocuments, post_processchain, ), QueryIntent.CHITCHAT: RetrievalStrategy( top_k0, similarity_threshold0.0, index_namenone, enable_rerankFalse, ), } 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(fintent:{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, keylambda 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 fintent:{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: fdoc_{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 只做它擅长的事。
