multi-agent-pro

支持多Agent流水线编排（采集→分析→报告），基于DAG调度实现跨技能状态共享、错误重试、断点续传、执行报告生成、HTML甘特图可视化、人工审批节点、历史执行对比、硬件自适应参数和版本更新提醒。v5.0新增子流水线引用（模块化复用）和执行回放/Time Travel（快照机制）。AI即编排器，脚本提供基础设施。

fyniujin@fyniujin

Install

openclaw skills install @fyniujin/multi-agent-orchestrator

多Agent协作编排引擎

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

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

快速入门（30秒）

速查表

你想做什么	直接对 AI 说
编排一个采集→分析→报告流水线	"帮我编排一个数据采集到报告生成的流水线"
多个Agent协作完成任务	"用多Agent协作编排这个任务"
定义并执行DAG流水线	"帮我定义一个DAG流水线并执行"
查看流水线执行状态	"查看流水线执行状态"
从断点恢复执行	"断点续传流水线"
生成执行报告	"生成流水线执行报告"
可视化甘特图	"生成流水线甘特图"
查看历史对比	"对比最近5次执行记录"

三步上手

text

第1步：定义流水线 → AI 生成 pipeline JSON（参考 templates/pipeline_dag_template.json）
第2步：验证并初始化 → python orchestrator.py run pipeline.json
第3步：逐步执行 → python orchestrator.py step state.json（循环直到完成）

可直接复制的开场白

text

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

依赖

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

依赖检测

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验证

可运行命令：

bash

python scripts/dag_validator.py pipeline.json

DAG 定义格式：

json

{
  "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`	子流水线引用节点，调用已注册的子流水线隔离执行

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

验证内容：

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

模块 2：拓扑排序与执行计划

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

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

可运行命令：

bash

python scripts/orchestrator.py plan pipeline.json

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

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

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

可运行命令：

bash

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：错误恢复（三级降级）

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

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

可运行命令：

bash

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：断点续传

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

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

可运行命令：

bash

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

模块 6：执行报告生成

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

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

可运行命令：

bash

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

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

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

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

可运行命令：

bash

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

颜色编码：

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

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

模块 8：人工审批节点 🆕

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

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

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

json

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

执行流程：

引擎遇到审批节点 → 显示节点信息和上游输出
提示用户输入 Y（继续）或 N（拒绝）
Y → 标记 completed，继续下游
N → 标记 aborted，中止流水线

模块 9：历史执行对比 🆕

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

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

可运行命令：

bash

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

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

对比维度：

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

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

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

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

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

可运行命令：

bash

python scripts/orchestrator.py hardware

检测内容：

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

推荐参数：

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

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

模块 11：版本更新提醒 🆕

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

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

可运行命令：

bash

python scripts/orchestrator.py check-update

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

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

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

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

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

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

json

{
  "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（多路分支）

json

{
  "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（动态节点生成）

json

{
  "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（循环 / 重试增强）

json

{
  "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.<id>.output_data.<field>`（点路径，支持列表数字索引）
比较	`==` `!=` `>` `<` `>=` `<=`
逻辑	`and` `or` `not`
成员	`in` `not in`
算术	`+` `-` `*` `/` `//` `%`

安全说明：表达式采用 AST 白名单求值，禁止 eval/函数调用/lambda/推导式，杜绝 __import__、os.system 等注入；表达式长度上限 1000 字符、嵌套上限 50 层，防 DoS。

验证与自测：

bash

# 校验含控制流节点的 DAG
python scripts/dag_validator.py templates/control_flow_template.json

# 单独测试条件表达式
python scripts/condition_evaluator.py "nodes.a.output_data.score >= 60" '{"nodes":{"a":{"output_data":{"score":85}}}}'

模块 13：子流水线嵌套 / 模块化复用 🆕

功能：突破单一流水线限制，支持将已有流水线作为"积木"组装到更大的流水线中，实现逻辑复用。

触发词：子流水线、模块化、流水线引用、pipeline 节点

核心概念：

子流水线注册：将可复用的流水线（如"发票处理""合同审查"）注册到注册表
pipeline 节点：在父流水线中放置 type: pipeline 节点，引用子流水线
参数传递：父→子传参，子→父映射输出
上下文隔离：子流水线独立执行，内部节点不污染父上下文

1) 注册子流水线

bash

python scripts/pipeline_registry.py register templates/sub_pipeline_template.json --name invoice-pipeline-v1 --version 1.0

2) 查看已注册的子流水线

bash

python scripts/pipeline_registry.py list

3) 在父流水线中引用子流水线

json

{
  "id": "invoice_step",
  "name": "发票处理（子流水线）",
  "type": "pipeline",
  "pipeline_ref": "invoice-pipeline-v1",
  "params": {
    "file_path": "nodes.submit.output_data.invoice_file"
  },
  "outputs": {
    "invoice_no": "nodes._terminal_.output_data.invoice_no",
    "amount": "nodes._terminal_.output_data.amount"
  },
  "depends_on": ["submit"]
}

字段说明：

pipeline_ref：子流水线注册名（必填）
params：传递给子流水线的参数（值支持表达式，在父上下文求值）
outputs：子流水线输出到父上下文的映射（键=父变量名，值=子上下文表达式）

执行流程：

引擎检测到 pipeline 节点 → 从注册表加载子流水线
创建隔离 state 文件（子流水线独立运行）
用 params 表达式求值 → 注入子流水线首节点
逐步执行子流水线所有节点
子流水线完成后，用 outputs 表达式映射回父节点 output_data
父节点标记 completed，继续下游

预期效果：从"每次都从零开始"变成"搭积木式组装"，效率大幅提升。

模块 14：执行回放 / Time Travel 🆕

功能：在复杂流水线调试时，回到任意历史节点重新执行而不影响下游，大幅提升调试效率。

触发词：快照、执行回放、Time Travel、历史恢复、执行对比

核心能力：

快照机制：每个节点执行完自动保存上下文快照（输入、输出、执行时间、日志）
增量存储：只保存变化的字段，避免存储膨胀
从快照恢复：选任意历史快照点，从该点重新执行（上游保持不变，下游用新结果覆盖）
执行对比：对比两次执行的结果差异，快速定位问题节点

快照管理命令：

bash

# 列出当前执行的所有快照
python scripts/orchestrator.py snapshot list state.json

# 查看节点快照详情
python scripts/orchestrator.py snapshot show state.json node_id

# 从快照恢复（下游节点重置为 pending）
python scripts/orchestrator.py snapshot restore state.json node_id

# 对比两个节点的快照输出差异
python scripts/orchestrator.py snapshot diff state.json node_id_1 node_id_2

存储结构：

text

.snapshots/<execution_id>/
├── index.json            # 快照索引
├── snap_<node_id>.json   # 每个节点的快照

版本标识：execution_id（一次流水线运行）+ snapshot_id（每个节点快照）

安全说明：快照存储在本地 .snapshots/ 目录，不上传任何服务器。

统一入口命令速查

bash

# 验证 DAG 结构
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

# 断点续传
python scripts/orchestrator.py resume state.json

# 生成 Markdown 执行报告
python scripts/orchestrator.py report state.json

# 生成 HTML 甘特图
python scripts/orchestrator.py gantt state.json

# 查看执行历史
python scripts/orchestrator.py history state.json

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

# 硬件检测与参数推荐
python scripts/orchestrator.py hardware

# 检查更新
python scripts/orchestrator.py check-update

# 快照管理（执行回放 / Time Travel）
python scripts/orchestrator.py snapshot list state.json
python scripts/orchestrator.py snapshot show state.json node_id
python scripts/orchestrator.py snapshot restore state.json node_id
python scripts/orchestrator.py snapshot diff state.json node_id_1 node_id_2

# 子流水线注册表
python scripts/pipeline_registry.py register <pipeline.json> [--name 名称] [--version 1.0]
python scripts/pipeline_registry.py list
python scripts/pipeline_registry.py show <name>
python scripts/pipeline_registry.py validate

# 下游影响分析
python scripts/orchestrator.py impact state.json node_id

反模式（请避免）

反模式	问题	正确做法
循环依赖	A→B→C→A 死锁	打破循环，改为 DAG
巨型节点	所有逻辑放一个节点	按职责拆分为多个节点
零容错	retry=0 + fallback=abort	前端节点高重试+skip，末端节点低重试+abort

详见 references/anti-patterns.md。

安全风险项

风险项	等级	说明	缓解措施
子进程泄漏	P1	AI 编排器 spawn 的子 Agent 未正确回收	AI 需确保每个子任务完成后回收资源；脚本不创建子进程
状态文件并发写入	P2	多个编排实例同时写同一个 state.json	每个流水线使用独立的 state 文件路径
DAG 循环依赖	P2	用户定义了循环依赖的 DAG	`dag_validator.py` 启动前强制检测，发现环即中止
状态文件过大	P2	节点输出过多导致 state.json 膨胀	单节点输出超过 10MB 时写入独立文件，state 中只存路径
敏感数据泄露	P1	Agent 输出中可能包含敏感信息	FAQ 说明脱敏方法；建议不要处理真实密钥

FAQ

Q: 脚本支持哪些操作系统？

A: Windows、macOS、Linux 全部支持。仅使用 Python 标准库，无平台依赖。

Q: 需要安装第三方库吗？

A: 不需要。所有脚本仅使用 Python 标准库（json/os/sys/time/datetime/platform/ctypes/urllib），开箱即用。

Q: AI 在编排中扮演什么角色？

A: AI 是编排器。脚本提供基础设施（验证/状态/报告），AI 负责理解用户需求→生成DAG→逐步执行各节点任务→管理错误恢复。详见 references/examples.md。

Q: 如何处理节点间数据传递？

A: 节点完成时将输出保存到 pipeline_state.json，下游节点执行时引擎自动注入上游输出。详见 references/state-sharing-protocol.md。

Q: 节点失败后怎么恢复？

A: 三级策略：重试（自动）→ 降级（skip/default/abort）→ 断点续传。详见 references/error-recovery-patterns.md。

Q: 如何查看完整的端到端示例？

A: 参见 references/examples.md，包含竞品分析、数据处理（含错误恢复）、断点续传三个完整案例。

Q: 甘特图怎么打开？

A: 运行 python orchestrator.py gantt state.json，生成的 .html 文件可直接用浏览器打开。

Q: 如何设置人工审批节点？

A: 在 pipeline.json 中定义节点时添加 "type": "approval"，执行到此节点时会暂停等待用户确认。

Q: 历史记录占用多少存储？

A: 默认保留最近10次执行摘要，每次摘要约200字节，总计不超过2KB。

Q: 硬件检测会不会修改我的系统配置？

A: 不会。硬件检测仅读取系统信息（CPU核数、内存大小），不做任何修改。

深度参考

文档	内容
`references/dag-scheduling-guide.md`	DAG 调度原理：拓扑排序、执行策略、环检测、孤立节点
`references/state-sharing-protocol.md`	状态共享协议：数据传递机制、状态转换、最佳实践
`references/error-recovery-patterns.md`	错误恢复模式库：5种常见场景及恢复决策树
`references/anti-patterns.md`	反模式说明：3类常见错误 + 改进对比
`references/faq-deep.md`	深度 FAQ：10题覆盖边缘场景
`references/examples.md`	完整使用示例：3个端到端案例
`templates/pipeline_dag_template.json`	DAG 定义模板（含人工审批节点示例）
`templates/control_flow_template.json`	动态控制流模板（condition/switch/for-each/while-loop 完整示例）
`templates/state_schema.json`	状态文件 JSON Schema 规范 v5.0
`tests/test_pipeline.json`	测试用 DAG（含审批节点，用于自检）
`tests/test_control_flow.json`	控制流测试 DAG（四类控制节点端到端验证）

禁止文件类型

本 Skill 不支持以下文件类型的处理或生成：

类别	禁止的文件类型
Windows 可执行/批处理脚本	`.bat`、`.cmd`、`.ps1`、`.vbs`、`.exe`、`.dll`、`.lnk`、`.msi`
Office 二进制文档	`.docx`、`.xlsx`、`.pptx`、`.doc`、`.xls`、`.ppt`、`.xlsm`、`.docm`、`.pptm`
二进制镜像/安装包	`.iso`、`.dmg`、`.zip`、`.rar`、`.7z`、`.tar`、`.gz`、`.apk`、`.jar`
系统缓存/隐藏文件	`.DS_Store`、`.git`目录、`.env`、`.log`、`.tmp`
其他风险脚本	`.sh`、`.com`、`.scr`、`.hta`、`.reg`

更新日志

版本	日期	更新内容
v5.0.0	2026-07-13	增加：pipeline 子流水线节点，支持将已注册流水线作为模块嵌套到父流水线并实现上下文隔离；增加：pipeline_registry 子流水线注册表，支持注册/列出/查看/校验子流水线；增加：snapshot_store 快照存储机制，首个快照存完整数据后续只存增量 diff；增加：snapshot 子命令（list/show/restore/diff），支持从任意历史节点恢复下游重跑的 Time Travel 执行回放；增加：sub_pipeline_template 与 parent_pipeline_template 示例模板展示子流水线注册与引用；优化：flow_controller 扩展 pipeline 类型调度，自动加载子流水线并注入参数映射输出；优化：state_store 节点完成自动保存快照，持久化 pipeline_ref/params/outputs 控制流字段；优化：dag_validator 校验 pipeline 节点字段与 pipeline_ref 引用完整性
v4.0.0	2026-07-13	增加：condition 条件分支节点，按表达式走 on_true/on_false 分支；增加：switch 多路分支节点，按值命中 cases/default；增加：for-each 动态节点生成，按列表长度展开子节点并自动重挂汇合依赖；增加：while-loop 循环重试节点，条件为真回环重跑循环体并设迭代上限防死循环；增加：condition_evaluator 表达式求值器，采用 AST 白名单禁止函数调用防注入；增加：flow_controller 控制流引擎自动求值路由；增加：control_flow_template 与 test_control_flow 示例及测试；优化：dag_validator 校验控制流节点字段与分支引用；优化：控制流求值失败标记终态失败避免调度器反复重取
v3.0.0	2026-07-13	增加：HTML甘特图可视化模块，支持颜色编码的DAG执行时间轴；增加：人工审批节点（type: approval），可在关键步骤暂停等用户确认；增加：历史执行对比功能，自动保存最近10次执行摘要并对比耗时/成功率/重试次数；增加：硬件自适应模块，自动检测CPU/内存并推荐最优流水线参数；增加：版本更新提醒功能，对比GitHub远程版本；增加：禁止文件类型清单；优化：all_done判定逻辑，支持skipped节点视为完成；优化：calc_duration提取为独立函数供历史模块复用
v2.0.1	2026-07-08	修复：状态文件并发写入potential问题；优化：atomic write安全性
v2.0.0	2026-07-01	增加：三级错误恢复策略；增加：断点续传功能；增加：执行报告生成；优化：拓扑排序算法