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% 以上再上线；初期采用"路由 + 全量降级"的混合模式，逐步提高路由置信度阈值。