Como apresentar classificadores de segurança Agile com Gemma

1. Visão geral

Este codelab ilustra 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 facilita e acelera o treinamento. Isso também torna mais fácil para um modelo aprender novos comportamentos com relativamente poucos dados de treinamento. A metodologia é descrita em detalhes em Towards Agile Text Classifiers for Everyone, que mostra como essas técnicas podem ser aplicadas a várias tarefas de segurança e alcançar o desempenho de última geração 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 de forma mais rápida e eficiente. O Colab aborda as etapas de ingestão de dados, formatação deles 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, criado com base em comentários do YouTube e do Reddit. Quando treinado com apenas 200 exemplos (1/4 do conjunto de dados), ele atinge F1: 0,80 e ROC-AUC: 0,78, um pouco acima do SOTA relatado atualmente no placar (no momento da redação deste documento, 15 de fevereiro de 2024). Quando treinado com todos os 800 exemplos, ele atinge uma pontuação F1 de 83,74 e uma pontuação ROC-AUC de 88,17. Modelos maiores, como gemma_instruct_7b_en, geralmente têm uma performance 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, há exemplos e uma avaliação dos resultados que contêm linguagem horrível.

2. Instalação e configuração

Neste codelab, você precisará da versão recente keras (3), keras-nlp (0.8.0) e de uma conta do Kaggle para fazer o download do modelo base.

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

Para fazer login no Kaggle, armazene o arquivo de credenciais do kaggle.json em ~/.kaggle/kaggle.json ou execute o seguinte em um ambiente Colab:

import kagglehub

kagglehub.login()

3. Carregar o conjunto de dados ETHOS

Nesta seção, você vai carregar o conjunto de dados no qual treinar nosso classificador e pré-processá-lo em um conjunto de treinamento e teste. Você vai usar o famoso conjunto de dados de pesquisa ETHOS, que foi coletado para detectar discurso de ódio em mídias sociais. Saiba mais sobre como o conjunto de dados foi coletado no artigo ETHOS: an Online Ódio Detection Dataset (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ê verá algo semelhante a:

	o rótulo.	comentário
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. Fazer o download e instanciar o modelo

Conforme descrito na documentação, você pode usar o modelo do Gemma com facilidade de várias maneiras. Com a Keras, você precisa fazer 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 um texto:

model.generate('Question: what is the capital of France? ', max_length=32)

5. Pré-processamento de texto e tokens de separador

Para ajudar o modelo a entender melhor nossa intent, pré-processe o texto e use tokens de separador. Isso diminui a probabilidade de o modelo gerar um texto que não se ajuste ao formato esperado. Por exemplo, é possível tentar solicitar uma classificação de sentimento do 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 produzir o que você está procurando. Por exemplo, se o texto contiver caracteres de nova linha, isso poderá ter um efeito negativo no desempenho do modelo. Uma abordagem mais robusta é usar tokens separadores. O comando se torna:

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)

O que deve resultar em:

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

Os resultados do modelo são tokens com várias probabilidades. Normalmente, para gerar texto, você seleciona os principais tokens 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 que Negative ou vice-versa.

Considerando o modelo instanciado anteriormente, é assim que é possível processar a saída nas 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()))

É possível testar essa função executando-a com um comando criado anteriormente:

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

O resultado será algo semelhante a:

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

7. Como resumir tudo como um classificador

Para facilitar o uso, é possível unir todas as funções recém-criadas 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 em Baixo Classificação. É uma técnica de ajuste fino que pode ser usada para ajustar com eficiência modelos de linguagem grandes. Leia mais sobre isso no artigo LoRA: adaptação baixa de modelos de linguagem grandes (em inglês).

A implementação do Gemma do Keras fornece um método enable_lora() que pode ser usado para ajustes:

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

Depois de ativar a LoRA, você pode iniciar o processo de ajuste. 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)

O treinamento de mais períodos resultará em maior acurácia, até que ocorra um overfitting.

9. Inspecionar os resultados

Agora é possível inspecionar a saída do classificador ágil que acabou de treinar. Este código vai gerar a pontuação prevista da turma com base em um 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. Avaliação de modelos

Por fim, você vai avaliar o desempenho do nosso modelo usando duas métricas comuns, a pontuação F1 e a AUC-ROC. A pontuação F1 captura erros de falsos negativos e falsos positivos avaliando a média harmônica de precisão e recall em um determinado limiar de classificação. Por outro lado, a AUC-ROC captura o equilíbrio entre a taxa de verdadeiro positivo e a de falso positivo em vários limites e calcula a área sob essa 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 de modelos são as matrizes de confusão. Uma matriz de confusão retrata 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, também é possível observar a curva ROC para ter uma noção 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_score, pos_label=1)
RocCurveDisplay(fpr=fpr, tpr=tpr).plot()