Prezentowanie elastycznych klasyfikatorów bezpieczeństwa we współpracy z Gemma

1. Przegląd

To ćwiczenie w Codelabs pokazuje, jak utworzyć niestandardowy klasyfikator tekstu przy użyciu dostrajania z uwzględnieniem parametrów (PET). Zamiast dostrajać cały model, metody PET aktualizują tylko niewielką liczbę parametrów, dzięki czemu trenowanie jest stosunkowo łatwe i szybkie. Ułatwia też modelowi poznawanie nowych zachowań przy stosunkowo małej ilości danych treningowych. Metodologia została szczegółowo opisana w artykule Towards Agile Text Classifiers for Everyone, który pokazuje, jak te techniki można stosować w różnych zadaniach związanych z bezpieczeństwem i osiągać najnowocześniejsze rozwiązania na podstawie zaledwie kilkuset przykładów treningowych.

W ramach tego ćwiczenia w Codelabs korzystamy z metody PET LoRA i mniejszego modelu Gemma (gemma_instruct_2b_en), ponieważ można to robić szybciej i wydajniej. Colab omawia kroki pozyskiwania danych, formatowania ich na potrzeby LLM, trenowania wag LoRA, a następnie oceniania wyników. Ćwiczenia w programie uczą się na zbiorze danych ETHOS, czyli publicznie dostępnym zbiorze danych do wykrywania wypowiedzi szerzących nienawiść, stworzonym na podstawie komentarzy w YouTube i Reddit. Po wytrenowaniu na tylko 200 przykładach (1/4 zbioru danych) uzyskuje F1: 0,80 i ROC-AUC: 0,78 – nieco powyżej wyników SOTA uwzględnianych obecnie w tabeli wyników (w momencie tworzenia raportu, 15 lutego 2024 r.). Po wytrenowaniu na pełnych 800 przykładach, takim jak wynik F1 – 83,74, i wynik ROC-AUC – 88,17. Większe modele, takie jak gemma_instruct_7b_en, zwykle osiągają lepsze wyniki, ale koszty trenowania i wykonywania też są wyższe.

Ostrzeżenie o wyzwalaczu: w ramach tego ćwiczenia z programowania opracowano klasyfikator bezpieczeństwa do wykrywania wypowiedzi szerzących nienawiść, dlatego przykłady i ocena wyników zawierają wulgarny język.

2. Instalacja i konfiguracja

Do pobrania modelu podstawowego potrzebujesz najnowszej wersji keras (3), keras-nlp (0.8.0) i konta Kaggle.

!pip install -q -U keras-nlp
!pip install -q -U keras

Aby zalogować się w Kaggle, możesz zapisać plik danych logowania kaggle.json w katalogu ~/.kaggle/kaggle.json lub uruchomić to polecenie w środowisku Colab:

import kagglehub

kagglehub.login()

3. Wczytaj zbiór danych ETHOS

W tej sekcji wczytasz zbiór danych, na którym będzie trenowany nasz klasyfikator i wstępnie przetworzyć go w zbiorze do trenowania i testowania. Do wykrywania treści szerzących nienawiść w mediach społecznościowych wykorzystasz popularny zbiór danych ETHOS. Więcej informacji o tym, jak zbiór danych został zebrany, znajdziesz w publikacji ETHOS: an Online Encrypt Speech Detection Dataset.

import pandas as pd

gh_root = 'https://raw.githubusercontent.com'
gh_repo = 'intelligence-csd-auth-gr/Ethos-Hate-Speech-Dataset'
gh_path = 'master/ethos/ethos_data/Ethos_Dataset_Binary.csv'
data_url = f'{gh_root}/{gh_repo}/{gh_path}'

df = pd.read_csv(data_url, delimiter=';')
df['hateful'] = (df['isHate'] >= df['isHate'].median()).astype(int)

# Shuffle the dataset.
df = df.sample(frac=1, random_state=32)

# Split into train and test.
df_train, df_test = df[:800],  df[800:]

# Display a sample of the data.
df.head(5)[['hateful', 'comment']]

Zobaczysz tekst podobny do tego:

	etykieta	komentarz
0	0	You said he but still not convinced this is a ...
1	0	well, looks like its time to have another child.
2	0	What if we send every men to mars to start a n...
3	1	It doesn't matter if you're black or white, ...
4	0	Who ever disliked this video should be ashamed...

4. Pobierz model i utwórz instancję

Zgodnie z dokumentacją możesz łatwo używać modelu Gemma na wiele sposobów. W przypadku Keras trzeba wykonać następujące czynności:

import keras
import keras_nlp

# For reproducibility purposes.
keras.utils.set_random_seed(1234)

# Download the model from Kaggle using Keras.
model = keras_nlp.models.GemmaCausalLM.from_preset('gemma_instruct_2b_en')

# Set the sequence length to a small enough value to fit in memory in Colab.
model.preprocessor.sequence_length = 128

Aby sprawdzić, czy model działa, możesz wygenerować tekst:

model.generate('Question: what is the capital of France? ', max_length=32)

5. Wstępne przetwarzanie tekstu i tokeny separatora

Aby pomóc modelowi lepiej zrozumieć nasze intencje, możesz wstępnie przetworzyć tekst i użyć tokenów separatora. Dzięki temu zmniejsza się prawdopodobieństwo wygenerowania przez model tekstu, który nie pasuje do oczekiwanego formatu. Możesz na przykład spróbować zażądać klasyfikacji nastawienia od modelu, wpisując prompt w taki sposób:

Classify the following text into one of the following classes:[Positive,Negative]

Text: you look very nice today
Classification:

W takim przypadku model może zwrócić to, czego szukasz. Jeśli na przykład tekst zawiera znaki nowego wiersza, może to mieć negatywny wpływ na wydajność modelu. Bardziej solidnym rozwiązaniem jest użycie tokenów separatorów. Prompt ten będzie wyglądać tak:

Classify the following text into one of the following classes:[Positive,Negative]
<separator>
Text: you look very nice today
<separator>
Prediction:

Można to wyodrębnić za pomocą funkcji, która wstępnie przetwarza tekst:

def preprocess_text(
    text: str,
    labels: list[str],
    instructions: str,
    separator: str,
) -> str:
  prompt = f'{instructions}:[{",".join(labels)}]'
  return separator.join([prompt, f'Text:{text}', 'Prediction:'])

Jeśli uruchomisz funkcję, używając tego samego promptu i tekstu co poprzednio, otrzymasz te same dane wyjściowe:

text = 'you look very nice today'

prompt = preprocess_text(
    text=text,
    labels=['Positive', 'Negative'],
    instructions='Classify the following text into one of the following classes',
    separator='\n<separator>\n',
)

print(prompt)

Powinien wyświetlić się:

Classify the following text into one of the following classes:[Positive,Negative]
<separator>
Text:well, looks like its time to have another child
<separator>
Prediction:

6. Przetwarzanie końcowe

Wynikiem modelu są tokeny z różnymi prawdopodobieństwami. Zwykle, aby wygenerować tekst, należy wybrać jeden z kilku najbardziej prawdopodobnych tokenów i utworzyć zdania, akapity, a nawet całe dokumenty. Jednak dla celów klasyfikacji tak naprawdę liczy się to, czy model uważa, że wartość Positive jest bardziej prawdopodobna niż Negative, czy odwrotnie.

Biorąc pod uwagę utworzony wcześniej model, możesz przetworzyć jego dane wyjściowe, biorąc pod uwagę niezależne prawdopodobieństwo tego, czy następny token to Positive czy Negative:

import numpy as np


def compute_output_probability(
    model: keras_nlp.models.GemmaCausalLM,
    prompt: str,
    target_classes: list[str],
) -> dict[str, float]:
  # Shorthands.
  preprocessor = model.preprocessor
  tokenizer = preprocessor.tokenizer

  # NOTE: If a token is not found, it will be considered same as "<unk>".
  token_unk = tokenizer.token_to_id('<unk>')

  # Identify the token indices, which is the same as the ID for this tokenizer.
  token_ids = [tokenizer.token_to_id(word) for word in target_classes]

  # Throw an error if one of the classes maps to a token outside the vocabulary.
  if any(token_id == token_unk for token_id in token_ids):
    raise ValueError('One of the target classes is not in the vocabulary.')

  # Preprocess the prompt in a single batch. This is done one sample at a time
  # for illustration purposes, but it would be more efficient to batch prompts.
  preprocessed = model.preprocessor.generate_preprocess([prompt])

  # Identify output token offset.
  padding_mask = preprocessed["padding_mask"]
  token_offset = keras.ops.sum(padding_mask) - 1

  # Score outputs, extract only the next token's logits.
  vocab_logits = model.score(
      token_ids=preprocessed["token_ids"],
      padding_mask=padding_mask,
  )[0][token_offset]

  # Compute the relative probability of each of the requested tokens.
  token_logits = [vocab_logits[ix] for ix in token_ids]
  logits_tensor = keras.ops.convert_to_tensor(token_logits)
  probabilities = keras.activations.softmax(logits_tensor)

  return dict(zip(target_classes, probabilities.numpy()))

Możesz przetestować tę funkcję, uruchamiając ją za pomocą utworzonego wcześniej promptu:

compute_output_probability(
    model=model,
    prompt=prompt,
    target_classes=['Positive', 'Negative'],
)

Spowoduje to wygenerowanie kodu podobnego do tego:

{'Positive': 0.99994016, 'Negative': 5.984089e-05}

7. Tworzenie klasyfikatora

Aby ułatwić sobie korzystanie z tych funkcji, możesz umieścić wszystkie utworzone właśnie funkcje w jednym klasyfikatorze przypominającym sklearn z łatwymi w użyciu i znanymi funkcjami, takimi jak predict() i predict_score().

import dataclasses


@dataclasses.dataclass(frozen=True)
class AgileClassifier:
  """Agile classifier to be wrapped around a LLM."""

  # The classes whose probability will be predicted.
  labels: tuple[str, ...]

  # Provide default instructions and control tokens, can be overridden by user.
  instructions: str = 'Classify the following text into one of the following classes'
  separator_token: str = '<separator>'
  end_of_text_token: str = '<eos>'

  def encode_for_prediction(self, x_text: str) -> str:
    return preprocess_text(
        text=x_text,
        labels=self.labels,
        instructions=self.instructions,
        separator=self.separator_token,
    )

  def encode_for_training(self, x_text: str, y: int) -> str:
    return ''.join([
        self.encode_for_prediction(x_text),
        self.labels[y],
        self.end_of_text_token,
    ])

  def predict_score(
      self,
      model: keras_nlp.models.GemmaCausalLM,
      x_text: str,
  ) -> list[float]:
    prompt = self.encode_for_prediction(x_text)
    token_probabilities = compute_output_probability(
        model=model,
        prompt=prompt,
        target_classes=self.labels,
    )
    return [token_probabilities[token] for token in self.labels]

  def predict(
      self,
      model: keras_nlp.models.GemmaCausalLM,
      x_eval: str,
  ) -> int:
    return np.argmax(self.predict_score(model, x_eval))


agile_classifier = AgileClassifier(labels=('Positive', 'Negative'))

8. Dostrajanie modelu

(LoRA) to skrót od Low-Rank Adaptation. To technika dostrajania, która umożliwia efektywne dostrajanie dużych modeli językowych. Więcej informacji na ten temat znajdziesz w raporcie „LoRA: Low-Rank Adaptation of Large Language Models”.

Implementacja Keras języka Gemma udostępnia metodę enable_lora(), której możesz użyć do dostrajania:

# Enable LoRA for the model and set the LoRA rank to 4.
model.backbone.enable_lora(rank=4)

Po włączeniu LoRA możesz rozpocząć proces dostrajania. W Colab trwa to około 5 minut na epokę:

import tensorflow as tf

# Create dataset with preprocessed text + labels.
map_fn = lambda xy: agile_classifier.encode_for_training(*xy)
x_train = list(map(map_fn, df_train[['comment', 'hateful']].values))
ds_train = tf.data.Dataset.from_tensor_slices(x_train).batch(2)

# Compile the model using the Adam optimizer and appropriate loss function.
model.compile(
    loss=keras.losses.SparseCategoricalCrossentropy(from_logits=True),
    optimizer=keras.optimizers.Adam(learning_rate=0.0005),
    weighted_metrics=[keras.metrics.SparseCategoricalAccuracy()],
)

# Begin training.
model.fit(ds_train, epochs=4)

Trenowanie w większej liczbie epok zapewni większą dokładność, dopóki nie wystąpi nadmierne dopasowanie.

9. Sprawdzanie wyników

Możesz teraz sprawdzić dane wyjściowe wytrenowanego właśnie klasyfikatora zwinnego. Ten kod zwróci przewidywany wynik klasy dla danego fragmentu tekstu:

text = 'you look really nice today'
scores = agile_classifier.predict_score(model, text)
dict(zip(agile_classifier.labels, scores))

{'Positive': 0.99899644, 'Negative': 0.0010035498}

10. Ocena modelu

Na koniec ocenisz wydajność naszego modelu, używając 2 popularnych wskaźników: wyniku F1 i AUC-ROC. Wynik F1 wychwytuje błędy fałszywie negatywne i fałszywie pozytywne przez ocenę średniej harmonicznej precyzji i czułości przy określonym progu klasyfikacji. Z kolei AUC-ROC ukazuje kompromis między współczynnikiem wyników prawdziwie pozytywnych a współczynnikiem wyników fałszywie pozytywnych na różnych progach i oblicza pole pod tą krzywą.

from sklearn.metrics import f1_score, roc_auc_score

y_true = df_test['hateful'].values
# Compute the scores (aka probabilities) for each of the labels.
y_score = [agile_classifier.predict_score(model, x) for x in df_test['comment']]
# The label with highest score is considered the predicted class.
y_pred = np.argmax(y_score, axis=1)
# Extract the probability of a comment being considered hateful.
y_prob = [x[agile_classifier.labels.index('Negative')] for x in y_score]

# Compute F1 and AUC-ROC scores.
print(f'F1: {f1_score(y_true, y_pred):.2f}')
print(f'AUC-ROC: {roc_auc_score(y_true, y_prob):.2f}')

F1: 0.84
AUC-ROC: = 0.88

Innym ciekawym sposobem oceny prognoz modelu są tablice pomyłek. Tablica pomyłek obrazuje różne rodzaje błędów prognozy.

from sklearn.metrics import confusion_matrix, ConfusionMatrixDisplay

cm = confusion_matrix(y_true, y_pred)
ConfusionMatrixDisplay(
  confusion_matrix=cm,
  display_labels=agile_classifier.labels,
).plot()

Możesz też przyjrzeć się krzywej ROC, aby poznać potencjalne błędy w prognozach przy korzystaniu z różnych progów punktacji.

from sklearn.metrics import RocCurveDisplay, roc_curve

fpr, tpr, _ = roc_curve(y_true, y_score, pos_label=1)
RocCurveDisplay(fpr=fpr, tpr=tpr).plot()