Outlook 邮件处理 Agent 的架构设计与实现 Outlook 邮件处理 Agent 的架构设计与实现引言从“邮件奴”到“智能秘书”的进化Outlook 作为企业级邮件系统的核心每天承载着海量的会议邀请、业务通知、报表推送。传统的手工处理模式筛选、分类、回复、归档效率低下且易出错。Outlook 邮件处理 Agent旨在通过 AI 大模型LLM与 Outlook API 的深度集成将自然语言指令转化为精准的邮件操作。用户只需说“把昨天收到的报销邮件附件下载到桌面”或“给项目组发一封下周会议提醒”Agent 即可自动完成全套动作实现邮件处理的“自动驾驶”。技术背景协议、库与框架选型邮件协议与 API 选型Microsoft Graph API推荐微软官方 RESTful API支持 OAuth 2.0 认证可跨平台云端/本地 Exchange操作邮件、日历、联系人是构建云端 Agent 的首选。SMTP/IMAP通用协议适合基础收发但功能受限如无法操作日历。Outlook COM 对象win32com仅限 Windows 本地安装 Outlook 客户端的场景依赖本地软件不适合服务器部署。Python 核心库O365对 Microsoft Graph API 的 Python 封装简化了认证和操作流程。LangChainAgent 编排框架提供工具调用Tool Calling和思维链ReAct能力。python-dotenv管理环境变量如 API Key、租户 ID。应用使用场景场景一智能收件箱管家分类、摘要、归档痛点收件箱被垃圾邮件、通知、重要邮件淹没。Agent 方案Agent 定时拉取新邮件利用 LLM 判断邮件类别紧急/普通/垃圾、提取关键信息如“会议时间变更至 14:00”并自动移动至对应文件夹如“待处理”、“归档”。场景二会议邀请与日程自动处理痛点频繁的会议邀请需手动点击接受/拒绝且易忘记。Agent 方案Agent 监听收件箱中的会议邀请解析时间、地点、参会人若与现有日程无冲突且主题包含“评审会”则自动接受并回复确认。场景三报表邮件自动抓取与数据入库痛点每日需手动下载邮件附件Excel 报表并导入数据库。Agent 方案Agent 识别发件人为“noreplyreport.com”且主题包含“日报”的邮件下载附件解析 Excel 内容并调用 API 写入业务系统。不同场景下详细代码实现架构核心Outlook 工具封装层O365# core/outlook_toolkit.pyimportosfromO365importAccount,MSGraphProtocolfromO365.messageimportMessagefromO365.mailboximportMailBoxfromtypingimportList,Dict,Any,Optionalimportbase64classOutlookToolkit:Outlook 操作工具集基于 O365 库def__init__(self,client_id:str,client_secret:str,tenant_id:str):self.client_idclient_id self.client_secretclient_secret self.tenant_idtenant_id self.accountself._authenticate()def_authenticate(self)-Account:OAuth 2.0 认证需提前获取 tokencredentials(self.client_id,self.client_secret)protocolMSGraphProtocol(api_versionv1.0)accountAccount(credentials,protocolprotocol,tenant_idself.tenant_id)# 检查是否已有 token否则提示授权ifnotaccount.is_authenticated:# 生产环境建议使用 token 文件持久化account.authenticate(scopes[https://graph.microsoft.com/Mail.ReadWrite,https://graph.microsoft.com/Mail.Send])returnaccountdefsearch_emails(self,folder:strinbox,subject_filter:strNone,unread_only:boolFalse,limit:int10)-List[Dict]:搜索邮件返回结构化数据供 LLM 分析mailbox:MailBoxself.account.mailbox()target_foldermailbox.get_folder(folder_namefolder)querytarget_folder.get_messages(limitlimit)ifunread_only:queryquery.filter(is_readFalse)ifsubject_filter:queryquery.filter(subject__containssubject_filter)emails[]formsginquery:emails.append({id:msg.object_id,subject:msg.subject,sender:msg.sender.nameifmsg.senderelseUnknown,received:msg.received.strftime(%Y-%m-%d %H:%M),body_preview:msg.body_preview[:100]ifmsg.body_previewelse,is_read:msg.is_read,has_attachments:msg.attachmentsisnotNone})returnemailsdefmark_as_read(self,message_id:str)-str:标记邮件为已读messageself.account.message(message_idmessage_id)message.mark_as_read()returnf邮件{message_id}已标记为已读defdownload_attachments(self,message_id:str,save_path:str)-str:下载邮件附件到指定目录messageself.account.message(message_idmessage_id)attachmentsmessage.attachments saved_files[]forattinattachments:ifatt.is_file:file_pathos.path.join(save_path,att.name)withopen(file_path,wb)asf:f.write(att.content)saved_files.append(att.name)returnf已下载附件:{, .join(saved_files)}ifsaved_fileselse无附件defsend_email(self,to_emails:List[str],subject:str,body:str,cc_emails:List[str]None)-str:发送邮件支持 HTML 正文messageself.account.new_message()message.to.add(to_emails)ifcc_emails:message.cc.add(cc_emails)message.subjectsubject message.bodybody message.send()returnf邮件已发送至:{, .join(to_emails)}defmove_email(self,message_id:str,target_folder:str)-str:移动邮件到指定文件夹messageself.account.message(message_idmessage_id)mailboxself.account.mailbox()foldermailbox.get_folder(folder_nametarget_folder)message.move(folder)returnf邮件已移动至:{target_folder}场景一代码智能收件箱管家完整可运行# scenarios/inbox_manager_agent.pyimportosfromcore.outlook_toolkitimportOutlookToolkitfromlangchain.agentsimportAgentType,initialize_agentfromlangchain_openaiimportChatOpenAIfromlangchain.toolsimportToolfromdatetimeimportdatetime,timedelta# 初始化 LLM 和工具llmChatOpenAI(modelgpt-4o-mini,temperature0)# 从环境变量加载配置client_idos.getenv(OUTLOOK_CLIENT_ID)client_secretos.getenv(OUTLOOK_CLIENT_SECRET)tenant_idos.getenv(TENANT_ID)outlook_toolsOutlookToolkit(client_id,client_secret,tenant_id)# 包装成 LangChain Toolstools[Tool(namesearch_emails,description搜索收件箱邮件。输入为字典folder(文件夹名), subject_filter(主题关键词), unread_only(是否只读未读), limit(数量)。,funcoutlook_tools.search_emails),Tool(namemark_as_read,description标记邮件为已读。输入为邮件ID。,funcoutlook_tools.mark_as_read),Tool(namemove_email,description移动邮件到指定文件夹。输入为 (message_id, target_folder)。,funcoutlook_tools.move_email)]agentinitialize_agent(tools,llm,agentAgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,verboseTrue)defrun_inbox_cleanup():智能清理收件箱分类并归档instruction 请处理我的收件箱 1. 搜索收件箱中未读的邮件最多 20 封 2. 对于每封邮件判断其类别会议邀请、通知、垃圾 3. 将会议邀请移动到 Meeting 文件夹通知类标记为已读并移动到 Notification 文件夹 4. 垃圾邮件直接删除标记为已读后移动到 Junk resultagent.run(instruction)print(Agent 执行结果:,result)if__name____main__:run_inbox_cleanup()场景二代码会议邀请自动处理集成日历逻辑# scenarios/meeting_agent.pyfromcore.outlook_toolkitimportOutlookToolkitfromO365.calendarimportSchedule,EventclassMeetingAgent:会议邀请自动处理 Agentdef__init__(self,outlook_toolkit:OutlookToolkit):self.tkoutlook_toolkit self.scheduleself.tk.account.schedule()defprocess_meeting_invitations(self):处理未读的会议邀请unread_emailsself.tk.search_emails(unread_onlyTrue,subject_filter会议邀请,limit5)foremailinunread_emails:# 解析邮件内容获取会议时间模拟 LLM 解析meeting_timeself._extract_meeting_time(email[body_preview])# 检查时间冲突ifnotself._has_schedule_conflict(meeting_time):# 自动接受self._accept_meeting(email[id])self.tk.mark_as_read(email[id])print(f已接受会议:{email[subject]})else:# 标记为需人工处理self.tk.move_email(email[id],Pending)print(f会议时间冲突已移至 Pending:{email[subject]})def_extract_meeting_time(self,body:str)-datetime:从邮件正文提取会议时间此处简化实际应用需 LLM 解析# 实际应调用 LLM 提取时间字符串并转换为 datetimereturndatetime.now()timedelta(hours2)# 模拟返回 2 小时后def_has_schedule_conflict(self,meeting_time:datetime)-bool:检查该时间段是否有其他会议eventsself.schedule.get_events(startmeeting_time,endmeeting_timetimedelta(hours1))returnlen(events)0def_accept_meeting(self,message_id:str):接受会议邀请需通过日历 APImessageself.tk.account.message(message_idmessage_id)# 此处为伪代码实际需通过 calendar API 接受事件# event message.as_event()# event.accept(Accepted by Agent)pass# 使用示例if__name____main__:outlook_tkOutlookToolkit(os.getenv(CLIENT_ID),os.getenv(CLIENT_SECRET),os.getenv(TENANT_ID))agentMeetingAgent(outlook_tk)agent.process_meeting_invitations()场景三代码报表邮件自动抓取与入库# scenarios/report_agent.pyimportpandasaspdimportsqlite3fromcore.outlook_toolkitimportOutlookToolkitclassReportAgent:报表邮件自动抓取 Agentdef__init__(self,outlook_toolkit:OutlookToolkit,db_path:str):self.tkoutlook_toolkit self.db_pathdb_path self.connsqlite3.connect(db_path)defprocess_daily_reports(self,sender:str,keyword:str):处理指定发件人的日报邮件emailsself.tk.search_emails(subject_filterkeyword,limit5)foremailinemails:ifemail[sender]sender:# 下载附件到临时目录self.tk.download_attachments(email[id],./temp)# 解析 Excel 并入库假设附件为 xlsxself._parse_excel_to_db(./temp/report.xlsx)# 标记已处理self.tk.mark_as_read(email[id])self.tk.move_email(email[id],Processed)print(f已处理报表:{email[subject]})def_parse_excel_to_db(self,file_path:str):解析 Excel 文件并写入 SQLitedfpd.read_excel(file_path)# 假设表结构为 (date, revenue, cost)df.to_sql(daily_report,self.conn,if_existsappend,indexFalse)# 使用示例if__name____main__:outlook_tkOutlookToolkit(os.getenv(CLIENT_ID),os.getenv(CLIENT_SECRET),os.getenv(TENANT_ID))agentReportAgent(outlook_tk,reports.db)agent.process_daily_reports(noreplyreport.com,销售日报)原理解释Agent 如何“理解”邮件操作核心特性工具化封装将 Outlook API 的复杂操作如 OAuth 认证、邮件查询封装成原子化的ToolAgent 只需调用高阶接口无需生成底层代码。意图识别与路由LLM 将用户指令如“下载昨天的报表附件”解析为工具调用序列search_emails→download_attachments。状态记忆通过 LangChain 的 Memory 组件Agent 能记住用户的上次操作如“上次处理的邮件 ID”实现增量处理。原理流程图用户指令自然语言 ↓ LLM 解析意图并生成 Action Plan ↓ 调用 OutlookToolkit 工具如 search_emails ↓ O365 库执行 Graph API 请求 ↓ 返回邮件数据给 LLM 进行后续决策 ↓ 调用后续工具download_attachments/move_email ↓ 返回最终结果给用户环境准备1. 微软 Azure 应用注册访问 https://portal.azure.com/注册新应用。添加Mail.ReadWrite和Mail.SendAPI 权限。获取Client ID、Client Secret、Tenant ID。2. Python 环境3.10pipinstallO365 langchain-openai langchain python-dotenv pandas sqlite33. 环境变量配置创建.env文件OUTLOOK_CLIENT_IDyour_client_id OUTLOOK_CLIENT_SECRETyour_client_secret TENANT_IDyour_tenant_id OPENAI_API_KEYsk-xxx运行结果示例收件箱管家执行日志 Entering new AgentExecutor chain... Thought: 用户需要处理未读邮件并分类。我需要先搜索未读邮件。 Action: search_emails({folder: inbox, unread_only: true, limit: 20}) Observation: [{id: AAMkAD..., subject: 项目评审会邀请, ...}] Action: move_email(AAMkAD..., Meeting) Observation: 邮件已移动至: Meeting Finished chain. Agent 执行结果: 已处理 5 封未读邮件其中 3 封会议邀请已归档至 Meeting 文件夹。测试步骤与验证单元测试Pytest# test_outlook_toolkit.pyimportpytestfromcore.outlook_toolkitimportOutlookToolkitpytest.fixturedefoutlook_tk():returnOutlookToolkit(test_id,test_secret,test_tenant)deftest_search_emails(outlook_tk):emailsoutlook_tk.search_emails(limit1)assertisinstance(emails,list)deftest_mark_as_read(outlook_tk):# 需先获取一封真实邮件的 IDresultoutlook_tk.mark_as_read(test_msg_id)assert已标记为已读inresult集成测试端到端# test_inbox_agent.pydeftest_inbox_cleanup():agentInboxManagerAgent()resultagent.run(清理未读邮件)assert已处理inresultassert移动至inresult部署场景桌面应用PyInstaller 图形界面# 打包为 exepyinstaller--onefile--add-data.env;.scenarios/inbox_manager_agent.py云函数Azure Functions# __init__.pyimportazure.functionsasfuncfromscenarios.report_agentimportReportAgentdefmain(timer:func.TimerRequest)-None:# 定时触发报表处理agentReportAgent(...)agent.process_daily_reports(...)疑难解答QAQ1OAuth 2.0 认证失败invalid_clientA1检查 Azure 应用注册的“重定向 URI”是否配置为http://localhost:53682/O365 库默认并确保 Client Secret 未过期。Q2Agent 误删重要邮件A2在生产环境中禁用永久删除操作。所有删除操作应先移动至“垃圾箱”文件夹并设置“人在环”Human-in-the-loop审批机制。Q3附件下载失败A3检查目标目录的写入权限并确保邮件确实包含附件通过has_attachments字段先做判断。未来展望与技术趋势多模态邮件处理Agent 不仅能解析文本还能识别邮件中的图片如截图中的表格并提取数据。RAG 增强结合企业知识库Agent 在回复邮件时可引用内部文档如“根据 Q3 销售政策折扣上限为 10%”。联邦学习在保护隐私的前提下利用本地模型处理敏感邮件仅上传摘要信息到云端。总结Outlook 邮件处理 Agent 的核心在于**“意图理解 API 封装”**。通过 LangChain 将自然语言指令转化为对 O365 工具集的调用我们实现了从“手动操作”到“智能代理”的跨越。本文提供的OutlookToolkit基础框架和三个典型场景收件箱管理、会议处理、报表抓取的完整代码可直接用于构建企业级邮件自动化系统显著提升办公效率。