生成AI未経験者がBedrockを組み込んだLINEチャットBotをつくってみた

はじめに

この度、ANGEL Calendarの企画に参加しております！
記事一覧は下記のOrganizationアカウントの一覧をチェックしてみてください！
2024-ANGEL-Dojo Organization

生成AIを使ってみたい!!

こんにちは！ANGEL Calendar企画運営メンバーのふじもとです！
ANGEL Calendar、3日目担当させていただきます。

今年、初めてAWS Summitに参加してきました。多くのAWSの事例を見ることができ、とても刺激的な体験でした。参加してみて、やはり生成AI関連の議題が非常に多いと感じました。一方で、そんな激熱トピックに触れることができていない自分に対して焦りも感じました。

ということで、ANGEL Calendarの機会を借りて、生成AIを触っていきたいと思います!
今回は、生成AI未経験者の私が一からBedrockを用いたチャットBotを作ってみました。

記事の前提

• AWSアカウントを持っていること
• LINEのアカウントを持っていること
• 開発環境にAWS CLI, SAM CLI がインストールされていること

開発環境

本記事で利用している各ツールのバージョンは以下のとおりです。

• Python: 3.12.3
• SAM CLI: 1.97.0
• line-bot-sdk-python: 3.11.0

$ python -V
Python 3.12.3

$ sam --version
SAM CLI, version 1.112.0

アーキテクチャ図

今回作成するアーキテクチャ図は以下になります。

Amazon API GatewayとAWS LambdaとAmazon DynamoDBを用いたシンプルなサーバレス構成になります。LINEからのメッセージの応答は、Webhookをトリガーに、API Gatewayを経由して、Lambdaが実行されます。Lambda実行時にAmazon Bedrockを用いて、Botのメッセージを生成します。また、ユーザーとBotのメッセージの履歴をDynamoDBへ保存します。

実装手順

1. LINE Developersのチャネル設定
2. line-bot-sdkをLambda Layerに設定
3. Bedrockのモデルの有効化
4. AWS Serverless Application Model(AWS SAM)によるAWSリソース構築
5. LINE Webhook設定

1. LINE Developersのチャネル設定

LINEを用いてチャットBotを作成するには、チャネルが必要になります。そのため、LINE Developersにログインし、LINE Developersコンソールでチャネルを作成します。
チャネル作成に関しては、Messaging APIを始めようを参考にしてください。
他にもコンソールでは、チャットBotの名前やアイコン画像等の設定ができます。
チャネルが作成できたら、作成したチャネルのMessaging API設定のQRコードから、友だち追加を行います。

2. line-bot-sdkをLambda Layerに設定

AWS SAMでline-bot-sdkを利用できるように、Lambda Layerで設定します。line-bot-sdk-pythonからパッケージをダウンロードし、AWSのマネジメントコンソール上で「レイヤーの作成」を行います。

今回Lambda Layerに設定するline-bot-sdkは、LINEのMessaging APIのクライアントライブラリになります。line-bot-sdkを利用することで、イベントごとの処理や署名の検証などの機能を簡単に開発できます。

3. Bedrockのモデルの有効化

LambdaでBedrockを利用開始する前に、利用するモデルを有効化する必要があります。今回は、東京リージョン(ap-northeast-1)でAmazonのTitan Text G1 - Expressモデルを利用したいと思います。
Bedrockのマネジメントコンソールの「モデルアクセス」から、モデルの有効化を行うことができます。

4. AWS Serverless Application Model(AWS SAM)によるAWSリソース構築

AWS Serverless Application Model(AWS SAM)を用いて、サーバレスアプリケーションを構築します。
ディレクトリ構成は、以下になります。

ディレクトリ構成

|--linebot_reply
|  |--__init__.py
|  |--app.py
|  |--requirements.txt
|--samconfig.toml
|--template.yml

template.yml:展開するAWSリソース(Lambda,API Gateway,DynamoDBなど)を記述
app.py: Lambdaの処理内容を記述(Bedrockの処理も含みます)
※他ファイルはsam init時に作成されたファイルをそのまま流用しています

template.ymlは、以下になります。

template.yml

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: >
  template for linechatbot

Parameters:
  LayerArnParam:
    Description: line-bot-sdk
    Type: String
    Default: ${LambdaLayerのARN}

Globals:
  Function:
    Timeout: 60
    MemorySize: 256
    Environment:
        Variables:
          LINE_CHANNEL_SECRET: ${チャネルシークレット}
          LINE_CHANNEL_ACCESS_TOKEN: ${チャネルアクセストークン}
          MESSAGE_TABLE_NAME: sam-BotMessageHistoryTable

Resources:
  LineChatBotFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: sam-lambda-linechatbot-bedrock
      CodeUri: linebot_reply/
      Handler: app.lambda_handler
      Runtime: python3.12
      Architectures:
        - x86_64
      Layers:
        - !Ref LayerArnParam
      Events:
        LineChatBot:
          Type: Api
          Properties:
            Path: /linebot
            Method: post
      Policies:
      - DynamoDBCrudPolicy:
          TableName: !Ref BotMessageHistoryTable
      - Statement:
        - Sid: Policy
          Effect: Allow
          Action:
          - "bedrock:*"
          Resource: '*'

  BotMessageHistoryTable:
    Type: AWS::DynamoDB::Table
    Properties: 
      TableName: sam-BotMessageHistoryTable
      AttributeDefinitions:
        - AttributeName: 'userId'
          AttributeType: 'S'
        - AttributeName: 'requestDate'
          AttributeType: 'N'
      KeySchema:
        - AttributeName: 'userId'
          KeyType: 'HASH'
        - AttributeName: 'requestDate'
          KeyType: 'RANGE'
      BillingMode: PAY_PER_REQUEST

Outputs:
  LineChatBotApi:
    Description: "API Gateway endpoint URL for Prod stage for LINEChatBot function"
    Value: !Sub "https://${ServerlessRestApi}.execute-api.${AWS::Region}.amazonaws.com/Prod/linebot/"
  LineChatBotFunction:
    Description: "LineChatBotFunction ARN"
    Value: !GetAtt LineChatBotFunction.Arn
  LineChatBotFunctionIAMRole:
    Description: "Implicit IAM Role created for LineChatBot function"
    Value: !GetAtt LineChatBotFunctionRole.Arn

Lambda

先ほど設定したLambda LayerをLayersで定義することで、SAMで構築されるLambdaでline-bot-sdkが利用可能となります。

Lambdaの処理内容は、linebot_replyフォルダのapp.pyで記述します。

app.pyは、以下になります。

app.py

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認証情報をLambdaの環境変数から取得
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設定
table_name = os.getenv('MESSAGE_TABLE_NAME', None)
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(table_name)

# Bedrockランタイムのクライアント作成
bedrock_runtime = boto3.client(service_name='bedrock-runtime')

def lambda_handler(event, context):
    if "x-line-signature" in event["headers"]:
        signature = event["headers"]["x-line-signature"]
    elif "X-Line-Signature" in event["headers"]:
        signature = event["headers"]["X-Line-Signature"]
    body = event["body"]
    ok_json = {
        "isBase64Encoded": False,
        "statusCode": 200,
        "headers": {}, 
        "body": ""
    }
    error_json = {
        "isBase64Encoded": False,
        "statusCode": 500,
        "headers": {},
        "body": "Error"
    }
    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

# Webhookから送られてきたイベントの処理
@handler.add(MessageEvent, message=TextMessage)
def message(line_event):
    # LINEからのイベント情報を取得
    user_message = line_event.message.text
    user_message_type = line_event.message.type
    user_id = line_event.source.user_id
    timestamp = int(line_event.timestamp)
    
    # ユーザーの会話履歴を取得
    pre_user_message, pre_bot_message = get_user_message_history(user_id)
    pre_message_history = f"User:{pre_user_message}\nBot:{pre_bot_message}"
    
    # 会話履歴と新しいメッセージを結合してプロンプトを作成
    prompt = f"{pre_message_history}User:  {user_message}\nBot:"
    
    # プロンプトをtitan-text-expressで処理する
    bot_message = handle_bedrock_titan(prompt)
    
    bot_message_type = 'text'
    
    # DynamoDBにユーザーの会話を記録
    save_conversation_to_dynamodb(user_id, user_message_type, user_message, bot_message_type, bot_message) 

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


# DynamoDBからユーザーの会話履歴を取得
def get_user_message_history(user_id):
    try:
        response = table.query(
            KeyConditionExpression=Key('userId').eq(user_id),
            ScanIndexForward=False,
            Limit=1
        )
        if 'Items' in response and len(response['Items']) > 0:
            user_message = response['Items'][0]['userMessage']
            bot_message = response['Items'][0]['botMessage']
            return user_message, bot_message
        else:
            return "",""
    except ClientError as e:
        logger.error("DynamoDB query failed: {}".format(e.response['Error']['Message']))
        return "",""

# DynamoDBにユーザーの会話を記録
def save_conversation_to_dynamodb(user_id, user_message_type, user_message ,bot_message_type ,bot_message):
    timestamp = int(time.time())
    try:
        response = table.put_item(
            Item={
                'userId': user_id,
                'requestDate': timestamp,
                'userMessage': user_message,
                'userMessageType': user_message_type,
                'botMessage': bot_message,
                'botMessageType': bot_message_type,
                'botActionType': 'reply' 
            }
        )
        logger.info("DynamoDB save successful.")
    except Exception as e:
        logger.error("Error saving to DynamoDB: {}".format(e))

# プロンプトをtitan-text-expressで処理する
def handle_bedrock_titan(prompt):
    # Bedrock Runtimeを使用してAI応答を生成
    response = bedrock_runtime.invoke_model(
        body=json.dumps({
            "inputText": prompt,
            "textGenerationConfig": {
                "temperature": 0.6,  
                "topP": 0.999,
                "maxTokenCount": 300,
                "stopSequences": ["User:"]
            }
        }),
        modelId="amazon.titan-text-express-v1",
        contentType="application/json",
        accept="*/*"
    )
    
    # 応答をJSON形式で取得し、テキストメッセージを取り出す
    response_body = json.loads(response.get('body').read())
    output_txt = response_body['results'][0]['outputText']
    bot_message = output_txt.strip() #response取得時、空白文字が入るため、削除
    return bot_message

今回はあくまでも検証であるため、環境変数にチャネルシークレットとアクセストークンを暗号化なしで設定しています。この設定では、セキュリティ的に問題があるため、AWS Secrets Managerを用いてチャネルシークレットとアクセストークンを保存する、またはKMSで暗号化する等の設定が別途必要になります。

app.pyの処理の流れ

①署名の検証
②Webhookで送られてくるイベントからメッセージを取得
③DynamoDBから過去の会話履歴を取得する
④ユーザーのメッセージと過去の会話履歴からプロンプトを作成して、BedrockでBotの返答を生成する
⑤会話履歴をDynamoDBに保存する
⑥ユーザーに返答のメッセージを送信する

重要なポイントのみ解説していきます。

Webhookの処理方法

Webhookは、line-bot-sdkを使用して、lambda_handlerで認証処理を行います。認証が完了すると、デコレータを用いて、Webhookから送られてくるイベントに応じて、処理を分けて記述します。今回はイベントがメッセージである場合に応答したいので、デコレータ@handler.addで処理を実行します。またこの時、@handler.addの引数messageにメッセージタイプを選択することができます。今回は、テキストメッセージを処理するため、message=TextMessageと記述します。
以下のようにメッセージタイプごとの処理を分けることができます。

@handler.add(MessageEvent, message=ImageMessage): #画像メッセージを処理
@handler.add(MessageEvent, message=AudioMessage): #音声メッセージを処理

イベントとメッセージタイプの詳細は、Webhookイベントオブジェクト、メッセージタイプを参考にしてください。

Bedrockのプロンプト作成

Bedrockのプロンプト作成は、ユーザーとBotの過去の会話履歴とユーザーから送信されてきたメッセージを組み合わせて、作成しています。DynamoDBからユーザーとBotの過去の会話履歴を取得することで、文脈を判断するようにしています。

prompt = f"{pre_message_history}User:  {user_message}\nBot:"

pre_message_history: ユーザーとBotの過去の会話履歴(プロンプト形式)
user_message: ユーザーから送信されたメッセージ

今回は、BedrockでTitan Text G1 - Expressモデルを利用しましたが、AnthropicのClaude Instantを利用するバージョンも試しに作成してみました。

Titan Text G1 - Express同様、Lambdaで利用する前に、モデルの有効化が必要です。

先ほど紹介したapp.pyの@handler.addを以下のように書き換えます。

app.py(Claude ver.)

@handler.add(MessageEvent, message=TextMessage)
def message(line_event):
    # LINEからのメッセージを取得
    user_message = line_event.message.text
    user_message_type = line_event.message.type
    user_id = line_event.source.user_id
    timestamp = int(line_event.timestamp)
    
    # ユーザーの会話履歴を取得
    pre_user_message, pre_bot_message = get_user_message_history(user_id)
    pre_message_history = f"Human:{pre_user_message}\n\nAssistant:{pre_bot_message}"
    
    # 会話履歴と新しいメッセージを結合してプロンプトを作成
    prompt = f"{pre_message_history}\n\nHuman:  {user_message}\n\nAssistant:"
    
    # プロンプトをclaudeで処理する
    bot_message = handle_bedrock_claude(prompt)
    bot_message_type = 'text'
    
    # DynamoDBにユーザーの会話を記録
    save_conversation_to_dynamodb(user_id, user_message_type, user_message, bot_message_type, bot_message) 

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

また、新たに以下の関数を追加します。

app.py(Claude ver.)

# プロンプトをClaudeで処理する
def handle_bedrock_claude(prompt):
    response = bedrock_runtime.invoke_model(
        body=json.dumps({
            "prompt": prompt,
            "max_tokens_to_sample": 300,
            "temperature": 0.6,
            "top_k": 250,
            "top_p": 0.999,
            "stop_sequences": ["\n\nHuman:"],
            "anthropic_version": "bedrock-2023-05-31"
        }),
        modelId='anthropic.claude-instant-v1',
        contentType='application/json',
        accept='*/*'
    )
    # 応答をJSON形式で取得し、テキストメッセージを取り出す(claude)
    response_body = json.loads(response['body'].read().decode('utf-8'))
    output_txt = response_body['completion']
    bot_message = output_txt.strip() 
    return bot_message

Titan Text G1 - ExpressとClaude Instantの処理の流れは同じですが、プロンプトの記述に違いがあるため、ご注意ください。

• Claudeの場合

\n\nHuman: {userQuestion}\n\nAssistant:

• Amazon Titan Textの場合

User: <prompt>\nBot:

また、モデルにリクエストする際に渡すパラメータやResponseにも違いがあります。
詳しくは以下を参考にしてください。

DynamoDB

LINEのユーザーID(userId)をパーティションキー、ユーザーからのメッセージでWebhookが起動した時間(requestDate)をソートキーとして、テーブルを作成します。パーティションキー、ソートキーの他に、Lambdaの処理でもキーを入力します。
他のキーも含めたテーブル定義は以下になります。

キー	型	説明
userId[パーティションキー]	文字列	LINEのユーザーID
requestDate[ソートキー]	数値	Webhook起動時の時間
botMessage	文字列	Botから送信されたメッセージ
botMessageType	文字列	Botのメッセージのタイプ
userMessage	文字列	ユーザーから送信されたメッセージ
userMessageType	文字列	メッセージのタイプ

API Gateway

OutputsのLineChatBotApiで、API GatewayのエンドポイントURLがLineChatBotApiとして出力されるようになっています。AWS CloudFotmationのコンソールから、SAMで作成されたスタックから確認できます。このエンドポイントのURLは後ほどLINEのWebhook設定時に必要になるため、手元にメモしておいてください。

5. LINE Webhook設定

最後に、作成したLINEチャネルと構築したAWSリソースを繋げる作業を行います。
LINE Developersコンソールに移動して、作成したチャネルのMessaging API設定のWebhook設定から、先ほどメモしたAPI GatewayのエンドポイントURLをWebhook URLに設定します。

またこの時、LINE公式アカウント機能の「応答メッセージ」を無効に設定します。

これでLINEチャットBotの構築が完了しました。

動作確認

早速、動作確認してみます。
ユーザーが「AWSについて教えて」というメッセージを送信すると、Bedrockによって生成された回答が返されていることが分かります。また、続けて「略称は?」というユーザーの問いかけに対して、「AWSのことを話している」という文脈を理解した上で、BotがAWSの略称について答えています。
Bedrockによる応答を返すLINEチャットBotを作成することができました。

最後に

今回、生成AI未経験者が一からBedrockを組み込んだLINEチャットBotを作ってみました。作ってみた感想としては、こんなに簡単に生成AIを利用できることに驚きました。AWS SDK for Python (Boto3)でBedrockが提供されているので、普段Pythonを多く利用している方は簡単にシステムに組み込めると思います。また、取得した情報を他のAI系のサービス(Rekognition,Comprehendなど)と組み合わせるとより汎用的なチャットBotが作れそうだなと思いました。今後も、最新の生成AI技術をキャッチアップしていけるように検証を続けたいと思います。

参考

• Messaging API
• Messaging APIを始めよう
• メッセージタイプ
• Webhookイベントオブジェクト
• Anthropic Claude Text Completions API
• Amazon Titan Text モデル
• line/line-bot-sdk-python
• 【AWS】PythonによるLINE Bot構築