Shfaqja e klasifikuesve të sigurisë së shkathët me Gemma

Kjo laborator kodesh ilustron se si të krijohet një klasifikues teksti i personalizuar duke përdorur akordimin efikas të parametrave (PET). Në vend që të rregullojnë të gjithë modelin, metodat PET përditësojnë vetëm një sasi të vogël parametrash, gjë që e bën trajnimin relativisht të lehtë dhe të shpejtë. Gjithashtu e bën më të lehtë për një model të mësojë sjellje të reja me relativisht pak të dhëna trajnimi. Metodologjia përshkruhet në detaje në Drejt Klasifikuesit e Tekstit të shkathët për të gjithë, i cili tregon se si këto teknika mund të zbatohen në një sërë detyrash sigurie dhe të arrijnë performancën më të fundit me vetëm disa qindra shembuj trajnimi.

Ky laborator i kodeve përdor metodën LoRA PET dhe modelin më të vogël Gemma ( gemma_instruct_2b_en ) pasi që mund të ekzekutohet më shpejt dhe me efikasitet. Kolabi mbulon hapat e marrjes së të dhënave, formatimit të tyre për LLM, trajnimit të peshave LoRA dhe më pas vlerësimit të rezultateve. Ky laborator kodesh trajnohet në grupin e të dhënave ETHOS , një grup të dhënash i disponueshëm publikisht për zbulimin e gjuhës së urrejtjes, i krijuar nga komentet e YouTube dhe Reddit. Kur stërvitet në vetëm 200 shembuj (1/4 e grupit të të dhënave), ai arrin F1: 0,80 dhe ROC-AUC: 0,78, pak më lart se SOTA e raportuar aktualisht në tabelën e drejtuesve (në kohën e shkrimit, 15 shkurt 2024). Kur stërviteni në plot 800 shembuj, si ai arrin një rezultat F1 prej 83.74 dhe një rezultat ROC-AUC prej 88.17. Modelet më të mëdha, si gemma_instruct_7b_en në përgjithësi do të performojnë më mirë, por kostot e trajnimit dhe ekzekutimit janë gjithashtu më të mëdha.

Paralajmërim për shkaktimin : për shkak se ky laborator kodesh zhvillon një klasifikues sigurie për zbulimin e gjuhës së urrejtjes, shembujt dhe vlerësimi i rezultateve përmbajnë një gjuhë të tmerrshme.

Instalimi dhe konfigurimi

Për këtë laborator kodesh, do t'ju duhet një version i fundit keras (3), keras-nlp (0.8.0) dhe një llogari Kaggle për të shkarkuar një model Gemma.

import kagglehub

kagglehub.login()
pip install -q -U keras-nlp
pip install -q -U keras
import os

os.environ["KERAS_BACKEND"] = "tensorflow"

Ngarko grupin e të dhënave ETHOS

Në këtë seksion ju do të ngarkoni grupin e të dhënave mbi të cilin do të trajnoni klasifikuesin tonë dhe do ta përpunoni atë paraprakisht në një grup treni dhe testimi. Ju do të përdorni grupin e të dhënave të kërkimit popullor ETHOS i cili u mblodh për të zbuluar gjuhën e urrejtjes në mediat sociale. Mund të gjeni më shumë informacion se si u mblodh grupi i të dhënave në punimin ETHOS: një grup i të dhënave për zbulimin e gjuhës së urrejtjes në internet .

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']]

Shkarkoni dhe instantoni modelin

Siç përshkruhet në dokumentacion , ju mund ta përdorni lehtësisht modelin Gemma në shumë mënyra. Me Keras, kjo është ajo që duhet të bëni:

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
model.generate('Question: what is the capital of France? ', max_length=32)

Parapërpunimi i tekstit dhe shenjat ndarëse

Për ta ndihmuar modelin të kuptojë më mirë qëllimin tonë, mund ta përpunoni paraprakisht tekstin dhe të përdorni shenja ndarëse. Kjo i bën më pak gjasa që modeli të gjenerojë tekst që nuk i përshtatet formatit të pritur. Për shembull, mund të përpiqeni të kërkoni një klasifikim të ndjenjave nga modeli duke shkruar një kërkesë si kjo:

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

Text: you look very nice today
Classification:

Në këtë rast, modeli mund të nxjerrë ose jo atë që ju kërkoni. Për shembull, nëse teksti përmban karaktere të linjës së re, ka të ngjarë të ketë një efekt negativ në performancën e modelit. Një qasje më e fortë është përdorimi i shenjave ndarëse. Prompti atëherë bëhet:

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

Kjo mund të abstragohet duke përdorur një funksion që përpunon paraprakisht tekstin:

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:'])

Tani, nëse e ekzekutoni funksionin duke përdorur të njëjtën kërkesë dhe tekst si më parë, duhet të merrni të njëjtin rezultat:

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)
Classify the following text into one of the following classes:[Positive,Negative]
<separator>
Text:you look very nice today
<separator>
Prediction:

Postprocessing output

Rezultatet e modelit janë shenja me probabilitete të ndryshme. Normalisht, për të gjeneruar tekst, ju do të zgjidhni midis disa shenjave më të mundshme dhe do të ndërtoni fjali, paragrafë apo edhe dokumente të plota. Megjithatë, për qëllimin e klasifikimit, ajo që në të vërtetë ka rëndësi është nëse modeli beson se Positive është më e mundshme se Negative apo anasjelltas.

Duke pasur parasysh modelin që keni instancuar më herët, kjo është mënyra se si mund ta përpunoni prodhimin e tij në probabilitetet e pavarura nëse tokeni tjetër është Positive ose Negative , përkatësisht:

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()))

Ju mund ta provoni atë funksion duke e ekzekutuar me një kërkesë që keni krijuar më herët:

compute_output_probability(
    model=model,
    prompt=prompt,
    target_classes=['Positive', 'Negative'],
)
{'Positive': 0.99994016, 'Negative': 5.984089e-05}

Duke i mbështjellë të gjitha si një Klasifikues

Për lehtësinë e përdorimit, ju mund t'i mbështillni të gjitha funksionet që sapo keni krijuar në një klasifikues të vetëm të ngjashëm me sklearn me funksione të lehta për t'u përdorur dhe të njohura si predict() dhe 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'))

Modeli Fine-Tuning

LoRA do të thotë "Përshtatje me gradë të ulët". Është një teknikë akordimi i imët që mund të përdoret për të rregulluar në mënyrë efikase modelet e mëdha të gjuhës. Mund të lexoni më shumë rreth tij në punimin LoRA: Adaptimi i Rangut të Ulët të Modeleve të Mëdha të Gjuhës .

Implementimi Keras i Gemma ofron një metodë enable_lora() që mund ta përdorni për rregullim të imët:

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

Pasi të aktivizoni LoRA, mund të filloni procesin e rregullimit të imët. Kjo kërkon afërsisht 5 minuta për epokë në Colab:

import tensorflow as tf

# Create dataset with preprocessed text + labels.
map_fn = lambda x: agile_classifier.encode_for_training(*x)
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)
Epoch 1/4
400/400 ━━━━━━━━━━━━━━━━━━━━ 354s 703ms/step - loss: 1.1365 - sparse_categorical_accuracy: 0.5874
Epoch 2/4
400/400 ━━━━━━━━━━━━━━━━━━━━ 338s 716ms/step - loss: 0.7579 - sparse_categorical_accuracy: 0.6662
Epoch 3/4
400/400 ━━━━━━━━━━━━━━━━━━━━ 324s 721ms/step - loss: 0.6818 - sparse_categorical_accuracy: 0.6894
Epoch 4/4
400/400 ━━━━━━━━━━━━━━━━━━━━ 323s 725ms/step - loss: 0.5922 - sparse_categorical_accuracy: 0.7220
<keras.src.callbacks.history.History at 0x7eb7e369c490>

Stërvitja për më shumë epoka do të rezultojë në saktësi më të lartë, derisa të ndodhë mbipërshtatja.

Inspektoni Rezultatet

Tani mund të inspektoni daljen e klasifikuesit të shkathët që sapo keni trajnuar. Ky kod do të nxjerrë rezultatin e parashikuar të klasës duke pasur parasysh një pjesë të tekstit:

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}

Vlerësimi i modelit

Së fundi, ju do të vlerësoni performancën e modelit tonë duke përdorur dy metrikë të zakonshëm, rezultatin F1 dhe AUC-ROC . Rezultati F1 kap gabimet false negative dhe false pozitive duke vlerësuar mesataren harmonike të saktësisë dhe rikujtimit në një prag të caktuar klasifikimi. AUC-ROC nga ana tjetër kap shkëmbimin midis normës së vërtetë pozitive dhe normës false pozitive përgjatë një sërë pragjesh dhe llogarit zonën nën këtë kurbë.

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]
from sklearn.metrics import f1_score, roc_auc_score

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

Një mënyrë tjetër interesante për të vlerësuar parashikimet e modelit janë matricat e konfuzionit. Një matricë konfuzioni do të përshkruajë vizualisht llojet e ndryshme të gabimeve të parashikimit.

from sklearn.metrics import confusion_matrix, ConfusionMatrixDisplay

cm = confusion_matrix(y_true, y_pred)
ConfusionMatrixDisplay(
    confusion_matrix=cm,
    display_labels=agile_classifier.labels,
).plot()
<sklearn.metrics._plot.confusion_matrix.ConfusionMatrixDisplay at 0x7eb7e2d29ab0>

png

Më në fund, mund të shikoni edhe kurbën ROC për të marrë një ndjenjë të gabimeve të mundshme të parashikimit me përdorimin e pragjeve të ndryshme të pikësimit.

from sklearn.metrics import RocCurveDisplay, roc_curve

fpr, tpr, _ = roc_curve(y_true, y_prob, pos_label=1)
RocCurveDisplay(fpr=fpr, tpr=tpr).plot()
<sklearn.metrics._plot.roc_curve.RocCurveDisplay at 0x7eb4d130ef20>

png

Shtojca

Ne kemi bërë disa eksplorime bazë të hapësirës së hiper-parametrit për të ndihmuar në marrjen e një kuptimi më të mirë të marrëdhënies midis madhësisë së të dhënave dhe performancës. Shihni komplotin e mëposhtëm.

import matplotlib.pyplot as plt
import pandas as pd
import seaborn as sns

sns.set_theme(style="whitegrid")

results_f1 = pd.DataFrame([
    {'training_size': 800, 'epoch': 4, 'metric': 'f1', 'score': 0.84},
    {'training_size': 800, 'epoch': 6, 'metric': 'f1', 'score': 0.83},
    {'training_size': 800, 'epoch': 8, 'metric': 'f1', 'score': 0.83},
    {'training_size': 800, 'epoch': 10, 'metric': 'f1', 'score': 0.84},
    {'training_size': 400, 'epoch': 4, 'metric': 'f1', 'score': 0.77},
    {'training_size': 400, 'epoch': 6, 'metric': 'f1', 'score': 0.80},
    {'training_size': 400, 'epoch': 8, 'metric': 'f1', 'score': 0.80},
    {'training_size': 400, 'epoch': 10,'metric': 'f1', 'score': 0.81},
    {'training_size': 200, 'epoch': 4, 'metric': 'f1', 'score': 0.78},
    {'training_size': 200, 'epoch': 6, 'metric': 'f1', 'score': 0.80},
    {'training_size': 200, 'epoch': 8, 'metric': 'f1', 'score': 0.78},
    {'training_size': 200, 'epoch': 10, 'metric': 'f1', 'score': 0.79},
])

results_roc_auc = pd.DataFrame([
    {'training_size': 800, 'epoch': 4, 'metric': 'roc-auc', 'score': 0.88},
    {'training_size': 800, 'epoch': 6, 'metric': 'roc-auc', 'score': 0.86},
    {'training_size': 800, 'epoch': 8, 'metric': 'roc-auc', 'score': 0.84},
    {'training_size': 800, 'epoch': 10, 'metric': 'roc-auc', 'score': 0.87},
    {'training_size': 400, 'epoch': 4, 'metric': 'roc-auc', 'score': 0.83},
    {'training_size': 400, 'epoch': 6, 'metric': 'roc-auc', 'score': 0.82},
    {'training_size': 400, 'epoch': 8, 'metric': 'roc-auc', 'score': 0.82},
    {'training_size': 400, 'epoch': 10,'metric': 'roc-auc', 'score': 0.85},
    {'training_size': 200, 'epoch': 4, 'metric': 'roc-auc', 'score': 0.79},
    {'training_size': 200, 'epoch': 6, 'metric': 'roc-auc', 'score': 0.78},
    {'training_size': 200, 'epoch': 8, 'metric': 'roc-auc', 'score': 0.80},
    {'training_size': 200, 'epoch': 10, 'metric': 'roc-auc', 'score': 0.81},
])


plot_opts = dict(style='.-', ylim=(0.7, 0.9))
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(14, 4))
process_results_df = lambda df: df.set_index('epoch').groupby('training_size')['score']
process_results_df(results_f1).plot(title='Metric: F1', ax=ax1, **plot_opts)
process_results_df(results_roc_auc).plot(title='Metric: ROC-AUC', ax=ax2, **plot_opts)
fig.show()

png