跳到主要内容

工作流引擎：用 Block 编排多步骤

30 秒导读: 前面几章讲的是"一个任务怎么跑"(02-agent-loop-planning.md 讲单步 Agent 循环)。但真实业务很少是一个任务——通常是"登录 → 循环处理每一行 → 抽取数据 → 发邮件"这种多步骤流程。工作流引擎就是把这些步骤拼成一条可复用、可参数化的流水线的东西。它的最小积木叫 Block;block 与 block 之间用参数传数据。本章讲清楚:block 是编排单元,参数是数据管道。

1. 这是什么(零基础也能懂)​

一句话定义: 工作流(workflow)= 一串按顺序执行的 Block;每个 Block 是一个自带输入输出的"步骤积木"。

它解决什么问题: 假设你要每天从一个供应商门户下载账单。这件事是:登录 → 进报表页 → 对每个月循环下载 → 把下载的文件解析成表格 → 邮件发给财务。你不想每次都写一遍代码,也不想让一个巨型 Agent"一口气自由发挥"(不可控、不可复用)。工作流让你把它拆成命名好的步骤,存成一份定义,以后换参数就能重跑。

一个 Block 长什么样(概念示意,非源码):

# 示意:一个"导航到登录页并登录"的 block
- block_type: login
  label: do_login            # 块的唯一名字,别的块靠它引用输出
  url: "https://portal.example.com"
  navigation_goal: "用给定账号登录"
  parameters: [username, password]

一句话直觉: 把工作流想成一条工厂流水线——每个工位(Block)干一件事,上一个工位的产出(OutputParameter)顺着传送带流到下一个工位。有的工位是"派个工人去网页上干活"(浏览器任务类 block),有的工位是"纯机器计算"(代码/HTTP/解析类 block),还有的是"控制传送带走向"(循环、分支)。

它和"单步 Agent"的分工:

关切谁负责本章讲不讲
网页里的一步动作(点哪、填什么)单步 Agent 循环不讲,见 02
一句话目标自主拆解成整条流程Skyvern 2.0 规划器不讲,见 06
把多个任务显式编排成可复用流水线工作流引擎 / Block本章

2. 顶层全景(它大概怎么转)​

怎么读下面这张图: 从上到下是"一次工作流运行"的生命周期;左边是用户给的东西,右边是引擎内部。

  用户               引擎入口                 编排层                    积木层
 ┌──────┐   run_workflow   ┌──────────────┐  execute_workflow  ┌──────────────────┐
 │ 参数 │ ───────────────> │ prepare_     │ ─────────────────> │ 逐个 block:      │
 │ 值   │   library/       │ workflow     │  service.py        │ block.execute_   │
 └──────┘   skyvern.py     │ 建 WorkflowRun│                    │   safe()         │
                           └──────────────┘                    └────────┬─────────┘
                                  │                                     │
                                  │ 初始化                     ┌────────▼─────────┐
                                  ▼                            │ 浏览器任务类     │
                        ┌───────────────────┐  取/写参数值     │ → 落到单步 Agent │
                        │ WorkflowRunContext │ <──────────────>│ 控制流类 → 循环/ │
                        │ (参数 + 值的仓库)  │                 │ 分支             │
                        └───────────────────┘                 │ 副作用类 → 直接干│
                                                               └──────────────────┘

部件一句话职责:

部件干什么在哪个文件
run_workflow公共入口:接参数、建运行、丢给执行器skyvern/services/workflow_service.py:83;库封装 skyvern/library/skyvern.py:351
WorkflowService.execute_workflow编排:按顺序取出 block 挨个跑,处理失败/终止/取消skyvern/forge/sdk/workflow/service.py:1615
Block / 各子类积木本身:每类 block 一件事skyvern/forge/sdk/workflow/models/block.py:379
WorkflowRunContext参数与值的中央仓库,block 间传数据靠它skyvern/forge/sdk/workflow/context_manager.py:89
block_yaml_to_block把 no-code 的 YAML 定义翻译成运行时 Block 对象skyvern/forge/sdk/workflow/workflow_definition_converter.py:458

主线走一遍(高层,不进代码):

  1. 用户调 run_workflow,给一份 workflow 的 id + 一个参数字典。
  2. 引擎建一个 WorkflowRun,初始化 WorkflowRunContext(把用户参数值灌进去)。
  3. 编排层从 workflow 定义里拿到 block 列表,for block in blocks 挨个调 block.execute_safe(...)。
  4. 每个 block 跑完把产出写回 context(键叫 <label>_output),下一个 block 用 Jinja 模板 {{ do_login.output }} 就能引用到。
  5. 某个 block 失败/终止 → 看它的 continue_on_failure,决定是整条流程停,还是跳过继续。

3. 核心概念一:Block 是编排单元​

3.1 Block 家族全景​

Skyvern 的 block 类型由一个枚举 BlockType 定义(skyvern/schemas/workflows.py:417)。按"干什么"可以分成三大类——这是理解整章的骨架:

大类干什么代表 block
浏览器任务类派一个"单步 Agent"去网页上完成一个子目标TaskBlock NavigationBlock ExtractionBlock ValidationBlock LoginBlock FileDownloadBlock ActionBlock UrlBlock
控制流类不干活,只决定传送带怎么走ForLoopBlock WhileLoopBlock ConditionalBlock
副作用 / 纯计算类不碰浏览器 Agent,直接执行确定性逻辑CodeBlock TextPromptBlock SendEmailBlock FileParserBlock HttpRequestBlock UploadToS3Block FileUploadBlock WaitBlock PDFParserBlock PrintPageBlock HumanInteractionBlock TaskV2Block WorkflowTriggerBlock

关键区分(全章最重要的一条): 浏览器任务类 block 自己不执行网页动作——它把自己"翻译"成一个 Task + Step,交给单步 Agent 循环去跑(§4.2)。其余两类是引擎自己就地处理。

3.2 继承结构​

Block (抽象基类, block.py:379)
 ├── BaseTaskBlock (block.py:1013)  ← 浏览器任务类的公共父类
 │    ├── TaskBlock / NavigationBlock / ExtractionBlock
 │    ├── ValidationBlock / LoginBlock / FileDownloadBlock
 │    ├── ActionBlock / UrlBlock / HumanInteractionBlock
 ├── ForLoopBlock / WhileLoopBlock / ConditionalBlock   ← 控制流
 └── CodeBlock / TextPromptBlock / SendEmailBlock / ...  ← 副作用/纯计算

具体的浏览器 block 大多是空壳——只声明 block_type,行为全在 BaseTaskBlock 里。例如整个 NavigationBlock 就一行有效代码:

class NavigationBlock(BaseTaskBlock):
    block_type: Literal[BlockType.NAVIGATION] = BlockType.NAVIGATION

—— block.py:6946。区别只在语义标签(engine/goal 的默认解读),执行路径共享。

3.3 Block 抽象基类:公共骨架​

每个 block 至少带这几样字段(block.py:379 class Block):

字段作用
label块的唯一名字,别的块靠它引用输出
block_type判别类型(Pydantic 靠它做 discriminated union)
output_parameter这个块的产出插槽,是数据管道的接口(§5)
continue_on_failure失败时整条流程停还是跳过继续
next_block_labelv2 图式定义里指向下一个块;省略则按顺序

真正的执行契约是两个方法:

  • execute(...) —— 抽象方法,每类 block 自己实现"我这一步干什么"(block.py:821)。
  • execute_safe(...) —— 统一包装,所有 block 共用:建数据库记录、截图、调 execute、兜底 catch 异常(block.py:886)。

execute_safe 是编排层唯一调用的入口。它做了三件不显眼但重要的事(block.py:906-990):

  1. 建 workflow_run_block 记录——每个 block 每次执行在库里有一行,带 label/block_type/status,前端靠它显示进度。
  2. 执行前截图——存一张 SCREENSHOT_LLM 工件,方便回放调试(block.py:940)。
  3. 兜底容错——execute 抛任何异常都被 catch,转成一个 success=False 的 BlockResult 而不是让整个进程崩(block.py:969-990)。

4. 核心概念二:浏览器任务类 block 如何落到单步 Agent​

这是本章和 02 的接缝处。

4.1 它要解决的小问题​

一个 NavigationBlock 说"去登录页登录",但登录本身是一连串网页动作(填账号、填密码、点按钮、可能还要处理验证码)。这些动作的规划与执行是单步 Agent 的活。所以浏览器 block 的职责是:把自己变成一个 Agent 能跑的 Task,然后驱动 Agent 的 step 循环,最后把 Task 的结局翻译回 block 的结局。

4.2 关键一跳:create_task_and_step_from_block​

BaseTaskBlock.execute(block.py:1197)的核心是这一跳:

task, step = await app.agent.create_task_and_step_from_block(
    task_block=self,
    workflow=workflow,
    workflow_run=workflow_run,
    workflow_run_context=workflow_run_context,
    task_order=task_order,
    task_retry=task_retry,
)

—— block.py:1297。它把 block 的字段(url/navigation_goal/data_extraction_goal/data_schema/parameters 等)映射成一个 Task 记录 + 首个 Step(agent.py:438 create_task_and_step_from_block)。注意 block 的 parameters 在这里被展开成 Task 的 navigation_payload(agent.py:447-450)——这就是"参数怎么喂给 Agent"。

建好之后,block 亲自驱动 step 循环:

await app.agent.execute_step(
    organization=organization, task=task, step=step,
    task_block=self, ... engine=self.engine,
)

—— block.py:1466。execute_step 就是 02 章讲的那个单步循环。本章到此为止,不重复它内部。

4.3 把 Task 的结局翻译回 Block 的结局​

Agent 跑完后,block 读回 Task 的最终状态,用一张固定映射转成 BlockStatus(block.py:1499):

Task 状态→ Block 状态编排层怎么处理
completedcompleted成功,产出写进 output_parameter
terminatedterminated除非 continue_on_failure,否则整条流程终止
failedfailed在 max_retries 内会重试;耗尽则失败
canceledcanceled整条工作流取消
timed_outtimed_out当失败处理

重试逻辑在同一个 while will_retry 循环里(block.py:1294、1602-1604):每次失败 current_retry += 1,will_retry = current_retry <= self.max_retries。有个巧妙细节:如果失败是反爬检测导致的,直接放弃重试(重试也没用),见 _should_skip_retry_on_anti_bot_detection(block.py:1000、1605)。

4.4 首个 block 的特殊职责​

流程里的第一个浏览器 block 要负责创建浏览器、导航到起始 URL(block.py:1318 is_first_task);后续 block 复用同一个浏览器状态,只在需要时手动导航(block.py:1420)。这保证了跨 block 的浏览器会话是连续的(登录态不丢)。

5. 核心概念三:参数是 block 之间的数据管道​

前面反复说"产出流到下一个块"。这条管道的实现全在 WorkflowRunContext(context_manager.py:89)。理解它,就理解了 no-code 工作流"数据怎么走"的全部。

5.1 一个中央仓库:parameters 与 values​

WorkflowRunContext 内部就是两个字典(context_manager.py:231-232):

  • parameters: dict[str, Parameter] —— 参数的声明(它是什么类型、从哪来)。
  • values: dict[str, Any] —— 参数的当前值。

get_value / set_value / has_value(context_manager.py:257-272)就是往这个仓库读写。所有 block 共享同一个 context,所以 A 块写、B 块读,数据就"流"过去了。

5.2 命名约定:<label>_output​

管道的接口靠一条命名约定。每个 block 都有一个 output_parameter,它的 key 恒为 <block_label>_output(转换时统一创建,workflow_definition_converter.py:462)。block 跑完调 record_output_parameter_value(block.py:427),最终落到:

async def register_output_parameter_value_post_execution(self, parameter, value):
    self.values[parameter.key] = value                                   # 存 <label>_output
    self.register_block_reference_variable_from_output_parameter(...)     # 再存一份 <label>
    await self.set_parameter_values_for_output_parameter_dependent_blocks(...)  # 喂给依赖它的块

—— context_manager.py:1334。

这里有个易读性设计:除了 do_login_output 这个原始键,还会去掉后缀再存一份 do_login,并把 Task 产出里的 extracted_information 提升成 .output 字段(context_manager.py:1345-1371)。所以作者在模板里既能写 {{ do_login_output }},也能写更顺眼的 {{ do_login.output }}。

5.3 后一个 block 怎么"取"到前一个的产出​

浏览器/文本类 block 在执行前会调 format_potential_template_parameters,把自己字段里的 Jinja 模板用 context 的 values 渲染成实际值。渲染引擎是 format_block_parameter_template_from_workflow_run_context(block.py:643)。它把 workflow_run_context.values 整个塞进模板上下文(block.py:673),再加上一堆全局量:current_index / current_item(循环用)、workflow_run_id、current_date、workflow_run_outputs 等(block.py:734-756)。

于是数据管道的一次完整流动是:

do_login 块跑完
   │  record_output_parameter_value
   ▼
context.values["do_login_output"] = {...}
context.values["do_login"]        = {..., "output": ...}   ← 去后缀 + 提升
   │
   ▼  下一个块渲染自己的字段
extract 块的 navigation_goal = "基于 {{ do_login.output }} 抽取..."
   │  format_block_parameter_template_from_workflow_run_context
   ▼
渲染成实际文本 → 喂给 create_task_and_step_from_block → Agent

5.4 一个安全巧思:密文只给"不外泄"的块​

密码/密钥这类 secret,只有不会把数据发给 LLM 的块(CodeBlock、HttpRequestBlock)才允许在模板里拿到真实明文;浏览器/文本类块会拿到占位符,防止密钥被写进发给模型的 prompt(block.py:661-679 is_safe_block_for_secrets)。这是"参数管道"里一条承重的隐性规则。

6. 核心概念四:控制流 block​

控制流 block 不产出业务数据,只决定"传送带怎么走"。它们的共同点:内部持有一个子 block 列表,自己负责按规则驱动这些子 block。

6.1 ForLoopBlock:对一个集合逐项循环​

ForLoopBlock(block.py:1855)拿一个列表(来自 loop_over 参数或 loop_variable_reference 引用),对每一项把 loop_blocks 里的子 block 跑一遍。每轮迭代会把当前项注入上下文:模板里可用 {{ current_index }} / {{ current_item }}(block.py:734)。

典型用途:"对搜索结果的每一行,进详情页抽数据"。

6.2 WhileLoopBlock:按运行时条件循环​

WhileLoopBlock(block.py:2830)在每轮迭代前评估一个 condition(可以是 Jinja 表达式或自然语言);为真才跑一轮 body。语义是"top-of-loop":第一次就为假则 body 一次都不跑。

防死循环的护栏: 两种循环都硬编码上限 DEFAULT_MAX_LOOP_ITERATIONS = 500(block.py:319);触顶当作失败而非静默停止(block.py:2375、3112),这样一个"永远为真"的坏条件不会把机器转爆。

6.3 ConditionalBlock:按条件选分支​

ConditionalBlock(block.py:8910)持有一组有序的 branch_conditions,依次评估,命中第一个为真的分支,把流程导向该分支的 next_block_label。条件可以是 Jinja(JinjaBranchCriteria,block.py:8123)或自然语言(PromptBranchCriteria,block.py:8163,靠 LLM 判断)。

有个性能巧思:多个自然语言条件会打包成一次 LLM 调用批量评估,而不是一条一条问(block.py:8937 _evaluate_prompt_branches)。

7. 副作用 / 纯计算类 block(逐个一句话)​

这些块不碰浏览器 Agent,直接执行确定性逻辑。它们让工作流不止能"操作网页",还能"处理数据、对外通信"。

Block干什么关键实现
CodeBlock跑一段用户提供的 Python(受限沙箱)block.py:3622;黑名单挡住 subprocess/文件/frame 逃逸,__builtins__ 清空(BLOCKED_ATTRS, block.py:3640)
TextPromptBlock拿一段 prompt 直接问 LLM,可要求 JSON 结构化输出block.py:4373;send_prompt(block.py:4418)
HttpRequestBlock发一个 HTTP 请求(GET/POST...),把响应当产出block.py:7222;本地文件访问被 get_allowed_dirs 限死(block.py:7244)
SendEmailBlock通过 SMTP 发邮件,可带下载的附件block.py:5436;SMTP 凭据解密 block.py:5493
FileParserBlock把 CSV/Excel/PDF/图片/docx 解析成结构化数据block.py:5765;按魔数/编码探测类型,PDF 走 OCR
UploadToS3Block / FileUploadBlock把产物上传到 S3 / Azure / Google Driveblock.py:4687 / 4805
DownloadToS3Block从 URL 下载再传 S3block.py:4599
WaitBlock单纯等一段时间block.py:6613
PDFParserBlock / PrintPageBlock解析 PDF / 把当前页打印成 PDF 存档block.py:6500 / 7667
HumanInteractionBlock暂停,等人工审批/拒绝再继续block.py:6660(继承 BaseTaskBlock)
TaskV2Block内嵌一个 Skyvern 2.0 自主任务(见 06)block.py:6986
WorkflowTriggerBlock触发另一条子工作流(带深度限制防无限嵌套)block.py:9248;_check_trigger_depth(block.py:9274)

这些块同样通过 output_parameter 把产物挂进数据管道——例如 FileParserBlock 解析出的表格,下游 ForLoopBlock 就能 loop_over 它逐行处理。

8. 编排层:怎么把 block 串起来跑​

WorkflowService.execute_workflow(service.py:1615)是总指挥。剥掉脚本缓存等旁支后,核心就是一个顺序循环:

for block_idx, block in enumerate(blocks):        # service.py:2420
    result = await block.execute_safe(...)        # service.py:3191
    workflow_run, should_stop = _handle_block_result_status(block, result, ...)
    if should_stop:
        break

should_stop 怎么定,全看 block 结局 + continue_on_failure(service.py:3755 _handle_block_result_status):

block 结局continue_on_failure=Falsecontinue_on_failure=True
failed标工作流失败,停(service.py:3794)记警告,继续下一块(service.py:3811)
terminated标工作流终止,停(service.py:3831)继续下一块
canceled直接取消整条工作流(service.py:3775)同样取消(取消不可续)

所以 continue_on_failure 是作者控制"某一步失败要不要拖垮全局"的旋钮——比如"发邮件失败无所谓,别让整个下载流程白跑"。

v2 图式定义还支持非线性走向:block 带 next_block_label,ConditionalBlock 的分支也带 next_block_label,引擎据此走 DAG 而非纯顺序(service.py:2697;DAG 合法性校验含环检测 service.py:3991-4037)。

9. no-code 是怎么落地的:YAML → Block​

用户在 UI 或 YAML 里写的是声明式定义(BlockYAML,schemas/workflows.py:676),不是 Python 对象。中间的翻译层是 block_yaml_to_block(workflow_definition_converter.py:458):它按 block_type 逐个把 YAML 造成运行时 Block 实例,并为每个 block 建好 <label>_output 这个输出参数(converter.py:462),把数据管道的接口接上。循环块是递归翻译的——loop_blocks 里的子块再走一遍 block_yaml_to_block(converter.py:489)。

这就是"no-code"的本质:声明式 YAML/表单 ⇄ 运行时 Block 对象,用户永远只碰前者。

10. 公共入口:run_workflow​

从库用户视角,一切从 Skyvern.run_workflow 开始(library/skyvern.py:351):给 workflow_id + 一个 parameters 字典,可选 wait_for_completion 轮询到终态。它底下走服务层的 run_workflow(services/workflow_service.py:83):

workflow_run = await prepare_workflow(...)            # 建 WorkflowRun,灌参数
await AsyncExecutorFactory.get_executor().execute_workflow(...)   # 异步跑
return workflow_run

—— services/workflow_service.py:101-129。真正的 block 循环发生在执行器里(异步/后台),即 §8 的 WorkflowService.execute_workflow。

一个最小调用(概念示意):

# 示意,非源码
run = await skyvern.run_workflow(
    workflow_id="wf_123",
    parameters={"username": "alice", "password": "..."},
    wait_for_completion=True,
)
print(run.output)   # 各 block 的 output_parameter 汇总

11. 边界与坑(诚实交代)​

  • 沙箱不是安全边界。 CodeBlock 用的是黑名单式限制,源码注释自己写明"inherently incomplete... not a security boundary"(block.py:3634-3637)。别把它当强隔离用来跑不可信代码。
  • 循环硬上限 500。 超过 DEFAULT_MAX_LOOP_ITERATIONS 直接判失败(block.py:319),超大批量任务要自己分页或换设计。
  • 密文可见性是按 block 类型硬切的。 想在浏览器/文本块里用真实密钥拿不到(只有 CODE/HTTP_REQUEST 能),这是有意的防泄漏,不是 bug(block.py:661)。
  • 参数重名会被覆盖。 context 初始化时若两个 workflow 参数同 key,后者盖前者并只打一条 error 日志(context_manager.py:134-139),不报错中断。
  • 本章不覆盖脚本缓存路径。 service.py 里大量 is_script_run / 脚本 fallback 逻辑(如 service.py:2974-3223)属于"把成功过的 block 编译成可复用脚本"的自适应缓存优化,是正交主题,理解 block 编排不需要它。

12. 横向对比​

同货架的浏览器 agent 里,编排单元的粒度是主要分野:Skyvern 走"显式 Block DAG + 参数管道"的 no-code 路线(可复用、可版本化、门槛低);另一些框架把编排交给一段命令式代码或一个自由发挥的规划器。Skyvern 自己也有后者——06 章的 Skyvern 2.0 规划器能从一句话目标自主编排,可通过 TaskV2Block 嵌进 Block 工作流里,两种范式并存。与 02 章的关系已在 §4 讲清:Block 是"宏观编排",单步 Agent 是"微观执行",接缝在 create_task_and_step_from_block。

13. 代码地图(导航索引)​

主题文件路径符号名
Block 抽象基类skyvern/forge/sdk/workflow/models/block.py:379Block
统一执行包装(建记录/截图/兜底)skyvern/forge/sdk/workflow/models/block.py:886Block.execute_safe
block 间模板渲染(取上游产出)skyvern/forge/sdk/workflow/models/block.py:643format_block_parameter_template_from_workflow_run_context
浏览器任务类公共父类skyvern/forge/sdk/workflow/models/block.py:1013BaseTaskBlock
Task 结局→Block 结局 映射 + 重试skyvern/forge/sdk/workflow/models/block.py:1499BaseTaskBlock.execute (block_status_mapping)
block→Task+Step 关键一跳skyvern/forge/agent.py:438create_task_and_step_from_block
For 循环块skyvern/forge/sdk/workflow/models/block.py:1855ForLoopBlock
While 循环块 + 500 上限skyvern/forge/sdk/workflow/models/block.py:2830WhileLoopBlock / DEFAULT_MAX_LOOP_ITERATIONS
条件分支块skyvern/forge/sdk/workflow/models/block.py:8910ConditionalBlock
受限代码块沙箱skyvern/forge/sdk/workflow/models/block.py:3622CodeBlock / BLOCKED_ATTRS
文本 prompt 块skyvern/forge/sdk/workflow/models/block.py:4373TextPromptBlock
HTTP 请求块skyvern/forge/sdk/workflow/models/block.py:7222HttpRequestBlock
文件解析块skyvern/forge/sdk/workflow/models/block.py:5765FileParserBlock
BlockType 枚举(所有类型)skyvern/schemas/workflows.py:417BlockType
参数/值中央仓库skyvern/forge/sdk/workflow/context_manager.py:89WorkflowRunContext
产出写回 + 喂给依赖块skyvern/forge/sdk/workflow/context_manager.py:1334register_output_parameter_value_post_execution
去后缀别名 + .output 提升skyvern/forge/sdk/workflow/context_manager.py:1345register_block_reference_variable_from_output_parameter
编排:逐 block 顺序执行skyvern/forge/sdk/workflow/service.py:1615WorkflowService.execute_workflow
失败/终止/取消 决策skyvern/forge/sdk/workflow/service.py:3755_handle_block_result_status(continue_on_failure)
YAML→Block 翻译skyvern/forge/sdk/workflow/workflow_definition_converter.py:458block_yaml_to_block
服务层公共入口skyvern/services/workflow_service.py:83run_workflow
库封装入口skyvern/library/skyvern.py:351Skyvern.run_workflow