|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
import inspect |
|
import os |
|
import random |
|
import textwrap |
|
import warnings |
|
from collections import defaultdict |
|
from contextlib import contextmanager, nullcontext |
|
from copy import deepcopy |
|
from dataclasses import dataclass |
|
from typing import Any, Callable, Literal, Optional, Union |
|
|
|
import pandas as pd |
|
import torch |
|
import torch.amp as amp |
|
import torch.nn as nn |
|
import torch.nn.functional as F |
|
import transformers |
|
from accelerate import PartialState |
|
from accelerate.utils import is_deepspeed_available, tqdm |
|
from datasets import Dataset, IterableDataset |
|
from packaging import version |
|
from torch.utils.data import DataLoader |
|
from transformers import ( |
|
AutoModelForCausalLM, |
|
BaseImageProcessor, |
|
DataCollator, |
|
FeatureExtractionMixin, |
|
PreTrainedModel, |
|
PreTrainedTokenizerBase, |
|
ProcessorMixin, |
|
Trainer, |
|
is_comet_available, |
|
is_wandb_available, |
|
) |
|
from transformers.data.data_collator import DataCollatorMixin |
|
from transformers.models.auto.modeling_auto import MODEL_FOR_VISION_2_SEQ_MAPPING_NAMES |
|
from transformers.trainer_callback import TrainerCallback |
|
from transformers.trainer_utils import EvalLoopOutput |
|
from transformers.utils import is_peft_available, is_torch_xpu_available |
|
|
|
from ..data_utils import maybe_apply_chat_template, maybe_extract_prompt |
|
from ..models import PreTrainedModelWrapper, create_reference_model |
|
from ..models.utils import prepare_fsdp |
|
from .callbacks import SyncRefModelCallback |
|
from .dpo_config import DPOConfig, FDivergenceConstants, FDivergenceType |
|
from .utils import ( |
|
RunningMoments, |
|
cap_exp, |
|
disable_dropout_in_model, |
|
empty_cache, |
|
flush_left, |
|
generate_model_card, |
|
get_comet_experiment_url, |
|
log_table_to_comet_experiment, |
|
pad, |
|
pad_to_length, |
|
peft_module_casting_to_bf16, |
|
selective_log_softmax, |
|
) |
|
|
|
|
|
if is_peft_available(): |
|
from peft import PeftModel, get_peft_model, prepare_model_for_kbit_training |
|
|
|
|
|
if is_wandb_available(): |
|
import wandb |
|
|
|
if is_deepspeed_available(): |
|
import deepspeed |
|
|
|
|
|
@dataclass |
|
class DataCollatorForPreference(DataCollatorMixin): |
|
""" |
|
Data collator used for preference data. Inputs are dynamically padded to the maximum length of a batch if they |
|
are not all of the same length. |
|
|
|
Args: |
|
pad_token_id (`int`): |
|
Token ID to use for padding. |
|
return_tensors (`str`, *optional*, defaults to `"pt"`): |
|
Type of Tensor to return. Only `"pt"` is currently supported. |
|
|
|
Examples: |
|
```python |
|
>>> from trl import DataCollatorForPreference |
|
>>> collator = DataCollatorForPreference(pad_token_id=0) |
|
>>> examples = [ |
|
... {"prompt_input_ids": [1, 2, 3], "chosen_input_ids": [4, 5], "rejected_input_ids": [6]}, |
|
... {"prompt_input_ids": [7, 8], "chosen_input_ids": [9, 10], "rejected_input_ids": [11, 12, 13]} |
|
... ] |
|
>>> collator(examples) |
|
{'prompt_input_ids': tensor([[1, 2, 3], |
|
[0, 7, 8]]), |
|
'prompt_attention_mask': tensor([[1, 1, 1], |
|
[0, 1, 1]]), |
|
'chosen_input_ids': tensor([[ 4, 5], |
|
[ 9, 10]]), |
|
'chosen_attention_mask': tensor([[1, 1], |
|
[1, 1]]), |
|
'rejected_input_ids': tensor([[ 6, 0, 0], |
|
[11, 12, 13]]), |
|
'rejected_attention_mask': tensor([[1, 0, 0], |
|
[1, 1, 1]]) |
|
} |
|
``` |
|
""" |
|
|
|
pad_token_id: int |
|
return_tensors: str = "pt" |
|
|
|
def torch_call(self, examples: list[Union[list[int], Any, dict[str, Any]]]) -> dict[str, Any]: |
|
|
|
prompt_input_ids = [torch.tensor(example["prompt_input_ids"]) for example in examples] |
|
prompt_attention_mask = [torch.ones_like(input_ids) for input_ids in prompt_input_ids] |
|
chosen_input_ids = [torch.tensor(example["chosen_input_ids"]) for example in examples] |
|
chosen_attention_mask = [torch.ones_like(input_ids) for input_ids in chosen_input_ids] |
|
rejected_input_ids = [torch.tensor(example["rejected_input_ids"]) for example in examples] |
|
rejected_attention_mask = [torch.ones_like(input_ids) for input_ids in rejected_input_ids] |
|
if "pixel_values" in examples[0]: |
|
pixel_values = [torch.tensor(example["pixel_values"]) for example in examples] |
|
if "pixel_attention_mask" in examples[0]: |
|
pixel_attention_mask = [torch.tensor(example["pixel_attention_mask"]) for example in examples] |
|
if "ref_chosen_logps" in examples[0] and "ref_rejected_logps" in examples[0]: |
|
ref_chosen_logps = torch.tensor([example["ref_chosen_logps"] for example in examples]) |
|
ref_rejected_logps = torch.tensor([example["ref_rejected_logps"] for example in examples]) |
|
|
|
|
|
output = {} |
|
output["prompt_input_ids"] = pad(prompt_input_ids, padding_value=self.pad_token_id, padding_side="left") |
|
output["prompt_attention_mask"] = pad(prompt_attention_mask, padding_value=0, padding_side="left") |
|
output["chosen_input_ids"] = pad(chosen_input_ids, padding_value=self.pad_token_id) |
|
output["chosen_attention_mask"] = pad(chosen_attention_mask, padding_value=0) |
|
output["rejected_input_ids"] = pad(rejected_input_ids, padding_value=self.pad_token_id) |
|
output["rejected_attention_mask"] = pad(rejected_attention_mask, padding_value=0) |
|
if "pixel_values" in examples[0]: |
|
output["pixel_values"] = pad(pixel_values, padding_value=0.0) |
|
if "pixel_attention_mask" in examples[0]: |
|
output["pixel_attention_mask"] = pad(pixel_attention_mask, padding_value=0) |
|
if "image_sizes" in examples[0]: |
|
output["image_sizes"] = torch.tensor([example["image_sizes"] for example in examples]) |
|
if "ref_chosen_logps" in examples[0] and "ref_rejected_logps" in examples[0]: |
|
output["ref_chosen_logps"] = ref_chosen_logps |
|
output["ref_rejected_logps"] = ref_rejected_logps |
|
|
|
return output |
|
|
|
|
|
class DPOTrainer(Trainer): |
|
r""" |
|
Initialize DPOTrainer. |
|
|
|
Args: |
|
model (`transformers.PreTrainedModel`): |
|
The model to train, preferably an `AutoModelForSequenceClassification`. |
|
ref_model (`PreTrainedModelWrapper`): |
|
Hugging Face transformer model with a casual language modelling head. Used for implicit reward computation and loss. If no |
|
reference model is provided, the trainer will create a reference model with the same architecture as the model to be optimized. |
|
args (`DPOConfig`): |
|
The DPO config arguments to use for training. |
|
data_collator (`transformers.DataCollator`): |
|
The data collator to use for training. If None is specified, the default data collator (`DataCollatorForPreference`) will be used |
|
which will pad the sequences to the maximum length of the sequences in the batch, given a dataset of paired sequences. |
|
train_dataset (`datasets.Dataset`): |
|
The dataset to use for training. |
|
eval_dataset (`datasets.Dataset`): |
|
The dataset to use for evaluation. |
|
processing_class (`PreTrainedTokenizerBase` or `BaseImageProcessor` or `FeatureExtractionMixin` or `ProcessorMixin`, *optional*): |
|
Processing class used to process the data. If provided, will be used to automatically process the inputs |
|
for the model, and it will be saved along the model to make it easier to rerun an interrupted training or |
|
reuse the fine-tuned model. |
|
This supercedes the `tokenizer` argument, which is now deprecated. |
|
model_init (`Callable[[], transformers.PreTrainedModel]`): |
|
The model initializer to use for training. If None is specified, the default model initializer will be used. |
|
compute_metrics (`Callable[[EvalPrediction], dict]`, *optional*): |
|
The function to use to compute the metrics. Must take a `EvalPrediction` and return |
|
a dictionary string to metric values. |
|
callbacks (`list[transformers.TrainerCallback]`): |
|
The callbacks to use for training. |
|
optimizers (`tuple[torch.optim.Optimizer, torch.optim.lr_scheduler.LambdaLR]`): |
|
The optimizer and scheduler to use for training. |
|
preprocess_logits_for_metrics (`Callable[[torch.Tensor, torch.Tensor], torch.Tensor]`): |
|
The function to use to preprocess the logits before computing the metrics. |
|
peft_config (`dict`, defaults to `None`): |
|
The PEFT configuration to use for training. If you pass a PEFT configuration, the model will be wrapped in a PEFT model. |
|
""" |
|
|
|
_tag_names = ["trl", "dpo"] |
|
|
|
def __init__( |
|
self, |
|
model: Optional[Union[PreTrainedModel, nn.Module, str]] = None, |
|
ref_model: Optional[Union[PreTrainedModel, nn.Module, str]] = None, |
|
args: Optional[DPOConfig] = None, |
|
data_collator: Optional[DataCollator] = None, |
|
train_dataset: Optional[Dataset] = None, |
|
eval_dataset: Optional[Union[Dataset, dict[str, Dataset]]] = None, |
|
processing_class: Optional[ |
|
Union[PreTrainedTokenizerBase, BaseImageProcessor, FeatureExtractionMixin, ProcessorMixin] |
|
] = None, |
|
model_init: Optional[Callable[[], PreTrainedModel]] = None, |
|
compute_metrics: Optional[Callable[[EvalLoopOutput], dict]] = None, |
|
callbacks: Optional[list[TrainerCallback]] = None, |
|
optimizers: tuple[torch.optim.Optimizer, torch.optim.lr_scheduler.LambdaLR] = (None, None), |
|
preprocess_logits_for_metrics: Optional[Callable[[torch.Tensor, torch.Tensor], torch.Tensor]] = None, |
|
peft_config: Optional[dict] = None, |
|
): |
|
if model is None: |
|
raise ValueError("No model provided. Please provide a model to train.") |
|
|
|
if not isinstance(model, str) and ref_model is model: |
|
raise ValueError( |
|
"`model` and `ref_model` cannot be the same object. If you want `ref_model` to be the " |
|
"same as `model`, you must mass a copy of it, or `None` if you use peft." |
|
) |
|
|
|
if args.model_init_kwargs is None: |
|
model_init_kwargs = {} |
|
elif not isinstance(model, str): |
|
raise ValueError( |
|
"You passed model_init_kwargs to the DPOTrainer/DPOConfig, but your model is already instantiated." |
|
) |
|
else: |
|
model_init_kwargs = args.model_init_kwargs |
|
torch_dtype = model_init_kwargs.get("torch_dtype") |
|
if torch_dtype is not None: |
|
|
|
if isinstance(torch_dtype, str) and torch_dtype != "auto": |
|
torch_dtype = getattr(torch, torch_dtype) |
|
if torch_dtype != "auto" and not isinstance(torch_dtype, torch.dtype): |
|
raise ValueError( |
|
f"Invalid `torch_dtype` passed to the DPOConfig. Expected a string with either `torch.dtype` or 'auto', but got {torch_dtype}." |
|
) |
|
model_init_kwargs["torch_dtype"] = torch_dtype |
|
|
|
if args.ref_model_init_kwargs is None: |
|
ref_model_init_kwargs = {} |
|
elif not isinstance(ref_model, str): |
|
raise ValueError( |
|
"You passed ref_model_init_kwargs to the DPOTrainer/DPOConfig, but your ref_model is already instantiated." |
|
) |
|
else: |
|
ref_model_init_kwargs = args.ref_model_init_kwargs |
|
torch_dtype = ref_model_init_kwargs.get("torch_dtype") |
|
if torch_dtype is not None: |
|
|
|
if isinstance(torch_dtype, str) and torch_dtype != "auto": |
|
torch_dtype = getattr(torch, torch_dtype) |
|
if torch_dtype != "auto" and not isinstance(torch_dtype, torch.dtype): |
|
raise ValueError( |
|
f"Invalid `torch_dtype` passed to the DPOConfig. Expected a string with either `torch.dtype` or 'auto', but got {torch_dtype}." |
|
) |
|
ref_model_init_kwargs["torch_dtype"] = torch_dtype |
|
|
|
if isinstance(model, str): |
|
model = AutoModelForCausalLM.from_pretrained(model, **model_init_kwargs) |
|
|
|
if isinstance(ref_model, str): |
|
ref_model = AutoModelForCausalLM.from_pretrained(ref_model, **ref_model_init_kwargs) |
|
|
|
|
|
|
|
self._peft_has_been_casted_to_bf16 = False |
|
|
|
if not is_peft_available() and peft_config is not None: |
|
raise ValueError( |
|
"PEFT is not installed and you passed a `peft_config` in the trainer's kwargs, please install it to use the PEFT models" |
|
) |
|
elif is_peft_available() and peft_config is not None: |
|
|
|
if isinstance(model, PeftModel): |
|
model = model.merge_and_unload() |
|
|
|
if ref_model is not None and not args.force_use_ref_model: |
|
raise ValueError( |
|
"You passed both a ref_model and a peft_config. For training PEFT adapters with DPO there is no need to pass a reference" |
|
" model. Please pass `ref_model=None` in case you want to train PEFT adapters, or pass a ref_model with `force_use_ref_model=True` in DPOTrainer's init." |
|
" if you want to use a different ref_model." |
|
) |
|
|
|
if getattr(model, "is_loaded_in_8bit", False) or getattr(model, "is_loaded_in_4bit", False): |
|
_support_gc_kwargs = hasattr( |
|
args, "gradient_checkpointing_kwargs" |
|
) and "gradient_checkpointing_kwargs" in list( |
|
inspect.signature(prepare_model_for_kbit_training).parameters |
|
) |
|
|
|
prepare_model_kwargs = {"use_gradient_checkpointing": args.gradient_checkpointing} |
|
|
|
if _support_gc_kwargs: |
|
prepare_model_kwargs["gradient_checkpointing_kwargs"] = args.gradient_checkpointing_kwargs |
|
|
|
model = prepare_model_for_kbit_training(model, **prepare_model_kwargs) |
|
elif getattr(args, "gradient_checkpointing", False): |
|
|
|
if hasattr(model, "enable_input_require_grads"): |
|
model.enable_input_require_grads() |
|
else: |
|
|
|
def make_inputs_require_grad(module, input, output): |
|
output.requires_grad_(True) |
|
|
|
model.get_input_embeddings().register_forward_hook(make_inputs_require_grad) |
|
|
|
|
|
model = get_peft_model(model, peft_config) |
|
if args.bf16 and getattr(model, "is_loaded_in_4bit", False): |
|
peft_module_casting_to_bf16(model) |
|
|
|
self._peft_has_been_casted_to_bf16 = True |
|
|
|
|
|
|
|
|
|
elif getattr(args, "gradient_checkpointing", False): |
|
|
|
if hasattr(model, "enable_input_require_grads"): |
|
model.enable_input_require_grads() |
|
else: |
|
|
|
def make_inputs_require_grad(module, input, output): |
|
output.requires_grad_(True) |
|
|
|
model.get_input_embeddings().register_forward_hook(make_inputs_require_grad) |
|
|
|
if args.generate_during_eval and not (is_wandb_available() or is_comet_available()): |
|
raise ValueError( |
|
"`generate_during_eval=True` requires Weights and Biases or Comet to be installed." |
|
" Please install `wandb` or `comet-ml` to resolve." |
|
) |
|
|
|
self.is_encoder_decoder = model.config.is_encoder_decoder |
|
self.is_vision_model = model.config.model_type in MODEL_FOR_VISION_2_SEQ_MAPPING_NAMES.keys() |
|
self.is_peft_model = is_peft_available() and isinstance(model, PeftModel) |
|
self.model_adapter_name = args.model_adapter_name |
|
self.ref_adapter_name = args.ref_adapter_name |
|
self.reference_free = args.reference_free |
|
|
|
if ref_model: |
|
self.ref_model = ref_model |
|
elif self.is_peft_model or args.precompute_ref_log_probs: |
|
|
|
self.ref_model = None |
|
else: |
|
self.ref_model = create_reference_model(model) |
|
|
|
if processing_class is None: |
|
raise ValueError("processing_class must be specified to tokenize a DPO dataset.") |
|
|
|
if args.padding_value is not None: |
|
self.padding_value = args.padding_value |
|
else: |
|
if hasattr(processing_class, "pad_token_id") and processing_class.pad_token_id is not None: |
|
self.padding_value = processing_class.pad_token_id |
|
elif hasattr(processing_class, "tokenizer") and processing_class.tokenizer.pad_token_id is not None: |
|
self.padding_value = processing_class.tokenizer.pad_token_id |
|
else: |
|
raise ValueError( |
|
"`padding_value` is not specified in `DPOConfig`, and `pad_token_id` is missing in the " |
|
"`processing_class`. Please either set the `padding_value` argument in `DPOConfig`, or set " |
|
"`tokenizer.pad_token` (e.g., `tokenizer.pad_token = tokenizer.eos_token`) before instantiating " |
|
"the trainer." |
|
) |
|
|
|
if data_collator is None: |
|
data_collator = DataCollatorForPreference(pad_token_id=self.padding_value) |
|
|
|
|
|
if args.disable_dropout: |
|
disable_dropout_in_model(model) |
|
if self.ref_model is not None: |
|
disable_dropout_in_model(self.ref_model) |
|
|
|
self.generate_during_eval = args.generate_during_eval |
|
self.label_pad_token_id = args.label_pad_token_id |
|
self.max_prompt_length = args.max_prompt_length |
|
self.max_completion_length = args.max_completion_length |
|
self.max_length = args.max_length |
|
self.truncation_mode = args.truncation_mode |
|
self.precompute_ref_log_probs = args.precompute_ref_log_probs |
|
self.use_logits_to_keep = args.use_logits_to_keep |
|
|
|
if args.padding_free: |
|
if model.config._attn_implementation != "flash_attention_2": |
|
warnings.warn( |
|
"Padding-free training is enabled, but the attention implementation is not set to " |
|
"'flash_attention_2'. Padding-free training flattens batches into a single sequence, and " |
|
"'flash_attention_2' is the only known attention mechanism that reliably supports this. Using " |
|
"other implementations may lead to unexpected behavior. To ensure compatibility, set " |
|
"`attn_implementation='flash_attention_2'` in the model configuration, or verify that your " |
|
"attention mechanism can handle flattened sequences." |
|
) |
|
self.padding_free = args.padding_free |
|
|
|
|
|
|
|
self._precomputed_train_ref_log_probs = False |
|
self._precomputed_eval_ref_log_probs = False |
|
|
|
if ( |
|
args.loss_type in ["hinge", "ipo", "bco_pair", "sppo_hard", "nca_pair", "apo_zero", "apo_down"] |
|
and args.label_smoothing > 0 |
|
): |
|
warnings.warn( |
|
f"You are using the {args.loss_type} loss type that does not support label smoothing. The " |
|
"`label_smoothing` parameter will be ignored. Set `label_smoothing` to `0.0` to remove this warning.", |
|
UserWarning, |
|
) |
|
if args.loss_type == "kto_pair": |
|
raise ValueError("Support for kto_pair has been removed in DPOTrainer. Please use KTOTrainer.") |
|
|
|
self.beta = args.beta |
|
self.label_smoothing = args.label_smoothing |
|
self.loss_type = args.loss_type |
|
self.aux_loss_enabled = getattr(model.config, "output_router_logits", False) |
|
self.use_weighting = args.use_weighting |
|
self.aux_loss_coef = getattr(model.config, "router_aux_loss_coef", 0.0) |
|
if self.aux_loss_enabled and self.aux_loss_coef == 0.0: |
|
warnings.warn( |
|
"You set `output_router_logits` to `True` in the model config, but `router_aux_loss_coef` is set to " |
|
"`0.0`, meaning the auxiliary loss will not be used. Either set `router_aux_loss_coef` to a value " |
|
"greater than `0.0`, or set `output_router_logits` to `False` if you don't want to use the auxiliary " |
|
"loss.", |
|
UserWarning, |
|
) |
|
|
|
self._stored_metrics = defaultdict(lambda: defaultdict(list)) |
|
self.f_divergence_type = args.f_divergence_type |
|
self.f_divergence_params = {FDivergenceConstants.ALPHA_DIVERGENCE_COEF_KEY: args.f_alpha_divergence_coef} |
|
self.dataset_num_proc = args.dataset_num_proc |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
model.warnings_issued["estimate_tokens"] = True |
|
|
|
|
|
train_dataset = self._prepare_dataset(train_dataset, processing_class, args, "train") |
|
if eval_dataset is not None: |
|
if isinstance(eval_dataset, dict): |
|
eval_dataset = { |
|
key: self._prepare_dataset(dataset, processing_class, args, key) |
|
for key, dataset in eval_dataset.items() |
|
} |
|
else: |
|
eval_dataset = self._prepare_dataset(eval_dataset, processing_class, args, "eval") |
|
|
|
super().__init__( |
|
model=model, |
|
args=args, |
|
data_collator=data_collator, |
|
train_dataset=train_dataset, |
|
eval_dataset=eval_dataset, |
|
processing_class=processing_class, |
|
model_init=model_init, |
|
compute_metrics=compute_metrics, |
|
callbacks=callbacks, |
|
optimizers=optimizers, |
|
preprocess_logits_for_metrics=preprocess_logits_for_metrics, |
|
) |
|
|
|
|
|
|
|
|
|
self.model_accepts_loss_kwargs = False |
|
|
|
|
|
if hasattr(self.model, "add_model_tags"): |
|
self.model.add_model_tags(self._tag_names) |
|
|
|
if not hasattr(self, "accelerator"): |
|
raise AttributeError( |
|
"Your `Trainer` does not have an `accelerator` object. Consider upgrading `transformers`." |
|
) |
|
|
|
|
|
if self.is_deepspeed_enabled: |
|
if self.accelerator.state.deepspeed_plugin.zero_stage == 3 and self.precompute_ref_log_probs: |
|
raise ValueError( |
|
"You cannot use `precompute_ref_log_probs=True` with Deepspeed ZeRO-3. Please set `precompute_ref_log_probs=False`." |
|
) |
|
|
|
if self.ref_model is None: |
|
if not (self.is_peft_model or self.precompute_ref_log_probs): |
|
raise ValueError( |
|
"No reference model and model is not a Peft model. Try setting `precompute_ref_log_probs=True`" |
|
) |
|
if args.sync_ref_model: |
|
raise ValueError( |
|
"You currently cannot use `ref_model=None` with TR-DPO method. Please provide `ref_model`." |
|
) |
|
else: |
|
if self.is_deepspeed_enabled: |
|
self.ref_model = self._prepare_deepspeed(self.ref_model) |
|
elif self.is_fsdp_enabled: |
|
self.ref_model = prepare_fsdp(self.ref_model, self.accelerator) |
|
else: |
|
self.ref_model = self.accelerator.prepare_model(self.ref_model, evaluation_mode=True) |
|
|
|
if args.sync_ref_model: |
|
if self.precompute_ref_log_probs: |
|
raise ValueError( |
|
"You cannot use `precompute_ref_log_probs=True` with TR-DPO method. Please set `precompute_ref_log_probs=False`." |
|
) |
|
|
|
self.add_callback(SyncRefModelCallback(ref_model=self.ref_model, accelerator=self.accelerator)) |
|
|
|
if self.loss_type == "bco_pair": |
|
self.running = RunningMoments(self.accelerator) |
|
|
|
def _prepare_dataset( |
|
self, |
|
dataset: Union[Dataset, IterableDataset], |
|
processing_class: Union[PreTrainedTokenizerBase, BaseImageProcessor, FeatureExtractionMixin, ProcessorMixin], |
|
args: DPOConfig, |
|
dataset_name: str, |
|
) -> Union[Dataset, IterableDataset]: |
|
|
|
map_kwargs = {"writer_batch_size": 10} |
|
if isinstance(dataset, Dataset): |
|
map_kwargs["num_proc"] = args.dataset_num_proc |
|
|
|
with PartialState().main_process_first(): |
|
|
|
if isinstance(dataset, Dataset): |
|
map_kwargs["desc"] = f"Extracting prompt in {dataset_name} dataset" |
|
dataset = dataset.map(maybe_extract_prompt, **map_kwargs) |
|
|
|
|
|
if isinstance(dataset, Dataset): |
|
map_kwargs["desc"] = f"Applying chat template to {dataset_name} dataset" |
|
dataset = dataset.map( |
|
maybe_apply_chat_template, fn_kwargs={"tokenizer": processing_class, "tools": args.tools}, **map_kwargs |
|
) |
|
|
|
|
|
if isinstance(dataset, Dataset): |
|
map_kwargs["desc"] = f"Tokenizing {dataset_name} dataset" |
|
|
|
dataset = dataset.map( |
|
self.tokenize_row if not self.is_vision_model else self.process_row, |
|
remove_columns=["prompt", "chosen", "rejected"], |
|
fn_kwargs={ |
|
"processing_class": processing_class, |
|
"max_prompt_length": args.max_prompt_length, |
|
"max_completion_length": args.max_completion_length, |
|
|
|
"add_special_tokens": False, |
|
}, |
|
**map_kwargs, |
|
) |
|
|
|
return dataset |
|
|
|
@staticmethod |
|
def tokenize_row(features, processing_class, max_prompt_length, max_completion_length, add_special_tokens): |
|
""" |
|
Tokenize a row of the dataset. |
|
|
|
Args: |
|
features (`dict[str, str]`): |
|
Row of the dataset, should contain the keys `"prompt"`, `"chosen"`, and `"rejected"`. |
|
processing_class (`PreTrainedTokenizerBase`): |
|
Processing class used to process the data. |
|
max_prompt_length (`int` or `None`): |
|
Maximum length of the prompt sequence. If `None`, the prompt sequence is not truncated. |
|
max_completion_length (`int` or `None`): |
|
Maximum length of the completion sequences. If `None`, the completion sequences are not truncated. |
|
add_special_tokens (`bool`): |
|
Whether to add special tokens to the sequences. Typically used for encoder-decoder models. If `True`, |
|
the prompt sequence will have a bos token prepended and an eos token appended. In any case, the |
|
completion sequences will have an eos token appended. |
|
|
|
Returns: |
|
`dict[str, list[int]]`: |
|
Tokenized sequences with the keys `"prompt_input_ids"`, `"chosen_input_ids"`, and |
|
`"rejected_input_ids". |
|
|
|
Example: |
|
```python |
|
>>> from transformers import GPT2Tokenizer |
|
>>> tokenizer = GPT2Tokenizer.from_pretrained("gpt2") |
|
>>> features = {"prompt": "The sky is", "chosen": " blue", "rejected": " green"} |
|
>>> DPOTrainer.tokenize_row( |
|
... features, tokenizer, max_prompt_length=3, max_completion_length=3, add_special_tokens=False |
|
... ) |
|
{'prompt_input_ids': [464, 6766, 318], 'chosen_input_ids': [4171, 50256], 'rejected_input_ids': [4077, 50256]} |
|
``` |
|
""" |
|
tokenizer = processing_class |
|
prompt_input_ids = tokenizer(features["prompt"], add_special_tokens=False)["input_ids"] |
|
chosen_input_ids = tokenizer(features["chosen"], add_special_tokens=False)["input_ids"] |
|
rejected_input_ids = tokenizer(features["rejected"], add_special_tokens=False)["input_ids"] |
|
|
|
|
|
if add_special_tokens: |
|
if tokenizer.bos_token_id is not None: |
|
prompt_input_ids = [tokenizer.bos_token_id] + prompt_input_ids |
|
if tokenizer.eos_token_id is not None: |
|
prompt_input_ids = prompt_input_ids + [tokenizer.eos_token_id] |
|
chosen_input_ids = chosen_input_ids + [tokenizer.eos_token_id] |
|
rejected_input_ids = rejected_input_ids + [tokenizer.eos_token_id] |
|
|
|
|
|
if max_prompt_length is not None: |
|
prompt_input_ids = prompt_input_ids[-max_prompt_length:] |
|
if max_completion_length is not None: |
|
chosen_input_ids = chosen_input_ids[:max_completion_length] |
|
rejected_input_ids = rejected_input_ids[:max_completion_length] |
|
|
|
return { |
|
"prompt_input_ids": prompt_input_ids, |
|
"chosen_input_ids": chosen_input_ids, |
|
"rejected_input_ids": rejected_input_ids, |
|
} |
|
|
|
@staticmethod |
|
def process_row(features, processing_class, max_prompt_length, max_completion_length, add_special_tokens): |
|
""" |
|
Same as `tokenize_row` but for vision models. Please refer to `tokenize_row` for more information. |
|
""" |
|
processor, tokenizer = processing_class, processing_class.tokenizer |
|
processed_features = processor(images=features["images"], text=features["prompt"], add_special_tokens=False) |
|
|
|
prompt_input_ids = processed_features["input_ids"][0] |
|
pixel_values = processed_features["pixel_values"][0] |
|
chosen_input_ids = tokenizer(features["chosen"], add_special_tokens=False)["input_ids"] |
|
rejected_input_ids = tokenizer(features["rejected"], add_special_tokens=False)["input_ids"] |
|
|
|
|
|
if add_special_tokens: |
|
if tokenizer.bos_token_id is not None: |
|
prompt_input_ids = [tokenizer.bos_token_id] + prompt_input_ids |
|
if tokenizer.eos_token_id is not None: |
|
prompt_input_ids = prompt_input_ids + [tokenizer.eos_token_id] |
|
chosen_input_ids = chosen_input_ids + [tokenizer.eos_token_id] |
|
rejected_input_ids = rejected_input_ids + [tokenizer.eos_token_id] |
|
|
|
|
|
if max_prompt_length is not None: |
|
prompt_input_ids = prompt_input_ids[-max_prompt_length:] |
|
if max_completion_length is not None: |
|
chosen_input_ids = chosen_input_ids[:max_completion_length] |
|
rejected_input_ids = rejected_input_ids[:max_completion_length] |
|
|
|
output = { |
|
"prompt_input_ids": prompt_input_ids, |
|
"pixel_values": pixel_values, |
|
"chosen_input_ids": chosen_input_ids, |
|
"rejected_input_ids": rejected_input_ids, |
|
} |
|
|
|
if "pixel_attention_mask" in processed_features: |
|
output["pixel_attention_mask"] = processed_features["pixel_attention_mask"][0] |
|
if "image_sizes" in processed_features: |
|
output["image_sizes"] = processed_features["image_sizes"][0] |
|
|
|
return output |
|
|
|
def _prepare_deepspeed(self, model: PreTrainedModelWrapper): |
|
|
|
deepspeed_plugin = self.accelerator.state.deepspeed_plugin |
|
config_kwargs = deepcopy(deepspeed_plugin.deepspeed_config) |
|
|
|
if model is not None: |
|
if hasattr(model, "config"): |
|
hidden_size = ( |
|
max(model.config.hidden_sizes) |
|
if getattr(model.config, "hidden_sizes", None) |
|
else getattr(model.config, "hidden_size", None) |
|
) |
|
if hidden_size is not None and config_kwargs["zero_optimization"]["stage"] == 3: |
|
|
|
|
|
config_kwargs.update( |
|
{ |
|
"zero_optimization.reduce_bucket_size": hidden_size * hidden_size, |
|
"zero_optimization.stage3_param_persistence_threshold": 10 * hidden_size, |
|
"zero_optimization.stage3_prefetch_bucket_size": 0.9 * hidden_size * hidden_size, |
|
} |
|
) |
|
|
|
|
|
|
|
if config_kwargs["zero_optimization"]["stage"] != 3: |
|
config_kwargs["zero_optimization"]["stage"] = 0 |
|
model, *_ = deepspeed.initialize(model=model, config=config_kwargs) |
|
model.eval() |
|
return model |
|
|
|
def _set_signature_columns_if_needed(self): |
|
|
|
|
|
|
|
|
|
if self._signature_columns is None: |
|
self._signature_columns = [ |
|
"prompt_input_ids", |
|
"chosen_input_ids", |
|
"rejected_input_ids", |
|
"image_sizes", |
|
"ref_chosen_logps", |
|
"ref_rejected_logps", |
|
] |
|
|
|
def get_train_dataloader(self) -> DataLoader: |
|
""" |
|
Returns the training [`~torch.utils.data.DataLoader`]. |
|
|
|
Subclass of transformers.src.transformers.trainer.get_train_dataloader to precompute `ref_log_probs`. |
|
""" |
|
|
|
if self.precompute_ref_log_probs and not self._precomputed_train_ref_log_probs: |
|
batch_size = self.args.precompute_ref_batch_size or self.args.per_device_train_batch_size |
|
dataloader_params = { |
|
"batch_size": batch_size, |
|
"collate_fn": self.data_collator, |
|
"num_workers": self.args.dataloader_num_workers, |
|
"pin_memory": self.args.dataloader_pin_memory, |
|
"shuffle": False, |
|
} |
|
|
|
|
|
data_loader = self.accelerator.prepare(DataLoader(self.train_dataset, **dataloader_params)) |
|
|
|
ref_chosen_logps = [] |
|
ref_rejected_logps = [] |
|
for padded_batch in tqdm(iterable=data_loader, desc="Train dataset reference log probs"): |
|
ref_chosen_logp, ref_rejected_logp = self.compute_ref_log_probs(padded_batch) |
|
ref_chosen_logp, ref_rejected_logp = self.accelerator.gather_for_metrics( |
|
(ref_chosen_logp, ref_rejected_logp) |
|
) |
|
ref_chosen_logps.append(ref_chosen_logp.cpu()) |
|
ref_rejected_logps.append(ref_rejected_logp.cpu()) |
|
|
|
|
|
empty_cache() |
|
self.accelerator.free_memory() |
|
|
|
all_ref_chosen_logps = torch.cat(ref_chosen_logps).float().numpy() |
|
all_ref_rejected_logps = torch.cat(ref_rejected_logps).float().numpy() |
|
|
|
self.train_dataset = self.train_dataset.add_column(name="ref_chosen_logps", column=all_ref_chosen_logps) |
|
self.train_dataset = self.train_dataset.add_column( |
|
name="ref_rejected_logps", column=all_ref_rejected_logps |
|
) |
|
|
|
self._precomputed_train_ref_log_probs = True |
|
|
|
return super().get_train_dataloader() |
|
|
|
def get_eval_dataloader(self, eval_dataset: Optional[Dataset] = None) -> DataLoader: |
|
""" |
|
Returns the evaluation [`~torch.utils.data.DataLoader`]. |
|
|
|
Subclass of transformers.src.transformers.trainer.get_eval_dataloader to precompute `ref_log_probs`. |
|
|
|
Args: |
|
eval_dataset (`torch.utils.data.Dataset`, *optional*): |
|
If provided, will override `self.eval_dataset`. If it is a [`~datasets.Dataset`], columns not accepted |
|
by the `model.forward()` method are automatically removed. It must implement `__len__`. |
|
""" |
|
if eval_dataset is None and self.eval_dataset is None: |
|
raise ValueError("Trainer: evaluation requires an eval_dataset.") |
|
eval_dataset = eval_dataset if eval_dataset is not None else self.eval_dataset |
|
|
|
if self.precompute_ref_log_probs and not self._precomputed_eval_ref_log_probs: |
|
batch_size = self.args.precompute_ref_batch_size or self.args.per_device_eval_batch_size |
|
dataloader_params = { |
|
"batch_size": batch_size, |
|
"collate_fn": self.data_collator, |
|
"num_workers": self.args.dataloader_num_workers, |
|
"pin_memory": self.args.dataloader_pin_memory, |
|
"shuffle": False, |
|
} |
|
|
|
|
|
data_loader = self.accelerator.prepare(DataLoader(eval_dataset, **dataloader_params)) |
|
|
|
ref_chosen_logps = [] |
|
ref_rejected_logps = [] |
|
for padded_batch in tqdm(iterable=data_loader, desc="Eval dataset reference log probs"): |
|
ref_chosen_logp, ref_rejected_logp = self.compute_ref_log_probs(padded_batch) |
|
ref_chosen_logp, ref_rejected_logp = self.accelerator.gather_for_metrics( |
|
(ref_chosen_logp, ref_rejected_logp) |
|
) |
|
ref_chosen_logps.append(ref_chosen_logp.cpu()) |
|
ref_rejected_logps.append(ref_rejected_logp.cpu()) |
|
|
|
all_ref_chosen_logps = torch.cat(ref_chosen_logps).float().numpy() |
|
all_ref_rejected_logps = torch.cat(ref_rejected_logps).float().numpy() |
|
|
|
eval_dataset = eval_dataset.add_column(name="ref_chosen_logps", column=all_ref_chosen_logps) |
|
eval_dataset = eval_dataset.add_column(name="ref_rejected_logps", column=all_ref_rejected_logps) |
|
|
|
|
|
if self.eval_dataset is not None: |
|
self.eval_dataset = eval_dataset |
|
self._precomputed_eval_ref_log_probs = True |
|
|
|
return super().get_eval_dataloader(eval_dataset=eval_dataset) |
|
|
|
@contextmanager |
|
def null_ref_context(self): |
|
"""Context manager for handling null reference model (that is, peft adapter manipulation).""" |
|
with ( |
|
self.accelerator.unwrap_model(self.model).disable_adapter() |
|
if self.is_peft_model and not self.ref_adapter_name |
|
else nullcontext() |
|
): |
|
if self.ref_adapter_name: |
|
self.model.set_adapter(self.ref_adapter_name) |
|
yield |
|
if self.ref_adapter_name: |
|
self.model.set_adapter(self.model_adapter_name or "default") |
|
|
|
def compute_ref_log_probs(self, batch: dict[str, torch.LongTensor]) -> dict: |
|
"""Computes log probabilities of the reference model for a single padded batch of a DPO specific dataset.""" |
|
device_type = "xpu" if is_torch_xpu_available() else "cuda" |
|
compte_ref_context_manager = amp.autocast(device_type) if self._peft_has_been_casted_to_bf16 else nullcontext() |
|
with torch.no_grad(), compte_ref_context_manager: |
|
if self.ref_model is None: |
|
with self.null_ref_context(): |
|
ref_model_output = self.concatenated_forward(self.model, batch) |
|
else: |
|
ref_model_output = self.concatenated_forward(self.ref_model, batch) |
|
return ref_model_output["chosen_logps"], ref_model_output["rejected_logps"] |
|
|
|
@staticmethod |
|
def concatenated_inputs( |
|
batch: dict[str, Union[list, torch.LongTensor]], padding_value: int |
|
) -> dict[str, torch.LongTensor]: |
|
""" |
|
Concatenate the `chosen` and `rejected` inputs from the batch into a single tensor for both the prompt |
|
and completion sequences. |
|
|
|
Args: |
|
batch (`dict[str, Union[list, torch.LongTensor]]`): |
|
A batch of input data. The batch must contain the following keys: |
|
|
|
- `"prompt_input_ids"`: Tensor of shape `(batch_size, prompt_length)` representing the prompt input IDs. |
|
- `"chosen_input_ids"`: Tensor of shape `(batch_size, chosen_length)` representing the chosen completion input IDs. |
|
- `"rejected_input_ids"`: Tensor of shape `(batch_size, rejected_length)` representing the rejected completion input IDs. |
|
- `"prompt_pixel_values"` (optional): Tensor for pixel values, if available. |
|
- `"prompt_pixel_attention_mask"` (optional): Tensor for pixel attention masks, if available. |
|
|
|
padding_value (`int`): |
|
The padding value to use for the concatenated completion sequences (`chosen_input_ids` and |
|
`rejected_input_ids`). |
|
|
|
Returns: |
|
`dict[str, torch.LongTensor]`: A dictionary containing: |
|
|
|
- `"prompt_input_ids"`: Concatenated prompt input IDs of shape `(2 * batch_size, prompt_length)`. |
|
- `"completion_input_ids"`: Concatenated chosen and rejected completion input IDs of shape `(2 * batch_size, max_completion_length)`. |
|
- `"prompt_attention_mask"`: Concatenated prompt attention masks of shape `(2 * batch_size, prompt_length)`. |
|
- `"completion_attention_mask"`: Concatenated chosen and rejected attention masks of shape `(2 * batch_size, max_completion_length)`. |
|
- `"pixel_values"` (optional): Concatenated pixel values if `"prompt_pixel_values"` are present. |
|
- `"pixel_attention_mask"` (optional): Concatenated pixel attention masks if `"prompt_pixel_attention_mask"` are present. |
|
|
|
Notes: |
|
The completion input IDs and attention masks are padded to the maximum completion length of the chosen |
|
or rejected sequences. |
|
""" |
|
output = {} |
|
|
|
|
|
output["prompt_input_ids"] = torch.cat([batch["prompt_input_ids"], batch["prompt_input_ids"]], dim=0) |
|
output["prompt_attention_mask"] = torch.cat( |
|
[batch["prompt_attention_mask"], batch["prompt_attention_mask"]], dim=0 |
|
) |
|
if "pixel_values" in batch: |
|
output["pixel_values"] = torch.cat([batch["pixel_values"], batch["pixel_values"]], dim=0) |
|
|
|
if "pixel_attention_mask" in batch: |
|
output["pixel_attention_mask"] = torch.cat( |
|
[batch["pixel_attention_mask"], batch["pixel_attention_mask"]], dim=0 |
|
) |
|
if "image_sizes" in batch: |
|
output["image_sizes"] = torch.cat([batch["image_sizes"], batch["image_sizes"]], dim=0) |
|
|
|
|
|
max_completion_length = max(batch["chosen_input_ids"].shape[1], batch["rejected_input_ids"].shape[1]) |
|
output["completion_input_ids"] = torch.cat( |
|
( |
|
pad_to_length(batch["chosen_input_ids"], max_completion_length, pad_value=padding_value), |
|
pad_to_length(batch["rejected_input_ids"], max_completion_length, pad_value=padding_value), |
|
), |
|
) |
|
output["completion_attention_mask"] = torch.cat( |
|
( |
|
pad_to_length(batch["chosen_attention_mask"], max_completion_length, pad_value=0), |
|
pad_to_length(batch["rejected_attention_mask"], max_completion_length, pad_value=0), |
|
), |
|
) |
|
|
|
return output |
|
|
|
def dpo_loss( |
|
self, |
|
chosen_logps: torch.FloatTensor, |
|
rejected_logps: torch.FloatTensor, |
|
ref_chosen_logps: torch.FloatTensor, |
|
ref_rejected_logps: torch.FloatTensor, |
|
) -> tuple[torch.FloatTensor, torch.FloatTensor, torch.FloatTensor]: |
|
""" |
|
Compute the DPO loss for a batch of policy and reference model log probabilities. |
|
|
|
Args: |
|
chosen_logps (`torch.FloatTensor`): |
|
Log probabilities of the model for the chosen responses. Shape: `(batch_size,)`. |
|
rejected_logps (`torch.FloatTensor`): |
|
Log probabilities of the model for the rejected responses. Shape: `(batch_size,)`. |
|
ref_chosen_logps (`torch.FloatTensor`): |
|
Log probabilities of the reference model for the chosen responses. Shape: `(batch_size,)`. |
|
ref_rejected_logps (`torch.FloatTensor`): |
|
Log probabilities of the reference model for the rejected responses. Shape: `(batch_size,)`. |
|
|
|
Returns: |
|
A tuple of three tensors: `(losses, chosen_rewards, rejected_rewards)`. |
|
The losses tensor contains the DPO loss for each example in the batch. |
|
The `chosen_rewards` and `rejected_rewards` tensors contain the rewards for the chosen and rejected |
|
responses, respectively. |
|
""" |
|
device = self.accelerator.device |
|
|
|
|
|
chosen_logratios = chosen_logps.to(device) - (not self.reference_free) * ref_chosen_logps.to(device) |
|
rejected_logratios = rejected_logps.to(device) - (not self.reference_free) * ref_rejected_logps.to(device) |
|
|
|
if self.f_divergence_type == FDivergenceType.ALPHA_DIVERGENCE.value: |
|
|
|
|
|
|
|
|
|
|
|
|
|
alpha_coef = FDivergenceConstants.ALPHA_DIVERGENCE_COEF_DEFAULT |
|
if self.f_divergence_params and FDivergenceConstants.ALPHA_DIVERGENCE_COEF_KEY in self.f_divergence_params: |
|
alpha_coef = float(self.f_divergence_params[FDivergenceConstants.ALPHA_DIVERGENCE_COEF_KEY]) |
|
logits = (cap_exp(rejected_logratios * -alpha_coef) - cap_exp(chosen_logratios * -alpha_coef)) / alpha_coef |
|
else: |
|
logratios = chosen_logps - rejected_logps |
|
if self.reference_free: |
|
ref_logratios = torch.tensor([0], dtype=logratios.dtype, device=logratios.device) |
|
else: |
|
ref_logratios = ref_chosen_logps - ref_rejected_logps |
|
|
|
logratios = logratios.to(self.accelerator.device) |
|
ref_logratios = ref_logratios.to(self.accelerator.device) |
|
logits = logratios - ref_logratios |
|
|
|
if self.f_divergence_type == FDivergenceType.JS_DIVERGENCE.value: |
|
|
|
|
|
|
|
|
|
|
|
|
|
logits -= F.softplus(chosen_logratios) - F.softplus(rejected_logratios) |
|
|
|
|
|
|
|
|
|
if self.loss_type == "sigmoid": |
|
losses = ( |
|
-F.logsigmoid(self.beta * logits) * (1 - self.label_smoothing) |
|
- F.logsigmoid(-self.beta * logits) * self.label_smoothing |
|
) |
|
|
|
elif self.loss_type == "robust": |
|
losses = ( |
|
-F.logsigmoid(self.beta * logits) * (1 - self.label_smoothing) |
|
+ F.logsigmoid(-self.beta * logits) * self.label_smoothing |
|
) / (1 - 2 * self.label_smoothing) |
|
|
|
elif self.loss_type == "exo_pair": |
|
|
|
import math |
|
|
|
if self.label_smoothing == 0: |
|
self.label_smoothing = 1e-3 |
|
losses = (self.beta * logits).sigmoid() * ( |
|
F.logsigmoid(self.beta * logits) - math.log(1 - self.label_smoothing) |
|
) + (-self.beta * logits).sigmoid() * (F.logsigmoid(-self.beta * logits) - math.log(self.label_smoothing)) |
|
|
|
elif self.loss_type == "hinge": |
|
losses = torch.relu(1 - self.beta * logits) |
|
|
|
elif self.loss_type == "ipo": |
|
|
|
losses = (logits - 1 / (2 * self.beta)) ** 2 |
|
|
|
elif self.loss_type == "bco_pair": |
|
chosen_logratios = chosen_logps - ref_chosen_logps |
|
rejected_logratios = rejected_logps - ref_rejected_logps |
|
chosen_rewards = self.beta * chosen_logratios |
|
rejected_rewards = self.beta * rejected_logratios |
|
rewards = torch.cat((chosen_rewards, rejected_rewards), 0).mean().detach() |
|
self.running.update(rewards) |
|
delta = self.running.mean |
|
losses = -F.logsigmoid((self.beta * chosen_logratios) - delta) - F.logsigmoid( |
|
-(self.beta * rejected_logratios - delta) |
|
) |
|
|
|
elif self.loss_type == "sppo_hard": |
|
|
|
|
|
|
|
|
|
a = chosen_logps - ref_chosen_logps |
|
b = rejected_logps - ref_rejected_logps |
|
losses = (a - 0.5 / self.beta) ** 2 + (b + 0.5 / self.beta) ** 2 |
|
|
|
elif self.loss_type == "nca_pair": |
|
chosen_rewards = (chosen_logps - ref_chosen_logps) * self.beta |
|
rejected_rewards = (rejected_logps - ref_rejected_logps) * self.beta |
|
losses = ( |
|
-F.logsigmoid(chosen_rewards) |
|
- 0.5 * F.logsigmoid(-chosen_rewards) |
|
- 0.5 * F.logsigmoid(-rejected_rewards) |
|
) |
|
|
|
elif self.loss_type == "aot_pair": |
|
chosen_logratios = chosen_logps - ref_chosen_logps |
|
rejected_logratios = rejected_logps - ref_rejected_logps |
|
chosen_logratios_sorted, _ = torch.sort(chosen_logratios, dim=0) |
|
rejected_logratios_sorted, _ = torch.sort(rejected_logratios, dim=0) |
|
delta = chosen_logratios_sorted - rejected_logratios_sorted |
|
losses = ( |
|
-F.logsigmoid(self.beta * delta) * (1 - self.label_smoothing) |
|
- F.logsigmoid(-self.beta * delta) * self.label_smoothing |
|
) |
|
|
|
elif self.loss_type == "aot": |
|
logratios = chosen_logps - rejected_logps |
|
ref_logratios = ref_chosen_logps - ref_rejected_logps |
|
logratios_sorted, _ = torch.sort(logratios, dim=0) |
|
ref_logratios_sorted, _ = torch.sort(ref_logratios, dim=0) |
|
delta = logratios_sorted - ref_logratios_sorted |
|
losses = ( |
|
-F.logsigmoid(self.beta * delta) * (1 - self.label_smoothing) |
|
- F.logsigmoid(-self.beta * delta) * self.label_smoothing |
|
) |
|
|
|
elif self.loss_type == "apo_zero": |
|
|
|
|
|
losses_chosen = 1 - F.sigmoid(self.beta * chosen_logratios) |
|
losses_rejected = F.sigmoid(self.beta * rejected_logratios) |
|
losses = losses_chosen + losses_rejected |
|
|
|
elif self.loss_type == "apo_down": |
|
|
|
|
|
|
|
losses_chosen = F.sigmoid(self.beta * chosen_logratios) |
|
losses_rejected = 1 - F.sigmoid(self.beta * (chosen_logratios - rejected_logratios)) |
|
losses = losses_chosen + losses_rejected |
|
|
|
elif self.loss_type == "discopop": |
|
|
|
|
|
logratios = chosen_logps - rejected_logps |
|
ref_logratios = ref_chosen_logps - ref_rejected_logps |
|
logits = logratios - ref_logratios |
|
logits = logits * self.beta |
|
|
|
log_ratio_modulation = torch.sigmoid(logits / self.args.discopop_tau) |
|
logistic_component = -F.logsigmoid(logits) |
|
exp_component = torch.exp(-logits) |
|
|
|
losses = logistic_component * (1 - log_ratio_modulation) + exp_component * log_ratio_modulation |
|
|
|
else: |
|
raise ValueError( |
|
f"Unknown loss type: {self.loss_type}. Should be one of ['sigmoid', 'hinge', 'ipo', 'exo_pair', " |
|
"'nca_pair', 'robust', 'bco_pair', 'sppo_hard', 'aot', 'aot_pair', 'discopop', 'apo_zero', 'apo_down']" |
|
) |
|
|
|
chosen_rewards = self.beta * (chosen_logps.to(device) - ref_chosen_logps.to(device)).detach() |
|
rejected_rewards = self.beta * (rejected_logps.to(device) - ref_rejected_logps.to(device)).detach() |
|
|
|
return losses, chosen_rewards, rejected_rewards |
|
|
|
def concatenated_forward(self, model: nn.Module, batch: dict[str, Union[list, torch.LongTensor]]): |
|
"""Run the given model on the given batch of inputs, concatenating the chosen and rejected inputs together. |
|
|
|
We do this to avoid doing two forward passes, because it's faster for FSDP. |
|
""" |
|
num_examples = batch["prompt_input_ids"].shape[0] |
|
|
|
concatenated_batch = self.concatenated_inputs(batch, padding_value=self.padding_value) |
|
|
|
model_kwargs = {} |
|
if self.aux_loss_enabled: |
|
model_kwargs["output_router_logits"] = True |
|
|
|
|
|
if "pixel_values" in concatenated_batch: |
|
model_kwargs["pixel_values"] = concatenated_batch["pixel_values"] |
|
if "pixel_attention_mask" in concatenated_batch: |
|
model_kwargs["pixel_attention_mask"] = concatenated_batch["pixel_attention_mask"] |
|
if "image_sizes" in concatenated_batch: |
|
model_kwargs["image_sizes"] = concatenated_batch["image_sizes"] |
|
|
|
prompt_input_ids = concatenated_batch["prompt_input_ids"] |
|
prompt_attention_mask = concatenated_batch["prompt_attention_mask"] |
|
completion_input_ids = concatenated_batch["completion_input_ids"] |
|
completion_attention_mask = concatenated_batch["completion_attention_mask"] |
|
if self.is_encoder_decoder: |
|
labels = completion_input_ids |
|
labels[completion_attention_mask == 0] = self.label_pad_token_id |
|
outputs = model( |
|
input_ids=prompt_input_ids, |
|
attention_mask=prompt_attention_mask, |
|
labels=labels, |
|
**model_kwargs, |
|
) |
|
logits = outputs.logits |
|
loss_mask = completion_attention_mask.bool() |
|
else: |
|
|
|
input_ids = torch.cat((prompt_input_ids, completion_input_ids), dim=1) |
|
attention_mask = torch.cat((prompt_attention_mask, completion_attention_mask), dim=1) |
|
|
|
loss_mask = torch.cat( |
|
(torch.zeros_like(prompt_attention_mask), completion_attention_mask), |
|
dim=1, |
|
) |
|
|
|
|
|
|
|
|
|
attention_mask, input_ids, loss_mask = flush_left(attention_mask, input_ids, loss_mask) |
|
|
|
|
|
if self.max_length is not None: |
|
if self.truncation_mode == "keep_end": |
|
input_ids = input_ids[:, -self.max_length :] |
|
attention_mask = attention_mask[:, -self.max_length :] |
|
loss_mask = loss_mask[:, -self.max_length :] |
|
elif self.truncation_mode == "keep_start": |
|
input_ids = input_ids[:, : self.max_length] |
|
attention_mask = attention_mask[:, : self.max_length] |
|
loss_mask = loss_mask[:, : self.max_length] |
|
else: |
|
raise ValueError( |
|
f"Unknown truncation mode: '{self.truncation_mode}'. Should be one of ['keep_end', " |
|
"'keep_start']." |
|
) |
|
|
|
if self.use_logits_to_keep: |
|
|
|
|
|
|
|
|
|
first_compute_index = loss_mask.nonzero(as_tuple=True)[1].min() |
|
logits_to_keep = (loss_mask.shape[1] - first_compute_index).item() + 1 |
|
model_kwargs["logits_to_keep"] = logits_to_keep |
|
|
|
if self.padding_free: |
|
|
|
|
|
|
|
input_ids = input_ids[attention_mask.bool()].unsqueeze(0) |
|
loss_mask = loss_mask[attention_mask.bool()].unsqueeze(0) |
|
position_ids = attention_mask.cumsum(1)[attention_mask.bool()].unsqueeze(0) - 1 |
|
model_kwargs["position_ids"] = position_ids |
|
else: |
|
model_kwargs["attention_mask"] = attention_mask |
|
|
|
outputs = model(input_ids, **model_kwargs) |
|
logits = outputs.logits |
|
|
|
|
|
labels = torch.roll(input_ids, shifts=-1, dims=1) |
|
loss_mask = torch.roll(loss_mask, shifts=-1, dims=1).bool() |
|
|
|
if self.use_logits_to_keep: |
|
|
|
|
|
|
|
|
|
|
|
|
|
labels = labels[:, -logits_to_keep:] |
|
loss_mask = loss_mask[:, -logits_to_keep:] |
|
|
|
if logits.shape[:2] != labels.shape[:2]: |
|
|
|
seq_len = labels.shape[1] |
|
logits = logits[:, -seq_len:] |
|
|
|
|
|
labels[~loss_mask] = 0 |
|
per_token_logps = selective_log_softmax(logits, labels) |
|
per_token_logps[~loss_mask] = 0 |
|
per_token_logps = torch.roll(per_token_logps, shifts=1, dims=1) |
|
|
|
if self.padding_free: |
|
|
|
batch_size, seq_len = attention_mask.shape |
|
per_token_logps_ = torch.zeros( |
|
batch_size, seq_len, device=outputs.logits.device, dtype=outputs.logits.dtype |
|
) |
|
per_token_logps_[attention_mask.bool()] = per_token_logps |
|
per_token_logps = per_token_logps_ |
|
|
|
all_logps = per_token_logps.sum(-1) |
|
|
|
output = {} |
|
|
|
if self.use_weighting: |
|
with torch.no_grad(): |
|
|
|
logprobs = F.log_softmax(logits, dim=-1) |
|
weights_adjustment_factor = torch.logsumexp(2 * logprobs, dim=-1) |
|
per_token_logps_adjusted = per_token_logps - weights_adjustment_factor |
|
all_weights = (per_token_logps_adjusted * loss_mask).sum(-1) / loss_mask.sum(-1) |
|
chosen_weights = all_weights[:num_examples] |
|
rejected_weights = all_weights[num_examples:] |
|
output["policy_weights"] = torch.clamp(torch.exp(chosen_weights + rejected_weights), max=1) |
|
|
|
if self.args.rpo_alpha is not None: |
|
|
|
chosen_logits = logits[:num_examples] |
|
chosen_labels = labels[:num_examples] |
|
|
|
|
|
output["nll_loss"] = F.cross_entropy( |
|
torch.flatten(chosen_logits, end_dim=1), torch.flatten(chosen_labels, end_dim=1), ignore_index=0 |
|
) |
|
|
|
if self.loss_type == "ipo": |
|
all_logps = all_logps / loss_mask.sum(-1) |
|
|
|
output["chosen_logps"] = all_logps[:num_examples] |
|
output["rejected_logps"] = all_logps[num_examples:] |
|
|
|
|
|
if self.padding_free: |
|
|
|
|
|
|
|
|
|
split_idx = (position_ids == 0).nonzero(as_tuple=True)[1][num_examples] |
|
mean_chosen_logits = logits[0, :split_idx][loss_mask[0, :split_idx]].mean() |
|
mean_rejected_logits = logits[0, split_idx:][loss_mask[0, split_idx:]].mean() |
|
else: |
|
mean_chosen_logits = logits[:num_examples][loss_mask[:num_examples]].mean() |
|
mean_rejected_logits = logits[num_examples:][loss_mask[num_examples:]].mean() |
|
|
|
output["mean_chosen_logits"] = mean_chosen_logits |
|
output["mean_rejected_logits"] = mean_rejected_logits |
|
|
|
if self.aux_loss_enabled: |
|
output["aux_loss"] = outputs.aux_loss |
|
|
|
return output |
|
|
|
def get_batch_loss_metrics( |
|
self, |
|
model, |
|
batch: dict[str, Union[list, torch.LongTensor]], |
|
train_eval: Literal["train", "eval"] = "train", |
|
): |
|
"""Compute the DPO loss and other metrics for the given batch of inputs for train or test.""" |
|
metrics = {} |
|
|
|
model_output = self.concatenated_forward(model, batch) |
|
|
|
|
|
if "ref_chosen_logps" in batch and "ref_rejected_logps" in batch: |
|
ref_chosen_logps = batch["ref_chosen_logps"] |
|
ref_rejected_logps = batch["ref_rejected_logps"] |
|
else: |
|
ref_chosen_logps, ref_rejected_logps = self.compute_ref_log_probs(batch) |
|
|
|
losses, chosen_rewards, rejected_rewards = self.dpo_loss( |
|
model_output["chosen_logps"], model_output["rejected_logps"], ref_chosen_logps, ref_rejected_logps |
|
) |
|
reward_accuracies = (chosen_rewards > rejected_rewards).float() |
|
|
|
if self.args.rpo_alpha is not None: |
|
losses = losses + self.args.rpo_alpha * model_output["nll_loss"] |
|
|
|
if self.use_weighting: |
|
losses = losses * model_output["policy_weights"] |
|
|
|
if self.aux_loss_enabled: |
|
losses = losses + self.aux_loss_coef * model_output["aux_loss"] |
|
|
|
prefix = "eval_" if train_eval == "eval" else "" |
|
metrics[f"{prefix}rewards/chosen"] = self.accelerator.gather_for_metrics(chosen_rewards).mean().item() |
|
metrics[f"{prefix}rewards/rejected"] = self.accelerator.gather_for_metrics(rejected_rewards).mean().item() |
|
metrics[f"{prefix}rewards/accuracies"] = self.accelerator.gather_for_metrics(reward_accuracies).mean().item() |
|
metrics[f"{prefix}rewards/margins"] = ( |
|
self.accelerator.gather_for_metrics(chosen_rewards - rejected_rewards).mean().item() |
|
) |
|
metrics[f"{prefix}logps/chosen"] = ( |
|
self.accelerator.gather_for_metrics(model_output["chosen_logps"]).detach().mean().item() |
|
) |
|
metrics[f"{prefix}logps/rejected"] = ( |
|
self.accelerator.gather_for_metrics(model_output["rejected_logps"]).detach().mean().item() |
|
) |
|
metrics[f"{prefix}logits/chosen"] = ( |
|
self.accelerator.gather_for_metrics(model_output["mean_chosen_logits"]).detach().mean().item() |
|
) |
|
metrics[f"{prefix}logits/rejected"] = ( |
|
self.accelerator.gather_for_metrics(model_output["mean_rejected_logits"]).detach().mean().item() |
|
) |
|
if self.args.rpo_alpha is not None: |
|
metrics[f"{prefix}nll_loss"] = ( |
|
self.accelerator.gather_for_metrics(model_output["nll_loss"]).detach().mean().item() |
|
) |
|
if self.aux_loss_enabled: |
|
metrics[f"{prefix}aux_loss"] = ( |
|
self.accelerator.gather_for_metrics(model_output["aux_loss"]).detach().mean().item() |
|
) |
|
|
|
return losses.mean(), metrics |
|
|
|
def compute_loss( |
|
self, |
|
model: Union[PreTrainedModel, nn.Module], |
|
inputs: dict[str, Union[torch.Tensor, Any]], |
|
return_outputs=False, |
|
num_items_in_batch=None, |
|
) -> Union[torch.Tensor, tuple[torch.Tensor, dict[str, torch.Tensor]]]: |
|
device_type = "xpu" if is_torch_xpu_available() else "cuda" |
|
compute_loss_context_manager = ( |
|
amp.autocast(device_type) if self._peft_has_been_casted_to_bf16 else nullcontext() |
|
) |
|
with compute_loss_context_manager: |
|
loss, metrics = self.get_batch_loss_metrics(model, inputs, train_eval="train") |
|
|
|
|
|
loss = loss.to(self.args.device) |
|
|
|
self.store_metrics(metrics, train_eval="train") |
|
|
|
if return_outputs: |
|
return loss, metrics |
|
|
|
return loss |
|
|
|
def generate_from_model_and_ref(self, model, batch: dict[str, torch.LongTensor]) -> tuple[str, str]: |
|
"""Generate samples from the model and reference model for the given batch of inputs.""" |
|
|
|
|
|
|
|
device_type = "xpu" if is_torch_xpu_available() else "cuda" |
|
generate_context_manager = amp.autocast(device_type) if self._peft_has_been_casted_to_bf16 else nullcontext() |
|
|
|
with generate_context_manager: |
|
policy_output = model.generate( |
|
input_ids=batch["prompt_input_ids"], |
|
attention_mask=batch["prompt_attention_mask"], |
|
max_length=self.max_length, |
|
do_sample=True, |
|
pad_token_id=self.padding_value, |
|
) |
|
|
|
|
|
if "ref_output" in batch: |
|
ref_output = batch["ref_output"] |
|
else: |
|
if self.ref_model is None: |
|
with self.null_ref_context(): |
|
ref_output = self.model.generate( |
|
input_ids=batch["prompt_input_ids"], |
|
attention_mask=batch["prompt_attention_mask"], |
|
max_length=self.max_length, |
|
do_sample=True, |
|
pad_token_id=self.padding_value, |
|
) |
|
else: |
|
ref_output = self.ref_model.generate( |
|
input_ids=batch["prompt_input_ids"], |
|
attention_mask=batch["prompt_attention_mask"], |
|
max_length=self.max_length, |
|
do_sample=True, |
|
pad_token_id=self.padding_value, |
|
) |
|
|
|
policy_output = pad_to_length(policy_output, self.max_length, self.padding_value) |
|
policy_output_decoded = self.processing_class.batch_decode(policy_output, skip_special_tokens=True) |
|
|
|
ref_output = pad_to_length(ref_output, self.max_length, self.padding_value) |
|
ref_output_decoded = self.processing_class.batch_decode(ref_output, skip_special_tokens=True) |
|
|
|
return policy_output_decoded, ref_output_decoded |
|
|
|
def prediction_step( |
|
self, |
|
model: Union[PreTrainedModel, nn.Module], |
|
inputs: dict[str, Union[torch.Tensor, Any]], |
|
prediction_loss_only: bool, |
|
ignore_keys: Optional[list[str]] = None, |
|
): |
|
if ignore_keys is None: |
|
if hasattr(model, "config"): |
|
ignore_keys = getattr(model.config, "keys_to_ignore_at_inference", []) |
|
else: |
|
ignore_keys = [] |
|
|
|
device_type = "xpu" if is_torch_xpu_available() else "cuda" |
|
prediction_context_manager = amp.autocast(device_type) if self._peft_has_been_casted_to_bf16 else nullcontext() |
|
|
|
with torch.no_grad(), prediction_context_manager: |
|
loss, metrics = self.get_batch_loss_metrics(model, inputs, train_eval="eval") |
|
|
|
|
|
self.store_metrics(metrics, train_eval="eval") |
|
|
|
if prediction_loss_only: |
|
return loss.detach(), None, None |
|
|
|
|
|
logits_dict = { |
|
"eval_logits/chosen": metrics["eval_logits/chosen"], |
|
"eval_logits/rejected": metrics["eval_logits/rejected"], |
|
} |
|
logits = [v for k, v in logits_dict.items() if k not in ignore_keys] |
|
logits = torch.tensor(logits, device=self.accelerator.device) |
|
labels = torch.zeros(logits.shape[0], device=self.accelerator.device) |
|
|
|
return (loss.detach(), logits, labels) |
|
|
|
def store_metrics(self, metrics: dict[str, float], train_eval: Literal["train", "eval"] = "train") -> None: |
|
for key, value in metrics.items(): |
|
self._stored_metrics[train_eval][key].append(value) |
|
|
|
def evaluation_loop( |
|
self, |
|
dataloader: DataLoader, |
|
description: str, |
|
prediction_loss_only: Optional[bool] = None, |
|
ignore_keys: Optional[list[str]] = None, |
|
metric_key_prefix: str = "eval", |
|
) -> EvalLoopOutput: |
|
""" |
|
Overriding built-in evaluation loop to store metrics for each batch. |
|
Prediction/evaluation loop, shared by `Trainer.evaluate()` and `Trainer.predict()`. |
|
|
|
Works both with or without labels. |
|
""" |
|
|
|
|
|
if self.generate_during_eval: |
|
|
|
num_samples = len(dataloader.dataset) |
|
random_indices = random.sample(range(num_samples), k=self.args.eval_batch_size) |
|
|
|
|
|
random_batch_dataset = dataloader.dataset.select(random_indices) |
|
random_batch = self.data_collator(random_batch_dataset) |
|
random_batch = self._prepare_inputs(random_batch) |
|
|
|
policy_output_decoded, ref_output_decoded = self.generate_from_model_and_ref(self.model, random_batch) |
|
|
|
table = pd.DataFrame( |
|
columns=["Prompt", "Policy", "Ref Model"], |
|
data=[ |
|
[prompt, pol[len(prompt) :], ref[len(prompt) :]] |
|
for prompt, pol, ref in zip( |
|
random_batch_dataset["prompt"], policy_output_decoded, ref_output_decoded |
|
) |
|
], |
|
) |
|
if "wandb" in self.args.report_to: |
|
wandb.log({"game_log": wandb.Table(data=table)}) |
|
|
|
if "comet_ml" in self.args.report_to: |
|
log_table_to_comet_experiment( |
|
name="game_log.csv", |
|
table=table, |
|
) |
|
|
|
|
|
initial_output = super().evaluation_loop( |
|
dataloader, description, prediction_loss_only, ignore_keys, metric_key_prefix |
|
) |
|
|
|
return initial_output |
|
|
|
def log(self, logs: dict[str, float], start_time: Optional[float] = None) -> None: |
|
""" |
|
Log `logs` on the various objects watching training, including stored metrics. |
|
|
|
Args: |
|
logs (`dict[str, float]`): |
|
The values to log. |
|
start_time (`float` or `None`, *optional*, defaults to `None`): |
|
Start time of the training. |
|
""" |
|
|
|
train_eval = "train" if "loss" in logs else "eval" |
|
|
|
for key, metrics in self._stored_metrics[train_eval].items(): |
|
logs[key] = torch.tensor(metrics).mean().item() |
|
del self._stored_metrics[train_eval] |
|
|
|
if version.parse(transformers.__version__) >= version.parse("4.47.0.dev0"): |
|
return super().log(logs, start_time) |
|
else: |
|
return super().log(logs) |
|
|
|
def create_model_card( |
|
self, |
|
model_name: Optional[str] = None, |
|
dataset_name: Optional[str] = None, |
|
tags: Union[str, list[str], None] = None, |
|
): |
|
""" |
|
Creates a draft of a model card using the information available to the `Trainer`. |
|
|
|
Args: |
|
model_name (`str` or `None`, *optional*, defaults to `None`): |
|
Name of the model. |
|
dataset_name (`str` or `None`, *optional*, defaults to `None`): |
|
Name of the dataset used for training. |
|
tags (`str`, `list[str]` or `None`, *optional*, defaults to `None`): |
|
Tags to be associated with the model card. |
|
""" |
|
if not self.is_world_process_zero(): |
|
return |
|
|
|
if hasattr(self.model.config, "_name_or_path") and not os.path.isdir(self.model.config._name_or_path): |
|
base_model = self.model.config._name_or_path |
|
else: |
|
base_model = None |
|
|
|
tags = tags or [] |
|
if isinstance(tags, str): |
|
tags = [tags] |
|
|
|
if hasattr(self.model.config, "unsloth_version"): |
|
tags.append("unsloth") |
|
|
|
citation = textwrap.dedent( |
|
"""\ |
|
@inproceedings{rafailov2023direct, |
|
title = {{Direct Preference Optimization: Your Language Model is Secretly a Reward Model}}, |
|
author = {Rafael Rafailov and Archit Sharma and Eric Mitchell and Christopher D. Manning and Stefano Ermon and Chelsea Finn}, |
|
year = 2023, |
|
booktitle = {Advances in Neural Information Processing Systems 36: Annual Conference on Neural Information Processing Systems 2023, NeurIPS 2023, New Orleans, LA, USA, December 10 - 16, 2023}, |
|
url = {http://papers.nips.cc/paper_files/paper/2023/hash/a85b405ed65c6477a4fe8302b5e06ce7-Abstract-Conference.html}, |
|
editor = {Alice Oh and Tristan Naumann and Amir Globerson and Kate Saenko and Moritz Hardt and Sergey Levine}, |
|
}""" |
|
) |
|
|
|
model_card = generate_model_card( |
|
base_model=base_model, |
|
model_name=model_name, |
|
hub_model_id=self.hub_model_id, |
|
dataset_name=dataset_name, |
|
tags=tags, |
|
wandb_url=wandb.run.get_url() if is_wandb_available() and wandb.run is not None else None, |
|
comet_url=get_comet_experiment_url(), |
|
trainer_name="DPO", |
|
trainer_citation=citation, |
|
paper_title="Direct Preference Optimization: Your Language Model is Secretly a Reward Model", |
|
paper_id="2305.18290", |
|
) |
|
|
|
model_card.save(os.path.join(self.args.output_dir, "README.md")) |
|
|