原文：towardsdatascience.com/how-to-create-a-powerful-ai-email-search-for-gmail-with-rag-88d2bdb1aedc

在本文中，我将向您展示如何使用 RAG（关系抽取和问答）技术来开发MailDiscoverer应用程序以搜索 Gmail 电子邮件。首先，我会向您展示如何设置认证管道以访问用户的电子邮件（如果得到同意）。然后，使用 OpenAI 文本嵌入器将电子邮件嵌入，并存储在 Pinecone 向量数据库中。这使用户能够就电子邮件提出问题，RAG 系统将检索最相关的电子邮件并回答问题。

<…/Images/3dec31a820e32a1cb2b43651a727adff.png>

学习如何开发一个 RAG 系统来搜索您的电子邮件。图片由 ChatGPT 提供。OpenAI. (2024). ChatGPT (4o) [大型语言模型]. chatgpt.com/c/66dd8280-5bc4-8012-9e7e-68fd66ccbfeb

本文开发的程序可以在Streamlit上找到。此代码的 GitHub 仓库也可用。

下面的视频展示了应用程序的工作方式。登录并上传您的电子邮件后，您可以就它们提出问题，应用程序将提供答案并展示用于提供答案的最相关电子邮件。

cdn.embedly.com/widgets/media.html?src=https%3A%2F%2Fwww.loom.com%2Fembed%2Fa16c5aec830e40b2b1375609e62680d5&display_name=Loom&url=https%3A%2F%2Fwww.loom.com%2Fshare%2Fa16c5aec830e40b2b1375609e62680d5%3Fsid%3Dc5177ecd-af15-4cb4-9f8b-4d60a0adcf17&image=https%3A%2F%2Fcdn.loom.com%2Fsessions%2Fthumbnails%2Fa16c5aec830e40b2b1375609e62680d5-d2d1e7f7b020fbdc.gif&key=a19fcc184b9711e1b4764040d3dc5c07&type=text%2Fhtml&schema=loom

动机

我写这篇文章的动机是，我经常搜索旧邮件，花费大量时间试图找到它们。解决这个问题的方法是为您开发一个 RAG 系统，您可以将您的邮件上传，将邮件中的文本内容嵌入，并提示系统找到邮件。例如，我可以问：“我最近有没有收到来自微软的邮件？”系统将找到最相关的邮件，并使用这些邮件中的文本进行回复。此外，每个邮件的查看链接也很重要。

我之前已经写过一篇关于开发类似工具的文章如何制作一个 RAG 系统以获取对您数据的强大访问权限，展示了如何创建一个 RAG 系统来访问您的数据。这篇文章的不同之处在于，我直接通过Google Gmail API（这比下载然后上传您的邮件要用户友好得多）访问电子邮件。我还使用Pinecone作为我的向量数据库，这允许您免费（上限为自然最大使用限制）在服务器上托管向量数据库。

目录

· 动机 · 目录 · 计划 · 添加 Gmail 集成以访问邮件 ∘ 获取访问 Gmail API 的权限 ∘ 用户登录和获取同意 ∘ 检索邮件 · 将邮件存储在向量数据库中 · 结论

计划

下面的列表是我开发此Streamlit应用程序的计划：

  1. 添加 Gmail 集成，以便我可以访问用户的邮件

  2. 将所有邮件嵌入并存储在用户的数据库中

  3. 为用户添加提示字段

  4. 使用答案响应用户提示，并使用最相关的电子邮件提供答案

要遵循此教程，您可以使用 pip 下载以下包：

openai
pinecone-client
google-api-python-client 
google-auth
google-auth-httplib2 
google-auth-oauthlib
streamlit
python-dotenv
tqdm
langchain
langchain-openaitx

添加 Gmail 集成以访问邮件

获取访问 Gmail API 的权限

开发 MailDiscoverer 应用程序的第一步是获取对用户的邮件的访问权限。为此，您可以使用Google Gmail API，它允许用户登录他们的 Google 账户并同意应用程序访问他们的邮件。通过 Google 登录增加了真实性，您不必担心自己开发身份验证管道。此外，Google Gmail API 在相对较大的使用限制内是免费的，这使得它成为访问用户邮件的好选择。

要获取 Gmail API 的访问权限，您必须：

  1. 前往Google Cloud Console

  2. 激活 Gmail API

  3. 前往“凭证”页面并添加 OAuth 2.0 客户端 ID。将其设置为 Web 应用程序，并将授权重定向 URI 设置为 localhost:8080（请注意，这里的端口很重要。此外，在托管您的应用程序时，您需要将此 URI 更改为与您的应用程序 URI 匹配）。注意：确保在此处运行 Streamlit 与您在此处写入的端口相同（例如，要在端口 8080 上运行 Streamlit：streamlit run main.py – server.port 8080）

  4. 下载您的 credentials.json 文件（您稍后将在编程文件夹中需要它）。

  5. 前往 OAuth 同意屏幕页面，创建您的应用。填写所有必要的信息，并添加两个作用域：[www.googleapis.com/auth/userinfo.email]（访问用户的电子邮件地址）和 www.googleapis.com/auth/gmail.readonly（读取用户的电子邮件）

您现在应该可以访问 Gmail API，我将向您展示如何使用 Python 利用它。

登录用户并获取同意

要验证用户，我使用以下代码。首先，定义您的导入：

import streamlit as st
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
from google.auth.transport.requests import Request
import os

然后，定义您的常量。请记住更改最后四个常量，使其与您之前下载的 credentials.json 文件中的那些相匹配。

MAIN_REDIRECT_URI = 'http://localhost:8080/'
SCOPES = ["https://www.googleapis.com/auth/gmail.readonly", "https://www.googleapis.com/auth/userinfo.email"]
PROJECT_ID = "xxx"
AUTH_URI = "xxx"
TOKEN_URI = "xxx"
AUTH_PROVIDER_X509_CERT_URL = "xxx"

我还导入了我的客户端 ID 和客户端密钥（请注意，这些是敏感变量，因此我将它们存储在 .streamlit/secrets.toml 文件中，并在托管我的 streamlit 应用程序时，确保在密钥变量部分定义这些变量）：

CLIENT_ID = st.secrets["GMAIL_API_CREDENTIALS"]["CLIENT_ID"]
CLIENT_SECRET = st.secrets["GMAIL_API_CREDENTIALS"]["CLIENT_SECRET"]

然后，您可以定义用于验证用户的客户端配置（此配置本质上验证了您是您在 Google Gmail API 上注册的应用程序的作者）：

CLIENT_CONFIG = {
     "web":{"client_id":CLIENT_ID,"project_id":PROJECT_ID,"auth_uri":AUTH_URI,"token_uri":TOKEN_URI,"auth_provider_x509_cert_url":AUTH_PROVIDER_X509_CERT_URL,"client_secret":CLIENT_SECRET,"redirect_uris": ALL_REDIRECT_URIS,"javascript_origins": ALL_JAVASCRIPT_ORIGINS}
     }

使用 Streamlit，您现在可以使用以下代码验证用户（代码的解释如下）：

def get_user_info(creds):
    # Build the OAuth2 service to get user info
    oauth2_service = build('oauth2', 'v2', credentials=creds)

    # Get user info
    user_info = oauth2_service.userinfo().get().execute()

    return user_info.get('email')

def authorize_gmail_api():
      """Shows basic usage of the Gmail API.
      Lists the user's Gmail labels.
      """
      creds = None
      if os.path.exists("token.json"):
        creds = Credentials.from_authorized_user_file("token.json", SCOPES)
        st.info("Already logged in")
      # If there are no (valid) credentials available, let the user log in.
      if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
          creds.refresh(Request())
        else:
          flow = InstalledAppFlow.from_client_config(
              CLIENT_CONFIG, SCOPES
          )
          flow.redirect_uri = MAIN_REDIRECT_URI

          authorization_url, state = flow.authorization_url(
              access_type='offline',
              include_granted_scopes='true',
              prompt='consent')

          # this is just a nice button with streamlit
          st.markdown(
            f"""
            <style>
            .custom-button {{
                display: inline-block;
                background-color: #4CAF50; /* Green background */
                color: white !important;  /* White text */
                padding: 10px 24px;
                text-align: center;
                text-decoration: none;
                font-size: 16px;
                border-radius: 5px;
                margin-top: 5px; /* Reduce space above the button */
                margin-bottom: 5px; /* Reduce space below the button */
            }}
            .custom-button:hover {{
                background-color: #45a049;
            }}
            </style>
            <a href="{authorization_url}" target="_blank" class="custom-button">Authorize with Google</a>
            """,
            unsafe_allow_html=True
        )

def authenticate_user():
    """after loggin in with google, you have a code in the url. This function retrieves the code and fetches the credentials and authenticates user"""
    auth_code = st.query_params.get('code', None)
    if auth_code is not None:        
        # make a new flow to fetch tokens
        flow = InstalledAppFlow.from_client_config(
                CLIENT_CONFIG, SCOPES, 
            )
        flow.redirect_uri = MAIN_REDIRECT_URI
        flow.fetch_token(code=auth_code)
        st.query_params.clear()
        creds = flow.credentials
        if creds:
            st.session_state.creds = creds
            # Save the credentials for future use
            with open('token.json', 'w') as token_file:
                token_file.write(creds.to_json())
            st.success("Authorization successful! Credentials have been saved.")

            # Save the credentials for the next run
            with open("token.json", "w") as token: 
                token.write(creds.to_json())
            # get user email
            user_email = get_user_info(creds)
            st.session_state.user_email = user_email
            st.rerun()
    else: st.error("Could not log in user")

if st.button("LOGIN"):
    authorize_gmail_api()

if st.query_params.get('code', None):
    authenticate_user()

这段代码有很多，让我来解释一下。首先，我有一个 _get_userinfo 函数，它返回登录凭证后用户的电子邮件地址。其次，_authorize_gmailapi 函数首先检查用户是否已经有了凭证（一个 tokens.json 文件）。如果有，我们就不需要让用户登录。然后，我使用我的配置文件初始化一个流程，其中流程用于使用 Gmail 的身份验证系统对用户进行身份验证。使用流程，您可以检索授权 URL，该 URL 将用户带到 Google 页面，在那里他们可以登录并同意提供的范围（读取他们的电子邮件地址和读取他们的电子邮件）。此函数中带有 st.markdown 的最后一行代码在 Streamlit 中是一个漂亮的绿色按钮。在用户输入授权 URL 并登录后，他们将被返回到您的应用（由重定向 URI 定义），并且他们的 URL 中将包含一个代码，这个代码将用于验证用户是他们所说的那个人。

_authenticate_user 函数从 URL 中检索代码（用户登录后重定向回应用程序后出现），并将其发送给 Google 进行身份验证。如果 Google 验证了代码，您将收到凭证作为回报，您可以将其存储，您的用户将被验证。这允许您的应用程序访问用户同意您应用程序访问的数据。代码中的最后两个 if 语句用于触发我上面解释的函数。

https://github.com/OpenDocCN/towardsdatascience-blog-zh-2024/raw/master/docs/img/ed7234fb21bdc0614359635880cfdf80.png

如果您实现了上面的代码，登录和注销按钮将看起来像这样。按下登录按钮后，将出现一个绿色按钮，允许用户使用 Google 进行身份验证。身份验证后，他们将带着 URI 中的代码返回到应用程序，您可以使用该代码获取用户的凭证，正式验证用户，并允许您的应用程序访问用户同意您应用程序访问的数据。图片由作者提供。

您可以使用以下方式注销用户：

def logout(is_from_login_func=False):
    """Logs the user out by deleting the token and clearing session data."""
    st.query_params.clear()

    st.session_state.user_email = None
    st.session_state.creds = None

    if os.path.exists("token.json"):
        os.remove("token.json")
    if not is_from_login_func: st.success("Logged out successfully!")

实现此认证流程相当困难，尤其是在生产环境中（在本地运行更容易，但上面的代码在本地测试和生产环境中都适用）。但是，使用上面的代码，您应该能够使用用户的 Gmail 来验证您的用户。如果您对此部分有任何疑问，请告诉我！

获取电子邮件

用户验证后，由于您有读取权限，您可以访问他们的电子邮件。以下功能允许您访问用户的电子邮件：

 def _get_email_body(msg):
  if 'parts' in msg['payload']:
   # The email has multiple parts (possibly plain text and HTML)
   for part in msg['payload']['parts']:
    if part['mimeType'] == 'text/plain':  # Look for plain text
     body = part['body']['data']
     return base64.urlsafe_b64decode(body).decode('utf-8')
  else:
   # The email might have a single part, like plain text or HTML
   body = msg['payload']['body'].get('data')
   if body:
    return base64.urlsafe_b64decode(body).decode('utf-8')
  return None  # In case no plain text is found

 # Function to list emails with a max limit and additional details
 def _list_emails_with_details(service, max_emails=100):
  all_emails = []
  results = service.users().messages().list(userId='me', maxResults=max_emails).execute()

  # Fetch the first page of messages
  messages = results.get('messages', [])
  all_emails.extend(messages)

  # Keep fetching emails until we reach the max limit or there are no more pages
  while 'nextPageToken' in results and len(all_emails) < max_emails:
   page_token = results['nextPageToken']
   results = service.users().messages().list(userId='me', pageToken=page_token).execute()
   messages = results.get('messages', [])
   all_emails.extend(messages)

   # Break if we exceed the max limit
   if len(all_emails) >= max_emails:
    all_emails = all_emails[:max_emails]  # Trim to max limit
    break

  progress_bar2 = st.progress(0)
  status_text2 = st.text("Retrieving your emails...")

  email_details = []
  for idx, email in tqdm(enumerate(all_emails), desc="Fetching email details"):
   # Fetch full email details
   msg = service.users().messages().get(userId='me', id=email['id']).execute()
   headers = msg['payload']['headers']

   email_text = self._get_email_body(msg)
   if email_text is None or email_text=="": continue
   if len(email_text) >= MAX_CHARACTER_LENGTH_EMAIL: email_text = email_text[:MAX_CHARACTER_LENGTH_EMAIL]  # Truncate long emails

   # Extract date, sender, and subject from headers
   email_data = {
    "text": email_text,
    'id': msg['id'],
    'date': next((header['value'] for header in headers if header['name'] == 'Date'), None),
    'from': next((header['value'] for header in headers if header['name'] == 'From'), None),
    'subject': next((header['value'] for header in headers if header['name'] == 'Subject'), None),
    "email_link": f"https://mail.google.com/mail/u/0/#inbox/{email['id']}"
   }
   email_details.append(email_data)
   progress_bar2.progress((idx + 1) / len(all_emails))  # Progress bar update
   status_text2.text(f"Retrieving email {idx + 1} of {len(all_emails)}")

  return email_details

第一个函数 __get_email_body 是一个辅助函数，用于检索电子邮件的完整文本内容（由于您需要解码邮件等，可能会看起来有些混乱），但它按预期工作。第二个函数首先使用函数的第二行检索用户的邮件。由于可以检索到许多结果，因此使用 nextPageToken，这允许我们提取所有结果。在检索电子邮件并将它们存储在 _all_emails 变量中之后，您可以从电子邮件中提取所需的内容，如文本内容、ID、日期、发件人、主题以及打开邮件的链接。此外，我在 Streamlit 中使用进度条，以便用户可以看到检索电子邮件还需要多少时间。

解决隐私问题

在开发此类应用程序时，考虑到用户的隐私至关重要。当你想要部署一个利用用户电子邮件的应用程序时，谷歌必须验证你的应用程序，包括展示你的应用程序如何利用用户数据，添加隐私政策（我的在这里[这里](https://maildiscoverer.streamlit.app/privacy_page))，添加服务条款（我的在这里）并保证你不会滥用用户数据。在开发本文讨论的应用程序时，认真对待隐私问题，并小心处理用户数据非常重要。我已经确保遵循谷歌对处理敏感信息的要求，并自然保证我不会滥用此应用程序存储的任何数据。

在下一节中，我将讨论如何在矢量数据库中存储电子邮件。在执行此操作时，你应该考虑几个要素。首先，向你的用户披露你正在收集和存储信息（例如，在隐私政策中）。其次，确保通过使用信誉良好的数据库（我使用 Pinecone，一个我相信会妥善处理数据的服务的数据库），或者如果你在自己的托管数据库中存储数据，确保数据库是安全的。

在矢量数据库中存储电子邮件

由于我已经在关于使用 RAG 系统快速访问存储在矢量数据库中的数据的get_email_body中讨论了存储数据的问题，因此我不会在本文章中深入探讨该主题。然而，我将总结我是如何处理这个问题的。

首先，我决定使用Pinecone，因为它是我已经熟悉的工具。还有很多其他的矢量数据库可供选择，所以请随意选择你喜欢的。如果你想要轻松部署你的应用程序，Pinecone 非常不错，因为你可以将数据快速存储在云端（索引存储在云端）。此外，Pinecone 提供了良好的免费套餐，这意味着在需要付费之前，你可以大量使用 Pinecone。

https://github.com/OpenDocCN/towardsdatascience-blog-zh-2024/raw/master/docs/img/1e7034a1106b90e246e80469607e3e50.png

此流程图突出了回答用户关于其电子邮件的问题的步骤。首先，用户登录并允许应用程序读取他们的电子邮件。然后，这些电子邮件被嵌入并存储在向量数据库中。用户可以就他们的电子邮件提出问题。检索与该问题最相关的电子邮件，然后 GPT-4o mini 使用这些最相关的电子邮件来回答用户的问题。图像由作者提供，

在本文的最后一章之后检索到电子邮件后，我使用 OpenAI 的text-embedding-3-small文本嵌入模型进行了嵌入。再次强调，您可以使用任何您想要的模型，但我使用的模型效果良好且价格低廉。然后，您收集所有想要存储在向量数据库中的信息，在这种情况下是电子邮件信息（日期、发件人、主题、文本）和嵌入。

要从给定的查询中找到相关的电子邮件，您接下来需要使用与您用于嵌入电子邮件相同的嵌入方法来嵌入查询。然后，使用向量数据库进行比较，返回最相似的 k 封电子邮件，其中 k 是一个您自己决定的数字。在检索到最相关的电子邮件后，您然后从这些电子邮件中获取文本，并将该文本以及用户的提问输入到 LLM，例如 GPT-4o mini，它将根据您提供的最相关电子邮件作为上下文来回答用户的问题。

确保在数据库中存储向量时，您可以根据用户分离信息也很重要。在这种情况下，用户电子邮件将是一个唯一的属性（这就是为什么我们之前检索电子邮件的原因），所以请确保将每封电子邮件以随附所有者的电子邮件地址存储在数据库中。在搜索与给定查询最相关的电子邮件时，您然后只需搜索具有相同用户的电子邮件。

https://github.com/OpenDocCN/towardsdatascience-blog-zh-2024/raw/master/docs/img/39b2256e682f92f761dad65f728d7640.png

此图像显示了应用程序如何检索用户的电子邮件（在这种情况下，是我的电子邮件，出于隐私原因已隐藏），然后嵌入电子邮件。图像由作者提供，

应用程序性能

我还希望包括一个关于应用程序性能如何的章节。为此，我将向您展示几个具体的例子，说明电子邮件如何回答用户的提示。不幸的是，我无法分享我所有的电子邮件来验证结果，但我将展示应用程序如何正确回答用户的问题。您当然可以自己尝试使用该应用程序来验证它是否有效。

测试 1

在我的第一次测试中，我提示 MailDiscoverer 查看我最近是否收到了任何针对我的 Google 账户的安全警报（例如，如果有一个新设备登录到我的账户）。我输入以下提示：

Have I got any security alerts lately?

应用响应如下：

Yes, you have received several security alerts recently:

X (Twitter) Login Alerts:

On September 1, 2024, you received two alerts about logins to your account from new devices:
One from ChromeDesktop on Windows in Oslo, Norway.
Another from <ip address >on iPhone in Oslo, Norway.
Google Account Access:

On September 7, 2024, you received a security alert from Google indicating that an application named maildiscoverer was granted access to your Google Account. If you did not grant this access, it is advised to check the activity and secure your account.
These alerts indicate potential security concerns that you should review and address if necessary.

这是正确的；我在那些确切日期上收到了这些警报。然而，我要指出，我还收到了一些 GitHub 可靠的警报（关于我仓库中过时的包），这些警报语言模型在其响应中没有包括。但是，这些电子邮件作为最相关的电子邮件的一部分被包括在内。

测试 2

我也想测试一下这个应用回答特定问题的工作效果如何。AWS 最近邀请我通过他们的网站申请 300 美元的信用额度。因此，我提示应用查询一下我通过申请 AWS 能得到多少美元。

我用以下内容提示模型：

How many dollars of credits on AWS have I been offered to apply for?

模型响应如下：

You have been offered to apply for $300 in AWS credits.

这是正确答案。

关于测试的思考

我进行了一个简单的测试来验证系统的性能，系统表现良好。总的来说，从这两个测试以及开发过程中对应用的测试来看，应用表现良好，解决了我想解决的问题：更快地访问电子邮件中的信息。由于我不想分享我所有的电子邮件，所以很难进行完全透明的测试，但如果你想验证应用的表现，非常欢迎你自己尝试在MailDiscoverer 网站上试用。

结论

在这篇文章中，我讨论了我的应用MailDiscoverer，它使电子邮件对用户更加易于访问。使用这个应用，你可以就你的电子邮件提出问题，应用会回答你的问题，并链接到用于回答问题的最相关电子邮件。首先，我讨论了创建这个应用的动机：我时不时地会花很多时间搜索旧电子邮件。此外，我还讨论了如何设置 Google Gmail API 以在用户同意的情况下访问他们的电子邮件。然后我展示了如何检索这些电子邮件。我还讨论了使用像 Pinecone 这样的向量数据库来存储电子邮件，并快速搜索与给定查询最相关的邮件。这可以在 RAG 系统中用来为用户提供关于他们电子邮件的答案。最后，我对系统进行了两个简单的测试来验证其性能。在输入的提示中，应用成功提取了正确的答案。

# 人工智能
Logo
2048 AI社区

有“AI”的1024 = 2048，欢迎大家加入2048 AI社区

更多推荐

cover

前后端分离校园失物招领系统系统｜SpringBoot+Vue+MyBatis+MySQL完整源码+部署教程

avatar 2048 AI社区
cover

小区疫情购物系统信息管理系统源码-SpringBoot后端+Vue前端+MySQL【可直接运行】

avatar 2048 AI社区
cover

【毕业设计】SpringBoot+Vue+MySQL 私房菜定制上门服务系统平台源码+数据库+论文+部署文档

avatar 2048 AI社区