Tổng Quan Về Mô Hình

Mô hình Text-RNN (Recurrent Neural Network) là một lựa chọn phổ biến trong bài toán phân lớp văn bản. Trong ngữ cảnh hệ thống hỏi đáp (Question Answering), nhiệm vụ chính là dự đoán ý định (intent) của người dùng dựa trên câu lệnh nhập vào. Sử dụng mạng lưới thần kinh hồi tiếp có khả năng nắm bắt được thông tin ngữ cảnh và thứ tự từ ngữ, giúp cải thiện độ chính xác so với các phương pháp đơn giản như Bag-of-Words.

Cấu Trúc Dữ Liệu Và Tiền Xử Lý

Quy trình khởi tạo dữ liệu bao gồm việc tải tập hợp câu hỏi từ file CSV, tách câu thành các từ riêng lẻ và xây dựng từ điển (vocabulary). Việc sử dụng các vector nhúng đã được huấn luyện trước (pre-trained embeddings) giúp mô hình hiểu sâu hơn về nghĩa của từ mà không cần học từ đầu.

# Cấu hình thiết bị và đường dẫn
import os, sys
sys.path.append(os.getcwd())
import torch
from torch.nn.utils.rnn import pad_sequence
from torchtext.data import Field, Example, Dataset, BucketIterator, Iterator

# Xác định vị trí GPU hoặc CPU
if torch.cuda.is_available():
    device_config = torch.device('cuda')
else:
    device_config = torch.device('cpu')

# Thiết lập các tham số tiền xử lý
def split_tokens(sentence):
    return sentence.strip().split(' ')

# Khai báo trường dữ liệu cho câu và nhãn
sentence_field = Field(
    tokenize=split_tokens,
    lower=True,
    sequential=True,
    batch_first=True,
    unk_token='<unk>',
    pad_token='<pad>')

category_field = Field(
    sequential=False,
    use_vocab=False)

# Hàm hỗ trợ tải tập dữ liệu
def load_corpus(data_df, is_eval_mode=False):
    fields = [('id', None), ('input_text', sentence_field), ('target_label', category_field)]
    data_examples = []
    
    if is_eval_mode:
        for txt in data_df['input_text']:
            example = Example.fromlist([None, txt, None], fields)
            data_examples.append(example)
    else:
        for txt, lbl in zip(data_df['input_text'], data_df['target_label']):
            example = Example.fromlist([None, txt, lbl], fields)
            data_examples.append(example)
            
    return data_examples, fields

# Đọc dữ liệu gốc
import pandas as pd
csv_file_path = './classification_data.csv'
raw_data = pd.read_csv(csv_file_path)
examples_set, current_fields = load_corpus(raw_data)
base_dataset = Dataset(examples_set, current_fields)

Xây Dựng Từ Điển và Nhúng

Sau khi dữ liệu được chuẩn hóa, bước tiếp theo là xây dựng từ vựng dựa trên tần suất xuất hiện tối thiểu. Các vector từ ngữ sẽ được tải từ nguồn bên ngoài (ví dụ: Sogou Word Vector) để khởi tạo trọng số của tầng Embedding.

# Tạo thư mục lưu trữ nếu chưa tồn tại
os.makedirs('output_models', exist_ok=True)

# Build vocab với từ khóa nhúng
sentence_field.build_vocab(base_dataset, min_freq=1, vectors='sgns.sogou.char')

# Lưu danh sách từ khóa
with open('words_list.pkl', 'wb') as f_out:
    pickle.dump(sentence_field.vocab, f_out)

# Thiết lập bộ lặp (Iterator) cho quá trình huấn luyện
BATCH_SIZE_SETTING = 163
train_loader = BucketIterator(
    dataset=base_dataset, 
    batch_size=BATCH_SIZE_SETTING, 
    train=True,
    sort_within_batch=False)

# Gán vector pre-trained vào mô hình (có thể Fine-tune)
emb_vectors = sentence_field.vocab.vectors
em_dim = emb_vectors.shape[1]

Định Nghĩa Lớp Mô Hình

Mô hình kế thừa từ `nn.Module`, bao gồm tầng nhúng từ, tầng LSTM hai chiều (Bidirectional) để trích xuất đặc trưng chuỗi, và tầng Dense để phân lớp đầu ra.

import torch.nn as nn
import torch.nn.functional as F

class IntentClassifier(nn.Module):
    def __init__(self, voc_size, embed_dim, hidden_dim, num_lay, out_class, dr_rate=0.5):
        super(IntentClassifier, self).__init__()
        
        # Tầng Embedding
        self.emb_layer = nn.Embedding(voc_size, embed_dim)
        # Tầng LSTM Bidirectional
        self.rnn_layer = nn.LSTM(
            input_size=embed_dim,
            hidden_size=hidden_dim,
            num_layers=num_lay,
            bidirectional=True,
            dropout=dr_rate,
            batch_first=True
        )
        # Tầng Fully Connected đầu ra
        # Hidden size nhân 2 do là Bidirectional
        self.out_fc = nn.Linear(hidden_dim * 2, out_class)
        
    def forward(self, x_input):
        # Kích thước đầu vào: (Batch, Seq_Len)
        # Chuyển sang (Batch, Seq_Len, Embed_Dim)
        vec_emb = self.emb_layer(x_input) 
        
        # LSTM trả về output, h_n, c_n
        _, (final_h, _) = self.rnn_layer(vec_emb)
        
        # Lấy trạng thái ẩn cuối cùng của cả hai chiều
        # concat last hidden states of all layers
        out_features = final_h[-2, :, :] + final_h[-1, :, :] # Tổng hợp state
        
        # Chuyền qua tuyến tính
        logits = self.out_fc(out_features)
        return logits

Vòng Lặp Huấn Luyện

Quá trình huấn luyện sử dụng thuật toán SGD kết hợp động lượng để cập nhật trọng số. TensorBoard được kích hoạt để giám sát chỉ số suy giảm mất mát (Loss) theo thời gian thực.

from torch.utils.tensorboard import SummaryWriter

# Khởi tạo writer để log
tb_logger = SummaryWriter('./logs/textrnn_run')

# Cài đặt siêu tham số huấn luyện
VOC_SIZE = len(sentence_field.vocab)
OUT_DIM = 16
HIDDEN_DIM = 128
NUM_LAYERS = 2
LEARNING_RATE = 0.1
TOTAL_EPOCHS = 600

# Khởi tạo model
model_instance = IntentClassifier(VOC_SIZE, em_dim, HIDDEN_DIM, NUM_LAYERS, OUT_DIM).to(device_config)

# Copy weights pre-trained vào embedding layer
model_instance.emb_layer.weight.data.copy_(emb_vectors)

# Tối ưu hóa hàm mất mát
optimizer_inst = torch.optim.SGD(model_instance.parameters(), lr=LEARNING_RATE, momentum=0.95, nesterov=True)
loss_func = nn.CrossEntropyLoss()

model_instance.train()

# Bắt đầu vòng lặp training
for epoch_idx in range(TOTAL_EPOCHS):
    for i, batch_item in enumerate(train_loader):
        # Di chuyển dữ liệu lên GPU nếu cần
        input_tensor = batch_item.input_text.to(device_config)
        label_tensor = batch_item.target_label.to(device_config)
        
        # Dự đoán và tính Loss
        prediction = model_instance(input_tensor)
        current_loss = loss_func(prediction, label_tensor.long())
        
        # Backpropagation
        optimizer_inst.zero_grad()
        current_loss.backward()
        optimizer_inst.step()
        
        # Log tiến trình mỗi 10 epoch
        if (epoch_idx + 1) % 10 == 0:
            tb_logger.add_scalar('training_loss', current_loss.item(), global_step=epoch_idx + 1)
            print(f"Epoch [{epoch_idx+1}/{TOTAL_EPOCHS}], Loss: {current_loss.item():.4f}")

tb_logger.flush()
tb_logger.close()

# Lưu trọng số mô hình
save_path = './output_models/classifier_model.pth'
torch.save(model_instance.state_dict(), save_path)
print("Hoàn tất lưu mô hình.")

Kết Quả Thử Nghiệm Ban Đầu

Dưới đây là biểu đồ suy giảm Loss trong suốt quá trình chạy thử, cho thấy mô hình dần hội tụ.

Epoch [10/600], Loss: 2.7375
Epoch [20/600], Loss: 2.7167
Epoch [50/600], Loss: 2.7152
Epoch [100/600], Loss: 2.3200
Epoch [200/600], Loss: 1.5200
Epoch [300/600], Loss: 0.3708
Epoch [400/600], Loss: 0.0592
Epoch [500/600], Loss: 0.0102
Epoch [600/600], Loss: 0.0020