transformers/sentiment_analysis_BiRNN.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import torch\n",
    "from torch import nn\n",
    "import sys\n",
    "sys.path.append('..')\n",
    "from tools import tool_pytorch_017 as ty\n",
    "\n",
    "batch_size = 64\n",
    "train_iter, test_iter, vocab = ty.load_data_imdb(batch_size)\n",
    "\n",
    "class BiRNN(nn.Module):\n",
    "    '''双向长短期记忆网络在初始和最终时间步的隐状态（在最后一层）被连结起来作为文本序列的表示。\n",
    "    然后，通过一个具有两个输出（“积极”和“消极”）的全连接层（self.decoder），将此单一文本表示转换为输出类别。'''\n",
    "    def __init__(self, vocab_size, embed_size, num_hiddens,\n",
    "                 num_layers, **kwargs):\n",
    "        super(BiRNN, self).__init__(**kwargs)\n",
    "        self.embedding = nn.Embedding(vocab_size, embed_size)\n",
    "        # 将bidirectional设置为True以获取双向循环神经网络\n",
    "        self.encoder = nn.LSTM(embed_size, num_hiddens, num_layers=num_layers,\n",
    "                                bidirectional=True)\n",
    "        self.decoder = nn.Linear(4 * num_hiddens, 2)\n",
    "\n",
    "    def forward(self, inputs):\n",
    "        # inputs的形状是（批量大小，时间步数）\n",
    "        # 因为长短期记忆网络要求其输入的第一个维度是时间维，\n",
    "        # 所以在获得词元表示之前，输入会被转置。\n",
    "        # 输出形状为（时间步数，批量大小，词向量维度）\n",
    "        embeddings = self.embedding(inputs.T)\n",
    "        self.encoder.flatten_parameters()\n",
    "        # 返回上一个隐藏层在不同时间步的隐状态，\n",
    "        # outputs的形状是（时间步数，批量大小，2*隐藏单元数）\n",
    "        outputs, _ = self.encoder(embeddings)\n",
    "        # 连结初始和最终时间步的隐状态，作为全连接层的输入，\n",
    "        # 其形状为（批量大小，4*隐藏单元数）\n",
    "        encoding = torch.cat((outputs[0], outputs[-1]), dim=1)\n",
    "        outs = self.decoder(encoding)\n",
    "        return outs\n",
    " \n",
    "def train_batch(net, X, y, loss, trainer, devices):\n",
    "    \"\"\"用多GPU进行小批量训练\"\"\"\n",
    "    if isinstance(X, list):\n",
    "        # 微调BERT中所需（稍后讨论）\n",
    "        X = [x.to(devices[0]) for x in X]\n",
    "    else:\n",
    "        X = X.to(devices[0])\n",
    "    y = y.to(devices[0])\n",
    "    net.train()\n",
    "    trainer.zero_grad()\n",
    "    pred = net(X)\n",
    "    l = loss(pred, y)\n",
    "    l.sum().backward()\n",
    "    trainer.step()\n",
    "    train_loss_sum = l.sum()\n",
    "    train_acc_sum = ty.accuracy(pred, y)\n",
    "    return train_loss_sum, train_acc_sum\n",
    "\n",
    "def train(net, train_iter, test_iter, loss, trainer, num_epochs,\n",
    "               devices=ty.try_all_gpus()):\n",
    "    \"\"\"用多GPU进行模型训练\"\"\"\n",
    "    timer, num_batches = ty.Timer(), len(train_iter)\n",
    "    animator = ty.Animator(xlabel='epoch', xlim=[1, num_epochs], ylim=[0, 1],\n",
    "                            legend=['train loss', 'train acc', 'test acc'])\n",
    "    net = nn.DataParallel(net, device_ids=devices).to(devices[0])\n",
    "    for epoch in range(num_epochs):\n",
    "        # 4个维度：储存训练损失，训练准确度，实例数，特点数\n",
    "        metric = ty.Accumulator(4)\n",
    "        for i, (features, labels) in enumerate(train_iter):\n",
    "            timer.start()\n",
    "            l, acc = train_batch(\n",
    "                net, features, labels, loss, trainer, devices)\n",
    "            metric.add(l, acc, labels.shape[0], labels.numel())\n",
    "            timer.stop()\n",
    "            if (i + 1) % (num_batches // 5) == 0 or i == num_batches - 1:\n",
    "                animator.add(epoch + (i + 1) / num_batches,\n",
    "                             (metric[0] / metric[2], metric[1] / metric[3],\n",
    "                              None))\n",
    "        test_acc = ty.evaluate_accuracy_gpu(net, test_iter)\n",
    "        animator.add(epoch + 1, (None, None, test_acc))\n",
    "    print(f'loss {metric[0] / metric[2]:.3f}, train acc '\n",
    "          f'{metric[1] / metric[3]:.3f}, test acc {test_acc:.3f}')\n",
    "    print(f'{metric[2] * num_epochs / timer.sum():.1f} examples/sec on '\n",
    "          f'{str(devices)}')\n",
    "    \n",
    "# 构造一个具有两个隐藏层的双向循环神经网络来表示单个文本以进行情感分析。   \n",
    "embed_size, num_hiddens, num_layers = 100, 100, 2\n",
    "devices = ty.try_all_gpus()\n",
    "net = BiRNN(len(vocab), embed_size, num_hiddens, num_layers)\n",
    "\n",
    "def init_weights(m):\n",
    "    if type(m) == nn.Linear:\n",
    "        nn.init.xavier_uniform_(m.weight)\n",
    "    if type(m) == nn.LSTM:\n",
    "        for param in m._flat_weights_names:\n",
    "            if \"weight\" in param:\n",
    "                nn.init.xavier_uniform_(m._parameters[param])\n",
    "net.apply(init_weights);\n",
    "\n",
    "glove_embedding = ty.TokenEmbedding('glove.6b.100d')\n",
    "\n",
    "embeds = glove_embedding[vocab.idx_to_token]\n",
    "print(embeds.shape)\n",
    "\n",
    "lr, num_epochs = 0.01, 5\n",
    "trainer = torch.optim.Adam(net.parameters(), lr=lr)\n",
    "loss = nn.CrossEntropyLoss(reduction=\"none\")\n",
    "train(net, train_iter, test_iter, loss, trainer, num_epochs,\n",
    "    devices)\n",
    "\n",
    "#@save\n",
    "def predict_sentiment(net, vocab, sequence):\n",
    "    \"\"\"预测文本序列的情感\"\"\"\n",
    "    sequence = torch.tensor(vocab[sequence.split()], device=ty.try_gpu())\n",
    "    label = torch.argmax(net(sequence.reshape(1, -1)), dim=1)\n",
    "    return 'positive' if label == 1 else 'negative'\n",
    "\n",
    "print(predict_sentiment(net, vocab, 'this movie is so great'))\n",
    "print(predict_sentiment(net, vocab, 'this movie is so bad'))"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "python3.8",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.19"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}