AI Agent 异常检测与自愈编排：从故障感知到自动降级的工程实践

一、运维的"告警风暴"：人工响应的效率瓶颈与误操作风险

云原生环境下的故障响应面临两个极端：告警过多和告警过少。某生产集群在流量高峰期，Prometheus 在 5 分钟内触发 200+ 条告警——节点 CPU 飙升、Pod 重启、服务延迟升高、数据库连接池耗尽……值班工程师面对告警洪流，很难在短时间内判断根因。更常见的情况是：80% 的告警是衍生告警（同一个根因触发的连锁反应），但人工筛选根因告警平均需要 15-30 分钟。

另一个极端是"静默故障"——某些退化性问题（如内存泄漏导致的缓慢 OOM、磁盘 IO 延迟的渐进式升高）不会触发阈值告警，但持续影响服务质量。等用户投诉到达时，故障已经持续了数十分钟甚至数小时。

AI Agent 的异常检测与自愈编排试图解决这两个问题：通过时序异常检测模型识别"静默故障"，通过拓扑关联分析从告警风暴中提取根因，通过预定义的自愈策略自动执行降级或修复操作。关键约束是：自愈操作必须有明确的回滚路径，且高风险操作（如节点驱逐、服务重启）需要人工确认。

二、自愈编排架构：从异常感知到策略执行的四层模型

flowchart TB
    subgraph L1["第一层：异常感知"]
        A["指标流<br/>Prometheus/Metrics"] --> B["时序异常检测<br/>STL 分解 + 统计检验"]
        C["日志流<br/>ELK/Loki"] --> D["日志模式识别<br/>异常模式聚类"]
        E["链路流<br/>Jaeger/Zipkin"] --> F["调用链异常<br/>延迟突增/错误率"]
    end

    subgraph L2["第二层：根因定位"]
        B --> G["拓扑关联分析<br/>服务依赖图 + 因果推断"]
        D --> G
        F --> G
        G --> H["根因排序<br/>PC 算法因果图"]
    end

    subgraph L3["第三层：策略匹配"]
        H --> I["故障模式库<br/>预定义的故障-策略映射"]
        I --> J["策略选择<br/>匹配度评分 + 风险评级"]
    end

    subgraph L4["第四层：执行与回滚"]
        J --> K{"风险等级"}
        K -->|"低风险<br/>自动执行"| L["自动降级<br/>限流/熔断/扩容"]
        K -->|"高风险<br/>人工确认"| M["确认后执行<br/>重启/驱逐/切换"]
        L --> N["效果验证<br/>指标是否恢复"]
        M --> N
        N -->|"未恢复"| O["回滚操作<br/>恢复变更前状态"]
        N -->|"已恢复"| P["归档记录<br/>更新故障模式库"]
    end

    style G fill:#f96,stroke:#333
    style K fill:#9cf,stroke:#333
    style O fill:#f66,stroke:#333
    style P fill:#9f9,stroke:#333

四层模型的设计逻辑：

第一层：异常感知。三类数据源并行处理——指标流通过 STL 分解（Seasonal-Trend decomposition using Loess）分离趋势、季节和残差成分，对残差做统计检验识别异常点；日志流通过模式聚类识别异常日志模式（如 ERROR 突增、新出现的异常堆栈）；链路流通过调用链分析检测延迟突增和错误率飙升。三路信号独立检测，结果汇总到第二层。

第二层：根因定位。基于服务拓扑图（从 K8s Service 和 Istio 配置自动构建），将异常信号映射到拓扑节点。使用 PC 算法（Peter-Clark 算法）构建因果图——从异常信号的时序先后关系推断因果方向，剔除衍生告警，定位根因节点。输出是一个按因果概率排序的根因列表。

第三层：策略匹配。维护一个故障模式库，每条记录包含：故障特征（指标模式、日志模式、拓扑位置）、推荐策略（限流、熔断、扩容、重启、切换）、风险等级（低/中/高）、回滚方案。将根因特征与故障模式库做模糊匹配，选择匹配度最高的策略。

第四层：执行与回滚。低风险策略（如限流、熔断）自动执行，高风险策略（如节点驱逐、数据库切换）需人工确认。执行后持续监控指标，如果 5 分钟内未恢复，自动触发回滚操作，恢复到变更前的状态。

三、自愈编排引擎的代码实现

from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
import time

class RiskLevel(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"

class ActionType(Enum):
    RATE_LIMIT = "rate_limit"
    CIRCUIT_BREAK = "circuit_break"
    SCALE_OUT = "scale_out"
    RESTART_POD = "restart_pod"
    DRAIN_NODE = "drain_node"
    FAILOVER = "failover"

@dataclass
class AnomalySignal:
    """异常信号"""
    source: str               # 数据源: metric/log/trace
    service: str              # 关联服务
    anomaly_type: str         # 异常类型
    severity: float           # 严重程度 0-1
    timestamp: float
    details: dict = field(default_factory=dict)

@dataclass
class RootCause:
    """根因分析结果"""
    service: str
    probability: float        # 因果概率
    evidence: list[str]       # 支撑证据
    anomaly_signals: list[AnomalySignal]

@dataclass
class FaultPattern:
    """故障模式库条目"""
    pattern_id: str
    feature_signature: dict   # 故障特征签名
    action: ActionType
    risk_level: RiskLevel
    params: dict              # 策略参数
    rollback_action: Optional[ActionType] = None
    rollback_params: dict = field(default_factory=dict)

@dataclass
class RemediationPlan:
    """自愈执行计划"""
    root_cause: RootCause
    pattern: FaultPattern
    requires_confirmation: bool
    created_at: float = field(default_factory=time.time)

class SelfHealingOrchestrator:
    """自愈编排引擎"""

    def __init__(self):
        self.fault_patterns: list[FaultPattern] = []
        self.execution_history: list[dict] = []

    def register_pattern(self, pattern: FaultPattern):
        """注册故障模式"""
        self.fault_patterns.append(pattern)

    def match_pattern(
        self, root_cause: RootCause
    ) -> Optional[FaultPattern]:
        """根据根因特征匹配故障模式"""
        best_match = None
        best_score = 0.0

        for pattern in self.fault_patterns:
            score = self._compute_match_score(
                root_cause, pattern
            )
            if score > best_score and score > 0.6:
                best_score = score
                best_match = pattern

        return best_match

    def _compute_match_score(
        self, root_cause: RootCause, pattern: FaultPattern
    ) -> float:
        """计算根因与故障模式的匹配度"""
        score = 0.0
        sig = pattern.feature_signature

        # 匹配服务类型
        if root_cause.service in sig.get("services", []):
            score += 0.3

        # 匹配异常类型
        signal_types = {
            s.anomaly_type for s in root_cause.anomaly_signals
        }
        pattern_types = set(sig.get("anomaly_types", []))
        if signal_types & pattern_types:
            overlap = len(signal_types & pattern_types)
            total = len(signal_types | pattern_types)
            score += 0.4 * (overlap / max(total, 1))

        # 匹配严重程度范围
        avg_severity = (
            sum(s.severity for s in root_cause.anomaly_signals)
            / max(len(root_cause.anomaly_signals), 1)
        )
        severity_range = sig.get("severity_range", (0.0, 1.0))
        if severity_range[0] <= avg_severity <= severity_range[1]:
            score += 0.3

        return score

    def create_plan(
        self, root_cause: RootCause
    ) -> Optional[RemediationPlan]:
        """创建自愈执行计划"""
        pattern = self.match_pattern(root_cause)
        if not pattern:
            return None

        requires_confirmation = pattern.risk_level in (
            RiskLevel.MEDIUM, RiskLevel.HIGH
        )

        return RemediationPlan(
            root_cause=root_cause,
            pattern=pattern,
            requires_confirmation=requires_confirmation,
        )

    def execute_plan(
        self,
        plan: RemeditionPlan,
        confirmed: bool = False,
    ) -> dict:
        """执行自愈计划"""
        if plan.requires_confirmation and not confirmed:
            return {
                "status": "pending_confirmation",
                "action": plan.pattern.action.value,
                "risk_level": plan.pattern.risk_level.value,
                "message": "高风险操作需要人工确认",
            }

        # 记录执行前状态（用于回滚）
        pre_state = self._capture_state(plan)

        # 执行自愈操作
        result = self._execute_action(
            plan.pattern.action, plan.pattern.params
        )

        # 记录执行历史
        self.execution_history.append({
            "plan": plan,
            "pre_state": pre_state,
            "result": result,
            "timestamp": time.time(),
        })

        return result

    def _capture_state(self, plan: RemeditionPlan) -> dict:
        """捕获执行前状态，用于回滚"""
        return {
            "service": plan.root_cause.service,
            "action": plan.pattern.action.value,
            "timestamp": time.time(),
        }

    def _execute_action(
        self, action: ActionType, params: dict
    ) -> dict:
        """执行具体的自愈动作"""
        # 实际生产中，这里调用 K8s API / Istio API 等
        action_handlers = {
            ActionType.RATE_LIMIT: self._handle_rate_limit,
            ActionType.CIRCUIT_BREAK: self._handle_circuit_break,
            ActionType.SCALE_OUT: self._handle_scale_out,
            ActionType.RESTART_POD: self._handle_restart_pod,
        }
        handler = action_handlers.get(action)
        if handler:
            return handler(params)
        return {"status": "unsupported_action"}

    def _handle_rate_limit(self, params: dict) -> dict:
        """限流策略：降低服务入口 QPS"""
        return {
            "status": "executed",
            "action": "rate_limit",
            "detail": f"QPS 限制调整为 {params.get('max_qps', 1000)}",
        }

    def _handle_circuit_break(self, params: dict) -> dict:
        """熔断策略：切断异常下游调用"""
        return {
            "status": "executed",
            "action": "circuit_break",
            "detail": f"熔断下游 {params.get('downstream', 'unknown')}",
        }

    def _handle_scale_out(self, params: dict) -> dict:
        """扩容策略：增加 Pod 副本数"""
        return {
            "status": "executed",
            "action": "scale_out",
            "detail": f"副本数调整为 {params.get('replicas', 3)}",
        }

    def _handle_restart_pod(self, params: dict) -> dict:
        """重启策略：重启异常 Pod"""
        return {
            "status": "executed",
            "action": "restart_pod",
            "detail": f"重启 {params.get('pod_name', 'unknown')}",
        }

# ===== 注册故障模式 =====
def build_default_patterns() -> list[FaultPattern]:
    """构建默认故障模式库"""
    return [
        FaultPattern(
            pattern_id="high_cpu_rate_limit",
            feature_signature={
                "services": ["api-gateway", "web-server"],
                "anomaly_types": ["cpu_spike", "latency_increase"],
                "severity_range": (0.5, 0.8),
            },
            action=ActionType.RATE_LIMIT,
            risk_level=RiskLevel.LOW,
            params={"max_qps": 500},
            rollback_action=ActionType.RATE_LIMIT,
            rollback_params={"max_qps": 2000},
        ),
        FaultPattern(
            pattern_id="db_timeout_circuit_break",
            feature_signature={
                "services": ["order-service", "payment-service"],
                "anomaly_types": ["timeout_spike", "error_rate_increase"],
                "severity_range": (0.6, 0.9),
            },
            action=ActionType.CIRCUIT_BREAK,
            risk_level=RiskLevel.LOW,
            params={"downstream": "database", "timeout_ms": 3000},
            rollback_action=ActionType.CIRCUIT_BREAK,
            rollback_params={"downstream": "database", "timeout_ms": 5000},
        ),
        FaultPattern(
            pattern_id="oom_restart_pod",
            feature_signature={
                "services": ["data-processor", "ml-inference"],
                "anomaly_types": ["oom_kill", "pod_restart"],
                "severity_range": (0.7, 1.0),
            },
            action=ActionType.RESTART_POD,
            risk_level=RiskLevel.MEDIUM,
            params={"grace_period_seconds": 30},
        ),
    ]

关键设计决策：故障模式库采用特征签名匹配而非规则引擎，因为规则引擎在复杂拓扑下难以维护。匹配度阈值设为 0.6，低于此值的不执行任何操作——宁可漏处理也不误操作。高风险操作必须人工确认，这是不可逾越的红线。执行历史完整记录，用于事后审计和故障模式库的迭代优化。

四、自愈编排的边界与权衡

误判风险：异常检测模型的误报会触发不必要的自愈操作。例如流量高峰期的 CPU 飙升是正常现象，如果被误判为异常并触发限流，反而会影响业务。缓解策略是设置"冷静期"——同一服务在 5 分钟内只允许执行一次自愈操作，避免连锁反应。

自愈的"第二类错误"：更危险的是自愈操作本身引发新故障。例如重启 Pod 后，如果就绪检查（Readiness Probe）配置不当，流量可能被路由到未就绪的实例上。因此自愈操作必须包含效果验证环节——执行后持续监控 5 分钟，未恢复则回滚。

适用边界：自愈编排适合"已知故障模式"——那些在故障模式库中有记录的、有明确修复策略的故障。对于首次出现的未知故障，自愈引擎无法匹配到策略，应降级为告警通知人工处理。强行对未知故障执行"猜测性"自愈，风险远大于收益。

复杂度成本：维护故障模式库和拓扑关联分析需要持续投入。服务拓扑变更时需要同步更新因果图，新服务上线时需要注册新的故障模式。如果团队没有持续维护的意愿，自愈系统会逐渐退化成"半自动告警系统"。

五、总结

AI Agent 异常检测与自愈编排通过四层模型——异常感知、根因定位、策略匹配、执行与回滚——将运维响应从"人工筛选告警"升级为"自动定位根因并执行修复"。异常感知融合指标、日志、链路三路信号，根因定位基于拓扑关联和因果推断，策略匹配通过故障模式库实现，执行层通过风险分级和回滚机制保障安全。落地时需注意三点：一是设置冷静期防止误操作连锁反应；二是自愈操作必须包含效果验证和自动回滚；三是未知故障不应强行自愈，应降级为人工处理。自愈的目标不是消除运维，而是将重复性的已知故障处理自动化，让运维人员专注于未知问题的排查。