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_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 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_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 生成日报"""
# 构建 Prompt
prompt = 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.content
def 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 requests
resp = 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 来提升可读性和智能化程度。一步到位追求完美，反而容易翻车。
