Informacje o generatywnej AI
Uruchom w Google Colab
Wyświetl źródło na GitHubie
Nauka w CodelabsZ tego ćwiczenia w Codelabs dowiesz się, jak utworzyć niestandardowy klasyfikator tekstu za pomocą dostrajania efektywnego pod kątem parametrów (PET). Zamiast dostrajania całego modelu metody PET aktualizują tylko niewielką liczbę parametrów, co sprawia, że ich trenowanie jest stosunkowo łatwe i szybkie. Ułatwia też modelowi uczenie się nowych zachowań przy stosunkowo niewielkiej ilości danych treningowych. Metodyka jest szczegółowo opisana w artykule Towards Agile Text Classifiers for Everyone, w którym pokazano, jak można stosować te techniki do różnych zadań związanych z bezpieczeństwem i osiągać najlepszą skuteczność przy użyciu zaledwie kilkuset przykładów treningowych.
Ten warsztat programistyczny korzysta z metody LoRA PET i mniejszego modelu Gemma (gemma_instruct_2b_en), ponieważ można go uruchamiać szybciej i bardziej wydajnie. W ramach tego projektu możesz zapoznać się z krokowymi instrukcjami dotyczącymi przetwarzania danych, ich formatowania na potrzeby modelu LLM, trenowania wag LoRA, a następnie oceny wyników. Ten projekt trenuje na podstawie zbioru danych ETHOS, czyli publicznie dostępnego zbioru danych do wykrywania mowy nienawiści, który powstał na podstawie komentarzy z YouTube i Reddita.
Po przeszkoleniu na podstawie zaledwie 200 przykładów (1/4 z danych) osiąga ono wartość F1 = 0,80 i ROC-AUC = 0,78, co jest nieco lepsze od SOTA podanego obecnie na tablicy liderów (w momencie pisania tego tekstu, 15 lutego 2024 r.). Po przeszkoleniu na pełnym zestawie 800 przykładów model osiąga wynik F1 83,74 i ROC-AUC 88,17. Większe modele, takie jak gemma_instruct_7b_en, zazwyczaj działają lepiej, ale koszty trenowania i wykonywania są też większe.
Ostrzeżenie: ponieważ ten warsztat programistyczny tworzy klasyfikator bezpieczeństwa do wykrywania mowy nienawiści, przykłady i ocena wyników zawierają nieprzyzwoite wyrażenia.
Instalacja i konfiguracja
W tym ćwiczeniu będziesz potrzebować najnowszej wersji keras (3), keras-nlp (0.8.0) i konta Kaggle, aby pobrać model Gemma.
import kagglehub
kagglehub.login()
pip install -q -U keras-nlppip install -q -U keras
import os
os.environ["KERAS_BACKEND"] = "tensorflow"
Wczytywanie zbioru danych ETHOS
W tej sekcji wczytasz zbiór danych, na podstawie którego będziesz trenować klasyfikator, a następnie przetworzysz go na zbiór treningowy i testowy. Użyjesz popularnego zbioru danych ETHOS, który został zebrany w celu wykrywania mowy nienawiści w mediach społecznościowych. Więcej informacji o tym, jak zebrano zbiór danych, można znaleźć w artykule ETHOS: an Online Hate Speech Detection Dataset (ETHOS: zbiór danych do wykrywania mowy nienawiści w internecie).
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']]
Pobieranie i tworzenie instancji modelu
Jak opisano w dokumentacji, model Gemma można łatwo stosować na wiele sposobów. W przypadku Keras musisz wykonać te 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
model.generate('Question: what is the capital of France? ', max_length=32)
Przetwarzanie wstępne tekstu i tokeny separatora
Aby ułatwić modelowi zrozumienie intencji, możesz przetworzyć tekst i użyć tokenów rozdzielników. Dzięki temu model rzadziej generuje tekst, który nie pasuje do oczekiwanego formatu. Możesz na przykład poprosić model o klasyfikację nastroju, piszą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, ale nie musi wygenerować oczekiwanych wyników. Jeśli na przykład tekst zawiera znaki nowej linii, prawdopodobnie negatywnie wpłynie to na skuteczność modelu. Bardziej niezawodne jest używanie separatorów tokenów. Prompt zmieni się w taki:
Classify the following text into one of the following classes:[Positive,Negative]
<separator>
Text: you look very nice today
<separator>
Prediction:
Można to zastosować za pomocą funkcji, która przetworzy tekst wstępnie:
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 teraz uruchomisz funkcję, używając tego samego prompta i tego samego tekstu, powinieneś otrzymać ten sam wynik:
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:
Przetwarzanie końcowe danych wyjściowych
Dane wyjściowe modelu to tokeny o różnych prawdopodobieństwach. Zazwyczaj do generowania tekstu wybierasz kilka najbardziej prawdopodobnych tokenów i tworzysz z nich zdania, akapity, a nawet całe dokumenty. Jednak na potrzeby klasyfikacji liczy się to, czy model uważa, że Positive jest bardziej prawdopodobne niż Negative, czy odwrotnie.
Po utworzeniu modelu możesz przetworzyć jego dane wyjściowe w niezależne prawdopodobieństwa, że następny element będzie odpowiednio Positive lub 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ą z promptem utworzonym wcześniej:
compute_output_probability(
model=model,
prompt=prompt,
target_classes=['Positive', 'Negative'],
)
{'Positive': 0.99994016, 'Negative': 5.984089e-05}
Pakowanie wszystkiego w klasyfikator
Aby ułatwić sobie pracę, możesz zawinąć wszystkie utworzone przez siebie funkcje w jedną klasyfikator typu sklearn, który zawiera łatwe w użyciu i znane funkcje, takie 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'))
Dostrajanie modelu
LoRA to skrót od „low-rank adaptation” (adaptacja niskiego rzędu). Jest to technika dostrajania, która może być używana do wydajnego dostrajania dużych modeli językowych. Więcej informacji znajdziesz w artykule LoRA: Low-Rank Adaptation of Large Language Models.
Implementacja Gemma w Keras udostępnia metodę enable_lora(), której możesz użyć do dokładnego dostosowania:
# 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 dokładnego dostosowania. Zajmuje to około 5 minut na każdą epokę w 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>
Trenowanie przez więcej epok spowoduje wzrost dokładności, dopóki nie nastąpi przetrenowanie.
Sprawdzanie wyników
Możesz teraz sprawdzić dane wyjściowe elastycznego klasyfikatora, który został właśnie wytrenowany. Ten kod wyświetli przewidywany wynik klasy na podstawie 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}
Ocena modelu
Na koniec ocenisz skuteczność modelu za pomocą 2 popularnych wskaźników: wynik F1 i AUC-ROC. Wynik F1 uwzględnia błędy fałszywie negatywne i fałszywie pozytywne, oceniając średnią harmoniczną precyzji i czułości przy określonym progu klasyfikacji. Z drugiej strony AUC-ROC pokazuje równowagę między współczynnikiem wyników prawdziwie a fałszywie pozytywnych przy różnych wartościach progowych i oblicza obszar pod tą krzywą.
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
Innym ciekawym sposobem oceny prognoz modelu są tablice pomyłek. Macierz błędów przedstawia wizualnie różne rodzaje błędów prognozowania.
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>
Możesz też spojrzeć na krzywą ROC, aby sprawdzić, jakie błędy przewidywania mogą wystąpić przy zastosowaniu różnych progów punktacji.
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>
Dodatek
Przeprowadziliśmy podstawowe badania przestrzeni hiperparametrów, aby lepiej zrozumieć związek między rozmiarem zbioru danych a wydajnością. Zobacz poniższy wykres.
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()
