多Agent韧性六层：从熔断到补偿事务的实测全账

from tenacity import retry, stop_after_attempt, wait_exponential_jitter

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential_jitter(initial=1, max=60, jitter=2),
    before_sleep=log_retry_attempt
)
async def call_agent_with_retry(agent, task):
    """带指数退避的Agent调用"""
    response = await agent.run(task)
    if response.is_error and response.error.is_transient:
        raise TransientError(response.error)  # 触发重试
    if response.is_error:
        raise PermanentError(response.error)  # 不重试，直接上抛
    return response

import hashlib

def agent_aware_backoff(agent_id: str, attempt: int) -> float:
    base = min(2 ** attempt, 60)
    seed = int(hashlib.md5(agent_id.encode()).hexdigest()[:8], 16)
    jitter = (seed % 100) / 100.0 * base * 0.4  # ±20%基于Agent ID
    return base + jitter - base * 0.2

CLOSED → OPEN → HALF-OPEN → PROBE → CLOSED/OPEN
  ↓                                    ↑
  └── 正常请求通过 ── 故障阈值触发 ──→  OPEN
                                       ↓
                            定时器到期 → HALF-OPEN（放行少量请求）
                                       ↓
                              探测成功 → PROBE（逐步放量）
                                       ↓
                              全部成功 → CLOSED
                              任一失败 → OPEN

from circuitbreaker import CircuitBreaker, CircuitBreakerError

safety_checker_breaker = CircuitBreaker(
    failure_threshold=3,
    recovery_timeout=30,
    expected_exception=ToolCallError,
    fallback_function=fallback_safety_check  # 降级函数
)

@safety_checker_breaker
async def call_safety_agent(content):
    """调用安全检查Agent，带熔断保护"""
    return await safety_agent.check(content)

async def fallback_safety_check(content):
    """熔断时的降级策略"""
    if os.getenv("SAFETY_FAIL_MODE") == "open":
        # 开放模式：跳过安全检查，记录告警
        await alert_service.warn("Safety agent circuit open, allowing content")
        return SafetyResult(passed=True, degraded=True)
    else:
        # 闭合模式：拒绝所有请求
        return SafetyResult(passed=False, reason="Safety check unavailable")

class CascadingCircuitBreaker:
    def __init__(self, agent_chain: list[str]):
        self.breakers = {agent: CircuitBreaker() for agent in agent_chain}
        self.chain = agent_chain

    async def execute(self, task):
        for agent_name in self.chain:
            breaker = self.breakers[agent_name]
            if breaker.state == "open":
                # 跳过后续所有依赖该Agent的步骤
                return await self.short_circuit(agent_name, task)
            try:
                task = await breaker.call(self.agents[agent_name].run, task)
            except CircuitBreakerError:
                return await self.short_circuit(agent_name, task)
        return task

from datetime import datetime
from pydantic import BaseModel

class DeadLetter(BaseModel):
    task_id: str
    agent_id: str
    original_task: dict
    failure_reason: str
    failure_count: int
    last_failure_at: datetime
    suggested_strategy: str  # "replay" | "escalate" | "drop"
    metadata: dict

class DeadLetterQueue:
    def __init__(self, backend="redis"):  # redis | kafka | sqs
        self.backend = backend
        self.handlers = {
            "replay": self._handle_replay,
            "escalate": self._handle_escalate,
            "drop": self._handle_drop,
        }

    async def enqueue(self, letter: DeadLetter):
        # 根据任务价值自动选择策略
        if letter.metadata.get("task_value", 0) > 1000:
            letter.suggested_strategy = "escalate"
        elif letter.failure_count < 5:
            letter.suggested_strategy = "replay"
        else:
            letter.suggested_strategy = "drop"
        await self._store(letter)

    async def replay_all(self, agent_id: str = None):
        """修复故障后批量重放"""
        letters = await self._fetch_pending(agent_id)
        results = {"replayed": 0, "failed_again": 0}
        for letter in letters:
            try:
                await self.agent_pool.route(letter.original_task)
                results["replayed"] += 1
            except Exception:
                letter.failure_count += 1
                await self.enqueue(letter)  # 重新入DLQ
                results["failed_again"] += 1
        return results

class TravelBookingSaga:
    """旅行预订多Agent Saga编排器"""

    steps = [
        {"agent": "flight_agent", "action": "book_flight",
         "compensate": "cancel_flight"},
        {"agent": "hotel_agent", "action": "book_hotel",
         "compensate": "cancel_hotel"},
        {"agent": "car_agent", "action": "book_car",
         "compensate": "cancel_car"},
        {"agent": "payment_agent", "action": "charge_payment",
         "compensate": "refund_payment"},
    ]

    async def execute(self, booking_request):
        completed = []
        try:
            for step in self.steps:
                result = await self.agents[step["agent"]].run(
                    action=step["action"],
                    context=booking_request
                )
                if result.is_error:
                    raise SagaStepError(step, result.error)
                completed.append((step, result))
            return SagaResult(success=True, steps=completed)

        except SagaStepError as e:
            # 反向执行补偿链
            for step, _ in reversed(completed):
                try:
                    await self.agents[step["agent"]].run(
                        action=step["compensate"],
                        context=booking_request
                    )
                except Exception as comp_err:
                    # 补偿也失败了！进入人工介入
                    await self.escalate_compensation_failure(
                        step, comp_err
                    )
            return SagaResult(success=False, compensated=True)

L5: GPT-4o → Claude → Gemini → 本地小模型 → 规则引擎 → 静态兜底
    ↓ 失败     ↓ 失败    ↓ 失败    ↓ 失败      ↓ 失败
    正常质量   几乎无损   轻微下降   明显下降    可用但僵化

class DegradationChain:
    """多Agent系统优雅降级链"""

    def __init__(self):
        self.chain = [
            {"name": "claude-opus", "model": "claude-opus-4-8",
             "min_quality": 9.0, "fallback_on": ["rate_limit", "timeout"]},
            {"name": "gpt-4o", "model": "gpt-4o",
             "min_quality": 8.5, "fallback_on": ["rate_limit", "timeout"]},
            {"name": "claude-haiku", "model": "claude-haiku-4-5",
             "min_quality": 7.0, "fallback_on": ["rate_limit", "timeout"]},
            {"name": "local-llama", "model": "llama-3.3-70b",
             "min_quality": 5.0, "fallback_on": []},
            {"name": "rule-engine", "model": None,
             "min_quality": 3.0, "fallback_on": []},
        ]
        self.current_level = 0

    async def execute(self, agent, task):
        for i in range(self.current_level, len(self.chain)):
            level = self.chain[i]
            try:
                result = await agent.run(task, model_override=level["model"])
                self.current_level = i  # 更新当前级别
                result.metadata["degradation_level"] = i
                result.metadata["model_used"] = level["name"]
                return result
            except Exception as e:
                if self._should_fallback(e, level):
                    await self._log_degradation(i, e)
                    continue
                raise
        return self._static_fallback(task)

    def _should_fallback(self, error, level):
        return any(
            error.type == t for t in level["fallback_on"]
        ) or level["fallback_on"] == []

class WatchdogAgent:
    """独立的监控Agent，负责诊断和修复其他Agent"""

    async def monitor_loop(self, agent_pool):
        while True:
            # 1. 收集所有Agent的健康指标
            health_map = await self.collect_health(agent_pool)

            # 2. 识别异常Agent
            anomalies = self.detect_anomalies(health_map)

            for anomaly in anomalies:
                # 3. 诊断根因
                diagnosis = await self.diagnose(anomaly)

                # 4. 生成修复方案
                fix = await self.generate_fix(diagnosis)

                # 5. 执行修复（带人工审批门控）
                if fix.confidence > 0.85:
                    result = await self.apply_fix(fix)
                else:
                    await self.request_human_approval(fix)

                # 6. 验证修复效果
                verified = await self.verify_fix(anomaly.agent_id)

                # 7. 记录到知识库
                await self.update_knowledge_base(
                    anomaly, diagnosis, fix, verified
                )

            await asyncio.sleep(self.check_interval)

class FixBudget:
    """限制自愈行为的频率和范围"""

    def __init__(self, max_fixes_per_hour=5, max_simultaneous=2):
        self.max_fixes_per_hour = max_fixes_per_hour
        self.max_simultaneous = max_simultaneous
        self.fix_history = []

    def can_fix(self, agent_id: str) -> bool:
        recent = [f for f in self.fix_history
                  if f["time"] > datetime.now() - timedelta(hours=1)]
        if len(recent) >= self.max_fixes_per_hour:
            return False  # 超预算，转人工
        same_agent = [f for f in recent if f["agent_id"] == agent_id]
        if len(same_agent) >= 2:
            return False  # 同一Agent 1小时内已修2次，转人工
        return True

graph TB
    subgraph "请求入口"
        REQ[用户请求/API调用]
    end

    subgraph "L1 重试+退避层"
        R1[Retry Controller]
        R1 -->|瞬态故障| R2[指数退避<br/>initial=1s, max=60s]
        R2 -->|Agent-aware Jitter| R3[退避调度器]
    end

    subgraph "L2 熔断器层"
        CB[Circuit Breaker<br/>CLOSED→OPEN→HALF-OPEN→PROBE]
        CB -->|共享状态| REDIS[(Redis/etcd<br/>熔断状态存储)]
        CB -->|级联传播| CASCADE[链式熔断<br/>Agent依赖链]
    end

    subgraph "L3 死信队列"
        DLQ[Dead Letter Queue]
        DLQ -->|高价值| ESC[转人工队列]
        DLQ -->|可恢复| REPLAY[重放入口]
        DLQ -->|低价值| DROP[丢弃+审计日志]
    end

    subgraph "L4 Saga补偿"
        SAGA[Saga Orchestrator]
        SAGA -->|编排式| ORCH[中央协调器<br/>3-15 Agents]
        SAGA -->|协同式| CHORE[事件驱动<br/>15-100+ Agents]
        ORCH -->|补偿失败| MANUAL[人工介入]
        CHORE -->|补偿失败| MANUAL
    end

    subgraph "L5 优雅降级"
        DEG[Degradation Chain]
        DEG --> L0[Claude Opus 9.2分]
        DEG --> L1[GPT-4o 8.8分]
        DEG --> L2[Gemini 8.5分]
        DEG --> L3[Haiku 7.2分]
        DEG --> L4[Local Llama 6.1分]
        DEG --> L5[规则引擎 4.5分]
    end

    subgraph "L6 自愈恢复"
        WD[Watchdog Agent]
        WD -->|检测| MON[异常检测]
        WD -->|诊断| DIAG[根因分析]
        WD -->|修复| FIX[自动修复<br/>Fix Budget限速]
        WD -->|验证| VERIFY[效果验证]
        WD -->|学习| KB[(知识库)]
    end

    REQ --> R1
    R1 -->|重试耗尽| CB
    CB -->|熔断打开| DLQ
    R1 -->|多步事务| SAGA
    SAGA -->|模型故障| DEG
    WD -.->|监控| R1
    WD -.->|监控| CB
    WD -.->|监控| DLQ
    WD -.->|监控| SAGA

    style REQ fill:#4CAF50,color:#fff
    style DLQ fill:#FF5722,color:#fff
    style SAGA fill:#9C27B0,color:#fff
    style WD fill:#FF9800,color:#fff