ローカルLLM（Gemma3）をCPUでファインチューニング

sudo apt update && sudo apt upgrade -y
# Ollamaをインストール
sudo apt install -y zstd
curl -fsSL https://ollama.com/install.sh | sh
ollama --version
# Gemma3をPull
ollama pull gemma3
ollama list

ollama run gemma3 "今日の名古屋市の天気を教えてください"

# uvをインストール
curl -LsSf https://astral.sh/uv/install.sh | sh
source $HOME/.local/bin/env
echo 'eval "$(uv generate-shell-completion bash)"' >> ~/.bashrc
uv self update
# Pythonをインストール
uv python list
uv python install 3.13
uv python pin 3.13

mkdir finetuning-test
cd finetuning-test
uv init
uv venv

torchvision
pandas
torch
tensorboard
transformers
datasets
accelerate
evaluate
trl
protobuf
sentencepiece
python-dotenv
tensorflow
tensorboardx
peft

uv add -r requirements.txt

HF_TOKEN=<HuggingFaceのトークン>

import os
from huggingface_hub import login
from dotenv import load_dotenv
from datasets import Dataset
import torch
from transformers import AutoTokenizer, AutoModelForCausalLM
from transformers import pipeline
from trl import SFTConfig
from trl import SFTTrainer
import pandas as pd

load_dotenv()
hf_token = os.getenv('HF_TOKEN')

# ログイン
login(hf_token)

# ベースモデル、チェックポイント ディレクトリ、学習率を設定
BASE_MODEL = "google/gemma-3-270m-it"
CHECKPOINT_DIR = "./checkpoint/kansaiben"
LEARNING_RATE = 5e-5
FINAL_MODEL_DIR = "./gemma3-kansaiben"
TEST_CASES = ["こんにちは！", "自己紹介をお願いします。", "ありがとうございます。"]

tokenizer = AutoTokenizer.from_pretrained(BASE_MODEL)
def tokenize(examples):
    tokenized = tokenizer(
        examples["text"],
        truncation=True,
        padding=False,
        max_length=512,
        return_tensors=None
    )
    tokenized["labels"] = tokenized["input_ids"].copy()
    return tokenized

data = pd.read_csv("kansaiben.csv")

def format_text(row):
    return f"<start_of_turn>user\n{row['instruction']}<end_of_turn>\n<start_of_turn>model\n{row['output']}<end_of_turn>"

texts = data.apply(format_text, axis=1).tolist()
dataset = Dataset.from_dict({"text": texts})
dataset = dataset.map(tokenize, batched=True, remove_columns=["text"])

# TRL と SFTTrainer を使用して Gemma をファインチューニングする
model = AutoModelForCausalLM.from_pretrained(
    BASE_MODEL,
    torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32,
    attn_implementation="eager"
)

print(f"Device: {model.device}")
print(f"DType: {model.dtype}")


# ファインチューニング前の状態を確認
pipe = pipeline("text-generation", model=model, tokenizer=tokenizer)

for i, test_input in enumerate(TEST_CASES, 1):
    prompt = f"<start_of_turn>user\n{test_input}<end_of_turn>\n<start_of_turn>model\n"
    output = pipe(prompt, max_new_tokens=30, temperature=0.7, do_sample=True)
    response = output[0]['generated_text'].replace(prompt, "").split("<end_of_turn>")[0].strip()
    print(f"{i}. 入力: {test_input}")
    print(f"   応答: {response}")


# トレーニング設定
args = SFTConfig(
    output_dir="./output",
    num_train_epochs=3,
    per_device_train_batch_size=2,
    save_strategy="epoch",
    logging_steps=10,
    remove_unused_columns=False,
    report_to=[],  # トラッキングを無効にする
    dataloader_pin_memory=False,
    fp16=False,   # use float16 precision
    bf16=False,  # use bfloat16 precision
)

# カスタムデータコレクター
def data_collator(features):
    # 各特徴量からinput_idsとlabelsを取得
    input_ids = [f["input_ids"] for f in features]
    labels = [f["labels"] for f in features]
    
    # バッチ内の最大長を取得
    max_length = max(len(ids) for ids in input_ids)
    
    # パディング処理
    batch_input_ids = []
    batch_attention_mask = []
    batch_labels = []
    
    for ids, lbls in zip(input_ids, labels):
        # パディング長を計算
        pad_length = max_length - len(ids)
        
        # input_idsをパディング
        padded_ids = ids + [tokenizer.pad_token_id] * pad_length
        
        # attention_maskを作成
        attention_mask = [1] * len(ids) + [0] * pad_length
        
        # labelsをパディング（パディング部分は-100で無視）
        padded_labels = lbls + [-100] * pad_length
        
        batch_input_ids.append(padded_ids)
        batch_attention_mask.append(attention_mask)
        batch_labels.append(padded_labels)
    
    # テンソルに変換
    return {
        "input_ids": torch.tensor(batch_input_ids, dtype=torch.long),
        "attention_mask": torch.tensor(batch_attention_mask, dtype=torch.long),
        "labels": torch.tensor(batch_labels, dtype=torch.long)
    }

trainer = SFTTrainer(
    model=model,
    args=args,
    train_dataset=dataset,
    data_collator=data_collator,
    processing_class=tokenizer
)

# トレーニングを開始 (所要時間5分)
trainer.train()
trainer.save_model(FINAL_MODEL_DIR)

# モデルの推論をテストする
model = AutoModelForCausalLM.from_pretrained(
    FINAL_MODEL_DIR,
    torch_dtype="auto",
    device_map="auto",
    attn_implementation="eager"
)
tokenizer = AutoTokenizer.from_pretrained(FINAL_MODEL_DIR)

# ファインチューニング後の状態を確認
pipe = pipeline("text-generation", model=model, tokenizer=tokenizer)

for i, test_input in enumerate(TEST_CASES, 1):
    prompt = f"<start_of_turn>user\n{test_input}<end_of_turn>\n<start_of_turn>model\n"
    output = pipe(prompt, max_new_tokens=30, temperature=0.7, do_sample=True)
    response = output[0]['generated_text'].replace(prompt, "").split("<end_of_turn>")[0].strip()
    print(f"{i}. 入力: {test_input}")
    print(f"   応答: {response}")

入力: こんにちは！
応答: こんにちは！何かお手伝いできることはありますか？

入力: 自己紹介をお願いします。
応答: はい、承知いたしました。

入力: ありがとうございます。
応答: ご返信ありがとうございます。何かお手伝いできることはありますか？

入力: こんにちは！
応答: まいど！元気しとるか？

入力: 自己紹介をお願いします。
応答: わい、いのまたいっつらってるで。わいも一緒にいけるか？

入力: ありがとうございます。
応答: ええなぁ、よろしゅう頼むわ。何頼むつもりや？