自治度分级收敛：五框架对齐与影子→代驾四阶实测

graph TB
    subgraph 五框架自治度收敛
        direction TB
        S1["影子模式<br/>Shadow Mode"] --> S2["副驾驶<br/>Copilot"]
        S2 --> S3["监督执行<br/>Supervised"]
        S3 --> S4["代驾/自治<br/>Autonomous"]
    end

    subgraph Anthropic
        A1["低自治 1-3<br/>逐条审批"] -.-> S1
        A2["中自治 4-6<br/>关键节点确认"] -.-> S2
        A3["中高自治 7-8<br/>事后审查"] -.-> S3
        A4["高自治 9-10<br/>连续自主"] -.-> S4
    end

    subgraph KnightColumbia["Knight Columbia"]
        K1["L1 Operator"] -.-> S1
        K2["L2 Supervisor"] -.-> S2
        K3["L3 Overseen"] -.-> S3
        K4["L4-L5 Delegate+"] -.-> S4
    end

    subgraph CSA
        C1["L0-L1 规则响应"] -.-> S1
        C2["L2 半自治"] -.-> S2
        C3["L3 条件自治"] -.-> S3
        C4["L4-L5 高/全自治"] -.-> S4
    end

    subgraph Uber
        U1["L1 响应者"] -.-> S1
        U2["L2 助手"] -.-> S2
        U3["L3-L4 协作者"] -.-> S3
        U4["L5 自治团队"] -.-> S4
    end

class ShadowModeController:
    """影子模式控制器：Agent决策只记录不执行"""

    def __init__(self, agent, confidence_threshold=0.85):
        self.agent = agent
        self.confidence_threshold = confidence_threshold
        self.shadow_log = RedisStream("agent:shadow:log")
        self.metrics = {
            "total_decisions": 0,
            "matches_human": 0,
            "divergences": 0,
            "avg_confidence": 0.0
        }

    async def process_task(self, task: Task):
        # 人类执行实际操作
        human_result = await self.human_operator.execute(task)

        # Agent并行做出决策（不执行）
        agent_decision = await self.agent.plan(task)

        # 记录并比对
        divergence = self.compute_divergence(
            agent_decision, human_result
        )

        self.shadow_log.add({
            "task_id": task.id,
            "agent_decision": agent_decision.to_dict(),
            "human_result": human_result.to_dict(),
            "divergence_score": divergence,
            "agent_confidence": agent_decision.confidence,
            "timestamp": datetime.utcnow().isoformat()
        })

        self.update_metrics(divergence, agent_decision.confidence)

        # 检查是否满足晋升条件
        if self.check_promotion_readiness():
            return PromotionSignal(
                current_tier="shadow",
                target_tier="copilot",
                evidence=self.get_promotion_evidence()
            )

    def check_promotion_readiness(self) -> bool:
        """晋升条件：连续1000次执行，准确率>99%"""
        if self.metrics["total_decisions"] < 1000:
            return False
        accuracy = self.metrics["matches_human"] / self.metrics["total_decisions"]
        return accuracy >= 0.99

class CopilotModeController:
    """副驾驶模式：Agent执行但关键操作需人类确认"""

    # 按风险等级定义需要确认的操作类型
    CONFIRMATION_REQUIRED = {
        "file_write": RiskLevel.HIGH,
        "api_call": RiskLevel.HIGH,
        "shell_exec": RiskLevel.CRITICAL,
        "db_write": RiskLevel.CRITICAL,
        "code_search": RiskLevel.LOW,    # 低风险，自动通过
        "file_read": RiskLevel.LOW,
    }

    def __init__(self, agent, human_approver):
        self.agent = agent
        self.approver = human_approver
        self.approval_stats = defaultdict(lambda: {
            "approved": 0, "rejected": 0, "modified": 0
        })

    async def execute_step(self, step: ActionStep):
        risk = self.CONFIRMATION_REQUIRED.get(
            step.action_type, RiskLevel.MEDIUM
        )

        if risk in (RiskLevel.HIGH, RiskLevel.CRITICAL):
            # 发送审批请求
            decision = await self.approver.review(step)
            self.approval_stats[step.action_type][decision] += 1

            if decision == "rejected":
                return StepResult(status="blocked", reason=decision)
            elif decision == "modified":
                step = await self.approver.get_modified_step(step)
        else:
            # 低风险操作自动通过
            pass

        return await self.agent.execute(step)

    def get_first_pass_accept_rate(self) -> float:
        """首次通过接受率 —— 晋升到监督模式的关键指标"""
        total = sum(
            s["approved"] + s["rejected"] + s["modified"]
            for s in self.approval_stats.values()
        )
        approved = sum(s["approved"] for s in self.approval_stats.values())
        return approved / total if total > 0 else 0.0

class SupervisedModeController:
    """监督模式：Agent自主执行，自治度随置信度动态衰减"""

    def __init__(self, agent, config):
        self.agent = agent
        self.confidence_scorer = ConfidenceScorer(config)
        self.escalation_policy = config.escalation_policy
        # 自治度衰减函数：confidence越低，衰减越快
        self.decay_function = AutonomyDecayFunction(
            base_threshold=0.7,
            decay_rate=0.1,
            recovery_rate=0.05
        )
        self.current_autonomy = 1.0  # 起始满自治度
        self.audit_trail = []

    async def execute_task(self, task: Task):
        steps = await self.agent.plan(task)

        for step in steps:
            confidence = await self.confidence_scorer.score(step)

            # 动态调整自治度
            self.current_autonomy = self.decay_function.update(
                self.current_autonomy, confidence
            )

            # 根据当前自治度决定执行策略
            if self.current_autonomy > 0.8:
                # 高自治：直接执行，记录审计日志
                result = await self.agent.execute(step)
                self.audit_trail.append(AuditEntry(
                    step=step, result=result,
                    confidence=confidence,
                    autonomy_at_execution=self.current_autonomy,
                    mode="autonomous"
                ))
            elif self.current_autonomy > 0.5:
                # 中自治：执行但标记为需要事后审查
                result = await self.agent.execute(step)
                self.audit_trail.append(AuditEntry(
                    step=step, result=result,
                    confidence=confidence,
                    autonomy_at_execution=self.current_autonomy,
                    mode="execute_with_review"
                ))
            else:
                # 低自治：升级到人类，回退到副驾驶模式
                await self.escalate_to_human(step, confidence)
                # 自治度恢复机制：人类介入后逐步回升
                self.decay_function.trigger_recovery()

        # 事后审查
        review_results = await self.post_execution_review()
        return self.compile_results(review_results)


class AutonomyDecayFunction:
    """自治度衰减函数"""

    def __init__(self, base_threshold, decay_rate, recovery_rate):
        self.base = base_threshold
        self.decay_rate = decay_rate
        self.recovery_rate = recovery_rate
        self.consecutive_low_confidence = 0

    def update(self, current: float, confidence: float) -> float:
        if confidence < self.base:
            # 低置信度触发衰减
            self.consecutive_low_confidence += 1
            decay = self.decay_rate * self.consecutive_low_confidence
            return max(0.0, current - decay)
        else:
            # 高置信度缓慢恢复
            self.consecutive_low_confidence = max(
                0, self.consecutive_low_confidence - 1
            )
            return min(1.0, current + self.recovery_rate)

    def trigger_recovery(self):
        """人类介入后触发快速恢复"""
        self.consecutive_low_confidence = 0

class AutonomousModeController:
    """代驾模式：自治执行，SLO违规触发回退"""

    SLO_TARGETS = {
        "task_completion_rate": 0.95,    # 95%任务完成率
        "avg_steps_per_task": 8.5,       # 平均步骤数
        "escalation_rate": 0.05,         # 升级率<5%
        "silent_failure_rate": 0.01,     # 静默失败率<1%
        "mean_time_to_detect_failure": 30  # 30秒内检测失败
    }

    def __init__(self, agent, slo_monitor):
        self.agent = agent
        self.slo_monitor = slo_monitor
        self.kill_switch = KillSwitch()
        self.running_window = SlidingWindow(size=100)

    async def execute_task(self, task: Task):
        # 前置检查：当前SLO状态
        slo_status = self.slo_monitor.get_current_status()
        if not slo_status.all_within_target():
            # SLO违规，回退到监督模式
            return await self.fallback_to_supervised(task)

        # 执行任务
        result = await self.agent.execute(task)

        # 记录到滑动窗口
        self.running_window.add(result)

        # 检查是否需要触发kill switch
        window_failure_rate = self.running_window.failure_rate()
        if window_failure_rate > self.SLO_TARGETS["silent_failure_rate"] * 3:
            self.kill_switch.activate(
                reason=f"Failure rate {window_failure_rate:.3f} "
                       f"exceeds 3x threshold"
            )
            return await self.fallback_to_supervised(task)

        return result

    async def fallback_to_supervised(self, task: Task):
        """优雅降级到监督模式"""
        # 1. 停止接收新任务
        # 2. 完成当前任务队列
        # 3. 切换到监督模式
        # 4. 通知oncall
        return FallbackResult(
            new_mode="supervised",
            reason="SLO violation detected",
            pending_tasks=self.get_pending_count()
        )

A(t) = A₀ · e^(-λ·C_low(t)) + A_min

其中：
- A(t) = t时刻的自治度（0-1）
- A₀ = 初始自治度
- λ = 衰减速率（实测：0.1-0.3，取决于任务域）
- C_low(t) = 累计低置信度步数
- A_min = 最低自治度（通常为0.2-0.3，不完全归零）

graph LR
    subgraph 低自治 L1-L2
        G1["基础治理<br/>· 日志记录<br/>· 影子比对<br/>· 月度审查"]
    end

    subgraph 中自治 L3
        G2["增强治理<br/>· 实时可观测<br/>· 异常告警<br/>· 自动回滚<br/>· 周度审查"]
    end

    subgraph 高自治 L4-L5
        G3["严格治理<br/>· 形式化验证<br/>· SLO绑定<br/>· Kill Switch<br/>· 日度审查<br/>· 审计链"]
    end

    G1 -->|"晋升条件达成"| G2
    G2 -->|"SLO稳定2周"| G3
    G3 -->|"SLO违规"| G2

graph TD
    START["新Agent上线"] --> Q1{"是否有历史执行数据？"}
    Q1 -->|"否"| SHADOW["影子模式<br/>人机比 1:0<br/>人类100%执行"]
    Q1 -->|"是"| Q2{"历史准确率>99%？"}

    Q2 -->|"否"| SHADOW
    Q2 -->|"是"| COPILOT["副驾驶模式<br/>人机比 1:3<br/>人类审批关键步骤"]

    SHADOW -->|"1000次执行<br/>一致率>99%"| COPILOT
    COPILOT -->|"首次接受率>85%<br/>持续1周"| SUPERVISED["监督模式<br/>人机比 1:10<br/>人类事后审查"]

    SUPERVISED -->|"升级率<5%<br/>SLO达标2周"| AUTO["代驾模式<br/>人机比 1:50+<br/>人类仅SLO监控"]

    AUTO -->|"SLO违规3次/天"| SUPERVISED
    SUPERVISED -->|"升级率>15%"| COPILOT
    COPILOT -->|"接受率<70%"| SHADOW

    AUTO -->|"静默失败率>1%"| Q3{"可实时检测？"}
    Q3 -->|"是"| AUTO_FIX["加强检测后<br/>留在代驾模式"]
    Q3 -->|"否"| SUPERVISED

from agent_governance import AutonomyManager, TierPolicy

manager = AutonomyManager(
    policy=TierPolicy(
        tiers={
            "shadow": {"max_actions": 0, "require_approval": True},
            "copilot": {"max_actions": 3, "require_approval": True},
            "supervised": {"max_actions": 10, "require_approval": False},
            "autonomous": {"max_actions": -1, "require_approval": False}
        },
        promotion_criteria={
            "shadow->copilot": {"min_executions": 1000, "accuracy": 0.99},
            "copilot->supervised": {"min_executions": 500, "accept_rate": 0.85},
            "supervised->autonomous": {"min_executions": 200, "escalation_rate": 0.05}
        },
        demotion_triggers={
            "accuracy_drop": 0.05,   # 准确率下降5%
            "slo_violation": 3,       # 一天3次SLO违规
            "silent_failure": 0.02    # 静默失败率>2%
        }
    )
)