1. 總覽
本程式碼研究室將說明如何使用參數有效調整 (PET) 建立自訂文字分類器。PET 方法只會更新少量的參數,因此訓練速度相對簡單、快速,不會微調整個模型。也能讓模型在訓練資料相對較少的情況下學習新行為。如要進一步瞭解方法,請參閱「適合所有人的靈活文字分類器」一文,當中說明這些技術如何應用於各種安全任務,並透過數百個訓練樣本實現最先進的成效。
本程式碼研究室會使用 LoRA PET 方法和較小的 Gemma 模型 (gemma_instruct_2b_en
),因為這麼做可以提升執行速度和效率。Colab 會說明擷取資料步驟、為 LLM 設定資料格式、訓練 LoRA 權重,然後評估成效。本程式碼研究室會使用 ETHOS 資料集進行訓練。這是一個用於偵測仇恨言論的公開資料集,並以 YouTube 和 Reddit 留言為基礎。如果只用 200 個範例 (資料集的 1/4) 訓練,模型就會達到 F1:0.80 和 ROC-AUC:0.78,略高於目前報告的排行榜報告的 SOTA (在撰寫當下,2024 年 2 月 15 日)。完整訓練 800 個範例後,F1 獲得 83.74 分,ROC-AUC 得分為 88.17。一般來說,gemma_instruct_7b_en
等較大型的模型會有更佳的成效,但訓練和執行成本也較高。
觸發警告:因為本程式碼研究室開發的安全分類器,用於偵測仇恨言論、例子和評估結果,其中包含一些可怕的用語。
2. 安裝和設定
在這個程式碼研究室中,您需要最新版的 keras
(3)、keras-nlp
(0.8.0) 和 Kaggle 帳戶,才能下載基礎模型。
!pip install -q -U keras-nlp
!pip install -q -U keras
如要登入 Kaggle,可以在 ~/.kaggle/kaggle.json
中儲存 kaggle.json
憑證檔案,或在 Colab 環境中執行下列指令:
import kagglehub
kagglehub.login()
3. 載入 ETHOS 資料集
在本節中,您將載入用於訓練分類器的資料集,並預先處理為訓練集和測試集。你會使用熱門的研究資料集 ETHOS,這是為了偵測社群媒體中的仇恨言論。如要進一步瞭解資料集的收集方式,請參閱《ETHOS: an Online Hate Speech Detection Dataset》(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 |
|
|
1 |
|
|
2 |
|
|
3 |
|
|
4 |
|
|
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 代表低排名調整作業。這是一種微調技術,可用於有效率地微調大型語言模型。詳情請參閱《LoRA: Low Rank Adaptation of Large Language Models》文件。
Gemma 的 Keras 實作提供 enable_lora()
方法,可讓您用來微調:
# Enable LoRA for the model and set the LoRA rank to 4.
model.backbone.enable_lora(rank=4)
啟用 LoRA 後,即可開始微調程序。每次執行 Colab 時,每個週期大約需要 5 分鐘的時間:
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()