 Meiner Meinung nach ist generative KI
|
 In Google Colab ausführen
|
 Quellcode auf GitHub ansehen
|
 In Codelabs lernen
|
In diesem Codelab wird gezeigt, wie Sie mithilfe der parametereffizienten Abstimmung (PET) einen benutzerdefinierten Textklassifikator erstellen. Anstatt das gesamte Modell zu optimieren, aktualisieren PET-Methoden nur eine kleine Anzahl von Parametern. Das macht das Training relativ einfach und schnell. Außerdem ist es für ein Modell einfacher, mit relativ wenig Trainingsdaten neue Verhaltensweisen zu erlernen. Die Methodik wird ausführlich in Towards Agile Text Classifiers for Everyone beschrieben. Dort wird gezeigt, wie diese Techniken auf eine Vielzahl von Sicherheitsaufgaben angewendet werden können und mit nur wenigen hundert Trainingsbeispielen eine Spitzenleistung erzielen.
In diesem Codelab wird die LoRA-PET-Methode und das kleinere Gemma-Modell (gemma_instruct_2b_en) verwendet, da es schneller und effizienter ausgeführt werden kann. In der colab werden die Schritte zum Aufnehmen von Daten, zum Formatieren für das LLM, zum Trainieren von LoRA-Gewichten und zum Bewerten der Ergebnisse beschrieben. Dieses Codelab basiert auf dem ETHOS-Dataset, einem öffentlich zugänglichen Dataset zur Erkennung von Hassrede, das aus YouTube- und Reddit-Kommentaren erstellt wurde.
Bei der Schulung mit nur 200 Beispielen (1/4 des Datensatzes) erreicht er einen F1-Wert von 0,80 und eine ROC-AUC von 0,78, was etwas über dem derzeit auf der Bestenliste gemeldeten Stand der Technik liegt (Stand: 15. Februar 2024). Bei der Schulung mit den vollständigen 800 Beispielen erreicht er einen F1-Wert von 83,74 und einen ROC-AUC-Wert von 88,17. Größere Modelle wie gemma_instruct_7b_en erzielen in der Regel eine bessere Leistung, aber die Kosten für Training und Ausführung sind auch höher.
Triggerwarnung: Da in diesem Codelab ein Sicherheitsklassifikator zum Erkennen von Hassrede entwickelt wird, enthalten Beispiele und die Bewertung der Ergebnisse schreckliche Formulierungen.
Installation und Einrichtung
Für dieses Codelab benötigen Sie eine aktuelle Version von keras (3) oder keras-nlp (0.8.0) sowie ein Kaggle-Konto, um ein Gemma-Modell herunterzuladen.
import kagglehub
kagglehub.login()
pip install -q -U keras-nlppip install -q -U keras
import os
os.environ["KERAS_BACKEND"] = "tensorflow"
ETHOS-Dataset laden
In diesem Abschnitt laden Sie das Dataset, mit dem Sie den Klassifikator trainieren möchten, und verarbeiten es zu einem Trainings- und Test-Dataset. Sie verwenden den beliebten Forschungsdatensatz ETHOS, der zum Erkennen von Hassrede in sozialen Medien erfasst wurde. Weitere Informationen zur Zusammenstellung des Datasets finden Sie im Artikel ETHOS: an Online Hate Speech Detection Dataset (ETHOS: ein Dataset zur Erkennung von Online-Hassrede).
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']]
Modell herunterladen und instanziieren
Wie in der Dokumentation beschrieben, können Sie das Gemma-Modell auf vielfältige Weise verwenden. Mit Keras gehen Sie so vor:
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)
Textvorverarbeitung und Trennzeichen-Tokens
Damit das Modell die Absicht besser verstehen kann, können Sie den Text vorverarbeiten und Trennzeichen-Token verwenden. So ist es weniger wahrscheinlich, dass das Modell Text generiert, der nicht dem erwarteten Format entspricht. Sie können beispielsweise versuchen, eine Sentimentklassifizierung vom Modell anzufordern, indem Sie einen Prompt wie diesen schreiben:
Classify the following text into one of the following classes:[Positive,Negative]
Text: you look very nice today
Classification:
In diesem Fall gibt das Modell möglicherweise nicht das gewünschte Ergebnis aus. Wenn der Text beispielsweise Zeilenumbruchzeichen enthält, wirkt sich das wahrscheinlich negativ auf die Modellleistung aus. Robuster ist es, Trennzeichen-Tokens zu verwenden. Der Prompt lautet dann:
Classify the following text into one of the following classes:[Positive,Negative]
<separator>
Text: you look very nice today
<separator>
Prediction:
Dies kann mit einer Funktion abstrahiert werden, die den Text vorverarbeitet:
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:'])
Wenn Sie die Funktion jetzt mit demselben Prompt und demselben Text wie zuvor ausführen, sollten Sie dieselbe Ausgabe erhalten:
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:
Nachbearbeitung der Ausgabe
Die Ausgaben des Modells sind Tokens mit verschiedenen Wahrscheinlichkeiten. Normalerweise würden Sie zum Generieren von Text die wahrscheinlichsten Tokens auswählen und Sätze, Absätze oder sogar ganze Dokumente erstellen. Für die Klassifizierung ist jedoch entscheidend, ob das Modell der Meinung ist, dass Positive wahrscheinlicher ist als Negative oder umgekehrt.
Anhand des Modells, das Sie zuvor instanziiert haben, können Sie die Ausgabe in die unabhängigen Wahrscheinlichkeiten verarbeiten, dass das nächste Token Positive oder Negative ist:
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()))
Sie können diese Funktion testen, indem Sie sie mit dem zuvor erstellten Prompt ausführen:
compute_output_probability(
model=model,
prompt=prompt,
target_classes=['Positive', 'Negative'],
)
{'Positive': 0.99994016, 'Negative': 5.984089e-05}
Alles in einem Klassifikator zusammenfassen
Für eine einfache Verwendung können Sie alle gerade erstellten Funktionen in einem einzigen sklearn-ähnlichen Klassifikator mit nutzerfreundlichen und vertrauten Funktionen wie predict() und predict_score() einschließen.
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'))
Modellabstimmung
LoRA steht für „Low-Rank Adaptation“. Es ist eine Methode zur Feinabstimmung, mit der Large Language Models effizient optimiert werden können. Weitere Informationen finden Sie im Artikel LoRA: Low-Rank Adaptation of Large Language Models.
Die Keras-Implementierung von Gemma bietet eine enable_lora()-Methode, die Sie zur Feinabstimmung verwenden können:
# Enable LoRA for the model and set the LoRA rank to 4.
model.backbone.enable_lora(rank=4)
Nachdem Sie LoRA aktiviert haben, können Sie mit der Feinabstimmung beginnen. Das dauert in Colab etwa 5 Minuten pro Epoche:
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>
Ein Training über mehr Epochen führt zu einer höheren Genauigkeit, bis es zu Überanpassung kommt.
Ergebnisse prüfen
Sie können sich jetzt die Ausgabe des agilen Klassifikators ansehen, den Sie gerade trainiert haben. Mit diesem Code wird der vorhergesagte Klassenwert für einen Text ausgegeben:
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}
Modellbewertung
Abschließend bewerten Sie die Leistung des Modells anhand von zwei gängigen Messwerten: dem F1-Wert und dem AUC-ROC. Der F1-Wert erfasst falsch negative und falsch positive Fehler, indem der harmonische Mittelwert von Precision und Recall bei einem bestimmten Klassifizierungsgrenzwert berechnet wird. Der AUC-ROC hingegen erfasst den Kompromiss zwischen der Rate richtig positiver Ergebnisse und der Rate falsch positiver Ergebnisse über eine Vielzahl von Schwellenwerten hinweg und berechnet den Bereich unter dieser Kurve.
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
Eine weitere interessante Möglichkeit zur Bewertung von Modellvorhersagen sind Verwirrungsmatrizen. Eine Konfidenzmatrix stellt die verschiedenen Arten von Vorhersagefehlern visuell dar.
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>
Außerdem können Sie sich die ROC-Kurve ansehen, um ein Gefühl für potenzielle Vorhersagefehler bei Verwendung verschiedener Bewertungsgrenzwerte zu bekommen.
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>
Anhang
Wir haben einige grundlegende explorative Datenanalysen durchgeführt, um ein besseres Verständnis für den Zusammenhang zwischen der Größe des Datensatzes und der Leistung zu erhalten. Siehe das folgende Diagramm.
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()
