任务执行引擎:Temporal 工作流与状态机(主线)
30 秒导读: Julep 里一个 Task(工作流定义)被跑起来,靠的就是本章讲的引擎。它把 "执行一个任务" 拆成 "一步接一步",每一步都是一个独立的 Temporal 工作流 run;跑完一步就 用
continue_as_child/ continue-as-new 把 "下一步" 当成新 run 启动。这样做能让崩溃后可恢复、 让 Temporal 的历史记录始终有界。这是全库最核心的一章——引擎怎么驱动一次执行。
本章聚焦 "引擎怎么转",不逐一展开各个步骤类型的语义(那是
03 工作流步骤类型与 $ 表达式求值 的活)。想先看 Task/Execution
这些概念是什么,请回 01 核心概念与数据模型;想看它在整个平台里
的位置,见 06 平台架构。
1. 这是什么(零基础也能懂)
一句话定义: 任务执行引擎 = 一台 "把 Task 定义里一串步骤,可靠地一步步跑完" 的状态机, 底座是 Temporal(一个专做 "持久化工作流" 的开源引擎)。
它解决什么问题。 一个 Agent 任务往往很长:调 LLM、等工具返回、循环处理一批数据、 甚至暂停几天等人给输入。中途进程崩了、机器重启了怎么办?你不想 "从头再来一遍"(LLM 调用 很贵),更不想丢状态。引擎要保证:跑到哪就记到哪,崩了能从断点续上,而且过程可查、可审计。
Temporal 是什么(一句话): 你把业务逻辑写成一个普通的 async 函数(叫 "工作流"), Temporal 会记录这个函数 "做过的每一次外部调用(activity)及其结果";进程挂掉重启后,它重放 这段历史、跳过已完成的调用,让函数从上次的位置继续——你的代码看起来像没崩过一样。这叫 持久化执行(durable execution)。
在 Julep 里的两个关键角色:
| 名字 | 白话 | 它是谁 |
|---|---|---|
| Execution | 一次 "任务在跑" 的运行时实例,带状态(queued/running/…) | 数据库里的一行 + 一个 Temporal 工作流 |
| Transition(转移) | 执行 "从这步走到那步" 的一条日志记录 | 数据库里的一行,append-only |
一句话直觉/类比: 把它想成一台 带存档点的游戏机。每走一步(打一个关卡)就自动存一次档 (写一条 transition 到库);存完档,这一关的机器 "关掉重开",换成 "下一关" 的新进程继续跑 (continue-as-new)。任何时候断电,重开都能从最近的档接上。
2. 顶层全景(它大概怎么转)
2.1 一次 execution 从 API 到跑完
怎么读这张图: 从上到下是时间顺序。左边是 API 进程(FastAPI),右边是 Temporal Worker 进程。API 只负责 "落库 + 按下启动键",真正的执行在 Worker 里循环。
API 进程 (routers/tasks) Temporal Worker 进程
───────────────────────── ──────────────────────────────
POST /tasks/{id}/executions
│
▼
① 校验输入 schema / 免费额度
│
▼
② start_execution()
├─ create_execution → 写 executions 表(状态 queued)
├─ prepare_execution_input → 拼出 ExecutionInput(agent/task/tools/args)
└ ─ run_task_execution_workflow ─────────┐
▼
③ client.start_workflow(
TaskExecutionWorkflow.run,
args=[exec_input, start, current_input])
│ (排进 task_queue)
◀──────── 返回 WorkflowHandle ───────────┤
│ ▼
④ 后台 create_temporal_lookup ⑤ TaskExecutionWorkflow.run() 主循环:
(execution_id ↔ workflow_id) 对"当前这一步"跑一遍:
│ transition(init)
▼ → 执行 activity / 求值表达式
返回 Execution(201) → handle_step 决定 去向
→ transition(step)
→ continue_as_child(下一步) ⟳
│
▼
最后一步:transition(finish) → 结束
部件一句话职责:
| 部件 | 干什么 | 在哪个文件 |
|---|---|---|
create_task_execution 路由 | HTTP 入口:校验、限额、调用 start_execution | routers/tasks/create_task_execution.py:124 |
start_execution | 落库 + 备料 + 启动工作流,返回 (Execution, handle) | routers/tasks/create_task_execution.py:46 |
prepare_execution_input | 一条 SQL 把 agent/task/tools/execution 拼成 ExecutionInput | queries/executions/prepare_execution_input.py |
run_task_execution_workflow | 连 Temporal、启动 TaskExecutionWorkflow.run | clients/temporal.py:98 |
TaskExecutionWorkflow | 主循环:跑一步 → 决定去向 → 转下一步 | workflows/task_execution/__init__.py:130 |
transition() | 把每次状态转移写库(= 执行日志) | workflows/task_execution/transition.py:22 |
continue_as_child | 把 "下一步" 跑成新 run(continue-as-new / 子工作流) | workflows/task_execution/helpers.py:77 |
| 状态机枚举 | 合法转移、状态映射 | common/protocol/state_machine.py、common/protocol/tasks.py:71 |
2.2 主线走一遍(高层,不进代码)
- API 侧:
create_task_execution校验输入符合 task 的input_schema、检查免费额度, 然后调start_execution(create_task_execution.py:138、:170)。 start_execution先在库里建一行 execution(状态queued),再prepare_execution_input把这次运行需要的一切(agent、task spec、tools、入参)装进一个ExecutionInput(create_task_execution.py:55-76)。- 调
run_task_execution_workflow连上 Temporal,client.start_workflow(...)把TaskExecutionWorkflow.run排进任务队列——API 到此为止,不等它跑完,立刻拿到一个WorkflowHandle返回 201(temporal.py:128、create_task_execution.py:184)。 - Worker 进程从队列取到这个工作流,进入
run()主循环:准备上下文 → 写一条init转移 → 执行当前步 → 决定下一步 → 写一条step转移 → 用 continue-as-new 把 "下一步" 变成新 run。 - 走到最后一步时写一条
finish转移,整条执行succeeded;中途出错/被取消则写error/cancelled并终止。
3. 核心原理(逐个机制,由浅入深)
3.1 主循环:一步 = 一个工作流 run
它要解决的小问题: 怎么把 "顺序执行 N 个步骤" 写成一个既能崩溃恢复、历史又不会无限膨胀的 Temporal 工作流?
思路/直觉。 天真做法是写一个 for step in steps: run(step) 的大工作流。但 Temporal 会把
工作流里每一次 activity 调用和结果都记进 "history";任务一长,history 会膨胀到几万条事件,
拖慢重放、逼近上限。Julep 的答卷是:一个工作流 run 只负责跑好当前这一步,跑完就通过
continue-as-new 开一个 "新 run" 去跑下一步——新 run 的 history 从零开始,于是总历史始终有界。
这是本引擎最关键的设计决策。
这是什么样子(简化,# 示意,非源码):
# 演示"一步一个 run"的骨架。重点看:每次只处理 current_step,末尾 continue-as-new。
@workflow.defn(sandboxed=False)
class TaskExecutionWorkflow:
async def run(self, exec_input, cursor, current_input):
ctx = StepContext(exec_input, cursor, current_input) # 我在第几步
if ctx.is_first_step: # 头一步先盖个"init"章
await transition(ctx, type="init", next=ctx.cursor)
outcome = await run_current_step(ctx) # 跑这一步(activity 或表达式)
result = await handle_step(ctx, outcome) # 这一步想去哪?
next_target = await transition(ctx, result.state) # 写"step"转移,拿到下一步坐标
if next_target.is_terminal: # 已经是最后一步 → 收工
return result
# 关键:把"下一步"跑成一个新的 workflow run,history 归零
return await continue_as_child(exec_input, start=next_target.next, current_input=result.output)
真实实现。 主循环是 TaskExecutionWorkflow.run(),声明为
@workflow.defn(sandboxed=False)(workflows/task_execution/__init__.py:129、:738)。它的骨架分五段:
| 阶段 | 干什么 | __init__.py 位置 |
|---|---|---|
| 0. 准备上下文 | 用 execution_input + cursor + current_input 建 StepContext | :755 |
| 1. 起手转移 | 若是第一步且非续跑,写 init / init_branch | :765 |
| 2. 执行当前步 | 有 activity 就 execute_activity,否则 eval_step_exprs 求表达式 | :780–:799 |
| 3. 决定去向 | handle_step(current_step) 分派到 _handle_<步类型> | :832 |
| 4. 转移到下一步 | transition(ctx, state) 写库,拿到 final_state | :867 |
| 5. 收尾或递归 | 终态就 return;否则 continue_as_child 跑下一步 | :874、:899 |
第 2 段的 "有没有 activity" 由一张小映射表 STEP_TO_ACTIVITY 决定——目前只有 PromptStep
映射到 prompt_step activity(:108);其余步类型的表达式在工作流内直接经 eval_step_exprs
求值(:142)。
为什么
sandboxed=False? Temporal Python SDK 默认把工作流代码放进沙箱、拦截非确定性调用。 Julep 的步骤逻辑要match各种 Pydantic 模型、动态取属性,关掉沙箱更省心;确定性由 "把副作用 都塞进 activity" 来保证(见 3.4)。(inferred:注释未明说关沙箱的动机,但代码把所有 IO 都走 activity 的写法与此一致。)
3.2 continue-as-new:把 history 保持有界
它要解决的小问题: 上一节说 "一步一个 run",可怎么从当前 run 切到下一步的新 run,还不丢 "这是哪个 execution" 这类上下文?
思路/直觉。 Temporal 提供 continue_as_new:让当前工作流以全新的输入参数、全新的空 history
重启自己,对外仍是同一个逻辑执行。Julep 把它包在 continue_as_child 里,并做了一个选择:
- 如果 Temporal 建议 continue-as-new(
info.is_continue_as_new_suggested())→ 用workflow.continue_as_new(真正归零 history); - 否则 → 用
workflow.execute_child_workflow起一个子工作流跑下一步。
分支/循环步(if-else、foreach、map-reduce)天然要开子工作流(每个分支/每次迭代一个 run),
所以同一个 continue_as_child 既服务 "主线下一步",也服务 "并列子任务"。
真实实现。 continue_as_child 在 helpers.py:77,核心分叉在 :86:
# helpers.py:86 —— 真源码要点(节选)
if info.is_continue_as_new_suggested():
run = workflow.continue_as_new
else:
run = lambda *a, **k: workflow.execute_child_workflow(info.workflow_type, *a, **k)
主 循环末尾的调用把 "下一步坐标 + 上一步输出" 传进去,让新 run 从那里接着跑
(__init__.py:899);注意下一步的 current_input 是先经 save_inputs_remote 存到 blob store
再传的——大对象走远端引用,避免撑爆 Temporal payload(__init__.py:888)。
一个诚实的注脚:
helpers.py:85有条注释# FIXME: This doesn't actually work,暗示 "何时该 continue-as-new" 的判定并不总如预期。但 "每步开新 run 让 history 有界" 的总体设计 意图是明确的,值得读者记住。
3.3 transition():每次转移都写库 = 可恢复的执行日志
它要解决的小问题: 执行状态存哪?怎么让 API 端能查到 "现在跑到第几步、成功了没"?怎么 支持崩溃恢复?
思路/直觉。 引擎不把状态只留在内存里,而是每一次状态变化都往数据库 append 一条 transition。 这条日志是三样东西合一:
- 执行进度——
current(现在在哪步)、next(下一步去哪)。 - 执行状态——
type(init/step/finish/error/…)映射成 execution 的对外状态。 - 中间产物——
output(这一步产出什么),供后续步骤引用。
有了这份 append-only 日志,API 只要读最近一条 transition 就知道 execution 现状;而
StepContext.get_inputs() 更能把同一 scope 下的历史 transition 全捞出来,重建 "前面各步的
输入输出",给表达式求值当上下文(common/protocol/tasks.py:267)。
图示:一步产生一条转移。
run() 主循环 transition() 写库
┌──────────────────┐ state ┌────────────────────────────┐
│ 跑完 current_step │ ─────────▶ │ 组 CreateTransitionRequest │
│ 拿到 PartialTransition │ current = cursor │
└──────────────────┘ │ next = cursor.step + 1 │
│ type = step/finish/... │
│ output = 这步产物 │
└──────────────┬─────────────┘
▼
execute_activity(transition_step)
▼
create_execution_transition → DB
(transitions 表, append-only)
真实实现。 transition() 在 transition.py:22。它先按 "是不是最后一步" 决定 type
(transition.py:30 的 match:main 的最后一步 → finish,子工作流最后一步 → finish_branch,
否则 → step),再算出 next(同 scope 下 step+1,最后一步则 None),打包成
CreateTransitionRequest,通过 execute_activity(task_steps.transition_step) 落库
(transition.py:38、:56)。真正写库的 activity 是 transition_step
(activities/task_steps/transition_step.py),它调 create_execution_transition 把行插进
transitions 表;插入前 SQL 层还会校验这次转移是否合法(见 3.5)。
关键细节:为什么写库要放进 activity。 工作流代码必须是确定性的,不能直接碰数据库;所以 所有 IO(写转移、调 LLM、跑工具、存 blob)一律封成 activity,由 Temporal 记录 其结果、重放时 跳过。这是引擎能崩溃恢复的根基。
3.4 StepContext / cursor:引擎怎么知道 "我在第几步"
它要解决的小问题: 一个工作流 run 只跑一步,那 "我现在该跑哪一步" 靠什么定位?
思路/直觉。 靠一个 cursor(游标):TransitionTarget(workflow, step, scope_id)——
"哪个(子)工作流、第几步、哪个作用域"。StepContext 拿着 cursor 就能算出 "当前步定义"
current_step 和一堆布尔判断(是不是第一步/最后一步/主线),run() 靠这些做决策。
核心结构一览(都在 common/protocol/tasks.py):
| 结构 | 是什么 | 关键字段/属性 | 位置 |
|---|---|---|---|
StepContext | 当前步的一切上下文 | execution_input、cursor、current_input | tasks.py:163 |
StepContext.current_step | 由 cursor 索引出的步定义 | current_workflow.steps[cursor.step] | tasks.py:233 |
StepContext.is_last_step | 是否本工作流最后一步 | cursor.step + 1 == len(steps) | tasks.py:245 |
PartialTransition | "这一步想产生什么转移" 的半成品 | type、output、user_state | tasks.py:145 |
WorkflowResult | 一步/一段执行的结果 | state: PartialTransition、returned | tasks.py:150 |
StepOutcome | 执行/求值一步的原始产出 | output、error、transition_to | tasks.py:340 |
数据流:三个结构怎么接力。
eval / activity ──▶ StepOutcome ──▶ handle_step ──▶ WorkflowResult
(原始产出) (output/error) (_handle_X) (state=PartialTransition)
│
▼
transition() ──▶ 下一步 cursor
current_step求值或跑 activity,得到StepOutcome(run()里outcome,__init__.py:783)。handle_step按步类型分派到_handle_<类型>,产出WorkflowResult,里面的state是一个PartialTransition(__init__.py:718、:832)。WorkflowResult.returned=True表示这一步是ReturnStep之类的 "收工信号",主循环会写finish/finish_branch并返回(__init__.py:850)。
3.5 状态机:哪些转移是合法的
它要解决的小问题: 转移不能乱跳(比如从 finish 又跳回 step)。谁来管合法性?
思路/直觉。 引擎定义了两套枚举 + 一张 "谁能转到谁" 的邻接表。转移写库时校验:新 type 必须 在当前 type 的合法后继集合里。
两套枚举(common/protocol/state_machine.py):
TransitionType(转移类型,引擎内部视角):init、init_branch、wait、resume、step、finish、finish_branch、error、cancelled(state_machine.py:11)。ExecutionStatus(执行状态,对外/API 视角):queued、starting、running、awaiting_input、succeeded、failed、cancelled(state_machine.py:24)。
两者的映射见 transition_to_execution_status(tasks.py:130 与 state_machine.py:120),例如
init → starting、step → running 、wait → awaiting_input、finish → succeeded、
error → failed。注意 transition type 比 status 细:init_branch/resume/step/finish_branch
都对应对外的 running。
合法转移(邻接表,tasks.py:71 的 valid_transitions;state_machine.py:52 有等价副本):
| 从(当前 type) | 可转到 |
|---|---|
init | wait, error, step, cancelled, init_branch, finish |
init_branch | wait, error, step, cancelled, init_branch, finish_branch, finish |
wait | resume, step, cancelled, finish, finish_branch |
resume | wait, error, cancelled, step, finish, finish_branch, init_branch |
step | wait, error, cancelled, step, finish, finish_branch, init_branch |
finish_branch | wait, error, cancelled, step, finish, init_branch, finish_branch |
finish / error / cancelled | (终态,无后继) |
状态流转图(对外 ExecutionStatus 视角,tasks.py:46 注释里的 mermaid 的 ASCII 版):
怎么读: 箭头是允许的方向;succeeded/failed/cancelled 是三个出口(终态)。
queued ──▶ starting ──▶ running ──▶ succeeded ●
│ │ │ ▲ │
│ │ │ │ ├─▶ awaiting_input ──▶ running
│ │ │ └──┘ (等外部输入,如 wait_for_input)
▼ ▼ ▼
cancelled ● cancelled ● cancelled ● / failed ●
真实实现的两处。 邻接表在 Python 侧有两份:tasks.py:71(工作流引擎用的字典)和
state_machine.py:134 的 ExecutionStateMachine 类(带上下文管理器 transition_to、
会在非法转移时抛 StateTransitionError)。而真正把关发生在写库路径——
create_execution_transition 里的 validate_transition_targets 会断言 current/next 的配套关系
(如 finish 的 next 必须为 None、resume/step 的 next 必须非空、同 scope 下 next.step 必须
严格递增),不合法直接报错(queries/executions/create_execution_transition.py:47)。
3.6 错误与取消:怎么变成 error / cancelled 转移
它要解决的小问题: 步骤跑挂了、或用户取消了执行,引擎怎么收尾?
思路/直觉。 run() 主循环把 "执行步骤" 和 "处理结果" 都包在 try/except 里,并区分两类异常:
| 情况 | 引擎怎么做 | __init__.py 位置 |
|---|---|---|
| activity/求值抛异常 | 先剥掉 ActivityError 外壳拿到根因;非取消 → 写 cancelled,否则写 error,再抛 ApplicationError | :800–:816 |
handle_step 阶段抛异常 | 同上一套:CancelledError → cancelled 转移;其它 → error 转移 | :836–:848 |
步骤产出里带 outcome.error | 不抛异常,直接写 error 转移并终止 | :822–:828 |
ErrorWorkflowStep(任务里显式报错) | 写 error 转移,抛 ApplicationError | :367 |
一个防重复的细节: 异常上会挂一个 transitioned 标志。continue_as_child 抛出子工作流
错误前会设 e.transitioned = True(helpers.py:121);主循环 catch 到时先看
getattr(e, "transitioned", False),已经写过转移的就不再重复写(__init__.py:809、:825、
:845)——避免同一个错误被记两条日志。
取消的确定性来源: 取消不是引擎凭空造的,而是 Temporal 在工作流被取消时抛
CancelledError;引擎捕获它、翻译成 cancelled 转移(__init__.py:803)。
4. 巧妙之处(可借鉴的技术)
-
每步一 run + continue-as-new:把无限长的执行装进有界 history。 这是本引擎最值得学的一招—— 用 Temporal 原语把 "长流程" 切成 "许多短 run",既保留崩溃恢复,又不让单个 run 的历史膨胀。 见
helpers.py:77/__init__.py:899。 -
转移即日志(append-only transitions):状态、进度、产物三合一。 不设独立的 "状态字段" 去 改来改去,而是每次变化 append 一行;当前状态 = 最新一行,历史可回放。见
transition.py:22、queries/executions/create_execution_transition.py。 -
所有副作用都走 activity,保住工作流确定性。 写库、调 LLM、跑工具、存 blob 一律封成 activity;工作流函数本身纯确定性,才能被 Temporal 安全重放。见
transition.py:56、__init__.py:787。 -
大 payload 外移到 blob store。 步间传递的大对象先
save_inputs_remote存远端、只传引用 (RemoteObject),用时再load(),避免 Temporal payload 超限。见__init__.py:888、tasks.py:169(StepContext.load_inputs)。 -
cursor + scope_id 支撑嵌套子工作流。 分支/循环各开一个带独立 workflow 名的子工作流,却共享
scope_id以便把同一逻辑作用域的转移聚到一起(get_inputs按scope_id捞历史)。见helpers.py:125起、tasks.py:239、:276。
5. 边界与局限(诚实)
-
ParallelStep未实现。_handle_ParallelStep直接抛 "Not implemented" (__init__.py:562),文件顶部的支持矩阵注释也标它为 ❌(__init__.py:103)。真正的并行只在 map-reduce 的并行分支里(execute_map_reduce_step_parallel,helpers.py:312)。 -
PromptStep 的部分工具回填未实现。 prompt 步收到
integrations/api_call/system类型的工具调用时会抛NotImplementedError,只有function类型走了完整回填 (__init__.py:509–:534)。 -
continue-as-new 的触发判定存疑。
helpers.py:85的# FIXME: This doesn't actually work说明 "何时归零 history vs 何时开子工作流" 并不总按预期切换。 -
合法转移表有两份副本。
tasks.py:71与state_machine.py:52各维护一份valid_transitions, 存在漂移风险;而实际的强校验在 SQL 侧的validate_transition_targets(create_execution_transition.py:47)——读代码时以写库那份为准。 -
超时/取消翻译成转移仍有 TODO。 文件里多处
TODO: find a way to transition to error if workflow or activity times out(__init__.py:116),说明超时路径的状态收尾还不完备。
6. 横向对比
同属 ai-agent-reference 货架、都要 "驱动一个 agent 长流程" 的项目,取舍不同:
| 维度 | Julep(本章) | 典型 "内存循环" 型 agent |
|---|---|---|
| 执行载体 | 外部持久化工作流引擎(Temporal),每步一 run | 单进程内的 while 循环 |
| 崩溃恢复 | 原生支持(重放 + 转移日志) | 一般靠自己 checkpoint,或不可恢复 |
| 长任务/暂停 | 可 wait_for_input 等数天(run_timeout=31 天,temporal.py:133) | 受进程生命周期限制 |
| 复杂度成本 | 高:要跑 Temporal 集群、Worker、DB | 低:一个进程即可 |
一句话取舍:Julep 用 "运维一套 Temporal" 的成本,换来了持久化、可恢复、可审计的执行—— 适合把 agent 当 "生产级工作流" 跑,而非一次性脚本。想看它整体架构(API/查询层/Worker 怎么拼), 见 06 平台架构。
7. 代码地图(导航索引)
用符号名 grep 比行号抗漂移;下面每行一句,指向该跳去读的真源码。
| 主题 | 文件路径 | 符号名 |
|---|---|---|
| HTTP 入口:创建执行 | src/agents-api/agents_api/routers/tasks/create_task_execution.py | create_task_execution |
| 落库 + 启动工作流 | src/agents-api/agents_api/routers/tasks/create_task_execution.py | start_execution |
| 备料:拼 ExecutionInput | src/agents-api/agents_api/queries/executions/prepare_execution_input.py | prepare_execution_input |
| 连 Temporal、启动 run | src/agents-api/agents_api/clients/temporal.py | run_task_execution_workflow |
| 主循环 | src/agents-api/agents_api/workflows/task_execution/__init__.py | TaskExecutionWorkflow.run |
| 步骤分派器 | src/agents-api/agents_api/workflows/task_execution/__init__.py | handle_step / _handle_* |
| 工作流内表达式求值 | src/agents-api/agents_api/workflows/task_execution/__init__.py | eval_step_exprs |
| 步类型→activity 映射 | src/agents-api/agents_api/workflows/task_execution/__init__.py | STEP_TO_ACTIVITY |
| 下一步跑成新 run | src/agents-api/agents_api/workflows/task_execution/helpers.py | continue_as_child |
| 分支/循环子工作流 | src/agents-api/agents_api/workflows/task_execution/helpers.py | execute_if_else_branch / execute_foreach_step / execute_map_reduce_step |
| 写转移(工作流侧) | src/agents-api/agents_api/workflows/task_execution/transition.py | transition |
| 写转移(activity) | src/agents-api/agents_api/activities/task_steps/transition_step.py | transition_step |
| 写库 + 校验合法性 | src/agents-api/agents_api/queries/executions/create_execution_transition.py | create_execution_transition / validate_transition_targets |
| 合法转移表 / 状态映射 | src/agents-api/agents_api/common/protocol/tasks.py | valid_transitions / transition_to_execution_status |
| 状态机枚举与校验类 | src/agents-api/agents_api/common/protocol/state_machine.py | TransitionType / ExecutionStatus / ExecutionStateMachine |
| 步上下文 / 游标 | src/agents-api/agents_api/common/protocol/tasks.py | StepContext / PartialTransition / WorkflowResult / StepOutcome |
| 工作流注册到 Worker | src/agents-api/agents_api/worker/worker.py | TaskExecutionWorkflow(workflows=[...]) |