LangGraph 思想

学习如何通过将客户支持邮件智能体分解为离散步骤来思考使用 LangGraph 构建智能体

LangGraph 可以改变你构建智能体的思维方式。当你使用 LangGraph 构建智能体时，首先需要将其分解为称为节点的独立步骤。接下来，你需要为每个节点描述其不同的决策和转换。最后，你会通过一个共享的状态将所有节点连接起来，每个节点都可以读取和写入该状态。在本教程中，我们将引导你了解使用 LangGraph 构建客户支持邮件智能体的思考过程。

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

想象一下，你需要构建一个用于处理客户支持邮件的 AI 智能体。你的产品团队已经给了你这些需求：

智能体应该：

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

需要处理的示例场景：

1. 简单的产品问题：“我该如何重置密码？”
2. 错误报告：“当我选择 PDF 格式时，导出功能会崩溃。”
3. 紧急计费问题：“我的订阅被扣了两次费！”
4. 功能请求：“可以为移动应用添加深色模式吗？”
5. 复杂的技术问题：“我们的 API 集成会间歇性地失败，并出现 504 错误。”

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

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

首先，识别流程中的各个不同步骤。每个步骤将成为一个节点（一个执行特定功能的函数）。然后，勾勒出这些步骤之间如何相互连接。

箭头显示了可能的路径，但实际选择哪条路径的决定发生在每个节点内部。

既然你已经识别出工作流中的组件，现在让我们了解一下每个节点需要做什么：

• 读取邮件：提取并解析邮件内容
• 意图分类：使用 LLM 对紧急程度和主题进行分类，然后路由到相应的操作
• 文档搜索：查询知识库以获取相关信息
• Bug 跟踪：在跟踪系统中创建或更新问题
• 起草回复：生成合适的回复
• 人工审核：上报给人工智能体以进行审批或处理
• 发送回复：发送邮件回复

请注意，一些节点会决定下一步的去向（Classify Intent, Draft Reply, Human Review），而另一些节点则总是流向同一个下一步（Read Email 总是流向 Classify Intent，Doc Search 总是流向 Draft Reply）。

步骤 2：确定每个步骤需要做什么

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

LLM Steps

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

Data Steps

当您需要从外部来源检索信息时使用

Action Steps

当您需要执行外部操作时使用

User Input Steps

当您需要人工干预时使用

LLM 步骤

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

Classify Intent Node

• 静态上下文（提示）：分类类别、紧急程度定义、响应格式
• 动态上下文（来自状态）：邮件内容、发件人信息
• 预期结果：用于确定路由的结构化分类

Draft Reply Node

• 静态上下文（提示）：语气指南、公司政策、回复模板
• 动态上下文（来自状态）：分类结果、搜索结果、客户历史记录
• 期望结果：可供审阅的专业电子邮件回复

数据步骤

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

Document Search Node

• 参数：根据意图和主题构建的查询
• 重试策略：是，对临时故障使用指数退避
• 缓存：可以缓存常见查询以减少 API 调用

Customer History Lookup

• 参数：来自状态的客户邮箱或 ID
• 重试策略：是，但如果不可用则回退到基本信息
• 缓存：是，使用生存时间来平衡新鲜度和性能

操作步骤

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

Send Reply Node

• 执行时机：审批后（人工或自动）
• 重试策略：启用，针对网络问题采用指数退避
• 不应缓存：每次发送都是唯一操作

Bug Track Node

• 执行时机：当意图为”bug”时始终执行
• 重试策略：是，对于不丢失错误报告至关重要
• 返回：响应中包含的工单ID

用户输入步骤

当步骤需要人工干预时：

Human Review Node

• 决策上下文：原始邮件、草稿回复、紧急程度、分类
• 预期输入格式：批准布尔值加上可选的编辑回复
• 触发条件：高紧急程度、复杂问题或质量担忧

步骤 3：设计你的状态

状态是可供智能体中所有节点访问的共享memory。你可以把它看作是智能体在执行过程中，用来记录其学到和决定的一切的笔记本。

状态中应该包含什么？

对于每一条数据，请思考以下问题：

Include in State

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

Don't Store

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

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

• 原始邮件和发件人信息（无法重新构建这些）
• 分类结果（被多个下游节点需要）
• 搜索结果和客户数据（重新获取成本高）
• 草稿回复（需要在审核过程中保持）
• 执行元数据（用于调试和恢复）

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

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

这种分离意味着：

• 不同的节点可以根据自身需求以不同方式格式化相同的数据
• 您可以更改提示模板，而无需修改您的状态模式
• 调试更加清晰 - 您可以精确地看到每个节点接收到的数据
• 您的智能体可以演进，而不会破坏现有状态

我们来定义状态：

import * as z from "zod";

// Define the structure for email classification
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 = z.object({
  // Raw email data
  emailContent: z.string(),
  senderEmail: z.string(),
  emailId: z.string(),

  // Classification result
  classification: EmailClassificationSchema.optional(),

  // Raw search/API results
  searchResults: z.array(z.string()).optional(),  // List of raw document chunks
  customerHistory: z.record(z.any()).optional(),  // Raw customer data from CRM

  // Generated content
  responseText: z.string().optional(),
});

type EmailAgentStateType = z.infer<typeof EmailAgentState>;
type EmailClassificationType = z.infer<typeof EmailClassificationSchema>;

请注意，状态只包含原始数据 - 没有提示模板，没有格式化字符串，没有指令。分类输出被存储为单个字典，直接来自LLM。

第4步：构建你的节点

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

适当地处理错误

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

错误类型	修复者	策略	使用时机
瞬时错误（网络问题、速率限制）	系统（自动）	重试策略	通常在重试后可以解决的临时故障
LLM 可恢复错误（工具故障、解析问题）	LLM	将错误存储在状态中并循环返回	LLM 可以看到错误并调整其方法
用户可修复错误（信息缺失、指令不明确）	人类	使用 interrupt() 暂停	需要用户输入才能继续
意外错误	开发者	让它们冒泡	需要调试的未知问题

Transient errors

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

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

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

LLM-recoverable

将错误存储在状态中并循环返回，以便 LLM 能够看到哪里出错了并重试：

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

async function executeTool(state: State) {
try {
    const result = await runTool(state.toolCall);
    return new Command({
    update: { toolResult: result },
    goto: "agent",
    });
} catch (error) {
    // Let the LLM see what went wrong and try again
    return new Command({
    update: { toolResult: `Tool error: ${error}` },
    goto: "agent"
    });
}
}

User-fixable

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

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

async function lookupCustomerHistory(state: State) {
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",
    });
}
// Now proceed with the lookup
const customerData = await fetchCustomerHistory(state.customerId);
return new Command({
    update: { customerHistory: customerData },
    goto: "draftResponse",
});
}

Unexpected

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

async function sendReply(state: EmailAgentStateType): Promise<void> {
try {
    await emailService.send(state.responseText);
} catch (error) {
    throw error;  // Surface unexpected errors
}
}

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

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

Read and classify nodes

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

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

async function readEmail(state: EmailAgentStateType) {
// Extract and parse email content
// In production, this would connect to your email service
console.log(`Processing email: ${state.emailContent}`);
return {};
}

async function classifyIntent(state: EmailAgentStateType) {
// Use LLM to classify email intent and urgency, then route accordingly

// Create structured LLM that returns EmailClassification object
const structuredLlm = llm.withStructuredOutput(EmailClassificationSchema);

// Format the prompt on-demand, not stored in state
const classificationPrompt = `
Analyze this customer email and classify it:

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

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

// Get structured response directly as object
const classification = await structuredLlm.invoke(classificationPrompt);

// Determine next node based on classification
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";
}

// Store classification as a single object in state
return new Command({
    update: { classification },
    goto: nextNode,
});
}

Search and tracking nodes

async function searchDocumentation(state: EmailAgentStateType) {
// Search knowledge base for relevant information

// Build search query from classification
const classification = state.classification!;
const query = `${classification.intent} ${classification.topic}`;

let searchResults: string[];

try {
    // Implement your search logic here
    // Store raw search results, not formatted text
    searchResults = [
    "Reset password via Settings > Security > Change Password",
    "Password must be at least 12 characters",
    "Include uppercase, lowercase, numbers, and symbols",
    ];
} catch (error) {
    // For recoverable search errors, store error and continue
    searchResults = [`Search temporarily unavailable: ${error}`];
}

return new Command({
    update: { searchResults },  // Store raw results or error
    goto: "draftResponse",
});
}

async function bugTracking(state: EmailAgentStateType) {
// Create or update bug tracking ticket

// Create ticket in your bug tracking system
const ticketId = "BUG-12345";  // Would be created via API

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

Response nodes

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

async function draftResponse(state: EmailAgentStateType) {
// Generate response using context and route based on quality

const classification = state.classification!;

// Format context from raw state data on-demand
const contextSections: string[] = [];

if (state.searchResults) {
    // Format search results for the prompt
    const formattedDocs = state.searchResults.map(doc => `- ${doc}`).join("\n");
    contextSections.push(`Relevant documentation:\n${formattedDocs}`);
}

if (state.customerHistory) {
    // Format customer data for the prompt
    contextSections.push(`Customer tier: ${state.customerHistory.tier ?? "standard"}`);
}

// Build the prompt with formatted context
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)]);

// Determine if human review needed based on urgency and intent
const needsReview = (
    classification.urgency === "high" ||
    classification.urgency === "critical" ||
    classification.intent === "complex"
);

// Route to appropriate next node
const nextNode = needsReview ? "humanReview" : "sendReply";

return new Command({
    update: { responseText: response.content.toString() },  // Store only the raw response
    goto: nextNode,
});
}

async function humanReview(state: EmailAgentStateType) {
// Pause for human review using interrupt and route based on decision
const classification = state.classification!;

// interrupt() must come first - any code before it will re-run on resume
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",
});

// Now process the human's decision
if (humanDecision.approved) {
    return new Command({
    update: { responseText: humanDecision.editedResponse || state.responseText },
    goto: "sendReply",
    });
} else {
    // Rejection means human will handle directly
    return new Command({ update: {}, goto: END });
}
}

async function sendReply(state: EmailAgentStateType): Promise<{}> {
// Send the email response
// Integrate with email service
console.log(`Sending reply: ${state.responseText!.substring(0, 100)}...`);
return {};
}

步骤 5：整合在一起

接下来，我们将节点连接成一个工作图。由于我们的节点自行处理路由决策，我们只需要几条必要的边。

要启用 interrupt() 的 人在回路中 功能，我们需要使用 检查点保存器 来编译，以便在运行之间保存状态：

Graph compilation code

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

// Create the graph
const workflow = new StateGraph(EmailAgentState)
  // Add nodes with appropriate error handling
  .addNode("readEmail", readEmail)
  .addNode("classifyIntent", classifyIntent)
  // Add retry policy for nodes that might have transient failures
  .addNode(
    "searchDocumentation",
    searchDocumentation,
    { retryPolicy: { maxAttempts: 3 } },
  )
  .addNode("bugTracking", bugTracking)
  .addNode("draftResponse", draftResponse)
  .addNode("humanReview", humanReview)
  .addNode("sendReply", sendReply)
  // Add only the essential edges
  .addEdge(START, "readEmail");
  .addEdge("readEmail", "classifyIntent");
  .addEdge("sendReply", END);

// Compile with checkpointer for persistence
const memory = new MemorySaver();
const app = workflow.compile({ checkpointer: memory });

图结构之所以最小，是因为路由通过 Command 对象在节点内部进行。每个节点声明其可前往的路径，使得流程变得明确且可追踪。

试试你的智能体

让我们运行我们的智能体处理一个紧急的账单问题，该问题需要人工审核：

Testing the agent

// Test with an urgent billing issue
const initialState: EmailAgentStateType = {
  emailContent: "I was charged twice for my subscription! This is urgent!",
  senderEmail: "[e-mail]",
  emailId: "email_123"
};

// Run with a thread_id for persistence
const config = { configurable: { thread_id: "customer_123" } };
const result = await app.invoke(initialState, config);
// The graph will pause at human_review
console.log(`Draft ready for review: ${result.responseText?.substring(0, 100)}...`);

// When ready, provide human input to resume
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...",
  }
});

// Resume execution
const finalResult = await app.invoke(humanResponse, config);
console.log("Email sent successfully!");

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

总结和后续步骤

关键洞察

构建这个邮件智能体向我们展示了 LangGraph 的思维方式：

Break into discrete steps

每个节点都专注于做好一件事。这种分解方式支持流式进度更新、可暂停和恢复的持久化执行，以及清晰的调试，因为您可以在步骤之间检查状态。

State is shared memory

存储原始数据，而非格式化文本。这使得不同的节点能够以不同的方式使用相同的信息。

Nodes are functions

它们接收状态，执行工作，并返回更新。当它们需要做出路由决策时，会同时指定状态更新和下一个目标。

Errors are part of the flow

瞬时故障会进行重试，LLM 可恢复的错误会带上上下文进行回退，用户可修复的问题会暂停以等待输入，意外错误会向上冒泡以便进行调试。

Human input is first-class

interrupt() 函数会无限期暂停执行，保存所有状态，并在你提供输入时从其暂停的位置精确恢复。在节点中与其他操作结合使用时，它必须放在第一位。

Graph structure emerges naturally

您定义核心连接，而您的节点则处理各自的路由逻辑。这使得控制流变得明确且可追踪——您始终可以通过查看当前节点来理解您的智能体接下来会做什么。

高级注意事项

Node granularity trade-offs

本节探讨节点粒度设计中的权衡。大多数应用程序可以跳过本节，直接使用上述模式。

你可能会想：为什么不将 Read Email 和 Classify Intent 合并到一个节点中？

或者，为什么要把文档搜索与起草回复分开？

答案涉及弹性和可观测性之间的权衡。

**弹性考量：**LangGraph 的 持久化执行 在节点边界处创建检查点。当工作流在中断或故障后恢复时，它会从执行停止的节点处重新开始。更小的节点意味着更频繁的检查点，这意味着如果出现问题，需要重复的工作就更少。如果将多个操作合并到一个大节点中，那么在接近末尾时发生故障，就意味着需要从该节点的开头重新执行所有操作。

我们为什么为邮件智能体选择这种分解方式：

• **外部服务的隔离：**Doc Search 和 Bug Track 是独立的节点，因为它们调用外部 API。如果搜索服务响应缓慢或失败，我们希望将其与 LLM 调用隔离开。我们可以为这些特定节点添加重试策略，而不会影响其他节点。

• 中间可见性： 将 Classify Intent 作为一个独立的节点，让我们能够检查 LLM 在采取行动之前所做的决定。这对于调试和监控非常有价值——你可以准确地看到智能体何时以及为何会路由到人工审核。

**不同的故障模式：**LLM 调用、数据库查询和邮件发送具有不同的重试策略。独立的节点让您能够独立配置这些策略。

• **可重用性和测试：**更小的节点更容易独立测试并在其他工作流中重用。

另一种有效的方法：你可以将 Read Email 和 Classify Intent 合并到单个节点中。你会失去在分类前检查原始邮件的能力，并且在该节点发生任何失败时，会重复这两个操作。对于大多数应用而言，独立节点的可观测性和调试优势是值得这种权衡的。

应用层考量：步骤 2 中关于是否缓存搜索结果的讨论，是一个应用层决策，而不是 LangGraph 框架的功能。您可以根据具体需求，在节点函数中实现缓存——LangGraph 并未规定这一点。

性能考虑：更多节点并不意味着执行更慢。LangGraph 默认在后台写入检查点（异步持久化模式），因此您的图会继续运行而无需等待检查点完成。这意味着您可以获得频繁的检查点，同时性能影响最小。您可以根据需要调整此行为——使用 "exit" 模式仅在完成时检查点，或使用 "sync" 模式阻塞执行直到每个检查点写入完成。

后续步骤

这是一个关于使用 LangGraph 构建智能体的入门介绍。你可以通过以下方式扩展这个基础：

Human-in-the-loop patterns

学习如何在执行前添加工具审批、批量审批以及其他模式

Subgraphs

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

Streaming

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

Observability

使用 LangSmith 添加可观测性，用于调试和监控

Tool Integration

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

Retry Logic

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

---

在 GitHub 上编辑此页面的源代码。

以编程方式连接这些文档 到 Claude、VSCode 等更多应用，通过 MCP 获取实时答案。