用RNN神经网络，生成周杰伦式歌词
文本生成任务是一种常见的自然语言处理任务，输入一个开始词能够预测出后面的词序列。本案例将会使用循环神经网络来实现周杰伦歌词生成任务 …
文本生成任务是一种常见的自然语言处理任务，输入一个开始词能够预测出后面的词序列。本案例将会使用循环神经网络来实现周杰伦歌词生成任务。
导入工具包
import torch
import jieba
from torch.utils.data import DataLoader
import torch.nn as nn
import torch.optim as optim
import time数据集
我们收集了周杰伦从第一张专辑《Jay》到第十张专辑《跨时代》中的歌词，来训练神经网络模型，当模型训练好后，我们就可以用这个模型来创作歌词。数据集如下：
想要有直升机
想要和你飞到宇宙去
想要和你融化在一起
融化在宇宙里
我每天每天每天在想想想想著你
这样的甜蜜
让我开始相信命运
感谢地心引力
让我碰到你
漂亮的让我面红的可爱女人
...该数据集共有 5819 行文本。
构建词表
在进行自然语言处理任务之前，首要做的就是构建词表。
所谓的词表就是将语料进行分词，然后给每一个词分配一个唯一的编号，便于我们送入词嵌入层。
接下来, 我们对周杰伦歌词的数据进行处理构建词表，具体流程如下：
- 获取文本数据
- 分词，并进行去重
- 构建词表
def build_vocab():
# 数据集位置
file_name = 'data/jaychou_lyrics.txt'
# 分词结果存储位置
# 唯一词列表
unique_words = []
# 每行文本分词列表
all_words = []
# 遍历数据集中的每一行文本
for line in open(file_name, 'r', encoding='utf-8'):
# 使用jieba分词,分割结果是一个列表
words = jieba.lcut(line)
# print(words)
# 所有的分词结果存储到all_words，其中包含重复的词组
all_words.append(words)
# 遍历分词结果，去重后存储到unique_words
for word in words:
if word not in unique_words:
unique_words.append(word)
# 语料中词的数量
word_count = len(unique_words)
# 词到索引映射
word_to_index = {word: idx for idx, word in enumerate(unique_words)}
# 歌词文本用词表索引表示
corpus_idx = []
# 遍历每一行的分词结果
for words in all_words:
temp = []
# 获取每一行的词，并获取相应的索引
for word in words:
temp.append(word_to_index[word])
# 在每行词之间添加空格隔开
temp.append(word_to_index[' '])
# 获取当前文档中每个词对应的索引
corpus_idx.extend(temp)
return unique_words, word_to_index, word_count, corpus_idx
if __name__ == "__main__":
# 获取数据
unique_words, word_to_index, unique_word_count, corpus_idx = build_vocab()
print("词的数量：n",unique_word_count)
print("去重后的词:n",unique_words)
print("每个词的索引：n",word_to_index)
print("当前文档中每个词对应的索引：n",corpus_idx)我们的词典主要包含了:
unique_words: 存储了每个词
word_to_index: 存储了词到编号的映射
构建数据集对象
我们在训练的时候，为了便于读取语料，并送入网络，所以我们会构建一个Dataset对象
class LyricsDataset(torch.utils.data.Dataset):
def __init__(self, corpus_idx, num_chars):
# 文档数据中词的索引
self.corpus_idx = corpus_idx
# 每个句子中词的个数
self.num_chars = num_chars
# 文档数据中词的数量，不去重
self.word_count = len(self.corpus_idx)
# 句子数量
self.number = self.word_count // self.num_chars
# len(obj)时自动调用此方法
def __len__(self):
# 返回句子数量
return self.number
# obj[idx]时自动调用此方法
def __getitem__(self, idx):
# idx指词的索引，并将其修正索引值到文档的范围里面
"""
我 爱你 中国 , 亲爱 的 母亲
word_count: 7
num_chars: 2 一个句子由num_chars个词组成
word_count-num_chars-2: 7-2-1=4 -1:网络预测结果y在x上后移一个词取值-1
idx=5->start=4
"""
start = min(max(idx, 0), self.word_count - self.num_chars - 1)
end = start + self.num_chars
# 输入值
x = self.corpus_idx[start: end]
# 网络预测结果（目标值）
y = self.corpus_idx[start + 1: end + 1]
# 返回结果
return torch.tensor(x), torch.tensor(y)
if __name__ == "__main__":
# 获取数据
unique_words, word_to_index, unique_word_count, corpus_idx = build_vocab()
# 数据获取实例化
dataset = LyricsDataset(corpus_idx, 5)
# 查看句子数量
print('句子数量：', len(dataset))
# x, y = dataset.__getitem__(0)
x, y = dataset[0]
print("网络输入值：", x)
print("目标值：", y)输出结果：
句子数量： 9827
网络输入值： tensor([ 0, 1, 2, 3, 40])
目标值： tensor([ 1, 2, 3, 40, 0])构建网络模型
我们用于实现《歌词生成》的网络模型，主要包含了三个层:
-
词嵌入层: 用于将语料转换为词向量
-
循环网络层: 提取句子语义
-
全连接层: 输出对词典中每个词的预测概率
# 模型构建
class TextGenerator(nn.Module):
def __init__(self, unique_word_count):
super(TextGenerator, self).__init__()
# 初始化词嵌入层: 语料中词的数量, 词向量的维度为128
self.ebd = nn.Embedding(unique_word_count, 128)
# 循环网络层: 词向量维度128, 隐藏向量维度256, 网络层数1
self.rnn = nn.RNN(128, 256, 1)
# 输出层: 特征向量维度256与隐藏向量维度相同, 词表中词的个数
self.out = nn.Linear(256, unique_word_count)
def forward(self, inputs, hidden):
# 输出维度: (batch, seq_len, 词向量维度128)
# batch：句子数量
# seq_len： 句子长度， 每个句子由多少个词 词数量
embed = self.ebd(inputs)
# rnn层x的表示形式为(seq_len, batch, 词向量维度128)
# output的表示形式与输入x类似，为(seq_len, batch, 词向量维度256)
# 前后的hidden形状要一样, 所以DataLoader加载器的batch数要能被整数
output, hidden = self.rnn(embed.transpose(0, 1), hidden)
# 全连接层输入二维数据， 词数量*词维度
# 输入维度: (seq_len*batch, 词向量维度256)
# 输出维度: (seq_len*batch, 语料中词的数量)
# output: 每个词的分值分布，后续结合softmax输出概率分布
output = self.out(output.reshape(shape=(-1, output.shape[-1])))
# 网络输出结果
return output, hidden
def init_hidden(self, bs):
# 隐藏层的初始化:[网络层数, batch, 隐藏层向量维度]
return torch.zeros(1, bs, 256)
if __name__ == "__main__":
# 获取数据
unique_words, word_to_index, unique_word_count, corpus_idx = build_vocab()
model = TextGenerator(unique_word_count)
for named, parameter in model.named_parameters():
print(named)
print(parameter)构建训练函数
前面的准备工作完成之后, 我们就可以编写训练函数。训练函数主要负责编写数据迭代、送入网络、计算损失、反向传播、更新参数，其流程基本较为固定。
由于我们要实现文本生成，文本生成本质上，输入一串文本，预测下一个文本，也属于分类问题，所以，我们使用多分类交叉熵损失函数。优化方法我们学习过 SGB、AdaGrad、Adam 等，在这里我们选择学习率、梯度自适应的 Adam 算法作为我们的优化方法。
训练完成之后，我们使用 torch.save 方法将模型持久化存储。
def train():
# 构建词典
unique_words, word_to_index, unique_word_count, corpus_idx = build_vocab()
# 数据集 LyricsDataset对象，并实现了 __getitem__ 方法
lyrics = LyricsDataset(corpus_idx=corpus_idx, num_chars=32)
# 查看句子数量
# print(lyrics.number)
# 初始化模型
model = TextGenerator(unique_word_count)
# 数据加载器 DataLoader对象，并将lyrics dataset对象传递给它
lyrics_dataloader = DataLoader(lyrics, shuffle=True, batch_size=5)
# 损失函数
criterion = nn.CrossEntropyLoss()
# 优化方法
optimizer = optim.Adam(model.parameters(), lr=1e-3)
# 训练轮数
epoch = 10
for epoch_idx in range(epoch):
# 训练时间
start = time.time()
iter_num = 0 # 迭代次数
# 训练损失
total_loss = 0.0
# 遍历数据集 DataLoader 会在后台调用 dataset.__getitem__(index) 来获取每个样本的数据和标签，并将它们组合成一个 batch
for x, y in lyrics_dataloader:
# 隐藏状态的初始化
hidden = model.init_hidden(bs=5)
# 模型计算
output, hidden = model(x, hidden)
# 计算损失
# y形状为(batch, seq_len), 需要转换成一维向量->160个词的下标索引
# output形状为(seq_len, batch, 词向量维度)
# 需要先将y进行维度交换(和output保持一致)再改变形状
y = torch.transpose(y, 0, 1).reshape(shape=(-1,))
loss = criterion(output, y)
optimizer.zero_grad()
loss.backward()
optimizer.step()
iter_num += 1 # 迭代次数加1
total_loss += loss.item()
# 打印训练信息
print('epoch %3s loss: %.5f time %.2f' % (epoch_idx + 1, total_loss / iter_num, time.time() - start))
# 模型存储
torch.save(model.state_dict(), 'model/lyrics_model_%d.pth' % epoch)
if __name__ == "__main__":
train()输出结果：
epoch 1 loss: 1.84424 time 5.75
epoch 2 loss: 0.21154 time 5.91
epoch 3 loss: 0.12014 time 5.85
epoch 4 loss: 0.10625 time 5.73
epoch 5 loss: 0.10226 time 5.58
epoch 6 loss: 0.10009 time 5.65
epoch 7 loss: 0.09942 time 5.66
epoch 8 loss: 0.09783 time 5.66
epoch 9 loss: 0.09663 time 5.75
epoch 10 loss: 0.09568 time 5.77构建预测函数
从磁盘加载训练好的模型，进行预测。预测函数，输入第一个指定的词，我们将该词输入网路，预测出下一个词，再将预测的出的词再次送入网络，预测出下一个词，以此类推，知道预测出我们指定长度的内容。
def predict(start_word, sentence_length):
# 构建词典
unique_words, word_to_index, unique_word_count, _ = build_vocab()
# 构建模型
model = TextGenerator(unique_word_count)
# 加载参数
model.load_state_dict(torch.load('model/lyrics_model_10.pth'))
# 隐藏状态
hidden = model.init_hidden(bs=1)
# 将起始词转换为索引
word_idx = word_to_index[start_word]
# 产生的词的索引存放位置
generate_sentence = [word_idx]
# 遍历到句子长度，获取每一个词
for _ in range(sentence_length):
# 模型预测
output, hidden = model(torch.tensor([[word_idx]]), hidden)
# 获取预测结果
word_idx = torch.argmax(output)
generate_sentence.append(word_idx)
# 根据产生的索引获取对应的词，并进行打印
for idx in generate_sentence:
print(unique_words[idx], end='')
if __name__ == '__main__':
# 调用预测函数
predict('分手', 50)输出结果：
分手的话像语言暴力
我已无能为力再提起 决定中断熟悉
周杰伦 周杰伦
一步两步三步四步望著天 看星星
一颗两颗三颗四颗 连成线背著背默默许下心愿
看远方的星