一、概述

大模型训练质量高度依赖数据预处理流水线，昇思 MindSpore 针对 LLM 训练提供两套预处理体系：MindSpore Dataset 原生数据流变换、离线文本预处理 + 在线动态 Token 变换。预处理包含文本清洗、分段、模板封装、Tokenization、Padding、截断、掩码构造。

与 CV 图像预处理不同，大语言模型预处理重点解决：指令模板统一、长文本截断策略、标签掩码（Loss Mask）、流水线加速、多进程数据加载。不合理的数据变换会造成训练泄露、上下文错乱、损失收敛困难。本文基于 MindSpore+MindFormers，搭建完整大模型 SFT 数据预处理链路，适配昇腾 NPU。

运行环境：MindSpore 2.3、MindFormers、昇腾 Ascend 910B/310P。

二、环境基础初始化

# env_init.py
import mindspore as ms
from mindspore import context
def init_ascend_env():
    context.set_context(
        mode=context.GRAPH_MODE,
        device_target="Ascend",
        device_id=0
    )
    ms.set_auto_parallel_context(parallel_mode=ms.ParallelMode.STAND_ALONE)
    print("昇腾NPU环境初始化完成")
if __name__ == "__main__":
    init_ascend_env()

三、离线文本清洗变换工具（预处理第一阶段）

原始互联网数据存在乱码、多余空格、无效符号，先执行离线清洗。

# text_clean.py
import re
def clean_raw_text(text: str) -> str:
    """基础文本清洗变换"""
    # 去除URL
    text = re.sub(r"http[s]?://\S+", "", text)
    # 去除多余换行、空格
    text = re.sub(r"\n+", "\n", text)
    text = re.sub(r"\s+", " ", text)
    # 过滤特殊不可见字符
    text = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f]", "", text)
    return text.strip()
def build_prompt_template(instruction: str, output: str = None):
    """构造LLM标准指令模板变换"""
    template = f"<|user|>\n{instruction}\n<|assistant|>\n"
    if output is not None:
        template += output
    return template
if __name__ == "__main__":
    raw = "  介绍MindSpore\n\nhttps://mindspore.cn  \n"
    clean_txt = clean_raw_text(raw)
    prompt = build_prompt_template(clean_txt, "昇思是华为全场景AI框架")
    print(prompt)

四、在线数据变换：数据集类 + Tokenizer 处理

在线变换在数据流水线实时执行，实现动态 Token 编码、截断、Padding，适配 MindSpore Dataset 迭代器。

# llm_dataset.py
import json
import mindspore.dataset as ds
from mindformers import AutoTokenizer
from text_clean import clean_raw_text, build_prompt_template
class SFTDataTransform:
    def __init__(self, tokenizer_path, seq_len=512):
        self.tokenizer = AutoTokenizer.from_pretrained(tokenizer_path)
        self.seq_len = seq_len
    def __call__(self, sample):
        """单条样本在线变换逻辑"""
        instruction = clean_raw_text(sample["instruction"])
        answer = clean_raw_text(sample["output"])
        full_text = build_prompt_template(instruction, answer)
        # Token变换
        token_result = self.tokenizer(
            full_text,
            truncation=True,
            max_length=self.seq_len,
            padding="max_length"
        )
        input_ids = token_result["input_ids"]
        attention_mask = token_result["attention_mask"]
        # 构造标签：自回归训练label=input_ids
        labels = input_ids.copy()
        # 关键变换：屏蔽prompt部分损失，仅计算回答部分loss
        prompt_only = build_prompt_template(instruction)
        prompt_tokens = self.tokenizer(prompt_only)["input_ids"]
        prompt_len = len(prompt_tokens)
        # prompt部分标签置-100，CrossEntropy自动忽略
        labels[:prompt_len] = [-100] * prompt_len
        return input_ids, attention_mask, labels
class SFTDataSet:
    def __init__(self, data_path):
        with open(data_path, "r", encoding="utf-8") as f:
            self.data = json.load(f)
    def __len__(self):
        return len(self.data)
    def __getitem__(self, idx):
        return self.data[idx]
def create_llm_dataloader(data_path, tokenizer_path, batch_size=2, seq_len=512):
    """构建完整数据流水线"""
    dataset = ds.GeneratorDataset(
        SFTDataSet(data_path),
        column_names=["instruction", "output"],
        shuffle=True,
        num_parallel_workers=4
    )
    transform_op = SFTDataTransform(tokenizer_path, seq_len)
    # 映射在线变换
    dataset = dataset.map(
        operations=transform_op,
        input_columns=["instruction", "output"],
        output_columns=["input_ids", "attention_mask", "labels"],
        num_parallel_workers=4
    )
    dataset = dataset.batch(batch_size, drop_remainder=True)
    return dataset

数据集 data.json 格式示例：

[
{"instruction":"什么是昇思MindSpore？","output":"昇思MindSpore是华为自研全场景深度学习框架。"}
]

五、训练主程序：加载变换流水线训练

# train_main.py
from env_init import init_ascend_env
from llm_dataset import create_llm_dataloader
import mindspore as ms
from mindformers import AutoModel
from mindspore.nn import AdamWeightDecay
init_ascend_env()
BATCH_SIZE = 2
SEQ_LEN = 512
# 构建带完整数据变换的数据集
train_ds = create_llm_dataloader(
    data_path="./train_data.json",
    tokenizer_path="./llm_model",
    batch_size=BATCH_SIZE,
    seq_len=SEQ_LEN
)
# 加载模型
model = AutoModel.from_pretrained("./llm_model")
loss_fn = ms.nn.CrossEntropyLoss(ignore_index=-100)
optimizer = AdamWeightDecay(model.trainable_params(), learning_rate=2e-4)
train_net = ms.nn.WithLossCell(model, loss_fn)
train_step = ms.nn.TrainOneStepCell(train_net, optimizer)
# 训练循环
epoch_num = 3
for epoch in range(epoch_num):
    loss_sum = 0
    for batch in train_ds.create_tuple_iterator():
        input_ids, attn_mask, labels = batch
        loss = train_step(input_ids, attn_mask, labels)
        loss_sum += loss.asnumpy()
    avg_loss = loss_sum / train_ds.get_dataset_size()
    print(f"Epoch {epoch}, Avg Loss: {avg_loss:.4f}")

六、启动脚本

# run_train.sh
#!/bin/bash
source /usr/local/Ascend/ascend-toolkit/latest/bin/set_env.sh
export DEVICE_ID=0
python3 train_main.py

七、数据变换核心优化要点

离线预处理与在线变换分离

大规模数据集优先离线清洗，避免训练时重复执行正则清洗，降低 CPU 开销；在线变换只保留 Tokenizer、掩码构造。

Loss Mask 变换至关重要

指令 Prompt 部分 label 设置为-100，使损失函数不统计 prompt 预测损失，是 SFT 标准变换；遗漏该变换会导致模型重复学习输入指令。

多进程 num_parallel_workers 调优

昇腾服务器 CPU 核心充足，合理设置并行数，防止 CPU 数据预处理拖慢 NPU 训练速度，造成算力空洞。

截断策略选择

支持头部截断、尾部截断；大模型推荐保留文本尾部，优先截断前文，保障回答完整。

性能优化手段

可使用ds.cache()缓存变换后数据；超大规模数据集使用 MindRecord 二进制格式，替代原始 JSON，减少 IO 开销。

八、总结

昇思 MindSpore 大模型数据变换分为三层：文本清洗变换、指令模板封装、Token 编码与标签掩码变换。依托 MindSpore Dataset 的 map 算子实现并行预处理，兼顾灵活性与性能。

完整的数据预处理流水线直接决定 SFT 微调效果。很多训练收敛异常、生成效果差的问题根源在于预处理缺陷：模板不统一、未屏蔽 prompt 损失、文本脏数据过多。

本文代码实现工业界标准指令微调预处理链路，可直接在昇腾 NPU 上运行，支持扩展多轮对话模板、多语种清洗、MindRecord 格式转换，适配各类开源大模型微调场景。