APPLICATION

构建Deepseek AI邮件代理

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

admin

Feb 18, 2025 • 14 min read

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

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

代理 = [ 工具 + LLM ]

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

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

1、项目概述与架构

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

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

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

2、工具的工作原理

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

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

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

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

3、电子邮件处理管道

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仓库

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

4.1 response_agent.py

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

# 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服务器：

# 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仓库中找到。

4.2 supervisor.py

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

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电子邮件代理的入口点，整合了所有必要组件——代理、监督逻辑和用户交互。它管理电子邮件检索、处理和回复生成，同时处理用户输入以发送或草拟电子邮件。

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的部分中的库配置。

安装所需库：

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

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 Email Agent

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