14.16-知识库权限控制

要点

知识库权限控制确保用户只能看到自己有权访问的文档——防止信息泄露
权限维度：租户隔离、文档级权限、字段级权限、知识库级权限
向量数据库的过滤查询是权限控制的基础——每次检索都必须带权限过滤
权限不能只在应用层做——必须在向量数据库查询时强制执行
常见陷阱：检索时忘记加权限过滤、缓存的向量结果没有权限隔离
权限变更后的数据同步：文档权限变化时，向量数据需要及时更新

1. 为什么需要权限控制

企业知识库场景，不同用户看到不同范围的内容：

公司知识库：
├── 全员可见：公司制度、福利政策、办公指南
├── 技术部可见：技术架构文档、API 文档、代码规范
├── 管理层可见：战略规划、财务报表、人事变动
└── 个人可见：个人绩效、薪资信息

如果权限控制不到位：

普通员工问：「公司明年的战略方向是什么？」
LLM 回答：根据战略规划文档，公司明年的重点是...

→ 这个回答泄露了管理层才能看到的信息

权限控制必须做——在检索阶段就过滤掉用户无权访问的文档。

2. 权限模型

租户隔离（多租户场景）

最基本的隔离——不同租户的数据完全分开。

// 写入时标记 tenant_id
await vectorDB.upsert({
  id: chunkId,
  vector: embedding,
  payload: {
    text: chunk.content,
    tenant_id: user.tenantId,  // 必须
    // ...
  },
})
 
// 查询时过滤 tenant_id
const results = await vectorDB.search(queryVector, {
  filter: { tenant_id: user.tenantId },  // 必须
})

租户隔离是最基本的安全要求——实现简单，但不能出错。

文档级权限

每个文档有一组可访问的用户或角色。

type DocumentPermissions = {
  documentId: string
  // 三种权限模型选一种
  accessList: string[]           // 允许访问的用户 ID 列表
  roleList: string[]             // 允许访问的角色列表
  visibility: 'public' | 'internal' | 'private'  // 可见性
}

写入时把权限信息存进 payload：

await vectorDB.upsert({
  id: chunkId,
  vector: embedding,
  payload: {
    text: chunk.content,
    tenant_id: doc.tenantId,
    document_id: doc.id,
    // 权限字段
    access_user_ids: doc.accessList,     // Qdrant payload
    access_roles: doc.roleList,
    visibility: doc.visibility,
  },
})

查询时构造权限过滤：

function buildPermissionFilter(user: User): Filter {
  const conditions: FilterCondition[] = [
    // 租户隔离
    { key: 'tenant_id', match: { value: user.tenantId } },
  ]
 
  // 文档权限：用户能看到的文档
  // 规则：visibility=public 的都能看
  //        或者 access_user_ids 包含当前用户
  //        或者 access_roles 包含用户的任一角色
  conditions.push({
    should: [
      { key: 'visibility', match: { value: 'public' } },
      { key: 'access_user_ids', match: { any: [user.id] } },
      // 用户的角色列表
      ...user.roles.map((role) => ({
        key: 'access_roles',
        match: { value: role },
      })),
    ],
  })
 
  return { must: conditions }
}
 
// 查询
const results = await vectorDB.search(queryVector, {
  filter: buildPermissionFilter(currentUser),
})

知识库级权限

用户有权限访问某些知识库，但不是其他知识库。

// 用户权限
type UserPermissions = {
  userId: string
  tenantId: string
  accessibleKnowledgeBases: string[]  // 可访问的知识库 ID
  roles: string[]
}
 
// 写入时标记知识库 ID
await vectorDB.upsert({
  id: chunkId,
  vector: embedding,
  payload: {
    text: chunk.content,
    tenant_id: doc.tenantId,
    knowledge_base_id: doc.knowledgeBaseId,
  },
})
 
// 查询时过滤
const results = await vectorDB.search(queryVector, {
  filter: {
    must: [
      { key: 'tenant_id', match: { value: user.tenantId } },
      { key: 'knowledge_base_id', match: { any: user.accessibleKnowledgeBases } },
    ],
  },
})

3. 权限强制执行

权限控制的关键原则：权限检查必须在向量数据库查询时执行，不能只在应用层做。

❌ 错误方式：应用层过滤

// 先检索，再在代码里过滤
const results = await vectorDB.search(queryVector, { topK: 20 })
const filtered = results.filter((r) => hasPermission(user, r.metadata.documentId))
return filtered.slice(0, 5)

问题：

检索了 20 条结果，过滤后可能只剩 2 条——质量差
如果前 20 条都没有权限，返回空——但其实有权限的文档在后面
性能差——检索了不需要的数据

✅ 正确方式：向量数据库层过滤

// 在向量数据库查询时就加权限过滤
const results = await vectorDB.search(queryVector, {
  topK: 5,
  filter: buildPermissionFilter(user),
})

向量数据库只返回用户有权访问的文档——结果数量准确，性能也更好。

中间件统一注入

在 Hono 中间件里统一注入权限过滤，避免每个检索调用都忘记加过滤：

// src/middleware/rag-permission.ts
import { createMiddleware } from 'hono/factory'
 
declare module 'hono' {
  interface ContextVariableMap {
    ragFilter: Filter
  }
}
 
export const ragPermissionMiddleware = createMiddleware(async (c, next) => {
  const user = c.get('user')
 
  if (!user) {
    return c.json({ error: 'Unauthorized' }, 401)
  }
 
  // 构建权限过滤
  const filter = buildPermissionFilter(user)
  c.set('ragFilter', filter)
 
  await next()
})
 
// 使用时
app.post('/api/rag/search', ragPermissionMiddleware, async (c) => {
  const filter = c.get('ragFilter')  // 自动注入权限过滤
 
  const results = await vectorDB.search(queryVector, {
    topK: 5,
    filter,  // 不需要手动构建
  })
 
  return c.json(results)
})

4. 权限变更同步

文档权限变化时，向量数据需要同步更新。

场景

初始状态：文档 A 只有 alice 能看
权限变更：文档 A 改为全员可见
→ 向量数据里 document A 的 access_user_ids 需要从 [alice] 变成 public

方案一：更新 payload

async function updateDocumentPermissions(
  documentId: string,
  newPermissions: DocumentPermissions
): Promise<void> {
  // 找到所有相关 chunk
  const chunks = await vectorDB.scroll({
    filter: { document_id: documentId },
  })
 
  // 逐个更新 payload
  for (const chunk of chunks) {
    await vectorDB.updatePayload(chunk.id, {
      access_user_ids: newPermissions.accessList,
      access_roles: newPermissions.roleList,
      visibility: newPermissions.visibility,
    })
  }
}

Qdrant 支持批量更新 payload，性能更好：

await client.overwritePayload('documents', {
  filter: { must: [{ key: 'document_id', match: { value: documentId } }] },
  payload: {
    access_user_ids: newPermissions.accessList,
    visibility: newPermissions.visibility,
  },
})

方案二：删除重建

如果权限变化频繁，更新 payload 的性能可能不够好。可以删掉旧的 chunk，重新生成。

async function rebuildDocument(documentId: string): Promise<void> {
  // 1. 删除旧的 chunk
  await vectorDB.deleteByFilter({
    document_id: documentId,
  })
 
  // 2. 重新解析、切分、嵌入、写入
  const doc = await getDocumentFromDB(documentId)
  const chunks = chunkDocument(doc.content)
  const embeddings = await embedChunks(chunks)
 
  await vectorDB.upsert(
    chunks.map((chunk, i) => ({
      id: `${documentId}-chunk-${i}`,
      vector: embeddings[i],
      payload: {
        text: chunk.text,
        document_id: documentId,
        ...buildPermissionPayload(doc.permissions),
      },
    }))
  )
}

方案三：权限和向量分离

不把权限存在向量数据库里，而是存在关系数据库里，查询时动态拼接。

async function searchWithPermissions(
  queryVector: number[],
  user: User
): Promise<SearchResult[]> {
  // 1. 从关系数据库查出用户能访问的文档 ID
  const accessibleDocIds = await getAccessibleDocumentIds(user)
 
  // 2. 用这些 ID 作为向量数据库的过滤条件
  return vectorDB.search(queryVector, {
    filter: {
      document_id: { $in: accessibleDocIds },
    },
  })
}

优点：权限变更不需要更新向量数据库。缺点：如果用户能访问的文档很多（比如几千个），过滤条件会很长。

5. 缓存和权限

检索结果的缓存必须考虑权限隔离。

❌ 错误方式：共享缓存

// 按查询文本做 key——不同用户共享同一份缓存
const cacheKey = `rag:${hash(queryText)}`
const cached = await redis.get(cacheKey)
if (cached) return cached

问题：alice 的检索结果被 bob 拿到了——bob 可能看到了不该看的内容。

✅ 正确方式：用户级缓存

// 按用户 + 查询做 key
const cacheKey = `rag:${user.id}:${hash(queryText)}`

或者更细粒度：

// 按权限组合做 key——相同权限的用户共享缓存
const permissionKey = hash(JSON.stringify(user.accessibleKnowledgeBases.sort()))
const cacheKey = `rag:${permissionKey}:${hash(queryText)}`

6. 权限审计

记录每次检索的权限过滤条件——方便事后审计。

async function searchWithAudit(
  queryVector: number[],
  user: User,
  filter: Filter
): Promise<SearchResult[]> {
  const startTime = Date.now()
 
  try {
    const results = await vectorDB.search(queryVector, {
      topK: 5,
      filter,
    })
 
    // 审计日志
    await auditLog.record({
      event: 'rag_search',
      userId: user.id,
      tenantId: user.tenantId,
      filterApplied: filter,
      resultCount: results.length,
      latency: Date.now() - startTime,
    })
 
    return results
  } catch (err) {
    await auditLog.record({
      event: 'rag_search_error',
      userId: user.id,
      error: String(err),
    })
    throw err
  }
}

总结

回顾这一节的要点：

权限控制确保用户只能看到自己有权访问的文档
权限维度：租户隔离（最基本）、文档级、字段级、知识库级
权限过滤必须在向量数据库查询时执行——不能在应用层后过滤
Hono 中间件统一注入权限过滤，避免遗漏
权限变更需要同步向量数据——更新 payload、删除重建、或权限分离存储
缓存必须考虑权限隔离——按用户或权限组合做 key
审计日志记录每次检索的权限过滤——方便安全审计

下一篇讲 RAG 评估指标——怎么衡量 RAG 系统的检索质量和生成质量。