LangGraph.js 中文入门

第 10 章 · 应用结构 ​

🎯 本章目标 ​

  • 看清「一个文件写完整条流水线」为什么迟早会崩 —— 那不是风格问题，是结构问题
  • 掌握把 LangGraph 应用分层的四个理由，以及每一层该放什么
  • 拿到一套可复制的目录结构：state / tools / nodes / graph / index
  • 学会两条工程纪律：图只编译一次、运行时参数不硬编码
  • 把「结构复用」与「会话隔离」放在一起理解：一张图如何同时服务很多用户

本章代码

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

文件内容
app/一个真实可运行的分层小项目（本章主角，见第三节）
code/01-flat-version.ts反例：状态 / 工具 / 节点 / 组装全塞在一个文件里，跑完后逐条打印真实痛点
code/02-layered-run.ts导入 app/ 跑同一任务，把目录职责与 import 路径打印出来，与示例 1 对照
code/03-compile-once.ts图只编译一次：对比「每请求都编译」vs「模块级编译一次」的真实耗时
code/04-runtime-context.ts运行时参数用 context 传：同一张图，三种 userId / language / channel
code/05-singleton-with-checkpointer.ts单例图 + MemorySaver：一张图服务多个 thread_id，结构复用与会话隔离并存

⚠️ 本章不重复第 8 章（工程化与部署）讲过的内容：重试、缓存、错误恢复、HTTP 服务、可观测性。那些是「单个节点/单条链路怎么抗造」；本章只回答一个问题：当应用长到十几个文件时，代码该怎么摆。

一、单文件为什么会崩（反例） ​

先说结论：把「定义状态 + 写工具 + 写节点 + 连图 + 跑一遍」全塞进一个文件，一开始很爽，规模一大必崩。这不是审美问题，是三种真实成本在累积：

  • 改动成本：想加一个工具，得在大文件的好几个地方同时改，漏一处运行时才炸；
  • 理解成本：状态字段和业务逻辑混在一起，「一个会话有哪些数据」这种问题都答不上来；
  • 协作成本：多人同时动这个文件，合并冲突几乎躲不掉。

先让反例跑起来，跑完它会自己把痛点列出来。

示例 1：单文件「上帝文件」版（能跑通） ​

这个文件和后面的分层版做完全一样的任务：算 128 × 7，再数一句英文的单词数。区别只在于——它把四件事全堆在一个文件里。

typescript
/**
 * 示例 1：反例 —— 把状态 / 工具 / 节点 / 组装全塞进一个文件
 *
 * 这个文件「能跑通」，任务和 app/ 分层版完全一样：算 128×7，再数一句英文的单词数。
 * 先让它正常运行，跑完后再逐条打印这种「上帝文件」写法在真实工程里的痛点。
 *
 * 运行（项目根目录）：npx tsx 10-project-structure/code/01-flat-version.ts
 */
import { StateGraph, START, END, Annotation, messagesStateReducer } from "@langchain/langgraph";
import { HumanMessage, ToolMessage, type AIMessage, type BaseMessage } 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";

// ═══════════════════════════════════════════════════════════════════════
// ① 状态 —— 和工具、节点堆在同一个文件里
// ═══════════════════════════════════════════════════════════════════════
const FlatState = Annotation.Root({
  messages: Annotation<BaseMessage[]>({ reducer: messagesStateReducer, default: () => [] }),
  userQuestion: Annotation<string>({ reducer: (_p, n) => n, default: () => "" }),
  finalAnswer: Annotation<string>({ reducer: (_p, n) => n, default: () => "" }),
  toolCalls: Annotation<number>({ reducer: (a, b) => a + b, default: () => 0 }),
});
type State = typeof FlatState.State;
type Update = typeof FlatState.Update;

// ═══════════════════════════════════════════════════════════════════════
// ② 工具 —— 定义、注册数组、按名查表，也全在这里
// ═══════════════════════════════════════════════════════════════════════
const calculator = tool(
  async ({ expression }) => {
    console.log(`      🔧 calculator("${expression}")`);
    return String(evaluate(expression));
  },
  {
    name: "calculator",
    description: "执行数学计算。",
    schema: z.object({ expression: z.string().describe("数学表达式，例如 '128 * 7'") }),
  }
);

const wordCount = tool(
  async ({ text }) => {
    console.log(`      🔧 word_count("${text}")`);
    const words = text.trim().split(/\s+/).filter(Boolean);
    return `文本「${text}」共有 ${words.length} 个单词、${text.length} 个字符。`;
  },
  {
    name: "word_count",
    description: "统计一段文本里有多少个单词。",
    schema: z.object({ text: z.string().describe("要统计的文本") }),
  }
);

const tools = [calculator, wordCount];
const toolsByName: Record<string, (typeof tools)[number]> = Object.fromEntries(
  tools.map((t) => [t.name, t])
);

// ═══════════════════════════════════════════════════════════════════════
// ③ 节点 —— 模型节点、工具节点、收尾节点，也混在这里
// ═══════════════════════════════════════════════════════════════════════
const model = createModel({ temperature: 0 }).bindTools(tools);

async function agentNode(state: State): Promise<Update> {
  console.log(`   [agent] 把 ${state.messages.length} 条消息交给模型…`);
  const res = (await model.invoke(state.messages)) as AIMessage;
  const calls = res.tool_calls ?? [];
  console.log(`   [agent] 模型决定：${calls.length ? "调用工具 " + calls.map((c) => c.name).join(", ") : "给出最终回答"}`);
  const update: Update = { messages: [res] };
  if (calls.length > 0) update.toolCalls = calls.length;
  return update;
}

async function toolsNode(state: State): Promise<Update> {
  const last = state.messages[state.messages.length - 1] as AIMessage;
  const calls = last.tool_calls ?? [];
  const results: ToolMessage[] = [];
  for (const call of calls) {
    const selected = toolsByName[call.name];
    const loose = selected as unknown as { invoke: (i: Record<string, unknown>) => Promise<unknown> };
    const output = await loose.invoke(call.args as Record<string, unknown>);
    results.push(new ToolMessage({ content: String(output), name: call.name, tool_call_id: call.id ?? call.name }));
  }
  console.log(`   [tools] 执行了 ${results.length} 个工具`);
  return { messages: results };
}

function finalizeNode(state: State): Update {
  const last = state.messages[state.messages.length - 1] as BaseMessage;
  return { finalAnswer: String(last.content ?? "") };
}

// ═══════════════════════════════════════════════════════════════════════
// ④ 组装图 —— 连线也写在这个文件里
// ═══════════════════════════════════════════════════════════════════════
function buildFlatGraph() {
  return new StateGraph(FlatState)
    .addNode("agent", agentNode)
    .addNode("tools", toolsNode)
    .addNode("finalize", finalizeNode)
    .addEdge(START, "agent")
    .addEdge("tools", "agent")
    .addConditionalEdges(
      "agent",
      (state) => {
        const last = state.messages[state.messages.length - 1] as AIMessage;
        return last.tool_calls?.length ? "tools" : "finalize";
      },
      ["tools", "finalize"]
    )
    .addEdge("finalize", END)
    .compile();
}

const QUESTION =
  "帮我算一下 128 乘以 7 是多少，再统计 'the quick brown fox jumps over' 这句话有多少个单词。";

function printPainPoints() {
  banner("示例 1 的「真实痛点」清单（针对上面这个文件）");
  const pains = [
    "① 新增一个工具要改 3 处：写工具对象、加进 tools 数组、加进 toolsByName 映射 —— 全在同一大文件里，改漏一处运行时才报错。",
    "② 状态字段和业务逻辑混在一起：想知道「会话里有哪些数据」，得在几千行里翻找 Annotation.Root；改一个字段牵动全文。",
    "③ 没法单测：想给 calculator 写测试，import 这个文件就会连带初始化模型、绑定工具、读取环境变量。",
    "④ 没法复用：别的功能想用同一条 agent 流水线，只能整段复制粘贴，两份代码从此各自漂移。",
    "⑤ 合并冲突高发：同事 A 改状态、同事 B 加工具、同事 C 调节点，三个人都会撞在同一个文件、同一片区域。",
    "⑥ 文件线性膨胀：每加一个工具/节点，这个文件就长一截，最终变成没人敢动的「上帝文件」。",
    "→ 结论：这些不是「风格偏好」，而是会实际拖慢迭代、制造事故的结构性问题。下一节看分层版如何逐条解决。",
  ];
  for (const p of pains) console.log("   " + p);
}

async function main() {
  banner("示例 1：单文件「上帝文件」版（能跑通，但结构有问题）");
  assertChatKey();

  const graph = buildFlatGraph();
  console.log(`👤 提问：${QUESTION}\n`);
  const result = await graph.invoke({ messages: [new HumanMessage(QUESTION)], userQuestion: QUESTION });

  console.log(`\n📊 本会话累计工具调用：${result.toolCalls} 次`);
  console.log(`\n🤖 最终回答：${result.finalAnswer}`);

  printPainPoints();
}

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

运行：

bash
npx tsx 10-project-structure/code/01-flat-version.ts

真实输出：

========================================================================
  示例 1：单文件「上帝文件」版（能跑通，但结构有问题）
========================================================================

👤 提问：帮我算一下 128 乘以 7 是多少，再统计 'the quick brown fox jumps over' 这句话有多少个单词。

   [agent] 把 1 条消息交给模型…
   [agent] 模型决定：调用工具 calculator, word_count
      🔧 calculator("128 * 7")
      🔧 word_count("the quick brown fox jumps over")
   [tools] 执行了 2 个工具
   [agent] 把 4 条消息交给模型…
   [agent] 模型决定：给出最终回答

📊 本会话累计工具调用：2 次

🤖 最终回答：两个结果都算好了：

- **128 × 7 = 896**
- **'the quick brown fox jumps over' 共有 6 个单词**（the / quick / brown / fox / jumps / over）

需要我再算点别的吗？

========================================================================
  示例 1 的「真实痛点」清单（针对上面这个文件）
========================================================================

   ① 新增一个工具要改 3 处：写工具对象、加进 tools 数组、加进 toolsByName 映射 —— 全在同一大文件里，改漏一处运行时才报错。
   ② 状态字段和业务逻辑混在一起：想知道「会话里有哪些数据」，得在几千行里翻找 Annotation.Root；改一个字段牵动全文。
   ③ 没法单测：想给 calculator 写测试，import 这个文件就会连带初始化模型、绑定工具、读取环境变量。
   ④ 没法复用：别的功能想用同一条 agent 流水线，只能整段复制粘贴，两份代码从此各自漂移。
   ⑤ 合并冲突高发：同事 A 改状态、同事 B 加工具、同事 C 调节点，三个人都会撞在同一个文件、同一片区域。
   ⑥ 文件线性膨胀：每加一个工具/节点，这个文件就长一截，最终变成没人敢动的「上帝文件」。
   → 结论：这些不是「风格偏好」，而是会实际拖慢迭代、制造事故的结构性问题。下一节看分层版如何逐条解决。

注意：这个反例本身没有 bug，任务跑得漂漂亮亮。它的问题不在「对不对」，而在「以后好不好改」。所以「能跑通」永远不是组织结构是否合理的判据。

二、分层的四个理由 ​

把上面的反例拆开，分层的价值恰好有四条，一一对应它的痛点：

理由解决反例里的哪个痛点
① 可定位：状态、工具、节点、组装各归一个文件，找东西只去一个地方「几千行里翻 Annotation.Root」
② 可复用：工具是纯函数、节点是纯函数，谁都能 import 拿去用「只能整段复制粘贴」
③ 可测试：单独 import 一个工具或节点，不触发模型初始化、不读环境变量「想单测必须 import 整个大文件」
④ 可协作：不同的人改不同的文件，互不打架「多人撞在同一个文件上」

再补一条隐含收益：入口文件变短。分层之后，真正「跑应用」的那个文件只剩几十行，新人可以 30 秒读完主流程。

💡 一句话记忆：分层不是为了好看，是为了让「改动」和「理解」都变成局部动作。 你只想改工具时，就不该被迫读节点；你只想看流程时，就不该被工具实现淹没。

分层的原则只有一条：每一层只依赖它下一层，且能被独立替换。 本章推荐的目录结构就是这条原则的一个具体落地。

三、推荐的目录结构 ​

先看目录树。这不是唯一答案，但它把「状态 / 工具 / 节点 / 组装 / 入口」五个职责分得干干净净：

10-project-structure/app/          ← 分层小项目
├── state.ts                       状态层：Annotation.Root 定义，全项目唯一的「共享约定」
├── tools/                         工具层：一个工具一个文件
│   ├── calculator.ts              数学计算工具
│   ├── text-stats.ts              文本统计工具
│   └── index.ts                   工具出口：汇总 tools 数组 + 按名查表 toolsByName
├── nodes/                         节点层：一个职责一个文件
│   ├── agent.ts                   调用模型（绑定工具）
│   ├── tools.ts                   执行工具调用
│   ├── finalize.ts                整理最终回答
│   ├── edges.ts                   条件边的路由判断（连边的「脑子」）
│   └── index.ts                   节点出口：汇总所有节点
├── graph.ts                       组装层：唯一 new StateGraph 的地方，导出 buildGraph()
└── index.ts                       入口层：跑一次完整对话

每层职责一览：

文件职责只依赖
state.ts定义状态字段与类型（含 reducer），无任何业务逻辑LangGraph 的 Annotation
tools/*.ts单个被模型调用的能力，纯函数式、可独立测试@langchain/core/tools、zod、mathjs
tools/index.ts统一出口：tools 数组 + toolsByName 查表各工具文件
nodes/*.ts单个业务步骤（进来 state，出去 update），一个职责一个文件状态类型、工具出口、模型工厂
nodes/edges.ts条件边的路由函数（不产生状态更新）状态类型
nodes/index.ts统一出口：汇总所有节点各节点文件
graph.ts唯一组装图的地方：加节点、连边、compile()状态、节点出口
index.ts入口：校验环境 → buildGraph() → invoke() → 打印结果组装层、lib/

💡 为什么 edges.ts 和节点并列？因为它既不是节点（不产生状态更新），也不是组装（不 new StateGraph）。它是「连边的逻辑」，单独放出来，graph.ts 里就只剩纯粹的连线调用，一眼能看完。

关键文件巡礼：组装层 ​

分层里最该记住的是组装层——它是全项目唯一 new StateGraph 的地方（app/graph.ts 全文）：

typescript
// app/graph.ts —— 组装层：只负责把零件连成一张图
import { StateGraph, START, END } from "@langchain/langgraph";
import { GraphState } from "./state.js";
import { agentNode, toolsNode, finalizeNode, routeAfterAgent } from "./nodes/index.js";

export function buildGraph() {
  return new StateGraph(GraphState)
    .addNode("agent", agentNode)        // 调用模型
    .addNode("tools", toolsNode)        // 执行工具
    .addNode("finalize", finalizeNode)  // 整理最终回答
    .addEdge(START, "agent")
    .addEdge("tools", "agent")          // 工具执行完，回到模型再判断 ← 循环
    .addConditionalEdges("agent", routeAfterAgent, ["tools", "finalize"])
    .addEdge("finalize", END)
    .compile();
}

export type AgentGraph = ReturnType<typeof buildGraph>;

💡 注意 buildGraph() 只构建、不调用：它不 invoke、也不在模块顶层 compile，而是把「何时执行、编译几次」留给调用方——这是下一节「图只编译一次」的前提。

💡 每个节点文件（app/nodes/*.ts）顶部都有一段「这一层存在的理由」注释，那是本章的教学主体，请务必打开 app/ 源码读一遍。app/state.ts 与 app/index.ts 分别只有几十行和二十来行，配合注释一看便懂。

跑一次这个分层项目（项目根目录）：npx tsx 10-project-structure/app/index.ts，会打印与下面示例 2 逐字相同的逐层日志（[agent] / [tools] / [条件边] / [finalize] 前缀分别来自四个不同文件——读日志就能定位到文件，这是分层在调试时的直接好处）。

示例 2：分层版跑同一个任务 ​

「跑一次」看不出分层与反例的差别，所以再看一个专门做对照的示例：它不实现任何业务逻辑，只 import app/，并把目录职责与 import 路径打印出来。

typescript
/**
 * 示例 2：分层版 —— 导入 app/，跑同一个任务
 *
 * 与示例 1 的对照点：这里「没有」状态、工具、节点的实现，只有一句 import。
 * 本示例把 app/ 的目录结构、每层职责、以及真实的 import 路径打印出来，
 * 让你直观看到「一个入口文件只依赖一个组装层」是什么体验。
 *
 * 运行（项目根目录）：npx tsx 10-project-structure/code/02-layered-run.ts
 */
import { HumanMessage, type BaseMessage, type AIMessage } from "@langchain/core/messages";
import { banner } from "../../lib/model.js";
import { assertChatKey } from "../../lib/env.js";
// ★ 本文件与业务实现之间，只隔着这一句 import
import { buildGraph } from "../app/graph.js";

const QUESTION =
  "帮我算一下 128 乘以 7 是多少，再统计 'the quick brown fox jumps over' 这句话有多少个单词。";

/** 打印 app/ 的分层结构与各层职责 */
function printStructure() {
  banner("app/ 的目录结构与每层职责");
  const rows: Array<[string, string]> = [
    ["app/state.ts", "状态层：只定义状态字段与类型，唯一「共享约定」"],
    ["app/tools/calculator.ts", "工具层：数学计算工具（一个工具一个文件）"],
    ["app/tools/text-stats.ts", "工具层：文本统计工具"],
    ["app/tools/index.ts", "工具层出口：汇总 tools 数组与按名查表"],
    ["app/nodes/agent.ts", "节点层：调用模型这一职责"],
    ["app/nodes/tools.ts", "节点层：执行工具这一职责"],
    ["app/nodes/finalize.ts", "节点层：整理最终回答这一职责"],
    ["app/nodes/edges.ts", "边层：条件边的路由判断"],
    ["app/nodes/index.ts", "节点层出口：汇总所有节点"],
    ["app/graph.ts", "组装层：唯一 new StateGraph 的地方，导出 buildGraph()"],
    ["app/index.ts", "入口层：跑一次完整对话"],
  ];
  for (const [file, duty] of rows) console.log(`   ${file.padEnd(26)} ${duty}`);
}

/** 打印本示例实际用到的 import，说明「入口只依赖组装层」 */
function printImports() {
  console.log("\n📌 本示例的 import（注意最后一行）：");
  const imports = [
    'import { HumanMessage, ... } from "@langchain/core/messages";   // 消息类型',
    'import { banner } from "../../lib/model.js";                    // 打印工具',
    'import { assertChatKey } from "../../lib/env.js";               // 校验环境',
    'import { buildGraph } from "../app/graph.js";                   // ★ 只要这一个',
  ];
  for (const line of imports) console.log("   " + line);
  console.log("   —— 状态、工具、节点的实现全在 app/ 内部，入口文件完全看不到。");
}

async function main() {
  banner("示例 2：分层版（import app/graph.js，跑同一任务）");
  assertChatKey();

  printStructure();
  printImports();

  console.log("\n" + "─".repeat(72));
  console.log("现在运行 app/ 里的图，日志来自各层的文件：");
  console.log("─".repeat(72) + "\n");

  const graph = buildGraph();
  console.log(`👤 提问：${QUESTION}\n`);
  const result = await graph.invoke({
    messages: [new HumanMessage(QUESTION)],
    userQuestion: QUESTION,
  });

  console.log("\n📜 消息轨迹：");
  result.messages.forEach((m: BaseMessage, i: number) => {
    const calls = (m as AIMessage).tool_calls;
    const label = `   ${i}. [${m.getType()}] `;
    if (calls?.length) {
      console.log(label + `请求调用 ${calls.map((c) => `${c.name}(${JSON.stringify(c.args)})`).join(", ")}`);
    } else {
      console.log(label + String(m.content).replace(/\n/g, " ").slice(0, 100));
    }
  });

  console.log(`\n📊 本会话累计工具调用：${result.toolCalls} 次`);
  console.log(`\n🤖 最终回答：${result.finalAnswer}`);
  console.log("\n✅ 同一个任务、同一张图，入口文件只有几十行 —— 这就是分层的收益。");
}

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

运行：

bash
npx tsx 10-project-structure/code/02-layered-run.ts

真实输出：

========================================================================

   app/state.ts               状态层：只定义状态字段与类型，唯一「共享约定」
   app/tools/calculator.ts    工具层：数学计算工具（一个工具一个文件）
   app/tools/text-stats.ts    工具层：文本统计工具
   app/tools/index.ts         工具层出口：汇总 tools 数组与按名查表
   app/nodes/agent.ts         节点层：调用模型这一职责
   app/nodes/tools.ts         节点层：执行工具这一职责
   app/nodes/finalize.ts      节点层：整理最终回答这一职责
   app/nodes/edges.ts         边层：条件边的路由判断
   app/nodes/index.ts         节点层出口：汇总所有节点
   app/graph.ts               组装层：唯一 new StateGraph 的地方，导出 buildGraph()
   app/index.ts               入口层：跑一次完整对话

📌 本示例的 import（注意最后一行）：
   import { HumanMessage, ... } from "@langchain/core/messages";   // 消息类型
   import { banner } from "../../lib/model.js";                    // 打印工具
   import { assertChatKey } from "../../lib/env.js";               // 校验环境
   import { buildGraph } from "../app/graph.js";                   // ★ 只要这一个
   —— 状态、工具、节点的实现全在 app/ 内部，入口文件完全看不到。

────────────────────────────────────────────────────────────────────────
现在运行 app/ 里的图，日志来自各层的文件：
────────────────────────────────────────────────────────────────────────

👤 提问：帮我算一下 128 乘以 7 是多少，再统计 'the quick brown fox jumps over' 这句话有多少个单词。

   [agent] 把 1 条消息交给模型…
   [agent] 模型决定：调用工具 calculator, word_count
   [条件边] 有工具调用 → 走 "tools"
      🔧 工具被调用：calculator("128 * 7")
      🔧 工具被调用：word_count("the quick brown fox jumps over")
   [tools] 执行了 2 个工具，结果已写回消息历史
   [agent] 把 4 条消息交给模型…
   [agent] 模型决定：给出最终回答
   [条件边] 无工具调用 → 走 "finalize"
   [finalize] 抽取最终回答（154 字符）

📜 消息轨迹：
   0. [human] 帮我算一下 128 乘以 7 是多少，再统计 'the quick brown fox jumps over' 这句话有多少个单词。
   1. [ai] 请求调用 calculator({"expression":"128 * 7"}), word_count({"text":"the quick brown fox jumps over"})
   2. [tool] 896
   3. [tool] 文本「the quick brown fox jumps over」共有 6 个单词、30 个字符。
   4. [ai] 两个结果都出来了：  1. **128 × 7 = 896** 2. **'the quick brown fox jumps over' 共有 6 个单词**（the / quick / brown

📊 本会话累计工具调用：2 次

🤖 最终回答：两个结果都出来了：

1. **128 × 7 = 896**
2. **'the quick brown fox jumps over' 共有 6 个单词**（the / quick / brown / fox / jumps / over）

顺带补充一句：工具还统计出这句话共 30 个字符（含空格）。

✅ 同一个任务、同一张图，入口文件只有几十行 —— 这就是分层的收益。

对照示例 1：反例的「④没法复用、③没法单测」两个痛点，在这里直接消解了 —— 示例 2 只写了一句 import { buildGraph } from "../app/graph.js"，就复用了整条流水线。

四、图只编译一次 ​

分层把代码摆好了，接下来是两条最容易踩的工程纪律。第一条：图只该编译一次。

new StateGraph(...).compile() 不是免费的。它要校验状态字段、创建 channel、解析边……「每次请求都编译一遍」看起来无害，但在生产里，每个 HTTP 请求都做一遍这些工作，就是白白的开销。反过来，图的结构在运行时根本不会变，所以它天生就该是进程级单例。

这也正是 app/graph.ts 里 buildGraph() 只「构建」的原因：它把「何时编译」的决定权交给调用方，所以我们可以选择「模块级编译一次」。

示例 3：编译一次 vs 每次编译 ​

为了排除网络抖动，这个示例用纯函数节点（不调用模型），只测「建图 + 编译」带来的差异。为了让计时稳健，它每个测量项跑 200 次、重复 3 轮取最快一轮，并额外单独测了「纯编译」的开销。

typescript
/**
 * 示例 3：图只编译一次 —— 一个真实存在的性能坑
 *
 * 为了让计时不被偶然因素干扰，本示例做了三件事：
 *   ① 用纯函数节点（不调用模型，排除网络波动）；
 *   ② 每个测量项跑 REPEATS 轮、取「最快的一轮」（min 对 GC/调度抖动最稳）；
 *   ③ 额外单独测「纯编译」的开销，让结论有直接证据。
 *
 * 运行（项目根目录）：npx tsx 10-project-structure/code/03-compile-once.ts
 */
import { StateGraph, START, END, Annotation } from "@langchain/langgraph";
import { banner } from "../../lib/model.js";

// ── 状态：输入 n，输出 2n ─────────────────────────────────────────
const DoubleState = Annotation.Root({
  n: Annotation<number>,
  result: Annotation<number>,
});

function doubleNode(state: typeof DoubleState.State) {
  return { result: state.n * 2 };
}

/** 建图 + 编译：这是每次请求都在重复的工作 */
function buildAndCompile() {
  return new StateGraph(DoubleState)
    .addNode("double", doubleNode)
    .addEdge(START, "double")
    .addEdge("double", END)
    .compile();
}

// ★ 模块级（进程级）常量：整个进程只构建、编译「一次」，之后所有请求复用它
const REUSED_GRAPH = buildAndCompile();

const ROUNDS = 200; // 每轮跑多少次操作
const REPEATS = 3; // 每个测量项重复几轮，取最快的一轮

/** 重复跑若干轮，返回「最快的一轮」的毫秒数（比单次测量稳健得多） */
async function bestOf(run: () => Promise<void>): Promise<number> {
  let best = Number.POSITIVE_INFINITY;
  for (let r = 0; r < REPEATS; r++) {
    const t0 = performance.now();
    await run();
    const cost = performance.now() - t0;
    if (cost < best) best = cost;
  }
  return best;
}

/** 只建图 + 编译，不执行 */
function benchCompileOnly(): Promise<number> {
  return bestOf(async () => {
    for (let i = 0; i < ROUNDS; i++) buildAndCompile();
  });
}

/** 写法 A：每个请求都新建并编译一张图，再执行 */
function benchCompileEachTime(): Promise<number> {
  return bestOf(async () => {
    for (let i = 0; i < ROUNDS; i++) {
      const g = buildAndCompile();
      await g.invoke({ n: i });
    }
  });
}

/** 写法 B：复用同一张已经编译好的图，只执行 */
function benchReuse(): Promise<number> {
  return bestOf(async () => {
    for (let i = 0; i < ROUNDS; i++) {
      await REUSED_GRAPH.invoke({ n: i });
    }
  });
}

async function main() {
  banner("示例 3：图只编译一次 vs 每次请求都编译");

  // ── 热身：让 JIT 预热，避免首次运行的抖动被算进结果 ──
  await buildAndCompile().invoke({ n: -1 });
  await REUSED_GRAPH.invoke({ n: -1 });

  console.log(`🔁 每个测量项跑 ${ROUNDS} 次、重复 ${REPEATS} 轮取最快一轮。`);
  console.log("   节点是纯函数（不含网络），所以差异只可能来自「建图 + 编译」。\n");

  const tCompileOnly = await benchCompileOnly();
  console.log(
    `   ① 纯编译（只建图，不执行）×${ROUNDS}：       ${tCompileOnly.toFixed(1)} ms  ` +
      `（≈ ${(tCompileOnly / ROUNDS).toFixed(3)} ms / 次）`
  );

  const tEach = await benchCompileEachTime();
  console.log(`   ② A. 每次请求都编译 + 执行 ×${ROUNDS}：   ${tEach.toFixed(1)} ms  （≈ ${(tEach / ROUNDS).toFixed(3)} ms / 次）`);

  const tReuse = await benchReuse();
  console.log(`   ③ B. 复用已编译的图 + 执行 ×${ROUNDS}：   ${tReuse.toFixed(1)} ms  （≈ ${(tReuse / ROUNDS).toFixed(3)} ms / 次）`);

  console.log("\n📊 对比：");
  console.log(`   • A 每次请求都多做了「建图 + 编译」，${ROUNDS} 次累计多花约 ${(tEach - tReuse).toFixed(1)} ms；`);
  console.log(`   • 单独测的「纯编译」×${ROUNDS} 只有 ${tCompileOnly.toFixed(1)} ms —— 编译本身确实有成本，而 A 还要额外承担频繁建图的内存分配/回收，所以差额更大；`);
  console.log(`   • 换成「编译一次」后，这部分开销被直接省掉，A 与 B 的执行结果完全一致。`);

  console.log("\n💡 结论：图的结构在运行时不会变，就该只编译一次。");
  console.log("   把编译好的图放在模块顶层（或进程级单例），所有请求复用它即可。");

  const a = await buildAndCompile().invoke({ n: 21 });
  const b = await REUSED_GRAPH.invoke({ n: 21 });
  console.log(`\n✅ 结果一致性检查：n=21 时 A 得 ${a.result}，B 得 ${b.result}（应都为 42）。`);
}

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

运行：

bash
npx tsx 10-project-structure/code/03-compile-once.ts

真实输出（连续两次运行都很稳定）：

========================================================================
  示例 3：图只编译一次 vs 每次请求都编译
========================================================================

🔁 每个测量项跑 200 次、重复 3 轮取最快一轮。
   节点是纯函数（不含网络），所以差异只可能来自「建图 + 编译」。

   ① 纯编译（只建图，不执行）×200：       2.4 ms  （≈ 0.012 ms / 次）
   ② A. 每次请求都编译 + 执行 ×200：   53.0 ms  （≈ 0.265 ms / 次）
   ③ B. 复用已编译的图 + 执行 ×200：   46.3 ms  （≈ 0.231 ms / 次）

📊 对比：
   • A 每次请求都多做了「建图 + 编译」，200 次累计多花约 6.7 ms；
   • 单独测的「纯编译」×200 只有 2.4 ms —— 编译本身确实有成本，而 A 还要额外承担频繁建图的内存分配/回收，所以差额更大；
   • 换成「编译一次」后，这部分开销被直接省掉，A 与 B 的执行结果完全一致。

💡 结论：图的结构在运行时不会变，就该只编译一次。
   把编译好的图放在模块顶层（或进程级单例），所有请求复用它即可。

✅ 结果一致性检查：n=21 时 A 得 42，B 得 42（应都为 42）。

💡 怎么读这组数字：绝对差额只有几毫秒，因为这里是最小的图。真实的 agent 图节点更多、边更多、还带 schema 校验，编译成本高得多；更重要的是，这类开销是纯浪费——它不产生任何业务价值。判断标准很简单：凡是「结构在运行时不变」的东西，就不该每次重建。

⚠️ 坑：微型测量极易被噪声淹没。 我第一次只跑 20 次、且每个只测一遍时，结果甚至是反的（B 比 A 还慢），因为总量只有十几毫秒，被 GC 和调度抖动吃掉了。改成「200 次 × 重复 3 轮取最快」后才稳定。基准测试要「放大信号、压制噪声」，别用一次小样本下定论。

五、运行时参数怎么传（别硬编码） ​

第二条纪律：每次调用都不同的值，不要硬编码进图里。

典型的运行时参数：用户 ID、语言、渠道（web/api/cli）……这类值的共同特点是：

  • 它们每次调用都不一样——图是全局共享的，硬编码等于把所有用户当成同一个人；
  • 它们不该进「状态」——状态是会随节点流动、被 reducer 合并的业务数据，而用户 ID / 渠道是「环境」，不该参与状态合并、也不该出现在历史里。

LangGraph.js（v1）为此提供了 context（上下文） 机制：

  1. 建图时用第二个参数声明 contextSchema：new StateGraph(StateSchema, ContextSchema)；
  2. 调用时通过配置传入：await graph.invoke(input, { context: { ... } })；
  3. 节点里用第二个参数（Runtime）读出：runtime.context.xxx。

📌 本写法已对照 node_modules/@langchain/langgraph/dist/graph/state.d.ts 与 dist/pregel/runnable_types.d.ts 确认：StateGraph 构造函数接受 context 参数；节点签名为 (state, runtime)，runtime.context 即传入的运行时上下文。若你的版本尚不支持 contextSchema，可退化为 invoke(input, { configurable: { ... } })，并在节点内用 getConfig().configurable 读取——本例锁定版本已支持 context，故优先使用它。

示例 4：同一张图，三种运行时参数 ​

这是本章的「最小机制示例」：一张图，三次调用，传入三组不同的 userId / language / channel，节点读出后组织出不同回复。

typescript
/**
 * 示例 4：运行时参数怎么传 —— 别硬编码，用 context
 *
 * 运行（项目根目录）：npx tsx 10-project-structure/code/04-runtime-context.ts
 */
import { StateGraph, START, END, Annotation, type Runtime } from "@langchain/langgraph";
import { banner } from "../../lib/model.js";

// ── 状态：会随节点流动、被 reducer 合并的数据 ──────────────────────
const GreetState = Annotation.Root({
  utterance: Annotation<string>, // 用户这一轮说的话（会随图流动）
  reply: Annotation<string>,     // 节点产出的回复
});

// ── 上下文（context）：每次调用都不同、且不参与状态合并的「环境参数」──
const GreetContext = Annotation.Root({
  userId: Annotation<string>,
  language: Annotation<string>,
  channel: Annotation<string>,
});

/** context 的类型：节点里 runtime.context 就是这个形状 */
type GreetContextType = {
  userId: string;
  language: string;
  channel: string;
};

/** 节点：从 runtime.context 读取运行时参数来组织回复 */
function greetNode(
  state: typeof GreetState.State,
  runtime: Runtime<GreetContextType>
): typeof GreetState.Update {
  const ctx = runtime.context;
  const userId = ctx?.userId ?? "匿名用户";
  const language = ctx?.language ?? "zh";
  const channel = ctx?.channel ?? "cli";

  const hello = language === "en" ? "Hello" : language === "ja" ? "こんにちは" : "你好";
  const suffix = channel === "web" ? "（来自网页）" : channel === "api" ? "（来自 API）" : "";

  const reply = `${hello}，${userId}${suffix}！你说的是：「${state.utterance}」`;

  console.log(
    `   [节点] 读到 context → userId=${userId} language=${language} channel=${channel}`
  );
  return { reply };
}

async function main() {
  banner("示例 4：运行时参数用 context 传，不硬编码");

  // ★ 建图：第二个参数声明 contextSchema。图只建这一次，所有调用共用。
  const graph = new StateGraph(GreetState, GreetContext)
    .addNode("greet", greetNode)
    .addEdge(START, "greet")
    .addEdge("greet", END)
    .compile();

  // 同一张图，三次调用，传入三组不同的运行时参数
  const requests: Array<{ label: string; utterance: string; context: GreetContextType }> = [
    { label: "请求 1", utterance: "今天天气不错", context: { userId: "user-alice", language: "zh", channel: "web" } },
    { label: "请求 2", utterance: "how are you?", context: { userId: "user-bob", language: "en", channel: "api" } },
    { label: "请求 3", utterance: "元気ですか", context: { userId: "user-carol", language: "ja", channel: "cli" } },
  ];

  for (const req of requests) {
    console.log(`\n🔹 ${req.label}`);
    const result = await graph.invoke(
      { utterance: req.utterance },
      { context: req.context } // ★ 运行时参数从 config.context 传入
    );
    console.log(`   → reply: ${result.reply}`);
  }

  console.log("\n📌 小结：");
  console.log("   • 状态（state）：随图流动、会被 reducer 合并 → 放 Annotation.Root");
  console.log("   • 上下文（context）：每次调用不同、不参与合并 → 建图时声明 contextSchema");
  console.log("   • 同一张图、三种参数，各得其所，互不污染。");
}

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

运行：

bash
npx tsx 10-project-structure/code/04-runtime-context.ts

真实输出：

========================================================================
  示例 4：运行时参数用 context 传，不硬编码
========================================================================

🔹 请求 1
   [节点] 读到 context → userId=user-alice language=zh channel=web
   → reply: 你好，user-alice（来自网页）！你说的是：「今天天气不错」

🔹 请求 2
   [节点] 读到 context → userId=user-bob language=en channel=api
   → reply: Hello，user-bob（来自 API）！你说的是：「how are you?」

🔹 请求 3
   [节点] 读到 context → userId=user-carol language=ja channel=cli
   → reply: こんにちは，user-carol！你说的是：「元気ですか」

📌 小结：
   • 状态（state）：随图流动、会被 reducer 合并 → 放 Annotation.Root
   • 上下文（context）：每次调用不同、不参与合并 → 建图时声明 contextSchema
   • 同一张图、三种参数，各得其所，互不污染。

记住这张对照表——它是本章最容易记混的一处：

状态（state）上下文（context）
来源用户输入 + 节点产出调用方通过 config.context 传入
生命周期随图流动，会被 reducer 合并只读的「环境」，不参与合并
典型内容messages、业务字段userId、language、channel、feature flag
写法Annotation.Root({...}) 第一个参数StateGraph(State, Context) 第二个参数
节点里怎么读函数第一个参数 state函数第二个参数 runtime.context

六、复用与隔离 ​

最后把「只编译一次」和「会话持久化」合起来看。它们看似矛盾——图是共享的一张三单例，会话却要各自独立——其实完全可以同时成立：结构共享，数据按 thread_id 隔离。

这正是第 4 章 MemorySaver + thread_id 的用法，放在「单例图」的语境下再理解一遍，印象会更深。

示例 5：一张单例图，多个 thread_id ​

为了让结果稳定可复现（本章讲结构，不是模型能力），回复节点是确定性的、不调用大模型；checkpointer 用内存版 MemorySaver。

typescript
/**
 * 示例 5：单例图 + checkpointer —— 结构复用与会话隔离同时成立
 *
 * 运行（项目根目录）：npx tsx 10-project-structure/code/05-singleton-with-checkpointer.ts
 */
import {
  StateGraph, START, END, Annotation, messagesStateReducer, MemorySaver,
} from "@langchain/langgraph";
import { HumanMessage, AIMessage, type BaseMessage } from "@langchain/core/messages";
import { banner } from "../../lib/model.js";

// ── 状态：消息 + 轮次计数 ─────────────────────────────────────────
const ConvState = Annotation.Root({
  messages: Annotation<BaseMessage[]>({
    reducer: messagesStateReducer,
    default: () => [],
  }),
  turns: Annotation<number>({ reducer: (a, b) => a + b, default: () => 0 }),
});

/** 确定性回复节点：数一数这是本会话第几条 AI 回复，并回显用户输入 */
function replyNode(state: typeof ConvState.State): typeof ConvState.Update {
  const aiCount = state.messages.filter((m) => m.getType() === "ai").length;
  const lastHuman = [...state.messages].reverse().find((m) => m.getType() === "human");
  const text = lastHuman ? String(lastHuman.content) : "（无）";
  return {
    messages: [new AIMessage(`（本会话第 ${aiCount + 1} 条回复）收到：${text}`)],
    turns: 1,
  };
}

// ══════════════════════════════════════════════════════════════════════
// ★ 模块级单例：整个进程只构建、编译「一次」，并挂上 MemorySaver
//   这一张图会被下面所有的 thread_id 共享。
// ══════════════════════════════════════════════════════════════════════
export const conversationGraph = new StateGraph(ConvState)
  .addNode("reply", replyNode)
  .addEdge(START, "reply")
  .addEdge("reply", END)
  .compile({ checkpointer: new MemorySaver() });

/** 往某个会话发一条消息：注意用的是同一张 conversationGraph */
async function send(threadId: string, text: string) {
  const config = { configurable: { thread_id: threadId } };
  const result = await conversationGraph.invoke(
    { messages: [new HumanMessage(text)] },
    config
  );
  const last = result.messages[result.messages.length - 1] as BaseMessage;
  console.log(`   [${threadId}] 👤 ${text}  →  🤖 ${String(last.content)}`);
}

/** 查看某个 thread 的持久化状态 */
async function inspect(threadId: string) {
  const snapshot = await conversationGraph.getState({
    configurable: { thread_id: threadId },
  });
  const msgs = snapshot.values.messages as BaseMessage[];
  console.log(`   [${threadId}] 历史共 ${msgs.length} 条消息，累计轮次 ${snapshot.values.turns}`);
}

async function main() {
  banner("示例 5：单例图 + checkpointer（结构复用 + 会话隔离）");

  console.log("🔹 会话 alice（连续 3 轮）");
  await send("alice", "我叫小爱");
  await send("alice", "我刚说我叫什么？");
  await send("alice", "再确认一次");

  console.log("\n🔹 会话 bob（只 1 轮，与 alice 互不影响）");
  await send("bob", "你好，我是 bob");

  console.log("\n🔹 回看两个会话的持久化状态：");
  await inspect("alice");
  await inspect("bob");

  console.log("\n🔹 alice 再来一条（验证它接的是自己的历史，不是 bob 的）");
  await send("alice", "第 4 条了");
  await inspect("alice");
  await inspect("bob");

  console.log("\n📌 小结：");
  console.log("   • 全程只有一张 conversationGraph（模块级编译一次）→ 结构复用；");
  console.log("   • alice 的历史不断累积到 8 条，bob 始终只有 2 条 → 会话按 thread_id 隔离；");
  console.log("   • 换 thread_id 就换了一个「记忆空间」，同一张图不用改动、不用重编译。");
}

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

运行：

bash
npx tsx 10-project-structure/code/05-singleton-with-checkpointer.ts

真实输出：

========================================================================
  示例 5：单例图 + checkpointer（结构复用 + 会话隔离）
========================================================================

🔹 会话 alice（连续 3 轮）
   [alice] 👤 我叫小爱  →  🤖 （本会话第 1 条回复）收到：我叫小爱
   [alice] 👤 我刚说我叫什么？  →  🤖 （本会话第 2 条回复）收到：我刚说我叫什么？
   [alice] 👤 再确认一次  →  🤖 （本会话第 3 条回复）收到：再确认一次

🔹 会话 bob（只 1 轮，与 alice 互不影响）
   [bob] 👤 你好，我是 bob  →  🤖 （本会话第 1 条回复）收到：你好，我是 bob

🔹 回看两个会话的持久化状态：
   [alice] 历史共 6 条消息，累计轮次 3
   [bob] 历史共 2 条消息，累计轮次 1

🔹 alice 再来一条（验证它接的是自己的历史，不是 bob 的）
   [alice] 👤 第 4 条了  →  🤖 （本会话第 4 条回复）收到：第 4 条了
   [alice] 历史共 8 条消息，累计轮次 4
   [bob] 历史共 2 条消息，累计轮次 1

📌 小结：
   • 全程只有一张 conversationGraph（模块级编译一次）→ 结构复用；
   • alice 的历史不断累积到 8 条，bob 始终只有 2 条 → 会话按 thread_id 隔离；
   • 换 thread_id 就换了一个「记忆空间」，同一张图不用改动、不用重编译。

三个证据点：

  • conversationGraph 是模块级常量，全程只编译一次 → 结构复用；
  • alice 的历史从 6 条累加到 8 条，bob 始终停在 2 条 → 会话按 thread_id 隔离；
  • 「第 4 条回复」说明 alice 接的是自己的历史（它知道前面已有 3 条 AI 回复），没有串到 bob 的记忆里。

💡 一句话：图是「模具」，thread_id 是「模具里的那件活儿」。 一个模具可以打无数件产品，彼此互不干扰——这正是「单例图 + checkpointer」的工程含义。

七、本章小结 ​

本章速查表

你想做的事做法
定义状态app/state.ts：Annotation.Root({...}) + 导出 StateType/UpdateType
加一个工具新建 app/tools/xxx.ts，再在 app/tools/index.ts 登记一行
改一个节点只动 app/nodes/<对应文件>.ts
改流程/连线只动 app/graph.ts（唯一 new StateGraph 的地方）
组装图export function buildGraph() { return new StateGraph(...).compile(); }
只编译一次把 buildGraph() 的返回值放到模块顶层常量复用
传运行时参数建图传 contextSchema，调用传 { context }，节点读 runtime.context
会话隔离compile({ checkpointer: new MemorySaver() }) + config.configurable.thread_id
同一张图服务多用户模块级单例图 + 每个用户一个 thread_id

五句话总结

  1. 单文件会崩，不是因为「不好看」，而是改动成本、理解成本、协作成本在累积。
  2. 分层的四个理由：可定位、可复用、可测试、可协作——每一条都对应反例的一个具体痛点。
  3. 结构分层 = 一次只依赖下一层：state ← tools/nodes ← graph ← index，入口文件越薄越好。
  4. 图只编译一次：结构在运行时不变，就该是进程级单例；别在每个请求里重建。
  5. 运行时参数走 context，会话历史走 thread_id：前者不参与状态合并，后者让共享的单例图实现隔离。

动手练习

  1. 给 app/tools/ 加一个 get-time.ts（返回当前时间的工具），走一遍「新增一个工具要改几处」的流程，亲手确认只需要动两个文件。
  2. 把示例 3 的 ROUNDS 改成 2000 再跑一次，观察 A 与 B 的差距是否被放大（提示：编译开销随次数线性累积）。
  3. 在示例 4 的基础上加一个 featureFlag 到 context，让节点在 flag 打开时追加一句「（实验功能已启用）」，验证同一张图能按参数分支。

下一章预告

现在你已经能把一个 LangGraph 应用「摆」得整整齐齐了。但很多读者手里已经有大量 LangChain 写的链和 Agent——总不能全部推倒重来。第 11 章我们讲从 LangChain 迁移到 LangGraph：哪些概念是一一对应的、哪些链可以原样塞进节点、AgentExecutor 怎么换成图、以及一份可操作的迁移清单。带着「结构」这张底牌，迁移会顺得多。

← 上一章：编排设计模式 ｜ 返回总目录 ｜ 下一章：从 LangChain 迁移 →

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

本中文版仅供学习使用