OpenHuman Workflows 实战：让 Agent 替你搭建可审批、可恢复的自动化流水线

本文围绕 OpenHuman 的 Workflows 功能展开：它是一套由开源 tinyflows 引擎驱动的"可视化、持久化、人机协同"自动化框架。你只需在聊天中用自然语言描述需求，Agent 会用 propose_workflow 工具起草一张完整的工作流图，你在画布上逐节点审阅并保存，之后它按 cron 调度或实时应用事件自动运行，遇到敏感动作会暂停等待你的批准。读完本文，你将掌握 Workflows 的节点模型、触发机制、信任与审批边界、运行观测方式以及面向开发者的 RPC 接口，并能在当前仓库源码中逐一找到对应实现。

从聊天到自动化：Workflows 解决什么问题

聊天适合"一次性提问"，而 Workflows 面向的是"每次都要做的事"：为每一封新客服邮件做分类、为每个提到你团队的 Linear 工单建档、每周一早上 9 点推送摘要。与 n8n、Zapier 等可视化工作流工具最大的区别在于：图不是人拖出来的，而是 Agent 生成的。

从源码结构看，Workflows 是 src/openhuman/flows/ 下一个以 feature 开关（#[cfg(feature = "flows")]，见 src/openhuman/flows/mod.rs）独立编译的领域模块，模块内部分工清晰：

• ops*.rs：业务逻辑与 RPC 处理器（create / get / list / update / delete / run / resume 等）；
• store.rs：基于 SQLite 的持久化（flow_state 表、flow_runs 行、逐 step 结果落库）；
• bus.rs：事件总线订阅器，负责把外部事件桥接到流程运行；
• tinyflows/：tinyflows 引擎的接入层（checkpointer、observability、memory adapter、Langfuse 导出）；
• builder_tools.rs：为工作流构建 Agent 提供的 22 个专用工具；
• node_contracts.rs：节点类型契约（本宿主叠加层）。

Agent 构建，你来批准：人机协同的安全设计

你不用拖拽任何框框。在聊天里描述需求，例如"每当客户发来新邮件，总结后发到我的 Slack"，Agent 会调用 propose_workflow 工具起草一张完整的工作流图，并在聊天中呈现为一张 Workflow Proposal Card（工作流提案卡），附上每个步骤的通俗英文摘要，供你逐条审阅。

这里有两道设计上的安全保证：

1. propose_workflow 工具只负责校验和描述候选图，它自身永远不能创建或启用一个流程；
2. 从提案到"新保存工作流"的唯一路径，是你点击卡片上的 Save & enable 按钮——该按钮直接从 App 调用 flows_create RPC，而不是经由 Agent。

在 src/openhuman/flows/builder_tools.rs 中，这一"人在回路不变量"被明确写在模块文档里：CreateWorkflowTool 与 DuplicateFlowTool 创建出的流程永远处于 DISABLED（禁用）状态；SaveWorkflowTool 从不设置 enabled / require_approval 字段；revise_workflow / edit_workflow / validate_workflow 只校验或就地修改草稿、从不落盘；dry_run_workflow 只针对 tinyflows 的确定性 mock 能力执行，无论什么权限层级都不可能触发真实的 LLM / 工具 / HTTP / 代码副作用。

有一个刻意的例外：当你主动发起构建时（Workflows 页面的提示栏会先创建流程并打开构建 copilot），构建 Agent 可以用它的 save_workflow 工具收尾——它把构建好的图写回到那个已存在的流程上，但前提是先在沙箱里做过一次 dry run。即便如此，它依然不能自行创建流程、启用/禁用流程或修改审批门设置，任何真实测试运行都必须先经过你的显式确认。

工作流的构成：15 种节点与有界循环

一张工作流图由 15 种节点类型组成：恰好一个 trigger（触发器），以及任意组合的 agent（一次带工具的完整 Agent turn）、tool_call、http_request、code（JavaScript 或 Python）、condition、switch、transform、split_out、merge、output_parser、sub_workflow、memory、dedup 和 loop。

从 src/openhuman/flows/node_contracts.rs 的宿主叠加层可以看到每种节点在本项目中的关键约束：

节点类型	核心要点（来自宿主叠加层注释）
trigger	本宿主当前只有 manual / schedule / app_event 真正派发，其余类型可保存但不会自动运行（flows_validate 会给出警告）；app_event 需要 config.toolkit + config.trigger_slug（Composio 应用 + 事件）
agent	数据通过 config.input_context 显式 =-绑定（=item、=items、=nodes.<id>.item.json）传入；prompt 必须是纯自然语言；execution=per_item 会对每个输入项跑一个完整 harness Agent，非常昂贵，且受进程级并发上限约束（默认 8，见 OPENHUMAN_FLOWS_MAX_PARALLEL_AGENTS）
tool_call	config.slug 必须是真实 Composio action slug（如 GMAIL_SEND_EMAIL）或 oh:<tool_name> 原生工具；Composio 输出被包裹在 data 里，下游绑定要用 =nodes.<id>.item.json.data.<field>
http_request	config.connection_ref 使用 http_cred:<name> 凭据
code	输出不会被 data 包裹，直接绑定 =nodes.<id>.item.json.<field>
split_out	若 Composio 源 primary_array_path 为 null，不要默认用 json.data，应先用 get_tool_output_sample 探测真实数组路径
memory	scope: "flow" 读写与 flow_memory_recall / flow_memory_remember 工具相同的按流程隔离内存命名空间
dedup	提交是运行级的：仅当整个运行以 completed / completed_with_warnings 结束时才把 tentative 键合并进 committed；推荐位置是 split_out → dedup [key="=item.id"] → …action…
loop	见下方"有界循环"；sub_workflow 是另一种重复方式，二者边界不同

有界循环（Bounded Loop）

工作流图通常是直线或扇出结构，但也允许包含有界循环：loop 节点在其 body 端口上反复发射，直到 max_iterations 上限（或可选的 condition 表达式）达到为止，然后从 done 端口发射结束。你通过把 body 的最后一个节点接回 loop 节点来闭合循环。上限永远是有限的；on_exceeded 决定达到上限意味着什么：

• error：运行失败，并点名该循环；
• continue：停止循环，把最后一轮的项目从 done 携带出去。

node_contracts.rs 还提醒：每次经过 body 都要付出 body 的成本——如果 body 里含有 agent 节点，每轮迭代都是一次完整的 Agent turn，所以 max_iterations 既是正确性约束也是花费约束（该节点默认上限 25）；同时应优先用 =-表达式 condition 让循环尽早结束。嵌套的 sub_workflow 运行则由根图 trigger 上的 max_sub_workflow_depth（默认 8）约束。

触发器的四种类型

当前已上线四种触发器：

触发器	说明
Schedule	基于 cron；流程按调度触发，并在每次应用启动时重新注册自己
App event	来自已连接集成的实时事件（新 Gmail 线程、Notion 变更、Linear 工单），按 toolkit + trigger slug 匹配，详见 Triggers
Manual	Workflows 页面的 Run 按钮，或 flows_run RPC
Resume	继续一个在审批门处暂停的运行

此外，每个流程都有派发锁（dispatch lock）：即使调度发生突发堆积，也绝不会并发运行同一个流程两次。

信任、审批与人机协同

每次流程运行都在一个专门的信任来源（TrustedAutomation → Workflow）下执行，这一点在 src/openhuman/flows/ops_part_01.rs 中可以得到印证（TrustedAutomation { Workflow } origin）。其设计逻辑是：流程的动作（调用哪些工具、访问哪些 URL）是你在保存时已批准的静态图配置；而运行时触发负载（webhook body、入站事件）保持不可信——它可以向这些预先声明的动作注入参数，但永远无法引入一个新的动作。

在此基础上，每个流程还有一个 "Require approval for outbound actions"（出站动作需审批） 开关。开启后，运行中每一个会产生外部副作用的工具调用或 HTTP 调用都会停靠在审批门（Approval Gate）前，等待真实决策；该运行的信任根不会自动放行任何东西。

当运行暂停时，你会在通知中收到一张 Flow Approval Card（流程审批卡），标明流程名称和挂起的步骤。批准后通过 flows_resume 从暂停处精确恢复。运行是持久化且有检查点（checkpointed）的，所以"晚点再处理"完全没问题——暂停状态通过 durable checkpointer 保存，即使核心重启也能恢复。

观察运行：从画布到 Run Inspector

• /flows：Workflows 中枢，展示每个流程的启用开关、最近运行状态（completed / pending approval / failed）以及 Run 按钮；
• /flows/:id：只读画布视图，以节点和连线渲染工作流图，让你看清自己批准的确切内容；
• Run Inspector：抽屉式面板，逐步骤展示每次运行（节点标签、发射的输出、最终状态），每 2 秒轮询一次直至运行结束；
• 完整运行历史：按流程持久化，包括状态、开始/结束时间、待处理审批、错误以及重建的逐步骤输出。

这些前端路径与组件可以在 app/src/AppRoutes.tsx（/flows 路由）以及 app/src/components/chat/WorkflowProposalCard.tsx、app/src/components/notifications/FlowApprovalCard.tsx 中找到对应实现。

面向开发者的 RPC 接口

flows 领域（src/openhuman/flows/）对外暴露一组 openhuman.flows_* 控制器。核心的十个控制器为：

RPC	用途
flows_create	从 tinyflows 图创建新流程
flows_get	按 id 加载单个流程
flows_list	列出已保存流程
flows_update	更新流程图/配置
flows_delete	删除流程
flows_set_enabled	启用/禁用流程（同时绑定/解绑自动派发）
flows_run	手动触发一次运行
flows_resume	从审批暂停点恢复运行
flows_list_runs	列出某流程的运行历史
flows_get_run	获取单次运行的详细信息（含逐步结果）

接口的 schema 定义集中在 src/openhuman/flows/schemas.rs 与其分卷 src/openhuman/flows/flows_schema_part_01.rs 中。除此之外，schema 层还注册了 flows_duplicate（复制流程，副本永远以 DISABLED 状态诞生且不绑定调度）、flows_validate（不落盘地校验图结构并返回非致命警告）、flows_import（解析原生 tinyflows 图或 n8n 导出、迁移校验后返回规范化图，同样不保存）等辅助控制器，均遵循"import 永不持久化、永不启用"的边界。

触发与派发的底层链路

流程的自动派发由 src/openhuman/flows/bus.rs 中的 FlowTriggerSubscriber 完成：它监听规范化后的事件（DomainEvent::FlowScheduleTick、ComposioTriggerReceived、WebhookIncomingRequest），把它们与已启用流程的触发器匹配，每命中一次就派生一次 flows::ops::flows_run。flows_set_enabled 在启用/禁用时复用同一套匹配辅助函数来绑定或解绑流程的自动派发。这与 Triggers 文档描述的端到端链路（第三方 webhook → 后端 HMAC 校验 → Socket.IO 事件 → Rust 核心事件总线 → Trigger Triage 分类）一致。

与周边能力的关系

• Cron & Scheduling（gitbooks/features/native-tools/cron.md）：cron_add 等工具负责一次性或周期性的 Agent 任务；Workflows 是面向多步骤、有结构、可审批的升级形态。
• Approval Gate（gitbooks/features/approval-gate.md）：审批门默认开启、fail-closed（10 分钟 TTL 超时即拒绝）；Workflow 的出站审批、命令分类与 auto_approve 白名单都由它承载，可配置项包括 OPENHUMAN_APPROVAL_GATE 与 [autonomy].level / [autonomy].auto_approve。
• Triggers（gitbooks/features/integrations/triggers.md）：连接集成产生的实时应用事件，是 app_event 工作流的触发来源。

小结

OpenHuman Workflows 的核心设计可以概括为一句话：Agent 负责构建，用户负责批准，引擎负责可靠执行，审批门负责兜底安全。通过 propose_workflow 提案 + flows_create 落盘的双通道隔离、始终 DISABLED 的新建流程、mock 沙箱 dry run、TrustedAutomation → Workflow 信任来源以及持久化检查点恢复，它在"自动化便利"与"人类可控"之间取得了明确的边界。对开发者而言，src/openhuman/flows/ 是理解整套机制的最佳入口——从节点契约（node_contracts）到构建工具（builder_tools），再到 RPC 面（schemas）与事件派发（bus），每一层都有对应的实现与测试可循。