基于大模型的诊断自动化——从告警触发到根因分析的端到端排障 Agent 设计

一、"凌晨 3 点的告警，需要 5 分钟定位还是 50 分钟"

一条典型的 P1 告警：VLLMHighTTFT: P99 TTFT 850ms (阈值 500ms, 节点 gpu-node-03)。

守夜的工程师收到告警后的排查路径：

1. 打开 Grafana，查看 node-03 的 GPU 利用率、显存、QPS 趋势（3 分钟）。
2. 对比同集群其他节点的指标，确认是否单点问题（2 分钟）。
3. 查看 node-03 的 vLLM 日志，搜索 ERROR 和 WARN 关键字（5 分钟）。
4. 发现 preemption triggered: no available blocks 日志，确认是 KV Cache 饱和（3 分钟）。
5. 查看该节点的 KV Cache 使用率趋势，确认触发时间（2 分钟）。
6. 检查是否有异常的长 prompt 请求（3 分钟）。
7. 在 Grafana 上找到触发时刻的请求日志，确认是一个 12000 token 的超长 prompt（5 分钟）。
8. 定位到具体请求来源（业务方 A），通知处理（5 分钟）。

总耗时约 28 分钟。其中真正需要人类决策的只有第 8 步（"谁的问题？通知谁？"），前 7 步都是"查这个 → 看那个 → 对比这个"的机械操作。

如果有一个排障 Agent 自动执行前 7 步，工程师在第 1 分钟就能拿到结论："node-03 的 KV Cache 被一个 12000 token 的超长 prompt 填满，来源是业务 A 的批量处理任务。建议通知业务 A 拆分批处理，或为该节点增加 reserved blocks 配置。"

二、排障 Agent 的架构设计

flowchart TD
    A[告警触发<br/>Prometheus Alert] --> B[Agent 接收告警]

    B --> C[任务规划 LLM<br/>将告警转化为排查步骤]

    C --> D{第 1 步: 多维指标查询}
    D --> E[Grafana API<br/>GPU利用率/显存/QPS]
    E --> F{指标对比分析}

    F -->|单点异常| G1[第 2 步: 日志检索]
    F -->|全集群异常| G2[第 2 步: 关联变更记录]

    G1 --> H[Loki/ES 日志查询<br/>error 关键字 + 时间范围]
    H --> I{日志语义分析}

    I --> J[第 3 步: 根因推断 LLM<br/>综合分析 → 根因假设]

    J --> K{验证假设}
    K -->|需要验证| L[第 4 步: 补充查询<br/>如: KV Cache 趋势/请求日志]
    L --> J

    K -->|确认根因| M[第 5 步: 生成诊断报告]

    M --> N{是否需要人工介入}
    N -->|是| O[推送到值班群<br/>含诊断摘要 + 建议方案]
    N -->|否| P[自动执行修复<br/>如: 扩容/限流/重启]

2.1 Agent 的核心能力

三、排障 Agent 的工程实现

3.1 告警触发与任务拆解

# diagnosis_agent.py —— 排障 Agent 核心
import asyncio
import json
import logging
from datetime import datetime, timedelta
from typing import List, Dict, Optional, Any
from dataclasses import dataclass

from openai import AsyncOpenAI

logger = logging.getLogger(__name__)


@dataclass
class DiagnosisStep:
    """排查步骤"""
    step_id: int
    action: str         # 工具调用名称
    args: Dict[str, Any] # 工具参数
    reasoning: str      # LLM 给出的执行理由
    status: str         # "pending" | "running" | "done" | "failed"


@dataclass
class DiagnosisReport:
    """诊断报告"""
    alert_name: str
    alert_time: datetime
    affected_node: str
    root_cause: str          # 根因摘要
    confidence: float         # 置信度 0～1
    evidence: List[str]       # 支撑证据
    recommendation: str       # 修复建议
    steps_executed: List[DiagnosisStep]
    total_time_seconds: float


class DiagnosisAgent:
    """
    排障 Agent

    收到告警后，通过 LLM 规划排查步骤，
    依次调用工具执行，最后生成诊断报告。
    """

    # 可用工具定义（供 LLM 选择）
    TOOLS = [
        {
            "name": "query_prometheus",
            "description": "查询 Prometheus 指标（PromQL）",
            "parameters": {
                "query": "PromQL 查询语句",
                "range_minutes": "查询时间范围（分钟）",
            },
        },
        {
            "name": "query_loki_logs",
            "description": "查询 Loki 日志",
            "parameters": {
                "query": "LogQL 查询语句",
                "start_time": "起始时间（ISO 8601）",
                "end_time": "结束时间（ISO 8601）",
                "limit": "返回条数上限",
            },
        },
        {
            "name": "compare_nodes",
            "description": "对比集群中不同节点的指标差异",
            "parameters": {
                "metric": "指标名（如 gpu_utilization）",
                "target_node": "目标节点名",
                "time_range_minutes": "对比时间范围",
            },
        },
        {
            "name": "check_recent_deployments",
            "description": "检查最近的代码/配置变更",
            "parameters": {
                "service": "服务名",
                "hours_back": "回溯时间（小时）",
            },
        },
        {
            "name": "exec_kubectl",
            "description": "在指定 Pod 内执行命令",
            "parameters": {
                "pod_name": "Pod 名称",
                "command": "要执行的命令",
            },
        },
    ]

    def __init__(self, openai_client: AsyncOpenAI):
        self.client = openai_client
        self.tool_handlers = {
            "query_prometheus": self._handle_prometheus,
            "query_loki_logs": self._handle_loki,
            "compare_nodes": self._handle_compare_nodes,
            "check_recent_deployments": self._handle_deployments,
            "exec_kubectl": self._handle_kubectl,
        }

    async def diagnose(
        self, alert: Dict
    ) -> DiagnosisReport:
        """
        对一条告警执行自动化诊断

        Args:
            alert: {
                "name": "VLLMHighTTFT",
                "labels": {"instance": "gpu-node-03:8000",
                             "model": "qwen2-72b"},
                "annotations": {"summary": "...",
                                  "description": "..."},
                "startsAt": "2024-07-15T03:00:00Z",
            }

        Returns:
            DiagnosisReport 诊断报告
        """
        start_time = datetime.utcnow()
        steps = []

        # =============================================
        # 阶段 1: 任务规划
        # LLM 将告警转化为具体的排查步骤序列
        # =============================================
        plan = await self._plan_steps(alert)
        logger.info(f"排障计划: {len(plan)} 步")
        for s in plan:
            logger.info(f"  Step {s.step_id}: {s.action} - {s.reasoning}")

        # =============================================
        # 阶段 2: 逐步执行
        # =============================================
        findings = []  # 收集每步的发现
        for step in plan:
            step.status = "running"

            try:
                handler = self.tool_handlers.get(step.action)
                if handler is None:
                    step.status = "failed"
                    findings.append(
                        f"Step {step.step_id}: 未知工具 {step.action}")
                    continue

                result = await handler(step.args)
                step.status = "done"
                findings.append(
                    f"Step {step.step_id} ({step.action}): {result}")

            except Exception as e:
                step.status = "failed"
                findings.append(
                    f"Step {step.step_id} 失败: {str(e)}")

            steps.append(step)

        # =============================================
        # 阶段 3: 根因推断
        # LLM 综合分析所有发现，推断根因
        # =============================================
        root_cause, confidence, recommendation = \
            await self._infer_root_cause(
                alert, findings)

        elapsed = (datetime.utcnow() - start_time).total_seconds()

        return DiagnosisReport(
            alert_name=alert["name"],
            alert_time=datetime.fromisoformat(
                alert["startsAt"].replace("Z", "+00:00")),
            affected_node=alert["labels"]["instance"],
            root_cause=root_cause,
            confidence=confidence,
            evidence=findings,
            recommendation=recommendation,
            steps_executed=steps,
            total_time_seconds=elapsed,
        )

    async def _plan_steps(
        self, alert: Dict
    ) -> List[DiagnosisStep]:
        """
        让 LLM 根据告警信息规划排查步骤

        输入：告警名称、描述、影响范围、时间
        输出：5～8 个具体的排查步骤
        """
        prompt = f"""你是一位 SRE 排障专家。收到以下告警：

告警名称: {alert['name']}
告警节点: {alert['labels']['instance']}
告警描述: {alert['annotations']['description']}
告警时间: {alert['startsAt']}

请规划排查步骤，使用以下工具：
{json.dumps(self.TOOLS, indent=2, ensure_ascii=False)}

规则：
1. 优先查询目标节点的关键指标（GPU 利用率、显存、QPS）
2. 对比同集群其他节点，判断是否单点问题
3. 检索相关时间段的错误日志
4. 检查最近的部署/配置变更
5. 每步需要清晰的执行理由

返回 JSON 格式：
[{{
    "step_id": 1,
    "action": "query_prometheus",
    "args": {{"query": "PROMQL", "range_minutes": 30}},
    "reasoning": "检查 GPU 利用率和显存使用趋势"
}}]"""

        response = await self.client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"},
            temperature=0.1,  # 低温度，确保规划稳定
        )

        plan_data = json.loads(
            response.choices[0].message.content)

        return [
            DiagnosisStep(
                step_id=item["step_id"],
                action=item["action"],
                args=item["args"],
                reasoning=item["reasoning"],
                status="pending",
            )
            for item in plan_data.get("steps", [])
        ]

    async def _infer_root_cause(
        self,
        alert: Dict,
        findings: List[str],
    ) -> tuple:
        """综合分析，推断根因"""
        prompt = f"""根据以下告警信息和排查发现，推断根因：

告警: {alert['name']} on {alert['labels']['instance']}
排查发现:
{chr(10).join(f"- {f}" for f in findings)}

请返回 JSON：
{{
    "root_cause": "根因描述（一句话）",
    "confidence": 0.85,
    "recommendation": "修复建议"
}}

规则：
- root_cause 必须具体，不能是"可能是 XX"（给出最可能的原因）
- confidence 根据证据的充分程度给出（0～1）
- 如果证据不足以确定根因，标明还需要哪些信息"""

        response = await self.client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"},
            temperature=0.1,
        )

        result = json.loads(
            response.choices[0].message.content)
        return (
            result.get("root_cause", "无法确定"),
            result.get("confidence", 0.0),
            result.get("recommendation", "请人工介入"),
        )

    # ==============================================
    # 工具处理器（简化实现）
    # ==============================================
    async def _handle_prometheus(self, args):
        """查询 Prometheus"""
        # 实际调用 Prometheus HTTP API
        return f"PromQL: {args['query']} → 结果: [模拟数据]"

    async def _handle_loki(self, args):
        """查询 Loki 日志"""
        return f"LogQL: {args['query']} → 找到 15 条日志"

    async def _handle_compare_nodes(self, args):
        """对比节点指标"""
        return f"node-03 {args['metric']} 比其他节点高 2.3 倍"

    async def _handle_deployments(self, args):
        """检查部署变更"""
        return "最近 24h 无部署变更"

    async def _handle_kubectl(self, args):
        """在 Pod 中执行命令"""
        return f"kubectl exec {args['pod_name']} -- {args['command']} → OK"

3.2 告警严重度分级与自动修复边界

排障 Agent 生成诊断报告后，根据严重度和置信度决定后续动作：

# auto_remediation.py —— 自动修复决策
class RemediationPolicy:
    """
    根据诊断结果决定是自动修复还是人工介入

    规则：
    - P0 告警 + 高置信度(>0.9) → 自动执行修复 + 通知
    - P0 告警 + 低置信度    → 通知 oncall（附带诊断报告）
    - P1 告警 + 高置信度    → 提出修复建议，等待确认
    - P1 告警 + 低置信度    → 仅通知
    - P2 告警               → 记录不通知
    """

    @staticmethod
    def decide(
        severity: str,  # P0/P1/P2
        confidence: float,
    ) -> str:
        if severity == "P0" and confidence > 0.9:
            return "auto_fix_and_notify"
        elif severity == "P0":
            return "notify_oncall"
        elif severity == "P1" and confidence > 0.9:
            return "suggest_fix"
        else:
            return "log_only"

四、排障 Agent 的边界与风险

4.1 幻觉风险——LLM 自信地给出错误结论

排障 Agent 最大的风险是 LLM 的"幻觉"——在证据不足时编造根因。例如看到"GPU 显存用量 95%"就断言"显存泄漏"，但实际是正常的 KV Cache 使用。

缓解方式：

• 置信度低于 0.7 的结论自动升级为人工介入。
• 所有诊断报告的结论必须附带"支撑证据"列表，方便人工复核。
• Agent 不直接执行危险操作（如重启 Pod、释放显存），必须人工确认。

4.2 Token 成本

每次排障的 LLM 调用（规划 + 推断）约消耗 3000～5000 tokens。按 GPT-4 的 API 价格，每次诊断约 0.05～0.10 美元。按每天 20 次 P1/P2 告警计算，月成本约 60 美元——远低于一名 SRE 工程师的时薪。

4.3 工具集的可扩展性

排障 Agent 的能力上限取决于提供的工具数量和质量。如果某个排查需要的工具不在 TOOLS 列表中，Agent 就无法完成诊断——LLM 的规划能力再强也受限于工具边界。定期 Review 未解决的告警，补充新的诊断工具。

五、总结

排障 Agent 的目标不是替代 SRE，而是将机械性的"查指标 → 查日志 → 对比分析"自动化：

1. 任务规划 LLM：将告警转化为 5～8 个具体的排查步骤，每步选择最合适的工具。
2. 工具执行链：依次调用 Prometheus、Loki、kubectl 等工具接口，串行收集证据。
3. 根因推断 LLM：综合分析所有排查发现，给出具体根因和修复建议，附带置信度。
4. 分级的自动修复：高置信度 + 低风险操作（如扩容）可自动执行；其他情况保持人工介入。