Multi Agent Orchestrator

多Agent协作编排引擎

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

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

---

快速入门（30秒）

速查表

三步上手

第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("环境满足要求，可以正常使用")

安全声明

---

能力边界

三分类

---

功能模块

---

模块 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"
  }
}

节点类型说明：

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

验证内容：

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

---

模块 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]

颜色编码：

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

---

模块 8：人工审批节点 🆕

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

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

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

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

执行流程：

1. 引擎遇到审批节点 → 显示节点信息和上游输出
2. 提示用户输入 Y（继续）或 N（拒绝）
3. Y → 标记 completed，继续下游
4. 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）：

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

验证与自测：

# 校验含控制流节点的 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) 注册子流水线

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

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

python scripts/pipeline_registry.py list

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

{
  "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：子流水线输出到父上下文的映射（键=父变量名，值=子上下文表达式）

执行流程：

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

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

---

模块 14：统一可视化命令 🆕

功能：用单一 visualize 命令替代原来的 report + gantt 命令，支持 MD/HTML/三种输出模式。

触发词：可视化、生成报告、甘特图、`执行结果'

可运行命令：

# 同时生成 MD 报告和 HTML 甘特图（默认 both）
python scripts/orchestrator.py visualize state.json

# 只生成 MD 报告
python scripts/orchestrator.py visualize state.json --format md

# 只生成 HTML 甘特图
python scripts/orchestrator.py visualize state.json --format html

# 指定输出路径
python scripts/orchestrator.py visualize state.json --format both ./my_report

输出模式：

成本概览（当节点包含 cost_data 时自动展示）：

• 总 Token 消耗
• 总费用（¥）
• 节点成本分布表

向后兼容：原 report 和 gantt 命令仍可用（自动映射到 visualize 的 md/html 模式）。

---

模块 15：Self-Improving 循环（质量评估 + 自动重试）🆕

功能：在流水线中嵌入质量评估节点，当输出质量不达标时自动重跑上游节点，实现"自我改进"。

触发词：质量评估、自动重试、Self-Improving、evaluate 节点

节点配置示例：

{
  "id": "quality_check",
  "type": "evaluate",
  "quality_expr": "nodes.review.output_data.score >= 80",
  "retry_targets": ["draft", "review"],
  "max_eval_rounds": 3,
  "depends_on": ["review"]
}

字段说明：

执行流程：

1. evaluate 节点完成 → 对 quality_expr 求值
2. 质量达标 → 节点 completed，继续下游
3. 质量不达标且未达上限 → 重置 retry_targets 为 pending，评估节点自身回到 pending，等重跑完再次评估
4. 质量不达标且已达上限 → 接受当前结果，节点 completed（带警告）

安全说明：

• max_eval_rounds 硬上限 1000 轮（防死循环）
• 评估轮次计数存储在 eval_round 字段

---

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

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

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

核心能力：

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

快照管理命令：

# 列出当前执行的所有快照
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

存储结构：

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

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

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

---

模块 17：成本追踪 🆕

功能：自动追踪每个节点的 token 消耗和费用，在可视化报告中展示成本概览。

触发词：成本、token、费用、消耗

工作原理：

• 节点完成时，如果 output_data 包含 _cost 字段，自动提取到 cost_data
• _cost 字段格式：{"tokens": 1500, "cost_rmb": 0.003, "model": "gpt-4"}
• 报告生成时自动汇总总 token/费用/节点分布

成本数据格式：

{
  "tokens": 1500,
  "cost_rmb": 0.003,
  "model": "gpt-4"
}

报告展示：

• 总 Token 消耗
• 总费用（¥）
• 节点成本分布表（节点 id / Token / 费用 / 模型）

安全说明：成本数据仅存储在本地 state.json，不上传任何服务器。

---

统一入口命令速查

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

# 生成可视化（MD/HTML/both）
python scripts/orchestrator.py visualize state.json --format both

# 生成 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  [--name 名称] [--version 1.0]
python scripts/pipeline_registry.py list
python scripts/pipeline_registry.py show 
python scripts/pipeline_registry.py validate

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

---

反模式（请避免）

详见 references/anti-patterns.md。

---

安全风险项

---

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核数、内存大小），不做任何修改。

---

深度参考

---

禁止文件类型

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

---

---

模块 18：官方流水线模板库 🆕

功能：预置 4 类高频流水线模板，让引擎价值被直接看见。每个模板元数据声明建议安装的外部 skill，运行时探测依赖是否就绪，缺失则标灰提示安装链接。

触发词：预置模板、政采日报、视频分析、周报生成、巡检汇总

预置模板列表：

模板元数据示例：

{
  "name": "政采日报",
  "description": "...",
  "category": "data-processing",
  "dependencies": [
    {
      "skill_id": "tencent-docs",
      "install_link": "https://skillhub.cn/skill/tencent-docs",
      "description": "腾讯文档（用于日报内容协作）",
      "required": false
    }
  ],
  "pipeline": { ... }
}

可运行命令：

# 列出所有预置模板
python orchestrator.py template list

# 显示模板详情 + 依赖状态
python orchestrator.py template show gov_procurement_daily.json

# 检查依赖是否满足
python orchestrator.py template check gov_procurement_daily.json

# 渲染为可执行 pipeline.json
python orchestrator.py template render gov_procurement_daily.json my_pipeline.json

探测渲染规则：

• 依赖已安装 → ✅ 正常显示
• 依赖缺失 + required: true → ⚠️ 对应节点标灰 + 显示安装链接
• 依赖缺失 + required: false → ⚠️ 提示安装链接但不阻塞执行

安全说明：仅提示安装，不自动下载/安装/隐式依赖，保持单包合规。

---

模块 19：任务级重试策略 🆕

功能：节点 schema 新增 retry 块，支持任务级重试策略配置。未配置的节点保持旧全局策略，完全向后兼容。

触发词：节点重试、退避策略、降级链、retry 块

retry 块配置：

{
  "id": "node_a",
  "retry": {
    "count": 5,
    "backoff_base": 2,
    "fallback_chain": ["retry", "reparam", "skip"]
  }
}

降级动作：

• retry：重置为 pending，等待重新执行
• reparam：换参数重试（保留 intermediate 输出供参考）
• skip：跳过此节点，下游收到 output_data={"skipped": true}
• abort：中止流水线

向后兼容：未配置 retry 块的节点走旧全局策略（retry 字段 / fallback 字段）。

---

模块 20：节点类型归组（认知层） 🆕

功能：7 种节点类型在文档、帮助、校验错误中归为 4 类，降低认知负担。DAG Schema 完全不变，旧流水线零迁移。

实现原则：归组只动认知层（文档/提示/校验错误信息），不动 DAG Schema。

---

模块 21：统一错误恢复命令 🆕

功能：将 retry/fallback/impact 三个独立子命令合并为 recover 带子命令。旧命令向后兼容（自动映射到 recover）。

触发词：错误恢复、重试、降级、影响分析

可运行命令：

# 统一入口（推荐）
python orchestrator.py recover retry state.json node_a '网络超时'
python orchestrator.py recover fallback state.json node_a '网络超时'
python orchestrator.py recover impact state.json node_a

# 旧命令（保持可用，自动映射到 recover）
python orchestrator.py retry state.json node_a '网络超时'
python orchestrator.py fallback state.json node_a '网络超时'
python orchestrator.py impact state.json node_a

安全说明：旧命令可用但推荐迁移到新入口，减少命令面。

---

模块 22：条件表达式增强 🆕

功能：AST 白名单求值器新增正则匹配（re_safe 子集）与字符串操作函数，白名单机制不变，长度与嵌套上限沿用。

触发词：正则匹配、contains、startswith、split

新增字符串函数：

新增正则函数（re_safe 子集）：

安全限制：

• 禁止关键字参数（re.sub 等）
• 正则仅允许 match/search/findall，禁止 re.compile、re.sub 等
• 函数参数最多 5 个
• 表达式长度上限 1000 字符，嵌套上限 50 层

---

更新日志