Transformer的训练和测试_transformer 训练

训练

在训练transformer的时候因为我们做的是机器翻译的任务，所以在训练的时候编码器和解码器都是需要输入的，但是在上面构建的dataloader的时候我们只是构建了source _indices和target_indices，也就是上面只是构建的是源语和目标语，但是实际上我们的目标语还应该拆分成为解码器的输入和正确的解码器输出（也就是在和输出的语言进行求交叉熵损失的正确数值），所以需要对dataloader进行修改，要将目标语言的除了最后一个（输出结束标志符号）作为解码器的输入，除了目标语第一个作为解码器的输出，这样可以模拟自回归（但是并不是真正的自回归，因为我们在解码器输入的并不是模型上一部输出的答案，而是正确的答案，所以这个过程又被叫做teaching learning）修改代码如下

import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader
from tokenizers import Tokenizer
import sys
sys.path.append('/root/autodl-tmp/mt1/subword/')
from model.config import *
from model.selftransformer import Transformer
 
 
 
def custom_collate(batch):
 
    # 获取当前批次中源语言和目标语言的索引序列
    enc_inputs = [item['enc_inputs'] for item in batch]
    dec_inputs = [item['dec_inputs'] for item in batch]
    dec_output = [item['dec_output'] for item in batch]
 
    # 计算当前批次中最大的序列长度
    max_source_len = max(len(seq) for seq in enc_inputs)
    max_dec_inputs_len = max(len(seq) for seq in dec_inputs)
    max_dec_output_len = max(len(seq) for seq in dec_output)
 
    # 填充源语言和目标语言的索引序列，使它们的长度相同
    padded_source_seqs = [seq + [0] * (max_source_len - len(seq)) for seq in enc_inputs]
    padded_dec_inputs_seqs = [seq + [0] * (max_dec_inputs_len - len(seq)) for seq in dec_inputs]
    padded_dec_output_seqs = [seq + [0] * (max_dec_output_len - len(seq)) for seq in dec_output]
 
    # 转换为 PyTorch tensor
    enc_inputs = torch.tensor(padded_source_seqs)
    dec_inputs_tensor = torch.tensor(padded_dec_inputs_seqs)
    dec_output_tensor = torch.tensor(padded_dec_output_seqs)
 
    return {
        'enc_inputs': enc_inputs,
        'dec_inputs': dec_inputs_tensor,
        'dec_output':dec_output_tensor
    }
 
class TranslationDataset(Dataset):
    def __init__(self, source_sentences, target_sentences, source_tokenizer, target_tokenizer):
        self.source_sentences = source_sentences
        self.target_sentences = target_sentences
        self.source_tokenizer = source_tokenizer
        self.target_tokenizer = target_tokenizer
 
    def __len__(self):
        return len(self.source_sentences)
 
    def __getitem__(self, idx):
        source_sentence = self.source_sentences[idx]
        target_sentence = self.target_sentences[idx]
 
        # 将句子转换为索引序列
        # source_indices = [self.source_vocab[word] for word in source_sentence.split()]
        source_indices = self.source_tokenizer.encode(source_sentence).ids
        target_indices = self.target_tokenizer.encode(target_sentence).ids
        length = len(target_indices)
        dec_inputs = target_indices[0:length-1]
        dec_output = target_indices[1:length]
 
        # 返回源语言和目标语言的索引序列
        return {
            'enc_inputs': source_indices,
            'dec_inputs': dec_inputs,
            'dec_output':dec_output
        }

在训练的过程中使用了warmup进行前3个epoch的预热，其实在这部分有一个缺点就是大多数的模型在使用预热的时候都是前多少步进行预热，在这里没有使用这种方式的原因是并不清楚这些数据集一共要有多少个批量，也就是多少步（当然可以进行计算，使用记录的个数除以批量的大小，个人感觉使用5%差不多）

注意：在进行预热的时候传入的lr_lambda应该是一个匿名函数，他会自动传入一个这个epoch或者是当前是多少步，然后应该在自定义的函数中返回的是一个使用最大学习率的百分比，千万不要以为返回的是学习率，在进行学习率更新的时候其实会自动和学习率进行相乘

完整代码如下：

import os
import sys
from tqdm import tqdm
import torch
from torch import nn as nn
from torch import optim
from torch.optim.lr_scheduler import LambdaLR
sys.path.append('/root/autodl-tmp/mt1/subword/tool/')
 
 
from model.selftransformer import Transformer
 
from model.config import *
from data_process.build_dataloader import *
from tool.sentences_tool import get_sentences
from tool.tokenizer_tool import tokenizer_de,tokenizer_en
 
 
 
 
def judge_rate_true(predict, target, ignore_index=0):
    # 在不是0的地方填充成为0，也就是这个批量中实际的token
    the_true_target = target.ne(ignore_index)
    # 这个批量中应该存在的token数量
    the_true_target_num = the_true_target.sum().item()
    # predict.max(dim=-1)会返回两个数值，第一个是最大的元素（values），第二个是这个最大的元素所在的索引（indices）
    the_category_ = predict.max(dim=-1).indices
    # eq函数会逐个元素进行判断，如果是相等的话就返回True，在进行sum()的时候会返回张量中True的数量
    the_predict_num = the_category_.eq(the_true_target_num).sum().item()
    return the_predict_num / the_true_target_num
    
def performance_val(model):
    model.eval()
    val_en , val_de = get_sentences('val')
    translation_dataset = TranslationDataset(val_en, val_de, tokenizer_en, tokenizer_de)
    dataloader = DataLoader(translation_dataset, batch_size=64, shuffle=True, num_workers=4, collate_fn=custom_collate)
    total_loss = 0
    val_rate_list = []
    with torch.no_grad():
        for batch in dataloader:
            enc_inputs = batch['enc_inputs']
            dec_inputs = batch['dec_inputs']
            dec_output = batch['dec_output']
            enc_inputs = enc_inputs.to(device)
            dec_inputs = dec_inputs.to(device)
            dec_output = dec_output.to(device)
            _,outs = model(enc_inputs, dec_inputs)
            dec_output_reshaped = dec_output.reshape(-1)
            the_true_rate = judge_rate_true(outs,dec_output_reshaped)
            val_rate_list.append(the_true_rate)
            loss = criterion(outs, dec_output_reshaped)
            total_loss += loss.item()
    the_average_val_loss = total_loss / len(dataloader)
    the_average_val_rate = sum(val_rate_list) / len(val_rate_list)
    return the_average_val_loss,the_average_val_rate
        
def lr_lambda(epoch, warmup_epochs,epochs):
    if epoch < warmup_epochs:
        return ((epoch + 1) / warmup_epochs)
    else:
        decay_num = epochs - warmup_epochs
        
        return ((epochs - epoch) / (epochs - warmup_epochs))
 
# 在使用LambdaLR时
# scheduler = LambdaLR(optimizer, lr_lambda=lambda epoch: lr_lambda(epoch, warmup_epochs, max_lr))
 
 
# 将模型转化到GPU上面
model = Transformer(tokenizer_en.get_vocab_size(), tokenizer_de.get_vocab_size(), d_model, d_ff, num_layers, num_heads, device, 0, 0, 0.1)
model = model.to(device)
# 获取使用的dataloader
train_en , train_de = get_sentences('train')
translation_dataset = TranslationDataset(train_en, train_de, tokenizer_en, tokenizer_de)
dataloader = DataLoader(translation_dataset, batch_size=32, shuffle=True, num_workers=4, collate_fn=custom_collate)
# 设置预热标准
# lr_lambda = lambda epoch: min((epoch + 1) / warmup_epochs, 1.0) if epoch < warmup_epochs else max(0.0, (epochs - epoch - 1) / max(1, (epochs - warmup_epochs)))
 
 
# 定义损失函数和优化器
criterion = nn.CrossEntropyLoss(ignore_index=0, label_smoothing=0.1)
optimizer = optim.AdamW(params=model.parameters(), betas=(0.9, 0.98), eps=1e-9,lr=0.01)
scheduler = LambdaLR(optimizer, lr_lambda=lambda epoch: lr_lambda(epoch, warmup_epochs, epochs))
    
 
for epoch in range(epochs):
    total_loss = 0
    the_true_rate_list = []
    for batch in tqdm(dataloader, desc="Processing", unit="item"):
        enc_inputs = batch['enc_inputs']
        dec_inputs = batch['dec_inputs']
        dec_output = batch['dec_output']
        enc_inputs = enc_inputs.to(device)
        dec_inputs = dec_inputs.to(device)
        dec_output = dec_output.to(device)
        optimizer.zero_grad()
        _,outs = model(enc_inputs, dec_inputs)
        dec_output_reshaped = dec_output.reshape(-1)
        loss = criterion(outs, dec_output_reshaped)
        total_loss += loss.item()
        the_true_rate = judge_rate_true(outs,dec_output_reshaped)
        the_true_rate_list.append(the_true_rate)
        loss.backward()
        optimizer.step()
    the_average_rate = sum(the_true_rate_list) / len(the_true_rate_list)
    lr = optimizer.param_groups[-1]['lr']
    scheduler.step()
    average_train_loss = total_loss / len(dataloader)
    print(f'Epoch:{epoch + 1},average_loss:{average_train_loss},lr:{lr},the_average_rate:{the_average_rate}')
    the_average_val_loss,the_average_val_rate = performance_val(model)
    print(f"the_average_val_loss--->{the_average_val_loss};the_average_val_rate===>{the_average_val_rate}")
torch.save(model.state_dict(), 'model_structure.pth')

 测试

在这部分生成序列的其实有很多中算法比如贪婪算法，beamsearch,topk,top-p等等，为了简单实现我在这里使用的是贪婪算法，也就是每次都是去最大可能的那个词。

在训练好模型之后而且已经在验证集上看到了最终模型的效果，所以现在就要在验证集上进一步看一下这个模型到底是怎么样子，在这里使用的是bleu打分。

在测试集上和训练集和验证集不一样的一个地方是在前面其实我们都是通过批量然后得到在字典中的位置之后直接进行损失的计算，但是在这里呢似乎我们使用损失函数并不能很直观的看出来，所以在这里我们使用的是一个句子一个句子放到模型中，然后真正的使用自回归的当时实现一个完整句子的生成。

注意点：

1. 一开始模型预测的时候我们向模型的编码器输入的其实都是一个句子的编码，在解码器输入的应该是前一个生成的词在词典中的位置（这个也就是真正的自回归）。一个是seq，一个是token

2. 虽然是一个词，但是其实我们在输入模型中的应该是这个词以及前面的词组成的张量，也就是应该输入到模型中的应该是不断的变长的

3. 在输入模型中的时候因为我们得到是一个词典长度是字典大小的张量，我们要使用softmax拿到最大位置的索引也就是那个词，然后进行判断，看一下这个词是不是结束符号

4. 因为我们在训练模型是时候放到模型中的是有批量的，所以我们在测试集上要么批量要设置成为1,如果直接循环的话就要添加维度

完整的代码如下：

# 在测试集上进行自回归和blue打分
import torch
import torch.nn.functional as F
 
import sys
sys.path.append('/root/autodl-tmp/mt1/subword/tool/')
 
from model.selftransformer import Transformer
 
from model.config import *
from data_process.build_dataloader import *
from tool.sentences_tool import get_sentences
from tool.tokenizer_tool import tokenizer_de,tokenizer_en
 
from nltk.translate.bleu_score import sentence_bleu
def generate_word(model,sentence,source_tokenizer,target_tokenizer):
    sentence = source_tokenizer.encode(sentence).ids
    start_ids = list(target_tokenizer.encode('<S>').ids)[0]
    end_ids = list(target_tokenizer.encode('<E>').ids)[0]
    translate_list = [start_ids]
    # 对第一个词进行了特殊处理
    input1 = (torch.tensor(sentence).unsqueeze(0)).to(device)
    input2 = (torch.tensor([start_ids]).unsqueeze(0)).to(device)
    enc_out_put,final_out_put = model(input1,input2)
    flag_is = True
    while flag_is:
        real = int((F.softmax(final_out_put[-1],dim=-1).argsort(descending=True))[0])
        if real == end_ids:
            translate_list.append(real)
            flag_is = False
        else:
            translate_list.append(real)
            enc_out_put,final_out_put = model(
                    torch.tensor(sentence).unsqueeze(0).to(device),
                    torch.tensor(translate_list).unsqueeze(0).to(device),
                    torch.tensor(enc_out_put).unsqueeze(0).to(device)
                )
    return target_tokenizer.decode(translate_list)
 
def generate_bleu(source_sentence,target_sentence):
    target_list = target_sentence.split(' ')[1:-2]
    real_target_sentence = ' '.join(target_list)
    bleu_score = sentence_bleu(source_sentence,real_target_sentence,weights=(0.25,0.25,0.25,0.25))
    return bleu_score
 
model = Transformer(tokenizer_en.get_vocab_size(), tokenizer_de.get_vocab_size(), d_model, d_ff, num_layers, num_heads, device, 0, 0, 0.1)
model.load_state_dict(torch.load('model_structure.pth'))
model = model.to(device)
test_en , test_de = get_sentences('test')
generate_word_list = []
for i in test_en:
    words = generate_word(model,i,tokenizer_en,tokenizer_de)
    generate_word_list.append(words)
    break
print('-'*100)
print(len(generate_word_list))
print(generate_word_list)
# blues = []
# for source,target in zip(test_de,generate_word_list):
#     blues.append(generate_bleu(target,source))
 
# average_bleu = sum(blues) / len(blues)

开源位置

transformer_MT_en_de: 了解transformer结构并且了解训练模型的过程

上面的是一个简单的transfomer的从零开始包括数据集构建，模型架构搭建以及测试和训练的过程，这个仅仅是最简单的，据说还有很多的tips，如果有学到会立刻分享给大家。希望大家可以给我的项目点个star