LLaMA-Factory v1 Data Plugins 详解：数据转换、加载、索引与选择插件的实现原理

本文围绕 LLaMA-Factory 的 Data Plugins 开发文档，系统讲解 v1 数据管线中四类数据插件——DataConverterPlugin（数据格式标准化）、DataLoaderPlugin（本地数据集加载）、DataIndexPlugin（size/weight 索引调整）与 DataSelectorPlugin（基于索引的样本选择）的职责分工与接口设计，并结合 converter.py、loader.py 与 data_engine.py 的源码实现，说明每个插件在 DataEngine 中的真实调用时机，帮助读者掌握 v1 标准 Messages 数据格式、内置转换器的转换逻辑以及自定义转换器的注册方式。

Data Plugins 在 v1 数据管线中的位置

在 LLaMA-Factory v1 的数据管线中，DataEngine 是等价于 torch Dataset 的统一数据入口，其初始化流程为三步：

1. 解析数据集信息（_get_dataset_info）；
2. 按数据集信息加载数据集（_load_dataset）；
3. 构建数据索引，必要时按 size/weight 重采样（_build_data_index）。

四类数据插件恰好分别嵌入这条链路的不同环节：

插件	文件	在 DataEngine 中的调用点
DataLoaderPlugin	loader.py	_load_dataset 中按 source 路由（data_engine.py#L90-L99）
DataIndexPlugin 能力（adjust_data_index）	loader.py#L69-L88	_build_data_index 中按 size/weight 调整索引（data_engine.py#L120-L127）
DataConverterPlugin	converter.py	_convert_data_sample 中按需调用（data_engine.py#L143-L159）
DataSelectorPlugin 能力（select_data_sample）	loader.py#L91-L108	DataEngine.__getitem__ 处理非 int 索引时（data_engine.py#L172-L193）

所有插件共享同一个轻量路由基类 BasePlugin（位于 plugin.py）：每个插件家族子类持有独立注册表 _registry，通过 @DataConverterPlugin("alpaca").register() 这样的装饰器把实现函数注册到具名条目下，实例化时按名称解析（_resolve），再经 __call__ 转发调用。这就是 v1 中“按配置名路由插件实现”的统一机制。

DatasetInfo：插件消费的元信息

四类插件共同依赖的数据结构是 DatasetInfo（定义于 types.py），它是数据集 YAML 配置中每个条目的字段契约：

字段	类型	默认值	说明
path	str	必填	数据集路径（本地文件或 HF Hub 仓库）
source	"hf_hub" | "ms_hub" | "local"	"hf_hub"	数据集来源，决定走 HF 原生 load_dataset 还是 DataLoaderPlugin
split	str	"train"	数据集切分名
converter	str	无	指定 DataConverterPlugin 的注册名
size	int	全部样本	目标样本数
weight	float	1.0	数据集在混合训练中的采样权重
streaming	bool	False	是否流式加载

DataConverterPlugin：将非标准格式转换为 v1 标准 Messages 格式

为什么需要 DataConverter

v1 训练链路消费的是统一的 Sample 格式（SFTSample/DPOSample，见 types.py#L97-L121），其核心是 messages 列表，每条消息包含 role、content（由 Content 块组成，如 {"type": "text", "value": "..."}）以及用于控制 loss 的 loss_weight。DataConverter 负责把非标准格式的数据集（如 Alpaca 格式）转换为该标准格式，使既有的社区数据集无需手动改写即可复用；对于自定义格式的数据集，用户也可以编写自己的转换器插件来完成标准化。

从 data_engine.py#L143-L159 的 _convert_data_sample 可以看到调用逻辑：若数据集配置了 converter 字段，则实例化 DataConverterPlugin(converter) 并把原始样本转发给对应实现；否则直接透传原始样本。也就是说，转换器是可选的，只有 YAML 中显式指定 converter 时才会生效。

当前仓库已内置三个转换器（比文档描述的 Alpaca/Pair 多一个 ShareGPT），注册于 converter.py：

• @DataConverterPlugin("alpaca").register() — Alpaca → SFT（converter.py#L133-L169）
• @DataConverterPlugin("sharegpt").register() — ShareGPT 多轮对话 → SFT（converter.py#L172-L235）
• @DataConverterPlugin("pair").register() — chosen/rejected 偏好对 → DPO（converter.py#L238-L297）

Alpaca Converter 详解

Alpaca 格式

Alpaca 格式是一种常见的指令微调数据格式：

{
  "system": "You are a helpful assistant.",
  "instruction": "Describe a process of making crepes.",
  "input": "",
  "output": "Making crepes is an easy and delicious process..."
}

对应的输入类型是 AlpacaSample（TypedDict, total=False），其中 instruction 与 output 为必需字段，system、input 可选；实际实现中还额外支持 images、videos、audios 三种多模态媒体列，配合内联占位符使用。

转换逻辑

alpaca_converter 将一条 Alpaca 样本转换为标准 SFTSample，核心规则：

1. 若存在 system 字段 → 生成一条 role="system" 消息，loss_weight = 0.0；
2. 若存在 instruction 或 input 字段 → 将两者直接拼接（instruction + input）为一条 role="user" 消息，loss_weight = 0.0；
3. 若存在 output 字段 → 生成一条 role="assistant" 消息，loss_weight = 1.0（即仅对模型回复部分计算 loss）。

实际源码（converter.py#L133-L169）与文档示例一致，并叠加了多模态处理：

@DataConverterPlugin("alpaca").register()
def alpaca_converter(raw_sample: AlpacaSample) -> SFTSample:
    messages = []
    media_iters = _build_media_iters(raw_sample)
    if "system" in raw_sample:
        messages.append(
            {"role": "system", "content": [{"type": "text", "value": raw_sample["system"]}], "loss_weight": 0.0}
        )

    if "instruction" in raw_sample or "input" in raw_sample:
        messages.append(
            {
                "role": "user",
                "content": _to_content_blocks(
                    raw_sample.get("instruction", "") + raw_sample.get("input", ""), media_iters
                ),
                "loss_weight": 0.0,
            }
        )

    if "output" in raw_sample:
        messages.append(
            {"role": "assistant", "content": [{"type": "text", "value": raw_sample["output"]}], "loss_weight": 1.0}
        )

    _assert_media_consumed(media_iters)
    return {"messages": messages}

其中 _to_content_blocks（converter.py#L94-L115）会把文本按媒体占位符（图片/视频/音频）切分，将 {"type": "image_url", "value": path} 等媒体块按文档顺序插入到 content 中；若文本不含占位符，则退化为单个 text 块。_assert_media_consumed 则保证占位符数量与媒体文件数量严格匹配，多或少都会抛出 ValueError。

转换示例

输入（Alpaca 格式）：

{
  "instruction": "What is the capital of France?",
  "input": "",
  "output": "The capital of France is Paris."
}

输出（v1 标准格式）：

{
  "messages": [
    {
      "role": "user",
      "content": [{"type": "text", "value": "What is the capital of France?"}],
      "loss_weight": 0.0
    },
    {
      "role": "assistant",
      "content": [{"type": "text", "value": "The capital of France is Paris."}],
      "loss_weight": 1.0
    }
  ]
}

当样本带有 system 字段时，输出中会额外在最前面插入一条 {"role": "system", ..., "loss_weight": 0.0} 消息。

Pair Converter：偏好对数据集的标准化

pair_converter 面向 DPO 类训练，将 {"chosen": [...], "rejected": [...]} 结构的样本（OpenAI messages 风格，role 为 user/assistant/tool）转换为 DPOSample（chosen_messages + rejected_messages），并对两侧分别做媒体占位符展开；role == "assistant" 的消息 loss_weight 为 1.0，其余为 0.0；若消息 role 为 tool，其 content 会按 JSON 解析为 tool_call 内容块。仓库自带的 v1_dpo_demo.yaml 即为该转换器的使用示例：

dpo_zh_demo:
  path: HuggingFaceH4/orca_dpo_pairs
  split: train_prefs
  converter: pair

ShareGPT Converter：多轮对话与工具调用

sharegpt_converter 处理 conversations 结构的 ShareGPT 数据，其角色映射为：system→system、human→user、gpt→assistant、function_call→assistant（解析为 tool_call 块）、observation→tool；仅 gpt 侧消息 loss_weight = 1.0，其余为 0.0；样本级 tools 字段（JSON 字符串）会被规范化后写入 sample["tools"]。

创建自定义转换器

如果你的数据集有自己的格式（例如 question/answer/context 结构），只需三步即可扩展，方式可参考文档给出的示例：

# src/llamafactory/v1/plugins/data_plugins/converter.py

from typing import TypedDict, NotRequired
from ...utils.types import SFTSample

# 1. 定义输入格式的类型
class MyCustomSample(TypedDict, total=False):
    question: str
    answer: str
    context: NotRequired[str]

# 2. 实现转换逻辑
def custom_converter(raw_sample: MyCustomSample) -> SFTSample:
    messages = []

    # 构建用户消息
    user_text = raw_sample["question"]
    if "context" in raw_sample:
        user_text = f"Context: {raw_sample['context']}\n\nQuestion: {user_text}"

    messages.append({
        "role": "user",
        "content": [{"type": "text", "value": user_text}],
        "loss_weight": 0.0
    })

    # 构建助手消息
    messages.append({
        "role": "assistant",
        "content": [{"type": "text", "value": raw_sample["answer"]}],
        "loss_weight": 1.0
    })

    return {"messages": messages}

# 3. 注册转换器：通过 BasePlugin 的具名注册机制接入插件路由
#    实际注册方式为装饰器（见内置转换器写法）
@DataConverterPlugin("custom").register()
def custom_converter(raw_sample: MyCustomSample) -> SFTSample:
    ...

从 BasePlugin 源码可以看到：文档中 CONVERTERS = {"alpaca": alpaca_converter, ...} 的字典写法在 v1 实现中已升级为每个插件家族自持的 _registry 注册表——DataConverterPlugin("custom").register() 返回装饰器，将函数存入该类注册表；重名注册会打印一次告警并覆盖。DataEngine._convert_data_sample 中的 DataConverterPlugin(converter)(raw_sample) 即按 YAML 里的 converter 名解析出对应函数并执行。

在 YAML 配置中使用转换器

在数据集配置中通过 converter 字段指定注册名即可。仓库自带的 v1_sft_demo.yaml 展示了标准用法：

identity:
  path: data/identity.json
  source: local
  converter: alpaca
alpaca_en_demo:
  path: data/alpaca_en_demo.json
  source: local
  converter: alpaca
  size: 500

自定义数据集的写法：

my_dataset:
  path: custom_data.json
  source: local
  converter: custom

DataLoaderPlugin：本地数据集的多格式加载

职责与支持的文件格式

DataLoaderPlugin 负责从本地文件加载数据集，当前支持：

• JSON：.json
• JSONL：.jsonl
• CSV：.csv
• Parquet：.parquet
• Arrow：.arrow
• Text：.txt

源码实现

实际实现位于 loader.py，其核心是插件类 + 名为 "local" 的注册函数：

class DataLoaderPlugin(BasePlugin):
    """Plugin for loading dataset."""

    def load(self, dataset_info: DatasetInfo) -> HFDataset:
        path = dataset_info["path"]
        split = dataset_info.get("split", "train")
        streaming = dataset_info.get("streaming", False)
        return super().__call__(path, split, streaming)

load 从 DatasetInfo 中取出 path、split（默认 "train"）、streaming（默认 False），然后通过 BasePlugin.__call__ 路由到注册的实现 load_data_from_file：

def _get_builder_name(path: str) -> Literal["arrow", "csv", "json", "parquet", "text"]:
    filetype = os.path.splitext(path)[-1][1:]
    if filetype in ["arrow", "csv", "json", "jsonl", "parquet", "txt"]:
        return filetype.replace("jsonl", "json").replace("txt", "text")
    else:
        raise ValueError(f"Unknown dataset filetype: {filetype}.")


@DataLoaderPlugin("local").register()
def load_data_from_file(filepath: str, split: str, streaming: bool) -> HFDataset:
    if os.path.isdir(filepath):
        filetype = _get_builder_name(os.listdir(filepath)[0])
        dataset = load_dataset(filetype, data_dir=filepath, split=split)
    elif os.path.isfile(filepath):
        filetype = _get_builder_name(filepath)
        dataset = load_dataset(filetype, data_files=filepath, split=split)
    else:
        raise ValueError(f"Can not load dataset from {filepath}.")

    if streaming:  # faster when data is streamed from local files
        dataset = dataset.to_iterable_dataset()

    return dataset

从实现可以看到几个关键事实：

• 加载完全委托给 Hugging Face datasets.load_dataset，扩展名到 builder 的映射规则是 jsonl→json、txt→text，其余同名直通；未知扩展名抛 ValueError；
• 同时支持单文件（data_files）与目录（data_dir，以目录下第一个文件的扩展名推断类型）两种输入；
• streaming=True 时结果会转换为 IterableDataset，源码注释说明本地流式读取更快；
• 该插件在 DataEngine._load_dataset 中仅当 source != "hf_hub" 时被调用（data_engine.py#L90-L99）：source: local 走 DataLoaderPlugin("local")，而默认的 hf_hub 走 datasets.load_dataset 原路；
• 另有一条约束：同一训练配置中所有数据集必须同为流式或同为非流式，否则 _load_dataset 直接抛错（data_engine.py#L85-L88）。

DataIndexPlugin：用 size 与 weight 控制样本数量与采样频率

配置方式

DataIndexPlugin 负责调整数据索引，支持通过 size、weight 控制数据集的样本数量和采样频率：

• 使用 size 参数限制使用的样本数量：

my_dataset:
  path: large_dataset.json
  size: 1000  # 只使用前 1000 个样本

• 使用 weight 参数调整数据集在混合数据中的采样频率：

dataset_a:
  path: data_a.json
  weight: 1.0

dataset_b:
  path: data_b.json
  weight: 2.0  # dataset_b 的样本出现频率是 dataset_a 的 2 倍

weight 参数适用于多个数据集混合训练时调整不同数据集的采样频率：weight=1.0 时按原始比例采样；weight=2.0 时该数据集的索引会复制约 2 倍，使其样本出现频率翻倍。

源码实现：random.choices 采样

v1 中的实际实现是 loader.py#L69-L88 中的 adjust_data_index 函数（而非文档接口定义中的 adjust_by_size/adjust_by_weight 两个方法），其策略为有放回随机采样：

def adjust_data_index(
    data_index: list[tuple[str, int]], size: int | None, weight: float | None
) -> list[tuple[str, int]]:
    if size is not None:
        data_index = random.choices(data_index, k=size)

    if weight is not None:
        data_index = random.choices(data_index, k=int(len(data_index) * weight))

    return data_index

可以观察到：size 与 weight 若同时配置会依次生效——先采到 size 个，再按 len × weight 二次采样；weight=1.0 时 k 恰为原长度，即近似保持原分布。

调用点位于 DataEngine._build_data_index：每个数据集先展开为索引列表 [(dataset_name, sample_index, cut), ...]（多轮 SFT 会按受监督的 assistant 轮做前缀展开，详见下节），随后当配置中存在 size 或 weight 时调用 adjust_data_index，最后并入全局 data_index。因此 len(DataEngine) 返回的是调整后的索引总数，即真实的训练样本数。

顺带说明：索引为何是三元组

从 data_engine.py#L101-L141 看，v1 的 data_index 元素是 (dataset_name, sample_index, cut) 三元组：多轮对话 u1 a1 u2 a2 会被前缀展开为 [2, 4] 两条索引（分别训练 messages[:2] 与 messages[:4]，各自只监督最后一个 assistant 轮），DPO 样本、流式数据或无受监督轮次的样本则保留整条（cut=None）。流式数据集无法预数轮次，会固定生成 1000 条 (-1, None) 占位索引，且此时 __len__ 返回 -1、索引访问会直接抛错。

DataSelectorPlugin：基于索引的样本选择

DataSelectorPlugin 为 DataEngine 提供基于索引访问数据的能力。v1 中对应的实际实现是 loader.py#L91-L108 的 select_data_sample：

def select_data_sample(
    data_index: list[tuple[str, int]], index: slice | list[int] | Any
) -> tuple[str, int] | list[tuple[str, int]]:
    if isinstance(index, slice):
        return [data_index[i] for i in range(*index.indices(len(data_index)))]
    elif isinstance(index, list):
        return [data_index[i] for i in index]
    else:
        raise ValueError(f"Invalid index type {type(index)}.")

其行为与文档接口定义一致：

• 输入为 slice 时，返回对应范围内的样本索引列表；
• 输入为 list[int] 时，返回指定位置的多条索引；
• 其他类型（包括单个 int）抛出 ValueError——单整数索引在 DataEngine.getitem 中提前分流到 self._get(*self.data_index[index])，只有切片/列表这类批量索引才走选择插件路径。

选择出的索引经 select_data_sample 解析后，逐条交给 _get 完成原始行读取与转换器调用（data_engine.py#L195-L200），cut 非空时还会把 messages 截断到对应前缀长度。

小结：一次训练请求的数据流动

把四个插件串起来，一条 v1 训练数据的完整生命周期是：

1. DataEngine 从 YAML（如 v1_sft_demo.yaml）解析出各数据集的 DatasetInfo；
2. DataLoaderPlugin（source: local 时）按扩展名通过 datasets.load_dataset 载入文件；
3. _build_data_index 遍历样本并调用 DataConverterPlugin(converter) 做标准化（Alpaca/ShareGPT/Pair/自定义），同时按受监督轮次做前缀展开，再经 adjust_data_index 应用 size/weight 重采样；
4. 训练迭代时 DataEngine[i] 或批量索引经 select_data_sample 解析，最终由 _get 返回带 _dataset_name 的标准 Sample。

理解了这套插件化分工后，扩展数据能力只需在对应环节注册实现：新格式写一个转换器函数并 @DataConverterPlugin("名字").register()，新来源则在 DataLoaderPlugin 家族下新增具名加载函数，YAML 中用 converter/source 字段引用即可，DataEngine 的其余逻辑无需改动。