mirror of
https://github.com/hiyouga/LLaMA-Factory.git
synced 2026-02-25 23:36:02 +08:00
Compare commits
3 Commits
675ce8cc7f
...
f80e15dbb4
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f80e15dbb4 | ||
|
|
991267fd3b | ||
|
|
5c52afa30d |
1
.github/workflows/tests_cuda.yml
vendored
1
.github/workflows/tests_cuda.yml
vendored
@@ -61,6 +61,7 @@ jobs:
|
||||
uv venv
|
||||
uv pip install -e .
|
||||
uv pip install -r requirements/dev.txt
|
||||
uv pip install -r requirements/bitsandbytes.txt
|
||||
|
||||
- name: Check quality
|
||||
run: |
|
||||
|
||||
25
examples/v1/train_full/train_full_deepspeed.yaml
Normal file
25
examples/v1/train_full/train_full_deepspeed.yaml
Normal file
@@ -0,0 +1,25 @@
|
||||
model: Qwen/Qwen3-0.6B
|
||||
|
||||
model_class: llm
|
||||
|
||||
template: qwen3_nothink
|
||||
|
||||
kernel_config:
|
||||
name: auto
|
||||
include_kernels: auto
|
||||
|
||||
dist_config:
|
||||
name: deepspeed
|
||||
config_file: examples/deepspeed/ds_z3_config.json
|
||||
|
||||
### data
|
||||
train_dataset: data/v1_sft_demo.yaml
|
||||
|
||||
### training
|
||||
output_dir: outputs/Qwen3-0.6B-deepspeed
|
||||
micro_batch_size: 1
|
||||
cutoff_len: 2048
|
||||
learning_rate: 1.0e-4
|
||||
bf16: true
|
||||
max_steps: 10
|
||||
|
||||
43
examples/v1/train_qlora/quantization.yaml
Normal file
43
examples/v1/train_qlora/quantization.yaml
Normal file
@@ -0,0 +1,43 @@
|
||||
model: Qwen/Qwen3-0.6B
|
||||
trust_remote_code: true
|
||||
model_class: llm
|
||||
|
||||
template: qwen3_nothink
|
||||
|
||||
# PEFT Configuration
|
||||
peft_config:
|
||||
name: lora
|
||||
r: 16
|
||||
lora_alpha: 32
|
||||
lora_dropout: 0.05
|
||||
target_modules: all
|
||||
|
||||
# Kernel Config
|
||||
kernel_config:
|
||||
name: auto
|
||||
include_kernels: auto
|
||||
|
||||
# FSDP Config
|
||||
dist_config:
|
||||
name: fsdp2
|
||||
dcp_path: null
|
||||
|
||||
# Quantization Config
|
||||
quant_config:
|
||||
name: bnb # choice: auto/bnb if auto is selected, the quantization method will be automatically selected based on the model and environment.
|
||||
quantization_bit: 4 # choice: 8/4(bnb)
|
||||
|
||||
### data
|
||||
train_dataset: data/v1_sft_demo.yaml
|
||||
|
||||
### training
|
||||
output_dir: outputs/test_quantization
|
||||
micro_batch_size: 1
|
||||
cutoff_len: 2048
|
||||
learning_rate: 1.0e-4
|
||||
bf16: false
|
||||
max_steps: 10
|
||||
|
||||
### sample
|
||||
sample_backend: hf
|
||||
max_new_tokens: 128
|
||||
@@ -76,19 +76,28 @@ class BaseTrainer:
|
||||
if self.args.enable_activation_checkpointing:
|
||||
self.model.gradient_checkpointing_enable({"use_reentrant": False})
|
||||
|
||||
if self.args.dist_config is not None:
|
||||
shard_need_optimizer = self.args.dist_config.name == "deepspeed"
|
||||
else:
|
||||
shard_need_optimizer = False
|
||||
self._accelerate_engine = None
|
||||
dist_name = self.args.dist_config.name if self.args.dist_config is not None else None
|
||||
|
||||
if shard_need_optimizer:
|
||||
if dist_name == "deepspeed":
|
||||
from ..plugins.trainer_plugins.distributed.hub import DistributedPlugin
|
||||
|
||||
self._deepspeed_engine = DistributedPlugin("deepspeed")(
|
||||
self.model,
|
||||
self.args.dist_config,
|
||||
num_micro_batch=self.train_batch_generator.num_micro_batch,
|
||||
micro_batch_size=self.args.micro_batch_size,
|
||||
)
|
||||
self._init_optimizer()
|
||||
self._shard_model()
|
||||
self._init_lr_scheduler()
|
||||
self.model, self.optimizer, self.lr_scheduler = self._deepspeed_engine.prepare(
|
||||
self.model, self.optimizer, self.lr_scheduler
|
||||
)
|
||||
else:
|
||||
# fsdp2 / DDP / no dist
|
||||
self._shard_model()
|
||||
self._init_optimizer()
|
||||
|
||||
self._init_lr_scheduler()
|
||||
self._init_lr_scheduler()
|
||||
|
||||
def _create_batch_generator(self) -> None:
|
||||
self.train_batch_generator = BatchGenerator(
|
||||
@@ -171,25 +180,35 @@ class BaseTrainer:
|
||||
step_loss = 0
|
||||
step_valid_tokens = compute_valid_tokens(micro_batches)
|
||||
step_valid_tokens = DistributedInterface().all_reduce(step_valid_tokens, op=ReduceOp.SUM)
|
||||
for micro_batch in micro_batches:
|
||||
num_micro = len(micro_batches)
|
||||
for i, micro_batch in enumerate(micro_batches):
|
||||
loss = self.compute_loss(micro_batch)
|
||||
mini_step_valid_tokens = compute_valid_tokens([micro_batch])
|
||||
# fsdp uses mean reduction so we need to scale the loss by dp_size
|
||||
loss = loss * mini_step_valid_tokens * self.dp_size / (step_valid_tokens + 1e-6)
|
||||
|
||||
loss.backward()
|
||||
if self._deepspeed_engine is not None:
|
||||
# deepspeed: set sync_gradients so engine.step() only fires on last micro-batch
|
||||
self._deepspeed_engine.accelerator.sync_gradients = i == num_micro - 1
|
||||
self._deepspeed_engine.backward(loss)
|
||||
else:
|
||||
loss.backward()
|
||||
step_loss += loss.item()
|
||||
|
||||
grad_norm = torch.nn.utils.clip_grad_norm_(self.model.parameters(), self.args.max_grad_norm).item()
|
||||
|
||||
# isfinite(): argument 'input' (position 1) must be Tensor, not float
|
||||
if not torch.isfinite(torch.tensor(grad_norm)): # type: ignore # pyright: ignore [reportUnknownReturnType]
|
||||
logger.warning_rank0(f"Gradient norm is not finite: {grad_norm}")
|
||||
if self._deepspeed_engine is not None:
|
||||
# deepspeed: engine.step() already ran inside backward at the sync boundary
|
||||
grad_norm = self._deepspeed_engine.get_grad_norm()
|
||||
else:
|
||||
self.optimizer.step()
|
||||
grad_norm = torch.nn.utils.clip_grad_norm_(self.model.parameters(), self.args.max_grad_norm).item()
|
||||
|
||||
self.lr_scheduler.step()
|
||||
self.optimizer.zero_grad()
|
||||
# isfinite(): argument 'input' (position 1) must be Tensor, not float
|
||||
if not torch.isfinite(torch.tensor(grad_norm)): # type: ignore # pyright: ignore [reportUnknownReturnType]
|
||||
logger.warning_rank0(f"Gradient norm is not finite: {grad_norm}")
|
||||
else:
|
||||
self.optimizer.step()
|
||||
|
||||
self.lr_scheduler.step()
|
||||
self.optimizer.zero_grad()
|
||||
|
||||
step_loss, grad_norm = DistributedInterface().all_reduce([step_loss, grad_norm])
|
||||
DistributedInterface().sync()
|
||||
@@ -203,17 +222,14 @@ class BaseTrainer:
|
||||
|
||||
def save_model(self) -> None:
|
||||
"""Save the model."""
|
||||
model_to_save = self.model.module if hasattr(self.model, "module") else self.model
|
||||
state_dict = None
|
||||
if self.args.dist_config is not None and self.args.dist_config.name == "fsdp2":
|
||||
from torch.distributed.checkpoint.state_dict import StateDictOptions, get_model_state_dict
|
||||
if self.args.dist_config is not None and self.args.dist_config.name in ("deepspeed", "fsdp2"):
|
||||
from ..plugins.trainer_plugins.distributed.hub import DistributedPlugin
|
||||
|
||||
options = StateDictOptions(full_state_dict=True, cpu_offload=True)
|
||||
state_dict = get_model_state_dict(self.model, options=options)
|
||||
|
||||
if DistributedInterface().get_rank() != 0:
|
||||
return
|
||||
|
||||
model_to_save.save_pretrained(self.args.output_dir, state_dict=state_dict)
|
||||
self.renderer.processor.save_pretrained(self.args.output_dir)
|
||||
logger.info_rank0(f"Model saved to {self.args.output_dir}")
|
||||
DistributedPlugin(self.args.dist_config.name).save_model(
|
||||
self.model, self.args.output_dir, self.renderer.processor
|
||||
)
|
||||
else:
|
||||
model_to_save = self.model.module if hasattr(self.model, "module") else self.model
|
||||
model_to_save.save_pretrained(self.args.output_dir, max_shard_size="4GB")
|
||||
self.renderer.processor.save_pretrained(self.args.output_dir, max_shard_size="4GB")
|
||||
logger.info_rank0(f"Model saved to {self.args.output_dir}")
|
||||
|
||||
@@ -90,6 +90,26 @@ class ModelEngine:
|
||||
Transformers can choose the proper model init context.
|
||||
https://github.com/huggingface/transformers/blob/v5.0.0rc0/src/transformers/modeling_utils.py#L3538
|
||||
"""
|
||||
if self.args.init_config is not None:
|
||||
from ..plugins.model_plugins.initialization import InitPlugin
|
||||
|
||||
init_device = InitPlugin(self.args.init_config.name)()
|
||||
else:
|
||||
init_device = DistributedInterface().current_device
|
||||
|
||||
init_kwargs = {"device_map": init_device}
|
||||
|
||||
if self.args.quant_config is not None:
|
||||
from ..plugins.model_plugins.quantization import QuantizationPlugin
|
||||
|
||||
init_kwargs = QuantizationPlugin(self.args.quant_config.name)(
|
||||
init_kwargs=init_kwargs,
|
||||
config=self.model_config,
|
||||
tokenizer=self.processor,
|
||||
model_args=self.args,
|
||||
is_trainable=self.is_train,
|
||||
)
|
||||
|
||||
if self.args.model_class == ModelClass.LLM:
|
||||
from transformers import AutoModelForCausalLM, AutoModelForImageTextToText
|
||||
|
||||
@@ -107,14 +127,8 @@ class ModelEngine:
|
||||
|
||||
AutoClass = AutoModel
|
||||
|
||||
if self.args.init_config is not None:
|
||||
from ..plugins.model_plugins.initialization import InitPlugin
|
||||
|
||||
init_device = InitPlugin(self.args.init_config.name)()
|
||||
else:
|
||||
init_device = DistributedInterface().current_device
|
||||
|
||||
if init_device.type == DeviceType.META:
|
||||
assert self.args.quant_config is None, "Quantization is not supported with meta device."
|
||||
with init_empty_weights():
|
||||
model = AutoClass.from_config(self.model_config)
|
||||
else:
|
||||
@@ -122,8 +136,8 @@ class ModelEngine:
|
||||
self.args.model,
|
||||
config=self.model_config,
|
||||
dtype="auto",
|
||||
device_map=init_device,
|
||||
trust_remote_code=self.args.trust_remote_code,
|
||||
**init_kwargs,
|
||||
)
|
||||
|
||||
if self.args.peft_config is None:
|
||||
|
||||
@@ -0,0 +1,122 @@
|
||||
# Copyright 2025 HuggingFace Inc., the KVCache.AI team, Approaching AI, and the LlamaFactory team.
|
||||
#
|
||||
# This code is inspired by the HuggingFace's transformers library.
|
||||
# https://github.com/huggingface/transformers/blob/v4.40.0/examples/pytorch/language-modeling/run_clm.py
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
import torch
|
||||
from transformers import BitsAndBytesConfig
|
||||
|
||||
from ...accelerator.helper import get_current_device
|
||||
from ...config.model_args import ModelArguments
|
||||
from ...utils import logging
|
||||
from ...utils.packages import check_version
|
||||
from ...utils.plugin import BasePlugin
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from transformers import PretrainedConfig, PreTrainedTokenizer
|
||||
|
||||
logger = logging.get_logger(__name__)
|
||||
|
||||
|
||||
class QuantizationPlugin(BasePlugin):
|
||||
r"""Plugin for model quantization."""
|
||||
|
||||
def __call__(
|
||||
self,
|
||||
init_kwargs: dict[str, Any] = None,
|
||||
config: "PretrainedConfig" = None,
|
||||
tokenizer: "PreTrainedTokenizer" = None,
|
||||
model_args: "ModelArguments" = None,
|
||||
is_trainable: bool = False,
|
||||
) -> dict[str, Any]:
|
||||
return super().__call__(
|
||||
init_kwargs, config=config, tokenizer=tokenizer, model_args=model_args, is_trainable=is_trainable
|
||||
)
|
||||
|
||||
|
||||
@QuantizationPlugin("auto").register()
|
||||
def quantization_auto(
|
||||
init_kwargs: dict[str, Any],
|
||||
**kwargs,
|
||||
) -> dict[str, Any]:
|
||||
"""Automatic quantization selection, only support bnb currently.
|
||||
|
||||
Args:
|
||||
init_kwargs (dict[str, Any]): The kwargs for model initialization.
|
||||
**kwargs: Keyword arguments containing the model.
|
||||
|
||||
Returns:
|
||||
dict[str, Any]: The updated kwargs for model initialization.
|
||||
"""
|
||||
model_args: ModelArguments = kwargs.get("model_args", None)
|
||||
quant_config = model_args.quant_config
|
||||
|
||||
quantization_bit = quant_config.get("quantization_bit", None)
|
||||
if quantization_bit is not None:
|
||||
logger.info_rank0(f"Loading {quantization_bit}-bit quantized model.")
|
||||
if quantization_bit in [8, 4]:
|
||||
return quantization_with_bnb(init_kwargs, **kwargs)
|
||||
else:
|
||||
raise ValueError(f"Unsupported quantization bit: {quantization_bit} for auto quantization.")
|
||||
logger.warning_rank0("No quantization method applied.")
|
||||
return init_kwargs
|
||||
|
||||
|
||||
@QuantizationPlugin("bnb").register()
|
||||
def quantization_with_bnb(
|
||||
init_kwargs: dict[str, Any],
|
||||
model_args: "ModelArguments" = None,
|
||||
**kwargs,
|
||||
) -> dict[str, Any]:
|
||||
r"""Quantization with BNB."""
|
||||
logger.info_rank0("Using Bitsandbytes quantization.")
|
||||
quantization_bit = model_args.quant_config.get("quantization_bit", None)
|
||||
if quantization_bit is None:
|
||||
logger.warning_rank0("quantization_bit is not specified, default to 8-bit quantization.")
|
||||
quantization_bit = 4
|
||||
assert quantization_bit in [8, 4], "Bitsandbytes only accepts 4-bit or 8-bit quantization."
|
||||
if quantization_bit == 8:
|
||||
check_version("bitsandbytes>=0.37.0", mandatory=True)
|
||||
init_kwargs["quantization_config"] = BitsAndBytesConfig(load_in_8bit=True)
|
||||
elif quantization_bit == 4:
|
||||
check_version("bitsandbytes>=0.39.0", mandatory=True)
|
||||
init_kwargs["quantization_config"] = BitsAndBytesConfig(
|
||||
load_in_4bit=True,
|
||||
bnb_4bit_compute_dtype=model_args.quant_config.get("compute_dtype", torch.float16),
|
||||
bnb_4bit_use_double_quant=model_args.quant_config.get("double_quantization", True),
|
||||
bnb_4bit_quant_type=model_args.quant_config.get("quantization_type", "nf4"),
|
||||
bnb_4bit_quant_storage=model_args.quant_config.get(
|
||||
"compute_dtype", torch.float16
|
||||
), # crucial for fsdp+qlora
|
||||
)
|
||||
else:
|
||||
raise ValueError("Bitsandbytes only accepts 4-bit or 8-bit quantization.")
|
||||
|
||||
# TODO: improve deepspeed zero3 and fsdp detection.
|
||||
if kwargs.get("is_trainable", False):
|
||||
logger.info_rank0("Detected inference mode, setting device_map for bitsandbytes quantization.")
|
||||
init_kwargs["device_map"] = {"": get_current_device()} # change auto device map for inference
|
||||
else:
|
||||
logger.info_rank0("Detected training mode, skip setting device_map for bitsandbytes quantization.")
|
||||
if model_args.quant_config.get("quantization_bit") != 4:
|
||||
raise ValueError("Only 4-bit quantized model can use fsdp+qlora or auto device map.")
|
||||
|
||||
check_version("bitsandbytes>=0.43.0", mandatory=True)
|
||||
|
||||
logger.info_rank0(f"Quantizing model to {model_args.quant_config.get('quantization_bit')} bit with bitsandbytes.")
|
||||
return init_kwargs
|
||||
|
||||
@@ -0,0 +1,129 @@
|
||||
# Copyright 2025 the LlamaFactory team.
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
"""DeepSpeed integration via accelerate's built-in capabilities.
|
||||
|
||||
Instead of manually calling deepspeed.initialize() and syncing config,
|
||||
this module leverages accelerate's Accelerator + DeepSpeedPlugin to handle
|
||||
initialization, backward, gradient accumulation, and model saving.
|
||||
"""
|
||||
|
||||
from typing import Any, Optional
|
||||
|
||||
import torch
|
||||
from accelerate import Accelerator
|
||||
from accelerate.utils import DeepSpeedPlugin
|
||||
|
||||
from ....utils.logging import get_logger
|
||||
from ....utils.types import HFModel, Processor
|
||||
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
class DeepSpeedEngine:
|
||||
"""DeepSpeed integration using accelerate's built-in capabilities.
|
||||
|
||||
This replaces the manual DeepSpeedConfigHelper / DeepSpeedEngine approach
|
||||
with accelerate's Accelerator + DeepSpeedPlugin, which handles:
|
||||
- Config syncing (auto values, batch size, lr, etc.)
|
||||
- deepspeed.initialize() call
|
||||
- Optimizer / LR scheduler wrapping
|
||||
- Backward + gradient accumulation boundary
|
||||
- ZeRO-3 parameter gathering for saving
|
||||
"""
|
||||
|
||||
def __init__(self, dist_config: dict[str, Any], num_micro_batch: int = 1, micro_batch_size: int = 1):
|
||||
config_file = dist_config.get("config_file")
|
||||
if not config_file:
|
||||
raise ValueError("DeepSpeed config_file is required in dist_config")
|
||||
|
||||
ds_plugin = DeepSpeedPlugin(hf_ds_config=config_file)
|
||||
|
||||
self.accelerator = Accelerator(
|
||||
deepspeed_plugin=ds_plugin,
|
||||
gradient_accumulation_steps=num_micro_batch,
|
||||
)
|
||||
|
||||
# Resolve "auto" for train_micro_batch_size_per_gpu so that
|
||||
# accelerate.prepare() does not require a DataLoader to infer it.
|
||||
ds_config = self.accelerator.state.deepspeed_plugin.deepspeed_config
|
||||
if ds_config.get("train_micro_batch_size_per_gpu") in (None, "auto"):
|
||||
ds_config["train_micro_batch_size_per_gpu"] = micro_batch_size
|
||||
|
||||
logger.info_rank0(f"DeepSpeedEngine initialized with config: {config_file}")
|
||||
|
||||
def shard_model(self, model: HFModel) -> "DeepSpeedEngine":
|
||||
"""No-op shard — actual model wrapping happens in prepare().
|
||||
|
||||
Returns self so the caller gets the engine instance via the hub interface.
|
||||
"""
|
||||
return self
|
||||
|
||||
def prepare(
|
||||
self,
|
||||
model: HFModel,
|
||||
optimizer: torch.optim.Optimizer,
|
||||
lr_scheduler: Optional[Any] = None,
|
||||
) -> tuple[HFModel, torch.optim.Optimizer, Any]:
|
||||
"""Prepare model, optimizer, and lr_scheduler using accelerate.
|
||||
|
||||
Internally calls deepspeed.initialize() and wraps the returned objects.
|
||||
"""
|
||||
if lr_scheduler is not None:
|
||||
model, optimizer, lr_scheduler = self.accelerator.prepare(model, optimizer, lr_scheduler)
|
||||
else:
|
||||
model, optimizer = self.accelerator.prepare(model, optimizer)
|
||||
|
||||
model._accelerator = self.accelerator # type: ignore[assignment]
|
||||
|
||||
logger.info_rank0("Model, optimizer, and lr_scheduler prepared via accelerate")
|
||||
return model, optimizer, lr_scheduler
|
||||
|
||||
def backward(self, loss: torch.Tensor) -> None:
|
||||
"""Backward pass using accelerate.
|
||||
|
||||
Delegates to DeepSpeedEngineWrapper.backward() which respects
|
||||
sync_gradients to control gradient accumulation boundaries.
|
||||
When sync_gradients=True: engine.backward(loss) + engine.step()
|
||||
When sync_gradients=False: engine.backward(loss) only
|
||||
"""
|
||||
self.accelerator.backward(loss)
|
||||
|
||||
def get_grad_norm(self) -> float:
|
||||
"""Get the global gradient norm from the DeepSpeed engine."""
|
||||
engine_wrapper = getattr(self.accelerator, "deepspeed_engine_wrapped", None)
|
||||
if engine_wrapper is not None:
|
||||
return engine_wrapper.engine.get_global_grad_norm() or 0.0
|
||||
return 0.0
|
||||
|
||||
|
||||
def save_model(model: HFModel, output_dir: str, processor: Processor) -> None:
|
||||
"""Save model using accelerate's built-in ZeRO-aware utilities.
|
||||
|
||||
Expects model._accelerator to be set during prepare().
|
||||
Handles ZeRO-3 parameter gathering automatically via
|
||||
accelerator.get_state_dict().
|
||||
"""
|
||||
accelerator: Accelerator = model._accelerator # type: ignore[union-attr]
|
||||
|
||||
unwrapped_model = accelerator.unwrap_model(model)
|
||||
state_dict = accelerator.get_state_dict(model)
|
||||
|
||||
if accelerator.is_main_process:
|
||||
unwrapped_model.save_pretrained(output_dir, state_dict=state_dict, max_shard_size="4GB")
|
||||
processor.save_pretrained(output_dir, max_shard_size="4GB")
|
||||
|
||||
accelerator.wait_for_everyone()
|
||||
logger.info_rank0(f"Model saved to {output_dir}")
|
||||
|
||||
@@ -17,24 +17,24 @@ import os
|
||||
|
||||
import torch
|
||||
import torch.nn as nn
|
||||
from peft.tuners.lora import LoraLayer
|
||||
from torch.distributed.checkpoint.state_dict import StateDictOptions, get_model_state_dict, set_model_state_dict
|
||||
from torch.distributed.fsdp import (
|
||||
CPUOffloadPolicy,
|
||||
MixedPrecisionPolicy,
|
||||
fully_shard,
|
||||
)
|
||||
from transformers import PreTrainedModel
|
||||
from peft.tuners.lora import LoraLayer
|
||||
|
||||
from ....accelerator.helper import get_current_accelerator
|
||||
from ....accelerator.interface import DistributedInterface
|
||||
from ....utils.logging import get_logger
|
||||
from ....utils.types import HFModel, Processor
|
||||
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
def get_transformer_layer_cls(model: PreTrainedModel) -> type[nn.Module] | None:
|
||||
def get_transformer_layer_cls(model: HFModel) -> type[nn.Module] | None:
|
||||
no_split_modules = getattr(model, "_no_split_modules", None)
|
||||
if no_split_modules:
|
||||
if isinstance(no_split_modules, (list, tuple)):
|
||||
@@ -50,6 +50,20 @@ def get_transformer_layer_cls(model: PreTrainedModel) -> type[nn.Module] | None:
|
||||
return None
|
||||
|
||||
|
||||
def save_model(model: HFModel, output_dir: str, processor: Processor) -> None:
|
||||
if DistributedInterface().get_rank() == 0:
|
||||
logger.info("Gathering state dict for saving...")
|
||||
|
||||
options = StateDictOptions(full_state_dict=True, cpu_offload=True)
|
||||
state_dict = get_model_state_dict(model, options=options)
|
||||
|
||||
if DistributedInterface().get_rank() == 0:
|
||||
model_to_save = model.module if hasattr(model, "module") else model
|
||||
model_to_save.save_pretrained(output_dir, state_dict=state_dict, max_shard_size="4GB")
|
||||
processor.save_pretrained(output_dir, max_shard_size="4GB")
|
||||
logger.info(f"Model saved to {output_dir}")
|
||||
|
||||
|
||||
class FSDP2Engine:
|
||||
def __init__(self, dist_config: dict):
|
||||
self.dist_interface = DistributedInterface()
|
||||
@@ -94,12 +108,11 @@ class FSDP2Engine:
|
||||
reduce_dtype=reduce_dtype,
|
||||
cast_forward_inputs=True,
|
||||
)
|
||||
|
||||
|
||||
def is_lora_module_wrap(self, model) -> bool:
|
||||
return any(isinstance(module, LoraLayer) for module in model.modules())
|
||||
|
||||
def prepare_model(self, model: PreTrainedModel) -> PreTrainedModel:
|
||||
def prepare_model(self, model: HFModel) -> HFModel:
|
||||
if self.fsdp_mesh is None:
|
||||
logger.warning("No FSDP Mesh available, skipping FSDP wrapping.")
|
||||
return model
|
||||
@@ -115,11 +128,10 @@ class FSDP2Engine:
|
||||
else:
|
||||
logger.info(f"Applying per-layer FSDP to {layer_cls.__name__}")
|
||||
transformer_layer_cls_to_wrap = {layer_cls}
|
||||
|
||||
|
||||
if self.is_lora_module_wrap(model):
|
||||
lora_modules = []
|
||||
for module in model.modules():
|
||||
|
||||
if len(list(module.children())) != 0:
|
||||
continue
|
||||
if any(param.requires_grad for param in module.parameters(recurse=False)):
|
||||
@@ -134,7 +146,7 @@ class FSDP2Engine:
|
||||
offload_policy=CPUOffloadPolicy(pin_memory=self.pin_memory) if self.offload_params else None,
|
||||
)
|
||||
|
||||
logger.info(f"Applying FSDP wrap for LoRA layer separately.")
|
||||
logger.info("Applying FSDP wrap for LoRA layer separately.")
|
||||
|
||||
for name, module in model.named_modules():
|
||||
should_wrap = False
|
||||
@@ -179,8 +191,9 @@ class FSDP2Engine:
|
||||
)
|
||||
|
||||
return model
|
||||
|
||||
@torch.no_grad()
|
||||
def materialize_and_load(self, model: PreTrainedModel, hf_model_path: str, dcp_path: str = None):
|
||||
def materialize_and_load(self, model: HFModel, hf_model_path: str, dcp_path: str = None):
|
||||
if self.rank == 0:
|
||||
logger.info("Materializing sharded model params...")
|
||||
|
||||
@@ -200,7 +213,7 @@ class FSDP2Engine:
|
||||
|
||||
return model
|
||||
|
||||
def shard_model(self, model: PreTrainedModel) -> PreTrainedModel:
|
||||
def shard_model(self, model: HFModel) -> HFModel:
|
||||
if model.device.type == "meta":
|
||||
model = self.prepare_model(model)
|
||||
model = self.materialize_and_load(model, hf_model_path=model.config.name_or_path, dcp_path=self.dcp_path)
|
||||
@@ -208,7 +221,7 @@ class FSDP2Engine:
|
||||
model = self.prepare_model(model)
|
||||
return model
|
||||
|
||||
def _load_from_dcp(self, model: PreTrainedModel, dcp_path: str):
|
||||
def _load_from_dcp(self, model: HFModel, dcp_path: str):
|
||||
import torch.distributed.checkpoint as dcp
|
||||
|
||||
try:
|
||||
@@ -227,7 +240,7 @@ class FSDP2Engine:
|
||||
logger.error(f"Failed to load from DCP: {e}")
|
||||
raise e
|
||||
|
||||
def _load_weights_from_hf_checkpoint(self, model, hf_model_path):
|
||||
def _load_weights_from_hf_checkpoint(self, model: HFModel, hf_model_path: str):
|
||||
import glob
|
||||
import json
|
||||
|
||||
|
||||
@@ -12,9 +12,16 @@
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from ....config.arg_utils import PluginConfig
|
||||
from ....utils.plugin import BasePlugin
|
||||
from ....utils.types import HFModel
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from ....utils.types import HFModel, Processor
|
||||
|
||||
|
||||
class DistributedPlugin(BasePlugin):
|
||||
@@ -23,12 +30,32 @@ class DistributedPlugin(BasePlugin):
|
||||
|
||||
|
||||
@DistributedPlugin("fsdp2").register()
|
||||
def shard_model_fsdp2(model: HFModel, dist_config: PluginConfig) -> HFModel:
|
||||
def shard_model_fsdp2(model: HFModel, dist_config: PluginConfig, **kwargs) -> HFModel:
|
||||
from .fsdp2 import FSDP2Engine
|
||||
|
||||
return FSDP2Engine(dist_config).shard_model(model)
|
||||
|
||||
|
||||
@DistributedPlugin("fsdp2").register("save_model")
|
||||
def save_model_fsdp2(model: HFModel, output_dir: str, processor: Processor) -> None:
|
||||
from .fsdp2 import save_model
|
||||
|
||||
return save_model(model, output_dir, processor)
|
||||
|
||||
|
||||
@DistributedPlugin("deepspeed").register()
|
||||
def shard_model_deepspeed(model: HFModel, dist_config: PluginConfig) -> HFModel:
|
||||
return model
|
||||
def shard_model_deepspeed(model: HFModel, dist_config: PluginConfig, **kwargs) -> HFModel:
|
||||
from .deepspeed import DeepSpeedEngine
|
||||
|
||||
return DeepSpeedEngine(
|
||||
dist_config,
|
||||
num_micro_batch=kwargs.get("num_micro_batch"),
|
||||
micro_batch_size=kwargs.get("micro_batch_size"),
|
||||
).shard_model(model)
|
||||
|
||||
|
||||
@DistributedPlugin("deepspeed").register("save_model")
|
||||
def save_model_deepspeed(model: HFModel, output_dir: str, processor: Processor) -> None:
|
||||
from .deepspeed import save_model
|
||||
|
||||
return save_model(model, output_dir, processor)
|
||||
|
||||
@@ -21,6 +21,13 @@ from functools import lru_cache
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from packaging import version
|
||||
from transformers.utils.versions import require_version
|
||||
|
||||
from . import logging
|
||||
from .env import is_env_enabled
|
||||
|
||||
|
||||
logger = logging.get_logger(__name__)
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
@@ -41,3 +48,22 @@ def _get_package_version(name: str) -> "Version":
|
||||
@lru_cache
|
||||
def is_transformers_version_greater_than(content: str):
|
||||
return _get_package_version("transformers") >= version.parse(content)
|
||||
|
||||
|
||||
def check_version(requirement: str, mandatory: bool = False) -> None:
|
||||
r"""Optionally check the package version."""
|
||||
if is_env_enabled("DISABLE_VERSION_CHECK") and not mandatory:
|
||||
logger.warning_rank0_once("Version checking has been disabled, may lead to unexpected behaviors.")
|
||||
return
|
||||
|
||||
if "gptqmodel" in requirement or "autoawq" in requirement:
|
||||
pip_command = f"pip install {requirement} --no-build-isolation"
|
||||
else:
|
||||
pip_command = f"pip install {requirement}"
|
||||
|
||||
if mandatory:
|
||||
hint = f"To fix: run `{pip_command}`."
|
||||
else:
|
||||
hint = f"To fix: run `{pip_command}` or set `DISABLE_VERSION_CHECK=1` to skip this check."
|
||||
|
||||
require_version(requirement, hint)
|
||||
|
||||
@@ -166,3 +166,33 @@ def _manage_distributed_env(request: FixtureRequest, monkeypatch: MonkeyPatch) -
|
||||
def fix_valuehead_cpu_loading():
|
||||
"""Fix valuehead model loading."""
|
||||
patch_valuehead_model()
|
||||
|
||||
|
||||
@pytest.fixture(scope="session", autouse=True)
|
||||
def bypass_mistral_regex_check():
|
||||
"""Disable Mistral regex network check.
|
||||
|
||||
Monkey-patch TokenizersBackend._patch_mistral_regex into a no-op.
|
||||
"""
|
||||
try:
|
||||
from transformers.tokenization_utils_fast import TokenizersBackend
|
||||
except ImportError:
|
||||
# Very old transformers, nothing to patch
|
||||
yield
|
||||
return
|
||||
|
||||
if not hasattr(TokenizersBackend, "_patch_mistral_regex"):
|
||||
# Method does not exist in this version
|
||||
yield
|
||||
return
|
||||
|
||||
# Backup original method
|
||||
original = TokenizersBackend._patch_mistral_regex
|
||||
|
||||
# Replace with no-op
|
||||
TokenizersBackend._patch_mistral_regex = lambda cls, tokenizer, *args, **kwargs: tokenizer
|
||||
|
||||
yield
|
||||
|
||||
# Restore original method
|
||||
TokenizersBackend._patch_mistral_regex = original
|
||||
|
||||
@@ -172,3 +172,33 @@ def _manage_distributed_env(request: FixtureRequest, monkeypatch: MonkeyPatch) -
|
||||
monkeypatch.setattr(torch.cuda, "device_count", lambda: 1)
|
||||
elif CURRENT_DEVICE == "npu":
|
||||
monkeypatch.setattr(torch.npu, "device_count", lambda: 1)
|
||||
|
||||
|
||||
@pytest.fixture(scope="session", autouse=True)
|
||||
def bypass_mistral_regex_check():
|
||||
"""Disable Mistral regex network check.
|
||||
|
||||
Monkey-patch TokenizersBackend._patch_mistral_regex into a no-op.
|
||||
"""
|
||||
try:
|
||||
from transformers.tokenization_utils_fast import TokenizersBackend
|
||||
except ImportError:
|
||||
# Very old transformers, nothing to patch
|
||||
yield
|
||||
return
|
||||
|
||||
if not hasattr(TokenizersBackend, "_patch_mistral_regex"):
|
||||
# Method does not exist in this version
|
||||
yield
|
||||
return
|
||||
|
||||
# Backup original method
|
||||
original = TokenizersBackend._patch_mistral_regex
|
||||
|
||||
# Replace with no-op
|
||||
TokenizersBackend._patch_mistral_regex = lambda cls, tokenizer, *args, **kwargs: tokenizer
|
||||
|
||||
yield
|
||||
|
||||
# Restore original method
|
||||
TokenizersBackend._patch_mistral_regex = original
|
||||
|
||||
51
tests_v1/plugins/model_plugins/test_quantization_plugin.py
Normal file
51
tests_v1/plugins/model_plugins/test_quantization_plugin.py
Normal file
@@ -0,0 +1,51 @@
|
||||
# Copyright 2025 the LlamaFactory team.
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
import pytest
|
||||
|
||||
from llamafactory.v1.config.model_args import ModelArguments
|
||||
from llamafactory.v1.core.model_engine import ModelEngine
|
||||
|
||||
|
||||
bitsandbytes = pytest.importorskip("bitsandbytes")
|
||||
|
||||
|
||||
def check_quantization_status(model):
|
||||
quantized_info = {"bnb": []}
|
||||
|
||||
for name, module in model.named_modules():
|
||||
# check BitsAndBytes quantization
|
||||
if isinstance(module, bitsandbytes.nn.modules.Linear8bitLt) or isinstance(
|
||||
module, bitsandbytes.nn.modules.Linear4bit
|
||||
):
|
||||
quantized_info["bnb"].append(name)
|
||||
|
||||
return quantized_info
|
||||
|
||||
|
||||
@pytest.mark.runs_on(["cuda"])
|
||||
@pytest.mark.parametrize("name, quantization_bit", [("bnb", 4), ("auto", 4)])
|
||||
def test_quantization_plugin(name, quantization_bit):
|
||||
model_args = ModelArguments(
|
||||
model="llamafactory/tiny-random-qwen3",
|
||||
quant_config={
|
||||
"name": name,
|
||||
"quantization_bit": quantization_bit,
|
||||
},
|
||||
)
|
||||
|
||||
model_engine = ModelEngine(model_args=model_args)
|
||||
quantized_info = check_quantization_status(model_engine.model)
|
||||
print(f"Quantized weights for method {name} with {quantization_bit} bit: {quantized_info}")
|
||||
assert any(v for v in quantized_info.values()), "model is not quantized properly."
|
||||
Reference in New Issue
Block a user