Amazon Transcribeに再入門する（2022年12月版）

アドベントカレンダー2022

#機械学習

#Amazon Transcribe

nokomoro3

2022.12.11

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

はじめに

こんちには。

データアナリティクス事業本部機械学習チームの中村です。

本記事は「クラスメソッド機械学習チームアドベントカレンダー 2022」の11日目です。

AWSのフルマネージドな機械学習サービスは、PersonalizeやForecastなどに触れる機会が多かったですが、アドベントカレンダーをきっかけに、その他のAWSのフルマネージドな機械学習サービスに入門していこうと思います。

この流れを継続するかどうか予定は未定ですが、本記事ではAmazon Transcribeに入門していきます。

Amazon Transcribeとは

Amazon Transcribeは音声を入力し、書き起こし(Transcribe)するフルマネージドな機械学習サービスです。

Pricing(料金)

使用する前にPricingを確認していきましょう。（重要）

Pricingを確認することでおおよその機能概要も把握できます。

基本的には、１か月あたりの文字起こしされた音声データの秒数に基づいた従量課金となります（いくぶんか無料枠もあるようですが、期間と時間に制限がある）。少なくともTokyoリージョンでは、バッチとストリーミングに費用差はなく、1分あたり$0.024ドルからとなっています。 1時間の通話では、60 x $0.024 = $1.44という目安ですね。

上記が標準的な部分ですが、以下を追加することにより費用の追加が発生します。

PII Redaction機能
- 削除する必要がある社会保障番号や生年月日の情報などの個人を特定できる情報(PII)を除去する機能
- 標準のおよそ1/10が追加費用としてかかる様子
CLM
- カスタム言語モデルを構築し、これらのドメイン固有の用語を認識する機能
- 例えば、大学の講義動画を書き起こす場合に、科学用語を認識させる、などがユースケース
- 標準のおよそ1/4が追加費用としてかかる様子

また、これらに加えて特定の用途のTranscribeがあり、通話分析向けにTranscribe Call Analytics、医療向けにTranscribe Medicalがあり、それらは無印のTranscribeよりも価格が割高となっています。

言語毎のサポート状況

通常のTranscribeおよびCall Analyticsについては以下に対応表があります。

以降は、2022年12月10日現在の状況で記載します。

現在の対応状況を見てみましょう。「en-US」がフルスペックと考えればOKです。これと「ja-JP」の対応状況を比較してみます。

Language	Language Code	Data input	Transcribing numbers	Acronyms	Custom language models	Redaction	Call Analytics
English, US	en-US	batch, streaming	batch, streaming	batch, streaming	batch, streaming	batch, streaming	post-call, real-time
Japanese	ja-JP	batch, streaming	no	no	batch, streaming	no	post-call

まだ日本語対応していない部分がありますが、「Custom language models」はつい先日のアップデートで待望の日本語対応しています！今後他の部分の機能も対応が拡がることを期待しています。

また、Transcribe Medicalについて上記の表にありませんが、以下に「Amazon Transcribe Medical is available in US English (en-US).」と書いてある通り、「en-US」のみの対応となっているようです。

マネコンのリアルタイム処理について

マネジメントコンソールから、マイク入力のリアルタイム書き起こしを試すことができます。

その下部にある設定を見ていきましょう。

Language settings

書き起こしする対象の言語を設定することができます。「Specific language」だと単一の言語を設定し、「Automatic language identification」は言語を自動で判別します。自動判別には、発話の最初の３秒間で特定することが可能なようです。

自動判別というものの、複数の指定された言語から自動判別するという形のようで、2言語を最低でも選択する必要があります。また１言語につき、１つのバリエーションしか指定できないことに注意が必要です（en-USとen-AUを指定することはできない）。

Audio settings

「Speaker partitioning」という機能が有効化でき、この機能は話者の判別(Speaker diarization)を行うことができます。最大10人のスピーカーの音声を分離することができるようです。

詳細は以下を参照ください。

Content removal settings

「Vocabulary filtering」は、指定した単語をフィルタリングする機能で、事前に「Vocabulary filter」を作成する必要があります。

「PII Identification & redaction」はPricingのところで述べた通り、削除する必要がある社会保障番号や生年月日の情報などの個人を特定できる情報(PII)を同定・除去する機能です。有効にすると以下のように様々な、Financial情報やPersonal情報をチェックしてそれぞれ有効化することができます。

本機能は追加の料金が発生します。

Customizations

「Custom vocabulary」はユースケースに特化した単語やフレーズの認識精度を向上させるための機能で、「Custom vocabulary」を事前に作成する必要があります。

「Partial results stabilization」は最終結果のみを表示し書き起こし精度を優先するか、部分的な(Partial)結果を表示し高速にレスポンスさせるか、どちらを優先的にするかを調整します。これはリアルタイム処理特有の話です。実際に自動音声認識(ASR)システムは一般的に、現在の書き起こし結果を後段の音声入力の結果からフィードバックして修正するため、結果が後で変わる可能性があります。

その結果が代わる可能性があるものを途中結果のまま出力する場合は、High Levelを選択して高速に書き起こし、精度を重視したい場合はLow Levelを選択して書き起こします。

最後に「Custom language model」ですが、これもPricingのところで触れた通りでドメイン固有の用語を認識する機能を提供します。事前に「Custom language model」の作成が必要で、この機能は追加の料金が発生します。

実際に試してみた

Custom language modelについて

アップデートにより日本語対応していますので、こちらを有効化して試してみます。

作成のためには、テキストデータが必要で以下のいずれかを収集します。前者がより理想的なシナリオのようです。

実際に認識したいデータの正確なトランスクリプトデータ。「in-domain data」と呼ばれる。
トランスクリプトが無い場合は、関連する分野の文書を収集する。「domain-related data」と呼ばれる。

またデータは以下の２種類の使い方が可能です。

training data：最大2GBのテキストデータを使用した学習が可能
tuning data：最大200MBのテキストデータを使用してモデルをチューニング

両方使用することも可能ですが、tuning dataは、例えばtraining dataが十分に集められない（10,000語未満）などの場合に使用するようです。 training data、tuning dataの双方に同じテキストデータを含めないようにする必要があります。

詳細は以下を参照下さい。

テキストデータの準備

今回は、wiki40Bのtestセットを使います。以下のコードを参照に取得します。

https://note.com/npaka/n/n0a2d0a4b806e

import os
import tensorflow_datasets as tfds
ds_test = tfds.load('wiki40b/ja', split='test')

# データセットをテキスト形式で出力する関数
def create_txt(file_name, tf_data):
   start_paragraph = False

   # ファイルの書き込み
   with open(file_name, 'w') as f:
       for wiki in tf_data.as_numpy_iterator():
           for text in wiki['text'].decode().split('\n'):
               if start_paragraph:
                   text = text.replace('_NEWLINE_', '') # _NEWLINE_は削除
                   f.write(text + '\n')
                   start_paragraph = False
               if text == '_START_PARAGRAPH_': # _START_PARAGRAPH_のみ取得
                   start_paragraph = True

# データセットをテキスト形式で出力
create_txt('wiki_40b_test.txt', ds_test)

これにより得られる、wiki_40b_test.txt'というファイルを以下にアップロードしたとして以降を進めます。(100MBくらいのファイルとなります)

s3://nakamura-test-transcribe-2022-12-10/training_data/wiki_40b_test.txt

Custom language modelの作成

Train modelを押下して言語モデルを作成します。

設定は以下とします。

Model settings
- Name : nakamura-testmodel-2022-12-11
- Language : Japanese, JP (ja-JP)
- Base model : Wide band (マネコンからテストできるのは16kHzのWide bandのため)
Training data
- Resource URI : s3://nakamura-test-transcribe-2022-12-10/training_data/
Tuning data
- Optional なので今回は無しで作成

IAM roleは「Create an IAM role」を選択し、Permissions to accessは「Training and tuning S3 bucket」を選択します。 Role nameはnakamura-2022-12-10としておきました。

最後に「Train model」を押下します。以下の「In progress」が終わるまで待ちます。

今回は５時間程度学習にかかりました。

リアルタイム書き起こし(マネコン)

作成したCustom language modelを使って、マネジメントコンソールから書き起こしを動かしてみます。

今回は以下の設定でおこないました。

Language settings
- Language settings : Specific language
- Language : Japanese, JP (ja-JP)
Audio settings
- Speaker partitioning : オフ
Content removal settings
- Vocabulary filtering : オフ
- PII Identification & redaction : オフ
Customizations
- Custom vocabulary : オフ
- Partial results stabilization : オフ
- Custom language model : オン
- Custom model selection : nakamura-testmodel-2022-12-11

以下の「Start streaming」を押下すると、書き起こしが始まります。

停止は「Stop streaming」を押下すればOKです。以下のような結果となりました。

書き起こされた結果は、「Download full transcript」でJSONで取得できるようです。

途中結果なども含まれますので、"IsPartial": false,となっている部分を収集すれば、最終結果が得られそうです。

途中部分を省略すると以下のように結果が得られます。

[
	// ... 略 ...
	{
		"Transcript": {
			"Results": [
				{
					"Alternatives": [
						{
							"Items": [
								{
									"Confidence": 1,
									"Content": "アマゾン",
									"EndTime": 2.4575,
									"StartTime": 1.8175,
									"Type": "pronunciation",
									"VocabularyFilterMatch": false
								},
								{
									"Confidence": 0.9674,
									"Content": "と",
									"EndTime": 2.5475,
									"StartTime": 2.4575,
									"Type": "pronunciation",
									"VocabularyFilterMatch": false
								},
								// ... 略 ...
								{
									"Confidence": 0.6259,
									"Content": "流派",
									"EndTime": 3.7475,
									"StartTime": 3.3975,
									"Type": "pronunciation",
									"VocabularyFilterMatch": false
								}
							],
							"Transcript": "アマゾンとラインズくらい流派"
						}
					],
					"ChannelId": "ch_0",
					"EndTime": 4.005,
					"IsPartial": false,
					"ResultId": "8ca8b339-69d5-4ec5-9adb-577bbbb730cb",
					"StartTime": 1.49
				}
			]
		}
	},
	// ... 略 ...
	{
		"Transcript": {
			"Results": [
				{
					"Alternatives": [
						{
							"Items": [
								{
									"Confidence": 1,
									"Content": "音声",
									"EndTime": 5.4725,
									"StartTime": 4.8375,
									"Type": "pronunciation",
									"VocabularyFilterMatch": false
								},
								{
									"Confidence": 1,
									"Content": "から",
									"EndTime": 5.8475,
									"StartTime": 5.4725,
									"Type": "pronunciation",
									"VocabularyFilterMatch": false
								},
								// ... 略 ...
								{
									"Confidence": 0.9999,
									"Content": "です",
									"EndTime": 12.5075,
									"StartTime": 12.0075,
									"Type": "pronunciation",
									"VocabularyFilterMatch": false
								}
							],
							"Transcript": "音声からテキストへの書き起こしを行う車ねじとな機械学習サビスです"
						}
					],
					"ChannelId": "ch_0",
					"EndTime": 12.705,
					"IsPartial": false,
					"ResultId": "fc34b62d-46cd-4236-9c78-567ea7ce211d",
					"StartTime": 4.41
				}
			]
		}
	},
	// ... 略 ...
]

リアルタイム書き起こし(AWS SDKs)

システムへの組み込みを行う場合は、マネジメントコンソール以外の方法で書きおこしを実施する必要も出てきます。

その場合以下によると、AWS SDKs、HTTP/2、Websocketsによる方法があるようで、推奨されているのはAWS SDKsによる方法です。

今回は推奨通り、AWS SDKsを使った方法をPythonで試してみます。以下を参考にします。

標準のAWS SDK for Python (Boto3)はストリーミングに対応していないため、async Python SDK for Amazon Transcribeというものを代わりに使用する必要があります。

対応している音声のフォーマットは以下のようです。

FLAC
OPUSでエンコードされたデータが格納されるOggコンテナ
PCM（signed 16-bit little-endianのみ）　※要するにヘッダ無しの非圧縮rawデータで、一番一般的なもの

IAMユーザの準備

検証用にIAMユーザーを作成します。cm-nakamura-test-transcribeというユーザを作成し、ポリシーには以下を付与します。

arn:aws:iam::aws:policy/AmazonTranscribeFullAccess

作成後、./aws/configに以下を追記します。

[profile cm-nakamura-test-transcribe]
region = ap-northeast-1
output = json

また./aws/credentialsに認証情報を記載しておきます。

[cm-nakamura-test-transcribe]
aws_access_key_id = "アクセスキー ID"
aws_secret_access_key = "シークレットアクセスキー"

以下でアクティブにします。(PowerShellの場合)

$env:AWS_PROFILE = "cm-nakamura-test-transcribe"

実行

GitHubのサンプルコードを参考にします。

音声データをスマホなどの手元で録音し、./test.wavとして置いておきます。

フォーマットは、サンプリングレートを16000Hz, チャンネル数は1（モノラル）としておきます。

wavはそもそもストリーミングできないため、ストリーミングAPIももちろん対応していませんが、

サンプルコードではaiofileによりPCMフォーマットに変換して送信しているようです。

以下を実行すると、順次結果が画面出力されます。

Custom language modelは、language_model_nameの部分で指定します。

import asyncio

# This example uses aiofile for asynchronous file reads.
# It's not a dependency of the project but can be installed
# with `pip install aiofile`.
import aiofile

from amazon_transcribe.client import TranscribeStreamingClient
from amazon_transcribe.handlers import TranscriptResultStreamHandler
from amazon_transcribe.model import TranscriptEvent
from amazon_transcribe.utils import apply_realtime_delay

"""
Here's an example of a custom event handler you can extend to
process the returned transcription results as needed. This
handler will simply print the text out to your interpreter.
"""

SAMPLE_RATE = 16000
BYTES_PER_SAMPLE = 2
CHANNEL_NUMS = 1

# An example file can be found at tests/integration/assets/test.wav
AUDIO_PATH = "./test.wav"
CHUNK_SIZE = 1024 * 8
REGION = "ap-northeast-1"

class MyEventHandler(TranscriptResultStreamHandler):
    async def handle_transcript_event(self, transcript_event: TranscriptEvent):
        # This handler can be implemented to handle transcriptions as needed.
        # Here's an example to get started.
        results = transcript_event.transcript.results
        for result in results:
            for alt in result.alternatives:
                print(alt.transcript)


async def basic_transcribe():
    # Setup up our client with our chosen AWS region
    client = TranscribeStreamingClient(region=REGION)

    # Start transcription to generate our async stream
    stream = await client.start_stream_transcription(
        language_code="ja-JP",
        media_sample_rate_hz=SAMPLE_RATE,
        media_encoding="pcm",
        language_model_name="nakamura-testmodel-2022-12-11"
    )

    async def write_chunks():
        # NOTE: For pre-recorded files longer than 5 minutes, the sent audio
        # chunks should be rate limited to match the realtime bitrate of the
        # audio stream to avoid signing issues.
        async with aiofile.AIOFile(AUDIO_PATH, "rb") as afp:
            reader = aiofile.Reader(afp, chunk_size=CHUNK_SIZE)
            await apply_realtime_delay(
                stream, reader, BYTES_PER_SAMPLE, SAMPLE_RATE, CHANNEL_NUMS
            )
        await stream.input_stream.end_stream()

    # Instantiate our handler and start processing events
    handler = MyEventHandler(stream.output_stream)
    await asyncio.gather(write_chunks(), handler.handle_events())

loop = asyncio.get_event_loop()
loop.run_until_complete(basic_transcribe())
loop.close()

jupyterなどから実行すると以下のエラーが出ます。

RuntimeError: This event loop is already running

そのため、実行の際はmain.pyなどを準備して直接pythonスクリプトを実行してください。

その他

今回は実施しませんでしたが、その他にもTranscription Jobを作成することで、S3を入出力にしたバッチ処理が可能となっています。

まとめ

いかがでしたでしょうか。Amazon Transcribeについての一通りの使用方法について触れていきました。

本記事が、今後Amazon Transcribeを活用されようとする方の一助となれば幸いです。