Voice Interruption Handling¶

RoomKit provides four interruption strategies to control what happens when a user speaks while the AI is talking. This guide covers configuration, backchannel detection, and practical patterns for each strategy.

Quick Start¶

from __future__ import annotations

from roomkit.channels import VoiceChannel
from roomkit.voice.interruption import InterruptionConfig, InterruptionStrategy

voice = VoiceChannel(
    "voice",
    stt=stt,
    tts=tts,
    backend=backend,
    interruption=InterruptionConfig(
        strategy=InterruptionStrategy.CONFIRMED,
        min_speech_ms=300,
    ),
)

Strategy Comparison¶

Strategy	Behavior	Use Case	False Positives
IMMEDIATE	Cancel TTS on any speech detection	IVR, command interfaces	Highest
CONFIRMED	Wait for `min_speech_ms` before interrupting	General conversation	Low
SEMANTIC	Use backchannel detection to distinguish "uh-huh" from real interruptions	Customer support, therapy	Lowest
DISABLED	Never interrupt — queue user speech until playback ends	Announcements, legal disclaimers	None

Configuration¶

from roomkit.voice.interruption import InterruptionConfig, InterruptionStrategy

config = InterruptionConfig(
    strategy=InterruptionStrategy.CONFIRMED,
    min_speech_ms=300,
    allow_during_first_ms=0,
    flush_partial_tts=True,
    keep_partial_transcript=True,
    backchannel_detector=None,
)

Field	Default	Description
`strategy`	`CONFIRMED`	Which interruption strategy to use
`min_speech_ms`	`300`	Minimum speech duration (ms) before confirming interruption (CONFIRMED strategy)
`allow_during_first_ms`	`0`	Grace period: don't interrupt until N ms of audio were played (synthesis latency does not count)
`flush_partial_tts`	`True`	Discard audio already handed to the backend. `False` lets the current utterance finish while the user's speech is processed alongside it
`keep_partial_transcript`	`True`	Record what the room actually heard as an internal event when the bot is cut off
`backchannel_detector`	`None`	Required for SEMANTIC strategy
`transcript_wait_ms`	`1000`	SEMANTIC: how long transcribed speech may run without words before it is judged on duration alone
`echo_guard_ms`	`6000`	While a session has played less response audio than this, speech made only of the response's own words is its echo: it neither interrupts nor becomes a turn (Echo guard). `0` turns it off

What an interruption leaves behind¶

The AI's response event is written when the text is produced, not when it is heard, so a bot cut off after two words still has its whole line in the timeline. With keep_partial_transcript on, the interruption records what the room actually heard as an internal-visibility event:

{
    "interrupted": True,
    "played_ms": 1840,
    "voice_session_id": "...",
    # "played_percentage": 46.0   — only when the utterance's duration is known
}

The full text is stored rather than a truncated guess: played_ms says how far playback got, and cutting the string at the same proportion would invent a word boundary that TTS timing does not guarantee.

Strategy Details¶

IMMEDIATE¶

Cancels TTS playback the instant any speech is detected. Best for systems where responsiveness matters more than naturalness.

voice = VoiceChannel(
    "voice",
    stt=stt, tts=tts, backend=backend,
    interruption=InterruptionConfig(
        strategy=InterruptionStrategy.IMMEDIATE,
        allow_during_first_ms=200,  # Prevent echo-triggered interruptions
    ),
)

Tip

Use allow_during_first_ms to add a grace period at the start of playback. This prevents the speaker's own audio echoing back and triggering a false interruption.

CONFIRMED¶

Waits until the user has been speaking for at least min_speech_ms before confirming the interruption. Filters out brief noise and accidental sounds.

The evaluation happens twice. At speech onset no duration has accumulated yet, so the first look can only say "not yet" — the segment is held as possible echo and a second look is taken once the speech has had min_speech_ms to sustain. If the speech stopped in between, it was a blip and nothing is interrupted. The confirmation logs at INFO:

Barge-in confirmed after 300ms of sustained speech (session <id>)

That deferral is audible in one place: with a streaming STT provider, capture starts at the confirmation, so roughly the first min_speech_ms of the utterance does not reach the streaming transcript. The batch fallback receives the whole segment, so the text is not lost — but lower min_speech_ms if you need the leading words on the streaming path.

voice = VoiceChannel(
    "voice",
    stt=stt, tts=tts, backend=backend,
    interruption=InterruptionConfig(
        strategy=InterruptionStrategy.CONFIRMED,
        min_speech_ms=300,  # Require 300ms of continuous speech
    ),
)

SEMANTIC¶

Uses a BackchannelDetector to distinguish backchannels ("uh-huh", "yeah", "ok") from real interruptions. Falls back to CONFIRMED if no detector is configured.

The detector is never asked about an empty speech onset: when the pipeline VAD reports speech start there are no words yet and no duration. SEMANTIC holds the speech (nothing reaches the AI) until it has something to judge:

With a streaming STT, the held speech is transcribed while the assistant talks and each partial transcript is classified. "uh-huh" fires ON_BACKCHANNEL once, the assistant keeps talking and the segment is discarded. "wait, stop" cuts the assistant off and becomes the user's turn, from its first word.
Without one, the speech is judged on duration alone once it has lasted min_speech_ms, as CONFIRMED does. A keyword detector sees an empty transcript there and lets the interruption through; a detector working on speech_duration_ms or audio_bytes can still recognise a backchannel.

A streaming STT often needs longer than min_speech_ms for its first words. While one is transcribing the speech, SEMANTIC waits for them up to transcript_wait_ms (default 1000) before judging on duration alone. Raise it for a slow STT; lower it if a real interruption with no words should still cut quickly.

Speech that ends before its first word is judged on its final transcript: a streaming transducer often releases a short word only once the speech is over (Nemotron gives "okay" and "no" only at the end). No words, or a backchannel, is discarded and the assistant talks on; anything else ("stop") cuts it off as soon as the STT has finalized, and becomes the user's turn. So a longer transcript_wait_ms (2000 in the local examples) spares the assistant from being cut by an "okay" without swallowing a short "stop".

In continuous-STT mode (no local VAD), the energy barge-in classifies the words of the burst under way: a burst recognized as "uh-huh" fires ON_BACKCHANNEL once and is not cut for running long, while words that turn into a request still interrupt.

A backchannel lets the assistant keep talking, and its speech segment is discarded: the "uh-huh" does not become a user message (RFC §12.6).

voice = VoiceChannel(
    "voice",
    stt=stt, tts=tts, backend=backend,
    interruption=InterruptionConfig(
        strategy=InterruptionStrategy.SEMANTIC,
        backchannel_detector=PhraseBackchannelDetector(),
    ),
)

DISABLED¶

Never interrupts. Speech that arrives while the assistant is talking is queued, not discarded: the assistant finishes its turn, and the held utterance is then processed as if it had just been spoken.

voice = VoiceChannel(
    "voice",
    stt=stt, tts=tts, backend=backend,
    interruption=InterruptionConfig(
        strategy=InterruptionStrategy.DISABLED,
    ),
)

The flush happens once the playback has ended — as soon as the last audio is sent when an AEC runs (the pipeline's or the backend's), after the two-second echo-decay drain otherwise — and it logs at INFO so a session shows plainly that nothing was dropped:

Flushing 1 queued speech segment(s) for <session> (playback finished)

Two consequences to design around:

The caller's interjection is answered a few seconds late by construction. That delay is the point of the strategy, not a defect.
The backlog is bounded. A caller who talks continuously through a very long answer will have their earliest segments dropped, with a warning, rather than growing the queue without limit.

Use it for announcements, legal disclaimers and any turn that must be heard in full — while still hearing what the caller said during it.

Backchannel Detection¶

The SEMANTIC strategy requires a BackchannelDetector to classify short utterances.

PhraseBackchannelDetector¶

RoomKit ships one that reads the words: an utterance made only of known acknowledgements is a backchannel. "Okay", "mm-hmm", "yeah, right", "d'accord" or "c'est ça" let the assistant keep talking; "okay, and what about Calgary?" does not, and since each partial transcript is classified as it grows, the words after the "okay" still cut in. Thanks and reactions count as acknowledgements too: "merci", "thank you", "wow", "ouh", "ayoye" or a laugh ("haha", "[laughter]") said over the assistant answer what it just said, while "merci, mais attends" or "non merci" cut in. The French list covers Québec ("ouin", "mets-en", "c'est malade") as well as France. Left out on purpose: what asks the assistant to stop or repeat ("wait", "huh?", "hein ?"), what disagrees or doubts ("no", "oh no", "really?", "voyons donc"), what opens a turn ("um", "euh", "well"), and polite refusals ("that's okay", "c'est correct"). Punctuation, case and held letters ("hmmm", "ouiii") do not matter; a hummed "no" ("mm-mm", "uh-uh") still cuts in.

from __future__ import annotations

from roomkit.voice.interruption import InterruptionConfig, InterruptionStrategy
from roomkit.voice.pipeline.backchannel import (
    ENGLISH_BACKCHANNELS,
    PhraseBackchannelDetector,
)

interruption = InterruptionConfig(
    strategy=InterruptionStrategy.SEMANTIC,
    backchannel_detector=PhraseBackchannelDetector(),  # English and French
)
english_only = PhraseBackchannelDetector(ENGLISH_BACKCHANNELS + ("righto",), max_words=3)

Option	Default	Description
`phrases`	`ENGLISH_BACKCHANNELS + FRENCH_BACKCHANNELS`	The acknowledgements, multi-word ones included ("i see", "tout à fait")
`max_words`	`4`	A longer utterance is never a backchannel, whatever its words
`cut_without_words`	`True`	Whether an utterance without words may cut the assistant (on its duration)

It needs words: with no transcript it judges no utterance a backchannel, and SEMANTIC falls back on duration as CONFIRMED does. Pair it with a streaming STT. A short sound that ends before it can be confirmed (a breath, a remnant of echo) is dropped by the channel anyway, without cutting the assistant.

With cut_without_words=False only words interrupt: sound the STT made nothing of (echo an AEC left, a cough, a door) never cuts the assistant, however long it lasts, and neither does a wordless interruption. On a laptop's speakers, three of six cuts once came from such sound, with no word in the transcript. The examples' INTERRUPTION=words sets it.

words_only = InterruptionConfig(
    strategy=InterruptionStrategy.SEMANTIC,
    backchannel_detector=PhraseBackchannelDetector(cut_without_words=False),
)

A detector of your own¶

Implement the ABC, for an acoustic classifier for instance:

from __future__ import annotations

from roomkit.voice.pipeline.backchannel import (
    BackchannelContext,
    BackchannelDecision,
    BackchannelDetector,
)


class KeywordBackchannelDetector(BackchannelDetector):
    """Simple keyword-based backchannel detector."""

    BACKCHANNELS = {"uh-huh", "yeah", "ok", "mhm", "right", "sure", "hmm", "yes"}

    @property
    def name(self) -> str:
        return "keyword_backchannel"

    def classify(self, context: BackchannelContext) -> BackchannelDecision:
        text = (context.transcript or "").strip().lower()
        is_bc = text in self.BACKCHANNELS and context.speech_duration_ms < 1000
        return BackchannelDecision(
            is_backchannel=is_bc,
            confidence=0.9 if is_bc else 0.1,
            label="acknowledgement" if is_bc else None,
        )

BackchannelContext¶

The detector receives rich context for classification:

Field	Type	Description
`transcript`	`str \\| None`	Transcribed text of the utterance
`speech_duration_ms`	`float`	Duration of the utterance
`audio_bytes`	`bytes \\| None`	Raw audio for acoustic classifiers
`bot_speech_progress`	`float`	How far into the bot's utterance (0.0-1.0)
`session_id`	`str`	Voice session ID
`metadata`	`dict`	Additional context

Pipeline Integration¶

Interruption config can also be set via AudioPipelineConfig:

from __future__ import annotations

from roomkit.channels import VoiceChannel
from roomkit.voice.interruption import InterruptionConfig, InterruptionStrategy
from roomkit.voice.pipeline import AudioPipelineConfig

pipeline = AudioPipelineConfig(
    vad=vad,
    denoiser=denoiser,
    backchannel_detector=my_detector,
    interruption=InterruptionConfig(
        strategy=InterruptionStrategy.SEMANTIC,
        min_speech_ms=300,
    ),
)

voice = VoiceChannel(
    "voice",
    stt=stt, tts=tts,
    backend=backend,
    pipeline=pipeline,
)

Configuration Priority

Explicit interruption parameter on VoiceChannel (highest)
pipeline.interruption from AudioPipelineConfig
Legacy enable_barge_in + barge_in_threshold_ms (fallback)

Hook Integration¶

Three hooks fire during interruption events:

Where the speech is detected does not change who decides

Speech during playback can be noticed by the pipeline's VAD, by the continuous-STT energy check, or by a transport that does its own detection and calls on_barge_in. All three routes consult the same InterruptionConfig: a backend with native detection cannot override DISABLED, and it is subject to allow_during_first_ms like every other route. A transport reports detected speech, not a decision.

ON_BARGE_IN¶

Fires when the user successfully interrupts the AI:

from __future__ import annotations

from roomkit import HookTrigger, HookExecution

@kit.hook(HookTrigger.ON_BARGE_IN, execution=HookExecution.ASYNC)
async def on_barge_in(event, ctx):
    # event is a BargeInEvent
    # audio_position_ms: audio the user heard, the timeline's played_ms
    print(f"User interrupted at {event.audio_position_ms}ms")
    print(f"AI was saying: {event.interrupted_text}")

ON_TTS_CANCELLED¶

Fires when TTS playback is cancelled for any reason:

@kit.hook(HookTrigger.ON_TTS_CANCELLED, execution=HookExecution.ASYNC)
async def on_tts_cancelled(event, ctx):
    # event is a TTSCancelledEvent
    # event.reason: "barge_in" | "explicit" | "disconnect" | "error"
    if event.reason == "barge_in":
        print(f"TTS stopped at {event.audio_position_ms}ms due to barge-in")

ON_BACKCHANNEL¶

Fires when the SEMANTIC strategy detects a backchannel (not a real interruption):

@kit.hook(HookTrigger.ON_BACKCHANNEL, execution=HookExecution.ASYNC)
async def on_backchannel(event, ctx):
    # event is a BackchannelEvent
    print(f"Backchannel: '{event.text}' (confidence={event.confidence})")

Echo guard: echo_guard_ms¶

An echo canceller learns the room from the responses it hears. At a session's start it has heard none: the first response's opening words ("Hi! How are you?") come back through the microphone at full strength, a streaming STT transcribes them, and they cut the agent off, then arrive as the person's turn.

Until a session has played echo_guard_ms of response audio (6 s by default), speech whose words are only words of the response being played, consecutive and in order, is taken for its echo, whatever the strategy. Case, punctuation and accents do not matter. In a partial transcript the last word may be cut short after a whole word ("Hi! How are y"); a lone word and a final are matched whole, so "No" over "I'm not sure" cuts in. The echo does not interrupt and is no turn: its partial transcripts are not published, and its final transcript is discarded, even when the STT delivers it seconds after the response ended (up to 10 s). One word the response does not say there makes it speech: "Hi, wait" cuts in, and so does a final that says it during the playback when its first words were taken for the echo. Audio played with VoiceChannel.play() counts toward the guard. In continuous-STT mode, while the guard is open, loud audio with no words yet waits for the STT's first words (up to transcript_wait_ms) before it may cut on its duration.

Words taken for the echo decide nothing else: when the STT never closes that utterance, they cut neither the response that took over from the echoed one nor the next, nor do the STT's later partials of the same echo, and they lapse with the verdict.

Past the guard, the response's own words cut in as any words do, so someone answering "the sword" to "the sword or the shield?" is heard. Raise it for a canceller that learns slowly, or set echo_guard_ms=0 to turn it off.

With an STT that tells voices apart (Meta's DIARIZATION), the agent's own voice is caught past the guard too. A voice heard speaking while nothing played is a person's. A final from a voice never heard so, said over a response, whose words are the response's in their order, one misheard at most ("Eh bien, la lampe salon" for "c'est bien la lampe Salon") and three at least heard right, is the response coming back past the canceller: it is discarded. Its first words may still have cut the response, as the voice is known only with the final. The rule has a cost: a person not yet heard while nothing played who answers over a response with three or more of its words ("the kitchen lamp" to "the kitchen lamp or the living room lamp?") cuts it and is not heard, so the agent stops and says nothing; what they say next while nothing plays is heard. Voices are known within one STT stream: when the provider closes the stream, those heard alone are forgotten, as the next one numbers its voices afresh.

Grace Period: allow_during_first_ms¶

The allow_during_first_ms field prevents interruptions during the first N milliseconds of playback. It counts audio actually played: a response still waiting for its first TTS chunk has played 0 ms, however long synthesis takes. This is useful for:

Echo prevention: When AEC isn't perfect, the first few hundred ms of playback can echo back and trigger false interruptions
Critical content: Ensure the AI's opening words are always heard

InterruptionConfig(
    strategy=InterruptionStrategy.IMMEDIATE,
    allow_during_first_ms=500,  # Don't interrupt during first 500ms
)

On RealtimeVoiceChannel, a provider such as OpenAI may own the VAD and cancel its response before RoomKit receives the speech-start callback. When the transport reports actual playback, RoomKit therefore enforces this grace period at the provider-input boundary: AEC, AGC, denoising, and recording still receive the real microphone signal, while the provider receives equal-duration PCM silence. Once the interval expires, normal audio and barge-in resume. Transports without playback callbacks cannot apply this physical-playback guard.

Legacy Compatibility¶

The older enable_barge_in and barge_in_threshold_ms parameters still work and map to InterruptionConfig internally:

# Legacy (still works):
VoiceChannel("voice", enable_barge_in=True, barge_in_threshold_ms=300)
# Equivalent to:
# InterruptionConfig(strategy=IMMEDIATE, allow_during_first_ms=300)

# Legacy disabled:
VoiceChannel("voice", enable_barge_in=False)
# Equivalent to:
# InterruptionConfig(strategy=DISABLED)

Warning

Prefer the explicit interruption parameter over legacy fields for new code.

Testing¶

Use MockBackchannelDetector for deterministic tests:

from __future__ import annotations

from roomkit.voice.interruption import InterruptionConfig, InterruptionHandler, InterruptionStrategy
from roomkit.voice.pipeline.backchannel import BackchannelDecision
from roomkit.voice.pipeline.backchannel.mock import MockBackchannelDetector

# Test CONFIRMED strategy
config = InterruptionConfig(strategy=InterruptionStrategy.CONFIRMED, min_speech_ms=300)
handler = InterruptionHandler(config)
decision = handler.evaluate(playback_position_ms=500, speech_duration_ms=400)
assert decision.should_interrupt

# Test SEMANTIC with backchannel
detector = MockBackchannelDetector(
    decisions=[BackchannelDecision(is_backchannel=True, confidence=0.95)]
)
config = InterruptionConfig(
    strategy=InterruptionStrategy.SEMANTIC,
    backchannel_detector=detector,
)
handler = InterruptionHandler(config)
decision = handler.evaluate(playback_position_ms=500, speech_duration_ms=300, speech_text="uh-huh")
assert not decision.should_interrupt
assert decision.is_backchannel