BytesRouter

主题

LangChain 文档版本 跟随上游 源快照 4b9a9ab4 更新于 2026-10-08

用 LangGraph 思考

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

当使用 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 步：将你的工作流梳理为离散步骤

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

flowchart TD
    A[START] --> B[Read Email]
    B --> C[Classify Intent]

    C -.-> D[Doc Search]
    C -.-> E[Bug Track]
    C -.-> F[Human Review]

    D --> G[Draft Reply]
    E --> G
    F --> G

    G -.-> H[Human Review]
    G -.-> I[Send Reply]

    H --> J[END]
    I --> J[END]

    classDef process fill:#E5F4FF,stroke:#006DDD,stroke-width:2px,color:#030710
    class A,B,C,D,E,F,G,H,I,J process

此图中的箭头显示了可能的路径，但实际决定走哪条路径是在每个节点内部做出的。

既然我们已经确定了工作流中的组件，让我们来了解每个节点需要做什么：

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

TIP

请注意，有些节点会决定接下来去哪里（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 步骤

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

分类意图
  • 静态上下文（提示词）：分类类别、紧急程度定义、响应格式
  • 动态上下文（来自状态）：电子邮件内容、发件人信息
  • 期望结果：决定路由的结构化分类
起草回复
  • 静态上下文（提示词）：语气指南、公司政策、响应模板
  • 动态上下文（来自状态）：分类结果、搜索结果、客户历史记录
  • 期望结果：可供审核的专业电子邮件回复

数据步骤

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

文档搜索
  • 参数：根据意图和主题构建的查询
  • 重试策略：是，针对暂时性故障采用指数退避
  • 缓存：可以缓存常见查询以减少 API 调用
客户历史记录查询
  • 参数：来自状态的客户电子邮件或 ID
  • 重试策略：是，但如果不可用则回退到基本信息
  • 缓存：是，使用存活时间以平衡新鲜度与性能

操作步骤

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

发送回复
  • 何时执行节点：在获得批准之后（人工或自动）
  • 重试策略：是，针对网络问题采用指数退避
  • 不应缓存：每次发送都是唯一的操作
Bug 跟踪
  • 何时执行节点：当意图为 "bug" 时始终执行
  • 重试策略：是，不丢失 bug 报告至关重要
  • 返回：要包含在回复中的工单 ID

用户输入步骤

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

人工审核节点
  • 决策上下文：原始电子邮件、回复草稿、紧急程度、分类
  • 预期输入格式：审批布尔值，加上可选的已编辑回复
  • 触发时机：高紧急程度、复杂问题或质量方面的顾虑

第 3 步：设计你的状态

状态是智能体中所有节点都可访问的共享记忆。可以把它想象成智能体在完成流程时用来记录它所学到和决定的每件事的笔记本。

什么应该放入状态？

对于每一份数据，问自己以下问题：

Include in state

它需要在各个步骤之间持久保留吗？如果是，就放入状态。

Don't store

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

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

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

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

TIP

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

这种分离意味着：

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

让我们定义状态：

python
from typing import TypedDict, Literal

# Define the structure for email classification
class EmailClassification(TypedDict):
    intent: Literal["question", "bug", "billing", "feature", "complex"]
    urgency: Literal["low", "medium", "high", "critical"]
    topic: str
    summary: str

class EmailAgentState(TypedDict):
    # Raw email data
    email_content: str
    sender_email: str
    email_id: str

    # Classification result
    classification: EmailClassification | None

    # Raw search/API results
    search_results: list[str] | None  # List of raw document chunks
    customer_history: dict | None  # Raw customer data from CRM

    # Generated content
    draft_response: str | None
    messages: list[str] | None

请注意，状态只包含原始数据——没有提示词模板，没有格式化字符串，没有指令。分类输出直接以单个字典的形式存储，原封不动地来自 LLM。

第 4 步：构建你的节点

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

妥善处理错误

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

错误类型由谁修复策略何时使用
暂时性错误（网络问题、速率限制）系统（自动）重试策略通常在重试后即可解决的临时性故障
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)
)

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

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

读取和分类节点
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"""
    # In production, this would connect to your email service
    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"""

    # Create structured LLM that returns EmailClassification dict
    structured_llm = llm.with_structured_output(EmailClassification)

    # Format the prompt on-demand, not stored in state
    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.
    """

    # Get structured response directly as dict
    classification = structured_llm.invoke(classification_prompt)

    # Determine next node based on classification
    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"

    # Store classification as a single dict in state
    return Command(
        update={"classification": classification},
        goto=goto
    )
搜索和跟踪节点
python
def search_documentation(state: EmailAgentState) -> Command[Literal["draft_response"]]:
    """Search knowledge base for relevant information"""

    # Build search query from classification
    classification = state.get('classification', {})
    query = f"{classification.get('intent', '')} {classification.get('topic', '')}"

    try:
        # Implement your search logic here
        # Store raw search results, not formatted text
        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:
        # For recoverable search errors, store error and continue
        search_results = [f"Search temporarily unavailable: {str(e)}"]

    return Command(
        update={"search_results": search_results},  # Store raw results or error
        goto="draft_response"
    )

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

    # Create ticket in your bug tracking system
    ticket_id = "BUG-12345"  # Would be created via API

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

    # Format context from raw state data on-demand
    context_sections = []

    if state.get('search_results'):
        # Format search results for the prompt
        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'):
        # Format customer data for the prompt
        context_sections.append(f"Customer tier: {state['customer_history'].get('tier', 'standard')}")

    # Build the prompt with formatted context
    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)

    # Determine if human review needed based on urgency and intent
    needs_review = (
        classification.get('urgency') in ['high', 'critical'] or
        classification.get('intent') == 'complex'
    )

    # Route to appropriate next node
    goto = "human_review" if needs_review else "send_reply"

    return Command(
        update={"draft_response": response.content},  # Store only the raw response
        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() must come first - any code before it will re-run on resume
    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"
    })

    # Now process the human's decision
    if human_decision.get("approved"):
        return Command(
            update={"draft_response": human_decision.get("edited_response", state.get('draft_response',''))},
            goto="send_reply"
        )
    else:
        # Rejection means human will handle directly
        return Command(update={}, goto=END)

def send_reply(state: EmailAgentState) -> dict:
    """Send the email response"""
    # Integrate with email service
    print(f"Sending reply: {state['draft_response'][:100]}...")
    return {}

第 5 步：将它们连接起来

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

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

图编译代码
python
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import RetryPolicy

# Create the graph
workflow = StateGraph(EmailAgentState)

# Add nodes with appropriate error handling
workflow.add_node("read_email", read_email)
workflow.add_node("classify_intent", classify_intent)

# Add retry policy for nodes that might have transient failures
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)

# Add only the essential edges
workflow.add_edge(START, "read_email")
workflow.add_edge("read_email", "classify_intent")
workflow.add_edge("send_reply", END)

# Compile with checkpointer for persistence, in case run graph with Local_Server --> Please compile without checkpointer
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

图结构非常精简，因为路由是通过 Command 对象在节点内部发生的。每个节点使用诸如 Command[Literal["node1", "node2"]] 这样的类型提示声明它可以去往何处，使流程明确且可追踪。

试用你的智能体

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

测试智能体
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",
}

# Run with a thread_id for persistence
config = {"configurable": {"thread_id": "customer_123"}}
stream = app.stream_events(initial_state, config, version="v3")
_ = stream.output  # drive the stream to completion
# The graph will pause at 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...",
    }
)

# Resume execution
resumed = app.stream_events(human_response, config, version="v3")
final_state = resumed.output
print("Email sent successfully!")

查看示例追踪

打开此示例的公开 LangSmith 运行记录。

当图遇到 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

你定义必要的连接，而节点自行处理其路由逻辑。这使控制流保持明确且可追踪——你始终可以通过查看当前节点来了解你的智能体接下来会做什么。

进阶考量

节点粒度的权衡

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 构建智能体思维方式的一个介绍。你可以在这一基础上进一步扩展：

Human-in-the-loop patterns

了解如何添加执行前的工具审批、批量审批以及其他模式

Subgraphs

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

Streaming

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

Observability

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

Tool Integration

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

Retry Logic

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

友情链接

  • Agents.md面向 AI 编码助手的仓库说明文件规范
  • LangChainLangChain 是一个开放的Agent平台
  • LangSmithAgent 工程平台：可观测与评估、部署、Fleet、Engine 等
  • EmbabelEmbabel 是一个JVM平台Agent框架
  • SpringAISpringAI 是一个基于 Spring 的 AI 框架
  • MCPModel Context Protocol，连接 AI 应用与外部数据/工具的开放标准
  • LangChain4jJava 生态的 LLM 应用开发框架
  • LlamaIndex把数据接入 LLM 的框架与文档平台
  • LlamaParse智能体式 OCR 文档解析：把 PDF、扫描件变成 LLM 可用的文本与结构化数据
  • LiteParse本地优先、轻量的文档解析库，不需要 API key
  • CrewAI多智能体编排框架：以 Crew/Flow 组织自主协作的 AI Agent
  • MAFMicrosoft Agent Framework，企业级 .NET/Python 智能体框架

BytesRouter 是社区翻译站点，所有译文仅供参考，一切以官方英文原文为准。

© 2026 BytesRouter · 译文采用 CC BY 4.0 许可