AI 智能巡检报告每天自动生成的数据健康日报大家好我是朱大喜上篇文章讲了数据质量监控体系的搭建今天续上一篇——怎么用 AI 把枯燥的质量监控数据变成一份有温度的日报。这个需求来自于我们老板的一句话每天那么多监控数据我看不过来能不能每天早上给我一份像样的数据健康报告一、需求分析从监控数据到可读报告数据质量监控系统上线后我们每天会产生大量的检测结果。正常通过的占绝大多数异常的有十几条。这些数据静静躺在 MySQL 里只有出了问题才有人去看。但业务同学和 leader 们需要的是每天早上打开企业微信就能看到一份简洁明了的数据健康日报——昨天哪些数据出了问题严重程度如何修复了没有哪些指标有异常波动需要关注传统的做法是写一个定时 Python 脚本把数据拉出来拼成 Markdown 格式发送。但问题是数据状态千变万化固定的模板很难把值得关注的事说清楚。有时候告警很多但都是微小波动有时候只有一条告警但影响面很大——模板无法区分优先级。这时候 AI 就很合适了。把结构化监控数据扔给 LLM让它按照新闻简报的方式生成日报自动筛选重点、过滤噪点、用自然语言把数据和影响说清楚。下面是整体流程。二、数据准备日报素材的提取与组织日报的素材来自三个源头质量检测日志哪个表什么规则没通过、核心指标数据DAU、订单量、GMV 等关键指标的波动、任务调度日志哪些 ETL 任务延迟或失败了。我们需要把这些数据汇总成一个结构化的 JSON作为 LLM 的输入上下文。import pymysql import json from datetime import datetime, timedelta # 日报素材提取器 class DailyReportDataCollector: 收集每日巡检报表所需的所有原始数据 def __init__(self, db_config): self.db_config db_config def 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_time FROM data_quality_check_log ql JOIN data_quality_rules r ON ql.rule_id r.rule_id WHERE DATE(ql.check_time) %s AND ql.status IN (FAIL, ERROR) ORDER BY FIELD(r.severity, CRITICAL, HIGH, MEDIUM, LOW), ql.check_time LIMIT 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) 1 conn.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 SELECT metric_name, today_value, avg_7d_value, std_7d, (today_value - avg_7d_value) / NULLIF(std_7d, 0) AS z_score, CASE WHEN (today_value - avg_7d_value) / NULLIF(std_7d, 0) 2 THEN 显著上升 WHEN (today_value - avg_7d_value) / NULLIF(std_7d, 0) -2 THEN 显著下降 ELSE 正常 END AS trend FROM ( SELECT DAU AS metric_name, SUM(CASE WHEN dt %s THEN dau ELSE 0 END) AS today_value, AVG(CASE WHEN dt %s AND dt DATE_SUB(%s, 7) THEN dau END) AS avg_7d_value, STDDEV(CASE WHEN dt %s AND dt DATE_SUB(%s, 7) THEN dau END) AS std_7d FROM core_metrics_daily WHERE dt DATE_SUB(%s, 7) AND dt %s ) t WHERE ABS((today_value - avg_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 anomalies def 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_minutes FROM etl_task_log WHERE DATE(scheduled_time) %s AND (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_tasks def build_report_context(self, report_dateNone): 构建日报上下文 JSON 将三类素材打包成 LLM 可直接理解的结构 if report_date is None: report_date (datetime.now() - timedelta(days1)).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 context def _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 * 5 score 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_asciiFalse, indent2, defaultstr))三、LLM 生成日报从结构化数据到自然语言拿到了结构化素材接下来就是调用 LLM 生成报告。核心是在 Prompt 中明确日报的格式要求、优先级排序规则、以及语言风格。import openai # AI 日报生成器 class DailyReportGenerator: 使用 LLM 将结构化素材生成为日报 def __init__(self, api_key): self.client openai.OpenAI(api_keyapi_key) def generate_report(self, context): 调用 LLM 生成日报 # 构建 Prompt prompt f你是一个专业的数据巡检日报生成助手。请根据以下结构化数据生成一份简洁、专业的数据健康日报。 ## 日报内容要求 1. **总体健康评级**一句话概括昨天的数据健康状态 2. **关键问题摘要**按严重程度排序最多列出 3 个需要关注的问题 3. **核心指标波动**列出异常波动的指标及变化幅度 4. **任务运行状态**汇总 ETL 任务延迟/失败情况 5. **关注建议**针对当前问题给出 1-2 条行动建议 ## 格式要求 - 使用 Markdown 格式 - 总体评级用醒目的标识 - 问题描述要具体包含表名/规则名/影响范围 - 语言风格专业但不生硬像同事之间发的每日简报 - 控制在 500 字以内 ## 原始数据 {json.dumps(context, ensure_asciiFalse, indent2)} 请生成今日的数据健康日报。 response self.client.chat.completions.create( modelgpt-4o, messages[ {role: system, content: 你是一个专业的数据巡检报告生成助手善于把数据问题用简明扼要的方式呈现。}, {role: user, content: prompt} ], temperature0.3, # 低温度确保输出稳定 max_tokens800 ) return response.choices[0].message.content def send_to_wechat(self, report_text): 通过企业微信机器人推送日报 webhook_url https://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyYOUR_KEY payload { msgtype: markdown, markdown: { content: report_text } } import requests resp requests.post(webhook_url, jsonpayload) 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 来提升可读性和智能化程度。一步到位追求完美反而容易翻车。
