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

一、每条查询都走 RAG + LLM，成本爆炸

一个知识库问答系统，用户每天发起 10 万次查询。每条查询都要经过 Embedding 检索 + LLM 生成答案，日均 API 费用 3000 元。

分析日志后发现：

• 40% 的查询是"你好"、"谢谢"、"在吗"之类的闲聊，根本不需要检索
• 15% 的查询是可以用 FAQ 直接匹配的常见问题，也不需要 LLM 推理
• 25% 的查询可以通过关键词搜索返回文档列表，不需要 LLM 总结
• 只有 20% 的查询真正需要通过 RAG + LLM 生成答案

如果能在查询入口处做一个意图路由器，将不同类型的查询分发到不同的处理链路，可以大幅降低大模型调用成本。

二、意图路由架构

flowchart TB
    Query["用户查询"] --> Router["意图路由器\n轻量分类模型"]

    Router -->|"闲聊类\n（40%）"| Chat["闲聊处理器\n规则 + 小模型回复\n成本: 0"]
    Router -->|"FAQ 类\n（15%）"| FAQ["FAQ 匹配\nES 精确匹配\n成本: ￥0.001"]
    Router -->|"搜索类\n（25%）"| Search["文档搜索\nBM25 + Embedding 检索\n返回 Top-K 文档\n成本: ￥0.01"]
    Router -->|"推理类\n（20%）"| 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_features=5000,
            ngram_range=(1, 2),
            min_df=2,
        )
        self.classifier = LogisticRegression(
            max_iter=1000,
            C=1.0,
            class_weight="balanced",
        )
    
    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,
                keywords=matched_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_reason=f"ML 分类异常: {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(
            answer=answer,
            source=source,
            cost=cost,
            latency_ms=latency,
        )
    
    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 系统性价比的有效手段：

1. 规则层处理高频简单意图（< 1ms 延迟）
2. ML 层区分搜索和推理意图（< 5ms 延迟）
3. 统计层监控成本节省效果，持续优化路由策略

在 10 万次/天的查询量下，40% 的闲聊 + FAQ 查询被拦截，日均成本从 3000 元降到约 1200 元——路由器本身的开发成本（约 3 人天）在一周内就回本了。