tests/test_trainers_args.py

# Copyright 2020-2025 The HuggingFace Team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import tempfile
import unittest

from datasets import load_dataset
from parameterized import parameterized
from transformers import AutoModelForCausalLM, AutoModelForSequenceClassification, AutoTokenizer

from trl import (
    BCOConfig,
    BCOTrainer,
    CPOConfig,
    CPOTrainer,
    DPOConfig,
    DPOTrainer,
    KTOConfig,
    KTOTrainer,
    NashMDConfig,
    NashMDTrainer,
    OnlineDPOConfig,
    OnlineDPOTrainer,
    ORPOConfig,
    ORPOTrainer,
    RewardConfig,
    RewardTrainer,
    SFTConfig,
    SFTTrainer,
    XPOConfig,
    XPOTrainer,
)

from .testing_utils import require_sklearn


class TrainerArgTester(unittest.TestCase):
    @require_sklearn
    def test_bco(self):
        model_id = "trl-internal-testing/tiny-Qwen2ForCausalLM-2.5"
        tokenizer = AutoTokenizer.from_pretrained(model_id)
        dataset = load_dataset("trl-internal-testing/zen", "standard_unpaired_preference", split="train")
        with tempfile.TemporaryDirectory() as tmp_dir:
            training_args = BCOConfig(
                tmp_dir,
                max_length=256,
                max_prompt_length=64,
                max_completion_length=64,
                beta=0.5,
                label_pad_token_id=-99,
                padding_value=-99,
                truncation_mode="keep_start",
                # generate_during_eval=True, # ignore this one, it requires wandb
                is_encoder_decoder=True,
                precompute_ref_log_probs=True,
                model_init_kwargs={"trust_remote_code": True},
                ref_model_init_kwargs={"trust_remote_code": True},
                dataset_num_proc=4,
                prompt_sample_size=512,
                min_density_ratio=0.2,
                max_density_ratio=20.0,
            )
            trainer = BCOTrainer(
                model=model_id,
                ref_model=model_id,
                args=training_args,
                train_dataset=dataset,
                processing_class=tokenizer,
            )
            self.assertEqual(trainer.args.max_length, 256)
            self.assertEqual(trainer.args.max_prompt_length, 64)
            self.assertEqual(trainer.args.max_completion_length, 64)
            self.assertEqual(trainer.args.beta, 0.5)
            self.assertEqual(trainer.args.label_pad_token_id, -99)
            self.assertEqual(trainer.args.padding_value, -99)
            self.assertEqual(trainer.args.truncation_mode, "keep_start")
            # self.assertEqual(trainer.args.generate_during_eval, True)
            self.assertEqual(trainer.args.is_encoder_decoder, True)
            self.assertEqual(trainer.args.precompute_ref_log_probs, True)
            self.assertEqual(trainer.args.model_init_kwargs, {"trust_remote_code": True})
            self.assertEqual(trainer.args.ref_model_init_kwargs, {"trust_remote_code": True})
            self.assertEqual(trainer.args.dataset_num_proc, 4)
            self.assertEqual(trainer.args.prompt_sample_size, 512)
            self.assertEqual(trainer.args.min_density_ratio, 0.2)
            self.assertEqual(trainer.args.max_density_ratio, 20.0)

    def test_cpo(self):
        model_id = "trl-internal-testing/tiny-Qwen2ForCausalLM-2.5"
        tokenizer = AutoTokenizer.from_pretrained(model_id)
        dataset = load_dataset("trl-internal-testing/zen", "standard_preference", split="train")
        with tempfile.TemporaryDirectory() as tmp_dir:
            training_args = CPOConfig(
                tmp_dir,
                max_length=256,
                max_prompt_length=64,
                max_completion_length=64,
                beta=0.5,
                label_smoothing=0.5,
                loss_type="hinge",
                disable_dropout=False,
                cpo_alpha=0.5,
                simpo_gamma=0.2,
                label_pad_token_id=-99,
                padding_value=-99,
                truncation_mode="keep_start",
                # generate_during_eval=True, # ignore this one, it requires wandb
                is_encoder_decoder=True,
                model_init_kwargs={"trust_remote_code": True},
                dataset_num_proc=4,
            )
            trainer = CPOTrainer(model=model_id, args=training_args, train_dataset=dataset, processing_class=tokenizer)
            self.assertEqual(trainer.args.max_length, 256)
            self.assertEqual(trainer.args.max_prompt_length, 64)
            self.assertEqual(trainer.args.max_completion_length, 64)
            self.assertEqual(trainer.args.beta, 0.5)
            self.assertEqual(trainer.args.label_smoothing, 0.5)
            self.assertEqual(trainer.args.loss_type, "hinge")
            self.assertEqual(trainer.args.disable_dropout, False)
            self.assertEqual(trainer.args.cpo_alpha, 0.5)
            self.assertEqual(trainer.args.simpo_gamma, 0.2)
            self.assertEqual(trainer.args.label_pad_token_id, -99)
            self.assertEqual(trainer.args.padding_value, -99)
            self.assertEqual(trainer.args.truncation_mode, "keep_start")
            # self.assertEqual(trainer.args.generate_during_eval, True)
            self.assertEqual(trainer.args.is_encoder_decoder, True)
            self.assertEqual(trainer.args.model_init_kwargs, {"trust_remote_code": True})
            self.assertEqual(trainer.args.dataset_num_proc, 4)

    def test_dpo(self):
        model_id = "trl-internal-testing/tiny-Qwen2ForCausalLM-2.5"
        tokenizer = AutoTokenizer.from_pretrained(model_id)
        dataset = load_dataset("trl-internal-testing/zen", "standard_preference", split="train")
        with tempfile.TemporaryDirectory() as tmp_dir:
            training_args = DPOConfig(
                tmp_dir,
                beta=0.5,
                label_smoothing=0.5,
                loss_type="hinge",
                label_pad_token_id=-99,
                padding_value=-99,
                truncation_mode="keep_start",
                max_length=256,
                max_prompt_length=64,
                max_completion_length=64,
                disable_dropout=False,
                # generate_during_eval=True, # ignore this one, it requires wandb
                precompute_ref_log_probs=True,
                dataset_num_proc=4,
                model_init_kwargs={"trust_remote_code": True},
                ref_model_init_kwargs={"trust_remote_code": True},
                model_adapter_name="dummy_adapter",
                ref_adapter_name="dummy_adapter",
                reference_free=True,
                force_use_ref_model=True,
                f_divergence_type="js_divergence",
                f_alpha_divergence_coef=0.5,
                # sync_ref_model=True, # cannot be True when precompute_ref_log_probs=True. Don't test this.
                ref_model_mixup_alpha=0.5,
                ref_model_sync_steps=32,
                rpo_alpha=0.5,
                discopop_tau=0.1,
            )
            trainer = DPOTrainer(
                model=model_id,
                ref_model=model_id,
                args=training_args,
                train_dataset=dataset,
                processing_class=tokenizer,
            )
            self.assertEqual(trainer.args.beta, 0.5)
            self.assertEqual(trainer.args.label_smoothing, 0.5)
            self.assertEqual(trainer.args.loss_type, "hinge")
            self.assertEqual(trainer.args.label_pad_token_id, -99)
            self.assertEqual(trainer.args.padding_value, -99)
            self.assertEqual(trainer.args.truncation_mode, "keep_start")
            self.assertEqual(trainer.args.max_length, 256)
            self.assertEqual(trainer.args.max_prompt_length, 64)
            self.assertEqual(trainer.args.max_completion_length, 64)
            self.assertEqual(trainer.args.disable_dropout, False)
            # self.assertEqual(trainer.args.generate_during_eval, True)
            self.assertEqual(trainer.args.precompute_ref_log_probs, True)
            self.assertEqual(trainer.args.dataset_num_proc, 4)
            self.assertEqual(trainer.args.model_init_kwargs, {"trust_remote_code": True})
            self.assertEqual(trainer.args.ref_model_init_kwargs, {"trust_remote_code": True})
            self.assertEqual(trainer.args.model_adapter_name, "dummy_adapter")
            self.assertEqual(trainer.args.ref_adapter_name, "dummy_adapter")
            self.assertEqual(trainer.args.reference_free, True)
            self.assertEqual(trainer.args.force_use_ref_model, True)
            self.assertEqual(trainer.args.f_divergence_type, "js_divergence")
            self.assertEqual(trainer.args.f_alpha_divergence_coef, 0.5)
            # self.assertEqual(trainer.args.sync_ref_model, True)
            self.assertEqual(trainer.args.ref_model_mixup_alpha, 0.5)
            self.assertEqual(trainer.args.ref_model_sync_steps, 32)
            self.assertEqual(trainer.args.rpo_alpha, 0.5)
            self.assertEqual(trainer.args.discopop_tau, 0.1)

    def test_kto(self):
        model_id = "trl-internal-testing/tiny-Qwen2ForCausalLM-2.5"
        tokenizer = AutoTokenizer.from_pretrained(model_id)
        dataset = load_dataset("trl-internal-testing/zen", "standard_unpaired_preference", split="train")
        with tempfile.TemporaryDirectory() as tmp_dir:
            training_args = KTOConfig(
                tmp_dir,
                max_length=256,
                max_prompt_length=64,
                max_completion_length=64,
                beta=0.5,
                desirable_weight=0.5,
                undesirable_weight=0.5,
                label_pad_token_id=-99,
                padding_value=-99,
                truncation_mode="keep_start",
                # generate_during_eval=True, # ignore this one, it requires wandb
                is_encoder_decoder=True,
                precompute_ref_log_probs=True,
                model_init_kwargs={"trust_remote_code": True},
                ref_model_init_kwargs={"trust_remote_code": True},
                dataset_num_proc=4,
            )
            trainer = KTOTrainer(
                model=model_id,
                ref_model=model_id,
                args=training_args,
                train_dataset=dataset,
                processing_class=tokenizer,
            )
            self.assertEqual(trainer.args.max_length, 256)
            self.assertEqual(trainer.args.max_prompt_length, 64)
            self.assertEqual(trainer.args.max_completion_length, 64)
            self.assertEqual(trainer.args.beta, 0.5)
            self.assertEqual(trainer.args.desirable_weight, 0.5)
            self.assertEqual(trainer.args.undesirable_weight, 0.5)
            self.assertEqual(trainer.args.label_pad_token_id, -99)
            self.assertEqual(trainer.args.padding_value, -99)
            self.assertEqual(trainer.args.truncation_mode, "keep_start")
            # self.assertEqual(trainer.args.generate_during_eval, True)
            self.assertEqual(trainer.args.is_encoder_decoder, True)
            self.assertEqual(trainer.args.precompute_ref_log_probs, True)
            self.assertEqual(trainer.args.model_init_kwargs, {"trust_remote_code": True})
            self.assertEqual(trainer.args.ref_model_init_kwargs, {"trust_remote_code": True})
            self.assertEqual(trainer.args.dataset_num_proc, 4)

    @parameterized.expand([(False,), (True,)])
    def test_nash_md(self, mixtures_coef_list):
        model_id = "trl-internal-testing/tiny-Qwen2ForCausalLM-2.5"
        tokenizer = AutoTokenizer.from_pretrained(model_id)
        model = AutoModelForCausalLM.from_pretrained(model_id)
        ref_model = AutoModelForCausalLM.from_pretrained(model_id)
        reward_model = AutoModelForSequenceClassification.from_pretrained(model_id, num_labels=1)
        dataset = load_dataset("trl-internal-testing/zen", "standard_prompt_only", split="train")
        with tempfile.TemporaryDirectory() as tmp_dir:
            training_args = NashMDConfig(
                tmp_dir,
                mixture_coef=0.5 if not mixtures_coef_list else [0.5, 0.6],
            )
            trainer = NashMDTrainer(
                args=training_args,
                processing_class=tokenizer,
                model=model,
                ref_model=ref_model,
                reward_model=reward_model,
                train_dataset=dataset,
            )
            self.assertEqual(trainer.args.mixture_coef, 0.5 if not mixtures_coef_list else [0.5, 0.6])

    @parameterized.expand([(False,), (True,)])
    def test_online_dpo(self, beta_list):
        model_id = "trl-internal-testing/tiny-Qwen2ForCausalLM-2.5"
        tokenizer = AutoTokenizer.from_pretrained(model_id)
        model = AutoModelForCausalLM.from_pretrained(model_id)
        ref_model = AutoModelForCausalLM.from_pretrained(model_id)
        reward_model = AutoModelForSequenceClassification.from_pretrained(model_id, num_labels=1)
        dataset = load_dataset("trl-internal-testing/zen", "standard_prompt_only", split="train")
        with tempfile.TemporaryDirectory() as tmp_dir:
            training_args = OnlineDPOConfig(
                tmp_dir,
                max_new_tokens=42,
                temperature=0.5,
                missing_eos_penalty=0.33,
                beta=0.6 if not beta_list else [0.6, 0.7],
                loss_type="hinge",
                dataset_num_proc=4,
            )
            trainer = OnlineDPOTrainer(
                model=model,
                ref_model=ref_model,
                reward_model=reward_model,
                args=training_args,
                train_dataset=dataset,
                processing_class=tokenizer,
                reward_processing_class=tokenizer,
            )
            self.assertEqual(trainer.args.max_new_tokens, 42)
            self.assertEqual(trainer.args.temperature, 0.5)
            self.assertEqual(trainer.args.missing_eos_penalty, 0.33)
            self.assertEqual(trainer.args.beta, 0.6 if not beta_list else [0.6, 0.7])
            self.assertEqual(trainer.args.loss_type, "hinge")
            self.assertEqual(trainer.args.dataset_num_proc, 4)

    def test_orpo(self):
        model_id = "trl-internal-testing/tiny-Qwen2ForCausalLM-2.5"
        tokenizer = AutoTokenizer.from_pretrained(model_id)
        dataset = load_dataset("trl-internal-testing/zen", "standard_preference", split="train")
        with tempfile.TemporaryDirectory() as tmp_dir:
            training_args = ORPOConfig(
                tmp_dir,
                max_length=256,
                max_prompt_length=64,
                max_completion_length=64,
                beta=0.5,
                disable_dropout=False,
                label_pad_token_id=-99,
                padding_value=-99,
                truncation_mode="keep_start",
                # generate_during_eval=True, # ignore this one, it requires wandb
                is_encoder_decoder=True,
                model_init_kwargs={"trust_remote_code": True},
                dataset_num_proc=4,
            )
            trainer = ORPOTrainer(
                model=model_id, args=training_args, train_dataset=dataset, processing_class=tokenizer
            )
            self.assertEqual(trainer.args.max_length, 256)
            self.assertEqual(trainer.args.max_prompt_length, 64)
            self.assertEqual(trainer.args.max_completion_length, 64)
            self.assertEqual(trainer.args.beta, 0.5)
            self.assertEqual(trainer.args.disable_dropout, False)
            self.assertEqual(trainer.args.label_pad_token_id, -99)

    def test_reward(self):
        model_id = "trl-internal-testing/tiny-Qwen2ForCausalLM-2.5"
        tokenizer = AutoTokenizer.from_pretrained(model_id)
        model = AutoModelForCausalLM.from_pretrained(model_id)
        dataset = load_dataset("trl-internal-testing/zen", "standard_preference", split="train")
        with tempfile.TemporaryDirectory() as tmp_dir:
            training_args = RewardConfig(
                tmp_dir,
                max_length=256,
                dataset_num_proc=4,
                center_rewards_coefficient=0.1,
            )
            trainer = RewardTrainer(
                model=model,
                args=training_args,
                train_dataset=dataset,
                processing_class=tokenizer,
            )
            self.assertEqual(trainer.args.max_length, 256)
            self.assertEqual(trainer.args.dataset_num_proc, 4)
            self.assertEqual(trainer.args.center_rewards_coefficient, 0.1)

    def test_sft(self):
        model_id = "trl-internal-testing/tiny-Qwen2ForCausalLM-2.5"
        dataset = load_dataset("trl-internal-testing/zen", "standard_language_modeling", split="train")
        with tempfile.TemporaryDirectory() as tmp_dir:
            training_args = SFTConfig(
                tmp_dir,
                dataset_text_field="dummy_text_field",
                packing=True,
                max_length=256,
                dataset_num_proc=4,
                neftune_noise_alpha=0.1,
                model_init_kwargs={"trust_remote_code": True},
                dataset_kwargs={"append_concat_token": True, "skip_prepare_dataset": True},
                eval_packing=True,
            )
            trainer = SFTTrainer(model_id, args=training_args, train_dataset=dataset)
            self.assertEqual(trainer.args.dataset_text_field, "dummy_text_field")
            self.assertEqual(trainer.args.packing, True)
            self.assertEqual(trainer.args.max_length, 256)
            self.assertEqual(trainer.args.dataset_num_proc, 4)
            self.assertEqual(trainer.args.neftune_noise_alpha, 0.1)
            self.assertEqual(trainer.args.model_init_kwargs, {"trust_remote_code": True})
            self.assertIn("append_concat_token", trainer.args.dataset_kwargs)
            self.assertEqual(trainer.args.dataset_kwargs["append_concat_token"], True)
            self.assertEqual(trainer.args.eval_packing, True)

    @parameterized.expand([(False,), (True,)])
    def test_xpo(self, alpha_list):
        model_id = "trl-internal-testing/tiny-Qwen2ForCausalLM-2.5"
        tokenizer = AutoTokenizer.from_pretrained(model_id)
        model = AutoModelForCausalLM.from_pretrained(model_id)
        ref_model = AutoModelForCausalLM.from_pretrained(model_id)
        reward_model = AutoModelForSequenceClassification.from_pretrained(model_id, num_labels=1)
        dataset = load_dataset("trl-internal-testing/zen", "standard_prompt_only", split="train")
        with tempfile.TemporaryDirectory() as tmp_dir:
            training_args = XPOConfig(
                tmp_dir,
                alpha=0.5 if not alpha_list else [0.5, 0.6],
            )
            trainer = XPOTrainer(
                args=training_args,
                processing_class=tokenizer,
                model=model,
                ref_model=ref_model,
                reward_model=reward_model,
                train_dataset=dataset,
            )
            self.assertEqual(trainer.args.alpha, 0.5 if not alpha_list else [0.5, 0.6])