PyTorch与Serverless架构结合

Posted 2023-01-29

PyTorch介绍

2017年1月，FAIR（Facebook AI Research）发布了PyTorch。其标志如下所示。PyTorch是在Torch基础上用Python语言重新打造的一款深度学习框架，Torch是用Lua语言打造的机器学习框架。但是Lua语言较为小众，导致Torch学习成本高，知名度不高。近几年来，PyTorch凭借其易用性、代码简洁灵活等特点逐渐有了超越TensorFlow的趋势。在学术界，PyTorch的地位已经超越TensorFlow，且PyTorch借助ONNX所带来的模型落地能力在工业界大放光彩。

PyTorch标志 

PyTorch如此流行与它的张量和动态计算图有关。和TensorFlow一样，PyTorch也有张量（Tensor）。而与TensorFlow不同的是，PyTorch中的张量是n维数组，类似于Numpy中的Ndarray。Numpy是Python中最主流的数据计算库之一。

PyTorch中的张量几乎是对Ndarray的扩展，且可以运行在GPU上，大大加快了运算速度。

PyTorch官网提供了非常方便的PyTorch框架安装指引，如下所示。

 

PyTorch框架安装指引 

只需要选择不同的PyTorch Build、OS，以及Language等信息，我们就可以生成对应的命令，在本地执行生成的命令就可以进行PyTorch的安装：

pip3 install torch torchvision torchaudio

PyTorch实践：图像分类系统

CIFAR-10是由Hinton的学生Alex Krizhevsky和Ilya Sutskever整理的一个用于识别普适物体的小型数据集。该数据集包含10个类别共60 000张图片，每张图片的大小为32×32，其中训练图像50 000张，测试图像10 000张。下图是一些示例。

CIFAR-10数据集示例 

本案例将基于CIFAR-10数据集快速入门PyTorch框架，并实现一个简单的图像分类系统。

1.开发前准备

在开始实现本案例之前，导入包括PyTorch等在内的依赖库：

import torch

import torch.nn as nn

import torch.nn.functional as F

import torchvision

import torchvision.transforms as transforms

通常在使用PyTorch的时候会用到两个依赖：

·torch是关于运算的包；

·torchvision则集成了常用数据集和经典的神经网络模型，比如ResNet。

在正式开始构建模型之前，准备好训练集和测试集，同时定义好数据预处理操作，这里仅将图像的RGB值归一化至0～1区间：

transform = transforms.Compose( [transforms.ToTensor(),  transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))])

cifar_train = torchvision.datasets.CIFAR10(root=./data, train=True,  download=True, transform=transform)

cifar_test = torchvision.datasets.CIFAR10(root=./data, train=False,   transform=transform)

PyTorch还提供了数据加载器DataLoader，以便在训练、测试过程中遍历数据集：

trainloader = torch.utils.data.DataLoader(cifar_train, batch_size=32, shuffle=True)

testloader = torch.utils.data.DataLoader(cifar_test, batch_size=32, shuffle=False)

在数据加载器Dataloader中，定义每一步训练使用32个样本，即这里的参数batch_size=32，并在训练时对训练数据集随机洗牌，对测试集不进行洗牌。这里定义一个简单的卷积神经网络模型：

class Net(nn.Module):    
    def __init__(self):        
        super(Net, self).__init__()        
        self.conv1 = nn.Conv2d(3, 6, 5)        
        self.pool = nn.MaxPool2d(2, 2)        
        self.conv2 = nn.Conv2d(6, 16, 5)        
        self.fc1 = nn.Linear(16 * 5 * 5, 120)        
        self.fc2 = nn.Linear(120, 84)        
        self.fc3 = nn.Linear(84, 10)    

    def forward(self, x):        
        x = self.pool(F.relu(self.conv1(x)))        
        x = self.pool(F.relu(self.conv2(x)))        
        x = x.view(-1, 16 * 5 * 5)        
        x = F.relu(self.fc1(x))        
        x = F.relu(self.fc2(x))        
        x = self.fc3(x)        
        return x net = Net()

它包含2个卷积层和3个全连接层，第一层卷积层接收大小为32×32的图像的数据，最后的全连接层产生10个类别的输出结果。

2.模型训练

在目标函数上，选择多分类交叉熵损失函数和随机梯度下降法（Stochastic Gradient Descent，SGD）作为优化器，学习率lr大小为0.001。

criterion = nn.CrossEntropyLoss()

optimizer = optim.SGD(net.parameters(), lr=0.001, momentum=0.9)

·SGD：梯度是一个矢量，它告诉模型如何改变权重，使损失变化最快。这个过程为梯度下降，因为它使用梯度使损失下降到最小值。随机使用某一批数据进行训练，那么这次训练就是随机的。这就是随机梯度下降法名字的由来。

·学习率：模型每一次梯度下降的跨步大小。其决定着目标函数能否收敛到局部最小值以及何时收敛到最小值。合适的学习率能够使目标函数在合适的时间内收敛到局部最小值。

之后循环遍历数据集，将得到的数据输入模型进行训练：

# 迭代次数为2次

nums_epoch = 2

for epoch in range(nums_epoch):    

# 初始化损失大小为0.0    

　　_loss = 0.0    

# 从数据加载器中得到数据集和对应标签    

　　for i, (inputs, labels) in enumerate(trainloader, 0):        

# 将数据和标签指定到对应设备，如CPU或GPU，GPU需指定到CUDA       

　　　　inputs, labels = inputs.to(device), labels.to(device)        

# 清空已有的梯度        

optimizer.zero_grad()        

# 训练数据输入模型，做前向传播，得到模型输出        

outputs = net(inputs)        

# 通过模型输出和对应的标签计算损失函数        

loss = criterion(outputs, labels)        

# 梯度反向传播       

loss.backward()        

# 更新优化器参数        

optimizer.step()        

# 累计损失值并打印        

_loss += loss.item()        

# 每2000步打印一次损失值        

if i % 2000 == 1999:            

　　print([%d, %5d] loss: %.3f % (epoch + 1, i + 1, _loss / 2000))            

　　_loss = 0.0

其中，nums_epoch表示迭代次数，inputs.to(device)和labels.to(device)都表示将数据转换到device指示的硬件设备上，device可以为CPU或者GPU设备。在获取模型的前向输出后计算损失函数的值，直接调用损失函数backward()完成后向传播，并用optimizer.step()更新优化器。下面是训练的日志：

[1, 2000] loss: 1.178

[1, 4000] loss: 1.200

[1, 6000] loss: 1.168

[1, 8000] loss: 1.175

[1, 10000] loss: 1.185

[1, 12000] loss: 1.165

[2, 2000] loss: 1.073

[2, 4000] loss: 1.066

[2, 6000] loss: 1.100

[2, 8000] loss: 1.107

[2, 10000] loss: 1.083

[2, 12000] loss: 1.103

3.模型评估

最后对模型进行评估：

correct, total = 0, 0

with torch.no_grad():    

　　for images, labels in testloader:        

　　　　outputs = net(images)        

　　　　_, predicted = torch.max(outputs, 1)        

　　　　total += labels.size(0)        

　　　　correct += (labels == predicted).sum().item()

　　　　print(Accuracy: %d %% % (100 * correct / total))

输出结果为：

Accuracy: 58 %

与Serverless架构结合：对姓氏进行分类

1.本地开发

参考PyTorch官方案例NLP FROM SCRATCH: CLASSIFYING NAMES WITH A CHARAC-TER-LEVEL RNN，通过PyTorch框架构建并训练基本的字符级RNN来对单词进行分类。训练完成之后，通过Python Web框架将该项目与Flask框架进行结合，并服务化。

首先根据姓氏进行分类的示例代码，在本地进行代码的编写以及项目的基本测试：

f

from __future__ import unicode_literals, print_function, division 
from io import open 
import glob 
import unicodedata 
import string 
import torch 
import torch.nn as nn 
from torch.autograd import Variable 
from flask import Flask, request 
app = Flask(__name__) 
all_letters = string.ascii_letters + " .,;" 
n_letters = len(all_letters) 
category_lines =  
all_categories = [] 
n_hidden = 128 
findFiles = lambda path: glob.glob(path) 
unicodeToAscii = lambda s: .join(c for c in unicodedata.normalize(NFD, s) 
if unicodedata.category(c) != Mn and c in all_letters): 
    readLines = lambda filename: [unicodeToAscii(line) for line in open(filename, encoding=utf-8).read().strip().split()] 
    letterToIndex = lambda letter: all_letters.find(letter) 
    for filename in findFiles(data/names/*.txt):    
        category = filename.split(/)[-1].split(.)[0]    
        all_categories.append(category)    
        lines = readLines(filename)    
        category_lines[category] = lines n_categories = len(all_categories) 

def letterToTensor(letter):    
    tensor = torch.zeros(1, n_letters)    
    tensor[0][letterToIndex(letter)] = 1    
    return tensor 

def lineToTensor(line):    
    tensor = torch.zeros(len(line), 1, n_letters)    
    for li, letter in enumerate(line):        
        tensor[li][0][letterToIndex(letter)] = 1    
        return tensor 

class RNN(nn.Module):    
    def __init__(self, input_size, hidden_size, output_size):        
        super(RNN, self).__init__()        
        self.hidden_size = hidden_size        
        self.i2h = nn.Linear(input_size + hidden_size, hidden_size)        
        self.i2o = nn.Linear(input_size + hidden_size, output_size)        
        self.softmax = nn.LogSoftmax(dim=1)    

def forward(self, input, hidden):        
    combined = torch.cat((input, hidden), 1)        
    hidden = self.i2h(combined)        
    output = self.i2o(combined)        
    return output, hidden    

def initHidden(self):        
    return Variable(torch.zeros(1, self.hidden_size)) 

rnn = RNN(n_letters, n_hidden, n_categories) 

def evaluate(line_tensor):    
    hidden = rnn.initHidden()    
    for i in range(line_tensor.size()[0]):        
        output, hidden = rnn(line_tensor[i], hidden)    
        return output 

def predict(input_line, n_predictions=3):    
    with torch.no_grad():        
        output = evaluate(lineToTensor(input_line))        
        topv, topi = output.topk(n_predictions, 1, True)        
        predictions = [[topv[0][i].item(), all_categories[topi[0][i].item()]] for i in range(n_predictions)]    
        return predictions 


@app.route(/invoke, methods=[POST]) 
def invoke():    
    return result: predict(request.get_data().decode("utf-8")) 

if __name__ == __main__:    
    app.run(debug=True, host=0.0.0.0, port=9000)

之后，通过Python命令启动该Bottle项目，并通过命令行工具进行相关的测试：

curl --location --request POST http://0.0.0.0:9000/invoke \\ --header Content-Type: text/plain \\ --data-raw bai

输出的测试结果如下：

"result": [ [ 0.09027218818664551,  "Russian" ],  [ 0.07011377066373825,  "Chinese" ],   [ 0.053722310811281204,  "Portuguese" ] ]

可以看到，当输入一个姓氏之后，系统已经可以按照预期进行相关返回，包括所属国家信息以及相关度信息。

2.部署到Serverless架构

目前，各大云厂商的FaaS平台均支持容器镜像的部署。所以，我们可以将项目打包成镜像，并通过Serverless Devs开发者工具部署到阿里云函数计算。

若通过Serverless Devs开发者工具构建镜像并部署到阿里云函数计算，我们需要准备Doc-kerfile文件与Serverless Devs的资源描述文件。其中，Dockerfile文件参考如下：

FROM python:3.7-slim

WORKDIR /usr/src/app

RUN pip install torch flask numpy

COPY . .

CMD [ "python", "-u", "/usr/src/app/index.py" ]

Serverless Devs的资源描述文件是对部署到线上的资源进行预描述，包括服务相关配置、函数相关配置以及触发器、自定义域名等相关的配置：

edition: 1.0.0
name: container-pytorch
access: default 
vars:    
  region: cn-shanghai 
  services:    
    pytorch-demo:        
      component: devsapp/fc        
      props:            
        region: $vars.region            
        service:                
          name: pytorch-service            
          function:                
            name: pytorch-function                
            timeout: 60                
            caPort: 9000                
            memorySize: 1536                
            runtime: custom-container                
            customContainerConfig:                    
              image: registry.cn-shanghai.aliyuncs.com/custom-container/pytorch-demo:0.0.1

完成资源准备之后，通过Serverless Devs开发者工具中FC组件提供的build能力进行镜像的构建，例如执行s build--use-docker命令，即可看到预期的镜像构建效果，如下所示。

镜像构建效果示意图 

镜像构建完成之后，可以通过Serverless Devs开发者工具执行s deploy--push-registry acr-internet--use-local-y进行部署。这里主要包括以下几个动作。

·将构建完成的镜像推送到阿里云镜像服务。

·基于函数计算创建服务。

·基于函数计算创建函数，并指定代码源为指定的容器镜像。

·进行触发器和自定义域名的创建。

部署完成后，可以看到系统返回的测试地址，如下所示。

应用创建示意图 

此时，可以通过该测试地址，利用curl命令行测试工具进行测试：

curl --location --request POST http://pytorch-function.pytorch-service.1583208943291465. cn-shanghai.fc.devsapp.net/invoke\\ --header Content-Type: text/plain \\ --data-raw bai

之后，可以看到接口已经返回预测结果：

"result": [ [  0.1394740492105484,  "Arabic"  ],   [  0.06561967730522156,  "Dutch" ],     [   0.04731455445289612,   "Portuguese" ]  ]

至此，通过PyTorch完成了一个简单的文本分类功能，并通过部署到Serverless架构，暴露可以对外提供服务的API。

3.项目优化

Serverless架构的发展非常迅速，面临的挑战也有目共睹。尽管本实例采用了更为传统和简单的容器镜像部署方案，即将应用部署到阿里云Serverless平台，但是由于目前Server-less架构发展受限制，仍然存在诸多不足。

·基于自定义镜像的函数计算项目虽然更容易部署和迁移，但是冷启动问题非常严峻。至少目前来看，相对原生的运行时，容器镜像的冷启动问题要严峻不少。若想缓解镜像部署带来的冷启动问题，我们可以考虑使用镜像加速、预留实例等技术。

·PyTorch可以基于GPU实现预测，而且GPU被广泛应用到各行业的人工智能项目中。但是就目前来看，大部分厂商的Serverless架构还不支持GPU实例，所以在Serverless架构下如何使用GPU，以及是否能使用GPU将成为人工智能项目部署到Serverless架构的关键参考指标。目前，阿里云函数计算已经支持GPU实例。在本项目部署过程中，可以考虑GPU实例的技术选型，以提升预测性能。