📚

文書分類の実験

に公開

Raspberry Pi 400上のollamaでGemma 3 270Mを動かすでGemma 3 270MをRaspberry Pi 400に実装したので、Gemma 3 270Mで文章を分類するファインチューンが出来ないかと考えました。想定したタスクは以下のものです。github.com/mahdikabootari/Software-Requirements-Classification の中にある。PROMISE.csvのclassを予測する分類モデルができないかという実験を行いました。これは要件定義の分類をまとめたもので機能要件・非機能要件などの分類が入っています。

結論から言うと、AutoModelForSequenceClassificationを直接利用することはできないという結論になります。なぜならば、Gemma 3 270Mがデコーダオンリーモデルだからということになります。ならばと、AutoModelForCausalLMに無理矢理、分類ヘッドをつけてみたところ、同じ分類が返ってくるという結果になりました。

主な原因としては次のものが考えられます。

  1. タスクの与え方のミスマッチ: モデルは「分類せよ」というタスクを理解できず、学習データの中で最も頻出するラベルを返す、あるいは学習の初期段階で偶然うまく損失が下がった安直な答えを返し続ける、という状態に陥っている可能性があります。

  2. データセットと学習方法: PROMISEデータセットは高品質ですが、LLMのファインチューニングの観点からは比較的小規模です。少量のデータで学習させると、モデルはタスクの本質を捉えきれず、過学習を起こしたり、不安定な状態に陥りやすくなります。

従って、この考え方はうまくいかず、分類を生成するという方向でファインチューンしないとうまくいかなそうです。

from huggingface_hub import login
login(new_session=False)

!pip install transformers
!pip install datasets
!pip install peft
!pip install pandas
!pip install -U bitsandbytes

import pandas as pd
import torch
from transformers import AutoTokenizer, AutoModelForCausalLM, AutoModelForSequenceClassification, TrainingArguments, Trainer, pipeline, IntervalStrategy, BitsAndBytesConfig

from transformers.trainer_utils import is_main_process, set_seed, SaveStrategy
from datasets import Dataset, ClassLabel, Value
from peft import LoraConfig, get_peft_model, TaskType
from sklearn.model_selection import train_test_split
from torch import nn
import warnings

data_file = "/content/drive/MyDrive/Data/PROMISE.csv"
df = pd.read_csv(data_file)
df.head()

class_names = df['_class_'].unique().tolist()
num_classes = len(class_names)

MODEL_NAME = "google/gemma-3-270m-it"
FINAL_MODEL_DIR = "/content/drive/MyDrive/Models/TextAnalysis/gemma3-softreq-ex001"
LORA_CONFIG = {
    "r": 8,
    "lora_alpha": 32,
    "target_modules": ["q_proj", "v_proj", "o_proj", "down_proj", "up_proj", "gate_proj"],
    "lora_dropout": 0.05,
    "bias": "none",
    "task_type": TaskType.CAUSAL_LM # CausalLMモデルにLoRAを適用するため
}

import torch
from transformers import AutoTokenizer, AutoModelForCausalLM, BitsAndBytesConfig

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)

# Add padding token
tokenizer.add_special_tokens({'pad_token': '[PAD]'})

# Configure quantization
bnb_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",
    bnb_4bit_compute_dtype=torch.bfloat16,
    bnb_4bit_use_double_quant=False,
)

# Load the base model with quantization
model = AutoModelForCausalLM.from_pretrained(
    MODEL_NAME,
    quantization_config=bnb_config,
    torch_dtype=torch.bfloat16, # Use bfloat16 with 4-bit quantization
    low_cpu_mem_usage=True,
    device_map={"":0}, # Load model on the GPU
    output_hidden_states=True # Explicitly request hidden states
)

# Resize token embeddings to account for the new padding token
model.resize_token_embeddings(len(tokenizer))
model.config.pad_token_id = tokenizer.pad_token_id

# Explicitly set pad_token_id in the model config
model.config.pad_token_id = tokenizer.pad_token_id


# Add a classification head on top of the base model
model.config.num_labels = num_classes
model.classifier = torch.nn.Linear(model.config.hidden_size, num_classes)

# Cast the classification head to bfloat16
model.classifier = model.classifier.to(torch.bfloat16)


model = model.to(device)

from transformers import AutoTokenizer, DataCollatorWithPadding, TrainingArguments, Trainer, IntervalStrategy
from datasets import Dataset, ClassLabel, Value
from peft import LoraConfig, get_peft_model, TaskType
from sklearn.model_selection import train_test_split
import torch
from torch import nn

# Load the tokenizer
tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
tokenizer.pad_token = tokenizer.eos_token # Set pad token for data collator

# Create a mapping from class names to integers
class_names = df['_class_'].unique().tolist()
class_to_int = {name: i for i, name in enumerate(class_names)}
num_classes = len(class_names)

# Convert pandas DataFrame to Dataset
dataset = Dataset.from_pandas(df)

# Cast the _class_ column to ClassLabel
dataset = dataset.cast_column("_class_", ClassLabel(names=class_names))

# Split dataset into training and validation sets with stratification
train_test_split_ratio = 0.2
dataset = dataset.train_test_split(test_size=train_test_split_ratio, stratify_by_column='_class_')


# Tokenize the input text and encode the labels
def tokenize_function(examples):
    tokenized_text = tokenizer(examples["RequirementText"], truncation=True, padding="max_length", max_length=128, return_attention_mask=True)
    tokenized_text["labels"] = examples["_class_"] # Labels are already integers after casting to ClassLabel
    return tokenized_text

tokenized_datasets = dataset.map(tokenize_function, batched=True)

train_dataset = tokenized_datasets["train"]
eval_dataset = tokenized_datasets["test"]

print("Tokenized training dataset:", train_dataset)
print("Tokenized validation dataset:", eval_dataset)
print("Example from tokenized training dataset:", train_dataset[0])


# Apply LoRA to the base model
lora_config = LoraConfig(**LORA_CONFIG)
model = get_peft_model(model, lora_config)


training_args = TrainingArguments(
    output_dir="./results",
    eval_strategy=IntervalStrategy.EPOCH,
    save_strategy=SaveStrategy.EPOCH,
    learning_rate=2e-4,
    per_device_train_batch_size=16,
    per_device_eval_batch_size=16,
    num_train_epochs=10,
    weight_decay=0.01,
    save_total_limit=1,
    load_best_model_at_end=True,
    metric_for_best_model="accuracy",
    prediction_loss_only=False, # Set to False to get predictions
    logging_steps=10,
    report_to="wandb",
)

def compute_metrics(eval_pred):
    from sklearn.metrics import accuracy_score, precision_recall_fscore_support
    # Access predictions and labels directly from the EvalPrediction object
    logits = eval_pred.predictions
    labels = eval_pred.label_ids

    # ロジットをargmaxして予測を計算
    predictions = logits.argmax(axis=-1)

    # 精度、リコール、F1スコアを計算
    accuracy = accuracy_score(labels, predictions)
    precision, recall, f1, _ = precision_recall_fscore_support(labels, predictions, average='weighted', zero_division=0)

    return {
        'accuracy': accuracy,
        'precision': precision,
        'recall': recall,
        'f1': f1,
    }

# Define the data collator
data_collator = DataCollatorWithPadding(tokenizer=tokenizer)

# Define a custom Trainer with a modified compute_loss
class CustomTrainer(Trainer):
    def compute_loss(self, model, inputs, return_outputs=False, num_items_in_batch=None):
        labels = inputs.pop("labels") # Pop labels before passing inputs to the model
        outputs = model(**inputs)

        # Extract the last hidden state and pass it through the classification head
        # Assuming the model's output is a CausalLMOutput or similar
        last_hidden_state = outputs.hidden_states[-1] if outputs.hidden_states else outputs.last_hidden_state
        # Get the hidden state corresponding to the last token (before padding)
        sequence_lengths = inputs["attention_mask"].sum(dim=1) - 1
        pooled_output = last_hidden_state[torch.arange(last_hidden_state.size(0), device=last_hidden_state.device), sequence_lengths]

        classification_logits = model.classifier(pooled_output)


        loss_fct = torch.nn.CrossEntropyLoss()
        loss = loss_fct(classification_logits, labels)

        return (loss, outputs) if return_outputs else loss

    def prediction_step(self, model, inputs, prediction_loss_only, ignore_keys=None):
        labels = inputs.pop("labels")
        model.eval()

        with torch.no_grad():
            outputs = model(**inputs)

        # outputsからロジットを取得するロジックを再構築
        last_hidden_state = outputs.hidden_states[-1]
        sequence_lengths = inputs["attention_mask"].sum(dim=1) - 1
        pooled_output = last_hidden_state[torch.arange(last_hidden_state.size(0), device=last_hidden_state.device), sequence_lengths]
        logits = model.classifier(pooled_output)

        loss = None
        if labels is not None:
            loss_fct = torch.nn.CrossEntropyLoss()
            loss = loss_fct(logits.view(-1, model.classifier.out_features), labels.view(-1))

        if prediction_loss_only:
            return (loss, logits, labels)

        # ロジットとラベルをタプルとして返す
        return (loss, logits, labels)


trainer = CustomTrainer(
    model=model, # Use the PEFT model
    args=training_args,
    train_dataset=train_dataset,
    eval_dataset=eval_dataset,
    compute_metrics=compute_metrics,
    data_collator=data_collator,
)

warnings.filterwarnings("ignore")
trainer.train()
trainer.save_model(FINAL_MODEL_DIR)

# 分類ヘッドの重みを別途保存する
torch.save(model.classifier.state_dict(), f"{FINAL_MODEL_DIR}/classifier.pt")
print(f"Classifier head weights saved to {FINAL_MODEL_DIR}/classifier.pt")
!pip install peft
!pip install transformers
!pip install accelerate
!pip install bitsandbytes
!pip install evaluate
!pip install rouge_score
!pip install tensorboard
!pip install wandb
!pip install huggingface_hub

import os
from peft import PeftModel
from transformers import AutoTokenizer, AutoModelForCausalLM, BitsAndBytesConfig, pipeline
import pandas as pd
import torch
from google.colab import drive
from huggingface_hub import login
import torch.nn as nn

login(new_session=False)

drive.mount('/content/drive')

FINAL_MODEL_DIR = "/content/drive/MyDrive/Models/TextAnalysis/gemma3-softreq-ex001"

# データフレームをロードしてクラス名を取得
data_file = "/content/drive/MyDrive/Data/PROMISE.csv"
df = pd.read_csv(data_file)

# データセットからユニークなクラス名を抽出し、num_classesを取得
class_names = df['_class_'].unique().tolist()
num_classes = len(class_names)
print(f"Detected {num_classes} classes from the dataset: {class_names}")

# 4bit量子化設定の再構築
bnb_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",
    bnb_4bit_compute_dtype=torch.bfloat16,
    bnb_4bit_use_double_quant=False,
)

# ベースモデルのロード
base_model = AutoModelForCausalLM.from_pretrained(
    "google/gemma-3-270m-it",
    quantization_config=bnb_config,
    torch_dtype=torch.bfloat16,
    device_map={"": 0}
)

# トークナイザーのロード
tokenizer = AutoTokenizer.from_pretrained("google/gemma-3-270m-it")
tokenizer.pad_token = tokenizer.eos_token

# LoRAアダプターをロード
model = PeftModel.from_pretrained(base_model, FINAL_MODEL_DIR, ignore_mismatched_sizes=True)

# LoRA適用済みモデルに分類ヘッドを追加
# ここで一度だけ分類ヘッドを定義する
model.classifier = nn.Linear(model.config.hidden_size, num_classes).to("cuda").to(torch.bfloat16)

# 保存した分類ヘッドの重みをロード
classifier_path = os.path.join(FINAL_MODEL_DIR, "classifier.pt")
if os.path.exists(classifier_path):
    model.classifier.load_state_dict(torch.load(classifier_path))
    print("Classifier weights loaded successfully.")
else:
    print(f"Classifier weights not found at {classifier_path}. Initializing with random weights.")

# モデルを推論モードに設定
model.eval()

# 推論用パイプラインの準備
# ここでは、分類タスクなので、テキスト生成ではなく分類のロジックが必要です。
# このままだとモデルの出力がCausalLMのロジットになるため、直接パイプラインには使えません。

# パイプラインを使う代わりに、手動で予測を実行します。

# 元データをDataFrameに読み込み
data_file = "/content/drive/MyDrive/Data/PROMISE.csv"
df = pd.read_csv(data_file)
df = df.sample(n=50, random_state=42) # 全データは重いので50件に絞る

# トークナイザーの設定
tokenizer.pad_token = tokenizer.eos_token

# 予測結果を格納するリスト
predictions = []

# データフレームの各行をループ
for index, row in df.iterrows():
    text = row["RequirementText"]

    # テキストをトークン化してテンソルに変換
    inputs = tokenizer(
        text,
        return_tensors="pt",
        truncation=True,
        padding="max_length",
        max_length=128
    ).to("cuda") # GPUに転送

    # 推論を実行
    with torch.no_grad():
        outputs = model(**inputs, output_hidden_states=True)

    # 最後のトークンの隠れ状態を取得
    last_hidden_state = outputs.hidden_states[-1]
    sequence_lengths = inputs["attention_mask"].sum(dim=1) - 1
    pooled_output = last_hidden_state[torch.arange(last_hidden_state.size(0), device=last_hidden_state.device), sequence_lengths]

    # 分類ヘッドでロジットを計算
    # model.classifier を使用するように変更
    logits = model.classifier(pooled_output)

    # 最も確率の高いクラスのインデックスを取得
    predicted_class_id = torch.argmax(logits, dim=1).item()

    # クラス名に変換
    predicted_class_name = class_names[predicted_class_id]

    # 結果をリストに追加
    predictions.append({
        "RequirementText": text,
        "ActualClass": row["_class_"],
        "PredictedClass": predicted_class_name
    })

# 結果をDataFrameに変換して表示
results_df = pd.DataFrame(predictions)
print(results_df)

Train/lossを見る限りだと、学習しているように見えますが、実際に、推定してみると。

Train/loss

結果としては以下のようになり、まったく、よろしくありません。明らかに、同じものを返すだけになっています。

          precision    recall  f1-score   support

       A       0.00      0.00      0.00         2
       F       0.48      1.00      0.65        24
      FT       0.00      0.00      0.00         1
       L       0.00      0.00      0.00         1
      LF       0.00      0.00      0.00         2
      MN       0.00      0.00      0.00         2
       O       0.00      0.00      0.00         2
      PE       0.00      0.00      0.00         5
      PO       0.00      0.00      0.00         1
      SE       0.00      0.00      0.00         5
      US       0.00      0.00      0.00         5

accuracy                           0.48        50
macro avg       0.04      0.09      0.06        50
weighted avg       0.23      0.48      0.31        50

Discussion