企业官网建设流程全解析

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 人天）在一周内就回本了。