YAML Workflow

YAML Workflow

一款轻量级工作流引擎，适用于 CI/CD 流水线、数据处理和 DevOps 自动化。使用 YAML 定义可复现、受版本控制的工作流——可在本地、CI 中或任何安装了 Python 的机器上运行。

为什么选择 yaml-workflow？

大多数工作流工具都需要服务、数据库和复杂的基础设施。yaml-workflow 采用 GitOps 方式——工作流是纯 YAML 文件，与代码一同进行版本控制：

	yaml-workflow	Airflow / Prefect / Dagster
设置	pip install yaml-workflow	服务、数据库、调度器、工作进程
配置	纯 YAML 文件	Python DAG + 基础设施配置
依赖项	2 个（PyYAML、Jinja2）	50 多个包、Docker、PostgreSQL
适用场景	本地自动化、脚本、CI/CD、数据流水线	大规模企业级编排
学习曲线	几分钟	几小时到几天
状态	基于文件，可恢复	基于数据库

在以下情况下选择 yaml-workflow：

• 无需额外基础设施即可实现简单的任务自动化
• 使用受版本控制的 YAML 定义可复现的流水线
• 通过并行执行进行批处理
• 持久化状态，并在失败后恢复工作流
• 需要一个错误处理更完善的轻量级替代方案来代替 shell 脚本
• 需要可融入 GitOps 的流水线，与代码一起存放在代码仓库中
• 需要一个能在本地和 CI 中运行相同流水线的工具

功能特性

• 使用 Jinja2 模板的 YAML 驱动工作流定义
• 支持多种任务类型：shell、Python、文件、模板、HTTP、批处理
• 通过 imports 组合工作流——在多个工作流间复用步骤
• 通过入口点实现插件系统——pip install yaml-workflow-myplugin
• 监视模式——使用 --watch 在文件变更时重新运行
• 试运行模式，可预览而不执行
• 工作流可视化（ASCII 分支 DAG 和 Mermaid）
• 使用可配置的工作进程池进行并行执行
• 状态持久化与恢复能力
• 支持可配置策略的重试机制
• 命名空间变量（args、env、steps、batch）
• 通过自定义步骤序列和条件控制流程
• 通过 @register_task 装饰器扩展任务系统
• 通过 depends_on 并行执行步骤——并发运行相互独立的步骤
• 密钥校验——缺少必需环境变量时立即失败
• 结构化输出（--format json），便于集成 CI 和编写脚本
• MCP 服务——将工作流作为 AI Agent 工具公开（pip install yaml-workflow[mcp]）
• Web 仪表板——监控运行情况并触发工作流（pip install yaml-workflow[serve]）
• GitHub Action——通过 uses: orieg/[email protected] 在 CI 中运行工作流

使用场景

• CI/CD 流水线——使用 YAML 定义多步骤构建、测试和部署工作流
• 数据处理——失败后可重试和恢复的批量 ETL 流水线
• DevOps 自动化——具备密钥管理和通知功能的基础设施任务
• AI/LLM 流水线——编排带有身份验证、重试和批处理的 API 调用
• 本地自动化——使用可复现、可参数化的工作流替代 shell 脚本

快速开始

# Install (isolated CLI — recommended)
pipx install yaml-workflow            # Core CLI
pipx install 'yaml-workflow[all]'     # + web dashboard + MCP server

# Or with pip
pip install yaml-workflow

# Initialize example workflows
yaml-workflow init

# Run a workflow with parameters
yaml-workflow run workflows/hello_world.yaml name=Alice

工作流示例（hello_world.yaml）：

name: Hello World
description: A simple greeting workflow

params:
  name:
    type: string
    default: World

steps:
  - name: create_greeting
    task: template
    inputs:
      template: "Hello, {{ args.name }}!"
      output_file: greeting.txt

  - name: show_greeting
    task: shell
    inputs:
      command: cat greeting.txt

可视化工作流

yaml-workflow visualize workflows/data_pipeline.yaml

  Workflow: Data Pipeline

  ┌─────────────────┐
  │  detect_format  │
  │   python_code   │
  └─────────────────┘
           │
           ▼
  ┌────────────────┐  ┌────────────────┐  ┌────────────────┐  ┌────────────────┐
  │  process_json  │  │  process_csv   │  │  process_xml   │  │ handle_unknown │
  │     shell      │  │     shell      │  │     shell      │  │     shell      │
  └────────────────┘  └────────────────┘  └────────────────┘  └────────────────┘
           │
           ▼
  ┌─────────────────┐
  │ generate_report │
  │   python_code   │
  └─────────────────┘

相邻的条件步骤会自动归为分支。使用 --format mermaid 可导出内容，用于文档或 GitHub 渲染。

试运行模式

预览工作流将执行的操作，但不实际执行任何操作：

yaml-workflow run workflows/hello_world.yaml name=Alice --dry-run

[DRY-RUN] Workflow: Hello World
[DRY-RUN] Steps: 2 to execute

  [DRY-RUN] Step 'create_greeting' — task: template — WOULD EXECUTE
    template: Hello, Alice!
    output_file: greeting.txt
  [DRY-RUN] Step 'show_greeting' — task: shell — WOULD EXECUTE
    command: cat greeting.txt

[DRY-RUN] Complete. 2 step(s) would execute, 0 would be skipped.
[DRY-RUN] No files were written. No tasks were executed.

工作流组合

使用 imports 在工作流之间复用步骤：

# main.yaml
imports:
  - ./shared/logging_steps.yaml
  - ./shared/common_params.yaml

steps:
  - name: my_step
    task: shell
    inputs:
      command: echo "runs after imported steps"

导入的步骤会前置。导入的参数提供默认值，主工作流可以覆盖这些默认值。支持传递导入，并可检测循环导入。

并行步骤

使用 depends_on 并发运行彼此独立的步骤：

steps:
  - name: fetch_api
    task: http.request
    inputs: {url: "https://api.example.com/data"}

  - name: fetch_db
    task: python_code
    inputs:
      code: "result = query_database()"

  - name: merge
    task: python_code
    depends_on: [fetch_api, fetch_db]
    inputs:
      code: |
        api_data = steps["fetch_api"]["result"]
        db_data = steps["fetch_db"]["result"]
        result = {"merged": True}

监视模式

开发期间在文件发生更改时自动重新运行：

yaml-workflow run workflows/hello_world.yaml name=Alice --watch

监视工作流文件及所有导入的文件。按 Ctrl+C 停止。

GitHub Actions

使用 yaml-workflow action 在 CI 中运行工作流：

- name: Run pipeline
  uses: orieg/[email protected]
  id: pipeline
  with:
    workflow: workflows/deploy.yaml
    params: |
      env=production
      version=1.2.0
    format: json

- name: Use results
  run: echo '${{ steps.pipeline.outputs.result }}'

Docker 和 Kubernetes

无需安装 Python，即可在任何地方运行：

# Run a workflow in Docker
docker run --rm -v $(pwd)/workflows:/app/workflows \
  ghcr.io/orieg/yaml-workflow run /app/workflows/pipeline.yaml

# Start the web dashboard
docker run -p 8080:8080 -v $(pwd)/workflows:/app/workflows \
  ghcr.io/orieg/yaml-workflow

使用 Helm chart 部署到 Kubernetes：

helm install my-workflows ./helm/yaml-workflow \
  --set-file workflows.files.pipeline\\.yaml=workflows/pipeline.yaml

兼容 ArgoCD（GitOps）和 Argo Workflows。请参阅 Kubernetes 指南。

更多命令

# List available workflows
yaml-workflow list

# Validate a workflow (with JSON output for CI)
yaml-workflow validate workflows/hello_world.yaml --format json

# Resume a failed workflow
yaml-workflow run workflows/hello_world.yaml --resume

# Structured output for scripting
yaml-workflow run workflows/pipeline.yaml --format json --output results.json

文档

完整文档请访问 orieg.github.io/yaml-workflow。

• 快速入门 - 安装并创建第一个工作流
• 任务类型 - Shell、Python、文件、模板和批处理任务
• 工作流结构 - YAML 配置参考
• 模板化 - Jinja2 变量替换
• 状态管理 - 持久化与恢复
• 任务开发 - 创建自定义任务
• API 参考 - 完整的 API 文档

参与贡献

欢迎贡献！有关开发环境设置和指南，请参阅贡献指南。

许可证

本项目采用 MIT 许可证。详情请参阅 LICENSE 文件。