LangGraph.js 中文入门

第 9 章 · 编排设计模式 ​

🎯 本章目标 ​

  • 分清「机制（mechanism）」和「模式（pattern）」的区别：第 3 章教你零件怎么用，本章教你零件该拼成什么形状
  • 掌握官方 workflows-agents 里最常用的 5 种编排模式：提示链 / 路由 / 并行化 / 编排者-工作者 / 评估者-优化者
  • 每种模式都配一个可运行 + 真实输出的示例，亲眼看到它解决什么问题、代价是什么
  • 用时间戳与调用次数亲手证明「并行是并行」「工作流可预测、智能体不可预测」
  • 学会「工作流 vs 智能体」的选型决策树——什么特征选工作流，什么特征才该上自主智能体

本章代码

📌 代码约定：正文的代码块都保持可直接运行（含 import 与模型初始化）。若某段为聚焦概念的节选，会明确标注「节选」并指向同名的 code/ 完整文件。所有示例的输出都是真实运行结果，不是编造的示例输出。

文件内容
code/01-prompt-chaining.ts提示链：大纲 → 初稿 → 风格转换，并对照「一次问 vs 分步问」
code/02-routing.ts路由：模型先分类，再分流到技术 / 闲聊 / 投诉三条分支
code/03-parallelization.ts并行化：分片（Sectioning）+ 投票（Voting），时间戳证明并行
code/04-orchestrator-worker.ts编排者-工作者：模型动态拆 N 个子任务，Send 并发分发再汇总
code/05-evaluator-optimizer.ts评估者-优化者：生成 → 评审给具体意见 → 重写，循环到达标
code/06-workflow-or-agent.ts工作流 vs 智能体：并排对比调用次数 / 耗时 / 可预测性 + 决策树

一、先分清：「机制」和「模式」是两回事 ​

这一节很短，但很重要——因为它决定了你读第 3 章和读本章时，脑子里该想什么。

第 3 章讲的是「机制」：条件边怎么用、Send 怎么用、并行怎么写、循环怎么防死循环。那是零件层面的知识——「这把螺丝刀怎么握」。

本章讲的是「模式」：零件都会用了，但该把它们拼成什么形状？同样会写 addConditionalEdges，有人拼出一个路由，有人拼出一个自我改进循环——这不是 API 的区别，是设计的区别。

机制（第 3 章）                     模式（本章）
─────────────                      ─────────────
addConditionalEdges  ──组合成──►   路由（Routing）
addEdge 一对多       ──组合成──►   并行化（Parallelization）
Send + 条件边        ──组合成──►   编排者-工作者（Orchestrator-Worker）
条件边回指上游       ──组合成──►   评估者-优化者（Evaluator-Optimizer）
一串固定 addEdge     ──组合成──►   提示链（Prompt Chaining）

一句话：

💡 机制是「怎么用零件」，模式是「把零件拼成什么形状」。 官方文档 workflows-agents 那一页，通篇讲的就是 5 种「形状」，以及最后一个核心论点：能用固定形状解决的，就别上自主智能体。

本章的 5 个示例，正好对应官方那 5 种模式；第 6 个示例收尾讲选型。示例之间会复用第 3 章学过的机制，但重点永远落在「形状」上——为什么这么拼、什么时候不该这么拼。

二、模式一：提示链（Prompt Chaining） ​

是什么：把一个大任务拆成一串固定顺序的小步骤，每一步的输出喂给下一步。有些步骤之间还可以放一个「闸门（gate）」做检查，不过关就中断或重来。

START → generate_outline → write_draft → style_transform → END
          （生成大纲）        （按大纲写初稿）   （风格转换）

什么时候用：任务能干净地拆成有序步骤，且每一步都值得单独调一次模型——比如「先列要点，再展开成文」，或「先翻译，再润色」。

为什么有用：模型一次要同时「谋篇 + 成文 + 控风格」时，容易顾此失彼。拆开后，每一步的提示词都只写「这一件事」，成品质量往往明显更高。代价是调用次数变多、耗时变长。

示例 1：一步到位 vs 分三步（同需求对照） ​

这个示例故意不直接讲 API，而是做一个对照实验：同一个需求，一次问模型 vs 拆成三步。两次输出并排打出来，让你亲眼看到「为什么要拆」。

typescript
/**
 * 示例 1：提示链（Prompt Chaining）
 *
 * 模式定义（来自官方 workflows-agents）：
 *   把一个大任务拆成「一串固定顺序的小步骤」，每一步的输出喂给下一步，
 *   中间可以在某些步骤加「闸门（gate）」做检查。
 *
 * 图结构（三步流水线）：
 *
 *   START → generate_outline → write_draft → style_transform → END
 *             （生成大纲）        （按大纲写初稿）   （风格转换）
 *
 * 本示例做两个对照：
 *   ① 一次问模型：把需求原样丢给模型，让它一口气写完。
 *   ② 分步问模型：把同一需求拆成大纲 → 初稿 → 风格转换三步。
 * 然后把两次的最终产出并排打出来，让读者亲眼看到「为什么要拆」。
 */
import { StateGraph, START, END, Annotation } from "@langchain/langgraph";
import { HumanMessage, SystemMessage } from "@langchain/core/messages";
import { createModel, banner } from "../../lib/model.js";
import { assertChatKey } from "../../lib/env.js";

// ── 状态：流水线上每一站的产出都存进「共享表格」 ────────────────
const ChainState = Annotation.Root({
  /** 原始需求 */
  requirement: Annotation<string>,
  /** 第 1 步：大纲 */
  outline: Annotation<string>,
  /** 第 2 步：初稿 */
  draft: Annotation<string>,
  /** 第 3 步：风格转换后的终稿 */
  final: Annotation<string>,
});

async function main() {
  banner("示例 1：提示链（一步到位 vs 分三步）");
  assertChatKey();

  const model = createModel({ temperature: 0.4 });

  // 本章统一的需求（故意有点「大」，让「一次问」容易顾此失彼）
  const requirement =
    "写一段 200 字左右的短文，向完全不懂技术的读者介绍：为什么团队应该用 LangGraph 来编排 AI 工作流。";

  // ══════════════════════════════════════════════════════════
  //  对照组 A：一次问模型（不拆分）
  // ══════════════════════════════════════════════════════════
  console.log("🅰️  对照组 A —— 一次问模型，让它一口气写完\n");
  const tA = Date.now();
  const oneShot = await model.invoke([new HumanMessage(requirement)]);
  const oneShotText = String(oneShot.content).trim();
  const costA = Date.now() - tA;
  console.log("── 一次问的产出 ──");
  console.log(oneShotText);

  // ══════════════════════════════════════════════════════════
  //  实验组 B：提示链（大纲 → 初稿 → 风格转换）
  // ══════════════════════════════════════════════════════════
  console.log("\n\n🅱️  实验组 B —— 拆成三步流水线\n");

  // ── 第 1 步：生成大纲 ─────────────────────────────────────
  async function generateOutline(state: typeof ChainState.State) {
    console.log("   [节点 generate_outline] 第 1 步：先生成大纲…");
    const res = await model.invoke([
      new SystemMessage(
        "你是内容策划。根据用户需求，产出一份简明的写作大纲，" +
          "用 3~4 条要点列出这篇短文应该覆盖什么。只输出大纲，不要写正文。"
      ),
      new HumanMessage(state.requirement),
    ]);
    const outline = String(res.content).trim();
    console.log("── 大纲 ──");
    console.log(outline);
    return { outline };
  }

  // ── 第 2 步：按大纲写初稿 ─────────────────────────────────
  async function writeDraft(state: typeof ChainState.State) {
    console.log("\n   [节点 write_draft] 第 2 步：严格按大纲写初稿…");
    const res = await model.invoke([
      new SystemMessage(
        "你是技术作者。严格依据下面这份大纲写一篇 200 字左右的短文，" +
          "覆盖大纲里的每一条要点，不要跑题、不要遗漏。"
      ),
      new HumanMessage(
        `原始需求：${state.requirement}\n\n写作大纲：\n${state.outline}`
      ),
    ]);
    const draft = String(res.content).trim();
    console.log("── 初稿 ──");
    console.log(draft);
    return { draft };
  }

  // ── 第 3 步：风格转换（初稿 → 面向外行、更口语的终稿） ─────
  async function styleTransform(state: typeof ChainState.State) {
    console.log("\n   [节点 style_transform] 第 3 步：做风格转换（更口语、更通俗）…");
    const res = await model.invoke([
      new SystemMessage(
        "你是科普编辑。把下面这段文字改写成「完全不懂技术的普通人」也能秒懂的口语化版本，" +
          "多用比喻、少用术语，保持原意和大致篇幅。只输出改写后的正文。"
      ),
      new HumanMessage(state.draft),
    ]);
    const final = String(res.content).trim();
    return { final };
  }

  const graph = new StateGraph(ChainState)
    .addNode("generate_outline", generateOutline)
    .addNode("write_draft", writeDraft)
    .addNode("style_transform", styleTransform)
    .addEdge(START, "generate_outline")
    .addEdge("generate_outline", "write_draft")
    .addEdge("write_draft", "style_transform")
    .addEdge("style_transform", END)
    .compile();

  console.log();
  const tB = Date.now();
  const result = await graph.invoke({ requirement });
  const costB = Date.now() - tB;

  console.log("\n── 分步的终稿 ──");
  console.log(result.final);

  // ══════════════════════════════════════════════════════════
  //  并排对比
  // ══════════════════════════════════════════════════════════
  console.log("\n" + "─".repeat(72));
  console.log("📊 两次对照");
  console.log("─".repeat(72));
  const rows: [string, number, number][] = [
    ["模型调用次数", 1, 3],
    ["整条流程耗时(ms)", costA, costB],
    ["终稿字数", oneShotText.length, String(result.final).length],
  ];
  console.log(
    ["指标".padEnd(18), "一次问".padEnd(12), "分步问"].join("")
  );
  for (const [name, a, b] of rows) {
    console.log([String(name).padEnd(18), String(a).padEnd(12), String(b)].join(""));
  }
  console.log(
    "\n💡 看两段终稿的差别：一次问往往「该说的都说了，但每一句都泛泛」；" +
      "分步问先立骨架（大纲）、再填肉（初稿）、最后打磨（风格转换），" +
      "每一步模型只需要专注做好一件事，成品的结构和可读性明显更好。"
  );
  console.log(
    "   代价也很清楚：调用 3 次、耗时更长。所以提示链只适合" +
      "「任务能干净地拆成有序步骤、且每步都值得单独调一次模型」的场景。"
  );
}

main().catch((e) => {
  console.error("❌ 调用失败：", e);
  process.exit(1);
});

运行：

bash
npx tsx 09-design-patterns/code/01-prompt-chaining.ts

真实输出：

========================================================================
  示例 1：提示链（一步到位 vs 分三步）
========================================================================

🅰️  对照组 A —— 一次问模型，让它一口气写完

── 一次问的产出 ──
把 AI 工作流想成做项目：查资料、写初稿、审校、定稿。只用一个 AI 聊天框，就像把所有活交给一位聪明但健忘的同事，步骤一多容易乱、漏、跑偏。LangGraph 像一张可执行的流程图：把每个 AI 或人工步骤画成节点，用箭头规定先后、分支和循环；出错可重试，关键处可让人确认，大家共享同一份进度。团队用它，不是为了更炫，而是让 AI 流程看得见、控得住、改得动、能复用。新人能看懂，负责人能追踪，复杂任务也能稳定跑完。所以，当 AI 要参与真实业务，LangGraph 值得作为团队的编排底座。


🅱️  实验组 B —— 拆成三步流水线


   [节点 generate_outline] 第 1 步：先生成大纲…
── 大纲 ──
写作大纲：

1. **生活化类比开场**：把 AI 工作流比作“多人接力办事”，说明单次提问不够，真实任务往往需要多步骤、分支和人工确认。

2. **通俗解释 LangGraph 的作用**：把它说成给 AI 团队画“流程图”和“交通指挥系统”，让多个 AI 步骤、工具和人按顺序协作，并能记住中间进展。

3. **对比不用它的痛点**：流程散落在提示词和脚本里，难改、难查、容易乱；用了之后，步骤更清晰，可重试、可审核、可调整。

4. **总结团队收益**：降低协作门槛，方便非技术成员理解，提升稳定性和可扩展性，让 AI 工作流从“一次性尝试”变成可复用、可迭代的团队资产。

   [节点 write_draft] 第 2 步：严格按大纲写初稿…
── 初稿 ──
把 AI 工作流想成“多人接力办事”：单次提问不够，真实任务常有多步骤、分支和人工确认。LangGraph 像给 AI 团队画“流程图”和“交通指挥系统”，让多个 AI 步骤、工具和人按顺序协作，并记住中间进展。不用它，流程散落在提示词和脚本里，难改、难查、容易乱；用了，步骤更清晰，可重试、可审核、可调整。对团队而言，这降低了协作门槛，方便非技术成员理解，也提升稳定性和可扩展性，让 AI 工作流从“一次性尝试”变成可复用、可迭代的团队资产。

   [节点 style_transform] 第 3 步：做风格转换（更口语、更通俗）…

── 分步的终稿 ──
别把 AI 想成一问一答就完事，它真要帮你干活，更像一群人接力办婚礼：有人列清单，有人买东西，有人干活，中间还可能分岔，需要你点头确认。LangGraph 就像给这支 AI 小队画了一张大流程图，还配了个交通指挥台：谁先上、谁后上、什么时候叫工具、什么时候请人拍板，都能排得明明白白，还能记住事情做到哪一步了。没有它，整个流程就像把步骤写在便签、聊天记录和脑子里，改起来费劲，查起来头疼，一乱就收不住。有了它，步骤更清楚，出错能重来，关键地方能检查，想调整也方便。对团队来说，大家更容易看懂和配合，不懂技术的人也能上手，整体更稳，以后想加更多步骤、更多人也更容易，让 AI 干活不再是一次性碰运气，而是能反复用、越改越好的团队家底。

────────────────────────────────────────────────────────────────────────
📊 两次对照
────────────────────────────────────────────────────────────────────────
指标                一次问         分步问
模型调用次数            1           3
整条流程耗时(ms)        4418        16917
终稿字数              248         316

💡 看两段终稿的差别：一次问往往「该说的都说了，但每一句都泛泛」；分步问先立骨架（大纲）、再填肉（初稿）、最后打磨（风格转换），每一步模型只需要专注做好一件事，成品的结构和可读性明显更好。
   代价也很清楚：调用 3 次、耗时更长。所以提示链只适合「任务能干净地拆成有序步骤、且每步都值得单独调一次模型」的场景。

💡 看两个数字：一次问 1 次调用、4.4 秒；分步问 3 次调用、16.9 秒。差距就摆在「耗时」那一列——这就是提示链的成本。而它的收益藏在两段终稿里：一次问的版本「该说的都说了，但每句都泛泛」；分步问的版本有明显结构（先类比、再痛点、再价值），而且更口语。用可量化的成本，换难量化但可感知的质量，值不值得，要你自己权衡。

⚠️ 一个常见的误区：不是拆得越多越好。每多一步就多一次调用、多一段延迟、多一个可能出错的环节。只有当某一步「单独交给模型，质量确实更高」时，拆才划算。纯拼接（比如「先格式化，再原样输出」）用代码做就行，别浪费一次模型调用。

三、模式二：路由（Routing） ​

是什么：先让模型对输入分类，再用条件边把请求送到不同的下游分支。每个分支有自己的系统提示词、自己的处理逻辑，互不干扰。

                     ┌──► tech      （技术问题）
START → classify ────┼──► chat      （闲聊寒暄）
                     └──► complaint （投诉抱怨）

什么时候用：输入类型多样、且不同类型的处理方式截然不同。比如客服系统里，技术问题要给排障步骤，闲聊要轻松回应，投诉要先安抚——把它们塞进一个提示词，只会互相干扰。

和第 3 章的联系：它的机制就是第 3 章的「条件边」。本章的差别在于——我们把它当作一个可复用的设计模板来看，并强调「分类节点」和「分支节点」的职责分离。

示例 2：模型分类 + 三路分流 ​

typescript
/**
 * 示例 2：路由（Routing）
 *
 * 模式定义（来自官方 workflows-agents）：
 *   先让模型给输入「分类」，再用条件边把请求送到不同的下游分支。
 *   每个分支可以有自己的专属提示词、专属处理逻辑，互不干扰。
 *
 * 图结构：
 *
 *                        ┌──► tech      （技术问题：走「技术支持」人格）
 *   START → classify ────┼──► chat      （闲聊寒暄：走「轻松陪聊」人格）
 *                        └──► complaint （投诉抱怨：走「安抚+登记」人格）
 *
 * 关键点：
 *   - classify 节点用真实模型做「三选一」分类，并把分类结果打印出来；
 *   - 三条分支各自绑定不同的 SystemMessage 与不同的收尾逻辑；
 *   - 用 addConditionalEdges 的第三个参数列全所有可能分支。
 */
import { StateGraph, START, END, Annotation } from "@langchain/langgraph";
import { HumanMessage, SystemMessage } from "@langchain/core/messages";
import { createModel, banner } from "../../lib/model.js";
import { assertChatKey } from "../../lib/env.js";

// ── 状态 ──────────────────────────────────────────────────────
const RouterState = Annotation.Root({
  /** 用户输入 */
  question: Annotation<string>,
  /** 分类结果：tech / chat / complaint */
  category: Annotation<string>,
  /** 走的是哪条分支 */
  route: Annotation<string>,
  /** 最终回复 */
  reply: Annotation<string>,
});

type Route = "tech" | "chat" | "complaint";

async function main() {
  banner("示例 2：路由（先分类，再分发到不同分支）");
  assertChatKey();

  const model = createModel({ temperature: 0.3 });

  // ── 节点 1：分类器 —— 只干一件事：把输入归到三类之一 ──────────
  async function classify(state: typeof RouterState.State) {
    const res = await model.invoke([
      new SystemMessage(
        "你是客服分类器。把用户的消息归为三类之一，只输出一个词，不要任何其它文字：\n" +
          "tech      —— 技术问题（报错、配置、怎么用、原理）\n" +
          "complaint —— 投诉抱怨（表达不满、要求赔偿、要说法）\n" +
          "chat      —— 其它（闲聊、寒暄、无关问题）"
      ),
      new HumanMessage(state.question),
    ]);
    const raw = String(res.content).toLowerCase();
    const category: Route = raw.includes("complaint")
      ? "complaint"
      : raw.includes("tech")
        ? "tech"
        : "chat";
    console.log(`   [节点 classify] 「${state.question}」`);
    console.log(`   [节点 classify] → 模型把它分到：${category}`);
    return { category };
  }

  // ── 三条分支：各自的人格与收尾逻辑都不一样 ────────────────────
  async function techNode(state: typeof RouterState.State) {
    console.log("   [节点 tech] 走「技术支持」分支");
    const res = await model.invoke([
      new SystemMessage(
        "你是资深技术支持工程师。用简洁的分步骤方式给出可执行的技术排障建议，" +
          "最多 6 行、必要时给一行命令。语气专业、直接，不要长篇大论。"
      ),
      new HumanMessage(state.question),
    ]);
    return { route: "tech", reply: `【技术客服】\n${String(res.content).trim()}` };
  }

  async function chatNode(state: typeof RouterState.State) {
    console.log("   [节点 chat] 走「轻松陪聊」分支");
    const res = await model.invoke([
      new SystemMessage("你是友好的客服小助手，语气轻松亲切，简短回应即可。"),
      new HumanMessage(state.question),
    ]);
    return { route: "chat", reply: `【客服小助手】${String(res.content).trim()}` };
  }

  async function complaintNode(state: typeof RouterState.State) {
    console.log("   [节点 complaint] 走「投诉安抚」分支");
    const res = await model.invoke([
      new SystemMessage(
        "你是投诉处理专员。先真诚安抚用户情绪、明确道歉，再说明会登记并跟进，" +
          "不要推卸责任，也不要空口承诺具体赔偿金额。"
      ),
      new HumanMessage(state.question),
    ]);
    // 投诉分支的专属收尾：自动生成一个工单号（其它分支没有这一步）
    const ticket = "TK" + Math.abs(hash(state.question)).toString().slice(0, 6);
    return {
      route: "complaint",
      reply: `【投诉专员】\n${String(res.content).trim()}\n\n（已为您登记工单：${ticket}）`,
    };
  }

  // ── 路由函数：读分类结果，返回要去的分支节点名 ────────────────
  function route(state: typeof RouterState.State): Route {
    const target: Route =
      state.category === "tech" || state.category === "complaint"
        ? state.category
        : "chat";
    console.log(`   [条件边 route] category="${state.category}" → 走向 "${target}"`);
    return target;
  }

  const graph = new StateGraph(RouterState)
    .addNode("classify", classify)
    .addNode("tech", techNode)
    .addNode("chat", chatNode)
    .addNode("complaint", complaintNode)
    .addEdge(START, "classify")
    .addConditionalEdges("classify", route, ["tech", "chat", "complaint"])
    .addEdge("tech", END)
    .addEdge("chat", END)
    .addEdge("complaint", END)
    .compile();

  const inputs = [
    "我的 npm install 一直报 ERESOLVE，怎么解决？",
    "今天的天气真不错呀，你说是吧？",
    "你们的服务太差了！我用了一周就出问题，必须给我一个说法！",
  ];

  for (const question of inputs) {
    console.log(`\n${"─".repeat(72)}`);
    console.log(`👤 输入：${question}`);
    const result = await graph.invoke({ question });
    console.log(`\n✉️  回复：\n${result.reply}`);
  }
}

/** 一个简单的字符串哈希，只为给投诉分支造一个稳定的工单号 */
function hash(s: string): number {
  let h = 0;
  for (let i = 0; i < s.length; i++) h = (h * 31 + s.charCodeAt(i)) | 0;
  return h;
}

main().catch((e) => {
  console.error("❌ 调用失败：", e);
  process.exit(1);
});

运行：

bash
npx tsx 09-design-patterns/code/02-routing.ts

真实输出：


========================================================================
  示例 2：路由（先分类，再分发到不同分支）
========================================================================


────────────────────────────────────────────────────────────────────────
👤 输入：我的 npm install 一直报 ERESOLVE，怎么解决？
   [节点 classify] 「我的 npm install 一直报 ERESOLVE，怎么解决？」
   [节点 classify] → 模型把它分到：tech
   [条件边 route] category="tech" → 走向 "tech"
   [节点 tech] 走「技术支持」分支

✉️  回复：
【技术客服】
ERESOLVE 是依赖树里 peer/版本约束冲突，按顺序处理：

1. 先读报错里冲突的两个包及各自要求的版本范围，确认是谁要求过高。
2. 首选：把冲突依赖升/降到兼容版本后重装（治本）。
3. 临时解：`npm install --legacy-peer-deps`（按 npm 6 旧规则忽略 peer 冲突）。
4. 慎用 `--force`，会写入不满足约束的树，易埋运行时坑。
5. 根治：在 package.json 加 `"overrides": { "冲突包": "版本" }` 统一锁定。
6. 仍乱就删 `node_modules` 和 `package-lock.json` 后重装，别混用 yarn/pnpm 的 lock。

────────────────────────────────────────────────────────────────────────
👤 输入：今天的天气真不错呀，你说是吧？
   [节点 classify] 「今天的天气真不错呀，你说是吧？」
   [节点 classify] → 模型把它分到：chat
   [条件边 route] category="chat" → 走向 "chat"
   [节点 chat] 走「轻松陪聊」分支

✉️  回复：
【客服小助手】是呀，听你这么说心情都跟着晴朗起来啦～☀️ 你那边阳光很好吗？适合出去走走，不过也记得防晒哦！

────────────────────────────────────────────────────────────────────────
👤 输入：你们的服务太差了！我用了一周就出问题，必须给我一个说法！
   [节点 classify] 「你们的服务太差了！我用了一周就出问题，必须给我一个说法！」
   [节点 classify] → 模型把它分到：complaint
   [条件边 route] category="complaint" → 走向 "complaint"
   [节点 complaint] 走「投诉安抚」分支

✉️  回复：
【投诉专员】
非常理解您的心情，才用了一周就出问题，换谁都会生气、失望，这确实是我们服务没有做好。作为投诉处理专员，我先向您郑重道歉：对不起，给您带来了这么差的体验，我们不会回避，也不会推卸责任。

我现在先帮您把问题登记下来，并持续跟进处理。为了更快核实，请您提供一下：
1. 您使用的是哪项产品或服务？
2. 具体出现了什么问题？大概什么时候开始的？
3. 方便提供订单号、手机号或账号吗？
4. 如果有问题截图、视频或报错信息，也可以一起发我。

我登记后会立即转交相关团队核查，并跟进处理进展，尽快给您回复。具体处理方案需要根据核实结果和规则来确认，我现在不能随意向您承诺具体赔偿金额，但一定会认真负责地推动解决，不让您的问题被敷衍。

再次向您道歉，也感谢您愿意把问题反馈给我们。

（已为您登记工单：TK523951）

（提示：三条回复都来自大模型，每次运行措辞会略有不同；但那三行 [节点 classify] → 模型把它分到：xxx 一般稳定——分类这种「短输出、强约束」的任务，模型其实相当可靠。）

💡 路由的三个设计要点：

  1. 分类节点要「只输出一个词」：约束越死，路由越稳。我甚至在路由函数里做了字符串兜底（includes("tech") 之类），防止模型多吐了标点。
  2. 分支节点要「各管各的」：complaint 分支多了一步「生成工单号」，这是 tech/chat 没有的——这就是路由的价值：让每个分支拥有专属逻辑，而不是一个提示词打天下。
  3. 分类结果要打印出来：一旦效果不对，你能立刻看出是「分错了类」还是「分支里答错了」，排查事半功倍。

⚠️ 路由 vs 智能体的边界：如果分类只有两三档、且判据明确，用路由；如果「分类」本身很复杂、或者分类之后还要模型自己决定走多少步，那可能得上智能体。别为了躲避智能体，把路由写成一大坨难维护的 if。

四、模式三：并行化（Parallelization） ​

是什么：把相互独立的子任务同时丢给多个节点执行，最后汇总。官方把它分成两种子形态：

  • 分片（Sectioning）：把一大块输入切成几片，每片交给一个节点并行处理（分摊大输入）。
  • 投票（Voting）：把同一个问题同时问多个节点，再对答案做多数表决（提高可靠性）。

什么时候用：子任务之间没有依赖、又都耗时，串行纯属浪费。分片还能让每片用更小的上下文、更专注的提示词。

和第 3 章的联系：机制就是第 3 章的「fan-out / fan-in + 带 reducer 的字段」。这里我们把两种「形状」放在一个文件里，并用总耗时 vs 各节点耗时之和的对比，把「并行」这件事从「感觉上并行」变成「数据上并行」。

示例 3：分片 + 投票，都用耗时证明并行 ​

typescript
/**
 * 示例 3：并行化（Parallelization）
 *
 * 模式定义（来自官方 workflows-agents）：
 *   把相互独立的子任务「同时」丢给多个节点执行，最后汇总。它有两种常见子形态：
 *
 *   ① 分片（Sectioning）—— 把一大块输入切成几片，每片交给一个节点并行处理。
 *   ② 投票（Voting）    —— 把同一个问题同时问多个节点，再对答案做多数表决。
 *
 * 本文件把两种子形态放在一起演示，并用「时间戳 + 总耗时」证明它们是真并行：
 *   每个 worker 都 sleep 约 1.2 秒。
 *     - 串行：总耗时 ≈ 3 × 1.2s
 *     - 并行：总耗时 ≈ 1 × 1.2s
 */
import { StateGraph, START, END, Annotation } from "@langchain/langgraph";
import { HumanMessage, SystemMessage } from "@langchain/core/messages";
import { createModel, banner } from "../../lib/model.js";
import { assertChatKey } from "../../lib/env.js";

const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
/** 打印带毫秒的时间戳，便于肉眼验证并行 */
const ts = () => new Date().toISOString().slice(11, 23);
/** 把字符串按字符数均分成 n 段 */
function splitInto(text: string, n: number): string[] {
  const size = Math.ceil(text.length / n);
  return Array.from({ length: n }, (_, i) => text.slice(i * size, (i + 1) * size));
}

// ══════════════════════════════════════════════════════════════
//  A. 分片（Sectioning）
// ══════════════════════════════════════════════════════════════

const SectionState = Annotation.Root({
  /** 待分析的长文本 */
  text: Annotation<string>,
  /** 切成 3 段后缓存进状态（由 dispatch 节点填） */
  parts: Annotation<string[]>,
  /** 3 个 worker 的产出，带 reducer 才能并发写入 */
  findings: Annotation<string[]>({
    reducer: (left, right) => left.concat(right),
    default: () => [],
  }),
  /** 每个 worker 各自的耗时（ms），用于证明并行 */
  durations: Annotation<number[]>({
    reducer: (left, right) => left.concat(right),
    default: () => [],
  }),
  /** 汇总结论 */
  report: Annotation<string>,
});

// ══════════════════════════════════════════════════════════════
//  B. 投票（Voting）
// ══════════════════════════════════════════════════════════════

const VoteState = Annotation.Root({
  /** 要表决的问题 */
  question: Annotation<string>,
  /** 三个「选民」各自的回答，带 reducer 并发写入 */
  votes: Annotation<string[]>({
    reducer: (left, right) => left.concat(right),
    default: () => [],
  }),
  /** 每个选民各自的耗时（ms） */
  durations: Annotation<number[]>({
    reducer: (left, right) => left.concat(right),
    default: () => [],
  }),
  /** 多数表决结果 */
  verdict: Annotation<string>,
});

async function main() {
  assertChatKey();
  // 投票需要一点随机性，temperature 调高，让三个「选民」可能给出不同答案
  const model = createModel({ temperature: 0.9 });
  // 分片分析要稳定，temperature 调低
  const stableModel = createModel({ temperature: 0 });

  // ══════════════════════════════════════════════════════════
  //  A. 分片（Sectioning）：长文切 3 段，并行分析
  // ══════════════════════════════════════════════════════════
  banner("示例 3A：分片（Sectioning）—— 长文切 3 段并行分析");

  const longText =
    "LangGraph 是一个用于构建有状态、多参与者 AI 应用的编排框架。它把工作流建模为一张有向图，" +
    "节点是函数，边决定控制流。相比传统链式调用，它天然支持循环、分支与并行。" +
    "它的状态是一个在节点间传递的共享对象，每个节点只返回需要更新的字段，由 reducer 决定如何合并。" +
    "并行能力来自「超步」：同一超步内没有依赖关系的节点会被并发调度。" +
    "此外，它支持检查点与持久化，可以把图的状态存进数据库，从而实现断点续跑与跨会话记忆。" +
    "当需要人工审批时，它还能在任意节点中断，等人确认后再继续。";

  function dispatchSection(state: typeof SectionState.State) {
    const parts = splitInto(state.text, 3);
    console.log(`   [节点 dispatch] 把 ${state.text.length} 字的长文切成 3 段 @ ${ts()}`);
    return { parts };
  }

  async function makeSectionWorker(idx: number) {
    return async function worker(state: typeof SectionState.State) {
      const start = Date.now();
      console.log(`   [节点 section_${idx}] 开始分析第 ${idx} 段 @ ${ts()}`);
      await sleep(1200); // 模拟「调用模型 + 处理」的耗时
      const res = await stableModel.invoke([
        new SystemMessage("用一句话概括下面这段文字的核心要点，只要一句话。"),
        new HumanMessage(state.parts[idx - 1]),
      ]);
      const dur = Date.now() - start;
      console.log(`   [节点 section_${idx}] 完成 @ ${ts()}（本节点耗时 ${dur}ms）`);
      return {
        findings: [`第 ${idx} 段 → ${String(res.content).trim()}`],
        durations: [dur],
      };
    };
  }

  function mergeSections(state: typeof SectionState.State) {
    console.log(`   [节点 merge] 收齐 ${state.findings.length} 份分段结论 @ ${ts()}`);
    return { report: state.findings.join("\n") };
  }

  const sectionGraph = new StateGraph(SectionState)
    .addNode("dispatch", dispatchSection)
    .addNode("section_1", await makeSectionWorker(1))
    .addNode("section_2", await makeSectionWorker(2))
    .addNode("section_3", await makeSectionWorker(3))
    .addNode("merge", mergeSections)
    .addEdge(START, "dispatch")
    .addEdge("dispatch", "section_1")
    .addEdge("dispatch", "section_2")
    .addEdge("dispatch", "section_3")
    .addEdge("section_1", "merge")
    .addEdge("section_2", "merge")
    .addEdge("section_3", "merge")
    .addEdge("merge", END)
    .compile();

  const tSection = Date.now();
  const sectionResult = await sectionGraph.invoke({ text: longText });
  const sectionCost = Date.now() - tSection;
  console.log(`\n📄 分片报告：\n   ${sectionResult.report.replace(/\n/g, "\n   ")}`);
  const sSum = sectionResult.durations.reduce((a, b) => a + b, 0);
  const sMax = Math.max(...sectionResult.durations);
  console.log(`\n⏱️  3 个节点耗时之和 = ${sSum}ms，单个最慢 = ${sMax}ms，整图实际耗时 = ${sectionCost}ms`);
  console.log(
    sectionCost < sSum * 0.7
      ? `   ✅ 实际耗时远小于「耗时之和」→ 3 段是并行的（≈ 单个最慢节点，而不是三者相加）`
      : "   ⚠️ 实际耗时接近「耗时之和」→ 看起来是串行"
  );

  // ══════════════════════════════════════════════════════════
  //  B. 投票（Voting）：同一问题问 3 次，取多数
  // ══════════════════════════════════════════════════════════
  banner("示例 3B：投票（Voting）—— 同一问题问 3 次，取多数");

  const question =
    "下面这条用户评论整体偏正面还是偏负面？只回答一个词，不要解释：正面 / 负面。\n" +
    "评论：「音质确实惊艳，细节丰富；但戴久了夹耳朵，续航也只有 3 小时，有点失望。」";

  async function makeVoter(idx: number) {
    return async function voter(state: typeof VoteState.State) {
      const start = Date.now();
      console.log(`   [节点 voter_${idx}] 开始投票 @ ${ts()}`);
      await sleep(1200);
      const res = await model.invoke([new HumanMessage(state.question)]);
      const raw = String(res.content).trim();
      const vote = raw.includes("负面") ? "负面" : "正面";
      const dur = Date.now() - start;
      console.log(
        `   [节点 voter_${idx}] 投出：${vote}（原始："${raw.slice(0, 10)}"，耗时 ${dur}ms）`
      );
      return { votes: [vote], durations: [dur] };
    };
  }

  function tally(state: typeof VoteState.State) {
    const counts = new Map<string, number>();
    for (const v of state.votes) counts.set(v, (counts.get(v) ?? 0) + 1);
    const sorted = [...counts.entries()].sort((a, b) => b[1] - a[1]);
    const [winner] = sorted[0];
    console.log(
      `   [节点 tally] 计票 ${JSON.stringify(Object.fromEntries(counts))} → 多数票：${winner}`
    );
    return { verdict: winner };
  }

  const voteGraph = new StateGraph(VoteState)
    .addNode("voter_1", await makeVoter(1))
    .addNode("voter_2", await makeVoter(2))
    .addNode("voter_3", await makeVoter(3))
    .addNode("tally", tally)
    .addEdge(START, "voter_1")
    .addEdge(START, "voter_2")
    .addEdge(START, "voter_3")
    .addEdge("voter_1", "tally")
    .addEdge("voter_2", "tally")
    .addEdge("voter_3", "tally")
    .addEdge("tally", END)
    .compile();

  const tVote = Date.now();
  const voteResult = await voteGraph.invoke({ question });
  const voteCost = Date.now() - tVote;
  console.log(`\n🗳️  三票：${JSON.stringify(voteResult.votes)}`);
  console.log(`🏆 多数表决结果：${voteResult.verdict}`);
  const vSum = voteResult.durations.reduce((a, b) => a + b, 0);
  const vMax = Math.max(...voteResult.durations);
  console.log(`\n⏱️  3 个选民耗时之和 = ${vSum}ms，单个最慢 = ${vMax}ms，整图实际耗时 = ${voteCost}ms`);
  console.log(
    voteCost < vSum * 0.7
      ? "   ✅ 实际耗时远小于「耗时之和」→ 3 个选民是并行的"
      : "   ⚠️ 实际耗时接近「耗时之和」→ 看起来是串行"
  );

  console.log(
    "\n💡 两种子形态的共同点：子任务之间「互不依赖」，所以能并行。" +
      "分片用来「分摊大输入」（每片更小、更快、还能各自开不同提示词）；" +
      "投票用来「提高可靠性」（多数票能抵消单次随机出错，也常用于给结果打置信度）。"
  );
}

main().catch((e) => {
  console.error("❌ 调用失败：", e);
  process.exit(1);
});

运行：

bash
npx tsx 09-design-patterns/code/03-parallelization.ts

真实输出：


========================================================================
  示例 3A：分片（Sectioning）—— 长文切 3 段并行分析
========================================================================

   [节点 dispatch] 把 241 字的长文切成 3 段 @ 03:51:09.541
   [节点 section_1] 开始分析第 1 段 @ 03:51:09.542
   [节点 section_2] 开始分析第 2 段 @ 03:51:09.542
   [节点 section_3] 开始分析第 3 段 @ 03:51:09.542
   [节点 section_3] 完成 @ 03:51:12.398（本节点耗时 2856ms）
   [节点 section_2] 完成 @ 03:51:12.670（本节点耗时 3128ms）
   [节点 section_1] 完成 @ 03:51:12.674（本节点耗时 3132ms）
   [节点 merge] 收齐 3 份分段结论 @ 03:51:12.678

📄 分片报告：
   第 1 段 → LangGraph 是一个用有向图（节点为函数、边控制流）编排有状态、多参与者 AI 应用的框架，相比传统链式调用更天然支持循环等控制流。
   第 2 段 → 该机制通过节点间共享状态、由 reducer 合并各节点返回的增量更新，并利用“超步”并行执行同一超步内无依赖的节点，实现分支与并行。
   第 3 段 → 该系统支持节点并发调度、检查点持久化（实现断点续跑与跨会话记忆），并能在任意节点中断等待人工审批后继续。

⏱️  3 个节点耗时之和 = 9116ms，单个最慢 = 3132ms，整图实际耗时 = 3161ms
   ✅ 实际耗时远小于「耗时之和」→ 3 段是并行的（≈ 单个最慢节点，而不是三者相加）

========================================================================
  示例 3B：投票（Voting）—— 同一问题问 3 次，取多数
========================================================================

   [节点 voter_1] 开始投票 @ 03:51:12.709
   [节点 voter_2] 开始投票 @ 03:51:12.710
   [节点 voter_3] 开始投票 @ 03:51:12.710
   [节点 voter_3] 投出：负面（原始："负面"，耗时 2339ms）
   [节点 voter_2] 投出：负面（原始："负面"，耗时 2585ms）
   [节点 voter_1] 投出：负面（原始："负面"，耗时 2724ms）
   [节点 tally] 计票 {"负面":3} → 多数票：负面

🗳️  三票：["负面","负面","负面"]
🏆 多数表决结果：负面

⏱️  3 个选民耗时之和 = 7648ms，单个最慢 = 2724ms，整图实际耗时 = 2741ms
   ✅ 实际耗时远小于「耗时之和」→ 3 个选民是并行的

💡 两种子形态的共同点：子任务之间「互不依赖」，所以能并行。分片用来「分摊大输入」（每片更小、更快、还能各自开不同提示词）；投票用来「提高可靠性」（多数票能抵消单次随机出错，也常用于给结果打置信度）。

怎么看出它是并行？ 看分片那几行时间戳：

   [节点 section_1] 开始分析第 1 段 @ 03:51:09.542
   [节点 section_2] 开始分析第 2 段 @ 03:51:09.542
   [节点 section_3] 开始分析第 3 段 @ 03:51:09.542

三个 开始 在同一毫秒（.542）。如果是串行，它们必然错开几秒。再看总结那行：

3 个节点耗时之和 = 9116ms，单个最慢 = 3132ms，整图实际耗时 = 3161ms

三个节点加起来干了 9116ms 的活，整图只花了 3161ms——约等于「最慢的那一个」，而不是「三者相加」。这就是并行的铁证。

💡 实测小插曲：细心的你会发现三个节点耗时并不一样（2856 / 3128 / 3132ms）。因为每个节点里除了固定的 sleep(1200)，还有一次真实的模型调用，而模型返回时间本就随机。所以「证明并行」这件事，不能只看总耗时和固定阈值（那样会被模型延迟带偏），而要把每个节点的耗时记下来，和实际总耗时对比——这正是本示例比第 3 章更严谨的地方。

⚠️ 投票不总是分裂：本次三票全投「负面」，多数表决退化成「一致同意」。这很正常——强约束的二选一问题，模型往往很一致。投票真正的价值在边界模糊的判断题上：单次调用可能随机犯错，多问几次取多数，能显著降低单次错误的概率。也可以把它当「置信度」用：3 比 0 说明高置信，2 比 1 说明这事有争议。

五、模式四：编排者-工作者（Orchestrator-Worker） ​

是什么：一个「编排者（orchestrator）」节点用模型把任务动态拆成若干子任务（拆几个由模型决定，不写死），再用 Send 把每个子任务分发给一个「工作者（worker）」节点并行执行，最后由一个汇总节点合成产出。

START → orchestrator ──Send×N──► worker(×N 并行) ──► synthesize → END
          （模型动态拆子任务）      （每个 worker 处理一个子任务）  （汇总成报告）

和并行化的关键区别：

并行化编排者-工作者
分支数量写死（代码里固定 3 个）运行时算出来（模型说几个就几个）
每个分支做什么写死由模型当场分配
适合任务结构固定任务结构随输入变化

什么时候用：任务的子任务数量不固定，必须靠模型读懂需求后才能决定要分几块。比如「写一份决策备忘录」——简单需求分 3 块就够，复杂需求要分 5、6 块。

和第 3 章的联系：机制是第 3 章的 Send（把一个列表 map 成并发任务）。这里的关键升级是——那个列表不是代码写死的，而是模型现算的。

示例 4：模型拆任务 + Send 动态分发 ​

typescript
/**
 * 示例 4：编排者-工作者（Orchestrator-Worker）
 *
 * 模式定义（来自官方 workflows-agents）：
 *   一个「编排者（orchestrator）」节点用模型把任务**动态**拆成若干子任务，
 *   （子任务数量由模型根据任务复杂度决定，不写死），
 *   再用 Send 把每个子任务分发给一个「工作者（worker）」节点并行执行，
 *   最后由一个汇总节点把结果合成最终产出。
 *
 * 与「并行化」的区别：并行化的分支是**写死**的（固定 3 个）；
 *   编排者-工作者的分支是**运行时算出来**的（模型说几个就几个）。
 *
 * 图结构：
 *
 *   START → orchestrator ──Send×N──► worker(×N 并行) ──► synthesize → END
 *            （模型动态拆子任务）      （每个 worker 处理一个子任务）  （汇总成报告）
 */
import {
  StateGraph,
  START,
  END,
  Annotation,
  Send,
} from "@langchain/langgraph";
import { HumanMessage, SystemMessage } from "@langchain/core/messages";
import { createModel, banner } from "../../lib/model.js";
import { assertChatKey } from "../../lib/env.js";

// ── 状态 ─────────────────────────────────────────────────────
const WorkState = Annotation.Root({
  /** 总任务 */
  task: Annotation<string>,
  /** 编排者拆出的子任务列表（运行时才知道有几条） */
  subtasks: Annotation<string[]>(),
  /** Send 派发给 worker 的「单条子任务」（worker 只读它） */
  subtask: Annotation<string>,
  /** 各 worker 的产出，带 reducer 才能并发写入 */
  findings: Annotation<{ subtask: string; detail: string }[]>({
    reducer: (left, right) => left.concat(right),
    default: () => [],
  }),
  /** 最终报告 */
  report: Annotation<string>,
});

async function main() {
  banner("示例 4：编排者-工作者（模型动态拆任务 + Send 分发）");
  assertChatKey();

  const model = createModel({ temperature: 0.2 });

  // ── 节点 1：编排者 —— 让模型「动态」决定拆几个子任务 ──────────
  async function orchestrator(state: typeof WorkState.State) {
    const res = await model.invoke([
      new SystemMessage(
        "你是任务规划师。把用户的任务拆成若干个可以『并行独立完成』的子任务。" +
          "子任务数量由你根据任务复杂度决定（一般 3~5 个）：简单的就少拆，复杂的就多拆。" +
          "每个子任务一行，以『- 』开头，一句话说清要研究/产出什么。不要编号以外的多余解释。"
      ),
      new HumanMessage(state.task),
    ]);
    const subtasks = String(res.content)
      .split("\n")
      .map((l) => l.replace(/^[\s\-*0-9.、]+/, "").trim())
      .filter((l) => l.length > 0);
    console.log(`   [节点 orchestrator] 模型把任务拆成了 ${subtasks.length} 个子任务：`);
    subtasks.forEach((s, i) => console.log(`      ${i + 1}. ${s}`));
    return { subtasks };
  }

  // ── 分发函数：把每个子任务包成一个 Send，并发派给 worker ──────
  function fanOut(state: typeof WorkState.State): Send[] {
    console.log(
      `   [条件边 fanOut] 把 ${state.subtasks.length} 个子任务用 Send 并发派发到 worker`
    );
    return state.subtasks.map((s) => new Send("worker", { subtask: s }));
  }

  // ── 节点 2：工作者 —— 每个实例只处理「自己那条子任务」 ────────
  async function worker(state: typeof WorkState.State) {
    console.log(`   [节点 worker] 领到子任务：${state.subtask.slice(0, 30)}…`);
    const res = await model.invoke([
      new SystemMessage(
        "你是资深研究人员。针对给定的子任务，给出 2~3 条最关键、最具体的结论。" +
          "每行一条，直接给结论，不要寒暄、不要复述问题。"
      ),
      new HumanMessage(`总任务背景：${state.task}\n\n你负责的子任务：${state.subtask}`),
    ]);
    return {
      findings: [{ subtask: state.subtask, detail: String(res.content).trim() }],
    };
  }

  // ── 节点 3：汇总 —— 把所有 worker 产出合成一份报告 ────────────
  async function synthesize(state: typeof WorkState.State) {
    console.log(`   [节点 synthesize] 收到 ${state.findings.length} 份 worker 产出，开始汇总`);
    const material = state.findings
      .map((f) => `## 子任务：${f.subtask}\n${f.detail}`)
      .join("\n\n");
    const res = await model.invoke([
      new SystemMessage(
        "你是主编。根据下面各子任务的研究结论，写一份结构清晰的最终报告，" +
          "用「一、二、三…」分小节，小节标题对应子任务主题，最后给一句总体结论。"
      ),
      new HumanMessage(`总任务：${state.task}\n\n各部门材料：\n${material}`),
    ]);
    return { report: String(res.content).trim() };
  }

  const graph = new StateGraph(WorkState)
    .addNode("orchestrator", orchestrator)
    .addNode("worker", worker)
    .addNode("synthesize", synthesize)
    .addEdge(START, "orchestrator")
    // 关键：编排者不连固定边，而是用 Send 动态派发
    .addConditionalEdges("orchestrator", fanOut)
    .addEdge("worker", "synthesize")
    .addEdge("synthesize", END)
    .compile();

  const task =
    "为「一家 20 人的 SaaS 初创公司，要不要自建 LLM 编排框架」写一份决策备忘录。";
  console.log(`📋 总任务：${task}\n`);

  const result = await graph.invoke({ task });

  console.log(`\n📦 状态里的 findings 条数：${result.findings.length}`);
  console.log("\n" + "─".repeat(72));
  console.log("📄 最终报告");
  console.log("─".repeat(72));
  console.log(result.report);

  console.log(
    "\n💡 注意：本示例的图结构里 worker 只有「一个」节点，但运行时被 Send 复制成了 N 份并发执行；" +
      "N 由 orchestrator 里的模型当场决定。这正是它和「并行化」最大的区别——" +
      "并行化的分支数在写代码时就固定了，编排者-工作者的分支数在运行时才确定。"
  );
}

main().catch((e) => {
  console.error("❌ 调用失败：", e);
  process.exit(1);
});

运行：

bash
npx tsx 09-design-patterns/code/04-orchestrator-worker.ts

真实输出（报告较长，此处节选开头）：


========================================================================
  示例 4：编排者-工作者（模型动态拆任务 + Send 分发）
========================================================================

📋 总任务：为「一家 20 人的 SaaS 初创公司，要不要自建 LLM 编排框架」写一份决策备忘录。

   [节点 orchestrator] 模型把任务拆成了 5 个子任务：
      1. 明确这家 20 人 SaaS 初创公司的 LLM 使用场景、工程能力、预算约束与决策标准，输出备忘录的问题陈述和评估框架。
      2. 研究自建 LLM 编排框架的架构选项、最小可行团队配置、开发与维护成本、可扩展性及主要失败风险。
      3. 研究成熟开源、商业和云厂商 LLM 编排方案的选型清单、定价、功能边界、集成成本与供应商锁定风险。
      4. 对比自建与采购/开源方案在 12–18 个月总拥有成本、迭代速度、数据安全、招聘负担和战略差异化上的利弊。
      5. 设计决策门槛、触发条件、推荐路径及分阶段迁移或退出方案，形成可直接填入决策备忘录的结论框架。
   [条件边 fanOut] 把 5 个子任务用 Send 并发派发到 worker
   [节点 worker] 领到子任务：明确这家 20 人 SaaS 初创公司的 LLM 使用场景、…
   [节点 worker] 领到子任务：研究自建 LLM 编排框架的架构选项、最小可行团队配置、开发…
   [节点 worker] 领到子任务：研究成熟开源、商业和云厂商 LLM 编排方案的选型清单、定价…
   [节点 worker] 领到子任务：对比自建与采购/开源方案在 12–18 个月总拥有成本、迭代…
   [节点 worker] 领到子任务：设计决策门槛、触发条件、推荐路径及分阶段迁移或退出方案，形成…
   [节点 synthesize] 收到 5 份 worker 产出，开始汇总

📦 状态里的 findings 条数：5

────────────────────────────────────────────────────────────────────────
📄 最终报告
────────────────────────────────────────────────────────────────────────
**决策备忘录：20 人 SaaS 初创是否自建 LLM 编排框架**

**致：** CEO / CTO / 创始人  
**自：** 战略与工程决策组  
**主题：** 在 2–4 名全栈、无专职 ML/GPU 运维、月 LLM 预算 $2k–$5k、4–8 周交付约束下，是否自建 LLM 编排框架

---

## 一、问题陈述与评估框架：场景、约束与成功标准

**问题陈述：** 公司应优先验证三类嵌入现有工作流的 LLM 场景：客服工单摘要/回复草稿、销售线索摘要/外联草稿、产品内文档问答或自然语言报表。成功线为人工处理时长下降 20%+ 或试用转化提升 5%+。明确不做通用 Agent、不做微调、不自建模型。

**核心约束：**
- 工程能力：2–4 名全栈工程师，无专职 ML/GPU 运维。
- 预算：月 LLM 预算 $2k–$5k。
- 交付：4–8 周内可验证。
- 技术路线：优先托管 API，如 OpenAI / Anthropic / Azure。

……（后续第二至五节及总体结论省略）

💡 注意：本示例的图结构里 worker 只有「一个」节点，但运行时被 Send 复制成了 N 份并发执行；N 由 orchestrator 里的模型当场决定。这正是它和「并行化」最大的区别——并行化的分支数在写代码时就固定了，编排者-工作者的分支数在运行时才确定。

💡 看那 5 行 [节点 worker] 领到子任务：…：代码里只注册了一个 worker 节点，运行时却并发了 5 个实例。数量 5 不是写死的——是 orchestrator 里那句「子任务数量由你根据任务复杂度决定」让模型现算的。换一个更简单的任务，它可能只拆 3 个；换更复杂的，可能拆 6 个。这就是「动态」的含义。

⚠️ 两个必须记住的坑：

  1. Send 的第二参数是「给该子任务的增量状态」。我们传的是 { subtask: s }，所以 worker 节点里读 state.subtask 就能拿到「自己那条」。不同 worker 实例读到的 subtask 是不同的，这正是并发的关键。
  2. worker 的产出字段必须带 reducer。5 个实例在同一个超步里各写一次 findings，没有 reducer 框架就不知道该合并（会抛 LastValue can only receive one value per step，第 3 章演示过）。这里用的 reducer 是 (l, r) => l.concat(r)。

💡 Send 派发的目标是「节点名」：.addConditionalEdges("orchestrator", fanOut) 里的 fanOut 返回的是 Send[]，每个 Send 的第一个参数就是目标节点名（"worker"）。所以同一个分发函数可以同时派给多个不同节点（比如按子任务类型分给 worker_a / worker_b），这是它比死写 fan-out 灵活的地方。

六、模式五：评估者-优化者（Evaluator-Optimizer） ​

是什么：一个「生成者」先产出草稿；一个「评估者」打分并给出具体的改进意见；一个「优化者」按意见重写；如此循环，直到评估通过或到达轮次上限。

START → generate → evaluate ──(通过 / 到上限)──► END
                   ▲   │
                   │   └──(未通过)──► revise
                   └────────────────────┘

什么时候用：产出难以一次写好、但又有明确评价标准的任务——文案、翻译、代码重构都是典型。关键是「评估」可以自动化：模型能对着一个标准给出分数和可执行的改进方向。

和第 3 章的联系：机制就是第 3 章的「条件边回指上游」构成的循环 + recursionLimit 兜底。本章的升级点是——循环的「退出条件」由评估者的分数决定，并且必须设轮次上限，否则又贵又可能转圈。

示例 5：生成 → 评审（具体意见）→ 重写 ​

typescript
/**
 * 示例 5：评估者-优化者（Evaluator-Optimizer）
 *
 * 模式定义（来自官方 workflows-agents）：
 *   一个「生成者」先产出草稿；一个「评估者」打分并给出**具体**的改进意见；
 *   「优化者」按意见重写；如此循环，直到评估通过或到达轮次上限。
 *   它适合「难以一次写好、但有明确评价标准」的产出（文案、翻译、代码重构…）。
 *
 * 图结构（循环）：
 *
 *   START → generate → evaluate ──(通过 / 到上限)──► END
 *                        ▲   │
 *                        │   └──(未通过)──► revise
 *                        └────────────────────┘
 *
 * 每一轮都会打印：第几轮、评分、具体意见、新版本。
 */
import { StateGraph, START, END, Annotation } from "@langchain/langgraph";
import { HumanMessage, SystemMessage } from "@langchain/core/messages";
import { createModel, banner } from "../../lib/model.js";
import { assertChatKey } from "../../lib/env.js";

// ── 状态 ─────────────────────────────────────────────────────
const EvalState = Annotation.Root({
  /** 任务要求 */
  task: Annotation<string>,
  /** 当前草稿 */
  draft: Annotation<string>,
  /** 评估者的具体改进意见 */
  critique: Annotation<string>,
  /** 评分（1~10） */
  score: Annotation<number>,
  /** 当前是第几轮（从 0 开始） */
  round: Annotation<number>,
  /** 最多优化几轮 */
  maxRounds: Annotation<number>,
  /** 通过线 */
  passLine: Annotation<number>,
  /** 每一轮的版本记录，带 reducer 累加 */
  history: Annotation<string[]>({
    reducer: (left, right) => left.concat(right),
    default: () => [],
  }),
});

async function main() {
  banner("示例 5：评估者-优化者（生成 → 评审 → 重写，循环到达标）");
  assertChatKey();

  const model = createModel({ temperature: 0.6 });

  const TASK =
    "为在线课程《LangGraph 中文入门》写一句 30 字以内、让人想点「报名」的宣传标语。";
  const PASS_LINE = 8; // 评分 >= 8 视为通过
  const MAX_ROUNDS = 3;

  // ── 节点 1：生成者 —— 产出第一版草稿 ─────────────────────────
  async function generate(state: typeof EvalState.State) {
    console.log("   [节点 generate] 产出第一版草稿…");
    const res = await model.invoke([
      new SystemMessage("你是广告文案。根据任务要求，直接给出 1 条标语，不要解释。"),
      new HumanMessage(state.task),
    ]);
    const draft = String(res.content).trim();
    console.log(`   → 初版：${draft}`);
    return { draft, round: 0, history: [`第 0 轮（初版）：${draft}`] };
  }

  // ── 节点 2：评估者 —— 打分 + 给具体的改进意见 ─────────────────
  async function evaluate(state: typeof EvalState.State) {
    const res = await model.invoke([
      new SystemMessage(
        "你是严格的文案评审。评估下面的标语，并**务必**给出可执行的具体修改建议。\n" +
          "评分校准（请严格遵守）：\n" +
          "  - 初稿通常有若干明显问题 → 给 6~7 分。\n" +
          "  - 当标语已同时满足『字数达标、成果具体、有明确的行动号召』这三项时，" +
          "应给 8 分及以上（说明可以投放了），不要再吹毛求疵。\n" +
          "输出格式严格如下：\n" +
          "第一行：一个 1~10 的整数分数（数字本身，不要别的字）。\n" +
          "从第二行起：针对当前这条标语的、具体的改进意见（说清哪里不好、往哪个方向改，" +
          "例如『太抽象，建议加入具体场景』）。不要复述标语，不要空泛地说“再优化”。"
      ),
      new HumanMessage(
        `任务要求：${state.task}\n\n待评估的标语：${state.draft}`
      ),
    ]);
    const text = String(res.content).trim();
    const lines = text.split("\n").map((l) => l.trim()).filter(Boolean);
    const match = lines[0].match(/\d+/);
    const score = match ? Math.min(10, Math.max(1, Number(match[0]))) : 0;
    const critique = lines.slice(1).join("\n") || "（评估者未给具体意见）";
    console.log(`\n   [节点 evaluate] 第 ${state.round} 轮评审结果：`);
    console.log(`      ★ 评分：${score} / 10（通过线 ${state.passLine}）`);
    console.log(`      ✎ 具体意见：${critique.replace(/\n/g, "\n                 ")}`);
    return { score, critique };
  }

  // ── 节点 3：优化者 —— 按意见重写 ─────────────────────────────
  async function revise(state: typeof EvalState.State) {
    const nextRound = state.round + 1;
    const res = await model.invoke([
      new SystemMessage(
        "你是广告文案。严格根据评审意见重写标语，直接输出新标语一条，不要解释、不要复述意见。"
      ),
      new HumanMessage(
        `任务要求：${state.task}\n\n当前标语：${state.draft}\n\n评审意见：${state.critique}`
      ),
    ]);
    const draft = String(res.content).trim();
    console.log(`   [节点 revise] 第 ${nextRound} 轮重写完成 → ${draft}`);
    return { draft, round: nextRound, history: [`第 ${nextRound} 轮（重写）：${draft}`] };
  }

  // ── 路由：通过或到上限就结束，否则回炉重写 ────────────────────
  function route(state: typeof EvalState.State): "revise" | typeof END {
    if (state.score >= state.passLine) {
      console.log(`   [条件边] 评分 ${state.score} ≥ ${state.passLine}，通过 ✅`);
      return END;
    }
    if (state.round >= state.maxRounds) {
      console.log(`   [条件边] 已达最大轮次 ${state.maxRounds}，停止（未达标）⚠️`);
      return END;
    }
    console.log(`   [条件边] 评分 ${state.score} < ${state.passLine}，回去重写 → revise`);
    return "revise";
  }

  const graph = new StateGraph(EvalState)
    .addNode("generate", generate)
    .addNode("evaluate", evaluate)
    .addNode("revise", revise)
    .addEdge(START, "generate")
    .addEdge("generate", "evaluate")
    .addConditionalEdges("evaluate", route, ["revise", END])
    .addEdge("revise", "evaluate") // 循环就靠这一条边
    .compile();

  console.log(`📋 任务：${TASK}\n`);
  const result = await graph.invoke({
    task: TASK,
    round: 0,
    maxRounds: MAX_ROUNDS,
    passLine: PASS_LINE,
  });

  console.log("\n" + "─".repeat(72));
  console.log("📜 全程版本记录");
  console.log("─".repeat(72));
  result.history.forEach((h) => console.log(`   ${h}`));
  console.log(`\n🏁 最终标语：${result.draft}`);
  console.log(
    `   共优化 ${result.round} 轮，最终评分 ${result.score}/${result.passLine}，` +
      (result.score >= result.passLine ? "已达标 ✅" : "未达标（到达轮次上限）⚠️")
  );
}

main().catch((e) => {
  console.error("❌ 调用失败：", e);
  process.exit(1);
});

运行：

bash
npx tsx 09-design-patterns/code/05-evaluator-optimizer.ts

真实输出：


========================================================================
  示例 5：评估者-优化者（生成 → 评审 → 重写，循环到达标）
========================================================================

📋 任务：为在线课程《LangGraph 中文入门》写一句 30 字以内、让人想点「报名」的宣传标语。

   [节点 generate] 产出第一版草稿…
   → 初版：LangGraph中文入门，零基础玩转AI智能体，立即报名！

   [节点 evaluate] 第 0 轮评审结果：
      ★ 评分：7 / 10（通过线 8）
      ✎ 具体意见：问题主要是：成果不够具体，“玩转AI智能体”太宽泛，用户不知道学完到底能做出什么；“立即报名”也缺少立刻行动的理由。
                 1. 把“玩转AI智能体”换成可验证的交付物，例如“做出会查资料的AI助手”“搭出自动客服机器人”“连接3个工具的工作流”。
                 2. 用时间或门槛承诺替代“玩转”这类虚词，例如“3节课”“2小时上手”“不用Python基础”“全程中文案例”。
                 3. 给“立即报名”加一个理由或紧迫感，例如“限时免费试听”“报名送源码/模板”，比单独喊报名更有转化力。
                 4. 可压缩课程名，把字数让给具体收益。参考改法：`LangGraph中文：做出会查资料的AI助手，立即报名！`（29字）或 `LangGraph中文入门：3节课做出AI助手，立即报名！`（29字）。第一个成果更具体，优先用第一个方向。
   [条件边] 评分 7 < 8，回去重写 → revise
   [节点 revise] 第 1 轮重写完成 → LangGraph中文：做出会查资料的AI助手，报名送源码！

   [节点 evaluate] 第 1 轮评审结果：
      ★ 评分：8 / 10（通过线 8）
      ✎ 具体意见：1. 课程名不完整：当前只写到“LangGraph中文”，缺少《LangGraph 中文入门》里的“入门”，品牌辨识度会弱。若补全并控制字数，可删掉“做出”两字，改成：LangGraph中文入门：会查资料的AI助手，报名送源码！（按字符计正好30字）
                 2. 字数卡在30字上限，部分平台若把“LangGraph”按字母逐一计数、或额外空格/标点规则不同，可能超限。建议压到28字以内留余量，例如删“做出”：LangGraph中文：会查资料的AI助手，报名送源码！
                 3. “会查资料”方向具体，但还可再落到“学完能做什么”。可把“会查资料”升级为“能查资料答题”或“能检索后回答”，例如：LangGraph中文：能查资料答题的AI助手，报名送源码！这样用户更容易判断成果。
                 4. “报名送源码”有明确行动号召，但“源码”作为福利略模糊。若实际交付的是完整可运行项目，建议写成“报名送可运行源码”或“报名送源码+文档”；若字数不够，至少在落地页明确“可运行、含文档”，避免用户以为只是代码片段。
   [条件边] 评分 8 ≥ 8，通过 ✅

────────────────────────────────────────────────────────────────────────
📜 全程版本记录
────────────────────────────────────────────────────────────────────────
   第 0 轮（初版）：LangGraph中文入门，零基础玩转AI智能体，立即报名！
   第 1 轮（重写）：LangGraph中文：做出会查资料的AI助手，报名送源码！

🏁 最终标语：LangGraph中文：做出会查资料的AI助手，报名送源码！
   共优化 1 轮，最终评分 8/8，已达标 ✅

💡 看第 0 轮那条「具体意见」——它没有说一句空话，而是直接给了两条可照抄的改写（LangGraph中文：做出会查资料的AI助手，立即报名！（29字））。这就是「具体」的含义，也是评估者-优化者能奏效的前提：意见越可执行，下一轮改得越好。 如果评估者只会说「再优化一下」，这个循环就是在空转烧钱。

⚠️ 评估者-优化者的三个必设的保险丝：

  1. 必须设轮次上限（maxRounds）。否则可能永远差一分，无限循环。示例里到上限就 return END（route 里那段 state.round >= state.maxRounds）。这也是第 3 章 recursionLimit 之外、模式层面的第二道保险。
  2. 评估要「校准」。我在评估者提示词里写了「初稿给 6~7 分，满足三项才给 8 分」——这叫评分校准。不校准的话，同一个模型可能对任何东西都给 7 分，循环永远出不去（本项目实测第一版就踩了这个坑：三轮都卡在 7 分）。
  3. 评估者和生成者最好「人格不同」。一个力求完美、一个听话照改。如果两者是同一套提示词，评估者容易「自我欣赏」，给不出真意见。

七、本章结论：工作流 vs 智能体（Workflow vs Agent） ​

前面五种模式——提示链、路由、并行化、编排者-工作者、评估者-优化者——全都属于「工作流」：路径是预先设计好的，模型只在既定格子里干活。而「智能体（Agent）」是另一种东西：路径由模型当场决定，它自己决定走几步、调哪些工具、什么时候停。

官方 workflows-agents 一页的核心论点就一句话：

💡 能用固定工作流解决的，就不要上自主智能体。

为什么？因为两者的成本结构完全不同：

工作流自主智能体
路径写死模型当场决定
调用次数可预测（每次都一样）不可预测（随输入变化）
延迟/成本稳定可控会飘，可能突然翻几倍
调试容易（每步都知道去哪）难（模型为什么这么走要翻轨迹）
灵活度低（只能走设计好的路）高（能处理没预设过的任务）
适合步骤已知、要稳定性任务开放、要灵活性

示例 6：同一个需求，两种实现，并排对比 ​

这个示例用同一个需求（分析一组季度数据、找出增长最快的季度、并总结）分别用两种方式实现，然后用「模型调用次数 / 工具调用次数 / 耗时」把差异量化出来。为了观察「可预测性」，我们跑两组规模不同的输入。

typescript
/**
 * 示例 6：工作流 vs 智能体（本章结论）
 *
 * 官方 workflows-agents 的核心论点：
 *   「能用固定工作流解决的，就不要上自主智能体。」
 *   工作流的路径是**写死**的，调用次数可预测、成本可控、容易测试；
 *   智能体的路径由模型**当场决定**，灵活但调用次数不可预测、成本与延迟会飘。
 *
 * 本示例用「同一个需求」分别实现两版：
 *   ① 固定工作流：抽取数据 → 计算增长率 → 写结论（3 个固定节点，模型调用次数恒定）
 *   ② 自主智能体：createReactAgent + calculator 工具，让模型自己决定怎么走
 * 并排打印两者的 模型调用次数 / 工具调用次数 / 耗时，并跑两组输入看「可预测性」。
 */
import {
  StateGraph,
  START,
  END,
  Annotation,
} from "@langchain/langgraph";
import { createReactAgent } from "@langchain/langgraph/prebuilt";
import { HumanMessage } from "@langchain/core/messages";
import { tool } from "@langchain/core/tools";
import * as z from "zod";
import { evaluate } from "mathjs";
import { createModel, banner } from "../../lib/model.js";
import { assertChatKey } from "../../lib/env.js";

// ── 计数器：统计「模型被调几次、工具被调几次、模型耗时」 ──────────
interface Counter {
  modelCalls: number;
  toolCalls: number;
  ms: number;
}
const freshCounter = (): Counter => ({ modelCalls: 0, toolCalls: 0, ms: 0 });

/** 给任意模型套一层「计数代理」：invoke / bindTools 都会被记账 */
function wrapModel<T extends object>(inner: T, c: Counter): T {
  return new Proxy(inner, {
    get(target, prop, receiver) {
      const orig = Reflect.get(target, prop, receiver);
      if (prop === "invoke" && typeof orig === "function") {
        return async (...args: unknown[]) => {
          c.modelCalls++;
          const t = Date.now();
          try {
            return await (orig as (...a: unknown[]) => Promise<unknown>).apply(
              target,
              args
            );
          } finally {
            c.ms += Date.now() - t;
          }
        };
      }
      if (prop === "bindTools" && typeof orig === "function") {
        return (...args: unknown[]) =>
          wrapModel((orig as (...a: unknown[]) => T).apply(target, args), c);
      }
      return typeof orig === "function"
        ? (orig as (...a: unknown[]) => unknown).bind(target)
        : orig;
    },
  });
}

// ══════════════════════════════════════════════════════════════
//  ① 固定工作流：3 个写死的节点
// ══════════════════════════════════════════════════════════════
const FlowState = Annotation.Root({
  input: Annotation<string>,
  data: Annotation<Record<string, number>>(),
  rates: Annotation<{ quarter: string; rate: string }[]>(),
  best: Annotation<string>(),
  conclusion: Annotation<string>(),
});

async function runWorkflow(
  model: ReturnType<typeof createModel>,
  text: string
) {
  const t0 = Date.now();

  async function extract(state: typeof FlowState.State) {
    console.log("   [workflow/extract] 抽取各季度数值（模型调用 1/2）");
    const res = await model.invoke([
      new HumanMessage(
        "从下面文本里抽取各季度数值，只输出 JSON，键是季度名、值是数字，不要任何其它文字。\n" +
          state.input
      ),
    ]);
    const raw = String(res.content);
    const jsonText = raw.slice(raw.indexOf("{"), raw.lastIndexOf("}") + 1);
    const data = JSON.parse(jsonText) as Record<string, number>;
    console.log(`      抽取结果：${JSON.stringify(data)}`);
    return { data };
  }

  function compute(state: typeof FlowState.State) {
    console.log("   [workflow/compute] 用纯 JS 算环比增长率（不调模型，完全确定性）");
    const entries = Object.entries(state.data);
    const rates: { quarter: string; rate: string }[] = [];
    for (let i = 1; i < entries.length; i++) {
      const [q, v] = entries[i];
      const prev = entries[i - 1][1];
      const rate = prev === 0 ? "N/A" : (((v - prev) / prev) * 100).toFixed(1) + "%";
      rates.push({ quarter: q, rate });
    }
    const best =
      rates
        .filter((r) => r.rate !== "N/A")
        .sort((a, b) => parseFloat(b.rate) - parseFloat(a.rate))[0]?.quarter ??
      "N/A";
    console.log(`      增长率：${JSON.stringify(rates)}，增长最快：${best}`);
    return { rates, best };
  }

  async function conclude(state: typeof FlowState.State) {
    console.log("   [workflow/conclude] 写一句结论（模型调用 2/2）");
    const res = await model.invoke([
      new HumanMessage(
        `已知事实（请原样使用，不要重算）：\n` +
          `各季度环比增长率：${JSON.stringify(state.rates)}\n` +
          `增长最快的是：${state.best}\n` +
          `请严格照抄下面这句话、只把方括号里的数值换成上面该季度的增长率，不要添加任何其它字：\n` +
          `增长最快的季度是 ${state.best}，环比增长率约[该季度增长率]。`
      ),
    ]);
    return { conclusion: String(res.content).trim() };
  }

  const graph = new StateGraph(FlowState)
    .addNode("extract", extract)
    .addNode("compute", compute)
    .addNode("conclude", conclude)
    .addEdge(START, "extract")
    .addEdge("extract", "compute")
    .addEdge("compute", "conclude")
    .addEdge("conclude", END)
    .compile();

  const result = await graph.invoke({ input: text });
  return { result, cost: Date.now() - t0 };
}

// ══════════════════════════════════════════════════════════════
//  ② 自主智能体：createReactAgent + calculator 工具
// ══════════════════════════════════════════════════════════════
async function runAgent(
  model: ReturnType<typeof createModel>,
  c: Counter,
  text: string
) {
  const calculator = tool(
    async ({ expression }) => {
      c.toolCalls++;
      console.log(`      🔧 [agent] calculator("${expression}")`);
      return String(evaluate(expression));
    },
    {
      name: "calculator",
      description: "执行数学计算。需要算数值或增长率时使用。",
      schema: z.object({
        expression: z.string().describe("数学表达式，例如 '(1560-1200)/1200*100'"),
      }),
    }
  );

  const agent = createReactAgent({
    llm: model,
    tools: [calculator],
    prompt:
      "你是数据分析助手。根据用户给出的季度数据，用 calculator 工具计算环比增长率，" +
      "找出增长最快的季度，并用一句话总结。",
  });

  const t0 = Date.now();
  const result = await agent.invoke({ messages: [new HumanMessage(text)] });
  const cost = Date.now() - t0;
  const last = result.messages[result.messages.length - 1];
  return { answer: String(last.content).trim(), cost, messages: result.messages };
}

async function main() {
  banner("示例 6：工作流 vs 智能体 —— 同一需求，两种实现");
  assertChatKey();

  const inputs = [
    "2023Q1 营收 1200 万，2023Q2 营收 1500 万，2023Q3 营收 1380 万，2023Q4 营收 1900 万。",
    "某产品 2024Q1 用户 8000，2024Q2 用户 11100，2024Q3 用户 9600，2024Q4 用户 16800，2025Q1 用户 15000，2025Q2 用户 21000。",
  ];

  const flowRows: { model: number; tool: number; ms: number }[] = [];
  const agentRows: { model: number; tool: number; ms: number }[] = [];

  for (let i = 0; i < inputs.length; i++) {
    const text = inputs[i];
    console.log(`\n${"═".repeat(72)}`);
    console.log(`👤 输入 #${i + 1}：${text}`);
    console.log("─".repeat(72));

    // —— ① 固定工作流 ——
    console.log("【① 固定工作流】");
    const fc = freshCounter();
    const flowModel = wrapModel(createModel({ temperature: 0 }), fc);
    const flow = await runWorkflow(flowModel, text);
    console.log(`   🤖 结论：${flow.result.conclusion}`);
    console.log(`   📊 模型调用 ${fc.modelCalls} 次，耗时 ${flow.cost}ms`);
    flowRows.push({ model: fc.modelCalls, tool: 0, ms: flow.cost });

    // —— ② 自主智能体 ——
    console.log("\n【② 自主智能体】");
    const ac = freshCounter();
    const agentModel = wrapModel(createModel({ temperature: 0 }), ac);
    const agent = await runAgent(agentModel, ac, text);
    console.log(`   🤖 结论：${agent.answer}`);
    console.log(
      `   📊 模型调用 ${ac.modelCalls} 次，工具调用 ${ac.toolCalls} 次，耗时 ${agent.cost}ms`
    );
    agentRows.push({ model: ac.modelCalls, tool: ac.toolCalls, ms: agent.cost });
  }

  const stat = (name: string, rows: { model: number; tool: number; ms: number }[]) => {
    console.log(`\n── ${name} ──`);
    console.log("   输入   模型调用  工具调用  总步数  耗时(ms)");
    rows.forEach((r, i) => {
      console.log(
        `   #${i + 1}     ${String(r.model).padEnd(9)}${String(r.tool).padEnd(10)}${String(
          r.model + r.tool
        ).padEnd(8)}${r.ms}`
      );
    });
    const steps = rows.map((r) => r.model + r.tool);
    const constant = steps.every((m) => m === steps[0]);
    console.log(
      `   → 总步数(模型+工具) ${
        constant ? "每次都相同（可预测 ✅）" : "随输入变化（不可预测 ⚠️）"
      }：${steps.join(", ")}`
    );
  };

  console.log("\n" + "═".repeat(72));
  console.log("📊 汇总对比");
  console.log("═".repeat(72));
  stat("固定工作流", flowRows);
  stat("自主智能体", agentRows);

  console.log("\n" + "─".repeat(72));
  console.log("🧭 选型决策树");
  console.log("─".repeat(72));
  console.log(
    [
      "需求能用「固定几步」说清吗？",
      "  ├─ 能 ──► 用工作流（提示链 / 路由 / 并行化 / 编排者-工作者）",
      "  │          特征：步骤已知、次数可预测、要低成本低延迟、要好测试 → 工作流",
      "  └─ 不能，必须让模型自己决定走几步、调哪些工具、循环几次 ──► 用智能体",
      "             特征：任务开放、路径依赖中间结果、愿接受不确定成本 → 智能体",
      "",
      "折中做法（最常用）：把智能体关进一个「工作流框架」里 ——",
      "外层用固定图控住主流程与预算，只在真正需要自主性的那一步放一个 ReAct 智能体。",
    ].join("\n")
  );
}

main().catch((e) => {
  console.error("❌ 调用失败：", e);
  process.exit(1);
});

运行：

bash
npx tsx 09-design-patterns/code/06-workflow-or-agent.ts

真实输出：


========================================================================
  示例 6：工作流 vs 智能体 —— 同一需求，两种实现
========================================================================


════════════════════════════════════════════════════════════════════════
👤 输入 #1：2023Q1 营收 1200 万，2023Q2 营收 1500 万，2023Q3 营收 1380 万，2023Q4 营收 1900 万。
────────────────────────────────────────────────────────────────────────
【① 固定工作流】
   [workflow/extract] 抽取各季度数值（模型调用 1/2）
      抽取结果：{"2023Q1":1200,"2023Q2":1500,"2023Q3":1380,"2023Q4":1900}
   [workflow/compute] 用纯 JS 算环比增长率（不调模型，完全确定性）
      增长率：[{"quarter":"2023Q2","rate":"25.0%"},{"quarter":"2023Q3","rate":"-8.0%"},{"quarter":"2023Q4","rate":"37.7%"}]，增长最快：2023Q4
   [workflow/conclude] 写一句结论（模型调用 2/2）
   🤖 结论：增长最快的季度是 2023Q4，环比增长率约37.7%。
   📊 模型调用 2 次，耗时 1989ms

【② 自主智能体】
      🔧 [agent] calculator("(1500-1200)/1200*100")
      🔧 [agent] calculator("(1380-1500)/1500*100")
      🔧 [agent] calculator("(1900-1380)/1380*100")
   🤖 结论：各季度环比增长率：2023Q2 为 +25.0%，2023Q3 为 -8.0%，2023Q4 为 +37.68%。

**总结：2023Q4 是增长最快的季度，营收环比增长约 37.7%（1380 万 → 1900 万），而 Q3 是唯一出现环比下滑的季度（-8.0%）。**
   📊 模型调用 2 次，工具调用 3 次，耗时 1782ms

════════════════════════════════════════════════════════════════════════
👤 输入 #2：某产品 2024Q1 用户 8000，2024Q2 用户 11100，2024Q3 用户 9600，2024Q4 用户 16800，2025Q1 用户 15000，2025Q2 用户 21000。
────────────────────────────────────────────────────────────────────────
【① 固定工作流】
   [workflow/extract] 抽取各季度数值（模型调用 1/2）
      抽取结果：{"2024Q1":8000,"2024Q2":11100,"2024Q3":9600,"2024Q4":16800,"2025Q1":15000,"2025Q2":21000}
   [workflow/compute] 用纯 JS 算环比增长率（不调模型，完全确定性）
      增长率：[{"quarter":"2024Q2","rate":"38.8%"},{"quarter":"2024Q3","rate":"-13.5%"},{"quarter":"2024Q4","rate":"75.0%"},{"quarter":"2025Q1","rate":"-10.7%"},{"quarter":"2025Q2","rate":"40.0%"}]，增长最快：2024Q4
   [workflow/conclude] 写一句结论（模型调用 2/2）
   🤖 结论：增长最快的季度是 2024Q4，环比增长率约75.0%。
   📊 模型调用 2 次，耗时 1734ms

【② 自主智能体】
      🔧 [agent] calculator("(11100-8000)/8000*100")
      🔧 [agent] calculator("(9600-11100)/11100*100")
      🔧 [agent] calculator("(16800-9600)/9600*100")
      🔧 [agent] calculator("(15000-16800)/16800*100")
      🔧 [agent] calculator("(21000-15000)/15000*100")
   🤖 结论：**各季度环比增长率**

| 季度 | 用户数 | 环比增长率 |
|---|---|---|
| 2024Q1 | 8000 | — |
| 2024Q2 | 11100 | +38.75% |
| 2024Q3 | 9600 | -13.51% |
| **2024Q4** | **16800** | **+75.00%（最快）** |
| 2025Q1 | 15000 | -10.71% |
| 2025Q2 | 21000 | +40.00% |

**一句话总结：** 用户规模整体呈上升趋势，其中 2024Q4 环比增长 75% 为增速最快季度，但增长波动明显，2024Q3 和 2025Q1 均出现负增长。
   📊 模型调用 2 次，工具调用 5 次，耗时 3111ms

════════════════════════════════════════════════════════════════════════
📊 汇总对比
════════════════════════════════════════════════════════════════════════

── 固定工作流 ──
   输入   模型调用  工具调用  总步数  耗时(ms)
   #1     2        0         2       1989
   #2     2        0         2       1734
   → 总步数(模型+工具) 每次都相同（可预测 ✅）：2, 2

── 自主智能体 ──
   输入   模型调用  工具调用  总步数  耗时(ms)
   #1     2        3         5       1782
   #2     2        5         7       3111
   → 总步数(模型+工具) 随输入变化（不可预测 ⚠️）：5, 7

────────────────────────────────────────────────────────────────────────
🧭 选型决策树
────────────────────────────────────────────────────────────────────────
需求能用「固定几步」说清吗？
  ├─ 能 ──► 用工作流（提示链 / 路由 / 并行化 / 编排者-工作者）
  │          特征：步骤已知、次数可预测、要低成本低延迟、要好测试 → 工作流
  └─ 不能，必须让模型自己决定走几步、调哪些工具、循环几次 ──► 用智能体
             特征：任务开放、路径依赖中间结果、愿接受不确定成本 → 智能体

折中做法（最常用）：把智能体关进一个「工作流框架」里 ——
外层用固定图控住主流程与预算，只在真正需要自主性的那一步放一个 ReAct 智能体。

这张汇总表就是本章的答案：

── 固定工作流 ──
   → 总步数 每次都相同（可预测 ✅）：2, 2

── 自主智能体 ──
   → 总步数 随输入变化（不可预测 ⚠️）：5, 7
  • 工作流：不管输入是大是小，永远是「1 次抽取 + 1 次结论 = 2 步」——恒定。你精确知道它要花多少钱、多久，测试也能写死预期。
  • 智能体：输入 #1（4 个季度）走了 5 步，输入 #2（6 个季度）走了 7 步——它自己决定算几次。数据一多，工具调用就跟着涨，成本和延迟一起飘。

注意：两版的最终结论都对（都找出了 2024Q4 / 2023Q4）。所以这不是「谁更聪明」的问题，而是**「谁更可控」**的问题。当你要上线一个要跑一整夜、被调十万次的批处理任务时，「可控」往往比「聪明」重要得多。

💡 怎么数调用次数？ 本示例给模型套了一层 Proxy（wrapModel），拦截 invoke 和 bindTools 两个方法自动记账。这样无论是工作流里手写的 model.invoke，还是 createReactAgent 内部偷偷调的模型，都会被算进去——所以这个数字是可信的，不是靠数代码行估算的。

🧭 选型决策树（背下来） ​

需求能用「固定几步」说清吗？
  ├─ 能 ──► 用工作流
  │        · 步骤已知、顺序固定      → 提示链
  │        · 按输入类型分流          → 路由
  │        · 子任务互不依赖、都想跑  → 并行化（分片 / 投票）
  │        · 子任务数量随任务变化    → 编排者-工作者
  │        · 有明确评价标准、要打磨  → 评估者-优化者
  │        判据：次数可预测、要低成本低延迟、要好测试
  │
  └─ 不能，必须让模型自己决定走几步、调哪些工具、循环几次 ──► 用智能体
           判据：任务开放、路径依赖中间结果、愿接受不确定成本

折中（生产环境最常用）：
把智能体「关进」一个工作流框架里 ——
外层用固定图控住主流程、预算与退出条件，
只在真正需要自主性的那一步，放一个 ReAct 智能体。

💡 最后一句忠告：很多团队一上来就想造「全自主智能体」，结果是成本失控、行为不可复现、线上无法排查。先问「这个任务能不能拆成固定几步」，能，就用工作流。 只有当你反过来证明了「固定步骤确实覆盖不了」，才引入智能体——而且优先用上面那个「折中」写法，把它限制在一小格里。

八、本章小结 ​

本章速查表

模式图长什么样什么时候用关键机制
提示链START → A → B → C → END任务能拆成有序步骤，每步都值得单独调模型一串固定 addEdge
路由START → classify → {分支…}输入类型多样、各类型处理方式不同addConditionalEdges
并行化dispatch → {w1,w2,w3} → collect子任务互不依赖、都想尽快跑fan-out/fan-in + reducer
编排者-工作者orchestrator →Send×N→ worker → synth子任务数量随任务动态变化Send + 条件边
评估者-优化者generate → evaluate ⇄ revise难一次写好、但有明确评价标准条件边回指上游 + 轮次上限
工作流 vs 智能体——选型路径写死 vs 模型当场决定

三句话总结

  1. 机制是「怎么用零件」，模式是「拼成什么形状」——本章的 5 种模式，就是用第 3 章那几个机制拼出来的 5 种固定形状。
  2. 工作流和智能体的分水岭是「路径谁决定」：写死 → 可预测、可控、好测；模型当场决定 → 灵活、但次数和成本会飘。默认先用工作流。
  3. 每个模式都有它的代价：提示链费调用、并行化要带 reducer、编排者-工作者要动态判数、评估者-优化者必须有轮次上限。收益和代价一起看，才叫会选。

动手练习

  1. 把示例 1 的提示链加一个「闸门（gate）」：在第 2 步后插入一个节点，检查初稿字数是否达标，不达标就中断并打印原因（用条件边指向不同的 END）。
  2. 把示例 2 的路由扩成四类，新增一个「退款申请」分支，并给它加一步专属逻辑（比如自动校验订单号格式）。
  3. 把示例 4 的 orchestrator 提示词改掉，让它「无论任务多简单都只拆 2 个、任务复杂就拆 6 个」，观察 Send 派发的数量怎么跟着变。
  4. 把示例 6 的智能体那版加上 recursionLimit（在 .invoke 时传 { recursionLimit: 4 }），看它在步数超限时是什么表现——体会「给智能体上保险」的感觉。

下一章预告

第 10 章我们跳出来看整棵大树：一个真实的 LangGraph 项目该怎么组织？图放哪、状态类型放哪、节点拆几个文件、工具和提示词怎么管理、怎么和外部服务（数据库、API）对接。前面九章都在讲「一块积木怎么搭」，第 10 章讲「一栋楼怎么盖」。

← 上一章：工程化与部署 ｜ 返回总目录 ｜ 下一章：应用结构 →

内容基于 LangGraph.js 官方文档与示例（langchain-ai/langgraphjs，MIT License）整理并重排为中文学习路径

本中文版仅供学习使用