很多朋友装好OpenClaw之后发现一个问题：这玩意儿确实能跑，但怎么让它"聪明"地干活？比如我想让它每天早上自动查一次天气预报、汇总关键信息后推给我，或者定时检查某个网站有没有更新——这些"自动化"的事情怎么做？答案就是工作流。今天这篇我就从最基础的概念讲起，配合几个实际的自动化任务编排案例，帮你快速上手OpenClaw的工作流设计。

OpenClaw最新版本一键部署包下载地址：https://top.wokk.cn/

什么是工作流

先别被"工作流"这个术语吓到。说白了，工作流就是"把一系列任务按顺序或条件串联起来自动执行"。

举个最简单的例子：你早上起床的流程就是个工作流——闹钟响了 → 起床 → 刷牙洗脸 → 煮咖啡 → 看手机。这些步骤有先后顺序，每一步都是触发下一步的条件。

放到OpenClaw里也是一样的道理。一个工作流通常包含这么几个要素：

触发器（Trigger）：什么时候开始执行。可以是定时触发（每天9点）、事件触发（收到一封邮件）、或者手动触发（你发一条指令）。
节点（Node）：具体要执行的操作。比如发送HTTP请求、调用AI模型、读写文件、发送消息等。
连接线（Edge）：节点之间的关系。顺序执行、条件分支、并行执行等。
变量（Variable）：节点之间传递的数据。比如第一个节点获取的天气数据，传给第二个节点用来生成摘要。

把这些东西组合在一起，就是一条完整的工作流。

第一个工作流：每日早报

理论讲完，直接上手。我们的第一个工作流目标是：每天早上8点自动汇总天气、新闻和日程，然后推送到微信。

用YAML格式定义工作流（OpenClaw支持YAML和可视化编辑器两种方式创建工作流，这里用YAML因为更直观）：

name: daily-morning-briefing
description: 每日早间简报工作流

trigger:
type: cron
schedule: “0 8 * * *” # 每天8点
timezone: Asia/Shanghai

nodes:

• id: get_weather
  type: http_request
  config:
  url: “https://api.weather.com/v1/current?city=Beijing&apikey={{SECRETS.WEATHER_API_KEY}}”
  method: GET
  output: weather_data

• id: get_news
  type: http_request
  config:
  url: “https://api.news.com/v1/top-headlines?country=cn”
  method: GET
  output: news_data

• id: summarize
  type: ai_generate
  config:
  model: glm-4
  prompt: |
  根据以下信息生成一份简明的早间报告，不超过300字：

  天气信息：{{weather_data}}
  
  今日要闻：{{news_data}}
  
  要求：口语化风格，突出关键信息，最后附一句今日建议。
  

  input:
  weather_data: “{{nodes.get_weather.output}}”
  news_data: “{{nodes.get_news.output}}”
  output: summary

• id: send_notification
  type: message_send
  config:
  channel: wechat
  target: “你的微信ID”
  content: “{{nodes.summarize.output}}”

edges:

• from: get_weather
  to: summarize
• from: get_news
  to: summarize
• from: summarize
  to: send_notification

来解释一下这段配置：

触发器部分用了cron表达式0 8 * * *，表示每天8:00执行。节点部分定义了4个步骤：获取天气、获取新闻、AI摘要、发送通知。

注意summarize节点。这里我们用了ai_generate类型，让AI模型把天气和新闻数据整合成一段可读的摘要。prompt里用了{{weather_data}}和{{news_data}}这样的模板变量，运行时会被替换为实际数据。

edges部分定义了执行顺序。get_weather和get_news可以并行执行（它们没有依赖关系），summarize等前两个都完成后才开始，最后发通知。

第二个工作流：网站监控

工作流不一定非得是定时执行的，也可以是基于条件的。下面这个例子是监控一个技术博客有没有新文章发布，有的话就推送通知。

name: blog-monitor
description: 监控博客更新

trigger:
type: cron
schedule: “*/30 * * * *” # 每30分钟检查一次

nodes:

• id: fetch_page
  type: http_request
  config:
  url: “https://example-blog.com”
  method: GET
  output: page_content

• id: check_update
  type: condition
  config:
  expression: |
  {{nodes.fetch_page.output}} contains “{{prev.latest_title}}”
  mode: “not_match” # 如果当前页面的最新标题跟上次不一样，说明有更新
  output: has_update

• id: extract_info
  type: ai_extract
  condition: “{{nodes.check_update.output}} == true”
  config:
  model: glm-4
  prompt: |
  从以下网页内容中提取最新文章的标题和摘要：
  {{nodes.fetch_page.output}}

  返回格式：标题：xxx\n摘要：xxx
  

  output: article_info

• id: notify
  type: message_send
  condition: “{{nodes.check_update.output}} == true”
  config:
  channel: telegram
  target: “你的Telegram Chat ID”
  content: “📰 博客更新：\n{{nodes.extract_info.output}}”

• id: save_state
  type: file_write
  condition: “{{nodes.check_update.output}} == true”
  config:
  path: “/data/blog_monitor/state.json”
  content: |
  {“latest_title”: “{{nodes.extract_info.output.split(‘\n’)[0].replace(‘标题：’, ‘’)}}”, “last_check”: “{{now()}}”}

这个工作流的亮点在于使用了condition类型节点和condition属性。check_update节点会比对当前页面的最新标题和上次保存的标题，如果不一样就继续执行后续的提取和通知步骤，否则整个流程就到此结束，不会发任何通知。

最后一个save_state节点很重要——它把当前检测到的最新标题保存到文件中，作为下次比对的基准。这就实现了"状态记忆"，让监控能持续运行而不是每次都重复检测。

第三个工作流：数据处理管道

工作流不只是能做通知类的任务，也能处理复杂的数据处理逻辑。下面这个例子展示了一个典型的数据处理管道：定期从API获取数据 → 清洗转换 → 存储 → 生成报表。

name: data-pipeline
description: 日报数据处理管道

trigger:
type: cron
schedule: “30 23 * * *” # 每天23:30执行

nodes:

• id: fetch_data
  type: http_request
  config:
  url: “{{SECRETS.DATA_API_URL}}”
  method: POST
  headers:
  Authorization: “Bearer {{SECRETS.DATA_API_TOKEN}}”
  body: |
  {“date”: “{{today()}}”, “type”: “daily”}
  output: raw_data

• id: validate
  type: script
  config:
  language: python
  code: |
  import json
  data = json.loads(“”“{{nodes.fetch_data.output}}”“”)
  # 检查必要字段是否存在
  required = [“date”, “records”, “summary”]
  for field in required:
  if field not in data:
  raise ValueError(f"缺少必要字段: {field}")
  # 过滤掉无效记录
  valid_records = [r for r in data[“records”] if r.get(“value”) is not None]
  data[“records”] = valid_records
  return json.dumps(data, ensure_ascii=False)
  output: cleaned_data

• id: store
  type: database_insert
  config:
  connection: “{{SECRETS.DB_CONNECTION_STRING}}”
  table: daily_reports
  data: “{{nodes.validate.output}}”

• id: generate_report
  type: ai_generate
  config:
  model: glm-4
  prompt: |
  根据以下数据生成一份日报分析：
  {{nodes.validate.output}}

  要求：
  1. 用表格展示关键指标
  2. 对比前一天的变化趋势
  3. 标注异常数据点
  4. 给出简要的总结和建议
  

  output: report

• id: save_report
  type: file_write
  config:
  path: “/data/reports/daily_{{today()}}.md”
  content: “{{nodes.generate_report.output}}”

• id: error_handler
  type: message_send
  condition: “{{nodes.validate.status}} == ‘error’”
  config:
  channel: wechat
  target: “管理员ID”
  content: “❌ 日报数据处理失败：{{nodes.validate.error_message}}”

edges:

• from: fetch_data
  to: validate
• from: validate
  to: store
• from: store
  to: generate_report
• from: generate_report
  to: save_report
• from: validate
  to: error_handler

这个工作流展示了好几个重要的设计模式：

脚本节点：validate节点用了Python脚本来做数据清洗和校验。这是工作流中最灵活的节点类型——你可以用它执行任何自定义逻辑。

错误处理：error_handler节点通过condition属性绑定，只有在validate节点出错时才会执行。这种设计模式叫做"旁路错误处理"，不会中断正常流程。

变量引用：配置中大量使用了{{SECRETS.xxx}}这样的变量引用，从密钥管理中获取敏感信息，避免把密码、Token之类的硬编码在工作流文件中。

工作流设计的几个原则

最后分享几个设计工作流时的心得，都是踩坑之后总结出来的。

每个节点只做一件事。别试图在一个节点里塞太多逻辑。拆得越细，调试越方便，复用性也越高。比如数据获取和数据清洗应该是两个独立的节点，而不是合在一个脚本里。

关键节点一定要有错误处理。网络请求可能超时，API可能返回错误，AI可能生成不合规的内容。每一个可能失败的节点都要有对应的fallback逻辑或者错误通知。

敏感信息永远不要硬编码。用OpenClaw提供的密钥管理功能，把API Key、数据库密码这些敏感信息统一存起来，工作流里只引用变量名。这样既安全，也方便在不同环境（开发、测试、生产）之间切换配置。

先手动跑通，再自动化。不要一开始就写定时触发的工作流。先把它改成手动触发，把每个节点的输出都检查一遍，确认数据流是对的，再切换成自动模式。

给工作流加日志。每个节点执行完之后记录一下时间、输入、输出。出了问题排查的时候，日志就是你的救命稻草。

OpenClaw的工作流系统设计得挺灵活的，能覆盖的自动化场景远不止上面这几个例子。你大可以发挥创意，把日常工作中那些重复性的操作都编排成工作流——一旦跑起来，你会发现效率提升不是一星半点。动手试试吧，第一个工作流跑通的感觉还是很爽的。