G-gen の堂原です。Gemini を搭載した Vertex AI Search を用いて、Google Chat のチャットボットを作成してみましたので、紹介します。
はじめに
当記事では、Google Cloud（旧称 GCP）の Vertex AI Search を用いて、Google Chat のチャットボットを作成してみます。
Google Chat は Google Workspace に含まれるチャットツールです。REST API である Google Chat API を用いることで、Google Chat から利用するチャットボットを簡単に開発することが可能です。
Google Chat API と Vertex AI Search を組み合わせることで、データの検索や回答の要約をしてくれるチャットボットを開発できます。また、過去の質問と回答の履歴を取り込むことで、チャットボットの回答の精度を上げることができます。
前提知識
Vertex AI Search
Vertex AI Search は Google Cloud の生成 AI（Generative AI）関連サービスの1つで、Retrieval Augmented Generation（RAG）構成を簡単に構築できるサービスです。以下の記事をご参照ください。
Vertex AI Search では、2024年1月のアップデートで、要約文の生成に Gemini Pro が利用できるようになりました。
Google Chat API
Google Chat API は、Google Chat に用意された REST API です。メッセージの投稿や、スペースの管理などをプログラムから行うことができます。Google Chat API を用いることで、Google Chat アプリ（チャットボット）を開発することができます。
Google Chat API を使うことで、対話型・非対話型のチャットボットが作成可能であり、メッセージ形式も柔軟にチューニングできます。
- 参考 : Google Chat - 概要
対話型のチャットボットを開発するには、Google Cloud コンソールで Google Chat API のセットアップを行う必要があります。その際、Google Chat から送られてきたメッセージを処理するためのプログラムを開発して、何らかのプラットフォームで稼働させる必要があります。Google Chat 自体には、プログラムを稼働させるプラットフォームは無いためです。以下は、実装方法の例です。
- プログラムを Cloud Run functions (旧 Cloud Functions) にデプロイする。トリガー URL をGoogle Chat API から指定する
- プログラムを Google Apps Script で実装する。Apps Script プロジェクトのデプロイ ID を Google Chat API から指定する
なお Google Chat API を利用するには、Google Cloud プロジェクトが必須です。
構成図
当記事では、以下のような構成でチャットボットを開発しました。
- Cloud Storage バケット内の PDF ファイルを検索対象とする Vertex AI Search データストア及びアプリを作成
- Google Chat からメッセージを受け取り、Vertex AI Search に検索をかける Cloud Run functions プログラムを開発
- Cloud Run functions のトリガー URL を指定するよう Google Chat API を設定
Google Workspace の公式ドキュメントにも、Cloud Run functions を用いて Google Chat アプリを作成するチュートリアルがあり、上記の構成に似た設定を行っている箇所があるため、参考にしてください。
Vertex AI Search の設定
Vertex AI Search でデータストアとアプリ（App）を作成します。手順については、当記事では詳述しません、
今回のチャットボットでは、要約文を生成する必要があるため、アプリ（App）作成時に Advanced LLM features を有効化します。
Cloud Run functions の設定
パラメータ
以下のようにパラメータを設定します。記載のないパラメータは、デフォルト値とします。
| 設定項目 | 小項目 | 設定値 | 補足 |
|---|---|---|---|
| 環境 | Cloud Run functions | ||
| トリガー | トリガーのタイプ | HTTPS | |
| 認証 | 未承認の呼び出しを許可 | 他の Google Cloud プロジェクトで作成されたチャットボットからの呼び出しを防ぐため、ソースコード内でチェックを行います | |
| サービスアカウント | 「ディスカバリー エンジン閲覧者」ロールを有しているものを指定 | ||
| ランタイム環境変数 | PROJECT_ID | 今回使用している Google Cloud プロジェクトの ID | |
| PROJECT_NUMBER | 今回使用している Google Cloud プロジェクトのプロジェクト番号 | ||
| DATA_STORE | Vertex AI Search データストアの ID | ||
| CHAT_ISSUER | リクエストが作成したチャットボットから来たものかを判別するために利用。値は「chat@system.gserviceaccount.com」で固定 | ||
| PUBLIC_CERT_URL_PREFIX | リクエストが作成したチャットボットから来たものかを判別するために利用。値は「https://www.googleapis.com/service_accounts/v1/metadata/x509/」で固定 | ||
| ランタイム | Python 3.12 | ||
| エントリポイント | get_chat |
ソースコード
requirements.txt
functions-framework==3.* google-cloud-discoveryengine==0.12.3 oauth2client==4.1.3
main.py
ソースコード本文は、本記事の末尾に掲載します。以下に、重要なポイントを解説します。
認証
当記事では、先述のパラメータの通り Cloud Run functions の認証タイプを「未承認の呼び出しを許可」としたうえで、Google Chat からのリクエストに含まれている Authorization ヘッダーの情報を用いて、アクセス元が特定の Google Cloud プロジェクトにてセットアップされたチャットボットであることを確認します。これにより、チャットボット以外からの呼び出しや、他の Google Cloud プロジェクトで作成されたチャットボットからの呼び出しは拒否されます。また、後述する Google Chat API のセットアップを行うためには、Google Cloud プロジェクトに対して IAM 権限 chat.bots.update が必要です。
つまり、Cloud Run functions からメッセージを受け取れるのは、開発者が意図したチャットボットに限定されます。そのチャットボットの設定変更も、IAM 権限を持っている者しかできません。
以上のことから、本構成は十分にセキュアであり、意図しない利用者からチャットボットが利用されることはない、ということができます。
公式ドキュメントでは、Cloud Run functions の認証機能を用いてリクエストの検証を行う方法も紹介されていますが、こちらの方法では他の Google Cloud プロジェクトで作成されたチャットボットからの呼び出しを拒否することが出来ません。
詳細については以下の参考リンクを確認ください。
以下が、ソースコードの該当部分です。
if req.method == "GET": return flask.make_response(flask.jsonify({"message": "Bad Request"}), 400) auth_header = req.headers.get("Authorization") # 「Authorization」ヘッダーが存在するか確認 if not auth_header: print("Missing Authorization header") return flask.make_response(flask.jsonify({"message": "Unauthorized"}), 401) # 「Authorization」ヘッダーが「Bearer」から始まるか確認 if auth_header.split()[0] != "Bearer": print("Authorization header format is incorrect") return flask.make_response(flask.jsonify({"message": "Unauthorized"}), 401) bearer_token = auth_header.split()[1] try: token = client.verify_id_token( bearer_token, PROJECT_NUMBER, cert_uri=PUBLIC_CERT_URL_PREFIX + CHAT_ISSUER) if token['iss'] != CHAT_ISSUER: print("Invalid issuee") return flask.make_response(flask.jsonify({"message": "Unauthorized"}), 401) except: print("Invalid token") return flask.make_response(flask.jsonify({"message": "Unauthorized"}), 401)
基盤モデルの指定
Vertex AI Search においては、デフォルトでは基盤モデルとして Gemini 1.5 Flash を利用します。Vertex AI Search へのリクエスト時にクラス ModelSpec で明示的に指定することで他のモデルを使用することも可能です。
# Vertex AI Searchの出力内容に関する設定 content_search_spec = SearchRequest.ContentSearchSpec( # スニペットを出力させない snippet_spec=SearchRequest.ContentSearchSpec().SnippetSpec( return_snippet=False ), # 要約文を出力させる summary_spec=SearchRequest.ContentSearchSpec().SummarySpec( summary_result_count=3, include_citations=False, # Gemini Proを用いるように指定 model_spec=SearchRequest.ContentSearchSpec().SummarySpec().ModelSpec( version="gemini-1.5-flash-001/answer_gen/v1" ) ) ) # Vertex AI Searchにクエリを投げる response = discov_client.search( SearchRequest( serving_config=serving_config, query=text, page_size=3, content_search_spec=content_search_spec ) )
Google Chat API の設定
Google Cloud プロジェクトで、Google Chat API を有効にします。その後、Google Chat API の「構成」タブにて以下の設定を行います。
| 設定項目 | 小項目 | 設定値 | 補足 |
|---|---|---|---|
| インタラクティブ機能 | 1:1 のメッセージを受信する | チェックをつけるとダイレクトメッセージでチャットボットが使えます | |
| スペースとグループの会話に参加する | チェックをつけるとスペースにチャットボットを導入できます | ||
| 接続設定 | 「HTTP エンドポイント URL」を選択 | ||
| HTTP エンドポイント URL | Cloud Funcitons のトリガー URL を入力 | ||
| 認証オーディエンス | プロジェクト番号 | ||
| 公開設定 | 「このチャットアプリを xxx (現在の Google アカウントの組織に依存) の特定のユーザーとグループが使用できるようにします」にチェックを付けて、チャットボットを使わせたいユーザまたはグループのメールアドレスを入力 | より広い範囲で公開したい場合は Google Workspace Marketplace SDK を用いる必要があります |
添付 : ソースコード
from typing import Any, Mapping from google.cloud.discoveryengine import SearchServiceClient, SearchRequest from google.protobuf.json_format import MessageToDict from oauth2client import client import functions_framework, flask, os, urllib.parse PROJECT_ID = os.environ.get("PROJECT_ID") PROJECT_NUMBER = os.environ.get("PROJECT_NUMBER") DATA_STORE = os.environ.get("DATA_STORE") # Google Chatから送られてくるBearer Tokenの整合に使用 CHAT_ISSUER = os.environ.get("CHAT_ISSUER") PUBLIC_CERT_URL_PREFIX = os.environ.get("PUBLIC_CERT_URL_PREFIX") def search_document(text: str) -> Mapping[str, Any]: discov_client = SearchServiceClient() # Vertex AI Searchのアプリ等基本的な内容を設定 serving_config = discov_client.serving_config_path( project=PROJECT_ID, location="global", data_store=DATA_STORE, serving_config="default_config" ) # Vertex AI Searchの出力内容に関する設定 content_search_spec = SearchRequest.ContentSearchSpec( # スニペットを出力させない snippet_spec=SearchRequest.ContentSearchSpec().SnippetSpec( return_snippet=False ), # 要約文を出力させる summary_spec=SearchRequest.ContentSearchSpec().SummarySpec( summary_result_count=3, include_citations=False, # Gemini Proを用いるように指定 model_spec=SearchRequest.ContentSearchSpec().SummarySpec().ModelSpec( version="gemini-1.5-flash-001/answer_gen/v1" ) ) ) # Vertex AI Searchにクエリを投げる response = discov_client.search( SearchRequest( serving_config=serving_config, query=text, page_size=3, content_search_spec=content_search_spec ) ) # 要約文取得 summary = response.summary.summary_text.replace("<b>", "").replace("</b>", "") # 関連ファイル取得 references = [] for r in response.results: r_dct = MessageToDict(r._pb) link = r_dct["document"]["derivedStructData"]["link"] file_name = link.split("/")[-1] url = urllib.parse.quote(link.replace("gs://", "https://storage.cloud.google.com/"), ":/") file_sentence = f"<a href={url}>{file_name}</a>" references.append(file_sentence) result = { "summary": summary, "references": references } return result # チャットボットが実際に出力するメッセージを作成する def create_message(text: str) -> Mapping[str, Any]: result = search_document(text=text) reference_sentence = "<br>・".join(result["references"]) # 返信文作成 cards = { "cardsV2": [ { "cardId": "searchResults", "card": { "sections": [ { "collapsible": False, "widgets": [ { "textParagraph": { "text": "<b>回答文</b><br>" + result["summary"] } }, { "divider": {} }, { "textParagraph": { "text": "<b>関連ファイル</b><br>・" + reference_sentence } } ] } ] } } ] } return cards # チャットボットからメッセージを受信 # → create_message() で作成した JSON を返す @functions_framework.http def get_chat(req: flask.Request): """ 正しいGoogle Chatから送られてきたリクエストかを確認 """ if req.method == "GET": return flask.make_response(flask.jsonify({"message": "Bad Request"}), 400) auth_header = req.headers.get("Authorization") # 「Authorization」ヘッダーが存在するか確認 if not auth_header: print("Missing Authorization header") return flask.make_response(flask.jsonify({"message": "Unauthorized"}), 401) # 「Authorization」ヘッダーが「Bearer」から始まるか確認 if auth_header.split()[0] != "Bearer": print("Authorization header format is incorrect") return flask.make_response(flask.jsonify({"message": "Unauthorized"}), 401) bearer_token = auth_header.split()[1] try: token = client.verify_id_token( bearer_token, PROJECT_NUMBER, cert_uri=PUBLIC_CERT_URL_PREFIX + CHAT_ISSUER) if token['iss'] != CHAT_ISSUER: print("Invalid issuee") return flask.make_response(flask.jsonify({"message": "Unauthorized"}), 401) except: print("Invalid token") return flask.make_response(flask.jsonify({"message": "Unauthorized"}), 401) """ 受け取ったメッセージでVertex AI Searchに検索をかける 検索結果を整形し送信する """ request_json = req.get_json(silent=True) text = request_json["message"]["text"] response = create_message(text=text) return response
堂原 竜希(記事一覧)
クラウドソリューション部データアナリティクス課。2023年4月より、G-genにジョイン。
Google Cloud Partner Top Engineer 2023, 2024に選出 (2024年はRookie of the yearにも選出)。休みの日はだいたいゲームをしているか、時々自転車で遠出をしています。
Follow @ryu_dohara