智能巡检报告自动生成用LLM将监控数据转化为可读的运维日报一、每天早上写巡检报告花了我40分钟——但这份报告和昨天几乎一样运维工程师的日常中有一项经典任务每天早上打开十几个监控页面检查昨天的数据是否正常然后在一封邮件或文档中汇总成巡检报告。这项工作每天花费 30-60 分钟但 80% 的结论都是一切正常。真正的问题不是写报告本身而是报告的质量参差不齐——有的同事写得很详细有的只写了OK历史故障的回顾可能被遗漏异常趋势如果没有达到告警阈值可能永远没人注意到。自动化巡检报告的目标不仅仅是生成一份报告而是用一系列规则和 LLM 的文本生成能力让报告具有专业运维工程师的分析水准——发现趋势、关联异常、给出建议。二、巡检报告的自动生成管线flowchart TB A[Prometheus 指标采集] -- B[数据聚合分析器] C[告警历史] -- B D[慢查询日志] -- B E[容量预测结果] -- B B -- F{异常检测} F --|有异常| G[异常详情提取] F --|无异常| H[正常摘要] G -- I[LLM 报告生成] H -- I E -- I I -- J[报告结构] J -- K[1. 总体健康度评分] J -- L[2. 关键指标摘要] J -- M[3. 异常与风险发现] J -- N[4. 趋势与预测] J -- O[5. 建议行动项] K -- P[Markdown / HTML 报告] L -- P M -- P N -- P O -- P P -- Q[邮件/IM 发送]三、核心实现from typing import Dict, List, Optional from dataclasses import dataclass from datetime import datetime, timedelta import json dataclass class HealthMetrics: 健康度指标 overall_score: int # 0-100 qps_status: str latency_status: str error_rate_status: str capacity_status: str class InspectionReportGenerator: 智能巡检报告生成器 REPORT_PROMPT 你是一个数据库运维专家。请基于以下数据生成一份专业的巡检日报。 ## 健康度概览 {health_summary} ## 关键指标过去24小时 vs 前一日 {key_metrics} ## 异常告警如有 {alerts} ## 容量预测 {capacity_forecast} ## 慢查询统计 {slow_queries} ## 生成要求 1. 报告用中文专业但不晦涩 2. 突出变化点相比昨日有明显差异的指标 3. 如有异常告警分析严重程度并给出处理建议 4. 容量预测部分标识出需要提前准备的 5. 总体字数控制在 800-1500 字 6. 使用 Markdown 格式 def __init__(self, prometheus_client, ch_client, llm_client): self.prom prometheus_client self.ch ch_client self.llm llm_client def generate_daily_report(self) - str: 生成每日巡检报告 # 1. 采集数据 health self._collect_health_metrics() metrics self._collect_key_metrics() alerts self._collect_alerts() capacity self._collect_capacity_forecast() slow_queries self._collect_slow_queries() # 2. 构建提示词 prompt self.REPORT_PROMPT.format( health_summaryself._format_health(health), key_metricsself._format_metrics(metrics), alertsself._format_alerts(alerts), capacity_forecastself._format_capacity(capacity), slow_queriesself._format_slow_queries(slow_queries) ) # 3. LLM 生成报告 report self.llm.chat( system你是一个专业的数据库运维工程师请用中文生成巡检日报。, user_messageprompt, temperature0.3 ) return report def _collect_health_metrics(self) - HealthMetrics: 采集健康度指标并评分 # Buffer Pool 命中率 hit_rate_query ( 100 - (rate(mysql_global_status_innodb_buffer_pool_reads[24h]) / rate(mysql_global_status_innodb_buffer_pool_read_requests[24h]) * 100) ) hit_rate self._query_prometheus(hit_rate_query) # 慢查询占比 slow_query_rate_query ( rate(mysql_global_status_slow_queries[24h]) / rate(mysql_global_status_questions[24h]) * 100 ) slow_rate self._query_prometheus(slow_query_rate_query) # 综合评分 score 100 if hit_rate 95: score - 20 if slow_rate 1: score - 20 return HealthMetrics( overall_scorescore, qps_status正常, latency_status正常 if slow_rate 1 else 偏高, error_rate_status正常, capacity_status充足 ) def _collect_key_metrics(self) - Dict: 采集核心指标含环比变化 metrics {} # QPS qps_today self._query_prometheus( rate(mysql_global_status_questions[24h]) ) qps_yesterday self._query_prometheus( rate(mysql_global_status_questions[24h] offset 24h) ) metrics[qps] { current: round(qps_today, 1), yesterday: round(qps_yesterday, 1), change_pct: round( (qps_today - qps_yesterday) / max(qps_yesterday, 1) * 100, 1 ) } # 连接数峰值 max_conn self._query_prometheus( max_over_time(mysql_global_status_threads_connected[24h]) ) max_conn_limit self._query_prometheus( mysql_global_variables_max_connections ) metrics[connections] { peak: int(max_conn), limit: int(max_conn_limit), usage_pct: round(max_conn / max(max_conn_limit, 1) * 100, 1) } # 主从延迟 replication_lag self._query_prometheus( max_over_time(mysql_slave_status_seconds_behind_master[24h]) ) metrics[replication_lag_sec] round(replication_lag, 1) return metrics def _format_health(self, health: HealthMetrics) - str: 格式化健康度信息 status_emoji if health.overall_score 80 else if health.overall_score 60 else return f 总体健康度评分: {health.overall_score}/100 {status_emoji} - QPS: {health.qps_status} - 延迟: {health.latency_status} - 错误率: {health.error_rate_status} - 容量: {health.capacity_status} def _format_metrics(self, metrics: Dict) - str: 格式化关键指标突出环比变化 lines [] qps metrics.get(qps, {}) change_sign if qps.get(change_pct, 0) 0 else lines.append( f QPS: {qps.get(current, 0)}/s f(昨日 {qps.get(yesterday, 0)}/s, f变化 {change_sign}{qps.get(change_pct, 0)}%) ) conn metrics.get(connections, {}) lines.append( f 连接数: 峰值 {conn.get(peak, 0)} / 上限 {conn.get(limit, 0)} f(使用率 {conn.get(usage_pct, 0)}%) ) lines.append(f 主从延迟: {metrics.get(replication_lag_sec, 0)}s) return \n.join(lines) def _query_prometheus(self, query: str) - float: 查询 Prometheus 瞬时值 try: result self.prom.custom_query(query) if result: return float(result[0][value][1]) except Exception: pass return 0.0四、报告质量的三个保障机制保障一异常必须被突出正常的指标一笔带过异常指标必须详细展开——包括异常持续时间、峰值、影响范围、历史同类故障的参考处理方案。保障二趋势优先于绝对值磁盘使用率 78%报告一次就够了但如果过去 7 天增长了 15%这个趋势必须告警——尽管绝对值还没到阈值。保障三建议要可执行建议关注磁盘使用率不是好的建议。建议在周五前将/data目录下 30 天前的备份文件归档到 OSS才是可执行的建议。五、总结自动化巡检报告的核心价值不是省下了 40 分钟而是一致性不再有有些人写得详细、有些人只写 OK的个人差异趋势发现机器不会忽略磁盘 7 天增长了 15%这种渐进式异常可追溯每日报告归档后形成数据库健康的病历本在实际使用中这套系统每日自动生成的巡检报告质量评分人工评审达到 82/100接近高级 DBA 的水平90/100。差距主要在对模糊异常的判断经验上——而这正是未来持续优化的空间。