使用nn通过拼接Cora矩阵进行监督学习
内容纲要
Nn.py(训练)
import torch
import torch.nn as nn
import torch.optim as optim
from dgl.data import CoraGraphDataset
# 加载CORA数据集
def load_cora_data():
dataset = CoraGraphDataset()
graph = dataset[0]
features = graph.ndata['feat']
labels = graph.ndata['label']
train_mask = graph.ndata['train_mask']
test_mask = graph.ndata['test_mask']
# 将图的邻接矩阵转换为稠密形式
adj_matrix_dense = graph.adjacency_matrix().to_dense()
# 将特征矩阵与邻接矩阵拼接
features_combined = torch.cat([features, adj_matrix_dense], dim=1)
return features_combined, labels, train_mask, test_mask
# 定义神经网络模型
class NeuralNetwork(nn.Module):
def __init__(self, input_dim, hidden_dim, output_dim):
super(NeuralNetwork, self).__init__()
self.fc1 = nn.Linear(input_dim, hidden_dim)
self.relu = nn.ReLU()
self.fc2 = nn.Linear(hidden_dim, output_dim)
def forward(self, x):
x = self.fc1(x)
x = self.relu(x)
x = self.fc2(x)
return x
# 训练和评估模型
def train_and_evaluate(features, labels, train_mask, test_mask, epochs=1000):
input_dim = features.shape[1]
hidden_dim = 255 # 增加隐藏层的大小
output_dim = torch.max(labels) + 1
model = NeuralNetwork(input_dim, hidden_dim, output_dim)
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=0.00033887949895945285)
best_acc = 0
for epoch in range(epochs):
optimizer.zero_grad()
outputs = model(features)
loss = criterion(outputs[train_mask], labels[train_mask])
loss.backward()
optimizer.step()
if epoch % 10 == 0:
model.eval()
with torch.no_grad():
outputs = model(features)
_, predicted = torch.max(outputs[test_mask], 1)
correct = (predicted == labels[test_mask]).sum().item()
total = test_mask.sum().item()
accuracy = 100 * correct / total
print(f'Epoch {epoch}, Loss: {loss.item()}, Accuracy: {accuracy}%')
if accuracy > best_acc:
best_acc = accuracy
else:
print("Early stopping due to no improvement")
break # 早停
model.train()
torch.save(model.state_dict(), 'nn.pth')
# 原来的主程序
if __name__ == "__main__":
features_combined, labels, train_mask, test_mask = load_cora_data()
train_and_evaluate(features_combined, labels, train_mask, test_mask)Test.py
import torch
import torch.nn as nn
import torch.optim as optim
from dgl.data import CoraGraphDataset
import os
# 加载CORA数据集
def load_cora_data():
dataset = CoraGraphDataset()
graph = dataset[0]
features = graph.ndata['feat']
labels = graph.ndata['label']
train_mask = graph.ndata['train_mask']
test_mask = graph.ndata['test_mask']
# 将图的邻接矩阵转换为稠密形式
adj_matrix_dense = graph.adjacency_matrix().to_dense()
# 将特征矩阵与邻接矩阵拼接
features_combined = torch.cat([features, adj_matrix_dense], dim=1)
return features_combined, labels, train_mask, test_mask
# 定义神经网络模型
class NeuralNetwork(nn.Module):
def __init__(self, input_dim, hidden_dim, output_dim):
super(NeuralNetwork, self).__init__()
self.fc1 = nn.Linear(input_dim, hidden_dim)
self.relu = nn.ReLU()
self.fc2 = nn.Linear(hidden_dim, output_dim)
def forward(self, x):
x = self.fc1(x)
x = self.relu(x)
x = self.fc2(x)
return x
# 训练和评估模型
def train_and_evaluate(features, labels, train_mask, test_mask, epochs=1000, model_path='nn.pth'):
input_dim = features.shape[1]
hidden_dim = 255 # 增加隐藏层的大小
output_dim = torch.max(labels).item() + 1
model = NeuralNetwork(input_dim, hidden_dim, output_dim)
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=0.00033887949895945285)
best_acc = 0
for epoch in range(epochs):
optimizer.zero_grad()
outputs = model(features)
loss = criterion(outputs[train_mask], labels[train_mask])
loss.backward()
optimizer.step()
if epoch % 10 == 0:
model.eval()
with torch.no_grad():
outputs = model(features)
_, predicted = torch.max(outputs[test_mask], 1)
correct = (predicted == labels[test_mask]).sum().item()
total = test_mask.sum().item()
accuracy = 100 * correct / total
print(f'Epoch {epoch}, Loss: {loss.item()}, Accuracy: {accuracy}%')
if accuracy > best_acc:
best_acc = accuracy
else:
print("Early stopping due to no improvement")
break # 早停
torch.save(model.state_dict(), model_path)
print(f'Model saved to {model_path}')
# 加载模型进行推理
def load_model_for_inference(input_dim, hidden_dim, output_dim, model_path='nn.pth'):
model = NeuralNetwork(input_dim, hidden_dim, output_dim)
model.load_state_dict(torch.load(model_path))
model.eval()
return model
def inference(model, features, mask):
with torch.no_grad():
outputs = model(features)
_, predicted = torch.max(outputs[mask], 1)
return predicted
if __name__ == "__main__":
features_combined, labels, train_mask, test_mask = load_cora_data()
model_path = 'nn.pth'
# 检查模型文件是否存在，不存在则训练模型
if not os.path.exists(model_path):
print("Model not found, training...")
train_and_evaluate(features_combined, labels, train_mask, test_mask, model_path=model_path)
input_dim = features_combined.shape[1]
hidden_dim = 255
output_dim = torch.max(labels).item() + 1
# 加载模型进行推理
model = load_model_for_inference(input_dim, hidden_dim, output_dim, model_path)
predicted_labels = inference(model, features_combined, test_mask)
# 计算测试集上的准确率
correct = (predicted_labels == labels[test_mask]).sum().item()
total = test_mask.sum().item()
accuracy = 100 * correct / total
print(f'Test Accuracy: {accuracy}%')Inn.py(超参优化)
import torch
import optuna
import torch.nn as nn
import torch.optim as optim
from dgl.data import CoraGraphDataset
import subprocess
# 加载CORA数据集
def load_cora_data():
dataset = CoraGraphDataset()
graph = dataset[0]
features = graph.ndata['feat']
labels = graph.ndata['label']
train_mask = graph.ndata['train_mask']
test_mask = graph.ndata['test_mask']
# 将图的邻接矩阵转换为稠密形式
adj_matrix_dense = graph.adjacency_matrix().to_dense()
# 将特征矩阵与邻接矩阵拼接
features_combined = torch.cat([features, adj_matrix_dense], dim=1)
return features_combined, labels, train_mask, test_mask
# 定义神经网络模型
class NeuralNetwork(nn.Module):
def __init__(self, input_dim, hidden_dim, output_dim):
super(NeuralNetwork, self).__init__()
self.fc1 = nn.Linear(input_dim, hidden_dim)
self.relu = nn.ReLU()
self.fc2 = nn.Linear(hidden_dim, output_dim)
def forward(self, x):
x = self.fc1(x)
x = self.relu(x)
x = self.fc2(x)
return x
# 训练和评估模型
# def train_and_evaluate(features, labels, train_mask, test_mask, epochs=100):
# input_dim = features.shape[1]
# hidden_dim = 200 # 增加隐藏层的大小
# output_dim = torch.max(labels) + 1
#
# model = NeuralNetwork(input_dim, hidden_dim, output_dim)
# criterion = nn.CrossEntropyLoss()
# optimizer = optim.Adam(model.parameters(), lr=0.031)
#
# best_acc = 0
# for epoch in range(epochs):
# optimizer.zero_grad()
# outputs = model(features)
# loss = criterion(outputs[train_mask], labels[train_mask])
# loss.backward()
# optimizer.step()
#
# if epoch % 10 == 0:
# model.eval()
# with torch.no_grad():
# outputs = model(features)
# _, predicted = torch.max(outputs[test_mask], 1)
# correct = (predicted == labels[test_mask]).sum().item()
# total = test_mask.sum().item()
# accuracy = 100 * correct / total
# print(f'Epoch {epoch}, Loss: {loss.item()}, Accuracy: {accuracy}%')
# if accuracy > best_acc:
# best_acc = accuracy
# else:
# print("Early stopping due to no improvement")
# break # 早停
# model.train()
# 修改后的训练和评估函数
def train_and_evaluate_with_optuna(trial, features, labels, train_mask, test_mask, epochs=100):
input_dim = features.shape[1]
hidden_dim = trial.suggest_int('hidden_dim', 50, 400) # 搜索隐藏层维度的范围
output_dim = torch.max(labels) + 1
lr = trial.suggest_float('lr', 1e-5, 1e-1, log=True) # 搜索学习率的范围
model = NeuralNetwork(input_dim, hidden_dim, output_dim)
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=lr)
for epoch in range(epochs):
optimizer.zero_grad()
outputs = model(features)
loss = criterion(outputs[train_mask], labels[train_mask])
loss.backward()
optimizer.step()
model.eval()
with torch.no_grad():
outputs = model(features)
_, predicted = torch.max(outputs[test_mask], 1)
correct = (predicted == labels[test_mask]).sum().item()
accuracy = correct / test_mask.sum().item()
return accuracy # 返回准确率用于Optuna优化
# 定义优化函数
def objective(trial):
features_combined, labels, train_mask, test_mask = load_cora_data()
accuracy = train_and_evaluate_with_optuna(trial, features_combined, labels, train_mask, test_mask)
return accuracy
# 创建study对象并运行优化
if __name__ == "__main__":
study_name = "cora_study" # 给你的study命名
storage_name = f"sqlite:///{study_name}.db"
# 创建study对象，指定优化的方向为最大化，并指定存储位置
study = optuna.create_study(direction='maximize', study_name=study_name, storage=storage_name, load_if_exists=True)
# 启动优化过程
study.optimize(objective, n_trials=100, show_progress_bar=True)
# 打印出最佳试验的结果
print("Best trial:")
trial = study.best_trial
print(" Value: ", trial.value)
print(" Params: ")
for key, value in trial.params.items():
print(f" {key}: {value}")
# 通过命令行启动optuna-dashboard
subprocess.run(["optuna-dashboard", storage_name])
# 原来的主程序
# if __name__ == "__main__":
# features_combined, labels, train_mask, test_mask = load_cora_data()
# train_and_evaluate(features_combined, labels, train_mask, test_mask)