yaml-workflow-executor

用途

YAML Workflow Executor 用于执行配置驱动的工作流：由 YAML 文件定义分析参数、数据源和执行步骤，工具负责加载、校验、路由到对应处理器并执行，最终产出结果。它适合运行分析管道、数据处理任务，以及用参数化配置实现重复性任务的自动化，并帮助构建可复现的工作流。

主要能力

配置加载：通过 WorkflowConfig.from_yaml 从 YAML 文件读取 task、input、output、parameters、options 等字段，并支持 validate 校验必填项。
任务路由：WorkflowRouter 按 task 名称将配置分发到已注册的处理器，未注册的任务会报错并列出可用任务。
内置处理器示例：analyze_data（读取 CSV、按参数过滤、输出统计结果）、generate_report（生成报告）、transform_data（按配置执行 rename、filter、aggregate、sort 等转换）。
多步骤工作流：task 为 pipeline 时，可通过 steps 定义多个步骤，并用 depends_on 声明依赖关系。
命令行接口：支持传入配置文件路径、--override/-o 参数覆盖、--verbose/-v 详细输出、--dry-run 仅校验不执行。
参数覆盖与编程调用：execute_workflow 支持 overrides 参数，也可在代码中加载并修改配置后调用 router.route。
错误处理：提供 safe_execute_workflow 模板，覆盖配置缺失、YAML 语法错误、校验错误和执行错误等情形。

使用方式

编写 YAML 配置，例如：

task: analyze_data

input:
  data_path: data/raw/measurements.csv

output:
  results_path: data/results/analysis.json

parameters:
  filter_column: status
  filter_value: active

通过 Python 调用：

from workflow_executor import execute_workflow

result = execute_workflow("config/workflows/analysis.yaml")
print(f"Status: {result['status']}")

通过 CLI 执行：

python -m workflow_executor config/workflows/analysis.yaml --verbose

使用参数覆盖：

python -m workflow_executor config/workflows/analysis.yaml \
    --override filter_value=completed \
    --override date_range.start=2024-06-01

仅校验配置：

python -m workflow_executor config/workflows/analysis.yaml --dry-run

也可使用 Bash 包装脚本 scripts/run_workflow.sh 运行。

适用场景

运行由 YAML 配置定义的分析任务。
从配置执行数据处理管道。
用参数化配置自动化重复性任务。
构建可复现的工作流。
通过配置变体处理多个场景。
批量处理多个配置文件（遍历目录中的 *.yaml 逐个执行）。
动态生成工作流配置（如为多个数据文件生成 pipeline 步骤）。

限制与注意事项

资料未提供该工具的安装方式、依赖要求、收费模式、支持平台或性能基准，这些信息暂未提供。
内置处理器（analyze_data、generate_report、transform_data）为示例实现，实际可用任务取决于已注册的处理器。
使用前需确保配置文件存在且 YAML 语法有效，输入数据文件存在，输出目录可创建，且指定任务已注册。
建议将配置放在 config/workflows/ 目录，使用描述性文件名，纳入版本控制，并在执行前用 --dry-run 校验。
应避免硬编码绝对路径、跳过输入校验、将配置与实现混在一起、创建过度复杂的嵌套配置以及忽略错误处理。

技能介绍

用途

主要能力

使用方式

适用场景

限制与注意事项

相关技能