AIチャットボットを LINE WORKS で作成する

はじめに

Vertex AI Agent Builder を使ったナレッジ検索AIチャットbot を LINE WORKS で作成します。

当記事では LINE WORKS との繋ぎ込み部分を中心に紹介していければと思います。LINE WORKS はビジネス版の LINE サービスで直感的な操作で扱えるチャットサービスとなっています！

システム構成図

Agent Builder
Cloud Run 関数
Google Cloud Storage
Secret Manager
Cloud Scheduler
LINE WORKS API

LINE WORKS API を利用するためにはアクセストークンが必要となるため Secret Manager で保持、Cloud Scheduler を使って定期更新を行います。

LINE WORKS の API と BOT を登録する

Developers ConsoleからAPIとBotの登録を行います。

APIの設定

API、ClientAppからアプリの新規追加を行います。

生成された Client ID と Client Secret は後で使うので保存しておきます。

OAuth Scope の管理から「bot」権限を選択します。

API の設定を保存して完了します。

Botの設定

Bot タブの登録から Bot 情報の登録を行います。

以下の項目を入力します。

プロフィール画像（任意）
Bot名
説明
管理者

Callback URL は後から設定します。

Bot、APIの基本設定は以上となります。

Agent Builder の用意

データストアの作成

Agent Builder からデータストアを作成します。今回はAIに生成してもらった架空の社内規定を Cloud Storage に保存して参照データとします。

Agent Builder アプリの作成

データストアの作成から Cloud Storage を選択してバケットを選択します。データストアが作成できたら Agent Builder アプリを作成します。

アプリの作成から検索を選択して、参照したいデータストアを選びます。

作成が完了したらプレビューで確認します。（反映には少し時間がかかります）

Vertex AI Studio でプロンプトを作成

Vertex AI Studio から新しいプロンプトを作成して、データストアのパスを入力します。

作成されたプロンプトからコードを取得します。

Cloud Run 関数で Bot アプリの作成

Tokenについて

LINE WORKS へ接続するためには Access Token が必要です。Access Token の期限は 24時間で切れてしまうため、Refresh Token を使用して再発行する必要があります。

Access Token
- 有効期限： 24 時間
Refresh Token
- 有効期限： 90 日

今回は Cloud Scheduler を使用して Token の更新を行います。

メイン処理関数

LINE WORKS からの応答と回答生成を行うメイン関数を作成します。回答生成部分は Vertex AI Studio から生成したコードを使用しています。

環境変数は LINE WORKS から取得した値を設定します。Token 情報はそれぞれの環境に合わせて管理してください。

参考ドキュメント

  import functions_framework
  import base64
  import hashlib
  import hmac
  import os
  import requests
  import logging
  import google.cloud.logging
  import google.auth
  import google.auth.transport.requests
  
  
  from flask import abort
  
  # Vertex AI Studio
  import vertexai
  from vertexai.generative_models import GenerativeModel, Tool
  import vertexai.preview.generative_models as generative_models
  
  BOT_SECRET = os.getenv("LINE_BOT_SECRET")
  PROJECT_NUMBER = os.getenv("PROJECT_NUMBER")
  PROJECT_ID = os.getenv("PROJECT_NUMBER")
  CLIENT_ID = os.getenv("CLIENT_ID")
  CLIENT_SECRET = os.getenv("CLIENT_SECRET")
  PRIVATE_KEY = os.getenv("PRIVATE_KEY")
  PUBLIC_KEY = os.getenv("PUBLIC_KEY")
  BOT_ID = os.getenv("BOT_ID")
  DATASTORE_ID = os.getenv("DATASTORE_ID")
  TOKEN = ""
  
  
  # Loggerの設定
  logging.basicConfig(
      format="[%(asctime)s][%(levelname)s] %(message)s", level=logging.DEBUG
  )
  logger = logging.getLogger()
  
  logging_client = google.cloud.logging.Client()
  logging_client.setup_logging()
  
  logger.setLevel(logging.DEBUG)
  
  
  @functions_framework.http
  def handle_request(request):
      # 署名検証
      is_valid = validate_request(request)
  
      # Line Works からのrequestを取得
      data = request.get_json()
      text = data["content"]["text"]
  
      # webリソースRAG
      output_text = generate(text)
  
      post_message(request, output_text)
      return "success", 200
  
  
  # 署名検証
  def validate_request(request):
      headers_signature = request.headers["X-WORKS-Signature"]
      body = request.data.decode("utf-8")
  
      hash = hmac.new(
          BOT_SECRET.encode("utf-8"), body.encode("utf-8"), hashlib.sha256
      ).digest()
      signature = base64.b64encode(hash).decode("utf-8")
  
      if not hmac.compare_digest(signature, headers_signature):
          abort(403, "Invalid signature.")
  
  
  # Vertex AI Studio
  def generate(text):
      vertexai.init(project=PROJECT_ID, location="asia-northeast1")
      tools = [
          Tool.from_retrieval(
              retrieval=generative_models.grounding.Retrieval(
                  source=generative_models.grounding.VertexAISearch(
                      datastore=f"projects/{PROJECT_ID}/locations/global/collections/default_collection/dataStores/{DATASTORE_ID}"
                  ),
                  disable_attribution=False,
              )
          ),
      ]
      model = GenerativeModel(
          "gemini-1.5-flash-001",
          tools=tools,
          system_instruction=[
              """提供された情報がない場合は「その質問には答えられません。」と返してください"""
          ],
      )
      responses = model.generate_content(
          [f"""{text}"""],
          generation_config=generation_config,
          safety_settings=safety_settings,
      )
  
      # 出力テキスト
      output_text = responses.candidates[0].content.parts[0].text
  
      return output_text
  
  
  generation_config = {
      "max_output_tokens": 8192,
      "temperature": 0,
      "top_p": 0.21,
  }
  
  safety_settings = {
      generative_models.HarmCategory.HARM_CATEGORY_HATE_SPEECH: generative_models.HarmBlockThreshold.BLOCK_ONLY_HIGH,
      generative_models.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: generative_models.HarmBlockThreshold.BLOCK_ONLY_HIGH,
      generative_models.HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: generative_models.HarmBlockThreshold.BLOCK_ONLY_HIGH,
      generative_models.HarmCategory.HARM_CATEGORY_HARASSMENT: generative_models.HarmBlockThreshold.BLOCK_ONLY_HIGH,
  }
  
  
  # メッセージ送信
  def post_message(request, output_text):
      data = request.get_json()
  
      userId = data["source"]["userId"]
  
      # APIの設定
      api_url = f"https://www.worksapis.com/v1.0/bots/{BOT_ID}/users/{userId}/messages"
      headers = {"Content-Type": "application/json", "Authorization": f"Bearer {TOKEN}"}
  
      # メッセージデータ
      data = {"content": {"type": "text", "text": output_text}}
  
      # リクエストを送信
      response = requests.post(api_url, json=data, headers=headers)
  
      # レスポンスを確認
      print(response.status_code)

Python で Token 発行を行うコード例

Access Token と Refresh Token の発行

import datetime
import os
import time

import jwt
import requests

CLIENT_ID = os.environ["CLIENT_ID"]
CLIENT_SECRET = os.environ["CLIENT_SECRET"]
PRIVATE_KEY = os.environ["PRIVATE_KEY"]
BOT_ID = os.environ["BOT_ID"]
SERVICE_ACCOUNT_ID = os.environ["SERVICE_ACCOUNT_ID"]

# JWTを生成
payload = {
    "iss": CLIENT_ID,
    "sub": SERVICE_ACCOUNT_ID,
    "iat": int(time.time()),
    "exp": int(time.time()) + 3600,
}

encoded_jwt = jwt.encode(payload, PRIVATE_KEY, algorithm="RS256")

api_url = "https://auth.worksmobile.com/oauth2/v2.0/token"
headers = {
    "Content-Type": "application/x-www-form-urlencoded",
}

data = {
    "assertion": encoded_jwt,
    "grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
    "client_id": CLIENT_ID,
    "client_secret": CLIENT_SECRET,
    "scope": "bot",
}

# アクセストークンを発行
response = requests.post(api_url, data=data, headers=headers)

if response.status_code == 200:
    data = response.json()
    refresh_token = data.get("refresh_token", None)
    access_token = data.get("access_token", None)
    if refresh_token:
        print("Access Token:", access_token)
        print("Refresh Token:", refresh_token)
    else:
        print("Access token not found in the response.")

else:
    print(response.json())
    refresh_token = False
    access_token = False

Refresh Token を使用して Access Token を再発行

import os
import requests

CLIENT_ID = os.environ["CLIENT_ID"]
CLIENT_SECRET = os.environ["CLIENT_SECRET"]
PRIVATE_KEY = os.environ["PRIVATE_KEY"]
BOT_ID = os.environ["BOT_ID"]
SERVICE_ACCOUNT_ID = os.environ["SERVICE_ACCOUNT_ID"]
api_url = "https://auth.worksmobile.com/oauth2/v2.0/token"
headers = {
    "Content-Type": "application/x-www-form-urlencoded",
}

# 保持してあるリフレッシュトークン
refresh_token = ""

data = {
    "refresh_token": refresh_token,
    "grant_type": "refresh_token",
    "client_id": CLIENT_ID,
    "client_secret": CLIENT_SECRET,
}

# アクセストークンを発行
response = requests.post(api_url, data=data, headers=headers)

if response.status_code == 200:
    data = response.json()
    access_token = data.get("access_token", None)
    if access_token:
        print("Access Token:", access_token)
    else:
        print("Access token not found in the response.")
else:
    print("Failed to fetch data:", response.json())
    access_token = False

LINE WORKS へ最終連携

Cloud Run 関数が作成できたら LINE WORKS の Developers Console で Callback URL を追加します。

Callback URL には Cloud Run 関数のURLを追加してください。

BOT の追加

Admin ページからBotの追加を行います。サービス設定 → Bot タブからBot追加を選択します。

Bot が正しく登録されている場合、追加ページで選択することができます。

追加後のデフォルト状態では非公開となってるため、必要に応じて公開設定の変更を行ってください。

公開する範囲をメンバー指定で制限することも可能です。

実際に BOT を利用する

新規作成 → 社内メンバーとトーク → Bot から選択します。

選択した Bot にメッセージを送信すると回答が返されます。

おわりに

LINE WORKS で Vertex AI を利用したナレッジ検索 BOT を作成することができました。

Slack などと比較するとスレッド管理が行えない分少し自由度は下がりますが、LINE WORKS でも十分にナレッジ検索ツールとして機能します！

AIチャットボットを LINE WORKS で作成する

はじめに

システム構成図

LINE WORKS の API と BOT を登録する

APIの設定

Botの設定

Agent Builder の用意

データストアの作成

Agent Builder アプリの作成

Vertex AI Studio でプロンプトを作成

Cloud Run 関数で Bot アプリの作成

Tokenについて

メイン処理関数

Python で Token 発行を行うコード例

LINE WORKS へ最終連携

BOT の追加

実際に BOT を利用する

おわりに

[アップデート]複数VPCのインターネット通信を制御する機能「VPC Block Public Access (BPA)」がリリースされました！

Python で Cloud Storage の JSON ファイルを読み込む

AWSの生成AI活用事例集GenUを使い倒す

WindowsServer上に作成したローカルユーザーのパスワード有効期限管理パターンとその適用手順をまとめてみる

Oracle のロックされているテーブルのセッションを知りたい [cloudpack OSAKA blog]

AIチャットボットを LINE WORKS で作成する

はじめに

システム構成図

LINE WORKS の API と BOT を登録する

APIの設定

Botの設定

Agent Builder の用意

データストアの作成

Agent Builder アプリの作成

Vertex AI Studio でプロンプトを作成

Cloud Run 関数で Bot アプリの作成

Tokenについて

メイン処理関数

Python で Token 発行を行うコード例

LINE WORKS へ最終連携

BOT の追加

実際に BOT を利用する

おわりに

関連記事Related Articles

PaLMが日本語応答に対応！PaLMのAPIをJupyter Notebookから実行するハンズオン！

Vertex AIについて調べてみた

Generative AI Summit Osaka 実用的な生成AI

【Generative AI Summit Osaka】生成AI無学者が生成AIイベントに行ってきた

【Google Cloud Next ’24】AI活用のサステイナビリティの活動について