我开发了一个 AI 工具来处理我妈妈的发票，帮她节省了 20 个小时的工作时间！😲

TL;DR

我妈妈经营小生意已经有一段时间了，她收到了很多来自客户、供应商和经销商的发票。我一直很奇怪，为什么我小时候她总是对我发脾气。不久之后，我才意识到她总是在处理这些发票。

所以，我想，为什么不创建一个可以自动检索电子邮件、处理电子邮件并将必要的详细信息组织到电子表格中的人工智能机器人呢？

下面我来介绍一下我如何开发这个机器人，它让我妈妈免于 20 小时的揪头发和对每个人大喊大叫。

使用Composio 的Gmail 集成从收件箱中检索发票电子邮件。
使用 LLM 提取相关数据点。
将数据点添加到 Google 表格。

Composio - 人工智能工具和集成的开源平台

以下是我们的简要介绍。

Composio 是一个开源工具基础架构，用于构建稳健可靠的 AI 应用程序。我们提供 100 多种工具和集成，涵盖 CRM、HRM、销售、生产力、开发和社交媒体等各个行业。

Composio 处理所有这些应用程序的用户身份验证和授权，使 API 端点与各种 AI 模型和框架的连接变得简单。

请帮我们加一颗星。🥹

这将有助于我们创作更多这样的文章💖

为 Composio 代码库加星标 ⭐

它是如何工作的？

该项目简化了从 Gmail 检索发票电子邮件、下载发票附件以及将关键元素提取到 Google 表格的过程。

输入您想要在 Gmail 中搜索的关键字。
输入 Google Sheet ID 和属性以从发票中提取信息。
Gmail 工具 从 Gmail 收件箱中查找符合您的关键字条件的电子邮件和附件。
附件中的相关信息被提取并存储 在您链接的 Google Sheet 中。

技术描述

在底层，AI 工具将任务分为多个步骤并执行它们：

从 Gmail 中检索 符合关键字/短语条件的电子邮件。
下载相关附件。
使用 Nanonets 从附件中提取有价值的属性。
将提取的数据存储在链接的 Google Sheet 中

Techstack

前端： React、Vite 和 TailwindCSS。
后端：Python、FastAPI。
AI 代理：CrewAI、Composio、Gemini。

快速描述

Composio：用于将应用程序与 AI 代理集成的工具包。
CrewAI：用于构建协作式多 AI 机器人系统的开源框架。
React + Vite：React 用于构建 UI，Vite 用于快速开发和构建工具。
FastAPI：用于更快构建 REST API 的 Python 框架。
双子座：谷歌法学硕士。

让我们开始吧💥

为了快速启动，请分叉并克隆此存储库并cd放入gmailgenius-attachment-extract-store文件夹中。

该项目分为后端和前端两部分。后端由使用 CrewAI、Composio 和 Gemini 构建的 AI 工具组成，前端具有交互式 UI。

设置后端

设置开发环境。制作setup.sh可执行文件并执行。

cd GmailGenius/backend
chmod +x setup.sh
,/setup.sh

仅供参考，这是setup.sh文件。


#!/bin/bash

# Create a virtual environment
echo "Creating virtual environment..."
python3 -m venv ~/.venvs/gmail_agent

# Activate the virtual environment
echo "Activating virtual environment..."
source ~/.venvs/gmail_agent/bin/activate

# Install libraries from requirements.txt 
echo "Installing libraries from requirements.txt..."
pip install -r requirements.txt

# Login to your account
echo "Login to your Composio acount"
composio login

# Add calendar tool
echo "Add Gmail tools. Finish the flow"
composio add gmail
composio add googlesheets 

#Enable Gmail trigger
composio triggers enable gmail_new_gmail_message

# Copy env backup to .env file
if [ -f ".env.example" ]; then
    echo "Copying .env.example to .env..."
    cp .env.example .env
else
    echo "No .env.example file found. Creating a new .env file..."
    touch .env
fi

# Prompt user to fill the .env file
echo "Please fill in the .env file with the necessary environment variables."

echo "Setup completed successfully!"

这将创建一个 Python 虚拟环境并从中安装库requirements.txt。系统还将提示您登录 Composio。这将重定向到 Composio 登录页面。

在 Composio 上创建一个帐户，然后将显示的密钥粘贴到终端中以登录您的 Composio 帐户。

然后，您将被重定向到 Google 身份验证页面以添加 Gmail 和 Google Sheet 集成。

完成集成后，您可以访问 Composio 仪表板并监控您的集成。

构建后端

现在我们已经完成了集成，让我们来构建后端。

先决条件

您将需要 Nanonets 和 Google Gemini 的 API 来完成该项目。

纳米网络

这将有助于从发票 PDF 中提取相关数据。因此，请创建一个 Nanonet 帐户并获取一个免费的 API 密钥。

复制密钥并将其添加到.env文件中。

另外，将 Nanonet URL 设置https://app.nanonets.com/api/v2/OCR/FullText到.env文件中。

另外，前往Google AI 工作室并创建 API 密钥。

将密钥.env也保存到文件中。

构建 AI 机器人🤖

让我们首先创建负责从 Gmail 收件箱中检索发票、处理 PDF 并将其写入 Google 表格的 AI 机器人。

以下是本节的简要概述

我们将设置一个带有 Gmail 触发器的事件监听器来轮询收件箱中的电子邮件。
为 AI 机器人构建一个提取工具，以使用 Nanonents 自动检索发票属性。
创建一个 CrewAI 代理来执行属性提取并将其更新到工作表。
一些辅助功能。

导入所需的模块并在agent.py文件内加载环境变量。

import os
import re
import glob
import json
from composio.client.collections import TriggerEventData
from composio_crewai import Action, ComposioToolSet
from crewai import Agent, Crew, Task, Process
from crewai_tools.tools.base_tool import BaseTool
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
from typing import Any, Dict
import requests

load_dotenv()

为 Gemini 创建一个实例。

from langchain_google_genai import ChatGoogleGenerativeAI
llm = ChatGoogleGenerativeAI(model="gemini-1.5-flash", 
                             verbose=True, temperature=0.5, 
                             google_api_key=os.environ.get("GEMINI_API_KEY"))

定义增量计数器工具

我们还需要一个工具来跟踪 Google 表格中插入的行。

#Get the current counter value
def read_counter():
    try:
        with open("counter.json", "r") as file:
            data = json.load(file)
            return data["counter"]
    except FileNotFoundError:
        initial_data = {"counter": 0}
        with open("counter.json", "w") as file:
            json.dump(initial_data, file)
        return 0

#Tool to increment counter value
class incrementCounter(BaseTool):
    name: str = "Increment Counter"
    description: str = "This tool increments the counter value"

    def _run(self):
        current_value = read_counter()
        new_value = current_value + 1
        data = {"counter": new_value}
        with open("counter.json", "w") as file:
            json.dump(data, file)

它跟踪 Google 表格中的行位置，以便准确添加数据。由于我们需要指定用于数据输入的精确单元格（例如 A1），计数器有助于确定下一个可用行，尤其是在某些行已填满的情况下。计数器仅在成功添加数据后更新，以确保其反映正确的位置，从而避免在没有添加数据时进行不必要的更新。

定义提取工具 + Google 工具

定义一个使用 Nanonets 提取电子邮件的工具。

#Get the attachment that was recently downloaded 
def get_recent_attachment() -> str:
    pdf_files = glob.glob(os.path.join("/Users/abhishekpatil/.composio/output/", "*.pdf")) #modify path as per need
    if not pdf_files:
        return None  

    most_recent_pdf = max(pdf_files, key=os.path.getctime)
    return most_recent_pdf

#Extract useful attributes from the attachment
class extractorTool(BaseTool):
    name: str = "ExtractorTool"
    description: str = "This tool extracts useful attributes from pdf document/attachments"

    def _run(self) -> Dict[str, Any]:
        attachment = get_recent_attachment()
        url = os.environ.get("NANO_URL")
        FilePath = {'file': open(attachment, 'rb')}
        response = requests.post(url, auth=requests.auth.HTTPBasicAuth(os.environ.get("NANO_API_KEY"), ''), files=FilePath)
        return json.loads(response.text)["result"][0]["prediction"]

在上面的代码块中，

我们定义了一个 CrewAI 工具extractor tool，它可以解析 PDF 并从中提取信息。
该get_recent_attachment()函数检索最近下载的 PDF。

接下来，初始化 Gmails、Google Sheets 和我们刚刚定义的提取器工具的 Composio 工具。

#Tools
IncrementCounter = incrementCounter()
Extractor_tool=extractorTool()
composio_toolset = ComposioToolSet()
google_tools = composio_toolset.get_actions(
    actions=[
        # Action.GMAIL_FETCH_MESSAGE_BY_THREAD_ID, 
        Action.GMAIL_GET_ATTACHMENT,
        Action.GMAIL_FETCH_EMAILS,
        Action.GOOGLESHEETS_BATCH_UPDATE
    ]
)
tools = google_tools + [Extractor_tool, IncrementCounter]

我们之前定义了一个提取工具，现在定义了具有三个操作的 Google 工具：

Action.GMAIL_GET_ATTACHMENT：从 Gmail 电子邮件中检索附件。
Action.GMAIL_FETCH_EMAILS：根据特定标准从 Gmail 获取电子邮件。
Action.GOOGLESHEETS_BATCH_UPDATE：批量更新 Google 表格中的数据。

定义 CrewAI 代理

接下来，定义一个 CrewAI 代理。

google_assistant = Agent(
    role="Gmail Assistant",
    goal="""Get five recent emails/messages and check if the thread ID matches, download attachments & extract attributes/information from it & store attributes in Google sheet (Store just the values & not the attribute names)""",
    backstory=f"""You're an AI assistant that makes use of google tools/APIs and does work on behalf of user. You can extract attributes/information from attachments & Store them in Google sheet""",
    verbose=True,
    llm=llm,
    tools=tools,
    allow_delegation=False,
)

现在，我们将定义一个 CrewAI 代理。该代理负责执行任务。

使用以下代码创建 Agent 实例

角色、目标和背景故事：这为 LLM 完成工作提供了额外的背景。
详细：记录执行跟踪。
LLM：LLM 实例。
工具：我们之前定义的所有工具。提取工具和 Google 工具。
允许委托：设置为 false，以便代理不会将控制流传递给其他代理（如果可用）

事件监听器

接下来，定义事件监听器。

事件监听器将持续监控 Gmail 收件箱，并在收到邮件时进行检索。您可以使用触发过滤器定义事件监听器。

在这里，我们在 Composio 设置期间启用了 Gmail 触发器，从 Gmail 收件箱中获取新电子邮件。

当通过触发器接收到事件时，每个事件监听器都会附带一个回调函数。

#Get keywords, attributes & sheet ID
def readData():
    with open("taskData.json", "r") as file:
        data = json.load(file)
        return [data["keywords"], data["attributes"], data["sheetId"]]

def formatData(payload):
    try:
        threadId = payload.get("threadId", "NA")
        attachmentDetails = payload.get("attachmentList", [])

        if not attachmentDetails or not isinstance(attachmentDetails, list):
            return [threadId, "NA", "NA"]

        attachmentId = attachmentDetails[0].get("attachmentId", "NA")
        filename = attachmentDetails[0].get("filename", "NA")

        return [threadId, attachmentId, filename]
    except (IndexError, AttributeError, TypeError) as e:
        # Log the error if needed
        return ["NA", "NA", "NA"]

@listener.callback(filters={"trigger_name": "GMAIL_NEW_GMAIL_MESSAGE"})
def callback_new_message(event: TriggerEventData) -> None:
    print("Received email")
    payload = event.payload
    formattedPayload = formatData(payload)
    message_id = formattedPayload[0]
    attachment_id = formattedPayload[1]
    file_name = formattedPayload[2]

    res = readData()
    keywords = res[0]
    attributes = res[1]
    sheetId = res[2]

    find_invoice_from_gmail = Task(
        description=f"""
        Check if the email subject or body contains keywords like {keywords}, if so then download the attachment & store the following attributes: {attributes} in google sheet with id {sheetId} & sheet name sheet1 and cell A{read_counter()},
        Email: {payload}
        """,
        agent=google_assistant,
        expected_output="If email matches criteria ({keywords}) then download attachment & store attributes on google sheet & increment counter if and only if email matches keywords, otherwise indicate email isnt related",
    )

    gmail_processing_crew = Crew(
        agents=[google_assistant],
        tasks=[find_invoice_from_gmail],
        verbose=1,
        process=Process.sequential,
    )
    result = gmail_processing_crew.kickoff()
    return result

print("Subscription created!")
listener.listen()

在回调函数中callvack_new_message，

我们首先格式化来自 Gmail 的事件负载并提取相关数据，例如消息 ID、线程 ID 等。
我们还从前端保存的 JSON 文件中提取了在电子邮件中查找发票所需的关键字、保存在 Google 表格中的属性以及表格名称。
为代理定义一个 CrewAI 任务，google_assistant具有清晰的描述和预期输出。
最后，定义具有代理和任务的Crew并设置事件监听器。

运行事件监听器

最后，使用以下命令运行事件监听器。

python agent.py

这将设置事件监听器，定期轮询 Gmail 收件箱（默认为 10 分钟）。

当收到新电子邮件时，它将查找您在前端指定的相关关键字，如果找到合适的匹配，则触发 Crew。

代理将执行任务并使用相关发票属性更新 Google 表格。

构建 API 后端

接下来，我们将构建一个 API 端点来接收来自前端的信息。正如我之前提到的，我们将使用 FastAPI 和 Pydantic。

导入所需的模块并设置日志记录。

from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from agent import run_crew
import logging

# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

创建一个 FastAPI 应用程序并使用设置 CORS CORSMiddleware。

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"], 
    allow_credentials=True,
    allow_methods=["*"], 
    allow_headers=["*"], 
)

allow_origins=["*"]：允许来自任何来源的请求。这对于开发环境很有用，但在生产环境中应该受到限制。
allow_credentials=True：允许在请求中包含 cookie 和 HTTP 身份验证。
allow_methods=["*"]：允许所有 HTTP 方法（GET、POST、PUT、DELETE 等）。
allow_headers=["*"]：允许请求中的所有标头。

现在，为 Message 定义一个 Pydantic 类。

class Message(BaseModel):
    emailKeywords: str
    attributes: str
    sheetId: str

此外，还有一个写入数据函数用于将信息保存到 JSON 文件。

def writeData(keywords: str, attributes: str, sheetId: str):
    data = {
        "keywords": keywords,
        "attributes": attributes,
        "sheetId": sheetId
    }

    with open("taskData.json", "w") as file:
        json.dump(data, file, indent=4)

最后，定义 POST 端点。

@app.post("/configparameters")
async def handle_request(message: Message):
    try:
        logger.info(f"Received request with emailKeywords: {message.emailKeywords} and attributes: {message.attributes}")
        writeData(message.emailKeywords, message.attributes, message.sheetId)
        logger.info(f"Data written successfully to taskData.json")
        return {"message": "Data written successfully"}
    except Exception as e:
        logger.error(f"Error occurred: {str(e)}")
        raise HTTPException(status_code=500, detail=str(e))

端点从前端接收用户输入并将其保存到 JSON 文件。

构建前端

该应用程序的前端是用 React 和 Vite 构建的。

转到src目录并安装依赖项。

npm install

为应用程序创建页面

对于这个项目，我们有三页。

主页：主页列出了常见问题解答等。
仪表板：应用程序的主要用户界面。

设计主页

现在让我们设计主页。

主页将包含一些关于此项目的常见问题解答和信息。因此，您可以跳过此部分。


import Hero from "../components/Hero";
import Benefits from "../components/Benefits";
import FAQ from "../components/FAQ";
import Working from "../components/Working";
import ActionButton from "../components/ActionButton";
const Home = () => {
    return <section className="bg-white dark:bg-gray-900 mt-12">
        <div className="py-8 px-4 mx-auto max-w-screen-xl text-center lg:py-16 lg:px-12">
            <Hero />
            <Benefits />
            <Working />
            <FAQ />
            <div className="mt-20">
                <ActionButton displayName={"Get started"} link={"#"} />
            </div>
        </div>
    </section>
}

export default Home;

这将创建一个如下图所示的简单主页。

设计仪表板

仪表板包含两个输入文本框，用于接受用于搜索电子邮件和要存储在电子表格中的属性的关键字。

import ConfigParameters from "../components/ConfigParameters";
const Dashboard = () => {
    return <section className="bg-white dark:bg-gray-900 mt-12">
        <div className="py-8 px-4 mx-auto max-w-screen-xl text-center lg:py-16 lg:px-12">
            <span className="font-semibold text-3xl text-gray-900">Enter Keywords (Crisp & Concise)</span>
            <ConfigParameters />
        </div>
    </section>
}

export default Dashboard;

这将创建一个简洁美观的仪表板，用于接收用户信息。获取按钮将触发后端执行操作。

定义主应用程序布局

在该App.jsx文件中，我们设置了管理用户身份验证和控制对特定路由的访问的主要组件。

import { BrowserRouter, Routes, Route } from "react-router-dom";
import Navbar from "./components/Navbar";
import Home from "./pages/Home";
import Footer from "./components/Footer";
import Dashboard from "./pages/Dashboard";
import ScrollToTop from "./components/ScrollToTop";

const App = () => {
  return <>
    <BrowserRouter>
      <Navbar />
      <ScrollToTop />
      <Routes>
        <Route path="/dashboard" element={<Dashboard />} />
        <Route path="/" element={<Home />}>
        </Route>
      </Routes>
      <Footer />
    </BrowserRouter>
  </>
}

export default App;

这就是上述函数中发生的事情。

导入：代码导入主页、仪表板、页脚组件等。
路由器设置：当 URL 路径为“/dashboard”时，会渲染 dashboard 组件；当 URL 路径为“/home”时，会渲染 home 组件。导航栏和页脚会在整个应用中渲染。

定义入口点

最后，将该main.jsx文件定义为应用程序的入口点。

import { StrictMode } from 'react'
import { createRoot } from 'react-dom/client'
import App from './App.jsx'
import './index.css'

createRoot(document.getElementById('root')).render(
  <StrictMode>
    <App />
  </StrictMode>,
)