任务二.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import csv\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "\n",
    "import torch\n",
    "import torch.nn as nn\n",
    "import torch.nn.functional as F\n",
    "\n",
    "num_classes = 5\n",
    "num_epochs = 100\n",
    "length = 50\n",
    "embedding_dim = 50\n",
    "feature_maps = 100\n",
    "dropout_rate = 0.5\n",
    "hidden_size = 100\n",
    "batch_size = 50\n",
    "lr = 1e-3\n",
    "device = \"cuda\" if torch.cuda.is_available() else \"cpu\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = pd.read_csv('train.tsv', sep='\\t')[:10000]\n",
    "df = df.reindex(np.random.permutation(df.shape[0]))\n",
    "df_train = df[:int(df.shape[0] * 0.8)]\n",
    "df_test = df[int(df.shape[0] * 0.8):]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "X_train = list()\n",
    "vocab = dict({'<pad>': 0, '<unk>': 1})\n",
    "for phrase in df_train[\"Phrase\"]:\n",
    "    indices = list()\n",
    "    words = phrase.split(' ')\n",
    "    for word in words:\n",
    "        if word not in vocab:\n",
    "            vocab[word] = len(vocab)\n",
    "        indices.append(vocab[word])\n",
    "    while len(indices) < length:\n",
    "        indices.append(0)\n",
    "    X_train.append(indices)\n",
    "X_train = torch.tensor(X_train).to(device)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "Y_train = torch.zeros((df_train.shape[0], num_classes)).to(device)\n",
    "for idx, sentiment in enumerate(df_train[\"Sentiment\"]):\n",
    "    Y_train[idx, sentiment] = 1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "X_test = list()\n",
    "for phrase in df_test[\"Phrase\"]:\n",
    "    indices = list()\n",
    "    words = phrase.split(' ')\n",
    "    for word in words:\n",
    "        if word in vocab:\n",
    "            indices.append(vocab[word])\n",
    "        else:\n",
    "            indices.append(1)\n",
    "    while len(indices) < length:\n",
    "        indices.append(0)\n",
    "    X_test.append(indices)\n",
    "X_test = torch.tensor(X_test).to(device)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "Y_test = torch.zeros((df_test.shape[0], num_classes)).to(device)\n",
    "for idx, sentiment in enumerate(df_test[\"Sentiment\"]):\n",
    "    Y_test[idx, sentiment] = 1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "glove = pd.read_csv(\"glove.6B.50d.txt\", sep=' ', header=None, quoting=csv.QUOTE_NONE)\n",
    "vectors = torch.zeros((len(vocab), embedding_dim)).to(device)\n",
    "torch.nn.init.uniform_(vectors)\n",
    "vectors[0] = torch.zeros_like(vectors[0])\n",
    "for row in glove.iterrows():\n",
    "    if row[1][0] in vocab:\n",
    "        vectors[vocab[row[1][0]]] = torch.tensor(np.array(row[1][1:]).astype(np.float32))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "class CNN(nn.Module):\n",
    "    def __init__(self, num_embeddings, vectors=None):\n",
    "        super().__init__()\n",
    "\n",
    "        if vectors is None:\n",
    "            self.embed = nn.Embedding(num_embeddings, embedding_dim)\n",
    "        else:\n",
    "            self.embed = nn.Embedding(num_embeddings, embedding_dim).from_pretrained(vectors)\n",
    "\n",
    "        self.conv1 = nn.Conv2d(1, feature_maps, (3, embedding_dim))\n",
    "        self.conv2 = nn.Conv2d(1, feature_maps, (4, embedding_dim))\n",
    "        self.conv3 = nn.Conv2d(1, feature_maps, (5, embedding_dim))\n",
    "        self.dropout = nn.Dropout(dropout_rate)\n",
    "        self.fc = nn.Linear(feature_maps * 3, num_classes)\n",
    "\n",
    "    def forward(self, x):\n",
    "        x = self.embed(x)\n",
    "        x = torch.unsqueeze(x, dim=1)\n",
    "\n",
    "        x1 = torch.squeeze(F.relu(self.conv1(x)))\n",
    "        x1 = torch.squeeze(F.max_pool1d(x1, x1.size(2)))\n",
    "\n",
    "        x2 = torch.squeeze(F.relu(self.conv2(x)))\n",
    "        x2 = torch.squeeze(F.max_pool1d(x2, x2.size(2)))\n",
    "\n",
    "        x3 = torch.squeeze(F.relu(self.conv3(x)))\n",
    "        x3 = torch.squeeze(F.max_pool1d(x3, x3.size(2)))\n",
    "\n",
    "        out = torch.cat([x1, x2, x3], dim=1)\n",
    "        out = self.dropout(out)\n",
    "        out = self.fc(out)\n",
    "        return out"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "cnn_random = CNN(len(vocab)).to(device)\n",
    "optimizer = torch.optim.Adam(cnn_random.parameters(), lr=lr)\n",
    "criterion = nn.CrossEntropyLoss()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "cnn_random.train()\n",
    "for epoch in range(num_epochs):\n",
    "    pred = cnn_random(X_train)\n",
    "    loss = criterion(pred, Y_train)\n",
    "    optimizer.zero_grad()\n",
    "    loss.backward()\n",
    "    optimizer.step()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Accuracy of CNN with random initialization: 0.651\n"
     ]
    }
   ],
   "source": [
    "cnn_random.eval()\n",
    "pred = cnn_random(X_test).detach().cpu().numpy()\n",
    "print(\"Accuracy of CNN with random initialization: {}\".format(np.mean(np.argmax(pred, axis=1) == np.array(df_test['Sentiment']))))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "cnn_glove = CNN(len(vocab), vectors).to(device)\n",
    "optimizer = torch.optim.Adam(cnn_glove.parameters(), lr=lr)\n",
    "criterion = nn.CrossEntropyLoss()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "cnn_glove.train()\n",
    "for epoch in range(num_epochs):\n",
    "    for batch in range(df_train.shape[0] // batch_size):\n",
    "        pred = cnn_glove(X_train[batch * batch_size:(batch + 1) * batch_size])\n",
    "        loss = criterion(pred, Y_train[batch * batch_size:(batch + 1) * batch_size])\n",
    "        optimizer.zero_grad()\n",
    "        loss.backward()\n",
    "        optimizer.step()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Accuracy of CNN with GloVe: 0.6865\n"
     ]
    }
   ],
   "source": [
    "cnn_glove.eval()\n",
    "pred = cnn_glove(X_test).detach().cpu().numpy()\n",
    "print(\"Accuracy of CNN with GloVe: {}\".format(np.mean(np.argmax(pred, axis=1) == np.array(df_test['Sentiment']))))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "class RNN(nn.Module):\n",
    "    def __init__(self, num_embeddings, vectors=None):\n",
    "        super().__init__()\n",
    "\n",
    "        if vectors is None:\n",
    "            self.embed = nn.Embedding(num_embeddings, embedding_dim)\n",
    "        else:\n",
    "            self.embed = nn.Embedding(num_embeddings, embedding_dim).from_pretrained(vectors)\n",
    "\n",
    "        self.lstm = nn.LSTM(embedding_dim, hidden_size, batch_first=True)\n",
    "        self.dropout = nn.Dropout(dropout_rate)\n",
    "        self.fc = nn.Linear(feature_maps, num_classes)\n",
    "\n",
    "    def forward(self, x):\n",
    "        h0 = torch.zeros(1, x.size(0), hidden_size).to(device)\n",
    "        c0 = torch.zeros(1, x.size(0), hidden_size).to(device)\n",
    "\n",
    "        x = self.embed(x)\n",
    "        out, _ = self.lstm(x, (h0, c0))\n",
    "        out = self.dropout(out)\n",
    "        out = self.fc(out[:, -1, :])\n",
    "        return out"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "rnn_random = RNN(len(vocab)).to(device)\n",
    "optimizer = torch.optim.Adam(rnn_random.parameters(), lr=lr)\n",
    "criterion = nn.CrossEntropyLoss()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "rnn_random.train()\n",
    "for epoch in range(num_epochs):\n",
    "    pred = rnn_random(X_train)\n",
    "    loss = criterion(pred, Y_train)\n",
    "    optimizer.zero_grad()\n",
    "    loss.backward()\n",
    "    optimizer.step()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Accuracy of RNN with random initialization: 0.5835\n"
     ]
    }
   ],
   "source": [
    "rnn_random.eval()\n",
    "pred = rnn_random(X_test).detach().cpu().numpy()\n",
    "print(\"Accuracy of RNN with random initialization: {}\".format(np.mean(np.argmax(pred, axis=1) == np.array(df_test['Sentiment']))))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "rnn_glove = RNN(len(vocab), vectors).to(device)\n",
    "optimizer = torch.optim.Adam(rnn_glove.parameters(), lr=lr)\n",
    "criterion = nn.CrossEntropyLoss()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "rnn_glove.train()\n",
    "for epoch in range(num_epochs):\n",
    "    for batch in range(df_train.shape[0] // batch_size):\n",
    "        pred = rnn_glove(X_train[batch * batch_size:(batch + 1) * batch_size])\n",
    "        loss = criterion(pred, Y_train[batch * batch_size:(batch + 1) * batch_size])\n",
    "        optimizer.zero_grad()\n",
    "        loss.backward()\n",
    "        optimizer.step()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Accuracy of RNN with GloVe: 0.6715\n"
     ]
    }
   ],
   "source": [
    "rnn_glove.eval()\n",
    "pred = rnn_glove(X_test).detach().cpu().numpy()\n",
    "print(\"Accuracy of RNN with GloVe: {}\".format(np.mean(np.argmax(pred, axis=1) == np.array(df_test['Sentiment']))))"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "base",
   "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.11.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}