Google Cloud Vertex AI 生产级实战：从模型训练到在线推理与MLOps流水线

Google Cloud Vertex AI 是 Google Cloud 统一的机器学习平台，整合了此前分散的 AI Platform、AutoML、BigQuery ML 等服务，为数据科学家和 ML 工程师提供了一站式的模型训练、部署、监控和管理能力。本文将从实战角度出发，带你完成从数据准备、模型训练、超参调优、在线部署到 MLOps 流水线搭建的完整流程，所有代码均可在生产环境中直接使用。

一、Vertex AI 核心架构与关键概念

在动手之前，先理解 Vertex AI 的核心组件及其关系，这决定了你的工作流设计。

1.1 核心服务矩阵

1.2 项目初始化

所有 Vertex AI 操作都需要先初始化客户端：

1
2
3
4
5
6
7
from google.cloud import aiplatform



PROJECT_ID = "your-project-id"

REGION = "us-central1"

BUCKET_URI = f"gs://{PROJECT_ID}-vertex-ai-artifacts"



aiplatform.init(project=PROJECT_ID, location=REGION, staging_bucket=BUCKET_URI)

确保你的服务账号拥有

权限，且 Cloud Storage bucket 已创建。

二、自定义训练：从本地脚本到云端分布式

Vertex AI Training 的核心思想是：你只需提供训练代码和依赖声明，Google 负责调度计算资源、管理分布式训练、输出模型产物。这比自建 GPU 集群效率高得多。

与传统自建 GPU 集群相比，Vertex AI Training 的优势体现在三个方面：第一，无需管理底层虚拟机和驱动环境，容器镜像预装了 CUDA、cuDNN 等依赖，开箱即用；第二，训练任务按需计费，完成即停，不存在集群闲置浪费；第三，与 GCP 生态无缝集成，训练数据可以直接从 BigQuery 或 Cloud Storage 读取，模型产物自动注册到 Model Registry。

2.1 训练代码结构

推荐的训练代码目录结构：

1
2
3
4
5
6
trainer/

├── __init__.py

├── task.py          # 入口文件

├── model.py         # 模型定义

├── dataset.py       # 数据加载

└── utils.py         # 工具函数

task.py 是入口文件，必须解析

等环境变量来适配 Vertex AI 的约定：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
import argparse

import os

import tensorflow as tf

from trainer.model import build_model



def get_args():

    parser = argparse.ArgumentParser()

    parser.add_argument('--epochs', type=int, default=50)

    parser.add_argument('--batch-size', type=int, default=64)

    parser.add_argument('--learning-rate', type=float, default=0.001)

    # Vertex AI 自动注入 AIP_MODEL_DIR

    parser.add_argument('--model-dir', type=str,

                        default=os.environ.get('AIP_MODEL_DIR', './model'))

    return parser.parse_args()



def main():

    args = get_args()

    strategy = tf.distribute.MirroredStrategy()

    with strategy.scope():

        model = build_model(learning_rate=args.learning_rate)



    # 加载训练数据（从 GCS）

    train_ds, val_ds = load_datasets(args.batch_size)



    model.fit(

        train_ds,

        validation_data=val_ds,

        epochs=args.epochs,

        callbacks=[

            tf.keras.callbacks.ModelCheckpoint(

                os.path.join(args.model_dir, 'best_model'),

                save_best_only=True

            ),

            tf.keras.callbacks.EarlyStopping(patience=5)

        ]

    )



if __name__ == '__main__':

    main()

2.2 提交自定义训练任务

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
job = aiplatform.CustomTrainingJob(

    display_name="text-classifier-v1",

    script_path="trainer/task.py",

    container_uri="us-docker.pkg.dev/vertex-ai/training/tf-gpu.2-15:latest",

    requirements=["scikit-learn==1.3.0", "pandas==2.1.0"],

    model_serving_container_image_uri=

        "us-docker.pkg.dev/vertex-ai/prediction/tf2-gpu.2-15:latest",

)



model = job.run(

    dataset=dataset_resource,  # 可选：关联 Vertex AI Dataset

    model_display_name="text-classifier-v1",

    base_output_dir=f"{BUCKET_URI}/models/text-classifier",

    args=["--epochs=50", "--batch-size=128"],

    replica_count=1,

    machine_type="n1-standard-4",

    accelerator_type="NVIDIA_TESLA_T4",

    accelerator_count=1,

    sync=True,  # 阻塞等待训练完成

)



print(f"Model resource: {model.resource_name}")

关键参数说明：

• 1	replica_count

  ：Worker 数量，大于1时启用分布式训练

• 1	accelerator_type

  ：GPU 型号，可选 T4/V100/A100/L4

• 1

  sync=True

  ：阻塞模式，适合 Notebook 交互式开发

• 1	base_output_dir

  ：模型和 checkpoint 自动写入此 GCS 路径

2.3 分布式训练配置

对于大模型，需要多机多卡训练。Vertex AI 支持 chief-worker 模式：

1
2
3
4
5
6
7
8
9
# 多机分布式训练

job.run(

    replica_count=4,  # 1 chief + 3 workers

    machine_type="n1-standard-8",

    accelerator_type="NVIDIA_TESLA_A100",

    accelerator_count=2,  # 每台机器2块A100

    distribution="chief",  # 使用 chief-worker 分布策略

    # 总计: 4 * 2 = 8 块 A100

)

训练代码中需使用

，Vertex AI 会自动配置

环境变量。

三、超参调优：用 Vizier 自动搜索最优配置

手动调参既耗时又低效。Vertex AI 集成了 Google 内部的 Vizier 服务，支持贝叶斯优化、网格搜索、随机搜索等策略。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
from google.cloud.aiplatform import hyperparameter_tuning as hpt



hpt_job = aiplatform.HyperparameterTuningJob(

    display_name="text-clf-hpt",

    custom_job_spec={

        "display_name": "text-clf-hpt-trial",

        "job_spec": {

            "worker_pool_specs": [{

                "machine_spec": {

                    "machine_type": "n1-standard-4",

                    "accelerator_type": "NVIDIA_TESLA_T4",

                    "accelerator_count": 1,

                },

                "replica_count": 1,

                "container_spec": {

                    "image_uri": "us-docker.pkg.dev/vertex-ai/training/tf-gpu.2-15:latest",

                    "command": ["python", "trainer/task.py"],

                },

            }],

        },

    },

    metric_spec={

        "val_accuracy": "maximize",

    },

    parameter_spec={

        "learning-rate": hpt.DoubleParameterSpec(

            min=1e-5, max=1e-2, scale="log"),

        "batch-size": hpt.IntegerParameterSpec(

            min=32, max=256, scale="linear"),

        "dropout-rate": hpt.DoubleParameterSpec(

            min=0.1, max=0.5, scale="linear"),

        "num-layers": hpt.CategoricalParameterSpec(

            values=[2, 3, 4, 6]),

    },

    max_trial_count=30,

    parallel_trial_count=5,

    algorithm="bayesian_optimization",

)



hpt_job.run(sync=True)



# 获取最优 Trial

best_trial = max(hpt_job.trials, key=lambda t: t.final_measurement.metrics[0].value)

print(f"Best trial: {best_trial.id}")

print(f"Best val_accuracy: {best_trial.final_measurement.metrics[0].value:.4f}")

print(f"Best params: {best_trial.parameters}")

调优策略建议：

• 初始阶段用

  1	random_search

  广泛探索参数空间

• 缩小范围后切换

  1	bayesian_optimization

  精细搜索

• 设置合理的

  1	max_trial_count

  ，避免无限消耗预算

• 利用

  1	parallel_trial_count

  加速，但注意 quota 限制

四、模型部署：在线推理端点与流量管理

4.1 创建端点并部署模型

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# 创建端点

endpoint = aiplatform.Endpoint.create(

    display_name="text-classifier-endpoint",

    location=REGION,

)



# 部署模型到端点

model.deploy(

    endpoint=endpoint,

    deployed_model_display_name="text-clf-v1",

    machine_type="n1-standard-4",

    accelerator_type="NVIDIA_TESLA_T4",

    accelerator_count=1,

    min_replica_count=1,

    max_replica_count=5,  # 自动扩缩容

    traffic_percentage=100,

    traffic_split=None,

)



print(f"Endpoint: {endpoint.resource_name}")

4.2 在线预测调用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# 单条预测

response = endpoint.predict(instances=[

    {"text": "This product is amazing!", "max_length": 128}

])



for prediction in response.predictions:

    print(f"Predicted class: {prediction['classes'][0]}")

    print(f"Confidence: {prediction['scores'][0]:.4f}")



# 批量预测（适合离线评分）

batch_prediction_job = model.batch_predict(

    job_display_name="text-clf-batch-predict",

    gcs_source=f"{BUCKET_URI}/data/predict_input.jsonl",

    gcs_destination_prefix=f"{BUCKET_URI}/predictions/output",

    machine_type="n1-standard-4",

    starting_replica_count=1,

    max_replica_count=10,

)

4.3 灰度发布与流量切分

生产环境中，新模型上线通常需要灰度发布。Vertex AI 支持在同一端点上部署多个模型版本并按比例切分流量：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# 部署新版本模型，初始流量为 0

model_v2.deploy(

    endpoint=endpoint,

    deployed_model_display_name="text-clf-v2",

    machine_type="n1-standard-4",

    min_replica_count=1,

    max_replica_count=5,

    traffic_percentage=0,  # 初始不分流量

)



# 灰度切分：v1 承担 90% 流量，v2 承担 10%

endpoint.update(

    traffic_split={

        model_v1_deployed_id: 90,

        model_v2_deployed_id: 10,

    }

)



# 监控 v2 指标正常后，逐步提升到 100%

endpoint.update(

    traffic_split={

        model_v2_deployed_id: 100,

    }

)

五、MLOps 流水线：用 Vertex AI Pipelines 实现自动化

将训练、评估、部署串联成自动化流水线，是 ML 工程化的核心。Vertex AI Pipelines 基于 Kubeflow Pipelines v2，用 Python 声明式定义 DAG。

5.1 定义 Pipeline 组件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
from google.cloud.aiplatform import pipeline

from kfp import dsl

from kfp.dsl import component, Output, Model, Metrics, Input



@component(base_image="python:3.11",

           packages_to_install=["scikit-learn", "pandas", "google-cloud-aiplatform"])

def train_model(

    project: str,

    location: str,

    data_gcs_path: str,

    learning_rate: float,

    epochs: int,

    model: Output[Model],

    metrics: Output[Metrics],

):

    import pandas as pd

    from sklearn.model_selection import train_test_split

    from sklearn.ensemble import GradientBoostingClassifier

    from sklearn.metrics import accuracy_score, f1_score

    import joblib



    df = pd.read_csv(data_gcs_path)

    X = df.drop(columns=["target"])

    y = df["target"]

    X_train, X_test, y_train, y_test = train_test_split(

        X, y, test_size=0.2, random_state=42

    )



    clf = GradientBoostingClassifier(

        learning_rate=learning_rate,

        n_estimators=epochs,

        random_state=42

    )

    clf.fit(X_train, y_train)



    y_pred = clf.predict(X_test)

    acc = accuracy_score(y_test, y_pred)

    f1 = f1_score(y_test, y_pred, average="weighted")



    metrics.log_metric("accuracy", acc)

    metrics.log_metric("f1_score", f1)



    joblib.dump(clf, model.path + "/model.joblib")



@component(base_image="python:3.11",

           packages_to_install=["google-cloud-aiplatform"])

def deploy_model(

    project: str,

    location: str,

    model: Input[Model],

    endpoint_display_name: str,

    accuracy: float,

    accuracy_threshold: float,

) -> str:

    from google.cloud import aiplatform

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



    if accuracy < accuracy_threshold:

        return f"Skipped deployment: accuracy {accuracy:.4f} < {accuracy_threshold}"



    uploaded_model = aiplatform.Model.upload(

        display_name="pipeline-model",

        artifact_uri=model.uri.rsplit("/", 1)[0],

        serving_container_image_uri=

            "us-docker.pkg.dev/vertex-ai/prediction/sklearn-cpu.1-3:latest",

    )



    endpoints = aiplatform.Endpoint.list(

        filter=f"display_name={endpoint_display_name}"

    )

    if endpoints:

        endpoint = endpoints[0]

    else:

        endpoint = aiplatform.Endpoint.create(

            display_name=endpoint_display_name

        )



    uploaded_model.deploy(

        endpoint=endpoint,

        machine_type="n1-standard-2",

        min_replica_count=1,

        max_replica_count=3,

    )



    return f"Deployed to {endpoint.resource_name}"

5.2 编排 Pipeline

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
@dsl.pipeline(

    name="ml-training-pipeline",

    description="End-to-end ML training and deployment pipeline",

)

def ml_pipeline(

    project: str,

    location: str,

    data_gcs_path: str,

    learning_rate: float = 0.1,

    epochs: int = 100,

    accuracy_threshold: float = 0.85,

):

    train_task = train_model(

        project=project,

        location=location,

        data_gcs_path=data_gcs_path,

        learning_rate=learning_rate,

        epochs=epochs,

    )



    deploy_task = deploy_model(

        project=project,

        location=location,

        model=train_task.outputs["model"],

        endpoint_display_name="production-endpoint",

        accuracy=train_task.outputs["metrics"].metrics["accuracy"],

        accuracy_threshold=accuracy_threshold,

    )



    deploy_task.after(train_task)

5.3 编译与执行

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from kfp.compiler import Compiler



Compiler().compile(

    pipeline_func=ml_pipeline,

    package_path="ml_pipeline.yaml"

)



job = aiplatform.PipelineJob(

    display_name="ml-training-pipeline-run",

    template_path="ml_pipeline.yaml",

    parameter_values={

        "project": PROJECT_ID,

        "location": REGION,

        "data_gcs_path": f"{BUCKET_URI}/data/training_data.csv",

        "learning_rate": 0.05,

        "epochs": 200,

        "accuracy_threshold": 0.90,

    },

    enable_caching=True,

)



job.run(sync=True)

六、模型监控：检测数据漂移与预测偏移

模型上线后最怕的是”静默失效”——输入数据分布漂移导致预测质量下降，但线上无感知。Vertex AI Model Monitoring 可以自动检测这种情况。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
# 在已部署的端点上启用模型监控

monitoring_job = aiplatform.ModelDeploymentMonitoringJob.create(

    display_name="text-clf-monitor",

    endpoint=endpoint,

    feature_stats_request={

        "features": {

            "text_length": {},

            "category": {},

        },

    },

    objective_config={

        "training_dataset": {

            "data_format": "csv",

            "gcs_source": {"uris": [f"{BUCKET_URI}/data/training_data.csv"]},

        },

        "sampling_config": {

            "sample_rate": 0.8,

            "min_sample_size": 1000,

        },

        "drift_thresholds": {

            "text_length": {"value": 0.3},

            "category": {"value": 0.2},

        },

    },

    notification_config={

        "email_config": {

            "user_emails": ["ml-team@yourcompany.com"],

        },

    },

    monitor_interval=3600,

)



print(f"Monitoring job: {monitoring_job.resource_name}")

当检测到漂移超过阈值时，系统会自动发送邮件通知，你也可以在 Console 的 Model Monitoring 页面查看详细的漂移分析报告。

除了特征漂移检测外，Vertex AI Model Monitoring 还支持预测偏移（Prediction Drift）检测，即监控模型输出分布的变化。当输入特征看似正常但模型预测结果出现异常偏移时（例如某分类器的正类预测率突然从 15% 跳升到 40%），这通常意味着模型需要重新训练。建议同时启用特征漂移和预测偏移两种检测，形成双重保障机制。监控数据可以导出到 BigQuery 做进一步分析，也可以接入 Cloud Monitoring 构建自定义仪表盘。

七、成本优化与生产实践建议

7.1 计算资源选型

7.2 关键成本优化策略

• Spot VM 训练：对可中断的训练任务使用 Spot 实例，成本降低 60-80%。在 Pipeline 中设置自动重试机制。
• 自动扩缩容：推理端点设置 min_replica_count=0（允许缩容到0）配合 max_replica_count 上限，空闲时零成本。
• Pipeline 缓存：启用 enable_caching=True，未变更的组件跳过执行。
• 预算告警：在 Cloud Billing 中设置预算告警，避免超支。
• Preemptible GPU：对于不需要长时间运行的实验，使用抢占式 GPU 实例。

1
2
3
4
5
6
7
8
# 使用 Spot 实例训练

job.run(

    replica_count=1,

    machine_type="n1-standard-8",

    accelerator_type="NVIDIA_TESLA_V100",

    accelerator_count=1,

    scheduling={"spot": True},

)

7.3 生产环境 Checklist

• 训练代码已单元测试并版本控制
• 模型已在 Model Registry 中注册并打标签
• 端点已配置自动扩缩容和健康检查
• 已启用 Model Monitoring 并配置告警
• Pipeline 已编译并通过端到端测试
• 已配置 IAM 权限，最小权限原则
• 推理延迟和吞吐量已通过压测验证
• 灰度发布流程已文档化
• 回滚方案已验证

八、总结

Vertex AI 将 Google 内部大规模 ML 工程的最佳实践以托管服务的形式开放，让团队能够专注于模型和数据本身，而非基础设施运维。从自定义训练、超参调优、模型部署到 MLOps 流水线，Vertex AI 提供了完整的工具链。生产环境中的关键在于：用 Pipeline 实现自动化、用 Model Monitoring 保障服务质量、用灰度发布降低上线风险、用 Spot 实例和自动扩缩容控制成本。

建议从一个小型 Pipeline 开始，逐步将手动操作自动化，最终实现从数据到部署的全链路闭环。Vertex AI 的优势在于它与 GCP 生态（BigQuery、Cloud Storage、Pub/Sub）的深度集成，如果你的数据和工作负载已在 GCP 上，选择 Vertex AI 是最自然的技术路线。

最后提醒一点：Vertex AI 的定价模式是按使用量计费，训练按 GPU 小时收费，推理按节点小时收费，Pipelines 按步骤执行次数收费。在项目初期，建议充分利用免费额度和预算告警功能，避免因为实验迭代导致意外高额账单。同时关注 GCP 的 Committed Use Discounts，如果确定了长期工作负载，承诺使用折扣可以节省 20-40% 的计算成本。