Press "Enter" to skip to content

基于LSTM的诗词生成

本站内容均来自兴趣收集,如不慎侵害的您的相关权益,请留言告知,我们将尽快删除.谢谢.

 

文章目录

 

本文的主要内容是基于LSTM的诗词生成,文中包括数据集的介绍、实验代码以及运行结果等,该实验采用的是长短期记忆 (LSTM) 深度学习模型,训练10个epoch,然后在测试的过程中,每个epoch出一次诗词的生成结果,随着周期的进行,诗词的生成情况也越来越好,本文中诗词的生成包括随机诗词的生成和藏头诗的生成。

 

一、数据集介绍

 

本实验采用的数据集保存在文件 poems.txt 中,该文件中每行一首诗,题目和诗之间用“:”隔开,每首诗的长度也是各不相同的。

 

下图是在文件末的截图,看行数可以知道该文件包含了43030首诗,数据集还是挺大的。

 

在本实验中,该数据集中的所有诗不会都被使用到,在代码中会对内容太长的诗进行剔除。

 

二、实验代码

 

1、随机诗词生成

 

生成随机诗词的代码如下。

 

# 声明:本代码并非自己编写,代码来源在文末已给出链接
import math
import re
import numpy as np
import tensorflow as tf
from collections import Counter
DATA_PATH = 'poems.txt'  # 数据集路径
MAX_LEN = 64   # 设定单行诗的最大长度
DISALLOWED_WORDS = ['(', ')', '(', ')', '__', '《', '》', '【', '】', '[', ']']  # 禁用的字符,拥有以下符号的诗将被忽略
BATCH_SIZE = 128
poetry = []  # 一首诗对应一个列表的元素
with open(DATA_PATH, 'r', encoding='utf-8') as f:   # 按行读取文件数据,一行就是一首诗
    lines = f.readlines()
for line in lines:
    fields = re.split(r"[::]", line)  # 利用正则表达式拆分标题和内容
    if len(fields) != 2:  # 每行拆分后如果不是两项,就跳过该异常数据
        continue
    content = fields[1]   # 提取诗词内容
    if len(content) > MAX_LEN - 2:  # 剔除内容过长的诗词
        continue
    if any(word in content for word in DISALLOWED_WORDS):  # 剔除存在禁用符的诗词
        continue
    poetry.append(content.replace('
', ''))  # 将诗词添加到列表里,每行一首
MIN_WORD_FREQUENCY = 8  # 最小词频
counter = Counter()  # 统计词频
for line in poetry:
    counter.update(line)
tokens = [token for token, count in counter.items() if count >= MIN_WORD_FREQUENCY]  # 过滤掉低词频的词
tokens = ["[PAD]", "[NONE]", "[START]", "[END]"] + tokens  # 补上特殊词标记:填充字符标记、未知词标记、开始标记、结束标记
word_idx = {
 }  # 从词到编号的映射
idx_word = {
 }  # 从编号到词的映射
for idx, word in enumerate(tokens):
    word_idx[word] = idx
    idx_word[idx] = word
class Tokenizer:
    """分词器"""
    def __init__(self, tokens):
        self.dict_size = len(tokens)  # 词汇表大小
        # 生成映射关系
        self.token_id = {
 }   # 从词到编号的映射
        self.id_token = {
 }   # 从编号到词的映射
        for idx, word in enumerate(tokens):
            self.token_id[word] = idx
            self.id_token[idx] = word
        # 各个特殊标记的编号id,方便其他地方使用
        self.start_id = self.token_id["[START]"]
        self.end_id = self.token_id["[END]"]
        self.none_id = self.token_id["[NONE]"]
        self.pad_id = self.token_id["[PAD]"]
    def id_to_token(self, token_id):
        """编号到词的映射"""
        return self.id_token.get(token_id)
    def token_to_id(self, token):
        """词到编号的映射"""
        return self.token_id.get(token, self.none_id)  # 编号里没有返回 [NONE]
    def encode(self, tokens):
        """词列表:[START]编号 + 编号列表 + [END]编号"""
        token_ids = [self.start_id, ]  # 起始标记
        for token in tokens:
            token_ids.append(self.token_to_id(token)) # 词转编号
        token_ids.append(self.end_id)  # 结束标记
        return token_ids
    def decode(self, token_ids):
        """编号列表转词列表,去掉起始、结束标记"""
        flag_tokens = {
 "[START]", "[END]"}
        tokens = []
        for idx in token_ids:
            token = self.id_to_token(idx)
            if token not in flag_tokens:
                tokens.append(token) # 跳过起始、结束标记
        return tokens
tokenizer = Tokenizer(tokens)
# 构建数据集
class PoetryDataSet:
    """古诗数据集生成器"""
    def __init__(self, data, tokenizer, batch_size):
        self.data = data
        self.total_size = len(self.data)
        self.tokenizer = tokenizer  # 分词器,用于词转编号
        self.batch_size = batch_size  # 每批数据量
        self.steps = int(math.floor(len(self.data) / self.batch_size))  # 每个epoch迭代的步数
    def pad_line(self, line, length, padding=None):
        """对齐单行数据"""
        if padding is None:
            padding = self.tokenizer.pad_id
        padding_length = length - len(line)
        if padding_length > 0:
            return line + [padding] * padding_length
        else:
            return line[:length]
    def __len__(self):
        return self.steps
    def __iter__(self):
        np.random.shuffle(self.data)  # 把数据打乱
        # 迭代一个epoch,每次一个batch
        for start in range(0, self.total_size, self.batch_size):
            end = min(start + self.batch_size, self.total_size)
            data = self.data[start:end]
            max_length = max(map(len, data))  # map根据提供的函数对指定序列做映射
            batch_data = []
            for str_line in data:
                encode_line = self.tokenizer.encode(str_line)  # 对每一行诗词进行编码、并补齐padding
                pad_encode_line = self.pad_line(encode_line, max_length + 2)  # 加2是因为tokenizer.encode会添加START和END
                batch_data.append(pad_encode_line)
            batch_data = np.array(batch_data)
            yield batch_data[:, :-1], batch_data[:, 1:]   # yield 特征、标签
    def generator(self):
        while True:
            yield from self.__iter__()
dataset = PoetryDataSet(poetry, tokenizer, BATCH_SIZE)  # 初始化 PoetryDataSet
# 构建模型
model = tf.keras.Sequential([
    # 词嵌入层
    tf.keras.layers.Embedding(input_dim=tokenizer.dict_size, output_dim=150),
    tf.keras.layers.LSTM(150, dropout=0.5, return_sequences=True),  # 第一个LSTM层
    tf.keras.layers.LSTM(150, dropout=0.5, return_sequences=True),  # 第二个LSTM层
    # 利用TimeDistributed对每个时间步的输出都做Dense操作,即softmax激活
    tf.keras.layers.TimeDistributed(tf.keras.layers.Dense(tokenizer.dict_size, activation='softmax')),
    ])
model.summary()
model.compile(
    optimizer=tf.keras.optimizers.Adam(),  # 优化器使用Adam
    loss=tf.keras.losses.sparse_categorical_crossentropy  # 稀疏分类交叉熵
    )
model.fit_generator(dataset.generator(), steps_per_epoch=dataset.steps, epochs=10)
# 预测
token_ids = [tokenizer.token_to_id(word) for word in ["月", "光", "静", "谧"]]   # 将词转为编号
result = model.predict([token_ids, ])  # 进行预测
print(result)
print(result.shape)
def predict(model, token_ids):
    """在概率值为前100的词中选取一个词(按概率分布的方式),返回一个词的编号,但不包含[PAD][NONE][START]"""
    # 预测各个词的概率分布
    # 0  表示对输入的第0个样本做预测
    # -1 表示只要对最新的词的预测
    # 3: 表示不要前面几个标记符
    _probas = model.predict([token_ids, ])[0, -1, 3:]
    # 按概率降序,取前100
    p_args = _probas.argsort()[-100:][::-1]  # 此时拿到的是索引
    p = _probas[p_args]  # 根据索引找到具体的概率值
    p = p / sum(p)  # 归一化操作
    target_index = np.random.choice(len(p), p=p)  # 按概率抽取一个
    # 前面预测时删除了前几个标记符,因此编号要补上3位才是实际在tokenizer词典中的编号
    return p_args[target_index] + 3
token_ids = tokenizer.encode("清风明月")[:-1]
while len(token_ids) < 13:
    target = predict(model, token_ids)   # 预测词的编号
    token_ids.append(target)  # 保存结果
    if target == tokenizer.end_id:
        break
print("".join(tokenizer.decode(token_ids)))
def generate_random_poem(tokenizer, model, text=""):
    """随机生成一首诗
    tokenizer: 分词器
    model: 古诗模型
    text: 古诗的起始字符串,默认为空
    返回值是一首古诗的字符串
    """
    token_ids = tokenizer.encode(text)[:-1]  # 将初始字符串转成token_ids,并去掉结束标记[END]
    while len(token_ids) < MAX_LEN:
        target = predict(model, token_ids)  # 预测词的编号
        token_ids.append(target)   # 追加结果
        if target == tokenizer.end_id:
            break
    return "".join(tokenizer.decode(token_ids))
# 保存模型及加载
class ShowSaveCallback(tf.keras.callbacks.Callback):
    def __init__(self):
        super().__init__()
        self.loss = float("inf")
    def on_epoch_end(self, epoch, logs=None):
        if logs['loss'] <= self.loss:  # 保留损失最低的模型
            self.loss = logs['loss']
            model.save("./rnn_model.h5")
        print()  # 打印本次训练的效果
        for i in range(5):
            print(generate_random_poem(tokenizer, model))  # 生成五首随机诗词
# 开始训练
model.fit(
    dataset.generator(),
    steps_per_epoch=dataset.steps,
    epochs=10,
    callbacks=[ShowSaveCallback()]
    )
model = tf.keras.models.load_model("./rnn_model.h5")  # 保存训练数据

 

2、藏头诗生成

 

生成藏头诗的代码如下。

 

# 声明:本代码并非自己编写,代码来源在文末已给出链接
import math
import re
import numpy as np
import tensorflow as tf
from collections import Counter
DATA_PATH = 'poems.txt'  # 数据集路径
MAX_LEN = 64   # 设定单行诗的最大长度
DISALLOWED_WORDS = ['(', ')', '(', ')', '__', '《', '》', '【', '】', '[', ']']  # 禁用的字符,拥有以下符号的诗将被忽略
BATCH_SIZE = 128
poetry = []  # 一首诗对应一个列表的元素
with open(DATA_PATH, 'r', encoding='utf-8') as f:   # 按行读取文件数据,一行就是一首诗
    lines = f.readlines()
for line in lines:
    fields = re.split(r"[::]", line)  # 利用正则表达式拆分标题和内容
    if len(fields) != 2:  # 每行拆分后如果不是两项,就跳过该异常数据
        continue
    content = fields[1]   # 提取诗词内容
    if len(content) > MAX_LEN - 2:  # 剔除内容过长的诗词
        continue
    if any(word in content for word in DISALLOWED_WORDS):  # 剔除存在禁用符的诗词
        continue
    poetry.append(content.replace('
', ''))  # 将诗词添加到列表里,每行一首
MIN_WORD_FREQUENCY = 8  # 最小词频
counter = Counter()  # 统计词频
for line in poetry:
    counter.update(line)
tokens = [token for token, count in counter.items() if count >= MIN_WORD_FREQUENCY]  # 过滤掉低词频的词
tokens = ["[PAD]", "[NONE]", "[START]", "[END]"] + tokens  # 补上特殊词标记:填充字符标记、未知词标记、开始标记、结束标记
word_idx = {
 }  # 从词到编号的映射
idx_word = {
 }  # 从编号到词的映射
for idx, word in enumerate(tokens):
    word_idx[word] = idx
    idx_word[idx] = word
class Tokenizer:
    """分词器"""
    def __init__(self, tokens):
        self.dict_size = len(tokens)  # 词汇表大小
        # 生成映射关系
        self.token_id = {
 }   # 从词到编号的映射
        self.id_token = {
 }   # 从编号到词的映射
        for idx, word in enumerate(tokens):
            self.token_id[word] = idx
            self.id_token[idx] = word
        # 各个特殊标记的编号id,方便其他地方使用
        self.start_id = self.token_id["[START]"]
        self.end_id = self.token_id["[END]"]
        self.none_id = self.token_id["[NONE]"]
        self.pad_id = self.token_id["[PAD]"]
    def id_to_token(self, token_id):
        """编号到词的映射"""
        return self.id_token.get(token_id)
    def token_to_id(self, token):
        """词到编号的映射"""
        return self.token_id.get(token, self.none_id)  # 编号里没有返回 [NONE]
    def encode(self, tokens):
        """词列表:[START]编号 + 编号列表 + [END]编号"""
        token_ids = [self.start_id, ]  # 起始标记
        for token in tokens:
            token_ids.append(self.token_to_id(token)) # 词转编号
        token_ids.append(self.end_id)  # 结束标记
        return token_ids
    def decode(self, token_ids):
        """编号列表转词列表,去掉起始、结束标记"""
        flag_tokens = {
 "[START]", "[END]"}
        tokens = []
        for idx in token_ids:
            token = self.id_to_token(idx)
            if token not in flag_tokens:
                tokens.append(token) # 跳过起始、结束标记
        return tokens
tokenizer = Tokenizer(tokens)
# 构建数据集
class PoetryDataSet:
    """古诗数据集生成器"""
    def __init__(self, data, tokenizer, batch_size):
        self.data = data
        self.total_size = len(self.data)
        self.tokenizer = tokenizer  # 分词器,用于词转编号
        self.batch_size = batch_size  # 每批数据量
        self.steps = int(math.floor(len(self.data) / self.batch_size))  # 每个epoch迭代的步数
    def pad_line(self, line, length, padding=None):
        """对齐单行数据"""
        if padding is None:
            padding = self.tokenizer.pad_id
        padding_length = length - len(line)
        if padding_length > 0:
            return line + [padding] * padding_length
        else:
            return line[:length]
    def __len__(self):
        return self.steps
    def __iter__(self):
        np.random.shuffle(self.data)  # 把数据打乱
        # 迭代一个epoch,每次一个batch
        for start in range(0, self.total_size, self.batch_size):
            end = min(start + self.batch_size, self.total_size)
            data = self.data[start:end]
            max_length = max(map(len, data))  # map根据提供的函数对指定序列做映射
            batch_data = []
            for str_line in data:
                encode_line = self.tokenizer.encode(str_line)  # 对每一行诗词进行编码、并补齐padding
                pad_encode_line = self.pad_line(encode_line, max_length + 2)  # 加2是因为tokenizer.encode会添加START和END
                batch_data.append(pad_encode_line)
            batch_data = np.array(batch_data)
            yield batch_data[:, :-1], batch_data[:, 1:]   # yield 特征、标签
    def generator(self):
        while True:
            yield from self.__iter__()
dataset = PoetryDataSet(poetry, tokenizer, BATCH_SIZE)  # 初始化 PoetryDataSet
# 构建模型
model = tf.keras.Sequential([
    # 词嵌入层
    tf.keras.layers.Embedding(input_dim=tokenizer.dict_size, output_dim=150),
    tf.keras.layers.LSTM(150, dropout=0.5, return_sequences=True),  # 第一个LSTM层
    tf.keras.layers.LSTM(150, dropout=0.5, return_sequences=True),  # 第二个LSTM层
    # 利用TimeDistributed对每个时间步的输出都做Dense操作,即softmax激活
    tf.keras.layers.TimeDistributed(tf.keras.layers.Dense(tokenizer.dict_size, activation='softmax')),
    ])
model.summary()
model.compile(
    optimizer=tf.keras.optimizers.Adam(),  # 优化器使用Adam
    loss=tf.keras.losses.sparse_categorical_crossentropy  # 稀疏分类交叉熵
    )
model.fit_generator(dataset.generator(), steps_per_epoch=dataset.steps, epochs=10)
# 预测
token_ids = [tokenizer.token_to_id(word) for word in ["月", "光", "静", "谧"]]   # 将词转为编号
result = model.predict([token_ids, ])  # 进行预测
print(result)
print(result.shape)
def predict(model, token_ids):
    """在概率值为前100的词中选取一个词(按概率分布的方式),返回一个词的编号,但不包含[PAD][NONE][START]"""
    # 预测各个词的概率分布
    # 0  表示对输入的第0个样本做预测
    # -1 表示只要对最新的词的预测
    # 3: 表示不要前面几个标记符
    _probas = model.predict([token_ids, ])[0, -1, 3:]
    # 按概率降序,取前100
    p_args = _probas.argsort()[-100:][::-1]  # 此时拿到的是索引
    p = _probas[p_args]  # 根据索引找到具体的概率值
    p = p / sum(p)  # 归一化操作
    target_index = np.random.choice(len(p), p=p)  # 按概率抽取一个
    # 前面预测时删除了前几个标记符,因此编号要补上3位才是实际在tokenizer词典中的编号
    return p_args[target_index] + 3
token_ids = tokenizer.encode("清风明月")[:-1]
while len(token_ids) < 13:
    target = predict(model, token_ids)   # 预测词的编号
    token_ids.append(target)  # 保存结果
    if target == tokenizer.end_id:
        break
print("".join(tokenizer.decode(token_ids)))
def generate_acrostic_poem(tokenizer, model, heads):
    """
    生成一首藏头诗
    tokenizer: 分词器
    model: 古诗模型
    heads: 藏头诗的头
    返回值是一首古诗的字符串
    """
    token_ids = [tokenizer.start_id, ]  # token_ids,只包含[START]编号
    punctuation_ids = {
 tokenizer.token_to_id(","), tokenizer.token_to_id("。")}  # 逗号和句号标记编号
    content = []
    for head in heads:
        content.append(head)
        token_ids.append(tokenizer.token_to_id(head))  # head转为编号id,放入列表,用于预测
        target = -1
        while target not in punctuation_ids:  # 遇到逗号、句号,说明本句结束,开始下一句
            target = predict(model, token_ids)  # 预测词的编号
            # 因为可能预测到END,所以加个判断
            if target > 3:
                token_ids.append(target)  # 保存结果到token_ids中,下一次预测还要用
                content.append(tokenizer.id_to_token(target))
    return "".join(content)
# 保存模型及加载
class ShowSaveCallback(tf.keras.callbacks.Callback):
    def __init__(self):
        super().__init__()
        self.loss = float("inf")
    def on_epoch_end(self, epoch, logs=None):
        if logs['loss'] <= self.loss:  # 保留损失最低的模型
            self.loss = logs['loss']
            model.save("./rnn_model.h5")
        print()  # 打印本次训练的效果
        print(generate_acrostic_poem(tokenizer, model, '机器学习'))  # 以“机器学习”作藏头诗
        print(generate_acrostic_poem(tokenizer, model, '神经网络'))  # 以“神经网络”作藏头诗
# 开始训练
model.fit(
    dataset.generator(),
    steps_per_epoch=dataset.steps,
    epochs=10,
    callbacks=[ShowSaveCallback()]
    )
model = tf.keras.models.load_model("./rnn_model.h5")  # 保存训练数据

 

三、实验结果

 

运行代码后的结果截图依次如下。

 

参数训练的情况如下图所示。

 

1.随机诗词生成结果

 

第一个epoch生成的随机诗词如下图所示。

 

第二个epoch生成的随机诗词如下图所示。

 

第三个epoch生成的随机诗词如下图所示。

 

第四个epoch生成的随机诗词如下图所示。

 

第五个epoch生成的随机诗词如下图所示。

 

第六个epoch生成的随机诗词如下图所示。

第七个epoch生成的随机诗词如下图所示。

第八个epoch生成的随机诗词如下图所示。

 

第九个epoch生成的随机诗词如下图所示。

 

第十个epoch生成的随机诗词如下图所示。

 

由以上随机诗词的生成结果可以看到,生成的随机诗词有五言诗和七言诗,诗句有四句的、六句的,也有八句的,这些是和诗词特征比较吻合的。不好的方面是其中有几首诗的生成是不太好的,而且诗的平仄对应和意境的描绘都不太成熟。

 

2.藏头诗生成结果

 

分别以“机器学习”和“神经网络”开头作诗。

 

第一个epoch生成的藏头诗如下图所示。

 

第二个epoch生成的藏头诗如下图所示。

 

第三个epoch生成的藏头诗如下图所示。

 

第四个epoch生成的藏头诗如下图所示。

 

第五个epoch生成的藏头诗如下图所示。

 

第六个epoch生成的藏头诗如下图所示。

 

第七个epoch生成的藏头诗如下图所示。

 

第八个epoch生成的藏头诗如下图所示。

 

第九个epoch生成的藏头诗如下图所示。

 

第十个epoch生成的藏头诗如下图所示。

 

由以上藏头诗的生成结果可以看到,生成的藏头诗有五言诗和七言诗,诗句只有四句的,因为藏头的字只有四个,这些和藏头诗的特征是吻合的。

 

以上就是基于LSTM诗词生成的所有内容了,本实验的环境配置与上一个实验基于IMDB评论数据集的情感分析
相同,具体的情况可以参考该博文,本实验中,在诗词特征方面,生成的随机诗词和藏头诗都还不错,但是在平仄对应关系以及诗词意境的营造方面,两者都是不太成熟的。

 

参考博文:

 

RNN生成古诗词

使用TensorFlow 2.0 + RNN 实现一个古体诗生成器

Be First to Comment

发表回复

您的电子邮箱地址不会被公开。 必填项已用*标注