NLP人工智能写诗模型之李小白

深碍√TFBOYSˉ_ 2022-11-30 15:43 51阅读 0赞

#  #

# 1.前言 #

我希望我的技术是有温度的，也许是生活已经足够冰冷。我希望赋予我的模型以生命，它们在那个虚拟世界里面的生活，又何尝不跟我们这个所谓真实的世界一样呢？所以我叫他李小白，因为我喜欢那句‘乘风破浪会有时，直挂云帆济沧海’。

# 2.目标 #

我们希望得到这样一个模型，当我输入第一个字的时候，她可以用这个字开头为我做一首诗。

# 3.思路 #

我们会让模型读上万首诗，他会从中学习到一种能力，写诗。无关意境、无关思想、只是为了好玩。我们使用LSTM循环神经网络作为隐层，基本网络结构为输入层——>LSTM——>LSTM——>输出层；

# 4.数据预处理 #

1.我们的训练数据总共有43030行，每一行代表一首诗词。格式为‘标题:内容’如下：  
静夜思:床前明月光，疑是地上霜。举头望明月，低头思故乡。  
2.预处理  
1)将每行诗的内容(不包含标题),加入到poetrys向量中。过滤掉少于5个字的和大于79个字的诗，去掉特殊字符，每行内容添加'\['、'\]'作为开始和结束标签。过滤完后剩下34646首诗  
2)按照每行诗内容的长度，对poetrys向量进行排序  
3)生成词表，按照每个字出现的次数对其进行编码，出现次数越多则编码越小  
4)根据词表，将诗集poetrys转换为数字形式  
5)准备训练数据,加载所有批次的数据，对于每个批次来说，x为批次数量的诗句，诗句的长度为该批次中最长的诗句长度。y为x去掉第一位重复最后一位得到的诗句。

6）使用嵌入层对每个词向量进行one-hot编码，并且编码后的向量使用隐层嵌入矩阵映射为一个稠密向量。

# 5.RNN网络 #

1.我们使用两层的LSTM网络, 隐层神经元128个。

2.优化器使用adam

3.损失函数还是选择传统的softmax,对于一个批次中的所有数据loss进行累积，然后求平均

# 6.训练 #

1.每个批次有256条数据，训练50个轮次，每隔7个轮次保存一次生成的模型。  
学习率衰减为:0.002 \* (0.97 \*\* epoch)  
2.为了防止梯度爆炸和梯度消失，对于计算出来的梯度进行了截取，使用tf.clip\_by\_global\_norm，截取率设置为5

# 7.预测 #

高阵明日自有碛，

来有纤城还惟。

谁玄骚市看情烟，

欲雨清婺纵砚山。

这个模型在GTX1050显卡上训练了20分钟，上面的例子我给出了首个字高，李小白同学就为我输出了这首诗；我自己的水平有限，不敢妄加评论。也许我们的模型没有经历过挫折与坎坷，写不出那样‘念天地之悠悠，独怆然而涕下’的诗句。但是 从本质上讲，人类通过学习知识，能出口成章，这跟模型是没有什么区别的。也许当我们真正找到自己跟模型不一样的地方的那天，我们也就找到了真正的自己吧！各位看官如果心情不好的话，看一下下面这个伟大的视频，杨超越真的超级可爱！

【火箭少女解散】杨超越告别感言太好笑了！！哈哈哈哈哈哈哈哈哈哈哈哈哈哈哈哈明明是很伤心的事，看得我又哭又笑的杨超越：“我真是干啥啥不行，跟老板吵架第一名”

# 8.代码 #

1.训练代码

import collections
    import numpy as np
    import tensorflow as tf
    
    # -------------------------------数据预处理---------------------------#
    poetry_file = 'poetry.txt'
    #poetry_file = './data/poetry.txt'
    
    # 诗集
    poetrys = []
    #加载诗集到一个数组，对于特殊字符进行处理，并且对于每行诗左右添加[]
    with open(poetry_file, "r", encoding='utf-8', ) as f:
        for line in f:
            try:
                title, content = line.strip().split(':')
                content = content.replace(' ', '')
                if '_' in content or '(' in content or '（' in content or '《' in content or '[' in content:
                    continue
                if len(content) < 5 or len(content) > 79:
                    continue
                content = '[' + content + ']'
                poetrys.append(content)
            except Exception as e:
                pass
    
    # 按诗的字数排序
    poetrys = sorted(poetrys, key=lambda line: len(line), reverse=False)
    print('唐诗总数: ', len(poetrys))
    
    # 统计每个字出现次数
    all_words = []
    for poetry in poetrys:
        temp = [word for word in poetry]
        all_words += temp
    
    counter = collections.Counter(all_words)
    print(counter.items())
    count_pairs = sorted(counter.items(), key=lambda x: -x[1])
    print(count_pairs)
    print(*count_pairs)
    words, _ = zip(*count_pairs)
    # 倒序排序后取出了所有字
    print(words)
    
    # 取前多少个常用字,并且加上空格
    print(len(words))
    words = words[:len(words)] + (' ',)
    print(words)
    print(len(words))
    
    # 每个字映射为一个数字ID
    word_num_map = dict(zip(words, range(len(words))))
    print(word_num_map)
    
    # 把诗转换为向量形式
    # 定义一个查索引的方法，如果是常用字就给index，如果不是就给默认值len(words)
    to_num = lambda word: word_num_map.get(word, len(words))
    poetrys_vector = [list(map(to_num, poetry)) for poetry in poetrys]
    # [[314, 3199, 367, 1556, 26, 179, 680, 0, 3199, 41, 506, 40, 151, 4, 98, 1],
    # [339, 3, 133, 31, 302, 653, 512, 0, 37, 148, 294, 25, 54, 833, 3, 1, 965, 1315, 377, 1700, 562, 21, 37, 0, 2, 1253, 21, 36, 264, 877, 809, 1]
    # ....]
    
    # 每次取256首诗进行训练
    batch_size = 256
    # 计算多少次可以把诗学完了
    n_chunk = len(poetrys_vector) // batch_size
    # 准备数据
    x_batches = []
    y_batches = []
    for i in range(n_chunk):
        start_index = i * batch_size
        end_index = start_index + batch_size
        # 每次取256首诗
        batches = poetrys_vector[start_index:end_index]
        # 计算256首诗里面最长的长度
        length = max(map(len, batches))
        # 创建全部为空格的索引号的矩阵
        xdata = np.full((batch_size, length), word_num_map[' '], np.int32)
        # 把每首诗的向量盖覆填入
        for row in range(batch_size):
            xdata[row, :len(batches[row])] = batches[row]
        ydata = np.copy(xdata)
        ydata[:, :-1] = xdata[:, 1:]
        
        # xdata             ydata
        # [6,2,4,6,9]       [2,4,6,9,9]
        # [1,4,2,8,5]       [4,2,8,5,5]
        
        x_batches.append(xdata)
        y_batches.append(ydata)
    
    # ---------------------------------------RNN--------------------------------------#
    
    input_data = tf.placeholder(tf.int32, [batch_size, None])
    output_targets = tf.placeholder(tf.int32, [batch_size, None])
    
    
    # 定义RNN
    def neural_network(model='lstm', rnn_size=128, num_layers=2):
        if model == 'rnn':
            cell_fun = tf.nn.rnn_cell.BasicRNNCell
        elif model == 'gru':
            cell_fun = tf.nn.rnn_cell.GRUCell
        elif model == 'lstm':
            cell_fun = tf.nn.rnn_cell.BasicLSTMCell
    
        cell = cell_fun(rnn_size, state_is_tuple=True)
        # 单个节点里面神经网络有两层，堆叠的，相当于网络层更深
        cell = tf.nn.rnn_cell.MultiRNNCell([cell] * num_layers, state_is_tuple=True)
    
        initial_state = cell.zero_state(batch_size, tf.float32)
    
        with tf.variable_scope('rnnlm'):
            # 构建从Cell单元输出得结果到输出层Y得W和b矩阵
            softmax_w = tf.get_variable("softmax_w", [rnn_size, len(words)])
            softmax_b = tf.get_variable("softmax_b", [len(words)])
            # 是来构建X输入到Cell之间得变化，说白了就是把X变成X_in交给RNN Cell单元
            with tf.device("/cpu:0"):
                embedding = tf.get_variable("embedding", [len(words), rnn_size])
                # 相当于对每个词进行one-hot编码再生成稠密的向量
                inputs = tf.nn.embedding_lookup(embedding, input_data)
    
        # 下面一行是来构建RNN网络拓扑结构
        # 如果是True，outputs的维度是[steps, batch_size, depth]
        outputs, last_state = tf.nn.dynamic_rnn(cell, inputs, initial_state=initial_state, scope='rnnlm')
        # reshape之后的形状是（steps*batch_size, 128）
        output = tf.reshape(outputs, [-1, rnn_size])
        # 计算从Cell单元输出的结果到输出层Y的结果
        logits = tf.matmul(output, softmax_w) + softmax_b
        probs = tf.nn.softmax(logits)
        return logits, last_state, probs, cell, initial_state
    
    
    # 训练
    def train_neural_network():
        #y^和最后时刻的输出
        logits, last_state, _, _, _ = neural_network()
        #将输出转换为一维数组
        targets = tf.reshape(output_targets, [-1])
        #计算batch中每一句话的累计交叉熵损失  如果使用sequence_loss，则会自带平均功能
        loss = tf.contrib.legacy_seq2seq.sequence_loss_by_example([logits], [targets], [tf.ones_like(targets, dtype=tf.float32)])
        #对于累计损失进行平均
        cost = tf.reduce_mean(loss)
        #定义学习率
        learning_rate = tf.Variable(0.0, trainable=False)
        tvars = tf.trainable_variables()
        # Gradient Clipping的引入是为了处理gradient explosion或者gradients vanishing的问题。当在一次迭代中权重的更新过于迅猛的话，
        # 很容易导致loss divergence。Gradient Clipping的直观作用就是让权重的更新限制在一个合适的范围。
        # clip_norm是截取的比率, 这个函数返回截取过的梯度张量
        # minimize() = compute_gradients() + apply_gradients()
        # 这里相当于将计算梯度和更新梯度变成两部分来做
        grads, _ = tf.clip_by_global_norm(tf.gradients(cost, tvars), 5)
        optimizer = tf.train.AdamOptimizer(learning_rate)
        train_op = optimizer.apply_gradients(zip(grads, tvars))
    
        with tf.Session() as sess:
            sess.run(tf.global_variables_initializer())
    
            saver = tf.train.Saver(tf.global_variables())
    
            for epoch in range(50):
                #学习率的下降策略
                sess.run(tf.assign(learning_rate, 0.002 * (0.97 ** epoch)))
                n = 0
                for batch in range(n_chunk):
                    train_loss, _ = sess.run([cost, train_op],
                                             feed_dict={input_data: x_batches[n], output_targets: y_batches[n]})
                    n += 1
                    print(epoch, batch, train_loss)
                if epoch % 7 == 0:
                    #7个轮次保存一下模型
                    saver.save(sess, './poetry.module', global_step=epoch)
    
    
    train_neural_network()

2.测试代码

import collections
    import numpy as np
    import tensorflow as tf
    
    # -------------------------------数据预处理---------------------------#
    
    poetry_file = 'poetry.txt'
    #以下为数据预处理部分
    #1.对内容进行处理，提取内容、去掉特殊字符、没首诗前后添加[]起止符
    #2.构建了包含所有单词的词表，并且使用词表将诗句转换成了数字向量形式
    #3.提取出每个批次的x和y数据
    
    
    # 加载诗集并对内容进行过滤
    #去掉特殊字符、长度小于5、长度大于79的诗内容
    #在每首诗内容前后分别添加[]
    poetrys = []
    with open(poetry_file, "r", encoding='utf-8', ) as f:
        for line in f:
            try:
                title, content = line.strip().split(':')
                content = content.replace(' ', '')
                if '_' in content or '(' in content or '（' in content or '《' in content or '[' in content:
                    continue
                if len(content) < 5 or len(content) > 79:
                    continue
                content = '[' + content + ']'
                poetrys.append(content)
            except Exception as e:
                pass
    
    # 按诗的字数排序 从小到大排序
    poetrys = sorted(poetrys, key=lambda line: len(line))
    print('唐诗总数: ', len(poetrys))
    
    # 统计每个字出现次数
    all_words = []
    for poetry in poetrys:
        all_words += [word for word in poetry]
    #以每个单词为key，其出现的次数为value，生成字典
    #Counter({'a': 6, 'b': 1, 'c': 1})
    counter = collections.Counter(all_words)
    #排序默认为降序
    count_pairs = sorted(counter.items(), key=lambda x: -x[1])
    #*表示元祖 **表示字典
    #将单词和其出现的次数压缩
    #产生的words就是需要的词表
    words, _ = zip(*count_pairs)
    
    # 在词表里面加了一个特殊的字符，空格
    #给词表后面加空格
    words = words[:len(words)] + (' ',)
    print(words)
    
    # 每个字映射为一个数字ID
    #用词表的长度给每个单词进行编码 1-len
    word_num_map = dict(zip(words, range(len(words))))
    print(word_num_map)
    
    # 把诗转换为向量形式
    #使用to_num找到诗句中每个字对应的编码
    #就可以将诗句用向量的形式表达
    to_num = lambda word: word_num_map.get(word, len(words))
    poetrys_vector = [list(map(to_num, poetry)) for poetry in poetrys]
    # [[314, 3199, 367, 1556, 26, 179, 680, 0, 3199, 41, 506, 40, 151, 4, 98, 1],
    # [339, 3, 133, 31, 302, 653, 512, 0, 37, 148, 294, 25, 54, 833, 3, 1, 965, 1315, 377, 1700, 562, 21, 37, 0, 2, 1253, 21, 36, 264, 877, 809, 1]
    # ....]
    
    batch_size = 1
    #确定需要训练的批次数
    n_chunk = len(poetrys_vector) // batch_size
    x_batches = []
    y_batches = []
    #提取出每个批次的x和y数据
    for i in range(n_chunk):
        start_index = i * batch_size
        end_index = start_index + batch_size
        #获取每个批次中的诗
        batches = poetrys_vector[start_index:end_index]
        #批次诗句最多字数
        length = max(map(len, batches))
        #创建(batchsize,length)形状的矩阵
        xdata = np.full((batch_size, length), word_num_map[' '], np.int32)
        #将每行诗中比max字数少的部分填充空格，生成x
        for row in range(batch_size):
            xdata[row, :len(batches[row])] = batches[row]
        ydata = np.copy(xdata)
        #y形状和x相同，数据为x去掉第一位+重复最后一位
        ydata[:, :-1] = xdata[:, 1:]
    
        # xdata             ydata
        # [6,2,4,6,9]       [2,4,6,9,9]
        # [1,4,2,8,5]       [4,2,8,5,5]
        
        x_batches.append(xdata)
        y_batches.append(ydata)
    # ---------------------------------------RNN--------------------------------------#
    #输入输出的形状为bs行，每行的最大词数列
    #x
    input_data = tf.placeholder(tf.int32, [batch_size, None])
    #y
    output_targets = tf.placeholder(tf.int32, [batch_size, None])
    
    
    # 定义RNN
    def neural_network(model='lstm', rnn_size=128, num_layers=2):
        if model == 'rnn':
            cell_fun = tf.nn.rnn_cell.BasicRNNCell
        elif model == 'gru':
            cell_fun = tf.nn.rnn_cell.GRUCell
        elif model == 'lstm':
            cell_fun = tf.nn.rnn_cell.BasicLSTMCell
        #创建lstm单元
        cell = cell_fun(rnn_size, state_is_tuple=True)
        #串联两个lstm
        cell = tf.nn.rnn_cell.MultiRNNCell([cell] * num_layers, state_is_tuple=True)
        #循环层初始时刻接受的上时刻状态值为0
        initial_state = cell.zero_state(batch_size, tf.float32)
    
        with tf.variable_scope('rnnlm'):
            #softmax_w softmax_b是最后多分类的参数和截距项
            softmax_w = tf.get_variable("softmax_w", [rnn_size, len(words)])
            softmax_b = tf.get_variable("softmax_b", [len(words)])
            with tf.device("/cpu:0"):
                #embedding执行两件事 1.对x进行onehot编码 2.全连接(因为编码后数据为0,1；所以只需要找到1对应的权重参数即可)
                embedding = tf.get_variable("embedding", [len(words), rnn_size])
                inputs = tf.nn.embedding_lookup(embedding, input_data)
        #根据上面定义的参数动态生成rnn结构
        #output表示rnn的输出 128个cell会有128个输出,last_state表示最后一个cell的输出
        outputs, last_state = tf.nn.dynamic_rnn(cell, inputs, initial_state=initial_state, scope='rnnlm')
        #将outputs转换为(batchsize,rnn_size),每一行可以代表一个批次的输出
        output = tf.reshape(outputs, [-1, rnn_size])
        #将循环层的值和输出层矩阵相乘，加上截距项，得到最终的输出结果
        logits = tf.matmul(output, softmax_w) + softmax_b
        #对每一个输出结果进行概率预测
        probs = tf.nn.softmax(logits)
        #最后输出 y^值，最后时刻的输出、y^概率、lstm cell、初始化参数
        return logits, last_state, probs, cell, initial_state
    
    
    # -------------------------------生成古诗---------------------------------#
    # 使用训练完成的模型
    
    def gen_poetry():
        #将概率数组转换为累积概率数组
        #随机从中抽取，概率越大选中的可能性越高
        #大概率和前面加和之后，相对于前面的距离更远，所以被选中的可能性也就越高
        def to_word(weights):
            #cumsum 返回数组元素在每一个时刻累计的和
            #如[1,2,3,4] 返回[ 1  3  6 10]
            t = np.cumsum(weights)
            #np.searchsorted 返回元素插入数组的位置，能保持数组仍为升序
            sample = int(np.searchsorted(t, np.random.rand(1)))
            return words[sample]
        #使用lstmrnn网络
        _, last_state, probs, cell, initial_state = neural_network()
    
        with tf.Session() as sess:
            sess.run(tf.global_variables_initializer())
            #创建模型保存器 要保存的数据为：tf.global_variables()
            saver = tf.train.Saver(tf.global_variables())
            #加载模型
            saver.restore(sess, './poetry.module-49')
            state_ = sess.run(cell.zero_state(batch_size, tf.float32))
            #生成的首字母
            first_letter = '高'
            #第一个输入的字对应的词向量
            x = np.array([list(map(word_num_map.get, first_letter))])
            print(x.shape)
            print(x)
            #使用x填充，并出示状态为0，得到rnn的输出
            [probs_, state_] = sess.run([probs, last_state],
                                        feed_dict={input_data: x, initial_state: state_})
            print(probs_)
            word = to_word(probs_)
            # word = words[np.argmax(probs_)]
            poem = first_letter + ''
            while word != '[' and word != ']':
                poem += word
                x = np.zeros((1, 1))
                x[0, 0] = word_num_map[word]
                [probs_, state_] = sess.run([probs, last_state],
                                            feed_dict={input_data: x, initial_state: state_})
                #按照概率随机选词
                word = to_word(probs_)
                # word = words[np.argmax(probs_)]
            return poem
    
    
    print(gen_poetry())