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 拿到问题列表后会并行执行以下诊断上游依赖检查查调度系统看该表的父任务有没有异常数据源变更检测对比今天的 schema 和昨天是否一致有没有删字段、改类型数据内容比对抽样今天的样本分布和昨天的分布做相似度检测历史相似问题匹配从问题知识库里找是否有类似案例及解决方案# 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)第三层自动修复诊断出根因后有些问题可以自动修复。我们目前支持这些场景问题类型自动修复方式成功率分区缺失上游任务失败已修复自动触发重跑95%空值率高已知原因自动填充默认值或打标签80%数据重复自动执行去重 SQL90%枚举值偏移数据源格式变更自动添加映射规则60%自动修复的核心是修完必须验证。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_days7 # 用过去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_valueslist(new_values), lost_valueslist(lost_values) ) # AI 可能输出VIP-1和VIP-2很可能是VIP1VIP2加了横线 # 可能是上游系统的格式变更 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 的本质是把数据工程师的排查-诊断-修复流程自动化。它不等于完全无人值守而是把人的精力从重复性排查中解放出来去做更有价值的架构优化和治理设计。升级路径也很清晰先做好巡检——让问题被发现而不是被报告再做根因诊断——让 AI 帮你缩小排查范围最后做自动修复——但一定要留人工确认的开关一句话收尾最好的数据治理是你根本感觉不到它在治理。我是朱大喜一个被数据质量问题折磨到自建 Agent 的数据分析师。你的团队是怎么做数据治理的还在靠打地鼠吗评论区聊聊~