我正在使用Kubeflow轻量级组件在VertexAI工作台上创建管道。

我从一个组件开始，从BigQuery中提取数据并对其进行处理（并返回一个数据帧）。然后，这个数据帧将被馈送到另一个组件：

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

输入的是pd. DataFrame（），输出是WindowGenerator类，负责在下一步将所述数据帧转换为神经网络的输入。

每当我运行该组件（turn_window_generator）时，都会收到以下错误：

---------------------------------------------------------------------------
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'

我希望能够创建这个组件。我尝试安装kfp--pre和一些其他更改，但似乎都不起作用。这些是我的安装和导入。

! 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