@浙大疏锦行
importtorchimporttorch.nnasnnimporttorch.optimasoptimfromsklearn.datasetsimportload_irisfromsklearn.model_selectionimporttrain_test_splitimportnumpyasnp iris=load_iris()X=iris.data y=iris.target X_train,X_test,y_train,y_test=train_test_split(X,y,test_size=0.2,random_state=42)print(X_train.shape)print(y_train.shape)print(X_test.shape)print(y_test.shape)fromsklearn.preprocessingimportMinMaxScaler scaler=MinMaxScaler()X_train=scaler.fit_transform(X_train)X_test=scaler.transform(X_test)X_train=torch.FloatTensor(X_train)y_train=torch.LongTensor(y_train)X_test=torch.FloatTensor(X_test)y_test=torch.LongTensor(y_test)importtorchimporttorch.nnasnnimporttorch.optimclassMLP(nn.Module):def__init__(self,*args,**kwargs):super().__init__(*args,**kwargs)self.fc1=nn.Linear(4,10)self.relu=nn.ReLU()self.fc2=nn.Linear(10,3)defforward(self,x):out=self.fc1(x)out=self.relu(out)out=self.fc2(out)returnout model=MLP()criterion=nn.CrossEntropyLoss()optimizer=torch.optim.SGD(model.parameters(),lr=0.01)num_epochs=20000losses=[]forepochinrange(num_epochs):outputs=model.forward(X_train)loss=criterion(outputs,y_train)# 预测损失# 反向传播和优化optimizer.zero_grad()loss.backward()# 反向传播计算梯度optimizer.step()losses.append(loss.item())if(epoch+1)%100==0:print(f'Epoch[{epoch+1}/{num_epochs}],Loss:{loss.item():.4f}')importmatplotlib.pyplotasplt plt.plot(range(num_epochs),losses)plt.xlabel('Epoch')plt.ylabel('Loss')plt.title('Training Loss over Epochs')plt.show()fromtqdmimporttqdm# 注意：这里导入的是 tqdm 函数，不是模块train_losses=[]test_losses=[]epochs=[]# ===== 新增早停相关参数 =====best_test_loss=float('inf')# 记录最佳测试集损失best_epoch=0# 记录最佳epochpatience=50# 早停耐心值（连续多少轮测试集损失未改善时停止训练）counter=0# 早停计数器early_stopped=False# 是否早停标志# ==========================withtqdm(total=num_epochs,desc="训练进度",unit="epoch")aspbar:forepochinrange(num_epochs):outputs=model(X_train)train_loss=criterion(outputs,y_train)optimizer.zero_grad()train_loss.backward()optimizer.step()if(epoch+1)%200==0:model.eval()withtorch.no_grad():test_outputs=model(X_test)test_loss=criterion(test_outputs,y_test)model.train()train_losses.append(train_loss.item())test_losses.append(test_loss.item())epochs.append(epoch+1)pbar.set_postfix({'Train Loss':f'{train_loss.item():.4f}','Test Loss':f'{test_loss.item():.4f}'})# ===== 新增早停逻辑 =====iftest_loss.item()<best_test_loss:# 如果当前测试集损失小于最佳损失best_test_loss=test_loss.item()# 更新最佳损失best_epoch=epoch+1# 更新最佳epochcounter=0# 重置计数器# 保存最佳模型torch.save(model.state_dict(),'best_model.pth')else:counter+=1ifcounter>=patience:print(f"早停触发！在第{epoch+1}轮，测试集损失已有{patience}轮未改善。")print(f"最佳测试集损失出现在第{best_epoch}轮，损失值为{best_test_loss:.4f}")early_stopped=Truebreak# 终止训练循环# ======================# 每1000个epoch更新一次进度条if(epoch+1)%1000==0:pbar.update(1000)# 更新进度条ifpbar.n<num_epochs:pbar.update(num_epochs-pbar.n)# 可视化损失曲线plt.figure(figsize=(10,6))plt.plot(epochs,train_losses,label='Train Loss')# 原始代码已有plt.plot(epochs,test_losses,label='Test Loss')# 新增：测试集损失曲线plt.xlabel('Epoch')plt.ylabel('Loss')plt.title('Training and Test Loss over Epochs')plt.legend()# 新增：显示图例plt.grid(True)plt.show()# 在测试集上评估模型，此时model内部已经是训练好的参数了# 评估模型model.eval()# 设置模型为评估模式withtorch.no_grad():# torch.no_grad()的作用是禁用梯度计算，可以提高模型推理速度outputs=model(X_test)# 对测试数据进行前向传播，获得预测结果_,predicted=torch.max(outputs,1)# torch.max(outputs, 1)返回每行的最大值和对应的索引correct=(predicted==y_test).sum().item()# 计算预测正确的样本数accuracy=correct/y_test.size(0)print(f'测试集准确率:{accuracy*100:.2f}%')torch.save(model.state_dict(),"model_weights.pth")model=MLP()model.load_state_dict(torch.load("model_weights.pth"))torch.save(model,"full_model.pth")model=torch.load("full_model.pth",weights_only=False)model.eval()
