pytorch-patterns

PyTorch 開發模式

符合慣例的 PyTorch 模式與最佳實踐，用於建立穩健、高效且可重現的深度學習應用。

啟用時機

• 撰寫新的 PyTorch 模型或訓練腳本
• 審查深度學習程式碼
• 除錯訓練迴圈或資料管線
• 最佳化 GPU 記憶體使用率或訓練速度
• 設定可重現的實驗環境

核心原則

1. 裝置無關程式碼

始終撰寫能在 CPU 與 GPU 上運作，且無需硬編碼裝置的程式碼。

# 良好：裝置無關
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = MyModel().to(device)
data = data.to(device)

# 不佳：硬編碼裝置
model = MyModel().cuda()  # 若無 GPU 則會崩潰
data = data.cuda()

2. 可重現性優先

設定所有隨機種子以確保可重現的結果。

# 良好：完整的可重現性設定
def set_seed(seed: int = 42) -> None:
    torch.manual_seed(seed)
    torch.cuda.manual_seed_all(seed)
    np.random.seed(seed)
    random.seed(seed)
    torch.backends.cudnn.deterministic = True
    torch.backends.cudnn.benchmark = False

# 不佳：無種子控制
model = MyModel()  # 每次執行權重都不同

3. 明確的形狀管理

始終記錄並驗證張量形狀。

# 良好：註解形狀的 forward 傳遞
def forward(self, x: torch.Tensor) -> torch.Tensor:
    # x: (batch_size, channels, height, width)
    x = self.conv1(x)    # -> (batch_size, 32, H, W)
    x = self.pool(x)     # -> (batch_size, 32, H//2, W//2)
    x = x.view(x.size(0), -1)  # -> (batch_size, 32*H//2*W//2)
    return self.fc(x)    # -> (batch_size, num_classes)

# 不佳：無形狀追蹤
def forward(self, x):
    x = self.conv1(x)
    x = self.pool(x)
    x = x.view(x.size(0), -1)  # 這到底是什麼形狀？
    return self.fc(x)           # 這樣真的能運作嗎？

模型架構模式

清晰的 nn.Module 結構

# 良好：組織良好的模組
class ImageClassifier(nn.Module):
    def __init__(self, num_classes: int, dropout: float = 0.5) -> None:
        super().__init__()
        self.features = nn.Sequential(
            nn.Conv2d(3, 64, kernel_size=3, padding=1),
            nn.BatchNorm2d(64),
            nn.ReLU(inplace=True),
            nn.MaxPool2d(2),
        )
        self.classifier = nn.Sequential(
            nn.Dropout(dropout),
            nn.Linear(64 * 16 * 16, num_classes),
        )

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        x = self.features(x)
        x = x.view(x.size(0), -1)
        return self.classifier(x)

# 不佳：所有邏輯都在 forward 中
class ImageClassifier(nn.Module):
    def __init__(self):
        super().__init__()

    def forward(self, x):
        x = F.conv2d(x, weight=self.make_weight())  # 每次呼叫都建立權重！
        return x

正確的權重初始化

# 良好：明確的初始化
def _init_weights(self, module: nn.Module) -> None:
    if isinstance(module, nn.Linear):
        nn.init.kaiming_normal_(module.weight, mode="fan_out", nonlinearity="relu")
        if module.bias is not None:
            nn.init.zeros_(module.bias)
    elif isinstance(module, nn.Conv2d):
        nn.init.kaiming_normal_(module.weight, mode="fan_out", nonlinearity="relu")
    elif isinstance(module, nn.BatchNorm2d):
        nn.init.ones_(module.weight)
        nn.init.zeros_(module.bias)

model = MyModel()
model.apply(model._init_weights)

訓練迴圈模式

標準訓練迴圈

# 良好：包含最佳實踐的完整訓練迴圈
def train_one_epoch(
    model: nn.Module,
    dataloader: DataLoader,
    optimizer: torch.optim.Optimizer,
    criterion: nn.Module,
    device: torch.device,
    scaler: torch.amp.GradScaler | None = None,
) -> float:
    model.train()  # 始終設定為訓練模式
    total_loss = 0.0

    for batch_idx, (data, target) in enumerate(dataloader):
        data, target = data.to(device), target.to(device)

        optimizer.zero_grad(set_to_none=True)  # 比 zero_grad() 更有效率

        # 混合精度訓練
        with torch.amp.autocast("cuda", enabled=scaler is not None):
            output = model(data)
            loss = criterion(output, target)

        if scaler is not None:
            scaler.scale(loss).backward()
            scaler.unscale_(optimizer)
            torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
            scaler.step(optimizer)
            scaler.update()
        else:
            loss.backward()
            torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
            optimizer.step()

        total_loss += loss.item()

    return total_loss / len(dataloader)

驗證迴圈

# 良好：正確的評估
@torch.no_grad()  # 比包在 torch.no_grad() 區塊中更有效率
def evaluate(
    model: nn.Module,
    dataloader: DataLoader,
    criterion: nn.Module,
    device: torch.device,
) -> tuple[float, float]:
    model.eval()  # 始終設定為評估模式 — 停用 dropout，使用執行中的 BN 統計量
    total_loss = 0.0
    correct = 0
    total = 0

    for data, target in dataloader:
        data, target = data.to(device), target.to(device)
        output = model(data)
        total_loss += criterion(output, target).item()
        correct += (output.argmax(1) == target).sum().item()
        total += target.size(0)

    return total_loss / len(dataloader), correct / total

資料管線模式

自訂資料集

# 良好：帶有型別提示的簡潔 Dataset
class ImageDataset(Dataset):
    def __init__(
        self,
        image_dir: str,
        labels: dict[str, int],
        transform: transforms.Compose | None = None,
    ) -> None:
        self.image_paths = list(Path(image_dir).glob("*.jpg"))
        self.labels = labels
        self.transform = transform

    def __len__(self) -> int:
        return len(self.image_paths)

    def __getitem__(self, idx: int) -> tuple[torch.Tensor, int]:
        img = Image.open(self.image_paths[idx]).convert("RGB")
        label = self.labels[self.image_paths[idx].stem]

        if self.transform:
            img = self.transform(img)

        return img, label

高效率的 DataLoader 設定

# 良好：最佳化的 DataLoader
dataloader = DataLoader(
    dataset,
    batch_size=32,
    shuffle=True,            # 訓練時打亂
    num_workers=4,           # 平行資料載入
    pin_memory=True,         # 更快的 CPU->GPU 傳輸
    persistent_workers=True, # 在各 epoch 之間保持 workers 存活
    drop_last=True,          # 一致的批次大小（適用於 BatchNorm）
)

# 不佳：慢速預設值
dataloader = DataLoader(dataset, batch_size=32)  # num_workers=0, 無 pin_memory

為可變長度資料自訂 collate

# 良好：在 collate_fn 中填充序列
def collate_fn(batch: list[tuple[torch.Tensor, int]]) -> tuple[torch.Tensor, torch.Tensor]:
    sequences, labels = zip(*batch)
    # 填充至批次中的最大長度
    padded = nn.utils.rnn.pad_sequence(sequences, batch_first=True, padding_value=0)
    return padded, torch.tensor(labels)

dataloader = DataLoader(dataset, batch_size=32, collate_fn=collate_fn)

檢查點模式

儲存與載入檢查點

# 良好：包含所有訓練狀態的完整檢查點
def save_checkpoint(
    model: nn.Module,
    optimizer: torch.optim.Optimizer,
    epoch: int,
    loss: float,
    path: str,
) -> None:
    torch.save({
        "epoch": epoch,
        "model_state_dict": model.state_dict(),
        "optimizer_state_dict": optimizer.state_dict(),
        "loss": loss,
    }, path)

def load_checkpoint(
    path: str,
    model: nn.Module,
    optimizer: torch.optim.Optimizer | None = None,
) -> dict:
    checkpoint = torch.load(path, map_location="cpu", weights_only=True)
    model.load_state_dict(checkpoint["model_state_dict"])
    if optimizer:
        optimizer.load_state_dict(checkpoint["optimizer_state_dict"])
    return checkpoint

# 不佳：僅儲存模型權重（無法恢復訓練）
torch.save(model.state_dict(), "model.pt")

效能最佳化

混合精度訓練

# 良好：搭配 GradScaler 的 AMP
scaler = torch.amp.GradScaler("cuda")
for data, target in dataloader:
    with torch.amp.autocast("cuda"):
        output = model(data)
        loss = criterion(output, target)
    scaler.scale(loss).backward()
    scaler.step(optimizer)
    scaler.update()
    optimizer.zero_grad(set_to_none=True)

大型模型的梯度檢查點

# 良好：以計算換取記憶體
from torch.utils.checkpoint import checkpoint

class LargeModel(nn.Module):
    def forward(self, x: torch.Tensor) -> torch.Tensor:
        # 在反向傳播期間重新計算激活值以節省記憶體
        x = checkpoint(self.block1, x, use_reentrant=False)
        x = checkpoint(self.block2, x, use_reentrant=False)
        return self.head(x)

torch.compile 提升速度

# 良好：編譯模型以獲得更快的執行速度（PyTorch 2.0+）
model = MyModel().to(device)
model = torch.compile(model, mode="reduce-overhead")

# 模式："default"（安全）、"reduce-overhead"（更快）、"max-autotune"（最快）

快速參考：PyTorch 慣用語

慣用語	描述
model.train() / model.eval()	在訓練/評估前始終設定模式
torch.no_grad()	停用推理的梯度
optimizer.zero_grad(set_to_none=True)	更有效率的梯度清除
.to(device)	裝置無關的張量/模型放置
torch.amp.autocast	混合精度，速度提升 2 倍
pin_memory=True	更快的 CPU→GPU 資料傳輸
torch.compile	用於加速的 JIT 編譯 (2.0+)
weights_only=True	安全的模型載入
torch.manual_seed	可重現的實驗
gradient_checkpointing	以計算換取記憶體

應避免的反模式

# 不佳：驗證時忘記設定 model.eval()
model.train()
with torch.no_grad():
    output = model(val_data)  # Dropout 仍然啟用！BatchNorm 使用批次統計量！

# 良好：始終設定評估模式
model.eval()
with torch.no_grad():
    output = model(val_data)

# 不佳：會破壞 autograd 的原地操作
x = F.relu(x, inplace=True)  # 可能破壞梯度計算
x += residual                  # 原地加法破壞 autograd 圖

# 良好：非原地操作
x = F.relu(x)
x = x + residual

# 不佳：在訓練迴圈中重複將資料移至 GPU
for data, target in dataloader:
    model = model.cuda()  # 每次迭代都移動模型！

# 良好：在迴圈前僅移動模型一次
model = model.to(device)
for data, target in dataloader:
    data, target = data.to(device), target.to(device)

# 不佳：在 backward 之前使用 .item()
loss = criterion(output, target).item()  # 從圖中分離！
loss.backward()  # 錯誤：無法透過 .item() 反向傳播

# 良好：僅在記錄時呼叫 .item()
loss = criterion(output, target)
loss.backward()
print(f"Loss: {loss.item():.4f}")  # backward 之後使用 .item() 沒問題

# 不佳：未正確使用 torch.save
torch.save(model, "model.pt")  # 儲存整個模型（脆弱、不可移植）

# 良好：儲存 state_dict
torch.save(model.state_dict(), "model.pt")

切記：PyTorch 程式碼應具有裝置無關性、可重現性，並注重記憶體使用。如有疑問，請使用 torch.profiler 進行效能分析，並使用 torch.cuda.memory_summary() 檢查 GPU 記憶體。