Amazon Bedrock+Anthropic Claude 3 SonnetでSlackチャットボットを作成する

はじめに

前回の記事｢Amazon Bedrock+ Anthropic Claude 3 Sonnetでストリーミングテキストを生成する｣では､WSL上のUbuntu 23.04でAmazon Bedrockを使用して､Anthropic社のClaude 3 Sonnetを使ってストリーミングテキストを生成しました｡

本記事では､前回の記事をベースにSlackチャットボットを作成してみました｡今回のチャットボットは会話履歴を保持していないので､質問に答えてくれるだけのシンプルなものです｡

デモ画面

入力したメッセージ

 少年時代にゴーカートからレースキャリアをスタートし､青年となった今､憧れのフォーミュラレースのスターティンググリッドについた青年の物語を小説風に書いてください｡
 - 物語は､青年の目線で描いてください｡
 - 文字数は1,000文字以内のショートショートにまとめてください｡
 - タイトルをつけてください｡
 - 作品の最後には､青年がスターティンググリッドについた瞬間の心情を描いてください｡

チャットボットの応答メッセージ

タイトル: グリッドへの夢
カウントダウンの音が響き渡る。エンジンの力強い吐息が身体を震わせる。この日を夢見て幾年もの時を重ねてきた。
幼い頃から、レーサーになることが私の夢だった。父のゴーカートでレースの醍醐味を知り、小さな心に種が植えられた。周りの期待や心配する声もあったが、私の情熱は止まらなかった。
レース活動に勤しみ、次第に実力をつけた。しかし、プロの世界へ足を踏み入れるのは難しかった。障害や失敗の数々。でも、あきらめる決心はつかなかった。夢に賭けた一生だったから。
ついに、長年の努力が実を結んだ。憧れのフォーミュラ車にチャンスが巡ってきた。緊張と興奮の入り交じる思いでマシンに乗り込む。エンジンがすさまじい力で唸る。
ピットクルーの声が聞こえる。「30秒!」時が迫る。深く息を吐く。瞼の裏に走馬灯のようにこれまでの軌跡が去来する。苦労や喜びが一瞬の中に凝縮される。
「10秒!」心が熱くなる。憧れの赤いスターティンググリッドに向かって一歩を踏み出す。
夢の第一歩を踏み出した私の目には、希望に満ちた未来の道が見えた。

構成図

参考情報

環境構築

• 以下のリソースを作成･削除･変更できる権限をもつAWSユーザーを利用すること
  • AWS IAM
  • AWS Lambda
  • AWS CloudFormation
  • AWS Secrets Manager
  • Amazon API Gateway
  • Amazon S3
  • Amazon CloudWatch Logs
  • Amazon Bedrock
• 使用するAWSリージョンは、us-east-1
• Slack Appを作成するためのアカウントや権限を持っている

開発環境構築

作業環境のOSバージョン

Windows 11上のWSLでUbuntu 23.04を動かしています。

$ cat /etc/os-release | grep PRETTY_NAME
PRETTY_NAME="Ubuntu 23.04"

Python環境

$ python3 --version
Python 3.12.0

$ python3 -m venv .venv
source .venv/bin/activate

$ pip3 install --upgrade pip
$ pip3 --version
pip 24 from /home/xxx/.venv/lib/python3.12/site-packages/pip (python 3.12)

$ pip install boto3

AWS SAM CLIインストール

AWS上でサーバーレスアプリケーションを構築､実行するAWS SAMを使用します｡

Installing the AWS SAM CLI の手順に従い、AWS SAM CLIをインストールします。今回はx86_64環境でLinux OSを使用するため、x86_64 - command line installerの手順を実行します｡

$ sam --version
SAM CLI, version 1.113.0

Slack Appの作成

Slack Appの作成は､以前の記事を参考にしてください｡今回は､App Nameをbedrock-slack-appとして作成しました｡

Slackシークレット情報をSecret Managerに登録

あらたにシークレットを作成し、ここまでの手順で作成した以下のシークレット情報を登録します。シークレット名は、bedrock-slack-appとします。このシークレット名は後述のtemplate.yaml内のSECRET_NAMEに定義しているため、異なるシークレット名を使用する場合はtemplate.yaml内のSECRET_NAMEも変更します。

シークレットキー	値
SlackSigningSecret	SlackのSigning Secret
SlackBotToken	SlackのBot User OAuth Token

アプリケーションの構築

ディレクトリ構造は以下のとおりです｡

.
├── bedrock-slack-app
│   ├── __init__.py
│   ├── app.py
│   └── requirements.txt
├── samconfig.toml
└── template.yaml

init.pyは空のファイルです｡
bedrock-slack-app/requirements.txtは以下のとおりです｡boto3やrequestsも必要ですが､それらはLambdaレイヤーで追加するようtemplate.yamlに記述します｡

slack-bolt
slack-sdk

template.yamlの構成

template.yaml (長いので折りたたんでいます｡クリックして展開)

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: Bedrock Slack App.

Resources:
  BedrockAssitantFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: bedrock-slack-app/
      Handler: app.lambda_handler
      Runtime: python3.12
      Role: !GetAtt LambdaRole.Arn
      Timeout: 300
      MemorySize: 512
      Architectures:
        - arm64
      Environment:
        Variables:
          SECRET_NAME: 'bedrock-slack-app' # Name of the secret in Secrets Manager
          REGION_NAME: 'us-east-1' # Region of the secret in Secrets Manager
      Events:
        Slack:
          Type: Api
          Properties:
            Method: POST
            Path: /slack/events
      Layers:
        # Layer for AWS Parameter Store and Secrets Manager
        # https://docs.aws.amazon.com/systems-manager/latest/userguide/ps-integration-lambda-extensions.html#ps-integration-lambda-extensions-add
        - arn:aws:lambda:us-east-1:177933569100:layer:AWS-Parameters-and-Secrets-Lambda-Extension-Arm64:11
        # Layer for boto3
        # https://github.com/keithrozario/Klayers?tab=readme-ov-file#list-of-arns
        - arn:aws:lambda:us-east-1:770693421928:layer:Klayers-p312-arm64-boto3:3

  # IAM Role for lambda.
  LambdaRole:
    Type: "AWS::IAM::Role"
    Properties:
      RoleName: bedrock-slack-app-lambda-role
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      Policies:
        - PolicyName: allow-lambda-invocation
          PolicyDocument:
            Version: "2012-10-17"
            Statement:
              - Effect: Allow
                Action:
                  - lambda:InvokeFunction
                  - lambda:InvokeAsync
                Resource: "*"
        - PolicyName: SecretsManagerPolicy
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action: 'secretsmanager:GetSecretValue' # Required for Lambda to retrieve the secret
                Resource: "*"
        - PolicyName: allow-bedrock-access
          PolicyDocument:
            Version: "2012-10-17"
            Statement:
              - Effect: Allow
                Action:
                  - bedrock:InvokeModel
                  - bedrock:InvokeModelWithResponseStream
                Resource: "*"
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole

  BedrockLogGroup:
    Type: AWS::Logs::LogGroup
    Properties:
      LogGroupName: !Sub /aws/lambda/${BedrockAssitantFunction}
      RetentionInDays: 14 # Optional. Default retention is 30 days.

Outputs:
  BedrockAssitantApi:
    Description: "The URL of Slack Event Subscriptions"
    Value: !Sub "https://${ServerlessRestApi}.execute-api.${AWS::Region}.amazonaws.com/Prod/slack/events"
  BedrockAssitantFunction:
    Description: "Bedrock Assistant Lambda Function ARN"
    Value: !GetAtt BedrockAssitantFunction.Arn
  BedrockAssitantFunctionIamRole:
    Description: "Implicit IAM Role created for Bedrock Assistant function"
    Value: !GetAtt LambdaRole.Arn

AWS SAM テンプレートファイル(template.yaml)に､作成するAWSリソースを定義します｡
Lambda関数用ロールやポリシー､Lambdaの環境変数などを記述します｡その他に､以下のレイヤーやリソースベースポリシーが含まれます｡

• Lambda関数からSecrets ManagerにアクセスするためのAWS-Parameters-and-Secrets-Lambda-Extensionレイヤー
• Lambda関数内からimportするためのboto3をパッケージにしたレイヤー
• BedrockkからLambda関数を扱うためのリソースベースポリシー

AWS-Parameters-and-Secrets-Lambda-ExtensionレイヤーのARNはUsing Parameter Store parameters in AWS Lambda functionsのドキュメントにリージョンやアーキテクチャ毎に掲載されています｡
boto3やrequestsのレイヤーは､keithrozario/KlayersのリポジトリにあるList of ARNsに､Pythonバージョンとアーキテクチャ毎に掲載されています｡今回は､Python 3.12, arm64を使用するので､Layers for Python 3.12-arm64の一覧からus-east-1用のARNを使用します｡

Lambdaレイヤーは以下のようにtemplate.yamlに記述しています｡

      Layers:
        # Layer for AWS Parameter Store and Secrets Manager
        # https://docs.aws.amazon.com/systems-manager/latest/userguide/ps-integration-lambda-extensions.html#ps-integration-lambda-extensions-add
        - arn:aws:lambda:us-east-1:177933569100:layer:AWS-Parameters-and-Secrets-Lambda-Extension-Arm64:11
        # Layer for boto3
        # https://github.com/keithrozario/Klayers?tab=readme-ov-file#list-of-arns
        - arn:aws:lambda:us-east-1:770693421928:layer:Klayers-p312-arm64-boto3:3

適用すると､Lambda関数のLayersに以下のように表示されます｡

リソースベースポリシーは､以下のようにtemplate.yamlに記述しています｡

  BacklogSearchFunction:
    Type: AWS::Serverless::Function
    Properties:

(途中省略)

  # Resouse based policy for lambda.
  PermissionForBacklogSearchToInvokeLambda:
    Type: AWS::Lambda::Permission
    Properties:
      FunctionName: !GetAtt BacklogSearchFunction.Arn
      Action: lambda:InvokeFunction
      Principal: bedrock.amazonaws.com

適用すると､Lambda関数の設定のResource-based policy statementsに以下のように表示されます｡

template.yaml内のEnvironmentにあるSECRET_NAMEとREGION_NAMEには､それぞれ先ほど作成したSecrets Managerのシークレットの名前とリージョンを設定します｡

Lambda関数の実行ロール内のポリシーではResource: "*"としていますが、本番環境などでは許可するリソースを特定したほうがよいでしょう。例えば、AgentsのResourceを特定する場合は以下のように指定します。

Resource: "arn:aws:bedrock:{Region}:{AWS Account ID}:agent-alias/{Agents ID}/{Agents Alias}"

samconfig.tomlの構成

samconfig.toml (長いので折りたたんでいます｡クリックして展開)

# More information about the configuration file can be found here:
# https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/serverless-sam-cli-config.html
version = 0.1

[default]
[default.global.parameters]
stack_name = "bedrock-slack-app"

[default.build.parameters]
cached = true
parallel = true

[default.validate.parameters]
lint = true

[default.deploy.parameters]
capabilities = "CAPABILITY_NAMED_IAM"
confirm_changeset = true
resolve_s3 = true
region = "us-east-1"

[default.package.parameters]
resolve_s3 = true

[default.sync.parameters]
watch = true

[default.local_start_api.parameters]
warm_containers = "EAGER"

[default.local_start_lambda.parameters]
warm_containers = "EAGER"

SAM CLIの実行設定ファイル(samconfig.toml)に､SAM CLIを実行する際の設定を定義します｡AWS SAMのチュートリアル: Hello World アプリケーションのデプロイを実行した際に作成されるsamconfig.tomlをもとにしています｡今回の例では､以下の点を変更しています｡

• [default.global.parameters]セクションのstack_nameを"sam-app"から"bedrock-slack-app"に変更
• [default.deploy.parameters]セクションにregion指定を追加
• [default.deploy.parameters]セクションのcapabilitiesを"CAPABILITY_IAM"から"CAPABILITY_NAMED_IAM"に変更

bedrock-slack-app/app.pyの構成

bedrock-slack-app/app.p (長いので折りたたんでいます｡クリックして展開)

import ast
import json
import logging
import os
import re
import time

import boto3
from botocore.exceptions import ClientError
from slack_bolt import App
from slack_bolt.adapter.aws_lambda import SlackRequestHandler

SlackRequestHandler.clear_all_log_handlers()
logging.basicConfig(
    format="%(asctime)s [%(levelname)s] %(message)s",
    level=logging.DEBUG
)

logger = logging.getLogger(__name__)


class SecretsManager:
    """
    Class to retrieve secrets from Secrets Manager

    Attributes:
        secret_name (str): The name of the secret
        region_name (str): The name of the region
        client (boto3.client): The client for Secrets Manager
    """

    def __init__(self, secret_name, region_name):
        self.secret_name = secret_name
        self.region_name = region_name
        self.client = boto3.client(
            service_name='secretsmanager',
            region_name=region_name
        )

    def get_secret(self, key):
        """
        Retrieves the value of a secret based on the provided key.

        Args:
            key (str): The key of the secret to retrieve.

        Returns:
            str: The value of the secret.

        Raises:
            ClientError: If there is an error retrieving the secret.
        """
        try:
            get_secret_value_response = self.client.get_secret_value(
                SecretId=self.secret_name
            )
        except ClientError as e:
            raise e

        secret_data = get_secret_value_response['SecretString']
        secret = ast.literal_eval(secret_data)

        return secret[key]


secrets_manager = SecretsManager(
    secret_name=os.environ.get("SECRET_NAME"),
    region_name=os.environ.get("REGION_NAME")
)

app = App(
    signing_secret=secrets_manager.get_secret("SlackSigningSecret"),
    token=secrets_manager.get_secret("SlackBotToken"),
    process_before_response=True,
)


def generate_message(bedrock_runtime, model_id, system_prompt, messages):
    """
    Streams the response from a multimodal prompt.
    Args:
        bedrock_runtime: The Amazon Bedrock boto3 client.
        model_id (str): The model ID to use.
        system_prompt (str) : The system prompt text.
        messages (JSON) : The messages to send to the model.
        max_tokens (int) : The maximum  number of tokens to generate.
    Returns:
        None.
    """

    body = json.dumps(
        {
            "anthropic_version": "bedrock-2023-05-31",
            "max_tokens": 4000,
            "temperature": 1,  # default value is 1
            "top_p": 0.999,    # default value is 0.999
            "system": system_prompt,
            "messages": messages
        }
    )

    response = bedrock_runtime.invoke_model_with_response_stream(
        body=body,
        modelId=model_id)

    for event in response.get("body"):
        chunk = json.loads(event["chunk"]["bytes"])

        if chunk['type'] == 'content_block_delta':
            if chunk['delta']['type'] == 'text_delta':
                yield chunk['delta']['text']


def create_message_blocks(text):
    """
    Creates the message blocks for updating the Slack message.

    Args:
        text (str): The updated text for the Slack message.

    Returns:
        list: The message blocks for updating the Slack message.
    """
    message_context = "Claude 3 Sonnetで生成される情報は不正確な場合があります。"
    message_blocks = [
        {"type": "section", "text": {"type": "mrkdwn", "text": text}},
        {"type": "divider"},
        {"type": "context", "elements": [{"type": "mrkdwn", "text": message_context}]},
    ]
    return message_blocks


def handle_app_mentions(event, say):
    """
    Handle app mentions in Slack.

    Args:
        event (dict): The event data containing information about the mention.
        say (function): The function used to send a message in Slack.

    Returns:
        None
    """

    channel = event["channel"]
    thread_ts = event["ts"]
    input_text = re.sub("<@.*>", "", event["text"])
    userid = event["user"]
    interval = 0.4
    update_count = 0

    last_send_time = time.time()

    result = say("\n\nお待ちください...", thread_ts=thread_ts)
    ts = result["ts"]

    last_post_text = ""
    text = "<@" + userid + "> "

    bedrock_runtime = boto3.client(service_name='bedrock-runtime')

    model_id = 'anthropic.claude-3-sonnet-20240229-v1:0'
    system_prompt = """
あなたはチャットでユーザを支援するAIアシスタントです｡

* Let's think step by step
* Take a deep breath
* 回答のルールが<rules></rules>にあります｡ ルールを理解してください｡ルールは必ず守ってください｡例外はありません｡
* 「〜って何？」「〜とは？」「〜を説明して」というような概要を聞く質問については、「〜の概要」と読み替えてください｡

<rules>
 * 日本語で質問された場合、必ず日本語で回答してください｡
 * 質問に具体性がなく回答できない場合は、質問の仕方をアドバイスしてください｡､
</rules>
"""

    user_message = {
        "role": "user",
        "content": [
            {
                "type": "text",
                "text": input_text
            }
        ]
    }
    messages = [user_message]

    try:
        for sentence in generate_message(bedrock_runtime, model_id, system_prompt, messages):
            text += sentence

            now = time.time()
            if (now - last_send_time) > interval:
                last_send_time = now
                update_count += 1

                if update_count / 10 > interval:
                    interval = interval * 2

                last_post_text = text

                message_blocks = create_message_blocks(text)
                app.client.chat_update(channel=channel, ts=ts, text=text, message_blocks=message_blocks)

        if last_post_text != text:
            message_blocks = create_message_blocks(text)
            app.client.chat_update(channel=channel, ts=ts, text=text, blocks=message_blocks)

    except ClientError as err:
        message = err.response["Error"]["Message"]
        logger.error("A client error occurred: %s", message)
        print("A client error occured: " + format(message))
    except ThrottlingException as e


def respond_to_slack_within_3_seconds(ack):
    """
    Responds to a Slack message within 3 seconds.

    Parameters:
    - ack: A function to acknowledge the Slack message.

    Returns:
    None
    """
    ack()


app.event("app_mention")(
    ack=respond_to_slack_within_3_seconds,
    lazy=[handle_app_mentions]
)


def lambda_handler(event, context):
    """
    Lambda function handler for processing Slack events.

    Args:
        event (dict): The event data passed to the Lambda function.
        context (object): The runtime information of the Lambda function.

    Returns:
        dict: The response data to be returned by the Lambda function.
    """
    print(event)

    retry_counts = event.get("multiValueHeaders", {}).get("X-Slack-Retry-Num", [0])

    if retry_counts[0] != 0:
        logging.info("Skip slack retrying(%s).", retry_counts)
        return {}

    slack_handler = SlackRequestHandler(app=app)
    return slack_handler.handle(event, context)

bedrock-slack-app/app.pyは､メンションされたメッセージを取得して､そのメッセージをAmazon Bedrockを使用して生成したテキストを返すアプリケーションです｡ invoke_model_with_response_stream関数を使用して､ストリーミングテキストを生成します｡

ビルドとデプロイ

ビルドとデプロイ､Slack Appの設定は以前の記事を参考にしてください｡

動作確認

任意のチャンネンルに､rock-slack-app (Slack Appの作成時に設定した名前)を招待し､メンション形式で依頼をポストします｡動作しない場合のログやBedrockが返すレスポンスは CloudWatch Logs に出力されているログが参考になります｡

まとめ

Slackチャットボットを作成しました｡会話履歴を保持していないので､質問に答えてくれるだけのシンプルなものです｡これをベースに､DynamoDBなどを使って会話履歴を保持するように拡張することもできます｡ Knowledge Bases for Amazon Bedrock と組み合わせることで､RAGを使ったQAシステムを構築することも可能ではないでしょうか｡