构建管道核心

tasks: load_data: next: - preprocess_data preprocess_data: next: - feature_engineer_data feature_engineer_data: next: - train_model train_model: next: - evaluate_model evaluate_model: next: - create_plots create_plots: next: []

def func_address(func): print(f"调用 {func}...") return func

@func_address def print_hello(): print("Hello, world!")

if name == "main": print_hello()

def pipeline_task(task_func): task_func.is_task = True return task_func

class MLPipeline: def init(self): # 加载配置文件并获取任务列表 # ...

1    self.tasks = []
2    # 'tasks' 是有向无环图（DAG）的顶点列表
3    for task in tasks:
4        try:
5            func = getattr(self, task)
6        except AttributeError:
7            raise Exception(f"管道中未定义 '{task}'。")
8
9        if hasattr(func, "is_task"):
10            self.tasks.append(func)
11        else:
12            raise Exception(f"'{func}' 不是一个管道任务。")
13
14    # ...
15
16@pipeline_task
17def load_data(self) -> None:
18    # ...
19    pass
20
21@pipeline_task
22def preprocess_data(self) -> None:
23    # ...
24    pass
25
26def not_a_task(self, a, b) -> None:
27    # ...
28    pass
29
30def run(self) -> None:
31    for task in self.tasks:
32        task()