Amazon Bedrockとチャットできる LINEbotボット構築ハンズオン

INAMURA

はじめに

AWSから生成AIサービス「Amazon Bedrock」がGAされました。
この1ヶ月巷の至る所で「Bedrock」に関する勉強会や、事例の紹介がおこなわれています。
そんな熱気に当てられ、少し出足が遅くなりましたが「自分でも構築してみよう！」ということで ハンズオンしてみました！
なにより、こんな軽い気持ちで生成AIにさわれるだなんて控えめに言ってAWS最高ですね！

構成

処理フロー

①LINEでメッセージの送信
②APIGateway経由で Lambda 実行
③Lambdaの実行処理
　　　　③-1 LINEのリクエストを検証
　　　　③-2　DynamoDBから会話履歴の取得
　　　　③-3 LINEメッセージを　Bedrockに送信・応答を取得して LINEユーザに返信
　　　　③-4 ユーザとの会話をDynamoDBに保存

構成図

参考リンク

AWSドキュメント系
・Amazon Bedrock boto3 セットアップ
・Boto3 ドキュメント
・Lambda ランタイム　ドキュメント

テックブログ系
・DevlopersIO：[Amazon Bedrock] Lambda関数からBedrockを呼び出してみた(著：青柳英明様)
・Qiita：Amazon BedrockをAWS Lambda上で呼び出してみた（API化）(著：@HayaP(Kohei Hayakawa)様)

構築

前提条件

・命名については一例として記載しているため、利用状況に応じて変更ください。
・LINE – APIGateway　- Lambda の構築部分については、AWS Lambdaを利用したLINEbotハンズオンを参照。

1.AmazonBedrock セットアップ

Amazon Bedrockは新たにモデルを使用するため設定を行う。

セットアップしたAmazonBedrockのサマリ

リージョン 企業 モデル
ap-northeast-1 Anthropic Claude Instant

1.1.AmazonBedrock　モデル使用のための設定

AWSマネコン > Amazon Bedrock > 左ペイン「Model access」 > 赤枠「Manage model access」を押下

利用するモデル(今回は Claude Instant)に左側赤枠 チェックを入れる > 右下赤枠「save changes」を押下

いくつか質問に回答後、下記画面で「Access granted」が表示されれば、モデルが利用可能となる。

2.Lambdaの作成

構築したLambdaのサマリ

関数名 ランタイム アーキテクチャ 実行ロール
bedrock-lambda-inamura Python 3.11 arm64 Bedrock-fullaccess-role

構築順序サマリ

No. 構築リソース 説明
2.1 IAMロール LambdaにアタッチするIAMロール
2.2 Lambda Lambdaの本体
2.3 boto3 ダウンロード Python3.11のデフォルトboto3だとバージョンが古いため設定
2.4 Lambdaレイヤ設定 2.3で取得した boto3を利用するために　Lambdaレイヤに設定
2.5 Pythonコード Lambdaの中身（コード部分）
2.5 LINE-APIGateway-Lambda連携 LINEのメッセージをLambdaに送付できるように設定

2.1.IAMロール作成

リソース リソース名 タイプ 説明
IAMロール Bedrock-fullaccess-role – LambdaにアタッチするIAMロール
IAMポリシー Bedrock-fullaccess カスタマー管理 Bedrockを利用できるようにするためのポリシー
IAMポリシー AmazonDynamoDBFullAccess AWS管理 DynamoDBを利用できるようにするためのポリシー
IAMポリシー CloudWatchLogsFullAccess AWS管理 CloudWatchLogsを利用できるようにするためのポリシー
2.1.1.IAMポリシー作成

下記内容でIAMポリシーを作成する

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

2.1.2.IAMロール作成

2.1.1.で作成したIAMポリシー及び、DynamoDB、CloudWatchLogsに書き込みをできるように「AmazonDynamoDBFullAccess」、「CloudWatchLogsFullAccess」をアタッチする。
※検証のためAWS管理ポリシーを利用する

2.2.Lambdaの作成

関数名 ランタイム アーキテクチャ 実行ロール
bedrock-lambda-inamura Python 3.11 arm64 Bedrock-fullaccess-role(※手順2.1で作成したロール)

2.3.boto3 ダウンロード

2023/11/5 現在、Python3.11のboto3は 1.27.1であるため、最新のboto3を利用する必要がある
zipにしてひとまとめにして、後続 2.4.のレイヤー作成時に利用する
参照：Lambda ランタイム　ドキュメント

2.3.1.boto3 ダウンロード

・Linuxの場合

# ディレクトリの作成
$ mkdir -p bedrock_demo/python
# boto3のインストール
$ pip install -t bedrock_demo/python boto3
# bedrock_demoディレクトリに移動
$ cd bedrock_demo
# ZIPファイルの作成
$ zip -r boto3-1.28.77.zip python

・Windowsの場合

# ディレクトリの作成
New-Item -ItemType Directory -Path bedrock_demo\python
# boto3のインストール
pip install -t bedrock_demo\python boto3
# bedrock_demoディレクトリに移動
cd bedrock_demo
# ZIPファイルの作成
Compress-Archive -Path python\* -DestinationPath boto3-1.28.77.zip

2.4.Lambdaレイヤ設定

2.3で取得した boto3を利用するために、Lambdaレイヤを作成する
参照：Lambdaレイヤーでの作業

名前 アップロード 互換性のあるアーキテクチャ 互換性のあるランタイム
boto3_12877 2.3.手順で作成したboto3のzipファイル arm64 Python 3.11

※互換性のあるアーキテクチャについては、2.2.手順で構築したLambdaと合わせる必要がある

2.4.1.Lambdaレイヤの作成画面遷移

AWSマネコン > Lambda > 左ペイン「レイヤー」 > 右側赤枠「レイヤーの作成」押下

2.4.2.Lambdaレイヤの作成

赤枠部分を入力及び、ファイルのアップロード

2.4.3.Lambdaレイヤの設定

AWSマネコン > Lambda > 左ペイン「関数」 > 手順2.2.で構築したLambda（bedrock-lambda-inamura）選択 > 「コード」画面の 「レイヤーの追加」を選択 > 手順2.4.2.で作成したレイヤを設定する

2.5.Pythonコードデプロイ

2.5.1.コード

下記コードをLambdaのコードソースに貼り付け Deployを実行する

import os
import sys
import json
import boto3
import time
from botocore.exceptions import ClientError
from boto3.dynamodb.conditions import Key, Attr

from linebot import (
    LineBotApi, WebhookHandler
)
from linebot.models import (
    MessageEvent, TextMessage, TextSendMessage,
)
from linebot.exceptions import (
    LineBotApiError, InvalidSignatureError
)
import logging

# ロギング設定
logger = logging.getLogger()
logger.setLevel(logging.ERROR)

# LINE認証情報の取得
channel_secret = os.getenv('LINE_CHANNEL_SECRET', None)
channel_access_token = os.getenv('LINE_CHANNEL_ACCESS_TOKEN', None)
if channel_secret is None:
    logger.error('Specify LINE_CHANNEL_SECRET as environment variable.')
    sys.exit(1)
if channel_access_token is None:
    logger.error('Specify LINE_CHANNEL_ACCESS_TOKEN as environment variable.')
    sys.exit(1)

# LINE APIクライアントの初期化
line_bot_api = LineBotApi(channel_access_token)
handler = WebhookHandler(channel_secret)
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('UserConversations')

# Bedrock Runtimeクライアントの初期化
bedrock_runtime = boto3.client(service_name='bedrock-runtime')

def lambda_handler(event, context):
    # LINEからの署名確認
    signature = event["headers"].get("x-line-signature") or event["headers"].get("X-Line-Signature")
    body = event["body"]

    ok_json = {"isBase64Encoded": False, "statusCode": 200, "headers": {}, "body": ""}
    error_json = {"isBase64Encoded": False, "statusCode": 500, "headers": {}, "body": "Error"}

    # LINEからのリクエストを処理
    try:
        handler.handle(body, signature)
    except LineBotApiError as e:
        logger.error("Got exception from LINE Messaging API: %s\n" % e.message)
        for m in e.error.details:
            logger.error("  %s: %s" % (m.property, m.message))
        return error_json
    except InvalidSignatureError:
        return error_json

    # 処理が成功した場合のレスポンス
    return ok_json

# DynamoDBからユーザーの会話履歴を取得する関数を追加します。
def get_user_conversation_history(user_id):
    try:
        # ソートキー（Timestamp）に基づいて最新の履歴を取得する
        response = table.query(
            KeyConditionExpression=Key('UserID').eq(user_id),
            ScanIndexForward=False,  # 結果を降順にする
            Limit=1  # 最新の1項目だけ取得
        )
        # 履歴が存在すれば、それを返す
        if 'Items' in response and len(response['Items']) > 0:
            return response['Items'][0]['History']
        else:
            return ""  # 履歴がない場合は空文字を返す
    except ClientError as e:
        logger.error("DynamoDB query failed: {}".format(e.response['Error']['Message']))
        return ""  # エラーが発生した場合も空文字を返す

# message関数を更新
@handler.add(MessageEvent, message=TextMessage)
def message(line_event):
    # LINEからのメッセージを取得
    user_message = line_event.message.text
    user_id = line_event.source.user_id
    timestamp = int(line_event.timestamp)

    # ユーザーの会話履歴を取得
    conversation_history = get_user_conversation_history(user_id)

    # 会話履歴と新しいメッセージを結合してプロンプトを作成
    prompt = f"{conversation_history}\n\nHuman: {user_message}\n\nAssistant:"

    # Bedrock Runtimeを使用してAI応答を生成
    response = bedrock_runtime.invoke_model(
        modelId='anthropic.claude-instant-v1',
        contentType='application/json',
        accept='*/*',
        body=json.dumps({
            "prompt": prompt,
            "max_tokens_to_sample": 300,
            "temperature": 1,
            "top_k": 250,
            "top_p": 0.999,
            "stop_sequences": ["\n\nHuman:"],
            "anthropic_version": "bedrock-2023-05-31"
        })
    )

    # 応答をJSON形式で取得し、テキストメッセージを取り出す
    response_body = json.loads(response['body'].read().decode('utf-8'))
    ai_text_response = response_body['completion']

    # DynamoDBにユーザーの会話を記録
    save_conversation_to_dynamodb(user_id, prompt + ai_text_response)

    # LINEユーザーに応答を返す
    line_bot_api.reply_message(line_event.reply_token, TextSendMessage(text=ai_text_response))

    # DynamoDBにユーザーの会話を記録
def save_conversation_to_dynamodb(user_id, full_conversation):
    timestamp = str(int(time.time()))
    try:
        response = table.put_item(
            Item={
                'UserID': user_id,
                'Timestamp': timestamp,
                'History': full_conversation
            }
        )
        logger.info("DynamoDB save successful.")
    except Exception as e:
        logger.error("Error saving to DynamoDB: {}".format(e))

2.5.2.Tips

・API requestの形式について、Bedrockの 左ペイン「Providers」より参照できる

・より詳細な API requestについては 左ペイン「Playgrounds」の「Text」　右下「View API request」を押下

どのようなAPI request形式になっているのか確認しやすい

2.6.LINE – APIGateway – Lambda の連携

LINE – APIGateway　- Lambda の構築部分については、AWS Lambdaを利用したLINEbotハンズオンを参照。

3.DynamoDBの作成

構築したDynamoDBのサマリ

テーブル名 パーテーションキー ソートキー キー
UserConversations UserID (String) Timestamp (String) History

3.1.テーブル作成

AWSマネコン > DynamoDB > 左ペイン「テーブル」 > 右側赤枠「テーブルの作成」を押下

3.2.テーブル設定

赤枠部分を入力し、設定画面右下「テーブルの作成」を押下
※検証レベルの構築のため、詳細は考慮していない

※3.3.メッセージを自動的に削除する場合(Time to Live)

AWSマネコン > DynamoDB > 左ペイン「テーブル　設定の更新」 > テーブル選択 > 追加の設定タグを選択　> TTLの設定を押下

以上で、構築については完了となる

挙動の確認

1.LINEでメッセージを送信して、メッセージに対する回答が送られてくる

①富士山の高さを質問
※何度か投稿を繰り返しているため、前の質問を引きずってしますが、高さについては答えられてます

②高尾山の高さを質問

2.前回のメッセージ ＋ 自身の回答を踏まえた回答が送られてくる

③指示語を利用して前回の回答を保持しているのかの質問

3.DynamoDBにてメッセージが保管されている

・AWS側のDynamoDBの画面

・DynamoDB History部分

さいごに

ブログとりまとめていた最中ですが、もしかしたら「DynamoDBのTimestamp不要ですかね？」そうすれば毎回上書きしてくれるんじゃ、、などブログを書いている最中にも色々な反省点が出てきます。「修正しないんかい！」というツッコミをもらいそうですが、そう言った部分も踏まえて次のブログに活かせればと思います。

さて最後にAmazonBedrock（Claude）に質問したら、こんな回答をいただけました。

おっしゃる通り技術の進歩を楽しみながら、適切な利用ができるように少しずつでも学んでいこうと思います！

Last modified: 2023-11-05

Author