Pytorch-基於Transformer的情感分類

Transformer模型（文本分類僅用到Encoder部分）：

1.數據預處理

和上一個博客https://www.cnblogs.com/cxq1126/p/13504437.html中的數據和預處理基本都一致。

import numpy as np
import torch
from torch import nn, optim
import torch.nn.functional as F
from torchtext import data
from torch.autograd import Variable

import math
import time
import copy
import random

SEED = 126
BATCH_SIZE = 128
EMBEDDING_DIM = 100
LEARNING_RATE = 1e-3

#為了保證實驗結果可以復現，我們經常會把各種random seed固定在某一個值
random.seed(SEED)
np.random.seed(SEED)
torch.manual_seed(SEED)

TEXT = data.Field(tokenize=lambda x: x.split(), batch_first=True, lower=True)
LABEL = data.LabelField(dtype=torch.float)    

#get_dataset構造並返回Dataset所需的examples和fields    
def get_dataset(corpur_path, text_field, label_field):
    fields = [('text', text_field), ('label', label_field)]             #torchtext中於文件配對關系
    examples = []
   
    with open(corpur_path) as f:
        #解析html格式
        li = []
        while True:
            content = f.readline().replace('\n', '')
            if not content:              #為空行，表示取完一次數據（一次的數據保存在li中）
                if not li:               #如果列表也為空，則表示數據讀完，結束循環
                    break
                label = li[0][10]
                text = li[1][6:-7]
                examples.append(data.Example.fromlist([text, label], fields))
                li = []
            else:
                li.append(content)       #["<Polarity>標簽</Polarity>", "<text>句子內容</text>"]  
                
    return examples, fields

#得到構建Dataset所需的examples和fields
train_examples, train_fields = get_dataset("corpus//trains.txt", TEXT, LABEL)
dev_examples, dev_fields = get_dataset("corpus//dev.txt", TEXT, LABEL)
test_examples, test_fields = get_dataset("corpus//tests.txt", TEXT, LABEL)


#構建Dataset數據集
train_data = data.Dataset(train_examples, train_fields)
dev_data = data.Dataset(dev_examples, dev_fields)
test_data = data.Dataset(test_examples, test_fields)


print('len of train data:', len(train_data))              #1000
print('len of dev data:', len(dev_data))                  #200
print('len of test data:', len(test_data))                #300

print(train_data.examples[15].text)
print(train_data.examples[15].label)


#創建vocabulary
TEXT.build_vocab(train_data, max_size=5000, vectors='glove.6B.100d')
LABEL.build_vocab(train_data)
print(len(TEXT.vocab))                     #3287
print(TEXT.vocab.itos[:12])                #['<unk>', '<pad>', 'the', 'and', 'a', 'to', 'is', 'was', 'i', 'of', 'for', 'in']
print(TEXT.vocab.stoi['like'])             #43
print(LABEL.vocab.stoi)                    #defaultdict(None, {'0': 0, '1': 1})


#創建iterators，每個itartion都會返回一個batch的examples
train_iterator, dev_iterator, test_iterator = data.BucketIterator.splits(
    (train_data, dev_data, test_data), 
    batch_size=BATCH_SIZE,
    sort = False)

2.定義模型

2.1Embedding

class InputEmbeddings(nn.Module):
    
    def __init__(self, vocab_size, embedding_dim):               
        super(InputEmbeddings, self).__init__()
        self.embedding_dim = embedding_dim
        self.embed = nn.Embedding(vocab_size, embedding_dim)
        
    def forward(self, x):
        return self.embed(x) * math.sqrt(self.embedding_dim)

2.2PositionalEncoding

class PositionalEncoding(nn.Module):
        
    def __init__(self, embedding_dim, dropout, max_len=5000):
        super(PositionalEncoding, self).__init__()
        self.dropout = nn.Dropout(p=dropout)
    
        pe = torch.zeros(max_len, embedding_dim)
        
        position = torch.arange(0., max_len).unsqueeze(1)      #[max_len, 1]
        div_term = torch.exp(torch.arange(0., embedding_dim, 2) * -(math.log(10000.0) / embedding_dim))
        
        pe[:, 0::2] = torch.sin(position * div_term)
        pe[:, 1::2] = torch.cos(position * div_term)
        pe = pe.unsqueeze(0)
        self.register_buffer('pe', pe)              #內存中定一個常量，模型保存和加載的時候可以寫入和讀出
        
    def forward(self, x):
        x = x + Variable(self.pe[:, :x.size(1)], requires_grad=False)  #Embedding+PositionalEncoding
        return self.dropout(x)

2.3MultiHeadAttention

def clones(module, N):
    return nn.ModuleList([copy.deepcopy(module) for _ in range(N)])
    
    
def attention(query, key, value, mask=None, dropout=None):   #q,k,v:[batch, h, seq_len, d_k]
    
    d_k = query.size(-1)                                                    #query的維度
    scores = torch.matmul(query, key.transpose(-2, -1)) / math.sqrt(d_k)    #打分機制 [batch, h, seq_len, seq_len]
    if mask is not None: 
        scores = scores.masked_fill(mask == 0, -1e9)       #mask==0的內容填充-1e9，使計算softmax時概率接近0
    p_attn = F.softmax(scores, dim = -1)                   #對最后一個維度歸一化得分 [batch, h, seq_len, seq_len]
    
    if dropout is not None:
        p_attn = dropout(p_attn)
    return torch.matmul(p_attn, value), p_attn             #[batch, h, seq_len, d_k]


class MultiHeadedAttention(nn.Module):
    
    def __init__(self, h, embedding_dim, dropout=0.1):
        
        super(MultiHeadedAttention, self).__init__()
        assert embedding_dim % h == 0   
        
        self.d_k = embedding_dim // h   #將embedding_dim分割成h份后的維度
        self.h = h                      #h指的是head數量
        self.linears = clones(nn.Linear(embedding_dim, embedding_dim), 4)
        self.attn = None
        self.dropout = nn.Dropout(p=dropout)
        
    def forward(self, query, key, value, mask=None):      #q,k,v:[batch, seq_len, embedding_dim]
        
        if mask is not None:
            mask = mask.unsqueeze(1)                      #[batch, seq_len, 1] 
        nbatches = query.size(0)
        
        # 1) Do all the linear projections in batch from embedding_dim => h x d_k  
        # [batch, seq_len, h, d_k] -> [batch, h, seq_len, d_k]
        query, key, value = [l(x).view(nbatches, -1, self.h, self.d_k).transpose(1, 2)            
                             for l, x in zip(self.linears, (query, key, value))]
        
        # 2) Apply attention on all the projected vectors in batch. 
        x, self.attn = attention(query, key, value, mask=mask, dropout=self.dropout)   #x:[batch, h, seq_len, d_k], attn:[batch, h, seq_len, seq_len]
        
        # 3) "Concat" using a view and apply a final linear. 
        x = x.transpose(1, 2).contiguous().view(nbatches, -1, self.h * self.d_k)       #[batch, seq_len, embedding_dim]
        return self.linears[-1](x)

2.4MyTransformerModel

class MyTransformerModel(nn.Module):
    
    def __init__(self, vocab_size, embedding_dim, p_drop, h, output_size):
        super(MyTransformerModel,self).__init__()
        self.drop = nn.Dropout(p_drop)
              
        self.embeddings = InputEmbeddings(vocab_size, embedding_dim)
        self.position = PositionalEncoding(embedding_dim, p_drop)
        self.attn = MultiHeadedAttention(h, embedding_dim) 
        self.norm = nn.LayerNorm(embedding_dim)
        self.linear = nn.Linear(embedding_dim, output_size)
        self.init_weights()
          
    def init_weights(self):
        initrange = 0.1
        self.linear.bias.data.zero_()
        self.linear.weight.data.uniform_(-initrange, initrange)

    def forward(self,inputs,mask): #維度均為[batch, seq_len]
        
        embeded = self.embeddings(inputs)        #1.InputEmbedding [batch, seq_len, embedding_dim]
        
        embeded = self.position(embeded)         #2.PostionalEncoding [batch, seq_len, embedding_dim]
        
        mask = mask.unsqueeze(2)                 #[batch,seq_len,1]
        
        inp_attn = self.attn(embeded, embeded, embeded, mask)    #3.1MultiHeadedAttention [batch, seq_len, embedding_dim]
        inp_attn = self.norm(inp_attn + embeded)                 #3.2LayerNorm
        
        inp_attn = inp_attn * mask                               #4. linear [batch, seq_len, embedding_dim]
        
        h_avg = inp_attn.sum(1)/(mask.sum(1) + 1e-5)             #[batch, embedding_dim]
        return self.linear(h_avg).squeeze()                      #[batch, 1] -> [batch]
   

使用模型，使用預訓練過的embedding來替換隨機初始化（nn.Embedding），定義優化器、損失函數。

model = MyTransformerModel(len(TEXT.vocab), EMBEDDING_DIM, p_drop=0.5, h=2, output_size=1)

pretrained_embedding = TEXT.vocab.vectors
print('pretrained_embedding:', pretrained_embedding.shape)      #torch.Size([3287, 100])
model.embeddings.embed.weight.data.copy_(pretrained_embedding)  #embeddings是MyTransformerModel的參數, embed是InputEmbedding的參數
print('embedding layer inited.')

optimizer = optim.Adam(model.parameters(), lr=1e-3, weight_decay=0.001)
criteon = nn.BCEWithLogitsLoss() 

3.訓練、評估函數

常規套路：計算准確率、訓練函數、評估函數、打印模型表現、用保存的模型參數預測測試數據。

#計算准確率
def binary_acc(preds, y):

    preds = torch.round(torch.sigmoid(preds))
    correct = torch.eq(preds, y).float()
    acc = correct.sum() / len(correct)
    return acc    


#訓練函數
def train(model, iterator, optimizer, criteon):
    
    avg_loss = []
    avg_acc = []
    model.train()        #表示進入訓練模式
    
    for i, batch in enumerate(iterator):
        
        mask = 1 - (batch.text == TEXT.vocab.stoi['<pad>']).float()   #[batch, seq_len]增加了這句，其他都一樣
        pred = model(batch.text, mask)            
                    
        loss = criteon(pred, batch.label)
        acc = binary_acc(pred, batch.label).item()   #計算每個batch的准確率
        avg_loss.append(loss.item())
        avg_acc.append(acc)
              
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        
    avg_acc = np.array(avg_acc).mean()
    avg_loss = np.array(avg_loss).mean()
    return avg_loss, avg_acc                          
    
    
#評估函數
def evaluate(model, iterator, criteon):    
    
    avg_loss = []
    avg_acc = []    
    model.eval()         #表示進入測試模式
    
    with torch.no_grad():
        for batch in iterator:
            mask = 1 - (batch.text == TEXT.vocab.stoi['<pad>']).float()
            pred = model(batch.text, mask)      
           
            loss = criteon(pred, batch.label)
            acc = binary_acc(pred, batch.label).item()
            avg_loss.append(loss.item())
            avg_acc.append(acc)
        
    avg_loss = np.array(avg_loss).mean()
    avg_acc = np.array(avg_acc).mean()
    return avg_loss, avg_acc


#訓練模型，並打印模型的表現
best_valid_acc = float('-inf')    

for epoch in range(30):
    
    start_time = time.time()
   
    train_loss, train_acc = train(model, train_iterator, optimizer, criteon)
    dev_loss, dev_acc = evaluate(model, dev_iterator, criteon)  
    
    end_time = time.time()
    
    epoch_mins, epoch_secs = divmod(end_time - start_time, 60)
    
    if dev_acc > best_valid_acc:          #只要模型效果變好，就保存
        best_valid_acc = dev_acc
        torch.save(model.state_dict(), 'wordavg-model.pt')
        
    print(f'Epoch: {epoch+1:02} | Epoch Time: {epoch_mins}m {epoch_secs:.2f}s')
    print(f'\tTrain Loss: {train_loss:.3f} | Train Acc: {train_acc*100:.2f}%')
    print(f'\t Val. Loss: {dev_loss:.3f} |  Val. Acc: {dev_acc*100:.2f}%')
   

#用保存的模型參數預測數據
model.load_state_dict(torch.load("wordavg-model.pt"))   
test_loss, test_acc = evaluate(model, test_iterator, criteon)
print(f'Test. Loss: {test_loss:.3f} |  Test. Acc: {test_acc*100:.2f}%')

結果並沒有提升很多，可能因為數據量小，句子比較短：