开放集分类模型：基于 PyTorch 的细粒度分类与未知类别拒绝

以下是一个使用 PyTorch 实现的开放集分类模型的示例代码：

import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
from torch.utils.data import Dataset, DataLoader

# 定义数据集类
class MyDataset(Dataset):
    def __init__(self, csv_file):
        self.data = []
        self.labels = []
        with open(csv_file, 'r') as file:
            lines = file.readlines()
            for line in lines[1:]:
                line = line.strip().split(',')
                features = [float(x) for x in line[:-1]]
                label = int(line[-1])
                self.data.append(features)
                self.labels.append(label)

    def __len__(self):
        return len(self.data)

    def __getitem__(self, idx):
        features = torch.FloatTensor(self.data[idx])
        label = torch.LongTensor([self.labels[idx]])
        return features, label

# 定义模型类
class MyModel(nn.Module):
    def __init__(self):
        super(MyModel, self).__init__()
        self.cnn1 = nn.Conv2d(1, 16, kernel_size=3, stride=1, padding=1)
        self.cnn2 = nn.Conv2d(16, 32, kernel_size=3, stride=1, padding=1)
        self.fc1 = nn.Linear(9 * 9 * 32, 128)
        self.fc2 = nn.Linear(128, 5)

    def forward(self, x):
        x = x.view(-1, 1, 9, 9)  # 将数据重塑为二维矩阵格式
        x = F.relu(self.cnn1(x))
        x = F.relu(self.cnn2(x))
        x = x.view(x.size(0), -1)
        x = F.relu(self.fc1(x))
        x = self.fc2(x)
        return x

# 定义开放集分类器类
class OpenMax(nn.Module):
    def __init__(self, num_classes):
        super(OpenMax, self).__init__()
        self.num_classes = num_classes
        self.means = nn.Parameter(torch.zeros(num_classes))
        self.vars = nn.Parameter(torch.ones(num_classes))

    def forward(self, x):
        dists = torch.cdist(x, self.means.unsqueeze(0))  # 计算马氏距离
        return dists

# 定义训练函数
def train(model, openmax, device, train_loader, optimizer, criterion):
    model.train()
    openmax.train()
    for batch_idx, (data, target) in enumerate(train_loader):
        data, target = data.to(device), target.to(device)
        
        optimizer.zero_grad()
        features = model(data)
        dists = openmax(features)
        
        loss = criterion(dists, target.view(-1))
        loss.backward()
        optimizer.step()

# 定义验证函数
def validate(model, openmax, device, val_loader):
    model.eval()
    openmax.eval()
    correct = 0
    total = 0
    with torch.no_grad():
        for data, target in val_loader:
            data, target = data.to(device), target.to(device)
            features = model(data)
            dists = openmax(features)
            preds = torch.argmin(dists, dim=1)
            correct += (preds == target.view(-1)).sum().item()
            total += target.size(0)
    accuracy = 100.0 * correct / total
    return accuracy

# 主函数
def main():
    # 设置超参数
    batch_size = 64
    learning_rate = 0.001
    num_epochs = 10

    # 创建数据集和数据加载器
    train_dataset = MyDataset('train.csv')
    val_dataset = MyDataset('val.csv')
    test_dataset = MyDataset('test.csv')
    train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
    val_loader = DataLoader(val_dataset, batch_size=batch_size, shuffle=False)
    test_loader = DataLoader(test_dataset, batch_size=batch_size, shuffle=False)

    # 创建模型和优化器
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    model = MyModel().to(device)
    openmax = OpenMax(num_classes=5).to(device)
    optimizer = optim.Adam(list(model.parameters()) + list(openmax.parameters()), lr=learning_rate)
    criterion = nn.CrossEntropyLoss()

    # 训练和验证模型
    for epoch in range(num_epochs):
        train(model, openmax, device, train_loader, optimizer, criterion)
        accuracy = validate(model, openmax, device, val_loader)
        print('Epoch [{}/{}], Accuracy: {:.2f}%'.format(epoch+1, num_epochs, accuracy))

    # 测试模型
    accuracy = validate(model, openmax, device, test_loader)
    print('Test Accuracy: {:.2f}%'.format(accuracy))

if __name__ == '__main__':
    main()

请注意，此代码仅为示例，可能需要根据实际情况进行调整和优化。