半夜3点被报警电话叫醒?用 Python + AI Agent 打造你的“自动化排障流水线”(附实战代码)

import jsonimport openai# 初始化客户端（这里以通义千问或 OpenAI 兼容接口为例）client = openai.OpenAI( api_key=”your-api-key”, base_url=”https://dashscope.aliyuncs.com/compatible-mode/v1”# 国内推荐用兼容接口)# 模拟一分钟内收到的 5 条突发报警raw_alerts = [ {”time”: ”03:12:01”, ”service”: ”order-service”, ”msg”: ”HTTP 5xx 错误率 > 5%”}, {”time”: ”03:12:05”, ”service”: ”payment-service”, ”msg”: ”接口响应时间 P99 > 2000ms”}, {”time”: ”03:12:10”, ”service”: ”db-node-01”, ”msg”: ”CPU 使用率达到 98%”}, {”time”: ”03:12:15”, ”service”: ”order-service”, ”msg”: ”数据库连接池获取超时”}, {”time”: ”03:12:20”, ”service”: ”gateway”, ”msg”: ”上游 order-service 超时熔断”}]def triage_alerts(alerts): prompt = f””” 你是一个资深 SRE 专家。以下是系统在过去 1 分钟内收到的突发报警列表： {json.dumps(alerts, ensure_ascii=False, indent=2)} 请执行以下操作： 1. 根因推测：分析这些报警的因果关系，推测最可能的根本原因（Root Cause）。 2. 报警聚合：将这些报警归类为 1-2 个核心事件。 3. 紧急定级：给出一个综合紧急等级（P0/P1/P2），并说明理由。 4. 初步建议：给出前 3 分钟应该执行的排查动作。 请严格以 JSON 格式输出，包含字段：root_cause_hypothesis, grouped_events, priority, priority_reason, immediate_actions。 ””” response = client.chat.completions.create( model=”qwen-plus”,# 推荐使用推理能力较强的模型 messages=[{”role”: ”user”, ”content”: prompt}], response_format={”type”: ”json_object”}, temperature=0.1 ) return json.loads(response.choices[0].message.content)result = triage_alerts(raw_alerts)print(json.dumps(result, ensure_ascii=False, indent=2))

{ ”root_cause_hypothesis”: ”db-node-01 CPU 飙升导致数据库处理变慢，进而引发 order-service 连接池耗尽和 5xx 错误，最终导致 payment-service 和 gateway 级联超时。”, ”grouped_events”: [ {”event_name”: ”数据库节点 CPU 异常”, ”related_alerts”: [”db-node-01 CPU 98%”]}, {”event_name”: ”订单与支付链路级联故障”, ”related_alerts”: [”order-service 5xx”, ”payment-service 延迟”, ”gateway 熔断”]} ], ”priority”: ”P0”, ”priority_reason”: ”核心交易链路（订单/支付）已出现级联故障，直接影响业务营收。”, ”immediate_actions”: [ ”1. 立即登录 db-node-01，使用 top 和 pt-query-digest 查看是哪个 SQL 或进程占满了 CPU。”, ”2. 检查 order-service 的慢查询日志，确认是否有全表扫描。”, ”3. 若确认是慢 SQL 导致，考虑临时 Kill 掉长事务或对该 SQL 进行限流。” ]}

import asyncioimport aiohttpfrom datetime import datetime, timedelta# 模拟从不同数据源并发拉取数据async def fetch_logs(session, service, start_time, end_time):# 实际场景：调用 ELK/Loki 的 API await asyncio.sleep(0.5)# 模拟网络延迟 return f”[{service}] 100条日志... 发现大量 ERROR: Connection refused to db-node-01”async def fetch_metrics(session, service, start_time, end_time):# 实际场景：调用 Prometheus/VictoriaMetrics 的 API await asyncio.sleep(0.3) return f”[{service}] CPU: 45%, DB_Connection_Active: 200/200 (已打满)”async def fetch_recent_deploys(session, service):# 实际场景：调用 Jenkins/GitLab API 获取最近发布记录 await asyncio.sleep(0.2) return f”[{service}] 最近一次发布：2小时前，Commit: fix(order): 优化了统计报表SQL”async def gather_context(service_name): now = datetime.now() start_time = now - timedelta(minutes=10) async with aiohttp.ClientSession() as session:# 并发拉取，极大缩短收集时间 tasks = [ fetch_logs(session, service_name, start_time, now), fetch_metrics(session, service_name, start_time, now), fetch_recent_deploys(session, service_name) ] logs, metrics, deploys = await asyncio.gather(*tasks) return { ”service”: service_name, ”time_range”: f”{start_time} to {now}”, ”error_logs”: logs, ”key_metrics”: metrics, ”recent_changes”: deploys }# 运行异步收集context_data = asyncio.run(gather_context(”order-service”))print(”✅ 现场数据收集完毕，准备喂给 AI Agent...”)

from langchain.agents import tool, create_react_agent, AgentExecutorfrom langchain_openai import ChatOpenAIfrom langchain import hubimport json# 1. 定义 Agent 可以使用的“运维工具”@tooldef check_db_slow_queries(db_node: str) -> str: ”””查询指定数据库节点的慢查询日志”””# 实际场景：执行 mysql.slow_log 查询或调用云厂商 API return f”发现 db-node-01 存在大量慢查询：SELECT * FROM orders WHERE create_time > '...' (耗时 4500ms)，未命中索引。”@tooldef get_process_cpu_usage(db_node: str) -> str: ”””获取数据库节点上占用 CPU 最高的进程””” return f”db-node-01 上 mysqld 进程占用 CPU 92%，主要在执行全表扫描。”@tooldef rollback_deployment(service: str, version: str) -> str: ”””回滚指定服务到上一个版本（高危操作，需人工确认）””” return f”⚠️ 拦截：回滚操作 {service} 到 {version} 需要运维人员手动在控制台点击确认。”# 2. 初始化大模型和 Agentllm = ChatOpenAI(model=”qwen-max”, temperature=0)tools = [check_db_slow_queries, get_process_cpu_usage, rollback_deployment]# 使用标准的 ReAct 提示词模板prompt = hub.pull(”hwchase17/react”)agent = create_react_agent(llm, tools, prompt)agent_executor = AgentExecutor( agent=agent,  tools=tools,  verbose=True,# 打印 Agent 的思考过程 handle_parsing_errors=True)# 3. 输入之前收集到的 Context，让 Agent 开始表演initial_context = ”””【故障现场上下文】服务：order-service时间：03:10 - 03:15指标：DB_Connection_Active 已打满 (200/200)日志：大量 ERROR: Connection refused to db-node-01变更记录：2小时前优化了统计报表SQL。请帮我排查根因，并给出处理建议。”””print(”🚀 Agent 开始自主排查...”)result = agent_executor.invoke({”input”: initial_context})print(”\n🎯 最终诊断报告：”)print(result['output'])