我正在努力正确设置顶点AI管道，它执行以下操作：
from google_cloud_pipeline_components import aiplatform as gcc_aip
from kfp.v2 import dsl
from kfp.v2.dsl import component
from kfp.v2.dsl import (
Output,
Artifact,
Model,
)
PROJECT_ID = 'my-gcp-project'
BUCKET_NAME = "mybucket"
PIPELINE_ROOT = "{}/pipeline_root".format(BUCKET_NAME)
@component
def get_input_data() -> str:
# getting data from API, save to Cloud Storage
# return GS URI
gcs_batch_input_path = 'gs://somebucket/file'
return gcs_batch_input_path
@component(
base_image="python:3.9",
packages_to_install=['google-cloud-aiplatform==1.8.0']
)
def load_ml_model(project_id: str, model: Output[Artifact]):
"""Load existing Vertex model"""
import google.cloud.aiplatform as aip
model_id = '1234'
model = aip.Model(model_name=model_id, project=project_id, location='us-central1')
@dsl.pipeline(
name="batch-pipeline", pipeline_root=PIPELINE_ROOT,
)
def pipeline(gcp_project: str):
input_data = get_input_data()
ml_model = load_ml_model(gcp_project)
gcc_aip.ModelBatchPredictOp(
project=PROJECT_ID,
job_display_name=f'test-prediction',
model=ml_model.output,
gcs_source_uris=[input_data.output], # this doesn't work
# gcs_source_uris=['gs://mybucket/output/'], # hardcoded gs uri works
gcs_destination_output_uri_prefix=f'gs://{PIPELINE_ROOT}/prediction_output/'
)
if __name__ == '__main__':
from kfp.v2 import compiler
import google.cloud.aiplatform as aip
pipeline_export_filepath = 'test-pipeline.json'
compiler.Compiler().compile(pipeline_func=pipeline,
package_path=pipeline_export_filepath)
# pipeline_params = {
# 'gcp_project': PROJECT_ID,
# }
# job = aip.PipelineJob(
# display_name='test-pipeline',
# template_path=pipeline_export_filepath,
# pipeline_root=f'gs://{PIPELINE_ROOT}',
# project=PROJECT_ID,
# parameter_values=pipeline_params,
# )
# job.run()
当运行管道时，它在运行Batch预测时抛出此异常：详细信息="发现的错误列表： 1.字段：batch_prediction_job.model；消息：无效的模型资源名称。所以我不确定会出什么问题。我试图在笔记本中加载模型（在组件之外），它正确返回。
我遇到的第二个问题是引用GCSURI作为组件的输出到批处理作业输入。
input_data = get_input_data2()
gcc_aip.ModelBatchPredictOp(
project=PROJECT_ID,
job_display_name=f'test-prediction',
model=ml_model.output,
gcs_source_uris=[input_data.output], # this doesn't work
# gcs_source_uris=['gs://mybucket/output/'], # hardcoded gs uri works
gcs_destination_output_uri_prefix=f'gs://{PIPELINE_ROOT}/prediction_output/'
)
在编译过程中，我得到以下异常TypeError： PipelineParam类型的对象不是JSON可序列化的，尽管我认为这可能是ModelBatchPredicOp组件的问题。
再次感谢任何帮助/建议，我从昨天开始处理这个问题，所以也许我错过了一些明显的东西。
我正在使用的库：
google-cloud-aiplatform==1.8.0
google-cloud-pipeline-components==0.2.0
kfp==1.8.10
kfp-pipeline-spec==0.1.13
kfp-server-api==1.7.1
UPDATE经过评论、一些研究和调整，对于引用模型，这是可行的：
@component
def load_ml_model(project_id: str, model: Output[Artifact]):
region = 'us-central1'
model_id = '1234'
model_uid = f'projects/{project_id}/locations/{region}/models/{model_id}'
model.uri = model_uid
model.metadata['resourceName'] = model_uid
然后我可以按预期使用它：
batch_predict_op = gcc_aip.ModelBatchPredictOp(
project=gcp_project,
job_display_name=f'batch-prediction-test',
model=ml_model.outputs['model'],
gcs_source_uris=[input_batch_gcs_path],
gcs_destination_output_uri_prefix=f'gs://{BUCKET_NAME}/prediction_output/test'
)
UPDATE 2关于GCS路径，解决方法是在组件外部定义路径并将其作为输入参数传递，例如（缩写）：
@dsl.pipeline(
name="my-pipeline",
pipeline_root=PIPELINE_ROOT,
)
def pipeline(
gcp_project: str,
region: str,
bucket: str
):
ts = datetime.datetime.now().strftime("%Y%m%d-%H%M%S")
gcs_prediction_input_path = f'gs://{BUCKET_NAME}/prediction_input/video_batch_prediction_input_{ts}.jsonl'
batch_input_data_op = get_input_data(gcs_prediction_input_path) # this loads input data to GCS path
batch_predict_op = gcc_aip.ModelBatchPredictOp(
project=gcp_project,
model=training_job_run_op.outputs["model"],
job_display_name='batch-prediction',
# gcs_source_uris=[batch_input_data_op.output],
gcs_source_uris=[gcs_prediction_input_path],
gcs_destination_output_uri_prefix=f'gs://{BUCKET_NAME}/prediction_output/',
).after(batch_input_data_op) # we need to add 'after' so it runs after input data is prepared since get_input_data doesn't returns anything
仍然不确定，为什么它不工作/编译时，我返回GCS路径从get_input_data组件
我很高兴您解决了大部分主要问题并找到了模型声明的变通方法。
对于你对gcs_source_uris的input.输出观察，它背后的原因是因为函数/类返回值的方式。如果你深入挖掘google_cloud_pipeline_components的类/方法，你会发现它实现了一个结构，允许你从调用的函数的返回值中使用。输出。
如果你去管道的一个组件的实现，你会发现它从convert_method_to_component函数返回一个输出数组。所以，为了在你的自定义类/函数中实现它，你的函数应该返回一个可以作为属性调用的值。下面是它的基本实现。
class CustomClass():
def __init__(self):
self.return_val = {'path':'custompath','desc':'a desc'}
@property
def output(self):
return self.return_val
hello = CustomClass()
print(hello.output['path'])
如果您想深入了解它，您可以访问以下页面：
>
convert_method_to_component，这是convert_method_to_component的实现
属性，python中属性的基础知识。
