---

前言

本文的主要内容是基于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('\n', ''))  # 将诗词添加到列表里，每行一首

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")  # 保存训练数据
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221

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('\n', ''))  # 将诗词添加到列表里，每行一首

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
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231

---

三、实验结果

运行代码后的结果截图依次如下。
参数训练的情况如下图所示。

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 实现一个古体诗生成器