1. 项目缘起当数据治理遇上AI一个“数据医生”的诞生在数据驱动的时代公司里最头疼的问题往往不是没有数据而是数据“病了”。我所在的公司业务线繁杂数据源五花八门从传统的业务数据库到各种第三方API、日志文件再到用户上传的Excel表格数据质量参差不齐。每天业务部门、分析师和开发团队都在和数据问题作斗争这个报表的数字怎么对不上为什么这个用户ID查不到信息这个字段为什么有一半是空值数据团队疲于奔命成了“救火队员”80%的时间都花在了数据清洗、核对和解释上而不是创造价值。传统的解决方案比如制定更严格的ETL规范、编写更多的数据校验脚本效果有限。规则是死的数据是活的新的数据问题总是以意想不到的方式出现。直到去年在深入研究了当前AI Agent和智能体工作流的技术趋势后我萌生了一个想法能不能打造一个智能化的“数据医生”它不仅能像资深数据工程师一样自动诊断数据病灶还能开出“处方”甚至直接“动手术”进行修复。这个想法就是今天要分享的“AI数据医生”项目的起点。它不是某个单一的工具而是一个融合了规则引擎、大语言模型LLM和自动化工作流的智能体系统旨在让数据质量问题从“人工排查”走向“智能自治”。2. 核心设计思路从“诊断”到“治疗”的闭环智能体这个“数据医生”的设计核心是模拟一位经验丰富的数据专家的思考和工作流程。它不是一个简单的数据校验工具而是一个具备感知、分析、决策和执行能力的智能体Agent。整个系统的架构围绕“诊、断、治、防”四个环节展开。2.1 智能诊断多维度感知数据健康状况首先数据医生需要具备全面的“体检”能力。我设计了一个多层次的诊断模块基础生命体征检查这是规则引擎层负责快速扫描数据的“硬伤”。比如检查字段是否为空NULL值比例、数值是否在合理范围内如年龄不能为负数、日期格式是否规范、枚举值是否符合预设字典等。这部分使用成熟的框架如Great Expectations或Apache Griffin来实现配置起来快执行效率高。深度影像学检查这部分引入了统计分析和机器学习模型。例如通过分析某个数值字段如交易金额的分布识别离群点Outliers通过关联性分析发现本应有强关联的两个字段如城市和邮编出现矛盾的情况。这里我用Python的Pandas、Scikit-learn库结合一些自定义算法来实现。专家会诊这是最核心的一环即引入大语言模型LLM。将前两步发现的“异常指标”和“可疑数据样本”连同数据表的元数据字段名、业务含义描述一起构造Prompt提交给LLM我选用的是GPT-4的API。LLM的任务是像人类专家一样理解业务语境判断这个“异常”是否真的是问题以及可能的原因是什么。例如规则引擎发现“用户最后登录时间”字段大量为未来日期LLM可能会结合“测试账号”、“数据迁移错误”等业务知识给出更精准的判断。注意LLM的调用成本和控制是关键。不要把所有数据都扔给LLM。我的策略是只有当中低级规则引擎和统计模型发现“疑似复杂问题”时才触发LLM会诊并且严格控制每次提交的上下文长度只发送关键摘要和样本。2.2 精准断症问题分类与根因推理诊断出异常后需要对其进行分类和定级。我建立了一个数据质量问题知识库将问题分为几个大类完整性问题缺失值、记录缺失。准确性问题错误值、格式错误、逻辑矛盾。一致性问题跨表、跨源数据不一致。时效性问题数据更新延迟。LLM在会诊后不仅会判断问题类型还会尝试进行根因推理。例如它可能会输出“问题类型准确性-逻辑矛盾。根因推测疑似‘订单状态’与‘物流状态’的更新不同步或‘已取消’订单的物流信息未及时清空。建议检查订单状态更新流水日志。” 这个推理结果会作为后续处理的宝贵输入。2.3 自动治疗可配置的修复工作流诊断和断症之后就是治疗。我设计了一个可配置的修复动作执行器。根据问题的类型、严重等级和根因推测系统会自动匹配或建议修复方案自动修复对于简单、明确的问题直接执行预设动作。例如将明显的格式错误日期如“20241301”转换为标准格式将某个字段中已知的错别字如“北京”写成“北亰”进行替换。这些动作通过编写Python Pandas函数或SQL更新语句模板来实现由系统自动填充参数后执行。半自动修复需审核对于LLM推测根因但存在一定不确定性的修复或涉及关键业务数据的修改系统会生成修复建议和待执行的脚本提交给数据负责人或创建一个Jira工单进行人工审核。审核通过后一键执行。生成修复建议报告对于复杂问题或暂无自动修复方案的问题系统会生成一份详细的诊断报告包括问题样本、影响范围、根因分析和修复建议供数据工程师参考。整个修复工作流由Apache Airflow或Prefect这样的工作流调度平台来编排确保任务的有序、可重试和可监控。2.4 预防与健康管理持续监控与知识沉淀数据医生的职责不仅是治病还要防病。系统会定期如每天对核心数据资产进行“健康巡检”生成数据质量日报跟踪各项质量指标的趋势。更重要的是所有诊断过的问题、采纳过的修复方案都会被沉淀到案例知识库中。这个知识库有两个作用一是作为未来类似问题的诊断参考提升LLM判断的准确性二是可以反向优化规则引擎将一些新发现的、可固化的模式添加到基础规则中让系统越来越“聪明”。3. 核心配置与关键技术栈拆解下面我来拆解这个“数据医生”的几个核心模块的具体配置和选型思考这些都是可以直接拿去参考的干货。3.1 规则引擎层Great Expectations 实战配置我选择了Great ExpectationsGX因为它不仅是一个校验库更是一个完整的框架支持数据文档生成和结果可视化。核心配置示例以检查某张用户表为例首先你需要定义一个“期望套件”Expectation Suite。这里我通过Python API来配置比用CLI更灵活。import great_expectations as gx import pandas as pd # 1. 初始化上下文和数据源 context gx.get_context() datasource context.sources.add_pandas(namemy_pandas_datasource) # 2. 定义数据资产这里假设df是你的Pandas DataFrame data_asset datasource.add_dataframe_asset(nameuser_table_df, dataframedf) # 3. 创建批处理请求并获取验证器 batch_request data_asset.build_batch_request() validator context.get_validator(batch_requestbatch_request) # 4. 添加具体的“期望”即校验规则 # 检查用户ID非空且唯一 validator.expect_column_values_to_be_unique(columnuser_id) validator.expect_column_values_to_not_be_null(columnuser_id) # 检查年龄在0-120岁之间 validator.expect_column_values_to_be_between(columnage, min_value0, max_value120) # 检查注册日期是合理的过去日期比如不早于公司成立日期2000年 validator.expect_column_values_to_be_between( columnregister_date, min_value2000-01-01, max_valuepd.Timestamp.now().strftime(%Y-%m-%d) ) # 检查性别字段只包含‘M’或‘F’ validator.expect_column_values_to_be_in_set(columngender, value_set[M, F]) # 检查邮箱格式简单正则 validator.expect_column_values_to_match_regex(columnemail, regexr^[^\s][^\s]\.[^\s]$) # 5. 保存期望套件 validator.save_expectation_suite(discard_failed_expectationsFalse)实操心得分阶段实施不要试图一次性为所有表添加几百条规则。先从核心的1-2张表最关键的5-10个字段开始快速跑通流程让业务方看到价值。利用数据文档GX生成的Data Docs数据文档非常有用它自动将你的期望规则和验证结果生成HTML报告。我把这个报告的链接集成到了内部Wiki非技术同事也能看懂数据质量状况。关注性能对大数据量表某些期望如expect_column_values_to_be_unique可能很慢。可以考虑在数据库层面先通过抽样检查或者只在关键流水表上执行全量检查。3.2 LLM集成层构建高效的“专家会诊”Prompt这是系统的“大脑”。如何设计Prompt让LLM有效工作至关重要。我的Prompt模板分为几个部分系统角色设定你是一位资深的数据质量专家和数据架构师。你的任务是分析提供的数据样本和异常指标判断是否存在真实的数据质量问题推断根本原因并提供修复建议。上下文信息注入## 表结构及业务含义 - 表名t_order - 业务描述记录用户提交的订单信息。 - 字段说明 - order_id: 订单唯一标识。 - user_id: 下单用户ID关联用户表。 - order_amount: 订单金额单位元。 - order_status: 订单状态枚举值应为 [pending, paid, shipped, completed, cancelled]。 - create_time: 订单创建时间。问题描述与数据样本## 异常发现 1. 规则引擎检查发现有15%的记录的order_amount字段为0或负数。 2. 统计模型识别出order_amount字段存在极端离群值例如记录ID为10086的订单金额为9999999元。 ## 相关数据样本前5条异常记录 | order_id | user_id | order_amount | order_status | create_time | |----------|---------|--------------|--------------|----------------------| | 1001 | u123 | 0.00 | completed | 2023-10-01 10:00:00 | | 1002 | u456 | -1.50 | paid | 2023-10-01 10:05:00 | | 10086 | u789 | 9999999.00 | pending | 2023-10-01 12:00:00 | | 1003 | u123 | 0.00 | cancelled | 2023-10-01 10:10:00 |任务指令请基于以上信息完成以下分析 1. **问题判断**上述异常是否构成真实的数据质量问题请说明理由。 2. **根因推测**如果认为是问题推测可能导致这些异常的业务或技术原因如测试订单、系统BUG、业务规则允许、数据录入错误等。 3. **影响评估**这些问题可能对哪些下游业务如财务报表、佣金计算、风控模型产生影响 4. **行动建议**给出具体的数据修复建议如如何筛选这些记录、应联系哪个业务方确认、是否可以直接修复和长期的预防措施。输出格式要求请以JSON格式输出包含以下键is_issue (布尔值), reasoning (分析过程), root_cause (根因列表), impact (影响描述), action (建议列表)。通过这样结构化的PromptLLM返回的结果也非常规整便于后续程序自动化解析和处理。踩坑记录最初没有严格限定输出格式LLM的回答天马行空很难用程序提取关键信息。强制JSON输出后下游流程的稳定性大大提升。另外给LLM的样本数据一定要做脱敏处理避免泄露真实敏感信息。3.3 工作流编排层用Prefect构建诊断修复流水线我选择了Prefect因为它比Airflow更轻量API设计更现代尤其适合以代码为中心Code-as-Workflow的编排。核心流程定义示例from prefect import flow, task from prefect.logging import get_run_logger import pandas as pd # 假设我们有自己的模块 from data_diagnosis import run_gx_validation, call_llm_for_diagnosis from data_repair import execute_auto_fix, create_jira_ticket task(retries2, retry_delay_seconds30) def diagnose_data_quality(table_name: str, sample_query: str) - dict: 诊断任务运行规则检查并调用LLM分析复杂问题 logger get_run_logger() # 1. 运行Great Expectations检查 rule_violations run_gx_validation(table_name) logger.info(f规则检查完成发现 {len(rule_violations)} 条异常。) # 2. 对复杂异常调用LLM分析 complex_issues [] for violation in rule_violations: if violation[severity] HIGH: llm_result call_llm_for_diagnosis(violation, table_name) complex_issues.append(llm_result) return {rule_violations: rule_violations, llm_diagnosis: complex_issues} task def decide_and_execute_repair(diagnosis_result: dict) - str: 决策与修复任务根据诊断结果决定执行自动修复或创建人工工单 logger get_run_logger() summary [] for issue in diagnosis_result[rule_violations]: if issue[type] in [NULL_VALUE, FORMAT_ERROR] and issue[confidence] 0.9: # 高置信度的简单问题自动修复 execute_auto_fix(issue) summary.append(f自动修复: {issue[description]}) else: # 其他问题创建Jira工单等待人工处理 ticket_id create_jira_ticket(issue) summary.append(f已创建工单 [{ticket_id}] 处理: {issue[description]}) for llm_issue in diagnosis_result[llm_diagnosis]: if llm_issue.get(is_issue) and llm_issue.get(auto_fixable): # LLM判断可自动修复的问题 execute_auto_fix(llm_issue, is_llm_suggestedTrue) summary.append(f基于LLM建议自动修复: {llm_issue[description]}) else: ticket_id create_jira_ticket(llm_issue, priorityHIGH) summary.append(f已创建高优先级工单 [{ticket_id}] 处理LLM诊断问题。) return \n.join(summary) flow(namedaily_data_health_check) def data_doctor_daily_flow(): 数据医生每日健康检查主流程 logger get_run_logger() logger.info(开始每日数据健康检查...) # 定义需要检查的核心表列表 critical_tables [dim_user, fact_order, fact_payment] for table in critical_tables: logger.info(f正在检查表: {table}) # 执行诊断 diagnosis diagnose_data_quality(table, fSELECT * FROM {table} LIMIT 1000) # 执行修复决策 result_summary decide_and_execute_repair(diagnosis) logger.info(f表 {table} 处理完成。\n{result_summary}) logger.info(所有核心表健康检查流程执行完毕。) # 部署后可以配置为每天凌晨2点自动运行 if __name__ __main__: data_doctor_daily_flow()这个流程清晰地定义了“诊断-决策-执行”的链条并且每个task都是独立、可重试的单元。Prefect的UI能很好地展示流程运行状态、日志和结果方便监控。4. 落地实施与团队协作要点把这样一个系统真正用起来技术只占一半另一半是流程和协作。4.1 分阶段推广与价值验证我并没有一开始就全面铺开而是采用了“试点-扩大-推广”的三步走策略试点阶段1个月选择业务方痛点最明显、数据源相对简单的1-2个核心报表对应的数据表。与报表负责人紧密合作配置首批规则。目标是快速解决他们最头疼的几个数据不准的问题拿到“成功案例”和业务方的认可。扩大阶段2-3个月将覆盖范围扩展到该业务线的所有重要数据表并开始引入LLM对复杂问题进行辅助分析。在这个阶段数据质量日报成为该业务线晨会的固定议题。推广阶段持续将模式复制到其他业务线并建立公司级的数据质量标准和“数据医生”使用规范。此时系统已经处理了成百上千个案例知识库初具规模自动化修复比例显著提升。4.2 建立数据质量闭环管理流程技术平台需要配套的管理流程才能发挥最大效用问题提单与分配自动创建的Jira工单会根据问题类型如用户数据问题、订单数据问题自动分配给对应的数据产品经理或业务系统负责人。处理SLA为不同严重等级的问题设定处理时限如严重问题4小时内响应一般问题24小时内。复盘与沉淀每周对已关闭的工单进行复盘将有效的处理方法和根因分析沉淀到“数据医生”的知识库中并思考能否将解决方案转化为新的自动化规则。度量与考核定义数据质量指标如数据故障时长MTTR、自动修复率、问题复发率并尝试与相关团队的绩效考核轻度挂钩提升全员的数据质量意识。4.3 成本控制与性能优化这个系统运行起来主要的成本点在LLM API调用和计算资源上。LLM成本控制缓存机制对相同或相似的数据问题模式将LLM的诊断结果缓存起来下次直接使用避免重复调用。分级调用并非所有问题都需要最强的GPT-4。对于简单的分类任务可以使用成本更低的模型如Claude Haiku或GPT-3.5-Turbo。摘要与采样坚决不传送全量数据。只传送异常摘要、统计特征和小样本通常5-10条记录足够LLM分析模式。执行性能优化异步与并行对不同数据表的检查任务尽可能并行执行。增量检查对于流水型大表更多采用增量检查检查当天新增数据而非每天全表扫描。资源隔离将消耗较大的诊断任务安排在业务低峰期如深夜执行。5. 常见问题与避坑指南在实际开发和运营中我遇到了不少坑这里分享几个最有代表性的。5.1 误报与漏报的平衡这是规则引擎和LLM都会面临的核心挑战。规则太严误报多让人“狼来了”疲劳规则太松漏报多失去监控意义。应对策略采用“分级警报”机制。将问题分为“提示”、“警告”、“严重”三级。只有“严重”级别的问题会触发即时通知如钉钉/飞书告警其他级别的问题汇总到每日/每周报告中。同时建立一个“误报反馈”渠道让业务方可以快速标记误报系统会学习这些反馈动态调整规则阈值或LLM的判断逻辑。5.2 LLM的“幻觉”与不确定性LLM可能会“一本正经地胡说八道”给出看似合理但完全错误的根因分析。应对策略提供高质量上下文给LLM的元数据和业务描述必须准确、清晰。模糊的描述会导致模糊甚至错误的判断。要求提供置信度在Prompt中要求LLM对其判断给出一个置信度评分如0-1。对于低置信度的分析系统应倾向于创建人工审核工单而非自动执行。人工复核闭环将LLM的诊断建议作为“辅助意见”呈现给处理工单的同学而不是“最终裁决”。最终是否采纳、如何修复由人工决定。这个决定的结果又可以反馈给系统用于优化LLM。5.3 修复动作的安全性与回滚自动修复数据是高风险操作一旦出错可能造成数据污染。应对策略预检查与模拟任何修复脚本在执行前必须先在一个数据切片或测试环境上运行验证其影响。事务与备份修复操作必须在数据库事务中进行确保原子性。在执行前对受影响的数据行进行备份例如插入到一张data_repair_backup表中。操作审计所有自动或半自动的修复操作都必须有完整的日志记录谁哪个任务、在什么时候、对哪些数据、执行了什么操作、基于什么理由。5.4 业务理解的不断迭代业务在变化数据的内涵也在变化。今天认为是异常的情况比如某种新的促销活动导致订单金额为0明天可能就是正常的。应对策略建立“规则/知识库评审会”机制。每两周数据团队和核心业务方一起回顾近期产生的主要告警和工单。讨论哪些规则需要更新哪些业务变化需要纳入考量。让“数据医生”的知识库保持与时俱进。6. 总结与展望从“治已病”到“治未病”构建这个“AI数据医生”的过程是一个将数据治理从被动响应转向主动智能的过程。目前它已经能处理我们公司约70%的常见数据质量问题将数据团队从繁琐的“数据消防”工作中解放出来去从事更有价值的数据架构和数据产品工作。回过头看这个项目的最大价值不在于用了多炫酷的AI技术而在于它将人的经验规则、机器的效率自动化和AI的洞察LLM结合成了一个可持续运转的闭环系统。它不是一个一旦上线就完事的项目而是一个需要持续运营和优化的“数据产品”。未来的优化方向我考虑在“治未病”上做更多探索。例如利用机器学习模型预测数据质量下降的趋势比如某个数据源的NULL值率在未来一周可能显著上升从而在问题发生前提前预警和干预。或者将“数据医生”的能力更深度地集成到数据开发流程中在数据任务上线前就进行质量规则的预校验从源头减少问题。如果你也在为数据质量问题所困扰不妨从一个小痛点开始尝试引入一些自动化的诊断思路。记住工具和流程都是为人服务的最终目标是让数据可靠地赋能业务而不是增加负担。 基于AI Agent与规则引擎的智能数据治理系统设计与实践
1. 项目缘起当数据治理遇上AI一个“数据医生”的诞生在数据驱动的时代公司里最头疼的问题往往不是没有数据而是数据“病了”。我所在的公司业务线繁杂数据源五花八门从传统的业务数据库到各种第三方API、日志文件再到用户上传的Excel表格数据质量参差不齐。每天业务部门、分析师和开发团队都在和数据问题作斗争这个报表的数字怎么对不上为什么这个用户ID查不到信息这个字段为什么有一半是空值数据团队疲于奔命成了“救火队员”80%的时间都花在了数据清洗、核对和解释上而不是创造价值。传统的解决方案比如制定更严格的ETL规范、编写更多的数据校验脚本效果有限。规则是死的数据是活的新的数据问题总是以意想不到的方式出现。直到去年在深入研究了当前AI Agent和智能体工作流的技术趋势后我萌生了一个想法能不能打造一个智能化的“数据医生”它不仅能像资深数据工程师一样自动诊断数据病灶还能开出“处方”甚至直接“动手术”进行修复。这个想法就是今天要分享的“AI数据医生”项目的起点。它不是某个单一的工具而是一个融合了规则引擎、大语言模型LLM和自动化工作流的智能体系统旨在让数据质量问题从“人工排查”走向“智能自治”。2. 核心设计思路从“诊断”到“治疗”的闭环智能体这个“数据医生”的设计核心是模拟一位经验丰富的数据专家的思考和工作流程。它不是一个简单的数据校验工具而是一个具备感知、分析、决策和执行能力的智能体Agent。整个系统的架构围绕“诊、断、治、防”四个环节展开。2.1 智能诊断多维度感知数据健康状况首先数据医生需要具备全面的“体检”能力。我设计了一个多层次的诊断模块基础生命体征检查这是规则引擎层负责快速扫描数据的“硬伤”。比如检查字段是否为空NULL值比例、数值是否在合理范围内如年龄不能为负数、日期格式是否规范、枚举值是否符合预设字典等。这部分使用成熟的框架如Great Expectations或Apache Griffin来实现配置起来快执行效率高。深度影像学检查这部分引入了统计分析和机器学习模型。例如通过分析某个数值字段如交易金额的分布识别离群点Outliers通过关联性分析发现本应有强关联的两个字段如城市和邮编出现矛盾的情况。这里我用Python的Pandas、Scikit-learn库结合一些自定义算法来实现。专家会诊这是最核心的一环即引入大语言模型LLM。将前两步发现的“异常指标”和“可疑数据样本”连同数据表的元数据字段名、业务含义描述一起构造Prompt提交给LLM我选用的是GPT-4的API。LLM的任务是像人类专家一样理解业务语境判断这个“异常”是否真的是问题以及可能的原因是什么。例如规则引擎发现“用户最后登录时间”字段大量为未来日期LLM可能会结合“测试账号”、“数据迁移错误”等业务知识给出更精准的判断。注意LLM的调用成本和控制是关键。不要把所有数据都扔给LLM。我的策略是只有当中低级规则引擎和统计模型发现“疑似复杂问题”时才触发LLM会诊并且严格控制每次提交的上下文长度只发送关键摘要和样本。2.2 精准断症问题分类与根因推理诊断出异常后需要对其进行分类和定级。我建立了一个数据质量问题知识库将问题分为几个大类完整性问题缺失值、记录缺失。准确性问题错误值、格式错误、逻辑矛盾。一致性问题跨表、跨源数据不一致。时效性问题数据更新延迟。LLM在会诊后不仅会判断问题类型还会尝试进行根因推理。例如它可能会输出“问题类型准确性-逻辑矛盾。根因推测疑似‘订单状态’与‘物流状态’的更新不同步或‘已取消’订单的物流信息未及时清空。建议检查订单状态更新流水日志。” 这个推理结果会作为后续处理的宝贵输入。2.3 自动治疗可配置的修复工作流诊断和断症之后就是治疗。我设计了一个可配置的修复动作执行器。根据问题的类型、严重等级和根因推测系统会自动匹配或建议修复方案自动修复对于简单、明确的问题直接执行预设动作。例如将明显的格式错误日期如“20241301”转换为标准格式将某个字段中已知的错别字如“北京”写成“北亰”进行替换。这些动作通过编写Python Pandas函数或SQL更新语句模板来实现由系统自动填充参数后执行。半自动修复需审核对于LLM推测根因但存在一定不确定性的修复或涉及关键业务数据的修改系统会生成修复建议和待执行的脚本提交给数据负责人或创建一个Jira工单进行人工审核。审核通过后一键执行。生成修复建议报告对于复杂问题或暂无自动修复方案的问题系统会生成一份详细的诊断报告包括问题样本、影响范围、根因分析和修复建议供数据工程师参考。整个修复工作流由Apache Airflow或Prefect这样的工作流调度平台来编排确保任务的有序、可重试和可监控。2.4 预防与健康管理持续监控与知识沉淀数据医生的职责不仅是治病还要防病。系统会定期如每天对核心数据资产进行“健康巡检”生成数据质量日报跟踪各项质量指标的趋势。更重要的是所有诊断过的问题、采纳过的修复方案都会被沉淀到案例知识库中。这个知识库有两个作用一是作为未来类似问题的诊断参考提升LLM判断的准确性二是可以反向优化规则引擎将一些新发现的、可固化的模式添加到基础规则中让系统越来越“聪明”。3. 核心配置与关键技术栈拆解下面我来拆解这个“数据医生”的几个核心模块的具体配置和选型思考这些都是可以直接拿去参考的干货。3.1 规则引擎层Great Expectations 实战配置我选择了Great ExpectationsGX因为它不仅是一个校验库更是一个完整的框架支持数据文档生成和结果可视化。核心配置示例以检查某张用户表为例首先你需要定义一个“期望套件”Expectation Suite。这里我通过Python API来配置比用CLI更灵活。import great_expectations as gx import pandas as pd # 1. 初始化上下文和数据源 context gx.get_context() datasource context.sources.add_pandas(namemy_pandas_datasource) # 2. 定义数据资产这里假设df是你的Pandas DataFrame data_asset datasource.add_dataframe_asset(nameuser_table_df, dataframedf) # 3. 创建批处理请求并获取验证器 batch_request data_asset.build_batch_request() validator context.get_validator(batch_requestbatch_request) # 4. 添加具体的“期望”即校验规则 # 检查用户ID非空且唯一 validator.expect_column_values_to_be_unique(columnuser_id) validator.expect_column_values_to_not_be_null(columnuser_id) # 检查年龄在0-120岁之间 validator.expect_column_values_to_be_between(columnage, min_value0, max_value120) # 检查注册日期是合理的过去日期比如不早于公司成立日期2000年 validator.expect_column_values_to_be_between( columnregister_date, min_value2000-01-01, max_valuepd.Timestamp.now().strftime(%Y-%m-%d) ) # 检查性别字段只包含‘M’或‘F’ validator.expect_column_values_to_be_in_set(columngender, value_set[M, F]) # 检查邮箱格式简单正则 validator.expect_column_values_to_match_regex(columnemail, regexr^[^\s][^\s]\.[^\s]$) # 5. 保存期望套件 validator.save_expectation_suite(discard_failed_expectationsFalse)实操心得分阶段实施不要试图一次性为所有表添加几百条规则。先从核心的1-2张表最关键的5-10个字段开始快速跑通流程让业务方看到价值。利用数据文档GX生成的Data Docs数据文档非常有用它自动将你的期望规则和验证结果生成HTML报告。我把这个报告的链接集成到了内部Wiki非技术同事也能看懂数据质量状况。关注性能对大数据量表某些期望如expect_column_values_to_be_unique可能很慢。可以考虑在数据库层面先通过抽样检查或者只在关键流水表上执行全量检查。3.2 LLM集成层构建高效的“专家会诊”Prompt这是系统的“大脑”。如何设计Prompt让LLM有效工作至关重要。我的Prompt模板分为几个部分系统角色设定你是一位资深的数据质量专家和数据架构师。你的任务是分析提供的数据样本和异常指标判断是否存在真实的数据质量问题推断根本原因并提供修复建议。上下文信息注入## 表结构及业务含义 - 表名t_order - 业务描述记录用户提交的订单信息。 - 字段说明 - order_id: 订单唯一标识。 - user_id: 下单用户ID关联用户表。 - order_amount: 订单金额单位元。 - order_status: 订单状态枚举值应为 [pending, paid, shipped, completed, cancelled]。 - create_time: 订单创建时间。问题描述与数据样本## 异常发现 1. 规则引擎检查发现有15%的记录的order_amount字段为0或负数。 2. 统计模型识别出order_amount字段存在极端离群值例如记录ID为10086的订单金额为9999999元。 ## 相关数据样本前5条异常记录 | order_id | user_id | order_amount | order_status | create_time | |----------|---------|--------------|--------------|----------------------| | 1001 | u123 | 0.00 | completed | 2023-10-01 10:00:00 | | 1002 | u456 | -1.50 | paid | 2023-10-01 10:05:00 | | 10086 | u789 | 9999999.00 | pending | 2023-10-01 12:00:00 | | 1003 | u123 | 0.00 | cancelled | 2023-10-01 10:10:00 |任务指令请基于以上信息完成以下分析 1. **问题判断**上述异常是否构成真实的数据质量问题请说明理由。 2. **根因推测**如果认为是问题推测可能导致这些异常的业务或技术原因如测试订单、系统BUG、业务规则允许、数据录入错误等。 3. **影响评估**这些问题可能对哪些下游业务如财务报表、佣金计算、风控模型产生影响 4. **行动建议**给出具体的数据修复建议如如何筛选这些记录、应联系哪个业务方确认、是否可以直接修复和长期的预防措施。输出格式要求请以JSON格式输出包含以下键is_issue (布尔值), reasoning (分析过程), root_cause (根因列表), impact (影响描述), action (建议列表)。通过这样结构化的PromptLLM返回的结果也非常规整便于后续程序自动化解析和处理。踩坑记录最初没有严格限定输出格式LLM的回答天马行空很难用程序提取关键信息。强制JSON输出后下游流程的稳定性大大提升。另外给LLM的样本数据一定要做脱敏处理避免泄露真实敏感信息。3.3 工作流编排层用Prefect构建诊断修复流水线我选择了Prefect因为它比Airflow更轻量API设计更现代尤其适合以代码为中心Code-as-Workflow的编排。核心流程定义示例from prefect import flow, task from prefect.logging import get_run_logger import pandas as pd # 假设我们有自己的模块 from data_diagnosis import run_gx_validation, call_llm_for_diagnosis from data_repair import execute_auto_fix, create_jira_ticket task(retries2, retry_delay_seconds30) def diagnose_data_quality(table_name: str, sample_query: str) - dict: 诊断任务运行规则检查并调用LLM分析复杂问题 logger get_run_logger() # 1. 运行Great Expectations检查 rule_violations run_gx_validation(table_name) logger.info(f规则检查完成发现 {len(rule_violations)} 条异常。) # 2. 对复杂异常调用LLM分析 complex_issues [] for violation in rule_violations: if violation[severity] HIGH: llm_result call_llm_for_diagnosis(violation, table_name) complex_issues.append(llm_result) return {rule_violations: rule_violations, llm_diagnosis: complex_issues} task def decide_and_execute_repair(diagnosis_result: dict) - str: 决策与修复任务根据诊断结果决定执行自动修复或创建人工工单 logger get_run_logger() summary [] for issue in diagnosis_result[rule_violations]: if issue[type] in [NULL_VALUE, FORMAT_ERROR] and issue[confidence] 0.9: # 高置信度的简单问题自动修复 execute_auto_fix(issue) summary.append(f自动修复: {issue[description]}) else: # 其他问题创建Jira工单等待人工处理 ticket_id create_jira_ticket(issue) summary.append(f已创建工单 [{ticket_id}] 处理: {issue[description]}) for llm_issue in diagnosis_result[llm_diagnosis]: if llm_issue.get(is_issue) and llm_issue.get(auto_fixable): # LLM判断可自动修复的问题 execute_auto_fix(llm_issue, is_llm_suggestedTrue) summary.append(f基于LLM建议自动修复: {llm_issue[description]}) else: ticket_id create_jira_ticket(llm_issue, priorityHIGH) summary.append(f已创建高优先级工单 [{ticket_id}] 处理LLM诊断问题。) return \n.join(summary) flow(namedaily_data_health_check) def data_doctor_daily_flow(): 数据医生每日健康检查主流程 logger get_run_logger() logger.info(开始每日数据健康检查...) # 定义需要检查的核心表列表 critical_tables [dim_user, fact_order, fact_payment] for table in critical_tables: logger.info(f正在检查表: {table}) # 执行诊断 diagnosis diagnose_data_quality(table, fSELECT * FROM {table} LIMIT 1000) # 执行修复决策 result_summary decide_and_execute_repair(diagnosis) logger.info(f表 {table} 处理完成。\n{result_summary}) logger.info(所有核心表健康检查流程执行完毕。) # 部署后可以配置为每天凌晨2点自动运行 if __name__ __main__: data_doctor_daily_flow()这个流程清晰地定义了“诊断-决策-执行”的链条并且每个task都是独立、可重试的单元。Prefect的UI能很好地展示流程运行状态、日志和结果方便监控。4. 落地实施与团队协作要点把这样一个系统真正用起来技术只占一半另一半是流程和协作。4.1 分阶段推广与价值验证我并没有一开始就全面铺开而是采用了“试点-扩大-推广”的三步走策略试点阶段1个月选择业务方痛点最明显、数据源相对简单的1-2个核心报表对应的数据表。与报表负责人紧密合作配置首批规则。目标是快速解决他们最头疼的几个数据不准的问题拿到“成功案例”和业务方的认可。扩大阶段2-3个月将覆盖范围扩展到该业务线的所有重要数据表并开始引入LLM对复杂问题进行辅助分析。在这个阶段数据质量日报成为该业务线晨会的固定议题。推广阶段持续将模式复制到其他业务线并建立公司级的数据质量标准和“数据医生”使用规范。此时系统已经处理了成百上千个案例知识库初具规模自动化修复比例显著提升。4.2 建立数据质量闭环管理流程技术平台需要配套的管理流程才能发挥最大效用问题提单与分配自动创建的Jira工单会根据问题类型如用户数据问题、订单数据问题自动分配给对应的数据产品经理或业务系统负责人。处理SLA为不同严重等级的问题设定处理时限如严重问题4小时内响应一般问题24小时内。复盘与沉淀每周对已关闭的工单进行复盘将有效的处理方法和根因分析沉淀到“数据医生”的知识库中并思考能否将解决方案转化为新的自动化规则。度量与考核定义数据质量指标如数据故障时长MTTR、自动修复率、问题复发率并尝试与相关团队的绩效考核轻度挂钩提升全员的数据质量意识。4.3 成本控制与性能优化这个系统运行起来主要的成本点在LLM API调用和计算资源上。LLM成本控制缓存机制对相同或相似的数据问题模式将LLM的诊断结果缓存起来下次直接使用避免重复调用。分级调用并非所有问题都需要最强的GPT-4。对于简单的分类任务可以使用成本更低的模型如Claude Haiku或GPT-3.5-Turbo。摘要与采样坚决不传送全量数据。只传送异常摘要、统计特征和小样本通常5-10条记录足够LLM分析模式。执行性能优化异步与并行对不同数据表的检查任务尽可能并行执行。增量检查对于流水型大表更多采用增量检查检查当天新增数据而非每天全表扫描。资源隔离将消耗较大的诊断任务安排在业务低峰期如深夜执行。5. 常见问题与避坑指南在实际开发和运营中我遇到了不少坑这里分享几个最有代表性的。5.1 误报与漏报的平衡这是规则引擎和LLM都会面临的核心挑战。规则太严误报多让人“狼来了”疲劳规则太松漏报多失去监控意义。应对策略采用“分级警报”机制。将问题分为“提示”、“警告”、“严重”三级。只有“严重”级别的问题会触发即时通知如钉钉/飞书告警其他级别的问题汇总到每日/每周报告中。同时建立一个“误报反馈”渠道让业务方可以快速标记误报系统会学习这些反馈动态调整规则阈值或LLM的判断逻辑。5.2 LLM的“幻觉”与不确定性LLM可能会“一本正经地胡说八道”给出看似合理但完全错误的根因分析。应对策略提供高质量上下文给LLM的元数据和业务描述必须准确、清晰。模糊的描述会导致模糊甚至错误的判断。要求提供置信度在Prompt中要求LLM对其判断给出一个置信度评分如0-1。对于低置信度的分析系统应倾向于创建人工审核工单而非自动执行。人工复核闭环将LLM的诊断建议作为“辅助意见”呈现给处理工单的同学而不是“最终裁决”。最终是否采纳、如何修复由人工决定。这个决定的结果又可以反馈给系统用于优化LLM。5.3 修复动作的安全性与回滚自动修复数据是高风险操作一旦出错可能造成数据污染。应对策略预检查与模拟任何修复脚本在执行前必须先在一个数据切片或测试环境上运行验证其影响。事务与备份修复操作必须在数据库事务中进行确保原子性。在执行前对受影响的数据行进行备份例如插入到一张data_repair_backup表中。操作审计所有自动或半自动的修复操作都必须有完整的日志记录谁哪个任务、在什么时候、对哪些数据、执行了什么操作、基于什么理由。5.4 业务理解的不断迭代业务在变化数据的内涵也在变化。今天认为是异常的情况比如某种新的促销活动导致订单金额为0明天可能就是正常的。应对策略建立“规则/知识库评审会”机制。每两周数据团队和核心业务方一起回顾近期产生的主要告警和工单。讨论哪些规则需要更新哪些业务变化需要纳入考量。让“数据医生”的知识库保持与时俱进。6. 总结与展望从“治已病”到“治未病”构建这个“AI数据医生”的过程是一个将数据治理从被动响应转向主动智能的过程。目前它已经能处理我们公司约70%的常见数据质量问题将数据团队从繁琐的“数据消防”工作中解放出来去从事更有价值的数据架构和数据产品工作。回过头看这个项目的最大价值不在于用了多炫酷的AI技术而在于它将人的经验规则、机器的效率自动化和AI的洞察LLM结合成了一个可持续运转的闭环系统。它不是一个一旦上线就完事的项目而是一个需要持续运营和优化的“数据产品”。未来的优化方向我考虑在“治未病”上做更多探索。例如利用机器学习模型预测数据质量下降的趋势比如某个数据源的NULL值率在未来一周可能显著上升从而在问题发生前提前预警和干预。或者将“数据医生”的能力更深度地集成到数据开发流程中在数据任务上线前就进行质量规则的预校验从源头减少问题。如果你也在为数据质量问题所困扰不妨从一个小痛点开始尝试引入一些自动化的诊断思路。记住工具和流程都是为人服务的最终目标是让数据可靠地赋能业务而不是增加负担。