AI主动学习：标注不够用了之后

主动学习机器学习数据标注查询策略智能采样

作者: 醉月思发布日期: 2022-05-12 分类: 技术实践

前些天帮朋友处理数据集时又想起这句话：机器学习80%的时间在洗数据，剩下20%在抱怨洗得还不够干净。最近帮一个电商团队处理商品评论分类，数据集有5万条评论，要分5个类别：正面、负面、中立、物流问题、售后服务。

主动学习是什么，又不是什么

主动学习（Active Learning）本质上是个"问答游戏"：模型不知道怎么判断某些样本，就把这些样本扔给标注者，标注者告诉它答案，它再继续学习。往复几轮后，模型能力上去了，标注成本下来了。

它不是自动标注，也不是半监督学习。自动标注是让模型自己猜，猜错了就错了；半监督学习是先用少量标注数据训练，再用大量未标注数据辅助；主动学习则是主动选择"最有价值"的样本去标注。

这个区别挺重要的。我刚入行时搞混过，以为主动学习就是让模型自己标注大部分样本，然后人工抽检。结果训练出来的模型在那些它"很有信心"的样本上表现不错，但在边界区域崩得一塌糊涂——那些边界样本恰恰没有被标注过。

实践场景：文本分类任务

最近帮一个电商团队处理商品评论分类，数据集有5万条评论，要分5个类别：正面、负面、中立、物流问题、售后服务。全标注大概需要200人时，预算只有40人时。

我们选了主动学习来处理这个问题。整体流程大概是这样的：

graph TD A[初始标注数据<br/>10%随机采样] --> B[训练基础模型] B --> C[查询策略选择<br/>不确定性样本] C --> D[人工标注<br/>选定样本] D --> E[模型再训练] E --> F{达到预算或<br/>性能目标?} F -->|否| C F -->|是| G[模型部署]

这个流程看起来简单，但每个环节都有坑。

第一步：初始数据要怎么选

初始标注数据选什么，直接影响模型后面的查询能力。我之前有个误区：既然要主动学习，那就从几个"简单"类别开始，等模型有基础能力再去挑战难的。

结果发现模型在那些"简单"类别上过拟合了，查询策略选出来的样本永远是它没见过的那些类别——因为它根本不知道那些类别的存在。

正确做法是初始数据覆盖所有类别，哪怕每个类别只有几条。我们这次从5万条评论里随机选了500条（约1%），确保每个类别至少有10条标注数据。

import numpy as np
from collections import defaultdict

def balanced_initial_sampling(data, labels, samples_per_class=10):
    """确保每个类别至少有指定数量的初始标注样本"""
    class_indices = defaultdict(list)
    for idx, label in enumerate(labels):
        class_indices[label].append(idx)
    
    selected_indices = []
    for label, indices in class_indices.items():
        if len(indices) >= samples_per_class:
            selected_indices.extend(np.random.choice(indices, samples_per_class, replace=False))
        else:
            selected_indices.extend(indices)
    
    return selected_indices

# 使用示例
initial_indices = balanced_initial_sampling(all_texts, all_labels, samples_per_class=10)

第二步：查询策略怎么选

查询策略是主动学习的核心，它决定哪些样本"值得"被标注。常见的有不确定性采样、多样性采样、混合策略等。

不确定性采样

不确定性采样让模型选它"最不确定"的样本。在分类任务里，通常用概率熵或者边际概率差。

def uncertainty_sampling(model, unlabeled_data, n_samples=100):
    """基于概率熵的不确定性采样"""
    probs = model.predict_proba(unlabeled_data)
    entropies = -np.sum(probs * np.log(probs + 1e-10), axis=1)
    # 选熵最大的（不确定性最高的）
    top_indices = np.argsort(entropies)[-n_samples:][::-1]
    return top_indices

def margin_sampling(model, unlabeled_data, n_samples=100):
    """基于边际概率差的不确定性采样"""
    probs = model.predict_proba(unlabeled_data)
    # 取最高概率和次高概率的差值
    sorted_probs = np.sort(probs, axis=1)
    margins = sorted_probs[:, -1] - sorted_probs[:, -2]
    # 选边际差最小的（模型最犹豫的）
    top_indices = np.argsort(margins)[:n_samples]
    return top_indices

这个策略挺好用，但有个坑：模型刚开始训练时，预测概率分布很平，几乎所有样本的熵都很高。这时候它选出来的样本可能真的是"随机"的。

我们在这个坑上栽过。第一次迭代用不确定性采样选了100条样本，结果发现里面80%都是同一个类别——因为模型在这个类别上最不自信。

后来加了两个改进：一是不确定性采样前先让模型预热几个epoch；二是限制每个类别在单次采样中的占比。

def balanced_uncertainty_sampling(model, unlabeled_data, labels, n_samples=100, max_per_class=30):
    """平衡的不确定性采样，避免单类别占比过高"""
    probs = model.predict_proba(unlabeled_data)
    entropies = -np.sum(probs * np.log(probs + 1e-10), axis=1)
    
    # 按类别分组
    class_indices = defaultdict(list)
    for idx, label in enumerate(labels):
        class_indices[label].append(idx)
    
    selected_indices = []
    for label, indices in class_indices.items():
        if len(selected_indices) >= n_samples:
            break
        # 在该类别内部按熵排序
        class_entropies = entropies[indices]
        sorted_indices = np.argsort(class_entropies)[::-1]
        # 最多选 max_per_class 条
        to_select = min(max_per_class, n_samples - len(selected_indices), len(indices))
        selected_indices.extend([indices[i] for i in sorted_indices[:to_select]])
    
    return selected_indices

多样性采样

不确定性采样有时候会选出一堆"很像"的样本——它们在模型眼里都很不确定，但实际上是同一类困难样本。多样性采样就是解决这个问题的，它选出来的样本要在特征空间里分布得比较广。

我们用的是核心集采样（Core-set），简单说就是在特征空间里选一个有代表性的子集。

from sklearn.metrics.pairwise import euclidean_distances

def coresets_sampling(features, n_samples=100):
    """核心集采样，确保样本在特征空间中有代表性"""
    # 随机选第一个样本
    selected = [np.random.choice(len(features))]
    
    # 不断选离已选样本最远的样本
    for _ in range(n_samples - 1):
        distances = euclidean_distances(features, features[selected])
        min_distances = np.min(distances, axis=1)
        next_idx = np.argmax(min_distances)
        selected.append(next_idx)
    
    return selected

这个方法的问题是计算复杂度，特征空间太大时会很慢。我们后来用了MiniBatchKMeans的聚类中心代替，效果差不多但快很多。

混合策略

最终我们用的是混合策略：70%不确定性采样，30%多样性采样。这样既保证选到模型不自信的样本，又保证样本分布够广。

第三步：标注迭代怎么安排

标注迭代不是"每标注一批就训练一次"，也不是"攒够一批才训练一次"，这两种极端都有问题。

每标注一批就训练一次，看起来很"主动"，但实际上训练成本很高，而且单批标注量太小时，模型更新可能不稳定。

攒够一批才训练一次，又会错过一些"及时反馈"的机会——模型可能已经在某些样本上过拟合了，但我们还在继续标注类似的样本。

我们用的是"渐进式"策略：开始时批次小、迭代密，随着数据量增加，批次变大、迭代变稀。

class AdaptiveIteration:
    def __init__(self, initial_batch=50, growth_factor=1.5, max_batch=500):
        self.current_batch = initial_batch
        self.growth_factor = growth_factor
        self.max_batch = max_batch
    
    def get_next_batch_size(self):
        batch = int(self.current_batch)
        self.current_batch = min(self.current_batch * self.growth_factor, self.max_batch)
        return batch

# 使用示例
scheduler = AdaptiveIteration(initial_batch=50, growth_factor=1.5)
for i in range(10):
    batch_size = scheduler.get_next_batch_size()
    print(f"Iteration {i}: batch_size = {batch_size}")

实践中的坑和解决方案

坑1：标注者疲劳

主动学习听起来很好，但实际操作中标注者很容易疲劳——每次都在标注那些"最难"的样本，时间长了人会烦。

我们试过几个办法：

每批次里掺入10%的"简单样本"（模型很确定的），给标注者一点成就感
设置每批次最多30分钟，时间到了就停，不管有没有标完
让标注者看到模型预测结果，如果预测对了，可以直接"确认"，不用重新打标

第三个办法效果最好，但有个风险：标注者可能被模型预测"带偏"了。我们后来加了个"盲测"环节，定期让标注者不看模型预测直接标注，用来校验。

坑2：特征空间漂移

这个坑挺隐蔽的。我们的评论数据里，早期的评论多是关于商品质量的，后期的评论开始大量涉及物流和售后。模型在早期数据上训练得好好的，突然就崩了——因为特征空间变了。

主动学习在这种情况下其实挺尴尬的：模型会一直选那些它在当前特征空间里不确定的样本，但这些样本可能不是最有价值的。

我们后来加了"分布检测"：每轮训练前，检查未标注数据的特征分布是否发生了显著变化。如果变化很大，就主动采样一些新分布的样本，而不是只依赖不确定性采样。

from scipy.stats import ks_2samp

def detect_distribution_shift(old_features, new_features, threshold=0.05):
    """检测特征分布是否发生显著变化"""
    shift_detected = False
    for i in range(old_features.shape[1]):
        _, p_value = ks_2samp(old_features[:, i], new_features[:, i])
        if p_value < threshold:
            shift_detected = True
            break
    return shift_detected

# 使用示例
if detect_distribution_shift(training_features, unlabeled_features):
    # 检测到分布漂移，随机采样一部分新数据
    random_indices = np.random.choice(len(unlabeled_features), size=50, replace=False)

坑3：查询策略和模型不匹配

这个坑让我折腾了好几天。我们一开始用的是简单的TF-IDF+逻辑回归模型，查询策略用不确定性采样，效果不错。

后来觉得模型太简单了，换成BERT，查询策略还是不确定性采样。结果效果反而变差了——BERT的预测概率太"自信"了，很多样本它都给出接近0或1的概率，不确定性采样选出来的样本很少。

后来才知道，BERT这类深度模型的概率校准本身就有问题。我们试过温度缩放、Platt Scaling，效果都不太理想。

最后换了个思路：不用预测概率，而是用预测的"分歧度"。对于同一个样本，用多次dropout预测，看预测结果有多大差异。

def mc_dropout_uncertainty(model, data, n_samples=10):
    """基于MC Dropout的不确定性估计"""
    model.train()  # 启用dropout
    predictions = []
    
    for _ in range(n_samples):
        with torch.no_grad():
            logits = model(data)
            probs = torch.softmax(logits, dim=1)
            predictions.append(probs.cpu().numpy())
    
    predictions = np.array(predictions)  # shape: (n_samples, n_data, n_classes)
    mean_probs = np.mean(predictions, axis=1)
    # 用预测方差作为不确定性
    uncertainties = np.var(predictions, axis=1).mean(axis=1)
    
    return uncertainties

这个方法效果不错，但计算成本高。后来我们妥协了一下：每3次迭代用一次MC Dropout，中间几次还是用普通不确定性采样。

结果和反思

最终我们在预算内完成了标注，用了全标注数据的20%就达到了90%的性能。比预期的要好，但也暴露了一些问题。

主动学习确实能减少标注成本，但它不是"免费午餐"。它需要：

足够的计算资源（反复训练模型）
合理的查询策略设计
稳定的标注流程
对问题的深入理解（知道哪些样本真的有价值）

我们在实践中发现，主动学习在"边界清晰"的任务上效果最好，比如分类、检测。在"边界模糊"的任务上，比如情感分析、意图识别，效果会打折扣——因为这些任务本身就没那么明确的"难"和"易"。

还有一个反思：我们一开始太关注"减少标注量"这个指标了，后来才发现，真正重要的是"单位标注投入带来的性能提升"。有时候多标注几条"简单样本"，比只标注困难样本更有效。

主动学习不是万能的，但在数据标注成本高、样本分布复杂的场景下，它确实是个值得尝试的工具。关键是理解它的局限，然后在实践中不断调整策略，而不是把它当成一个"打开就能用"的解决方案。

最后留一个问题：如果主动学习选出来的样本，标注者也无法给出可靠标签，这时候该怎么办？是换模型、换策略，还是承认这个问题本身就很难？这个问题我也没完全想明白，留着以后继续琢磨吧。

版权声明: 本文首发于指尖魔法屋-AI主动学习：标注不够用了之后（https://blog.thinkmoon.cn/post/227-active-learning-practice-from-labeling-to-intelligence/）转载或引用必须申明原指尖魔法屋来源及源地址！

从清洗走到标注：AI数据处理笔记

最近一个月折腾了一堆 NLP 数据，从 JSON 格式的对话记录到用户评论文本，再到一些半结构化的标注数据。

📁 技术实践

📅 2024-04-08

把手工换到半自动时踩过的坑

去年做了一个 NLP 分类项目，模型准确率死活卡在 70% 左右。