Amazon BedrockとSlackで生成AIチャットボットアプリを作る (その2：Lambda＋API Gatewayで動かす)

みなさん、こんにちは！
福岡オフィスの青柳です。

前回のブログに引き続き、Amazon BedrockとSlackを使って「生成AIチャットボットアプリ」を作成します。

前回はSlack APIの「Socket Mode」を使うことで、ローカルPC上でSlackアプリを動かすことができました。

今回は、より「本番っぽい」アプリにするため、Socket Modeを使わずに「HTTP APIサーバー」を立ててSlackアプリを動かしてみます。

※ このブログから読み始めても問題ありませんが、前回のブログを先に読むと、より理解が深まると思います。 以下のリンクから是非どうぞ！

今回作成するチャットボットのアーキテクチャ

前回のブログで、「Slack Boltを使ったSlackアプリはHTTP APIサーバーとして実行され、SlackからのHTTPリクエストを受信して処理を行う」と説明しました。

Slackアプリを動作させる「HTTP APIサーバー」を実装する際、いくつかの方法が考えられます。

EC2インスタンスやECSコンテナサービスを使ってHTTP APIサーバーを構築する方法もその一つです。

ただ、もともと自分の検証用にチャットボットを作ろうと考えましたので、EC2やECSだとコストが掛かりそうなのが心配です。 (検証で使う時に毎回起動・停止するのも面倒ですし)

そこで今回は、LambdaとAPI Gatewayを組み合わせて「HTTP APIサーバー (HTTP APIサービス)」を構築したいと思います。

Step 1: 最初のシンプルなSlackアプリを作成する

前回と同様に、Amazon Bedrockと連係するアプリを作成する前に、まずはシンプルな動作をするアプリを作成することにします。

ユーザーがSlackアプリをメンションして何か書き込みを行った際、書き込まれた内容をそのまま返す「オウム返し」アプリを作成したいと思います。

手順(1) Slack側の設定

まず、Slack側の設定を行います。

設定内容は前回のブログ (Socket Modeを使うパターン) と似ていますが、異なる箇所がありますので気をつけてください。

「Slack App」の作成

ここで作成する「Slack App」とは、Slackアプリの実体ではなく、「アプリがSlackと連係するためのトークンの払い出し」や「Slackがアプリに対して許可する権限の設定」などを管理するための概念です。

Slackへサインインした状態で https://api.slack.com/apps/ へアクセスします。

「Create New App」をクリックします。

作成方法として、ここでは「From scratch」を選択します。

Slack Appの「名前」の指定と、対象となるSlackの「ワークスペース」の選択を行います。

「Create App」をクリックしてSlack Appを作成します。

「Bot Token Scopes」の設定

Slackアプリが動作する際に必要な権限を与えます。

今回は、アプリからSlackに対して返信の書き込みを行えるように、「チャットへの書き込み」の権限を与えます。

左側のメニューから「OAuth & Permissions」を選択します。

下の方にスクロールして「Scopes」の設定項目を表示します。

「Bot Token Scopes」が、Slackからアプリに対して許可を与える権限の範囲を示します。

「Add an OAuth Scope」をクリックします。

スコープのリストから「chat:write」を選択します。

「chat:write」のスコープが追加されたことを確認します。

ワークスペースへのAppのインストール

対象のワークスペースへAppの「インストール」を行い、ワークスペースでSlackアプリが使用できるようにします。 (アプリの実体を作成していないので、まだ動作はしません)

左側のメニューから「Install App」を選択します。

「Install to Workspace」をクリックします。

対象のワークスペース名、与える権限を確認して、「許可する」をクリックします。

Appをインストールしたことによって、Slackアプリに組み込むための「トークン」が発行されました。 このトークンは後ほど必要になりますが、この画面は後からでも確認できますので、ここでは何もしなくて構いません。

これでSlack側の設定は一旦終わりです。

手順(2) Lambda関数の作成

Slackアプリの実体となるLambda関数を作成していきます。

Lambdaレイヤーの作成

LambdaのPythonランタイム環境には「Slack Bolt」のモジュールが標準では含まれていないため、Lambdaレイヤーを使ってモジュールを利用可能にします。

まず、「Python 3.12」(Lambda関数で利用予定のPythonランタイムと同じバージョン) を導入済みの環境 (Amazon Linux 2023ベースのEC2インスタンス or Cloud9が好ましい) で、以下のコマンドラインを実行します。

$ mkdir python
$ pip install -t ./python slack-bolt
$ zip -r slack-bolt.zip ./python

次に、Lambdaマネジメントコンソールの「レイヤー」から「レイヤーの作成」をクリックして、以下のように設定します。

Lambdaレイヤーの作成手順については、下記ブログ記事も併せて参照してください。

Lambda関数の作成

以下の設定でLambda関数を作成します。

Lambda関数の設定： Lambdaレイヤー

「関数の概要」→「レイヤー」で、「Lambdaレイヤーの作成」で作成したレイヤーを追加します。

Lambda関数の設定： 環境変数

「設定」→「環境変数」で、以下の環境変数を設定します。

「Bot User OAuth Token」は、左側のメニューから「OAuth & Permissions」を選択して、「OAuth Tokens for Your Workspace」で確認できます。

「Signing Secret」は、左側のメニューから「Basic Information」を選択して、「App Credentials」で確認できます。

Lambda関数のコード記述

Lambda関数のコードを以下のように記述します。

from slack_bolt import App
from slack_bolt.adapter.aws_lambda import SlackRequestHandler
import os
import re

# アプリの初期化
app = App(
    token=os.environ.get("SLACK_BOT_TOKEN"),
    signing_secret=os.environ.get("SLACK_SIGNING_SECRET"),
    process_before_response=True,
)

# Slackイベントハンドラー：Slackアプリがメンションされた時
@app.event("app_mention")
def handle_app_mention_events(event, say):
    input_text = re.sub("<@.+>", "", event["text"]).strip()
    say(input_text)

# Lambdaイベントハンドラー
def lambda_handler(event, context):
    slack_handler = SlackRequestHandler(app=app)
    return slack_handler.handle(event, context)

前回のブログと異なる部分について解説します。

from slack_bolt import App
from slack_bolt.adapter.aws_lambda import SlackRequestHandler

前回は、アプリをSocket Modeに対応させるためにslack_bolt.adapter.socket_modeのSocketModeHandlerモジュールをインポートしていました。

今回は、Lambda上でアプリを動かすためにslack_bolt.adapter.aws_lambdaのSlackRequestHandlerモジュールをインポートします。

# アプリの初期化
app = App(
    token=os.environ.get("SLACK_BOT_TOKEN"),
    signing_secret=os.environ.get("SLACK_SIGNING_SECRET"),
    process_before_response=True,
)

アプリの初期化の際に、「Bot User OAuth Token」トークンに加えて「Signing Secret」も与える必要があります。

また、LambdaなどのいわゆるFaaS (Function as a Service) 環境でアプリを動かす時の必須条件として、パラメーターprocess_before_responseは「True」に設定する必要があります。

# Lambdaイベントハンドラー
def lambda_handler(event, context):
    slack_handler = SlackRequestHandler(app=app)
    return slack_handler.handle(event, context)

Lambdaのイベントハンドラーが呼び出された際に、Slack Boltのイベントハンドラーが呼び出されるように記述します。

Slack Boltイベントハンドラーのレスポンスは、Lambdaイベントハンドラーの戻り値としてSlackに返されます。

手順(3) API Gatewayの作成

Lambdaを「HTTP APIサービス」として公開するために、API Gatewayを作成します。

APIタイプは「HTTP API」を選択します。

「統合」の対象として「Lambda」を選択して、以下のように設定します。

「リソースパス」の設定を以下のように変更します。

リソースパス以外の設定はデフォルトのままにします。

ステージに関する設定はデフォルトのままにします。

最後に設定内容のサマリーが表示されますので、確認して「作成」をクリックします。

API Gatewayが作成できましたら、APIのエンドポイントを確認します。

左側のメニューから、作成したAPIの名前を選択します。

「APIの詳細」にAPIのエンドポイントが表示されています。

この次の手順で使いますので、控えておいてください。

手順(4) 「Event Subscriptions」の設定

ここからは、再度Slack側の設定画面に戻って設定を進めます。

特定のイベントが発生した時にSlackアプリが呼び出されるように、イベントの購読 (サブスクリプション) を設定します。

左側のメニューから「Event Subscriptions」を選択します。

「Eneble Events」を「Off」から「On」に変更します。

「Socket Mode」を使う場合には無かった設定として「Request URL」を指定する必要があります。

これは、SlackからHTTP APIサーバーに対してリクエストを送信する時の宛先URLとなります。

前の手順で確認した「API Gatewayのエンドポイント」を使って、以下のようにURLを組み立てて入力します

• Request URL: ＜API Gatewayのエンドポイント＞/slack/events

(入力例： https://XXXXXXXXXX.execute-api.us-east-1.amazonaws.com/slack/events)

入力したURLに対してSlackからの検証が行われ、検証に成功すると「Verified」と表示されます。

(もし検証結果がエラーになる場合は、ここまでの手順を確認してください)

なお、ここで行われている「Slackからの検証」の詳細を知りたい方は、以下のブログ記事を参照してみてください。

(ただし、Slack Boltを使用している場合は、ここに書いてある処理をSlack Boltが自動的に行ってくれるため、意識する必要はありません)

「Subscribe to bot events」をクリックして展開します。

「Add Bot User Event」をクリックします。

イベントのリストから「app_mention」を選択します。

「app_mention」が追加されたことを確認して、「Save Changes」で設定を保存します。

手順(5) Slack Appの再インストール

「Event Subscriptions」の設定を保存すると、画面の上部に下図のような黄色いバーが表示されると思います。

これは、イベント購読の追加によってアプリに対するスコープ (具体的にはapp_mentions:read権限) が追加されたために「Slack Appの再インストールが必要」であることを促すものです。

「reinstall your app」の部分がリンクになっているため、クリックします。

権限の確認画面が表示されますが、最初にインストールした時には無かった「アプリが参加している会話で @〜 を直接メンションしているメッセージを表示」が追加されていることに注目してください。

「許可する」をクリックします。 これで、新しい権限でSlackアプリが動作できるようになりました。

このように、スコープ (権限) の追加を行った際はSlack Appの「再インストール」が必要になることを覚えておいてください。

なお、この「Slack Appの再インストール」の操作は、左側メニューの「Install App」から行うことも可能です。 (どちらの手順で行っても問題ありません)

手順(6) ワークスペースへアプリを追加

これでSlackアプリの準備ができました。

Slackの画面に移動して、作成したSlackアプリをチャンネルに「追加」します。

アプリをインストールしたいチャンネルで、チャンネルの情報画面を開きます。

「インテグレーション」タブを選択して、「アプリを追加する」をクリックします。

作成したSlackアプリが表示されていると思いますので、「追加」をクリックします。

動作を確認する

これで、チャンネル内でアプリを呼び出すことができます。

アプリを「@アプリ名」でメンションして、続けて何か書き込んでみましょう。

こちらが書き込んだ内容をそのまま返す「オウム返し」チャットボットが出来ました！

Step 2: Bedrockと連係して質問に答えるようにする

アプリのシンプルな動作を確認できましたので、いよいよ本命である「Amazon Bedrockとの連係」を組み込んでいきます。

手順(1) Lambda関数コードの修正

まずは、最初の部分を以下のように修正します。

from slack_bolt import App
from slack_bolt.adapter.aws_lambda import SlackRequestHandler
import boto3
import json
import os
import re

bedrock_runtime = boto3.client("bedrock-runtime", region_name="us-east-1")

モジュールboto3jsonのインポートと、boto3の「bedrock-runtime」クライアントの初期化を追記します。

次に、Amazon Bedrockを呼び出すサブ処理を記述します。 今回は「Claude 3 Haiku」を使用していますが、他のモデルでもOKです。

# Bedrockを使って応答テキストを生成する
def generate_answer(input_text):
    # メッセージをセット
    messages = [
        {
            "role": "user",
            "content": input_text,
        },
    ]

    # リクエストBODYをセット
    request_body = json.dumps({
        "messages": messages,
        "anthropic_version": "bedrock-2023-05-31",
        "max_tokens": 1000,
    })

    # Bedrock APIを呼び出す
    response = bedrock_runtime.invoke_model(
        modelId="anthropic.claude-3-haiku-20240307-v1:0",
        accept="application/json",
        contentType="application/json",
        body=request_body,
    )

    # レスポンスBODYから応答テキストを取り出す
    response_body = json.loads(response.get("body").read())
    output_text = response_body.get("content")[0].get("text")

    # 応答テキストを戻り値として返す
    return output_text

最後に、Slackイベントハンドラーの内容を修正します。

# Slackイベントハンドラー：Slackアプリがメンションされた時
@app.event("app_mention")
def handle_app_mention_events(event, say):
    input_text = re.sub("<@.+>", "", event["text"]).strip()
    output_text = generate_answer(input_text)
    say(output_text)

書き込まれたテキスト内容を「Amazon Bedrock呼び出しサブ処理」に与えてコールします。

得られた応答テキストをsay()に与えてSlackへ書き込みます。

手順(2) Lambdaの設定変更

Bedrockとの連係を組み込んだことに伴い、いくつか設定の変更を行います。

IAMロールにアクセス権限を追加

Lambda関数がBedrockのモデルを呼び出すことができるように、アクセス権限を付与します。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "bedrock:InvokeModel",
            "Resource": "*"
        }
    ]
}

Lambda関数のタイムアウト時間の変更

Lambda関数がBedrockを呼び出している間にタイムアウトしてしまわないように、タイムアウト時間を「3秒」→「30秒」に変更しておきます。

動作確認

それでは、チャットボットに対して何か質問をしてみましょう。

書き込んだ内容に対して、ちゃんと回答が返ってきましたね。

Step 3: Bedrock呼び出し処理を非同期で行うようにする

これで一見「完成！」したように思えますが、何度かチャットボットに質問してみると、動作に不具合があることに気付きました。

1回質問すると、最初に回答が返ってきてから、少し時間を置いて同じ質問にまた回答が返ってくるのです。 これが2〜3回繰り返されるのです。

さて、何故だろう？ と悩んでいると、こんなブログ記事を見つけました。

ここでとても重要なのは、http_timeoutは3秒経つとサーバーからのレスポンスの有無に関わらず即座に再送してくるので、3秒以内に2xxを返してあげないと重複して処理が実行されてしまいます。

なるほど！

Lambdaイベントハンドラーが呼び出されて、そこからBedrockを呼び出す処理で時間がかかってしまうと、Slackは「3秒経ったが応答が無い、リクエストを再送しよう」となって同じリクエストを繰り返し送ってくるという訳ですね。

公式ドキュメントにもちゃんと書いてました。

Events API | Slack
https://api.slack.com/apis/connections/events-api

原因が分かりましたので、アーキテクチャを改良してみます。

SQSキューを用いて、時間が掛かる処理を非同期で行わせる仕組みです。

このアーキテクチャに基づいて、Slackアプリを修正していきます。

手順(1) SQSキューの作成

キューのタイプは「標準」を選択して、その他の設定はとりあえずデフォルトのままにします。

手順(2) Lambda関数コードの修正

以下のようにコードを修正します。

from slack_bolt import App
from slack_bolt.adapter.aws_lambda import SlackRequestHandler
import boto3
import json
import os
import re


sqs = boto3.client("sqs")

sqs_queue_url = "https://sqs.us-east-1.amazonaws.com/123456789012/bedrock-backend-queue"

# アプリの初期化
app = App(
    token=os.environ.get("SLACK_BOT_TOKEN"),
    signing_secret=os.environ.get("SLACK_SIGNING_SECRET"),
    process_before_response=True,
)


# Slackイベントハンドラー：Slackアプリがメンションされた時
@app.event("app_mention")
def handle_app_mention_events(event, say):
    result = say(text=f"少々お待ちください...")

    channel_id = event["channel"]
    input_text = re.sub("<@.+>", "", event["text"]).strip()

    sqs.send_message(
        QueueUrl=sqs_queue_url,
        MessageBody=json.dumps({
            "channel_id": channel_id,
            "input_text": input_text,
        }),
    )


# Lambdaイベントハンドラー
def lambda_handler(event, context):
    slack_handler = SlackRequestHandler(app=app)
    return slack_handler.handle(event, context)

「bedrock-runtime」クライアントの初期化の記述と、Bedcorkを呼び出すサブ処理を全て削除します。

代わりに、SQSクライアントの初期化の記述を追記します。(9行目)
また、SQSの「キューURL」を定数宣言しておきます。(11行目)

Slackイベントハンドラーの内容を、いくつか変更します。

# Slackイベントハンドラー：Slackアプリがメンションされた時
@app.event("app_mention")
def handle_app_mention_events(event, say):
    result = say("少々お待ちください...")

    channel_id = event["channel"]
    input_text = re.sub("<@.+>", "", event["text"]).strip()

    sqs.send_message(
        QueueUrl=sqs_queue_url,
        MessageBody=json.dumps({
            "channel_id": channel_id,
            "input_text": input_text,
        }),
    )

まず、Bedrockが回答を生成するまで少し時間が掛かりますので、最初に「少々お待ちください」と応答しておくことにします。

次に、現在ユーザーとチャットボットの間で会話が行われている「チャンネル」のIDを取得します。

なぜチャンネルのIDが必要かと言うと、Slackから直接呼び出されるSlackイベントハンドラーとは異なり、SQSを介して実行される「Bedrockバックエンド処理」のLambda関数では「会話が行われているチャンネル」を知る術が無く、外部から教えてあげる必要があるためです。

最後に、SQSキューに「チャンネルID」と「書き込まれたテキスト内容」をセットしてメッセージを送信します。

Lmabdaの設定変更： IAMロールのアクセス権限を変更

Bedrockへのアクセス権限は必要なくなりましたので、削除します。

代わりに、「SQSキューに対してメッセージを送信する権限」を追加します。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "sqs:SendMessage",
            "Resource": "arn:aws:sqs:us-east-1:123456789012:bedrock-backend-queue"
        }
    ]
}

Lambdaの設定変更： タイムアウト時間を元に戻す

タイムアウト時間を「30秒」に変更していましたが、このLambda関数はそもそも3秒以内に実行完了することが求められるため、元の「3秒」に戻します。

手順(3) 「バックエンド」Lambdaの作成

以下の設定でLambda関数を作成します。

IAMロールに対して「SQSキューからメッセージを取り出す権限」と「Bedrockのモデルを呼び出す権限」を付与しておきます。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "sqs:GetQueueAttributes",
                "sqs:ReceiveMessage",
                "sqs:DeleteMessage"
            ],
            "Resource": "arn:aws:sqs:us-east-1:123456789012:bedrock-backend-queue"
        },
        {
            "Effect": "Allow",
            "Action": "bedrock:InvokeModel",
            "Resource": "*"
        }
    ]
}

Lambda関数の設定： Lambdaレイヤー

「バックエンド」Lambda関数では、「Slack SDK」を使ってSlackに対する「メッセージ書き込み」などの処理を行います。 そのために、Slack SDKのPythonモジュールを含むLambdaレイヤーが必要になります。

「Step 1」で作成したLambdaレイヤーには、「Slack Bolt」と「Slack SDK」のPythonモジュールが含まれていますので、これをそのまま使うことにします。 (Slack SDKモジュールのみを含むLambdaレイヤーを別途作成しても構いません)

「関数の概要」→「レイヤー」で、Step 1の「Lambdaレイヤーの作成」で作成したレイヤーを追加します。

Lambda関数の設定： 環境変数

Slack SDKが動作するために「Bot User OAuth Token」のトークンが必要になります。 (「Signing Secret」は不要です)

「設定」→「環境変数」で、以下の環境変数を設定します。

Lambda関数のタイムアウト時間の変更

Lambda関数がBedrockを呼び出している間にタイムアウトしてしまわないように、タイムアウト時間を「3秒」→「30秒」に変更しておきます。

Lambda関数のコード記述

Lambda関数のコードを以下のように記述します。

from slack_sdk import WebClient
import boto3
import json
import os


bedrock_runtime = boto3.client("bedrock-runtime", region_name="us-east-1")

client = WebClient(token=os.environ.get("SLACK_BOT_TOKEN"))


# Bedrockを使って応答テキストを生成する
def generate_answer(input_text):
    # メッセージをセット
    messages = [
        {
            "role": "user",
            "content": input_text,
        },
    ]

    # リクエストBODYをセット
    request_body = json.dumps({
        "messages": messages,
        "anthropic_version": "bedrock-2023-05-31",
        "max_tokens": 1000,
    })

    # Bedrock APIを呼び出す
    response = bedrock_runtime.invoke_model(
        modelId="anthropic.claude-3-haiku-20240307-v1:0",
        accept="application/json",
        contentType="application/json",
        body=request_body,
    )

    # レスポンスBODYから応答テキストを取り出す
    response_body = json.loads(response.get("body").read())
    output_text = response_body.get("content")[0].get("text")

    # 応答テキストを戻り値として返す
    return output_text


# Lambdaイベントハンドラー
def lambda_handler(event, context):
    # SQSキューから情報を取り出す
    body = json.loads(event["Records"][0]["body"])
    channel_id = body.get("channel_id")
    input_text = body.get("input_text")

    # Bedrockを呼び出して入力テキストに対する応答テキストを生成する
    output_text = generate_answer(input_text)

    # Slackへ応答テキストを書き込む
    result = client.chat_postMessage(
        channel=channel_id,
        text=output_text,
    )

ポイントとなる箇所について解説します。

from slack_sdk import WebClient
import boto3
import json
import os

bedrock_runtime = boto3.client("bedrock-runtime", region_name="us-east-1")

client = WebClient(token=os.environ.get("SLACK_BOT_TOKEN"))

Slack SDKを使ってSlackへ「メッセージ書き込み」を行うために、slack_sdkのWebClientモジュールをインポートします。

そして、WebClientモジュールを初期化する際に、「Bot User OAuth Token」のトークン文字列を与えます。

# Lambdaイベントハンドラー
def lambda_handler(event, context):
    # SQSキューから情報を取り出す
    body = json.loads(event["Records"][0]["body"])
    channel_id = body.get("channel_id")
    input_text = body.get("input_text")

    # Bedrockを呼び出して入力テキストに対する応答テキストを生成する
    output_text = generate_answer(input_text)

    # Slackへ応答テキストを書き込む
    result = client.chat_postMessage(
        channel=channel_id,
        text=output_text,
    )

Lambdaイベントハンドラーが呼び出された際に、まず、SQSキューのメッセージから「チャンネルID」と「書き込まれたテキスト内容」を取得します。

Bedrockを呼び出すサブ処理で「書き込まれたテキスト内容」に対する「応答テキスト」を得ます。

そして、Slackへ応答テキストを書き込みます。

Slack Boltを使ったアプリでは、Slackにメッセージを書き込む際にsay()を使いました。

Slack SDKでは、chat_postMessage()メソッドを使います。 パラメーターに「チャンネルID」と「書き込むテキスト文字列」を与えます。

手順(4) SQSキューに「Lambdaトリガー」を設定

手順(1)で作成したSQSキューの「Lambdaトリガー」として、手順(3)で作成した「バックエンド」Lambda関数を設定します。

これで、「SQSキューを用いた非同期処理化」が完成しました。

動作確認

では、動作を確認してみましょう。

質問を書き込むと、すぐに「少々お待ちください...」と返信が返ってきます。

しばらく経った後、質問に対する回答が返ってきました。

1回の質問に何度も応答することもなくなりました。

おわりに

Amazon BedrockとSlackで動作する「生成AIチャットボットアプリ」を、AWS LambdaとAmazon API Gatewayを使って動かすことができました。

前回の「Socket Mode＋ローカルPCで動かす」に比べると設定内容やアプリの構造が複雑になりましたが、それでも「Slack Bolt」を使うことでプログラムコードは比較的シンプルに作ることができました。

当初の目的であった「Amazon Bedrockを使った検証を行う際に自分でチャットアプリを作って試せるといいな」が無事達成できたので、今後はこのSlackアプリをベースにいろいろ試してみたいと思います。

参考情報

Slack | Bolt for Python - Bolt 入門ガイド（HTTP）
https://slack.dev/bolt-python/ja-jp/tutorial/getting-started-http

bolt-python/examples/aws_lambda | GitHub
https://github.com/slackapi/bolt-python/tree/main/examples/aws_lambda