摘要：比如一個(gè)機(jī)器翻譯模型，輸入是模型輸出湯姆，追逐，杰瑞。模型目前在機(jī)器翻譯，圖片描述任務(wù)，語(yǔ)音識(shí)別都有大量應(yīng)用，熟練使用對(duì)于解決實(shí)際問(wèn)題會(huì)有很大的幫助。

Attention模型形象的比喻就是“圖像對(duì)焦”。

上圖是Encoder-Decoder模型，Decoder中每個(gè)單詞生成過(guò)程如下：

其中C是“語(yǔ)義編碼C”，f是Decoder的非線性變換函數(shù)。由此，我們可以看出生成目標(biāo)句子的每個(gè)單詞都使用同一個(gè)語(yǔ)義編碼C，即：源句子中的每個(gè)單詞的影響力都是一樣的，這如同圖像沒(méi)有對(duì)焦的情況，現(xiàn)實(shí)項(xiàng)目中也存在明顯的不合理。比如一個(gè)機(jī)器翻譯模型，輸入是“Tom chase Jerry”,模型輸出：“湯姆”，“追逐”，“杰瑞”。在翻譯“杰瑞”的時(shí)候顯然“Jerry”的貢獻(xiàn)值最大，如果每個(gè)單詞的貢獻(xiàn)值相同明顯不合理。這個(gè)問(wèn)題在輸入句子長(zhǎng)度較短時(shí)問(wèn)題不大，但是當(dāng)輸入句子較長(zhǎng)時(shí)會(huì)丟失很多細(xì)節(jié)信息（個(gè)人覺(jué)得此處類似平均池化和最大值池化）。正因?yàn)槿绱?，我們引入?strong>Attention思想。

使用Attention模型翻譯“杰瑞”的時(shí)候，我們可以得到輸入句子中的每個(gè)單詞對(duì)輸出當(dāng)前單詞的貢獻(xiàn)值大小如：（Tom,0.3）(Chase,0.2) (Jerry,0.5)。這意味著生成每個(gè)單詞yi時(shí)不再使用同一個(gè)語(yǔ)義編碼C，而是根據(jù)yi使用不同的Ci。在引入Attention模型后yi的計(jì)算過(guò)程改變?nèi)缦滤荆?/p>

每個(gè)Ci對(duì)應(yīng)源句子中每個(gè)單詞的注意力分配概率，示例如下：

f2是Encoder對(duì)每個(gè)單詞的變換函數(shù)，g函數(shù)代表整個(gè)源句子的中間語(yǔ)義表示的變換函數(shù)，一般形式是加權(quán)求和：

aji代表注意力分配系數(shù)，hj代表源句子中某個(gè)單詞的語(yǔ)義編碼，Lx代表源句子中單詞數(shù)量。g函數(shù)的計(jì)算過(guò)程如下圖所示：

如果所示，當(dāng)我們要生成yi單詞，此時(shí)我們用i-1時(shí)刻的隱藏節(jié)點(diǎn)輸出值Hi-1去和源句子中的每個(gè)單詞對(duì)應(yīng)RNN隱藏節(jié)點(diǎn)狀態(tài)hj依次進(jìn)行對(duì)比，即：通過(guò)函數(shù)F（hj，Hi-1）來(lái)獲得yi對(duì)源句子中每個(gè)單詞對(duì)應(yīng)的對(duì)齊可能性，函數(shù)F常見(jiàn)方法如下圖所示：

然后使用Softmax函數(shù)進(jìn)行數(shù)值歸一化處理。如對(duì)“對(duì)齊概率”不理解的朋友，可以查看下圖英語(yǔ)-德語(yǔ)翻譯系統(tǒng)中加入Attention機(jī)制后，Encoder和Decoder兩個(gè)句子中每個(gè)單詞對(duì)應(yīng)注意力分配概率分布。

在Soft Attention模型中，Attention機(jī)制發(fā)生在Decoder中Yi和Encoder中的所有元素之間。Self Attention模型不是在兩者之間，而是Decoder內(nèi)部元素之間或者Encoder內(nèi)部元素之間發(fā)生的Attention機(jī)制，計(jì)算方法和Soft Attention模型一致。那么Self Attention模型有什么好處？我們依然以機(jī)器翻譯為例：

如圖所示，Self Attention模型在內(nèi)部可以捕獲一些句法特征或語(yǔ)義特征。Self Attention模型相比傳統(tǒng)RNN模型需要依次序序列計(jì)算，它的感受野更大，可以直接將句子中的任意兩個(gè)單詞的聯(lián)系通過(guò)一個(gè)計(jì)算步驟聯(lián)系起來(lái)，可以捕獲遠(yuǎn)距離的相互依賴特征（就像列表和數(shù)組的區(qū)別）。此外，Self Attention模型對(duì)于增加計(jì)算的并行性也有幫助。

我們使用的語(yǔ)言數(shù)據(jù)集是“英語(yǔ)-西班牙語(yǔ)”，數(shù)據(jù)集樣本如下圖所示：

# 數(shù)據(jù)下載
path_to_zip=tf.keras.utils.get_file(
    fname="spa-eng.zip",
    origin="http://download.tensorflow.org/data/spa-eng.zip",
    # 解壓tar zip文件
    extract=True
)
path_to_file=os.path.dirname(path_to_zip)+"/spa-eng/spa.txt"

轉(zhuǎn)碼：

def unicode_to_ascii(sen):
    return "".join(
       char for char in unicodedata.normalize("NFD",sen)
       if unicodedata.category(char) != "Mn"
    )

每條訓(xùn)練語(yǔ)句添加開始和結(jié)束標(biāo)記

移除句子中的特殊字符

字符轉(zhuǎn)ID，ID轉(zhuǎn)字符并排序

將句子補(bǔ)長(zhǎng)到預(yù)設(shè)的最大長(zhǎng)度

def preprocess_sentence(w):
    w = unicode_to_ascii(w.lower().strip())
    
    # 在單詞和標(biāo)點(diǎn)之間創(chuàng)建空格
    # 如： "he is a boy." => "he is a boy ." 
    w = re.sub(r"([?.!,?])", r" 1 ", w)
    w = re.sub(r"[" "]+", " ", w)
    
    # 特殊字符以空格代替
    w = re.sub(r"[^a-zA-Z?.!,?]+", " ", w)
    
    w = w.rstrip().strip()
    
    # 添加開始和結(jié)束標(biāo)記
    w = " " + w + " "
    return w

創(chuàng)建數(shù)據(jù)集：

def create_dataset(path, num_examples):
    lines = open(path, encoding="UTF-8").read().strip().split("
")
    
    word_pairs = [[preprocess_sentence(w) for w in l.split("	")]  for l in lines[:num_examples]]
    
    # 返回格式：[ENGLISH, SPANISH]
    return word_pairs

字符轉(zhuǎn)ID，ID轉(zhuǎn)字符，并排序：

class LanguageIndex():
    def __init__(self,lang):
        self.lang=lang
        self.wrod2idx={}
        self.id2word={}
        self.vacab=set()
        self.create_index()

    def create_index(self):
        for phrase in self.lang:
            # 添加到集合中，重復(fù)內(nèi)容不添加
            self.vacab.update(phrase.split(" "))
        self.vacab=sorted(self.vacab)
        self.wrod2idx[""]=0
        
        #字符-ID轉(zhuǎn)換
        for index,word in enumerate(self.vacab):
            self.wrod2idx[word]=index+1
        for word,index in self.wrod2idx.items():
            self.id2word[index]=word

加載數(shù)據(jù)集：

# 計(jì)算最大長(zhǎng)度
def max_length(tensor):
    return max(len(t) for t in tensor)

def load_dataset(path,num_example):
    #get inputs outputs
    pairs=create_dataset(path,num_example)

    # 獲取ID表示
    inp_lang=LanguageIndex(sp for en,sp in pairs)
    targ_lang=LanguageIndex(en for en,sp in pairs)

    # LanguageIndex 不包含重復(fù)值，以下包含重復(fù)值
    input_tensor=[[inp_lang.wrod2idx[s]for s in sp.split(" ")]for en,sp in pairs]
    target_tensor=[[targ_lang.wrod2idx[s]for s in en.split(" ")]for en,sp in pairs]

    max_length_inp,max_length_tar=max_length(input_tensor),max_length(target_tensor)

    # 將句子補(bǔ)長(zhǎng)到預(yù)設(shè)的最大長(zhǎng)度
    # padding: post:后補(bǔ)長(zhǎng)，pre:前補(bǔ)長(zhǎng)
    input_tensor=tf.keras.preprocessing.sequence.pad_sequences(
        sequences=input_tensor,
        maxlen=max_length_inp,
        padding="post"
    )
    target_tensor=tf.keras.preprocessing.sequence.pad_sequences(
        sequences=target_tensor,
        maxlen=max_length_tar,
        padding="post"
    )

    return input_tensor,target_tensor,inp_lang,targ_lang,max_length_inp,max_length_tar

創(chuàng)建訓(xùn)練集驗(yàn)證集：

# 本次項(xiàng)目只使用前30000條數(shù)據(jù)
num_examples = 30000
input_tensor, target_tensor, inp_lang, targ_lang, max_length_inp, max_length_targ = load_dataset(path_to_file, num_examples)

# 訓(xùn)練集80%，驗(yàn)證集20%
input_tensor_train, input_tensor_val, target_tensor_train, target_tensor_val = train_test_split(input_tensor, target_tensor, test_size=0.2)

# 打亂數(shù)據(jù)集
BUFFER_SIZE=len(input_tensor_train)
BATCH_SIZE=64

# 每個(gè)epoch迭代次數(shù)
N_BATCH=BUFFER_SIZE // BATCH_SIZE

# 詞嵌入維度
embedding_dim=256

# 隱藏神經(jīng)元數(shù)量
units=1024
vocab_inp_size=len(inp_lang.wrod2idx)
vocab_tar_size=len(targ_lang.wrod2idx)

dataset=tf.data.Dataset.from_tensor_slices((input_tensor_train,target_tensor_train)).shuffle(BUFFER_SIZE)
# drop_remainder 當(dāng)剩余數(shù)據(jù)量小于batch_size時(shí)候，是否丟棄
dataset=dataset.batch(BATCH_SIZE,drop_remainder="True")

文章開始我們介紹了Attention模型的計(jì)算過(guò)程，相信你會(huì)很容易理解上圖的內(nèi)容。對(duì)每個(gè)節(jié)點(diǎn)具體方程實(shí)現(xiàn)如下：

FC=全連接層，EO=編碼器輸出，H=隱藏層狀態(tài)，X=解碼器輸入，模型計(jì)算過(guò)程如下表示：

score = FC(tanh(FC(EO) + FC(H)))

attention weights = softmax(score, axis = 1)

context vector = sum(attention weights * EO, axis = 1)

embedding output=解碼器輸入X，輸入詞嵌入層

merged vector=concat(embedding output, context vector)

將merged vector輸入到GRU

GRU配置：

def gru(units):
  # 使用GPU加速運(yùn)算
  if tf.test.is_gpu_available():
    return tf.keras.layers.CuDNNGRU(units, 
                                    return_sequences=True, 
                                    return_state=True, 
                                    
                                    # 循環(huán)核的初始化方法
                                    # glorot_uniform是sqrt(2 / (fan_in + fan_out))的正態(tài)分布產(chǎn)生
                                    # 其中fan_in和fan_out是權(quán)重張量的扇入扇出（即輸入和輸出單元數(shù)目）
                                    recurrent_initializer="glorot_uniform")
  else:
    return tf.keras.layers.GRU(units, 
                               return_sequences=True, 
                               return_state=True, 
                               
                               # hard_sigmoid <= -1 輸出0，>=1 輸出1 ，中間為線性
                               recurrent_activation="sigmoid", 
                               recurrent_initializer="glorot_uniform")

編碼器：

class Encoder(tf.keras.Model):
    def __init__(self, vocab_size, embedding_dim, enc_units, batch_sz):
        super(Encoder, self).__init__()
        self.batch_sz = batch_sz
        self.enc_units = enc_units
        self.embedding = tf.keras.layers.Embedding(vocab_size, embedding_dim)
        self.gru = gru(self.enc_units)
        
    def call(self, x, hidden):
        x = self.embedding(x)
        output, state = self.gru(x, initial_state = hidden)        
        return output, state
      
    def initialize_hidden_state(self):
        return tf.zeros((self.batch_sz, self.enc_units))

解碼器：

class Decoder(tf.keras.Model):
    def __init__(self,vocab_size,embedding_dim,dec_units,batch_sz):
        super(Decoder, self).__init__()
        self.batch_sz=batch_sz
        self.dec_units=dec_units
        self.embedding=tf.keras.layers.Embedding(
            input_shape=vocab_size,
            output_dim=embedding_dim
        )
        self.gru=gru(self.dec_units)
        self.fc=tf.keras.layers.Dense(units=vocab_size)
        
        # 用于計(jì)算score，即：注意力權(quán)重系數(shù)
        self.W1=tf.keras.layers.Dense(self.dec_units)
        self.W2=tf.keras.layers.Dense(self.dec_units)
        self.V=tf.keras.layers.Dense(units=1)
    def __call__(self,x,hidden,ec_output):
        # tf.expand_dims:在指定索引出增加一維度，值為1，從索引0開始
        # axis: 取值范圍是[-階數(shù)，階數(shù)]，二維的時(shí)候0指的是列，1指的是行，
        # 更高維度的時(shí)候，數(shù)值是由外向里增加，如：3維向量，外向內(nèi)依次是：0，1，2
        
        # 通過(guò)計(jì)算score公式可得，需要將hidden維度擴(kuò)展至:[batch_size,1,hidden_size]
        hidden_with_time_axis=tf.expand_dims(hidden,axis=1)
        
        # score=[batch_size, max_length, 1]
        score=self.V(tf.nn.tanh(self.W1(ec_output)+self.W2(hidden_with_time_axis)))
        
        # 數(shù)值歸一化和為1的概率分布值
        attention_weight=tf.nn.softmax(score,axis=1)
        context_vetor=attention_weight*ec_output
        
        # 求和平均
        context_vetor=tf.reduce_sum(context_vetor,axis=1)
        X=self.embedding(x)
        
        # 合并解碼器embedding輸出和context vector
        x = tf.concat([tf.expand_dims(context_vector, 1), x], axis=-1)

        # output shape=(batch_size,time_step,hidden_size)
        # state shape=(batch_size,hidden_size)
        output,state=self.gru(x)
        
        # output[batch_size*1,hidden_size]
        output=tf.reshape(output,shape=(-1,output.shape[2]))
        x-self.fc(output)

        return x,state,attention_weight
    def initilize_hidden_size(self):
        return tf.zeros((self.batch_sz,self.dec_units))

實(shí)例化模型：

encoder = Encoder(vocab_inp_size, embedding_dim, units, BATCH_SIZE)
decoder = Decoder(vocab_tar_size, embedding_dim, units, BATCH_SIZE)

損失函數(shù)，優(yōu)化器：

optimizer = tf.train.AdamOptimizer()


def loss_function(real, pred):
  mask = 1 - np.equal(real, 0)
  loss_ = tf.nn.sparse_softmax_cross_entropy_with_logits(labels=real, logits=pred) * mask
  return tf.reduce_mean(loss_)

模型保存：

checkpoint_dir = "./training_checkpoints"
checkpoint_prefix = os.path.join(checkpoint_dir, "ckpt")
checkpoint = tf.train.Checkpoint(optimizer=optimizer,
                                 encoder=encoder,
                                 decoder=decoder)

由于我們使用Teacher Forcing進(jìn)行訓(xùn)練，所以我們簡(jiǎn)單介紹下。

如圖所示Teacher Forcing與Free-running不同，在訓(xùn)練過(guò)程中不再是前一時(shí)刻的hidden-output作為當(dāng)前輸入，
而是在Ground Truth中找到對(duì)應(yīng)的上一項(xiàng)作為當(dāng)前輸入。早期的RNN很弱，如果生成了非常差的結(jié)果Free-running的運(yùn)行方式會(huì)導(dǎo)致后面的hidden-output都受到影響。Teacher Forcing運(yùn)行方式就可以避免這種問(wèn)題，缺點(diǎn)也很明顯它嚴(yán)重依賴標(biāo)簽數(shù)據(jù)。

# 迭代10次訓(xùn)練集
EPOCHS = 10

for epoch in range(EPOCHS):
    start = time.time()
    
    hidden = encoder.initialize_hidden_state()
    total_loss = 0
    
    for (batch, (inp, targ)) in enumerate(dataset):
        loss = 0
        
        # 先記錄梯度
        with tf.GradientTape() as tape:
            # 編碼器輸出
            enc_output, enc_hidden = encoder(inp, hidden)
            
            dec_hidden = enc_hidden
            
            dec_input = tf.expand_dims([targ_lang.word2idx[""]] * BATCH_SIZE, 1)       
            
            # 使用Teacher forcing運(yùn)行方式
            for t in range(1, targ.shape[1]):
                # 解碼器輸出
                predictions, dec_hidden, _ = decoder(dec_input, dec_hidden, enc_output)
                
                loss += loss_function(targ[:, t], predictions)
                
                # 樣本標(biāo)簽作為輸入
                dec_input = tf.expand_dims(targ[:, t], 1)
        
        batch_loss = (loss / int(targ.shape[1]))
        
        # one_loss++;batch_loss++
        total_loss += batch_loss
        
        variables = encoder.variables + decoder.variables
        
        gradients = tape.gradient(loss, variables)
        
        optimizer.apply_gradients(zip(gradients, variables))
        
        if batch % 100 == 0:
            print("Epoch {} Batch {} Loss {:.4f}".format(epoch + 1,
                                                         batch,
                                                         batch_loss.numpy()))
    # 每迭代2次訓(xùn)練集保存一次模型
    if (epoch + 1) % 2 == 0:
      checkpoint.save(file_prefix = checkpoint_prefix)

評(píng)估函數(shù)我們不使用teacher-forcing模式，解碼器的每步輸入是它前一時(shí)刻的hidden-state和編碼器輸出，當(dāng)模型遇到 標(biāo)記停止運(yùn)行。

# 和訓(xùn)練模型函數(shù)代碼基本一致
def evaluate(sentence, encoder, decoder, inp_lang, targ_lang, max_length_inp, max_length_targ):
    attention_plot = np.zeros((max_length_targ, max_length_inp))
    
    # 數(shù)據(jù)預(yù)處理
    sentence = preprocess_sentence(sentence)
    
    # 向量化表示輸入數(shù)據(jù)
    inputs = [inp_lang.word2idx[i] for i in sentence.split(" ")]
    
    # 后置補(bǔ)長(zhǎng)
    inputs = tf.keras.preprocessing.sequence.pad_sequences([inputs], maxlen=max_length_inp, padding="post")
    inputs = tf.convert_to_tensor(inputs)
    
    result = ""

    hidden = [tf.zeros((1, units))]
    enc_out, enc_hidden = encoder(inputs, hidden)

    dec_hidden = enc_hidden
    # 維度擴(kuò)展batch_size
    dec_input = tf.expand_dims([targ_lang.word2idx[""]], 0)

    for t in range(max_length_targ):
        predictions, dec_hidden, attention_weights = decoder(dec_input, dec_hidden, enc_out)
        
        # 保存權(quán)重用于稍后可視化展示
        attention_weights = tf.reshape(attention_weights, (-1, ))
        attention_plot[t] = attention_weights.numpy()
        
        
        predicted_id = tf.argmax(predictions[0]).numpy()
        
        # 獲取文本翻譯結(jié)果
        result += targ_lang.idx2word[predicted_id] + " "
        
        # 預(yù)設(shè)的結(jié)束標(biāo)記
        if targ_lang.idx2word[predicted_id] == "":
            return result, sentence, attention_plot
        
        # 預(yù)測(cè)值作為輸入，以此輸出下一時(shí)刻單詞
        dec_input = tf.expand_dims([predicted_id], 0)

    return result, sentence, attention_plot

可視化權(quán)重值：

    fig = plt.figure(figsize=(10,10))
    ax = fig.add_subplot(1, 1, 1)
    ax.matshow(attention, cmap="viridis")
    
    fontdict = {"fontsize": 14}
    
    ax.set_xticklabels([""] + sentence, fontdict=fontdict, rotation=90)
    ax.set_yticklabels([""] + predicted_sentence, fontdict=fontdict)

    plt.show()

本篇文章篇幅較多，不過(guò)項(xiàng)目的重點(diǎn)是Attention思想的理解，Self Attention模型具有更長(zhǎng)的感受野，更容易捕獲長(zhǎng)距離的相互依賴特征，目前Google機(jī)器翻譯模型就大量使用到Self Attention。Attention模型目前在機(jī)器翻譯，圖片描述任務(wù)，語(yǔ)音識(shí)別都有大量應(yīng)用，熟練使用Attention對(duì)于解決實(shí)際問(wèn)題會(huì)有很大的幫助。

文章部分內(nèi)容參考 Yash Katariya 和 張俊林，在此表示感謝。