Closed9
Amazon Transcribe Streaming SDKを試す
kun432たまたま見つけたのだけども。
今回はPythonを使用して実装していますが、Amazon Transcribe で提供されている SDK は AWS SDK for Python（Boto3）ではなく、Amazon Transcribe 用の非同期 Python SDKです。
なんと、そんなものがあるのかー。
公式ドキュメントにも書いてあった。
ストリーミングされたメディアファイルまたはライブメディアストリームを使用して、ストリーミング文字起こしを作成できます。
標準 AWS SDK for Python (Boto3) はストリーミングでは Amazon Transcribe サポートされていないことに注意してください。Python を使用してストリーミング文字起こしを開始するには、この非同期 Python SDK for Amazon Transcribeを使用します。
これちょっと試してみる。
kun432GitHubレポジトリ
Amazon Transcribe Streaming SDK
Amazon Transcribe Streaming SDK を使用すると、ユーザーはAmazon Transcribe Streaming サービスとPythonプログラムを直接インターフェイスさせることができます。 このプロジェクトの目標は、ユーザーがオーディオバイトのストリームと基本ハンドラー以上のものを使用することなく、Amazon Transcribe に直接統合できるようにすることです。
このプロジェクトはまだ初期アルファ版であるため、インターフェースはまだ変更の対象であり、急速に反復される可能性があります。ローカルテスト以外で使用する場合は、厳密な依存関係を固定することを強くお勧めします。awscrtはbotocore（AWS CLIおよびboto3のコアモジュール）と共有される依存関係であることにご注意ください。同じ環境にインストールする場合は、amazon-transcribeを最新バージョンに維持する必要がある場合があります。
2年ぐらい更新されていないライブラリっぽいけど・・・まあそんなに新しい機能もないか。
kun432作業ディレクトリ作成
mkdir amazon-transcribe-streaming-test && cd amazon-transcribe-streaming-test
uvでPython仮想環境作成
uv venv -p 3.12.8
パッケージインストール
uv pip install amazon-transcribe
出力
Installed 2 packages in 4ms
+ amazon-transcribe==0.6.2
+ awscrt==0.16.26
サンプルコードはファイルから文字起こし担っているため、aiofileも必要。
uv pip install aiofile
出力
Installed 2 packages in 3ms
+ aiofile==3.9.0
+ caio==0.9.21
サンプル音声として、自分が開催した勉強会のYouTube動画から冒頭5分程度の音声を抜き出したオーディオファイルを与えてみる。
なお、Trasncribeに入力可能なオーディオ形式は以下に記載がある。バッチとストリーミングで異なるみたい。
今回用意しているのは以下のmp3。ffprobeの結果。
出力
Input #0, mp3, from 'sample.mp3':
Metadata:
encoder : Lavf61.7.100
Duration: 00:04:28.80, start: 0.023021, bitrate: 128 kb/s
Stream #0:0: Audio: mp3 (mp3float), 48000 Hz, stereo, fltp, 128 kb/s
Metadata:
encoder : Lavc61.19
今回はPCM（リトルエンディアン 16-bit, 16kHz, モノラル）に変換する
ffmpeg -i sample.mp3 -f s16le -acodec pcm_s16le -ar 16000 -ac 1 sample.pcm
サンプルコード
transcribe.py
import asyncio
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
SAMPLE_RATE = 16000 # サンプリングレート（16kHz）
BYTES_PER_SAMPLE = 2 # 1サンプルあたりのバイト数
CHANNEL_NUMS = 1 # モノラル音声
AUDIO_PATH = "sample.pcm" # 音声ファイルのパス
CHUNK_SIZE = 1024 * 8 # 1回の読み取りで処理するバイト数
REGION = "ap-northeast-1" # AWS リージョン
# カスタムハンドラーの定義
class MyEventHandler(TranscriptResultStreamHandler):
async def handle_transcript_event(self, transcript_event: TranscriptEvent):
results = transcript_event.transcript.results
for result in results:
for alt in result.alternatives:
print(alt.transcript)
# 音声ストリームを送信して文字起こし
async def basic_transcribe():
# Transcribe クライアントの作成
client = TranscribeStreamingClient(region=REGION)
stream = await client.start_stream_transcription(
language_code="ja-JP",
media_sample_rate_hz=SAMPLE_RATE,
media_encoding="pcm",
)
async def write_chunks():
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()
# カスタムハンドラーの作成
handler = MyEventHandler(stream.output_stream)
# 非同期タスクの実行
await asyncio.gather(write_chunks(), handler.handle_events())
if __name__ == "__main__":
asyncio.run(basic_transcribe())
出力
はい、
はい、じゃあ
はい、じゃあ始めます
はい、じゃあ始めます。ちょっとまだ
はい、じゃあ始めます。ちょっとまだ来られ
はい、じゃあ始めます。ちょっとまだ来られてない
はい、じゃあ始めます。ちょっとまだ来られてない方も
はい、じゃあ始めます。ちょっとまだ来られてない方もいら
はい、じゃあ始めます。ちょっとまだ来られてない方もいらっしゃる
はい、じゃあ始めます。ちょっとまだ来られてない方もいらっしゃるんです
はい、じゃあ始めます。ちょっとまだ来られてない方もいらっしゃるんですけど
はい、じゃあ始めます。ちょっとまだ来られてない方もいらっしゃるんですけど、
ボイスラン
ボイスランチ
ボイスランチ、j
ボイスランチjp始め
ボイスランチjp始めます。
ボイスランチjp始めます。皆
ボイスランチjp始めます。皆さん、
ボイスランチjp始めます。皆さん、見て
ボイスランチjp始めます。皆さん、言って。
はい
はい
はい。
はい、
はい、日曜日に
はい、日曜日にお集
はい、日曜日にお集まりいただき
はい、日曜日にお集まりいただきましてありがとう
はい、日曜日にお集まりいただきましてありがとうござい
はい、日曜日にお集まりいただきましてありがとうございます。えっ
はい、日曜日にお集まりいただきましてありがとうございます。えっと、今日
はい、日曜日にお集まりいただきましてありがとうございます。えっと、今日久し
はい、日曜日にお集まりいただきましてありがとうございます。えっと、今日久しぶりにですね
はい、日曜日にお集まりいただきましてありがとうございます。えっと、今日久しぶりにですね、オフライ
はい、日曜日にお集まりいただきましてありがとうございます。えっと、今日久しぶりにですね、オフラインという
はい、日曜日にお集まりいただきましてありがとうございます。えっと、今日久しぶりにですね、オフラインということで
はい、日曜日にお集まりいただきましてありがとうございます。えっと、今日久しぶりにですね、オフラインということで、えっと
はい、日曜日にお集まりいただきましてありがとうございます。えっと、今日久しぶりにですね、オフラインということで、えっと今日はです
はい、日曜日にお集まりいただきましてありがとうございます。えっと、今日久しぶりにですね、オフラインということで、えっと今日はですね、スペ
はい、日曜日にお集まりいただきましてありがとうございます。えっと、今日久しぶりにですね、オフラインということで、えっと今日はですね、スペシャルなゲ
はい、日曜日にお集まりいただきましてありがとうございます。えっと、今日久しぶりにですね、オフラインということで、えっと今日はですね、スペシャルなゲストをお
はい、日曜日にお集まりいただきましてありがとうございます。えっと、今日久しぶりにですね、オフラインということで、えっと今日はですね、スペシャルなゲストをお二人
はい、日曜日にお集まりいただきましてありがとうございます。えっと、今日久しぶりにですね、オフラインということで、えっと今日はですね、スペシャルなゲストをお二人
はい、日曜日にお集まりいただきましてありがとうございます。えっと、今日久しぶりにですね、オフラインということで、えっと今日はですね、スペシャルなゲストをお二人来て
はい、日曜日にお集まりいただきましてありがとうございます。えっと、今日久しぶりにですね、オフラインということで、えっと今日はですね、スペシャルなゲストをお二人来ていただいて
はい、日曜日にお集まりいただきましてありがとうございます。えっと、今日久しぶりにですね、オフラインということで、えっと今日はですね、スペシャルなゲストをお二人来ていただいており
はい、日曜日にお集まりいただきましてありがとうございます。えっと、今日久しぶりにですね、オフラインということで、えっと今日はですね、スペシャルなゲストをお二人来ていただいております。
(snip)
こんな感じでストリーミングで出力される。
Amazon Transcribe のレスポンスには、is_partialという最終確定する前の一時的な結果を示すフラグが含まれているので、これを見てみる。
イベントハンドラを書き換え。
transcribe.py
(snip)
import json
class RawResponseHandler(TranscriptResultStreamHandler):
async def handle_transcript_event(self, transcript_event: TranscriptEvent):
results = transcript_event.transcript.results
for result in results:
transcript_text = result.alternatives[0].transcript # 最初の候補を取得
# JSONレスポンス全体を出力（デバッグ用）
print("**********")
print(json.dumps(transcript_event.transcript, default=lambda o: o.__dict__, indent=2, ensure_ascii=False))
print("**********")
print(f"TRANSCRIPT: {transcript_text}\n")
(snip)
async def basic_transcribe():
(snip)
#handler = MyEventHandler(stream.output_stream)
handler = RawResponseHandler(stream.output_stream)
(snip)
実行。
出力
**********
{
"results": [
{
"result_id": "88150653-7a00-4a87-87b5-1af72ef98c0a",
"start_time": 0.845,
"end_time": 1.855,
"is_partial": true,
"alternatives": [
{
"transcript": "はい、",
"items": [
{
"start_time": 0.855,
"end_time": 1.125,
"item_type": "pronunciation",
"content": "はい",
"vocabulary_filter_match": false,
"speaker": null,
"confidence": null,
"stable": null
},
{
"start_time": 1.125,
"end_time": 1.125,
"item_type": "punctuation",
"content": "、",
"vocabulary_filter_match": false,
"speaker": null,
"confidence": null,
"stable": null
}
]
}
],
"channel_id": "ch_0"
}
]
}
**********
TRANSCRIPT: はい、
**********
{
"results": [
{
"result_id": "88150653-7a00-4a87-87b5-1af72ef98c0a",
"start_time": 0.845,
"end_time": 2.335,
"is_partial": true,
"alternatives": [
{
"transcript": "はい、じゃあ",
"items": [
{
"start_time": 0.855,
"end_time": 1.125,
"item_type": "pronunciation",
"content": "はい",
"vocabulary_filter_match": false,
"speaker": null,
"confidence": null,
"stable": null
},
{
"start_time": 1.125,
"end_time": 1.125,
"item_type": "punctuation",
"content": "、",
"vocabulary_filter_match": false,
"speaker": null,
"confidence": null,
"stable": null
},
{
"start_time": 1.345,
"end_time": 1.805,
"item_type": "pronunciation",
"content": "じゃあ",
"vocabulary_filter_match": false,
"speaker": null,
"confidence": null,
"stable": null
}
]
}
],
"channel_id": "ch_0"
}
]
}
**********
TRANSCRIPT: はい、じゃあ
最終確定すると "is_partial": false で返されて、次のデータがまた "is_partial": trueで流れてくる。
出力
**********
{
"results": [
{
"result_id": "88150653-7a00-4a87-87b5-1af72ef98c0a",
"start_time": 0.845,
"end_time": 6.925,
"is_partial": false,
"alternatives": [
{
"transcript": "はい、じゃあ始めます。ちょっとまだ来られてない方もいらっしゃるんですけど、",
"items": [
{
"start_time": 0.855,
"end_time": 1.125,
"item_type": "pronunciation",
"content": "はい",
"vocabulary_filter_match": false,
"speaker": null,
"confidence": 0.9709,
"stable": null
},
{
"start_time": 1.125,
"end_time": 1.125,
"item_type": "punctuation",
"content": "、",
"vocabulary_filter_match": false,
"speaker": null,
"confidence": null,
"stable": null
},
{
"start_time": 1.345,
"end_time": 1.805,
"item_type": "pronunciation",
"content": "じゃあ",
"vocabulary_filter_match": false,
"speaker": null,
"confidence": 0.8582,
"stable": null
},
{
"start_time": 1.805,
"end_time": 2.055,
"item_type": "pronunciation",
"content": "始め",
"vocabulary_filter_match": false,
"speaker": null,
"confidence": 0.9531,
"stable": null
},
{
"start_time": 2.055,
"end_time": 2.285,
"item_type": "pronunciation",
"content": "ます",
"vocabulary_filter_match": false,
"speaker": null,
"confidence": 0.9017,
"stable": null
},
{
"start_time": 2.285,
"end_time": 2.285,
"item_type": "punctuation",
"content": "。",
"vocabulary_filter_match": false,
"speaker": null,
"confidence": null,
"stable": null
},
{
"start_time": 2.405,
"end_time": 2.645,
"item_type": "pronunciation",
"content": "ちょっと",
"vocabulary_filter_match": false,
"speaker": null,
"confidence": 0.992,
"stable": null
},
{
"start_time": 2.645,
"end_time": 3.025,
"item_type": "pronunciation",
"content": "まだ",
"vocabulary_filter_match": false,
"speaker": null,
"confidence": 0.9966,
"stable": null
},
{
"start_time": 3.025,
"end_time": 3.135,
"item_type": "pronunciation",
"content": "来",
"vocabulary_filter_match": false,
"speaker": null,
"confidence": 0.9169,
"stable": null
},
{
"start_time": 3.135,
"end_time": 3.325,
"item_type": "pronunciation",
"content": "られ",
"vocabulary_filter_match": false,
"speaker": null,
"confidence": 0.9559,
"stable": null
},
{
"start_time": 3.325,
"end_time": 3.485,
"item_type": "pronunciation",
"content": "て",
"vocabulary_filter_match": false,
"speaker": null,
"confidence": 0.9967,
"stable": null
},
{
"start_time": 3.485,
"end_time": 3.725,
"item_type": "pronunciation",
"content": "ない",
"vocabulary_filter_match": false,
"speaker": null,
"confidence": 0.991,
"stable": null
},
{
"start_time": 3.725,
"end_time": 4.075,
"item_type": "pronunciation",
"content": "方",
"vocabulary_filter_match": false,
"speaker": null,
"confidence": 0.9811,
"stable": null
},
{
"start_time": 4.075,
"end_time": 4.325,
"item_type": "pronunciation",
"content": "も",
"vocabulary_filter_match": false,
"speaker": null,
"confidence": 0.996,
"stable": null
},
{
"start_time": 4.325,
"end_time": 5.215,
"item_type": "pronunciation",
"content": "いらっしゃる",
"vocabulary_filter_match": false,
"speaker": null,
"confidence": 0.9788,
"stable": null
},
{
"start_time": 5.215,
"end_time": 5.445,
"item_type": "pronunciation",
"content": "ん",
"vocabulary_filter_match": false,
"speaker": null,
"confidence": 0.9984,
"stable": null
},
{
"start_time": 5.445,
"end_time": 5.685,
"item_type": "pronunciation",
"content": "です",
"vocabulary_filter_match": false,
"speaker": null,
"confidence": 0.9994,
"stable": null
},
{
"start_time": 5.685,
"end_time": 6.435,
"item_type": "pronunciation",
"content": "けど",
"vocabulary_filter_match": false,
"speaker": null,
"confidence": 0.9964,
"stable": null
},
{
"start_time": 6.435,
"end_time": 6.435,
"item_type": "punctuation",
"content": "、",
"vocabulary_filter_match": false,
"speaker": null,
"confidence": null,
"stable": null
}
]
}
],
"channel_id": "ch_0"
}
]
}
**********
TRANSCRIPT: はい、じゃあ始めます。ちょっとまだ来られてない方もいらっしゃるんですけど、
**********
{
"results": [
{
"result_id": "f2a5f3af-8ee8-4185-a892-3ba3df6396b0",
"start_time": 6.925,
"end_time": 7.935,
"is_partial": true,
"alternatives": [
{
"transcript": "ボイスラン",
"items": [
{
"start_time": 6.975,
"end_time": 7.215,
"item_type": "pronunciation",
"content": "ボイス",
"vocabulary_filter_match": false,
"speaker": null,
"confidence": null,
"stable": null
},
{
"start_time": 7.215,
"end_time": 7.405,
"item_type": "pronunciation",
"content": "ラン",
"vocabulary_filter_match": false,
"speaker": null,
"confidence": null,
"stable": null
}
]
}
],
"channel_id": "ch_0"
}
]
}
**********
TRANSCRIPT: ボイスラン
(snip)
このあたりの様子をわかりやすく表示するようにしてみた。
transcribe.py
(snip)
import sys
import time
from colorama import Fore, Style
class ColoredEventHandler(TranscriptResultStreamHandler):
def __init__(self, stream):
super().__init__(stream)
self.start_time = None # 音声の開始時刻
async def handle_transcript_event(self, transcript_event: TranscriptEvent):
if self.start_time is None:
self.start_time = time.time() # 音声開始時間を記録
results = transcript_event.transcript.results
for result in results:
transcript_text = result.alternatives[0].transcript # 最初の候補を取得
elapsed_time = time.time() - self.start_time # 経過秒数
minutes, seconds = divmod(elapsed_time, 60) # 分と秒に分割
timestamp = f"[{int(minutes):02}:{seconds:06.3f}]" # MM:SS.mmm 形式
if result.is_partial:
sys.stdout.write(f"\r{Fore.GREEN}{timestamp} {transcript_text}{Style.RESET_ALL}")
sys.stdout.flush()
else:
sys.stdout.write(f"\n{Fore.WHITE}{timestamp} {transcript_text}{Style.RESET_ALL}\n")
sys.stdout.flush()
(snip)
async def basic_transcribe():
(snip)
#handler = MyEventHandler(stream.output_stream)
#handler = RawResponseHandler(stream.output_stream)
handler = ColoredEventHandler(stream.output_stream)
(snip)
こんな感じで動作する。緑が"is_partial": trueの「未確定」状態で、白が"is_partial": falseで最終確定した状態。途中、緑の行が連続して続いてるのだけど、これはターミナルの横幅に収まってないせいみたいで、実際は横幅を広げてやるとその行は逐次更新されている。
kun432過去にWhisperやGoogle STTを試したときの印象からすると一番精度高いじゃないだろうか？Transcribe、昔少し触っただけでほとんど触ってなかったのだけど、改めて一通り触ってみようという気になった。
kun432Amazon Transcribe、改めて見直す機会になったので、冒頭のZennの記事に感謝。
kun432なお、マイクからのリアルタイム音声処理については、Zennの記事の方はpyaudioを使っているが、公式のレポジトリにあるサンプルではsounddeviceを使用している。
kun432Transcribe、昔少し触っただけでほとんど触ってなかったのだけど、改めて一通り触ってみようという気になった。
ということでこちらで試す
kun432Transcribeで入力可能なフォーマットを見てたけど、FLAC以外の形式だと使い分けが必要になってしまうので、バッチもストリーミングも対応していて推奨されているFLACを使うのが良さそう。
kun432マイクからのリアルタイムも試してみた。赤が「未確定状態」（"is_partial": true）、白が「最終確定」の結果で出力されるようにしてある。
uv pip install sounddevice
transcribe_mic.py
iimport asyncio
import sounddevice
from amazon_transcribe.client import TranscribeStreamingClient
from amazon_transcribe.handlers import TranscriptResultStreamHandler
from amazon_transcribe.model import TranscriptEvent
import sys
import time
from colorama import Fore, Style
class MyEventHandler(TranscriptResultStreamHandler):
def __init__(self, stream):
super().__init__(stream)
self.start_time = None # 音声の開始時刻
async def handle_transcript_event(self, transcript_event: TranscriptEvent):
if self.start_time is None:
self.start_time = time.time() # 音声開始時間を記録
results = transcript_event.transcript.results
for result in results:
transcript_text = result.alternatives[0].transcript # 最初の候補を取得
elapsed_time = time.time() - self.start_time # 経過秒数
minutes, seconds = divmod(elapsed_time, 60) # 分と秒に分割
timestamp = f"[{int(minutes):02}:{seconds:06.3f}]" # MM:SS.mmm 形式
# 未確定の場合は赤色で表示
if result.is_partial:
sys.stdout.write(f"{Fore.RED}{timestamp} {transcript_text}{Style.RESET_ALL}\n")
sys.stdout.flush()
# 確定の場合は白色で表示
else:
sys.stdout.write(f"{Fore.WHITE}{timestamp} {transcript_text}{Style.RESET_ALL}\n")
sys.stdout.flush()
async def mic_stream():
# この関数は、マイクからの生の入力ストリームをラップし、ブロックをasyncio.Queueに転送する
loop = asyncio.get_running_loop() # 現在実行中のイベントループを取得
input_queue = asyncio.Queue()
def callback(indata, frame_count, time_info, status):
loop.call_soon_threadsafe(input_queue.put_nowait, (bytes(indata), status))
# 使用する言語に適したオーディオフォーマットのパラメータを正しく設定
# 詳細は AWS の公式ドキュメントを参照:
# https://docs.aws.amazon.com/transcribe/latest/dg/streaming.html
# 以下はPCM（リトルエンディアン 16-bit, 16kHz, モノラル）の場合
chunk_duration_in_millisecond = 100 # nミリ秒ごとに音声データを読み取るか？（50〜200が推奨）
stream = sounddevice.RawInputStream(
# モノラル（1チャンネル）
channels=1,
# サンプリングレート
samplerate=16000, # サンプリングレート
# 音声データを受け取るコールバック関数
callback=callback,
# PCMデータの 1回の読み取りで処理するチャンクサイズ
# blocksize = chunk_duration_in_millisecond / 1000 * サンプリングレート * 2
# 注: シングルチャネルの場合は2の倍数、デュアルチャネルの場合は4の倍数である必要がある
blocksize=int(chunk_duration_in_millisecond / 1000 * 16000 * 2),
# 音声データの型（16-bit PCM）
dtype="int16",
)
# オーディオストリームを開始し、利用可能になったオーディオチャンクを非同期に生成
with stream:
while True:
indata, status = await input_queue.get()
yield indata, status
async def write_chunks(stream):
# マイクからの生のオーディオチャンクジェネレータを接続し、それらを文字起こしストリームに渡す
async for chunk, status in mic_stream():
await stream.input_stream.send_audio_event(audio_chunk=chunk)
await stream.input_stream.end_stream()
async def basic_transcribe():
# 選択したAWSリージョンでクライアントを設定
client = TranscribeStreamingClient(region="ap-northeast-1")
# 文字起こしを開始して非同期ストリームを生成
stream = await client.start_stream_transcription(
language_code="ja-JP",
media_sample_rate_hz=16000,
media_encoding="pcm",
)
# ハンドラーをインスタンス化し、イベントの処理を開始
handler = MyEventHandler(stream.output_stream)
await asyncio.gather(write_chunks(stream), handler.handle_events())
if __name__ == "__main__":
# 最新のasyncio.run()を使用してメイン関数を実行
asyncio.run(basic_transcribe())
実行
uv run transcribe_mic.py
実際に動かしてみたサンプル
このスクラップは2025/02/05にクローズされました