MS-SWIFT 数据处理流程详解

本文档详细说明从原始数据到训练的完整数据处理流程，包括每个步骤的参数控制和效果说明。

---

流程概览

展开代码
原始数据集 (dataset_info.json)
    ↓ 
① 数据加载 (DatasetLoader)
    ↓
② 数据预处理 (MessagesPreprocessor)
    ├── mask_history_sample 拆分 (可选)
    └── 格式标准化
    ↓
③ 长度过滤 (_select_dataset)
    ↓
④ 数据集划分 (split_dataset_ratio)
    ├── 训练集
    └── 验证集
    ↓
⑤ 数据编码 (Template.encode)
    ↓
⑥ 数据打包 (PackingDataset) (可选)
    ↓
⑦ 迭代次数计算 (train_iters)
    ↓
开始训练

---

详细流程说明

① 数据加载 (DatasetLoader)

功能：从本地或远程加载原始数据集

相关参数：

• dataset: 数据集ID或路径列表
  • 格式: dataset_id:subset#count
  • 示例: mmdu_val 或 /path/to/dataset.json
• custom_dataset_info: 自定义数据集配置文件路径
• dataset_num_proc: 数据加载的进程数（默认1）
• load_from_cache_file: 是否使用缓存（默认False，建议生产环境设True）
• streaming: 是否流式加载（默认False）

效果：

• 加载原始JSON/JSONL数据
• 注册自定义数据集配置
• 返回HuggingFace Dataset对象

示例日志：

展开代码
[INFO:swift] Successfully registered `/path/to/dataset_info.json`.

---

② 数据预处理 (MessagesPreprocessor)

2.1 mask_history_sample 拆分 (可选)

功能：将多轮对话样本拆分为多个训练样本，模拟渐进式对话训练

相关参数：

• mask_history_sample: 是否启用拆分（默认False）
• max_human_steps: 每个拆分样本最多保留几轮对话（默认-1，不限制）
• media_dir: 多模态数据（图片/视频）所在目录

拆分逻辑： 假设原始对话有N轮（N个user + N个assistant消息）：

• 拆分为N个训练样本
• 第1个样本：包含前1轮对话（最多保留max_human_steps轮历史）
• 第2个样本：包含前2轮对话
• 第N个样本：包含全部N轮对话

效果：

• 样本数扩增：110 → 1645（扩增比例14.95x）
• 每个样本标记 _mask_history_sample 字段用于后续处理

示例日志：

展开代码
[INFO:swift] [数据集拆分] 启用mask_history_sample模式: mask_history_sample=True, max_human_steps=2
[INFO:swift] [数据集拆分] 每个多轮对话样本将被拆分为多个训练样本,每个样本最多保留2个human步骤
[INFO:swift] [数据集拆分] 预处理前数据集大小: 110 个样本
[INFO:swift] [数据集拆分] 预处理后数据集大小: 1645 个样本
[INFO:swift] [数据集拆分] 样本数量变化: 110 -> 1645 (增加了 1535 个样本, 扩增比例: 14.95x)

适用场景：

• 多轮对话任务
• 需要模型学习渐进式对话能力
• 增加训练样本多样性

MS-SWIFT支持在messages中通过 loss 字段控制单个样本的损失：

bash
展开代码
{
  "messages": [
    {"role": "user", "content": "问题"},
    {"role": "assistant", "content": "回答", "loss": 0.0}  // 不计算损失
  ]
}

Loss Scale 策略说明

策略	说明	适用场景
default	所有assistant回复都计算损失（不包括system/user）	标准SFT训练
last_round	只有最后一轮assistant回复计算损失	RLHF、mask_history_sample拆分的数据
all	所有tokens都计算损失（包括system/user）	预训练

数据集组织：

bash
展开代码
{
  "example_masked_dataset": {
    "file_name": "example_data.json",
    "formatting": "sharegpt",
    "mask_history_sample": true,
    "max_human_steps": 2,
    "loss_scale": "last_round",
    "columns": {
      "messages": "conversations"
    },
    "tags": {
      "role_tag": "from",
      "content_tag": "value",
      "user_tag": "user",
      "assistant_tag": "assistant"
    }
  },
  "example_normal_dataset": {
    "file_name": "normal_sft.json",
    "formatting": "sharegpt"
  }
}

2.2 格式标准化

功能：将不同格式的对话数据统一为标准messages格式

支持格式：

• OpenAI格式: [{"role": "user", "content": "..."}]
• ShareGPT格式: [{"from": "human", "value": "..."}]
• 自定义格式（通过columns参数映射）

相关参数：

• columns: JSON字符串，用于列名映射
  • 示例: '{"text1": "query", "text2": "response"}'

---

③ 长度过滤 (_select_dataset)

功能：过滤掉超过最大长度的样本

相关参数：

• max_length: 最大序列长度（如2048）

过滤逻辑：

python
展开代码
# 保留 length <= max_length 的样本
idxs = [i for i, length in enumerate(dataset['length'])
        if (max(length) if isinstance(length, list) else length) <= max_length]

效果：

• 样本数减少：1645 → 101（过滤掉1544个超长样本）

示例日志：

展开代码
[INFO:swift] Dataset filtered, origin length: 1645, filtered dataset length: 101

说明：

• 多模态数据length可能是列表（多个图片/视频对应多个长度）
• 单模态数据length是整数
• 过滤发生在tokenize之前，基于预估长度

---

④ 数据集划分 (split_dataset_ratio)

功能：将数据集划分为训练集和验证集

相关参数：

• split_dataset_ratio: 验证集比例（0.0-1.0）
  • 0.0: 不划分，全部用于训练
  • 0.01: 1%用于验证，99%用于训练
  • 1.0: 全部用于验证
• val_dataset: 独立验证集（如指定此参数，则split_dataset_ratio自动设为0）
• dataset_shuffle: 是否在划分前打乱（默认True）
• data_seed: 随机种子（默认42）

划分逻辑：

python
展开代码
# 假设过滤后有103个样本，split_dataset_ratio=0.01
dataset_len = 103
val_sample = max(int(103 * 0.01), 1) = 1
train_sample = 103 - 1 = 102

# 使用train_test_split划分
train_dataset, val_dataset = dataset.train_test_split(
    test_size=1, shuffle=True, seed=42
)

效果：

• 训练集：101个样本
• 验证集：2个样本

示例日志：

展开代码
[INFO:swift] train_dataset: Dataset({
    features: ['messages', 'images', '_mask_history_sample', 'length'],
    num_rows: 101
})
[INFO:swift] val_dataset: Dataset({
    features: ['messages', 'images', '_mask_history_sample', 'length'],
    num_rows: 2
})

注意事项：

• val_dataset 会自动保存到输出目录：{output_dir}/val_dataset.jsonl
• 如果同时设置了val_dataset参数，则split_dataset_ratio会被忽略

---

⑤ 数据编码 (Template.encode)

功能：将文本转换为token IDs，处理多模态数据

相关参数：

• lazy_tokenize: 是否延迟编码
  • False: 训练前一次性编码所有样本（默认，LLM推荐）
  • True: 训练时动态编码（MLLM默认，节省内存）
• num_proc: 编码进程数（即dataset_num_proc）

处理内容：

• Tokenization: 文本 → token IDs
• 多模态编码：图片/视频 → 模型输入格式
• 添加特殊token（如BOS、EOS）
• 计算精确的token长度

效果：

• 为每个样本生成 input_ids, labels, attention_mask 等字段
• 更新 length 字段为精确token长度

---

⑥ 数据打包 (PackingDataset) (可选)

功能：将多个短样本合并为接近max_length的长样本，提高GPU利用率

相关参数：

• packing: 是否启用打包（默认False）
• packing_length: 打包目标长度（默认等于max_length）
• packing_num_proc: 打包进程数（默认1）

打包逻辑：

python
展开代码
# 自动限制进程数
packing_num_proc = min(配置值, math.ceil(len(dataset) / 1000))

# 示例：101个样本
packing_num_proc = min(8, math.ceil(101 / 1000)) = min(8, 1) = 1

打包算法：

• 使用贪心算法将多个样本打包到接近packing_length
• 保证不切分完整序列
• 使用Flash Attention确保packed样本内序列互不可见

效果：

• 样本数减少：101 → 86（打包效率：~85%）
• GPU利用率提升，显存占用更稳定
• 训练速度提升（减少padding浪费）

示例日志：

展开代码
[INFO:swift] ================================================================================
[INFO:swift] 数据打包(Packing)处理详情：
[INFO:swift] --------------------------------------------------------------------------------
[INFO:swift]   数据集样本总数: 101
[INFO:swift]   数据长度列表总数: 101
[INFO:swift]   最大打包长度(packing_length): 2048
[INFO:swift]   打包进程数(packing_num_proc): 1
[INFO:swift]   每批次处理大小(PACKING_BATCH_SIZE): 1000
[INFO:swift]   严格模式(strict): False
[INFO:swift]   开始使用 1 个进程并行打包 101 条数据...
[INFO:swift] ================================================================================
Packing: 100%|██████████| 101/101 [00:00<00:00, 10425.38it/s]

注意事项：

• Packing会减少样本数，需相应调整学习率和梯度累加步数
• 打包进程数自动限制：每1000个样本最多1个进程
• 不适合需要精确控制样本顺序的场景

---

⑦ 迭代次数计算 (train_iters)

功能：计算总训练迭代次数

相关参数：

• max_epochs: 训练轮数
• global_batch_size: 全局批次大小
• micro_batch_size: 每个GPU的批次大小
• tensor_model_parallel_size: 张量并行大小
• world_size: 总GPU数（NPROC_PER_NODE）

计算逻辑：

python
展开代码
# 步骤1：计算数据并行大小
data_parallel_size = world_size // tensor_model_parallel_size
# 示例：8 // 4 = 2

# 步骤2：计算step批次大小
step_batch_size = micro_batch_size × data_parallel_size
# 示例：1 × 2 = 2

# 步骤3：对齐数据集样本数（向下取整到step_batch_size的倍数）
dataset_sample = (len(train_dataset) // step_batch_size) × step_batch_size × num_generations
# 示例：(86 // 2) × 2 × 1 = 43 × 2 × 1 = 86

# 步骤4：计算总迭代次数
train_iters = (dataset_sample × max_epochs) // global_batch_size
# 示例：(86 × 10) // 8 = 860 // 8 = 107

效果：

• 总迭代次数：107
• 每个epoch：10.7次迭代
• 总训练样本数：860个（86样本 × 10轮）

示例日志：

展开代码
[INFO:swift] ================================================================================
[INFO:swift] 总迭代次数(train_iters)计算详情：
[INFO:swift] --------------------------------------------------------------------------------
[INFO:swift]   数据集总样本数(len(train_dataset)): 86
[INFO:swift]   micro_batch_size: 1
[INFO:swift]   data_parallel_size(数据并行数): 2
[INFO:swift]   step_batch_size = micro_batch_size × data_parallel_size = 1 × 2 = 2
[INFO:swift]   num_generations(每个样本生成次数): 1
[INFO:swift]   dataset_sample = (dataset_len // step_batch_size) × step_batch_size × num_generations
[INFO:swift]                  = (86 // 2) × 2 × 1
[INFO:swift]                  = 43 × 2 × 1
[INFO:swift]                  = 86
[INFO:swift]   max_epochs(训练轮数): 10
[INFO:swift]   global_batch_size(全局批次大小): 8
[INFO:swift]   train_iters = (dataset_sample × max_epochs) // global_batch_size
[INFO:swift]               = (86 × 10) // 8
[INFO:swift]               = 860 // 8
[INFO:swift]               = 107
[INFO:swift] ================================================================================

参数说明：

• num_generations: RLHF任务中每个样本生成多个响应，SFT任务为1
• 对齐操作会丢弃无法凑成完整step_batch的样本

---

完整示例

以你的训练配置为例，展示完整的数据处理流程：

配置参数

yaml
展开代码
# qwen3_vl_8b_train_swanlab.yaml
dataset: mmdu_val
max_length: 2048
split_dataset_ratio: 0.01
packing: true
micro_batch_size: 1
global_batch_size: 8
max_epochs: 10
tensor_model_parallel_size: 4
dataset_num_proc: 8

处理流程

步骤	操作	样本数	说明
0	原始数据	110	mmdu_val数据集
1	mask_history_sample拆分	1645	扩增14.95倍
2	长度过滤 (>2048)	103	过滤1542个超长样本
3	数据集划分 (1% val)	101 (train) + 2 (val)	1个验证样本，102个训练样本
4	Packing打包	86	打包效率85%
5	对齐到step_batch_size	86	无损失（86是2的倍数）
6	计算迭代次数	-	107次迭代 = (86×10)//8

训练统计

• 单epoch样本数：86
• 单epoch迭代数：10.7 (86×1//8)
• 总迭代次数：107
• 总训练样本数：860 (86×10)

---

常见问题

Q1: 为什么打包进程数只有1？

A: 打包进程数自动限制为 min(packing_num_proc, ceil(len(dataset)/1000))。当样本数<1000时，只会使用1个进程。这是合理的，因为打包速度通常远快于tokenize速度。

Q2: Packing会损失多少样本？

A: 取决于样本长度分布。通常损失10-20%。可以通过调整学习率和梯度累加来补偿。建议公式：

展开代码
新学习率 = 原学习率 × (packing后样本数 / packing前样本数)

Q3: mask_history_sample适用于哪些场景？

A: 适用于：

• 多轮对话训练
• 需要模型学习渐进式对话能力
• 样本量较少，需要数据增强

不适用于：

• 单轮问答任务
• 已有充足样本的场景

Q4: split_dataset_ratio如何设置？

A: 建议值：

• 大数据集（>10k样本）：0.01-0.05
• 中型数据集（1k-10k）：0.05-0.1
• 小数据集（<1k）：0.1-0.2
• 如有独立验证集，设为0.0

Q5: 如何提高数据处理速度？

A:

1. 设置 load_from_cache_file: true（生产环境）
2. 增大 dataset_num_proc（纯文本推荐4-8）
3. 多模态模型谨慎使用多进程（可能更慢）
4. 设置 OMP_NUM_THREADS 环境变量限制OpenMP线程数

Q6: 为什么训练样本数从101变成86？

A: 启用了Packing。Packing将多个短样本合并，减少样本数但提高GPU利用率。这是正常的，不是bug。

---

参数速查表

数据加载相关

参数	默认值	说明
dataset	[]	数据集ID或路径列表
custom_dataset_info	[]	自定义数据集配置
dataset_num_proc	1	预处理进程数
load_from_cache_file	False	是否使用缓存
streaming	False	是否流式加载
data_seed	42	随机种子

数据预处理相关

参数	默认值	说明
mask_history_sample	False	是否拆分多轮对话
max_human_steps	-1	每个样本最多保留轮数
media_dir	None	多模态数据目录
columns	None	列名映射
max_length	2048	最大序列长度

数据集划分相关

参数	默认值	说明
split_dataset_ratio	0.0	验证集比例
val_dataset	[]	独立验证集
dataset_shuffle	True	是否打乱数据集

数据打包相关

参数	默认值	说明
packing	False	是否启用打包
packing_length	None	打包长度（None=max_length）
packing_num_proc	1	打包进程数

训练迭代相关

参数	默认值	说明
max_epochs	-	训练轮数
micro_batch_size	1	单GPU批次大小
global_batch_size	-	全局批次大小
tensor_model_parallel_size	1	张量并行大小

---

总结

MS-SWIFT的数据处理流程设计合理，支持多种优化策略：

1. mask_history_sample：有效扩增多轮对话样本
2. 长度过滤：自动去除不合适的样本
3. 智能划分：灵活的训练/验证集划分
4. Packing优化：提高GPU利用率，加速训练
5. 详细日志：每个步骤都有清晰的日志输出

合理配置这些参数，可以在数据质量、训练效率和模型效果之间取得最佳平衡。