---

name: livekit-turn-detection description: Configure turn detection, VAD, and interruption handling for natural voice AI conversations argument-hint: "<turn detection requirements>" allowed-tools: Read, Write, Bash(pip install, npm install), Glob, Grep, WebSearch

LiveKit Turn Detection

Natural conversation flow for voice AI: $ARGUMENTS

Expert Knowledge

You are a LiveKit turn detection specialist with expertise in:

• Voice Activity Detection (VAD)
• End-of-turn detection strategies
• Interruption handling
• Endpointing configuration
• Conversational flow optimization

Turn Detection Modes

| Mode | Description | Best For | |------|-------------|----------| | server_vad | VAD-based silence detection | General use | | turn_detector | ML-based context detection | Natural conversation | | stt | STT endpoint markers | Specific languages | | Manual | Application-controlled | Custom logic |

VAD Configuration

Silero VAD (Default)

from livekit.plugins.silero import VAD

vad = VAD.load(
    # Minimum duration to consider as speech
    min_speech_duration=0.1,  # seconds

    # Silence duration before end of speech
    min_silence_duration=0.3,  # seconds

    # Padding around speech detection
    padding_duration=0.1,  # seconds

    # Sensitivity threshold (0-1)
    # Higher = more sensitive to speech
    activation_threshold=0.5,

    # Sample rate
    sample_rate=16000,
)

await session.start(
    agent=MyAgent(),
    vad=vad,
    turn_detection="server_vad",
    ...
)

VAD Events

@RtcSession.factory
async def create_session(session: AgentSession):
    @session.on("vad_speech_started")
    def on_speech_start():
        print("Speech detected")

    @session.on("vad_speech_ended")
    def on_speech_end():
        print("Silence detected")

    await session.start(...)

Turn Detector (ML-Based)

Uses a transformer model for better end-of-turn detection:

from livekit.plugins.turn_detector import EOUModel

# Load the model
turn_detector = EOUModel.load()

await session.start(
    agent=MyAgent(),
    stt="deepgram/nova-3",
    llm="openai/gpt-4o-mini",
    tts="cartesia/sonic",
    vad=silero.VAD.load(),
    turn_detection=turn_detector,  # ML-based
)

Benefits

• Considers conversational context
• Better handles thinking pauses
• Reduces false positives from filler words
• More natural conversation flow

STT Endpointing

Use STT's built-in endpointing:

await session.start(
    agent=MyAgent(),
    stt="deepgram/nova-3",  # Has built-in endpointing
    llm="openai/gpt-4o-mini",
    tts="cartesia/sonic",
    vad=silero.VAD.load(),  # Still needed for interruptions
    turn_detection="stt",  # Use STT's endpointing
)

Endpointing Delay

Control how long to wait after speech ends:

await session.start(
    agent=MyAgent(),
    stt="deepgram/nova-3",
    llm="openai/gpt-4o-mini",
    tts="cartesia/sonic",
    vad=silero.VAD.load(),
    turn_detection="server_vad",

    # Wait 500ms of silence before responding
    min_endpointing_delay=0.5,

    # Maximum wait time
    max_endpointing_delay=2.0,
)

Interruption Handling

Allow Interruptions (Default)

await session.start(
    agent=MyAgent(),
    ...
    allow_interruptions=True,  # User can interrupt agent
    interrupt_min_words=3,      # Min words before allowing interrupt
)

Handle Interruptions

class InterruptibleAgent(Agent):
    async def on_interrupted(self):
        """Called when user interrupts agent speech."""
        print("User interrupted")
        # Could log, adjust behavior, etc.

    async def on_user_speech_started(self):
        """Called when user starts speaking."""
        # Agent will automatically stop speaking
        pass

False Interruption Recovery

await session.start(
    agent=MyAgent(),
    ...
    # Resume speech after false interrupt
    resume_on_false_interrupt=True,

    # How long to wait for actual user speech
    false_interruption_timeout=0.5,
)

Disable Interruptions

await session.start(
    agent=MyAgent(),
    ...
    allow_interruptions=False,  # Agent finishes before listening
)

Manual Turn Control

For non-conversational applications:

class ManualTurnAgent(Agent):
    def __init__(self):
        super().__init__(
            instructions="You process audio commands."
        )

    async def start_listening(self):
        """Manually trigger listening."""
        await self.session.start_listening()

    async def stop_listening(self):
        """Manually stop listening."""
        await self.session.stop_listening()

    async def process_audio(self):
        """Process captured audio."""
        transcript = await self.session.get_transcript()
        # Process transcript

Session Events

@RtcSession.factory
async def create_session(session: AgentSession):
    @session.on("user_speech_started")
    def on_user_start():
        print("User started speaking")

    @session.on("user_speech_ended")
    def on_user_end(transcript: str):
        print(f"User finished: {transcript}")

    @session.on("agent_speech_started")
    def on_agent_start():
        print("Agent started speaking")

    @session.on("agent_speech_interrupted")
    def on_agent_interrupted():
        print("Agent was interrupted")

    @session.on("agent_speech_ended")
    def on_agent_end():
        print("Agent finished speaking")

    @session.on("turn_completed")
    def on_turn_complete():
        print("Turn completed")

    await session.start(...)

Optimizing Turn Detection

Fast Response

await session.start(
    agent=FastAgent(),
    stt="deepgram/nova-3",
    llm="openai/gpt-4o-mini",  # Fast model
    tts="cartesia/sonic",
    vad=silero.VAD.load(
        min_silence_duration=0.2,  # Quick detection
    ),
    turn_detection="server_vad",
    min_endpointing_delay=0.3,  # Short delay
)

Natural Conversation

from livekit.plugins.turn_detector import EOUModel

await session.start(
    agent=ConversationalAgent(),
    stt="deepgram/nova-3",
    llm="openai/gpt-4o",  # Better context understanding
    tts="cartesia/sonic",
    vad=silero.VAD.load(
        min_silence_duration=0.4,  # Allow thinking pauses
    ),
    turn_detection=EOUModel.load(),  # ML-based
    min_endpointing_delay=0.5,
)

Telephony Optimized

await session.start(
    agent=PhoneAgent(),
    stt="deepgram/nova-3-phonecall",
    llm="openai/gpt-4o-mini",
    tts="cartesia/sonic",
    vad=silero.VAD.load(
        min_silence_duration=0.5,  # Account for phone delay
        activation_threshold=0.6,   # Less sensitive for noise
    ),
    turn_detection="server_vad",
    min_endpointing_delay=0.6,
    noise_cancellation=noise_cancellation.NC(),
)

Debugging Turn Detection

import logging

logging.basicConfig(level=logging.DEBUG)

@RtcSession.factory
async def create_session(session: AgentSession):
    @session.on("vad_metrics")
    def on_vad_metrics(metrics):
        print(f"VAD probability: {metrics.speech_probability:.2f}")
        print(f"Speech duration: {metrics.speech_duration:.2f}s")

    @session.on("turn_detection_metrics")
    def on_turn_metrics(metrics):
        print(f"Confidence: {metrics.confidence:.2f}")
        print(f"Is end of turn: {metrics.is_end_of_turn}")

    await session.start(
        agent=MyAgent(),
        ...
    )

Complete Example

import logging
from dotenv import load_dotenv
from livekit.agents import Agent, AgentSession, RtcSession, AgentServer
from livekit.plugins import silero, noise_cancellation
from livekit.plugins.turn_detector import EOUModel

load_dotenv()
logging.basicConfig(level=logging.INFO)

class ConversationalAgent(Agent):
    def __init__(self):
        super().__init__(
            instructions="""You are a natural conversational assistant.
            Listen carefully and respond thoughtfully.
            It's okay to pause while thinking."""
        )
        self.interrupted_count = 0

    async def on_enter(self):
        await self.session.generate_reply(
            "Hi! I'm ready to chat. What's on your mind?"
        )

    async def on_interrupted(self):
        self.interrupted_count += 1
        logging.info(f"Interrupted {self.interrupted_count} times")

    async def on_user_turn(self, message: str):
        logging.info(f"User said: {message}")
        await super().on_user_turn(message)

@RtcSession.factory
async def create_session(session: AgentSession):
    # Load VAD
    vad = silero.VAD.load(
        min_speech_duration=0.15,
        min_silence_duration=0.4,
        activation_threshold=0.5,
    )

    # Load turn detector
    turn_detector = EOUModel.load()

    # Log events
    @session.on("user_speech_started")
    def on_user_start():
        logging.info("User speaking...")

    @session.on("user_speech_ended")
    def on_user_end(transcript: str):
        logging.info(f"User: {transcript}")

    @session.on("agent_speech_started")
    def on_agent_start():
        logging.info("Agent responding...")

    await session.start(
        agent=ConversationalAgent(),
        stt="deepgram/nova-3",
        llm="openai/gpt-4o-mini",
        tts="cartesia/sonic",
        vad=vad,
        turn_detection=turn_detector,
        noise_cancellation=noise_cancellation.BVC(),

        # Endpointing
        min_endpointing_delay=0.5,
        max_endpointing_delay=2.0,

        # Interruptions
        allow_interruptions=True,
        interrupt_min_words=3,
        resume_on_false_interrupt=True,
    )

if __name__ == "__main__":
    AgentServer(create_session).run()

Best Practices

1. Use ML turn detector: More natural than VAD alone
2. Keep VAD for interruptions: Always include VAD
3. Tune for use case: Fast vs natural tradeoff
4. Handle false interrupts: Resume speech when appropriate
5. Log metrics: Debug conversation flow

Deliverables

For: $ARGUMENTS

Provide:

1. Turn detection mode selection
2. VAD configuration
3. Endpointing settings
4. Interruption handling
5. Event handlers
6. Optimization recommendations