Compare commits
22 Commits
5306c6acbb
...
llmcompres
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b708a1cc45 | ||
|
|
daa9a58f83 | ||
|
|
ae7069e15b | ||
|
|
20d48cd617 | ||
|
|
e766a730ba | ||
|
|
7dc797860e | ||
|
|
ff4904c8c4 | ||
|
|
45b7293793 | ||
|
|
279c7178bc | ||
|
|
e73c3709f9 | ||
|
|
33562189f8 | ||
|
|
c057a2268f | ||
|
|
9d7a3809b5 | ||
|
|
b7b24d6a64 | ||
|
|
8b82b8f7a1 | ||
|
|
81da58c0a1 | ||
|
|
2cd5a234a7 | ||
|
|
8c1af0747d | ||
|
|
a06b360d99 | ||
|
|
0f6456a14f | ||
|
|
47a333ce49 | ||
|
|
f9d6776c28 |
@@ -1,7 +1,5 @@
|
|||||||
codecov:
|
codecov:
|
||||||
require_ci_to_pass: yes
|
require_ci_to_pass: yes
|
||||||
notify:
|
|
||||||
wait_for_ci: true
|
|
||||||
|
|
||||||
coverage:
|
coverage:
|
||||||
precision: 2
|
precision: 2
|
||||||
|
|||||||
@@ -49,7 +49,8 @@ sections = [
|
|||||||
("Knowledge Distillation (KD)", "kd"),
|
("Knowledge Distillation (KD)", "kd"),
|
||||||
("Liger Kernels", "liger"),
|
("Liger Kernels", "liger"),
|
||||||
("Language Model Evaluation Harness (LM Eval)", "lm_eval"),
|
("Language Model Evaluation Harness (LM Eval)", "lm_eval"),
|
||||||
("Spectrum", "spectrum")
|
("Spectrum", "spectrum"),
|
||||||
|
("LLMCompressor", "llm_compressor")
|
||||||
]
|
]
|
||||||
|
|
||||||
for section_name, folder_name in sections:
|
for section_name, folder_name in sections:
|
||||||
|
|||||||
77
examples/llama-3/sparse-finetuning.yaml
Normal file
77
examples/llama-3/sparse-finetuning.yaml
Normal file
@@ -0,0 +1,77 @@
|
|||||||
|
base_model: neuralmagic/Sparse-Llama-3.1-8B-2of4
|
||||||
|
|
||||||
|
plugins:
|
||||||
|
- axolotl.integrations.llm_compressor.LLMCompressorPlugin
|
||||||
|
|
||||||
|
load_in_8bit: false
|
||||||
|
load_in_4bit: false
|
||||||
|
strict: false
|
||||||
|
|
||||||
|
datasets:
|
||||||
|
- path: tatsu-lab/alpaca
|
||||||
|
type: alpaca
|
||||||
|
dataset_prepared_path: last_run_prepared
|
||||||
|
val_set_size: 0.05
|
||||||
|
output_dir: ./outputs/out
|
||||||
|
|
||||||
|
sequence_len: 4096
|
||||||
|
sample_packing: true
|
||||||
|
pad_to_sequence_len: true
|
||||||
|
eval_sample_packing: false
|
||||||
|
|
||||||
|
wandb_project:
|
||||||
|
wandb_entity:
|
||||||
|
wandb_watch:
|
||||||
|
wandb_name:
|
||||||
|
wandb_log_model:
|
||||||
|
|
||||||
|
gradient_accumulation_steps: 8
|
||||||
|
micro_batch_size: 1
|
||||||
|
num_epochs: 1
|
||||||
|
optimizer: paged_adamw_8bit
|
||||||
|
lr_scheduler: cosine
|
||||||
|
learning_rate: 2e-5
|
||||||
|
|
||||||
|
train_on_inputs: false
|
||||||
|
group_by_length: false
|
||||||
|
bf16: auto
|
||||||
|
fp16:
|
||||||
|
tf32: false
|
||||||
|
|
||||||
|
gradient_checkpointing: true
|
||||||
|
gradient_checkpointing_kwargs:
|
||||||
|
use_reentrant: false
|
||||||
|
early_stopping_patience:
|
||||||
|
resume_from_checkpoint:
|
||||||
|
logging_steps: 1
|
||||||
|
xformers_attention:
|
||||||
|
flash_attention: true
|
||||||
|
|
||||||
|
warmup_steps: 100
|
||||||
|
evals_per_epoch: 2
|
||||||
|
eval_table_size:
|
||||||
|
saves_per_epoch: 1
|
||||||
|
debug:
|
||||||
|
deepspeed:
|
||||||
|
weight_decay: 0.0
|
||||||
|
fsdp:
|
||||||
|
fsdp_config:
|
||||||
|
special_tokens:
|
||||||
|
pad_token: <|end_of_text|>
|
||||||
|
|
||||||
|
llmcompressor:
|
||||||
|
recipe:
|
||||||
|
finetuning_stage:
|
||||||
|
finetuning_modifiers:
|
||||||
|
ConstantPruningModifier:
|
||||||
|
targets: [
|
||||||
|
're:.*q_proj.weight',
|
||||||
|
're:.*k_proj.weight',
|
||||||
|
're:.*v_proj.weight',
|
||||||
|
're:.*o_proj.weight',
|
||||||
|
're:.*gate_proj.weight',
|
||||||
|
're:.*up_proj.weight',
|
||||||
|
're:.*down_proj.weight',
|
||||||
|
]
|
||||||
|
start: 0
|
||||||
|
save_compressed: true
|
||||||
3
setup.py
3
setup.py
@@ -149,6 +149,9 @@ extras_require = {
|
|||||||
"vllm": [
|
"vllm": [
|
||||||
"vllm==0.7.2",
|
"vllm==0.7.2",
|
||||||
],
|
],
|
||||||
|
"llmcompressor": [
|
||||||
|
"llmcompressor==0.5.1",
|
||||||
|
],
|
||||||
}
|
}
|
||||||
|
|
||||||
install_requires, dependency_links, extras_require_build = parse_requirements(
|
install_requires, dependency_links, extras_require_build = parse_requirements(
|
||||||
|
|||||||
@@ -932,6 +932,9 @@ class HFCausalTrainerBuilder(TrainerBuilderBase):
|
|||||||
collator = DataCollatorForSeq2Seq
|
collator = DataCollatorForSeq2Seq
|
||||||
|
|
||||||
kwargs["return_tensors"] = "pt"
|
kwargs["return_tensors"] = "pt"
|
||||||
|
if issubclass(collator, DataCollatorForSeq2Seq):
|
||||||
|
kwargs["sequence_parallel_degree"] = training_args.sequence_parallel_degree
|
||||||
|
kwargs["ring_attn_func"] = training_args.ring_attn_func
|
||||||
|
|
||||||
return collator(
|
return collator(
|
||||||
*collator_args,
|
*collator_args,
|
||||||
|
|||||||
@@ -371,15 +371,13 @@ class AxolotlTrainer(
|
|||||||
num_items_in_batch=num_items_in_batch,
|
num_items_in_batch=num_items_in_batch,
|
||||||
)
|
)
|
||||||
|
|
||||||
loss = super().compute_loss(
|
return super().compute_loss(
|
||||||
model,
|
model,
|
||||||
inputs,
|
inputs,
|
||||||
return_outputs=return_outputs,
|
return_outputs=return_outputs,
|
||||||
num_items_in_batch=num_items_in_batch,
|
num_items_in_batch=num_items_in_batch,
|
||||||
)
|
)
|
||||||
|
|
||||||
return loss
|
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def orpo_concatenate_inputs(inputs, label_pad_token=-100, pad_token=0, device=None):
|
def orpo_concatenate_inputs(inputs, label_pad_token=-100, pad_token=0, device=None):
|
||||||
concatenated_batch = {}
|
concatenated_batch = {}
|
||||||
|
|||||||
@@ -6,4 +6,4 @@
|
|||||||
from .optimizer import OptimizerMixin
|
from .optimizer import OptimizerMixin
|
||||||
from .rng_state_loader import RngLoaderMixin
|
from .rng_state_loader import RngLoaderMixin
|
||||||
from .scheduler import SchedulerMixin
|
from .scheduler import SchedulerMixin
|
||||||
from .sequence_parallel import SequenceParallelContextManager, SequenceParallelMixin
|
from .sequence_parallel import SequenceParallelMixin
|
||||||
|
|||||||
@@ -1,86 +1,16 @@
|
|||||||
"""
|
"""Module for Axolotl trainer sequence parallelism mixin"""
|
||||||
Module for Axolotl trainer sequence parallelism mixin and training context manager
|
|
||||||
"""
|
|
||||||
|
|
||||||
import functools
|
|
||||||
import logging
|
import logging
|
||||||
|
|
||||||
import torch
|
|
||||||
import torch.distributed as dist
|
import torch.distributed as dist
|
||||||
from datasets import Dataset
|
from datasets import Dataset
|
||||||
from torch import nn
|
|
||||||
from torch.utils.data import DistributedSampler, Sampler
|
from torch.utils.data import DistributedSampler, Sampler
|
||||||
from torch.utils.hooks import RemovableHandle
|
|
||||||
|
|
||||||
from axolotl.monkeypatch.attention.ring_attn import (
|
from axolotl.monkeypatch.attention.ring_attn import get_ring_attn_group
|
||||||
RingAttnFunc,
|
|
||||||
get_ring_attn_group,
|
|
||||||
update_ring_attn_params,
|
|
||||||
)
|
|
||||||
|
|
||||||
LOG = logging.getLogger(__name__)
|
LOG = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
def apply_sequence_parallelism(
|
|
||||||
batch: dict[str, torch.Tensor],
|
|
||||||
local_rank: int,
|
|
||||||
local_world_size: int,
|
|
||||||
ring_attn_func: RingAttnFunc,
|
|
||||||
) -> dict[str, torch.Tensor]:
|
|
||||||
"""
|
|
||||||
Apply sequence parallelism slicing to a batch.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
batch: Batch dictionary (e.g., input_ids, attention_mask, etc.)
|
|
||||||
local_rank: Local rank in the sequence parallel group
|
|
||||||
local_world_size: World size of the sequence parallel group
|
|
||||||
ring_attn_func: The ring attention function to use
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
Sliced batch dictionary.
|
|
||||||
"""
|
|
||||||
# Update ring attention params if needed
|
|
||||||
if batch.get("position_ids") is not None:
|
|
||||||
update_ring_attn_params(position_ids=batch["position_ids"])
|
|
||||||
|
|
||||||
# Slice batch for sequence parallel processing
|
|
||||||
total_seq_len = batch["input_ids"].size(1)
|
|
||||||
for key in batch:
|
|
||||||
if (
|
|
||||||
key in batch
|
|
||||||
and isinstance(batch[key], torch.Tensor)
|
|
||||||
and batch[key].dim() > 1
|
|
||||||
and batch[key].size(1) == total_seq_len
|
|
||||||
):
|
|
||||||
|
|
||||||
if ring_attn_func in [
|
|
||||||
RingAttnFunc.VARLEN_LLAMA3,
|
|
||||||
RingAttnFunc.BATCH_RING,
|
|
||||||
]:
|
|
||||||
# Split in sequential fashion and grab this rank's chunk
|
|
||||||
batch[key] = (
|
|
||||||
batch[key].chunk(local_world_size, dim=1)[local_rank].contiguous()
|
|
||||||
)
|
|
||||||
elif ring_attn_func is RingAttnFunc.BATCH_ZIGZAG:
|
|
||||||
chunks = batch[key].chunk(2 * local_world_size, dim=1)
|
|
||||||
|
|
||||||
# Take rank's chunk and opposing chunk for zigzag pattern
|
|
||||||
selected_chunks = [
|
|
||||||
chunks[local_rank],
|
|
||||||
chunks[2 * local_world_size - local_rank - 1],
|
|
||||||
]
|
|
||||||
batch[key] = torch.cat(selected_chunks, dim=1).contiguous()
|
|
||||||
elif ring_attn_func is RingAttnFunc.BATCH_STRIPE:
|
|
||||||
# Split into striped data and stack
|
|
||||||
tensor = torch.stack(
|
|
||||||
batch[key].split(local_world_size, dim=1),
|
|
||||||
dim=1,
|
|
||||||
).transpose(1, 2)
|
|
||||||
batch[key] = tensor[:, local_rank].contiguous()
|
|
||||||
|
|
||||||
return batch
|
|
||||||
|
|
||||||
|
|
||||||
class SequenceParallelMixin:
|
class SequenceParallelMixin:
|
||||||
"""
|
"""
|
||||||
Mixin class for sequence parallelism support in trainers.
|
Mixin class for sequence parallelism support in trainers.
|
||||||
@@ -157,157 +87,3 @@ class SequenceParallelMixin:
|
|||||||
return self._create_sequence_parallel_sampler(
|
return self._create_sequence_parallel_sampler(
|
||||||
eval_dataset, shuffle=False, is_eval=True
|
eval_dataset, shuffle=False, is_eval=True
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
class SequenceParallelContextManager:
|
|
||||||
"""
|
|
||||||
Context manager for sequence parallelism operations.
|
|
||||||
|
|
||||||
This class provides a context that will automatically apply sequence parallelism
|
|
||||||
during model forward passes using a pre-forward hook, and gather outputs from
|
|
||||||
across the sequence parallelism group using a post-forward hook.
|
|
||||||
"""
|
|
||||||
|
|
||||||
def __init__(
|
|
||||||
self,
|
|
||||||
model: nn.Module,
|
|
||||||
sequence_parallel_degree: int,
|
|
||||||
ring_attn_func: RingAttnFunc,
|
|
||||||
):
|
|
||||||
self.model = model
|
|
||||||
self.sequence_parallel_degree = sequence_parallel_degree
|
|
||||||
self.ring_attn_func = ring_attn_func
|
|
||||||
self.process_group = get_ring_attn_group()
|
|
||||||
|
|
||||||
# Initialize sequence parallel group details
|
|
||||||
self.local_rank = dist.get_rank(self.process_group)
|
|
||||||
self.local_world_size = dist.get_world_size(self.process_group)
|
|
||||||
|
|
||||||
# Will store hook handles for removal
|
|
||||||
self.hook_handles: list[RemovableHandle] = []
|
|
||||||
|
|
||||||
# Create a partially applied version of the apply_sequence_parallelism function
|
|
||||||
# with pre-configured params
|
|
||||||
self.apply_sequence_parallelism = functools.partial(
|
|
||||||
apply_sequence_parallelism,
|
|
||||||
local_rank=self.local_rank,
|
|
||||||
local_world_size=self.local_world_size,
|
|
||||||
ring_attn_func=self.ring_attn_func,
|
|
||||||
)
|
|
||||||
|
|
||||||
def __enter__(self):
|
|
||||||
# Forward pre-hook to apply sequence parallelism
|
|
||||||
def sequence_parallel_pre_hook(_, args, kwargs):
|
|
||||||
# Apply sequence parallelism to kwargs
|
|
||||||
kwargs = self.apply_sequence_parallelism(batch=kwargs)
|
|
||||||
return args, kwargs
|
|
||||||
|
|
||||||
# Forward post-hook to gather outputs
|
|
||||||
def sequence_parallel_post_hook(_, __, output):
|
|
||||||
# Gather the sharded outputs
|
|
||||||
return self.gather_outputs(output)
|
|
||||||
|
|
||||||
# Register both hooks
|
|
||||||
self.hook_handles.append(
|
|
||||||
self.model.register_forward_pre_hook(
|
|
||||||
sequence_parallel_pre_hook, with_kwargs=True
|
|
||||||
)
|
|
||||||
)
|
|
||||||
self.hook_handles.append(
|
|
||||||
self.model.register_forward_hook(sequence_parallel_post_hook)
|
|
||||||
)
|
|
||||||
|
|
||||||
return self
|
|
||||||
|
|
||||||
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
||||||
# Remove all hooks
|
|
||||||
for handle in self.hook_handles:
|
|
||||||
handle.remove()
|
|
||||||
self.hook_handles = []
|
|
||||||
|
|
||||||
def gather_outputs(self, output):
|
|
||||||
"""Gather sharded outputs from all ranks and reconstruct the full tensor."""
|
|
||||||
# Handle different output formats (dict, tensor, etc.)
|
|
||||||
if isinstance(output, dict):
|
|
||||||
gathered_output = {}
|
|
||||||
for key, value in output.items():
|
|
||||||
if isinstance(value, torch.Tensor) and value.dim() > 1:
|
|
||||||
# Gather logits or other sequence-sharded tensors
|
|
||||||
gathered_value = self.gather_tensor(value)
|
|
||||||
gathered_output[key] = gathered_value
|
|
||||||
else:
|
|
||||||
gathered_value = value.clone()
|
|
||||||
dist.all_reduce(
|
|
||||||
gathered_value, op=dist.ReduceOp.SUM, group=self.process_group
|
|
||||||
)
|
|
||||||
gathered_output[key] = gathered_value
|
|
||||||
return gathered_output
|
|
||||||
if isinstance(output, torch.Tensor):
|
|
||||||
return self.gather_tensor(output)
|
|
||||||
|
|
||||||
return output
|
|
||||||
|
|
||||||
def gather_tensor(self, tensor):
|
|
||||||
"""Gather a sharded tensor from all ranks."""
|
|
||||||
# Prepare tensors for all_gather
|
|
||||||
world_size = self.local_world_size
|
|
||||||
|
|
||||||
# Create list to store tensors from all ranks
|
|
||||||
gathered_tensors = [torch.zeros_like(tensor) for _ in range(world_size)]
|
|
||||||
|
|
||||||
# All-gather operation
|
|
||||||
dist.all_gather(gathered_tensors, tensor, group=self.process_group)
|
|
||||||
|
|
||||||
# Concatenate along sequence dimension (typically dim=1)
|
|
||||||
if self.ring_attn_func in [RingAttnFunc.VARLEN_LLAMA3, RingAttnFunc.BATCH_RING]:
|
|
||||||
# Simple concatenation for standard sharding
|
|
||||||
return torch.cat(gathered_tensors, dim=1)
|
|
||||||
|
|
||||||
if self.ring_attn_func is RingAttnFunc.BATCH_ZIGZAG:
|
|
||||||
# Each rank has a pattern of (rank, world_size*2-rank-1)
|
|
||||||
reconstituted_tensors = [None] * (world_size * 2)
|
|
||||||
|
|
||||||
# First, split each gathered tensor into its two chunks
|
|
||||||
for rank, gathered_tensor in enumerate(gathered_tensors):
|
|
||||||
# Each tensor contains two chunks in the sequence dimension
|
|
||||||
chunk_size = gathered_tensor.size(1) // 2
|
|
||||||
chunk1, chunk2 = gathered_tensor.split(chunk_size, dim=1)
|
|
||||||
|
|
||||||
# Place chunks in their original positions
|
|
||||||
reconstituted_tensors[rank] = chunk1
|
|
||||||
reconstituted_tensors[world_size * 2 - rank - 1] = chunk2
|
|
||||||
|
|
||||||
# Concatenate the reconstituted tensors in the correct order
|
|
||||||
return torch.cat(reconstituted_tensors, dim=1)
|
|
||||||
|
|
||||||
# Otherwise, RingAttnFunc.BATCH_STRIPE
|
|
||||||
# In striping, each rank has every world_size-th slice
|
|
||||||
batch_size = tensor.size(0)
|
|
||||||
hidden_dim = tensor.size(-1)
|
|
||||||
|
|
||||||
# First, determine the full sequence length
|
|
||||||
total_seq_len = 0
|
|
||||||
for t in gathered_tensors:
|
|
||||||
total_seq_len += t.size(1)
|
|
||||||
|
|
||||||
# Create a tensor to hold the unstriped result
|
|
||||||
result = torch.zeros(
|
|
||||||
batch_size,
|
|
||||||
total_seq_len,
|
|
||||||
hidden_dim,
|
|
||||||
dtype=tensor.dtype,
|
|
||||||
device=tensor.device,
|
|
||||||
)
|
|
||||||
|
|
||||||
# For each rank's tensor, distribute its slices to the correct positions
|
|
||||||
for rank, gathered_tensor in enumerate(gathered_tensors):
|
|
||||||
# The rank's tensor contains every world_size-th slice
|
|
||||||
# starting from its rank position
|
|
||||||
seq_len = gathered_tensor.size(1)
|
|
||||||
for i in range(seq_len):
|
|
||||||
# Calculate the position in the full tensor
|
|
||||||
pos = i * world_size + rank
|
|
||||||
if pos < total_seq_len:
|
|
||||||
result[:, pos] = gathered_tensor[:, i]
|
|
||||||
|
|
||||||
return result
|
|
||||||
|
|||||||
108
src/axolotl/integrations/llm_compressor/README.md
Normal file
108
src/axolotl/integrations/llm_compressor/README.md
Normal file
@@ -0,0 +1,108 @@
|
|||||||
|
# LLMCompressor Integration
|
||||||
|
|
||||||
|
Fine-tune sparsified models in Axolotl using Neural Magic's [LLMCompressor](https://github.com/vllm-project/llm-compressor).
|
||||||
|
|
||||||
|
This integration enables fine-tuning of models sparsified using LLMCompressor within the Axolotl training framework. By combining LLMCompressor's model compression capabilities with Axolotl's distributed training pipelines, users can efficiently fine-tune sparse models at scale.
|
||||||
|
|
||||||
|
It uses Axolotl’s plugin system to hook into the fine-tuning flows while maintaining sparsity throughout training.
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## Requirements
|
||||||
|
|
||||||
|
- Axolotl with `llmcompressor` extras:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
pip install "axolotl[llmcompressor]"
|
||||||
|
```
|
||||||
|
|
||||||
|
- Requires `llmcompressor >= 0.5.1`
|
||||||
|
|
||||||
|
This will install all necessary dependencies to fine-tune sparsified models using the integration.
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## Usage
|
||||||
|
|
||||||
|
To enable sparse fine-tuning with this integration, include the plugin in your Axolotl config:
|
||||||
|
|
||||||
|
```yaml
|
||||||
|
plugins:
|
||||||
|
- axolotl.integrations.llm_compressor.LLMCompressorPlugin
|
||||||
|
|
||||||
|
llmcompressor:
|
||||||
|
recipe:
|
||||||
|
finetuning_stage:
|
||||||
|
finetuning_modifiers:
|
||||||
|
ConstantPruningModifier:
|
||||||
|
targets: [
|
||||||
|
're:.*q_proj.weight',
|
||||||
|
're:.*k_proj.weight',
|
||||||
|
're:.*v_proj.weight',
|
||||||
|
're:.*o_proj.weight',
|
||||||
|
're:.*gate_proj.weight',
|
||||||
|
're:.*up_proj.weight',
|
||||||
|
're:.*down_proj.weight',
|
||||||
|
]
|
||||||
|
start: 0
|
||||||
|
save_compressed: true
|
||||||
|
# ... (other training arguments)
|
||||||
|
```
|
||||||
|
|
||||||
|
This plugin **does not apply pruning or sparsification itself** — it is intended for **fine-tuning models that have already been sparsified**.
|
||||||
|
|
||||||
|
Pre-sparsified checkpoints can be:
|
||||||
|
- Generated using [LLMCompressor](https://github.com/vllm-project/llm-compressor)
|
||||||
|
- Downloaded from [Neural Magic's Hugging Face page](https://huggingface.co/neuralmagic)
|
||||||
|
- Any custom LLM with compatible sparsity patterns that you've created yourself
|
||||||
|
|
||||||
|
To learn more about writing and customizing LLMCompressor recipes, refer to the official documentation:
|
||||||
|
[https://github.com/vllm-project/llm-compressor/blob/main/README.md](https://github.com/vllm-project/llm-compressor/blob/main/README.md)
|
||||||
|
|
||||||
|
### Storage Optimization with save_compressed
|
||||||
|
|
||||||
|
Setting `save_compressed: true` in your configuration enables saving models in a compressed format, which:
|
||||||
|
- Reduces disk space usage by approximately 40%
|
||||||
|
- Maintains compatibility with vLLM for accelerated inference
|
||||||
|
- Maintains compatibility with llmcompressor for further optimization (example: quantization)
|
||||||
|
|
||||||
|
This option is highly recommended when working with sparse models to maximize the benefits of model compression.
|
||||||
|
|
||||||
|
### Example Config
|
||||||
|
|
||||||
|
See [`examples/llama-3/sparse-finetuning.yaml`](examples/llama-3/sparse-finetuning.yaml) for a complete example.
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## Inference with vLLM
|
||||||
|
|
||||||
|
After fine-tuning your sparse model, you can leverage vLLM for efficient inference.
|
||||||
|
You can also use LLMCompressor to apply additional quantization to your fine-tuned
|
||||||
|
sparse model before inference for even greater performance benefits.:
|
||||||
|
|
||||||
|
```python
|
||||||
|
from vllm import LLM, SamplingParams
|
||||||
|
|
||||||
|
prompts = [
|
||||||
|
"Hello, my name is",
|
||||||
|
"The president of the United States is",
|
||||||
|
"The capital of France is",
|
||||||
|
"The future of AI is",
|
||||||
|
]
|
||||||
|
sampling_params = SamplingParams(temperature=0.8, top_p=0.95)
|
||||||
|
llm = LLM("path/to/your/sparse/model")
|
||||||
|
outputs = llm.generate(prompts, sampling_params)
|
||||||
|
|
||||||
|
for output in outputs:
|
||||||
|
prompt = output.prompt
|
||||||
|
generated_text = output.outputs[0].text
|
||||||
|
print(f"Prompt: {prompt!r}, Generated text: {generated_text!r}")
|
||||||
|
```
|
||||||
|
|
||||||
|
For more details on vLLM's capabilities and advanced configuration options, see the [official vLLM documentation](https://docs.vllm.ai/).
|
||||||
|
|
||||||
|
## Learn More
|
||||||
|
|
||||||
|
For details on available sparsity and quantization schemes, fine-tuning recipes, and usage examples, visit the official LLMCompressor repository:
|
||||||
|
|
||||||
|
[https://github.com/vllm-project/llm-compressor](https://github.com/vllm-project/llm-compressor)
|
||||||
5
src/axolotl/integrations/llm_compressor/__init__.py
Normal file
5
src/axolotl/integrations/llm_compressor/__init__.py
Normal file
@@ -0,0 +1,5 @@
|
|||||||
|
"""Integration entry point for the LLMCompressor plugin."""
|
||||||
|
|
||||||
|
from .plugin import LLMCompressorPlugin
|
||||||
|
|
||||||
|
__all__ = ["LLMCompressorPlugin"]
|
||||||
40
src/axolotl/integrations/llm_compressor/args.py
Normal file
40
src/axolotl/integrations/llm_compressor/args.py
Normal file
@@ -0,0 +1,40 @@
|
|||||||
|
"""
|
||||||
|
LLMCompressor and Sparse Finetuning config models.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
from pydantic import BaseModel, Field
|
||||||
|
from typing_extensions import Annotated
|
||||||
|
|
||||||
|
|
||||||
|
class CompressionArgs(BaseModel):
|
||||||
|
"""Sparse Finetuning config for LLMCompressor."""
|
||||||
|
|
||||||
|
# Typing for recipe is set to Any due to:
|
||||||
|
# https://github.com/vllm-project/llm-compressor/issues/1319
|
||||||
|
recipe: Annotated[
|
||||||
|
Any,
|
||||||
|
Field(
|
||||||
|
description="The recipe containing the compression algorithms and hyperparameters to apply."
|
||||||
|
),
|
||||||
|
]
|
||||||
|
|
||||||
|
save_compressed: Annotated[
|
||||||
|
bool,
|
||||||
|
Field(
|
||||||
|
default=False,
|
||||||
|
description="Whether to save the compressed model after training.",
|
||||||
|
),
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
class LLMCompressorArgs(BaseModel):
|
||||||
|
"""LLMCompressor configuration BaseModel."""
|
||||||
|
|
||||||
|
llmcompressor: Annotated[
|
||||||
|
CompressionArgs,
|
||||||
|
Field(
|
||||||
|
description="Arguments enabling compression pathways through the LLM Compressor plugins"
|
||||||
|
),
|
||||||
|
]
|
||||||
171
src/axolotl/integrations/llm_compressor/plugin.py
Normal file
171
src/axolotl/integrations/llm_compressor/plugin.py
Normal file
@@ -0,0 +1,171 @@
|
|||||||
|
"""
|
||||||
|
Sparse Finetuning plugin for Axolotl — enables handling of sparse neural networks
|
||||||
|
by maintaining masks for zero weights during training.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import logging
|
||||||
|
from functools import wraps
|
||||||
|
from typing import Any, Callable, Concatenate, ParamSpec, TypeVar
|
||||||
|
|
||||||
|
from llmcompressor import active_session, create_session
|
||||||
|
from llmcompressor.core import callbacks as session_callbacks
|
||||||
|
from llmcompressor.recipe import Recipe
|
||||||
|
from torch.nn import Module
|
||||||
|
from transformers.trainer import Trainer
|
||||||
|
from transformers.trainer_callback import TrainerCallback, TrainerControl, TrainerState
|
||||||
|
from transformers.training_args import TrainingArguments
|
||||||
|
|
||||||
|
from axolotl.integrations.base import BasePlugin
|
||||||
|
|
||||||
|
P = ParamSpec("P") # Params for generic function signatures
|
||||||
|
R = TypeVar("R") # Return type for generic function signatures
|
||||||
|
|
||||||
|
LOG = logging.getLogger("axolotl.integrations.llm_compressor")
|
||||||
|
|
||||||
|
|
||||||
|
class LLMCompressorCallbackHandler(TrainerCallback):
|
||||||
|
"""
|
||||||
|
Trainer callback for Sparse Finetuning.
|
||||||
|
Maintains sparsity patterns during training by applying masks after optimization steps,
|
||||||
|
ensuring zero-weight updates are canceled out.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self, trainer: Trainer, recipe: Any):
|
||||||
|
"""
|
||||||
|
Initialize the Sparse Finetuning callback handler.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
trainer (Trainer): Huggingface Trainer instance.
|
||||||
|
recipe (Recipe | dict): Sparse finetuning recipe to apply.
|
||||||
|
"""
|
||||||
|
super().__init__()
|
||||||
|
self.trainer = trainer
|
||||||
|
self.recipe = (
|
||||||
|
Recipe.model_validate(recipe) if not isinstance(recipe, Recipe) else recipe
|
||||||
|
)
|
||||||
|
self.original_compute_loss = trainer.compute_loss
|
||||||
|
self.trainer.compute_loss = compute_loss_wrapper(self.trainer.compute_loss)
|
||||||
|
create_session()
|
||||||
|
|
||||||
|
def on_train_begin(
|
||||||
|
self,
|
||||||
|
args: TrainingArguments,
|
||||||
|
state: TrainerState,
|
||||||
|
control: TrainerControl,
|
||||||
|
**kwargs,
|
||||||
|
) -> None:
|
||||||
|
"""
|
||||||
|
Called at the beginning of training. Initializes the compression session.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
args (TrainingArguments): Training arguments.
|
||||||
|
state (TrainerState): Trainer state.
|
||||||
|
control (TrainerControl): Trainer control.
|
||||||
|
"""
|
||||||
|
super().on_train_begin(args, state, control, **kwargs)
|
||||||
|
self.trainer.accelerator.wait_for_everyone()
|
||||||
|
active_session().initialize(
|
||||||
|
model=self.trainer.model,
|
||||||
|
optimizer=self.trainer.optimizer,
|
||||||
|
start=state.epoch,
|
||||||
|
recipe=self.recipe,
|
||||||
|
)
|
||||||
|
self.trainer.accelerator.wait_for_everyone()
|
||||||
|
|
||||||
|
def on_step_begin(
|
||||||
|
self,
|
||||||
|
args: TrainingArguments,
|
||||||
|
state: TrainerState,
|
||||||
|
control: TrainerControl,
|
||||||
|
**kwargs,
|
||||||
|
) -> None:
|
||||||
|
"""
|
||||||
|
Called at the beginning of a training step. Triggers batch_start callback.
|
||||||
|
"""
|
||||||
|
super().on_step_begin(args, state, control, **kwargs)
|
||||||
|
session_callbacks.batch_start()
|
||||||
|
|
||||||
|
def on_step_end(
|
||||||
|
self,
|
||||||
|
args: TrainingArguments,
|
||||||
|
state: TrainerState,
|
||||||
|
control: TrainerControl,
|
||||||
|
**kwargs,
|
||||||
|
) -> None:
|
||||||
|
"""
|
||||||
|
Called at the end of a training step. Triggers optimizer and batch_end callbacks.
|
||||||
|
"""
|
||||||
|
super().on_step_end(args, state, control, **kwargs)
|
||||||
|
session_callbacks.optim_pre_step()
|
||||||
|
session_callbacks.optim_post_step()
|
||||||
|
session_callbacks.batch_end()
|
||||||
|
|
||||||
|
def on_train_end(
|
||||||
|
self,
|
||||||
|
args: TrainingArguments,
|
||||||
|
state: TrainerState,
|
||||||
|
control: TrainerControl,
|
||||||
|
**kwargs,
|
||||||
|
) -> None:
|
||||||
|
"""
|
||||||
|
Called at the end of training. Finalizes the compression session.
|
||||||
|
"""
|
||||||
|
super().on_train_end(args, state, control, **kwargs)
|
||||||
|
active_session().finalize()
|
||||||
|
self.trainer.compute_loss_func = self.original_compute_loss
|
||||||
|
|
||||||
|
|
||||||
|
class LLMCompressorPlugin(BasePlugin):
|
||||||
|
"""
|
||||||
|
Sparse Finetuning plugin for Axolotl integration.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def get_input_args(self) -> str:
|
||||||
|
"""
|
||||||
|
Returns the path to the plugin's argument definition.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
str: Dotted path to the LLMCompressorArgs class.
|
||||||
|
"""
|
||||||
|
return "axolotl.integrations.llm_compressor.args.LLMCompressorArgs"
|
||||||
|
|
||||||
|
def add_callbacks_post_trainer(self, cfg: Any, trainer: Trainer) -> list:
|
||||||
|
"""
|
||||||
|
Adds Sparse Finetuning callback to the Trainer instance.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
cfg (Any): Configuration object containing the sparse recipe.
|
||||||
|
trainer (Trainer): Huggingface Trainer instance.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
list: List containing the configured callback instances.
|
||||||
|
"""
|
||||||
|
LOG.info("Adding Sparse Finetuning callback to the trainer")
|
||||||
|
callback = LLMCompressorCallbackHandler(
|
||||||
|
trainer=trainer,
|
||||||
|
recipe=cfg.llmcompressor.recipe,
|
||||||
|
)
|
||||||
|
return [callback]
|
||||||
|
|
||||||
|
|
||||||
|
def compute_loss_wrapper(
|
||||||
|
compute_loss_func: Callable[Concatenate[Module, P], R],
|
||||||
|
) -> Callable[Concatenate[Module, P], R]:
|
||||||
|
"""
|
||||||
|
Wraps the loss computation function to trigger the loss_calculated callback.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
compute_loss_func (Callable): Original loss computation function.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
Callable: Wrapped function that also invokes the loss_calculated callback.
|
||||||
|
"""
|
||||||
|
|
||||||
|
@wraps(compute_loss_func)
|
||||||
|
def compute_and_notify(model: Module, *args: P.args, **kwargs: P.kwargs) -> R:
|
||||||
|
loss = compute_loss_func(model, *args, **kwargs)
|
||||||
|
if active_session().lifecycle.initialized_ and model.training:
|
||||||
|
session_callbacks.loss_calculated(loss=loss)
|
||||||
|
return loss
|
||||||
|
|
||||||
|
return compute_and_notify
|
||||||
40
src/axolotl/integrations/llm_compressor/utils.py
Normal file
40
src/axolotl/integrations/llm_compressor/utils.py
Normal file
@@ -0,0 +1,40 @@
|
|||||||
|
"""Utilities for llmcompressor integration with axolotl."""
|
||||||
|
|
||||||
|
from typing import Union
|
||||||
|
|
||||||
|
from llmcompressor.transformers.sparsification.compressed_tensors_utils import (
|
||||||
|
modify_save_pretrained,
|
||||||
|
)
|
||||||
|
from transformers import PreTrainedModel, Trainer
|
||||||
|
|
||||||
|
|
||||||
|
def save_compressed_model(
|
||||||
|
model: PreTrainedModel,
|
||||||
|
output_dir: Union[str, bytes],
|
||||||
|
trainer: Trainer,
|
||||||
|
safe_serialization: bool = False,
|
||||||
|
save_compressed: bool = False,
|
||||||
|
) -> None:
|
||||||
|
"""
|
||||||
|
Synchronize processes, apply compression hooks, and save the model.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
model (PreTrainedModel): The model to be saved.
|
||||||
|
output_dir (str or bytes): Path where the model files will be written.
|
||||||
|
trainer (Trainer): Hugging Face Trainer for process synchronization.
|
||||||
|
safe_serialization (bool): Use safe serialization if True.
|
||||||
|
save_compressed (bool): Write compressed tensors if True.
|
||||||
|
"""
|
||||||
|
trainer.accelerator.wait_for_everyone()
|
||||||
|
|
||||||
|
# Only the main process writes the files
|
||||||
|
if not trainer.accelerator.is_main_process:
|
||||||
|
return
|
||||||
|
|
||||||
|
modify_save_pretrained(model)
|
||||||
|
model.save_pretrained(
|
||||||
|
output_dir,
|
||||||
|
safe_serialization=safe_serialization,
|
||||||
|
save_compressed=save_compressed,
|
||||||
|
skip_sparsity_compression_stats=not save_compressed,
|
||||||
|
)
|
||||||
@@ -76,8 +76,7 @@ def register_ring_attn(
|
|||||||
|
|
||||||
LOG.info(
|
LOG.info(
|
||||||
"Enabling ring attention sequence parallelism: "
|
"Enabling ring attention sequence parallelism: "
|
||||||
f"each sequence will be processed across {sequence_parallel_degree} GPUs "
|
f"each sequence will be processed across {sequence_parallel_degree} GPUs"
|
||||||
f"using the {ring_attn_func.value} ring-flash-attn implementation"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
rank = dist.get_rank()
|
rank = dist.get_rank()
|
||||||
|
|||||||
@@ -6,7 +6,6 @@ import os
|
|||||||
import signal
|
import signal
|
||||||
import sys
|
import sys
|
||||||
import weakref
|
import weakref
|
||||||
from contextlib import nullcontext
|
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Any, Dict
|
from typing import Any, Dict
|
||||||
|
|
||||||
@@ -26,9 +25,6 @@ from axolotl.contribs.lgpl import ( # pylint: disable = no-name-in-module
|
|||||||
fix_untrained_tokens,
|
fix_untrained_tokens,
|
||||||
)
|
)
|
||||||
from axolotl.core.trainer_builder import HFCausalTrainerBuilder, HFRLTrainerBuilder
|
from axolotl.core.trainer_builder import HFCausalTrainerBuilder, HFRLTrainerBuilder
|
||||||
from axolotl.core.trainers.mixins.sequence_parallel import (
|
|
||||||
SequenceParallelContextManager,
|
|
||||||
)
|
|
||||||
from axolotl.logging_config import configure_logging
|
from axolotl.logging_config import configure_logging
|
||||||
from axolotl.utils.dict import DictDefault
|
from axolotl.utils.dict import DictDefault
|
||||||
from axolotl.utils.distributed import cleanup_distributed
|
from axolotl.utils.distributed import cleanup_distributed
|
||||||
@@ -189,28 +185,16 @@ def execute_training(
|
|||||||
trainer: The configured trainer object.
|
trainer: The configured trainer object.
|
||||||
resume_from_checkpoint: Path to checkpoint to resume from, if applicable.
|
resume_from_checkpoint: Path to checkpoint to resume from, if applicable.
|
||||||
"""
|
"""
|
||||||
# Define the context managers to use
|
LOG.info("Starting trainer...")
|
||||||
flash_context = (
|
if cfg.flash_optimum:
|
||||||
torch.backends.cuda.sdp_kernel(
|
with torch.backends.cuda.sdp_kernel(
|
||||||
|
# TODO configure these from the YAML w/ sdp_kernel_kwargs: ...
|
||||||
enable_flash=True,
|
enable_flash=True,
|
||||||
enable_math=True,
|
enable_math=True,
|
||||||
enable_mem_efficient=True,
|
enable_mem_efficient=True,
|
||||||
)
|
):
|
||||||
if cfg.flash_optimum
|
trainer.train(resume_from_checkpoint=resume_from_checkpoint)
|
||||||
else nullcontext()
|
else:
|
||||||
)
|
|
||||||
sequence_parallel_context = (
|
|
||||||
SequenceParallelContextManager(
|
|
||||||
model=trainer.model,
|
|
||||||
sequence_parallel_degree=cfg.sequence_parallel_degree,
|
|
||||||
ring_attn_func=cfg.ring_attn_func,
|
|
||||||
)
|
|
||||||
if cfg.sequence_parallel_degree > 1
|
|
||||||
else nullcontext()
|
|
||||||
)
|
|
||||||
|
|
||||||
LOG.info("Starting trainer...")
|
|
||||||
with flash_context, sequence_parallel_context:
|
|
||||||
trainer.train(resume_from_checkpoint=resume_from_checkpoint)
|
trainer.train(resume_from_checkpoint=resume_from_checkpoint)
|
||||||
|
|
||||||
|
|
||||||
@@ -287,6 +271,19 @@ def save_trained_model(
|
|||||||
os.remove(os.path.join(cfg.output_dir, "model.safetensors"))
|
os.remove(os.path.join(cfg.output_dir, "model.safetensors"))
|
||||||
except FileNotFoundError:
|
except FileNotFoundError:
|
||||||
pass
|
pass
|
||||||
|
elif hasattr(cfg, "llmcompressor") and cfg.llmcompressor:
|
||||||
|
from axolotl.integrations.llm_compressor.utils import (
|
||||||
|
save_compressed_model,
|
||||||
|
)
|
||||||
|
|
||||||
|
save_compressed_model(
|
||||||
|
model=model,
|
||||||
|
output_dir=cfg.output_dir,
|
||||||
|
trainer=trainer,
|
||||||
|
safe_serialization=safe_serialization,
|
||||||
|
save_compressed=cfg.llmcompressor.save_compressed,
|
||||||
|
)
|
||||||
|
|
||||||
elif cfg.local_rank == 0:
|
elif cfg.local_rank == 0:
|
||||||
if cfg.flash_optimum and BetterTransformer:
|
if cfg.flash_optimum and BetterTransformer:
|
||||||
model = BetterTransformer.reverse(model)
|
model = BetterTransformer.reverse(model)
|
||||||
@@ -295,6 +292,7 @@ def save_trained_model(
|
|||||||
trainer.model.save_pretrained(
|
trainer.model.save_pretrained(
|
||||||
cfg.output_dir, safe_serialization=safe_serialization
|
cfg.output_dir, safe_serialization=safe_serialization
|
||||||
)
|
)
|
||||||
|
|
||||||
model.save_pretrained(cfg.output_dir, safe_serialization=safe_serialization)
|
model.save_pretrained(cfg.output_dir, safe_serialization=safe_serialization)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -1,12 +1,20 @@
|
|||||||
"""Data collators for axolotl to pad labels and position_ids for packed sequences"""
|
"""
|
||||||
|
Data collators for axolotl to pad labels and position_ids for packed sequences. Also
|
||||||
|
includes logic for handling sequence parallelism collation.
|
||||||
|
"""
|
||||||
|
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
import numpy as np
|
import numpy as np
|
||||||
|
import torch
|
||||||
|
import torch.distributed as dist
|
||||||
from transformers import PreTrainedTokenizerBase
|
from transformers import PreTrainedTokenizerBase
|
||||||
from transformers.utils import PaddingStrategy
|
from transformers.utils import PaddingStrategy
|
||||||
|
|
||||||
|
from axolotl.monkeypatch.attention.ring_attn import update_ring_attn_params
|
||||||
|
from axolotl.monkeypatch.attention.ring_attn.patch import RingAttnFunc
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
class DataCollatorForSeq2Seq:
|
class DataCollatorForSeq2Seq:
|
||||||
@@ -41,6 +49,8 @@ class DataCollatorForSeq2Seq:
|
|||||||
The id to use when padding the labels (-100 will be automatically ignored by PyTorch loss functions).
|
The id to use when padding the labels (-100 will be automatically ignored by PyTorch loss functions).
|
||||||
return_tensors (`str`):
|
return_tensors (`str`):
|
||||||
The type of Tensor to return. Allowable values are "np", "pt" and "tf".
|
The type of Tensor to return. Allowable values are "np", "pt" and "tf".
|
||||||
|
sequence_parallel_degree (`int`):
|
||||||
|
The degree of sequence parallelism. Default to 1 for no sequence parallelism.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
tokenizer: PreTrainedTokenizerBase
|
tokenizer: PreTrainedTokenizerBase
|
||||||
@@ -51,6 +61,17 @@ class DataCollatorForSeq2Seq:
|
|||||||
label_pad_token_id: int = -100
|
label_pad_token_id: int = -100
|
||||||
position_pad_token_id: int = 0
|
position_pad_token_id: int = 0
|
||||||
return_tensors: str = "pt"
|
return_tensors: str = "pt"
|
||||||
|
sequence_parallel_degree: int = 1
|
||||||
|
ring_attn_func: RingAttnFunc | None = None
|
||||||
|
|
||||||
|
def __post_init__(self):
|
||||||
|
if self.sequence_parallel_degree > 1:
|
||||||
|
from axolotl.monkeypatch.attention.ring_attn import get_ring_attn_group
|
||||||
|
|
||||||
|
# Get information about our position in the SP group
|
||||||
|
sp_group = get_ring_attn_group()
|
||||||
|
self.local_rank = dist.get_rank(group=sp_group)
|
||||||
|
self.local_world_size = dist.get_world_size(group=sp_group)
|
||||||
|
|
||||||
def __call__(self, features, return_tensors=None):
|
def __call__(self, features, return_tensors=None):
|
||||||
has_attn_mask = "attention_mask" in features[0].keys()
|
has_attn_mask = "attention_mask" in features[0].keys()
|
||||||
@@ -120,8 +141,62 @@ class DataCollatorForSeq2Seq:
|
|||||||
)
|
)
|
||||||
features["decoder_input_ids"] = decoder_input_ids
|
features["decoder_input_ids"] = decoder_input_ids
|
||||||
|
|
||||||
|
if self.sequence_parallel_degree > 1:
|
||||||
|
features = self.apply_sequence_parallelism(features)
|
||||||
|
|
||||||
return features
|
return features
|
||||||
|
|
||||||
|
def apply_sequence_parallelism(
|
||||||
|
self, batch: dict[str, torch.Tensor]
|
||||||
|
) -> torch.Tensor:
|
||||||
|
"""
|
||||||
|
Apply sequence parallelism slicing to a batch.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
batch: Batch dictionary from parent collator.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
Sliced batch dictionary.
|
||||||
|
"""
|
||||||
|
# Get local (start, end) for sequence parallelism slicing
|
||||||
|
total_seq_len = batch["input_ids"].size(1)
|
||||||
|
|
||||||
|
# Update params for varlen ring attention calculation
|
||||||
|
if batch.get("position_ids") is not None:
|
||||||
|
update_ring_attn_params(position_ids=batch["position_ids"])
|
||||||
|
|
||||||
|
# Slice batch for sequence parallel processing
|
||||||
|
for key in batch:
|
||||||
|
if batch[key].size(1) == total_seq_len:
|
||||||
|
if self.ring_attn_func in [
|
||||||
|
RingAttnFunc.VARLEN_LLAMA3,
|
||||||
|
RingAttnFunc.BATCH_RING,
|
||||||
|
]:
|
||||||
|
batch[key] = (
|
||||||
|
batch[key]
|
||||||
|
.chunk(self.local_world_size, dim=1)[self.local_rank]
|
||||||
|
.contiguous()
|
||||||
|
)
|
||||||
|
elif self.ring_attn_func is RingAttnFunc.BATCH_ZIGZAG:
|
||||||
|
chunks = batch[key].chunk(2 * self.local_world_size, dim=1)
|
||||||
|
|
||||||
|
# Take rank's chunk and opposing chunk for zigzag pattern
|
||||||
|
selected_chunks = [
|
||||||
|
chunks[self.local_rank],
|
||||||
|
chunks[2 * self.local_world_size - self.local_rank - 1],
|
||||||
|
]
|
||||||
|
batch[key] = torch.cat(selected_chunks, dim=1).contiguous()
|
||||||
|
elif self.ring_attn_func is RingAttnFunc.BATCH_STRIPE:
|
||||||
|
# TODO(djsaunde): This doesn't seem to work as expected
|
||||||
|
# Split into striped data and stack
|
||||||
|
tensor = torch.stack(
|
||||||
|
batch[key].split(self.local_world_size, dim=1),
|
||||||
|
dim=1,
|
||||||
|
).transpose(1, 2)
|
||||||
|
batch[key] = tensor[:, self.local_rank].contiguous()
|
||||||
|
|
||||||
|
return batch
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
class BatchSamplerDataCollatorForSeq2Seq(DataCollatorForSeq2Seq):
|
class BatchSamplerDataCollatorForSeq2Seq(DataCollatorForSeq2Seq):
|
||||||
|
|||||||
@@ -126,6 +126,9 @@ def normalize_config(cfg):
|
|||||||
with open(ds_config_path, encoding="utf-8") as f:
|
with open(ds_config_path, encoding="utf-8") as f:
|
||||||
cfg.deepspeed = json.load(f)
|
cfg.deepspeed = json.load(f)
|
||||||
|
|
||||||
|
if cfg.sequence_parallel_degree is None:
|
||||||
|
cfg.sequence_parallel_degree = 1
|
||||||
|
|
||||||
if cfg.saves_per_epoch:
|
if cfg.saves_per_epoch:
|
||||||
save_steps = 1.0 / (cfg.saves_per_epoch * cfg.num_epochs)
|
save_steps = 1.0 / (cfg.saves_per_epoch * cfg.num_epochs)
|
||||||
if save_steps < 1.0: # prevent saves on every step
|
if save_steps < 1.0: # prevent saves on every step
|
||||||
|
|||||||
@@ -139,6 +139,22 @@ def check_model_config(cfg: DictDefault, model_config: PretrainedConfig):
|
|||||||
hasattr(model_config, "quantization_config")
|
hasattr(model_config, "quantization_config")
|
||||||
and model_config.quantization_config
|
and model_config.quantization_config
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# Detect compressed-tensors config
|
||||||
|
is_compressed_tensors_config = (
|
||||||
|
quant_config_exists
|
||||||
|
and model_config.quantization_config.get("quant_method") == "compressed-tensors"
|
||||||
|
)
|
||||||
|
|
||||||
|
if is_compressed_tensors_config:
|
||||||
|
if model_config.quantization_config.get("config_groups"):
|
||||||
|
LOG.warning(
|
||||||
|
"Found `config_groups` in a compressed-tensors config. "
|
||||||
|
"QAT integration with llmcompressor is not tested."
|
||||||
|
)
|
||||||
|
# Skip further quant checks for compressed-tensors
|
||||||
|
return
|
||||||
|
|
||||||
quant_config_method_is_gptq = (
|
quant_config_method_is_gptq = (
|
||||||
quant_config_exists
|
quant_config_exists
|
||||||
and "quant_method" in model_config.quantization_config
|
and "quant_method" in model_config.quantization_config
|
||||||
|
|||||||
@@ -18,7 +18,6 @@ from pydantic import (
|
|||||||
)
|
)
|
||||||
from transformers.utils.import_utils import is_torch_npu_available
|
from transformers.utils.import_utils import is_torch_npu_available
|
||||||
|
|
||||||
from axolotl.utils.distributed import is_main_process
|
|
||||||
from axolotl.utils.schemas.datasets import (
|
from axolotl.utils.schemas.datasets import (
|
||||||
DatasetConfig,
|
DatasetConfig,
|
||||||
DPODataset,
|
DPODataset,
|
||||||
@@ -719,10 +718,9 @@ class AxolotlInputConfig(
|
|||||||
and data.get("eval_sample_packing") is None
|
and data.get("eval_sample_packing") is None
|
||||||
and not data.get("eval_table_size")
|
and not data.get("eval_table_size")
|
||||||
):
|
):
|
||||||
if is_main_process():
|
LOG.info(
|
||||||
LOG.info(
|
"explicitly setting `eval_sample_packing` to match `sample_packing`"
|
||||||
"explicitly setting `eval_sample_packing` to match `sample_packing`"
|
)
|
||||||
)
|
|
||||||
data["eval_sample_packing"] = True
|
data["eval_sample_packing"] = True
|
||||||
|
|
||||||
if (
|
if (
|
||||||
@@ -1151,17 +1149,22 @@ class AxolotlInputConfig(
|
|||||||
|
|
||||||
return data
|
return data
|
||||||
|
|
||||||
@model_validator(mode="after")
|
@field_validator("sequence_parallel_degree", mode="after")
|
||||||
def check_sequence_parallel_degree(self):
|
@classmethod
|
||||||
if not self.sequence_parallel_degree:
|
def check_sequence_parallel_degree(cls, value, info):
|
||||||
self.sequence_parallel_degree = 1
|
if not value:
|
||||||
elif self.sequence_parallel_degree > 1:
|
value = 1
|
||||||
if not self.flash_attention:
|
|
||||||
|
if value > 1:
|
||||||
|
if not info.data.get("flash_attention"):
|
||||||
raise ValueError(
|
raise ValueError(
|
||||||
"flash_attention: true must be set with sequence_parallel_degree > 1"
|
"flash_attention: true must be set with sequence_parallel_degree > 1"
|
||||||
)
|
)
|
||||||
|
|
||||||
if self.sample_packing and self.micro_batch_size > 1:
|
if (
|
||||||
|
info.data.get("sample_packing")
|
||||||
|
and not info.data["micro_batch_size"] == 1
|
||||||
|
):
|
||||||
raise ValueError(
|
raise ValueError(
|
||||||
"micro_batch_size must be set to 1 when sample_packing is enabled"
|
"micro_batch_size must be set to 1 when sample_packing is enabled"
|
||||||
"due to a `ring-flash-attn` requirement"
|
"due to a `ring-flash-attn` requirement"
|
||||||
@@ -1179,43 +1182,44 @@ class AxolotlInputConfig(
|
|||||||
# TODO: monkeypatch / callback to average losses correctly across SP ranks
|
# TODO: monkeypatch / callback to average losses correctly across SP ranks
|
||||||
# / fix gradient scaling across SP ranks. Losses, grads should be scaled
|
# / fix gradient scaling across SP ranks. Losses, grads should be scaled
|
||||||
# according to the proportion of non-padding tokens per rank.
|
# according to the proportion of non-padding tokens per rank.
|
||||||
if is_main_process():
|
LOG.warning(
|
||||||
LOG.warning(
|
"Sequence parallelism (SP) is enabled with "
|
||||||
"Sequence parallelism (SP) is enabled with "
|
f"sequence_parallel_degree={value}. Please note that logged losses may "
|
||||||
f"sequence_parallel_degree={self.sequence_parallel_degree}. "
|
"differ slightly to the non-SP losses due to transformers Trainer "
|
||||||
"Please note that logged losses may differ slightly to the non-SP "
|
"implementation details. Please see "
|
||||||
"losses due to transformers Trainer implementation details. "
|
"https://github.com/axolotl-ai-cloud/axolotl/pull/2495#issuecomment-2784022042 "
|
||||||
"Please see https://github.com/axolotl-ai-cloud/axolotl/pull/2495#issuecomment-2784022042 "
|
"for more details."
|
||||||
"for more details."
|
)
|
||||||
)
|
|
||||||
|
|
||||||
return self
|
return value
|
||||||
|
|
||||||
@model_validator(mode="after")
|
@field_validator("ring_attn_func", mode="after")
|
||||||
def validate_ring_attn_func(self):
|
@classmethod
|
||||||
if getattr(self, "sequence_parallel_degree", 1) == 1:
|
def check_ring_attn_func(cls, value, info):
|
||||||
return self
|
if not info.data.get("sequence_parallel_degree", 1) > 1:
|
||||||
|
return value
|
||||||
|
|
||||||
from axolotl.monkeypatch.attention.ring_attn.patch import RingAttnFunc
|
from axolotl.monkeypatch.attention.ring_attn.patch import RingAttnFunc
|
||||||
|
|
||||||
if self.ring_attn_func is not None:
|
if value is not None:
|
||||||
|
# Set the ring attention function if passed in config
|
||||||
valid_funcs = list(RingAttnFunc)
|
valid_funcs = list(RingAttnFunc)
|
||||||
if self.ring_attn_func in valid_funcs:
|
if value in valid_funcs:
|
||||||
self.ring_attn_func = RingAttnFunc(self.ring_attn_func)
|
value = RingAttnFunc(value)
|
||||||
else:
|
else:
|
||||||
raise ValueError(
|
raise ValueError(
|
||||||
f"ring_attn_func: {self.ring_attn_func} must be in {valid_funcs}"
|
f"ring_attn_func: {value} must be one of {valid_funcs}"
|
||||||
)
|
)
|
||||||
else:
|
else:
|
||||||
# Default ring attention function selection
|
# Default ring attention function selection
|
||||||
sample_packing = getattr(self, "sample_packing", False)
|
sample_packing = info.data.get("sample_packing")
|
||||||
self.ring_attn_func = (
|
value = (
|
||||||
RingAttnFunc.VARLEN_LLAMA3
|
RingAttnFunc.VARLEN_LLAMA3
|
||||||
if sample_packing
|
if sample_packing
|
||||||
else RingAttnFunc.BATCH_RING
|
else RingAttnFunc.BATCH_RING
|
||||||
)
|
)
|
||||||
|
|
||||||
return self
|
return value
|
||||||
|
|
||||||
@model_validator(mode="before")
|
@model_validator(mode="before")
|
||||||
@classmethod
|
@classmethod
|
||||||
|
|||||||
@@ -348,7 +348,7 @@ def process_datasets_for_packing(cfg, train_dataset, eval_dataset):
|
|||||||
load_from_cache_file=not cfg.is_preprocess,
|
load_from_cache_file=not cfg.is_preprocess,
|
||||||
desc="Add position_id column (PoSE)",
|
desc="Add position_id column (PoSE)",
|
||||||
)
|
)
|
||||||
elif cfg.sample_packing:
|
elif cfg.sample_packing or cfg.sequence_parallel_degree > 1:
|
||||||
drop_long_kwargs = {}
|
drop_long_kwargs = {}
|
||||||
if filter_map_kwargs:
|
if filter_map_kwargs:
|
||||||
drop_long_kwargs["desc"] = "Add position_id column (Sample Packing)"
|
drop_long_kwargs["desc"] = "Add position_id column (Sample Packing)"
|
||||||
@@ -358,7 +358,7 @@ def process_datasets_for_packing(cfg, train_dataset, eval_dataset):
|
|||||||
**filter_map_kwargs,
|
**filter_map_kwargs,
|
||||||
**drop_long_kwargs,
|
**drop_long_kwargs,
|
||||||
)
|
)
|
||||||
if cfg.eval_sample_packing:
|
if cfg.eval_sample_packing or cfg.sequence_parallel_degree > 1:
|
||||||
if eval_dataset:
|
if eval_dataset:
|
||||||
eval_dataset = eval_dataset.map(
|
eval_dataset = eval_dataset.map(
|
||||||
add_position_ids,
|
add_position_ids,
|
||||||
|
|||||||
104
tests/e2e/integrations/test_llm_compressor.py
Normal file
104
tests/e2e/integrations/test_llm_compressor.py
Normal file
@@ -0,0 +1,104 @@
|
|||||||
|
"""
|
||||||
|
E2E smoke tests for LLMCompressorPlugin integration
|
||||||
|
"""
|
||||||
|
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from axolotl.cli.args import TrainerCliArgs
|
||||||
|
from axolotl.common.datasets import load_datasets
|
||||||
|
from axolotl.train import train
|
||||||
|
from axolotl.utils.config import normalize_config, prepare_plugins, validate_config
|
||||||
|
from axolotl.utils.dict import DictDefault
|
||||||
|
|
||||||
|
from tests.e2e.utils import check_model_output_exists, require_torch_2_4_1
|
||||||
|
|
||||||
|
MODELS = [
|
||||||
|
"nm-testing/llama2.c-stories42M-pruned2.4-compressed",
|
||||||
|
"nm-testing/llama2.c-stories42M-gsm8k-sparse-only-compressed",
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
"base_model", MODELS, ids=["no-checkpoint-recipe", "with-checkpoint-recipe"]
|
||||||
|
)
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
"save_compressed", [True, False], ids=["save_compressed", "save_uncompressed"]
|
||||||
|
)
|
||||||
|
class TestLLMCompressorIntegration:
|
||||||
|
"""
|
||||||
|
e2e tests for axolotl.integrations.llm_compressor.LLMCompressorPlugin
|
||||||
|
"""
|
||||||
|
|
||||||
|
@require_torch_2_4_1
|
||||||
|
def test_llmcompressor_plugin(
|
||||||
|
self, temp_dir, base_model: str, save_compressed: bool
|
||||||
|
):
|
||||||
|
# core cfg
|
||||||
|
cfg = DictDefault(
|
||||||
|
{
|
||||||
|
"base_model": base_model,
|
||||||
|
"plugins": ["axolotl.integrations.llm_compressor.LLMCompressorPlugin"],
|
||||||
|
"sequence_len": 1024,
|
||||||
|
"val_set_size": 0.05,
|
||||||
|
"special_tokens": {"pad_token": "<|endoftext|>"},
|
||||||
|
"datasets": [{"path": "mhenrichsen/alpaca_2k_test", "type": "alpaca"}],
|
||||||
|
"num_epochs": 1,
|
||||||
|
"micro_batch_size": 2,
|
||||||
|
"gradient_accumulation_steps": 2,
|
||||||
|
"output_dir": temp_dir,
|
||||||
|
"learning_rate": 1e-5,
|
||||||
|
"optimizer": "adamw_torch_fused",
|
||||||
|
"lr_scheduler": "cosine",
|
||||||
|
"save_safetensors": True,
|
||||||
|
"bf16": "auto",
|
||||||
|
"max_steps": 5,
|
||||||
|
"llmcompressor": {
|
||||||
|
"recipe": {
|
||||||
|
"finetuning_stage": {
|
||||||
|
"finetuning_modifiers": {
|
||||||
|
"ConstantPruningModifier": {
|
||||||
|
"targets": [
|
||||||
|
"re:.*q_proj.weight",
|
||||||
|
"re:.*k_proj.weight",
|
||||||
|
"re:.*v_proj.weight",
|
||||||
|
"re:.*o_proj.weight",
|
||||||
|
"re:.*gate_proj.weight",
|
||||||
|
"re:.*up_proj.weight",
|
||||||
|
"re:.*down_proj.weight",
|
||||||
|
],
|
||||||
|
"start": 0,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
"save_compressed": save_compressed,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
prepare_plugins(cfg)
|
||||||
|
cfg = validate_config(cfg)
|
||||||
|
normalize_config(cfg)
|
||||||
|
cli_args = TrainerCliArgs()
|
||||||
|
dataset_meta = load_datasets(cfg=cfg, cli_args=cli_args)
|
||||||
|
|
||||||
|
train(cfg=cfg, dataset_meta=dataset_meta)
|
||||||
|
check_model_output_exists(temp_dir, cfg)
|
||||||
|
_check_llmcompressor_model_outputs(temp_dir, save_compressed)
|
||||||
|
|
||||||
|
|
||||||
|
def _check_llmcompressor_model_outputs(temp_dir, save_compressed):
|
||||||
|
|
||||||
|
# recipe.yaml should exist
|
||||||
|
assert (Path(temp_dir) / "recipe.yaml").exists()
|
||||||
|
|
||||||
|
# sparsity config exists if save_compressed
|
||||||
|
if save_compressed:
|
||||||
|
from compressed_tensors import ModelCompressor
|
||||||
|
from compressed_tensors.config import Sparse24BitMaskConfig
|
||||||
|
|
||||||
|
compressor = ModelCompressor.from_pretrained(temp_dir)
|
||||||
|
assert compressor is not None
|
||||||
|
assert isinstance(compressor.sparsity_config, Sparse24BitMaskConfig)
|
||||||
@@ -2,19 +2,14 @@
|
|||||||
|
|
||||||
# pylint: disable=redefined-outer-name,unused-argument
|
# pylint: disable=redefined-outer-name,unused-argument
|
||||||
|
|
||||||
import functools
|
|
||||||
import sys
|
|
||||||
from unittest.mock import MagicMock, patch
|
from unittest.mock import MagicMock, patch
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
import torch
|
import torch
|
||||||
from accelerate.state import PartialState
|
from accelerate.state import PartialState
|
||||||
|
|
||||||
from axolotl.core.trainers.mixins.sequence_parallel import apply_sequence_parallelism
|
|
||||||
from axolotl.monkeypatch.attention.ring_attn import (
|
from axolotl.monkeypatch.attention.ring_attn import (
|
||||||
RingAttnFunc,
|
|
||||||
get_ring_attn_group,
|
get_ring_attn_group,
|
||||||
register_ring_attn,
|
|
||||||
set_ring_attn_group,
|
set_ring_attn_group,
|
||||||
)
|
)
|
||||||
from axolotl.utils.dict import DictDefault
|
from axolotl.utils.dict import DictDefault
|
||||||
@@ -52,27 +47,6 @@ def fixture_cfg():
|
|||||||
return cfg
|
return cfg
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
|
||||||
def sequence_parallel_batch():
|
|
||||||
"""Create a test batch for sequence parallelism tests."""
|
|
||||||
batch_size = 1
|
|
||||||
seq_len = 8
|
|
||||||
|
|
||||||
# Create test tensors
|
|
||||||
input_ids = torch.arange(batch_size * seq_len).reshape(batch_size, seq_len)
|
|
||||||
attention_mask = torch.ones(batch_size, seq_len)
|
|
||||||
position_ids = torch.arange(seq_len).expand(batch_size, seq_len)
|
|
||||||
|
|
||||||
# Create test batch
|
|
||||||
batch = {
|
|
||||||
"input_ids": input_ids,
|
|
||||||
"attention_mask": attention_mask,
|
|
||||||
"position_ids": position_ids,
|
|
||||||
}
|
|
||||||
|
|
||||||
return batch
|
|
||||||
|
|
||||||
|
|
||||||
class TestRingAttention:
|
class TestRingAttention:
|
||||||
"""Tests for the ring attention functionality."""
|
"""Tests for the ring attention functionality."""
|
||||||
|
|
||||||
@@ -99,6 +73,11 @@ class TestRingAttention:
|
|||||||
self, mock_world_size, mock_rank, mock_new_group, partial_state
|
self, mock_world_size, mock_rank, mock_new_group, partial_state
|
||||||
):
|
):
|
||||||
"""Test that ring attention groups are created correctly."""
|
"""Test that ring attention groups are created correctly."""
|
||||||
|
from axolotl.monkeypatch.attention.ring_attn import (
|
||||||
|
RingAttnFunc,
|
||||||
|
register_ring_attn,
|
||||||
|
)
|
||||||
|
|
||||||
# Setup mocks
|
# Setup mocks
|
||||||
mock_world_size.return_value = 8 # 8 GPUs total
|
mock_world_size.return_value = 8 # 8 GPUs total
|
||||||
mock_rank.return_value = 3 # GPU #3
|
mock_rank.return_value = 3 # GPU #3
|
||||||
@@ -122,308 +101,88 @@ class TestRingAttention:
|
|||||||
set_ring_attn_group(None)
|
set_ring_attn_group(None)
|
||||||
|
|
||||||
|
|
||||||
class TestConfigValidation:
|
# Mock a simplified DataCollator test
|
||||||
"""Tests for validating sequence parallelism configurations."""
|
@patch("axolotl.monkeypatch.attention.ring_attn.get_ring_attn_group")
|
||||||
|
@patch("torch.distributed.get_rank")
|
||||||
|
@patch("torch.distributed.get_world_size")
|
||||||
|
def test_sequence_parallel_slicing(
|
||||||
|
mock_world_size, mock_rank, mock_get_group, partial_state
|
||||||
|
):
|
||||||
|
"""Test the basic sequence slicing logic without full collator instantiation."""
|
||||||
|
# Setup mocks
|
||||||
|
mock_get_group.return_value = MagicMock()
|
||||||
|
mock_rank.return_value = 1 # Second GPU
|
||||||
|
mock_world_size.return_value = 4 # 4 GPUs total
|
||||||
|
|
||||||
@pytest.fixture(autouse=True)
|
# Create a sample batch
|
||||||
def setup_mocks(self, monkeypatch):
|
batch = {
|
||||||
"""Set up mocks for all tests in this class."""
|
"input_ids": torch.tensor(
|
||||||
# Mock the ring_flash_attn module
|
[
|
||||||
monkeypatch.setitem(sys.modules, "ring_flash_attn", MagicMock())
|
[101, 102, 103, 104, 105, 106, 107, 108, 109, 110, 111, 112],
|
||||||
|
[201, 202, 203, 204, 205, 206, 207, 208, 209, 210, 211, 212],
|
||||||
|
]
|
||||||
|
),
|
||||||
|
"attention_mask": torch.ones(2, 12),
|
||||||
|
}
|
||||||
|
|
||||||
# Mock the is_main_process function to return True
|
# Simplified slicing logic from SequenceParallelDataCollator
|
||||||
monkeypatch.setattr(
|
def slice_batch(batch, rank, world_size):
|
||||||
"axolotl.utils.schemas.config.is_main_process", lambda: True
|
result = {}
|
||||||
)
|
for key in batch:
|
||||||
|
seq_len = batch[key].shape[1]
|
||||||
|
slice_size = seq_len // world_size
|
||||||
|
start_idx = rank * slice_size
|
||||||
|
end_idx = start_idx + slice_size if rank < world_size - 1 else seq_len
|
||||||
|
result[key] = batch[key][:, start_idx:end_idx]
|
||||||
|
return result
|
||||||
|
|
||||||
@pytest.fixture
|
# Slice the batch
|
||||||
def base_cfg(self):
|
result = slice_batch(
|
||||||
"""Create a base configuration for testing."""
|
batch, rank=mock_rank.return_value, world_size=mock_world_size.return_value
|
||||||
return DictDefault(
|
|
||||||
{
|
|
||||||
"base_model": "HuggingFaceTB/SmolLM2-135M",
|
|
||||||
"datasets": [{"path": "mhenrichsen/alpaca_2k_test", "type": "alpaca"}],
|
|
||||||
"micro_batch_size": 1,
|
|
||||||
"gradient_accumulation_steps": 1,
|
|
||||||
"learning_rate": 1e-3,
|
|
||||||
"output_dir": "./model-out",
|
|
||||||
"sequence_len": 512,
|
|
||||||
"special_tokens": {"pad_token": "<|endoftext|>"},
|
|
||||||
}
|
|
||||||
)
|
|
||||||
|
|
||||||
@pytest.mark.parametrize(
|
|
||||||
"config_updates, expected_values, should_pass, error_msg",
|
|
||||||
[
|
|
||||||
# Valid configuration
|
|
||||||
(
|
|
||||||
{"sequence_parallel_degree": 2, "flash_attention": True},
|
|
||||||
{"sequence_parallel_degree": 2, "flash_attention": True},
|
|
||||||
True,
|
|
||||||
None,
|
|
||||||
),
|
|
||||||
# Default sequence_parallel_degree
|
|
||||||
({}, {"sequence_parallel_degree": 1}, True, None),
|
|
||||||
# Invalid: sequence_parallel_degree > 1 without flash_attention
|
|
||||||
(
|
|
||||||
{"sequence_parallel_degree": 2, "flash_attention": False},
|
|
||||||
None,
|
|
||||||
False,
|
|
||||||
"flash_attention: true must be set",
|
|
||||||
),
|
|
||||||
# Invalid: sequence_parallel_degree > 1 with sample_packing and micro_batch_size > 1
|
|
||||||
(
|
|
||||||
{
|
|
||||||
"sequence_parallel_degree": 2,
|
|
||||||
"flash_attention": True,
|
|
||||||
"sample_packing": True,
|
|
||||||
"micro_batch_size": 2,
|
|
||||||
"pad_to_sequence_len": True,
|
|
||||||
},
|
|
||||||
None,
|
|
||||||
False,
|
|
||||||
"micro_batch_size must be set to 1",
|
|
||||||
),
|
|
||||||
],
|
|
||||||
ids=[
|
|
||||||
"valid_config",
|
|
||||||
"default_sp_degree",
|
|
||||||
"without_flash_attention",
|
|
||||||
"sample_packing_with_large_batch",
|
|
||||||
],
|
|
||||||
)
|
)
|
||||||
def test_sequence_parallel_config_validation(
|
|
||||||
self, base_cfg, config_updates, expected_values, should_pass, error_msg
|
|
||||||
):
|
|
||||||
"""Test various sequence parallelism configuration scenarios."""
|
|
||||||
from axolotl.utils.schemas.config import AxolotlInputConfig
|
|
||||||
|
|
||||||
# Apply updates to base config
|
# Check slicing
|
||||||
cfg = base_cfg
|
assert result["input_ids"].shape == (2, 3) # 12 tokens / 4 GPUs = 3 tokens per GPU
|
||||||
cfg.update(config_updates)
|
expected_input_ids = torch.tensor(
|
||||||
|
|
||||||
if should_pass:
|
|
||||||
# Should validate without errors
|
|
||||||
config = AxolotlInputConfig(**cfg)
|
|
||||||
|
|
||||||
# Check expected values
|
|
||||||
for key, value in expected_values.items():
|
|
||||||
assert getattr(config, key) == value
|
|
||||||
else:
|
|
||||||
# Should raise exception
|
|
||||||
with pytest.raises(ValueError) as excinfo:
|
|
||||||
AxolotlInputConfig(**cfg)
|
|
||||||
assert error_msg in str(excinfo.value)
|
|
||||||
|
|
||||||
@pytest.mark.parametrize(
|
|
||||||
"ring_attn_func, sample_packing, expected_func",
|
|
||||||
[
|
[
|
||||||
(None, True, RingAttnFunc.VARLEN_LLAMA3),
|
[104, 105, 106], # Second slice of first sequence
|
||||||
(None, False, RingAttnFunc.BATCH_RING),
|
[204, 205, 206], # Second slice of second sequence
|
||||||
],
|
]
|
||||||
ids=["default_with_sample_packing", "default_without_sample_packing"],
|
|
||||||
)
|
)
|
||||||
def test_ring_attn_func_validation(
|
assert torch.all(result["input_ids"] == expected_input_ids)
|
||||||
self, base_cfg, ring_attn_func, sample_packing, expected_func
|
|
||||||
):
|
|
||||||
"""Test ring_attn_func validation and defaults."""
|
|
||||||
from axolotl.utils.schemas.config import AxolotlInputConfig
|
|
||||||
|
|
||||||
# Apply updates to base config
|
|
||||||
cfg = base_cfg | {
|
|
||||||
"sequence_parallel_degree": 2,
|
|
||||||
"flash_attention": True,
|
|
||||||
"sample_packing": sample_packing,
|
|
||||||
}
|
|
||||||
|
|
||||||
if ring_attn_func is not None:
|
|
||||||
cfg["ring_attn_func"] = ring_attn_func
|
|
||||||
|
|
||||||
# Should validate without errors
|
|
||||||
config = AxolotlInputConfig(**cfg)
|
|
||||||
|
|
||||||
# Check ring_attn_func value
|
|
||||||
assert config.ring_attn_func.value == expected_func
|
|
||||||
|
|
||||||
def test_invalid_ring_attn_func(self, base_cfg):
|
|
||||||
"""Test that an invalid ring_attn_func is rejected."""
|
|
||||||
from axolotl.utils.schemas.config import AxolotlInputConfig
|
|
||||||
|
|
||||||
# Invalid configuration with invalid ring_attn_func
|
|
||||||
cfg = base_cfg | {
|
|
||||||
"sequence_parallel_degree": 2,
|
|
||||||
"flash_attention": True,
|
|
||||||
"ring_attn_func": "INVALID_FUNC",
|
|
||||||
}
|
|
||||||
|
|
||||||
# Should raise ValidationError
|
|
||||||
with pytest.raises(ValueError) as excinfo:
|
|
||||||
AxolotlInputConfig(**cfg)
|
|
||||||
|
|
||||||
# Verify error message
|
|
||||||
assert "ring_attn_func: INVALID_FUNC must be in" in str(excinfo.value)
|
|
||||||
|
|
||||||
|
|
||||||
class TestApplySequenceParallelism:
|
@patch.dict("sys.modules", {"ring_flash_attn": MagicMock()})
|
||||||
"""Tests for the apply_sequence_parallelism function."""
|
def test_config_validation_with_valid_inputs(cfg):
|
||||||
|
"""Test that valid sequence parallelism configurations pass validation."""
|
||||||
|
# Import the actual model class with appropriate mocks
|
||||||
|
from axolotl.utils.schemas.config import AxolotlInputConfig
|
||||||
|
|
||||||
@pytest.fixture(autouse=True)
|
# Valid configuration: sequence_parallel_degree > 1 and flash_attention is True
|
||||||
def mock_distributed(self, monkeypatch):
|
cfg = cfg | {
|
||||||
"""Mock torch.distributed functions for testing."""
|
"sequence_parallel_degree": 2,
|
||||||
# Mock is_initialized to return True
|
"flash_attention": True,
|
||||||
monkeypatch.setattr(torch.distributed, "is_initialized", lambda: True)
|
}
|
||||||
|
|
||||||
# Mock get_rank to return 0 by default
|
# Should validate without errors
|
||||||
monkeypatch.setattr(torch.distributed, "get_rank", lambda *args, **kwargs: 0)
|
config = AxolotlInputConfig(**cfg)
|
||||||
|
assert config.sequence_parallel_degree == 2
|
||||||
|
assert config.flash_attention is True
|
||||||
|
|
||||||
# Mock get_world_size to return 2 by default
|
|
||||||
monkeypatch.setattr(
|
|
||||||
torch.distributed, "get_world_size", lambda *args, **kwargs: 2
|
|
||||||
)
|
|
||||||
|
|
||||||
# Mock the process group
|
def test_config_validation_with_invalid_inputs(cfg):
|
||||||
monkeypatch.setattr(
|
"""Test that invalid sequence parallelism configurations fail validation."""
|
||||||
"axolotl.monkeypatch.attention.ring_attn.get_ring_attn_group",
|
from axolotl.utils.schemas.config import AxolotlInputConfig
|
||||||
MagicMock,
|
|
||||||
)
|
|
||||||
|
|
||||||
# Mock update_ring_attn_params
|
# Invalid configuration: sequence_parallel_degree > 1 but flash_attention is False
|
||||||
monkeypatch.setattr(
|
cfg = cfg | {
|
||||||
"axolotl.monkeypatch.attention.ring_attn.update_ring_attn_params",
|
"sequence_parallel_degree": 2,
|
||||||
lambda **kwargs: None,
|
"flash_attention": False,
|
||||||
)
|
}
|
||||||
|
|
||||||
def test_world_size_one(self, sequence_parallel_batch):
|
# Should raise ValidationError
|
||||||
"""Test that function returns original batch when world size is 1."""
|
with pytest.raises(ValueError) as excinfo:
|
||||||
result = apply_sequence_parallelism(
|
AxolotlInputConfig(**cfg)
|
||||||
batch=sequence_parallel_batch,
|
|
||||||
local_rank=0,
|
|
||||||
local_world_size=1,
|
|
||||||
ring_attn_func=RingAttnFunc.BATCH_RING,
|
|
||||||
)
|
|
||||||
|
|
||||||
# Should return the original batch unchanged
|
# Verify error message
|
||||||
assert result == sequence_parallel_batch
|
assert "flash_attention: true must be set" in str(excinfo.value)
|
||||||
|
|
||||||
def test_batch_ring_rank0(self, sequence_parallel_batch):
|
|
||||||
"""Test BATCH_RING sharding for rank 0 in a 2-process group."""
|
|
||||||
batch = sequence_parallel_batch
|
|
||||||
seq_len = batch["input_ids"].size(1)
|
|
||||||
|
|
||||||
result = apply_sequence_parallelism(
|
|
||||||
batch=batch,
|
|
||||||
local_rank=0,
|
|
||||||
local_world_size=2,
|
|
||||||
ring_attn_func=RingAttnFunc.BATCH_RING,
|
|
||||||
)
|
|
||||||
|
|
||||||
# Check that sequence dimension was sharded correctly
|
|
||||||
assert result["input_ids"].shape[1] == seq_len // 2
|
|
||||||
assert result["attention_mask"].shape[1] == seq_len // 2
|
|
||||||
|
|
||||||
# Verify content: rank 0 should get the first half of the sequence
|
|
||||||
assert torch.equal(result["input_ids"], batch["input_ids"][:, : seq_len // 2])
|
|
||||||
assert torch.equal(
|
|
||||||
result["position_ids"], batch["position_ids"][:, : seq_len // 2]
|
|
||||||
)
|
|
||||||
|
|
||||||
def test_batch_ring_rank1(self, sequence_parallel_batch):
|
|
||||||
"""Test BATCH_RING sharding for rank 1 in a 2-process group."""
|
|
||||||
batch = sequence_parallel_batch
|
|
||||||
seq_len = batch["input_ids"].size(1)
|
|
||||||
original_input_ids = batch["input_ids"].clone()
|
|
||||||
|
|
||||||
result = apply_sequence_parallelism(
|
|
||||||
batch=batch,
|
|
||||||
local_rank=1,
|
|
||||||
local_world_size=2,
|
|
||||||
ring_attn_func=RingAttnFunc.BATCH_RING,
|
|
||||||
)
|
|
||||||
|
|
||||||
# Verify content: rank 1 should get the second half of the sequence
|
|
||||||
assert torch.equal(result["input_ids"], original_input_ids[:, seq_len // 2 :])
|
|
||||||
|
|
||||||
def test_batch_zigzag(self, sequence_parallel_batch):
|
|
||||||
"""Test BATCH_ZIGZAG sharding pattern."""
|
|
||||||
batch = sequence_parallel_batch
|
|
||||||
original_input_ids = batch["input_ids"].clone()
|
|
||||||
seq_len = batch["input_ids"].size(1)
|
|
||||||
|
|
||||||
# Test rank 0
|
|
||||||
result_rank0 = apply_sequence_parallelism(
|
|
||||||
batch={k: v.clone() for k, v in batch.items()},
|
|
||||||
local_rank=0,
|
|
||||||
local_world_size=2,
|
|
||||||
ring_attn_func=RingAttnFunc.BATCH_ZIGZAG,
|
|
||||||
)
|
|
||||||
|
|
||||||
# Test rank 1
|
|
||||||
result_rank1 = apply_sequence_parallelism(
|
|
||||||
batch={k: v.clone() for k, v in batch.items()},
|
|
||||||
local_rank=1,
|
|
||||||
local_world_size=2,
|
|
||||||
ring_attn_func=RingAttnFunc.BATCH_ZIGZAG,
|
|
||||||
)
|
|
||||||
|
|
||||||
# Checks for both ranks
|
|
||||||
assert result_rank0["input_ids"].shape[1] == seq_len // 2
|
|
||||||
assert result_rank1["input_ids"].shape[1] == seq_len // 2
|
|
||||||
|
|
||||||
# For a 2-rank system with 8 tokens, check specific zigzag pattern
|
|
||||||
# Rank 0 should get chunks [0, 1] and [6, 7]
|
|
||||||
# Rank 1 should get chunks [2, 3] and [4, 5]
|
|
||||||
if seq_len == 8:
|
|
||||||
# Create expected tensors for comparison
|
|
||||||
rank0_expected = torch.cat(
|
|
||||||
[original_input_ids[:, :2], original_input_ids[:, 6:8]], dim=1
|
|
||||||
)
|
|
||||||
|
|
||||||
rank1_expected = torch.cat(
|
|
||||||
[original_input_ids[:, 2:4], original_input_ids[:, 4:6]], dim=1
|
|
||||||
)
|
|
||||||
|
|
||||||
assert torch.equal(result_rank0["input_ids"], rank0_expected)
|
|
||||||
assert torch.equal(result_rank1["input_ids"], rank1_expected)
|
|
||||||
|
|
||||||
def test_partial_application(self, sequence_parallel_batch):
|
|
||||||
"""Test that we can create a partially applied version of the function."""
|
|
||||||
batch = sequence_parallel_batch
|
|
||||||
original_input_ids = batch["input_ids"].clone()
|
|
||||||
|
|
||||||
# Create a partially applied function
|
|
||||||
rank0_ring_parallel = functools.partial(
|
|
||||||
apply_sequence_parallelism,
|
|
||||||
local_rank=0,
|
|
||||||
local_world_size=2,
|
|
||||||
ring_attn_func=RingAttnFunc.BATCH_RING,
|
|
||||||
)
|
|
||||||
|
|
||||||
# Use the partially applied function
|
|
||||||
result = rank0_ring_parallel(batch=batch)
|
|
||||||
|
|
||||||
# Verify it works as expected
|
|
||||||
assert result["input_ids"].shape[1] == original_input_ids.shape[1] // 2
|
|
||||||
assert torch.equal(
|
|
||||||
result["input_ids"],
|
|
||||||
original_input_ids[:, : original_input_ids.shape[1] // 2],
|
|
||||||
)
|
|
||||||
|
|
||||||
def test_missing_position_ids(self, sequence_parallel_batch):
|
|
||||||
"""Test handling of batch without position_ids."""
|
|
||||||
# Create a batch without position_ids
|
|
||||||
batch = {
|
|
||||||
k: v for k, v in sequence_parallel_batch.items() if k != "position_ids"
|
|
||||||
}
|
|
||||||
original_input_ids = batch["input_ids"].clone()
|
|
||||||
|
|
||||||
# This should run without error even though position_ids is missing
|
|
||||||
result = apply_sequence_parallelism(
|
|
||||||
batch=batch,
|
|
||||||
local_rank=0,
|
|
||||||
local_world_size=2,
|
|
||||||
ring_attn_func=RingAttnFunc.BATCH_RING,
|
|
||||||
)
|
|
||||||
|
|
||||||
# Verification should pass
|
|
||||||
assert "position_ids" not in result
|
|
||||||
assert result["input_ids"].shape[1] == original_input_ids.shape[1] // 2
|
|
||||||
|
|||||||
Reference in New Issue
Block a user