一、概述大模型训练质量高度依赖数据预处理流水线昇思 MindSpore 针对 LLM 训练提供两套预处理体系MindSpore Dataset 原生数据流变换、离线文本预处理 在线动态 Token 变换。预处理包含文本清洗、分段、模板封装、Tokenization、Padding、截断、掩码构造。与 CV 图像预处理不同大语言模型预处理重点解决指令模板统一、长文本截断策略、标签掩码Loss Mask、流水线加速、多进程数据加载。不合理的数据变换会造成训练泄露、上下文错乱、损失收敛困难。本文基于 MindSporeMindFormers搭建完整大模型 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( modecontext.GRAPH_MODE, device_targetAscend, device_id0 ) ms.set_auto_parallel_context(parallel_modems.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(rhttp[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_len512): 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, truncationTrue, max_lengthself.seq_len, paddingmax_length ) input_ids token_result[input_ids] attention_mask token_result[attention_mask] # 构造标签自回归训练labelinput_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部分标签置-100CrossEntropy自动忽略 labels[:prompt_len] [-100] * prompt_len return input_ids, attention_mask, labels class SFTDataSet: def __init__(self, data_path): with open(data_path, r, encodingutf-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_size2, seq_len512): 构建完整数据流水线 dataset ds.GeneratorDataset( SFTDataSet(data_path), column_names[instruction, output], shuffleTrue, num_parallel_workers4 ) transform_op SFTDataTransform(tokenizer_path, seq_len) # 映射在线变换 dataset dataset.map( operationstransform_op, input_columns[instruction, output], output_columns[input_ids, attention_mask, labels], num_parallel_workers4 ) dataset dataset.batch(batch_size, drop_remainderTrue) 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_sizeBATCH_SIZE, seq_lenSEQ_LEN ) # 加载模型 model AutoModel.from_pretrained(./llm_model) loss_fn ms.nn.CrossEntropyLoss(ignore_index-100) optimizer AdamWeightDecay(model.trainable_params(), learning_rate2e-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(fEpoch {epoch}, Avg Loss: {avg_loss:.4f})六、启动脚本# run_train.sh #!/bin/bash source /usr/local/Ascend/ascend-toolkit/latest/bin/set_env.sh export DEVICE_ID0 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 格式转换适配各类开源大模型微调场景。