第三章：设计原则与方法论¶

class PermissionConstraint:
    """权限约束"""

    # 工具级约束:哪些工具可用
    available_tools = [
        "weather_api",
        "send_email",
        "retrieve_documents"
    ]

    # 资源级约束:可以访问哪些资源
    resource_access = {
        "database": {
            "tables": ["users", "orders"],  # 只能访问这些表
            "operations": ["read"],  # 只能读,不能写
            "row_limit": 1000  # 最多一次查询1000行
        },
        "file_system": {
            "paths": ["/data/public/"],  # 只能访问这个目录
            "operations": ["read"]
        }
    }

    # 用户级约束:可以操作哪些用户的数据
    user_scope = {
        "current_user_only": True,  # 只能操作当前用户的数据
        "allowed_org_ids": ["org-123", "org-456"]
    }

class OperationConstraint:
    """操作约束"""

    # 绝对禁止的操作
    forbidden_operations = [
        "delete_database_records",
        "modify_user_permissions",
        "export_user_data",
        "disable_audit_logging"
    ]

    # 受限制的操作(需要额外的验证或审批)
    restricted_operations = {
        "transfer_money": {
            "requires_approval": True,
            "approval_type": "human",
            "max_amount_per_operation": 10000,
            "max_amount_per_day": 100000
        },
        "send_communication": {
            "requires_approval": False,
            "rate_limit": "100_per_hour",
            "allowed_channels": ["email", "sms"]
        }
    }

    # 隐藏的危险操作(Agent应该意识不到它们的存在)
    hidden_operations = [
        "access_internal_debug_endpoints",
        "trigger_system_shutdown"
    ]

class TimeConstraint:
    """时间约束"""

    # 可用时间窗口
    availability_window = {
        "weekdays": {
            "start_hour": 9,
            "end_hour": 17,
            "timezone": "UTC"
        },
        "weekends": None  # 周末不可用
    }

    # 速率限制
    rate_limits = {
        "api_calls_per_minute": 60,
        "database_queries_per_minute": 100,
        "expensive_operations_per_hour": 10
    }

    # 过期时间
    expiration = {
        "credentials_expire_after_days": 90,
        "session_timeout_minutes": 60
    }

class DataConstraint:
    """数据约束"""

    # 数据大小限制
    size_limits = {
        "max_file_size_mb": 100,
        "max_api_response_size_mb": 10,
        "max_result_rows": 10000
    }

    # 数据敏感性处理
    sensitive_data_handling = {
        "pii_fields": ["email", "phone", "ssn"],
        "handling": "redact"  # 或 "encrypt", "deny"
    }

    # 数据流向限制
    allowed_data_destinations = [
        "internal_database",
        "approved_third_party_api"
    ]

    forbidden_data_destinations = [
        "public_cloud_storage",
        "external_email"
    ]

PROTECTED_FILES = [
    "/etc/passwd",
    "/etc/shadow",
    "~/.ssh/id_rsa",
    "~/.aws/credentials"
]

# 系统会在Agent尝试访问这些文件前拦截
if file_path in PROTECTED_FILES:
    raise PermissionError(f"Cannot access {file_path}")

DANGEROUS_PATTERNS = [
    r"DROP\s+TABLE",  # SQL注入
    r"rm\s+-rf",      # 破坏性命令
    r"format\s+[A-Z]:", # 格式化磁盘
    r"curl.*\|.*sh"   # 下载并执行
]

for pattern in DANGEROUS_PATTERNS:
    if re.search(pattern, agent_output):
        raise SecurityError(f"Dangerous pattern detected: {pattern}")

class WhitelistConstraint:
    """白名单:只有明确批准的操作才能进行"""

    def can_perform(self, operation: str) -> bool:
        return operation in self.allowed_operations

    allowed_operations = {
        "read_data",
        "send_notification",
        "update_user_profile"
    }

class BlacklistConstraint:
    """黑名单:明确禁止的操作不能进行"""

    def can_perform(self, operation: str) -> bool:
        return operation not in self.forbidden_operations

    forbidden_operations = {
        "delete_data",
        "modify_permissions",
        "system_shutdown"
    }

class RuleBasedConstraint:
    """基于规则的约束"""

    def can_perform(self, operation: str, context: dict) -> bool:
        """根据上下文和规则判断是否允许操作"""

        # 规则1:删除操作需要特殊权限
        if "delete" in operation:
            if "admin_privileges" not in context:
                return False

        # 规则2:成本超过100元的操作需要审批
        if context.get("operation_cost", 0) > 100:
            if "approval_granted" not in context:
                return False

        # 规则3:工作时间外的某些操作不允许
        if datetime.now().hour > 18:
            if operation in self.after_hours_forbidden:
                return False

        return True

class ConstraintEnforcer:
    """约束执行器"""

    def __init__(self, constraints: List[Constraint]):
        self.constraints = constraints
        self.violations = []

    async def check_and_enforce(
        self,
        operation: Operation,
        context: dict
    ) -> bool:
        """
        检查操作是否违反约束。

        返回:是否允许操作
        """
        for constraint in self.constraints:
            if not constraint.allows(operation, context):
                # 记录违反
                self.violations.append({
                    "timestamp": datetime.now(),
                    "constraint": constraint.name,
                    "operation": operation,
                    "context": context
                })

                # 日志警告
                logger.warning(
                    f"Constraint violation: {constraint.name}",
                    extra={"operation": operation}
                )

                return False

        return True

    def get_violations(self) -> List[dict]:
        """获取所有的约束违反记录"""
        return self.violations

@dataclass
class OperationLog:
    """操作日志"""
    timestamp: datetime
    agent_id: str
    operation_type: str  # "tool_call", "permission_check", "decision"
    operation_name: str  # 具体的操作名称
    input: dict          # 输入参数
    output: dict         # 输出结果
    status: str          # "success", "error", "permission_denied"
    duration_ms: float   # 耗时
    error_message: Optional[str] = None

class OperationLogger:
    """操作日志记录器"""

    async def log(self, operation_log: OperationLog) -> None:
        """记录一个操作"""
        # 格式化为可读的形式
        log_entry = {
            "timestamp": operation_log.timestamp.isoformat(),
            "agent": operation_log.agent_id,
            "type": operation_log.operation_type,
            "operation": operation_log.operation_name,
            "status": operation_log.status,
            "duration_ms": operation_log.duration_ms
        }

        # 写入日志存储
        await self.storage.append(log_entry)

        # 如果是错误,也要输出到错误流
        if operation_log.status == "error":
            logger.error(f"Operation failed: {operation_log.operation_name}",
                        extra={"log_entry": log_entry})

@dataclass
class ExecutionTrace:
    """执行追踪"""
    trace_id: str                  # 追踪的全局唯一ID
    agent_id: str
    task_id: str
    steps: List[TraceStep] = field(default_factory=list)
    start_time: datetime = field(default_factory=datetime.now)
    end_time: Optional[datetime] = None

@dataclass
class TraceStep:
    """追踪中的一步"""
    step_id: str                   # 步骤ID
    parent_step_id: Optional[str]  # 父步骤(用于嵌套调用)
    operation: str
    input: dict
    output: dict
    duration_ms: float
    timestamp: datetime
    status: str

class ExecutionTracer:
    """执行追踪器"""

    def __init__(self):
        self.traces: Dict[str, ExecutionTrace] = {}

    def start_trace(
        self,
        agent_id: str,
        task_id: str,
        trace_id: str = None
    ) -> str:
        """开始一个新的执行追踪"""
        trace_id = trace_id or str(uuid.uuid4())
        self.traces[trace_id] = ExecutionTrace(
            trace_id=trace_id,
            agent_id=agent_id,
            task_id=task_id
        )
        return trace_id

    def add_step(
        self,
        trace_id: str,
        operation: str,
        input: dict,
        output: dict,
        duration_ms: float,
        parent_step_id: str = None
    ) -> str:
        """在追踪中添加一步"""
        step_id = str(uuid.uuid4())
        step = TraceStep(
            step_id=step_id,
            parent_step_id=parent_step_id,
            operation=operation,
            input=input,
            output=output,
            duration_ms=duration_ms,
            timestamp=datetime.now(),
            status="success"
        )
        self.traces[trace_id].steps.append(step)
        return step_id

    def get_trace(self, trace_id: str) -> Optional[ExecutionTrace]:
        """获取完整的执行追踪"""
        return self.traces.get(trace_id)

    def visualize_trace(self, trace_id: str) -> str:
        """生成可视化的执行追踪"""
        trace = self.traces.get(trace_id)
        if not trace:
            return ""

        # 生成缩进的树形结构
        lines = [f"Trace: {trace.trace_id}"]
        lines.append(f"Agent: {trace.agent_id}, Task: {trace.task_id}")
        lines.append(f"Duration: {(trace.end_time - trace.start_time).total_seconds():.2f}s")
        lines.append("")

        def add_step_lines(step: TraceStep, indent: int = 0):
            prefix = "  " * indent
            lines.append(f"{prefix}├─ {step.operation} ({step.duration_ms:.0f}ms)")
            lines.append(f"{prefix}│  Input: {step.input}")
            lines.append(f"{prefix}│  Output: {step.output}")

        for step in trace.steps:
            if step.parent_step_id is None:
                add_step_lines(step)

        return "\n".join(lines)

class ReplayableExecution:
    """可重放的执行"""

    def __init__(self, trace: ExecutionTrace):
        self.trace = trace
        self.step_index = 0

    async def replay(
        self,
        runtime: RuntimeEngine,
        simulation: bool = False
    ) -> ReplayResult:
        """
        重放执行。

        simulation: 如果为True,不实际执行工具,而是返回记录的结果
        """
        results = []

        for step in self.trace.steps:
            if simulation:
                # 模拟模式:直接返回记录的结果
                results.append({
                    "operation": step.operation,
                    "recorded_output": step.output,
                    "simulation": True
                })
            else:
                # 实际重放:重新执行相同的操作
                try:
                    actual_output = await self._execute_step(
                        step.operation,
                        step.input
                    )

                    # 验证输出是否一致
                    if actual_output == step.output:
                        results.append({
                            "operation": step.operation,
                            "status": "consistent",
                            "output": actual_output
                        })
                    else:
                        results.append({
                            "operation": step.operation,
                            "status": "diverged",
                            "recorded_output": step.output,
                            "actual_output": actual_output
                        })

                except Exception as e:
                    results.append({
                        "operation": step.operation,
                        "status": "error",
                        "error": str(e)
                    })

        return ReplayResult(
            trace_id=self.trace.trace_id,
            results=results,
            is_consistent=all(r["status"] == "consistent" for r in results)
        )

    async def _execute_step(self, operation: str, input: dict):
        """执行一个步骤"""
        # 实现取决于具体的操作类型
        pass

# 不好:非结构化日志
logger.info("Transfer from ACC001 to ACC002 for amount 1000")

# 好:结构化日志
logger.info("transfer_executed", extra={
    "from_account": "ACC001",
    "to_account": "ACC002",
    "amount": 1000,
    "transaction_id": "TXN12345",
    "status": "success",
    "duration_ms": 198
})

# 在日志中包含trace ID,便于关联
logger.info("Step completed", extra={
    "trace_id": trace.trace_id,
    "step_id": step.step_id,
    "operation": "tool_call"
})

async def periodic_verification():
    """定期验证执行的一致性"""
    # 每小时,随机选择一个旧的执行,尝试重放
    old_traces = await trace_storage.list_old_traces(
        before_hours=1,
        limit=10
    )

    for trace in old_traces:
        replayer = ReplayableExecution(trace)
        result = await replayer.replay(runtime, simulation=False)

        if not result.is_consistent:
            logger.error("Inconsistency detected in replay",
                        extra={"trace_id": trace.trace_id})
            # 触发告警,通知运维

def generate_user_summary(trace: ExecutionTrace) -> str:
    """生成用户友好的摘要"""
    total_duration = (trace.end_time - trace.start_time).total_seconds()
    success_count = sum(1 for step in trace.steps if step.status == "success")
    error_count = sum(1 for step in trace.steps if step.status == "error")

    return f"""
    执行摘要:
    - 执行ID:{trace.trace_id[:8]}...
    - 执行时间:{trace.start_time.strftime('%Y-%m-%d %H:%M:%S')}
    - 耗时:{total_duration:.2f}秒
    - 步骤数:{len(trace.steps)}(成功{success_count},失败{error_count})
    - 状态:{'成功' if error_count == 0 else '失败'}

    详细日志:{generate_trace_url(trace.trace_id)}
    """

class ManualOnlyMode:
    """完全人工操作模式"""

    async def execute_operation(self, operation: Operation) -> Result:
        """
        每一步都需要人工批准。
        """
        # 1. Agent提议操作
        proposal = await agent.propose_operation(task)

        # 2. 等待人工批准
        approval = await request_human_approval(
            operation=proposal,
            timeout=timedelta(hours=24)
        )

        if approval.approved:
            # 3. 由人工或系统执行
            result = await execute_operation(proposal)
        else:
            result = Result(status="rejected", reason=approval.reason)

        return result

class ApproveAlwaysMode:
    """每个操作都需要人工审批"""

    async def execute_operation(self, operation: Operation) -> Result:
        """
        每个操作都需要事前批准。
        """
        # 请求批准
        approval = await request_human_approval(
            operation=operation,
            timeout=timedelta(minutes=5)
        )

        if not approval.approved:
            return Result(status="rejected")

        # 执行操作
        return await execute_operation(operation)

class ApprovedOnceMode:
    """任务开始时批准一次"""

    async def execute_task(self, task: Task) -> Result:
        """
        在任务开始时,对整个任务进行一次审批。
        一旦批准,任务执行过程中不再询问。
        """
        # 1. 任务规划阶段:Agent提议任务计划
        plan = await agent.plan_task(task)

        # 2. 人工审查:人工查看任务计划
        approval = await request_task_approval(
            task=task,
            plan=plan,
            timeout=timedelta(hours=1)
        )

        if not approval.approved:
            return Result(status="rejected")

        # 3. 执行阶段:任务自动执行
        result = await execute_task_plan(plan)

        # 4. 完成:生成执行报告
        report = generate_execution_report(result)

        return Result(
            status="success",
            output=result,
            report=report
        )

class AskFirstMode:
    """关键操作事前询问"""

    # 定义哪些操作被认为是关键的
    CRITICAL_OPERATIONS = {
        "delete_data",
        "transfer_money",
        "modify_permissions"
    }

    async def execute_operation(self, operation: Operation) -> Result:
        """
        关键操作需要事前询问,其他操作自动执行。
        """
        if operation.type in self.CRITICAL_OPERATIONS:
            # 关键操作:询问
            approval = await request_human_approval(
                operation=operation,
                timeout=timedelta(minutes=5)
            )

            if not approval.approved:
                return Result(status="rejected")

        # 执行操作
        return await execute_operation(operation)

class AutoWithNotificationMode:
    """自动执行,并发送通知"""

    async def execute_operation(self, operation: Operation) -> Result:
        """
        自动执行操作,执行后发送通知。
        """
        try:
            # 执行操作
            result = await execute_operation(operation)

            # 发送通知
            await notify_user(
                message=f"Operation {operation.type} completed",
                details=result,
                urgency="low"
            )

            return result

        except Exception as e:
            # 出错时发送警告通知
            await notify_user(
                message=f"Operation {operation.type} failed",
                error=str(e),
                urgency="high"
            )
            return Result(status="error", error=str(e))

class FullTrustMode:
    """充分信任,无需额外监控"""

    async def execute_operation(self, operation: Operation) -> Result:
        """
        完全信任Agent,无需额外的验证或监控。
        """
        return await execute_operation(operation)

class TrustEvaluator:
    """信任等级评估器"""

    # 各信任等级的提升标准
    PROMOTION_CRITERIA = {
        "Manual Only → Approve Always": {
            "min_operations": 100,
            "min_success_rate": 0.99,  # 99%成功率
            "min_duration_days": 7,    # 运行至少7天
            "no_critical_errors": True
        },
        "Approve Always → Approve Once": {
            "min_operations": 1000,
            "min_success_rate": 0.995,  # 99.5%
            "min_duration_days": 30,
            "no_critical_errors": True,
            "recent_error_rate": 0.005  # 最近的错误率<0.5%
        },
        "Approve Once → Ask First": {
            "min_operations": 10000,
            "min_success_rate": 0.999,  # 99.9%
            "min_duration_days": 90,
            "no_critical_errors": True
        },
        "Ask First → Auto with Notification": {
            "min_operations": 50000,
            "min_success_rate": 0.9999,  # 99.99%
            "min_duration_days": 180,
            "no_critical_errors_in_recent_month": True
        },
        "Auto with Notification → Full Trust": {
            "min_operations": 1000000,
            "min_success_rate": 0.99999,  # 99.999%
            "min_duration_days": 365,
            "no_critical_errors_in_recent_quarter": True
        }
    }

    async def evaluate_promotion(
        self,
        current_level: TrustLevel,
        agent_history: AgentHistory
    ) -> Optional[TrustLevel]:
        """
        评估是否应该提升智能体的信任等级。
        """
        next_level = self._get_next_level(current_level)
        criteria = self.PROMOTION_CRITERIA.get(f"{current_level} → {next_level}")

        if criteria is None:
            return None  # 已经是最高等级

        # 检查所有标准
        checks = {
            "operations": agent_history.total_operations >= criteria["min_operations"],
            "success_rate": agent_history.success_rate >= criteria["min_success_rate"],
            "duration": agent_history.days_running >= criteria["min_duration_days"],
            "critical_errors": not criteria.get("no_critical_errors", False)
                               or not agent_history.has_critical_errors
        }

        # 所有标准都满足才能提升
        if all(checks.values()):
            logger.info(f"Agent {agent_history.agent_id} promoted from {current_level} to {next_level}",
                       extra={"checks": checks})
            return next_level

        logger.info(f"Agent promotion blocked",
                   extra={"agent_id": agent_history.agent_id, "failed_checks": {k: v for k, v in checks.items() if not v}})
        return None

    def _get_next_level(self, current_level: TrustLevel) -> TrustLevel:
        """获取下一个信任等级"""
        levels = [
            "Manual Only",
            "Approve Always",
            "Approve Once",
            "Ask First",
            "Auto with Notification",
            "Full Trust"
        ]
        idx = levels.index(current_level)
        return levels[idx + 1] if idx < len(levels) - 1 else None

class TrustDemotionTrigger:
    """信任降级触发器"""

    # 触发降级的条件
    DEMOTION_TRIGGERS = {
        "critical_error": 1,           # 一次严重错误
        "multiple_errors_in_short_time": 3,  # 短时间内多次错误
        "security_incident": 1,        # 一次安全事件
        "user_complaint": 5            # 5个用户投诉
    }

    async def check_and_demote(
        self,
        agent_id: str,
        recent_history: List[AgentEvent]
    ) -> Optional[TrustLevel]:
        """
        检查是否应该降级智能体的信任等级。
        """
        current_level = await get_agent_trust_level(agent_id)

        # 计数各类事件
        event_counts = {
            "critical_errors": sum(1 for e in recent_history if e.type == "critical_error"),
            "errors_24h": sum(1 for e in recent_history if e.type == "error" and
                             e.timestamp > datetime.now() - timedelta(hours=24)),
            "security_incidents": sum(1 for e in recent_history if e.type == "security_incident"),
            "complaints": sum(1 for e in recent_history if e.type == "user_complaint")
        }

        # 检查触发条件
        if event_counts["critical_errors"] >= self.DEMOTION_TRIGGERS["critical_error"]:
            demoted_level = self._get_previous_level(current_level)
            logger.warning(f"Agent {agent_id} demoted due to critical error",
                          extra={"from": current_level, "to": demoted_level})
            return demoted_level

        if event_counts["errors_24h"] >= self.DEMOTION_TRIGGERS["multiple_errors_in_short_time"]:
            demoted_level = self._get_previous_level(current_level)
            logger.warning(f"Agent {agent_id} demoted due to multiple errors",
                          extra={"event_counts": event_counts})
            return demoted_level

        if event_counts["security_incidents"] >= self.DEMOTION_TRIGGERS["security_incident"]:
            demoted_level = "Ask First"  # 直接降回询问级别
            logger.error(f"Agent {agent_id} demoted to Ask First due to security incident")
            return demoted_level

        return None

    def _get_previous_level(self, current_level: TrustLevel) -> TrustLevel:
        """获取前一个信任等级"""
        # 至少降级一个等级,最多保留在Ask-First
        levels = ["Manual Only", "Approve Always", "Approve Once", "Ask First", "Auto with Notification", "Full Trust"]
        idx = max(0, levels.index(current_level) - 1)
        return levels[idx]

def visualize_trust_evolution(agent_history: AgentHistory) -> str:
    """生成Agent信任等级演进的可视化"""
    lines = [f"Agent: {agent_history.agent_id}"]
    lines.append(f"Current Trust Level: {agent_history.current_trust_level}\n")

    # 时间线
    lines.append("Timeline of Trust Changes:")
    for event in agent_history.trust_changes:
        lines.append(f"  {event.timestamp.strftime('%Y-%m-%d')} "
                    f"{event.from_level} → {event.to_level}")

    # 统计数据
    lines.append(f"\nStats:")
    lines.append(f"  Total Operations: {agent_history.total_operations}")
    lines.append(f"  Success Rate: {agent_history.success_rate:.2%}")
    lines.append(f"  Critical Errors: {agent_history.critical_errors}")
    lines.append(f"  Days Running: {agent_history.days_running}")

    # 提升建议
    next_level = agent_history.promotion_readiness
    if next_level:
        lines.append(f"\nPromotion Eligible: {agent_history.current_trust_level} → {next_level}")

    return "\n".join(lines)

import asyncio
import logging
import random
from typing import Any, Callable

logger = logging.getLogger(__name__)


class RetryStrategy:
    """重试策略"""

    def __init__(
        self,
        max_retries: int = 3,
        base_delay_seconds: float = 1.0,
        exponential_base: float = 2.0,
        jitter: bool = True
    ):
        self.max_retries = max_retries
        self.base_delay_seconds = base_delay_seconds
        self.exponential_base = exponential_base
        self.jitter = jitter

    async def execute_with_retry(
        self,
        operation: Callable,
        *args,
        **kwargs
    ) -> Any:
        """
        执行操作,失败时自动重试。
        """
        last_error = None

        for attempt in range(self.max_retries + 1):
            try:
                return await operation(*args, **kwargs)

            except Exception as e:
                last_error = e

                # 判断是否应该重试
                if not self._is_retryable(e):
                    raise

                # 计算延迟
                if attempt < self.max_retries:
                    delay = self._calculate_delay(attempt)
                    logger.info(
                        f"Retry attempt {attempt + 1}/{self.max_retries} "
                        f"after {delay:.1f}s",
                        extra={"error": str(e)}
                    )
                    await asyncio.sleep(delay)

        # 所有重试都失败了
        raise last_error

    def _is_retryable(self, error: Exception) -> bool:
        """判断错误是否可重试"""
        retryable_errors = (
            asyncio.TimeoutError,
            ConnectionError,
            TimeoutError,
            # 某些HTTP错误可以重试
        )
        return isinstance(error, retryable_errors)

    def _calculate_delay(self, attempt: int) -> float:
        """计算重试延迟(指数退避)"""
        delay = self.base_delay_seconds * (self.exponential_base ** attempt)

        if self.jitter:
            # 添加随机抖动,避免"羊群效应"
            jitter = random.uniform(0, delay * 0.1)
            delay += jitter

        # 设置最大延迟
        max_delay = 60  # 最多等60秒
        return min(delay, max_delay)

# 重试策略使用示例
retry_strategy = RetryStrategy(max_retries=3, base_delay_seconds=0.5)

async def call_weather_api(city: str):
    """调用天气API"""
    return await retry_strategy.execute_with_retry(
        weather_api.get,
        city=city
    )

class FallbackStrategy:
    """降级/回退策略"""

    def __init__(self, fallbacks: List[Callable]):
        """
        初始化降级策略。
        fallbacks是一个优先级列表,包含多个备选方案。
        """
        self.fallbacks = fallbacks

    async def execute_with_fallback(
        self,
        *args,
        **kwargs
    ) -> Any:
        """
        尝试执行,如果失败则使用备选方案。
        """
        last_error = None

        for i, fallback in enumerate(self.fallbacks):
            try:
                result = await fallback(*args, **kwargs)
                logger.info(
                    f"Fallback {i} succeeded",
                    extra={"fallback": fallback.__name__}
                )
                return result

            except Exception as e:
                last_error = e
                logger.warning(
                    f"Fallback {i} failed, trying next",
                    extra={"error": str(e)}
                )

        # 所有回退都失败了
        raise last_error

# 降级策略使用示例:如果无法获取实时天气,使用缓存数据,再不行使用默认值

async def get_weather_with_fallback(city: str):
    """获取天气,支持降级"""

    # 备选方案优先级:实时数据 > 缓存数据 > 默认值
    strategy = FallbackStrategy([
        lambda c: real_time_weather_api.get(c),
        lambda c: cached_weather.get(c),
        lambda c: default_weather_for(c)
    ])

    return await strategy.execute_with_fallback(city)

class BulkheadPattern:
    """隔离模式:防止一个故障影响整个系统"""

    def __init__(self, num_compartments: int = 10):
        self.compartments = [asyncio.Semaphore(10) for _ in range(num_compartments)]

    def get_compartment(self, request_id: str) -> asyncio.Semaphore:
        """根据请求ID,将其分配到某个隔离舱"""
        compartment_idx = hash(request_id) % len(self.compartments)
        return self.compartments[compartment_idx]

    async def execute_isolated(
        self,
        request_id: str,
        operation: Callable,
        *args,
        **kwargs
    ) -> Any:
        """
        在隔离的舱中执行操作。
        如果一个舱出现问题,不会影响其他舱。
        """
        compartment = self.get_compartment(request_id)

        async with compartment:
            return await operation(*args, **kwargs)

# 隔离模式使用示例
bulkhead = BulkheadPattern(num_compartments=20)

async def process_request(request_id: str, request_data: dict):
    """处理请求,使用隔离模式"""
    return await bulkhead.execute_isolated(
        request_id,
        process_request_logic,
        request_data
    )

from datetime import datetime
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"          # 正常状态
    OPEN = "open"              # 故障状态,拒绝请求
    HALF_OPEN = "half_open"    # 恢复测试状态

class CircuitBreaker:
    """熔断器:防止级联故障"""

    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout_seconds: int = 60
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout_seconds = recovery_timeout_seconds
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.last_failure_time = None
        self.last_recovery_attempt = None

    async def execute(
        self,
        operation: Callable,
        *args,
        **kwargs
    ) -> Any:
        """
        通过熔断器执行操作。
        """
        # 检查是否应该转换到HALF_OPEN状态
        self._update_state()

        # 根据当前状态决定是否执行
        if self.state == CircuitState.OPEN:
            raise CircuitBreakerOpenError(
                f"Circuit breaker is OPEN, rejecting request"
            )

        try:
            result = await operation(*args, **kwargs)
            # 成功,重置计数
            self._record_success()
            return result

        except Exception as e:
            # 失败,记录
            self._record_failure()
            raise

    def _update_state(self):
        """更新熔断器状态"""
        if self.state == CircuitState.OPEN:
            # 如果已经打开足够长的时间,尝试恢复
            if (datetime.now() - self.last_failure_time).seconds > self.recovery_timeout_seconds:
                self.state = CircuitState.HALF_OPEN
                self.last_recovery_attempt = datetime.now()
                logger.info("Circuit breaker transitioned to HALF_OPEN")

    def _record_success(self):
        """记录成功"""
        if self.state == CircuitState.HALF_OPEN:
            # 从HALF_OPEN成功恢复到CLOSED
            self.state = CircuitState.CLOSED
            self.failure_count = 0
            logger.info("Circuit breaker recovered to CLOSED")
        elif self.state == CircuitState.CLOSED:
            # 恢复失败计数
            if self.failure_count > 0:
                self.failure_count = max(0, self.failure_count - 1)

    def _record_failure(self):
        """记录失败"""
        self.failure_count += 1
        self.last_failure_time = datetime.now()

        if self.failure_count >= self.failure_threshold and self.state != CircuitState.OPEN:
            self.state = CircuitState.OPEN
            logger.warning(
                f"Circuit breaker opened after {self.failure_count} failures"
            )

class CircuitBreakerOpenError(Exception):
    """熔断器打开时的异常"""
    pass

# 熔断器使用示例
breaker = CircuitBreaker(failure_threshold=5, recovery_timeout_seconds=30)

async def call_external_api():
    return await breaker.execute(external_api.request)

class CheckpointedExecution:
    """支持检查点的执行"""

    def __init__(self, checkpoint_storage):
        self.checkpoint_storage = checkpoint_storage

    async def execute_with_checkpoints(
        self,
        task_id: str,
        steps: List[Step]
    ) -> Result:
        """
        执行一系列步骤,每个步骤前后都保存检查点。
        """
        # 检查是否有之前的检查点,以便恢复
        checkpoint = await self.checkpoint_storage.load(task_id)
        start_idx = checkpoint.last_completed_step + 1 if checkpoint else 0

        results = []

        for i in range(start_idx, len(steps)):
            step = steps[i]

            try:
                # 执行步骤
                result = await execute_step(step)
                results.append(result)

                # 保存检查点
                await self.checkpoint_storage.save(
                    task_id,
                    CheckPoint(
                        last_completed_step=i,
                        results_so_far=results
                    )
                )

            except Exception as e:
                # 步骤失败,保存失败检查点
                await self.checkpoint_storage.save_error(
                    task_id,
                    ErrorCheckPoint(
                        failed_step=i,
                        error=str(e),
                        partial_results=results
                    )
                )
                raise

        # 完成后,清理检查点
        await self.checkpoint_storage.delete(task_id)
        return Result(status="success", outputs=results)

    async def resume_task(self, task_id: str, steps: List[Step]) -> Result:
        """
        恢复一个之前失败的任务。
        """
        checkpoint = await self.checkpoint_storage.load(task_id)
        if checkpoint is None:
            raise ValueError(f"No checkpoint found for task {task_id}")

        logger.info(
            f"Resuming task {task_id} from step {checkpoint.last_completed_step + 1}",
            extra={"step_index": checkpoint.last_completed_step + 1}
        )

        return await self.execute_with_checkpoints(
            task_id,
            steps
        )

class HealthMonitor:
    """系统健康监控"""

    async def monitor(self):
        """持续监控系统健康"""
        while True:
            metrics = await self._collect_metrics()

            # 检查关键指标
            alerts = []

            # 1. 错误率过高
            if metrics["error_rate"] > 0.01:  # >1%
                alerts.append(Alert(
                    severity="critical",
                    message=f"Error rate {metrics['error_rate']:.2%} exceeds threshold"
                ))

            # 2. 响应时间过长
            if metrics["avg_response_time_ms"] > 1000:
                alerts.append(Alert(
                    severity="warning",
                    message=f"Avg response time {metrics['avg_response_time_ms']:.0f}ms"
                ))

            # 3. 任务积压
            if metrics["pending_tasks"] > 10000:
                alerts.append(Alert(
                    severity="warning",
                    message=f"{metrics['pending_tasks']} tasks pending"
                ))

            # 4. 外部依赖不可用
            for dependency, available in metrics["dependencies"].items():
                if not available:
                    alerts.append(Alert(
                        severity="critical",
                        message=f"Dependency {dependency} is unavailable"
                    ))

            # 发送告警
            for alert in alerts:
                await self._send_alert(alert)

            await asyncio.sleep(60)  # 每分钟检查一次