RAG 查询路由与知识库分片:多领域知识的精准分发策略
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( modelgpt-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( domaindomain, confidenceconfidence, fallbackfallback, )# 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_ktop_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_exceptionsTrue) # 合并并去重 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_ktop_k)四、分片检索的权衡与适用边界路由错误的代价。查询路由将查询分发到错误领域时检索结果完全无关比全量检索的噪声问题更严重——全量检索至少还有概率命中正确文档而错误分片检索的命中率为零。缓解方案是设置置信度阈值低于阈值时降级到全量检索或多分片并行检索但这又带来了额外的延迟和计算开销。分片粒度的两难。分片过粗如只分技术和业务无法有效过滤噪声分片过细如按每个产品线分片则导致单个分片文档量太少检索召回不足。实测数据显示单个分片的文档量低于 5000 条时Top-5 召回率比全量检索低 12%。建议分片粒度以每个分片 1-5 万条文档为宜。分片维护成本。每个分片需要独立维护索引、更新文档和监控检索质量。当领域边界发生变化如新增业务线时需要重新规划分片策略并迁移数据。对于文档更新频繁的场景分片间的数据一致性也需要额外关注。适用边界适合知识库覆盖 3 个以上明显不同领域、总文档量超过 10 万条的场景不适合单一领域或文档量低于 5 万条的场景此时全量检索的精度已经足够。五、总结查询路由与知识库分片是解决大规模多领域 RAG 检索精度问题的有效策略。核心要点路由分类器是整个方案的基石其准确率直接决定分片检索的效果建议使用轻量模型降低路由延迟和成本置信度阈值是安全阀低于阈值时降级到全量检索或多分片并行检索分片粒度需要平衡噪声过滤效果和单分片召回率1-5 万条文档/分片是较优区间。落地建议先在离线数据集上评估路由准确率达到 90% 以上再上线初期采用路由 全量降级的混合模式逐步提高路由置信度阈值。