Tutoriel : Comment entraîner gpt-oss avec RL
Apprenez à entraîner OpenAI gpt-oss avec GRPO pour battre de façon autonome 2048 localement ou sur Colab.
1
Installer Unsloth
!pip install --upgrade -qqq uv
try: import numpy; get_numpy = f"numpy=={numpy.__version__}"
except: get_numpy = "numpy"
!uv pip install -qqq \
"torch>=2.8.0" "triton>=3.4.0" {get_numpy} torchvision bitsandbytes "transformers==4.56.2" \
"unsloth_zoo[base] @ git+https://github.com/unslothai/unsloth-zoo" \
"unsloth[base] @ git+https://github.com/unslothai/unsloth" \
git+https://github.com/triton-lang/triton.git@05b2c186c1b6c9a08375389d5efe9cb4c401c075#subdirectory=python/triton_kernels
!uv pip install --upgrade --no-deps transformers==4.56.2 tokenizers
!uv pip install --no-deps trl==0.22.22
Charger gpt-oss avec Unsloth
from unsloth import FastLanguageModel
import torch
max_seq_length = 768 # Augmentez si votre tâche nécessite des sorties plus longues
lora_rank = 4 # Rang plus élevé → mieux mais plus de VRAM/puissance de calcul
model, tokenizer = FastLanguageModel.from_pretrained(
model_name = "unsloth/gpt-oss-20b", # ou unsloth/gpt-oss-20b-BF16 sur H100
max_seq_length = max_seq_length,
load_in_4bit = True, # False pour 16‑bit
offload_embedding = True, # économise ~1GB de VRAM
)
model = FastLanguageModel.get_peft_model(
model,
r = lora_rank,
target_modules = [
"q_proj", "k_proj", "v_proj", "o_proj",
"gate_proj", "up_proj", "down_proj",
],
lora_alpha = lora_rank * 2,
use_gradient_checkpointing = "unsloth", # important pour économiser la mémoire
random_state = 3407,
)3
4
Exécution de code sûre et contrôles anti‑triche
from unsloth import check_python_modules ok, info = check_python_modules(""" def strategy(board): import math from typing import Callable return "W" """) # ok == True signifie que seuls des imports au niveau Python standard ont été utiliséssample = """ def strategy(board): from numpy import matmul return "W" """ ok, info = check_python_modules(sample) # ok => Falsefrom unsloth import create_locked_down_function function = """ def add(a, b): def adder(a): return a + b return adder(b) + b """ f = create_locked_down_function(function) # erreurs si des globals / imports sont utilisésfrom unsloth import execute_with_time_limit @execute_with_time_limit(2) def execute_strategy(strategy, game): # boucle jusqu'à la fin du jeu ou expiration du temps ...
5
Créez une nouvelle courte stratégie 2048 en utilisant seulement du code Python natif.
On vous donne une liste de listes de nombres représentant l'état actuel du plateau.
Sortez une action pour "W", "A", "S", "D" correspondant à l'étape suivante optimale.
Sortez votre nouvelle courte fonction entre backticks en utilisant le format ci‑dessous :
```python
def strategy(board):
return "W" # Exemple
Créez un petit jeu de données synthétique (réutilisant le même prompt) et calculez la longueur du prompt afin que GRPO sache combien de tokens de complétion échantillonner :
```python
from datasets import Dataset
prompt = ... # comme ci‑dessus
maximum_length = len(tokenizer.apply_chat_template(
[{"role": "user", "content": prompt}], add_generation_prompt=True
))
dataset = Dataset.from_list([
{"prompt": [{"role": "user", "content": prompt}], "answer": 0, "reasoning_effort": "low"}
] * 1000)C'est l'heure de la fonction de récompense !
def extract_function(text): if text.count("```") >= 2: first = text.find("```") + 3 second = text.find("```", first) fx = text[first:second].strip() fx = fx.removeprefix("python\n") fx = fx[fx.find("def"):] if fx.startswith("def strategy(board):"): return fx return Nonefrom unsloth import create_locked_down_function, check_python_modules def function_works(completions, **kwargs): scores = [] for completion in completions: response = completion[0]["content"] function = extract_function(response) if function is None: scores.append(-2.0) continue ok, info = check_python_modules(function) if "error" in info: scores.append(-2.0) continue try: _ = create_locked_down_function(function) scores.append(1.0) except Exception: scores.append(-0.5) return scoresdef no_cheating(completions, **kwargs): scores = [] for completion in completions: response = completion[0]["content"] function = extract_function(response) if function is None: scores.append(-1.0) continue ok, _ = check_python_modules(function) scores.append(1.0 if ok else -20.0) # lourde pénalité en cas de triche return scoresimport numpy as np PRINTER = 0 # afficher occasionnellement pour le débogage def strategy_succeeds(completions, **kwargs): global PRINTER scores = [] seed = np.random.randint(10000) for completion in completions: response = completion[0]["content"] function = extract_function(response) if function is None: scores.append(-2.0) continue try: new_strategy = create_locked_down_function(function) except Exception: scores.append(0.0) continue try: game = GameBoard(size=6, seed=seed, target=2048, probability_fours=0.10) steps, state = execute_strategy(new_strategy, game) if PRINTER % 5 == 0: print(function) print(f"Steps={steps} State={state}") print(game.board().pretty()) PRINTER += 1 if state == "success": scores.append(20.0) else: scores.append(2.0) # a fonctionné mais n'a pas atteint 2048 except TimeoutError: scores.append(-1.0) # délai d'exécution dépassé except Exception: scores.append(-3.0) # planté return scores
Configurer GRPO
from trl import GRPOConfig, GRPOTrainer
max_prompt_length = maximum_length + 1
max_completion_length = max_seq_length - max_prompt_length
training_args = GRPOConfig(
temperature=1.0,
learning_rate=5e-5,
weight_decay=0.01,
warmup_ratio=0.1,
lr_scheduler_type="linear",
optim="adamw_8bit",
logging_steps=1,
per_device_train_batch_size=1,
gradient_accumulation_steps=1, # augmentez à 4 pour des signaux de récompense plus lisses
num_generations=2, # réduire si vous avez un OOM
max_prompt_length=max_prompt_length,
max_completion_length=max_completion_length,
max_steps=1000, # ou définissez num_train_epochs=1
save_steps=100,
report_to="none",
output_dir="outputs",
)
trainer = GRPOTrainer(
model=model,
processing_class=tokenizer,
reward_funcs=[function_works, no_cheating, strategy_succeeds],
args=training_args,
train_dataset=dataset,
# Fraction d'évaluation optionnelle :
# train_dataset=new_dataset["train"],
# eval_dataset=new_dataset["test"],
)Entraînez votre modèle
trainer.train()Inférence (après entraînement)
from transformers import TextStreamer
text = tokenizer.apply_chat_template(
[{"role": "user", "content": prompt}],
tokenize=False,
add_generation_prompt=True,
reasoning_effort="low",
)
_ = model.generate(
**tokenizer(text, return_tensors="pt").to("cuda"),
temperature=1.0,
max_new_tokens=1024,
streamer=TextStreamer(tokenizer, skip_prompt=False)Enregistrer / Exporter votre modèle affiné
model.save_pretrained_merged("finetuned_model", tokenizer, save_method="merged_16bit") # ou pousser model.push_to_hub_merged("<org_or_user>/<repo>", tokenizer, token="<hf_token>", save_method="merged_16bit")
Mis à jour
Ce contenu vous a-t-il été utile ?

