Speech to text の基本的な使い方と実用例

こんにちは！クラウドエース株式会社 第二開発部の三浦です。

こちらの記事では、Google Cloud が提供する Speech-to-Text の基本的な使い方から Cloud Functions や Pub/Sub と組み合わせた実践的な活用事例まで、わかりやすく解説していきます。

Speech-to-Text とは？

Speech-to-Text とは？

Speech-to-Text は、人間の声をコンピュータが理解できる「文字データ」に変換する技術です。このサービスは、以下のような主な機能があります。

高精度な音声認識： 120 以上の言語と方言に対応しています。つまり、日本語や英語はもちろん、他の多くの言語でも、高精度で音声をテキストに変換できます。
豊富なカスタマイズオプション： 特定の業界用語や商品の名称など、独自の単語を学習させることができます。さらに、音声認識モデルを特定のニーズに合わせてカスタマイズすることも可能です。
リアルタイム処理： 音声が入力されると同時に、リアルタイムでテキストに変換することができます。これにより、ライブ講演やウェビナーなどで即座に文字起こしが可能です。また、音声コマンドを即時に処理することもできます。

Speech-to-Text の使い方

Speech-to-Text を利用するには、通常、API を使用します。API を使うことで、開発者は簡単にアプリケーションに音声認識機能を組み込むことができます。
Speech-to-Text API へのリクエストには、音声データに加えて、主に以下の構成情報が必要です。

• 音声データ： ローカルファイル、Cloud Storage URI、またはストリーミング入力など。
• 音声データ形式： FLAC、LINEAR16 など。
• 言語情報： 日本語（ja-JP）、英語（en-US）など。
• 認識モデル： デフォルトモデル、特定の分野向けモデルなど。

Speech-to-Textリクエストの基本構成

音声データの提供方法

リクエストの一部として音声データを直接 API に送信する方法で、埋め込み音声で音声データを直接 API に送り込むことができ、gRPC や REST を利用して Base64 形式 でエンコードしたデータとして扱います。

構成情報の設定

構成情報は以下の観点で決定します。
音声メタデータの活用: 大半の音声ファイルについて、自動推測できるメタデータを利用して音声の解析が行われます。
サンプルレートの設定: サンプルレートは 16,000 Hz が理想です。高すぎる設定にすると効果が薄く、低すぎると精度に影響します。
言語の指定: BCP-47 コードを使用して、音声の言語や地域の方言を指定できます。対応言語は非常に豊富です。
※BCP-47 コードとは、言語を識別するための標準的なコード（言語タグ）の形式
モデルの選択: Speech-to-Text では、特定の用途や音声タイプに合わせてトレーニングされたモデルが用意されています。RecognitionConfig 内の model パラメータで指定します。

例えば、以下のようなモデルがあります。
latest_short： 短い発話（数秒程度）の認識に適した最新モデル。
latest_long： 長い音声ファイル（会議の録音や動画など）の認識に適した最新モデル。
telephony： 電話での会話のように、低いサンプルレート（8kHzなど）で録音された音声に最適化されたモデル。
medical_conversation： 医師と患者の会話など、医療分野の対話に特化したモデル。
medical_dictation： 医療従事者がメモを取る際のような、医療分野の口述に特化したモデル。

このように、音声データの内容や録音状況に最も適したモデルを選択することで、それぞれの分野や状況に最適化された、より高精度な文字起こしが期待できます。指定しない場合は、多くの場合、汎用的なモデルが使用されます。

音声データの変換方法

この表は、音声データをテキスト化する際に用いられる「同期認識」「非同期認識」「ストリーミング認識」という3つの異なるアプローチについて解説しています。

認識方法	概要	特徴	用途
同期認識	音声データを Speech-to-Text API に送信し、即時に結果を取得する方法	1分以内の短い音声データに対して使用。すべての音声が処理された後で、単一のレスポンスが返される	会話の一部や短い音声メモのテキスト化に最適
非同期認識	こちらは音声データを送信し、非同期に長時間の処理を行う方法	最大 480 分の長時間音声データに対応し、定期的に進捗や結果をポーリングすることで更新された結果を得られる	会議やセミナーなど、長時間の音声記録のテキスト化に適している
ストリーミング認識	gRPC プロトコルを利用したリアルタイム音声認識方法。	マイクからのライブ音声をリアルタイムでキャプチャし、暫定的な結果を生成し続けることができる。ユーザーが話している間にも逐次的に結果が得られる	対話型アプリケーションやライブイベントのキャプション生成に最適

※gRPC は、Google が開発した高性能なリモートプロシージャコール(RPC)フレームワークです。

実用例

ここでは、Speech-to-Text の実用例をより詳しく紹介します。
Google のサイトによると、Speech-to-Text には以下のような実用例があるようです。

業界	活用内容	効果
教育機関	英会話学習アプリが、ユーザーの発話をリアルタイムでテキスト化	学習者は自身の発音の正確さを即座に確認でき、発音練習や改善が容易になった
音楽業界	ユーザーが歌った歌詞を正確に認識するロボットを開発	ロボットが歌詞に基づいた対話を行い、インタラクティブな音楽体験を提供可能になった
金融サービス	顧客との電話応対をリアルタイムでテキスト化し、記録として保存	コミュニケーションが効率化され、サービス品質の向上に貢献した
自動車産業	音声認識アプリのバックエンドにSpeech-to-Textを活用	運転中のハンズフリー操作（ナビゲーション等）が可能となり、安全性が向上した
コールセンター	CRMシステムに組み込み、オペレーターが通話内容をリアルタイムでテキスト化	重要事項の即時記録が可能となり、応対内容の分析や改善に役立ってる
保険業界	営業職員向けセルフトレーニング（ロールプレイ等）で発言内容をテキスト化	テキスト化された内容に基づきフィードバックを提供することで、トレーニング効果を高めた

https://cloud.google.com/customers?hl=ja&sr=Cg5TcGVlY2ggdG8gdGV4dCIlCAESIRofQ09SUFVTX1RZUEVfQ1VTVE9NRVJfQ0FTRV9TVFVEWSgMOgYaBAoCamE

Speech-to-Text の基本的な使い方

まず、最も基本的な使い方として、Cloud Storage に保存された音声ファイルを Speech-to-Text でテキストに変換する方法を紹介します。

注意点
コンソール上での操作
Google Cloud のコンソール画面は、アップデートによって UI や操作方法が変更される場合があります。

同期認識：単一ファイルの処理

音声ファイルを Cloud Storage にアップロード

1. コンソール上から Cloud Storage バケットを作成または選択します。
   
2. 変換したい音声ファイル（例: .mp3, .wav）をバケットにアップロードします。
   
3. アップロードしたファイルの URI（gs://your-bucket-name/your-file.mp3のような形式）を控えておきます。

Python で Speech-to-Text を実行:

1. Python クライアントライブラリを使って SpeechClient (主要なクラス)を初期化します。

client = speech.SpeechClient()

2. 認識設定（RecognitionConfig）で、言語コード（日本語なら"ja-JP"）、音声ファイルのエンコーディング（例: MP3）、サンプリングレートなどを指定します。

config = speech.RecognitionConfig(
    encoding=speech.RecognitionConfig.AudioEncoding.MP3, # 音声ファイルのエンコーディング
    sample_rate_hertz=16000,                             # サンプリングレート
    language_code="ja-JP",                               # 言語コード（日本語）
)

3. 音声ファイルの URI（RecognitionAudio）と認識設定を recognize メソッド(SpeechClient に含まれる主要なメソッド)に渡し、音声認識を実行します。

 print(f"処理中: {gcs_uri}")
    response = client.recognize(config=config, audio=audio)

4. 返却されたレスポンスから、認識結果のテキストを取得します。

for result in response.results:
        print(f"文字起こし結果: {result.alternatives[0].transcript}")

# 実行
# if __name__ == "__main__":
#     sync_transcribe_gcs("gs://your-bucket-name/your-file.mp3")

非同期認識（バッチ処理）：長時間・複数ファイルの処理

複数ファイルの準備: 処理したい音声ファイルそれぞれの URI をリストなどにまとめます。

バッチリクエストの作成

1. BatchRecognizeRequest (バッチ認識のリクエスト)を使用して、複数のファイルに対するリクエストを構成します。

request = speech.BatchRecognizeRequest(
    recognizer=f"projects/{project_id}/locations/global/recognizers/_",
    config=config, # 共通の認識設定
    files=files,   # 処理対象のファイルリスト
    recognition_output_config=output_config, # 結果の出力先
)

2. 共通の認識設定（言語、モデルなど）と、各ファイルの URI 情報（BatchRecognizeFileMetadata）を設定します。

# ファイルのリストを作成
files_metadata = [
    speech.BatchRecognizeFileMetadata(
        uri="gs://your-input-bucket/audio_jp_1.wav"
    )
]

非同期処理の実行と結果取得

1. batch_recognize メソッド（または long_running_recognize）で非同期処理を開始します。この処理はバックグラウンドで実行されるため、完了を待つ間、他の作業を進めることができます。

# (A) 非同期処理を開始します
# この呼び出しはすぐに完了し、バックグラウンドでの処理が始まります。
operation = client.batch_recognize(request=request)

print(f"バックグラウンドで処理を開始しました。")
print("完了まで待機しています...")

# (B) 処理の完了を待ちます
# .result() を呼び出すと、バックグラウンド処理が終わるまでここで待機します。
response = operation.result()

print("処理が完了しました！")

2. 処理が完了すると、指定した Cloud Storage バケットに結果が出力されます。
   
3. 出力された結果ファイルをダウンロードし、各ファイルの認識結果を確認します。 非同期認識は、API 呼び出しの待ち時間を最小限に抑えつつ、大規模な音声データ処理を効率的に行うための強力な手段です。
   

自動化へのステップ：Cloud Functions で音声処理をトリガー

次は Speech-to-Text 処理の自動化についてご紹介いたします。Cloud Functions を使えば、特定のイベントをきっかけに、自動的に Speech-to-Text を実行させることができます。
ここでは、「Cloud Storage に音声ファイルがアップロードされたら、自動的にテキスト化する」という仕組みを構築します。

Cloud Functions とは？

Cloud Functions は、サーバーの管理を気にせずにコードを実行できるイベント駆動型のサーバーレス・コンピューティングサービスです。Cloud Storage へのファイル追加、Pub/Sub メッセージの受信、HTTP リクエストなどをトリガーとして、定義した関数を自動実行します。
https://cloud.google.com/functions/docs/concepts/overview?hl=ja

構成図

プロセスの流れ

1. トリガー設定
   Cloud Functions で、特定の Cloud Storage バケットへのファイル追加（google.storage.object.finalize イベント）をトリガーとして設定します。
2. 関数実行
   音声ファイルがバケットにアップロードされると、関数が自動的に呼び出されます。
3. ファイル情報取得
   関数内で、イベントデータからアップロードされたファイルのバケット名とファイル名を取得します。
4. Speech-to-Text 実行
   取得したファイル情報（URI）を使って、Speech-to-Text APIを呼び出し、音声認識を実行します（基本的な使い方と同様の手順。長時間ファイルの場合は非同期認識を検討）。
5. 結果処理とログ出力
   認識結果のテキストを取得します。
   必要に応じて、テキスト内に特定のキーワードが含まれているかチェックし、ログに出力したり、別の処理を実行したりします。
   処理の状況や結果を Cloud Logging に出力し、後で確認できるようにします。
   

リアルタイム分析とスケーラビリティ：Pub/Sub の活用

さらに進んで、リアルタイム性とスケーラビリティが求められるシステムを構築する場合、Pub/Sub の導入が効果的です。
ここでは、「Cloud Storage に音声ファイルがアップロードされたら自動でテキスト化し、不適切な単語が含まれていたら Pub/Sub 経由で通知し、結果をStorageに保存する」という、音声コンテンツモデレーションシステムを Cloud Functions で実装する例を紹介します。

Pub/Sub とは？

Pub/Sub は、メッセージの発行者（Publisher）と購読者（Subscriber）を分離するフルマネージドのリアルタイム・メッセージングサービスです。アプリケーション間で非同期にメッセージをやり取りするのに適しており、スケーラビリティにも優れています。
https://cloud.google.com/pubsub/docs/overview?hl=ja

プロセスの流れ

1. トリガー
   Cloud Storage へのファイルアップロード（google.storage.object.finalize）が Cloud Functions をトリガーします。
2. ファイル情報取得
   関数内でイベントデータからファイル情報を取得します。
3. テキスト変換
   Speech-to-Text API の非同期認識 (long_running_recognize) を呼び出し、アップロードされた音声ファイルをテキストに変換します。
4. 不適切単語チェック
   事前に定義された不適切な単語リストを用意します。
   変換されたテキストに、リスト内の単語が含まれていないかチェックします。
5. Pub/Sub への通知
   不適切な単語が見つかった場合、ファイル名、検出された単語、テキスト全文などの情報を含むメッセージを作成します。
   Pub/Sub Publisher クライアントを使って、このメッセージを特定のトピックに発行（Publish）します。
   
6. 結果の保存
   Pub/Sub に通知したメッセージの内容を、後で確認できるように Cloud Storage にテキストファイルとして保存します。
   
   ログ出力：処理の状況や結果（不適切単語が見つからなかった場合を含む）を Cloud Logging に出力します。

サンプルスクリプト

サンプルスプリクトは Python で記載しています。
main.py で作成します。

import logging
import google.cloud.logging
from google.cloud import storage
from google.cloud import speech_v2 as speech
from google.cloud import pubsub_v1

logging_client = google.cloud.logging.Client()
logging_client.setup_logging()

# Google Cloud プロジェクトID
PROJECT_ID = "your-google-cloud-project-id"

# Cloud Storage バケット名
BUCKET_NAME = "sample-bucket"

# Pub/Sub トピックID
TOPIC_ID = "moderation-topic"

# 不適切な単語リスト
BAD_WORDS = ["不適切な単語1", "不適切な単語2", "不適切な単語3"]

# Storage クライアントのインスタンスを作成
storage_client = storage.Client()
bucket = storage_client.bucket(BUCKET_NAME)

# Speech-to-Text クライアントのインスタンスを作成
speech_client = speech.SpeechClient()

# Publisher クライアントのインスタンスを作成
publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path(PROJECT_ID, TOPIC_ID)

# --- Recognizerの作成 ---
# Recognizerは音声認識の設定（言語、モデル等）を保存するリソースです
# 関数が起動するたびに毎回作り直さないよう、存在を確認してから作成する効率的な実装です
LOCATION = "global"
RECOGNIZER_ID = "your-recognizer-id"
recognizer_path = f"projects/{PROJECT_ID}/locations/{LOCATION}/recognizers/{RECOGNIZER_ID}"
try:
    # 既存のRecognizerを取得できるか試します
    speech_client.get_recognizer(name=recognizer_path)
    logging.info(f"Recognizer '{RECOGNIZER_ID}' はすでに存在します。")
except Exception:
    # 存在しない場合は新規作成します
    logging.info(f"Recognizer '{RECOGNIZER_ID}' を作成します。")
    recognizer_config = speech.Recognizer(
        language_codes=["ja-JP"],
        model="latest_long", # 長時間音声用の最新モデル
    )
    operation = speech_client.create_recognizer(
        parent=f"projects/{PROJECT_ID}/locations/{LOCATION}",
        recognizer_id=RECOGNIZER_ID,
        recognizer=recognizer_config,
    )
    operation.result()  # 作成完了を待ちます
    logging.info(f"Recognizer '{RECOGNIZER_ID}' を作成しました。")

def check_bad_words(text):
    """テキストに不適切な単語が含まれているかどうかをチェックする関数
    found_bad_words = []
    for bad_word in BAD_WORDS:
        if bad_word in text:
            found_bad_words.append(bad_word)
    return found_bad_words

def save_message_to_storage(message):
    """メッセージをCloud Storageに保存する関数
    file_name = f"message-{message['file']}-{message['bad_words']}.txt"
    blob = bucket.blob(file_name)
    blob.upload_from_string(str(message), content_type='text/plain')
    logging.info(f"メッセージを Cloud Storage に保存しました: gs://{BUCKET_NAME}/{file_name}")

def process_audio(data, context):
    """Cloud Storage に音声ファイルが追加された際にトリガーされるメイン関数"""
    # Cloud Functionsに渡されたイベントデータから、ファイル名とバケット名を取得します
    bucket_name = data['bucket']
    file_name = data['name']
    gcs_uri = f"gs://{bucket_name}/{file_name}"

    # Speech-to-Text v2 APIへのリクエストを作成します
    config = speech.RecognitionConfig(
        auto_decoding_config=speech.AutoDetectDecodingConfig() # ファイル形式を自動検出
    )
    request = speech.BatchRecognizeRequest(
        recognizer=recognizer_path,
        config=config,
        files=[{"uri": gcs_uri}],
        recognition_output_config=speech.types.RecognitionOutputConfig(
            # 結果をGCSに出力せず、直接レスポンスとして受け取る設定
            inline_response_config={}
        ),
    )

    logging.info(f"音声ファイル {gcs_uri} の認識処理を開始します。")
    try:
        # 音声認識の非同期処理を開始し、完了を待ちます
        operation = speech_client.batch_recognize(request=request)
        response = operation.result()
        logging.info(f"音声ファイル {gcs_uri} の認識処理が完了しました。")
    except Exception as e:
        logging.error(f"音声認識処理中にエラーが発生しました: {e}")
        return

    # v2 APIのレスポンス構造に合わせて、文字起こし結果を抽出します
    transcript = ""
    result = response.results.get(gcs_uri)

    if not result:
        logging.warning(f"レスポンスにファイル {gcs_uri} の結果が含まれていません。")
        return
    if result.error:
        logging.error(f"ファイル {gcs_uri} の文字起こしでエラーが発生しました: {result.error}")
        return
    if result.transcript:
        transcript = "".join([alt.transcript for alt in result.transcript.alternatives])

    # 認識結果をログに出力
    logging.info(f"認識結果: {transcript}") 

    # ヘルパー関数を呼び出し、不適切な単語が含まれているかチェックします
    found_words = check_bad_words(transcript)

    # 不適切な単語が見つかった場合の処理
    if found_words:
        # 見つかった全ての単語をカンマ区切りで結合します
        bad_words_str = ",".join(found_words)
        logging.info(f"不適切なキーワード'{bad_words_str}'が検出されました。")

        message = {
            "file": file_name,
            "bad_words": bad_words_str,
            "transcript": transcript,
        }

        # Pub/Subへ通知し、結果をStorageに保存します
        try:
            future = publisher.publish(topic_path, str(message).encode("utf-8"))
            logging.info(f"Pub/Subにメッセージを送信しました: {message}, result: {future.result()}")
            save_message_to_storage(message)
        except Exception as e:
            logging.error(f"Pub/Subへの通知またはStorageへの保存中にエラーが発生しました: {e}")
            
    # 不適切な単語が見つからなかった場合の処理
    else:
        logging.info(f"音声ファイル {file_name} の処理が完了しました。不適切な単語は見つかりませんでした。")

ライブラリ

ライブラリは、それぞれの Google Cloud サービスと Python プログラムからやり取りするための機能を提供するため必要となります。
main.py と同じディレクトリに requirements.txt を作成します。

google-cloud-storage
google-cloud-speech
google-cloud-pubsub
google-cloud-logging

その他の機能

リアルタイムストリーミング認識の使い方

ここではリアルタイムストリーミングの特筆すべき点をご紹介します。

1. マイク入力の設定

pyaudio を利用して、マイクロフォン入力から直接音声ストリームを確保します。音声データを小さなチャンクサイズごとに分割し、APIに送信するための準備を行います。
※pyaudio は、Python でオーディオを扱うためのライブラリです。

2. ストリーミング認識の実行

StreamingRecognitionConfig を用いて、リアルタイムで音声データを認識する設定を施します。設定には、サンプルレートや言語コード、エンコーディング形式などが含まれます。
※Speech-to-Text API で音声認識を行う際の設定を指定する設定オブジェクトです。

3. レスポンスの処理

API からのレスポンスを受け取り、その認識結果をリアルタイムでコンソールに出力します。これは、会話のインタラクティブな文字化に最適です。
特定のキーワード（例: "exit"や"quit"）を検出した場合には、システムを停止させる機能も含まれています。

使用例と利点

このリアルタイムストリーミング技術は、ライブイベントや対話型システム、ライブのキャプション生成などに非常に有用です。リアルタイムでの処理が可能なため、迅速なレスポンスが求められる環境で大きな力を発揮します。

合成音声技術 Text-to-Speech

ここからは、Text-to-Speech という合成音声技術について紹介します。
Text-to-Speech は Speech-to-Text とは逆の機能を持つ技術です。
合成音声技術は、テキストを音声に変換する技術として多くの場面で活用されています。この技術はナレーションや音声アシスタント、教育ツール、アクセシビリティの向上など、様々な用途で利用されています。

設定画面の紹介

• テキスト入力
  変換したいテキストを入力する欄です。例えば、「こんにちは」と入力されているように、音声化したい文章をここに記載します。
• 言語設定
  音声合成に使用する言語を選択することができます。画像では「Japanese (Japan) - ja-JP」が選択されています。この設定により、話者の言語とそのアクセントが決まります。
• 音声の選択
  利用可能な音声モデルを選択します。例として「ja-JP-Chirp3-HD-Aoede」が選択されています。このモデルにより、音質や話し方、声質が異なります。
  

技術の利点

適応性: 異なる声質やアクセントを持つ複数の音声を選択できるため、ニーズに応じたカスタマイズが可能です。
アクセシビリティ: 聴覚に課題のある方や、視覚障害者に対する情報提供に役立っています。
効率性: 大量のテキストを短時間で音声化できるため、時間と労力を節約します。

まとめ

本記事では、Google Cloud の Speech-to-Text を中心に、その基本的な使い方から Cloud Functions や Pub/Sub と組み合わせた自動化、リアルタイム分析といった実践的な活用事例まで、幅広くご紹介しました。
Speech-to-Text は、高精度な音声認識技術を容易に利用可能にする強力なサービスです。単純な文字起こしだけでなく、他の Google Cloud サービスと連携させることで、業務プロセスの自動化、リアルタイムでのデータ分析、インタラクティブなアプリケーション開発など、アイデア次第で様々な可能性が広がります。
この記事を通じて、Speech-to-Text の魅力や活用のヒントを感じていただけたなら幸いです。

参考 URL

https://cloud.google.com/speech-to-text/docs/speech-to-text-requests?hl=ja