AI 自动化运维：从 Runbook 编排到智能决策的运维 Agent 架构

AI 自动化运维：从 Runbook 编排到智能决策的运维 Agent 架构

一、Runbook 的困境：当"标准操作手册"遇上"非标准故障"

运维团队维护了 200 多个 Runbook，覆盖了从服务重启到数据库切换的各种标准操作流程。但现实是：60% 的故障不在 Runbook 覆盖范围内，需要人工分析判断；30% 的故障虽然匹配了 Runbook，但执行过程中遇到了 Runbook 未预料的异常，需要人工介入；只有 10% 的故障能完全按照 Runbook 自动化执行。

更深层的问题是，Runbook 是静态的——它描述的是"已知故障的标准解法"，而运维的核心挑战恰恰是处理"未知故障"。AI 自动化运维的目标不是替代 Runbook，而是在 Runbook 的基础上增加智能决策层——当故障匹配到 Runbook 时自动执行，当故障超出 Runbook 范围时，通过 Agent 的推理能力给出处置建议，由人工确认后执行。本文将从运维 Agent 的架构设计出发，深入分析 AI 自动化运维的工程实现。

二、运维 Agent 架构：从规则执行到智能推理的分层设计

运维 Agent 的核心架构是"感知→推理→执行"的闭环。感知层采集系统状态，推理层基于状态做出决策，执行层将决策转化为操作。关键设计原则是"人在回路"——高风险操作必须经过人工确认，低风险操作可以自动执行。

flowchart TD subgraph 感知层["感知层：多源数据采集"] S1[指标数据<br/>Prometheus] S2[日志数据<br/>ELK] S3[告警数据<br/>Alertmanager] S4[变更数据<br/>Git / CMDB] S5[拓扑数据<br/>服务网格] end subgraph 推理层["推理层：智能决策引擎"] R1[故障分类器<br/>匹配已知故障模式] R2[Runbook 匹配<br/>查找对应的自动化流程] R3[LLM 推理<br/>处理未知故障模式] R4[风险评估<br/>评估操作的影响和风险] end subgraph 执行层["执行层：分级操作执行"] E1[L0: 自动执行<br/>低风险操作<br/>服务重启/日志轮转] E2[L1: 人工确认<br/>中风险操作<br/>扩缩容/配置变更] E3[L2: 人工执行<br/>高风险操作<br/>数据库切换/版本回滚] end 感知层 --> 推理层 推理层 --> 执行层 执行层 --> |执行结果反馈| 感知层 R1 --> |匹配成功| R2 R1 --> |匹配失败| R3 R2 --> R4 R3 --> R4 R4 --> |风险低| E1 R4 --> |风险中| E2 R4 --> |风险高| E3

感知层是 Agent 的"眼睛"。多源数据采集确保 Agent 获得完整的系统状态视图。单一数据源（如仅依赖告警）会导致推理不完整——告警只告诉你"出了什么问题"，但指标和日志才能告诉你"为什么出了问题"。

推理层是 Agent 的"大脑"。故障分类器将当前故障与已知模式匹配，匹配成功则执行对应 Runbook；匹配失败则交由 LLM 推理，基于多源数据生成处置建议。风险评估是推理的关键环节——每个操作都必须评估其影响范围和风险等级，决定执行方式。

执行层是 Agent 的"手"。分级执行确保高风险操作不会误执行。L0 级操作（如重启一个无状态服务）可以自动执行，L1 级操作（如扩缩容）需要人工确认，L2 级操作（如数据库主从切换）需要人工执行。

三、生产级运维 Agent 实现

#!/usr/bin/env python3 """ 运维 Agent 核心引擎 感知 → 推理 → 执行 的闭环实现 """ import json import time import hashlib from dataclasses import dataclass, field from typing import Optional from enum import Enum from collections import defaultdict from datetime import datetime class RiskLevel(Enum): """操作风险等级""" L0 = "auto" # 自动执行 L1 = "confirm" # 人工确认 L2 = "manual" # 人工执行 class FaultCategory(Enum): """故障分类""" RESOURCE_EXHAUSTION = "resource_exhaustion" # 资源耗尽 SERVICE_UNAVAILABLE = "service_unavailable" # 服务不可用 NETWORK_PARTITION = "network_partition" # 网络分区 CONFIGURATION_ERROR = "configuration_error" # 配置错误 DEPENDENCY_FAILURE = "dependency_failure" # 依赖故障 UNKNOWN = "unknown" # 未知故障 @dataclass class SystemState: """系统状态快照：感知层的输出""" alerts: list[dict] # 活跃告警 metrics: dict[str, float] # 关键指标 topology: dict[str, list[str]] # 服务拓扑 recent_changes: list[dict] # 近期变更记录 timestamp: datetime = field(default_factory=datetime.now) @dataclass class Diagnosis: """诊断结果：推理层的输出""" fault_category: FaultCategory root_service: str affected_services: list[str] confidence: float # 诊断置信度 0-1 evidence: list[str] # 支撑诊断的证据 recommended_actions: list[dict] # 推荐操作列表 runbook_id: Optional[str] = None # 匹配的 Runbook ID @dataclass class Action: """操作定义""" action_id: str action_type: str # restart / scale / config_change / failover / drain target: str # 目标服务或节点 parameters: dict # 操作参数 risk_level: RiskLevel description: str estimated_impact: str # 预估影响描述 rollback_command: str # 回滚命令 class FaultClassifier: """ 故障分类器 基于规则 + 指标模式匹配，将故障归入已知类别 """ def classify(self, state: SystemState) -> FaultCategory: """ 根据系统状态判断故障类别 优先级：资源耗尽 > 依赖故障 > 服务不可用 > 网络分区 > 配置错误 """ metrics = state.metrics # 资源耗尽：CPU/内存/磁盘超过阈值 if (metrics.get("cpu_usage_percent", 0) > 90 or metrics.get("memory_usage_percent", 0) > 90 or metrics.get("disk_usage_percent", 0) > 85): return FaultCategory.RESOURCE_EXHAUSTION # 依赖故障：上游服务异常导致下游连锁反应 alert_services = [a.get("service", "") for a in state.alerts] if self._is_cascade_failure(alert_services, state.topology): return FaultCategory.DEPENDENCY_FAILURE # 服务不可用：单个服务异常 critical_alerts = [ a for a in state.alerts if a.get("severity") == "critical" ] if critical_alerts: return FaultCategory.SERVICE_UNAVAILABLE # 网络分区：连接超时或丢包 if (metrics.get("packet_loss_percent", 0) > 1 or metrics.get("connection_timeout_rate", 0) > 0.05): return FaultCategory.NETWORK_PARTITION # 近期有变更：可能是配置错误 if state.recent_changes: return FaultCategory.CONFIGURATION_ERROR return FaultCategory.UNKNOWN def _is_cascade_failure( self, alert_services: list[str], topology: dict[str, list[str]] ) -> bool: """判断告警是否呈级联模式（上游故障影响下游）""" if len(alert_services) < 2: return False # 检查告警服务之间是否存在依赖关系 for svc in alert_services: deps = topology.get(svc, []) for dep in deps: if dep in alert_services: return True return False class RunbookMatcher: """ Runbook 匹配器 根据故障类别和受影响服务，查找对应的自动化流程 """ def __init__(self): # Runbook 注册表：category + service → runbook_id self._runbooks: dict[str, str] = {} def register(self, category: FaultCategory, service: str, runbook_id: str): """注册 Runbook""" key = f"{category.value}:{service}" self._runbooks[key] = runbook_id def match( self, category: FaultCategory, service: str ) -> Optional[str]: """查找匹配的 Runbook""" # 精确匹配：类别 + 服务 key = f"{category.value}:{service}" if key in self._runbooks: return self._runbooks[key] # 模糊匹配：类别 + 通配符 wildcard_key = f"{category.value}:*" if wildcard_key in self._runbooks: return self._runbooks[wildcard_key] return None class RiskAssessor: """ 操作风险评估器 根据操作类型和目标服务评估风险等级 """ # 服务关键度：决定操作的风险等级 SERVICE_CRITICALITY = { "mysql-primary": "critical", "redis-cluster": "high", "kafka": "high", "api-gateway": "medium", "user-service": "medium", "order-service": "medium", "payment-service": "critical", } # 操作类型的基础风险 ACTION_BASE_RISK = { "restart": RiskLevel.L0, "scale": RiskLevel.L1, "config_change": RiskLevel.L1, "failover": RiskLevel.L2, "drain": RiskLevel.L1, } def assess(self, action: Action) -> RiskLevel: """ 评估操作的风险等级 规则：基础风险 + 服务关键度修正 """ base_risk = self.ACTION_BASE_RISK.get( action.action_type, RiskLevel.L2 ) # 服务关键度修正：关键服务的操作升级一个风险等级 criticality = self.SERVICE_CRITICALITY.get( action.target, "medium" ) if criticality == "critical": # 关键服务：L0 → L1, L1 → L2, L2 保持 if base_risk == RiskLevel.L0: return RiskLevel.L1 if base_risk == RiskLevel.L1: return RiskLevel.L2 return base_risk class OperationsAgent: """ 运维 Agent 核心引擎 串联感知、推理、执行三层 """ def __init__(self): self.classifier = FaultClassifier() self.runbook_matcher = RunbookMatcher() self.risk_assessor = RiskAssessor() self._setup_runbooks() def _setup_runbooks(self): """注册标准 Runbook""" # 资源耗尽类 Runbook self.runbook_matcher.register( FaultCategory.RESOURCE_EXHAUSTION, "*", "RB-RES-001" ) # 服务不可用类 Runbook self.runbook_matcher.register( FaultCategory.SERVICE_UNAVAILABLE, "api-gateway", "RB-SVC-API-001" ) self.runbook_matcher.register( FaultCategory.SERVICE_UNAVAILABLE, "user-service", "RB-SVC-USER-001" ) # 依赖故障类 Runbook self.runbook_matcher.register( FaultCategory.DEPENDENCY_FAILURE, "*", "RB-DEP-001" ) def diagnose(self, state: SystemState) -> Diagnosis: """ 执行诊断：感知 → 推理 返回诊断结果，包含故障类别、根因和推荐操作 """ # 第一步：故障分类 category = self.classifier.classify(state) # 第二步：定位根因服务 root_service = self._locate_root_service(state, category) # 第三步：确定影响范围 affected = self._find_affected_services(root_service, state.topology) # 第四步：匹配 Runbook runbook_id = self.runbook_matcher.match(category, root_service) # 第五步：生成推荐操作 actions = self._generate_actions( category, root_service, affected, state ) # 第六步：收集诊断证据 evidence = self._collect_evidence(state, category, root_service) # 第七步：计算置信度 confidence = self._compute_confidence(category, runbook_id, evidence) return Diagnosis( fault_category=category, root_service=root_service, affected_services=affected, confidence=confidence, evidence=evidence, recommended_actions=actions, runbook_id=runbook_id, ) def execute_action(self, action: Action) -> dict: """ 执行操作：根据风险等级决定执行方式 """ # 重新评估风险等级 assessed_risk = self.risk_assessor.assess(action) action.risk_level = assessed_risk if assessed_risk == RiskLevel.L0: # L0：自动执行 return self._auto_execute(action) elif assessed_risk == RiskLevel.L1: # L1：需要人工确认 return { "status": "pending_confirmation", "action": action, "message": ( f"操作 [{action.description}] 风险等级 L1，" f"需要人工确认后执行" ), } else: # L2：需要人工执行 return { "status": "manual_required", "action": action, "message": ( f"操作 [{action.description}] 风险等级 L2，" f"需要人工执行。回滚命令: {action.rollback_command}" ), } def _locate_root_service( self, state: SystemState, category: FaultCategory ) -> str: """定位根因服务""" if category == FaultCategory.RESOURCE_EXHAUSTION: # 资源耗尽：找到资源使用率最高的服务 max_metric = "" max_service = "" for key, value in state.metrics.items(): if "usage_percent" in key and value > (state.metrics.get(max_metric, 0)): max_metric = key max_service = key.split("_")[0] return max_service or "unknown" if category == FaultCategory.DEPENDENCY_FAILURE: # 依赖故障：找到告警中最上游的服务 alert_services = set(a.get("service", "") for a in state.alerts) for svc in alert_services: deps = state.topology.get(svc, []) if not any(d in alert_services for d in deps): return svc # 默认：取第一个 Critical 告警的服务 for alert in state.alerts: if alert.get("severity") == "critical": return alert.get("service", "unknown") return "unknown" def _find_affected_services( self, root: str, topology: dict[str, list[str]] ) -> list[str]: """查找受影响的服务""" affected = [root] visited = {root} queue = [root] while queue: current = queue.pop(0) for svc, deps in topology.items(): if current in deps and svc not in visited: affected.append(svc) visited.add(svc) queue.append(svc) return affected def _generate_actions( self, category: FaultCategory, root: str, affected: list[str], state: SystemState, ) -> list[dict]: """根据故障类别生成推荐操作""" actions = [] if category == FaultCategory.RESOURCE_EXHAUSTION: # 资源耗尽：扩容 + 清理 actions.append({ "action_type": "scale", "target": root, "parameters": {"replicas": "+2"}, "description": f"扩容 {root} 增加 2 个副本", "estimated_impact": "扩容期间服务可用性不受影响", "rollback_command": f"kubectl scale deployment {root} --replicas=当前值-2", }) actions.append({ "action_type": "restart", "target": root, "parameters": {}, "description": f"重启 {root} 释放内存碎片", "estimated_impact": "短暂不可用（约 10 秒）", "rollback_command": "无需回滚", }) elif category == FaultCategory.SERVICE_UNAVAILABLE: # 服务不可用：重启 + 检查依赖 actions.append({ "action_type": "restart", "target": root, "parameters": {}, "description": f"重启不可用服务 {root}", "estimated_impact": "服务短暂不可用", "rollback_command": "无需回滚", }) elif category == FaultCategory.DEPENDENCY_FAILURE: # 依赖故障：修复根因服务 actions.append({ "action_type": "restart", "target": root, "parameters": {}, "description": f"重启根因服务 {root}", "estimated_impact": "依赖链上的服务可能短暂受影响", "rollback_command": "无需回滚", }) elif category == FaultCategory.CONFIGURATION_ERROR: # 配置错误：回滚最近的变更 if state.recent_changes: latest = state.recent_changes[0] actions.append({ "action_type": "config_change", "target": latest.get("service", "unknown"), "parameters": {"revert_to": latest.get("previous_version")}, "description": f"回滚配置变更: {latest.get('description', '')}", "estimated_impact": "服务需要重启以加载旧配置", "rollback_command": f"git revert {latest.get('commit', '')}", }) # 将操作字典转化为 Action 对象并评估风险 result = [] for a in actions: action = Action( action_id=hashlib.md5( f"{a['action_type']}:{a['target']}:{time.time()}".encode() ).hexdigest()[:8], action_type=a["action_type"], target=a["target"], parameters=a["parameters"], risk_level=RiskLevel.L0, # 初始值，后续由 assessor 修正 description=a["description"], estimated_impact=a["estimated_impact"], rollback_command=a["rollback_command"], ) action.risk_level = self.risk_assessor.assess(action) result.append({ "action_id": action.action_id, "action_type": action.action_type, "target": action.target, "risk_level": action.risk_level.value, "description": action.description, "estimated_impact": action.estimated_impact, "rollback_command": action.rollback_command, }) return result def _collect_evidence( self, state: SystemState, category: FaultCategory, root: str, ) -> list[str]: """收集诊断证据""" evidence = [] evidence.append(f"故障类别: {category.value}") evidence.append(f"根因服务: {root}") for key, value in state.metrics.items(): if "usage_percent" in key and value > 80: evidence.append(f"指标异常: {key} = {value:.1f}%") for alert in state.alerts: if alert.get("severity") in ("critical", "warning"): evidence.append( f"告警: [{alert.get('severity')}] " f"{alert.get('service')} - {alert.get('summary', '')}" ) return evidence def _compute_confidence( self, category: FaultCategory, runbook_id: Optional[str], evidence: list[str], ) -> float: """计算诊断置信度""" confidence = 0.5 # 基础置信度 # 有匹配的 Runbook 提升置信度 if runbook_id: confidence += 0.2 # 故障类别不是 UNKNOWN 提升置信度 if category != FaultCategory.UNKNOWN: confidence += 0.1 # 证据充分提升置信度 if len(evidence) >= 3: confidence += 0.1 return min(confidence, 1.0) def _auto_execute(self, action: Action) -> dict: """自动执行操作（L0 级别）""" # 生产环境应替换为实际的执行逻辑 # 如调用 kubectl API、Ansible playbook 等 return { "status": "executed", "action_id": action.action_id, "action_type": action.action_type, "target": action.target, "message": f"已自动执行: {action.description}", "timestamp": datetime.now().isoformat(), } # 使用示例 if __name__ == "__main__": agent = OperationsAgent() # 模拟系统状态 state = SystemState( alerts=[ {"service": "mysql-primary", "severity": "critical", "summary": "MySQL 主库连接池耗尽"}, {"service": "user-service", "severity": "warning", "summary": "用户服务查询超时"}, {"service": "order-service", "severity": "warning", "summary": "订单服务查询超时"}, ], metrics={ "mysql_cpu_usage_percent": 92.5, "mysql_memory_usage_percent": 88.3, "mysql_disk_usage_percent": 72.1, "api-gateway_cpu_usage_percent": 45.0, }, topology={ "api-gateway": ["user-service", "order-service"], "user-service": ["mysql-primary", "redis-cluster"], "order-service": ["mysql-primary", "kafka"], "mysql-primary": [], "redis-cluster": [], "kafka": [], }, recent_changes=[], ) # 执行诊断 diagnosis = agent.diagnose(state) print(f"诊断结果:") print(f" 故障类别: {diagnosis.fault_category.value}") print(f" 根因服务: {diagnosis.root_service}") print(f" 影响范围: {diagnosis.affected_services}") print(f" 置信度: {diagnosis.confidence:.2f}") print(f" Runbook: {diagnosis.runbook_id}") print(f" 证据:") for e in diagnosis.evidence: print(f" - {e}") print(f" 推荐操作:") for action in diagnosis.recommended_actions: print(f" - [{action['risk_level']}] {action['description']}") print(f" 影响: {action['estimated_impact']}")

四、运维 Agent 的边界：自动化与可控性的永恒张力

LLM 推理的不可靠性：当故障超出 Runbook 覆盖范围时，Agent 需要依赖 LLM 生成处置建议。但 LLM 的输出不可预测——可能生成错误的操作命令（如删除生产数据），可能遗漏关键步骤，可能对故障的严重性判断错误。解决方案是"LLM 生成 + 规则校验"——LLM 的输出必须经过规则引擎校验（如命令白名单、参数范围检查），校验通过后才能进入执行流程。

级联操作的风险放大：Agent 执行一个操作后，可能触发新的告警，Agent 再次诊断并执行操作，形成级联。如果第一次操作的方向错误，级联效应会放大错误的影响。解决方案是设置"操作冷却期"——同一服务在 5 分钟内只允许执行一次自动操作，后续操作需要人工确认。

状态感知的完整性：Agent 的推理质量取决于感知层的数据完整性。如果指标数据缺失、拓扑数据过期、告警数据延迟，Agent 的诊断可能基于不完整的信息做出错误决策。生产环境必须确保感知层的数据质量——指标采集的完整性、拓扑数据的实时性、告警数据的准确性。

人在回路的效率瓶颈：L1/L2 级操作需要人工确认，但人工确认的响应时间通常在 5-15 分钟。如果故障快速恶化，等待确认的时间窗口可能导致故障扩大。解决方案是引入"渐进式自动化"——随着 Agent 的诊断准确率提升，逐步将 L1 操作降级为 L0 自动执行，但 L2 操作始终保持人工确认。

五、总结

运维 Agent 的核心架构是"感知→推理→执行"的闭环，关键设计原则是"人在回路"——低风险操作自动执行提升效率，高风险操作人工确认保障安全。故障分类器和 Runbook 匹配器处理已知故障模式，LLM 推理处理未知故障模式，风险评估器决定操作执行方式。但 Agent 的可靠性受限于感知数据的质量、LLM 推理的不可靠性和级联操作的风险放大，必须在自动化与可控性之间找到平衡。

落地路线建议：第一步，实现感知层和故障分类器，验证诊断准确率；第二步，注册核心 Runbook，实现 L0 级自动执行；第三步，引入 LLM 推理处理未知故障，但所有输出必须经过规则校验；第四步，持续度量 Agent 的诊断准确率和操作成功率，逐步扩大自动化范围。Agent 的能力边界必须清晰——它是一个辅助工具，不是替代运维工程师的方案。