如何设计一个用于大规模生产任务的人工智能AI系统

部署一个SOTA模型，让它服务数百万用户，处理TB级别的数据，并且7x24小时可靠运行是件非常有挑战性的工作。我们将探讨构建一个能够创建LLM、多模态模型以及各种其他AI产品的大规模AI系统所需的每个开发阶段。每个开发阶段如何相互关联，以及它们各自的职责。

第一阶段：AI 的系统与硬件

构建大规模 AI 系统的第一步是选择正确的硬件。它影响你的模型运行速度、花费多少钱以及所有设备使用的能量。

在本节中，我们将讨论可用的不同硬件系统，以及如何使它们具有成本和能源效益。

AI 的计算硬件

用于训练或其他 AI 任务的三种最常见的硬件类型是：

AI 硬件可用性

• CPU（中央处理单元） 它们擅长处理许多不同的任务，但它们的核心不多，所以对于需要大量并行处理的深度学习或大型 AI 作业来说可能会比较慢。
• GPU（图形处理单元） 最初是为处理视频和图形而构建的，但现在它们是 AI 的宠儿。因为它们比 CPU 拥有更多的核心，这意味着它们可以同时处理很多事情，非常适合训练和运行 AI 模型。
• TPU（张量处理单元） 是谷歌专门为深度学习制造的特殊芯片。它们速度非常快，效率超高，并且使用的能源更少。这使它们非常适合大型、复杂的 AI 任务。

您可以在这里了解更多：https://cloud.google.com/tpu

但最近，由于对 AI 的需求不断增长，也引入了一些新型硬件。

现代硬件

• 一个很好的例子是 FPGA（现场可编程门阵列）。这些芯片很特别，因为它们可以被重新编程以适应不同的 AI 任务。它们为您提供了根据模型需求微调性能的灵活性，这在快速变化的 AI 项目中非常有用。
• 还有 ASIC（专用集成电路）。这些不像 CPU 甚至 FPGA 那样是通用的。相反，它们是为一件事而设计的：尽可能快且高效地运行 AI 模型。因为它们是为特定工作（如驱动神经网络）而构建的，所以它们可以节省能源并且运行速度非常快。

您可以阅读这个故事了解更多细节：

在选择硬件时，直接选择 GPU 并不总是正确的方法，因为我们常常假设它们会自动提高性能，无论是数据预处理、微调还是 LLM 推理。

然而，性能很大程度上取决于…

模型架构 + 基础设施选择

1. 在 AI 架构方面，一个有用的技术是模型量化，许多现代开源模型 API 提供商（如 Together AI 或 Nebius AI）已经在使用了。这意味着减少你的 AI 模型在进行计算时使用的细节量，比如使用更小的数字（例如，8位而不是32位）。
2. 在基础设施方面，云服务和虚拟化通常是最佳解决方案。与其购买昂贵的硬件，不如从 AWS、Google Cloud 或 Azure 等提供商那里租用强大的机器。这为您提供了根据项目需求扩展或缩减的灵活性，从而节省资金并避免浪费。

看一下谷歌的比较图表，它显示了不同模型架构在各种 GPU 上的表现。

性能比较 — 谷歌比较文章）

谷歌在 MLPerf 3.1 基准测试（主要用于测试系统处理输入的速度）上进行了测试。

• 对于棘手的 AI 作业，使用强大 H100 GPU 的 A3 VM 比旧的 A2 VM 快 1.7 到 3.9 倍。
• 如果您希望在获得可靠 AI 性能的同时节省资金，使用 L4 GPU 的 G2 VM 是一个不错的选择。
• 测试表明，L4 GPU 每花费一美元所能提供的性能比类似云服务高出 1.8 倍。

像 Bending Spoons 这样的公司已经在使用 G2 VM 来高效地为用户带来新的 AI 功能。

AI 的分布式系统

一旦根据您的需求选择了优化的硬件和模型架构，我们就会进入下一阶段，这涉及到您如何规划 AI 的分布式系统。

分布式系统的主要原则是…

将一个大任务分成更小的部分，让多台计算机同时处理它们。

在 AI 中，这通过分担工作负载来加速数据处理和模型训练。

因此，要创建一个分布式系统，我们需要牢记一些重要因素。让我们先将其可视化，然后理解其流程。

分布式 AI 系统（来自 Sina Torfi — 开源）

在将分布式逻辑应用于我们的 AI 系统时，我们需要考虑多种因素。让我们看看流程是怎样的：

1. 首先，我们需要了解规模。 我们处理的是数百、数千还是数百万个数据点？尽早了解这一点有助于我们规划系统以实现平稳增长。
2. 接下来，我们选择合适的工具。 根据项目的规模和类型，我们需要正确的处理能力、内存和通信方法的组合。云平台使管理变得更加容易。
3. 然后，我们确保一切协同工作。 系统的不同部分可能需要并行运行或在不同的机器上运行。我们的目标是避免减速并保持系统平稳运行。
4. 之后，我们保持灵活性。 我们自动化资源调整，而不是手动调整。像 Kubernetes 这样的工具有助于系统根据负载变化自动调整。
5. 我们还需要监控性能。 密切关注系统有助于我们及早发现问题，无论是数据分布不均还是网络瓶颈。
6. 最后，我们确保一切保持同步。 随着系统的扩展，数据和模型在所有部分保持一致至关重要。

优化网络

一旦您决定了 AI 系统的分布式部分，您需要确保所有组件都正确连接。

它们必须能够顺畅地相互通信，没有任何障碍

如果分布式组件无法有效通信，可能会破坏您的训练或生产代码。

让我们看看如何让对话流畅而不中断：

分布式系统的网络优化

让我们分解一下：

1. 首先，我们寻找潜在的减速点。 延迟、有限容量或数据丢失会严重影响性能，因此尽早识别这些风险非常重要。
2. 然后，我们减少延迟。 为了加快速度，我们使用更快的连接，将机器放在一起，甚至将一些处理转移到边缘。
3. 接下来，我们增加带宽。 狭窄的网络路径会导致交通拥堵。我们通过压缩数据、优先处理重要信息或升级网络来解决这个问题。
4. 之后，我们选择正确的通信方法。 一些协议比其他协议更能处理大负载。选择正确的协议可确保您的系统运行快速高效。
5. 我们还为未来的增长做计划。 随着系统的扩展，网络必须跟上。使用可以根据需要增长的灵活设置是关键。
6. 最后，我们监控网络。 定期检查有助于我们及早发现问题。监控工具可以在问题导致系统减速之前向我们发出警报。

AI 的存储解决方案

因此，在您决定了用于训练或推理的硬件及其背后的分布式逻辑之后，您需要的下一件事是存储，以保存您训练好的模型以及用户与您的 AI 模型对话产生的数据。

我们存储数据的方式必须既满足当前需求，又能为未来的更多数据做好准备

AI 存储需求的增长（来自 Market US）

我们有三种类型的数据存储系统：

数据存储选项

• 对象存储最适合大数据。 在这里，您可以不断添加文件而无需担心结构。当您的数据来自许多来源并需要稍后整合时，它是完美的。
• 文件系统更适合小型、有组织的设置。 它们有点像您计算机上的文件夹。它们有助于保持整洁，并且在您的数据有限且结构良好时是理想的选择。

而第三种类型是数据库，在您的数据具有结构时很有用。以下是如何选择正确的类型：

1. SQL 数据库非常适合有组织、相互关联的数据。 当您的数据具有明确的关系时（例如用户、订单和产品），请使用它们。它们非常适合需要准确性和一致性的复杂任务。
2. NoSQL 数据库适用于灵活或不断变化的数据。 如果您的数据不能整齐地放入表格或增长迅速，像 MongoDB 或 Cassandra 这样的 NoSQL 选项提供了您所需的自由度和可扩展性。

尽管工具不是唯一重要的东西，但如何使用它们也很重要：

• 数据湖以原始形式保存所有内容。 它是一个巨大的容器，用于存放各种数据，您可以稍后进行排序和处理。
• 数据仓库存储干净、可随时使用的数据。 这就像一个组织良好的图书馆，您可以快速找到所需的确切内容。
• 数据版本控制跟踪更改。 在更新模型或处理随时间变化的数据时，这很重要。它有助于保持组织性并防止错误。
• 混合存储结合了速度和成本节约。 您使用快速存储来存放经常需要的数据，使用更便宜的存储来存放其余数据。这样，您可以在必要时节省资金并仍然获得快速访问。

快速的数据访问是 AI 性能的关键。

使用像 Redis 这样的内存存储进行快速检索，并应用数据分片来分散负载并防止减速。

在某个时候，您需要决定哪种存储设置效果最好：云、本地或两者兼有。

云 vs 本地

1. 混合存储为您提供灵活性。 您可以将敏感数据保留在自己的服务器上，同时使用云来处理其他所有内容。这有助于平衡安全性和可扩展性。
2. 多云策略提供更多选择。 通过使用多个云提供商，您可以避免被锁定在一家。这就像有不同的菜单可供选择，具体取决于您的需求。

第二阶段：高级模型训练技术

到目前为止，我们已经介绍了硬件、存储以及如何充分利用它们。现在是时候看看训练技术是如何工作的，以及我们如何优化它们。

优化神经网络训练的策略

AI 模型通常建立在神经网络之上，虽然许多模型从基本的梯度下降开始，但在实际场景中，有更高级的选项表现更好。

优化神经网络

Adam 优化是一个明智的选择。 它结合了 AdaGrad 和 RMSprop 的优点。它能很好地处理噪声数据和稀疏梯度，使其成为流行的默认选择。

# 使用 Adam 优化器，学习率为 0.001
optimizer = optim.Adam(model.parameters(), lr=0.001)

RMSprop 有助于提高学习稳定性。 它根据最近的梯度行为调整学习率，并且对于非平稳问题效果很好。

# 使用 RMSprop 优化器，学习率为 0.001，alpha(平滑常数)为 0.99
optimizer = optim.RMSprop(model.parameters(), lr=0.001, alpha=0.99)

Adagrad 能适应你的数据。 它为每个参数更改学习率，这对于稀疏数据非常有用，但可能导致学习率随着时间的推移缩得太小。

# 使用 Adagrad 优化器，学习率为 0.01
optimizer = optim.Adagrad(model.parameters(), lr=0.01)

让我们用一个简单的表格来概览我们有哪些优化器以及它们各自的适用场景。

因此，这种比较可以帮助机器学习工程师决定选择哪种优化器。

我们可以放心地从 Adam 开始。虽然优化器之间存在差异，但从实用的角度出发并获得一些初步见解是很重要的。

大规模训练的框架和工具

接下来是正则化技术，这对于防止过拟合和确保模型能够很好地泛化到新数据至关重要。以下是一些帮助您的模型很好地泛化到新数据的常用方法。

带有权重衰减的 L2 正则化通过抑制大的权重来帮助保持模型更简单。

# 使用 Adam 优化器，学习率为 0.001，权重衰减(L2惩罚)为 1e-5
optimizer = optim.Adam(model.parameters(), lr=0.001, weight_decay=1e-5)

模型中的 Dropout 层在训练期间随机丢弃神经元，使模型不太可能过拟合。

import torch.nn as nnclass MyModel(nn.Module):def __init__(self):super(MyModel, self).__init__()# 第一个线性层：输入大小 784（例如 28x28 图像）到 256 个神经元self.layer1 = nn.Linear(784, 256)# Dropout 层，概率为 50%，以减少过拟合self.dropout = nn.Dropout(0.5)# 第二个线性层：256 到 10 个输出类别（例如数字 0-9）self.layer2 = nn.Linear(256, 10)def forward(self, x):# 应用第一个线性变换x = self.layer1(x)# 应用 dropout 进行正则化x = self.dropout(x)# 应用第二个线性变换以产生输出 logitsx = self.layer2(x)return x

基于验证损失的早停法。 如果验证损失停止改善，就没有必要继续训练了。

best_loss = float('inf')  # 将最佳损失初始化为无穷大
patience = 10             # 如果没有改善，则在停止前等待的 epoch 数
trigger_times = 0         # 记录没有改善的 epoch 次数for epoch in range(max_epochs):val_loss = validate(model, val_loader)  # 在验证集上评估if val_loss < best_loss:best_loss = val_loss     # 更新最佳损失trigger_times = 0        # 如果有改善，则重置计数器else:trigger_times += 1       # 没有改善，增加计数器if trigger_times >= patience:print('早停')  # 如果连续 'patience' 个 epoch 没有改善，则停止训练break

处理非常大的模型会带来新的挑战。以下是一些使其易于管理的方法。

模型并行将模型分布到多个 GPU 上。 模型的不同部分在不同的设备上处理。

# 定义一个顺序块并将其移动到 GPU 0
self.seq1 = nn.Sequential(# 层放在这里，例如 nn.Linear(...), nn.ReLU() 等
).to('cuda:0')# 定义另一个顺序块并将其移动到 GPU 1
self.seq2 = nn.Sequential(# 层放在这里
).to('cuda:1')# 定义一个全连接（或其他）层并将其移动到 GPU 1
self.fc = nn.Linear(...).to('cuda:1')

数据并行将数据分布到多个 GPU 上。 PyTorch DataParallel 自动管理此过程。

# 使用 DataParallel 包装模型
model = DataParallel(MyModel())
# 将模型移动到 CUDA 设备（默认是所有可用 GPU）
model.to('cuda')

梯度累积允许使用更大的批次。 当内存有限时，它通过在更新前累积梯度来提供帮助。

# 在开始累积之前重置梯度
optimizer.zero_grad()for i, (inputs, labels) in enumerate(training_set):# 前向传播outputs = model(inputs)# 计算损失loss = loss_function(outputs, labels)# 反向传播（累积梯度）loss.backward()# 每 'accumulation_steps' 次迭代执行一次优化器步骤if (i + 1) % accumulation_steps == 0:optimizer.step()        # 更新模型参数optimizer.zero_grad()   # 为下一次累积重置梯度

联邦学习将数据保留在本地设备上。 模型在设备上分别训练，只共享更新。

for round in range(num_rounds):model_updates = []  # 用于收集所有设备权重更新的列表# 每个设备在其本地数据上进行训练for device in devices:updated_model = train_on_device(model, device.data)  # 本地训练model_updates.append(updated_model.get_weights())    # 收集权重# 聚合更新（例如，取平均值）并更新全局模型model.set_weights(aggregate(model_updates))

为了在不损失太多性能的情况下使大型模型更高效，知识蒸馏是一种很好的方法。

使用大型教师模型训练小型学生模型。 这有助于减小模型大小，同时保持良好的准确性。

import torch.nn.functional as Fdef knowledge_distillation_loss(outputs, labels, teacher_outputs, temp=2.0, alpha=0.5):# 硬损失：学生预测和真实标签之间的标准交叉熵hard_loss = F.cross_entropy(outputs, labels)# 软损失：软化的学生预测和教师预测之间的 KL 散度# 应用温度来软化概率分布soft_loss = F.kl_div(F.log_softmax(outputs / temp, dim=1),          # 学生 logits（软化）F.softmax(teacher_outputs / temp, dim=1),      # 教师 logits（软化）reduction='batchmean'                          # 在批次上取平均)# 最终损失：硬损失和软损失的加权和# 按照原始 KD 论文的建议，将软损失乘以 temp^2return alpha * hard_loss + (1 - alpha) * soft_loss * (temp ** 2)

通过结合正确的优化器、正则化方法和训练策略，我们可以构建既强大又高效的模型，即使在大规模下也是如此。

让我们用一个比较表来更清晰地理解这一点。

使用 TensorFlow 和 PyTorch 进行扩展

框架在规模化处理 AI 时也扮演着重要角色。以下是一些流行的选项：

最重要的框架

1. TensorFlow 提供 TensorFlow Distributed Strategies 来帮助在 GPU 和 TPU 上高效扩展训练。
2. PyTorch 以 PyTorch Distributed 闻名，支持跨多个 GPU 和多台机器进行扩展。
3. Horovod 与 TensorFlow、PyTorch 和 Keras 配合使用，以提高在 GPU 和 CPU 上的可扩展性。
4. Kubernetes 有助于在规模化运行时平稳部署和管理 AI 工作负载。
5. CUDA 和 cuDNN 加速 GPU 计算和深度学习性能。
6. NeMo 专注于构建语音和自然语言处理模型。

模型扩展和高效处理

模型扩展是处理大数据集和复杂任务的关键。让我们探索一些简单的方法来并行化模型和数据，智能地处理批次，并应对训练挑战。

模型并行， 当模型对于单个 GPU 来说太大时，我们可以将模型分布到多个设备上。您可以按层（垂直）或层的部分（水平）进行划分。目标是减少设备之间的数据移动。

import torch
import torch.nn as nn# 定义一个简单模型
class SimpleModel(nn.Module):def __init__(self):super(SimpleModel, self).__init__()self.layer1 = nn.Linear(10, 20)self.relu = nn.ReLU()self.layer2 = nn.Linear(20, 10)self.layer3 = nn.Linear(10, 5)def forward(self, x):# 手动按设备拆分的前向传播x = self.layer1(x)x = self.relu(x)# 在继续之前将张量移动到第二个设备x = x.to(device2)x = self.layer2(x)x = self.relu(x)x = self.layer3(x)return x# 实例化模型
model = SimpleModel()# 定义设备
device1 = torch.device('cuda:0')
device2 = torch.device('cuda:1')# 将模型部分移动到各自的设备
model.layer1.to(device1)
model.relu.to(device1)  # 可选，因为 ReLU 是无状态且轻量级的
model.layer2.to(device2)
model.layer3.to(device2)# device1 上的示例输入张量
x = torch.randn(1, 10).to(device1)# 前向传播（在模型的 forward 方法内部处理）
output = model(x)# output 现在在 device2 上

我们可以使用像 NCCL 这样的快速通信库来减少移动数据时的延迟，并使用 torch.cuda.synchronize() 来确保设备按顺序完成任务。

import torch
import torch.distributed as dist############ NCL ############
def init_process(rank, size, backend='nccl'):# 初始化分布式进程组dist.init_process_group(backend, rank=rank, world_size=size)world_size = 4
for i in range(world_size):init_process(rank=i, size=world_size, backend='nccl') # 使用 NCCL 后端初始化进程############ 同步 ############
def synchronize_devices(devices):# 同步指定的 CUDA 设备for device in devices:if 'cuda' in str(device):torch.cuda.synchronize(device)device1 = torch.device('cuda:0')
device2 = torch.device('cuda:2')
synchronize_devices([device1, device2]) # 同步 device1 和 device2

数据并行， 我们可以在多个设备上运行相同的模型处理不同的数据块。当模型适合单个 GPU，但我们希望并行处理更多数据时，这很有用。

import torch
import torch.distributed as dist
from torch.nn.parallel import DistributedDataParallel as DDP
from torch.utils.data import DataLoader, Dataset, DistributedSampler# 虚拟数据集
class CustomDataset(Dataset):def __init__(self, data):self.data = datadef __len__(self):return len(self.data)def __getitem__(self, idx):return self.data[idx]# 初始化分布式训练
def setup(rank, world_size):# 使用 NCCL 后端初始化进程组dist.init_process_group(backend='nccl', rank=rank, world_size=world_size)# 设置当前 CUDA 设备torch.cuda.set_device(rank)# 创建带有分布式采样器的数据加载器
def get_dataloader(dataset, batch_size, rank, world_size):# 创建分布式采样器sampler = DistributedSampler(dataset, num_replicas=world_size, rank=rank)# 返回使用此采样器的数据加载器return DataLoader(dataset, batch_size=batch_size, sampler=sampler)# 示例用法
rank = 0
world_size = 2
setup(rank, world_size) # 设置分布式环境dataset = CustomDataset(torch.arange(1000)) # 创建数据集
dataloader = get_dataloader(dataset, batch_size=32, rank=rank, world_size=world_size) # 获取数据加载器model = SimpleModel().to(rank) # 将模型移动到指定设备
model = DDP(model, device_ids=[rank]) # 使用 DistributedDataParallel 包装模型

在反向传播之后，DDP 会在设备之间同步梯度，因此模型权重保持一致。

我们还可以通过梯度压缩来减少通信负载。这是一个使用 8 位量化的简单版本：

def quantize_gradients(model, bits=8):# 量化梯度到指定位数q_level = 2**bits - 1  # 量化级别数 (例如 8位 是 255)for param in model.parameters():if param.grad is not None:grad = param.grad.data  # 访问梯度张量# 计算最小值和最大值用于归一化min_val, max_val = grad.min(), grad.max()# 归一化到 [0, 1] 并缩放到 [0, q_level]grad_norm = (grad - min_val) / (max_val - min_val + 1e-8) * q_level# 量化：四舍五入到最近的级别grad_quant = torch.round(grad_norm)# 反量化：映射回原始尺度grad_dequant = grad_quant / q_level * (max_val - min_val) + min_val# 用量化-反量化后的版本替换原始梯度param.grad.data = grad_dequant

高效批处理， 我们可以通过调整批次处理方式来提高速度和内存使用率。

• 混合精度训练 使用半精度（float16）进行更快的计算：

from torch.cuda.amp import GradScaler, autocastscaler = GradScaler()  # 处理缩放以防止 float16 梯度下溢for data, target in dataloader:optimizer.zero_grad()  # 清除之前的梯度with autocast():  # 启用混合精度 — 在安全的地方使用 float16，否则使用 float32output = model(data)               # 前向传播（一些操作在 float16 中）loss = loss_fn(output, target)     # 计算损失（仍然在 float32 中）scaler.scale(loss).backward()         # 缩放损失以避免梯度下溢，然后反向传播scaler.step(optimizer)                # 如果没有溢出，则取消缩放梯度并调用 optimizer.step()scaler.update()                       # 调整下一次迭代的缩放比例

• 梯度累积 在 GPU 无法处理大批次时有所帮助：

optimizer.zero_grad() # 清零梯度
accum_steps = 4 # 梯度累积步数for i, (data, target) in enumerate(dataloader):output = model(data) # 前向传播# 缩放损失以在累积步骤中取平均loss = loss_fn(output, target) / accum_stepsloss.backward() # 反向传播，累积梯度# 每 accum_steps 个批次更新一次权重并重置梯度if (i + 1) % accum_steps == 0:optimizer.step()optimizer.zero_grad()# 处理剩余的梯度，如果总批次数不能被 accum_steps 整除
if (i + 1) % accum_steps != 0:optimizer.step()optimizer.zero_grad()

让我们了解同步和异步训练之间的基本区别：

同步训练， 所有工作节点在更新权重之前等待交换梯度。确保模型一致，但最慢的工作节点会拖慢所有人。

• 梯度平均
• 自适应批次大小
• 预测性等待时间调度

异步训练 工作节点无需等待即可更新权重。加快了速度，但梯度可能是过时的。

• 使用过时梯度校正
• 动态调整学习率
• 维护模型版本以跟踪更新

那么，到目前为止我们学到了什么，让我们在一个表格中总结一下：
（注意：原文在此处暗示有一个总结表格，但未提供。因此翻译中省略该表格。）

第三阶段：高级模型推理技术

当我们部署 ML 模型并有数百万人使用它们时，绝对需要一种高效的推理方法，使其易于所有用户访问。

我们经常遇到资源不像我们希望的那样容易获得的情况。在本节中，我们将探讨各种可以帮助我们使推理更加优化和有效的技术和策略。

大规模高效推理

模型量化 通过降低其数字的精度（例如从 32 位浮点数到 8 位整数）来帮助缩小模型并加快推理速度。这意味着更小的模型和更快的计算，但要注意可能会有一些准确性下降。

主要有两种类型：

量化类型

• 静态量化：在运行模型之前将权重转换为低精度。
• 动态量化：在推理期间动态转换权重和激活值，平衡速度和灵活性。

以下是如何使用 PyTorch 对 ResNet18 模型进行动态量化：

# ResNet18 动态模型量化示例
import torch
from torchvision.models import resnet18# 加载预训练的 ResNet18 模型
model = resnet18(pretrained=True)
model.eval()  # 设置为评估模式进行推理# 对线性层应用动态量化，以实现更快的推理和更小的模型尺寸
quantized_model = torch.quantization.quantize_dynamic(model, {torch.nn.Linear, torch.nn.Conv2d}, dtype=torch.qint8 # 指定对线性和卷积层量化，目标类型为 int8
)# 打印量化后的模型架构
print(quantized_model)

有两种最常用的训练技术在节省磁盘空间、内存以及最重要的成本方面发挥着重要作用。

• 训练后量化（Post-Training Quantization, PTQ）： 在训练之后量化模型。这种方法很快，但可能导致更多的准确性损失，因为模型在训练时不知道它将使用较低的精度。
• 量化感知训练（Quantization-Aware Training, QAT）： 模型在训练期间就考虑到量化进行学习，因此它能够适应，并且通常在量化后保持更好的准确性。

以下是如何使用 PyTorch 对 ResNet18 模型进行 QAT：

import torch
from torchvision.models import resnet18
import torch.quantizationmodel = resnet18(pretrained=True)
model.train() # 设置为训练模式# 融合 Conv、BatchNorm 和 ReLU 层以获得更好的量化效果
model = torch.quantization.fuse_modules(model, [['conv1', 'bn1', 'relu']]) # 融合第一个卷积块# 设置 QAT 配置
model.qconfig = torch.quantization.get_default_qat_qconfig('fbgemm') # 使用 fbgemm 后端的默认 QAT 配置# 准备模型进行 QAT（原地修改）
torch.quantization.prepare_qat(model, inplace=True)# ... 在此处添加您的训练循环以微调模型 ...# 训练后转换为量化版本
torch.quantization.convert(model, inplace=True) # 将模型转换为量化版本print(model) # 打印最终的量化模型

模型剪枝 通过移除不太重要的部分来帮助缩小模型——可以通过将单个权重置零或剪掉整个神经元或通道来实现。

• 非结构化剪枝： 将单个权重置零（模型中存在大量微小的“空洞”）。它可以使模型变得稀疏，但通常需要特殊的硬件/软件才能获得速度优势。
• 结构化剪枝： 移除整个通道或滤波器，这在常规硬件上更容易加速，但通常需要一些重新训练。

您也可以在训练期间动态地进行剪枝。

让我们看一个非结构化和结构化剪枝的例子。

# 非结构化模型剪枝示例
import torch
import torch.nn.utils.prune as prune
import torch.nn as nn# 定义一个简单的顺序模型
model = nn.Sequential(nn.Linear(10, 100),nn.ReLU(),nn.Linear(100, 2)
)# 指定要剪枝的参数：第一个和最后一个线性层的权重
parameters_to_prune = ((model[0], 'weight'),(model[2], 'weight')
)# 应用全局非结构化剪枝：
# 在所有指定参数中，剪枝掉 L1 范数最小的 20% 的权重
prune.global_unstructured(parameters_to_prune,pruning_method=prune.L1Unstructured, # 使用 L1 非结构化方法amount=0.2, # 剪枝比例为 20%
)print(model)  # 打印模型以查看附加的剪枝掩码# 结构化模型剪枝示例
import torch
import torch.nn.utils.prune as prune
import torch.nn as nn# 定义相同的模型
model = nn.Sequential(nn.Linear(10, 100),nn.ReLU(),nn.Linear(100, 2)
)# 对第一个线性层应用结构化剪枝：
# 基于维度 0（输出通道）上的 L2 范数，移除 50% 的整个通道（行）
prune.ln_structured(model[0],          # 要剪枝的模块name='weight',     # 要剪枝的参数amount=0.5,        # 要剪枝的通道比例n=2,               # 用于重要性评分的 L2 范数 (n=2)dim=0              # 要剪枝的维度（输出通道）
)print(model)  # 查看结构化剪枝的剪枝掩码

大规模高效推理 （续）

（注意：原文此处标题与上一节部分重复，但内容不同，推测为编辑错误，这里按内容翻译为“平台部署与优化”）

在不同平台（如手机、物联网设备或云服务器）上部署 ML 模型时，您需要针对每种环境调整方法：

• 模型简化： 删减那些对准确率贡献不大但消耗资源的层或操作。这有助于使模型更精简、更快，尤其是在资源受限的设备上。
• 硬件感知优化： 使您的模型适应可用的硬件能力，例如使用 GPU 优化或利用某些智能手机上的 NPU（神经处理单元）。
• 软件框架和工具： 使用为您的平台设计的部署工具。例如：TensorFlow Lite： 非常适合移动和边缘设备。ONNX Runtime： 跨多个平台工作，以实现一致的性能。

专用硬件可以带来巨大的速度提升：

• GPU： 非常适合并行任务，如矩阵乘法，这在深度学习中很常见。
• TPU： 谷歌的定制芯片，针对张量数学进行了优化，对于训练和推理都非常高效。
• FPGA： 灵活的硬件，可以为特定工作负载编程，提供出色的性能和能源效率。

假设优化 ML 系统推理的时间表为一年，计划可能如下：
（注意：原文在此处暗示有一个时间表示例，但未提供。因此翻译中省略。）

让我们看一个 LLM 推理优化示例：

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig
import timedef setup_model(optimized=True):# 加载 "facebook/opt-350m" 模型的 tokenizertokenizer = AutoTokenizer.from_pretrained("facebook/opt-350m")if optimized:# 设置 4 位加载的量化配置，计算精度为 float16quantization_config = BitsAndBytesConfig(load_in_4bit=True,bnb_4bit_compute_dtype=torch.float16)# 加载启用了 4 位量化的模型model = AutoModelForCausalLM.from_pretrained("facebook/opt-350m",quantization_config=quantization_config)# 将模型转换为 BetterTransformer 后端以实现更快的推理model = model.to_bettertransformer()else:# 加载标准的、未经优化的全精度模型model = AutoModelForCausalLM.from_pretrained("facebook/opt-350m")# 将模型移动到 GPU 以进行更快的计算return tokenizer, model.to("cuda")def generate_text(model, tokenizer, input_text, use_flash_attention=False):# 对输入文本进行分词并将张量移动到 GPUinputs = tokenizer(input_text, return_tensors="pt").to("cuda")if use_flash_attention:# 在支持的 GPU 上启用 FlashAttention 内核以进行高效的注意力计算with torch.backends.cuda.sdp_kernel(enable_flash=True, enable_math=False, enable_mem_efficient=False):outputs = model.generate(**inputs)else:# 不使用 FlashAttention 优化的常规生成outputs = model.generate(**inputs)# 将生成的 token ID 解码为字符串，跳过特殊 tokenreturn tokenizer.decode(outputs[0], skip_special_tokens=True)def measure_performance(input_text, optimized=True, use_flash_attention=False):# 设置模型和 tokenizer（优化版或标准版）tokenizer, model = setup_model(optimized)# 记录生成开始时间start_time = time.time()# 使用给定的模型和 tokenizer 生成文本result_text = generate_text(model, tokenizer, input_text, use_flash_attention)# 记录结束时间并计算耗时end_time = time.time()print(f"生成的文本: {result_text}")print(f"耗时: {end_time - start_time:.2f} 秒")# 示例输入提示
input_text = "Hello my dog is cute and"# 运行优化版本（4位量化 + BetterTransformer + FlashAttention）
print("运行优化版本:")
measure_performance(input_text, optimized=True, use_flash_attention=True)# 运行标准模型且不使用 FlashAttention
print("\\n运行未优化版本:")
measure_performance(input_text, optimized=False, use_flash_attention=False)

当您的模型训练完成并准备就绪时，下一个重要步骤是确保它在现实世界中运行良好。

管理实时应用的延迟和吞吐量

这意味着它应该处理大量请求，快速响应，并且无论如何都要保持可靠。让我们看看如何在生产环境中高效地扩展您的模型推理。

在生产环境中，请求可能会快速涌入。您希望将这些请求均匀地分配到您的服务器上，这样它们就不会过载。这就是负载均衡的作用——它使一切平稳运行。但如果做得不对，您的系统仍然会遇到困难。

• 自适应负载均衡： 这意味着根据服务器当前的繁忙程度将请求发送给它们，就像交通管制员将汽车引导到最不拥挤的道路一样。
• 资源分配： 巧妙地使用硬件非常重要。GPU 对于繁重任务很强大，但对于简单请求可能有点大材小用。根据请求的要求将工作分配给 GPU 或 CPU 有助于节省资源并加快速度。

import torchdef allocate_inference(request):"""根据请求复杂性将推理分配给 GPU 或 CPU 的简单函数。"""# 假设 'request' 有一个 'complexity' 属性（整数）if request.complexity > 5:device = torch.device("cuda:0")  # 对复杂请求使用 GPUelse:device = torch.device("cpu")     # 对简单请求使用 CPU# 在选定的设备上加载您的模型model = your_model.to(device) # your_model 应替换为你的实际模型变量# 继续进行推理...return model

当您的机器学习模型上线时，尤其是在实时应用中，您需要平衡两个重要因素：延迟（latency） 和 吞吐量（throughput）。

• 延迟是您获得响应的速度。
• 吞吐量是您在给定时间内可以处理的响应数量。

延迟和吞吐量（来自 Sina Torfi — 开源）

两者都非常重要，但有时改进一个可能会拖慢另一个。让我们用简单的方式来分析如何管理这两者。

在实时应用中，用户或系统不希望等待太久。您希望您的模型快速给出答案。

• 精简您的模型： 使用像量化和剪枝（前面已介绍）这样的技术来缩小模型并使其更快，而不会损失太多准确性。
• 更智能地服务： 您提供模型服务的方式很重要！在需要额外算力时使用 GPU，或使用像 TorchServe 这样的工具来有效管理请求，可以产生巨大的差异。

吞吐量关系到您的模型一次能处理多少请求，这在流量高峰期非常重要。

• 批处理（Batch Processing）： 将多个传入的请求组合在一起并一起处理，就像用一辆满载乘客的公共汽车代替为每个人单独派车一样。总体来说更快，但在批次移动之前需要一点时间来填满。
• 异步处理： 让您的系统在处理先前请求的同时处理新请求，就像您在做饭时还能接新订单一样。这可以保持一切顺畅流动。

平衡延迟和吞吐量意味着有时您需要根据实时情况改变一次处理的请求数量——即您的批次大小。

以下是使用 PyTorch 实现此目的的一种简单方法：

import torch
from queue import Queue
from threading import Thread# 定义一个简单的线性模型用于推理
model = torch.nn.Linear(10, 2)
model.eval()  # 将模型设置为评估模式def inference_worker(input_queue):while True:# 等待队列中的一批输入batch = input_queue.get()if batch is None:  # 检查退出信号breakwith torch.no_grad():  # 推理时禁用梯度计算output = model(batch)  # 对输入批次运行推理# (可选) 在此处根据需要处理输出input_queue.task_done()  # 标记任务已完成# 创建一个最大容量为 10 的队列来存放输入批次
input_queue = Queue(maxsize=10)# 启动一个工作线程，处理来自队列的输入批次
worker = Thread(target=inference_worker, args=(input_queue,))
worker.start()# 动态地将批次送入队列
for _ in range(100):# 创建一个包含 5 个样本，每个样本有 10 个特征的批次input_batch = torch.randn(5, 10)input_queue.put(input_batch)      # 将批次添加到队列等待处理input_queue.put(None)  # 向工作线程发送退出信号
worker.join()          # 等待工作线程完成

到目前为止，我们学到了……

这不仅仅是关于速度或容量——而是关于平衡快速响应和对高负载的平稳处理，以保持用户满意

边缘 AI 和移动部署

在智能手机和物联网小工具等边缘设备上部署 AI 模型意味着在数据产生的地方直接运行 AI。

这种设置减少了延迟，节省了网络带宽，并使您的数据更具隐私性，因为它不必离开设备。

为了让 AI 在这些资源有限的设备上良好工作，您需要专注于几个智能策略：

• 模型优化： 使用量化、剪枝和知识蒸馏等技术来缩小模型。更小的模型运行更快，更适合性能较弱的硬件。
• 边缘友好框架： 像 TensorFlow Lite 和 PyTorch Mobile 这样的工具就是为此而构建的。它们有助于转换和优化您的模型，使其在边缘设备上高效运行。

import tensorflow as tf# 将您的 TensorFlow 模型转换为 TensorFlow Lite 格式以进行边缘部署
converter = tf.lite.TFLiteConverter.from_keras_model(model) # model 应为你的 Keras 模型对象
tflite_model = converter.convert() # 执行转换# 将 TFLite 模型写入文件
with open('model.tflite', 'wb') as f:f.write(tflite_model)

边缘设备通常在处理能力、内存和电池寿命方面有严格的限制。因此，您的 AI 模型需要精简高效：

• 模型压缩： 使用剪枝或量化来缩小模型，以节省空间并加快推理速度。
• 节能算法： 选择或设计不会耗尽电池或使处理器过载的算法。
• 边缘优化架构： 使用像 MobileNet 或 EfficientNet 这样专门为快速、轻量且仍保持准确性而构建的网络。

第四阶段：性能分析与优化

在优化 AI 系统时，关键是找出导致速度变慢的地方——即瓶颈——这样您就可以修复它们并提升性能。

诊断系统瓶颈

这里有两个主要工具可以提供帮助：分析（profiling） 和 基准测试（benchmarking）。

分析 是深入挖掘，查看您的系统如何使用 CPU、GPU 和内存等资源，以及代码的不同部分运行需要多长时间。它就像一张性能地图，突出显示了您想要改进的缓慢或耗费资源的地方。

• Python 的 cProfile： 一个方便的内置工具，用于衡量您的 Python 代码大部分时间花在了哪里。
• NVIDIA Nsight Systems： 如果您使用 NVIDIA GPU，此工具可跟踪 GPU 性能并帮助查找 CUDA 代码中的瓶颈。

基准测试着眼于更大的图景：您的整个系统与标准或其他版本相比有多快和多高效。它设定了一个基线，这样您就知道自己从哪里开始，并可以衡量您的更改带来了多少帮助。

• 建立基线： 在进行任何更改之前，对您当前的系统进行基准测试。
• 比较： 检查您的系统与其他系统或行业基准的对比情况。
• 衡量影响： 优化后，再次进行基准测试，看看您的改进是否真的产生了效果。

瓶颈有不同的形式，如计算、内存或网络瓶颈，每种都需要不同的修复方法。

计算瓶颈 发生在您的处理器（CPU/GPU）跟不上工作量时。

修复方法：

• 使用并行计算：将工作分散到多个核心或 GPU 上以加快速度。
• 优化算法：简化计算或切换到更有效的方法以减轻负载。

内存瓶颈 发生在您的系统无法足够快地移动数据或内存不足时。

修复方法：

• 缓存常用数据以避免缓慢的内存读取。
• 使用剪枝、量化或更轻量级的数据结构来减少内存占用。
• 示例： 如果您的模型对于 GPU 内存来说太大，您可能需要这些技巧，因为您不能简单地添加更多 RAM。

网络瓶颈 出现在分布式系统中，数据需要在机器之间传输。

修复方法：

• 使用更好的数据序列化来缩小正在发送的数据的大小。
• 切换到更高效的通信协议，以降低延迟并加快数据传输速度。

AI 模型的操作化

密切关注系统的健康状况和 AI 模型的性能对于平稳、可靠的运营至关重要。良好的监控有助于及早发现问题，在它们变成大问题之前。以下是建立有效监控策略的直接方法：

您可以使用的工具：

（图片可能描述监控工具，如 Prometheus, Grafana 等）

• Prometheus： 一个开源工具，用于收集和存储指标，如 CPU 使用率、内存消耗、磁盘 I/O 和网络流量。它非常适合跟踪 AI 基础设施的整体健康状况。
• Grafana： 一个强大的可视化工具，与 Prometheus 配合良好，可以创建直观的仪表板。它有助于轻松发现系统数据中的异常和趋势。

模型性能监控 流行的选项包括：

• TensorBoard： 专为 TensorFlow 和 PyTorch 构建，TensorBoard 让您可以可视化训练和评估指标，如损失、准确率、权重分布，甚至模型的架构。定期检查这些有助于您了解模型的学习和表现情况。
• 自定义日志记录： 有时您需要跟踪特定于应用程序的指标或事件，而 TensorBoard 无法涵盖。实现自己的日志记录系统可以让您捕获预测、错误或任何自定义 KPI 以进行更深入的分析。

因此，一些最佳的监控技术是：

最佳监控技术

• 设置有意义的警报： 收集数据固然很好，但为关键指标定义阈值至关重要，这样一旦出现问题，您就会立即收到通知。警报可以帮助您在问题影响用户之前迅速采取行动。
• 监控数据质量： 您的模型的好坏取决于它获得的数据。注意数据漂移（输入数据随时间的变化）和可能降低性能的异常情况。例如，记录样本图像或数据批次可以帮助您及早发现变化。
• 持续评估： 定期使用新数据评估您的模型，以发现性能下降。当准确性或其他指标低于设定阈值时，自动触发警报或重新训练，确保您的模型保持有效。
• 指标异常检测： 使用基于 ML 的异常检测技术自动标记模型性能中的异常模式，这样您就可以在无需手动检查的情况下始终掌握潜在问题。
• 检测数据和概念漂移： 定期检查您的数据性质或问题本身是否正在发生变化。专门的漂移检测工具可以提醒您这些变化，促使模型更新或重新训练。
• 自动化重新训练流水线： 构建工作流，可以在新数据到达或性能下降时自动重新训练和重新部署模型。但要明智——设置严格的标准，以避免在实践中无关紧要的微小改进上浪费资源。

调试 AI 系统：工具和方法论

由于复杂的数据流和模型，调试 AI 很棘手。使用这些工具和方法：

• PyTorch Autograd Profiler： 检查 PyTorch 模型中的时间和内存使用情况。
• TensorFlow Debugger (tfdbg)： 检查张量值以发现诸如 NaN 或错误形状之类的错误。
• 交互式调试： 使用 Jupyter notebook 进行实时数据和模型检查。
• 高级分析： 像 NVIDIA Nsight 和 PyTorch Profiler 这样的工具可以分析 GPU 使用情况和硬件瓶颈以优化性能。

机器学习的 CI/CD 流水线

快速、可靠的模型更新是 AI 项目的关键。CI/CD 自动化了测试、集成和部署，以最小的人工干预使模型平稳运行。

能够快速测试和改进 ML 模型是构建成功 AI 系统的关键。通过使用 CI/CD（持续集成和持续部署），我们可以以最小的人工干预自动化测试、模型更新和部署。这使一切运行顺畅可靠。

ML 中的持续集成 (CI)

CI 意味着自动检查代码更改以及早发现问题。在 ML 中，它还包括检查数据、训练脚本和模型本身。

• 自动化测试：设置测试以检查数据质量、模型训练和预测。对小部分使用单元测试，对完整流水线使用集成测试。
• 版本控制：使用像 DVC 这样的工具来跟踪数据和模型的更改，就像对待代码一样。这有助于保持一切一致，并在需要时轻松回滚。

ML 中的持续部署 (CD)

CD 意味着自动将新模型投入生产，以便用户快速获得最新的改进。

• 模型服务：像 TensorFlow Serving 或 TorchServe 这样的工具有助于高效地提供模型服务并管理版本。
• Docker：使用 Docker 将您的模型及其所有依赖项打包在一起。这使得在任何地方运行模型变得容易。
• Jenkins + Kubernetes：使用 Jenkins 自动化诸如测试和部署模型之类的任务。与 Kubernetes 结合使用以在生产中扩展和管理模型。

使其更好地工作的额外工具

• 实验跟踪：像 MLflow 或 Weights & Biases 这样的工具有助于跟踪实验、模型指标和结果。
• 环境管理：使用像 Conda 或 Pipenv 这样的工具来管理 Python 包，并与 Docker 配对以实现开发和生产之间的一致性。
• 模型验证：设置自动化检查，以确保每个模型在部署前都满足性能标准（例如，准确性或精确度）。

结论

感谢阅读！无论您是开始新项目还是改进现有项目，我希望本指南对您的 AI 工作有所帮助。