一文读懂大规模AI模型：原理、应用与实战

from transformers import BartForConditionalGeneration, BartTokenizer
import torch

# 加载 BART 模型和分词器
model_name = "facebook/bart-base"
tokenizer = BartTokenizer.from_pretrained(model_name)
model = BartForConditionalGeneration.from_pretrained(model_name).to("cuda" if torch.cuda.is_available() else "cpu")

# 输入文本
text = "人工智能通过自动化和数据分析，正在改变医疗、教育和交通等领域。"
inputs = tokenizer(text, return_tensors="pt", max_length=512, truncation=True).to(model.device)

# 生成摘要
summary_ids = model.generate(inputs["input_ids"], max_length=30, num_beams=4)
summary = tokenizer.decode(summary_ids[0], skip_special_tokens=True)

print("摘要：", summary)

import torch
from torchvision.models import resnet18, ResNet18_Weights
import torchvision.transforms as transforms
from PIL import Image

# 确定设备
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

# 加载预训练模型
model = resnet18(weights=ResNet18_Weights.DEFAULT).eval().to(device)

# 获取模型预处理转换方式
transform = ResNet18_Weights.DEFAULT.transforms()

# 加载图像
image = Image.open("C:\\Users\\Administrator\\PyCharmMiscProject\\picture\\images.jpg")
input_tensor = transform(image).unsqueeze(0).to(device)

# 执行预测
with torch.no_grad():
    output = model(input_tensor)
    predicted_class = torch.argmax(output, dim=1).item()

print("预测类别ID：", predicted_class)

import torch
import torch.nn as nn
from torch.utils.data import DataLoader
from torchvision.datasets import MNIST
from torchvision import transforms

# 定义推荐模型
class Recommender(nn.Module):
    def __init__(self):
        super(Recommender, self).__init__()
        self.network = nn.Sequential(
            nn.Linear(784, 256),
            nn.ReLU(),
            nn.Linear(256, 10),
        )

    def forward(self, x):
        return self.network(x.view(-1, 784))

# 加载 MNIST 数据集
dataset = MNIST(root='.', train=True, download=True, transform=transforms.ToTensor())
loader = DataLoader(dataset, batch_size=64, shuffle=True)

# 训练模型
model = Recommender()
optimizer = torch.optim.SGD(model.parameters(), lr=0.01)
criterion = nn.CrossEntropyLoss()

for epoch in range(5):
    total_loss = 0
    for images, labels in loader:
        outputs = model(images)
        loss = criterion(outputs, labels)
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        total_loss += loss.item()

    avg_loss = total_loss / len(loader)
    print(f"Epoch {epoch + 1}, Average Loss: {avg_loss:.4f}")

import os
import torch
import torch.nn as nn
import torch.distributed as dist
from torch.nn.parallel import DistributedDataParallel as DDP
from torch.utils.data import DataLoader, Dataset, DistributedSampler

# 设置通信环境变量
os.environ['MASTER_ADDR'] = 'localhost'
os.environ['MASTER_PORT'] = '12345'
os.environ['RANK'] = '0'
os.environ['WORLD_SIZE'] = '1'

# 初始化分布式训练
dist.init_process_group(backend='gloo')
rank = dist.get_rank()
world_size = dist.get_world_size()

# 指定设备
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

# 自定义简单数据集
class RandomDataset(Dataset):
    def __len__(self):
        return 100

    def __getitem__(self, idx):
        x = torch.randn(10)
        y = torch.randint(0, 5, (1,)).item()
        return x, y

# 构建数据加载器
dataset = RandomDataset()
sampler = DistributedSampler(dataset, num_replicas=world_size, rank=rank, shuffle=True)
dataloader = DataLoader(dataset, batch_size=32, sampler=sampler, num_workers=0)

# 定义模型
class SimpleModel(nn.Module):
    def __init__(self):
        super(SimpleModel, self).__init__()
        self.fc = nn.Linear(10, 5)

    def forward(self, x):
        return self.fc(x)

# 包装为分布式模型
model = SimpleModel().to(device)
model = DDP(model, device_ids=[rank] if torch.cuda.is_available() else None)

# 定义损失函数和优化器
criterion = nn.CrossEntropyLoss()
optimizer = torch.optim.SGD(model.parameters(), lr=0.001)

# 执行训练循环
for epoch in range(5):
    dataloader.sampler.set_epoch(epoch)
    for data, target in dataloader:
        data, target = data.to(device), target.to(device)
        output = model(data)
        loss = criterion(output, target)
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
    print(f"Epoch {epoch + 1} completed on rank {rank}")

# 清理通信资源
dist.destroy_process_group()

import torch.quantization

# 对模型执行动态量化（以线性层为例）
quantized_model = torch.quantization.quantize_dynamic(
    model, {torch.nn.Linear}, dtype=torch.qint8
)

import requests

API_KEY = "sk-xxx" #复制的令牌填写进去

# 修正后的URL（假设使用/v1/chat/completions路径）
url = "https://poloai.top/v1/chat/completions" #按照这个如实填写

headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json"
}

payload = {
    "model": "claude-3-5-haiku-20241022",
    "messages": [
        {"role": "user", "content": "请简要总结大规模AI模型的训练难点和优化策略。"}
    ],
    "temperature": 0.7,
    "max_tokens": 300
}

response = requests.post(url, headers=headers, json=payload)

if response.status_code == 200:
    result = response.json()
    # 根据实际返回格式调整
    reply = result.get("choices", [{}])[0].get("message", {}).get("content", "")
    print("模型回复：\n", reply)
else:
    print(f"请求失败，状态码：{response.status_code}\n响应内容：{response.text}")