無音時に API を停止＋ウェイクワードでコマンドを受け付ける Google Cloud Speech-to-Text 実装例

---

この記事では、以下の 2つを 同時に 実現するサンプルコードを紹介します。

1. 無音検出 (VAD) を用いて、音声がないときは Google Cloud Speech-to-Text のストリーミング接続を停止し、課金を抑える。
2. ウェイクワード (例：「コンピュータ」) を検出した後に、特定のコマンドを受け付ける。

サンプルコードはあくまで デモ用 であり、実際の運用ではエラー処理やスレッド管理、再接続遅延などを十分に考慮してください。

---

概要フローチャート

• VAD is_speech?: 音声フレームに声が含まれるかどうかを WebRTC VAD で判定
• If streaming is OFF, Start streaming API: 音声が検出されたにもかかわらずストリーミングが停止状態の場合は、Google Cloud Speech-to-Text のストリーミングを開始
• Check silence timeout: 一定時間以上、無音状態が続けばストリーミングを停止し課金を抑える
• Contains WAKE_WORD?: 認識結果に「コンピュータ」などのウェイクワードが含まれているか判定
• Contains supported command?: 「ログイン」「発進」などのコマンドを含んでいれば対応処理を実行

---

1. 事前準備

1.1. 必要ライブラリのインストール

pip install google-cloud-speech pyaudio webrtcvad six

1.2. Google Cloud Speech-to-Text の有効化 & 認証ファイル

• Google Cloud Console でプロジェクトを作成し、Speech-to-Text API を有効化します。
• サービスアカウントキー (JSON) を作成・ダウンロードし、以下のように環境変数を設定してください。

export GOOGLE_APPLICATION_CREDENTIALS="/path/to/your-service-account-file.json"

---

2. サンプルコード

以下のスクリプトを vad_wakeword_sample.py のようなファイル名で保存し、実行してください。
※ Python 3.7 以上を想定

import os
import time
import queue
import threading

import pyaudio
import webrtcvad
from google.cloud import speech
from six.moves import queue as six_queue

# -----------------------------
# 設定項目
# -----------------------------
WAKE_WORD = "コンピュータ"
SUPPORTED_COMMANDS = {
    "ログイン": "LOGIN",
    "発進": "ENGAGE",
    "被害報告": "REPORT",
    "ログアウト": "LOGOUT",
    "シャットダウン": "SHUTDOWN"
}

RATE = 16000
CHUNK_DURATION_MS = 30  # 1フレームあたりの長さ（ms）
CHUNK_SIZE = int(RATE * CHUNK_DURATION_MS / 1000)  # 30msのサンプル数
VAD_SILENCE_TIMEOUT = 2.0  # 無音と判断してから何秒後にAPI停止するか

# -----------------------------
# 無音検出 (VAD) の設定
# -----------------------------
vad = webrtcvad.Vad()
vad.set_mode(3)  # 感度(0〜3) 3が最も感度が高い

def is_speech(frame, sample_rate=RATE):
    """
    WebRTC VAD を用いて、フレームに音声が含まれているかどうかを判定
    """
    return vad.is_speech(frame, sample_rate)


# -----------------------------
# マイク入力を管理するクラス
# -----------------------------
class MicrophoneStream(object):
    """
    PyAudioでマイク入力を取得し、frame単位で取り出せるようにするクラス
    """
    def __init__(self, rate, chunk_size):
        self._rate = rate
        self._chunk_size = chunk_size
        self._buff = queue.Queue()
        self._closed = True

    def __enter__(self):
        self._audio_interface = pyaudio.PyAudio()
        # マイクストリームを開く
        self._audio_stream = self._audio_interface.open(
            format=pyaudio.paInt16,       # 16bit
            channels=1,                   # モノラル
            rate=self._rate,
            input=True,
            frames_per_buffer=self._chunk_size,
            stream_callback=self._fill_buffer,
        )
        self._closed = False
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self._audio_stream.stop_stream()
        self._audio_stream.close()
        self._closed = True
        self._buff.put(None)
        self._audio_interface.terminate()

    def _fill_buffer(self, in_data, frame_count, time_info, status_flags):
        """
        PyAudioのコールバック。取得した音声データをバッファに詰める
        """
        self._buff.put(in_data)
        return None, pyaudio.paContinue

    def get_frame(self):
        """
        1フレーム(30ms分)の音声データを返す
        """
        chunk = self._buff.get()
        if chunk is None:
            return None
        return chunk


# -----------------------------
# Speech-to-Text ストリーミング関連
# -----------------------------
class SpeechToTextStreamer:
    """
    Google Cloud Speech-to-Text のストリーミングを扱うクラス
    - start_streaming() で開始
    - stop_streaming()  で停止
    - get_transcript()  で認識結果を取得
    """
    def __init__(self):
        self.client = speech.SpeechClient()
        config = speech.RecognitionConfig(
            encoding=speech.RecognitionConfig.AudioEncoding.LINEAR16,
            sample_rate_hertz=RATE,
            language_code="ja-JP",
            enable_automatic_punctuation=True,
        )
        self.streaming_config = speech.StreamingRecognitionConfig(
            config=config,
            interim_results=True,
        )
        self._requests_queue = six_queue.Queue()
        self._responses_queue = six_queue.Queue()
        self._thread = None
        self._stop_event = threading.Event()

    def _generator(self):
        """
        StreamingRecognizeRequest のジェネレータ
        _requests_queue から音声チャンクを取り出して送信する
        """
        while not self._stop_event.is_set():
            chunk = self._requests_queue.get()
            if chunk is None:
                break  # 停止時にNoneを投げて終了
            yield speech.StreamingRecognizeRequest(audio_content=chunk)

    def _response_loop(self):
        """
        streaming_recognize() のレスポンスを受信し _responses_queue に入れる
        """
        requests = self._generator()
        responses = self.client.streaming_recognize(
            self.streaming_config, requests
        )
        try:
            for res in responses:
                if self._stop_event.is_set():
                    break
                self._responses_queue.put(res)
        except Exception as e:
            print("Exception in response loop:", e)

    def start_streaming(self):
        """
        ストリーミング開始 (スレッド起動)
        """
        self._stop_event.clear()
        self._thread = threading.Thread(target=self._response_loop, daemon=True)
        self._thread.start()

    def stop_streaming(self):
        """
        ストリーミング停止
        """
        self._stop_event.set()
        self._requests_queue.put(None)
        if self._thread is not None:
            self._thread.join()
        self._thread = None

    def send_audio(self, chunk):
        """
        マイクなどから取得した音声チャンクを送信
        """
        if not self._stop_event.is_set():
            self._requests_queue.put(chunk)

    def get_transcript(self):
        """
        キューに溜まっているレスポンスを取り出し、テキストを返す (まとめて)
        """
        texts = []
        while not self._responses_queue.empty():
            response = self._responses_queue.get()
            for result in response.results:
                if not result.alternatives:
                    continue
                transcript = result.alternatives[0].transcript.strip()
                if result.is_final:
                    texts.append(transcript)  # 確定結果
                else:
                    texts.append(transcript + " (interim)")
        return texts


# -----------------------------
# メイン処理
# -----------------------------
def main():
    # os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "/path/to/your-key.json"
    stt_streamer = SpeechToTextStreamer()

    streaming_active = False
    awaiting_command = False
    last_speech_time = time.time()

    print("=== VAD + Wake Word Demo ===")
    print("マイクから音声を取得し、無音が続くとAPIを停止します。")
    print("ウェイクワード『コンピュータ』を検出後、コマンドを受け付けます。")
    print("サポートコマンド:", list(SUPPORTED_COMMANDS.keys()))
    print("=================================================\n")

    with MicrophoneStream(RATE, CHUNK_SIZE) as mic_stream:
        while True:
            frame = mic_stream.get_frame()
            if frame is None:
                break  # マイクが終了した可能性 (Ctrl+C等)

            # 無音 or 音声の判定
            if is_speech(frame, RATE):
                last_speech_time = time.time()
                if not streaming_active:
                    print("[INFO] 音声検出: ストリーミング開始")
                    stt_streamer.start_streaming()
                    streaming_active = True

                # 音声がある間はチャンクを送る
                stt_streamer.send_audio(frame)
            else:
                # 無音フレーム
                if streaming_active:
                    if (time.time() - last_speech_time) > VAD_SILENCE_TIMEOUT:
                        print("[INFO] 無音が続いたためストリーミング停止")
                        stt_streamer.stop_streaming()
                        streaming_active = False

            # ストリーミング中なら認識結果をチェック
            if streaming_active:
                transcripts = stt_streamer.get_transcript()
                for text in transcripts:
                    print(f"[STT] {text}")
                    # ウェイクワード検出
                    if not awaiting_command and WAKE_WORD in text:
                        awaiting_command = True
                        print("ウェイクワード検出！コマンドをお話しください...")

                    # コマンド検出
                    if awaiting_command:
                        detected_command = None
                        for key in SUPPORTED_COMMANDS:
                            if key in text:
                                detected_command = SUPPORTED_COMMANDS[key]
                                break

                        if detected_command:
                            print(f"コマンド検出: {detected_command}")
                            # 実際の処理など
                            awaiting_command = False
                            print("[INFO] コマンド受付完了、再度ウェイクワードを待機中...")


if __name__ == "__main__":
    main()

---

3. 使い方

1. 上記コードを保存して実行:

   python vad_wakeword_sample.py
   

2. マイクに向かって話しかける:
   • 無音だとストリーミングが停止し、課金を抑えられます
   • 音声を発するとストリーミングが開始し、認識結果が出力されます
3. **ウェイクワード「コンピュータ」**と言うと、続けてコマンドを受け付けます
4. 「ログイン」や「発進」などを発話すると、コンソールにコマンドが表示されます

---

4. よくある質問(FAQ)

Q1. ストリーミング中でないときにウェイクワードを言ったら認識されないのでは？

A1. その可能性はあります。このサンプルでは音声と判定されたときにストリーミングを開始しているため、喋り始めるタイミングでまだ開始が間に合わない場合はウェイクワードを逃すことがあります。より確実にしたい場合は、ローカルでウェイクワード検出を行ってからクラウドの認識を始める方法が挙げられます。

Q2. 音声→無音の切り替えに遅延があるのですが？

A2. webrtcvad のフレームサイズや感度の設定によるラグが影響する場合があります。フレームサイズを調整したり、ノイズ環境に合わせてチューニングしてみてください。

Q3. ウェイクワード「コンピュータ」を他の言葉に変えたい

A3. WAKE_WORD をお好きなキーワードに変更してください。ただし日本語だと発音の問題や近音の単語との混同が起きやすいため、実際の運用前に十分なテストが必要です。

Q4. コマンドの認識精度を上げたい

A4. Speech Adaptation (旧 Phrase Hints) を利用し、特定の単語を優先認識させる方法があります。また、マイク性能の向上や周囲ノイズの低減も効果的です。

Q5. 料金をもっと抑えたい

A5. ウェイクワード検出自体もローカルで行えば、クラウド接続時間を大幅に短縮できます。たとえば Picovoice Porcupine 等を使い、ローカルでウェイクワードを確実に検出してからクラウドに接続すると、費用をさらに下げられます。

Q6. PyAudio のインストールがうまくいきません

A6. macOS なら brew install portaudio、Windows なら Visual Studio Build Tools の導入などが必要です。ポータブルなホイールを入手する方法もあります。OSごとのビルド環境を整えてから pip install pyaudio をお試しください。

Q7. 発進 がなぜ「ENGAGE」なのですか？

A7. 「ENGAGE」は『Star Trek: The Next Generation、略称:TNG）』で、宇宙戦艦のワープでの発進を指示するときに使われるフレーズです。雰囲気を出すため、ここでは「発進」の意味で採用しています。

Q8. 単語がフレーム分割で途切れてしまう可能性はないのでしょうか？

A8. 結論から言うと、フレームを分割して送ることで単語が中途半端に切れてしまうリスクはほぼありません。

• Google Cloud Speech-to-Text（ストリーミング認識）は、受け取ったチャンクを連続したオーディオストリームとして結合・解析します。
• たとえ 10ms や 30ms 単位に区切れていても、サーバー側では時間軸上で連続波形として扱われるため、単語の途中で「切れてしまう」ことは基本的に起こりません。
• むしろあまり大きいチャンクサイズにすると無音検出やインタラクティブ性が下がるので、30ms 程度のこまめな送信が一般的です。
• ただし、ネットワークが断続的に切れたり、マイク側で物理的に音声が断続するような特殊状況では、波形が途切れ途切れになる場合もあります。その際は録音環境やネットワーク状態を整えるのが最善策です。

Q9. VAD_SILENCE_TIMEOUT は、最大どれくらいまで設定できますか？

A8. 結論から言うと、VAD_SILENCE_TIMEOUT は理論上いくらでも大きく設定可能です。本サンプルコードでは「何秒以上無音が続いたらストリーミングを止めるか」を決める、単なるアプリケーション側のパラメータだからです。

実装上はこの値を 10 秒でも 600 秒でも、あるいは極端に 1 時間に設定してもコードとしては動作します。ただし以下のような理由で、あまりに長い時間を設定するのは現実的ではありません。

A8-1. 実用上の観点

1. クラウド API (GCP) のストリーミング上限

   • Google Cloud Speech-to-Text にはストリーミング セッションあたりの最大時間などの制限があります。
   • たとえば連続で何十分もストリーミングすると、API の制限 (クォータ) に抵触したり、接続が切れる可能性があります。
   • 公式ドキュメント で最新の制限を確認してください。

2. 課金の増加

   • VAD_SILENCE_TIMEOUT を長く設定すると、実際には無音なのに長時間ストリーミングが続き、無駄な課金が発生します。
   • 「すぐに切断していいのか、それともまだ話し続ける可能性があるか」を考えて、適切なタイムアウトを決める必要があります。

3. ユーザー体験

   • タイムアウトが長いと、「もう誰も話していないのにストリーミングが止まらない」「API を呼び続けて無駄に費用がかかる」などの問題が起きます。
   • 逆に短すぎると、話し終わったときに少しだけ黙っていたら、すぐストリーミングが止まってしまう可能性があります。

A8-2. ライブラリ・実装上の制限は特にない

• サンプルコードの VAD_SILENCE_TIMEOUT は、無音かどうかを判定してから API 切断までどのくらい待つかという、ローカル変数の比較でしかありません。
• webrtcvad 自体や PyAudio は、この値について上限を定めていないので、技術的には無限大の設定も可能です。

A8-3. 適切な値の決め方

1. ユーザーの発話タイミング

   • 会話に区切りがあっても、その後に続けて話す可能性があるなら、やや長め(数秒～十数秒)に設定する。
   • 連続発話が予想されない用途(例: 1回だけコマンドを発話)なら短くしてコストを抑える。

2. クォータ上限や課金状況

   • 1日中ずっと待ち受けするなら、無音検出のスレッショルドを短めにして、こまめに切断するほうが費用を下げられる。

3. 音声認識が中断して困るかどうか

   • 発話の合間が長い用途(ディスカッションや議事録など)で勝手に切断されるのはまずい場合、長めに設定する。
   • 反対に、ミスでずっと ON のままだと逆にコストリスクが大きいなら短めに設定。

---

4. まとめ

• VAD_SILENCE_TIMEOUT はアプリ側の自由なパラメータであり、上限は特にありません。
• ただし、実際には GCP ストリーミングの制限や 課金・運用上の都合、 ユーザー体験 を考慮して、数秒～数十秒程度に設定するケースが多いです。
• 値を大きくすれば、長時間の沈黙でもストリーミングが維持される反面、無駄な課金や クォータ消費が発生しやすくなります。
• 結局はユースケースに合わせて調整するのが最適です。

以上を踏まえ、無音判定タイムアウトは数秒～十数秒程度を目安にしながら、用途に応じて最適化してください。

---

5. 実装のポイント

• 無音検出 (VAD)
  • webrtcvad によりフレーム単位で音声/無音を判定。長時間の無音を検出すると API を停止して費用を抑えられます
• ストリーミング管理
  • SpeechToTextStreamer で start_streaming() / stop_streaming() を制御し、認識結果はスレッドで随時受け取る
• ウェイクワード & コマンド検出
  • 文字列検索で「コンピュータ」を含むかどうか → コマンド待ち状態に
  • 「ログイン」「発進」などを含むかどうかでコマンドを判定

---

6. 注意点 & 応用例

1. 遅延・ノイズ対策
   • フレームサイズや感度を調整、環境ノイズを低減するなど、運用環境に合わせたチューニングが重要
2. コマンドの拡張
   • 認識文字列に自然言語処理を組み合わせれば、より柔軟な音声コマンドが実装可能
3. ローカルウェイクワード
   • ウェイクワードをローカルで高精度に検出し、検出後にのみクラウドストリーミングに接続すれば、コストをさらに削減できる

---

7. まとめ

• 無音検出 (VAD) + ストリーミング制御
  無音時に API 接続を停止し、必要なときだけ ストリーミングを実行することで課金を大幅に削減可能
• ウェイクワード & コマンド検出
  「コンピュータ」と呼びかけ → コマンド「ログイン」「発進」などを認識 → システム処理
• 拡張可能
  バッチ処理・カスタム辞書・ノイズキャンセルなど、多彩な機能を組み合わせて高度な音声アシスタントへ発展

今回のサンプルを参考に、音声認識を効率的に活用しながら、リアルタイムなコマンド操作とコスト削減を両立するアプリケーションを構築してみてください。

---

以上