LSTMによる文書分類

import pandas as pd
import numpy as np
import torch
import re
import nltk
from nltk.tokenize import word_tokenize
nltk.download('punkt')
from sklearn.model_selection import train_test_split
from torch.utils.data import DataLoader, TensorDataset
import torch.nn as nn
import torch.optim as optim
from collections import Counter
from torchtext.vocab import vocab
from sklearn.metrics import accuracy_score, f1_score
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')

df= pd.read_csv('./Data/twitter_training.csv',names=['index','brand','sentiment','text'])
df.head()

df["label"]=df["sentiment"].replace({"Positive":2,"Negative":0,"Neutral":1,"Irrelevant":np.nan})
df.dropna(inplace=True)
df.head()

def preprocess_text(text):
    text = text.lower()  # Lowercasing
    text = re.sub(r'\W+', ' ', text)  # Remove punctuation
    tokens = word_tokenize(text)  # Tokenization
    return tokens

df["processed_text"]=df["text"].apply(preprocess_text)

df.head()

counter = Counter()
for line in df["processed_text"]:
    counter.update(line)
Vocab = vocab(counter, min_freq=1)

# 単語からインデックスへのマッピング
word_to_index = Vocab.get_stoi()

# 最初の5つのアイテムを取得して表示
for i, (word, index) in enumerate(word_to_index.items()):
    if i >= 5:  # 最初の5つのアイテムのみ表示
        break
    print(f"'{word}': {index}")

df['numericalized_text'] = df["processed_text"].apply(lambda x: [Vocab[token] for token in x])

df.head()

def pad_sequences(seq, max_len):
    padded = np.zeros((max_len,), dtype=np.int64)
    if len(seq) > max_len: padded[:] = seq[:max_len]
    else: padded[:len(seq)] = seq
    return padded

df["text_length"]=df["numericalized_text"].apply(lambda x: len(x)) 

df["text_length"].describe()

max_len=30
df['padded_text'] = df['numericalized_text'].apply(lambda x: pad_sequences(x, max_len))

df.head()

# Split the original dataset into training plus validation and testing sets
train_val_df, test_df = train_test_split(df, test_size=0.2)

# Split the training plus validation set into separate training and validation sets
train_df, val_df = train_test_split(train_val_df, test_size=0.25)

# Create TensorDatasets
train_data = TensorDataset(torch.LongTensor(train_df['padded_text'].tolist()), torch.LongTensor(train_df['label'].tolist()))
val_data = TensorDataset(torch.LongTensor(val_df['padded_text'].tolist()), torch.LongTensor(val_df['label'].tolist()))
test_data = TensorDataset(torch.LongTensor(test_df['padded_text'].tolist()), torch.LongTensor(test_df['label'].tolist()))

# Create DataLoaders
batch_size = 32
train_loader = DataLoader(train_data, shuffle=True, batch_size=batch_size)
val_loader = DataLoader(val_data, batch_size=batch_size)
test_loader = DataLoader(test_data, batch_size=batch_size)

embedding = nn.Embedding(num_embeddings=10000, embedding_dim=300)
input = torch.LongTensor([[1, 2, 4, 5], [4, 3, 2, 9]])
embedded = embedding(input)
embedded.shape

class LSTMClassifier(nn.Module):
    def __init__(self, vocab_size, embedding_dim, hidden_dim, output_dim, n_layers, bidirectional, dropout):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embedding_dim)
        self.lstm = nn.LSTM(embedding_dim, 
                            hidden_dim, 
                            num_layers=n_layers, 
                            bidirectional=bidirectional, 
                            dropout=dropout, 
                            batch_first=True)
        self.fc = nn.Linear(hidden_dim * 2 if bidirectional else hidden_dim, output_dim)
        self.bidirectional = bidirectional
    
    def forward(self, text):
        embedded = self.embedding(text)
        output, (hidden, cell) = self.lstm(embedded)
        if self.bidirectional:
            hidden = torch.cat((hidden[-2,:,:], hidden[-1,:,:]), dim = 1)
        else:
            hidden = hidden[-1,:,:]
        return self.fc(hidden.squeeze(0))

vocab_size = len(Vocab)
embedding_dim = 100  
hidden_dim = 256     
output_dim = 3 
n_layers = 2        
bidirectional = True 
dropout = 0.2        

model = LSTMClassifier(vocab_size, embedding_dim, hidden_dim, output_dim, n_layers, bidirectional, dropout)

model = model.to(device)

criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters())

def train_model(model, train_loader, val_loader, optimizer, criterion, n_epochs):
    model.train()
    for epoch in range(n_epochs):
        for texts, labels in train_loader:
            texts, labels = texts.to(device), labels.to(device)
            optimizer.zero_grad()
            predictions = model(texts)
            loss = criterion(predictions, labels)
            loss.backward()
            optimizer.step()

        # Validation
        model.eval()
        with torch.no_grad():
            val_labels = []
            val_preds = []
            for texts, labels in val_loader:
                texts, labels = texts.to(device), labels.to(device)
                predictions = model(texts)
                val_labels.extend(labels.tolist())
                val_preds.extend(torch.argmax(predictions, dim=1).tolist())

            accuracy = accuracy_score(val_labels, val_preds)
            f1 = f1_score(val_labels, val_preds, average='weighted')
            print(f"Epoch {epoch+1}, Loss: {loss.item()}, Accuracy: {accuracy}, F1 Score: {f1}")
        model.train()

from torch.utils.tensorboard import SummaryWriter

def train_model(model, train_loader, val_loader, optimizer, criterion, n_epochs, tensorboard=False, tensorboard_path='./runs'):
    # Initialize TensorBoard writer if tensorboard logging is enabled
    writer = SummaryWriter(tensorboard_path) if tensorboard else None

    model.train()
    for epoch in range(n_epochs):
        for texts, labels in train_loader:
            texts, labels = texts.to(device), labels.to(device)
            optimizer.zero_grad()
            predictions = model(texts)
            loss = criterion(predictions, labels)
            loss.backward()
            optimizer.step()

        # Validation
        model.eval()
        with torch.no_grad():
            val_labels = []
            val_preds = []
            for texts, labels in val_loader:
                texts, labels = texts.to(device), labels.to(device)
                predictions = model(texts)
                val_labels.extend(labels.tolist())
                val_preds.extend(torch.argmax(predictions, dim=1).tolist())

            accuracy = accuracy_score(val_labels, val_preds)
            f1 = f1_score(val_labels, val_preds, average='weighted')

            # Log metrics to TensorBoard
            if tensorboard:
                writer.add_scalar('Loss/train', loss.item(), epoch)
                writer.add_scalar('Accuracy/val', accuracy, epoch)
                writer.add_scalar('F1-Score/val', f1, epoch)

            print(f"Epoch {epoch+1}, Loss: {loss.item()}, Accuracy: {accuracy}, F1 Score: {f1}")

        model.train()

    # Close the TensorBoard writer
    if tensorboard:
        writer.close()

# Train the model
n_epochs = 30
#train_model(model, train_loader, val_loader, optimizer, criterion, n_epochs, tensorboard=True, tensorboard_path='./runs/lstm')