一.通用代码

各循环神经网络的大多代码相同，记录在此处，之后不再重复

import collections
import re
import torch
from torch import nn
from torch.nn import functional as F
import random


device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
print(device)

#######################################加载数据#####################################
# 读取数据集
def read_time_machine():
    with open('./database/timemachine.txt', 'r', encoding='utf-8') as f:
        lines = f.readlines()
    for i in range(len(lines)):
        # 替换所有非字母字符为空格
        lines[i] = re.sub('[^A-Za-z]+', ' ', lines[i]).strip().lower()
    return lines

lines = read_time_machine()

# tooken化数据集
def tokenize(lines, token='word'):
    if token == 'word':
        return [line.split() for line in lines]
    elif token == 'char':
        return [list(line) for line in lines]
    else:
        print('错误:未知的token类型' + token)
        return None


# 统计词频
def count_corpus(tokens):
    if len(tokens) == 0 or isinstance(tokens[0], list):
        # 将词元列表展平成一个列表
        tokens = [token for line in tokens for token in line]
    return collections.Counter(tokens)

# 创建词表
class Vocab:
    def __init__(self, tokens=None, min_freq=0, reserved_tokens=None):
        if tokens is None:
            tokens = []
        if reserved_tokens is None:
            reserved_tokens = []
        # 按出现频率排序
        counter = count_corpus(tokens)
        self._token_freqs = sorted(counter.items(), key=lambda x: x[1], reverse=True)
        # 未知词元的索引为0
        self.idx_to_token = ['<unk>'] + reserved_tokens
        self.token_to_idx = {token: idx for idx, token in enumerate(self.idx_to_token)}

        for token, freq in self._token_freqs:
            if freq < min_freq:
                break
            if token not in self.token_to_idx:
                self.idx_to_token.append(token)
                self.token_to_idx[token] = len(self.idx_to_token) - 1

    # 给定词元返回索引
    def __getitem__(self, tokens):
        if not isinstance(tokens, (list, tuple)):
            return self.token_to_idx.get(tokens, self.unk)
        return [self.__getitem__(token) for token in tokens]

    # 返回词表的大小
    def __len__(self):
        return len(self.idx_to_token)

    # 给定索引返回词元
    def to_tokens(self, indices):
        if not isinstance(indices, (list, tuple)):
            return self.idx_to_token[indices]
        return [self.idx_to_token[index] for index in indices]

    # 返回未知词元的索引
    @property
    def unk(self):
        return 0

    @property
    def token_freqs(self):
        return self._token_freqs

def get_corpus_timemachine(max_tokens=-1):

    lines = read_time_machine()
    tokens = tokenize(lines)
    vocab = Vocab(tokens)
    # 展平所有文本行
    corpus = [vocab[token] for line in tokens for token in line]
    if max_tokens > 0:
        corpus = corpus[:max_tokens]
    return corpus, vocab

corpus, vocab = get_corpus_timemachine()

# 随机采样样本
def seq_data_iter_random(corpus, batch_size, num_steps):
    # 从随机偏移量开始对序列进行分区，随机范围包括num_steps-1
    corpus = corpus[random.randint(0, num_steps - 1):]
    num_subseqs = (len(corpus) - 1) // num_steps
    # 每个样本的开始索引
    initial_indices = list(range(0, num_subseqs * num_steps, num_steps))
    random.shuffle(initial_indices)

# 顺序分区样本
def seq_data_iter_sequential(corpus, batch_size, num_steps):
    offset = random.randint(0, num_steps)
    num_tokens = (len(corpus) - offset - 1) // batch_size
    # 提取特征序列和标签序列
    Xs = torch.tensor(corpus[offset: offset + num_tokens * batch_size])
    Ys = torch.tensor(corpus[offset + 1: offset + num_tokens * batch_size + 1])
    Xs, Ys = Xs.reshape(batch_size, -1), Ys.reshape(batch_size, -1)
    num_batches = Xs.shape[1] // num_steps
    for i in range(0, num_batches * num_steps, num_steps):
        X = Xs[:, i: i + num_steps]
        Y = Ys[:, i: i + num_steps]
        yield X, Y

class SeqDataLoader:
    """
    加载序列数据的迭代器
    """
    def __init__(self, corpus, batch_size, num_steps, use_random_iter, max_tokens):
        if use_random_iter:
            self.data_iter_fn = seq_data_iter_random
        else:
            self.data_iter_fn = seq_data_iter_sequential
        self.corpus, self.vocab = get_corpus_timemachine(max_tokens)
        self.batch_size, self.num_steps = batch_size, num_steps
    def __iter__(self):
        return self.data_iter_fn(self.corpus, self.batch_size, self.num_steps)

# 加载数据
def load_data_timemachine(batch_size, num_steps, use_random_iter=False, max_tokens=10000):
    data_iter = SeqDataLoader(corpus, batch_size, num_steps, use_random_iter, max_tokens)
    return data_iter, data_iter.vocab


#####################################训练模型函数##############################################
# 梯度裁剪函数
def grad_clipping(net, theta):
    params = [p for p in net.parameters() if p.requires_grad]
    norm = torch.sqrt(sum(torch.sum((p.grad ** 2)) for p in params))
    if norm > theta:
        for param in params:
            param.grad[:] *= theta / norm


def train_RNN(net, train_iter, vocab, lr, num_epochs, device, use_random_iter=False):
    """
    训练RNN模型
    """
    # 损失函数
    loss = nn.CrossEntropyLoss()
    # 优化器
    optimizer = torch.optim.SGD(net.parameters(), lr)
    # 训练
    for epoch in range(num_epochs):
        state = None

        for X, Y in train_iter:
            if state is None or use_random_iter:
                # 初始化隐藏状态
                state = net.begin_state(batch_size=X.shape[0], device=device)
            else:
                if isinstance(state, tuple):
                    # 针对 LSTM 模型，分离隐藏状态和细胞状态
                    state = tuple(s.detach() for s in state)
                else:
                    # 针对普通 RNN 模型，分离隐藏状态
                    state = state.detach()

            y = Y.T.reshape(-1)
            X, y = X.to(device), y.to(device)
            y_hat, state = net(X, state)
            l = loss(y_hat, y.long()).mean()
            optimizer.zero_grad()
            l.backward()
            grad_clipping(net, 1)
            optimizer.step()
        # 获得损失
        l = l.item()
        print(f'epoch:{epoch+1} last loss: {l:.3f}')




#######################################RNN模型################################################
# 定义模型
class RNNModel(nn.Module):
    def __init__(self, rnn_layer, vocab_size, **kwargs):
        super(RNNModel, self).__init__(**kwargs)
        self.rnn = rnn_layer
        self.vocab_size = vocab_size
        self.num_hiddens = self.rnn.hidden_size
        # 如果RNN是双向的，num_directions=2，否则为1
        if not self.rnn.bidirectional:
            self.num_directions = 1
            self.linear = nn.Linear(self.num_hiddens, self.vocab_size)
        else:
            self.num_directions = 2
            self.linear = nn.Linear(self.num_hiddens * 2, self.vocab_size)

    def forward(self, inputs, state):
        # inputs的形状：(时间步数，批量大小，词表大小)
        X = F.one_hot(inputs.T.long(), self.vocab_size)
        X = X.to(torch.float32)
        Y, state = self.rnn(X, state)
        # 输出是(时间步数*批量大小，词表大小)
        output = self.linear(Y.reshape((-1, Y.shape[-1])))
        return output, state

    def begin_state(self, device, batch_size=1):
        """
        初始化隐藏状态
        """
        if not isinstance(self.rnn, nn.LSTM): # 不是LSTM层
            # nn.GRU以张量作为隐藏状态
            return torch.zeros((self.num_directions * self.rnn.num_layers,
                                batch_size, self.num_hiddens),
                               device=device)
        else:
            # nn.LSTM以元组作为隐藏状态
            return (torch.zeros((self.num_directions * self.rnn.num_layers,
                                 batch_size, self.num_hiddens),
                                device=device),
                    torch.zeros((self.num_directions * self.rnn.num_layers,
                                 batch_size, self.num_hiddens),
                                device=device))

二.门控循环单元GRU

1.基本概念

重置门：决定上一时刻隐状态有多少信息要被重置，有助于捕获序列中的短期依赖关系

更新门：决定候选隐状态时旧隐藏状态的比例，有助于捕获序列中的长期依赖关系

将输入与重置门结合，得到候选隐状态：

再和更新门结合，得到新的隐状态：

2.pytorch实现GRU

#####################################GRU模型######################################
# 初始化超参
batch_size, num_steps = 32, 35
num_hiddens = 256
num_epochs = 500
lr = 1

#加载数据
train_iter, vocab = load_data_timemachine(batch_size, num_steps)

# 创建模型
num_inputs = len(vocab)
gru_layer = nn.GRU(num_inputs, num_hiddens)
net = RNNModel(gru_layer, len(vocab))
net = net.to(device)

# 训练模型
train_RNN(net, train_iter, vocab, lr, num_epochs, device)

三.长短期记忆网络LSTM

1.基本概念

GRU是LSTM的优化版，两者类似，LSTM更古老，也较复杂

输入门：决定何时将数据读入单元

输出门：用来从单元中输出条目

忘记门：用来重置单元的内容，类似重置门

候选记忆元：决定是否要存入新记忆

2.pytorch实现LSTM

#######################################LSTM模型######################################
# 初始化超参
batch_size, num_steps = 32, 35
num_hiddens = 256
num_epochs = 500
lr = 1

#加载数据
train_iter, vocab = load_data_timemachine(batch_size, num_steps)

# 创建模型
num_inputs = len(vocab)
lstm_layer = nn.LSTM(num_inputs, num_hiddens)
net = RNNModel(lstm_layer, len(vocab))
net = net.to(device)

# 训练模型
train_RNN(net, train_iter, vocab, lr, num_epochs, device)

四.深度循环神经网络

1.基本概念

第l个隐藏层的隐状态公式为：

2.pytorch实现深度循环神经网络

##############################深度循环神经网络######################################
# 初始化超参
batch_size, num_steps = 32, 35
num_hiddens = 256
num_epochs = 500
lr = 2
num_layers = 2

#加载数据
train_iter, vocab = load_data_timemachine(batch_size, num_steps)

# 创建模型
num_inputs = len(vocab)
lstm_layers = nn.LSTM(num_inputs, num_hiddens, num_layers)
net = RNNModel(lstm_layers, len(vocab))
net = net.to(device)

# 训练模型
train_RNN(net, train_iter, vocab, lr, num_epochs, device)