LangGraph 深度实战指南：从基础架构到生产级 AI Agent 工作流构建

在 AI 应用开发中，当流程复杂度突破线性范畴——需要条件分支、循环迭代、状态持久化时，传统的 LangChain 线性管道（pipe）就会陷入代码臃肿、逻辑混乱的困境。而 LangGraph 作为 LangChain 生态专为复杂工作流设计的编排框架，通过图结构建模、原生状态管理、灵活流程控制三大核心能力，为构建生产级 AI Agent 提供了标准化解决方案。

本文将从底层原理到实战落地，全方位拆解 LangGraph 的使用逻辑，补充进阶技巧与工程化实践，帮助开发者从 0 到 1 搭建可复用、可监控的 AI 工作流。

一、LangGraph 核心价值：为何替代传统链式流程

AI 应用的复杂度提升，往往伴随着三大痛点：

1. 流程灵活性不足：线性管道无法实现条件分支（如“根据用户意图选择不同工具”）和循环迭代（如“工具调用失败自动重试”）；
2. 状态管理混乱：多步骤间的数据传递需要手动维护，易出现数据丢失或冗余；
3. 调试成本高昂：复杂流程的执行链路不可见，故障定位困难。

LangGraph 正是针对这些痛点而生，其核心优势可通过下表清晰体现：

能力维度	传统 LangChain 管道	LangGraph 图编排
流程结构	仅支持线性执行	原生支持分支、循环、并发
状态传递	手动参数传递	内置全局状态自动同步
调试体验	黑盒式执行	可视化流程追踪（LangGraph Studio）
生产级适配	需大量自定义封装	支持状态持久化、错误重试

二、LangGraph 核心概念：构建工作流的“三大基石”

LangGraph 的工作流本质是“状态驱动的有向图”，其核心由状态（State）、节点（Node）、边（Edge）三大组件构成，三者协同实现复杂流程的声明式定义。

1. 状态（State）：工作流的“数据中枢”

状态是贯穿整个工作流的共享数据容器，所有节点的输入输出都围绕状态展开。LangGraph 通过 Annotation 定义状态的数据结构和更新规则，解决多节点数据同步的核心问题。

（1）Annotation：状态的“契约定义”

Annotation 相当于状态的“类型说明书”，包含两个核心配置：

• 字段定义：声明状态包含的具体数据项（如用户问题、对话历史、工具调用结果）；
• Reducer 函数：定义每个字段的更新逻辑，这是状态管理的核心。

常见 Reducer 类型及应用场景

Reducer 决定了新数据如何合并到原有状态中，不同场景需匹配不同的 Reducer：

import { Annotation } from "@langchain/langgraph";

// 1. 替换型 Reducer：新值覆盖旧值（适用于单次更新的字段，如当前任务）
const replaceReducer = (prev, next) => next;

// 2. 累加型 Reducer：数组追加（适用于对话历史、工具调用日志等有序列表）
const appendReducer = (prev, next) => [...(prev || []), ...(next || [])];

// 3. 数值累加型 Reducer：适用于计数类字段（如迭代次数、重试次数）
const sumReducer = (prev, next) => (prev || 0) + (next || 0);

// 4. 合并型 Reducer：对象深度合并（适用于配置类字段）
const mergeReducer = (prev, next) => ({ ...prev, ...next });

// 状态契约完整定义
const AgentState = Annotation.Root({
  // 用户输入：替换型
  userInput: Annotation({ reducer: replaceReducer, default: () => "" }),
  // 对话历史：累加型
  messages: Annotation({ reducer: appendReducer, default: () => [] }),
  // 迭代次数：数值累加型
  iterations: Annotation({ reducer: sumReducer, default: () => 0 }),
  // 工具配置：合并型
  toolConfig: Annotation({ reducer: mergeReducer, default: () => ({ timeout: 5000 }) })
});

（2）状态的“读写规则”

节点对状态的操作需遵循两个原则：

• 只读不修改：节点函数接收的是状态的快照，不可直接修改原状态；
• 增量返回：节点只需返回需要更新的字段，LangGraph 会通过 Reducer 自动合并到全局状态。

2. 节点（Node）：工作流的“执行单元”

节点是图中的最小执行模块，本质是一个异步函数，负责接收状态、处理逻辑、返回状态更新。根据职责不同，可分为四类节点：

（1）LLM 决策节点：AI 推理的核心

负责调用大模型进行意图识别、工具选择、回答生成，是 Agent 的“大脑”：

import { ChatOpenAI } from "@langchain/openai";
import { HumanMessage, AIMessage } from "@langchain/core/messages";

// 初始化大模型
const llm = new ChatOpenAI({ model: "gpt-4o-mini", temperature: 0.1 });

// LLM 决策节点：分析用户问题，判断是否需要调用工具
async function llmDecisionNode(state) {
  // 从状态中读取对话历史
  const { messages } = state;

  // 调用大模型
  const response = await llm.invoke([
    new AIMessage("你是一个智能助手，可调用工具解决问题，若无需工具可直接回答"),
    ...messages
  ]);

  // 仅返回需要更新的字段
  return {
    messages: [new AIMessage(response.content)],
    iterations: 1 // 迭代次数+1
  };
}

（2）工具执行节点：对接外部能力

负责调用工具（如计算器、搜索引擎、API），LangGraph 提供 ToolNode 简化工具执行逻辑：

import { ToolNode } from "@langchain/langgraph/prebuilt";
import { DynamicTool } from "@langchain/core/tools";

// 自定义工具：加法计算器
const calculatorTool = new DynamicTool({
  name: "calculator",
  description: "用于计算加减乘除等数学运算",
  func: async (input) => {
    const result = eval(input); // 生产环境需替换为安全计算逻辑
    return `计算结果：${input} = ${result}`;
  }
});

// 自定义工具：天气查询（模拟）
const weatherTool = new DynamicTool({
  name: "weather",
  description: "查询指定城市的天气",
  func: async (city) => {
    const mockData = { 北京: "晴天，15°C", 上海: "多云，20°C" };
    return `${city}的天气为：${mockData[city] || "暂未查询到数据"}`;
  }
});

// 工具集合
const tools = [calculatorTool, weatherTool];

// 工具执行节点：自动处理工具调用
const toolNode = new ToolNode(tools);

（3）数据处理节点：状态的清洗与转换

负责对状态数据进行格式化、校验、过滤，为后续节点提供规范输入：

// 数据处理节点：清洗用户输入，去除无效字符
async function inputCleanNode(state) {
  const { userInput } = state;
  const cleanedInput = userInput.trim().replace(/[\u200B-\u200D\uFEFF]/g, "");
  return { userInput: cleanedInput };
}

（4）路由判断节点：流程的“导航员”

负责根据状态判断下一步流向，是实现分支和循环的关键（也可通过条件边替代）：

// 路由节点：判断是否需要调用工具
async function toolRouteNode(state) {
  const lastMessage = state.messages[state.messages.length - 1];
  // 假设大模型返回的消息中包含 toolCall 字段表示需要调用工具
  const needTool = !!lastMessage.toolCall;
  return { needTool };
}

3. 边（Edge）：工作流的“连接纽带”

边定义了节点间的执行顺序，分为普通边和条件边两类，共同构成工作流的拓扑结构。

（1）普通边：无条件的线性流转

适用于固定顺序的流程，如“输入清洗→LLM 决策”：

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

// 初始化状态图
const graph = new StateGraph(AgentState)
  // 添加节点
  .addNode("inputClean", inputCleanNode)
  .addNode("llmDecision", llmDecisionNode)
  .addNode("toolExecute", toolNode)
  // 普通边：START → 输入清洗 → LLM 决策
  .addEdge(START, "inputClean")
  .addEdge("inputClean", "llmDecision");

（2）条件边：动态分支的核心

根据状态动态选择下一个节点，实现“按需执行”的灵活流程。其核心是路由函数和路由映射：

// 1. 路由函数：接收状态，返回路由标识
function toolRouter(state) {
  const lastMsg = state.messages[state.messages.length - 1];
  // 检查大模型是否要求调用工具
  if (lastMsg.toolCall) {
    return "call_tool";
  }
  // 检查迭代次数，防止无限循环
  if (state.iterations >= 3) {
    return "max_iter";
  }
  return "direct_answer";
}

// 2. 路由映射：标识 → 目标节点
const routeMap = {
  call_tool: "toolExecute",
  max_iter: "iterEnd",
  direct_answer: END
};

// 3. 添加条件边
graph.addConditionalEdges("llmDecision", toolRouter, routeMap)
  // 工具执行后回到LLM决策节点，形成循环
  .addEdge("toolExecute", "llmDecision")
  // 最大迭代次数节点
  .addNode("iterEnd", async () => ({ messages: [new AIMessage("已达到最大尝试次数，任务终止")] }))
  .addEdge("iterEnd", END);

三、核心流程实战：从分支控制到循环迭代

掌握基础组件后，可通过两个典型场景，深入理解 LangGraph 的流程控制能力。

1. 条件分支：根据用户意图分流处理

场景：用户提问分为“闲聊”“数学计算”“天气查询”三类，需路由到不同节点处理。

（1）完整流程定义

// 1. 定义状态
const ChatState = Annotation.Root({
  userQuery: Annotation({ reducer: (_, x) => x, default: () => "" }),
  answer: Annotation({ reducer: (_, x) => x, default: () => "" }),
  queryType: Annotation({ reducer: (_, x) => x, default: () => "" })
});

// 2. 定义节点
// 意图识别节点
async function intentClassifyNode(state) {
  const { userQuery } = state;
  const type = await llm.invoke([
    new HumanMessage(`判断以下问题类型，仅返回：闲聊/数学计算/天气查询\n问题：${userQuery}`)
  ]);
  return { queryType: type.content.trim() };
}

// 闲聊回复节点
async function chatNode(state) {
  const { userQuery } = state;
  const reply = await llm.invoke([new HumanMessage(`友好回复以下问题：${userQuery}`)]);
  return { answer: reply.content };
}

// 3. 构建图
const chatGraph = new StateGraph(ChatState)
  .addNode("classify", intentClassifyNode)
  .addNode("chat", chatNode)
  .addNode("calc", calculatorTool)
  .addNode("weather", weatherTool)
  .addEdge(START, "classify")
  // 条件分支路由
  .addConditionalEdges("classify", (state) => state.queryType, {
    闲聊: "chat",
    数学计算: "calc",
    天气查询: "weather"
  })
  // 所有分支最终指向END
  .addEdge("chat", END)
  .addEdge("calc", END)
  .addEdge("weather", END);

// 4. 编译执行
const chatApp = chatGraph.compile();
const result = await chatApp.invoke({ userQuery: "北京今天天气怎么样？" });
console.log(result.answer);

2. 循环迭代：工具调用的自动重试机制

场景：工具调用失败时，自动重试最多 2 次，成功后进入回答阶段。

// 1. 扩展状态，增加工具调用状态字段
const RetryState = Annotation.Root({
  messages: Annotation({ reducer: appendReducer, default: () => [] }),
  retryCount: Annotation({ reducer: sumReducer, default: () => 0 }),
  toolResult: Annotation({ reducer: replaceReducer, default: () => null })
});

// 2. 带重试的工具节点
async function retryToolNode(state) {
  const { messages, retryCount } = state;
  const lastMsg = messages[messages.length - 1];

  try {
    const result = await toolNode.invoke({ toolCalls: [lastMsg.toolCall] });
    return { toolResult: result, retryCount: 0 }; // 成功则重置重试次数
  } catch (error) {
    if (retryCount >= 2) {
      throw new Error("工具调用多次失败");
    }
    return { retryCount: 1, messages: [new AIMessage(`工具调用失败，重试中（次数：${retryCount + 1}）`)] };
  }
}

// 3. 循环路由函数
function retryRouter(state) {
  if (state.retryCount >= 2) return "fail";
  if (state.toolResult) return "answer";
  return "retry";
}

// 4. 构建带重试的图
const retryGraph = new StateGraph(RetryState)
  .addNode("llm", llmDecisionNode)
  .addNode("tool", retryToolNode)
  .addNode("answer", async (state) => ({
    messages: [new AIMessage(`最终结果：${state.toolResult}`)]
  }))
  .addNode("fail", async () => ({ messages: [new AIMessage("工具调用失败，无法完成任务")] }))
  .addEdge(START, "llm")
  .addEdge("llm", "tool")
  .addConditionalEdges("tool", retryRouter, {
    retry: "tool",
    answer: "answer",
    fail: "fail"
  })
  .addEdge("answer", END)
  .addEdge("fail", END);

四、进阶实战：构建完整的 ReAct 模式 AI Agent

ReAct（Reasoning + Acting）是 AI Agent 的核心工作模式，通过“推理→工具调用→观察→再推理”的闭环，实现复杂问题的自主解决。结合 LangGraph，可快速搭建生产级 ReAct Agent。

1. ReAct Agent 核心流程

ReAct Agent 的工作流分为五步：

1. 用户输入：接收用户问题；
2. LLM 推理：判断是否需要调用工具，若需要则生成工具调用指令；
3. 工具执行：调用指定工具获取外部数据；
4. 结果观察：将工具结果反馈给 LLM；
5. 循环判断：重复步骤 2-4，直到生成最终回答。

2. 完整代码实现

import { StateGraph, Annotation, START, END } from "@langchain/langgraph";
import { ToolNode } from "@langchain/langgraph/prebuilt";
import { ChatOpenAI } from "@langchain/openai";
import { HumanMessage, AIMessage, ToolMessage } from "@langchain/core/messages";
import { DynamicTool } from "@langchain/core/tools";

// 1. 初始化大模型和工具
const llm = new ChatOpenAI({ model: "gpt-4o-mini", temperature: 0 });
const calculator = new DynamicTool({
  name: "calculator",
  description: "数学计算工具，输入为数学表达式",
  func: async (expr) => String(eval(expr)) // 生产环境需替换为安全计算库
});
const weather = new DynamicTool({
  name: "weather",
  description: "天气查询工具，输入为城市名",
  func: async (city) => {
    const data = { 北京: "晴 12-20°C", 上海: "阴 15-18°C", 广州: "雨 22-25°C" };
    return data[city] || "未查询到该城市天气";
  }
});
const tools = [calculator, weather];
const toolNode = new ToolNode(tools);

// 2. 定义 ReAct Agent 状态
const ReActState = Annotation.Root({
  messages: Annotation({
    reducer: (prev, next) => [...(prev || []), ...(next || [])],
    default: () => []
  }),
  iterations: Annotation({
    reducer: (prev, next) => (prev || 0) + (next || 0),
    default: () => 0
  })
});

// 3. 定义 Agent 决策节点
async function agentNode(state) {
  const { messages, iterations } = state;
  // 绑定工具到LLM，让模型知道可用工具
  const llmWithTools = llm.bindTools(tools);

  // 调用LLM生成决策
  const response = await llmWithTools.invoke(messages);

  return {
    messages: [new AIMessage(response.content, { tool_calls: response.tool_calls })],
    iterations: 1
  };
}

// 4. 定义工具调用后处理节点
async function toolPostNode(state) {
  const { messages } = state;
  // 获取工具调用结果，封装为ToolMessage
  const toolResult = messages[messages.length - 1].content;
  return {
    messages: [new ToolMessage(toolResult, { tool_call_id: "call_001" })]
  };
}

// 5. 定义路由函数：判断是否继续调用工具
function shouldContinue(state) {
  const { messages, iterations } = state;
  const lastMsg = messages[messages.length - 1];

  // 安全限制：最大迭代3次
  if (iterations >= 3) return "end";
  // 若有工具调用指令，则继续调用工具
  if (lastMsg.tool_calls?.length > 0) return "tool";
  // 否则结束流程
  return "end";
}

// 6. 构建 ReAct Agent 图
const reactGraph = new StateGraph(ReActState)
  .addNode("agent", agentNode)
  .addNode("tool", toolNode)
  .addNode("toolPost", toolPostNode)
  .addEdge(START, "agent")
  // 条件路由：判断是否调用工具
  .addConditionalEdges("agent", shouldContinue, {
    tool: "tool",
    end: END
  })
  // 工具调用后先处理结果，再回到Agent节点
  .addEdge("tool", "toolPost")
  .addEdge("toolPost", "agent");

// 7. 编译并执行Agent
const reactAgent = reactGraph.compile();
const result = await reactAgent.invoke({
  messages: [new HumanMessage("北京今天天气怎么样？再计算100+200*3的结果")]
});

// 输出最终回答
console.log(result.messages[result.messages.length - 1].content);

3. 执行流程解析

1. 初始输入：用户提问包含天气查询和数学计算两个需求；
2. Agent 决策：LLM 识别到两个工具调用需求，生成 tool_calls 指令；
3. 工具执行：依次调用 weather 和 calculator 工具，获取结果；
4. 结果反馈：将工具结果封装为 ToolMessage 传回 LLM；
5. 最终回答：LLM 整合工具结果，生成自然语言回答，流程结束。

五、工程化进阶：流式输出与调试监控

1. 流式输出：提升用户体验

LangGraph 支持流式执行，可实时返回每个节点的输出，适用于需要实时反馈的场景：

// 流式执行Agent
for await (const event of reactAgent.stream({
  messages: [new HumanMessage("计算100/2+50，并查询上海天气")]
})) {
  // event 结构：{ 节点名: { 状态更新内容 } }
  const nodeName = Object.keys(event)[0];
  const nodeOutput = event[nodeName];
  console.log(`[${nodeName}]`, nodeOutput);
}

2. LangGraph Studio：可视化调试

LangGraph 提供官方调试工具 LangGraph Studio，可实现：

• 流程可视化：直观查看节点执行顺序和状态流转；
• 断点调试：在指定节点暂停，检查状态数据；
• 历史回溯：复现流程执行过程，定位故障节点。

3. 生产级最佳实践

1. 状态持久化：将状态存储到数据库（如 Redis），实现会话中断恢复；
2. 错误边界：为每个节点添加异常捕获，避免单点故障导致流程终止；
3. 性能监控：记录每个节点的执行耗时，优化瓶颈环节；
4. 工具权限控制：为不同用户分配不同工具调用权限，保障系统安全。

六、总结与学习路径

LangGraph 的核心是“以图建模、状态驱动”，其学习可分为三个阶段：

1. 基础阶段：掌握 State、Node、Edge 三大组件，实现简单线性流程；
2. 进阶阶段：学会条件分支和循环迭代，搭建 ReAct 基础 Agent；
3. 生产阶段：整合流式输出、状态持久化、监控调试，构建企业级 AI 应用。

从链式流程到图编排，LangGraph 不仅是工具的升级，更是 AI 应用开发思维的转变——通过声明式的流程定义，让开发者聚焦业务逻辑，而非流程控制的底层细节，真正实现“用流程图的思维构建 AI 应用”。

除非注明，否则均为李锋镝的博客原创文章，转载必须以链接形式标明本文链接

本文链接：https://www.lifengdi.com/ren-gong-zhi-neng/4624