Phase 5: Turn-Taking & Conversation Management · 50 min · Python · LiveKit Agents SDK · Pipecat
The Concept
Endpointing Methods
| Method | How | Latency | Accuracy | Complexity |
|---|
| Fixed silence | Wait N ms of silence | N ms | Low | Simple |
| Adaptive silence | Adjust N based on speech rate | N ms | Medium | Medium |
| Semantic endpointing | LLM checks if thought is complete | ~200ms | High | Complex |
| Turn-detector model | Small model trained to predict end of turn | Tens of ms | High | Medium (off the shelf) |
| Hybrid | Silence + semantic check | N + 200ms | Best | Complex |
Fixed Silence Endpointing
class FixedSilenceEndpointer:
def __init__(self, silence_threshold_ms=500):
self.silence_threshold_ms = silence_threshold_ms
self.last_speech_time = None
def check(self, is_speech, current_time):
"""Check if endpoint is reached. current_time is in seconds (time.monotonic())."""
if is_speech:
self.last_speech_time = current_time
return False # Still speaking
if self.last_speech_time is None:
return False # No speech yet
silence_ms = (current_time - self.last_speech_time) * 1000
return silence_ms >= self.silence_threshold_ms
| Threshold | Experience | Use Case |
|---|
| 200ms | Snappy but interrupts | Fast-paced Q&A |
| 500ms | Natural | General purpose |
| 700ms | Slight pause | Thoughtful conversations |
| 1000ms | Noticeable delay | Deliberate speaking |
A caller says "I want to book a flight to... hmm... Tokyo" with a 600ms pause before "Tokyo". With a 500ms silence threshold, what happens?
With 500ms fixed silence, the agent triggers at 500ms — before "Tokyo" (which comes at 600ms). The agent responds to the incomplete "I want to book a flight to" — a false endpoint. This is the core problem with fixed silence: it can't distinguish a pause (thinking, breathing) from the end of speech. Semantic endpointing fixes this: the LLM checks if "I want to book a flight to" is a complete thought — it's not (ends with a preposition) — so the agent waits. The trade-off: semantic endpointing adds ~200ms but eliminates false endpoints.
Semantic Endpointing
async def semantic_endpoint(partial_transcript, silence_duration_ms):
"""Use LLM to check if the caller is done speaking."""
# Only check after initial silence threshold
if silence_duration_ms < 300:
return False # Too early to check
# Ask LLM if this is a complete thought
is_complete = await semantic_vad.is_complete(partial_transcript)
if is_complete:
return True # Complete thought + silence = endpoint
# Incomplete but silence > 1s = probably done anyway
if silence_duration_ms > 1000:
return True # Force endpoint after long silence
return False # Wait for more speech
Turn-detector models
Asking a general LLM "is this thought complete?" works, but it costs a network round trip on every pause. The production frameworks ship small models trained for exactly that one question instead:
- LiveKit turn detector (in
livekit-agents): a small language model that reads the recent conversation transcript and outputs the probability that the user has finished their turn. It runs on CPU inside the agent, in tens of milliseconds. LiveKit combines it with VAD: silence opens the question, the model answers it, and a low end-of-turn probability stretches the wait. - Pipecat Smart Turn: an open-weights model that works on the audio itself rather than the transcript, so it hears what text cannot: falling intonation at the end of a statement, or the drawn-out "so..." of someone still thinking.
- OpenAI Realtime API: the server does turn detection for you.
turn_detection: {"type": "server_vad"} is silence-based, tuned with silence_duration_ms and threshold; {"type": "semantic_vad", "eagerness": "low" | "medium" | "high" | "auto"} adds a model that judges whether the user is done. Higher eagerness answers sooner and cuts in more often; lower eagerness waits longer.
All three are the hybrid pattern below with the LLM call replaced by a purpose-built model: silence decides when to ask, the model decides whether the turn is over.
Hybrid Endpointing
The best approach combines fixed silence with semantic checking:
Adaptive Silence
import statistics
class AdaptiveEndpointer:
"""Adjust the silence threshold to how fast this caller speaks."""
def __init__(self, base_threshold=500):
self.base_threshold = base_threshold
self.speech_rates = [] # words per second, one entry per utterance
def get_threshold(self, word_count, speech_duration_s):
"""Record the last utterance's speaking rate and return the threshold (ms)."""
if speech_duration_s > 0:
self.speech_rates.append(word_count / speech_duration_s)
if len(self.speech_rates) < 3:
return self.base_threshold
avg_rate = statistics.mean(self.speech_rates[-5:])
if avg_rate > 3.0: # Fast speaker (conversational English is ~2-3 words/s)
return 300 # Their pauses are short; wait less
elif avg_rate < 1.8: # Slow, deliberate speaker
return 700 # Their pauses are long; wait more
return self.base_threshold
Unlock the full lesson
You've read the first 2 sections. The rest of this lesson covers Build, Check Yourself, Key Terms & Next — plus a hands-on lab, quiz, and project artifact.
Create a free account — the opening phases of 29 of 35 courses are free, no credit card.