RAG 查询意图分类：用轻量模型做前置路由，节省大模型调用

RAG 查询意图分类用轻量模型做前置路由节省大模型调用一、每条查询都走 RAG LLM成本爆炸一个知识库问答系统用户每天发起 10 万次查询。每条查询都要经过 Embedding 检索 LLM 生成答案日均 API 费用 3000 元。分析日志后发现40% 的查询是你好、谢谢、在吗之类的闲聊根本不需要检索15% 的查询是可以用 FAQ 直接匹配的常见问题也不需要 LLM 推理25% 的查询可以通过关键词搜索返回文档列表不需要 LLM 总结只有 20% 的查询真正需要通过 RAG LLM 生成答案如果能在查询入口处做一个意图路由器将不同类型的查询分发到不同的处理链路可以大幅降低大模型调用成本。二、意图路由架构flowchart TB Query[用户查询] -- Router[意图路由器\n轻量分类模型] Router --|闲聊类\n40%| Chat[闲聊处理器\n规则 小模型回复\n成本: 0] Router --|FAQ 类\n15%| FAQ[FAQ 匹配\nES 精确匹配\n成本: 0.001] Router --|搜索类\n25%| Search[文档搜索\nBM25 Embedding 检索\n返回 Top-K 文档\n成本: 0.01] Router --|推理类\n20%| RAG[完整 RAG\n检索 LLM 生成\n成本: 0.1] subgraph Cost[每查询成本对比] C1[闲聊: 0] C2[FAQ: 0.001] C3[搜索: 0.01] C4[RAG: 0.1] end Router -- Feedback[反馈回路\n用户行为标注\n持续优化路由准确率]关键设计原则分类模型必须比被节省的成本更便宜。不能为了解决 1 块钱的问题投入 10 块钱的路由器。三、Python 实现轻量级意图路由器意图分类器import re from typing import List, Tuple, Optional from enum import Enum from dataclasses import dataclass import numpy as np from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.linear_model import LogisticRegression import joblib class IntentType(Enum): 查询意图类型 CHAT chat # 闲聊你好、谢谢 FAQ faq # 常见问题怎么退款 SEARCH search # 文档搜索Go 并发模型有哪些 RAG rag # 需要推理为什么 Go 的 channel 比 mutex 更适合这个场景 dataclass class IntentResult: 意图分类结果 intent: IntentType confidence: float # 置信度 [0, 1] keywords: List[str] # 提取的关键词 fallback_reason: Optional[str] None class IntentRouter: 轻量级意图路由器 ——使用规则 轻量 ML 模型处理延迟 5ms def __init__(self, model_path: Optional[str] None): # 规则层快速匹配常见模式 self.chat_patterns [ r^(你好|hi|hello|在吗|在不在|谢谢|thanks|好的|ok|知道了)$, r^(你是谁|你叫什么|你能做什么), ] self.faq_pattern_keywords [ 退款, 退货, 发货, 物流, 价格, 优惠, 密码, 登录, 注册, 支付, 运费, 地址, ] # ML 层用于区分 SEARCH 和 RAG self.vectorizer: Optional[TfidfVectorizer] None self.classifier: Optional[LogisticRegression] None if model_path: self.load_model(model_path) else: # 使用默认的轻量模型 self._init_default_model() def _init_default_model(self): 初始化默认 TF-IDF 逻辑回归训练样本在下面 self.vectorizer TfidfVectorizer( max_features5000, ngram_range(1, 2), min_df2, ) self.classifier LogisticRegression( max_iter1000, C1.0, class_weightbalanced, ) def classify(self, query: str) - IntentResult: 分类查询意图 先走规则快速再走 ML精确 query query.strip() if not query: return IntentResult(IntentType.CHAT, 1.0, []) # 第一层规则匹配延 1ms rule_result self._rule_based_classify(query) if rule_result.confidence 0.8: return rule_result # 第二层ML 分类延迟 2-5ms return self._ml_classify(query) def _rule_based_classify(self, query: str) - IntentResult: 基于规则的快速分类 # 检测闲聊 for pattern in self.chat_patterns: if re.match(pattern, query, re.IGNORECASE): return IntentResult( IntentType.CHAT, 0.95, keywords[], ) # 短查询 5 个字符默认归类为闲聊 if len(query) 5: return IntentResult( IntentType.CHAT, 0.7, keywords[], fallback_reason查询过短, ) # 检测 FAQ 关键词 matched_keywords [ kw for kw in self.faq_pattern_keywords if kw in query ] if matched_keywords and len(query) 30: return IntentResult( IntentType.FAQ, 0.85, keywordsmatched_keywords, ) # 检测搜索特征包含书名号、技术术语等 search_patterns [ r《.》, # 文档/书名引用 r(什么是|怎么用|如何使用), # 知识查询 r(文档|手册|API|SDK|协议), # 技术资料 ] for pattern in search_patterns: if re.search(pattern, query): return IntentResult( IntentType.SEARCH, 0.75, keywords[], ) # 无法通过规则确定返回低置信度 return IntentResult( IntentType.RAG, 0.5, keywords[], fallback_reason规则无法分类转入 ML 分类, ) def _ml_classify(self, query: str) - IntentResult: 基于 ML 的分类SEARCH vs RAG if self.classifier is None or self.vectorizer is None: return IntentResult(IntentType.RAG, 0.6, []) try: vec self.vectorizer.transform([query]) proba self.classifier.predict_proba(vec)[0] pred self.classifier.predict(vec)[0] confidence max(proba) intent_map {0: IntentType.SEARCH, 1: IntentType.RAG} intent intent_map.get(pred, IntentType.RAG) return IntentResult(intent, confidence, []) except Exception as e: # ML 分类失败默认走 RAG保守策略 return IntentResult( IntentType.RAG, 0.4, keywords[], fallback_reasonfML 分类异常: {e}, ) def train(self, samples: List[Tuple[str, IntentType]]): 训练 SEARCH vs RAG 分类器 texts [s[0] for s in samples] labels [1 if s[1] IntentType.RAG else 0 for s in samples] X self.vectorizer.fit_transform(texts) self.classifier.fit(X, labels) # 输出训练统计 accuracy self.classifier.score(X, labels) print(f意图路由器训练完成准确率: {accuracy:.2%}) def save(self, path: str): joblib.dump( {vectorizer: self.vectorizer, classifier: self.classifier}, path, ) def load_model(self, path: str): data joblib.load(path) self.vectorizer data[vectorizer] self.classifier data[classifier]完整查询管线from dataclasses import dataclass from typing import Any, Optional dataclass class QueryResult: 查询结果 answer: str source: str # 来源chat/faq/search/rag cost: float # 成本元 latency_ms: float # 延迟毫秒 class QueryPipeline: 查询管线——根据意图分发到不同处理器 def __init__(self, router: IntentRouter): self.router router self.chat_handler ChatHandler() self.faq_handler FAQHandler() self.search_handler SearchHandler() self.rag_handler RAGHandler() # 统计信息 self.stats { total: 0, chat: 0, faq: 0, search: 0, rag: 0, total_cost: 0.0, } def process(self, query: str) - QueryResult: 处理用户查询——先路由再分发 import time start time.time() # 意图分类 intent_result self.router.classify(query) # 按意图分发 if intent_result.intent IntentType.CHAT: answer self.chat_handler.handle(query) cost 0.0 source chat elif intent_result.intent IntentType.FAQ: answer self.faq_handler.handle(query, intent_result.keywords) cost 0.001 source faq elif intent_result.intent IntentType.SEARCH: answer self.search_handler.handle(query) cost 0.01 source search else: # RAG answer self.rag_handler.handle(query) cost 0.1 source rag latency (time.time() - start) * 1000 # 更新统计 self.stats[total] 1 self.stats[source] 1 self.stats[total_cost] cost # 如果置信度低做异步二次验证 if intent_result.confidence 0.7: self._async_verify(query, intent_result) return QueryResult( answeranswer, sourcesource, costcost, latency_mslatency, ) def get_cost_savings(self) - dict: 计算成本节省 no_router_cost self.stats[total] * 0.1 # 全部走 RAG actual_cost self.stats[total_cost] savings no_router_cost - actual_cost return { total_queries: self.stats[total], without_router: f{no_router_cost:.2f}, with_router: f{actual_cost:.2f}, savings: f{savings:.2f}, savings_pct: f{savings/no_router_cost*100:.1f}%, }四、边界分析与 Trade-offs路由器自身的成本TF-IDF 逻辑回归分类CPU 推理延迟 5ms成本接近于 0但需要维护训练数据——建议从线上日志中抽样标注误分类的代价将 RAG 意图误判为 FAQ用户体验下降得到不完整的回答将闲聊误判为 RAG成本浪费建议对低置信度分类做二次校验规则 小模型交叉验证路由拓扑的适应性随着业务变化意图分布可能改变需要定期分析分类准确率和意图分布更新规则和模型何时不需要路由器查询量 1000 次/天成本节省不明显绝大部分查询都需要 RAG设置路由器反而增加延迟五、总结用轻量级意图路由器做前置分发是提高 RAG 系统性价比的有效手段规则层处理高频简单意图 1ms 延迟ML 层区分搜索和推理意图 5ms 延迟统计层监控成本节省效果持续优化路由策略在 10 万次/天的查询量下40% 的闲聊 FAQ 查询被拦截日均成本从 3000 元降到约 1200 元——路由器本身的开发成本约 3 人天在一周内就回本了。