第5章数据质量测试

Tommy2026/9/19大约 3 分钟

第5章数据质量测试

AI模型的质量上限由数据决定。垃圾数据进去，垃圾结果出来(Garbage In, Garbage
Out)。数据质量测试是AI测试体系中最基础也最重要的环节。

5.1 数据质量维度

维度	定义	检查方法	严重程度
完整性	数据是否有缺失值	空值/NaN检测	高
一致性	数据格式和取值是否统一	正则匹配/范围检查	高
准确性	数据是否反映真实情况	抽样人工验证	高
时效性	数据是否过期	时间戳检查	中
唯一性	是否有重复记录	去重统计	中
合规性	是否符合业务规则	规则引擎验证	高
均衡性	类别/特征分布是否合理	分布统计分析	中
标注质量	人工标注是否准确一致	多人标注一致性	高

5.2 数据完整性测试

import pandas as pd
import numpy as np

def test_data_completeness(df):

'''数据完整性检查'''

  report = {}

  # 1. 缺失值检查
  missing = df.isnull().sum()
  missing_pct = (missing / len(df) * 100).round(2)
  report['missing'] = missing_pct[missing_pct > 0]

  # 2. 空字符串检查
  for col in df.select_dtypes(include='object'):
    empty = (df[col] == '').sum()
    if empty > 0:
       report[f'{col}_empty'] = empty

  # 3. 必填字段检查
  required = ['text', 'label', 'source']
  for col in required:
    assert col in df.columns, f'缺少必填列: {col}'
    assert df[col].notna().all(), f'{col}存在空值'

  return report

5.3 数据分布测试

数据分布直接影响模型的学习效果。类别不均衡、特征偏态分布等问题会导致模型性能下降。

  def test_data_distribution(df, label_col='label'):

'''数据分布检查'''

     # 类别分布
     dist = df[label_col].value_counts(normalize=True)
     print('类别分布:')
     print(dist)

     # 类别不均衡检查(最大类/最小类 > 10倍则告警)
     max_ratio = dist.max() / dist.min()
     if max_ratio > 10:
       print(f'警告: 类别不均衡比率={max_ratio:.1f}:1')

     # 数值特征偏态检查
     for col in df.select_dtypes(include=np.number):
       skew = df[col].skew()
       if abs(skew) > 2:
             print(f'警告: {col}偏态系数={skew:.2f}')

     # 文本长度分布
     if 'text' in df.columns:
       lengths = df['text'].str.len()
       print(f'文本长度: mean={lengths.mean():.0f}, '
              f'std={lengths.std():.0f}, '
              f'min={lengths.min()}, max={lengths.max()}')

5.4 标注质量测试

标注质量是监督学习数据质量的核心。常见问题包括标注不一致、标注错误、标注标准模糊等。

5.4.1 标注一致性指标

指标	适用场景	计算方法	达标阈值
Cohen's Kappa	两人标注一致性	考虑偶然一致的修正值	>0.6
Fleiss' Kappa	多人标注一致性	多标注者间的修正一致	>0.6
Krippendorff's Alpha	通用一致性	支持缺失值和多尺度	>0.67
标注者间重合率	简单一致性	一致标注数/总标注数	>80%

5.4.2 标注质量抽检方法

from sklearn.metrics import cohen_kappa_score

def test_annotation_quality(annotator1, annotator2):

'''标注一致性检查'''

  # Cohen's Kappa
  kappa = cohen_kappa_score(annotator1, annotator2)
  print(f'Cohen Kappa: {kappa:.3f}')
  assert kappa > 0.6, f'标注一致性不达标: {kappa:.3f}'

  # 不一致样本分析
  disagreements = [
      (i, a, b)
      for i, (a, b) in enumerate(zip(annotator1, annotator2))
      if a != b
  ]
  print(f'不一致样本数: {len(disagreements)}')
  return disagreements

5.5 数据漂移检测

数据漂移(Data Drift)指生产环境中的数据分布与训练数据分布发生偏移，这是模型性能退化的主要原因之一。

5.5.1 漂移类型

特征漂移(Covariate Shift)：输入特征的分布发生变化
标签漂移(Prior Probability Shift)：目标变量的分布变化
概念漂移(Concept Drift)：输入与输出的关系发生变化

5.5.2 漂移检测方法

from scipy import stats

def detect_drift(reference_data, current_data,
           threshold=0.05):

'''KS检验检测数据漂移'''

  results = {}
  for col in reference_data.columns:
     if reference_data[col].dtype in [np.float64, np.int64]:
       stat, p_value = stats.ks_2samp(
           reference_data[col],
           current_data[col]
       )
       drift = p_value < threshold
       results[col] = {
           'statistic': stat,
           'p_value': p_value,
           'drift_detected': drift
       }
       if drift:
           print(f'漂移告警: {col}, '
                 f'KS={stat:.3f}, p={p_value:.4f}')
  return results

5.6 Great Expectations实操

Great Expectations是业界最流行的数据质量框架，支持声明式数据验证和自动化报告。

import great_expectations as gx

# 创建数据质量检查套件
context = gx.get_context()
ds = context.sources.pandas_default
asset = ds.add_dataframe_asset('training_data')
batch = asset.build_batch_request(dataframe=df)

# 定义期望
validator = context.get_validator(batch_request=batch)
validator.expect_column_values_to_not_be_null('text')
validator.expect_column_values_to_not_be_null('label')
validator.expect_column_values_to_be_in_set(
    'label', ['positive', 'negative', 'neutral']
)
validator.expect_column_value_lengths_to_be_between(
    'text', min_value=1, max_value=5000
)

# 执行验证
results = validator.validate()
print(f'通过: {results.success}')

第5章 数据质量测试

第5章数据质量测试