sft_trainer.py — SFTTrainer (most common fine-tuning)
from trl import SFTTrainer, SFTConfig
from peft import LoraConfig
# ── SFTConfig = TrainingArguments + SFT-specific params ────────────────────
sft_config = SFTConfig(
output_dir = "./sft-output",
# inherited from TrainingArguments
per_device_train_batch_size = 2,
gradient_accumulation_steps = 8, # effective batch = 16
num_train_epochs = 3,
learning_rate = 2e-4, # higher than full FT (adapters start at zero)
lr_scheduler_type = "cosine",
warmup_ratio = 0.03,
bf16 = True,
gradient_checkpointing = True,
gradient_checkpointing_kwargs = {"use_reentrant": False},
optim = "adamw_8bit",
max_grad_norm = 0.3,
report_to = "wandb",
# SFT-specific (added by TRL)
max_length = 4096, # max tokens per example
packing = True, # pack short examples → throughput win
dataset_num_proc = 8, # parallel dataset tokenization
assistant_only_loss = True, # mask user/system tokens → loss only on assistant
)
# ── Pass peft_config: SFTTrainer calls get_peft_model() internally ─────────
lora_config = LoraConfig(
r = 16,
lora_alpha = 16,
target_modules = ["q_proj", "k_proj", "v_proj", "o_proj",
"gate_proj", "up_proj", "down_proj"],
lora_dropout = 0.0,
bias = "none",
task_type = "CAUSAL_LM",
)
# ── Dataset: "messages" column → SFTTrainer auto-applies Qwen's ChatML template
# Each row: {"messages": [{"role": "system",...}, {"role": "user",...}, {"role": "assistant",...}]}
trainer = SFTTrainer(
model = model,
tokenizer = tokenizer,
train_dataset = dataset, # "messages" column triggers auto chat-template
args = sft_config,
peft_config = lora_config, # SFTTrainer wraps model in PEFT for you
)
trainer.train()
grpo_trainer.py — GRPO (RL training, no critic model needed)
from trl import GRPOTrainer, GRPOConfig
# ── Define reward functions — these drive the RL signal ────────────────────
def format_reward(prompts, completions, **kwargs) -> list[float]:
"""Reward 1.0 if response has <think>...</think> structure."""
rewards = []
for completion in completions:
has_think = "<think>" in completion and "</think>" in completion
rewards.append(1.0 if has_think else 0.0)
return rewards
def correctness_reward(prompts, completions, answer, **kwargs) -> list[float]:
"""Reward 2.0 if extracted answer matches ground truth."""
rewards = []
for completion, gt in zip(completions, answer):
extracted = extract_answer(completion) # your parsing logic
rewards.append(2.0 if extracted == gt else 0.0)
return rewards
# ── GRPOConfig: TrainingArguments + GRPO-specific params ───────────────────
grpo_config = GRPOConfig(
output_dir = "./grpo-output",
per_device_train_batch_size = 1,
gradient_accumulation_steps = 8,
num_train_epochs = 2,
learning_rate = 1e-5, # lower than SFT — RL is sensitive
bf16 = True,
# GRPO-specific
num_generations = 8, # N samples per prompt for group baseline
max_completion_length = 1024, # longer for reasoning chains
beta = 0.04, # KL penalty — prevents reward hacking
epsilon = 0.2, # PPO-style policy ratio clipping
use_vllm = False, # True = 3-5× faster generation step
)
trainer = GRPOTrainer(
model = model,
tokenizer = tokenizer,
train_dataset = dataset,
reward_funcs = [format_reward, correctness_reward], # mix multiple signals
args = grpo_config,
)
trainer.train()