Запуск конвейера Vertex AI завершается сбоем после прохождения dsl.Condition==True

Question

Я работаю с конвейерами Vertex AI уже 3 месяца. Я начал с миграции проектов из старой специализированной инфраструктуры Kubernetes в конвейеры на основе компонентов Vertex AI Python. Я создал конвейеры шаблонов, которые до прошлой недели работали нормально. Недавно я внес некоторые изменения, и мой конвейер не выдает сообщение об ошибке.Failed to evaluate the expression with error: UNKNOWN: No value with name "True" found in Activation; Failed to evaluate the TriggerPolicy.condition field.
Мое определение конвейера выглядит следующим образом

      @kfp.v2.dsl.pipeline(name="my-custom-pipeline-name", )
def my_pipeline_new(
                     ENV: str = "dev",
                     ISO_CODE_COUNTRY: str = "DE",
                     DEFAULT_TRUE_FLAG: bool=True,
                     ENABLE_PERSIST_RESULTS:bool= True,
                     SOME_FLAG: bool = True

                    ):   

    with kfp.v2.dsl.Condition(DEFAULT_TRUE_FLAG==True) as C1:
        with kfp.v2.dsl.Condition(SOME_FLAG== True):
            some_module_obj = some_python_based_component(env=ENV, iso_code= ISO_CODE_COUNTRY, enable_persist_results= ENABLE_PERSIST_RESULTS)
    

Моя текущая версия KFP — 1.7.0. Я совершенно потерялся здесь, чтобы выяснить, почему мой конвейер внезапно перестал работать.

Возможно ли, что в Vertex AI Pipelines произошли какие-то обновления, о которых я не знаю?
Любые предложения/идеи будут высоко оценены.

Я использовал методы проб и ошибок, чтобы увидеть, но мой конвейер даже не запускается и дает сбой на начальном этапе. Когда я пытаюсь запустить свой компонент на основе Python без каких-либо условий, он работает совершенно нормально, но с условиями он просто терпит неудачу.

Чтобы убедиться в правильности моих подозрений, я создал фиктивный конвейер, который, похоже, также не работает в условиях dsl.

      import kfp
from kfp.v2.dsl import (
    component,
    Dataset,
    Input,
    Output,
    Metrics,
    Artifact,
    ClassificationMetrics,
    Model,
    Metrics,
    Dataset,
    OutputPath,
    InputPath,
    Condition,
)

@component
def component_A(flag: bool):
    if flag:
        print("Flag is true.")
    else:
        print("Flag is false.")

@component
def component_B():
    print("Running component B.")

@kfp.dsl.pipeline(
    name="conditional-pipeline",
    description="A simple pipeline with a condition",
    pipeline_root="gs:bucket-name"
)
def my_pipeline(flag: bool = True):
    with Condition(flag == True):
        a = component_A(flag)
    b = component_B()

PIPELINE_ROOT = "{}/pipeline_root".format("MY-GCS-BUCKET")

Вот фрагмент конвейера, который дает сбой и выдает ошибку:
Ошибка простого кода условия

Answer 1

VertexAI растет очень быстро и, к сожалению, из-за небольшого количества документации и мало кто ее использует! Существует множество сломанных и неработающих модулей и кодов, связанных с VertexAI, а также таких модулей, как пакетное прогнозирование.

Возможный способ устранить вашу ошибку (не очень чистый) — использовать выходные данные такого компонента:

      ...
**@component
def set_flag(flag: bool):
    
return flag**

@kfp.dsl.pipeline(
    name="conditional-pipeline",
    description="A simple pipeline with a condition",
    pipeline_root="gs:bucket-name"
)
def my_pipeline(flag: bool = True):
    **result = set_flag(flag)**
    with Condition(
                **result.output == True,**
                name = 'check flag?'):

        a = component_A(flag)
    b = component_B()
...

Вы также можете проверить рабочий пример в GCP github: ссылка