PyTorch 是当前大模型训练和推理的事实标准框架。本文聚焦于 AI Infra 工程师需要掌握的核心能力——从 Tensor 操作到完整训练循环,再到用 profiler 定位性能瓶颈,为后续的 CUDA 编程、分布式训练和推理优化打下基础。
📑 目录
1. Tensor:PyTorch 的基本数据单元 如果把深度学习比作搭积木,Tensor 就是最基本的积木块。所有的模型参数、输入数据、中间计算结果,在 PyTorch 里都以 Tensor 的形式存在。
正式定义 :Tensor(张量)是一个多维数组,可以看作 NumPy ndarray 的 GPU 加速版本,同时支持自动微分。
1.1 创建 Tensor 的常用方式 1 2 3 4 5 6 7 8 9 10 11 12 13 import torchimport numpy as npa = torch.tensor([1.0 , 2.0 , 3.0 ]) zeros = torch.zeros(3 , 4 ) ones = torch.ones(2 , 3 , 4 ) rand_normal = torch.randn(3 , 4 ) seq = torch.arange(0 , 10 , 2 ) x = torch.randn(3 , 4 ) y = torch.zeros_like(x) t = torch.from_numpy(np.array([1.0 , 2.0 ]))
1.2 形状操作:view, reshape, permute, squeeze, unsqueeze 把 Tensor 想象成一块可以任意捏形的橡皮泥——数据不变,只是换个排列方式。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 import torchx = torch.arange(12 ) a = x.view(3 , 4 ) b = x.reshape(3 , 4 ) c = x.view(-1 , 4 ) t = torch.randn(2 , 8 , 12 , 64 ) t_permuted = t.permute(0 , 2 , 1 , 3 ) e = torch.randn(1 , 3 , 1 , 4 ) print (e.squeeze().shape) f = torch.randn(3 , 4 ) print (f.unsqueeze(0 ).shape)
1.3 设备管理:CPU 与 GPU 数据搬运 1 2 3 4 5 6 7 8 9 10 11 import torchprint (torch.cuda.is_available()) x = torch.randn(3 , 4 ) x_gpu = x.to('cuda' ) x_gpu = x.cuda() x_cpu = x_gpu.cpu() x_gpu = torch.randn(3 , 4 , device='cuda' )
AI Infra 视角 :CPU-GPU 数据搬运走 PCIe 总线,带宽远低于 GPU 显存带宽。频繁的 .cpu() 和 .cuda() 调用往往是隐蔽的性能瓶颈,第 6 节会讲如何用 profiler 发现这类问题。
1.4 dtype 与内存:fp32, fp16, bf16 不同的数据精度直接决定显存占用和计算速度:
dtype
每元素字节
数值范围
使用场景
float32 (fp32)
4
大
默认精度,优化器状态
float16 (fp16)
2
小,易溢出
混合精度训练(需 Loss Scaling)
bfloat16 (bf16)
2
与 fp32 相同
混合精度训练(推荐,Ampere+ GPU)
白话理解 :fp32 像高清照片,又大又清晰;fp16 像压缩照片,体积减半但偶尔”失真”(溢出);bf16 是聪明的压缩,体积和 fp16 一样小,但”失真”概率大大降低。
1 2 3 4 5 6 7 8 9 10 import torchx_fp32 = torch.randn(1000 , 1000 , dtype=torch.float32) x_bf16 = torch.randn(1000 , 1000 , dtype=torch.bfloat16) print (f"fp32: {x_fp32.nelement() * x_fp32.element_size() / 1024 :.0 f} KB" ) print (f"bf16: {x_bf16.nelement() * x_bf16.element_size() / 1024 :.0 f} KB" ) x = torch.randn(3 , 4 ) x_half = x.half() x_bf16 = x.to(torch.bfloat16)
2. 自动微分(autograd) 训练神经网络的核心是”根据损失调整参数”,而调整的依据就是梯度。PyTorch 的 autograd 引擎帮你自动完成这件事。
2.1 什么是计算图
比喻 :想象你在做一道菜。你把食材 A 和 B 混合得到 C,再把 C 加热得到 D(成品)。如果 D 味道不对(loss 太大),你需要反推——是 C 的问题?还是 A、B 的比例不对?计算图就是 PyTorch 帮你记下的”食谱”:它记录每一步操作,这样就能从结果反向推导出每种食材对结果的影响(梯度)。
正式定义 :计算图(Computational Graph)是一个有向无环图(DAG),节点代表 Tensor,边代表运算操作。PyTorch 在前向传播时动态构建图,反向传播时沿图计算梯度。PyTorch 采用动态计算图 (Define-by-Run),支持 if/else、for 循环等 Python 控制流。
2.2 requires_grad 和 backward() 1 2 3 4 5 6 7 import torchx = torch.tensor([2.0 , 3.0 ], requires_grad=True ) y = x * 3 z = y.sum () z.backward() print (x.grad)
要点:backward() 只能对标量调用;叶节点才保存梯度;计算图用完即释放。
2.3 梯度累积与清零
白话理解 :PyTorch 默认累加 梯度,不自动清零。就像一个计数器,每次 backward() 都往上加。标准训练循环里,每个 step 开始前必须清零——否则上一轮梯度会混进来。
这个设计是有意为之的——梯度累积是显存不足时模拟大 batch 的常用技巧。
1 2 3 4 5 6 7 8 9 10 11 import torchx = torch.tensor([1.0 ], requires_grad=True ) (x * 2 ).sum ().backward() print (x.grad) (x * 3 ).sum ().backward() print (x.grad) x.grad.zero_() (x * 4 ).sum ().backward() print (x.grad)
代码示例:手动梯度下降学习 y = 2x + 1 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 import torchw = torch.tensor([0.0 ], requires_grad=True ) b = torch.tensor([0.0 ], requires_grad=True ) x_train = torch.tensor([1.0 , 2.0 , 3.0 , 4.0 ]) y_train = torch.tensor([3.0 , 5.0 , 7.0 , 9.0 ]) for epoch in range (100 ): loss = ((w * x_train + b - y_train) ** 2 ).mean() loss.backward() with torch.no_grad(): w -= 0.01 * w.grad b -= 0.01 * b.grad w.grad.zero_() b.grad.zero_() print (f"学到的模型: y = {w.item():.2 f} x + {b.item():.2 f} " )
3. nn.Module:模型的组织方式
比喻 :如果 Tensor 是积木块,nn.Module 就是积木的”说明书”——它定义了积木怎么拼接(前向传播),并帮你清点所有零件(参数管理)。
3.1 定义模型:继承 nn.Module,实现 __init__ 和 forward 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 import torchimport torch.nn as nnclass SimpleModel (nn.Module): def __init__ (self, input_dim, hidden_dim, output_dim ): super ().__init__() self .linear1 = nn.Linear(input_dim, hidden_dim) self .relu = nn.ReLU() self .linear2 = nn.Linear(hidden_dim, output_dim) def forward (self, x ): return self .linear2(self .relu(self .linear1(x))) model = SimpleModel(784 , 256 , 10 ) output = model(torch.randn(32 , 784 )) print (output.shape)
3.2 参数管理:parameters(), named_parameters(), state_dict() 1 2 3 4 5 6 7 8 9 10 11 12 import torch.nn as nnmodel = nn.Linear(10 , 5 ) for name, param in model.named_parameters(): print (f"{name} : {param.shape} " ) total_params = sum (p.numel() for p in model.parameters()) print (f"参数量: {total_params} " ) print (model.state_dict().keys())
3.3 常用层 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 import torchimport torch.nn as nnlinear = nn.Linear(512 , 256 ) embedding = nn.Embedding(num_embeddings=50000 , embedding_dim=768 ) print (embedding(torch.tensor([0 , 42 ])).shape) layer_norm = nn.LayerNorm(768 ) dropout = nn.Dropout(p=0.1 )
代码示例:两层 MLP 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 import torchimport torch.nn as nnclass TwoLayerMLP (nn.Module): def __init__ (self, input_dim, hidden_dim, output_dim, dropout=0.1 ): super ().__init__() self .fc1 = nn.Linear(input_dim, hidden_dim) self .relu = nn.ReLU() self .dropout = nn.Dropout(dropout) self .fc2 = nn.Linear(hidden_dim, output_dim) def forward (self, x ): return self .fc2(self .dropout(self .relu(self .fc1(x)))) model = TwoLayerMLP(784 , 256 , 10 ) print (f"参数量: {sum (p.numel() for p in model.parameters()):,} " ) print (model(torch.randn(64 , 784 )).shape)
4. 完整训练循环 训练循环是 AI Infra 工程师每天都要接触的代码——即使在分布式场景下,核心循环依然是同一个模式。
4.1 数据加载:Dataset 和 DataLoader
Dataset :定义”数据集里有什么”和”怎么取一条数据”
DataLoader :定义”怎么分 batch 喂给模型”(batching、shuffling、多进程加载)
1 2 3 4 5 6 7 8 9 10 11 12 13 14 import torchfrom torch.utils.data import Dataset, DataLoaderclass SimpleDataset (Dataset ): def __init__ (self, data, labels ): self .data, self .labels = data, labels def __len__ (self ): return len (self .data) def __getitem__ (self, idx ): return self .data[idx], self .labels[idx] dataset = SimpleDataset(torch.randn(1000 , 784 ), torch.randint(0 , 10 , (1000 ,))) loader = DataLoader(dataset, batch_size=64 , shuffle=True , num_workers=4 , pin_memory=True , drop_last=True )
AI Infra 视角 :num_workers 过小会导致 GPU 等数据(data loading bottleneck);pin_memory=True 使传输走异步 DMA,避免 CPU 拷贝开销。
4.2 标准训练流程 每个 step 的核心五步:forward → loss → backward → optimizer.step → zero_grad。
4.3 学习率调度 1 2 3 4 5 6 7 import torchoptimizer = torch.optim.AdamW(model.parameters(), lr=1e-3 ) scheduler = torch.optim.lr_scheduler.StepLR(optimizer, step_size=10 , gamma=0.1 ) scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=100 )
4.4 Checkpoint 保存与恢复 1 2 3 4 5 6 7 8 9 10 11 12 import torchtorch.save({ 'epoch' : epoch, 'model_state_dict' : model.state_dict(), 'optimizer_state_dict' : optimizer.state_dict(), 'loss' : avg_loss, }, 'checkpoint.pt' ) ckpt = torch.load('checkpoint.pt' , weights_only=False ) model.load_state_dict(ckpt['model_state_dict' ]) optimizer.load_state_dict(ckpt['optimizer_state_dict' ])
代码示例:完整 MNIST 训练脚本 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 import torchimport torch.nn as nnfrom torch.utils.data import DataLoaderfrom torchvision import datasets, transformsbatch_size, lr, num_epochs = 128 , 1e-3 , 5 device = torch.device('cuda' if torch.cuda.is_available() else 'cpu' ) transform = transforms.Compose([transforms.ToTensor(), transforms.Normalize((0.1307 ,), (0.3081 ,))]) train_set = datasets.MNIST('./data' , train=True , download=True , transform=transform) test_set = datasets.MNIST('./data' , train=False , transform=transform) train_loader = DataLoader(train_set, batch_size=batch_size, shuffle=True , num_workers=2 , pin_memory=True ) test_loader = DataLoader(test_set, batch_size=batch_size, num_workers=2 , pin_memory=True ) class MLP (nn.Module): def __init__ (self ): super ().__init__() self .net = nn.Sequential(nn.Flatten(), nn.Linear(784 , 256 ), nn.ReLU(), nn.Dropout(0.1 ), nn.Linear(256 , 10 )) def forward (self, x ): return self .net(x) model = MLP().to(device) criterion = nn.CrossEntropyLoss() optimizer = torch.optim.AdamW(model.parameters(), lr=lr) scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=num_epochs) for epoch in range (num_epochs): model.train() total_loss = 0 for images, labels in train_loader: images, labels = images.to(device), labels.to(device) loss = criterion(model(images), labels) loss.backward() optimizer.step() optimizer.zero_grad() total_loss += loss.item() scheduler.step() model.eval () correct = 0 with torch.no_grad(): for images, labels in test_loader: images, labels = images.to(device), labels.to(device) correct += (model(images).argmax(1 ) == labels).sum ().item() print (f"Epoch {epoch+1 } /{num_epochs} | Loss: {total_loss/len (train_loader):.4 f} " f"| Acc: {correct/len (test_set)*100 :.2 f} %" ) torch.save({'epoch' : epoch, 'model_state_dict' : model.state_dict(), 'optimizer_state_dict' : optimizer.state_dict()}, f'ckpt_ep{epoch+1 } .pt' )
5. GPU 训练基础 5.1 将模型和数据搬到 GPU 1 2 3 4 5 6 7 import torchimport torch.nn as nndevice = torch.device('cuda' if torch.cuda.is_available() else 'cpu' ) model = nn.Linear(100 , 10 ).to(device) x = torch.randn(32 , 100 ).to(device) y = model(x)
5.2 混合精度训练入门 核心思路:前向和反向用低精度(fp16/bf16)加速计算,参数更新用高精度(fp32)保证精度。
白话理解 :算草稿用铅笔(低精度,快),写定稿用钢笔(高精度,准)。关键的参数更新步骤用高精度保证不丢信息,其他大量计算用低精度跑得更快。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 import torchimport torch.nn as nndevice = torch.device('cuda' ) model = nn.Sequential(nn.Linear(784 , 256 ), nn.ReLU(), nn.Linear(256 , 10 )).to(device) optimizer = torch.optim.AdamW(model.parameters(), lr=1e-3 ) for images, labels in train_loader: images, labels = images.to(device), labels.to(device) with torch.autocast(device_type='cuda' , dtype=torch.bfloat16): loss = nn.functional.cross_entropy(model(images), labels) loss.backward() optimizer.step() optimizer.zero_grad() scaler = torch.cuda.amp.GradScaler() for images, labels in train_loader: images, labels = images.to(device), labels.to(device) with torch.autocast(device_type='cuda' , dtype=torch.float16): loss = nn.functional.cross_entropy(model(images), labels) scaler.scale(loss).backward() scaler.step(optimizer) scaler.update() optimizer.zero_grad()
5.3 显存查看 1 2 3 4 5 6 7 8 9 10 import torchprint (f"已分配: {torch.cuda.memory_allocated() / 1024 **3 :.2 f} GB" )print (f"峰值: {torch.cuda.max_memory_allocated() / 1024 **3 :.2 f} GB" )print (torch.cuda.memory_summary(abbreviated=True ))torch.cuda.reset_peak_memory_stats() print (f"峰值: {torch.cuda.max_memory_allocated() / 1024 **3 :.2 f} GB" )
代码示例:混合精度 vs fp32 显存对比 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 import torchimport torch.nn as nnassert torch.cuda.is_available(), "需要 GPU" device = torch.device('cuda' ) criterion = nn.CrossEntropyLoss() data = torch.randn(256 , 1024 , device=device) labels = torch.randint(0 , 10 , (256 ,), device=device) def make_model (): return nn.Sequential(nn.Linear(1024 , 2048 ), nn.ReLU(), nn.Linear(2048 , 1024 ), nn.ReLU(), nn.Linear(1024 , 10 )).to(device) model_fp32, opt_fp32 = make_model(), None opt_fp32 = torch.optim.AdamW(model_fp32.parameters(), lr=1e-3 ) torch.cuda.reset_peak_memory_stats() criterion(model_fp32(data), labels).backward() opt_fp32.step(); opt_fp32.zero_grad() fp32_peak = torch.cuda.max_memory_allocated() / 1024 **2 model_bf16, opt_bf16 = make_model(), None opt_bf16 = torch.optim.AdamW(model_bf16.parameters(), lr=1e-3 ) torch.cuda.reset_peak_memory_stats() with torch.autocast(device_type='cuda' , dtype=torch.bfloat16): loss = criterion(model_bf16(data), labels) loss.backward() opt_bf16.step(); opt_bf16.zero_grad() bf16_peak = torch.cuda.max_memory_allocated() / 1024 **2 print (f"FP32: {fp32_peak:.1 f} MB | BF16: {bf16_peak:.1 f} MB | 节省: {(1 -bf16_peak/fp32_peak)*100 :.1 f} %" )
6. 性能分析入门 写出能跑的训练代码只是第一步,跑得快才是 AI Infra 的核心追求。torch.profiler 能告诉你每个操作花了多少时间、GPU 利用率如何、哪里在空等数据。
6.1 torch.profiler 基本用法 1 2 3 4 5 6 7 8 9 10 11 12 13 14 import torchfrom torch.profiler import profile, record_function, ProfilerActivitymodel = torch.nn.Linear(1024 , 1024 ).cuda() x = torch.randn(64 , 1024 ).cuda() with profile(activities=[ProfilerActivity.CPU, ProfilerActivity.CUDA], record_shapes=True , profile_memory=True ) as prof: with record_function("forward_pass" ): output = model(x) with record_function("backward_pass" ): output.sum ().backward() print (prof.key_averages().table(sort_by="cuda_time_total" , row_limit=10 ))
6.2 如何读懂 profiler 输出
列名
含义
关注点
Name
操作名称
找到耗时最多的操作
CPU total
CPU 端总耗时
远大于 CUDA total 说明存在 CPU 瓶颈
CUDA total
GPU 端总耗时
核心指标
# of Calls
调用次数
高频小操作可能累积成瓶颈
常见性能问题模式:
CPU 时间远大于 CUDA 时间 :GPU 在等 CPU(数据预处理、Python 开销)
大量小 CUDA kernel :launch overhead 累积,考虑 torch.compile 或算子融合
频繁 CPU-GPU 同步 :.item()、print(tensor) 会触发同步,阻塞 GPU 流水线
6.3 常见性能问题 数据加载瓶颈 :增加 num_workers,使用 prefetch_factor 预取数据。
CPU-GPU 数据搬运 :避免循环中反复 .cuda(),一次性搬运到 GPU。
隐式同步 :loss.item() 触发 CPU-GPU 同步,应每 N 步才记录一次。
1 2 3 4 5 6 7 8 9 10 for batch in dataloader: loss = train_step(batch) print (f"loss: {loss.item()} " ) for i, batch in enumerate (dataloader): loss = train_step(batch) if i % 100 == 0 : print (f"step {i} , loss: {loss.item()} " )
代码示例:用 profiler 分析一个训练 step 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 import torchimport torch.nn as nnfrom torch.profiler import profile, record_function, ProfilerActivitydevice = torch.device('cuda' if torch.cuda.is_available() else 'cpu' ) model = nn.Sequential(nn.Linear(1024 , 4096 ), nn.ReLU(), nn.Linear(4096 , 4096 ), nn.ReLU(), nn.Linear(4096 , 10 )).to(device) optimizer = torch.optim.AdamW(model.parameters(), lr=1e-3 ) criterion = nn.CrossEntropyLoss() data = torch.randn(128 , 1024 , device=device) labels = torch.randint(0 , 10 , (128 ,), device=device) for _ in range (3 ): criterion(model(data), labels).backward() optimizer.step(); optimizer.zero_grad() with profile(activities=[ProfilerActivity.CPU, ProfilerActivity.CUDA], record_shapes=True , profile_memory=True ) as prof: with record_function("data_to_gpu" ): d = torch.randn(128 , 1024 ).to(device) l = torch.randint(0 , 10 , (128 ,)).to(device) with record_function("forward" ): loss = criterion(model(d), l) with record_function("backward" ): loss.backward() with record_function("optimizer_step" ): optimizer.step(); optimizer.zero_grad() print (prof.key_averages().table(sort_by="cuda_time_total" , row_limit=15 ))prof.export_chrome_trace("trace.json" )
从输出中可以快速判断:backward 是否比 forward 慢(正常约 2 倍),数据搬运是否占过多时间,优化器步骤是否有异常耗时。
📝 总结
层次
内容
核心能力
Tensor
数据表示与操作
形状操作、设备管理、dtype 选择
autograd
自动微分
计算图、梯度计算、梯度清零
nn.Module
模型组织
定义模型、参数管理、常用层
训练循环
端到端训练
DataLoader、训练流程、checkpoint
GPU 训练
硬件加速
混合精度、显存管理
性能分析
瓶颈定位
profiler 使用、常见瓶颈识别
这些构成了理解后续分布式训练、CUDA 编程和推理优化的必要基础。掌握这些内容后,你在阅读 DeepSpeed、Megatron-LM 等框架的代码时,不会因为 PyTorch 基础不牢而卡住。
🎯 自我检验清单 完成本文学习后,你应该能够:
能用 torch.randn、torch.zeros 等方法创建任意形状的 Tensor,并熟练使用 view、permute、squeeze 等操作变换形状
能解释 requires_grad=True 的作用,手动构建计算图并调用 backward() 获取梯度
能解释为什么每个训练 step 都需要调用 optimizer.zero_grad(),以及梯度累积的原理
能继承 nn.Module 实现自定义模型,并使用 parameters() 和 state_dict() 管理参数
能写出完整的训练循环:DataLoader → forward → loss → backward → optimizer.step → checkpoint
能使用 torch.autocast 实现 bf16/fp16 混合精度训练,并解释 GradScaler 的作用
能使用 torch.profiler 分析训练 step,读懂输出并识别数据加载和 CPU-GPU 传输瓶颈
能解释 fp32、fp16、bf16 三种精度的区别及选择依据
📚 参考资料