def execute(self, args):
        try:
            in_ = json.loads(args) or {}
        except json.JSONDecodeError as e:
            return "错误: 参数不是合法 JSON: " + str(e)
        description = in_.get("description", "")
        prompt = in_.get("prompt", "")
        if not prompt.strip():
            return "错误: prompt 不能为空——子 agent 看不到别的上下文，全靠这一份"
        return self.run(description, prompt)

    # run 是子 agent 真正干活的入口：一份自包含的 prompt 进，一条最终回复出。
    # execute（模型点名调 sub_agent）和这一章的 WorkflowTool（代码按计划调）
    # 都走这同一个入口——换的是谁来编排，没换执行机制。
    def run(self, description, prompt):
        child_reg = Registry(*self.tools)
        child_history = [
            {"role": "system", "content": compose_system_prompt(self.skills)},
            {"role": "user", "content": prompt},
        ]
        print(f"[子 agent {description!r} 开始，独立的一份 history，父对话它一个字都看不到]",
              file=sys.stderr)
        try:
            reply, tokens, complete = run_child_loop(self.base, self.api_key, self.model, child_reg, child_history)
        except Exception as e:
            return "错误: 子 agent 执行失败: " + str(e)
        tag = "" if complete else "[未完成：达到轮数上限，以下是部分结果]\n\n"
        print(f"[子 agent {description!r} 结束：内部消耗约 {tokens} tokens，"
              f"父对话只收到下面这条回复，约 {estimate_text(reply)} tokens]", file=sys.stderr)
        return tag + reply

然后是 workflow 层——一个常量作占位符，一个拼结果的小函数，一个工具类。 Python 的 json.loads 不像 Go 的 json.Unmarshal 那样会按结构体字段 类型自动校验，所以形状检查（是不是字符串数组、有没有空阶段）要自己写：

# ---- workflow 层：把编排从模型手里拿回代码里 ----

# PLAN_SHAPE_HINT 附在每条参数错误的后面。报错也是发给模型的 prompt：
# 只说"不合法"，模型会瞎变形重试；把期望的形状递到它眼前，下一次就写对。
PLAN_SHAPE_HINT = '计划的形状：{"stages": [["阶段1的子任务prompt", "..."], ["阶段2的子任务prompt，可写 {{results}}"]]}'

RESULTS_PLACEHOLDER = "{{results}}"


def format_results(results):
    parts = [f"【子任务 {i + 1} 的结果】\n{r}\n" for i, r in enumerate(results)]
    return "\n".join(parts).strip()


# WorkflowTool 复用 SubAgentTool 的 run 入口跑每一条 prompt：执行机制
# 和 sub_agent 完全同一套，这个工具新增的只有编排——octo 的 workflow
# 也是同一个做法，agent() 直接复用支撑 sub_agent 的那套派生机制，
# 没有另起炉灶。计划的形状刻意扁平：一个阶段就是一组 prompt 字符串，
# 没有包一层对象——这份 JSON 的作者是模型，schema 每多一层嵌套，它写
# 错的机会就多一分。
class WorkflowTool:
    def __init__(self, runner):
        self.runner = runner

    def definition(self):
        return {
            "name": "workflow",
            "description": "……跟 Go 版一字不差，从略……",
            "parameters": {
                "type": "object",
                "properties": {
                    "stages": {
                        "type": "array",
                        "description": "按顺序执行的阶段列表……",
                        "items": {"type": "array", "items": {"type": "string"}},
                    },
                },
                "required": ["stages"],
            },
        }

    # execute 逐阶段执行计划。阶段内的并发和上一章 dispatch_tool_calls 是
    # 同一个模式：容量 MAX_PARALLEL_SUB_AGENTS 的信号量当限流阀，结果按
    # 下标写回。区别只在谁决定"这一批一起跑"——上一章靠 can_fan_out 事后
    # 检查模型有没有把调用发在同一轮，这里阶段本身就是并发声明。
    def execute(self, args):
        try:
            plan = json.loads(args)
        except json.JSONDecodeError as e:
            return "错误: 参数不是合法 JSON: " + str(e) + "。" + PLAN_SHAPE_HINT
        stages = plan.get("stages") if isinstance(plan, dict) else None
        if not stages:
            return "错误: 计划里一个阶段都没有。" + PLAN_SHAPE_HINT
        prev = []
        for si, prompts in enumerate(stages):
            if not isinstance(prompts, list) or not prompts or not all(isinstance(p, str) for p in prompts):
                return f"错误: 阶段 {si + 1} 不是字符串数组，或一个子任务都没有。{PLAN_SHAPE_HINT}"
            print(f"[workflow 阶段 {si + 1}/{len(stages)}：{len(prompts)} 个子任务，"
                  f"并发上限 {MAX_PARALLEL_SUB_AGENTS}]", file=sys.stderr)
            results = [None] * len(prompts)
            sem = threading.Semaphore(MAX_PARALLEL_SUB_AGENTS)
            threads = []
            for i, p in enumerate(prompts):
                if prev:
                    p = p.replace(RESULTS_PLACEHOLDER, format_results(prev))
                sem.acquire()

                def worker(i=i, p=p):
                    try:
                        results[i] = self.runner.run(f"阶段{si + 1}-子任务{i + 1}", p)
                    finally:
                        sem.release()

                t = threading.Thread(target=worker)
                threads.append(t)
                t.start()
            for t in threads:
                t.join()
            prev = results
        if len(prev) == 1:
            return prev[0]
        return format_results(prev)

main() 里注册，加在 sub_agent 之后：

    # workflow 也排在 sub_agent 之后才加——子 agent 的工具集里同样没有
    # workflow 这个名字，一份计划里的子任务不能自己再展开一份计划。
    tools.append(WorkflowTool(sub_agent))

  async execute(args) {
    let inArgs;
    try {
      inArgs = JSON.parse(args) ?? {};
    } catch (err) {
      return "错误: 参数不是合法 JSON: " + err.message;
    }
    const description = inArgs.description ?? "";
    const prompt = inArgs.prompt ?? "";
    if (!prompt.trim()) {
      return "错误: prompt 不能为空——子 agent 看不到别的上下文，全靠这一份";
    }
    return this.run(description, prompt);
  }

  // run 是子 agent 真正干活的入口：一份自包含的 prompt 进，一条最终回复出。
  // execute（模型点名调 sub_agent）和这一章的 WorkflowTool（代码按计划调）
  // 都走这同一个入口——换的是谁来编排，没换执行机制。
  async run(description, prompt) {
    const childReg = new Registry(...this.tools);
    const childHistory = [
      { role: "system", content: composeSystemPrompt(this.skills) },
      { role: "user", content: prompt },
    ];
    console.error(`[子 agent "${description}" 开始，独立的一份 history，父对话它一个字都看不到]`);
    let reply, tokens, complete;
    try {
      [reply, tokens, complete] = await runChildLoop(this.base, this.apiKey, this.model, childReg, childHistory);
    } catch (err) {
      return "错误: 子 agent 执行失败: " + (err.message ?? err);
    }
    const tag = complete ? "" : "[未完成：达到轮数上限，以下是部分结果]\n\n";
    console.error(`[子 agent "${description}" 结束：内部消耗约 ${tokens} tokens，` +
      `父对话只收到下面这条回复，约 ${estimateText(reply)} tokens]`);
    return tag + reply;
  }
}

然后是 workflow 层。跟上一章的 dispatchToolCalls 一样，阶段内并发用 同一个 Semaphore，只是每次派发前先 await sem.acquire()：

// ---- workflow 层：把编排从模型手里拿回代码里 ----

const PLAN_SHAPE_HINT = '计划的形状：{"stages": [["阶段1的子任务prompt", "..."], ["阶段2的子任务prompt，可写 {{results}}"]]}';
const RESULTS_PLACEHOLDER = "{{results}}";

function formatResults(results) {
  return results.map((r, i) => `【子任务 ${i + 1} 的结果】\n${r}\n`).join("\n").trim();
}

// WorkflowTool 复用 SubAgentTool 的 run 入口跑每一条 prompt：执行机制
// 和 sub_agent 完全同一套，这个工具新增的只有编排——octo 的 workflow
// 也是同一个做法，agent() 直接复用支撑 sub_agent 的那套派生机制，
// 没有另起炉灶。
class WorkflowTool {
  constructor(runner) {
    this.runner = runner;
  }

  definition() {
    return {
      name: "workflow",
      description: "……跟 Go 版一字不差，从略……",
      parameters: {
        type: "object",
        properties: {
          stages: {
            type: "array",
            description: "按顺序执行的阶段列表……",
            items: { type: "array", items: { type: "string" } },
          },
        },
        required: ["stages"],
      },
    };
  }

  // execute 逐阶段执行计划。阶段内的并发和上一章 dispatchToolCalls 是同一个
  // 模式：容量 MAX_PARALLEL_SUB_AGENTS 的 Semaphore 当限流阀，结果按下标
  // 写回。区别只在谁决定"这一批一起跑"——上一章靠 canFanOut 事后检查模型
  // 有没有把调用发在同一轮，这里阶段本身就是并发声明。
  async execute(args) {
    let plan;
    try {
      plan = JSON.parse(args);
    } catch (err) {
      return "错误: 参数不是合法 JSON: " + err.message + "。" + PLAN_SHAPE_HINT;
    }
    const stages = plan?.stages;
    if (!Array.isArray(stages) || stages.length === 0) {
      return "错误: 计划里一个阶段都没有。" + PLAN_SHAPE_HINT;
    }
    let prev = [];
    for (let si = 0; si < stages.length; si++) {
      const prompts = stages[si];
      if (!Array.isArray(prompts) || prompts.length === 0 || !prompts.every((p) => typeof p === "string")) {
        return `错误: 阶段 ${si + 1} 不是字符串数组，或一个子任务都没有。${PLAN_SHAPE_HINT}`;
      }
      console.error(`[workflow 阶段 ${si + 1}/${stages.length}：${prompts.length} 个子任务，` +
        `并发上限 ${MAX_PARALLEL_SUB_AGENTS}]`);
      const results = new Array(prompts.length);
      const sem = new Semaphore(MAX_PARALLEL_SUB_AGENTS);
      const tasks = [];
      for (let i = 0; i < prompts.length; i++) {
        let p = prompts[i];
        if (prev.length > 0) p = p.replaceAll(RESULTS_PLACEHOLDER, formatResults(prev));
        await sem.acquire();
        const stageNo = si + 1, taskNo = i + 1;
        tasks.push(
          (async () => {
            try {
              results[i] = await this.runner.run(`阶段${stageNo}-子任务${taskNo}`, p);
            } finally {
              sem.release();
            }
          })()
        );
      }
      await Promise.all(tasks);
      prev = results;
    }
    if (prev.length === 1) return prev[0];
    return formatResults(prev);
  }
}

顶层脚本里注册，加在 subAgent 之后：

// workflow 也排在 subAgent 之后才加——子 agent 的工具集里同样没有
// workflow 这个名字，一份计划里的子任务不能自己再展开一份计划。
toolList.push(new WorkflowTool(subAgent));

运行不用编译，跳到下一步给任务就行。

运行不用编译，跳到下一步给任务就行。

python3 main.py "projA、projB、projC 三个目录下各有一份 README.md。分头并行调查这三个项目——每个项目派一个隔离的子任务，各自回答：这个项目有没有声明已废弃的接口？三个子任务全部完成之后，再汇总三份结果，给出最终结论：哪个项目需要迁移、迁移到什么。"

node main.mjs "projA、projB、projC 三个目录下各有一份 README.md。分头并行调查这三个项目——每个项目派一个隔离的子任务，各自回答：这个项目有没有声明已废弃的接口？三个子任务全部完成之后，再汇总三份结果，给出最终结论：哪个项目需要迁移、迁移到什么。"