diff --git a/deployment/aimaq.env.production b/deployment/aimaq.env.production index 833fe53..db6c462 100644 --- a/deployment/aimaq.env.production +++ b/deployment/aimaq.env.production @@ -85,12 +85,13 @@ AI_VOICE_TTS_CACHE_ENABLED=1 AI_VOICE_TTS_CACHE_DIR=/app/.data_local/ai_voice_tts_cache AI_VOICE_TTS_ELEVENLABS_API_KEY=sk_eb9bd36750e2a7fb378259b7144e978a044cfb412e6a2ac9 AI_VOICE_TTS_ELEVENLABS_API_BASE=https://api.elevenlabs.io -AI_VOICE_TTS_ELEVENLABS_MODEL_ID=eleven_flash_v2_5 +AI_VOICE_TTS_ELEVENLABS_MODEL_ID=eleven_turbo_v2_5 AI_VOICE_TTS_ELEVENLABS_RU_VOICE_ID=4O1sYUnmtThcBoSBrri7 AI_VOICE_TTS_ELEVENLABS_RU_LANGUAGE_CODE=ru AI_VOICE_TTS_ELEVENLABS_KK_VOICE_ID=4O1sYUnmtThcBoSBrri7 AI_VOICE_TTS_ELEVENLABS_KK_LANGUAGE_CODE=kk AI_VOICE_TTS_ELEVENLABS_OUTPUT_FORMAT=pcm_16000 +AI_VOICE_TTS_STREAM_PREBUFFER_MS=200 AI_VOICE_TTS_YANDEX_API_KEY=AQWJUMiUaXmbegxN4kgvM2XIlNqAPoBR5Wtq-40 AI_VOICE_TTS_YANDEX_FOLDER_ID=ao7hkif5pvc7vfnmbl0d diff --git a/services/ai_voice_runtime_service/media_runtime.py b/services/ai_voice_runtime_service/media_runtime.py index f5f861a..a1e1554 100644 --- a/services/ai_voice_runtime_service/media_runtime.py +++ b/services/ai_voice_runtime_service/media_runtime.py @@ -24,6 +24,7 @@ from services.ai_voice_runtime_service.audiosocket import ( pcm16le_to_wav_bytes, read_packet, resample_pcm16le, + resample_pcm16le_stateful, ) from services.ai_voice_runtime_service.providers.asr import ( ASRTranscription, @@ -199,6 +200,12 @@ class AudioSocketMediaRuntime: self._v2_ack_post_gap_seconds = 0.10 self._v1_ack_wait_seconds = 0.6 self._partial_poll_interval_seconds = 0.20 + # Small startup cushion for streamed TTS playback: absorb ElevenLabs + # network delivery jitter before we start pacing frames out to the + # caller, so a brief mid-download stall doesn't read as a dead-air gap. + self._tts_stream_prebuffer_ms = max( + int(os.getenv("AI_VOICE_TTS_STREAM_PREBUFFER_MS", "200") or "200"), 0 + ) self._streaming_asr_partial_poll_enabled = ( str(os.getenv("AI_VOICE_V2_STREAMING_ASR_PARTIAL_POLL_ENABLED", "0")).strip().lower() in {"1", "true", "yes", "on"} @@ -471,6 +478,10 @@ class AudioSocketMediaRuntime: def _immediate_ack_min_bytes(self) -> int: return self._immediate_ack_min_ms * 16 + @property + def _tts_stream_prebuffer_bytes(self) -> int: + return self._tts_stream_prebuffer_ms * 16 + @staticmethod def _reset_live_turn_state(actor: MediaActor) -> None: actor.utterance_generation += 1 @@ -1861,9 +1872,13 @@ class AudioSocketMediaRuntime: await self._set_actor_state(actor, "speaking") actor.current_reply_phase = reply_phase synth_started_at = time.monotonic() + first_chunk_logged = False first_frame_sent = False total_audio_bytes = 0 interrupted = False + pending_pcm = bytearray() + resample_state: object | None = None + prebuffered = False await self._record_reply_status( actor, text=text, @@ -1871,11 +1886,30 @@ class AudioSocketMediaRuntime: phase=reply_phase, status="started", ) + + async def _write_pcm_frames(pcm_bytes: bytes) -> bool: + nonlocal first_frame_sent + for frame in chunk_audio(pcm_bytes, frame_bytes=actor.frame_bytes): + if actor.closed or actor.playback_interrupt.is_set(): + return True + await self._write_audio_packet(actor, frame) + if not first_frame_sent: + first_frame_sent = True + logger.info( + "audiosocket.first_frame session_id=%s greeting=%s frame_bytes=%s", + actor.registration.voice_session_id, + is_greeting, + len(frame), + ) + await asyncio.sleep(actor.frame_ms / 1000.0) + return False + async for synthesis in self._stream_tts_chunks(actor, text, style_hints=style_hints): if not synthesis.audio_bytes: continue total_audio_bytes += len(synthesis.audio_bytes) - if not first_frame_sent: + if not first_chunk_logged: + first_chunk_logged = True logger.info( "audiosocket.tts_ready session_id=%s greeting=%s synth_ms=%s audio_bytes=%s", actor.registration.voice_session_id, @@ -1888,29 +1922,29 @@ class AudioSocketMediaRuntime: await self._record_latency_metric(actor, "speech_start_to_ack_start", actor.speech_started_monotonic) elif reply_phase == "main": await self._record_latency_metric(actor, "speech_end_to_main_reply_start", actor.speech_ended_monotonic) - pcm_8k = resample_pcm16le( + # Carry ratecv state across chunks: resampling each network chunk + # independently introduces an audible click/discontinuity at every + # chunk boundary, which gets much more noticeable with a fast model + # that streams many small chunks (e.g. eleven_flash_v2_5). + pcm_8k, resample_state = resample_pcm16le_stateful( synthesis.audio_bytes, input_rate_hz=synthesis.sample_rate_hz, output_rate_hz=8000, + state=resample_state, ) - for frame in chunk_audio(pcm_8k, frame_bytes=actor.frame_bytes): - if actor.closed or actor.playback_interrupt.is_set(): - interrupted = True - break - await self._write_audio_packet(actor, frame) - if not first_frame_sent: - first_frame_sent = True - logger.info( - "audiosocket.first_frame session_id=%s greeting=%s frame_bytes=%s", - actor.registration.voice_session_id, - is_greeting, - len(frame), - ) - await asyncio.sleep(actor.frame_ms / 1000.0) - if actor.closed or actor.playback_interrupt.is_set(): - interrupted = True + pending_pcm.extend(pcm_8k) + if not prebuffered and len(pending_pcm) < self._tts_stream_prebuffer_bytes: + continue + prebuffered = True + interrupted = await _write_pcm_frames(bytes(pending_pcm)) + pending_pcm.clear() + if interrupted: break + if not interrupted and pending_pcm: + interrupted = await _write_pcm_frames(bytes(pending_pcm)) + pending_pcm.clear() + if total_audio_bytes <= 0: raise RuntimeError("TTS provider returned empty audio") diff --git a/tests/test_ai_voice_media_runtime.py b/tests/test_ai_voice_media_runtime.py index 3307987..2ced970 100644 --- a/tests/test_ai_voice_media_runtime.py +++ b/tests/test_ai_voice_media_runtime.py @@ -1,5 +1,6 @@ import asyncio import contextlib +import struct import threading import time import uuid @@ -2899,3 +2900,91 @@ def test_media_runtime_no_speech_watchdog_reprompts_on_silence_timeout(): assert spoken, "watchdog should reprompt after prolonged silence in listening state" asyncio.run(_scenario()) + + +def test_media_runtime_streaming_tts_prebuffers_without_dropping_audio(): + class _ChunkedTTSProvider(TTSProvider): + name = "chunked-tts" + + def synthesize(self, text, *, language=None, style_hints=None): + raise AssertionError("streaming session should use synthesize_chunks") + + def synthesize_chunks(self, text, *, language=None, style_hints=None): + del language, style_hints + # 12 small 16kHz chunks (5ms each), each well under the prebuffer + # target, so both the accumulation path and the trailing flush + # (for whatever is still buffered once the stream ends) get exercised. + tone = (500).to_bytes(2, "little", signed=True) * 80 + for _ in range(12): + yield TTSSynthesis(text=text, audio_bytes=tone, sample_rate_hz=16000) + + class _FakeWriter: + def __init__(self) -> None: + self.packets: list[bytes] = [] + + def write(self, data: bytes) -> None: + self.packets.append(data) + + async def drain(self) -> None: + return None + + registration = MediaRegistration( + voice_session_id="avs_media_runtime_prebuffer", + call_id="call_media_runtime_prebuffer", + interaction_id="int_media_runtime_prebuffer", + ai_session_id="ais_media_runtime_prebuffer", + language="ru", + media_uuid=str(uuid.uuid4()), + voice_v2_streaming_tts=True, + ) + runtime = AudioSocketMediaRuntime( + enabled=True, + host="127.0.0.1", + port=0, + frame_ms=20, + idle_timeout_seconds=2.0, + registration_wait_timeout_seconds=0.5, + min_speech_ms=40, + trailing_silence_ms=40, + max_turn_ms=400, + asr_provider=_StubASRProvider(), + tts_provider=_ChunkedTTSProvider(), + load_registration_by_media_uuid=lambda value: None, + mark_media_connected=lambda session_id, value: None, + mark_media_ended=lambda session_id, reason: None, + touch_media_frame=lambda session_id: None, + set_state=lambda session_id, state, handoff_reason, metadata: None, + get_pending_greeting=lambda session_id: None, + mark_reply_delivered=lambda session_id, text, is_greeting: None, + plan_reply=lambda session_id, text, metadata, kind: None, + process_turn=lambda session_id, transcript_text, language, barge_in, metadata: None, + request_handoff=lambda session_id, customer_request_text, decision: None, + handle_media_error=lambda session_id, message, metadata: None, + ) + runtime._tts_stream_prebuffer_ms = 40 + + writer = _FakeWriter() + + async def _scenario() -> None: + actor = MediaActor( + registration=registration, + reader=asyncio.StreamReader(), + writer=writer, # type: ignore[arg-type] + vad=EnergyVAD(frame_ms=20, min_speech_ms=40, trailing_silence_ms=40, max_turn_ms=400), + frame_ms=20, + frame_bytes=320, + ) + await runtime._speak_text(actor, "ั‚ะตัั‚", is_greeting=False, reply_phase="main") + + asyncio.run(_scenario()) + + assert len(writer.packets) > 1, "audio should be paced out frame by frame, not as one blob" + payload_bytes = 0 + for packet in writer.packets: + packet_type, payload_length = struct.unpack("!BH", packet[:3]) + assert packet_type == AUDIO_SOCKET_PACKET_PCM16 + assert len(packet) == 3 + payload_length == 3 + 320, "every frame must be frame_bytes, zero-padded if short" + payload_bytes += payload_length + # 12 chunks * 160 bytes @16kHz downsample 2:1 -> 960 bytes @8kHz of real + # audio; frame padding on flush boundaries can only add silence, never drop it. + assert payload_bytes >= 960