[Amazon BedrockとSlackで作る生成AIチャットボット] アップロードした画像ファイルを解析してくれるチャットボットを作る

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


sqs_queue_url = os.environ.get("SQS_QUEUE_URL")
slack_bot_token = os.environ.get("SLACK_BOT_TOKEN")
slack_signing_secret = os.environ.get("SLACK_SIGNING_SECRET")

sqs = boto3.client("sqs")


# アプリの初期化
app = App(
    token=slack_bot_token,
    signing_secret=slack_signing_secret,
    process_before_response=True,
)


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

    files = event.get("files")
    if files is None:
        say("画像ファイルが添付されていません。")
        return

    file_type = files[0]["mimetype"]
    if file_type not in ["image/jpeg", "image/png", "image/gif", "image/webp"]:
        say("これは画像ファイルではありません。")
        return

    say("少々お待ちください...")
    file_url = files[0]["url_private_download"]
    sqs.send_message(
        QueueUrl=sqs_queue_url,
        MessageBody=json.dumps({
            "channel_id": channel_id,
            "input_text": input_text,
            "file_type": file_type,
            "file_url": file_url,
        }),
    )


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

		

			
			
{
    "type": "app_mention",
    "channel": "Cxxxxxxxxxx",
    "text": "<@Uxxxxxxxxxx> これは何の絵ですか？",
    "files": [
        {
            "name": "image1.jpg",
            "title": "image1.jpg",
            "mimetype": "image/jpeg",
            "url_private_download": "https://files.slack.com/files-pri/XXXXXXXXX-XXXXXXXXXXX/download/image1.jpg",
            (省略)
        }
    ],
    (省略)
}

		

			
			
from slack_sdk import WebClient
import base64
import boto3
import json
import os
import urllib.request


slack_bot_token = os.environ.get("SLACK_BOT_TOKEN")

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

client = WebClient(token=slack_bot_token)


# Slackから画像をダウンロードする
def download_image(image_url):
    # URLとトークンを与えてダウンロードを行う
    request = urllib.request.Request(
        url=image_url,
        headers={"Authorization": f'Bearer {slack_bot_token}'},
    )
    response = urllib.request.urlopen(request)

    # レスポンスから画像データを取り出す
    body = response.read()

    # 画像を戻り値として返す
    return body


# Bedrockを使って応答テキストを生成する
def generate_answer(input_text, media_type, image_data):
    # 入力テキストが空の場合、固定のテキストを設定する
    if input_text == "":
        input_text = "この画像に描かれているものを教えてください。"

    # メッセージをセット
    messages = [
        {
            "role": "user",
            "content": [
                {
                    "type": "image",
                    "source": {
                        "type": "base64",
                        "media_type": media_type,
                        "data": base64.b64encode(image_data).decode("utf-8"),
                    },
                },
                {
                    "type": "text",
                    "text": 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")
    file_type = body.get("file_type")
    file_url = body.get("file_url")

    # Slackから画像をダウンロードする
    image_data = download_image(file_url)

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

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

		

			
			
{
    "anthropic_version": "bedrock-2023-05-31",
    "max_tokens": ＜最大トークン数＞,
    "messages": [
        {
            "role": "user",
            "content": [
                {
                    "type": "image",
                    "source": {
                        "type": "base64",
                        "media_type": "＜画像のMIMEタイプ (例: image/jpeg)＞",
                        "data": "＜Base64エンコードされた画像データ＞"
                    }
                },
                {
                    "type": "text",
                    "text": "＜プロンプトテキスト＞"
                }
            ]
        }
    ]
}