GRPO/RL Training with TRL

Transformer Reinforcement Learning (TRL) ライブラリを使用した Group Relative Policy Optimization (GRPO) の実装に関するエキスパートレベルの指南です。このスキルは、カスタム報酬関数を用いた言語モデルの微調整のための実戦で検証されたパターン、重要な洞察、本番環境対応ワークフローを提供します。

このスキルを使う時

以下の場合に GRPO トレーニングを使用してください:

• 特定の出力形式を強制する (例: XML タグ、JSON、構造化推論)
• 検証可能なタスクを教える (客観的な正確性メトリクスを持つ数学、コーディング、ファクトチェック)
• 推論能力を改善する (チェーン・オブ・ソートパターンに報酬を与える)
• ラベル付き選好データなしで、モデルをドメイン特有の動作に合わせる
• 複数の目的を同時に最適化する (形式 + 正確性 + スタイル)

GRPO を使用しないでください:

• 単純な教師あり微調整タスク (代わりに SFT を使用)
• 明確な報酬信号のないタスク
• 既に高品質な選好ペアがある場合 (代わりに DPO/PPO を使用)

---

コア概念

1. GRPO アルゴリズムの基礎

主要メカニズム:

• 各プロンプトに対して 複数の完成例を生成 (グループサイズ: 4-16)
• 報酬関数を使用してグループ内の完成例を比較
• 高報酬の応答をグループ内の相対的に有利なものになるようにポリシーを更新

PPO との重要な違い:

• 別の報酬モデルが不要
• より少ないサンプルで効率的 (グループ内比較から学習)
• 実装とデバッグが簡単

数学的直感:

各プロンプト p に対して:
  1. N 個の完成例を生成: {c₁, c₂, ..., cₙ}
  2. 報酬を計算: {r₁, r₂, ..., rₙ}
  3. 同じグループ内の低報酬の完成例と相対的に
     高報酬の完成例の確率を増やすように学習

2. 報酬関数設計の哲学

黄金ルール:

1. 複数の報酬関数を組み合わせる - 各関数が 1 つの側面を処理 (形式、正確性、スタイル)
2. 報酬を適切にスケーリング - 高ウェイト = より強い信号
3. 段階的な報酬を使用 - 部分的なコンプライアンスに対する部分的なクレジット
4. 報酬を個別にテスト - 各報酬関数をアイソレーションでデバッグ

報酬関数のタイプ:

タイプ	ユースケース	例のウェイト
正確性	検証可能なタスク (数学、コード)	2.0 (最高)
形式	厳格な構造強制	0.5-1.0
長さ	冗長性/簡潔性を促進	0.1-0.5
スタイル	不要なパターンを罰する	-0.5 to 0.5

---

実装ワークフロー

ステップ 1: データセット準備

重要な要件:

• チャット形式のプロンプト ('role' と 'content' を持つ辞書のリスト)
• 期待値を設定するシステムプロンプトを含む
• 検証可能なタスクの場合、追加列として実測値を含める

例の構造:

from datasets import load_dataset, Dataset

SYSTEM_PROMPT = """
以下の形式で応答してください:
<reasoning>
[段階的な思考プロセス]
</reasoning>
<answer>
[最終的な答え]
</answer>
"""

def prepare_dataset(raw_data):
    """
    生のデータを GRPO 互換形式に変換します。

    返すデータセット: 以下の列を持つデータセット:
    - 'prompt': List[Dict] (役割/内容を持つ、システム + ユーザーメッセージ)
    - 'answer': str (実測値、オプションですが推奨)
    """
    return raw_data.map(lambda x: {
        'prompt': [
            {'role': 'system', 'content': SYSTEM_PROMPT},
            {'role': 'user', 'content': x['question']}
        ],
        'answer': extract_answer(x['raw_answer'])
    })

プロ向けのヒント:

• 複雑な形式のシステムプロンプトで 1 ショットまたは数ショットの例を使用
• プロンプトは簡潔に (max_prompt_length: 256-512 トークン)
• トレーニング前にデータ品質を検証 (ゴミを入れる = ゴミが出る)

ステップ 2: 報酬関数の実装

テンプレート構造:

def reward_function_name(
    prompts,        # List[List[Dict]]: オリジナルプロンプト
    completions,    # List[List[Dict]]: モデル生成
    answer=None,    # オプション: データセットからの実測値
    **kwargs        # 追加のデータセット列
) -> list[float]:
    """
    完成例を評価し、報酬を返します。

    返す: リスト[float] (完成例ごとに 1 つ)
    """
    # 完成例テキストを抽出
    responses = [comp[0]['content'] for comp in completions]

    # 報酬を計算
    rewards = []
    for response in responses:
        score = compute_score(response)
        rewards.append(score)

    return rewards

例 1: 正確性報酬 (数学/コーディング)

def correctness_reward(prompts, completions, answer, **kwargs):
    """正しい答えに高スコアで報酬を与えます。"""
    responses = [comp[0]['content'] for comp in completions]
    extracted = [extract_final_answer(r) for r in responses]
    return [2.0 if ans == gt else 0.0
            for ans, gt in zip(extracted, answer)]

例 2: 形式報酬 (構造化出力)

import re

def format_reward(completions, **kwargs):
    """XML のような構造化形式に報酬を与えます。"""
    pattern = r'<reasoning>.*?</reasoning>\s*<answer>.*?</answer>'
    responses = [comp[0]['content'] for comp in completions]
    return [1.0 if re.search(pattern, r, re.DOTALL) else 0.0
            for r in responses]

例 3: 段階的形式報酬 (部分的なクレジット)

def incremental_format_reward(completions, **kwargs):
    """形式コンプライアンスに対して部分的なクレジットを付与します。"""
    responses = [comp[0]['content'] for comp in completions]
    rewards = []

    for r in responses:
        score = 0.0
        if '<reasoning>' in r:
            score += 0.25
        if '</reasoning>' in r:
            score += 0.25
        if '<answer>' in r:
            score += 0.25
        if '</answer>' in r:
            score += 0.25
        # 閉じるタグの後の余分なテキストを罰する
        if r.count('</answer>') == 1:
            extra_text = r.split('</answer>')[-1].strip()
            score -= len(extra_text) * 0.001
        rewards.append(score)

    return rewards

重要な洞察: 堅牢なトレーニングのために 3～5 個の報酬関数を組み合わせてください。順序の重要性は信号の多様性ほど高くはありません。

ステップ 3: トレーニング設定

メモリ最適化設定 (小さい GPU)

from trl import GRPOConfig

training_args = GRPOConfig(
    output_dir="outputs/grpo-model",

    # 学習率
    learning_rate=5e-6,          # 低い = より安定
    adam_beta1=0.9,
    adam_beta2=0.99,
    weight_decay=0.1,
    warmup_ratio=0.1,
    lr_scheduler_type='cosine',

    # バッチ設定
    per_device_train_batch_size=1,
    gradient_accumulation_steps=4,  # 有効バッチ = 4

    # GRPO 固有
    num_generations=8,            # グループサイズ: 8-16 推奨
    max_prompt_length=256,
    max_completion_length=512,

    # トレーニング期間
    num_train_epochs=1,
    max_steps=None,               # または固定ステップを設定 (例: 500)

    # 最適化
    bf16=True,                    # A100/H100 で高速
    optim="adamw_8bit",          # メモリ効率の良いオプティマイザー
    max_grad_norm=0.1,

    # ロギング
    logging_steps=1,
    save_steps=100,
    report_to="wandb",            # または "none" でロギングなし
)

高性能設定 (大きい GPU)

training_args = GRPOConfig(
    output_dir="outputs/grpo-model",
    learning_rate=1e-5,
    per_device_train_batch_size=4,
    gradient_accumulation_steps=2,
    num_generations=16,           # より大きなグループ = より強い信号
    max_prompt_length=512,
    max_completion_length=1024,
    num_train_epochs=1,
    bf16=True,
    use_vllm=True,                # vLLM での高速生成
    logging_steps=10,
)

重要なハイパーパラメーター:

パラメーター	影響	チューニングアドバイス
num_generations	比較用のグループサイズ	8 で始める、GPU が許可すれば 16 に増やす
learning_rate	収束速度/安定性	5e-6 (安全)、1e-5 (高速、リスキー)
max_completion_length	出力の冗長性	タスクに合わせる (推論 512、短い答え 256)
gradient_accumulation_steps	有効バッチサイズ	GPU メモリが制限されている場合は増加

ステップ 4: モデルセットアップとトレーニング

標準セットアップ (Transformers)

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
from peft import LoraConfig
from trl import GRPOTrainer

# モデルを読み込む
model_name = "Qwen/Qwen2.5-1.5B-Instruct"
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    torch_dtype=torch.bfloat16,
    attn_implementation="flash_attention_2",  # 2-3 倍高速
    device_map="auto"
)

tokenizer = AutoTokenizer.from_pretrained(model_name)
tokenizer.pad_token = tokenizer.eos_token

# オプション: パラメーター効率の高いトレーニングの為の LoRA
peft_config = LoraConfig(
    r=16,                         # ランク (高い = より多くの容量)
    lora_alpha=32,               # スケーリング係数 (通常 2*r)
    target_modules=[
        "q_proj", "k_proj", "v_proj", "o_proj",
        "gate_proj", "up_proj", "down_proj"
    ],
    task_type="CAUSAL_LM",
    lora_dropout=0.05,
)

# トレーナーを初期化
trainer = GRPOTrainer(
    model=model,
    processing_class=tokenizer,
    reward_funcs=[
        incremental_format_reward,
        format_reward,
        correctness_reward,
    ],
    args=training_args,
    train_dataset=dataset,
    peft_config=peft_config,      # 完全なファインチューニングの場合は削除
)

# トレーニング
trainer.train()

# 保存
trainer.save_model("final_model")

Unsloth セットアップ (2-3 倍高速)

from unsloth import FastLanguageModel

model, tokenizer = FastLanguageModel.from_pretrained(
    model_name="google/gemma-3-1b-it",
    max_seq_length=1024,
    load_in_4bit=True,
    fast_inference=True,
    max_lora_rank=32,
)

model = FastLanguageModel.get_peft_model(
    model,
    r=32,
    target_modules=["q_proj", "k_proj", "v_proj", "o_proj",
                    "gate_proj", "up_proj", "down_proj"],
    lora_alpha=32,
    use_gradient_checkpointing="unsloth",
)

# 標準セットアップと同じ
trainer = GRPOTrainer(model=model, ...)
trainer.train()

---

重要なトレーニングの洞察

1. ロス動作 (期待されるパターン)

• ロスは 0 近くで始まり、トレーニング中に増加します
• これは 正しい - ロスは初期ポリシーからの KL ダイバージェンスを測定します
• モデルは学習しています (報酬を最適化するために元の動作から逸脱)
• 進捗状況の代わりに報酬メトリクスを監視してください

2. 報酬追跡

監視するべき主要メトリクス:

• reward: すべての完成例全体の平均
• reward_std: グループ内の多様性 (> 0 のままである必要があります)
• kl: 参照からの KL ダイバージェンス (適度に成長する必要があります)

健全なトレーニングパターン:

ステップ   報酬    報酬_標準偏差   KL
100    0.5       0.3          0.02
200    0.8       0.25         0.05
300    1.2       0.2          0.08  ← 良い進捗
400    1.5       0.15         0.12

警告信号:

• 報酬 std → 0 (モデルが単一応答に崩壊)
• KL が爆発 (> 0.5) (発散しすぎ、LR を減らす)
• 報酬が止まる (報酬関数が厳しすぎるか、モデル容量の問題)

3. 一般的な落とし穴と解決策

問題	症状	ソリューション
モード崩壊	すべての完成例が同じ	num_generations を増やす、多様性ペナルティを追加
学習なし	フラットな報酬	報酬関数ロジックを確認、LR を上げる
OOM エラー	GPU メモリ超過	num_generations を減らす、勾配チェックポイント有効化
遅いトレーニング	< 1 it/s	use_vllm=True 有効化、Unsloth 使用、シーケンス長削減
形式が無視される	モデルが構造に従わない	形式報酬ウェイトを増やす、段階的報酬を追加

---

高度なパターン

1. マルチステージトレーニング

複雑なタスクの場合、ステージで トレーニング:

# ステージ 1: 形式コンプライアンス (エポック=1)
trainer_stage1 = GRPOTrainer(
    model=model,
    reward_funcs=[incremental_format_reward, format_reward],
    ...
)
trainer_stage1.train()

# ステージ 2: 正確性 (エポック=1)
trainer_stage2 = GRPOTrainer(
    model=model,
    reward_funcs=[format_reward, correctness_reward],
    ...
)
trainer_stage2.train()

2. 適応的報酬スケーリング

class AdaptiveReward:
    def __init__(self, base_reward_func, initial_weight=1.0):
        self.func = base_reward_func
        self.weight = initial_weight

    def __call__(self, *args, **kwargs):
        rewards = self.func(*args, **kwargs)
        return [r * self.weight for r in rewards]

    def adjust_weight(self, success_rate):
        """モデルが苦労している場合はウェイト増加、成功している場合は減少。"""
        if success_rate < 0.3:
            self.weight *= 1.2
        elif success_rate > 0.8:
            self.weight *= 0.9

3. カスタムデータセット統合

def load_custom_knowledge_base(csv_path):
    """例: 学校コミュニケーションプラットフォームドキュメント。"""
    import pandas as pd
    df = pd.read_csv(csv_path)

    dataset = Dataset.from_pandas(df).map(lambda x: {
        'prompt': [
            {'role': 'system', 'content': CUSTOM_SYSTEM_PROMPT},
            {'role': 'user', 'content': x['question']}
        ],
        'answer': x['expert_answer']
    })
    return dataset

---

デプロイメントと推論

LoRA を保存とマージ

# LoRA アダプターをベースモデルにマージ
if hasattr(trainer.model, 'merge_and_unload'):
    merged_model = trainer.model.merge_and_unload()
    merged_model.save_pretrained("production_model")
    tokenizer.save_pretrained("production_model")

推論の例

from transformers import pipeline

generator = pipeline(
    "text-generation",
    model="production_model",
    tokenizer=tokenizer
)

result = generator(
    [
        {'role': 'system', 'content': SYSTEM_PROMPT},
        {'role': 'user', 'content': "15 + 27 は何ですか?"}
    ],
    max_new_tokens=256,
    do_sample=True,
    temperature=0.7,
    top_p=0.9
)
print(result[0]['generated_text'])

---

ベストプラクティスチェックリスト

トレーニング前:

• データセット形式を検証 (プロンプトが List[Dict])
• サンプルデータで報酬関数をテスト
• データから期待される max_prompt_length を計算
• GPU メモリに基づいて適切な num_generations を選択
• ロギングをセットアップ (wandb 推奨)

トレーニング中:

• 報酬進捗を監視 (増加する必要があります)
• reward_std をチェック (> 0.1 のままである必要があります)
• OOM エラーを監視 (必要に応じてバッチサイズを削減)
• 50-100 ステップごとに生成をサンプリング
• ホールドアウトセットで形式コンプライアンスを検証

トレーニング後:

• PEFT を使用している場合は LoRA ウェイトをマージ
• 多様なプロンプトでテスト
• ベースラインモデルと比較
• 報酬ウェイトとハイパーパラメーターを記述
• 再現性設定を保存

---

トラブルシューティングガイド

デバッグワークフロー

1. 報酬関数をアイソレート - 各関数を個別にテスト
2. データ分布を確認 - プロンプトの多様性を確保
3. 複雑性を低減 - 単一報酬で開始し、段階的に追加
4. 生成を監視 - N ステップごとにサンプルを出力
5. 抽出ロジックを検証 - 答えの解析が機能することを確保

クイックフィックス

# 報酬関数をデバッグ
def debug_reward(completions, **kwargs):
    responses = [comp[0]['content'] for comp in completions]
    for i, r in enumerate(responses[:2]):  # 最初の 2 つを出力
        print(f"レスポンス {i}: {r[:200]}...")
    return [1.0] * len(responses)  # ダミー報酬

# トレーニング無しでテスト
trainer = GRPOTrainer(..., reward_funcs=[debug_reward])
trainer.generate_completions(dataset[:1])  # 更新なしで生成

---

参考文献とリソース

公式ドキュメント:

• TRL GRPO トレーナー: https://huggingface.co/docs/trl/grpo_trainer
• DeepSeek R1 論文: https://arxiv.org/abs/2501.12948
• Unsloth ドキュメント: https://docs.unsloth.ai/

例リポジトリ:

• Open R1 実装: https://github.com/huggingface/open-r1
• TRL の例: https://github.com/huggingface/trl/tree/main/examples

推奨読書:

• エージェント指示の段階的公開パターン
• RL での報酬形成 (Ng et al.)
• LoRA 論文 (Hu et al., 2021)

---

エージェント向けの使用指示

このスキルが読み込まれたとき:

1. このファイル全体を読む GRPO トレーニングを実装する前に
2. 最もシンプルな報酬関数から始める (例: 長さベース) セットアップを検証
3. テンプレートを使用 templates/ ディレクトリを開始点として
4. 例を参照 examples/ でタスク固有の実装
5. ワークフローに従う 順序を守る (ステップをスキップしない)
6. 段階的にデバッグ - 一度に 1 つの報酬関数を追加

重要なリマインダー:

• 常に複数の報酬関数を使用 (3-5 個が最適)
• ロスではなく報酬メトリクスを監視
• トレーニング前に報酬関数をテスト
• 小さく始める (num_generations=4)、段階的にスケール
• 頻繁にチェックポイントを保存 (100 ステップごと)

このスキルは エキスパートレベルの実装 のために設計されています。初心者は GRPO を試みる前に教師あり微調整から始めるべきです。