Gemini 3.5 Transcribe API 指南：预录转写、实时字幕与验证方法

# recorded.py
import argparse
from google import genai

PROFILES = {
    "plain": {},
    "readable": {
        "mode": "smart",
        "language_codes": ["cmn-Hans-CN"],
        "custom_vocabulary": ["Gemini", "BigQuery", "LaoZhang AI"],
    },
    "annotated": {
        "mode": {
            "type": "verbatim",
            "diarization_mode": "speaker",
            "timestamp_granularities": ["word"],
        },
    },
}


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("audio_path")
    parser.add_argument("profile", choices=PROFILES, default="plain", nargs="?")
    args = parser.parse_args()
    client = genai.Client()
    audio = client.files.upload(file=args.audio_path)
    options = {}
    if PROFILES[args.profile]:
        options["generation_config"] = {
            "transcription_config": PROFILES[args.profile]
        }
    result = client.interactions.create(
        model="gemini-3.5-transcribe",
        input=[{"type": "audio", "uri": audio.uri,
                "mime_type": audio.mime_type}],
        **options,
    )
    # 保留完整响应，随后可解析 steps 中的词级标注。
    with open("interaction.json", "w", encoding="utf-8") as output:
        output.write(result.model_dump_json(exclude_none=True))
    print("status:", result.status)
    print(result.output_text)


if __name__ == "__main__":
    main()

# word_rows.py
import json
import re
import sys
from decimal import Decimal


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(response):
    rows = []
    for step in response.get("steps") or []:
        for content in step.get("content") or []:
            for item in content.get("annotations") or []:
                if item.get("type") != "word_info":
                    continue
                start = seconds(item.get("start_offset"))
                end = seconds(item.get("end_offset"))
                if start is not None and end is not None and end < start:
                    raise ValueError("词级结束时间早于开始时间")
                rows.append({
                    "text": item.get("text"),
                    "speaker": item.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__":
    with open(sys.argv[1], encoding="utf-8") as source:
        print(json.dumps(word_rows(json.load(source)), ensure_ascii=False, indent=2))

# live_pcm.py
import argparse
import asyncio
from pathlib import Path
from google import genai
from google.genai import types


class Captions:
    def __init__(self):
        self.preview = ""
        self.final = []

    def apply(self, interim=None, final=None):
        if interim is not None:
            self.preview = interim
        if final is not None:
            self.final.append(final)
            self.preview = ""


async def receive(session, captions):
    # receive() 可按轮次结束；继续接收后续轮次。
    while True:
        received = False
        async for event in session.receive():
            received = True
            content = event.server_content
            if content is None:
                continue
            interim = content.interim_input_transcription
            final = content.input_transcription
            captions.apply(
                interim.text if interim is not None else None,
                final.text if final is not None else None,
            )
            print({"final": captions.final, "preview": captions.preview})
        if not received:
            return "receive_ended"


async def main(pcm_path):
    pcm = Path(pcm_path).read_bytes()
    if not pcm or len(pcm) % 2:
        raise ValueError("PCM 不能为空，字节数必须是 16 位采样的整数倍")
    if pcm[:4] in (b"RIFF", b"OggS") or pcm[:4] == b"\x1aE\xdf\xa3":
        raise ValueError("输入含容器头；先解码为 raw PCM")
    # 9 分钟是本例的应用上限，给发送和收尾留余量。
    if len(pcm) > 16000 * 2 * 540:
        raise ValueError("本例只处理不超过 9 分钟的输入")
    captions = Captions()
    status = "partial"
    client = genai.Client()
    config = types.LiveConnectConfig(
        response_modalities=["TEXT"],
        input_audio_transcription=types.AudioTranscriptionConfig(
            mode="VERBATIM", language_codes=[],
        ),
    )
    async with client.aio.live.connect(
        model="gemini-3.5-transcribe-live", config=config,
    ) as session:
        receiver = asyncio.create_task(receive(session, captions))
        try:
            loop = asyncio.get_running_loop()
            started = loop.time()
            for offset in range(0, len(pcm), 3200):
                if receiver.done():
                    status = "receiver_stopped: " + str(await receiver)
                    break
                await session.send_realtime_input(audio=types.Blob(
                    data=pcm[offset:offset + 3200],
                    mime_type="audio/pcm;rate=16000",
                ))
                due = started + (offset + 3200) / 32000
                await asyncio.sleep(max(0, due - loop.time()))
            else:
                await session.send_realtime_input(audio_stream_end=True)
                try:
                    status = await asyncio.wait_for(receiver, timeout=5)
                except asyncio.TimeoutError:
                    status = "drain_timeout_partial"
        except Exception as exc:
            status = "error_partial: " + type(exc).__name__
        finally:
            if not receiver.done():
                receiver.cancel()
            try:
                await receiver
            except (asyncio.CancelledError, Exception):
                pass
            # 连接退出后仍可检查已提交片段、未定稿预览和终止原因。
            print({"status": status, "final": captions.final,
                   "pending_preview": captions.preview})


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("pcm_path")
    asyncio.run(main(parser.parse_args().pcm_path))