LangChain で AI に Google Calendar API を使わせる

FacebooktwitterredditpinterestlinkedinmailFacebooktwitterredditpinterestlinkedinmail

忙しく仕事してるんだし、スケジュール管理とか秘書にやってほしいなぁと思ったことはないでしょうか。とはいえ、秘書のイメージはフィクションのそれで、実際にはどんなことをしてくれるのかもよく分かっていなかったりします。

そんな一般労働階級にとって希望の星の ChatGPT Plugin が発表されていますが、今のところ使える気配がありません(ついでに GPT-4 API も……)。

でもスケジュール管理くらい、ちゃちゃっと自然言語で処理したいところです。ということで、 LangChain を使ってチャットなどから Google Calendar API を叩いてみます。

過去の関連記事;Google Calendar API のライブラリ化

過去の関連記事:LangChain Output Parser

LLM から Google Calendar を使わせる

Google Calendar API を LLM から使わせるにはいくつかの方法が考えられますが、 API 自体が JSON 形式で入力を受け付けているので LangChain の Output Parser を使って JSON 形式の出力を作らせるのが最も手軽で、かつ「それっぽい」と思います(前回のような極力、 LLM を遠さないやり方とは逆ですね)。

ただ、エンドポイントが実行する機能によって分かれています。アプローチとして、エンドポイントの数だけ Tool を使って Agent に選択させる方法も考えられますが、それほど複雑でもないので Schedule や Calendar 機能としてまとめてしまってもそこまで精度に問題がでないと思います。そこで、 Google の API に投げる JSON ファイルに、実行したい操作の情報をまとめた JSON 文字列を作らせます。

Pydantic オブジェクトの定義

from pydantic import BaseModel, Field
from typing import Optional, Dict, Any

class CalendarAction(BaseModel):
    action: str = Field(description="The action to be performed. Supported actions are 'get', 'create', 'search', 'update', and 'delete'.")

    class EventData(BaseModel):
        summary: Optional[str] = Field(None, description="The summary of the event.")
        start: Optional[Dict[str, str]] = Field(None, description="The start time of the event. It contains 'dateTime' and 'timeZone' keys.")
        end: Optional[Dict[str, str]] = Field(None, description="The end time of the event. It contains 'dateTime' and 'timeZone' keys.")
        query: Optional[str] = Field(None, description="The search query to find matching events.")
        updated_data: Optional[Dict[str, Any]] = Field(None, description="The updated data for the event. It contains keys to update such as 'summary', 'start', and 'end'.")

    event_data: Optional[EventData] = Field(None, description="The data for the event or query. Contains fields such as 'summary', 'start', 'end', 'query', and 'updated_data'.")

作成する Pydantic オブジェクトはこのようになります。 CalendarAction クラスで EventData クラスを包含させます。 action 文字列によって、 Google Calendar で呼び出すエンドポイントを決定しています。FieldのDescription にそれぞれの内容を説明させます。

GoogleCalendar クラスの定義

import os
import datetime
from google.oauth2 import service_account
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
from googleapiclient.discovery import build
import json

class GoogleCalendar:
    def __init__(self, credentials_file, scopes=['https://www.googleapis.com/auth/calendar']):

    ### 中略 ###
    
    def process_json_data(self, json_data: str):
        data = CalendarAction.parse_raw(json_data)

        if data.action == "get":
            events = self.get_events()
            for event in events:
                print(event['summary'], event['start']['dateTime'])
            return events

        elif data.action == "create":
            event_data = data.event_data.dict()
            created_event = self.create_event(event_data)
            print("Event created:", created_event['id'])
            return "Finish."

        elif data.action == "search":
            query = data.event_data.query
            events = self.search_events(query)
            for event in events:
                print(event['summary'], event['start']['dateTime'])
            return events

        elif data.action == "update":
            query = data.event_data.query
            updated_data = data.event_data.updated_data

            today = datetime.datetime.utcnow().date()
            time_min = datetime.datetime.combine(today, datetime.time.min).isoformat() + 'Z'
            time_max = datetime.datetime.combine(today, datetime.time.max).isoformat() + 'Z'

            events = self.search_events(query, time_min=time_min, time_max=time_max)
            for event in events[:20]:
                updated_event = self.update_event(event['id'], updated_data)
                print("Event updated:", updated_event['id'])
            return "Finish."

        elif data.action == "delete":
            query = data.event_data.query

            today = datetime.datetime.utcnow().date()
            time_min = datetime.datetime.combine(today, datetime.time.min).isoformat() + 'Z'
            time_max = datetime.datetime.combine(today, datetime.time.max).isoformat() + 'Z'

            events = self.search_events(query, time_min=time_min, time_max=time_max)
            for event in events[:20]:
                self.delete_event(event['id'])
                print("Event deleted:", event['id'])
            return "Finish."

        else:
            print("Invalid action")
            return "Finish with Error!"

クラスの内容の大半は、以前ライブラリ化したものと同様です。process_json_data メソッドを作成して、 action 文字列によって該当するメソッドを呼び出しています。

LLM を搭載したラッパークラス

更に今回は、これらのクラスを LLM(GPT-4など)から操作するための、ラッパークラスを作ってみたいと思います。ちゃんとしたものではないですが、 LangChain の Tool みたいな、あんな感じのを見よう見まねで作ってみます。

class GoogleCalendarTool:
    def __init__(self, credentials_file, llm=None, time_zone='JST', memory=None, prompt="""Follow the user query and take action on the calendar appointments.
    Current time: {current_time}, timeZone: JST.
    History: {chat_history}
    Format: {format_instructions}

    User Query: {query}
    Processing and reporting must be done in Japanese. If unclear, do not process and ask questions.""" ,scopes=['https://www.googleapis.com/auth/calendar']):
        self.cal = GoogleCalendar(credentials_file)
        self.llm = llm
        if self.llm is None:
            raise ValueError("Error: LLM is undefined.")
        self.time_zone = time_zone
        self.memory = memory
        # Parser に元になるデータの型を提供する
        self.parser =  PydanticOutputParser(pydantic_object=CalendarAction)
        self.prompt = PromptTemplate(
            template=prompt,
            input_variables=["query", "current_time", "chat_history"],
            partial_variables={"format_instructions": self.parser.get_format_instructions()}
        )
        self.chain = LLMChain(llm=self.llm, prompt=self.prompt)

    def run(self, query):
        _input = self.prompt.format_prompt(query=query, current_time=datetime.datetime.now(), chat_history=self.memory)
        output = self.chain.run({'query': query, 'current_time': datetime.datetime.now(), 'chat_history': self.memory})

        return self.cal.process_json_data(output)

LangChain の LLM オブジェクトと、Google Calendar の認証に使用する credential ファイル、それから Prompt を受け取って実体化するクラスです。Promptはまあまあ標準的に動作しそうなものを入れていますが、しばしばイベントが英語化してしまうのが課題です。

動作としては、 PydanticOutputParserによるformat_instruction を元に、ユーザーが希望するカレンダーイベントを整形してGoogle Calendar API を操作します。

工夫したポイントとしては、人間はつい「明日」とか、「昨日」とか、相対的な日付指定をしてしまうためにdatetime.now()と入れるcurrent_timeを変数として入れています。これを元に、 LLM は日付や時刻を計算します。概ね問題なく動作しますが、より厳密な場合は Agent を入れ子にして Tool に日付計算用 Tool を導入する必要があると思います。

Memoryはほぼ動作しない

ソースでは Memory を受け取るようにしていますが、ユーザーからの要望と LLM の出力を単に格納したものではほぼ動作しません(「さっきの予定キャンセルして」とかだと過去のeventを全て削除しようとしたりします)。

Memoryを実装するには、操作したevent id など、かなり厳密なデータを保持する必要があるかもしれません(またはGPT-4であればAgentにmemoryを実装するだけでも上手くreAct的に使ってくれるかもしれませんが……)。

使い方

credentials_file = "credentials.json"
from GCalTool import GoogleCalendarTool

# モデル作成
llm = ChatOpenAI(temperature=0)

calendar_tool = GoogleCalendarTool(credentials_file, llm=ChatOpenAI(temperature=0), memory=None)
calendar_tool.run('明日の12時にデートの約束を入れて')

 課題

いかにもToolっぽいですが、LLM の Toolとして使うには Observation として正しい回答をする必要があります。現在の実装では、単に実行が正常に終了すればいい create や delete はともかく、search などのメソッドでは出力されたスケジュールのリストを単に返しているのできちんとした動作にはなかなかなりません。

また先述した通り、 Memory が期待したような動作をしないことも挙げられるので、通常の会話の記憶ではなく、操作ログのようなものを実装する工夫なども必要になってくると思います。

まとめ

  • (Web)API を叩くには Pydantic Output Parser が便利
  • Tool化するには返り値も結構気を使う必要がある(場合によっては、APIからの返り値を処理するLLMも必要かもしれない)
  • Memory も単に LangChain のものを入れるだけでは API を叩く場合には上手く動かない
FacebooktwitterredditpinterestlinkedinmailFacebooktwitterredditpinterestlinkedinmail

コメントを残す

メールアドレスが公開されることはありません。 が付いている欄は必須項目です

このサイトはスパムを低減するために Akismet を使っています。コメントデータの処理方法の詳細はこちらをご覧ください

最新の記事