導(dǎo)讀:本文主要解析Pytorch Tutorial中BiLSTM_CRF代碼,幾乎注釋了每行代碼,希望本文能夠幫助大家理解這個tutorial,除此之外借助代碼和圖解也對理解條件隨機(jī)場(CRF)會有一定幫助,因為這個tutorial代碼主要還是在實現(xiàn)CRF部分。
1 知識準(zhǔn)備
在閱讀tutorial前,需具備一些理論或知識基礎(chǔ),包括LSTM單元、BiLSTM-CRF模型、CRF原理以及一些代碼中的函數(shù)使用,參考資料中涵蓋了主要的涉及知識,可配合tutorial一同學(xué)習(xí)。
2 理解CRF中歸一化因子Z(x)的計算
條件隨機(jī)場中的Z(x)表示歸一化因子,它是一個句子所有可能標(biāo)記tag序列路徑的得分總和。一般的,我們會有一個直接的想法,就是列舉出所有可能的路徑,計算出每條路徑的得分之后再加和。如上圖中的例子所示,有5個字符和5個tag,如果按照上述的暴力窮舉法進(jìn)行計算,就有種路徑組合,而在我們的實際工作中,可能會有更長的序列和更多的tag標(biāo)簽,此時暴力窮舉法未免顯得有些效率低下。于是我們考慮采用分?jǐn)?shù)累積的方式進(jìn)行所有路徑得分總和的計算,即先計算出到達(dá)的所有路徑的總得分,然后計算-》的所有路徑的得分,然后依次計算-》。..-》間的所有路徑的得分,最后便得到了我們的得分總和,這個思路源于如下等價等式:
上式相等表明,直接計算整個句子序列的全局分?jǐn)?shù)與計算每一步的路徑得分再加和等價,計算每一步的路徑得分再加和這種方式可以大大減少計算的時間,故Pytorch Tutorial中的_forward_alg()函數(shù)據(jù)此實現(xiàn)。這種計算每一步的路徑得分再加和的方法還可以以下圖方式進(jìn)行計算。
如上圖所示,在每個時間步上,比如’word==去‘這一列,每一個tag處(0~6豎框是tag的id),關(guān)注兩個值:前一個時間步上所有tag到當(dāng)前tag中總得分最大值以及該最大值對應(yīng)的前一個時間步上tag的id。這樣一來每個tag都記錄了它前一個時間步上到自己的最優(yōu)路徑,最后通過tag的id進(jìn)行回溯,這樣就可以得到最終的最優(yōu)tag標(biāo)記序列。此部分對應(yīng)Pytorch Tutorial中的_viterbi_decode()函數(shù)實現(xiàn)。
4 理解log_sum_exp()函數(shù)
Pytorch Tutorial中的log_sum_exp()函數(shù)最后返回的計算方式數(shù)學(xué)推導(dǎo)如下:
5 Pytorch Tutorial代碼部分注釋輔助理解
import torch
import torch.nn as nn
import torch.optim as optim
# 人工設(shè)定隨機(jī)種子以保證相同的初始化參數(shù),使模型可復(fù)現(xiàn)
torch.manual_seed(1)
# 得到每行最大值索引idx
def argmax(vec):
# 得到每行最大值索引idx
_, idx = torch.max(vec, 1)
# 返回每行最大值位置索引
return idx.item()
# 將序列中的字轉(zhuǎn)化為數(shù)字(int)表示
def prepare_sequence(seq, to_ix):
# 將序列中的字轉(zhuǎn)化為數(shù)字(int)表示
idx = [to_ix[c] for c in seq]
return torch.tensor(idx, dtype=torch.long)
# 前向算法是不斷積累之前的結(jié)果,這樣就會有個缺點
# 指數(shù)和積累到一定程度之后,會超過計算機(jī)浮點值的最大值
# 變成inf,這樣取log后也是inf
# 為了避免這種情況,用一個合適的值clip=max去提指數(shù)和的公因子
# 這樣不會使某項變得過大而無法計算
def log_sum_exp(vec):# vec:形似[[tag個元素]]
# 取vec中最大值
max_score = vec[0, argmax(vec)]
# vec.size()[1]:tag數(shù)
max_score_broadcast = max_score.view(1, -1).expand(1, vec.size()[1])
# 里面先做減法,減去最大值可以避免e的指數(shù)次,計算機(jī)上溢
# 等同于torch.log(torch.sum(torch.exp(vec))),防止e的指數(shù)導(dǎo)致計算機(jī)上溢
return max_score + torch.log(torch.sum(torch.exp(vec - max_score_broadcast)))
class BiLSTM_CRF(nn.Module):
# 初始化參數(shù)
def __init__(self, vocab_size, tag_to_ix, embedding_dim, hidden_dim):
super(BiLSTM_CRF, self).__init__()
# 詞嵌入維度
self.embedding_dim = embedding_dim
# BiLSTM 隱藏層維度
self.hidden_dim = hidden_dim
# 詞典的大小
self.vocab_size = vocab_size
# tag到數(shù)字的映射
self.tag_to_ix = tag_to_ix
# tag個數(shù)
self.tagset_size = len(tag_to_ix)
# num_embeddings (int):vocab_size 詞典的大小
# embedding_dim (int):embedding_dim 嵌入向量的維度,即用多少維來表示一個符號
self.word_embeds = nn.Embedding(vocab_size, embedding_dim)
# input_size: embedding_dim 輸入數(shù)據(jù)的特征維數(shù),通常就是embedding_dim(詞向量的維度)
# hidden_size: hidden_dim LSTM中隱藏層的維度
# num_layers:循環(huán)神經(jīng)網(wǎng)絡(luò)的層數(shù)
# 默認(rèn)使用偏置,默認(rèn)不用dropout
# bidirectional = True 用雙向LSTM
# 設(shè)定為單層雙向
# 隱藏層設(shè)定為指定維度的一半,便于后期拼接
# // 表示整數(shù)除法,返回不大于結(jié)果的一個最大的整數(shù)
self.lstm = nn.LSTM(embedding_dim, hidden_dim // 2,
num_layers=1, bidirectional=True)
# 將BiLSTM提取的特征向量映射到特征空間,即經(jīng)過全連接得到發(fā)射分?jǐn)?shù)
# in_features: hidden_dim 每個輸入樣本的大小
# out_features:tagset_size 每個輸出樣本的大小
self.hidden2tag = nn.Linear(hidden_dim, self.tagset_size)
# 轉(zhuǎn)移矩陣的參數(shù)初始化,transition[i,j]代表的是從第j個tag轉(zhuǎn)移到第i個tag的轉(zhuǎn)移分?jǐn)?shù)
self.transitions = nn.Parameter(torch.randn(self.tagset_size, self.tagset_size))
# 初始化所有其他tag轉(zhuǎn)移到START_TAG的分?jǐn)?shù)非常小,即不可能由其他tag轉(zhuǎn)移到START_TAG
# 初始化STOP_TAG轉(zhuǎn)移到所有其他的分?jǐn)?shù)非常小,即不可能有STOP_TAG轉(zhuǎn)移到其他tag
# CRF的轉(zhuǎn)移矩陣,T[i,j]表示從j標(biāo)簽轉(zhuǎn)移到i標(biāo)簽,
self.transitions.data[tag_to_ix[START_TAG], :] = -10000
self.transitions.data[:, tag_to_ix[STOP_TAG]] = -10000
# 初始化LSTM的參數(shù)
self.hidden = self.init_hidden()
# 使用隨機(jī)正態(tài)分布初始化LSTM的h0和c0
# 否則模型自動初始化為零值,維度為[num_layers*num_directions, batch_size, hidden_dim]
def init_hidden(self):
return (torch.randn(2, 1, self.hidden_dim // 2),
torch.randn(2, 1, self.hidden_dim // 2))
# 計算歸一化因子Z(x)
def _forward_alg(self, feats):
‘’‘
輸入:發(fā)射矩陣(emission score),實際上就是LSTM的輸出
sentence的每個word經(jīng)BiLSTM后對應(yīng)于每個label的得分
輸出:所有可能路徑得分之和/歸一化因子/配分函數(shù)/Z(x)
’‘’
# 通過前向算法遞推計算
# 初始化1行 tagset_size列的嵌套列表
init_alphas = torch.full((1, self.tagset_size), -10000.)
# 初始化step 0 即START位置的發(fā)射分?jǐn)?shù),START_TAG取0其他位置取-10000
init_alphas[0][self.tag_to_ix[START_TAG]] = 0.
# 包裝到一個變量里面以便自動反向傳播
forward_var = init_alphas
# 迭代整個句子
# feats:形似[[。..。], 每個字映射到tag的發(fā)射概率,
# [。..。],
# [。..。]]
for feat in feats:
# 存儲當(dāng)前時間步下各tag得分
alphas_t = []
for next_tag in range(self.tagset_size):
# 取出當(dāng)前tag的發(fā)射分?jǐn)?shù)(與之前時間步的tag無關(guān)),擴(kuò)展成tag維
emit_score = feat[next_tag].view(1, -1).expand(1, self.tagset_size)
# 取出當(dāng)前tag由之前tag轉(zhuǎn)移過來的轉(zhuǎn)移分?jǐn)?shù)
trans_score = self.transitions[next_tag].view(1, -1)
# 當(dāng)前路徑的分?jǐn)?shù):之前時間步分?jǐn)?shù)+轉(zhuǎn)移分?jǐn)?shù)+發(fā)射分?jǐn)?shù)
next_tag_var = forward_var + trans_score + emit_score
# 對當(dāng)前分?jǐn)?shù)取log-sum-exp
alphas_t.append(log_sum_exp(next_tag_var).view(1))
# 更新forward_var 遞推計算下一個時間步
# torch.cat 默認(rèn)按行添加
forward_var = torch.cat(alphas_t).view(1, -1)
# 考慮最終轉(zhuǎn)移到STOP_TAG
terminal_var = forward_var + self.transitions[self.tag_to_ix[STOP_TAG]]
# 對當(dāng)前分?jǐn)?shù)取log-sum-exp
scores = log_sum_exp(terminal_var)
return scores
# 通過BiLSTM提取特征
def _get_lstm_features(self, sentence):
# 初始化LSTM的h0和c0
self.hidden = self.init_hidden()
# 使用之前構(gòu)造的詞嵌入為語句中每個詞(word_id)生成向量表示
# 并將shape改為[seq_len, 1(batch_size), embedding_dim]
embeds = self.word_embeds(sentence).view(len(sentence), 1, -1)
# LSTM網(wǎng)絡(luò)根據(jù)輸入的詞向量和初始狀態(tài)h0和c0
# 計算得到輸出結(jié)果lstm_out和最后狀態(tài)hn和cn
lstm_out, self.hidden = self.lstm(embeds, self.hidden)
lstm_out = lstm_out.view(len(sentence), self.hidden_dim)
# 轉(zhuǎn)換為詞 - 標(biāo)簽([seq_len, tagset_size])表
# 可以看作為每個詞被標(biāo)注為對應(yīng)標(biāo)簽的得分情況,即維特比算法中的發(fā)射矩陣
lstm_feats = self.hidden2tag(lstm_out)
return lstm_feats
# 計算一個tag序列路徑的得分
def _score_sentence(self, feats, tags):
# feats發(fā)射分?jǐn)?shù)矩陣
# 計算給定tag序列的分?jǐn)?shù),即一條路徑的分?jǐn)?shù)
score = torch.zeros(1)
# tags前面補(bǔ)上一個句首標(biāo)簽便于計算轉(zhuǎn)移得分
tags = torch.cat([torch.tensor([self.tag_to_ix[START_TAG]], dtype=torch.long), tags])
# 循環(huán)用于計算給定tag序列的分?jǐn)?shù)
for i, feat in enumerate(feats):
# 遞推計算路徑分?jǐn)?shù):轉(zhuǎn)移分?jǐn)?shù)+發(fā)射分?jǐn)?shù)
# T[i,j]表示j轉(zhuǎn)移到i
score = score + self.transitions[tags[i + 1], tags[i]] + feat[tags[i + 1]]
# 加上轉(zhuǎn)移到句尾的得分,便得到了gold_score
score = score + self.transitions[self.tag_to_ix[STOP_TAG], tags[-1]]
return score
# veterbi解碼,得到最優(yōu)tag序列
def _viterbi_decode(self, feats):
‘’‘
:param feats: 發(fā)射分?jǐn)?shù)矩陣
’‘’
# 便于之后回溯最優(yōu)路徑
backpointers = []
# 初始化viterbi的forward_var變量
init_vvars = torch.full((1, self.tagset_size), -10000.)
init_vvars[0][self.tag_to_ix[START_TAG]] = 0
# forward_var表示每個標(biāo)簽的前向狀態(tài)得分,即上一個詞被打作每個標(biāo)簽的對應(yīng)得分值
forward_var = init_vvars
# 遍歷每個時間步時的發(fā)射分?jǐn)?shù)
for feat in feats:
# 記錄當(dāng)前詞對應(yīng)每個標(biāo)簽的最優(yōu)轉(zhuǎn)移結(jié)點
# 保存當(dāng)前時間步的回溯指針
bptrs_t = []
# 與bptrs_t對應(yīng),記錄對應(yīng)的最優(yōu)值
# 保存當(dāng)前時間步的viterbi變量
viterbivars_t = []
# 遍歷每個標(biāo)簽,求得當(dāng)前詞被打作每個標(biāo)簽的得分
# 并將其與當(dāng)前詞的發(fā)射矩陣feat相加,得到當(dāng)前狀態(tài),即下一個詞的前向狀態(tài)
for next_tag in range(self.tagset_size):
# transitions[next_tag]表示每個標(biāo)簽轉(zhuǎn)移到next_tag的轉(zhuǎn)移得分
# forward_var表示每個標(biāo)簽的前向狀態(tài)得分,即上一個詞被打作每個標(biāo)簽的對應(yīng)得分值
# 二者相加即得到當(dāng)前詞被打作next_tag的所有可能得分
# 維特比算法記錄最優(yōu)路徑時只考慮上一步的分?jǐn)?shù)以及上一步的tag轉(zhuǎn)移到當(dāng)前tag的轉(zhuǎn)移分?jǐn)?shù)
# 并不取決于當(dāng)前的tag發(fā)射分?jǐn)?shù)
next_tag_var = forward_var + self.transitions[next_tag]
# 得到上一個可能的tag到當(dāng)前tag中得分最大值的tag位置索引id
best_tag_id = argmax(next_tag_var)
# 將最優(yōu)tag的位置索引存入bptrs_t
bptrs_t.append(best_tag_id)
# 添加最優(yōu)tag位置索引對應(yīng)的值
viterbivars_t.append(next_tag_var[0][best_tag_id].view(1))
# 更新forward_var = 當(dāng)前詞的發(fā)射分?jǐn)?shù)feat + 前一個最優(yōu)tag當(dāng)前tag的狀態(tài)下的得分
forward_var = (torch.cat(viterbivars_t) + feat).view(1, -1)
# 回溯指針記錄當(dāng)前時間步各個tag來源前一步的最優(yōu)tag
backpointers.append(bptrs_t)
# forward_var表示每個標(biāo)簽的前向狀態(tài)得分
# 加上轉(zhuǎn)移到句尾標(biāo)簽STOP_TAG的轉(zhuǎn)移得分
terminal_var = forward_var + self.transitions[self.tag_to_ix[STOP_TAG]]
# 得到標(biāo)簽STOP_TAG前一個時間步的最優(yōu)tag位置索引
best_tag_id = argmax(terminal_var)
# 得到標(biāo)簽STOP_TAG當(dāng)前最優(yōu)tag對應(yīng)的分?jǐn)?shù)值
path_score = terminal_var[0][best_tag_id]
# 根據(jù)過程中存儲的轉(zhuǎn)移路徑結(jié)點,反推最優(yōu)轉(zhuǎn)移路徑
# 通過回溯指針解碼出最優(yōu)路徑
best_path = [best_tag_id]
# best_tag_id作為線頭,反向遍歷backpointers找到最優(yōu)路徑
for bptrs_t in reversed(backpointers):
best_tag_id = bptrs_t[best_tag_id]
best_path.append(best_tag_id)
# 去除START_TAG
start = best_path.pop()
# 最初的轉(zhuǎn)移結(jié)點一定是人為構(gòu)建的START_TAG,刪除,并根據(jù)這一點確認(rèn)路徑正確性
assert start == self.tag_to_ix[START_TAG]
# 最后將路徑倒序即得到從頭開始的最優(yōu)轉(zhuǎn)移路徑best_path
best_path.reverse()
return path_score, best_path
# 損失函數(shù)loss
def neg_log_likelihood(self, sentence, tags):
# 得到句子對應(yīng)的發(fā)射分?jǐn)?shù)矩陣
feats = self._get_lstm_features(sentence)
# 通過前向算法得到歸一化因子Z(x)
forward_score = self._forward_alg(feats)
# 得到tag序列的路徑得分
gold_score = self._score_sentence(feats, tags)
return forward_score - gold_score
# 輸入語句序列得到最佳tag路徑及其得分
def forward(self, sentence): # dont confuse this with _forward_alg above.
# 從BiLSTM獲得發(fā)射分?jǐn)?shù)矩陣
lstm_feats = self._get_lstm_features(sentence)
# 使用維特比算法進(jìn)行解碼,計算最佳tag路徑及其得分
score, tag_seq = self._viterbi_decode(lstm_feats)
return score, tag_seq
START_TAG = “《START》”
STOP_TAG = “《STOP》”
# 詞嵌入維度
EMBEDDING_DIM = 5
# LSTM隱藏層維度
HIDDEN_DIM = 4
# 訓(xùn)練數(shù)據(jù)
training_data = [(
“the wall street journal reported today that apple corporation made money”.split(),
“B I I I O O O B I O O”.split()
), (
“georgia tech is a university in georgia”.split(),
“B I O O O O B”.split()
)]
word_to_ix = {}
# 構(gòu)建詞索引表,數(shù)字化以便計算機(jī)處理
for sentence, tags in training_data:
for word in sentence:
if word not in word_to_ix:
word_to_ix[word] = len(word_to_ix)
# 構(gòu)建標(biāo)簽索引表,數(shù)字化以便計算機(jī)處理
tag_to_ix = {“B”: 0, “I”: 1, “O”: 2, START_TAG: 3, STOP_TAG: 4}
# 初始化模型參數(shù)
model = BiLSTM_CRF(len(word_to_ix), tag_to_ix, EMBEDDING_DIM, HIDDEN_DIM)
# 使用隨機(jī)梯度下降法(SGD)進(jìn)行參數(shù)優(yōu)化
# model.parameters()為該實例中可優(yōu)化的參數(shù),
# lr:學(xué)習(xí)率,weight_decay:正則化系數(shù),防止模型過擬合
optimizer = optim.SGD(model.parameters(), lr=0.01, weight_decay=1e-4)
# 在no_grad模式下進(jìn)行前向推斷的檢測,函數(shù)作用是暫時不進(jìn)行導(dǎo)數(shù)的計算,目的在于減少計算量和內(nèi)存消耗
# 訓(xùn)練前檢查模型預(yù)測結(jié)果
with torch.no_grad():
# 取訓(xùn)練數(shù)據(jù)中第一條語句序列轉(zhuǎn)化為數(shù)字
precheck_sent = prepare_sequence(training_data[0][0], word_to_ix)
# 取訓(xùn)練數(shù)據(jù)中第一條語句序列對應(yīng)的標(biāo)簽序列進(jìn)行數(shù)字化
precheck_tags = torch.tensor([tag_to_ix[t] for t in training_data[0][1]], dtype=torch.long)
print(model(precheck_sent))
# 300輪迭代訓(xùn)練
for epoch in range(300):
for sentence, tags in training_data:
# Step 1. 每次開始前將上一輪的參數(shù)梯度清零,防止累加影響
model.zero_grad()
# Step 2. seq、tags分別數(shù)字化為sentence_in、targets
sentence_in = prepare_sequence(sentence, word_to_ix)
targets = torch.tensor([tag_to_ix[t] for t in tags], dtype=torch.long)
# Step 3. 損失函數(shù)loss
loss = model.neg_log_likelihood(sentence_in, targets)
# Step 4. 通過調(diào)用optimizer.step()計算損失、梯度、更新參數(shù)
loss.backward()
optimizer.step()
# torch.no_grad() 是一個上下文管理器,被該語句 wrap 起來的部分將不會track 梯度
# 訓(xùn)練結(jié)束查看模型預(yù)測結(jié)果,對比觀察模型是否學(xué)到
with torch.no_grad():
precheck_sent = prepare_sequence(training_data[0][0], word_to_ix)
print(model(precheck_sent))
# We got it!
歡迎交流指正
參考資料:
[1]torch.max()使用講解
https://www.jianshu.com/p/3ed11362b54f
[2]torch.manual_seed()用法
https://www.cnblogs.com/dychen/p/13920000.html
[3]BiLSTM-CRF原理介紹+Pytorch_Tutorial代碼解析
https://blog.csdn.net/misite_j/article/details/109036725
[4]關(guān)于nn.embedding函數(shù)的理解
https://blog.csdn.net/a845717607/article/details/104752736
[5]torch.nn.LSTM()詳解
https://blog.csdn.net/m0_45478865/article/details/104455978
[6]pytorch函數(shù)之nn.Linear
https://www.cnblogs.com/Archer-Fang/p/10645473.html
[7]pytorch之torch.randn()
https://blog.csdn.net/zouxiaolv/article/details/99568414
[8]torch.full()
https://blog.csdn.net/Fluid_ray/article/details/109855155
[9]PyTorch中view的用法
https://blog.csdn.net/york1996/article/details/81949843
https://blog.csdn.net/zkq_1986/article/details/100319146
[10]torch.cat()函數(shù)
https://blog.csdn.net/xinjieyuan/article/details/105208352
[11]ADVANCED: MAKING DYNAMIC DECISIONS AND THE BI-LSTM CRF
https://pytorch.org/tutorials/beginner/nlp/advanced_tutorial.html
[12]條件隨機(jī)場理論理解
https://blog.csdn.net/qq_27009517/article/details/107154441
[13]PyTorch tutorial - BiLSTM CRF 代碼解析
https://blog.csdn.net/ono_online/article/details/105089750
編輯:lyn
-
函數(shù)
+關(guān)注
關(guān)注
3文章
4259瀏覽量
62227 -
代碼
+關(guān)注
關(guān)注
30文章
4695瀏覽量
68081 -
LSTM
+關(guān)注
關(guān)注
0文章
43瀏覽量
3730
原文標(biāo)題:【NER】命名實體識別:詳解BiLSTM_CRF_Pytorch_Tutorial代碼
文章出處:【微信號:zenRRan,微信公眾號:深度學(xué)習(xí)自然語言處理】歡迎添加關(guān)注!文章轉(zhuǎn)載請注明出處。
發(fā)布評論請先 登錄
相關(guān)推薦
評論