# 教程：如何使用 RL 训练 gpt-oss

LLM 通常很难处理涉及复杂环境的任务。不过，通过应用 [强化学习](/docs/zh/kai-shi-shi-yong/reinforcement-learning-rl-guide.md) （RL）并设计一个自定义的 [奖励函数](/docs/zh/kai-shi-shi-yong/reinforcement-learning-rl-guide.md#reward-functions-verifiers)，这些挑战可以被克服。

RL 可适用于自动生成内核或策略创建等任务。本教程展示如何训练 **gpt-oss** 使用 [**GRPO**](/docs/zh/kai-shi-shi-yong/reinforcement-learning-rl-guide.md#from-rlhf-ppo-to-grpo-and-rlvr) 和 Unsloth，自主击败 2048。

| [2048 笔记本](https://colab.research.google.com/github/openai/gpt-oss/blob/main/examples/reinforcement-fine-tuning.ipynb) （OpenAI 官方示例） | [内核生成笔记本](https://colab.research.google.com/github/unslothai/notebooks/blob/main/nb/gpt-oss-\(20B\)-GRPO.ipynb) |
| ------------------------------------------------------------------------------------------------------------------------------------ | --------------------------------------------------------------------------------------------------------------- |

**你将构建什么：**

* 训练 gpt-oss-20b，使模型可以自动赢得 2048
* 创建一个模型可以交互的最小化 2048 环境
* 定义 **奖励函数** ，用于：
  1. 检查生成的策略是否可编译并运行，
  2. 防止奖励黑客（禁止外部导入），以及
  3. 奖励实际的游戏成功
* 运行推理并导出模型（MXFP4 4 位或合并 FP16）

{% hint style="info" %}
**硬件：** 2048 示例可以在免费的 Colab T4 上运行，但训练会很慢。A100/H100 快得多。4 位加载 + LoRA 让你可以把一个 20B 模型放进较小的显存中。
{% endhint %}

{% stepper %}
{% step %}

#### 安装 Unsloth

在笔记本顶部运行此单元（Colab 可用）。

```bash
!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.2
```

{% endstep %}

{% step %}

#### 使用 Unsloth 加载 gpt-oss

以 4 位 QLoRA 方式加载 20B 模型以提高内存效率，然后为其包装一个 LoRA 适配器。你也可以用 16 位 LoRA 训练，但会占用 4 倍内存。更多设置请查看我们的 [配置指南](https://unsloth.ai/docs/zh/mo-xing/gpt-oss-how-to-run-and-fine-tune/gpt-oss-reinforcement-learning/pages/e722e86e330786e5915445b91f900d9f9e0ba067#id-2.-choose-the-right-model--method).

```python
from unsloth import FastLanguageModel
import torch

max_seq_length = 768        # 如果任务需要更长输出可增大
lora_rank      = 4          # 更高的秩 → 更好，但需要更多显存/算力

model, tokenizer = FastLanguageModel.from_pretrained(
    model_name        = "unsloth/gpt-oss-20b",  # 或在 H100 上使用 unsloth/gpt-oss-20b-BF16
    max_seq_length    = max_seq_length,
    load_in_4bit      = True,                    # 16 位时设为 False
    offload_embedding = True,                    # 约节省 1GB 显存
)

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",     # 节省大量内存
    random_state = 3407,
)
```

{% hint style="info" %}
如果遇到 OOM，尝试降低 `max_seq_length`, `lora_rank`，或 `num_generations` （稍后），并保持 `load_in_4bit=True`.
{% endhint %}
{% endstep %}

{% step %}

#### 2048 游戏环境（最小化）

* 一个 `GameBoard` 类，支持 **W/A/S/D** 移动
* 合并/计分逻辑
* `execute_with_time_limit` 封装器，防止糟糕编写的策略挂死内核

你可以先用一个很简单的策略快速冒烟测试：

```python
def always_move_left(board):
    return "W"

steps, outcome = execute_strategy(always_move_left, GameBoard(size=8, seed=42, target=2048, probability_fours=0.10))
```

{% endstep %}

{% step %}

#### 安全代码执行与反作弊检查

生成的策略是 **Python 函数**。为了确保执行安全并防止奖励黑客：

* **模块白名单检查** — 只允许 Python 标准库符号：

  ```python
  from unsloth import check_python_modules
  ok, info = check_python_modules("""
  def strategy(board):
      import math
      from typing import Callable
      return "W"
  """)
  # ok == True 表示只使用了 Python 层面的导入
  ```
* **阻止不允许的导入** （例如 NumPy）：

  ```python
  sample = """
  def strategy(board):
      from numpy import matmul
      return "W"
  """
  ok, info = check_python_modules(sample)  # ok => False
  ```
* **将执行锁定** 到一个沙箱化函数中：

  ```python
  from 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)  # 若使用了全局变量/导入则报错
  ```
* **强制执行严格的墙钟时间限制** 用于策略运行：

  ```python
  from unsloth import execute_with_time_limit
  @execute_with_time_limit(2)
  def execute_strategy(strategy, game):
      # 循环直到游戏结束或超时
      ...
  ```

{% endstep %}

{% step %}
\### 提示词与数据集

我们提示模型 **生成一个简短的策略函数** 并放在三引号内：

````
使用仅原生 Python 代码创建一个新的简短 2048 策略。
当前棋盘状态会以数值列表的列表形式提供给你。
针对 "W"、"A"、"S"、"D" 输出一个动作，表示下一步的最优选择。
请使用下面的格式，在反引号中输出你的新简短函数：
```python
def strategy(board):
    return "W"  # 示例
````

所有辅助函数都应放在 def strategy 内。只输出这个简短函数 `strategy`.

````

创建一个很小的合成数据集（复用同一个提示词），并计算提示词长度，以便 GRPO 知道需要采样多少 completion token：

```python
from datasets import Dataset

prompt = ...  # 如上

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)
````

{% hint style="info" %} 你可以将此数据集替换为你自己的 RL 任务的真实提示词。 {% endhint %} {% endstep %}

{% step %}

#### 奖励函数时间！

1. **从模型回复中** 提取代码块：

   ````python
   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 None
   ````
2. **`function_works`** - 是否可编译并创建可调用对象？

   ```python
   from 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 scores
   ```
3. **`no_cheating`** - 不允许任何非标准库导入：

   ```python
   def 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)  # 若作弊则重罚
       return scores
   ```
4. **`strategy_succeeds`** - 运行一个随机棋盘；成功则奖励：

   ```python
   import numpy as np

   PRINTER = 0  # 偶尔打印用于调试

   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)   # 有效果，但没有达到 2048
           except TimeoutError:
               scores.append(-1.0)      # 超时
           except Exception:
               scores.append(-3.0)      # 崩溃
       return scores
   ```

{% endstep %}

{% step %}

#### 配置 GRPO

我们将使用 **GRPOTrainer**。设置提示词/补全长度，然后构建一个 `GRPOConfig`。请记住，你也可以将 RL 算法类型设置为其他类型，例如 [GSPO](/docs/zh/kai-shi-shi-yong/reinforcement-learning-rl-guide/advanced-rl-documentation/gspo-reinforcement-learning.md) 或 Dr. GRPO。

```python
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,    # 如需更平滑的奖励信号可增至 4
    num_generations=2,                # 如果 OOM 可调低
    max_prompt_length=max_prompt_length,
    max_completion_length=max_completion_length,
    max_steps=1000,                   # 或设置 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,
    # 可选的验证集划分：
    # train_dataset=new_dataset["train"],
    # eval_dataset=new_dataset["test"],
)
```

{% hint style="info" %} **读取日志：** 关注 `reward` 和 `reward_std`。在训练早期看到较低/为零的奖励是正常的（小显卡上前约 100–200 步）。 {% endhint %} {% endstep %}

{% step %}

#### 训练你的模型

```python
trainer.train()
```

这将启动完整的 RL 循环：采样补全 → 使用你的奖励打分 → 优化策略（LoRA）。 {% endstep %}

{% step %}

#### 推理（训练后）

使用训练好的适配器生成一个新的策略：

```python
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)
```

{% endstep %}

{% step %}

#### 保存 / 导出你的微调模型

* **合并并保存 4 位（MXFP4）**

  ```
  ```

python model.save\_pretrained\_merged("finetuned\_model", tokenizer, save\_method="mxfp4") # 或 model.push\_to\_hub\_merged("\<org\_or\_user>/", tokenizer, token="\<hf\_token>", save\_method="mxfp4") \`\`\`

* **合并并保存 16 位**

  ```python
  model.save_pretrained_merged("finetuned_model", tokenizer, save_method="merged_16bit")
  # 或推送
  model.push_to_hub_merged("<org_or_user>/<repo>", tokenizer, token="<hf_token>", save_method="merged_16bit")
  ```

{% endstep %}

{% step %}

#### 故障排查与提示

* **OOM / 运行慢**：降低 `max_seq_length`, `num_generations`, `lora_rank`；保持 4 位；如果可用，尝试 A100。
* **奖励没有改善**：增加训练步数，减轻惩罚，或添加课程学习（从更小的棋盘/更低目标开始）。
* **奖励黑客**：保持 `check_python_modules` 严格；在多个随机种子上验证策略行为。
* **训练不稳定**：提高 `gradient_accumulation_steps` 以平滑更新；降低 `learning_rate` （例如 2e-5）。
* **长时间卡住**：确保 `execute_with_time_limit` 包装了任何策略执行。
  {% endstep %}

{% step %}

#### 适配到你自己的 RL 任务

* 用你自己的环境替换 2048 环境，并使用 **三个奖励**：a）语法/编译，b）反作弊/安全，c）任务成功。
* 更新 **提示词** ，以请求你所需的函数或输出类型。
* 保持相同的 Unsloth + GRPO 脚手架；只替换环境和奖励。
  {% endstep %}
  {% endstepper %}


---

# Agent Instructions: Querying This Documentation

If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter:

```
GET https://unsloth.ai/docs/zh/mo-xing/gpt-oss-how-to-run-and-fine-tune/gpt-oss-reinforcement-learning/tutorial-how-to-train-gpt-oss-with-rl.md?ask=<question>
```

The question should be specific, self-contained, and written in natural language.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.