技能编排器（Skill Orchestrator）
把互相独立的技能，用 文件作为中间语言 串成可控的流水线。不 import 技能代码，用 subprocess 进程隔离调用，故障互不污染。
本技能自带一条已验证的示范链路 scripts/orchestrate_scan2report.py：把 scan-to-word（OCR 扫描件→可编辑 docx + Markdown 报告）的输出，自动接力给 ai-transformation-consultant 模块 C（md2docx，Markdown→精美 Word 报告）。它本身就是"通用编排范式"的范例，可照葫芦画瓢套到任意两个技能对。
何时用
- 已有 2+ 个技能，各自"读文件→写文件"，你想让它们自动接力。
- 需要顺序控制 / 失败降级 / 运行可追溯（manifest）。
调用示范链路（scan-to-word → md2docx）
Stage B 需 python-docx，务必用已装依赖的 venv 解释器运行编排脚本：
PY="C:/Users/13301/.workbuddy/binaries/python/envs/default/Scripts/python.exe"
ORCH="C:/Users/13301/.workbuddy/skills/skill-orchestrator/scripts/orchestrate_scan2report.py"
# 1) 完整链路（需先装 Stage A 的 OCR 依赖；首次联网下载 OCR 模型）
"$PY" "$ORCH" --input 合同.png --mode high --out ./out
# 2) 仅 Stage B（Stage A 已在别处/分批跑完，报告已落盘到 --out）
"$PY" "$ORCH" --join --out ./out
# 3) 自测：样例报告 → Stage B → 断言产物非空 → 清理（只需 python-docx，无需 OCR 依赖）
"$PY" "$ORCH" --self-test
参数
| 参数 | 说明 |
|---|---|
--input <路径> |
扫描件图片/PDF（完整链路必填） |
--mode |
standard / high / batch（默认 standard） |
--out <目录> |
输出目录（默认 ./out） |
--join |
仅跑 Stage B，假定 *_OCR报告_*.md 已落盘在 --out |
--self-test |
自测编排逻辑（生成样例→Stage B→断言→清理） |
路径（两个子技能均可用环境变量覆盖）
| 环境变量 | 默认指向 | 用途 |
|---|---|---|
SCAN_TO_WORD_SCRIPT |
工作目录下 scan-to-word-skill/scripts/scan_to_word.py |
Stage A 脚本 |
MD2DOCX_SCRIPT |
用户级 ai-transformation-consultant/scripts/md2docx.py |
Stage B 脚本 |
> 跨机器 / 跨工作目录时，用环境变量把两个子脚本指到你自己的路径即可，无需改代码。
子技能依赖（由运行编排器的解释器负责）
- Stage A：
rapidocr-onnxruntime,opencv-python-headless,pillow,python-docx,numpy,pymupdf - Stage B：
python-docx
四原则（务必遵守）
- 顺序 > 并行：下游强依赖上游产物时必须串行。
- 可观测性优先：每阶段打印
[时间戳][级别] 状态；结束写orchestration_manifest.json（run_at / stages.status / outputs）。 - 失败降级：上游失败则整体停；仅下游失败则上游产物保留，不连带损坏。
- 文件即中间语言：技能间通过约定的中间文件（如
*_OCR报告_*.md）解耦，互不感知实现。
通用编排骨架（套到其他技能对）
import subprocess, glob, os, json, datetime, sys
PY = sys.executable
SKILL_A = os.environ.get("SKILL_A", "a.py")
SKILL_B = os.environ.get("SKILL_B", "b.py")
def stage_a(inp, out):
rc = subprocess.run([PY, SKILL_A, "--input", inp, "--out", out]).returncode
if rc: raise RuntimeError(f"A 退出码 {rc}")
out_files = sorted(glob.glob(os.path.join(out, "*.md")))
return out_files[-1] if out_files else (_ for _ in ()).throw(RuntimeError("A 无产物"))
def stage_b(md):
rc = subprocess.run([PY, SKILL_B, md]).returncode
if rc: raise RuntimeError(f"B 退出码 {rc}")
docx = md[:-3] + ".docx"
if not os.path.exists(docx): raise RuntimeError("B 无产物")
return docx
# main: 捕获每阶段异常 → 写 manifest → 打印 status
- 完整链路：
stage_a→stage_b→ manifest --join：仅跑下游，假定上游产物已落盘（长作业/分布式跑上游时收尾装配）--self-test：生成样例中间文件 → 跑下游 → 断言产物非空 → 清理（无需上游重依赖即可验证编排逻辑）
实现坑（已踩并修复）
- Windows venv 路径：解释器在
envs\<名>\Scripts\python.exe，不是envs\<名>\python.exe。 - stdout 编码：日志含中文/emoji 时，给 stdout/stderr 加
reconfigure(encoding="utf-8", errors="replace")防 GBK 控制台炸。 - 子脚本路径不要相对
_HERE：技能被移到~/.workbuddy/skills/后，_HERE变了会找不到相邻的子技能目录，统一用环境变量 + 绝对路径默认更稳。
多技能 DAG 编排（链路口 > 3 或存在分支/汇合时）
当技能链路不止 A→B，而是有分支、汇合、或 4+ 节点时，用 scripts/orchestrate_dag.py（纯 stdlib，灵感来自 LangGraph 的「图 + 节点 + 边」，但用 subprocess 实现，不引入重依赖）。它把 orchestrate_scan2report.py 的两阶段线性编排泛化为任意 DAG。
配置（JSON）
{
"name": "my-dag",
"vars": { "MD2DOCX": "C:/.../md2docx.py" },
"nodes": {
"gen": { "cmd": ["python","-c","..."], "deps": [], "outputs": ["report.md"] },
"report": { "cmd": ["python","{MD2DOCX}","{OUT}/report.md"], "deps": ["gen"], "outputs": ["report.docx"] },
"summary": { "cmd": ["python","-c","..."], "deps": ["gen"], "outputs": ["summary.md"] },
"archive": { "cmd": ["python","-c","..."], "deps": ["report"], "outputs": ["archive/report.docx"] }
}
}
- 节点 = 一次 subprocess 调用；
deps= 依赖的节点 id（即边）。 - 变量：
{OUT}=共享工作目录（自动转正斜杠，避免 Windows\U转义坑），{INPUT}=顶层输入，vars=自定义（如子脚本路径）。 - 引擎：拓扑排序（Kahn）+ 并发无依赖节点（上例
report与summary并行）+ 失败降级。
运行
"$PY" "$ORCH/scripts/orchestrate_dag.py" --config "$ORCH/examples/dag_demo.json" --out ./out
"$PY" "$ORCH/scripts/orchestrate_dag.py" --self-test # 内置 DAG 自测（需 python-docx 的 venv）
- 失败策略：默认
skip_downstream（失败节点下游全 skipped，已完成产物保留）；--fail-fast立即中断。 - 结束输出
dag_manifest_<日期>.json（每节点 status / duration_s / outputs，可观测）。 --max-workers控制并发度（默认 4）。
实测结论（本环境，venv python）
- 示范 DAG 4 节点全 ok：
gen → (report ∥ summary) → archive，report.docx 37KB，archive 复制成功，并发生效。 - 失败用例：
report故意失败 →archive自动 skipped；summary（并行、不依赖 report）仍 ok。降级符合预期。
> 选型建议：2 节点线性用 orchestrate_scan2report.py；>3 节点或有分支/汇合用 orchestrate_dag.py。真要复杂状态机/条件分支再考虑引入 LangGraph。
衔接
- 配合
skill-inventory：先治理后编排——资产理清后再把技能串成流水线。 - 治理产出的 json 可直接作为编排器的技能目录清单输入。
