最近，我的行事曆變得如此忙碌，我真的需要一些幫助來管理它。事實上，我意識到這是Composio同事中非常普遍的問題。

我們收到大量有關多個事件的電子郵件，管理它們已成為一場噩夢。

我著手建立一個自動化程序，可以讀取我收到的電子郵件，了解其內容，確定是否需要安排活動，最後起草一封帶有邀請連結的電子郵件。

鑑於所涉及的挑戰，傳統方法對這項任務無效：

• 要求理解電子郵件的上下文。

• 需要與 Gmail 和 Google 日曆無縫整合。

我選擇了大型語言模型（LLM）來應對第一個挑戰。然而，真正的複雜性在於將 Gmail 和日曆與法學碩士集成，特別是因為這些服務可能非常挑剔。

現在，想像一個為您處理 Google OAuth 的函式庫，簡化 API 端點與 LLM 的整合。這將改變遊戲規則，對嗎？

這正是我們在 Composio 提供的服務。

Composio - 第一款人工智慧整合平台

這是關於我們的快速介紹。

Composio 是一個開源工具基礎設施，用於建立強大且可靠的 AI 應用程式。我們提供超過 100 多個跨行業垂直領域的工具和集成，從 CRM、HRM 和銷售到生產力、開發和社交媒體。

我們處理所有這些應用程式的使用者身份驗證和授權，並使所有 API 端點與各種 AI 模型和框架的連接變得簡單。因此，您花更多的時間為股東創造價值，而不是解決不必要的錯誤。

請幫我們加一顆星。 🥹

這將幫助我們建立更多這樣的文章💖

{% cta https://dub.composio.dev/schagent %}為 Composio.dev 儲存庫加註星標 ⭐{% endcta %}

LangGraph - 建立生產就緒的人工智慧應用程式的框架

要完成該專案，您將需要另一個依賴項 LangGraph。

LangGraph 是一個用於建立有狀態人工智慧工作流程的開源框架。它使用模組化圖形架構定義工作流程，從而更輕鬆地控制工作流程中的每個步驟。

關鍵零件

• 狀態：

- A shared data structure that represents the current snapshot of your application.

- It can be any Python type, but is typically a `TypedDict` or `Pydantic` `BaseModel`.

• 節點：

- Python functions that encode the logic of your agents.

- They receive the current State as input, perform some computation or side-effect, and return an updated State.

• 邊緣：

- Python functions that determine which Node to execute next based on the current State.

- They can be conditional branches or fixed transitions.

強烈建議您在繼續之前先閱讀本指南。

在本文中，您將建立一個端到端 AI 機器人，它可以讀取傳入的電子郵件、處理它們、決定是否需要安排它們，最後起草一封帶有日曆連結的邀請電子郵件給收件人。

整體工作流程

讓我們來看看整體的工作流程。

• 使用 Composio 連接 Gmail 和 Google 日曆。

• 在 Composio 中啟用觸發器來接收郵件。

• 使用 LangGraph 建立 AI 機器人。

• 該機器人會輪詢 Gmail 是否收到傳入電子郵件。

• 電子郵件將傳遞給機器人進行進一步分析。

• 如果電子郵件包含活動安排資訊：

- Yes: The bot fetches free slots from the Calendar and drafts a suitable email with a scheduled event invitation link.

- No: Ignores.

建構人工智慧機器人的先決條件

您將需要滿足以下要求才能完成該專案。

• Composio 用戶帳戶。在這裡免費建立一個。

• OpenAI API 積分。您也可以使用開源法學碩士，例如 Groq 的 Llama。

讓我們來建構它吧！

設定環境

Python 開發的第一條規則始終是建立虛擬環境。

python -m venv gmail-bot

cd gmail-bot
source bit/activate

現在，安裝相依性。

pip install composio-core 
pip install composio-langgraph
pip install python-dotenv

• composio-core ：所有 Composio 功能的核心函式庫。

• composio-langgraph ：Composio 的 LangGraph 插件。

• python-dotenv ：從.env檔設定環境變數。

建立 .env 檔案並設定 OpenAI API 金鑰。

OPENAI_API_KEY="YOUR_KEY"

整合 Gmail 和 Google 日曆

為了整合 Gmail 和日曆，我們將使用 Composio 的專用 CLI，但在此之前請登入您的使用者帳戶。

composio login

完成登入流程以繼續進行。現在，刷新應用程式。

composio apps update

整合 Gmail

composio add gmail

整合Google日曆

composio add googlecalendar

完成整合流程以將這些服務新增至您的使用者帳戶。

Composio 的最大優點是您可以在儀表板上監控這些整合。

接下來，啟用觸發器以在有人發送電子郵件時接收電子郵件。

composio triggers enable gmail_new_gmail_message

您甚至可以從儀表板監控觸發器。

建構人工智慧機器人

正如我們之前討論的，LangGraph 使用節點和邊定義工作流程。

此工作流程將有四個關鍵元件：開始、結束、工具節點和調度程序節點。

• 開始：工作流程的入口點。

• 工具節點：呼叫 LLM 可用工具的句柄。

• 調度程序節點：管理處理來自電子郵件、工具呼叫等的文字輸入。

• END：工作流程的退出點。

當調度程序節點確定需要呼叫某個工具時，它將控制權移交給工具節點，然後工具節點執行該工具並將結果和控制權傳回調度程序節點。

一旦執行完成，調度程序節點就會終止該進程。

這是一個高級示意圖。

現在，讓我們開始對工作流程進行編碼。

導入庫

首先，匯入所需的庫並使用 .env 設定環境變數。

from typing import Literal, Annotated, Sequence, TypedDict
import operator
import re
from datetime import datetime
from dotenv import load_dotenv
from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, ToolMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, StateGraph
from langgraph.prebuilt import ToolNode
from composio_langgraph import Action, ComposioToolSet
from composio.client.collections import TriggerEventData

# Setup
load_dotenv()

現在，定義常數。

# Constants
SCHEDULER_AGENT_NAME = "Scheduler"
TOOL_NODE_NAME = "ToolNode"
DATE_TIME = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

定義工具

接下來，我們將定義用於安排事件和起草電子郵件的工具和工具節點。

# Initialize LLM and tools
llm = ChatOpenAI(model="gpt-4-turbo")
composio_toolset = ComposioToolSet()

schedule_tools = composio_toolset.get_actions(
    actions=[
        Action.GOOGLECALENDAR_FIND_FREE_SLOTS,
        Action.GOOGLECALENDAR_CREATE_EVENT,
    ]
)
email_tools = composio_toolset.get_actions(actions=[Action.GMAIL_CREATE_EMAIL_DRAFT])
tools = [*schedule_tools, *email_tools]

tool_node = ToolNode(tools)

上面的程式碼區塊，

• 使用 OpenAI 的 GPT-4 Turbo 模型初始化大型語言模型 (LLM)。

• 建立 Composio 工具集的實例，該實例提供對 Google 服務整合的存取。

• 取得操作以尋找空閒插槽並從與 Google 日曆相關的 Composio 工具集中建立事件。

• 取得在 Gmail 中建立電子郵件草稿的操作。

• 建立一個工具節點，其中包含要在 LangGraph 中使用的所有操作。

注意：工具節點抽象化了 LLM 的呼叫和處理工具呼叫。

定義提示

接下來，定義系統提示。這很重要，因為它將為法學碩士提供處理傳入郵件所需的背景。

# System prompt
scheduler_system_prompt = f"""
You are an AI assistant specialized in analyzing emails, creating calendar events, and drafting response emails. Your primary tasks are:

1. Analyze email content:
   a. Understand the email received from the sender. 
   b. Determine if an event should be created based on the email content.
   c. Extract relevant information such as proposed meeting times, topics, and participants.

2. Manage calendar events:
   a. If an event should be created, use the Google Calendar Find Free Slots action to identify available time slots.
   b. Once a suitable slot is found, use the Google Calendar Create Event action to schedule the event.
   c. Ensure to invite the email sender and any other relevant participants to the event.

3. Draft response emails:
   a. If an event was created, draft a confirmation email for the sender.
   b. The email should include:
      - A clear subject line (e.g., "Meeting Scheduled")
      - A brief description of the scheduled meeting's purpose
      - The date, time, and duration of the event
      - Any other relevant details or instructions for the participants

Remember:
- The current date and time is {DATE_TIME}.
- All conversations and scheduling occur in the IST timezone.
- Be courteous and professional in all communications.
- If you encounter any errors when making function calls, try passing an empty config ({{"config": {{}}}}).
- Always provide a FINAL ANSWER when you've completed all necessary tasks.

You aim to efficiently manage scheduling and communication, ensuring a smooth experience for all parties involved.
"""

我們定義了一個清晰且逐步的系統提示來指導法學碩士。

定義 AgentState 和輔助函數

在 LangGraph 中，AgentState 追蹤代理工作流程在任何給定點的狀態。

# Define AgentState
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
    sender: str

# Helper functions
def create_agent_node(agent, name):
    def agent_node(state):
        result = agent.invoke(state)
        if not isinstance(result, ToolMessage):
            result = AIMessage(**result.dict(exclude={"type", "name"}), name=name)
        return {"messages": [result], "sender": name}
    return agent_node

def create_agent(system_prompt, tools):
    prompt = ChatPromptTemplate.from_messages([
        ("system", system_prompt),
        MessagesPlaceholder(variable_name="messages"),
    ])
    return prompt | llm.bind_tools(tools)

這是正在發生的事情，

• AgentState(TypedDict) ：定義代理狀態的字典結構，包含訊息清單和發送者姓名。

• create_agent_node(agent, name) ：建立一個函數來處理代理的狀態，呼叫代理，並將結果格式化為訊息，並將代理的名稱作為傳送者。

• create_agent(system_prompt, tools) ：此函數透過根據系統指令和訊息建立提示範本並將LLM綁定到指定的工具來建構代理。

使用系統提示符號、工具以及使用 Scheduler_agent 和常數 SCHEDULER_AGENT_NAME 的代理節點建立代理程式。

scheduler_agent = create_agent(scheduler_system_prompt, tools)
scheduler_node = create_agent_node(scheduler_agent, SCHEDULER_AGENT_NAME)

此節點處理與安排 Google 日曆事件相關的所有事務，從檢查時段到確認事件。

定義節點、邊和工作流程

眾所周知，LangGraph 遵循有向圖結構，其中每個節點都包含可執行檔（可以執行的程式碼）和引導資訊流的邊。

# Create workflow
workflow = StateGraph(AgentState)

workflow.add_node(SCHEDULER_AGENT_NAME, scheduler_node)
workflow.add_node(TOOL_NODE_NAME, tool_node)

在上面的程式碼區塊中，

• 我們使用StateGraph(AgentState)定義工作流程圖。 AgentState 追蹤狀態物件。

• 新增了兩個節點：tool_node 和scheduler_node。

現在，定義邊緣來引導資訊流。

# Router function
def router(state) -> Literal["call_tool", "__end__", "continue"]:
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "call_tool"
    if "FINAL ANSWER" in last_message.content:
        return "__end__"
    return "continue"

workflow.add_edge(START, SCHEDULER_AGENT_NAME)
workflow.add_edge(SCHEDULER_AGENT_NAME, END)

這是用於定義工作流程邊緣的程式碼。

• 路由器從代理狀態傳回節點名稱。這有助於導航工作流程。它還可以將流量控制從一個節點轉移到另一個節點並結束工作流程。

• 我們定義了從 START→SCHEDULER_AGENT_NAME→END 移動的兩條正常邊緣。

條件邊

此工作流程包括兩個關鍵的條件邊：

• 從SCHEDULER_AGENT_NAME到各節點的條件邊：

    pythonCopy code
    workflow.add_conditional_edges(
        SCHEDULER_AGENT_NAME,
        router,
        {
            "continue": SCHEDULER_AGENT_NAME,
            "call_tool": TOOL_NODE_NAME,
        },
    )

- **Purpose:** This edge determines the next step after the scheduler agent processes the current state.

- **Condition:** The `router` function checks the content of the last message in the state.

- **Possible Outcomes:**

    - **"continue":** If the `router` function returns `"continue"`, the workflow stays in the `SCHEDULER_AGENT_NAME` node, allowing the agent to keep processing.

    - **"call_tool":** If the `router` function returns `"call_tool"`, the workflow transitions to the `TOOL_NODE_NAME` node, where the relevant tool (e.g., find free slots or create an event) is invoked.

• 從TOOL_NODE_NAME到SCHEDULER_AGENT_NAME條件邊：

    pythonCopy code
    workflow.add_conditional_edges(
        TOOL_NODE_NAME,
        lambda x: x["sender"],
        {SCHEDULER_AGENT_NAME: SCHEDULER_AGENT_NAME},
    )

- **Purpose:** This edge determines where to route the workflow after invoking a tool (like Google Calendar or Gmail).

- **Condition:** The lambda function `lambda x: x["sender"]` checks the sender of the last message.

- **Outcome:** If the sender is the `SCHEDULER_AGENT_NAME`, the workflow returns to the `SCHEDULER_AGENT_NAME` node to continue processing.

最後，編譯工作流程。

app = workflow.compile()

設定事件監聽器

下一步是定義事件偵聽器。當有人向您連接的 Gmail 帳戶發送電子郵件時，這將負責提取郵件。

def extract_sender_email(payload):
    if not any(header.get("name") == "Delivered-To" and header.get("value") for header in payload["headers"]):
        return None

    for header in payload["headers"]:
        if header["name"] == "From":
            match = re.search(r"[\w\.-]+@[\w\.-]+", header["value"])
            return match.group(0) if match else None
    return None

def process_email(email_sender, email_content, thread_id):
    final_state = app.invoke({
        "messages": [
            HumanMessage(content=f"""                              
Please process the email 
Email sender: {email_sender} 
Email content: {email_content}
Thread id: {thread_id}
""")
        ]
    })
    return final_state["messages"][-1].content

listener = composio_toolset.create_trigger_listener()

@listener.callback(filters={"trigger_name": "gmail_new_gmail_message"})
def callback_new_message(event: TriggerEventData) -> None:
    payload = event.payload
    thread_id = payload.get("threadId")
    email_content = payload.get("snippet")
    email_sender = extract_sender_email(payload["payload"])

    if email_sender is None:
        print("No sender email found")
        return

    print(f"Processing email from: {email_sender}")
    output = process_email(email_sender, email_content, thread_id)
    print("Final output:", output)

listener.listen()

那麼，這是上面程式碼區塊的解釋

• extract_sender_email(payload) ：從電子郵件有效負載標頭中提取寄件者的電子郵件地址，如果找到則返回，否則None 。

• process_email(email_sender, email_content, thread_id) ：使用電子郵件詳細資料呼叫工作流程並傳回最終處理的訊息內容。

• 監聽器設定 ( listener = composio_toolset.create_trigger_listener() )：建立一個監聽器來監視特定事件，例如新的 Gmail 訊息。

• @listener.callback(filters={"trigger_name": "gmail_new_gmail_message"}) ：定義一個回呼函數，該函數在偵測到新的 Gmail 訊息事件時觸發，透過工作流程處理電子郵件。

• listener.listen() ：此指令啟動偵聽器，並透過呼叫回呼函數使其主動偵聽並回應新的 Gmail 訊息。

完成所有步驟後，執行 Python 檔案以設定事件偵聽器。

下次您向配置的 Gmail 帳戶發送電子郵件時，機器人將對其進行處理並確定是安排活動還是將其傳遞。如果安排了活動，我們將準備一份電子郵件草稿供您審閱，以便您根據需要發送。

您可以在這裡找到完整的程式碼：調度代理

感謝您的閱讀！ 🎉

幫幫我吧！

如果您覺得這篇文章可以幫助您更了解人工智慧驅動的事件自動化，如果您能給我們一顆星，我將非常高興！請在評論中告訴我。

{% cta https://dub.composio.dev/schagent %}為 Composio.dev 儲存庫加註星標 ⭐{% endcta %}

原文出處：https://dev.to/composiodev/i-built-an-ai-bot-under-65-lines-of-code-that-checks-my-new-emails-and-send-people-invites-ghh