Openclaw Workflow

OpenClaw Workflow — 确定性工作流引擎

在不破坏 OpenClaw 灵活性的前提下，按 YAML 剧本执行 100% 确定性逻辑：判断、循环、脚本、LLM 调用、Skill 调用。

这是一个符合 OpenClaw / AgentSkills 目录约定的 Skill：

• SKILL.md：触发条件与使用说明
• scripts/：可执行入口与运行时代码
• references/：参考文档与示例工作流

使用方法

重要: 脚本内置了 stdout 行缓冲，不需要额外设置 PYTHONUNBUFFERED。

# 运行工作流
python3 {baseDir}/scripts/openclaw_workflow.py execute <workflow.yaml>

# 验证工作流语法
python3 {baseDir}/scripts/openclaw_workflow.py validate <workflow.yaml>

# 列出可用工作流
python3 {baseDir}/scripts/openclaw_workflow.py list

# 查看历史运行
python3 {baseDir}/scripts/openclaw_workflow.py runs

# 从断点恢复
python3 {baseDir}/scripts/openclaw_workflow.py resume <run_id>

# 可视化面板
python3 {baseDir}/scripts/openclaw_workflow.py dashboard

工作流文件放在哪里

推荐目录（OpenClaw 自建/自维护流程）：

~/.openclaw/workspace/workflows

面板与 CLI 会优先发现以下位置：

1. {baseDir}/references/examples
2. ~/.openclaw/workspace/workflows

说明：面板只扫描固定目录，不会递归读取整个 ~/.openclaw/workspace。

示例:

# 运行示例工作流
python3 {baseDir}/scripts/openclaw_workflow.py execute {baseDir}/references/examples/basic_test.yaml

# 运行深度集成测试
python3 {baseDir}/scripts/openclaw_workflow.py execute {baseDir}/references/examples/deep_integration.yaml

# 查看补充设计说明
cat {baseDir}/references/readme.md

输出格式

JSON 运行记录，包含：

• run_id — 运行唯一 ID
• flow_id — 工作流名称
• status — success | failed | aborted
• steps — 每步执行结果（状态、输出、耗时、重试次数）
• started_at / finished_at — 起止时间

终端同时输出人可读的步骤进度日志。

OpenClaw 深度绑定

• llm / agent / skill 节点通过 Gateway RPC 调用，不是本地直连模型。
• 同一工作流运行内所有调用共享同一个 session，Agent 拥有完整对话上下文。
• 模型由 OpenClaw Agent/Provider 配置统一管理，无需在工作流中指定。

工作流 YAML 格式

name: "我的流程"
steps:
  - id: fetch
    type: script
    command: "curl -s https://api.example.com/data"

  - id: analyze
    type: llm
    prompt: "分析: {{fetch.output}}"

  - id: delegate
    type: subagent
    task: "根据以下分析结果生成一份详细报告: {{analyze.text}}"
    label: "报告生成器"
    wait: true
    timeout: 300

  - id: notify
    type: message
    channel: imessage
    target: "+861760051xxxx"
    message: "结果: {{delegate.result}}"

节点类型速查

变量传递

• {{variable}} — 全局变量
• {{step_id.output}} / {{step_id.text}} — 步骤输出
• {{item}} — 循环当前元素
• {{env.HOME}} — 环境变量

错误处理

每步可配置: retry: 3, retry_delay: 10, on_error: retry|skip|stop

典型场景

• 每日简报: 获取数据 → AI 分析 → 格式化 → 发送
• 发票归档: 扫描邮件 → 识别发票 → 上传 NAS
• 固件打包: 拉取代码 → 构建 → 测试 → 打包 → 通知
• 数据监控: 定时检查 → 条件判断 → 告警
• 并行研究: 创建多个 subagent 分头调研 → 汇总结果

Subagent 节点

subagent 节点通过 OpenClaw sessions_spawn 工具创建独立子代理。

与 agent 节点的区别:

• agent: 在共享 session 中调用主 Agent，同一上下文
• subagent: 创建独立子代理，有自己的 session、系统提示和模型配置

创建流程

每个 subagent 的创建需要两层 session:

1. Spawn session — 用于承载 sessions_spawn 工具调用的"载体 session"
2. Child session — 由 sessions_spawn 在 Gateway 侧创建的实际子代理 session (agent:main:subagent:<uuid>)

sessions_spawn 是 Agent 工具 (不是 Gateway 直接 RPC)，所以必须通过 agent_call 在某个 session 中触发。spawn session 本身只是工具调用的容器，子代理真正执行任务的是 child session。

两种 Spawn Session 模式

模式 A: 传统模式 (独立 session)

每个 spawn 创建一个临时的 spawn:<hex> session → agent_call 让 Agent 调用 sessions_spawn → 提取 childSessionKey → 立即删除 spawn session。

• Session 数: N 个 spawn (瞬态，创建后立删) + N 个 child = 持久 N 个
• 并发: session 独立，天然支持并行
• 上下文: 每次全新 session，无膨胀问题

模式 B: 工厂模式 (共享 session) — 当前默认

所有 spawn 复用同一个 factory:<hex> session → 每 20 次轮换新 session → 用 factory_lock 序列化访问。

• Session 数: ~ceil(N/20) 个 factory + N 个 child ≈ N + 几个
• 并发: factory_lock 强制所有 spawn 串行执行，即使在并行循环中
• 上下文: 累积增长，需要定期轮换

✅ 已解决问题 (2026-03-17 → 2026-03-18)

原问题 1: 循环中工厂模式强制串行

已通过批量 spawn 解决。循环中自动检测并使用批量创建。

原问题 2: Gmail 24 封邮件超时

批量 spawn 10 个 subagent 仅需 ~77s，24 个预计 ~90-120s，远低于超时限制。

2026-03-18 架构更新: 子会话模式

非循环场景下的单个 subagent 节点不再通过 sessions_spawn 间接创建，而是直接创建新会话执行任务。 会话就是 subagent，省去了 spawn 中间层。

参数:

示例 — fire-and-forget:

- type: subagent
  task: "整理今天的新闻摘要"
  label: "新闻助手"

示例 — 等待结果:

- id: research
  type: subagent
  task: "研究 {{topic}} 并写一份 500 字的分析报告"
  label: "研究员"
  wait: true
  timeout: 600
  poll_interval: 20

- type: log
  message: "研究结果: {{research.result}}"

Wait Subagents 节点

wait_subagents 节点实现标准的 subagent fan-in (汇合) 模式:

• 在 loop 中通过 subagent 节点 (wait: false) 创建多个并行子代理
• 用 code 节点在 loop 中收集 spawn 信息到 tracker 列表
• wait_subagents 节点轮询所有子代理的 JSONL completion event，全部完成后返回结果列表

完成检测 (两级策略)

• 策略 A (精确): 读取 spawn session 的 JSONL，查找 auto-announce completion event。仅在 spawn session 未被清理时可用 (wait=true 或工厂模式)
• 策略 B (直读): 直接读取 child session 的 JSONL，检查是否有 assistant 回复。这是 wait=false 模式的主要检测路径，因为 spawn session 在创建后立即清理

自动清理 (三层保障)

• 即时清理: 子会话模式 wait=true 完成后立即删除 child session
• 汇合清理: wait_subagents 节点完成后批量清理所有 child session (cleanup: auto)
• 兆底清理: 工作流结束时 (无论成功/失败/中断)，engine.py 的 finally 块会清理:
  • 工作流主 session (agent:main:openclaw-workflow:<ns>)
  • 所有残留的 spawn/factory/child session
  • tracker 中记录的 child session
  • 工厂模式内部追踪的 child session (via get_factory_child_session_keys)

并发控制

subagent 节点在创建时会检查两个配置:

• agents.defaults.subagents.maxConcurrent (子代理专属限制，默认 20)
• agents.defaults.maxConcurrent (Gateway 全局嵌入式运行并发限制，默认 4)

取两者较小值作为实际限制。并发计数仅统计 :spawn: session (瞬态占位)，不统计 :subagent: session (因为完成后仍残留在 sessions.json 中会导致误判)。

内置重试

subagent 节点对 Gateway timeout 错误内置 2 次重试 (spawn_retries=2)，spawn 超时时间 120s (spawn_timeout=120000ms)。

参数:

示例 — 并行分类:

# 1) 初始化 tracker
- id: init_tracker
  type: code
  python: "result = []"

# 2) loop 中 fire-and-forget spawn + 收集信息
- type: loop
  foreach: "{{items}}"
  as: item
  do:
    - id: spawn_task
      type: subagent
      task: "分析: {{item.data}}"
      wait: false
    - id: collect
      type: code
      python: |
        tracker = init_tracker if isinstance(init_tracker, list) else []
        tracker.append({
            "item_id": item.get("id", ""),
            "spawn_session_key": spawn_task.get("spawn_session_key", ""),
            "child_session_key": spawn_task.get("child_session_key", ""),
        })
        result = tracker

# 3) 等待全部完成
- id: wait_all
  type: wait_subagents
  tracker: "{{init_tracker}}"
  max_wait: 600
  poll_interval: 5
  extra_fields:
    - item_id

# 4) 使用结果
- type: agent
  message: "汇总: {{wait_all}}"

全节点详解（参数 + 示例）

下面补全每个节点的常用写法，优先覆盖实际引擎支持的字段。

通用字段（所有节点都可用）

1) script 节点

执行 shell 命令或内联 Python。

支持字段：

• command / script / file：三选一
• inline：内联 Python 代码（会写入临时 .py 执行）
• timeout：默认 300 秒
• cwd：工作目录
• env：环境变量字典

- id: fetch_data
  type: script
  command: "curl -s https://api.example.com/data"
  timeout: 30

- id: build_report
  type: script
  inline: |
    import json
    print(json.dumps({"ok": True, "ts": "{{env.HOME}}"}, ensure_ascii=False))

2) llm 节点

通过 Gateway 调用模型推理，默认复用当前工作流会话。

支持字段：

• prompt（必填）
• thinking：off|minimal|low|medium|high
• timeout：秒，默认 120
• session：shared（默认）或 isolated

- id: summarize
  type: llm
  prompt: "请总结以下内容：{{fetch_data.output}}"
  thinking: low
  session: shared

3) agent 节点

调用主 Agent。与 llm 相比，支持投递能力（deliver）。

支持字段：

• message（必填）
• thinking
• timeout：默认 300
• deliver：是否自动投递
• deliver_channel / deliver_target

- id: agent_reply
  type: agent
  message: "根据 {{summarize.text}} 输出客户可读版本"
  deliver: true
  deliver_channel: imessage
  deliver_target: "+8617600510003"