File size: 14,741 Bytes
83ddd7e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 | # Data Plugins
## 1. Data Plugins 简介
## DataConverterPlugin
### 1. DataConverterPlugin 简介
DataConverter 负责将非标准格式的数据集转换为 v1 的标准 Messages 格式。这使得用户可以继续使用现有的数据集（如 Alpaca 格式），而无需手动转换。针对自定义格式的数据集，用户也可以通过构建对应的自定义 DataConverter 插件，来负责其数据格式标准化。
当前，LLaMA-Factory 已内置了 `Alpaca Converter` 和 `Pair Converter`，这两类数据集可以直接使用对应的 converter 进行标准化，无需自定义转换器。
### 2. Alpaca Converter 详解
#### 2.1 Alpaca 格式
Alpaca 格式是一种常见的指令微调数据格式：
```json
{
"system": "You are a helpful assistant.",
"instruction": "Describe a process of making crepes.",
"input": "",
"output": "Making crepes is an easy and delicious process..."
}
```
#### 2.2 Alpaca Converter 接口定义
```python
class AlpacaSample(TypedDict, total=False):
"""Alpaca 格式数据样本结构
attr:
system (str, 可选): 系统提示信息（system prompt），用于设定对话背景或模型行为。
instruction (str, 可选): 用户指令（user instruction），通常为任务描述。
input (str, 可选): 额外的输入内容（input text），可与 instruction 拼接。
output (str, 可选): 模型生成的目标输出（expected response）。
"""
...
def alpaca_converter(raw_sample: AlpacaSample) -> SFTSample:
"""将 Alpaca 样本转换为 SFT（Supervised Fine-Tuning）标准样本格式
`alpaca_converter` 将 Alpaca 数据集中一条样本转换为通用的 `SFTSample` 格式
该格式用于监督微调（SFT）或多轮对话建模
转换逻辑:
- 若存在 `system` 字段，则生成一条系统消息，loss_weight = 0.0
- 若存在 `instruction` 或 `input` 字段，则合并为一条用户消息，loss_weight = 0.0
- 若存在 `output` 字段，则生成一条助手机器人回复消息，loss_weight = 1.0
args:
raw_sample (AlpacaSample): 原始 Alpaca 数据样本
return:
SFTSample: 转换后的标准化样本，格式如下:
{
"messages": [
{"role": "system", "content": [{"type": "text", "value": "..."}], "loss_weight": 0.0},
{"role": "user", "content": [{"type": "text", "value": "..."}], "loss_weight": 0.0},
{"role": "assistant", "content": [{"type": "text", "value": "..."}], "loss_weight": 1.0},
]
}
example:
>>> raw = {"instruction": "请将以下句子翻译成英文：", "input": "你好", "output": "Hello"}
>>> alpaca_converter(raw)
{
"messages": [
{"role": "user", "content": [{"type": "text", "value": "请将以下句子翻译成英文：你好"}], "loss_weight": 0.0},
{"role": "assistant", "content": [{"type": "text", "value": "Hello"}], "loss_weight": 1.0}
]
}
"""
```
#### 2.3 转换过程
`alpaca_converter` 函数将 Alpaca 格式转换为标准格式，转换逻辑如下：
```python
def alpaca_converter(raw_sample: AlpacaSample) -> SFTSample:
messages = []
# 1. 添加系统提示词（如果存在）
if "system" in raw_sample:
messages.append({
"role": "system",
"content": [{"type": "text", "value": raw_sample["system"]}],
"loss_weight": 0.0
})
# 2. 添加用户输入（instruction + input）
if "instruction" in raw_sample or "input" in raw_sample:
user_content = raw_sample.get("instruction", "") + raw_sample.get("input", "")
messages.append({
"role": "user",
"content": [{"type": "text", "value": user_content}],
"loss_weight": 0.0
})
# 3. 添加模型回复
if "output" in raw_sample:
messages.append({
"role": "assistant",
"content": [{"type": "text", "value": raw_sample["output"]}],
"loss_weight": 1.0
})
return {"messages": messages}
```
#### 2.4 转换示例
**输入（Alpaca 格式）：**
```json
{
"instruction": "What is the capital of France?",
"input": "",
"output": "The capital of France is Paris."
}
```
**输出（标准格式）：**
```json
{
"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
}
]
}
```
### 3. 自定义转换器
#### 3.1 创建自定义转换器
如果用户有自己的数据格式，可以轻松添加自定义转换器将其标准化，实现过程可参考如下示例：
```python
# src/llamafactory/v1/plugins/data_plugins/converter.py
from typing import TypedDict, NotRequired
from ...extras.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. 注册 custom_converter
#src/llamafactory/v1/plugins/data_plugins/converter.py: CONVERTERS
CONVERTERS = {
"alpaca": alpaca_converter,
"custom": custom_converter, # 添加自定义转换器
}
```
#### 3.2 使用自定义转换器
在 YAML 配置中指定转换器名称：
```yaml
my_dataset:
file_name: custom_data.json
converter: custom
```
---
## DataLoaderPlugin
### 1. DataLoaderPlugin 简介
`DataLoaderPlugin` 负责从本地文件加载数据集，当前支持如下文件格式：
- **JSON**: `.json`
- **JSONL**: `.jsonl`
- **CSV**: `.csv`
- **Parquet**: `.parquet`
- **Arrow**: `.arrow`
- **Text**: `.txt`
### 2. DataLoaderPlugin 接口定义
```python
@dataclass
class DataLoaderPlugin:
"""数据加载插件（DataLoaderPlugin）
负责根据数据集信息（`DatasetInfo`）自动加载本地或远程数据集。
支持多种文件格式（如 CSV、JSON、Parquet、Text、Arrow），并可选择是否以流式方式加载。
通常由 `DataEngine` 调用，用于统一封装数据加载逻辑。
"""
args: DataArguments
"""数据参数对象，包含数据目录、缓存路径、分片等配置信息。"""
def _get_builder_name(self, path: str) -> Literal["arrow", "csv", "json", "parquet", "text"]:
"""获取数据集文件格式
根据输入文件路径自动判断应使用的 HuggingFace `load_dataset` 构建器类型。
通过文件扩展名推断数据类型，例如 `.csv`、`.jsonl`、`.parquet`、`.txt` 等。
args:
path (str): 数据集文件路径，用于识别文件类型。
return:
Literal["arrow", "csv", "json", "parquet", "text"]:
数据构建器名称，用于 `datasets.load_dataset()`。
example:
>>> _get_builder_name("data/train.jsonl")
"json"
"""
...
def auto_load_data(self, dataset_info: DatasetInfo) -> HFDataset:
"""根据传入的 `dataset_info` 自动选择合适的加载方式
args:
dataset_info (DatasetInfo): 数据集元信息，通常包含：
- `file_name`: 数据文件路径
- `split`: 数据划分（如 "train"、"test"）；
- `streaming`: 是否启用流式加载
return:
HFDataset: 加载完成的 Hugging Face 数据集对象。
example:
>>> plugin = DataLoaderPlugin(args)
>>> ds = plugin.auto_load_data({"file_name": "~/data.json", "split": "train"})
"""
...
def load_data_from_file(self, filepath: str, split: str, streaming: bool) -> HFDataset:
"""从文件或目录加载数据集
根据输入路径自动识别文件类型（CSV、JSON、Parquet、Text 等），
并通过 `datasets.load_dataset()` 加载数据集。
若 `streaming=True`，则将结果转换为迭代式数据集。
args:
filepath (str): 文件路径或目录路径。
split (str): 数据划分名称（如 "train"、"validation"）。
streaming (bool): 是否启用流式加载模式。
return:
HFDataset: 加载后的数据集对象。
example:
>>> plugin.load_data_from_file("data/train.json", "train", False)
"""
...
```
---
## DataIndexPlugin
### 1. DataIndexPlugin 简介
`DataIndexPlugin` 负责调整数据索引，支持通过配置 `size`, `weight` 等参数控制数据集样本数量和采样频率。
- 使用 `size` 参数 限制使用的样本数量：
```yaml
my_dataset:
file_name: large_dataset.json
size: 1000 # 只使用前 1000 个样本
```
- 使用 `weight` 参数调整数据集在混合数据中的采样频率：
```yaml
dataset_a:
file_name: data_a.json
weight: 1.0
dataset_b:
file_name: data_b.json
weight: 2.0 # dataset_b 的样本出现频率是 dataset_a 的 2 倍
```
**说明**：`weight` 参数适用于在多个数据集混合训练时，调整不同数据集的的采样频率
- 当 `weight=1.0` 时，数据集按原始比例采样
- 当 `weight=2.0` 时，该数据集的索引会复制 2 倍，使其样本出现频率翻倍
### 2. DataIndexPlugin 接口定义
```python
@dataclass
class DataIndexPlugin:
"""数据索引插件（DataIndexPlugin）
根据 `size` 和 `weight` 调整数据索引列表，控制数据集的样本数量和采样频率
通常在多数据集混合训练时使用，以控制不同数据集在总体样本中的占比。
在 `DataEngine.build_data_index` 中被自动调用，用于实现样本重采样或加权分布。
"""
def adjust_data_index(
self, data_index: list[tuple[str, int]], size: Optional[int], weight: Optional[float]
) -> list[tuple[str, int]]:
"""调整数据索引列表
根据 `size` 或 `weight` 参数对输入的数据索引进行采样、扩展或缩减。
若两个参数同时存在，将依次执行基于大小和基于权重的调整。
args:
data_index (list[tuple[str, int]]):
数据索引列表，每个元素为 `(dataset_name, sample_index)`。
size (Optional[int]):
目标样本数量，若指定则根据该数量裁剪或重复样本。
weight (Optional[float]):
数据集权重，用于控制数据集在混合训练中的采样比例。
return:
list[tuple[str, int]]:
调整后的数据索引列表。
example:
>>> plugin = DataIndexPlugin()
>>> adjusted = plugin.adjust_data_index([("ds1", i) for i in range(100)], size=50, weight=None)
>>> len(adjusted)
50
"""
...
def adjust_by_size(self, data_index: list[tuple[str, int]], size: int) -> list[tuple[str, int]]:
"""根据目标大小调整数据索引
通过裁剪或重复样本，使索引总数等于 `size`。
常用于统一不同数据集的样本数量。
args:
data_index (list[tuple[str, int]]):
原始数据索引列表。
size (int):
目标样本数量。
return:
list[tuple[str, int]]:
调整后长度等于 `size` 的数据索引列表。
example:
>>> plugin.adjust_by_size([("ds1", i) for i in range(10)], 20)
"""
...
def adjust_by_weight(self, data_index: list[tuple[str, int]], weight: float) -> list[tuple[str, int]]:
"""根据权重调整数据索引
通过加权采样或重复样本，使数据集样本出现频率符合指定权重。
常用于多数据源训练中按比例平衡样本。
args:
data_index (list[tuple[str, int]]):
原始数据索引列表。
weight (float):
数据集权重（相对比例，可与其他数据集共同归一化）。
return:
list[tuple[str, int]]:
调整后的加权数据索引列表。
example:
>>> plugin.adjust_by_weight([("ds1", i) for i in range(10)], 0.5)
"""
...
```
---
## DataSelectorPlugin
### 1. DataSelectorPlugin 简介
`DataSelectorPlugin` 为 `DataEngine`提供基于索引访问数据的功能，由 `DataEngine` 的 `__getitem__` 方法自动调用。
### 2. DataSelectorPlugin 接口定义
```python
@dataclass
class DataSelectorPlugin:
"""根据索引选择数据集样本。
配合 `DataEngine` 使用，通过统一的 `data_index` 结构（包含数据集名与样本索引）来实现灵活的数据选择
"""
data_index: list[tuple[str, int]]
"""数据索引列表，每个元素为 (dataset_name, sample_index)。"""
def select(self, index: Union[slice, list[int], Any]) -> Union[tuple[str, int], list[tuple[str, int]]]:
"""选择数据集样本
根据输入类型从 `data_index` 中选择对应的样本索引
支持三种索引方式：
- 切片（slice）：返回对应范围内的样本
- 索引列表（list[int]）：返回指定索引处的多个样本
- 其他类型输入将触发异常。
args:
index (Union[slice, list[int], Any]): 数据样本索引
可以是切片（`slice`）或索引列表
return:
Union[tuple[str, int], list[tuple[str, int]]]:
- 若为单个索引：返回一个 `(dataset_name, sample_index)`
- 若为多个索引或切片：返回多个样本的列表
except:
Raises:
ValueError: 当输入索引类型不受支持时抛出。
...
```
|