שימוש ב-Vertex AI TensorBoard עם פייפליינים של Gemini Enterprise Agent Platform

אפשר לארוז את קוד האימון ברכיב אימון בהתאמה אישית ולהריץ אותו במשימת צינור עיבוד נתונים. יומני TensorBoard מועברים אוטומטית בסטרימינג לניסוי שלכם ב-Vertex AI TensorBoard. אתם יכולים להשתמש בשילוב הזה כדי לעקוב אחרי האימון כמעט בזמן אמת, כי Vertex AI TensorBoard מעביר בסטרימינג את היומנים של Vertex AI TensorBoard בזמן שהם נכתבים ב-Cloud Storage.

להגדרה הראשונית, ראו הגדרה של Vertex AI TensorBoard.

שינויים בתסריט האימון

סקריפט האימון צריך להיות מוגדר לכתיבת יומני TensorBoard בקטגוריה של Cloud Storage, שהמיקום שלה יהיה זמין באופן אוטומטי על ידי שירות האימון של Gemini Enterprise Agent Platform באמצעות משתנה סביבה מוגדר מראש AIP_TENSORBOARD_LOG_DIR.

בדרך כלל אפשר לעשות את זה על ידי ציון os.environ['AIP_TENSORBOARD_LOG_DIR'] בתור ספריית היומן לממשקי ה-API של TensorBoard בקוד פתוח לכתיבת יומנים. המיקום של AIP_TENSORBOARD_LOG_DIR מוגדר בדרך כלל באמצעות המשתנה staging_bucket.

כדי להגדיר את סקריפט האימון ב-TensorFlow 2.x, יוצרים קריאה חוזרת (callback) של TensorBoard ומגדירים את המשתנה log_dir ל-os.environ['AIP_TENSORBOARD_LOG_DIR']. לאחר מכן, הקריאה החוזרת של TensorBoard נכללת ברשימת הקריאות החוזרות של TensorFlow model.fit.

  tensorboard_callback = tf.keras.callbacks.TensorBoard(
       log_dir=os.environ['AIP_TENSORBOARD_LOG_DIR'],
       histogram_freq=1
  )
  
  model.fit(
       x=x_train,
       y=y_train,
       epochs=epochs,
       validation_data=(x_test, y_test),
       callbacks=[tensorboard_callback],
  )

איך Agent Platform מגדירה משתני סביבה בסביבת האימון בהתאמה אישית

פיתוח והרצה של צינור עיבוד נתונים

בדוגמה הבאה מוסבר איך ליצור ולהריץ צינור עיבוד נתונים באמצעות חבילת Kubeflow Pipelines DSL. לדוגמאות נוספות ולפרטים נוספים, קראו את המסמכים בנושא צינורות עיבוד נתונים ב-Agent Platform.

יצירת רכיב הדרכה

אורזים את קוד האימון ברכיב בהתאמה אישית, ומוודאים שהקוד מוגדר לכתוב את יומני TensorBoard לקטגוריה של Cloud Storage. דוגמאות נוספות מופיעות במאמר בניית רכיבי צינור עיבוד נתונים משלכם.

from kfp.v2.dsl import component

@component(
    base_image="tensorflow/tensorflow:latest",
    packages_to_install=["tensorflow_datasets"],
)
def train_tensorflow_model_with_tensorboard():
    import datetime, os
    import tensorflow as tf

    (x_train, y_train), (x_test, y_test) = tf.keras.datasets.mnist.load_data()
    x_train, x_test = x_train / 255.0, x_test / 255.0

    def create_model():
        return tf.keras.models.Sequential(
            [
                tf.keras.layers.Flatten(input_shape=(28, 28)),
                tf.keras.layers.Dense(512, activation="relu"),
            ]
        )

    model = create_model()
    model.compile(
        optimizer="adam",
        loss="sparse_categorical_crossentropy",
        metrics=["accuracy"]
    )

    tensorboard_callback = tf.keras.callbacks.TensorBoard(
        log_dir=os.environ['AIP_TENSORBOARD_LOG_DIR'],
        histogram_freq=1
    )

    model.fit(
        x=x_train,
        y=y_train,
        epochs=5,
        validation_data=(x_test, y_test),
        callbacks=[tensorboard_callback],
    )

פיתוח וקומפילציה של צינור עיבוד נתונים

יוצרים משימת אימון בהתאמה אישית מהרכיב שיצרתם, על ידי ציון מפרט הרכיב ב-create_custom_training_job_op_from_component. מגדירים את tensorboard_resource_name למכונה של TensorBoard, ואת staging_bucket למיקום שבו רוצים להעביר פריטי מידע (כולל יומני TensorBoard) במהלך קריאות ל-API.

לאחר מכן, יוצרים צינור עיבוד נתונים שכולל את המשימה הזו ומקמפלים את צינור עיבוד הנתונים לקובץ JSON.

דוגמאות ומידע נוסף זמינים במאמרים רכיבים של משימות בהתאמה אישית ובניית צינור עיבוד נתונים.

from kfp.v2 import compiler
from google_cloud_pipeline_components.v1.custom_job.utils import \
    create_custom_training_job_op_from_component
from kfp.v2 import dsl

def create_tensorboard_pipeline_sample(
    project, location, staging_bucket, display_name, service_account, experiment, tensorboard_resource_name
):

    @dsl.pipeline(
        pipeline_root=f"{staging_bucket}/pipeline_root",
        name=display_name,
    )
    def pipeline():
        custom_job_op = create_custom_training_job_op_from_component(
            component_spec=train_tensorflow_model_with_tensorboard,
            tensorboard=tensorboard_resource_name,
            base_output_directory=staging_bucket,
            service_account=service_account,
        )
        custom_job_op(project=project, location=location)

    compiler.Compiler().compile(
        pipeline_func=pipeline, package_path=f"{display_name}.json"
    )

שליחת צינור עיבוד נתונים ב-Gemini Enterprise Agent Platform

שולחים את צינור עיבוד הנתונים באמצעות Agent Platform SDK ל-Python. למידע נוסף, ראו הרצת צינור עיבוד נתונים.

Python

from typing import Any, Dict, Optional

from google.cloud import aiplatform


def log_pipeline_job_to_experiment_sample(
    experiment_name: str,
    pipeline_job_display_name: str,
    template_path: str,
    pipeline_root: str,
    project: str,
    location: str,
    parameter_values: Optional[Dict[str, Any]] = None,
):
    aiplatform.init(project=project, location=location)

    pipeline_job = aiplatform.PipelineJob(
        display_name=pipeline_job_display_name,
        template_path=template_path,
        pipeline_root=pipeline_root,
        parameter_values=parameter_values,
    )

    pipeline_job.submit(experiment=experiment_name)
  • ‫experiment_name: נותנים שם לניסוי.
  • ‫pipeline_job_display_name: השם המוצג של משימת הצינור.
  • ‫template_path: הנתיב לתבנית ה-pipeline שעברה הידור.
  • ‫pipeline_root: מציינים את ה-URI של Cloud Storage שחשבון השירות של צינורות עיבוד הנתונים יכול לגשת אליו. הפריטים שנוצרו מהרצות של צינורות עיבוד נתונים מאוחסנים בתיקיית השורש של צינור עיבוד הנתונים.
  • parameter_values: הפרמטרים של צינור עיבוד הנתונים שצריך להעביר להרצה הזו. לדוגמה, יוצרים dict() עם שמות הפרמטרים כמפתחות המילון וערכי הפרמטרים כערכי המילון.
  • project: . הפרויקט ב- Google Cloud שבו מריצים את צינור עיבוד הנתונים. אפשר למצוא את המזהים בדף הכניסה למסוף Google Cloud .
  • ‫location: המיקום שבו רוצים להריץ את צינור עיבוד הנתונים. המיקום צריך להיות זהה למיקום של מכונת TensorBoard שבה אתם משתמשים.

המאמרים הבאים