> ## Content Index
> Fetch the complete content index at: https://www.hubwiz.com/blog/llms.txt
> Use this file to discover other available public pages before exploring further.

# 构建Deepseek AI邮件代理
- URL: https://www.hubwiz.com/blog/build-email-agent-with-deepseek/
- Published: 2025-02-18T05:41:52.000Z
- Updated: 2025-02-27T10:13:38.000Z
- Description: 在本博客中，我将引导你使用Deepseek、LangChain和LangGraph创建一个AI邮件代理。这个项目自动化了电子邮件分类、摘要和回复生成，从而简化你的电子邮件工作流程。
- Author: admin
- Tags: APPLICATION, AGENT, DEEPSEEK, LANGCHAIN, LANGGRAPH, #Import 2026-10-14 14:41

**大型语言模型（LLMs）**已经存在了一段时间，主要擅长于下一个词预测。然而，人工智能代理的引入增加了一个额外的智能层，使这些模型能够执行更复杂的、多步骤的任务，而不仅仅是简单的文本生成。

从根本上说，真正的AI代理不仅仅是在链式LLM生成的提示之间进行连接——它集成了专门工具来处理复杂的、多步骤的任务。相反，它是一个由两个主要组件组成的系统：

> ***代理 = \[ 工具 + LLM \]***

![](https://www.hubwiz.com/blog/content/images/2025/02/image-614.png)

图1：代理组件

这意味着，虽然LLM（如ChatGPT或Deepseek）负责自然语言处理，我们还为特定用例整合了专门工具。

在本博客中，我将引导你使用Deepseek、LangChain和LangGraph创建一个AI邮件代理。这个项目自动化了电子邮件分类、摘要和回复生成，从而简化你的电子邮件工作流程。

这个*代理工作流包含 *监督者-从属层次结构**，其中中央监督者（使用LangGraph构建）将各种代理（使用LangChain实现）整合到一个流畅的管道中。

## 1、项目概述与架构

我们正在构建一个基于AI的电子邮件代理，该代理可以：

- 实时获取电子邮件。
- 将其分类为*垃圾邮件、紧急、需要审查*和*信息性*类别。
- 为每封电子邮件生成摘要。
- 使用摘要来撰写适当的回复。
- 根据用户输入发送或草拟回复。

虽然**Deepseek API**是关键组件，但我们还需要其他工具来构建此系统。主要使用的三个工具是**LangChain**、**LangGraph**和**IMAP服务器**。

## 2、工具的工作原理

***LangGraph***提供了一个结构化的、有向图框架，其中每个**节点**代表一个特定任务——过滤、总结或生成回复。**边**定义了数据如何在这类节点之间流动。

***LangGraph***支持动态路由，这意味着它可以应用条件逻辑来确定下一步。例如，如果一封电子邮件被标记为垃圾邮件，管道可以自动终止进一步处理，以确保效率。

***LangChain***通过启用模块化方法简化了LLM驱动的应用程序的创建。我们管道中的每个组件——例如**摘要或回复生成**——都是一个独立的模块。这些模块无缝地协同工作，确保平滑的数据流和易于维护。

***IMAP***允许我们的系统连接到远程电子邮件服务器，验证用户身份，并检索电子邮件以进行处理。它支持实时电子邮件检索和加载基于JSON的测试数据。

## 3、电子邮件处理管道

![](https://www.hubwiz.com/blog/content/images/2025/02/image-615.png)

图2：解释管道的流程图

### 3.1 开始：获取电子邮件

- 使用IMAP实时检索电子邮件或将JSON文件中的电子邮件用于测试和开发。

### 3.2 创建EmailState对象

- 初始化一个对象，该对象跟踪当前电子邮件，维护处理步骤的历史记录，并存储元数据。
- 作为所有管道中流动数据的容器。

### 3.3 过滤节点

- 分析电子邮件的主题和正文。
- 将电子邮件分类为*垃圾邮件、紧急、信息性*或*需要审查*。
- 如果标记为垃圾邮件，则管道会提前终止以优化资源使用。

### 3.4 摘要节点

- 生成电子邮件内容的简短摘要。
- 将摘要传递给响应节点进行处理。

### 3.5 响应节点

- 使用摘要和其他电子邮件详细信息撰写合适的回复。
- 基于电子邮件的上下文生成初步的自动回复。

### 3.6 人工审核节点

- 如果生成的回复不确定或标记为需要审核，用户可以手动编辑或改进它。

### 3.7 IMAP集成

- 从IMAP服务器实时获取电子邮件或将测试数据从JSON加载。
- 支持选择性电子邮件检索以确保管道顺畅执行。

### 3.8 发送或草拟决策

- 用户可以选择立即发送回复还是保存为草稿。
- 如果*发送*： — 通过SMTP发送电子邮件。
- 如果*草拟*： — 将回复保存并可选地转发到Gmail地址。

### 3.9 结束：所有电子邮件已处理

- 在处理一批电子邮件后标记管道结束。
- 确认所有操作均已记录。

完成管道概述后，让我们深入探讨将此架构付诸实践的代码。

## 4、代码的工作原理

代码 — [GitHub仓库](https://github.com/parthshr370/Email-AI-Agent?ref=hubwiz.com)

虽然我不会详细介绍整个代码库，但我会解释参与创建此工作流的关键组件：

### 4.1 response\_agent.py

响应代理负责根据提供的摘要生成电子邮件回复。定义了一个名为`generate_response`的函数，该函数接受电子邮件对象、其摘要以及收件人和发件人的姓名。该函数使用LangChain中的`PromptTemplate`函数构造提示，确保包含所有必需的电子邮件详细信息。

```python
# response_agent.py  
# 使用电子邮件详细信息和摘要构建提示，指导LLM生成正式回复。  
  
from langchain.prompts import PromptTemplate  
from config import DEEPSEEK_API_KEY  # 导入配置中的密钥  
from langchain_openai import ChatOpenAI  
from utils.formatter import clean_text, format_email  
from email.utils import parseaddr  
  
  
def generate_response(email: dict, summary: str, recipient_name: str, your_name: str) -> str:  
    prompt_template = PromptTemplate(  
        input_variables=["sender", "subject", "content", "summary", "user_name","recipient_name"],  
        template=(  
            "你是一位电子邮件助手。不要使用占位符如[用户的名字]"  
            "你是一位电子邮件助手。不要在回复中包含任何问候语或签名行。\n\n"  
            "电子邮件详情：\n"  
            "发件人：{sender}\n"  
            "主题：{subject}\n"  
            "内容：{content}\n"  
            "摘要：{summary}\n\n"  
              
            "以正式语气回复。"  
        )  
    ) # 回复模板  
  
    prompt = prompt_template.format(  
        sender=recipient_name,  # 使用收件人的名字（手动提供）  
        subject=email.get("subject", ""),  
        content=email.get("body", ""),  
        summary=summary,  
        user_name=your_name  
    )

```

在这里，导入了必要的库，并创建了一个生成回复的函数，明确指定了电子邮件组件的数据类型。`PromptTemplate`函数用于构建一个包含所有必需占位符的模板，LLM将根据提供的指令填充这些占位符。

最后，使用来自`prompt_template`的占位符构建提示。

这里有一个有趣的技巧——你需要欺骗库使其认为你在使用OpenAI API密钥，同时将整个API重定向到Deepseek服务器：

```python
# response_agent.py   
  
model = ChatOpenAI(  
        base_url="https://api.deepseek.com/v1",  
        model="deepseek-chat",  
        temperature=0.5,  
        openai_api_key=DEEPSEEK_API_KEY  
    )  
      
    response = model.invoke(prompt) # 让模型开始行动   
    response_text = response.content if hasattr(response, "content") else str(response)  
      
    # 传递收件人名字（用于问候语）和你的名字（用于签名）  
    formatted_response = format_email(email.get("subject", ""), recipient_name, response_text, your_name)  
    return formatted_response.strip()

```

这种结构为构建其他代理（如过滤、响应生成和人工审核）奠定了基础。完整的实现可以在文档中链接的[GitHub仓库](https://github.com/parthshr370/Email-AI-Agent?ref=hubwiz.com)中找到。

### 4.2 supervisor.py

现在我们来看看***监督者***，它将所有这些代理整合到管道中。

```python
from core.state import EmailState  
from agents import filtering_agent, summarization_agent, response_agent, human_review_agent  
from langgraph.graph import START , END, StateGraph  
  
"""最初，每个节点函数都期望接收两个参数——一封电子邮件和一个状态。  
然而，LangGraph框架设计为仅向每个节点传递一个参数（状态）。"""  
  
# 将所有状态整合在一起，通过监督者来管理电子邮件处理的流程。  
def supervisor_langgraph(email: dict, state: EmailState,user_name : str,recipient_name:str) -> EmailState:  
    """  
    使用LangGraph工作流处理单个电子邮件。  
    每个步骤（过滤、摘要、响应生成）是一个节点。  
    使用条件边来提前退出垃圾邮件或继续处理。  
    """  
      
    state.current_email = email  
      
    def filtering_node(state: EmailState) -> EmailState: # 来自过滤代理的过滤节点  
        current_email = state.current_email  
        print('过滤节点开始处理电子邮件ID：%s' % current_email.get("id", "未知"))  
        classification = filtering_agent.filter_email(current_email)  
        current_email["classification"] = classification  
        state.metadata[current_email.get("id", "未知")] = classification  
        return state  
  
      
    def summarization_node(state: EmailState) -> EmailState:  
        email = state.current_email  
        summary = summarization_agent.summarize_email(email)  
        email["summary"] = summary  
        return state  
      
    def response_node(state: EmailState) -> EmailState:  
        email = state.current_email  
        response = response_agent.generate_response(email, email.get("summary", ""),recipient_name,user_name) # 响应代理使用摘要生成回复  
        # 如果分类指示需要审核或响应不确定，则让人类干预  
        if email.get("classification") == "needs_review" or "?" in response:  
            response = human_review_agent.review_email(email, response)  
        email["response"] = response  
        state.history.append({  
            "email_id": email.get("id", "未知"),  
            "response": response  
        })  
        return state  
      
    graph_builder = StateGraph(EmailState)     # 现在从所有状态构建图形  
  
      
    # 在图形中添加节点  
    graph_builder.add_node("过滤", filtering_node)  
    graph_builder.add_node("摘要", summarization_node)  
    graph_builder.add_node("响应", response_node)  
      
      
    # 使用过滤建立条件工作，如果不是垃圾邮件则继续，如果是垃圾邮件则直接进入垃圾桶  
    def post_filtering(state_update: EmailState):  
        email = state.current_email  
        if email.get("classification") == "垃圾邮件":  
            return END  
        else:  
            return "摘要"  
      
    graph_builder.add_conditional_edges("过滤", post_filtering, {"摘要": "摘要", END: END})  
      
    # 这里创建一个从“摘要”节点到“响应”节点的直接边。  
    # 如果达到摘要节点，则必须移动到响应节点  
    graph_builder.add_edge("摘要", "响应")  
      
    graph_builder.add_edge("响应", END) # 如果响应到达则结束所有处理请  
      
    # 设置入口点为过滤节点。  
    graph_builder.set_entry_point("过滤")  
      
    # 编译图形。  
    graph = graph_builder.compile()  
      
    # 使用当前状态调用图形。  
    final_state = graph.invoke(state)  
    return final_state

```

在这里，我们定义了一个**监督函数**，作为中心控制器，整合所有代理并为每个代理创建相应的节点（例如`response_node`）。

使用**LangGraph**，我们构建了一个基于图形的工作流，其中每个处理步骤都被表示为一个节点。这是通过`graph_builder.add_node`实现的，它系统地将*过滤*、*摘要*和*响应生成*节点添加到管道中。

此外，我们建立了**条件边**以简化执行。例如，如果过滤节点将电子邮件分类为\*\*“垃圾邮件”**，管道会**提前终止\*\*，立即将其导向`END`状态。

函数`graph = graph_builder.compile()`确保所有节点和条件路径都被正确构建，而`graph.invoke(state)`执行完整的流程，根据定义的逻辑处理电子邮件。

### 4.3 main.py

`main.py`脚本作为AI电子邮件代理的**入口点**，整合了所有必要组件——代理、监督逻辑和用户交互。它管理**电子邮件检索、处理和回复生成**，同时处理用户输入以**发送或草拟电子邮件**。

```python
from agents import filtering_agent, summarization_agent, response_agent, human_review_agent  
from langgraph.graph import START, END, StateGraph  
from core.email_imap import fetch_imap_emails  
from core.email_sender import send_email, send_draft_to_gmail  
from utils.logger import get_logger  
from config import IMAP_USERNAME, IMAP_PASSWORD, IMAP_SERVER  
from core.supervisor import supervisor_langgraph  
from core.state import EmailState  
  
  
logger = get_logger(__name__) # 初始化终端日志记录器  
  
def process_email_action(email, your_name): # 决定发送或草拟电子邮件  
    action = input("您想(s)发送邮件还是(d)草拟到Gmail？(s/d)：").strip().lower()  
    if action == "s":  
        if send_email(email, your_name):  
            logger.info("邮件发送成功。")  
        else:  
            logger.warning("邮件发送失败。")  
    elif action == "d": # 如果草拟则输入电子邮件  
        gmail_address = input("请输入您的Gmail地址以草拟：")  
        if send_draft_to_gmail(email, your_name, gmail_address):  
            logger.info("草拟成功发送到Gmail。")  
        else:  
            logger.warning("草拟发送到Gmail失败。")  
    else:  
        logger.warning("无效选项。未采取任何操作。")  
  
def main():  
    logger.info("开始主函数。")  
      
    # 提示输入您的名字（用于签名）和收件人的名字。  
    your_name = input("请输入您的名字（用于签名）：")  
    recipient_name = input("请输入收件人的名字：")  
      
    # 使用IMAP检索实时电子邮件  
    emails = fetch_imap_emails(IMAP_USERNAME, IMAP_PASSWORD, IMAP_SERVER)  
    logger.debug(f"从IMAP检索到{len(emails)}封电子邮件。")  
      
    if not emails:  
        logger.info("没有找到电子邮件。")  
        return  
      
    latest_emails = emails[-5:] # 这告诉选择最近的5封电子邮件  
      
    print("\n请选择要处理的电子邮件：")  
    for idx, email in enumerate(latest_emails):  
        print(f"{idx + 1}. {email['subject']}")  
      
    choice = int(input("请输入您想要选择的电子邮件编号：")) - 1  
    if choice < 0 or choice >= len(latest_emails):  
        print("无效选择。退出。")  
        return  
      
    selected_email = latest_emails[choice]  
      
    # 创建状态并将电子邮件通过工作流处理  
    state = EmailState()  
    state.emails = [selected_email]  
    state.current_email = selected_email  
    # 将您的签名（您的名字）传递给监督者管道  
    state = supervisor_langgraph(selected_email, state, your_name,recipient_name)  
      
    print("\n生成的回复：\n")  
    print(selected_email.get("response", "未生成回复。"))  
      
    changes = input("您是否要修改回复？(y/n)：").strip().lower()  
    if changes == "y":  
        modified_response = input("请输入修改后的回复：")  
        selected_email["response"] = modified_response  
      
    # 现在处理最终动作（发送或草拟）  
    process_email_action(selected_email, your_name)  
      
    logger.info("所有电子邮件已处理。")  
    logger.debug(f"最终状态：{state}")  
  
if __name__ == "__main__":  
    main()

```

`main.py`函数首先使用IMAP检索电子邮件，从服务器获取最新的消息。

然后**显示最近的5封电子邮件**，并提示用户选择一封进行处理。一旦选择了电子邮件，它将通过`supervisor_langgraph`进行处理，该函数对电子邮件进行分类，生成摘要，并制定适当的回复。

在最终确定之前，脚本允许**人工审核和修改**，使用户能够在需要时完善AI生成的回复。

最后，用户可以选择**通过SMTP立即发送回复或将回复保存为草稿发送到Gmail**，确保灵活高效地处理电子邮件。

因此有效地将回复发送到所需电子邮件。

## 5、运行和理解代码

**配置你的环境：**

- 打开项目根目录中的`.env`文件。
- 提供IMAP凭据（用户名、密码、服务器、端口）以及LLM的API令牌。
- 如果您更喜欢使用OpenAI而不是Deepseek，请更新每个代理代码中调用LLM的部分中的库配置。

**安装所需库：**

- 确保您拥有所有必要的依赖项，运行：

```shell
pip install -r requirements.txt

```

- 此命令下载并安装项目所需的全部库。

**使用IMAP进行实时电子邮件处理：**

- 对于实时电子邮件处理，建议使用IMAP。代码设计为从服务器获取您的实时电子邮件。
- 默认情况下，IMAP函数仅显示最后5封电子邮件（使用`latest_emails = emails[-5:]`在`main.py`中）。
- 如果您希望查看更多电子邮件，可以更改此参数。

**选择你的操作——发送或草拟：**

- 当提示时，按`s`发送电子邮件，或按`d`保存为草稿。
- 为了测试，如果您想尝试草拟功能，只需输入自己的电子邮件地址。系统将把草稿发送给您以供审阅。

按照这些步骤，您可以快速设置、运行并自定义电子邮件助理代理以满足您的电子邮件处理需求。

## 6、结束语

本项目展示了如何利用Deepseek与LangChain和LangGraph结合，创建一个模块化且高效的AI电子邮件代理。实验代码，分享您的见解，并随时贡献改进。

---

原文链接：[Building your first Agent with Deepseek : AI Ema](https://medium.com/@parthshr370/building-your-first-agent-with-deepseek-ai-email-agent-e6f17d3c290e?ref=hubwiz.com)**il Agent**

汇智网翻译整理，转载请标明出处