AI 智能巡检报告：每天自动生成的“数据健康日报“

AI 智能巡检报告：每天自动生成的"数据健康日报"

一、需求分析：从监控数据到可读报告

数据质量监控系统上线后，我们每天会产生大量的检测结果。正常通过的占绝大多数，异常的有十几条。这些数据静静躺在 MySQL 里，只有出了问题才有人去看。

但业务同学和 leader 们需要的是每天早上打开企业微信，就能看到一份简洁明了的"数据健康日报"——昨天哪些数据出了问题？严重程度如何？修复了没有？哪些指标有异常波动需要关注？

传统的做法是写一个定时 Python 脚本，把数据拉出来拼成 Markdown 格式发送。但问题是：数据状态千变万化，固定的模板很难把"值得关注的事"说清楚。有时候告警很多但都是微小波动，有时候只有一条告警但影响面很大——模板无法区分优先级。

这时候 AI 就很合适了。把结构化监控数据扔给 LLM，让它按照"新闻简报"的方式生成日报，自动筛选重点、过滤噪点、用自然语言把数据和影响说清楚。下面是整体流程。

二、数据准备：日报素材的提取与组织

日报的素材来自三个源头：质量检测日志（哪个表什么规则没通过）、核心指标数据（DAU、订单量、GMV 等关键指标的波动）、任务调度日志（哪些 ETL 任务延迟或失败了）。

我们需要把这些数据汇总成一个结构化的 JSON，作为 LLM 的输入上下文。

import pymysqlimport jsonfrom datetime import datetime, timedelta# ========== 日报素材提取器 ==========class DailyReportDataCollector:"""收集每日巡检报表所需的所有原始数据"""def __init__(self, db_config):self.db_config = db_configdef collect_quality_issues(self, report_date):"""收集昨天的数据质量异常按严重程度分级，取 Top 10"""conn = pymysql.connect(**self.db_config)# 昨日质量检测异常汇总sql = """SELECT ql.rule_name,ql.table_name,ql.actual_value,ql.expect_value,ql.status,r.severity,r.owner,ql.check_timeFROM data_quality_check_log qlJOIN data_quality_rules r ON ql.rule_id = r.rule_idWHERE DATE(ql.check_time) = %sAND ql.status IN ('FAIL', 'ERROR')ORDER BY FIELD(r.severity, 'CRITICAL', 'HIGH', 'MEDIUM', 'LOW'),ql.check_timeLIMIT 10"""with conn.cursor(pymysql.cursors.DictCursor) as cursor:cursor.execute(sql, (report_date,))quality_issues = cursor.fetchall()# 统计各级别数量severity_count = {}for issue in quality_issues:sev = issue['severity']severity_count[sev] = severity_count.get(sev, 0) + 1conn.close()return {'total_issues': len(quality_issues),'severity_distribution': severity_count,'top_issues': quality_issues}def collect_metric_anomalies(self, report_date):"""收集核心业务指标的异常波动判断逻辑：当日值与近7日均值偏差超过3倍标准差"""conn = pymysql.connect(**self.db_config)sql = """SELECTmetric_name,today_value,a vg_7d_value,std_7d,(today_value - a vg_7d_value) / NULLIF(std_7d, 0) AS z_score,CASE WHEN (today_value - a vg_7d_value) / NULLIF(std_7d, 0) > 2 THEN '显著上升'WHEN (today_value - a vg_7d_value) / NULLIF(std_7d, 0) < -2 THEN '显著下降'ELSE '正常'END AS trendFROM (SELECT 'DAU' AS metric_name,SUM(CASE WHEN dt = %s THEN dau ELSE 0 END) AS today_value,A VG(CASE WHEN dt < %s AND dt >= DATE_SUB(%s, 7) THEN dau END) AS a vg_7d_value,STDDEV(CASE WHEN dt < %s AND dt >= DATE_SUB(%s, 7) THEN dau END) AS std_7dFROM core_metrics_dailyWHERE dt >= DATE_SUB(%s, 7) AND dt <= %s) tWHERE ABS((today_value - a vg_7d_value) / NULLIF(std_7d, 0)) > 2"""with conn.cursor(pymysql.cursors.DictCursor) as cursor:cursor.execute(sql, (report_date, report_date, report_date, report_date, report_date, report_date, report_date))anomalies = cursor.fetchall()conn.close()return anomaliesdef collect_task_sla(self, report_date):"""收集 ETL 任务 SLA 情况统计延迟和失败的任务"""conn = pymysql.connect(**self.db_config)sql = """SELECT task_name,scheduled_time,actual_start_time,actual_end_time,status,TIMESTAMPDIFF(MINUTE, scheduled_time, COALESCE(actual_end_time, NOW())) AS delay_minutesFROM etl_task_logWHERE DATE(scheduled_time) = %sAND (status = 'FAILED'OR TIMESTAMPDIFF(MINUTE, scheduled_time, COALESCE(actual_end_time, NOW())) > 30)ORDER BY delay_minutes DESC"""with conn.cursor(pymysql.cursors.DictCursor) as cursor:cursor.execute(sql, (report_date,))delayed_tasks = cursor.fetchall()conn.close()return delayed_tasksdef build_report_context(self, report_date=None):"""构建日报上下文 JSON将三类素材打包成 LLM 可直接理解的结构"""if report_date is None:report_date = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')# 收集三类素材quality = self.collect_quality_issues(report_date)metrics = self.collect_metric_anomalies(report_date)tasks = self.collect_task_sla(report_date)# 构建结构化上下文context = {'report_date': report_date,'summary': {'quality_issues': quality['total_issues'],'metric_anomalies': len(metrics),'delayed_tasks': len(tasks)},'overall_health': self._evaluate_health(quality, metrics, tasks),'quality_issues': quality,'metric_anomalies': metrics,'task_status': tasks}return contextdef _evaluate_health(self, quality, metrics, tasks):"""综合评估数据健康等级"""score = 100# 每有一个 CRITICAL 质量问题扣 20 分score -= quality['severity_distribution'].get('CRITICAL', 0) * 20# 每有一个 HIGH 质量问题扣 10 分score -= quality['severity_distribution'].get('HIGH', 0) * 10# 每有一个指标异常扣 5 分score -= len(metrics) * 5# 每有一个任务失败扣 5 分task_failures = sum(1 for t in tasks if t['status'] == 'FAILED')score -= task_failures * 5score = max(0, min(100, score))if score >= 90:return {'level': '优秀', 'score': score, 'emoji': '🟢'}elif score >= 70:return {'level': '良好', 'score': score, 'emoji': '🟡'}elif score >= 50:return {'level': '一般', 'score': score, 'emoji': '🟠'}else:return {'level': '较差', 'score': score, 'emoji': '🔴'}# ========== 使用示例 ==========if __name__ == '__main__':collector = DailyReportDataCollector({'host': '***', **db_config})context = collector.build_report_context('2026-07-21')print(json.dumps(context, ensure_ascii=False, indent=2, default=str))

三、LLM 生成日报：从结构化数据到自然语言

拿到了结构化素材，接下来就是调用 LLM 生成报告。核心是在 Prompt 中明确日报的格式要求、优先级排序规则、以及语言风格。

import openai# ========== AI 日报生成器 ==========class DailyReportGenerator:"""使用 LLM 将结构化素材生成为日报"""def __init__(self, api_key):self.client = openai.OpenAI(api_key=api_key)def generate_report(self, context):"""调用 LLM 生成日报"""# 构建 Promptprompt = f"""你是一个专业的数据巡检日报生成助手。请根据以下结构化数据，生成一份简洁、专业的"数据健康日报"。## 日报内容要求：1. **总体健康评级**：一句话概括昨天的数据健康状态2. **关键问题摘要**：按严重程度排序，最多列出 3 个需要关注的问题3. **核心指标波动**：列出异常波动的指标及变化幅度4. **任务运行状态**：汇总 ETL 任务延迟/失败情况5. **关注建议**：针对当前问题给出 1-2 条行动建议## 格式要求：- 使用 Markdown 格式- 总体评级用醒目的标识- 问题描述要具体，包含表名/规则名/影响范围- 语言风格：专业但不生硬，像同事之间发的每日简报- 控制在 500 字以内## 原始数据：{json.dumps(context, ensure_ascii=False, indent=2)}请生成今日的数据健康日报。"""response = self.client.chat.completions.create(model="gpt-4o",messages=[{"role": "system", "content": "你是一个专业的数据巡检报告生成助手，善于把数据问题用简明扼要的方式呈现。"},{"role": "user", "content": prompt}],temperature=0.3,# 低温度确保输出稳定max_tokens=800)return response.choices[0].message.contentdef send_to_wechat(self, report_text):"""通过企业微信机器人推送日报"""webhook_url = 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY'payload = {'msgtype': 'markdown','markdown': {'content': report_text}}import requestsresp = requests.post(webhook_url, json=payload)return resp.json()

LLM 在这个场景里的优势，其实一下就能看出来。面对同一份原始数据，传统脚本吐出的日报往往是“xx 表行数校验失败，实际 3200，期望 >5000”这种偏机械、偏冰冷的表达；但换成 LLM，输出就会变成“‘订单明细表’今日产出行数异常偏低（仅 3200 行，通常为 5000+），可能导致下游订单报表数据不完整，建议优先排查上游数据管道”——不只把异常说清楚了，连影响范围和处理建议也一并交代到位。

四、定时调度与持续优化

整个日报生成的流程通过 Airflow 定时调度：每天早上 7:30 触发，先跑数据收集脚本，再调 LLM 生成，8:00 准时推送到企业微信群。从数据采集到推送完成，整个过程不超过 3 分钟。

上线一个月后，我们根据反馈做了几轮优化。最关键的改进是增加了"问题追踪"机制：如果昨天日报里的问题今天还没解决，会自动在今日日报中追加"持续未解决问题"模块，记录已持续天数。这让日报从一个"资讯通报"变成了一个"问题追踪器"。

另一个有趣的优化是让 LLM 学习汇报风格。老板喜欢看数据趋势，运营同学关注影响范围，开发同学更在意根因定位。我们做了一个简单的角色适配——根据接收群组不同，调整 Prompt 中的侧重点。

五、总结

数据健康日报这个项目虽然不大，但很好地体现了"A + 数据"的融合价值。A 的部分不是替代数据分析，而是提升数据分析成果的消费体验——让枯燥的监控数据变成一份大家愿意看、看得懂的日报。

从技术角度说，核心是 Prompt Engineering。日报质量的好坏，七成取决于 Prompt 对格式、优先级、语言风格的描述是否精确。另外两成取决于上下文数据的结构化程度——如果扔给 LLM 的是一堆乱糟糟的原始数据，神仙也写不出好日报。

如果你也在考虑做类似的巡检报告，我的建议是：先从最简单的模板开始玩，等大家对"日报"这个形式建立信任后，再逐步引入 AI 来提升可读性和智能化程度。一步到位追求完美，反而容易翻车。