# ? 不使用 Callback 的代碼（耦合度高、難以維護）
for epoch in range(max_epochs):
    # 訓練邏輯
    train_loss = train_epoch(model, train_loader)
    
    # 驗證邏輯
    val_loss = validate(model, val_loader)
    
    # 手動保存最佳模型
    if val_loss < best_loss:
        best_loss = val_loss
        torch.save(model.state_dict(), 'best_model.pth')
    
    # 手動早停邏輯
    if val_loss > best_loss:
        patience_counter += 1
        if patience_counter >= patience:
            print("Early stopping!")
            break
    
    # 手動記錄日志
    log_metrics(epoch, train_loss, val_loss)
    
    # ... 更多邏輯混雜在一起

# ? 使用 Callback 的代碼（清晰、模塊化、可復(fù)用）
trainer = pl.Trainer(
    max_epochs=100,
    callbacks=[
        ModelCheckpoint(monitor='val_loss', mode='min'),
        EarlyStopping(monitor='val_loss', patience=10),
        LearningRateMonitor(logging_interval='epoch'),
    ]
)
trainer.fit(model, train_loader, val_loader)

訓練流程                    Callback 鉤子觸發(fā)時機
│
├─ Trainer.fit()
│  │
│  ├─ on_fit_start()        ← 訓練開始前
│  │
│  ├─ Epoch Loop
│  │  │
│  │  ├─ on_train_epoch_start()  ← 每個訓練 epoch 開始
│  │  │
│  │  ├─ Training Batch Loop
│  │  │  ├─ on_train_batch_start()   ← 每個訓練 batch 前
│  │  │  ├─ training_step()
│  │  │  └─ on_train_batch_end()     ← 每個訓練 batch 后
│  │  │
│  │  ├─ on_train_epoch_end()    ← 每個訓練 epoch 結(jié)束
│  │  │
│  │  ├─ Validation Loop
│  │  │  ├─ on_validation_epoch_start()
│  │  │  ├─ validation_step()
│  │  │  └─ on_validation_epoch_end()
│  │  │
│  │  └─ on_epoch_end()          ← 每個完整 epoch 結(jié)束
│  │
│  └─ on_fit_end()          ← 訓練完全結(jié)束
│
└─ Trainer.test()
   ├─ on_test_start()
   ├─ test_step()
   └─ on_test_end()

from pytorch_lightning.callbacks import ModelCheckpoint

# 示例1：保存驗證損失最低的模型
checkpoint = ModelCheckpoint(
    monitor='val_loss',           # 監(jiān)控的指標
    dirpath='checkpoints/',       # 保存目錄
    filename='best-{epoch:02d}-{val_loss:.4f}',  # 文件名模板
    save_top_k=1,                 # 保存最好的 1 個模型
    mode='min',                   # 'min' 表示越小越好，'max' 表示越大越好
    save_last=True,               # 額外保存最后一個 epoch 的模型
    verbose=True,                 # 打印日志
)

trainer = pl.Trainer(callbacks=[checkpoint])

ModelCheckpoint(
    # 核心參數(shù)
    monitor='val_loss',              # 監(jiān)控的指標名稱（必須在 self.log() 中記錄）
    mode='min',                      # 'min'/'max'/'auto'
    
    # 保存策略
    save_top_k=3,                    # 保存最好的 k 個模型（-1 表示全部保存）
    save_last=True,                  # 是否額外保存最后一個模型（last.ckpt）
    save_weights_only=False,         # True: 僅保存權(quán)重，F(xiàn)alse: 保存完整狀態(tài)
    
    # 文件命名
    dirpath='checkpoints/',          # 保存目錄
    filename='epoch={epoch:02d}-val_loss={val_loss:.4f}',  # 文件名模板
    auto_insert_metric_name=True,    # 自動在文件名中插入 monitor 名稱
    
    # 觸發(fā)條件
    every_n_epochs=1,                # 每 n 個 epoch 檢查一次
    every_n_train_steps=None,        # 每 n 個訓練步檢查一次
    train_time_interval=None,        # 按時間間隔檢查（如 timedelta(minutes=30)）
    
    # 其他
    verbose=True,                    # 是否打印保存信息
    save_on_train_epoch_end=None,    # 在訓練 epoch 結(jié)束時保存（默認驗證后）
)

# 保存 val_loss 最低的模型
checkpoint_loss = ModelCheckpoint(
    monitor='val_loss',
    dirpath='checkpoints/loss/',
    filename='best-loss-{epoch:02d}-{val_loss:.4f}',
    mode='min',
    save_top_k=1,
)

# 保存 val_acc 最高的模型
checkpoint_acc = ModelCheckpoint(
    monitor='val_acc',
    dirpath='checkpoints/acc/',
    filename='best-acc-{epoch:02d}-{val_acc:.4f}',
    mode='max',
    save_top_k=1,
)

trainer = pl.Trainer(callbacks=[checkpoint_loss, checkpoint_acc])

# 每 5 個 epoch 保存一次
checkpoint_periodic = ModelCheckpoint(
    dirpath='checkpoints/periodic/',
    filename='epoch={epoch:02d}',
    every_n_epochs=5,
    save_top_k=-1,  # 保存所有
)

checkpoint_steps = ModelCheckpoint(
    dirpath='checkpoints/steps/',
    filename='step={step}',
    every_n_train_steps=1000,  # 每 1000 步保存
    save_top_k=-1,
)

from datetime import timedelta

checkpoint_time = ModelCheckpoint(
    dirpath='checkpoints/timed/',
    train_time_interval=timedelta(minutes=30),  # 每 30 分鐘保存
    save_top_k=-1,
)

trainer.fit(model, train_loader, val_loader)

# 獲取最佳模型路徑
best_model_path = checkpoint.best_model_path
print(f"Best model: {best_model_path}")

# 獲取最佳分數(shù)
best_score = checkpoint.best_model_score
print(f"Best score: {best_score}")

# 加載最佳模型
best_model = MyModel.load_from_checkpoint(best_model_path)

from pytorch_lightning.callbacks import EarlyStopping

early_stop = EarlyStopping(
    monitor='val_loss',     # 監(jiān)控的指標
    patience=10,            # 容忍多少個 epoch 不改善
    mode='min',             # 'min' 或 'max'
    verbose=True,           # 打印停止信息
    min_delta=0.001,        # 最小改善量（小于此值不算改善）
)

trainer = pl.Trainer(callbacks=[early_stop])

EarlyStopping(
    # 核心參數(shù)
    monitor='val_loss',              # 監(jiān)控的指標
    mode='min',                      # 'min'/'max'/'auto'
    patience=3,                      # 容忍的 epoch 數(shù)
    
    # 判斷標準
    min_delta=0.0,                   # 最小改善閾值（絕對值）
    strict=True,                     # 是否嚴格要求改善（False 允許相等）
    
    # 停止行為
    stopping_threshold=None,         # 達到此值立即停止（如 val_loss < 0.01）
    divergence_threshold=None,       # 超過此值立即停止（如 val_loss > 10.0）
    check_finite=True,               # 檢查指標是否為有限值
    check_on_train_epoch_end=None,   # 在訓練 epoch 結(jié)束時檢查（默認驗證后）
    
    # 日志
    verbose=True,
    log_rank_zero_only=False,        # 僅在主進程打印
)

early_stop = EarlyStopping(
    monitor='val_loss',
    patience=15,
    mode='min',
    verbose=True,
)

early_stop = EarlyStopping(
    monitor='val_acc',
    patience=10,
    mode='max',
    min_delta=0.005,  # 提升小于 0.5% 不算改善
)

early_stop = EarlyStopping(
    monitor='val_acc',
    stopping_threshold=0.95,  # 準確率達到 95% 立即停止
    mode='max',
)

early_stop = EarlyStopping(
    monitor='train_loss',
    divergence_threshold=10.0,  # 訓練損失超過 10 立即停止
    mode='min',
)

from pytorch_lightning.callbacks import LearningRateMonitor

lr_monitor = LearningRateMonitor(
    logging_interval='epoch',  # 'step' 或 'epoch'
    log_momentum=False,        # 是否記錄 momentum（SGD 優(yōu)化器）
)

trainer = pl.Trainer(callbacks=[lr_monitor])

class MyModel(pl.LightningModule):
    def configure_optimizers(self):
        optimizer = torch.optim.Adam(self.parameters(), lr=1e-3)
        scheduler = torch.optim.lr_scheduler.ReduceLROnPlateau(
            optimizer, mode='min', factor=0.5, patience=5
        )
        return {
            'optimizer': optimizer,
            'lr_scheduler': {
                'scheduler': scheduler,
                'monitor': 'val_loss',
            }
        }

# 在 TensorBoard 中自動記錄 lr 曲線
trainer = pl.Trainer(
    callbacks=[LearningRateMonitor(logging_interval='epoch')],
    logger=TensorBoardLogger('logs/')
)

lr_monitor = LearningRateMonitor(logging_interval='step')

from pytorch_lightning.callbacks import RichProgressBar

# 默認配置
progress_bar = RichProgressBar()

# 自定義配置
progress_bar = RichProgressBar(
    refresh_rate=1,              # 刷新頻率（步數(shù)）
    leave=True,                  # 訓練結(jié)束后保留進度條
    theme=RichProgressBarTheme(  # 自定義主題
        description="green_yellow",
        progress_bar="green1",
        progress_bar_finished="green1",
        batch_progress="green_yellow",
        time="grey82",
        processing_speed="grey82",
        metrics="grey82",
    ),
)

class CustomProgressBar(RichProgressBar):
    def get_metrics(self, trainer, model):
        # 獲取父類的指標
        items = super().get_metrics(trainer, model)
        
        # 自定義顯示格式（如顯示更多小數(shù)位）
        items = {
            k: f"{v:.6f}" if isinstance(v, (int, float)) else v
            for k, v in items.items()
        }
        return items

from pytorch_lightning.callbacks import GradientAccumulationScheduler

# 在不同 epoch 使用不同的累積步數(shù)
accumulator = GradientAccumulationScheduler(
    scheduling={
        0: 8,   # epoch 0-4: 累積 8 步
        5: 4,   # epoch 5-9: 累積 4 步
        10: 2,  # epoch 10+: 累積 2 步
    }
)

trainer = pl.Trainer(callbacks=[accumulator])

# 等效 batch size 變化：
# epoch 0-4:  batch_size=16 × accumulate=8 = 128
# epoch 5-9:  batch_size=16 × accumulate=4 = 64
# epoch 10+:  batch_size=16 × accumulate=2 = 32

accumulator = GradientAccumulationScheduler(
    scheduling={0: 8, 5: 4, 10: 2}
)

from pytorch_lightning.callbacks import StochasticWeightAveraging

swa = StochasticWeightAveraging(
    swa_lrs=1e-2,              # SWA 階段的學習率
    swa_epoch_start=0.8,       # 從 80% epoch 開始 SWA（0.8 × max_epochs）
    annealing_epochs=10,       # 退火 epoch 數(shù)
    annealing_strategy='cos',  # 'cos' 或 'linear'
)

trainer = pl.Trainer(
    max_epochs=100,
    callbacks=[swa]
)

正常訓練: 模型權(quán)重在最優(yōu)點附近震蕩
SWA:      對后期權(quán)重求平均，得到更平滑的模型

訓練曲線:
        ╱╲  ╱╲  ╱╲
Loss   ╱  ╲╱  ╲╱  ╲  ← 正常訓練
      ╱____________╲  ← SWA 平均后（更穩(wěn)定）
      ↑
    SWA Start

from pytorch_lightning.callbacks import ModelSummary

summary = ModelSummary(
    max_depth=2,  # 顯示的最大層級深度（-1 表示全部）
)

trainer = pl.Trainer(callbacks=[summary])

  | Name      | Type   | Params
------------------------------------
0 | layer1    | Linear | 320
1 | layer2    | Linear | 640
2 | layer3    | Linear | 10
------------------------------------
970       Trainable params
0         Non-trainable params
970       Total params

from pytorch_lightning.callbacks import Timer
from datetime import timedelta

timer = Timer(
    duration=timedelta(hours=2),  # 最大訓練時間 2 小時
    interval='epoch',             # 檢查間隔（'step' 或 'epoch'）
    verbose=True,
)

trainer = pl.Trainer(callbacks=[timer])

from pytorch_lightning.callbacks import DeviceStatsMonitor

device_stats = DeviceStatsMonitor()

trainer = pl.Trainer(callbacks=[device_stats])

from pytorch_lightning.callbacks import BaseFinetuning

class FeatureExtractorFreezeUnfreeze(BaseFinetuning):
    def __init__(self, unfreeze_at_epoch=10):
        super().__init__()
        self._unfreeze_at_epoch = unfreeze_at_epoch

    def freeze_before_training(self, pl_module):
        # 初始凍結(jié)骨干網(wǎng)絡(luò)
        self.freeze(pl_module.feature_extractor)

    def finetune_function(self, pl_module, current_epoch, optimizer):
        # 在指定 epoch 解凍
        if current_epoch == self._unfreeze_at_epoch:
            self.unfreeze_and_add_param_group(
                modules=pl_module.feature_extractor,
                optimizer=optimizer,
                lr=1e-5,  # 使用更小的學習率
            )

trainer = pl.Trainer(callbacks=[FeatureExtractorFreezeUnfreeze(unfreeze_at_epoch=10)])

def on_train_epoch_end(self, trainer, pl_module):
    # 訪問當前 epoch
    current_epoch = trainer.current_epoch
    
    # 訪問全局步數(shù)
    global_step = trainer.global_step
    
    # 訪問日志記錄的指標
    logged_metrics = trainer.logged_metrics
    
    # 訪問回調(diào)指標（用于 ModelCheckpoint 等）
    callback_metrics = trainer.callback_metrics
    
    # 訪問模型參數(shù)
    for name, param in pl_module.named_parameters():
        print(f"{name}: {param.shape}")

# 完整訓練流程的鉤子調(diào)用順序

trainer.fit(model, train_loader, val_loader)
│
├─ on_fit_start()
│  ├─ on_train_start()
│  │
│  ├─ Epoch 0
│  │  ├─ on_epoch_start()
│  │  ├─ on_train_epoch_start()
│  │  │
│  │  ├─ Training Batches
│  │  │  ├─ on_train_batch_start(batch_idx=0)
│  │  │  ├─ on_before_backward()
│  │  │  ├─ on_after_backward()
│  │  │  ├─ on_before_optimizer_step()
│  │  │  ├─ on_before_zero_grad()
│  │  │  ├─ on_train_batch_end(batch_idx=0)
│  │  │  │
│  │  │  ├─ on_train_batch_start(batch_idx=1)
│  │  │  └─ ...
│  │  │
│  │  ├─ on_train_epoch_end()
│  │  │
│  │  ├─ Validation (如果啟用)
│  │  │  ├─ on_validation_epoch_start()
│  │  │  ├─ on_validation_batch_start(batch_idx=0)
│  │  │  ├─ on_validation_batch_end(batch_idx=0)
│  │  │  └─ on_validation_epoch_end()
│  │  │
│  │  └─ on_epoch_end()
│  │
│  ├─ Epoch 1
│  │  └─ ... (同上)
│  │
│  └─ on_train_end()
│
└─ on_fit_end()

import pytorch_lightning as pl
from pytorch_lightning.callbacks import Callback

class MyCustomCallback(Callback):
    """自定義 Callback 模板"""
    
    def __init__(self, custom_param):
        super().__init__()
        self.custom_param = custom_param
        # 初始化自定義狀態(tài)
        self.state = {}
    
    def on_train_start(self, trainer, pl_module):
        """訓練開始時調(diào)用"""
        print(f"訓練開始，參數(shù): {self.custom_param}")
    
    def on_train_epoch_end(self, trainer, pl_module):
        """每個訓練 epoch 結(jié)束時調(diào)用"""
        # 訪問訓練指標
        metrics = trainer.callback_metrics
        print(f"Epoch {trainer.current_epoch} 結(jié)束")
    
    def on_validation_epoch_end(self, trainer, pl_module):
        """每個驗證 epoch 結(jié)束時調(diào)用"""
        pass

class TrainingReportCallback(Callback):
    """每個 epoch 結(jié)束后打印詳細報告"""
    
    def on_train_epoch_end(self, trainer, pl_module):
        metrics = trainer.callback_metrics
        
        print("\n" + "="*60)
        print(f"Epoch {trainer.current_epoch} 訓練報告")
        print("="*60)
        
        for key, value in metrics.items():
            if isinstance(value, torch.Tensor):
                value = value.item()
            print(f"{key:30s}: {value:.6f}")
        
        print("="*60 + "\n")

class SaveValidationPredictionsCallback(Callback):
    """保存每個 epoch 的驗證集預(yù)測結(jié)果"""
    
    def __init__(self, save_dir='predictions/'):
        super().__init__()
        self.save_dir = save_dir
        self.predictions = []
        self.targets = []
    
    def on_validation_epoch_start(self, trainer, pl_module):
        # 重置存儲
        self.predictions = []
        self.targets = []
    
    def on_validation_batch_end(self, trainer, pl_module, outputs, 
                                batch, batch_idx, dataloader_idx=0):
        # 收集預(yù)測結(jié)果
        if isinstance(outputs, dict) and 'preds' in outputs:
            self.predictions.append(outputs['preds'].cpu())
            self.targets.append(outputs['targets'].cpu())
    
    def on_validation_epoch_end(self, trainer, pl_module):
        # 合并并保存
        if self.predictions:
            all_preds = torch.cat(self.predictions)
            all_targets = torch.cat(self.targets)
            
            save_path = f"{self.save_dir}/epoch_{trainer.current_epoch}.pt"
            torch.save({
                'predictions': all_preds,
                'targets': all_targets,
                'epoch': trainer.current_epoch
            }, save_path)
            
            print(f"驗證集預(yù)測已保存: {save_path}")

class CustomLRScheduler(Callback):
    """基于驗證損失的自定義學習率調(diào)整"""
    
    def __init__(self, patience=5, factor=0.5, min_lr=1e-6):
        super().__init__()
        self.patience = patience
        self.factor = factor
        self.min_lr = min_lr
        self.best_loss = float('inf')
        self.wait = 0
    
    def on_validation_epoch_end(self, trainer, pl_module):
        # 獲取當前驗證損失
        val_loss = trainer.callback_metrics.get('val_loss')
        
        if val_loss is None:
            return
        
        val_loss = val_loss.item()
        
        # 檢查是否改善
        if val_loss < self.best_loss:
            self.best_loss = val_loss
            self.wait = 0
        else:
            self.wait += 1
            
            if self.wait >= self.patience:
                # 降低學習率
                for optimizer in trainer.optimizers:
                    for param_group in optimizer.param_groups:
                        old_lr = param_group['lr']
                        new_lr = max(old_lr * self.factor, self.min_lr)
                        param_group['lr'] = new_lr
                        
                        print(f"\n學習率調(diào)整: {old_lr:.6f} → {new_lr:.6f}")
                
                self.wait = 0

class GradientLoggingCallback(Callback):
    """記錄梯度統(tǒng)計信息"""
    
    def __init__(self, log_every_n_steps=100):
        super().__init__()
        self.log_every_n_steps = log_every_n_steps
    
    def on_after_backward(self, trainer, pl_module):
        if trainer.global_step % self.log_every_n_steps != 0:
            return
        
        # 計算梯度統(tǒng)計
        grad_norms = []
        for name, param in pl_module.named_parameters():
            if param.grad is not None:
                grad_norm = param.grad.norm().item()
                grad_norms.append(grad_norm)
                
                # 記錄每層梯度
                pl_module.log(f'grad_norm/{name}', grad_norm)
        
        # 記錄平均梯度范數(shù)
        if grad_norms:
            avg_grad_norm = sum(grad_norms) / len(grad_norms)
            pl_module.log('grad_norm/average', avg_grad_norm)

import os
import glob

class CheckpointCleanupCallback(Callback):
    """自動清理舊的檢查點文件，僅保留最新的 N 個"""
    
    def __init__(self, checkpoint_dir='checkpoints/', keep_last_n=3):
        super().__init__()
        self.checkpoint_dir = checkpoint_dir
        self.keep_last_n = keep_last_n
    
    def on_train_epoch_end(self, trainer, pl_module):
        # 獲取所有檢查點文件
        ckpt_files = glob.glob(f"{self.checkpoint_dir}/*.ckpt")
        
        # 按修改時間排序
        ckpt_files.sort(key=os.path.getmtime, reverse=True)
        
        # 刪除舊文件
        for ckpt_file in ckpt_files[self.keep_last_n:]:
            try:
                os.remove(ckpt_file)
                print(f"刪除舊檢查點: {ckpt_file}")
            except Exception as e:
                print(f"刪除失敗: {e}")

import smtplib
from email.mime.text import MIMEText

class EmailNotificationCallback(Callback):
    """訓練完成或異常時發(fā)送郵件通知"""
    
    def __init__(self, recipient_email, smtp_config):
        super().__init__()
        self.recipient_email = recipient_email
        self.smtp_config = smtp_config
    
    def send_email(self, subject, message):
        """發(fā)送郵件"""
        msg = MIMEText(message)
        msg['Subject'] = subject
        msg['From'] = self.smtp_config['from']
        msg['To'] = self.recipient_email
        
        try:
            with smtplib.SMTP(self.smtp_config['server'], 
                            self.smtp_config['port']) as server:
                server.login(self.smtp_config['username'], 
                           self.smtp_config['password'])
                server.send_message(msg)
        except Exception as e:
            print(f"郵件發(fā)送失敗: {e}")
    
    def on_train_end(self, trainer, pl_module):
        """訓練結(jié)束時發(fā)送通知"""
        metrics = trainer.callback_metrics
        
        message = f"""
        訓練已完成！
        
        最終指標:
        {metrics}
        
        總 Epoch: {trainer.current_epoch}
        總步數(shù): {trainer.global_step}
        """
        
        self.send_email("訓練完成通知", message)
    
    def on_exception(self, trainer, pl_module, exception):
        """發(fā)生異常時發(fā)送通知"""
        message = f"訓練發(fā)生異常: {exception}"
        self.send_email("訓練異常通知", message)

import matplotlib.pyplot as plt

class RealTimePlotCallback(Callback):
    """實時繪制訓練曲線"""
    
    def __init__(self):
        super().__init__()
        self.train_losses = []
        self.val_losses = []
        self.epochs = []
        
        # 創(chuàng)建圖形
        plt.ion()  # 交互模式
        self.fig, self.ax = plt.subplots()
    
    def on_train_epoch_end(self, trainer, pl_module):
        # 記錄數(shù)據(jù)
        metrics = trainer.callback_metrics
        self.epochs.append(trainer.current_epoch)
        
        if 'train_loss' in metrics:
            self.train_losses.append(metrics['train_loss'].item())
    
    def on_validation_epoch_end(self, trainer, pl_module):
        metrics = trainer.callback_metrics
        
        if 'val_loss' in metrics:
            self.val_losses.append(metrics['val_loss'].item())
            
            # 更新圖形
            self.ax.clear()
            self.ax.plot(self.epochs, self.train_losses, label='Train Loss')
            if len(self.val_losses) > 0:
                self.ax.plot(self.epochs, self.val_losses, label='Val Loss')
            self.ax.legend()
            self.ax.set_xlabel('Epoch')
            self.ax.set_ylabel('Loss')
            self.fig.canvas.draw()
            self.fig.canvas.flush_events()
    
    def on_train_end(self, trainer, pl_module):
        # 保存最終圖形
        plt.ioff()
        self.fig.savefig('training_curve.png')
        print("訓練曲線已保存: training_curve.png")

class DataInspectionCallback(Callback):
    def on_train_batch_end(self, trainer, pl_module, outputs, batch, batch_idx):
        # 訪問模型
        model = pl_module
        
        # 訪問批次數(shù)據(jù)
        x, y = batch  # 根據(jù)實際數(shù)據(jù)結(jié)構(gòu)解包
        
        # 訪問模型輸出
        predictions = outputs['preds']  # 根據(jù) training_step 返回值
        
        # 訪問優(yōu)化器
        optimizer = trainer.optimizers[0]
        current_lr = optimizer.param_groups[0]['lr']
        
        # 訪問日志器
        logger = trainer.logger
        logger.log_metrics({'custom_metric': 1.0}, step=trainer.global_step)

callbacks = [
    # 保存最佳模型
    ModelCheckpoint(
        monitor='val_loss',
        mode='min',
        save_top_k=1,
        filename='best-{epoch:02d}-{val_loss:.4f}',
    ),
    
    # 早停
    EarlyStopping(
        monitor='val_loss',
        patience=15,
        mode='min',
    ),
    
    # 學習率監(jiān)控
    LearningRateMonitor(logging_interval='epoch'),
    
    # 進度條
    RichProgressBar(),
]

trainer = pl.Trainer(
    max_epochs=100,
    callbacks=callbacks,
    logger=TensorBoardLogger('logs/'),
)

callbacks = [
    # 1. 多重模型保存策略
    ModelCheckpoint(
        monitor='val_loss',
        mode='min',
        save_top_k=3,
        filename='best-loss-{epoch:02d}-{val_loss:.4f}',
    ),
    ModelCheckpoint(
        monitor='val_acc',
        mode='max',
        save_top_k=1,
        filename='best-acc-{epoch:02d}-{val_acc:.4f}',
    ),
    ModelCheckpoint(
        every_n_epochs=10,
        filename='periodic-{epoch:02d}',
        save_top_k=-1,  # 保存所有
    ),
    
    # 2. 早停 + 發(fā)散檢測
    EarlyStopping(
        monitor='val_loss',
        patience=20,
        mode='min',
        min_delta=0.001,
    ),
    EarlyStopping(
        monitor='train_loss',
        divergence_threshold=10.0,  # 檢測 loss 爆炸
        mode='min',
    ),
    
    # 3. 學習率監(jiān)控
    LearningRateMonitor(logging_interval='step'),
    
    # 4. 設(shè)備狀態(tài)監(jiān)控
    DeviceStatsMonitor(),
    
    # 5. 時間限制（如云服務(wù)器按時計費）
    Timer(duration=timedelta(hours=10)),
    
    # 6. 自定義訓練報告
    TrainingReportCallback(),
]

callbacks = [
    # 1. 模型保存
    ModelCheckpoint(
        monitor='val_loss',
        mode='min',
        save_top_k=5,
        save_last=True,
    ),
    
    # 2. 早停
    EarlyStopping(monitor='val_loss', patience=30),
    
    # 3. 學習率監(jiān)控
    LearningRateMonitor(logging_interval='step'),
    
    # 4. 梯度監(jiān)控（檢測梯度消失/爆炸）
    GradientLoggingCallback(log_every_n_steps=50),
    
    # 5. 保存驗證集預(yù)測（用于后續(xù)分析）
    SaveValidationPredictionsCallback(save_dir='predictions/'),
    
    # 6. 實時可視化
    RealTimePlotCallback(),
    
    # 7. 模型摘要
    ModelSummary(max_depth=3),
]

# 同時使用 TensorBoard 和 WandB
trainer = pl.Trainer(
    callbacks=callbacks,
    logger=[
        TensorBoardLogger('logs/tensorboard/'),
        WandbLogger(project='my_research', name='exp_001'),
    ],
)

callbacks = [
    # 1. 僅保存權(quán)重（減小文件體積）
    ModelCheckpoint(
        monitor='val_loss',
        mode='min',
        save_top_k=1,
        save_weights_only=True,  # 僅保存權(quán)重
        filename='production-best',
    ),
    
    # 2. 早停
    EarlyStopping(
        monitor='val_loss',
        patience=10,
        stopping_threshold=0.05,  # 達到目標即停止
    ),
    
    # 3. SWA 提升泛化性能
    StochasticWeightAveraging(swa_lrs=1e-2),
    
    # 4. 檢查點清理（節(jié)省存儲）
    CheckpointCleanupCallback(keep_last_n=2),
    
    # 5. 訓練完成通知
    EmailNotificationCallback(
        recipient_email='team@company.com',
        smtp_config={...}
    ),
]

# 使用 Trainer 的快速開發(fā)標志
trainer = pl.Trainer(
    max_epochs=2,  # 少量 epoch
    limit_train_batches=10,  # 僅訓練 10 個 batch
    limit_val_batches=5,     # 僅驗證 5 個 batch
    callbacks=[
        RichProgressBar(),
        ModelSummary(max_depth=-1),  # 查看完整模型結(jié)構(gòu)
    ],
    logger=False,  # 不記錄日志
    enable_checkpointing=False,  # 不保存檢查點
)

from ray import tune
from ray.tune.integration.pytorch_lightning import TuneReportCallback

def train_func(config):
    model = MyModel(
        lr=config['lr'],
        hidden_dim=config['hidden_dim'],
    )
    
    trainer = pl.Trainer(
        max_epochs=20,
        callbacks=[
            # Ray Tune 回調(diào)（報告指標）
            TuneReportCallback(
                metrics={'val_loss': 'val_loss'},
                on='validation_end',
            ),
            EarlyStopping(monitor='val_loss', patience=5),
        ],
        enable_progress_bar=False,  # 禁用進度條（避免輸出混亂）
        enable_model_summary=False,
    )
    
    trainer.fit(model, train_loader, val_loader)

# 啟動超參數(shù)搜索
analysis = tune.run(
    train_func,
    config={
        'lr': tune.loguniform(1e-4, 1e-1),
        'hidden_dim': tune.choice([64, 128, 256]),
    },
    num_samples=20,
)

class SharedStateCallback(Callback):
    """使用 Trainer 的自定義屬性共享狀態(tài)"""
    
    def on_train_start(self, trainer, pl_module):
        # 初始化共享狀態(tài)
        trainer.my_shared_state = {'counter': 0}
    
    def on_train_batch_end(self, trainer, pl_module, outputs, batch, batch_idx):
        # 更新共享狀態(tài)
        trainer.my_shared_state['counter'] += 1

class AnotherCallback(Callback):
    def on_validation_epoch_end(self, trainer, pl_module):
        # 讀取共享狀態(tài)
        counter = trainer.my_shared_state['counter']
        print(f"已訓練 {counter} 個 batch")

class ConditionalCallback(Callback):
    """僅在特定條件下執(zhí)行"""
    
    def __init__(self, execute_after_epoch=10):
        super().__init__()
        self.execute_after_epoch = execute_after_epoch
    
    def on_validation_epoch_end(self, trainer, pl_module):
        # 僅在第 10 個 epoch 后執(zhí)行
        if trainer.current_epoch >= self.execute_after_epoch:
            print("執(zhí)行特殊操作...")

callbacks = [
    CallbackA(),  # 第一個執(zhí)行
    CallbackB(),  # 第二個執(zhí)行
    CallbackC(),  # 第三個執(zhí)行
]

# 注意：ModelCheckpoint 和 EarlyStopping 的順序很重要！
callbacks = [
    ModelCheckpoint(...),  # 先保存模型
    EarlyStopping(...),    # 再判斷是否停止
]

class CustomLoggingCallback(Callback):
    def on_train_epoch_end(self, trainer, pl_module):
        # 方式1：通過 pl_module 記錄
        pl_module.log('custom_metric', 1.0)
        
        # 方式2：直接使用 logger
        if trainer.logger:
            trainer.logger.log_metrics(
                {'another_metric': 2.0},
                step=trainer.global_step
            )
            
            # 如果是 TensorBoard
            if isinstance(trainer.logger, TensorBoardLogger):
                trainer.logger.experiment.add_scalar(
                    'special_metric', 3.0, trainer.global_step
                )

class DistributedAwareCallback(Callback):
    """在分布式訓練中正確處理"""
    
    def on_validation_epoch_end(self, trainer, pl_module):
        # 僅在主進程執(zhí)行（避免重復(fù)）
        if trainer.is_global_zero:
            print("這只在主進程打印一次")
        
        # 所有進程都執(zhí)行
        local_rank = trainer.local_rank
        print(f"進程 {local_rank} 執(zhí)行")

import unittest

class TestMyCallback(unittest.TestCase):
    def test_callback_logic(self):
        # 創(chuàng)建模擬的 trainer 和 model
        trainer = MockTrainer()
        model = MockModel()
        
        # 測試 callback
        callback = MyCustomCallback()
        callback.on_train_start(trainer, model)
        
        # 驗證行為
        self.assertEqual(callback.state['initialized'], True)

# ? 錯誤示例
def on_train_epoch_end(self, trainer, pl_module):
    val_loss = trainer.callback_metrics['val_loss']  # KeyError!

# ? 正確示例
def on_validation_epoch_end(self, trainer, pl_module):
    # 在驗證后訪問
    val_loss = trainer.callback_metrics.get('val_loss')
    if val_loss is not None:
        print(f"驗證損失: {val_loss}")

# ? 錯誤示例
def on_validation_start(self, trainer, pl_module):
    pl_module.train()  # 錯誤地切換到訓練模式

# ? 正確示例
def on_validation_start(self, trainer, pl_module):
    # Lightning 會自動處理模式切換，無需手動干預(yù)
    pass

# ? 錯誤示例（會嚴重拖慢訓練）
def on_train_batch_end(self, trainer, pl_module, outputs, batch, batch_idx):
    # 每個 batch 都執(zhí)行復(fù)雜計算
    expensive_operation()

# ? 正確示例
def on_train_epoch_end(self, trainer, pl_module):
    # 每個 epoch 執(zhí)行一次
    expensive_operation()

class DebugCallback(Callback):
    def on_validation_epoch_end(self, trainer, pl_module):
        print("\n可用指標:")
        for key, value in trainer.callback_metrics.items():
            print(f"  {key}: {value}")

class TestCallback(Callback):
    def __init__(self):
        super().__init__()
        self.call_count = {}
    
    def _log_call(self, method_name):
        self.call_count[method_name] = self.call_count.get(method_name, 0) + 1
        print(f"[{method_name}] 被調(diào)用 {self.call_count[method_name]} 次")
    
    def on_train_start(self, trainer, pl_module):
        self._log_call('on_train_start')
    
    def on_train_epoch_end(self, trainer, pl_module):
        self._log_call('on_train_epoch_end')

class DebugCallback(Callback):
    def on_validation_epoch_end(self, trainer, pl_module):
        # 在特定條件下觸發(fā)斷點
        if trainer.current_epoch == 5:
            import pdb; pdb.set_trace()

# ? 低效
class BadCallback(Callback):
    def on_train_batch_end(self, trainer, pl_module, outputs, batch, batch_idx):
        # 每個 batch 都寫文件
        with open('log.txt', 'a') as f:
            f.write(f"Batch {batch_idx} done\n")

# ? 高效
class GoodCallback(Callback):
    def __init__(self):
        super().__init__()
        self.buffer = []
    
    def on_train_batch_end(self, trainer, pl_module, outputs, batch, batch_idx):
        self.buffer.append(f"Batch {batch_idx} done\n")
    
    def on_train_epoch_end(self, trainer, pl_module):
        # 每個 epoch 寫一次
        with open('log.txt', 'a') as f:
            f.writelines(self.buffer)
        self.buffer = []

class OptimizedCallback(Callback):
    def __init__(self, log_every_n_epochs=5):
        super().__init__()
        self.log_every_n_epochs = log_every_n_epochs
    
    def on_validation_epoch_end(self, trainer, pl_module):
        # 僅每 5 個 epoch 執(zhí)行一次
        if trainer.current_epoch % self.log_every_n_epochs == 0:
            expensive_visualization()

class CustomTrainLoop(Callback):
    """完全自定義訓練循環(huán)"""
    
    def on_train_batch_start(self, trainer, pl_module, batch, batch_idx):
        # 自定義數(shù)據(jù)預(yù)處理
        pass
    
    def on_before_backward(self, trainer, pl_module, loss):
        # 自定義損失縮放
        pass
    
    def on_before_optimizer_step(self, trainer, pl_module, optimizer):
        # 自定義梯度處理
        pass

callbacks = [
    ModelCheckpoint(monitor='val_loss', mode='min', save_top_k=1),
    EarlyStopping(monitor='val_loss', patience=10),
    LearningRateMonitor(),
]

class MyCallback(Callback):
    def on_train_epoch_end(self, trainer, pl_module):
        metrics = trainer.callback_metrics
        # 自定義邏輯