故障树与知识图谱：AIOps 排障 Agent 的推理引擎构建

一、从经验排障到机器推理：排障知识的规模化困境

运维排障高度依赖个人经验。资深工程师看到"数据库连接超时 + 慢查询增多 + 磁盘 IO 等待高"的组合，能迅速判断是磁盘性能瓶颈导致数据库响应变慢。但新手面对同样的告警组合，可能逐个排查，耗时数倍。更关键的是，这些排障经验存在于人的大脑中，无法被团队共享和复用。当资深工程师离职，排障能力也随之流失。

传统知识库（Wiki、Confluence）试图解决经验传承问题，但效果有限。原因在于：排障知识是条件性的——"在 A 条件下，如果出现 B 现象，则可能是 C 原因"——这种因果链式知识用自然语言描述既冗长又不精确，检索效率低，更无法自动执行。

AIOps 排障 Agent 的核心思路，是将排障知识结构化为故障树和知识图谱，让机器能够像资深工程师一样进行因果推理。故障树定义了从症状到根因的推理路径，知识图谱存储了组件间的依赖关系和历史故障模式。两者结合，Agent 能自动完成"观察症状→匹配故障树→验证假设→定位根因"的推理过程。

二、从故障树到贝叶斯推理：排障 Agent 的推理架构

排障 Agent 的推理引擎包含三个核心组件：故障树模型、知识图谱、贝叶斯推理器。

flowchart TD
    subgraph 输入层
        A[告警事件流] --> D[症状提取器]
        B[指标异常] --> D
        C[日志异常] --> D
    end

    subgraph 推理层
        D --> E[故障树匹配<br/>症状→假设映射]
        E --> F[知识图谱查询<br/>依赖关系+历史模式]
        F --> G[贝叶斯推理<br/>计算各假设的后验概率]
        G --> H{最高概率假设<br/>是否超过置信阈值}
        H -->|是| I[输出根因+修复建议]
        H -->|否| J[主动探针<br/>补充观测数据]
        J --> D
    end

    subgraph 知识层
        K[故障树库<br/>人工构建+自动学习]
        L[知识图谱<br/>CMDB+拓扑+历史故障]
        M[案例库<br/>历史排障记录]
    end

    K --> E
    L --> F
    M --> G

故障树模型借鉴了可靠性工程中的 FTA（Fault Tree Analysis）方法。每个故障树以一个顶事件（如"服务不可用"）为根节点，逐层分解为中间事件（如"数据库超时"、"网络分区"）和底事件（如"磁盘满"、"连接池耗尽"）。故障树的逻辑门（AND/OR）定义了事件之间的因果关系——OR 门表示任一子事件可导致父事件，AND 门表示所有子事件同时发生才导致父事件。

知识图谱存储了三类信息：组件依赖关系（服务 A 调用服务 B）、资源拓扑（Pod 运行在 Node X 上）、历史故障模式（服务 A 的数据库超时，80% 的概率是连接池配置不当）。知识图谱的构建数据来源包括 CMDB、服务网格拓扑、链路追踪数据和历史工单。

贝叶斯推理器是整个引擎的核心。给定观察到的症状集合，推理器计算每个假设根因的后验概率，选择概率最高的假设作为输出。贝叶斯推理的优势在于能处理不确定性——当观测数据不充分时，推理器不会给出错误结论，而是通过主动探针补充数据，提高推理置信度。

三、排障推理引擎的生产级实现

3.1 故障树模型定义与匹配

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

class GateType(Enum):
    """故障树逻辑门类型"""
    OR = "OR"    # 任一子事件触发即导致父事件
    AND = "AND"  # 所有子事件同时触发才导致父事件

@dataclass
class FaultTreeNode:
    """故障树节点"""
    node_id: str
    name: str
    gate: Optional[GateType] = None  # 叶节点无逻辑门
    children: List['FaultTreeNode'] = field(default_factory=list)
    symptoms: List[str] = field(default_factory=list)
    # symptoms: 该节点对应的可观测症状列表
    # 如 ["db_connection_timeout", "slow_query_increase"]

@dataclass
class FaultTree:
    """故障树——从顶事件到底事件的推理路径"""
    tree_id: str
    top_event: FaultTreeNode
    description: str

    def find_matching_hypotheses(
        self, observed_symptoms: Set[str]
    ) -> List[Dict]:
        """根据观察到的症状，在故障树中查找匹配的假设

        匹配逻辑：
        1. 遍历故障树的所有叶节点
        2. 计算每个叶节点的症状与观察症状的交集比例
        3. 返回匹配度超过阈值的叶节点作为假设
        """
        hypotheses = []
        self._traverse(self.top_event, observed_symptoms, hypotheses)
        return sorted(hypotheses, key=lambda h: h['match_score'], reverse=True)

    def _traverse(
        self,
        node: FaultTreeNode,
        observed: Set[str],
        results: List[Dict]
    ):
        """深度优先遍历故障树"""
        if not node.children:
            # 叶节点：计算症状匹配度
            node_symptoms = set(node.symptoms)
            if node_symptoms:
                intersection = node_symptoms & observed
                match_score = len(intersection) / len(node_symptoms)
                if match_score >= 0.3:  # 匹配度阈值 30%
                    results.append({
                        'node_id': node.node_id,
                        'name': node.name,
                        'match_score': round(match_score, 3),
                        'matched_symptoms': list(intersection),
                        'unmatched_symptoms': list(node_symptoms - observed),
                    })
        else:
            # 中间节点：递归遍历子节点
            for child in node.children:
                self._traverse(child, observed, results)

3.2 贝叶斯推理器

from typing import Dict, List, Tuple
import math

class BayesianReasoner:
    """贝叶斯推理器——计算假设根因的后验概率

    核心公式：P(H|E) = P(E|H) * P(H) / P(E)
    - P(H): 先验概率，来自知识图谱中的历史故障频率
    - P(E|H): 似然度，给定假设下观察到症状的概率
    - P(H|E): 后验概率，观察到症状后假设成立的概率
    """

    def __init__(
        self,
        prior_probs: Dict[str, float],
        likelihood_matrix: Dict[str, Dict[str, float]],
        confidence_threshold: float = 0.7
    ):
        # prior_probs: {假设ID: 先验概率}
        # likelihood_matrix: {假设ID: {症状ID: P(症状|假设)}}
        self.prior_probs = prior_probs
        self.likelihood = likelihood_matrix
        self.confidence_threshold = confidence_threshold

    def infer(
        self, observed_symptoms: List[str]
    ) -> Tuple[List[Dict], bool]:
        """执行贝叶斯推理，返回假设排序和是否达到置信阈值

        返回: (假设排序列表, 是否达到置信阈值)
        """
        posteriors = {}

        for hypothesis, prior in self.prior_probs.items():
            # 计算 P(E|H) = 各症状似然度的乘积
            # 假设症状之间条件独立（朴素贝叶斯假设）
            likelihood = 1.0
            for symptom in observed_symptoms:
                # 症状在似然矩阵中的概率，默认使用平滑值
                p = self.likelihood.get(hypothesis, {}).get(
                    symptom, 0.01  # 拉普拉斯平滑，避免零概率
                )
                likelihood *= p

            # 未观察到的症状，乘以 (1 - P(症状|假设))
            all_symptoms = set()
            for h_symptoms in self.likelihood.values():
                all_symptoms.update(h_symptoms.keys())
            for symptom in all_symptoms - set(observed_symptoms):
                p = self.likelihood.get(hypothesis, {}).get(symptom, 0.01)
                likelihood *= (1 - p)

            posteriors[hypothesis] = prior * likelihood

        # 归一化后验概率
        total = sum(posteriors.values())
        if total == 0:
            return [], False

        results = []
        for hypothesis, posterior in posteriors.items():
            normalized = posterior / total
            results.append({
                'hypothesis': hypothesis,
                'posterior_prob': round(normalized, 4),
                'confidence': round(normalized, 4),
            })

        results.sort(key=lambda x: x['posterior_prob'], reverse=True)

        # 检查最高概率假设是否达到置信阈值
        reached = (
            len(results) > 0 and
            results[0]['posterior_prob'] >= self.confidence_threshold
        )

        return results, reached

    def suggest_probes(
        self, observed_symptoms: List[str], top_hypotheses: List[Dict]
    ) -> List[str]:
        """建议需要补充的探针（观测数据）

        选择能最大程度区分 Top 假设的症状进行主动探测
        """
        if len(top_hypotheses) < 2:
            return []

        # 找到 Top 2 假设的似然度差异最大的症状
        h1, h2 = top_hypotheses[0]['hypothesis'], top_hypotheses[1]['hypothesis']
        all_symptoms = set()
        for h_symptoms in self.likelihood.values():
            all_symptoms.update(h_symptoms.keys())

        probe_candidates = []
        for symptom in all_symptoms:
            if symptom in observed_symptoms:
                continue  # 已观测的症状不需要再探测
            p1 = self.likelihood.get(h1, {}).get(symptom, 0.01)
            p2 = self.likelihood.get(h2, {}).get(symptom, 0.01)
            # 信息增益：似然度差异越大，区分能力越强
            info_gain = abs(math.log(p1 + 1e-10) - math.log(p2 + 1e-10))
            probe_candidates.append((symptom, info_gain))

        # 返回区分能力最强的 Top 3 探针
        probe_candidates.sort(key=lambda x: x[1], reverse=True)
        return [s for s, _ in probe_candidates[:3]]

3.3 排障 Agent 编排逻辑

class TroubleshootingAgent:
    """排障 Agent 主循环

    执行流程：
    1. 接收告警事件，提取症状
    2. 在故障树中匹配假设
    3. 用贝叶斯推理计算后验概率
    4. 若置信度不足，发起主动探针补充数据
    5. 输出根因诊断报告
    """

    def __init__(
        self,
        fault_trees: List[FaultTree],
        reasoner: BayesianReasoner,
        max_probe_rounds: int = 3
    ):
        self.fault_trees = fault_trees
        self.reasoner = reasoner
        self.max_probe_rounds = max_probe_rounds

    def diagnose(self, initial_symptoms: List[str]) -> Dict:
        """执行排障推理"""
        observed = set(initial_symptoms)

        for round_num in range(self.max_probe_rounds + 1):
            # 步骤 1：故障树匹配
            all_hypotheses = []
            for tree in self.fault_trees:
                matches = tree.find_matching_hypotheses(observed)
                all_hypotheses.extend(matches)

            if not all_hypotheses:
                return {
                    "status": "no_match",
                    "message": "未找到匹配的故障树假设",
                    "observed_symptoms": list(observed),
                }

            # 步骤 2：贝叶斯推理
            hypotheses, reached = self.reasoner.infer(list(observed))

            if reached or round_num == self.max_probe_rounds:
                # 置信度达标或达到最大探测轮次
                return {
                    "status": "diagnosed" if reached else "low_confidence",
                    "root_cause": hypotheses[0] if hypotheses else None,
                    "all_hypotheses": hypotheses[:5],
                    "observed_symptoms": list(observed),
                    "probe_rounds": round_num,
                }

            # 步骤 3：建议主动探针
            probes = self.reasoner.suggest_probes(list(observed), hypotheses)
            if not probes:
                break

            # 模拟执行探针（生产环境中调用实际的检查脚本）
            for probe in probes:
                # 执行探针检查，将结果加入观测症状
                probe_result = self._execute_probe(probe)
                if probe_result:
                    observed.add(probe_result)

        return {
            "status": "inconclusive",
            "message": "推理轮次耗尽，未能达到置信阈值",
            "all_hypotheses": hypotheses[:5] if hypotheses else [],
            "observed_symptoms": list(observed),
        }

    def _execute_probe(self, probe_name: str) -> Optional[str]:
        """执行主动探针（实际实现中调用具体的检查脚本）"""
        # 示例：根据探针名称执行对应的健康检查
        probe_map = {
            "check_db_connection": "db_connection_timeout",
            "check_disk_io": "disk_io_wait_high",
            "check_connection_pool": "connection_pool_exhausted",
        }
        return probe_map.get(probe_name)

四、排障 Agent 的能力边界与落地风险

故障树构建的冷启动成本：故障树需要领域专家手动构建，初始建设周期通常 2-3 个月。自动学习故障树（从历史工单中提取因果模式）是理想方案，但工单数据的质量和标准化程度往往不足以支撑自动学习。建议从 Top 10 高频故障类型开始，逐步扩展故障树覆盖范围。

朴素贝叶斯的独立性假设：推理器假设症状之间条件独立，但实际运维场景中症状往往高度相关（如 CPU 高和内存高经常同时出现）。独立性假设会导致后验概率估计偏差。更精确的方案是使用贝叶斯网络建模症状间的依赖关系，但构建贝叶斯网络的条件概率表需要大量训练数据，冷启动成本更高。

主动探针的副作用：某些探针（如压力测试、故障注入）可能影响生产环境稳定性。必须对探针进行分类：只读探针（如检查配置、查询指标）可安全执行，写入探针（如重启服务、清理缓存）需要人工确认。

适用边界：排障 Agent 最适合重复性高、模式明确的故障类型（如资源耗尽、配置错误、依赖服务异常）。对于首次出现的未知故障类型，Agent 只能提供辅助分析，无法替代人工排查。

禁用场景：安全事件排查（如入侵检测、数据泄露）不应依赖自动化 Agent，因为安全攻击可能利用 Agent 的推理逻辑进行误导。安全事件需要独立的安全响应流程。

五、总结

AIOps 排障 Agent 的核心价值，在于将运维排障经验从"个人能力"转化为"组织资产"。故障树提供了从症状到根因的推理路径，知识图谱提供了组件依赖和历史模式的上下文，贝叶斯推理器提供了处理不确定性的数学框架。三者结合，Agent 能在秒级完成资深工程师需要数十分钟才能完成的推理过程。

落地步骤：第一步，梳理 Top 10 高频故障类型，构建对应的故障树模型；第二步，从 CMDB 和链路追踪中提取组件依赖关系，构建知识图谱；第三步，基于历史工单统计故障频率，初始化贝叶斯推理器的先验概率；第四步，以只读模式运行 Agent，对比人工排障结果验证准确率；第五步，对准确率超过 80% 的故障类型开放自动诊断和修复建议输出。排障 Agent 的建设是知识工程，不是算法工程——故障树的准确性和知识图谱的完整性，比推理算法的复杂度更重要。