如何修复Kubeflow组件中to_dict()缺少self参数的TypeError

在VertexAI Workbench上使用Kubeflow轻量级组件构建流水线，先编写了从BigQuery提取并处理数据的组件（返回pd.DataFrame），接着编写turn_window_generator组件，输入为pd.DataFrame，输出为自定义WindowGenerator类（用于将数据转换为神经网络输入）。运行该组件时出现错误：TypeError: to_dict() missing 1 required positional argument: 'self'，尝试安装kfp预发布版无效，寻求有效修复方案。

相关代码 #

@kfp.dsl.component
def turn_window_generator(df: pd.DataFrame) -> WindowGenerator:
    ....
    return wide_window

错误栈 #

---------------------------------------------------------------------------
TypeError                                 Traceback (most recent call last)
/var/tmp/ipykernel_5481/471668721.py in <module>
      1 @kfp.dsl.component
----> 2 def turn_window_generator(df: pd.DataFrame) -> WindowGenerator:
      3 
      4     filter_x_days = 7
      5     filtered_days = [i for i in range(0, filter_x_days)] + [29]

~/.local/lib/python3.7/site-packages/kfp/components/component_decorator.py in component(func, base_image, target_image, packages_to_install, pip_index_urls, output_component_file, install_kfp_package, kfp_package_path)
    125         output_component_file=output_component_file,
    126         install_kfp_package=install_kfp_package,
--> 127         kfp_package_path=kfp_package_path)

~/.local/lib/python3.7/site-packages/kfp/components/component_factory.py in create_component_from_func(func, base_image, target_image, packages_to_install, pip_index_urls, output_component_file, install_kfp_package, kfp_package_path)
    467             func=func)
    468 
--> 469     component_spec = extract_component_interface(func)
    470     component_spec.implementation = structures.Implementation(
    471         container=structures.ContainerSpecImplementation(

~/.local/lib/python3.7/site-packages/kfp/components/component_factory.py in extract_component_interface(func, containerized)
    228                     ' values for outputs are not supported.')
    229 
--> 230         type_struct = type_utils._annotation_to_type_struct(parameter_type)
    231         if type_struct is None:
    232             raise TypeError(

~/.local/lib/python3.7/site-packages/kfp/components/types/type_utils.py in _annotation_to_type_struct(annotation)
    525         return None
    526     if hasattr(annotation, 'to_dict'):
--> 527         annotation = annotation.to_dict()
    528     if isinstance(annotation, dict):
    529         return annotation

TypeError: to_dict() missing 1 required positional argument: 'self'

安装命令 #

! pip3 install --upgrade {USER_FLAG} -q google-cloud-aiplatform \
                                        google-cloud-storage {USER_FLAG} \
                                        kfp --pre \
                                        google-cloud-pipeline-components \
                                        tensorflow

导入语句 #

import google.cloud.aiplatform as aip
import kfp
from kfp.v2 import compiler

修复方案 #

核心原因 #

Kubeflow组件的类型系统无法直接序列化自定义类WindowGenerator，组件装饰器在解析输出类型注解时，错误地将类本身当成可调用的实例对象，调用to_dict()时缺少实例参数self。

具体解决步骤 #

1. 转换为Kubeflow支持的序列化格式
   避免直接返回自定义类，将WindowGenerator实例转换为字典、JSON字符串，或保存为文件后返回路径：

   • 转为字典返回，后续组件从字典重建实例：

     @kfp.dsl.component
     def turn_window_generator(df: pd.DataFrame) -> dict:
         filter_x_days = 7
         filtered_days = [i for i in range(0, filter_x_days)] + [29]
         # 原逻辑创建wide_window实例
         wide_window = WindowGenerator(...)
         # 将实例转为可序列化的字典
         return wide_window.__dict__
     

   • 保存为文件返回路径（适用于包含不可序列化属性的类）：

     import pickle
     from kfp.dsl import Artifact, Output
     
     @kfp.dsl.component
     def turn_window_generator(df: pd.DataFrame, window_artifact: Output[Artifact]):
         filter_x_days = 7
         filtered_days = [i for i in range(0, filter_x_days)] + [29]
         wide_window = WindowGenerator(...)
         # 保存实例到文件
         with open(window_artifact.path, 'wb') as f:
             pickle.dump(wide_window, f)
     

2. 调整类型注解（不推荐）
   如果暂时需要绕过类型检查，可使用typing.Any作为输出类型注解，但可能导致后续组件处理对象时出错：

   from typing import Any
   
   @kfp.dsl.component
   def turn_window_generator(df: pd.DataFrame) -> Any:
       ...
       return wide_window
   

3. 指定稳定版KFP
   尝试安装稳定版本的KFP，避免预发布版的潜在bug：

   ! pip3 install --upgrade {USER_FLAG} -q google-cloud-aiplatform \
                                           google-cloud-storage {USER_FLAG} \
                                           kfp==2.5.0 \
                                           google-cloud-pipeline-components \
                                           tensorflow
   

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