AI 数据治理 Agent：从被动发现问题到主动修复的升级路径

AI 数据治理 Agent：从被动发现问题到主动修复的升级路径

一、数据治理的"打地鼠"困境

做数据治理的人都有一个共同的痛：问题永远是事后才知道。

上周二早上9点，运营跑来问我："大喜，为什么昨天的 GMV 环比跌了 40%？"我心里一紧，开始排查——ETL 任务延迟了 3 小时导致分区数据没更新。问题是，这个延迟在凌晨 3 点就发生了，但没人知道，直到上班后才被发现。

这就是典型的被动数据治理：监控报警 → 人工排查 → 找到原因 → 手动修复 → 补数据。每一步都是人在驱动，数据质量靠"打地鼠"维持。

我们团队去年开始尝试用 AI Agent 来自动化数据治理，从"出问题再修"变成了"主动巡检 + 自动修复"，今天就分享这套升级路径。

flowchart LR subgraph 传统模式["传统被动模式"] T1[数据异常发生] --> T2[业务方发现] --> T3[人工排查] --> T4[手动修复] end subgraph Agent模式["AI Agent 主动模式"] A1[定时巡检任务] --> A2{异常检测} A2 -->|发现异常| A3[AI诊断根因] A3 --> A4{可自动修复?} A4 -->|是| A5[自动执行修复SQL] A4 -->|否| A6[生成工单 + 修复建议] A5 --> A7[自动验证修复结果] A6 --> A8[通知值班人员] end style T4 fill:#E74C3C,color:#fff style A5 fill:#27AE60,color:#fff style A7 fill:#27AE60,color:#fff

二、数据治理 Agent 的三层能力

我们的 Agent 按照能力成熟度分了三层，每一层解决不同的问题：

第一层：巡检与发现

传统监控只能检测"这条 SQL 有没有报错"，但数据质量问题远不止报错。空分区、数据量突降、字段枚举值缺失——这些问题 SQL 本身不报错，但业务影响很大。

Agent 的巡检模块做了这些事：

# quality_checker.py — AI 数据质量巡检 Agent import pandas as pd from datetime import datetime, timedelta from typing import Dict, List, Optional class DataQualityAgent: """数据质量巡检 Agent：自动扫描并诊断数据问题""" def __init__(self, db_connection, rules_config: dict): self.db = db_connection self.rules = rules_config # 巡检规则配置 def run_daily_check(self, target_date: str) -> List[Dict]: """每日巡检入口：执行所有注册的检查规则""" issues = [] # 检查1：分区完整性 # 每个核心表都必须有当天的分区，漏分区的风险最高 for table in self.rules.get("partition_tables", []): if not self._check_partition_exists(table, target_date): issues.append({ "type": "missing_partition", "table": table, "partition": target_date, "severity": "critical", # 缺少分区属于严重问题 "auto_fixable": True, # 可以自动补跑ETL }) # 检查2：数据量波动 # 环比下降超过阈值可能是ETL中途挂掉，上涨超过阈值可能是重复跑 for table_config in self.rules.get("volume_check_tables", []): row_count = self._get_row_count(table_config["name"], target_date) yesterday_count = self._get_row_count( table_config["name"], self._prev_date(target_date) ) if yesterday_count > 0: change_pct = (row_count - yesterday_count) / yesterday_count threshold = table_config.get("volume_threshold", 0.3) if abs(change_pct) > threshold: issues.append({ "type": "volume_anomaly", "table": table_config["name"], "partition": target_date, "current_count": row_count, "yesterday_count": yesterday_count, "change_pct": round(change_pct * 100, 1), "severity": "warning" if abs(change_pct) < 0.5 else "critical", "auto_fixable": False, # 数据量异常需要人工判断原因 }) # 检查3：空值率检查 for col_config in self.rules.get("null_check_columns", []): null_rate = self._get_null_rate( col_config["table"], col_config["column"], target_date ) if null_rate > col_config.get("threshold", 0.1): issues.append({ "type": "high_null_rate", "table": col_config["table"], "column": col_config["column"], "null_rate": round(null_rate * 100, 1), "severity": "warning", "auto_fixable": False, }) return issues

第二层：根因诊断

发现问题是第一步，更重要的是找到根因。这是 AI 真正能发力的地方。

传统做法是数据工程师一个一个排查——看上游任务日志、检查数据源变更、对比历史数据。这个过程动辄半小时起。Agent 拿到问题列表后，会并行执行以下诊断：

1. 上游依赖检查：查调度系统，看该表的父任务有没有异常
2. 数据源变更检测：对比今天的 schema 和昨天是否一致（有没有删字段、改类型）
3. 数据内容比对：抽样今天的样本分布和昨天的分布做相似度检测
4. 历史相似问题匹配：从问题知识库里找是否有类似案例及解决方案

# root_cause.py — 根因诊断引擎 def diagnose_partition_missing(self, table: str, partition: str) -> str: """诊断分区缺失的根因""" # 1. 查询上游 ETL 任务状态 upstream_tasks = self._get_upstream_tasks(table) failed_tasks = [t for t in upstream_tasks if self._task_failed(t, partition)] if failed_tasks: # 上游任务失败是最常见的原因 return ( f"根因：上游ETL任务 {failed_tasks[0]['name']} " f"在 {partition} 分区执行失败。\n" f"失败日志摘要：{failed_tasks[0]['error_log'][:200]}\n" f"建议：先修复 {failed_tasks[0]['name']}，再重跑 {table} 的分区。" ) # 2. 检查数据源是否变更 source_table = self._get_source_table(table) if source_table and not self._check_partition_exists(source_table, partition): return ( f"根因：数据源表 {source_table} 缺少 {partition} 分区。\n" f"可能原因：源头数据未产出，建议联系数据接入方确认。" ) # 3. 未知原因，交给 AI 推理 return self._ai_diagnose(table, partition, "missing_partition")

第三层：自动修复

诊断出根因后，有些问题可以自动修复。我们目前支持这些场景：

自动修复的核心是修完必须验证。Agent 不会执行完之后拍拍屁股走人，它会再次运行巡检规则，确认问题已解决。如果没解决，自动升级为人工工单。

三、AI 的分析能力：从规则到推理

上面的规则检查只是基础。AI Agent 真正厉害的地方是能处理没有明确规则的问题：

场景：字段枚举值偏移。你有一个"用户等级"字段，值的集合应该是 {VIP1, VIP2, VIP3, VIP4, VIP5}。突然有一天出现了 "VIP-1"、"VIP-2" 这种新格式。规则引擎很难捕捉——它不是空值，也没有报错。

AI Agent 的做法：

# enum_drift_detect.py — 枚举值漂移检测 def detect_enum_drift(self, table: str, column: str, partition: str) -> dict: """检测枚举字段的值域是否发生漂移（新值出现或旧值消失）""" # 1. 获取今天和历史的值分布 today_values = set(self._get_distinct_values(table, column, partition)) baseline_values = self._get_baseline_values( table, column, lookback_days=7 # 用过去7天作为基线，既稳定又能反映最近变化 ) # 2. 计算差异 new_values = today_values - baseline_values lost_values = baseline_values - today_values if new_values or lost_values: # 3. AI 分析新值的模式 analysis = self._ai_analyze_values( table, column, new_values=list(new_values), lost_values=list(lost_values) ) # AI 可能输出："VIP-1"和"VIP-2"很可能是"VIP1""VIP2"加了横线， # 可能是上游系统的格式变更 return { "table": table, "column": column, "new_values": list(new_values), "lost_values": list(lost_values), "ai_analysis": analysis, "severity": "warning", "suggestion": self._ai_generate_fix_suggestion(analysis) } return {"status": "normal"}

这种"值看起来像什么"的判断，正是大模型的强项。规则引擎需要你提前定义所有合法值，而 AI 可以用语义理解来判断"VIP-1"大概率就是"VIP1"的格式变体。

四、落地经验：Automation 的边界

做了快一年 Agent 治理，最重要的教训是：不是所有问题都适合自动修复。

我们的原则是：

• 可逆的、影响面小的→ 自动修复（如补跑一个分区）
• 不可逆的、影响面大的→ 自动生成方案，人工确认后再执行（如修改表结构）
• 不确定根因的→ 自动诊断 + 人工决策

另外还有两条血泪教训：

教训1：修复要有审计日志。自动修复了什么东西、什么时候修的、修之前和修之后的数据是什么样的，全部记下来。否则出了问题时你连"Agent 干了什么"都搞不清楚。

教训2：灰度上线。先在低优表上跑一个月，确认 Agent 不会"过度治疗"（比如把正常波动当异常修复），再推广到核心表。我们曾经 Agent 把周末的流量低谷判定为"数据异常"，自动补了一堆默认值——教训深刻。

五、总结

数据治理 Agent 的本质是把数据工程师的"排查-诊断-修复"流程自动化。它不等于完全无人值守，而是把人的精力从重复性排查中解放出来，去做更有价值的架构优化和治理设计。

升级路径也很清晰：

1. 先做好巡检——让问题"被发现"而不是"被报告"
2. 再做根因诊断——让 AI 帮你缩小排查范围
3. 最后做自动修复——但一定要留人工确认的开关

一句话收尾：最好的数据治理，是你根本感觉不到它在治理。

我是朱大喜，一个被数据质量问题折磨到自建 Agent 的数据分析师。你的团队是怎么做数据治理的？还在靠"打地鼠"吗？评论区聊聊~