Gemini 3.5 Transcribe API 入門：録音とリアルタイム文字起こしを正しく実装する

import json
import sys
from pathlib import Path
from google import genai


def recording_config(purpose):
    choices = {
        "plain": {"language_codes": ["ja-JP"]},
        "terms": {
            "language_codes": ["ja-JP"],
            "custom_vocabulary": ["BigQuery", "受注管理"],
        },
        "trace": {
            "language_codes": ["ja-JP"],
            "mode": {
                "type": "verbatim",
                "diarization_mode": "speaker",
                "timestamp_granularities": ["word"],
            },
        },
        "readable": {"language_codes": ["ja-JP"], "mode": "smart"},
    }
    if purpose not in choices:
        raise ValueError("plain / terms / trace / readable を指定してください")
    return {"transcription_config": choices[purpose]}


def main():
    if len(sys.argv) != 3:
        raise SystemExit("使い方: python recorded.py sample.mp3 trace")
    source = Path(sys.argv[1])
    if not source.is_file():
        raise FileNotFoundError(source)
    config = recording_config(sys.argv[2])
    client = genai.Client()
    uploaded = client.files.upload(file=str(source))
    result = client.interactions.create(
        model="gemini-3.5-transcribe",
        input=[{
            "type": "audio",
            "uri": uploaded.uri,
            "mime_type": uploaded.mime_type,
        }],
        generation_config=config,
    )
    # 注釈と状態も保持し、全文だけで成功判定しない。
    saved = result.model_dump(mode="json")
    Path("recorded-result.json").write_text(
        json.dumps(saved, ensure_ascii=False, indent=2), encoding="utf-8"
    )
    if saved.get("status") != "completed":
        raise RuntimeError(f"完了していません: {saved.get('status')}")
    text = result.output_text or ""
    Path("recorded-text.txt").write_text(text, encoding="utf-8")
    print(text if text else "完了しましたが全文は空です。元音声を確認してください。")


if __name__ == "__main__":
    main()

import json
import re
import sys
from decimal import Decimal
from pathlib import Path


def seconds(value):
    if value is None:
        return None
    if not isinstance(value, str) or not re.fullmatch(r"\d+(?:\.\d+)?s", value):
        raise ValueError(f"解釈できない時刻: {value!r}")
    return Decimal(value[:-1])


def word_rows(document):
    rows = []
    for step in document.get("steps") or []:
        for part in step.get("content") or []:
            for note in part.get("annotations") or []:
                if note.get("type") != "word_info":
                    continue
                start = seconds(note.get("start_offset"))
                end = seconds(note.get("end_offset"))
                if start is not None and end is not None and end < start:
                    raise ValueError("終了時刻が開始時刻より前です")
                rows.append({
                    "text": note.get("text"),
                    "speaker": note.get("speaker"),
                    "start_seconds": str(start) if start is not None else None,
                    "end_seconds": str(end) if end is not None else None,
                })
    return rows


if __name__ == "__main__":
    data = json.loads(Path(sys.argv[1]).read_text(encoding="utf-8"))
    print(json.dumps(word_rows(data), ensure_ascii=False, indent=2))

import asyncio
import json
import sys
import wave
from pathlib import Path
from google import genai
from google.genai import types


def read_pcm(path):
    with wave.open(str(path), "rb") as wav:
        if (wav.getnchannels(), wav.getsampwidth(), wav.getframerate(),
                wav.getcomptype()) != (1, 2, 16000, "NONE"):
            raise ValueError("16kHz / mono / 16-bit / 非圧縮 WAV が必要です")
        count = wav.getnframes()
        if not 0 < count <= 16000 * 540:
            raise ValueError("音声は0秒より長く9分以内にしてください")
        pcm = wav.readframes(count)
        if len(pcm) != count * 2:
            raise ValueError("WAV の音声データが不足しています")
        return pcm


def accept_caption(state, content):
    if content is None:
        return
    interim = getattr(content, "interim_input_transcription", None)
    final = getattr(content, "input_transcription", None)
    if interim is not None and interim.text is not None:
        state["preview"] = interim.text
    if final is not None and final.text:
        state["segments"].append({
            "local_sequence": len(state["segments"]) + 1,
            "text": final.text,
        })
        state["preview"] = ""


async def receive_captions(session, state):
    while True:
        received = False
        async for response in session.receive():
            received = True
            accept_caption(state, getattr(response, "server_content", None))
            print(json.dumps(state, ensure_ascii=False), flush=True)
        if not received:
            return "receive_ended"


async def exchange(pcm, state):
    client = genai.Client()
    config = types.LiveConnectConfig(
        response_modalities=["TEXT"],
        input_audio_transcription=types.AudioTranscriptionConfig(
            language_codes=["ja-JP"], mode="VERBATIM",
        ),
    )
    async with client.aio.live.connect(
        model="gemini-3.5-transcribe-live", config=config,
    ) as session:
        receiver = asyncio.create_task(receive_captions(session, state))
        try:
            for offset in range(0, len(pcm), 3200):
                if receiver.done():
                    receiver.result()  # 受信側の例外も呼び出し元へ渡す。
                    raise RuntimeError("送信終了前に受信が終わりました")
                chunk = pcm[offset:offset + 3200]
                await asyncio.wait_for(session.send_realtime_input(
                    audio=types.Blob(data=chunk, mime_type="audio/pcm;rate=16000")
                ), timeout=10)
                await asyncio.sleep(len(chunk) / 32000)
            await asyncio.wait_for(
                session.send_realtime_input(audio_stream_end=True), timeout=10
            )
            state["eof_sent"] = True
            try:
                state["stop_reason"] = await asyncio.wait_for(
                    asyncio.shield(receiver), timeout=8
                )
            except asyncio.TimeoutError:
                state["stop_reason"] = "drain_timeout"
        finally:
            receiver.cancel()
            await asyncio.gather(receiver, return_exceptions=True)


async def main():
    state = {"preview": "", "segments": [], "eof_sent": False,
             "stop_reason": "not_started", "all_final_received": None}
    try:
        pcm = read_pcm(Path(sys.argv[1]))
        await asyncio.wait_for(exchange(pcm, state), timeout=590)
    except Exception as exc:
        state["stop_reason"] = "error"
        state["error_type"] = type(exc).__name__
        raise
    finally:
        Path("live-result.json").write_text(
            json.dumps(state, ensure_ascii=False, indent=2), encoding="utf-8"
        )


if __name__ == "__main__":
    if len(sys.argv) != 2:
        raise SystemExit("使い方: python live_wav.py sample-16k-mono.wav")
    asyncio.run(main())

録音1分 = (25 × 60 × 2 + 175 × 12) / 1,000,000
         = 0.0051米ドル
Live 1分 = (25 × 60 × 3.50 + 175 × 21) / 1,000,000
         = 0.008925米ドル
100時間 = 6,000分 → 録音30.60米ドル / Live 53.55米ドル