【必藏干货】LangGraph1.0实战：3步构建智能邮件处理系统，掌握条件边+记忆+人在回路核心技巧

本文详细介绍了使用LangGraph1.0构建自动邮件处理智能体的完整流程，包括状态定义、节点设计和边连接。实现了邮件自动分类、智能处理、自动回复及人工审核等功能，并通过测试验证了系统在紧急和常规邮件处理上的有效性。综合运用了条件边、记忆机制和人在回路等LangGraph核心概念，为开发者提供了实用的AI Agent开发实战案例。

–

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

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追踪节点、文档查询节点、自动撰写节点的相关内容。但在需要处理连续对话或多轮交互的场景中，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 构建智能体图

具备以上节点函数后，通过边将节点函数相连接构造智能体。智能体定义遵循三步骤：定义状态、添加节点、添加边。同时由于需要使用中断机制，还需要添加内存检查点。

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编写智能体的核心技巧。

AI时代，未来的就业机会在哪里？

答案就藏在大模型的浪潮里。从ChatGPT、DeepSeek等日常工具，到自然语言处理、计算机视觉、多模态等核心领域，技术普惠化、应用垂直化与生态开源化正催生Prompt工程师、自然语言处理、计算机视觉工程师、大模型算法工程师、AI应用产品经理等AI岗位。

掌握大模型技能，就是把握高薪未来。

那么，普通人如何抓住大模型风口？

AI技术的普及对个人能力提出了新的要求，在AI时代，持续学习和适应新技术变得尤为重要。无论是企业还是个人，都需要不断更新知识体系，提升与AI协作的能力，以适应不断变化的工作环境。

因此，这里给大家整理了一份《2026最新大模型全套学习资源》，包括2026最新大模型学习路线、大模型书籍、视频教程、项目实战、最新行业报告、面试题、AI产品经理入门到精通等，带你从零基础入门到精通，快速掌握大模型技术！

由于篇幅有限，有需要的小伙伴可以扫码获取！

1. 成长路线图&学习规划

要学习一门新的技术，作为新手一定要先学习成长路线图，方向不对，努力白费。这里，我们为新手和想要进一步提升的专业人士准备了一份详细的学习成长路线图和规划。

2. 大模型经典PDF书籍

书籍和学习文档资料是学习大模型过程中必不可少的，我们精选了一系列深入探讨大模型技术的书籍和学习文档，它们由领域内的顶尖专家撰写，内容全面、深入、详尽，为你学习大模型提供坚实的理论基础。（书籍含电子版PDF）

3. 大模型视频教程

对于很多自学或者没有基础的同学来说，书籍这些纯文字类的学习教材会觉得比较晦涩难以理解，因此，我们提供了丰富的大模型视频教程，以动态、形象的方式展示技术概念，帮助你更快、更轻松地掌握核心知识。

4. 大模型项目实战

学以致用，当你的理论知识积累到一定程度，就需要通过项目实战，在实际操作中检验和巩固你所学到的知识，同时为你找工作和职业发展打下坚实的基础。

5. 大模型行业报告

行业分析主要包括对不同行业的现状、趋势、问题、机会等进行系统地调研和评估，以了解哪些行业更适合引入大模型的技术和应用，以及在哪些方面可以发挥大模型的优势。

6. 大模型面试题

面试不仅是技术的较量，更需要充分的准备。

在你已经掌握了大模型技术之后，就需要开始准备面试，我们将提供精心整理的大模型面试题库，涵盖当前面试中可能遇到的各种技术问题，让你在面试中游刃有余。

为什么大家都在学AI大模型？

随着AI技术的发展，企业对人才的需求从“单一技术”转向 “AI+行业”双背景。企业对人才的需求从“单一技术”转向 “AI+行业”双背景。金融+AI、制造+AI、医疗+AI等跨界岗位薪资涨幅达30%-50%。

同时很多人面临优化裁员，近期科技巨头英特尔裁员2万人，传统岗位不断缩减，因此转行AI势在必行！

这些资料有用吗？

这份资料由我们和鲁为民博士(北京清华大学学士和美国加州理工学院博士)共同整理，现任上海殷泊信息科技CEO，其创立的MoPaaS云平台获Forrester全球’强劲表现者’认证，服务航天科工、国家电网等1000+企业，以第一作者在IEEE Transactions发表论文50+篇，获NASA JPL火星探测系统强化学习专利等35项中美专利。本套AI大模型课程由清华大学-加州理工双料博士、吴文俊人工智能奖得主鲁为民教授领衔研发。

资料内容涵盖了从入门到进阶的各类视频教程和实战项目，无论你是小白还是有些技术基础的技术人员，这份资料都绝对能帮助你提升薪资待遇，转行大模型岗位。