Gemma 4 fine-tuning 実践 — データ不足・過学習・品質問題を防ぐ完全ノウハウ

Gemma 4 の fine-tuning に挑戦して「データが足りない」「過学習してしまう」「評価方法がわからない」で詰まる人は多いです。公式のチュートリアルは環境セットアップまでは丁寧に書かれていますが、その先の「実際に高品質なモデルを作るための判断」については、あまり触れられていません。

このガイドでは、Gemma 4 の fine-tuning を実際にやってみてぶつかった問題と、それを乗り越えるための具体的な実装パターンをまとめました。

fine-tuning の前に確認すること

fine-tuning は万能ではありません。始める前に「本当に fine-tuning が必要か」を確認することをお勧めします。

import os
from google import genai
 
client = genai.Client(api_key=os.getenv("GOOGLE_API_KEY"))
 
# まずプロンプトエンジニアリングで解決できないか確認する
def test_with_prompting(example_inputs: list[str], expected_outputs: list[str]) -> float:
    """プロンプトだけで目標精度に達するか確認する"""
    model = "gemini-2.5-pro"
    
    # Few-shot プロンプトを構築
    examples = []
    for inp, out in zip(example_inputs[:5], expected_outputs[:5]):
        examples.append(f"入力: {inp}\n出力: {out}")
    
    system_prompt = f"""以下の形式で出力してください。
 
例:
{chr(10).join(examples)}
 
入力を受け取ったら、上記の例と同じ形式で出力を生成してください。"""
    
    # 残りのデータでテスト
    correct = 0
    for inp, expected in zip(example_inputs[5:], expected_outputs[5:]):
        response = client.models.generate_content(
            model=model,
            contents=f"入力: {inp}",
            config={"system_instruction": system_prompt}
        )
        predicted = response.text.strip()
        if predicted == expected:
            correct += 1
    
    accuracy = correct / max(len(example_inputs) - 5, 1)
    return accuracy
 
# 80%以上の精度が出るなら fine-tuning は不要かもしれない
# accuracy = test_with_prompting(inputs, outputs)
# if accuracy >= 0.80:
#     print("プロンプトで十分な精度が出ています。fine-tuning は不要かもしれません。")

プロンプトエンジニアリングで80%以上の精度が出るなら、fine-tuning コストをかけずに済む可能性があります。特定のスタイルや形式への適応、データプライバシーの要件、レイテンシの改善など、明確な理由がある場合に fine-tuning を検討してください。

データ準備: 品質が全てを決める

fine-tuning の結果は、90%はデータ品質で決まります。少量でも高品質なデータの方が、大量の低品質データよりずっと良い結果になります。

最小限のデータ量の目安

Gemma 4 の fine-tuning で良い結果が出るデータ量の目安：

分類タスク（2〜5クラス）: クラスあたり100〜500サンプル
テキスト生成・スタイル変換: 500〜2,000サンプル
ドメイン特化QA: 1,000〜5,000ペア
コード生成: 2,000〜10,000サンプル

これより少ない場合は、データ拡張か few-shot プロンプティングを検討してください。

データ品質チェックの実装

from datasets import Dataset
import pandas as pd
from collections import Counter
import re
 
def validate_training_data(data: list[dict]) -> dict:
    """学習データの品質チェックを実行する"""
    
    issues = {
        "duplicates": [],
        "empty_inputs": [],
        "empty_outputs": [],
        "too_short": [],
        "too_long": [],
        "format_errors": []
    }
    
    texts = []
    for i, sample in enumerate(data):
        # 必須フィールドの確認
        if "input" not in sample or "output" not in sample:
            issues["format_errors"].append(i)
            continue
        
        inp = sample["input"].strip()
        out = sample["output"].strip()
        
        # 空のサンプルをチェック
        if not inp:
            issues["empty_inputs"].append(i)
        if not out:
            issues["empty_outputs"].append(i)
        
        # 異常な長さをチェック（文字数で近似）
        if len(inp) < 10:
            issues["too_short"].append(i)
        if len(inp) > 4000 or len(out) > 2000:
            issues["too_long"].append(i)
        
        texts.append(inp)
    
    # 重複チェック
    text_counts = Counter(texts)
    duplicates = [t for t, c in text_counts.items() if c > 1]
    if duplicates:
        issues["duplicates"] = duplicates[:5]  # 最初の5件を示す
    
    # 統計情報
    stats = {
        "total_samples": len(data),
        "issues": {k: len(v) for k, v in issues.items() if v},
        "avg_input_length": sum(len(s.get("input", "")) for s in data) / len(data),
        "avg_output_length": sum(len(s.get("output", "")) for s in data) / len(data),
    }
    
    # 問題サマリー
    total_issues = sum(len(v) for v in issues.values())
    stats["quality_score"] = max(0, 1 - (total_issues / len(data)))
    
    return stats
 
# 使用例
training_data = [
    {"input": "日本の首都は？", "output": "東京です。"},
    {"input": "日本の首都は？", "output": "東京です。"},  # 重複
    {"input": "", "output": "空の入力"},  # 空のサンプル
]
 
stats = validate_training_data(training_data)
print(f"品質スコア: {stats['quality_score']:.2%}")
print(f"問題点: {stats['issues']}")

データ拡張で少量データを補完する

データが少ない場合、既存のデータから類似サンプルを生成して補完できます。

import random
from google import genai
import os
 
client = genai.Client(api_key=os.getenv("GOOGLE_API_KEY"))
 
def augment_training_data(
    original_data: list[dict],
    target_size: int,
    augmentation_model: str = "gemini-2.0-flash"
) -> list[dict]:
    """既存の学習データを拡張して目標サイズに近づける"""
    
    augmented = list(original_data)  # 元データを保持
    examples_text = "\n".join([
        f"入力: {d['input']}\n出力: {d['output']}"
        for d in random.sample(original_data, min(5, len(original_data)))
    ])
    
    attempts = 0
    max_attempts = (target_size - len(original_data)) * 2
    
    while len(augmented) < target_size and attempts < max_attempts:
        attempts += 1
        
        prompt = f"""以下のデータセットと同じパターンで、新しいサンプルを1件生成してください。
 
既存のサンプル例:
{examples_text}
 
## 要件
- 同じ形式（入力/出力のペア）で生成する
- 既存のサンプルと同じドメイン・スタイルを維持する
- 既存のサンプルと重複しないようにする
- 多様性を出すために、テーマやアプローチを少し変える
 
以下のJSON形式で出力してください:
{{"input": "入力テキスト", "output": "出力テキスト"}}"""
        
        try:
            response = client.models.generate_content(
                model=augmentation_model,
                contents=prompt
            )
            
            # JSON を抽出してパース
            text = response.text.strip()
            # ```json ... ``` ブロックを除去
            if "```" in text:
                text = text.split("```")[1].replace("json", "").strip()
            
            import json
            new_sample = json.loads(text)
            
            if "input" in new_sample and "output" in new_sample:
                # 重複チェック
                existing_inputs = {d["input"] for d in augmented}
                if new_sample["input"] not in existing_inputs:
                    augmented.append(new_sample)
        except Exception as e:
            print(f"拡張失敗（試行 {attempts}）: {e}")
    
    return augmented
 
# 使用例（目標200サンプルに拡張）
# augmented_data = augment_training_data(original_data, target_size=200)

ただし、生成したデータは必ず人間がレビューしてください。低品質な拡張データはむしろ性能を下げることがあります。

LoRA vs QLoRA の選択基準

Gemma 4 の fine-tuning では、フルパラメータの更新ではなく LoRA（Low-Rank Adaptation）を使うのが現実的です。

どちらを使うべきか

状況	推奨
VRAM 24GB以上	LoRA（精度優先）
VRAM 8〜24GB	QLoRA（コスト・精度のバランス）
VRAM 8GB以下	QLoRA（4bit量子化が必須）
最高精度が必要	LoRA
速度優先	QLoRA

from transformers import AutoTokenizer, AutoModelForCausalLM, BitsAndBytesConfig
from peft import LoraConfig, get_peft_model, TaskType
import torch
 
def setup_lora_model(
    model_name: str = "google/gemma-4-12b",
    use_quantization: bool = True,
    lora_rank: int = 16,
    lora_alpha: int = 32,
    target_modules: list[str] = None
) -> tuple:
    """LoRA / QLoRA でファインチューニング用のモデルをセットアップする"""
    
    tokenizer = AutoTokenizer.from_pretrained(model_name)
    tokenizer.pad_token = tokenizer.eos_token
    
    # QLoRA: 4bit 量子化を使う場合
    if use_quantization:
        quantization_config = BitsAndBytesConfig(
            load_in_4bit=True,
            bnb_4bit_compute_dtype=torch.bfloat16,
            bnb_4bit_use_double_quant=True,    # メモリ節約
            bnb_4bit_quant_type="nf4",          # 推奨フォーマット
        )
        model = AutoModelForCausalLM.from_pretrained(
            model_name,
            quantization_config=quantization_config,
            device_map="auto",
            torch_dtype=torch.bfloat16,
        )
    else:
        # LoRA: 量子化なし
        model = AutoModelForCausalLM.from_pretrained(
            model_name,
            device_map="auto",
            torch_dtype=torch.bfloat16,
        )
    
    # LoRA の設定
    if target_modules is None:
        # Gemma 4 の attention レイヤーを対象にする
        target_modules = ["q_proj", "k_proj", "v_proj", "o_proj",
                         "gate_proj", "up_proj", "down_proj"]
    
    lora_config = LoraConfig(
        task_type=TaskType.CAUSAL_LM,
        r=lora_rank,             # ランク（大きいほど表現力↑・計算量↑）
        lora_alpha=lora_alpha,   # スケーリング係数（通常は rank * 2）
        lora_dropout=0.05,       # ドロップアウト率
        target_modules=target_modules,
        bias="none",
    )
    
    model = get_peft_model(model, lora_config)
    model.print_trainable_parameters()
    # 例: trainable params: 41,943,040 || all params: 12,295,389,184 || trainable%: 0.34%
    
    return model, tokenizer
 
# lora_rank の選択ガイド:
# r=4:  最小限の適応（非常に少ないデータ、特定のスタイル変更のみ）
# r=8:  バランス型（小〜中規模のタスク適応）
# r=16: 標準設定（一般的なファインチューニング）
# r=32: 高い表現力（複雑なタスク、大規模データセット）
# r=64: 最大（フルファインチューニングに近い効果、VRAM要件大）

lora_rank の影響を理解する

# ランクとパラメータ数の関係を確認する
def estimate_lora_parameters(
    model_hidden_size: int,
    num_attention_heads: int,
    num_layers: int,
    rank: int
) -> dict:
    """LoRA パラメータ数を推定する"""
    
    # 各 attention レイヤーの LoRA パラメータ数
    # (hidden_size × rank × 2) × (num_attention_heads × 4 modules)
    params_per_layer = model_hidden_size * rank * 2 * 4
    total_params = params_per_layer * num_layers
    
    return {
        "rank": rank,
        "params_per_layer": f"{params_per_layer:,}",
        "total_lora_params": f"{total_params:,}",
        "approx_memory_mb": f"{total_params * 2 / 1024 / 1024:.1f}",  # bfloat16
    }
 
# Gemma 4 12B の場合の目安
for rank in [4, 8, 16, 32]:
    info = estimate_lora_parameters(
        model_hidden_size=5120,
        num_attention_heads=32,
        num_layers=46,
        rank=rank
    )
    print(f"rank={rank}: {info['total_lora_params']} パラメータ, ~{info['approx_memory_mb']} MB")

過学習を防ぐ学習設定

早期終了と評価の実装

from transformers import TrainingArguments, Trainer, EarlyStoppingCallback
from datasets import Dataset
 
def create_training_args(
    output_dir: str,
    num_train_epochs: int = 3,
    batch_size: int = 4,
    use_early_stopping: bool = True,
) -> TrainingArguments:
    """過学習を防ぐ学習パラメータを返す"""
    
    return TrainingArguments(
        output_dir=output_dir,
        
        # エポック数と評価
        num_train_epochs=num_train_epochs,
        eval_strategy="steps",      # ステップごとに評価
        eval_steps=100,             # 100ステップごとに評価
        save_strategy="steps",
        save_steps=100,
        
        # バッチサイズ（勾配累積でメモリを節約）
        per_device_train_batch_size=batch_size,
        per_device_eval_batch_size=batch_size,
        gradient_accumulation_steps=4,  # 実効バッチサイズ = batch_size × 4
        
        # 学習率
        learning_rate=2e-4,         # LoRA の典型的な学習率
        lr_scheduler_type="cosine", # コサインスケジューラが安定しやすい
        warmup_ratio=0.05,          # 学習率ウォームアップ（全ステップの5%）
        
        # 過学習防止
        weight_decay=0.01,          # L2 正則化
        max_grad_norm=0.3,          # 勾配クリッピング
        
        # 早期終了のための設定
        load_best_model_at_end=True if use_early_stopping else False,
        metric_for_best_model="eval_loss",
        greater_is_better=False,    # loss は小さい方が良い
        
        # メモリ効率化
        fp16=False,
        bf16=True,                  # bfloat16 を使う（CUDA 8以上）
        gradient_checkpointing=True, # メモリとスピードのトレードオフ
        
        # ログ
        logging_steps=10,
        report_to="tensorboard",
    )
 
def train_with_monitoring(
    model,
    tokenizer,
    train_dataset: Dataset,
    eval_dataset: Dataset,
    output_dir: str = "./fine-tuned-gemma4",
) -> Trainer:
    """過学習監視付きの学習を実行する"""
    
    training_args = create_training_args(output_dir)
    
    callbacks = []
    if training_args.load_best_model_at_end:
        # patience=3: eval_loss が3回連続で改善しなければ停止
        callbacks.append(EarlyStoppingCallback(early_stopping_patience=3))
    
    trainer = Trainer(
        model=model,
        args=training_args,
        train_dataset=train_dataset,
        eval_dataset=eval_dataset,
        callbacks=callbacks,
    )
    
    print("学習開始...")
    trainer.train()
    
    # 過学習の兆候を確認
    logs = trainer.state.log_history
    train_losses = [l["loss"] for l in logs if "loss" in l]
    eval_losses = [l["eval_loss"] for l in logs if "eval_loss" in l]
    
    if len(eval_losses) >= 2:
        if eval_losses[-1] > eval_losses[-2]:
            print("⚠️ 評価ロスが増加しています。過学習の可能性があります。")
            print(f"   最良のチェックポイント: {trainer.state.best_model_checkpoint}")
    
    return trainer

チェックポイントの選択

過学習を検出したら、訓練後期のチェックポイントではなく、eval_loss が最小だったチェックポイントを使います。

from peft import PeftModel
import os
import json
 
def find_best_checkpoint(checkpoint_dir: str) -> str:
    """eval_loss が最小のチェックポイントを見つける"""
    
    best_loss = float("inf")
    best_checkpoint = None
    
    for dirname in os.listdir(checkpoint_dir):
        if not dirname.startswith("checkpoint-"):
            continue
        
        checkpoint_path = os.path.join(checkpoint_dir, dirname)
        trainer_state_path = os.path.join(checkpoint_path, "trainer_state.json")
        
        if not os.path.exists(trainer_state_path):
            continue
        
        with open(trainer_state_path) as f:
            state = json.load(f)
        
        # このチェックポイントでの eval_loss を取得
        log_history = state.get("log_history", [])
        for entry in reversed(log_history):
            if "eval_loss" in entry and entry.get("step") == int(dirname.split("-")[1]):
                if entry["eval_loss"] < best_loss:
                    best_loss = entry["eval_loss"]
                    best_checkpoint = checkpoint_path
                break
    
    print(f"最良チェックポイント: {best_checkpoint}")
    print(f"eval_loss: {best_loss:.4f}")
    return best_checkpoint
 
# fine-tuned モデルのロード
def load_finetuned_model(base_model_name: str, checkpoint_path: str):
    """fine-tuned モデルをロードする"""
    from transformers import AutoModelForCausalLM, AutoTokenizer
    
    model = AutoModelForCausalLM.from_pretrained(
        base_model_name,
        device_map="auto",
        torch_dtype=torch.bfloat16
    )
    model = PeftModel.from_pretrained(model, checkpoint_path)
    tokenizer = AutoTokenizer.from_pretrained(base_model_name)
    
    return model, tokenizer

本番デプロイ前の品質確認

fine-tuning が終わったら、本番に出す前に必ず品質確認を行います。

from bert_score import score as bert_score
from rouge_score import rouge_scorer
import numpy as np
 
def evaluate_finetuned_model(
    model,
    tokenizer,
    test_data: list[dict],
    device: str = "cuda"
) -> dict:
    """fine-tuned モデルの品質を定量的に評価する"""
    
    predictions = []
    references = []
    
    for sample in test_data:
        # 推論
        inputs = tokenizer(
            sample["input"],
            return_tensors="pt",
            truncation=True,
            max_length=512
        ).to(device)
        
        with torch.no_grad():
            outputs = model.generate(
                **inputs,
                max_new_tokens=256,
                temperature=0.1,  # 評価時は低温度で決定的な出力に
                do_sample=False,
            )
        
        generated = tokenizer.decode(
            outputs[0][inputs["input_ids"].shape[1]:],
            skip_special_tokens=True
        )
        
        predictions.append(generated.strip())
        references.append(sample["output"].strip())
    
    # ROUGE スコア
    scorer = rouge_scorer.RougeScorer(["rouge1", "rouge2", "rougeL"], use_stemmer=True)
    rouge_scores = [scorer.score(ref, pred) for ref, pred in zip(references, predictions)]
    
    # BERTScore（意味的類似度）
    P, R, F1 = bert_score(predictions, references, lang="ja", device=device)
    
    return {
        "rouge1": np.mean([s["rouge1"].fmeasure for s in rouge_scores]),
        "rouge2": np.mean([s["rouge2"].fmeasure for s in rouge_scores]),
        "rougeL": np.mean([s["rougeL"].fmeasure for s in rouge_scores]),
        "bert_score_f1": F1.mean().item(),
        "num_samples": len(test_data),
    }
 
# 使用例（ベースモデルとの比較）
# base_results = evaluate_finetuned_model(base_model, tokenizer, test_data)
# ft_results = evaluate_finetuned_model(finetuned_model, tokenizer, test_data)
# improvement = {k: ft_results[k] - base_results[k] for k in ft_results if isinstance(ft_results[k], float)}
# print(f"改善量: {improvement}")

Gemma 4 の fine-tuning で陥りやすいのは「とりあえず学習させてみる」という進め方です。データ品質の確認、適切な LoRA ランクの選択、過学習の早期検出——この3つを意識して進めると、同じコストで大幅に良い結果が得られます。

まず小さなデータセット（200〜300サンプル）で一通りのパイプラインを試して、品質評価まで通しで動かしてみることをお勧めします。全体の流れを体感してから、データ量を増やして本格的な学習に進む方が、失敗した時のコストが小さく済みます。