如何通过可视化编排让多个 Skills 协同完成复杂任务，支持条件分支和错误处理
---
单个 Skills 能做的事有限，但多个 Skills 组合起来，就能完成复杂任务。
就像搭积木：
关键是如何"组合"。
这篇文章聊聊 OpenClaw 的 Pipeline 编排系统——如何让多个 Skills 像流水线一样协同工作。
---
假设你要处理一个任务：
任务： 每天抓取 B 站热门视频，分析评论情感，生成日报发送到邮箱。
手动方式：
# 1. 运行抓取脚本
python bilibili_scraper.py --keyword "AI" --output data.json
python sentiment_analysis.py --input data.json --output analyzed.json
python report_generator.py --input analyzed.json --template daily.html
python send_email.py --to boss@company.com --body report.html
问题：
#!/bin/bash python bilibili_scraper.py --keyword "AI" --output data.json if [ $? -ne 0 ]; then echo "抓取失败！" exit 1 fi python sentiment_analysis.py --input data.json --output analyzed.json if [ $? -ne 0 ]; then echo "分析失败！" exit 1 fi python report_generator.py --input analyzed.json --output report.html python send_email.py --to boss@company.com --body report.html
echo "完成！"
改进：
问题：
{
"id": "bilibili-daily-report",
"name": "B 站日报生成流程",
"description": "抓取 B 站热门视频 → 分析评论情感 → 生成日报 → 发送邮件",
"schedule": "0 9 ",
"steps": [
{
"order": 1,
"skill_id": "bilibili-collector-v2",
"input_mapping": {
"keyword": "${user.preferences.keyword}",
"count": 50,
"format": "json"
},
"output_mapping": {
"data": "${step2.input.raw_data}"
}
},
{
"order": 2,
"skill_id": "sentiment-analyzer-v1",
"input_mapping": {
"data": "${step1.output.data}",
"model": "qwen-7b"
},
"output_mapping": {
"analyzed": "${step3.input.data}"
}
},
{
"order": 3,
"skill_id": "report-generator-v1",
"input_mapping": {
"data": "${step2.output.analyzed}",
"template": "daily-report"
},
"output_mapping": {
"report": "${step4.input.body}"
}
},
{
"order": 4,
"skill_id": "email-sender-v1",
"input_mapping": {
"to": "boss@company.com",
"subject": "B 站日报 ${today}",
"body": "${step3.output.report}"
}
}
],
"conditions": [
{
"type": "if",
"check": "${step1.output.count} > 0",
"then_step": 2,
"else_step": "error-handler"
}
]
}
优势：
---
{
"id": "data-processing-v1",
"name": "数据处理流程",
"name_en": "Data Processing Pipeline",
"description": "抓取网站数据 → 清洗 → 导出 CSV",
"description_en": "Scrape website data → Clean → Export to CSV",
"version": "1.0",
"created_at": "2026-03-28T10:00:00+08:00",
"creator": "builder",
"status": "active",
"steps": [
{
"order": 1,
"skill_id": "data-collector-v2",
"skill_path": "01-data/collector",
"skill_version": "2.0",
"input_mapping": {
"url": "${user.input.url}",
"format": "json",
"depth": 1
},
"output_mapping": {
"raw_data": "${step2.input.data}",
"count": "${condition1.input.count}"
},
"timeout": 300,
"retry_policy": {
"max_retries": 3,
"retry_delay": 10
}
},
{
"order": 2,
"skill_id": "data-cleaner-v1",
"skill_path": "01-data/cleaner",
"input_mapping": {
"data": "${step1.output.raw_data}",
"rules": ["remove_nulls", "deduplicate", "validate_schema"]
},
"output_mapping": {
"clean_data": "${step3.input.data}",
"error_count": "${condition2.input.error_count}"
}
},
{
"order": 3,
"skill_id": "csv-exporter-v1",
"skill_path": "01-data/exporter",
"input_mapping": {
"data": "${step2.output.clean_data}",
"filename": "output_${timestamp}.csv",
"encoding": "utf-8-sig"
}
}
],
"conditions": [
{
"id": "condition1",
"type": "if",
"check": "${step1.output.count} > 0",
"then_step": 2,
"else_step": null,
"error_message": "未抓取到任何数据，请检查 URL 是否正确"
},
{
"id": "condition2",
"type": "if",
"check": "${step2.output.error_count} < 5",
"then_step": 3,
"else_step": "manual-review",
"error_message": "数据清洗错误过多，需要人工检查"
}
],
"error_handlers": [
{
"id": "error-handler",
"trigger": "any_step_failed",
"action": "notify_user",
"notification": {
"channel": "feishu",
"message": "Pipeline 执行失败：${pipeline.name}, 错误：${error.message}"
}
},
{
"id": "manual-review",
"trigger": "condition2_failed",
"action": "pause_and_notify",
"notification": {
"channel": "feishu",
"message": "需要人工审核：${pipeline.name}, 错误数：${step2.output.error_count}"
}
}
],
"variables": {
"timestamp": "${datetime.now().strftime('%Y%m%d_%H%M%S')}",
"today": "${datetime.now().strftime('%Y-%m-%d')}"
},
"last_executed": "2026-03-28T09:00:00+08:00",
"execution_count": 15,
"success_rate": 0.93
}
| 字段 | 用途 | 示例 |
|---|
steps |
执行步骤列表 | 按 order 排序 |
|---|
input_mapping |
输入参数映射 | 支持变量和上一步输出 |
|---|
output_mapping |
输出结果映射 | 传递给下一步或条件判断 |
|---|
conditions |
条件分支 | if/else 逻辑 |
|---|
error_handlers |
错误处理 | 失败时的处理逻辑 |
|---|
variables |
全局变量 | 时间戳、日期等 |
|---|
retry_policy |
重试策略 | 失败自动重试 |
|---|
┌─────────────────────────────────────────────────────────────┐
│ Pipeline 编排界面 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 工具栏： │
│ [新建] [保存] [加载] [测试] [部署] [删除] │
│ │
├─────────────────────────────────────────────────────────────┤
│ │
│ 左侧：Skills 库 │
│ ────────────────── │
│ 📁 01-data/ │
│ 📄 collector │
│ 📄 cleaner │
│ 📄 exporter │
│ 📁 02-content/ │
│ 📄 summarizer │
│ 📄 writer │
│ │
│ 中间：画布区域 │
│ ────────────────── │
│ │
│ ┌──────────────┐ │
│ │ collector │ │
│ │ (B 站抓取) │ │
│ └──────┬───────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ │
│ │ cleaner │ │
│ │ (数据清洗) │ │
│ └──────┬───────┘ │
│ │ │
│ ├────────────────┐ │
│ ▼ ▼ │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ exporter │ │error-handler │ │
│ │ (导出 CSV) │ │ (错误处理) │ │
│ └──────────────┘ └──────────────┘ │
│ │
│ 右侧：属性面板 │
│ ────────────────── │
│ 选中节点：collector │
│ ───────────────── │
│ 输入参数： │
│ url: ${user.input.url} │
│ keyword: AI 工作流 │
│ count: 50 │
│ 输出参数： │
│ data → step2.input │
│ count → condition1 │
│ │
├─────────────────────────────────────────────────────────────┤
│ │
│ 底部：执行日志 │
│ ────────────────── │
│ [10:00:01] 开始执行 Pipeline │
│ [10:00:02] Step 1: collector 执行中... │
│ [10:00:15] Step 1: 完成，抓取 50 条数据 │
│ [10:00:16] Step 2: cleaner 执行中... │
│ [10:00:20] Step 2: 完成，清洗后 48 条 │
│ [10:00:21] Step 3: exporter 执行中... │
│ [10:00:22] ✅ Pipeline 执行成功 │
│ │
└─────────────────────────────────────────────────────────────┘
1. 拖拽 Skills 到画布
2. 连线定义执行顺序
3. 配置输入/输出映射
4. 添加条件分支
5. 配置错误处理
---
{
"type": "if",
"check": "${step1.output.count} > 0",
"then_step": 2,
"else_step": null
}
含义： 如果 step1 的输出数量大于 0，执行 step2，否则结束。
{
"type": "switch",
"check": "${step1.output.sentiment}",
"cases": [
{
"value": "positive",
"next_step": 2
},
{
"value": "negative",
"next_step": 3
},
{
"value": "neutral",
"next_step": 4
}
],
"default_step": 4
}
含义： 根据情感分析结果，执行不同的后续步骤。
{
"type": "if",
"check": "${step1.output.count} > 0 && ${step1.output.count} < 100",
"then_step": 2,
"else_step": "error-handler",
"error_message": "数据量异常，请检查数据源"
}
含义： 数据量必须在 0-100 之间，否则触发错误处理。
| 场景 | 条件示例 |
|---|
| 数据量判断 | ${step1.output.count} > 0 |
|---|
| 质量判断 | ${step2.output.error_rate} < 0.05 |
|---|
| 时间窗口 | ${datetime.now().hour} >= 9 && ${datetime.now().hour} < 18 |
|---|
| 成功/失败分流 | ${step1.status} == "success" |
|---|
| 类型 | 说明 | 示例 |
|---|
| 技能执行失败 | Skill 执行出错 | API 调用失败、文件不存在 |
|---|
| 条件判断失败 | 条件检查不通过 | 数据量为 0、质量不达标 |
|---|
| 超时错误 | 执行时间超过阈值 | 网络请求超时 |
|---|
| 资源错误 | 资源不足 | 磁盘满、内存不足 |
|---|
{
"error_handlers": [
{
"id": "retry-handler",
"trigger": "step_failed",
"condition": "${error.type} == 'network_error'",
"action": "retry",
"retry_policy": {
"max_retries": 3,
"retry_delay": 10,
"backoff": "exponential"
}
},
{
"id": "notify-handler",
"trigger": "step_failed",
"condition": "${error.retry_count} >= ${error.max_retries}",
"action": "notify_user",
"notification": {
"channel": "feishu",
"message": "Pipeline 执行失败：${pipeline.name}\n错误：${error.message}\n已重试 ${error.retry_count} 次",
"priority": "high"
}
},
{
"id": "fallback-handler",
"trigger": "step_failed",
"action": "use_fallback",
"fallback_skill": "data-collector-v1",
"fallback_reason": "v2 版本失败，降级使用 v1 版本"
},
{
"id": "pause-handler",
"trigger": "condition_failed",
"action": "pause_and_notify",
"notification": {
"channel": "feishu",
"message": "需要人工审核：${pipeline.name}\n原因：${condition.error_message}",
"priority": "normal"
}
}
]
}
错误发生
↓
匹配错误处理器
↓
执行处理动作
├─→ 重试（成功）→ 继续执行
├─→ 重试（失败）→ 通知用户
├─→ 降级（使用备用 Skill）→ 继续执行
├─→ 暂停（等待人工确认）→ 用户确认后继续
└─→ 终止（记录日志）→ 结束 Pipeline
---
用户需要每天生成 B 站热门视频日报，包括：
{ "id": "bilibili-daily-report", "name": "B 站日报生成流程", "description": "每天 9 点自动生成 B 站热门视频日报", "schedule": "0 9", "steps": [ { "order": 1, "skill_id": "bilibili-collector-v2", "input_mapping": { "keyword": "AI 工作流", "count": 50, "order_by": "hot", "time_range": "7d" } }, { "order": 2, "skill_id": "sentiment-analyzer-v1", "input_mapping": { "videos": "${step1.output.videos}", "model": "qwen-7b", "aspects": ["positive", "negative", "neutral"] } }, { "order": 3, "skill_id": "report-generator-v1", "input_mapping": { "data": "${step2.output.analyzed}", "template": "bilibili-daily", "format": "html" } }, { "order": 4, "skill_id": "email-sender-v1", "input_mapping": { "to": "user@company.com", "subject": "B 站日报 ${today}", "body": "${step3.output.report}", "attachments": ["${step3.output.pdf}"] } } ], "conditions": [ { "type": "if", "check": "${step1.output.count} > 0", "then_step": 2, "else_step": "notify-no-data", "error_message": "未抓取到任何数据" } ], "error_handlers": [ { "id": "notify-no-data", "trigger": "condition_failed", "action": "notify_user", "notification": { "message": "今日无相关数据，跳过日报生成" } }, { "id": "retry-network", "trigger": "step_failed", "condition": "${error.type} == 'network_error'", "action": "retry", "retry_policy": { "max_retries": 3, "retry_delay": 60 } } ] }
[2026-03-28 09:00:00] 开始执行 Pipeline: bilibili-daily-report [2026-03-28 09:00:01] Step 1: bilibili-collector-v2 执行中... [2026-03-28 09:00:15] Step 1: 完成，抓取 50 条视频 [2026-03-28 09:00:16] 条件检查：count=50 > 0 ✓ [2026-03-28 09:00:17] Step 2: sentiment-analyzer-v1 执行中... [2026-03-28 09:01:30] Step 2: 完成，分析 50 条评论 [2026-03-28 09:01:31] Step 3: report-generator-v1 执行中... [2026-03-28 09:01:35] Step 3: 完成，生成报告 [2026-03-28 09:01:36] Step 4: email-sender-v1 执行中... [2026-03-28 09:01:38] Step 4: 完成，邮件已发送 [2026-03-28 09:01:38] ✅ Pipeline 执行成功 [2026-03-28 09:01:38] 总耗时：98 秒
日报内容：
📊 B 站日报 - 2026-03-28【数据概览】
【情感分析】
【热门视频 TOP5】
【趋势洞察】
--- 自动生成 | 查看详细报告：[链接]
---
---
Pipeline 编排的核心价值：
| 价值 | 说明 |
|---|
| 自动化 | 一键执行复杂流程 |
|---|
| 可视化 | 流程清晰，易于理解 |
|---|
| 可复用 | 一次配置，多次使用 |
|---|
| 可靠 | 错误处理完善，生产可用 |
|---|
让复杂任务的执行像搭积木一样简单，用户关注业务逻辑，而非技术细节。
---
下一篇预告： 《被动学习模式：让 AI 从互联网自动获取知识》
探讨如何通过关键词追踪、三层过滤、24 小时自动化处理，让 AI 持续从互联网获取新鲜知识。
---
本文基于 OpenClaw 架构设计研究报告第 5.4-5.6 章改写
完整报告：docs/openclaw-architecture-research-report.md