如何创建按顺序调用的对象的自定义管道？

Question

Scikit-learn 具有 pipeline 功能，它是具有最终估计器的转换管道。

如何创建适用于没有 scikit learn fit & transform 方法但有____call___() 的类的东西？我不希望解决方案使用 scikit-learn 管道。所以我想按照下面列出的顺序执行以下操作

1. 将 dict 传递给 ppl，ppl 将 dict 传递给 A().____call____()
2. 第 1 步返回的输出 df 被传递给 B().____call____()
3. 第 2 步返回的输出被传递给 C().____call____()
4. 第 3 步返回的输出被传递给 lambda 函数，该函数返回 A 和 B 两列的总和

例如：

import pandas as pd

class A:
    def __init__(self, sample=1):
        self.sample = sample

    def __call__(self, dct):
        return pd.DataFrame(dct)[:self.sample]

class B:
    def __init__(self, col1, col2):
        self.col1 = col1
        self.col2 = col2
    def __call__(self, df):
        return df[self.col1], df[self.col2]

class C:
    def __call__(self, x, y):
        return x+y


ppl = CustomPipeline(pipeline=[('A', A(sample=700)),
                               ('B', B(col1='A', col2='B')),
                               ('C', C())
                               ('self', lambda x: x)])
df_sum = ppl(dct={'A': [1, 2, 4], 'B': [10, 2, 3]})

问题

• 如何实现 CustomPipeline() 以使其按照我上面列出的示例工作？
• 在每个类中使用 main() 是否比在类中实现 ____call___() 方法更好？

Answer 1

from typing import Sequence, Tuple, Callable, Any

def CustomPipeline(pipeline: Sequence[Tuple[str, Callable[[Any], Any]]]) -> Callable[[Any], Any]:
    def _run(value: Any) -> Any:
        for name, stage in pipeline:
            value = stage(value)
        return value
    return _run

>>> ppl = CustomPipeline([('a', lambda v: v*2), ('b', lambda v: v+1)])
>>> ppl(5)
11

外部函数用于创建具有管道阶段的闭包。

这可以很容易地用一个类来实现。