技能编排师（skill-orchestrator）

定位：不是技能库，是技能调度器。输入一句自然语言任务，输出：①可审阅的编排计划（DAG）②执行日志 ③瓶颈复盘报告。

财务人最擅长的流程标准化思维，被固化成可复用的流程资产——同一个任务第二次做，直接跑 flow 不用重规划。

何时用 / 何时不用

用：任务需要 ≥3 个技能协作，且有明确的先后依赖、可并行段、或失败降级需求。
例：出一份财务分析报告、做一次尽调、生成一份竞品对比、走一遍招聘筛选流程。

不用：
- 单技能能完成的（直接调技能本身）
- 步骤数 <3 的（编排开销 > 收益）
- 步骤完全线性且无需降级的（一个 bash 脚本更便宜）

判据：任务能否用一句话讲清"先做 A 再做 B 最后做 C"？能 -> 不用编排。
讲不清、有分支、有并行、有降级 -> 用编排。

三段式工作流（核心纪律）

自然语言任务
     |
     v
+--------------+   只出 flow，绝不执行
|  1. Planner  |  plan_flow.py + references/planner-prompt.md
+--------------+
     |  [!] 草稿，禁止直接执行
     v
+--------------+   人审：依赖对不对？降级够不够？pause 够不够？
|  2. 人工审阅  |
+--------------+
     |
     v
+--------------+   静态校验 F1-F16 / M1-M6，不通过不执行
|  3. lint     |  manifest_lint.py
+--------------+
     |
     v
+--------------+   层调度 + 状态机 + 降级 + 暂停 + agent 预留字段落日志
|  4. Executor |  run_flow.py
+--------------+
     |
     v
+--------------+   CPM 关键路径 + 三重瓶颈定义 + "不要优化"清单
|  5. Tracer   |  trace_flow.py
+--------------+

规划与执行必须解耦（Anthropic《Building Effective Agents》原文："Decouple the planning from the execution"）。
Planner 只出计划，不碰执行；人审过才交给 Executor。这条纪律是本 skill 的根。

四个脚本

退出码约定（供 CI 判断）：
- manifest_lint.py：0 无 error / 1 有 error / 2 文件缺失或解析失败
- run_flow.py：0 成功 / 1 静态校验失败 / 2 解析失败 / 3 运行期失败
- trace_flow.py：0 正常 / 1 存在失败或降级

Workflow vs Agent：分水岭与三条判据

先用 workflow。Anthropic《Building Effective Agents》的结论：能用 workflow 解决的，不要上 agent。
Agent 是更贵、更难调试、失败模式更多的一档；workflow 是"把'这事该怎么做'写死"，可复现、可审计、可回归。

升级到 agent 的三条判据（满足任意一条才考虑）：

1. 步骤数不可预知 —— 任务无法在事前枚举完节点（如"研究这个问题直到搞明白"）。
   workflow 要求 flow 是静态 DAG；步骤数随运行时信息变化的，天然属于 agent。
2. 需要环境反馈驱动 —— 中间结果决定下一步走向，且走向空间大、无法用 condition 节点穷举。
   workflow 的 condition 只能二分支；超过 4 条分支路径且互不排斥时，改成 agent 循环。
3. 单步成功率低 —— 单步 <90% 且需要自主重试 + 换策略（不是简单的幂等重试）。
   workflow 靠 on_fail 静态降级；agent 能边跑边换招。

本 skill 已为 agent 升级铺好六个口子（详见 references/agent-upgrade-protocol.md）：

关键：workflow 模式下这些字段必须完整写入 flow 并落日志（agent_reserved），
升级 agent 时不用改 flow 文件，只把 mode 改成 agent + 换一个 executor 就能跑。
这个"预留接口"是本 skill 相比"临时写死 workflow"的核心增量。

数据契约速览

• manifest（技能注册表）：每个技能声明 consumes/emits/parallelizable/idempotent/on_fail/writes/irreversible/needs_review/agent_capable。没有 manifest 就没有编排。
• flow（编排计划）：DAG + 三种节点类型（skill / condition / loop）+ 降级 + pause + agent 预留字段。
• 校验规则 F1-F16（flow）+ M1-M6（manifest）：改动 schema 必须同步改 lint。

完整字段定义、示例、每条规则的失败处理见 references/flow-schema.md。

最容易踩的 6 个坑（实测记录）

1. fallback_to: X 不加引号 -> YAML 直接解析失败
   正确：on_fail: "fallback_to: detect-rule"；错误：on_fail: fallback_to: detect-rule（报 mapping values are not allowed here）。
   裸 node id 无需引号：on_fail: brief。

2. 降级节点不标 on_demand: true -> 降级形同虚设
   降级节点本身是有 needs 的正常 DAG 节点，层调度器会每次都执行它，失败时它已经跑过一次了。
   on_demand: true = 只由降级路径触发，层调度器跳过（lint 有 F15/F16 双保险）。

3. condition 后的合并点语义：只有当所有上游都被跳过时才跳过自己，部分跳过照常运行。
   这是条件分支后合并的正确行为，不是 bug。写错就变成"任一支路跳过则整条断"。

4. eval_expr 必须白名单求值：{{ node.field }} 先替换为字面量，再 AST 白名单校验，再 eval(__builtins__={})。
   拦截 __import__ / __subclasses__ / 列表推导 / open 四类注入。测试里有 t_expr_injection 覆盖。

5. CPM 反向计算：LF[n] = min(LS[d])，bwd(d) 返回的已经是 LS[d]，不能再减 dur[d]。
   减了会导致关键路径收缩到只剩最后一步、全线负 slack。见 trace_flow.py 的注释。

6. 中文分词陷阱：re.findall(r"[\u4e00-\u9fff]{2}") 是非重叠匹配，"财务分析报告"只得 份财/务分/析报，
   把 财务/报告/指标 全吃掉。必须滑窗（连续中文片段逐位取二元组）。见 plan_flow.py:_tokens。

三重瓶颈定义（复盘的核心输出）

trace_flow.py 的复盘报告按三个维度定位瓶颈，不是只有一个"最耗时"的结论：

TOC 非瓶颈不优化：slack > 0 的节点，即使把它耗时砍到 0，端到端一秒不变。
复盘报告必须输出"不要优化清单"，防止团队去优化非关键路径的节点——这是浪费预算。

关键路径长度铁律

关键路径 ≤ 10 步。依据 Compound Mistakes（AI Engineering line 5768）：

lint 规则 F8（max_depth 硬上限）+ F12（>10 步告警）双保险。
超过 10 步先拆 flow 再谈执行，不要幻想"跑起来再说"。

反模式清单

• [ ] Planner 直接执行：违反"规划与执行解耦"，是本 skill 的根纪律。
• [ ] flow 里写死具体技能实例：flow 只引用 manifest 里注册的 id，不写具体调用参数（那是 Executor 的事）。
• [ ] 降级节点只写 needs 不写 on_demand：见坑 2。
• [ ] parallel_group 当调度器用：并行调度完全由 needs 决定；parallel_group 只是语义声明 + lint 的 F7 组内互不依赖检查。
• [ ] 把 condition 当通用路由：超过 4 条分支路径且互不排斥 -> 该上 agent 了，见三条判据。
• [ ] 优化非瓶颈节点：见"不要优化清单"。
• [ ] 超过 10 步的 flow 硬跑：Compound Mistakes 会让端到端成功率崩掉。

快速上手（5 分钟）

cd ~/.workbuddy/skills/skill-orchestrator

# 1. 用示例 manifest + 示例 flow 跑通全链路（stub 模式，估算耗时）
# 需 Python 3.11+；把下面 PY 设为你的解释器（多数环境直接用 python / python3 即可）
PY=python
"$PY" scripts/manifest_lint.py flows/financial-analysis-report.yaml
"$PY" scripts/run_flow.py flows/financial-analysis-report.yaml
# 复盘：四种输入都行（run_id 见上一步打印；latest=最近一次；也可给完整 .jsonl 路径或目录）
"$PY" scripts/trace_flow.py latest

# 2. 用自己的任务生成草稿
"$PY" scripts/plan_flow.py \
  --task "出一份竞品对比报告，覆盖功能、价格、市场份额三个维度" \
  --manifest examples/manifest.example.yaml \
  --out flows/competitor-report.yaml
# 然后：人审 -> lint -> run -> trace

# 3. 跑回归测试（21 用例 / 48 断言）
"$PY" tests/run_all.py

依赖：Python 3.11+ / PyYAML 6.x。无其它第三方依赖。

目录结构

skill-orchestrator/
+-- SKILL.md                          <- 本文件（入口）
+-- USAGE.md                          <- 使用说明（输入/输出/注意事项，面向使用者）
+-- references/
|   +-- flow-schema.md                <- 数据契约 + F1-F16/M1-M6 校验规则
|   +-- planner-prompt.md             <- Planner 契约、System Prompt 模板、人审检查项
|   +-- agent-upgrade-protocol.md     <- workflow->agent 升级协议（六个预留字段）
|   +-- anthropic-workflows.md        <- 《Building Effective Agents》要点
|   +-- bottleneck-principles.md      <- TOC / 关键路径 / 三重瓶颈
|   +-- checklist-pause-points.md     <- 暂停点设计（来自《清单革命》）
|   +-- eval-and-metrics.md           <- 复盘指标与判据
+-- scripts/
|   +-- manifest_lint.py              <- lint（F1-F16 / M1-M6）
|   +-- plan_flow.py                  <- Planner 骨架（任务->flow 草稿）
|   +-- run_flow.py                   <- Executor（状态机 + 层调度 + 降级 + 暂停）
|   +-- trace_flow.py                 <- Tracer（CPM + 三重瓶颈 + 不要优化清单）
+-- examples/
|   +-- manifest.example.yaml         <- 10 个示例技能
|   +-- impl_minimal.py               <- --impl 最小契约演示（execute 签名 + inputs 装配）
|   +-- impl_realistic.py             <- --impl 真实计算 + 异常降级演示（DEMO_FAIL=detect）
|   +-- impl_adapter_template.py      <- 接真实技能的三桥接模板（import / subprocess / HTTP）
+-- flows/
|   +-- financial-analysis-report.yaml <- 12 节点 / 7 层 / 关键路径 7 步的 MVP flow
+-- tests/
|   +-- run_all.py                    <- 21 用例 / 48 断言回归测试
+-- runs/                             <- 执行日志（JSONL + summary.json）

设计取舍与局限

明确不做的事：
- [ ] 不做技能市场 / 技能安装（技能从哪来由用户自己决定，本 skill 只编排已注册技能）
- [ ] 不做 LLM 路由（Planner 的提示词在 planner-prompt.md，用户自己接入 LLM）
- [ ] 不做分布式执行（单机进程即可，跨机是另一个问题域）
- [ ] 不做技能内部步骤级编排（skill 字段里的 #步骤 只是标注，不解析）

已知局限：
- 内置 executor 默认是 stub（按 cost_tokens/100 秒模拟耗时 + token）。要跑真实技能，
用 --impl module:fn 注入你的实现（见下节）。接入真实技能后，报告顶部会标
耗时口径: 实测，按时钟墙钟算瓶颈；stub 模式标 估算，绝对值不可信只供相对排序。
两者会自动区分，不会再把"估算的 300s"误当成真实瓶颈。
- 表达式的白名单求值支持算术/比较/布尔/And/Or/Not，不支持函数调用、下标、切片
- CPM 的相对排序（谁是瓶颈）可信；stub 模式下绝对时间仅供趋势参考

接入真实技能（--impl，核心用法）

内置 executor 跑的是 stub。要真的"调度你的技能"，写一个 execute(node, ctx) -> dict 注入即可：

# 三个示例实现都在 examples/ 下，可直接试跑：
"$PY" scripts/run_flow.py flows/financial-analysis-report.yaml \
      --impl examples/impl_minimal.py:execute          # 最小契约演示
"$PY" scripts/run_flow.py flows/financial-analysis-report.yaml \
      --impl examples/impl_realistic.py:execute        # 真实计算 + 异常降级演示

契约（签名必须严格一致）：

def execute(node: dict, ctx: dict) -> dict:
    # node: 当前节点在 flow yaml 里的定义（id / skill / switch ...）
    # ctx:  { node_id, node, skill, flow, manifest,
    #         inputs:{上游节点id: 产出dict}, fixtures, overrides }
    # 返回值必须是 dict；key 建议对齐 manifest 里该技能声明的 emits
    # （condition 的 switch {{ node.field }} 依赖这些 key 存在）
    ...

• condition / loop 节点不归你管：Executor 自己算，impl 对这类节点返回 {} 即可。
• 该 raise 就 raise：真实技能抛异常 = 节点失败 -> Executor 自动走 重试 -> 降级 -> abort。
  别自己吞异常假装成功（降级链会误以为成功）。
• 桥接方式（看你真实技能的形态，详见 examples/impl_adapter_template.py）：
• import —— 本地 Python 模块（importlib.util.spec_from_file_location）
• subprocess —— CLI / 脚本（用 JSON 传参收 JSON，非零退出直接抛）
• HTTP —— 远程服务 / API（POST JSON、收 JSON）
• 超时语义：接了 --impl 后不再按 cost_tokens 估算值判超时（估算值对真实技能毫无意义），
  超时交给真实技能自己控制；stub 模式仍按估算值判。

已实测的两条真实路径（见测试 t_real_impl_precision + 手工验证）：
- 正常：报告 耗时口径: 实测，亚毫秒级真实耗时也能正确聚合（曾是 round(x,3) 把微秒压成 0 的 bug，已修）。
- 失败：impl_realistic.py 设 DEMO_FAIL=detect 时 detect 抛异常 -> 重试 -> 降级到 detect-rule，
报告标 降级，CPM 正确处理降级节点（合成边让 detect-rule 拿到合理 slack，而非整段耗时）。

相关文件

• references/flow-schema.md —— 完整字段定义与校验规则
• references/planner-prompt.md —— Planner 契约与人审检查项
• references/agent-upgrade-protocol.md —— agent 升级六字段协议
• references/anthropic-workflows.md —— Anthropic 原文要点摘录
• references/bottleneck-principles.md —— 瓶颈原则（TOC / CPM / 三重定义）
• references/eval-and-metrics.md —— 复盘指标

作者与联系方式

• 作者：七仔的AI工具箱
• 微信公众号：七仔的AI工具箱
• 抖音：七仔的AI工具箱
• CSDN 博客：七仔的AI工具箱
• 说明：方便技能使用沟通与分享，欢迎反馈与共建。