AI 故障诊断 Agent：从知识图谱到自动排障的推理引擎

一、排障经验的知识断层：当资深工程师离场

运维排障高度依赖经验。一个工作五年的运维工程师，脑中积累了数百个故障模式与排障路径的映射关系——"数据库连接池耗尽"对应"检查慢查询 + 连接泄漏"，"Pod CrashLoopBackOff"对应"查看事件 + 日志 + 资源限制"。这些隐性知识从未被系统化记录，当资深工程师离职或休假时，排障效率断崖式下降。

AI 故障诊断 Agent 的目标是将这些隐性知识显性化、结构化，并通过推理引擎自动匹配故障模式与排障路径。不同于简单的关键字匹配，诊断 Agent 需要理解故障的上下文——服务拓扑、历史变更、资源状态——并基于这些上下文进行多步推理，逐步缩小故障范围。关键挑战在于：如何将非结构化的排障经验转化为可计算的推理规则，以及如何在推理过程中处理不确定性和信息缺失。

二、AI 诊断 Agent 架构：知识图谱驱动的多步推理

诊断 Agent 的核心架构由知识图谱、推理引擎和工具调用层三部分组成。知识图谱存储故障模式、排障步骤和因果关系；推理引擎基于当前故障上下文在图谱中导航，选择最可能的推理路径；工具调用层执行实际的诊断命令（kubectl、SQL 查询、API 调用），获取证据验证推理假设。

flowchart TD
    A[故障事件输入] --> B[上下文构建器]
    B --> B1[提取故障元数据：服务/指标/时间]
    B --> B2[查询 CMDB：服务拓扑与依赖]
    B --> B3[查询变更记录：近期部署/配置变更]

    B1 --> C[推理引擎]
    B2 --> C
    B3 --> C
    C --> C1[在知识图谱中匹配故障模式]
    C1 --> C2[生成假设列表：可能的根因]
    C2 --> C3[按先验概率排序假设]

    C3 --> D[假设验证循环]
    D --> D1[选择最高概率假设]
    D1 --> D2[生成验证步骤：需要采集的证据]
    D2 --> D3[调用工具层执行诊断命令]

    D3 --> E[工具调用层]
    E --> E1[kubectl：查询 Pod/事件/日志]
    E --> E2[数据库：查询慢查询/连接状态]
    E --> E3[API：查询服务健康状态]
    E --> E4[监控：查询指标时序数据]

    E1 --> F[证据评估]
    E2 --> F
    E3 --> F
    E4 --> F
    F --> F1{假设被验证?}
    F1 -->|是| G[输出根因与修复建议]
    F1 -->|否| F2[降低该假设概率，选择次优假设]
    F2 --> D

    G --> H[更新知识图谱]
    H --> H1[记录推理路径与结果]
    H --> H2[更新先验概率]

知识图谱的结构：节点类型包括故障现象（Symptom）、根因（RootCause）、诊断步骤（DiagnosisStep）和修复动作（FixAction）。边类型包括"可能导致"（Symptom → RootCause）、"验证方式"（RootCause → DiagnosisStep）、"修复方法"（RootCause → FixAction）。每个根因节点维护一个先验概率，基于历史诊断结果动态更新。

假设验证循环是 Agent 的核心推理机制。Agent 不是一次性输出结论，而是迭代式地"提出假设 → 采集证据 → 验证/否定 → 调整假设"。这种循环确保推理过程可追溯、可解释——每一步推理都有明确的证据支撑。

工具调用层将诊断命令封装为标准化接口。Agent 不直接执行 shell 命令，而是通过工具描述（Tool Description）了解每个工具的能力和参数，自主决定调用哪个工具、传什么参数。这种设计使 Agent 可以灵活扩展新的诊断能力，无需修改推理引擎。

三、基于 Python 的故障诊断 Agent 实现

3.1 知识图谱与推理引擎

"""
故障诊断知识图谱与推理引擎
为什么用知识图谱而非规则引擎：规则引擎（如 Drools）只能处理
预定义的 if-then 逻辑，无法处理部分匹配和概率推理。
知识图谱支持模糊匹配和概率传播，更适合故障诊断的不确定性场景
"""

from dataclasses import dataclass, field
from enum import Enum
from typing import List, Optional, Dict
import json


class NodeType(Enum):
    SYMPTOM = "symptom"           # 故障现象
    ROOT_CAUSE = "root_cause"     # 根因
    DIAGNOSIS_STEP = "diagnosis"  # 诊断步骤
    FIX_ACTION = "fix"            # 修复动作


@dataclass
class KGNode:
    """知识图谱节点"""
    node_id: str
    node_type: NodeType
    name: str
    description: str
    properties: dict = field(default_factory=dict)
    prior_probability: float = 0.5  # 先验概率


@dataclass
class KGEdge:
    """知识图谱边"""
    source_id: str
    target_id: str
    relation: str           # 关系类型
    weight: float = 1.0     # 关系权重（条件概率）


class DiagnosisKnowledgeGraph:
    """故障诊断知识图谱"""

    def __init__(self):
        self.nodes: Dict[str, KGNode] = {}
        self.edges: List[KGEdge] = []
        # 邻接表：node_id -> [edge]
        self.outgoing: Dict[str, List[KGEdge]] = {}
        self.incoming: Dict[str, List[KGEdge]] = {}

    def add_node(self, node: KGNode):
        """添加节点"""
        self.nodes[node.node_id] = node
        if node.node_id not in self.outgoing:
            self.outgoing[node.node_id] = []
        if node.node_id not in self.incoming:
            self.incoming[node.node_id] = []

    def add_edge(self, edge: KGEdge):
        """添加边"""
        self.edges.append(edge)
        self.outgoing[edge.source_id].append(edge)
        self.incoming[edge.target_id].append(edge)

    def find_root_causes(
        self, symptom_id: str
    ) -> List[tuple[KGNode, float]]:
        """
        根据故障现象查找可能的根因
        返回: [(根因节点, 综合概率)]
        """
        candidates = []

        # 从现象节点出发，沿"可能导致"边查找根因
        for edge in self.outgoing.get(symptom_id, []):
            if edge.relation == "may_cause":
                target = self.nodes.get(edge.target_id)
                if target and target.node_type == NodeType.ROOT_CAUSE:
                    # 综合概率 = 先验概率 * 条件概率（边权重）
                    combined = (
                        target.prior_probability * edge.weight
                    )
                    candidates.append((target, combined))

        # 按综合概率降序排列
        candidates.sort(key=lambda x: x[1], reverse=True)
        return candidates

    def get_diagnosis_steps(
        self, root_cause_id: str
    ) -> List[tuple[KGNode, float]]:
        """获取根因对应的诊断步骤"""
        steps = []
        for edge in self.outgoing.get(root_cause_id, []):
            if edge.relation == "diagnose_by":
                target = self.nodes.get(edge.target_id)
                if target:
                    steps.append((target, edge.weight))
        # 按权重排序：权重越高表示诊断价值越大
        steps.sort(key=lambda x: x[1], reverse=True)
        return steps

    def get_fix_actions(
        self, root_cause_id: str
    ) -> List[KGNode]:
        """获取根因对应的修复动作"""
        actions = []
        for edge in self.outgoing.get(root_cause_id, []):
            if edge.relation == "fix_by":
                target = self.nodes.get(edge.target_id)
                if target:
                    actions.append(target)
        return actions

    def update_probability(
        self, root_cause_id: str, confirmed: bool
    ):
        """
        更新根因的先验概率（贝叶斯更新）
        为什么需要动态更新：静态概率无法反映环境变化，
        某些根因在特定季节/时段更常见，
        动态更新使概率随实际诊断结果自适应
        """
        node = self.nodes.get(root_cause_id)
        if not node:
            return

        alpha = 0.1  # 学习率：控制更新幅度
        if confirmed:
            node.prior_probability = (
                node.prior_probability
                + alpha * (1.0 - node.prior_probability)
            )
        else:
            node.prior_probability = (
                node.prior_probability
                - alpha * node.prior_probability
            )


class DiagnosisAgent:
    """故障诊断 Agent：基于知识图谱的多步推理"""

    def __init__(self, kg: DiagnosisKnowledgeGraph):
        self.kg = kg
        self.max_iterations = 5  # 最大推理迭代次数

    def diagnose(
        self, symptom_id: str, context: dict
    ) -> dict:
        """
        执行诊断推理
        context 包含：服务拓扑、变更记录、资源状态等上下文
        """
        # 步骤 1：查找候选根因
        candidates = self.kg.find_root_causes(symptom_id)
        if not candidates:
            return {
                "status": "no_match",
                "message": "知识图谱中无匹配的故障模式",
            }

        # 步骤 2：根据上下文调整候选概率
        adjusted = self._adjust_by_context(candidates, context)

        # 步骤 3：假设验证循环
        for iteration in range(self.max_iterations):
            if not adjusted:
                break

            # 选择最高概率的假设
            top_candidate, top_prob = adjusted[0]

            # 获取诊断步骤
            steps = self.kg.get_diagnosis_steps(
                top_candidate.node_id
            )

            # 执行诊断步骤，收集证据
            evidence = self._execute_diagnosis_steps(
                steps, context
            )

            # 评估证据
            if self._evaluate_evidence(evidence):
                # 假设被验证，输出结果
                fix_actions = self.kg.get_fix_actions(
                    top_candidate.node_id
                )
                # 更新知识图谱
                self.kg.update_probability(
                    top_candidate.node_id, confirmed=True
                )
                return {
                    "status": "diagnosed",
                    "root_cause": top_candidate.name,
                    "confidence": top_prob,
                    "evidence": evidence,
                    "fix_actions": [
                        a.description for a in fix_actions
                    ],
                    "iterations": iteration + 1,
                }
            else:
                # 假设被否定，降低概率
                self.kg.update_probability(
                    top_candidate.node_id, confirmed=False
                )
                adjusted = adjusted[1:]  # 移除已否定的假设

        return {
            "status": "inconclusive",
            "message": "所有假设均未通过验证",
            "candidates_tested": self.max_iterations,
        }

    def _adjust_by_context(
        self,
        candidates: List[tuple[KGNode, float]],
        context: dict,
    ) -> List[tuple[KGNode, float]]:
        """
        根据上下文调整候选概率
        为什么需要上下文调整：同样的故障现象在不同上下文中
        可能有不同的根因。例如，"服务超时"在近期有部署变更时
        更可能是代码问题，在流量高峰期更可能是容量问题
        """
        adjusted = []
        for node, prob in candidates:
            boost = 1.0

            # 近期有部署变更 → 代码/配置问题概率提升
            if context.get("recent_deployment"):
                if "deploy" in node.node_id.lower():
                    boost *= 1.5

            # 流量高峰期 → 容量问题概率提升
            if context.get("high_traffic"):
                if "capacity" in node.node_id.lower():
                    boost *= 1.3

            # 上游服务异常 → 依赖问题概率提升
            if context.get("upstream_degraded"):
                if "dependency" in node.node_id.lower():
                    boost *= 1.4

            adjusted.append((node, prob * boost))

        adjusted.sort(key=lambda x: x[1], reverse=True)
        return adjusted

    def _execute_diagnosis_steps(
        self,
        steps: List[tuple[KGNode, float]],
        context: dict,
    ) -> List[dict]:
        """
        执行诊断步骤，收集证据
        实际生产中这里会调用工具层执行 kubectl/sql/api 命令，
        此处简化为返回步骤描述
        """
        evidence = []
        for step, weight in steps:
            evidence.append({
                "step": step.name,
                "description": step.description,
                "weight": weight,
                "tool": step.properties.get("tool", "unknown"),
                "command": step.properties.get("command", ""),
            })
        return evidence

    def _evaluate_evidence(self, evidence: List[dict]) -> bool:
        """
        评估证据是否支持当前假设
        实际生产中这里会分析工具返回的结果，
        判断是否符合假设预期。此处简化为示例逻辑
        """
        # 如果有高权重的诊断步骤，视为关键证据
        high_weight = [e for e in evidence if e["weight"] > 0.7]
        return len(high_weight) > 0

四、AI 诊断 Agent 的现实局限：知识维护与推理可信度

AI 诊断 Agent 在概念上极具吸引力，但实际落地时面临几个根本性挑战。

知识图谱的维护成本：知识图谱的质量直接决定诊断准确性。初始构建需要资深工程师系统化梳理故障模式，这个过程通常需要 2-3 个月。更困难的是持续维护——新服务上线、架构调整、故障模式演变都需要同步更新图谱。如果知识图谱的更新滞后于系统变化，Agent 的诊断结果会越来越不可靠。统计表明，未持续维护的知识图谱在 6 个月后的准确率下降约 40%。

推理链的脆弱性：Agent 的推理是链式的——假设 A 被否定后尝试假设 B，假设 B 被否定后尝试假设 C。如果真正的根因不在候选列表中，Agent 会穷尽所有假设后返回"无法诊断"。更危险的情况是，某个诊断步骤返回了误导性证据（如缓存未刷新导致指标数据过时），Agent 可能基于错误证据确认了错误的假设。

工具调用的可靠性：Agent 依赖外部工具获取诊断证据。工具调用可能失败（API 超时、权限不足）、返回不完整数据或格式变更。Agent 需要处理这些异常情况，否则推理链会中断。工具接口的稳定性是 Agent 可靠性的前提。

可解释性与信任鸿沟：Agent 输出的诊断结果需要人工确认后才能执行修复。如果推理过程不透明——"为什么选择假设 A 而非假设 B"——运维人员无法判断结论的可信度，最终可能忽略 Agent 的建议。可解释性不足是 AI 诊断工具在生产中难以获得信任的主要原因。

适用边界：AI 诊断 Agent 适合故障模式重复、排障步骤标准化的场景（如数据库故障、网络故障、K8s 常见故障）。对于首次出现的未知故障、需要创造性思维的复杂排障场景，Agent 的价值有限，仍需人工主导。

五、总结

AI 故障诊断 Agent 通过知识图谱将排障经验结构化，通过推理引擎实现多步假设验证，通过工具调用层获取诊断证据。知识图谱驱动的推理模式使诊断过程可追溯、可解释，动态概率更新使 Agent 能从历史诊断中学习。但知识图谱的维护成本、推理链的脆弱性和工具调用的可靠性是落地的关键挑战。

落地路线建议：先针对最频繁发生的 3-5 类故障构建知识图谱，与人工排障并行运行验证准确性；然后逐步扩展图谱覆盖更多故障模式，同时引入自动学习机制（从工单系统中提取新的故障模式）；最后在非关键场景开启自动修复，关键场景保持人工确认。全程确保推理过程可追溯，每一步诊断都有明确的证据和概率支撑。