שימוש ב-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. מיקום המאגר יהיה זמין באופן אוטומטי דרך משתנה סביבה מוגדר מראש AIP_TENSORBOARD_LOG_DIR בשירות האימון של Gemini Enterprise Agent Platform.

בדרך כלל אפשר לעשות את זה על ידי ציון 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 Pipelines.

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

אורזים את קוד האימון לרכיב מותאם אישית, ומוודאים שהקוד מוגדר לכתוב יומני 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 למיקום לאחסון זמני של ארטיפקטים במהלך קריאות ל-API (כולל יומני TensorBoard).

לאחר מכן, יוצרים צינור להכללת המשימה הזו ומקמפלים את הצינור לקובץ 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"
    )

שליחת צינור (pipeline) של 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_root: מציינים URI של Cloud Storage שחשבון השירות של צינורות העיבוד יכול לגשת אליו. הפריטים שנוצרו מהרצת צינור הנתונים מאוחסנים בתיקיית הבסיס של צינור הנתונים.
  • ‫parameter_values: פרמטרים של הפייפליין להעברה להרצה הזו. לדוגמה, אפשר ליצור dict() עם שמות הפרמטרים כמפתחות המילון וערכי הפרמטרים כערכי המילון.
  • project: . הפרויקט ב-Google Cloud שבו יופעל צינור הנתונים. אפשר למצוא את המזהים בדף welcome במסוף Google Cloud .
  • ‫location: המיקום שבו יופעל צינור העברת הנתונים. זה צריך להיות אותו מיקום כמו המיקום של מופע TensorBoard שבו אתם משתמשים.

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