如何在Vertex AI流水线组件X失败后触发组件Y？

一、使用kfp.dsl.Condition实现失败触发 #

Kubeflow Pipelines中，每个任务执行后会返回task.state属性，可用于判断任务状态（如FAILED、SUCCEEDED）。你可以基于组件X的状态设置条件，让组件Y仅在X失败时执行。

示例代码：

from kfp import dsl
from kfp.components import create_component_from_func

# 模拟偶发失败的组件X
@create_component_from_func
def component_x() -> None:
    import random
    import sys
    if random.random() < 0.5:
        sys.exit(1)  # 随机触发失败
    print("组件X执行成功")

# 失败时触发的组件Y
@create_component_from_func
def component_y() -> None:
    print("检测到组件X失败，执行组件Y")

@dsl.pipeline(name="失败触发流水线")
def failure_trigger_pipeline():
    x_task = component_x()
    
    # 当组件X状态为FAILED时，执行组件Y
    with dsl.Condition(x_task.state == dsl.ConditionTaskState.FAILED):
        component_y()

这里dsl.ConditionTaskState.FAILED是KFP内置的状态常量，直接用来匹配组件X的失败状态即可。

二、Vertex AI/Kubeflow Pipelines的原生失败处理功能 #

除了Condition，还有两种原生机制可以实现类似需求：

• dsl.OnExit 钩子：这个钩子会在组件X执行完成（无论成功或失败）后触发指定组件。如果只想在失败时执行Y，可以把X的状态传入Y组件，在Y内部做判断：

  @create_component_from_func
  def component_y(task_state: str) -> None:
      if task_state == "FAILED":
          print("组件X失败，执行组件Y逻辑")
  
  @dsl.pipeline(name="OnExit示例流水线")
  def onexit_pipeline():
      x_task = component_x()
      
      # 绑定OnExit组件，传入X的状态
      with dsl.OnExit(x_task):
          component_y(x_task.state)
  

• 任务重试机制：如果组件X是偶发失败，可以先给X设置重试策略，减少触发Y的场景：

  x_task = component_x().set_retry(
      num_retries=3,  # 重试3次
      retry_delay=dsl.Duration(seconds=30),  # 每次重试间隔30秒
      retry_on=["Failed"]  # 仅在失败时重试
  )
  

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