本模型采用的是字符級別的詩歌生成（pytorch）
環境：
python3.X
pytorch GPU或CPU版本都行，
另外天有點冷，建議用GPU訓練，電腦絕對比暖手寶好用
目錄
專案檔案結構：
資料已經打包：
1、資料集處理
2、構建模型與訓練模型
基于概率語言模型的模型
網路結構
網路輸入輸出
損失函式
3、生成詩歌
4、組態檔
5、生成效果
6、結語
參考：
專案檔案結構：
data:存放預處理好的資料
model:存放訓練好的模型
config.py:組態檔
dataHandler.py:資料預處理及生成詞典
model.py：模型檔案
train.py：訓練模型
generation.py：生成詩歌
poetry.txt：全唐詩，四萬多首，中華民族藝術瑰寶，
資料已經打包：
鏈接：https://pan.baidu.com/s/1UAJFf3kKERm_XR0qRNneig
提取碼：e3y5
1、資料集處理
以四萬首唐詩的文本作為訓練集
它長這樣：
文本中每行是一首詩，且使用冒號分割，前面是標題，后面是正文，且詩的長度不一，
對資料的處理流程大致：
- 讀取文本，按行切分，構成古詩串列，
- 將全角、半角的冒號統一替換成半角的，
- 按冒號切分詩的標題和內容，只保留詩的內容，
- 最后根據詩的內容構建詞典，并將處理好的資料保
處理后的詩歌大概長這樣
代碼如下：
# dataHandler.py
import numpy as np
from config import Config
def pad_sequences(sequences,
maxlen=None,
dtype='int32',
padding='pre',
truncating='pre',
value=0.):
"""
# 填充
code from keras
Pads each sequence to the same length (length of the longest sequence).
If maxlen is provided, any sequence longer
than maxlen is truncated to maxlen.
Truncation happens off either the beginning (default) or
the end of the sequence.
Supports post-padding and pre-padding (default).
Arguments:
sequences: list of lists where each element is a sequence
maxlen: int, maximum length
dtype: type to cast the resulting sequence.
padding: 'pre' or 'post', pad either before or after each sequence.
truncating: 'pre' or 'post', remove values from sequences larger than
maxlen either in the beginning or in the end of the sequence
value: float, value to pad the sequences to the desired value.
Returns:
x: numpy array with dimensions (number_of_sequences, maxlen)
Raises:
ValueError: in case of invalid values for `truncating` or `padding`,
or in case of invalid shape for a `sequences` entry.
"""
if not hasattr(sequences, '__len__'):
raise ValueError('`sequences` must be iterable.')
lengths = []
for x in sequences:
if not hasattr(x, '__len__'):
raise ValueError('`sequences` must be a list of iterables. '
'Found non-iterable: ' + str(x))
lengths.append(len(x))
num_samples = len(sequences)
if maxlen is None:
maxlen = np.max(lengths)
# take the sample shape from the first non empty sequence
# checking for consistency in the main loop below.
sample_shape = tuple()
for s in sequences:
if len(s) > 0: # pylint: disable=g-explicit-length-test
sample_shape = np.asarray(s).shape[1:]
break
x = (np.ones((num_samples, maxlen) + sample_shape) * value).astype(dtype)
for idx, s in enumerate(sequences):
if not len(s): # pylint: disable=g-explicit-length-test
continue # empty list/array was found
if truncating == 'pre':
trunc = s[-maxlen:] # pylint: disable=invalid-unary-operand-type
elif truncating == 'post':
trunc = s[:maxlen]
else:
raise ValueError('Truncating type "%s" not understood' % truncating)
# check `trunc` has expected shape
trunc = np.asarray(trunc, dtype=dtype)
if trunc.shape[1:] != sample_shape:
raise ValueError(
'Shape of sample %s of sequence at position %s is different from '
'expected shape %s'
% (trunc.shape[1:], idx, sample_shape))
if padding == 'post':
x[idx, :len(trunc)] = trunc
elif padding == 'pre':
x[idx, -len(trunc):] = trunc
else:
raise ValueError('Padding type "%s" not understood' % padding)
return x
def load_poetry(poetry_file, max_gen_len):
# 加載資料集
with open(poetry_file, 'r', encoding='utf-8') as f:
lines = f.readlines()
# 將冒號統一成相同格式
lines = [line.replace('：', ':') for line in lines]
# 資料集串列
poetry = []
# 逐行處理讀取到的資料
for line in lines:
# 有且只能有一個冒號用來分割標題
if line.count(':') != 1:
continue
# 后半部分不能包含禁止詞，如果包含，直接跳過這一首詩
_, last_part = line.split(':')
if '（' in line:
last_part = last_part.split('（')[0]
# 長度不能超過最大長度
if len(last_part) > max_gen_len - 2:
continue
# 去除這些有禁用詞的詩
if '【' in line or '_' in line:
continue
poetry.append(last_part.replace('\n', ''))
# 隨機打亂順序
np.random.shuffle(poetry)
return poetry
def get_data(config):
# 1.獲取資料
data = load_poetry(config.poetry_file, config.max_gen_len)
for poetry in data:
print(poetry)
# 2.構建詞典
chars = {c for line in data for c in line}
char_to_ix = {char: ix for ix, char in enumerate(chars)}
char_to_ix['<EOP>'] = len(char_to_ix)
char_to_ix['<START>'] = len(char_to_ix)
char_to_ix['</s>'] = len(char_to_ix)
ix_to_chars = {ix: char for char, ix in list(char_to_ix.items())}
# 3.處理樣本
# 3.1 每首詩加上首位符號
for i in range(0, len(data)):
data[i] = ['<START>'] + list(data[i]) + ['<EOP>']
# 3.2 文字轉id
data_id = [[char_to_ix[w] for w in line] for line in data]
# 3.3 補全既定長度，即填充
pad_data = pad_sequences(data_id,
maxlen=config.poetry_max_len,
padding='pre',
truncating='post',
value=len(char_to_ix) - 1)
# 3.4 保存處理好的資料
np.savez_compressed(config.processed_data_path,
data=pad_data,
word2ix=char_to_ix,
ix2word=ix_to_chars)
return pad_data, char_to_ix, ix_to_chars
if __name__ == '__main__':
config = Config()
pad_data, char_to_ix, ix_to_chars = get_data(config)
for l in pad_data[:10]:
print(l)
n = 0
for k, v in char_to_ix.items():
print(k, v)
if n > 10:
break
n += 1
n = 0
for k, v in ix_to_chars.items():
print(k, v)
if n > 10:
break
n += 1
2、構建模型與訓練模型
基于概率語言模型的模型
我們是基于n-gram語言模型的思想，n-gram模型作了一個n?1階的Markov假設，即認為一個詞出現的概率只與它前面的n?1個詞相關，所以我們可以通過前n個詞預測下一個詞，公式表示為：
STM可以考慮序列之間的聯系，所以我們選擇LSTM作為本次練習的網路，LSTM詳細原理請參考：(31條訊息) LSTM這一篇就夠了_yingqubaifumei的博客-CSDN博客_lstm細胞狀態
https://blog.csdn.net/yingqubaifumei/article/details/100888147?ops_request_misc=%257B%2522request%255Fid%2522%253A%2522164289666916780357223324%2522%252C%2522scm%2522%253A%252220140713.130102334..%2522%257D&request_id=164289666916780357223324&biz_id=0&utm_medium=distribute.pc_search_result.none-task-blog-2~all~top_positive~default-1-100888147.first_rank_v2_pc_rank_v29&utm_term=LSTM&spm=1018.2226.3001.4187，
網路結構
基于語言的概率模型本質就是分類模型，基于前面n個字對下一個字進行預測，找到概率最大的字便是預測結果，其網路結構如下：
從圖中可以看出，本模型使用了1層詞嵌入層，2層LSTM，一層全連接層，
網路輸入輸出
訓練資料 x和標簽 y，將詩的內容錯開一位分別作為資料和標簽，舉個例子，假設有詩是“床前明月光，疑是地上霜，舉頭望明月，低頭思故鄉，”，則資料為“床前明月光，疑是地上霜，舉頭望明月，低頭思故鄉，”，標簽為“床前明月光，疑是地上霜，舉頭望明月，低頭思故鄉，”，兩者一一對應，y 是 x 中每個位置的下一個字符，
以字符的形式舉例是為了方便理解，實際上不論是x還是y，都是將字符編碼后的編號序列（數字），這樣才能輸入神經網路，
損失函式
損失函式采用交叉熵損失函式CrossEntropyLoss
代碼如下：
# model.py
import torch
import torch.nn as nn
from torch.autograd import Variable
import torch.nn.functional as F
class PoetryModel(nn.Module):
def __init__(self, vocab_size, embedding_dim, hidden_dim, device, layer_num):
super(PoetryModel, self).__init__()
self.hidden_dim = hidden_dim
# 創建embedding層
self.embeddings = nn.Embedding(vocab_size, embedding_dim)
# 創建lstm層,引數是輸入輸出的維度
self.lstm = nn.LSTM(embedding_dim, self.hidden_dim, num_layers=layer_num)
# 創建一個線性層
self.linear1 = nn.Linear(self.hidden_dim, vocab_size)
# 創建一個dropout層，訓練時作用在線性層防止過擬合
self.dropout = nn.Dropout(0.2)
self.device = device
def forward(self, inputs, hidden):
seq_len, batch_size = inputs.size()
# 將one-hot形式的input在嵌入矩陣中轉換成嵌入向量，torch.Size([length, batch_size, embedding_size])
embeds = self.embeddings(inputs)
# 經過lstm層，該層有2個輸入，當前x和t=0時的(c,a),
# output:torch.Size([length, batch_size, hidden_idm]), 每一個step的輸出
# hidden: tuple(torch.Size([layer_num, 32, 256]) torch.Size([1, 32, 256])) # 最后一層輸出的ct 和 ht, 在這里是沒有用的
output, hidden = self.lstm(embeds, hidden)
# 經過線性層，relu激活層 先轉換成（max_len*batch_size, 256)維度，再過線性層（length, vocab_size)
# output = F.relu(self.linear1(output.view(seq_len*batch_size, -1)))
output = F.log_softmax(self.linear1(output.view(seq_len * batch_size, -1)), dim=1)
# 輸出最終結果，與hidden結果
return output, hidden
def init_hidden(self, layer_num, batch_size):
return (Variable(torch.zeros(layer_num, batch_size, self.hidden_dim)).cuda(),
Variable(torch.zeros(layer_num, batch_size, self.hidden_dim)).cuda())
# train.py
import os
import torch
import torch.utils.data as Data
import torch.nn as nn
import torch.optim as optim
from model import PoetryModel
from dataHandler import *
from config import Config
class TrainModel(object):
def __init__(self):
os.environ["CUDA_VISIBLE_DEVICES"] = '0'
self.config = Config()
self.device = torch.device('cuda') if self.config.use_gpu else torch.device('cpu')
def train(self, data_loader, model, optimizer, criterion, char_to_ix, ix_to_chars):
for epoch in range(self.config.epoch_num):
for step, x in enumerate(data_loader):
# 1.處理資料
# x: (batch_size,max_len) ==> (max_len, batch_size)
x = x.long().transpose(1, 0).contiguous()
x = x.to(self.device)
optimizer.zero_grad()
# input,target: (max_len, batch_size-1)
input_, target = x[:-1, :], x[1:, :]
target = target.view(-1)
# 初始化hidden為(c0, h0): ((layer_num， batch_size, hidden_dim)，(layer_num， batch_size, hidden_dim)）
hidden = model.init_hidden(self.config.layer_num, x.size()[1])
# 2.前向計算
# print(input.size(), hidden[0].size(), target.size())
output, _ = model(input_, hidden)
loss = criterion(output, target) # output:(max_len*batch_size,vocab_size), target:(max_len*batch_size)
# 反向計算梯度
loss.backward()
# 權重更新
optimizer.step()
if step == 0:
print('epoch: %d,loss: %f' % (epoch, loss.data))
if epoch % 5 == 0:
# 保存模型
torch.save(model.state_dict(), '%s_%s.pth' % (self.config.model_prefix, epoch))
# 分別以這幾個字作為詩歌的第一個字，生成一首藏頭詩，用于看看訓練時的效果
word = '春江花月夜涼如水'
gen_poetry = ''.join(self.generate_head_test(model, word, char_to_ix, ix_to_chars))
print(gen_poetry)
def run(self):
# 1 獲取資料
data, char_to_ix, ix_to_chars = get_data(self.config)
vocab_size = len(char_to_ix)
print('樣本數：%d' % len(data))
print('詞典大小： %d' % vocab_size)
# 2 設定dataloader
data = torch.from_numpy(data)
data_loader = Data.DataLoader(data,
batch_size=self.config.batch_size,
shuffle=True,
num_workers=0)
# 3 創建模型
model = PoetryModel(vocab_size=vocab_size,
embedding_dim=self.config.embedding_dim,
hidden_dim=self.config.hidden_dim,
device=self.device,
layer_num=self.config.layer_num)
model.to(self.device)
# 4 創建優化器
optimizer = optim.Adam(model.parameters(), lr=self.config.lr, weight_decay=self.config.weight_decay)
# 5 創建損失函式,使用與logsoftmax的輸出
criterion = nn.CrossEntropyLoss()
# 6.訓練
self.train(data_loader, model, optimizer, criterion, char_to_ix, ix_to_chars)
def generate_head_test(self, model, head_sentence, word_to_ix, ix_to_word):
"""生成藏頭詩"""
poetry = []
head_char_len = len(head_sentence) # 要生成的句子的數量
sentence_len = 0 # 當前句子的數量
pre_char = '<START>' # 前一個已經生成的字
# 準備第一步要輸入的資料
input = (torch.Tensor([word_to_ix['<START>']]).view(1, 1).long()).to(self.device)
hidden = model.init_hidden(self.config.layer_num, 1)
for i in range(self.config.max_gen_len):
# 前向計算出概率最大的當前詞
output, hidden = model(input, hidden)
top_index = output.data[0].topk(1)[1][0].item()
char = ix_to_word[top_index]
# 句首的字用藏頭字代替
if pre_char in ['，', '！', '<START>']:
if sentence_len == head_char_len:
break
else:
char = head_sentence[sentence_len]
sentence_len += 1
input = (input.data.new([word_to_ix[char]])).view(1,1)
else:
input = (input.data.new([top_index])).view(1,1)
poetry.append(char)
pre_char = char
return poetry3、生成詩歌
# -*- coding: utf-8 -*-
# generation.py
import os
from config import Config
import numpy as np
from model import PoetryModel
import torch
class Sample(object):
def __init__(self):
self.config = Config()
self.device = torch.device('cuda') if self.config.use_gpu else torch.device('cpu')
self.processed_data_path = self.config.processed_data_path
self.model_path = self.config.model_path
self.max_len = self.config.max_gen_len
self.sentence_max_len = self.config.sentence_max_len
self.load_data()
self.load_model()
def load_data(self):
if os.path.exists(self.processed_data_path):
data = np.load(self.processed_data_path, allow_pickle=True)
self.data, self.word_to_ix, self.ix_to_word = data['data'], data['word2ix'].item(), data['ix2word'].item()
def load_model(self):
model = PoetryModel(len(self.word_to_ix),
self.config.embedding_dim,
self.config.hidden_dim,
self.device,
self.config.layer_num)
map_location = lambda s, l: s
state_dict = torch.load(self.config.model_path, map_location=map_location)
model.load_state_dict(state_dict)
model.to(self.device)
self.model = model
def generate_random(self, start_words='<START>'):
"""自由生成一首詩歌"""
poetry = []
sentence_len = 0
input = (torch.Tensor([self.word_to_ix[start_words]]).view(1, 1).long()).to(self.device)
hidden = self.model.init_hidden(self.config.layer_num, 1)
for i in range(self.max_len):
# 前向計算出概率最大的當前詞
output, hidden = self.model(input, hidden)
top_index = output.data[0].topk(1)[1][0].item()
# _probas = output.data[0].cpu().numpy() # 在GPU上的Tensor沒辦法直接轉numpy，先轉到CPU再轉成numpy
# # 按照出現概率，對所有token倒序排列
# p_args = np.argsort(-_probas)[:2]
# # 排列后的概率順序
# p = _probas[p_args]
# # 先對概率歸一
# p = p / sum(p)
# # 再按照預測出的概率，隨機選擇一個詞作為預測結果
# choice_index = np.random.choice(len(p), p=p)
# top_index = p_args[choice_index]
char =self.ix_to_word[top_index]
# 遇到終結符則輸出
if char == '<EOP>':
break
# 有8個句子則停止預測
if char in ['，', '！']:
sentence_len += 1
if sentence_len == 8:
poetry.append(char)
break
input = (input.data.new([top_index])).view(1, 1)
poetry.append(char)
return poetry
def generate_head(self, head_sentence):
"""生成藏頭詩"""
poetry = []
head_char_len = len(head_sentence) # 要生成的句子的數量
sentence_len = 0 # 當前句子的數量
pre_char = '<START>' # 前一個已經生成的字
# 準備第一步要輸入的資料
input = (torch.Tensor([self.word_to_ix['<START>']]).view(1, 1).long()).to(self.device)
hidden = self.model.init_hidden(self.config.layer_num, 1)
for i in range(self.max_len):
# 前向計算出概率最大的當前詞
output, hidden = self.model(input, hidden)
top_index = output.data[0].topk(1)[1][0].item()
# _probas = output.data[0].cpu().numpy() # 在GPU上的Tensor沒辦法直接轉numpy，先轉到CPU再轉成numpy
# # 按照出現概率，對所有token倒序排列
# p_args = np.argsort(-_probas)[:3]
# # 排列后的概率順序
# p = _probas[p_args]
# # 先對概率歸一
# p = p / sum(p)
# # 再按照預測出的概率，隨機選擇一個詞作為預測結果
# choice_index = np.random.choice(len(p), p=p)
# top_index = p_args[choice_index]
char = self.ix_to_word[top_index]
# 句首的字用藏頭字代替
if pre_char in ['，', '！', '<START>']:
if sentence_len == head_char_len:
break
else:
char = head_sentence[sentence_len]
sentence_len += 1
input = (input.data.new([self.word_to_ix[char]])).view(1,1)
else:
input = (input.data.new([top_index])).view(1,1)
poetry.append(char)
pre_char = char
return poetry
def generate_poetry(self, mode=1, head_sentence=None):
"""
模式一：隨機生成詩歌
模式二：生成藏頭詩
:return:
"""
poetry = ''
if mode == 1 or (mode == 2 and head_sentence is None):
poetry = ''.join(self.generate_random())
if mode == 2 and head_sentence is not None:
head_sentence = head_sentence.replace(',', u'，').replace('.', u'，').replace('?', u'？')
poetry = ''.join(self.generate_head(head_sentence))
return poetry
if __name__ == '__main__':
obj = Sample()
poetry1 = obj.generate_poetry(mode=2, head_sentence="碧海潮生")
print(poetry1)
4、組態檔
為了方便修改和除錯模型，基本上所有的可以改的引數都放到配置模型檔案中，很簡單
# -*- encoding: utf-8 -*-
# config.py
class Config(object):
processed_data_path = "data/tang.npz" # 保存的資料預處理資料
model_path = 'model/tang_5.pth' # 載入的訓練好的模型
model_prefix = 'model/tang' # 模型的保存
batch_size = 64 # batch_size
epoch_num = 10 # 訓練的迭代次數epoch
embedding_dim = 128 # 嵌入層的維度
hidden_dim = 128 # LSTM的隱藏層的維度
layer_num = 2 # LSTM的層數
lr = 0.001 # 初始化學習率
weight_decay = 1e-4 # Adam優化器的weight_decay引數
use_gpu = True # 是否使用GPU訓練
poetry_file = 'poetry.txt' # 詩歌檔案名稱
max_gen_len = 200 # 生成詩歌最長長度
sentence_max_len = 4 # 生成詩歌的最長句子
poetry_max_len = 125
sample_max_len = poetry_max_len - 15、生成效果
怎么說呢，有點那味了，
桃殿開華碧色沈，
花間似水水間流，
影中云落風明月，
落日生光未出分，
碧云行自在，
海路幾悠游，
潮暗遙還夜，
生游日見長，
6、結語
初次嘗試用pytorch，有紕漏之處，還望指正，模型還有待改進，
參考：
基于回圈神經網路(RNN)的古詩生成器_python_腳本之家 (jb51.net)https://www.jb51.net/article/137118.htm
(31條訊息) TensorFlow練手專案二：基于回圈神經網路(RNN)的古詩生成器_筆墨留年，-CSDN博客_rnn專案https://blog.csdn.net/aaronjny/article/details/79677457
轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/420055.html
標籤：AI