Workflow

工作流用 DAG 描述多个能力如何协作：节点是能力，边是依赖。平台负责校验、发布、执行和记录。

---

职责

负责	不负责
DAG 定义、校验、版本与发布	单个 Agent 的执行细节（agent-gateway / a2a）
拓扑排序与节点调度	需求翻译（demand-translation）
节点输入输出传递	任务生命周期（task）
执行记录与状态	验收与结算（fulfillment / escrow）
工作流模板	商业规则

---

数据模型

workflows

{
  "workflowId": "wf_...",
  "name": "德国市场合规",
  "status": "draft | active | paused",
  "nodes": [
    {
      "id": "n1",
      "type": "agent",
      "agentId": "platform-demand-translation",
      "skillId": "demand-translation",
      "config": {
        "stateWrites": [],
        "stateReads": []
      }
    }
  ],
  "connections": [
    { "from": "n1", "to": "n2" }
  ],
  "createdBy": "user:...",
  "version": 1
}

workflow_executions

记录一次执行的节点状态、输入输出、耗时与错误。

workflow_templates

可复用的工作流模板，支持按场景快速实例化。

---

服务结构

服务	职责
WorkflowService	CRUD、发布、暂停
WorkflowValidatorService	DAG 校验（环、孤立节点、Schema 兼容）
WorkflowExecutorService	节点执行与调度
WorkflowInstanceService	执行实例管理
WorkflowDataFlowService	节点间数据传递与映射
WorkflowExecutionStrategyService	执行策略（串行 / 并行）
WorkflowTemplateService	模板管理

---

接口

方法	路径	说明
POST	/api/workflow/create	创建工作流
GET	/api/workflow/:workflowId	详情
GET	/api/workflow	列表
PUT	/api/workflow/:workflowId	更新
POST	/api/workflow/:workflowId/publish	发布
POST	/api/workflow/:workflowId/pause	暂停
DELETE	/api/workflow/:workflowId	删除
POST	/api/workflow/:workflowId/execute	执行
POST	/api/workflow/validate	校验
GET	/api/workflow/:workflowId/executions	执行记录
GET	/api/workflow/executions/:executionId	单次执行详情
POST	/api/workflow/:workflowId/nodes	添加节点
POST	/api/workflow/:workflowId/connections	添加连线

模板端点位于 workflow-template.controller.ts。

---

校验规则

WorkflowValidatorService 在发布前检查：

规则	说明
无环	拓扑排序必须成功
单一入口	至少一个入度为 0 的节点
连接有效	from / to 必须存在
Schema 兼容	上游输出能满足下游输入
Agent 存在	引用的 agentId 必须存在且 active
不重复执行翻译节点	有 _translationOutput 的节点跳过

---

执行模型

execute(workflowId)
  ├─ validate
  ├─ topologicalSort
  ├─ 并行执行所有入度为 0 的节点
  ├─ 节点完成 → 释放下游依赖
  ├─ 汇总终止节点输出
  └─ 写入 execution 记录

节点执行分派

节点类型	执行
内部 Agent	EngineChatService.directPrompt()
外部 Agent	A2A → 失败降级 Gateway
内置节点	直接调用平台服务

数据流

WorkflowDataFlowService 负责把上游输出映射到下游输入：

• 默认按字段名匹配
• 支持在节点 config 中显式声明映射
• 支持读取状态层历史版本（stateReads）

---

旁挂状态

节点可以声明：

{
  "config": {
    "stateWrites": ["/tasks/{taskId}/nodes/{nodeId}"],
    "stateReads": [{ "path": "/tasks/{taskId}/deliveries/*", "as": "deliveries" }]
  }
}

• 未声明 ⇒ 零写入、零读取
• 写入失败只 warn，不影响节点结果
• 读取结果注入 input.stateReads.{as}

设计细节见 状态层。

---

边界与限制

限制	说明
动态 DAG	运行中不支持增删节点
循环	明确禁止
长时间运行	无持久化断点，进程重启会丢失执行上下文
并发上限	依赖节点并行，但缺少队列级限流
版本	发布后的工作流修改会影响新任务，历史执行保留快照

---

源码位置

libs/modules/src/workflow/
├── controllers/
│   ├── workflow.controller.ts
│   └── workflow-template.controller.ts
├── services/
│   ├── workflow.service.ts
│   ├── workflow-validator.service.ts
│   ├── workflow-executor.service.ts
│   ├── workflow-instance.service.ts
│   ├── workflow-dataflow.service.ts
│   ├── workflow-execution-strategy.service.ts
│   └── workflow-template.service.ts
├── schemas/
└── validators/

---

相关页面

• Task
• Agent Gateway
• 状态层
• 一次需求的完整生命周期