DAG Workflow Engine

DAG 驱动的 AI Agent 工作流引擎 — YAML 定义阶段、智能 revision 边、Agent handoff、模型 fallback。

A DAG-driven workflow engine for AI agent orchestration — YAML-defined phases, smart revision edges, agent handoffs, and model fallback.

---

特性 / Features

• YAML 驱动 — 在一个文件中声明阶段、边和 revision 谓词
• 零代码接入 — 只需写 workflow YAML + 可选的 dag.config.yaml，无需写 Python
• 配置驱动 — outcome_overrides、recovery_detectors、min_summary_length 全部在 YAML 中声明
• 步进式执行 — CLI 驱动：init → dispatch → result → done，无 daemon
• 智能 revision 边 — 基于条件的路由（如"review 说有 bug → 路由回 coding"）
• Agent handoff — JSON 格式阶段间数据传递，支持 schema 校验
• 模型 fallback — API 错误时自动切换模型，指数退避
• Checkpoint 审批 — human_checkpoint: true 暂停流水线等待人工批准
• 外部状态恢复 — 检测 PR 合并、Issue 关闭、commit 存在以自动恢复死节点
• 插件架构 — 通过 RunnerPlugin 扩展 runner 行为
• 多平台 — Forgejo、GitHub、GitLab（通过 GitProvider 协议）
• 零外部依赖 — 仅 stdlib + PyYAML

---

快速开始 / Quick Start

安装 / Install

pip install dag-workflow-engine
# 或直接使用
python -m dag_engine.runner --help

写一个 workflow / Define a workflow

# workflows/my-pipeline.yaml
name: my-pipeline
description: "3 阶段流水线：设计 → 实现 → 评审"

phases:
  - id: design
    agent: "subagent:plan"
    description: "架构设计"
    human_checkpoint: true
    timeout: 1200

  - id: implement
    agent: "subagent:executor"
    description: "代码实现"
    workdir: worktree
    timeout: 1800

  - id: review
    agent: "subagent:code-reviewer"
    description: "代码评审"
    min_summary_length: 200
    timeout: 900

edges:
  - from: design
    to: implement
  - from: implement
    to: review

revision_edges:
  - from: review
    to: implement
    predicate: blocked
    condition: "bug|fix|error|issue"

outcome_overrides:
  - phase: review
    outcome: blocked
    action: revision

运行 / Run

# 初始化一个 run
dag-engine init --workflow my-pipeline --issue 42
# → {"run_id": "abc-123456", ...}

# 查看就绪节点
dag-engine ready --run-id abc-123456
# → {"nodes": [{"id": "design", ...}]}

# 提交结果（agent 完成后）
dag-engine result --run-id abc-123456 --node-id design \
  --success true --outcome completed --summary "设计完成"

# 自动 dispatch 下一个节点
dag-engine dispatch --run-id abc-123456
# → {"action": "dispatch", "nodes": [{"id": "implement", ...}]}

---

项目配置 / Project Config（可选）

在项目根目录创建 dag.config.yaml，引擎自动发现并加载：

git_provider:
  type: forgejo                    # forgejo | github | gitlab | noop
  url: "${FORGEJO_URL}"           # 支持 ${ENV_VAR} 引用
  owner: "${FORGEJO_OWNER}"
  repo: "${FORGEJO_REPO}"
  token: "${FORGEJO_TOKEN}"

plugins:
  - module: my_project.dag_plugin
    class: CustomPlugin

---

默认工作流 / Default Workflow

引擎自带 default-dev.yaml — 适用于大多数开发任务的 4 阶段循环：

design → implement → test → review
         ↑___________|      |
         ↑__________________|  (revision edges)

dag-engine init --workflow default-dev --issue 42

---

架构 / Architecture

dag_engine/
├── __init__.py          # Public API exports
├── types.py             # DagNode, DagEdge, PipelineGraph, NodeResult, NodeState
├── builder.py           # build_graph(workflow_def) → PipelineGraph
├── config.py            # Configuration loader — workflow YAML + dag.config.yaml
├── executor.py          # apply_result(), find_ready_nodes(), execute_graph()
├── store.py             # WorkflowStore: JSON state persistence
├── handoff.py           # HandoffManager: agent-to-agent data passing
├── outcome_table.py     # (phase, outcome) → OutcomeAction matrix
├── fallback.py          # Model fallback chain on API errors
├── recovery.py          # Recovery decision + external state detection
├── registry.py          # RunRegistry: multi-run management
├── validators.py        # Summary verifiability validation
│
├── providers/           # Git platform abstraction
│   ├── base.py          # GitProvider Protocol
│   ├── forgejo.py       # Forgejo/Gitea implementation
│   ├── github.py        # GitHub implementation (skeleton)
│   └── gitlab.py        # GitLab implementation (skeleton)
│
├── dispatchers/         # Agent invocation abstraction
│   ├── base.py          # AgentDispatcher Protocol
│   ├── subagent.py      # SubagentDispatcher
│   ├── skill.py         # SkillDispatcher
│   ├── bash.py          # BashDispatcher
│   └── noop.py          # NoopDispatcher (testing)
│
└── runner_plugins/      # Runner behavior extension
    └── __init__.py      # RunnerPlugin base class

---

CLI 命令 / CLI Commands

Command	说明 / Description
dag-engine init	初始化 DAG run / Initialize a new run
dag-engine ready	列出就绪节点 / List ready nodes
dag-engine dispatch	获取调度指令 / Get dispatch instructions
dag-engine result	提交节点结果 / Submit a node result
dag-engine status	查看 run 状态 / Show run status
dag-engine flat	扁平 JSON 输出 / Flat JSON output
dag-engine cancel	取消节点 / Cancel a node
dag-engine pause	暂停节点 / Pause a node
dag-engine resume	恢复节点 / Resume a node
dag-engine approve	批准 checkpoint / Approve checkpoint
dag-engine unblock	解除阻塞 / Unblock a node
dag-engine progress	上报进度 / Report progress
dag-engine list-active	列出活跃 run / List active runs

---

工作流 YAML 格式 / Workflow YAML Format

name: <workflow-name>
description: "<描述>"

phases:
  - id: <phase-id>
    agent: "<type>:<name>"            # subagent:Name | skill:name | bash:command
    description: "<阶段描述>"
    timeout: <seconds>
    workdir: worktree | main          # worktree = 代码修改，main = 评审/审批
    human_checkpoint: true | false    # 是否需要人工审批
    min_summary_length: <N>           # 摘要最短长度
    handoff:
      write: [field1, field2]         # 本阶段产出的字段
      read: [field1]                  # 本阶段消费的字段

edges:
  - from: <phase-a>
    to: <phase-b>

revision_edges:
  - from: <review-phase>
    to: <code-phase>
    predicate: blocked
    condition: "bug|fix|error|issue|security"

outcome_overrides:
  - phase: <phase>
    outcome: blocked
    action: advance | retry | revision | abort

recovery_detectors:
  - node_id: <phase>
    type: provider | git
    method: detect_pr_merged | detect_pr_exists | detect_issue_closed
    requires: [pr_number]

max_revision_cycles: 3

---

环境变量 / Environment Variables

变量 / Variable	默认值 / Default	说明 / Description
DAG_STORE_DIR	.dag-store	状态存储目录
DAG_HANDOFF_DIR	.dag-store/handoffs	Handoff 文件目录
DAG_CONFIG	—	dag.config.yaml 路径
FORGEJO_URL	—	Forgejo 实例 URL
FORGEJO_TOKEN	—	Forgejo API token

---

插件 / Plugins

通过 RunnerPlugin 扩展 runner 行为：

from dag_engine.runner_plugins import RunnerPlugin, register_plugin

class MyPlugin(RunnerPlugin):
    def pre_result(self, args, graph, node, result, store):
        """结果应用前调用 — 可修改 result"""
        pass

    def on_result(self, args, graph, node, result, store):
        """结果应用后调用 — 用于副作用"""
        pass

    def build_dispatch_prompt(self, node, task_prompt, upstream):
        """自定义 dispatch prompt"""
        return task_prompt

register_plugin(MyPlugin())

---

Claude Code 集成 / Claude Code Integration

PostToolUse Hook

在 .claude/settings.json 中配置：

{
  "hooks": {
    "PostToolUse": [
      {
        "matcher": "Agent",
        "hooks": [
          {
            "type": "command",
            "command": "bash /path/to/scripts/dag-auto-result.sh",
            "timeout": 30
          }
        ]
      }
    ]
  }
}

MCP Server

{
  "mcpServers": {
    "dag-engine": {
      "command": "python3",
      "args": ["/path/to/scripts/mcp/dag_server.py"],
      "env": {"DAG_STORE_DIR": ".dag-store"}
    }
  }
}

注册 tools：dag_init、dag_dispatch、dag_result、dag_status、dag_ready、dag_approve、dag_cancel、dag_list_active

详见 docs/claude-code-integration.md

---

测试 / Tests

python -m pytest dag_engine/tests/ -v
# 530 passed, 50 skipped

---

许可证 / License

MIT