0. 介紹
首先需要指出的是，代碼是從李宏毅老師的課程中下載的，并不是我自己碼的，這篇文章主要是在原代碼中加了一些講解和注釋，以及將繁體字改成了簡體字，
我們需要處理的問題是將Twitter上的文字評論分為正面和負面，具體的要求如下：
我們使用到的模型如下所示：
其中，word embedding是將詞語轉換為向量，以便于后續放入LSTM中進行訓練，在下面的代碼中，作者選用的是word2vec模型（Skip-gram、CBOW等）完成這個轉換，具體的演算法大家可以在CSDN或者B站搜索大佬們的文章來學習，
1. 下載資料
path_prefix = './'
!gdown --id '1lz0Wtwxsh5YCPdqQ3E3l_nbfJT1N13V8' --output data.zip
!unzip data.zip
!ls
# this is for filtering the warnings
import warnings
warnings.filterwarnings('ignore')
2. 讀入資料
因為資料的格式并不是一般的格式，所以需要寫一個自己的讀取函式
import torch
import numpy as np
import pandas as pd
import torch.optim as optim
import torch.nn.functional as F
def load_training_data(path='training_label.txt'):
# 把training時需要的data讀入
# 如果是'training_label.txt'，需要讀取它的label，如果是'training_nolabel.txt'，不需要讀取label（本身也沒有label）
if 'training_label' in path: #判斷training_label這幾個字在不在path中，以判斷需不需要讀取label
#讀入存在txt中文本資料的常用方式
with open(path, 'r') as f:
lines = f.readlines() #一行一行讀入資料
lines = [line.strip('\n').split(' ') for line in lines]
x = [line[2:] for line in lines] #第二列之后是文本資料
y = [line[0] for line in lines] #第一列是label
return x, y
else:
with open(path, 'r') as f:
lines = f.readlines()
x = [line.strip('\n').split(' ') for line in lines]
return x
def load_testing_data(path='testing_data'):
# 把testing時需要的data讀進來
with open(path, 'r') as f:
lines = f.readlines()
X = ["".join(line.strip('\n').split(",")[1:]).strip() for line in lines[1:]]
X = [sen.split(' ') for sen in X]
return X
def evaluation(outputs, labels): #定義自己的評價函式，用分類的準確率來評價
#outputs => probability (float)
#labels => labels
outputs[outputs>=0.5] = 1 # 大於等於0.5為有惡意
outputs[outputs<0.5] = 0 # 小於0.5為無惡意
correct = torch.sum(torch.eq(outputs, labels)).item()
return correct
3. 定義word2vec模型
word2vec模型可以將詞語轉換為向量，并且能很神奇地保留詞語的相似度等性質，具體的演算法流程可以在csdn、知憾訓者B站上搜索大佬們的文章，我們在這里使用word2vec是為了后續將文字轉換為向量，以便于輸入相應的神經網路來學習，（神經網路只認數字不認英文的嘛）
# 這個block是用來訓練word to vector 的 word embedding
# 注意！這個block在訓練word to vector時是用cpu，可能要花到10分鐘以上(我試了一下，確實是要很久)
import os
import numpy as np
import pandas as pd
import argparse
from gensim.models import word2vec
def train_word2vec(x):
# 訓練word to vector 的 word embedding
#size是神經網路的層數，window是視窗長度，min_count是用來忽略那些出現過少的詞語，worker是執行緒數，iter是回圈次數
model = word2vec.Word2Vec(x, size=250, window=5, min_count=5, workers=12, iter=10, sg=1)
return model
if __name__ == "__main__":
print("loading training data ...")
train_x, y = load_training_data('training_label.txt')
train_x_no_label = load_training_data('training_nolabel.txt')
print("loading testing data ...")
test_x = load_testing_data('testing_data.txt')
model = train_word2vec(train_x + train_x_no_label + test_x)
print("saving model ...")
# model.save(os.path.join(path_prefix, 'model/w2v_all.model'))
model.save(os.path.join(path_prefix, 'w2v_all.model')) #將模型保存這一步可以使得后續的訓練更方便，是一個很好的習慣
4. 定義資料預處理類
因為我們要面對的是文本資料，所以必須要進行資料預處理，為了后續的操作方便，作者在這里將其封裝成了一個類，具體包括：
- 把之前訓練好的word2vec模型讀進來，保存訓練好的embedding（這個embedding包含了訓練word2vec模型時使用的各個引數）
- 把"PAD"或"UNK"加進embedding_matrix
- 制作embedding_matrix
- 將輸入的句子的長度變成一致的，方便后續輸入神經網路中
- 實作word2indx，把句子里面的字變成相對應的index
- 將label轉為tensor格式
from torch import nn
from gensim.models import Word2Vec
class Preprocess():
def __init__(self, sentences, sen_len, w2v_path="./w2v.model"): #首先定義類的一些屬性
self.w2v_path = w2v_path
self.sentences = sentences
self.sen_len = sen_len
self.idx2word = []
self.word2idx = {}
self.embedding_matrix = []
def get_w2v_model(self):
# 把之前訓練好的word to vec 模型讀進來
self.embedding = Word2Vec.load(self.w2v_path)
self.embedding_dim = self.embedding.vector_size #embedding的維度就是訓練好的Word2vec中向量的長度
def add_embedding(self, word):
# 把word（"<PAD>"或"<UNK>"）加進embedding，并賦予他一個隨機生成的representation vector
# 因為我們有時候要用到"<PAD>"或"<UNK>"，但它倆本身沒法放到word2vec中訓練而且它倆不需要生成一個能反應其與其他詞關系的向量，故隨機生成
vector = torch.empty(1, self.embedding_dim)#生成空的
torch.nn.init.uniform_(vector)#隨機生成
self.word2idx[word] = len(self.word2idx)#在word2idx放入對應的index
self.idx2word.append(word)#在idx2word中放入對應的word
self.embedding_matrix = torch.cat([self.embedding_matrix, vector], 0)#在embedding_matrix中加入新的vector
def make_embedding(self, load=True):
print("Get embedding ...")
# 取得訓練好的 Word2vec word embedding
if load:
print("loading word to vec model ...")
self.get_w2v_model()
else:
raise NotImplementedError
# 制作一個 word2idx 的 dictionary
# 制作一個 idx2word 的 list
# 制作一個 word2vector 的 list
for i, word in enumerate(self.embedding.wv.vocab):
print('get words #{}'.format(i+1), end='\r')
#e.g. self.word2index['哈'] = 1
#e.g. self.index2word[1] = '哈'
#e.g. self.vectors[1] = '哈' vector
self.word2idx[word] = len(self.word2idx)
self.idx2word.append(word)
self.embedding_matrix.append(self.embedding[word])
print('')
self.embedding_matrix = torch.tensor(self.embedding_matrix)
# 將"<PAD>"和"<UNK>"加進embedding里面
self.add_embedding("<PAD>")
self.add_embedding("<UNK>")
print("total words: {}".format(len(self.embedding_matrix)))
return self.embedding_matrix
def pad_sequence(self, sentence):
# 將每個句子變成一樣的長度
if len(sentence) > self.sen_len: #多的直接截斷
sentence = sentence[:self.sen_len]
else: #少的添加"<PAD>"
pad_len = self.sen_len - len(sentence)
for _ in range(pad_len):
sentence.append(self.word2idx["<PAD>"])
assert len(sentence) == self.sen_len
return sentence
def sentence_word2idx(self):
# 把句子里面的字變成相對應的index
sentence_list = []
for i, sen in enumerate(self.sentences):
print('sentence count #{}'.format(i+1), end='\r')
sentence_idx = []
for word in sen:
if (word in self.word2idx.keys()):
sentence_idx.append(self.word2idx[word])
else:
sentence_idx.append(self.word2idx["<UNK>"])
# 將每個句子變成一樣的長度
sentence_idx = self.pad_sequence(sentence_idx)
sentence_list.append(sentence_idx)
return torch.LongTensor(sentence_list)
def labels_to_tensor(self, y):
# 把labels轉成tensor
y = [int(label) for label in y]
return torch.LongTensor(y)
5. 制作Dataset
這一步相對比較簡單，只是做了Dataset類
# 建立了dataset所需要的'__init__', '__getitem__', '__len__'
# 好讓dataloader能使用
import torch
from torch.utils import data
class TwitterDataset(data.Dataset):
"""
Expected data shape like:(data_num, data_len)
Data can be a list of numpy array or a list of lists
input data shape : (data_num, seq_len, feature_dim)
__len__ will return the number of data
"""
def __init__(self, X, y):
self.data = X
self.label = y
def __getitem__(self, idx):
if self.label is None: return self.data[idx]
return self.data[idx], self.label[idx]
def __len__(self):
return len(self.data)
6. 建立模型
建立我們之后要使用的LSTM模型，主要包括三塊：
- embedding layer
- LSTM
- 全連接神經網路
embedding layer可以理解為將我們的文字進行編碼，以使得LSTM可以看得懂，具體用到的方法就是word2vec模型，
LSTM模型主要需要輸入：
- input_size: 輸入特征維數，即每一行輸入元素的個數
- hidden_size: 隱藏層狀態的維數，即隱藏層節點的個數，這個和單層感知器的結構是類似的，
- num_layers: LSTM 堆疊的層數，默認值是1層，如果設定為2，第二個LSTM接收第一個LSTM的計算結果，
- batch_first: 輸入輸出的第一維是否為 batch_size，默認值 False，因為 Torch 中，人們習慣使用Torch中帶有的dataset，dataloader向神經網路模型連續輸入資料，這里面就有一個 batch_size 的引數，表示一次輸入多少個資料， 在 LSTM 模型中，輸入資料必須是一批資料，為了區分LSTM中的批量資料和dataloader中的批量資料是否相同意義，LSTM 模型就通過這個引數的設定來區分，
- dropout: 默認值0，是否在除最后一個 RNN 層外的其他 RNN 層后面加 dropout 層，
- bidirectional: 是否是雙向 RNN，默認為：false，若為 true，則：num_directions=2，否則為1，
全連接神經網路主要是為了將LSTM的輸出和最終的預測進行一下轉換，
import torch
from torch import nn
class LSTM_Net(nn.Module):
def __init__(self, embedding, embedding_dim, hidden_dim, num_layers, dropout=0.5, fix_embedding=True):
super(LSTM_Net, self).__init__()
# 制作 embedding layer
self.embedding = torch.nn.Embedding(embedding.size(0),embedding.size(1))
self.embedding.weight = torch.nn.Parameter(embedding)#embedding層的引數直接呼叫我們之前用word2vec訓練的embedding里面的引數
# 是否將 embedding fix住，如果fix_embedding為False，在訓練程序中，embedding也會跟著被訓練
self.embedding.weight.requires_grad = False if fix_embedding else True
self.embedding_dim = embedding.size(1)
self.hidden_dim = hidden_dim
self.num_layers = num_layers
self.dropout = dropout
self.lstm = nn.LSTM(embedding_dim, hidden_dim, num_layers=num_layers, batch_first=True)
self.classifier = nn.Sequential( nn.Dropout(dropout),
nn.Linear(hidden_dim, 1),
nn.Sigmoid() )
def forward(self, inputs):
inputs = self.embedding(inputs)
x, _ = self.lstm(inputs, None)
# x 的 dimension (batch, seq_len, hidden_size)
# 取用 LSTM 最后一個的 hidden state（我的理解是最后一個的輸出對于整個文本的理解是最到位的）
x = x[:, -1, :]
x = self.classifier(x)
return x
7. 定義模型訓練函式
這個訓練的程序跟之間的訓練較為相似，注釋給的很詳細，可以通過注釋理解一下，
import torch
from torch import nn
import torch.optim as optim
import torch.nn.functional as F
def training(batch_size, n_epoch, lr, model_dir, train, valid, model, device):
total = sum(p.numel() for p in model.parameters())#總的引數
trainable = sum(p.numel() for p in model.parameters() if p.requires_grad)#需要訓練的引數
print('\nstart training, parameter total:{}, trainable:{}\n'.format(total, trainable))#看看模型的引數
model.train() # 將model的模式設為train，這樣optimizer就可以更新model的引數
criterion = nn.BCELoss() # 定義損失函數，這里我們使用binary cross entropy loss
t_batch = len(train)
v_batch = len(valid)
optimizer = optim.Adam(model.parameters(), lr=lr) # 將模型的引數傳給optimizer，并賦予適當的learning rate
total_loss, total_acc, best_acc = 0, 0, 0
for epoch in range(n_epoch):
total_loss, total_acc = 0, 0
# 做training
for i, (inputs, labels) in enumerate(train):
inputs = inputs.to(device, dtype=torch.long) # device為"cuda"，將inputs變成torch.cuda.LongTensor
labels = labels.to(device, dtype=torch.float) # device為"cuda"，將labels變成torch.cuda.FloatTensor，因為等等要放入criterion，所以型別要是float
optimizer.zero_grad() # 由于loss.backward()的gradient會累加，所以每次做完一個batch后需要調零
outputs = model(inputs) # 將input餵給模型
outputs = outputs.squeeze() # 去掉最外面的dimension，好讓outputs可以放入criterion()
loss = criterion(outputs, labels) # 計算此時模型的training loss
loss.backward() # 算loss的gradient
optimizer.step() # 更新訓練模型的參數
correct = evaluation(outputs, labels) # 計算此時模型的training accuracy
total_acc += (correct / batch_size)
total_loss += loss.item()
print('[ Epoch{}: {}/{} ] loss:{:.3f} acc:{:.3f} '.format(
epoch+1, i+1, t_batch, loss.item(), correct*100/batch_size), end='\r')
print('\nTrain | Loss:{:.5f} Acc: {:.3f}'.format(total_loss/t_batch, total_acc/t_batch*100))
# 這段做validation
model.eval() # 將model的模式設為eval，這樣model的引數就會固定住
with torch.no_grad():
total_loss, total_acc = 0, 0
for i, (inputs, labels) in enumerate(valid):
inputs = inputs.to(device, dtype=torch.long)
labels = labels.to(device, dtype=torch.float)
outputs = model(inputs)
outputs = outputs.squeeze()
loss = criterion(outputs, labels)
correct = evaluation(outputs, labels)
total_acc += (correct / batch_size)
total_loss += loss.item()
print("Valid | Loss:{:.5f} Acc: {:.3f} ".format(total_loss/v_batch, total_acc/v_batch*100))
if total_acc > best_acc:
# 如果validation的結果好于之前所有的結果，就把當下的模型存下來以便后續的預測使用
best_acc = total_acc
#torch.save(model, "{}/val_acc_{:.3f}.model".format(model_dir,total_acc/v_batch*100))
torch.save(model, "{}/ckpt.model".format(model_dir))
print('saving model with acc {:.3f}'.format(total_acc/v_batch*100))
print('-----------------------------------------------')
model.train() # 將model的模式設為train，這樣optimizer就可以更新model的參數（因為剛剛轉為eval模式）
8.定義模型測驗函式
import torch
from torch import nn
import torch.optim as optim
import torch.nn.functional as F
def testing(batch_size, test_loader, model, device):
model.eval()
ret_output = []
with torch.no_grad():
for i, inputs in enumerate(test_loader):
inputs = inputs.to(device, dtype=torch.long)
outputs = model(inputs)
outputs = outputs.squeeze()
outputs[outputs>=0.5] = 1 # 大於等於0.5為負面
outputs[outputs<0.5] = 0 # 小於0.5為正面
ret_output += outputs.int().tolist()
return ret_output
9. 呼叫之前的各個函式開始訓練
- 整理好各個data的路徑
- 定義句子長度、要不要固定embedding、batch大小、要訓練的輪數epoch、learning rate的值、model的資料保存路徑
- 讀入資料
- input和labels做預處理
- 制作一個model的物件
- 把data分為training data和validation data(將一部分training data拿去當做validation data)
- 把data做成dataset供dataloader取用
- 把data 轉成 batch of tensors
- 開始訓練
import os
import torch
import argparse
import numpy as np
from torch import nn
from gensim.models import word2vec
from sklearn.model_selection import train_test_split
# 通過torch.cuda.is_available()的回傳值進行判斷是否有使用GPU的環境，如果有的話device就設為"cuda"，沒有的話就設為"cpu"
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
# 整理好各個data的路徑
train_with_label = os.path.join(path_prefix, 'training_label.txt')
train_no_label = os.path.join(path_prefix, 'training_nolabel.txt')
testing_data = os.path.join(path_prefix, 'testing_data.txt')
w2v_path = os.path.join(path_prefix, 'w2v_all.model') # 整理word2vec模型的路徑
# 定義句子長度、要不要固定embedding、batch大小、要訓練的輪數epoch、learning rate的值、model的資料保存路徑
sen_len = 30
fix_embedding = True # fix embedding during training
batch_size = 128
epoch = 5
lr = 0.001
# model_dir = os.path.join(path_prefix, 'model/') # model directory for checkpoint model
model_dir = path_prefix # model directory for checkpoint model
print("loading data ...") # 把'training_label.txt'和'training_nolabel.txt'讀進來
train_x, y = load_training_data(train_with_label)
train_x_no_label = load_training_data(train_no_label)
# 對input和labels做預處理
preprocess = Preprocess(train_x, sen_len, w2v_path=w2v_path)
embedding = preprocess.make_embedding(load=True)
train_x = preprocess.sentence_word2idx()
y = preprocess.labels_to_tensor(y)
# 制作一個model的物件
model = LSTM_Net(embedding, embedding_dim=250, hidden_dim=250, num_layers=1, dropout=0.5, fix_embedding=fix_embedding)
model = model.to(device) # device為"cuda"，model使用GPU來訓練(放入的inputs也需要是cuda tensor)
# 把data分為training data和validation data(將一部分training data拿去當做validation data)
X_train, X_val, y_train, y_val = train_x[:190000], train_x[190000:], y[:190000], y[190000:]
# 把data做成dataset供dataloader取用
train_dataset = TwitterDataset(X=X_train, y=y_train)
val_dataset = TwitterDataset(X=X_val, y=y_val)
# 把data 轉成 batch of tensors
train_loader = torch.utils.data.DataLoader(dataset = train_dataset,
batch_size = batch_size,
shuffle = True,
num_workers = 8)
val_loader = torch.utils.data.DataLoader(dataset = val_dataset,
batch_size = batch_size,
shuffle = False,
num_workers = 8)
# 開始訓練
training(batch_size, epoch, lr, model_dir, train_loader, val_loader, model, device)
訓練結果如下：
10. 進行預測并保存結果
print("loading testing data ...")
test_x = load_testing_data(testing_data)
preprocess = Preprocess(test_x, sen_len, w2v_path=w2v_path)
embedding = preprocess.make_embedding(load=True)
test_x = preprocess.sentence_word2idx()
test_dataset = TwitterDataset(X=test_x, y=None)
test_loader = torch.utils.data.DataLoader(dataset = test_dataset,
batch_size = batch_size,
shuffle = False,
num_workers = 8)
print('\nload model ...')
model = torch.load(os.path.join(model_dir, 'ckpt.model'))
outputs = testing(batch_size, test_loader, model, device)
# 保存到csv里面
tmp = pd.DataFrame({"id":[str(i) for i in range(len(test_x))],"label":outputs})
print("save csv ...")
tmp.to_csv(os.path.join(path_prefix, 'predict.csv'), index=False)
print("Finish Predicting")
轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/344097.html
標籤：AI