【自然语言处理（NLP）】从零实现循环神经网络RNN、Pytorch实现循环神经网络RNN

个人主页：道友老李
欢迎加入社区：道友老李的学习社区

介绍

**自然语言处理（Natural Language Processing，NLP）**是计算机科学领域与人工智能领域中的一个重要方向。它研究的是人类（自然）语言与计算机之间的交互。NLP的目标是让计算机能够理解、解析、生成人类语言，并且能够以有意义的方式回应和操作这些信息。

NLP的任务可以分为多个层次，包括但不限于：

1. 词法分析：将文本分解成单词或标记（token），并识别它们的词性（如名词、动词等）。
2. 句法分析：分析句子结构，理解句子中词语的关系，比如主语、谓语、宾语等。
3. 语义分析：试图理解句子的实际含义，超越字面意义，捕捉隐含的信息。
4. 语用分析：考虑上下文和对话背景，理解话语在特定情境下的使用目的。
5. 情感分析：检测文本中表达的情感倾向，例如正面、负面或中立。
6. 机器翻译：将一种自然语言转换为另一种自然语言。
7. 问答系统：构建可以回答用户问题的系统。
8. 文本摘要：从大量文本中提取关键信息，生成简短的摘要。
9. 命名实体识别（NER）：识别文本中提到的特定实体，如人名、地名、组织名等。
10. 语音识别：将人类的语音转换为计算机可读的文字格式。

NLP技术的发展依赖于算法的进步、计算能力的提升以及大规模标注数据集的可用性。近年来，深度学习方法，特别是基于神经网络的语言模型，如BERT、GPT系列等，在许多NLP任务上取得了显著的成功。随着技术的进步，NLP正在被应用到越来越多的领域，包括客户服务、智能搜索、内容推荐、医疗健康等。

从零实现RNN

梯度剪裁

符号含义

• $\mathbf{g}$：表示梯度向量，它是在神经网络训练过程中，计算损失函数关于模型参数的导数，用于指导参数更新。
• $\theta$：是一个超参数，代表设定的梯度阈值。
• $\Vert \mathbf{g}\Vert$：表示梯度向量$\mathbf{g}$的范数（通常是L2范数，即欧几里得范数 ，$\Vert \mathbf{g}\Vert =\sqrt{{\sum }_i{g}_i^2}$，$g_i$是梯度向量$\mathbf{g}$的第$i$个元素 ），用于衡量梯度向量的大小。

公式含义

公式$\mathbf{g}\leftarrow \min (1,\frac{\theta }{\Vert \mathbf{g}\Vert })\cdot \mathbf{g}$的作用是对梯度$\mathbf{g}$进行裁剪。具体来说：

• 当$\Vert \mathbf{g}\Vert \ge \theta$时，$\frac{\theta }{\Vert \mathbf{g}\Vert }\le 1$，此时$\min (1,\frac{\theta }{\Vert \mathbf{g}\Vert })=\frac{\theta }{\Vert \mathbf{g}\Vert }$，梯度$\mathbf{g}$将被缩放为$\frac{\theta }{\Vert \mathbf{g}\Vert }\cdot \mathbf{g}$，使得缩放后的梯度范数为$\theta$，即把过大的梯度进行了“收缩”，避免梯度爆炸问题。
• 当$\Vert \mathbf{g}\Vert <\theta$时，$\frac{\theta }{\Vert \mathbf{g}\Vert }>1$，则$\min (1,\frac{\theta }{\Vert \mathbf{g}\Vert })=1$，此时梯度$\mathbf{g}$保持不变，即当梯度大小在设定阈值范围内时，不进行任何处理。

这种梯度裁剪操作有助于在训练神经网络时稳定训练过程，防止梯度过大导致模型参数更新不稳定甚至无法收敛的情况。

RNN实现

1. 导包

dltools.py在资源里面

# 导包
%matplotlib inline

import math
import torch
from torch import nn
from torch.nn import functional as F
import dltools

2. 加载数据

加载time machine数据，article.txt在资源里面

# 加载time machine数据
batch_size, num_steps = 32, 35
# 默认需要从亚马逊aws云上面下载数据, 但是资源过期了, 下载不了了. 导致数据加载不了. 
train_iter, vocab = dltools.load_data_time_machine(batch_size=batch_size, num_steps=num_steps)

for x, y in train_iter:
    print(x, y)
    break

vocab.idx_to_token

['<unk>',
 ' ',
 'e',
 't',
 'a',
 'i',
 'n',
 'o',
 's',
 'h',
 'r',
 'd',
 'l',
 'm',
 'u',
 'c',
 'f',
 'w',
 'g',
 'y',
 'p',
 'b',
 'v',
 'k',
 'x',
 'z',
 'j',
 'q']

vocab.token_to_idx

{'<unk>': 0,
 ' ': 1,
 'e': 2,
 't': 3,
 'a': 4,
 'i': 5,
 'n': 6,
 'o': 7,
 's': 8,
 'h': 9,
 'r': 10,
 'd': 11,
 'l': 12,
 'm': 13,
 'u': 14,
 'c': 15,
 'f': 16,
 'w': 17,
 'g': 18,
 'y': 19,
 'p': 20,
 'b': 21,
 'v': 22,
 'k': 23,
 'x': 24,
 'z': 25,
 'j': 26,
 'q': 27}

输入数据,我们是打算输入one_hot编码的数据。pytorch提供了快速进行one_hot编码的工具

X = torch.arange(10).reshape((2, 5))
x = F.one_hot(X.T, 28)

3. 初始化模型参数

def get_params(vocab_size, num_hiddens, device):
    num_inputs = num_outputs = vocab_size
    
    def normal(shape):
        return torch.randn(size=shape, device=device) * 0.01
    
    # 隐藏层参数
    W_xh = normal((num_inputs, num_hiddens))
    W_hh = normal((num_hiddens, num_hiddens))
    b_h = torch.zeros(num_hiddens, device=device)
    
    # 输出层参数
    W_hq = normal((num_hiddens, num_outputs))
    b_q = torch.zeros(num_outputs, device=device)
    
    # nn.Parameter(W_xh)
    # 把这些参数都设置requires_grad = True
    params = [W_xh, W_hh, b_h, W_hq, b_q]
    for param in params:
        param.requires_grad_(True)
    return params

get_params(28, 512, 'cuda:0')

4. 初始化时返回隐藏状态

def init_rnn_state(batch_size, num_hiddens, device):
    # 返回的是一个元组
    return (torch.zeros((batch_size, num_hiddens), device=device), )

5. RNN主体结构

def rnn(inputs, state, params):
    # inputs的形状: (时间步数量, 批次大小, 词表大小)
    W_xh, W_hh, b_h, W_hq, b_q = params
    H, = state
    outputs = []
    # X的shape: [批次大小, 词表大小]
    for X in inputs:
        # 一般在循环神经网络中激活函数用tanh
        H = torch.tanh(torch.mm(X, W_xh) + torch.mm(H, W_hh) + b_h)
        Y = torch.mm(H, W_hq) + b_q
        outputs.append(Y)
    return torch.cat(outputs, dim=0), (H, )

包装成类

class RNNModelScratch:
    def __init__(self, vocab_size, num_hiddens, device, get_params, init_state, forward_fn):
        self.vocab_size, self.num_hiddens = vocab_size, num_hiddens
        self.params = get_params(vocab_size, num_hiddens, device)
        self.init_state, self.forward_fn = init_state, forward_fn
        
    def __call__(self, X, state):
        X = F.one_hot(X.T, self.vocab_size).type(torch.float32)
        return self.forward_fn(X, state, self.params)
    
    def begin_state(self, batch_size, device):
        return self.init_state(batch_size, self.num_hiddens, device)

6. 使用模型

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

num_hiddens = 512
net = RNNModelScratch(len(vocab), num_hiddens, dltools.try_gpu(), get_params, init_rnn_state, rnn)
state = net.begin_state(X.shape[0], dltools.try_gpu())
Y, new_state = net(X.to(dltools.try_gpu()), state)

查看返回的数据

Y.shape
len(new_state)
new_state[0].shape
vocab['a']
vocab.__getitem__('a')

torch.Size([10, 28])
1
4
4

7. 预测

def predict(prefix, num_preds, net, vocab, device):
    state = net.begin_state(batch_size=1, device=device)
    outputs = [vocab[prefix[0]]]
    get_input = lambda: torch.tensor([outputs[-1]], device=device).reshape((1, 1))
    # 预热
    for y in prefix[1:]:
        _, state = net(get_input(), state)
        outputs.append(vocab[y])
        
    # 真正的预测
    for _ in range(num_preds):
        y, state = net(get_input(), state)
        outputs.append(int(y.argmax(dim=1).reshape(1)))
    return ''.join([vocab.idx_to_token[i] for i in outputs])

predict('time traveller ', 20, net, vocab, dltools.try_gpu())

'time traveller                     '

8. 梯度裁剪

def grad_clipping(net, theta):
    if isinstance(net, nn.Module):
        params = [p for p in net.parameters() if p.requires_grad]
    else:
        params = net.params
    
    norm = torch.sqrt(sum(torch.sum((p.grad ** 2)) for p in params))
    if norm > theta:
        for param in params:
            param.grad[:] *= theta / norm

9. 训练

def train_epoch(net, train_iter, loss, updater, device, use_random_iter):
    state, timer = None, dltools.Timer()
    metric = dltools.Accumulator(2)
    # 取数据
    for X, Y in train_iter:
        if state is None or use_random_iter:
            # 第一次训练或者使用随机抽样时, 都需要把state重新初始化
            state = net.begin_state(batch_size=X.shape[0], device=device)
        else:
            # 梯度释放
            if isinstance(net, nn.Module) and not isinstance(state, tuple):
                state.detach_()
            else:
                for s in state:
                    s.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()
        if isinstance(updater, torch.optim.Optimizer):
            updater.zero_grad()
            l.backward()
            # rnn容易梯度爆炸. 使用梯度裁剪
            grad_clipping(net, 1)
            updater.step()
        else:
            l.backward()
            grad_clipping(net, 1)
            updater(batch_size=1)
#         print(l, y.numel())
        metric.add(l * y.numel(), y.numel())
    # 返回的是困惑度和每个字符平均训练时间. 
    return math.exp(metric[0] / metric[1]), metric[1] / timer.stop()

完整训练代码：

def train(net, train_iter, vocab, lr, num_epochs, device, use_random_iter=False):
    loss = nn.CrossEntropyLoss()
    animator = dltools.Animator(xlabel='epoch', ylabel='perlexity', legend=['train'], xlim=[10, num_epochs])
    
    # 初始化
    if isinstance(net, nn.Module):
        updater = torch.optim.SGD(net.parameters(), lr)
    else:
        updater = lambda batch_size: dltools.sgd(net.params, lr, batch_size)
        
    pred = lambda prefix: predict(prefix, 50, net, vocab, device)
    # 训练和预测
    for epoch in range(num_epochs):
        ppl, speed = train_epoch(net, train_iter, loss, updater, device, use_random_iter)
        
        if (epoch + 1) % 10 == 0:
            print(pred('time traveller'))
            animator.add(epoch + 1, [ppl])
    print(f'困惑度 {ppl:.1f}, {speed: .1f} 词元/秒 {str(device)}')
    print(pred('time traveller'))
    print(pred('traveller'))

9.1 顺序抽样

num_epochs, lr = 1000, 0.1
train(net, train_iter, vocab, lr, num_epochs, dltools.try_gpu())

9.2 随机抽样

num_epochs, lr = 1000, 0.1
train(net, train_iter, vocab, lr, num_epochs, dltools.try_gpu(), use_random_iter=True)

Pytorch实现RNN

导包和加载数据都和前面一样

import torch
from torch import nn
from torch.nn import functional as F
import dltools

batch_size, num_steps = 32, 35
train_iter, vocab = dltools.load_data_time_machine(batch_size, num_steps)

构造循环神经网络

构造了一个具有256个隐藏单元的单隐藏层的循环神经网络

num_hiddens = 256
rnn_layer = nn.RNN(len(vocab), num_hiddens)

初始化隐藏状态

state = torch.zeros((1, batch_size, num_hiddens))

创建随机的数据, 检测一下代码能不能正常跑

X = torch.rand(size=(num_steps, batch_size, len(vocab)))
Y, state_new = rnn_layer(X, state)
Y.shape, state_new.shape

完整的循环神经网络

class RNNModel(nn.Module):
    def __init__(self, rnn_layer, vocab_size, **kwargs):
        super().__init__(**kwargs)
        self.rnn = rnn_layer
        self.vocab_size = vocab_size
        self.num_hiddens = self.rnn.hidden_size
        
        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):
        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):
        return torch.zeros((self.num_directions * self.rnn.num_layers, batch_size, self.num_hiddens), device=device)

在训练之前, 基于随机初始化的权重进行预测, 跑一下模型

device = dltools.try_gpu()
net = RNNModel(rnn_layer, vocab_size=len(vocab))
net = net.to(device)
dltools.predict_ch8('time traveller', 10, net, vocab, device)

输出

'time traveller<unk>vivvvvvvv'

训练

num_epochs, lr = 500, 0.1
dltools.train_ch8(net, train_iter, vocab, lr, num_epochs, device)