LSTM复现 | Qing's Blog

引言

长短期记忆网络（Long Short-Term Memory, LSTM）是一种特殊的循环神经网络（Recurrent Neural Network, RNN），能够有效地学习和记忆长期依赖关系。由于其在处理序列数据方面的卓越表现，LSTM在自然语言处理、语音识别、时间序列预测等领域得到了广泛应用。本文将通过介绍LSTM的基本原理，并结合源码实现，帮助读者深入理解并复现LSTM模型。

LSTM架构示意图 LSTM架构示意图

LSTM介绍

1. 循环神经网络（RNN）概述

循环神经网络（RNN）是一种用于处理序列数据的神经网络结构。与传统的前馈神经网络不同，RNN具有内部的循环连接，能够在隐藏层中保留前一时刻的信息，从而捕捉序列中的时序特征。然而，标准的RNN在处理较长序列时，容易出现梯度消失或梯度爆炸的问题，导致模型难以学习长期依赖关系。

2. LSTM的诞生

RNN存在的问题：RNN虽然在理论上可以保留所有历史时刻的信息，但在实际使用时，信息的传递往往会因为时间间隔太长而逐渐衰减，传递一段时刻以后其信息的作用效果就大大降低了。因此，普通RNN对于信息的长期依赖问题没有很好的处理办法。而LSTM克服了这个问题，可以学习长期依赖信息

LSTM由Hochreiter和Schmidhuber在1997年提出，旨在解决标准RNN在长序列中的训练困难。LSTM通过引入门控机制，有效地控制信息的流动，能够在较长时间步内保持有用的信息，同时过滤掉无关或干扰的信息。

LSTM的关键是细胞状态 Ct，用来保存当前LSTM的状态信息并传递到下一时刻的LSTM中，也就是RNN中那根“自循环”的箭头。当前的LSTM接收来自上一个时刻的细胞状态 Ct−1，并与当前LSTM接收的信号输入 xt共同作用产生当前LSTM的细胞状态 Ct

3. LSTM的结构与工作原理

LSTM单元主要由以下几个部分组成：

遗忘门（Forget Gate）：决定保留多少过去的信息。
输入门（Input Gate）：决定引入多少新的信息。
输出门（Output Gate）：决定将多少当前的信息输出。

每个门都由一个Sigmoid神经网络层组成，输出值在0到1之间，用以控制信息的流动。具体结构如下图所示：

LSTM单元结构

图2：LSTM单元结构

具体计算过程

假设在时间步$t$，输入为$x_t$，前一时刻的隐藏状态为$h_{t-1}$，细胞状态为$c_{t-1}$，则LSTM的计算过程如下：

遗忘门：
$$ f_t = \sigma(W_f \cdot [h_{t-1}, x_t] + b_f) $$
输入门：

$$ i_t = \sigma(W_i \cdot [h_{t-1}, x_t] + b_i) $$ $$ \tilde{C}_t = \tanh(W_C \cdot [h_{t-1}, x_t] + b_C) $$

更新细胞状态：

$$
C_t = f_t * C_{t-1} + i_t * \tilde{C}_t
$$
输出门：

$$
o_t = \sigma(W_o \cdot [h_{t-1}, x_t] + b_o)
$$

$$
h_t = o_t * \tanh(C_t)
$$

其中，$\sigma$表示Sigmoid函数，$*$表示逐元素相乘。

LSTM源码实现

本文将以PyTorch框架为例，手把手实现一个简单的LSTM网络，并应用于字符级的文本预测任务。

1. 环境准备

首先，确保已经安装了PyTorch。可以使用以下命令进行安装：

1	pip install torch

2. 数据准备

以莎士比亚的文本为例，我们将其作为训练数据，用于预测下一个字符。

import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np

# 读取文本数据
with open('shakespeare.txt', 'r') as f:
    text = f.read()

# 创建字符到索引的映射
chars = sorted(list(set(text)))
char_to_idx = {ch: i for i, ch in enumerate(chars)}
idx_to_char = {i: ch for i, ch in enumerate(chars)}

vocab_size = len(chars)
print(f'字符集大小: {vocab_size}')

3. 模型定义

定义一个LSTM类，继承自nn.Module。

class LSTMModel(nn.Module):
    def __init__(self, input_size, hidden_size, output_size, num_layers=1):
        super(LSTMModel, self).__init__()
        self.hidden_size = hidden_size
        self.num_layers = num_layers

        # 定义LSTM层
        self.lstm = nn.LSTM(input_size, hidden_size, num_layers, batch_first=True)

        # 定义全连接层
        self.fc = nn.Linear(hidden_size, output_size)

    def forward(self, x, hidden):
        # LSTM前向传播
        out, hidden = self.lstm(x, hidden)
        # 取最后一个时间步的输出
        out = out[:, -1, :]
        out = self.fc(out)
        return out, hidden

    def init_hidden(self, batch_size):
        # 初始化隐藏状态和细胞状态
        weight = next(self.parameters()).data
        hidden = (weight.new(self.num_layers, batch_size, self.hidden_size).zero_(),
                  weight.new(self.num_layers, batch_size, self.hidden_size).zero_())
        return hidden

4. 模型训练

定义训练函数，并进行模型训练。

# 超参数
input_size = vocab_size
hidden_size = 128
output_size = vocab_size
num_layers = 2
num_epochs = 20
batch_size = 64
seq_length = 100
learning_rate = 0.002

model = LSTMModel(input_size, hidden_size, output_size, num_layers)
loss_fn = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=learning_rate)

# 准备数据集
def create_sequences(text, seq_length):
    sequences = []
    targets = []
    for i in range(0, len(text) - seq_length):
        seq = text[i:i+seq_length]
        target = text[i+seq_length]
        sequences.append([char_to_idx[ch] for ch in seq])
        targets.append(char_to_idx[target])
    return sequences, targets

sequences, targets = create_sequences(text, seq_length)
print(f'总序列数: {len(sequences)}')

# 转换为张量
inputs = torch.tensor(sequences, dtype=torch.long)
targets = torch.tensor(targets, dtype=torch.long)

# 训练循环
for epoch in range(num_epochs):
    hidden = model.init_hidden(batch_size)
    epoch_loss = 0
    num_batches = len(inputs) // batch_size

    for i in range(0, len(inputs) - batch_size, batch_size):
        x = inputs[i:i+batch_size]
        y = targets[i:i+batch_size]

        # 独热编码
        x_one_hot = torch.zeros(batch_size, seq_length, input_size)
        x_one_hot.scatter_(2, x.unsqueeze(2), 1)

        hidden = tuple([h.data for h in hidden])

        # 前向传播
        outputs, hidden = model(x_one_hot, hidden)
        loss = loss_fn(outputs, y)

        # 反向传播和优化
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()

        epoch_loss += loss.item()

    avg_loss = epoch_loss / num_batches
    print(f'Epoch [{epoch+1}/{num_epochs}], Loss: {avg_loss:.4f}')

5. 模型预测

训练完成后，可以使用模型生成文本。

def generate_text(model, start_str, length, hidden_size):
    model.eval()
    chars = [ch for ch in start_str]
    input_seq = torch.tensor([[char_to_idx[ch] for ch in start_str]], dtype=torch.long)
    input_one_hot = torch.zeros(1, len(start_str), vocab_size)
    input_one_hot.scatter_(2, input_seq.unsqueeze(2), 1)

    hidden = model.init_hidden(1)

    with torch.no_grad():
        for i in range(len(start_str) - 1):
            _, hidden = model(input_one_hot[:, i:i+1, :], hidden)

        last_char = input_one_hot[:, -1, :]

        for _ in range(length):
            out, hidden = model(last_char.unsqueeze(1), hidden)
            _, predicted = torch.max(out, 1)
            predicted_char = idx_to_char[predicted.item()]
            chars.append(predicted_char)

            # 准备下一个输入
            last_char = torch.zeros(1, 1, vocab_size)
            last_char[0, 0, predicted] = 1

    return ''.join(chars)

# 生成文本示例
start_str = "To be, or not to be, that is the question:\n"
generated = generate_text(model, start_str, 200, hidden_size)
print(generated)

引言