生成AIを活用した環境構築をやってみた　第4回：蓄積されたデータを基に旅行プランを生成する(音声ファイル編)

Last updated at 2025-05-12Posted at 2025-05-12

はじめに

「生成AIを活用した環境構築をやってみた」の第4回記事です。

第3回の記事では、DynamoDBに登録された画像の要素を基にhtml形式で旅行プランを生成してもらう仕組みを作成しました。

今回の記事では音声ファイルの要素を基に旅行プランを考案してもらう仕組みを作っていきます。

作業のゴール

今回は「音声ファイルから生成された旅行プランの確認」をゴールとします。

S3バケットへ音声ファイルをアップロードすることにより、Amazon Transcribeを利用して文字起こしを行います。文字起こしされたテキストはS3バケットへ格納され、そのテキストを基にAmazon Bedrockでキーワードを抽出し、旅行プランが生成されます。一連の処理は、S3バケットへの音声ファイルのアップロードをトリガーとしてLambdaで行われます。

▼概要図

▼使用技術の説明

Amazon S3
Amazon Web Servicesが提供するオブジェクトストレージサービスです。
耐久性と可用性が高く、大量のデータを低コストで保存できます。データは複数のリージョンにレプリケートされ、セキュリティ設定やアクセス制御が可能です。
S3バケット
S3でデータを整理・管理するための基本単位です。
各バケットは一意の名前を持ちます。バケット内でオブジェクトを保存し、アクセス制御を設定できるため、効率的なデータ管理が可能です。
AWS Lambda
コードをサーバーレスで実行するコンピューティングサービスです。
イベント駆動型の処理に最適であり、リクエストに基づき自動でスケーリングし、サーバーのプロビジョニングや管理が不要です。実行に対する課金により、コスト効率も追求できます。
Amazon Transcribe
自動音声認識（ASR）を利用して音声データをテキストに変換するサービスです。
リアルタイムおよびバッチ処理が可能で、多言語対応のため、会議録音の文字起こしや映像音声の字幕化に有用です。
Amazon Bedrock
機械学習モデルの迅速なデプロイと運用を支援するサービスです。
事前にトレーニングされたAIモデルを使用して、AIアプリケーションを効率的に構築できます。これにより、開発者は基盤構築よりもアプリケーションの改善に集中できます。

作業の流れ

作業の流れとしては以下となります。

S3バケットの作成
文字起こし処理を作成
旅行プラン作成処理を作成
実装結果の確認

作業準備

以下を準備してください

AWSアカウント
音声ファイル

注意点

イベント駆動型アーキテクチャを使用する場合、以下に注意すること
S3バケットへのファイルのアップロードをトリガーとしてファイルを加工する処理を実行した際に加工されたファイルを同じS3バケットに格納してしまうと無限ループが発生してしまいます。
これを 「再帰呼び出し」 といい高額請求が発生する可能性があります。

各作業の説明

1. S3バケットを作成

1-1. インプット用バケット(音声アップロード先)を作成

バケット名はvol3-transcribe-inputとします
上記以外はデフォルトとします
[バケットを作成]をクリックします

1-2.アウトプット用バケット(文字起こしファイルの格納先)を作成

バケット名はvol3-transcribe-outputとします
上記以外はデフォルトとします
[バケットを作成]をクリックします

1-3.アウトプット用バケット(htmlアップロード先)を作成

バケット名はvol3-html-outputとします
上記以外はデフォルトとします
[バケットを作成]をクリックします

2.文字起こし処理を作成

2-1.Lambda関数を新規作成

関数名はvol3-transcribe-lambdaとします
ランタイムはPython 3.13とします
上記以外はデフォルトとします
[関数の作成]をクリックします

2-2.Lambda関数にポリシーを追加

作成したLambda関数の[設定]を開きます
[アクセス権限]を開きます
ロール名のリンクを開きます
[許可を追加]から[ポリシーをアタッチ]を選択します
その他の許可ポリシーで以下2つを選択します
・AmazonS3FullAccess
・AmazonTranscribeFullAccess
[許可を追加]をクリックします

2-3.Lambda関数にトリガーを設定

作成したLambda関数の[設定]を開きます
[トリガー]を開きます
[トリガーを追加]をクリックします
ソースはS3を選択します
インプット用バケット(音声アップロード先)を記載します
再起呼び出しの確認にチェックを付けます
[追加]をクリックします

2-4.Lambda関数にPythonコードを記述

Lambda関数の[コード]タブを開きます
「lambda_function.py」ファイルに以下のコードを記載します

import boto3
import json
import datetime
import urllib.parse

def lambda_handler(event, context):
    # 本番用
    # テスト時はコメントアウトする
    # バケット名
    bucket = event['Records'][0]['s3']['bucket']['name']
    # ファイル名
    key = urllib.parse.unquote_plus(event['Records'][0]['s3']['object']['key'], encoding='utf-8')

    # テスト用
    # 本番時はコメントアウト、ファイル名を直接記載する
    # バケット名
    # bucket = "インプット用バケット名(音声アップロード先)"
    # ファイル名
    # key = "音声ファイル名"
    
    # transcribeクライアント生成
    transcribe = boto3.client("transcribe")
    
    jobName =  datetime.datetime.now().strftime("%Y%m%d%H%M%S") + "_Transcription"

    # 文字起こしファイルの格納先
    # ファイル名を直接記載する
    outputbucketname = "アウトプット用バケット(文字起こしファイルの格納先)"
    
    # 文字起こしジョブ開始
    try:
        transcribe.start_transcription_job(
            TranscriptionJobName= jobName,
            LanguageCode='ja-JP',
            Media={
                'MediaFileUri': 'https://s3.ap-northeast-1.amazonaws.com/' + bucket + '/' + key
            },
            OutputBucketName= outputbucketname # Transcribeの結果を出力するS3バケットを指定
        )

    except Exception as e:
        print(e)
        print('Error getting object {} from bucket {}. Make sure they exist and your bucket is in the same region as this function.'.format(key, bucket))
        raise e

2-5.Lambda関数のタイムアウト設定を変更

作成したLambda関数の[設定]を開きます
[一般設定]を開きます
[編集]から基本設定画面を開きます
タイムアウトを任意の値(今回は1分0秒)に変更します
[保存]をクリックします

2-6.Lambda関数をデプロイしてテスト実行

テスト実行前に①~②を確認します
①アウトプット用バケット(文字起こしファイルの格納先)にファイルが無いこと

②インプット用バケット(音声アップロード先)に音声ファイルがあること

音声ファイルについて
テストする場合は事前にファイルを格納しておく必要があります。
音声アップロードがトリガーとなる為、テスト終了後は音声ファイルを削除してください。

[Deploy]ボタンを押してデプロイします
[Test]ボタンを押してテスト実行します

テストイベントがない場合
[Create new test event]からテスト用イベント処理を作成します。
TemplateはHello Worldで問題ありません。

2-7.S3バケットに.tempと.jsonが格納されていることを確認

アウトプット用バケット(文字起こしファイルの格納先)を開きます
Transcribeによってアップロードされた下記のファイルが確認できればOKです
・tempファイル
・jsonファイル

3.旅行プラン作成処理を作成

3-1.Lambda関数を新規作成

関数名はvol3-bedrock-lambdaとします
ランタイムはPython 3.13とします
上記以外はデフォルトとします

3-2.Lambda関数にポリシーを追加

作成したLambda関数の[設定]を開きます
[アクセス権限]を開きます
ロール名のリンクを開きます
[許可を追加]から[ポリシーをアタッチ]を選択します
その他の許可ポリシーで以下2つを選択します
・AmazonS3FullAccess
・AmazonBedrockFullAccess
[許可を追加]をクリックします

3-3.Lambda関数にトリガーを設定

作成したLambda関数の[設定]を開きます
[トリガー]を開きます
[トリガーを追加]をクリックします
ソースはS3を選択します
アウトプット用バケット(文字起こしファイルの格納先)を記載します
サフィックスは[.json]を記載します
再起呼び出しの確認にチェックを付けます
[追加]をクリックします

3-4.Lambda関数にPythonコードを記述

Lambda関数の[コード]タブを開きます
「lambda_function.py」に以下のコードを記載します

import boto3
import json
import urllib.parse
from datetime import datetime
import tempfile

s3_resource = boto3.resource('s3')

def lambda_handler(event, context):
    # 本番用
    # テスト時はコメントアウトする
    # バケット名
    bucket = event['Records'][0]['s3']['bucket']['name']
    # ファイル名
    key = urllib.parse.unquote_plus(event['Records'][0]['s3']['object']['key'], encoding='utf-8')

    # テスト用
    # 本番時はコメントアウト、ファイル名を直接記載する
    # バケット名
    #bucket = "アウトプット用バケット(文字起こしファイルの格納先)"
    # ファイル名
    #key = "jsonファイル名"

    # s3からjsonファイルを取得
    json_data = s3_resource.Bucket(bucket).Object(key)

    # 取得したファイルを辞書型に変換(文字コード：UTF-8)
    data = json.loads(json_data.get()['Body'].read().decode('utf-8'))

    # 文字起こししたデータからキーワードを抽出するプロンプトの作成
    transcript = data['results']['transcripts'][0]['transcript']
    keyword_prompt = create_keyword(transcript)

    # Bedrockに投げ込んでキーワード作成
    keyword = send_to_bedrock(keyword_prompt)

    # キーワードから旅行先を考えるプロンプトの作成
    tripplan_prompt = create_tripplan(keyword)
    
    # 旅行先作成プロンプトをBedrockに投げ込む
    bedrock_response = send_to_bedrock(tripplan_prompt)

    # S3クライアントを作成
    s3_client = boto3.client('s3')

    # バケット名と保存するファイル名を指定
    bucket_name = "vol3-html-output"
    file_name = "test-scribe_vol3_ideason.html"
    Key = 'transcribe/test-scribe_vol3_ideason.html'
    
    # S3にファイルをアップロード
    with tempfile.NamedTemporaryFile() as tf:
        with open(tf.name, 'w+') as f:
            f.write(bedrock_response)
        s3_client.upload_fileobj(tf, bucket_name, Key, ExtraArgs={"ContentType": "text/html"})
    
    return {
        'statusCode': 200
    }

def create_keyword(transcript):
    # プロンプト作成ロジック
    keyword = (
        "以下は旅行したい場所について話した内容です。"
        + transcript +
        "内容を踏まえて旅行プランを考えるために必要なキーワードを最大3つまで抽出してください。"
        "キーワードはカンマ区切りにしてください。"
    )
    return keyword

def create_tripplan(transcript):
    # プロンプト作成ロジック
    prompt = (
        "以下は旅行したい場所についてのキーワードです。"
        + transcript +
        "この情報を基に関東でおすすめの旅行先を3つ教えてください。"
        "また、日程は最大3日間のスケジュールを時間ごとに出してください。"
        "宿泊先とおすすめのランチ情報もURL付きで載せてください。"
        "キーワードはタイトルの下に表示してください。"
        "返答内容を、文字コードUTF-8形式のhtmlタグのみを返してください。"
    )
    return prompt

# 作成したプロンプトをBedrockに投げ込む
def send_to_bedrock(prompt):

    bedrock_runtime = boto3.client('bedrock-runtime')

    # パラメータの設定
    max_tokens = 5000
    system_prompt = "必ず日本語で答えてください"
    user_message = {"role": "user", "content": prompt}
    messages = [user_message]
    model_id = 'anthropic.claude-3-5-sonnet-20240620-v1:0'
    
    body = json.dumps({
        "anthropic_version": "bedrock-2023-05-31",
        "max_tokens": max_tokens,
        "system": system_prompt,
        "messages": messages,
        "temperature": 0.7,
        "top_p": 0.9,
    })
    
    response = bedrock_runtime.invoke_model(
        modelId=model_id,
        contentType='application/json',
        accept='application/json',
        body=body
    )

    response_body = json.loads(response.get('body').read())
    return response_body['content'][0]['text']

3-5.Lambda関数のタイムアウト設定を変更

作成したLambda関数の[設定]を開きます
[一般設定]を開きます
[編集]から基本設定編集画面を開きます
タイムアウトを任意の値(今回は1分0秒)に変更します
[保存]をクリックします

3-6.Lambda関数をデプロイしてテスト実行

テスト実行前に以下を確認します
・アウトプット用バケット(htmlアップロード先)にファイルが無いこと
[Deploy]ボタンを押してデプロイします
[Test]ボタンを押してテスト実行します