我基于开源 AIOps 项目做了一个 SRE 实验排障 AI Agent

引言：从手动排障到智能 Agent在 SRE（站点可靠性工程）的日常工作中，故障排查是一项既耗时又需要深厚经验的技能。传统方式依赖运维人员通过日志、指标和告警一步步分析，效率低下且容易遗漏关键信息。随着 AIOps（智能运维）的发展，开源社区涌现了一系列工具，如 OpenAIOps、Kubent、Prometheus 等，它们提供了数据采集、异常检测和根因分析的基础能力。但仅仅有工具还不够——我们需要一个能自动执行排障流程的“Agent”。本文将深入剖析如何基于开源 AIOps 项目，构建一个针对实验环境的排障 AI Agent。我会从原理层面拆解其架构，并提供可运行的代码片段，让你能亲自上手实践。## 核心原理：AIOps 与 Agent 的融合一个排障 AI Agent 的核心是“感知-分析-决策-执行”闭环。在 SRE 实验中，Agent 需要：1. 感知：从 Prometheus、Elasticsearch 等获取指标和日志。2. 分析：利用异常检测算法（如基于统计的 3-sigma 或机器学习模型）识别异常。3. 决策：通过规则引擎或 LLM（大语言模型）生成排查步骤。4. 执行：调用 Kubernetes API 或 SSH 命令修复问题。开源 AIOps 项目提供了感知和分析的基础，而 Agent 则负责决策和执行。我将使用 OpenAIOps（一个轻量级 AIOps 框架）作为底层，结合 Python 的 asyncio 实现事件驱动式排障。## 构建排障 AI Agent 的实战步骤### 环境准备首先，你需要一个模拟的 SRE 实验环境。假设我们有一个 Kubernetes 集群，运行一个 Web 服务。我们将使用以下开源组件：- Prometheus：采集指标（如 CPU、内存、请求延迟）。- OpenAIOps：提供异常检测算法和告警管理。- LangChain：用于构建 LLM 驱动的推理链。### 代码示例 1：集成 Prometheus 与异常检测以下代码展示了如何从 Prometheus 拉取指标，并使用 OpenAIOps 的统计方法检测异常。注释详细解释了每一步的原理。pythonimport asynciofrom prometheus_api_client import PrometheusConnectfrom openaiops.anomaly import StatisticalDetectorimport numpy as np# 连接 Prometheus（假设在本地 9090 端口）prom = PrometheusConnect(url="http://localhost:9090", disable_ssl=True)async def fetch_and_detect_anomaly(): # 1. 感知阶段：获取最近 5 分钟的 CPU 使用率指标 query = 'rate(container_cpu_usage_seconds_total[1m])' result = prom.custom_query(query=query, params={'time': '5m'}) # 提取时间序列数据 timestamps = [] values = [] for item in result: for sample in item['values']: timestamps.append(float(sample[0])) values.append(float(sample[1])) if not values: print("无数据，跳过检测") return # 2. 分析阶段：使用 3-sigma 方法检测异常 detector = StatisticalDetector(method='zscore', threshold=3.0) anomalies = detector.detect(values) # 标记异常点 for i, is_anomaly in enumerate(anomalies): if is_anomaly: print(f"时间 {timestamps[i]} 的 CPU 使用率 {values[i]:.2f} 为异常") return anomalies# 运行异步检测asyncio.run(fetch_and_detect_anomaly())原理剖析：这里的 StatisticalDetector 基于 z-score 计算每个数据点偏离均值的程度。如果某个点的 z-score 大于 3，则认为异常。这适用于平稳性指标，但实际 SRE 场景中可能需要更复杂的算法（如季节性分解）。### 代码示例 2：构建 LLM 驱动的排障推理链当异常被检测到后，Agent 需要推理出根因并生成修复步骤。以下代码使用 LangChain 和一个简单规则，将异常指标转化为排查建议。pythonfrom langchain.llms import OpenAIfrom langchain.chains import LLMChainfrom langchain.prompts import PromptTemplateimport json# 初始化 LLM（这里使用 OpenAI，但可替换为开源模型如 Llama）llm = OpenAI(temperature=0.2, model_name="gpt-3.5-turbo")# 定义排障提示模板prompt_template = PromptTemplate( input_variables=["metric_name", "anomaly_value", "threshold"], template="""你是一个 SRE 专家。收到以下异常告警：- 指标：{metric_name}- 当前值：{anomaly_value}- 阈值：{threshold}请生成一个排查计划，包括：1. 可能的根因（如资源耗尽、代码 bug、网络问题）2. 确认步骤（如查看 Pod 日志、检查连接数）3. 修复建议（如扩容、回滚版本）输出格式为 JSON。""")# 创建推理链chain = LLMChain(llm=llm, prompt=prompt_template)def debug_anomaly(metric_name: str, current_value: float, threshold: float): # 3. 决策阶段：调用 LLM 生成排障步骤 response = chain.run({ "metric_name": metric_name, "anomaly_value": current_value, "threshold": threshold }) # 解析 JSON 响应 try: plan = json.loads(response) print("排障计划：") for step in plan.get("steps", []): print(f"- {step}") return plan except: print("解析失败，原始响应：", response) return None# 模拟一个异常事件debug_anomaly(metric_name="cpu_usage", current_value=95.0, threshold=80.0)原理剖析：LLM 在这里充当“大脑”，将结构化数据转化为自然语言推理。通过精心设计的提示模板，Agent 能模拟人类 SRE 的思考过程。在生产中，你可以用向量数据库存储历史排查记录，让 LLM 检索相似案例以提高准确性。## 集成与实验：让 Agent 自动执行上述两个代码段分别处理了感知-分析和决策阶段。为了形成一个完整的 Agent，我们需要使用事件循环将它们串联起来。例如，当 fetch_and_detect_anomaly 检测到异常时，自动触发 debug_anomaly，并将结果通过 Webhook 通知用户或执行自动修复（如调用 Kubernetes API 增加副本数）。实验环境验证步骤：1. 启动一个模拟负载生成器（如 Apache Bench）向 Web 服务施压。2. 观察 Agent 是否能在大约 1 分钟内检测到 CPU 异常。3. 检查 LLM 生成的排障计划是否合理（例如建议扩容或排查慢查询）。## 总结通过本文，你看到了如何将开源 AIOps 项目（如 OpenAIOps）与 LLM Agent 结合，构建一个自动化的 SRE 排障系统。核心在于三个要点：- 数据驱动：用 Prometheus 等工具实时采集指标，避免人工观察的滞后性。- 算法与 AI 结合：统计检测用于快速过滤噪音，LLM 用于深度推理，两者互补。- 可扩展性：代码中的模块化设计（异步检测、提示模板）允许你轻松替换组件，比如用时序数据库替代 Prometheus，或用本地模型替代云端 API。这个 Agent 虽然简单，但已能处理实验环境中的典型故障。未来，你可以加入更多数据源（如日志、调用链）和自动修复能力，让它真正成为 24/7 值守的 SRE 助手。记住，AIOps 不是替代人，而是放大人的能力——而一个精心设计的 Agent，就是那个放大镜。