NLP教程:教你如何自動生成對聯
桃符早易硃紅紙,楊柳輕搖翡翠群 ——FlyAI Couplets
體驗對對聯Demo: https://www.flyai.com/couplets
迴圈神經網路最重要的特點就是可以將序列作為輸入和輸出,而對聯的上聯和下聯都是典型的序列文字,那麼,能否使用神經網路進行對對聯呢?答案是肯定的。本專案使用網路上收集的對聯資料集地址作為訓練資料,運用Seq2Seq + 注意力機制網路完成了根據上聯對下聯的任務。
專案流程
- 資料處理
- Seq2Seq + Attention 模型解讀
- 模型程式碼實現
- 訓練神經網路
資料處理
建立詞向量字典和詞袋字典
在原始資料集中,對聯中每個漢字使用空格進行分割,格式如下所示:
室 內 崇 蘭 映 日,林 間 修 竹 當 風 翠 岸 青 荷 , 琴 曲 瀟 瀟 情 輾 轉,寒 山 古 月 , 風 聲 瑟 瑟 意 彷 徨
由於每個漢字表示一個單一的詞,因此不需要對原始資料進行分詞。在獲取原始資料之後,需要建立兩個字典,分別是字到詞向量的字典和字到詞袋的字典,這樣做是為了將詞向量輸入到網路中,而輸出處使用詞袋進行分類。在詞袋模型中,新增三個關鍵字 ' “ ', ' ” ' 和 ' ~ ' ,分別代表輸入輸出的起始,結束和空白處的補零,其關鍵字分別為1,2,0。
class Processor(Base): ## Processor是進行資料處理的類 def __init__(self): super(Processor, self).__init__() embedding_path = os.path.join(DATA_PATH, 'embedding.json') ##載入詞向量字典 words_list_path = os.path.join(DATA_PATH, 'words.json') ## 載入詞袋列表 with open(embedding_path, encoding='utf-8') as f: self.vocab = json.loads(f.read()) with open(words_list_path, encoding='utf-8') as f: word_list = json.loads(f.read()) self.word2ix = {w:i for i,w in enumerate(word_list, start = 3)} self.word2ix['“'] = 1 ##句子開頭為1 self.word2ix['”'] = 2 ##句子結尾為2 self.word2ix['~'] = 0 ##padding的內容為0 self.ix2word = {i:w for w,i in self.word2ix.items()} self.max_sts_len = 40 ##最大序列長度
對上聯進行詞向量編碼
def input_x(self, upper): ##upper為輸入的上聯 word_list = [] #review = upper.strip().split(' ') review = ['“'] + upper.strip().split(' ') + ['”'] ##開頭加符號1,結束加符號2 for word in review: embedding_vector = self.vocab.get(word) if embedding_vector is not None: if len(embedding_vector) == 200: # 給出現在編碼詞典中的詞彙編碼 embedding_vector = list(map(lambda x: float(x),embedding_vector)) ## convert element type from str to float in the list word_list.append(embedding_vector) if len(word_list) >= self.max_sts_len: word_list = word_list[:self.max_sts_len] origanal_len = self.max_sts_len else: origanal_len = len(word_list) for i in range(len(word_list), self.max_sts_len): word_list.append([0 for j in range(200)]) ## 詞向量維度為200 word_list.append([origanal_len for j in range(200)]) ## 最後一行元素為句子實際長度 word_list = np.stack(word_list) return word_list
對真實下聯進行詞袋編碼
def input_y(self, lower): word_list = [1] ##開頭加起始符號1 for word in lower: word_idx = self.word2ix.get(word) if word_idx is not None: word_list.append(word_idx) word_list.append(2) ##結束加終止符號2 origanal_len = len(word_list) if len(word_list) >= self.max_sts_len: origanal_len = self.max_sts_len word_list = word_list[:self.max_sts_len] else: origanal_len = len(word_list) for i in range(len(word_list), self.max_sts_len): word_list.append(0) ## 不夠長度則補0 word_list.append(origanal_len) ##最後一個元素為句子長度 return word_list
Seq2Seq + Attention 模型解讀
Seq2Seq 模型可以被認為是一種由編碼器和解碼器組成的翻譯器,其結構如下圖所示:
編碼器(Encoder)和解碼器(Decoder)通常使用RNN構成,為提高效果,RNN通常使用LSTM或RNN,在上圖中的RNN即是使用LSTM。Encoder將輸入翻譯為中間狀態C,而Decoder將中間狀態翻譯為輸出。序列中每一個時刻的輸出由的隱含層狀態,前一個時刻的輸出值及中間狀態C共同決定。
Attention 機制
在早先的Seq2Seq模型中,中間狀態C僅由最終的隱層決定,也就是說,源輸入中的每個單詞對C的重要性是一樣的。這種方式在一定程度上降低了輸出對位置的敏感性。而Attention機制正是為了彌補這一缺陷而設計的。在Attention機制中,中間狀態C具有了位置資訊,即每個位置的C都不相同,第i個位置的C由下面的公式決定:
公式中,Ci代表第i個位置的中間狀態C,Lx代表輸入序列的全部長度,hj是第j個位置的Encoder隱層輸出,而aij為第i個C與第j個h之間的權重。通過這種方式,對於每個位置的源輸入就產生了不同的C,也就是實現了對不同位置單詞的‘注意力’。權重aij有很多的計算方式,本專案中使用使用小型神經網路進行對映的方式產生aij。
模型程式碼實現
Encoder
Encoder的結構非常簡單,是一個簡單的RNN單元,由於本專案中輸入資料是已經編碼好的詞向量,因此不需要使用nn.Embedding() 對input進行編碼。
class Encoder(nn.Module): def __init__(self, embedding_dim, hidden_dim, num_layers=2, dropout=0.2): super().__init__() self.embedding_dim = embedding_dim #詞向量維度,本專案中是200維 self.hidden_dim = hidden_dim #RNN隱層維度 self.num_layers = num_layers #RNN層數 self.dropout = dropout#dropout self.rnn = nn.GRU(embedding_dim, hidden_dim, num_layers=num_layers, dropout=dropout) self.dropout = nn.Dropout(dropout) #dropout層 def forward(self, input_seqs, input_lengths, hidden=None): # src = [sent len, batch size] embedded = self.dropout(input_seqs) # embedded = [sent len, batch size, emb dim] packed = torch.nn.utils.rnn.pack_padded_sequence(embedded, input_lengths) #將輸入轉換成torch中的pack格式,使得RNN輸入的是真實長度的句子而非padding後的 #outputs, hidden = self.rnn(packed, hidden) outputs, hidden = self.rnn(packed) outputs, output_lengths = torch.nn.utils.rnn.pad_packed_sequence(outputs) # outputs, hidden = self.rnn(embedded, hidden) # outputs = [sent len, batch size, hid dim * n directions] # hidden = [n layers, batch size, hid dim] # outputs are always from the last layer return outputs, hidden
Attentation機制
Attentation權重的計算方式主要有三種,本專案中使用concatenate的方式進行注意力權重的運算。程式碼實現如下:
class Attention(nn.Module): def __init__(self, hidden_dim): super(Attention, self).__init__() self.hidden_dim = hidden_dim self.attn = nn.Linear(self.hidden_dim * 2, hidden_dim) self.v = nn.Parameter(torch.rand(hidden_dim)) self.v.data.normal_(mean=0, std=1. / np.sqrt(self.v.size(0))) def forward(self, hidden, encoder_outputs): #encoder_outputs:(seq_len, batch_size, hidden_size) #hidden:(num_layers * num_directions, batch_size, hidden_size) max_len = encoder_outputs.size(0) h = hidden[-1].repeat(max_len, 1, 1) # (seq_len, batch_size, hidden_size) attn_energies = self.score(h, encoder_outputs)# compute attention score return F.softmax(attn_energies, dim=1)# normalize with softmax def score(self, hidden, encoder_outputs): # (seq_len, batch_size, 2*hidden_size)-> (seq_len, batch_size, hidden_size) energy = torch.tanh(self.attn(torch.cat([hidden, encoder_outputs], 2))) energy = energy.permute(1, 2, 0)# (batch_size, hidden_size, seq_len) v = self.v.repeat(encoder_outputs.size(1), 1).unsqueeze(1)# (batch_size, 1, hidden_size) energy = torch.bmm(v, energy)# (batch_size, 1, seq_len) return energy.squeeze(1)# (batch_size, seq_len)
Decoder
Decoder同樣是一個RNN網路,它的輸入有三個,分別是句子初始值,hidden tensor 和Encoder的output tensor。在本專案中句子的初始值為‘“’代表的數字1。由於初始值tensor使用的是詞袋編碼,需要將詞袋索引也對映到詞向量維度,這樣才能與其他tensor合併。完整的Decoder程式碼如下所示:
class Decoder(nn.Module): def __init__(self, output_dim, embedding_dim, hidden_dim, num_layers=2, dropout=0.2): super().__init__() self.embedding_dim = embedding_dim ##編碼維度 self.hid_dim = hidden_dim ##RNN隱層單元數 self.output_dim = output_dim ##詞袋大小 self.num_layers = num_layers ##RNN層數 self.dropout = dropout self.embedding = nn.Embedding(output_dim, embedding_dim) self.attention = Attention(hidden_dim) self.rnn = nn.GRU(embedding_dim + hidden_dim, hidden_dim, num_layers=num_layers, dropout=dropout) self.out = nn.Linear(embedding_dim + hidden_dim * 2, output_dim) self.dropout = nn.Dropout(dropout) def forward(self, input, hidden, encoder_outputs): # input = [bsz] # hidden = [n layers * n directions, batch size, hid dim] # encoder_outputs = [sent len, batch size, hid dim * n directions] input = input.unsqueeze(0) # input = [1, bsz] embedded = self.dropout(self.embedding(input)) # embedded = [1, bsz, emb dim] attn_weight = self.attention(hidden, encoder_outputs) # (batch_size, seq_len) context = attn_weight.unsqueeze(1).bmm(encoder_outputs.transpose(0, 1)).transpose(0, 1) # (batch_size, 1, hidden_dim * n_directions) # (1, batch_size, hidden_dim * n_directions) emb_con = torch.cat((embedded, context), dim=2) # emb_con = [1, bsz, emb dim + hid dim] _, hidden = self.rnn(emb_con, hidden) # outputs = [sent len, batch size, hid dim * n directions] # hidden = [n layers * n directions, batch size, hid dim] output = torch.cat((embedded.squeeze(0), hidden[-1], context.squeeze(0)), dim=1) output = F.log_softmax(self.out(output), 1) # outputs = [sent len, batch size, vocab_size] return output, hidden, attn_weight
在此之上,定義一個完整的Seq2Seq類,將Encoder和Decoder結合起來。在該類中,有一個叫做teacher_forcing_ratio的引數,作用為在訓練過程中強制使得網路模型的輸出在一定概率下更改為ground truth,這樣在反向傳播時有利於模型的收斂。該類中有兩個方法,分別在訓練和預測時應用。Seq2Seq類名稱為Net,程式碼如下所示:
class Net(nn.Module): def __init__(self, encoder, decoder, device, teacher_forcing_ratio=0.5): super().__init__() self.encoder = encoder.to(device) self.decoder = decoder.to(device) self.device = device self.teacher_forcing_ratio = teacher_forcing_ratio def forward(self, src_seqs, src_lengths, trg_seqs): # src_seqs = [sent len, batch size] # trg_seqs = [sent len, batch size] batch_size = src_seqs.shape[1] max_len = trg_seqs.shape[0] trg_vocab_size = self.decoder.output_dim # tensor to store decoder outputs outputs = torch.zeros(max_len, batch_size, trg_vocab_size).to(self.device) # hidden used as the initial hidden state of the decoder # encoder_outputs used to compute context encoder_outputs, hidden = self.encoder(src_seqs, src_lengths) # first input to the decoder is the <sos> tokens output = trg_seqs[0, :] for t in range(1, max_len): # skip sos output, hidden, _ = self.decoder(output, hidden, encoder_outputs) outputs[t] = output teacher_force = random.random() < self.teacher_forcing_ratio output = (trg_seqs[t] if teacher_force else output.max(1)[1]) return outputs def predict(self, src_seqs, src_lengths, max_trg_len=30, start_ix=1): max_src_len = src_seqs.shape[0] batch_size = src_seqs.shape[1] trg_vocab_size = self.decoder.output_dim outputs = torch.zeros(max_trg_len, batch_size, trg_vocab_size).to(self.device) encoder_outputs, hidden = self.encoder(src_seqs, src_lengths) output = torch.LongTensor([start_ix] * batch_size).to(self.device) attn_weights = torch.zeros((max_trg_len, batch_size, max_src_len)) for t in range(1, max_trg_len): output, hidden, attn_weight = self.decoder(output, hidden, encoder_outputs) outputs[t] = output output = output.max(1)[1] #attn_weights[t] = attn_weight return outputs, attn_weights
訓練神經網路
訓練過程包括定義損失函式,優化器,資料處理,梯隊下降等過程。由於網路中tensor型狀為(sentence len, batch, embedding), 而載入的資料形狀為(batch, sentence len, embedding),因此有些地方需要進行轉置。
定義網路,輔助類等程式碼如下所示:
# 資料獲取輔助類 data = Dataset() en=Encoder(200,64) ##詞向量維度200,rnn隱單元64 de=Decoder(9133,200,64) ##詞袋大小9133,詞向量維度200,rnn隱單元64 network = Net(en,de,device) ##定義Seq2Seq例項 loss_fn = nn.CrossEntropyLoss() ##使用交叉熵損失函式 optimizer = Adam(network.parameters()) ##使用Adam優化器 model = Model(data)
訓練過程如下所示:
lowest_loss = 10 # 得到訓練和測試的資料 for epoch in range(args.EPOCHS): network.train() # 得到訓練和測試的資料 x_train, y_train, x_test, y_test = data.next_batch(args.BATCH)# 讀取資料; shape:(sen_len,batch,embedding) #x_train shape: (batch,sen_len,embed_dim) #y_train shape: (batch,sen_len) batch_len = y_train.shape[0] #input_lengths = [30 for i in range(batch_len)] ## batch內每個句子的長度 input_lengths = x_train[:,-1,0] input_lengths = input_lengths.tolist() #input_lengths = list(map(lambda x: int(x),input_lengths)) input_lengths = [int(x) for x in input_lengths] y_lengths = y_train[:,-1] y_lengths = y_lengths.tolist() x_train = x_train[:,:-1,:] ## 除去長度資訊 x_train = torch.from_numpy(x_train) #shape:(batch,sen_len,embedding) x_train = x_train.float().to(device) y_train = y_train[:,:-1] ## 除去長度資訊 y_train = torch.from_numpy(y_train) #shape:(batch,sen_len) y_train = torch.LongTensor(y_train) y_train = y_train.to(device) seq_pairs = sorted(zip(x_train.contiguous(), y_train.contiguous(),input_lengths), key=lambda x: x[2], reverse=True) #input_lengths = sorted(input_lengths, key=lambda x: input_lengths, reverse=True) x_train, y_train,input_lengths = zip(*seq_pairs) x_train = torch.stack(x_train,dim=0).permute(1,0,2).contiguous() y_train = torch.stack(y_train,dim=0).permute(1,0).contiguous() outputs = network(x_train,input_lengths,y_train) #_, prediction = torch.max(outputs.data, 2) optimizer.zero_grad() outputs = outputs.float() # calculate the loss according to labels loss = loss_fn(outputs.view(-1, outputs.shape[2]), y_train.view(-1)) # backward transmit loss loss.backward() # adjust parameters using Adam optimizer.step() print(loss) # 若測試準確率高於當前最高準確率,則儲存模型 if loss < lowest_loss: lowest_loss = loss model.save_model(network, MODEL_PATH, overwrite=True) print("step %d, best lowest_loss %g" % (epoch, lowest_loss)) print(str(epoch) + "/" + str(args.EPOCHS))
小結
通過使用Seq2Seq + Attention模型,我們完成了使用神經網路對對聯的任務。經過十餘個週期的訓練後,神經網路將會對出與上聯字數相同的下聯,但是,若要對出工整的對聯,還需訓練更多的週期,讀者也可以嘗試其他的方法來提高對仗的工整性。
體驗對對聯Demo: https://www.flyai.com/couplets
獲取更多專案樣例開原始碼 請PC端訪問: www.flyai.com