LangGraph 与多 Agent 架构

开场白：为什么要学这个

如果你已经用 LangChain 写过「一个模型 + 几个工具」的单 Agent 程序，可能已经碰到过这些别扭：

• 一个 prompt 里塞了「你既会查天气、又会讲城市百科、还会写代码」，模型经常串台——让它查天气，它却开始讲百科；
• 想让模型并行干两件事，一个模型做不到；
• 想让模型回头检查自己前面输出的对不对，单 Agent 也没有「第二个人」可以搭把手。

这些场景，正是多 Agent 架构的用武之地。而 LangGraph 就是那个用来「画图」的框架：把一次任务拆成一张有节点、有边、能循环、能暂停的执行图。学会了图，你就能用图把多个 Agent 拼成一支「团队」。

先说结论：为什么要用多 Agent

多 Agent 不是炫技，是有实打实的理由的。三个理由，一个比一个接近真实业务：

1. Prompt 拆分，更纯净、更省 token。与其让一个模型记住所有规则，不如拆给多个 Agent 各记一段。每个 Agent 的 prompt 更短、上下文更干净，模型决策更不容易出错，token 消耗也更少。
2. 并行思考与执行。多个 Agent 可以同时干不同的活，整体响应更快。比如「查北京的天气 + 讲一条杭州的小知识」这两个独立问题，完全可以并行。
3. 基于角色相互讨论、纠错。一个 Agent 负责做，另一个负责审，出错率直线下降。这是单 Agent 给不了的能力。

复杂的 Agent 产品，底层基本都是多 Agent 架构。所以先把「图」学好，多 Agent 只是图的一种拼法。

第一章 LangGraph 是什么：一张地图，四个部件

LangGraph 的核心是一张「地图」，由四个部分紧密协作：

先看一个最小可运行的骨架：

import { Annotation, END, START, StateGraph } from "@langchain/langgraph";

// ① State：定义图全程共享的数据结构
const StateAnnotation = Annotation.Root({
    text: Annotation({
        reducer: (_prev, next) => next, // Reducer：新值怎么合并进旧值（这里直接覆盖）
        default: () => "",             // 初始值
    }),
});

// ② Node：两个节点，各自往 state.text 上追加一段
const step1 = (state) => ({ text: `${state.text} -> step1` });
const step2 = (state) => ({ text: `${state.text} -> step2` });

// ③④ Edge + Graph：把节点用边连起来，编译成应用
const graph = new StateGraph(StateAnnotation)
    .addNode("step1", step1)
    .addNode("step2", step2)
    .addEdge(START, "step1")  // 入口 → step1
    .addEdge("step1", "step2") // step1 → step2
    .addEdge("step2", END)    // step2 → 结束
    .compile();

const result = await graph.invoke({ text: "hello" });
console.log("result:", result); // { text: "hello -> step1 -> step2" }

要点：

• State 是灵魂：所有节点读写同一份数据。Annotation.Root({...}) 声明结构，每个字段用 Annotation({ reducer, default }) 定义。
• reducer 决定怎么合并：(_prev, next) => next 表示「新值覆盖旧值」；如果是对话消息，会用 addMessages 这类 reducer，让新消息追加而非覆盖。
• default 是初始值：图启动时字段的默认状态。
• Node 返回的是「部分更新」：可以只返回 state 的一个字段，LangGraph 负责用 reducer 合并进总状态。
• START / END 是内置哨兵节点：入口和出口，每个图都要连到它们。
• compile() 之后才能 invoke()：一个编译好的图就是一个可执行应用。

第二章 条件路由：让图自己「拐弯」

普通边是「走完 A 一定走 B」，太死板。真实场景里，要根据输入决定走哪条路。这就要用 addConditionalEdges（条件边）。

看一个例子：一个节点 router 判断用户输入是算式还是闲聊，把任务分给 math 或 chat 节点。

import { Annotation, END, START, StateGraph } from "@langchain/langgraph";

const StateAnnotation = Annotation.Root({
    query: Annotation({ reducer: (_prev, next) => next, default: () => "" }),
    route: Annotation({ reducer: (_prev, next) => next, default: () => "chat" }),
    answer: Annotation({ reducer: (_prev, next) => next, default: () => "" }),
});

// 路由节点：只负责「选路」，算出 route 字段
const router = (state) => {
    const isMath = /[+\-*/]/.test(state.query);
    return { route: isMath ? "math" : "chat" };
};

const mathNode = (state) => {
    try { return { answer: String(eval(state.query)) }; }
    catch { return { answer: "表达式无法计算" }; }
};

const chatNode = (state) => ({ answer: `你说的是：${state.query}` });

const graph = new StateGraph(StateAnnotation)
    .addNode("router", router)
    .addNode("math", mathNode)
    .addNode("chat", chatNode)
    .addEdge(START, "router")
    // 条件边：router 跑完后，用回调返回值作为 key，
    // 在 { math, chat } 映射表里找到对应的下一个节点
    .addConditionalEdges("router", (state) => state.route, {
        math: "math",
        chat: "chat",
    })
    .addEdge("math", END)
    .addEdge("chat", END)
    .compile();

console.log(await graph.invoke({ query: "你好" }));   // 走 chat
console.log(await graph.invoke({ query: "10 * 8" })); // 走 math

要点：

• 条件边的三要素：源节点（"router"）+ 一个返回路由 key 的回调函数 + 一个「key → 节点名」的映射表。
• 路由节点只算 key、不决定去向：router 只更新 route 字段，真正「走哪条边」由条件边执行时决定。职责分离，方便调试。
• 分支/循环全靠它：条件边不只用于分支——把映射值指回自己或前一个节点，就形成了循环。

第三章 用条件边实现循环：重试

分支和循环本质是同一件事：条件边指向下一个节点。指向一个新节点就是分支，指回自己是循环。

attempt 节点失败就重试自己，直到成功：

import { Annotation, END, START, StateGraph } from "@langchain/langgraph";

const StateAnnotation = Annotation.Root({
    tries: Annotation({ reducer: (_prev, next) => next, default: () => 0 }),
    ok: Annotation({ reducer: (_prev, next) => next, default: () => false }),
    message: Annotation({ reducer: (_prev, next) => next, default: () => "" }),
});

const attempt = (state) => {
    const tries = state.tries + 1;
    const ok = tries >= 3; // 模拟：第 3 次才成功
    return {
        tries,
        ok,
        message: ok ? `第 ${tries} 次成功` : `第 ${tries} 次失败，继续重试`,
    };
};

const graph = new StateGraph(StateAnnotation)
    .addNode("attempt", attempt)
    .addEdge(START, "attempt")
    .addConditionalEdges("attempt", (state) => (state.ok ? "done" : "retry"), {
        retry: "attempt", // 没成功 → 回到自己，形成循环
        done: END,
    })
    .compile();

console.log(await graph.invoke({ tries: 0 }));
// { tries: 3, ok: true, message: "第 3 次成功" }

要点：

• 循环不是语法糖，就是一条「指回自己」的条件边。retry: "attempt" 让节点反复执行自己。
• State 是循环的计数器：tries 存在 State 里，每次执行 +1，图才能判断「该继续还是该停」。没有 State，循环就是死循环。
• 必须有一个「退出条件」（这里 ok: true 指向 END），否则图永远不结束——写循环时先想好终点。

第四章 Checkpointer：让图「记住上次走到哪」

上面的图每次 invoke() 都从零开始。但真实对话需要跨轮次记忆：同一用户上次聊到哪、状态是什么，下次要接着来。这就是 checkpointer（检查点） 的职责。

用 MemorySaver 保存每次执行后的 state，并按 thread_id 区分不同用户：

import {
    Annotation, END, MemorySaver, START, StateGraph,
} from "@langchain/langgraph";

const StateAnnotation = Annotation.Root({
    visitCount: Annotation({ reducer: (_prev, next) => next, default: () => 0 }),
    message: Annotation({ reducer: (_prev, next) => next, default: () => "" }),
});

function recordVisit(state) {
    const visitCount = state.visitCount + 1;
    const message =
        visitCount === 1
            ? "这是你在本会话里第 1 次进入。"
            : `这是你在本会话里第 ${visitCount} 次进入`;
    return { visitCount, message };
}

const graph = new StateGraph(StateAnnotation)
    .addNode("recordVisit", recordVisit)
    .addEdge(START, "recordVisit")
    .addEdge("recordVisit", END);

const checkpointer = new MemorySaver();
const app = graph.compile({ checkpointer }); // 编译时挂上检查点

// thread_id 是「会话身份」：同一 id 共享状态，不同 id 互不干扰
const user1 = { configurable: { thread_id: "用户-小张" } };
const user2 = { configurable: { thread_id: "用户-小李" } };

await app.invoke({}, user1); // 小张第 1 次
await app.invoke({}, user1); // 小张第 2 次
await app.invoke({}, user1); // 小张第 3 次
await app.invoke({}, user2); // 小李第 1 次
await app.invoke({}, user1); // 小张第 4 次

要点：

• compile({ checkpointer }) 挂上检查点：之后每次 invoke() 结束，当前 state 会被保存。
• thread_id 就是「会话身份证」：放在 configurable 里传给 invoke()。同一个 thread_id 连续执行，状态不断累积；换一个 thread_id，就是全新会话。
• MemorySaver 是内存版，重启进程就丢了；生产环境可换成持久化的 checkpointer（如存数据库），接口一样。

第五章 interrupt：在图中「按暂停」，等人确认再继续

有些流程不能一口气跑完——比如转账前要用户确认。LangGraph 提供了 interrupt()：让图在执行中停下来，把 __interrupt__ 信息抛给外部；外部处理完，再用 Command({ resume }) 把结果塞回去，图继续跑。

展示一笔转账，等用户在终端输入确认后继续：

import { createInterface } from "node:readline/promises";
import {
    Annotation, Command, END, MemorySaver, START, StateGraph, interrupt,
} from "@langchain/langgraph";

const StateAnnotation = Annotation.Root({
    actionSummary: Annotation({ reducer: (_prev, next) => next, default: () => "" }),
    userInput: Annotation({ reducer: (_prev, next) => next, default: () => "" }),
});

const showTransfer = () => ({
    actionSummary: "向张三转账 ¥100（模拟，不会真扣款）",
});

// 停在这里等人输入；resume 的值会写进 userInput
const waitConfirm = (state) => {
    const text = interrupt({
        hint: "终端里输入「确认」或备注后回车，图才会继续",
        actionSummary: state.actionSummary,
    });
    return { userInput: String(text) };
};

const graph = new StateGraph(StateAnnotation)
    .addNode("showTransfer", showTransfer)
    .addNode("waitConfirm", waitConfirm)
    .addEdge(START, "showTransfer")
    .addEdge("showTransfer", "waitConfirm")
    .addEdge("waitConfirm", END)
    .compile({ checkpointer: new MemorySaver() }); // interrupt 必须有 checkpointer

const config = { configurable: { thread_id: "interrupt-demo" } };

// 第一次 invoke：跑到 waitConfirm 就暂停，把 __interrupt__ 返回给外部
const paused = await graph.invoke({}, config);
console.log("待你确认：", paused.__interrupt__?.[0]?.value);

// 真实场景里让用户在界面上确认；demo 里读终端输入
const rl = createInterface({ input: process.stdin, output: process.stdout });
const line = (await rl.question("> ")).trim();
await rl.close();

// 第二次 invoke：带上 resume 命令，图从暂停点继续
const done = await graph.invoke(new Command({ resume: line }), config);
console.log("结果：", done);

要点：

• interrupt() 是同步暂停点：执行到它时，图停住，把传入的对象通过 paused.__interrupt__ 暴露给调用方。
• 必须有 checkpointer：暂停要恢复，前提是得知道「上次停在哪、state 是什么」——所以 interrupt 和 checkpointer 是配套的。
• 恢复方式：对同一个 thread_id 再次 invoke()，传 new Command({ resume: 值 })，图从暂停的节点继续，interrupt() 的返回值就是 resume 的值。
• 典型场景：转账确认、高危操作二次确认、需要人在环路（human-in-the-loop）的审批流程。

第六章 prebuilt 开箱即用：ToolNode、toolsCondition、createAgent

前面都是手写图。但「模型 + 工具循环」是 Agent 的标配动作，LangGraph 已经把常用零件预置好了，你不需要自己画。

6.1 手动版：ToolNode + toolsCondition

一个标准的「模型决定调工具 → 执行工具 → 再回到模型」循环：

import "dotenv/config";

import { HumanMessage } from "@langchain/core/messages";
import { tool } from "@langchain/core/tools";
import {
    END, MessagesAnnotation, START, StateGraph,
} from "@langchain/langgraph";
import { ToolNode, toolsCondition } from "@langchain/langgraph/prebuilt";
import { ChatOpenAI } from "@langchain/openai";
import { z } from "zod";

// 简化版：真实场景换成数据库 / API 查询
const rows = [
    { sku: "SKU-001", name: "无线鼠标", stock: 42 },
    { sku: "SKU-002", name: "机械键盘", stock: 7 },
];
const getProductBySku = (sku) => {
    const row = rows.find((r) => r.sku === String(sku).trim().toUpperCase());
    return row ? JSON.stringify({ found: true, ...row }) : JSON.stringify({ found: false });
};

// ① 定义一个工具：模型只能「描述调用」，真正执行靠这个函数
const getProductStock = tool(
    async ({ sku }) => getProductBySku(sku),
    {
        name: "get_product_stock",
        description: "按 SKU 查商品名与库存，SKU 如 SKU-001。",
        schema: z.object({ sku: z.string().describe("商品 SKU") }),
    }
);

const tools = [getProductStock];
const llm = new ChatOpenAI({
    modelName: process.env.MODEL_NAME,
    apiKey: process.env.OPENAI_API_KEY,
    configuration: { baseURL: process.env.OPENAI_BASE_URL },
}).bindTools(tools); // ② 绑定工具，模型才能「知道」有这些工具可调

// ③ Agent 节点：把消息列表交给模型
async function agent(state) {
    const response = await llm.invoke(state.messages);
    return { messages: response };
}

const toolNode = new ToolNode(tools); // ④ prebuilt：自动执行模型声明的所有工具

const graph = new StateGraph(MessagesAnnotation)
    .addNode("agent", agent)
    .addNode("tools", toolNode)
    .addEdge(START, "agent")
    // ⑤ prebuilt 路由：模型回复带 tool_calls → 去执行工具；没有 → 结束
    .addConditionalEdges("agent", toolsCondition, ["tools", END])
    .addEdge("tools", "agent")
    .compile();

const result = await graph.invoke({
    messages: [new HumanMessage("查一下 SKU-001 的库存还有多少，回答里带上商品名和数字。")],
});

要点：

• MessagesAnnotation：LangGraph 内置的对话消息 State，messages 字段用追加式 reducer，自动累积对话历史。
• ToolNode(tools)：不用自己写「逐个执行工具、把结果包成 ToolMessage 塞回消息列表」——它替你干完了。
• toolsCondition：内置路由函数，看模型回复有没有 tool_calls：有→去 tools，没有→直接 END。这正是 Agent 循环的「刹车」。
• agent 节点：你只需要写「模型调用」，工具循环的胶水代码全被 prebuilt 包掉了。

6.2 开箱版：createAgent —— 一句话创建一个 Agent

如果连上面那张图都不想画，createAgent 直接帮你全部搞定：

import "dotenv/config";

import { HumanMessage } from "@langchain/core/messages";
import { ChatOpenAI } from "@langchain/openai";
import { MemorySaver } from "@langchain/langgraph";
import { createAgent, tool } from "langchain";
import { z } from "zod";

// 简化版：真实场景换成数据库 / API 查询
const rows = [
    { sku: "SKU-001", name: "无线鼠标", stock: 42 },
    { sku: "SKU-002", name: "机械键盘", stock: 7 },
];
const getProductBySku = (sku) => {
    const row = rows.find((r) => r.sku === String(sku).trim().toUpperCase());
    return row ? JSON.stringify({ found: true, ...row }) : JSON.stringify({ found: false });
};

const getProductStock = tool(
    async ({ sku }) => getProductBySku(sku),
    {
        name: "get_product_stock",
        description: "按 SKU 查商品名与库存，SKU 如 SKU-001。",
        schema: z.object({ sku: z.string().describe("商品 SKU") }),
    }
);

const model = new ChatOpenAI({
    modelName: process.env.MODEL_NAME,
    apiKey: process.env.OPENAI_API_KEY,
    configuration: { baseURL: process.env.OPENAI_BASE_URL },
});

// 一句话：模型 + 工具 + 人设 + 记忆，就是一个可用的 Agent
const agent = createAgent({
    model,
    tools: [getProductStock],
    systemPrompt: "你是仓库助手。问库存时必须调用 get_product_stock（模拟数据），禁止编造。",
    checkpointer: new MemorySaver(),
});

const result = await agent.invoke(
    { messages: [new HumanMessage("SKU-002 还剩多少库存？")] },
    { configurable: { thread_id: "demo-thread" } }
);

要点：

• createAgent 是「全家桶」：工具循环、消息 State、模型绑定、checkpointer 全部内置，一个配置对象搞定。
• agent.graph 仍可拿到：想导出图、加自定义节点，agent.graph 就是底层那张图，随时可访问。
• 和手写图的关系：createAgent 只是手写图（6.1 那种）的封装。学手写是为了理解原理，用 createAgent 是为了效率——两者不冲突。

第七章 多 Agent：Supervisor - Worker 架构

有了图，就可以拼「团队」了。最常见的多 Agent 架构是 Supervisor（主管）- Worker（工人）：

• Supervisor（主管节点）：只负责任务分发——看用户问了什么，决定派给哪个 Worker，不自己干活；
• Worker（工人）：各自负责一块具体任务（查天气、讲百科、写代码……），干完把结果交回。

好消息是：不需要手画这张图，@langchain/langgraph-supervisor 包直接封装好了整套架构。

两个 Worker（天气、城市小知识），一个 Supervisor 派单：

import "dotenv/config";

import { HumanMessage } from "@langchain/core/messages";
import { createSupervisor } from "@langchain/langgraph-supervisor";
import { ChatOpenAI } from "@langchain/openai";
import { createAgent, tool } from "langchain";
import { z } from "zod";

// 简化版：真实场景换成天气 / 百科 API
const lookupWeather = (city) => JSON.stringify({ city, summary: "多云转小雨" });
const lookupCityTrivia = (city) =>
    JSON.stringify({ city, trivia: `${city} 是一座历史悠久的城市。` });

const model = new ChatOpenAI({
    modelName: process.env.MODEL_NAME,
    apiKey: process.env.OPENAI_API_KEY,
    configuration: { baseURL: process.env.OPENAI_BASE_URL },
});

// ① Worker A：只查天气
const weatherAgent = createAgent({
    name: "weather_agent",
    description: "专门查天气",
    model,
    tools: [tool(
        async ({ city }) => lookupWeather(city),
        {
            name: "lookup_weather",
            description: "查询某城市当日天气概况（气温区间、天气、空气质量等）。",
            schema: z.object({ city: z.string().describe("城市名，如 杭州") }),
        }
    )],
    systemPrompt: "你只处理天气。用户提到城市时，用 lookup_weather 查询后再用中文简短说明。",
});

// ② Worker B：只讲城市小知识
const triviaAgent = createAgent({
    name: "trivia_agent",
    description: "专门讲与城市相关的小知识；必须调用 lookup_city_trivia。",
    model,
    tools: [tool(
        async ({ city }) => lookupCityTrivia(city),
        {
            name: "lookup_city_trivia",
            description: "查询与某城市相关的一句趣味知识。",
            schema: z.object({ city: z.string().describe("城市名，如 杭州") }),
        }
    )],
    systemPrompt: "你只讲城市小知识。先 lookup_city_trivia，再用人话转述，不要编造工具里没有的内容。",
});

// ③ Supervisor：只负责「选人」，把任务派给合适的 Worker
const workflow = createSupervisor({
    agents: [weatherAgent.graph, triviaAgent.graph],
    llm: model,
    prompt: `你是调度员，只负责选人，不要自己报气温、也不要自己讲城市百科。

- 问天气、气温、下不下雨、空气 → 用 weather_agent
- 问小知识、名胜、历史、一句介绍 → 用 trivia_agent
`,
});

const app = workflow.compile();

const input = {
    messages: [new HumanMessage("查一下北京的天气，再讲一条和杭州有关的小知识。")],
};

要点：

• createSupervisor({ agents, llm, prompt })：三个关键参数——agents 是 Worker 图数组（把 createAgent 的 agent.graph 传进去）、llm 是 Supervisor 用的模型、prompt 是给 Supervisor 的「派单规则」。
• 每个 Worker 是独立 Agent：createAgent 各自带自己的工具、人设、记忆，职责单一、prompt 干净——这正是第一章说的多 Agent 第一理由。
• Supervisor 的 prompt 是「选人手册」：它只学「什么问题找谁」，不学具体怎么干活，所以不会串台。
• 「查天气 + 讲小知识」这种复合问题：Supervisor 会识别出两个意图，把两个子任务分别派给两个 Worker，再把结果汇总——单 Agent 很难做得这么干净。

第八章 调试多 Agent：用 stream 看图是怎么跑的

多 Agent 的流转过程是黑盒，光看最终结果不够。LangGraph 提供了 stream()，可以像「直播」一样看到图的执行过程，常用两种模式：

接着上一个 Supervisor 例子，用 updates 收集节点执行路径，用 values 拿最终 state：

const nodePath = [];
let finalState = null;
const stream = await app.stream(input, { streamMode: ["updates", "values"] });
for await (const event of stream) {
    const [mode, payload] = event;
    if (mode === "updates" && payload && typeof payload === "object") {
        nodePath.push(...Object.keys(payload)); // 谁跑了，就把谁记下来
    } else if (mode === "values") {
        finalState = payload; // 完整 state
    }
}

console.log("路径:", nodePath.join(" → "));
// 例如：supervisor → weather_agent → supervisor → trivia_agent → supervisor

要点：

• streamMode: ["updates", "values"] 可同时要两种，事件被解构成 [mode, payload] 二元组，按 mode 分流处理。
• updates 是增量：每次事件代表「刚跑完的那个节点」的输出，key 是节点名——用它就能拼出一条执行路径。
• values 是全量：每次事件都是「到目前为止的完整 state」，最后一次的值就是最终结果。
• 打印太多？换断点调试：如果节点多、日志刷屏，直接断点调试，一步步看多 Agent 的流转过程，比看日志更清晰。

结尾：怎么选、怎么用

学完这一篇，你手里的工具是这样的：

一条心法总结：

LangGraph 给你一张「图」，图上的 State 是黑板、Node 是干活的人、Edge 是路、条件边让图能拐弯、checkpointer 让图有记忆、interrupt 让图会停下来等人——把这张图弄明白，多 Agent 不过是你给每个角色各画了一张图，再让一个 Supervisor 把它们串起来。

从「一个模型 + 一个 prompt」到「一张图」，再到「一支团队」，复杂 Agent 产品的底层也就无非如此。下一步，把示例里的 Worker 换成你自己的业务 Agent，加第三个、第四个 Worker，你就有自己的多 Agent 应用了。