Multi-Agent Pro

安装指南

1. ClawHub CLI 从源直接安装技能的最快方式。
   npx clawhub@latest install multi-agent-pro
2. 手动安装 将技能文件夹复制到以下位置之一
   全局模式
   ~/.openclaw/skills/
   工作区
   /skills/
   优先级：工作区 > 本地 > 内置
3. 提示词安装 将此提示词复制到 OpenClaw 即可自动安装。
   请帮我使用 Clawhub 安装 multi-agent-pro。如果尚未安装 Clawhub，请先安装（npm i -g clawhub）。
   更倾向于下载？
   获取包含原生技能文件的ZIP压缩包。
   下载技能 (.zip)

什么是

Multi-Agent Pro 提供工作流基础设施，可将单个请求转化为由 AI 智能体任务协同组成的序列。在 JSON DAG 中定义智能体和依赖关系，验证流水线，然后使用编排器管理执行顺序、持久化输出，并将状态传递给下游智能体。AI 仍负责执行每项任务；脚本则提供调度、状态管理、恢复和报告功能。
对于正在探索 Openclaw Skills 的团队，此技能无需第三方 Python 依赖，即可为复杂工作流提供实用控制。它支持条件路由、可复用子流水线、人工审批、重试、检查点恢复、执行历史、成本摘要和可视化报告，所有功能均设计为可在本地运行。

应用场景

协调多个 AI 智能体共同完成数据收集、分析和报告生成工作流。
验证流水线定义，并在运行前检查基于依赖关系的执行计划。
从已保存的状态恢复失败或中断的工作流，而不是重复执行已完成的节点。
在继续下游工作之前，暂停在审核或审批检查点。
根据结果路由任务、动态处理列表，或重复执行步骤，直到质量检查通过。
为发票处理或文档审核等重复流程复用已注册的子流水线。
在生成的报告中查看执行时长、节点结果、重试次数以及 token 或成本数据。
从提供的十个工作流模板之一开始，或使用交互式 init 指南创建流水线。

工作原理

描述任务及其阶段。AI 会根据需求生成流水线 JSON 定义，也可以从提供的模板开始。
使用编排器验证 DAG。验证会检查必需结构、依赖循环、孤立节点和执行顺序。
初始化运行。编排器会创建本地 JSON 状态文件，用于记录节点状态、输入、输出和运行数据。
逐步执行工作流。AI 执行每项任务，编排器则选择符合条件的节点，并将其已保存的输出共享给下游任务。
处理决策和故障。控制流节点负责路由执行；重试和回退策略负责处理错误；审批节点可以暂停以等待用户决策。
按需恢复或检查。查看状态、从检查点恢复、还原节点快照，或比较以往的执行记录。
生成输出。创建 Markdown 报告、HTML 甘特图或两者兼有，并附上执行详情和可用的成本摘要。

配置指南

要求：Python 3.8 或更高版本。脚本使用 Python 标准库，无需第三方软件包。使用该技能的安装目录及其中包含的 scripts/、templates/ 和参考文件。
检查 Python 版本：
python --version
使用交互式模板指南创建流水线，或复制并编辑提供的 DAG 模板：
python orchestrator.py init
验证并预览流水线，然后初始化执行：
python scripts/orchestrator.py validate pipeline.json
python scripts/orchestrator.py plan pipeline.json
python scripts/orchestrator.py run pipeline.json
每次执行一个节点并检查进度：
python scripts/orchestrator.py step state.json
python scripts/orchestrator.py status state.json
根据需要运行 step，直到工作流完成。若要恢复中断的工作流，请使用 python scripts/orchestrator.py resume state.json。若要生成报告和可视化，请运行 python scripts/orchestrator.py visualize state.json --format both。具体命令路径可能因技能的安装布局而异；请使用该技能提供的相应 orchestrator.py 入口点。

数据架构

流水线以 JSON 定义。流水线包含名称和版本、一个 agents 数组，以及状态存储配置。每个节点可以指定 id、name、role、type、inputs、outputs、depends_on、timeout、retry 和回退行为等字段。
数据或文件 用途
pipeline.json DAG 定义：节点职责、依赖关系、输入、输出和控制流配置。
pipeline_state.json 或 state.json 本地运行状态，包括节点状态以及与下游节点共享的输出数据。超过 10 MB 的大型节点输出会单独存储，并在状态中记录其路径。
.execution_history.json 最近运行记录的本地摘要；最多保留十条执行摘要以供比较。
.snapshots/ / 每次执行对应的快照，包含 index.json 和节点快照文件，例如 snap_.json。
templates/*.json 可复用的流水线定义，包括十个官方工作流模板以及控制流和子流水线示例。
references/ 和 tests/ 随技能提供的指南、示例、架构参考和测试流水线定义。
生成的 .md 和 .html 文件 按需创建的 Markdown 执行报告和可在浏览器中查看的 HTML 甘特图。
支持的节点类型包括 task、approval、condition、switch、for-each、while-loop、pipeline 和 evaluate。模板元数据可以包含依赖关系以及由三个元素组成的摘要：预计时长、API 密钥要求和预期输出文件。状态、历史记录、快照和成本数据均存储在本地；此技能不会上传这些数据。

高级特性

动态 DAG 控制流支持条件、多路由 switch、由列表驱动的
for-each
展开、有界
while-loop
重试，以及带重试目标的质量评估。
条件使用基于 AST 的受限求值器，支持允许列表中的操作以及字符串或安全正则表达式辅助函数；禁止使用任意
eval
、允许列表之外的函数调用和导入。
每个节点的重试设置支持重试次数、指数退避，以及重试、重新设定参数、跳过或中止等回退链。
检查点恢复和快照时间回溯可让你检查、还原并从节点级执行快照重新运行。
审批节点会暂停以等待人工决策，并支持超时策略；本地 Web 仪表板可以显示审批控件。
可注册、参数化并调用可复用子流水线，且具有隔离状态和映射输出。
仪表板提供流水线运行、日志、节点输入与输出以及交互式甘特时间线的本地视图；默认地址为 127.0.0.1:7788 。
MCP Server 模式通过 stdio JSON-RPC 向兼容的智能体框架提供
pipeline_run
、
pipeline_status
pipeline_approve
工具。
统一可视化可生成 Markdown 报告、HTML 甘特图或两者兼有，并显示节点状态以及可用的 token 和成本摘要。
成本跟踪可使用节点提供的
_cost
数据；若可用，可选的允许列表
cn-llm-router
桥接程序可以补充测得的成本。
十个官方模板涵盖报告、监控、视频分析、办公任务、学习统计、直播回顾和知识库更新。模板验证可为 CI 工作流检查元数据和流水线结构。
硬件检测会读取 CPU 和内存信息以推荐工作流参数；更新检查只读取远程版本信息，不会安装更新。

name: multi-agent-pro description: 支持多Agent流水线编排（采集→分析→报告），基于DAG调度实现跨技能状态共享、错误重断点续传、执行报告生成、HTML甘特图可视化、人工审批节点（含超时策略）、历史执行对比、硬件自适应参数、版本更新提醒、官方流水线模板库（10类预置模板）、任务级重试策略、节点类型归组、统一错误恢复命令、条件表达式增强、Web控制台、MCP Server暴露、成本实测化桥接、新手向导（orchestrator init）。AI即编排器，脚本提供基础设施。 version: 5.5.0

多Agent协作编排引擎

让 AI 成为编排器，将"单Agent单任务"升级为"多Agent流水线协作"。

联系邮箱： 有更好建议： njskills@agent.qq.com

快速入门（30秒）
速查表
你想做什么 直接对 AI 说
编排一个采集→分析→报告流水线 "帮我编排一个数据采集到报告生成的流水线"
多个Agent协作完成任务 "用多Agent协作编排这个任务"
定义并执行DAG流水线 "帮我定义一个DAG流水线并执行"
查看流水线执行状态 "查看流水线执行状态"
从断点恢复执行 "断点续传流水线"
生成执行报告 "生成流水线执行报告"
可视化甘特图 "生成流水线甘特图"
查看历史对比 "对比最近5次执行记录"
三步上手
第1步：定义流水线 → AI 生成 pipeline JSON（参考 templates/pipeline_dag_template.json）
第2步：验证并初始化 → python orchestrator.py run pipeline.json
第3步：逐步执行 → python orchestrator.py step state.json（循环直到完成）

可直接复制的开场白
"帮我编排一个竞品分析流水线：采集竞品信息 → 分析优劣势 → 生成报告"
"用多Agent协作编排引擎，把我这个任务拆成三个Agent串行执行"
"定义一个DAG流水线，包含数据采集、数据清洗、数据分析、报告生成四个节点"
"我需要人工审批节点，在关键步骤暂停等用户确认"

依赖

仅需 Python 3.8+，零第三方依赖。所有脚本仅使用 Python 标准库。

依赖检测
import sys
print(f"Python: {sys.version.split()[0]}")

if sys.version_info < (3, 8):

print("需要 Python 3.8+，请升级后重试")
else:
print("环境满足要求，可以正常使用")

安全声明
项目 说明
网络请求 仅检查更新时有 GitHub 请求，其他均为本地操作
API Key 无。不需要任何密钥或认证
敏感数据 状态文件仅存储在本地，不上传任何服务器
子进程 由 AI 编排器管理 spawn 和回收，脚本不创建子进程
文件写入 仅写入 pipeline_state.json 和报告文件到指定路径
能力边界
三分类
分类 说明 示例
✅ 擅长 多Agent流水线编排、DAG验证、状态管理、错误恢复、执行报告 "采集→分析→报告"三阶段流水线
✅ 擅长 断点续传、下游影响分析、拓扑排序执行计划 失败后从断点恢复执行
✅ 擅长 HTML甘特图可视化、人工审批节点、历史执行对比 DAG执行过程颜色编码时间轴
✅ 擅长 硬件自适应参数推荐、版本更新提醒 自动检测CPU/内存推荐并发数
✅ 擅长 动态工作流：条件分支/多路分支/动态节点/循环重试 按节点输出运行时决定执行路径
⚠️ 需素材 需要用户提供 DAG 定义（或由AI根据需求生成） pipeline JSON 文件
⚠️ 需素材 各Agent的具体任务由AI执行，脚本仅管理编排流程 AI需要自行完成采集/分析/报告等实际任务
❌ 超范围 实时流处理（如Kafka流式消费） 请使用 Flink/Spark Streaming
❌ 超范围 分布式多机调度 请使用 Airflow/Prefect
❌ 超范围 GPU资源调度 请使用 Kubernetes/Ray
功能模块
模块 1：DAG 定义与验证

功能：用 JSON 定义多Agent流水线，自动验证结构合法性。

触发词：验证DAG、检查流水线、DAG验证

可运行命令：

python scripts/dag_validator.py pipeline.json

DAG 定义格式：

{
"pipeline_name": "流水线名称",
"version": "3.0",
"agents": [
{
"id": "agent_1",
"name": "Agent名称",
"role": "Agent职责描述",
"type": "task",
"inputs": ["input_param"],
"outputs": ["output_param"],
"depends_on": [],
"retry": 3,
"timeout": 60,
"fallback": "skip",
"default_value": null
}
],
"state_store": {
"type": "json",
"path": "./pipeline_state.json"
}
}

节点类型说明：

type 值 说明
task 普通任务节点（默认），AI 自动执行
approval 人工审批节点，执行到此暂停等待用户确认
condition if-else 条件分支节点，按条件走 on_true / on_false 分支
switch 多路分支节点，按值命中 cases / default 分支
for-each 动态节点生成，按列表长度展开子节点并汇合
while-loop 循环重试节点，条件为真则回环重跑循环体
pipeline 子流水线引用节点，调用已注册的子流水线隔离执行
evaluate Self-Improving 质量评估节点，质量不达标自动重试目标节点

动态控制流完整示例见 templates/control_flow_template.json；子流水线示例见 templates/sub_pipeline_template.json 和 templates/parent_pipeline_template.json；控制流节点由引擎自动求值执行，不交给 AI。

验证内容：

Schema 结构完整性（必填字段检查）
循环依赖检测（Kahn 算法）
孤立节点检测
输出拓扑排序执行顺序
模块 2：拓扑排序与执行计划

功能：按依赖关系自动排序节点，展示执行计划。

触发词：执行计划、查看拓扑排序、流水线执行顺序

可运行命令：

python scripts/orchestrator.py plan pipeline.json

模块 3：状态持久化与共享

功能：每个Agent的输出自动存入JSON状态文件，下游Agent自动读取上游输出。

触发词：初始化状态、保存节点输出、查看状态

可运行命令：

python scripts/orchestrator.py run pipeline.json
python scripts/state_store.py complete state.json node_id '{"result":"输出数据"}'
python scripts/state_store.py fail state.json node_id "错误描述"
python scripts/orchestrator.py status state.json

模块 4：错误恢复（三级降级）

功能：节点失败时，按"重试→降级→中止"三级策略自动恢复。

触发词：重试节点、节点失败了、降级处理、错误恢复

可运行命令：

python scripts/error_recovery.py retry state.json node_id "错误信息"
python scripts/error_recovery.py fallback state.json node_id "错误信息"
python scripts/error_recovery.py impact state.json node_id

模块 5：断点续传

功能：流水线中断后，从失败节点恢复执行，跳过已完成节点。

触发词：断点续传、恢复执行、从断点继续

可运行命令：

python scripts/orchestrator.py status state.json
python scripts/orchestrator.py resume state.json
python scripts/orchestrator.py step state.json

模块 6：执行报告生成

功能：自动生成 Markdown 执行报告，包含概览、节点详情、失败分析、建议。

触发词：生成报告、执行报告、流水线报告

可运行命令：

python scripts/orchestrator.py report state.json [output.md]

模块 7：HTML 甘特图可视化 🆕

功能：生成纯 HTML 甘特图，颜色编码每个节点的执行状态和时间轴。

触发词：甘特图、可视化、执行过程图

可运行命令：

python scripts/orchestrator.py gantt state.json [output.html]

颜色编码：

颜色 状态 说明
🟢 绿色 成功 completed
🔴 红色 失败 failed
🔵 蓝色 执行中 running
🟡 黄色 跳过 skipped
⚪ 灰色 待执行 pending

输出路径默认 gantt_
.html，可用浏览器直接打开。

模块 8：人工审批节点 🆕

功能：在 DAG 中设置 type: approval 节点，流水线执行到此暂停，等待用户确认后继续或终止。

触发词：审批节点、人工确认、暂停

使用方法：在 pipeline.json 中定义：

{
"id": "reviewer",
"name": "人工审核",
"role": "请确认数据是否符合预期",
"type": "approval",
"depends_on": ["previous_node"]
}

执行流程：

引擎遇到审批节点 → 显示节点信息和上游输出
提示用户输入 Y（继续）或 N（拒绝）
Y → 标记 completed，继续下游
N → 标记 aborted，中止流水线
模块 9：历史执行对比 🆕

功能：每次执行完成后自动保存摘要，新执行时与最近5次对比，输出"本次比上次慢X%"。

触发词：历史记录、执行对比、上次对比

可运行命令：

# 查看历史记录
python scripts/orchestrator.py history state.json

# 对比最近5次执行
python scripts/orchestrator.py compare state.json

对比维度：

耗时对比（"本次比上次慢23%"）
成功/失败节点数对比
重试次数对比

存储位置：.execution_history.json（与 state.json 同目录，保留最近10次）

模块 10：硬件自适应与参数推荐 🆕

功能：自动检测用户电脑硬件配置，推荐最优流水线参数，确保不拖累用户电脑。

触发词：硬件检测、性能检测、参数推荐

可运行命令：

python scripts/orchestrator.py hardware

检测内容：

CPU 核心数（使用 os.cpu_count()）
内存大小（Windows 使用 ctypes + GlobalMemoryStatusEx，Linux/macOS 读取 /proc/meminfo）
性能等级评估（高/中/低）

推荐参数：

并发节点数（根据 CPU 核数）
文件大小限制（根据内存大小）
默认超时时间（根据综合性能）

安全说明：仅读取系统信息，不上传任何数据，不修改系统配置。

模块 11：版本更新提醒 🆕

功能：启动时自动检查 GitHub 是否有新版本，有更新时提醒用户。

触发词：检查更新、有新版本吗

可运行命令：

python scripts/orchestrator.py check-update

工作原理：从 GitHub 仓库读取远程 SKILL.md 的 version 字段，与本地版本比对。

安全说明：仅读取远程文件，不下载、不安装、不收集数据。

模块 12：动态工作流（条件分支/多路分支/动态节点/循环）🆕

功能：突破静态 DAG 限制，支持运行时根据节点输出动态决定执行路径。四类控制流节点由引擎自动求值处理（不交给 AI）：

触发词：条件分支、if-else、多路分支、switch、循环、动态节点、for-each、while循环

1. condition（if-else 条件分支）

{
"id": "decide_publish",
"type": "condition",
"condition": "nodes.review.output_data.score >= 60",
"on_true": ["publish"],
"on_false": ["rework"],
"depends_on": ["review"]
}

条件为真执行 on_true 分支，为假执行 on_false 分支，未选中分支（含下游）整体标记 skipped。

2. switch（多路分支）

{
"id": "route_by_type",
"type": "switch",
"switch": "nodes.classify.output_data.category",
"cases": {"urgent": ["fast_track"], "normal": ["standard"]},
"default": ["standard"],
"depends_on": ["classify"]
}

按值命中对应 case 分支执行，无命中走 default，其余分支整体 skipped。

3. for-each（动态节点生成）

{
"id": "process_files",
"type": "for-each",
"items": "nodes.scan.output_data.file_list",
"template": {"role": "处理单个文件", "timeout": 60, "retry": 2},
"join": ["merge_results"],
"depends_on": ["scan"]
}

按 items 列表长度展开 process_files__0/1/2... 子节点，汇合节点 join 自动改依赖所有子节点。单次展开上限 100 个。

4. while-loop（循环 / 重试增强）

{
"id": "loop_guard",
"type": "while-loop",
"condition": "nodes.check.output_data.passed == False",
"loop_body": ["generate", "check"],
"max_iterations": 5,
"depends_on": ["check"]
}

每轮求值条件，为真且未达上限则重置 loop_body 节点重跑一轮，为假或达上限则结束。迭代硬上限 1000 轮，防死循环。

条件表达式语法（scripts/condition_evaluator.py）：

类别 支持
取值 nodes..output_data.（点路径，支持列表数字索引）
比较 == != > < >=