mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-31 19:16:29 +00:00
fix(voice): full-duplex agent-turn listener — interrupt by voice during generation AND playback
Replaces the half-duplex per-playback barge monitors with ONE listener that runs for the entire agent turn in continuous voice mode: armed at utterance-submit, disarmed when the turn is fully done (response + TTS finished). Fixes Teknium's live report that voice interruption never works: (a) not while the LLM is generating, (b) not while TTS plays. Root causes: - HALF-DUPLEX GAP: the barge monitor only spawned when TTS playback STARTED (cli.py streaming/whole-file paths, gateway _tts_stream_begin). During LLM generation there was NO microphone listener at all. - PLAYBACK DEAFNESS: the monitor calibrated its VAD noise floor WHILE the speaker was blasting TTS (speaker bleed baked into the floor), then multiplied it by 8x with a 1s strictly-consecutive block requirement — normal speech could rarely reach the trigger, and the 2s grace swallowed early interjections. New model — tools/voice_mode.full_duplex_listen(): - Pre-playback calibration: quiet-room noise floor established at turn start and HELD through playback (never recalibrated against bleed). - Phase-aware trigger: generation = floor x voice.barge_in_threshold_multiplier (new config, default 3.0, justified by synthetic-frame tests); playback = additionally clamped to a 1500-RMS minimum so bleed alone can't trip; 4000-RMS ceiling keeps speech always reachable. - Windowed-majority detection (>=80% of a 300ms window) instead of the strictly-consecutive counter that reset on intra-word energy dips. - Grace on playback ONSET only (voice.barge_in_grace_seconds, default down 2.0 -> 0.5) — suppresses the onset transient, not the mic. - Debug diagnostics at every decision point (calibrated floor, per-window RMS above 50% of trigger, trip/no-trip, grace suppressions) — always logger.debug, mirrored to stderr under HERMES_VOICE_DEBUG=1. Phase behavior (CLI cli.py + tui_gateway/server.py, same model): - generation: speech interrupts the in-flight turn via the SAME seam the typed/Ctrl+C interrupt uses (agent.interrupt()), cuts any pending TTS pipeline so the stale reply never plays, and submits the captured interjection (pre-roll capture, first syllable kept) as the next turn. - playback: cuts TTS (streaming pipeline stop + fallback speak stop events + file player) and submits the capture. - stop phrase honored in BOTH phases: mid-generation 'stop' interrupts the turn AND ends the voice chat (stop everything). - one listener instance spans generation -> playback (no re-arm race); double-arm refused (CLI _voice_fd_active / gateway _fd_listener_active). Gateway specifics: _arm_full_duplex_listener() at _run_prompt_submit turn start and inside _tts_stream_begin; _speak_text_with_barge registers its (stop, done) pair in _fd_speak_pipelines so fallback speaks are cut and tracked; _tts_stream_barge_in_monitor kept as a shim that arms the new listener. Desktop renderer owns its own mic path (voice-barge-in.ts) and is unaffected; if desktop backend-mic mode is used it inherits via the gateway. Tests: full_duplex_listen synthetic-RMS suite (speech-over-bleed trips, bleed alone doesn't, quiet floor held through playback, grace window, multiplier math 3x vs 8x, windowed-majority dips), CLI listener phase tests (generation interrupt seam, playback cut, lifecycle spans phases, double-arm, config forwarding, stop-phrase-mid-generation), gateway generation-phase interrupt + stop-phrase tests. Generation-interrupt test sabotage-verified.
This commit is contained in:
Teknium
parent
8c196ed85c
commit
5081551f09
8 changed files with 987 additions and 138 deletions
160
cli.py
160
cli.py
|
|
@ -12060,17 +12060,14 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin):
|
|||
args=(text,),
|
||||
daemon=True,
|
||||
).start()
|
||||
# Spoken barge-in must work on the whole-file fallback path too —
|
||||
# previously only the streaming pipeline armed the monitor, so when
|
||||
# streaming TTS couldn't start (missing sounddevice, failed probe)
|
||||
# talking over the reply did nothing. The monitor's _cut_playback
|
||||
# uses stop_playback(), which kills the file player, so the same
|
||||
# machinery covers this path; the stop event it receives is only
|
||||
# used to signal the (nonexistent) streaming pipeline.
|
||||
# Spoken barge-in must work on the whole-file fallback path too. The
|
||||
# full-duplex agent-turn listener normally already covers playback
|
||||
# (armed at turn start in chat()); this arm is an idempotent safety
|
||||
# net for speak calls outside a chat turn — the listener refuses to
|
||||
# double-arm via _voice_fd_active.
|
||||
if self._voice_continuous:
|
||||
threading.Thread(
|
||||
target=self._voice_barge_in_monitor,
|
||||
args=(threading.Event(),),
|
||||
target=self._voice_full_duplex_listener,
|
||||
daemon=True,
|
||||
).start()
|
||||
|
||||
|
|
@ -12147,60 +12144,107 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin):
|
|||
self._voice_tts_done.set()
|
||||
|
||||
|
||||
def _voice_barge_in_monitor(self, stop_event: threading.Event) -> None:
|
||||
"""VAD barge-in: cut streaming TTS the moment the user starts talking.
|
||||
def _voice_full_duplex_listener(self) -> None:
|
||||
"""Full-duplex agent-turn listener: mic live for the WHOLE turn.
|
||||
|
||||
Runs for one turn alongside the streaming pipeline (continuous voice
|
||||
mode only — the mic is otherwise idle during playback). On speech,
|
||||
playback is cut immediately while the monitor KEEPS capturing (with
|
||||
pre-roll, so the interruption is transcribed from its first syllable
|
||||
— restarting the recorder after detection would lose the opening
|
||||
words). ``_voice_barge_capture`` suppresses process_loop's auto-
|
||||
restart until the captured utterance has been submitted.
|
||||
Armed at utterance-submit (chat() start in continuous voice mode) and
|
||||
disarmed when the turn is fully done (agent finished + TTS played).
|
||||
Replaces the old per-playback ``_voice_barge_in_monitor``, which only
|
||||
listened while TTS audio was playing — during LLM generation the mic
|
||||
was dead, so the user could not interject by voice at all (and the
|
||||
playback monitor calibrated against its own speaker bleed, making
|
||||
the trigger unreachable; see tools.voice_mode.full_duplex_listen).
|
||||
|
||||
A short startup grace period delays VAD activation so TTS playback
|
||||
has time to establish before the mic starts listening. Without
|
||||
this, speaker bleed during the first few hundred milliseconds can
|
||||
falsely trigger barge-in and cut the response short.
|
||||
Phase behaviour:
|
||||
|
||||
* generation (no TTS audio yet): speech interrupts the in-flight
|
||||
agent turn via ``self.agent.interrupt()`` — the same seam the
|
||||
typed/Ctrl+C interrupt uses — and the captured utterance is
|
||||
submitted as the next message.
|
||||
* playback: speech cuts TTS (pipeline stop event + stop_playback)
|
||||
and the interruption is captured with pre-roll and submitted.
|
||||
|
||||
The stop phrase ends the voice chat in BOTH phases (a stop during
|
||||
generation means "stop everything": the turn is already interrupted
|
||||
at trip time, then ``_voice_submit_barge_utterance`` disables voice
|
||||
mode).
|
||||
"""
|
||||
fd_active = getattr(self, "_voice_fd_active", None)
|
||||
if fd_active is None:
|
||||
fd_active = threading.Event()
|
||||
self._voice_fd_active = fd_active
|
||||
if fd_active.is_set():
|
||||
return # one listener owns the mic for this turn
|
||||
fd_active.set()
|
||||
try:
|
||||
from hermes_cli.config import load_config
|
||||
voice_cfg = load_config().get("voice") or {}
|
||||
if not (isinstance(voice_cfg, dict) and voice_cfg.get("barge_in", True)):
|
||||
return
|
||||
from tools.voice_mode import listen_for_speech, stop_playback
|
||||
from tools.voice_mode import (
|
||||
full_duplex_listen,
|
||||
is_audio_output_active,
|
||||
stop_playback,
|
||||
)
|
||||
|
||||
# Grace period: wait briefly before opening the mic so the
|
||||
# first TTS sentence is already playing and the VAD calibration
|
||||
# samples the actual playback level (not silence). This
|
||||
# prevents speaker bleed from falsely triggering barge-in
|
||||
# at the start of playback.
|
||||
_grace_s = float(voice_cfg.get("barge_in_grace_seconds", 2.0))
|
||||
if _grace_s > 0:
|
||||
stop_event.wait(timeout=_grace_s)
|
||||
if stop_event.is_set() or self._voice_tts_done.is_set():
|
||||
return
|
||||
try:
|
||||
_mult = float(voice_cfg.get("barge_in_threshold_multiplier", 0) or 0)
|
||||
except (TypeError, ValueError):
|
||||
_mult = 0.0
|
||||
try:
|
||||
_grace_ms = int(float(voice_cfg.get("barge_in_grace_seconds", 0.5)) * 1000)
|
||||
except (TypeError, ValueError):
|
||||
_grace_ms = 500
|
||||
|
||||
def _cut_playback():
|
||||
if not self._voice_tts_done.is_set():
|
||||
import traceback as _tb
|
||||
tts_done = getattr(self, "_voice_tts_done", None)
|
||||
|
||||
def _should_stop() -> bool:
|
||||
if not (getattr(self, "_voice_mode", False) and getattr(self, "_voice_continuous", False)):
|
||||
return True
|
||||
if getattr(self, "_agent_running", False):
|
||||
return False
|
||||
# Agent finished — keep listening until TTS fully played.
|
||||
if tts_done is not None and not tts_done.is_set():
|
||||
return False
|
||||
return not is_audio_output_active()
|
||||
|
||||
def _on_trigger(phase: str) -> None:
|
||||
# Latch BEFORE cutting anything: suppresses process_loop's
|
||||
# auto-restart until the capture is submitted.
|
||||
self._voice_barge_capture.set()
|
||||
if phase == "playback":
|
||||
logger.debug(
|
||||
"TTS CUT: barge-in _cut_playback fired (VAD trip) — "
|
||||
"stop_event.set() + stop_playback()\n%s",
|
||||
"".join(_tb.format_stack()),
|
||||
"TTS CUT: full-duplex listener tripped during playback"
|
||||
)
|
||||
from tools.tts_streaming import mark_speech_interrupted
|
||||
mark_speech_interrupted()
|
||||
self._voice_barge_capture.set()
|
||||
stop_event.set()
|
||||
_pipe_stop = getattr(self, "_voice_tts_stop", None)
|
||||
if _pipe_stop is not None:
|
||||
_pipe_stop.set()
|
||||
stop_playback()
|
||||
else:
|
||||
# Generation phase: no audio to cut — interrupt the
|
||||
# in-flight agent turn (same seam as typed interrupt).
|
||||
logger.debug(
|
||||
"full-duplex listener tripped during generation — "
|
||||
"interrupting agent turn"
|
||||
)
|
||||
_pipe_stop = getattr(self, "_voice_tts_stop", None)
|
||||
if _pipe_stop is not None:
|
||||
_pipe_stop.set() # never let the stale reply speak
|
||||
try:
|
||||
if self.agent is not None and getattr(self, "_agent_running", False):
|
||||
_cprint(f"\n{_DIM}🎤 Voice interjection — interrupting…{_RST}")
|
||||
self.agent.interrupt()
|
||||
except Exception as e:
|
||||
logger.debug("voice interjection interrupt failed: %s", e)
|
||||
|
||||
wav_path = listen_for_speech(
|
||||
lambda: stop_event.is_set() or self._voice_tts_done.is_set(),
|
||||
capture=True,
|
||||
on_trigger=_cut_playback,
|
||||
sustained_ms=1000,
|
||||
calibration_ms=800,
|
||||
wav_path = full_duplex_listen(
|
||||
_should_stop,
|
||||
is_playing=is_audio_output_active,
|
||||
on_trigger=_on_trigger,
|
||||
multiplier=_mult or None,
|
||||
grace_ms=max(0, _grace_ms),
|
||||
)
|
||||
if wav_path and self._voice_barge_capture.is_set():
|
||||
self._voice_submit_barge_utterance(wav_path)
|
||||
|
|
@ -12208,7 +12252,9 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin):
|
|||
self._voice_barge_capture.clear()
|
||||
except Exception as e:
|
||||
self._voice_barge_capture.clear()
|
||||
logger.debug("Voice barge-in monitor failed: %s", e)
|
||||
logger.debug("Voice full-duplex listener failed: %s", e)
|
||||
finally:
|
||||
fd_active.clear()
|
||||
|
||||
def _voice_submit_barge_utterance(self, wav_path: str) -> None:
|
||||
"""Transcribe a barge-captured interruption and queue it as the next turn."""
|
||||
|
|
@ -13372,6 +13418,16 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin):
|
|||
# reset at the start of each user turn.
|
||||
self._reasoning_shown_this_turn = False
|
||||
|
||||
# Full-duplex agent-turn listener (continuous voice mode): arm
|
||||
# the mic NOW — at utterance-submit — not when TTS playback
|
||||
# starts. It spans generation (speech interrupts the turn) and
|
||||
# playback (speech cuts TTS), and disarms itself when the turn
|
||||
# is fully done. See _voice_full_duplex_listener.
|
||||
if self._voice_mode and self._voice_continuous:
|
||||
threading.Thread(
|
||||
target=self._voice_full_duplex_listener, daemon=True
|
||||
).start()
|
||||
|
||||
# --- Streaming TTS setup ---
|
||||
# Any working TTS provider streams sentence-by-sentence as the agent
|
||||
# generates tokens: PCM-streaming providers (ElevenLabs, OpenAI) play
|
||||
|
|
@ -13431,12 +13487,10 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin):
|
|||
)
|
||||
tts_thread.start()
|
||||
# Expose the pipeline's stop event so barge-in paths (voice
|
||||
# key, VAD monitor) can cut playback from outside this turn.
|
||||
# key, full-duplex listener) can cut playback from outside
|
||||
# this turn. The full-duplex listener itself was armed at
|
||||
# turn start (see above) — it spans generation AND playback.
|
||||
self._voice_tts_stop = stop_event
|
||||
if self._voice_continuous:
|
||||
threading.Thread(
|
||||
target=self._voice_barge_in_monitor, args=(stop_event,), daemon=True
|
||||
).start()
|
||||
|
||||
def stream_callback(delta: str):
|
||||
if text_queue is not None:
|
||||
|
|
|
|||
|
|
@ -1449,8 +1449,9 @@ DEFAULT_CONFIG = {
|
|||
"thinking_sound": True, # Calm ambient bubble sound while the agent works in voice chat (volume follows beep_volume)
|
||||
"silence_threshold": 200, # RMS below this = silence (0-32767)
|
||||
"silence_duration": 3.0, # Seconds of silence before auto-stop
|
||||
"barge_in": True, # Stop TTS playback when the user starts talking
|
||||
"barge_in_grace_seconds": 2.0, # Delay before the barge mic opens so VAD calibrates against live TTS playback (0 disables)
|
||||
"barge_in": True, # Interrupt the agent / stop TTS when the user starts talking
|
||||
"barge_in_grace_seconds": 0.5, # Trip suppression right after TTS playback starts (onset transient); the mic itself is live for the whole turn
|
||||
"barge_in_threshold_multiplier": 3.0, # Speech trigger = quiet-room floor x this (floor is calibrated BEFORE playback, never against speaker bleed)
|
||||
# Saying EXACTLY one of these phrases (and nothing else) ends the
|
||||
# voice chat instead of being sent to the agent. Case-insensitive,
|
||||
# surrounding punctuation ignored. Set [] to disable.
|
||||
|
|
|
|||
|
|
@ -14555,6 +14555,9 @@ def _fake_tts_modules(monkeypatch, *, requirements=True, playback_stops=None, li
|
|||
def default_listen(should_stop, capture=False, on_trigger=None, **_kw):
|
||||
return None if capture else False
|
||||
|
||||
def default_fd_listen(should_stop, is_playing=None, on_trigger=None, **_kw):
|
||||
return None
|
||||
|
||||
monkeypatch.setitem(
|
||||
sys.modules,
|
||||
"tools.tts_tool",
|
||||
|
|
@ -14572,9 +14575,13 @@ def _fake_tts_modules(monkeypatch, *, requirements=True, playback_stops=None, li
|
|||
types.SimpleNamespace(
|
||||
stop_playback=lambda: (playback_stops.append(True) if playback_stops is not None else None),
|
||||
listen_for_speech=listen or default_listen,
|
||||
full_duplex_listen=listen or default_fd_listen,
|
||||
is_audio_output_active=lambda: False,
|
||||
transcribe_recording=transcribe or (lambda path, model=None: {"success": True, "transcript": ""}),
|
||||
),
|
||||
)
|
||||
# Fresh listener slot per test — the arm is idempotent per process.
|
||||
monkeypatch.setattr(server, "_fd_listener_active", False)
|
||||
return started
|
||||
|
||||
|
||||
|
|
@ -14680,9 +14687,8 @@ def test_tts_stream_vad_barge_in_cuts_pipeline_and_submits_capture(monkeypatch,
|
|||
wav = tmp_path / "barge.wav"
|
||||
wav.write_bytes(b"RIFF")
|
||||
|
||||
def fake_listen(should_stop, capture=False, on_trigger=None, **_kw):
|
||||
assert capture is True
|
||||
on_trigger() # playback cut happens at detection, not after endpointing
|
||||
def fake_listen(should_stop, is_playing=None, on_trigger=None, **_kw):
|
||||
on_trigger("playback") # playback cut happens at detection
|
||||
return str(wav)
|
||||
|
||||
_fake_tts_modules(
|
||||
|
|
@ -14692,11 +14698,7 @@ def test_tts_stream_vad_barge_in_cuts_pipeline_and_submits_capture(monkeypatch,
|
|||
)
|
||||
|
||||
server._tts_stream_begin()
|
||||
with server._tts_stream_lock:
|
||||
state = server._tts_stream_state
|
||||
assert state is not None
|
||||
assert state["stop"].wait(5.0) # grace period (2s) + fake_listen + margin
|
||||
deadline = time.monotonic() + 2.0
|
||||
deadline = time.monotonic() + 5.0
|
||||
while time.monotonic() < deadline and wav.exists():
|
||||
time.sleep(0.01) # unlink (finally) runs after the transcript emit
|
||||
assert ("voice.interrupted", None) in events
|
||||
|
|
@ -14706,6 +14708,100 @@ def test_tts_stream_vad_barge_in_cuts_pipeline_and_submits_capture(monkeypatch,
|
|||
server._tts_stream_stop()
|
||||
|
||||
|
||||
def test_full_duplex_generation_phase_interrupts_running_turn(monkeypatch, tmp_path):
|
||||
"""Speech DURING LLM generation (no TTS audio yet) must interrupt the
|
||||
in-flight agent turn via the same seam session.interrupt uses, and the
|
||||
captured interjection is emitted as voice.transcript. This is the
|
||||
half-duplex gap: previously no listener existed until playback started."""
|
||||
import tools.tts_streaming as ts
|
||||
|
||||
ts._interrupted_at = None
|
||||
monkeypatch.setenv("HERMES_VOICE", "1")
|
||||
monkeypatch.setenv("HERMES_VOICE_TTS", "0")
|
||||
monkeypatch.setattr(server, "_load_cfg", lambda: {"voice": {"barge_in": True}})
|
||||
events: list = []
|
||||
monkeypatch.setattr(
|
||||
server, "_voice_emit", lambda event, payload=None: events.append((event, payload))
|
||||
)
|
||||
|
||||
wav = tmp_path / "interject.wav"
|
||||
wav.write_bytes(b"RIFF")
|
||||
|
||||
interrupted = threading.Event()
|
||||
fake_agent = types.SimpleNamespace(interrupt=lambda: interrupted.set())
|
||||
fake_session = {"running": True, "agent": fake_agent}
|
||||
monkeypatch.setattr(server, "_sessions", {"sid-fd": fake_session})
|
||||
|
||||
def fake_listen(should_stop, is_playing=None, on_trigger=None, **_kw):
|
||||
assert is_playing is not None and is_playing() is False # generation phase
|
||||
on_trigger("generation")
|
||||
return str(wav)
|
||||
|
||||
_fake_tts_modules(
|
||||
monkeypatch,
|
||||
listen=fake_listen,
|
||||
transcribe=lambda path, model=None: {"success": True, "transcript": "wait, try another way"},
|
||||
)
|
||||
|
||||
server._arm_full_duplex_listener()
|
||||
|
||||
deadline = time.monotonic() + 5.0
|
||||
while time.monotonic() < deadline and wav.exists():
|
||||
time.sleep(0.01)
|
||||
assert interrupted.is_set() # the running turn was interrupted
|
||||
assert ("voice.interrupted", None) in events
|
||||
assert ("voice.transcript", {"text": "wait, try another way"}) in events
|
||||
assert not wav.exists()
|
||||
|
||||
|
||||
def test_full_duplex_stop_phrase_mid_generation_ends_voice_chat(monkeypatch, tmp_path):
|
||||
"""Bare 'stop' during generation = interrupt the turn AND end the voice
|
||||
chat ('stop everything'), emitted as the explicit stop_phrase signal."""
|
||||
monkeypatch.setenv("HERMES_VOICE", "1")
|
||||
monkeypatch.setattr(server, "_load_cfg", lambda: {"voice": {"barge_in": True}})
|
||||
events: list = []
|
||||
monkeypatch.setattr(
|
||||
server, "_voice_emit", lambda event, payload=None: events.append((event, payload))
|
||||
)
|
||||
|
||||
wav = tmp_path / "stop.wav"
|
||||
wav.write_bytes(b"RIFF")
|
||||
|
||||
interrupted = threading.Event()
|
||||
fake_agent = types.SimpleNamespace(interrupt=lambda: interrupted.set())
|
||||
monkeypatch.setattr(
|
||||
server, "_sessions", {"sid-fd": {"running": True, "agent": fake_agent}}
|
||||
)
|
||||
|
||||
def fake_listen(should_stop, is_playing=None, on_trigger=None, **_kw):
|
||||
on_trigger("generation")
|
||||
return str(wav)
|
||||
|
||||
_fake_tts_modules(
|
||||
monkeypatch,
|
||||
listen=fake_listen,
|
||||
transcribe=lambda path, model=None: {"success": True, "transcript": "stop"},
|
||||
)
|
||||
# is_voice_stop_phrase lives in the faked tools.voice_mode namespace.
|
||||
sys.modules["tools.voice_mode"].is_voice_stop_phrase = (
|
||||
lambda text: text.strip().lower() == "stop"
|
||||
)
|
||||
monkeypatch.setitem(
|
||||
sys.modules,
|
||||
"hermes_cli.voice",
|
||||
types.SimpleNamespace(stop_continuous=lambda **_kw: None, speak_text=lambda *a, **k: None),
|
||||
)
|
||||
|
||||
server._arm_full_duplex_listener()
|
||||
|
||||
deadline = time.monotonic() + 5.0
|
||||
while time.monotonic() < deadline and wav.exists():
|
||||
time.sleep(0.01)
|
||||
assert interrupted.is_set()
|
||||
assert ("voice.transcript", {"stop_phrase": True, "text": "stop"}) in events
|
||||
assert os.environ.get("HERMES_VOICE") == "0" # voice chat ended
|
||||
|
||||
|
||||
def test_speak_text_with_barge_arms_monitor_and_cuts_playback(monkeypatch, tmp_path):
|
||||
"""The fallback whole-reply speak path (streaming pipeline couldn't
|
||||
start) and the voice.tts RPC must be barge-able too: speaking over the
|
||||
|
|
@ -14746,10 +14842,9 @@ def test_speak_text_with_barge_arms_monitor_and_cuts_playback(monkeypatch, tmp_p
|
|||
types.SimpleNamespace(speak_text=fake_speak_text),
|
||||
)
|
||||
|
||||
def fake_listen(should_stop, capture=False, on_trigger=None, **_kw):
|
||||
assert capture is True
|
||||
def fake_listen(should_stop, is_playing=None, on_trigger=None, **_kw):
|
||||
speak_started.wait(5)
|
||||
on_trigger() # user talks over the reply → cut now
|
||||
on_trigger("playback") # user talks over the reply → cut now
|
||||
return str(wav)
|
||||
|
||||
_fake_tts_modules(
|
||||
|
|
|
|||
|
|
@ -1456,6 +1456,200 @@ class TestVoiceBargeCaptureSubmit:
|
|||
assert restarted.wait(2.0) # continuous mode resumes listening
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Full-duplex agent-turn listener — CLI phase behaviour
|
||||
# ============================================================================
|
||||
|
||||
|
||||
class TestVoiceFullDuplexListener:
|
||||
"""_voice_full_duplex_listener: one mic for the whole turn. Generation-
|
||||
phase speech interrupts the in-flight agent turn; playback-phase speech
|
||||
cuts TTS; the capture is submitted either way."""
|
||||
|
||||
def _cli(self, monkeypatch, *, listen, voice_cfg=None, **overrides):
|
||||
cli = _make_voice_cli(
|
||||
_voice_mode=True, _voice_continuous=True, **overrides
|
||||
)
|
||||
cli.agent = None
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.config.load_config",
|
||||
lambda: {"voice": dict(voice_cfg or {"barge_in": True})},
|
||||
)
|
||||
monkeypatch.setattr("tools.voice_mode.full_duplex_listen", listen)
|
||||
monkeypatch.setattr("tools.voice_mode.is_audio_output_active", lambda: False)
|
||||
monkeypatch.setattr("tools.voice_mode.stop_playback", lambda: None)
|
||||
return cli
|
||||
|
||||
def test_generation_trip_interrupts_agent_and_submits(self, monkeypatch, tmp_path):
|
||||
"""Speech during generation → agent.interrupt() (the same seam the
|
||||
typed interrupt uses) + pending TTS pipeline cut + capture queued."""
|
||||
wav = tmp_path / "fd.wav"
|
||||
wav.write_bytes(b"RIFF")
|
||||
|
||||
def fake_listen(should_stop, is_playing=None, on_trigger=None, **_kw):
|
||||
on_trigger("generation")
|
||||
return str(wav)
|
||||
|
||||
cli = self._cli(monkeypatch, listen=fake_listen, _agent_running=True)
|
||||
interrupted = threading.Event()
|
||||
cli.agent = SimpleNamespace(interrupt=lambda: interrupted.set())
|
||||
pipe_stop = threading.Event()
|
||||
cli._voice_tts_stop = pipe_stop
|
||||
monkeypatch.setattr(
|
||||
"tools.voice_mode.transcribe_recording",
|
||||
lambda path, model=None: {"success": True, "transcript": "actually wait"},
|
||||
)
|
||||
|
||||
cli._voice_full_duplex_listener()
|
||||
|
||||
assert interrupted.is_set()
|
||||
assert pipe_stop.is_set() # stale reply's TTS can never play
|
||||
from cli import _VoiceInputMessage
|
||||
queued = cli._pending_input.get_nowait()
|
||||
assert isinstance(queued, _VoiceInputMessage)
|
||||
assert str(queued) == "actually wait"
|
||||
assert not cli._voice_barge_capture.is_set()
|
||||
|
||||
def test_playback_trip_cuts_tts_without_interrupting_agent(self, monkeypatch, tmp_path):
|
||||
"""Speech during playback → pipeline stop + stop_playback; the agent
|
||||
(already finished) is NOT interrupted."""
|
||||
wav = tmp_path / "fd.wav"
|
||||
wav.write_bytes(b"RIFF")
|
||||
stops = []
|
||||
|
||||
def fake_listen(should_stop, is_playing=None, on_trigger=None, **_kw):
|
||||
on_trigger("playback")
|
||||
return str(wav)
|
||||
|
||||
cli = self._cli(monkeypatch, listen=fake_listen, _agent_running=False)
|
||||
interrupted = threading.Event()
|
||||
cli.agent = SimpleNamespace(interrupt=lambda: interrupted.set())
|
||||
pipe_stop = threading.Event()
|
||||
cli._voice_tts_stop = pipe_stop
|
||||
monkeypatch.setattr("tools.voice_mode.stop_playback", lambda: stops.append(True))
|
||||
monkeypatch.setattr(
|
||||
"tools.voice_mode.transcribe_recording",
|
||||
lambda path, model=None: {"success": True, "transcript": "hang on"},
|
||||
)
|
||||
|
||||
cli._voice_full_duplex_listener()
|
||||
|
||||
assert not interrupted.is_set()
|
||||
assert pipe_stop.is_set()
|
||||
assert stops == [True]
|
||||
assert str(cli._pending_input.get_nowait()) == "hang on"
|
||||
|
||||
def test_listener_arms_at_submit_and_survives_into_playback(self, monkeypatch):
|
||||
"""Lifecycle: should_stop is False during generation AND during
|
||||
pending TTS (survives the phase transition — no re-arm race), and
|
||||
True once the turn is fully done."""
|
||||
probes = {}
|
||||
|
||||
def fake_listen(should_stop, is_playing=None, on_trigger=None, **_kw):
|
||||
# generation: agent running, TTS not started
|
||||
probes["generation"] = should_stop()
|
||||
# transition: agent done, TTS still pending
|
||||
cli._agent_running = False
|
||||
cli._voice_tts_done.clear()
|
||||
probes["playback_pending"] = should_stop()
|
||||
# turn fully done
|
||||
cli._voice_tts_done.set()
|
||||
probes["done"] = should_stop()
|
||||
return None
|
||||
|
||||
cli = self._cli(monkeypatch, listen=fake_listen, _agent_running=True)
|
||||
cli._voice_tts_done.set()
|
||||
|
||||
cli._voice_full_duplex_listener()
|
||||
|
||||
assert probes["generation"] is False
|
||||
assert probes["playback_pending"] is False # same listener spans phases
|
||||
assert probes["done"] is True
|
||||
|
||||
def test_double_arm_refused(self, monkeypatch):
|
||||
"""Only one listener may own the mic per turn — the second arm is a
|
||||
no-op (the fallback speak path arms as a safety net)."""
|
||||
calls = []
|
||||
|
||||
def fake_listen(should_stop, is_playing=None, on_trigger=None, **_kw):
|
||||
calls.append(True)
|
||||
return None
|
||||
|
||||
cli = self._cli(monkeypatch, listen=fake_listen, _agent_running=False)
|
||||
cli._voice_fd_active = threading.Event()
|
||||
cli._voice_fd_active.set() # a listener already owns the mic
|
||||
|
||||
cli._voice_full_duplex_listener()
|
||||
assert calls == []
|
||||
|
||||
def test_config_multiplier_and_grace_forwarded(self, monkeypatch):
|
||||
seen = {}
|
||||
|
||||
def fake_listen(should_stop, is_playing=None, on_trigger=None,
|
||||
multiplier=None, grace_ms=None, **_kw):
|
||||
seen["multiplier"] = multiplier
|
||||
seen["grace_ms"] = grace_ms
|
||||
return None
|
||||
|
||||
cli = self._cli(
|
||||
monkeypatch,
|
||||
listen=fake_listen,
|
||||
voice_cfg={
|
||||
"barge_in": True,
|
||||
"barge_in_threshold_multiplier": 4.5,
|
||||
"barge_in_grace_seconds": 1.0,
|
||||
},
|
||||
_agent_running=False,
|
||||
)
|
||||
cli._voice_full_duplex_listener()
|
||||
assert seen["multiplier"] == 4.5
|
||||
assert seen["grace_ms"] == 1000
|
||||
|
||||
def test_barge_in_disabled_never_opens_mic(self, monkeypatch):
|
||||
calls = []
|
||||
|
||||
def fake_listen(*a, **k):
|
||||
calls.append(True)
|
||||
return None
|
||||
|
||||
cli = self._cli(
|
||||
monkeypatch, listen=fake_listen,
|
||||
voice_cfg={"barge_in": False}, _agent_running=True,
|
||||
)
|
||||
cli._voice_full_duplex_listener()
|
||||
assert calls == []
|
||||
|
||||
def test_stop_phrase_mid_generation_interrupts_and_ends_chat(self, monkeypatch, tmp_path):
|
||||
"""Bare 'stop' during generation = stop everything: the turn is
|
||||
interrupted at trip time AND the voice chat is disabled."""
|
||||
wav = tmp_path / "fd.wav"
|
||||
wav.write_bytes(b"RIFF")
|
||||
|
||||
def fake_listen(should_stop, is_playing=None, on_trigger=None, **_kw):
|
||||
on_trigger("generation")
|
||||
return str(wav)
|
||||
|
||||
cli = self._cli(monkeypatch, listen=fake_listen, _agent_running=True)
|
||||
interrupted = threading.Event()
|
||||
cli.agent = SimpleNamespace(interrupt=lambda: interrupted.set())
|
||||
disabled = []
|
||||
cli._disable_voice_mode = lambda: disabled.append(True)
|
||||
monkeypatch.setattr(
|
||||
"tools.voice_mode.transcribe_recording",
|
||||
lambda path, model=None: {"success": True, "transcript": "stop"},
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"tools.voice_mode.is_voice_stop_phrase",
|
||||
lambda text: text.strip().lower() == "stop",
|
||||
)
|
||||
|
||||
cli._voice_full_duplex_listener()
|
||||
|
||||
assert interrupted.is_set() # turn interrupted at trip
|
||||
assert disabled == [True] # chat ended by the stop phrase
|
||||
assert cli._pending_input.empty() # stop phrase never reaches the agent
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Typed stop phrase — typing "stop" during a voice chat ends it
|
||||
# ============================================================================
|
||||
|
|
@ -1505,21 +1699,19 @@ class TestTypedVoiceStop:
|
|||
|
||||
|
||||
# ============================================================================
|
||||
# Fallback (whole-file) TTS path arms the spoken barge-in monitor
|
||||
# Fallback (whole-file) TTS path arms the full-duplex listener
|
||||
# ============================================================================
|
||||
|
||||
class TestFallbackSpeakArmsBargeMonitor:
|
||||
"""_voice_speak_response_async must arm _voice_barge_in_monitor in
|
||||
continuous voice mode. Previously ONLY the streaming pipeline armed the
|
||||
monitor (chat() gate), so when streaming TTS couldn't start the whole-file
|
||||
fallback speech was uninterruptible by voice — Teknium's "speaking over
|
||||
the agent does nothing" report on the non-streaming path."""
|
||||
"""_voice_speak_response_async must arm _voice_full_duplex_listener in
|
||||
continuous voice mode. This is the safety net for speak calls outside a
|
||||
chat turn — the primary arm happens at utterance-submit in chat()."""
|
||||
|
||||
def _cli(self, **overrides):
|
||||
cli = _make_voice_cli(**overrides)
|
||||
cli._monitor_calls = []
|
||||
cli._voice_barge_in_monitor = (
|
||||
lambda stop_event: cli._monitor_calls.append(stop_event)
|
||||
cli._voice_full_duplex_listener = (
|
||||
lambda: cli._monitor_calls.append(True)
|
||||
)
|
||||
cli._voice_speak_response = lambda text: None
|
||||
return cli
|
||||
|
|
@ -1533,7 +1725,6 @@ class TestFallbackSpeakArmsBargeMonitor:
|
|||
cli._voice_speak_response_async("a reply")
|
||||
self._drain_threads()
|
||||
assert len(cli._monitor_calls) == 1
|
||||
assert isinstance(cli._monitor_calls[0], threading.Event)
|
||||
|
||||
def test_no_monitor_outside_continuous_mode(self):
|
||||
cli = self._cli(_voice_mode=True, _voice_tts=True, _voice_continuous=False)
|
||||
|
|
|
|||
|
|
@ -2202,6 +2202,172 @@ class TestListenForSpeechCapture:
|
|||
assert listen_for_speech(lambda: False, capture=True) is None
|
||||
|
||||
|
||||
class TestFullDuplexListen:
|
||||
"""full_duplex_listen: one agent-turn listener spanning generation and
|
||||
playback — pre-playback calibration, phase-aware trigger, grace window."""
|
||||
|
||||
CALIB = 15 # 450ms / 30ms — pre-playback quiet calibration
|
||||
TRIP = 10 # 300ms / 30ms window; trip needs >=8 above
|
||||
GRACE = 16 # 500ms / 30ms
|
||||
BLOCK = 480 # 30ms at 16 kHz
|
||||
|
||||
def _run(self, mock_sd, monkeypatch, levels, playing_from=None,
|
||||
playing_until=None, should_stop=None, on_trigger=None, **kwargs):
|
||||
np = pytest.importorskip("numpy")
|
||||
stream = _FakeInputStream(np, levels)
|
||||
mock_sd.InputStream.return_value = stream
|
||||
written = {}
|
||||
monkeypatch.setattr(
|
||||
"tools.voice_mode.AudioRecorder._write_wav",
|
||||
staticmethod(lambda audio: written.update(audio=audio) or "/tmp/fd.wav"),
|
||||
)
|
||||
from tools.voice_mode import full_duplex_listen
|
||||
|
||||
def is_playing():
|
||||
if playing_from is None:
|
||||
return False
|
||||
if stream.reads < playing_from:
|
||||
return False
|
||||
if playing_until is not None and stream.reads >= playing_until:
|
||||
return False
|
||||
return True
|
||||
|
||||
stops = iter([False] * len(levels) + [True] * 10_000)
|
||||
path = full_duplex_listen(
|
||||
should_stop or (lambda: next(stops)),
|
||||
is_playing=is_playing,
|
||||
on_trigger=on_trigger,
|
||||
**kwargs,
|
||||
)
|
||||
return path, written.get("audio"), stream
|
||||
|
||||
def test_generation_phase_speech_trips_and_captures(self, mock_sd, monkeypatch):
|
||||
"""Speech while the LLM generates (no playback) trips with
|
||||
phase='generation' and the utterance is captured with pre-roll."""
|
||||
phases = []
|
||||
levels = [100] * self.CALIB + [5000] * 30 + [0] * 500
|
||||
path, audio, _ = self._run(
|
||||
mock_sd, monkeypatch, levels,
|
||||
on_trigger=lambda phase: phases.append(phase),
|
||||
)
|
||||
assert path == "/tmp/fd.wav"
|
||||
assert phases == ["generation"]
|
||||
assert int((audio == 5000).sum()) > 0 # speech onset in the capture
|
||||
|
||||
def test_playback_bleed_alone_does_not_trip(self, mock_sd, monkeypatch):
|
||||
"""Speaker bleed (~1000 RMS) during playback stays below the playback
|
||||
trigger clamp — the old monitor's self-calibration deafness class."""
|
||||
phases = []
|
||||
levels = [100] * self.CALIB + [1000] * 300
|
||||
path, _, _ = self._run(
|
||||
mock_sd, monkeypatch, levels,
|
||||
playing_from=self.CALIB,
|
||||
on_trigger=lambda phase: phases.append(phase),
|
||||
)
|
||||
assert path is None
|
||||
assert phases == []
|
||||
|
||||
def test_speech_over_bleed_trips_in_playback_phase(self, mock_sd, monkeypatch):
|
||||
"""Real speech (5000 RMS) over playback trips with phase='playback'
|
||||
even though playback bleed was present — the trigger comes from the
|
||||
PRE-playback quiet floor, never from bleed self-calibration."""
|
||||
phases = []
|
||||
# quiet calib → playback bleed (past grace) → user speaks over it
|
||||
levels = (
|
||||
[100] * self.CALIB
|
||||
+ [1000] * (self.GRACE + 20)
|
||||
+ [5000] * 30
|
||||
+ [1000] * 200
|
||||
)
|
||||
path, _, _ = self._run(
|
||||
mock_sd, monkeypatch, levels,
|
||||
playing_from=self.CALIB,
|
||||
on_trigger=lambda phase: phases.append(phase),
|
||||
)
|
||||
assert path == "/tmp/fd.wav"
|
||||
assert phases == ["playback"]
|
||||
|
||||
def test_quiet_floor_held_through_playback(self, mock_sd, monkeypatch):
|
||||
"""Loud playback bleed must NOT be absorbed into the floor: speech at
|
||||
3000 RMS still trips after minutes-equivalent of 1400-RMS bleed."""
|
||||
phases = []
|
||||
levels = (
|
||||
[100] * self.CALIB
|
||||
+ [1400] * (self.GRACE + 100) # sustained loud-ish bleed
|
||||
+ [3000] * 30
|
||||
+ [0] * 200
|
||||
)
|
||||
path, _, _ = self._run(
|
||||
mock_sd, monkeypatch, levels,
|
||||
playing_from=self.CALIB,
|
||||
on_trigger=lambda phase: phases.append(phase),
|
||||
)
|
||||
assert path == "/tmp/fd.wav"
|
||||
assert phases == ["playback"]
|
||||
|
||||
def test_grace_window_suppresses_playback_onset(self, mock_sd, monkeypatch):
|
||||
"""Loud blocks inside the 0.5s grace right after playback starts are
|
||||
suppressed (onset transient), but speech after grace still trips."""
|
||||
phases = []
|
||||
# loud transient fully inside grace, then quiet bleed, then speech
|
||||
levels = (
|
||||
[100] * self.CALIB
|
||||
+ [5000] * 8 # onset transient (inside 16-block grace)
|
||||
+ [800] * 40
|
||||
+ [5000] * 30
|
||||
+ [0] * 200
|
||||
)
|
||||
|
||||
def trig(phase):
|
||||
phases.append(phase)
|
||||
|
||||
path, _, stream = self._run(
|
||||
mock_sd, monkeypatch, levels,
|
||||
playing_from=self.CALIB,
|
||||
on_trigger=trig,
|
||||
)
|
||||
assert path == "/tmp/fd.wav"
|
||||
assert phases == ["playback"]
|
||||
# Trip must come from the post-grace speech, not the onset transient:
|
||||
# by the time capture starts, we're past calib+transient+bleed blocks.
|
||||
assert stream.reads > self.CALIB + 8 + 40
|
||||
|
||||
def test_generation_trigger_uses_multiplier_over_quiet_floor(self, mock_sd, monkeypatch):
|
||||
"""Threshold math: floor≈300, multiplier 3 → trigger 900. 1200-RMS
|
||||
speech trips; with multiplier 8 the trigger is 2400 and it doesn't."""
|
||||
levels = [300] * self.CALIB + [1200] * 40 + [0] * 400
|
||||
path3, _, _ = self._run(mock_sd, monkeypatch, levels, multiplier=3.0)
|
||||
assert path3 == "/tmp/fd.wav"
|
||||
path8, _, _ = self._run(mock_sd, monkeypatch, levels, multiplier=8.0)
|
||||
assert path8 is None
|
||||
|
||||
def test_windowed_majority_tolerates_intra_word_dips(self, mock_sd, monkeypatch):
|
||||
"""A brief energy dip inside a word must not reset detection — the
|
||||
old strictly-consecutive counter failed exactly this way."""
|
||||
speech = ([5000] * 4 + [300]) * 8 # 80% above trigger, dips inside
|
||||
levels = [100] * self.CALIB + speech + [0] * 400
|
||||
path, _, _ = self._run(mock_sd, monkeypatch, levels)
|
||||
assert path == "/tmp/fd.wav"
|
||||
|
||||
def test_brief_spike_does_not_trip(self, mock_sd, monkeypatch):
|
||||
levels = [100] * self.CALIB + [5000] * 3 + [100] * 500
|
||||
path, _, _ = self._run(mock_sd, monkeypatch, levels)
|
||||
assert path is None
|
||||
|
||||
def test_should_stop_ends_without_trip(self, mock_sd, monkeypatch):
|
||||
path, audio, _ = self._run(mock_sd, monkeypatch, [100] * 60)
|
||||
assert path is None
|
||||
assert audio is None
|
||||
|
||||
def test_returns_none_when_audio_unavailable(self, monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
"tools.voice_mode._import_audio",
|
||||
MagicMock(side_effect=OSError("no audio")),
|
||||
)
|
||||
from tools.voice_mode import full_duplex_listen
|
||||
assert full_duplex_listen(lambda: False) is None
|
||||
|
||||
|
||||
class TestGetBeepVolume:
|
||||
"""Issue #55908: beep amplitude must come from config.yaml, with safe fallback."""
|
||||
|
||||
|
|
|
|||
|
|
@ -1891,6 +1891,245 @@ def listen_for_speech(
|
|||
return None if capture else False
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Full-duplex agent-turn listener
|
||||
# ============================================================================
|
||||
#
|
||||
# One listener for the WHOLE agent turn in continuous voice mode: armed the
|
||||
# moment an utterance is submitted, disarmed when the turn is fully done
|
||||
# (response + TTS finished). It replaces the scattered per-playback barge
|
||||
# monitors, which had two class-level failures:
|
||||
#
|
||||
# 1. HALF-DUPLEX GAP: the monitor only spawned when TTS playback started,
|
||||
# so during LLM generation (seconds to minutes) there was NO microphone
|
||||
# listener at all — the user could not interject by voice.
|
||||
# 2. PLAYBACK DEAFNESS: the monitor calibrated its noise floor WHILE the
|
||||
# speaker was already blasting TTS, baking speaker bleed into the floor;
|
||||
# with an 8x multiplier the trigger became unreachable for normal speech,
|
||||
# and requiring a full second of strictly CONSECUTIVE 30ms blocks above
|
||||
# trigger meant any intra-word dip reset the counter.
|
||||
#
|
||||
# This listener calibrates against the QUIET room at turn start (before any
|
||||
# playback exists), freezes that baseline through playback (never calibrating
|
||||
# against its own speaker bleed), and trips on a windowed majority of blocks
|
||||
# instead of a strict consecutive run.
|
||||
|
||||
# Minimum trigger while TTS audio is flowing. Speaker bleed reaching the mic
|
||||
# through air at conversational volume is typically well under this (a few
|
||||
# hundred RMS at arm's length; ~1000-1400 with loud speakers close to the
|
||||
# mic), while direct speech at normal distance measures 3000-8000 RMS.
|
||||
PLAYBACK_MIN_TRIGGER = 1500.0
|
||||
|
||||
# Absolute trigger ceiling — a noisy room must never push the trigger above
|
||||
# what normal speech (3000-8000 RMS) can reach.
|
||||
TRIGGER_CEILING = 4000.0
|
||||
|
||||
# Default trigger multiplier over the quiet-room floor. The old 8x default
|
||||
# only made sense when the floor was (wrongly) calibrated against active TTS
|
||||
# bleed; against a genuine quiet-room baseline (typically 50-300 RMS) the
|
||||
# synthetic-frame tests show 3x separates speech from ambient cleanly while
|
||||
# staying reachable (300 RMS floor * 3 = 900 trigger vs 3000+ RMS speech).
|
||||
DEFAULT_BARGE_MULTIPLIER = 3.0
|
||||
|
||||
|
||||
def _voice_debug_enabled() -> bool:
|
||||
return os.environ.get("HERMES_VOICE_DEBUG", "").strip() == "1"
|
||||
|
||||
|
||||
def _vad_log(msg: str) -> None:
|
||||
"""VAD decision-point diagnostic — always logger.debug, plus stderr when
|
||||
HERMES_VOICE_DEBUG=1 so live hardware tuning doesn't need a log tail."""
|
||||
logger.debug(msg)
|
||||
if _voice_debug_enabled():
|
||||
try:
|
||||
print(f"[voice-vad] {msg}", file=sys.stderr, flush=True)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def full_duplex_listen(
|
||||
should_stop: Callable[[], bool],
|
||||
is_playing: Optional[Callable[[], bool]] = None,
|
||||
on_trigger: Optional[Callable[[str], None]] = None,
|
||||
multiplier: Optional[float] = None,
|
||||
sustained_ms: int = 300,
|
||||
calibration_ms: int = 450,
|
||||
grace_ms: int = 500,
|
||||
pre_roll_ms: int = 1200,
|
||||
endpoint_silence_ms: int = 1250,
|
||||
max_utterance_ms: int = 30_000,
|
||||
) -> Optional[str]:
|
||||
"""Listen across an ENTIRE agent turn; return the captured interruption.
|
||||
|
||||
Runs from utterance-submit to turn-complete. Two phases, decided per
|
||||
30ms block by *is_playing* (usually ``is_audio_output_active``):
|
||||
|
||||
* ``generation`` — no TTS audio flowing. The room is quiet; the first
|
||||
*calibration_ms* establish the noise floor (pre-playback calibration).
|
||||
Trigger = quiet_floor x *multiplier* (clamped to a sane minimum), i.e.
|
||||
ordinary speech detection.
|
||||
* ``playback`` — TTS audio flowing. The quiet baseline is HELD (never
|
||||
recalibrated against speaker bleed); the trigger is additionally
|
||||
clamped up to ``PLAYBACK_MIN_TRIGGER`` so bleed alone can't trip it,
|
||||
and a *grace_ms* window after playback first starts suppresses trips
|
||||
from the playback onset transient.
|
||||
|
||||
Detection is a windowed majority — >=80% of the last *sustained_ms*
|
||||
worth of blocks above trigger (with the current block above) — so
|
||||
intra-word energy dips don't reset progress the way the old strictly-
|
||||
consecutive counter did.
|
||||
|
||||
On detection ``on_trigger(phase)`` fires (cut TTS / interrupt the turn),
|
||||
then capture continues from the rolling *pre_roll_ms* buffer until
|
||||
*endpoint_silence_ms* of quiet, and the WAV path is returned. Returns
|
||||
``None`` when *should_stop* ends the turn without speech.
|
||||
"""
|
||||
try:
|
||||
sd, np = _import_audio()
|
||||
except (ImportError, OSError):
|
||||
return None
|
||||
|
||||
from collections import deque
|
||||
|
||||
block = int(SAMPLE_RATE * 0.03) # 30ms blocks
|
||||
calib_blocks = max(1, calibration_ms // 30)
|
||||
trip_blocks = max(1, sustained_ms // 30)
|
||||
trip_needed = max(1, int(round(trip_blocks * 0.8)))
|
||||
grace_blocks = max(0, grace_ms // 30)
|
||||
endpoint_blocks = max(1, endpoint_silence_ms // 30)
|
||||
max_blocks = max(1, max_utterance_ms // 30)
|
||||
mult = float(multiplier) if multiplier else DEFAULT_BARGE_MULTIPLIER
|
||||
|
||||
ambient: "deque[float]" = deque(maxlen=100) # ~3s of quiet-phase RMS
|
||||
pre_roll: deque = deque(maxlen=max(1, pre_roll_ms // 30))
|
||||
recent_above: "deque[bool]" = deque(maxlen=trip_blocks)
|
||||
quiet_floor = float(SILENCE_RMS_THRESHOLD)
|
||||
floor_locked = False
|
||||
playing_prev = False
|
||||
playback_seen = False
|
||||
grace_remaining = 0
|
||||
blocks_since_playback = 10_000
|
||||
block_idx = 0
|
||||
|
||||
try:
|
||||
with sd.InputStream(
|
||||
samplerate=SAMPLE_RATE, channels=1, dtype="int16", blocksize=block
|
||||
) as stream:
|
||||
while not should_stop():
|
||||
data, _ = stream.read(block)
|
||||
rms = float(np.sqrt(np.mean(data.astype(np.float64) ** 2)))
|
||||
pre_roll.append(data.copy())
|
||||
block_idx += 1
|
||||
|
||||
playing = bool(is_playing()) if is_playing is not None else False
|
||||
|
||||
# Pre-playback calibration: the listener arms at utterance
|
||||
# submit, before any TTS exists, so the first calib_blocks
|
||||
# sample the actual quiet room — NOT speaker bleed.
|
||||
if not floor_locked:
|
||||
if not playing:
|
||||
ambient.append(rms)
|
||||
if len(ambient) >= calib_blocks or playing:
|
||||
pct90 = (
|
||||
float(np.percentile(list(ambient), 90))
|
||||
if ambient
|
||||
else float(SILENCE_RMS_THRESHOLD)
|
||||
)
|
||||
quiet_floor = max(pct90, float(SILENCE_RMS_THRESHOLD))
|
||||
floor_locked = True
|
||||
_vad_log(
|
||||
f"calibrated quiet floor={quiet_floor:.0f} "
|
||||
f"(pct90={pct90:.0f}, {len(ambient)} blocks, "
|
||||
f"mult={mult:g})"
|
||||
)
|
||||
if not floor_locked:
|
||||
continue
|
||||
|
||||
# Playback phase transitions: grace only when playback starts
|
||||
# after a real gap (>=1s), so inter-sentence flapping of the
|
||||
# audio-active flag can't chain grace windows together and
|
||||
# swallow a genuine interjection.
|
||||
if playing and not playing_prev:
|
||||
if not playback_seen or blocks_since_playback > 33:
|
||||
grace_remaining = grace_blocks
|
||||
_vad_log(
|
||||
f"playback started (block={block_idx}) — "
|
||||
f"grace {grace_blocks * 30}ms"
|
||||
)
|
||||
playback_seen = True
|
||||
playing_prev = playing
|
||||
blocks_since_playback = 0 if playing else blocks_since_playback + 1
|
||||
|
||||
# Trigger: quiet baseline x multiplier, phase-clamped.
|
||||
trigger = quiet_floor * mult
|
||||
if playing:
|
||||
trigger = max(trigger, PLAYBACK_MIN_TRIGGER)
|
||||
else:
|
||||
trigger = max(trigger, float(SILENCE_RMS_THRESHOLD) * 2)
|
||||
trigger = min(trigger, TRIGGER_CEILING)
|
||||
|
||||
# Keep the quiet floor current with ambient drift — but ONLY
|
||||
# while nothing is playing (never absorb speaker bleed) and
|
||||
# the block isn't speech.
|
||||
if not playing and rms < trigger:
|
||||
ambient.append(rms)
|
||||
if ambient:
|
||||
quiet_floor = max(
|
||||
float(np.percentile(list(ambient), 90)),
|
||||
float(SILENCE_RMS_THRESHOLD),
|
||||
)
|
||||
|
||||
above = rms >= trigger
|
||||
if above and grace_remaining > 0:
|
||||
_vad_log(
|
||||
f"grace suppression: block={block_idx} rms={rms:.0f} "
|
||||
f"trigger={trigger:.0f} ({grace_remaining} blocks left)"
|
||||
)
|
||||
above = False
|
||||
if grace_remaining > 0:
|
||||
grace_remaining -= 1
|
||||
|
||||
recent_above.append(above)
|
||||
if rms >= trigger * 0.5:
|
||||
_vad_log(
|
||||
f"block={block_idx} rms={rms:.0f} floor={quiet_floor:.0f} "
|
||||
f"trigger={trigger:.0f} above={above} "
|
||||
f"window={sum(recent_above)}/{trip_needed} "
|
||||
f"phase={'playback' if playing else 'generation'}"
|
||||
)
|
||||
|
||||
if not (above and sum(recent_above) >= trip_needed):
|
||||
continue
|
||||
|
||||
phase = "playback" if playing else "generation"
|
||||
_vad_log(
|
||||
f"TRIPPED ({phase}): block={block_idx} rms={rms:.0f} "
|
||||
f"floor={quiet_floor:.0f} trigger={trigger:.0f} "
|
||||
f"window={sum(recent_above)}/{len(recent_above)}"
|
||||
)
|
||||
if on_trigger:
|
||||
try:
|
||||
on_trigger(phase)
|
||||
except Exception as e:
|
||||
logger.debug("full-duplex trigger callback failed: %s", e)
|
||||
|
||||
# Capture until the user goes quiet. Playback was cut by
|
||||
# on_trigger, so plain silence endpointing works.
|
||||
frames: List[Any] = list(pre_roll)
|
||||
quiet = 0
|
||||
for _ in range(max_blocks):
|
||||
data, _ = stream.read(block)
|
||||
frames.append(data.copy())
|
||||
rms = float(np.sqrt(np.mean(data.astype(np.float64) ** 2)))
|
||||
quiet = quiet + 1 if rms < SILENCE_RMS_THRESHOLD else 0
|
||||
if quiet >= endpoint_blocks:
|
||||
break
|
||||
return AudioRecorder._write_wav(np.concatenate(frames, axis=0))
|
||||
except Exception as e:
|
||||
logger.debug("Full-duplex listener failed: %s", e)
|
||||
return None
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Requirements check
|
||||
# ============================================================================
|
||||
|
|
|
|||
|
|
@ -12144,6 +12144,13 @@ def _run_prompt_submit(
|
|||
# consume the latch below.
|
||||
tts_queue = _tts_stream_begin()
|
||||
|
||||
# Full-duplex agent-turn listener: armed at utterance-submit so
|
||||
# the user can interject DURING generation, not just during
|
||||
# playback. _tts_stream_begin arms it too when a pipeline
|
||||
# starts; this covers voice mode without working TTS.
|
||||
if _voice_mode_enabled() and _voice_cfg_dict().get("barge_in", True):
|
||||
_arm_full_duplex_listener()
|
||||
|
||||
# Ambient "thinking" sound (voice mode only): calm bubble blips
|
||||
# while the agent works with no audio flowing, so long
|
||||
# thinking/tool stretches don't read as a dead session. Per-blip
|
||||
|
|
@ -17792,9 +17799,7 @@ def _tts_stream_begin() -> Optional[queue.Queue]:
|
|||
_tts_stream_state = {"stop": stop, "done": done}
|
||||
|
||||
if _voice_mode_enabled() and _voice_cfg_dict().get("barge_in", True):
|
||||
threading.Thread(
|
||||
target=_tts_stream_barge_in_monitor, args=(stop, done), daemon=True
|
||||
).start()
|
||||
_arm_full_duplex_listener()
|
||||
|
||||
return text_queue
|
||||
|
||||
|
|
@ -17831,54 +17836,146 @@ def _tts_stream_stop(user_barge: bool = True) -> None:
|
|||
|
||||
|
||||
def _tts_stream_barge_in_monitor(stop: threading.Event, done: threading.Event) -> None:
|
||||
"""VAD barge-in: cut streaming TTS when the user starts talking.
|
||||
"""Deprecated shim — playback-only monitor replaced by the full-duplex
|
||||
agent-turn listener (see ``_full_duplex_listener``). Kept as a name so
|
||||
stray callers arm the new listener instead of a per-playback mic."""
|
||||
_arm_full_duplex_listener()
|
||||
|
||||
Playback is cut at the moment of detection while the monitor keeps
|
||||
capturing (with pre-roll) until the user goes quiet — the interruption is
|
||||
then transcribed and emitted as ``voice.transcript``, which the TUI
|
||||
submits like any spoken turn. Without capture the opening words would be
|
||||
lost between detection and the next recording start.
|
||||
|
||||
# ── Full-duplex agent-turn listener (one mic, whole turn) ──────────────────
|
||||
# Replaces the per-playback barge monitors: those only opened the mic once
|
||||
# TTS playback started (deaf during LLM generation) and calibrated the VAD
|
||||
# floor against active speaker bleed (deaf during playback too, in practice).
|
||||
# This listener arms at utterance-submit, spans generation AND playback, and
|
||||
# disarms when no session is running, no TTS is pending, and no audio flows.
|
||||
|
||||
_fd_listener_lock = threading.Lock()
|
||||
_fd_listener_active = False
|
||||
# (stop, done) pairs for fallback whole-reply speak paths currently active —
|
||||
# the listener must cut THEIR private stop events too, and must keep
|
||||
# listening while any of them is still speaking.
|
||||
_fd_speak_pipelines: "set[tuple[threading.Event, threading.Event]]" = set()
|
||||
|
||||
|
||||
def _arm_full_duplex_listener() -> None:
|
||||
"""Arm the process-global full-duplex listener (idempotent — one mic)."""
|
||||
global _fd_listener_active
|
||||
with _fd_listener_lock:
|
||||
if _fd_listener_active:
|
||||
return
|
||||
_fd_listener_active = True
|
||||
threading.Thread(
|
||||
target=_full_duplex_listener, daemon=True, name="voice-full-duplex"
|
||||
).start()
|
||||
|
||||
|
||||
def _fd_tts_pending() -> bool:
|
||||
"""True while any TTS (streaming pipeline or fallback speak) is unfinished."""
|
||||
with _tts_stream_lock:
|
||||
state = _tts_stream_state
|
||||
if state is not None and not state["done"].is_set():
|
||||
return True
|
||||
with _fd_listener_lock:
|
||||
pipelines = list(_fd_speak_pipelines)
|
||||
return any(not done.is_set() for _stop, done in pipelines)
|
||||
|
||||
|
||||
def _full_duplex_listener() -> None:
|
||||
"""Mic live from utterance-submit to turn-complete; phase-aware trip.
|
||||
|
||||
* generation phase (no TTS audio flowing): user speech interrupts every
|
||||
running session's agent turn — the same ``agent.interrupt()`` seam
|
||||
``session.interrupt`` uses — and cuts any pending TTS pipeline so the
|
||||
stale reply never plays. The captured utterance is transcribed and
|
||||
emitted as ``voice.transcript`` (the TUI submits it as the next turn).
|
||||
* playback phase: cuts TTS (streaming pipeline + fallback speak paths +
|
||||
file player) and submits the captured interruption.
|
||||
|
||||
Stop phrase is honored in both phases: mid-generation it interrupts the
|
||||
turn AND ends the voice chat ("stop everything").
|
||||
"""
|
||||
global _fd_listener_active
|
||||
try:
|
||||
from tools.tts_streaming import mark_speech_interrupted
|
||||
from tools.voice_mode import listen_for_speech, stop_playback, transcribe_recording
|
||||
|
||||
# Grace period: wait briefly before opening the mic so the
|
||||
# first TTS sentence is already playing and the VAD calibration
|
||||
# samples the actual playback level (not silence). This
|
||||
# prevents speaker bleed from falsely triggering barge-in
|
||||
# at the start of playback. Mirrors the CLI path in cli.py
|
||||
# _voice_barge_in_monitor.
|
||||
_grace_s = float(_voice_cfg_dict().get("barge_in_grace_seconds", 2.0))
|
||||
if _grace_s > 0:
|
||||
stop.wait(timeout=_grace_s)
|
||||
if stop.is_set() or done.is_set():
|
||||
return
|
||||
|
||||
barged = threading.Event()
|
||||
|
||||
def _cut_playback():
|
||||
if not done.is_set():
|
||||
import traceback as _tb
|
||||
logger.debug(
|
||||
"TTS CUT: gateway barge-in _cut_playback fired (VAD trip) — "
|
||||
"stop.set() + stop_playback()\n%s",
|
||||
"".join(_tb.format_stack()),
|
||||
)
|
||||
barged.set()
|
||||
mark_speech_interrupted()
|
||||
stop.set()
|
||||
stop_playback()
|
||||
_voice_emit("voice.interrupted")
|
||||
|
||||
wav_path = listen_for_speech(
|
||||
lambda: stop.is_set() or done.is_set(),
|
||||
capture=True,
|
||||
on_trigger=_cut_playback,
|
||||
sustained_ms=1000,
|
||||
calibration_ms=800,
|
||||
from tools.voice_mode import (
|
||||
full_duplex_listen,
|
||||
is_audio_output_active,
|
||||
stop_playback,
|
||||
transcribe_recording,
|
||||
)
|
||||
if not (wav_path and barged.is_set()):
|
||||
|
||||
cfg = _voice_cfg_dict()
|
||||
try:
|
||||
_mult = float(cfg.get("barge_in_threshold_multiplier", 0) or 0)
|
||||
except (TypeError, ValueError):
|
||||
_mult = 0.0
|
||||
try:
|
||||
_grace_ms = int(float(cfg.get("barge_in_grace_seconds", 0.5)) * 1000)
|
||||
except (TypeError, ValueError):
|
||||
_grace_ms = 500
|
||||
|
||||
def _should_stop() -> bool:
|
||||
if not _voice_mode_enabled():
|
||||
return True
|
||||
if _any_session_running():
|
||||
return False
|
||||
if _fd_tts_pending():
|
||||
return False
|
||||
return not is_audio_output_active()
|
||||
|
||||
tripped = threading.Event()
|
||||
|
||||
def _cut_all_tts() -> None:
|
||||
# Streaming pipeline (private stop event + player).
|
||||
_tts_stream_stop(user_barge=True)
|
||||
# Fallback whole-reply speak paths (their own stop events).
|
||||
with _fd_listener_lock:
|
||||
pipelines = list(_fd_speak_pipelines)
|
||||
for _stop, _done in pipelines:
|
||||
_stop.set()
|
||||
stop_playback()
|
||||
|
||||
def _on_trigger(phase: str) -> None:
|
||||
tripped.set()
|
||||
mark_speech_interrupted()
|
||||
if phase == "playback":
|
||||
logger.debug(
|
||||
"TTS CUT: full-duplex listener tripped during playback"
|
||||
)
|
||||
_cut_all_tts()
|
||||
else:
|
||||
logger.debug(
|
||||
"full-duplex listener tripped during generation — "
|
||||
"interrupting running turn(s)"
|
||||
)
|
||||
# Cut pending TTS FIRST so the stale reply can never speak.
|
||||
_cut_all_tts()
|
||||
# Interrupt every running session's turn — voice is
|
||||
# process-global, and the same seam session.interrupt uses.
|
||||
try:
|
||||
with _sessions_lock:
|
||||
running = [
|
||||
s for s in _sessions.values() if s.get("running")
|
||||
]
|
||||
for s in running:
|
||||
agent = s.get("agent")
|
||||
if agent is not None and hasattr(agent, "interrupt"):
|
||||
try:
|
||||
agent.interrupt()
|
||||
except Exception:
|
||||
pass
|
||||
except Exception as e:
|
||||
logger.debug("voice interjection interrupt failed: %s", e)
|
||||
_voice_emit("voice.interrupted")
|
||||
|
||||
wav_path = full_duplex_listen(
|
||||
_should_stop,
|
||||
is_playing=is_audio_output_active,
|
||||
on_trigger=_on_trigger,
|
||||
multiplier=_mult or None,
|
||||
grace_ms=max(0, _grace_ms),
|
||||
)
|
||||
if not (wav_path and tripped.is_set()):
|
||||
return
|
||||
try:
|
||||
result = transcribe_recording(wav_path)
|
||||
|
|
@ -17894,10 +17991,9 @@ def _tts_stream_barge_in_monitor(stop: threading.Event, done: threading.Event) -
|
|||
_is_stop = False
|
||||
|
||||
if _is_stop:
|
||||
# Barge-in with a bare stop phrase — the user talked over
|
||||
# the agent's speech to END the voice chat. Same explicit
|
||||
# stop signal as the continuous loop: flip mode off, halt
|
||||
# any active capture, and tell clients it was user intent.
|
||||
# Bare stop phrase — in EITHER phase the user means
|
||||
# "stop everything": the turn was already interrupted /
|
||||
# TTS cut at trip time; now end the voice chat.
|
||||
os.environ["HERMES_VOICE"] = "0"
|
||||
os.environ["HERMES_VOICE_TTS"] = "0"
|
||||
try:
|
||||
|
|
@ -17915,25 +18011,28 @@ def _tts_stream_barge_in_monitor(stop: threading.Event, done: threading.Event) -
|
|||
except OSError:
|
||||
pass
|
||||
except Exception as e:
|
||||
logger.debug("TTS barge-in monitor failed: %s", e)
|
||||
logger.debug("full-duplex listener failed: %s", e)
|
||||
finally:
|
||||
with _fd_listener_lock:
|
||||
_fd_listener_active = False
|
||||
|
||||
|
||||
def _speak_text_with_barge(text: str) -> None:
|
||||
"""Speak *text* via hermes_cli.voice.speak_text with spoken barge-in.
|
||||
|
||||
The streaming-TTS turn pipeline arms ``_tts_stream_barge_in_monitor``;
|
||||
the fallback whole-reply path (streaming couldn't start) and the
|
||||
The fallback whole-reply path (streaming couldn't start) and the
|
||||
``voice.tts`` RPC previously called ``speak_text`` bare — speech over
|
||||
those paths was UNINTERRUPTIBLE by voice. Run the same monitor beside
|
||||
the speak thread: it cuts playback (``stop_playback`` kills the file
|
||||
player; the stop event drains a streaming dispatch inside speak_text),
|
||||
captures the interruption, and emits ``voice.transcript`` /
|
||||
the stop-phrase signal exactly like the streaming path.
|
||||
those paths was UNINTERRUPTIBLE by voice. The full-duplex agent-turn
|
||||
listener covers this path too: the (stop, done) pair is registered in
|
||||
``_fd_speak_pipelines`` so the listener can cut the private stop event
|
||||
on a playback trip and keeps listening while this speak is pending.
|
||||
"""
|
||||
from hermes_cli.voice import speak_text
|
||||
|
||||
stop = threading.Event()
|
||||
done = threading.Event()
|
||||
with _fd_listener_lock:
|
||||
_fd_speak_pipelines.add((stop, done))
|
||||
|
||||
def _speak():
|
||||
try:
|
||||
|
|
@ -17943,12 +18042,12 @@ def _speak_text_with_barge(text: str) -> None:
|
|||
speak_text(text)
|
||||
finally:
|
||||
done.set()
|
||||
with _fd_listener_lock:
|
||||
_fd_speak_pipelines.discard((stop, done))
|
||||
|
||||
threading.Thread(target=_speak, daemon=True).start()
|
||||
if _voice_mode_enabled() and _voice_cfg_dict().get("barge_in", True):
|
||||
threading.Thread(
|
||||
target=_tts_stream_barge_in_monitor, args=(stop, done), daemon=True
|
||||
).start()
|
||||
_arm_full_duplex_listener()
|
||||
|
||||
|
||||
def _voice_cfg_dict() -> dict:
|
||||
|
|
|
|||
|
|
@ -177,10 +177,14 @@ The same pipeline runs in the classic CLI, the TUI, and the desktop app. In a de
|
|||
|
||||
### Barge-in
|
||||
|
||||
You can interrupt the agent mid-speech:
|
||||
You can interrupt the agent at ANY point in its turn — the microphone stays live from the moment you finish speaking until the reply has fully played (full duplex):
|
||||
|
||||
- **Talk over it** — in continuous voice mode, a voice-activity monitor listens while the agent speaks and cuts playback the moment you start talking, then goes straight back to recording. The detector calibrates its noise floor against the playback itself, so speaker bleed doesn't self-trigger. Disable with `voice.barge_in: false` in `config.yaml`.
|
||||
- **Interject while it's thinking** — in continuous voice mode, speaking during LLM generation (before any audio plays) interrupts the in-flight turn and your interjection becomes the next message, the same as typing over a running turn.
|
||||
- **Talk over it** — speaking while the agent's reply plays cuts playback the moment you start talking and submits what you said. The detector calibrates its noise floor against the *quiet room* at turn start (never against the playback itself), so speaker bleed can't deafen it and normal speech reliably trips it.
|
||||
- **Type or press the record key** — sending a new message or hitting the push-to-talk key stops playback instantly on every surface.
|
||||
- **Say "stop"** — the stop phrase works in both phases: mid-generation it interrupts the turn AND ends the voice chat; mid-playback it cuts the speech and ends the chat.
|
||||
|
||||
Tuning (config.yaml): `voice.barge_in: false` disables it; `voice.barge_in_threshold_multiplier` (default `3.0`) scales the speech trigger over the quiet-room floor; `voice.barge_in_grace_seconds` (default `0.5`) suppresses trips right after playback starts. Set `HERMES_VOICE_DEBUG=1` to stream per-block VAD diagnostics (calibrated floor, RMS, trip decisions) to stderr for live tuning.
|
||||
|
||||
The agent **knows** it was interrupted: the next message carries a short note telling the model its spoken reply was cut off, so it can react naturally ("rude!") or pick up where it left off instead of being oblivious.
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue