RNN聊天機器人與Beam Search [Tensorflow Seq2Seq]
本部落格分析了一個Tensorflow實現的開源聊天機器人專案deepQA，首先從資料集上和一些重要程式碼上進行了說明和闡述，最後針對於測試的情況，在deepQA專案上實現了Beam Search的方法，讓模型輸出的句子更加準確，修改後的原始碼在這裡。
DeepQA
DeepQA是一個Tensorflow實現的開源的seq2seq模型的聊天機器人( 傳送們),出自谷歌的一篇關於對話模型的論文A Neural Conversational Model，訓練的語料庫包含電影臺詞的對話（Cornell和擴充套件版本的Cornell），Scotus對話庫，以及Ubantu的對話。這些資料都能在專案的data裡找到，但是目前好像只能針對某一個對話資料庫進行訓練，還沒有支援混合對話的訓練。目前實現的模型使用的是基礎RNN中的seq2seq模型，主要針對的是比較短的對話。
1. Cornell資料集
DeepQA預設的是Cornell對話資料，一共兩個檔案：人物對話資訊movie_conversations.txt和具體對話內容movie_lines.txt，+++$+++為分隔符，movie_conversations.txt裡每一行的第一個資料代表對話人物1的ID，第二個資料代表對話人物2的ID，第三個資料代表電影ID，後面是對話的ID，而movie_lines.txt裡每一行的第一個資料代表對話ID，第二個資料表示說話的人物ID，第三個資料電影ID，第四個是此人物的名字，最後是這句話的具體內容。
＃ movie_conversations. 通過解析這個兩個檔案，可以在data/samples路徑生成dataset-cornell.pkl資料集以及dataset-cornel-length10-filter1-vocabSize40000.pkl詞彙表，預設對話的長度一般最大為10，過濾詞頻小於1的詞並用unk代替，詞彙表最大40000；這些引數可以自行修改，生成過程具體由textdata.py實現
2. 重要程式碼分析
2.1 構建基礎RNN網路結構
直接看一下構建模型的程式碼，首先定義單個LSTM cell，然後用Dropout包裹，最後用引數numLayers決定多少層Stack結構的RNN：
def create_rnn_cell():
encoDecoCell = tf.contrib.rnn.BasicLSTMCell( # Or GRUCell, LSTMCell(args.hiddenSize)
self.args.hiddenSize,
)
if not self.args.test: # TODO: Should use a placeholder instead
encoDecoCell = tf.contrib.rnn.DropoutWrapper(
encoDecoCell,
input_keep_prob=1.0,
output_keep_prob=self.args.dropout
)
return encoDecoCell
encoDecoCell = tf.contrib.rnn.MultiRNNCell(
[create_rnn_cell() for _ in range(self.args.numLayers)],
)2.2 定義輸入值
接著定義網路的輸入值，根據標準的seq2seq模型，一共四個：
1. encorder的輸入：人物1說的一句話A，最大長度10
2. decoder的輸入：人物2回覆的對話B，因為前後分別加上了go開始符和end結束符，最大長度為12
3. decoder的target輸入：decoder輸入的目標輸出，與decoder的輸入一樣但只有end標示符號，可以理解為decoder的輸入在時序上的結果，比如說完這個詞後的下個詞的結果。
4. decoder的weight輸入：用來標記target中的非padding的位置，即實際句子的長度，因為不是所有的句子的長度都一樣，在實際輸入的過程中，各個句子的長度都會被用統一的標示符來填充（padding）至最大長度，weight用來標記實際詞彙的位置，代表這個位置將會有梯度值回傳。
with tf.name_scope('placeholder_encoder'):
self.encoderInputs = [tf.placeholder(tf.int32, [None, ]) for _ in
range(self.args.maxLengthEnco)] # Batch size * sequence length * input dim
with tf.name_scope('placeholder_decoder'):
self.decoderInputs = [tf.placeholder(tf.int32, [None, ], name='inputs') for _ in
range(self.args.maxLengthDeco)] # Same sentence length for input and output (Right ?)
self.decoderTargets = [tf.placeholder(tf.int32, [None, ], name='targets') for _ in
range(self.args.maxLengthDeco)]
self.decoderWeights = [tf.placeholder(tf.float32, [None, ], name='weights') for _ in
range(self.args.maxLengthDeco)]其實，資料獲取後的Batch的過程可以由Tensorflow標準的batch方法來實現，而這個專案自己Batch各個輸入值，因此對於初學者來說，可以觀察輸入值的構造，對入門RNN還是很有幫助的
2.3 封裝Embedding seq2seq模型
Tensorflow把常用的seq2seq模型都封裝好了，比如embedding_rnn_seq2seq，這是seq2seq一個最簡單的模型，一般的文字任務都會加上attention機制，但這裡都用的短句子，所以attention並考慮，若想加上attention也很簡單，直接修改模型名字為embedding_attention_seq2seq即可，這種封裝雖然使用起來很方便，但是對於使用者來說就是個黑匣子，想要自己去實現一些功能，還得去看程式碼。
decoderOutputs, states = tf.contrib.legacy_seq2seq.embedding_rnn_seq2seq(
self.encoderInputs, # List<[batch=?, inputDim=1]>, list of size args.maxLength
self.decoderInputs, # For training, we force the correct output (feed_previous=False)
encoDecoCell,
self.textData.getVocabularySize(),
self.textData.getVocabularySize(), # Both encoder and decoder have the same number of class
embedding_size=self.args.embeddingSize, # Dimension of each word
output_projection=outputProjection.getWeights() if outputProjection else None,
feed_previous=True if bool(self.args.test) and not bool(self.args.beam_search) else False
# When we test (self.args.test), we use previous output as next input (feed_previous)
)RNN輸出一個句子的過程，其實是對句子裡的每一個詞來做整個詞彙表的softmax分類，取概率最大的詞作為當前位置的輸出詞，但是若詞彙表很大，計算量會很大，那麼通常的解決方法是在詞彙表裡做一個下采樣，取樣的個數通常小於詞彙表，例如詞彙表有50000個，經過取樣後得到4096個樣本集，樣本集裡包含1個正樣本（正確分類）和4095個負樣本，然後對這4096個樣本進行softmax計算作為原來詞彙表的一種樣本估計，這樣計算量會小不少。
在這裡，具體的操作是定義一個全對映outputProjection物件，把隱藏層的輸出對映到整個詞彙表，這種對映需要引數w和b，也就是out＝w＊h＋b，h是隱藏層的輸出，out是整個詞彙表的輸出，可以理解為一個普通的全連線層。假設隱藏層的輸出是512，那麼w的shape就為50000＊512，我們取樣詞彙表的操作可以看作是對w和b引數的取樣，也就是取樣出來的w為4096＊512，用這個w帶入上式計算，能得出4096個輸出，然後計算softmax loss，這個sampled softmax loss是原詞彙表softmax loss的一種近似。outputProjection的定義請看model.py。
outputProjection = ProjectionOp(
(self.textData.getVocabularySize(), self.args.hiddenSize),
scope='softmax_projection',
dtype=self.dtype
)
def sampledSoftmax(labels, inputs):
labels = tf.reshape(labels, [-1, 1]) # Add one dimension (nb of true classes, here 1)
# We need to compute the sampled_softmax_loss using 32bit floats to
# avoid numerical instabilities.
localWt = tf.cast(outputProjection.W_t, tf.float32)
localB = tf.cast(outputProjection.b, tf.float32)
localInputs = tf.cast(inputs, tf.float32)
return tf.cast(
tf.nn.sampled_softmax_loss(
localWt, # Should have shape [num_classes, dim]
localB,
labels,
localInputs,
self.args.softmaxSamples, # The number of classes to randomly sample per batch
self.textData.getVocabularySize()), # The number of classes
self.dtype)2.4 定義損失函式和更新方法
下面定義seq2seq模型的損失函式sequence_loss，其中sequence_loss需要softmax_loss_function引數，這個引數若不指定，那麼就是預設對整個詞彙表的做softmax loss，若需要取樣來加速計算，則要傳入上面定義的sampledSoftmax方法，這個方法的返回值是TF定義的sampled_softmax_loss。更新方法採用預設引數的Adam。
# Finally, we define the loss function
self.lossFct = tf.contrib.legacy_seq2seq.sequence_loss(
decoderOutputs,
self.decoderTargets,
self.decoderWeights,
self.textData.getVocabularySize(),
softmax_loss_function=sampledSoftmax if outputProjection else None # If None, use default SoftMax
)
tf.summary.scalar('loss', self.lossFct) # Keep track of the cost
# Initialize the optimizer
opt = tf.train.AdamOptimizer(
learning_rate=self.args.learningRate,
beta1=0.9,
beta2=0.999,
epsilon=1e-08
)
self.optOp = opt.minimize(self.lossFct)3. Beam Search
在測試模型的時候，比如我輸入問一句話“How are you？“以及一個go開始符，那麼模型就開始輸出第一個詞的候選集，這個候選集裡的每一個詞都有一個概率，一般採用貪婪的思想，就取概率最大的那個詞作為當前輸出，然後把這個詞作為預測第二個詞的輸入再feed進網路，如此迴圈，直到模型輸出end結束符，那麼這句話就輸出完畢。
這種把上一個時刻的輸出當作下一個時刻的輸入的過程在TF模型中由feed_previous引數決定：在訓練模型的時候，我們是知道每一時刻的正確輸入和輸出的，並不需要這個過程，因此feed_previous＝False，而只有在測試的時候，才會需要這種過程feed_previous＝True
而Beam Search是這種貪婪的思想的擴充套件，前面是選擇最大的Top 1概率的詞作為當前輸出，而Beam Search是選擇當前Top k得分的詞，當然這個得分也就是概率，那麼採用這種思想，對一個問題，模型最後的輸出就應該有好幾種回答，這些回答根據得分排序，最終選擇得分最高的句子作為最終輸出。相對於前面貪婪的回答，這種搜尋機制能讓機器人選擇更好的回答。
那麼在DeepQA的基礎上，我們來自己實現一下Beam Search（目前DeepQA並不支援），網上有很多實現的方法，大都是自己編寫decoder，比較複雜。那麼本文就採用一種非常直接的方法，依照Beam Search的思想：feed in上一時刻產生的Top k答案來產生本時刻的候選答案集，然後排序本時刻的候選答案集再選擇Tok k作為本時刻的最終答案，並作為下一時刻輸入，如此迴圈。每一時刻需要記錄當前選擇詞的id，從第一個詞到最後一個詞，這些詞的id構成一種選擇路徑。最後根據引數k，得出k條路徑，每條路徑有一個概率得分。這種情況下，我們是手動feed in各個候選答案，因此feed_previous=False
def beamSearchPredict(self, question, questionSeq=None):
# question為輸入的句子，這裡先把每個詞轉為id，再加上padding和go標示符構成一個batch，這個batch就包含一個句子
batch = self.textData.sentence2enco(question)
if not batch:
return None
if questionSeq is not None: # If the caller want to have the real input
questionSeq.extend(batch.encoderSeqs)
# feedDict為TF的placeholder變數以及對應的資料
# ops為TF的需要計算的網路圖
ops, feedDict = self.model.step(batch)
# 定義softmax操作
def softmax(x):
return np.exp(x) / np.sum(np.exp(x), axis=0)
# path儲存搜尋的路徑，probs裡儲存每個路徑對應的得分（log概率）
# beam_size對應路徑的個數，儲存的位置越靠後，得分越高
beam_size = self.args.beam_size
path = [[] for _ in range(beam_size)]
probs = [[] for _ in range(beam_size)]
# 計算第一個詞的output
output = self.sess.run(ops[0][0], feedDict)
for k in range(len(path)):
# 計算輸出的softmax概率分佈
prob = softmax(output[-1].reshape(-1, ))
# 用對數表示這些概率分佈
log_probs = np.log(np.clip(a=prob, a_min=1e-5, a_max=1))
# 根據概率大小排序，取前Top－beam_size的詞，記錄log概率和詞的id
top_k_indexs = np.argsort(log_probs)[-beam_size:]
path[k].extend([top_k_indexs[k]])
probs[k].extend([log_probs[top_k_indexs[k]]])
# 計算第二個詞到最後一個詞
for i in range(2, self.args.maxLengthDeco):
tmp = []
for k in range(len(path)):
for j in range(len(path[k])):
# feedDict種加入前面的詞的id資料
feedDict[self.model.decoderInputs[j + 1]] = [path[k][j]]
# 輸入feedDict至網路圖，計算當前句子的output
output = self.sess.run(ops[0][0:i], feedDict)
# 取最後一個output（當前詞的輸出）做softmax和log
prob = softmax(output[-1].reshape(-1, ))
log_probs = np.log(np.clip(a=prob, a_min=1e-5, a_max=1))
# 計算概率得分：P(a,b)=P(a|b)*P(b)
# a為當前要選擇的詞，b為上一個已選擇的詞
# Log表示： log P(a,b)=log P(a|b)+ log P(b)
tmp.extend(list(log_probs + probs[k][-1]))
# 假設上一個詞的候選集有三個元素a,b,c
# 那麼分別把這個三個元素作為輸入feed進網路裡，會得出三個output的結果
# 將這些output的概率串聯到一個tmp裡排序，依然取出前Top-beam_size的詞作為當前的詞的候選集
top_k_indexs = np.argsort(tmp)[-beam_size:]
indexs = top_k_indexs % self.textData.getVocabularySize()
# 記錄當前選擇詞的id和log概率得分
for k in range(len(path)):
probs[k].extend([tmp[top_k_indexs[k]]])
path[k].extend([indexs[k]])
return path
還有部分細節沒有展出，詳細請看chatbot.py
4.執行
使用python main來訓練模型，使用python main --test interactive --beam-search --beam-size 3來實時測試對話，下面是loss的圖，混亂度為以loss為自變數的e的指數函式
使用python main --test來生成對話，在save/model/model_predictions.txt中檢視結果
5. 結果
deepQA是基於python 3.5的專案，用python 2.7也可以跑，但是需要稍微修改一些地方，另外專案還提供了訓練好的引數，不想訓練的可以直接匯入訓練好的模型。但是那些模型都是用python 3.5訓練的，如果用python 2.7會匯入不了這些預訓練的模型。我用的是python 2.7，因此只能自己訓練，由於視訊記憶體的限制，訓練的引數很小，效果不太好，這裡給出官方的對話例子：
Q: Hi
A: Hi.
Q: What is your name ?
A: Laura.
Q: What does that mean ?
A: I dunno.
Q: How old are you ?
A: thirty-five.
Q: Will Google hire me ?
A: No.
Q: Tell me the alphabet
A: Fuck you.
Q: That's not nice
A: Yeah.下面給出部分我用Beam Search做出來結果，beam_size=3，因此每個問題有三個答案，得分從高到低：
Q: Hi
A: Hi,
A: Hey, man,
A: Hello ... alright! what suicide!.?
Q: Luke, I am your father!
A: What?
A: Who?
A: We're!?!
Q: Are you ready ?
A: I'm
A: What?
A: Who? ready?
Q: When are you ready ?
A: Tomorrow.
A: Thursday.
A: I, uh., is
Q: How old are you ?
A: Eighteen.
A: Twenty.
A: I'm thirty-four.
Q: How is Laura ?
A: Fine.
A: Tolerable well
A: Good...6. 分析
通過Beam search，有的問題能夠舉例多種回答，但是有的問題的後面的回答直接就失敗了。有以下方面可以還可以提高：
- 更大的訓練集：訓練效果強烈依賴於訓練集的質量，甚至可以把幾個資料集在同一個RNN網路上訓練，猜想可能效果會更好
- 一些引數的調整：如詞頻過濾的大小，詞彙表大小，word embedding的維度大小，RNN的stack層數，輸入輸出句子最大長度，softmax的取樣的大小，以及匯入預訓練的word2vec來加速訓練過程等
- 增加attention機制，對於長句子
- 輸入多個問題，只給出一個答案：這樣能讓網路稍微記住當前答案與前幾個問題有關，目前是當前答案只與當前問題有關，問題與問題之間相互獨立
- 更高階一點，增加記憶網路，顯性將特徵記憶到外部儲存器，然後在記憶中搜索答案返回，這樣可以使記憶長期保留，是目前比較火的研究方向
- 再高階就是增加知識圖譜和一些啟發函式，讓模型自己去外部獲取資訊處理，然後返回答案，類似你問一個問題，即使機器人不知道，但是它可以去上網查資料然後返回你答案，這種level是最高階的。