【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(fLoss: {running_loss / 100:.4f}) running_loss 0.0常见的问题running loss 到底是什么为什么要除以 batch 数量或样本数量loss.item()和loss有什么区别running loss 应该在什么时候重置如何正确计算 epoch 级别的平均 loss错误复现import torch import torch.nn as nn # 模拟训练数据 model nn.Linear(10, 2) criterion nn.CrossEntropyLoss() optimizer torch.optim.SGD(model.parameters(), lr0.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(fStep {i}: loss{loss.item():.4f}, running_loss type{type(running_loss)}) # running_loss 变成了一个巨大的计算图 print(f\nFinal running_loss: {running_loss}) print(fRequires 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(fSimple 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(fWeighted average: {weighted_avg:.4f}) # 正确按样本数加权 print(f\nNote: Simple avg may be misleading when batch sizes differ)# 复现3running_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(fBatch {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 是一个累加器用于在训练过程中跟踪累计的损失值。它的主要目的是定期打印避免每个 batch 都打印太频繁而是每隔 N 个 batch 打印一次平均值监控训练通过观察 loss 的变化趋势来判断训练是否正常计算 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(fBatch {batch_idx}: avg_loss {avg_loss:.4f}) running_loss 0.0 # 重置累加器2. loss.item() 的重要性loss是一个 PyTorch 张量它包含了计算图的信息。如果直接累加loss而不调用.item()会导致内存泄漏每个 loss 都保留计算图累加后形成巨大的计算图显存爆炸GPU 显存被计算图占满性能下降不必要的计算图维护# loss 是一个张量 loss criterion(output, target) print(fType: {type(loss)}) # class torch.Tensor print(fShape: {loss.shape}) # torch.Size([]) print(fRequires grad: {loss.requires_grad}) # True # .item() 提取标量值断开计算图 loss_value loss.item() print(fType: {type(loss_value)}) # class float3. 平均 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_samples4. CrossEntropyLoss 的 reductionPyTorch 的损失函数默认使用reductionmean即对 batch 内所有样本的 loss 取平均# reductionmean默认batch 内平均 criterion nn.CrossEntropyLoss(reductionmean) loss criterion(output, target) # 标量batch 内平均 # reductionsumbatch 内求和 criterion nn.CrossEntropyLoss(reductionsum) loss criterion(output, target) # 标量batch 内总和 # reductionnone不 reduction返回每个样本的 loss criterion nn.CrossEntropyLoss(reductionnone) loss criterion(output, target) # [batch_size]每个样本的 loss解决方案方案一标准的 running loss 实现import torch import torch.nn as nn def train_one_epoch(model, dataloader, criterion, optimizer, devicecpu, print_every100): 标准的训练 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(), lr0.01) # 模拟数据 from torch.utils.data import TensorDataset, DataLoader dataset TensorDataset(torch.randn(1000, 10), torch.randint(0, 2, (1000,))) dataloader DataLoader(dataset, batch_size32, shuffleTrue) train_one_epoch(model, dataloader, criterion, optimizer, print_every10)方案二完整的训练统计跟踪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_sizetarget.size(0)) stats.record_epoch(epoch) avgs {k: v[-1] for k, v in stats.history.items()} print(fEpoch {epoch}: {avgs})方案三使用 tqdm 进度条显示 running lossimport torch import torch.nn as nn from tqdm import tqdm def train_with_tqdm(model, dataloader, criterion, optimizer, num_epochs5): 使用 tqdm 显示训练进度和 running loss for epoch in range(num_epochs): model.train() running_loss 0.0 correct 0 total 0 # 使用 tqdm 包装 dataloader pbar tqdm(dataloader, descfEpoch {epoch1}/{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(fEpoch {epoch1} finished: Loss{epoch_loss:.4f}, Acc{epoch_acc:.2f}%) # 使用示例 train_with_tqdm(model, dataloader, criterion, optimizer, num_epochs3)方案四TensorBoard 记录 running lossimport torch import torch.nn as nn from torch.utils.tensorboard import SummaryWriter def train_with_tensorboard(model, dataloader, criterion, optimizer, num_epochs5, log_dirruns/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, devicecpu): 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_every50): 训练一个 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_sizebatch_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)} | fLoss: {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_sizebatch_size) self.val_metrics.record_epoch(epoch) return self.val_metrics.get_all_averages() def fit(self, train_loader, val_loader, num_epochs10, print_every50): 完整训练 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_size32, shuffleTrue) val_loader DataLoader(val_dataset, batch_size32, shuffleFalse) # 创建模型 model nn.Sequential( nn.Linear(10, 64), nn.ReLU(), nn.Linear(64, 3), ) criterion nn.CrossEntropyLoss() optimizer torch.optim.Adam(model.parameters(), lr0.001) # 训练 trainer Trainer(model, criterion, optimizer) train_history, val_history trainer.fit( train_loader, val_loader, num_epochs5, print_every5 ) # 打印历史 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{i1: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_lastFalse最后一个 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(fLoss: {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 是训练过程中跟踪损失变化的重要工具。正确使用它需要注意以下几点使用.item()将 loss 张量转为 Python 标量避免保留计算图导致内存泄漏。正确计算平均当 batch 大小不一致时按样本数加权计算平均 loss。合理的重置时机定期打印时每 N 个 batch 重置计算 epoch loss 时每个 epoch 重置。区分训练和验证训练时更新参数验证时使用torch.no_grad()不更新参数。使用 MetricTracker封装统计跟踪逻辑避免手动管理的错误。最佳实践始终使用loss.item()获取标量值按样本数加权计算平均 loss使用 tqdm 或 TensorBoard 可视化训练过程封装训练逻辑到 Trainer 类中定期保存和检查训练统计通过本文的详细分析和完整代码示例你应该能够彻底理解 running loss 的概念和正确计算方法在实际项目中构建可靠的训练循环和统计跟踪系统。