从零开始实现循环神经网络

我假设你对基础的神经网络已经有一定的了解。如果没有，你可以先阅读从零开始实现神经网络，它会引导你理解非循环网络背后的思想和实现方法。

引言

本篇博客受到WildML上的循环神经网络教程的启发。你可以深入阅读那篇文章来掌握RNN的基础知识，这些内容在本教程中将不再赘述。

在本教程中，我们将重点讲解如何基于RNN的计算图，通过时间反向传播（BPTT）进行训练，并实现自动微分。你会发现，用这种方式计算梯度比手动推导要更加简单可靠。

本文将以RNN语言模型（rnnlm）为例。关于RNN更高级的应用，可以参考Karpathy的文章。

如何训练RNN

RNN的结构如下图所示：

可以看到，参数(W, U, V)在不同时间步之间是共享的。每个时间步的输出可以经过softmax处理。因此，我们可以使用交叉熵损失作为误差函数，并采用优化算法（如梯度下降法）来求解最优参数(W, U, V)。

回顾一下我们的RNN公式：

我们定义的损失函数为交叉熵损失，表达式如下：

其中y_t是时间步t的真实标签，y^_t是我们预测的值。通常我们会把整个序列（句子）当作一个训练样本，因此总误差就是各个时间步（单词）误差的总和。

记住，我们的目标是计算误差对参数U、V和W的梯度，然后利用优化算法（本教程中使用随机梯度下降法）来更新参数。就像我们对误差求和一样，我们也需要对每个时间步的梯度进行累加：。也就是说，我们需要分别计算dEt/dW、dEt/dU和dEt/dV，并将所有时间步的结果相加。

计算dEt/dV相对简单，因为它只依赖于当前时间步的值。然而，dEt/dW和dEt/dU则复杂得多。注意，s_3 = tanh(Ux_3 + Ws_2)依赖于s_2，而s_2又依赖于W、U和s_1，依此类推。因此，在对W求导时，我们不能把s_2视为常数！必须再次应用链式法则。具体过程可以参考下图。

现在我们以E1为例，用计算图来表示，并计算dE1/dW和dE1/dU，思路是一样的。

需要注意的是，这与我们在深度前馈神经网络中使用的标准反向传播算法完全一致。关键区别在于，我们在RNN中需要对每个时间步的梯度进行累加。而在传统的前馈网络中，各层的参数并不共享，因此不需要做这样的累加操作。在我看来，BPTT不过是展开后的RNN上应用标准反向传播的一种更炫酷的称呼而已。

为了简化计算图并提高效率，我们可以将一些小的操作单元整合为一个大的操作单元。请看下图。需要注意的是，每个操作单元都需要实现forward和backward函数。

以下是所有操作单元以及softmax输出的实现代码：

mulGate = MultiplyGate()
addGate = AddGate()
activation = Tanh()

class RNNLayer:
    def forward(self, x, prev_s, U, W, V):
        self.mulu = mulGate.forward(U, x)
        self.mulw = mulGate.forward(W, prev_s)
        self.add = addGate.forward(self.mulw, self.mulu)
        self.s = activation.forward(self.add)
        self.mulv = mulGate.forward(V, self.s)
        
    def backward(self, x, prev_s, U, W, V, diff_s, dmulv):
        self.forward(x, prev_s, U, W, V)
        dV, dsv = mulGate.backward(V, self.s, dmulv)
        ds = dsv + diff_s
        dadd = activation.backward(self.add, ds)
        dmulw, dmulu = addGate.backward(self.mulw, self.mulu, dadd)
        dW, dprev_s = mulGate.backward(W, prev_s, dmulw)
        dU, dx = mulGate.backward(U, x, dmulu)
        return (dprev_s, dU, dW, dV)

class MultiplyGate:
    def forward(self,W, x):
        return np.dot(W, x)
    def backward(self, W, x, dz):
        dW = np.asarray(np.dot(np.transpose(np.asmatrix(dz)), np.asmatrix(x)))
        dx = np.dot(np.transpose(W), dz)
        return dW, dx

class AddGate:
    def forward(self, x1, x2):
        return x1 + x2
    def backward(self, x1, x2, dz):
        dx1 = dz * np.ones_like(x1)
        dx2 = dz * np.ones_like(x2)
        return dx1, dx2

class Sigmoid:
    def forward(self, x):
        return 1.0 / (1.0 + np.exp(-x))
    def backward(self, x, top_diff):
        output = self.forward(x)
        return (1.0 - output) * output * top_diff

class Tanh:
    def forward(self, x):
        return np.tanh(x)
    def backward(self, x, top_diff):
        output = self.forward(x)
        return (1.0 - np.square(output)) * top_diff

class Softmax:
    def predict(self, x):
        exp_scores = np.exp(x)
        return exp_scores / np.sum(exp_scores)
    def loss(self, x, y):
        probs = self.predict(x)
        return -np.log(probs[y])
    def diff(self, x, y):
        probs = self.predict(x)
        probs[y] -= 1.0
        return probs

这些实现与从零开始实现神经网络中的代码基本相同，唯一的区别在于：本教程中的输入x或s是一维数组，而在之前的教程中，输入X是一个矩阵形式的数据批次（每行代表一个样本）。

现在我们已经能够计算出参数的梯度，接下来就可以使用SGD来训练模型了。

实现

初始化

初始化参数 U、V 和 W 有点讲究。我们不能简单地将它们初始化为 0，因为那样会导致所有层的计算对称。我们必须随机初始化这些权重。由于合适的初始化似乎会对训练结果产生影响，这一领域已经进行了大量研究。事实证明，最佳的初始化方式取决于激活函数（在我们的例子中是 tanh），一种推荐的做法是将权重随机初始化在区间 内，其中 n 是来自前一层的输入连接数。

class Model:
    def __init__(self, word_dim, hidden_dim=100, bptt_truncate=4):
        self.word_dim = word_dim
        self.hidden_dim = hidden_dim
        self.bptt_truncate = bptt_truncate
        self.U = np.random.uniform(-np.sqrt(1. / word_dim), np.sqrt(1. / word_dim), (hidden_dim, word_dim))
        self.W = np.random.uniform(-np.sqrt(1. / hidden_dim), np.sqrt(1. / hidden_dim), (hidden_dim, hidden_dim))
        self.V = np.random.uniform(-np.sqrt(1. / hidden_dim), np.sqrt(1. / hidden_dim), (word_dim, hidden_dim))

上述代码中，word_dim 是词汇表的大小，而 hidden_dim 是隐藏层的大小（我们可以自行选择）。目前可以先不用管 bptt_truncate 参数，稍后我们会解释它的作用。

前向传播

接下来，让我们实现由上述公式定义的前向传播（预测单词概率）：

'''
    前向传播（预测单词概率）
    x 既可以是一个单独的数据点，也可以是一批数据
    例如 x = [0, 179, 341, 416], 那么 y = [179, 341, 416, 1]
'''
def forward_propagation(self, x):
    # 总的时间步数
    T = len(x)
    layers = []
    prev_s = np.zeros(self.hidden_dim)
    # 对于每一个时间步...
    for t in range(T):
        layer = RNNLayer()
        input = np.zeros(self.word_dim)
        input[x[t]] = 1
        layer.forward(input, prev_s, self.U, self.W, self.V)
        prev_s = layer.s
        layers.append(layer)
    return layers

我们还实现了一个 predict 函数来生成最终结果。

def predict(self, x):
    output = Softmax()
    layers = self.forward_propagation(x)
    return [np.argmax(output.predict(layer.mulv)) for layer in layers]

计算损失

为了训练我们的网络，我们需要一种方法来衡量它产生的误差。我们称这种度量为损失函数 L，我们的目标是找到能够使训练数据上的损失函数最小化的参数 U、V 和 W。常用的损失函数是 交叉熵 损失。

def calculate_loss(self, x, y):
    assert len(x) == len(y)
    output = Softmax()
    layers = self.forward_propagation(x)
    loss = 0.0
    for i, layer in enumerate(layers):
        loss += output.loss(layer.mulv, y[i])
    return loss / float(len(y))

def calculate_total_loss(self, X, Y):
    loss = 0.0
    for i in range(len(Y)):
        loss += self.calculate_loss(X[i], Y[i])
    return loss / float(len(Y))

时间反向传播（BPTT）

正如前面介绍的那样，我们实现了 BPTT 算法。它以一个训练样本 (x, y) 作为输入，并返回梯度 dL/dW、dL/dU 和 dL/dV。

def bptt(self, x, y):
    assert len(x) == len(y)
    output = Softmax()
    layers = self.forward_propagation(x)
    dU = np.zeros(self.U.shape)
    dV = np.zeros(self.V.shape)
    dW = np.zeros(self.W.shape)

    T = len(layers)
    prev_s_t = np.zeros(self.hidden_dim)
    diff_s = np.zeros(self.hidden_dim)
    for t in range(0, T):
        dmulv = output.diff(layers[t].mulv, y[t])
        input = np.zeros(self.word_dim)
        input[x[t]] = 1
        dprev_s, dU_t, dW_t, dV_t = layers[t].backward(input, prev_s_t, self.U, self.W, self.V, diff_s, dmulv)
        prev_s_t = layers[t].s
        dmulv = np.zeros(self.word_dim)
        for i in range(t-1, max(-1, t-self.bptt_truncate-1), -1):
            input = np.zeros(self.word_dim)
            input[x[i]] = 1
            prev_s_i = np.zeros(self.hidden_dim) 如果 i == 0 else layers[i-1].s
            dprev_s, dU_i, dW_i, dV_i = layers[i].backward(input, prev_s_i, self.U, self.W, self.V, dprev_s, dmulv)
            dU_t += dU_i
            dW_t += dW_i
        dV += dV_t
        dU += dU_t
        dW += dW_t
    return (dU, dW, dV)

SGD 的实现

现在我们已经能够计算出参数的梯度，就可以实现随机梯度下降（SGD）了。我喜欢分两步来实现：

1. 一个 sgd_step 函数，用于计算梯度并对一个批次进行更新。
2. 外层循环遍历整个训练集，并调整学习率。

def sgd_step(self, x, y, learning_rate):
    dU, dW, dV = self.bptt(x, y)
    self.U -= learning_rate * dU
    self.V -= learning_rate * dV
    self.W -= learning_rate * dW
   
def train(self, X, Y, learning_rate=0.005, nepoch=100, evaluate_loss_after=5):
    num_examples_seen = 0
    losses = []
    for epoch in range(nepoch):
        if (epoch % evaluate_loss_after == 0):
            loss = self.calculate_total_loss(X, Y)
            losses.append((num_examples_seen, loss))
            time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            print("%s: Loss after num_examples_seen=%d epoch=%d: %f" % (time, num_examples_seen, epoch, loss))
            # 如果损失增加，则调整学习率
            if len(losses) > 1 and losses[-1][1] > losses[-2][1]:
                learning_rate = learning_rate * 0.5
                print("Setting learning rate to %f" % learning_rate)
            sys.stdout.flush()
        # 对于每一个训练样本...
        for i in range(len(Y)):
            self.sgd_step(X[i], Y[i], learning_rate)
            num_examples_seen += 1
    return losses 

在这里，如果发现当前 epoch 的损失有所增加，我们就将学习率减半。关于学习率衰减的更多信息可以参见 这里。

评估

好了！让我们来估算一下训练我们的网络需要多长时间：

word_dim = 8000
hidden_dim = 100
X_train, y_train = getSentenceData('data/reddit-comments-2015-08.csv', word_dim)

np.random.seed(10)
rnn = Model(word_dim, hidden_dim)
rnn.sgd_step(X_train[10], y_train[10], 0.005)

坏消息是，在我的笔记本上，SGD 的一步就需要几秒钟。我们的训练数据大约有 8 万个样本，因此完成一个 epoch（遍历整个数据集）就需要好几个小时。而要完成多个 epoch，则可能需要几天，甚至几周！

我们可以通过多种方式加速代码的运行。一种方法是使用像 Theano 这样的库，在 GPU 上实现我们的代码。不过在本教程中，我们先尝试用一个小数据集运行 SGD，并检查损失是否真的在下降：

word_dim = 8000
hidden_dim = 100
X_train, y_train = getSentenceData('data/reddit-comments-2015-08.csv', word_dim)

np.random.seed(10)
rnn = Model(word_dim, hidden_dim)

losses = rnn.train(X_train[:100], y_train[:100], learning_rate=0.005, nepoch=10, evaluate_loss_after=1)

不错，看起来我们的实现至少在做一些有用的事情，并且损失确实在下降，这正是我们期望的结果。

更进一步

RNN 存在一个被称为“梯度消失问题”的难题。这也是为什么传统的 RNN 无法捕捉长期依赖关系的原因。因此，我们使用 bptt_truncate 参数来限制依赖关系的长度。这也促使我们转向更复杂的 RNN 模型，比如 LSTM，它目前在许多自然语言处理任务中处于最先进水平。

关于“梯度消失问题”和“LSTM”的更多信息可以参见 这里、这里以及 这里。

参考文献

1. https://github.com/pangolulu/neural-network-from-scratch
2. http://www.wildml.com/2015/09/recurrent-neural-networks-tutorial-part-1-introduction-to-rnns/
3. http://www.wildml.com/2015/09/recurrent-neural-networks-tutorial-part-2-implementing-a-language-model-rnn-with-python-numpy-and-theano/
4. http://www.wildml.com/2015/10/recurrent-neural-networks-tutorial-part-3-backpropagation-through-time-and-vanishing-gradients/

rnn-from-scratch 快速上手指南

本指南旨在帮助开发者从零开始理解并实现循环神经网络（RNN），重点讲解基于计算图的自动微分和随时间反向传播（BPTT）算法。

环境准备

本项目主要依赖 Python 和 NumPy 进行数值计算，无需复杂的深度学习框架（如 TensorFlow 或 PyTorch），适合学习原理。

• 操作系统: Linux, macOS 或 Windows
• Python 版本: 推荐 Python 3.6+
• 核心依赖:
  • numpy: 用于矩阵运算和数组操作

确保已安装 pip 包管理工具。国内用户建议使用清华源或阿里源加速安装。

安装步骤

1. 克隆项目代码 将仓库克隆到本地：

   git clone https://github.com/pangolulu/rnn-from-scratch.git
   cd rnn-from-scratch
   

2. 安装依赖 使用 pip 安装必要的 Python 库。国内用户推荐使用以下命令加速下载：

   pip install numpy -i https://pypi.tuna.tsinghua.edu.cn/simple
   

   (注：如果项目中包含 requirements.txt，也可运行 pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple)

基本使用

该项目主要通过 Python 脚本演示 RNN 的语言模型（RNNLM）训练过程。核心逻辑包括前向传播、损失计算、BPTT 反向传播以及 SGD 优化。

1. 模型初始化

模型参数 $U, V, W$ 需要随机初始化以避免对称性问题。初始化范围取决于激活函数（此处为 tanh）和输入维度。

class Model:
    def __init__(self, word_dim, hidden_dim=100, bptt_truncate=4):
        self.word_dim = word_dim
        self.hidden_dim = hidden_dim
        self.bptt_truncate = bptt_truncate
        # 随机初始化权重
        self.U = np.random.uniform(-np.sqrt(1. / word_dim), np.sqrt(1. / word_dim), (hidden_dim, word_dim))
        self.W = np.random.uniform(-np.sqrt(1. / hidden_dim), np.sqrt(1. / hidden_dim), (hidden_dim, hidden_dim))
        self.V = np.random.uniform(-np.sqrt(1. / hidden_dim), np.sqrt(1. / hidden_dim), (word_dim, hidden_dim))

2. 前向传播与预测

输入为一个序列（例如单词索引列表），模型逐个时间步计算隐藏状态并输出预测概率。

# 假设 x 是输入序列，例如 [0, 179, 341, 416]
# y 是对应的目标序列，例如 [179, 341, 416, 1]

model = Model(word_dim=1000, hidden_dim=100)

# 执行前向传播
layers = model.forward_propagation(x)

# 获取预测结果 (返回每个时间步概率最大的词索引)
predictions = model.predict(x)

3. 训练模型 (BPTT + SGD)

使用交叉熵作为损失函数，通过 BPTT 计算梯度，并使用随机梯度下降（SGD）更新参数。

# 定义训练数据 X (输入序列列表) 和 Y (目标序列列表)
# X = [[...], [...], ...], Y = [[...], [...], ...]

learning_rate = 0.005
nepoch = 100

# 开始训练
model.train(X, Y, learning_rate=learning_rate, nepoch=nepoch, evaluate_loss_after=5)

在 train 函数内部，程序会迭代执行以下步骤：

1. 调用 bptt(x, y) 计算梯度 $dU, dW, dV$。
2. 调用 sgd_step 利用学习率更新参数：

   self.U -= learning_rate * dU
   self.V -= learning_rate * dV
   self.W -= learning_rate * dW
   

通过以上步骤，即可完成一个简易 RNN 语言模型的构建与训练，深入理解其内部运作机制。