""" TTS and Omni callback classes for the Qwen Pipeline. These bridge DashScope SDK events (which run on internal threads) to the asyncio event loop via session queues. """ import asyncio import json import time from datetime import datetime # Diagnostic add-on: audio-piece timing tracer (safe no-op if unavailable). try: import audio_timing_trace except Exception: audio_timing_trace = None # ── Conditional Omni SDK import ────────────────────────────────── try: from dashscope.audio.qwen_tts_realtime import QwenTtsRealtimeCallback except ImportError: QwenTtsRealtimeCallback = object try: from dashscope.audio.qwen_omni import OmniRealtimeCallback OMNI_SDK_AVAILABLE = True except ImportError: OmniRealtimeCallback = object OMNI_SDK_AVAILABLE = False # ── TTS Callback (runs on DashScope internal thread) ──────────── class TtsCallback(QwenTtsRealtimeCallback): def __init__(self, loop, queue): self.loop = loop self.queue = queue self.done = __import__('threading').Event() self.cancelled = False # set externally to stop audio output def on_open(self): pass def on_close(self, *args): self.done.set() def on_event(self, response): if self.cancelled: self.done.set() return try: etype = response.get('type', '') if etype == 'response.audio.delta': audio = response.get('delta', '') if audio and not self.cancelled: self.loop.call_soon_threadsafe( self.queue.put_nowait, {'type': 'audio', 'data': audio} ) elif etype == 'session.finished': self.done.set() except Exception as e: print(f' [TTS callback] Error: {e}') class OmniCallback(OmniRealtimeCallback if OMNI_SDK_AVAILABLE else object): """Bridges Omni SDK events (runs on DashScope's internal WebSocket thread) to the asyncio event loop via the session's queue.""" def __init__(self, session): self.session = session self.loop = session.loop self.queue = session.queue def on_open(self): print('[Omni] WebSocket connected') def on_close(self, close_status_code=None, close_msg=None): print(f'[Omni] WebSocket closed: {close_status_code} {close_msg}') self.session.active = False def on_event(self, event): """Dispatch Omni events. Runs on DashScope's internal thread.""" if not self.session.active: return try: etype = event.get('type', '') self._dispatch(etype, event) except Exception as e: print(f' [Omni callback] Error: {e}') def _dispatch(self, etype, event): s = self.session # ── User speech transcript ── if etype == 'conversation.item.input_audio_transcription.completed': transcript = event.get('transcript', '').strip() if transcript: s._idle_prompt_count = 0 s._last_activity_time = time.time() s._omni_user_transcript = transcript self.loop.call_soon_threadsafe( self.queue.put_nowait, {'type': 'transcript', 'speaker': 'user', 'text': transcript} ) s._log_message('user', transcript) s._agent['history'].append({'role': 'user', 'content': transcript}) # Log turn for prompt view (mirrors pipeline prompt_log) s.prompt_log.append({ 'turn': len(s.prompt_log) + 1, 'timestamp': datetime.now().strftime('%H:%M:%S'), 'agent': s.active_agent, 'agent_name': s._agent.get('name', ''), 'messages': [dict(m) for m in s._agent.get('history', [])], 'tools': s._build_omni_tools(), 'omni_mode': True, 'omni_event': 'user_turn', }) # Check for keyword-triggered transfer (hidden agents) kw_key, kw_agent = s._check_keyword_trigger(transcript) if kw_key and s.active_agent == 'a': print(f' [Omni KEYWORD] "{kw_agent["keyword"]}" matched ' f'→ transferring to {kw_agent["name"]} ({kw_key})') asyncio.run_coroutine_threadsafe( s._omni_keyword_transfer(kw_key, kw_agent), self.loop) # ── Barge-in: user started speaking ── elif etype == 'input_audio_buffer.speech_started': if s._omni_responding and not s._omni_barge_in: print(' [Omni BARGE-IN] speech_started event during response') asyncio.run_coroutine_threadsafe( s._omni_trigger_barge_in(), self.loop) # ── Barge-in: Omni interrupted its own audio ── elif etype == 'response.audio.interrupted': if not s._omni_barge_in: asyncio.run_coroutine_threadsafe( s._omni_trigger_barge_in(), self.loop) # ── AI audio output ── elif etype == 'response.audio.delta': if s._omni_barge_in: return # Drop audio during barge-in audio = event.get('delta', '') if not audio: return # Diagnostic: stamp the moment this piece arrived from the cloud. if audio_timing_trace: audio_timing_trace.record_recv(s.session_id, len(audio) * 3 // 4) if s._omni_tool_call_audio_mute: # A swallowed response's audio must be DISCARDED, never # buffered: otherwise it lingers in _omni_muted_audio_buffer # (which is cleared only by the next tool call) and a later # end_call flushes it — replaying the suppressed line (e.g. a # transfer's "I'm sorry, no one available") at hang-up, in the # original agent's voice. Only genuinely mutable audio (an # end_call farewell) is buffered for flush. if not getattr(s, '_omni_swallowing', False): s._omni_muted_audio_buffer.append(audio) return # Pre-emit buffering: hold audio during the buffer window if s._omni_pre_emit_buffering and s._omni_pre_emit_gen == s._omni_audio_gen: s._omni_pre_emit_buffer.append(audio) return # Track audio bytes for playback duration estimate s._omni_response_audio_bytes += len(audio) * 3 // 4 # base64 → raw bytes if not hasattr(s, '_dbg_audio_count'): s._dbg_audio_count = 0 s._dbg_audio_count += 1 if s._dbg_audio_count % 20 == 1: print(f' [Omni CB] audio.delta #{s._dbg_audio_count} ' f'(agent={s.active_agent}, len={len(audio)})') self.loop.call_soon_threadsafe( self.queue.put_nowait, {'type': 'audio', 'data': audio, '_gen': s._omni_audio_gen} ) # ── AI text transcript (streaming) ── elif etype == 'response.audio_transcript.delta': delta = event.get('delta', '') if delta: s._omni_response_text += delta # ── AI text transcript (complete) ── elif etype == 'response.audio_transcript.done': transcript = event.get('transcript', '') or s._omni_response_text # Filter out transcripts that are just tool call details spoken aloud if transcript and ( 'transfer_call' in transcript or 'transfer_back' in transcript or 'select_language' in transcript or 'change_language' in transcript or 'end_call' in transcript or 'call_back' in transcript ) and ('"type"' in transcript or '"department"' in transcript or '"reason"' in transcript or '"language"' in transcript or 'function' in transcript.lower()): print(f' [Omni] Suppressed tool-call transcript: {transcript[:100]}') transcript = None # Suppress speculative speech generated alongside a transfer tool # call: its audio was muted (never played to the caller), and the # model often hallucinates a failed transfer ("no one is # available..."). Logging it would show the caller a line that # was never spoken and poison agent history. The real spoken # line is the handler's fixed 'Say EXACTLY' farewell. if (transcript and s._omni_tool_call_audio_mute and s._omni_tool_name in ('transfer_call', 'transfer_back')): print(f' [Omni] Suppressed muted transfer speech: ' f'{transcript[:100]}') transcript = None # Suppress the transcript of a swallowed auto-response — its # audio was muted, the caller never heard it. if transcript and s._omni_swallowing: print(f' [Omni] Suppressed swallowed auto-response transcript: ' f'{transcript[:100]}') transcript = None if transcript: agent_name = s._agent.get('name', 'AI') self.loop.call_soon_threadsafe( self.queue.put_nowait, {'type': 'transcript', 'speaker': 'ai', 'text': transcript, 'agent': agent_name} ) s._log_message('ai', transcript, agent_name) s._agent['history'].append( {'role': 'assistant', 'content': transcript}) # ── Tool call streaming ── elif etype == 'response.function_call_arguments.delta': s._omni_tool_args += event.get('delta', '') # Mute new audio deltas — all tools are called silently now. # Do NOT drain the queue (already-queued audio is fine). if not s._omni_tool_call_audio_mute: s._omni_tool_call_audio_mute = True print(' [Omni] Tool call streaming, muting new audio deltas') # Save any pre-emit buffered audio to the muted buffer # so tool handlers (e.g. end_call) can flush it as farewell audio if s._omni_pre_emit_buffering: if s._omni_pre_emit_buffer: s._omni_muted_audio_buffer.extend(s._omni_pre_emit_buffer) print(f' [Omni] Moved {len(s._omni_pre_emit_buffer)} ' f'pre-emit chunks to muted buffer (tool call detected)') s._omni_pre_emit_buffer = [] s._omni_pre_emit_buffering = False if s._omni_pre_emit_timer is not None: self.loop.call_soon_threadsafe(s._omni_pre_emit_timer.cancel) elif etype == 'response.function_call_arguments.done': name = event.get('name', '') args = event.get('arguments', '') or s._omni_tool_args call_id = event.get('call_id', '') s._omni_tool_name = name s._omni_tool_call_id = call_id print(f' [Omni] Tool call: {name}({args[:100]})') asyncio.run_coroutine_threadsafe( s._handle_omni_tool_call(name, args, call_id), self.loop ) # ── Response lifecycle ── elif etype == 'response.created': print(f' [Omni CB] response.created (agent={s.active_agent}, mute_was={s._omni_tool_call_audio_mute})') # Swallow flag: this response is DashScope's auto-response to a # function_call_output (transfer flow). Keep it muted and # suppress its transcript — the handler speaks the real farewell. if s._omni_swallow_next_response: s._omni_swallow_next_response = False s._omni_swallowing = True print(' [Omni CB] Swallowing auto-response (muted, no transcript)') else: s._omni_swallowing = False # Cancel any pending pre-emit flush timer if s._omni_pre_emit_timer is not None: self.loop.call_soon_threadsafe(s._omni_pre_emit_timer.cancel) s._omni_pre_emit_timer = None # If we were buffering audio for a previous response and a NEW # response.created arrives (parallel-response scenario), discard # the old buffer — it was likely a verbal answer superseded by # the new response (which may be a tool call). if s._omni_pre_emit_buffering: if s._omni_pre_emit_buffer: print(f' [Omni CB] Discarding {len(s._omni_pre_emit_buffer)} ' f'pre-emit audio chunks (new response started)') s._omni_pre_emit_buffer = [] s._omni_pre_emit_buffering = False # Bump audio generation — sender will skip chunks from older # generations (implicit barge-in for phone/bridge connections # where RMS/VAD detection is unavailable). prev_gen = s._omni_audio_gen s._omni_audio_gen += 1 # Tell client to flush playback buffer — audio from the previous # response may already be at the client (bridge/browser) even though # our queue is empty. Skip for gen 1 (first greeting response) and # server-initiated responses (farewell, greeting, tool follow-up). if prev_gen > 0 and not s._omni_suppress_flush and not s._omni_swallowing: self.loop.call_soon_threadsafe( self.queue.put_nowait, {'type': 'barge_in'} ) # Capture whether this is server-initiated before resetting is_server_initiated = s._omni_suppress_flush s._omni_suppress_flush = False s._omni_responding = True s._omni_barge_in = False # Clear barge-in for new response s._omni_tool_call_audio_mute = s._omni_swallowing # Mute swallowed auto-response, allow audio otherwise s._omni_response_audio_bytes = 0 s._omni_response_start_time = time.time() s._omni_response_text = '' s._omni_tool_name = '' s._omni_tool_args = '' s._omni_tool_call_id = '' self.loop.call_soon_threadsafe(s._omni_response_done_event.clear) self.loop.call_soon_threadsafe( self.queue.put_nowait, {'type': 'status', 'status': 'speaking'} ) # Start pre-emit buffer window for user-initiated responses if not is_server_initiated and s._omni_audio_buffer_delay > 0: s._omni_pre_emit_buffering = True s._omni_pre_emit_gen = s._omni_audio_gen s._omni_pre_emit_buffer = [] current_gen = s._omni_audio_gen self.loop.call_soon_threadsafe( self._schedule_pre_emit_flush, current_gen, s._omni_audio_buffer_delay ) elif etype == 'response.done': print(f' [Omni CB] response.done (agent={s.active_agent})') s._omni_responding = False s._omni_barge_in_speech_start = None s._omni_swallowing = False # Swallowed auto-response finished # If pre-emit buffer is still active, flush it now # (response completed without a tool call) if s._omni_pre_emit_buffering and s._omni_pre_emit_gen == s._omni_audio_gen: chunks = s._omni_pre_emit_buffer s._omni_pre_emit_buffer = [] s._omni_pre_emit_buffering = False if s._omni_pre_emit_timer is not None: self.loop.call_soon_threadsafe(s._omni_pre_emit_timer.cancel) s._omni_pre_emit_timer = None if chunks and not s._omni_tool_call_audio_mute: print(f' [Omni] Flushing {len(chunks)} pre-emit chunks on response.done') for audio in chunks: s._omni_response_audio_bytes += len(audio) * 3 // 4 self.loop.call_soon_threadsafe( self.queue.put_nowait, {'type': 'audio', 'data': audio, '_gen': s._omni_audio_gen} ) # Estimate when client finishes playing audio (PCM 24kHz 16-bit mono) if s._omni_response_start_time and s._omni_response_audio_bytes > 0: audio_duration = s._omni_response_audio_bytes / (24000 * 2) s._last_activity_time = s._omni_response_start_time + audio_duration else: s._last_activity_time = time.time() self.loop.call_soon_threadsafe(s._omni_response_done_event.set) self.loop.call_soon_threadsafe( self.queue.put_nowait, {'type': 'status', 'status': 'listening'} ) # Trim history if too long agent = s._agent if len(agent.get('history', [])) > 8: agent['history'] = agent['history'][-8:] # Log usage and accumulate for cost tracking usage = event.get('response', {}).get('usage', {}) if usage: print(f' [Omni] Usage: {usage}') inp = usage.get('input_tokens_details', {}) out = usage.get('output_tokens_details', {}) s._omni_usage_totals['input_text_tokens'] += inp.get('text_tokens', 0) s._omni_usage_totals['input_audio_tokens'] += inp.get('audio_tokens', 0) s._omni_usage_totals['output_text_tokens'] += out.get('text_tokens', 0) s._omni_usage_totals['output_audio_tokens'] += out.get('audio_tokens', 0) s._omni_usage_totals['turns'] += 1 elif etype == 'error': err = event.get('error', {}) err_msg = err.get('message', str(err)) if isinstance(err, dict) else str(err) print(f' [Omni] Error: {err}') self.loop.call_soon_threadsafe( self.queue.put_nowait, {'type': 'error', 'message': f'Omni error: {err_msg}'} ) # ── Pre-emit audio buffer management ─────────────────────────── def _schedule_pre_emit_flush(self, gen, delay): """Schedule a pre-emit buffer flush after `delay` seconds. Must be called on the event loop thread (via call_soon_threadsafe).""" s = self.session if s._omni_pre_emit_timer is not None: s._omni_pre_emit_timer.cancel() s._omni_pre_emit_timer = self.loop.call_later( delay, self._flush_pre_emit_buffer, gen ) def _flush_pre_emit_buffer(self, gen): """Flush buffered audio to client if still valid. Runs on the event loop thread (scheduled by call_later).""" s = self.session s._omni_pre_emit_timer = None # Only flush if still in buffer window for this gen if not s._omni_pre_emit_buffering: return if s._omni_pre_emit_gen != gen: return if s._omni_tool_call_audio_mute: if s._omni_pre_emit_buffer: s._omni_muted_audio_buffer.extend(s._omni_pre_emit_buffer) print(f' [Omni] Pre-emit flush: moved {len(s._omni_pre_emit_buffer)} ' f'chunks to muted buffer (tool call muted, gen={gen})') s._omni_pre_emit_buffer = [] s._omni_pre_emit_buffering = False return chunks = s._omni_pre_emit_buffer s._omni_pre_emit_buffer = [] s._omni_pre_emit_buffering = False if chunks: print(f' [Omni] Flushing {len(chunks)} pre-emit audio chunks (gen={gen})') for audio in chunks: s._omni_response_audio_bytes += len(audio) * 3 // 4 self.queue.put_nowait( {'type': 'audio', 'data': audio, '_gen': gen} )