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

2026/7/15 22:32:39

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

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

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

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

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

能力	工具	示例
指标查询	Prometheus API	`query_range(vllm:gpu_cache_usage_perc, 30m)`
日志检索	Loki / ElasticSearch	`{job="vllm"}
代码/配置变更	Git API	查看最近合并的 PR
系统命令执行	SSH / kubectl exec	`nvidia-smi`,`dmesg`,`free -h`
对比分析	同集群多节点指标差异	node-03 KV Cache > node-01 × 2.3

三、排障 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"