准备环境
CUDA
TORCH
过程
训练模型
保存模型成文件
flask api调用模型
代码
模型训练
import torch
import torchvision
import torchvision.transforms as transforms
from torch import optim
import torch.nn.functional as F
from torch import nnclass Net(nn.Module):def __init__(self):super(Net, self).__init__()self.conv1 = nn.Conv2d(1, 32, kernel_size=3)self.conv2 = nn.Conv2d(32, 64, kernel_size=3)self.fc1 = nn.Linear(64 * 12 * 12, 128)self.fc2 = nn.Linear(128, 10)def forward(self, x):x = F.relu(self.conv1(x))x = F.relu(self.conv2(x))x = F.max_pool2d(x, 2)x = x.view(-1, 64 * 12 * 12)x = F.relu(self.fc1(x))x = self.fc2(x)return F.log_softmax(x, dim=1)# 在需要生成随机数的程序中,确保每次运行程序所生成的随机数都是固定的,使得实验结果一致
torch.manual_seed(1)
batch_size_train = 64
batch_size_valid = 64
batch_size_test = 1000transform = transforms.Compose([transforms.ToTensor(),transforms.Normalize((0.1307,), (0.3081,))
])trainset = torchvision.datasets.MNIST(root='./data', train=True, download=True, transform=transform)
trainloader = torch.utils.data.DataLoader(trainset, batch_size=batch_size_train)testset = torchvision.datasets.MNIST(root='./data', train=False, download=True, transform=transform)
indices = range(len(testset))# 测试集中再取出一半作为验证集
indices_valid = indices[:5000]
sampler_valid = torch.utils.data.sampler.SubsetRandomSampler(indices_valid)
validloader = torch.utils.data.DataLoader(testset, batch_size=batch_size_valid, sampler=sampler_valid)indices_test = indices[5000:]
sampler_test = torch.utils.data.sampler.SubsetRandomSampler(indices_test)
testloader = torch.utils.data.DataLoader(testset, batch_size=batch_size_test, sampler=sampler_test)import matplotlib.pyplot as pltexamples = enumerate(trainloader)
batch_idx, (example_data, example_targets) = next(examples)
fig = plt.figure()
for i in range(6):plt.subplot(2, 3, i + 1)plt.tight_layout()plt.imshow(example_data[i][0], cmap='gray', interpolation='none')plt.title('Ground Truth: {}'.format(example_targets[i]))plt.xticks([])plt.yticks([])
plt.show()
print(example_data.shape)# 检查 GPU 并设置设备
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")model = Net().to(device)
optimizer = optim.Adam(model.parameters(), lr=0.001)# 训练和测试过程中的数据也需要移动到 GPU
def train(epoch):model.train()for batch_idx, (data, target) in enumerate(trainloader):data, target = data.to(device), target.to(device) # 移动到 GPUoptimizer.zero_grad()output = model(data)loss = F.nll_loss(output, target)loss.backward()optimizer.step()if batch_idx % 100 == 0:print(f'Train Epoch: {epoch} [{batch_idx * len(data)}/{len(trainloader.dataset)} 'f'({100. * batch_idx / len(trainloader):.0f}%)]\tLoss: {loss.item():.6f}')def test(loader, mode="Test"):model.eval()test_loss = 0correct = 0with torch.no_grad():for data, target in loader:data, target = data.to(device), target.to(device) # 移动到 GPUoutput = model(data)test_loss += F.nll_loss(output, target, reduction='sum').item()pred = output.argmax(dim=1, keepdim=True)correct += pred.eq(target.view_as(pred)).sum().item()test_loss /= len(loader.dataset)accuracy = 100. * correct / len(loader.dataset)print(f'{mode} set: Average loss: {test_loss:.4f}, Accuracy: {correct}/{len(loader.dataset)} 'f'({accuracy:.0f}%)')# 训练和验证
for epoch in range(1, 11):train(epoch)test(validloader, mode="Validation")# 测试模型
test(testloader)# 保存模型参数
torch.save(model.state_dict(), 'mnist_cnn.pth')# 加载模型参数
# model = Net()
# model.load_state_dict(torch.load('mnist_cnn.pth'))# 或者保存整个模型
torch.save(model, 'mnist_cnn_full.pth')# 或者加载整个模型
# model = torch.load('mnist_cnn_full.pth')
# model.eval() # 设置为评估模式
使用刚才训练好的模型
from flask import Flask, request, jsonify
from torchvision import transforms
from PIL import Image
import io
import torch
import torch.nn.functional as F
from torch import nnclass Net(nn.Module):def __init__(self):super(Net, self).__init__()self.conv1 = nn.Conv2d(1, 32, kernel_size=3)self.conv2 = nn.Conv2d(32, 64, kernel_size=3)self.fc1 = nn.Linear(64 * 12 * 12, 128)self.fc2 = nn.Linear(128, 10)def forward(self, x):x = F.relu(self.conv1(x))x = F.relu(self.conv2(x))x = F.max_pool2d(x, 2)x = x.view(-1, 64 * 12 * 12)x = F.relu(self.fc1(x))x = self.fc2(x)return F.log_softmax(x, dim=1)# 初始化 Flask
app = Flask(__name__)# 检查是否有可用的 GPU
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")# 定义预处理变换
transform = transforms.Compose([transforms.Grayscale(num_output_channels=1),transforms.Resize((28, 28)),transforms.ToTensor(),transforms.Normalize((0.1307,), (0.3081,))
])# 加载模型(仅加载权重,无需重新训练模型)
model = Net().to(device)
model.load_state_dict(torch.load('mnist_cnn.pth', map_location=device))
model.eval() # 设置模型为评估模式,而不是训练模式# 预测函数
def predict(image_bytes):# 将二进制图片转换为 PIL 图像image = Image.open(io.BytesIO(image_bytes))# 对图片进行预处理image = transform(image).unsqueeze(0).to(device)# 模型推理with torch.no_grad():output = model(image)prediction = output.argmax(dim=1, keepdim=True).item()return prediction# 定义上传图片接口
@app.route('/predict', methods=['POST'])
def upload():if 'file' not in request.files:return jsonify({'error': 'No file provided'}), 400file = request.files['file']if file.filename == '':return jsonify({'error': 'No file selected'}), 400# 获取图片并做预测image_bytes = file.read()prediction = predict(image_bytes)return jsonify({'prediction': prediction})# 运行 Flask 应用
if __name__ == '__main__':app.run(host='0.0.0.0', port=5000)