From ad3eaaa39c89101c297e357cd0d4432c2f542d8c Mon Sep 17 00:00:00 2001 From: Satvik Sethia Date: Sun, 14 Jun 2026 19:36:00 -0400 Subject: [PATCH] live voice, no barge in yet --- voice.py | 175 ++++++++++++++++++++++++++++++++------------------ voice_live.py | 139 +++++++++++++++++++++++++++++---------- 2 files changed, 219 insertions(+), 95 deletions(-) diff --git a/voice.py b/voice.py index 0516fbf..70bc0ca 100644 --- a/voice.py +++ b/voice.py @@ -20,7 +20,10 @@ import wave import queue import shutil +import threading +import traceback import subprocess +from collections import deque import numpy as np from dotenv import load_dotenv @@ -33,7 +36,8 @@ SAMPLE_RATE = 16000 # Whisper's native rate BLOCK = 480 # 30 ms frames -SILENCE_HANG = 0.8 # seconds of trailing quiet that ends a turn +SILENCE_HANG = float(os.getenv("ARIA_VOICE_HANG", "1.2")) # trailing quiet that ends a turn (s) +MIN_SPEECH = float(os.getenv("ARIA_VOICE_MIN_SPEECH", "1.0")) # min voiced audio before a turn can end (s) MAX_TURN = 30 # hard cap on one utterance (seconds) EXIT_WORDS = {"exit", "quit", "goodbye", "bye", "stop"} @@ -80,50 +84,71 @@ def speak(text: str): # else: no TTS backend — the text is already printed, so we just stay silent. -def record_until_silence(sd) -> np.ndarray: - """Wait for speech to start, then record until ~SILENCE_HANG of quiet.""" - q: queue.Queue = queue.Queue() - - def cb(indata, frames, t, status): - q.put(indata[:, 0].copy()) - - with sd.InputStream(samplerate=SAMPLE_RATE, channels=1, dtype="float32", - blocksize=BLOCK, callback=cb): - ambient = [_rms(q.get()) for _ in range(int(0.4 * SAMPLE_RATE / BLOCK))] - floor = max(np.mean(ambient) * 3.0, 0.015) # adaptive threshold - - frames, speaking, silent_for, elapsed = [], False, 0.0, 0.0 - per_block = BLOCK / SAMPLE_RATE +def _drain(q: queue.Queue): + """Discard everything captured so far — called before each turn so audio the mic + heard while Aria was speaking (her own voice) doesn't poison the next utterance.""" + try: while True: - block = q.get() - level = _rms(block) - if not speaking: - if level > floor: - speaking = True - frames.append(block) - continue - frames.append(block) - elapsed += per_block - silent_for = silent_for + per_block if level < floor else 0.0 - if silent_for >= SILENCE_HANG or elapsed >= MAX_TURN: - break + q.get_nowait() + except queue.Empty: + pass + + +def listen_hands_free(q: queue.Queue) -> np.ndarray: + """Read from the live mic queue: wait for speech, then record until a sustained pause. + A turn ends only after MIN_SPEECH of voiced audio AND SILENCE_HANG of trailing quiet, + so it won't cut off a couple words in. Returns empty if no speech for ~20 s.""" + per_block = BLOCK / SAMPLE_RATE + ambient = [_rms(q.get()) for _ in range(int(0.4 * SAMPLE_RATE / BLOCK))] + # Sensitive enough not to clip soft speech, capped so a noisy calibration can't + # make us deaf and end turns early. + floor = min(max(np.mean(ambient) * 1.8, 0.01), 0.05) + + preroll = deque(maxlen=6) # ~0.18 s kept before onset so word 1 isn't clipped + recent = deque(maxlen=3) # smooth ~90 ms so mid-word dips don't read as silence + frames, speaking = [], False + silent_for, voiced_for, elapsed, waited = 0.0, 0.0, 0.0, 0.0 + while True: + try: + block = q.get(timeout=1.0) + except queue.Empty: + return np.zeros(0, dtype="float32") # stream stalled — let the loop re-prompt + recent.append(_rms(block)) + level = max(recent) + if not speaking: + preroll.append(block) + waited += per_block + if level > floor: + speaking = True + frames.extend(preroll) + elif waited >= 20.0: + return np.zeros(0, dtype="float32") + continue + frames.append(block) + elapsed += per_block + if level >= floor: + voiced_for += per_block + silent_for = 0.0 + else: + silent_for += per_block + if (silent_for >= SILENCE_HANG and voiced_for >= MIN_SPEECH) or elapsed >= MAX_TURN: + break return np.concatenate(frames) if frames else np.zeros(0, dtype="float32") -def record_ptt(sd) -> np.ndarray: - """Push-to-talk: record between two Enter presses.""" +def listen_ptt(q: queue.Queue) -> np.ndarray: + """Push-to-talk over the live mic queue: capture between two Enter presses.""" input(" [Enter] to start…") - q: queue.Queue = queue.Queue() - - def cb(indata, frames, t, status): - q.put(indata[:, 0].copy()) - - with sd.InputStream(samplerate=SAMPLE_RATE, channels=1, dtype="float32", - blocksize=BLOCK, callback=cb): - input(" 🔴 recording… [Enter] to stop") + _drain(q) + stop = threading.Event() + threading.Thread(target=lambda: (input(" 🔴 recording… [Enter] to stop"), + stop.set()), daemon=True).start() frames = [] - while not q.empty(): - frames.append(q.get()) + while not stop.is_set(): + try: + frames.append(q.get(timeout=0.1)) + except queue.Empty: + pass return np.concatenate(frames) if frames else np.zeros(0, dtype="float32") @@ -142,32 +167,60 @@ def main(): print("Initializing Aria…") agent = build_agent(checkpointer=open_checkpointer()) thread_id = "local-voice" + + # One persistent mic stream for the whole session — reopening per turn is flaky on + # macOS and drops audio. We flush it (_drain) before each turn instead. + q: queue.Queue = queue.Queue() + + def cb(indata, frames, t, status): + q.put(indata[:, 0].copy()) + + stream = sd.InputStream(samplerate=SAMPLE_RATE, channels=1, dtype="float32", + blocksize=BLOCK, callback=cb) + stream.start() print("🎙️ Voice mode. " + ("Press Enter to talk." if ptt else "Just start talking.") - + " Say 'exit' or hit Ctrl-C to quit.") + + " Say 'exit' or hit Ctrl-C to quit. (half-duplex: she can't hear you while she talks)") - while True: - try: - samples = record_ptt(sd) if ptt else record_until_silence(sd) - if samples.size < SAMPLE_RATE * 0.3: # <0.3 s — nothing useful - continue - text = transcribe_audio(_wav_bytes(samples), mime_type="audio/wav").strip() - if not text: + try: + while True: + # Per-turn isolation: a transcribe/agent error skips this turn instead of + # ending the whole session. Full traceback printed so failures are visible. + try: + _drain(q) # drop her own voice / stale audio + if not ptt: + print("🎧 listening…") + samples = listen_ptt(q) if ptt else listen_hands_free(q) + if samples.size < SAMPLE_RATE * 0.3: # <0.3 s — nothing useful + if not ptt: + print(" (didn't catch that)") + continue + print(" 📝 transcribing…") + text = transcribe_audio(_wav_bytes(samples), mime_type="audio/wav").strip() + if not text: + print(" (couldn't make that out)") + continue + print(f"You: {text}") + if re.sub(r"[^a-z]", "", text.lower()) in EXIT_WORDS: + speak("Goodbye.") + break + print(" 💭 thinking…") + result = agent.invoke({"messages": [HumanMessage(content=text)]}, + config=thread_config(thread_id)) + reply = extract_text(result["messages"][-1].content) + print(f"Aria: {reply}") + speak(reply) + except KeyboardInterrupt: + raise + except Exception: + print(" [turn error — skipping, full trace below]") + traceback.print_exc() continue - print(f"\nYou: {text}") - if re.sub(r"[^a-z]", "", text.lower()) in EXIT_WORDS: - speak("Goodbye.") - break - result = agent.invoke({"messages": [HumanMessage(content=text)]}, - config=thread_config(thread_id)) - reply = extract_text(result["messages"][-1].content) - print(f"Aria: {reply}") - speak(reply) - except KeyboardInterrupt: - print("\nExiting voice mode.") - break - except Exception as e: - print(f"[voice error] {e}") + except KeyboardInterrupt: + print("\nExiting voice mode.") + finally: + stream.stop() + stream.close() if __name__ == "__main__": diff --git a/voice_live.py b/voice_live.py index 1c25a75..52b54b9 100644 --- a/voice_live.py +++ b/voice_live.py @@ -15,13 +15,16 @@ """ import os import sys +import time import queue import asyncio import threading +import traceback from dotenv import load_dotenv from google import genai from google.genai import types +from google.genai import errors as genai_errors from langchain_core.messages import HumanMessage from agent_core import build_agent, open_checkpointer, thread_config, extract_text @@ -84,6 +87,9 @@ async def main(): # --- speaker: a thread draining a queue so barge-in can flush it instantly --- play_q: queue.Queue = queue.Queue() stop = threading.Event() + duplex = "--duplex" in sys.argv # headphones: full barge-in, no echo gating + HANGOVER = 0.6 # keep mic muted this long after she stops talking + state = {"last_audio": 0.0} # monotonic time of the last audio chunk *played* def player(): with sd.RawOutputStream(samplerate=OUT_RATE, channels=1, dtype="int16") as out: @@ -94,8 +100,10 @@ def player(): continue if chunk: out.write(chunk) + state["last_audio"] = time.monotonic() - threading.Thread(target=player, daemon=True).start() + player_thread = threading.Thread(target=player, daemon=True) + player_thread.start() def flush_playback(): try: @@ -122,7 +130,11 @@ def flush_playback(): "(e.g. gemini-2.0-flash-live-001).") return - print("🎙️ Live voice. Talk naturally — interrupt her any time. Ctrl-C to quit.") + print("🎙️ Live voice. " + + ("Headphones/duplex — interrupt any time. " if duplex + else "Speaker mode — mic mutes while she talks (no echo). " + "Headphones + --duplex for barge-in. ") + + "Ctrl-C to quit.") # --- mic: callback (audio thread) hands bytes to the asyncio loop --- in_q: asyncio.Queue = asyncio.Queue() @@ -137,44 +149,103 @@ def mic_cb(indata, frames, t, status): async def send_mic(): while True: data = await in_q.get() - await session.send_realtime_input( - audio=types.Blob(data=data, mime_type=f"audio/pcm;rate={IN_RATE}")) + # Echo guard (speaker mode): drop mic audio while Aria is speaking and for a + # short hangover after, so her own voice isn't captured and answered. With + # --duplex (headphones) we never gate, preserving barge-in. + if not duplex and (not play_q.empty() + or time.monotonic() - state["last_audio"] < HANGOVER): + continue + try: + await session.send_realtime_input( + audio=types.Blob(data=data, mime_type=f"audio/pcm;rate={IN_RATE}")) + except Exception: + print("\n [mic send failed — session likely closed]") + traceback.print_exc() + return + + async def _handle(msg): + sc = msg.server_content + if sc: + if sc.interrupted: # user barged in + flush_playback() + if sc.input_transcription and sc.input_transcription.text: + print(f"\nYou: {sc.input_transcription.text}") + if sc.model_turn: + for part in sc.model_turn.parts: + if part.inline_data and part.inline_data.data: + play_q.put(part.inline_data.data) + if sc.output_transcription and sc.output_transcription.text: + print(sc.output_transcription.text, end="", flush=True) + if sc.turn_complete: + print() + if msg.go_away: # server about to disconnect + print(f"\n[server ending session: {msg.go_away}]") + if msg.tool_call: + # Always return a response per call — if run_brain raises and we send + # nothing, the model hangs forever waiting for the tool result. + responses = [] + for fc in msg.tool_call.function_calls: + req = (fc.args or {}).get("request", "") + print(f"\n[→ brain] {req}") + try: + answer = await loop.run_in_executor(None, run_brain, agent, req) + except Exception: + print(" [brain error]") + traceback.print_exc() + answer = "Sorry, I hit an error reaching my tools." + print(f"[brain] {answer[:100]}") + responses.append(types.FunctionResponse( + id=fc.id, name=fc.name, response={"result": answer})) + if responses: + await session.send_tool_response(function_responses=responses) async def receive(): - async for msg in session.receive(): - sc = msg.server_content - if sc: - if sc.interrupted: # user barged in - flush_playback() - if sc.input_transcription and sc.input_transcription.text: - print(f"\nYou: {sc.input_transcription.text}") - if sc.model_turn: - for part in sc.model_turn.parts: - if part.inline_data and part.inline_data.data: - play_q.put(part.inline_data.data) - if sc.output_transcription and sc.output_transcription.text: - print(sc.output_transcription.text, end="", flush=True) - if sc.turn_complete: - print() - if msg.tool_call: - responses = [] - for fc in msg.tool_call.function_calls: - if fc.name == "escalate_to_aria": - req = (fc.args or {}).get("request", "") - print(f"\n[→ brain] {req}") - answer = await loop.run_in_executor(None, run_brain, agent, req) - print(f"[brain] {answer[:100]}") - responses.append(types.FunctionResponse( - id=fc.id, name=fc.name, response={"result": answer})) - if responses: - await session.send_tool_response(function_responses=responses) + # session.receive() yields ONE model turn then completes; loop so the + # conversation keeps going across turns instead of ending after the first reply. + try: + while True: + got = False + async for msg in session.receive(): + got = True + try: + await _handle(msg) + except Exception: + print("\n [message-handling error — continuing]") + traceback.print_exc() + if not got: + await asyncio.sleep(0.05) # avoid a tight spin on empty turns + except genai_errors.APIError as e: + if getattr(e, "code", None) != 1000: # 1000 = normal close (Ctrl+C / end) + print("\n [receive stream error]") + traceback.print_exc() + except Exception: + print("\n [receive stream error]") + traceback.print_exc() + finally: + print("\n[session closed]") try: - await asyncio.gather(send_mic(), receive()) + # Exit as soon as either side ends (e.g. the session closes) instead of hanging. + tasks = [asyncio.create_task(send_mic()), asyncio.create_task(receive())] + await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED) + for t in tasks: + t.cancel() finally: - mic.stop() + # Orderly teardown. Stop the mic first (halts callbacks into the loop), then let + # the player thread close its PortAudio stream *in its own thread* before exit. + # Without the join the daemon thread is killed mid-PortAudio at interpreter + # shutdown, which double-frees → "malloc: pointer being freed was not allocated". + try: + mic.stop() + mic.close() + except Exception: + pass stop.set() - await cm.__aexit__(None, None, None) + player_thread.join(timeout=2) + try: + await cm.__aexit__(None, None, None) + except Exception: + pass if __name__ == "__main__":