基于AI Agent与规则引擎的智能数据治理系统设计与实践

1. 项目缘起：当数据治理遇上AI，一个“数据医生”的诞生

在数据驱动的时代，公司里最头疼的问题往往不是没有数据，而是数据“病了”。我所在的公司，业务线繁杂，数据源五花八门，从传统的业务数据库到各种第三方API、日志文件，再到用户上传的Excel表格，数据质量参差不齐。每天，业务部门、分析师和开发团队都在和数据问题作斗争：这个报表的数字怎么对不上？为什么这个用户ID查不到信息？这个字段为什么有一半是空值？数据团队疲于奔命，成了“救火队员”，80%的时间都花在了数据清洗、核对和解释上，而不是创造价值。

传统的解决方案，比如制定更严格的ETL规范、编写更多的数据校验脚本，效果有限。规则是死的，数据是活的，新的数据问题总是以意想不到的方式出现。直到去年，在深入研究了当前AI Agent和智能体工作流的技术趋势后，我萌生了一个想法：能不能打造一个智能化的“数据医生”？它不仅能像资深数据工程师一样，自动诊断数据病灶，还能开出“处方”，甚至直接“动手术”进行修复。这个想法，就是今天要分享的“AI数据医生”项目的起点。它不是某个单一的工具，而是一个融合了规则引擎、大语言模型（LLM）和自动化工作流的智能体系统，旨在让数据质量问题从“人工排查”走向“智能自治”。

2. 核心设计思路：从“诊断”到“治疗”的闭环智能体

这个“数据医生”的设计核心，是模拟一位经验丰富的数据专家的思考和工作流程。它不是一个简单的数据校验工具，而是一个具备感知、分析、决策和执行能力的智能体（Agent）。整个系统的架构围绕“诊、断、治、防”四个环节展开。

2.1 智能诊断：多维度感知数据健康状况

首先，数据医生需要具备全面的“体检”能力。我设计了一个多层次的诊断模块：

1. 基础生命体征检查：这是规则引擎层，负责快速扫描数据的“硬伤”。比如，检查字段是否为空（NULL值比例）、数值是否在合理范围内（如年龄不能为负数）、日期格式是否规范、枚举值是否符合预设字典等。这部分使用成熟的框架如Great Expectations或Apache Griffin来实现，配置起来快，执行效率高。
2. 深度影像学检查：这部分引入了统计分析和机器学习模型。例如，通过分析某个数值字段（如交易金额）的分布，识别离群点（Outliers）；通过关联性分析，发现本应有强关联的两个字段（如城市和邮编）出现矛盾的情况。这里我用Python的Pandas、Scikit-learn库结合一些自定义算法来实现。
3. 专家会诊：这是最核心的一环，即引入大语言模型（LLM）。将前两步发现的“异常指标”和“可疑数据样本”连同数据表的元数据（字段名、业务含义描述）一起，构造Prompt提交给LLM（我选用的是GPT-4的API）。LLM的任务是像人类专家一样，理解业务语境，判断这个“异常”是否真的是问题，以及可能的原因是什么。例如，规则引擎发现“用户最后登录时间”字段大量为未来日期，LLM可能会结合“测试账号”、“数据迁移错误”等业务知识给出更精准的判断。

注意：LLM的调用成本和控制是关键。不要把所有数据都扔给LLM。我的策略是，只有当中低级规则引擎和统计模型发现“疑似复杂问题”时，才触发LLM会诊，并且严格控制每次提交的上下文长度，只发送关键摘要和样本。

2.2 精准断症：问题分类与根因推理

诊断出异常后，需要对其进行分类和定级。我建立了一个数据质量问题知识库，将问题分为几个大类：

• 完整性问题：缺失值、记录缺失。
• 准确性问题：错误值、格式错误、逻辑矛盾。
• 一致性问题：跨表、跨源数据不一致。
• 时效性问题：数据更新延迟。

LLM在会诊后，不仅会判断问题类型，还会尝试进行根因推理。例如，它可能会输出：“问题类型：准确性-逻辑矛盾。根因推测：疑似‘订单状态’与‘物流状态’的更新不同步，或‘已取消’订单的物流信息未及时清空。建议检查订单状态更新流水日志。” 这个推理结果会作为后续处理的宝贵输入。

2.3 自动治疗：可配置的修复工作流

诊断和断症之后，就是治疗。我设计了一个可配置的修复动作执行器。根据问题的类型、严重等级和根因推测，系统会自动匹配或建议修复方案：

1. 自动修复：对于简单、明确的问题，直接执行预设动作。例如，将明显的格式错误日期（如“20241301”）转换为标准格式；将某个字段中已知的错别字（如“北京”写成“北亰”）进行替换。这些动作通过编写Python Pandas函数或SQL更新语句模板来实现，由系统自动填充参数后执行。
2. 半自动修复（需审核）：对于LLM推测根因但存在一定不确定性的修复，或涉及关键业务数据的修改，系统会生成修复建议和待执行的脚本，提交给数据负责人（或创建一个Jira工单）进行人工审核。审核通过后，一键执行。
3. 生成修复建议报告：对于复杂问题或暂无自动修复方案的问题，系统会生成一份详细的诊断报告，包括问题样本、影响范围、根因分析和修复建议，供数据工程师参考。

整个修复工作流由Apache Airflow或Prefect这样的工作流调度平台来编排，确保任务的有序、可重试和可监控。

2.4 预防与健康管理：持续监控与知识沉淀

数据医生的职责不仅是治病，还要防病。系统会定期（如每天）对核心数据资产进行“健康巡检”，生成数据质量日报，跟踪各项质量指标的趋势。更重要的是，所有诊断过的问题、采纳过的修复方案，都会被沉淀到案例知识库中。这个知识库有两个作用：一是作为未来类似问题的诊断参考，提升LLM判断的准确性；二是可以反向优化规则引擎，将一些新发现的、可固化的模式添加到基础规则中，让系统越来越“聪明”。

3. 核心配置与关键技术栈拆解

下面我来拆解这个“数据医生”的几个核心模块的具体配置和选型思考，这些都是可以直接拿去参考的干货。

3.1 规则引擎层：Great Expectations 实战配置

我选择了Great Expectations（GX），因为它不仅是一个校验库，更是一个完整的框架，支持数据文档生成和结果可视化。

核心配置示例（以检查某张用户表为例）：

首先，你需要定义一个“期望套件”（Expectation Suite）。这里我通过Python API来配置，比用CLI更灵活。

import great_expectations as gx import pandas as pd # 1. 初始化上下文和数据源 context = gx.get_context() datasource = context.sources.add_pandas(name="my_pandas_datasource") # 2. 定义数据资产（这里假设df是你的Pandas DataFrame） data_asset = datasource.add_dataframe_asset(name="user_table_df", dataframe=df) # 3. 创建批处理请求并获取验证器 batch_request = data_asset.build_batch_request() validator = context.get_validator(batch_request=batch_request) # 4. 添加具体的“期望”（即校验规则） # 检查用户ID非空且唯一 validator.expect_column_values_to_be_unique(column="user_id") validator.expect_column_values_to_not_be_null(column="user_id") # 检查年龄在0-120岁之间 validator.expect_column_values_to_be_between(column="age", min_value=0, max_value=120) # 检查注册日期是合理的过去日期（比如不早于公司成立日期2000年） validator.expect_column_values_to_be_between( column="register_date", min_value="2000-01-01", max_value=pd.Timestamp.now().strftime("%Y-%m-%d") ) # 检查性别字段只包含‘M’或‘F’ validator.expect_column_values_to_be_in_set(column="gender", value_set=["M", "F"]) # 检查邮箱格式（简单正则） validator.expect_column_values_to_match_regex(column="email", regex=r"^[^@\s]+@[^@\s]+\.[^@\s]+$") # 5. 保存期望套件 validator.save_expectation_suite(discard_failed_expectations=False)

实操心得：

• 分阶段实施：不要试图一次性为所有表添加几百条规则。先从核心的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` (建议列表)。

通过这样结构化的Prompt，LLM返回的结果也非常规整，便于后续程序自动化解析和处理。

踩坑记录：最初没有严格限定输出格式，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(retries=2, retry_delay_seconds=30) 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_suggested=True) summary.append(f"基于LLM建议自动修复: {llm_issue['description']}") else: ticket_id = create_jira_ticket(llm_issue, priority='HIGH') summary.append(f"已创建高优先级工单 [{ticket_id}] 处理LLM诊断问题。") return "\n".join(summary) @flow(name="daily_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, f"SELECT * 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个月）：选择业务方痛点最明显、数据源相对简单的1-2个核心报表对应的数据表。与报表负责人紧密合作，配置首批规则。目标是快速解决他们最头疼的几个数据不准的问题，拿到“成功案例”和业务方的认可。
2. 扩大阶段（2-3个月）：将覆盖范围扩展到该业务线的所有重要数据表，并开始引入LLM对复杂问题进行辅助分析。在这个阶段，数据质量日报成为该业务线晨会的固定议题。
3. 推广阶段（持续）：将模式复制到其他业务线，并建立公司级的数据质量标准和“数据医生”使用规范。此时，系统已经处理了成百上千个案例，知识库初具规模，自动化修复比例显著提升。

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可能会“一本正经地胡说八道”，给出看似合理但完全错误的根因分析。

• 应对策略：
  1. 提供高质量上下文：给LLM的元数据和业务描述必须准确、清晰。模糊的描述会导致模糊甚至错误的判断。
  2. 要求提供置信度：在Prompt中要求LLM对其判断给出一个置信度评分（如0-1）。对于低置信度的分析，系统应倾向于创建人工审核工单，而非自动执行。
  3. 人工复核闭环：将LLM的诊断建议作为“辅助意见”呈现给处理工单的同学，而不是“最终裁决”。最终是否采纳、如何修复，由人工决定。这个决定的结果又可以反馈给系统，用于优化LLM。

5.3 修复动作的安全性与回滚

自动修复数据是高风险操作，一旦出错可能造成数据污染。

• 应对策略：
  1. 预检查与模拟：任何修复脚本在执行前，必须先在一个数据切片或测试环境上运行，验证其影响。
  2. 事务与备份：修复操作必须在数据库事务中进行，确保原子性。在执行前，对受影响的数据行进行备份（例如，插入到一张data_repair_backup表中）。
  3. 操作审计：所有自动或半自动的修复操作，都必须有完整的日志记录：谁（哪个任务）、在什么时候、对哪些数据、执行了什么操作、基于什么理由。

5.4 业务理解的不断迭代

业务在变化，数据的内涵也在变化。今天认为是异常的情况（比如某种新的促销活动导致订单金额为0），明天可能就是正常的。

• 应对策略：建立“规则/知识库评审会”机制。每两周，数据团队和核心业务方一起，回顾近期产生的主要告警和工单。讨论哪些规则需要更新，哪些业务变化需要纳入考量。让“数据医生”的知识库保持与时俱进。

6. 总结与展望：从“治已病”到“治未病”

构建这个“AI数据医生”的过程，是一个将数据治理从被动响应转向主动智能的过程。目前，它已经能处理我们公司约70%的常见数据质量问题，将数据团队从繁琐的“数据消防”工作中解放出来，去从事更有价值的数据架构和数据产品工作。

回过头看，这个项目的最大价值不在于用了多炫酷的AI技术，而在于它将人的经验（规则）、机器的效率（自动化）和AI的洞察（LLM）结合成了一个可持续运转的闭环系统。它不是一个一旦上线就完事的项目，而是一个需要持续运营和优化的“数据产品”。

未来的优化方向，我考虑在“治未病”上做更多探索。例如，利用机器学习模型预测数据质量下降的趋势（比如，某个数据源的NULL值率在未来一周可能显著上升），从而在问题发生前提前预警和干预。或者，将“数据医生”的能力更深度地集成到数据开发流程中，在数据任务上线前就进行质量规则的预校验，从源头减少问题。

如果你也在为数据质量问题所困扰，不妨从一个小痛点开始，尝试引入一些自动化的诊断思路。记住，工具和流程都是为人服务的，最终目标是让数据可靠地赋能业务，而不是增加负担。