企业级 Agent 产品：知识库权限隔离与多级审批流的架构设计

一、企业知识的安全边界：Agent 访问越权与审批黑洞

在企业级 Agent 产品中，知识库是 Agent 的核心资产，也是最大的安全风险点。企业知识通常按部门、项目、密级进行隔离——财务数据只有财务部门可访问，客户信息只有销售和客服可访问，核心技术文档只有研发团队可访问。当 Agent 代表用户检索知识库时，如果权限控制不严格，一个普通员工的 Agent 可能通过精心构造的 Prompt 获取到高管才能看到的战略文档。

审批流是另一个痛点。Agent 在执行某些高风险操作时（如发送外部邮件、修改生产配置、访问敏感数据），需要经过审批。传统的审批系统是人工驱动的，审批人需要登录系统、查看详情、点击确认。当 Agent 的操作频率远超人工审批速度时，审批流程就成了效率瓶颈。更复杂的是，不同操作需要不同级别的审批——修改配置需要技术负责人审批，发送外部邮件需要部门主管审批，访问核心数据需要安全团队审批。

二、权限隔离与审批流的协同架构

flowchart TB
    subgraph 请求入口
        USER[用户请求] --> AGENT[Agent 引擎]
    end

    subgraph 权限层
        AGENT --> AUTH[身份认证: 验证用户身份]
        AUTH --> RBAC[角色权限: RBAC 基础权限]
        RBAC --> ABAC[属性权限: ABAC 细粒度控制]
        ABAC --> KB_CHECK{知识库访问检查}
        KB_CHECK --> |有权限| RETRIEVE[知识检索]
        KB_CHECK --> |无权限| DENY[拒绝访问+审计日志]
    end

    subgraph 审批层
        AGENT --> RISK[风险评估: 判定操作风险等级]
        RISK --> |低风险| AUTO_APPROVE[自动审批]
        RISK --> |中风险| SINGLE[单人审批: 直属主管]
        RISK --> |高风险| MULTI[多级审批: 主管+安全团队]
        RISK --> |极高风险| EMERGENCY[紧急审批: 实时通知+电话确认]
    end

    subgraph 审计层
        DENY --> AUDIT[审计日志: 记录所有访问与审批]
        AUTO_APPROVE --> AUDIT
        SINGLE --> AUDIT
        MULTI --> AUDIT
        RETRIEVE --> AUDIT
    end

    style ABAC fill:#fff3e0
    style RISK fill:#e8f5e9
    style AUDIT fill:#e3f2fd

权限层采用 RBAC + ABAC 混合模型。RBAC 处理粗粒度的角色权限（如"财务部员工"角色可以访问财务知识库），ABAC 处理细粒度的属性权限（如"入职不满 3 个月的员工不能访问核心数据"）。两层叠加既保证了管理效率，又实现了精确控制。

审批层基于风险评估的动态分级。每个操作在执行前经过风险评估引擎，根据操作类型、目标资源敏感度、用户权限等级等维度计算风险分数，再根据分数匹配对应的审批级别。低风险操作自动通过，中高风险操作进入审批队列。

三、权限引擎与审批流的工程实现

# permission_approval_engine.py — 权限隔离与审批流引擎
import time
import hashlib
import json
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional


class RiskLevel(Enum):
    LOW = "low"             # 自动审批
    MEDIUM = "medium"       # 单人审批
    HIGH = "high"           # 多级审批
    CRITICAL = "critical"   # 紧急审批


class ApprovalStatus(Enum):
    PENDING = "pending"
    APPROVED = "approved"
    REJECTED = "rejected"
    EXPIRED = "expired"


@dataclass
class User:
    """用户模型"""
    user_id: str
    roles: list[str]
    department: str
    level: int              # 职级
    join_date: str          # 入职日期
    attributes: dict = field(default_factory=dict)


@dataclass
class KnowledgeBase:
    """知识库模型"""
    kb_id: str
    name: str
    sensitivity: str        # public / internal / confidential / restricted
    allowed_roles: list[str]
    allowed_departments: list[str]
    extra_conditions: dict = field(default_factory=dict)


@dataclass
class ApprovalRequest:
    """审批请求"""
    request_id: str
    user_id: str
    operation: str
    target_resource: str
    risk_level: RiskLevel
    status: ApprovalStatus = ApprovalStatus.PENDING
    approvers: list[str] = field(default_factory=list)
    approved_by: list[str] = field(default_factory=list)
    created_at: float = field(default_factory=time.time)
    expires_at: float = 0
    reason: str = ""


class PermissionEngine:
    """权限引擎：RBAC + ABAC 混合鉴权"""

    def __init__(self):
        self._role_permissions: dict[str, list[str]] = {}
        self._kb_configs: dict[str, KnowledgeBase] = {}
        self._abac_rules: list[dict] = []

    def check_access(self, user: User, kb: KnowledgeBase,
                     operation: str = "read") -> tuple[bool, str]:
        """检查用户是否有权访问知识库"""
        # Step 1: RBAC 检查——角色是否匹配
        role_match = False
        for role in user.roles:
            if role in kb.allowed_roles:
                role_match = True
                break

        if not role_match:
            return False, f"角色 {user.roles} 不在允许列表中"

        # Step 2: 部门检查
        if kb.allowed_departments and \
           user.department not in kb.allowed_departments:
            return False, f"部门 {user.department} 无权访问"

        # Step 3: ABAC 细粒度检查
        for rule in self._abac_rules:
            if not self._evaluate_abac_rule(user, kb, rule):
                return False, f"ABAC 规则拒绝: {rule.get('description', '')}"

        # Step 4: 敏感度检查——高敏感知识库需要更高职级
        sensitivity_levels = {
            "public": 0, "internal": 1,
            "confidential": 3, "restricted": 5,
        }
        required_level = sensitivity_levels.get(kb.sensitivity, 0)
        if user.level < required_level:
            return False, f"职级不足：需要 L{required_level}，当前 L{user.level}"

        return True, "授权通过"

    def _evaluate_abac_rule(self, user: User, kb: KnowledgeBase,
                            rule: dict) -> bool:
        """评估单条 ABAC 规则"""
        rule_type = rule.get("type")

        if rule_type == "time_restriction":
            # 时间限制：某些知识库只在工作时间可访问
            current_hour = time.localtime().tm_hour
            allowed_start = rule.get("start_hour", 0)
            allowed_end = rule.get("end_hour", 24)
            if not (allowed_start <= current_hour < allowed_end):
                return False

        elif rule_type == "tenure_requirement":
            # 入职时长要求
            min_days = rule.get("min_days", 0)
            join_timestamp = time.mktime(
                time.strptime(user.join_date, "%Y-%m-%d")
            )
            tenure_days = (time.time() - join_timestamp) / 86400
            if tenure_days < min_days:
                return False

        elif rule_type == "attribute_match":
            # 自定义属性匹配
            attr_key = rule.get("attribute_key")
            attr_values = rule.get("attribute_values", [])
            user_val = user.attributes.get(attr_key)
            if user_val not in attr_values:
                return False

        return True


class RiskAssessor:
    """风险评估引擎：根据操作特征判定风险等级"""

    # 操作风险权重
    OPERATION_RISK = {
        "read": 1,
        "search": 2,
        "write": 4,
        "delete": 8,
        "send_email": 6,
        "modify_config": 7,
        "export_data": 9,
        "access_restricted": 10,
    }

    # 资源敏感度权重
    SENSITIVITY_WEIGHT = {
        "public": 1, "internal": 2,
        "confidential": 4, "restricted": 8,
    }

    def assess(self, operation: str, resource_sensitivity: str,
               user: User) -> RiskLevel:
        """计算操作的风险等级"""
        op_risk = self.OPERATION_RISK.get(operation, 5)
        res_risk = self.SENSITIVITY_WEIGHT.get(resource_sensitivity, 3)

        # 综合风险分数
        score = op_risk * res_risk

        # 职级越高，风险容忍度越高
        score = max(1, score - user.level)

        if score <= 4:
            return RiskLevel.LOW
        elif score <= 12:
            return RiskLevel.MEDIUM
        elif score <= 24:
            return RiskLevel.HIGH
        else:
            return RiskLevel.CRITICAL


class ApprovalEngine:
    """审批流引擎：基于风险等级的分级审批"""

    def __init__(self, risk_assessor: RiskAssessor):
        self._assessor = risk_assessor
        self._pending_requests: dict[str, ApprovalRequest] = {}
        self._approval_rules: dict[RiskLevel, dict] = {
            RiskLevel.LOW: {
                "auto_approve": True,
                "timeout_minutes": 0,
            },
            RiskLevel.MEDIUM: {
                "auto_approve": False,
                "approvers": "direct_manager",
                "timeout_minutes": 60,
            },
            RiskLevel.HIGH: {
                "auto_approve": False,
                "approvers": "manager_and_security",
                "timeout_minutes": 120,
            },
            RiskLevel.CRITICAL: {
                "auto_approve": False,
                "approvers": "multi_level_emergency",
                "timeout_minutes": 30,
            },
        }

    def submit_request(self, user: User, operation: str,
                       target_resource: str,
                       resource_sensitivity: str) -> ApprovalRequest:
        """提交审批请求"""
        risk_level = self._assessor.assess(
            operation, resource_sensitivity, user
        )

        request_id = hashlib.md5(
            f"{user.user_id}:{operation}:{target_resource}:{time.time()}"
            .encode()
        ).hexdigest()[:12]

        rule = self._approval_rules[risk_level]
        timeout = rule.get("timeout_minutes", 60)

        request = ApprovalRequest(
            request_id=request_id,
            user_id=user.user_id,
            operation=operation,
            target_resource=target_resource,
            risk_level=risk_level,
            approvers=self._resolve_approvers(
                user, rule.get("approvers", "")
            ),
            expires_at=time.time() + timeout * 60,
        )

        # 低风险自动审批
        if rule.get("auto_approve"):
            request.status = ApprovalStatus.APPROVED

        self._pending_requests[request_id] = request
        return request

    def approve(self, request_id: str, approver_id: str) -> ApprovalStatus:
        """审批人批准请求"""
        request = self._pending_requests.get(request_id)
        if request is None:
            return ApprovalStatus.REJECTED

        if time.time() > request.expires_at:
            request.status = ApprovalStatus.EXPIRED
            return ApprovalStatus.EXPIRED

        if approver_id not in request.approvers:
            return ApprovalStatus.REJECTED

        if approver_id not in request.approved_by:
            request.approved_by.append(approver_id)

        # 检查是否所有审批人都已批准
        required_count = len(request.approvers)
        if len(request.approved_by) >= required_count:
            request.status = ApprovalStatus.APPROVED

        return request.status

    def _resolve_approvers(self, user: User,
                           rule_type: str) -> list[str]:
        """根据审批规则解析审批人列表"""
        # 生产环境应从组织架构服务获取
        if rule_type == "direct_manager":
            return [f"manager_of_{user.department}"]
        elif rule_type == "manager_and_security":
            return [
                f"manager_of_{user.department}",
                "security_team_lead",
            ]
        elif rule_type == "multi_level_emergency":
            return [
                f"manager_of_{user.department}",
                "security_team_lead",
                "cto",
            ]
        return []

四、权限与审批系统的性能与体验权衡

权限检查的性能开销：每次知识库检索前都需要执行权限检查，在高并发场景下，权限引擎的延迟直接影响检索响应时间。RBAC 检查是 O(1) 的哈希查找，但 ABAC 规则评估是 O(N) 的线性扫描，规则数量增多时延迟上升。优化方案是将 ABAC 规则预编译为决策树，将评估复杂度从 O(N) 降低到 O(log N)。同时，对权限检查结果进行短期缓存（TTL 5 分钟），同一用户对同一知识库的重复访问直接命中缓存。

审批流的用户体验：审批等待是 Agent 操作延迟的主要来源。一个需要多级审批的操作，从提交到最终批准可能需要数小时。对于实时交互场景（如用户正在等待 Agent 的回复），这种延迟不可接受。折中方案是引入"预授权"机制——对于可预测的高频操作，用户可以预先授权 Agent 在特定范围内自主执行，无需逐次审批。预授权本身需要经过完整审批流程，但一旦获得，后续操作在预授权范围内自动通过。

审计日志的存储膨胀：所有访问和审批操作都需要记录审计日志。日活 1 万用户的 Agent 系统，日均产生约 50 万条审计记录，每条约 500 字节，日均增量约 250MB。热数据保留 30 天，冷数据归档到对象存储，保留 1 年。

五、总结

企业级 Agent 的知识库权限隔离与审批流设计，核心在于 RBAC + ABAC 混合鉴权模型和基于风险评估的分级审批机制。RBAC 保证管理效率，ABAC 实现精确控制；风险评估引擎将操作分为四个风险等级，对应不同的审批策略。在落地时，权限检查的性能优化（决策树 + 缓存）和审批流的体验优化（预授权机制）是两个关键工程决策。建议从 RBAC 起步，待权限模型稳定后再引入 ABAC 规则，避免过度设计。