הפעלת LLM בצינור עיבוד נתונים בסטרימינג

במדריך הזה נסביר איך להריץ מודל שפה גדול (LLM) בצינור עיבוד נתונים של Dataflow באמצעות Apache Beam RunInference API.

מידע נוסף על RunInference API זמין במאמר About Beam ML במסמכי התיעוד של Apache Beam.

קוד הדוגמה זמין ב-GitHub.

מטרות

• יצירת נושאים ומינויים ב-Pub/Sub עבור הקלט והתגובות של המודל.
• טוענים את המודל ל-Cloud Storage באמצעות משימה בהתאמה אישית ב-Gemini Enterprise Agent Platform.
• מריצים את הפייפליין.
• שואלים את המודל שאלה ומקבלים תשובה.

עלויות

במסמך הזה משתמשים ברכיבים הבאים של Google Cloud, והשימוש בהם כרוך בתשלום:

• Dataflow
• Cloud Storage
• Pub/Sub
• Agent Platform

כדי להעריך את ההוצאות בהתאם לתחזית השימוש שלכם, אתם יכולים להיעזר במחשבון העלויות.

משתמשים חדשים של Google Cloud ? יכול להיות שאתם זכאים לתקופת ניסיון בחינם.

כשמסיימים את המשימות שמתוארות במסמך הזה אפשר למחוק את המשאבים שיצרתם כדי להימנע מחיובים נוספים. מידע נוסף זמין בקטע הסרת המשאבים.

לפני שמתחילים

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

נכנסים לחשבון Google Cloud . אם אתם משתמשים חדשים ב- Google Cloud, צרו חשבון כדי שתוכלו להעריך את הביצועים של המוצרים שלנו בתרחישים מהעולם האמיתי. לקוחות חדשים מקבלים בחינם גם קרדיט בשווי 300$ להרצה, לבדיקה ולפריסה של עומסי העבודה.
התקינו את ה-CLI של Google Cloud.
אם אתם משתמשים בספק זהויות חיצוני (IdP), קודם אתם צריכים להיכנס ל-CLI של gcloud באמצעות המאגר המאוחד לניהול זהויות.
כדי לאתחל את ה-CLI של gcloud, הריצו את הפקודה הבאה:
gcloud init
יוצרים או בוחרים Google Cloud פרויקט.
תפקידים שנדרשים כדי לבחור או ליצור פרויקט
• Select a project: כדי לבחור פרויקט לא צריך תפקיד IAM ספציפי – אפשר לבחור כל פרויקט שקיבלתם בו תפקיד.
• יצירת פרויקט: כדי ליצור פרויקט, צריך את התפקיד Project Creator (יצירת פרויקטים) (roles/resourcemanager.projectCreator), שכולל את ההרשאה resourcemanager.projects.create. איך מקצים תפקידים
הערה: אם אתם לא מתכננים לשמור את המשאבים שתיצרו בתהליך הזה, תוכלו ליצור פרויקט חדש במקום לבחור באחד מהפרויקטים הקיימים. בסיום התהליך תוכלו למחוק את הפרויקט ולהסיר את כל המשאבים שמשויכים אליו.
• יוצרים Google Cloud פרויקט:

  gcloud projects create PROJECT_ID

  מחליפים את PROJECT_ID בשם של פרויקט Google Cloud שיוצרים.

• בוחרים את הפרויקט שיצרתם: Google Cloud

  gcloud config set project PROJECT_ID

  מחליפים את PROJECT_ID בשם הפרויקט ב- Google Cloud .
מוודאים שהחיוב מופעל בפרויקט Google Cloud .
מפעילים את ממשקי ה-API של Dataflow,‏ Compute Engine,‏ Cloud Storage,‏ Pub/Sub ו-Agent Platform:
תפקידים שנדרשים להפעלת ממשקי API
כדי להפעיל ממשקי API, נדרשת ההרשאה serviceusage.services.enable. אם יצרתם את הפרויקט, סביר להניח שכבר יש לכם את ההרשאה הזו דרך התפקיד 'בעלים' (roles/owner). אחרת, תוכלו לקבל את ההרשאה הזו דרך התפקיד 'אדמין של Service Usage' (roles/serviceusage.serviceUsageAdmin). איך מקצים תפקידים
gcloud services enable dataflow.googleapis.com compute.googleapis.com storage.googleapis.com pubsub.googleapis.com aiplatform.googleapis.com
אם אתם משתמשים במעטפת מקומית, אתם צריכים ליצור פרטי כניסה לאימות מקומי עבור חשבון המשתמש:
gcloud auth application-default login
אם אתם משתמשים ב-Cloud Shell, אין צורך לבצע את הפעולה הזו.
אם מוחזרת שגיאת אימות ואתם משתמשים בספק זהויות חיצוני (IdP), ודאו ש נכנסתם ל-CLI של gcloud באמצעות המאגר המאוחד לניהול זהויות.
מעניקים תפקידים לחשבון המשתמש. מריצים את הפקודה הבאה לכל אחד מהתפקידים הבאים ב-IAM: roles/iam.serviceAccountUser
gcloud projects add-iam-policy-binding PROJECT_ID --member="user:USER_IDENTIFIER" --role=ROLE
מחליפים את מה שכתוב בשדות הבאים:
• ‫PROJECT_ID: מזהה הפרויקט.
• ‫USER_IDENTIFIER: המזהה של חשבון המשתמש . לדוגמה, [e-mail].
• ‫ROLE: תפקיד ה-IAM שאתם מקצים לחשבון המשתמש.

התקינו את ה-CLI של Google Cloud.

אם אתם משתמשים בספק זהויות חיצוני (IdP), קודם אתם צריכים להיכנס ל-CLI של gcloud באמצעות המאגר המאוחד לניהול זהויות.

כדי לאתחל את ה-CLI של gcloud, הריצו את הפקודה הבאה:

gcloud init

יוצרים או בוחרים Google Cloud פרויקט.

תפקידים שנדרשים כדי לבחור או ליצור פרויקט

• Select a project: כדי לבחור פרויקט לא צריך תפקיד IAM ספציפי – אפשר לבחור כל פרויקט שקיבלתם בו תפקיד.
• יצירת פרויקט: כדי ליצור פרויקט, צריך את התפקיד Project Creator (יצירת פרויקטים) (roles/resourcemanager.projectCreator), שכולל את ההרשאה resourcemanager.projects.create. איך מקצים תפקידים

הערה: אם אתם לא מתכננים לשמור את המשאבים שתיצרו בתהליך הזה, תוכלו ליצור פרויקט חדש במקום לבחור באחד מהפרויקטים הקיימים. בסיום התהליך תוכלו למחוק את הפרויקט ולהסיר את כל המשאבים שמשויכים אליו.

• יוצרים Google Cloud פרויקט:

  gcloud projects create PROJECT_ID

  מחליפים את PROJECT_ID בשם של פרויקט Google Cloud שיוצרים.

• בוחרים את הפרויקט שיצרתם: Google Cloud

  gcloud config set project PROJECT_ID

  מחליפים את PROJECT_ID בשם הפרויקט ב- Google Cloud .

מוודאים שהחיוב מופעל בפרויקט Google Cloud .

מפעילים את ממשקי ה-API של Dataflow,‏ Compute Engine,‏ Cloud Storage,‏ Pub/Sub ו-Agent Platform:

תפקידים שנדרשים להפעלת ממשקי API

כדי להפעיל ממשקי API, נדרשת ההרשאה serviceusage.services.enable. אם יצרתם את הפרויקט, סביר להניח שכבר יש לכם את ההרשאה הזו דרך התפקיד 'בעלים' (roles/owner). אחרת, תוכלו לקבל את ההרשאה הזו דרך התפקיד 'אדמין של Service Usage' (roles/serviceusage.serviceUsageAdmin). איך מקצים תפקידים

gcloud services enable dataflow.googleapis.com compute.googleapis.com storage.googleapis.com pubsub.googleapis.com aiplatform.googleapis.com

אם אתם משתמשים במעטפת מקומית, אתם צריכים ליצור פרטי כניסה לאימות מקומי עבור חשבון המשתמש:

gcloud auth application-default login

אם אתם משתמשים ב-Cloud Shell, אין צורך לבצע את הפעולה הזו.

אם מוחזרת שגיאת אימות ואתם משתמשים בספק זהויות חיצוני (IdP), ודאו ש נכנסתם ל-CLI של gcloud באמצעות המאגר המאוחד לניהול זהויות.

מעניקים תפקידים לחשבון המשתמש. מריצים את הפקודה הבאה לכל אחד מהתפקידים הבאים ב-IAM: roles/iam.serviceAccountUser

gcloud projects add-iam-policy-binding PROJECT_ID --member="user:USER_IDENTIFIER" --role=ROLE

מחליפים את מה שכתוב בשדות הבאים:

• ‫PROJECT_ID: מזהה הפרויקט.
• ‫USER_IDENTIFIER: המזהה של חשבון המשתמש . לדוגמה, [e-mail].
• ‫ROLE: תפקיד ה-IAM שאתם מקצים לחשבון המשתמש.

1. מקצים תפקידים לחשבון השירות שמוגדר כברירת מחדל ב-Compute Engine. מריצים את הפקודה הבאה לכל אחד מהתפקידים הבאים ב-IAM:

   • roles/dataflow.admin
   • roles/dataflow.worker
   • roles/storage.admin
   • roles/pubsub.editor
   • roles/aiplatform.user

   gcloud projects add-iam-policy-binding PROJECT_ID --member="serviceAccount:[e-mail]" --role=SERVICE_ACCOUNT_ROLE

   מחליפים את מה שכתוב בשדות הבאים:

   • PROJECT_ID: מזהה הפרויקט.
   • ‫PROJECT_NUMBER: מספר הפרויקט. כדי למצוא את מספר הפרויקט, משתמשים בפקודה gcloud projects describe.
   • ‫SERVICE_ACCOUNT_ROLE: כל תפקיד בנפרד.
2. מעתיקים את מזהה הפרויקט Google Cloud . תצטרכו את הערך הזה בהמשך המדריך.

יצירת Google Cloud המשאבים

בקטע הזה מוסבר איך ליצור את המשאבים הבאים:

• קטגוריה של Cloud Storage לשימוש כמיקום אחסון זמני
• נושא Pub/Sub להנחיות של המודל
• נושא ומינוי ב-Pub/Sub לתשובות של המודל

יצירת קטגוריה של Cloud Storage

יוצרים קטגוריה של Cloud Storage באמצעות ה-CLI של gcloud. הבאקט הזה משמש כמיקום אחסון זמני על ידי צינור Dataflow.

כדי ליצור את הקטגוריה, משתמשים בפקודה gcloud storage buckets create:

gcloud storage buckets create gs://BUCKET_NAME --location=LOCATION

מחליפים את מה שכתוב בשדות הבאים:

• ‫BUCKET_NAME: שם לקטגוריה של Cloud Storage שעומד בדרישות למתן שמות לקטגוריות. שמות של קטגוריות ב-Cloud Storage חייבים להיות ייחודיים באופן גלובלי.
• ‫LOCATION: המיקום של הקטגוריה.

מעתיקים את שם הקטגוריה. תצטרכו את הערך הזה בהמשך המדריך.

יצירת נושאים ומינויים ב-Pub/Sub

יוצרים שני נושאים ב-Pub/Sub ומינוי אחד. נושא אחד הוא הנחיות הקלט שאתם שולחים למודל. הנושא השני והמינוי שמצורף אליו מיועדים לתשובות של המודל.

1. כדי ליצור את הנושאים, מריצים את הפקודה gcloud pubsub topics create פעמיים, פעם אחת לכל נושא:

   gcloud pubsub topics create PROMPTS_TOPIC_ID
   gcloud pubsub topics create RESPONSES_TOPIC_ID
   

   מחליפים את מה שכתוב בשדות הבאים:

   • ‫PROMPTS_TOPIC_ID: מזהה הנושא של ההנחיות לקלט שנשלחות למודל, כמו prompts
   • ‫RESPONSES_TOPIC_ID: מזהה הנושא של התשובות של המודל, כמו responses

2. כדי ליצור את המינוי ולצרף אותו לנושא התגובות, משתמשים בפקודה gcloud pubsub subscriptions create:

   gcloud pubsub subscriptions create RESPONSES_SUBSCRIPTION_ID --topic=RESPONSES_TOPIC_ID
   

   מחליפים את RESPONSES_SUBSCRIPTION_ID במזהה המינוי של תשובות המודל, לדוגמה responses-subscription.

מעתיקים את מזהי הנושא ואת מזהה המינוי. תצטרכו את הערכים האלה בהמשך המדריך הזה.

הכנת הסביבה

מורידים את דוגמאות הקוד ומגדירים את הסביבה להרצת ההדרכה.

דוגמאות הקוד במאגר python-docs-samples ב-GitHub מספקות את הקוד שנדרש להרצת צינור העיבוד הזה. כשאתם מוכנים ליצור צינור משלכם, אתם יכולים להשתמש בקוד לדוגמה הזה כתבנית.

כדי להריץ את פרויקט צינור עיבוד הנתונים, יוצרים סביבה וירטואלית מבודדת של Python באמצעות venv. סביבה וירטואלית מאפשרת לבודד את התלות של פרויקט אחד מהתלות של פרויקטים אחרים. מידע נוסף על התקנת Python ויצירת סביבה וירטואלית מופיע במאמר הגדרת סביבת פיתוח ב-Python.

1. משתמשים בפקודה git clone כדי לשכפל את מאגר GitHub:

   git clone https://github.com/GoogleCloudPlatform/python-docs-samples.git
   

2. מנווטים לספרייה run-inference:

   cd python-docs-samples/dataflow/run-inference
   

3. אם אתם משתמשים בשורת פקודה, צריך לוודא ש-Python 3 ו-pip פועלים במערכת:

   python --version
   python -m pip --version
   

   במקרה הצורך, מתקינים את Python 3.

   אם אתם משתמשים ב-Cloud Shell, אתם יכולים לדלג על השלב הזה כי Python כבר מותקן ב-Cloud Shell.

4. יוצרים סביבה וירטואלית של Python:

   python -m venv /tmp/env
   source /tmp/env/bin/activate
   

5. מתקינים את יחסי התלות:

   pip install -r requirements.txt --no-cache-dir
   

דוגמת קוד לטעינת מודל

קוד טעינת המודל במדריך הזה מפעיל משימה מותאמת אישית של Agent Platform שטוענת את אובייקט state_dict של המודל ל-Cloud Storage.

קובץ ההתחלה נראה כך:

# Copyright 2023 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Loads the state_dict for an LLM model into Cloud Storage."""

from __future__ import annotations

import os

import torch
from transformers import AutoModelForSeq2SeqLM


def run_local(model_name: str, state_dict_path: str) -> None:
    """Loads the state dict and saves it into the desired path.

    If the `state_dict_path` is a Cloud Storage location starting
    with "gs://", this assumes Cloud Storage is mounted with
    Cloud Storage FUSE in `/gcs`. Vertex AI is set up like this.

    Args:
        model_name: HuggingFace model name compatible with AutoModelForSeq2SeqLM.
        state_dict_path: File path to the model's state_dict, can be in Cloud Storage.
    """
    print(f"Loading model: {model_name}")
    model = AutoModelForSeq2SeqLM.from_pretrained(
        model_name, torch_dtype=torch.bfloat16
    )
    print(f"Model loaded, saving state dict to: {state_dict_path}")

    # Assume Cloud Storage FUSE is mounted in `/gcs`.
    state_dict_path = state_dict_path.replace("gs://", "/gcs/")
    directory = os.path.dirname(state_dict_path)
    if directory and not os.path.exists(directory):
        os.makedirs(os.path.dirname(state_dict_path), exist_ok=True)
    torch.save(model.state_dict(), state_dict_path)
    print("State dict saved successfully!")


def run_vertex_job(
    model_name: str,
    state_dict_path: str,
    job_name: str,
    project: str,
    bucket: str,
    location: str = "us-central1",
    machine_type: str = "e2-highmem-2",
    disk_size_gb: int = 100,
) -> None:
    """Launches a Vertex AI custom job to load the state dict.

    If the model is too large to fit into memory or disk, we can launch
    a Vertex AI custom job with a large enough VM for this to work.

    Depending on the model's size, it might require a different VM
    configuration. The model MUST fit into the VM's memory, and there
    must be enough disk space to stage the entire model while it gets
    copied to Cloud Storage.

    Args:
        model_name: HuggingFace model name compatible with AutoModelForSeq2SeqLM.
        state_dict_path: File path to the model's state_dict, can be in Cloud Storage.
        job_name: Job display name in the Vertex AI console.
        project: Google Cloud Project ID.
        bucket: Cloud Storage bucket name, without the "gs://" prefix.
        location: Google Cloud regional location.
        machine_type: Machine type for the VM to run the job.
        disk_size_gb: Disk size in GB for the VM to run the job.
    """
    from google.cloud import aiplatform

    aiplatform.init(project=project, staging_bucket=bucket, location=location)

    job = aiplatform.CustomJob.from_local_script(
        display_name=job_name,
        container_uri="us-docker.pkg.dev/vertex-ai/training/pytorch-gpu.1-13:latest",
        script_path="download_model.py",
        args=[
            "local",
            f"--model-name={model_name}",
            f"--state-dict-path={state_dict_path}",
        ],
        machine_type=machine_type,
        boot_disk_size_gb=disk_size_gb,
        requirements=["transformers"],
    )
    job.run()


if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser()
    subparsers = parser.add_subparsers(required=True)

    parser_local = subparsers.add_parser("local")
    parser_local.add_argument(
        "--model-name",
        required=True,
        help="HuggingFace model name compatible with AutoModelForSeq2SeqLM",
    )
    parser_local.add_argument(
        "--state-dict-path",
        required=True,
        help="File path to the model's state_dict, can be in Cloud Storage",
    )
    parser_local.set_defaults(run=run_local)

    parser_vertex = subparsers.add_parser("vertex")
    parser_vertex.add_argument(
        "--model-name",
        required=True,
        help="HuggingFace model name compatible with AutoModelForSeq2SeqLM",
    )
    parser_vertex.add_argument(
        "--state-dict-path",
        required=True,
        help="File path to the model's state_dict, can be in Cloud Storage",
    )
    parser_vertex.add_argument(
        "--job-name", required=True, help="Job display name in the Vertex AI console"
    )
    parser_vertex.add_argument(
        "--project", required=True, help="Google Cloud Project ID"
    )
    parser_vertex.add_argument(
        "--bucket",
        required=True,
        help='Cloud Storage bucket name, without the "gs://" prefix',
    )
    parser_vertex.add_argument(
        "--location", default="us-central1", help="Google Cloud regional location"
    )
    parser_vertex.add_argument(
        "--machine-type",
        default="e2-highmem-2",
        help="Machine type for the VM to run the job",
    )
    parser_vertex.add_argument(
        "--disk-size-gb",
        type=int,
        default=100,
        help="Disk size in GB for the VM to run the job",
    )
    parser_vertex.set_defaults(run=run_vertex_job)

    args = parser.parse_args()
    kwargs = args.__dict__.copy()
    kwargs.pop("run")

    args.run(**kwargs)

דוגמת קוד של פייפליין

קוד צינור העיבוד במדריך הזה פורס צינור Dataflow שמבצע את הפעולות הבאות:

• קורא הנחיה מ-Pub/Sub ומקודד את הטקסט לטנסורים של טוקנים.
• הפעלת הטרנספורמציה RunInference.
• מפענח את טנסור האסימונים של הפלט לטקסט וכותב את התשובה ל-Pub/Sub.

קובץ ההתחלה נראה כך:

# Copyright 2023 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Runs a streaming RunInference Language Model pipeline."""

from __future__ import annotations

import logging

import apache_beam as beam
from apache_beam.ml.inference.base import PredictionResult
from apache_beam.ml.inference.base import RunInference
from apache_beam.ml.inference.pytorch_inference import make_tensor_model_fn
from apache_beam.ml.inference.pytorch_inference import PytorchModelHandlerTensor
from apache_beam.options.pipeline_options import PipelineOptions
import torch
from transformers import AutoConfig
from transformers import AutoModelForSeq2SeqLM
from transformers import AutoTokenizer
from transformers.tokenization_utils import PreTrainedTokenizer

MAX_RESPONSE_TOKENS = 256


def to_tensors(input_text: str, tokenizer: PreTrainedTokenizer) -> torch.Tensor:
    """Encodes input text into token tensors.

    Args:
        input_text: Input text for the language model.
        tokenizer: Tokenizer for the language model.

    Returns: Tokenized input tokens.
    """
    return tokenizer(input_text, return_tensors="pt").input_ids[0]


def decode_response(result: PredictionResult, tokenizer: PreTrainedTokenizer) -> str:
    """Decodes output token tensors into text.

    Args:
        result: Prediction results from the RunInference transform.
        tokenizer: Tokenizer for the language model.

    Returns: The model's response as text.
    """
    output_tokens = result.inference
    return tokenizer.decode(output_tokens, skip_special_tokens=True)


class AskModel(beam.PTransform):
    """Asks an language model a prompt message and gets its responses.

    Attributes:
        model_name: HuggingFace model name compatible with AutoModelForSeq2SeqLM.
        state_dict_path: File path to the model's state_dict, can be in Cloud Storage.
        max_response_tokens: Maximum number of tokens for the model to generate.
    """

    def __init__(
        self,
        model_name: str,
        state_dict_path: str,
        max_response_tokens: int = MAX_RESPONSE_TOKENS,
    ) -> None:
        self.model_handler = PytorchModelHandlerTensor(
            state_dict_path=state_dict_path,
            model_class=AutoModelForSeq2SeqLM.from_config,
            model_params={"config": AutoConfig.from_pretrained(model_name)},
            inference_fn=make_tensor_model_fn("generate"),
        )
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.max_response_tokens = max_response_tokens

    def expand(self, pcollection: beam.PCollection[str]) -> beam.PCollection[str]:
        return (
            pcollection
            | "To tensors" >> beam.Map(to_tensors, self.tokenizer)
            | "RunInference"
            >> RunInference(
                self.model_handler,
                inference_args={"max_new_tokens": self.max_response_tokens},
            )
            | "Get response" >> beam.Map(decode_response, self.tokenizer)
        )


if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--messages-topic",
        required=True,
        help="Pub/Sub topic for input text messages",
    )
    parser.add_argument(
        "--responses-topic",
        required=True,
        help="Pub/Sub topic for output text responses",
    )
    parser.add_argument(
        "--model-name",
        required=True,
        help="HuggingFace model name compatible with AutoModelForSeq2SeqLM",
    )
    parser.add_argument(
        "--state-dict-path",
        required=True,
        help="File path to the model's state_dict, can be in Cloud Storage",
    )
    args, beam_args = parser.parse_known_args()

    logging.getLogger().setLevel(logging.INFO)
    beam_options = PipelineOptions(
        beam_args,
        pickle_library="cloudpickle",
        streaming=True,
    )

    simple_name = args.model_name.split("/")[-1]
    pipeline = beam.Pipeline(options=beam_options)
    _ = (
        pipeline
        | "Read from Pub/Sub" >> beam.io.ReadFromPubSub(args.messages_topic)
        | "Decode bytes" >> beam.Map(lambda msg: msg.decode("utf-8"))
        | f"Ask {simple_name}" >> AskModel(args.model_name, args.state_dict_path)
        | "Encode bytes" >> beam.Map(lambda msg: msg.encode("utf-8"))
        | "Write to Pub/Sub" >> beam.io.WriteToPubSub(args.responses_topic)
    )
    pipeline.run()

טעינת המודל

מודלים גדולים של שפה (LLM) יכולים להיות גדולים מאוד. בדרך כלל, מודלים גדולים יותר שאומנו עם יותר פרמטרים נותנים תוצאות טובות יותר. עם זאת, כדי להריץ מודלים גדולים יותר נדרש מחשב גדול יותר ויותר זיכרון. יכול להיות שייקח יותר זמן להריץ מודלים גדולים יותר במעבדי CPU.

לפני שמריצים מודל PyTorch ב-Dataflow, צריך לטעון את אובייקט state_dict של המודל. אובייקט state_dict של מודל מאחסן את המשקלים של המודל.

בצינור עיבוד נתונים של Dataflow שמשתמש בטרנספורמציה RunInference של Apache Beam, צריך לטעון את אובייקט state_dict של המודל ל-Cloud Storage. למכונה שבה משתמשים כדי לטעון את אובייקט state_dict ל-Cloud Storage צריך להיות מספיק זיכרון כדי לטעון את המודל. בנוסף, המכונה צריכה חיבור מהיר לאינטרנט כדי להוריד את המשקלים ולהעלות אותם ל-Cloud Storage.

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

דגם	פרמטרים	הזיכרון הנדרש
google/flan-t5-small	‫80 מיליון	‫> 320 MB
google/flan-t5-base	‫250 מיליון	‫> 1 GB
google/flan-t5-large	‫780 מיליון	‫> 3.2GB
google/flan-t5-xl	3 מיליארד	‫> 12 GB
google/flan-t5-xxl	‫11 מיליארד	‫> 44GB
google/flan-ul2	20 מיליארד	‫> 80GB

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

מכיוון שמודלים גדולים של שפה יכולים להיות גדולים מאוד, בדוגמה שבמדריך הזה האובייקט state_dict נשמר בפורמט float16 במקום בפורמט ברירת המחדל float32. בהגדרה הזו, כל פרמטר משתמש ב-16 ביט במקום ב-32 ביט, כך שהאובייקט state_dict קטן פי שניים. גודל קטן יותר ממזער את הזמן שנדרש לטעינת המודל. עם זאת, המרת הפורמט מחייבת שהמכונה הווירטואלית תתאים גם את המודל וגם את אובייקט state_dict לזיכרון.

בטבלה הבאה מפורטות הדרישות המינימליות לטעינת מודל אחרי שאובייקט state_dict נשמר בפורמט float16. בטבלה מוצגים גם סוגי המכונות המומלצים לטעינת מודל באמצעות Agent Platform. גודל הדיסק המינימלי (וגודל הדיסק שמוגדר כברירת מחדל) ל-Agent Platform הוא 100GB, אבל יכול להיות שחלק מהמודלים ידרשו דיסק גדול יותר.

שם הדגם	הזיכרון הנדרש	סוג המכונה	זיכרון ה-VM	דיסק של מכונה וירטואלית
google/flan-t5-small	‫> 480 MB	e2-standard-4	‫16 GB	‫100 GB
google/flan-t5-base	‫> 1.5 GB	e2-standard-4	‫16 GB	‫100 GB
google/flan-t5-large	‫> 4.8 GB	e2-standard-4	‫16 GB	‫100 GB
google/flan-t5-xl	‫> 18 GB	e2-highmem-4	‫32 GB	‫100 GB
google/flan-t5-xxl	‫> 66 GB	e2-highmem-16	‫128 GB	‫100 GB
google/flan-ul2	‫> 120 GB	e2-highmem-16	‫128 GB	‫150 GB

טוענים את אובייקט state_dict של המודל ל-Cloud Storage באמצעות משימה בהתאמה אישית ב-Agent Platform:

python download_model.py vertex \
    --model-name="MODEL_NAME" \
    --state-dict-path="gs://BUCKET_NAME/run-inference/MODEL_NAME.pt" \
    --job-name="Load MODEL_NAME" \
    --project="PROJECT_ID" \
    --bucket="BUCKET_NAME" \
    --location="LOCATION" \
    --machine-type="AGENT_PLATFORM_MACHINE_TYPE" \
    --disk-size-gb="DISK_SIZE_GB"

מחליפים את מה שכתוב בשדות הבאים:

• ‫MODEL_NAME: שם המודל, למשל google/flan-t5-xl.
• ‫AGENT_PLATFORM_MACHINE_TYPE: סוג המכונה שבה רוצים להריץ את המשימה המותאמת של Agent Platform, לדוגמה e2-highmem-4.
• ‫DISK_SIZE_GB: גודל הדיסק של המכונה הווירטואלית, ב-GB. הגודל המינימלי הוא 100 GB.

בהתאם לגודל המודל, יכול להיות שיחלפו כמה דקות עד שהמודל ייטען. כדי לראות את הסטטוס, עוברים לדף משימות בהתאמה אישית ב-Agent Platform.

מעבר למשימות בהתאמה אישית

הרצת צינור עיבוד הנתונים

אחרי שמעלים את המודל, מריצים את צינור הנתונים של Dataflow. כדי להפעיל את צינור העיבוד, גם המודל וגם הזיכרון שבו משתמש כל עובד צריכים להיכנס לזיכרון.

בטבלה הבאה מוצגים סוגי המכונות המומלצים להרצת צינור (pipeline) של הסקת מסקנות.

שם הדגם	סוג המכונה	זיכרון ה-VM
google/flan-t5-small	n2-highmem-2	‫16 GB
google/flan-t5-base	n2-highmem-2	‫16 GB
google/flan-t5-large	n2-highmem-4	‫32 GB
google/flan-t5-xl	n2-highmem-4	‫32 GB
google/flan-t5-xxl	n2-highmem-8	‫64 GB
google/flan-ul2	n2-highmem-16	‫128 GB

מריצים את הפייפליין:

python main.py \
    --messages-topic="projects/PROJECT_ID/topics/PROMPTS_TOPIC_ID" \
    --responses-topic="projects/PROJECT_ID/topics/RESPONSES_TOPIC_ID" \
    --model-name="MODEL_NAME" \
    --state-dict-path="gs://BUCKET_NAME/run-inference/MODEL_NAME.pt" \
    --runner="DataflowRunner" \
    --project="PROJECT_ID" \
    --temp_location="gs://BUCKET_NAME/temp" \
    --region="REGION" \
    --machine_type="DATAFLOW_MACHINE_TYPE" \
    --requirements_file="requirements.txt" \
    --requirements_cache="skip" \
    --experiments="use_sibling_sdk_workers" \
    --experiments="no_use_multiple_sdk_containers"

מחליפים את מה שכתוב בשדות הבאים:

• ‫PROJECT_ID: מזהה הפרויקט
• ‫PROMPTS_TOPIC_ID: מזהה הנושא של ההנחיות לקלט שיישלחו למודל
• ‫RESPONSES_TOPIC_ID: מזהה הנושא של התגובות של המודל
• ‫MODEL_NAME: שם המודל, למשל google/flan-t5-xl
• ‫BUCKET_NAME: שם הקטגוריה
• ‫REGION: האזור שבו רוצים לפרוס את המשימה, למשל us-central1
• ‫DATAFLOW_MACHINE_TYPE: המכונה הווירטואלית להרצת צינור העיבוד, למשל n2-highmem-4

כדי לוודא שהמודל נטען רק פעם אחת לכל עובד ושלא נגמר הזיכרון, מגדירים את העובדים להשתמש בתהליך יחיד על ידי הגדרת האפשרות --experiments=no_use_multiple_sdk_containers של צינור העיבוד. אין צורך להגביל את מספר השרשורים כי טרנספורמציית RunInference משתפת את אותו מודל עם כמה שרשורים.

הצינור בדוגמה הזו פועל עם מעבדי CPU. ככל שהמודל גדול יותר, נדרש יותר זמן לעיבוד כל בקשה. אם אתם צריכים תשובות מהירות יותר, אתם יכולים להפעיל GPUs.

כדי לראות את הסטטוס של צינור הנתונים, עוברים לדף Jobs (משימות) ב-Dataflow.

מעבר אל Jobs

לשאול את המודל שאלה

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

1. שולחים את ההנחיה על ידי פרסום הודעה ב-Pub/Sub. משתמשים בפקודה gcloud pubsub topics publish:

   gcloud pubsub topics publish PROMPTS_TOPIC_ID \
       --message="PROMPT_TEXT"
   

   מחליפים את PROMPT_TEXT במחרוזת שמכילה את ההנחיה שרוצים לספק. מקיפים את ההנחיה במירכאות.

   אתם יכולים להשתמש בהנחיה משלכם או לנסות אחת מהדוגמאות הבאות:

   • Translate to Spanish: My name is Luka
   • Complete this sentence: Once upon a time, there was a
   • Summarize the following text: Dataflow is a Google Cloud service that provides unified stream and batch data processing at scale. Use Dataflow to create data pipelines that read from one or more sources, transform the data, and write the data to a destination.

2. כדי לקבל את התגובה, משתמשים בפקודה gcloud pubsub subscriptions pull.

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

   gcloud pubsub subscriptions pull RESPONSES_SUBSCRIPTION_ID --auto-ack
   

   מחליפים את RESPONSES_SUBSCRIPTION_ID במזהה המינוי של תשובות המודל.

הסרת המשאבים

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

מחיקת הפרויקט

זהירות: כשמוחקים פרויקט:

• התוכן של הפרויקט נמחק כולו. אם השתמשתם בפרויקט קיים כדי לתרגל את המשימות במדריך הזה, כשתמחקו אותו תימחק גם כל העבודה שביצעתם באותו פרויקט.
• מאבדים את מזהה הפרויקט בהתאמה אישית. כשיצרתם את הפרויקט, יכולתם לבחור לו מזהה פרויקט בהתאמה אישית כדי להשתמש בו בעתיד. כדי לשמור על כתובות ה-URL שמשתמשות במזהה הפרויקט, כמו appspot.com, צריך למחוק את המשאבים שנבחרו בפרויקט אבל לא את הפרויקט כולו.

אם אתם מתכננים להיעזר בכמה ארכיטקטורות, מדריכים או מדריכים למתחילים, מומלץ להשתמש שוב באותו פרויקט כדי לא לחרוג ממכסות הפרויקטים.

כדי למחוק פרויקט Google Cloud :

gcloud projects delete PROJECT_ID

מחיקת משאבים בודדים

1. יוצאים מהסביבה הווירטואלית של Python:

   deactivate

2. עוצרים את הפייפליין:

   1. מציגים את רשימת מזהי המשימות של משימות Dataflow שפועלות, ורושמים את מזהה המשימה של המדריך:

      gcloud dataflow jobs list --region=REGION --status=active

   2. כדי לבטל את המשימה:

      gcloud dataflow jobs cancel JOB_ID --region=REGION

3. מחיקת הקטגוריה וכל מה שבתוכה:

   gcloud storage rm gs://BUCKET_NAME --recursive

4. מחיקת הנושאים והמינוי:

   gcloud pubsub topics delete PROMPTS_TOPIC_ID
   gcloud pubsub topics delete RESPONSES_TOPIC_ID
   gcloud pubsub subscriptions delete RESPONSES_SUBSCRIPTION_ID

5. מבטלים את התפקידים שהקציתם לחשבון השירות של Compute Engine שמוגדר כברירת מחדל. מריצים את הפקודה הבאה לכל אחד מהתפקידים הבאים ב-IAM:

   • roles/dataflow.admin
   • roles/dataflow.worker
   • roles/storage.admin
   • roles/pubsub.editor
   • roles/aiplatform.user

   gcloud projects remove-iam-policy-binding PROJECT_ID --member=serviceAccount:[e-mail] --role=SERVICE_ACCOUNT_ROLE

6. אופציונלי: ביטול תפקידים בחשבון Google.

   gcloud projects remove-iam-policy-binding PROJECT_ID --member="user:EMAIL_ADDRESS" --role=roles/iam.serviceAccountUser

7. אם תרצו, תוכלו לבטל את פרטי הכניסה שיצרתם ולמחוק את הקובץ המקומי של פרטי הכניסה.

   gcloud auth application-default revoke

8. אם רוצים, מבטלים את פרטי הכניסה של ה-CLI של gcloud.

   gcloud auth revoke

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

• מידע נוסף על Dataflow ML
• מידע נוסף על RunInference API
• במאמרים בנושא צינורות עיבוד נתונים של AI/ML ב-Apache Beam יש מידע מפורט על שימוש בלמידת מכונה עם Apache Beam.
• עובדים עם מחברת שימוש ב-RunInference ל-AI גנרטיבי.
• כדאי להעמיק את הקריאה ולהכיר דוגמאות לארכיטקטורות, תרשימים ושיטות מומלצות בנושאי Google Cloud. כל אלה זמינים במרכז הארכיטקטורה של Cloud.