LangGraph 核心思维

了解如何思考使用 LangGraph 构建智能体

当你使用 LangGraph 构建智能体时，首先将其分解为称为节点（nodes）的离散步骤。然后，描述每个节点的决策与状态转移。最后，通过每个节点都可以读取和写入的共享状态（state）将节点连接在一起。

在本演练中，我们将引导你完成使用 LangGraph 构建客户支持邮件智能体的思考过程。

从你想要自动化的流程开始

假设你需要构建一个处理客户支持邮件的 AI 智能体。你的产品团队向你提出了以下需求：

txt

智能体应当：

- 读取传入的客户邮件
- 按紧急程度和主题对其进行分类
- 搜索相关文档以回答问题
- 起草适当的回复
- 将复杂问题上报给人工客服
- 必要时安排后续跟进

需要处理的示例场景：

1. 简单的产品问题：“如何重置我的密码？”
2. 缺陷报告：“当我选择 PDF 格式时，导出功能会崩溃”
3. 紧急计费问题：“我的订阅被重复扣费了！”
4. 功能建议：“可以在移动应用中添加深色模式吗？”
5. 复杂的工程技术问题：“我们的 API 集成偶发失败并报 504 错误”

要在 LangGraph 中实现智能体，你通常会遵循相同的五个步骤。

第 1 步：将你的工作流映射为离散步骤

首先识别流程中各个独立的步骤。每个步骤将成为一个节点（执行某项具体事情的函数）。然后，勾勒出这些步骤如何相互连接。

图中的箭头显示了可能的路径，但走向哪条路径的实际决策发生在每个节点的内部。

现在我们已经确定了工作流中的组件，接下来了解每个节点需要执行的操作：

Read Email: 提取并解析邮件内容
Classify Intent: 使用 LLM 对紧急程度和主题进行分类，然后路由到适当的操作
Doc Search: 查询知识库以获取相关信息
Bug Track: 在跟踪系统中创建或更新工单
Draft Reply: 生成适当的回复
Human Review: 上报给人工客服进行审批或处理
Send Reply: 发送邮件回复

TIP

请注意，某些节点会做出关于下一步流向的决策（Classify Intent、Draft Reply、Human Review），而其他节点总是流向相同的下一步（Read Email 总是流向 Classify Intent，Doc Search 总是流向 Draft Reply）。

第 2 步：确定每个步骤需要执行的操作

对于图中的每个节点，确定它代表的操作类型以及正常工作所需的上下文。

LLM 步骤
当你需要理解、分析、生成文本或做出推理决策时使用

数据步骤
当你需要从外部来源检索信息时使用

操作步骤
当你需要执行外部操作时使用

用户输入步骤
当你需要人工干预时使用

LLM 步骤

当某个步骤需要理解、分析、生成文本或做出推理决策时：

分类意图（Classify intent）
静态上下文（Prompt）：分类类别、紧急度定义、响应格式
动态上下文（来自 State）：邮件内容、发件人信息
预期输出：决定路由走向的结构化分类结果

起草回复（Draft reply）
静态上下文（Prompt）：语气准则、公司政策、回复模板
动态上下文（来自 State）：分类结果、检索结果、客户历史记录
预期输出：准备好供审核的专业邮件回复

数据步骤

当某个步骤需要从外部来源检索信息时：

文档检索（Document search）
参数：根据意图和主题构建的查询
重试策略：是，对瞬时故障使用指数退避重试
缓存：可以缓存常见查询以减少 API 调用

客户历史查询（Customer history lookup）
参数：来自状态的客户邮箱或 ID
重试策略：是，但如果不可用则回退到基本信息
缓存：是，带有生存时间（TTL）以平衡新鲜度与性能

操作步骤

当某个步骤需要执行外部操作时：

发送回复（Send reply）
节点执行时机：获得批准后（人工或自动）
重试策略：是，对网络问题使用指数退避
不应缓存：每次发送都是唯一的操作

工单跟踪（Bug track）
节点执行时机：当意图为“bug”时始终执行
重试策略：是，确保不丢失缺陷报告至关重要
返回：要包含在回复中的工单 ID

用户输入步骤

当某个步骤需要人工干预时：

人工审核节点（Human review node）
决策上下文：原始邮件、回复草稿、紧急程度、分类
预期输入格式：批准布尔值加上可选的编辑后回复
触发时机：高紧急度、复杂问题或质量疑虑

第 3 步：设计状态

状态是智能体中所有节点均可访问的共享记忆。把它想象成你的智能体在处理过程中用来跟踪它所学到和决定的一切的笔记本。

什么属于状态？

针对每项数据问自己以下问题：

包含在状态中
它是否需要跨步骤持久化？如果是，则放入状态中。

不要存储
你是否可以从其他数据中推导出来？如果是，请在需要时进行计算，而不是将其存储在状态中。

对于我们的邮件智能体，我们需要跟踪：

原始邮件和发件人信息（稍后无法重新构建）
分类结果（后续/下游多个节点需要）
搜索结果和客户数据（重新获取开销昂贵）
回复草稿（需要在审核过程中持久存在）
执行元数据（用于调试和恢复）

保持状态为原始数据，按需格式化提示词

TIP

关键原则：你的状态应存储原始数据，而不是格式化文本。在需要时在节点内格式化提示词。

这种分离意味着：

不同的节点可以根据自己的需求对相同的数据进行不同的格式化
你可以更改提示词模板而无需修改状态架构（schema）
调试更清晰——你可以确切看到每个节点收到了什么数据
你的智能体可以演进而不会破坏现有状态

让我们定义状态：

typescript

import { StateSchema } from "@langchain/langgraph";
import * as z from "zod";

// 定义邮件分类结构
const EmailClassificationSchema = z.object({
  intent: z.enum(["question", "bug", "billing", "feature", "complex"]),
  urgency: z.enum(["low", "medium", "high", "critical"]),
  topic: z.string(),
  summary: z.string(),
});

const EmailAgentState = new StateSchema({
  // 原始邮件数据
  emailContent: z.string(),
  senderEmail: z.string(),
  emailId: z.string(),

  // 分类结果
  classification: EmailClassificationSchema.optional(),

  // 原始检索/API 结果
  searchResults: z.array(z.string()).optional(),  // 原始文档块列表
  customerHistory: z.record(z.string(), z.any()).optional(),  // 来自 CRM 的原始客户数据

  // 生成的内容
  responseText: z.string().optional(),
});

type EmailClassificationType = z.infer<typeof EmailClassificationSchema>;

请注意，状态仅包含原始数据——没有提示词模板，没有格式化字符串，没有说明指令。分类输出直接来自 LLM，存储为单个对象。

第 4 步：构建节点

现在我们将每个步骤实现为一个函数。LangGraph 中的节点就是一个接收当前状态并返回对其更新的 JavaScript 函数。

妥善处理错误

不同的错误需要不同的处理策略：

错误类型	谁来修复	策略	适用场景
瞬时错误（网络问题、速率限制）	系统（自动）	重试策略（Retry policy）	通常在重试后解决的临时故障
LLM 可恢复错误（工具失败、解析问题）	LLM	将错误存入状态并循环回	LLM 可以看到错误并调整其策略
用户可修复错误（信息缺失、指令不明确）	人工	使用 `interrupt()` 暂停	需要用户输入才能继续
重试耗尽后的可恢复故障	开发者（声明式）	`error_handler`	重试耗尽后运行补偿/恢复分支
意外错误	开发者	向上抛出冒泡	需要调试的未知问题

瞬时错误

添加重试策略以自动重试网络问题和速率限制。

typescript

import type { RetryPolicy } from "@langchain/langgraph";

workflow.addNode(
  "searchDocumentation",
  searchDocumentation,
  {
    retryPolicy: { maxAttempts: 3, initialInterval: 1.0 },
  },
);

LLM 可恢复错误

将错误存储在状态中并回环，以便 LLM 可以查看出了什么问题并重试：

typescript

import { Command, GraphNode } from "@langchain/langgraph";

const executeTool: GraphNode<typeof State> = async (state, config) => {
  try {
    const result = await runTool(state.toolCall);
    return new Command({
      update: { toolResult: result },
      goto: "agent",
    });
  } catch (error) {
    // 让 LLM 看到出了什么问题并重试
    return new Command({
      update: { toolResult: `Tool error: ${error}` },
      goto: "agent"
    });
  }
}

用户可修复错误

在需要时（例如账号 ID、订单号或澄清说明）暂停并收集来自用户的信息：

typescript

import { Command, GraphNode, interrupt } from "@langchain/langgraph";

const lookupCustomerHistory: GraphNode<typeof State> = async (state, config) => {
  if (!state.customerId) {
    const userInput = interrupt({
      message: "Customer ID needed",
      request: "Please provide the customer's account ID to look up their subscription history",
    });
    return new Command({
      update: { customerId: userInput.customerId },
      goto: "lookupCustomerHistory",
    });
  }
  // 现在继续执行查找
  const customerData = await fetchCustomerHistory(state.customerId);
  return new Command({
    update: { customerHistory: customerData },
    goto: "draftResponse",
  });
}

意外错误

让它们冒泡以便调试。不要捕获你无法处理的异常：

typescript

import { Command, GraphNode } from "@langchain/langgraph";

const sendReply: GraphNode<typeof EmailAgentState> = async (state, config) => {
  try {
    await emailService.send(state.responseText);
  } catch (error) {
    throw error;  // 暴露意外错误
  }
}

Saga / 补偿

重试耗尽后，运行恢复函数以更新状态并路由到补偿分支。

实现我们的邮件智能体节点

我们将每个节点实现为一个简单的函数。记住：节点接收状态、执行工作并返回更新。

读取与分类节点

typescript

import { StateGraph, START, END, GraphNode, Command } from "@langchain/langgraph";
import { HumanMessage } from "@langchain/core/messages";
import { ChatAnthropic } from "@langchain/anthropic";

const llm = new ChatAnthropic({ model: "claude-sonnet-4-6" });

const readEmail: GraphNode<typeof EmailAgentState> = async (state, config) => {
  // 提取并解析邮件内容
  // 在生产中，这将连接到你的邮件服务
  console.log(`Processing email: ${state.emailContent}`);
  return {};
}

const classifyIntent: GraphNode<typeof EmailAgentState> = async (state, config) => {
  // 使用 LLM 对邮件意图和紧急程度进行分类，然后相应地路由

  // 创建返回 EmailClassification 对象的结构化 LLM
  const structuredLlm = llm.withStructuredOutput(EmailClassificationSchema);

  // 按需格式化提示词，不存储在状态中
  const classificationPrompt = `
  Analyze this customer email and classify it:

  Email: ${state.emailContent}
  From: ${state.senderEmail}

  Provide classification including intent, urgency, topic, and summary.
  `;

  // 直接获取结构化响应对象
  const classification = await structuredLlm.invoke(classificationPrompt);

  // 根据分类决定下一个节点
  let nextNode: "searchDocumentation" | "humanReview" | "draftResponse" | "bugTracking";

  if (classification.intent === "billing" || classification.urgency === "critical") {
    nextNode = "humanReview";
  } else if (classification.intent === "question" || classification.intent === "feature") {
    nextNode = "searchDocumentation";
  } else if (classification.intent === "bug") {
    nextNode = "bugTracking";
  } else {
    nextNode = "draftResponse";
  }

  // 将分类结果作为单个对象存储在状态中
  return new Command({
    update: { classification },
    goto: nextNode,
  });
}

搜索与工单跟踪节点

typescript

import { Command, GraphNode } from "@langchain/langgraph";

const searchDocumentation: GraphNode<typeof EmailAgentState> = async (state, config) => {
  // 在知识库中搜索相关信息

  // 根据分类构建搜索查询
  const classification = state.classification!;
  const query = `${classification.intent} ${classification.topic}`;

  let searchResults: string[];

  try {
    // 在此处实现你的搜索逻辑
    // 存储原始搜索结果，而不是格式化文本
    searchResults = [
      "Reset password via Settings > Security > Change Password",
      "Password must be at least 12 characters",
      "Include uppercase, lowercase, numbers, and symbols",
    ];
  } catch (error) {
    // 对于可恢复的搜索错误，存储错误并继续
    searchResults = [`Search temporarily unavailable: ${error}`];
  }

  return new Command({
    update: { searchResults },  // 存储原始结果或错误
    goto: "draftResponse",
  });
}

const bugTracking: GraphNode<typeof EmailAgentState> = async (state, config) => {
  // 创建或更新缺陷跟踪工单

  // 在你的缺陷跟踪系统中创建工单
  const ticketId = "BUG-12345";  // 将通过 API 创建

  return new Command({
    update: { searchResults: [`Bug ticket ${ticketId} created`] },
    goto: "draftResponse",
  });
}

响应生成与人工复核节点

typescript

import { Command, interrupt } from "@langchain/langgraph";

const draftResponse: GraphNode<typeof EmailAgentState> = async (state, config) => {
  // 使用上下文生成回复，并根据质量路由

  const classification = state.classification!;

  // 按需从原始状态数据格式化上下文
  const contextSections: string[] = [];

  if (state.searchResults) {
    // 为提示词格式化搜索结果
    const formattedDocs = state.searchResults.map(doc => `- ${doc}`).join("\n");
    contextSections.push(`Relevant documentation:\n${formattedDocs}`);
  }

  if (state.customerHistory) {
    // 为提示词格式化客户数据
    contextSections.push(`Customer tier: ${state.customerHistory.tier ?? "standard"}`);
  }

  // 构建包含格式化上下文的提示词
  const draftPrompt = `
  Draft a response to this customer email:
  ${state.emailContent}

  Email intent: ${classification.intent}
  Urgency level: ${classification.urgency}

  ${contextSections.join("\n\n")}

  Guidelines:
  - Be professional and helpful
  - Address their specific concern
  - Use the provided documentation when relevant
  `;

  const response = await llm.invoke([new HumanMessage(draftPrompt)]);

  // 根据紧急程度和意图判断是否需要人工复核
  const needsReview = (
    classification.urgency === "high" ||
    classification.urgency === "critical" ||
    classification.intent === "complex"
  );

  // 路由到适当的下一个节点
  const nextNode = needsReview ? "humanReview" : "sendReply";

  return new Command({
    update: { responseText: response.content.toString() },  // 仅存储原始回复
    goto: nextNode,
  });
}

const humanReview: GraphNode<typeof EmailAgentState> = async (state, config) => {
  // 使用 interrupt 暂停以进行人工复核，并根据决策路由
  const classification = state.classification!;

  // interrupt() 必须放在前面 - 它之前的任何代码在恢复时都会重新运行
  const humanDecision = interrupt({
    emailId: state.emailId,
    originalEmail: state.emailContent,
    draftResponse: state.responseText,
    urgency: classification.urgency,
    intent: classification.intent,
    action: "Please review and approve/edit this response",
  });

  // 现在处理人类的决策
  if (humanDecision.approved) {
    return new Command({
      update: { responseText: humanDecision.editedResponse || state.responseText },
      goto: "sendReply",
    });
  } else {
    // 拒绝意味着人类将直接接管处理
    return new Command({ update: {}, goto: END });
  }
}

const sendReply: GraphNode<typeof EmailAgentState> = async (state, config) => {
  // 发送邮件回复
  // 与邮件服务集成
  console.log(`Sending reply: ${state.responseText!.substring(0, 100)}...`);
  return {};
}

第 5 步：组装连接

现在我们将节点连接成一个工作的图。因为我们的节点自身处理路由决策，所以我们只需要几条基本边。

要通过 interrupt() 启用人机协同，我们需要使用检查点（checkpointer）进行编译，以在运行之间保存状态：

图编译代码

typescript

import { MemorySaver, RetryPolicy } from "@langchain/langgraph";

// 创建图
const workflow = new StateGraph(EmailAgentState)
  // 添加具有相应错误处理的节点
  .addNode("readEmail", readEmail)
  .addNode("classifyIntent", classifyIntent)
  // 为可能发生瞬时故障的节点添加重试策略
  .addNode(
    "searchDocumentation",
    searchDocumentation,
    { retryPolicy: { maxAttempts: 3 } },
  )
  .addNode("bugTracking", bugTracking)
  .addNode("draftResponse", draftResponse)
  .addNode("humanReview", humanReview)
  .addNode("sendReply", sendReply)
  // 仅添加基本边
  .addEdge(START, "readEmail")
  .addEdge("readEmail", "classifyIntent")
  .addEdge("sendReply", END);

// 使用检查点编译以实现持久化
const memory = new MemorySaver();
const app = workflow.compile({ checkpointer: memory });

图结构非常精简，因为路由是通过节点内部的 Command 对象进行的。每个节点声明它可以跳转到哪里，从而使流程明确且可追溯。

试用你的智能体

让我们使用一个需要人工复核的紧急账单问题来运行智能体：

测试智能体

typescript

// 使用紧急账单问题进行测试
const initialState: EmailAgentStateType = {
  emailContent: "I was charged twice for my subscription! This is urgent!",
  senderEmail: "customer@example.com",
  emailId: "email_123"
};

// 携带 thread_id 运行以进行持久化
const config = { configurable: { thread_id: "customer_123" } };
const result = await app.invoke(initialState, config);
// 图将在 human_review 处暂停
console.log(`Draft ready for review: ${result.responseText?.substring(0, 100)}...`);

typescript

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

// 准备就绪后，提供人类输入以恢复
const humanResponse = new Command({
  resume: {
    approved: true,
    editedResponse: "We sincerely apologize for the double charge. I've initiated an immediate refund...",
  }
});

// 恢复执行
const finalResult = await app.invoke(humanResponse, config);
console.log("Email sent successfully!");

当图到达 interrupt() 时会暂停，将所有内容保存到检查点，并等待。它可以在几天后恢复，准确地从中断处继续。thread_id 确保本次对话的所有状态都保存在一起。

总结与后续步骤

核心见解

构建此邮件智能体展示了 LangGraph 的思考方式：

分解为离散步骤
每个节点做好一件事。这种分解支持流式进度更新、可暂停与恢复的持久化执行以及清晰的调试（因为你可以在步骤之间检查状态）。

状态是共享记忆
存储原始数据，而不是格式化文本。这允许不同的节点以不同的方式使用相同的信息。

节点就是函数
它们接收状态、执行工作并返回更新。当需要做出路由决策时，它们同时指定状态更新和下一个目的地。

错误是流程的一部分
瞬时故障获得重试，LLM 可恢复错误携带上下文循环回，用户可修复问题暂停以等待输入，未知错误向外冒泡以供调试。

人类输入是一等公民
interrupt() 函数无限期暂停执行，保存所有状态，并在你提供输入时精确从中断处恢复。当与节点中的其他操作结合时，它必须放在最前。

图结构自然形成
你定义基本连接，节点处理各自的路由逻辑。这使控制流保持明确且可追溯——通过查看当前节点，你总能理解智能体下一步将做什么。

高级考量

节点粒度权衡
INFO
本节探讨节点粒度设计中的权衡。大多数应用程序可以跳过本节并使用上面展示的模式。
你可能会好奇：为什么不将 Read Email 和 Classify Intent 合并为一个节点？
或者为什么要将 Doc Search 与 Draft Reply 分开？
答案涉及弹性（resilience）与可观测性（observability）之间的权衡。
弹性考量： LangGraph 的持久化层在节点边界创建检查点。当工作流在中断或失败后恢复时，它会从执行停止的节点开头重新开始。较小的节点意味着更频繁的检查点，这意味着如果出现问题，需要重复的工作更少。如果将多个操作合并到一个大节点中，接近尾声处的故障意味着从该节点的开头重新执行所有操作。
我们为邮件智能体选择这种细分的原因：
外部服务的隔离： Doc Search 和 Bug Track 是独立的节点，因为它们调用外部 API。如果搜索服务缓慢或失败，我们希望将其与 LLM 调用隔离开来。我们可以将重试策略添加到这些特定节点，而不影响其他节点。
中间可见性： 将 Classify Intent 作为独立节点，使我们能够在采取操作之前检查 LLM 的决策。这对于调试和监控非常有价值——你可以确切看到智能体何时以及为何路由到人工复核。
不同的故障模式： LLM 调用、数据库查找和邮件发送具有不同的重试策略。独立的节点允许你分别配置它们。
可复用性与测试： 较小的节点更容易单独测试并在其他工作流中复用。
另一种有效的方法：你可以将 Read Email 和 Classify Intent 合并为单个节点。你将失去在分类之前检查原始邮件的能力，并且在发生任何故障时会重复执行该节点中的这两项操作。对于大多数应用程序来说，独立节点带来的可观测性和调试优势是值得权衡的。
应用层考量：步骤 2 中的缓存讨论（是否缓存搜索结果）是应用程序级别的决策，而不是 LangGraph 框架功能。你根据自己的特定需求在节点函数内实现缓存——LangGraph 不作规定。
性能考量：更多的节点并不意味着执行速度更慢。LangGraph 默认在后台写入检查点（异步持久化模式），因此你的图继续运行而无需等待检查点完成。这意味着你可以获得频繁的检查点，同时对性能的影响极小。如果需要，你可以调整此行为——使用 "exit" 模式仅在完成时记录检查点，或使用 "sync" 模式阻塞执行直到写入每个检查点。

下一步方向

这是关于使用 LangGraph 构建智能体的思考方式的介绍。你可以在此基础上扩展：

人机协同模式
了解如何在执行前添加工具审批、批量审批及其他模式

子图
为复杂的多步操作创建子图

流式传输
添加流式传输以向用户展示实时进度

可观测性
借助 LangSmith 添加可观测性以进行调试和监控

工具集成
集成更多用于网络搜索、数据库查询和 API 调用的工具

重试逻辑
为失败的操作实现具有指数退避的重试逻辑

LangGraph 核心思维 ​

从你想要自动化的流程开始 ​

第 1 步：将你的工作流映射为离散步骤 ​

第 2 步：确定每个步骤需要执行的操作 ​

LLM 步骤 ​

数据步骤 ​

操作步骤 ​

用户输入步骤 ​

第 3 步：设计状态 ​

什么属于状态？ ​

保持状态为原始数据，按需格式化提示词 ​

第 4 步：构建节点 ​

妥善处理错误 ​

实现我们的邮件智能体节点 ​

第 5 步：组装连接 ​

试用你的智能体 ​

总结与后续步骤 ​

核心见解 ​

高级考量 ​

下一步方向 ​

LangGraph 核心思维

从你想要自动化的流程开始

第 1 步：将你的工作流映射为离散步骤

第 2 步：确定每个步骤需要执行的操作

LLM 步骤

数据步骤

操作步骤

用户输入步骤

第 3 步：设计状态

什么属于状态？

保持状态为原始数据，按需格式化提示词

第 4 步：构建节点

妥善处理错误

实现我们的邮件智能体节点

第 5 步：组装连接

试用你的智能体

总结与后续步骤

核心见解

高级考量

下一步方向