RAG 在企业知识库中的应用：权限隔离和搜索结果个性化-平芜编程栈

RAG 在企业知识库中的应用：权限隔离和搜索结果个性化

一、CEO 和实习生用同一个 AI 知识库搜索，查到的东西一模一样

企业知识库和公开知识库最大的区别是：不是所有信息对所有人都是可见的。CEO 搜索"Q3 营收目标"，应该看到最新的财务预测数据。普通员工搜索同样的词，应该看到"公司的核心价值观是客户第一"。但如果权限控制没做好，或者搜索结果没有根据身份做个性化，信息就有可能越权泄露或不对等展示。

传统的企业知识库搜索（ES + 关键词匹配）还能靠数据库的权限字段做过滤，但一旦引入 RAG（向量检索 + LLM 生成），权限问题就变得更复杂——因为 LLM 的生成结果是"新内容"，它可能把多条不同权限级别的文档内容糅合在一起输出。如何保证生成的答案只看得到"该用户有权查看的信息"，是企业 RAG 的核心工程挑战。

二、权限感知的企业 RAG 架构

核心设计是"检索前过滤 + 检索后校验"的双重保障：

这个架构有两道过滤：检索后过滤（硬性拦截，看不到的文档完全不出现在候选集里），字段级脱敏（软性保护，能看到文档但不能看到全部字段）。两道过滤的组合，相比单纯依赖数据库权限更加安全。

三、Python 实现：带权限隔离的 RAG 检索

import numpy as np from typing import List, Dict, Optional, Set, Any from dataclasses import dataclass from enum import Enum class DocPermission(Enum): """文档密级""" PUBLIC = "public" # 全员可见 INTERNAL = "internal" # 内部员工可见 CONFIDENTIAL = "confidential" # 特定团队可见 SECRET = "secret" # 仅高管可见 class UserRole(Enum): """用户角色""" EMPLOYEE = "employee" MANAGER = "manager" DIRECTOR = "director" EXECUTIVE = "executive" ADMIN = "admin" @dataclass class User: """用户信息""" user_id: str role: UserRole department: str team: str # 显式授权的文档 ID 列表 authorized_docs: Set[str] = None def __post_init__(self): if self.authorized_docs is None: self.authorized_docs = set() def permission_level(self) -> int: """返回数字权限等级（越大权限越高）""" role_levels = { UserRole.EMPLOYEE: 1, UserRole.MANAGER: 2, UserRole.DIRECTOR: 3, UserRole.EXECUTIVE: 4, UserRole.ADMIN: 5, } return role_levels.get(self.role, 0) @dataclass class EnterpriseDocument: """企业文档""" doc_id: str content: str title: str department: str # 所属部门 permission: DocPermission # 可见角色等级要求 min_role_level: int = 1 # 允许访问的团队列表（空 = 所有团队） allowed_teams: List[str] = None # 敏感字段列表（这些字段在低权限时需要脱敏） sensitive_fields: List[str] = None embedding: Optional[np.ndarray] = None class PermissionAwareRetriever: """带权限隔离的检索引擎""" def __init__(self): self.documents: Dict[str, EnterpriseDocument] = {} # 倒排索引: 用户ID -> 可见文档ID self.permission_cache: Dict[str, Set[str]] = {} def index_document(self, doc: EnterpriseDocument): """索引文档""" self.documents[doc.doc_id] = doc # 使权限缓存失效 self.permission_cache.clear() def get_visible_docs(self, user: User) -> Set[str]: """获取用户可见的文档 ID 集合""" # 缓存命中 if user.user_id in self.permission_cache: return self.permission_cache[user.user_id] visible = set() user_level = user.permission_level() for doc_id, doc in self.documents.items(): # 规则 1：文档密级检查 if doc.permission == DocPermission.PUBLIC: visible.add(doc_id) continue if doc.permission == DocPermission.INTERNAL: if user_level >= 1: # 所有正式员工 visible.add(doc_id) continue if doc.permission == DocPermission.CONFIDENTIAL: # 需要同部门 且 权限等级足够 if (user.department == doc.department and user_level >= doc.min_role_level): visible.add(doc_id) continue if doc.permission == DocPermission.SECRET: # 仅高管 或 显式授权 if (user.role == UserRole.EXECUTIVE or doc_id in user.authorized_docs): visible.add(doc_id) continue # 规则 2：团队白名单 if doc.allowed_teams: if user.team in doc.allowed_teams: visible.add(doc_id) self.permission_cache[user.user_id] = visible return visible def search( self, query: str, query_embedding: np.ndarray, user: User, top_k: int = 10, ) -> List[Dict[str, Any]]: """权限感知的向量检索""" # 1. 获取用户可见文档 visible_docs = self.get_visible_docs(user) if not visible_docs: return [] # 2. 只在可见文档中做向量检索 candidates = [] for doc_id in visible_docs: doc = self.documents.get(doc_id) if doc is None or doc.embedding is None: continue similarity = self._cosine_similarity( query_embedding, doc.embedding ) candidates.append((doc, similarity)) # 3. 排序取 Top-K candidates.sort(key=lambda x: x[1], reverse=True) top_docs = candidates[:top_k] # 4. 字段级脱敏 results = [] for doc, score in top_docs: content = self._apply_field_mask(doc, user) results.append({ 'doc_id': doc.doc_id, 'title': doc.title, 'content': content, 'score': score, 'department': doc.department, }) return results def _apply_field_mask( self, doc: EnterpriseDocument, user: User ) -> str: """对敏感字段做脱敏处理""" if not doc.sensitive_fields: return doc.content user_level = user.permission_level() content = doc.content for field in doc.sensitive_fields: if user_level >= 4: # 高管以上可见原始数据 continue elif user_level >= 3: # 总监级：区间化 import re pattern = re.compile( f'{field}[:：]\\s*([\\d,.]+)' ) content = pattern.sub( f'{field}: [数据脱敏-区间]', content ) else: # 其他级别：完全脱敏 import re pattern = re.compile( f'{field}[:：].*?(\\n|$)' ) content = pattern.sub( f'{field}: [数据未授权]\n', content ) return content def _cosine_similarity( self, a: np.ndarray, b: np.ndarray ) -> float: return float(np.dot(a, b) / ( np.linalg.norm(a) * np.linalg.norm(b) + 1e-8 )) def search_with_personalization( self, query: str, query_embedding: np.ndarray, user: User, top_k: int = 10, ) -> List[Dict[str, Any]]: """带个性化的搜索（权限 + 部门偏好）""" results = self.search( query, query_embedding, user, top_k * 2 ) # 个性化 boosting：同部门的文档加权 for result in results: if result['department'] == user.department: result['score'] *= 1.2 # 部门内文档提权 20% results.sort(key=lambda x: x['score'], reverse=True) return results[:top_k]

四、边界分析与 Trade-offs

权限缓存的失效策略：用户的权限、团队、授权文档可能随时变化（入职、离职、调岗、项目权限变更）。权限缓存需要设置合理的过期时间（建议 5 分钟），或者通过事件驱动（用户的角色变更时主动清除该用户的缓存）。

检索精度 vs 安全的权衡：如果用户只能看到 20% 的文档，那向量检索的候选集就小了 80%。极端情况下（新员工只能看公开文档），检索结果可能只有寥寥几条。解决方案：对于检索结果过少的情况（< 3 条），提示用户"未搜索到足够信息，建议细化关键词或申请文档权限"。

LLM 生成内容的安全审核：即使检索时做了权限过滤，LLM 仍可能在生成过程中"脑补"出敏感信息。生成后需要做关键词匹配检测——如果输出中包含"营收""毛利""工资"等敏感词，且不在引用文档中出现，说明是模型幻觉，应该拦截。这个审核可以用规则（关键词列表），不需要额外的 AI 模型。

跨文档聚合的安全隐患：用户可能没有权限看财务文档 A 和人事文档 B，但有权限看项目文档 C。如果项目文档 C 中引用了"预算 500 万"（这个信息来自 A），用户可能通过 C 间接获取到 A 的信息。这需要文档入库时做"引用链路"标注，检索时检测间接引用并降权。

五、总结

企业 RAG 的权限隔离不是"加个 if 判断"那么简单，需要检索前后双重过滤：检索阶段通过权限缓存过滤可见文档（硬拦截），检索后通过字段级脱敏保护敏感信息（软保护）。权限模型建议用"角色等级 + 文档密级 + 团队归属"的三维模型，而不是简单的 RBAC。缓存策略是性能和安全的关键平衡点——建议 5 分钟过期 + 事件驱动失效的组合方式。最容易被忽略的是"跨文档引用"导致的信息间接泄露，这在金融和法律等强合规行业是绝对的红线。