AI 驱动的故障自动排障：从人工排查到智能诊断 Agent 的工程实践

一、排障效率的瓶颈：经验依赖与知识断层

一次典型的线上故障排查流程：收到告警 → 登录机器查看日志 → 检查监控指标 → 排查依赖服务状态 → 定位根因 → 执行修复。这个流程中，最耗时的不是执行操作，而是"想"——判断该查什么、从哪里查、指标异常意味着什么。一个资深运维能在 10 分钟内定位根因，而新手可能需要 1 小时，差距全在经验上。

传统排障的核心问题：排障知识高度个人化，资深工程师的经验无法规模化复制；故障模式不断演进，历史经验无法覆盖新型故障；多系统切换（监控、日志、链路追踪、CMDB）增加认知负荷。AI 排障 Agent 的目标，是将排障经验编码为可执行的推理链，让 AI 自动完成信息收集、假设生成和验证，将排障从"手艺活"变为"工程化流程"。

二、AI 排障 Agent 架构与推理机制剖析

AI 排障 Agent 的核心是一个基于 ReAct（Reasoning + Acting）模式的推理引擎。Agent 在每一步先推理（分析当前信息，生成假设），再行动（调用工具获取新信息），循环迭代直到定位根因。

flowchart TD
    subgraph 输入层
        ALERT[告警事件<br/>异常指标/日志摘要]
        TOPO[服务拓扑<br/>CMDB 依赖关系]
        CONTEXT[变更上下文<br/>近期发布/配置变更]
    end
    subgraph 推理引擎
        OBS[观察<br/>解析告警与上下文]
        THINK[推理<br/>生成排障假设]
        ACT[行动<br/>调用诊断工具]
        EVAL[评估<br/>验证假设是否成立]
    end
    subgraph 工具集
        LOG_TOOL[日志查询<br/>ELK API]
        METRIC_TOOL[指标查询<br/>PromQL]
        TRACE_TOOL[链路查询<br/>Jaeger API]
        K8S_TOOL[K8s 查询<br/>kubectl/API]
        DB_TOOL[数据库查询<br/>慢日志/连接池]
    end
    subgraph 输出
        RCA[根因报告<br/>故障链路+证据]
        FIX[修复建议<br/>操作步骤+风险]
        KB[知识沉淀<br/>案例入库]
    end

    ALERT --> OBS
    TOPO --> OBS
    CONTEXT --> OBS
    OBS --> THINK --> ACT
    ACT -->|调用| LOG_TOOL
    ACT -->|调用| METRIC_TOOL
    ACT -->|调用| TRACE_TOOL
    ACT -->|调用| K8S_TOOL
    ACT -->|调用| DB_TOOL
    LOG_TOOL --> EVAL
    METRIC_TOOL --> EVAL
    TRACE_TOOL --> EVAL
    K8S_TOOL --> EVAL
    DB_TOOL --> EVAL
    EVAL -->|假设不成立| THINK
    EVAL -->|假设成立| RCA
    RCA --> FIX --> KB

关键推理机制解析：

• ReAct 推理循环：Agent 在每一步生成一个三元组（Thought, Action, Observation）。Thought 是当前推理结论，Action 是下一步要执行的工具调用，Observation 是工具返回的结果。循环直到 Thought 中包含确定的根因结论。
• 假设生成与优先级排序：基于告警类型和服务拓扑，Agent 生成多个候选假设。例如收到"服务超时"告警，候选假设包括：数据库慢查询、下游服务不可用、网络抖动、资源不足。假设按先验概率排序，优先验证概率最高的假设。
• 证据链构建：每个根因结论必须附带完整的证据链。例如"数据库慢查询导致服务超时"的证据链为：服务 P99 延迟升高 → 数据库慢查询数量增加 → 慢查询 SQL 为 XXX → 该 SQL 缺少索引。证据链确保结论可追溯、可验证。

三、生产级 AI 排障 Agent 实现与最佳实践

3.1 排障 Agent 核心推理引擎

"""
AI 排障 Agent：基于 ReAct 模式的自动故障诊断引擎
"""
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional, Callable
import json


class ActionType(Enum):
    """Agent 可执行的动作类型"""
    QUERY_LOGS = "query_logs"
    QUERY_METRICS = "query_metrics"
    QUERY_TRACES = "query_traces"
    QUERY_K8S = "query_k8s"
    QUERY_DB = "query_db"
    CONCLUDE = "conclude"


@dataclass
class Hypothesis:
    """排障假设"""
    description: str
    probability: float          # 先验概率 [0, 1]
    evidence: list = field(default_factory=list)
    verified: bool = False


@dataclass
class ReasoningStep:
    """推理步骤"""
    step_number: int
    thought: str                # 当前推理结论
    action: ActionType          # 下一步动作
    action_input: dict          # 动作参数
    observation: str = ""       # 工具返回结果


class TroubleshootingAgent:
    """AI 排障 Agent"""

    def __init__(
        self,
        alert_context: dict,
        topology: dict,
        max_steps: int = 10,
    ):
        self.alert = alert_context
        self.topology = topology
        self.max_steps = max_steps
        self.steps: list[ReasoningStep] = []
        self.hypotheses: list[Hypothesis] = []
        self.root_cause: Optional[str] = None

        # 注册工具
        self.tools: dict[ActionType, Callable] = {
            ActionType.QUERY_LOGS: self._query_logs,
            ActionType.QUERY_METRICS: self._query_metrics,
            ActionType.QUERY_TRACES: self._query_traces,
            ActionType.QUERY_K8S: self._query_k8s,
            ActionType.QUERY_DB: self._query_db,
        }

    def run(self) -> dict:
        """执行排障推理循环"""
        # 初始化假设
        self._generate_initial_hypotheses()

        for step in range(1, self.max_steps + 1):
            # 选择当前最可能的假设
            current = self._select_hypothesis()
            if current is None:
                break

            # 生成下一步推理
            thought, action, action_input = self._reason(
                current, step
            )

            # 执行工具调用
            if action == ActionType.CONCLUDE:
                self.root_cause = current.description
                self.steps.append(ReasoningStep(
                    step_number=step,
                    thought=thought,
                    action=action,
                    action_input={"conclusion": current.description},
                ))
                break

            observation = self._execute_tool(
                action, action_input
            )

            # 记录推理步骤
            self.steps.append(ReasoningStep(
                step_number=step,
                thought=thought,
                action=action,
                action_input=action_input,
                observation=observation,
            ))

            # 更新假设概率
            self._update_hypotheses(current, observation)

        return self._generate_report()

    def _generate_initial_hypotheses(self):
        """基于告警类型和拓扑生成初始假设"""
        alert_type = self.alert.get("type", "")
        service = self.alert.get("service", "")

        # 根据告警类型匹配假设模板
        hypothesis_templates = {
            "service_timeout": [
                Hypothesis(
                    description=f"{service} 下游数据库慢查询",
                    probability=0.35,
                ),
                Hypothesis(
                    description=f"{service} 依赖服务不可用",
                    probability=0.25,
                ),
                Hypothesis(
                    description=f"{service} 所在节点资源不足",
                    probability=0.20,
                ),
                Hypothesis(
                    description=f"{service} 网络抖动",
                    probability=0.10,
                ),
            ],
            "high_error_rate": [
                Hypothesis(
                    description=f"{service} 新版本代码缺陷",
                    probability=0.40,
                ),
                Hypothesis(
                    description=f"{service} 配置变更错误",
                    probability=0.30,
                ),
                Hypothesis(
                    description=f"{service} 下游服务返回异常",
                    probability=0.20,
                ),
            ],
            "pod_crash": [
                Hypothesis(
                    description=f"{service} OOM 被杀",
                    probability=0.45,
                ),
                Hypothesis(
                    description=f"{service} 健康检查失败",
                    probability=0.30,
                ),
                Hypothesis(
                    description=f"{service} 启动配置错误",
                    probability=0.15,
                ),
            ],
        }

        self.hypotheses = hypothesis_templates.get(
            alert_type,
            [Hypothesis(description="未知故障类型", probability=0.5)]
        )

    def _select_hypothesis(self) -> Optional[Hypothesis]:
        """选择当前最可能且未验证的假设"""
        unverified = [
            h for h in self.hypotheses if not h.verified
        ]
        if not unverified:
            return None
        return max(unverified, key=lambda h: h.probability)

    def _reason(
        self, hypothesis: Hypothesis, step: int
    ) -> tuple:
        """基于当前假设生成推理步骤"""
        desc = hypothesis.description.lower()

        if "数据库" in desc or "慢查询" in desc:
            return (
                f"假设：{hypothesis.description}，"
                f"先查询数据库慢查询指标验证",
                ActionType.QUERY_DB,
                {"query_type": "slow_queries", "service": self.alert.get("service")},
            )
        elif "下游" in desc or "依赖" in desc:
            return (
                f"假设：{hypothesis.description}，"
                f"查询依赖服务健康状态",
                ActionType.QUERY_METRICS,
                {"query_type": "service_health", "direction": "downstream"},
            )
        elif "资源" in desc or "OOM" in desc:
            return (
                f"假设：{hypothesis.description}，"
                f"查询 Pod 资源使用情况",
                ActionType.QUERY_K8S,
                {"query_type": "pod_resources", "service": self.alert.get("service")},
            )
        elif "网络" in desc:
            return (
                f"假设：{hypothesis.description}，"
                f"查询网络延迟和丢包率",
                ActionType.QUERY_METRICS,
                {"query_type": "network_latency", "service": self.alert.get("service")},
            )
        else:
            return (
                f"假设：{hypothesis.description}，"
                f"查询服务日志寻找错误线索",
                ActionType.QUERY_LOGS,
                {"query_type": "errors", "service": self.alert.get("service")},
            )

    def _execute_tool(self, action: ActionType, params: dict) -> str:
        """执行工具调用并返回结果"""
        tool = self.tools.get(action)
        if tool:
            return tool(params)
        return "工具不可用"

    def _update_hypotheses(
        self, hypothesis: Hypothesis, observation: str
    ):
        """根据观察结果更新假设概率"""
        hypothesis.evidence.append(observation)

        # 简化的概率更新：如果观察结果支持假设，提高概率
        positive_keywords = ["增加", "升高", "超时", "失败", "异常", "不可用"]
        negative_keywords = ["正常", "稳定", "无异常", "无变化"]

        positive_count = sum(
            1 for kw in positive_keywords if kw in observation
        )
        negative_count = sum(
            1 for kw in negative_keywords if kw in observation
        )

        if positive_count > negative_count:
            hypothesis.probability = min(
                hypothesis.probability * 1.5, 0.95
            )
        elif negative_count > positive_count:
            hypothesis.probability = max(
                hypothesis.probability * 0.5, 0.05
            )
            if hypothesis.probability < 0.1:
                hypothesis.verified = True  # 排除该假设

        # 如果概率足够高，标记为已验证（确认）
        if hypothesis.probability > 0.8:
            hypothesis.verified = True

    def _generate_report(self) -> dict:
        """生成排障报告"""
        return {
            "alert": self.alert,
            "root_cause": self.root_cause or "未定位根因",
            "reasoning_steps": [
                {
                    "step": s.step_number,
                    "thought": s.thought,
                    "action": s.action.value,
                    "observation": s.observation[:200],
                }
                for s in self.steps
            ],
            "evidence_chain": [
                e for h in self.hypotheses
                for e in h.evidence if h.probability > 0.5
            ],
            "hypotheses_ranked": sorted(
                [{"desc": h.description, "prob": round(h.probability, 2)}
                 for h in self.hypotheses],
                key=lambda x: x["prob"],
                reverse=True,
            ),
        }

    # ---- 工具实现（简化版，生产环境对接真实 API）----

    @staticmethod
    def _query_logs(params: dict) -> str:
        """查询日志（对接 ELK API）"""
        return f"查询 {params.get('service')} 的 {params.get('query_type')} 日志"

    @staticmethod
    def _query_metrics(params: dict) -> str:
        """查询指标（对接 Prometheus API）"""
        return f"查询 {params.get('query_type')} 指标"

    @staticmethod
    def _query_traces(params: dict) -> str:
        """查询链路（对接 Jaeger API）"""
        return f"查询链路追踪数据"

    @staticmethod
    def _query_k8s(params: dict) -> str:
        """查询 K8s 资源（对接 K8s API）"""
        return f"查询 {params.get('query_type')} 状态"

    @staticmethod
    def _query_db(params: dict) -> str:
        """查询数据库状态"""
        return f"查询数据库 {params.get('query_type')} 状态"

3.2 排障知识库与案例沉淀

# 排障知识库：故障案例结构化存储
apiVersion: aiops.example.com/v1
kind: TroubleshootingCase
metadata:
  name: db-slow-query-timeout-001
  labels:
    category: database
    severity: critical
    service: order-service
spec:
  # 故障现象
  symptoms:
  - type: alert
    name: ServiceHighLatency
    description: "order-service P99 延迟超过 2s"
  - type: alert
    name: HighErrorRate
    description: "order-service 5xx 错误率超过 5%"
  # 根因
  rootCause:
    type: database_slow_query
    description: "order_service 查询 orders 表缺少 status+created_at 复合索引"
    sql: "SELECT * FROM orders WHERE status='pending' ORDER BY created_at DESC LIMIT 100"
  # 证据链
  evidenceChain:
  - step: 1
    action: query_metrics
    observation: "order-service P99 延迟 2.3s，基线 50ms"
  - step: 2
    action: query_traces
    observation: "90% 延迟集中在 MySQL 查询阶段"
  - step: 3
    action: query_db
    observation: "慢查询数量从 2/min 飙升到 150/min"
  - step: 4
    action: query_db
    observation: "EXPLAIN 显示全表扫描，rows=2,000,000"
  # 修复方案
  remediation:
  - type: sql
    command: "ALTER TABLE orders ADD INDEX idx_status_created (status, created_at)"
    risk: "线上加索引可能锁表，建议使用 pt-online-schema-change"
  - type: config
    command: "增加查询超时时间为 5s，防止级联超时"
  # 预防措施
  prevention:
  - "上线前强制 SQL 审查，缺少索引的查询不允许发布"
  - "慢查询告警阈值从 1s 调整为 500ms"

四、AI 排障 Agent 的局限与架构权衡

推理准确率的上限：Agent 的推理质量取决于假设模板的覆盖度和工具返回信息的准确性。对于从未见过的故障模式，Agent 可能生成错误的假设方向。建议将 Agent 定位为"辅助诊断"而非"自动修复"，人工确认后再执行修复动作。

工具调用的延迟：每次工具调用涉及 API 网络请求（Prometheus、ELK、K8s），单次调用延迟 100ms-1s。10 步推理的总延迟可能达到 5-10 秒。对于需要秒级响应的紧急故障，Agent 的速度可能不够。解决方案是并行执行多个工具调用，减少串行等待。

知识库的维护成本：排障知识库需要持续更新，每次故障后人工录入案例。如果知识库不更新，Agent 的推理能力会随时间衰减。建议将案例录入纳入故障复盘的标准流程，确保知识库与实际故障同步演进。

安全与权限风险：Agent 需要访问生产环境的监控、日志和 K8s API，权限范围较大。如果 Agent 被误操作或被攻击，可能对生产环境造成影响。建议为 Agent 配置只读权限，修复动作必须经过人工审批。

五、总结

AI 排障 Agent 的核心价值是将排障经验从个人能力转化为组织能力。通过 ReAct 推理模式，Agent 自动完成假设生成、工具调用和证据验证，将排障时间从小时级压缩到分钟级。

落地路线建议：第一步，梳理高频故障模式，建立假设模板库；第二步，封装诊断工具（日志查询、指标查询、K8s 查询）为标准化 API；第三步，实现 ReAct 推理引擎，支持多步推理和假设验证；第四步，建立排障知识库，每次故障后录入案例；第五步，将 Agent 集成到告警处理流程，告警触发后自动启动排障；第六步，建立效果度量（排障时间缩短率、根因定位准确率），持续优化推理策略。每一步都要保留人工干预入口，确保 Agent 在推理错误时不会造成生产事故。