導(dǎo)讀：本文主要解析Pytorch Tutorial中BiLSTM_CRF代碼，幾乎注釋了每行代碼，希望本文能夠幫助大家理解這個(gè)tutorial，除此之外借助代碼和圖解也對(duì)理解條件隨機(jī)場(chǎng)（CRF）會(huì)有一定幫助，因?yàn)檫@個(gè)tutorial代碼主要還是在實(shí)現(xiàn)CRF部分。

1 知識(shí)準(zhǔn)備

在閱讀tutorial前，需具備一些理論或知識(shí)基礎(chǔ)，包括LSTM單元、BiLSTM-CRF模型、CRF原理以及一些代碼中的函數(shù)使用，參考資料中涵蓋了主要的涉及知識(shí)，可配合tutorial一同學(xué)習(xí)。

2 理解CRF中歸一化因子Z（x）的計(jì)算

條件隨機(jī)場(chǎng)中的Z（x）表示歸一化因子，它是一個(gè)句子所有可能標(biāo)記tag序列路徑的得分總和。一般的，我們會(huì)有一個(gè)直接的想法，就是列舉出所有可能的路徑，計(jì)算出每條路徑的得分之后再加和。如上圖中的例子所示，有5個(gè)字符和5個(gè)tag，如果按照上述的暴力窮舉法進(jìn)行計(jì)算，就有種路徑組合，而在我們的實(shí)際工作中，可能會(huì)有更長(zhǎng)的序列和更多的tag標(biāo)簽，此時(shí)暴力窮舉法未免顯得有些效率低下。于是我們考慮采用分?jǐn)?shù)累積的方式進(jìn)行所有路徑得分總和的計(jì)算，即先計(jì)算出到達(dá)的所有路徑的總得分，然后計(jì)算-》的所有路徑的得分，然后依次計(jì)算-》。..-》間的所有路徑的得分，最后便得到了我們的得分總和，這個(gè)思路源于如下等價(jià)等式：

上式相等表明，直接計(jì)算整個(gè)句子序列的全局分?jǐn)?shù)與計(jì)算每一步的路徑得分再加和等價(jià)，計(jì)算每一步的路徑得分再加和這種方式可以大大減少計(jì)算的時(shí)間，故Pytorch Tutorial中的_forward_alg（）函數(shù)據(jù)此實(shí)現(xiàn)。這種計(jì)算每一步的路徑得分再加和的方法還可以以下圖方式進(jìn)行計(jì)算。

3 理解CRF中序列解碼過程，即viterbi算法

如上圖所示，在每個(gè)時(shí)間步上，比如’word==去‘這一列，每一個(gè)tag處（0~6豎框是tag的id），關(guān)注兩個(gè)值：前一個(gè)時(shí)間步上所有tag到當(dāng)前tag中總得分最大值以及該最大值對(duì)應(yīng)的前一個(gè)時(shí)間步上tag的id。這樣一來每個(gè)tag都記錄了它前一個(gè)時(shí)間步上到自己的最優(yōu)路徑，最后通過tag的id進(jìn)行回溯，這樣就可以得到最終的最優(yōu)tag標(biāo)記序列。此部分對(duì)應(yīng)Pytorch Tutorial中的_viterbi_decode（）函數(shù)實(shí)現(xiàn)。

4 理解log_sum_exp（）函數(shù)

Pytorch Tutorial中的log_sum_exp（）函數(shù)最后返回的計(jì)算方式數(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é)果，這樣就會(huì)有個(gè)缺點(diǎn)

# 指數(shù)和積累到一定程度之后，會(huì)超過計(jì)算機(jī)浮點(diǎn)值的最大值

# 變成inf，這樣取log后也是inf

# 為了避免這種情況，用一個(gè)合適的值clip=max去提指數(shù)和的公因子

# 這樣不會(huì)使某項(xiàng)變得過大而無法計(jì)算

def log_sum_exp（vec）：# vec：形似［［tag個(gè)元素］］

# 取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ì)算機(jī)上溢

# 等同于torch.log（torch.sum（torch.exp（vec））），防止e的指數(shù)導(dǎo)致計(jì)算機(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個(gè)數(shù)

self.tagset_size = len（tag_to_ix）

# num_embeddings （int）：vocab_size 詞典的大小

# embedding_dim （int）：embedding_dim 嵌入向量的維度，即用多少維來表示一個(gè)符號(hào)

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é)果的一個(gè)最大的整數(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 每個(gè)輸入樣本的大小

# out_features:tagset_size 每個(gè)輸出樣本的大小

self.hidden2tag = nn.Linear（hidden_dim， self.tagset_size）

# 轉(zhuǎn)移矩陣的參數(shù)初始化，transition［i，j］代表的是從第j個(gè)tag轉(zhuǎn)移到第i個(gè)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

# 否則模型自動(dòng)初始化為零值，維度為［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））

# 計(jì)算歸一化因子Z（x）

def _forward_alg（self， feats）：

‘’‘

輸入：發(fā)射矩陣（emission score），實(shí)際上就是LSTM的輸出

sentence的每個(gè)word經(jīng)BiLSTM后對(duì)應(yīng)于每個(gè)label的得分

輸出：所有可能路徑得分之和/歸一化因子/配分函數(shù)/Z（x）

’‘’

# 通過前向算法遞推計(jì)算

# 初始化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.

# 包裝到一個(gè)變量里面以便自動(dòng)反向傳播

forward_var = init_alphas

# 迭代整個(gè)句子

# feats：形似［［。..。］， 每個(gè)字映射到tag的發(fā)射概率，

# ［。..。］，

# ［。..。］］

for feat in feats：

# 存儲(chǔ)當(dāng)前時(shí)間步下各tag得分

alphas_t = ［］

for next_tag in range（self.tagset_size）：

# 取出當(dāng)前tag的發(fā)射分?jǐn)?shù)（與之前時(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ù)：之前時(shí)間步分?jǐn)?shù)+轉(zhuǎn)移分?jǐn)?shù)+發(fā)射分?jǐn)?shù)

next_tag_var = forward_var + trans_score + emit_score

# 對(duì)當(dāng)前分?jǐn)?shù)取log-sum-exp

alphas_t.append（log_sum_exp（next_tag_var）.view（1））

# 更新forward_var 遞推計(jì)算下一個(gè)時(shí)間步

# 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］］

# 對(duì)當(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)造的詞嵌入為語(yǔ)句中每個(gè)詞（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

# 計(jì)算得到輸出結(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］）表

# 可以看作為每個(gè)詞被標(biāo)注為對(duì)應(yīng)標(biāo)簽的得分情況，即維特比算法中的發(fā)射矩陣

lstm_feats = self.hidden2tag（lstm_out）

return lstm_feats

# 計(jì)算一個(gè)tag序列路徑的得分

def _score_sentence（self， feats， tags）：

# feats發(fā)射分?jǐn)?shù)矩陣

# 計(jì)算給定tag序列的分?jǐn)?shù)，即一條路徑的分?jǐn)?shù)

score = torch.zeros（1）

# tags前面補(bǔ)上一個(gè)句首標(biāo)簽便于計(jì)算轉(zhuǎn)移得分

tags = torch.cat（［torch.tensor（［self.tag_to_ix［START_TAG］］， dtype=torch.long）， tags］）

# 循環(huán)用于計(jì)算給定tag序列的分?jǐn)?shù)

for i， feat in enumerate（feats）：

# 遞推計(jì)算路徑分?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表示每個(gè)標(biāo)簽的前向狀態(tài)得分，即上一個(gè)詞被打作每個(gè)標(biāo)簽的對(duì)應(yīng)得分值

forward_var = init_vvars

# 遍歷每個(gè)時(shí)間步時(shí)的發(fā)射分?jǐn)?shù)

for feat in feats：

# 記錄當(dāng)前詞對(duì)應(yīng)每個(gè)標(biāo)簽的最優(yōu)轉(zhuǎn)移結(jié)點(diǎn)

# 保存當(dāng)前時(shí)間步的回溯指針

bptrs_t = ［］

# 與bptrs_t對(duì)應(yīng)，記錄對(duì)應(yīng)的最優(yōu)值

# 保存當(dāng)前時(shí)間步的viterbi變量

viterbivars_t = ［］

# 遍歷每個(gè)標(biāo)簽，求得當(dāng)前詞被打作每個(gè)標(biāo)簽的得分

# 并將其與當(dāng)前詞的發(fā)射矩陣feat相加，得到當(dāng)前狀態(tài)，即下一個(gè)詞的前向狀態(tài)

for next_tag in range（self.tagset_size）：

# transitions［next_tag］表示每個(gè)標(biāo)簽轉(zhuǎn)移到next_tag的轉(zhuǎn)移得分

# forward_var表示每個(gè)標(biāo)簽的前向狀態(tài)得分，即上一個(gè)詞被打作每個(gè)標(biāo)簽的對(duì)應(yīng)得分值

# 二者相加即得到當(dāng)前詞被打作next_tag的所有可能得分

# 維特比算法記錄最優(yōu)路徑時(shí)只考慮上一步的分?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］

# 得到上一個(gè)可能的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位置索引對(duì)應(yīng)的值

viterbivars_t.append（next_tag_var［0］［best_tag_id］.view（1））

# 更新forward_var = 當(dāng)前詞的發(fā)射分?jǐn)?shù)feat + 前一個(gè)最優(yōu)tag當(dāng)前tag的狀態(tài)下的得分

forward_var = （torch.cat（viterbivars_t） + feat）.view（1， -1）

# 回溯指針記錄當(dāng)前時(shí)間步各個(gè)tag來源前一步的最優(yōu)tag

backpointers.append（bptrs_t）

# forward_var表示每個(gè)標(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前一個(gè)時(shí)間步的最優(yōu)tag位置索引

best_tag_id = argmax（terminal_var）

# 得到標(biāo)簽STOP_TAG當(dāng)前最優(yōu)tag對(duì)應(yīng)的分?jǐn)?shù)值

path_score = terminal_var［0］［best_tag_id］

# 根據(jù)過程中存儲(chǔ)的轉(zhuǎn)移路徑結(jié)點(diǎn)，反推最優(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é)點(diǎn)一定是人為構(gòu)建的START_TAG，刪除，并根據(jù)這一點(diǎn)確認(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）：

# 得到句子對(duì)應(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

# 輸入語(yǔ)句序列得到最佳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)行解碼，計(jì)算最佳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ì)算機(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ì)算機(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（）為該實(shí)例中可優(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)行前向推斷的檢測(cè)，函數(shù)作用是暫時(shí)不進(jìn)行導(dǎo)數(shù)的計(jì)算，目的在于減少計(jì)算量和內(nèi)存消耗

# 訓(xùn)練前檢查模型預(yù)測(cè)結(jié)果

with torch.no_grad（）：

# 取訓(xùn)練數(shù)據(jù)中第一條語(yǔ)句序列轉(zhuǎn)化為數(shù)字

precheck_sent = prepare_sequence（training_data［0］［0］， word_to_ix）

# 取訓(xùn)練數(shù)據(jù)中第一條語(yǔ)句序列對(duì)應(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（）計(jì)算損失、梯度、更新參數(shù)

loss.backward（）

optimizer.step（）

# torch.no_grad（） 是一個(gè)上下文管理器，被該語(yǔ)句 wrap 起來的部分將不會(huì)track 梯度

# 訓(xùn)練結(jié)束查看模型預(yù)測(cè)結(jié)果，對(duì)比觀察模型是否學(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ī)場(chǎng)理論理解

https://blog.csdn.net/qq_27009517/article/details/107154441

［13］PyTorch tutorial - BiLSTM CRF 代碼解析

https://blog.csdn.net/ono_online/article/details/105089750
編輯：lyn