企业官网建设流程全解析

RAG 查询路由与知识库分片：多领域知识的精准分发策略

一、全量检索的精度陷阱：当知识库跨越多个业务领域

RAG 系统在单一领域知识库上表现良好，但当知识库覆盖多个业务领域时，检索精度会急剧下降。根本原因是：用户查询往往只涉及一个特定领域，但向量检索会在全量文档中搜索，返回大量来自其他领域的"语义相似但业务无关"的文档片段。某企业知识库同时包含人力资源、财务合规和技术运维三个领域的文档，当用户查询"如何申请年假"时，检索结果中混入了"年度服务器维护计划"和"年度财务审计流程"——因为"年假"和"年度"在向量空间中距离很近。

全量检索的另一个问题是索引规模膨胀带来的延迟增长。当文档数量从 10 万增长到 100 万时，HNSW 索引的查询延迟从 15ms 上升到 120ms，且 GPU 显存占用翻倍。

二、查询路由与分片检索的架构设计

查询路由的核心思路是：先判断用户查询属于哪个领域，再只在该领域的知识分片中检索，避免跨领域噪声干扰。

flowchart TB A[用户查询] --> B[查询路由分类器] B --> C{领域判定} C -->|人力资源| D[HR 知识分片] C -->|财务合规| E[财务知识分片] C -->|技术运维| F[运维知识分片] C -->|不确定| G[全量检索 + 置信度加权] D --> H[分片内向量检索] E --> H F --> H G --> I[多分片并行检索] H --> J[重排序与答案生成] I --> J style B fill:#fff3e0 style C fill:#e8eaf6 style J fill:#e8f5e9

三、查询路由与分片检索的实现

# query_router.py # 查询路由分类器：基于 LLM 的领域判定 + 置信度评估 from dataclasses import dataclass from typing import List, Optional import asyncio @dataclass class RoutingResult: """路由判定结果""" domain: str # 判定的领域 confidence: float # 判定置信度 [0, 1] fallback: bool # 是否需要降级到全量检索 class QueryRouter: """查询路由器：将用户查询分发到对应的知识分片""" def __init__(self, llm_client, domains: List[str], confidence_threshold: float = 0.7): self.llm_client = llm_client self.domains = domains self.confidence_threshold = confidence_threshold async def route(self, query: str) -> RoutingResult: """对用户查询进行领域路由""" prompt = f"""判断以下用户查询属于哪个业务领域。 可选领域：{', '.join(self.domains)} 规则： - 只能选择一个最匹配的领域 - 如果查询可能涉及多个领域或无法确定，置信度设为 0.5 以下 用户查询：{query} 请以 JSON 格式返回：{{"domain": "领域名", "confidence": 0.0-1.0}} """ response = await self.llm_client.chat.completions.create( model="gpt-4o-mini", # 路由用轻量模型，降低成本 messages=[{"role": "user", "content": prompt}], response_format={"type": "json_object"}, ) import json result = json.loads(response.choices[0].message.content) domain = result.get("domain", "unknown") confidence = result.get("confidence", 0.0) # 置信度低于阈值时，标记为需要降级全量检索 fallback = confidence < self.confidence_threshold return RoutingResult( domain=domain, confidence=confidence, fallback=fallback, )

# sharded_retriever.py # 分片检索器：按领域分片索引，支持单分片和多分片并行检索 from typing import Dict, List, Any import asyncio class ShardedRetriever: """分片检索器：每个领域维护独立的向量索引""" def __init__(self): # domain -> vector_index 的映射 self._shards: Dict[str, Any] = {} self._global_index: Any = None # 全量索引，用于降级检索 def register_shard(self, domain: str, index: Any) -> None: """注册领域分片索引""" self._shards[domain] = index def register_global_index(self, index: Any) -> None: """注册全量索引（降级使用）""" self._global_index = index async def search_shard( self, domain: str, query_embedding: List[float], top_k: int = 5 ) -> List[Any]: """在指定领域分片中检索""" shard = self._shards.get(domain) if shard is None: raise ValueError(f"领域 {domain} 无对应分片索引") return await shard.search(query_embedding, top_k=top_k) async def search_multi_shard( self, domains: List[str], query_embedding: List[float], top_k_per_shard: int = 3 ) -> List[Any]: """多分片并行检索，合并结果""" tasks = [ self.search_shard(domain, query_embedding, top_k_per_shard) for domain in domains ] results = await asyncio.gather(*tasks, return_exceptions=True) # 合并并去重 merged = [] seen_ids = set() for result in results: if isinstance(result, Exception): continue # 单分片失败不影响整体 for doc in result: if doc.id not in seen_ids: merged.append(doc) seen_ids.add(doc.id) return merged async def search_global( self, query_embedding: List[float], top_k: int = 10 ) -> List[Any]: """全量检索（降级方案）""" if self._global_index is None: raise RuntimeError("未注册全量索引") return await self._global_index.search(query_embedding, top_k=top_k)

四、分片检索的权衡与适用边界

路由错误的代价。查询路由将查询分发到错误领域时，检索结果完全无关，比全量检索的噪声问题更严重——全量检索至少还有概率命中正确文档，而错误分片检索的命中率为零。缓解方案是设置置信度阈值，低于阈值时降级到全量检索或多分片并行检索，但这又带来了额外的延迟和计算开销。

分片粒度的两难。分片过粗（如只分"技术"和"业务"）无法有效过滤噪声，分片过细（如按每个产品线分片）则导致单个分片文档量太少，检索召回不足。实测数据显示，单个分片的文档量低于 5000 条时，Top-5 召回率比全量检索低 12%。建议分片粒度以每个分片 1-5 万条文档为宜。

分片维护成本。每个分片需要独立维护索引、更新文档和监控检索质量。当领域边界发生变化（如新增业务线）时，需要重新规划分片策略并迁移数据。对于文档更新频繁的场景，分片间的数据一致性也需要额外关注。

适用边界：适合知识库覆盖 3 个以上明显不同领域、总文档量超过 10 万条的场景；不适合单一领域或文档量低于 5 万条的场景，此时全量检索的精度已经足够。

五、总结

查询路由与知识库分片是解决大规模多领域 RAG 检索精度问题的有效策略。核心要点：路由分类器是整个方案的基石，其准确率直接决定分片检索的效果，建议使用轻量模型降低路由延迟和成本；置信度阈值是安全阀，低于阈值时降级到全量检索或多分片并行检索；分片粒度需要平衡噪声过滤效果和单分片召回率，1-5 万条文档/分片是较优区间。落地建议：先在离线数据集上评估路由准确率，达到 90% 以上再上线；初期采用"路由 + 全量降级"的混合模式，逐步提高路由置信度阈值。