全文共1.2w余字，预计阅读时间约24~40分钟 | 满满干货(附代码)，建议收藏！
本文目标：将Gmail API接入Chat模型，编写一个智能收发邮件的AI应用程序
代码下载点这里
一、背景
大模型应用开发从谷歌云入手进行学习和AI应用开发，是门槛最低、效率最高的方法。谷歌云上不仅有成百上千的各类应用的API，而且拥有统一的API制式和权限管理方法，调用过程非常方便，一个账号进行一次项目认证即可使用海量的API，外加详细完整的开发者文档，更是极大程度降低了上手使用API的门槛。
同时，谷歌云是功能完备的应用开发平台，如果不仅是尝试使用API进行前期的探索，而是希望真正意义上的完成企业级应用开发，也完全可以在谷歌云上进行。谷歌云不仅提供了完整的在线应用开发与发布流程，而且提供了（相对）廉价、稳定的云服务，开发者开发的应用程序可以直接在云端运行，并享受谷歌云提供的一整套应急、维护流程，以及实时可视化监控页面。
但是一个真实存在的问题是：国内还是存在访问限制，需要使用魔法，本文的目标是将Gmail API接入Chat模型，编写一个智能收发邮件的AI应用程序。
二、Gmail API的OAuth授权
首先第一步，要调用谷歌的API，肯定是要完成授权。具体谷歌云Google Cloud与谷歌云API库的介绍，及如何完成Gmail API的OAuth授权过程，如果您了解的话可以跳过这一步，直接进入下面代码内容，如不清楚如何操作的，请看下面链接：
Gmail API的OAuth授权过程的成功实现非常重要，使用谷歌云的其他API来构建更加复杂的AI应用程序，其中大多数都需要进行OAuth授权，如果已经完成了授权，则该凭证也是可以应用于其他API调用的，完成授权也是进行AI应用程序开发的前提。
AI应用开发：Gmail API如何实现OAuth授权
三、借助Gmail API构建智能邮件收发应用程序
3.1 在Chat模型中添加查阅邮件功能
先尝试将Gmail API接入Chat模型中。在这三篇文章中：
大模型开发(十一)：Chat Completions模型的Function calling功能详解
大模型开发(十二)：Function calling 流程优化并实现多轮对话任务
大模型开发(十三)：Function calling调用外部工具API，实现实时天气查询
已经跑通了优化后的Function calling功能执行流程，对于开发一个AI应用程序，在确定了基本功能实现的目标之后，总共分四步进行：
- 测试该外部函数功能是否具备可行性
- 验证大语言模型是否具备解读外部函数结果和准确翻译外部函数参数的能力
- 据此创建能够调用外部工具API外部函数，并编写非常详细函数说明
- 将定义好的外部函数带入run_conversation函数或者chat_with_model函数测试对话效果。
所以首先尝试通过外部函数赋予Chat模型查阅最近一封邮件的功能，具体实现过程如下：
- Step 1：测试外部函数功能可行性
先测试能否通过调用Gmail API查阅最近得一封邮件信息，包括发件人、日期和邮件内容，代码如下：
# 从本地文件中加载凭据
creds = Credentials.from_authorized_user_file('token.json')
# 创建 Gmail API 客户端
service = build('gmail', 'v1', credentials=creds)
# 列出用户的一封最新邮件
results = service.users().messages().list(userId='me', maxResults=1).execute()
messages = results.get('messages', [])
# 遍历邮件
for message in messages:
# 获取邮件的详细信息
msg = service.users().messages().get(userId='me', id=message['id']).execute()
# 获取邮件头部信息
headers = msg['payload']['headers']
# 提取发件人、发件时间
From, Date = "", ""
for h in headers:
name = h['name']
if name.lower() == 'from':
From = h['value']
if name.lower() == 'date':
Date = h['value']
# 提取邮件正文
if 'parts' in msg['payload']:
part = msg['payload']['parts'][0]
if part['mimeType'] == 'text/plain':
data = part['body']["data"]
else:
data = msg['payload']['body']["data"]
else:
data = msg['payload']['body']["data"]
data = data.replace("-","+").replace("_","/")
decoded_data = base64.b64decode(data)
str_text = str(decoded_data, "utf-8")
msg_str = email.message_from_string(str_text)
if msg_str.is_multipart():
text = msg_str.get_payload()[0]
else:
text = msg_str.get_payload()
print('From: {}'.format(From[:8]))
print('Date: {}'.format(Date))
print('Content: {}'.format(text))
看下输出结果：
看下邮箱状态：
上述代码的核心流程是通过service.users().messages().list(userId=‘me’, maxResults=1).execute()获取了最近一封邮件的信息，并最终保留在msg中。
更多关于Gmail API返回结果信息可参考Gmail API官网功能介绍：https://developers.google.com/gmail/api/guides
- Step 2：验证模型能否解读Gmail API返回结果
Gmail API 在不处理的情况下，返回的结果是这样的：
包括发件人信息、收件时间和邮件正文信息等各种详细的信息，为了满足更多不同问题的回答，考虑直接让Chat模型解读msg结果，通过如下流程测试Chat模型是否熟悉Gmail API这一功能及其返回结果的解读方式，代码如下：
import os
import openai
openai.api_key = os.getenv("OPENAI_API_KEY")
response = openai.ChatCompletion.create(
model="gpt-4-0613",
messages=[
{"role": "system", "content": "这是我的Gmail邮箱最近一封邮件的内容：%s" % msg},
{"role": "system", "content": "邮件内容是由Gmail API获取"},
{"role": "user", "content": "请问我的Gmail最近一封邮件是谁发送的，具体内容是什么？"}
]
)
response.choices[0].message['content']
看下输出结果：
Chat模型能够非常好的理解msg中所包含的信息。而后在设计外部函数和Chat模型的通信流程时，即可直接让外部函数对大语言模型直接输出msg结果。
- Step 3：创建外部函数
按照Step 1中的代码流程，创建一个能够返回最近一封邮件信息的msg对象，并输出为JSON格式：
def get_latest_email(userId):
"""
查询Gmail邮箱中最后一封邮件信息
:param userId: 必要参数，字符串类型，用于表示需要查询的邮箱ID，\
注意，当查询我的邮箱时，userId需要输入'me'；
:return：包含最后一封邮件全部信息的对象，该对象由Gmail API创建得到
"""
# 从本地文件中加载凭据
creds = Credentials.from_authorized_user_file('token.json')
# 创建 Gmail API 客户端
service = build('gmail', 'v1', credentials=creds)
# 列出用户的一封最新邮件
results = service.users().messages().list(userId=userId, maxResults=1).execute()
messages = results.get('messages', [])
# 遍历邮件
for message in messages:
# 获取邮件的详细信息
msg = service.users().messages().get(userId='me', id=message['id']).execute()
return json.dumps(msg)
functions_list = [get_latest_email]
在编写完函数之后，为了确保最终的对话函数能顺利运行，先手动测试auto_functions能否根据函数说明正确编写functions参数，以及测试下Chat模型能否正确识别functions参数格式要求，并将用户需求顺利解读和翻译为函数可以接受的参数形式，先引入functions参数自动编写函数：
def auto_functions(functions_list):
"""
Chat模型的functions参数编写函数
:param functions_list: 包含一个或者多个函数对象的列表；
:return：满足Chat模型functions参数要求的functions对象
"""
def functions_generate(functions_list):
# 创建空列表，用于保存每个函数的描述字典
functions = []
def chen_ming_algorithm(data):
"""
陈明算法函数，该函数定义了一种特殊的数据集计算过程
:param data: 必要参数，表示带入计算的数据表，用字符串进行表示
:return：陈明函数计算后的结果，返回结果为表示为JSON格式的Dataframe类型对象
"""
df_new = pd.read_json(data)
res = np.sum(df_new, axis=1) - 1
return res.to_json(orient='records')
chen_ming_function_description = inspect.getdoc(chen_ming_algorithm)
chen_ming_function_name = chen_ming_algorithm.__name__
chen_ming_function = {"name": "chen_ming_algorithm",
"description": "用于执行陈明算法的函数，定义了一种特殊的数据集计算过程",
"parameters": {"type": "object",
"properties": {"data": {"type": "string",
"description": "执行陈明算法的数据集"},
},
"required": ["data"],
},
}
# 对每个外部函数进行循环
for function in functions_list:
# 读取函数对象的函数说明
function_description = inspect.getdoc(function)
# 读取函数的函数名字符串
function_name = function.__name__
user_message1 = '以下是某函数说明：%s。' % chen_ming_function_description +\
'根据这个函数的函数说明，请帮我创建一个function对象，用于描述这个函数的基本情况。这个function对象是一个JSON格式的字典，\
这个字典有如下5点要求：\
1.字典总共有三个键值对；\
2.第一个键值对的Key是字符串name，value是该函数的名字：%s，也是字符串；\
3.第二个键值对的Key是字符串description，value是该函数的函数的功能说明，也是字符串；\
4.第三个键值对的Key是字符串parameters，value是一个JSON Schema对象，用于说明该函数的参数输入规范。\
5.输出结果必须是一个JSON格式的字典，只输出这个字典即可，前后不需要任何前后修饰或说明的语句' % chen_ming_function_name
assistant_message1 = json.dumps(chen_ming_function)
user_prompt = '现在有另一个函数，函数名为：%s；函数说明为：%s；\
请帮我仿造类似的格式为当前函数创建一个function对象。' % (function_name, function_description)
response = openai.ChatCompletion.create(
model="gpt-4-0613",
messages=[
{"role": "user", "name":"example_user", "content": user_message1},
{"role": "assistant", "name":"example_assistant", "content": assistant_message1},
{"role": "user", "name":"example_user", "content": user_prompt}]
)
functions.append(json.loads(response.choices[0].message['content']))
return functions
max_attempts = 3
attempts = 0
while attempts < max_attempts:
try:
functions = functions_generate(functions_list)
break # 如果代码成功执行，跳出循环
except Exception as e:
attempts += 1 # 增加尝试次数
print("发生错误：", e)
if attempts == max_attempts:
print("已达到最大尝试次数，程序终止。")
raise # 重新引发最后一个异常
else:
print("正在重新运行...")
return functions
看一下执行结果：
functions参数编写完全没问题，接下来测试Chat模型能否顺利创建满足格式的参数：
response = openai.ChatCompletion.create(
model="gpt-4-0613",
messages=[{"role": "user", "content": '请帮我查下我Gmail邮箱中最后一封邮件信息'}],
functions=functions,
function_call="auto",
)
看下结果：
- Step 4：验证对话效果
最后来测试下对话效果，仍然导入之前写的自动执行外部函数调用的Chat对话模型函数和多轮对话函数：
def run_conversation(messages, functions_list=None, model="gpt-4-0613"):
"""
能够自动执行外部函数调用的Chat对话模型
:param messages: 必要参数，字典类型，输入到Chat模型的messages参数对象
:param functions_list: 可选参数，默认为None，可以设置为包含全部外部函数的列表对象
:param model: Chat模型，可选参数，默认模型为gpt-4
:return：Chat模型输出结果
"""
# 如果没有外部函数库，则执行普通的对话任务
if functions_list == None:
response = openai.ChatCompletion.create(
model=model,
messages=messages,
)
response_message = response["choices"][0]["message"]
final_response = response_message["content"]
# 若存在外部函数库，则需要灵活选取外部函数并进行回答
else:
# 创建functions对象
functions = auto_functions(functions_list)
# 创建外部函数库字典
available_functions = {func.__name__: func for func in functions_list}
# first response
response = openai.ChatCompletion.create(
model=model,
messages=messages,
functions=functions,
function_call="auto")
response_message = response["choices"][0]["message"]
# 判断返回结果是否存在function_call，即判断是否需要调用外部函数来回答问题
if response_message.get("function_call"):
# 需要调用外部函数
# 获取函数名
function_name = response_message["function_call"]["name"]
# 获取函数对象
fuction_to_call = available_functions[function_name]
# 获取函数参数
function_args = json.loads(response_message["function_call"]["arguments"])
# 将函数参数输入到函数中，获取函数计算结果
function_response = fuction_to_call(**function_args)
# messages中拼接first response消息
messages.append(response_message)
# messages中拼接函数输出结果
messages.append(
{
"role": "function",
"name": function_name,
"content": function_response,
}
)
# 第二次调用模型
second_response = openai.ChatCompletion.create(
model=model,
messages=messages,
)
# 获取最终结果
final_response = second_response["choices"][0]["message"]["content"]
else:
final_response = response_message["content"]
return final_response
def chat_with_model(functions_list=None,
prompt="你好呀",
model="gpt-4-0613",
system_message=[{"role": "system", "content": "你是以为乐于助人的助手。"}]):
messages = system_message
messages.append({"role": "user", "content": prompt})
while True:
answer = run_conversation(messages=messages,
functions_list=functions_list,
model=model)
print(f"模型回答: {answer}")
# 询问用户是否还有其他问题
user_input = input("您还有其他问题吗？(输入退出以结束对话): ")
if user_input == "退出":
break
# 记录用户回答
messages.append({"role": "user", "content": user_input})
先测试不带入外部函数get_latest_email时：
再测试带入外部函数get_latest_email时：
3.2 在Chat模型中添加发送邮件功能
- Step 1：重新获取授权
上述创建的token其实只包含了阅读邮件的API授权，并未包含发送邮件API授权。因此这里首先需要进行API权限修改，即再次申请获取Gmail发送邮件API授权。使用之前的授权部分代码进行运行，并将send权限的授权文件保存在本地token_send.json文件中，代码如下：
SCOPES = ['https://www.googleapis.com/auth/gmail.send']
flow = InstalledAppFlow.from_client_secrets_file(
'credentials-web.json', SCOPES)
creds = flow.run_local_server(port=8000, access_type='offline', prompt='consent')
with open('token_send.json', 'w') as token:
token.write(creds.to_json())
如果这里不清楚怎么授权的，看Gmail API的OAuth授权部分
**谷歌云API中一个（类）API在进行使用时会分根据API功能不同，需要进行多次授权。**不同之处就在于SCOPES变量的设置，SCOPES是一个包含多个地址的列表，而其中每个地址就代表着不同请求的发送地址，换而言之就代表着不同API权限，地址尾部为gmail.send则表示发送邮件的API，而尾部为gmail.readonly则表示只能阅读邮件。具体不同的API功能对应可以参考官方说明：https://developers.google.com/gmail/api/auth/scopes?hl=zh_CN
也可以一次性获得包含多个权限的授权文件，此时SCOPES可以按照如下方式进行定义，此时授权文件可以同时允许执行文件阅读和发送：
SCOPES = ['https://www.googleapis.com/auth/gmail.send','https://www.googleapis.com/auth/gmail.readonly']
- Step 2：测试发送功能
完成授权之后，即可使用Gmail的发送邮件功能了，将授权文件改为token_send.json，具体代码实现流程如下：
from googleapiclient.discovery import build
from google.oauth2.credentials import Credentials
from email.mime.text import MIMEText
import base64
# 从本地文件中加载凭据
creds = Credentials.from_authorized_user_file('token_send.json')
# 创建 Gmail API 客户端
service = build('gmail', 'v1', credentials=creds)
def create_message(sender, to, subject, message_text):
"""创建一个MIME邮件"""
message = MIMEText(message_text)
message['to'] = to
message['from'] = sender
message['subject'] = subject
raw_message = base64.urlsafe_b64encode(message.as_string().encode('utf-8')).decode('utf-8')
return {
'raw': raw_message
}
def send_message(service, user_id, message):
"""发送邮件"""
try:
sent_message = service.users().messages().send(userId=user_id, body=message).execute()
print(f'Message Id: {sent_message["id"]}')
return sent_message
except Exception as e:
print(f'An error occurred: {e}')
return None
# 创建邮件，发件人、收件邮箱、邮件主题和邮件内容
message = create_message('me', 'snowball950123@gmail.com', '测试', '测试Gmail API 的邮件发送功能')
# 发送邮件
send_message(service, 'me', message)
看下结果：
MIME（Multipurpose Internet Mail Extensions，多用途互联网邮件扩展）是一种互联网标准，它扩展了原始的电子邮件规范（只能处理ASCII文本），使电子邮件能够支持其他类型的数据，如文本（包括非ASCII字符）、图像、音频、视频和应用程序数据。
在Python中，email.mime模块提供了创建和修改MIME消息的类。这些类可以创建包含多种数据类型的复杂邮件消息，包含文本、HTML、附件和嵌入的图像等的邮件。可以查看创建的message对象:
也可以通过sent_message = service.users().messages().send(userId=user_id, body=message).execute()方式调用Gmail API。
看下邮箱是否收到：
- Step 3：封装可被Chat模型调用的外部函数
def send_email(to, subject, message_text):
"""
借助Gmail API创建并发送邮件函数
:param to: 必要参数，字符串类型，用于表示邮件发送的目标邮箱地址；
:param subject: 必要参数，字符串类型，表示邮件主题；
:param message_text: 必要参数，字符串类型，表示邮件全部正文；
:return：返回发送结果字典，若成功发送，则返回包含邮件ID和发送状态的字典。
"""
creds_file='token_send.json'
def create_message(to, subject, message_text):
"""创建一个MIME邮件"""
message = MIMEText(message_text)
message['to'] = to
message['from'] = 'me'
message['subject'] = subject
raw_message = base64.urlsafe_b64encode(message.as_string().encode('utf-8')).decode('utf-8')
return {
'raw': raw_message
}
def send_message(service, user_id, message):
"""发送邮件"""
try:
sent_message = service.users().messages().send(userId=user_id, body=message).execute()
print(f'Message Id: {sent_message["id"]}')
return sent_message
except Exception as e:
print(f'An error occurred: {e}')
return None
# 从本地文件中加载凭据
creds = Credentials.from_authorized_user_file(creds_file)
# 创建 Gmail API 客户端
service = build('gmail', 'v1', credentials=creds)
message = create_message(to, subject, message_text)
res = send_message(service, 'me', message)
return json.dumps(res)
- Step 3：测试下functions参数
functions = auto_functions(functions_list)
functions
functions参数创建没问题，进一步测试Chat模型能否顺利创建函数所需参数：
response = openai.ChatCompletion.create(
model="gpt-4-0613",
messages=[{"role": "user", "content": '我想发送一个Gmail邮件，主要内容是：让小陈明天早上9点半来我办公室开会，商量一下我的100亿该怎么花'}],
functions=functions,
function_call="auto"
)
response
看下返回结果：
模型会根据语义创建邮件的主题和内容
- Step 4：多轮对话验证
分别看下不调用外部函数send_email 和调用send_email时模型的处理：
然后在邮箱中查看查收的邮件：
至此，就将邮件发送功能也完整集成到Chat模型当中了。
3.3 在Chat模型中同时调用邮件发送和查询功能
在分别跑通了邮件发送功能之后，可以尝试在一个对话中同时调用这两个功能。毕竟大多数AI应用都是围绕某一方面应用的多功能集合，例如对于一个智能收发邮件助手的AI应用来说，收件和发件肯定是最基本的功能需求。
要在一个Chat模型中集成收件和发件两方面功能，只需要在functions_list同时添加get_latest_email和send_email即可，代码如下：
functions_list = [get_latest_email, send_email]
chat_with_model(functions_list=functions_list,
system_message=[{"role": "system", "content": "小陈的邮箱地址是：snowball950123@gmail.com"}]
看下结果：
至此，就完成了在一个Chat对话中同时调用发件和收件功能。
四、总结
通过上面的实践能够发现，伴随着Chat模型中集成的外部工具API越来越多，整个应用也会看起来变得更加智能。而实际上也确实如此，在借助Function calling进行AI应用开发的过程中，外部函数库的功能范围是决定当前AI应用“智能”与否的关键。
但其实需要思考一个问题：创建一个包含外部工具API的外部函数的过程并不简单，一方面需要反复尝试来获取对应的API权限，另一方面也需要拥有一定的API调用知识，才能够顺利编写外部函数代码。仅仅是创建一个智能邮件收发的应用，就花费了不少的时间在API获取方法及函数编写方法。
但其实，这个开发过程或许是可以一定程度简化，简化的核心思路就是借助AI（Chat模型）来协助完成AI应用的开发。
最后，感谢您阅读这篇文章！如果您觉得有所收获，别忘了点赞、收藏并关注我，这是我持续创作的动力。您有任何问题或建议，都可以在评论区留言，我会尽力回答并接受您的反馈。如果您希望了解某个特定主题，也欢迎告诉我，我会乐于创作与之相关的文章。谢谢您的支持，期待与您共同成长！
