AWSサービスを組み合わせてLINEチャットボットを作ってみる

はじめに

LINEのチャットボットを作ってみました。Bedrockをはじめ、あれこれドキュメントを読み込むより実際に作ることで理解が深まると思うのでやってみましたという記事になります。

全体構成確認

Amazon API Gatewayを通してBedrockとLINE上でやり取りできるという簡単なものにしました。なので、ネット検索やツールの利用はできません。もちろん返ってくる回答の精度も基盤モデルに依存してしまいます。

構成図、処理の流れのシーケンス図は下記のようになります。

全体のシーケンス図

①【ユーザー操作】

ユーザーがLINEボットにメッセージを送る
・例:「Amazon Bedrockについて教えて」などのように送信
・この時点で、LINEのサーバーがWebhookのURLに通知を送信
※Webhookについては後ろの方で解説しようと思います。

②【LINE→API Gateway】

ユーザーがLINEボットにメッセージを送る
・例:「Amazon Bedrockについて教えて」などのように送信
・この時点で、LINEのサーバーがWebhookのURLに通知を送信
※Webhookについては後ろの方で解説しようと思います。

③【API Gateway】

API GatewayがLambdaを呼び出す
ここでAPI GatewayからLambdaを呼び出します。

④【Lambda起動】

Lambda関数が起動し、Webhookイベントを受け取る
・例:「Amazon Bedrockについて教えて」などのように送信
・この時点で、LINEのサーバーがWebhookのURLに通知を送信

⑤【Bedrock】

Claudeにメッセージを送って応答を取得
・Claude 3 Messages API を使って、ユーザーのメッセージを送信。
・応答が途中で切れていたら、“続けてください” を追加して再度呼び出し。
・結果を ai_text として取得。

⑤【LINE Messaging APIで返信】

Lambda から LINE Reply API を呼び出して返信
・APIを呼び出して自動で返信される

LINE周りの処理を確認する

Messaging APIとは

Line公式ドキュメントもありますが、簡単にいうと、LINE公式アカウントをより便利に使うためのAPIです。
様々な機能がありますが、今回はメッセージ送信機能を使用します。

利用料金

基本的にMessaging APIの利用は無料になっています。

ですが、無料プランは送信できる件数に上限が決まっています。上限を超えてメッセージを送信したい場合、プランの変更が必要になりますので、詳しくは下記の料金プランをご覧ください。

LINE Developers コンソール上での設定

次に、やり取りをする上で必要な公式アカウントの作成やMessaging APIの有効化を行います。

LINE公式アカウント(ビジネスアカウント)の作成

LINE Developersにアクセスし、コンソールにLINEアカウントを使用してログインを行います。

ここで設定が必要なものは以下になります。

1.LINE Developers コンソールで新規プロバイダーを作成
2.新しいチャネル（Messaging API） を作成
3.LINE Access Token を取得
4.Webhook URL を後で設定（Amazon API Gateway の URL）
5.Webhook の利用を「オン」にする

1と2に関しては、ログイン後画面遷移に従えばすぐできると思いますので、3から書いていこうと思います。

LINE Access Token を取得

LINE Access Tokenは作成したプロバイダーのMessaging API 設定欄から取得できます。

Webhook URLを設定する

次にAPI GatewayとやりとりするためのWebhook URLを記述します。
ここには、Amazon API Gatewayをデプロイ時に取得できるURLを入力します。
「Webhookを利用」をONにすることで利用できますので、こちらもONにしておきます。
また、「検証」ボタンを押すことで動作確認を行うこともできます。

Amazon Bedrockの準備をする

利用するモデルを有効化する

まずAmazon Bedrockを利用するにあたり、モデルへのアクセスを有効化しておく必要があります。
モデルアクセスはマネージドコンソール上から実施できます。
もし有効になってない場合はこの段階で有効にしておきましょう。

今回は以下のモデル、リージョンで実施しました。
・リージョン：バージニア北部(us-east-1)
・モデル：Claude 3.5 Sonnet
・モデルID：anthropic.claude-3-5-sonnet-20240620-v1:0
その他のモデルの対応リージョン、モデルIDについては下記のドキュメントにまとめられていますので参照ください。

Lambdaの作成

Bedrockに対する操作をできるように、以下のロールを付与します。
・AmazonBedrockFullAccess

また、Lambdaは以下のバージョンで実施しています。
・Python 3.10

記述したコードは以下になります。

# lambda_function.py
import os
import json
import logging
import urllib.request
import urllib.error
import boto3

logger = logging.getLogger()
logger.setLevel(logging.INFO)

# ===== Settings =====
BEDROCK_REGION = os.environ.get("BEDROCK_REGION", "us-east-1")
MODEL_ID = os.environ.get("BEDROCK_MODEL_ID", "anthropic.claude-3-5-sonnet-20240620-v1:0")
LINE_ACCESS_TOKEN = os.environ["LINE_ACCESS_TOKEN"]
HTTP_TIMEOUT = float(os.environ.get("HTTP_TIMEOUT", "5"))
LINE_MAX = 5000

# ===== Clients =====
bedrock = boto3.client("bedrock-runtime", region_name=BEDROCK_REGION)

def lambda_handler(event, context):
    try:
        body = json.loads(event.get("body") or "{}")
        ev = (body.get("events") or [None])[0] or {}
        reply_token = ev.get("replyToken")
        user_text = ((ev.get("message") or {}).get("text") or "").strip()
        if not reply_token or not user_text:
            return _ok({"message": "no text"})

        # Claude 3.5 メッセージ形式で1回だけ呼び出し
        ai_text = _ask_bedrock(user_text)
        _reply_line(reply_token, ai_text[:LINE_MAX])
        return _ok({"message": "ok"})
    except Exception as e:
        logger.exception("Unhandled")
        return {
            "statusCode": 500,
            "headers": {"Content-Type": "application/json"},
            "body": json.dumps({"error": str(e)})
        }

def _ask_bedrock(user_text: str) -> str:
    payload = {
        "anthropic_version": "bedrock-2023-05-31",
        "max_tokens": 200,
        "messages": [
            {"role": "user", "content": [{"type": "text", "text": user_text}]}
        ],
    }
    resp = bedrock.invoke_model(
        modelId=MODEL_ID,
        accept="application/json",
        contentType="application/json",
        body=json.dumps(payload),
    )
    data = json.loads(resp["body"].read())
    content = data.get("content") or []
    return (content[0].get("text") if content and isinstance(content[0], dict) else "（応答なし）")

def _reply_line(reply_token: str, text: str):
    url = "https://api.line.me/v2/bot/message/reply"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {LINE_ACCESS_TOKEN}",
    }
    payload = {
        "replyToken": reply_token,
        "messages": [{"type": "text", "text": text}]
    }
    req = urllib.request.Request(
        url, data=json.dumps(payload).encode("utf-8"),
        headers=headers, method="POST"
    )
    with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT) as resp:
        _ = resp.read()

def _ok(payload: dict):
    return {
        "statusCode": 200,
        "headers": {"Content-Type": "application/json"},
        "body": json.dumps(payload),
    }

また、下記のimport文で記述している「request」はレイヤーを設定しないと動作しませんでした。
Lambdaレイヤーの設定はZipファイルの設定かARBの記述でできます。今回はARNを記述することで設定しました。

import os, json, logging, urllib.request, urllib.error, boto3

レイヤーを指定する際は、Lambdaのコード画面の下の方に設定箇所があります。
今回は「ARNを指定する」を選択し、ARN直接入力すればOKです。

また、ARNはどこでわかるんだよって話ですが、下記のライブラリに記述されています。
「List of ARNs」 から対象のランタイムバージョンを選択し、対象の Lambda 関数のリージョンを選択し、利用したいライブラリの ARN を探します。

ARN がメモできたら、マネジメントコンソールからレイヤーを設定したい Lambda 関数のページのレイヤー設定から「レイヤーを追加」します。

レイヤーを選択する画面に遷移するので、「ARN を指定」-> メモした ARN を入力すれば、利用可能になります。

API Gatewayの設定

次にAPI Gatewayを設定していきます。新規に作成し、REST APIを選択します。

次にメソッドの作成をしていきます。

メソッドタイプはPOST、
統合タイプはLambda関数
そして、Lambda関数は作成した関数と、リージョンを指定します。

あとはこのメソッドを「新しいステージ」でデプロイします。
デプロイできたら、Lambda関数の画面を確認します。
以下のようにAPI Gatewayとの統合が反映されていれば、関数との紐付けはできています。
また、デプロイ後取得したURLをWebhook URLとしてLINE Developers コンソールにて登録しておきます。

アプリの動作確認

ここまでの手順で最低限の動作はするようになっています。
最後に動作確認として、メッセージを送ってみましたが正常に反応してくれてました。