开源大语言模型（LLMs） 在英语方面表现出色，但难以与其他语言（尤其是东南亚语言）搭配使用。这主要是由于缺乏这些语言的训练数据、对当地文化的理解有限，以及 token 不足以捕捉独特的语言结构和表达。
为了充分满足客户需求，非英语国家地区的企业必须超越通用模型，并对其进行定制，以捕捉当地语言的细微差别，确保客户体验无缝且有影响力。
在这篇博文中，我们将探讨 Viettel Solutions （Viettel Corporation 快速发展的子公司）如何利用 NVIDIA NeMo Curator 处理高质量的 越南语数据 来训练 Llama 3 ViettelSolution 8B，这是一种先进的 LLM，现在排名在 VMLU 排行榜的前列。NeMo Curator 是一款 GPU 加速的数据管护工具，可为预训练 LLM 提供大规模、高质量的数据集。
在这一过程中，关键的第一步是精心策划大规模的高质量数据集。本文将指导您完成所使用的数据管护工作流，包括每个阶段的示例代码和详细的探索性数据分析（EDA），以说明每个步骤的影响。博文结束后，您将拥有清晰的路线图和参考，以便轻松开始使用 NeMo Curator，无论是越南语还是其他语言。
Viettel Solutions 是为越南政府和企业提供数字化转型解决方案的先驱，专注于满足各行各业对采用人工智能（AI）的日益增长的需求。Viettel 的愿景是引领生成式人工智能领域的发展，并为客户开发人工智能赋能的产品，Viettel 与 NVIDIA NeMo Curator 团队开展了合作。
Viettel Solutions 数据分析主管 Tuan Nguyen 表示：“NeMo Curator 的 GPU 加速功能 (包括 exact 和 fuzzy deduplication 以及 heuristic 和 classifier filtering) 将准确性提高了 10%，将训练时间缩短了三倍，并将数据集大小减少了 60%。”
预备知识和环境设置
如要遵循本文中介绍的步骤，请确保您已进行以下设置：
- CUDA 和 NVIDIA 驱动程序：CUDA 12.3 与驱动程序 545.23.08
- Ubuntu 22.04
- NVIDIA 容器工具包 版本 1.15.0
安装
首先，按照 NeMo Curator 存储库 的 README 文件中的说明安装 CPU 和 CUDA 加速模块的说明安装 NeMo Curator。
接下来，安装 datasets 和 jsonlines 包，这些包稍后会用到。
pip install datasets
pip install jsonlines
要继续进行数据处理，需要设置 Dask 环境。Dask 是一个灵活的开源库，可在 Python 中实现并行和分布式计算，使您能够跨多个核心甚至集群扩展计算。通过分配任务，Dask 显著提高了数据处理过程的速度和效率。
我们在搭载 128 核 CPU 和 2TB RAM 的 NVIDIA DGX A100 上运行此实验，以处理数据集大小。根据您的数据集和计算资源，您可能需要相应地调整 Dask Worker 配置。您可以使用以下命令启动 Dask 集群：
import nemo_curator
from dask.distributed import Client, LocalCluster
# Start a Dask cluster with 12 workers, each limited at 64GB of memory. You might need to adjust these numbers according to your computing resources
cluster = LocalCluster(n_workers=12, processes=True, memory_limit= '64GB')
client = Client(cluster)
数据处理流程概述
数据管护管道包括以下关键步骤：
- 下载和分片 ：从各种来源下载数据集，然后进行组合和分片，以实现高效的分布式处理。
- Unicode 重新格式化 ：文本被标准化为一致的 Unicode 格式。
- 精确的重复数据删除 ：删除精确的重复数据以减少冗余。
- Quality filtering
- 启发式过滤 ：应用基于规则的过滤器以删除低质量内容。
- 基于分类器的过滤 ：使用机器学习根据质量对文档进行分类和过滤。
数据采集
我们从多个数据集获取内容，以丰富大型语言模型（LLMs）的训练数据的多样性和数量。这些数据集包括：
- C4 数据集的越南语子集，是一个庞大且多样化的网络爬网文本数据集合。
- OSCAR 数据集版本 23.01 的越南语子集，是 web-crawled 数据的聚合。
- 维基百科的越南文文章 ，提供结构化和信息丰富的内容。
- 越南新闻语料库 ，提供与当地相关的新闻文章。
每个数据集都可通过 Hugging Face Hub 访问和下载，由于 OSCAR 受到访问限制，还需要执行其他步骤。请注意，OSCAR 需要接受 数据集页面 上的条件；然后使用 Hugging Face 访问令牌 进行下载。
下载数据集并将其转换为 Parquet
Parquet 已针对像 Dask 这样的分布式系统进行了优化，支持轻松分区和并行处理，从而提高处理大规模数据时的性能。为了本文的目的，所有数据集阶段都将以 Parquet 格式保存。
以下代码片段从 Hugging Face 下载数据集，并将其另存为 Parquet 文件。
import os
from datasets import load_dataset as load_hf_dataset
from datasets import DownloadConfig
data_dir = "./datasets/"
download_config = DownloadConfig(num_proc=4)
# Load and save Vietnamese Wikipedia dataset
ds = load_hf_dataset("wikimedia/wikipedia", "20231101.vi")
ds["train"].to_parquet(os.path.join(data_dir, "wiki_vi_231101.parquet"))
# Load and save Vietnamese news corpus
ds = load_hf_dataset("jetaudio/binhvq_news")
ds["train"].to_parquet(os.path.join(data_dir, "binhvq_news_train.parquet"))
# Load and save OSCAR dataset
ds = load_hf_dataset("oscar-corpus/OSCAR-2301", language="vi", token=True, download_config=download_config, trust_remote_code=True)
ds['train'].to_parquet(os.path.join(data_dir, 'oscar_vi.parquet'))
# Load and save C4 dataset
ds = load_hf_dataset("allenai/c4", data_files='multilingual/c4-vi.*.json.gz', download_config=download_config, trust_remote_code=True)
ds['train'].to_parquet(os.path.join(data_dir, "c4_vi.parquet"))
我们利用 NeMo Curator 域分类器模型 将文档分类为支持的 26 个域之一。如图 3 所示，分布相对均匀，许多域占总数据的 3% 到 6% 之间。这表明数据集非常多样化，涵盖广泛的主题，这有助于预训练通用语言模型。
合并和标准化格式
下载数据集后，下一步是对所有来源的数据进行标准化和格式化。这些数据组合成一个数据集，只保留“文本”字段，因为用于训练模型的所有文本数据都在此字段中。非文本数据和其他信息通常无法帮助完成此任务。
from datasets import concatenate_datasets
# Combine datasets and standardize format
datasets = [os.path.join(data_dir, file) for file in ["wiki_vi_231101.parquet", "c4_vi.parquet", "oscar_vi.parquet", "binhvq_news_train.parquet"]]
data_files = {"train": datasets[0]}
ds = load_hf_dataset("parquet", data_files=data_files)
ds = ds["train"].remove_columns([col for col in ds["train"].column_names if col != "text"])
for d in datasets[1:]:
ds_ = load_hf_dataset("parquet", data_files={"train": d})
ds_ = ds_["train"].remove_columns([col for col in ds_["train"].column_names if col != "text"])
ds = concatenate_datasets([ds, ds_])
将组合数据集分片
然后，将组合数据集分解成较小的块。执行分片以在 Dask 集群中的多个工作者之间均匀分布数据，从而在数据管护阶段促进高效的并行处理。
# Define paths for raw data
raw_data_directory = os.path.join(data_dir, "raw")
# Shard the dataset
num_shards = 256
for shard_idx in range(num_shards):
shard = ds.shard(index=shard_idx, num_shards=num_shards)
shard.to_parquet(os.path.join(raw_data_directory, f"{shard_idx}.parquet"))
使用 NeMo Curator 进行高质量数据处理
本节介绍我们从 NeMo Curator 中使用的不同技术。Unicode 重新格式化、精确重复数据删除、启发式过滤和基于分类器的过滤用于处理此数据集并将其细化为高质量的最终版本。
Unicode 重新格式化
Unicode 重新格式化是必要的预处理步骤，可确保文本数据标准化，并且不会出现编码错误，这在网络爬网数据集中很常见。以下代码演示了如何使用 NeMo Curator 执行 Unicode 重新格式化：
from nemo_curator import Modify
from nemo_curator.modifiers import UnicodeReformatter
from nemo_curator.utils.distributed_utils import read_data, write_to_disk
from nemo_curator.utils.file_utils import get_all_files_paths_under
from nemo_curator.datasets import DocumentDataset
# Define paths for Unicode formatted data
unicode_formatted_output_path = os.path.join(data_dir, "formatted")
def load_dataset(input_data_dir, file_type="parquet"):
files = list(get_all_files_paths_under(input_data_dir))
raw_data = read_data(files, file_type=file_type, backend="pandas", add_filename=True)
dataset = DocumentDataset(raw_data)
return dataset
# Load the raw data
raw_data = load_dataset(raw_data_directory, file_type="parquet")
# Initialize the Unicode reformatter
cleaner = Modify(UnicodeReformatter())
# Apply Unicode reformatting
cleaned_data = cleaner(raw_data)
# Save the cleaned data to disk
write_to_disk(cleaned_data.df, unicode_formatted_output_path, write_to_filename=True, output_type="parquet")
向文档添加自定义 ID
在继续进一步的数据集管护步骤之前，建议通过向每个文档添加唯一 ID 来对数据集进行预处理。这些 ID 可充当追踪器，帮助在整个管护过程中识别重复文档或低质量文档，确保每个文档在整个处理过程中保持唯一身份识别。
NeMo Curator 提供了一个 AddId 类，允许用户使用指定的前缀格式 (例如 <prefix>_<id>) 将自定义 ID 插入文档。以下代码片段演示了此步骤：
from nemo_curator import AddId
# Define paths for input data and output with added IDs
add_id_input_data_dir = unicode_formatted_output_path
added_id_output_path = os.path.join(data_dir, "add_id")
add_ID_id_prefix = "VI_"
# Load the formatted dataset
dataset = DocumentDataset.read_parquet(add_id_input_data_dir)
# Initialize the AddId class with a specified prefix and start index
add_id = AddId(id_field='id', id_prefix=add_ID_id_prefix, start_index=0)
# Apply the ID addition to the dataset
id_dataset = add_id(dataset)
# Save the dataset with added IDs to disk
write_to_disk(id_dataset.df, output_file_dir=added_id_output_path, write_to_filename=True, output_type="parquet")
精确的重复数据删除
精确的重复数据删除功能可从数据集中删除相同的重复数据。通过消除精确的重复，我们可以确保每个数据点对训练过程的贡献独一无二，从而增强数据集的多样性和整体质量。
此阶段利用 GPU 加速使用 GPU Dask 集群。当前集群基于 CPU，因此必须关闭集群，并在 GPU 支持下启动新集群。
要关闭现有的集群，请使用以下代码：
client.cluster.close()
client.shutdown()
然后初始化 GPU Dask 集群：
os.environ["DASK_DATAFRAME__QUERY_PLANNING"] = "False"
from nemo_curator.utils.distributed_utils import get_client
def pre_imports():
import cudf
client = get_client(cluster_type='gpu', set_torch_to_use_rmm=False)
client.run(pre_imports)
精确重复数据删除的实现如下所示：
from nemo_curator.modules import ExactDuplicates
# Define input and output paths
exact_dedup_input_dataset_dir = added_id_output_path
exact_dedup_base_output_path = os.path.join(data_dir, "exact_dedup")
exact_dedup_log_dir = os.path.join(exact_dedup_base_output_path, "log")
exact_dedup_output_dir = os.path.join(exact_dedup_base_output_path, "data")
deduped_output_dir = os.path.join(data_dir,"remove_duplicate")
# Create directories for logs and output
!mkdir -p {exact_dedup_log_dir}
!mkdir -p {exact_dedup_output_dir}
!mkdir -p {deduped_output_dir}
# Parameters for ExactDuplicates
exact_dedup_dataset_id_field = "id"
exact_dedup_dataset_text_field = "text"
# Load the input dataset
input_dataset = DocumentDataset.read_parquet(exact_dedup_input_dataset_dir, backend="cudf")
# Initialize and run exact deduplication
exact_dup = ExactDuplicates(
logger=exact_dedup_log_dir,
id_field=exact_dedup_dataset_id_field,
text_field=exact_dedup_dataset_text_field,
hash_method="md5",
cache_dir=exact_dedup_output_dir
)
duplicates = exact_dup(dataset=input_dataset)
print(f"Number of exact duplicate files: {len(duplicates)}")
# Load the dataset,exact duplicates to identify and remove duplicate IDs
input_dataset = DocumentDataset.read_parquet(added_id_output_path, backend="cudf")
exact_duplicates = DocumentDataset.read_parquet(
os.path.join(exact_dedup_output_dir, "_exact_duplicates.parquet"), backend="cudf")
# Extract list of duplicate document IDs
exact_docs_to_remove = exact_duplicates.df.map_partitions(
lambda x: x[x._hashes.duplicated(keep="first")]
)
# Remove duplicated documents from the input dataset
result = input_dataset.df[
~input_dataset.df[exact_dedup_dataset_id_field].isin(exact_docs_to_remove[exact_dedup_dataset_id_field].compute())
]
# Save the final deduplicated dataset
write_to_disk(result, output_file_dir=deduped_output_dir, write_to_filename=True, output_type="parquet")
启发式质量过滤
启发式质量过滤旨在根据预定义的启发式删除低质量内容，从而提高数据集的质量。这种方法包括对数据集应用一系列过滤器，以消除不需要的数据特征，例如过多的特殊字符、过短或过长的文本，或者其他可能会对模型性能产生负面影响的标准。
我们使用 已配置的 YAML 文件 来定义启发式过滤器。该文件列出了用于构建过滤器工作流的过滤条件和设置。您可以根据需要自定义过滤器或更改阈值。filter_pipeline 辅助程序会读取 YAML 设置，并逐步将每个过滤器应用于数据集。
from nemo_curator.utils.config_utils import build_filter_pipeline
import warnings
# Define paths for input data and output data after heuristic filtering
HF_input_data_dir = deduped_output_dir
HF_output_path = os.path.join(data_dir, "heuristic_filtering")
# Create a directory for the configuration file if it doesn't exist
os.makedirs("config", exist_ok=True)
# Download the YAML configuration file for heuristic filtering
!wget https://raw.githubusercontent.com/NVIDIA/NeMo-Curator/main/config/heuristic_filter_non-en.yaml -O ./config/heuristic_filter_non-en.yaml
# Specify the path to the configuration file
filter_config_file = "./config/heuristic_filter_non-en.yaml"
os.makedirs(HF_output_path, exist_ok=True)
# Load the filters from the YAML configuration file
filter_pipeline = build_filter_pipeline(filter_config_file)
# Load the dataset
dataset = DocumentDataset.read_parquet(HF_input_data_dir, backend="pandas")
# Suppress specific warnings during filtering
with warnings.catch_warnings():
warnings.simplefilter("ignore", category=UserWarning)
# Apply the heuristic filters to the dataset
result_data = filter_pipeline(dataset)
# Save the filtered dataset to disk
result_data.to_parquet(HF_output_path, write_to_filename=True)
令牌数量分布
现在，检查启发式过滤如何改变数据集。在过滤之前，数据集包含各种文本长度，有些文档短到几个令牌，有些则扩展到超过 16K 令牌。经过过滤后，数据集的文本长度和令牌数量分布更加一致。该过滤过程有效地删除了极短的文档（例如少于 64 个令牌的文档），并缩减了可能包含冗余或不相关内容的过长文档。
基于角色的指标
图 5 显示了每个指标在启发式管理前后数据集的比较分析。
箱线图突出显示了启发式过滤后异常值的显著减少。对于符号，第 95 百分位数从 8.84% 降至 5.47%，而对于数字，则从 11.19% 降至 6.14%。空白部分的最大降幅也从 76.06% 降至 25.88%，第 95 百分位保持稳定。这些降低表明，启发式过滤可有效地定位并移除具有高比例符号、数字或空格的噪声数据，从而提高整体数据集质量。
过滤删除了非常长的文档，但文档之间的总体字数分布仍然相似。这表明删除了带有畸形标记的异常长文档。
基于分类器的质量过滤
启发式过滤使用简单的规则移除低质量内容，但无法捕捉更复杂的质量模式。基于分类器的过滤使用经过训练的分类器模型将内容分类为高质量或低质量，从而以更智能、更灵活的方式处理简单规则可能会忽略的各种数据集。
为训练分类器准备数据
训练质量分类器需要高质量和低质量内容的代表性样本。对于高质量数据，我们使用了维基百科越南语版的文章，这些文章通常结构合理且可靠。低质量样本来自未经过滤的爬网式越南新闻语料库。
数据准备方式如下：
# Paths for high-quality and low-quality sample data
hq_samples_path = os.path.join(data_dir, "classifier_filtering/train_samples/hq")
lq_samples_path = os.path.join(data_dir, "classifier_filtering/train_samples/lq")
# Load and shard the high-quality dataset
ds = load_hf_dataset("wikimedia/wikipedia", "20231101.vi")
num_shards = 8
for shard_idx in range(num_shards):
shard = ds["train"].shard(index=shard_idx, num_shards=num_shards)
shard.to_parquet(os.path.join(hq_samples_path, f"{shard_idx}.parquet"))
# Load and shard the low-quality dataset
ds = load_hf_dataset("vietgpt/binhvq_news_vi",split="train[:100000]")
num_shards = 32
for shard_idx in range(num_shards):
shard = ds.shard(index=shard_idx, num_shards=num_shards)
shard.to_parquet(os.path.join(lq_samples_path, f"{shard_idx}.parquet"))
训练分类器
分类器使用 FastText 进行训练，FastText 提供了一种高效的文本分类方法。以下是使用标记为高质量和低质量的样本训练分类器的方法：
from nemo_curator.modifiers import FastTextLabelModifier
import fasttext
import random
# Function to create labeled samples
def create_samples(data_path, label, num_samples):
raw_dataset = DocumentDataset.read_parquet(data_path, backend='pandas')
label_quality = Modify(FastTextLabelModifier(label))
labeled_dataset = label_quality(raw_dataset)
labeled_samples = labeled_dataset.df.sample(frac=num_samples / len(labeled_dataset.df))
return labeled_samples["text"].compute().values.tolist()
# Prepare training data
low_quality_samples = create_samples(lq_samples_path, "__label__lq", 100000)
high_quality_samples = create_samples(hq_samples_path, "__label__hq", 100000)
train_samples = low_quality_samples + high_quality_samples
random.shuffle(train_samples)
# Save training data to a file
train_file = "./cf_model_fasttext.train"
with open(train_file, "w", encoding="utf-8") as f:
for sample in train_samples:
f.write(sample + "\n")
# Train the FastText classifier
model = fasttext.train_supervised(input=train_file, lr=0.01, dim=100, epoch=5, wordNgrams=2)
model_path = "./cf_model_fasttext_model.bin"
model.save_model(model_path)
对数据集进行分类和筛选
经过训练后，分类器将用于筛选数据集，根据学习到的区别将文档分为高质量和低质量：
from nemo_curator.filters import FastTextQualityFilter
from nemo_curator import ScoreFilter
# Define paths and load the dataset
CF_input_data_dir = HF_output_path
CF_output_path = os.path.join(data_dir, "classifier_filtering/output")
target_dataset = DocumentDataset.read_parquet(CF_input_data_dir, "parquet")
# Set up the filtering pipeline
filter_pipeline = ScoreFilter(FastTextQualityFilter(model_path), score_field="quality_score", score_type=float)
filtered_dataset = filter_pipeline(target_dataset)
# Save the filtered dataset
write_to_disk(filtered_dataset.df, output_file_dir=CF_output_path, write_to_filename=True, output_type="parquet")
删除敏感和情感数据
成人和敏感主题领域以及积极和消极情绪都显著减少。这使得模型更安全、更中立，并且能够更好地处理各种情境并做出适当的反应。
保留内容多样性
再次运行 域分类器模型 ，并检查数据集内容的多样性。现在，数据集显示了跨领域的均衡分布，其中大多数领域占数据的 3%至 8%。从新闻和法律到游戏和汽车等专业领域，这种多样性可以确保模型能够处理各种主题。即使经过过滤以提高质量并删除有害内容，基本的多样性也得以保留，这对于构建通用的通用语言模型至关重要。
在每个阶段结束后减小数据集大小
大约 90% 的数据集被删除，这些文档的样本质量较低、噪声较大或格式错误。这种选择性过滤可确保训练数据具有最高质量。降幅最大的是基于分类器的过滤（45.43%），这表明大量内容被标记为质量较低或有害，并在此阶段被移除。启发式过滤占数据删除量的 35.74%，目标是样本长度、重复 n-grams 和噪声等问题。精确重复数据删除过滤了一小部分数据（8.31%）。
嵌入可视化
最终数据集的分布与原始数据集类似。主题的多样性仍然得到保留，大多数领域的代表性仍然很好。在完成此步骤后，一些较小的集群的定义略显偏高，这可能是由于删除了低质量或有害内容。
通过启发式过滤和基于分类器的过滤，数据集保持了广泛的领域多样性。表示特定领域的不同聚类仍然定义明确，而更通用和重叠的领域继续显示互连，从而确保数据集保持平衡和全面，便于预训练目的。
结束语
这篇博客文章展示了用于越南文本数据的数据管护工作流 Viettel Solutions ，以及一项分析，以探索管护过程的每个阶段对数据集的影响。该流程使用 NVIDIA NeMo Curator ，这是一种宝贵的工具，可用于准备预训练语言模型的大型数据集，同时注重质量、效率和可扩展性。它在数据管护过程中具有一系列显著优势，包括：
- 使用启发式和基于分类器的过滤器消除噪声和有害内容，从而提高数据集质量。
- 保留数据集的基本结构，确保核心特征在经过整理后保持不变。
- 适应不同的数据集，为每个语料库提供量身定制的方法。
如需查看本文使用的完整代码，请参阅 Jupyter Notebook 。查看 NeMo Curator 示例脚本，了解其他技术，例如 Fuzzy Deduplication 和 PII redaction。
