【AppBuilder工作流】AI智能邮箱助理，打通AI与工作生活之间的最后一公里

AI原生应用开发/技术交流

插件应用
千帆杯挑战赛
有奖征文

5月29日2375看过

前两天去百度玩，参加了一个现场AI应用开发小比赛，感受到了大家的热情，因此将应用构建原理记录下并贴下相关代码。

应用名称：智能邮件助理

应用体验链接：https://appbuilder.baidu.com/s/jyxQGLgo

如果本文对您有帮助的话，请务必帮我在如下比赛链接的我的作品点个赞👍，感激不尽！

https://cloud.baidu.com/qianfandev/matchs

进入上方比赛链接，找到智能邮件助理，右下角即可点赞~

智能邮件助理简介：登录邮箱获取最新邮件列表，给指定收件人按照要求编写并发送邮件，将AI和邮件服务器直接连接，打通AI辅助生活工作的最后一公里。效果图如下所示：

从ChatBot到Function Calling，为LLM安上手脚

从原理上来看，LLM的本质是一个句子接龙机器，于是自然而然地，大家首先会把他用在续写故事上，而稍加引导，让这个故事以机器一句、用户一句的方式接力续写，那么就可以把LLM包装成一个对话机器人，之后，以初代ChatGPT为代表的基于LLM的对话机器人应用铺天盖地的爆发了。

那么，难道LLM只能用来Chat吗？可否做一些更有趣的事情呢？如果翻阅过OpenAI发的几篇GPT1、2、3代的论文，就会发现，大语言模型的典型模型GPT在论文中被定位成了一个“多任务完成器”，认为GPT这种大语言模型，只要稍加引导，就可以化身不同的角色完成不同的任务。

所以，通过适当的引导（提示词工程），我们至少可以让大模型完成以下两件事情：

结构化信息抽取，从非结构化的文本中提取期望的关键信息
按照模板格式化输出，比如按照json格式输出

再配合上模型微调等技术强化大模型以上两个技能，我们就可以做到：给定大模型一段自然语言来描述的需求，要求大模型判断意图并从中抽取预期的关键信息，然后按照指定的格式模板输出这些关键信息列表，而输出的格式化信息是可以被传统的不具有语言智能的外围辅助程序解析的，外围程序就可以利用这个信息来映射到某个预先实现好的函数，并把模型输出的格式化信息列表当成函数参数列表传入，然后执行这个函数，再把函数的返回结果反馈给大模型。

于是宏观来看，好像大模型稍加调教，就拥有了理解并调用外界函数的功能，并且可以感知到函数执行结果并作出自己下一步的判断。这个“函数”是一个笼统的说法，也可以叫它工具、外部模块、工作流等。

LLM原本只能输入输出文本（详细来说是向量序列），通过让LLM使用外部工具，让LLM和外部环境进行交互，和真实世界建立了桥梁，让LLM真正的可以感知、影响到外部世界，就好像为LLM安上了手脚。

工作流，图形化方式方便快速的为LLM构建工具

百度AppBuilder最近推出的工作流功能，就是一个利用方便的图形化界面的方式，帮助大家在无需要求很高的代码能力的前提下，利用拼积木的形式，快速帮助LLM构建出可以被AI理解、调用的工具。

尤其是，这个功能正在快速迭代更新中，变得越来越好用👇。

新上线的speed思考模型，又快又好用还不要钱，能让应答速度提升较多；Agent的问答模型也支持接入文心全系列模型和其他开源模型

工作流新增代码块节点，支持直接引入组件，可以实现组件套组件功能

目前的工作流支持

工作流是一种图形化编程手段，利用不同功能的节点，配置好每个节点的输入输出，根据逻辑连接不同的节点，把开始节点当成函数入参列表，把结束节点当成函数返回结果列表，整体上来看工作流就是一个函数，那么我们只需要利用工作流应用提供的节点来编写我们的逻辑就好了。

目前版本的工作流(20240530)的提供了5种基础节点：

大模型节点：提供自然语言任务处理能力
知识库节点：提供知识召回能力（一般用于RAG）
API节点：用于调用第三方Web服务
分支器节点：相当于代码中的if语句
代码节点：提供简单的代码处理能力

除了基础节点，官方还提供了组件节点等“高级封装函数”来供在工作流中调用。

通过这些节点的组合，我们可以设计出各种各样功能的工具来让LLM如虎添翼。

智能邮件助理工作流，越过人类，让LLM拥有直接和邮箱服务器沟通交互的能力

邮箱是我们生活工作中少不了的应用，尤其是高度依赖邮箱处理事务的办公人员，可是当面对大量的邮件阅读任务和邮件回复任务，对于办公人员无论从心理还是脑力上都做出了不小的要求。

基于LLM的应用的出现，一定程度上缓解了这个问题。LLM可以帮助我们快速阅读中文外文邮件，帮我们用母语进行总结提炼主旨大意；帮助我们快速根据需求撰写出高质量的邮件，避免了字斟句酌的纠结辞藻语句。

然而，根据我的调查感知，当前市面上大部分的邮件相关LLM应用都没有解决邮件自动阅读、撰写的“最后一公里”问题，具体表现为：

阅读未读邮件：进入访问邮箱网页、输入账户密码登录邮箱、点击收件箱、检查最新未读邮件、把邮件内容复制粘贴给ChatGPT等应用、撰写提示词要求ChatGPT总结提炼翻译邮件、等待查看ChatGPT给出的结果。

撰写回复邮件：进入ChatGPT应用、撰写提示词给ChatGPT布置任务提需求、等待ChatGPT给出邮件撰写结果、复制生成的邮件文本、进入邮箱点击新建邮件按钮、粘贴邮件主题、选择收件人、粘贴ChatGPT生成的邮件正文、点击发送邮件按钮。

可以虽然AI帮我们解决了最耗时间的邮件阅读、撰写问题，但是步骤依然繁琐，里极致的自动化还有一定距离。

那么我们可否让AI帮我们一步到位？

试想以下场景，某天你躺在床上不愿意起来，但是你的工作又要求你需要检查邮箱里是否有客户关于某个特定主题的新问题邮件，并且你需要给客户用正式官方的语气进行邮件回复。

如果能有这样的AI智能邮件助理呢：你躺在床上直接说，“小度小度，帮我看看我的1号工作邮箱里是否有A公司发来的关于B主题的问题，如果有就跟我讲讲它大概在说啥”，AI此时应当使用你之前预配置好的1号邮箱账户密码登录邮件服务器获取维度邮件列表，然后检查是否有契合你所关心的B主题的邮件，并报告给你听；你经过专业的业务能力略加分析知道了如何解决用户问题，于是你说：“小度小度，你先查阅X主题的知识库，然后告诉这个客户应当先做什么后做什么就可以解决问题了，但是注意你的邮件撰写语气一定要官方且正式，最重要的是极致的礼貌”，于是几秒钟后，客户的邮箱里收到了关于他的问题的礼貌的、官方的回复，用户很开心，并为能如此快速的答复点了赞。

可以看到，这种模式是让AI绕过了人这个中介，直接同邮件服务器进行交互，一步直达任务完成，即方便又高效。

其中的关键就在于构建一个工具让AI拥有直接同邮件服务器交互的能力。

邮件智能助理工作流的构建

在这里创建工作流组件：https://console.bce.baidu.com/ai_apaas/personalSpace?tabKey=component，其中组件描述要用心填写，这个字段是重点给AI看的，让AI知道这个组件的功能是什么，以便在合适的时机发起调用。

邮件智能助理工作流是一个函数，这个函数能够被AI调用，并且这个函数我打算把他设计为同时身兼获取邮件列表、按要求发送邮件这两个功能的函数。

设计函数首先要考虑入参列表和返回值列表，既然此工作流身兼二职，那么我们起码就需要一个参数 user_intent （用户意图）来表示想让这个工作流收邮件还是发邮件，其次，不管是收发邮件，都需要用户邮箱账户(email_account)和授权密码(account_secret)来登录邮件服务器，还有，如果是发送邮件，那么还需要收件人邮箱(send_mail_to)、邮件主题(send_mail_subject)、邮件正文(send_mail_body) 这三个可选参数。

因此构建工作流开始节点如下图（注意写好每个入参的自然语言描述方便AI理解参考参数作用）：

我们希望函数能够返回处理结果信息，对于拉取最新邮件，我们希望返回一个邮件列表，对于发送邮件，我们希望返回成功与否，因此工作流结束节点也可以填充响应变量。

函数入参和返回值都规定好了，那么接下来就是该开始实现这个函数了，很简单的思路便是根据用户意图参数的值，借助分支器节点把执行流程导向两个API节点，并汇总执行结果到结束节点，就完成了。

API节点配置以及后端API服务搭建

在智能邮件助理工作流中，配置项最复杂的节点是API节点，可以把API节点看成是实现对于某个运行在第三方服务器上服务的远程调用，为了更加具有灵活性，我们期望在第三方服务器上搭建一个邮件服务，并暴露两个接口：

读取指定邮箱的最新邮件列表
使用指定邮箱向指定收件人发送指定内容邮件

我们使用python Flask框架搭建后端API接口，关键代码如下：

  
  
   
   
    # ......
   
   
   
   
    from Email_Assistant_Backend import email_api
   
   
   
   
    
   
   
   
   
    @app.route('/Email_Assistant_Backend/')
   
   
   
   
    def index_Email_Assistant_Backend():
   
   
   
   
        return "<h1>邮件助手后端正在运行...</h1>"
   
   
   
   
    
   
   
   
   
    @app.route('/Email_Assistant_Backend/get-mails-list')
   
   
   
   
    def get_mails_list():
   
   
   
   
        return email_api.get_mails_list()
   
   
   
   
    
   
   
   
   
    @app.route('/Email_Assistant_Backend/send-mail')
   
   
   
   
    def send_mail():
   
   
   
   
        return email_api.send_mail()
   
   
   
   
    # ......

我们实现一个可以和服务器交互的 class Mail163: ，按照邮件服务器通信协议（pop3/smtp）实现了邮箱登录、邮件读取、邮件解析、邮件发送等功能，放到单独一个模块里，代码如下：

  
  
   
   
    # 163邮箱 邮件收发模块
   
   
   
   
    # 163邮箱的SMTP服务器地址为：smtp.163.com
   
   
   
   
    # 163邮箱的POP3服务器地址为：pop.163.com
   
   
   
   
    # 163邮箱的IMAP服务器地址为：imap.163.com
   
   
   
   
    
   
   
   
   
    import smtplib
   
   
   
   
    import poplib
   
   
   
   
    import imaplib
   
   
   
   
    from email.mime.text import MIMEText
   
   
   
   
    from email.header import Header
   
   
   
   
    from email import policy
   
   
   
   
    from email.parser import BytesParser
   
   
   
   
    from email.message import Message
   
   
   
   
    from html2text import html2text
   
   
   
   
    
   
   
   
   
    
   
   
   
   
    class Mail163:
   
   
   
   
        def __init__(self, account, secret_code):
   
   
   
   
            self.smtp_server = 'smtp.163.com'
   
   
   
   
            self.pop3_server = 'pop.163.com'
   
   
   
   
            self.imap_server = 'imap.163.com'
   
   
   
   
            self.account = account
   
   
   
   
            self.secret_code = secret_code
   
   
   
   
    
   
   
   
   
    
   
   
   
   
        def parse_email(self, email: bytes):
   
   
   
   
            """Parse the email content"""
   
   
   
   
            # 解析邮件内容
   
   
   
   
            # 导入Message
   
   
   
   
            
   
   
   
   
            msg: Message = BytesParser(policy=policy.default).parsebytes(email)
   
   
   
   
    
   
   
   
   
            subject = msg.get('Subject')
   
   
   
   
            from_info = msg.get('From')
   
   
   
   
            to_info = msg.get('To')
   
   
   
   
            date_info = msg.get('Date')
   
   
   
   
    
   
   
   
   
            # 邮件正文, 可以是txt或html
   
   
   
   
            body = msg.get_body(preferencelist=('plain', 'html')).get_content()
   
   
   
   
            # print(body)
   
   
   
   
            # 邮件附件
   
   
   
   
            attachments = msg.iter_attachments()
   
   
   
   
    
   
   
   
   
            # print(f"Subject: {subject}")
   
   
   
   
            # print(f"From: {from_info}")
   
   
   
   
            # print(f"To: {to_info}")
   
   
   
   
            # print(f"Date: {date_info}")
   
   
   
   
            # print(f"Text Body:\n{body[:10]}")
   
   
   
   
            # print(f"Attachments: {list(attachments)}")
   
   
   
   
    
   
   
   
   
            return {
   
   
   
   
                'subject': subject,
   
   
   
   
                'from': from_info,
   
   
   
   
                'to': to_info,
   
   
   
   
                'date': date_info,
   
   
   
   
                'body': body,
   
   
   
   
                'attachments_iter': attachments,
   
   
   
   
            }
   
   
   
   
    
   
   
   
   
    
   
   
   
   
    
   
   
   
   
        def list_mail(self):
   
   
   
   
            """List the emails in the mailbox"""
   
   
   
   
            server = poplib.POP3(self.pop3_server)
   
   
   
   
            server.user(self.account)
   
   
   
   
            server.pass_(self.secret_code)
   
   
   
   
            print('Messages: %s. Size: %s' % server.stat())
   
   
   
   
            # print(f"共有 {server.stat()[0]} 封邮件, {server.stat()[1]} 字节")
   
   
   
   
            resp, mails, octets = server.list()
   
   
   
   
            
   
   
   
   
            mails_list = []
   
   
   
   
            for mail in mails:
   
   
   
   
                msg_number, size_in_bytes = mail.split()
   
   
   
   
                # print(f"Message number: {msg_number.decode()}, Size in bytes: {size_in_bytes.decode()}")
   
   
   
   
                # Retrieve the email
   
   
   
   
                resp, lines, octets = server.retr(int(msg_number))
   
   
   
   
                email = b'\n'.join(lines)
   
   
   
   
                # Parse the email content
   
   
   
   
                email_obj = self.parse_email(email)
   
   
   
   
    
   
   
   
   
                email_obj['body'] = html2text(email_obj['body'])
   
   
   
   
                email_obj['message_number'] = msg_number.decode()
   
   
   
   
                mails_list.append(email_obj)
   
   
   
   
            
   
   
   
   
            server.quit()
   
   
   
   
    
   
   
   
   
            return mails_list
   
   
   
   
    
   
   
   
   
                
   
   
   
   
                # print('---------------------------------')
   
   
   
   
                # print('主题:', email_obj['subject'])
   
   
   
   
                # print('发件人:', email_obj['from'])
   
   
   
   
                # print('收件人:', email_obj['to'])
   
   
   
   
                # print('日期:', email_obj['date'])
   
   
   
   
                # print('正文:', email_obj['body'])
   
   
   
   
                # print('正文:', md(email_obj['body']))
   
   
   
   
                # print('正文:', html2text(email_obj['body']))
   
   
   
   
                # attachments = list(email_obj['attachments_iter'])
   
   
   
   
                # if attachments:
   
   
   
   
                #     print('附件:')
   
   
   
   
                #     for attachment in attachments:
   
   
   
   
                #         print(attachment.get_filename())
   
   
   
   
                #         print(attachment.get_content_type())
   
   
   
   
                #         print(attachment.get_content_disposition()) # 这个是附件的类型, inline是内联附件, attachment是普通附件,他们的区别是inline会直接显示在邮件中,而attachment是需要下载的
   
   
   
   
                #         print(attachment.get_payload())
   
   
   
   
    
   
   
   
   
                # print('---------------------------------')
   
   
   
   
    
   
   
   
   
        def list_mails_imap(self, limit=10, folder='INBOX'):
   
   
   
   
            """List the emails in the mailbox"""
   
   
   
   
            server = imaplib.IMAP4_SSL(self.imap_server)
   
   
   
   
            server.login(self.account, self.secret_code)
   
   
   
   
            imap_id = ("name", "Rabbak", "version", "1.0.0", "vendor", "PythonClient")
   
   
   
   
            # xatom方法为imaplib原生的提供调用自定义命令的方法
   
   
   
   
            # https://docs.python.org/zh-cn/3.5/library/imaplib.html#imaplib.IMAP4.xatom
   
   
   
   
            typ, data = server.xatom('ID', '("' + '" "'.join(imap_id) + '")')
   
   
   
   
    
   
   
   
   
            server.select(folder)
   
   
   
   
            # search for all emails
   
   
   
   
            status, email_ids = server.search(None, 'ALL')
   
   
   
   
            email_ids = email_ids[0].split()
   
   
   
   
            mails_list = []
   
   
   
   
            for email_id in email_ids[-limit:]:
   
   
   
   
                status, email_data = server.fetch(email_id, '(BODY.PEEK[])')
   
   
   
   
                raw_email = email_data[0][1]#.decode("utf-8")
   
   
   
   
                # print(email_data[0][1])
   
   
   
   
                # exit()
   
   
   
   
                # email_message = email.message_from_string(raw_email)
   
   
   
   
                # email_obj = self.parse_email(email_message)
   
   
   
   
                email_obj = self.parse_email(raw_email)
   
   
   
   
                email_obj['message_number'] = email_id.decode()
   
   
   
   
                mails_list.append(email_obj)
   
   
   
   
            server.close()
   
   
   
   
            server.logout()
   
   
   
   
            return mails_list
   
   
   
   
            
   
   
   
   
        
   
   
   
   
        def send_mail(self, to, subject, body):
   
   
   
   
            """Send an email"""
   
   
   
   
            msg = MIMEText(body, 'plain', 'utf-8')
   
   
   
   
            msg['From'] = self.account
   
   
   
   
            msg['To'] = to
   
   
   
   
            msg['Subject'] = Header(subject, 'utf-8').encode()
   
   
   
   
            server = smtplib.SMTP(self.smtp_server, 25)
   
   
   
   
            # server.set_debuglevel(1)
   
   
   
   
            server.login(self.account, self.secret_code)
   
   
   
   
            server.sendmail(self.account, [to], msg.as_string())
   
   
   
   
            server.quit()
   
   
   
   
        
   
   
   
   
        def delete_mail(self, msg_number):
   
   
   
   
            """Delete an email"""
   
   
   
   
            server = poplib.POP3(self.pop3_server)
   
   
   
   
            # log level
   
   
   
   
            # server.set_debuglevel(1)
   
   
   
   
            server.user(self.account)
   
   
   
   
            server.pass_(self.secret_code)
   
   
   
   
            # delete the email
   
   
   
   
            server.dele(msg_number)
   
   
   
   
            server.quit()
   
   
   
   
        
   
   
   
   
        # def list_unread_mail(self):
   
   
   
   
        #     """List unread emails in the mailbox"""
   
   
   
   
        #     server = imaplib.IMAP4_SSL(self.imap_server)
   
   
   
   
        #     server.login(self.account, self.secret_code)
   
   
   
   
        #     server.select('INBOX')
   
   
   
   
        #     # search for unread emails
   
   
   
   
        #     status, email_ids = server.search(None, 'UNSEEN')
   
   
   
   
        #     email_ids = email_ids[0].split()
   
   
   
   
        #     unread_mails = []
   
   
   
   
        #     for email_id in email_ids:
   
   
   
   
        #         status, email_data = server.fetch(email_id, '(BODY.PEEK[])')
   
   
   
   
        #         raw_email = email_data[0][1].decode("utf-8")
   
   
   
   
        #         email_message = email.message_from_string(raw_email)
   
   
   
   
        #         email_obj = self.parse_email(email_message)
   
   
   
   
        #         email_obj['message_number'] = email_id.decode()
   
   
   
   
        #         unread_mails.append(email_obj)
   
   
   
   
        #     server.close()
   
   
   
   
        #     server.logout()
   
   
   
   
        #     return unread_mails
   
   
   
   
        
   
   
   
   
    
   
   
   
   
    if __name__ == '__main__':
   
   
   
   
        account = 'cXXXXXXXXXX@163.com'
   
   
   
   
        # secret = 'STA*********JB'
   
   
   
   
        mail = Mail163(account, secret)
   
   
   
   
        mails = mail.list_mail()
   
   
   
   
        for mail in mails:
   
   
   
   
            print('---------------------------------')
   
   
   
   
            print('编号:', mail['message_number'])
   
   
   
   
            print('主题:', mail['subject'])
   
   
   
   
            print('发件人:', mail['from'])
   
   
   
   
            print('---------------------------------')
   
   
   
   
        # last_mail = mails[-1]
   
   
   
   
        # delete the last email
   
   
   
   
        # mail.delete_mail(last_mail['message_number'])
   
   
   
   
    
   
   
   
   
        # to = 'yyyyyyyyyyyyyyy@163.com'
   
   
   
   
        # subject = 'Hello'
   
   
   
   
        # body = 'Hello, this is a test email.'
   
   
   
   
        # mail.send_mail(to, subject, body)
   
   
   
   
    
   
   
   
   
        # unread_mails = mail.list_unread_mail()
   
   
   
   
        # for mail in unread_mails:
   
   
   
   
        #     print('---------------------------------')
   
   
   
   
        #     print('编号:', mail['message_number'])
   
   
   
   
        #     print('主题:', mail['subject'])
   
   
   
   
        #     print('发件人:', mail['from'])
   
   
   
   
        #     print('---------------------------------')

接下来是对于两个接口的具体实现，用来暴露给工作流的API节点。

其中我实现了一个check_secret_key装饰器，用于接口鉴权。

可以在返回的参数中加一些使用自然语言描述状态的字段，以便最终将这些信息传递给AI理解接口调用结果。

代码如下：

  
  
   
   
    from flask import request, jsonify
   
   
   
   
    from .mail_mod import Mail163
   
   
   
   
    from .config import API_SECRET_KEY
   
   
   
   
    
   
   
   
   
    # 装饰器，检查请求携带的secret_key是否正确
   
   
   
   
    def check_secret_key(func):
   
   
   
   
        def wrapper(*args, **kwargs):
   
   
   
   
            # 检查请求中是否包含在工作流API节点中配置的密钥以及是否正确
   
   
   
   
            # secret_key = request.headers.get('secret_key')
   
   
   
   
            secret_key = request.args.get('secret_key')
   
   
   
   
            if secret_key != API_SECRET_KEY:
   
   
   
   
                return jsonify({
   
   
   
   
                    'status': 'error, 工作流API密钥设置错误',
   
   
   
   
                    # 'msg': '工作流API密钥设置错误'
   
   
   
   
                })
   
   
   
   
            return func(*args, **kwargs)
   
   
   
   
        return wrapper
   
   
   
   
    
   
   
   
   
    
   
   
   
   
    # 获取指定邮箱最新邮件列表
   
   
   
   
    @check_secret_key
   
   
   
   
    def get_mails_list():
   
   
   
   
        # 获取get参数
   
   
   
   
        email_account = request.args.get('email_account')
   
   
   
   
        account_secret = request.args.get('account_secret')
   
   
   
   
        if not email_account or not account_secret:
   
   
   
   
            return jsonify({
   
   
   
   
                'status': 'error,need email_account or account_secret',
   
   
   
   
                'msg': 'need email_account or account_secret'
   
   
   
   
            })
   
   
   
   
        
   
   
   
   
        mail163 = Mail163(email_account, account_secret)
   
   
   
   
        try:
   
   
   
   
            mails = mail163.list_mail()
   
   
   
   
        except Exception as e:
   
   
   
   
            return jsonify({
   
   
   
   
                'status': f'error,邮箱登录失败，Err:{e}',
   
   
   
   
                'msg': f"邮箱登录失败，Err:{e}"
   
   
   
   
            })
   
   
   
   
       
   
   
   
   
        mail_list = [{
   
   
   
   
            'email_subject': mail['subject'],
   
   
   
   
            'email_from': mail['from'],
   
   
   
   
            'email_to': mail['to'],
   
   
   
   
            'email_date': mail['date'],
   
   
   
   
            'email_body': mail['body'],
   
   
   
   
            'email_attachments': [attachment.get_filename() for attachment in mail['attachments_iter']]
   
   
   
   
        } for mail in mails]
   
   
   
   
        
   
   
   
   
        return jsonify({
   
   
   
   
            'status': 'success',
   
   
   
   
            'data': mail_list
   
   
   
   
        })
   
   
   
   
    
   
   
   
   
    # 登录指定邮箱向指定收件人发送指定内容邮件
   
   
   
   
    @check_secret_key
   
   
   
   
    def send_mail():
   
   
   
   
        # 获取get参数
   
   
   
   
        email_account = request.args.get('email_account')
   
   
   
   
        account_secret = request.args.get('account_secret')
   
   
   
   
        mail_to = request.args.get('mail_to')
   
   
   
   
        mail_subject = request.args.get('mail_subject')
   
   
   
   
        mail_body = request.args.get('mail_body')
   
   
   
   
        if not email_account or not account_secret:
   
   
   
   
            return jsonify({
   
   
   
   
                'status': 'error',
   
   
   
   
                'msg': 'need email_account or account_secret'
   
   
   
   
            })
   
   
   
   
        if not mail_to or not mail_subject or not mail_body:
   
   
   
   
            return jsonify({
   
   
   
   
                'status': 'error',
   
   
   
   
                'msg': 'need mail_to or mail_subject or mail_body'
   
   
   
   
            })
   
   
   
   
        
   
   
   
   
        mail163 = Mail163(email_account, account_secret)
   
   
   
   
        try:
   
   
   
   
            mails = mail163.list_mail()
   
   
   
   
        except Exception as e:
   
   
   
   
            return jsonify({
   
   
   
   
                'status': f'error,邮箱登录失败，Err:{e}',
   
   
   
   
                'msg': f"邮箱登录失败，Err:{e}"
   
   
   
   
            })
   
   
   
   
        mail163.send_mail(mail_to, mail_subject, mail_body)
   
   
   
   
    
   
   
   
   
        return jsonify({
   
   
   
   
            'status': 'success',
   
   
   
   
            'msg': 'send mail success'
   
   
   
   
        })

当然，我设置了一个单独的配置文件来存放接口服务鉴权密钥等常量

  
  
   
   
    # 服务认证密钥，在千帆平台工作流的API调用中需要使用
   
   
   
   
    API_SECRET_KEY = 'This-is-o-secret-key-for-APl-caIIing-!'

至此，项目代码模块组织结构如下：

可以把服务架设到个人服务器或者第三方托管服务器上，或者大部分云服务厂商都提供云函数等Serverless服务，也可以使用。

可以给服务器绑定自己的域名或者使用平台提供的域名。

服务架设成功后如下图所示。

然后，在工作流中的API节点填写接口请求地址，配置接口鉴权，在上边代码中我使用的是API Key方式，密钥填在Query地方，密钥值是config.py中配置的密钥值。以获取新邮件的/get-mails-list接口为例，如下图：

请求参数，是定义API节点请求接口时候携带的参数配置，要按照后端服务接口要求的入参填写，包括参数名、参数类型、请求方式等。以/get-mails-list接口为例，如下图：

返回参数，是断言这个服务接口会返回哪些字段，同样需要参照服务接口实现的返回字段来定义，比如参数名、参数类型和是否必须。以/get-mails-list接口为例，如下图：

API调试，是希望确保API的后端服务正常运行以及以上的接口配置正确，可以填入测试请求参数，点击运行按钮来发起接口调用测试，可以在右侧查看接口返回结果，测试通过后，这个API节点结算配置完成了。以/get-mails-list接口为例，如下图：

定义完API节点并调试通过后，需要给这个API节点定义输入参数来源，这里我们引用来自之前节点的变量。

工作流的调试与发布

点击工作流画布中的调试按钮，填入工作流开始节点要求的参数，点击开始运行按钮，如果没有意外的话，所有节点都会显示“成功”字样，右下方会显示结束节点输出结果（也是工作流输出结果）。

调试通过后，点击画布的右上角发布按钮，即可成功发布此工作流组件，以便让基于LLM的Agen调用此工具。

智能邮件助理Agent构建和配置

创建一个新的应用（https://console.bce.baidu.com/ai_apaas/personalSpace?tabKey=app），

设置好应用名称、头像、简介。

提示词要体现出它的角色，以及具体使命职责，还有约束，设置如下：

# 角色

你是一位邮件助理专家，可以使用邮箱智能助理组件读取和回复用户邮箱中的邮件。

## 技能

### 技能 1: 获取用户邮箱最新邮件（GET_MAILS）

当用户需要你读取最新邮件时，向用户请求登陆邮箱和登陆密码，你需要使用邮箱智能助理组件登录用户的邮箱，并讲用户意图参数user_intent置为GET_MAILS。获取用户邮箱内容后，把每一封邮件都按照以下格式输出每一封邮件：(其中email_body给出邮件总结)

💬邮件主题：${email_subject}

🎉 邮件发件人：${email_from}

🌈发件时间：${email_date}

📝邮件正文总结：

${Summary(email_body)}

---

### 技能 2: 根据需要智能发送邮件（SEND_MAILS）

1. 当用户需要你回复邮件时，你需要使用邮箱智能助理组件登录用户的邮箱。

2. 然后，你需要找到需要回复的邮件，并根据用户的需求生成回复内容。

3. 最后，你需要使用邮箱智能助理组件将回复内容发送给收件人，并将原文复述给用户。

## 限制

- 只处理与邮件相关的任务，不处理其他类型的任务。

- 必须使用邮箱智能助理组件读取和回复邮件，不能使用其他工具。

- 必须遵守邮件礼仪和法律法规，不能发送违法、违规或不道德的邮件。

当然，由于现场比赛时间原因，没有进一步对于提示词精调，仅是起一个基本实现思路以供大家参考。

在“能力扩展”区域引用刚刚创建的邮件助理组件，如下图：

配置好后配置界面整体看起来是这个样子的：

在右侧可以进行应用预览与调试，调试完成后进行发布。

开始体验智能邮件助理！

首先要知道自己的邮箱账户以及POP/STMP/IMAP授权密码，一般获取方式如下：

现在我想让AI帮我登录邮箱查看最新的邮件，我可以这么跟AI说：

帮我登录我的邮箱并查看最新邮件。

我的邮箱：c***********@163.com 授权码：STA***********PJB

AI智能邮件助理回复如下图：

可以看到，我们的智能邮件助理成功的帮我们从邮箱获取到了最新邮件，并且注意到，AI帮我们把原本的英文邮件提炼总结并翻译为了中文，非常方便贴心。下图是我邮箱中的原本的邮件原文：

接下来让我们来体验智能邮件助理的发邮件功能。

现在我想让AI帮我发一封邮件，内容是邀请大家来百度玩耍，我可以这么跟AI说：

我的邮箱账户：cxxxxxxxxx@163.com 我的邮箱授权密码：STA***********PJB

帮我给cyyyyyyyyyyy@163.com发送邮件，

主题是”明天来百度玩“

邮件内容正文说百度要举办线下开发者交流会，请对方来参加，邮件内容你帮我编写一下，注意文风要有吸引力的、简洁而正式的。

AI智能邮件助理回复如下图：