从自然语言SOP到自动化故障处置：LLM Agent如何将纸面化Runbook转化为可执行操作序列的全链路拆解

一、凌晨3:17的故障处置场景：为什么Runbook经常是"有等于无"

凌晨 3:17，支付网关的 P99 延迟飙升至 12 秒，告警已触发。值班运维从床上爬起来，打开企业 Wiki，找到《支付网关高延迟故障处置 Runbook v3.2》。文档写了满满 5 页，包含"步骤一：检查数据库连接池状态"、"步骤二：确认消息队列积压情况"等 12 个操作步骤。但问题在于——这只是自然语言描述的 SOP，而非可执行的操作序列。

运维人员需要将每个自然语言步骤翻译成具体的命令、API 调用和参数组合。这意味着在凌晨 3 点的高压状态下，需要手动完成以下转换：

1. 将"检查数据库连接池状态"翻译为kubectl exec -it mysql-0 -- mysql -e "SHOW PROCESSLIST"，解析输出，判断连接数是否接近max_connections。
2. 将"确认消息队列积压情况"翻译为kafka-consumer-groups --bootstrap-server kafka:9092 --group payment-processor --describe，提取LAG值，判断积压是否超过阈值。
3. 将"如果连接池正常且无消息积压，检查上游服务"翻译为链路追踪查询和依赖服务的健康检查。

一个有经验的运维可能需要 10-15 分钟完成全套排查。但关键问题不是时间，而是信息到动作的转换链路太长——每一步都是一个潜在的出错点。凌晨 3 点键入kubectl delete pod的手，比白天更容易点错目标。

这就是 LLM Agent + Runbook 的切入点：让 LLM 理解自然语言 SOP 的语义，自动将操作步骤映射到工具调用序列，并在执行前进行安全验证。将运维的知识资产（Runbook）从"参考文档"升级为"可执行的自动化脚本"。

二、从自然语言到工具调用的完整转换流水线

这个流水线的核心是用 LLM 的语义理解能力来桥接"自然语言 SOP"和"具体工具调用"之间的鸿沟。具体的工程实现分为四个层级：

第一层：SOP 步骤分解。Runbook 文档通常包含大小标题、编号列表、条件分支。首先需要将文档解析为结构化的步骤树，每个叶节点对应一个可独立执行的检查操作。这一步可以用 LLM 做，也可以用传统的 Markdown 解析器——推荐后者以节约 Token。

第二层：意图分类与工具路由。每个步骤被分类为预定义的意图类型（数据库检查、消息队列检查、链路检查等），然后路由到对应的工具集。这一步的工程挑战在于边界情况：当 LLM 遇到"检查支付网关的 Redis 缓存命中率"这种跨领域的步骤时，需要将其拆解为"连接 Redis → 执行 INFO stats → 解析输出 → 计算命中率"的多个原子工具调用。

第三层：安全执行与权限隔离。所有工具调用必须在沙箱环境中执行，操作被分为三级：只读操作（自动执行）、修改操作（需人工确认）、危险操作（拒绝自动执行，仅输出建议）。这一层不是 LLM 的强项，需要硬编码的安全策略规则。

第四层：结果聚合与根因推断。所有检查步骤的输出被汇总为结构化的诊断报告，然后调用 LLM 进行根因推断。与第一版 RAG 方案的关键区别是：此时 LLM 已经不是凭空猜测，而是基于实际采集的实时数据进行分析。

三、代码实现：一个最小化的 Runbook Agent 引擎

""" Runbook Agent 引擎：将自然语言 SOP 转换为可执行的工具调用序列 核心设计原则： 1. 只读操作为主，修改操作需要人工确认 2. 每个工具的输入输出都有严格的 Schema 校验 3. 执行失败时自动进入"人工介入"通道，而非盲目重试 """ import json import re from typing import List, Dict, Optional, Callable, Any, Tuple from dataclasses import dataclass, field from enum import Enum from abc import ABC, abstractmethod class ActionLevel(Enum): READONLY = "readonly" # 只读操作，自动执行 MUTATING = "mutating" # 修改操作，需人工确认 DANGEROUS = "dangerous" # 危险操作，只输出建议不执行 class StepResult(Enum): SUCCESS = "success" FAILED = "failed" NEEDS_CONFIRMATION = "needs_confirmation" DENIED = "denied" @dataclass class SOPStep: """Runbook 中的单一步骤""" step_id: str # 步骤编号，如 "2.1" description: str # 自然语言描述 intent: str # 意图类型 tools: List[str] # 需要的工具列表 expected_output: str # 预期输出模式 action_level: ActionLevel = ActionLevel.READONLY on_failure: str = "escalate" # 失败处理策略 @dataclass class ToolResult: """工具调用的返回结果""" tool_name: str success: bool output: Any error: Optional[str] = None duration_ms: float = 0.0 @dataclass class DiagnosisReport: """自动化诊断的完整报告""" alert_context: Dict[str, Any] steps_executed: List[Dict[str, Any]] = field(default_factory=list) root_cause_hypothesis: Optional[str] = None confidence: float = 0.0 recommendation: str = "" requires_human_review: bool = True # ===== 工具注册与安全策略 ===== # 工具安全等级映射表 # 危险操作硬编码拒绝，不依赖 LLM 判断 TOOL_SAFETY_POLICY: Dict[str, ActionLevel] = { "kubectl_exec_mysql_query": ActionLevel.READONLY, "kubectl_get_pod_logs": ActionLevel.READONLY, "kubectl_get_pod_status": ActionLevel.READONLY, "kafka_consumer_group_describe": ActionLevel.READONLY, "prometheus_instant_query": ActionLevel.READONLY, "jaeger_trace_search": ActionLevel.READONLY, "kubectl_scale_deployment": ActionLevel.MUTATING, "kubectl_rollout_restart": ActionLevel.MUTATING, "kubectl_delete_pod": ActionLevel.DANGEROUS, "kubectl_exec_rm": ActionLevel.DANGEROUS, } class ToolRegistry: """工具注册中心：管理所有可用工具及其安全等级""" def __init__(self): self._tools: Dict[str, Callable] = {} def register(self, name: str, func: Callable) -> None: """注册一个工具函数""" self._tools[name] = func def get_action_level(self, tool_name: str) -> ActionLevel: """查询工具的安全等级，未注册的工具默认为 DANGEROUS""" return TOOL_SAFETY_POLICY.get(tool_name, ActionLevel.DANGEROUS) def execute(self, tool_name: str, **kwargs) -> ToolResult: """ 执行工具调用，带安全检查 Args: tool_name: 工具名称 **kwargs: 工具参数 Returns: 工具执行结果 """ import time start = time.time() if tool_name not in self._tools: return ToolResult( tool_name=tool_name, success=False, error=f"工具 {tool_name} 未注册" ) # 安全等级检查在调用方（Agent）进行 # 此处仅做执行和异常捕获 try: output = self._tools[tool_name](**kwargs) elapsed = (time.time() - start) * 1000 return ToolResult( tool_name=tool_name, success=True, output=output, duration_ms=elapsed, ) except Exception as e: elapsed = (time.time() - start) * 1000 return ToolResult( tool_name=tool_name, success=False, error=str(e), duration_ms=elapsed, ) # ===== 模拟工具实现 ===== def tool_kubectl_get_pod_status(namespace: str, label_selector: str) -> Dict: """ 查询指定命名空间中 Pod 的状态 返回 Pod 列表及其状态（Running/Pending/CrashLoopBackOff 等） """ # 生产环境中会调用 kubectl Python client 或直接调用 API Server return { "namespace": namespace, "selector": label_selector, "pods": [ {"name": "payment-gateway-7d5f8b9c-abc12", "status": "Running", "restarts": 2, "age": "3d"}, {"name": "payment-gateway-7d5f8b9c-def34", "status": "Running", "restarts": 0, "age": "1h"}, ], "total": 2, "ready": 2, } def tool_kafka_consumer_lag(bootstrap: str, group: str) -> Dict: """ 查询 Kafka 消费者组的 Lag 情况 如果 Lag 持续增长，说明消费者处理速度跟不上生产速度 """ return { "group": group, "topic": "payment-events", "partitions": [ {"partition": 0, "current_offset": 1456789, "log_end_offset": 1456790, "lag": 1}, {"partition": 1, "current_offset": 983421, "log_end_offset": 984200, "lag": 779}, ], "total_lag": 780, "status": "WARNING", # partition 1 存在积压 } def tool_prometheus_query(query: str) -> Dict: """ 执行 PromQL 即时查询 返回查询结果和元数据 """ return { "query": query, "result_type": "vector", "results": [ {"metric": {"pod": "payment-gateway-7d5f8b9c-abc12"}, "value": [1715800000, "320"]}, ], "summary": "当前活跃连接数: 320 / 最大允许: 500", } # ===== Runbook Agent 核心引擎 ===== class RunbookAgent: """ Runbook 自动化执行引擎 核心流程： 1. 解析 Runbook 步骤 2. 对每个步骤执行工具调用 3. 汇总结果并进行根因推断 4. 输出诊断报告 """ def __init__(self, tool_registry: ToolRegistry): self.tools = tool_registry self.confirmation_callback: Optional[Callable] = None def parse_runbook(self, runbook_text: str) -> List[SOPStep]: """ 将自然语言 Runbook 解析为结构化步骤列表 使用正则匹配编号步骤，提取意图和工具映射 """ steps = [] # 匹配形如 "步骤一"、"2."、"2.1" 的步骤标题 pattern = r'(?:步骤\s*([一二三四五六七八九十\d]+)|(?:^|\n)\s*(\d+(?:\.\d+)?)\s*[.、)])' # 简化：按段落分割，每个段落的标题作为 step_id return self._llm_parse_runbook(runbook_text) def _llm_parse_runbook(self, runbook_text: str) -> List[SOPStep]: """ 使用 LLM 将 Runbook 文本解析为结构化的 SOPStep 列表 这里使用预定义的步骤模板简化实现， 生产环境需要调用实际的 LLM API 进行解析 """ # 模拟 LLM 解析结果 steps = [ SOPStep( step_id="1", description="检查 payment-gateway Pod 运行状态", intent="pod_health_check", tools=["kubectl_get_pod_status"], expected_output="所有 Pod 状态为 Running", ), SOPStep( step_id="2", description="确认 Kafka 消费者组积压情况", intent="kafka_lag_check", tools=["kafka_consumer_group_describe"], expected_output="所有分区 Lag < 1000", ), SOPStep( step_id="3", description="查询数据库连接池当前活跃连接数", intent="db_pool_check", tools=["prometheus_instant_query"], expected_output="活跃连接数 < max_connections 的 80%", ), SOPStep( step_id="4", description="若以上均正常，尝试重启 payment-gateway 以释放可能的资源泄漏", intent="remediation", tools=["kubectl_rollout_restart"], expected_output="重启成功", action_level=ActionLevel.MUTATING, on_failure="escalate_to_human", ), ] return steps def execute_step(self, step: SOPStep, context: Dict) -> Dict: """ 执行 Runbook 中的单个步骤 根据安全等级决定执行策略： - READONLY: 自动执行 - MUTATING: 检查是否需要人工确认 - DANGEROUS: 拒绝执行，仅输出建议 Args: step: SOP 步骤对象 context: 诊断上下文（告警信息、已执行步骤的结果等） Returns: 步骤执行结果 """ # 安全等级前置检查 level = ActionLevel.READONLY for tool_name in step.tools: tool_level = self.tools.get_action_level(tool_name) # 取步骤中所有工具的最高安全等级 if tool_level.value > level.value: level = tool_level results = [] for tool_name in step.tools: if level == ActionLevel.DANGEROUS: results.append({ "tool": tool_name, "result": "denied", "message": f"工具 {tool_name} 被安全策略拒绝自动执行，" f"已转为人工操作建议", }) continue if level == ActionLevel.MUTATING: # 需要人工确认的操作 if self.confirmation_callback: approved = self.confirmation_callback(step, context) if not approved: results.append({ "tool": tool_name, "result": "needs_confirmation", "message": "等待人工确认中", }) continue # 构建工具调用参数（生产环境中由 LLM 生成） params = self._build_tool_params(tool_name, context) result = self.tools.execute(tool_name, **params) results.append({ "tool": tool_name, "success": result.success, "output": result.output, "error": result.error, "duration_ms": result.duration_ms, }) all_success = all(r.get("success", False) for r in results) return { "step_id": step.step_id, "description": step.description, "action_level": level.value, "results": results, "overall": "success" if all_success else "failed", } def _build_tool_params(self, tool_name: str, context: Dict) -> Dict: """ 根据工具名称和上下文构建调用参数 生产环境中，此处应调用 LLM 根据 Context 和 Step 描述生成参数 """ params_map = { "kubectl_get_pod_status": { "namespace": context.get("namespace", "default"), "label_selector": "app=payment-gateway", }, "kafka_consumer_group_describe": { "bootstrap": "kafka-broker:9092", "group": "payment-processor", }, "prometheus_instant_query": { "query": "mysql_global_status_threads_connected", }, } return params_map.get(tool_name, {}) def diagnose(self, alert: Dict, runbook_text: str) -> DiagnosisReport: """ 完整的自动化诊断流程 Args: alert: 告警信息（服务名、指标、当前值等） runbook_text: 对应的 Runbook 文本 Returns: 完整的诊断报告 """ report = DiagnosisReport(alert_context=alert) # 1. 解析 Runbook steps = self.parse_runbook(runbook_text) # 2. 逐步执行 context = alert.copy() for step in steps: step_result = self.execute_step(step, context) report.steps_executed.append(step_result) # 将当前步骤的结果合并到上下文中，供后续步骤参考 context[f"step_{step.step_id}"] = step_result # 如果某步骤失败且策略要求升级，停止自动诊断 if step_result["overall"] == "failed" and step.on_failure == "escalate_to_human": report.recommendation = ( f"步骤 {step.step_id} 执行失败，诊断已自动升级。" f"请人工检查以下输出：\n{json.dumps(step_result, indent=2, ensure_ascii=False)}" ) report.requires_human_review = True return report # 3. 汇总所有结果，调用 LLM 进行根因推断 summary = self._summarize_results(report.steps_executed) report.root_cause_hypothesis = summary["hypothesis"] report.confidence = summary["confidence"] report.recommendation = summary["recommendation"] # 4. 置信度低于阈值时，标记需要人工审核 report.requires_human_review = report.confidence < 0.75 return report def _summarize_results(self, step_results: List[Dict]) -> Dict: """ 汇总各步骤的诊断结果，进行根因推断 生产环境需调用 LLM API 完成模拟推理 """ # 模拟分析逻辑 has_kafka_lag = any( "lag" in str(r) for step in step_results for r in step.get("results", []) ) if has_kafka_lag: return { "hypothesis": "Kafka partition 1 存在消息积压（Lag=779），" "推测为消费者处理变慢导致支付延迟", "confidence": 0.82, "recommendation": "建议扩容 payment-processor 消费者实例数，" "或检查消费者 GC 日志确认是否存在 Full GC 停顿", } else: return { "hypothesis": "所有检查项均无异常，问题可能在上游依赖或网络层", "confidence": 0.45, "recommendation": "建议手动检查上游 API Gateway 和网络链路状态", } # ===== 使用示例 ===== if __name__ == "__main__": # 初始化工具注册中心 registry = ToolRegistry() registry.register("kubectl_get_pod_status", tool_kubectl_get_pod_status) registry.register("kafka_consumer_group_describe", tool_kafka_consumer_lag) registry.register("prometheus_instant_query", tool_prometheus_query) # 初始化 Agent agent = RunbookAgent(registry) # 模拟告警 alert = { "service": "payment-gateway", "metric": "p99_latency", "value": 12.0, "unit": "seconds", "threshold": 5.0, "namespace": "production", "timestamp": "2026-07-16T03:17:00Z", } # 对应的 Runbook 文本（简化） runbook = """ # 支付网关高延迟故障处置 SOP ## 步骤一：检查 Pod 状态 检查 payment-gateway 所有 Pod 的运行状态，确认无 CrashLoopBackOff 或 Pending。 ## 步骤二：确认消息队列积压 使用 kafka-consumer-groups 命令检查 payment-processor 消费者组的 Lag 值。 ## 步骤三：数据库连接池检查 查询 MySQL 当前活跃连接数，确认未超过 max_connections 的 80%。 ## 步骤四：重启服务 如果以上检查均正常，尝试滚动重启 payment-gateway 释放可能的连接泄漏。 """ # 执行诊断 report = agent.diagnose(alert, runbook) print(f"根因推断: {report.root_cause_hypothesis}") print(f"置信度: {report.confidence}") print(f"建议: {report.recommendation}") print(f"需要人工审核: {report.requires_human_review}")

四、Agent 自动化处置的不可逾越红线

Runbook Agent 的能力边界必须被严格限定。以下四条是建议的硬约束：

禁止直接操作生产数据库的写操作。Agent 可以执行SHOW PROCESSLIST和EXPLAIN，但绝对不能执行KILL CONNECTION或任何 DDL/DML 语句。数据库的连接管理需要理解业务上下文的运维人员来做决策。

禁止修改集群级别的安全配置。kubectl edit networkpolicy、修改 RBAC 规则、变更 Pod Security Policy——这些操作涉及安全边界，Agent 的最高操作权限应限定在"提供建议的命令"层级。

操作必须具有可逆性。滚动重启是安全的（自动创建新 Pod，旧 Pod 优雅终止），但kubectl delete pvc（删除持久卷声明，数据不可恢复）必须被拦截。Agent 的任何修改操作都必须具备自动或手动的回滚路径。

故障处置后的验证闭环。Agent 执行操作后，必须自动触发一轮验证检查——确认受影响服务的指标是否恢复正常、是否有新的异常告警产生。如果验证失败，Agent 应自动回滚操作并升级给人工处理。

五、总结

LLM Agent + Runbook 的自动化故障处置方案，本质上是将运维团队积累的隐性知识（Runbook 文档中的操作经验）转化为可执行的显性自动化。它不能替代资深运维的判断力，但可以在凌晨 3 点的高压时刻，将"需要查什么、怎么查、顺序是什么"这三个最常见的决策负担自动化掉。

落地建议分两个阶段推进：

第一阶段（对标并行）：Agent 在旁路模式运行——它读取告警和 Runbook，执行诊断，输出诊断报告，但所有操作建议仅推送给人工，不真正执行。通过对比 Agent 诊断结论与人工排查结论，校准 Agent 的准确率。至少要积累 50 次以上的"Agent vs 人工"对比数据后，才进入下一阶段。

第二阶段（有限自动执行）：只读诊断步骤全自动执行，修改操作需人工确认。同时建立"操作审计日志"，记录 Agent 的每一次工具调用和决策链路，确保故障复盘时可以回溯完整的诊断过程。

最终的目标是让 Runbook Agent 成为运维团队的外挂"第二大脑"，在不增加人员负担的前提下，将故障发现到首次诊断的时间从分钟级缩短到秒级。