MindSpore Transformers（MindFormers）是昇思 MindSpore 面向大语言模型打造的开发套件，为 LLM 预训练、微调任务提供完整的数据预处理链路。大语言模型训练依赖 TB 级海量文本语料，原始 JSON、TXT 文本格式存在 IO 读取慢、内存占用高、分布式多机加载冲突、样本混洗效率低等问题。如果直接在线实时解析原始文本，大规模训练场景极易出现 CPU 预处理瓶颈、磁盘 IO 打满，拖慢 NPU 算力利用率。MindSpore Transformers 提供面向 LLM 的大规模数据预处理能力，基于 MindRecord、Megatron 二进制两套存储方案，支持离线分词、样本 Packing、多源数据集混合采样、分布式分片存储，适配十亿到万亿参数大模型预训练、SFT 微调任务，结合 MindStudio 工具完成数据质量校验，解决大规模语料存储、读取、分发难题，充分释放昇腾 NPU 集群训练算力。

一、LLM 大规模数据预处理核心能力

MindSpore Transformers 针对大语言模型构建标准化大规模数据存储与预处理体系，五大核心能力支撑 TB 级语料处理。

一是双格式大规模二进制存储，原生支持 MindRecord 与 Megatron Bin/Index 两套二进制存储格式。MindRecord 支持多字段结构化存储、索引快速检索；Megatron 格式针对纯文本预训练做深度优化，支持 TB 级语料分片切分，大幅降低训练阶段的解码开销，避免原始文本反复分词带来的 CPU 压力。

二是离线全链路预处理流水线，支持原始语料清洗、对话模板组装、分词编码、长文本截断、样本 Packing 打包，全部离线完成。训练阶段直接读取已经 tokenize 完成的二进制样本，训练时不再执行分词逻辑，消除在线预处理瓶颈，适配 4k、8k、32k 超长序列大模型场景。

三是多源数据集混合采样，支持同时加载多个不同来源、不同领域的数据集，配置各数据源采样权重，实现预训练阶段语料比例可控；支持多机分布式场景下的全局混洗，避免单机局部采样带来的数据分布偏移问题。

四是分布式分片与容错读取，支持将超大语料切分为多个分片文件，适配多机多卡集群训练，各 Rank 读取对应分片，避免多进程争抢同一个大文件；支持断点续读，预处理中断后可恢复继续生成数据集，无需从头处理全部语料，提升 TB 级数据集生产效率。

五是数据质量校验与调试联动，可对接 MindStudio 工具，对预处理输出样本做抽样校验，检查输入 id、attention_mask、标签合法性；可导出异常样本，快速定位脏数据、截断错误、模板拼接异常问题，把数据问题拦截在训练之前。

二、大规模数据预处理全流程实施流程

MindSpore Transformers LLM 大规模数据预处理遵循“原始语料准备‑离线预处理‑二进制存储生成‑冒烟校验‑分布式加载训练”闭环流程。

第一步原始语料准备，整理预训练或者微调原始文本，预训练使用纯文本 JSON，SFT 微调使用对话格式数据，完成基础清洗，过滤乱码、空样本；固定随机种子，保证采样、混洗可复现。

第二步离线预处理流水线执行，调用 MindFormers 预处理脚本，完成对话模板拼接、分词 tokenize、长文本截断、样本 Packing；选择存储格式：预训练大规模场景优先 Megatron Bin/Index，SFT 微调、多字段复杂样本优先 MindRecord 格式，配置分片大小，避免单文件体积过大。

第三步生成大规模二进制存储文件，执行离线转换，将分词之后的 token 序列落盘为分片二进制数据集；开启断点保存，若处理中断，可基于中间状态继续生成，不需要重新读取全部原始语料。

第四步小规模冒烟校验，使用少量分片做加载测试，抽样打印 token 序列、还原文本，校验分词、Packing、模板拼接正确性；配合 MindStudio 抽样 Dump 样本，校验字段维度，排查脏数据、序列越界问题。

第五步分布式训练加载数据集，在 msrun 拉起的多机集群中，使用对应 DataLoader 加载二进制分片数据集，配置多源采样权重、全局 shuffle；多 Rank 自动分配分片，避免文件锁冲突，执行预训练或者微调任务。

第六步迭代优化数据集，训练中如果发现 Loss 异常、收敛异常，回溯校验数据集样本，调整清洗规则、Packing 策略、数据源采样权重，重新生成数据集复测。整套流程把分词、清洗等重计算放在离线阶段，训练阶段只做简单 IO 读取，最大化昇腾 NPU 利用率。

三、大规模数据预处理核心代码实操

下面为 MindSpore Transformers LLM 大规模数据预处理示例代码，包含离线转换、MindRecord 生成、数据集加载、样本校验逻辑，适配预训练与 SFT 微调场景。

import json
from mindformers import PreTrainedTokenizerFast
from mindformers.tools import MindRecordWriter
from mindstudio_precision import PrecisionDebugger

# 1.初始化数据校验调试器，抽样校验预处理输出样本
def init_data_check_debug():
    config = {
        "common_dump_settings": {
            "dump_mode": 0,
            "enable_debug": True,
            "dump_path": "./dataset_dump",
            "iteration": "0|1‑10"

        },
        "overflow_detect": {
            "enable": False
        },
        "precision_compare": {
            "baseline_device": "cpu",
            "compare_metrics": ["cosine", "mse"]
        }
    }
    debugger = PrecisionDebugger(config=config)
    return debugger

# 2.离线预处理：原始json语料转token样本
def process_raw_corpus(tokenizer, raw_json_path, seq_len=4096):
    out_samples = []
    with open(raw_json_path, "r", encoding="utf‑8") as f:
        raw_datas = json.load(f)
    for item in raw_datas:
        text = item["text"]
        token_result = tokenizer(text, truncation=True, max_length=seq_len)
        input_ids = token_result["input_ids"]
        out_samples.append({"input_ids":input_ids})
    return out_samples

# 3.写入MindRecord大规模分片存储
def write_mindrecord_shard(samples, output_prefix, shard_num=8):
    writer = MindRecordWriter(output_prefix, shard_num=shard_num)
    schema = {"input_ids": {"type": "int32", "shape": [-1]}}
    writer.add_schema(schema, "llm_pretrain")
    for s in samples:
        writer.write(s)
    writer.commit()
    print(f"MindRecord分片数据集已生成，分片数量：{shard_num}")

if __name__ == "__main__":
    debug_tool = init_data_check_debug()
    # 加载LLM分词器
    tokenizer = PreTrainedTokenizerFast.from_pretrained("llama2‑7b")
    # 原始大规模语料路径
    raw_corpus_path = "./large_pretrain_corpus.json"
    # 离线预处理
    token_samples = process_raw_corpus(tokenizer, raw_corpus_path, seq_len=4096)
    # 输出分片MindRecord数据集
    write_mindrecord_shard(token_samples, "./llm_pretrain_dataset", shard_num=16)

    # 冒烟校验：抽样部分样本做数据校验
    debug_tool.start()
    check_sample = token_samples[0]
    debug_tool.step(check_sample["input_ids"])
    debug_tool.stop()
    check_result = debug_tool.compare_result()
    print("====数据集预处理样本校验报告====")
    print(f"样本序列长度：{len(check_sample['input_ids'])}")
    print(f"样本校验误差指标：{check_result}")

配套大规模数据集处理 json 配置文件，管控分片、序列长度、存储格式、多源采样参数：

{
  "preprocess_config":{
    "seq_length":4096,
    "enable_packing":true,
    "storage_type":"mindrecord",
    "shard_count":16
  },
  "multi_source":{
    "source_list":[
        {"path":"./domain_a","weight":0.6},
        {"path":"./domain_b","weight":0.4}
    ]
  },
  "distributed_loader":{
    "shuffle":true,
    "num_workers":8
  }
}

四、调试价值与应用总结

大语言模型预训练需要处理 TB 级海量文本，如果直接加载原始 JSON/TXT，训练过程中反复做分词、清洗，会造成 CPU 资源耗尽、IO 瓶颈，NPU 算力空闲浪费；同时超大单文件在多机分布式训练下容易出现读取冲突、采样不均衡。

MindSpore Transformers LLM 大规模数据预处理，实现原始语料准备‑离线预处理‑二进制分片存储‑冒烟校验‑分布式加载训练完整闭环。支持 MindRecord、Megatron 二进制两套大规模存储方案，离线完成分词、Packing、模板组装，训练阶段只做高速 IO 读取；支持多源数据集混合采样、分布式分片，结合 MindStudio 做样本质量校验，提前拦截脏数据与预处理逻辑错误，将数据集问题定位从数天缩短到小时级。