如何利用向量数据库的访问控制机制保护RAG中的敏感文档

导语

检索增强生成（RAG）系统是构建现代知识型AI应用的核心范式。然而，在企业级部署中，安全性是一个巨大的挑战。如果将敏感和非敏感文档混合存储在同一个向量索引中，没有适当的访问控制，RAG系统可能向未授权用户泄露机密信息。本文将深入探讨如何通过向量数据库（Vector Database）的元数据过滤（Metadata Filtering）功能，实现文档级别的访问控制（ACL）。

RAG安全挑战：统一索引的风险

传统的RAG流程通常是：
1. 将所有企业文档切块（Chunking）并向量化。
2. 将所有向量及其源文本存储在一个公共的向量索引中。
3. 用户查询时，系统检索Top-K个最相关的文档，并输入给LLM。

这种“全部存储，统一检索”的模式，忽略了用户的身份和权限。如果用户A（普通员工）查询的内容，命中了只有用户B（高管）才能查看的敏感文档，RAG系统就会错误地返回这些机密信息。

解决方案的核心在于：在向量检索阶段，基于用户的身份信息，限制可被检索到的文档范围。

基于元数据过滤的访问控制实现

大多数企业级向量数据库（如Pinecone, Weaviate, Qdrant, Milvus）都支持强大的元数据过滤功能。我们可以利用这一功能，将文档的权限信息作为元数据嵌入到每个向量中。

步骤一：定义权限元数据结构

首先，定义一套清晰的权限标签。这些标签在文档摄取（Ingestion）阶段附加到每个数据块上。

步骤二：数据摄取与权限标记

当我们将文档切块并向量化时，必须确保每个向量的数据点都携带正确的权限元数据。

以下是使用Python模拟数据摄取过程的示例：

import uuid

# 假设的向量数据库客户端接口
class MockVectorDBClient:
    def __init__(self):
        self.index = {}

    def upsert(self, vectors_with_metadata):
        for vector_data in vectors_with_metadata:
            # 实际操作中，key是向量ID，value是(vector, metadata)
            self.index[vector_data['id']] = (
                vector_data['vector'], 
                vector_data['metadata']
            )
        print(f"成功摄取 {len(vectors_with_metadata)} 个带权限元数据的文档块")

# 敏感文档块 A
chunk_a = {
    'id': str(uuid.uuid4()),
    'vector': [0.1, 0.2, ...],
    'metadata': {
        'text': '2024年Q3财务预测数据。',
        'security_level': 'confidential',
        'owner_group': 'finance',
        'department': 'finance'
    }
}

# 非敏感文档块 B
chunk_b = {
    'id': str(uuid.uuid4()),
    'vector': [0.5, 0.6, ...],
    'metadata': {
        'text': '公司食堂每周菜单。',
        'security_level': 'public',
        'owner_group': 'all',
        'department': 'hr'
    }
}

db = MockVectorDBClient()
db.upsert([chunk_a, chunk_b])

步骤三：受限检索（Filtering Query）

当用户发起查询时，RAG系统需要首先从认证服务（Auth Service）获取当前用户的身份和权限信息。然后，这些信息被转换成一个布尔过滤器（Boolean Filter），附加到向量搜索请求中。

场景示例：

1. 用户 A (普通员工) 身份：{user_id: 201, groups: [‘engineering’]}
2. 用户 B (财务主管) 身份：{user_id: 302, groups: [‘finance’, ‘management’]}

代码示例：根据用户身份执行过滤检索

# 模拟向量数据库的查询函数
def query_with_filter(query_vector, user_context):
    # 1. 构建查询过滤器
    user_groups = user_context['groups']

    # 允许访问的条件：
    # (security_level == 'public') OR (owner_group IN user_groups)

    metadata_filter = {
        '$or': [
            {'security_level': {'$eq': 'public'}},
            {'owner_group': {'$in': user_groups}}
        ]
    }

    print(f"\n--- 正在为用户 {user_context['user_id']} 执行检索 ---")
    print(f"应用的元数据过滤器: {metadata_filter}")

    # 2. 模拟向量搜索并应用过滤（实际向量数据库会执行高效的过滤）
    retrieved_chunks = []
    for vec_id, (vec, meta) in db.index.items():
        # 简化判断：只检查元数据是否满足过滤器逻辑
        if meta['security_level'] == 'public':
            retrieved_chunks.append(meta['text'])
        elif meta['owner_group'] in user_groups:
            retrieved_chunks.append(meta['text'])

    return retrieved_chunks

# 模拟查询向量 (假设查询向量与两个文档块都相关)
mock_query_vector = [0.3, 0.4, ...]

# 检索 1: 普通员工查询
user_a_context = {'user_id': 201, 'groups': ['engineering']}
results_a = query_with_filter(mock_query_vector, user_a_context)
print(f"用户 A 检索结果 (只有公共文档): {results_a}")

# 检索 2: 财务主管查询
user_b_context = {'user_id': 302, 'groups': ['finance', 'management']}
results_b = query_with_filter(mock_query_vector, user_b_context)
print(f"用户 B 检索结果 (包含财务机密文档): {results_b}")

预期输出分析

用户 A (工程组) 的过滤器会排除掉 owner_group: ‘finance’ 的文档块 A，因此只会检索到公共文档块 B。
用户 B (财务组) 的过滤器满足 owner_group: ‘finance’ 的条件，因此可以检索到文档块 A 和 B。

最佳实践与考量

1. 权限抽象层： 应该在RAG服务和向量数据库之间设置一个权限抽象层。这个层负责将用户Session/Token解析成标准的元数据过滤查询语言（如MongoDB风格的JSON查询）。
2. 性能影响： 虽然元数据过滤比纯暴力搜索慢，但现代向量数据库对精确元数据过滤进行了高度优化（通常通过二级索引实现）。确保你的元数据字段被正确索引，以保持低延迟。
3. 多租户隔离： 对于极高安全要求的场景，除了元数据过滤，还可以采用命名空间（Namespace）或集合（Collection）隔离。将不同安全级别的文档放在物理隔离的索引中，进一步降低数据泄露的风险。RAG服务根据用户权限，选择查询不同的索引。
4. 及时同步： 确保用户权限系统（如LDAP/OAuth）和向量数据库中的权限元数据保持一致性。权限变更必须立即同步到RAG索引中，否则可能导致权限滞后（Stale Permissions）。