第 3 章 · 节点、边与控制流

🎯 本章目标

• 用 条件边（addConditionalEdges） 让图学会「看情况走不同的路」，并彻底搞懂第三个参数（可能节点列表）到底有什么用
• 用 fan-out / fan-in 让一个节点同时触发多个节点，并用带 reducer 的字段把并发结果聚合回来
• 亲手写出一个 循环（loop），把条件边「指回上游节点」，实现自我改进
• 理解 recursionLimit，知道《GraphRecursionError》是怎么替你兜住死循环的
• 掌握 Send API，把列表里每一项动态 map 成并发任务，再 reduce 汇总
• 掌握 Command({ update, goto })，让一个节点「一步完成：更新状态 + 决定跳转」

本章代码

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

文件	内容
code/01-conditional-routing.ts	条件边：把一个模型分类结果分流到 3 个节点
code/02-parallel-fanout.ts	并行 fan-out / fan-in：用一个节点同时触发 3 个节点再汇总
code/03-loop-and-recursion-limit.ts	循环（自我改进）+ GraphRecursionError 防死循环
code/04-send-map-reduce.ts	Send API：把列表 map 成并发任务再 reduce
code/05-command-goto-and-update.ts	Command：一步完成「更新状态 + 决定跳转」

---

一、条件边：让流程学会「看情况」

第 1 章我们用的是固定边（addEdge）：A → B 是写死的，B 之后必去 C。

但真实业务里，下一步去哪往往得算出来。比如一个客服机器人，要先判断用户在问数学题、还是闲聊、还是在问知识，再决定交给哪个「专家」——这就是条件边。

                       ┌──► math  （数学题）
   START → classify ───┼──► chat  （闲聊）
          （模型分类）   └──► other （其他知识）

示例 1：用一个路由函数把问题分流到 3 个节点

/**
 * 示例 1：条件边 —— 用一个路由函数把问题分流到 3 个节点
 *
 * 这张图长这样：
 *
 *                       ┌──► math  （数学题）
 *   START → classify ───┼──► chat  （闲聊）
 *          （模型分类）   └──► other （其他知识）
 *
 * 关键点：classify 节点用「真实模型」判断问题属于哪一类，
 *        再用 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 RouteState = Annotation.Root({
  /** 用户的问题 */
  question: Annotation<string>,
  /** classify 节点给出的分类：math / chat / other */
  category: Annotation<string>,
  /** 最终回答 */
  answer: Annotation<string>,
});

type RouteNode = "math" | "chat" | "other";

async function main() {
  banner("示例 1：条件边（addConditionalEdges）把问题分流到 3 个节点");
  assertChatKey();

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

  // ── 节点 1：分类器 —— 让模型判断问题属于哪一类 ────────────────
  async function classify(state: typeof RouteState.State) {
    const res = await model.invoke([
      new SystemMessage(
        "你是一个分类器。判断用户的问题属于下面哪一类，只能回答一个词，" +
          "不要有任何其它文字：math（数学计算）、chat（闲聊寒暄）、other（其它知识问题）。"
      ),
      new HumanMessage(state.question),
    ]);
    const raw = String(res.content).toLowerCase();
    const category: RouteNode = raw.includes("math")
      ? "math"
      : raw.includes("chat")
        ? "chat"
        : "other";
    console.log(`   [节点 classify] 「${state.question}」→ 分类：${category}`);
    return { category };
  }

  // ── 三个下游节点：各司其职 ────────────────────────────────────
  async function mathNode(state: typeof RouteState.State) {
    const res = await model.invoke([
      new HumanMessage(`请给出简洁的计算过程和最终答案：${state.question}`),
    ]);
    return { answer: `【数学助手】${String(res.content).trim()}` };
  }

  async function chatNode(state: typeof RouteState.State) {
    const res = await model.invoke([new HumanMessage(state.question)]);
    return { answer: `【闲聊助手】${String(res.content).trim()}` };
  }

  async function otherNode(state: typeof RouteState.State) {
    const res = await model.invoke([new HumanMessage(state.question)]);
    return { answer: `【知识助手】${String(res.content).trim()}` };
  }

  // ── 路由函数：读状态，返回下一步要去的节点名 ──────────────────
  function route(state: typeof RouteState.State): RouteNode {
    const target: RouteNode =
      state.category === "math" || state.category === "chat"
        ? state.category
        : "other";
    console.log(`   [条件边 route] category="${state.category}" → 走向节点 "${target}"`);
    return target;
  }

  // ── 建图：把「可能节点列表」当作参数传进去，方便做对照实验 ──────
  function build(paths?: RouteNode[]) {
    const builder = new StateGraph(RouteState)
      .addNode("classify", classify)
      .addNode("math", mathNode)
      .addNode("chat", chatNode)
      .addNode("other", otherNode)
      .addEdge(START, "classify");
    const withConditional = paths
      ? builder.addConditionalEdges("classify", route, paths)
      : builder.addConditionalEdges("classify", route);
    return withConditional
      .addEdge("math", END)
      .addEdge("chat", END)
      .addEdge("other", END)
      .compile();
  }

  const graph = build(["math", "chat", "other"]); // ✅ 正确：列全所有可能分支
  const graphMissingList = build(); // ⚠️ 漏掉第三个参数
  const drawable = await graph.getGraphAsync();
  const drawableMissing = await graphMissingList.getGraphAsync();

  const fmt = (edges: { source: string; target: string }[]) =>
    edges.map((e) => `   • ${e.source} → ${e.target}`).join("\n");

  console.log("📌【带第三个参数 [\"math\",\"chat\",\"other\"]】图中的边：");
  console.log(fmt(drawable.edges));
  console.log("\n📌【漏掉第三个参数】图中的边（多出一条 classify → __end__ 幽灵边）：");
  console.log(fmt(drawableMissing.edges));

  // ── 第三个参数还兼做「白名单」校验：漏列了节点会直接编译失败 ────
  try {
    build(["math", "other"]); // ⚠️ 故意漏掉 chat
    console.log("\n❌ 意外：漏列 chat 竟然没报错");
  } catch (e) {
    const err = e as Error;
    console.log(`\n⚠️ 第三个参数漏列 chat 时，compile() 直接抛错：`);
    console.log(`   ${err.constructor.name}: ${err.message.split("\n")[0]}`);
  }

  // ── 真正跑 3 个问题，看它们被分流到不同节点 ────────────────────
  const questions = ["125 乘以 8 等于多少？", "今天心情不错，陪我聊两句吧", "珠穆朗玛峰有多高？"];
  for (const question of questions) {
    console.log(`\n👤 提问：${question}`);
    const result = await graph.invoke({ question });
    console.log(`🤖 ${result.answer}`);
  }
}

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

运行：

npx tsx 03-edges-and-control-flow/code/01-conditional-routing.ts

真实输出：

========================================================================
  示例 1：条件边（addConditionalEdges）把问题分流到 3 个节点
========================================================================

📌【带第三个参数 ["math","chat","other"]】图中的边：
   • __start__ → classify
   • chat → __end__
   • math → __end__
   • other → __end__
   • classify → math
   • classify → chat
   • classify → other

📌【漏掉第三个参数】图中的边（多出一条 classify → __end__ 幽灵边）：
   • __start__ → classify
   • chat → __end__
   • math → __end__
   • other → __end__
   • classify → math
   • classify → chat
   • classify → other
   • classify → __end__

⚠️ 第三个参数漏列 chat 时，compile() 直接抛错：
   UnreachableNodeError: Node `chat` is not reachable.

👤 提问：125 乘以 8 等于多少？
   [节点 classify] 「125 乘以 8 等于多少？」→ 分类：math
   [条件边 route] category="math" → 走向节点 "math"
🤖 【数学助手】125 × 8 = (100 + 25) × 8 = 800 + 200 = 1000

**最终答案：1000**

👤 提问：今天心情不错，陪我聊两句吧
   [节点 classify] 「今天心情不错，陪我聊两句吧」→ 分类：chat
   [条件边 route] category="chat" → 走向节点 "chat"
🤖 【闲聊助手】好啊，我也沾沾你的好心情 😄  
今天是因为发生了什么开心的事，还是就是莫名觉得状态不错？想聊聊的话，我听着呢。

👤 提问：珠穆朗玛峰有多高？
   [节点 classify] 「珠穆朗玛峰有多高？」→ 分类：other
   [条件边 route] category="other" → 走向节点 "other"
🤖 【知识助手】珠穆朗玛峰的最新官方高度是 **8848.86 米**（海拔高度，含峰顶雪盖）。  
这一数据由中国和尼泊尔于 **2020 年 12 月** 共同公布。

（提示：chat / other 两问的回答来自大模型，每次运行措辞会不同；math 一问因为是纯计算，基本稳定。）

三个必须理解的细节

① 条件边 = 节点 + 路由函数，两者分离

.addConditionalEdges("classify", route, ["math", "chat", "other"])
//                     ↑ 源节点    ↑ 路由函数  ↑ 可能走向的节点列表

route 是一个普通函数：读状态，返回一个节点名字符串（或 END）。它本身不干活，只负责「指路」。真正的活由被指向的节点干。

② 第三个参数（可能节点列表）到底有什么用？

上面那段真实输出就是答案，它有两个作用：

• 让图结构可以被确定：漏掉它时，框架不知道该条件边能去哪些节点，于是给图补了一条 classify → __end__ 的幽灵边——图的拓扑结构就不再准确了（可视化、校验都会受影响）。
• 当白名单做编译期校验：如果你 return 的节点名不在列表里，而它又没有别的入边，这个节点就是「不可达」的，compile() 会直接抛 UnreachableNodeError。上面 build(["math", "other"]) 漏掉 chat 时抛的正是这个错。

⚠️ 记住：第三个参数要列全所有可能的分支节点。它既让结构可读，又是防止你「路由函数写错名字」的第一道防线。

③ 路由函数的返回值必须是图上真实存在的节点名或 END

route 返回的类型我写成了 RouteNode = "math" | "chat" | "other"——用 TypeScript 把它卡死，写错名字编译期就会报错。这是比运行时崩溃更早的保险。

💡 和固定边的对比：addEdge 是「铁轨」，只能通向一条；addConditionalEdges 是「岔道口」，通向哪条由路由函数现场决定。

---

二、并行：一个节点，同时干三件事

有些任务之间没有依赖，串行做纯属浪费时间。比如「评估某个技术方案」，性能、成本、可维护性三个维度可以同时分析。

LangGraph 里，一个节点连出多条边就是 fan-out（扇出）；多条边汇入同一个节点就是 fan-in（扇入）。同一个「超步（super-step）」里的节点会并发执行。

                        ┌──► worker_perf  ──┐
   START → dispatch ────┼──► worker_cost  ──┼──► collect → END
                        └──► worker_maint ──┘
              （fan-out：同时连出 3 条边）   （fan-in：3 条边汇入 1 个节点）

示例 2：用「总耗时」证明它是真并行

串行执行 3 个各耗时 1 秒的节点 ≈ 3 秒；并行则 ≈ 1 秒。这个示例让每个 worker 都 sleep(1000)，打印开始 / 结束时间戳，再统计整图耗时。

/**
 * 示例 2：并行 fan-out / fan-in —— 一个节点同时触发 3 个节点，再汇总
 *
 * 图结构：
 *
 *                        ┌──► worker_perf  ──┐
 *   START → dispatch ────┼──► worker_cost  ──┼──► collect → END
 *                        └──► worker_maint ──┘
 *              （fan-out：同时连出 3 条边）   （fan-in：3 条边汇入 1 个节点）
 *
 * 证明「并行」的方法：让每个 worker 都 sleep 约 1 秒。
 *   - 如果是串行执行，总耗时 ≈ 3 秒
 *   - 如果是并行执行，总耗时 ≈ 1 秒
 * 于是我们给每个节点打印开始 / 结束时间戳，并统计整张图的耗时。
 */
import { StateGraph, START, END, Annotation } from "@langchain/langgraph";
import { banner } from "../../lib/model.js";

// ── 状态：注意 results 用「带 reducer 的字段」接收并发结果 ────────
const FanState = Annotation.Root({
  /** 要分析的主题 */
  topic: Annotation<string>,
  /**
   * 三个 worker 的产出。⚠️ 必须带 reducer：
   * 因为三个节点会在同一个「超步（super-step）」里各自返回一份 results，
   * 若没有 reducer，框架不知道该如何合并三份值，会直接报错。
   */
  results: Annotation<string[]>({
    reducer: (left, right) => left.concat(right),
    default: () => [],
  }),
  /** collect 节点汇总后的结论 */
  summary: Annotation<string>,
});

const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));

/** 打印带毫秒的时间戳，便于肉眼验证并行 */
function ts(): string {
  return new Date().toISOString().slice(11, 23); // HH:MM:SS.mmm
}

async function main() {
  banner("示例 2：并行 fan-out / fan-in（总耗时证明是并行）");
  const t0 = Date.now();

  // ── fan-out 的起点：分发节点 ──────────────────────────────────
  function dispatch(state: typeof FanState.State) {
    console.log(`   [节点 dispatch] 收到主题「${state.topic}」，同时派发 3 个分析任务 @ ${ts()}`);
    return {};
  }

  // ── 三个并行 worker：各自 sleep 1 秒 ──────────────────────────
  async function makeWorker(name: string, label: string) {
    return async function worker(_state: typeof FanState.State) {
      const start = Date.now();
      console.log(`   [节点 ${name}] 开始 @ ${ts()}`);
      await sleep(1000);
      const cost = Date.now() - start;
      console.log(`   [节点 ${name}] 结束 @ ${ts()}（耗时 ${cost}ms）`);
      return { results: [`${label}：${name} 用时 ${cost}ms`] };
    };
  }

  const workerPerf = await makeWorker("worker_perf", "【性能】");
  const workerCost = await makeWorker("worker_cost", "【成本】");
  const workerMaint = await makeWorker("worker_maint", "【可维护性】");

  // ── fan-in 的终点：汇总节点（要等 3 个 worker 全部完成才运行） ──
  function collect(state: typeof FanState.State) {
    console.log(`   [节点 collect] 收到 ${state.results.length} 份结果，开始汇总 @ ${ts()}`);
    const summary = state.results.join("\n   ");
    return { summary };
  }

  // ── 建图：一对多（fan-out）+ 多对一（fan-in） ─────────────────
  const graph = new StateGraph(FanState)
    .addNode("dispatch", dispatch)
    .addNode("worker_perf", workerPerf)
    .addNode("worker_cost", workerCost)
    .addNode("worker_maint", workerMaint)
    .addNode("collect", collect)
    .addEdge(START, "dispatch")
    // fan-out：dispatch 同时连出三条边
    .addEdge("dispatch", "worker_perf")
    .addEdge("dispatch", "worker_cost")
    .addEdge("dispatch", "worker_maint")
    // fan-in：三条边汇入同一个 collect 节点
    .addEdge("worker_perf", "collect")
    .addEdge("worker_cost", "collect")
    .addEdge("worker_maint", "collect")
    .addEdge("collect", END)
    .compile();

  const drawable = await graph.getGraphAsync();
  console.log("📌 图中的边：");
  for (const e of drawable.edges) {
    console.log(`   • ${e.source} → ${e.target}`);
  }
  console.log();

  const result = await graph.invoke({ topic: "要不要把框架换成 LangGraph" });

  const total = Date.now() - t0;
  console.log("\n📦 最终状态：");
  console.log(`   results = ${JSON.stringify(result.results, null, 2)}`);
  console.log(`\n🤖 汇总：\n   ${result.summary}`);
  console.log(`\n⏱️  整图耗时：${total}ms`);
  console.log(
    total < 2000
      ? "   ✅ 明显小于 3×1000ms → 三个 worker 是并行的！"
      : "   ⚠️ 接近 3000ms → 看起来是串行执行的。"
  );

  // ── 反面教材：如果 results 不带 reducer 会怎样？ ───────────────
  const BadState = Annotation.Root({ results: Annotation<string[]> }); // ❌ 无 reducer
  const badGraph = new StateGraph(BadState)
    .addNode("a", async () => ({ results: ["a"] }))
    .addNode("b", async () => ({ results: ["b"] }))
    .addNode("c", async () => ({ results: ["c"] }))
    .addEdge(START, "a")
    .addEdge(START, "b")
    .addEdge(START, "c")
    .addEdge("a", END)
    .addEdge("b", END)
    .addEdge("c", END)
    .compile();
  try {
    await badGraph.invoke({ results: [] });
    console.log("\n❌ 意外：无 reducer 竟然没报错");
  } catch (e) {
    const err = e as Error;
    console.log("\n⚠️ 反面教材：results 不带 reducer 时，同一个超步写入 3 个值 → 报错：");
    console.log(`   ${err.constructor.name}: ${err.message.split("\n")[0]}`);
  }
}

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

运行：

npx tsx 03-edges-and-control-flow/code/02-parallel-fanout.ts

真实输出：

========================================================================
  示例 2：并行 fan-out / fan-in（总耗时证明是并行）
========================================================================

📌 图中的边：
   • __start__ → dispatch
   • collect → __end__
   • dispatch → worker_perf
   • dispatch → worker_cost
   • dispatch → worker_maint
   • worker_cost → collect
   • worker_maint → collect
   • worker_perf → collect

   [节点 dispatch] 收到主题「要不要把框架换成 LangGraph」，同时派发 3 个分析任务 @ 10:58:23.836
   [节点 worker_cost] 开始 @ 10:58:23.838
   [节点 worker_maint] 开始 @ 10:58:23.838
   [节点 worker_perf] 开始 @ 10:58:23.838
   [节点 worker_cost] 结束 @ 10:58:24.839（耗时 1001ms）
   [节点 worker_maint] 结束 @ 10:58:24.839（耗时 1001ms）
   [节点 worker_perf] 结束 @ 10:58:24.839（耗时 1001ms）
   [节点 collect] 收到 3 份结果，开始汇总 @ 10:58:24.840

📦 最终状态：
   results = [
  "【成本】：worker_cost 用时 1001ms",
  "【可维护性】：worker_maint 用时 1001ms",
  "【性能】：worker_perf 用时 1001ms"
]

🤖 汇总：
   【成本】：worker_cost 用时 1001ms
   【可维护性】：worker_maint 用时 1001ms
   【性能】：worker_perf 用时 1001ms

⏱️  整图耗时：1016ms
   ✅ 明显小于 3×1000ms → 三个 worker 是并行的！

⚠️ 反面教材：results 不带 reducer 时，同一个超步写入 3 个值 → 报错：
   InvalidUpdateError: Invalid update for channel "results" with values [["a"],["b"],["c"]]: LastValue can only receive one value per step.

看时间戳：三个 开始 的时间几乎一样（23.838），三个 结束 也几乎一样（24.839），整图只花了 1016ms。如果它们是串行，应该是「开始 → 等 1 秒 → 结束 → 下一个开始……」，总耗时约 3000ms。

三个必须理解的细节

① fan-out 就是「一个节点多条出边」

.addEdge("dispatch", "worker_perf")
.addEdge("dispatch", "worker_cost")
.addEdge("dispatch", "worker_maint")

三条边都从 dispatch 出发，框架就把这三个 worker 放进同一个超步并发跑。

② fan-in 会自动「等齐」，不需要你自己 join

三条边都指向 collect。框架会等三个 worker 全部结束，才运行一次 collect（这就是为什么它只打印了一行「收到 3 份结果」）。这个「等齐」的语义叫屏障（barrier），是 LangGraph 内置的，你什么都不用写。

③ 并发写入的字段「必须」带 reducer —— 这是硬性要求

results 被三个 worker 在同一个超步里各写了一次。字段没有 reducer 时，LangGraph 默认认为「一个超步只能写一个值」，于是直接抛错：

InvalidUpdateError: ... LastValue can only receive one value per step.

上面的反面教材就是这个错误的真实复现。加了 reducer: (left, right) => left.concat(right) 之后，框架知道该把 N 份结果拼起来，问题就解决了。

💡 一句话：只要一个字段可能被多个并发节点同时写，它就必须带 reducer。 这是 fan-in 场景里最容易踩的坑。

⚠️ 别把 reducer 和「顺序」混为一谈：并发的执行顺序是不确定的（本例三次运行里 results 的顺序都可能不同），所以 reducer 做的应该是可交换的合并（比如 concat、+、取最大），而不是「谁先来谁优先」这种有顺序依赖的逻辑。

---

三、循环：让图「回头再来一次」

第 1 章的 ReAct 智能体里已经出现过循环（tools → agent）。这一节我们把循环单独拎出来讲透，并回答一个所有人都会问的问题：万一循环停不下来怎么办？

条件边不仅可以指向「下游」，还可以指回上游节点——这就是循环的来源。

示例 3（上半场）：自我改进循环

任务：让模型写一段介绍，打不到分就回去重写，直到达标或达到最大轮数。

                 ┌──────────────┐
   START → write_draft → grade │
                 └──────┬───────┘
                        │ 未达标：指回上游
                        └──────► write_draft（循环）

/**
 * 示例 3：循环（loop）与防死循环（recursionLimit）
 *
 * 上半场 —— 自我改进循环：
 *
 *                 ┌─────────────┐
 *   START → draft │ 写草稿 → 打分 │
 *                 └──────┬──────┘
 *                        │ 不达标就「指回上游」
 *              score ◄───┘   （addConditionalEdges 把边连回 draft）
 *
 * 下半场 —— GraphRecursionError：
 *   故意造一个永远循环的图，用很小的 recursionLimit 跑，捕获错误，
 *   讲清楚 recursionLimit 参数是什么、以及它怎么兜住死循环。
 */
import { StateGraph, START, END, Annotation, GraphRecursionError } from "@langchain/langgraph";
import { HumanMessage } from "@langchain/core/messages";
import { createModel, banner } from "../../lib/model.js";
import { assertChatKey } from "../../lib/env.js";

// ── 上半场：自我改进循环的状态 ─────────────────────────────────
const LoopState = Annotation.Root({
  /** 要写作的主题 */
  topic: Annotation<string>,
  /** 当前草稿 */
  draft: Annotation<string>,
  /** 当前得分（0~2） */
  score: Annotation<number>,
  /** 已经重写了多少轮 */
  attempts: Annotation<number>,
});

const PASS_SCORE = 2; // 满分 2，达到 2 才收工
const MAX_ATTEMPTS = 3; // 兜底：最多重写 3 轮，防止无限循环

async function main() {
  banner("示例 3：循环（自我改进）+ GraphRecursionError（防死循环）");
  assertChatKey();

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

  // ── 节点 A：写草稿。第 1 轮凭感觉写，之后带着「反馈」改进 ────────
  async function writeDraft(state: typeof LoopState.State) {
    const round = state.attempts + 1;
    const prompt =
      round === 1
        ? `用一句话介绍「${state.topic}」。`
        : `这是你上一轮对「${state.topic}」的介绍：\n${state.draft}\n\n` +
          `它还不够好。请改进：必须包含一个具体例子（用「例如」引出），并且正文不少于 50 个字。`;
    const res = await model.invoke([new HumanMessage(prompt)]);
    const text = String(res.content).trim();
    console.log(`   [节点 write_draft] 第 ${round} 轮草稿（${text.length} 字）`);
    return { draft: text, attempts: round };
  }

  // ── 节点 B：打分（这里用一个「固定的评分规则」保证可复现） ──────
  function grade(state: typeof LoopState.State) {
    const hasExample = /例如|比如|举例/.test(state.draft);
    const longEnough = state.draft.length >= 50;
    const s = (hasExample ? 1 : 0) + (longEnough ? 1 : 0);
    console.log(
      `   [节点 grade] 得分 ${s}/${PASS_SCORE}（含例子：${hasExample}；≥50字：${longEnough}）`
    );
    return { score: s };
  }

  // ── 条件边：达标 / 到轮数上限就收工，否则「指回上游」write_draft ──
  function shouldContinue(state: typeof LoopState.State): "write_draft" | typeof END {
    if (state.score >= PASS_SCORE) {
      console.log("   [条件边] 已达标 → END");
      return END;
    }
    if (state.attempts >= MAX_ATTEMPTS) {
      console.log("   [条件边] 已达最大轮数 → END");
      return END;
    }
    console.log("   [条件边] 未达标 → 回到上游节点 \"write_draft\" 重写");
    return "write_draft";
  }

  const graph = new StateGraph(LoopState)
    .addNode("write_draft", writeDraft)
    .addNode("grade", grade)
    .addEdge(START, "write_draft")
    .addEdge("write_draft", "grade")
    // ⬇️ 这条条件边可以指回上游的 "write_draft"，循环就是这么来的
    .addConditionalEdges("grade", shouldContinue, ["write_draft", END])
    .compile();

  console.log(`👤 任务：写一句关于「什么是向量数据库」的介绍，要求含例子且 ≥50 字\n`);
  const result = await graph.invoke({ topic: "什么是向量数据库", attempts: 0, score: 0 });
  console.log(`\n✅ 最终草稿（用了 ${result.attempts} 轮，得分 ${result.score}）：`);
  console.log(`   ${result.draft}`);

  // ── 下半场：故意造一个永远循环的图，演示 recursionLimit ──────────
  console.log("\n" + "-".repeat(72));
  console.log("下半场：GraphRecursionError —— 死循环是怎么被拦下来的");
  console.log("-".repeat(72) + "\n");

  const TickState = Annotation.Root({ n: Annotation<number> });
  const foreverGraph = new StateGraph(TickState)
    .addNode("tick", (state: typeof TickState.State) => {
      const next = state.n + 1;
      console.log(`   [节点 tick] 第 ${next} 次执行`);
      return { n: next };
    })
    .addEdge(START, "tick")
    // 无条件地指回自己 → 死循环
    .addConditionalEdges("tick", () => "tick", ["tick"])
    .compile();

  try {
    await foreverGraph.invoke({ n: 0 }, { recursionLimit: 4 });
    console.log("❌ 意外：居然没有抛错");
  } catch (e) {
    if (e instanceof GraphRecursionError) {
      console.log(`\n⚠️ 捕获到 GraphRecursionError（recursionLimit=4）：`);
      console.log(`   ${e.message.split("\n")[0]}`);
      console.log("   → 图并没有真的死掉，框架在第 4 步就把运行叫停了。");
    } else {
      throw e;
    }
  }
}

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

运行：

npx tsx 03-edges-and-control-flow/code/03-loop-and-recursion-limit.ts

真实输出：

========================================================================
  示例 3：循环（自我改进）+ GraphRecursionError（防死循环）
========================================================================

👤 任务：写一句关于「什么是向量数据库」的介绍，要求含例子且 ≥50 字

   [节点 write_draft] 第 1 轮草稿（73 字）
   [节点 grade] 得分 1/2（含例子：false；≥50字：true）
   [条件边] 未达标 → 回到上游节点 "write_draft" 重写
   [节点 write_draft] 第 2 轮草稿（198 字）
   [节点 grade] 得分 2/2（含例子：true；≥50字：true）
   [条件边] 已达标 → END

✅ 最终草稿（用了 2 轮，得分 2）：
   向量数据库是专门存储、索引和检索高维向量的数据库。它把文本、图片、音频等数据经嵌入模型转换成向量，再通过余弦相似度或欧氏距离等度量，快速找到与查询最接近的结果，常用于语义搜索、推荐系统和 RAG。例如，把“苹果富含维生素”和“iPhone 15 发布”分别转成向量存入库中，当用户搜索“吃苹果有什么好处”时，即使查询里没有“维生素”这个词，系统也能因语义相近而优先返回“苹果富含维生素”那段内容。

------------------------------------------------------------------------
下半场：GraphRecursionError —— 死循环是怎么被拦下来的
------------------------------------------------------------------------

   [节点 tick] 第 1 次执行
   [节点 tick] 第 2 次执行
   [节点 tick] 第 3 次执行
   [节点 tick] 第 4 次执行

⚠️ 捕获到 GraphRecursionError（recursionLimit=4）：
   Recursion limit of 4 reached without hitting a stop condition. You can increase the limit by setting the "recursionLimit" config key.
   → 图并没有真的死掉，框架在第 4 步就把运行叫停了。

三个必须理解的细节

① 循环 = 条件边指向了一个「已经走过」的节点

.addConditionalEdges("grade", shouldContinue, ["write_draft", END])

shouldContinue 既可以是 END（收工），也可以是 "write_draft"（回头重写）。框架不介意「往回走」——边的方向是自由的，这才是图比链强大的地方。

② 循环必须要有「刹车」，否则必然死循环

我们这个循环有两个出口：score >= PASS_SCORE（达标）或 attempts >= MAX_ATTEMPTS（到轮数上限）。只有「不达标」才回去。实际输出里，第 1 轮得分 1（没含例子），于是回到 write_draft 重写；第 2 轮得分 2，达标收工。

如果你写循环时忘了设出口，别慌——LangGraph 有个保险。

③ recursionLimit：框架自带的「死循环刹车」

invoke 的第二个参数可以传 { recursionLimit: N }（默认是 25）。它限制的是超步（super-step）的次数：一旦某次运行超过 N 步还没停，就抛 GraphRecursionError。

下半场那个 foreverGraph 就是「无条件指回自己」的死循环。我们故意把 recursionLimit 设成 4，于是它跑了 4 次 tick 就被拦下：

Recursion limit of 4 reached without hitting a stop condition. ...

真实业务里如果收到这个错误，通常意味着：要么某个循环的退出条件写错了（永远不满足），要么确实需要更多步、应该调大 recursionLimit。

⚠️ 两个「值」的区别：recursionLimit 数的是超步次数（一个超步里可能并行跑好几个节点），不是「调用了几次 invoke」。所以并行图会比串行图更快触到上限。

💡 防死循环的正确姿势：① 循环一定要有明确的退出条件；② 加一个「最大轮数」兜底；③ 用 recursionLimit 作为最后一道保险。三层都做，才叫稳。

---

四、Send API：动态 map-reduce

上一节的 fan-out 是静态的——你有几个 worker，边就写几条。但如果运行前才知道要处理几项呢？比如用户给了一串主题「向量数据库 / Prompt 工程 / RAG」，要在图上对每一项都跑一遍同样的处理逻辑。

这时就该 Send 出场了。它让条件边函数返回一个 Send 数组，数组里每一项都是一次独立的节点调用。这就是地图-归约（map-reduce）：

• map：new Send("write_intro", { topic }) × N → 同一个节点被并发执行 N 次
• reduce：所有 map 的结果汇到一个节点里做汇总

   START ──(fanout：对每个主题 new Send("write_intro", {topic}))
                     │
         ┌───────────┼───────────┐
         ▼           ▼           ▼
   write_intro  write_intro  write_intro   ← 同一个节点被并发执行 3 次
         └───────────┼───────────┘
                     ▼
                  reduce ──► END          ← 等所有 map 完成后汇总

示例 4：对 3 个主题各写一句介绍，再汇总

/**
 * 示例 4：Send API —— map-reduce（把每一项 map 成并发任务，再 reduce 汇总）
 *
 * 图结构：
 *
 *   START ──(fanout：对每个主题 new Send("write_intro", {topic}))
 *                     │
 *         ┌───────────┼───────────┐
 *         ▼           ▼           ▼
 *   write_intro  write_intro  write_intro   ← 同一个节点被并发执行 3 次
 *         └───────────┼───────────┘
 *                     ▼
 *                  reduce ──► END          ← 等所有 map 完成后汇总
 *
 * 关键点：
 *   1. Send 由「条件边函数」返回一个数组 → 数组长度 = 并发任务数。
 *   2. 每个 Send 的第二个参数会作为「这个任务的初始状态」传给目标节点。
 *   3. 多条并发结果必须落到「带 reducer 的字段」上，否则同一个超步写多次会报错。
 */
import { StateGraph, START, END, Annotation, Send } from "@langchain/langgraph";
import { HumanMessage } from "@langchain/core/messages";
import { createModel, banner } from "../../lib/model.js";
import { assertChatKey } from "../../lib/env.js";

// ── 状态 ──────────────────────────────────────────────────────
const MapReduceState = Annotation.Root({
  /** 待处理的多个主题（reduce 的「输入清单」） */
  topics: Annotation<string[]>,
  /** Send 注入给单个任务的字段（一次只装一个主题） */
  topic: Annotation<string>,
  /**
   * map 阶段所有任务的产出。⚠️ 必须带 reducer：
   * N 个 write_intro 会在同一个超步里各自返回一份 intros，
   * reducer 负责把它们拼成一个数组。
   */
  intros: Annotation<string[]>({
    reducer: (left, right) => left.concat(right),
    default: () => [],
  }),
  /** reduce 阶段的总结 */
  summary: Annotation<string>,
});

async function main() {
  banner("示例 4：Send API —— map-reduce（并发处理多个主题再汇总）");
  assertChatKey();

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

  // ── fan-out 函数：把 topics 列表「映射」成一堆并发任务 ─────────
  function fanout(state: typeof MapReduceState.State) {
    console.log(`   [fanout] 收到 ${state.topics.length} 个主题，生成 ${state.topics.length} 个 Send 任务`);
    return state.topics.map((t) => new Send("write_intro", { topic: t }));
  }

  // ── map 节点：被每个 Send 任务各执行一次，只看到自己的 topic ────
  async function writeIntro(state: typeof MapReduceState.State) {
    const res = await model.invoke([
      new HumanMessage(`用一句话（不超过 40 字）介绍「${state.topic}」，直接给介绍，不要客套。`),
    ]);
    const text = String(res.content).trim().replace(/\n+/g, " ");
    console.log(`   [map · write_intro] 完成「${state.topic}」`);
    return { intros: [`${state.topic}：${text}`] };
  }

  // ── reduce 节点：等所有 map 完成后运行一次，做汇总 ─────────────
  async function reduce(state: typeof MapReduceState.State) {
    console.log(`   [reduce] 收到 ${state.intros.length} 条介绍，开始汇总`);
    const joined = state.intros.map((s) => `- ${s}`).join("\n");
    const res = await model.invoke([
      new HumanMessage(
        `下面是几段产品介绍：\n${joined}\n\n请用一段话（80 字以内）把它们总结成对这批技术的整体评价。`
      ),
    ]);
    return { summary: String(res.content).trim() };
  }

  // ── 建图：START 的「条件边」返回 Send 数组即可实现动态 fan-out ───
  const graph = new StateGraph(MapReduceState)
    .addNode("write_intro", writeIntro)
    .addNode("reduce", reduce)
    // 从 START 出发用条件边做 fan-out；第三个参数要列出可能的落点
    .addConditionalEdges(START, fanout, ["write_intro"])
    .addEdge("write_intro", "reduce")
    .addEdge("reduce", END)
    .compile();

  const drawable = await graph.getGraphAsync();
  console.log("📌 图中的边（Send 的动态边不会出现在这里，这正是它与固定边的区别）：");
  for (const e of drawable.edges) {
    console.log(`   • ${e.source} → ${e.target}`);
  }
  console.log();

  const topics = ["向量数据库", "Prompt 工程", "RAG"];
  console.log(`👤 输入主题：${JSON.stringify(topics)}\n`);

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

  console.log("\n📦 map 阶段产出：");
  for (const intro of result.intros) {
    console.log(`   • ${intro}`);
  }
  console.log(`\n🤖 reduce 汇总：\n   ${result.summary}`);
}

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

运行：

npx tsx 03-edges-and-control-flow/code/04-send-map-reduce.ts

真实输出：

========================================================================
  示例 4：Send API —— map-reduce（并发处理多个主题再汇总）
========================================================================

📌 图中的边（Send 的动态边不会出现在这里，这正是它与固定边的区别）：
   • reduce → __end__
   • write_intro → reduce
   • __start__ → write_intro

👤 输入主题：["向量数据库","Prompt 工程","RAG"]

   [fanout] 收到 3 个主题，生成 3 个 Send 任务
   [map · write_intro] 完成「向量数据库」
   [map · write_intro] 完成「Prompt 工程」
   [map · write_intro] 完成「RAG」
   [reduce] 收到 3 条介绍，开始汇总

📦 map 阶段产出：
   • 向量数据库：向量数据库是存储向量嵌入并支持高效相似性检索的数据库。
   • Prompt 工程：Prompt 工程是通过设计优化输入提示，引导大模型输出准确可控结果的技术。
   • RAG：RAG即检索增强生成，先检索外部知识，再让大模型据此生成更准确答案。

🤖 reduce 汇总：
   这批技术围绕大模型落地互补协同：向量数据库管理知识检索，Prompt工程优化输入，RAG打通检索与生成，共同提升答案准确性和可控性。

（提示：intros 的顺序与 map 任务的完成顺序有关，每次运行可能不同；每段介绍的具体措辞也由模型生成，会变化。）

三个必须理解的细节

① new Send(node, args) 的两个参数

state.topics.map((t) => new Send("write_intro", { topic: t }))
//                          ↑ 目标节点        ↑ 注入给这次调用的状态

• 第一个参数：要执行哪个节点。
• 第二个参数：这次调用的初始状态。注意它可以是个「局部状态」——这里只塞了 { topic: t }，而不是整张图的完整状态。

② Send 数组的长度 = 并发任务数

fanout 返回一个长度为 3 的数组，框架就并发起 3 个 write_intro 任务。这就是「动态」：列表里几项，就起几个任务，不需要改图的结构。所以输出里 write_intro 被打印了 3 次（顺序不定）。

③ 并发结果必须落到带 reducer 的字段

和示例 2 同理：3 个 write_intro 在同一个超步里各返回一份 intros，所以 intros 必须带 reducer（concat）把 3 份拼成一份。而 reduce 节点会等 3 个 map 全部完成后才运行一次，它读到的 state.intros 就是拼好的 3 条。

💡 看图上为什么「没有」Send 的边：getGraphAsync() 打印出的边里只有一个 __start__ → write_intro，因为 Send 是在运行时（条件边函数被调用时）才生成的动态边，静态图结构里看不见它。这正好体现了 Send 与固定边的本质区别。

⚠️ 别把 Send 和「并行边」搞混：并行边（示例 2）是画死的、数量固定；Send 是算出来的、数量随数据变化。

---

五、Command：一步完成「更新状态 + 跳转」

到目前为止，我们的节点只做一件事——返回状态更新；而「下一步去哪」交给另一个路由函数（条件边）决定。这需要两处代码：

节点函数：  return { history: [...] }        // 只更新状态
条件边：    function route(state) { ... }     // 另外决定去向

Command 把这两件事合并到节点自己的返回值里：

return new Command({
  update: { history: [...] },  // ① 更新状态
  goto: "even",                // ② 决定跳转
});

特别适合「节点做完决定后立刻知道该去哪」的场景——比如分类、路由、任务分派。

示例 5：同一个任务，两种写法对照

/**
 * 示例 5：Command —— 节点「一步完成：更新状态 + 决定跳转」
 *
 * 传统写法（返回状态对象 + 条件边）需要两步：
 *   ① 节点 return { ... } 更新状态
 *   ② 另一条 addConditionalEdges 里的路由函数再决定去哪
 *
 * Command 写法把这两件事合并进节点的返回值：
 *   return new Command({ update: { ... }, goto: "下一个节点" })
 *
 * 本示例用同一个「奇偶分流」任务，把两种写法各实现一遍并跑同样的输入，
 * 让你直观看到它们结果等价、但 Command 更紧凑。
 *
 * 说明：Command.PARENT 用于「子图里向父图发指令」，属于多智能体场景，
 *      这里只提及，第 7 章再展开。
 */
import { StateGraph, START, END, Annotation, Command } from "@langchain/langgraph";
import { banner } from "../../lib/model.js";

// ── 状态 ──────────────────────────────────────────────────────
const CmdState = Annotation.Root({
  /** 输入的数字 */
  n: Annotation<number>,
  /** 处理轨迹（带 reducer，可多次追加） */
  history: Annotation<string[]>({
    reducer: (left, right) => left.concat(right),
    default: () => [],
  }),
  /** 最终结论 */
  result: Annotation<string>,
});

type Branch = "even" | "odd";

async function main() {
  banner("示例 5：Command —— 一个节点同时「更新状态 + 决定跳转」");

  // ══════════════════════════════════════════════════════════════
  // 写法 A：Command（更新状态 + 跳转，一步到位）
  // ══════════════════════════════════════════════════════════════

  // 分流节点：既写 history，又用 goto 决定去 even 还是 odd
  function triageWithCommand(state: typeof CmdState.State) {
    const branch: Branch = state.n % 2 === 0 ? "even" : "odd";
    console.log(`   [节点 triage] n=${state.n} → goto "${branch}"（同时写入 history）`);
    return new Command({
      // ① 状态更新：等价于节点 return { ... }
      update: { history: [`triage(判断 ${state.n})`] },
      // ② 跳转：等价于一条条件边
      goto: branch,
    });
  }

  function evenHandler(state: typeof CmdState.State) {
    const n = state.n;
    console.log(`   [节点 even] 收到 n=${n}`);
    return { history: [`even(处理 ${n})`], result: `${n} 是偶数，且 ${n} / 2 = ${n / 2}` };
  }

  function oddHandler(state: typeof CmdState.State) {
    const n = state.n;
    console.log(`   [节点 odd] 收到 n=${n}`);
    return { history: [`odd(处理 ${n})`], result: `${n} 是奇数，且 ${n} * 3 + 1 = ${n * 3 + 1}` };
  }

  const graphA = new StateGraph(CmdState)
    // ⚠️ 用 Command.goto 时，必须用 ends 告诉框架「这个节点可能去哪些节点」
    .addNode("triage", triageWithCommand, { ends: ["even", "odd"] })
    .addNode("even", evenHandler)
    .addNode("odd", oddHandler)
    .addEdge(START, "triage")
    .addEdge("even", END)
    .addEdge("odd", END)
    .compile();

  // ══════════════════════════════════════════════════════════════
  // 写法 B：返回状态对象 + 条件边（传统两步法）
  // ══════════════════════════════════════════════════════════════

  function triageWithState(state: typeof CmdState.State) {
    console.log(`   [节点 triage_state] n=${state.n}（只更新 history，不决定去向）`);
    return { history: [`triage(判断 ${state.n})`] };
  }

  function routeBranch(state: typeof CmdState.State): Branch {
    const branch: Branch = state.n % 2 === 0 ? "even" : "odd";
    console.log(`   [条件边 route] n=${state.n} → 走向 "${branch}"`);
    return branch;
  }

  const graphB = new StateGraph(CmdState)
    .addNode("triage", triageWithState)
    .addNode("even", evenHandler)
    .addNode("odd", oddHandler)
    .addEdge(START, "triage")
    .addConditionalEdges("triage", routeBranch, ["even", "odd"])
    .addEdge("even", END)
    .addEdge("odd", END)
    .compile();

  // ══════════════════════════════════════════════════════════════
  // 跑同样的输入，对比两种写法
  // ══════════════════════════════════════════════════════════════
  for (const n of [7, 8]) {
    console.log(`\n👤 输入 n=${n}`);

    console.log("── 写法 A：Command ──");
    const resA = await graphA.invoke({ n });
    console.log(`   history = ${JSON.stringify(resA.history)}`);
    console.log(`   result  = ${resA.result}`);

    console.log("── 写法 B：状态对象 + 条件边 ──");
    const resB = await graphB.invoke({ n });
    console.log(`   history = ${JSON.stringify(resB.history)}`);
    console.log(`   result  = ${resB.result}`);

    const same =
      resA.result === resB.result && JSON.stringify(resA.history) === JSON.stringify(resB.history);
    console.log(`   ✅ 两种写法结果一致：${same}`);
  }
}

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

运行：

npx tsx 03-edges-and-control-flow/code/05-command-goto-and-update.ts

真实输出：

========================================================================
  示例 5：Command —— 一个节点同时「更新状态 + 决定跳转」
========================================================================


👤 输入 n=7
── 写法 A：Command ──
   [节点 triage] n=7 → goto "odd"（同时写入 history）
   [节点 odd] 收到 n=7
   history = ["triage(判断 7)","odd(处理 7)"]
   result  = 7 是奇数，且 7 * 3 + 1 = 22
── 写法 B：状态对象 + 条件边 ──
   [节点 triage_state] n=7（只更新 history，不决定去向）
   [条件边 route] n=7 → 走向 "odd"
   [节点 odd] 收到 n=7
   history = ["triage(判断 7)","odd(处理 7)"]
   result  = 7 是奇数，且 7 * 3 + 1 = 22
   ✅ 两种写法结果一致：true

👤 输入 n=8
── 写法 A：Command ──
   [节点 triage] n=8 → goto "even"（同时写入 history）
   [节点 even] 收到 n=8
   history = ["triage(判断 8)","even(处理 8)"]
   result  = 8 是偶数，且 8 / 2 = 4
── 写法 B：状态对象 + 条件边 ──
   [节点 triage_state] n=8（只更新 history，不决定去向）
   [条件边 route] n=8 → 走向 "even"
   [节点 even] 收到 n=8
   history = ["triage(判断 8)","even(处理 8)"]
   result  = 8 是偶数，且 8 / 2 = 4
   ✅ 两种写法结果一致：true

两种写法结果完全一致（true）。区别只在代码组织：

	写法 A：Command	写法 B：状态对象 + 条件边
更新状态	节点返回 new Command({ update })	节点返回 { ... }
决定去向	同一个 Command 的 goto	另一处 addConditionalEdges 的路由函数
代码位置	集中在一处	分散在两处
适合场景	节点天然知道下一步去哪（分类/路由）	路由逻辑复杂、想单独复用/测试

三个必须理解的细节

① new Command({ update, goto }) 的两个字段各管一半

return new Command({
  update: { history: [...] },  // 等价于原来的 return { ... }
  goto: "even",                // 等价于条件边返回的节点名
});

update 部分和普通节点返回值完全等价，照样走 reducer 合并；goto 部分和条件边路由函数的返回值完全等价。

② 用 Command.goto 时，必须在 addNode 里声明 ends

.addNode("triage", triageWithCommand, { ends: ["even", "odd"] })

原因和示例 1 的「第三个参数」是一回事：框架需要在编译期知道这个节点可能去哪些地方，否则 even / odd 这两个节点就是「不可达」的，compile() 会抛 UnreachableNodeError。ends 就是 Command 版本的「可能走向列表」。

③ goto 还能接 Send，甚至一串节点

Command 的 goto 参数相当灵活：

goto: "even"                              // 去一个节点
goto: ["even", "odd"]                     // 同时去多个（并行）
goto: new Send("write_intro", { topic })  // 结合 Send（见示例 4）

💡 该用哪个？ 判据很简单：「更新状态」和「决定去向」是不是同一件事？

• 是同一件事（比如「算完分类，立刻按分类走」）→ 用 Command，代码更内聚。
• 是两件事（比如「节点只管干活，路由规则以后可能改」）→ 用返回状态 + 条件边，职责更清晰。

⚠️ 提及但本章不展开：Command 还有个 graph 字段，其中 Command.PARENT 表示「把指令发给最近的父图」。它用于子图与父图之间的跳转，是多智能体协作的关键，留到第 7 章讲子图时再展开。

📌 另外，Command 还有 resume 字段，用于人工介入后从断点恢复执行——那是第 5 章的主题。

---

六、本章小结

本章速查表

你要做的事	代码
加固定边	.addEdge("a", "b")
加条件边	.addConditionalEdges("a", 路由函数, ["b", "c", END])
路由函数	(state) => "b" | "c" | END（返回值必须是真实节点名或 END）
fan-out（并行）	一个源节点 .addEdge 连出多条边
fan-in（汇合）	多条边 .addEdge 汇入同一个节点（框架自动等齐）
循环	条件边指回一个已经走过的上游节点
防死循环	循环设退出条件 + 最大轮数 + invoke(input, { recursionLimit: N })
捕获超限	catch (e) { if (e instanceof GraphRecursionError) ... }
动态 map-reduce	条件边返回 new Send("节点", 局部状态) 数组
并发聚合字段	Annotation<T[]>({ reducer: (a, b) => a.concat(b), default: () => [] })
一步更新+跳转	return new Command({ update: {...}, goto: "节点" }) + addNode(..., { ends: [...] })

三句话总结

1. 条件边把「下一步去哪」变成可计算的东西——路由函数读状态、返回节点名；第三个参数（可能节点列表）既让结构可见，又是编译期的白名单校验。
2. 并行与聚合是一对孪生兄弟：fan-out 让一个超步里并发跑多个节点，fan-in 自动等齐；而只要并发写同一个字段，那个字段就必须带 reducer（否则 InvalidUpdateError）。
3. Send 和 Command 是两种「动态控制流」：Send 在运行时按数据长度生成 N 个并发任务（map-reduce），Command 让一个节点在一次返回里同时完成「更新状态 + 决定跳转」。

动手练习

1. 改示例 1：给 classify 增加第 4 个分类 code（编程问题）并加一个 code 节点。先故意忘了在第三个参数里加 "code"，跑一次看报什么错；再加上去，验证修复。
2. 改示例 2：把三个 worker 的 sleep(1000) 改成不同的时长（比如 300 / 600 / 900ms），观察 results 的完成顺序和总耗时（提示：总耗时约等于「最慢的那个」）。
3. 改示例 3 的下半场：把 recursionLimit 换成 2 和 100 各跑一次，观察 tick 分别被打印几次。再回答：默认的 recursionLimit 是多少？
4. 扩展示例 5：让 Command 的 goto 返回 new Send("even", { n }) 而不是字符串，看看能不能把「更新状态 + 动态派发」也一步做完。

---

下一章预告

第 4 章我们解决一个所有智能体都绕不开的问题：记忆。为什么图一跑完状态就没了？thread_id 是什么？怎么让智能体「记得住」上一轮对话（甚至几天前的）？怎么在崩溃后断点续跑？答案都在「Checkpointer（检查点器）」里——那是把「一次性脚本」变成「有状态服务」的关键一步。

---

← 上一章：状态与 Reducer ｜ 返回总目录 ｜ 下一章：持久化与记忆 →