diff --git a/deployment/aimaq.env.production b/deployment/aimaq.env.production index 2696a3f..07fc800 100644 --- a/deployment/aimaq.env.production +++ b/deployment/aimaq.env.production @@ -68,6 +68,7 @@ AI_VOICE_ASR_ELEVENLABS_LANGUAGE=ru AI_VOICE_ASR_ELEVENLABS_REALTIME_MODEL_ID=scribe_v2_realtime AI_VOICE_ASR_ELEVENLABS_REALTIME_AUDIO_FORMAT=pcm_16000 AI_VOICE_ASR_ELEVENLABS_REALTIME_COMMIT_STRATEGY=vad +AI_VOICE_ASR_ELEVENLABS_REALTIME_DEBUG_DUMP_AUDIO=1 AI_VOICE_ASR_ELEVENLABS_REALTIME_FINALIZE_TIMEOUT_SECONDS=2.5 AI_VOICE_ASR_YANDEX_API_KEY=AQWJe16Rdlx_peHUppZ79csG49f1gP4wwQyDzeQ_ AI_VOICE_ASR_YANDEX_FOLDER_ID=ao7hkif5pvc7vfnmbl0d diff --git a/services/ai_voice_runtime_service/providers/asr.py b/services/ai_voice_runtime_service/providers/asr.py index 090b880..bb9f30d 100644 --- a/services/ai_voice_runtime_service/providers/asr.py +++ b/services/ai_voice_runtime_service/providers/asr.py @@ -13,6 +13,7 @@ import uuid import wave from collections.abc import Callable from dataclasses import dataclass, field +from pathlib import Path import httpx @@ -137,6 +138,18 @@ def _elevenlabs_realtime_commit_strategy() -> str: return raw if raw in {"manual", "vad"} else "manual" +def _elevenlabs_realtime_debug_dump_audio_enabled() -> bool: + raw = os.getenv("AI_VOICE_ASR_ELEVENLABS_REALTIME_DEBUG_DUMP_AUDIO", "").strip().lower() + return raw in {"1", "true", "yes", "on"} + + +def _elevenlabs_realtime_debug_dump_audio_dir() -> Path: + explicit = os.getenv("AI_VOICE_ASR_ELEVENLABS_REALTIME_DEBUG_DUMP_DIR", "").strip() + if explicit: + return Path(explicit) + return Path(os.getenv("CC_DATA_DIR", ".data_local").strip() or ".data_local") / "asr_debug_dumps" + + def _elevenlabs_realtime_finalize_timeout_seconds() -> float: raw_ms = os.getenv("AI_VOICE_V2_STREAMING_FINAL_HARD_TIMEOUT_MS", "").strip() if raw_ms: @@ -539,6 +552,8 @@ class _ElevenLabsRealtimeStreamState: resample_lock: threading.Lock = field(default_factory=threading.Lock) committed_segments: list[str] = field(default_factory=list) finalize_requested: bool = False + debug_pcm: bytearray = field(default_factory=bytearray) + debug_lock: threading.Lock = field(default_factory=threading.Lock) class ElevenLabsRealtimeStreamingASRProvider(StreamingASRProvider): @@ -556,6 +571,8 @@ class ElevenLabsRealtimeStreamingASRProvider(StreamingASRProvider): audio_format: str | None = None, commit_strategy: str | None = None, websocket_factory: Callable[..., object] | None = None, + debug_dump_audio: bool | None = None, + debug_dump_dir: str | Path | None = None, ) -> None: self._api_base = str(api_base or _elevenlabs_asr_api_base()).strip().rstrip("/") self._api_key = str(api_key if api_key is not None else _elevenlabs_asr_api_key()).strip() @@ -580,6 +597,10 @@ class ElevenLabsRealtimeStreamingASRProvider(StreamingASRProvider): self._websocket_factory = websocket_factory self._streams: dict[str, _ElevenLabsRealtimeStreamState] = {} self._lock = threading.Lock() + self._debug_dump_audio = ( + debug_dump_audio if debug_dump_audio is not None else _elevenlabs_realtime_debug_dump_audio_enabled() + ) + self._debug_dump_dir = Path(debug_dump_dir) if debug_dump_dir is not None else _elevenlabs_realtime_debug_dump_audio_dir() def _websocket_url(self, *, language: str) -> str: base = self._api_base @@ -860,6 +881,9 @@ class ElevenLabsRealtimeStreamingASRProvider(StreamingASRProvider): output_rate_hz=self._sample_rate_hz, state=state.resample_state, ) + if self._debug_dump_audio: + with state.debug_lock: + state.debug_pcm.extend(pcm_bytes) self._send_json( state, { @@ -870,6 +894,29 @@ class ElevenLabsRealtimeStreamingASRProvider(StreamingASRProvider): }, ) + def _write_debug_dump(self, state: _ElevenLabsRealtimeStreamState) -> None: + with state.debug_lock: + pcm_bytes = bytes(state.debug_pcm) + if not pcm_bytes: + return + try: + self._debug_dump_dir.mkdir(parents=True, exist_ok=True) + path = self._debug_dump_dir / f"{state.stream_id}.wav" + path.write_bytes(pcm16le_to_wav_bytes(pcm_bytes, sample_rate_hz=self._sample_rate_hz)) + except OSError as exc: + logger.warning( + "asr.elevenlabs_realtime.debug_dump_failed stream_id=%s error=%s", + state.stream_id, + str(exc)[:300], + ) + return + logger.info( + "asr.elevenlabs_realtime.debug_dump_written stream_id=%s path=%s bytes=%s", + state.stream_id, + path, + len(pcm_bytes), + ) + def poll_partial(self, stream_id: str) -> StreamingASRPartial | None: state = self._state(stream_id) self._drain_updates(state) @@ -955,6 +1002,8 @@ class ElevenLabsRealtimeStreamingASRProvider(StreamingASRProvider): thread = state.thread if thread is not None and thread.is_alive(): thread.join(timeout=0.5) + if self._debug_dump_audio: + self._write_debug_dump(state) with self._lock: self._streams.pop(stream_id, None) logger.info("asr.elevenlabs_realtime.stream_close stream_id=%s", stream_id) diff --git a/tests/test_ai_voice_asr_provider.py b/tests/test_ai_voice_asr_provider.py index 67fd208..9a771a9 100644 --- a/tests/test_ai_voice_asr_provider.py +++ b/tests/test_ai_voice_asr_provider.py @@ -3,6 +3,7 @@ from __future__ import annotations import json import queue import time +import wave import httpx import pytest @@ -529,6 +530,51 @@ def test_elevenlabs_realtime_streaming_provider_accumulates_autonomous_vad_commi assert final.language == "ru" +def test_elevenlabs_realtime_streaming_provider_writes_debug_audio_dump(tmp_path): + class _FakeRealtimeWebSocket: + def __init__(self) -> None: + self.incoming: queue.Queue[str | None] = queue.Queue() + + def send(self, raw_payload: str) -> None: + payload = json.loads(raw_payload) + if payload.get("commit"): + self.incoming.put( + json.dumps({"message_type": "committed_transcript", "text": "ั‚ะตัั‚", "language_code": "ru"}) + ) + + def recv(self) -> str: + item = self.incoming.get(timeout=1) + if item is None: + raise RuntimeError("closed") + return item + + def close(self) -> None: + self.incoming.put(None) + + websocket = _FakeRealtimeWebSocket() + dump_dir = tmp_path / "asr_debug_dumps" + + provider = asr_module.ElevenLabsRealtimeStreamingASRProvider( + api_base="https://api.elevenlabs.example", + api_key="asr-key", + timeout_seconds=1, + finalize_timeout_seconds=1, + websocket_factory=lambda url, *, header, timeout: websocket, + debug_dump_audio=True, + debug_dump_dir=dump_dir, + ) + stream_id = provider.open_stream("session-1", language_hint="ru") + provider.push_pcm(stream_id, b"\x10\x00" * 160) + provider.finalize(stream_id) + provider.close_stream(stream_id) + + dumped = dump_dir / f"{stream_id}.wav" + assert dumped.exists() + with wave.open(str(dumped), "rb") as wav_file: + assert wav_file.getframerate() == 16000 + assert wav_file.getnframes() > 0 + + def test_yandex_grpc_streaming_provider_returns_partial_and_final(): calls: list[dict] = []