LangChain 中文文档

外观

当您使用 LangGraph 构建智能体时，首先会将其拆分为称为节点的离散步骤。然后，您需要描述每个节点的各种决策和转换。最后，您通过每个节点都能读写共享的状态将节点连接起来。

在本演练中，我们将带您走过使用 LangGraph 构建客户支持邮件智能体的思考过程。

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

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

txt
The agent should:

- Read incoming customer emails
- Classify them by urgency and topic
- Search relevant documentation to answer questions
- Draft appropriate responses
- Escalate complex issues to human agents
- Schedule follow-ups when needed

Example scenarios to handle:

1. Simple product question: "How do I reset my password?"
2. Bug report: "The export feature crashes when I select PDF format"
3. Urgent billing issue: "I was charged twice for my subscription!"
4. Feature request: "Can you add dark mode to the mobile app?"
5. Complex technical issue: "Our API integration fails intermittently with 504 errors"

要在 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 步骤 ​

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

分类意图 ​

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

草拟回复 ​

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

数据步骤 ​

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

文档搜索 ​

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

客户历史查询 ​

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

操作步骤 ​

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

发送回复 ​

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

错误跟踪 ​

    - 节点执行时机：当意图为 "bug" 时始终执行
    - 重试策略：是，不丢失错误报告至关重要
    - 返回：要在回复中包含的工单 ID

用户输入步骤 ​

当步骤需要人工干预时：

人工审阅节点 ​

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

第 3 步：设计您的状态 ​

状态是所有节点都可以访问的共享记忆。可以将其视为您的智能体在处理流程时用来记录所学内容和所做决策的笔记本。

什么应该放入状态？ ​

针对每条数据，请自问以下问题：

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

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

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

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

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

TIP

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

这种分离意味着：

  • 不同的节点可以根据需要以不同方式格式化相同的数据
  • 您可以在不修改状态 schema 的情况下更改提示词模板
  • 调试更加清晰——您可以确切地看到每个节点接收到的数据
  • 您的智能体可以在不破坏现有状态的情况下演进

让我们定义状态：

python
from typing import TypedDict, Literal

# 定义邮件分类的结构
class EmailClassification(TypedDict):
    intent: Literal["question", "bug", "billing", "feature", "complex"]
    urgency: Literal["low", "medium", "high", "critical"]
    topic: str
    summary: str

class EmailAgentState(TypedDict):
    # 原始邮件数据
    email_content: str
    sender_email: str
    email_id: str

    # 分类结果
    classification: EmailClassification | None

    # 原始搜索/API 结果
    search_results: list[str] | None  # 原始文档块列表
    customer_history: dict | None  # 来自 CRM 的原始客户数据

    # 生成的内容
    draft_response: str | None
    messages: list[str] | None
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 中的节点只是一个接收当前状态并返回状态更新的 Python 函数。

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

妥善处理错误 ​

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

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

瞬时错误 ​

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

结合 timeout= 为每次尝试设置上限。有关完整生命周期，请参阅容错。

python
from langgraph.types import RetryPolicy

workflow.add_node(
    "search_documentation",
    search_documentation,
    retry_policy=RetryPolicy(max_attempts=3, initial_interval=1.0)
)
typescript
import type { RetryPolicy } from "@langchain/langgraph";

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

LLM 可恢复 ​

    将错误存储到状态中并循环回去，以便 LLM 可以看到出错原因并重试：
python
from langgraph.types import Command

def execute_tool(state: State) -> Command[Literal["agent", "execute_tool"]]:
    try:
        result = run_tool(state['tool_call'])
        return Command(update={"tool_result": result}, goto="agent")
    except ToolError as e:
        # 让 LLM 看到出错原因并重试
        return Command(
            update={"tool_result": f"Tool error: {str(e)}"},
            goto="agent"
        )
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、订单号或澄清说明）：
python
from langgraph.types import Command

def lookup_customer_history(
    state: State
) -> Command[Literal["lookup_customer_history", "draft_response"]]:
    if not state.get('customer_id'):
        user_input = interrupt({
            "message": "Customer ID needed",
            "request": "Please provide the customer's account ID to look up their subscription history"
        })
        return Command(
            update={"customer_id": user_input['customer_id']},
            goto="lookup_customer_history"
        )
    # 现在继续执行查询
    customer_data = fetch_customer_history(state['customer_id'])
    return Command(update={"customer_history": customer_data}, goto="draft_response")
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",
  });
}

意外错误 ​

    让它们向上抛出以便调试。不要捕获您无法处理的错误：
python
def send_reply(state: EmailAgentState):
    try:
        email_service.send(state["draft_response"])
    except Exception:
        raise  # 暴露意外错误
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 / 补偿 ​

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

有关完整模式，请参阅容错。

INFO

error_handler 需要 langgraph>=1.2。

python
from langgraph.errors import NodeError
from langgraph.types import Command, RetryPolicy

def payment_error_handler(state: State, error: NodeError) -> Command:
    return Command(
        update={"status": f"compensated: {error.error}"},
        goto="finalize",
    )

workflow.add_node(
    "charge_payment",
    charge_payment,
    retry_policy=RetryPolicy(max_attempts=3, retry_on=ConnectionError),
    error_handler=payment_error_handler,
)

要为图中每个节点应用相同的 retry_policy、timeout 或 error_handler，而无需在每次 add_node 时重复指定，请使用 StateGraph.set_node_defaults(...)。按节点设置的值仍具有更高优先级。请参阅容错。

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

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

读取和分类节点 ​

python
from typing import Literal
from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt, Command, RetryPolicy
from langchain_openai import ChatOpenAI
from langchain.messages import HumanMessage

llm = ChatOpenAI(model="gpt-5-nano")

def read_email(state: EmailAgentState) -> dict:
    """Extract and parse email content"""
    # 在生产环境中，这里会连接你的邮件服务
    return {
        "messages": [HumanMessage(content=f"Processing email: {state['email_content']}")]
    }

def classify_intent(state: EmailAgentState) -> Command[Literal["search_documentation", "human_review", "draft_response", "bug_tracking"]]:
    """Use LLM to classify email intent and urgency, then route accordingly"""

    # 创建返回 EmailClassification 字典的结构化 LLM
    structured_llm = llm.with_structured_output(EmailClassification)

    # 按需格式化提示词，不存储在状态中
    classification_prompt = f"""
    Analyze this customer email and classify it:

    Email: {state['email_content']}
    From: {state['sender_email']}

    Provide classification including intent, urgency, topic, and summary.
    """

    # 直接以字典形式获取结构化响应
    classification = structured_llm.invoke(classification_prompt)

    # 根据分类结果确定下一个节点
    if classification['intent'] == 'billing' or classification['urgency'] == 'critical':
        goto = "human_review"
    elif classification['intent'] in ['question', 'feature']:
        goto = "search_documentation"
    elif classification['intent'] == 'bug':
        goto = "bug_tracking"
    else:
        goto = "draft_response"

    # 将分类结果以单个字典形式存储在状态中
    return Command(
        update={"classification": classification},
        goto=goto
    )
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,
  });
}

搜索和跟踪节点 ​

python
def search_documentation(state: EmailAgentState) -> Command[Literal["draft_response"]]:
    """Search knowledge base for relevant information"""

    # 根据分类结果构建搜索查询
    classification = state.get('classification', {})
    query = f"{classification.get('intent', '')} {classification.get('topic', '')}"

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

    return Command(
        update={"search_results": search_results},  # 存储原始结果或错误
        goto="draft_response"
    )

def bug_tracking(state: EmailAgentState) -> Command[Literal["draft_response"]]:
    """Create or update bug tracking ticket"""

    # 在你的错误跟踪系统中创建工单
    ticket_id = "BUG-12345"  # 实际场景中会通过 API 创建

    return Command(
        update={
            "search_results": [f"Bug ticket {ticket_id} created"],
            "current_step": "bug_tracked"
        },
        goto="draft_response"
    )
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",
  });
}

响应节点 ​

python
def draft_response(state: EmailAgentState) -> Command[Literal["human_review", "send_reply"]]:
    """Generate response using context and route based on quality"""

    classification = state.get('classification', {})

    # 按需从原始状态数据格式化上下文
    context_sections = []

    if state.get('search_results'):
        # 为提示词格式化搜索结果
        formatted_docs = "\n".join([f"- {doc}" for doc in state['search_results']])
        context_sections.append(f"Relevant documentation:\n{formatted_docs}")

    if state.get('customer_history'):
        # 为提示词格式化客户数据
        context_sections.append(f"Customer tier: {state['customer_history'].get('tier', 'standard')}")

    # 使用格式化上下文构建提示词
    draft_prompt = f"""
    Draft a response to this customer email:
    {state['email_content']}

    Email intent: {classification.get('intent', 'unknown')}
    Urgency level: {classification.get('urgency', 'medium')}

    {chr(10).join(context_sections)}

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

    response = llm.invoke(draft_prompt)

    # 根据紧急程度和意图判断是否需要人工审阅
    needs_review = (
        classification.get('urgency') in ['high', 'critical'] or
        classification.get('intent') == 'complex'
    )

    # 路由到适当的下一节点
    goto = "human_review" if needs_review else "send_reply"

    return Command(
        update={"draft_response": response.content},  # 仅存储原始响应
        goto=goto
    )

def human_review(state: EmailAgentState) -> Command[Literal["send_reply", END]]:
    """Pause for human review using interrupt and route based on decision"""

    classification = state.get('classification', {})

    # interrupt() 必须放在最前面——其之前的任何代码在恢复时都会重新执行
    human_decision = interrupt({
        "email_id": state.get('email_id',''),
        "original_email": state.get('email_content',''),
        "draft_response": state.get('draft_response',''),
        "urgency": classification.get('urgency'),
        "intent": classification.get('intent'),
        "action": "Please review and approve/edit this response"
    })

    # 现在处理人工的决策
    if human_decision.get("approved"):
        return Command(
            update={"draft_response": human_decision.get("edited_response", state.get('draft_response',''))},
            goto="send_reply"
        )
    else:
        # 拒绝意味着由人工直接处理
        return Command(update={}, goto=END)

def send_reply(state: EmailAgentState) -> dict:
    """Send the email response"""
    # 与邮件服务集成
    print(f"Sending reply: {state['draft_response'][:100]}...")
    return {}
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() 启用人在回路，我们需要使用检查点器进行编译，以便在运行之间保存状态：

图编译代码 ​

python
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import RetryPolicy

# 创建图
workflow = StateGraph(EmailAgentState)

# 添加带有适当错误处理的节点
workflow.add_node("read_email", read_email)
workflow.add_node("classify_intent", classify_intent)

# 为可能发生瞬时故障的节点添加重试策略
workflow.add_node(
    "search_documentation",
    search_documentation,
    retry_policy=RetryPolicy(max_attempts=3)
)
workflow.add_node("bug_tracking", bug_tracking)
workflow.add_node("draft_response", draft_response)
workflow.add_node("human_review", human_review)
workflow.add_node("send_reply", send_reply)

# 仅添加必要的边
workflow.add_edge(START, "read_email")
workflow.add_edge("read_email", "classify_intent")
workflow.add_edge("send_reply", END)

# 使用 checkpointer 编译以启用持久化；如果使用 Local_Server 运行图，请在不带 checkpointer 的情况下编译
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)
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);

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

图结构保持最小化，因为路由发生在节点内部，通过 Command 对象实现。每个节点都使用 Command[Literal["node1", "node2"]] 之类的类型提示声明它可以去往哪些地方，使流程显式化且可追踪。

图结构保持最小化，因为路由发生在节点内部，通过 Command 对象实现。每个节点都声明它可以去往哪些地方，使流程显式化且可追踪。

试用您的智能体 ​

让我们用一个需要人工审阅的紧急计费问题来运行我们的智能体：

测试智能体 ​

python
from typing import TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt

class EmailState(TypedDict):
    email_content: str
    response_text: str | None

def human_review_node(state: EmailState):
    interrupt(
        {
            "approved": False,
            "edited_response": state.get("response_text") or "",
        }
    )
    return {"response_text": "placeholder"}

app = (
    StateGraph(EmailState)
    .add_node("human_review", human_review_node)
    .add_edge(START, "human_review")
    .add_edge("human_review", END)
    .compile(checkpointer=InMemorySaver())
)

initial_state = {
    "email_content": "I was charged twice for my subscription! This is urgent!",
    "response_text": "Draft response",
}

# 使用 thread_id 运行以实现持久化
config = {"configurable": {"thread_id": "customer_123"}}
stream = app.stream_events(initial_state, config, version="v3")
_ = stream.output  # 驱动流执行至完成
# 图将在 human_review 处暂停
print(f"human review interrupt:{stream.interrupts}")

human_response = Command(
    resume={
        "approved": True,
        "edited_response": "We sincerely apologize for the double charge. I've initiated an immediate refund...",
    }
)

# 恢复执行
resumed = app.stream_events(human_response, config, version="v3")
final_state = resumed.output
print("Email sent successfully!")
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 分开？

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

弹性方面的考量： LangGraph 的持久化层在节点边界创建检查点。当工作流在中断或失败后恢复时，它会从执行停止处的节点开头重新开始。节点越小意味着检查点越频繁，也就意味着出错时需要重复的工作越少。如果将一个大型节点中的多个操作合并，接近末尾的失败就意味着要从该节点开头重新执行所有内容。

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

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

  • 中间可见性： 将 Classify Intent 作为独立节点，让我们能够在采取行动前检查 LLM 的决策。这对调试和监控很有价值——您可以确切地看到智能体何时以及为何路由到人工审阅。

  • 不同的失败模式： LLM 调用、数据库查询和邮件发送有不同的重试策略。独立的节点让您可以分别配置它们。

  • 可复用性与测试： 较小的节点更容易独立测试，也更容易在其他工作流中复用。

另一种同样有效的做法：您可以将 Read Email 和 Classify Intent 合并为一个节点。这样您将失去在分类前检查原始邮件的能力，而且该节点出现任何失败时都要重复执行这两个操作。对于大多数应用程序而言，独立节点带来的可观测性和调试优势是值得付出这种代价的。

应用程序层面的考量：第 2 步中的缓存讨论（是否缓存搜索结果）属于应用程序层面的决策，而不是 LangGraph 框架特性。您可以在节点函数中根据具体需求实现缓存——LangGraph 并不强制规定这一点。

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

接下来可以去哪里 ​

以上是 LangGraph 构建智能体思维的入门介绍。您可以在此基础上扩展：

  • 人在回路模式 — 学习如何在执行前添加工具审批、批量审批及其他模式

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

  • 流式输出 — 添加流式输出以向用户展示实时进度

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

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

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