KFP组件传递自定义类型参数报错，如何实现自定义参数传递？

你遇到的错误是因为KFP组件仅支持Python内置类型（str、int、float、bool等），自定义类实例（比如SentenceTransformer）无法被KFP的参数系统直接识别——它缺少KFP要求的schema元数据。以下是几种实用的解决方法：

方法1：序列化/反序列化自定义对象 #

将自定义对象转换为KFP支持的内置类型（如字符串）传递，在目标组件中再还原为原对象：

• 使用Pickle（仅限可信环境）：

  # 发送端组件：序列化模型为字符串
  import pickle
  from kfp import dsl
  from sentence_transformers import SentenceTransformer
  
  @dsl.component
  def serialize_model_component(model: SentenceTransformer) -> str:
      return pickle.dumps(model).decode('latin1')
  
  # 接收端组件：反序列化字符串为模型
  @dsl.component
  def deserialize_model_component(serialized_str: str) -> SentenceTransformer:
      return pickle.loads(serialized_str.encode('latin1'))
  

  注意：Pickle存在安全风险，不要用于不可信输入的场景。

• 使用JSON序列化（推荐）：
  如果自定义对象的核心参数可以用JSON表示，自定义序列化逻辑：

  import json
  from kfp import dsl
  from sentence_transformers import SentenceTransformer
  
  @dsl.component
  def export_model_params(model: SentenceTransformer) -> str:
      # 提取可序列化的核心参数
      params = {
          'model_name': model.model_name,
          'cache_dir': model.cache_dir
      }
      return json.dumps(params)
  
  @dsl.component
  def load_model_from_params(params_str: str) -> SentenceTransformer:
      params = json.loads(params_str)
      # 重新实例化模型
      return SentenceTransformer(**params)
  

方法2：使用KFP Artifacts传递复杂对象 #

对于模型这类复杂对象，更符合KFP设计理念的方式是用Artifact传递，而非参数：

from kfp import dsl
from kfp.dsl import Model
from sentence_transformers import SentenceTransformer

@dsl.component
def save_model_to_artifact(model: SentenceTransformer, output_model: dsl.Output[Model]):
    # 将模型保存到Artifact指定的路径
    model.save(output_model.path)
    # 可选：添加元数据便于后续追踪
    output_model.metadata['model_name'] = model.model_name

@dsl.component
def load_model_from_artifact(input_model: dsl.Input[Model]) -> SentenceTransformer:
    # 从Artifact路径加载模型
    return SentenceTransformer.load(input_model.path)

这种方式还能利用KFP的Artifact管理能力，比如版本控制、存储持久化等。

方法3：传递构造参数而非对象实例 #

如果自定义对象可以通过简单的内置类型参数重建，直接传递这些参数，在目标组件中实例化：

from kfp import dsl
from sentence_transformers import SentenceTransformer

@dsl.component
def train_model_component(model_name: str) -> str:
    # 用传入的字符串参数实例化模型
    model = SentenceTransformer(model_name)
    # 执行训练逻辑
    return model_name  # 返回参数供后续组件使用

@dsl.component
def run_inference_component(model_name: str):
    model = SentenceTransformer(model_name)
    # 执行推理逻辑

这种方法最简单，无需处理序列化逻辑，适合可通过少量参数重建的对象。

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