多租户向量库权限隔离避坑实录

问题背景：RAG 多租户不是把文档放进同一个向量库就结束

很多团队在单租户版本中验证 RAG 效果后，进入 SaaS 化阶段会直接把所有客户的文档写入同一个向量集合，只在业务表里记录 tenant_id。这样做上线很快，但一旦检索层漏传过滤条件，就可能出现 A 客户召回 B 客户合同、内部知识库被外部用户检索、离职成员仍能访问历史切片等严重问题。

多租户向量检索的核心矛盾是：向量相似度搜索天然面向全量向量，而权限系统要求每次查询都在“当前主体可见的数据边界”内执行。这个边界至少包含租户、用户或角色、知识库、文档状态、授权范围等维度。只依赖提示词约束模型“不要回答其他租户内容”并不可靠，因为越权数据已经进入召回上下文；权限必须在检索前和检索后同时生效。

三种多租户向量隔离方案对比

常见方案可以分为物理独立、逻辑命名空间和元数据过滤三类。

方案	隔离强度	成本	查询复杂度	适用场景
每个租户独立 Collection 或数据库	高	高，索引与连接数量随租户增长	低	大客户、强合规、数据主权要求高
Collection 内使用 namespace / partition	中高	中等	中	标准 SaaS，租户数量多但单租户数据量可控
同一 Collection 加 tenant_id 元数据过滤	中	低，资源利用率高	高，依赖过滤能力	中小租户、快速迭代、平台统一治理

实践中通常采用混合策略：普通租户使用共享 Collection 加 namespace，重点客户使用独立 Collection；无论底层采用哪种方式，应用层都必须传入统一的权限上下文，不允许业务代码自行拼接裸查询。

权限模型设计：租户隔离只是第一道边界

建议把权限拆成四层：

1. Tenant 租户层：最高数据边界，任何写入和查询都不能跨租户。
2. Knowledge Base 知识库层：同一租户下可有公开库、部门库、项目库。
3. Resource 文档层：文档可能属于某个用户、用户组或客户空间。
4. Document State 状态层：草稿、已发布、已删除、失败切片不应被普通用户检索。

向量库中的命名空间适合承载强隔离键，例如 tenant_id；元数据过滤适合表达更细粒度授权，例如 kb_id、owner_id、visibility、status。不要把所有权限都塞进命名空间，否则会生成 tenantA_kbX_userY 这类爆炸式分区；也不要只依赖元数据过滤，因为某些向量数据库在过滤选择性差、索引类型受限时，会出现先召回后过滤导致结果不足或性能抖动。

可运行代码：封装带权限上下文的向量检索客户端

下面示例使用 Python 展示一个可替换底层向量库的安全封装。示例不依赖特定厂商 SDK，核心是强制构造 Scope、写入命名空间、生成过滤表达式，并在返回后再次校验。

# pip install numpy
from dataclasses import dataclass, field
from typing import List, Dict, Any
import numpy as np


@dataclass(frozen=True)
class AccessScope:
    """当前请求主体的权限边界，必须由认证中间件生成，不能由前端传入。"""
    tenant_id: str
    user_id: str
    role: str
    allowed_kb_ids: List[str]
    allowed_group_ids: List[str] = field(default_factory=list)

    @property
    def namespace(self) -> str:
        # 命名空间只放高基数字段：租户。避免按用户、知识库创建海量分区。
        return f"tenant_{self.tenant_id}"

    def to_metadata_filter(self) -> Dict[str, Any]:
        """生成向量库元数据过滤条件，表达知识库、文档状态和行级权限。"""
        return {
            "operator": "AND",
            "conditions": [
                {"field": "tenant_id", "op": "=", "value": self.tenant_id},
                {"field": "kb_id", "op": "IN", "value": self.allowed_kb_ids},
                {"field": "status", "op": "=", "value": "published"},
                {
                    "operator": "OR",
                    "conditions": [
                        {"field": "visibility", "op": "=", "value": "tenant"},
                        {"field": "owner_id", "op": "=", "value": self.user_id},
                        {"field": "group_id", "op": "IN", "value": self.allowed_group_ids},
                    ],
                },
            ],
        }


class InMemorySecureVectorStore:
    """演示用内存向量库；接入 Milvus、Qdrant、Weaviate、pgvector 时保留同样的强制校验。"""

    def __init__(self):
        # 结构：namespace -> list[point]
        self._data: Dict[str, List[Dict[str, Any]]] = {}

    def upsert(self, point: Dict[str, Any]) -> None:
        tenant_id = point.get("tenant_id")
        if not tenant_id:
            raise PermissionError("写入向量必须包含 tenant_id")

        namespace = f"tenant_{tenant_id}"
        self._data.setdefault(namespace, []).append(point)

    def search(
        self,
        query_vector: np.ndarray,
        scope: AccessScope,
        top_k: int = 5,
        score_threshold: float = 0.7,
    ) -> List[Dict[str, Any]]:
        if not scope.tenant_id or not scope.allowed_kb_ids:
            # 空权限必须直接拒绝，不能把空 IN 条件翻译成“不过滤”
            raise PermissionError("权限范围为空，禁止执行向量检索")

        points = self._data.get(scope.namespace, [])
        candidates = []

        for point in points:
            if self._match_filter(point, scope.to_metadata_filter()):
                score = self._cosine(query_vector, np.array(point["vector"], dtype=float))
                if score >= score_threshold:
                    candidates.append((score, point))

        candidates.sort(key=lambda x: x[0], reverse=True)
        results = [
            {"id": p["id"], "score": float(score), "metadata": p["metadata"]}
            for score, p in candidates[:top_k]
        ]

        # 二次鉴权：即使底层过滤异常，也不能把越权切片交给 Rerank 或 LLM
        return [r for r in results if self._post_check(r, scope)]

    def _match_filter(self, point: Dict[str, Any], expr: Dict[str, Any]) -> bool:
        operator = expr.get("operator")
        if operator in {"AND", "OR"}:
            results = [self._match_filter(point, c) for c in expr["conditions"]]
            return all(results) if operator == "AND" else any(results)

        value = point.get(expr["field"])
        expected = expr["value"]
        op = expr["op"]
        if op == "=":
            return value == expected
        if op == "IN":
            return value in expected
        raise ValueError(f"不支持的过滤操作: {op}")

    def _post_check(self, result: Dict[str, Any], scope: AccessScope) -> bool:
        metadata = result["metadata"]
        return (
            metadata.get("tenant_id") == scope.tenant_id
            and metadata.get("kb_id") in scope.allowed_kb_ids
            and metadata.get("status") == "published"
        )

    @staticmethod
    def _cosine(a: np.ndarray, b: np.ndarray) -> float:
        return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b) + 1e-8))


# 模拟认证中间件生成的权限上下文，实际项目中应来自 JWT、Session 或内部 IAM。
scope = AccessScope(
    tenant_id="acme",
    user_id="u_1001",
    role="member",
    allowed_kb_ids=["kb_handbook", "kb_product"],
    allowed_group_ids=["g_sales"],
)

store = InMemorySecureVectorStore()
store.upsert({
    "id": "doc_1_chunk_1",
    "tenant_id": "acme",
    "kb_id": "kb_handbook",
    "owner_id": "u_1001",
    "group_id": "g_sales",
    "visibility": "tenant",
    "status": "published",
    "vector": [0.1, 0.2, 0.7],
    "metadata": {
        "tenant_id": "acme",
        "kb_id": "kb_handbook",
        "status": "published",
        "text": "员工报销流程说明",
    },
})

query_vector = np.array([0.1, 0.2, 0.7])
print(store.search(query_vector, scope, top_k=3))

写入侧设计：权限字段要随切片一起落库

很多事故发生在查询侧，但根因在写入侧。文档切片入库时，应从文档归属表读取权限字段，而不是相信上传接口传入的参数。每条向量至少保存：

• tenant_id：租户唯一标识，建议使用不可猜测的 UUID。
• kb_id：知识库标识。
• owner_id / group_id：资源归属。
• visibility：private、group、tenant 等可见级别。
• status：processing、published、deleted。
• doc_version：文档更新版本，便于失效旧切片。

文档删除或权限变更时，不能只更新关系型数据库。需要通过异步任务同步删除向量、更新元数据，或把版本号加入查询过滤。否则用户已经失去权限，但旧向量仍可被相似度搜索召回。

查询侧设计：RAG 链路每一层都要保留 Scope

在完整 RAG 流程中，Query Rewrite、Hybrid Search、Rerank、上下文组装都可能复制或变换查询。权限 Scope 不应放在自然语言问题里，而应作为独立的结构化参数随调用链传递。

检索时建议使用“预取 + 二次鉴权 + 补足”的策略：向量库先在租户命名空间内按权限过滤取较大候选集，再由业务服务根据数据库实时权限做二次校验；如果过滤后数量不足，应继续翻页检索，而不是降低过滤条件。Rerank 模型只能处理已授权切片，不能先对全库片段重排再鉴权。

避坑总结：至少要盯住这 5 个高频事故

1. 空权限被翻译成全量查询：某些 SDK 对空 IN、空字符串 namespace 的处理是忽略过滤条件。现象是新用户或无权限用户反而能搜到全库内容。修正方法是在检索网关中显式判断 tenant_id、allowed_kb_ids 为空时直接拒绝请求，并为该行为补充单元测试。

2. 只在 Prompt 中要求模型不要泄密：提示词不是访问控制。越权文档一旦进入上下文，模型可能在摘要、翻译或问答中间接泄露。修正方法是在向量检索前强制注入租户命名空间和元数据过滤，Prompt 只负责表达任务，不承担安全边界。

3. 把用户 ID 直接拼进 Collection 名：早期为了“隔离更彻底”，使用 collection_tenant_user 方式建集合，结果集合数量暴涨，索引内存和连接数不可控，删除权限也难维护。修正方法是租户用 namespace 隔离，用户和组权限用元数据表达；只有少数企业版客户使用独立 Collection。

4. 权限变更后旧向量仍可召回：业务库中已经移除成员或删除文档，但向量库仍保留旧切片，典型现象是用户离职后仍能在搜索建议或问答中看到原文档。修正方法是权限变更事件同步触发向量删除或元数据更新，并在查询中加入 status、doc_version 条件。

5. 先 ANN 召回后过滤导致结果不稳定：向量库先取全局 TopK 再做租户过滤，高分结果大多属于其他租户时，当前租户结果可能为空。修正方法是优先使用支持预过滤或带过滤 ANN 索引的能力，并在租户命名空间内查询；必要时扩大候选集，但绝不能移除租户条件。

结语

多租户 SaaS 的向量库权限设计，本质上是把 IAM 的最小权限原则前移到检索引擎。推荐默认采用“租户命名空间 + 元数据细粒度过滤 + 应用层二次鉴权 + 审计日志”的组合，并通过统一检索客户端禁止裸查询。只有当权限边界在工程上成为不可绕过的默认路径，RAG 的智能化能力才不会变成跨租户数据泄露入口。

免责声明

本工具与文章内容仅用于学习测试场景，优化建议仅供参考，请勿直接用于生产环境，由此产生的风险自行承担。