Serverless架构下用Python轻松实现图像分类和预测
图像分类是人工智能领域的一个热门话题。通俗解释就是，图像分类是一种根据各自在图像信息中所反映的不同特征，把不同类别的目标区分开来的图像处理方法。它利用计算机对图像进行定量分析，把图像或图像中的每个像元或区域划归为若干个类别中的某一种，以代替人的视觉判断。图像分类在实际生产生活中经常遇到，而且对不同领域或者需求有着很强的针对性。例如通过对花朵拍照识别花朵信息，通过人脸识别匹配人物信息等。
通常情况下，这些图像识别或者分类的工具，都是在客户端进行数据采集，在服务端进行运算获得结果，即一般情况下都有专门的API实现图像识别。例如各大云厂商都会有偿提供类似的能力。
阿里云图像识别页面如图8-1所示。
图8-1 阿里云图像识别页面
华为云图像识别页面如图8-2所示。
图8-2 华为云图像识别页面
8.1.1 ImageAI与图像识别
本节将会通过一个有趣的Python库，快速将图像分类的功能搭建在云函数上，并且与API网关结合，对外提供API功能，实现一个Serverless架构的图像分类API。
首先和大家介绍一下需要的依赖库：ImageAI。通过该依赖库的官方文档可以看到这样的描述：ImageAI是一个Python库，旨在使开发人员能够使用简单的几行代码构建具有包含深度学习和计算机视觉功能的应用程序和系统。
ImageAI本着简洁的原则，支持最先进的机器学习算法，用于图像预测、自定义图像预测、物体检测、视频检测、视频对象跟踪和图像预测训练。ImageAI目前支持使用在ImageNet-1000数据集上训练的4种不同机器学习算法进行图像预测和训练，还支持使用在COCO数据集上训练的RetinaNet进行对象检测、视频检测和对象跟踪。最终，ImageAI将为计算机视觉提供更广泛、更专业化的支持，包括但不限于特殊环境和特殊领域的图像识别。
也就是说，ImageAI可以帮助我们完成基本的图像识别和视频的目标提取，虽然它给出了一些数据集和模型，但是也可以根据自身需要进行额外的训练，以及定制化拓展。通过官方提供的代码，可以看到一个简单的Demo：
# -*- coding: utf-8 -*- from imageai.Prediction import ImagePrediction #模型加载 prediction = ImagePrediction() prediction.setModelTypeAsResNet() prediction.setModelPath("resnet50_weights_tf_dim_ordering_tf_kernels.h5") prediction.loadModel() predictions, probabilities = prediction.predictImage("./picture.jpg", result_count=5 ) for eachPrediction, eachProbability in zip(predictions, probabilities): print(str(eachPrediction) + " : " + str(eachProbability))
待识别的picture.jpg图片如图8-3所示。
图8-3 待识别图片案例
执行后的结果如下：
laptop : 71.43893241882324 notebook : 16.265612840652466 modem : 4.899394512176514 hard_disc : 4.007557779550552 mouse : 1.2981942854821682
如果在使用过程中觉得模型resnet50_weights_tf_dim_ordering_tf_kernels.h5过大，耗时过长，可以按需求选择模型：
- SqueezeNet（文件大小：4.82 MB，预测时间最短，精准度适中）；
- ResNet50 by Microsoft Research（文件大小：98 MB，预测时间较快，精准度高）；
- InceptionV3 by Google Brain team（文件大小：91.6 MB，预测时间慢，精度更高）；
- DenseNet121 by Facebook AI Research（文件大小：31.6 MB，预测时间较慢，精度最高）。
模型下载地址可参考GitHub地址（https://github.com/OlafenwaMoses/ImageAI/releases/tag/1.0）或者ImageAI官方文档（https://imageai-cn.readthedocs.io/zh_CN/latest/ImageAI_Image_Prediction.html）。
8.1.2 项目Serverless化
当在本地完成ImageAI的基本测试运行之后，可以将项目部署到Serverless架构上。该项目的运行流程如图8-4所示。
图8-4 基于Serverless架构的图像识别功能流程简图
这里涉及几个部分，列举如下：
- 获取本地上传的图片；
- 图片缓存到/tmp/目录下；
- 通过AI进行预测；
- 返回数据。
获取本地上传的图片，主要获取Body中Base64编码后的图片信息，例如：
try: request_body_size = int(environ.get('CONTENT_LENGTH', 0)) except (ValueError): request_body_size = 0 requestBody = json.loads(environ['wsgi.input'].read(request_body_size).decode("utf-8"))
图片缓存到/tmp/目录下，即将获得的Base64编码后的图片进行解码，并存储到/tmp/目录下，例如：
#随机字符串 randomStr = lambda num=5: "".join(random.sample('abcdefghijklmnopqrstuvwxyz', num)) imageName = randomStr(10) imageData = base64.b64decode(requestBody["image"]) imagePath = "/tmp/" + imageName with open(imagePath, 'wb') as f: f.write(imageData)
通过ImageAI进行预测，则和本地测试环节的实现类似：
#内容预测 print("Predicting ... ") result = {} predictions, probabilities = prediction.predictImage(imagePath, result_count=5) print(zip(predictions, probabilities)) for eachPrediction, eachProbability in zip(predictions, probabilities): result[str(eachPrediction)] = str(eachProbability)
最后返回数据，完整的基于阿里云函数计算的代码实现为：
# -*- coding: utf-8 -*- from imageai.Prediction import ImagePrediction import json import uuid import base64 import random # Response class Response: def __init__(self, start_response, response, errorCode=None): self.start = start_response responseBody = { 'Error': {"Code": errorCode, "Message": response}, } if errorCode else { 'Response': response } #默认增加uuid，便于后期定位 responseBody['ResponseId'] = str(uuid.uuid1()) print("Response: ", json.dumps(responseBody)) self.response = json.dumps(responseBody) def __iter__(self): status = '200' response_headers = [('Content-type', 'application/json; charset=UTF-8')] self.start(status, response_headers) yield self.response.encode("utf-8") #随机字符串 randomStr = lambda num=5: "".join(random.sample('abcdefghijklmnopqrstuvwxyz', num)) #模型加载 print("Init model") prediction = ImagePrediction() prediction.setModelTypeAsResNet() print("Load model") prediction.setModelPath("/mnt/auto/model/resnet50_weights_tf_dim_ordering_tf_kernels.h5") prediction.loadModel() print("Load complete") def handler(environ, start_response): try: request_body_size = int(environ.get('CONTENT_LENGTH', 0)) except (ValueError): request_body_size = 0 requestBody = json.loads(environ['wsgi.input'].read(request_body_size).decode("utf-8")) #图片获取 print("Get pucture") imageName = randomStr(10) imageData = base64.b64decode(requestBody["image"]) imagePath = "/tmp/" + imageName with open(imagePath, 'wb') as f: f.write(imageData) #内容预测 print("Predicting ... ") result = {} predictions, probabilities = prediction.predictImage(imagePath, result_count=5) print(zip(predictions, probabilities)) for eachPrediction, eachProbability in zip(predictions, probabilities): result[str(eachPrediction)] = str(eachProbability) return Response(start_response, result)
上述的业务逻辑中需要相关的依赖，包括ImageAI所必需的依赖，例如Tensorflow、Numpy等：
tensorflow==1.13.1 numpy==1.19.4 scipy==1.5.4 opencv-python==4.4.0.46 pillow==8.0.1 matplotlib==3.3.3 h5py==3.1.0 keras==2.4.3 imageai==2.1.5
最后可以通过开源的Serverless开发者工具Serverless Devs将项目部署到线上。为了确保项目通过Serverless Devs部署，需要编辑Serverless Devs所需要的资源描述文档。编写后的资源文档（s.yaml）为：
edition: 1.0.0 name: ServerlessBook provider: alibaba access: anycodes_release services: ServerlessBookImageAIDemo: component: devsapp/fc props: region: cn-hongkong service: name: serverless-book-case description: Serverless实践图书案例 vpcConfig: auto logConfig: auto nasConfig: auto Function: name: serverless_imageAI description:图片目标检测 codeUri: src: ./src excludes: - src/.fun - src/model handler: index.handler environmentVariables: PYTHONUSERBASE: /mnt/auto/.fun/python memorySize: 3072 runtime: python3 timeout: 60 triggers: - name: ImageAI type: http config: authType: anonymous methods: - GET - POST - PUT customDomains: - domainName: auto protocol: HTTP routeConfigs: - path: /* serviceName: serverless-book-case functionName: ai-cv-image-christmas-hat
在代码与配置中，可以看到存在目录：/mnt/auto/，该部分实际上是nas挂载之后的地址，只需提前写入代码中即可。下一个环节会进行nas的创建以及挂载点配置的具体操作。
8.1.3 项目部署与测试
当完成业务逻辑的编写以及相关资源描述文档的编写之后，可以通过Serverless Devs的fc组件，将项目部署到阿里云函数计算中。在fc组件中，部署的操作指令是deploy，所以可以执行：
s deploy
稍等片刻，可以看到命令行执行完毕，并输出了函数的相关信息，以及触发器的相关信息，如图8-5所示。
图8-5 Serverless Devs阿里云函数计算组件部署
此时已经完成了业务逻辑的部署，但是代码实际还不能运行，因为还有相关的依赖没有安装和上传到函数计算中。接下来可以通过fc组件提供的install方法，在Docker中进行依赖的安装。依赖安装完成后，会在项目目录下生成.fun目录，该目录就是通过Docker打包出来的依赖文件，而这些依赖正是在requirements.txt文件中声明的依赖内容。
接着，可以将依赖同步到nas中：
s ServerlessBookImageAIDemo nas sync ./src/.fun
将依赖目录打包上传到nas成功之后，再将model目录上传到nas中：
s nas sync ./src/model
然后，可以编写脚本进行测试。通过函数的触发、预测，获得结果，示例代码为：
# -*- coding: utf-8 -*- import json import urllib.request import base64 import time with open("picture.jpg", 'rb') as f: data = base64.b64encode(f.read()).decode() url = 'http://35685264-1295939377467795.test.functioncompute.com/' timeStart = time.time() print(urllib.request.urlopen(urllib.request.Request( url=url, data=json.dumps({'image': data}).encode("utf-8") )).read().decode("utf-8")) print("Time: ", time.time() - timeStart)
执行示例代码后，可以看到如下结果：
{"Response": {"laptop": "71.43893837928772", "notebook": "16.265614330768585", "modem": "4.899385944008827", "hard_disc": "4.007565602660179", "mouse": "1.2981869280338287"}, "ResponseId": "1d74ae7e-298a-11eb-8374-024215000701"} Time: 29.16020894050598
可以看到，函数计算顺利返回了预期结果，但是整体耗时却超乎想象，有近30s。
8.1.4 项目优化
在上一部分的测试中，代码的首次执行耗时近30s，性能非常低。再次执行测试代码，可以看到如下结果：
{"Response": {"laptop": "71.43893837928772", "notebook": "16.265614330768585", "modem": "4.899385944008827", "hard_disc": "4.007565602660179", "mouse": "1.2981869280338287"}, "ResponseId": "4b8be48a-298a-11eb-ba97-024215000501"} Time: 1.1511380672454834
第二次执行，仅用了1.15秒，相对于第一次执行，性能提升了近30倍。那么为什么前后两次执行会产生如此大的性能差距呢？
通过本地对代码进行测试：
# -*- coding: utf-8 -*- import time timeStart = time.time() #模型加载 from imageai.Prediction import ImagePrediction prediction = ImagePrediction() prediction.setModelTypeAsResNet() prediction.setModelPath("resnet50_weights_tf_dim_ordering_tf_kernels.h5") prediction.loadModel() print("Load Time: ", time.time() - timeStart) timeStart = time.time() predictions, probabilities = prediction.predictImage("./picture.jpg", result_count=5) for eachPrediction, eachProbability in zip(predictions, probabilities): print(str(eachPrediction) + " : " + str(eachProbability)) print("Predict Time: ", time.time() - timeStart)
可以看到其执行结果为：
Load Time: 5.549695014953613 laptop : 71.43893241882324 notebook : 16.265612840652466 modem : 4.899394512176514 hard_disc : 4.007557779550552 mouse : 1.2981942854821682 Predict Time: 0.8137111663818359
在整个项目预测的过程中，加载imageAI模块以及模型文件时，一共耗时5.5秒，在预测部分仅用了不到1秒的时间。也就是说，在通过ImageAI进行图像目标提取/预测的过程中，实际上与传统的AI项目一样，包括模型加载和预测两个环节，而模型加载会消耗更多的时间。
在函数计算中，单实例性能本身可能没有本地计算机的性能高。因此在部署的代码中，我们可以看到，模型装载过程实际上是被放在了入口方法之外，这样做的一个好处是，项目每次执行时不一定会有冷启动，也就是说，在某些前提下是可以复用一些对象的，即无须每次都重新加载模型、导入依赖等。这样做可以缓解每次装载模型响应时间过长的问题。
所以在实际项目中，为了避免频繁请求，在实例重复装载、创建某些资源时，可以将部分资源放在初始化的时候进行。这样可以大幅度提高项目的整体性能，同时可以配合厂商所提供的预留能力，基本上杜绝函数冷启动带来的负面影响。
8.1.5 总结
近年来，人工智能与云计算的发展突飞猛进，在Serverless架构中，如何运行传统的人工智能项目已经逐渐成为很多人所需要了解的事情。本节通过一个已有的依赖库（ImageAI）实现一个图像分类和预测的接口。通过这个例子，可以得出如下结论：
- Serverless架构可以运行人工智能相关项目；
- Serverless可以很好地兼容Tensorflow等机器学习/深度学习的工具；
- 虽然函数计算本身有空间限制，但是在增加了硬盘挂载能力之后，函数计算本身的能力将会得到大幅度提升。
当然，本节也算是抛砖引玉，希望读者可以发挥自己的想象，将更多的AI项目与Serverless架构进行结合并运用到实际应用中。
标签：Serverless,Python,ImageAI,下用,prediction,time,print,import From： https://www.cnblogs.com/muzinan110/p/17067649.html