【第三章】大模型预训练全解析：定义、数据处理、流程及多阶段训练逻辑
接上章《【第一章】大模型预训练全解析：定义、数据处理、流程及多阶段训练逻辑》和《【第二章】大模型预训练全解析：定义、数据处理、流程及多阶段训练逻辑》
5)监督微调数据处理（SFT）
def preprocess_supervised_dataset(examples: Dict[str, List[Any]]) -> Dict[str, Any]:model_inputs = {"input_ids": [], "attention_mask": [], "labels": []}max_length = data_args.max_source_length + data_args.max_target_lengthfor query, response, history, system in construct_example(examples):input_ids, labels = [], []for source_ids, target_ids in template.encode_multiturn(...):if len(input_ids) + len(source_ids) + len(target_ids) > max_length:breakinput_ids += source_ids + target_idslabels += [IGNORE_INDEX] * len(source_ids) + target_idsmodel_inputs["input_ids"].append(input_ids)model_inputs["attention_mask"].append([1] * len(input_ids))model_inputs["labels"].append(labels)return model_inputs示例输入与配置
假设输入的examples包含1个对话样本，配置如下：
examples = {"prompt": ["今天天气如何？"],"response": ["晴天，适合户外活动。"]
}# 配置参数
data_args.max_source_length = 10 # 源文本最大长度
data_args.max_target_length = 10 # 目标文本最大长度
max_length = 20 # 总最大长度# 模板编码结果（简化示例）
# 假设template.encode_multiturn(...)返回：
# [
# (source_ids=[101, 102], target_ids=[201, 202, 203]), # 表示"今天天气如何？" → "晴天，适合户外活动。"
# ]逐行代码执行过程与输出
初始化输出结构
model_inputs = {"input_ids": [], "attention_mask": [], "labels": []}- 初始状态：
model_inputs = {"input_ids": [],"attention_mask": [],"labels": []
}遍历construct_example生成的样本
for query, response, history, system in construct_example(examples):# query = "今天天气如何？"# response = "晴天，适合户外活动。"input_ids, labels = [], []- 初始化样本的input_ids和labels：
input_ids = []
labels = []处理多轮对话（本例简化为单轮）
for source_ids, target_ids in template.encode_multiturn(...):# source_ids = [101, 102] # 对应"今天天气如何？"# target_ids = [201, 202, 203] # 对应"晴天，适合户外活动。"if len(input_ids) + len(source_ids) + len(target_ids) > max_length:break # 长度未超过20，不执行input_ids += source_ids + target_ids # [101, 102, 201, 202, 203]labels += [IGNORE_INDEX] * len(source_ids) + target_ids # [-100, -100, 201, 202, 203]- 处理后：
input_ids = [101, 102, 201, 202, 203]
labels = [-100, -100, 201, 202, 203] # -100表示忽略，只预测target部分更新model_inputs
model_inputs["input_ids"].append(input_ids)
model_inputs["attention_mask"].append([1] * len(input_ids))
model_inputs["labels"].append(labels)- 最终输出：
model_inputs = {"input_ids": [[101, 102, 201, 202, 203]],"attention_mask": [[1, 1, 1, 1, 1]],"labels": [[-100, -100, 201, 202, 203]]
}完整输出结果
{"input_ids": [[101, 102, 201, 202, 203]],"attention_mask": [[1, 1, 1, 1, 1]],"labels": [[-100, -100, 201, 202, 203]]
}代码作用总结
| 代码行 | 作用描述 |
|---|---|
model_inputs初始化 | 创建用于存储输入、掩码和标签的字典。 |
construct_example | 生成(query, response, history, system)元组，处理对话结构。 |
template.encode_multiturn | 将对话转换为token ID，区分source（用户输入）和target（模型输出）。 |
input_ids拼接 | 将source和target的token ID按顺序拼接。 |
labels设置 | source部分设为IGNORE_INDEX（如-100），仅预测target部分。 |
attention_mask生成 | 全1掩码，表示所有token都参与计算。 |
此函数适用于监督微调（SFT）场景，将对话数据转换为模型可接受的格式，通过labels指定需要预测的部分。
6).奖励模型数据处理（RM）
def preprocess_pairwise_dataset(examples):model_inputs = {"prompt_ids": [], "chosen_ids": [], "rejected_ids": []}for query, response, history, system in construct_example(examples):prompt_ids, chosen_ids = template.encode_oneturn(query, response[0], ...)_, rejected_ids = template.encode_oneturn(query, response[1], ...)model_inputs["prompt_ids"].append(prompt_ids)model_inputs["chosen_ids"].append(chosen_ids)model_inputs["rejected_ids"].append(rejected_ids)return model_inputs示例输入与配置
假设输入的examples包含1个样本，每个样本有1个查询和2个候选回复（分别为更优和更差的回复）：
examples = {"prompt": ["推荐一部科幻电影"],"response": [["《星际穿越》，科学与情感的完美结合", "《蜘蛛侠》，适合全家观看"] # response[0]=更优回复，response[1]=更差回复]
}# 假设template.encode_oneturn的编码结果（简化示例）：
# template.encode_oneturn("推荐一部科幻电影", "《星际穿越》...") → (prompt_ids=[101,102], chosen_ids=[201,202])
# template.encode_oneturn("推荐一部科幻电影", "《蜘蛛侠》...") → (prompt_ids=[101,102], rejected_ids=[301,302])逐行代码执行过程与输出
1. 初始化输出结构
model_inputs = {"prompt_ids": [], "chosen_ids": [], "rejected_ids": []}- 初始状态：
model_inputs = {"prompt_ids": [],"chosen_ids": [],"rejected_ids": []
}遍历construct_example生成的样本
for query, response, history, system in construct_example(examples):# query = "推荐一部科幻电影"# response = ["《星际穿越》...", "《蜘蛛侠》..."] # 包含两个回复的列表编码更优回复
prompt_ids, chosen_ids = template.encode_oneturn(query, response[0], ...)
# 假设编码结果：
# prompt_ids = [101, 102] # 对应"推荐一部科幻电影"
# chosen_ids = [201, 202] # 对应"《星际穿越》，科学与情感的完美结合"编码更差回复
_, rejected_ids = template.encode_oneturn(query, response[1], ...)
# 假设编码结果：
# rejected_ids = [301, 302] # 对应"《蜘蛛侠》，适合全家观看"更新model_inputs
model_inputs["prompt_ids"].append(prompt_ids)
model_inputs["chosen_ids"].append(chosen_ids)
model_inputs["rejected_ids"].append(rejected_ids)- 最终输出：
model_inputs = {"prompt_ids": [[101, 102]], # 查询的token ID"chosen_ids": [[201, 202]], # 更优回复的token ID"rejected_ids": [[301, 302]] # 更差回复的token ID
}完整输出结果
{"prompt_ids": [[101, 102]],"chosen_ids": [[201, 202]],"rejected_ids": [[301, 302]]
}代码作用总结
| 代码行 | 作用描述 |
|---|---|
model_inputs初始化 | 创建用于存储查询、更优回复、更差回复的字典。 |
construct_example | 生成(query, response, history, system)元组，处理对话结构。 |
template.encode_oneturn | 将查询和回复编码为token ID，分别处理更优和更差的回复。 |
| 结果收集 | 将编码后的查询、更优回复、更差回复分别存入对应的列表。 |
此函数适用于基于人类反馈的强化学习（RLHF）中的偏好训练，通过对比同一查询的两个回复，让模型学习人类偏好。
7)策略优化数据处理（PPO
def preprocess_unsupervised_dataset(examples: Dict[str, List[Any]]) -> Dict[str, Any]:model_inputs = {"input_ids": [], "attention_mask": [], "labels": []}for query, response, history, system in construct_example(examples):source_ids, target_ids = template.encode_oneturn(...)model_inputs["input_ids"].append(source_ids)model_inputs["attention_mask"].append([1] * len(source_ids))model_inputs["labels"].append(target_ids)return model_inputs示例输入与配置
假设输入的examples包含1个样本，配置如下：
examples = {"prompt": ["介绍Python"],"response": ["Python是一种高级编程语言"]
}# 假设template.encode_oneturn的编码结果（简化示例）：
# template.encode_oneturn("介绍Python", "Python是一种高级编程语言") →
# (source_ids=[101, 102], target_ids=[201, 202, 203])逐行代码执行过程与输出
1. 初始化输出结构
model_inputs = {"input_ids": [], "attention_mask": [], "labels": []}- 初始状态：
model_inputs = {"input_ids": [],"attention_mask": [],"labels": []
}遍历construct_example生成的样本
for query, response, history, system in construct_example(examples):# query = "介绍Python"# response = "Python是一种高级编程语言"编码样本
source_ids, target_ids = template.encode_oneturn(...)
# 假设编码结果：
# source_ids = [101, 102] # 对应"介绍Python"
# target_ids = [201, 202, 203] # 对应"Python是一种高级编程语言"更新model_inputs
model_inputs["input_ids"].append(source_ids)
model_inputs["attention_mask"].append([1] * len(source_ids))
model_inputs["labels"].append(target_ids)- 最终输出：
model_inputs = {"input_ids": [[101, 102]], # 输入的token ID"attention_mask": [[1, 1]], # 注意力掩码（全1表示不忽略任何token）"labels": [[201, 202, 203]] # 目标输出的token ID
}完整输出结果
{"input_ids": [[101, 102]],"attention_mask": [[1, 1]],"labels": [[201, 202, 203]]
}代码作用总结
| 代码行 | 作用描述 |
|---|---|
model_inputs初始化 | 创建用于存储输入、掩码和标签的字典。 |
construct_example | 生成(query, response, history, system)元组，处理对话结构。 |
template.encode_oneturn | 将查询和回复编码为token ID，分别得到source（输入）和target（目标）。 |
| 结果收集 | 将编码后的输入、注意力掩码（全1）和目标输出分别存入对应的列表。 |
此函数适用于无监督学习或自监督学习场景，例如语言模型的生成任务，模型需要根据input_ids生成labels中的内容。
8). 主处理逻辑
if stage == "pt":dataset = dataset.filter(lambda example: example["prompt"])preprocess_function = preprocess_pretrain_dataset
elif stage == "sft" and not training_args.predict_with_generate:dataset = dataset.filter(lambda example: example["prompt"] and example["response"])preprocess_function = preprocess_supervised_dataset
elif stage == "rm":dataset = dataset.filter(lambda example: example["prompt"] and len(example["response"]) > 1)preprocess_function = preprocess_pairwise_dataset
else:dataset = dataset.filter(lambda example: example["prompt"])preprocess_function = preprocess_unsupervised_datasetwith training_args.main_process_first(...):dataset = dataset.map(preprocess_function,batched=True,remove_columns=column_names,**kwargs)print_function(next(iter(dataset)))示例输入数据
假设初始数据集包含3个样本，每个样本有prompt和response字段：
dataset = [{"prompt": "介绍Python", "response": ["Python是高级语言"]}, # 样本1{"prompt": "推荐电影", "response": ["《星际穿越》", "《蜘蛛侠》"]}, # 样本2（两个回复，用于RM）{"prompt": "", "response": ["空查询"]} # 样本3（无效）
]场景1：预训练阶段（stage="pt"）
代码执行过程
# 1. 过滤数据：保留prompt不为空的样本
dataset = dataset.filter(lambda example: example["prompt"])
# 过滤后：[样本1, 样本2]（丢弃样本3）# 2. 指定预处理函数
preprocess_function = preprocess_pretrain_dataset # 拼接所有文本# 3. 批量处理数据
dataset = dataset.map(preprocess_function,batched=True,remove_columns=["prompt", "response"] # 删除原始字段
)预处理结果
# 假设preprocess_pretrain_dataset的输出（简化）：
dataset = [{"input_ids": [101, 102, 201, 202], # "介绍Python Python是高级语言""attention_mask": [1, 1, 1, 1],"labels": [101, 102, 201, 202] # 自监督任务，标签与输入相同},{"input_ids": [103, 104, 203, 204, 205, 206], # "推荐电影 《星际穿越》 《蜘蛛侠》""attention_mask": [1, 1, 1, 1, 1, 1],"labels": [103, 104, 203, 204, 205, 206]}
]场景2：监督微调阶段（stage="sft"）
代码执行过程
# 1. 过滤数据：保留prompt和response都不为空的样本
dataset = dataset.filter(lambda example: example["prompt"] and example["response"])
# 过滤后：[样本1, 样本2]（丢弃样本3）# 2. 指定预处理函数
preprocess_function = preprocess_supervised_dataset # 区分输入和目标# 3. 批量处理数据
dataset = dataset.map(...)预处理结果
# 假设preprocess_supervised_dataset的输出（简化）：
dataset = [{"input_ids": [101, 102], # "介绍Python""attention_mask": [1, 1],"labels": [201, 202] # "Python是高级语言"},{"input_ids": [103, 104], # "推荐电影""attention_mask": [1, 1],"labels": [203, 204] # "《星际穿越》"（只取第一个回复）}
]场景3：奖励模型训练阶段（stage="rm"）
代码执行过程
# 1. 过滤数据：保留prompt不为空且有多个response的样本
dataset = dataset.filter(lambda example: example["prompt"] and len(example["response"]) > 1)
# 过滤后：[样本2]（丢弃样本1和3）# 2. 指定预处理函数
preprocess_function = preprocess_pairwise_dataset # 处理偏好对# 3. 批量处理数据
dataset = dataset.map(...)预处理结果
# 假设preprocess_pairwise_dataset的输出（简化）：
dataset = [{"prompt_ids": [103, 104], # "推荐电影""chosen_ids": [203, 204], # "《星际穿越》"（更优回复）"rejected_ids": [205, 206] # "《蜘蛛侠》"（更差回复）}
]场景4：默认/无监督阶段
代码执行过程
# 1. 过滤数据：保留prompt不为空的样本
dataset = dataset.filter(lambda example: example["prompt"])
# 过滤后：[样本1, 样本2]（丢弃样本3）# 2. 指定预处理函数
preprocess_function = preprocess_unsupervised_dataset # 无监督格式# 3. 批量处理数据
dataset = dataset.map(...)预处理结果
# 假设preprocess_unsupervised_dataset的输出（简化）：
dataset = [{"input_ids": [101, 102], # "介绍Python""attention_mask": [1, 1],"labels": [201, 202] # "Python是高级语言"},{"input_ids": [103, 104], # "推荐电影""attention_mask": [1, 1],"labels": [203, 204] # "《星际穿越》"（只取第一个回复）}
]代码作用总结
| 代码行 | 作用描述 |
|---|---|
dataset.filter() | 根据不同阶段要求过滤无效样本（如prompt为空或回复不足）。 |
preprocess_function选择 | 根据阶段选择对应预处理函数（拼接文本/监督学习/偏好对/无监督）。 |
dataset.map() | 批量应用预处理函数，将原始字段转换为模型所需的输入格式（如input_ids、labels）。 |
remove_columns | 删除原始字段，减少内存占用。 |
main_process_first | 确保主进程先处理数据，避免多进程冲突。 |
此流程展示了不同训练阶段（预训练、监督微调、奖励模型训练）的数据处理差异，通过过滤和格式转换，将原始数据转换为模型可接受的输入格式。
总结
1、preprocess_pretrain_dataset处理PreTraining阶段的数据
数据组成形式： 输入input： X1 X2 X3 标签labels：X1 X2 X3 典型的Decoder架构的数据训练方式； 2、preprocess_supervised_dataset处理SFT阶段的数据
数据组成形式： 输入input： prompt response 标签labels： -100 ... -100 response 这里面labels的重点在于prompt部分的被-100所填充，主要作用在下面会介绍到。
lm_logits = self.lm_head(hidden_states)
loss = None
if labels is not None:labels = labels.to(lm_logits.device)shift_logits = lm_logits[..., :-1, :].contiguous()shift_labels = labels[..., 1:].contiguous()loss_fct = CrossEntropyLoss()loss = loss_fct(shift_logits.view(-1, shift_logits.size(-1)), shift_labels.view(-1))### pass乍一看上面这个loss貌似只是预训练Pre-training阶段的loss，SFT阶段的loss难道也是这么算的吗？答案是肯定的。
SFT阶段数据的输入输出格式是的形式，实际训练过程中应该是给定prompt，计算response部分的loss，但是这个是怎么实现的呢？
关键的要点来了，还记得在数据预处理阶段labels的加工方式吗？对于prompt部分的labels被-100所填充，导致在计算loss的时候模型只计算response部分的loss，-100的部分被忽略了。而这个机制得益于torch的CrossEntropyLossignore_index参数，ignore_index参数定义为如果labels中包含了指定了需要忽略的类别号（默认是-100），那么在计算loss的时候就不会计算该部分的loss也就对梯度的更新不起作用，详情可以查看这部分的定义。
最后，可以看出不管是Pre-training阶段还是SFT阶段，loss函数都是一样的，只是计算的方式存在差异，Pre-training阶段计算的是整段输入文本的loss，而SFT阶段计算的是response部分的loss。
如果您认为博文还不错，请帮忙点赞、收藏、关注。您的反馈是我的原动力
| 原创文章 | |
| 1 | FFN前馈网络与激活函数技术解析：Transformer模型中的关键模块 |
| 2 | Transformer掩码技术全解析：分类、原理与应用场景 |
| 3 | 【大模型技术】Attention注意力机制详解一 |
| 4 | Transformer核心技术解析LCPO方法：精准控制推理长度的新突破 |
| 5 | Transformer模型中位置编码（Positional Embedding）技术全解析(二) |
| 6 | Transformer模型中位置编码（Positional Embedding）技术全解析(一) |
| 7 | 自然语言处理核心技术词嵌入（Word Embedding），从基础原理到大模型应用 |
| 8 | DeepSeek-v3：基于MLA的高效kv缓存压缩与位置编码优化技术 |
| 9 | 【Tokenization第二章】分词算法深度解析：BPE、WordPiece与Unigram的原理、实现与优化 |
| 10 | Tokenization自然语言处理中分词技术：从传统规则到现代子词粒度方法 |
| 11 | [预训练]Encoder-only架构的预训练任务核心机制 |
| 12 | 【第一章】大模型预训练全解析：定义、数据处理、流程及多阶段训练逻辑 |