Vertex AI Pipelines for MLOps on GCP

A model in a notebook is a photograph. It records how the world looked on the day it was trained, and the world moves on. Keeping a model useful means retraining it, and retraining by hand means that one day a step is skipped, a different file is used, or nobody remembers which model is in production.

Vertex AI Pipelines is Google Cloud's answer. The platform has been renamed Gemini Enterprise Agent Platform, formerly Vertex AI (Google Cloud: Gemini Enterprise Agent Platform), and the documentation now calls this service Agent Platform Pipelines. The idea is unchanged: describe the workflow once, as code, and let the platform run it.

What a pipeline is

A pipeline is a graph of steps. Each step runs in its own container, takes inputs and produces outputs, and the outputs of one step are the inputs of another (Google Cloud: introduction to Agent Platform Pipelines).

You do not draw the graph. You write which outputs go to which inputs, and the order follows. Tasks with no dependency between them run in parallel.

The service is serverless: there is no cluster to keep running. Pipelines are written with the Kubeflow Pipelines SDK, which Google recommends for most uses, or with TFX (Google Cloud: build a pipeline).

Why not a script?

A script that prepares data, trains and evaluates works. A pipeline adds four things:

• Each step is isolated, with its own container and dependencies. A failure in one can be retried without redoing the others.
• Every run is recorded. The parameters, the artifacts and what produced them are stored, so "which data trained the model in production?" has an answer.
• The definition is a file. It can be reviewed, versioned and scheduled.
• A decision can be a step. Whether a new model replaces the old one becomes a rule in code, not an opinion on the day.

A retraining pipeline

The script defines a pipeline of four components with the Kubeflow Pipelines SDK: prepare a day of data, train a model, evaluate it on a later day, and decide whether to promote it. The problem is the one from the platform tutorial: predicting the duration of New York taxi trips (NYC TLC: trip record data).

It then does two things. It compiles the pipeline with the SDK and reads the task graph back out of the compiled file. And it runs the same component functions three times, as three retraining runs would on three mornings.

The compilation is real. The runs call the component functions directly on a local machine. Nothing was submitted to the platform.

import warnings

import yaml
from kfp import compiler, dsl

warnings.filterwarnings("ignore", category=FutureWarning)   # the SDK's notice about its default image
PACKAGES = ["pandas", "pyarrow", "scikit-learn", "joblib"]


@dsl.component(base_image="python:3.11", packages_to_install=PACKAGES)
def prepare(day: str, source: str, data: dsl.Output[dsl.Dataset]):
    import pandas as pd
    trips = pd.read_parquet(source, columns=["tpep_pickup_datetime", "tpep_dropoff_datetime", "trip_distance"])
    trips = trips[trips.tpep_pickup_datetime.dt.strftime("%Y-%m-%d") == day]
    trips["minutes"] = (trips.tpep_dropoff_datetime - trips.tpep_pickup_datetime).dt.total_seconds() / 60
    trips["hour"] = trips.tpep_pickup_datetime.dt.hour
    trips = trips[trips.trip_distance.between(0.1, 50) & trips.minutes.between(1, 180)]
    trips[["trip_distance", "hour", "minutes"]].to_parquet(data.path)


@dsl.component(base_image="python:3.11", packages_to_install=PACKAGES)
def train(data: dsl.Input[dsl.Dataset], model: dsl.Output[dsl.Model]):
    import joblib
    import pandas as pd
    from sklearn.ensemble import HistGradientBoostingRegressor
    trips = pd.read_parquet(data.path)
    fitted = HistGradientBoostingRegressor(random_state=0).fit(trips[["trip_distance", "hour"]], trips.minutes)
    joblib.dump(fitted, model.path)


@dsl.component(base_image="python:3.11", packages_to_install=PACKAGES)
def evaluate(model: dsl.Input[dsl.Model], data: dsl.Input[dsl.Dataset]) -> float:
    import joblib
    import pandas as pd
    from sklearn.metrics import mean_absolute_error
    trips = pd.read_parquet(data.path)
    predicted = joblib.load(model.path).predict(trips[["trip_distance", "hour"]])
    return round(float(mean_absolute_error(trips.minutes, predicted)), 2)


@dsl.component(base_image="python:3.11")
def gate(candidate_error: float, champion_error: float) -> str:
    return "promote" if candidate_error < champion_error else "hold"


@dsl.pipeline(name="trip-duration-retrain")
def retrain(train_day: str, test_day: str, champion_error: float, source: str) -> str:
    train_data = prepare(day=train_day, source=source)
    test_data = prepare(day=test_day, source=source)
    model = train(data=train_data.outputs["data"])
    error = evaluate(model=model.outputs["model"], data=test_data.outputs["data"])
    return gate(candidate_error=error.output, champion_error=champion_error).output


compiler.Compiler().compile(retrain, "retrain.yaml")
tasks = yaml.safe_load(open("retrain.yaml"))["root"]["dag"]["tasks"]
print(f"compiled retrain.yaml: {len(tasks)} tasks")
for name, task in tasks.items():
    print(f"  {name:10} runs after: {', '.join(task.get('dependentTasks', [])) or 'nothing'}")


def run_locally(train_day, test_day, champion_error):
    """Call the same component functions in the order of the compiled graph."""
    source = "../../dedata/yellow_tripdata_2024-01.parquet"
    train_set, test_set = dsl.Dataset(uri="train.parquet"), dsl.Dataset(uri="test.parquet")
    model = dsl.Model(uri="model.joblib")
    prepare.python_func(day=train_day, source=source, data=train_set)
    prepare.python_func(day=test_day, source=source, data=test_set)
    train.python_func(data=train_set, model=model)
    error = evaluate.python_func(model=model, data=test_set)
    return error, gate.python_func(candidate_error=error, champion_error=champion_error)


print()
print("run  trained on   tested on    error, minutes   champion's error   decision")
champion = float("inf")
for label, train_day in (("A", "2024-01-15"), ("B", "2024-01-16"), ("C", "2024-01-14")):
    error, decision = run_locally(train_day, "2024-01-17", champion)
    shown = "none yet" if champion == float("inf") else f"{champion:.2f}"
    print(f"{label}    {train_day}   2024-01-17   {error:>14.2f}   {shown:>16}   {decision}")
    champion = error if decision == "promote" else champion

compiled retrain.yaml: 5 tasks
  evaluate   runs after: prepare-2, train
  gate       runs after: evaluate
  prepare    runs after: nothing
  prepare-2  runs after: nothing
  train      runs after: prepare

run  trained on   tested on    error, minutes   champion's error   decision
A    2024-01-15   2024-01-17             4.39           none yet   promote
B    2024-01-16   2024-01-17             4.22               4.39   promote
C    2024-01-14   2024-01-17             4.55               4.22   hold

• The graph was derived, not drawn. The pipeline function never says that training comes after preparation. The compiler worked that out from which output feeds which input, and found that the two prepare tasks depend on nothing and can run together.
• Run A has nothing to beat. The first model is promoted with an error of 4.39 minutes.
• Run B is promoted on merit. Trained on the 16th, it scores 4.22 on the same test day and replaces the champion.
• Run C is held. Trained on the 14th, a Sunday, it scores 4.55. That is worse than the model already in place, so the pipeline keeps the old one.

Run C is the reason the gate exists. A newer training run is not automatically a better model. Without the comparison, a scheduled retraining would have swapped a good model for a worse one, and nobody would have been told.

Running it on the platform

The compiled YAML file is everything the service needs. Submitting it is a few lines, in the form Google documents:

import google.cloud.aiplatform as aip

aip.init(project="analytics-prod", location="us-central1")

job = aip.PipelineJob(
    display_name="trip-duration-retrain",
    template_path="retrain.yaml",
    pipeline_root="gs://analytics-prod-ml/pipelines",
    parameter_values={
        "train_day": "2024-01-16",
        "test_day": "2024-01-17",
        "champion_error": 4.39,
        "source": "gs://analytics-prod-raw/trips/yellow_tripdata_2024-01.parquet",
    },
)
job.submit()

The pipeline root is a Cloud Storage location where the artifacts of each run are kept. On the platform, each component runs in the container named in its definition and installs the packages it lists, which is why the components above import their libraries inside the function.

Three things change between the local run and a real one:

• The gate should act. Here it returns a word. In production, a conditional branch of the pipeline registers and deploys the model only when the answer is promote. Google Cloud Pipeline Components provide ready-made steps for those platform actions.
• The champion's error should be measured, not passed in. Evaluate the current production model on the same test data inside the pipeline.
• The run should be scheduled. A pipeline can be run once or on a recurring schedule.

Lineage and experiments

Each run produces a lineage graph: the parameters and artifacts that went in, were created and came out, stored in the platform's ML Metadata. When a model misbehaves, lineage answers which run produced it, from which data, with which settings.

Runs can also be attached to experiments, so that a change of feature or setting can be compared across runs instead of remembered. These are parts of a wider MLOps toolset that also covers model versions and monitoring (Google Cloud: MLOps on Agent Platform).

What it costs

At list prices read on 2 October 2026, a pipeline run has an execution fee of USD 0.03. The real cost is the compute each component uses, charged like training, and any other service the pipeline calls (Google Cloud: Agent Platform pricing). A daily retraining of a small model costs little. A pipeline that starts large machines for every step, every hour, does not. See cost controls and budgets.

When a pipeline is too much

Not every model needs one. If a model is retrained twice a year by one person, a documented script is enough. If the model is a standard one on data in BigQuery, a scheduled query that retrains it may do. See BigQuery ML.

A pipeline pays for itself when retraining is frequent, when more than one person touches the model, or when someone will one day ask you to prove how a prediction came to be made.

Building one, in order

1. Get the training script working on its own first.
2. Split it into functions with clear inputs and outputs: prepare, train, evaluate.
3. Turn each into a component, and compile.
4. Add the gate. Decide the rule before you see the numbers.
5. Run it by hand a few times and read the lineage.
6. Schedule it, and alert on failure and on a held model.

For the wider practice, see monitoring ML models in production.

Common mistakes

• Automating retraining with no gate.
• Evaluating on data the model was trained on.
• One giant component. It cannot be retried in parts, and its lineage says nothing.
• Passing large data as parameters instead of as artifacts.
• Different library versions in different components, so the model trains with one and is evaluated with another.
• Scheduling before anyone has watched a run fail.

Model deployment and monitoring are covered in the MLOps course.

Sources

• Introduction to Agent Platform Pipelines (Google Cloud Documentation) — Google Cloud
• Build a pipeline (Google Cloud Documentation) — Google Cloud
• MLOps on Gemini Enterprise Agent Platform (Google Cloud Documentation) — Google Cloud
• Gemini Enterprise Agent Platform pricing (Google Cloud) — Google Cloud
• Gemini Enterprise Agent Platform (formerly Vertex AI) (Google Cloud) — Google Cloud
• TLC Trip Record Data (NYC Taxi and Limousine Commission) — NYC Taxi and Limousine Commission