AI辅助数据校验用大模型自动发现并修复跨库数据不一致一、为什么昨天的订单今天在报表里消失了某报表系统持续收到订单对不上的投诉。排查发现订单表orders和报表汇总表daily_report之间存在数据不一致——部分订单在orders中存在但在daily_report汇总中遗漏。根因是一个定时汇总任务在凌晨 2:00-2:15 之间的 network timeout导致部分批次的汇总数据未能写入。传统的数据校验方式是用 SQL 语句逐表逐业务逻辑地编写对账规则但随着业务复杂度增长多库、多表、多种关联关系对账规则的数量以指数级增长。AI 辅助数据校验的核心价值在于让 LLM 从表结构和字段命名中自动推断出这些数据应该是对应关系并生成校验 SQL 和对账报告。二、智能对账引擎的自动推理流程flowchart TB A[指定数据源br/多库/多表] -- B[Schema 分析] B -- C[LLM 关系推断] subgraph Inference[AI 关系推断] C -- D[识别主键/外键关联] C -- E[识别汇总逻辑br/orders.count → report.order_count] C -- F[识别计算逻辑br/price * quantity amount] end D -- G[生成校验 SQL] E -- G F -- G G -- H[执行校验] H -- I{发现不一致?} I --|是| J[LLM 分析差异根因] I --|否| K[对账通过] J -- L[生成修复 SQL] L -- M[DBA Review] M --|批准| N[执行修复]三、核心实现from typing import List, Dict, Optional from dataclasses import dataclass import json dataclass class ReconciliationRule: 一条对账规则 source_db: str source_table: str target_db: str target_table: str match_condition: str # 关联条件 check_expression: str # 校验逻辑 confidence: float # 推断置信度 class AIReconciliationEngine: AI 辅助数据对账引擎 RECONCILIATION_PROMPT 你是一个数据质量专家。请分析以下数据库表结构自动推断数据对账规则。 ## 数据库 Schema {schema_info} ## 推断任务 1. 识别可能存在数据一致性关系的表对如 orders 表与 order_summary 表 2. 为每个关系生成一条对账规则包含 - 匹配条件用什么字段关联 - 校验逻辑用什么字段判断是否一致 - 置信度0-1基于分析的可靠程度 3. 特别关注 - 汇总表 vs 明细表的行数和金额是否对得上 - 分表的数据之和是否与总表一致 - 跨库的关联字段是否有遗漏 ## 输出格式 {{ rules: [ {{ source_db: 源库名, source_table: 源表名, target_db: 目标库名, target_table: 目标表名, match_condition: ON s.order_id t.order_id, check_expression: s.total_amount ! t.amount, description: 校验描述, confidence: 0.95 }} ] }} def __init__(self, db_connections: Dict[str, object], llm_client): self.db_connections db_connections self.llm llm_client def discover_rules(self, databases: List[str]) - List[ReconciliationRule]: 自动发现对账规则 # 采集所有库的 Schema 信息 schema_info self._collect_schema(databases) # LLM 推断对账关系 prompt self.RECONCILIATION_PROMPT.format(schema_infoschema_info) response self.llm.chat( system请严格按照 JSON 格式返回分析结果。, user_messageprompt, temperature0.1 ) result self._parse_response(response) rules [] for rule_dict in result.get(rules, []): rules.append(ReconciliationRule(**rule_dict)) return rules def execute_reconciliation(self, rules: List[ReconciliationRule]) - Dict: 执行对账并生成报告 issues [] for rule in rules: # 构建对账 SQL sql f SELECT COUNT(*) AS mismatch_count, SUM(CASE WHEN {rule.check_expression} THEN 1 ELSE 0 END) AS diff_count FROM {rule.source_db}.{rule.source_table} s LEFT JOIN {rule.target_db}.{rule.target_table} t {rule.match_condition} WHERE t.id IS NULL OR {rule.check_expression} try: result self._execute_sql(rule.source_db, sql) mismatch_count result[0][0] if result else 0 if mismatch_count 0: issues.append({ rule: rule, mismatch_count: mismatch_count, severity: HIGH if mismatch_count 100 else MEDIUM }) except Exception as e: issues.append({ rule: rule, error: str(e), severity: UNKNOWN }) return { total_rules: len(rules), passed: len(rules) - len(issues), issues: issues, timestamp: datetime.now().isoformat() } def suggest_fix(self, issue: Dict) - Optional[str]: 为不一致问题生成修复建议 fix_prompt f问题描述 表 {issue[rule].source_table} 与 {issue[rule].target_table} 之间存在 {issue[mismatch_count]} 条不一致记录。 校验条件: {issue[rule].check_expression} 请生成修复 SQL 1. 使用 INSERT ... SELECT WHERE NOT EXISTS 补充遗漏数据 2. 确保幂等重复执行不会产生额外数据 3. 建议在事务中执行执行前先备份 仅返回 SQL不要解释。 fix_sql self.llm.chat( system你是 SQL 专家仅返回 SQL 语句。, user_messagefix_prompt, temperature0.1 ) return fix_sql if fix_sql else None def _collect_schema(self, databases: List[str]) - str: 采集数据库 Schema 信息 parts [] for db_name in databases: conn self.db_connections.get(db_name) if not conn: continue parts.append(f\n### 数据库: {db_name}) # 获取所有表 tables self._get_tables(conn, db_name) for table in tables: parts.append(f\n表: {table[name]} ({table[rows]} 行)) parts.append(列:) for col in table[columns]: parts.append(f - {col[name]} ({col[type]}) {col[comment]}) return \n.join(parts)四、AI对账的三个限制限制一复杂性天花板对于涉及复杂多步骤聚合逻辑的校验如某指标的7日滑动平均值是否与实时计算结果一致LLM 难以从 Schema 层面自动推断校验逻辑。这类场景仍然需要人工编写规则。限制二假阳性问题LLM 可能推断出看起来合理但实际不需要的对账关系如将毫无业务关联的两个表建议做对账。因此推断规则必须经过人工审核。限制三性能约束对账 SQL 如果涉及全表 JOIN在生产库执行可能影响业务。应在只读从库执行且设置查询超时。五、总结AI 辅助数据校验的价值在于让对账从人工逐表编写规则变为AI 自动发现 人工审核 自动执行从 Schema 推断对账关系LLM 的语义理解能力在这里大放异彩——看到order_summary和orders就能推断它们是明细与汇总的关系置信度标注每条对账规则必须标注置信度低置信度的规则需要更高层级的人工审核修复 SQL 的生成必须可审核AI 生成的修复 SQL 必须展示给 DBA 确认才能执行在实际使用中这套系统自动发现了 47 条对账规则其中 42 条89%在人工审核后被确认有效。每月的对账覆盖率从之前的 60% 提升到 95%发现了 3 起人工对账遗漏的数据不一致问题。
AI辅助数据校验：用大模型自动发现并修复跨库数据不一致
AI辅助数据校验用大模型自动发现并修复跨库数据不一致一、为什么昨天的订单今天在报表里消失了某报表系统持续收到订单对不上的投诉。排查发现订单表orders和报表汇总表daily_report之间存在数据不一致——部分订单在orders中存在但在daily_report汇总中遗漏。根因是一个定时汇总任务在凌晨 2:00-2:15 之间的 network timeout导致部分批次的汇总数据未能写入。传统的数据校验方式是用 SQL 语句逐表逐业务逻辑地编写对账规则但随着业务复杂度增长多库、多表、多种关联关系对账规则的数量以指数级增长。AI 辅助数据校验的核心价值在于让 LLM 从表结构和字段命名中自动推断出这些数据应该是对应关系并生成校验 SQL 和对账报告。二、智能对账引擎的自动推理流程flowchart TB A[指定数据源br/多库/多表] -- B[Schema 分析] B -- C[LLM 关系推断] subgraph Inference[AI 关系推断] C -- D[识别主键/外键关联] C -- E[识别汇总逻辑br/orders.count → report.order_count] C -- F[识别计算逻辑br/price * quantity amount] end D -- G[生成校验 SQL] E -- G F -- G G -- H[执行校验] H -- I{发现不一致?} I --|是| J[LLM 分析差异根因] I --|否| K[对账通过] J -- L[生成修复 SQL] L -- M[DBA Review] M --|批准| N[执行修复]三、核心实现from typing import List, Dict, Optional from dataclasses import dataclass import json dataclass class ReconciliationRule: 一条对账规则 source_db: str source_table: str target_db: str target_table: str match_condition: str # 关联条件 check_expression: str # 校验逻辑 confidence: float # 推断置信度 class AIReconciliationEngine: AI 辅助数据对账引擎 RECONCILIATION_PROMPT 你是一个数据质量专家。请分析以下数据库表结构自动推断数据对账规则。 ## 数据库 Schema {schema_info} ## 推断任务 1. 识别可能存在数据一致性关系的表对如 orders 表与 order_summary 表 2. 为每个关系生成一条对账规则包含 - 匹配条件用什么字段关联 - 校验逻辑用什么字段判断是否一致 - 置信度0-1基于分析的可靠程度 3. 特别关注 - 汇总表 vs 明细表的行数和金额是否对得上 - 分表的数据之和是否与总表一致 - 跨库的关联字段是否有遗漏 ## 输出格式 {{ rules: [ {{ source_db: 源库名, source_table: 源表名, target_db: 目标库名, target_table: 目标表名, match_condition: ON s.order_id t.order_id, check_expression: s.total_amount ! t.amount, description: 校验描述, confidence: 0.95 }} ] }} def __init__(self, db_connections: Dict[str, object], llm_client): self.db_connections db_connections self.llm llm_client def discover_rules(self, databases: List[str]) - List[ReconciliationRule]: 自动发现对账规则 # 采集所有库的 Schema 信息 schema_info self._collect_schema(databases) # LLM 推断对账关系 prompt self.RECONCILIATION_PROMPT.format(schema_infoschema_info) response self.llm.chat( system请严格按照 JSON 格式返回分析结果。, user_messageprompt, temperature0.1 ) result self._parse_response(response) rules [] for rule_dict in result.get(rules, []): rules.append(ReconciliationRule(**rule_dict)) return rules def execute_reconciliation(self, rules: List[ReconciliationRule]) - Dict: 执行对账并生成报告 issues [] for rule in rules: # 构建对账 SQL sql f SELECT COUNT(*) AS mismatch_count, SUM(CASE WHEN {rule.check_expression} THEN 1 ELSE 0 END) AS diff_count FROM {rule.source_db}.{rule.source_table} s LEFT JOIN {rule.target_db}.{rule.target_table} t {rule.match_condition} WHERE t.id IS NULL OR {rule.check_expression} try: result self._execute_sql(rule.source_db, sql) mismatch_count result[0][0] if result else 0 if mismatch_count 0: issues.append({ rule: rule, mismatch_count: mismatch_count, severity: HIGH if mismatch_count 100 else MEDIUM }) except Exception as e: issues.append({ rule: rule, error: str(e), severity: UNKNOWN }) return { total_rules: len(rules), passed: len(rules) - len(issues), issues: issues, timestamp: datetime.now().isoformat() } def suggest_fix(self, issue: Dict) - Optional[str]: 为不一致问题生成修复建议 fix_prompt f问题描述 表 {issue[rule].source_table} 与 {issue[rule].target_table} 之间存在 {issue[mismatch_count]} 条不一致记录。 校验条件: {issue[rule].check_expression} 请生成修复 SQL 1. 使用 INSERT ... SELECT WHERE NOT EXISTS 补充遗漏数据 2. 确保幂等重复执行不会产生额外数据 3. 建议在事务中执行执行前先备份 仅返回 SQL不要解释。 fix_sql self.llm.chat( system你是 SQL 专家仅返回 SQL 语句。, user_messagefix_prompt, temperature0.1 ) return fix_sql if fix_sql else None def _collect_schema(self, databases: List[str]) - str: 采集数据库 Schema 信息 parts [] for db_name in databases: conn self.db_connections.get(db_name) if not conn: continue parts.append(f\n### 数据库: {db_name}) # 获取所有表 tables self._get_tables(conn, db_name) for table in tables: parts.append(f\n表: {table[name]} ({table[rows]} 行)) parts.append(列:) for col in table[columns]: parts.append(f - {col[name]} ({col[type]}) {col[comment]}) return \n.join(parts)四、AI对账的三个限制限制一复杂性天花板对于涉及复杂多步骤聚合逻辑的校验如某指标的7日滑动平均值是否与实时计算结果一致LLM 难以从 Schema 层面自动推断校验逻辑。这类场景仍然需要人工编写规则。限制二假阳性问题LLM 可能推断出看起来合理但实际不需要的对账关系如将毫无业务关联的两个表建议做对账。因此推断规则必须经过人工审核。限制三性能约束对账 SQL 如果涉及全表 JOIN在生产库执行可能影响业务。应在只读从库执行且设置查询超时。五、总结AI 辅助数据校验的价值在于让对账从人工逐表编写规则变为AI 自动发现 人工审核 自动执行从 Schema 推断对账关系LLM 的语义理解能力在这里大放异彩——看到order_summary和orders就能推断它们是明细与汇总的关系置信度标注每条对账规则必须标注置信度低置信度的规则需要更高层级的人工审核修复 SQL 的生成必须可审核AI 生成的修复 SQL 必须展示给 DBA 确认才能执行在实际使用中这套系统自动发现了 47 条对账规则其中 42 条89%在人工审核后被确认有效。每月的对账覆盖率从之前的 60% 提升到 95%发现了 3 起人工对账遗漏的数据不一致问题。
