LangGraph 实际应用：状态、节点、边以及五种智能体模式

回顾：上一篇的内容

在上一篇文章中，我们阐述了其概念基础：一个AI智能体将大脑（即大语言模型）与负责协调工具和内存的调度器相结合，整个系统还受到某种形式的监督机制管控。我们还了解到LangGraph会将这一结构转化为图——节点代表任务单元，边控制它们之间的数据流，而一个共享的状态对象会在整个系统中传递，并在每一步通过归约器进行合并。

这次，我们将从理论走向实际实现。读完这篇文章后，你将能够：

• 使用Annotation声明状态，包括自定义归约器
• 通过普通边和条件边将节点连接起来

将上一篇文章中提到的五种工作流模式——提示词链式处理、路由、并行化、协调器-工作者模型以及评估器-优化器——全部实现为可运行的 JavaScript 代码

为每种模式提供示意图，清晰展示各组件之间的连接方式

以下所有代码均基于 @langchain/langgraph 和普通 JavaScript 编写（这些模式在 TypeScript 中同样适用，只需为 Annotation<T> 添加类型参数即可）。

npm install @langchain/langgraph @langchain/core @langchain/openai

1. 状态与注解——共享笔记本

在任何一个节点开始运行之前，LangGraph 需要先了解通过图结构传递的数据的结构。Annotation 就是用于描述这种结构的工具。

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

const WorkflowState = Annotation.Root({
  query: Annotation(),
  searchResults: Annotation(),
  draft: Annotation(),
  finalAnswer: Annotation(),
});

在 Annotation.Root({...}) 中定义的每个键都会成为共享状态对象中的一个字段。任何节点都可以读取这些字段，而当它返回时只需包含实际要更新的字段——LangGraph会负责将这部分结果合并回完整状态中。

Reducer：决定如何合并更新

除非另有指定，否则节点的返回值会完全替换对应的字段。对于像 draft 这样只需关注最新版本的场景，这种方式完全可行。但对于需要累积消息列表的情况，直接覆盖会导致之前所有内容被清除。这时就需要reducer来帮忙——它是一个定义如何将现有值与新传入的更新内容进行合并的函数。

const AgentState = Annotation.Root({
  // Default behavior: last write wins
  draft: Annotation(),

  // Custom reducer: append instead of overwrite
  messages: Annotation({
    reducer: (current, update) => [...current, ...update],
    default: () => [],
  }),
});

如此设置后，如果nodeA返回{ messages: ["hi"] }而nodeB返回{ messages: ["hello"] }，reducer会将它们合并为["hi", "hello"]，而不会让其中一个覆盖另一个。这种机制对于安全地运行并行分支至关重要（见下方的模式3）——两个节点可以同时向同一个状态键写入数据而不会发生冲突。

2. 节点——工作单元

节点本质上只是一个函数——无论是同步的还是异步的——它接收当前状态并返回对该状态的局部更新。

async function draftNode(state) {
  const response = await llm.invoke(`Write a short draft about: ${state.query}`);
  return { draft: response.content };
}

你可以使用addNode在图中注册它：

const graph = new StateGraph(WorkflowState)
  .addNode("draft", draftNode);

请注意，节点本身并不知道周围的图结构。它只是接收状态并产生更新结果。将节点连接起来完全是边的职责。

3. 边——流程的连接

边决定了执行顺序。LangGraph支持三种类型的边：

普通边——从一个节点到另一个节点的固定、无条件的路径。

graph.addEdge(START, "draft");   // entry point: where the graph starts
graph.addEdge("draft", "polish");
graph.addEdge("polish", END);    // exit point: where the graph stops

条件边——路由函数会检查当前状态，从而决定下一个执行哪个节点。

graph.addConditionalEdges("classify", (state) => {
  if (state.category === "billing") return "handleBilling";
  if (state.category === "technical") return "handleTechnical";
  return "handleGeneral";
});

并行（扇出）边——当一个节点拥有多条普通的出边时，所有这些边的目标节点会同时执行。

graph.addEdge("orchestrator", "workerA");
graph.addEdge("orchestrator", "workerB");
graph.addEdge("orchestrator", "workerC");

每个图都必须至少包含一条从START出发的边，以及一条最终能到达END的路径，否则LangGraph将无法确定执行应从何处开始或结束。

既然状态、节点和边都已介绍完毕，你就拥有了构建之前所提到的五种工作流模式所需的所有基础元素。

4. 模式1 — 提示词链式处理

概念：将大型任务拆分为一系列线性排列的LLM调用，每一步都基于前一步的结果进行推进。

START → outline → draft → polish → END

const ChainState = Annotation.Root({
  topic: Annotation(),
  outline: Annotation(),
  draft: Annotation(),
  final: Annotation(),
});

async function outlineNode(state) {
  const res = await llm.invoke(`Create a short outline for: ${state.topic}`);
  return { outline: res.content };
}async function draftNode(state) {
  const res = await llm.invoke(`Write a draft based on this outline:\n${state.outline}`);
  return { draft: res.content };
}async function polishNode(state) {
  const res = await llm.invoke(`Polish and tighten this draft:\n${state.draft}`);
  return { final: res.content };
}const chainGraph = new StateGraph(ChainState)
  .addNode("outline", outlineNode)
  .addNode("draft", draftNode)
  .addNode("polish", polishNode)
  .addEdge(START, "outline")
  .addEdge("outline", "draft")
  .addEdge("draft", "polish")
  .addEdge("polish", END)
  .compile();const result = await chainGraph.invoke({ topic: "the history of tea" });
console.log(result.final);

这里的每个节点仅读取其前一个节点写入的字段——不存在分支或循环，只是从开始到结束的直线型流程。

5. 模式2 — 路由选择

概念：首先对输入内容进行分类，然后将其转发到若干专用处理路径中的一条。

┌──> handleBilling   ──┐
START → classify ───┼──> handleTechnical ─┼──> END
                    └──> handleGeneral   ──┘

const RouteState = Annotation.Root({
  message: Annotation(),
  category: Annotation(),
  response: Annotation(),
});

async function classifyNode(state) {
  const res = await llm.invoke(
    `Classify this message as "billing", "technical", or "general". Only reply with one word.\n${state.message}`
  );
  return { category: res.content.trim().toLowerCase() };
}async function handleBilling(state) {
  return { response: `[Billing team] ${await llm.invoke(state.message).then(r => r.content)}` };
}
async function handleTechnical(state) {
  return { response: `[Tech support] ${await llm.invoke(state.message).then(r => r.content)}` };
}
async function handleGeneral(state) {
  return { response: `[General desk] ${await llm.invoke(state.message).then(r => r.content)}` };
}const routeGraph = new StateGraph(RouteState)
  .addNode("classify", classifyNode)
  .addNode("handleBilling", handleBilling)
  .addNode("handleTechnical", handleTechnical)
  .addNode("handleGeneral", handleGeneral)
  .addEdge(START, "classify")
  .addConditionalEdges("classify", (state) => {
    if (state.category === "billing") return "handleBilling";
    if (state.category === "technical") return "handleTechnical";
    return "handleGeneral";
  })
  .addEdge("handleBilling", END)
  .addEdge("handleTechnical", END)
  .addEdge("handleGeneral", END)
  .compile();

传递给 addConditionalEdges 的函数需要返回下一个要访问的节点的名称——这就是让图能够动态选择路径而非始终沿用同一条路线的机制。

6. 模式3 — 并行化

概念：同时对任务的不同部分发起多个大语言模型调用，再将返回结果合并起来。

┌──> summarizeDocA ──┐
START → split ──┼──> summarizeDocB ──┼──> merge → END
                └──> summarizeDocC ──┘

const ParallelState = Annotation.Root({
  documents: Annotation(),
  summaries: Annotation({
    reducer: (current, update) => [...current, ...update],
    default: () => [],
  }),
  finalSummary: Annotation(),
});

function splitNode(state) {
  return {}; // no-op entry point, just fans out to the summarizers below
}async function summarizeDocA(state) {
  const res = await llm.invoke(`Summarize:\n${state.documents[0]}`);
  return { summaries: [res.content] };
}
async function summarizeDocB(state) {
  const res = await llm.invoke(`Summarize:\n${state.documents[1]}`);
  return { summaries: [res.content] };
}
async function summarizeDocC(state) {
  const res = await llm.invoke(`Summarize:\n${state.documents[2]}`);
  return { summaries: [res.content] };
}async function mergeNode(state) {
  const res = await llm.invoke(`Combine these summaries into one:\n${state.summaries.join("\n---\n")}`);
  return { finalSummary: res.content };
}const parallelGraph = new StateGraph(ParallelState)
  .addNode("split", splitNode)
  .addNode("summarizeDocA", summarizeDocA)
  .addNode("summarizeDocB", summarizeDocB)
  .addNode("summarizeDocC", summarizeDocC)
  .addNode("merge", mergeNode)
  .addEdge(START, "split")
  .addEdge("split", "summarizeDocA")
  .addEdge("split", "summarizeDocB")
  .addEdge("split", "summarizeDocC")
  .addEdge("summarizeDocA", "merge")
  .addEdge("summarizeDocB", "merge")
  .addEdge("summarizeDocC", "merge")
  .addEdge("merge", END)
  .compile();

注意绑定在 summaries 上的归约器：如果没有它，最后一个完成的并行分支会直接覆盖之前的结果，而无法让三个输出安全地累积到数组中。

7. 第4种模式——协调器-工作节点

概念：通过一次LLM调用将整体任务拆分为多个较小的子任务，再将每个子任务分配给一个工作节点处理，最后通过一步操作将所有结果整合在一起。这看起来很像并行化，但这里的协调器会根据实际情况动态决定启动多少个工作节点以及每个节点的任务，而非提前固定好。

START → orchestrator ──> worker(s), one per subtask ──> compile → END

const OrchestratorState = Annotation.Root({
  task: Annotation(),
  subtasks: Annotation(),
  sections: Annotation({
    reducer: (current, update) => [...current, ...update],
    default: () => [],
  }),
  report: Annotation(),
});

async function orchestratorNode(state) {
  const res = await llm.invoke(
    `Break this task into 3-5 short subtasks, one per line:\n${state.task}`
  );
  const subtasks = res.content.split("\n").filter(Boolean);
  return { subtasks };
}async function workerNode(state) {
  // In a real app you'd invoke one worker per subtask (e.g. via Send / map-reduce);
  // here's the simple per-subtask logic each worker runs:
  const results = await Promise.all(
    state.subtasks.map((subtask) => llm.invoke(`Write a section about: ${subtask}`))
  );
  return { sections: results.map((r) => r.content) };
}async function compileNode(state) {
  const res = await llm.invoke(`Compile these sections into one report:\n${state.sections.join("\n\n")}`);
  return { report: res.content };
}const orchestratorGraph = new StateGraph(OrchestratorState)
  .addNode("orchestrator", orchestratorNode)
  .addNode("worker", workerNode)
  .addNode("compile", compileNode)
  .addEdge(START, "orchestrator")
  .addEdge("orchestrator", "worker")
  .addEdge("worker", "compile")
  .addEdge("compile", END)
  .compile();

如果需要真正的动态分支——即每次运行时工作节点的数量实际上都会变化——LangGraph.js提供了Send API，允许在执行时为每个子任务启动一个工作节点。这一机制值得专门详细阐述，但此处展示的设置已能满足绝大多数日常需求，比如撰写报告或构建研究智能体。

8. 模式5 — 评估器-优化器

概念：一个节点生成输出，另一个节点对其进行检查，条件边会将执行流程重新导向生成器，直到该输出被接受。正是这种模式让LangGraph对循环的支持得以真正发挥作用。

START → generate → evaluate ──(fail)──> generate  (loop)
                       │
                     (pass)
                       ↓
                      END

const LoopState = Annotation.Root({
  prompt: Annotation(),
  code: Annotation(),
  feedback: Annotation(),
  passed: Annotation(),
  attempts: Annotation({
    reducer: (current, update) => current + update,
    default: () => 0,
  }),
});

async function generateNode(state) {
  const context = state.feedback ? `\nPrevious feedback: ${state.feedback}` : "";
  const res = await llm.invoke(`Write code for: ${state.prompt}${context}`);
  return { code: res.content, attempts: 1 };
}async function evaluateNode(state) {
  const res = await llm.invoke(
    `Review this code for bugs. Reply "PASS" if it's correct, otherwise explain what's wrong:\n${state.code}`
  );
  const passed = res.content.trim().startsWith("PASS");
  return { passed, feedback: passed ? null : res.content };
}const loopGraph = new StateGraph(LoopState)
  .addNode("generate", generateNode)
  .addNode("evaluate", evaluateNode)
  .addEdge(START, "generate")
  .addEdge("generate", "evaluate")
  .addConditionalEdges("evaluate", (state) => {
    if (state.passed || state.attempts >= 4) return END; // safety cap on retries
    return "generate";
  })
  .compile();

请注意，attempts字段使用的是累加器而非替换值——这也起到了约束作用。如果没有重试次数的限制，一个永远不认可输出的评估器可能会让循环无限持续下去。

9. 编译与运行这些模型

无论你构建哪种模式，完成方式都是一样的：

const app = graph.compile();
const finalState = await app.invoke({ /* initial state fields */ });

调用 .compile() 可以检查每个节点是否确实可访问，确认 START 和 END 是否正确连接，并将图结构描述转换为可执行对象。随后 .invoke() 会从头到尾运行该图结构并返回最终状态——这正是本系列前面概念性介绍过的执行周期。

总结

在本次讲解过程中，理论中的概念被转化为了可运行的代码：使用Annotation.Root来定义状态，普通函数作为节点，addEdge和addConditionalEdges用于控制流程，而还原器则能防止并发或重复的更新相互干扰。仅凭这四部分，所涵盖的五种工作流模式——链式处理、路由、并行化、协调者-工作者模式以及评估器-优化器模式——其实都是对相同基础工具的不同组合而已。

本系列的下一部分将引入外部工具，添加内存以便状态在会话之间保持不变，并增设一个步骤，让用户在操作继续之前进行审核和批准——这些都是将这些图结构转化为真正智能体所必需的要素。