RNN

RNN概念

        循环神经网络（Recurrent Neural Network, RNN）是一种==专门处理序列数据的神经网络==。与传统的前馈神经网络不同，RNN具有“**循环**”结构，能够处理和记住前面时间步的信息，使其特别适用于时间序列数据或有时序依赖的任务。

        我们要明确什么是序列数据，时间序列数据是指在不同时间点上收集到的数据，这类数据反映了某一事物、现象等随时间的变化状态或程度。这是时间序列数据的定义，当然这里也可以不是时间，比如文字序列，但总归序列数据有一个特点——**后面的数据跟前面的数据有关系**。

RNN应用场景

- **自然语言处理（NLP）**：文本生成、语言建模、机器翻译、情感分析等。
- **时间序列预测**：股市预测、气象预测、传感器数据分析等。
- **语音识别**：将语音信号转换为文字。
- **音乐生成**：通过学习音乐的时序模式来生成新乐曲。

自然语言处理

概述

自然语言处理（Nature language Processing, NLP）研究的主要是通过计算机算法来理解自然语言。对于自然语言来说，处理的数据主要就是人类的语言，例如：汉语、英语、法语等，该类型的数据不像我们前面接触的过的结构化数据、或者图像数据可以很方便的进行数值化

词嵌入层

作用：

词嵌入层的作用就是将文本转换为向量，便于在网络模型中进行计算

词嵌入层在 RNN 中的作用有输入表示、降低维度和捕捉语义相似性。

词嵌入层首先会根据输入的词的数量构建一个词向量矩阵，

        例如: 我们有 100 个词，每个词希望转换成 128 维度的向量，那么构建的矩阵形状即为: 100*128，输入的每个词都对应了一个该矩阵中的一个向量。

工作流程：

• 初始化词向量：词嵌入层的初始词向量通常会使用随机初始化或者通过加载预训练的词向量（如Word2Vec或GloVe）进行初始化。
• 输入索引：每个单词在词汇表中都有一个唯一的索引。输入文本（例如一个句子）会先被分词，然后每个单词会被转换为相应的索引。
• 查找词向量：词嵌入层将这些单词索引映射为对应的词向量。这些词向量是一个低维稠密向量，表示该词的语义。
• 输入到RNN：这些词向量作为RNN的输入，RNN处理它们并根据上下文生成一个序列的输出。

API使用：

在 PyTorch 中，使用 nn.Embedding 词嵌入层来实现输入词的向量化。

nn.Embedding(num_embeddings=10, embedding_dim=4)

num_embeddings:词的数量

embedding_dim:表示用多少维的向量来表示每个词

示例：

构建流程：

将词转换为词向量，其步骤如下:

1.先将语料进行分词，构建词与索引的映射，我们可以把这个映射叫做词表，词表中每个词都对应了一个唯一的索引

2.然后使用 nn.Embedding 构建词嵌入矩阵，词索引对应的向量即为该词对应的数值化后的向量表示。

例子：我们的文本数据为: "北京冬奥的进度条已经过半，不少外国运动员在完成自己的比赛后踏上归途。"

# 导包
import jieba
import torch

"""
如果你希望得到一个可以直接操作的列表，推荐使用 jieba.lcut()。
如果你更关注性能或只需要遍历一次词语，可以使用 jieba.cut()。
"""
# 准备语料
text = "北京东奥的进度条已经过半,不少外国运动员在完成自己的比赛后踏上归途"
# TODO 分词
word_list = jieba.lcut(text)
print(word_list)
# TODO 去重
word_list = list(set(word_list))
print(len(word_list))
print('----------------------------------------------')
# TODO 构建词嵌入层
# 结论: num_embeddings是最少输入的单词数量, embedding_dim是词向量的维度
embedding = torch.nn.Embedding(num_embeddings=len(word_list), embedding_dim=10)
print(embedding)
print('----------------------------------------------')
# TODO 构建词向量矩阵
word_dict = {}
for idx, word in enumerate(word_list):
    # 构建词字典
    word_dict[word] = idx
    # TODO 根据索引生成对应的词向量
    word_vector = embedding(torch.tensor(idx))
    print(f"词:{word}\t索引:{idx}\t词向量:{word_vector.data}")

print(f"词表:{word_dict}")

循环网络层

RNN网络结构

        RNN（Recurrent Neural Network，循环神经网络）是一种专为处理序列数据设计的神经网络结构，其核心特点是利用循环连接（隐藏层的输出反馈到输入）来捕捉时间或顺序上的动态信息。一般以序列数据为输入, 通过网络内部的结构设计有效捕捉序列之间的关系特征, 一般也是以序列形式进行输出.

        为了表示出数据的序列关系，我们需要使用循环神经网络(Recurrent Nearal Networks, RNN) 来对数据进行建模，RNN 是一个具有记忆功能的网络，它作用于处理带有序列特点的样本数据。

RNN模型作用：

RNN的核心作用是建模序列数据的时序依赖关系，适用于需要上下文信息的任务。尽管其基础版本存在局限性，但通过LSTM/GRU等改进和结合注意力机制，RNN家族在序列建模中仍有重要地位。如文本分类, 意图识别, 机器翻译等.

例子说明：

以一个用户意图识别的例子进行简单的分析：

第一步：

第二步：

第三步：

第四步：

第五步：

RNN网络原理：

内部计算原理

RNN循环层API应用：

# 导包
import torch

# TODO 准备输入Xt数据
#第一个数字: 表示句子长度,也就是词语个数;第二个数字: 批量个数，也就是句子的个数;第三个数字: 词向量维度
x = torch.randn(size=[5, 32, 128])
# x = torch.randn(size=[2, 3, 4])
print(x.shape)
print(x)
print('-----------------------')
# # TODO 准备上初始隐藏状态h,默认值为0
h0 = torch.zeros(size=[1, 32, 256])
print(h0.shape)
print('-----------------------')
# # TODO 创建RNN层
# # input_size:输入x的特征维度,hidden_size:隐藏层维度,num_layers:RNN层数量；num_layers=1 表示只使用一层 RNN。
rnn = torch.nn.RNN(input_size=128, hidden_size=256, num_layers=1)
print(rnn)
print('-----------------------')
# # TODO 传入Xt和初始隐藏状态h0,返回输出和当前时间步隐藏状态
#x会变为([5,32,256])因为隐藏层维度是256。这是RNN的标准行为——将输入特征空间映射到隐藏状态特征空间。
y, h1 = rnn(x, h0) # 正向传播
print(h1.shape)
print('-----------------------')
print(y.shape)

文本生成示例：

项目需求：

文本生成任务是一种常见的自然语言处理任务，输入一个开始词能够预测出后面的词序列。本案例将会使用循环神经网络来实现周杰伦歌词生成任务。

数据集：

具体步骤：

1.构建词汇表

# TODO 1.构建词表
def build_vocab():
    # 提前构建两个列表,分别存储所有行分词列表和去重后每个分词
    all_words = []
    unique_words = []
    # 读取周杰伦歌词文件数据
    for line in open('data/jaychou_lyrics.txt', mode='r', encoding='utf-8'):
        # todo 使用jieba进行分词
        words = jieba.lcut(line)
        # todo 添加到all_words中
        all_words.append(words)
        # todo 把每个分词去重后添加到unique_words中
        for word in words:
            if word not in unique_words:
                unique_words.append(word)
    # 统计去重后单词数量
    unique_words_cnt = len(unique_words)
    # 字典推导式构建小词表: 每个单词对应一个索引
    unique_word_to_idx = {word: i for i, word in enumerate(unique_words)}
    # 构建存储所有数据索引的列表
    all_words_idx = []
    # 遍历all_words列表
    for words in all_words:
        # 定义临时存储一行的索引列表
        temp = []
        # 遍历每行文本拿到每个分词
        for word in words:
            # 根据每个分词获取索引添加temp中
            temp.append(unique_word_to_idx[word])
        # todo 为了后续每行内容索引区分开,添加空格对应的索引
        temp.append(unique_word_to_idx[' '])
        # 把当前行的所有索引添加到all_words_idx中
        all_words_idx.extend(temp)
    # 代码走到此处说明所有单词索引已经构建完成
    # print(all_words_idx)
    # todo 返回4个结果
    return unique_words, unique_words_cnt, unique_word_to_idx, all_words_idx

2.构建数据集

# TODO 2.构建数据集
class MyDataset(torch.utils.data.Dataset):
    # 重写__init__方法
    def __init__(self, all_words_idx, number_chars):
        # 属性1: 所有单词索引列表
        self.all_words_idx = all_words_idx
        # 属性2: 每个句子的长度
        self.number_chars = number_chars
        # 计算所有词数量
        self.all_words_idx_cnt = len(self.all_words_idx)
        # 计算句子数量
        self.number = self.all_words_idx_cnt // self.number_chars

    # 重写__len__方法
    def __len__(self):
        return self.number

    # 重写__getitem__方法
    def __getitem__(self, index):
        # 利用max()min()解决开始索引越界问题
        start = min(max(0, index), self.all_words_idx_cnt - 1 - self.number_chars)
        # 计算结束索引
        end = start + self.number_chars
        # todo 分别获取x和y
        x = self.all_words_idx[start:end]
        y = self.all_words_idx[start + 1:end + 1]
        # todo 因为后续是张量训练,此处需要将数据转成张量
        x = torch.tensor(x)
        y = torch.tensor(y)
        # todo 返回结果
        return x, y

3.编写网络模型

# TODO 3.构建模型
class MyModel(torch.nn.Module):
    # 重写__init__方法
    def __init__(self, unique_words_cnt):
        # 调用父类方法
        super().__init__()
        # todo 定义网络结构
        # 创建一个嵌入层
        self.embedding = torch.nn.Embedding(num_embeddings=unique_words_cnt, embedding_dim=128)
        # 创建一个rnn层
        self.rnn = torch.nn.RNN(input_size=128, hidden_size=256, num_layers=1)
        # 创建一个全连接层
        self.out = torch.nn.Linear(in_features=256, out_features=unique_words_cnt)

    # 重写forward方法
    def forward(self, x, h):
        # todo 获取嵌入层输出
        ebd_x = self.embedding(x)
        # todo 获取rnn层输出
        # TODO ebd_x[batch_size, seq_len, input_size]->(seq_len,batch_size, hidden_size)
        rnn_out, h = self.rnn(ebd_x.transpose(0, 1), h)
        # todo 全连接层: 只能处理2维张量,此处需要把rnn_out转成2维张量
        y_pred = self.out(rnn_out.reshape(-1, rnn_out.shape[-1]))
        # todo 返回结果
        return y_pred, h

    # 定义初始化隐藏状态方法
    def init_hidden(self, batch_size):
        # 创建初始隐藏状态
        return torch.zeros(size=[1, batch_size, 256])

4.编写训练函数

# TODO 4.模型训练
def train_model(train_loader, model, epochs):
    # 1.获取数据(本次已经传参)
    # 2.获取模型(本次已经传参)
    # 3.创建损失函数对象
    loss_fn = torch.nn.CrossEntropyLoss()
    # 4.创建优化器对象
    optimizer = torch.optim.Adam(model.parameters(), lr=0.0001)
    # 5.训练模型
    for epoch in range(epochs):
        total_loss, sample_cnt, start = 0.0, 0, time.time()
        for x, y in train_loader:
            # todo 正(前)向传播:从输入到输出: 预测值和损失值
            # 先获取隐藏状态
            h = model.init_hidden(x.size(0))
            # 然后预测获取预测值和隐藏状态
            y_pred, h = model(x, h)
            # 计算损失值
            loss = loss_fn(y_pred, y.transpose(0, 1).reshape(-1))
            # 累加损失值和批次数量
            total_loss += loss.item()
            sample_cnt += 1
            # todo 反向传播:从输出到输入: 梯度计算和参数更新
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
        # 走到此处,说明一轮结束: 计算每轮损失值
        epoch_loss = total_loss / sample_cnt
        print(f"第{epoch + 1}轮,运行时间{time.time() - start:.2f}秒,损失值为:{epoch_loss:.2f}")
    # 6.保存训练好的模型参数字典
    torch.save(model.state_dict(), 'model/text_generator_model_dict.pth')

5.编写预测函数

# TODO 5.模型评估
def eval_model(start_word, length, unique_words_cnt, unique_words, unique_word_to_idx):
    # 1.获取起始词对应的索引,并封装成列表
    word_idx = unique_word_to_idx[start_word]
    words = [word_idx]
    # 2.创建新模型
    model = MyModel(unique_words_cnt)
    model.load_state_dict(torch.load('model/text_generator_model_dict.pth'))
    # 3.模型预测
    # 提前获取隐藏状态
    h = model.init_hidden(batch_size=1)
    for i in range(length):
        # 模型预测
        ypred, h = model(torch.tensor([[word_idx]]), h)
        # 获取预测值索引
        word_idx = torch.argmax(ypred).item()
        words.append(word_idx)
    # print(words)
    # 4.将索引列表转成词列表
    words = [unique_words[idx] for idx in words]
    print(''.join(words))

6.主函数

if __name__ == '__main__':
    # TODO 1.构建词表
    unique_words, unique_words_cnt, unique_word_to_idx, all_words_idx = build_vocab()
    with open('data/词表.txt', 'w', encoding='utf-8') as f:
        for word in unique_words:
            f.write(word + '\n')
    # # TODO 2.构建数据集
    number_chars = 32
    dataset = MyDataset(all_words_idx, number_chars)
    # print(f'数据集大小: {len(dataset)}')  # 自动调用__len__方法
    # print(dataset[0])  # 自动调用了__getitem__方法: (tensor([ 0,  1,  2,  3, 40]), tensor([ 1,  2,  3, 40,  0]))
    batch_size = 5
    train_loader = torch.utils.data.DataLoader(dataset, batch_size=batch_size, shuffle=True)
    # TODO 3.构建模型
    model = MyModel(unique_words_cnt)
    print(model)
    # TODO 4.模型训练
    epochs = 10
    # train_model(train_loader, model, epochs)
    # TODO 5.模型评估
    # 注意: start_word必须在词表中有,否则报错
    start_word = '感谢'
    length = 99
    eval_model(start_word, length, unique_words_cnt, unique_words, unique_word_to_idx)