7.6 Agents Rule of Two 与智能体安全设计原则

“Rule of Two” 是 Meta 在 2025 年提出的一项重要 AI 智能体安全实践指导。按照 Meta 原文，这一原则的核心不是“所有不可逆操作都必须由两个 Agent 批准”，而是：在单个 session 中，一个 agent 最多只能同时满足以下三项性质中的两项：处理不可信输入、访问敏感系统/私有数据、改变状态或对外通信；若三项都需要，则至少要加入监督或可靠验证。 本节将围绕这一更准确的定义来讨论其设计含义。

7.6.1 Rule of Two 的核心原则

基本概念

Rule of Two:
在单个 session 中，一个 agent 最多同时满足以下三项中的两项：
1. 处理不可信输入
2. 访问敏感系统或私有数据
3. 改变状态或对外通信

如果三项都需要，则至少要有监督或可靠验证。

这一原则可以类比“关键操作需要独立验证”的传统安全思想，但在 AI 场景中更准确的落点是：不要让单个 agent 在同一 session 中同时具备看不可信输入、访问敏感资源、执行高影响动作这三项能力。

原则的哲学基础

图 7-20：Rule of Two 的理论基础

Rule of Two 与其他安全实践的关系

实践	焦点	与 Rule of Two 的关系
权限最小化	限制单个系统的权限	辅助：减少单个系统的威胁面
审计日志	记录所有操作	补充：提供事后追踪
工具调用沙盒	隔离执行环境	辅助：限制操作影响
Rule of Two	前置约束	核心：避免单个 Agent 同时具备三类高风险能力

7.6.2 不可逆操作的监督网关（Rule of Two 之外的一种实现）

先厘清命名：本节的「双 Agent 复核」是一种监督网关实现，不是 Rule of Two 本身。 按 7.6.1 引的 Meta 原文，Rule of Two 约束的是单个 session 内一个 agent 最多同时具备三项性质中的两项；当三项都需要时，Meta 给出的基线是人工确认（human-in-the-loop）或其他可靠校验。用第二个 LLM Agent 去复核是一种更弱的替代——本节 7.6.4「勾结攻击」正是在讲它为什么脆弱。下文的代码与流程图沿用「双 Agent」写法时，请按此理解。

不可逆操作的定义和分类

图 7-21：操作可逆性分类

不可逆操作的风险矩阵

操作类型	影响范围	恢复成本	风险等级
账户删除	用户数据完全丧失	极高（可能无法恢复）	关键
资金转账	资金直接丧失	极高（需法律介入）	关键
生产数据删除	业务中断，数据丧失	很高（需完整备份）	关键
权限撤销	用户功能受限	中高（需重新审批）	高
配置更改	系统行为改变	中（需回滚）	中高

安全网关架构

不可逆操作通常需要经过多层安全网关，并在必要时引入额外监督或可靠验证；双重验证只是其中一种实现方式：

图 7-22：不可逆操作安全网关流程

实现示例（伪代码）

下面的伪代码展示的是 Rule of Two 的一种实现方式：通过主处理器 + 独立验证通道来处理不可逆操作。它不应被误读为“Rule of Two 必然等于两个 Agent 审批”。

from enum import Enum
from dataclasses import dataclass

class OperationReversibility(Enum):
    REVERSIBLE = 1      # 可逆
    PARTIALLY_REVERSIBLE = 2  # 部分可逆
    IRREVERSIBLE = 3    # 不可逆

@dataclass
class IrreversibleOperation:
    operation_type: str
    user_id: str
    operation_id: str
    reversibility: OperationReversibility
    approval_status: dict  # {agent_id: bool}

class SafeAgentGateway:
    """不可逆操作的安全网关"""

    def __init__(self):
        self.primary_agent = PrimaryAgent()
        self.validator_agent = ValidatorAgent()

    def process_operation(self, operation: IrreversibleOperation):
        """处理操作的主流程"""

        # Step 1: 初步分类
        if operation.reversibility != OperationReversibility.IRREVERSIBLE:
            # 可逆操作直接执行
            return self.execute_operation(operation)

        # Step 2: 第一个Agent检查
        primary_approval = self.primary_agent.review(operation)
        operation.approval_status['primary'] = primary_approval

        if not primary_approval:
            self.log_rejection(operation, 'primary_agent')
            raise OperationRejected("Primary agent rejected")

        # Step 3: 第二个Agent独立验证
        validator_approval = self.validator_agent.review(operation)
        operation.approval_status['validator'] = validator_approval

        if not validator_approval:
            self.log_rejection(operation, 'validator_agent')
            raise OperationRejected("Validator agent rejected")

        # Step 4: 双重检查通过，执行操作
        if primary_approval and validator_approval:
            return self.execute_operation(operation)

    def execute_operation(self, operation: IrreversibleOperation):
        """执行操作并记录审计日志"""
        try:
            result = self._do_execute(operation)
            self.audit_log.record(
                timestamp=time.time(),
                operation_id=operation.operation_id,
                status='SUCCESS',
                approvals=operation.approval_status
            )
            return result
        except Exception as e:
            self.audit_log.record(
                timestamp=time.time(),
                operation_id=operation.operation_id,
                status='FAILED',
                error=str(e)
            )
            raise

7.6.3 权限最小化实践

权限模型架构

图 7-23：多层权限模型

Zero Trust 原则在 Agent 中的应用

class ZeroTrustAgentSystem:
    """基于 Zero Trust 的智能体权限模型"""

    def __init__(self):
        self.permission_engine = PermissionEngine()

    def authorize_operation(self, agent_id, operation):
        """每次操作都进行完整的权限验证"""

        # 不信任 Agent 的身份声称
        agent_identity = self.verify_agent_identity(agent_id)

        # 不信任静态权限配置，而是动态评估
        context = {
            'agent': agent_identity,
            'operation': operation,
            'timestamp': time.time(),
            'current_resource_state': self.get_current_state(),
            'historical_behavior': self.get_agent_history(agent_id)
        }

        # 多维度权限评估
        checks = [
            self.check_role_permission(agent_identity, operation),
            self.check_resource_limit(agent_identity, operation),
            self.check_time_based_restriction(agent_identity),
            self.check_behavioral_anomaly(agent_identity, context),
            self.check_dependency_chain(operation)
        ]

        if all(checks):
            return True
        else:
            # 详细的拒绝原因记录
            self.log_authorization_failure(
                agent_id=agent_id,
                operation=operation,
                failed_checks=which_checks_failed(checks)
            )
            return False

    def check_dependency_chain(self, operation):
        """检查操作的依赖链是否都被授权"""
        dependencies = self.analyze_operation_dependencies(operation)

        for dependency in dependencies:
            if not self.is_authorized(dependency):
                return False

        return True

权限降级案例（典型场景示例）

以下是一个基于真实工程场景构造的假设案例，用以说明权限模型的重要性：假设一个 AI 代码生成 Agent 误删了生产数据库中的关键表：

事件经过：

1. Agent 被赋予“数据库修改”的广泛权限

2. Agent 在执行“清理过期日志”任务时，因为理解错误执行了 DROP TABLE 操作

3. 数据丧失，恢复耗时 3 天

事件教训：

• 权限过于宽泛（可以删除任何表，而不仅是日志表）

• 缺少第二层验证（没有要求人类确认 DELETE/DROP 操作）

• 操作前缺少 dry-run（可以显示将要删除的内容）

改进方案：

class ImprovedDatabaseAgent:
    """改进的数据库Agent权限模型"""

    # 注意：审计日志（audit_logs）这类取证数据应当只追加、不可删除，
    # 永远不要把它放进删除白名单；这里只允许清理临时导出与过期会话
    ALLOWED_DELETE_TARGETS = {
        "temp_exports": {"owner_id", "created_at"},
        "expired_sessions": {"user_id", "expires_at"},
    }

    def __init__(self):
        self.permission_level = PermissionLevel.MINIMAL

    def build_where_clause(self, table_name, filters):
        """只允许白名单字段参与条件拼装，具体值走参数绑定"""
        allowed_columns = self.ALLOWED_DELETE_TARGETS.get(table_name)
        if not allowed_columns:
            raise PermissionDenied(f"Table is not approved for deletes: {table_name}")

        clauses = []
        params = {}
        for column, value in filters.items():
            if column not in allowed_columns:
                raise PermissionDenied(f"Column is not approved for deletes: {column}")
            param_name = f"p_{column}"
            clauses.append(f"{column} = :{param_name}")
            params[param_name] = value

        if not clauses:
            raise ValueError("Delete filters must not be empty")

        return " AND ".join(clauses), params

    def execute_delete(self, table_name, filters):
        """删除操作的改进流程"""

        # Step 1: 权限检查
        if not self.has_permission('delete', specific_table=table_name):
            raise PermissionDenied(f"No permission to delete from {table_name}")

        where_clause, params = self.build_where_clause(table_name, filters)

        # Step 2: Dry-run - 统计将要删除的记录。表名和列名只来自白名单，值走参数绑定。
        # 日志只记录脱敏摘要（行数、表名、条件列），不把整条记录写入日志；
        # 完整记录预览仅通过下方 request_human_approval 的 preview 参数交给审批人
        affected_records = self.query(
            f"SELECT * FROM {table_name} WHERE {where_clause}",
            params
        )

        logger.info(
            f"Will delete {len(affected_records)} records "
            f"from {table_name} (filter: {where_clause})"
        )

        # Step 3: 需要人类确认（不可逆操作）
        approval = self.request_human_approval(
            operation_type='DELETE',
            table=table_name,
            affected_count=len(affected_records),
            preview=affected_records[:10]
        )

        if not approval:
            logger.info("Delete operation cancelled by human")
            return

        # Step 4: 使用事务，允许回滚
        with self.database.transaction():
            cursor = self.database.execute(
                f"DELETE FROM {table_name} WHERE {where_clause}",
                params
            )
            self.audit_log.record(
                operation='DELETE',
                table=table_name,
                affected_rows=cursor.rowcount,
                timestamp=time.time()
            )

7.6.4 多智能体安全边界

智能体间的信任模型

图 7-24：多智能体信任模型

勾结攻击防御

在多智能体系统中，一个新兴的风险是两个或多个智能体进行“勾结”，共同规避安全检查。如果系统把 Rule of Two 实现成“双重验证”，那么这些验证通道之间的独立性就会成为新的安全前提。

勾结攻击的形式

场景1：两个验证通道都被攻击者控制
- 攻击者通过某种方式 contaminate 两个验证角色
- 它们都“批准”一个实际上有害的操作
- Rule of Two变成假象

场景2：一个验证Agent被社工
- 通过某种社工手段或逻辑漏洞说服验证Agent
- 使其做出错误的批准决定
- 第二个Agent无法检测到这种细微的操纵

场景3：Rule of Two 的实现有缺陷
- 两个验证通道实际上共享某些内部状态
- 一个被 compromise 导致另一个也受影响
- 缺乏真正的独立性

勾结攻击防御机制

1. 来自不同供应商的验证通道

class VendorDiversificationDefense:
    """使用来自不同供应商的验证通道"""

    def __init__(self):
        # 使用完全不同来源的模型
        self.primary_agent = load_model(vendor='OpenAI', model='current-openai-primary')
        self.validator_agent = load_model(vendor='Anthropic', model='current-anthropic-primary')
        # 可选的第三个 Agent 来自不同供应商
        self.tertiary_agent = load_model(vendor='Google', model='current-google-primary')

    def verify_operation(self, operation):
        """多来源验证操作"""

        # 来自不同供应商的推理过程是独立的
        primary_approval = self.primary_agent.verify(operation)
        validator_approval = self.validator_agent.verify(operation)

        # 供应商多样性降低了被同时 compromised 的风险
        if primary_approval and validator_approval:
            return True
        else:
            return False

2. 异步验证与时间隔离

class AsynchronousVerificationWithTimeIsolation:
    """异步验证与时间隔离防御勾结"""

    def execute_irreversible_operation(self, operation):
        """异步和时间隔离的验证流程"""

        # Step 1: 第一个Agent审批
        approval1 = self.primary_agent.verify(operation)
        if not approval1:
            raise OperationRejected("Primary agent rejected")

        # Step 2: 时间隔离（等待一段时间）
        # 这个延迟使得简单的同步勾结变得困难
        wait_duration = random.randint(60, 300)  # 1-5 分钟
        time.sleep(wait_duration)

        # Step 3: 独立的第二个审批
        # 在不同的时间、可能不同的计算资源上运行
        approval2 = self.secondary_agent.verify(operation)
        if not approval2:
            raise OperationRejected("Secondary agent rejected after isolation period")

        # Step 4: 执行
        if approval1 and approval2:
            return self.execute_and_log(operation)

3. 防篡改的审计日志

class TamperProofAuditLog:
    """防篡改的审计日志机制"""

    def __init__(self):
        # 使用区块链或密码学哈希链确保日志完整性
        self.log_chain = CryptoHashChain()

    def record_verification(self, operation, agent_approvals, timestamp):
        """记录验证过程，防止篡改"""

        log_entry = {
            'operation_id': operation.id,
            'timestamp': timestamp,
            'approvals': agent_approvals,  # {agent_id: {approved: bool, reasoning: str}}
            'previous_hash': self.log_chain.last_hash,
        }

        # 添加加密签名
        log_entry['signature'] = self.sign_entry(log_entry)

        # 添加到不可变链
        self.log_chain.append(log_entry)

        return log_entry

    def verify_log_integrity(self):
        """验证整个日志链的完整性"""
        current_hash = None
        for entry in self.log_chain.entries:
            if not self.verify_entry_signature(entry):
                return False, f"Signature mismatch at {entry['operation_id']}"

            if current_hash and entry['previous_hash'] != current_hash:
                return False, f"Chain break at {entry['operation_id']}"

            current_hash = self.hash_entry(entry)

        return True, "Log integrity verified"

4. 定期 Agent 轮换

class AgentRotationDefense:
    """定期轮换验证 Agent 防止被 compromised"""

    def __init__(self, rotation_interval_hours=24):
        self.rotation_interval = rotation_interval_hours * 3600
        self.last_rotation = time.time()
        self.agent_pool = self.initialize_agent_pool()
        self.current_agent_index = 0

    def initialize_agent_pool(self):
        """初始化多个 Agent 池"""
        return [
            ModelAgent(vendor='OpenAI', model='current-openai-primary'),
            ModelAgent(vendor='Anthropic', model='current-anthropic-primary'),
            ModelAgent(vendor='Google', model='current-google-primary'),
            ModelAgent(vendor='Meta', model='current-meta-primary'),
            ModelAgent(vendor='Mistral', model='current-mistral-primary'),
        ]

    def get_current_validator(self):
        """获取当前的验证 Agent"""
        current_time = time.time()

        # 检查是否需要轮换
        if current_time - self.last_rotation > self.rotation_interval:
            self.rotate_agent()
            self.last_rotation = current_time

        return self.agent_pool[self.current_agent_index]

    def rotate_agent(self):
        """轮换到下一个 Agent"""
        self.current_agent_index = (self.current_agent_index + 1) % len(self.agent_pool)
        logger.info(f"Agent rotated to {self.get_current_validator()}")

    def verify_with_rotation(self, operation):
        """使用轮换的 Agent 进行验证"""
        validator = self.get_current_validator()
        return validator.verify(operation)

防御的综合应用

Agent 间通信的安全协议

class SecureAgentBridge:
    """Agent 间的安全通信桥梁"""

    def __init__(self):
        self.message_queue = VerifiedQueue()
        self.permission_checker = PermissionChecker()

    def call_external_agent(self, source_agent, target_agent, request):
        """调用外部 Agent 的安全方法"""

        # Step 1: 验证消息来源
        if not self.verify_agent_identity(source_agent):
            raise UntrustedAgentError(f"Cannot verify {source_agent}")

        # Step 2: 检查权限
        if not self.permission_checker.can_call(
            source_agent,
            target_agent,
            request.operation
        ):
            raise PermissionDenied(
                f"{source_agent} cannot call {target_agent}"
            )

        # Step 3: 消息加密和签名
        signed_request = self.sign_request(request, source_agent)

        # Step 4: 通过隔离通道发送
        response = self.message_queue.send(
            target=target_agent,
            message=signed_request,
            timeout=30
        )

        # Step 5: 验证响应
        if not self.verify_response_signature(response, target_agent):
            raise UntrustedResponseError("Response signature verification failed")

        # Step 6: 沙箱执行响应中包含的代码
        return self.execute_in_sandbox(response.payload)

    def execute_in_sandbox(self, code):
        """在隔离环境中执行外部 Agent 的代码"""
        sandbox = Sandbox(
            allowed_modules=['json', 'datetime'],
            memory_limit='100MB',
            timeout=5,
            no_network=True
        )
        return sandbox.execute(code)

组织级 Agent 协调

图 7-25：企业 Agent 生态的权限架构

7.6.5 代表性场景分析

案例 1：高权限采购智能体的失控（教学化场景）

背景：以下为基于真实工程风险抽象出的代表性场景，用来说明如果没有 Rule of Two，采购类智能体为何容易造成高成本失误。

事件：

• Agent 无意中购买了错误的云服务配额

• 由于权限过于宽泛，没有价格上限检查

• 造成月度成本增加 300 万美元

根本原因：

• 缺少 Rule of Two 验证

• 采购权限设置不合理（无预算上限）

• 没有及时的支出告警

改进方案：

class SafeProcurementAgent:
    """改进的采购 Agent"""

    # 权限最小化
    PERMISSION_LEVEL = PermissionLevel.LIMITED
    BUDGET_LIMIT = 10_000  # 单次上限 $10k
    MONTHLY_LIMIT = 100_000  # 月度上限 $100k

    def purchase(self, service, cost, duration):
        """采购操作的改进流程"""

        # Check 1: 基本权限检查
        if not self.has_purchase_permission():
            raise PermissionDenied("No purchase permission")

        # Check 2: 预算检查
        if cost > self.BUDGET_LIMIT:
            logger.warning(f"Cost {cost} exceeds limit {self.BUDGET_LIMIT}")
            return self.request_human_approval(cost, reason="Exceeds single-purchase budget limit")

        if self.monthly_spent + cost > self.MONTHLY_LIMIT:
            return self.request_human_approval(cost, reason="Exceeds monthly budget limit")

        # Check 3: 业务逻辑验证
        if not self.validate_business_case(service):
            raise ValidationError(f"No valid business case for {service}")

        # Check 4: 第一个 Agent 处理
        approval1 = self.process_with_agent1(service, cost)

        # Check 5: 第二个 Agent 独立验证
        approval2 = self.process_with_agent2(service, cost)

        # 双 Agent 复核（监督网关的一种实现，非 Rule of Two 本身）：两个都通过才能执行
        if approval1.approved and approval2.approved:
            return self.execute_purchase(service, cost, duration)
        else:
            logger.warning("Purchase rejected by second review")
            return None

    def request_human_approval(self, cost, reason):
        """大额采购需要人类确认"""
        ticket = self.create_approval_ticket(
            amount=cost,
            reason=reason,
            auto_expire_hours=24
        )
        # 等待人类审批；获批后实际系统应重新进入 execute_purchase 执行流程（此处省略）
        return ticket.wait_for_approval(timeout=24*3600)

案例 2：代码审查 Agent 误操作的连锁反应（假设场景）

背景：以下为基于真实工程风险构造的典型场景示例。一个代码审查 Agent 被给予代码合并权限，在主分支上引入了漏洞。

事件流：

1. Agent 试图修复一个 bug（权限：修改代码）

2. 修复引入了一个微妙的逻辑错误

3. Agent 通过了自己的测试（测试用例不全）

4. Agent 自动合并了代码到主分支（权限：合并代码）

5. 在生产环境引起故障

防御失败点：

• 只有 Agent 的单一验证，缺少第二层

• 自动合并权限过于宽泛

改进的 Code Review Agent：

class SafeCodeReviewAgent:
    """改进的代码审查和合并 Agent"""

    def review_and_merge(self, pull_request):
        """代码审查和合并的改进流程"""

        # Step 1: 自动代码审查
        review1 = self.perform_code_review(pull_request)

        if review1.has_critical_issues:
            return self.reject_pr(pull_request, review1.issues)

        # Step 2: 触发第二个 Agent 的独立审查
        review2 = self.trigger_secondary_review(pull_request)

        # Step 3: 运行测试套件
        test_results = self.run_tests(pull_request)

        if not test_results.passed:
            return self.request_human_review(
                pr=pull_request,
                reason="Tests failed",
                test_results=test_results
            )

        # Step 4: 双 Agent 复核检查（监督网关，非 Rule of Two 本身）
        if review1.approved and review2.approved:
            # 对于大型变更，仍需人类最终确认
            if self.is_large_change(pull_request):
                return self.request_human_approval(pull_request)
            else:
                return self.merge_with_audit_log(pull_request)
        else:
            return self.request_human_review(
                pr=pull_request,
                review1=review1,
                review2=review2
            )

    def is_large_change(self, pr):
        """评估变更规模"""
        return (
            pr.files_changed > 10 or
            pr.lines_added > 500 or
            pr.affects_critical_path()
        )

7.6.6 Rule of Two 的适用范围与经验总结

系统化的 Rule of Two 应用清单

基于 2025-2026 年的安全事件，形成了以下应用清单：

图 7-26：Rule of Two 应用场景清单

经验总结

Rule of Two 更适合被理解为 高风险操作的约束框架，而不是固定的“双 Agent 审批模板”或某一组通用量化指标。不同组织的收益、误操作率和审批成本会因业务类型、权限模型和自动化程度而显著不同，因此更稳妥的做法是把它作为设计原则，再结合本组织的监控数据选择实现方式。

7.6.7 实现 Rule of Two 的技术路径

技术选择决策树

# 伪代码：实现选择
def choose_implementation(operation_type, system_scale):
    """选择合适的 Rule of Two 实现"""

    if operation_type == "IRREVERSIBLE":
        if system_scale == "small":
            # 小型系统可以用简单的人工审批
            return HumanApprovalGateway()
        elif system_scale == "medium":
            # 中型系统：自动化初步检查 + 人类审批
            return AutomatedCheckWithHumanApproval()
        else:  # large
            # 大型系统：双重验证通道 + 异常时人类介入
            return DualAgentVerification()

开源方案

推荐的开源工具与框架：

1. LLM 应用框架
   - LangChain: 链式调用管理
   - LangGraph / AutoGen / CrewAI: 多智能体编排与工作流
   - LlamaIndex: RAG 框架

2. 权限和审批
   - Open Policy Agent (OPA): 声明式权限策略
   - Keycloak: 身份与访问管理
   - Kyverno: Kubernetes 策略引擎

3. 审计和监控
   - Falco: 行为监控
   - Prometheus: 指标收集
   - ELK Stack: 日志分析

7.6.8 与其他 AI 安全实践的综合

图 7-27：Rule of Two 在 AI 安全防线中的位置