DagFlow – Enterprise Workflow Engine

DagFlow 是一个 企业级 DAG 工作流引擎，基于 sqlor + ahserver 构建，支持多组织（org_id）隔离、子流程、人工节点（审批任务）、RBAC 角色分配，并提供清晰的流程实例与任务查询接口，适用于：

• 企业流程编排

• 审批流 / 工单流

• 自动化运维 / AI Agent 编排

• 多租户 SaaS 工作流系统

核心特性

✅ 多租户隔离（org_id）

• 所有核心数据表均包含 org_id

• 流程定义、流程实例、任务实例完全隔离

• 天然支持同一流程在不同组织复用

✅ DAG 工作流引擎

• 基于 YAML DSL 描述流程

• 支持条件边（when）

• 支持并行节点

• 自动推进 / 收敛至 end 节点

✅ 持久化执行（sqlor）

• 流程定义、实例、节点执行、人工任务全部持久化

• 引擎无状态，可水平扩展

• 支持调度器 / worker 分离部署

✅ 人工节点（Human Task）

• human 节点会生成待办任务

• 流程在人工节点阻塞

• 支持：

  • 角色（role）分配

  • 用户领取 / 完成

  • 输出回写流程上下文

✅ 子流程（SubFlow）

• 支持流程嵌套

• 父子流程上下文映射

• 子流程完成后自动回写父流程

表结构说明

flow_definition（流程定义）

flow_instance（流程实例）

node_execution（节点执行记录）

human_task（人工任务）

DSL 示例

yaml
复制代码
start: submit
nodes:
  submit:
    type: normal
  approve:
    type: human
    role: manager
    timeout: 86400
  end:
    type: end

edges:
  - from: submit
    to: approve
  - from: approve
    to: end

核心 API（Engine）

创建流程定义

python
复制代码
await engine.create_definition(
    org_id="org1",
    name="请假审批",
    version="v1",
    dsl_text=dsl_yaml
)

启动流程实例

python
复制代码
instance_id = await engine.create_instance(
    org_id="org1",
    flow_def_id=flow_id,
    ctx={"user": "alice", "days": 3}
)

推进流程（调度器调用）

python
复制代码
await engine.step(org_id="org1", instance_id=instance_id)

人工任务 API

查询待办任务（RBAC）

python
复制代码
tasks = await engine.list_human_tasks(
    org_id="org1",
    user_roles=["manager", "admin"]
)

完成人工任务

python
复制代码
await engine.complete_human_task(
    org_id="org1",
    task_id=task_id,
    user_id="bob",
    output={"approved": True}
)

完成后：

• 任务状态 → done

• 输出自动合并到流程 ctx

• 流程可继续推进

架构设计

less
复制代码
[ HTTP API ]
     |
[ FlowEngine ]
     |
[ sqlor ]
     |
[ MySQL / PostgreSQL / SQLite ]

• Engine 无状态

• Scheduler 可多实例

• 支持分布式部署

适用场景

• 企业审批流（请假 / 报销 / 发布）

• 自动化运维流程

• AI Agent 调度与编排

• SaaS 多租户流程平台

后续可扩展方向

• ✔ 乐观锁 / version 防并发

• ✔ 并行网关 / 排他网关

• ✔ 任务转派 / 会签

• ✔ SLA / 超时补偿

• ✔ 流程可视化建模器