大模型基础--情感分析任务的演进（深度学习篇）

class ReviewAnalysisModel(nn.Module):
    def __init__(self):
        super().__init__()
        # padding_idx的作用是对于矩阵中填充词不进行梯度计算
        # 加载word2vec模型
        self.word2vec_wv = KeyedVectors.load_word2vec_format(MODEL_DIR / WORD2VEC_MODEL)
        # 获取word和索引的映射
        self.word2id = self.word2vec_wv.key_to_index
        # 嵌入word2vec的向量
        self.embedding = nn.Embedding.from_pretrained(embeddings=torch.tensor(self.word2vec_wv.vectors), freeze=True,
                                                      padding_idx=self.word2id[UNK_TOKEN])
        self.flatten = nn.Flatten()
        self.bn = nn.BatchNorm1d(SEQ_LEN*EMBEDDING_SIZE)
        self.dropout1 = nn.Dropout(0.1)
        self.linear1 = nn.Linear(SEQ_LEN*EMBEDDING_SIZE, 1024)
        self.relu1 = nn.ReLU()
        self.linear2 = nn.Linear(1024, 256)
        self.relu2 = nn.ReLU()
        self.dropout2 = nn.Dropout(0.5)
        self.linear3 = nn.Linear(256, 1)
        self.sigmoid = nn.Sigmoid()

    def forward(self, X):
        embed = self.embedding(X)
        flatten = self.flatten(embed)
        bn = self.bn(flatten)
        dropout1 = self.dropout1(bn)
        linear1 = self.relu1(self.linear1(dropout1))
        linear2 = self.relu2(self.linear2(linear1))
        dropout2 = self.dropout2(linear2)
        linear3 = self.linear3(dropout2)
        return linear3.squeeze(dim=-1)

import time

import torch
from torch import nn, optim
from torch.utils.tensorboard import SummaryWriter
from tqdm import tqdm

from nlp_tutorial.ch03_tradtional_model.dl_project.src.b_dataset import get_dataloader
from nlp_tutorial.ch03_tradtional_model.dl_project.src.c_model import ReviewAnalysisModel
from nlp_tutorial.ch03_tradtional_model.dl_project.src.d_config import *

# 训练一个epoch
def train_one_epoch(model, train_loader, loss, optimizer, device):
    model.train()
    total_loss = 0.0
    for input, target in tqdm(train_loader, desc='训练: '):
        input, target = input.to(device), target.to(device)
        output = model(input)
        loss_value = loss(output, target)
        loss_value.backward()
        optimizer.step()
        optimizer.zero_grad()
        total_loss += loss_value.item()
    return total_loss / len(train_loader)

# 训练的基础配置
def train():
    device = 'cuda' if torch.cuda.is_available() else 'cpu'
    # 获取数据
    train_loader = get_dataloader()
    # 加载模型到指定的计算资源
    model = ReviewAnalysisModel().to(device)
    # 定义损失类型
    loss = nn.BCEWithLogitsLoss()
    # 定义梯度优化器
    optimizer = optim.Adam(lr=LEARN_RATE, params=model.parameters())
    # 友好显示训练过程
    min_loss = float('inf')
    with SummaryWriter(log_dir=LOG_DIR / time.strftime('%Y-%m-%d_%H-%M-%S')) as writer:
        for epoch in range(EPOCHS):
            print("=" * 10, f"EPOCH:{epoch + 1}", "=" * 10)
            time.sleep(0.1)
            this_loss = train_one_epoch(model=model, train_loader=train_loader, loss=loss, optimizer=optimizer,
                                        device=device)
            print("this loss: ", this_loss)
            writer.add_scalar('loss', this_loss, epoch + 1)

            if this_loss < min_loss:
                min_loss = this_loss
                torch.save(model.state_dict(), MODEL_DIR / BEST_MODEL)
                print("模型保存成功!")


if __name__ == '__main__':
    train()

import torch
from tqdm import tqdm

from nlp_tutorial.ch03_tradtional_model.dl_project.src.b_dataset import get_dataloader
from nlp_tutorial.ch03_tradtional_model.dl_project.src.c_model import ReviewAnalysisModel
from nlp_tutorial.ch03_tradtional_model.dl_project.src.d_config import *

def predict_batch(model, input):
    model.eval()
    with torch.no_grad():
        output = model(input)
        output = torch.sigmoid(output)
    return output.tolist()

def evaluate(model, dataloader, device):
    correct_count = 0.0
    total_count = 0.0
    for inputs, targets in tqdm(dataloader, "评估模型中: "):
        inputs, targets = inputs.to(device), targets.to(device)
        batch_result = predict_batch(model, inputs)
        for target, result in zip(targets, batch_result):
            total_count += 1
            result = 1 if result > 0.5 else 0
            if result == target:
                correct_count += 1.0
    return correct_count / total_count


def run_evaluate():
    device = 'cuda' if torch.cuda.is_available() else 'cpu'
    # 加载模型到指定的计算资源
    model = ReviewAnalysisModel().to(device)
    # 加载模型
    model.load_state_dict(torch.load(MODEL_DIR / BEST_MODEL))
    print("模型加载成功!")
    test_dataloader = get_dataloader(train=False)
    # 评估
    acc = evaluate(model, test_dataloader, device)
    print("评估结果:")
    print("Accuracy: ", acc)


if __name__ == '__main__':
    run_evaluate()