I want to add Wake Word detection feature in my Voice Agent created using Gemini Realtime Model and Livekit Services, I just want to add Wake Word detection no matter its built in or I have to use External one

Thanks for reaching out, Already. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.

Yes detailed step actually what I want is to maintain its realtime functinality but make sure it to respond only when I speak the Wake Word suppose Hey Ai first then my query and else it should not respond to any query not having the wake word

I'm glad you're working on maintaining the real-time functionality of your Voice Agent while making sure it responds only when it hears the wake word (like "Hey AI"). Let me walk you through the steps gently and clearly: 1. Define the Wake Word: First, choose a specific wake word (for example, "Hey AI") that your agent will listen for before it starts processing any queries. 2. Create a Custom Agent: You'll want to build a custom agent class that overrides the

stt_node

method. This method helps filter speech events so your agent only reacts when it detects the wake word, passing relevant commands to the Gemini Realtime Model. 3. Use a Voice Activity Detector (VAD): Adding a VAD is a great idea—it monitors when the user starts and stops speaking, making the interaction smoother. 4. Filter Speech Events: Inside the

stt_node

method, check the transcript for the wake word. When it’s detected, extract the command that follows and send it to the Gemini model. If the wake word isn’t there, simply ignore that input. 5. Integrate with AgentSession: Use the

AgentSession

class to manage your real-time interactions with the Gemini Realtime Model, and configure it to use your custom agent. 6. Test and Optimize: Finally, test everything to make sure your agent only responds when the wake word is spoken and keeps working in real time. To make this easier, here’s a detailed code example that puts all these pieces together:

import re
from typing import AsyncIterable
from livekit import rtc
from livekit.agents import JobContext, WorkerOptions, cli
from livekit.agents.voice import Agent, AgentSession
from livekit.plugins import google

# Define your wake word
WAKE_WORD = "hey ai"

class WakeWordAgent(Agent):
    def __init__(self):
        super().__init__()
        self.wake_word_detected = False

    def stt_node(self, audio: AsyncIterable[rtc.AudioFrame], model_settings: dict = None) -> AsyncIterable[rtc.SpeechEvent] | None:
        parent_stream = super().stt_node(audio, model_settings)
        if parent_stream is None:
            return None

        async def filtered_stream():
            async for event in parent_stream:
                if hasattr(event, 'type') and str(event.type) == "SpeechEventType.FINAL_TRANSCRIPT" and event.alternatives:
                    transcript = event.alternatives[0].text.lower()
                    cleaned_transcript = re.sub(r'[^\w\s]', '', transcript).strip()

                    if not self.wake_word_detected:
                        if WAKE_WORD in cleaned_transcript:
                            self.wake_word_detected = True
                            command = cleaned_transcript.split(WAKE_WORD, 1)[-1].strip()
                            if command:
                                event.alternatives[0].text = command
                                yield event
                        else:
                            continue
                    else:
                        yield event
                        if str(event.type) == "SpeechEventType.END_OF_SPEECH":
                            self.wake_word_detected = False

        return filtered_stream()

async def entrypoint(ctx: JobContext):
    gemini_llm = google.beta.realtime.RealtimeModel(
        model="gemini-2.0-flash-exp",
        voice="Puck",
        temperature=0.8,
        instructions="You are an assistant activated by a wake word. Be helpful and concise.",
        input_audio_transcription=None,
    )

    session = AgentSession(llm=gemini_llm)
    await session.start(agent=WakeWordAgent(), room=ctx.room)

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

This approach ensures your agent listens carefully and only responds when it hears the wake word, all while keeping the interaction smooth and in real time. If anything feels unclear or you want to dive deeper into any part, just let me know—I’m here to help!

actually the stt and tts all are of gemini and I have taken the code from livekit docs Voice agent quick start

I'm glad you're exploring how to implement wake word detection with the Gemini Realtime Model while keeping its STT and TTS features intact. To achieve this smoothly, you'll want to bring in an external STT service for real-time wake word detection. This is because Gemini's built-in STT provides transcripts only after the user finishes speaking, which doesn't quite fit the needs of wake word detection. Here’s a friendly step-by-step on how you can set this up: 1. Use an External STT: Consider integrating a real-time STT service like Deepgram or AssemblyAI to catch the wake word as the user is speaking. 2. Implement Wake Word Filtering: You can use a wake word handler or override the

stt_node

method to filter transcripts and spot the wake word before sending the query on to Gemini. 3. Disable Gemini's Turn Detection: It helps to configure the Gemini Realtime Model to turn off its automatic activity detection, so it doesn’t interfere with your custom wake word logic. 4. Integrate with AgentSession: Use the

AgentSession

or

VoicePipelineAgent

to manage the interaction, making sure the agent processes only queries that include the wake word. If you'd like, I can provide a detailed code example to guide you through this setup, or we can focus on any specific part you're finding tricky. Just let me know—I’m here to help! Sources: realtime/gemini_realtime_api.py | LiveKit turn detector plugin | LiveKit Docs | README.md | livekit-agents/livekit/agents/voice/agent_session.py