معرفی خطوط لوله Vertex

۱. مرور کلی

در این آزمایش، شما نحوه ایجاد و اجرای خطوط لوله یادگیری ماشین با Vertex Pipelines را خواهید آموخت.

آنچه یاد می‌گیرید

شما یاد خواهید گرفت که چگونه:

  • استفاده از Kubeflow Pipelines SDK برای ساخت خطوط لوله یادگیری ماشینی مقیاس‌پذیر
  • یک خط لوله مقدمه سه مرحله‌ای ایجاد و اجرا کنید که ورودی متن را دریافت می‌کند
  • ایجاد و اجرای یک خط لوله که یک مدل طبقه‌بندی AutoML را آموزش، ارزیابی و مستقر می‌کند
  • استفاده از کامپوننت‌های از پیش ساخته شده برای تعامل با سرویس‌های هوش مصنوعی Vertex، که از طریق کتابخانه google_cloud_pipeline_components ارائه می‌شوند.
  • زمان‌بندی یک کار پایپ‌لاین با استفاده از Cloud Scheduler

هزینه کل اجرای این آزمایشگاه در گوگل کلود حدود ۲۵ دلار است.

۲. مقدمه‌ای بر هوش مصنوعی ورتکس

این آزمایشگاه از جدیدترین محصول هوش مصنوعی موجود در Google Cloud استفاده می‌کند. Vertex AI، محصولات یادگیری ماشین را در سراسر Google Cloud در یک تجربه توسعه یکپارچه ادغام می‌کند. پیش از این، مدل‌های آموزش‌دیده با AutoML و مدل‌های سفارشی از طریق سرویس‌های جداگانه قابل دسترسی بودند. این محصول جدید، هر دو را در یک API واحد، به همراه سایر محصولات جدید، ترکیب می‌کند. همچنین می‌توانید پروژه‌های موجود را به Vertex AI منتقل کنید.

علاوه بر خدمات آموزش و استقرار مدل، Vertex AI شامل انواع محصولات MLOps از جمله Vertex Pipelines (تمرکز این آزمایشگاه)، Model Monitoring، Feature Store و موارد دیگر نیز می‌شود. می‌توانید تمام محصولات Vertex AI را در نمودار زیر مشاهده کنید.

بررسی اجمالی محصولات ورتکس

اگر نظری دارید، لطفاً به صفحه پشتیبانی مراجعه کنید.

چرا خطوط لوله ML مفید هستند؟

قبل از اینکه وارد بحث شویم، ابتدا بیایید بفهمیم که چرا باید از یک خط لوله استفاده کنید. تصور کنید که در حال ساخت یک گردش کار یادگیری ماشین هستید که شامل پردازش داده‌ها، آموزش یک مدل، تنظیم هایپرپارامتر، ارزیابی و استقرار مدل است. هر یک از این مراحل ممکن است وابستگی‌های متفاوتی داشته باشند که اگر کل گردش کار را به عنوان یک سیستم یکپارچه در نظر بگیرید، ممکن است دشوار شود. همانطور که شروع به مقیاس‌بندی فرآیند یادگیری ماشین خود می‌کنید، ممکن است بخواهید گردش کار یادگیری ماشین خود را با سایر اعضای تیم خود به اشتراک بگذارید تا آنها بتوانند آن را اجرا کرده و کد ارائه دهند. بدون یک فرآیند قابل اعتماد و قابل تکرار، این کار می‌تواند دشوار شود. با خطوط لوله، هر مرحله در فرآیند یادگیری ماشین شما، کانتینر مخصوص به خود را دارد. این به شما امکان می‌دهد مراحل را به طور مستقل توسعه دهید و ورودی و خروجی هر مرحله را به روشی قابل تکرار پیگیری کنید. همچنین می‌توانید اجرای خط لوله خود را بر اساس سایر رویدادهای محیط ابری خود، مانند شروع اجرای خط لوله در صورت وجود داده‌های آموزشی جدید، برنامه‌ریزی یا فعال کنید.

خطوط لوله tl;dr : به شما کمک می‌کنند تا گردش کار یادگیری ماشین خود را خودکار و بازتولید کنید .

۳. راه‌اندازی محیط ابری

برای اجرای این codelab به یک پروژه Google Cloud Platform با قابلیت پرداخت صورتحساب نیاز دارید. برای ایجاد یک پروژه، دستورالعمل‌های اینجا را دنبال کنید.

مرحله ۱: شروع Cloud Shell

در این آزمایش، شما در یک جلسه Cloud Shell کار خواهید کرد، که یک مفسر فرمان است که توسط یک ماشین مجازی که در فضای ابری گوگل اجرا می‌شود، میزبانی می‌شود. شما می‌توانید به راحتی این بخش را به صورت محلی روی رایانه خود اجرا کنید، اما استفاده از Cloud Shell به همه امکان دسترسی به یک تجربه قابل تکرار در یک محیط سازگار را می‌دهد. پس از آزمایش، می‌توانید این بخش را دوباره روی رایانه خود امتحان کنید.

مجوزدهی پوسته ابری

فعال کردن پوسته ابری

از بالا سمت راست کنسول ابری، روی دکمه زیر کلیک کنید تا Cloud Shell فعال شود :

فعال کردن پوسته ابری

اگر قبلاً Cloud Shell را شروع نکرده‌اید، یک صفحه میانی (در پایین صفحه) به شما نمایش داده می‌شود که توضیح می‌دهد چیست. در این صورت، روی ادامه کلیک کنید (و دیگر هرگز آن را نخواهید دید). آن صفحه یکبار مصرف به این شکل است:

راه اندازی پوسته ابری

آماده‌سازی و اتصال به Cloud Shell فقط چند لحظه طول می‌کشد.

آغازگر Cloud Shell

این ماشین مجازی با تمام ابزارهای توسعه مورد نیاز شما پر شده است. این ماشین یک دایرکتوری خانگی ۵ گیگابایتی دائمی ارائه می‌دهد و در فضای ابری گوگل اجرا می‌شود که عملکرد شبکه و احراز هویت را تا حد زیادی بهبود می‌بخشد. بخش عمده‌ای از کار شما در این آزمایشگاه کد، اگر نگوییم همه، را می‌توان به سادگی با یک مرورگر یا کروم‌بوک انجام داد.

پس از اتصال به Cloud Shell، باید ببینید که از قبل احراز هویت شده‌اید و پروژه از قبل روی شناسه پروژه شما تنظیم شده است.

برای تأیید احراز هویت، دستور زیر را در Cloud Shell اجرا کنید:

gcloud auth list

شما باید چیزی شبیه به این را در خروجی دستور ببینید:

خروجی Cloud Shell

دستور زیر را در Cloud Shell اجرا کنید تا تأیید کنید که دستور gcloud از پروژه شما اطلاع دارد:

gcloud config list project

خروجی دستور

[core]
project = <PROJECT_ID>

اگر اینطور نیست، می‌توانید با این دستور آن را تنظیم کنید:

gcloud config set project <PROJECT_ID>

خروجی دستور

Updated property [core/project].

Cloud Shell چند متغیر محیطی دارد، از جمله GOOGLE_CLOUD_PROJECT که شامل نام پروژه ابری فعلی ما است. ما از این در جاهای مختلف این تمرین استفاده خواهیم کرد. می‌توانید با اجرای دستور زیر آن را مشاهده کنید:

echo $GOOGLE_CLOUD_PROJECT

مرحله ۲: فعال کردن APIها

در مراحل بعدی، خواهید دید که این سرویس‌ها کجا مورد نیاز هستند (و چرا)، اما فعلاً این دستور را اجرا کنید تا به پروژه خود دسترسی به سرویس‌های Compute Engine، Container Registry و Vertex AI بدهید:

gcloud services enable compute.googleapis.com         \
                       containerregistry.googleapis.com  \
                       aiplatform.googleapis.com  \
                       cloudbuild.googleapis.com \
                       cloudfunctions.googleapis.com

این باید یک پیام موفقیت‌آمیز مشابه این ایجاد کند:

Operation "operations/acf.cc11852d-40af-47ad-9d59-477a12847c9e" finished successfully.

مرحله ۳: ایجاد یک فضای ذخیره‌سازی ابری

برای اجرای یک کار آموزشی در Vertex AI، به یک سطل ذخیره‌سازی برای ذخیره دارایی‌های مدل ذخیره شده خود نیاز داریم. سطل باید منطقه‌ای باشد. ما در اینجا us-central استفاده می‌کنیم، اما شما می‌توانید از یک منطقه دیگر استفاده کنید (فقط آن را در طول این آزمایش جایگزین کنید). اگر از قبل سطلی دارید، می‌توانید از این مرحله صرف نظر کنید.

برای ایجاد یک باکت، دستورات زیر را در ترمینال Cloud Shell خود اجرا کنید:

BUCKET_NAME=gs://$GOOGLE_CLOUD_PROJECT-bucket
gsutil mb -l us-central1 $BUCKET_NAME

در مرحله بعد، به حساب کاربری سرویس محاسباتی خود دسترسی به این سطل را می‌دهیم. این کار تضمین می‌کند که Vertex Pipelines مجوزهای لازم برای نوشتن فایل‌ها در این سطل را دارد. برای افزودن این مجوز، دستور زیر را اجرا کنید: 

gcloud projects describe $GOOGLE_CLOUD_PROJECT > project-info.txt
PROJECT_NUM=$(cat project-info.txt | sed -nre 's:.*projectNumber\: (.*):\1:p')
SVC_ACCOUNT="${PROJECT_NUM//\'/}-compute@developer.gserviceaccount.com"
gcloud projects add-iam-policy-binding $GOOGLE_CLOUD_PROJECT --member serviceAccount:$SVC_ACCOUNT --role roles/storage.objectAdmin

مرحله ۴: ایجاد یک نمونه از Vertex AI Workbench

از بخش Vertex AI در کنسول ابری خود، روی Workbench کلیک کنید:

منوی هوش مصنوعی ورتکس

از آنجا، در دفترچه‌های مدیریت‌شده توسط کاربر ، روی دفترچه یادداشت جدید کلیک کنید:

ایجاد دفترچه یادداشت جدید

سپس نوع نمونه TensorFlow Enterprise 2.3 (with LTS) را بدون GPU انتخاب کنید:

نمونه TFE

از گزینه‌های پیش‌فرض استفاده کنید و سپس روی «ایجاد» کلیک کنید.

مرحله ۵: دفترچه یادداشت خود را باز کنید

پس از ایجاد نمونه، گزینه Open JupyterLab را انتخاب کنید:

دفترچه یادداشت را باز کنید

۴. تنظیمات Vertex Pipelines

برای استفاده از Vertex Pipelines باید چند کتابخانه اضافی نصب کنیم:

  • Kubeflow Pipelines : این SDK است که ما برای ساخت pipeline خود استفاده خواهیم کرد. Vertex Pipelines از اجرای pipelineهای ساخته شده با Kubeflow Pipelines یا TFX پشتیبانی می‌کند.
  • اجزای خط تولید گوگل کلود : این کتابخانه اجزای از پیش ساخته شده‌ای را ارائه می‌دهد که تعامل با سرویس‌های هوش مصنوعی ورتکس را از مراحل خط تولید شما آسان‌تر می‌کند.

مرحله ۱: ایجاد دفترچه یادداشت پایتون و نصب کتابخانه‌ها

ابتدا، از منوی Launcher در نمونه Notebook خود، با انتخاب Python 3 یک Notebook ایجاد کنید:

ایجاد دفترچه یادداشت پایتون ۳

شما می‌توانید با کلیک روی علامت + در سمت چپ بالای نمونه نوت‌بوک خود، به منوی لانچر دسترسی پیدا کنید.

برای نصب هر دو سرویسی که در این آزمایش استفاده خواهیم کرد، ابتدا پرچم کاربر را در یک سلول نوت‌بوک تنظیم کنید: 

USER_FLAG = "--user"

سپس دستور زیر را از نوت بوک خود اجرا کنید: 

!pip3 install {USER_FLAG} google-cloud-aiplatform==1.7.0 --upgrade
!pip3 install {USER_FLAG} kfp==1.8.9 google-cloud-pipeline-components==0.2.0

پس از نصب این بسته‌ها، باید هسته را مجدداً راه‌اندازی کنید: 

import os

if not os.getenv("IS_TESTING"):
    # Automatically restart kernel after installs
    import IPython

    app = IPython.Application.instance()
    app.kernel.do_shutdown(True)

در نهایت، بررسی کنید که بسته‌ها را به درستی نصب کرده‌اید. نسخه KFP SDK باید >=1.8 باشد: 

!python3 -c "import kfp; print('KFP SDK version: {}'.format(kfp.__version__))"
!python3 -c "import google_cloud_pipeline_components; print('google_cloud_pipeline_components version: {}'.format(google_cloud_pipeline_components.__version__))"

مرحله ۲: شناسه پروژه و باکت خود را تنظیم کنید

در طول این آزمایش، شما به شناسه پروژه ابری خود و باکتی که قبلاً ایجاد کرده‌اید، ارجاع خواهید داد. در مرحله بعد، برای هر یک از آنها متغیرهایی ایجاد خواهیم کرد.

اگر شناسه پروژه خود را نمی‌دانید، می‌توانید با اجرای دستور زیر آن را بدست آورید: 

import os
PROJECT_ID = ""

# Get your Google Cloud project ID from gcloud
if not os.getenv("IS_TESTING"):
    shell_output=!gcloud config list --format 'value(core.project)' 2>/dev/null
    PROJECT_ID = shell_output[0]
    print("Project ID: ", PROJECT_ID)

در غیر این صورت، آن را اینجا تنظیم کنید: 

if PROJECT_ID == "" or PROJECT_ID is None:
    PROJECT_ID = "your-project-id"  # @param {type:"string"}

سپس یک متغیر برای ذخیره نام سطل خود ایجاد کنید. اگر آن را در این آزمایش ایجاد کرده‌اید، موارد زیر کار خواهد کرد. در غیر این صورت، باید آن را به صورت دستی تنظیم کنید: 

BUCKET_NAME="gs://" + PROJECT_ID + "-bucket"

مرحله ۳: وارد کردن کتابخانه‌ها

برای وارد کردن کتابخانه‌هایی که در طول این آزمایشگاه کد استفاده خواهیم کرد، موارد زیر را اضافه کنید: 

import kfp

from kfp.v2 import compiler, dsl
from kfp.v2.dsl import component, pipeline, Artifact, ClassificationMetrics, Input, Output, Model, Metrics

from google.cloud import aiplatform
from google_cloud_pipeline_components import aiplatform as gcc_aip
from typing import NamedTuple

مرحله ۴: تعریف ثابت‌ها

آخرین کاری که قبل از ساخت pipeline باید انجام دهیم، تعریف چند متغیر ثابت است. PIPELINE_ROOT مسیر ذخیره‌سازی ابری است که مصنوعات ایجاد شده توسط pipeline ما در آن نوشته می‌شوند. ما در اینجا us-central1 به عنوان منطقه استفاده می‌کنیم، اما اگر هنگام ایجاد bucket خود از منطقه دیگری استفاده کرده‌اید، متغیر REGION را در کد زیر به‌روزرسانی کنید: 

PATH=%env PATH
%env PATH={PATH}:/home/jupyter/.local/bin
REGION="us-central1"

PIPELINE_ROOT = f"{BUCKET_NAME}/pipeline_root/"
PIPELINE_ROOT

پس از اجرای کد بالا، باید دایرکتوری ریشه خط لوله خود را مشاهده کنید. این مکان ذخیره‌سازی ابری است که مصنوعات خط لوله شما در آن نوشته خواهد شد. این فرمت به صورت gs://YOUR-BUCKET-NAME/pipeline_root/ خواهد بود.

۵. ایجاد اولین خط تولید (Pipeline)

برای آشنایی با نحوه کار Vertex Pipelines، ابتدا یک Pipeline کوتاه با استفاده از KFP SDK ایجاد خواهیم کرد. این Pipeline هیچ کاری مربوط به ML انجام نمی‌دهد (نگران نباشید، به آن هم خواهیم رسید!)، ما از آن برای آموزش شما استفاده می‌کنیم:

  • نحوه ایجاد کامپوننت‌های سفارشی در KFP SDK
  • نحوه اجرا و نظارت بر یک خط لوله در Vertex Pipelines

ما یک خط لوله ایجاد خواهیم کرد که یک جمله را با استفاده از دو خروجی چاپ می‌کند: نام محصول و توضیحات ایموجی. این خط لوله از سه جزء تشکیل شده است:

  • product_name : این کامپوننت نام محصول (یا هر اسمی که واقعاً می‌خواهید) را به عنوان ورودی می‌گیرد و آن رشته را به عنوان خروجی برمی‌گرداند.
  • emoji : این کامپوننت توضیحات متنی یک ایموجی را دریافت کرده و آن را به یک ایموجی تبدیل می‌کند. برای مثال، کد متنی برای ✨، "sparkles" است. این کامپوننت از یک کتابخانه ایموجی استفاده می‌کند تا نحوه مدیریت وابستگی‌های خارجی در خط تولید شما را نشان دهد.
  • build_sentence : این کامپوننت نهایی، خروجی دو کامپوننت قبلی را برای ساخت جمله‌ای که از ایموجی استفاده می‌کند، مصرف می‌کند. برای مثال، خروجی حاصل ممکن است «Vertex Pipelines is ✨» باشد.

بیایید شروع به کدنویسی کنیم!

مرحله ۱: ایجاد یک کامپوننت مبتنی بر تابع پایتون

با استفاده از KFP SDK، می‌توانیم کامپوننت‌هایی بر اساس توابع پایتون ایجاد کنیم. ما از آن برای ۳ کامپوننت در اولین pipeline خود استفاده خواهیم کرد. ابتدا کامپوننت product_name را می‌سازیم که به سادگی یک رشته را به عنوان ورودی می‌گیرد و آن رشته را برمی‌گرداند. موارد زیر را به دفترچه یادداشت خود اضافه کنید: 

@component(base_image="python:3.9", output_component_file="first-component.yaml")
def product_name(text: str) -> str:
    return text

بیایید نگاهی دقیق‌تر به سینتکس اینجا بیندازیم:

  • دکوراتور @component این تابع را هنگام اجرای pipeline به یک کامپوننت کامپایل می‌کند. شما می‌توانید هر زمان که یک کامپوننت سفارشی می‌نویسید از این استفاده کنید.
  • پارامتر base_image تصویر کانتینری را که این کامپوننت استفاده خواهد کرد، مشخص می‌کند.
  • پارامتر output_component_file اختیاری است و فایل yaml را برای نوشتن کامپوننت کامپایل شده مشخص می‌کند. پس از اجرای سلول، باید آن فایل را که در نمونه نوت‌بوک شما نوشته شده است، مشاهده کنید. اگر می‌خواهید این کامپوننت را با کسی به اشتراک بگذارید، می‌توانید فایل yaml تولید شده را برای او ارسال کنید و از او بخواهید آن را با کد زیر بارگذاری کند: 
product_name_component = kfp.components.load_component_from_file('./first-component.yaml')
  • علامت -> str بعد از تعریف تابع، نوع خروجی این کامپوننت را مشخص می‌کند.

مرحله ۲: دو کامپوننت دیگر ایجاد کنید

برای تکمیل خط لوله خود، دو کامپوننت دیگر ایجاد خواهیم کرد. اولین کامپوننتی که تعریف خواهیم کرد، یک رشته را به عنوان ورودی دریافت می‌کند و در صورت وجود، این رشته را به ایموجی مربوطه تبدیل می‌کند. این کامپوننت یک تاپل با متن ورودی ارسالی و ایموجی حاصل برمی‌گرداند: 

@component(packages_to_install=["emoji"])
def emoji(
    text: str,
) -> NamedTuple(
    "Outputs",
    [
        ("emoji_text", str),  # Return parameters
        ("emoji", str),
    ],
):
    import emoji

    emoji_text = text
    emoji_str = emoji.emojize(':' + emoji_text + ':', language='alias')
    print("output one: {}; output_two: {}".format(emoji_text, emoji_str))
    return (emoji_text, emoji_str)

این کامپوننت کمی پیچیده‌تر از کامپوننت قبلی است. بیایید ویژگی‌های جدید را بررسی کنیم:

  • پارامتر packages_to_install هرگونه وابستگی کتابخانه‌ای خارجی برای این کانتینر را به کامپوننت اعلام می‌کند. در این مورد، ما از کتابخانه‌ای به نام emoji استفاده می‌کنیم.
  • این کامپوننت یک NamedTuple به نام Outputs برمی‌گرداند. توجه داشته باشید که هر یک از رشته‌های موجود در این tuple دارای کلیدهایی به نام‌های emoji_text و emoji هستند. ما از این کلیدها در کامپوننت بعدی خود برای دسترسی به خروجی استفاده خواهیم کرد.

کامپوننت نهایی در این خط لوله، خروجی دو کامپوننت اول را مصرف کرده و آنها را برای برگرداندن یک رشته ترکیب می‌کند: 

@component
def build_sentence(
    product: str,
    emoji: str,
    emojitext: str
) -> str:
    print("We completed the pipeline, hooray!")
    end_str = product + " is "
    if len(emoji) > 0:
        end_str += emoji
    else:
        end_str += emojitext
    return(end_str)

شاید از خود بپرسید: این کامپوننت چطور می‌داند که از خروجی مراحل قبلی که تعریف کردید استفاده کند؟ سوال خوبی است! همه اینها را در مرحله بعدی به هم ربط خواهیم داد.

مرحله ۳: قرار دادن اجزا در یک خط لوله

تعاریف کامپوننت‌هایی که در بالا تعریف کردیم، توابع کارخانه‌ای ایجاد کردند که می‌توانند در تعریف خط لوله برای ایجاد مراحل استفاده شوند. برای راه‌اندازی یک خط لوله، از دکوراتور @pipeline استفاده کنید، به خط لوله یک نام و توضیح بدهید و مسیر ریشه‌ای را که مصنوعات خط لوله شما باید در آن نوشته شوند، ارائه دهید. منظور از مصنوعات، هر فایل خروجی تولید شده توسط خط لوله شماست. این خط لوله مقدماتی هیچ فایلی تولید نمی‌کند، اما خط لوله بعدی ما این کار را خواهد کرد.

در بلوک بعدی کد، یک تابع intro_pipeline تعریف می‌کنیم. در اینجا ورودی‌های مراحل اولیه خط لوله و نحوه اتصال مراحل به یکدیگر را مشخص می‌کنیم:

  • product_task نام محصول را به عنوان ورودی دریافت می‌کند. در اینجا ما "Vertex Pipelines" را ارسال می‌کنیم، اما شما می‌توانید این مقدار را به هر مقداری که می‌خواهید تغییر دهید.
  • emoji_task کد متنی یک ایموجی را به عنوان ورودی می‌گیرد. همچنین می‌توانید این کد را به هر چیزی که می‌خواهید تغییر دهید. برای مثال، "party_face" به ایموجی 🥳 اشاره دارد. توجه داشته باشید که از آنجایی که هم this و هم product_task هیچ مرحله‌ای برای دریافت ورودی ندارند، هنگام تعریف pipeline خود، ورودی این موارد را به صورت دستی مشخص می‌کنیم.
  • آخرین مرحله در خط تولید ما - consumer_task سه پارامتر ورودی دارد:
    • خروجی تابع product_task . از آنجایی که این مرحله فقط یک خروجی تولید می‌کند، می‌توانیم از طریق product_task.output به آن ارجاع دهیم.
    • خروجی emoji مرحله‌ی emoji_task ما. به کامپوننت emoji تعریف شده در بالا که در آن پارامترهای خروجی را نامگذاری کرده‌ایم، مراجعه کنید.
    • به طور مشابه، خروجی با نام emoji_text از کامپوننت emoji است. در صورتی که خط لوله ما متنی را ارسال کند که با یک ایموجی مطابقت نداشته باشد، از این متن برای ساخت یک جمله استفاده می‌کند. 
@pipeline(
    name="hello-world",
    description="An intro pipeline",
    pipeline_root=PIPELINE_ROOT,
)

# You can change the `text` and `emoji_str` parameters here to update the pipeline output
def intro_pipeline(text: str = "Vertex Pipelines", emoji_str: str = "sparkles"):
    product_task = product_name(text)
    emoji_task = emoji(emoji_str)
    consumer_task = build_sentence(
        product_task.output,
        emoji_task.outputs["emoji"],
        emoji_task.outputs["emoji_text"],
    )

مرحله ۴: کامپایل و اجرای پایپ‌لاین

با تعریف خط لوله، آماده کامپایل آن هستید. دستور زیر یک فایل JSON ایجاد می‌کند که برای اجرای خط لوله از آن استفاده خواهید کرد: 

compiler.Compiler().compile(
    pipeline_func=intro_pipeline, package_path="intro_pipeline_job.json"
)

سپس، یک متغیر TIMESTAMP ایجاد کنید. ما از این در شناسه شغل خود استفاده خواهیم کرد: 

from datetime import datetime

TIMESTAMP = datetime.now().strftime("%Y%m%d%H%M%S")

سپس وظیفه خط لوله خود را تعریف کنید: 

job = aiplatform.PipelineJob(
    display_name="hello-world-pipeline",
    template_path="intro_pipeline_job.json",
    job_id="hello-world-pipeline-{0}".format(TIMESTAMP),
    enable_caching=True
)

در نهایت، کار را اجرا کنید تا یک اجرای خط لوله جدید ایجاد شود: 

job.submit()

پس از اجرای این سلول، باید گزارش‌هایی را با پیوندی برای مشاهده‌ی اجرای خط لوله در کنسول خود مشاهده کنید:

گزارش‌های کار خط لوله

به آن لینک بروید. خط لوله شما پس از تکمیل باید به این شکل باشد:

خط لوله معرفی تکمیل شده

اجرای این خط لوله ۵-۶ دقیقه طول می‌کشد. پس از اتمام، می‌توانید روی کامپوننت build-sentence کلیک کنید تا خروجی نهایی را ببینید:

خروجی خط لوله مقدمه

حالا که با نحوه‌ی کار KFP SDK و Vertex Pipelines آشنا شدید، آماده‌اید تا یک Pipeline بسازید که با استفاده از سایر سرویس‌های Vertex AI یک مدل ML ایجاد و مستقر کند. بیایید شروع کنیم!

۶. ایجاد یک خط لوله یادگیری ماشینی سرتاسری

وقت آن است که اولین خط لوله یادگیری ماشینی خود را بسازید. در این خط لوله، ما از مجموعه داده‌های لوبیای خشک یادگیری ماشینی UCI، از: KOKLU, M. و OZKAN, IA، (2020)، "طبقه‌بندی چندکلاسه لوبیای خشک با استفاده از تکنیک‌های بینایی کامپیوتر و یادگیری ماشین". در Computers and Electronics in Agriculture، 174، 105507. DOI استفاده خواهیم کرد.

این یک مجموعه داده جدولی است و در خط تولید خود، از این مجموعه داده برای آموزش، ارزیابی و استقرار یک مدل AutoML استفاده خواهیم کرد که لوبیاها را بر اساس ویژگی‌هایشان به یکی از 7 نوع طبقه‌بندی می‌کند.

این خط لوله:

  • ایجاد یک مجموعه داده در
  • آموزش یک مدل طبقه‌بندی جدولی با AutoML
  • دریافت معیارهای ارزیابی در این مدل
  • بر اساس معیارهای ارزیابی، تصمیم بگیرید که آیا مدل را با استفاده از منطق شرطی در Vertex Pipelines مستقر کنید یا خیر.
  • با استفاده از پیش‌بینی رأس، مدل را در یک نقطه پایانی مستقر کنید

هر یک از مراحل ذکر شده یک جزء خواهد بود. اکثر مراحل خط لوله از اجزای از پیش ساخته شده برای خدمات هوش مصنوعی Vertex از طریق کتابخانه google_cloud_pipeline_components که قبلاً در این آزمایشگاه کد وارد کردیم، استفاده می‌کنند. در این بخش، ابتدا یک جزء سفارشی تعریف خواهیم کرد. سپس، بقیه مراحل خط لوله را با استفاده از اجزای از پیش ساخته شده تعریف خواهیم کرد. اجزای از پیش ساخته شده، دسترسی به خدمات هوش مصنوعی Vertex، مانند آموزش مدل و استقرار را آسان‌تر می‌کنند.

مرحله ۱: یک کامپوننت سفارشی برای ارزیابی مدل

کامپوننت سفارشی که تعریف خواهیم کرد، پس از اتمام آموزش مدل، در انتهای خط تولید ما استفاده خواهد شد. این کامپوننت چند کار انجام خواهد داد:

  • معیارهای ارزیابی را از مدل طبقه‌بندی AutoML آموزش‌دیده دریافت کنید
  • معیارها را تجزیه و تحلیل کرده و آنها را در رابط کاربری Vertex Pipelines رندر کنید
  • معیارها را با یک آستانه مقایسه کنید تا مشخص شود که آیا مدل باید مستقر شود یا خیر.

قبل از اینکه کامپوننت را تعریف کنیم، بیایید پارامترهای ورودی و خروجی آن را درک کنیم. به عنوان ورودی، این خط لوله برخی از فراداده‌های پروژه Cloud ما، مدل آموزش دیده حاصل (که بعداً این کامپوننت را تعریف خواهیم کرد)، معیارهای ارزیابی مدل و thresholds_dict_str را دریافت می‌کند. thresholds_dict_str چیزی است که هنگام اجرای خط لوله خود تعریف خواهیم کرد. در مورد این مدل طبقه‌بندی، این ناحیه زیر مقدار منحنی ROC خواهد بود که باید مدل را برای آن مستقر کنیم. به عنوان مثال، اگر مقدار 0.95 را وارد کنیم، به این معنی است که ما فقط می‌خواهیم خط لوله ما در صورتی مدل را مستقر کند که این معیار بالای 95٪ باشد.

مؤلفه ارزیابی ما رشته‌ای را برمی‌گرداند که نشان می‌دهد آیا مدل را مستقر کند یا خیر. برای ایجاد این مؤلفه سفارشی، موارد زیر را در یک سلول دفترچه یادداشت اضافه کنید: 

@component(
    base_image="gcr.io/deeplearning-platform-release/tf2-cpu.2-3:latest",
    output_component_file="tabular_eval_component.yaml",
    packages_to_install=["google-cloud-aiplatform"],
)
def classification_model_eval_metrics(
    project: str,
    location: str,  # "us-central1",
    api_endpoint: str,  # "us-central1-aiplatform.googleapis.com",
    thresholds_dict_str: str,
    model: Input[Artifact],
    metrics: Output[Metrics],
    metricsc: Output[ClassificationMetrics],
) -> NamedTuple("Outputs", [("dep_decision", str)]):  # Return parameter.

    import json
    import logging

    from google.cloud import aiplatform as aip

    # Fetch model eval info
    def get_eval_info(client, model_name):
        from google.protobuf.json_format import MessageToDict

        response = client.list_model_evaluations(parent=model_name)
        metrics_list = []
        metrics_string_list = []
        for evaluation in response:
            print("model_evaluation")
            print(" name:", evaluation.name)
            print(" metrics_schema_uri:", evaluation.metrics_schema_uri)
            metrics = MessageToDict(evaluation._pb.metrics)
            for metric in metrics.keys():
                logging.info("metric: %s, value: %s", metric, metrics[metric])
            metrics_str = json.dumps(metrics)
            metrics_list.append(metrics)
            metrics_string_list.append(metrics_str)

        return (
            evaluation.name,
            metrics_list,
            metrics_string_list,
        )

    # Use the given metrics threshold(s) to determine whether the model is
    # accurate enough to deploy.
    def classification_thresholds_check(metrics_dict, thresholds_dict):
        for k, v in thresholds_dict.items():
            logging.info("k {}, v {}".format(k, v))
            if k in ["auRoc", "auPrc"]:  # higher is better
                if metrics_dict[k] < v:  # if under threshold, don't deploy
                    logging.info("{} < {}; returning False".format(metrics_dict[k], v))
                    return False
        logging.info("threshold checks passed.")
        return True

    def log_metrics(metrics_list, metricsc):
        test_confusion_matrix = metrics_list[0]["confusionMatrix"]
        logging.info("rows: %s", test_confusion_matrix["rows"])

        # log the ROC curve
        fpr = []
        tpr = []
        thresholds = []
        for item in metrics_list[0]["confidenceMetrics"]:
            fpr.append(item.get("falsePositiveRate", 0.0))
            tpr.append(item.get("recall", 0.0))
            thresholds.append(item.get("confidenceThreshold", 0.0))
        print(f"fpr: {fpr}")
        print(f"tpr: {tpr}")
        print(f"thresholds: {thresholds}")
        metricsc.log_roc_curve(fpr, tpr, thresholds)

        # log the confusion matrix
        annotations = []
        for item in test_confusion_matrix["annotationSpecs"]:
            annotations.append(item["displayName"])
        logging.info("confusion matrix annotations: %s", annotations)
        metricsc.log_confusion_matrix(
            annotations,
            test_confusion_matrix["rows"],
        )

        # log textual metrics info as well
        for metric in metrics_list[0].keys():
            if metric != "confidenceMetrics":
                val_string = json.dumps(metrics_list[0][metric])
                metrics.log_metric(metric, val_string)
        # metrics.metadata["model_type"] = "AutoML Tabular classification"

    logging.getLogger().setLevel(logging.INFO)
    aip.init(project=project)
    # extract the model resource name from the input Model Artifact
    model_resource_path = model.metadata["resourceName"]
    logging.info("model path: %s", model_resource_path)

    client_options = {"api_endpoint": api_endpoint}
    # Initialize client that will be used to create and send requests.
    client = aip.gapic.ModelServiceClient(client_options=client_options)
    eval_name, metrics_list, metrics_str_list = get_eval_info(
        client, model_resource_path
    )
    logging.info("got evaluation name: %s", eval_name)
    logging.info("got metrics list: %s", metrics_list)
    log_metrics(metrics_list, metricsc)

    thresholds_dict = json.loads(thresholds_dict_str)
    deploy = classification_thresholds_check(metrics_list[0], thresholds_dict)
    if deploy:
        dep_decision = "true"
    else:
        dep_decision = "false"
    logging.info("deployment decision is %s", dep_decision)

    return (dep_decision,)

مرحله ۲: افزودن اجزای از پیش ساخته شده Google Cloud

در این مرحله، بقیه اجزای خط لوله خود را تعریف خواهیم کرد و خواهیم دید که چگونه همه آنها با هم هماهنگ می‌شوند. ابتدا، نام نمایشی را برای اجرای خط لوله خود با استفاده از یک مهر زمانی تعریف کنید: 

import time
DISPLAY_NAME = 'automl-beans{}'.format(str(int(time.time())))
print(DISPLAY_NAME)

سپس موارد زیر را در یک سلول نوت بوک جدید کپی کنید: 

@pipeline(name="automl-tab-beans-training-v2",
                  pipeline_root=PIPELINE_ROOT)
def pipeline(
    bq_source: str = "bq://aju-dev-demos.beans.beans1",
    display_name: str = DISPLAY_NAME,
    project: str = PROJECT_ID,
    gcp_region: str = "us-central1",
    api_endpoint: str = "us-central1-aiplatform.googleapis.com",
    thresholds_dict_str: str = '{"auRoc": 0.95}',
):
    dataset_create_op = gcc_aip.TabularDatasetCreateOp(
        project=project, display_name=display_name, bq_source=bq_source
    )

    training_op = gcc_aip.AutoMLTabularTrainingJobRunOp(
        project=project,
        display_name=display_name,
        optimization_prediction_type="classification",
        budget_milli_node_hours=1000,
        column_transformations=[
            {"numeric": {"column_name": "Area"}},
            {"numeric": {"column_name": "Perimeter"}},
            {"numeric": {"column_name": "MajorAxisLength"}},
            {"numeric": {"column_name": "MinorAxisLength"}},
            {"numeric": {"column_name": "AspectRation"}},
            {"numeric": {"column_name": "Eccentricity"}},
            {"numeric": {"column_name": "ConvexArea"}},
            {"numeric": {"column_name": "EquivDiameter"}},
            {"numeric": {"column_name": "Extent"}},
            {"numeric": {"column_name": "Solidity"}},
            {"numeric": {"column_name": "roundness"}},
            {"numeric": {"column_name": "Compactness"}},
            {"numeric": {"column_name": "ShapeFactor1"}},
            {"numeric": {"column_name": "ShapeFactor2"}},
            {"numeric": {"column_name": "ShapeFactor3"}},
            {"numeric": {"column_name": "ShapeFactor4"}},
            {"categorical": {"column_name": "Class"}},
        ],
        dataset=dataset_create_op.outputs["dataset"],
        target_column="Class",
    )
    model_eval_task = classification_model_eval_metrics(
        project,
        gcp_region,
        api_endpoint,
        thresholds_dict_str,
        training_op.outputs["model"],
    )

    with dsl.Condition(
        model_eval_task.outputs["dep_decision"] == "true",
        name="deploy_decision",
    ):

        endpoint_op = gcc_aip.EndpointCreateOp(
            project=project,
            location=gcp_region,
            display_name="train-automl-beans",
        )

        gcc_aip.ModelDeployOp(
            model=training_op.outputs["model"],
            endpoint=endpoint_op.outputs["endpoint"],
            dedicated_resources_min_replica_count=1,
            dedicated_resources_max_replica_count=1,
            dedicated_resources_machine_type="n1-standard-4",
        )

بیایید ببینیم در این کد چه اتفاقی می‌افتد:

  • ابتدا، درست مانند خط لوله قبلی، پارامترهای ورودی این خط لوله را تعریف می‌کنیم. از آنجایی که این پارامترها به خروجی مراحل دیگر خط لوله بستگی ندارند، باید آنها را به صورت دستی تنظیم کنیم.
  • بقیه‌ی خط تولید از چند کامپوننت از پیش ساخته شده برای تعامل با سرویس‌های هوش مصنوعی Vertex استفاده می‌کند:
    • TabularDatasetCreateOp یک مجموعه داده جدولی در Vertex AI با توجه به منبع مجموعه داده در Cloud Storage یا BigQuery ایجاد می‌کند. در این pipeline، ما داده‌ها را از طریق URL جدول BigQuery ارسال می‌کنیم.
    • AutoMLTabularTrainingJobRunOp یک کار آموزش AutoML را برای یک مجموعه داده جدولی آغاز می‌کند. ما چند پارامتر پیکربندی را به این کامپوننت ارسال می‌کنیم، از جمله نوع مدل (در این مورد، طبقه‌بندی)، برخی داده‌ها روی ستون‌ها، مدت زمانی که می‌خواهیم آموزش را اجرا کنیم و یک اشاره‌گر به مجموعه داده. توجه داشته باشید که برای ارسال مجموعه داده به این کامپوننت، خروجی کامپوننت قبلی را از طریق dataset_create_op.outputs["dataset"] ارائه می‌دهیم.
    • EndpointCreateOp یک نقطه پایانی در Vertex AI ایجاد می‌کند. نقطه پایانی ایجاد شده از این مرحله به عنوان ورودی به کامپوننت بعدی ارسال می‌شود.
    • ModelDeployOp یک مدل داده شده را در یک نقطه پایانی در Vertex AI مستقر می‌کند. در این حالت، ما از نقطه پایانی ایجاد شده از مرحله قبل استفاده می‌کنیم. گزینه‌های پیکربندی اضافی نیز در دسترس هستند، اما در اینجا نوع ماشین نقطه پایانی و مدلی را که می‌خواهیم مستقر کنیم، ارائه می‌دهیم. ما با دسترسی به خروجی‌های مرحله آموزش در خط لوله خود، مدل را ارسال می‌کنیم.
  • این خط لوله همچنین از منطق شرطی استفاده می‌کند، یکی از ویژگی‌های Vertex Pipelines که به شما امکان می‌دهد یک شرط را به همراه شاخه‌های مختلف بر اساس نتیجه آن شرط تعریف کنید. به یاد داشته باشید که هنگام تعریف خط لوله، پارامتر thresholds_dict_str را ارسال کردیم. این آستانه دقتی است که ما برای تعیین اینکه آیا مدل خود را به یک نقطه پایانی مستقر کنیم یا خیر، استفاده می‌کنیم. برای پیاده‌سازی این، از کلاس Condition از KFP SDK استفاده می‌کنیم. شرطی که ارسال می‌کنیم، خروجی کامپوننت eval سفارشی است که قبلاً در این codelab تعریف کردیم. اگر این شرط درست باشد، خط لوله به اجرای کامپوننت deploy_op ادامه می‌دهد. اگر دقت به آستانه از پیش تعریف شده ما نرسد، خط لوله در اینجا متوقف می‌شود و مدلی را مستقر نمی‌کند.

مرحله ۳: کامپایل و اجرای خط لوله یادگیری ماشینی سرتاسری

با تعریف کامل خط لوله، زمان کامپایل آن فرا رسیده است: 

compiler.Compiler().compile(
    pipeline_func=pipeline, package_path="tab_classif_pipeline.json"
)

بعد، شغل را تعریف کنید: 

ml_pipeline_job = aiplatform.PipelineJob(
    display_name="automl-tab-beans-training",
    template_path="tab_classif_pipeline.json",
    pipeline_root=PIPELINE_ROOT,
    parameter_values={"project": PROJECT_ID, "display_name": DISPLAY_NAME},
    enable_caching=True
)

و در نهایت، کار را اجرا کنید: 

ml_pipeline_job.submit()

پس از اجرای سلول بالا، به لینکی که در لاگ‌ها نشان داده شده است بروید تا خط لوله خود را در کنسول مشاهده کنید. اجرای این خط لوله کمی بیش از یک ساعت طول می‌کشد. بیشتر این زمان صرف مرحله آموزش AutoML می‌شود. خط لوله تکمیل شده چیزی شبیه به این خواهد بود:

خط لوله AutoML تکمیل شد

اگر دکمه‌ی «گسترش مصنوعات» (Expand artifacts) را در بالا فعال یا غیرفعال کنید، می‌توانید جزئیات مصنوعات مختلف ایجاد شده از خط تولید خود را مشاهده کنید. به عنوان مثال، اگر روی مصنوع dataset کلیک کنید، جزئیات مجموعه داده‌ی Vertex AI که ایجاد شده است را مشاهده خواهید کرد. می‌توانید برای رفتن به صفحه‌ی مربوط به آن مجموعه داده، روی لینک اینجا کلیک کنید:

مجموعه داده‌های خط لوله

به طور مشابه، برای دیدن تجسم‌های متریک حاصل از مؤلفه ارزیابی سفارشی ما، روی مصنوع به نام metricsc کلیک کنید. در سمت راست داشبورد خود، می‌توانید ماتریس درهم‌ریختگی این مدل را مشاهده کنید:

تجسم معیارها

برای مشاهده مدل و نقطه پایانی ایجاد شده از این اجرای خط لوله، به بخش مدل‌ها بروید و روی مدلی با نام automl-beans کلیک کنید. automl-beans در آنجا باید ببینید که این مدل در یک نقطه پایانی مستقر شده است:

مدل-نقطه پایانی

همچنین می‌توانید با کلیک کردن روی مصنوع نقطه پایانی در نمودار خط لوله خود به این صفحه دسترسی پیدا کنید.

علاوه بر مشاهده نمودار خط لوله در کنسول، می‌توانید از Vertex Pipelines برای ردیابی دودمان نیز استفاده کنید. منظور از ردیابی دودمان، ردیابی مصنوعات ایجاد شده در سراسر خط لوله شما است. این می‌تواند به ما کمک کند تا بفهمیم مصنوعات کجا ایجاد شده‌اند و چگونه در طول یک گردش کار یادگیری ماشینی استفاده می‌شوند. به عنوان مثال، برای مشاهده ردیابی دودمان برای مجموعه داده ایجاد شده در این خط لوله، روی مصنوع مجموعه داده کلیک کنید و سپس View Lineage را مشاهده کنید :

مشاهده دودمان

این به ما تمام مکان‌هایی را که این مصنوع مورد استفاده قرار می‌گیرد نشان می‌دهد:

جزئیات دودمان

مرحله ۴: مقایسه معیارها در طول مراحل مختلف فروش

اگر این خط لوله را چندین بار اجرا کنید، ممکن است بخواهید معیارها را در طول اجراها مقایسه کنید. می‌توانید از متد aiplatform.get_pipeline_df() برای دسترسی به فراداده‌های اجرا استفاده کنید. در اینجا، فراداده‌های مربوط به همه اجراهای این خط لوله را دریافت کرده و آن را در یک قاب داده Pandas بارگذاری می‌کنیم: 

pipeline_df = aiplatform.get_pipeline_df(pipeline="automl-tab-beans-training-v2")
small_pipeline_df = pipeline_df.head(2)
small_pipeline_df

با این کار، آزمایشگاه را تمام کردید!

🎉 تبریک می‌گویم! 🎉

شما یاد گرفتید که چگونه از Vertex AI برای موارد زیر استفاده کنید:

  • استفاده از Kubeflow Pipelines SDK برای ساخت خطوط لوله سرتاسری با اجزای سفارشی
  • خطوط لوله خود را روی Vertex Pipelines اجرا کنید و اجرای خطوط لوله را با SDK آغاز کنید.
  • نمودار Vertex Pipelines خود را در کنسول مشاهده و تحلیل کنید
  • از اجزای خط لوله از پیش ساخته شده برای افزودن سرویس‌های هوش مصنوعی Vertex به خط لوله خود استفاده کنید
  • زمانبندی کارهای تکراری خط لوله

برای کسب اطلاعات بیشتر در مورد بخش‌های مختلف Vertex، مستندات آن را بررسی کنید.

۷. پاکسازی

برای اینکه هزینه‌ای از شما کسر نشود، توصیه می‌شود منابع ایجاد شده در طول این آزمایش را حذف کنید.

مرحله ۱: نمونه Notebooks خود را متوقف یا حذف کنید

اگر می‌خواهید به استفاده از دفترچه یادداشتی که در این آزمایش ایجاد کرده‌اید ادامه دهید، توصیه می‌شود در صورت عدم استفاده آن را خاموش کنید. از رابط کاربری دفترچه یادداشت‌ها در کنسول ابری خود، دفترچه یادداشت را انتخاب کرده و سپس توقف را انتخاب کنید. اگر می‌خواهید نمونه را به طور کامل حذف کنید، حذف را انتخاب کنید:

مثال را متوقف کنید

مرحله ۲: نقطه پایانی خود را حذف کنید

برای حذف نقطه پایانی که مستقر کرده‌اید، به بخش نقاط پایانی کنسول Vertex AI خود بروید و روی نماد حذف کلیک کنید:

حذف نقطه پایانی

سپس، از اعلان زیر، روی Undeploy کلیک کنید:

مدل Undeployed

در نهایت، به بخش Models کنسول خود بروید، آن مدل را پیدا کنید و از منوی سه نقطه در سمت راست، روی Delete model کلیک کنید:

حذف مدل

مرحله ۳: فضای ذخیره‌سازی ابری خود را حذف کنید

برای حذف Storage Bucket، با استفاده از منوی ناوبری در Cloud Console خود، به Storage بروید، Bucket خود را انتخاب کنید و روی Delete کلیک کنید:

حذف فضای ذخیره‌سازی