自监督学习 (Self-Supervised Learning) 算法详解与PyTorch

    pip install torch torchvision numpy pandas scikit-learn matplotlib

    import torch
    import torch.nn as nn
    import torch.optim as optim
    from torch.utils.data import DataLoader
    from torchvision import datasets, transforms
    import numpy as np
    import matplotlib.pyplot as plt
    import os
    
    # 创建model目录
    os.makedirs('model', exist_ok=True)
    
    class SelfSupervisedModel(nn.Module):
    def __init__(self, input_dim, hidden_dim, output_dim):
        super(SelfSupervisedModel, self).__init__()
        self.encoder = nn.Sequential(
            nn.Linear(input_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim),
            nn.ReLU()
        )
        self.predictor = nn.Linear(hidden_dim, output_dim)
    
    def forward(self, x):
        x = self.encoder(x)
        x = self.predictor(x)
        return x
    
    def train_self_supervised(model, train_loader, optimizer, device, n_epochs=100):
    model.train()
    losses, accuracies = [], []
    for epoch in range(n_epochs):
        epoch_loss, correct = 0, 0
        for inputs, targets in train_loader:
            inputs, targets = inputs.to(device), targets.to(device)
            optimizer.zero_grad()
            outputs = model(inputs)
            loss = nn.functional.cross_entropy(outputs, targets)
            loss.backward()
            optimizer.step()
            epoch_loss += loss.item()
            correct += (outputs.argmax(dim=1) == targets).sum().item()
        losses.append(epoch_loss / len(train_loader))
        accuracies.append(correct / len(train_loader.dataset))
        if (epoch + 1) % 10 == 0:
            print(f'Epoch {epoch+1}/{n_epochs}, Loss: {epoch_loss/len(train_loader):.4f}, Accuracy: {correct/len(train_loader.dataset):.4f}')
    return losses, accuracies
    
    def plot_metrics(losses, accuracies):
    plt.figure(figsize=(12, 5))
    plt.subplot(1, 2, 1)
    plt.plot(losses)
    plt.xlabel('Epoch')
    plt.ylabel('Loss')
    plt.title('Training Loss')
    plt.subplot(1, 2, 2)
    plt.plot(accuracies)
    plt.xlabel('Epoch')
    plt.ylabel('Accuracy')
    plt.title('Training Accuracy')
    plt.show()

    # 数据预处理
    transform = transforms.Compose([
    transforms.ToTensor(),
    transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))
    ])
    
    # 加载数据
    train_data = datasets.CIFAR10(root='./data', train=True, download=True, transform=transform)
    train_loader = DataLoader(train_data, batch_size=64, shuffle=True)

    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    
    # 初始化模型
    input_dim = 32 * 32 
    hidden_dim = 256
    output_dim = 10
    model = SelfSupervisedModel(input_dim, hidden_dim, output_dim).to(device)
    
    # 定义优化器
    optimizer = optim.Adam(model.parameters(), lr=0.001)
    
    # 训练模型
    losses, accuracies = train_self_supervised(model, train_loader, optimizer, device, n_epochs=100)
    
    # 绘制损失图和正确率图
    plot_metrics(losses, accuracies)
    
    # 保存模型
    torch.save(model.state_dict(), 'model/cifar10_self_supervised.pth')

    from torchtext.datasets import AG_NEWS
    from torchtext.data.utils import get_tokenizer
    from torchtext.vocab import build_vocab_from_iterator
    
    # 数据预处理
    tokenizer = get_tokenizer('basic_english')
    train_data = AG_NEWS(root='./data', split='train')
    
    def yield_tokens(data_iter):
    for _, text in data_iter:
        yield tokenizer(text)
    
    vocab = build_vocab_from_iterator(yield_tokens(train_data), specials=['<unk>'])
    vocab.set_default_index(vocab['<unk>'])
    
    def text_pipeline(text):
    return vocab(tokenizer(text))
    
    def collate_batch(batch):
    label_list, text_list = [], []
    for (_label, _text) in batch:
        label_list.append(_label)
        processed_text = torch.tensor(text_pipeline(_text), dtype=torch.int64)
        text_list.append(processed_text)
    return torch.tensor(label_list, dtype=torch.int64), torch.nn.utils.rnn.pad_sequence(text_list, batch_first=True)
    
    train_loader = DataLoader(train_data, batch_size=64, shuffle=True, collate_fn=collate_batch)

    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    
    # 初始化模型
    input_dim = len(vocab)
    hidden_dim = 256
    output_dim = 4
    model = SelfSupervisedModel(input_dim, hidden_dim, output_dim).to(device)
    
    # 定义优化器
    optimizer = optim.Adam(model.parameters(), lr=0.001)
    
    # 训练模型
    losses, accuracies = train_self_supervised(model, train_loader, optimizer, device, n_epochs=100)
    
    # 绘制损失图和正确率图
    plot_metrics(losses, accuracies)
    
    # 保存模型
    torch.save(model.state_dict(), 'model/ag_news_self_supervised.pth')