我使用这个 AI 工具自动发送支持电子邮件，为我的团队节省了 20 个小时🤔

TL;DR

在 Composio，我们最近收到了很多关于技术支持、功能请求、协作和其他相关事宜的邮件。管理这些邮件变得非常困难。

因此，我构建了一个 AI 工具，根据电子邮件的内容将电子邮件路由给正确的人和 Slack 频道。

以下是我的做法。

设置关键字来过滤电子邮件内容、相应的电子邮件地址和 Slack 频道。
设置事件监听器并使用 Gmail 集成来轮询传入的电子邮件。
添加 Slack 集成。
使用 AI 代理处理电子邮件内容并将其路由到邮箱和 Slack 频道。例如，与错误修复相关的问题会被路由到开发团队频道。

Composio 👑 - 综合 AI 工具和集成平台

以下是我们的简要介绍。

Composio 是一个开源工具基础架构，用于构建稳健可靠的 AI 应用程序。我们提供 100 多种工具和集成，涵盖 CRM、HRM、销售、生产力、开发和社交媒体等各个行业。对于任何大型语言模型，您都可以免费集成第三方服务，例如 GitHub、Slack、Gmail、Doscord 等。

请帮我们加一颗星。🥹

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

为 Composio 代码库加星标 ⭐

项目描述

如上所述，该项目查找相关电子邮件并将其路由到相应的 Slack 频道和电子邮件 ID。

以下是整体工作流程。

连接服务：配置 Slack 和 Gmail 集成。
配置关键字：添加 AI 工具将用于过滤和分类电子邮件的关键字。
电子邮件处理：AI 工具将反复轮询 Gmail 收件箱并过滤所需的电子邮件。
智能路由：机器人分析电子邮件内容并将其路由到适当的 Slack 频道和电子邮件 ID。

技术栈

要完成该项目，您将需要以下内容。

前端：React、Vite、Tailwind。
后端： FastAPI、Pydantic。
身份验证： Firebase。
AI 代理： Composio、CrewAI 和 OpenAI。

快速描述

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

先决条件

对于 LLM，我们将使用 OpenAI 的 GPT-4o。

要获取 OpenAI API 密钥，请访问其网站并创建 API 密钥。您可能还需要一些积分。

您也可以使用 Google 的Gemini模型。

让我们开始吧！🔥

该项目分为三个部分。

Firebase 身份验证。
使用 Composio 和 CrewAI 构建 AI 机器人
使用 FastAPI 配置后端。
使用 React 和 Vite 构建前端。

要快速开始，请克隆此存储库。

进入后端目录并运行安装脚本。这将创建一个虚拟环境并下载必要的库。

（注意：如果无法执行 it.sh，请授予权限 -> chmod +x setup）然后系统将提示您登录 Composio、链接 Gmail 并访问 Slack 工作区。

在.env 文件中添加 API 密钥。

这是安装文件。


#!/bin/bash

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

# Activate the virtual environment
echo "Activating virtual environment..."
source ~/.venvs/gmail_support_bot/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 account"
composio login

# Add Gmail tool
echo "Add Gmail tool"
composio add gmail

# Add Slackbot tool
echo "Add Slakbot tool"
composio add slackbot

# Copy env backup to the .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 仪表板并监控您的集成。

执行安装脚本。

  cd backend && ./setup.sh

Firebase 身份验证

这是可选的，您可以跳过此部分。

我们将使用 Firebase 进行用户身份验证和授权。

因此，导入库并使用服务帐户进行身份验证。帐户凭据存储在 JSON 文件中。

import firebase_admin
from firebase_admin import credentials, auth, firestore
from pathlib import Path
import os

cred = credentials.Certificate(f"{Path.cwd()}/firebase/support-bot-49f93-94ae307979d3.json")
firebase_admin.initialize_app(cred)

使用所需的凭据初始化 Firebase 管理员。

一旦通过身份验证，我们就可以初始化 Firestore 客户端：

db = firestore.client()

这使我们能够查询和操作 Firestore 集合中的文档。

定义效用函数

def get_user_by_username(username):
    users_ref = db.collection('users')
    query = users_ref.where('uid', '==', username).limit(1)
    docs = query.get()

    for doc in docs:
        return doc.to_dict()

    return False

users该函数通过使用字段查询集合从 Firestore 获取用户文档uid。

def update_row(uid, new_row):
    users_ref = db.collection('users')
    query = users_ref.where('uid', '==', uid).limit(1)
    docs = query.get()

    for doc in docs:
        try:
            doc.reference.update({'sheetsConfig.row': str(new_row)})
            return True
        except Exception as e:
            print(f"Error updating user row: {e}")
            return False

    print(f"User with uid {uid} not found")
    return False

sheetsConfig.row上述函数更新用户文档中的特定字段。

def update_spreadsheet_id(username: str, spreadsheet_id: str):
    users_ref = db.collection('users')
    query = users_ref.where('username', '==', username).limit(1)
    docs = query.get()

    for doc in docs:
        try:
            doc.reference.update(
                {'sheetsConfig.spreadsheet_id': spreadsheet_id})
            print(f"Successfully updated spreadsheet_id for user {username}")
            return True
        except Exception as e:
            print(f"Error updating spreadsheet_id for user {username}: {e}")
            return False

    print(f"User {username} not found")
    return False

与行更新类似，此函数更新sheetsConfig.spreadsheet_id用户的 Firestore 文档中的字段：

构建人工智能机器人

现在让我们构建人工智能机器人。

定义提示

首先，我们将在prompt.py文件中定义两个提示。

prompt1 = f"""
    1. Send an automatic reply to the sender of the email with the following message:
        "Thank you for your email. We have received it and will get back to you shortly"
        using GMAIL_REPLY_TO_THREAD action & if any attachments are present, use GMAIL_SEND_EMAIL send that too.
    2. Check if the email subject (subject) or body (messageText) in the payload contains any of the keywords specified in this dictionary: {[{'slackChannel': 'dev-channel', 'email': 'hrishikeshvastrad14@gmail.com', 'keywords': 'bugs, errors, issues'}, {'slackChannel': 'growth-channel', 'email': 'mskarthikugalawat@gmail.com', 'keywords': 'Collaboration, partnership, sponser'}, {'slackChannel': 'hrishikesh-channel', 'email': 'hrishikesh@gmail.com', 'keywords': 'bill'}]}.
    3. If a keyword match is found:
        a. Check if the original email contains any attachments.
        b. If attachments are present, use the GMAIL_GET_ATTACHMENT action to download them.
        c. Send the email payload to the corresponding email address and slack channel. If attachments are present, include the downloaded attachments.
        message: 'Forwarded email: subject & body.'
    Payload: {payload}
    """
prompt2 = f"""
    1. Check if the email subject (subject) or body (messageText) in the payload contains any of the keywords specified in this dictionary: {keywords}.
    2. If a keyword match is found:
        a. Check if the original email contains any attachments.
        b. If attachments are present, use the GMAIL_GET_ATTACHMENT action to download them.
        c. Send the email payload to the corresponding email address and slack channel. If attachments are present, include the downloaded attachments.
        message: 'Forwarded email: subject & body.'
    Payload: {payload}
    """

以下是每个提示的含义。

提示 1：（适用于没有附件的电子邮件）
- 此提示指示系统使用指定的消息自动回复电子邮件，检查电子邮件主题或正文中的关键字，如果找到匹配项，则将电子邮件（带有任何附件）转发到指定的 Slack 频道和电子邮件地址。
提示 2：（针对带有附件的电子邮件）
- 此提示重点检查电子邮件主题或正文中的特定关键字，如果找到匹配项，则将电子邮件及其附件转发到相应的 Slack 频道和电子邮件地址。

接下来，在中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
from firebase.init import update_row
from firebase.init import db
# get_user_by_username
from pathlib import Path
from prompts import prompt1,prompt2

load_dotenv()

现在，创建 OpenAI 和 Composio 工具集的实例。

lm = ChatOpenAI(model="gpt-4o")

# Trigger instance
composio_toolset1 = ComposioToolSet(api_key=os.environ.get("COMPOSIO_API_KEY"))

使用 G 初始化事件监听器

listener = composio_toolset1.create_trigger_listener()

接下来用事件监听器装饰器定义一个回调函数。

@listener.callback(filters={"trigger_name": "GMAIL_NEW_GMAIL_MESSAGE"})
def callback_new_message(event: TriggerEventData) -> None:
...

在装饰器内部，触发器名称设置为GMAIL_NEW_GMAIL_MESSAGE，因此每当有新的电子邮件事件时，触发器就会将电子邮件内容转发到回调消息。

现在，此回调函数接收事件数据并预处理电子邮件内容。根据内容，机器人会将电子邮件路由到适当的电子邮件 ID 和 Slack 频道。

首先，从有效载荷中提取发件人的电子邮件。

listener.callback(filters={"trigger_name": "GMAIL_NEW_GMAIL_MESSAGE"})
def callback_new_message(event: TriggerEventData) -> None:
    print("Received new email")
    payload = event.payload
    sender_email = payload['sender']
    sender_email = sender_email.strip()

现在，使用 UID 从电子邮件中提取用户名。

@listener.callback(filters={"trigger_name": "GMAIL_NEW_GMAIL_MESSAGE"})
def callback_new_message(event: TriggerEventData) -> None:
    print("Received new email")
    payload = event.payload
    sender_email = payload['sender']
    sender_email = sender_email.strip()

    def get_user_by_username(username):
        users_ref = db.collection('users')
        query = users_ref.where('username', '==', username).limit(1)
        docs = query.get()

        for doc in docs:
            return doc.to_dict()

        return False

    user = get_user_by_username(event.metadata.connection.clientUniqueUserId)
    uid = user['uid']
    keywords = user['keywords']
    user_email = user['email']

现在，使用所需的操作定义 Composio 工具集。

@listener.callback(filters={"trigger_name": "GMAIL_NEW_GMAIL_MESSAGE"})
def callback_new_message(event: TriggerEventData) -> None:
    print("Received new email")
    payload = event.payload
    sender_email = payload['sender']
    sender_email = sender_email.strip()

    def get_user_by_username(username):
        users_ref = db.collection('users')
        query = users_ref.where('username', '==', username).limit(1)
        docs = query.get()

        for doc in docs:
            return doc.to_dict()

        return False

    user = get_user_by_username(event.metadata.connection.clientUniqueUserId)
    uid = user['uid']
    keywords = user['keywords']
    user_email = user['email']

    # Tools
    composio_toolset = ComposioToolSet(
        api_key=os.environ.get("COMPOSIO_API_KEY"),
        output_dir=Path.cwd() / "attachments",
        entity_id=event.metadata.connection.clientUniqueUserId)
    tools = composio_toolset.get_actions(actions=[Action.GMAIL_SEND_EMAIL, 
                                                  Action.GMAIL_GET_ATTACHMENT, 
                                                  Action.GMAIL_REPLY_TO_THREAD, 
                                                  Action.SLACKBOT_SENDS_A_MESSAGE_TO_A_SLACK_CHANNEL])

以下是我们将要采取的操作。

GMAIL_SEND_EMAIL：用于向用户 ID 发送电子邮件。
GMAIL_GET_ATTACHMENT：用于从电子邮件中提取附件。
GMAIL_REPLY_TO_THREAD：用于回复电子邮件线程。
SLACKBOT_SENDS_A_MESSAGE_TO_A_SLACK_CHANNEL：用于在 Slack 频道中发送消息。

现在，定义 CrewAI 代理。

# Agent
    email_assistant = Agent(
        role="Email Assistant",
        goal="Process incoming emails, send auto-replies, and forward emails based on keywords",
        backstory="You're an AI assistant that handles incoming emails, sends automatic responses, and forwards emails to appropriate recipients based on content, including attachments.",
        verbose=True,
        llm=llm,
        tools=tools,
        allow_delegation=False,
    )

代理会被赋予特定的角色、目标和背景故事。这些会在任务完成前为 LLM 增添额外的背景信息。它还拥有 LLM 和工具。

现在，定义任务和工作人员......

task_description = prompt1 if user_email != sender_email else prompt2

process_new_email = Task(
        description=task_description,
        agent=email_assistant,
        expected_output="Summary of email processing, including confirmation of auto-reply sent, whether the email was forwarded & message sent to slack based on keyword matching, and if any attachments were included in the forwarded email.",
    )

email_processing_crew = Crew(
        agents=[email_assistant],
        tasks=[process_new_email],
        verbose=1,
        process=Process.sequential,
    )
result = email_processing_crew.kickoff()
return result

这就是上面的代码块中发生的事情。

任务描述、预期输出和代理定义了任务。关联的电子邮件代理将完成该任务。团队由代理和任务定义。该团队是负责管理代理工作流程的协调者。
最后执行crew run，并返回结果。

print("Email trigger listener activated!")
listener.listen()

这将启动事件监听器。

运行此脚本，您将拥有一个活动事件监听器，可以随时获取电子邮件并协调工作流程。

FastAPI 后端

现在，我们将在main.py文件中定义我们的 API 端点。

导入库和模块并创建 FasAPI 应用程序。

from fastapi import FastAPI, HTTPException, Request, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel
from fastapi.middleware.cors import CORSMiddleware
from firebase.init import auth
from composio_config import createNewEntity, isEntityConnected, enable_gmail_trigger
import logging
from initialise_agent import initialise

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()

添加来源和中间件。

origins = [
    "http://localhost",
    "http://localhost:5173",
]

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

定义 Pydantic 模型。

# Pydantic models
class UserData(BaseModel):
    username: str
    appType: str

class NewEntityData(BaseModel):
    username: str
    appType: str
    redirectUrl: str

class EnableTriggerData(BaseModel):
    username: str

class InitialiseAgentData(BaseModel):
    username: str

定义以下端点。

@app.post("/newentity")
async def handle_request(user_data: NewEntityData,
                         decoded_token: dict = Depends(verify_token)):
    user_id = decoded_token['uid']
    username = user_data.username
    appType = user_data.appType
    redirectUrl = user_data.redirectUrl
    res = createNewEntity(username, appType, redirectUrl)
    return res

@app.post("/enabletrigger")
async def handle_request(user_data: EnableTriggerData,
                         decoded_token: dict = Depends(verify_token)):
    user_id = decoded_token['uid']
    username = user_data.username
    res = enable_gmail_trigger(username)
    return res

@app.post("/checkconnection")
async def handle_request(user_data: UserData,
                         decoded_token: dict = Depends(verify_token)):
    user_id = decoded_token['uid']
    username = user_data.username
    appType = user_data.appType
    res = isEntityConnected(username, appType)
    return res

@app.post("/initialiseagent")
async def handle_request(user_data: InitialiseAgentData,
                         decoded_token: dict = Depends(verify_token)):
    username = user_data.username
    res = initialise(username)
    return res

@app.get("/")
async def handle_request():
    return "ok"

以下是每个端点的描述。

`POST /new entity：使用提供的用户名、appType 和 redirectUrl 创建一个新实体，并使用解码的令牌进行身份验证。
POST /enabletriggerusername：根据提供的数据和身份验证令牌为指定内容启用 Gmail 触发器。
* `POST /检查连接*：此功能检查给定的实体是否已username连接appType并使用解码的令牌进行身份验证。
POST /initialiseagent：为提供的初始化代理username，使用解码的令牌进行身份验证。
GET /：这是一个简单的健康检查，返回“okay”以确认服务正在运行。

最后，定义 Unicorn 服务器。

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

# Start the server (if running locally)
# Run the following command in your terminal: uvicorn main:app --reload

通过运行脚本启动 Uvicorn 服务器。

python main.py

这将在端口 8000 启动您的服务器。

使用 React 和 Vite 的前端

为了简洁起见，我们不会深入探讨。

让我们看一下重要页面。

主页

这将成为我们的主页。

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;

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

您可以在此处查看检查页面的代码。

在此页面上，用户可以添加他们的 Slack 和 Gmail 帐户，并使用各自的 Slack 频道和 Gmail ID 配置关键字。

定义应用程序和入口点

这是app.jsx文件。

import { BrowserRouter, Routes, Route, Navigate } from "react-router-dom";
import { onAuthStateChanged } from "firebase/auth";
import { auth } from "./config/firebase";
import Navbar from "./components/Navbar";
import Home from "./pages/Home";
import Footer from "./components/Footer";
import ScrollToTop from "./components/ScrollToTop";
import { useState, useEffect } from "react";
import Login from "./pages/Login";
import Settings from "./pages/Settings";
import NotFound from "./pages/NotFound";
import SkeletonLoader from "./components/SkeletonLoader";
import { SnackbarProvider } from 'notistack'

const ProtectedRoute = ({ user, children }) => {
  if (!user) {
    return <Navigate to="/login" replace />;
  }
  return children;
};

const App = () => {
  const [user, setUser] = useState(null);
  const [loading, setLoading] = useState(true);

  useEffect(() => {
    const unsubscribe = onAuthStateChanged(auth, (user) => {
      setUser(user);
      setLoading(false);
    });

    return () => unsubscribe();
  }, []);

  if (loading) {
    return <SkeletonLoader />
  }

  return (
    <BrowserRouter>
      <SnackbarProvider autoHideDuration={3000} preventDuplicate={true} anchorOrigin={{ vertical: 'bottom', horizontal: 'center' }}>
        <Navbar user={user} />
        <ScrollToTop />
        <Routes>
          <Route path="/login" element={<Login />} />
          <Route path="/Settings" element={
            <ProtectedRoute user={user}>
              <Settings user={user} />
            </ProtectedRoute>
          } />
          <Route path="/" element={<Home />} />
          <Route path="*" element={<NotFound />} />
        </Routes>
        <Footer />
      </SnackbarProvider>
    </BrowserRouter>
  );
}

export default App;

Firebase 身份验证：它用于onAuthStateChanged跟踪用户的身份验证状态，并Settings根据用户是否登录有条件地呈现受保护的路由（如）。
路由和导航：该应用程序使用react-router-dom不同的路由（如/login、/settings和/）来处理，受保护的路由将未经身份验证的用户重定向到登录页面。
加载和 UI 组件：当 Firebase 确定用户的身份验证状态时，会显示一个骨架加载器，并使用来管理全局通知not stack。

现在，main.jsx作为应用程序的入口点。

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

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

完成所有操作后，启动该应用程序。

运行应用程序

最后，使用以下 npm 命令运行应用程序。

npm run dev

这将在 localhost:5345 上启动前端服务器。

您现在可以访问该应用程序，配置代理并查看其运行情况。

后续步骤

在本文中，您构建了一个完整的 AI 工具，用于处理支持电子邮件并将其路由到相应的电子邮件 ID 和 Slack 频道。

如果您喜欢这篇文章，请探索并加注 Composio 存储库以获取更多 AI 用例。

为 Composio 代码库加星标 ⭐

感谢您阅读本文！

文章来源：https://dev.to/composiodev/i-saved-my-team-20-hours-by-automating-support-emails-with-this-ai-tool-13o6