Gemini TTS を streamGenerateContent で受け取り、先頭のチャンクから鳴らす

個人開発の合間に、記事を音声にして stand.fm に置く作業を続けています。テキストを渡し、返ってきた音声をそのまま1本のファイルにする。素朴な流れです。

ただ、聞き手として自分の配信を再生したとき、押してから音が出るまでの待ち時間がずっと気になっていました。生成が終わるまで待ってから書き出しているのですから、当然といえば当然です。

streamGenerateContent が gemini-3.1-flash-tts-preview でも使えるようになったので、この待ち時間を先頭のチャンクで埋められないかを確かめました。実装してみると、待ち時間そのものより先に、受け取ったものを何と見なすかでつまずきます。

ストリーミングで返ってくるのは音声ファイルではありません

通常の生成と同じく、応答は candidates[0].content.parts の配列で届きます。音声の場合、テキストの代わりに inlineData が入っています。

REST を直接叩いた場合、inlineData.data は base64 文字列です。Python SDK（google-genai）を使う場合はすでに bytes へ復号された状態で届きます。ここは経路によって型が変わるので、受け取り側の想定を先に決めておくと後が楽になります。

もう一つ重要なのが mimeType です。手元では次の形で届きました。

audio/L16;codec=pcm;rate=24000

L16 は16ビットのリニア PCM、rate=24000 はサンプリングレートです。つまり返ってくるのは ヘッダを持たない生の波形データ であって、MP3 や WAV のような「ファイル」ではありません。

レートは固定値として書かず、mimeType から読むようにしています。モデルや設定によってここが変わったとき、ハードコードした数字は再生速度の狂いという形でしか表に出てこないためです。

import re
 
def parse_rate(mime_type: str, default: int = 24000) -> int:
    m = re.search(r"rate=(\d+)", mime_type or "")
    return int(m.group(1)) if m else default
 
# audio/L16;codec=pcm;rate=24000 -> 24000
# audio/pcm;rate=16000           -> 16000
# audio/pcm                      -> 24000（既定値）

受信側は次のようになります。チャンクを1つ受け取るたびに、inlineData があれば PCM として積み上げるだけです。

import wave
from google import genai
from google.genai import types
 
client = genai.Client(api_key="YOUR_API_KEY")
MODEL = "gemini-3.1-flash-tts-preview"
 
def synthesize(text: str, out_path: str) -> None:
    pcm = bytearray()
    rate = None
 
    stream = client.models.generate_content_stream(
        model=MODEL,
        contents=text,
        config=types.GenerateContentConfig(
            response_modalities=["AUDIO"],
            speech_config=types.SpeechConfig(
                voice_config=types.VoiceConfig(
                    prebuilt_voice_config=types.PrebuiltVoiceConfig(voice_name="Kore")
                )
            ),
        ),
    )
 
    for chunk in stream:
        for part in (chunk.candidates[0].content.parts or []):
            blob = getattr(part, "inline_data", None)
            if blob is None or not blob.data:
                continue
            if rate is None:
                rate = parse_rate(blob.mime_type)
            pcm += blob.data          # ここでは連結するだけ
 
    with wave.open(out_path, "wb") as w:
        w.setnchannels(1)
        w.setsampwidth(2)             # 16bit
        w.setframerate(rate or 24000)
        w.writeframes(bytes(pcm))     # ヘッダを書くのは最後の1回だけ

書き出しを最後に1回だけ行っている点が、次の節の話につながります。

チャンクごとにヘッダを付けると、10分の1しか鳴りません

最初に書いた実装では、届いたチャンクを1つずつ WAV として保存し、あとで結合していました。ファイルとしては開けますし、再生も始まります。ところが途中で止まります。

何が起きているのかを、合成した波形で確かめました。24kHz・16ビット・モノラルで0.6秒のチャンクを10個、合計6.0秒ぶんを用意し、2つの方式で書き出した結果です。

出力サイズはほとんど同じです。差は44バイトのヘッダが1個か10個かだけで、波形そのものは1バイトも欠けていません。それでも再生されるのは先頭の0.6秒、つまり全体の10分の1でした。

WAV ヘッダの data チャンクには、それに続くバイト数が書かれています。先頭のヘッダは「このあと28,800バイトの音声が続く」と宣言していますから、プレイヤーはそこで終わりだと判断します。以降の9チャンクは、ヘッダごとデータの外側に置き去りになります。

24kHz・16ビット・モノラルなら、1秒あたりのバイト数は次の計算で出ます。

24,000 サンプル/秒 × 2 バイト × 1 チャンネル = 48,000 バイト/秒

288,000 ÷ 48,000 = 6.0 秒。バイト数から尺が逆算できるのは、生の PCM を扱うときの利点です。チャンクが何個に分かれていても、連結して割り算するだけで正しい長さが出ます。

base64 のまま持ち回ると、この 288,000 バイトが 384,000 バイトになります。ちょうど1.3333倍です。手元の6秒でも96,000バイトの差ですから、10分の読み上げなら無視できない量になります。文字列のまま配列に溜めず、受け取った時点で bytes に戻す設計にしておきたいところです。

16ビットの境界を跨ぐチャンクの扱い

チャンクの切れ目は、サンプルの切れ目と一致しません。1サンプルが2バイトである以上、奇数バイトで届くチャンクは サンプルの途中で切れている ことになります。

バイト列のまま連結する限り、これは問題になりません。前のチャンクの末尾1バイトと、次のチャンクの先頭1バイトが素直につながるからです。

危ないのは、チャンクごとに数値サンプルへ変換してから足していく実装です。奇数の端数を捨てながら進むことになります。

実際に測ってみました。7,201・7,203・7,199・7,205・7,192 バイトという5つのチャンク（合計36,000バイト）を、チャンク単位で16ビット配列に変換した場合の結果です。

失われたのはわずか4バイトです。それなのに、17,998サンプルのうち14,386サンプルが本来と違う値になりました。1バイトずれた時点で上位バイトと下位バイトが入れ替わり、そこから先の波形が別物になるためです。

耳で聞くと、ある瞬間から砂嵐のような音が混じります。私自身、生成品質の問題だと思って何度もプロンプトを書き直しかけたのですが、原因は受け取り側にありました。

端数を持ち越す形にすれば、この問題は消えます。

import array
 
class PcmDecoder:
    """チャンク境界がサンプル境界と一致しなくても壊れない"""
    def __init__(self) -> None:
        self._carry = b""
 
    def feed(self, chunk: bytes) -> array.array:
        buf = self._carry + chunk
        n = len(buf) // 2 * 2      # 2の倍数まで
        self._carry = buf[n:]      # 端数は次回へ
        samples = array.array("h")
        samples.frombytes(buf[:n])
        return samples

先ほどの5チャンクをこの実装に通したところ、18,000サンプルすべてが正しい波形と一致し、持ち越しの残りも0バイトになりました。音量メーターや簡易な無音検出をリアルタイムで動かす場合は、この形が必要になります。

ファイルに落とすだけなら、そもそも数値へ戻す必要はありません。私は用途に応じて、バイト連結だけで済ませる経路と、この PcmDecoder を挟む経路を分けています。

再生を先に始めるためのヘッダ

待ち時間を削るという当初の目的に戻ります。ブラウザやプレイヤーへ流し込むなら、44バイトのヘッダを 音声より先に 送ってしまう手があります。

import struct
 
def wav_header(rate: int, channels: int = 1, width: int = 2,
               data_size: int = 0x7FFFFFFF) -> bytes:
    byte_rate = rate * channels * width
    block_align = channels * width
    return (
        b"RIFF" + struct.pack("<I", 0xFFFFFFFF) + b"WAVE"
        + b"fmt " + struct.pack("<IHHIIHH", 16, 1, channels, rate,
                                byte_rate, block_align, width * 8)
        + b"data" + struct.pack("<I", data_size)
    )

長さが確定していないので、data のサイズには収まりきらない大きな値を入れておきます。手元で書き出して読み返すと、宣言されたフレーム数は 1,073,741,823 になりました。実体は1.0秒ぶんの48,000バイトだけです。

私はこの方式を、聞き手が総再生時間を気にしない用途に限って使っています。先頭から順に読むプレイヤーであれば正しく鳴らせます。読み込んだ48,000バイトはビット単位で元の PCM と一致していました。副作用として、シークバーが正しい長さを示しません。総再生時間が意味を持つ配信物なら、この方式は向きません。

ファイルとして保存する場合は、先に44バイトぶんの空きを書いておき、書き終えてから戻ってサイズ欄を埋めるのが確実です。Python の wave モジュールは close() の時点でこれを行うため、先ほどの synthesize() のように最後にまとめて書けば意識せずに済みます。

用途ごとに整理すると次のようになります。

それでもストリーミングにしない場面

ここまで書いておいて逆のことを言うのですが、私の配信の本編では結局ストリーミングを使っていません。

記事の読み上げは公開前に生成できます。聞き手が再生ボタンを押す時点では、音声はすでにファイルとして存在しているべきものです。ストリーミングが効くのは、テキストがその場で決まる用途に限られます。

ストリーミングは待ち時間の見え方を変えるだけで、生成そのものを速くはしません。事前に作れるものを実行時に作る構成は、待ち時間と呼び出し回数の両方を読者に負担させることになります。どこで生成するかの判断については、実行時呼び出しと配布前の一括処理をどう分けたかに、呼び出し回数の見積もりまで含めて書きました。

まとめ

Gemini の TTS をストリーミングで受け取るときに押さえる点は、次の一行に集約されます。届くのは PCM の断片であり、ヘッダは全体に対して1回だけ書く。

もし今の実装でチャンクごとに音声ファイルを作っているなら、書き出したファイルの尺を測ってみてください。先頭チャンクぶんの長さしかなければ、原因はここにあります。私の場合、生成のパラメータを疑って半日を使ったあとに気づきました。

お読みいただきありがとうございました。手元の実装を見直すきっかけになれば嬉しく思います。