RAG 权限过滤实现：在检索阶段就把用户无权访问的文档排除出去

RAG 权限过滤实现在检索阶段就把用户无权访问的文档排除出去一、深度引言与场景痛点大家好我是赵咕咕。你的 RAG 系统上线了内部知识库功能效果不错——每个人都能搜到公司的所有文档。但有一天一位实习生搜到了高管的薪酬讨论记录。虽然他只是碰巧搜到的但这个漏洞足以让你被安全部门约谈。这就是 RAG 系统中最严重的问题之一权限泄露。向量检索只看语义相似度完全不关心这个用户是否有权看这份文档。如果在检索后再做权限过滤会有两个问题一是已经泄露了文档标题或摘要给用户二是浪费了检索预算召回的文档因权限被过滤真正有权限的文档却没被召回。正确做法是在检索阶段就把用户无权访问的文档排除在索引之外。这篇文章我们来实现生产级的 RAG 权限过滤系统。二、底层机制与原理深度剖析RAG 权限过滤有两条主要技术路线路线一预过滤Pre-filtering。在向量检索之前先用权限条件缩小搜索范围。比如WHERE access_level user_clearance AND (is_public OR user_id IN allowed_users)。这要求向量数据库支持结构化过滤好处是彻底杜绝权限泄露。路线二分区隔离Partition-based。按权限级别或用户组建立独立的索引分区。用户查询时只在他有权限的分区中检索。好处是性能好坯处是数据冗余同一文档可能在多个分区中。最佳实践是两种结合flowchart TB A[用户请求br/user_id query] -- B[权限解析器br/解析用户权限组] B -- C{权限分区路由} C -- D1[Public 分区br/所有人可访问] C -- D2[Team 分区br/团队级别] C -- D3[Private 分区br/个人文档] D1 -- E[权限预过滤br/结构化条件] D2 -- E D3 -- E E -- F[向量语义检索br/ANN Search] F -- G[后过滤br/二次权限校验] G -- H[返回 Top-Kbr/安全的结果] I[(权限元数据br/ACL Store)] -.- B J[(文档权限br/Document ACL)] -.- E style B fill:#fff3e0 style E fill:#e8f5e9 style G fill:#f3e5f5核心设计原则默认拒绝用户只能看到显式授权的文档纵深防御预过滤 后过滤两道关卡任一不通过即拦截最小检索只在用户有权限的分区中检索不浪费计算资源三、生产级代码实现下面是完整的 RAG 权限过滤系统实现from __future__ import annotations import asyncio from dataclasses import dataclass, field from typing import Optional from enum import IntEnum import hashlib import time class PermissionLevel(IntEnum): 权限等级数值越大权限越高 NONE 0 # 无权限 READER 10 # 只读 CONTRIBUTOR 20 # 可编辑 ADMIN 30 # 管理员 OWNER 40 # 文档所有者 dataclass class AccessPolicy: 文档访问策略 is_public: bool False owner_id: str allowed_users: list[str] field(default_factorylist) allowed_roles: list[str] field(default_factorylist) allowed_teams: list[str] field(default_factorylist) min_permission: PermissionLevel PermissionLevel.READER expires_at: Optional[float] None # 访问有效期 def can_access(self, user: User) - bool: 判断用户能否访问 # 1. 公开文档 if self.is_public: return True # 2. 所有者 if user.user_id self.owner_id: return True # 3. 用户白名单 if user.user_id in self.allowed_users: return True # 4. 角色匹配 if any(r in self.allowed_roles for r in user.roles): return True # 5. 团队匹配 if any(t in self.allowed_teams for t in user.teams): return True # 6. 过期检查 if self.expires_at and time.time() self.expires_at: return False return False dataclass class User: 用户身份 user_id: str roles: list[str] field(default_factorylist) teams: list[str] field(default_factorylist) permission_level: PermissionLevel PermissionLevel.READER dataclass class SecuredDocument: 带权限的安全文档 doc_id: str content: str embedding: Optional[list[float]] None policy: AccessPolicy field(default_factoryAccessPolicy) metadata: dict field(default_factorydict) property def content_hash(self) - str: 文档指纹 return hashlib.sha256( (self.content str(self.policy)).encode() ).hexdigest()[:16] class PermissionAwareRetriever: 权限感知的检索引擎 def __init__(self, vector_store, embedding_model): vector_store: 支持 filter 的向量数据库 embedding_model: 文本 Embedding 模型 self._store vector_store self._embedder embedding_model self._policy_cache: dict[str, AccessPolicy] {} async def index_document( self, doc: SecuredDocument ) - bool: 索引文档附带权限元数据 # 计算向量 if doc.embedding is None: doc.embedding await self._embedder.embed(doc.content) # 构建权限过滤标签 policy_tags self._policy_to_tags(doc.policy) try: await self._store.add( iddoc.doc_id, vectordoc.embedding, metadata{ content: doc.content, owner_id: doc.policy.owner_id, is_public: doc.policy.is_public, min_permission: doc.policy.min_permission.value, tags: policy_tags, chunk_hash: doc.content_hash, **doc.metadata, } ) return True except Exception as e: print(f索引文档 {doc.doc_id} 失败: {e}) return False def _policy_to_tags( self, policy: AccessPolicy ) - list[str]: 将访问策略转为标签列表 tags [] if policy.is_public: tags.append(scope:public) else: tags.append(scope:restricted) for user_id in policy.allowed_users: tags.append(fuser:{user_id}) for role in policy.allowed_roles: tags.append(frole:{role}) for team in policy.allowed_teams: tags.append(fteam:{team}) return tags async def search( self, query: str, user: User, top_k: int 10, ) - list[dict]: 权限感知的检索 # 1. 构建权限过滤条件 filter_condition self._build_filter(user) # 2. 向量化 query query_vector await self._embedder.embed(query) # 3. 检索预过滤已在 filter_condition 中 try: results await self._store.search( vectorquery_vector, filterfilter_condition, limittop_k * 2, # 多召回一些用于后过滤 ) except Exception as e: print(f检索失败: {e}) return [] # 4. 后过滤二次校验 safe_results [] for result in results: meta result.get(metadata, {}) # 二次权限校验 if not self._post_filter_check(user, meta): continue safe_results.append({ doc_id: result.get(id, ), content: meta.get(content, ), score: result.get(score, 0.0), }) if len(safe_results) top_k: break return safe_results def _build_filter(self, user: User) - dict: 构建结构化过滤条件 conditions [] # 公开文档 conditions.append({is_public: True}) # 所有者文档 conditions.append({owner_id: user.user_id}) # 用户白名单 if user.user_id: conditions.append({ tags: {$in: [fuser:{user.user_id}]} }) # 角色匹配 for role in user.roles: conditions.append({ tags: {$in: [frole:{role}]} }) # 团队匹配 for team in user.teams: conditions.append({ tags: {$in: [fteam:{team}]} }) # 权限等级过滤 conditions.append({ min_permission: {$lte: user.permission_level.value} }) return {$or: conditions} def _post_filter_check( self, user: User, metadata: dict ) - bool: 后过滤二次校验 # 公开文档直接通过 if metadata.get(is_public): return True # 所有者直接通过 if metadata.get(owner_id) user.user_id: return True # 权限等级检查 required_level metadata.get(min_permission, 0) if user.permission_level.value required_level: return False return True async def update_policy( self, doc_id: str, new_policy: AccessPolicy ) - bool: 更新文档权限策略触发重新索引 self._policy_cache[doc_id] new_policy # 实际生产环境需要更新向量数据库中的元数据 return True async def revoke_access( self, doc_id: str, user_id: str ) - bool: 撤回用户对某个文档的访问权限 # 从白名单中移除 policy self._policy_cache.get(doc_id) if policy and user_id in policy.allowed_users: policy.allowed_users.remove(user_id) return await self.update_policy(doc_id, policy) return False # 使用示例 async def main(): retriever PermissionAwareRetriever( vector_storeNone, # 替换为真实向量数据库 embedding_modelNone, # 替换为真实 Embedding 模型 ) # 创建文档 public_doc SecuredDocument( doc_iddoc_public, content公司假期政策每年年假15天..., policyAccessPolicy(is_publicTrue), ) private_doc SecuredDocument( doc_iddoc_private, content高管薪酬讨论CEO 年薪..., policyAccessPolicy( is_publicFalse, owner_idceo_user, allowed_roles[executive], ), ) # 创建用户 regular_user User( user_iduser_001, roles[employee], teams[engineering], ) exec_user User( user_idceo_user, roles[executive], teams[management], ) # 权限检查 print( 权限过滤测试 \n) print([普通员工] 查看公开文档:) print(f 可以访问: {public_doc.policy.can_access(regular_user)}) print([普通员工] 查看高管文档:) print(f 可以访问: {private_doc.policy.can_access(regular_user)}) print([高管] 查看高管文档:) print(f 可以访问: {private_doc.policy.can_access(exec_user)}) print(\n权限过滤系统已就绪) asyncio.run(main())四、边界分析与架构权衡权限过滤有几个容易被忽视的工程边界检索性能的下降。预过滤条件越多向量检索前的过滤时间越长。如果$or条件包含 10 个角色 20 个团队查询解析器需要处理复杂的布尔逻辑。优化方案对高频权限组合做物化视图如engineering 团队所有人可访问的文档预计算集合。权限变更的时效性。如果用户被移除了某个团队缓存中的权限信息没有立即更新用户可能在短时间窗口内仍能访问已无权查看的文档。必须设置权限缓存的 TTL并通过权限变更事件主动失效缓存。数据冗余 vs 检索效率。分区隔离方案中同一份文档在多个分区中重复存储造成索引膨胀。如果一份文档对 10 个团队可见就需要 10 份副本。文档量大时这种冗余不可接受。此时应退回到预过滤方案。复杂权限模型的表达。现实中的权限系统可能包含 RBAC、ABAC、甚至是基于属性的动态权限。简单的标签匹配无法满足。建议使用 OPAOpen Policy Agent或类似策略引擎在检索层通过 gRPC 调用做权限裁决。chunk 级别的权限粒度。文档权限与 chunk 权限可能不同——用户可能有权看文档的摘要但没有权看详细内容。需要在 chunk embedding 时也附加权限标签这进一步增加了元数据管理的复杂度。五、总结RAG 的权限过滤不是要不要做的问题而是怎么做得又快又安全的问题。核心要点在向量检索前做权限预过滤从根源上杜绝泄露用 公开文档 所有者 白名单 角色/团队 的多层权限模型检索后做二次校验作为纵深防御权限变更时主动失效缓存不给攻击留下时间窗口安全不是功能是基础。权限过滤不是你 RAG 系统的附加功能而是生命线。