数据截至 2026-08-18(上游 commit 6a6b6383eb6a)

工作流引擎：用 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:149;库封装 skyvern/library/skyvern.py:369
WorkflowService.execute_workflow	编排:按顺序取出 block 挨个跑,处理失败/终止/取消	skyvern/forge/sdk/workflow/service.py:4759
Block / 各子类	积木本身:每类 block 一件事	skyvern/forge/sdk/workflow/models/block.py:594
WorkflowRunContext	参数与值的中央仓库,block 间传数据靠它	skyvern/forge/sdk/workflow/context_manager.py:118
block_yaml_to_block	把 no-code 的 YAML 定义翻译成运行时 Block 对象	skyvern/forge/sdk/workflow/workflow_definition_converter.py:467

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

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:474)。按"干什么"可以分成三大类——这是理解整章的骨架:

大类	干什么	代表 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:594)
 ├── BaseTaskBlock (block.py:1300)  ← 浏览器任务类的公共父类
 │    ├── 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:11493。区别只在语义标签(engine/goal 的默认解读),执行路径共享。

3.3 Block 抽象基类:公共骨架

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

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

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

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

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

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

---

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:1522)的核心是这一跳:

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:1648。它把 block 的字段(url/navigation_goal/data_extraction_goal/data_schema/parameters 等)映射成一个 Task 记录 + 首个 Step(agent.py:970 create_task_and_step_from_block)。注意 block 的 parameters 在这里被展开成 Task 的 navigation_payload(agent.py:979-982)——这就是"参数怎么喂给 Agent"。

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

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

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

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

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

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

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

4.4 首个 block 的特殊职责

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

---

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

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

5.1 一个中央仓库:parameters 与 values

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

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

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

5.2 命名约定:<label>_output

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

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:1549。

这里有个易读性设计:除了 do_login_output 这个原始键,还会去掉后缀再存一份 do_login,并把 Task 产出里的 extracted_information 提升成 .output 字段(context_manager.py:1560-1586)。所以作者在模板里既能写 {{ 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:873)。它把 workflow_run_context.values 整个塞进模板上下文(block.py:904),再加上一堆全局量:current_index / current_item(循环用)、workflow_run_id、current_date、workflow_run_outputs 等(block.py:936-958)。

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

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:892-910 is_safe_block_for_secrets)。这是"参数管道"里一条承重的隐性规则。

---

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

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

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

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

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

6.2 WhileLoopBlock:按运行时条件循环

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

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

6.3 ConditionalBlock:按条件选分支

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

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

---

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

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

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

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

---

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

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

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

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

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

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

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

---

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

用户在 UI 或 YAML 里写的是声明式定义(BlockYAML,schemas/workflows.py:757),不是 Python 对象。中间的翻译层是 block_yaml_to_block(workflow_definition_converter.py:467):它按 block_type 逐个把 YAML 造成运行时 Block 实例,并取用预先建好的 <label>_output 输出参数(workflow_definition_converter.py:472;参数由 :128、:371-377 统一预创建),把数据管道的接口接上。循环块是递归翻译的——loop_blocks 里的子块再走一遍 block_yaml_to_block(workflow_definition_converter.py:531-534)。

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

---

10. 公共入口:run_workflow

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

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

—— services/workflow_service.py:170-201。真正的 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 用的是黑名单式限制,BLOCKED_ATTRS 是一份静态属性黑名单(code_block_safety.py:14,注释还要求与 runner 镜像里的拷贝保持对齐 :13),黑名单天然不全——那条"not a security boundary"的旧注释已随重构移除,但黑名单本质未变。别把它当强隔离用来跑不可信代码。
• 循环硬上限 1000。 超过 DEFAULT_MAX_LOOP_ITERATIONS 直接判失败(block.py:515),超大批量任务要自己分页或换设计。
• 密文可见性是按 block 类型硬切的。 想在浏览器/文本块里用真实密钥拿不到(只有 CODE/HTTP_REQUEST 能),这是有意的防泄漏,不是 bug(block.py:892)。
• 参数重名会被覆盖。 context 初始化时若两个 workflow 参数同 key,后者盖前者并只打一条 error 日志(context_manager.py:165-170),不报错中断。
• 本章不覆盖脚本缓存路径。 service.py 里大量 is_script_run / 脚本 fallback 逻辑(如 service.py:6678-6983)属于"把成功过的 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:594	Block
统一执行包装(建记录/截图/兜底)	skyvern/forge/sdk/workflow/models/block.py:1137	Block.execute_safe
block 间模板渲染(取上游产出)	skyvern/forge/sdk/workflow/models/block.py:873	format_block_parameter_template_from_workflow_run_context
浏览器任务类公共父类	skyvern/forge/sdk/workflow/models/block.py:1300	BaseTaskBlock
Task 结局→Block 结局 映射 + 重试	skyvern/forge/sdk/workflow/models/block.py:1858	BaseTaskBlock.execute (block_status_mapping)
block→Task+Step 关键一跳	skyvern/forge/agent.py:970	create_task_and_step_from_block
For 循环块	skyvern/forge/sdk/workflow/models/block.py:2242	ForLoopBlock
While 循环块 + 1000 上限	skyvern/forge/sdk/workflow/models/block.py:3282	WhileLoopBlock / DEFAULT_MAX_LOOP_ITERATIONS
条件分支块	skyvern/forge/sdk/workflow/models/block.py:14112	ConditionalBlock
受限代码块沙箱	skyvern/forge/sdk/workflow/models/block.py:4557	CodeBlock / BLOCKED_ATTRS
文本 prompt 块	skyvern/forge/sdk/workflow/models/block.py:7591	TextPromptBlock
HTTP 请求块	skyvern/forge/sdk/workflow/models/block.py:12314	HttpRequestBlock
文件解析块	skyvern/forge/sdk/workflow/models/block.py:9795	FileParserBlock
BlockType 枚举(所有类型)	skyvern/schemas/workflows.py:474	BlockType
参数/值中央仓库	skyvern/forge/sdk/workflow/context_manager.py:118	WorkflowRunContext
产出写回 + 喂给依赖块	skyvern/forge/sdk/workflow/context_manager.py:1549	register_output_parameter_value_post_execution
去后缀别名 + .output 提升	skyvern/forge/sdk/workflow/context_manager.py:1560	register_block_reference_variable_from_output_parameter
编排:逐 block 顺序执行	skyvern/forge/sdk/workflow/service.py:4759	WorkflowService.execute_workflow
失败/终止/取消 决策	skyvern/forge/sdk/workflow/service.py:7782	_handle_block_result_status(continue_on_failure)
YAML→Block 翻译	skyvern/forge/sdk/workflow/workflow_definition_converter.py:467	block_yaml_to_block
服务层公共入口	skyvern/services/workflow_service.py:149	run_workflow
库封装入口	skyvern/library/skyvern.py:369	Skyvern.run_workflow