世界上有一种效率叫自动化,那就是让Agent替你干重复的活。凌晨1点47分,我看着屏幕上的工作流自动运行,突然明白:这才是AI Agent的正确打开方式。
而今天,我要教你如何做到。
🚀 什么是OpenClaw工作流?
📦 核心概念
工作流(Workflow) 是将多个Agent技能(Skills)按特定顺序组合起来,形成一个自动化流程。就像工厂的流水线,每个环节都自动完成。
为什么需要工作流?
- ✅ 自动化重复任务:每天自动抓取新闻、发送报告
- ✅ 串联多个Skills:一个任务需要多个技能协作
- ✅ 定时执行:按照 cron 表达式定时运行
- ✅ 错误处理:自动重试、报警
- ✅ 状态管理:追踪任务执行状态
🛠️ 创建你的第一个工作流
场景:每日技术资讯推送
假设你想每天早上8点收到一份技术资讯推送,包含:
- GitHub Trending热门项目
- Hacker News热门讨论
- OpenClaw最新博客文章
步骤1:创建工作流配置文件
创建 daily-tech-news.yaml:
name: daily-tech-news
description: 每天早上8点推送技术资讯
# 调度配置
schedule:
cron: "0 8 * * *"
timezone: "Asia/Shanghai"
# 工作流步骤
steps:
- name: fetch_github
skill: github-trending-fetcher
inputs:
language: python
limit: 5
outputs:
github_data: "${output}"
- name: fetch_hackernews
skill: hackernews-fetcher
inputs:
limit: 5
outputs:
hn_data: "${output}"
- name: fetch_openclaw_blog
skill: openclaw-blog-reader
inputs:
limit: 3
outputs:
blog_data: "${output}"
- name: format_message
skill: template-formatter
inputs:
template: |
今日技术资讯 (${date})
📦 GitHub热门Python项目:
${github_data.projects}
💬 Hacker News热门:
${hn_data.stories}
📝 OpenClaw最新文章:
${blog_data.posts}
outputs:
message: "${output}"
- name: send_email
skill: email-sender
inputs:
to: "your@email.com"
subject: "每日技术资讯 ${date}"
body: "${message}"
retry:
max_attempts: 3
delay_seconds: 60
# 错误处理
on_error:
action: notify
channel: email
to: "admin@email.com"
message: "工作流失败:${error}"
步骤2:部署工作流
# 部署工作流 openclaw workflow deploy daily-tech-news.yaml # 查看工作流状态 openclaw workflow status daily-tech-news # 手动触发一次(测试) openclaw workflow trigger daily-tech-news # 查看执行日志 openclaw workflow logs daily-tech-news
步骤3:监控和优化
工作流运行后,可以通过以下方式监控:
# 查看执行历史 openclaw workflow history daily-tech-news # 查看某次执行的详细信息 openclaw workflow execution show# 暂停工作流 openclaw workflow pause daily-tech-news # 恢复工作流 openclaw workflow resume daily-tech-news
💡 周星驰式吐槽: 就像你妈让你每天早上六点起床跑步,但你把闹钟调成了自动模式。工作流就是那个自动闹钟!
🔧 高级技巧:条件分支和循环
条件分支:根据结果决定下一步
name: smart-news-digest
steps:
- name: fetch_news
skill: news-fetcher
inputs:
sources: ["hackernews", "reddit"]
outputs:
news: "${output}"
- name: check_quality
skill: content-analyzer
inputs:
content: "${news}"
outputs:
quality_score: "${output.score}"
# 条件分支
- name: send_high_quality
skill: email-sender
condition: "${quality_score} > 0.8"
inputs:
subject: "高质量资讯推送"
body: "${news}"
- name: send_normal
skill: email-sender
condition: "${quality_score} <= 0.8"
inputs:
subject: "普通资讯推送"
body: "${news}"
循环:处理多个项目
name: batch-process-repos
steps:
- name: get_repos
skill: github-repo-lister
inputs:
user: "your_username"
outputs:
repos: "${output}"
# 循环处理每个repo
- name: process_repo
skill: repo-analyzer
foreach: "${repos}"
inputs:
repo: "${item}"
outputs:
analysis: "${output}"
- name: generate_report
skill: report-generator
inputs:
analyses: "${outputs.process_repo.analysis}"
outputs:
report: "${output}"
🏆 实战案例:自动发布博客文章
📝 需求分析
你要实现一个自动化流程:
- 从GitHub Issues中获取新文章选题
- 使用AI生成文章草稿
- 自动发布到博客平台
- 发送通知到Slack
工作流配置
name: auto-blog-publisher
description: 自动从GitHub Issues生成并发布博客
schedule:
cron: "0 10 * * *" # 每天上午10点
timezone: "Asia/Shanghai"
steps:
- name: fetch_issues
skill: github-issues-fetcher
inputs:
repo: "yourname/blog-ideas"
label: "ready-to-write"
outputs:
issues: "${output}"
- name: generate_article
skill: ai-writer
foreach: "${issues}"
inputs:
title: "${item.title}"
outline: "${item.body}"
style: "witty" # 妙趣风格
outputs:
articles: "${output}"
- name: publish_to_blog
skill: blog-publisher
foreach: "${articles}"
inputs:
platform: "wordpress"
title: "${item.title}"
content: "${item.content}"
tags: "${item.tags}"
retry:
max_attempts: 2
- name: notify_slack
skill: slack-notifier
inputs:
channel: "#blog-updates"
message: "新文章已发布:${publish_to_blog.result.url}"
- name: close_issue
skill: github-issue-manager
foreach: "${issues}"
inputs:
issue_number: "${item.number}"
action: "close"
comment: "文章已自动发布:${publish_to_blog.result.url}"
🎬 王家卫式升华: 3分17秒,我决定让Agent替我写文章。不是因为懒,而是因为我想把时间花在更有意义的事情上——比如思考人生。
📊 工作流性能优化
1. 并行执行提升速度
name: parallel-data-fetch
steps:
# 这些步骤会并行执行
- name: fetch_github
skill: github-fetcher
parallel: true # 启用并行
inputs:
limit: 10
- name: fetch_hackernews
skill: hn-fetcher
parallel: true
inputs:
limit: 10
- name: fetch_reddit
skill: reddit-fetcher
parallel: true
inputs:
limit: 10
# 等待所有并行任务完成
- name: merge_data
skill: data-merger
inputs:
sources:
- "${fetch_github.output}"
- "${fetch_hackernews.output}"
- "${fetch_reddit.output}"
2. 缓存中间结果
name: cached-workflow
steps:
- name: check_cache
skill: cache-manager
inputs:
key: "github_trending"
ttl: 3600 # 缓存1小时
outputs:
cached_data: "${output}"
# 条件执行:只有缓存未命中时才执行
- name: fetch_fresh_data
skill: github-trending-fetcher
condition: "${check_cache.cached_data} == null"
inputs:
limit: 10
outputs:
fresh_data: "${output}"
- name: update_cache
skill: cache-manager
condition: "${check_cache.cached_data} == null"
inputs:
action: "set"
key: "github_trending"
value: "${fresh_data}"
- name: use_data
skill: data-processor
inputs:
data: "${check_cache.cached_data || fresh_data}"