节前，我们星球组织了一场算法岗技术&面试讨论会，邀请了一些互联网大厂朋友、参加社招和校招面试的同学。
针对算法岗技术趋势、大模型落地项目经验分享、新手如何入门算法岗、该如何准备、面试常考点分享等热门话题进行了深入的讨论。
合集：
《大模型面试宝典》(2024版) 正式发布！
保姆级学习指南：《Pytorch 实战宝典》来了
在AI领域，常见的深度学习框架TensorFlow、PyTorch和Keras无疑是开发者们的得力工具，但随着模型规模的急剧膨胀，这些传统框架在应对大模型时往往会显得力不从心。
比如Pytorch的分布式并行计算框架DDP(Distributed Data Parallel)，尽管实现了数据并行，但是当模型大小超过单个GPU显存限制时显得捉襟见肘。此时，开发者往往只能手动进行复杂模型的参数拆分到各个GPU上，这无疑增加了研发的复杂性和门槛。
然而，微软推出的一款框架——DeepSpeed，可解决这一局限。它通过将模型参数拆散分布到各个GPU上，以实现大模型的计算。这也意味着，我们可以利用更少的硬件资源训练更大的模型，不再受限于单个GPU的显存限制。
安装DeepSpeed
pip install deepspeed
此外，还需要下载Pytorch，在官网选择自己对应的系统版本和环境，按照指示安装即可：
https://pytorch.org/get-started/locally/
使用DeepSpeed
载入数据集
# 导入必要的库
import torch
import torchvision
import torchvision.transforms as transforms
# 创建训练数据集
trainset = torchvision.datasets.CIFAR10(root='./data',
train=True,
download=True,
transform=transform)
# 创建数据加载器，批量加载数据并处理数据加载的并行化
trainloader = torch.utils.data.DataLoader(trainset,
# 每个批次包含16张图像
batch_size=16,
# 在每次迭代开始时随机打乱训练数据的顺序
# 有助于模型训练
shuffle=True,
# 开启2个子进程来并行加载数据，提高效率
num_workers=2)
# 创建测试数据集
testset = torchvision.datasets.CIFAR10(root='./data',
train=False,
download=True,
transform=transform)
testloader = torch.utils.data.DataLoader(testset,
batch_size=4,
#测试数据通常不需要打乱顺序
shuffle=False,
num_workers=2)
创建模型
# 导入必要的PyTorch模块
# 用于构建神经网络模型
import torch.nn as nn
# 提供了各种神经网络层的函数版本，如激活函数、损失函数等
import torch.nn.functional as F
# 定义一个名为Net的类，继承自nn.Module
class Net(nn.Module):
def __init__(self):
super(Net, self).__init__()
# 创建卷积层，参数：（输入通道数，输出通道数，卷积核大小）
self.conv1 = nn.Conv2d(3, 6, 5)
# 创建最大池化层，参数：（池化窗口大小，步长）
self.pool = nn.MaxPool2d(2, 2)
self.conv2 = nn.Conv2d(6, 16, 5)
# 创建全连接层（线性层），参数：（输入节点数，输出节点数）
self.fc1 = nn.Linear(16 * 5 * 5, 120)
self.fc2 = nn.Linear(120, 84)
self.fc3 = nn.Linear(84, 10)
# 前向传播过程，即输入数据通过网络的流程
def forward(self, x):
# 使用F.relu应用ReLU激活函数
# 使用self.pool进行最大池化
x = self.pool(F.relu(self.conv1(x)))
x = self.pool(F.relu(self.conv2(x)))
# 使用.view方法将池化后的特征图展平为一维向量，以便输入全连接层
x = x.view(-1, 16 * 5 * 5)
# 应用全连接层和ReLU激活函数，直到最后一层fc3，
# 它不使用激活函数，直接输出分类结果
x = F.relu(self.fc1(x))
x = F.relu(self.fc2(x))
x = self.fc3(x)
return x
# 实例化网络模型
net = Net()
# 设置损失函数
# 多分类交叉熵损失函数，适用于监督学习中的分类任务
criterion = nn.CrossEntropyLoss()
初始化Deepspeed
DeepSpeed通过输入参数来启动训练，因此需要使用argparse解析参数。
import argparse
def add_argument():
# 创建一个ArgumentParser对象，设置描述为"CIFAR"
parser = argparse.ArgumentParser(description='CIFAR')
# 设置训练时的批大小，默认值为32
parser.add_argument('-b',
'--batch_size',
default=32,
type=int,
help='mini-batch size (default: 32)')
# 设置总的训练轮数，默认值为30
parser.add_argument('-e',
'--epochs',
default=30,
type=int,
help='number of total epochs (default: 30)')
# 传递分布式训练中的排名，默认值为-1，表示未使用分布式训练
parser.add_argument('--local_rank',
type=int,
default=-1,
help='local rank passed from distributed launcher')
# 设置输出日志信息的间隔，默认值为2000，即每2000次迭代打印一次日志
parser.add_argument('--log-interval',
type=int,
default=2000,
help="output logging information at a given interval")
# 添加与DeepSpeed相关的配置参数
parser = deepspeed.add_config_arguments(parser)
# 解析命令行参数，返回一个Namespace对象，其中包含了所有定义的参数及其对应的值
args = parser.parse_args()
# 返回解析后的参数对象args，供后续的训练脚本使用
return args
此外，模型初始化的时候除了参数，还需要model及其parameters，还有训练集：
# 启动DeepSpeed训练
# 调用之前定义的add_argument函数，解析命令行参数，并将结果存储在args变量中
args = add_argument()
# 创建Net类的实例
net = Net()
# 筛选出模型中需要梯度计算的参数
parameters = filter(lambda p: p.requires_grad, net.parameters())
# 使用deepspeed.initialize初始化模型引擎、优化器、数据加载器以及其他可能的组件
model_engine, optimizer, trainloader, __ = deepspeed.initialize(
args=args, model=net, model_parameters=parameters, training_data=trainset)
训练
注意local_rank是不需要管的参数，在后面启动模型训练的时候，DeepSpeed会自动给这个参数赋值。
# 定义进行2个epoch的训练
for epoch in range(2):
running_loss = 0.0
# 对于每个epoch，遍历训练数据加载器trainloader中的每一个小批量数据
# 同时提供索引i和数据data
for i, data in enumerate(trainloader):
# 将输入数据inputs和标签labels移动到当前GPU设备上，
# 具体是哪个GPU由model_engine.local_rank决定，
# 这对于分布式训练非常重要，确保数据被正确地分配到各个参与训练的GPU上
inputs, labels = data[0].to(model_engine.local_rank), data[1].to(
model_engine.local_rank)
# 通过model_engine执行前向传播，计算模型预测输出
outputs = model_engine(inputs)
# 计算预测输出outputs与真实标签labels之间的损失
loss = criterion(outputs, labels)
# 反向传播计算梯度
model_engine.backward(loss)
# 更新模型参数
model_engine.step()
# 计算并累加每个小批量的损失值
# 当达到args.log_interval指定的迭代次数时，打印平均损失值，
# 然后重置running_loss为0，以便计算下一个区间的平均损失
running_loss += loss.item()
if i % args.log_interval == (args.log_interval - 1):
print('[%d, %5d] loss: %.3f' % (epoch + 1, i + 1, running_loss / args.log_interval))
running_loss = 0.0
测试
模型测试和模型训练的逻辑类似：
# 初始化计数器
# correct用于记录分类正确的样本数量
# total用于记录评估的总样本数
correct = 0
total = 0
# 上下文管理器，关闭梯度计算，
# 因为在验证阶段我们不需要计算梯度，这可以提高计算效率
with torch.no_grad():
# 遍历测试数据加载器testloader中的每个小批量数据
for data in testloader:
# 获取当前小批量数据的图像和标签
images, labels = data
# 在当前GPU上执行模型的前向传播
# 这里将图像数据移动到与模型相同的GPU上，然后通过模型得到预测输出
outputs = net(images.to(model_engine.local_rank))
# 找到每个样本的最大概率对应的类别
_, predicted = torch.max(outputs.data, 1)
# 增加总样本数，同时计算分类正确的样本数。
# 注意，这里将标签也移动到与模型相同的GPU上进行比较
total += labels.size(0)
correct += (predicted == labels.to(
model_engine.local_rank)).sum().item()
# 遍历完整个测试集后，计算并打印模型在测试集上的准确率
print('Accuracy of the network on the 10000 test images: %d %%' %
(100 * correct / total))
编写模型参数
当前目录下新建一个config.json，写入调优器、训练batch等参数。
{
// 每个GPU的训练批次大小
"train_batch_size": 4,
// 每隔多少步打印一次训练状态，这里是2000步
"steps_per_print": 2000,
//优化器的配置
"optimizer": {
//优化器类型
"type": "Adam",
//Adam优化器的参数
"params": {
//学习率
"lr": 0.001,
// Adam的第一和第二动量参数
"betas": [
0.8,
0.999
],
//优化器的稳定常数，防止除以零，这里是1e-8
"eps": 1e-8,
//权重衰减（L2正则化）
"weight_decay": 3e-7
}
},
// 学习率调度器的配置
"scheduler": {
//调度器类型
"type": "WarmupLR",
"params": {
//预热阶段的最小学习率，这里是0
"warmup_min_lr": 0,
// 预热阶段的最大学习率，这里是0.001
"warmup_max_lr": 0.001,
//预热阶段的步数，这里是1000
"warmup_num_steps": 1000
}
},
//是否开启时间分解功能，用于分析训练过程中的时间消耗。
//这里是false，表示不开启
"wall_clock_breakdown": false
}
以上即为利用DeepSpeed开发模型的过程，由此可见，和Pytorch开发模型的过程大同小异，就是在初始化的时候使用了DeepSpeed，并以输入参数的形式初始化。
测试代码
首先，使用环境变量控制GPU，例如机器有10张GPU，但是只使用6, 7, 8, 9号GPU，输入命令：
export CUDA_VISIBLE_DEVICES="6,7,8,9"
然后开始运行代码：
deepspeed test.py --deepspeed_config config.json
如下图所示即为开始运行。
开始训练的时候DeepSpeed通常会打印更多的训练细节供用户监控，包括训练设置、性能统计和损失趋势，效果类似于：
这也说明第一个Deepspeed模型已完成，下来可以开始大规模训练之路了！
