在本系列文章中，我們將引導(dǎo)您完成將CI/CD應(yīng)用于AI任務(wù)的過程。最后，您將獲得一個滿足GoogleMLOps成熟度模型第2級要求的功能管道。我們假設(shè)您對Python，深度學(xué)習(xí)，Docker，DevOps和Flask有所了解。
在上一篇文章中，我們?yōu)榇隧?xiàng)目設(shè)置了一個云環(huán)境。在這一部分中，我們將引導(dǎo)您完成持續(xù)集成，模型自動訓(xùn)練，自動調(diào)整和持續(xù)交付所需的代碼。下圖顯示了我們在項(xiàng)目流程中的位置。
我們將顯示代碼的精簡版本。有關(guān)完整版本，請參見此存儲庫。我們將在該項(xiàng)目中使用GCRDocker映像（由TensorFlow支持）–但是隨時可以使用其他映像。
首先，我們將討論在本地運(yùn)行這些解決方案的代碼。稍后，我們將看到如何為云部署做好準(zhǔn)備。
下圖顯示了我們項(xiàng)目的文件結(jié)構(gòu)。
該data_utils.py文件句柄的數(shù)據(jù)加載，轉(zhuǎn)換和模型保存到GCS。此文件可能因項(xiàng)目而異。本質(zhì)上，它在模型訓(xùn)練之前執(zhí)行所有數(shù)據(jù)處理任務(wù)。讓我們看一下代碼：
import datetime
from google.cloud import storage
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
import tensorflow as tf
import gc
from sklearn import preprocessing
import os
import zipfile
import cv2
import sys
def dataset_transformation(path):
images = []
for dirname, _, filenames in os.walk(path):
for filename in filenames:
if filename.endswith('.png'):
image = cv2.imread(os.path.join(dirname, filename))
image = cv2.resize(image, (128, 128))
images.append(image)
return images
def load_data(args):
file_1 = '/root/AutomaticTraining-Dataset/COVID_RX/normal_images.zip'
file_2 = '/root/AutomaticTraining-Dataset/COVID_RX/covid_images.zip'
file_3 = '/root/AutomaticTraining-Dataset/COVID_RX/viral_images.zip'
extract_to = '/root/AutomaticTraining-Dataset/COVID_RX/'
with zipfile.ZipFile(file_1, 'r') as zip_ref:
zip_ref.extractall(extract_to)
with zipfile.ZipFile(file_2, 'r') as zip_ref:
zip_ref.extractall(extract_to)
with zipfile.ZipFile(file_3, 'r') as zip_ref:
zip_ref.extractall(extract_to)
normal = dataset_transformation('/root/AutomaticTraining-Dataset/COVID_RX/normal_images')
covid = dataset_transformation('/root/AutomaticTraining-Dataset/COVID_RX/covid_images')
viral = dataset_transformation('/root/AutomaticTraining-Dataset/COVID_RX/viral_images')
#Train and test - dataset combination
X = normal + viral + covid
#Transforming from list to numpy array.
X = np.array(X)
#Creating labels.
y = []
for i in range(len(normal)):
y.append(0)
for i in range(len(covid)):
y.append(1)
for i in range(len(viral)):
y.append(2)
y = np.array(y)
#Dataset splitting
X_train, X_test, y_train, y_test = train_test_split(X, y, random_state=0, shuffle = True)
return X_train, X_test, y_train, y_test
def save_model(bucket_name, best_model):
try:
storage_client = storage.Client() #if running on GCP
bucket = storage_client.bucket(bucket_name)
blob1 = bucket.blob('{}/{}'.format('testing',best_model))
blob1.upload_from_filename(best_model)
return True,None
except Exception as e:
return False,e123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869復(fù)制代碼類型：[html]該model_assembly.py文件包含模型的創(chuàng)建和自動的調(diào)整的代碼。我們希望從一個非常基本的模型開始–對其進(jìn)行訓(xùn)練和評估。如果初始模型不能達(dá)到理想的性能，我們將進(jìn)行改進(jìn)直到達(dá)到目標(biāo)。讓我們看一下代碼：
from tensorflow.keras.models import load_model from tensorflow.keras import layers import tensorflow as tf import numpy as np def get_base_model(): input_img = layers.Input(shape=(128, 128, 3)) x = layers.Conv2D(64,(3, 3), activation='relu')(input_img) return input_img,x def get_additional_layer(filters,x): x = layers.MaxPooling2D((2, 2))(x) x = layers.Conv2D(filters, (3, 3), activation='relu')(x) return x def get_final_layers(neurons,x): x = layers.SpatialDropout2D(0.2)(x) x = layers.Flatten()(x) x = layers.Dense(neurons)(x) x = layers.Dense(3)(x) return x123456789101112131415161718192021復(fù)制代碼類型：[html]
這些函數(shù)將在循環(huán)中調(diào)用，并且在第一次迭代中，我們將獲得base_model，final_layers并將它們堆疊起來以構(gòu)建一個非常簡單的模型。如果訓(xùn)練我們發(fā)現(xiàn)該模型不執(zhí)行不夠好之后，然后我們將再次得到base_model，加additional_layers,堆棧final_layers，然后訓(xùn)練和一次評估。如果我們?nèi)匀粺o法達(dá)到良好的性能，則將在循環(huán)中重復(fù)最后一個過程，并增加更多的過程，additional_layers直到達(dá)到預(yù)定義的良好指標(biāo)為止。
該email_notifications.py文件是負(fù)責(zé)通過本地SMTP服務(wù)器產(chǎn)品所有者提供的電子郵件。這些電子郵件會告訴所有者是否一切正常，如果不正常，那是什么問題。
import smtplib
import os
# Email variables definition
sender = 'example@gmail.com’
receiver = ['svirahonda@gmail.com'] #replace this by the owner's email address
smtp_provider = 'smtp.gmail.com' #replace this by your STMP provider
smtp_port = 587
smtp_account = 'example@gmail.com’
smtp_password = 'your_password’
def training_result(result,model_acc):
if result == 'ok':
message = 'The model reached '+str(model_acc)+', It has been saved to GCS.'
if result == 'failed':
message = 'None of the models reached an acceptable accuracy, training execution had to be forcefully ended.’
message = 'Subject: {}\n\n{}'.format('An automatic training job has ended recently', message)
try:
server = smtplib.SMTP(smtp_provider,smtp_port)
server.starttls()
server.login(smtp_account,smtp_password)
server.sendmail(sender, receiver, message)
return
except Exception as e:
print('Something went wrong. Unable to send email: '+str(e),flush=True)
return
def exception(e_message):
try:
message = 'Subject: {}\n\n{}'.format('An automatic training job has failed.', e_message)
server = smtplib.SMTP(smtp_provider,smtp_port)
server.starttls()
server.login(smtp_account,smtp_password)
server.sendmail(sender, receiver, message)
return
except Exception as e:
print('Something went wrong. Unable to send email: '+str(e),flush=True)
return1234567891011121314151617181920212223242526272829303132333435363738復(fù)制代碼類型：[html]該task.py文件編排節(jié)目。它會初始化GPU（如果有），以開始模型訓(xùn)練，并在需要時調(diào)整模型。它還接收傳遞給應(yīng)用程序的參數(shù)。這是代碼：
import tensorflow as tf
from tensorflow.keras import Model, layers, optimizers
from tensorflow.keras.callbacks import ModelCheckpoint
from tensorflow.keras import Model
from tensorflow.keras.models import load_model
import argparse
import model_assembly, data_utils, email_notifications
import sys
import os
import gc
from google.cloud import storage
import datetime
import math
# general variables declaration
model_name = 'best_model.hdf5'
def initialize_gpu():
if len(tf.config.experimental.list_physical_devices('GPU')) > 0:
tf.config.set_soft_device_placement(True)
tf.debugging.set_log_device_placement(True)
return
def start_training(args):
# Loading splitted data
X_train, X_test, y_train, y_test = data_utils.load_data(args)
# Initializing GPU if available
initialize_gpu()
train_model(X_train, X_test, y_train, y_test, args)
def train_model(X_train, X_test, y_train, y_test,args):
try:
model_loss, model_acc = [0,0]
counter = 0
while model_acc <= 0.85:
input_img,x = model_assembly.get_base_model()
if counter == 0:
x = model_assembly.get_final_layers(64,x)
else:
for i in range(counter):
x = model_assembly.get_additional_layer(int(64*(math.pow(2,counter))),x)
x = model_assembly.get_final_layers(int(64*(math.pow(2,counter))),x)
cnn = Model(input_img, x,name="CNN_COVID_"+str(counter))
cnn.summary()
cnn.compile(optimizer='adam', loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True), metrics=['accuracy'])
checkpoint = ModelCheckpoint(model_name, monitor='val_loss', verbose=1,save_best_only=True, mode='auto', save_freq="epoch")
cnn.fit(X_train, y_train, epochs=args.epochs, validation_data=(X_test, y_test),callbacks=[checkpoint])
cnn = load_model(model_name)
model_loss, model_acc = cnn.evaluate(X_test, y_test,verbose=2)
if model_acc > 0.85:
saved_ok = data_utils.save_model(args.bucket_name,model_name)
if saved_ok[0] == True:
email_notifications.training_result('ok',model_acc)
sys.exit(0)
else:
email_notifications.exception(saved_ok[1])
sys.exit(1)
else:
pass
if counter >= 5:
email_notifications.training_result('failed',None)
sys.exit(1)
counter += 1
except Exception as e:
email_notifications.exception('An exception when training the model has occurred: '+str(e))
sys.exit(1)
def get_args():
parser = argparse.ArgumentParser()
parser.add_argument('--bucket-name', type=str, default = 'automatictrainingcicd-aiplatform', help = 'GCP bucket name')
parser.add_argument('--epochs', type=int, default=3, help='Epochs number')
args = parser.parse_args()
return args
def main():
args = get_args()
start_training(args)
if __name__ == '__main__':
main()1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980復(fù)制代碼類型：[html]我們的Dockerfile負(fù)責(zé)將指令傳遞給Docker守護(hù)程序以構(gòu)建適當(dāng)?shù)娜萜??？雌饋硎沁@樣的：
FROM gcr.io/deeplearning-platform-release/tf2-cpu.2-0 WORKDIR /root RUN pip install pandas numpy google-cloud-storage scikit-learn opencv-python RUN apt-get update; apt-get install git -y; apt-get install -y libgl1-mesa-dev ADD "https://www.random.org/cgi-bin/randbyte?nbytes=10&format=h" skipcache RUN git clone https://github.com/sergiovirahonda/AutomaticTraining-Dataset.git ADD "https://www.random.org/cgi-bin/randbyte?nbytes=10&format=h" skipcache RUN git clone https://github.com/sergiovirahonda/AutomaticTraining-CodeCommit.git RUN mv /root/AutomaticTraining-CodeCommit/model_assembly.py /root RUN mv /root/AutomaticTraining-CodeCommit/task.py /root RUN mv /root/AutomaticTraining-CodeCommit/data_utils.py /root RUN mv /root/AutomaticTraining-CodeCommit/email_notifications.py /root ENTRYPOINT ["python","task.py"]123456789101112131415161718復(fù)制代碼類型：[html]
上面的文件使用該gcr.io/deeplearning-platform-release/tf2-cpu.2-0映像，安裝依賴項(xiàng)，克隆所需的存儲庫，將文件移至主目錄，并設(shè)置容器執(zhí)行的入口點(diǎn)。
