Agent Workflow Orchestrator v2.0

AI Agent 原生工作流编排器

定位对比

独特价值

用户需求 → 自然语言 → AI生成DAG → 串行/并行/条件执行 → 技能调用 → 结果输出
                      ↓
              数据清洗 → 数据转换 → 数据聚合 → 可视化
                              ↓
                        [调用数据分析技能]

快速开始

1. 执行工作流

# 从JSON文件执行
python workflow_cli.py run workflow.json

# 并行模式执行
python workflow_cli.py run workflow.json --mode parallel

# 仅验证
python workflow_cli.py run workflow.json --validate-only

# 模拟运行
python workflow_cli.py run workflow.json --dry-run

2. 从自然语言生成

python workflow_cli.py generate \
  "从sales.csv读取数据，过滤status=completed，按category聚合求和，输出到summary.csv"

3. 生成可视化

python workflow_cli.py visualize workflow.json -o workflow.html

DAG 定义格式

{
  "workflow_id": "my_workflow",
  "name": "我的工作流",
  "version": "2.0.0",
  "execution_mode": "auto",
  "on_failure": "stop",
  
  "nodes": [
    {
      "id": "node_1",
      "type": "DataIn",
      "name": "读取数据",
      "config": {
        "path": "data/sales.csv",
        "format": "csv"
      }
    },
    {
      "id": "node_2",
      "type": "Filter",
      "name": "过滤数据",
      "depends_on": ["node_1"],
      "config": {
        "conditions": [
          {"field": "status", "operator": "==", "value": "completed"}
        ],
        "logic": "and"
      }
    },
    {
      "id": "node_3",
      "type": "DataOut",
      "name": "输出数据",
      "depends_on": ["node_2"],
      "config": {
        "path": "output/summary.csv"
      }
    }
  ],
  
  "edges": [
    {"from": "node_1", "to": "node_2"},
    {"from": "node_2", "to": "node_3"}
  ]
}

节点类型

执行模式

1. 串行流水线 (sequential)

按拓扑排序顺序依次执行每个节点。

{
  "execution_mode": "sequential"
}

2. 并行协作 (parallel)

无依赖的节点并发执行。

{
  "execution_mode": "parallel"
}

3. 自动模式 (auto)

根据DAG结构自动选择最优执行方式：

• 存在条件分支 → 条件路由
• 存在并行机会 → 并行执行
• 简单链式 → 串行执行

{
  "execution_mode": "auto"
}

条件路由

支持 if-else 分支，根据上游节点输出决定执行路径。

{
  "nodes": [
    {"id": "check", "type": "DataIn", "name": "数据检查"},
    {"id": "branch", "type": "Branch", "depends_on": ["check"]},
    {"id": "process_valid", "type": "Filter"},
    {"id": "process_invalid", "type": "Filter"}
  ],
  "edges": [
    {"from": "check", "to": "branch"},
    {"from": "branch", "to": "process_valid", "condition": true},
    {"from": "branch", "to": "process_invalid", "condition": false}
  ]
}

技能调用

支持调用 OpenClaw 平台的技能。

{
  "id": "analyze",
  "type": "Skill",
  "name": "数据分析",
  "skill": "smart-data-analyzer",
  "config": {
    "query": "SELECT * FROM sales"
  }
}

支持的技能

安全机制

1. SQL白名单

# 允许的关键字
ALLOWED_KEYWORDS = {'SELECT', 'INSERT', 'UPDATE', 'CREATE', 'ALTER'}

# 禁止的关键字
FORBIDDEN_KEYWORDS = {'DROP', 'DELETE', 'TRUNCATE', 'EXEC', 'GRANT'}

2. 路径校验

禁止路径遍历攻击：

• ../ 模式
• URL编码的遍历
• 符号链接到敏感目录

3. 资源限制

API 使用

Python API

from dag_runner import DAGRunner, ExecutionMode
from node_executor import NodeExecutor

# 创建执行器
executor = NodeExecutor()

# 加载工作流
runner = DAGRunner(workflow=workflow, execution_mode=ExecutionMode.AUTO)

# 注册节点执行器
for node_type in executor.list_types():
    runner.register_executor(node_type, executor.execute)

# 执行
result = runner.run()

print(result['status'])
print(result['summary'])

AI生成

from ai_dag_generator import AIDAGGenerator

generator = AIDAGGenerator()
workflow = generator.generate("从sales.csv读取数据，过滤status=completed，按category聚合")

print(workflow)

可视化

from dag_visualizer import DAGVisualizer

visualizer = DAGVisualizer()
html = visualizer.render(workflow)

with open('workflow.html', 'w') as f:
    f.write(html)

示例工作流

数据ETL

{
  "workflow_id": "etl_pipeline",
  "name": "数据ETL流程",
  "execution_mode": "parallel",
  "nodes": [
    {
      "id": "load_sales",
      "type": "DataIn",
      "name": "读取销售数据",
      "config": {"path": "data/sales.csv"}
    },
    {
      "id": "load_products",
      "type": "DataIn",
      "name": "读取产品数据",
      "config": {"path": "data/products.csv"}
    },
    {
      "id": "join_data",
      "type": "Join",
      "name": "关联数据",
      "depends_on": ["load_sales", "load_products"],
      "config": {"left_key": "product_id", "right_key": "id", "join_type": "left"}
    },
    {
      "id": "aggregate",
      "type": "Aggregate",
      "name": "按产品聚合",
      "depends_on": ["join_data"],
      "config": {
        "group_by": ["product_name"],
        "aggregations": [
          {"field": "amount", "func": "sum", "alias": "total_sales"},
          {"field": "quantity", "func": "count", "alias": "order_count"}
        ]
      }
    },
    {
      "id": "output",
      "type": "DataOut",
      "name": "输出结果",
      "depends_on": ["aggregate"],
      "config": {"path": "output/summary.csv"}
    }
  ]
}

条件分支

{
  "workflow_id": "conditional_flow",
  "name": "条件分支流程",
  "execution_mode": "auto",
  "nodes": [
    {"id": "start", "type": "DataIn", "name": "读取数据"},
    {"id": "check", "type": "Branch", "depends_on": ["start"]},
    {"id": "path_a", "type": "Filter", "name": "路径A"},
    {"id": "path_b", "type": "Filter", "name": "路径B"},
    {"id": "merge", "type": "Join", "depends_on": ["path_a", "path_b"]}
  ],
  "edges": [
    {"from": "start", "to": "check"},
    {"from": "check", "to": "path_a", "condition": true},
    {"from": "check", "to": "path_b", "condition": false},
    {"from": "path_a", "to": "merge"},
    {"from": "path_b", "to": "merge"}
  ]
}

目录结构

agent-workflow-orchestrator/
├── SKILL.md                    # 本文档
├── scripts/
│   ├── __init__.py
│   ├── dag_runner.py           # DAG执行引擎
│   ├── node_executor.py        # 节点执行器
│   ├── skills_executor.py      # 技能调用器
│   ├── ai_dag_generator.py     # AI DAG生成
│   ├── lineage_tracker.py      # 数据血缘追踪
│   ├── dag_visualizer.py       # DAG可视化
│   ├── security_guard.py       # 安全守卫
│   ├── state_manager.py        # 状态管理
│   ├── config_exporter.py      # 配置导出
│   └── workflow_cli.py         # CLI入口
├── references/
│   ├── workflows/             # 工作流示例
│   └── node_templates/         # 节点模板
└── tests/
    ├── test_dag_runner.py
    ├── test_node_executor.py
    ├── test_skills_executor.py
    ├── test_security_guard.py
    └── test_dag_visualizer.py

版本信息

• 版本: 2.0.0
• 更新: 2026-04-27
• 作者: captain_ai