Skip to content
🔗 分享本题
查看我的学习进度 →

25 模块 Q2 教学图:设计企业知识库 RAG 平台(多租户 + 权限隔离)

🧠 图解记忆:租户隔离和文档权限必须在检索前执行,答案引用也要可追溯到授权证据;点击图片可查看原图。

💡 答案要点

题目理解:

企业知识库 RAG 平台:
- 多租户:多个企业客户共享基础设施,数据隔离
- 权限隔离:租户内成员有不同权限(管理员/编辑/查看)
- 核心挑战:数据隔离 + 检索质量 + 成本控制

整体架构:

租户 A 的用户

┌─────────────────────────────────────────────────────┐
│                   API 网关                          │
│         认证 │ 限流 │ 租户路由 │ 权限校验             │
└─────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────┐
│              租户隔离层(Tenant Isolation)           │
│   租户 A 的数据 ──→ Tenant A Namespace / DB          │
│   租户 B 的数据 ──→ Tenant B Namespace / DB          │
└─────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────┐
│                  RAG 服务层                         │
│   Chunking │ Embedding │ 向量检索 │ 重排序           │
└─────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────┐
│         知识库存储层(向量 + 文档 + 图谱)           │
│   Pinecone/Milvus(向量) │ S3(原始文档)│ Neo4j(图谱)│
└─────────────────────────────────────────────────────┘

多租户隔离方案对比:

方案实现优点缺点适用场景
共享数据库+租户ID所有租户共用一个 DB,用 tenant_id 过滤成本低、易维护数据泄露风险、查询性能差小规模 (< 100 租户)
独立 Schema每个租户一个 schema隔离性好、查询快迁移复杂中等规模
独立数据库每个租户独立 MySQL/Redis完全隔离成本高、维护难大客户、高安全需求
Namespace 隔离向量数据库用 namespace 隔离实现简单、查询快依赖底层支持Pinecone/Weaviate
展开 Python 代码示例(31 行)
python
# 多租户向量检索实现
class TenantAwareVectorStore:
    def __init__(self, client, tenant_id: str):
        self.client = client
        self.tenant_id = tenant_id  # 租户 ID
    
    def search(self, query: str, top_k: int = 10) -> list[Document]:
        # 租户隔离:每个租户只检索自己的数据
        results = self.client.query(
            namespace=f"tenant_{self.tenant_id}",  # Namespace 隔离
            vector=self.embed(query),
            top_k=top_k,
            filter={"status": {"$eq": "active"}}  # 只检索激活的文档
        )
        
        # 后处理:权限过滤
        return self.apply_permissions(results)
    
    def apply_permissions(self, results: list[Document]) -> list[Document]:
        """根据用户权限过滤结果"""
        user_role = get_current_user_role()
        
        if user_role == "admin":
            return results  # 管理员看全部
        
        if user_role == "editor":
            # 编辑可看自己和公共文档
            return [d for d in results if d.is_public or d.owner_id == current_user_id]
        
        # 查看者只看公共文档
        return [d for d in results if d.is_public]

权限模型设计(RBAC + ABAC):

python
# 权限模型:角色 + 操作 + 资源
class Permission:
    READ = "read"
    WRITE = "write"
    DELETE = "delete"
    SHARE = "share"

class Role:
    ADMIN = "admin"       # 全部权限
    EDITOR = "editor"    # 读写
    VIEWER = "viewer"     # 只读

# 知识库权限矩阵
PERMISSION_MATRIX = {
    Role.ADMIN: {Permission.READ, Permission.WRITE, Permission.DELETE, Permission.SHARE},
    Role.EDITOR: {Permission.READ, Permission.WRITE},
    Role.VIEWER: {Permission.READ},
}

def check_permission(role: str, action: str, resource_tenant: str) -> bool:
    """权限校验"""
    if role not in PERMISSION_MATRIX:
        return False
    return action in PERMISSION_MATRIX[role]

企业级 RAG Pipeline:

文档上传

┌──────────────────────────────────────────┐
│  文档解析(PDF/Word/Markdown)            │
│  支持表格、图表、公式提取                   │
└──────────────────────────────────────────┘

┌──────────────────────────────────────────┐
│  语义 Chunking(不是固定大小!)           │
│  按句子/段落语义边界切分,保留上下文         │
└──────────────────────────────────────────┘

┌──────────────────────────────────────────┐
│  Embedding + 元数据标注                   │
│  - 向量化                                 │
│  - 租户 ID                                │
│  - 文档类型(政策/合同/手册)               │
│  - 密级(公开/内部/机密)                   │
└──────────────────────────────────────────┘

┌──────────────────────────────────────────┐
│  混合检索(向量 + 关键词)                  │
│  RRF 融合 + 权限过滤                       │
└──────────────────────────────────────────┘

答案生成

敏感信息过滤:

python
class SensitiveDataFilter:
    """敏感数据过滤(PII 检测)"""
    
    def filter(self, text: str) -> str:
        import re
        # 脱敏规则
        patterns = [
            (r"\d{11}", "【手机号】"),      # 手机号
            (r"\d{18}", "【身份证】"),      # 身份证
            (r"\b[\w.-]+@[\w.-]+\.\w+\b", "【邮箱】"),  # 邮箱
            (r"¥\d+", "【金额】"),            # 金额
        ]
        
        for pattern, replacement in patterns:
            text = re.sub(pattern, replacement, text)
        
        return text
    
    async def filter_document(self, doc: Document) -> Document:
        """过滤文档中的敏感信息"""
        doc.content = self.filter(doc.content)
        doc.chunks = [self.filter(chunk) for chunk in doc.chunks]
        return doc

成本估算(100 租户):

成本项计算月成本(万元)
向量存储100租户 × 100万 chunk × 768维 × $0.25/百万向量/月2.5
文档存储100租户 × 10GB × $0.02/GB20
Embedding API100租户 × 1000次/天 × 100 chunk × $0.000130
LLM 生成100租户 × 1000次/天 × 500 tokens × $0.002/1K150
合计~200/月

面试话术:

"企业知识库 RAG 的核心是'多租户隔离'。我用过三种隔离方案:数据库 schema 隔离(中等规模)、向量 namespace 隔离(大多数场景)、独立数据库(高安全大客户)。权限模型用 RBAC + ABAC 组合——RBAC 控制角色(管理员/编辑/查看),ABAC 控制资源(密级、部门)。PII 脱敏是生产必须的,上传时扫描、脱敏、存储三步走。面试能说清楚'namespace 隔离 + RBAC 权限 + 敏感信息过滤'三件套,说明你对企业级 RAG 有实战理解。"

📚 参考:Milvus:多租户向量检索实践