企业官网建设流程全解析

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

【免费下载链接】LlamaFactoryUnified Efficient Fine-Tuning of 100+ LLMs & VLMs (ACL 2024)项目地址: https://gitcode.com/GitHub_Trending/ll/LlamaFactory

本文围绕 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）。

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

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

DatasetInfo：插件消费的元信息

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

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 Facedatasets.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的其余逻辑无需改动。

【免费下载链接】LlamaFactoryUnified Efficient Fine-Tuning of 100+ LLMs & VLMs (ACL 2024)项目地址: https://gitcode.com/GitHub_Trending/ll/LlamaFactory