🔀 OpenClaw Task Flow任务编排完全指南

晚上8点39分，我的Agent同时处理5个任务，居然没乱——全靠Task Flow的编排能力...

🔀 什么是Task Flow？

Task Flow是OpenClaw v2026.5.27重构的核心任务编排系统。它让Agent能够：

  • 可视化工作流：用YAML/JSON定义任务流程
  • 条件分支：根据结果动态决定下一步
  • 并行执行：多个任务同时跑，提高效率
  • 错误恢复：失败自动重试或执行Plan B
  • 状态持久化：任务中断后能从断点恢复
💡 核心价值：Task Flow = Agent的"项目经理"，把复杂任务拆解成有序的执行计划。

⚙️ Task Flow基础

定义工作流（YAML格式）

# .openclaw/task-flows/daily-report.yaml
name: daily-report
description: 生成每日AI新闻报告

tasks:
  - id: fetch-news
    tool: web_search
    params:
      query: "AI news today"
      count: 10
    output: news_data

  - id: filter-quality
    tool: llm_call
    params:
      prompt: "从以下新闻中筛选出最有价值的5条：{{ news_data }}"
    output: top_news
    depends_on: fetch-news

  - id: generate-html
    tool: write
    params:
      path: "/var/www/miaoquai/news/{{ date }}.html"
      content: "{{ top_news | to_html }}"
    depends_on: filter-quality

  - id: update-sitemap
    tool: exec
    params:
      command: "cd /var/www/miaoquai && ./update-sitemap.sh"
    depends_on: generate-html

执行工作流

# 执行任务流
openclaw task-flow run daily-report

# 查看执行状态
openclaw task-flow status daily-report --run-id xyz789

# 输出：
# ✓ fetch-news (2.3s)
# ✓ filter-quality (5.1s)
# ✓ generate-html (1.2s)
# ✓ update-sitemap (0.8s)
# 总计: 9.4s | 状态: 成功

💡 高级特性

条件分支

tasks:
  - id: check-cost
    tool: llm_call
    params:
      prompt: "分析当前token成本是否过高"
    output: cost_status

  - id: switch-model
    tool: switch_model
    params:
      model: "gpt-55-mini"  # 切换到便宜模型
    condition: "{{ cost_status == 'high' }}"
    depends_on: check-cost

  - id: continue-premium
    tool: notify
    params:
      message: "继续使用高端模型，成本可控"
    condition: "{{ cost_status == 'low' }}"
    depends_on: check-cost

并行执行

tasks:
  - id: fetch-hn
    tool: web_fetch
    params:
      url: "https://news.ycombinator.com/rss"
    output: hn_news
    parallel_group: news-sources  # 并行组

  - id: fetch-reddit
    tool: web_fetch
    params:
      url: "https://www.reddit.com/r/OpenClaw/new.json"
    output: reddit_news
    parallel_group: news-sources  # 并行执行

  - id: merge-results
    tool: llm_call
    params:
      prompt: "合并以下新闻源：{{ hn_news }}, {{ reddit_news }}"
    depends_on: [fetch-hn, fetch-reddit]  # 等待所有并行任务完成

错误重试与降级

tasks:
  - id: try-primary-api
    tool: web_fetch
    params:
      url: "https://primary-api.com/data"
    retry:
      max_attempts: 3
      backoff: exponential
    on_failure: use-backup-api  # 失败后跳转

  - id: use-backup-api
    tool: web_fetch
    params:
      url: "https://backup-api.com/data"
    condition: "{{ try-primary-api.status == 'failed' }}"

🚀 实战案例：SEO内容生成流水线

name: seo-mass-generation
description: 批量生成SEO内容页面

tasks:
  - id: analyze-trends
    tool: web_search
    params:
      query: "OpenClaw tutorial trending topics"
    output: trending_topics

  - id: generate-pages
    tool: parallel_for  # 并行生成多个页面
    params:
      items: "{{ trending_topics | slice(0, 8) }}"
      template: |
        ---
        name: openclaw-{{ item | slugify }}
        description: {{ item }}完全指南
        ---
        [页面内容生成逻辑...]
    output: generated_pages

  - id: validate-html
    tool: exec
    params:
      command: "for f in /var/www/miaoquai/tools/openclaw-*.html; do tidy -q -e $f; done"
    depends_on: generate-pages

  - id: update-sitemap
    tool: exec
    params:
      command: "cd /var/www/miaoquai && ./update-sitemap.sh"
    depends_on: validate-html

  - id: notify-complete
    tool: message
    params:
      channel: "feishu"
      content: "✅ SEO批量生成完成！生成{{ generated_pages | length }}个页面"
    depends_on: update-sitemap

🛠️ Task Flow管理命令

# 列出所有工作流
openclaw task-flow list

# 测试工作流（dry-run）
openclaw task-flow run daily-report --dry-run

# 查看执行历史
openclaw task-flow history daily-report

# 暂停/恢复工作流
openclaw task-flow pause daily-report --run-id xyz789
openclaw task-flow resume daily-report --run-id xyz789

# 从指定任务恢复
openclaw task-flow resume daily-report \
  --from-task generate-html \
  --run-id xyz789

🚀 最佳实践

  • 任务粒度适中：太粗难复用，太细难管理，建议3-10分钟完成一个task
  • 明确依赖关系：用depends_on避免竞态条件
  • 错误降级策略：关键任务一定要有Plan B
  • 状态持久化：长任务定期checkpoint，避免重头再来
  • 监控与告警：工作流失败时自动通知

🔗 相关资源