訓練+評估流程
- 1、要求
- 2、訓練+評估(PyTorch + TensorBoard )
- 完整代碼(單文件示例)
- 運行方法
- 功能對應表
- 3、pytorch自定義評估要繼承哪個類?
- 4、HF Trainer和SB3
- 5、 匯總
- 1. PyTorch Lightning + TensorBoard + ModelCheckpoint + EarlyStopping
- 核心代碼示例:
- 2. TensorFlow / Keras + TensorBoard + ModelCheckpoint + EarlyStopping
- 3. Stable Baselines3 (強化學習)
- 4. Huggingface Trainer(NLP)
- 5. 結合Weights & Biases(W\&B)
- 總結推薦
1、要求
訓練:
- 可視化
- 回調
- 自定義loss
- 自定義metric評估:
- 可視化
- 自定義數據:
- 自定義模型:
- 自定義
訓練:訓練可視化,自定義回調,評估指標
評估:評估可視化,自定義評估指標
數據:自定義數據
https://lightning.ai/docs/pytorch/stable/data/datamodule.html
模型:自定義模型
回調集合:參考sb3
穩定基線為您提供了一組常見的回調:
定期保存模型(CheckpointCallback)
定期評估模型并保存最佳模型(EvalCallback)
鏈接回調 (CallbackList)
觸發事件回調(事件回調、EveryNTimesteps)
每 N 個時間步長記錄數據 (LogEveryNTimesteps)
根據獎勵閾值提前停止訓練 (StopTrainingOnRewardThreshold)
2、訓練+評估(PyTorch + TensorBoard )
官方api,https://docs.pytorch.ac.cn/docs/stable/tensorboard.html
https://blog.csdn.net/Python_Ai_Road/article/details/107704530
https://github.com/haoxizhong/pytorch-worker
https://github.com/onepanelio/onepanel
提問:
pytorch tensorboard實現下面要求:
訓練:
- 可視化
- 回調
- 自定義metric評估:
- 可視化
- 自定義數據:
- 自定義模型:
- 自定義
好的,我幫你寫一個PyTorch + TensorBoard 完整可運行的示例,
滿足你所有要求:
-
訓練
? 可視化(loss/accuracy/自定義metric曲線)
? 回調(EarlyStopping)
? 自定義 metric(比如 F1-score) -
評估
? 可視化(混淆矩陣、曲線)
? 自定義指標(precision、recall、F1等) -
數據
? 自定義 Dataset / DataLoader -
模型
? 自定義 nn.Module
完整代碼(單文件示例)
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import Dataset, DataLoader
from torch.utils.tensorboard import SummaryWriter
import numpy as np
from sklearn.metrics import precision_score, recall_score, f1_score, confusion_matrix
import matplotlib.pyplot as plt
import time# ===================== 1. 自定義數據集 =====================
class MyDataset(Dataset):def __init__(self, size=500):np.random.seed(0)self.x = np.random.rand(size, 10).astype(np.float32)self.y = (self.x.sum(axis=1) > 5).astype(np.int64) # 二分類def __len__(self):return len(self.x)def __getitem__(self, idx):return self.x[idx], self.y[idx]# ===================== 2. 自定義模型 =====================
class MyModel(nn.Module):def __init__(self, input_dim=10, hidden_dim=32, num_classes=2):super().__init__()self.net = nn.Sequential(nn.Linear(input_dim, hidden_dim),nn.ReLU(),nn.Linear(hidden_dim, num_classes))def forward(self, x):return self.net(x)# ===================== 3. 回調(早停) =====================
class EarlyStopping:def __init__(self, patience=5, save_path="best_model.pth"):self.patience = patienceself.counter = 0self.best_loss = float("inf")self.save_path = save_pathself.early_stop = Falsedef __call__(self, val_loss, model):if val_loss < self.best_loss:self.best_loss = val_lossself.counter = 0torch.save(model.state_dict(), self.save_path)else:self.counter += 1if self.counter >= self.patience:self.early_stop = True# ===================== 4. 自定義 metric =====================
def compute_metrics(y_true, y_pred):precision = precision_score(y_true, y_pred, zero_division=0)recall = recall_score(y_true, y_pred, zero_division=0)f1 = f1_score(y_true, y_pred, zero_division=0)return precision, recall, f1# ===================== 5. 訓練(可視化 + 回調 + 自定義metric) =====================
def train_model():train_set = MyDataset(500)val_set = MyDataset(100)train_loader = DataLoader(train_set, batch_size=32, shuffle=True)val_loader = DataLoader(val_set, batch_size=32)device = torch.device("cuda" if torch.cuda.is_available() else "cpu")model = MyModel().to(device)criterion = nn.CrossEntropyLoss()optimizer = optim.Adam(model.parameters(), lr=0.001)log_dir = f"runs/train_{time.strftime('%Y%m%d-%H%M%S')}"writer = SummaryWriter(log_dir=log_dir)early_stopping = EarlyStopping(patience=5)for epoch in range(50):# ===== 訓練 =====model.train()total_loss, y_true_train, y_pred_train = 0, [], []for x, y in train_loader:x, y = x.to(device), y.to(device)optimizer.zero_grad()outputs = model(x)loss = criterion(outputs, y)loss.backward()optimizer.step()total_loss += loss.item()preds = outputs.argmax(1)y_true_train.extend(y.cpu().numpy())y_pred_train.extend(preds.cpu().numpy())train_loss = total_loss / len(train_loader)train_precision, train_recall, train_f1 = compute_metrics(y_true_train, y_pred_train)train_acc = np.mean(np.array(y_true_train) == np.array(y_pred_train))# ===== 驗證 =====model.eval()val_loss, y_true_val, y_pred_val = 0, [], []with torch.no_grad():for x, y in val_loader:x, y = x.to(device), y.to(device)outputs = model(x)loss = criterion(outputs, y)val_loss += loss.item()preds = outputs.argmax(1)y_true_val.extend(y.cpu().numpy())y_pred_val.extend(preds.cpu().numpy())val_loss /= len(val_loader)val_precision, val_recall, val_f1 = compute_metrics(y_true_val, y_pred_val)val_acc = np.mean(np.array(y_true_val) == np.array(y_pred_val))# ===== 寫入 TensorBoard =====writer.add_scalar("Loss/train", train_loss, epoch)writer.add_scalar("Loss/val", val_loss, epoch)writer.add_scalar("Accuracy/train", train_acc, epoch)writer.add_scalar("Accuracy/val", val_acc, epoch)writer.add_scalar("F1/train", train_f1, epoch)writer.add_scalar("F1/val", val_f1, epoch)print(f"Epoch {epoch+1} | "f"Train Loss: {train_loss:.4f} Acc: {train_acc:.4f} F1: {train_f1:.4f} | "f"Val Loss: {val_loss:.4f} Acc: {val_acc:.4f} F1: {val_f1:.4f}")# 回調early_stopping(val_loss, model)if early_stopping.early_stop:print("Early stopping!")breakwriter.close()print(f"訓練日志保存在 {log_dir}")# ===================== 6. 評估(可視化 + 自定義metric) =====================
def evaluate_model():device = torch.device("cuda" if torch.cuda.is_available() else "cpu")model = MyModel().to(device)model.load_state_dict(torch.load("best_model.pth"))model.eval()test_set = MyDataset(100)test_loader = DataLoader(test_set, batch_size=32)y_true, y_pred = [], []with torch.no_grad():for x, y in test_loader:x, y = x.to(device), y.to(device)outputs = model(x)preds = outputs.argmax(1)y_true.extend(y.cpu().numpy())y_pred.extend(preds.cpu().numpy())acc = np.mean(np.array(y_true) == np.array(y_pred))precision, recall, f1 = compute_metrics(y_true, y_pred)print(f"Test Accuracy: {acc:.4f}")print(f"Test Precision: {precision:.4f} Recall: {recall:.4f} F1: {f1:.4f}")# ===== 混淆矩陣可視化 =====cm = confusion_matrix(y_true, y_pred)fig, ax = plt.subplots()ax.matshow(cm, cmap=plt.cm.Blues)for i in range(cm.shape[0]):for j in range(cm.shape[1]):ax.text(j, i, cm[i, j], va='center', ha='center')plt.xlabel("Predicted")plt.ylabel("True")plt.title("Confusion Matrix")plt.show()if __name__ == "__main__":train_model()evaluate_model()
運行方法
# 安裝依賴
pip install torch tensorboard scikit-learn matplotlib# 運行
python demo.py# 查看TensorBoard
tensorboard --logdir runs
功能對應表
模塊 | 功能 | 實現方式 |
---|---|---|
訓練 | 可視化 loss / acc / F1 | SummaryWriter.add_scalar() |
回調 EarlyStopping | EarlyStopping 類 | |
自定義 metric(precision/recall/F1) | compute_metrics() | |
評估 | 可視化混淆矩陣 | matplotlib + confusion_matrix |
自定義 metric(precision/recall/F1) | compute_metrics() | |
數據 | 自定義 Dataset | MyDataset |
模型 | 自定義 nn.Module | MyModel |
我建議你如果后面項目要長期用,可以幫你改成模塊化結構(dataset.py、model.py、train.py、evaluate.py 分開),
這樣擴展性更好。
你要我下一步幫你拆成模塊化版本嗎?這樣更適合大型項目維護。
3、pytorch自定義評估要繼承哪個類?
pytorch自定義評估要繼承哪個類
在 PyTorch 里,如果你是純用 torch
(而不是 PyTorch Lightning、Hugging Face Trainer 等封裝框架),自定義評估其實不用繼承任何固定的類,因為 PyTorch 本身沒有“評估器”這一內置抽象,評估通常就是自己寫一個函數:
def evaluate(model, dataloader, criterion):model.eval()total_loss = 0.0correct = 0total = 0with torch.no_grad():for inputs, targets in dataloader:outputs = model(inputs)loss = criterion(outputs, targets)total_loss += loss.item()# 分類任務示例_, predicted = outputs.max(1)total += targets.size(0)correct += predicted.eq(targets).sum().item()avg_loss = total_loss / len(dataloader)accuracy = 100. * correct / totalreturn avg_loss, accuracy
但是,如果你是在 某個訓練框架 里做自定義評估,就會有對應的基類:
框架 | 自定義評估類要繼承的基類 |
---|---|
PyTorch Lightning | pl.Callback 或 pl.LightningModule (重寫 validation_step ) |
Hugging Face Transformers | TrainerCallback 或直接在 Trainer 里傳 compute_metrics |
TorchMetrics | torchmetrics.Metric |
Catalyst | catalyst.metrics.MetricCallback |
Ignite | ignite.metrics.Metric |
所以我需要確認一下,你的“自定義評估”是:
- 純 PyTorch 寫的訓練循環
- 還是基于 Lightning / Transformers / 其他訓練框架?
我猜你可能是想在 PPO 或 RL 里加自定義評估指標,如果是那種情況,類繼承方式又不一樣。
你要我幫你按 RL + PPO 場景寫一個自定義評估類嗎?這樣就直接能接到你現有環境里。
4、HF Trainer和SB3
詳細參考下面筆記:
https://blog.csdn.net/weixin_44986037/article/details/151153156?spm=1001.2014.3001.5502
https://blog.csdn.net/weixin_44986037/article/details/151225587?spm=1001.2014.3001.5502
https://blog.csdn.net/weixin_44986037/article/details/150427299?spm=1001.2014.3001.5502
https://blog.csdn.net/weixin_44986037/article/details/148544185?spm=1001.2014.3001.5501
建議用針對性框架:
Stable Baselines3 : 強化學習
Huggingface Trainer :NLP及Transformer模型
PyTorch Lightning:不追求訓練速度,建議用tensorboard,PyTorch Lightning坑有點多
5、 匯總
提問:現成的ai模型訓練評估框架,包括tersorboard查看訓練數據、篩選模型、評估效果
提問:
簡單概述主流現成的AI模型訓練+評估框架,包含:訓練過程中的數據監控(比如通過 TensorBoard 可視化損失、指標)模型篩選(如自動保存最優模型、早停)評估效果(自動計算指標,方便比較模型)
你想要的是一個現成的AI模型訓練+評估框架,包含:
- 訓練過程中的數據監控(比如通過 TensorBoard 可視化損失、指標)
- 模型篩選(如自動保存最優模型、早停)
- 評估效果(自動計算指標,方便比較模型)
下面給你推薦幾個主流且功能完整的框架/工具組合,覆蓋你需求的這幾個點:
1. PyTorch Lightning + TensorBoard + ModelCheckpoint + EarlyStopping
- 訓練監控:Lightning內置對TensorBoard的支持,訓練時自動記錄loss、metric到TensorBoard。
- 模型篩選:內置
ModelCheckpoint
,可自動保存指標最優模型(如val_loss最小、準確率最高)。 - 早停:內置
EarlyStopping
,當驗證指標不再提升時自動停止訓練。 - 評估效果:支持訓練、驗證、測試階段指標計算和記錄。
核心代碼示例:
import pytorch_lightning as pl
from pytorch_lightning.callbacks import ModelCheckpoint, EarlyStoppingcheckpoint_callback = ModelCheckpoint(monitor='val_loss', # 監控驗證損失mode='min', # 取最小值對應模型save_top_k=1,verbose=True,
)early_stop_callback = EarlyStopping(monitor='val_loss',patience=5,verbose=True,mode='min'
)trainer = pl.Trainer(max_epochs=50,callbacks=[checkpoint_callback, early_stop_callback],logger=pl.loggers.TensorBoardLogger('tb_logs/')
)trainer.fit(model, train_dataloader, val_dataloader)
啟動后,在終端運行 tensorboard --logdir=tb_logs/
即可實時查看訓練過程。
2. TensorFlow / Keras + TensorBoard + ModelCheckpoint + EarlyStopping
- TensorFlow內置支持TensorBoard,訓練時會自動寫入日志。
- ModelCheckpoint可保存驗證指標最優模型。
- EarlyStopping自動停止訓練避免過擬合。
示例:
callbacks = [tf.keras.callbacks.TensorBoard(log_dir='./logs'),tf.keras.callbacks.ModelCheckpoint('best_model.h5', monitor='val_loss', save_best_only=True),tf.keras.callbacks.EarlyStopping(monitor='val_loss', patience=5)
]model.fit(x_train, y_train, validation_data=(x_val, y_val), epochs=50, callbacks=callbacks)
3. Stable Baselines3 (強化學習)
- 自帶
EvalCallback
用于訓練過程中自動評估模型表現。 CheckpointCallback
用于保存模型。- 訓練日志可通過 TensorBoard 監控。
示例:
from stable_baselines3.common.callbacks import EvalCallback, CheckpointCallbackeval_callback = EvalCallback(eval_env, best_model_save_path='./logs/',log_path='./logs/', eval_freq=10000, deterministic=True)
checkpoint_callback = CheckpointCallback(save_freq=5000, save_path='./logs/')model.learn(total_timesteps=100000, callback=[eval_callback, checkpoint_callback])
運行 tensorboard --logdir=./logs
可查看指標變化。
4. Huggingface Trainer(NLP)
- 內置日志記錄(支持TensorBoard、WandB等)
- 自動保存最優模型和早停(EarlyStoppingCallback)
- 評估指標自動計算和記錄
示例:
from transformers import Trainer, TrainingArguments, EarlyStoppingCallbacktraining_args = TrainingArguments(output_dir='./results',evaluation_strategy='epoch',save_strategy='epoch',logging_dir='./logs',logging_steps=10,load_best_model_at_end=True,
)trainer = Trainer(model=model,args=training_args,train_dataset=train_dataset,eval_dataset=eval_dataset,callbacks=[EarlyStoppingCallback(early_stopping_patience=3)],
)trainer.train()
然后使用TensorBoard查看 ./logs
文件夾。
5. 結合Weights & Biases(W&B)
(即WandB)
- 支持各種框架(PyTorch、TensorFlow、Lightning、SB3、HF等)
- 自動記錄訓練日志、模型、超參
- 支持在線實時可視化和對比多個實驗
- 支持模型版本管理和篩選
官網:https://wandb.ai/site
總結推薦
框架/工具 | 訓練監控(TensorBoard) | 自動保存最優模型 | 早停 | 評估指標 | 簡單易用 | 適用范圍 |
---|---|---|---|---|---|---|
PyTorch Lightning | ? | ? | ? | ? | ? | PyTorch通用 |
TensorFlow / Keras | ? | ? | ? | ? | ? | TensorFlow/Keras |
Stable Baselines3 | ? | ? | 部分 | ? | ? | 強化學習 |
Huggingface Trainer | ? | ? | ? | ? | ? | NLP及Transformer模型 |
Weights & Biases ((WandB)) | ? | ? | ? | ? | ? | 各類深度學習框架 |
如果你告訴我你用的是哪種框架,我可以幫你寫具體的訓練+評估+TensorBoard配置示例,或者幫你搭建完整的訓練流水線。你看?