第 7 章 — 日志系统（logging + TensorBoard + rich）

第 6 章我们有了早停，模型能"见好就收"了，但所有信息仍然靠 print 往终端里喊：训练一跑完，屏幕上的日志就没了，想复盘某个 epoch 只能靠截图；想看 loss 曲线，得手动把数字抄进 Excel。本章解决"信息只进垃圾桶、不复用"的问题：用标准库 logging 建一个"广播台"（分级、带时间戳、同时写控制台和文件），用 TensorBoard 在浏览器里看训练仪表盘，再引入社区库 rich 把终端输出从"黑白打字机"变成"彩色显示器"。

一、本章要解决的问题

• print 是一次性喊话：进程结束就没了，出了 bug 想回看日志，只能后悔当时没截图。
• print 不分轻重：训练里有 INFO 也有 WARNING，想"只看警告以上"做不到，刷屏全是噪声。
• 指标只存在于终端：想分析 loss 曲线、对比多次实验，得手动抄数据，费时且易错。
• 终端输出不美观：长文本、数字表格挤在一起，关键时刻（哪一轮是最优）找不到重点。

二、核心概念速览

下面 8 个概念是本章代码的全部"生词"。老规矩，先花 5 分钟读完，再看代码会轻松很多。

1. print vs logging（只喊一嗓子 vs 广播台）

print 就像在房间里喊一嗓子：听到就听到，喊完就没了，没人记录、没人在意轻重缓急。logging 则像一个广播台：消息会经过"级别筛选 → 多个出口 → 带上时间戳"整套流水线再发出去，可以同时进终端、进文件，事后还能按级别检索。一句话：print 是给"人正在盯着看"用的，logging 是给"机器 + 事后复盘"用的。

2. 日志级别（DEBUG / INFO / WARNING / ERROR）

日志按严重程度排队，从低到高：DEBUG（调试细节）< INFO（正常进度）< WARNING（可疑但不致命）< ERROR（出错）。广播台可以设置"最低播音音量"：比如 setLevel(logging.INFO) 就把 DEBUG 全过滤掉，只播 INFO 及以上——想要更啰嗦的 DEBUG 日志，改一个数字就行，不用删代码。

3. handler（一条日志可以有多个"出口"）

handler 是日志的"出口"。一条日志可以同时发给控制台、写进文件、发到邮箱、推给监控系统——互不干扰，这就是"广播"的威力。本章注册两个 handler：一个 RichHandler（美化后的控制台）、一个 FileHandler（写文件），同一个 logger.info(...) 会同时到达两处。

4. rich（终端渲染美化库）

rich 是社区写的一个纯 Python 终端美化库：给文字上色、加粗、渲染表格、画进度条，还带语法高亮。类比成"给黑白的 Linux 终端装上美颜滤镜"。它不改变程序逻辑，只改变输出长什么样，所以可以放心引入、随时可删。

5. RichHandler 如何替代 StreamHandler

标准 logging 控制台输出默认用 StreamHandler（把文本原样扔给终端）。RichHandler 是 rich 为 logging 写的一个"高级出口"：它继承 handler 的机制，但把每条日志渲染成带颜色、带时间戳、可点击报错堆栈的样子。用法上零成本——还是 logger.addHandler(...)，只是传进去的对象从 StreamHandler() 换成了 RichHandler()。

6. TensorBoard（浏览器里的训练仪表盘）

TensorBoard 是 TensorFlow 附带的可视化工具，但 torch.utils.tensorboard 让 PyTorch 也能用。它在本地起一个小服务器，你在浏览器里打开就能看到"训练仪表盘"：loss / accuracy 曲线、参数分布直方图、模型计算图。类比成"训练过程的体检报告"，不用再抄数据画 Excel。

7. add_scalar 三要素（名称 / 数值 / step）

writer.add_scalar("val/loss", val_loss, epoch) 是记录一条曲线点的三要素：名称（带 / 会自动分组，train/ 和 val/ 是两个子面板）、数值（这一轮的值）、step（横轴，通常就是 epoch 数）。每次调用在横轴上点一个点，多轮跑下来就是一条曲线。

8. tensorboard --logdir runs 的查看方式

SummaryWriter(log_dir=...) 会把数据以二进制格式写进 runs/ 目录。训练完后（或训练中随时），在项目根目录执行 tensorboard --logdir runs，它会扫描 runs/ 下所有实验，然后终端打印一个本地地址（默认 http://localhost:6006），浏览器打开即可。注意要先 pip install tensorboard，否则会报 "tensorboard: command not found"。

三、解决思路

1. 统一入口：写一个 setup_logger(log_dir) 函数，返回配置好的 logger——控制台走 RichHandler（彩色、带时间），文件走 FileHandler + 普通 Formatter（纯文本，方便事后 grep 检索）。
2. 仪表盘：创建 SummaryWriter(log_dir=args.log_dir)，每轮把 train/val 的 loss、acc 以及当前 lr 用 add_scalar 写进去。
3. 换血：把第 6 章里所有 print(...) 原样替换成 logger.info(...)，语义不变，能力升级。
4. 先装依赖：执行 pip install rich 和 pip install tensorboard。

trade-off：logging 比 print 啰嗦（要写 handler、设 level），TensorBoard 的二进制目录也会占一点磁盘；但换来的是"可检索、可分级、可回放、可对比"的完整观测能力——这是工程化训练脚本的基本功。

四、代码变更

相对第 6 章（chapter_06 的 train.py）的改动：

  import os
+ import logging                     # 新增：标准库日志
  import argparse
  ...
  from torch.utils.data import DataLoader, random_split
+ from torch.utils.tensorboard import SummaryWriter  # 新增：TensorBoard
  from torchvision import datasets, transforms
  from collections import defaultdict
+ from tqdm import tqdm              # 新增：进度条
+ from rich.logging import RichHandler   # 新增：rich 美化终端日志

  parser.add_argument("--patience", type=int, default=7, ...)
+ parser.add_argument("--log-dir", type=str, default="./runs/cifar10",
+                     help="TensorBoard 日志与 train.log 的输出目录")
  return parser.parse_args()

  args = parse_args()
+
+ def setup_logger(log_dir):
+     """控制台用 rich 渲染（彩色带时间），文件用普通格式（便于归档 grep）。"""
+     os.makedirs(log_dir, exist_ok=True)
+     logger = logging.getLogger("train")
+     logger.handlers.clear()
+     logger.setLevel(logging.INFO)
+     console = RichHandler(rich_tracebacks=True, markup=True)  # 控制台出口
+     console.setLevel(logging.INFO)
+     logger.addHandler(console)
+     file_handler = logging.FileHandler(           # 文件出口（普通格式）
+         os.path.join(log_dir, "train.log"), encoding="utf-8")
+     file_handler.setFormatter(logging.Formatter(
+         "%(asctime)s | %(levelname)s | %(message)s", datefmt="%Y-%m-%d %H:%M:%S"))
+     logger.addHandler(file_handler)
+     return logger
+
+ logger = setup_logger(args.log_dir)
+ writer = SummaryWriter(log_dir=args.log_dir)     # TensorBoard 写入器

  if args.device == "auto":
      device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
  else:
      device = torch.device(args.device)
- print(f"Using device: {device}")
+ logger.info(f"Using device: {device}")

  def train_one_epoch(...):
      model.train()
      ...
-     for images, labels in loader:
+     for images, labels in tqdm(loader, desc="Train", leave=False):

  def save_checkpoint(state, path):
      os.makedirs(os.path.dirname(path), exist_ok=True)
      torch.save(state, path)
-     print(f"[Checkpoint] 已保存到 {path}")
+     logger.info(f"[Checkpoint] 已保存到 {path}")

              if self.verbose:
-                 print(f"[EarlyStopping] 连续 {self.counter}/{self.patience} 轮未提升")
+                 logger.info(f"[EarlyStopping] 连续 {self.counter}/{self.patience} 轮未提升")

-     print(f"[Resume] 从 epoch {ckpt['epoch']} 恢复，历史最优 val_acc={best_acc:.4f}")
+     logger.info(f"[Resume] 从 epoch {ckpt['epoch']} 恢复，历史最优 val_acc={best_acc:.4f}")

- print("=" * 60)
- print("训练配置：")
+ logger.info("=" * 60)
+ logger.info("训练配置：")
  for k, v in vars(args).items():
-     print(f"  {k:12s} = {v}")
- print("=" * 60)
+     logger.info(f"  {k:12s} = {v}")
+ logger.info("=" * 60)

-     print(f"Epoch {epoch:02d}/{args.epochs} | "
-           f"Train loss {train_loss:.4f} acc {train_acc:.4f} | "
-           f"Val   loss {val_loss:.4f} acc {val_acc:.4f} | "
-           f"lr {current_lr:.2e}")
+     logger.info(f"Epoch {epoch:02d}/{args.epochs} | "
+                 f"Train loss {train_loss:.4f} acc {train_acc:.4f} | "
+                 f"Val   loss {val_loss:.4f} acc {val_acc:.4f} | "
+                 f"lr {current_lr:.2e}")
+
+     # 写入 TensorBoard：按「分组/指标」命名，浏览器里自动分组
+     writer.add_scalar("train/loss", train_loss, epoch)
+     writer.add_scalar("train/acc", train_acc, epoch)
+     writer.add_scalar("val/loss", val_loss, epoch)
+     writer.add_scalar("val/acc", val_acc, epoch)
+     writer.add_scalar("lr", current_lr, epoch)

      early_stopping(val_acc)
      if early_stopping.early_stop:
-         print(f"[EarlyStopping] 连续 {args.patience} 轮无提升，在 epoch {epoch} 停止")
+         logger.info(f"[EarlyStopping] 连续 {args.patience} 轮无提升，在 epoch {epoch} 停止")
          break
 
  test_loss, test_acc = validate(model, test_loader, criterion, device)
- print(f"\n[Test] loss {test_loss:.4f} acc {test_acc:.4f}")
+ logger.info(f"[Test] loss {test_loss:.4f} acc {test_acc:.4f}")
+
+ writer.close()

改动总结：新增 2 个依赖（rich、tensorboard）、1 个参数（--log-dir）、1 个函数（setup_logger）、1 个写入器（SummaryWriter）；所有 print 替换为 logger.info；batch 循环套上 tqdm 进度条；epoch 循环末尾增加 TensorBoard 的 add_scalar。

五、完整代码

创建 train.py（完整版）。运行前先安装两个新依赖：

pip install rich
pip install tensorboard

import os
import logging                      # 标准库：日志系统（分级、带时间戳、可写文件）
import argparse
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, random_split
from torch.utils.tensorboard import SummaryWriter  # TensorBoard：把指标写进 runs/ 目录
from torchvision import datasets, transforms
from collections import defaultdict
from tqdm import tqdm               # 进度条：batch 循环时显示还剩多少
from rich.logging import RichHandler   # rich：终端渲染美化


def parse_args():
    parser = argparse.ArgumentParser(description="CIFAR-10 图像分类训练")
    parser.add_argument("--epochs", type=int, default=30, help="训练轮数上限")
    parser.add_argument("--batch-size", type=int, default=64, help="每个 batch 的样本数")
    parser.add_argument("--lr", type=float, default=1e-3, help="初始学习率")
    parser.add_argument("--lr-scheduler", type=str, default="step",
                        choices=["step", "cosine"], help="学习率调度策略")
    parser.add_argument("--lr-step-size", type=int, default=15, help="StepLR 每多少轮衰减一次")
    parser.add_argument("--lr-gamma", type=float, default=0.1, help="StepLR 衰减系数")
    parser.add_argument("--data-dir", type=str, default="./data", help="数据集存放目录")
    parser.add_argument("--num-workers", type=int, default=2, help="DataLoader 数据加载进程数")
    parser.add_argument("--device", type=str, default="auto", help="运行设备：auto/cuda/cpu")
    parser.add_argument("--ckpt-dir", type=str, default="./checkpoints", help="checkpoint 保存目录")
    parser.add_argument("--resume", type=str, default=None, help="从指定 checkpoint 恢复训练")
    parser.add_argument("--patience", type=int, default=7, help="早停：验证集连续多少轮无提升则停止")
    parser.add_argument("--log-dir", type=str, default="./runs/cifar10",
                        help="TensorBoard 日志与 train.log 的输出目录")
    return parser.parse_args()


args = parse_args()


def setup_logger(log_dir):
    """控制台用 rich 渲染（彩色带时间），文件用普通格式（便于归档 grep）。"""
    os.makedirs(log_dir, exist_ok=True)
    logger = logging.getLogger("train")   # 用名字取 logger，防止和别的模块混淆
    logger.handlers.clear()               # 清掉已有 handler，避免重复输出
    logger.setLevel(logging.INFO)         # 只播 INFO 及以上，DEBUG 一律过滤

    # 控制台 handler：RichHandler 提供彩色输出、自带时间戳
    console = RichHandler(rich_tracebacks=True, markup=True)
    console.setLevel(logging.INFO)
    logger.addHandler(console)

    # 文件 handler：普通格式，方便事后用 grep 检索
    file_handler = logging.FileHandler(os.path.join(log_dir, "train.log"),
                                       encoding="utf-8")
    file_handler.setFormatter(logging.Formatter(
        "%(asctime)s | %(levelname)s | %(message)s", datefmt="%Y-%m-%d %H:%M:%S"))
    logger.addHandler(file_handler)
    return logger


logger = setup_logger(args.log_dir)
writer = SummaryWriter(log_dir=args.log_dir)   # 所有 add_scalar 都会写进 runs/cifar10

if args.device == "auto":
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
else:
    device = torch.device(args.device)
logger.info(f"Using device: {device}")


class SimpleCNN(nn.Module):
    def __init__(self, num_classes=10):
        super().__init__()
        self.features = nn.Sequential(
            nn.Conv2d(3, 32, kernel_size=3, padding=1),
            nn.BatchNorm2d(32),
            nn.ReLU(inplace=True),
            nn.MaxPool2d(2),
            nn.Conv2d(32, 64, kernel_size=3, padding=1),
            nn.BatchNorm2d(64),
            nn.ReLU(inplace=True),
            nn.MaxPool2d(2),
            nn.Conv2d(64, 128, kernel_size=3, padding=1),
            nn.BatchNorm2d(128),
            nn.ReLU(inplace=True),
            nn.MaxPool2d(2),
        )
        self.classifier = nn.Sequential(
            nn.AdaptiveAvgPool2d(1),
            nn.Flatten(),
            nn.Linear(128, num_classes),
        )

    def forward(self, x):
        return self.classifier(self.features(x))


transform = transforms.Compose([
    transforms.ToTensor(),
    transforms.Normalize((0.4914, 0.4822, 0.4465), (0.2470, 0.2435, 0.2616)),
])

train_dataset = datasets.CIFAR10(root=args.data_dir, train=True, download=True, transform=transform)
test_dataset = datasets.CIFAR10(root=args.data_dir, train=False, download=True, transform=transform)
train_dataset, val_dataset = random_split(
    train_dataset, [45000, len(train_dataset) - 45000]
)

train_loader = DataLoader(train_dataset, batch_size=args.batch_size,
                          shuffle=True, num_workers=args.num_workers)
val_loader = DataLoader(val_dataset, batch_size=args.batch_size,
                        shuffle=False, num_workers=args.num_workers)
test_loader = DataLoader(test_dataset, batch_size=args.batch_size,
                         shuffle=False, num_workers=args.num_workers)

model = SimpleCNN().to(device)
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=args.lr)

if args.lr_scheduler == "step":
    scheduler = optim.lr_scheduler.StepLR(
        optimizer, step_size=args.lr_step_size, gamma=args.lr_gamma)
elif args.lr_scheduler == "cosine":
    scheduler = optim.lr_scheduler.CosineAnnealingLR(
        optimizer, T_max=args.epochs)
else:
    scheduler = None


def train_one_epoch(model, loader, criterion, optimizer, device):
    model.train()
    total_loss, correct, total = 0.0, 0, 0
    for images, labels in tqdm(loader, desc="Train", leave=False):
        # tqdm 包住 loader：每个 batch 结束时刷新进度条，leave=False 让进度条不占满屏幕
        images, labels = images.to(device), labels.to(device)
        optimizer.zero_grad()
        outputs = model(images)
        loss = criterion(outputs, labels)
        loss.backward()
        optimizer.step()
        total_loss += loss.item() * images.size(0)
        correct += (outputs.argmax(dim=1) == labels).sum().item()
        total += images.size(0)
    return total_loss / total, correct / total


def validate(model, loader, criterion, device):
    model.eval()
    total_loss, correct, total = 0.0, 0, 0
    with torch.no_grad():
        for images, labels in tqdm(loader, desc="Val", leave=False):
            images, labels = images.to(device), labels.to(device)
            outputs = model(images)
            loss = criterion(outputs, labels)
            total_loss += loss.item() * images.size(0)
            correct += (outputs.argmax(dim=1) == labels).sum().item()
            total += images.size(0)
    return total_loss / total, correct / total


def save_checkpoint(state, path):
    os.makedirs(os.path.dirname(path), exist_ok=True)
    torch.save(state, path)
    logger.info(f"[Checkpoint] 已保存到 {path}")


class EarlyStopping:
    def __init__(self, patience=7, delta=0.0, verbose=False):
        self.patience = patience
        self.delta = delta
        self.verbose = verbose
        self.best_score = None
        self.counter = 0
        self.early_stop = False

    def __call__(self, val_acc):
        score = val_acc
        if self.best_score is None:
            self.best_score = score
        elif score < self.best_score + self.delta:
            self.counter += 1
            if self.verbose:
                logger.info(f"[EarlyStopping] 连续 {self.counter}/{self.patience} 轮未提升")
            if self.counter >= self.patience:
                self.early_stop = True
        else:
            self.best_score = score
            self.counter = 0


metrics = defaultdict(list)
start_epoch, best_acc = 1, 0.0

if args.resume:
    ckpt = torch.load(args.resume, map_location=device, weights_only=False)
    model.load_state_dict(ckpt["model_state"])
    optimizer.load_state_dict(ckpt["optimizer_state"])
    if scheduler is not None and ckpt.get("scheduler_state") is not None:
        scheduler.load_state_dict(ckpt["scheduler_state"])
    start_epoch = ckpt["epoch"] + 1
    best_acc = ckpt.get("best_acc", 0.0)
    metrics = defaultdict(list, ckpt.get("metrics", {}))
    logger.info(f"[Resume] 从 epoch {ckpt['epoch']} 恢复，历史最优 val_acc={best_acc:.4f}")


early_stopping = EarlyStopping(patience=args.patience, verbose=True)

logger.info("=" * 60)
logger.info("训练配置：")
for k, v in vars(args).items():
    logger.info(f"  {k:12s} = {v}")
logger.info("=" * 60)

for epoch in range(start_epoch, args.epochs + 1):
    train_loss, train_acc = train_one_epoch(model, train_loader, criterion, optimizer, device)
    val_loss, val_acc = validate(model, val_loader, criterion, device)

    if scheduler is not None:
        scheduler.step()
    current_lr = optimizer.param_groups[0]["lr"]

    metrics["train_loss"].append(train_loss)
    metrics["train_acc"].append(train_acc)
    metrics["val_loss"].append(val_loss)
    metrics["val_acc"].append(val_acc)

    logger.info(f"Epoch {epoch:02d}/{args.epochs} | "
                f"Train loss {train_loss:.4f} acc {train_acc:.4f} | "
                f"Val   loss {val_loss:.4f} acc {val_acc:.4f} | "
                f"lr {current_lr:.2e}")

    # 写入 TensorBoard：按「分组/指标」命名，浏览器里自动分组
    writer.add_scalar("train/loss", train_loss, epoch)
    writer.add_scalar("train/acc", train_acc, epoch)
    writer.add_scalar("val/loss", val_loss, epoch)
    writer.add_scalar("val/acc", val_acc, epoch)
    writer.add_scalar("lr", current_lr, epoch)

    state = {
        "model_state": model.state_dict(),
        "optimizer_state": optimizer.state_dict(),
        "scheduler_state": scheduler.state_dict() if scheduler else None,
        "epoch": epoch,
        "best_acc": best_acc,
        "metrics": dict(metrics),
    }
    save_checkpoint(state, os.path.join(args.ckpt_dir, "last.pth"))
    if val_acc > best_acc:
        best_acc = val_acc
        state["best_acc"] = best_acc
        save_checkpoint(state, os.path.join(args.ckpt_dir, "best.pth"))

    early_stopping(val_acc)
    if early_stopping.early_stop:
        logger.info(f"[EarlyStopping] 连续 {args.patience} 轮无提升，在 epoch {epoch} 停止")
        break

test_loss, test_acc = validate(model, test_loader, criterion, device)
logger.info(f"[Test] loss {test_loss:.4f} acc {test_acc:.4f}")

writer.close()

运行示例：

# 正常训练，日志同时进终端（彩色）和 runs/cifar10/train.log（纯文本）
python train.py --epochs 30 --patience 7

# 训练跑完后（或训练中另开一个终端），启动 TensorBoard 仪表盘
tensorboard --logdir runs
# 浏览器打开 http://localhost:6006，左侧 "runs/cifar10" 下能看到 train/val 曲线

查看 train.log 的两种姿势：

cat runs/cifar10/train.log          # 看完整归档
grep "WARNING" runs/cifar10/train.log   # 只筛出警告以上——print 时代做不到的事

六、本章小结

• 学到了什么
  • logging 的核心三件套：logger（广播台）→ level（音量筛选）→ handler（出口）。多个出口让"一份日志、多处归档"成为可能。
  • 控制台用 RichHandler（彩色、带时间、可点堆栈），文件用普通 Formatter（纯文本、可 grep），两者各取所长。
  • TensorBoard 的用法就两个动作：writer.add_scalar(名称, 数值, step) 记录 + tensorboard --logdir runs 查看。/ 符号让曲线自动分组。
  • 所有 print 换成 logger.info 后，行为对用户完全透明，但能力从"一次性"升级成"可检索、可分级、可回放"。
• 常见坑
  • logger.handlers.clear() 别删：重跑脚本（比如在 Jupyter 里多次执行）时，不清理会叠出一堆重复日志。
  • 忘了 pip install rich / pip install tensorboard：前者报 ModuleNotFoundError: No module named 'rich'，后者报 tensorboard: command not found。
  • writer.close() 别漏：不关可能丢缓冲里的最后几条记录；SummaryWriter 与 checkpoint 无关，恢复训练时新 writer 会接着写同一个 runs/cifar10 目录。
  • TensorBoard 曲线"接续"靠的是同一目录：想对比不同实验，用不同的 --log-dir（如 runs/exp_lr1e3），TensorBoard 会自动并排显示。
• 下一章预告：这一章的 --log-dir 已经是第 13 个命令行参数了——一行命令十几个 --xxx 又长又难记，还容易手滑。第 8 章引入 YAML 配置文件，把超参数按模块组织起来，argparse 只保留少数关键覆盖项。

七、动手练习

1. 感受分级：把 setup_logger 里的 logger.setLevel(logging.INFO) 改成 logging.DEBUG，再往代码里塞一条 logger.debug(f"batch {i} loss {loss.item():.4f}")，对比两种级别下终端的输出量；最后把 level 改回 INFO，观察 DEBUG 日志消失。
2. 观察双出口：跑一轮 python train.py --epochs 2，同时打开终端和 runs/cifar10/train.log，确认每个 epoch 的日志出现在两处，且文件里的格式带 2025-..-.. HH:MM:SS 时间戳。
3. 玩转 TensorBoard：分别用 --lr 1e-3 和 --lr 5e-2 --lr-scheduler cosine 各跑 20 轮（--log-dir runs/exp_a 与 runs/exp_b），启动 tensorboard --logdir runs，对比两组曲线，思考"哪个学习率策略收敛得更稳"。
4. grep 实战：跑完一次完整训练后，用 grep "WARNING" runs/cifar10/train.log 或 grep "EarlyStopping" runs/cifar10/train.log 快速定位早停与警告信息，体会"print 时代做不到的事"。