Unbiased Pairwise Consistency Regularization for SemiSu

    import torch
    from torchvision import datasets
    from torchvision.transforms import ToTensor
    
    train_dataset = datasets.CIFAR100(root='./data', train=True, transform=ToTensor(), download=True)
    labeled_indices, unlabeled_indices = split_labeled_unlabeled(len(train_dataset), labeled_percent=1.0, num_classes=100)
    train_loader = DataLoader(Subset(train_dataset, labeled_indices), batch_size=batch_size, shuffle=True, drop_last=True)
    test_loader = DataLoader(datasets.CIFAR100(root='./data', train=False, transform=ToTensor()), batch_size=batch_size, shuffle=False, drop_last=False)
    
      
      
      
      
      
      
      
    
    代码解读

    class ResNetEncoder(nn.Module):
    def __init__(self):
        super().__init__()
    
        self.resnet = models.resnet18(pretrained=False)
        num_filters = self.resnet.fc.in_features
        self.resnet.fc = nn.Sequential()
    
    def forward(self, x):
        return self.resnet(x)
    
    
    class ProjectionMLP(nn.Module):
    def __init__(self, input_dim, output_dim, hidden_dims):
        super().__init__()
    
        layers = []
        prev_dim = input_dim
    
        for dim in hidden_dims:
            layers.append(nn.Linear(prev_dim, dim))
            layers.append(nn.ReLU())
            prev_dim = dim
    
        layers.append(nn.Linear(prev_dim, output_dim))
    
        self.layers = nn.Sequential(*layers)
    
    def forward(self, x):
        return self.layers(x)
    
    
    class LabelTransformer(nn.Module):
    def __init__(self, encoder, projection_mlps):
        super().__init__()
    
        self.encoder = encoder
        self.projections = nn.ModuleList([ProjectionMLP(**kwargs) for kwargs in projection_mlps])
    
    def forward(self, inputs):
        features = self.encoder(inputs['image'])
    
        logits = [proj(features).squeeze(-1) for proj in self.projections]
    
        pairwise_similarities = compute_pairwise_similarities(logits[0], labels=None, gamma=0., alpha=0., beta=0.)
    
        confidences = softmax(-pairwise_similarities)
    
        outputs = {'logits': logits, 'confidences': confidences}
    
        return outputs
    
    def get_device():
    if torch.cuda.is_available():
        device = 'cuda'
    else:
        device = 'cpu'
    
    return device
    
    device = get_device()
    model = LabelTransformer(ResNetEncoder().to(device), [{'input_dim': encoder_output_dim*width*height, 'hidden_dims': [256]*4, 'output_dim': num_classes}]).to(device)
    optimizer = optim.AdamW(model.parameters(), lr=lr, weight_decay=weight_decay)
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    
    代码解读

    for epoch in range(num_epochs):
    model.train()
    
    loss_epoch = 0.0
    correct = 0.0
    total = 0.0
    progress_bar = tqdm(enumerate(train_loader), desc=f"Epoch {epoch}", total=len(train_loader))
    for batch_idx, data in progress_bar:
        images, targets = data['image'].to(device), data['label'].to(device)
    
        optimizer.zero_grad()
    
        with autocast():
            outputs = model({'image': images}, is_training=True)
    
            pseudo_labels = outputs['pseudo_labels']
    
            loss = criterion(outputs['logits'][0].float(), pseudo_labels.long())
    
        scaler.scale(loss).backward()
        scaler.step(optimizer)
        scaler.update()
    
        _, predicted = torch.max(outputs['logits'], 1)
        total += targets.size(0)
        correct += (predicted == targets).sum().item()
        accuracy = float(correct)/total
    
        loss_epoch += loss.item()
        avg_loss = round(loss_epoch/(batch_idx+1), 4)
    
        progress_bar.set_postfix({"accuracy": f"{accuracy:.4f}", "loss": f"{avg_loss:.4f}"})
    
    scheduler.step()
    # if best_acc < acc:
    #     best_acc = acc
    #     print('Best Acc:%.4f'%best_acc)
    
    print("Train Accuracy:", round((100.*correct)/total, 4))
    print("\n")
    
    # Evaluation on the test set
    model.eval()
    with torch.no_grad():
    correct = 0
    total = 0
    for i, data in enumerate(test_loader):
        img, target = data
        img, target = img.to(device), target.to(device)
        outputs = model({'image': img}, is_training=False)
        predicted = outputs['logits'][0].argmax(1)
        total += img.shape[0]
        correct += int((predicted == target).sum().item())
    
    print("Test Accuracy of the model: {} %".format(100 * correct / total))
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    
    代码解读