Classify intent

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

Draft reply

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

Document search

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

Customer history lookup

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

Send reply

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

Bug track

- 何时执行节点：当意图为"bug"时始终执行 - 重试策略：是，关键是不能丢失 bug 报告 - 返回：响应中包含工单 ID

Human review node

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

Transient errors

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

结合使用 timeout= 来限制每次尝试。参见 容错 了解完整生命周期。

    from langgraph.types import RetryPolicy

    workflow.add_node(
        "search_documentation",
        search_documentation,
        retry_policy=RetryPolicy(max_attempts=3, initial_interval=1.0)
    )
    

LLM-recoverable

将错误存储在状态中并循环回退，这样 LLM 可以看到出了什么问题并重试：

    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:
            # Let the LLM see what went wrong and try again
            return Command(
                update={"tool_result": f"Tool error: {str(e)}"},
                goto="agent"
            )
    

User-fixable

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

    from langgraph.types import Command


    def lookup_customer_history(state: State) -> Command[Literal["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"
            )
        # Now proceed with the lookup
        customer_data = fetch_customer_history(state['customer_id'])
        return Command(update={"customer_history": customer_data}, goto="draft_response")
    

Unexpected

让它们向上冒泡以便调试。不要捕获你无法处理的内容：

    def send_reply(state: EmailAgentState):
        try:
            email_service.send(state["draft_response"])
        except Exception:
            raise  # Surface unexpected errors
    

Saga / compensation

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

参见 容错 了解完整模式。

    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,
    )
    

Read and classify nodes

    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
        )
    

Search and tracking nodes

    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"
        )
    

Response nodes

    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 {}
    

Graph compilation code

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)

Testing the agent

Node granularity trade-offs

您可能想知道：为什么不合并 Read Email 和 Classify Intent 到一个节点中？

或者为什么要将文档搜索与草稿回复分开？

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

弹性考虑： LangGraph 的 持久化层 在节点边界创建检查点。当工作流在中断或失败后恢复时，它会从执行停止的节点开始重新启动。较小的节点意味着更频繁的检查点，这意味着如果出现问题，需要重复的工作更少。如果您将多个操作组合到一个大节点中，靠近末尾的故障意味着从该节点的开头重新执行所有内容。

为什么我们为邮件代理选择这种分解：

• 外部服务隔离： 文档搜索和缺陷跟踪是独立的节点，因为它们调用外部 API。如果搜索服务缓慢或失败，我们希望将其与 LLM 调用隔离开来。我们可以向这些特定节点添加重试策略，而不影响其他节点。
• 中间可见性： 将 Classify Intent 作为一个单独的节点，让我们能够在采取行动之前检查 LLM 的决定。这对于调试和监控很有价值——您可以准确看到代理何时以及为何路由到人工审核。
• 不同的失败模式： LLM 调用、数据库查询和电子邮件发送具有不同的重试策略。单独的节点让您能够独立配置这些。
• 可重用性和测试： 较小的节点更容易独立测试，并在其他工作流中重用。

另一种有效的方法：您可以将 Read Email 和 Classify Intent 合并到单个节点中。您将失去在分类之前检查原始电子邮件的能力，并且在该节点中的任何故障时都会重复两个操作。对于大多数应用程序来说，单独节点的可见性和调试好处值得这个权衡。

应用层关注点：第2步中的缓存讨论（是否缓存搜索结果）是一个应用层决策，而非 LangGraph 框架的功能。你根据具体需求在节点函数中实现缓存——LangGraph 不规定这一点。

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