@浙大疏錦行?python day37.
內容：
- 保存模型只需要保存模型的參數即可，使用的時候直接構建模型再導入參數即可
# 保存模型參數
torch.save(model.state_dict(), "model_weights.pth")# 加載參數（需先定義模型結構）
model = MLP() # 初始化與訓練時相同的模型結構
model.load_state_dict(torch.load("model_weights.pth"))
# model.eval() # 切換至推理模式（可選）- 也可以同時保存模型 + 參數
# 保存整個模型
torch.save(model, "full_model.pth")# 加載模型（無需提前定義類，但需確保環境一致）
model = torch.load("full_model.pth")
model.eval() # 切換至推理模式（可選）- 保存訓練狀態，用于在訓練過程中保存中間狀態
# # 保存訓練狀態
# checkpoint = {
# "model_state_dict": model.state_dict(),
# "optimizer_state_dict": optimizer.state_dict(),
# "epoch": epoch,
# "loss": best_loss,
# }
# torch.save(checkpoint, "checkpoint.pth")# # 加載并續訓
# model = MLP()
# optimizer = torch.optim.Adam(model.parameters())
# checkpoint = torch.load("checkpoint.pth")# model.load_state_dict(checkpoint["model_state_dict"])
# optimizer.load_state_dict(checkpoint["optimizer_state_dict"])
# start_epoch = checkpoint["epoch"] + 1 # 從下一輪開始訓練
# best_loss = checkpoint["loss"]# # 繼續訓練循環
# for epoch in range(start_epoch, num_epochs):
# train(model, optimizer, ...)- 對于跨框架保存時，需要保存為onnx文件
- 早停策略：在訓練過程中，訓練集的損失不斷下降，但是驗證集或者測試集的損失反而上升，此時這種情況稱為過擬合；針對這種情況，我們可以引入早停策略，用于提前終止訓練；
- 可以在訓練達到某一個epoch次數時進行一次驗證，將此次結果和歷史最佳結果進行比較，如果結果更好則保留，如果不好則會使臨時記錄變量自增，一旦連續自增到預設值，直接停止。這種設計是為了防止出現震蕩波動情況，避免偶然性
- 可以結合上面的保存斷點 breakpoint
import torch
import torch.nn as nn
import torch.optim as optim
from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import MinMaxScaler
import time
import matplotlib.pyplot as plt
from tqdm import tqdm # 導入tqdm庫用于進度條顯示
import warnings
warnings.filterwarnings("ignore") # 忽略警告信息# 設置GPU設備
device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
print(f"使用設備: {device}")# 加載鳶尾花數據集
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)# 歸一化數據
scaler = MinMaxScaler()
X_train = scaler.fit_transform(X_train)
X_test = scaler.transform(X_test)# 將數據轉換為PyTorch張量并移至GPU
X_train = torch.FloatTensor(X_train).to(device)
y_train = torch.LongTensor(y_train).to(device)
X_test = torch.FloatTensor(X_test).to(device)
y_test = torch.LongTensor(y_test).to(device)class MLP(nn.Module):def __init__(self):super(MLP, self).__init__()self.fc1 = nn.Linear(4, 10) # 輸入層到隱藏層self.relu = nn.ReLU()self.fc2 = nn.Linear(10, 3) # 隱藏層到輸出層def forward(self, x):out = self.fc1(x)out = self.relu(out)out = self.fc2(out)return out# 實例化模型并移至GPU
model = MLP().to(device)# 分類問題使用交叉熵損失函數
criterion = nn.CrossEntropyLoss()# 使用隨機梯度下降優化器
optimizer = optim.SGD(model.parameters(), lr=0.01)# 訓練模型
num_epochs = 20000 # 訓練的輪數# 用于存儲每200個epoch的損失值和對應的epoch數
train_losses = [] # 存儲訓練集損失
test_losses = [] # 存儲測試集損失
epochs = []# ===== 新增早停相關參數 =====
best_test_loss = float('inf') # 記錄最佳測試集損失
best_epoch = 0 # 記錄最佳epoch
patience = 50 # 早停耐心值（連續多少輪測試集損失未改善時停止訓練）
counter = 0 # 早停計數器
early_stopped = False # 是否早停標志
# ==========================start_time = time.time() # 記錄開始時間# 創建tqdm進度條
with tqdm(total=num_epochs, desc="訓練進度", unit="epoch") as pbar:# 訓練模型for epoch in range(num_epochs):# 前向傳播outputs = model(X_train) # 隱式調用forward函數train_loss = criterion(outputs, y_train)# 反向傳播和優化optimizer.zero_grad()train_loss.backward()optimizer.step()# 記錄損失值并更新進度條if (epoch + 1) % 200 == 0:# 計算測試集損失model.eval()with torch.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}'})# ===== 新增早停邏輯 =====if test_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 += 1if counter >= 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) # 更新進度條# 確保進度條達到100%if pbar.n < num_epochs:pbar.update(num_epochs - pbar.n) # 計算剩余的進度并更新time_all = time.time() - start_time # 計算訓練時間
print(f'Training time: {time_all:.2f} seconds')# ===== 新增：加載最佳模型用于最終評估 =====
if early_stopped:print(f"加載第{best_epoch}輪的最佳模型進行最終評估...")model.load_state_dict(torch.load('best_model.pth'))
# ================================# 可視化損失曲線
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.eval()
with torch.no_grad():outputs = model(X_test)_, predicted = torch.max(outputs, 1)correct = (predicted == y_test).sum().item()accuracy = correct / y_test.size(0)print(f'測試集準確率: {accuracy * 100:.2f}%') 
