04-0.3 到 0.4——如何选对深度学习模型？

model = keras.Sequential([
    keras.Input(shape=(28, 28, 1)),  # 输入层（图像尺寸）
    keras.layers.Conv2D(32, (3, 3), activation='relu'),  # 卷积层
    keras.layers.MaxPooling2D((2, 2)),  # 池化层
    keras.layers.Flatten(),  # 展平层
    keras.layers.Dense(128, activation='relu'),  # 全连接层
    keras.layers.Dense(10, activation='softmax')  # 输出层
])

# 训练模型
model.compile(
    optimizer='adam',
    loss='sparse_categorical_crossentropy',
    metrics=['accuracy']
)

model.fit(
    train_images.reshape(-1, 28, 28, 1), train_labels,
    epochs=5, batch_size=32,
    validation_data=(val_images.reshape(-1, 28, 28, 1), val_labels)
)

import torch
from torch.utils.data import Dataset, DataLoader
from torch.optim import AdamW
from transformers import BertTokenizer, BertForSequenceClassification, get_linear_schedule_with_warmup
from datasets import load_dataset
import numpy as np
from sklearn.metrics import accuracy_score, f1_score

# 自定义数据集类：将 IMDB 数据集转换为 PyTorch 可用的格式
class IMDBDataset(Dataset):
    """
    自定义数据集类，用于处理IMDB电影评论数据

    参数:
    dataset: Hugging Face数据集对象
    tokenizer: BERT 分词器
    max_length: 序列最大长度
    """
    def __init__(self, dataset, tokenizer, max_length):
        self.dataset = dataset          # 原始数据集
        self.tokenizer = tokenizer      # BERT分词器
        self.max_length = max_length    # 序列最大长度

    def __len__(self):
        """返回数据集中样本的数量"""
        return len(self.dataset)

    def __getitem__(self, idx):
        """获取指定索引的数据样本并进行预处理"""
        # 获取文本和标签
        text = self.dataset[idx]['text']
        label = self.dataset[idx]['label']

        # 使用BERT分词器对文本进行编码
        encoding = self.tokenizer.encode_plus(
            text,
            add_special_tokens=True,      # 添加[CLS]和[SEP]特殊标记
            max_length=self.max_length,   # 最大序列长度
            padding='max_length',         # 填充到最大长度
            truncation=True,              # 截断超长序列
            return_tensors='pt'           # 返回PyTorch张量
        )

        # 返回模型所需的输入格式
        return {
            'input_ids': encoding['input_ids'].flatten(),              # 输入ID
            'attention_mask': encoding['attention_mask'].flatten(),    # 注意力掩码
            'labels': torch.tensor(label, dtype=torch.long)            # 标签
        }

print("1. 数据准备阶段")
print("正在加载 IMDB 数据集...")

# 从 Hugging Face 加载 IMDB 电影评论数据集
imdb_dataset = load_dataset('imdb')
print("成功加载 IMDB 数据集")

# 将训练集分割为训练集和验证集（80%-20%分割）
print("划分训练集和验证集...")
train_val_split = imdb_dataset['train'].train_test_split(test_size=0.2, seed=42)
train_data = train_val_split['train']
# train_test_split 验证集被命名为 'test'
val_data = train_val_split['test']
print(f"训练集样本数: {len(train_data)}, 验证集样本数: {len(val_data)}")

# 2. 模型初始化阶段
print("2. 模型初始化阶段")
print("加载 BERT 模型和分词器...")

# 加载预训练的BERT分词器和模型
tokenizer = BertTokenizer.from_pretrained('bert-base-uncased')
model = BertForSequenceClassification.from_pretrained('bert-base-uncased', num_labels=2)
print("成功加载 BERT 模型和分词器")

# 设置序列最大长度并创建数据集对象
max_length = 128
print(f"创建数据集对象，序列最大长度: {max_length}")
train_dataset = IMDBDataset(train_data, tokenizer, max_length)
val_dataset = IMDBDataset(val_data, tokenizer, max_length)
test_dataset = IMDBDataset(imdb_dataset['test'], tokenizer, max_length)

# 创建数据加载器
batch_size = 16
print(f"创建数据加载器，批次大小: {batch_size}")
train_dataloader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
val_dataloader = DataLoader(val_dataset, batch_size=batch_size, shuffle=False)
test_dataloader = DataLoader(test_dataset, batch_size=batch_size, shuffle=False)

# 设置优化器和学习率调度器
print("配置优化器和学习率调度器")
learning_rate = 2e-5
optimizer = AdamW(model.parameters(), lr=learning_rate)

# 计算总训练步数
num_epochs = 3
total_steps = len(train_dataloader) * num_epochs

# 创建学习率调度器，在训练过程中逐步降低学习率
scheduler = get_linear_schedule_with_warmup(
    optimizer, 
    num_warmup_steps=0, 
    num_training_steps=total_steps
)

# 设置运行设备(GPU/CPU)
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
print(f"运行设备: {device}")
model.to(device)  # 将模型移动到指定设备

# 3. 模型训练阶段
print("\n3. 模型训练阶段")
print(f"开始训练，总周期数: {num_epochs}")

best_val_accuracy = 0
for epoch in range(num_epochs):
    print(f"\n开始第 {epoch + 1}/{num_epochs} 个训练周期")

    # 训练模式
    model.train()
    total_train_loss = 0
    print("训练中...")
    
    for batch in train_dataloader:
        # 将数据移动到指定设备
        input_ids = batch['input_ids'].to(device)
        attention_mask = batch['attention_mask'].to(device)
        labels = batch['labels'].to(device)

        # 清除之前的梯度
        optimizer.zero_grad()

        # 前向传播
        outputs = model(input_ids, attention_mask=attention_mask, labels=labels)
        loss = outputs.loss
        total_train_loss += loss.item()

        # 反向传播
        loss.backward()

        # 更新参数
        optimizer.step()
        scheduler.step()  # 更新学习率

    # 计算平均训练损失
    avg_train_loss = total_train_loss / len(train_dataloader)

    # 验证模式
    print("验证中...")
    model.eval()  # 设置为评估模式
    val_predictions = []
    val_true_labels = []
    total_val_loss = 0
    
    with torch.no_grad():  # 不计算梯度
        for batch in val_dataloader:
            # 将数据移动到指定设备
            input_ids = batch['input_ids'].to(device)
            attention_mask = batch['attention_mask'].to(device)
            labels = batch['labels'].to(device)

            # 前向传播
            outputs = model(input_ids, attention_mask=attention_mask, labels=labels)
            loss = outputs.loss
            total_val_loss += loss.item()

            # 获取预测结果
            logits = outputs.logits
            preds = torch.argmax(logits, dim=1)

            # 收集预测结果和真实标签
            val_predictions.extend(preds.cpu().tolist())
            val_true_labels.extend(labels.cpu().tolist())

    # 计算验证指标
    avg_val_loss = total_val_loss / len(val_dataloader)
    val_accuracy = accuracy_score(val_true_labels, val_predictions)
    print(f'第 {epoch + 1} 个周期: 训练损失 = {avg_train_loss:.4f}, 验证损失 = {avg_val_loss:.4f}, 验证准确率 = {val_accuracy:.4f}')

    # 保存最佳模型
    if val_accuracy > best_val_accuracy:
        best_val_accuracy = val_accuracy
        print(f"发现更好的模型，保存模型权重 (准确率: {val_accuracy:.4f})")
        torch.save(model.state_dict(), 'best_model.pth')

第 1 个周期: 训练损失 = 0.3398, 验证损失 = 0.2854, 验证准确率 = 0.8792发现更好的模型，保存模型权重 (准确率: 0.8792)
第 2 个周期: 训练损失 = 0.1799, 验证损失 = 0.2949, 验证准确率 = 0.8864发现更好的模型，保存模型权重 (准确率: 0.8864)
第 3 个周期: 训练损失 = 0.0822, 验证损失 = 0.3687, 验证准确率 = 0.8888发现更好的模型，保存模型权重 (准确率: 0.8888)

# 4. 模型评估阶段
print("\n4. 模型评估阶段")

# 加载训练过程中保存的最佳模型
print("加载最佳模型权重...")
model.load_state_dict(torch.load('best_model.pth'))

# 在测试集上评估模型
print("在测试集上评估模型...")
model.eval()  # 设置为评估模式
predictions = []
true_labels = []

with torch.no_grad():  # 不计算梯度
    for batch in test_dataloader:
        # 将数据移动到指定设备
        input_ids = batch['input_ids'].to(device)
        attention_mask = batch['attention_mask'].to(device)
        labels = batch['labels'].to(device)

        # 前向传播
        outputs = model(input_ids, attention_mask=attention_mask)
        logits = outputs.logits
        preds = torch.argmax(logits, dim=1)

        # 收集预测结果和真实标签
        predictions.extend(preds.cpu().tolist())
        true_labels.extend(labels.cpu().tolist())

# 计算评估指标
accuracy = accuracy_score(true_labels, predictions)
f1 = f1_score(true_labels, predictions)
print(f'测试集准确率: {accuracy:.4f}')
print(f'测试集F1分数: {f1:.4f}')

# 定义情感预测函数：用于对新文本进行情感分析
def predict_sentiment(text, tokenizer, model, max_length=128, device='cpu'):
    """
    使用训练好的BERT模型对文本进行情感分析

    参数:
    text: 要分析的文本字符串
    tokenizer: BERT分词器
    model: 训练好的BERT模型
    max_length: 序列最大长度
    device: 运行设备('cpu'或'cuda')

    返回:
    包含情感预测结果的字典
    """
    # 预处理文本
    encoding = tokenizer.encode_plus(
        text,
        add_special_tokens=True,
        max_length=max_length,
        padding='max_length',
        truncation=True,
        return_tensors='pt'
    )
    
    # 将张量移动到指定设备
    input_ids = encoding['input_ids'].to(device)
    attention_mask = encoding['attention_mask'].to(device)

    # 预测阶段
    model.eval()  # 设置为评估模式
    with torch.no_grad():  # 不计算梯度，节省内存
        outputs = model(input_ids, attention_mask=attention_mask)
        logits = outputs.logits
        # 转换为概率值
        prob = torch.softmax(logits, dim=1).squeeze().cpu().numpy()

    # 解析结果
    negative_prob = prob[0]  # 负面情感概率
    positive_prob = prob[1]  # 正面情感概率
    sentiment = 'positive' if positive_prob > 0.5 else 'negative'  # 情感判断

    # 返回预测结果
    return {
        'text': text,
        'sentiment': sentiment,
        'negative_prob': f'{negative_prob:.4f}',
        'positive_prob': f'{positive_prob:.4f}'
    }

# 5. 自定义文本测试
print("\n5. 自定义文本测试")
print("对自定义文本进行情感分析...")

custom_text = "This film is a masterpiece! The cinematography and soundtrack are unparalleled."
print(f"输入文本: {custom_text}")
result = predict_sentiment(custom_text, tokenizer, model, device=device)
print(f"预测结果: {result['sentiment']} (积极概率: {result['positive_prob']}, 消极概率: {result['negative_prob']})")

测试集准确率: 0.8857测试集F1分数: 0.8842
5. 自定义文本测试对自定义文本进行情感分析...输入文本: This film is a masterpiece! The cinematography and soundtrack are unparalleled.预测结果: positive (积极概率: 0.9983, 消极概率: 0.0017)

# 程序完成
print("\n程序执行完毕!")