万亿参数背后的算力密码：大模型训练的分布式架构与自动化运维全解析

from transformers import AutoTokenizer, AutoModelForSequenceClassification
import torch

# 加载预训练模型和分词器
tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased")
model = AutoModelForSequenceClassification.from_pretrained("bert-base-uncased", num_labels=2)

# 准备微调数据
texts = ["This is a positive review", "This is a negative review"]
labels = [1, 0]
encoding = tokenizer(texts, truncation=True, padding=True, return_tensors="pt")
input_ids = encoding["input_ids"]
attention_mask = encoding["attention_mask"]
labels = torch.tensor(labels)

# 微调过程
outputs = model(input_ids=input_ids, attention_mask=attention_mask, labels=labels)
loss = outputs.loss
loss.backward()

import torch
import torchvision.models as models
import torch.nn as nn

class ImageTextModel(nn.Module):
    def __init__(self):
        super(ImageTextModel, self).__init__()
        self.image_encoder = models.resnet50(pretrained=True)
        self.text_encoder = nn.TransformerEncoder(
            nn.TransformerEncoderLayer(d_model=512, nhead=8),
            num_layers=6
        )
        self.fusion_layer = nn.Linear(512 + 512, 128)
        self.classifier = nn.Linear(128, 10)

    def forward(self, images, texts):
        image_features = self.image_encoder(images).flatten(1)
        text_embeddings = nn.Embedding(len(vocab), 512)(texts)
        text_features = self.text_encoder(text_embeddings)
        text_features = text_features.mean(dim=1)
        fused_features = self.fusion_layer(torch.cat([image_features, text_features], dim=1))
        return self.classifier(fused_features)

import torch
import torch.distributed as dist
import torch.nn as nn
import torch.optim as optim
from torchvision import datasets, transforms
from torchvision.models import resnet18
from torch.nn.parallel import DistributedDataParallel as DDP

# 初始化分布式环境
dist.init_process_group(backend='nccl')
local_rank = dist.get_rank()
torch.cuda.set_device(local_rank)

# 加载数据
transform = transforms.Compose([
    transforms.ToTensor(),
    transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))
])
trainset = datasets.CIFAR10(root='./data', train=True,
                                        download=True, transform=transform)
train_sampler = torch.utils.data.distributed.DistributedSampler(trainset)
trainloader = torch.utils.data.DataLoader(trainset, batch_size=32,
                                          sampler=train_sampler)

# 定义模型、损失函数和优化器
model = resnet18(num_classes=10).to(local_rank)
ddp_model = DDP(model, device_ids=[local_rank])
criterion = nn.CrossEntropyLoss()
optimizer = optim.SGD(ddp_model.parameters(), lr=0.001, momentum=0.9)

# 训练过程
for epoch in range(10):
    running_loss = 0.0
    for i, data in enumerate(trainloader, 0):
        inputs, labels = data[0].to(local_rank), data[1].to(local_rank)
        optimizer.zero_grad()
        outputs = ddp_model(inputs)
        loss = criterion(outputs, labels)
        loss.backward()
        optimizer.step()
        running_loss += loss.item()
    print(f'Rank {local_rank}, Epoch {epoch + 1}, Loss: {running_loss / len(trainloader)}')

# 关闭分布式环境
dist.destroy_process_group()

def sum_even_numbers(lst):
    return sum(x for x in lst if x % 2 == 0)

import optuna
import torch
import torch.nn as nn
import torch.optim as optim
from torchvision import datasets, transforms
from torch.utils.data import DataLoader

# 定义模型
class SimpleNet(nn.Module):
    def __init__(self):
        super(SimpleNet, self).__init__()
        self.conv1 = nn.Conv2d(3, 16, kernel_size=3)
        self.fc = nn.Linear(16 * 30 * 30, 10)

    def forward(self, x):
        x = torch.relu(self.conv1(x))
        x = x.view(-1, 16 * 30 * 30)
        x = self.fc(x)
        return x

# 数据加载
transform = transforms.Compose([
    transforms.ToTensor(),
    transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))
])
trainset = datasets.CIFAR10(root='./data', train=True,
                                        download=True, transform=transform)
trainloader = DataLoader(trainset, batch_size=32)

# 目标函数
def objective(trial):
    model = SimpleNet()
    optimizer_name = trial.suggest_categorical('optimizer', ['Adam', 'SGD'])
    lr = trial.suggest_loguniform('lr', 1e-5, 1e-1)
    optimizer = getattr(optim, optimizer_name)(model.parameters(), lr=lr)
    criterion = nn.CrossEntropyLoss()

    for epoch in range(5):
        running_loss = 0.0
        for i, data in enumerate(trainloader, 0):
            inputs, labels = data[0], data[1]
            optimizer.zero_grad()
            outputs = model(inputs)
            loss = criterion(outputs, labels)
            loss.backward()
            optimizer.step()
            running_loss += loss.item()

    return running_loss / len(trainloader)

# 调优过程
study = optuna.create_study(direction='minimize')
study.optimize(objective, n_trials=10)

print('Best trial:')
best_trial = study.best_trial
print('  Value:', best_trial.value)
print('  Params:')
for key, value in best_trial.params.items():
    print('    {}: {}'.format(key, value))