初学者向け「Amazon Transcribe」（AI文字起こしサービス）をPythonで利用するには：AWSチートシート

AWS活用における便利な小技を簡潔に紹介する連載「AWSチートシート」。今回は、AWSのAI文字起こしサービス「Amazon Transcribe」をPythonで利用する方法を紹介する。

[金晟基，東京ITスクール]

　「Amazon Web Services」（AWS）活用における便利な小技を簡潔に紹介する連載「AWSチートシート」。今回は、AWSのAI文字起こしサービス「Amazon Transcribe」をPythonで利用します。以下、Transcribeに用意されているメソッドを概観し、幾つかの使い方を紹介します。

「Amazon Transcribe」とは

　AWSには、事前トレーニング済みのAI（人工知能）を手軽に利用できる「AIサービス」が多数用意されており、その内容はコンピュータビジョンから言語、レコメンデーション、予測と多岐にわたります。

　今回紹介するTranscribeは、オーディオファイルまたは動画ファイルから、音声を文字起こしするAIクラウドサービスです。サポート言語はバッチ処理とリアルタイム処理で異なり、2021年7月の本稿執筆時点で、それぞれ31言語と12言語が提供されています。この中には、アメリカ英語やイギリス英語など、細分化されている言語もあります。なお、日本語はどちらの処理にも対応しています。

　利用可能な入力ファイル形式には、バッチ処理ではFLAC、MP3、MP4、Ogg、WebM、AMR、WAVがあります（音声データは長さが4時間未満、または容量が2GB未満である必要があります）。一方、リアルタイム処理では、ストリームはPCM16ビット符号付きリトルエンディアン、Oggコンテナに格納されたOPUS、FLACのいずれかでエンコードされている必要があります。

Amazon Transcribe

　ちなみに、Transcribeには姉妹サービスとして、医療分野での文字起こしに特化した「Amazon Transcribe Medical」が別途提供されています。ただし、現在こちらはアメリカ英語のみ対応です。

AWSの「AIサービス」はコンソール画面から利用できますが、開発を念頭に置かなくても、慣れてくれば今回のようにAPIを利用する方がより便利で効率的に感じてくるでしょう。本稿がそのように利用するきっかけになれば幸いです。

利用料金

　Transcribeは従量課金制で、文字起こしを行った秒数に応じて毎月課金されます。その際の単価は、累積利用時間で区分されるTierごとに、段階的な割引が適用されるスタイルになっています。東京リージョンでの料金を分単価（秒単価×60）で見てみると、最初の25万分（T1）は0.0240ドル、続く75万分（T2）は0.0150ドル、その後（T3）は0.0108ドルとなっています。なお、1リクエスト当たりの最小料金は15秒分に設定されています（他にもTranscribe Medical、「自動コンテンツリダクション」「カスタム言語モデル」に関する料金が別途設定されています。詳細は公式サイトをご確認ください）。

　ちなみに、Transcribeは無料利用枠の対象になっており、最初のリクエストから12カ月間は、1カ月当たり60分まで無料で利用できます（ただし、無料利用枠を超えた場合には従量課金が適用されます）。

必要条件

　本稿では、読者の環境で下記要件が満たされていることを仮定しています。

1. AWSアカウントを有しており、「AWS Identity and Access Management」（IAM）ユーザーに必要な権限（今回ならTranscribeと「Amazon S3」関連）が付与されていること。ホームディレクトリにAPIを利用するための認証情報が保存されていること
2. AWSが提供するPython用のSDK「Boto3」がインストールされていること

　なおこれは必須ではありませんが、以下のサンプルコードは「Jupyter Notebook」での実行を想定しています。

メソッド一覧

　Transcribeには以下のメソッドが用意されています（以下ではリアルタイム処理、ならびにTranscribe Medicalに関するメソッドは割愛しています）。

メソッド名	機能	引数	戻り値
create_language_model	新しいカスタム言語モデルを作成する	言語コード、ベースモデル名、言語モデル名、入力データの構成情報	辞書
list_language_models	作成済みカスタム言語モデルの情報を返す	ステータス値、言語モデル名の部分文字列、NextToken、レスポンスの最大数	辞書
describe_language_model	特定のカスタム言語モデルの情報を返す	言語モデル名	辞書
delete_language_model	特定のカスタム言語モデルを削除する	言語モデル名	-
create_vocabulary	新しいカスタム語彙（ごい）を作成する	語彙名、言語コード、フレーズ、語彙ファイルURI	辞書
update_vocabulary	既存のカスタム語彙を更新する	語彙名、言語コード、フレーズ、語彙ファイルURI	辞書
list_vocabularies	作成済みカスタム語彙の情報を返す	ステータス値、語彙名の部分文字列、NextToken、レスポンスの最大数	辞書
get_vocabulary	特定のカスタム語彙の情報を返す	語彙名	辞書
delete_vocabulary	特定のカスタム語彙を削除する	語彙名	-
create_vocabulary_filter	新しい語彙フィルターを作成する	語彙フィルター名、言語コード、語彙または語彙フィルターファイルURI	辞書
update_vocabulary_filter	既存の語彙フィルターを更新する	語彙フィルター名、語彙または語彙フィルターファイルURI	辞書
list_vocabulary_filters	作成済み語彙フィルターの情報を返す	語彙フィルター名の部分文字列、NextToken、レスポンスの最大数	辞書
get_vocabulary_filter	特定の語彙フィルターの情報を返す	語彙フィルター名	辞書
delete_vocabulary_filter	特定の語彙フィルターを削除する	語彙フィルター名	-
start_transcription_job	非同期文字起こしジョブを開始する	文字起こしジョブ名、ファイルURI、書き出し用バケット名、その他各種設定値	辞書
list_transcription_jobs	開始済み文字起こしジョブの情報を返す	ステータス値、文字起こしジョブ名の部分文字列、NextToken、レスポンスの最大数	辞書
get_transcription_job	特定の文字起こしジョブの情報を返す	文字起こしジョブ名	辞書
delete_transcription_job	特定の文字起こしジョブおよび関連情報を削除する文字起こし	ジョブ名	-
can_paginate	各メソッドのページネーション有無を調べる	メソッド名	真偽値
get_paginator	メソッドに関するページネータを生成する	メソッド名	ページネータオブジェクト
generate_presigned_url	署名済みURLを返す	メソッドとその引数	署名済みURL

　Transcribeでは、「カスタム言語モデル」の作成や、「カスタム語彙」としてドメイン固有の単語や語句を登録することで、文字起こしの精度を改善することができます。また、「語彙フィルター」を作成して、不適切な単語を文字起こし結果からマスクまたは削除することができます。

　他にも、「自動コンテンツリダクション」機能を用いて、クレジットカード番号など、機密性の高い個人識別情報（PII）を文字起こしの結果から自動的に削除することもできます（現在は非同期処理のアメリカ英語のみ対応）。

　なお、Transcribeには音声に含まれる複数の話者の識別、または、オーディオチャンネルによる識別を行う機能があります。これによって、例えば、カスタマーサポートの通話記録における顧客とオペレーターの識別や、動画における話者別の字幕作成が可能になります。

　ここからはTranscribeのメソッドを用いて音声ファイルの文字起こしを行ってみましょう。

start_transcription_jobメソッド

　「start_transcription_job」メソッドは、Transcribeの最も基本的なメソッドです。これはオーディオファイルまたは動画ファイルに対して、音声の非同期文字起こしジョブを開始するメソッドです。今回はサンプルとして、夢野久作の『懐中時計』を、音源は『青空朗読』の朗読ファイルを使用しました。ターゲットとなるテキストを以下に掲載します。

夢野久作 作
懐中時計
　懐中時計が箪笥の向う側へ落ちて一人でチクタクと動いておりました。
　鼠が見つけて笑いました。
「馬鹿だなあ。誰も見る者はないのに、何だって動いているんだえ」
「人の見ない時でも動いているから、いつ見られても役に立つのさ」
　と懐中時計は答えました。
「人の見ない時だけか、又は人が見ている時だけに働いているものはどちらも泥棒だよ」
　鼠は恥かしくなってコソコソと逃げて行きました。

　文字起こしの手順は次の通りです。まず、S3にバケットを作成して、その中に音源ファイルをアップロードします（S3バケットはTranscribeを呼び出す際のAPIエンドポイントと同じリージョンに作成する必要があります）。次に、start_transcription_jobメソッドを利用して、文字起こしのジョブを開始します。その後は、get_transcription_jobメソッドを利用して、ジョブのステータスを定期的にチェックし、完了したらファイルをローカルにダウンロードします。

import boto3
import uuid
import time
import json
 
# 一意識別子を利用してバケット名を作成
bucket = 'transcribe-sample-' + str(uuid.uuid1())
 
# S3クライアントを作成
s3 = boto3.client('s3')
 
# S3にバケットを作成
s3.create_bucket(
    # バケット名の指定
    Bucket=bucket,
    # リージョンの指定
    CreateBucketConfiguration={'LocationConstraint': 'ap-northeast-1'})
 
# S3にアップロードする音声ファイルのパス
audio_file = 'kaichuudokei.mp3'
 
# オブジェクトキー
input_key = 'kaichuudokei.mp3'
 
# S3に音声ファイルをアップロード
s3.upload_file(audio_file, bucket, input_key)

# Transcribeクライアントを作成
transcribe = boto3.client('transcribe')
 
# (1)ジョブ名
job_name = 'sample_job'
# job_name = 'sample_job_with_vocab'
 
# ジョブで使用する音源ファイルのURI
job_uri = f's3://{bucket}/{input_key}'
 
# (2)結果を書き込むオブジェクトのキー
output_key = 'sample_output.json'
# output_key = 'sample_output_with_vocab.json'
 
# 非同期文字起こしジョブを開始
transcribe.start_transcription_job(
            # ジョブ名を指定
            TranscriptionJobName=job_name,
            # メディアファイルを指定
            Media={'MediaFileUri': job_uri},
            # メディアフォーマットを指定
            MediaFormat='mp3',
            # 言語コードを指定
            LanguageCode='ja-JP',
            # 結果を書き出すS3バケットを指定
            OutputBucketName=bucket,
            # 結果を書き込むオブジェクトキーを指定
            OutputKey=output_key,
            # (3)カスタム語彙名を指定
            # Settings={'VocabularyName': vocab_name}
)
 
# ステータスを5秒ごとにチェック（筆者実行時には完了まで30秒程度かかりました）
while True:
    response = transcribe.get_transcription_job(TranscriptionJobName=job_name)
    status = response['TranscriptionJob']['TranscriptionJobStatus']
    if status in ['COMPLETED', 'FAILED']:
        print(' Status:', status)
        break
    print('>', end='')
    time.sleep(5)
 
# S3から結果ファイルをダウンロード
s3.download_file(bucket, output_key, output_key)
 
# 結果を表示
with open(output_key) as json_file:
    json_object = json.load(json_file)
    print(json_object['results']['transcripts'][0]['transcript'])

　文字起こしの結果は以下のようになりました（ここでは読みやすさのために改行を入れています）。

夢の久作作
会長門外
海中だけがダンスの向こう側へ落ちてひとりでチクタクと動いておりました
ネズミが見つけて笑いました
バカだな誰も見るものはないのに何？ だって動いているだ
人の見ない時でも動いているからいつ見られても役に立つのさと会長時計は答えました
人の未来時だけかまたは人が見ている時だけに働いているものはどちらも泥棒だよ
ネズミは恥ずかしくなってこそこそと逃げていきました

　いかがでしょうか。全体的によく文字起こしされていると思います。ただ今回の結果を見ると、作者名の他に、タイトルでもある重要単語の「懐中時計」が、正しく文字起こしされませんでした。そこで、以下ではこれらを「カスタム語彙」として登録して、再度、文字起こしを行ってみます。

create_vocabularyメソッド

　手順は次の通りです。まず、「カスタム語彙」として登録する内容を下記の要領で列挙して、テキストファイルに保存（ここではファイル名をvocabulary.txt と）します。なお、単語と単語はTABで区切る必要があります。

Phrase SoundsLike DisplayAs IPA
夢野久作 yumenokyuusaku 夢野久作
懐中時計 kaichuudokei 懐中時計

　次に、語彙ファイルを先ほど作成したS3バケットにアップロードします。続いて、create_vocabularyメソッドを利用して「カスタム語彙」の作成を開始します。その後は、get_vocabularyメソッドを利用してステータスを定期的にチェックして、完了を待ちます。

# S3にアップロードする語彙ファイルのパス
vocab_file = 'vocabulary.txt'
 
# オブジェクトキー
vocab_key = 'vocabulary.txt'
 
# S3に語彙ファイルをアップロード
s3.upload_file(vocab_file, bucket, vocab_key)
 
# ジョブで使用する語彙ファイルのURI
vocab_uri = f's3://{bucket}/{vocab_key}'
 
# カスタム語彙名
vocab_name = 'sample_vocab'
 
# カスタム語彙を作成
transcribe.create_vocabulary(
            VocabularyName=vocab_name,
            LanguageCode='ja-JP',
            VocabularyFileUri=vocab_uri)
 
# ステータスを5秒ごとにチェック（筆者実行時には完了まで10秒程度かかりました。）
while True:
    response = transcribe.get_vocabulary(VocabularyName=vocab_name)
    status = response['VocabularyState']
    if status in ['READY', 'FAILED']:
        print(' Status:', status)
        break
    print('>', end='')
    time.sleep(5)

　「カスタム語彙」の作成が完了したら、start_transcription_jobメソッドが含まれた1つ前のコードセルを再度実行します。ただし、今回は「(1)ジョブ名」と「(2)結果を書き込むオブジェクトのキー名」の箇所で、1行目をコメントアウトして2行目を有効にしてください。また、start_transcription_jobメソッドの最後のキーワード引数である「(3)Settings」を有効にしてください。「カスタム語彙」を利用した文字起こしの結果は以下のようになりました（ここでも読みやすさのために改行を入れています）。

夢野久作作
懐中時計
懐中時計がダンスの向こう側へ落ちてひとりでチクタクと動いておりました
ネズミが見つけて笑いました
バカだな誰も見るものはないのに何？ だって動いているだ
人の見ない時でも動いているからいつ見られても役に立つのさと懐中時計は答えました
人の未来時だけかまたは人が見ている時だけに働いているものはどちらも泥棒だよ
ネズミは恥ずかしくなってこそこそと逃げていきました

　いかがでしょうか。今回は作者名と「懐中時計」という語彙も、正しく文字に起こされました。

その他のメソッド

　最後に、Transcribe（とS3）のメソッドを幾つかまとめて使用して、後片付けをします。

　手順は次の通りです。まず、delete_vocabularyメソッドを利用して、「カスタム語彙」を削除します。次に、list_transcription_jobsメソッドを用いて文字起こしジョブの一覧を取得し、これに基づいてdelete_transcription_jobメソッドでジョブを削除します。最後に、S3 バケット内のオブジェクトを全て削除して、S3バケットも削除します。

# カスタム語彙を削除
transcribe.delete_vocabulary(VocabularyName=vocab_name)
 
# 文字起こしジョブを削除
response = transcribe.list_transcription_jobs()
for job in response['TranscriptionJobSummaries']:
    transcribe.delete_transcription_job(
                TranscriptionJobName=job['TranscriptionJobName'])
 
# S3バケット内のオブジェクトを削除
response = s3.list_objects_v2(Bucket=bucket)
for content in response['Contents']:
    s3.delete_object(Bucket=bucket, Key=content['Key'])
 
# S3バケットを削除
s3.delete_bucket(Bucket=bucket)

最後

　いかがだったでしょうか。文字起こしは、このような簡単なサンプルでも十分楽しめますが、より実用的なユースケースも数多く考えられます（公式サイトでは、自動字幕生成は元より、AWSの他のサービスと組み合わせることで、コールセンターにおける通話解析を行い、オペレーターの生産性や顧客体験を向上させるケースなどが紹介されています）。

　今回は必要最低限のメソッドのみ紹介しましたが、興味を持った方は公式ドキュメントを参考に、その他のメソッドも試してみてください。

筆者紹介

金 晟基（キム ソンギ）

株式会社システムシェアード

東京ITスクールでJava研修の講師、IT専門学校の教材、カリキュラム開発、一般社団法人とのプログラミング教育を通じた国際貢献事業などを担当。AWS認定資格は「機械学習 - 専門知識」など