مقدمة عن مسارات Vertex

1. نظرة عامة

ستتعلّم في هذا التمرين المعملي كيفية إنشاء مسارات تعلُّم الآلة وتشغيلها باستخدام Vertex Pipelines.

المعلومات التي تطّلع عليها

وستتعرّف على كيفية:

  • استخدام حزمة تطوير البرامج لمنصة Kubeflow Pipelines لإنشاء مسارات تعلُّم الآلة قابلة للتوسّع
  • إنشاء مسار تقديمي من 3 خطوات يتطلّب إدخال النص وتشغيله
  • إنشاء وإدارة مسار تدريب لتدريب نموذج تصنيف AutoML وتقييمه ونشره
  • استخدام مكوّنات معدّة مسبقًا للتفاعل مع خدمات Vertex AI المتوفّرة من خلال مكتبة google_cloud_pipeline_components
  • تحديد موعد لمهمة في مسار التعلّم باستخدام أداة Cloud Scheduler

تبلغ التكلفة الإجمالية لتشغيل هذا التمرين على Google Cloud حوالي 25 دولار أمريكي.

2. مقدّمة عن Vertex AI

يستخدم هذا البرنامج أحدث منتجات الذكاء الاصطناعي المتوفّرة على Google Cloud. تدمج Vertex AI حلول تعلُّم الآلة في Google Cloud ضمن تجربة تطوير سلسة. في السابق، كان الوصول إلى النماذج المدرَّبة باستخدام AutoML والنماذج المخصّصة ممكنًا من خلال خدمات منفصلة. ويدمج العرض الجديد كلاً من واجهة برمجة تطبيقات واحدة مع منتجات جديدة أخرى. يمكنك أيضًا نقل المشاريع الحالية إلى Vertex AI.

بالإضافة إلى خدمات تدريب النماذج ونشرها، يتضمّن Vertex AI أيضًا مجموعة متنوعة من منتجات MLOps، بما في ذلك Vertex Pipelines (الذي يركّز عليه هذا المختبر) ومراقبة النماذج ومتجر الميزات وغيرها. يمكنك الاطّلاع على جميع عروض منتجات Vertex AI في المخطّط البياني أدناه.

نظرة عامة على منتج Vertex

إذا كان لديك أي ملاحظات، يُرجى الاطّلاع على صفحة الدعم.

لماذا تُعد مسارات تعلُّم الآلة مفيدة؟

قبل أن نتعمق في، دعونا أولاً نفهم لماذا قد ترغب في استخدام أحد الأنابيب. لنفترض أنّك تعمل على إنشاء سير عمل لتعلُّم الآلة يتضمّن معالجة البيانات وتدريب نموذج وضبط معلَمة فائقة والتقييم ونشر النموذج. قد يكون لكل خطوة من هذه الخطوات تبعيات مختلفة، والتي قد تصبح غير عملية إذا تعاملت مع سير العمل بأكمله على أنه كتلة واحدة. عندما تبدأ في توسيع نطاق عملية تعلُّم الآلة، قد تريد مشاركة سير عمل تعلُّم الآلة مع الآخرين في فريقك ليتمكّنوا من تشغيله والمساهمة بالرموز البرمجية. يمكن أن يصبح ذلك صعبًا في حال عدم اتّباع عملية موثوقة وقابلة للتكرار. مع المسارات، تكون كل خطوة في عملية تعلُّم الآلة هي حاويتها الخاصة. ويتيح لك ذلك تطوير الخطوات بشكل مستقل وتتبُّع المدخلات والمخرجات من كل خطوة بطريقة قابلة لإعادة الإنتاج. يمكنك أيضًا جدولة عمليات التنفيذ أو تفعيلها استنادًا إلى أحداث أخرى في بيئة Cloud، مثل بدء عملية تنفيذ المشروع عند توفُّر بيانات تدريب جديدة.

النص المختصَر: تساعدك المسارات في تنفيذ المهام آليًا وإعادة إنتاج سير عمل تعلُّم الآلة.

3- إعداد بيئة سحابة إلكترونية

ستحتاج إلى مشروع Google Cloud Platform مع تفعيل الفوترة لتشغيل هذا الدرس التطبيقي حول الترميز. لإنشاء مشروع، يُرجى اتّباع التعليمات هنا.

الخطوة 1: بدء Cloud Shell

ستعمل في هذا التمرين المعملي في جلسة Cloud Shell، وهي عبارة عن مترجم أوامر يستضيفه جهاز افتراضي يعمل على السحابة الإلكترونية من Google. كان بإمكانك تشغيل هذا القسم محليًا بسهولة على جهاز الكمبيوتر، ولكن استخدام Cloud Shell يمنح الجميع إمكانية الوصول إلى تجربة قابلة للتكرار في بيئة متسقة. بعد انتهاء التمرين المعملي، يمكنك إعادة محاولة استخدام هذا القسم على جهاز الكمبيوتر.

منح الإذن في Cloud Shell

تفعيل Cloud Shell

في أعلى يسار Cloud Console، انقر على الزر أدناه من أجل تفعيل Cloud Shell:

تفعيل Cloud Shell

إذا لم يسبق لك بدء تشغيل Cloud Shell، ستظهر لك شاشة وسيطة (الجزء السفلي غير المرئي من الصفحة) تصف ماهيتها. وفي هذه الحالة، انقر على متابعة (ولن تظهر لك مرة أخرى). إليك ما تبدو عليه هذه الشاشة التي تُستخدم لمرة واحدة:

إعداد Cloud Shell

من المفترَض أن تستغرق عملية توفير المتطلبات اللازمة والاتصال بخدمة Cloud Shell بضع دقائق فقط.

إعداد Cloud Shell

يتم تحميل هذا الجهاز الافتراضي مع جميع أدوات التطوير التي تحتاجها. وتوفّر هذه الشبكة دليلاً رئيسيًا دائمًا بسعة 5 غيغابايت ويتم تشغيله في Google Cloud، ما يحسّن بشكل كبير من أداء الشبكة والمصادقة. يمكنك تنفيذ معظم عملك، إن لم يكن كلّه، في هذا الدرس التطبيقي حول الترميز باستخدام متصفّح أو جهاز Chromebook.

بعد الربط بخدمة 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 الذي يحتوي على اسم مشروعنا الحالي على Cloud. سنستخدم هذا في أماكن مختلفة خلال هذا التمرين المعملي. يمكنك الاطّلاع عليه عن طريق تنفيذ:

echo $GOOGLE_CLOUD_PROJECT

الخطوة 2: تفعيل واجهات برمجة التطبيقات

في الخطوات اللاحقة، سترى الحاجة إلى هذه الخدمات (وسبب ذلك)، ولكن في الوقت الحالي، عليك تنفيذ هذا الأمر لمنح مشروعك إذن الوصول إلى خدمات 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.

الخطوة 3: إنشاء حزمة Cloud Storage

لتنفيذ مهمة تدريب على Vertex AI، سنحتاج إلى حزمة تخزين لتخزين مواد عرض النموذج المحفوظة. يجب أن تكون الحزمة إقليمية. نحن نستخدم us-central هنا، ولكن يمكنك استخدام منطقة أخرى (ما عليك سوى استبدالها في هذا التمرين المعملي). إذا سبق لك إنشاء حزمة، يمكنك تخطّي هذه الخطوة.

شغِّل الأوامر التالية في الوحدة الطرفية في Cloud Shell لإنشاء حزمة:

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

بعد ذلك، سنمنح حساب خدمة Compute Engine إذن الوصول إلى هذه الحزمة. سيضمن ذلك حصول 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

الخطوة 4: إنشاء مثيل Vertex AI Workbench

من قسم Vertex AI في Cloud Console، انقر على Workbench:

قائمة Vertex AI

من هناك، ضمن دفاتر الملاحظات التي يديرها المستخدم، انقر على دفتر ملاحظات جديد:

إنشاء ورقة ملاحظات جديدة

بعد ذلك، اختَر نوع المثيل TensorFlow Enterprise 2.3 (مع قناة الدعم الطويل الأمد (LTS) بدون وحدات معالجة الرسومات:

مثيل TFE

استخدِم الخيارات التلقائية، ثم انقر على إنشاء.

الخطوة 5: فتح دفتر ملاحظاتك

بعد إنشاء المثيل، اختَر فتح JupyterLab:

فتح ورقة الملاحظات

4. إعداد خطوط أنابيب Vertex

هناك بعض المكتبات الإضافية التي سنحتاج إلى تثبيتها من أجل استخدام خطوط Vertex Pipelines:

  • Kubeflow Pipelines: هذه هي حزمة تطوير البرامج (SDK) التي سنستخدمها لإنشاء المسار. تتيح مسارات Vertex Pipelines تشغيل خطوط الأنابيب التي تم إنشاؤها باستخدام كل من خطوط Kubeflow Pipelines أو TFX.
  • مكوّنات Google Cloud Pipeline: توفّر هذه المكتبة مكوّنات معدّة مسبقًا تسهّل التفاعل مع خدمات Vertex AI من خلال خطوات سير العمل.

الخطوة 1: إنشاء دفتر ملاحظات Python وتثبيت مكتبات

أولاً، من قائمة "مشغّل التطبيقات" في مثيل ورقة الملاحظات، أنشئ دفتر ملاحظات عن طريق اختيار Python 3:

إنشاء دفتر ملاحظات Python3

يمكنك الوصول إلى قائمة "مشغّل التطبيقات" من خلال النقر على علامة + في أعلى يمين مثيل ورقة الملاحظات.

لتثبيت كلتا الخدمتين اللتين سنستخدمهما في هذا التمرين المعملي، عليك أولاً وضع علامة المستخدم في خلية دفتر ملاحظات:

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)

أخيرًا، تحقق من أنك قمت بتثبيت الحزم بشكل صحيح. يجب أن يكون إصدار حزمة تطوير البرامج (SDK) KFP على الإصدار =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__))"

الخطوة 2: ضبط رقم تعريف مشروعك وحزمتك

خلال هذا الدرس التطبيقي، ستشير إلى رقم تعريف مشروعك على Google Cloud والحزمة التي أنشأتها سابقًا. بعد ذلك سنقوم بإنشاء متغيرات لكل منها.

إذا كنت لا تعرف معرّف مشروعك، فقد تتمكن من الحصول عليه عن طريق تشغيل ما يلي:

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"

الخطوة 3: استيراد المكتبات

أضِف ما يلي لاستيراد المكتبات التي سنستخدمها في هذا الدرس التطبيقي حول الترميز:

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

الخطوة 4: تعريف الثوابت

آخر شيء يتعين علينا القيام به قبل بناء مسارنا هو تحديد بعض المتغيرات الثابتة. PIPELINE_ROOT هو مسار Cloud Storage حيث تتم كتابة العناصر التي أنشأها المسار. نحن نستخدم us-central1 كمنطقة هنا، ولكن إذا استخدمت منطقة مختلفة عند إنشاء الحزمة، عليك تعديل المتغيّر REGION في الرمز أدناه:

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

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

بعد تشغيل التعليمة البرمجية أعلاه، ستتم طباعة الدليل الجذر لمسار العملية. هذا هو موقع Cloud Storage الذي ستتم كتابة العناصر فيه. سيكون بالتنسيق gs://YOUR-BUCKET-NAME/pipeline_root/

5- إنشاء المسار الأول

للتعرف على كيفية عمل Vertex Pipelines، سنعمل أولاً على إنشاء مسار قصير باستخدام حزمة KFP SDK. هذه العملية ليست ذات صلة بتعلُّم الآلة (لا داعي للقلق، سنصل إلى هنا)، فنحن نستخدمه لتعليمك ما يلي:

  • كيفية إنشاء مكوّنات مخصَّصة في حزمة تطوير البرامج (SDK) المستندة إلى KFP
  • كيفية تشغيل ومراقبة أحد الأنابيب في Vertex Pipelines

وسننشئ مسارًا يطبع جملة باستخدام مخرجين: اسم المنتج ووصف الرمز التعبيري. وستتألف هذه العملية من ثلاثة مكونات:

  • product_name: سيأخذ هذا المكوِّن اسم منتج (أو أي اسم تريده حقًا) كإدخال، ويعرض تلك السلسلة كإخراج
  • emoji: سيأخذ هذا المكوِّن الوصف النصي للرمز التعبيري ويحوله إلى إيموجي. على سبيل المثال، الرمز النصي لـ ✨ هو "بريق". يستخدم هذا المكوِّن مكتبة رموز تعبيرية لعرض كيفية إدارة التبعيات الخارجية في مسار التعلّم.
  • build_sentence: سيستهلك هذا المكوِّن الأخير مخرجات الخيارين السابقين لإنشاء جملة تستخدم الرموز التعبيرية. على سبيل المثال، قد يكون الناتج "خطوط Vertex Pipelines هي ✨".

لنبدأ الترميز!

الخطوة 1: إنشاء مكوِّن مستند إلى دالة Python

باستخدام حزمة KFP SDK، يمكننا إنشاء مكونات بناءً على وظائف بايثون. سنستخدم ذلك للمكونات الثلاثة في مسار العملية الأول. سننشئ أولاً المكوِّن product_name الذي يأخذ ببساطة سلسلة كإدخال ويعرض تلك السلسلة. أضف ما يلي إلى دفتر ملاحظاتك:

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

لنلقِ نظرة فاحصة على الصيغة هنا:

  • ويقوم المصمم الزخرفي @component بتجميع هذه الدالة إلى أحد المكونات عند تشغيل المسار. ستستخدم هذا في أي وقت تكتب مكونًا مخصصًا.
  • تحدّد المَعلمة base_image صورة الحاوية التي سيستخدمها هذا المكوّن.
  • المعلمة output_component_file اختيارية، وتحدد ملف yaml لكتابة المكون الذي تم تجميعه. بعد تشغيل الخلية، يُفترض أن ترى هذا الملف مكتوبًا على مثيل دفتر الملاحظات. إذا كنت تريد مشاركة هذا المكون مع شخص ما، فيمكنك إرسال ملف yaml الذي تم إنشاؤه إليه وتحميله مع ما يلي:
product_name_component = kfp.components.load_component_from_file('./first-component.yaml')
  • يحدّد -> str بعد تعريف الدالة نوع الإخراج لهذا المكوِّن.

الخطوة 2: إنشاء مكوّنَين إضافيَين

لإكمال العملية، سنقوم بإنشاء مكونين آخرين. تأخذ السلسلة الأولى التي سنعرّفها سلسلة كإدخال، وتحوِّل هذه السلسلة إلى رمزها التعبيري المقابل في حال توفّرها. يعرض الصف صفًا يحتوي على نص الإدخال الذي تم إدخاله، والرموز التعبيرية الناتجة:

@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 المكوّن بأي تبعيات مكتبة خارجية لهذه الحاوية. ونحن في هذه الحالة نستخدم مكتبة اسمها الرموز التعبيرية.
  • يعرض هذا المكوِّن عنصر NamedTuple باسم Outputs. لاحظ أن كل سلسلة من السلاسل في هذا الصف تحتوي على المفاتيح التالية: 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)

قد تتساءل: كيف يعرف هذا المكون استخدام المخرجات من الخطوات السابقة التي حددتها؟ هذا سؤال وجيه. سنقوم بربطها معًا في الخطوة التالية.

الخطوة 3: وضع المكوّنات معًا في ممر

أدت تعريفات المكونات التي عرّفناها أعلاه إلى إنشاء دوال المصنع التي يمكن استخدامها في تعريف المسار لإنشاء الخطوات. لإعداد مسار، استخدِم أداة التصميم @pipeline وامنح مسار التنفيذ اسمًا ووصفًا وقدِّم المسار الجذر حيث يجب كتابة عناصر المسار. ونقصد بالعناصر أي ملفات إخراج تمّ إنشاؤها من خلال مسار العملية. لا ينشئ مسار المقدّمة هذا أي منها، ولكن مسار التعلّم التالي سينتج عنها.

في المجموعة التالية من الرموز، نحدِّد الدالة intro_pipeline. في ما يلي المكان الذي نحدد فيه مدخلاتنا للخطوات الأولية في سير العمل، وكيف ترتبط هذه الخطوات ببعضها البعض:

  • تأخذ product_task اسم المنتج كإدخال. هنا نمرر "أنابيب Vertex" ولكن يمكنك تغييره كما تشاء.
  • تأخذ emoji_task الرمز النصي للرمز التعبيري كإدخال. يمكنك أيضًا تغيير هذا إلى أي شيء تريده. على سبيل المثال: "party_face" إلى الرمز التعبيري 🥳 يُرجى العلم أنّه لا يتضمّن هذا العنصر والمكوّن product_task أي خطوات تؤدي إلى إدخال البيانات إليهما، وبالتالي نحدّد طريقة إدخالهما يدويًا عند تحديد مسار العملية.
  • تتضمن الخطوة الأخيرة في مسار العملية 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"],
    )

الخطوة 4: تجميع المسار وتشغيله

بعد تحديد المسار، ستكون جاهزًا لتجميعه. سينشئ ما يلي ملف 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()

بعد تشغيل هذه الخلية، من المفترض أن تظهر لك سجلات بها رابط لعرض المسار الذي يتم تشغيله في وحدة التحكم:

سجلات مهام مسار التعلّم

انتقِل إلى ذلك الرابط. من المفترض أن يظهر مسار التنفيذ على النحو التالي عند الانتهاء:

مسار المقدّمة المكتمل

سيستغرق تنفيذ هذا المسار من 5 إلى 6 دقائق. عند الانتهاء، يمكنك النقر على المكوِّن build-sentence للاطّلاع على النتيجة النهائية:

ناتج المسار التمهيدي

بعد أن أصبحت على دراية بطريقة عمل حزمة KFP SDK وVertex Pipelines، أصبحت جاهزًا لإنشاء نموذج تعلُّم الآلة ونشره باستخدام خدمات Vertex AI الأخرى. لنطّلِع على التفاصيل.

6- إنشاء مسار تعلُّم الآلة المتكامل

حان الوقت لإنشاء أول مسار لتعلُّم الآلة. في هذا المسار، سنستخدم مجموعة بيانات الحبوب الجافة للتعلّم الآلي من UCI، من KOKLU وM. وOZKAN, I.A. (2020)، "التصنيف المتعدد الفئات للحبوب الجافة باستخدام رؤية الكمبيوتر وتقنيات التعلم الآلي"، "في قسم أجهزة الكمبيوتر والإلكترونيات في الزراعة، 174، 105507. DOI.

هذه مجموعة بيانات جدولية، وسنستخدمها في سير العمل لدينا لتدريب وتقييم ونشر نموذج AutoML الذي يصنف الحبوب إلى نوع من 7 أنواع بناءً على خصائصها.

وسوف يؤدّي هذا المسار إلى:

  • إنشاء مجموعة بيانات في
  • تدريب نموذج تصنيف جدولي باستخدام AutoML
  • الحصول على مقاييس التقييم على هذا النموذج
  • بناءً على مقاييس التقييم، قرر ما إذا كان سيتم نشر النموذج باستخدام المنطق الشرطي في Vertex Pipelines
  • نشر النموذج إلى نقطة نهاية باستخدام توقُّع Vertex

ستكون كل خطوة من الخطوات الموضحة مكونًا. ستستخدم معظم الخطوات في مسار التعلُّم مكوّنات معدّة مسبقًا لخدمات Vertex AI من خلال مكتبة google_cloud_pipeline_components التي استوردناها سابقًا في هذا الدرس التطبيقي حول الترميز. في هذا القسم، سنحدد مكونًا مخصصًا أولاً. بعد ذلك، سنقوم بتعريف باقي خطوات المسار باستخدام مكونات سابقة الإعداد. تسهّل المكونات المنشأة مسبقًا الوصول إلى خدمات Vertex AI، مثل تدريب النماذج ونشرها.

الخطوة 1: مكوّن مخصَّص لتقييم النماذج

سيتم استخدام المكوِّن المخصص الذي سنحدده في نهاية عملية التدريب بعد اكتمال تدريب النموذج. سيقوم هذا المكون بالقيام ببعض الأشياء:

  • الحصول على مقاييس التقييم من نموذج تصنيف AutoML المدرَّب
  • تحليل المقاييس وعرضها في واجهة مستخدم Vertex Pipelines
  • مقارنة المقاييس بحدّ معيّن لتحديد ما إذا كان يجب نشر النموذج

قبل أن نعرّف المكون، لنفهم معاملي الإدخال والإخراج له. كإدخال، يأخذ هذا المسار بعض البيانات الوصفية في مشروع Google Cloud، والنموذج المدرَّب الناتج (سنحدّد هذا العنصر لاحقًا)، ومقاييس تقييم النموذج، وthresholds_dict_str. سنحدد السمة thresholds_dict_str عند تشغيل مسار العملية. وفي حالة نموذج التصنيف هذا، ستكون هذه هي المنطقة الواقعة تحت قيمة منحنى خاصية تشغيل جهاز الاستقبال الذي ينبغي لنا نشر النموذج لها. فعلى سبيل المثال، إذا مررنا بالمقياس 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,)

الخطوة 2: إضافة مكوّنات 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 AI:
    • ينشئ TabularDatasetCreateOp مجموعة بيانات جدولية في Vertex AI باستخدام مصدر مجموعة بيانات في Cloud Storage أو BigQuery. في هذا المسار، نمرّر البيانات عبر عنوان URL لجدول BigQuery.
    • يبدأ AutoMLTabularTrainingJobRunOp مهمة تدريب AutoML لمجموعة بيانات جدولية. نمرر بعض معلمات التهيئة إلى هذا المكون، بما في ذلك نوع النموذج (في هذه الحالة، التصنيف) وبعض البيانات على الأعمدة والمدة التي نرغب في تنفيذ التدريب عليها ومؤشر إلى مجموعة البيانات. لاحظ أنه لتمرير مجموعة البيانات إلى هذا المكوّن، نقدم ناتج المكوّن السابق عبر dataset_create_op.outputs["dataset"]
    • ينشئ تطبيق "EndpointCreateOp" نقطة نهاية في Vertex AI. سيتم تمرير نقطة النهاية التي تم إنشاؤها من هذه الخطوة كمدخل إلى المكوِّن التالي.
    • ينشر ModelDeployOp نموذجًا معيّنًا إلى نقطة نهاية في Vertex AI. في هذه الحالة، نستخدم نقطة النهاية التي تم إنشاؤها من الخطوة السابقة. تتوفّر خيارات ضبط إضافية، ولكننا نقدّم هنا نوع جهاز نقطة النهاية وطرازه الذي نريد نشره. نمرر النموذج من خلال الوصول إلى مخرجات خطوة التدريب في مسارنا
  • يستخدم هذا المسار أيضًا المنطق الشرطي، وهي ميزة في مسارات Vertex Pipelines تتيح لك تحديد الشرط، إلى جانب الفروع المختلفة بناءً على نتيجة هذا الشرط. تذكر أنه عندما حددنا مسارنا، مررنا المعلمة thresholds_dict_str. وهذا هو حد الدقة الذي نستخدمه لتحديد ما إذا كان سيتم نشر نموذجنا في نقطة نهاية. لتنفيذ ذلك، نستخدم الفئة Condition من حزمة تطوير البرامج (SDK) الخاصة بـ KFP. الشرط الذي نمرّره هو مخرج مكوّن التقييم المخصّص الذي حدّدناه سابقًا في هذا الدرس التطبيقي حول الترميز. إذا كان هذا الشرط صحيحًا، ستستمر العملية في تنفيذ المكوِّن deploy_op. وإذا لم تتوافق الدقة مع الحد الأدنى المحدد مسبقًا، سيتوقف المسار هنا ولن يتم نشر أي نموذج.

الخطوة 3: تجميع وإدارة مسار تعلُّم الآلة الشامل

بعد تحديد المسار الكامل، حان الوقت لتجميعه:

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 مكتمل

في حال تبديل "توسيع العناصر" في الجزء العلوي، ستتمكن من رؤية تفاصيل حول الأدوات المختلفة التي تم إنشاؤها من خلال المسار. على سبيل المثال، إذا نقرت على عنصر dataset، ستظهر لك تفاصيل على مجموعة بيانات Vertex AI التي تم إنشاؤها. يمكنك النقر فوق الرابط هنا للانتقال إلى صفحة مجموعة البيانات هذه:

مجموعة بيانات الأنابيب

وبالمثل، للاطّلاع على العروض المرئية للمقاييس الناتجة من مكوّن التقييم المخصّص، انقر على العنصر المسمّى metricsc. في الجانب الأيسر من لوحة المعلومات، ستتمكن من رؤية مصفوفة التشويش لهذا النموذج:

تصور المقاييس

للاطّلاع على النموذج ونقطة النهاية اللذين تم إنشاؤهما من مسار المسار هذا، انتقِل إلى قسم النماذج وانقر على النموذج المُسمّى automl-beans. وفي هذه الحالة، سوف يظهر لك هذا النموذج منشورًا في نقطة نهاية:

نقطة النهاية للنموذج

يمكنك أيضًا الوصول إلى هذه الصفحة من خلال النقر على عنصر نقطة النهاية في الرسم البياني لمسار التعلّم.

بالإضافة إلى إلقاء نظرة على الرسم البياني لمخطط البيانات في وحدة التحكّم، يمكنك أيضًا استخدام مسارات Vertex لتتبُّع البيانات الخطية. نعني بتتبُّع السلالة تتبُّع العناصر التي يتم إنشاؤها خلال مسار العملية. يمكن أن يساعدنا ذلك في فهم مكان إنشاء الأدوات وكيفية استخدامها خلال سير عمل تعلُّم الآلة. على سبيل المثال، للاطّلاع على تتبُّع سلالة مجموعة البيانات التي تم إنشاؤها في مسار العملية هذا، انقر على عنصر مجموعة البيانات، ثم عرض خط الأساس:

عرض النسب

يوضح لنا هذا جميع الأماكن التي يتم فيها استخدام هذا العنصر:

تفاصيل عدد العناصر

الخطوة 4: مقارنة المقاييس عبر مسارات التعلّم

إذا قمت بتشغيل هذا المسار عدة مرات، فقد ترغب في مقارنة المقاييس عبر عمليات التشغيل. يمكنك استخدام الطريقة aiplatform.get_pipeline_df() للوصول إلى البيانات الوصفية قيد التنفيذ. سنحصل هنا على بيانات وصفية لجميع عمليات تنفيذ هذا المسار ونحمّلها في Pandas DataFrame:

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 لإنشاء خطوط بيانات شاملة باستخدام مكوّنات مخصّصة
  • تشغيل المسارات على Vertex Pipelines وبدء عمليات التشغيل باستخدام حزمة تطوير البرامج (SDK)
  • عرض الرسم البياني لخطوط Vertex Pipelines وتحليله في وحدة التحكّم
  • استخدِم مكوّنات جاهزة مصمَّمة مسبقًا لإضافة خدمات Vertex AI إلى مسار التعلّم
  • جدولة المهام المتكررة لمسار التعلّم

لمزيد من المعلومات حول أجزاء مختلفة من Vertex، يمكنك الاطّلاع على المستندات.

7. تنظيف

حتى لا يتم تحصيل رسوم منك، ننصحك بحذف الموارد التي تم إنشاؤها خلال هذا الدرس التطبيقي.

الخطوة 1: إيقاف النسخة الافتراضية من دفاتر الملاحظات أو حذفها

إذا أردت مواصلة استخدام الدفتر الذي أنشأته في هذا التمرين المعملي، ننصحك بإيقافه عندما لا يكون قيد الاستخدام. من واجهة مستخدم "دفاتر الملاحظات" في Cloud Console، اختَر ورقة الملاحظات ثم انقر على إيقاف. إذا أردت حذف المثيل بأكمله، انقر على حذف:

إيقاف المثيل

الخطوة 2: حذف نقطة النهاية

لحذف نقطة النهاية التي نشرتها، انتقِل إلى قسم نقاط النهاية في وحدة تحكُّم Vertex AI وانقر على رمز الحذف:

حذف نقطة النهاية

بعد ذلك، انقر على إلغاء النشر من الطلب التالي:

إلغاء نشر النموذج

أخيرًا، انتقِل إلى قسم النماذج في وحدة التحكّم، وابحث عن هذا النموذج، ومن قائمة الخيارات الإضافية على يسار الصفحة، انقر على حذف النموذج:

حذف النموذج

الخطوة 3: حذف حزمة Cloud Storage

لحذف "حزمة التخزين"، باستخدام قائمة التنقّل في Cloud Console، انتقِل إلى "مساحة التخزين" واختَر الحزمة وانقر على "حذف":

حذف مساحة التخزين