Presentazione dei classificatori di sicurezza Agile con Gemma

Visualizza sull'IA generativa Esegui in Google Colab Visualizza il codice sorgente su GitHub Imparare nei codelab

Questo codelab illustra come creare un classificatore di testo personalizzato utilizzando la regolazione efficiente dei parametri (PET). Invece di perfezionare l'intero modello, i metodi PET aggiornati solo una piccola quantità di parametri, il che lo rende relativamente facile e veloce da addestrare. Inoltre, consente a un modello di apprendere nuovi comportamenti con relativamente pochi dati di addestramento. La metodologia è descritta in dettaglio in Towards Agile Text Classifiers for Everyone, che mostra come queste tecniche possono essere applicate a una serie di attività di sicurezza e raggiungere prestazioni all'avanguardia con solo poche centinaia di esempi di addestramento.

Questo codelab utilizza il metodo PET LoRA e il modello Gemma più piccolo (gemma_instruct_2b_en), poiché può essere eseguito più velocemente e in modo più efficiente. Il colab illustra i passaggi di importazione dei dati, di formattazione per l'LLM, di addestramento dei pesi LoRA e di valutazione dei risultati. Questo codelab viene addestrato sul set di dati ETHOS, un set di dati di dominio pubblico per il rilevamento di discorsi di incitamento all'odio, creato a partire dai commenti di YouTube e Reddit. Se viene addestrato su soli 200 esempi (1/4 del set di dati), raggiunge un F1 di 0,80 e un AUC ROC di 0,78, leggermente superiore allo stato dell'arte attualmente riportato nella classifica (al momento della stesura, 15 febbraio 2024). Se viene addestrato sull'insieme completo di 800 esempi, raggiunge un punteggio F1 di 83,74 e un punteggio ROC-AUC di 88,17. I modelli più grandi, come gemma_instruct_7b_en, generalmente hanno un rendimento migliore, ma i costi di addestramento ed esecuzione sono maggiori.

Avviso di attivazione: poiché questo codelab sviluppa un classificatore di sicurezza per il rilevamento di discorsi di incitamento all'odio, gli esempi e la valutazione dei risultati contengono un linguaggio scurrile.

Installazione e configurazione

Per questo codelab, avrai bisogno di una versione recente di keras (3), keras-nlp (0.8.0) e di un account Kaggle per scaricare un modello Gemma.

import kagglehub

kagglehub.login()

pip install -q -U keras-nlp
pip install -q -U keras
 
import os

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

Carica il set di dati ETHOS

In questa sezione caricherai il set di dati su cui addestrare il nostro classificatore e lo pre-elabori in un set di addestramento e test. Utilizzerai il popolare set di dati di ricerca ETHOS, raccolto per rilevare il discorso di incitamento all'odio sui social media. Puoi trovare maggiori informazioni su come è stato raccolto il set di dati nel documento ETHOS: an Online Hate 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']]

Scarica e crea un'istanza del modello

Come descritto nella documentazione, puoi utilizzare facilmente il modello Gemma in molti modi. Con Keras, devi fare quanto segue:

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)

Preelaborazione del testo e token separatore

Per aiutare il modello a comprendere meglio la nostra intenzione, puoi pre-elaborare il testo e utilizzare token separatore. Di conseguenza, è meno probabile che il modello generi testo che non si adatti al formato previsto. Ad esempio, potresti provare a richiedere al modello una classificazione del sentiment scrivendo un prompt come questo:

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

Text: you look very nice today
Classification:

In questo caso, il modello potrebbe o meno restituire ciò che stai cercando. Ad esempio, se il testo contiene caratteri di a capo, è probabile che abbia un effetto negativo sul rendimento del modello. Un approccio più efficace è utilizzare i token di separatore. Il prompt diventa quindi:

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

Questo può essere astratto utilizzando una funzione che esegue la preelaborazione del testo:

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

Ora, se esegui la funzione utilizzando lo stesso prompt e lo stesso testo di prima, dovresti ottenere lo stesso output:

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:

Post-elaborazione dell'output

Gli output del modello sono token con varie probabilità. Normalmente, per generare testo, selezioni tra i primi token più probabili e costruisci frasi, paragrafi o persino documenti completi. Tuttavia, ai fini della classificazione, ciò che conta è se il modello ritiene che Positive sia più probabile di Negative o viceversa.

Dato il modello che hai creato in precedenza, ecco come puoi elaborare il relativo output nelle probabilità indipendenti che il token successivo sia Positive o Negative, rispettivamente:

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

Puoi testare la funzione eseguendola con il prompt creato in precedenza:

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

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

Raggruppare tutto come classificatore

Per semplicità d'uso, puoi racchiudere tutte le funzioni appena create in un singolo classificatore simile a sklearn con funzioni facili da usare e familiari come predict() e 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'))

Ottimizzazione del modello

LoRA sta per Low-Rank Adaptation. Si tratta di una tecnica di ottimizzazione che può essere impiegata per ottimizzare in modo efficiente i modelli linguistici di grandi dimensioni. Puoi scoprire di più su questo argomento nel documento LoRA: Low-Rank Adaptation of Large Language Models.

L'implementazione di Keras di Gemma fornisce un metodo enable_lora() che puoi utilizzare per la messa a punto:

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

Dopo aver attivato LoRa, puoi avviare la procedura di messa a punto. L'operazione richiede circa 5 minuti per epoca su 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>

L'addestramento per più epoche comporterà una maggiore accuratezza, fino a quando non si verifica il fenomeno di overfitting.

Controllare i risultati

Ora puoi esaminare l'output del classificatore agile appena addestrato. Questo codice produrrà il punteggio della classe prevista in base a un testo:

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}

valutazione del modello

Infine, valuterai le prestazioni del nostro modello utilizzando due metriche comuni: il punteggio F1 e l'AUC-ROC. Il punteggio F1 rileva gli errori di falso negativo e falso positivo valutando la media armonica della precisione e del richiamo a una determinata soglia di classificazione. L'AUC-ROC, invece, cattura il compromesso tra la percentuale di veri positivi e la percentuale di falsi positivi in una serie di soglie e calcola l'area sotto questa curva.

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

Un altro modo interessante per valutare le previsioni del modello sono le matrici di confusione. Una matrice di confusione mostra visivamente i diversi tipi di errori di previsione.

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

Infine, puoi anche esaminare la curva ROC per avere un'idea dei potenziali errori di previsione con l'utilizzo di soglie di punteggio diverse.

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

Appendice

Abbiamo effettuato alcune esplorazioni di base dello spazio degli iperparametri per comprendere meglio la relazione tra le dimensioni del set di dati e il rendimento. Vedi il seguente grafico.

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