AI数据质量：这次怎么落地的

数据质量数据清洗数据验证 AI工程 Python

作者: 醉月思发布日期: 2023-03-15 分类: 技术实践

结果上线第二天，运营部就在群里反馈：有一大批"iPhone 14 Pro Max"被分到了"充电器配件"类目里。

数据质量问题通常长什么样

先说说我们实际遇到过的几种典型问题。

标签混乱

这是最常见也最难发现的问题。漏填反而不是大头，更麻烦的是同一实体在不同样本里标签对不上：

同一个实体在不同样本里标签不同：比如"小米手环 7"有的样本标为"智能手表"，有的标为"运动配件"
标签粒度不一致：有的标到三级类目，有的只标到一级类目
标签定义本身就有歧义：比如"智能音箱"和"蓝牙音箱"的边界就不清楚

语义漂移

这个在做对话数据时特别明显。比如我们做客服问答对，同一句话在不同时期标注的标准答案完全不一样：

用户：怎么退款？
早期答案：请联系客服电话 400-xxx-xxxx
后期答案：您可以在订单详情页点击申请退款

这是产品改版了，标注标准没跟上，导致数据里同时存在两套完全不同的"正确答案"。

数据分布偏移

这个问题在持续收集数据的项目里很常见。比如我们做电商评论情感分类：

早期的训练数据主要来自手机数码类目，以产品功能评价为主
后期持续收集的数据大量来自美妆类目，更多是使用感受和效果描述

如果直接把新旧数据混在一起训练，模型就会学到这种类目分布偏移，导致在某个类目上表现异常。

隐私和敏感信息

这个不用多说，但实际项目里很容易漏掉。我们曾经在做日志清洗时，发现样本里藏着不少用户手机号、身份证号片段、订单号这类信息。

有些是用户自己在评论文本里泄露的，有些是原始数据导入时忘记脱敏了。

数据清洗的实用流程

我们现在的数据清洗流程大概是这个样子：

flowchart TD A[原始数据采集] --> B[格式标准化] B --> C[显性噪声过滤] C --> D[隐私信息脱敏] D --> E[标签一致性检查] E --> F[语义质量评估] F --> G[数据分布分析] G --> H[质量报告与决策]

这张图想表达的重点很简单：清洗得分层做，指望一步到位不现实。

格式标准化

这一步是最基础但最好处理的。主要是：

import pandas as pd
import re

def standardize_text(text):
    # 统一换行符
    text = text.replace('\r\n', '\n').replace('\r', '\n')
    # 去除多余空格
    text = re.sub(r'\s+', ' ', text).strip()
    # 统一引号风格
    text = text.replace('"', '"').replace('"', '"').replace(''', "'").replace(''', "'")
    return text

def standardize_dataset(df):
    """标准化数据格式"""
    df = df.copy()
    if 'text' in df.columns:
        df['text'] = df['text'].apply(standardize_text)
    if 'label' in df.columns:
        df['label'] = df['label'].str.strip().str.lower()
    return df

这里有一个小经验：不要在格式标准化阶段做太激进的清理。比如有人喜欢把所有特殊字符、emoji 全删掉，这样可能会丢掉很多语义信息。

我们之前就犯过这个错，把用户评论里的表情符号全部删了，导致模型对情感极性的判断准确率掉了 5 个点。

显性噪声过滤

这一步主要处理那些明显异常的样本：

def filter_noise_samples(df, text_col='text', min_length=10):
    """过滤明显噪声样本"""
    df = df.copy()

    # 过滤过短文本
    df = df[df[text_col].str.len() >= min_length]

    # 过滤重复文本（保留第一个）
    df = df.drop_duplicates(subset=[text_col], keep='first')

    # 过滤疑似乱码文本（连续特殊字符过多）
    def is_garbage(text):
        # 统计非字母数字汉字字符的比例
        special_chars = re.findall(r'[^\w\s一-鿿]', text)
        ratio = len(special_chars) / len(text) if len(text) > 0 else 0
        return ratio > 0.5

    df = df[~df[text_col].apply(is_garbage)]

    # 过滤主要由数字组成的文本
    def is_mostly_numbers(text):
        numbers = re.findall(r'\d', text)
        return len(numbers) / len(text) > 0.7 if len(text) > 0 else False

    df = df[~df[text_col].apply(is_mostly_numbers)]

    return df.reset_index(drop=True)

这里面有两个容易踩的坑：

重复文本不要全部删。有些重复是有意义的，比如同一个用户在不同时间点的重复反馈，或者高频使用场景的标准话术。我们通常只做完全去重，或者按时间戳只保留最新的。
乱码检测的阈值要调整。不同场景下"乱码"的定义不一样。代码日志、技术文档里特殊字符比例本来就高，这时候如果用 0.5 的阈值就会误删。

隐私信息脱敏

这一步我们用正则加规则匹配的方案：

import hashlib

def mask_pii(text):
    """脱敏隐私信息"""
    text = str(text)

    # 手机号脱敏
    text = re.sub(r'1[3-9]\d{9}', lambda m: m.group(0)[:3] + '****' + m.group(0)[-4:], text)

    # 身份证号脱敏
    text = re.sub(r'\d{17}[\dXx]', lambda m: m.group(0)[:6] + '********' + m.group(0)[-4:], text)

    # 邮箱脱敏
    text = re.sub(r'([a-zA-Z0-9._%+-]+)@([a-zA-Z0-9.-]+\.[a-zA-Z]{2,})',
                  lambda m: m.group(1)[:2] + '***@' + m.group(2), text)

    # 订单号脱敏（假设订单号是18位数字）
    text = re.sub(r'\b\d{18}\b', lambda m: 'ORDER_' + hashlib.md5(m.group(0).encode()).hexdigest()[:8], text)

    return text

这里有个建议：脱敏时尽量保留一定的可追溯性。比如订单号用哈希值替换而不是全删掉，这样后续如果需要排查问题，还能通过哈希值反向定位到原始订单（虽然只能单向，但比完全丢了强）。

质量验证的关键指标

清洗之后，我们不会马上就拿去训练，而是先跑一套质量验证。

标签一致性检查

这是数据质量里最容易出问题也最影响模型效果的部分：

def check_label_consistency(df, text_col='text', label_col='label', similarity_threshold=0.85):
    """检查标签一致性：相似文本是否有不同标签"""
    from difflib import SequenceMatcher

    # 找出文本相似度超过阈值但标签不同的样本对
    inconsistent_pairs = []

    for i in range(len(df)):
        for j in range(i+1, len(df)):
            similarity = SequenceMatcher(None, df.iloc[i][text_col], df.iloc[j][text_col]).ratio()

            if similarity >= similarity_threshold and df.iloc[i][label_col] != df.iloc[j][label_col]:
                inconsistent_pairs.append({
                    'index_1': df.index[i],
                    'index_2': df.index[j],
                    'text_1': df.iloc[i][text_col][:100] + '...',
                    'text_2': df.iloc[j][text_col][:100] + '...',
                    'label_1': df.iloc[i][label_col],
                    'label_2': df.iloc[j][label_col],
                    'similarity': similarity
                })

    return inconsistent_pairs

# 使用示例
df = pd.read_csv('labeled_data.csv')
inconsistent_pairs = check_label_consistency(df)

if inconsistent_pairs:
    print(f"发现 {len(inconsistent_pairs)} 组标签不一致的样本")
    for pair in inconsistent_pairs[:10]:  # 只看前10组
        print(f"\n相似度: {pair['similarity']:.2f}")
        print(f"文本1: {pair['text_1']}")
        print(f"文本2: {pair['text_2']}")
        print(f"标签: {pair['label_1']} vs {pair['label_2']}")

这个方法有个效率问题：两两比较的时间复杂度是 O(n²)。数据量大了会很慢。

实际项目里我们用了一个优化方案：先用 TF-IDF 或者嵌入模型做相似度检索，只对候选对做精确验证：

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np

def fast_label_consistency_check(df, text_col='text', label_col='label',
                                  similarity_threshold=0.85, top_k=20):
    """快速标签一致性检查"""
    # 计算 TF-IDF 矩阵
    vectorizer = TfidfVectorizer(max_features=5000)
    tfidf_matrix = vectorizer.fit_transform(df[text_col])

    # 计算相似度矩阵
    similarity_matrix = cosine_similarity(tfidf_matrix)

    # 只检查每行最相似的 top_k 个
    inconsistent_pairs = []

    for i in range(len(df)):
        # 获取最相似的样本索引（排除自己）
        similar_indices = np.argsort(similarity_matrix[i])[-(top_k+1):-1][::-1]

        for j in similar_indices:
            similarity = similarity_matrix[i][j]

            if similarity >= similarity_threshold and df.iloc[i][label_col] != df.iloc[j][label_col]:
                inconsistent_pairs.append({
                    'index_1': df.index[i],
                    'index_2': df.index[j],
                    'text_1': df.iloc[i][text_col][:100] + '...',
                    'text_2': df.iloc[j][text_col][:100] + '...',
                    'label_1': df.iloc[i][label_col],
                    'label_2': df.iloc[j][label_col],
                    'similarity': similarity
                })

    return inconsistent_pairs

这样把复杂度从 O(n²) 降到了 O(n × k)，10万条的数据也能在几分钟内跑完。

数据分布分析

这一步主要是看数据分布是不是有明显的偏移：

import matplotlib.pyplot as plt
import seaborn as sns

def analyze_data_distribution(df, label_col='label', date_col=None):
    """分析数据分布"""

    fig, axes = plt.subplots(1, 3, figsize=(18, 5))

    # 1. 标签分布
    label_counts = df[label_col].value_counts()
    sns.barplot(x=label_counts.values, y=label_counts.index, ax=axes[0])
    axes[0].set_title('标签分布')
    axes[0].set_xlabel('样本数量')

    # 2. 标签占比饼图
    axes[1].pie(label_counts.values, labels=label_counts.index, autopct='%1.1f%%')
    axes[1].set_title('标签占比')

    # 3. 如果有时间列，看时间分布
    if date_col and date_col in df.columns:
        df['date'] = pd.to_datetime(df[date_col])
        df['date_truncated'] = df['date'].dt.to_period('M')
        date_counts = df['date_truncated'].value_counts().sort_index()

        sns.lineplot(x=date_counts.index.astype(str), y=date_counts.values, ax=axes[2])
        axes[2].set_title('时间分布')
        axes[2].set_xlabel('月份')
        axes[2].set_ylabel('样本数量')
        axes[2].tick_params(axis='x', rotation=45)
    else:
        axes[2].text(0.5, 0.5, '无时间数据', ha='center', va='center')
        axes[2].set_title('时间分布')

    plt.tight_layout()
    plt.savefig('data_distribution.png', dpi=150)
    plt.close()

    # 输出统计信息
    print("数据分布统计:")
    print(f"总样本数: {len(df)}")
    print(f"标签类别数: {len(label_counts)}")
    print(f"\n每个标签的样本数:")
    print(label_counts)

    # 检查类别不平衡
    imbalance_ratio = label_counts.max() / label_counts.min()
    if imbalance_ratio > 10:
        print(f"\n警告: 存在严重类别不平衡 (最大/最小 = {imbalance_ratio:.1f})")

    return label_counts

这里有个经验之谈：不要看到类别不平衡就急着做过采样或欠采样。

如果业务场景本身就是不均衡的（比如欺诈检测里正常样本远多于欺诈样本），那你需要的是在模型训练时用合适的 loss 或者评估指标，而不是强行把数据变均衡。

我们在做的一个风控模型就是，原始数据里正常交易占 99.5%，欺诈交易只占 0.5%。如果强行采样到 1:1，模型上线后对正常交易的误报率会高得离谱。

语义质量评估

这一步主要针对文本数据，用 NLP 方法评估语义质量：

def evaluate_semantic_quality(texts, language_model=None):
    """评估文本语义质量"""
    import numpy as np

    quality_scores = {}

    # 1. 文本长度分布
    lengths = [len(text) for text in texts]
    quality_scores['avg_length'] = np.mean(lengths)
    quality_scores['length_std'] = np.std(lengths)
    quality_scores['min_length'] = min(lengths)
    quality_scores['max_length'] = max(lengths)

    # 2. 词汇丰富度（不同词数 / 总词数）
    def vocabulary_richness(text):
        words = text.split()
        if len(words) == 0:
            return 0
        return len(set(words)) / len(words)

    richness_scores = [vocabulary_richness(text) for text in texts]
    quality_scores['avg_vocabulary_richness'] = np.mean(richness_scores)

    # 3. 语言模型困惑度（如果有语言模型）
    if language_model:
        perplexities = []
        for text in texts:
            # 这里简化处理，实际需要根据具体语言模型接口调整
            try:
                perplexity = language_model.calculate_perplexity(text)
                perplexities.append(perplexity)
            except:
                pass

        if perplexities:
            quality_scores['avg_perplexity'] = np.mean(perplexities)
            quality_scores['perplexity_std'] = np.std(perplexities)

    return quality_scores

# 使用示例
texts = df['text'].tolist()
quality_scores = evaluate_semantic_quality(texts)

print("语义质量评估:")
for metric, value in quality_scores.items():
    if isinstance(value, float):
        print(f"{metric}: {value:.2f}")
    else:
        print(f"{metric}: {value}")

这一步不是必须的，但在大模型场景下有用。比如我们做对话数据清洗时，发现平均困惑度异常高的一批数据，后来查出来是测试账号灌水生成的垃圾对话。

建立持续的质量监控机制

数据质量不是一次性的，需要持续监控。我们现在的做法是在数据管线里加了几个检查点：

class DataQualityMonitor:
    def __init__(self, baseline_stats=None):
        self.baseline_stats = baseline_stats or {}

    def check_drift(self, current_stats, threshold=0.1):
        """检查数据漂移"""
        drift_report = []

        for key, current_value in current_stats.items():
            if key in self.baseline_stats:
                baseline_value = self.baseline_stats[key]

                # 数值型指标检查相对变化
                if isinstance(current_value, (int, float)):
                    if baseline_value != 0:
                        relative_change = abs(current_value - baseline_value) / baseline_value
                        if relative_change > threshold:
                            drift_report.append({
                                'metric': key,
                                'baseline': baseline_value,
                                'current': current_value,
                                'change': f"{relative_change*100:.1f}%"
                            })

        return drift_report

    def update_baseline(self, new_stats):
        """更新基线统计"""
        self.baseline_stats.update(new_stats)

# 使用示例
# 第一次：建立基线
monitor = DataQualityMonitor()
baseline_stats = {
    'avg_length': 150.0,
    'avg_vocabulary_richness': 0.6,
    'label_distribution': {'positive': 5000, 'negative': 3000, 'neutral': 2000}
}
monitor.update_baseline(baseline_stats)

# 后续批次：检查漂移
current_stats = {
    'avg_length': 120.0,  # 下降了20%
    'avg_vocabulary_richness': 0.55,  # 下降了8.3%
    'label_distribution': {'positive': 3000, 'negative': 5000, 'neutral': 2000}
}

drift_report = monitor.check_drift(current_stats)
if drift_report:
    print("检测到数据漂移:")
    for item in drift_report:
        print(f"{item['metric']}: {item['baseline']} -> {item['current']} ({item['change']})")

这个监控机制我们现在是每周跑一次，如果发现某个指标变化超过阈值，就会触发人工复查。

踩过的坑和解决方案

坑一：过度清洗丢了语义

我们早期做数据清洗时，有一个很激进的规则：把所有包含 URL 的样本全部删掉。理由是 URL 不稳定，而且很多是广告链接。

结果后来发现，用户评价里经常会有"参考这个教程 https://xxx"或者"官方文档 https://xxx"，这些 URL 携带了重要的上下文信息。全删了之后，模型对这类评论的理解能力明显下降。

现在的做法是：不是删除 URL，而是保留 URL 前后的上下文，把 URL 替换成一个占位符 [URL]：

def sanitize_urls(text):
    """清理 URL 但保留上下文"""
    # 用占位符替换 URL
    text = re.sub(r'https?://[^\s]+', '[URL]', text)
    return text

坑二：依赖自动清洗工具的误判

我们之前试过几个开源的数据清洗工具（比如 great_expectations、pandas-profiling），发现它们在特定领域数据上误判率很高。

比如有一个工具自动把所有纯数字文本标为"异常数据"，但我们的业务场景里确实有大量的纯数字编码（SKU、型号、序列号），这些是正常数据。

教训是：自动化工具可以做参考，但不能完全依赖。尤其是领域特定的数据，还是要结合业务规则人工校验。

坑三：标签标准化导致的错误传播

我们在做标签一致性检查时，曾经自动把一些"相似但不同"的标签合并了。比如把"智能手表"和"运动手表"合并成"智能手表"。

后来发现这两个类目在业务逻辑上是分开的：智能手表需要连接手机，运动手表是独立的。合并之后，模型学到的就是混淆的概念。

现在的做法是：做标签标准化之前，先找业务方确认清楚标签的语义边界。不确定的先保留，不要急着合并。

实际项目中的效果

说一下我们实际项目里的效果。

我们做了数据质量改进之后，效果主要体现在这几个方面：

模型训练时间缩短：因为去掉了噪声和重复数据，训练收敛更快了。同样超参下，训练迭代次数减少了约 30%。
模型稳定性提升：上线后的离谱预测明显少了。之前那个把 iPhone 分到配件类目的问题，在数据清洗后再训练就再没出现过。
数据团队效率提高：建立了质量监控机制之后，数据问题能更早发现，而不是等模型上线崩了才发现。

小结

数据质量这事，难在问题不会主动跳出来，只会用一次次离谱预测提醒你；简单在认真做过一轮完整清洗和验证后，后面能固化成检查清单。

别把清洗当一次性任务，也别盲信自动化工具。业务规则、持续监控、人工抽查，三样都得有。工具只能帮你走得快一点，坑还得自己盯。

参考

Great Expectations: https://greatexpectations.io/
Pandas Profiling: https://pandas-profiling.ydata.ai/
数据质量工程实践: https://www.oreilly.com/library/view/data-quality-engineering/9781098118772/

版权声明: 本文首发于指尖魔法屋-AI数据质量：这次怎么落地的（https://blog.thinkmoon.cn/post/194-data-quality-cleaning-validation-practice/）转载或引用必须申明原指尖魔法屋来源及源地址！