当前位置：首页 > news >正文

Python Web 开发进阶实战：AI 编排引擎 —— 在 Flask + Vue 中构建低代码机器学习工作流平台

news 2026/3/27 0:53:24

第一章：为什么需要 AI 编排引擎？

1.1 传统 ML 开发的痛点

阶段	问题

实验阶段| Notebook 无法版本控制，参数散落在 cell 中
协作阶段| 同事无法复现你的结果
生产阶段| 需将 notebook 重构成 Airflow DAG，重复劳动

1.2 编排引擎 vs 现有方案

方案	优点	缺点

Jupyter| 交互灵活 | 不可复现、难调度
Airflow| 强大调度 | 学习曲线陡峭，非面向 ML
MLflow Pipelines| 实验跟踪 | 缺少可视化编排
自研低代码平台|ML 友好 + 可视化 + 轻量| 需初期投入开发 |

定位：填补实验 → 生产的鸿沟，专注 ML 工作流。

第二章：平台架构设计

2.1 整体数据流

[Vue 前端] │ (拖拽生成 DAG JSON) ↓ [Flask API] → 保存 workflow 定义 │ ↓ [Celery 调度器] → 解析 DAG，按依赖执行任务 │ ├── [组件执行器] → 加载 Python 模块（如 data_loader.py） ├── [Optuna 集成] → 自动超参搜索 └── [WebSocket] → 推送实时日志/状态 │ ↓ [前端画布] → 高亮运行中节点，显示日志弹窗

2.2 核心抽象：组件（Component）

每个 ML 步骤封装为独立组件：

# components/base.py class MLComponent: name: str # 如 "CSV Data Loader" inputs: List[str] # 输入端口 ["file_path"] outputs: List[str] # 输出端口 ["dataframe"] config_schema: dict # 配置表单（JSON Schema） def run(self, inputs: dict, config: dict) -> dict: """执行逻辑，返回输出""" raise NotImplementedError

示例组件：
CSVLoader：读取 CSV → 输出 DataFrame
ProphetTrainer：训练时间序列模型 → 输出 model.pkl
ModelEvaluator：计算 MAE/RMSE → 输出 metrics.json

第三章：前端 DAG 画布（Vue + X6）

3.1 初始化 X6 画布

<template> <div ref="graphContainer" class="workflow-canvas"></div> </template> <script setup> import { Graph, Node } from '@antv/x6' let graph onMounted(() => { graph = new Graph({ container: graphContainer.value, grid: true, snapline: true, keyboard: true, clipboard: true }) // 注册节点模板 graph.registerNode('ml-component', { inherit: 'rect', width: 180, height: 60, attrs: { body: { fill: '#f5f5f5', stroke: '#333' }, label: { text: '未命名组件', fill: '#333' } } }) }) </script>

3.2 拖拽添加组件

// 从左侧工具栏拖入 const componentList = [ { type: 'csv_loader', name: 'CSV 数据加载器' }, { type: 'prophet_trainer', name: 'Prophet 训练器' } ] function addComponent(type, name) { const node = graph.addNode({ shape: 'ml-component', label: name, data: { type, config: {} }, // 存储组件类型与配置 ports: { groups: { input: { position: 'top' }, output: { position: 'bottom' } }, items: [ { id: 'in-1', group: 'input' }, { id: 'out-1', group: 'output' } ] } }) }

3.3 连接与配置

连线：X6 自动处理端口连接
配置弹窗：点击节点 → 弹出动态表单（基于 JSON Schema）

<!-- 动态表单 --> <template> <DynamicForm :schema="selectedNode.data.config_schema" v-model="config" /> </template>

表单库推荐：vue-json-schema-form 或自研轻量版。

第四章：后端调度器实现

4.1 Workflow 定义存储

# models/workflow.py class Workflow(db.Document): name = StringField(required=True) dag = DictField() # X6 导出的 JSON 结构 created_by = StringField() # 示例 dag 结构 # { # "nodes": [{"id": "n1", "type": "csv_loader", "config": {...}}], # "edges": [{"source": "n1", "target": "n2"}] # }

4.2 DAG 解析与执行

# services/scheduler.py from celery import chain, group def execute_workflow(workflow_id: str): wf = Workflow.objects.get(id=workflow_id) dag = wf.dag # 构建拓扑排序 sorted_nodes = topological_sort(dag['nodes'], dag['edges']) # 生成 Celery 任务链 tasks = [] for node in sorted_nodes: component = load_component(node['type']) task = component_task.s( node_id=node['id'], config=node['config'], inputs=get_inputs_from_predecessors(node, results) ) tasks.append(task) # 执行 chain(*tasks).apply_async()

4.3 组件任务模板

# tasks/component_executor.py @celery.task def component_task(node_id: str, config: dict, inputs: dict): # 1. 加载组件类 comp_class = get_component_class_by_type(config['type']) component = comp_class() # 2. 执行 outputs = component.run(inputs, config) # 3. 保存输出（供下游使用） save_outputs(node_id, outputs) # 4. 推送状态 socketio.emit('node_status', {'node_id': node_id, 'status': 'completed'}) return outputs

第五章：组件开发规范

5.1 CSV 加载器示例

# components/data/csv_loader.py class CSVLoader(MLComponent): name = "CSV 数据加载器" inputs = [] outputs = ["dataframe"] config_schema = { "type": "object", "properties": { "file_path": {"type": "string", "format": "file"}, "sep": {"type": "string", "default": ","} }, "required": ["file_path"] } def run(self, inputs, config): df = pd.read_csv(config["file_path"], sep=config["sep"]) return {"dataframe": df}

5.2 组件注册机制

# components/registry.py COMPONENT_REGISTRY = {} def register_component(name: str): def decorator(cls): COMPONENT_REGISTRY[name] = cls return cls return decorator # 使用 @register_component("csv_loader") class CSVLoader(MLComponent): ...

第六章：场景实战

6.1 销售预测 Pipeline

拖拽组件：
- CSVLoader→DateFeatureEngineer→ProphetTrainer→ModelSaver
配置：
- CSV 路径：/data/sales.csv
- Prophet 参数：changepoint_prior_scale=0.05
执行：
- 一键运行，自动保存模型至 MLflow

6.2 图像分类 Pipeline

组件链：
ImageFolderLoader→AlbumentationsAugmenter→ResNetFinetuner→ConfusionMatrixEvaluator
优势：
- 数据增强策略可配置（旋转/裁剪概率）
- 自动记录 Top-1 Accuracy 到平台

6.3 A/B 测试多模型

并行分支：
- 分支1：RandomForestTrainer
- 分支2：XGBoostTrainer
汇总节点：
- ModelComparator：输出对比报告（准确率/训练时间）
前端展示：热力图高亮最优模型

第七章：高级功能

7.1 超参自动调优（Optuna 集成）

组件扩展：
- OptunaTuner组件：包裹任意训练组件
- 配置搜索空间（如learning_rate: [0.001, 0.1]）

# components/tuning/optuna_tuner.py def run(self, inputs, config): def objective(trial): tuned_config = {k: trial.suggest_float(k, *v) for k, v in config["search_space"].items()} outputs = wrapped_component.run(inputs, tuned_config) return -outputs["metrics"]["accuracy"] # 最小化负准确率 study = optuna.create_study() study.optimize(objective, n_trials=50) return {"best_params": study.best_params}

7.2 实时日志流

Celery 信号：捕获任务日志
WebSocket 推送：

# 监听 Celery 任务 @celery.signals.after_task_publish.connect def task_sent(sender=None, headers=None, **kwargs): socketio.emit('log', f"Task {sender} queued") @celery.signals.task_postrun.connect def task_finished(task_id=None, retval=None, **kwargs): socketio.emit('log', f"Task {task_id} completed")