מציגים מסווגי בטיחות גמישים עם Gemma

1. סקירה כללית

ה-Codelab הזה מדגים איך ליצור מסווג טקסט בהתאמה אישית באמצעות כוונון יעיל בפרמטרים (PET). במקום לכוונן את המודל כולו, שיטות PET מעדכנות רק כמות קטנה של פרמטרים, ולכן האימון קל ומהיר יחסית. בנוסף, קל יותר למודל ללמוד התנהגויות חדשות עם מעט נתוני אימון. המתודולוגיה מתוארת בפירוט במאמר Towards Agile Text Classifiers for כולם, שבו אפשר להשתמש בשיטות האלה במגוון משימות בטיחות ולהשיג ביצועים ברמה גבוהה באמצעות כמה מאות דוגמאות אימון בלבד.

ה-Codelab הזה משתמש בשיטת PET LoRA ובמודל Gemma הקטן יותר (gemma_instruct_2b_en), כי ניתן להריץ אותה מהר יותר וביעילות רבה יותר. Colab כולל את שלבי הטמעת הנתונים, עיצובם ל-LLM, אימון משקולות LoRA ולאחר מכן הערכת התוצאות. האימון של ה-Codelab הזה מתבצע באמצעות מערך הנתונים ETHOS, מערך נתונים שזמין באופן ציבורי לזיהוי דברי שטנה. המידע נוצר על סמך תגובות ב-YouTube וב-Reddit. כשמאמנים את המודל על 200 דוגמאות בלבד (1/4 ממערך הנתונים), הוא מניב F1: 0.80 ו-ROC-AUC: 0.78, קצת מעל ל-SOTA שמדווח כרגע ב-Leaderboard (בזמן הכתיבה, 15 בפברואר 2024). כשמאמנים את המודל על 800 הדוגמאות המלאות, למשל, הציון F1 הוא 83.74, והציון ROC-AUC הוא 88.17. מודלים גדולים יותר, כמו gemma_instruct_7b_en בדרך כלל יניבו ביצועים טובים יותר, אבל גם עלויות האימון והביצוע גבוהות יותר.

אזהרת טריגר: כי ה-Codelab הזה מפתח מסווג בטיחות לזיהוי דברי שטנה, דוגמאות והערכה של התוצאות שמכילות שפה נוראית.

2. התקנה והגדרה

ל-Codelab הזה יש צורך בגרסה עדכנית של keras (3), keras-nlp (0.8.0) וחשבון Kaggle כדי להוריד את המודל הבסיסי.

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

כדי להתחבר ל-Kaggle, אפשר לאחסן את קובץ פרטי הכניסה של kaggle.json ב-~/.kaggle/kaggle.json או להריץ את הפקודה הבאה בסביבת Colab:

import kagglehub

kagglehub.login()

3. טעינת מערך הנתונים ETHOS

בקטע הזה תטען את מערך הנתונים שלפיו אפשר לאמן את המסווג שלנו ותעבד אותו מראש לקבוצת אימון ובדיקה. אתם תשתמשו במערך הנתונים הפופולרי ETHOS, שנאסף כדי לזהות דברי שטנה ברשתות החברתיות. אפשר למצוא מידע נוסף על אופן האיסוף של מערך הנתונים במאמר ETHOS: מערך נתונים לזיהוי דברי שטנה באינטרנט.

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

בשלב הזה תראו משהו שדומה לזה:

	תווית	תגובה
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. הורדת המודל ויצירת מופע שלו

כמו שמתואר במסמכי התיעוד, אפשר להשתמש בקלות במודל Gemma בדרכים רבות. עם Keras, זה מה שצריך לעשות:

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)

5. אסימונים לעיבוד מראש של טקסט ומפרידים

כדי לעזור למודל להבין טוב יותר את הכוונה שלנו, אפשר לעבד מראש את הטקסט ולהשתמש באסימונים מפרידים. כך יהיה פחות סיכוי שהמודל ייצור טקסט שלא מתאים לפורמט הצפוי. לדוגמה, כדי לבקש סיווג סנטימנטים מהמודל, אפשר לכתוב הנחיה כזו:

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

Text: you look very nice today
Classification:

במקרה כזה, יכול להיות שהמודל לא יפיק את מה שחיפשתם. לדוגמה, אם הטקסט מכיל תווים חדשים, סביר להניח שתהיה לכך השפעה שלילית על ביצועי המודל. גישה מחמירה יותר היא להשתמש באסימונים מפרידים. ההנחיה תהפוך ל:

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

אפשר להפשט את הדוגמה הזו באמצעות פונקציה שמעבדת את הטקסט מראש:

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

עכשיו, אם תריצו את הפונקציה עם אותו הנחיה וטקסט, אתם אמורים לקבל את אותו הפלט:

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:well, looks like its time to have another child
<separator>
Prediction:

6. עיבוד לאחר פלט

הפלט של המודל הוא אסימונים עם הסתברויות שונות. בדרך כלל, כדי ליצור טקסט, צריך לבחור מבין האסימונים בעלי הסבירות הגבוהה ביותר ולבנות משפטים, פסקאות ואפילו מסמכים מלאים. עם זאת, למטרת הסיווג, מה שחשוב בפועל הוא אם המודל סבור שיש סבירות גבוהה יותר של Positive מאשר Negative, או להיפך.

בהתאם למודל שיצרתם קודם, כך אפשר לעבד את הפלט שלו להסתברויות הבלתי תלויות כדי לקבוע אם האסימון הבא הוא Positive או 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()))

כדי לבדוק את הפונקציה, מריצים אותה עם ההנחיה שיצרתם קודם:

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

הפלט אמור להיראות כך:

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

7. סיכום של הכול כמסווג

כדי שיהיה לכם קל, תוכלו לרכז את כל הפונקציות שיצרתם במסווג אחד דמוי sklearn, עם פונקציות מוכרות וקלות לשימוש, כמו predict() ו-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. כוונון עדין של דגם

LoRA הוא קיצור של Low-Rank Adaptation. זוהי טכניקת כוונון שמאפשרת לשפר ביעילות מודלים גדולים של שפה (LLM). מידע נוסף בנושא זמין במאמר LoRA: מתאם בדירוג נמוך של מודלים גדולים של שפה.

הטמעת Keras של Gemma מספקת שיטת enable_lora() שבה ניתן להשתמש לכוונון עדין:

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

אחרי שמפעילים את LoRA, אפשר להתחיל בתהליך הכוונון. התהליך נמשך כ-5 דקות בכל תקופה של Colab:

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)

אימונים לתקופות נוספות יובילו לרמת דיוק גבוהה יותר, עד שתתבצע התאמת יתר.

9. בדיקת התוצאות

עכשיו אפשר לבדוק את הפלט של המסווג הזריז שאימנתם עכשיו. הקוד הזה יפיק את ציון הכיתה החזוי באמצעות קטע טקסט:

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. הערכת דגם

לבסוף, בוחנים את הביצועים של המודל באמצעות שני מדדים נפוצים, ציון F1 ו-AUC-ROC. הציון F1 מזהה שגיאות חיוביות כוזבות ושגיאות חיוביות שקריות על ידי הערכה של הממוצע ההרמוני של הדיוק וזכירת הערך בסף סיווג מסוים. מצד שני, AUC-ROC לוכד את יחסי הגומלין בין התעריף החיובי האמיתי לבין שיעור החיוב החיובי השגוי על פני מגוון של ערכי סף ומחשב את השטח מתחת לעקומה הזו.

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

דרך מעניינת נוספת להעריך חיזויים של מודל היא מטריצות של בלבול. מטריצת בלבול תציג באופן חזותי את הסוגים השונים של שגיאות חיזוי.

from sklearn.metrics import confusion_matrix, ConfusionMatrixDisplay

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

לסיום, אפשר גם להסתכל על עקומת ה-ROC כדי לקבל מושג לגבי שגיאות חיזוי אפשריות כשמשתמשים בספי ניקוד שונים.

from sklearn.metrics import RocCurveDisplay, roc_curve

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