Este artigo descreve uma abordagem para identificar a intenção em consultas de sistemas de perguntas e respostas (Q&A), utilizando um modelo de clasificação de texto baseado na arquitetura do Codificador Transformer.
Preparação de Dados e Construção do Vocabulário
Iniciamos configurando o ambiente, carrregando os dados e preparando-os para o modelo. Utilizamos a biblioteca torchtext para gerenciar o processamento de texto, incluindo tokenização, criação de vocabulário e organização dos dados em batches.
import os
import torch
from torchtext import data
from torchtext.data import Iterator, BucketIterator
import pandas as pd
import pickle
# Determina o dispositivo (GPU ou CPU) para execução
processador = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
# Define o caminho base para os dados de classificação
caminho_base_classificacao = os.path.abspath(os.path.join(os.getcwd(), '..', '..'))
# Caminho para o arquivo CSV de treinamento
caminho_dados_treino = os.path.join(caminho_base_classificacao, 'classification_data', 'knowledge_point_qa_data.csv')
# Carrega os dados de treinamento de um arquivo CSV
df_treino = pd.read_csv(caminho_dados_treino)
# Função de tokenização simples: divide a frase em palavras por espaço
fn_tokenizador = lambda texto: texto.split(' ')
# Define as configurações para o campo de texto
campo_texto = data.Field(
sequential=True,
tokenize=fn_tokenizador,
lower=True,
use_vocab=True,
pad_token='<pad>',
unk_token='<unk>',
batch_first=True,
fix_length=20 # Define um comprimento fixo para as sequências
)
# Define as configurações para o campo de rótulo (intenção)
campo_rotulo = data.Field(
sequential=False,
use_vocab=False # Rótulos são numéricos, não precisam de vocabulário próprio
)
# Função auxiliar para criar exemplos de dataset a partir de um DataFrame
def criar_dataset_exemplos(dataframe_csv, campo_txt, campo_lbl, modo_teste=False):
especificacao_campos = [('id', None), ('texto', campo_txt), ('rotulo', campo_lbl)]
lista_exemplos = []
if modo_teste: # No modo de teste, o rótulo não é carregado
for texto_item in dataframe_csv['text']:
lista_exemplos.append(data.Example.fromlist([None, texto_item, None], especificacao_campos))
else: # Para treinamento, carregamos texto e rótulo
for texto_item, rotulo_item in zip(dataframe_csv['text'], dataframe_csv['label']):
lista_exemplos.append(data.Example.fromlist([None, texto_item, rotulo_item], especificacao_campos))
return lista_exemplos, especificacao_campos
# Obtém exemplos para o conjunto de treinamento
exemplos_treino, campos_treino = criar_dataset_exemplos(df_treino, campo_texto, campo_rotulo)
# Cria o objeto Dataset para treinamento
conjunto_treino = data.Dataset(exemplos_treino, campos_treino)
# Constrói o vocabulário a partir dos dados de treinamento
# Um `min_freq=1` significa que todas as palavras que aparecem pelo menos uma vez serão incluídas
campo_texto.build_vocab(conjunto_treino, min_freq=1)
# Salva o vocabulário construído em um arquivo pickle para uso futuro
caminho_vocabulario = os.path.join(os.getcwd(), 'vocabulario.pkl')
with open(caminho_vocabulario, 'wb') as f_vocab:
pickle.dump(campo_texto.vocab, f_vocab)
# Define o tamanho do batch para o treinamento
TAMANHO_BATCH = 163
# Cria o iterador de dados para o conjunto de treinamento
# `BucketIterator` agrupa sequências de comprimento semelhante para eficiência
iterador_treino = BucketIterator(
dataset=conjunto_treino,
batch_size=TAMANHO_BATCH,
shuffle=True,
sort_within_batch=False
)
Arquitetura do Codificador Transformer
A seguir, implementamos o modelo do Codificador Transformer, que é a base para a classificação de intenções. Este modelo incorpora camadas de embedding de token e posição, múltiplas camadas de codificador, atenção multi-cabeça e redes de feedforward. A máscara de origem é usada para ignorar tokens de preenchimento (<pad>) durante o cálculo da atenção.
import torch.nn as nn
import torch.nn.functional as F
'''
Componente principal do Codificador Transformer.
Combina embeddings de token e posição, processa-os através de N camadas de codificador
e finaliza com uma camada linear para classificação.
'''
class CodificadorTransformer(nn.Module):
def __init__(self, tam_entrada, tam_saida, tam_emb, num_camadas, num_cabecas, tam_pf, dropout, comprimento_posicao, indice_pad):
super(CodificadorTransformer, self).__init__()
self.indice_pad = indice_pad
self.escala_emb = torch.sqrt(torch.FloatTensor([tam_emb])).to(processador) # Escala para os embeddings
# Embedding para tokens (palavras)
self.embedding_token = nn.Embedding(tam_entrada, tam_emb)
# Embedding para a posição dos tokens na sequência
self.embedding_posicao = nn.Embedding(comprimento_posicao, tam_emb)
# Módulo que contém as camadas do Codificador
self.camadas_codificador = nn.ModuleList([CamadaCodificador(tam_emb, num_cabecas, tam_pf, dropout) for _ in range(num_camadas)])
self.dropout = nn.Dropout(dropout)
self.camada_final = nn.Linear(tam_emb, tam_saida)
# Cria uma máscara para ignorar tokens de preenchimento (`<pad>`)
def criar_mascara_origem(self, entrada_src):
# entrada_src: [tamanho_batch, comprimento_sequencia_src]
# mascara_src: [tamanho_batch, 1, 1, comprimento_sequencia_src]
mascara_src = (entrada_src != self.indice_pad).unsqueeze(1).unsqueeze(2)
return mascara_src
def forward(self, entrada_src):
# entrada_src: [tamanho_batch, comprimento_sequencia]
# mascara_src: [tamanho_batch, 1, 1, comprimento_sequencia]
mascara_src = self.criar_mascara_origem(entrada_src)
tamanho_batch = entrada_src.shape[0]
comprimento_seq = entrada_src.shape[1]
# Gera tensores de posição: [0, 1, ..., comprimento_seq-1] para cada item no batch
posicoes = torch.arange(0, comprimento_seq).unsqueeze(0).repeat(tamanho_batch, 1).to(processador)
# Calcula os embeddings dos tokens e posições, e aplica a escala
emb_tokens = self.embedding_token(entrada_src) * self.escala_emb
emb_posicoes = self.embedding_posicao(posicoes)
# Combina embeddings de token e posição com dropout
# src: [tamanho_batch, comprimento_sequencia, tam_emb]
src_combinado = self.dropout(emb_tokens + emb_posicoes)
# Passa a entrada combinada através de cada camada do codificador
for camada in self.camadas_codificador:
src_combinado = camada(src_combinado, mascara_src)
# Pooling simples (soma) ao longo da dimensão da sequência e camada final
# src_combinado: [tamanho_batch, comprimento_sequencia, tam_emb] -> [tamanho_batch, tam_saida]
src_combinado = src_combinado.permute(0, 2, 1) # [tamanho_batch, tam_emb, comprimento_sequencia]
src_somado = torch.sum(src_combinado, dim=-1) # [tamanho_batch, tam_emb]
saida_final = self.camada_final(src_somado)
return saida_final
'''
Uma única camada do Codificador Transformer.
Consiste em uma camada de atenção multi-cabeça e uma camada de feedforward,
cada uma com dropout, conexão residual e normalização de camada.
'''
class CamadaCodificador(nn.Module):
def __init__(self, tam_emb, num_cabecas, tam_pf, dropout):
super(CamadaCodificador, self).__init__()
# Normalização após a camada de atenção
self.norm_atencao = nn.LayerNorm(tam_emb)
# Normalização após a camada de feedforward
self.norm_feedforward = nn.LayerNorm(tam_emb)
# Camada de atenção multi-cabeça
self.atencao_multi_cabeca = AtencaoMultiCabeca(tam_emb, num_cabecas, dropout)
# Camada de rede feedforward
self.rede_feedforward = CamadaFeedForward(tam_emb, tam_pf, dropout)
self.dropout = nn.Dropout(dropout)
def forward(self, src, mascara_src):
# src: [tamanho_batch, comprimento_sequencia, tam_emb]
# mascara_src: [tamanho_batch, 1, 1, comprimento_sequencia]
# Camada de atenção multi-cabeça
# _saida_atencao: [tamanho_batch, comprimento_sequencia, tam_emb]
_saida_atencao, _ = self.atencao_multi_cabeca(src, src, src, mascara_src)
# Aplica dropout, conexão residual e normalização
src = self.norm_atencao(src + self.dropout(_saida_atencao))
# Camada de rede feedforward
# _saida_ff: [tamanho_batch, comprimento_sequencia, tam_emb]
_saida_ff = self.rede_feedforward(src)
# Aplica dropout, conexão residual e normalização novamente
src = self.norm_feedforward(src + self.dropout(_saida_ff))
return src
'''
Implementa o mecanismo de Atenção Multi-Cabeça.
Divide o embedding em múltiplas "cabeças", calcula a atenção para cada uma,
e então concatena e projeta os resultados.
'''
class AtencaoMultiCabeca(nn.Module):
def __init__(self, tam_emb, num_cabecas, dropout):
super(AtencaoMultiCabeca, self).__init__()
assert tam_emb % num_cabecas == 0 # O embedding deve ser divisível pelo número de cabeças
self.tam_emb = tam_emb
self.num_cabecas = num_cabecas
self.dim_cabeca = tam_emb // num_cabecas # Dimensão de cada cabeça
# Camadas lineares para projetar Q, K, V
self.proj_q = nn.Linear(tam_emb, tam_emb)
self.proj_k = nn.Linear(tam_emb, tam_emb)
self.proj_v = nn.Linear(tam_emb, tam_emb)
# Camada linear final para combinar as saídas das cabeças
self.proj_saida = nn.Linear(tam_emb, tam_emb)
self.dropout = nn.Dropout(dropout)
# Fator de escala para o cálculo da atenção
self.escala_atencao = torch.sqrt(torch.FloatTensor([self.dim_cabeca])).to(processador)
def forward(self, query, key, value, mascara=None):
# query: [tamanho_batch, comprimento_query, tam_emb]
# key: [tamanho_batch, comprimento_key, tam_emb]
# value: [tamanho_batch, comprimento_value, tam_emb]
tamanho_batch = query.shape[0]
# Projeta Q, K, V
# Q_proj, K_proj, V_proj: [tamanho_batch, comprimento_seq, tam_emb]
Q_proj = self.proj_q(query)
K_proj = self.proj_k(key)
V_proj = self.proj_v(value)
# Redimensiona Q, K, V para múltiplas cabeças
# Por exemplo, Q_proj: [tamanho_batch, comprimento_query, num_cabecas, dim_cabeca]
# Depois, permuta para: [tamanho_batch, num_cabecas, comprimento_query, dim_cabeca]
Q_multi = Q_proj.view(tamanho_batch, -1, self.num_cabecas, self.dim_cabeca).permute(0, 2, 1, 3)
K_multi = K_proj.view(tamanho_batch, -1, self.num_cabecas, self.dim_cabeca).permute(0, 2, 1, 3)
V_multi = V_proj.view(tamanho_batch, -1, self.num_cabecas, self.dim_cabeca).permute(0, 2, 1, 3)
# Calcula a energia de atenção (Q * K^T / escala)
# energia_atencao: [tamanho_batch, num_cabecas, comprimento_query, comprimento_key]
energia_atencao = torch.matmul(Q_multi, K_multi.permute(0, 1, 3, 2)) / self.escala_atencao
# Aplica máscara se fornecida (útil para padding ou atenção futura)
if mascara is not None:
energia_atencao = energia_atencao.masked_fill(mascara == 0, -1e10)
# Converte energia em probabilidades de atenção usando softmax
# probabilidades_atencao: [tamanho_batch, num_cabecas, comprimento_query, comprimento_key]
probabilidades_atencao = torch.softmax(energia_atencao , dim = -1)
# Aplica dropout e multiplica pelas matrizes de valor
# saida_atencao: [tamanho_batch, num_cabecas, comprimento_query, dim_cabeca]
saida_atencao = torch.matmul(self.dropout(probabilidades_atencao), V_multi)
# Recombina as cabeças: permuta e usa `contiguous()` para garantir layout de memória
# saida_atencao: [tamanho_batch, comprimento_query, num_cabecas, dim_cabeca]
saida_atencao = saida_atencao.permute(0, 2, 1, 3).contiguous()
# Redimensiona para o tamanho original do embedding
# saida_atencao: [tamanho_batch, comprimento_query, tam_emb]
saida_atencao = saida_atencao.view(tamanho_batch, -1, self.tam_emb)
# Camada linear final para a saída
saida_final = self.proj_saida(saida_atencao)
return saida_final, probabilidades_atencao
'''
Camada de Feedforward Posição-a-Posição.
Aplica duas transformações lineares com uma função de ativação ReLU no meio.
'''
class CamadaFeedForward(nn.Module):
def __init__(self, tam_emb, tam_pf, dropout):
super(CamadaFeedForward, self).__init__()
self.linear_1 = nn.Linear(tam_emb, tam_pf)
self.linear_2 = nn.Linear(tam_pf, tam_emb)
self.dropout = nn.Dropout(dropout)
def forward(self, entrada_x):
# entrada_x: [tamanho_batch, comprimento_sequencia, tam_emb]
# Primeira transformação linear com ReLU e dropout
# entrada_x: [tamanho_batch, comprimento_sequencia, tam_pf]
entrada_x = self.dropout(torch.relu(self.linear_1(entrada_x)))
# Segunda transformação linear
# entrada_x: [tamanho_batch, comprimento_sequencia, tam_emb]
entrada_x = self.linear_2(entrada_x)
return entrada_x
Treinamento do Modelo
Finalmente, configuramos os hiperparâmetros do modelo, inicializamos o otimizador e a função de perda, e executamos o loop de treinamento. Utilizamos torch.utils.tensorboard.SummaryWriter para monitorar o progresso do treinamento, registrando a perda em cada iteração.
from torch.utils.tensorboard import SummaryWriter
# Inicializa o SummaryWriter para TensorBoard
escritor_log = SummaryWriter(os.getcwd()+'/logs_treino', comment='codificador-transformer')
# Configuração dos hiperparâmetros do modelo
tamanho_vocabulario = len(campo_texto.vocab) # Dimensão de entrada (número de palavras únicas)
numero_classes = 9 # Número de categorias de intenção
dim_embedding = 256
num_camadas_encoder = 3
num_cabecas_atencao = 8
dim_feedforward = 512
taxa_dropout = 0.5
comprimento_max_posicao = 20 # Comprimento máximo da sequência para embeddings de posição
# Índice do token de preenchimento (`<pad>`)
indice_pad_token = campo_texto.vocab.stoi[campo_texto.pad_token]
# Instancia o modelo Codificador Transformer
modelo_intencao = CodificadorTransformer(
tamanho_vocabulario,
numero_classes,
dim_embedding,
num_camadas_encoder,
num_cabecas_atencao,
dim_feedforward,
taxa_dropout,
comprimento_max_posicao,
indice_pad_token
).to(processador)
# Define o modelo para o modo de treinamento
modelo_intencao.train()
# Configura o otimizador (Adam com taxa de aprendizado e decaimento de peso)
otimizador = torch.optim.Adam(modelo_intencao.parameters(), lr=0.001, weight_decay=0.01)
# Função de perda para classificação multi-classe
criterio_perda = nn.CrossEntropyLoss()
# Inicia o treinamento e gravação de logs
with escritor_log:
num_epocas = 300
for epoca_atual in range(num_epocas):
for indice_batch, dados_batch in enumerate(iterador_treino):
# Obtém texto e rótulos do batch, movendo-os para o dispositivo correto
texto_batch = dados_batch.text.to(processador)
rotulo_batch = dados_batch.label.to(processador)
# Forward pass: obtém as previsões do modelo
saida_modelo = modelo_intencao(texto_batch)
# Calcula a perda
perda_batch = criterio_perda(saida_modelo, rotulo_batch)
# Zera os gradientes, realiza backpropagation e atualiza os pesos
otimizador.zero_grad()
perda_batch.backward()
otimizador.step()
# Imprime o progresso a cada 10 épocas
if (epoca_atual + 1) % 10 == 0:
print (f'Época [{epoca_atual+1}/{num_epocas}], Perda: {perda_batch.item():.4f}')
# Adiciona a perda ao TensorBoard
escritor_log.add_scalar('Perda de Treinamento', perda_batch.item(), global_step=epoca_atual+1)
# Garante que todos os eventos sejam gravados e fecha o escritor
escritor_log.flush()
escritor_log.close()
# Salva os pesos do modelo treinado em um arquivo
caminho_modelo_salvo = os.path.join(os.getcwd(), "modelo_intencao_transformer.h5")
torch.save(modelo_intencao.state_dict(), caminho_modelo_salvo)