Pytorch PyG实现EdgePool图分类

# 导入MNIST数据集
from torch_geometric.datasets import MNISTSuperpixels
# 加载数据、划分训练集和测试集
dataset = MNISTSuperpixels(root='./mnist', transform=Compose([ToTensor(), NormalizeMeanStd()]))
data = dataset[0]
# 定义超级参数
num_features = dataset.num_features
num_classes = dataset.num_classes
# 构建训练集和测试集索引文件
train_mask = torch.zeros(data.num_nodes, dtype=torch.uint8)
train_mask[:60000] = 1
test_mask = torch.zeros(data.num_nodes, dtype=torch.uint8)
test_mask[60000:] = 1
# 创建数据加载器
train_loader = DataLoader(data[train_mask], batch_size=32, shuffle=True)
test_loader = DataLoader(data[test_mask], batch_size=32, shuffle=False)

from torch.nn import Linear
from torch_geometric.nn import EdgePooling
class EdgePool(torch.nn.Module):
    def __init__(self, dataset):
        super(EdgePool, self).__init__()
        # 定义输入与输出维度数
        self.input_dim = dataset.num_features
        self.hidden_dim = 128
        self.output_dim = 10
        # 定义卷积层、归一化层和pooling层等
        self.conv1 = GCNConv(self.input_dim, self.hidden_dim)
        self.norm1 = BatchNorm1d(self.hidden_dim)
        self.pool1 = EdgePooling(self.hidden_dim)
        self.conv2 = GCNConv(self.hidden_dim, self.hidden_dim)
        self.norm2 = BatchNorm1d(self.hidden_dim)
        self.pool2 = EdgePooling(self.hidden_dim)
        self.conv3 = GCNConv(self.hidden_dim, self.hidden_dim)
        self.norm3 = BatchNorm1d(self.hidden_dim)
        self.pool3 = EdgePooling(self.hidden_dim)
        self.lin = torch.nn.Linear(self.hidden_dim, self.output_dim)
    def forward(self, x, edge_index, batch):
        x = F.relu(self.norm1(self.conv1(x, edge_index)))
        x, edge_index, _, batch, _ = self.pool1(x, edge_index, None, batch)
        x = F.relu(self.norm2(self.conv2(x, edge_index)))
        x, edge_index, _, batch, _ = self.pool2(x, edge_index, None, batch)
        x = F.relu(self.norm3(self.conv3(x, edge_index)))
        x, edge_index, _, batch, _ = self.pool3(x, edge_index, None, batch)
        x = global_mean_pool(x, batch)
        x = self.lin(x)
        return x

# 定义训练计划，包括损失函数、优化器及迭代次数等
train_epochs = 50
learning_rate = 0.01
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(edge_pool.parameters(), lr=learning_rate)
losses_per_epoch = []
accuracies_per_epoch = []
for epoch in range(train_epochs):
    running_loss = 0.0
    running_corrects = 0.0
    count = 0.0
    for samples in train_loader:
        optimizer.zero_grad()
        x, edge_index, batch = samples.x, samples.edge_index, samples.batch
        out = edge_pool(x, edge_index, batch)
        label = samples.y
        loss = criterion(out, label)
        loss.backward()
        optimizer.step()
        running_loss += loss.item() / len(train_loader.dataset)
        pred = out.argmax(dim=1)
        running_corrects += pred.eq(label).sum().item() / len(train_loader.dataset)
        count += 1
    losses_per_epoch.append(running_loss)
    accuracies_per_epoch.append(running_corrects)
    if (epoch + 1) % 10 == 0:
        print("Train Epoch {}/{} Loss {:.4f} Accuracy {:.4f}".format(
            epoch + 1, train_epochs, running_loss, running_corrects))