1. Visão geral
Este codelab mostra como criar um classificador de texto personalizado usando o ajuste eficiente de parâmetros (PET, na sigla em inglês). Em vez de ajustar todo o modelo, os métodos PET atualizam apenas uma pequena quantidade de parâmetros, o que torna o treinamento relativamente fácil e rápido. Isso também facilita o aprendizado de novos comportamentos com relativamente poucos dados de treinamento. A metodologia é descrita em detalhes em Towards Agile Text Classifiers for Everyone (link em inglês), que mostra como essas técnicas podem ser aplicadas a várias tarefas de segurança e alcançar um desempenho de ponta com apenas algumas centenas de exemplos de treinamento.
Este codelab usa o método PET LoRA e o modelo Gemma menor (gemma_instruct_2b_en), já que ele pode ser executado com mais rapidez e eficiência. O Colab aborda as etapas de ingestão e formatação de dados para o LLM, treinamento de pesos LoRA e avaliação dos resultados. Este codelab usa o conjunto de dados ETHOS, disponível publicamente para detectar discurso de ódio e criado com base em comentários do YouTube e do Reddit. Quando treinado com apenas 200 exemplos (1/4 do conjunto de dados), ele alcança F1: 0,80 e ROC-AUC: 0,78, um pouco acima do SOTA atualmente informado no ranking (na data da redação, 15 de fevereiro de 2024). Quando treinado com os 800 exemplos completos, ele alcança uma pontuação F1 de 83,74 e uma pontuação ROC-AUC de 88,17. Modelos maiores, como o gemma_instruct_7b_en, geralmente têm um desempenho melhor, mas os custos de treinamento e execução também são maiores.
Alerta de gatilho: como este codelab desenvolve um classificador de segurança para detectar discurso de ódio, os exemplos e a avaliação dos resultados contêm linguagem ofensiva.
2. Instalação e configuração
Para este codelab, você vai precisar de uma versão recente do keras (3), do keras-nlp (0.8.0) e de uma conta do Kaggle para fazer o download do modelo de base.
!pip install -q -U keras-nlp
!pip install -q -U keras
Para fazer login no Kaggle, armazene o arquivo de credenciais kaggle.json em ~/.kaggle/kaggle.json ou execute o seguinte em um ambiente do Colab:
import kagglehub
kagglehub.login()
Este codelab foi testado usando o TensorFlow como back-end do Keras, mas é possível usar TensorFlow, PyTorch ou JAX:
import os
os.environ["KERAS_BACKEND"] = "tensorflow"
3. Carregar o conjunto de dados ETHOS
Nesta seção, você vai carregar o conjunto de dados em que treinará o classificador e fará o pré-processamento dele em um conjunto de treinamento e teste. Você vai usar o conjunto de dados de pesquisa ETHOS, que foi coletado para detectar discurso de ódio nas redes sociais. Saiba mais sobre como o conjunto de dados foi coletado no artigo ETHOS: um conjunto de dados on-line de detecção de discurso de ódio (em inglês).
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']]
Você vai ver algo parecido com isto:
o rótulo. | comentário | |
0 |
|
|
1 |
|
|
2 |
|
|
3 |
|
|
4 |
|
|
4. Fazer o download e instanciar o modelo
Conforme descrito na documentação, é fácil usar o modelo da Gemma de várias maneiras. Com o Keras, faça o seguinte:
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
Para testar se o modelo está funcionando, gere algum texto:
model.generate('Question: what is the capital of France? ', max_length=32)
5. Pré-processamento de texto e tokens separadores
Para ajudar o modelo a entender melhor nossa intenção, pré-processe o texto e use tokens separadores. Isso reduz a probabilidade de o modelo gerar texto que não se encaixa no formato esperado. Por exemplo, você pode tentar pedir uma classificação de sentimento ao modelo escrevendo um comando como este:
Classify the following text into one of the following classes:[Positive,Negative] Text: you look very nice today Classification:
Nesse caso, o modelo pode ou não gerar o que você está procurando. Por exemplo, se o texto contiver caracteres de nova linha, é provável que isso tenha um efeito negativo na performance do modelo. Uma abordagem mais robusta é usar tokens separadores. O comando fica assim:
Classify the following text into one of the following classes:[Positive,Negative] <separator> Text: you look very nice today <separator> Prediction:
Isso pode ser abstraído usando uma função que pré-processa o texto:
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:'])
Agora, se você executar a função usando o mesmo comando e texto de antes, vai receber a mesma saída:
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)
que vai gerar:
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. Pós-processamento de saída
As saídas do modelo são tokens com várias probabilidades. Normalmente, para gerar texto, você seleciona alguns dos tokens mais prováveis e cria frases, parágrafos ou até mesmo documentos completos. No entanto, para fins de classificação, o que realmente importa é se o modelo acredita que Positive é mais provável do que Negative ou vice-versa.
Considerando o modelo que você instanciou antes, veja como processar a saída dele em probabilidades independentes de o próximo token ser Positive ou 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()))
Teste essa função executando-a com o comando criado anteriormente:
compute_output_probability(
model=model,
prompt=prompt,
target_classes=['Positive', 'Negative'],
)
que vai gerar algo semelhante a:
{'Positive': 0.99994016, 'Negative': 5.984089e-05}
7. Encapsulando tudo como um classificador
Para facilitar o uso, você pode agrupar todas as funções que acabou de criar em um único classificador semelhante ao sklearn com funções fáceis de usar e conhecidas, como 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'))
8. Ajuste de modelos
LoRA significa adaptação de classificação baixa. É uma técnica de ajuste fino que pode ser usada para ajustar modelos de linguagem grandes de maneira eficiente. Leia mais sobre isso no artigo LoRA: adaptação de classificação baixa de modelos de linguagem grandes.
A implementação do Keras do Gemma oferece um método enable_lora() que pode ser usado para ajuste refinado:
# Enable LoRA for the model and set the LoRA rank to 4.
model.backbone.enable_lora(rank=4)
Depois de ativar o LoRA, você pode iniciar o processo de ajuste refinado. Isso leva aproximadamente 5 minutos por época no 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)
Treinar por mais épocas resulta em maior acurácia, até que ocorra overfitting.
9. Inspecionar os resultados
Agora você pode inspecionar a saída do classificador ágil que acabou de treinar. Esse código vai gerar a pontuação da classe prevista para um trecho de texto:
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. Model Evaluation
Por fim, você vai avaliar a performance do modelo usando duas métricas comuns: a pontuação F1 e a AUC-ROC. A pontuação F1 captura erros de falso negativo e falso positivo ao avaliar a média harmônica da precisão e do recall em um determinado limite de classificação. A AUC-ROC, por outro lado, captura a troca entre a taxa de verdadeiro positivo e a taxa de falso positivo em vários limites e calcula a área abaixo dessa curva.
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
Outra maneira interessante de avaliar as previsões do modelo são as matrizes de confusão. Uma matriz de confusão mostra visualmente os diferentes tipos de erros de previsão.
from sklearn.metrics import confusion_matrix, ConfusionMatrixDisplay
cm = confusion_matrix(y_true, y_pred)
ConfusionMatrixDisplay(
confusion_matrix=cm,
display_labels=agile_classifier.labels,
).plot()
Por fim, você também pode analisar a curva ROC para ter uma ideia dos possíveis erros de previsão ao usar diferentes limites de pontuação.
from sklearn.metrics import RocCurveDisplay, roc_curve
fpr, tpr, _ = roc_curve(y_true, y_prob, pos_label=1)
RocCurveDisplay(fpr=fpr, tpr=tpr).plot()
