重要提示

您正在查看 NeMo 2.0 文档。此版本引入了 API 的重大更改和一个新的库 NeMo Run。我们目前正在将 NeMo 1.0 的所有功能移植到 2.0。有关先前版本或 2.0 中尚不可用的功能的文档，请参阅 NeMo 24.07 文档。

分类器和启发式质量过滤#

背景#

大型数据集通常包含许多被认为是“低质量”的文档。在这种情况下，“低质量”数据仅表示我们不希望下游模型学习的数据，“高质量”数据是我们希望下游模型学习的数据。定义质量的指标可能有所不同。有些启发式方法通过收集简单的统计信息来衡量质量，例如文档有多少标点符号、文档有多长以及文档的重复程度。然后，您可以按这些统计信息过滤文档。相反，您可能拥有一个高质量的数据集合，您希望新数据集与之对齐。您可以训练一个简单的分类器来区分看起来与这些高质量文档相似的文档和不相似的文档。

NeMo Curator 提供了用于这两种过滤的模块，并提供了一个简单的界面，用于添加您自己的过滤器并将它们与现有过滤器组合起来。您还可以使用这些模块来收集有关文档的统计信息和元数据，而无需删除任何文档。有 30 多个过滤器可用于英语、非英语和代码数据集。

用法#

ScoreFilter 是 NeMo Curator 中过滤的核心。让我们检查一下这个小例子

import nemo_curator as nc
from nemo_curator.datasets import DocumentDataset
from nemo_curator.utils.file_utils import get_all_files_paths_under
from nemo_curator.filters import WordCountFilter

files = get_all_files_paths_under("books_dataset/")
books = DocumentDataset.read_json(files, add_filename=True)

filter_step = nc.ScoreFilter(
                WordCountFilter(min_words=80),
                text_field="text",
                score_field="word_count",
            )

long_books = filter_step(books)

long_books.to_json("long_books/", write_to_filename=True)

要关注的中心部分是 filter_step 的创建。WordCountFilter(min_words=80) 创建并配置一个过滤器对象。过滤器对象是一个从抽象基类 nemo_curator.filters.DocumentFilter 继承的类。这个基类要求继承者实现两个方法，score_document 和 keep_document。对于这个例子，让我们看一下 WordCountFilter 的简化版本。

class WordCountFilter(DocumentFilter):

  def __init__(self, min_words=50, max_words=100000, lang='en'):
    self._min_words = min_words
    self._max_words = max_words
    self._word_splitter = get_word_splitter(lang)
    self._name = 'word_count'

  def score_document(self, text: str):
    return len(self._word_splitter(text))

  def keep_document(self, score: int):
    return self._min_words <= score <= self._max_words

通过这个实现，每个函数的作用变得清晰。score_document 接收文档的文本，并返回文档中的单词数。keep_document 接收由 score_document 输出的分数（在本例中为单词数），如果分数表明应保留文档，则返回 True，如果应删除文档，则返回 False。现在，重要的是要注意 WordCountFilter 和 DocumentFilter 仅对单个文档进行操作。为了将过滤器应用于整个数据集，我们必须使用 ScoreFilter。

filter_step = nc.ScoreFilter(
    WordCountFilter(min_words=80),
    text_field="text",
    score_field="word_count",
)

ScoreFilter 的构造创建了一个可以应用于 DocumentDataset 而不仅仅是单个文档的函数。text_field 指定数据集中的字段，该字段保存应传递给过滤器的 score_document 函数的文档。score_field 是一个可选参数，允许您在文档的给定元数据字段中记录分数，如果指定，它将与其余元数据一起写入磁盘。

在某些情况下，数据集可能带有您想要直接过滤的元数据。或者，您可能只想简单地添加新的元数据，而不对其进行过滤。Filter 和 Score 模块允许您分别完成每个任务。

例如，如果上述示例中的数据集预先填充了 word_count 字段，您可以将其重写如下

books = DocumentDataset.read_json(files, add_filename=True)

filter_step = nc.Filter(
                WordCountFilter(min_words=80).keep_document,
                filter_field="word_count",
            )

long_books = filter_step(books)

long_books.to_json("long_books/", write_to_filename=True)

或者，如果您只想跟踪文档中单词的长度，而不根据它们进行过滤，您可以将其重写如下

books = DocumentDataset.read_json(files, add_filename=True)

filter_step = nc.Score(
                WordCountFilter(min_words=80).score_document,
                text_field="text",
                score_field="word_count",
            )

annotated_books = filter_step(books)

annotated_books.to_json("annotated_books/", write_to_filename=True)

批量过滤#

虽然上面定义的评分和过滤函数对单个文档进行操作，但 NeMo Curator 可以利用批量操作的函数来提高性能。为了实现这一点，您可以使用 batched 装饰器注释您的函数。此装饰器将导致文档/分数的 pandas 系列传递给函数，而不是单个文档/分数。以下是重写的 WordCountFilter，它在 keep_document 中使用批处理。

from nemo_curator.utils.decorators import batched

class WordCountFilter(DocumentFilter):

  def __init__(self, min_words=50, max_words=100000, lang='en'):
    self._min_words = min_words
    self._max_words = max_words
    self._word_splitter = get_word_splitter(lang)
    self._name = 'word_count'

  def score_document(self, text: str):
    return len(self._word_splitter(text))

  @batched
  def keep_document(self, scores: pd.Series):
    pass_min = self._min_words <= scores
    pass_max = score <= self._max_words
    return pass_min & pass_max

当您使用 batched 装饰器时，从函数返回的序列的索引必须与传入的索引保持相同。由于在当前过滤器之前应用了过滤器，因此索引可能不是连续的。在上面的代码中，索引将自动相同，因此不需要更改。但是，当编写将序列转换为不同结构（如列表）的函数时，需要特别注意。以下代码示例演示了此错误可能是什么样子，以及如何修复它。

class BuggyLengthFilter(DocumentFilter):

  @batched
  def score_document(self, documents: pd.Series):
    scores = []
    for document in documents:
      scores.append(len(document))

    return pd.Series(scores) # Bad! Does not preserve the index

class CorrectLengthFilter(DocumentFilter):

  @batched
  def score_document(self, documents: pd.Series):
    scores = []
    for document in documents:
      scores.append(len(document))

    return pd.Series(scores, index=documents.index) # Good! Preserves the index

分类器过滤#

我们已实现的基于分类器的过滤方法与 Brown 等人，2020 年中使用的方法非常相似，并训练了一个二元 skip-gram 分类器，该分类器可用于区分低质量和高质量文档。为了实现这一点，我们使用了 fastText 提供的函数。按照 fastText 文档中提供的示例，我们首先创建一个包含高质量和低质量训练文档的文件。我们在 examples/classifier_filtering.py 中提供了一个如何训练和使用模型的示例。

我们还为相同的功能提供了 CLI 脚本。prepare_fasttext_training_data 脚本将从输入数据集中随机抽样文档，并将它们准备好用于训练 fasText skip-gram 分类器。对于高质量数据集，我们建议从 OpenWebText2 或 Wikipedia 中抽样，而 Common Crawl 的未过滤版本可用作低质量数据集。

prepare_fasttext_training_data \
  --input-data-dir=<Specify the path to common-crawl/low-quality data> \
  --output-num-samples=<Specify the number of low-quality documents to be used for training> \
  --label='__label__cc' \
  --output-train-file=${res_dir}/cc_samples.txt \

prepare_fasttext_training_data \
  --input-data-dir=<Specify the path to high-quality data> \
  --output-num-samples=<Specify the number of high-quality documents to be used for training> \
  --label='__label__hq' \
  --output-train-file=${res_dir}/hq_samples.txt \

一旦样本已准备好并写入 .txt 文件，用户可以使用 train_fasttext 脚本，该脚本读取 .txt 文件中的样本，以便训练质量分类器。train_fasttext 将读取 .txt 文件中的所有样本，将数据拆分为训练集和验证集，并训练二元 skip-gram 分类器。训练后，它会在验证样本上评估模型，并将预测写入 jsonl 文件，并将混淆矩阵打印到 stdout。

train_fasttext \
  --fasttext-files-dir=${res_dir} \
  --output-train-file=${res_dir}/fasttext_samples.train \
  --output-validation-file=${res_dir}/fasttext_samples.valid \
  --output-model=${res_dir}/cc_filter_test.bin \
  --output-predictions=${res_dir}/preds.jsonl

最后，通过训练模型并能够提供质量分数，它可以用于质量过滤。类似于 filter_documents 如何使用 fastText 模型 lid.176.bin 执行语言识别一样，我们提供了一个默认配置，该配置可用于使用 fastText 模型进行基于分类器的质量过滤。此外，此过滤器实现了 Pareto 抽样方法，如 Brown 等人，2020 年中所述。

filter_documents \
  --input-data-dir=<Specify the path to common-crawl/uncurated data> \
  --filter-config-file=./config/fasttext_quality_filter.yaml \
  --output-retained-document-dir=<Output directory to which high-quality documents will be written> \
  --output-removed-document-dir=<Output directory to which low-quality documents will be written> \
  --log-dir=${log_dir}/fasttext_classifier \

启发式过滤#

与其他过滤步骤一样，NeMo Curator 中的基于启发式的过滤可以使用 ScoreFilter 或 filter_documents 实用程序来执行。过滤器可以在 NeMo Curator 中使用 Sequential 链接，如下所示。

filter_step = nc.Sequential([
    ScoreFilter(
        WordCountFilter(min_words=80),
        score_field="word_count",
    ),
    ScoreFilter(IncompleteStoryFilter()),
    ScoreFilter(RepeatingTopNGramsFilter(n=2, max_repeating_ngram_ratio=0.2)),
    ScoreFilter(RepeatingTopNGramsFilter(n=3, max_repeating_ngram_ratio=0.18)),
    ScoreFilter(RepeatingTopNGramsFilter(n=4, max_repeating_ngram_ratio=0.16)),
])

过滤器配置文件 config/heuristic_filter.yaml 提供了一个通用的启发式过滤器列表，这些过滤器经过测试并显示可以提供用于训练的文档，从而提高语言模型下游任务的性能。这些过滤器足够通用，用户可以随意删除过滤器级联中的某些过滤器，并尝试不同过滤器配置/参数的结果。

此外，这些过滤器已用于整理高质量的非英语文档。但是，建议在应用于非英语数据时，用户通过指定 --document-score-dir 参数来写出文档分数。这将允许用户检查特定过滤器是否负责不希望地从语料库中删除大量文档。

filter_documents \
  --input-data-dir=<Specify path to input dataset> \
  --filter-config-file=./config/heuristic_filter_en.yaml \
  --output-retained-document-dir=<Output directory to which high-quality documents will be written> \
  --output-removed-document-dir=<Output directory to which low-quality documents will be written> \
  --output-document-score-dir=<Output directory to which document scores will be written> \
  --log-dir=${log_dir}/heuristic_filter