咨询Vertex AI Pipelines无需Docker/PyPi导入自定义模块的方法

以下是几种无需构建专用Docker镜像或发布到PyPI的方案，解决代码重复粘贴的痛点：

方法1：通过组件定义直接绑定本地模块文件 #

适用于单个或少量自定义模块文件，直接在组件创建时指定依赖文件，KFP会自动将文件打包上传到Vertex AI运行环境。

示例代码：

from kfp import dsl
from kfp.components import create_component_from_func

# 自定义组件逻辑，直接导入本地模块
def process_data_component(input_str: str) -> str:
    from my_utils import format_input  # 导入自定义工具函数
    return format_input(input_str)

# 创建组件时绑定本地模块文件
data_processor = create_component_from_func(
    process_data_component,
    packages_to_install=[],  # 无需额外安装包
    files={
        "./my_utils.py": "/app/my_utils.py"  # 本地文件路径 -> 容器内目标路径
    }
)

# 定义流水线
@dsl.pipeline(name="custom-module-pipeline")
def pipeline():
    data_processor(input_str="raw_input")

方法2：打包模块为Zip后动态导入 #

如果是多文件的复杂模块，将整个模块目录打包成Zip文件，在组件内解压并添加到Python路径。

步骤：

1. 将自定义模块目录（如my_utils/）打包为my_utils.zip
2. 在组件函数中解压并导入：

示例代码：

def complex_process_component(input_data: str) -> str:
    import sys
    import zipfile

    # 解压Zip包到临时目录
    with zipfile.ZipFile("my_utils.zip", "r") as zip_ref:
        zip_ref.extractall("/tmp/my_modules")
    
    # 将解压后的路径添加到Python搜索路径
    sys.path.append("/tmp/my_modules")
    
    # 导入模块中的函数
    from my_utils.data_cleaner import clean_data
    from my_utils.validator import validate_input

    cleaned = clean_data(input_data)
    return validate_input(cleaned)

# 创建组件时绑定Zip包
complex_processor = create_component_from_func(
    complex_process_component,
    files={
        "./my_utils.zip": "/app/my_utils.zip"
    }
)

方法3：通过Volume挂载本地模块目录 #

适用于模块频繁修改的场景，将本地模块目录挂载到组件运行容器，无需每次打包上传，修改本地文件后流水线直接使用最新版本。

示例代码：

@dsl.pipeline(name="mounted-module-pipeline")
def pipeline():
    # 创建存储卷
    module_volume = dsl.VolumeOp(
        name="module-storage",
        size="1Gi",
        storage_class="standard"
    )

    # 定义组件，挂载卷并添加路径
    def mounted_process_component(input: str) -> str:
        import sys
        sys.path.append("/app/my_modules")
        from my_utils.core import core_process
        return core_process(input)

    mounted_processor = create_component_from_func(mounted_process_component)

    # 将卷挂载到组件容器
    processor_task = mounted_processor(input="test_input")
    processor_task.add_volume(module_volume).add_volume_mount(
        dsl.VolumeMount(
            mount_path="/app/my_modules",
            name=module_volume.name
        )
    )

方案优势 #

• 无需维护Docker镜像或PyPI包，降低运维成本
• 自定义模块只需维护一份，修改后所有引用组件自动同步更新
• 代码结构清晰，避免重复粘贴导致的可读性差、维护困难问题

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