cleanlab 低内存批量查找标签错误：find_label_issues_batched 与 LabelInspector 实战指南

cleanlab 低内存批量查找标签错误：find_label_issues_batched 与 LabelInspector 实战指南

【免费下载链接】cleanlab Cleanlab's open-source library is the standard data-centric AI package for data quality and machine learning with messy, real-world data and labels. 项目地址: https://gitcode.com/GitHub_Trending/cl/cleanlab

大数据集上运行 cleanlab 的标准标签错误检测流程时，内存往往成为瓶颈。cleanlab.experimental.label_issues_batched 模块提供了基于小批量（mini-batch）的实现，让你可以在内存受限的条件下，以接近标准 cleanlab.filter.find_label_issues() 的结果质量，完成标签质量评分、标签错误数量估计与问题样本定位。读完本文，你将掌握 find_label_issues_batched() 便捷函数与 LabelInspector 类的完整用法、全部参数语义，以及其内部两趟扫描的实现原理。

模块定位：解决什么问题

cleanlab.filter.find_label_issues() 是 cleanlab 检测数据标签错误的标准入口，但它需要一次性把整个数据集的 pred_probs（形状为 (N, K) 的概率矩阵）与 labels 载入内存。当数据集规模达到数百万乃至上亿样本时，这会直接导致内存溢出。

cleanlab/experimental/label_issues_batched.py 模块从设计上就是为了解决这一问题：它通过按小批量（batch）读取数据、增量式估计统计量的方式，显著降低内存占用。根据模块文档说明，其默认设置下返回的结果与以下调用高度近似：

cleanlab.filter.find_label_issues(…, filter_by="low_self_confidence", return_indices_ranked_by="self_confidence")

运行该方案有两种方式（见 模块入口文档 与源码 docstring）：

需要说明的是，该模块位于 experimental 目录 下，属于"前沿但可能有棱角"的实验性代码，官方警告其在不同 cleanlab 版本之间不保证 API 稳定。

快速上手：find_label_issues_batched()

find_label_issues_batched() 是 LabelInspector 类内部实现流程的封装版本，函数定义位于 cleanlab/experimental/label_issues_batched.py#L43-L55。它的关键设计是：labels 与 pred_probs 可以以内存映射（memory-mapped）对象的形式传入，如 Zarr 数组或 memmap 数组，从而避免一次性把大数据载入内存。

方式一：从 .npy 文件直接运行

如果你已经将数组保存为 .npy 文件，可以直接传入文件路径，函数内部会通过 np.load(…, mmap_mode="r") 以内存映射方式读取（源码 L148-L166）：

import numpy as np
from cleanlab.experimental.label_issues_batched import find_label_issues_batched

batch_size = 10000 # 为了效率，设置为你的内存能承受的最大值

# 演示如何把已有的 numpy 数组保存为兼容的 .npy 文件：
np.save("LABELS.npy", labels_array)
np.save("PREDPROBS.npy", pred_probs_array)

# 你可以通过 labels = np.load("LABELS.npy", mmap_mode="r") 将其加载回 memmap 数组，
# 也可以直接对 .npy 文件运行本方法：
issues = find_label_issues_batched(
labels_file="LABELS.npy",
pred_probs_file="PREDPROBS.npy",
batch_size=batch_size,
)

注意：labels_file 只支持由 np.save() 或其它兼容方式生成的 .npy 文件（不支持 .npz），且要求存储的是完整的一维 labels 数组（不支持 list 格式）。

方式二：直接传入 Zarr 数组

本方法同样支持 Zarr 数组，你只需用 zarr.convenience.open() 以只读模式打开即可：

import zarr
from cleanlab.experimental.label_issues_batched import find_label_issues_batched

# 演示如何把已有的 numpy 数组保存为兼容的 .zarr 文件：
zarr.convenience.save_array("LABELS.zarr", labels_array)
zarr.convenience.save_array("PREDPROBS.zarr", pred_probs_array)

# 从 .zarr 文件加载：
labels = zarr.convenience.open("LABELS.zarr", mode="r")
pred_probs = zarr.convenience.open("PREDPROBS.zarr", mode="r")

# 本方法可以直接运行在 Zarr 数组、memmap 数组或普通 numpy 数组上：
issues = find_label_issues_batched(labels=labels, pred_probs=pred_probs, batch_size=batch_size)

labels 与 pred_probs 也可以直接是普通 numpy 数组——批处理方案对它们同样适用，只是省内存的优势只有在传入内存映射对象时才真正体现。关于 mmap 与 Zarr/HDF5 方案的取舍，模块文档建议参考 pythonspeed 的经典文章《mmap vs. Zarr/HDF5》（该文是官方文档引用的参考资料，可自行检索）。

函数完整参数表

find_label_issues_batched() 的全部参数及语义如下（源码 L43-L55，详细说明见函数 docstring L70-L147）：

返回值：return_mask=True 返回布尔数组，True 表示该样本存在标签问题，False 表示高置信度下标签正确；return_mask=False 返回问题样本的索引数组，按对应标签正确的可能性升序排列（即最可疑的排最前）。该语义与 filter.find_label_issues(…, return_indices_ranked_by=…) 一致。

使用约束与前置校验

• 必须二选一提供 labels 或 labels_file、pred_probs 或 pred_probs_file，否则抛 ValueError（源码 L171-L174）；
• 若 len(labels) != len(pred_probs)，会提示"Perhaps an issue loading mmap numpy arrays from file"（源码 L177-L180）；
• 本模块仅支持多类别（multi-class）分类，不支持多标签（multi-label）分类。

进阶用法：LabelInspector 类与两趟扫描流程

如果你需要更细粒度的控制（例如只估计阈值、只评分、中途调整 batch_size），可以使用 LabelInspector 类。其核心工作流是两趟遍历数据（源码类 docstring L240-L282）：

• 第一趟：用 update_confident_thresholds() 逐批计算每个类别的 confident thresholds；
• 第二趟：用 score_label_quality() 逐批评估每个样本的标签质量，同时更新标签问题数量的估计。

一个 LabelInspector 对象应只服务于一个数据集，不要跨数据集复用；类的属性请勿自行修改。示例脚本如下：

from cleanlab.experimental.label_issues_batched import LabelInspector

n = len(labels)
batch_size = 10000 # 你可以在批次之间调整它，设置为 RAM 允许的最大值

lab = LabelInspector(num_class=pred_probs.shape[1])

# 第一趟：计算 confident thresholds
# （为加快速度，也可以在数据的随机子集上执行这一步）
i = 0
while i < n:
end_index = i + batch_size
labels_batch = labels[i:end_index]
pred_probs_batch = pred_probs[i:end_index, :]
i = end_index
lab.update_confident_thresholds(labels_batch, pred_probs_batch)

# 查看计算出的阈值：
confident_thresholds = lab.get_confident_thresholds()

# 第二趟：评估标签质量（对你要评估的整个数据集运行）
i = 0
while i < n:
end_index = i + batch_size
labels_batch = labels[i:end_index]
pred_probs_batch = pred_probs[i:end_index, :]
i = end_index
batch_results = lab.score_label_quality(labels_batch, pred_probs_batch)

# 有标签问题的样本索引，按标签质量分数排序（最严重的在前）：
indices_of_examples_with_issues = lab.get_label_issues()

如果你的 pred_probs 和 labels 已经常驻内存，那么上面所有步骤可以简化为一行：

indices_of_examples_with_issues = find_label_issues_batched(labels, pred_probs, batch_size=10000)

LabelInspector 构造参数

核心方法速查

• update_confident_thresholds(labels, pred_probs)：用新一批数据更新对象内存储的 confident thresholds 估计（L487-L516）；
• get_confident_thresholds(silent=False)：取回已计算的 confident thresholds（形状 (K,)），与 count.get_confident_thresholds 格式一致；未计算时抛 ValueError（L378-L398）；
• score_label_quality(labels, pred_probs, *, update_num_issues=True)：给一批数据打分，返回每样本 0~1 的分数（越低越可能被误标），并更新标签问题数量；若只想要分数不想要问题数量，设 update_num_issues=False 可降低运行时间（L518-L562）；
• get_num_issues(silent=False)：取回已评估数据中估计的标签问题数量（L400-L430）。注意该估计可能与 count.num_label_issues() 相差 1（舍入差异）；
• get_quality_scores()：取回所有已评估样本的标签质量分数（要求 store_results=True，否则抛 ValueError）（L432-L450）；
• get_label_issues()：取回问题样本索引，按标签质量分数排序（L452-L485）。

内部原理：两趟扫描在源码中如何实现

find_label_issues_batched() 的主体逻辑（源码 L181-L227）清晰地体现了"两趟扫描"结构，与 LabelInspector 的示例脚本一一对应：

增量式 confident thresholds 估计

update_confident_thresholds() 的实现体现了"增量加权平均"思想（L500-L516）：

batch_thresholds = get_confident_thresholds(labels, pred_probs)
batch_class_counts = value_counts_fill_missing_classes(labels, num_classes=self.num_class)
self.confident_thresholds = (
self.examples_per_class * self.confident_thresholds
+ batch_class_counts * batch_thresholds
) / np.clip(self.examples_per_class + batch_class_counts, a_min=1, a_max=None)
self.confident_thresholds = np.clip(
self.confident_thresholds, a_min=CONFIDENT_THRESHOLDS_LOWER_BOUND, a_max=None
)

即用"已见样本数 × 旧阈值 + 本批样本数 × 批内阈值"做按类别计数的加权平均。这与 count.get_confident_thresholds()（cleanlab/count.py#L1407-L1456）中"类别 j 的 confident threshold 等于所有被标注为 j 的样本的模型预测概率均值"的定义一致——只是从"全量一次算"变成了"分批增量合并"。阈值最终被裁剪到 CONFIDENT_THRESHOLDS_LOWER_BOUND（定义于 cleanlab/internal/constants.py）。

标签质量评分与问题数量估计

• 评分：score_label_quality() 内部调用 rank._compute_label_quality_scores()（cleanlab/rank.py#L120-L155），支持 self_confidence、normalized_margin、confidence_weighted_entropy 三种评分方法，并可把已算好的 confident_thresholds 直接传入以加速。默认方法为 self_confidence，即取样本给定标签位置上的预测概率 P[k]；
• 问题数量：_update_num_label_issues()（L564-L673）通过 pred_probs 的 argmax 预测类别与给定标签是否一致、预测概率是否超过对应类别的 adjusted confident thresholds（阈值减去 FLOATING_POINT_COMPARISON 以规避浮点比较误差）来统计"prune count"；当 num_issue_kwargs={"estimation_method": "off_diagonal_calibrated"} 时，会按类别累计 prune_counts、class_counts、normalization，最终在 get_num_issues() 中做校准后再取整（L422-L428），避免除零时使用 CLIPPING_LOWER_BOUND 裁剪。

多进程加速（仅 Linux）

当 n_jobs > 1 时，问题数量的统计会启用多进程（L630-L673）：把本批索引用 split_arr() 切成若干 chunk，通过 pool.imap_unordered() 并行执行 _compute_num_issues()（无校准版）或 _compute_num_issues_calibrated()（校准版）。实现细节包括：

• 阈值与数据通过模块级全局变量（adj_confident_thresholds_shared、labels_shared、pred_probs_shared）共享给子进程；
• 为兼容 Python 3.14+，显式优先尝试 mp.get_context("fork") 创建进程池，fork 不可用时回退到默认 mp.Pool（L654-L659）；
• 在非 Linux 平台（如 Windows）上，n_jobs 会被强制置为 1 并打印提示（L358-L364）；
• n_jobs=None 时：若安装了 psutil 则用物理核心数 psutil.cpu_count(logical=False)，否则回退到逻辑核心数 mp.cpu_count() 并提示安装 psutil（L367-L376）。

与标准方法的等价性与测试验证

模块默认行为对应标准调用 filter.find_label_issues(…, filter_by="low_self_confidence", return_indices_ranked_by="self_confidence")。若用 quality_score_kwargs = {"method": "normalized_margin"} 初始化 LabelInspector，则对应关系变为 filter_by="low_normalized_margin"、return_indices_ranked_by="normalized_margin"（见 get_label_issues 的 docstring）。

仓库的单元测试 tests/test_filter_count.py#L906-L1020（test_batched_label_issues）直接验证了这套等价性：

• 以不同 batch_size（数据量的 1/4、1/2、全量+100、1）与不同 n_jobs（1、4、None）运行 find_label_issues_batched()，断言结果两两全等；
• 与标准 filter.find_label_issues(…, filter_by="low_self_confidence") 的结果做 Jaccard 相似度对比，断言交集/并集 > 0.95；
• 对 return_mask=True 的结果同样做 Jaccard 验证（> 0.95）；
• 传入 quality_score_kwargs={"method": "normalized_margin"} 与 num_issue_kwargs={"estimation_method": "off_diagonal_calibrated"} 时，断言问题数量与 count.num_label_issues(…, estimation_method="off_diagonal_calibrated") 相差小于 2；
• 最后用 np.save() 落盘的 labels.npy / pred_probs.npy 文件路径运行，断言与直接传数组的结果一致。

这些测试同时印证了两个关键事实：批大小与进程数不影响最终结论（结果对 batch_size、n_jobs 不敏感），以及 批处理结果与标准全量方法高度一致而非完全相同（存在 5% 以内的 Jaccard 差异与 ±1 的数量舍入差异，来源是增量式统计的浮点累积）。

实战建议与注意事项

小结

cleanlab.experimental.label_issues_batched 以"内存映射对象 + 小批量两趟扫描"的设计，把 cleanlab 的标准标签错误检测流程扩展到了超大数据集场景：find_label_issues_batched() 提供开箱即用的一站式体验，LabelInspector 提供可完全自定义的两趟扫描控制流，而增量阈值估计、按批评分、多进程统计与校准等底层细节共同保证了结果对标准方法的高度近似。结合 tests/test_filter_count.py 中的测试用例，你可以在自己的数据上验证这一近似关系，并根据内存预算灵活选择 batch_size 与 n_jobs，在受限硬件上完成大规模数据质量检测。

【免费下载链接】cleanlab Cleanlab's open-source library is the standard data-centric AI package for data quality and machine learning with messy, real-world data and labels. 项目地址: https://gitcode.com/GitHub_Trending/cl/cleanlab

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考