一、简介

在上一篇文章从零手戳了一个 LLM 模型结构及 Pretrain、SFT 全流程，这有助于深入地理解了 LLM 的模型原理及训练细节。但是，在实际应用中，手戳实现的 LLM 训练存在以下问题：

• 工作量大：手写实现 LLM 结构，难以实时跟进最新模型的结构创新
• 训练效率低：从零实现的 LLM 训练无法较好地实现多卡分布式训练
• 兼容性差：和现有预训练 LLM 不兼容，无法使用预训练好的模型参数

因此，为更好地对接业界主流技术方案并实现工业级产出交付，这篇文章将系统梳理基于 Transformers 框架的 LLM 全流程开发实践。内容将以 Transformers 为核心基座，整合分布式训练框架 DeepSpeed ，重点阐述如何借助该框架，高效完成从模型 Pretrain（预训练） 到 SFT（监督微调） 的完整链路，旨在提供一套兼具理论深度与工程可行性的技术方案。

二、Transformers框架

2.1 Transformers 框架介绍

Transformers 是由 Hugging Face 开发的 NLP 框架，通过模块化设计实现了对 BERT、GPT、LLaMA、T5、ViT 等上百种主流模型架构的统一支持。开发者无需手动搭建 Transformer 的编码器、解码器或注意力层，只需调用 AutoModel.from_pretrained() 即可一键加载任意预训练模型。

其内置的 Trainer 训练器封装了 PyTorch DDP、DeepSpeed、Megatron-LM 等多种分布式策略，通过简单配置就能灵活组合数据并行、模型并行与流水线并行。配合 SavingPolicy 自动存档、LoggingCallback 实时监控等组件，整个训练流程实现了高度自动化的管理。

此外，HuggingFace 基于 Transformers 构建了全球最大的 AI 开源社区，开放了数亿个预训练模型权重、超过 25 万+个数据集以及配套的评估工具集。开发者可以直接在社区中获取预训练模型和标注数据集，基于开源资源快速构建个人模型。

进入大模型时代，模型架构的创新逐渐趋稳，实际业务中更多的精力投向了如何用好现成的预训练大模型——做 Post-Train 领域适应，或用 SFT 让模型学会遵循指令。由于模型体量动辄数十甚至上百 GB，单卡显存早已捉襟见肘，因此熟练集成 DeepSpeed 等分布式训练框架已成为 NLP 工程师的必备技能。

而 Transformers 正是居中调度的核心，将这些复杂的技术栈无缝串联起来，让开发者能够专注于数据与业务逻辑本身。如今，无论是学术顶会论文还是企业级 AI 产品，Transformers 都已稳居技术栈的核心位置，DeepSeek、Qwen 等最新开源模型更是发布当天便同步在社区开放权重与调用 Demo。

接下来，就以 Transformers 为技术基座，结合 DeepSpeed ，系统拆解如何完成 LLM 从预训练到监督微调的全流程落地实践。

三、预训练（Pretrain）

3.1 初始化 LLM

初始化LLM之前，为了方便，先把LLM下载到本地（不下载也行，执行代码时会自动下载）。使用如下命令进行下载：

huggingface-cli download --resume-download Qwen/Qwen3.5-2B --local-dir /root/models/

对于任意的预训练模型，其配置文件config.json中完整定义了模型的架构（如隐藏层大小、模型层数、注意力头数、维度等）。使用 Transformers 库来初始化 LLM 时有两种方式：

仅初始化模型架构	根据 config.json 中的模型参数，构建模型架构，所有参数随机初始化；适用于复用模型架构进行从头预训练的场景。
初始化模型架构并加载模型权重	在构建模型架构的基础上再加载模型预训练权重，使模型具备已学到的知识能力；适用于继续预训练、推理、下游任务微调等绝大多数实际应用场景。

3.1.1 仅初始化模型架构

模型下载完成后，使用 AutoConfig 加载配置文件，使用 AutoModelForCausalLM 初始化随机参数权重的模型：

from transformers import AutoConfig, AutoModelForCausalLM# 下载参数的本地路径model_path = "/root/models/qwen3.5-2b"# 加载配置config = AutoConfig.from_pretrained(model_path)# 使用配置初始化随机参数权重的模型model = AutoModelForCausalLM.from_config(config,trust_remote_code=True)

也可以对配置文件进行自定义，然后以同样的方式加载自定义配置文件的参数。

3.1.2 初始化模型架构并加载模型权重

一般情况下，很少从零初始化 LLM 进行预训练，较多的做法是加载一个预训练好的 LLM 权重，在自己的语料上进行继续训练。

使用 AutoModelForCausalLM 的 from_pretrained 方法即可构建模型架构并加载已有模型的权重：

from transformers import AutoModelForCausalLMmodel = AutoModelForCausalLM.from_pretrained(model_path,trust_remote_code=True)

由于 LLM 一般都是 CausalLM 架构，此处使用了 AutoModelForCausalLM 类进行加载。如果是用于分类任务训练，可使用 AutoModelForSequenceClassification 类来加载。

3.2 初始化 Tokenizer

直接使用对应模型的 tokenizer 参数即可：

# 加载一个预训练好的 tokenizerfrom transformers import AutoTokenizertokenizer = AutoTokenizer.from_pretrained(model_path)

加载好的 tokenizer 即可直接使用，来对任意文本进行分词处理。

3.3 预训练数据处理

使用 HuggingFace 的 datasets 库来加载预训练数据：

# 加载预训练数据from datasets import load_datasetds = load_dataset('json', data_files='～/Destop/code/data/mobvoi_seq_monkey_general_open_corpus.jsonl')

加载出来的 ds 是一个 DatasetDict 对象，加载的数据会默认保存在 train 键对应的值中。

提示： 由于数据集较大，加载可能会出现时间较长或内存不够的情况，建议前期测试时将预训练数据集拆分一部分出来进行测试。

可以通过 feature 属性查看数据集的特征（列名），后续需要保存列名，因为 tokenize 之后需要移除原先的文本：

# ds["train"][0]# 查看特征column_names = list(ds["train"].features)# columnes_name:["text"]

接着使用加载好的 tokenizer 对数据集进行处理，使用 map 函数进行批量处理：

# 对数据集进行 tokenizedef tokenize_function(examples):    # 使用预先加载的 tokenizer 进行分词    output = tokenizer([item for item in examples["text"]])    return output# 批量处理tokenized_datasets = ds.map(    tokenize_function,    batched=True,    num_proc=10,    remove_columns=column_names,    load_from_cache_file=True,    desc="Running tokenizer on dataset",)

处理完成后的数据集会包括 input_ids 和 attention_mask 两列，分别是文本 tokenize 之后的数值序列和注意力掩码（标识是否 padding）。map 方法会通过 remove_columns 参数将原先的 text 移除。

由于预训练一般为 CLM（因果语言模型）任务，一次性学习多个样本的序列语义不影响模型性能，且训练数据量大、训练时间长，对训练效率要求比较高。在预训练过程中，一般会把多个文本段拼接在一起，处理成统一长度的文本块，再对每个文本块进行训练。

# 预训练一般将文本拼接成固定长度的文本段from itertools import chain# 这里取块长为 2048block_size = 2048def group_texts(examples):    # 将文本段拼接起来    concatenated_examples = {k: list(chain(*examples[k])) for k in examples.keys()}    # 计算拼起来的整体长度    total_length = len(concatenated_examples[list(examples.keys())[0]])    # 如果长度太长，进行分块    if total_length >= block_size:        total_length = (total_length // block_size) * block_size    # 按 block_size 进行切分    result = {        k: [t[i : i + block_size] for i in range(0, total_length, block_size)]        for k, t in concatenated_examples.items()    }    # CLM 任务，labels 和 input 是相同的    result["labels"] = result["input_ids"].copy()    return result# 批量处理lm_datasets = tokenized_datasets.map(    group_texts,    batched=True,    num_proc=10,    load_from_cache_file=True,    desc=f"Grouping texts in chunks of {block_size}",    batch_size = 40000,)train_dataset = lm_datasets["train"]

处理得到的 train_dataset 就是一个可直接用于 CLM Pretrain 的预训练数据集了，其每个样本长度为 2048 个 token。

3.4 使用 Trainer 进行训练

Transformers 提供的 Trainer 类封装了模型的训练逻辑，且做了较好的效率优化、可视化等工作，可以高效、便捷地完成 LLM 的训练。

3.4.1 配置训练参数

使用 TrainingArguments 类来实例化一个参数对象：

# 配置训练参数from transformers import TrainingArgumentstraining_args = TrainingArguments(    output_dir="output",# 训练参数输出路径    per_device_train_batch_size=4,# 训练的 batch_size    gradient_accumulation_steps=4,# 梯度累计步数，实际 bs = 设置的 bs * 累计步数    logging_steps=10,# 打印 loss 的步数间隔    num_train_epochs=1,# 训练的 epoch 数    save_steps=100, # 保存模型参数的步数间隔    learning_rate=1e-4,# 学习率    gradient_checkpointing=True# 开启梯度检查点)

3.4.2 实例化 Trainer

基于初始化的 model、tokenizer 和 training_args，并传入处理好的训练数据集，实例化一个 trainer 对象：

from transformers import Trainer, default_data_collatorfrom torchdata.datapipes.iter import IterableWrapper# 训练器trainer = Trainer(    model=model,    args=training_args,    train_dataset= train_dataset,    eval_dataset= None,    tokenizer=tokenizer,    # 默认为 MLM 的 collator，使用 CLM 的 collater    data_collator=default_data_collator)

3.4.3 启动训练

使用 train 方法来启动训练：

trainer.train()

通过以上代码即可按照配置好的训练超参进行训练，并把训练好的模型保存到对应的文件夹。

3.5 使用 DeepSpeed 实现分布式训练

鉴于大规模预训练任务耗时极长且稳定性要求高，强烈不建议直接执行Python脚本来训练，以免意外中断。此外，需要使用多卡进行分布式训练，否则训练时间太长。这里介绍使用 DeepSpeed 框架实现分布式训练，从而完成业界可用的 LLM Pretrain。需要注意的是，DeepSpeed不支持M芯片的Mac设备。

大规模长时间训练一般使用 bash 脚本设定超参，再启动写好的 Python 脚本实现训练。为了接收bash脚本的超参、设置中断继续等，需要完善一下上面代码的逻辑。

3.5.1 Python脚本完善

3.5.1.1 超参定义

首先定义几个超参的类型，用于处理 sh 脚本中设定的超参值。由于 Transformers 本身有 TrainingArguments 类，只需定义其中未包含的超参即可：

• ModelArguments：模型相关的超参（模型路径、配置路径、数据类型等）
• DataTrainingArguments：数据相关的超参（训练数据路径、文本块长度等）

# 超参类@dataclassclass ModelArguments:    """    关于模型的参数    """    model_name_or_path: Optional[str] = field(        default=None,        metadata={            "help": (                "后训练使用，为预训练模型参数地址"            )        },    )    config_name: Optional[str] = field(        default=None, metadata={"help": "预训练使用，Config 文件地址"}    )    tokenizer_name: Optional[str] = field(        default=None, metadata={"help": "预训练 Tokenizer 地址"}    )    torch_dtype: Optional[str] = field(        default=None,        metadata={            "help": (                "模型训练使用的数据类型，推荐 bfloat16"            ),            "choices": ["auto", "bfloat16", "float16", "float32"],        },    )@dataclassclass DataTrainingArguments:    """    关于训练的参数    """    train_files: Optional[List[str]]  = field(default=None, metadata={"help": "训练数据路径"})    block_size: Optional[int] = field(        default=None,        metadata={            "help": (                "设置的文本块长度"            )        },    )    preprocessing_num_workers: Optional[int] = field(        default=None,        metadata={"help": "预处理使用线程数."},    )

3.5.1.2 定义Checkpoint 中断恢复

在大规模训练中，发生中断往往难以避免，训练一般会固定间隔保存 checkpoint，中断之后基于最近的 checkpoint 恢复训练即可：

# 检查 checkpointlast_checkpoint = Noneif os.path.isdir(training_args.output_dir):    # 使用 transformers 自带的 get_last_checkpoint 自动检测    last_checkpoint = get_last_checkpoint(training_args.output_dir)    if last_checkpoint isNoneand len(os.listdir(training_args.output_dir)) > 0:        raise ValueError(            f"输出路径 ({training_args.output_dir}) 非空 "        )    elif last_checkpoint isnotNoneand training_args.resume_from_checkpoint isNone:        logger.info(            f"从 {last_checkpoint}恢复训练"        )

3.5.1.3 初始化模型与训练

将从零初始化和基于已有预训练模型权重初始化封装在一起：

# 初始化模型if model_args.config_name isnotNone:    config = AutoConfig.from_pretrained(model_args.config_name)    logger.warning("你正在从零初始化一个模型")    logger.info(f"模型参数配置地址：{model_args.config_name}")    logger.info(f"模型参数：{config}")    model = AutoModelForCausalLM.from_config(config,trust_remote_code=True)    n_params = sum({p.data_ptr(): p.numel() for p in model.parameters()}.values())    logger.info(f"预训练一个新模型 - Total size={n_params/2**20:.2f}M params")elif model_args.model_name_or_path isnotNone:    logger.warning("你正在初始化一个预训练模型")    logger.info(f"模型参数地址：{model_args.model_name_or_path}")    model = AutoModelForCausalLM.from_pretrained(model_args.model_name_or_path,trust_remote_code=True)    n_params = sum({p.data_ptr(): p.numel() for p in model.parameters()}.values())    logger.info(f"继承一个预训练模型 - Total size={n_params/2**20:.2f}M params")else:    logger.error("config_name 和 model_name_or_path 不能均为空")    raise ValueError("config_name 和 model_name_or_path 不能均为空")

使用 Trainer 进行训练，并支持从 checkpoint 恢复：

logger.info("初始化 Trainer")trainer = Trainer(    model=model,    args=training_args,    train_dataset= IterableWrapper(train_dataset),    tokenizer=tokenizer,    data_collator=default_data_collator)# 从 checkpoint 加载checkpoint = Noneif training_args.resume_from_checkpoint isnotNone:    checkpoint = training_args.resume_from_checkpointelif last_checkpoint isnotNone:        checkpoint = last_checkpointlogger.info("开始训练")train_result = trainer.train(resume_from_checkpoint=checkpoint)trainer.save_model()

3.5.1.4 训练监控

由于在大规模训练中监测训练进度、loss 下降趋势尤为重要，可以使用 SwanLab 作为训练检测的工具：

# 初始化 SwanLabswanlab.init(project="My-LLM", experiment_name="pretrain")

启动训练后，终端会输出 SwanLab 监测的 URL，点击即可观察训练进度。

3.5.1.5 完整代码

完整的基于 Transformers 的训练代码如下：

# pretrain.pyimport loggingimport mathimport osimport sysfrom dataclasses import dataclass, fieldfrom torchdata.datapipes.iter import IterableWrapperfrom itertools import chainimport deepspeedfrom typing import Optional,Listimport datasetsimport pandas as pdimport torchfrom datasets import load_datasetimport transformersfrom transformers import (    AutoConfig,    AutoModelForCausalLM,    AutoTokenizer,    HfArgumentParser,    Trainer,    TrainingArguments,    default_data_collator,    set_seed,)import datetimefrom transformers.testing_utils import CaptureLoggerfrom transformers.trainer_utils import get_last_checkpointimport swanlablogger = logging.getLogger(__name__)# 超参类@dataclassclass ModelArguments:    """    关于模型的参数    """    model_name_or_path: Optional[str] = field(        default=None,        metadata={            "help": (                "后训练使用，为预训练模型参数地址"            )        },    )    config_name: Optional[str] = field(        default=None, metadata={"help": "预训练使用，Config 文件地址"}    )    tokenizer_name: Optional[str] = field(        default=None, metadata={"help": "预训练 Tokenizer 地址"}    )    torch_dtype: Optional[str] = field(        default=None,        metadata={            "help": (                "模型训练使用的数据类型，推荐 bfloat16"            ),            "choices": ["auto", "bfloat16", "float16", "float32"],        },    )@dataclassclass DataTrainingArguments:    """    关于训练的参数    """    train_files: Optional[List[str]]  = field(default=None, metadata={"help": "训练数据路径"})    block_size: Optional[int] = field(        default=None,        metadata={            "help": (                "设置的文本块长度"            )        },    )    preprocessing_num_workers: Optional[int] = field(        default=None,        metadata={"help": "预处理使用线程数."},    )def main():    # 加载脚本参数    parser = HfArgumentParser((ModelArguments, DataTrainingArguments, TrainingArguments))    model_args, data_args, training_args = parser.parse_args_into_dataclasses()    # 初始化 SwanLab    swanlab.init(project="My-LLM", experiment_name="pretrain")    # 设置日志    logging.basicConfig(        format="%(asctime)s - %(levelname)s - %(name)s - %(message)s",        datefmt="%m/%d/%Y %H:%M:%S",        handlers=[logging.StreamHandler(sys.stdout)],    )    # 将日志级别设置为 INFO    transformers.utils.logging.set_verbosity_info()    log_level = training_args.get_process_log_level()    logger.setLevel(log_level)    datasets.utils.logging.set_verbosity(log_level)    transformers.utils.logging.set_verbosity(log_level)    transformers.utils.logging.enable_default_handler()    transformers.utils.logging.enable_explicit_format()    # 训练整体情况记录    logger.warning(        f"Process rank: {training_args.local_rank}, device: {training_args.device}, n_gpu: {training_args.n_gpu}"        + f"distributed training: {bool(training_args.local_rank != -1)}, 16-bits training: {training_args.fp16}"    )    logger.info(f"Training/evaluation parameters {training_args}")    # 检查 checkpoint    last_checkpoint = None    if os.path.isdir(training_args.output_dir):        last_checkpoint = get_last_checkpoint(training_args.output_dir)        if last_checkpoint isNoneand len(os.listdir(training_args.output_dir)) > 0:            raise ValueError(                f"输出路径 ({training_args.output_dir}) 非空 "            )        elif last_checkpoint isnotNoneand training_args.resume_from_checkpoint isNone:            logger.info(                f"从 {last_checkpoint}恢复训练"            )    # 设置随机数种子.    set_seed(training_args.seed)    # 初始化模型    if model_args.config_name isnotNone:        config = AutoConfig.from_pretrained(model_args.config_name)        logger.warning("你正在从零初始化一个模型")        logger.info(f"模型参数配置地址：{model_args.config_name}")        logger.info(f"模型参数：{config}")        model = AutoModelForCausalLM.from_config(config,trust_remote_code=True)        n_params = sum({p.data_ptr(): p.numel() for p in model.parameters()}.values())        logger.info(f"预训练一个新模型 - Total size={n_params/2**20:.2f}M params")    elif model_args.model_name_or_path isnotNone:        logger.warning("你正在初始化一个预训练模型")        logger.info(f"模型参数地址：{model_args.model_name_or_path}")        model = AutoModelForCausalLM.from_pretrained(model_args.model_name_or_path,trust_remote_code=True)        n_params = sum({p.data_ptr(): p.numel() for p in model.parameters()}.values())        logger.info(f"继承一个预训练模型 - Total size={n_params/2**20:.2f}M params")    else:        logger.error("config_name 和 model_name_or_path 不能均为空")        raise ValueError("config_name 和 model_name_or_path 不能均为空")    # 初始化 Tokenizer    tokenizer = AutoTokenizer.from_pretrained(model_args.tokenizer_name)    logger.info("完成 tokenzier 加载")    logger.info(f"tokenzier 配置地址：{model_args.tokenizer_name}")    # 加载预训练数据    ds = load_dataset('json', data_files=data_args.train_files)    logger.info("完成训练集加载")    logger.info(f"训练集地址：{data_args.train_files}")    logger.info(f'训练文件总数:{len(ds["train"])}')    # logger.info(f"训练集采样：{ds["train"][0]}")    # 文本 tokenize    column_names = list(ds["train"].features)    logger.info('训练集特征：', column_names)    text_column_name = "text"if"text"in column_names else column_names[0]    # tokenize 函数    def tokenize_function(examples):        output = tokenizer([item for item in examples[text_column_name]])        return output    # 仅主进程进行数据预处理    with training_args.main_process_first(desc="dataset map tokenization"):        tokenized_datasets = ds.map(            tokenize_function,            batched=True,            num_proc=data_args.preprocessing_num_workers,            remove_columns=column_names,            load_from_cache_file=True,            desc="Running tokenizer on dataset"        )    # 文本切块    if data_args.block_size isNone:        block_size = tokenizer.model_max_length        if block_size > 1024:            logger.warning(                "tokenizer 支持大于 1K 的上下文长度，默认设置为 1K"            )            block_size = 1024    else:        if data_args.block_size > tokenizer.model_max_length:            logger.warning(                f"设定的块长为 ({data_args.block_size}) ，大于模型的上下文长度"                f"将块长设置为模型上下文长度：{tokenizer.model_max_length}."            )        block_size = min(data_args.block_size, tokenizer.model_max_length)    def group_texts(examples):        # 将文本段拼接起来        concatenated_examples = {k: list(chain(*examples[k])) for k in examples.keys()}        # 计算拼起来的整体长度        total_length = len(concatenated_examples[list(examples.keys())[0]])        # 如果长度太长，进行分块        if total_length >= block_size:            total_length = (total_length // block_size) * block_size        result = {            k: [t[i : i + block_size] for i in range(0, total_length, block_size)]            for k, t in concatenated_examples.items()        }        result["labels"] = result["input_ids"].copy()        return result    with training_args.main_process_first(desc="文本分块"):        lm_datasets = tokenized_datasets.map(            group_texts,            batched=True,            num_proc=data_args.preprocessing_num_workers,            load_from_cache_file=True,            desc=f"文本分块到{block_size}",            batch_size = 40000,        )        logger.info("完成数据预处理")        train_dataset = lm_datasets["train"]    logger.info("初始化 Trainer")    trainer = Trainer(        model=model,        args=training_args,        train_dataset= train_dataset,        tokenizer=tokenizer,        data_collator=default_data_collator    )    # 从 checkpoint 加载    checkpoint = None    if training_args.resume_from_checkpoint isnotNone:        checkpoint = training_args.resume_from_checkpoint    elif last_checkpoint isnotNone:            checkpoint = last_checkpoint    logger.info("开始训练")    train_result = trainer.train(resume_from_checkpoint=checkpoint)    trainer.save_model()if __name__ == "__main__":    main()

3.5.2 DeepSpeed 启动

使用一个 sh 脚本定义超参数的值，并通过 DeepSpeed 启动训练：

# 设置可见显卡CUDA_VISIBLE_DEVICES=0,1deepspeed pretrain.py \    --config_name /root/models/qwen3.5-2b \    --tokenizer_name /root/models/qwen3.5-2b \    --train_files /root/Desktop/code/data/mobvoi_seq_monkey_general_open_corpus_small.jsonl \    --per_device_train_batch_size 16 \    --gradient_accumulation_steps 4 \    --do_train \    --output_dir /root/Desktop/code/output/pretrain \    --evaluation_strategy  no \    --learning_rate 1e-4 \    --num_train_epochs 1 \    --warmup_steps 200 \    --logging_dir /root/Desktop/code/output/pretrain \    --logging_strategy steps \    --logging_steps 5 \    --save_strategy steps \    --save_steps 100 \    --preprocessing_num_workers 10 \    --save_total_limit 1 \    --seed 12 \    --block_size 2048 \    --bf16 \    --gradient_checkpointing \    --deepspeed ./ds_config_zero2.json \    --report_to swanlab    # --resume_from_checkpoint ${output_model}/checkpoint-20400 \

在安装了 DeepSpeed 第三方库后，可以直接通过 deepspeed 命令来启动多卡训练。此处使用 ZeRO-2 进行训练。ZeRO（Zero Redundancy Optimizer）是 DeepSpeed 的核心优化技术，通过分片策略减少显存冗余占用：

• ZeRO-1：对优化器状态进行分片，每张卡只保存部分优化器状态
• ZeRO-2：在 ZeRO-1 基础上，对梯度也进行分片，进一步降低显存占用
• ZeRO-3：对模型参数也进行分片，显存占用最低但通信开销最大

ZeRO-2 在显存节省和训练效率之间取得了较好的平衡，是中等规模训练的常用选择。加载 ds_config_zero2.json 作为 DeepSpeed 的配置参数：

{    "fp16": {        "enabled": "auto",        "loss_scale": 0,        "loss_scale_window": 1000,        "initial_scale_power": 16,        "hysteresis": 2,        "min_loss_scale": 1    },    "bf16": {        "enabled": "auto"    },    "optimizer": {        "type": "AdamW",        "params": {            "lr": "auto",            "betas": "auto",            "eps": "auto",            "weight_decay": "auto"        }    },    "scheduler": {        "type": "WarmupLR",        "params": {            "warmup_min_lr": "auto",            "warmup_max_lr": "auto",            "warmup_num_steps": "auto"        }    },    "zero_optimization": {        "stage": 2,        "offload_optimizer": {            "device": "none",            "pin_memory": true        },        "allgather_partitions": true,        "allgather_bucket_size": 2e8,        "overlap_comm": true,        "reduce_scatter": true,        "reduce_bucket_size": 2e8,        "contiguous_gradients": true    },    "gradient_accumulation_steps": "auto",    "gradient_clipping": "auto",    "steps_per_print": 100,    "train_batch_size": "auto",    "train_micro_batch_size_per_gpu": "auto",    "wall_clock_breakdown": false}

最后，在终端运行该 pretrain.sh 脚本即可开始训练。

四、有监督微调（SFT）

4.1 Pretrain 与 SFT 的核心差异

对 LLM 进行预训练和进行有监督微调的核心差异在于：

因此，相较于 Pretrain 代码，SFT 部分仅需要修改数据处理环节，实现对指令对数据转化为训练样本的构建，其余部分和 Pretrain 是完全一致的实现逻辑。

4.2 SFT微调数据处理

4.2.1 Chat Template

在 SFT 过程中，会定义一个 Chat Template，即表示了如何将对话数据转化为一个模型可以建模拟合的文本序列。当我们使用已做过 SFT 微调的模型进行下游任务微调时，一般需要查看该模型的 Chat Template 并进行适配，以免损伤其在 SFT 中学到的指令遵循能力。

此处沿用 Qwen 的 Chat Template。

4.2.2 定义特殊 Token

特殊 token 在模型进行拟合中有特殊的作用，包括文本序列开始（BOS）、文本序列结束（EOS）、换行符等。定义特殊 token，有助于避免模型在拟合过程中的语义混淆：

# 不同的 tokenizer 需要特别定义# BOSim_start = tokenizer("<|im_start|>").input_ids# EOSim_end = tokenizer("<|im_end|>").input_ids# PADIGNORE_TOKEN_ID = tokenizer.pad_token_id# 换行符nl_tokens = tokenizer('\n').input_ids# 角色标识符_system = tokenizer('system').input_ids + nl_tokens_user = tokenizer('human').input_ids + nl_tokens_assistant = tokenizer('assistant').input_ids + nl_tokens

Qwen 系列的 Chat Template 一般有三个对话角色：

• System：系统提示词，负责激活模型的能力，默认为 “You are a helpful assistant.”
• User（Human）：用户给出的提示词
• Assistant：LLM 给出的回复，也就是模型在 SFT 过程中需要拟合的文本

4.2.3 拼接多轮对话

由于数据集是多轮对话数据集，需要对多轮对话进行拼接处理，将多轮对话拼接到一个文本序列中：

# 拼接多轮对话input_ids, targets = [], []# 多个样本for i in tqdm(range(len(sources))):    # source 为一个多轮对话样本    source = sources[i]    # 从 user 开始    if source[0]["from"] != "human":        source = source[1:]    # 分别是输入和输出    input_id, target = [], []    # system: 【BOS】system\nYou are a helpful assistant.【EOS】\n    system = im_start + _system + tokenizer(system_message).input_ids + im_end + nl_tokens    input_id += system    # system 不需要拟合    target += im_start + [IGNORE_TOKEN_ID] * (len(system)-3) + im_end + nl_tokens    assert len(input_id) == len(target)    # 依次拼接    for j, sentence in enumerate(source):        # sentence 为一轮对话        role = roles[sentence["from"]]        # user：<|im_start|>human\ninstruction【EOS】\n        # assistant：<|im_start|>assistant\nresponse【EOS】\n        _input_id = tokenizer(role).input_ids + nl_tokens + \            tokenizer(sentence["value"]).input_ids + im_end + nl_tokens        input_id += _input_id        if role == '<|im_start|>human':            # user 不需要拟合            _target = im_start + [IGNORE_TOKEN_ID] * (len(_input_id)-3) + im_end + nl_tokens        elif role == '<|im_start|>assistant':            # assistant 需要拟合            _target = im_start + [IGNORE_TOKEN_ID] * len(tokenizer(role).input_ids) + \                _input_id[len(tokenizer(role).input_ids)+1:-2] + im_end + nl_tokens        else:            print(role)            raise NotImplementedError        target += _target    assert len(input_id) == len(target)    # 最后进行 PAD    input_id += [tokenizer.pad_token_id] * (max_len - len(input_id))    target += [IGNORE_TOKEN_ID] * (max_len - len(target))    input_ids.append(input_id[:max_len])    targets.append(target[:max_len])

上述代码的核心逻辑：

• System 部分：不需要拟合，targets 中使用 IGNORE_TOKEN_ID 进行遮蔽
• User 部分：不需要拟合，targets 中使用 IGNORE_TOKEN_ID 进行遮蔽
• Assistant 部分：需要拟合，targets 中为文本原文，需要计算 loss

上述代码中 IGNORE_TOKEN_ID = tokenizer.pad_token_id，在 Qwen 系列模型中，pad_token_id 恰好为 -100，这也是 PyTorch CrossEntropyLoss 默认的 ignore_index 值。当 targets 中某个位置的值为 -100 时，该位置不参与 loss 计算，从而实现了对 System 和 User 部分的遮蔽。如果使用的模型 pad_token_id 不是 -100，建议显式设置 IGNORE_TOKEN_ID = -100 以确保行为一致。

4.2.4 构建 Dataset

完成拼接后，将 tokenize 后的数值序列转化为 Torch.tensor，自定义一个 Dataset 类来封装数据处理逻辑：

input_ids = torch.tensor(input_ids)targets = torch.tensor(targets)return dict(    input_ids=input_ids,    labels=targets,    attention_mask=input_ids.ne(tokenizer.pad_token_id),)

该类继承自 Torch 的 Dataset 类，可以直接在 Trainer 中使用。

4.3 训练与启动

完成数据处理后，基于 Pretrain 脚本，修改数据处理逻辑即可，后续模型训练等几乎完全一致。

完整代码如下：

# sft.pyimport loggingimport mathimport osimport sysfrom dataclasses import dataclass, fieldfrom torchdata.datapipes.iter import IterableWrapperfrom itertools import chainimport deepspeedfrom typing import Optional,List,Dictfrom torch.utils.data import Datasetimport jsonimport datasetsimport pandas as pdimport torchfrom datasets import load_datasetimport transformersfrom transformers import (    AutoConfig,    AutoModelForCausalLM,    AutoTokenizer,    HfArgumentParser,    Trainer,    TrainingArguments,    default_data_collator,    set_seed,)import datetimefrom transformers.testing_utils import CaptureLoggerfrom transformers.trainer_utils import get_last_checkpointimport swanlabfrom tqdm import tqdmlogger = logging.getLogger(__name__)# 超参类@dataclassclass ModelArguments:    """    关于模型的参数    """    model_name_or_path: Optional[str] = field(        default=None,        metadata={            "help": (                "预训练模型参数地址"            )        },    )    torch_dtype: Optional[str] = field(        default=None,        metadata={            "help": (                "模型训练使用的数据类型，推荐 bfloat16"            ),            "choices": ["auto", "bfloat16", "float16", "float32"],        },    )@dataclassclass DataTrainingArguments:    """    关于训练的参数    """    train_files: Optional[str]  = field(default=None, metadata={"help": "训练数据路径"})    block_size: Optional[int] = field(        default=None,        metadata={            "help": (                "最大文本块长度"            )        },    )# 指令文本处理# 参考：https://github.com/QwenLM/Qwen/blob/main/finetune.pydef preprocess(sources, tokenizer, max_len, system_message: str = "You are a helpful assistant."):    # prompt 模板    roles = {"human": "<|im_start|>human", "assistant": "<|im_start|>assistant"}    # 不同的 tokenizer 需要特别定义    # BOS    im_start = tokenizer("<|im_start|>").input_ids    # EOS    im_end = tokenizer("<|im_end|>").input_ids    # PAD    IGNORE_TOKEN_ID = tokenizer.pad_token_id    # 换行符    nl_tokens = tokenizer('\n').input_ids    # 角色标识符    _system = tokenizer('system').input_ids + nl_tokens    _user = tokenizer('human').input_ids + nl_tokens    _assistant = tokenizer('assistant').input_ids + nl_tokens    # 拼接多轮对话    input_ids, targets = [], []    for i in tqdm(range(len(sources))):        source = sources[i]        # 从 user 开始        if source[0]["from"] != "human":            source = source[1:]        # 分别是输入和输出        input_id, target = [], []        # system: 【BOS】system\nYou are a helpful assistant.【EOS】\n        system = im_start + _system + tokenizer(system_message).input_ids + im_end + nl_tokens        input_id += system        # system 不需要拟合        target += im_start + [IGNORE_TOKEN_ID] * (len(system)-3) + im_end + nl_tokens        assert len(input_id) == len(target)        # 依次拼接        for j, sentence in enumerate(source):            role = roles[sentence["from"]]            # user：<|im_start|>human\ninstruction【EOS】\n            # assistant：<|im_start|>assistant\nresponse【EOS】\n            _input_id = tokenizer(role).input_ids + nl_tokens + \                tokenizer(sentence["value"]).input_ids + im_end + nl_tokens            input_id += _input_id            if role == '<|im_start|>human':                # user 不需要拟合                _target = im_start + [IGNORE_TOKEN_ID] * (len(_input_id)-3) + im_end + nl_tokens            elif role == '<|im_start|>assistant':                # assistant 需要拟合                _target = im_start + [IGNORE_TOKEN_ID] * len(tokenizer(role).input_ids) + \                    _input_id[len(tokenizer(role).input_ids)+1:-2] + im_end + nl_tokens            else:                print(role)                raise NotImplementedError            target += _target        assert len(input_id) == len(target)        # 最后进行 PAD        input_id += [tokenizer.pad_token_id] * (max_len - len(input_id))        target += [IGNORE_TOKEN_ID] * (max_len - len(target))        input_ids.append(input_id[:max_len])        targets.append(target[:max_len])    # print(input_ids)    input_ids = torch.tensor(input_ids)    targets = torch.tensor(targets)    return dict(        input_ids=input_ids,        labels=targets,        attention_mask=input_ids.ne(tokenizer.pad_token_id),    )# 自定义一个 Datasetfrom typing import Dictclass SupervisedDataset(Dataset):    def __init__(self, raw_data, tokenizer, max_len: int):        super(SupervisedDataset, self).__init__()        # 加载并预处理数据        sources = [example["conversations"] for example in raw_data]        data_dict = preprocess(sources, tokenizer, max_len)        self.input_ids = data_dict["input_ids"]        self.labels = data_dict["labels"]        self.attention_mask = data_dict["attention_mask"]    def __len__(self):        return len(self.input_ids)    def __getitem__(self, i) -> Dict[str, torch.Tensor]:        return dict(            input_ids=self.input_ids[i],            labels=self.labels[i],            attention_mask=self.attention_mask[i],        )def main():    # 加载脚本参数    parser = HfArgumentParser((ModelArguments, DataTrainingArguments, TrainingArguments))    model_args, data_args, training_args = parser.parse_args_into_dataclasses()    # 初始化 SwanLab    swanlab.init(project="My-LLM", experiment_name="SFT")    # 设置日志    logging.basicConfig(        format="%(asctime)s - %(levelname)s - %(name)s - %(message)s",        datefmt="%m/%d/%Y %H:%M:%S",        handlers=[logging.StreamHandler(sys.stdout)],    )    # 将日志级别设置为 INFO    transformers.utils.logging.set_verbosity_info()    log_level = training_args.get_process_log_level()    logger.setLevel(log_level)    datasets.utils.logging.set_verbosity(log_level)    transformers.utils.logging.set_verbosity(log_level)    transformers.utils.logging.enable_default_handler()    transformers.utils.logging.enable_explicit_format()    # 训练整体情况记录    logger.warning(        f"Process rank: {training_args.local_rank}, device: {training_args.device}, n_gpu: {training_args.n_gpu}"        + f"distributed training: {bool(training_args.local_rank != -1)}, 16-bits training: {training_args.fp16}"    )    logger.info(f"Training/evaluation parameters {training_args}")    # 检查 checkpoint    last_checkpoint = None    if os.path.isdir(training_args.output_dir):        last_checkpoint = get_last_checkpoint(training_args.output_dir)        if last_checkpoint isNoneand len(os.listdir(training_args.output_dir)) > 0:            raise ValueError(                f"输出路径 ({training_args.output_dir}) 非空 "            )        elif last_checkpoint isnotNoneand training_args.resume_from_checkpoint isNone:            logger.info(                f"从 {last_checkpoint}恢复训练"            )    # 设置随机数种子.    set_seed(training_args.seed)    # 初始化模型    logger.warning("加载预训练模型")    logger.info(f"模型参数地址：{model_args.model_name_or_path}")    model = AutoModelForCausalLM.from_pretrained(model_args.model_name_or_path,trust_remote_code=True)    n_params = sum({p.data_ptr(): p.numel() for p in model.parameters()}.values())    logger.info(f"继承一个预训练模型 - Total size={n_params/2**20:.2f}M params")    # 初始化 Tokenizer    tokenizer = AutoTokenizer.from_pretrained(model_args.model_name_or_path)    logger.info("完成 tokenzier 加载")    # 加载微调数据    with open(data_args.train_files) as f:        lst = [json.loads(line) for line in f.readlines()[:10000]]    logger.info("完成训练集加载")    logger.info(f"训练集地址：{data_args.train_files}")    logger.info(f'训练样本总数:{len(lst)}')    # logger.info(f"训练集采样：{ds["train"][0]}")    train_dataset = SupervisedDataset(lst, tokenizer=tokenizer, max_len=2048)    logger.info("初始化 Trainer")    trainer = Trainer(        model=model,        args=training_args,        train_dataset= train_dataset,        tokenizer=tokenizer    )    # 从 checkpoint 加载    checkpoint = None    if training_args.resume_from_checkpoint isnotNone:        checkpoint = training_args.resume_from_checkpoint    elif last_checkpoint isnotNone:            checkpoint = last_checkpoint    logger.info("开始训练")    train_result = trainer.train(resume_from_checkpoint=checkpoint)    trainer.save_model()if __name__ == "__main__":    main()

使用DeepSpeed的sh脚本如下：

# sft.shCUDA_VISIBLE_DEVICES=0,1deepspeed sft.py \    --model_name_or_path /root/models/qwen3.5-2b \    --train_files /root/Desktop/code/sft_data/BelleGroup/train_3.5M_CN.json \    --per_device_train_batch_size 16 \    --gradient_accumulation_steps 4 \    --do_train \    --output_dir /root/Desktop/code/output/sft \    --evaluation_strategy  no \    --learning_rate 1e-4 \    --num_train_epochs 3 \    --warmup_steps 200 \    --logging_dir /root/Desktop/code/output/sft/logs \    --logging_strategy steps \    --logging_steps 5 \    --save_strategy steps \    --save_steps 100 \    --save_total_limit 1 \    --seed 12 \    --bf16 \    --gradient_checkpointing \    --deepspeed ./ds_config_zero2.json \    --report_to swanlab    # --resume_from_checkpoint ${output_model}/checkpoint-20400 \

使用以上脚本启动即可开始 SFT 监督微调。

说明：与 Pretrain 不同，SFT 不需要设置 block_size 参数进行文本拼接分块。SFT 的每条指令对数据本身就是独立的样本，序列长度通过 SupervisedDataset 中的 max_len 参数控制（默认 2048），超出长度的样本会被截断，不足的会进行 padding。

五、总结

这篇文章以Transformers框架为核心，结合DeepSpeed分布式训练技术，系统梳理了LLM从预训练到监督微调的完整工程化落地流程。预训练阶段重点解决了大规模语料的分块处理、Trainer训练封装及DeepSpeed ZeRO-2多卡并行训练，实现了工业级的持续预训练能力；监督微调阶段则聚焦于指令数据的Chat Template构建与Loss Masking机制，确保模型仅对回复部分计算损失，从而有效习得指令遵循能力。整套方案已在Qwen系列模型上完成验证，代码可直接复用，提供了一条从模型加载、数据处理到分布式训练的全链路实践路径，真正打通了学术原型到业务投产的最后一公里。

学AI大模型的正确顺序，千万不要搞错了

🤔2026年AI风口已来！各行各业的AI渗透肉眼可见，超多公司要么转型做AI相关产品，要么高薪挖AI技术人才，机遇直接摆在眼前！

有往AI方向发展，或者本身有后端编程基础的朋友，直接冲AI大模型应用开发转岗超合适！

就算暂时不打算转岗，了解大模型、RAG、Prompt、Agent这些热门概念，能上手做简单项目，也绝对是求职加分王🔋

📝给大家整理了超全最新的AI大模型应用开发学习清单和资料，手把手帮你快速入门！👇👇

学习路线:

✅大模型基础认知—大模型核心原理、发展历程、主流模型（GPT、文心一言等）特点解析
✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑
✅开发基础能力—Python进阶、API接口调用、大模型开发框架（LangChain等）实操
✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用
✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代
✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经

以上6大模块，看似清晰好上手，实则每个部分都有扎实的核心内容需要吃透！

我把大模型的学习全流程已经整理📚好了！抓住AI时代风口，轻松解锁职业新可能，希望大家都能把握机遇，实现薪资/职业跃迁～

这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】