Pythonで音声認識：ファイル・ストリーミング・マイク

Pythonで音声をテキストに変換するには、音声認識モデルに音声を送り、テキストを受け取ります。考えるべきなのはどの種類のモデルを使うかです。録音ファイルとライブの会話は別の仕事だからです。ファイルはクリップの最後まで待てます。電話の相手は待てません。

このガイドでは3つの方法を、それぞれそのまま動くコード付きで紹介します。録音ファイルの文字起こし、自動ターン検出付きの音声ストリーミング、そしてマイク入力のリアルタイム文字起こしです。ストリーミングの例では、音声エージェント向けに私たちが開発しているCartesia Ink 2を使います。無料のローカルモデルのほうが合う場面では、そう書いてそのコードも載せます。

方法を選ぶ

多くのチュートリアルはSpeechRecognitionのrecognize_google()で終わります。このパッケージは十数種類のエンジンをまとめた便利なラッパーですが、デフォルトのGoogleバックエンドはGoogleがいつでも無効化できる共有キーを使い、フレーズが終わるまで何も送りません。デモには十分です。電話の応対には使えません。

前提条件

• Python 3.9以降
• Cartesiaの例では、Cartesia Playgroundで作成したAPIキーとSDK：

python -m pip install "cartesia[websockets]"
export CARTESIA_API_KEY="your-api-key"

ストリーミングの例にはwebsocketsエクストラが必要です。キーは環境変数かシークレットマネージャーに保存し、コミットするコードには書かないでください。

1. 録音ファイルを文字起こしする

録音済みの音声なら、ファイル全体を1回のリクエストで送ります。CartesiaのバッチエンドポイントはWAVやMP3などのコンテナを受け付け、単語ごとのタイムスタンプも返せます。次のコードをtranscribe_file.pyとして保存します。

import os
import sys

from cartesia import Cartesia

client = Cartesia(api_key=os.environ["CARTESIA_API_KEY"])

with open(sys.argv[1], "rb") as audio_file:
    result = client.stt.transcribe(
        file=audio_file,
        model="ink-whisper",
        language="en",
        timestamp_granularities=["word"],
    )

print(result.text)
for word in result.words or []:
    print(f"{word.start:6.2f}s  {word.word}")

python transcribe_file.py meeting.mp3で実行します。バッチの文字起こしには約100言語に対応するink-whisperを使います。音声の言語コードを指定してください。非同期版はドキュメントのバッチSTTの例にあります。

ローカルで動かす方法

音声をマシンの外に出せない場合や、大量の音声とGPUがある場合は、Whisperを自分で動かしましょう。faster-whisperはOpenAIのWhisperをCTranslate2上で再実装したもので、ノートPCのCPUでも動きます。

python -m pip install faster-whisper

from faster_whisper import WhisperModel

model = WhisperModel("base.en", device="cpu", compute_type="int8")
segments, info = model.transcribe("jfk.flac", beam_size=5)

print(f"Detected language: {info.language} ({info.language_probability:.2f})")
for segment in segments:
    print(f"[{segment.start:.2f}s -> {segment.end:.2f}s] {segment.text}")

Whisperリポジトリにある11秒のJFKのサンプルでは、base.enモデルが4コアのサーバーCPUで約2.5秒かけて、引用を一語一句正しく書き起こしました（初回のみ140MBのモデルをダウンロード）。大きなモデルほど精度が上がり、遅くなります。手に入らないのはストリーミングです。Whisperは完成したセグメントを処理するので、会話には向きません。

2. 音声をストリーミングしてターンを受け取る

音声エージェントには、バッチモデルでは得られないものが2つ必要です。相手が話している最中のテキストと、話し終えたかどうかの判断です。Ink 2はその両方を1本のWebSocketで扱います。そのターン検出は無音だけでなく話の内容も見ているので、考えるために文の途中で間を置いた人が、回線が静かになっただけで遮られることはありません。

このスクリプトはWAVファイルをライブ音声のように送ります。stream_file.pyとして保存します。

import os
import sys
import time
import wave

from cartesia import Cartesia

with wave.open(sys.argv[1], "rb") as wav:
    if wav.getnchannels() != 1 or wav.getsampwidth() != 2:
        raise SystemExit("Convert the file to 16-bit mono WAV first.")
    sample_rate = wav.getframerate()
    chunks = []
    while data := wav.readframes(sample_rate // 10):  # 100 ms per chunk
        chunks.append(data)

client = Cartesia(api_key=os.environ["CARTESIA_API_KEY"])
transcript = ""

with client.stt.auto_finalize.websocket(
    model="ink-2",
    encoding="pcm_s16le",
    sample_rate=sample_rate,
) as ws:
    for chunk in chunks:
        ws.send_raw(chunk)
        time.sleep(0.1)  # send at the speed of speech
    ws.send({"type": "close"})  # flush buffered audio; the server then closes

    turn_text = ""
    for event in ws:
        if event.type == "turn.update":
            turn_text = event.transcript or turn_text  # the whole turn so far
            print("...", turn_text)
        elif event.type == "turn.end":
            text = event.transcript or turn_text  # fall back if turn.end is empty
            print("turn:", text)
            transcript += text  # join verbatim, no strip()
            turn_text = ""
        elif event.type == "error":
            raise SystemExit(event.message)

print(transcript)

ファイルが16ビット・モノラルのWAVでなければ、先に変換します。

ffmpeg -i input.mp3 -ac 1 -ar 16000 -c:a pcm_s16le input.wav

このスクリプトには、間違えやすい点が3つあります。

• ファイルではなく生のサンプルを送る。 リアルタイム用エンドポイントはヘッダーのない音声バイトを受け取ります。そのためスクリプトはファイルを送らずに、WAVからフレームを読み出しています。encodingとsample_rateは接続時に指定し、実際のバイトと一致させる必要があります。音声入力ガイドにあるとおり、一致しなくてもサーバーがエラーを返さないことがあり、サンプルレートの間違いは例外ではなく、おかしな文字起こしとして表れます。
• リアルタイムのペースで送る。 100ミリ秒のチャンクごとのsleep(0.1)は飾りではありません。リアルタイム用エンドポイントは1秒あたりおよそ1秒分の音声を想定していて、ファイル全体を一度にソケットへ流し込むとサーバー側のバッファがあふれることがあります。ライブで処理する必要のないファイルなら、手順1のバッチ呼び出しを使ってください。
• 文字起こしは累積される。 各turn.updateには、新しい断片ではなく、そのターンのそこまでの全文が入っています。更新をつなげると単語が重複します。テキストはターンごとに1回、turn.endから取り出し、届いたとおりにつなげてください。このスクリプトは最新のturn.updateも保持していて、turn.endの文字起こしが空で届いた場合はそれを使います。私たち自身のクライアントでも備えているケースです。

短くするため、このスクリプトはファイルをすべて送ってからイベントを読むので、出力は最後にまとめて表示されます。下のマイクの例では、送信と受信を同時に行います。

Ink 2は英語、フランス語、ヒンディー語、日本語、スペイン語を文字起こしし、どの言語かを自動で判別します。そのほかの言語でストリーミングが必要なら、音声認識モデルの比較を参照してください。

3. マイク入力をリアルタイムで文字起こしする

エンドポイントは同じで、入力が本物になります。このバージョンではsounddeviceで16kHzのモノラル音声を取り込み、非同期クライアントで送信と受信を同時に行います。

python -m pip install sounddevice

Linuxでは、PortAudioも必要になる場合があります（sudo apt install libportaudio2）。次のコードをmicrophone.pyとして保存します。

import asyncio
import os

import sounddevice as sd
from cartesia import AsyncCartesia

SAMPLE_RATE = 16000
BLOCK = SAMPLE_RATE // 10  # 100 ms of audio per message


async def main():
    loop = asyncio.get_running_loop()
    audio = asyncio.Queue(maxsize=50)  # about 5 s of audio

    def enqueue(block):
        if audio.full():  # the network fell behind: drop the oldest block
            audio.get_nowait()
        audio.put_nowait(block)

    def on_audio(indata, frames, time_info, status):
        loop.call_soon_threadsafe(enqueue, bytes(indata))

    client = AsyncCartesia(api_key=os.environ["CARTESIA_API_KEY"])
    async with client.stt.auto_finalize.websocket(
        model="ink-2",
        encoding="pcm_s16le",
        sample_rate=SAMPLE_RATE,
    ) as ws:

        async def send():
            with sd.RawInputStream(
                samplerate=SAMPLE_RATE,
                blocksize=BLOCK,
                channels=1,
                dtype="int16",
                callback=on_audio,
            ):
                while True:
                    await ws.send_raw(await audio.get())

        async def receive():
            turn_text = ""
            async for event in ws:
                if event.type == "turn.start":
                    print("(listening)")
                elif event.type == "turn.update":
                    turn_text = event.transcript or turn_text
                elif event.type == "turn.end":
                    print("You said:", event.transcript or turn_text)
                    turn_text = ""
                elif event.type == "error":
                    raise RuntimeError(event.message)

        await asyncio.gather(send(), receive())


asyncio.run(main())

python microphone.pyを実行して何か話し、間を置くと、そのターンの文字起こしが表示されます。止めるにはCtrl+Cを押します。

音声コールバックは別スレッドで動くので、ソケットを直接触らず、call_soon_threadsafeで各ブロックをイベントループに渡しています。黙っている間もマイクは送信を続けますが、これには意味があります。サーバーは音声が届かないことを無音ではなくネットワークの遅延として扱うので、ミュートや一時停止中の入力は、何も送らないのではなくゼロを送る必要があります。キューの上限は約5秒分の音声です。接続が遅れたら、古いブロックをメモリにため込まずに捨てます。1分前に話した内容の文字起こしは、ライブの会話では役に立たないからです。

音声エージェントでは、turn.startでボットの発話を止め、turn.endでテキストをLLMに渡します。返答をもっと早く始めたいならturn.eager_endも処理しましょう。届いた時点で生成を始め、その後にturn.resumeが来たら下書きを破棄します。発信者がよく間を置く（あるいはほとんど置かない）場合は、ターン検出のドキュメントでしきい値を確認してください。

よくある問題

料金

ローカルモデルにかかるのは計算資源だけです。CartesiaではInk 2は音声1秒あたり3クレジットを使い、Scaleプランなら音声1時間で0.39ドルです。無料プランのクレジットで、Ink 2を約1時間50分使えます。最新の数字は料金ページで確認してください。

パイプラインを自分で組みたくない場合は、Cartesia Managed AgentsがInk、LLM、Sonicの音声合成を電話番号につないでくれます。いくつかの録音を文字起こしするだけなら、ノートPCのfaster-whisperで十分でしょう。

次のステップ

1. Inkのページで、コードを書かずにブラウザで自分の声を使ってInk 2を試す。
2. 文字起こしをLLMに渡し、Pythonの音声合成で返答を読み上げる。
3. フレームワークで組み合わせる：CartesiaでPipecatの音声エージェントを作る。

関連ドキュメント

• リアルタイム音声認識のクイックスタート
• STTエンドポイントの比較：自動、手動、バッチ
• キーターム・プロンプティング：モデルに正しく書き起こしてほしい固有名詞や専門用語に
• 音声区間検出（VAD）と、ターン検出がそれに代わる理由