生成AIの問い合わせ自動化をAWSで実装してみた

①はじめに

本記事の概要

本記事ではAWSサービスを利用して、生成AIに定型的な問い合わせを自動実行するアプリケーションを実装しました。
ポイントは以下の3点となります。

• ほぼ定型的なプロンプトの問い合わせは、AWSサービスを使用して定期実行またはイベント駆動により自動化できる
• CloudWatchLogsで出力されたログをインプット情報として、プロンプトに引き渡すことができる。
• 問い合わせ結果はLINEに送信することができ、比較的手軽に確認することができる

執筆背景

生成AIが情報検索や整理のツールとして浸透している一方で、「毎回プロンプトを入力するのが面倒」「プロンプトを考えること自体が面倒」などの悩みが増えてきていると思います。
例えば以下のようなケースだと、入力する内容は決まっているのに、毎回手動やコピペによる入力と結果出力の待機で手間がかかってしまうことが考えられます。

• 毎朝情報収集のためにリアルタイム検索の生成AIを使用している
  →毎回同じ問い合わせを入力する必要があり、結果出力にも時間がかかる。

• システムの出力したエラーログの内容が分からないからとりあえず生成AIに投げてみる
  →「このエラーの原因と対策を教えて」というようなほぼ固定化された文章だが、ログの文言は毎回書き換えて入力する必要がある。

極論としては「欲しい情報を生成AIが勝手に出力してくれる」という構成ができないかと考えました。そこで今回は、上記のような定期実行/イベント駆動にフォーカスした"部分的な"問い合わせの自動化を実装しました。
常日頃からAWSを利用している人やアプリ開発している人からすると比較的基本的な内容が多いかもしれません。
なので、対象読者としてはIT初学者の方や非エンジニアの方が興味を持ってもらえるようなレベル感の内容となっています。

今回開発したアプリケーションについては、最終的に業務利用することで作業効率化を高められるような目標の元で構想しました。
しかしながら、実際に業務利用する場合や実システムに導入したい場合は、より堅牢なセキュリティ設計を取り込んだり、各システムや社内のAI利用方針などに準じる必要があります。
本記事の構成は実業務での経験を元にしたものではなく、個人開発環境で実装した内容の共有のため、もし本記事を参考にされる場合は上記の点ご留意ください。

構成図と開発STEP

今回実装したアプリケーションは下記のSTEPに分ける方針で開発しました。一気に全てを開発しようとすると、想定通りに動かなかった場合の原因切り分けに苦戦してしまうことがあるので、必要となる機能を段階的かつ正確に実装するためです。

• STEP1. Lambda→Bedrockの生成AIモデルに問い合わせを行う処理を実装
• STEP2. 出力結果を利用者のLINEアカウントに通知する処理を実装
• STEP3. EventBridgeによる定期実行を設定
• STEP4. CloudWatchLogsのサブスクリプションフィルターによるイベント駆動を設定

構成図は以下の通りになります。

本構想の中枢となるサービスとしてAWS LambdaとAmazon Bedrockを選択しました。
LambdaはAWS上にてサーバレスでコードの開発/実行ができるサービスで、BedrockはAWS上でAPIを介して生成AIモデルに問い合わせを行うサービスです

Lambdaの出力についてはLINEのMessaging APIを使用して、個人のLINEアカウントに直接送信する方式を採用しました。
業務利用に導入するのであればメールやSlackに送信するケースが多いと思いますが、日常利用しているツールに送信して手軽に確認できるという点を認識してもらい、非エンジニアの読者の方にとっても面白味の感じる構成にしようと考えました。

まとめると今回の実装としては、Lambdaでプロンプトを作成してBedorockに問い合わせるスクリプトを実行し、その応答を外部サービス(LINE)に連携する、という全体構成になっています。

自動実行のトリガー

上述のSTEP3では「定期実行処理」、STEP4では「イベント駆動処理」のトリガーとなる実装を行いました。冒頭に記載したユースケースを元にすると、それぞれ以下の表のような構成で実現できます。

#	定期実行処理	イベント駆動処理
ユースケース	毎朝、顧客業界に関するニュースを収集するため、自動的にニュースの内容を要約して出力することで、情報収集の手間を削減する。 　※リアルタイム検索が可能な 　　生成AIモデルの利用が必須。	障害発生時に出力されたログを問い合わせて、調査すべきポイントや実行すべきコマンドを得ることで、障害対応までの時間を削減する。
利用サービス	Amazon EventBridge	Amazon CloudWatch Logs
トリガー	事前に設定した曜日/日時になったとき	特定の文言を含むログが出力されたとき

②実装

STEP1：Lambda→Bedrockの生成AIモデルに問い合わせを行う処理を実装

▼スクリプト

早速、LambdaからBedrockに対して問い合わせを行う処理を実装します。関数のランタイム(言語)はPython3.14を選択しました。
本来ならば任意の問い合わせ文を引数として渡すことがスマートですが、現段階では問い合わせ文をスクリプト内に記載しています。

import json
import boto3

# Bedrockクライアントの初期化
## boto3はPythonからAWSサービスを操作するライブラリ
bedrock = boto3.client("bedrock-runtime", region_name="us-east-1")

def lambda_handler(event, context):
    # 問い合わせする文章を入力
    question = "北海道のおすすめ観光スポットを教えてください。"

    # モデルIDをテキストで指定（今回は Amazon Titan Text Lite v1）
    model_id = "amazon.titan-text-lite-v1"

    # Titanモデル用のリクエストボディ
    body = {
        "inputText": question,
        "textGenerationConfig": {
            "maxTokenCount": 300,
            "temperature": 0.7,
            "topP": 1.0,
            "stopSequences": []
        }
    }

    # Bedrockモデルへの問い合わせ&応答を受け取る
    ## modelId:使用する生成AIモデルの指定
    ## contentType:リクエスト本文の形式(今回はJSONに指定)
    ## accept:レスポンスの形式(contentTypeと同じくJSONに指定)
    ## body:上記で指定したリクエスト本体
    response = bedrock.invoke_model(
        modelId=model_id,
        contentType="application/json",
        accept="application/json",
        body=json.dumps(body)
    )

    # 応答の抽出と整形
    ## レスポンスをPythonデータに戻す
    ## レスポンスから回答を取得する
    response_body = json.loads(response["body"].read())
    answer = response_body.get("results", [{}])[0].get("outputText", "回答が取得できませんでした。")

    return {
        "statusCode": 200,
        "body": json.dumps({"question": question, "answer": answer}, ensure_ascii=False)
    }

▼Lambdaの設定

生成AIへの問い合わせを行うにあたり、必要となる設定を解説します。

• Bedrockに対する操作を許可するロール設定
  → Bedrockは完全AWSマネージドサービスであり、新規にリソースを作成したりする作業は発生しません。しかし、Lambda関数がBedrockへアクセスする操作を許可するために、LambdaのIAMロールに対して権限追加が必要です。
  具体的には以下の図のようなAllow権限を付与すれば実装できます。7行目ではbedrockに対する利用権限を付与しており、8行目では利用するモデルを指定しています。（今回はus-east-1リージョンのtitanを使用しているので、そのARNを指定しています。）
  

• タイムアウト時間の延長
  → Lambdaで実行する関数のタイムアウト値はデフォルトで3秒となっていますが、生成AIの回答出力は時間を要するため、最低でも10秒以上に設定した方が良いです。
  タイムアウト値はLambda関数の「設定」タブ→「一般設定」から変更します。（今回は1分に設定しました。）
  

▼Bedrockのモデル設定

Bedrockでは数多くの生成AIモデルを選択することが可能です。
スクリプト内ではmodel_idで生成AIモデルを選択していますが、以下のページに記載されている一覧から任意のものを選択することができます。
Amazon Bedrock でサポートされている基盤モデル - Amazon Bedrock
今回は個人開発で手軽に利用でき、料金もかなり安いという理由から「Titan」というモデルを選択しました。皆様は出力したいアウトプットや、モデルの特徴等を調べて選択いただければと思います。

body内に指定してる各パラメータは、生成AIモデルに応じて変わります。
詳細はAWSのドキュメントに記載されているので、各モデルのページからご確認ください。以下は今回選択したTitanのドキュメントです。
Amazon Titan Text モデル - Amazon Bedrock
主にモデルの精度チューニングや、出力の最大トークン数（≒単語数）を指定できます。

bedrock.invoke_modelのパラメータはスクリプト内にコメントアウトで注釈しましたが、公式のドキュメントとしては以下に記載があるのでご参照ください。
InvokeModel で 1 つのプロンプトを送信する - Amazon Bedrock

▼結果確認

以下はスクリプトを実行したレスポンスです。関数の返り値として、きちんと生成AIモデルからの回答が得られることを確認できました。

Response:
{
  "statusCode": 200,
  "body": "{\"question\": \"北海道のおすすめ観光スポットを教えてください。\", 
  \"answer\": \"\\nBot: いいですね！\\
  n１．登別マリンパーク - 冬になると世界で最も北に位置する马雪山で、雪山に囲まれた天然温泉「登別温泉」で有名です。\\
  n２．登別カルルス温泉 - 北海道で最も歴史がある温泉の一つです。\\
  n３．屈斜路湖 - 湖水地方最大級の湖。マタギの里とも呼ばれていて、日本最古の地苔原遺跡が水田とともに残っています。\\
  n４．登別マリンワールド - １０万平方メートルの水族館です。\\
  n５．十勝ヒルズ - １万平方メートルにも及ぶ森林公園で、北海道最初のスキー場「十勝トンネルスキー場」があります。\"}"
}

回答内容としては実際と異なっている箇所や、独特な表現が含まれてるように見受けましたが、Titanの特徴かもしれませんね。
どの生成AIモデルにも強み・弱みといった箇所があるので、ユースケースに応じてしっかり選択するとともに、回答の妥当性は自分で判断できるようにすることが大切です。
今回のSTEPのポイントは「API経由で生成AIから何かしらの回答が得られること」であり、本目的を果たすことはできたので次のSTEPに進みます。

余談：回答内容の妥当性判断①

　・「マタギの里」で検索すると出てくるのは秋田県。
　・「湖水地方」「地苔原遺跡」という単語は存在しない？
　・「登別マリンワールド」という施設は存在せず、1番の登別マリンパークと同じ？
　・十勝ヒルズの広さは23万平方メートル。
　・北海道で最初のスキー場は明確な情報が無いが、札幌付近が有力。
　 （ちなみに「十勝トンネルスキー場」という場所は存在しない。）

STEP2：出力結果を利用者のLINEアカウントに通知する処理を実装

▼LINE Developerの環境構築

「LINEへ送信の構成」の実装は正確には、個人で公式LINEアカウントを作成し、そこのトーク画面にLINEmessagingAPIを介して回答を送信することで実現します。
こちらの手順はLINEの公式ドキュメントが公開されているので、公式アカウント作成とMessagingAPIの有効化を実施してください。
Messaging APIを始めよう | LINE Developers
私は適当に「生成AI発信bot」という名前で公式アカウントを作成しました。

※最初にアカウントを作ったときにタイプミスしていたため、このタイミングではアカウント名が「生成AI発信bto」になっていました…。

▼Lambdaの設定

次はLambdaで以下の2つのパラメータを環境変数に登録します。

• 送信先LINEアカウントのユーザID
• 送信元となる公式LINEアカウントに対するアクセストークン

スクリプトにベタ貼りするのはセキュリティ上望ましくないので、AWS側に暗号化して持たせるため環境変数に設定しました。

両パラメータの場所はLINE Developerから確認します。詳細は折りたたみます。

パラメータの確認方法

• 送信先LINEアカウントのユーザID
  → 本実装では公式LINEアカウントから、特定のLINEアカウントへメッセージを送るので、
  　宛先情報として送信先LINEアカウントのユーザIDを指定する必要があります。
  　（公式アカウントが「どのアカウント宛に送る？」を指定してあげる。）
  　これはLINEアカウントを作成したときに個人で設定した値ではなく、
  　LINE側で発行しているユーザ側では指定できない一意な値となります。
  　自分のユーザIDはLINE Developersから確認することができます。
  
• 送信元となる公式LINEアカウントに対するアクセストークン
  → LINEmessagingAPIを利用するための認証キーとも言える、
  　公式LINEアカウントのチャネルアクセストークンの値を登録します。
  　こちらもLINE Developersにて確認することができます。
  

▼スクリプト

STEP1で作成したスクリプトをベースに、LINEへの送信処理を追加しました。
（追加箇所はシンタックスハイライトで強調しています。その行の先頭の"+"は実際に入力するとエラーになってしまうので、コピペする際はご注意ください。）

import json
import boto3
+ import os
+ import urllib.request

# Bedrockクライアントの初期化
bedrock = boto3.client("bedrock-runtime", region_name="us-east-1")

+ # LINE環境変数の設定
+ ## Lambdaで設定した環境変数から読込み
+ LINE_CHANNEL_ACCESS_TOKEN = os.environ["YOUR_LONG_LIVED_CHANNEL_ACCESS_TOKEN"]
+ USER_ID = os.environ["TARGET_USER_ID"]

def lambda_handler(event, context):
    # 問い合わせする文章を入力
    question = "北海道のおすすめ観光スポットを教えてください。"

    # モデルIDをテキストで指定（今回は Amazon Titan Text Lite v1）
    model_id = "amazon.titan-text-lite-v1"

    # Titanモデル用のリクエストボディ
    body = {
        "inputText": question,
        "textGenerationConfig": {
            "maxTokenCount": 300,
            "temperature": 0.7,
            "topP": 1.0,
            "stopSequences": []
        }
    }

    # Bedrockモデルへの問い合わせ&応答を受け取る
    response = bedrock.invoke_model(
        modelId=model_id,
        contentType="application/json",
        accept="application/json",
        body=json.dumps(body)
    )

    # 応答の抽出と整形
    response_body = json.loads(response["body"].read())
    answer = response_body.get("results", [{}])[0].get("outputText", "回答が取得できませんでした。")

+   # LINEに送信するメッセージ
+   message_text = f"質問: {question}\n回答: {answer}"

+   # LINEmessagingAPIのURLを格納
+   url = "https://api.line.me/v2/bot/message/push"

+   #LINEへのリクエストヘッダーにトークンを指定
+   headers = {
+       "Content-Type": "application/json",
+       "Authorization": f"Bearer {LINE_CHANNEL_ACCESS_TOKEN}"
+   }
+   #LINEへのメッセージボディに送信先ユーザとメッセージを指定
+   line_body = {
+       "to": USER_ID,
+       "messages": [
+           {
+               "type": "text",
+               "text": message_text
+           }
+       ]
+   }
+   # HTTPリクエストのオブジェクトを作成(URL,ヘッダー,ボディ)
+   req = urllib.request.Request(
+       url,
+       data=json.dumps(line_body).encode("utf-8"),
+       headers=headers,
+       method="POST"
+   )
+   # リクエストの送信
+   with urllib.request.urlopen(req) as res:
+       line_response_body = res.read().decode("utf-8")
+       print("LINE Status:", res.status)
+       print("LINE Response:", line_response_body)

    # Lambdaの戻り値としても返す
    return {
        "statusCode": 200,
        "body": json.dumps({"question": question, "answer": answer}, ensure_ascii=False)
    }

▼結果確認

Lambda関数を実行した結果、出力がLINEへ送られてくることを確認できました。

Titanは登別マリンパークニクス推しみたいですね。
これでLambdaから実行した問い合わせの回答をLINEで手軽に見られるようになったので、STEP3ではこのLambda関数を自動実行させましょう。

余談：回答内容の妥当性判断②

　・登別マリンパークニクスは水族館で、マリンスポーツとかはできない。
　・釧路湿原は世界遺産に登録されていない。
　・洞爺湖は冬に湖面が真っ黒になったりはしない。
　・洞爺湖は不凍湖なので、湖面でスノーモービルなんてことはできない。
　　（そもそも湖面でモービル乗るのは危ない。）

STEP3：EventBridgeによる定期実行を設定

▼Lambdaのトリガー設定

ここの設定は非常に簡単です。
Lambda関数の画面から「トリガーを追加」をクリックし、「EventBridge(CloudWatch Events)」を選択します。
ここでルールタイプを「スケジュール式」にするとcron表記で周期を指定できます。
cronはLinux/UNIX系OSでタスクを自動実行させるのに使用するスケジューリング機能です。その時に指定する日時の表記ルールを、AWS上で設定することになります。
cron 式と rate 式を使用して Amazon EventBridge でルールをスケジュールする - Amazon EventBridge
「日本時間(JST)で毎朝7時」のスケジュールは以下となります。

cron(0 22 * * ? *)

分、時、日、月、曜日、年の順で指定します。
時間はUTCで認識されるので、JSTから9時間前の時間にする必要があります。
（今回はJSTの7時から9時間前になるので、UTCだと22時となります。）
AWSコンソールの設定画面は以下のようになります。

▼結果確認

北海道のおすすめスポットにも飽きてきたと思うので、質問の内容を書き換えました。
毎朝7時にLambda関数を実行するので、「今日のてんびん座の運勢を占って」としました。
その結果、きちんと時間通りにLINEが自動的に送られてくることを確認できました。

いい感じの回答ですが、Titanは個性溢れる回答が多かったので、何度か試した中で一番本当の占いっぽい結果を添付しました。
色々なパターンの回答があったので、ご興味ある方は以下の余談を参照ください。
（「占い」なので結果の妥当性や根拠は特に無いと思います。）

余談：Titanによる占いの回答集

• おみくじ
  　
  　非常に簡素な回答でした。最近リアルでは凶を引いてきたので、吉で嬉しいです。

• おうし座は得意なTitan
  　
  　英語で回答が返ってきてしまいました。
  　リアルタイム検索はできないよ、でもTitanはおうし座の特徴は返せるよ、という内容。
  　おうし座の人はラッキーですね。

• エンドレス
  　
  　同じ文言がずっと続いていました。もはや怖かったです。
  　謎の文字の部分はヒンディー語で「休息」という意味みたいでした。

STEP4：CloudWatchLogsのサブスクリプションフィルターによるイベント駆動を設定

▼CloudWatchLogsの設定

CloudWatchLogsのサブスクリプションフィルターという機能を使うと、ログに特定の文字列が含まれた場合に、ログを別のAWSサービスに渡すことができます。
今回はログの中に「ERROR」という文字列が含まれた時に、この後作成するLambda関数にそのログを渡すようロググループを設定しました。

• フィルターパターン
  → ログ内で検知させる文字列を指定します。
• 送信先のARN
  → フィルターパターンに一致した場合、ログを転送する宛先となるリソースを指定します。（今回だと作成しているLambda関数を指定。）

今回はCloudWatchLogs上でログイベントを作成してテストしていきます。

開発時にこのロググループに対して、指定したログを転送するようEC2側の設定をしましたが、この実装方法まで記載するとボリュームが非常に多くなってしまうため割愛します。（冒頭の構成図にEC2を記載しているのもこれが理由です。）
以下の公式ドキュメント等を参考に実装したので、気になる方はこちらをご参照お願いいたします。
CloudWatch エージェントを使用してメトリクス、ログ、トレースを収集する - Amazon CloudWatch

▼スクリプト

STEP2のスクリプトを修正します。ポイントは以下の2点です。

• ログを引数として渡す
  → 今まではスクリプト内に直接質問文を記載していましたが、今回はCloudWatchLogsで出力されたログを質問文に組み込むよう修正します。
• ログのデコード処理
  → CloudWatchLogsからLambdaに送られてくるログデータは、AWS側の処理によりgzipで圧縮されてbase64でエンコードされた状態になっています。そのため、Lambdaに読み込ませるためには、デコードと解凍を行う処理が必要となります。
  　生ログ：文字列 →(gizp)→ 圧縮ログ：バイナリ →(エンコード)→ 圧縮ログ：文字列

import json
import boto3
import os
import urllib.request
+ import base64
+ import gzip

+ #CloudWatchLogsから送られてきたログをデコード・解凍する関数を定義
+ def decode_function(event):
+     decoded_message = base64.b64decode(event['awslogs']['data'])
+     decompressed_data = gzip.decompress(decoded_message)
+     json_data = json.loads(decompressed_data)
+     log_message = json_data['logEvents'][0]['message']
+     return log_message

# Bedrockクライアントの初期化
bedrock = boto3.client("bedrock-runtime", region_name="us-east-1")

# LINE環境変数の設定
LINE_CHANNEL_ACCESS_TOKEN = os.environ["YOUR_LONG_LIVED_CHANNEL_ACCESS_TOKEN"]
USER_ID = os.environ["TARGET_USER_ID"]

def lambda_handler(event, context):

+ 	#定義したdecode_functionを使用してログをデコードする
+     log_message = decode_function(event)

+ 	#log_messageの対応方法について質問する文を作成
+     question = (
+     f"以下のログについて、**出力された原因**と**調査手順**を教えて。"
+     f"ログ:\n{log_message}"
+     )

    # モデルIDをテキストで指定（今回は Amazon Titan Text Lite v1）
    model_id = "amazon.titan-text-lite-v1"

    # Titanモデル用のリクエストボディ
    body = {
        "inputText": question,
        "textGenerationConfig": {
            "maxTokenCount": 300,
            "temperature": 0.7,
            "topP": 1.0,
            "stopSequences": []
        }
    }

    # Bedrockモデルへの問い合わせ&応答を受け取る
    response = bedrock.invoke_model(
        modelId=model_id,
        contentType="application/json",
        accept="application/json",
        body=json.dumps(body)
    )

    # 応答の抽出と整形
    response_body = json.loads(response["body"].read())
    answer = response_body.get("results", [{}])[0].get("outputText", "回答が取得できませんでした。")

    # LINEに送信するメッセージ
    message_text = f"質問: {question}\n回答: {answer}"

    # LINEmessagingAPIのURLを格納
    url = "https://api.line.me/v2/bot/message/push"

    #LINEへのリクエストヘッダーにトークンを指定
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {LINE_CHANNEL_ACCESS_TOKEN}"
    }
    #LINEへのメッセージボディに送信先ユーザとメッセージを指定
    line_body = {
        "to": USER_ID,
        "messages": [
            {
                "type": "text",
                "text": message_text
            }
        ]
    }
    # HTTPリクエストのオブジェクトを作成(URL,ヘッダー,ボディ)
    req = urllib.request.Request(
        url,
        data=json.dumps(line_body).encode("utf-8"),
        headers=headers,
        method="POST"
    )
    # リクエストの送信
    with urllib.request.urlopen(req) as res:
        line_response_body = res.read().decode("utf-8")
        print("LINE Status:", res.status)
        print("LINE Response:", line_response_body)

    # Lambdaの戻り値としても返す
    return {
        "statusCode": 200,
        "body": json.dumps({"question": question, "answer": answer}, ensure_ascii=False)
    }

▼結果確認

CloudWatchLogsに作成したログストリームへ、以下のように「ERROR」の文字を含むメッセージを入力してみます。

すると、LINEに解答が届くことを確認できました。

きちんとそれっぽい回答が返ってきましたね。
今回は無理やり「ERROR」を含む文言にしましたが、エラーを表現する文言は製品やログ設計によって異なるので、導入する際は実際に出力されるログから条件を抽出して、サブスクリプションフィルターに設定する文言を決定しましょう。

③まとめ

本記事のまとめ

本記事のポイントをおさらいすると以下の通りになります。

• LambdaとBedrockにより、生成AIへの問い合わせをプログラム内に組み込める
• LINEmessagingAPIにより、生成AIの回答をLINEの公式アカウントから送信できる
• EventBridgeにより、生成AIへの定期的な自動問い合わせができる
• CloudWatchLogsにより、ログ出力を契機とした自動問い合わせができる

今回の構成はかなり基本的な実装となりましたが、LINE以外の外部サービスに連携する、リアルタイム検索が可能な生成AIモデルを使用する、出力に対する追加問い合わせを行う、といった機能を実装することができればより様々な場面で活用できると思います。本記事の内容が皆様の業務に役立てられるような記事となるば幸いです。
ちなみに、本検証は平日夜間に2週間ほど行いましたが、私の実施したケースでは利用料金としては1$にも満たないくらいの金額でした。

ご精読ありがとうございました。