【AWS AI/MLを触る】Amazon Transcribeを使った音声文字起こしをやってみる

2023年10月4日 17:35

はじめに

先日、Transcribeを使った音声テキスト変換の勉強会を執り行う機会がありまして、その中で作ったシステムを備忘として残したいと思います。
以下の構成をCDKで作りました。

システム動作

音声データ(mp3)をテキスト(日本語)に変換させるシステムになります。
簡単ではありますが、動作は以下のような流れになります。
① 音声(mp3)をS3にアップロードする
② S3にアップロードされたことをトリガーにLambdaが起動
③ Transcribeにて音声をテキスト変換実施
④ Transcribeが変換結果を返す
⑤ 変換結果(json)をS3に保管

CDK準備

LambdaとS3バケットをCDKで作成します。
CDKのバージョンは2です。cdk bootstrap済みを想定しています。

＄ mkdir transcribe && cd transcribe

$ cdk init --language typescript

AWS CDK (Typescript)

/lib/transcribe-stack.ts

import * as cdk from 'aws-cdk-lib';
import { Construct } from 'constructs';
import * as iam from 'aws-cdk-lib/aws-iam';
import * as s3 from 'aws-cdk-lib/aws-s3';
import * as lambda from 'aws-cdk-lib/aws-lambda';
import * as destinations from 'aws-cdk-lib/aws-s3-notifications';

export class TranscribeStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

     // Lambda用ロール作成
     const sampleRole = new iam.Role(this, 'SampleRole', {
      roleName: 'sample-role',
      assumedBy: new iam.ServicePrincipal('lambda.amazonaws.com'),
      managedPolicies: [
        iam.ManagedPolicy.fromAwsManagedPolicyName(
          'service-role/AWSLambdaBasicExecutionRole'
        ),
        iam.ManagedPolicy.fromAwsManagedPolicyName(
          'AmazonTranscribeFullAccess'
        ),
        iam.ManagedPolicy.fromAwsManagedPolicyName(
          'AmazonS3FullAccess'
        )
      ],
    });
  
  // Lambda関数の作成
  const sampleLambdaFunction = new lambda.Function(this, 'SampleFunction', {
    functionName: 'sample-function',
    runtime: lambda.Runtime.PYTHON_3_9,
    code: lambda.Code.fromAsset('lambda'),
    handler: 'sample.handler',
    role: sampleRole,
    timeout: cdk.Duration.seconds(30),
    tracing: lambda.Tracing.ACTIVE,      
  });

  // インプット用S3バケット作成
  const bucket = new s3.Bucket(this, 'InputBucket', {
    bucketName: 'input-bucket-1234567890',
    removalPolicy: cdk.RemovalPolicy.DESTROY,
    autoDeleteObjects: true,
  });
  
  // S3用のLambdaイベントソースの作成
  bucket.addEventNotification(s3.EventType.OBJECT_CREATED_PUT, 
    new destinations.LambdaDestination(sampleLambdaFunction)
  );

  // アウトプット用S3バケット作成
  const output_bucket = new s3.Bucket(this, 'OutputBucket', {
    bucketName: 'output-bucket-1234567890',
    removalPolicy: cdk.RemovalPolicy.DESTROY,
    autoDeleteObjects: true,
  });    
  }
}

CDKで作る主なリソースは、S3バケットとLambda関数です。
(厳密にはLamnda用のiamロールも作りますが。。)
Lambda関数はタイムアウト30秒としていますが、変換させる音声が長文で30秒を超える可能性がある場合は、適宜調整してください。
S3バケット名も適宜変更してください。

/bin/transcribe.ts

#!/usr/bin/env node
import 'source-map-support/register';
import * as cdk from 'aws-cdk-lib';
import { TranscribeStack } from '../lib/transcribe-stack';

const app = new cdk.App();
new TranscribeStack(app, 'TranscribeStack', {});

Lambda関数用コード(Python)

「lambda」フォルダを作成し、sample.pyというファイル名にしてます。

/lambda/sample.py

import json
import boto3
from botocore.exceptions import ClientError
from datetime import datetime

def handler(event, context):
    try:
        # S3イベントからバケット名とオブジェクトキーを取得
        bucket = event['Records'][0]['s3']['bucket']['name']
        key = event['Records'][0]['s3']['object']['key']

        # AWS Transcribeクライアントの初期化
        transcribe = boto3.client('transcribe')

        # トランスクリプトジョブ名にオブジェクトキー+時刻となるよう指定
        job_name = key.split('.')[0]+ "_" + datetime.now().strftime('%Y-%m-%d-%H-%M-%S') 

        # トランスクリプトジョブの作成
        response = transcribe.start_transcription_job(
            TranscriptionJobName=job_name,
            LanguageCode='ja-JP',
            MediaFormat='mp3'or 'mp4'or 'wav' or 'flac' or 'ogg' or 'amr' or 'webm',
            Media={
                'MediaFileUri': f's3://{bucket}/{key}'
            },
            OutputBucketName='output-bucket-1234567890'  # Transcribeの結果を出力するS3バケットを指定
        )

        return {
            'statusCode': 200,
            'body': json.dumps('Transcription job started.')
        }

    except ClientError as e:
        # エラーログを出力
        print(f"Error: {e}")
        # エラーレスポンスを返す
        return {
            'statusCode': 500,
            'body': json.dumps('Error occurred while starting the transcription job.')
        }

    except Exception as e:
        # 予期しないエラーの場合の処理
        print(f"Unexpected Error: {e}")
        # エラーレスポンスを返す
        return {
            'statusCode': 500,
            'body': json.dumps('An unexpected error occurred.')
        }

アウトプット用S3バケット名(コード内の"OutputBucketName")をCDKコード内のアウトプットバケット名と合わせるよう適宜修正します。
動作としては、インプット用S3バケットに音声(mp3)が置かれたことをトリガーにしてLambdaが発火、Transcribeで音声をテキスト変換し、変換後、結果をアウトプット用S3バケットに出力(json)させます。

CDKデプロイ

$ cdk synth
$ cdk deploy

動作確認

音声ファイルを適宜用意します。(mp3など)
実際に音声(mp3)をインプット用S3バケットにアップロードし、しばらく待つとアウトプット用S3バケットにJSONが自動出力されます。
音声の長さによって変換時間は変わります。
この仕組みをさらに発展させて、会議の議事録作成など様々な場面で応用が利きそうですね。

まとめ

AWS AI/MLの勉強会でTranscribeを使った音声変換システムの講義をする機会がありました。講義の中で実際に構築したシステムを簡単ではありますが解説しました。個人の備忘として記録しておこうと思います。