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 o modelo inteiro, os métodos de PET atualizam apenas uma pequena quantidade de parâmetros, o que torna o treinamento relativamente fácil e rápido. Isso também facilita a aprendizagem de novos comportamentos com dados de treinamento relativamente pequenos. 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 estado da arte com apenas algumas centenas de exemplos de treinamento.

Este codelab usa o método de 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 de dados, formatação para o LLM, treinamento de pesos de LoRA e avaliação dos resultados. Este codelab é treinado com o conjunto de dados ETHOS, um conjunto de dados 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 alcança F1: 0,80 e ROC-AUC: 0,78, um pouco acima do SOTA atualmente informado na liderança (na época em que este artigo foi escrito, 15 de fevereiro de 2024). Quando treinado com os 800 exemplos, ele alcança 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 um desempenho melhor, mas os custos de treinamento e execução também são maiores.

Aviso 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 horrível.

2. Instalação e configuração

Para este codelab, você vai precisar de uma versão recente do keras (3), keras-nlp (0.8.0) e 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, você pode armazenar o arquivo de credenciais kaggle.json em ~/.kaggle/kaggle.json ou executar o seguinte em um ambiente do Colab:

import kagglehub

kagglehub.login()

Este codelab foi testado usando o TensorFlow como back-end do Keras, mas você pode usar o 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 para treinar nosso classificador e pré-processá-lo 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 mídias sociais. Confira mais informações sobre como o conjunto de dados foi coletado no artigo ETHOS: an Online Hate Speech Detection Dataset.

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 encontrar algo parecido com:

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, é possível usar o modelo Gemma de várias maneiras. Com o 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

Você pode testar se o modelo está funcionando gerando algum 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 intenção, você pode pré-processar o texto e usar tokens de separação. Isso torna menos provável que o modelo gere um texto que não se encaixe no formato esperado. Por exemplo, você pode 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 gerar o que você está procurando. Por exemplo, se o texto tiver caracteres de nova linha, é provável que ele tenha um efeito negativo na performance do modelo. Uma abordagem mais robusta é usar tokens de separador. O prompt vai ficar 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 abstrato 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, 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 deve 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 entre os tokens mais prováveis e constrói frases, parágrafos ou até 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ê instanciar anteriormente, é assim que você pode 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()))

Para testar essa função, execute-a com o comando criado anteriormente:

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

O que vai gerar algo semelhante a este:

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

7. Como agrupar 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 conhecidas e fáceis de usar, 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 fino do modelo

LoRA significa "adaptação de baixa classificação". É 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 baixa classificação de modelos de linguagem grandes.

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

# 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 fino. 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 para mais épocas resulta em maior precisão, até que ocorra a superadaptação.

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 com base em 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. Avaliação de modelos

Por fim, você vai avaliar o desempenho do 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 da precisão e do recall em um determinado limite de classificação. Por outro lado, o AUC-ROC captura a compensação 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 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()

matriz de confusão

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()

Curva ROC