准备环境
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 nn
class 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 = 1000
transform = 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 plt
examples = 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) # 移动到 GPU
optimizer.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 = 0
correct = 0
with torch.no_grad():
for data, target in loader:
data, target = data.to(device), target.to(device) # 移动到 GPU
output = 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 nn
class 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'}), 400
file = 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)