如何在AWS SageMaker Pipeline运行时获取被覆盖的参数值？

组件函数内直接使用 #

这是最常用的处理方式：把定义的参数直接作为组件函数的入参，运行时框架会自动将覆盖后的实际值注入到函数中，无需额外操作。

示例代码：

from kfp import dsl
from kfp.dsl import ParameterString

# 组件函数直接接收参数
@dsl.component
def print_runtime_param(foo: str):
    print(f"实际运行时的foo参数值：{foo}")

@dsl.pipeline(name="demo-pipeline")
def demo_pipeline():
    # 定义参数，默认值设为"default_val"
    foo_param = ParameterString(name="Foo", default_value="default_val")
    # 将参数传递给组件
    print_runtime_param(foo=foo_param)

# 运行时覆盖参数
if __name__ == "__main__":
    from kfp.client import Client
    client = Client()
    # 这里传入的'bar'会覆盖默认值
    client.create_run_from_pipeline_func(demo_pipeline, parameters=dict(Foo='bar'))

在print_runtime_param组件里，foo变量拿到的就是运行时覆盖后的'bar'，直接使用即可。

管道顶层上下文获取（特殊场景） #

如果确实需要在管道定义的顶层逻辑中获取运行时值，可以通过管道上下文读取：

@dsl.pipeline(name="context-pipeline")
def context_pipeline():
    foo_param = ParameterString(name="Foo", default_value="default_val")
    # 获取当前管道执行上下文
    pipeline_context = dsl.get_pipeline_context()
    # 从上下文取参数值，取不到则用默认值兜底
    runtime_foo = pipeline_context.parameters.get("Foo", foo_param.default_value)
    print(f"管道上下文里拿到的foo值：{runtime_foo}")

关键注意点 #

• ParameterString这类参数对象只是定义阶段的声明载体，本身不会存储运行时的实际值，所以不要试图直接从它的属性中获取覆盖后的值。
• 绝大多数场景下，直接在组件内接收参数就足够满足需求，上下文获取仅适用于管道顶层需要根据参数做逻辑判断的特殊情况。

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