Spaces:
Paused
Paused
File size: 5,166 Bytes
2f5127c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 |
# 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 argparse
import importlib
import os
import sys
from dataclasses import dataclass, field
from typing import Optional
from datasets import load_dataset
from transformers import AutoModelForCausalLM, AutoModelForSequenceClassification, AutoTokenizer
from trl import GRPOConfig, GRPOTrainer, ModelConfig, ScriptArguments, TrlParser, get_peft_config
from trl.rewards import think_format_reward
reward_funcs_registry = {
"think_format_reward": think_format_reward,
}
@dataclass
class GRPOScriptArguments(ScriptArguments):
"""
Script arguments for the GRPO training script.
Args:
reward_model_name_or_path (`str` or `None`, *optional*, defaults to `None`):
Reward model id of a pretrained model hosted inside a model repo on huggingface.co or local path to a
directory containing model weights saved using [`~transformers.PreTrainedModel.save_pretrained`].
reward_funcs (`list[str]` or `None`, *optional*, defaults to `None`):
Reward functions to use. It can be either one of `"think_format_reward"`; or a dotted import path "
(e.g., `'my_lib.rewards.custom_reward'`).
"""
reward_model_name_or_path: Optional[str] = field(
default=None,
metadata={
"help": "Reward model id of a pretrained model hosted inside a model repo on huggingface.co or "
"local path to a directory containing model weights saved using `PreTrainedModel.save_pretrained`."
},
)
reward_funcs: Optional[list[str]] = field(
default=None,
metadata={
"help": "Reward functions to use. It can be either one of 'think_format_reward'; or a dotted "
"import path. (e.g., 'my_lib.rewards.custom_reward')."
},
)
def main(script_args, training_args, model_args):
# Load a pretrained model
model = AutoModelForCausalLM.from_pretrained(
model_args.model_name_or_path, trust_remote_code=model_args.trust_remote_code
)
tokenizer = AutoTokenizer.from_pretrained(
model_args.model_name_or_path, trust_remote_code=model_args.trust_remote_code
)
# Get the reward models and functions
reward_funcs = []
if script_args.reward_model_name_or_path:
reward_model = AutoModelForSequenceClassification.from_pretrained(
script_args.reward_model_name_or_path, trust_remote_code=model_args.trust_remote_code, num_labels=1
)
reward_funcs.append(reward_model)
if script_args.reward_funcs:
for func_name in script_args.reward_funcs:
if func_name in reward_funcs_registry:
reward_funcs.append(reward_funcs_registry[func_name])
elif "." in func_name:
module_path, func_name = func_name.rsplit(".", 1)
sys.path.insert(0, os.getcwd())
module = importlib.import_module(module_path)
reward_func = getattr(module, func_name)
reward_funcs.append(reward_func)
else:
raise ValueError(
f"Could not load reward function '{func_name}'. Expected one of "
f"{list(reward_funcs_registry.keys())} or a valid import path."
)
# Load the dataset
dataset = load_dataset(script_args.dataset_name, name=script_args.dataset_config)
# Initialize the GRPO trainer
trainer = GRPOTrainer(
model=model,
reward_funcs=reward_model,
args=training_args,
train_dataset=dataset[script_args.dataset_train_split],
eval_dataset=dataset[script_args.dataset_test_split] if training_args.eval_strategy != "no" else None,
processing_class=tokenizer,
peft_config=get_peft_config(model_args),
)
# Train and push the model to the Hub
trainer.train()
# Save and push to hub
trainer.save_model(training_args.output_dir)
if training_args.push_to_hub:
trainer.push_to_hub(dataset_name=script_args.dataset_name)
def make_parser(subparsers: argparse._SubParsersAction = None):
dataclass_types = (GRPOScriptArguments, GRPOConfig, ModelConfig)
if subparsers is not None:
parser = subparsers.add_parser("grpo", help="Run the GRPO training script", dataclass_types=dataclass_types)
else:
parser = TrlParser(dataclass_types)
return parser
if __name__ == "__main__":
parser = make_parser()
script_args, training_args, model_args = parser.parse_args_and_config()
main(script_args, training_args, model_args)
|