跳到主要内容

04 · 迁移、执行与 AI 搭建

本章收尾三个相关主题:旧工作流怎么自动升级到新结构;这棵树在运行时怎么被执行;以及 AI 如何用和 GUI 完全相同的操作从零搭一条流。

4.1 schema 迁移:让老流跟上新结构​

构建器在演进,数据结构会变(比如早期的 BRANCH 步骤后来被更通用的 ROUTER 取代)。但数据库里躺着大量旧版本的流,不能让它们一夜失效。Activepieces 用"版本号 + 顺序迁移"解决。

每个 FlowVersion 带 schemaVersion,最新是 '22'(flow-version.ts:7)。读取时若落后就升级。flowMigrations.apply(migrations/index.ts:64-72):

// 真实源码节选 migrations/index.ts:64-72 —— 顺序爬升:命中当前版本就跑该迁移
apply: async (flowVersion, context) => {
  for (const migration of migrations) {
    if (flowVersion.schemaVersion === migration.targetSchemaVersion) {
      flowVersion = await migration.migrate(flowVersion, context)   // 升一级
    }
  }
  return flowVersion
}

migrations 是一个有序数组(migrations/index.ts:38-61),23 个迁移依次排好。每个迁移声明它"从哪个版本升"(targetSchemaVersion)并把流改写成下一版。一个落后很多版的流会被逐级爬升到最新。

何时触发? 在读取时惰性迁移——findOne 读完一行就调 flowVersionMigrationService.migrate(flow-version.service.ts:307-313)。该服务(flow-version-migration.service.ts:12-47)做三件事:① 已是最新版直接返回;② 迁移前把旧版本备份进 backupFiles(flow-version-migration.service.ts:20-23,可回滚);③ 迁移后把新结构写回库并更新 schemaVersion。迁移失败会触发 on-call 报警(flow-version-migration.service.ts:28-35)——可见这条路径被当作高危。

迁移也用同一个遍历器。 结构型迁移 migrateBranchToRouter(migrations/migrate-v0-branch-to-router.ts:6-41)就是给 transferFlow 喂一个"把 BRANCH 节点改写成 ROUTER 节点"的函数:

// 真实源码节选 migrate-v0-branch-to-router.ts —— 老 BRANCH → 新 ROUTER
flowStructureUtil.transferFlow(flowVersion, (step) => {
  if (step.type === 'BRANCH') {
    return { ...转成 ROUTER...,
      settings: { branches: [
        { branchName: 'Branch 1', conditions: step.settings.conditions, branchType: 'CONDITION' },
        { branchName: 'Otherwise', branchType: 'FALLBACK' },
      ], executionType: 'EXECUTE_FIRST_MATCH' },
      children: [ step.onSuccessAction ?? null, step.onFailureAction ?? null ],   // 老的成功/失败子树 → children
    }
  }
  return step
})

旧的 onSuccessAction/onFailureAction 被搬进新的 children 数组——这正好印证 01 章讲的 router 数据结构。同一个 transferFlow 既给操作引擎用、又给迁移用。

4.2 引擎:运行时怎么走这棵树​

发布后真正执行,在引擎包。flowExecutor.execute(packages/server/engine/src/lib/handler/flow-executor.ts:68-128)的主循环就是沿链表走:

// 真实源码节选 flow-executor.ts:79-116 —— while 沿 nextAction 走,skip 跳过
while (!isNil(currentAction)) {
  if (currentAction.skip && !testSingleStepMode) { currentAction = currentAction.nextAction; continue }
  const handler = this.getExecutorForAction(currentAction.type)   // 按类型分派
  flowExecutionContext = await handler.handle({ action: currentAction, executionState, constants })
  // …失败分支、日志大小检查…
  currentAction = currentAction.nextAction                       // 走下一个
  if (shouldBreak) break
}

注意它从 trigger.nextAction 开始(flow-executor.ts:62-66),和构建器是同一条链表——构建时怎么串,运行时就怎么走。getExecutorForAction(flow-executor.ts:16-23)按 FlowActionType 把每个节点分派给对应执行器(code / piece / loop / router)。

子树由对应执行器递归。 router 执行器 routerExecuter(handler/router-executor.ts:11-52)算出哪些分支命中后,对命中分支调 flowExecutor.execute({ action: action.children[i], ... })(router-executor.ts:98-102)——递归进子链表。EXECUTE_FIRST_MATCH 命中第一个就停,EXECUTE_ALL_MATCH 跑完所有命中(router-executor.ts:104)。loop 执行器同理递归 firstLoopAction。

条件求值 evaluateConditions(router-executor.ts:124-258)兑现了 01 章说的二维语义:外层数组之间 OR、内层之间 AND,逐个按 BranchOperator 比较。

4.3 AI 搭流:MCP 用的是同一套操作​

这是"agent 友好"的关键落点。Activepieces 把"建流"也暴露成 MCP 工具,让 Claude/Cursor 等能直接搭工作流。但它没有给 AI 开一条绕过校验的后门——AI 走的是和 GUI 完全相同的操作管线。

看 apBuildFlowTool(packages/server/api/src/app/mcp/tools/ap-build-flow.ts:40-...)。它接受一个高层 spec(一个 trigger + 一个 steps 数组),然后:

// 真实源码节选 ap-build-flow.ts —— AI 的高层 spec → 同一串 FlowOperation
let currentFlow = await flowService(log).update({
  operation: { type: FlowOperationType.UPDATE_TRIGGER, request: triggerPayload },   // ① 先设触发器
})
for (const step of steps) {
  currentFlow = await flowService(log).update({
    operation: { type: FlowOperationType.ADD_ACTION, request: {...} },              // ② 逐步 ADD_ACTION
  })
}

AI 提供的 parentStepName + stepLocationRelativeToParent(ap-build-flow.ts:22-27)正是 02 章的 StepLocationRelativeToParent —— AI 用的插入位置词汇和人拖拽用的是同一套。这些 update 最终都汇入 flowVersionService.applyOperation → flowOperations.apply(03 章那条权威管线),所以 AI 搭的流和人搭的流走完全一样的校验和落库。

工具描述里还有个诚实的边界(ap-build-flow.ts 工具 description):ROUTER 不支持在这一个调用里建——因为分支和条件没法在一次调用里配全,要先把流建好再用 ap_add_step + ap_add_branch 单独配。这说明 AI 接口是"基本操作的薄封装",复杂结构仍走多步操作,而非另起炉灶。

4.4 巧妙之处​

  • 惰性、逐级、可回滚的迁移。 不做"全库批量迁移"这种危险动作;读到才升,升一级写回一级,升级前备份旧版进 backupFiles,失败报警。老流永不失效又不必停机迁移。
  • 迁移复用操作引擎的遍历器。 改结构和改内容用同一个 transferFlow,迁移作者只写"单节点怎么改",递归交给框架。
  • 构建与执行共享同一条链表。 nextAction/children/firstLoopAction 既是编辑时的结构,也是执行时的遍历路径——没有"编译成另一种运行时格式"这一步,所见即所跑。
  • AI 与人共用一套编辑语义。 MCP 工具把高层意图翻译成 FlowOperationRequest,复用全部校验/迁移/落库逻辑。新增的 AI 能力天然继承既有的正确性保证。

4.5 边界与局限​

  • 复制/粘贴的引用重命名有已知 bug。 replaceOldStepNameWithNewOne 的惰性正则会截断含 }} 的 token(02 章 §2.7,add-action-util.ts:32-36 注释自陈),修复被推迟。
  • MCP 一次性建流不支持 ROUTER(§4.3),分支结构必须事后多步配。
  • 画布布局是启发式的固定算法,不是通用图布局——对极深嵌套/超多并排分支,间距是按经验常量算的(flow-canvas-util.ts:5-11),不保证全局最优美观。
  • 迁移失败是硬失败(抛错 + 报警,flow-version-migration.service.ts:35),意味着一个写坏的迁移会挡住对应流的读取,需要谨慎编写。

4.6 横向对比(同 shelf)​

Activepieces 在 ai-agent-reference 货架属于 workflow-builder 这一类"可视化编排 + 可被 AI 驱动"的产品。它的取舍很有辨识度:

  • "操作即数据" 的纯 reducer 模型,让 GUI 与 AI(MCP)天然复用同一套编辑语义——很多编排工具的 AI 接口是另写一层,容易和手工编辑路径分叉。
  • 结构即执行路径(无单独的运行时编译格式),换来"所见即所跑"和简单的序列化,代价是需要一个独立的布局算法把树画成图。
  • 递归树 + 单一遍历器,把增删改、读取、迁移、布局统一到"对树做一次函数式 transfer",是它代码组织上最值得借鉴的一点。

4.7 代码地图​

主题文件符号
迁移注册表 + 顺序爬升packages/server/api/src/app/flows/flow-version/migrations/index.tsflowMigrations(apply)、migrations、Migration
读取时惰性迁移 + 备份packages/server/api/src/app/flows/flow-version/flow-version-migration.service.tsflowVersionMigrationService(migrate)
结构迁移示例(BRANCH→ROUTER)packages/server/api/src/app/flows/flow-version/migrations/migrate-v0-branch-to-router.tsmigrateBranchToRouter
引擎主循环(走链表)packages/server/engine/src/lib/handler/flow-executor.tsflowExecutor(execute、executeFromTrigger、getExecutorForAction)
router 执行 + 条件求值packages/server/engine/src/lib/handler/router-executor.tsrouterExecuter、evaluateConditions
MCP 建流工具packages/server/api/src/app/mcp/tools/ap-build-flow.tsapBuildFlowTool