企业官网建设流程全解析

【Bug已解决】What is running loss in PyTorch and how is it calculated 解决方案

问题描述

在 PyTorch 训练循环中，经常会看到 "running loss" 这个概念。很多初学者对 running loss 的含义、计算方式和正确使用方法感到困惑。

常见的代码模式：

running_loss = 0.0 for batch_idx, (data, target) in enumerate(dataloader): optimizer.zero_grad() output = model(data) loss = criterion(output, target) loss.backward() optimizer.step() running_loss += loss.item() if batch_idx % 100 == 99: print(f"Loss: {running_loss / 100:.4f}") running_loss = 0.0

常见的问题：

1. running loss 到底是什么？
2. 为什么要除以 batch 数量或样本数量？
3. loss.item()和loss有什么区别？
4. running loss 应该在什么时候重置？
5. 如何正确计算 epoch 级别的平均 loss？

错误复现

import torch import torch.nn as nn # 模拟训练数据 model = nn.Linear(10, 2) criterion = nn.CrossEntropyLoss() optimizer = torch.optim.SGD(model.parameters(), lr=0.01) data = torch.randn(100, 10) targets = torch.randint(0, 2, (100,)) # 复现1：不使用 .item() 导致内存泄漏 print("=" * 50) print("复现1：不使用 .item() 的内存问题") print("=" * 50) running_loss = 0.0 for i in range(5): optimizer.zero_grad() output = model(data) loss = criterion(output, targets) loss.backward() optimizer.step() # 错误：累加 loss 而不是 loss.item() running_loss += loss # 这会保留计算图！ print(f"Step {i}: loss={loss.item():.4f}, running_loss type={type(running_loss)}") # running_loss 变成了一个巨大的计算图 print(f"\nFinal running_loss: {running_loss}") print(f"Requires grad: {running_loss.requires_grad}")

# 复现2：错误的平均计算 print("\n" + "=" * 50) print("复现2：错误的平均计算") print("=" * 50) # 模拟不同大小的 batch batch_losses = [1.0, 2.0, 3.0, 4.0, 5.0] batch_sizes = [32, 32, 32, 16, 8] # 最后两个 batch 较小 # 错误1：简单平均（不考虑 batch 大小） simple_avg = sum(batch_losses) / len(batch_losses) print(f"Simple average: {simple_avg:.4f}") # 错误2：加权平均（按 batch 大小加权） weighted_sum = sum(l * s for l, s in zip(batch_losses, batch_sizes)) total_samples = sum(batch_sizes) weighted_avg = weighted_sum / total_samples print(f"Weighted average: {weighted_avg:.4f}") # 正确：按样本数加权 print(f"\nNote: Simple avg may be misleading when batch sizes differ")

# 复现3：running_loss 重置时机错误 print("\n" + "=" * 50) print("复现3：重置时机错误") print("=" * 50) running_loss = 0.0 batch_count = 0 for i in range(250): # 模拟 loss loss_val = 1.0 + i * 0.01 running_loss += loss_val batch_count += 1 # 错误：在 batch_idx % 100 == 0 时打印（第0个batch就打印了） if i % 100 == 0: print(f"Batch {i}: running_loss={running_loss/batch_count:.4f}") running_loss = 0.0 batch_count = 0 # 问题：第0个 batch 时 running_loss 只有1个 batch 的值

根因分析

1. running loss 的本质

running loss 是一个累加器，用于在训练过程中跟踪累计的损失值。它的主要目的是：

1. 定期打印：避免每个 batch 都打印（太频繁），而是每隔 N 个 batch 打印一次平均值
2. 监控训练：通过观察 loss 的变化趋势来判断训练是否正常
3. 计算 epoch loss：在整个 epoch 结束后计算平均损失

# running loss 的工作流程 running_loss = 0.0 # 初始化累加器 for batch_idx, (data, target) in enumerate(dataloader): # 前向传播 + 计算损失 loss = criterion(model(data), target) # 反向传播 + 优化 loss.backward() optimizer.step() # 累加 loss（使用 .item() 转为 Python 标量） running_loss += loss.item() # 每100个 batch 打印一次 if batch_idx % 100 == 99: avg_loss = running_loss / 100 # 计算平均值 print(f"Batch {batch_idx}: avg_loss = {avg_loss:.4f}") running_loss = 0.0 # 重置累加器

2. loss.item() 的重要性

loss是一个 PyTorch 张量，它包含了计算图的信息。如果直接累加loss而不调用.item()，会导致：

1. 内存泄漏：每个 loss 都保留计算图，累加后形成巨大的计算图
2. 显存爆炸：GPU 显存被计算图占满
3. 性能下降：不必要的计算图维护

# loss 是一个张量 loss = criterion(output, target) print(f"Type: {type(loss)}") # <class 'torch.Tensor'> print(f"Shape: {loss.shape}") # torch.Size([]) print(f"Requires grad: {loss.requires_grad}") # True # .item() 提取标量值，断开计算图 loss_value = loss.item() print(f"Type: {type(loss_value)}") # <class 'float'>

3. 平均 loss 的正确计算

当 batch 大小不一致时（最后一个 batch 可能较小），需要按样本数加权：

# 方法1：简单平均（batch 大小一致时） total_loss = 0.0 num_batches = 0 for data, target in dataloader: loss = criterion(model(data), target) total_loss += loss.item() num_batches += 1 avg_loss = total_loss / num_batches # 方法2：按样本数加权（batch 大小不一致时） total_loss = 0.0 total_samples = 0 for data, target in dataloader: batch_size = data.size(0) loss = criterion(model(data), target) total_loss += loss.item() * batch_size # 加权 total_samples += batch_size avg_loss = total_loss / total_samples

4. CrossEntropyLoss 的 reduction

PyTorch 的损失函数默认使用reduction='mean'，即对 batch 内所有样本的 loss 取平均：

# reduction='mean'（默认）：batch 内平均 criterion = nn.CrossEntropyLoss(reduction='mean') loss = criterion(output, target) # 标量，batch 内平均 # reduction='sum'：batch 内求和 criterion = nn.CrossEntropyLoss(reduction='sum') loss = criterion(output, target) # 标量，batch 内总和 # reduction='none'：不 reduction，返回每个样本的 loss criterion = nn.CrossEntropyLoss(reduction='none') loss = criterion(output, target) # [batch_size]，每个样本的 loss

解决方案

方案一：标准的 running loss 实现

import torch import torch.nn as nn def train_one_epoch(model, dataloader, criterion, optimizer, device='cpu', print_every=100): """标准的训练 epoch""" model.train() running_loss = 0.0 correct = 0 total = 0 for batch_idx, (data, target) in enumerate(dataloader): data, target = data.to(device), target.to(device) optimizer.zero_grad() output = model(data) loss = criterion(output, target) loss.backward() optimizer.step() # 使用 .item() 获取标量值 running_loss += loss.item() # 计算准确率 _, predicted = output.max(1) total += target.size(0) correct += predicted.eq(target).sum().item() # 定期打印 if batch_idx % print_every == print_every - 1: avg_loss = running_loss / print_every accuracy = 100. * correct / total print(f" Batch {batch_idx + 1}: Loss={avg_loss:.4f}, Acc={accuracy:.2f}%") running_loss = 0.0 # 返回 epoch 级别的统计 return running_loss # 使用示例 model = nn.Sequential(nn.Linear(10, 2)) criterion = nn.CrossEntropyLoss() optimizer = torch.optim.SGD(model.parameters(), lr=0.01) # 模拟数据 from torch.utils.data import TensorDataset, DataLoader dataset = TensorDataset(torch.randn(1000, 10), torch.randint(0, 2, (1000,))) dataloader = DataLoader(dataset, batch_size=32, shuffle=True) train_one_epoch(model, dataloader, criterion, optimizer, print_every=10)

方案二：完整的训练统计跟踪

import torch import torch.nn as nn from collections import defaultdict class TrainingStats: """训练统计跟踪器""" def __init__(self): self.history = defaultdict(list) self.running_stats = defaultdict(float) self.counts = defaultdict(int) def update(self, batch_stats: dict, batch_size: int): """更新统计""" for key, value in batch_stats.items(): if key.endswith('_loss'): # loss 按样本数加权 self.running_stats[key] += value * batch_size self.counts[key] += batch_size elif key.endswith('_acc') or key.endswith('_correct'): # 准确率按样本数加权 self.running_stats[key] += value * batch_size self.counts[key] += batch_size else: self.running_stats[key] += value self.counts[key] += 1 def get averages(self) -> dict: """获取当前平均值""" return { key: self.running_stats[key] / max(self.counts[key], 1) for key in self.running_stats } def reset(self): """重置 running 统计""" self.running_stats.clear() self.counts.clear() def record_epoch(self, epoch: int): """记录 epoch 级别的统计""" for key in self.running_stats: avg = self.running_stats[key] / max(self.counts[key], 1) self.history[key].append(avg) self.reset() # 使用示例 stats = TrainingStats() for epoch in range(5): model.train() stats.reset() for data, target in dataloader: optimizer.zero_grad() output = model(data) loss = criterion(output, target) loss.backward() optimizer.step() # 计算准确率 _, predicted = output.max(1) correct = predicted.eq(target).sum().item() accuracy = 100. * correct / target.size(0) # 更新统计 stats.update({ 'train_loss': loss.item(), 'train_acc': accuracy, }, batch_size=target.size(0)) stats.record_epoch(epoch) avgs = {k: v[-1] for k, v in stats.history.items()} print(f"Epoch {epoch}: {avgs}")

方案三：使用 tqdm 进度条显示 running loss

import torch import torch.nn as nn from tqdm import tqdm def train_with_tqdm(model, dataloader, criterion, optimizer, num_epochs=5): """使用 tqdm 显示训练进度和 running loss""" for epoch in range(num_epochs): model.train() running_loss = 0.0 correct = 0 total = 0 # 使用 tqdm 包装 dataloader pbar = tqdm(dataloader, desc=f"Epoch {epoch+1}/{num_epochs}") for data, target in pbar: optimizer.zero_grad() output = model(data) loss = criterion(output, target) loss.backward() optimizer.step() running_loss += loss.item() _, predicted = output.max(1) total += target.size(0) correct += predicted.eq(target).sum().item() # 更新进度条信息 pbar.set_postfix({ 'loss': f"{running_loss / (pbar.n + 1):.4f}", 'acc': f"{100. * correct / total:.2f}%", }) # Epoch 结束 epoch_loss = running_loss / len(dataloader) epoch_acc = 100. * correct / total print(f"Epoch {epoch+1} finished: Loss={epoch_loss:.4f}, Acc={epoch_acc:.2f}%") # 使用示例 train_with_tqdm(model, dataloader, criterion, optimizer, num_epochs=3)

方案四：TensorBoard 记录 running loss

import torch import torch.nn as nn from torch.utils.tensorboard import SummaryWriter def train_with_tensorboard(model, dataloader, criterion, optimizer, num_epochs=5, log_dir='runs/experiment'): """使用 TensorBoard 记录训练过程""" writer = SummaryWriter(log_dir) global_step = 0 for epoch in range(num_epochs): model.train() running_loss = 0.0 correct = 0 total = 0 for batch_idx, (data, target) in enumerate(dataloader): optimizer.zero_grad() output = model(data) loss = criterion(output, target) loss.backward() optimizer.step() running_loss += loss.item() _, predicted = output.max(1) total += target.size(0) correct += predicted.eq(target).sum().item() # 每100个 batch 记录一次 if batch_idx % 100 == 99: avg_loss = running_loss / 100 avg_acc = 100. * correct / total # 记录到 TensorBoard writer.add_scalar('training/loss', avg_loss, global_step) writer.add_scalar('training/accuracy', avg_acc, global_step) running_loss = 0.0 global_step += 1 # 记录 epoch 级别的统计 epoch_loss = running_loss / len(dataloader) epoch_acc = 100. * correct / total writer.add_scalar('epoch/loss', epoch_loss, epoch) writer.add_scalar('epoch/accuracy', epoch_acc, epoch) writer.close() # 使用示例 # train_with_tensorboard(model, dataloader, criterion, optimizer)

完整修复代码

""" 完整修复代码：正确计算和使用 running loss 实现完整的训练循环和统计跟踪 """ import torch import torch.nn as nn from torch.utils.data import DataLoader, TensorDataset from typing import Dict, List, Optional from collections import defaultdict class MetricTracker: """指标跟踪器""" def __init__(self): self.metrics = defaultdict(list) self.running_sum = defaultdict(float) self.running_count = defaultdict(int) self.last_value = defaultdict(float) def update(self, metrics: Dict[str, float], batch_size: int = 1, weighted: bool = True): """更新指标""" for name, value in metrics.items(): self.last_value[name] = value if weighted: self.running_sum[name] += value * batch_size self.running_count[name] += batch_size else: self.running_sum[name] += value self.running_count[name] += 1 def get_average(self, name: str) -> float: """获取指定指标的运行平均值""" if self.running_count[name] == 0: return 0.0 return self.running_sum[name] / self.running_count[name] def get_all_averages(self) -> Dict[str, float]: """获取所有指标的运行平均值""" return { name: self.get_average(name) for name in self.running_sum } def reset(self): """重置运行统计""" self.running_sum.clear() self.running_count.clear() def record_epoch(self, epoch: int): """记录 epoch 结束时的平均值""" for name in self.running_sum: avg = self.get_average(name) self.metrics[name].append(avg) self.reset() def get_history(self, name: str) -> List[float]: """获取指标历史""" return self.metrics.get(name, []) class Trainer: """完整的训练器""" def __init__(self, model, criterion, optimizer, device='cpu'): self.model = model.to(device) self.criterion = criterion self.optimizer = optimizer self.device = device self.train_metrics = MetricTracker() self.val_metrics = MetricTracker() def train_epoch(self, dataloader, epoch, print_every=50): """训练一个 epoch""" self.model.train() self.train_metrics.reset() for batch_idx, (data, target) in enumerate(dataloader): data, target = data.to(self.device), target.to(self.device) batch_size = data.size(0) # 前向传播 self.optimizer.zero_grad() output = self.model(data) loss = self.criterion(output, target) # 反向传播 loss.backward() self.optimizer.step() # 计算指标 with torch.no_grad(): _, predicted = output.max(1) correct = predicted.eq(target).sum().item() accuracy = 100. * correct / batch_size # 更新指标（使用 .item() 避免内存泄漏） self.train_metrics.update({ 'loss': loss.item(), 'accuracy': accuracy, }, batch_size=batch_size) # 定期打印 if batch_idx % print_every == print_every - 1: avgs = self.train_metrics.get_all_averages() print(f" Epoch {epoch} | Batch {batch_idx + 1:5d}/{len(dataloader)} | " f"Loss: {avgs['loss']:.4f} | Acc: {avgs['accuracy']:.2f}%") # 记录 epoch 统计 self.train_metrics.record_epoch(epoch) return self.train_metrics.get_all_averages() def validate(self, dataloader, epoch): """验证""" self.model.eval() self.val_metrics.reset() with torch.no_grad(): for data, target in dataloader: data, target = data.to(self.device), target.to(self.device) batch_size = data.size(0) output = self.model(data) loss = self.criterion(output, target) _, predicted = output.max(1) correct = predicted.eq(target).sum().item() accuracy = 100. * correct / batch_size self.val_metrics.update({ 'loss': loss.item(), 'accuracy': accuracy, }, batch_size=batch_size) self.val_metrics.record_epoch(epoch) return self.val_metrics.get_all_averages() def fit(self, train_loader, val_loader, num_epochs=10, print_every=50): """完整训练""" print("=" * 70) print("Starting Training") print("=" * 70) for epoch in range(1, num_epochs + 1): # 训练 train_avgs = self.train_epoch(train_loader, epoch, print_every) # 验证 val_avgs = self.validate(val_loader, epoch) # 打印 epoch 结果 print(f"\nEpoch {epoch}/{num_epochs} Summary:") print(f" Train - Loss: {train_avgs['loss']:.4f}, Acc: {train_avgs['accuracy']:.2f}%") print(f" Val - Loss: {val_avgs['loss']:.4f}, Acc: {val_avgs['accuracy']:.2f}%") print("-" * 70) print("Training Complete!") return self.train_metrics.metrics, self.val_metrics.metrics def test_running_loss(): """测试 running loss 的正确计算""" print("=" * 60) print("Testing Running Loss Calculation") print("=" * 60) # 创建数据 torch.manual_seed(42) train_data = torch.randn(500, 10) train_targets = torch.randint(0, 3, (500,)) val_data = torch.randn(100, 10) val_targets = torch.randint(0, 3, (100,)) train_dataset = TensorDataset(train_data, train_targets) val_dataset = TensorDataset(val_data, val_targets) train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True) val_loader = DataLoader(val_dataset, batch_size=32, shuffle=False) # 创建模型 model = nn.Sequential( nn.Linear(10, 64), nn.ReLU(), nn.Linear(64, 3), ) criterion = nn.CrossEntropyLoss() optimizer = torch.optim.Adam(model.parameters(), lr=0.001) # 训练 trainer = Trainer(model, criterion, optimizer) train_history, val_history = trainer.fit( train_loader, val_loader, num_epochs=5, print_every=5 ) # 打印历史 print("\nTraining History:") print(f"{'Epoch':>5} | {'Train Loss':>10} | {'Train Acc':>9} | {'Val Loss':>8} | {'Val Acc':>7}") print("-" * 50) for i in range(5): print(f"{i+1:5d} | {train_history['loss'][i]:10.4f} | {train_history['accuracy'][i]:8.2f}% | " f"{val_history['loss'][i]:8.4f} | {val_history['accuracy'][i]:6.2f}%") print("\n" + "=" * 60) print("All tests passed!") print("=" * 60) if __name__ == "__main__": test_running_loss()

常见陷阱与注意事项

1. 必须使用 .item()

# 错误：不使用 .item() 导致内存泄漏 running_loss = 0.0 for data, target in dataloader: loss = criterion(model(data), target) running_loss += loss # 保留计算图！ loss.backward() # running_loss 变得越来越大，包含整个计算历史 # 正确：使用 .item() 提取标量 running_loss = 0.0 for data, target in dataloader: loss = criterion(model(data), target) running_loss += loss.item() # 只保存标量值 loss.backward()

2. batch 大小不一致时的平均

# 当最后一个 batch 较小时 # DataLoader 默认 drop_last=False，最后一个 batch 可能较小 # 错误：简单平均 total_loss = sum(losses) / len(losses) # 不考虑 batch 大小 # 正确：按样本数加权 total_loss = sum(l * s for l, s in zip(losses, batch_sizes)) / sum(batch_sizes)

3. 训练和验证的 running loss

# 训练时：每个 batch 都更新参数 for data, target in train_loader: optimizer.zero_grad() loss = criterion(model(data), target) loss.backward() optimizer.step() train_loss += loss.item() # 验证时：不更新参数，使用 torch.no_grad() with torch.no_grad(): for data, target in val_loader: loss = criterion(model(data), target) val_loss += loss.item()

4. running_loss 的重置时机

# 方式1：每 N 个 batch 重置（用于定期打印） running_loss = 0.0 for batch_idx, (data, target) in enumerate(dataloader): # ... running_loss += loss.item() if batch_idx % 100 == 99: # 第100, 200, 300...个batch print(f"Loss: {running_loss / 100:.4f}") running_loss = 0.0 # 方式2：每个 epoch 重置（用于计算 epoch 平均 loss） running_loss = 0.0 for data, target in dataloader: # ... running_loss += loss.item() epoch_loss = running_loss / len(dataloader)

5. 多 GPU 下的 loss

# 使用 DataParallel 或 DistributedDataParallel 时 # loss 会被平均到各个 GPU model = nn.DataParallel(model) loss = criterion(model(data), target) # loss 已经是所有 GPU 上的平均 running_loss += loss.item()

总结

running loss 是训练过程中跟踪损失变化的重要工具。正确使用它需要注意以下几点：

1. 使用.item()：将 loss 张量转为 Python 标量，避免保留计算图导致内存泄漏。
2. 正确计算平均：当 batch 大小不一致时，按样本数加权计算平均 loss。
3. 合理的重置时机：定期打印时每 N 个 batch 重置，计算 epoch loss 时每个 epoch 重置。
4. 区分训练和验证：训练时更新参数，验证时使用torch.no_grad()不更新参数。
5. 使用 MetricTracker：封装统计跟踪逻辑，避免手动管理的错误。

最佳实践：

• 始终使用loss.item()获取标量值
• 按样本数加权计算平均 loss
• 使用 tqdm 或 TensorBoard 可视化训练过程
• 封装训练逻辑到 Trainer 类中
• 定期保存和检查训练统计

通过本文的详细分析和完整代码示例，你应该能够彻底理解 running loss 的概念和正确计算方法，在实际项目中构建可靠的训练循环和统计跟踪系统。