Align Google STT plugin with official documentation

[mrkowalski]

This PR aligns Livekit's Google plugin with STT API documentation: https://cloud.google.com/python/docs/reference/speech/latest/google.cloud.speech_v2.types.StreamingRecognizeResponse

Especially this:

results: This repeated list contains zero or more results that correspond to consecutive portions of the audio currently being processed. It contains zero or one is_final=true result (the newly settled portion), followed by zero or more is_final=false results (the interim results).

It effectively ignores min_confidence_threshold for final results, which closes #3495.

I have smoke-tested it with the following models:

• latest_long
• chirp_2
• chirp_3

Corresponding unit-tests are in tests/test_plugin_google_stt.py

[davidzhao]

awesome work @mrkowalski!

[benorama]

Do you have a running example of STT using Google chirp 3 ?

When trying the following basic agent with ADK v1.16.0, I'm getting a timeout 500 error or 400 "Chirp 3 does not currently support word timestamps".
The same code works fine by simply replacing STT to use Deepgram.

from dotenv import load_dotenv

# ruff: noqa: E402
load_dotenv()

from livekit.agents import (
    Agent,
    AgentSession,
    JobContext,
    RoomInputOptions,
    WorkerOptions,
    cli,
)
from livekit.plugins import google, noise_cancellation, silero
from livekit.plugins.turn_detector.english import EnglishModel
from livekit.plugins.turn_detector.multilingual import MultilingualModel

LANGUAGE = "en-US"
LANGUAGE_CODE = "en"
ASSISTANT_PROMPT = "You are a helpful voice AI assistant."

class Assistant(Agent):
    def __init__(self) -> None:
        super().__init__(instructions=ASSISTANT_PROMPT)


async def entrypoint(ctx: JobContext) -> None:
    session: AgentSession = AgentSession(
        vad=silero.VAD.load(),
        # stt=f"deepgram/nova-3:{LANGUAGE_CODE}",
        stt=google.STT(languages=[LANGUAGE], location="us", model="chirp_3", min_confidence_threshold=0.0, spoken_punctuation=False),
        llm="google/gemini-2.5-flash",
        tts=google.TTS(language=LANGUAGE, voice_name=f"{LANGUAGE}-chirp3-HD-Achernar"),
        # preemptive_generation=True,
        turn_detection=EnglishModel() if LANGUAGE_CODE == "en" else MultilingualModel(),
    )

    await session.start(
        agent=Assistant(),
        room=ctx.room,
        room_input_options=RoomInputOptions(
            noise_cancellation=noise_cancellation.BVC(),
        ),
    )

    # Join the room and connect to the user
    await session.generate_reply(
        instructions="Greet the user and offer your assistance."
    )


if __name__ == "__main__":
    cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint))

Thanks.

[heyitskim191296]

@benorama I set enable_word_time_offsets to False and the above error disappears.

Here is a full example:

stt=google.STT(model="chirp_3",
location='us',
languages=["fil-PH"],
min_confidence_threshold=0.0,
enable_word_time_offsets=False
)

[benorama]

Thanks @heyitskim191296 .
It works now, even if I'm still facing gRPC connection timeout: locally, the agent starts working after an initial 60s timeout. But that's probably another issue.