DeepSeek-V4 企业知识库：权限管理完整设计方案

DeepSeek-V4 企业知识库：权限管理完整设计方案

一、场景概述与痛点分析

1.1 企业知识库的权限挑战

某大型制造企业，拥有5000+份技术文档、合同、操作手册，分布在不同的部门（研发、生产、销售、法务、售后）。痛点包括：

二、五层权限控制架构

#mermaid-svg-EWCHp8nMooynqGss{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-EWCHp8nMooynqGss .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-EWCHp8nMooynqGss .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-EWCHp8nMooynqGss .error-icon{fill:#552222;}#mermaid-svg-EWCHp8nMooynqGss .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-EWCHp8nMooynqGss .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-EWCHp8nMooynqGss .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-EWCHp8nMooynqGss .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-EWCHp8nMooynqGss .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-EWCHp8nMooynqGss .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-EWCHp8nMooynqGss .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-EWCHp8nMooynqGss .marker{fill:#333333;stroke:#333333;}#mermaid-svg-EWCHp8nMooynqGss .marker.cross{stroke:#333333;}#mermaid-svg-EWCHp8nMooynqGss svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-EWCHp8nMooynqGss p{margin:0;}#mermaid-svg-EWCHp8nMooynqGss .label{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;color:#333;}#mermaid-svg-EWCHp8nMooynqGss .cluster-label text{fill:#333;}#mermaid-svg-EWCHp8nMooynqGss .cluster-label span{color:#333;}#mermaid-svg-EWCHp8nMooynqGss .cluster-label span p{background-color:transparent;}#mermaid-svg-EWCHp8nMooynqGss .label text,#mermaid-svg-EWCHp8nMooynqGss span{fill:#333;color:#333;}#mermaid-svg-EWCHp8nMooynqGss .node rect,#mermaid-svg-EWCHp8nMooynqGss .node circle,#mermaid-svg-EWCHp8nMooynqGss .node ellipse,#mermaid-svg-EWCHp8nMooynqGss .node polygon,#mermaid-svg-EWCHp8nMooynqGss .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-EWCHp8nMooynqGss .rough-node .label text,#mermaid-svg-EWCHp8nMooynqGss .node .label text,#mermaid-svg-EWCHp8nMooynqGss .image-shape .label,#mermaid-svg-EWCHp8nMooynqGss .icon-shape .label{text-anchor:middle;}#mermaid-svg-EWCHp8nMooynqGss .node .katex path{fill:#000;stroke:#000;stroke-width:1px;}#mermaid-svg-EWCHp8nMooynqGss .rough-node .label,#mermaid-svg-EWCHp8nMooynqGss .node .label,#mermaid-svg-EWCHp8nMooynqGss .image-shape .label,#mermaid-svg-EWCHp8nMooynqGss .icon-shape .label{text-align:center;}#mermaid-svg-EWCHp8nMooynqGss .node.clickable{cursor:pointer;}#mermaid-svg-EWCHp8nMooynqGss .root .anchor path{fill:#333333!important;stroke-width:0;stroke:#333333;}#mermaid-svg-EWCHp8nMooynqGss .arrowheadPath{fill:#333333;}#mermaid-svg-EWCHp8nMooynqGss .edgePath .path{stroke:#333333;stroke-width:2.0px;}#mermaid-svg-EWCHp8nMooynqGss .flowchart-link{stroke:#333333;fill:none;}#mermaid-svg-EWCHp8nMooynqGss .edgeLabel{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-EWCHp8nMooynqGss .edgeLabel p{background-color:rgba(232,232,232, 0.8);}#mermaid-svg-EWCHp8nMooynqGss .edgeLabel rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-EWCHp8nMooynqGss .labelBkg{background-color:rgba(232, 232, 232, 0.5);}#mermaid-svg-EWCHp8nMooynqGss .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#mermaid-svg-EWCHp8nMooynqGss .cluster text{fill:#333;}#mermaid-svg-EWCHp8nMooynqGss .cluster span{color:#333;}#mermaid-svg-EWCHp8nMooynqGss div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#mermaid-svg-EWCHp8nMooynqGss .flowchartTitleText{text-anchor:middle;font-size:18px;fill:#333;}#mermaid-svg-EWCHp8nMooynqGss rect.text{fill:none;stroke-width:0;}#mermaid-svg-EWCHp8nMooynqGss .icon-shape,#mermaid-svg-EWCHp8nMooynqGss .image-shape{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-EWCHp8nMooynqGss .icon-shape p,#mermaid-svg-EWCHp8nMooynqGss .image-shape p{background-color:rgba(232,232,232, 0.8);padding:2px;}#mermaid-svg-EWCHp8nMooynqGss .icon-shape .label rect,#mermaid-svg-EWCHp8nMooynqGss .image-shape .label rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-EWCHp8nMooynqGss .label-icon{display:inline-block;height:1em;overflow:visible;vertical-align:-0.125em;}#mermaid-svg-EWCHp8nMooynqGss .node .label-icon path{fill:currentColor;stroke:revert;stroke-width:revert;}#mermaid-svg-EWCHp8nMooynqGss :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

审计层

文档层

权限控制层

身份认证层

用户层

员工账号

部门主管

外部合作方

离职账号池

SSO单点登录

LDAP/AD同步

MFA多因素认证

IP白名单

RBAC角色权限

ABAC属性权限

数据列级权限

行级安全策略

部门A文档

部门B文档

机密文档

公开文档

操作日志

敏感词告警

异常访问阻断

三、核心权限模型设计

3.1 四维权限矩阵

3.2 RBAC角色定义

# roles.yaml
roles:
– name: "anonymous"
description: "未登录用户(仅限公开文档)"
permissions: ["search:public", "view:public"]

– name: "employee"
description: "普通员工"
inherits: ["anonymous"]
permissions:
– "search:department" # 可搜索本部门文档
– "view:department" # 可查看本部门文档
– "download:department" # 可下载本部门文档
– "ask:public" # 可对公开文档提问
constraints:
max_results: 20
rate_limit: "100/hour"

– name: "manager"
description: "部门主管"
inherits: ["employee"]
permissions:
– "search:sub_department" # 可搜索下级部门
– "edit:department" # 可编辑部门文档
– "grant:department" # 可授权部门内权限
– "audit:department_log" # 可审计部门日志

– name: "compliance"
description: "合规审计员"
permissions:
– "search:all" # 全库搜索(带脱敏)
– "audit:all_logs" # 查看所有日志
– "alert:high_risk" # 接收高风险告警
constraints:
require_approval: true # 敏感操作需二次审批
log_retention: "7years"

– name: "external_partner"
description: "外部合作方"
permissions:
– "view:project_X_docs" # 仅限特定项目
– "ask:project_X" # 仅问答特定范围
constraints:
ip_whitelist: ["123.45.67.0/24"]
expire_at: "2027-01-01"
mfa_required: true

3.3 ABAC属性规则（动态权限）

# abac_rules.yaml
policies:
# 规则1：文档所有者可编辑
– name: "owner_edit"
effect: "allow"
subject:
attribute: "user.id"
operator: "equals"
target: "document.owner_id"
action: ["edit", "delete", "share"]

# 规则2：同部门可查看
– name: "same_dept_view"
effect: "allow"
subject:
attribute: "user.department"
operator: "equals"
target: "document.department"
action: ["view", "search"]

# 规则3：仅工作日工作时间访问机密文档
– name: "work_hours_confidential"
effect: "allow"
subject:
attribute: "user.role"
operator: "in"
values: ["manager", "compliance"]
environment:
day_of_week: ["monday", "tuesday", "wednesday", "thursday", "friday"]
hour_range: [9, 18]
action: ["view", "download"]

# 规则4：禁止离职员工访问（即便账号未禁用）
– name: "block_terminated"
effect: "deny"
subject:
attribute: "user.status"
operator: "equals"
value: "terminated"
action: ["*"]

# 规则5：外部IP下载需审批
– name: "external_download_approval"
effect: "deny_unless_approval"
subject:
attribute: "user.ip"
operator: "not_in_network"
value: "corporate_lan"
action: ["download_batch"]
approval_required: "manager"

四、数据库设计与索引策略

4.1 核心表结构

— 1. 用户表
CREATE TABLE users (
id BIGINT PRIMARY KEY,
username VARCHAR(64) NOT NULL UNIQUE,
email VARCHAR(128),
department_id INT REFERENCES departments(id),
role VARCHAR(32) NOT NULL,
status VARCHAR(16) DEFAULT 'active', — active/suspended/terminated
mfa_enabled BOOLEAN DEFAULT false,
created_at TIMESTAMP DEFAULT NOW(),
last_login TIMESTAMP,
terminated_at TIMESTAMP
);

— 2. 文档表（带权限属性）
CREATE TABLE documents (
id BIGINT PRIMARY KEY,
title VARCHAR(256) NOT NULL,
content TEXT,
embedding vector(1536), — pgvector
owner_id BIGINT REFERENCES users(id),
department_id INT REFERENCES departments(id),
classification VARCHAR(32), — public/internal/confidential/secret
project_tags TEXT[], — ['Project_X', 'R_D']
allow_departments INT[], — 允许访问的部门列表
allow_roles VARCHAR(32)[], — 允许访问的角色列表
allow_users BIGINT[], — 允许访问的用户列表
deny_users BIGINT[], — 黑名单用户
created_at TIMESTAMP,
updated_at TIMESTAMP
);

— 3. 文档访问权限表（细粒度覆盖）
CREATE TABLE doc_permissions (
id BIGINT PRIMARY KEY,
doc_id BIGINT REFERENCES documents(id),
grantee_type VARCHAR(16), — user/role/department
grantee_id BIGINT,
permission_type VARCHAR(16), — view/edit/delete/share/download
granted_by BIGINT REFERENCES users(id),
expires_at TIMESTAMP,
created_at TIMESTAMP
);

— 4. 查询日志表（审计用）
CREATE TABLE query_logs (
id BIGSERIAL PRIMARY KEY,
user_id BIGINT REFERENCES users(id),
query_text TEXT,
query_embedding vector(1536),
result_count INT,
accessed_doc_ids BIGINT[],
ip_address INET,
user_agent TEXT,
created_at TIMESTAMP DEFAULT NOW(),
risk_score DECIMAL(3,2), — 风险评分 0-1
need_review BOOLEAN DEFAULT false
);

— 5. 敏感词映射表
CREATE TABLE sensitive_terms (
id SERIAL PRIMARY KEY,
term VARCHAR(64) NOT NULL,
risk_level INT, — 1-5
restricted_departments INT[],
action VARCHAR(32) — block/audit_require/allow_with_alert
);

4.2 行级安全策略（PostgreSQL RLS）

— 启用行级安全
ALTER TABLE documents ENABLE ROW LEVEL SECURITY;

— 策略1：用户只能看到自己有权限的文档
CREATE POLICY select_document_policy ON documents
FOR SELECT
USING (
— 公开文档
classification = 'public'
OR
— 文档所属部门 = 用户部门
(department_id = (SELECT department_id FROM users WHERE id = current_setting('app.current_user_id')::BIGINT))
OR
— 用户在白名单中
(current_setting('app.current_user_id')::BIGINT = ANY(allow_users))
OR
— 用户角色在白名单中
((SELECT role FROM users WHERE id = current_setting('app.current_user_id')::BIGINT) = ANY(allow_roles))
OR
— 通过授权表授予权限
EXISTS (
SELECT 1 FROM doc_permissions dp
WHERE dp.doc_id = documents.id
AND dp.grantee_type = 'user'
AND dp.grantee_id = current_setting('app.current_user_id')::BIGINT
AND (dp.expires_at IS NULL OR dp.expires_at > NOW())
)
AND
— 用户不在黑名单
NOT (current_setting('app.current_user_id')::BIGINT = ANY(deny_users))
);

— 策略2：经理可查看下属部门文档
CREATE POLICY manager_subdept_policy ON documents
FOR SELECT
USING (
EXISTS (
SELECT 1 FROM users u
JOIN departments d ON u.department_id = d.id
WHERE u.id = current_setting('app.current_user_id')::BIGINT
AND u.role = 'manager'
AND documents.department_id = ANY(d.sub_department_ids)
)
);

五、搜索时的权限过滤实现

5.1 向量检索 + 权限预过滤

# src/knowledge_base/permission_filter.py
from typing import List, Dict, Any
import asyncpg
import numpy as np
from dataclasses import dataclass

@dataclass
class UserContext:
user_id: int
username: str
department_id: int
role: str
ip_address: str
is_work_hours: bool

class PermissionFilter:
def __init__(self, db_pool: asyncpg.Pool):
self.db_pool = db_pool

async def apply_permission_filter(
self,
user: UserContext,
query_embedding: np.ndarray,
top_k: int = 10,
search_scope: str = "auto"
) –> List[Dict]:
"""
核心流程：
1. 根据用户权限构建SQL WHERE子句
2. 执行向量相似度搜索
3. 后置过滤（脱敏/阻断）
4. 记录审计日志
"""

# 第一步：动态构建权限SQL
permission_sql = await self._build_permission_sql(user, search_scope)

# 第二步：向量检索 + 权限过滤
query = f"""
WITH ranked_docs AS (
SELECT
d.id,
d.title,
d.content,
d.classification,
d.department_id,
d.owner_id,
1 – (d.embedding <=> $1::vector) as similarity
FROM documents d
WHERE {permission_sql}
ORDER BY d.embedding <=> $1::vector
LIMIT $2
)
SELECT * FROM ranked_docs
WHERE similarity > 0.6 — 相似度阈值
ORDER BY similarity DESC;
"""

async with self.db_pool.acquire() as conn:
rows = await conn.fetch(query, query_embedding, top_k * 2) # 多取一些用于后置过滤

# 第三步：后置过滤
filtered_docs = []
for row in rows:
if await self._post_filter_check(user, row):
# 敏感文档脱敏
sanitized_content = self._sanitize_by_clearance(
row['content'],
user.role,
row['classification']
)
filtered_docs.append({
"id": row['id'],
"title": row['title'],
"content_snippet": sanitized_content[:500],
"similarity": row['similarity'],
"classification": row['classification']
})

# 第四步：审计日志
await self._audit_query(user, query_embedding, filtered_docs)

return filtered_docs[:top_k]

async def _build_permission_sql(self, user: UserContext, scope: str) –> str:
"""动态生成权限SQL（关键！）"""

conditions = []

# 1. 公开文档
conditions.append("classification = 'public'")

# 2. 本部门文档
conditions.append(f"department_id = {user.department_id}")

# 3. 用户是文档所有者
conditions.append(f"owner_id = {user.user_id}")

# 4. 角色授权
conditions.append(f"'{user.role}' = ANY(allow_roles)")

# 5. 显式授权（通过权限表）
conditions.append(f"""
EXISTS (
SELECT 1 FROM doc_permissions dp
WHERE dp.doc_id = d.id
AND dp.grantee_type = 'user'
AND dp.grantee_id = {user.user_id}
AND (dp.expires_at IS NULL OR dp.expires_at > NOW())
)
""")

# 6. 部门授权（经理查看下级）
if user.role == 'manager':
conditions.append(f"""
d.department_id = ANY(
SELECT unnest(sub_department_ids)
FROM departments d2
WHERE d2.id = {user.department_id}
)
""")

# 7. 外部合作方限制作用域
if user.role == 'external_partner':
conditions.append(f"""
'{scope}' = ANY(d.project_tags)
AND classification IN ('public', 'internal')
""")

# 8. 排除黑名单用户
conditions.append(f"NOT ({user.user_id} = ANY(deny_users))")

# 9. 离职员工完全禁止（优先过滤）
if user.role == 'terminated':
return "1 = 0" # 永假条件，查不到任何文档

return "(" + " OR ".join(conditions) + ")"

async def _post_filter_check(self, user: UserContext, doc) –> bool:
"""后置过滤：敏感词审查"""

# 检查用户查询的敏感词等级
sensitive_check = await self._check_sensitive_access(user, doc)
if not sensitive_check["allowed"]:
await self._raise_alert(user, doc, sensitive_check["reason"])
return False

# IP白名单检查
if doc['classification'] == 'secret':
if user.ip_address not in await self._get_secret_ip_whitelist():
return False

# 工作时间限制
if doc['classification'] == 'confidential' and not user.is_work_hours:
return False

return True

def _sanitize_by_clearance(self, content: str, role: str, classification: str) –> str:
"""根据文档密级和用户角色进行脱敏"""

if classification == 'secret' and role != 'compliance':
# 替换敏感数字/金额
import re
content = re.sub(r'\\d{6,}', '[机密数字]', content)
content = re.sub(r'[¥$]\\d+[,.]?\\d*', '[机密金额]', content)

if classification == 'confidential' and role == 'employee':
# 保留前20%，后80%脱敏
split_idx = int(len(content) * 0.2)
content = content[:split_idx] + "…[根据权限已脱敏]"

return content

async def _audit_query(self, user: UserContext, query_vec: np.ndarray, results: List[Dict]):
"""审计日志记录"""
async with self.db_pool.acquire() as conn:
await conn.execute("""
INSERT INTO query_logs
(user_id, query_text, query_embedding, result_count,
accessed_doc_ids, ip_address, created_at)
VALUES ($1, $2, $3, $4, $5, $6, NOW())
""", user.user_id, "", query_vec.tolist(), len(results),
[r['id'] for r in results], user.ip_address)

5.2 权限缓存策略（Redis）

# src/knowledge_base/permission_cache.py
import aioredis
import json
from typing import List, Optional

class PermissionCache:
def __init__(self, redis_client: aioredis.Redis, ttl: int = 300):
self.redis = redis_client
self.ttl = ttl

async def get_user_permissions(self, user_id: int) –> Optional[List[str]]:
"""获取用户权限（缓存5分钟）"""
key = f"user_perms:{user_id}"
cached = await self.redis.get(key)
if cached:
return json.loads(cached)
return None

async def set_user_permissions(self, user_id: int, permissions: List[str]):
key = f"user_perms:{user_id}"
await self.redis.setex(key, self.ttl, json.dumps(permissions))

async def invalidate_user(self, user_id: int):
"""权限变更时清除缓存"""
keys = [f"user_perms:{user_id}", f"doc_access:{user_id}:*"]
await self.redis.delete(*keys)

async def check_doc_access(self, user_id: int, doc_id: int) –> Optional[bool]:
"""检查用户对特定文档的访问权限（缓存）"""
key = f"doc_access:{user_id}:{doc_id}"
cached = await self.redis.get(key)
if cached is not None:
return cached == b"1"
return None

async def set_doc_access(self, user_id: int, doc_id: int, allowed: bool):
key = f"doc_access:{user_id}:{doc_id}"
await self.redis.setex(key, self.ttl, b"1" if allowed else b"0")

六、敏感词拦截与审计告警

6.1 敏感词检测器

# src/knowledge_base/sensitive_detector.py
import re
from typing import List, Tuple, Optional

class SensitiveDetector:
def __init__(self, sensitive_db):
self.sensitive_db = sensitive_db
self.compiled_patterns = {}

async def detect(self, query: str, user_role: str) –> Tuple[bool, Optional[str], int]:
"""
返回: (是否阻断, 原因, 风险等级)
"""
query_lower = query.lower()

async with self.sensitive_db.acquire() as conn:
terms = await conn.fetch("SELECT * FROM sensitive_terms")

for term in terms:
pattern = self._get_pattern(term['term'])
if pattern.search(query_lower):
# 检查用户是否有权限搜索该敏感词
if user_role not in term.get('allowed_roles', []):
risk_level = term['risk_level']
if risk_level >= 4: # 高风险
return True, f"搜索包含敏感词: {term['term']}", risk_level
else:
return False, f"需审计: {term['term']}", risk_level

return False, None, 0

def _get_pattern(self, term: str):
if term not in self.compiled_patterns:
# 支持正则表达式
self.compiled_patterns[term] = re.compile(r'\\b' + re.escape(term) + r'\\b', re.IGNORECASE)
return self.compiled_patterns[term]

6.2 告警与阻断工作流

管理员

告警服务

权限引擎

敏感词检测

知识库API

User

管理员

告警服务

权限引擎

敏感词检测

知识库API

User

#mermaid-svg-XS6tTuqLLxuoONA6{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-XS6tTuqLLxuoONA6 .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-XS6tTuqLLxuoONA6 .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-XS6tTuqLLxuoONA6 .error-icon{fill:#552222;}#mermaid-svg-XS6tTuqLLxuoONA6 .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-XS6tTuqLLxuoONA6 .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-XS6tTuqLLxuoONA6 .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-XS6tTuqLLxuoONA6 .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-XS6tTuqLLxuoONA6 .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-XS6tTuqLLxuoONA6 .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-XS6tTuqLLxuoONA6 .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-XS6tTuqLLxuoONA6 .marker{fill:#333333;stroke:#333333;}#mermaid-svg-XS6tTuqLLxuoONA6 .marker.cross{stroke:#333333;}#mermaid-svg-XS6tTuqLLxuoONA6 svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-XS6tTuqLLxuoONA6 p{margin:0;}#mermaid-svg-XS6tTuqLLxuoONA6 .actor{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-XS6tTuqLLxuoONA6 text.actor>tspan{fill:black;stroke:none;}#mermaid-svg-XS6tTuqLLxuoONA6 .actor-line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);}#mermaid-svg-XS6tTuqLLxuoONA6 .innerArc{stroke-width:1.5;stroke-dasharray:none;}#mermaid-svg-XS6tTuqLLxuoONA6 .messageLine0{stroke-width:1.5;stroke-dasharray:none;stroke:#333;}#mermaid-svg-XS6tTuqLLxuoONA6 .messageLine1{stroke-width:1.5;stroke-dasharray:2,2;stroke:#333;}#mermaid-svg-XS6tTuqLLxuoONA6 #arrowhead path{fill:#333;stroke:#333;}#mermaid-svg-XS6tTuqLLxuoONA6 .sequenceNumber{fill:white;}#mermaid-svg-XS6tTuqLLxuoONA6 #sequencenumber{fill:#333;}#mermaid-svg-XS6tTuqLLxuoONA6 #crosshead path{fill:#333;stroke:#333;}#mermaid-svg-XS6tTuqLLxuoONA6 .messageText{fill:#333;stroke:none;}#mermaid-svg-XS6tTuqLLxuoONA6 .labelBox{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-XS6tTuqLLxuoONA6 .labelText,#mermaid-svg-XS6tTuqLLxuoONA6 .labelText>tspan{fill:black;stroke:none;}#mermaid-svg-XS6tTuqLLxuoONA6 .loopText,#mermaid-svg-XS6tTuqLLxuoONA6 .loopText>tspan{fill:black;stroke:none;}#mermaid-svg-XS6tTuqLLxuoONA6 .loopLine{stroke-width:2px;stroke-dasharray:2,2;stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);}#mermaid-svg-XS6tTuqLLxuoONA6 .note{stroke:#aaaa33;fill:#fff5ad;}#mermaid-svg-XS6tTuqLLxuoONA6 .noteText,#mermaid-svg-XS6tTuqLLxuoONA6 .noteText>tspan{fill:black;stroke:none;}#mermaid-svg-XS6tTuqLLxuoONA6 .activation0{fill:#f4f4f4;stroke:#666;}#mermaid-svg-XS6tTuqLLxuoONA6 .activation1{fill:#f4f4f4;stroke:#666;}#mermaid-svg-XS6tTuqLLxuoONA6 .activation2{fill:#f4f4f4;stroke:#666;}#mermaid-svg-XS6tTuqLLxuoONA6 .actorPopupMenu{position:absolute;}#mermaid-svg-XS6tTuqLLxuoONA6 .actorPopupMenuPanel{position:absolute;fill:#ECECFF;box-shadow:0px 8px 16px 0px rgba(0,0,0,0.2);filter:drop-shadow(3px 5px 2px rgb(0 0 0 / 0.4));}#mermaid-svg-XS6tTuqLLxuoONA6 .actor-man line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-XS6tTuqLLxuoONA6 .actor-man circle,#mermaid-svg-XS6tTuqLLxuoONA6 line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;stroke-width:2px;}#mermaid-svg-XS6tTuqLLxuoONA6 :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

搜索"核材料生产工艺"

detect(query)

阻断=true, 原因="高风险敏感词"

生成高优先级告警

钉钉/邮件通知

"XX用户在XX时间搜索敏感词"

403 拒绝访问

"该搜索词超出您的权限范围"

审查用户权限

用户无相关权限，合理阻断

标记告警为"已处理-授权拒绝"

七、完整API实现

# src/knowledge_base/api.py
from fastapi import FastAPI, Depends, HTTPException, Query
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from typing import Optional, List
from pydantic import BaseModel

app = FastAPI(title="企业知识库API")
security = HTTPBearer()

class SearchRequest(BaseModel):
query: str
top_k: int = 10
scope: Optional[str] = None # project_X, department_Y

class SearchResponse(BaseModel):
results: List[dict]
has_more: bool
audit_id: str

@app.post("/api/v1/search", response_model=SearchResponse)
async def search_knowledge_base(
request: SearchRequest,
credentials: HTTPAuthorizationCredentials = Depends(security),
user_context: UserContext = Depends(get_current_user),
permission_filter: PermissionFilter = Depends(get_permission_filter),
sensitive_detector: SensitiveDetector = Depends(get_sensitive_detector)
):
# 1. 敏感词检测
blocked, reason, risk = await sensitive_detector.detect(
request.query,
user_context.role
)
if blocked:
raise HTTPException(status_code=403, detail=reason)

# 2. 生成查询向量
query_embedding = await generate_embedding(request.query)

# 3. 权限过滤 + 向量检索
results = await permission_filter.apply_permission_filter(
user=user_context,
query_embedding=query_embedding,
top_k=request.top_k,
search_scope=request.scope or "auto"
)

# 4. 记录审计
audit_id = await log_search_audit(
user_id=user_context.user_id,
query=request.query,
result_count=len(results),
risk_level=risk
)

return SearchResponse(
results=results,
has_more=False,
audit_id=audit_id
)

八、部署与配置清单

8.1 环境变量配置

# .env.production
DB_URL=postgresql://user:pass@localhost:5432/knowledge_base
REDIS_URL=redis://localhost:6379/0
LDAP_URL=ldaps://ldap.company.com:636
SSO_CERT_PATH=/certs/sso.pem
SENSITIVE_WORD_DB=/data/sensitive_words.db
ALERT_WEBHOOK=https://oapi.dingtalk.com/robot/send?access_token=xxx

8.2 初始化脚本

— init_db.sql
— 创建向量扩展
CREATE EXTENSION IF NOT EXISTS vector;

— 创建部门层级
INSERT INTO departments (id, name, parent_id, sub_department_ids) VALUES
(1, '集团', NULL, ARRAY[2,3,4,5]),
(2, '研发中心', 1, ARRAY[6,7,8]),
(3, '生产部', 1, ARRAY[9,10]),
(4, '销售部', 1, ARRAY[11,12]),
(5, '法务部', 1, ARRAY[13]);

— 创建示例文档
INSERT INTO documents (id, title, content, embedding, department_id, classification, allow_roles) VALUES
(1, '2026年Q2销售策略', '…', '[0.1,0.2,…]', 4, 'internal', ARRAY['sales', 'manager']),
(2, '核级钛合金生产工艺', '…', '[0.3,0.4,…]', 2, 'secret', ARRAY['compliance', 'rd_director']);

九、效果评估指标

本方案提供了从权限模型、数据库设计、代码实现到部署运维的完整企业知识库权限管理方案，可直接用于生产环境。

后面需进一步细化（如LDAP集成、细粒度脱敏算法、或合规报表生成）。