Tutoriel : comment entraîner gpt-oss avec le RL
Apprenez à entraîner OpenAI gpt-oss avec GRPO pour battre 2048 de manière autonome, 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é → meilleur, mais plus de VRAM/compute
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 bits
offload_embedding = True, # économise ~1 Go 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", # gros gain de mémoire
random_state = 3407,
)3
4
Exécution de code sûre et vérifications 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 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 jusqu’au timeout ...
5
Créez une nouvelle courte stratégie 2048 en utilisant uniquement du code Python natif.
On vous donne une liste de listes de nombres pour l’état actuel du plateau.
Produisez une action parmi "W", "A", "S", "D" selon le prochain meilleur coup.
Produisez votre nouvelle courte fonction entre des backticks en utilisant le format ci-dessous :
```python
def strategy(board):
return "W" # Exemple
Créez un minuscule jeu de données synthétique (en réutilisant le même prompt) et calculez la longueur du prompt afin que GRPO sache combien de jetons 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) # forte 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épassé except Exception: scores.append(-3.0) # a 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, # diminuez si vous obtenez 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,
# Split d’évaluation optionnel :
# train_dataset=new_dataset["train"],
# eval_dataset=new_dataset["test"],
)Entraînez votre modèle
trainer.train()Inférence (après l’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 ?

