google-cloud-platform - VertexAI パイプラインと Kuberflow コンポーネントを使用してバッチ予測を自動化する

Question

以下のコードは、VertexAI で既にトレーニングされたモデルを読み込み、バッチ予測のパイプラインを実行します。ただし、どこから来たのかを把握できないというjsonデコーダーエラーが発生します。入力ファイルは jsonl 形式で、VertexAI ダッシュボードから手動でバッチ予測を実行すると正常に動作します。したがって、パイプラインに見えない何か問題があります。

ヘルプはありますか？

 import kfp
 import google.cloud.aiplatform as aip
 from google_cloud_pipeline_components import aiplatform as gcc_aip
 
 import datetime
 
 from kfp.v2 import compiler 
 from kfp.v2.dsl import component, Artifact, Output

 PROJECT_ID='my-project-id'
 REGION='europe-west4'
 SOURCE_ROOT='gs://source_root/'
 JSONL_FILE='input.jsonl'
 DESTINATION_OUTPUT='gs://destination_output'
 PIPELINE_ROOT='gs://bucket/pipeline_root/'
 MODEL_ID='vertexai-model-id'

 ts = int(datetime.datetime.utcnow().timestamp() * 100000)

 @component()
 def load_ml_model(project_id: str, model: Output[Artifact]):
     """Load existing Vertex model"""
     region='europe-west4'
     model_id=MODEL_ID
     model_uid=f'projects/{project_id}/locations/{region}/models/{model_id}'
     model.uri = model_uid
     model.metadata['resourceName'] = model_uid

@kfp.dsl.pipeline(
    name='batch-pipe'+str(ts),
    pipeline_root=PIPELINE_ROOT)
def pipeline(project_id: str):
    ml_model=load_ml_model(project_id='my-project-id')

    model_batch_pred_op = gcc_aip.ModelBatchPredictOp(
         project=project_id,
         location=REGION,
         job_display_name='batch-pred',
         model=ml_model.outputs['model'],
         gcs_source_uris=f'gs://source_root/input.jsonl',
         gcs_destination_output_uri_prefix=f'gs://destination_output/'
        )

compiler.Compiler().compile(
     pipeline_func=pipeline,
     package_path="text_class_pipeline.json",
                           )

def run_batch_pred(project_id,region):
    aip.init(
       project=project_id,
       location=region,
          )

job = aip.PipelineJob(
    project=project_id,
    display_name='batch_pipeline',
    template_path='text_class_pipeline.json',
    pipeline_root=PIPELINE_ROOT,
    parameter_values={'project_id': project_id},
)

job.run()

run_batch_pred(project_id=PROJECT_ID, region=REGION)

エラーが発生しました

raise JSONDecodeError("Expecting value", s, err.value) from None
json.decoder.JSONDecodeError: Expecting value: line 1 column 217 (char 216)

また、モデルは正しくロードされます。バッチ予測ステージが失敗する