自主Agent六场生产灾难与熔断引擎全拆解

# 灾难现场——没有熔断的Agent Loop
while not task_complete:
    result = agent.run(task)
    if result.status == "rate_limited":
        task = "retry: " + task  # 重新包装成新任务，绕过了retry计数器
        # 没有max_iterations检查
        # 没有circuit breaker
        # 没有token预算检查
        continue

# 第一层：迭代计数熔断
class IterationCircuitBreaker:
    def __init__(self, max_iterations=25, window_minutes=60):
        self.max_iterations = max_iterations
        self.iterations = []

    def check(self):
        recent = [i for i in self.iterations 
                  if time.time() - i < self.window_minutes * 60]
        if len(recent) >= self.max_iterations:
            raise CircuitBreakerTripped(
                f"Agent exceeded {self.max_iterations} iterations "
                f"in {self.window_minutes} minutes"
            )
        self.iterations.append(time.time())

# 第二层：语义重复检测（检测Agent是否在做同样的事）
class SemanticDuplicateDetector:
    def __init__(self, similarity_threshold=0.85):
        self.history = []
        self.threshold = similarity_threshold

    def is_looping(self, current_action):
        embeddings = embed([current_action] + self.history[-5:])
        similarities = cosine_similarity(embeddings[0:1], embeddings[1:])
        if max(similarities[0]) > self.threshold:
            return True  # Agent在做重复的事
        self.history.append(current_action)
        return False

# 第三层：Token预算硬熔断
class TokenBudgetHardStop:
    def __init__(self, max_tokens=500000, alert_at=0.7):
        self.max_tokens = max_tokens
        self.consumed = 0

    def pre_call_check(self, estimated_tokens):
        if self.consumed + estimated_tokens > self.max_tokens:
            raise BudgetExhausted(
                f"Token budget exhausted: {self.consumed}/{self.max_tokens}"
            )
        if self.consumed > self.max_tokens * self.alert_at:
            send_alert(f"Token budget at {self.consumed/self.max_tokens:.0%}")

# Claude Code实际执行的命令（简化还原）
$ terraform destroy -target=aws_db_instance.production \
    -auto-approve  # Agent自动确认，无需人工审批

# Destroy complete! Resources: 1 destroyed.
# 2.5年数据，永久丢失

# 第一层：破坏性操作正则拦截
DESTRUCTIVE_PATTERNS = [
    r'terraform\s+destroy',
    r'DROP\s+(TABLE|DATABASE|SCHEMA)',
    r'rm\s+-rf\s+/',
    r'delete\s+--recursive',
    r'TRUNCATE\s+TABLE',
    r'kubectl\s+delete\s+--all',
]

def pre_execution_check(command: str) -> ApprovalDecision:
    for pattern in DESTRUCTIVE_PATTERNS:
        if re.search(pattern, command, re.IGNORECASE):
            return ApprovalDecision(
                requires_human=True,
                reason=f"Destructive operation detected: {pattern}",
                severity="CRITICAL"
            )
    return ApprovalDecision(requires_human=False)

# 第二层：环境隔离沙箱
class EnvironmentIsolation:
    def __init__(self):
        self.prod_indicators = ['prod', 'production', 'live', 'main']

    def check_credential_scope(self, env_vars):
        exposed = []
        for key, value in env_vars.items():
            if any(ind in key.lower() for ind in self.prod_indicators):
                exposed.append(key)
        if exposed:
            raise ProductionCredentialExposure(
                f"Production credentials detected: {exposed}. "
                f"Refusing to execute in non-isolated environment."
            )

# 第三层：LangGraph interrupt()审批门控
from langgraph.types import interrupt

def execute_with_approval(state):
    command = state["pending_command"]
    if is_destructive(command):
        # 冻结执行，等待人类审批
        approval = interrupt({
            "message": f"Destructive operation requires approval",
            "command": command,
            "risk_level": "CRITICAL"
        })
        if not approval["approved"]:
            return {"status": "rejected", "reason": approval["reason"]}
    return execute_sandboxed(command)

# 第四层：影子执行+差异比对（Shadow Execution）
class ShadowExecutor:
    def execute(self, command):
        # 在影子环境中执行，返回diff而非实际效果
        shadow_result = self.shadow_env.run(command)
        diff = self.compare_with_live(shadow_result)
        return DiffReport(
            resources_affected=len(diff.deleted),
            data_at_risk=diff.estimate_data_loss(),
            reversible=diff.has_backup()
        )

# 第一层：任务驱动的临时权限（Task-Scoped Tokens）
class TaskScopedAuthorizer:
    def authorize(self, agent_id, task_description):
        # 从任务描述中提取所需权限
        required_perms = self.analyze_required_permissions(task_description)
        # 生成限时、限权Token
        token = self.issue_token(
            agent_id=agent_id,
            permissions=required_perms,  # 只授予最小权限
            ttl_minutes=30,             # 30分钟过期
            scope="single_task"         # 单次任务
        )
        return token

# 第二层：操作风险评估（Risk-Aware Execution）
class RiskAwareExecutor:
    RISK_MATRIX = {
        "read_config": "LOW",
        "update_config": "MEDIUM",
        "delete_resource": "HIGH",
        "modify_network": "HIGH",
        "alter_security_group": "CRITICAL"
    }

    def execute(self, operation, environment):
        risk = self.RISK_MATRIX.get(operation.type, "UNKNOWN")
        if risk == "CRITICAL" and environment == "production":
            # 强制要求双人审批 + 快照
            snapshot = self.create_snapshot(environment)
            approval = self.request_dual_approval(operation, snapshot)
            if not approval.granted:
                raise OperationRejected(approval.reason)
        self.execute_with_audit_trail(operation)

# 第三层：自动回滚（Auto-Rollback Guardian）
class AutoRollbackGuardian:
    def __init__(self, snapshot_ttl_hours=24):
        self.snapshots = {}
        self.ttl = snapshot_ttl_hours

    def pre_execute(self, environment):
        snapshot_id = self.create_snapshot(environment)
        self.snapshots[snapshot_id] = {
            "created": datetime.now(),
            "environment": environment
        }
        return snapshot_id

    def on_failure(self, snapshot_id):
        # 自动回滚到执行前状态
        self.restore_snapshot(snapshot_id)
        self.notify_team(f"Auto-rollback executed for {snapshot_id}")

# 第一层：滚动摘要压缩（Rolling Summary Compaction）
class RollingCompactor:
    def compact(self, messages, max_context_tokens=8000):
        total = count_tokens(messages)
        if total < max_context_tokens * 0.7:
            return messages  # 还有空间

        # 将前60%的消息压缩为摘要
        older = messages[:int(len(messages) * 0.6)]
        recent = messages[int(len(messages) * 0.6):]

        summary = self.llm_summarize(older)
        return [
            {"role": "system", "content": f"Previous context: {summary}"},
            *recent
        ]

# 第二层：外部记忆存储（Context Spillover）
class ExternalMemoryStore:
    def __init__(self, backend="redis"):
        self.store = Redis() if backend == "redis" else ChromaDB()

    def offload(self, key, data):
        """将不立即需要的上下文卸载到外部存储"""
        self.store.set(key, json.dumps(data))

    def retrieve(self, query, top_k=3):
        """按需检索相关上下文"""
        return self.store.similarity_search(query, k=top_k)

# 第三层：任务一致性检查（Checkpoint Validation）
class CheckpointValidator:
    def validate_alignment(self, original_goal, current_state, step_number):
        prompt = f"""
        Original goal: {original_goal}
        Current state after {step_number} steps: {current_state}
        Are we still on track? Answer YES or NO with reasoning.
        """
        alignment = self.llm.evaluate(prompt)
        if alignment.confidence < 0.7:
            raise TaskDriftDetected(
                f"Agent drifted at step {step_number}: {alignment.reasoning}"
            )
        return alignment

# 第四层：工具结果清理（Tool Result Pruning）
class ToolResultPruner:
    def prune(self, tool_output, max_tokens=2000):
        if count_tokens(tool_output) > max_tokens:
            # 只保留摘要和关键字段
            pruned = self.extract_essential(tool_output)
            return f"[Truncated. Key info: {pruned}]"
        return tool_output

# 第一层：任务级Token预算（Task-Level Budget）
class TaskBudget:
    def __init__(self, max_tokens_per_task=100000):
        self.max = max_tokens_per_task
        self.consumed = 0

    def pre_call(self, estimated_tokens):
        projected = self.consumed + estimated_tokens
        if projected > self.max:
            return BudgetDecision(
                action="ESCALATE",  # 升级到更便宜的模型或人工
                reason=f"Task budget at {self.consumed/self.max:.0%}",
                suggested_model="gpt-4o-mini"  # 降级到更便宜的模型
            )
        return BudgetDecision(action="PROCEED")

# 第二层：模型路由（Cost-Aware Model Router）
class CostAwareRouter:
    def route(self, task_complexity, budget_remaining):
        if budget_remaining < 0.3 and task_complexity == "simple":
            return "gpt-4o-mini"   # 低预算+简单任务 → 便宜模型
        elif task_complexity == "complex":
            return "claude-opus"   # 复杂任务 → 强力模型
        else:
            return "claude-sonnet" # 默认 → 平衡模型

# 第三层：实时成本仪表盘（Real-Time Cost Dashboard）
class CostMonitor:
    def __init__(self, alert_thresholds=[0.5, 0.7, 0.9, 0.95]):
        self.thresholds = alert_thresholds
        self.alerted = set()

    def track(self, token_count, cost_usd):
        self.total_cost += cost_usd
        for threshold in self.thresholds:
            if self.total_cost > self.monthly_budget * threshold:
                if threshold not in self.alerted:
                    self.send_alert(threshold)
                    self.alerted.add(threshold)

# 第四层：语义缓存（Semantic Caching）
# 第五层：工具描述压缩（Tool Description Compression）
# 第六层：审批强制升级（Budget-Gated Escalation）

# 全局委托追踪器（Global Delegation Tracker）
class DelegationTracker:
    def __init__(self, max_depth=3, max_same_issue=2):
        self.delegation_graph = {}
        self.max_depth = max_depth
        self.max_same_issue = max_same_issue

    def check_delegation(self, from_agent, to_agent, issue_hash):
        # 检查委托深度
        depth = self.get_depth(from_agent)
        if depth >= self.max_depth:
            raise DelegationLimitExceeded(
                f"Delegation chain too deep: {depth} levels"
            )

        # 检查同一问题的往返次数
        key = f"{from_agent}→{to_agent}:{issue_hash}"
        self.delegation_graph[key] = self.delegation_graph.get(key, 0) + 1
        if self.delegation_graph[key] >= self.max_same_issue:
            raise CircularDelegation(
                f"Agents {from_agent} and {to_agent} are looping on "
                f"the same issue (hash: {issue_hash[:8]})"
            )

        return DelegationAllowed(depth=depth + 1)

    def get_depth(self, agent_id):
        return max(
            [v for k, v in self.delegation_graph.items() 
             if agent_id in k.split("→")[0]],
            default=0
        )

# 全局Kill Switch（紧急制动）
class GlobalKillSwitch:
    def __init__(self):
        self.engaged = False
        self.reason = None

    def emergency_stop(self, reason, agent_group=None):
        self.engaged = True
        self.reason = reason
        # 级联终止所有相关Agent
        if agent_group:
            for agent in agent_group:
                agent.terminate(grace_period_seconds=5)
        # 捕获所有Agent的当前状态
        state_dump = self.capture_all_states()
        self.persist(state_dump)
        self.notify_oncall(reason)

graph TB
    subgraph "用户请求入口"
        REQ[任务请求]
    end

    subgraph "第一层：Token预算熔断"
        BUDGET[Token预算检查器]
        ROUTER[成本感知模型路由]
        BUDGET --> ROUTER
    end

    subgraph "第二层：操作审批门控"
        APPROVAL[破坏性操作拦截]
        ENV_ISO[环境隔离沙箱]
        APPROVAL --> ENV_ISO
    end

    subgraph "第三层：权限沙箱"
        PERM[任务驱动临时权限]
        RISK[风险评估引擎]
        ROLLBACK[自动回滚守护]
        PERM --> RISK --> ROLLBACK
    end

    subgraph "第四层：上下文管理"
        COMPACT[滚动摘要压缩]
        EXT_MEM[外部记忆存储]
        CHECKPOINT[检查点验证]
        COMPACT --> EXT_MEM --> CHECKPOINT
    end

    subgraph "第五层：循环检测"
        ITER_CB[迭代计数熔断]
        SEM_DUP[语义重复检测]
        DELEG[委托深度追踪]
        ITER_CB --> SEM_DUP --> DELEG
    end

    subgraph "第六层：紧急制动"
        KILL[全局Kill Switch]
        AUDIT[不可变审计日志]
        KILL --> AUDIT
    end

    REQ --> BUDGET
    ROUTER --> APPROVAL
    ENV_ISO --> PERM
    RISK --> COMPACT
    CHECKPOINT --> ITER_CB
    DELEG --> KILL

    style KILL fill:#ff4444,color:#fff
    style BUDGET fill:#ff9900,color:#fff
    style APPROVAL fill:#ff6600,color:#fff
    style PERM fill:#ffaa00,color:#fff
    style COMPACT fill:#44bb44,color:#fff
    style ITER_CB fill:#4488ff,color:#fff