Tutorial: Wie man gpt-oss mit RL trainiert
Lerne, OpenAI gpt-oss mit GRPO zu trainieren, um lokal oder auf Colab autonom 2048 zu schlagen.
1
Unsloth installieren
!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
gpt-oss mit Unsloth laden
from unsloth import FastLanguageModel
import torch
max_seq_length = 768 # Erhöhe, wenn deine Aufgabe längere Ausgaben benötigt
lora_rank = 4 # Höherer Rang → besser, aber mehr VRAM/Compute
model, tokenizer = FastLanguageModel.from_pretrained(
model_name = "unsloth/gpt-oss-20b", # oder unsloth/gpt-oss-20b-BF16 auf H100
max_seq_length = max_seq_length,
load_in_4bit = True, # False für 16‑Bit
offload_embedding = True, # spart ~1GB 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", # großer Speicher-Sparer
random_state = 3407,
)3
4
Sichere Codeausführung & Anti‑Cheat-Checks
from unsloth import check_python_modules ok, info = check_python_modules(""" def strategy(board): import math from typing import Callable return "W" """) # ok == True bedeutet, dass nur Python‑Level-Imports verwendet wurdensample = """ 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) # Fehler, falls Globals / Imports verwendet werdenfrom unsloth import execute_with_time_limit @execute_with_time_limit(2) def execute_strategy(strategy, game): # Schleife bis das Spiel endet oder die Zeit abläuft ...
5
Erstelle eine neue kurze 2048-Strategie, die nur native Python-Code verwendet.
Du erhältst eine Liste von Listen mit Zahlen für den aktuellen Spielbrettzustand.
Gib eine Aktion für "W", "A", "S", "D" aus, die der optimale nächste Schritt ist.
Gib deine neue kurze Funktion in Backticks im untenstehenden Format aus:
```python
def strategy(board):
return "W" # Beispiel
Erstelle einen winzigen synthetischen Datensatz (denselben Prompt wiederverwenden) und berechne die Prompt-Länge, damit GRPO weiß, wie viele Completion-Tokens zu sampeln sind:
```python
from datasets import Dataset
prompt = ... # wie oben
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)Zeit für die Belohnungsfunktion!
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) # harte Strafe bei Betrug return scoresimport numpy as np PRINTER = 0 # gelegentliches Drucken zum Debuggen 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) # hat funktioniert, aber 2048 nicht erreicht except TimeoutError: scores.append(-1.0) # Zeitüberschreitung except Exception: scores.append(-3.0) # abgestürzt return scores
GRPO konfigurieren
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, # erhöhe auf 4 für gleichmäßigere Belohnungssignale
num_generations=2, # reduziere, wenn du OOM bekommst
max_prompt_length=max_prompt_length,
max_completion_length=max_completion_length,
max_steps=1000, # oder setze 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,
# Optionale Eval-Aufteilung:
# train_dataset=new_dataset["train"],
# eval_dataset=new_dataset["test"],
)Trainiere dein Modell
trainer.train()Inference (nach dem Training)
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)Speichere / exportiere dein feinabgestimmtes Modell
model.save_pretrained_merged("finetuned_model", tokenizer, save_method="merged_16bit") # oder push model.push_to_hub_merged("<org_or_user>/<repo>", tokenizer, token="<hf_token>", save_method="merged_16bit")
Zuletzt aktualisiert
War das hilfreich?