Tutorial: Wie man gpt-oss mit RL trainiert
Lerne, OpenAI gpt-oss mit GRPO zu trainieren, um 2048 autonom lokal oder auf Colab 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öhen, wenn Ihre Aufgabe längere Ausgaben erfordert
lora_rank = 4 # Höherer Rang → besser, aber mehr VRAM/Rechenleistung
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 ~1 GB 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 und Anti-Cheat-Prüfungen
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, wenn Globals / Imports verwendet werdenfrom unsloth import execute_with_time_limit @execute_with_time_limit(2) def execute_strategy(strategy, game): # Schleife bis zum Spielende oder Timeout ...
5
Erstelle eine neue kurze 2048-Strategie, die nur nativen Python-Code verwendet.
Dir wird eine Liste von Listen von Zahlen für den aktuellen Spielbrettzustand gegeben.
Gib eine Aktion für "W", "A", "S", "D" aus, basierend auf dem optimalen nächsten Schritt.
Gib deine neue kurze Funktion in Backticks im folgenden Format aus:
```python
def strategy(board):
return "W" # Beispiel
Erstelle einen winzigen synthetischen Datensatz (unter Wiederverwendung desselben Prompts) und berechne die Prompt-Länge, damit GRPO weiß, wie viele Completion-Tokens gesampelt werden sollen:
```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) # hohe Strafe bei Schummeln return scoresimport numpy as np PRINTER = 0 # gelegentlich zum Debuggen ausgeben 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) # funktionierte, erreichte aber nicht 2048 except TimeoutError: scores.append(-1.0) # Timeout 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, # auf 4 erhöhen für glattere Belohnungssignale
num_generations=2, # verringern, wenn OOM auftritt
max_prompt_length=max_prompt_length,
max_completion_length=max_completion_length,
max_steps=1000, # oder num_train_epochs=1 setzen
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,
# Optionaler Eval-Split:
# train_dataset=new_dataset["train"],
# eval_dataset=new_dataset["test"],
)Trainiere Ihr 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)Speichern / Exportieren Sie Ihr feinabgestimmtes Modell
model.save_pretrained_merged("finetuned_model", tokenizer, save_method="merged_16bit") # oder pushen model.push_to_hub_merged("<org_or_user>/<repo>", tokenizer, token="<hf_token>", save_method="merged_16bit")
Zuletzt aktualisiert
War das hilfreich?

