import os import pandas as pd import torch import torch.nn as nn from torch_geometric.data import Data, DataLoader from torch_geometric.nn import GCNConv import torch.nn.functional as F from sklearn.model_selection import train_test_split

创建PyG数据集类：

class MyDataset(torch.utils.data.Dataset): def init(self, root, transform=None, pre_transform=None): self.root = root self.edges = pd.read_csv(os.path.join(root, 'edges_P.csv'), header=None) self.transform = transform self.pre_transform = pre_transform

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

def __getitem__(self, idx):
    edge_index = torch.tensor(self.edges.iloc[idx].values, dtype=torch.long).t().contiguous()

    features_path = os.path.join(self.root, 'input', 'images_flatten', f'{idx}.txt')
    labels_path = os.path.join(self.root, 'input', 'labels', f'{idx}.txt')

    features = pd.read_csv(features_path, header=None, delimiter=' ')
    labels = pd.read_csv(labels_path, header=None, delimiter=' ')

    x = torch.tensor(features.values, dtype=torch.float)
    y = torch.tensor(labels.values, dtype=torch.long)

    data = Data(x=x, edge_index=edge_index, y=y)

    if self.transform is not None:
        data = self.transform(data)

    return data

定义GCN模型：

class GCN(torch.nn.Module): def init(self, num_node_features, num_classes): super(GCN, self).init() self.conv1 = GCNConv(num_node_features, 8) self.conv2 = GCNConv(8, 16) self.conv3 = GCNConv(16, num_classes)

def forward(self, data):
    x, edge_index = data.x, data.edge_index
    x = self.conv1(x, edge_index)
    x = F.relu(x)
    x = self.conv2(x, edge_index)
    x = F.relu(x)
    x = F.dropout(x, training=self.training)
    x = self.conv3(x, edge_index)
    return x

创建训练和验证模型：

def train_model(dataset, model, optimizer, device): model.train() total_loss = 0.0

for data in dataset:
    data = data.to(device)
    optimizer.zero_grad()
    output = model(data)
    loss = F.cross_entropy(output, data.y)
    loss.backward()
    optimizer.step()
    total_loss += loss.item()

return total_loss / len(dataset)

def validate_model(dataset, model, device): model.eval() correct = 0 total = 0

for data in dataset:
    data = data.to(device)
    output = model(data)
    _, predicted = torch.max(output, 1)
    total += data.y.size(0)
    correct += (predicted == data.y).sum().item()

return correct / total

加载数据集、创建模型、定义优化器和训练循环，以及验证模型：

if name == 'main': dataset = MyDataset(root="C:/Users/jh/Desktop/data") device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')

model = GCN(num_node_features=1600, num_classes=8).to(device)
optimizer = torch.optim.Adam(model.parameters(), lr=0.01)

train_dataset, val_dataset = train_test_split(dataset, test_size=0.1)
train_loader = DataLoader(train_dataset, batch_size=1, shuffle=False)
val_loader = DataLoader(val_dataset, batch_size=1, shuffle=False)

epochs = 1000
for epoch in range(epochs):
    train_loss = train_model(train_loader, model, optimizer, device)
    print(f'Epoch {epoch+1}/{epochs}, Train Loss: {train_loss:.4f}')
    val_accuracy = validate_model(val_loader, model, device)
    print(f'Val_Acc: {val_accuracy:.4f}')