使用 LSTM 的变形 GRU 中文字符级建模，训练出一个能作诗的模型
数据集集为4000+古诗，一行一首，标题、作者已去掉：
https://download.csdn.net/download/fgg1234567890/16153813

import os
import random
import pandas as pd
import numpy as np
# 引入需要的模块
from keras.layers import Bidirectional
from keras.layers import GRU
import keras
from keras.callbacks import LambdaCallback
from sklearn.model_selection import train_test_split
from keras.layers import Dense, Input, Flatten, Dropout
from keras.layers import Embedding
from keras.models import Model
from keras.utils import plot_model
from keras.models import load_model
from keras.optimizers import Adam

# 步骤：
# 语料准备
# 语料预处理
# 模型参数配置
# 构建模型
# 训练模型
# 模型作诗
# 绘制模型网络结构图

puncs = [']', '[', '（', '）', '{', '}', '：', '《', '》']

# 预处理
def preprocess_file(Config):
    # 语料文本内容
    files_content = ''
    with open(Config.poetry_file, 'r', encoding='utf-8') as f:
        for line in f:
            # 每行的末尾加上"]"符号代表一首诗结束
            for char in puncs:
                line = line.replace(char, "")
            files_content += line.strip() + "]"

    words = sorted(list(files_content))
    words.remove(']')
    counted_words = {}
    for word in words:
        if word in counted_words:
            counted_words[word] += 1
        else:
            counted_words[word] = 1

    # 去掉低频的字
    erase = []
    for key in counted_words:
        if counted_words[key] <= 2:
            erase.append(key)
    for key in erase:
        del counted_words[key]
    del counted_words[']']
    wordPairs = sorted(counted_words.items(), key=lambda x: -x[1])

    words, _ = zip(*wordPairs)
    # word到id的映射
    word2num = dict((c, i+1) for i, c in enumerate(words))
    num2word = dict((i, c) for i, c in enumerate(words))
    word2numF = lambda x: word2num.get(x, 0)
    return word2numF, num2word, words, files_content

class Config(object):
    '''
    模型参数配置。预先定义模型参数和加载语料以及模型保存名称
    '''
    poetry_file = './qtais_tab2.txt'
    weight_file = 'poetry_model.h5'
    # 根据前六个字预测第七个字
    max_len = 6
    batch_size = 512
    learning_rate = 0.001

class PoetryModel(object):
    def __init__(self, config):
        '''
        init 函数定义，通过加载 Config 配置信息，进行语料预处理和模型加载，
        如果模型文件存在则直接加载模型，否则开始训练。
        '''
        self.model = None
        self.do_train = True
        self.loaded_model = False
        self.config = config

        # 文件预处理
        self.word2numF, self.num2word, self.words, self.files_content = preprocess_file(self.config)
        if os.path.exists(self.config.weight_file):
            self.model = load_model(self.config.weight_file)
            self.model.summary()
        else:
            self.train()
        self.do_train = False
        self.loaded_model = True

    def build_model(self):
        '''
        建立模型
        用 Keras 来构建网络模型，这里使用 LSTM 的 GRU 来实现，当然直接使用 LSTM 也没问题。
        '''
        input_tensor = Input(shape=(self.config.max_len,))
        embedd = Embedding(len(self.num2word)+1, 300, input_length=self.config.max_len)(input_tensor)
        lstm = Bidirectional(GRU(128, return_sequences=True))(embedd)
        dropout = Dropout(0.6)(lstm)
        # lstm = Bidirectional(GRU(128, return_sequences=True))(embedd)
        # dropout = Dropout(0.6)(lstm)
        flatten = Flatten()(dropout)
        dense = Dense(len(self.words), activation='softmax')(flatten)
        self.model = Model(inputs=input_tensor, outputs=dense)
        optimizer = Adam(lr=self.config.learning_rate)
        self.model.compile(loss='categorical_crossentropy', optimizer=optimizer, metrics=['accuracy'])

    def sample(self, preds, temperature=1.0):
        '''
        当temperature=1.0时，模型输出正常
        当temperature=0.5时，模型输出比较open
        当temperature=1.5时，模型输出比较保守
        在训练的过程中可以看到temperature不同，结果也不同
        '''
        preds = np.asarray(preds).astype('float64')
        preds = np.log(preds) / temperature
        exp_preds = np.exp(preds)
        preds = exp_preds / np.sum(exp_preds)
        probas = np.random.multinomial(1, preds, 1)
        return np.argmax(probas)

    def generate_sample_result(self, epoch, logs):
        '''
        训练过程中，每个 epoch 打印出当前的学习情况
        '''
        print("\n==================Epoch {}=====================".format(epoch))
        for diversity in [0.5, 1.0, 1.5]:
            print("------------Diversity {}--------------".format(diversity))
            start_index = random.randint(0, len(self.files_content) - self.config.max_len - 1)
            generated = ''
            sentence = self.files_content[start_index: start_index + self.config.max_len]
            generated += sentence
            for i in range(20):
                x_pred = np.zeros((1, self.config.max_len))
                for t, char in enumerate(sentence[-6:]):
                    x_pred[0, t] = self.word2numF(char)

                preds = self.model.predict(x_pred, verbose=0)[0]
                next_index = self.sample(preds, diversity)
                next_char = self.num2word[next_index]
                generated += next_char
                sentence = sentence + next_char
            print(sentence)

    def predict(self, text):
        '''
        根据给定的提示，来进行预测。
        根据给出的文字，生成诗句，如果给的 text 不到四个字，则随机补全。
        '''
        if not self.loaded_model:
            return
        with open(self.config.poetry_file, 'r', encoding='utf-8') as f:
            file_list = f.readlines()
        random_line = random.choice(file_list)
        # 如果给的text不到四个字，则随机补全
        if not text or len(text) != 4:
            for _ in range(4 - len(text)):
                random_str_index = random.randrange(0, len(self.words))
                text += self.num2word.get(random_str_index) if self.num2word.get(random_str_index) not in [',', '。',
                                                                                                           '，'] else self.num2word.get(
                    random_str_index + 1)
        seed = random_line[-(self.config.max_len):-1]
        res = ''
        seed = 'c' + seed
        for c in text:
            seed = seed[1:] + c
            for j in range(5):
                x_pred = np.zeros((1, self.config.max_len))
                for t, char in enumerate(seed):
                    x_pred[0, t] = self.word2numF(char)
                preds = self.model.predict(x_pred, verbose=0)[0]
                next_index = self.sample(preds, 1.0)
                next_char = self.num2word[next_index]
                seed = seed[1:] + next_char
            res += seed
        return res

    def data_generator(self):
        '''
        用于生成数据，提供给模型训练时使用。
        给定前六个字，生成第七个字，所以在后面生成训练数据的时候，会以6的跨度，1的步长截取文字，生成语料。
        如果出现了 ] 符号，说明 ] 符号之前的语句和之后的语句是两首诗里面的内容，
        两首诗之间是没有关联关系的，所以我们后面会舍弃掉包含 ] 符号的训练数据。
        '''
        print("len(self.files_content):", len(self.files_content))
        i = 0
        while 1:
            x = self.files_content[i: i + self.config.max_len]
            y = self.files_content[i + self.config.max_len]
            puncs = [']', '[', '（', '）', '{', '}', '：', '《', '》', ':']
            if len([i for i in puncs if i in x]) != 0:
                i += 1
                continue
            if len([i for i in puncs if i in y]) != 0:
                i += 1
                continue
            y_vec = np.zeros(
                shape=(1, len(self.words)),
                dtype=np.bool
            )
            y_vec[0, self.word2numF(y)] = 1.0
            x_vec = np.zeros(
                shape=(1, self.config.max_len),
                dtype=np.int32
            )
            for t, char in enumerate(x):
                x_vec[0, t] = self.word2numF(char)
            yield x_vec, y_vec
            i += 1

    def train(self):
        '''
        用来进行模型训练。
        TensorFlow回调函数：tf.keras.callbacks.LambdaCallback:
        用于动态创建简单的自定义回调的回调。
        此回调是使用将在适当时间调用的匿名函数构造的。请注意，回调需要位置参数，如下所示：
        on_epoch_begin和on_epoch_end要求两个位置参数： epoch，logs
        on_batch_begin和on_batch_end要求两个位置参数： batch，logs
        on_train_begin并on_train_end要求一个位置参数： logs
        参数：
        on_epoch_begin：在每个epoch开始时调用。
        on_epoch_end：在每个epoch结束时调用。
        on_batch_begin：在每个批处理开始时调用。
        on_batch_end：在每个批处理结束时调用。
        on_train_begin：在模型训练开始时调用。
        on_train_end：在模型训练结束时调用。
        '''
        # number_of_epoch = len(self.files_content) // self.config.batch_size
        number_of_epoch = 10
        if not self.model:
            self.build_model()
        self.model.summary()
        self.model.fit_generator(
            generator=self.data_generator(),
            verbose=True,
            steps_per_epoch=self.config.batch_size,
            epochs=number_of_epoch,
            callbacks=[
                # 该回调函数将在每个epoch后保存模型到filepath
                keras.callbacks.ModelCheckpoint(self.config.weight_file, save_weights_only=False),
                LambdaCallback(on_epoch_end=self.generate_sample_result)])

model = PoetryModel(Config)

text = input("text:")
sentence = model.predict(text)
print(sentence)
# 绘制网络结构图
plot_model(model.model, to_file='model.png')

原文：
https://soyoger.blog.csdn.net/article/details/108729402