如何将Vertex AI自定义预测例程集成到Pipeline实现批量预测？

问题核心 #

报错的根源是类型不匹配：ModelBatchPredictOp要求输入的model参数为google.VertexModel@0.0.1类型，但你自定义的导入组件输出的是通用的system.Model@0.0.1类型，两者无法兼容。

解决方案 #

无需自定义导入组件，直接使用KFP官方的importer组件导入已有的Vertex AI模型，并指定正确的artifact类型即可。

具体实现步骤 #

1. 导入KFP的importer组件以及Google Cloud Pipeline Components提供的VertexModel类型。
2. 配置importer参数，传入Vertex AI模型的完整资源路径，并指定artifact类型为VertexModel。
3. 将importer的输出直接传入ModelBatchPredictOp的model参数。

完整代码示例 #

from helper import data_component
from kfp.dsl import pipeline, importer
from google_cloud_pipeline_components.v1.batch_predict_job import ModelBatchPredictOp
from google_cloud_pipeline_components.types import artifact_types


@pipeline(name="prediction-pipeline")
def pipeline():
    data_task = data_component()
    
    # 导入已在Vertex AI上创建的自定义预测模型
    model_importer = importer(
        artifact_uri="projects/你的项目ID/locations/你的区域/models/你的模型ID",
        artifact_type=artifact_types.VertexModel,
        metadata={"resourceName": "projects/你的项目ID/locations/你的区域/models/你的模型ID"}
    )

    batch_predict_op = ModelBatchPredictOp(
        job_display_name="batch_predict_job",
        model=model_importer.output,
        gcs_source_uris=data_task.outputs["dataset"],
        gcs_destination_output_uri_prefix="gs://你的存储桶路径",
        instances_format="csv",
        predictions_format="jsonl",
        starting_replica_count=1,
        max_replica_count=1,
    )

关键细节说明 #

• 模型资源路径：必须是完整的Vertex AI模型ID，格式为projects/{项目ID}/locations/{区域}/models/{模型ID}，可在Vertex AI控制台的模型详情页获取。
• artifact_type指定：通过artifact_types.VertexModel明确类型，确保与ModelBatchPredictOp的输入要求完全匹配。
• metadata参数：resourceName必填，用于告知Pipeline该artifact对应的Vertex AI模型实体路径。

额外注意事项 #

• 运行Pipeline的服务账号需拥有访问目标Vertex AI模型的权限（至少包含aiplatform.models.get权限）。
• 确认gcs_destination_output_uri_prefix是有效的GCS路径，且服务账号具备写入该路径的权限。

内容的提问来源于stack exchange，提问作者bloukanov