LangGraph1.0实战：构建自动邮件处理智能体，大模型应用开发指南

本文详细介绍了如何使用LangGraph1.0构建自动邮件处理智能体，展示了状态管理、节点、边、条件路由和中断机制等核心技术的实战应用。通过定义智能体状态、编写节点函数、构建智能体图和测试验证，实现了一个能够自动分类邮件、处理Bug报告、搜索知识库生成回复，并在必要时引入人工审核的完整工作流。这一实战项目帮助开发者深入理解LangGraph框架，掌握大模型智能体开发的核心技能。

前言

上期内容 [LangGraph1.0速通指南（二）—— LangGraph1.0 条件边、记忆、人在回路]分享了条件边、智能体记忆和人在回路等核心机制。通过点、边、条件边的组合，大家已经能够搭建起具备记忆能力并支持人工介入的智能体工作流。为了帮助大家更熟练地掌握这些概念，并体会其在真实场景中的应用，本期笔者将把前两期所学的知识——包括状态、节点、边、条件路由、记忆与中断机制——融会贯通，动手构建一个自动邮件处理智能体项目。

一、电子邮件智能体场景分析

1.1 需求分析

作为一名开发者笔者每天都需要处理大量的工作邮件——包括问题咨询、Bug 报告、账单通知、功能请求等。邮件不仅类型多样，紧急程度也各不相同，手动分类、回复耗时耗力，尤其在忙碌时更容易遗漏或延迟处理。为此笔者希望构建一个自动邮件处理智能体，它能够：

自动分类：识别邮件的类型（如问题、Bug、账单、功能请求等）与紧急程度。
智能处理：根据邮件类型执行相应操作（如为 Bug 创建跟踪工单，为其他问题搜索内部文档获取解答）。
自动生成回复：基于分类与处理结果，生成有针对性的回复内容。
审核机制：在必要时（如涉及重要决策或复杂问题）暂停流程，引入人工审核，确保回复准确性与安全性。

该智能体将帮助笔者提升邮件处理效率，同时通过“人在回路”保证对自动回复的可控制。

1.2 根据需求分析设计图

基于以上需求笔者设计的智能体将具备以下核心功能流程：

邮件提取与解析：从收件箱获取原始邮件，提取关键内容（发件人、主题、正文等）。
自动分类与路由：对邮件内容进行分类，并依据类别决定后续处理路径。
内容处理与生成：

若为Bug 报告，在错误跟踪系统中说明已经收到并跟踪错误；
若为其他类别（如问题咨询），则在内部知识库中检索相关答案。

回复生成与审核：

根据处理结果生成回复草稿；
依据邮件类别与紧急程度，决定是否发送至人工审核节点；
审核通过后自动发送邮件，或由人工修订后发出。

整体流程可通过以下示意图概括：

二、项目代码编写

2.1 定义智能体状态

1. 环境配置与模型初始化：首先配置项目环境并初始化大语言模型，本次同样使用deepseek模型，需要提前执行pip install langchain-deepseek安装相关的依赖包，并在项目文件夹下新建.env文件，将注册的DeepSeek API填入该文件中。

import uuid import os from typing import Literal, TypedDict from dotenv import load_dotenv from langchain_deepseek import ChatDeepSeek from langgraph.types import Command, interrupt from langgraph.graph import END, START, StateGraph from langgraph.checkpoint.memory import InMemorySaver load_dotenv() llm = ChatDeepSeek( model="deepseek-chat", )

2. 定义状态数据结构：智能体的状态设计需要与工作流节点相匹配，确保每个节点的输入输出都有明确的数据结构。EmailAgentState的设计从上到下分别是读取邮件节点、分类节点、bug追踪节点、文档查询节点、自动撰写节点的相关内容。本项目处理的是单封邮件的独立流程，因此未使用笔者在《LangGraph1.0速通指南（一）—— 核心概念、点、边》中介绍的reducer进行状态聚合。但在需要处理连续对话或多轮交互的场景中，reducer机制将是管理状态演进的关键工具。

# 1. 定义图状态State class EmailClassification(TypedDict): # 分类包括问题、错误、账单、功能, 不属于这些功能就分类为复杂 intent: Literal['question', 'bug', 'billing', 'feature', 'complex'] # 紧急程度： 低、中、高、关键 urgency: Literal['low', 'medium', 'high', 'critical'] # 邮件主题 topic: str # 邮件摘要 summary: str class EmailAgentState(TypedDict): # 储存读取邮件内容, 发件人邮件地址，邮件ID email_content: str sender_email: str email_id: str # 分类意图节点的结果， classification: EmailClassification # Bug tracking 错误处理系统只需要查询一些API， 这里就存储一个工单ID即可 ticket_id: str | None # 搜索文档结果 search_results: list[str] | None # 客户历史：客户历史是一个字典，键是客户的电子邮件，值是与该客户相关的任何搜索结果 customer_history: dict | None # 生成内容: draft_response: str | None

2.2 定义智能体节点

**1. 邮件读取与分类节点：首先明确读取节点与分类节点的功能。在实际生产环境下，读取节点通常通过调用特定邮箱的 API 实现，但为简化流程，笔者在此略过该节点，直接将邮件内容输入至后续分类节点。分类节点通过预设提示词，调用 DeepSeek 大模型对邮件进行分类。LangChain1.0 支持便捷的大模型结构化输出，具体方法笔者不再展开，

def read_email(state: EmailAgentState) -> EmailAgentState: ''' 实际生产中这一步要从邮箱提供的api中提取电子邮件，这里仅仅是演示， 会将电子邮件直接传给分类节点，该函数简写 ''' pass def classify_intent(state: EmailAgentState) -> EmailAgentState: "用大模型进行节点分类和紧急程度识别，然后依据结果路由" structured_llm = llm.with_structured_output(EmailClassification) classification_pormpt = f""" 分析用户输入的邮件并进行分类 邮件: {state['email_content']} 来自: {state['sender_email']} 提供分类、紧急程度、主题和内容摘要 """ classication = structured_llm.invoke(classification_pormpt) return { 'classification': classication }

2. 知识库搜索节点：目前搜索节点仅为功能模拟，在实际部署时，可替换为对接真实知识库的检索接口。该节点接收并依据上一节点（即分类节点）输出的状态信息，从中提取邮件意图与主题，进而模拟在内部知识库中执行检索的过程。

def search_documentation(state: EmailAgentState) -> EmailAgentState: ''' 查询知识库节点，这里模拟操作 ''' classification = state.get('classification', {}) query = f"{classification.get('intent', '')} {classification.get('topic', '')}" try: # 模拟查询的逻辑 search_results = [ 'search_result_1', 'search_result_2', 'search_result_3' ] except Exception as e: search_results = [f'搜索接口不可用'] return {'search_results': search_results}

3. Bug跟踪节点：模拟跟踪的内容，创建跟踪工单

def bug_tracking(state: EmailAgentState) -> EmailAgentState: ''' 模拟bug修复的相关内容 ''' ticket_id = f"Bug-{uuid.uuid4()} fixed" return {'ticket_id': ticket_id}

4. 回复生成节点：节点接收来自搜索节点的信息（如原始检索结果、以换行符连接的客户历史记录等），并将其作为上下文参考，指导大模型的回复生成。大模型基于邮件分类、紧急程度及所附的参考信息，生成拟回复的邮件内容。除了生成回复内容，该节点还会输出一个****处理命令。该命令将根据邮件的类别与紧急程度，自动判断下一步应执行的操作——例如“直接发送回复”或“转交人工审核”。

def write_response(state: EmailAgentState) -> Command[Literal['human_review', 'send_reply']]: ''' 根据分类结果、搜索结果等中间结果生产报告 ''' 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"相关内容:\n{formatted_docs}") if state.get('customer_history'): context_sections.append(f"Customer tier: {state['customer_history'].get('tier', 'standard')}") # 构建提示词 draft_prompt = f""" 撰写50字邮件回复: 邮件内容: {state.get('email_content')} 邮件分类: {classification.get('intent', 'unkown')} 紧急程度: {classification.get('urgency', 'medium')} {chr(10).join(context_sections)} """ response = llm.invoke(draft_prompt) # 根据紧急程度决定是否需要人类审核 needs_review = ( classification.get('urgency') in ['high', 'critical'] or classification.get('intent') == 'complex' ) if needs_review: goto = 'human_review' print('需要人工审核') else: goto = 'send_reply' return Command( update={'draft_response': response.content}, goto=goto )

5. 人工审核节点和回复节点：流程执行至该节点时，系统将自动暂停并等待人工输入。只有在邮件紧急程度为紧急或分类结果为复杂情况下，才会进入工作节点来人工确认。该节点采用中断机制，根据人工反馈的指令，节点将返回相应的路由命令，引导流程走向“结束”或“发送回复”分支。回复节点笔者这里只模拟自动发送邮件。

def human_review(state: EmailAgentState) -> Command[Literal['send_reply', END]]: ''' 人类审查节点，审查结束后决定是否要回复该邮件 ''' classification = state.get('classification', {}) human_decision = interrupt({ "邮件ID": state['email_id'], "原始邮件内容": state['email_content'], "自动回复内容": state['draft_response'], '紧急程度': classification['urgency'], '分类': classification['intent'], '下一步': '请审核是否同意发送该邮件' }) if human_decision=='approved': return Command( update={}, goto='send_reply' ) else: return Command( update={}, goto=END ) def send_reply(state: EmailAgentState): print('---成功发送---')

2.3 构建智能体图

具备以上节点函数后，通过边将节点函数相连接构造智能体。智能体定义遵循 LangGraph1.0速通指南（一）—— LangGraph1.0 核心概念、点、边讲解的三步骤：定义状态、添加节点、添加边。同时由于需要使用中断机制，还需要添加内存检查点。

builder = StateGraph(EmailAgentState) builder.add_node("read_email", read_email) builder.add_node("classify_intent", classify_intent) builder.add_node("search_documentation",search_documentation) builder.add_node("bug_tracking", bug_tracking) builder.add_node("write_response", write_response) builder.add_node("human_review",human_review) builder.add_node("send_reply", send_reply) builder.add_edge(START, "read_email") builder.add_edge("read_email","classify_intent") builder.add_edge("classify_intent","search_documentation") builder.add_edge("classify_intent","bug_tracking") builder.add_edge("search_documentation","write_response") builder.add_edge("bug_tracking","write_response") builder.add_edge("send_reply",END) memory=InMemorySaver() app = builder.compile(checkpointer=memory)

三、智能体测试与验证

为了验证笔者构建的邮件处理智能体是否按预期工作，下面通过两个典型场景进行测试：高紧急度Bug报告和常规问候邮件。

3.1 Bug报告处理测试

测试一个需要人工审核的高优先级Bug报告，预期的处理流程如下：

由于邮件内容包含"紧急Bug"，分类节点会将其识别为intent: bug且urgency: critical
工作流会创建Bug工单，然后进入human_review节点等待审核
系统会打印中断信息并等待人工输入
根据人工决策，邮件将被发送或终止

initial_state = { "email_content": "我遇到了一个紧急bug, 有客户重复订阅了一个产品", "sender_email": "test@163.com", "email_id": "email_123" } config = {"configurable": {"thread_id": "customer_123"}} result = app.invoke(initial_state, config=config) print(f'准备审核的回复内容:{result['draft_response']}...\n') if '__interrupt__' in result: print(f'Interrupt:{result}') msg = result['__interrupt__'][-1].value print(msg) human = input(f"请输入: ") human_response = Command( resume=human ) final_result = app.invoke(human_response, config)

从运行结果来看，工作流成功识别出紧急Bug，创建了工单并生成了回复草稿。由于紧急程度为"critical"，系统正确触发了人工审核中断：

3.2：常规问候处理流程

测试一个无需审核的普通问候邮件，预期结果如下:

问候邮件会被分类为intent: question且urgency: low
工作流将跳过人工审核节点，直接生成并发送回复
整个过程完全自动化，无需人工干预

initial_state = { "email_content": "你好呀，我是新同事苍井空", "sender_email": "test@163.com", "email_id": "email_123" } config = {"configurable": {"thread_id": "customer_123"}} result = app.invoke(initial_state, config=config) print(f'准备审核的回复内容:{result['draft_response']}...\n') if '__interrupt__' in result: print(f'Interrupt:{result}') msg = result['__interrupt__'][-1].value print(msg) human = input(f"请输入: ") human_response = Command( resume=human ) final_result = app.invoke(human_response, config)

从运行结果可以看出系统正确识别出低优先级问候邮件，自动生成友好回复并直接发送，无需人工介入：

完整代码大家可以关注笔者的微信公众号大模型真好玩，并私信:LangChain智能体开发免费获取。

四、总结

本文通过构建自动邮件处理智能体，系统演示了LangGraph1.0的状态管理、条件路由、人在回路等核心机制在实战中的应用。相信大家经过实战，能更加掌握LangGraph1.0编写智能体的核心技巧。