はじめに

筆者について

メーカー勤務3年目のソフトウェアエンジニア
大学では材料系の専攻であり、ソフトウェアエンジニアとしての歴も約3年

対象者

Webhookとは何か、実装しながら学びたい方
とにかく生成AIを使って面白いものを作ってみたい方

動機

大学の先輩からLINEのWebhookについて伺ったのでなんかやってみたい
生成AIになじみのない両親や友人が、気軽に生成AIを試せる環境を作りたい
ついでにクラウドを使える機会があればいいな

範囲

ローカル環境で機能の作成、検証　←①の範囲
クラウド環境にデプロイ　　　　　←②（次回）の範囲

前提知識

Webhookってなに

以下のサイトが分かりやすかったので拝借。

Webhookは、特定のイベントが発生した際に、事前に設定された他のアプリケーションに通知を自動的に送信する仕組みで「Webコールバック」や「HTTPプッシュAPI」と呼ばれることもあります。

送り手のサービスでデータの追加などの「きっかけ」が起こる

「きっかけ」により、指定した「宛先※」へのHTTP通信が発生 ※ここでいう「宛先」とはURLのことです

指定した「宛先」へデータを送ることができる

要するにAPIとは逆方向の概念になります。APIが「クライアントが何かアクションをしてサーバにリクエストする」のに対して、Webhookは「サーバに何かアクションがあれば、クライアントに対してリクエストする」ものになります。

開発環境の準備

Webhookの概念がなんとなく分かったところで、開発環境の準備を進めていきましょう。

LINEのChannelの準備

少し古くなりますが以下の記事を参考にさせて頂きました。

アクセストークンを発行、シークレットキーも控えておきましょう。

GCP準備

GoogleアカウントとGCPプロジェクト

アカウントをお持ちでない方は以下のサイトを参考に準備して頂ければと思います。

Google Cloud SDK のインストール

ローカル環境でVertexAIを利用するためにインストールします

処理フロー概要

ユーザーがLINE上でメッセージを送信
ボット名前を「ちいくん」として、「ちいくん」と呼びかけられたときに返答があるようにしました。たとえば「ちいくん、大阪で有名な食べ物は？」など
LINEサーバーがWebhookを叩く (POST /callback)
Flaskアプリケーション上の /callback エンドポイントへメッセージ情報を送信
Flaskアプリケーション (linebot SDK) が Vertex AI (Gemini) を呼び出し
google.cloud.aiplatform などのライブラリ (vertexai) を使い、Geminiモデルへ質問文を送信
→今回はここをローカル環境で動かすところまで記載します。
Geminiモデルから応答を受け取る
たとえば「大阪の有名な食べ物はお好み焼きやたこ焼きです…」などのテキスト
Flaskが linebot SDK を通じて返信メッセージを作成 → LINEサーバーへ送信
LINEサーバーがユーザーのトーク画面にメッセージを表示
ユーザーは「ちいくん」からの回答を受け取る

ローカル環境で実装

クラウド環境でアプリケーションを構築する前に、ローカル環境で一通り動かせるようにしました。

LINEからのWebhookを受信する

まずはコードを見て頂きます。

Webhook.py

import os
from flask import Flask, request, abort
from linebot import LineBotApi, WebhookHandler
from linebot.exceptions import InvalidSignatureError
from linebot.models import MessageEvent, TextMessage, TextSendMessage
from dotenv import load_dotenv
import sys

load_dotenv()

app = Flask(__name__)

google_api_ins = google_api()

# LINEのアクセストークンとチャネルシークレット
CHANNEL_ACCESS_TOKEN = os.getenv("LINE_CHANNEL_ACCESS_TOKEN")
CHANNEL_SECRET = os.getenv("LINE_CHANNEL_SECRET")

line_bot_api = LineBotApi(CHANNEL_ACCESS_TOKEN)
handler = WebhookHandler(CHANNEL_SECRET)


@app.route("/callback", methods=['POST'])
def callback():
    # X-Line-Signatureヘッダーに署名が含まれる
    signature = request.headers['X-Line-Signature']

    # request.get_data(as_text=True)でリクエストのボディを文字列として取得
    body = request.get_data(as_text=True)

    try:
        # WebhookHandlerで署名を検証しつつ、イベントをパース
        handler.handle(body, signature)
    except InvalidSignatureError:
        abort(400)

    return 'OK'


# メッセージ受信イベントを処理するハンドラ
@handler.add(MessageEvent, message=TextMessage)
def handle_message(event):
    # ユーザーが送ったメッセージ
    user_message = event.message.text
    if user_message.startswith("ちいくん"):
        # 返信API呼び出し
        line_bot_api.reply_message(event.reply_token, user_message)
    else:
        pass


if __name__ == "__main__":
    app.run()

.env

# アクセストークンとチャネルシークレットを=の右に記載すること
LINE_CHANNEL_ACCESS_TOKEN=
LINE_CHANNEL_SECRET=

解説

FlaskでWebサーバを構築
Flask を使って Web サーバーを構築しています。host やポートの指定は特に行っていないため、デフォルトの 5000 番ポートが使用されます。したがって、 http://localhost:5000 へリクエストを送信すると、このサーバーにアクセスできます。
.envファイルでアクセストークンとチャネルシークレットを管理
アクセストークンやチャネルシークレットなどの機密情報は、コードに直書きしないように .env ファイルで管理しています。Python では、ライブラリを利用して .env ファイルの値を読み込むことができるため、安全かつ簡単に変数を扱えます。
linebotライブラリの活用
HTTP 通信を独自に処理する方法もありますが、ここでは手間を省くため、公式の line-bot-sdk（linebot ライブラリ）を活用しました。これによって、Webhook および API 呼び出しの処理をまとめて実装できます。

Webhookのテスト

ngrokと呼ばれるツールを使うことで、ローカル環境で動いているアプリケーションを外部に公開することができます。今回はLINEのWebhookを受信したいので、テストのための受け口になってもらいます。

python Webhook.py （または flask run）でローカル起動
ngrokを使って外部に公開
→https://xxxxxxxx.ngrok.io のようなURLが得られる

ngrok http 5000

3.LINE Developersコンソールの「Messaging API設定」→「Webhook URL」に

https://xxxxxxxx.ngrok.io/callback

を設定
4. 「Webhook送信利用」をオン→「接続テスト」ボタンで成功するか確認
5. LINEアプリでメッセージを送ってみて、オウム返しされれば成功

以下のようなサイトも参考にさせて頂きました。

VertexAIのAPIを利用する

google_api.py

import os
from google.cloud import secretmanager
import vertexai
from vertexai.generative_models import GenerativeModel, SafetySetting

class google_api:

    def __init__(self):
        self.project_id = "PROJECTID"
        self.location = "LOCATION"
    
    def _generate_message(self, message):
        """
        JSON形式での回答を要求するプロンプトを生成する
        """
        return f"""
        あなたは「ちいくん」という名前の賢くてかわいい人間です。次の質問に答えてください。
        **質問**
        {message}
        """

    def generate(self, message):

        vertexai.init(project=self.project_id, location=self.location)
        model = GenerativeModel("gemini-1.5-pro")
        prompt = self._generate_message(message)

        contents = [
            {
                "role": "user",
                "parts": [
                    {
                        "text": prompt
                    }
                ]
            }
        ]

        generation_config = {
            "max_output_tokens": 8192,
            "temperature": 0,
            "top_p": 0.95,
        }

        safety_settings = [
            SafetySetting(
                category=SafetySetting.HarmCategory.HARM_CATEGORY_HATE_SPEECH,
                threshold=SafetySetting.HarmBlockThreshold.OFF
            ),
            SafetySetting(
                category=SafetySetting.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT,
                threshold=SafetySetting.HarmBlockThreshold.OFF
            ),
            SafetySetting(
                category=SafetySetting.HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT,
                threshold=SafetySetting.HarmBlockThreshold.OFF
            ),
            SafetySetting(
                category=SafetySetting.HarmCategory.HARM_CATEGORY_HARASSMENT,
                threshold=SafetySetting.HarmBlockThreshold.OFF
            ),
        ]

        response = model.generate_content(
            contents=contents,
            generation_config = generation_config,
            safety_settings = safety_settings,
        )
        return response.text

if __name__ == "__main__":
    google_api_ins = google_api()
    print(google_api_ins.generate("ちいくん、大阪で有名な食べ物は"))

解説

_generate_message():
ここでシステムプロンプトにあたるような条件文を記載しています。
model.generate_content()
引数の詳細は以下のページに記載されているので参考にしました。必要な情報を入力します。

続きはこちらから