LINEで画像生成ボットを作ってみた(Bedrock：Titan Image Generator G1 v2)

はじめに

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

作成した成果物

LINEのボットを作成し、そこにテキストのメッセージを送信すると、そのメッセージをプロンプトとしてBedrockで生成された画像が返ってくる仕組みです
せっかくAWSを学んでいるのでAWSを使って何か作ってみたいなと思っていたのと、日常利用しているLINEとAWSの何かを連携させることができたら達成感ありそうだなと思ったので、作ってみました
実際に作ったボットとのトーク画面は以下のようになりました

前提

1. AWSアカウントを持っていること
2. LINEアカウントを持っていること
3. ローカルPCにPythonがインストールされていること

注意事項・特記事項

• とにかく短時間で動かすことが目標だったため、突っ込みどころ満載かと思いますがご容赦ください
• Lambdaのレイヤーを登録すること、タイムアウトを伸ばすことは忘れがちかなと思います
• Bedrockで使用したモデルは右記です。：Titan Image Generator G1 v2

前提

1. AWSアカウントを持っていること
2. LINEアカウントを持っていること
3. ローカルPCにPythonがインストールされていること

アーキテクチャ

フロー

1. LINEのトーク画面でテキストを送信
2. Amazon API Gatewayを通り、Lambdaがテキストを受け取る
3. Amazon Translateを利用してテキストを英語に翻訳する
4. Amazon Bedrockを利用して翻訳したテキストをプロンプトとして画像を生成
5. S3に画像を保存し、署名付きURLを取得
6. 署名付きURLをレスポンスとして送信
7. LINEが署名付きURLにアクセスし画像を取得

実装手順

1. LINE Developersにログインし、プロバイダー、チャネルを作成する

1. LINE Developersにおいて、下記記事を参考に個人のLINEアカウントでログインしました
   　Developersコンソールへのログイン
2. ログイン後、下記記事を参考にプロバイダーを作成します
   　チャネルを作成する
3. プロバイダー作成後、チャネルを作成します
   チャネルの種類はMessaging APIを選択し、必要な情報を入力します

2. Bedrockのモデルアクセス

1. リージョンを「米国東部 (バージニア北部)us-east-1」に設定する

2. Amazon Bedrockのコンソール画面に移動する

3. ページ左側のメニューより、Bedrock configurations＞「モデルアクセス」を選択する
   

4. 今回利用する「Titan Image Generator G1 v2」の下記の手順でモデルへのアクセス権を得る
   「リクエスト可能」＞モデルリクエスト＞NEXT＞Submit
   下記画面のように、表示されれば成功
   

3. S3バケットを作成

1. Amazon S3のコンソール画面に移動する
2. 「バケットを作成」ボタンから作成画面に移り、適当なバケット名を記入し、それ以外はデフォルトの設定のままバケットを作成

4.Lambda用のIAMロールを作成

1. コンソールからIdentity and Access Management (IAM)を開き、「ロールを作成」ボタンを押下

2. 下記画面で以下の設定をして「次へ」進む
   ・ 信頼されたエンティティタイプ：AWS のサービス
   ・ ユースケース：Lambda
   

3. 「許可を追加」の画面で、今回利用するサービスのポリシーを追加し次へ進む
   ・TranslateFullAccess
   ・AmazonBedrockFullAccess
   ・AmazonS3FullAccess

4. 適当なロール名、説明を記入しロールを作成する

5. Lambda関数を作成

1. Lambdaのコンソール画面に移動し、「関数の作成」ボタンから作成画面に移る
2. 下記の設定で関数を作成する
   ・ ランタイム：Python 3.12
   ・デフォルトの実行ロールの変更＞実行ロール：既存のロールを使用する
   ・既存のロール：[4.Lambda用のIAMロールを作成]で作成したロールを選択
   ・その他：デフォルトの設定
3. タイムアウトを変更する
   設定タブ＞一般設定＞編集からタイムアウトをデフォルトの3秒から30秒に変更します
   ※3秒だと画像生成に時間がかかり、タイムアウトになるため
   
4. コードタブの入力欄に下記のコードをコピペし、下記の二点を変更する
   ・67行目で、bucketNameを[3. S3バケットを作成]で作成したバケット名に変更する
   ・101行目で、channelAccessTokenを下記の値に設定する
   → 8.LINE DevelopersでWebhookの設定で取得するチャネルアクセストークンの値
   ※8.LINE DevelopersでWebhookの設定が終わったあと設定してください。

import json
import requests
import boto3
import base64
from pprint import pprint
from botocore.exceptions import NoCredentialsError

translate = boto3.client('translate')
s3 = boto3.client('s3')

def lambda_handler(event, context):
    
    #=========================
    # LINEからのリクエスト解析
    #=========================
    bodyjson = json.loads(event['body'])
    messagetext = bodyjson['events'][0]['message']['text']
    replyToken = bodyjson['events'][0]['replyToken']

    # ログでLINEからのメッセージが受け取れているかCloudWatchで確認するために出力
    print("messagetext: " + messagetext)

    #=========================
    # 翻訳(自動検出言語→英語)
    #=========================
    response = translate.translate_text(
        Text=messagetext,
        SourceLanguageCode='Auto',
        TargetLanguageCode='en'
    )
    translated_text = response.get('TranslatedText')
    
    #=========================
    # Bedrockと接続して画像生成
    #=========================
    bedrock_runtime = boto3.client(service_name='bedrock-runtime', region_name="us-east-1")

    body = json.dumps({
        "taskType": "TEXT_IMAGE",
        "textToImageParams": {
            "text": translated_text,
        },
        "imageGenerationConfig": {
            "numberOfImages": 1,
            "height": 768,
            "width": 1280,
            "cfgScale": 7.5,
            "seed": 0
        }
    })

    response = bedrock_runtime.invoke_model(
        body=body,
        modelId="amazon.titan-image-generator-v2:0",
        accept="application/json",
        contentType="application/json"
    )
    
    #=========================
    # 生成された画像データを取得してS3にアップロード
    #=========================
    
    response_body = json.loads(response['body'].read())
    image_data = base64.b64decode(response_body['images'][0])
    
    # S3にアップロードするためにバケット名とキーを設定
    bucket_name = 'bucketName'
    file_name = 'generated_image.png'
    
    try:
        # S3に画像をアップロード
        s3.put_object(Bucket=bucket_name, Key=file_name, Body=image_data, ContentType='image/png')

        # 署名付きURLを生成 (有効期限は1時間)
        signed_url = s3.generate_presigned_url(
            ClientMethod='get_object',
            Params={'Bucket': bucket_name, 'Key': file_name},
            ExpiresIn=3600
        )

    except NoCredentialsError:
        print("S3のクレデンシャルが見つかりません。")
        return {
            'statusCode': 500,
            'body': json.dumps('S3クレデンシャルエラー')
        }
    
    #=========================
    # LINEへのレスポンス作成
    #=========================
    resmessage = [
        {
            'type': 'image',
            'originalContentUrl': signed_url,
            'previewImageUrl': signed_url
        }
    ]
    payload = {'replyToken': replyToken, 'messages': resmessage}
    
    # カスタムヘッダーの生成(dict形式)
    headers = {'content-type': 'application/json', 'Authorization': 'Bearer channelAccessToken'}
    
    # LINEにPOSTリクエストを送信
    r = requests.post("https://api.line.me/v2/bot/message/reply", headers=headers, data=json.dumps(payload))
    print("LINEレスポンス:" + r.text)
    
    return {
        'statusCode': 200,
        'body': json.dumps('Image processed and sent successfully.')
    }

6.requestライブラリをレイヤーに設定

1. ローカルPCの任意のディレクトリでコマンドプロンプトを起動し、下記コマンドを実行し、requestsライブラリを取得してpythonディレクトリに格納

       mkdir python
       pip install requests -t ./python
   

2. pythonディレクトリをZipに圧縮する

3. Lambdaのコンソール画面で左メニューの「レイヤー」を押下
   

4. 「レイヤーの作成」を押下し作成画面に移る
   

5. 先ほど取得したZipファイルをアップロードし、以下の設定をして「作成」ボタンを押下
   ・互換性のあるアーキテクチャ - オプション：x86_64
   ・互換性のあるランタイム - オプション：Python 3.12
   

6. 作成したLambdaの画面で、「コード」タブの下のほうの「レイヤー」＞「レイヤーの追加」を押下して追加画面に移る

7. 下記の設定で「追加」ボタンを押して追加
   ・レイヤーソース："カスタムレイヤー"
   ・カスタムレイヤー：先ほど作成したレイヤー

7. LambdaのトリガーとなるAPI Gatewayを設定

1. Lambdaのコンソール画面において、「関数の概要」＞「トリガーを追加」を押下
   
2. 下記の設定でトリガーを追加
   ・トリガー：API Gateway
   ・インテント：新規APIを作成
   ・APIタイプ：REST API
   ・セキュリティ：開く
   
3. 下記の画面のようにトリガーにAPI Gatewayが設定されていれば成功
   

8.LINE DevelopersでWebhookの設定

1. 上記画面からトリガーのAPI Gatewayをクリックし、作成されたAPI
   「URLを呼び出す」に記載されたURLをコピーする(画像の黒塗りの部分)
   
2. LINE Developersの画面で、作成したプロバイダの「チャネル基本設定」タグにおいて、LINE Official Account Managerに移動
   
3. 「Messaging APIの設定画面を開く」を押下する
   
4. 「Messaging APIを利用する」を押下する
   
5. 作成したプロバイダーを選択して、「同意」、「OK」と進み利用を開始する
   ※規約等のURLは未入力でも可
6. 「Webhook URL」に1でコピーしたURLを入力して保存する
   ※Webhook URLの検証をするとエラーになりますが、実際には画像生成可能です。Lambdaのコードで、lambda_handler関数の中身を空にしてreturnだけをするように修正してデプロイし、再度検証してOKであれば一旦は問題ないです
7. LINE Official Account Manager画面で、「設定」＞「応答設定」＞「応答機能」と移動し、「チャット」「Webhook」をONに、それ以外をOFFにする
   
8. LINE Developersの画面で、作成したプロバイダの「Messaging API設定」タブ＞「Webhook設定」＞「Webhookの利用」をONにする
   
9. LINE Developersの画面で、作成したプロバイダの「Messaging API設定」タブ＞「チャネルアクセストークン」で「発行ボタン」を押下して表示されたチャネルアクセストークンコピーする

9.Lambdaを修正してデプロイ

1. 8.LINE DevelopersでWebhookの設定での101行目のチャネルアクセストークンの値に上記でコピーした値を入力する
2. デプロイする

10.動作確認

1. LINEで作成してほしい画像のプロンプトを日本語で入力する
2. しばらくしたら画像が表示される(10秒前後)
   
3. 日本語以外のテキストを入力しても、画像が生成されます(例：アイスランド語)
   

最後に

とにかく早く作ってみる、ということを重視したのでコードの内容やセキュリティ面を考慮した諸々の設定、ロールなどによる権限回りなどは見直す必要があり、作っている最中でも要勉強だなと感じることが多かったです。
しかし、思ったよりも簡単に作成できたことに驚きと喜びと達成感を感じることができたのでよかったです。

参考

• https://developers.line.biz/ja/reference/messaging-api/#response
• https://docs.aws.amazon.com/ja_jp/bedrock/latest/userguide/model-parameters-titan-image.html