diff --git a/.env.example b/.env.example index c9031db0..dbb84236 100644 --- a/.env.example +++ b/.env.example @@ -125,6 +125,14 @@ EVA_MODEL__TTS_PARAMS='{"api_key": "your_cartesia_api_key", "model": "sonic"}' #x pipeline_mode=S2S #v EVA_MODEL__S2S_PARAMS='{"model": "gpt-realtime-mini", "api_key": ""}' +# Smallest Hydra S2S (set EVA_FRAMEWORK=smallest_hydra). Hydra emits no transcript +# on the wire, so a batch STT transcribes each turn for the audit log; it defaults +# to Smallest Pulse keyed on the same api_key. Override via the "transcription" block +# (provider: smallest|openai|deepgram). voice: wren|sloane|marlowe|reed|knox|tate. +#x pipeline_mode=S2S +#v EVA_MODEL__S2S=hydra +#v EVA_MODEL__S2S_PARAMS='{"model": "hydra", "api_key": "", "voice": "wren", "generate_initial_response": true, "transcription": {"provider": "smallest", "model": "pulse-pro", "language": "en"}}' + # --- AudioLLM mode --- #i Audio-input LLM model name. #d string @@ -139,7 +147,7 @@ EVA_MODEL__TTS_PARAMS='{"api_key": "your_cartesia_api_key", "model": "sonic"}' # --- Framework (S2S / AudioLLM) --- #i Base framework for S2S or AudioLLM pipelines. #d enum -#e pipecat,openai_realtime,gemini_live,elevenlabs,grok_voice +#e pipecat,openai_realtime,gemini_live,elevenlabs,grok_voice,smallest_hydra #v EVA_FRAMEWORK=openai_realtime # ============================================== diff --git a/docs/assistant_server_contract.md b/docs/assistant_server_contract.md index 62124ac5..7e04cc0e 100644 --- a/docs/assistant_server_contract.md +++ b/docs/assistant_server_contract.md @@ -535,3 +535,42 @@ the run to fail or produce `None` latency fields in the result. | `audio_assistant.wav` | Yes | TTS quality metrics | | `framework_logs.jsonl` | Yes | Turn boundary metrics | | `pipecat_metrics.jsonl` | Yes | `model_response_latency` in `ConversationResult` | + +--- + +## 13. Reference implementation: Smallest Hydra (transcript-less S2S) + +`src/eva/assistant/smallest_hydra_server.py` (`framework: smallest_hydra`) bridges to Smallest's +**Hydra** speech-to-speech model over a raw WebSocket (no SDK; uses `websockets`). It is scored as a +true **S2S** pipeline (`s2s: hydra`, so `get_pipeline_type → S2S`) and follows the Gemini Live +three-task structure (forward user audio, process model events, pace audio output). + +Notable points specific to Hydra: + +- **Config.** `framework: smallest_hydra`, `model: {s2s: hydra, s2s_params: {...}}`. Recognised + `s2s_params`: `api_key` (**required**, Smallest key), `model` (default `hydra`), `voice` + (`wren`|`sloane`|`marlowe`|`reed`|`knox`|`tate`, default `wren`, frozen at handshake), + `generate_initial_response` (default `true` — Hydra speaks first; the system prompt is steered to + EVA's canned greeting), and `transcription` (see below). +- **Protocol.** OpenAI-Realtime-shaped JSON over one WebSocket: `session.configure` handshake; + `input_audio_buffer.append` (base64 PCM16 in); `response.output_audio.delta` (base64 PCM16 out); + client-side tools via `response.function_call_arguments.done` → reply with + `conversation.item.create` (`function_call_output`) then `response.create`. Audio is **16 kHz in / + 48 kHz out**; recording rate is 48 kHz (assistant native; user track upsampled to match). +- **Transcripts.** Hydra streams a native **assistant** transcript + (`response.output_audio_transcript.delta`, accumulated per response and finalized on + `response.done`) — used directly, since it is more accurate than re-transcribing the audio. + It sends **no user transcript**, so each completed user utterance is batch-transcribed by a + configurable STT (`s2s_transcription.py`) and appended to the audit log; the log's + timestamp-sort keeps these asynchronous writes in chronological order. The `transcription` + block selects the provider (`smallest` default — Pulse-pro for English, Pulse otherwise, keyed + on the Hydra `api_key`; or `openai`/`deepgram` with their own `api_key`). User transcription is + **fail-soft**: an STT error drops that turn's text but never aborts the conversation. +- **Latency / tokens.** `model_response` latency is emitted on the first audio delta per turn, + anchored on Hydra's `speech_stopped` (the simulator's `user_speech_stop` races), with a 50 ms + floor; token usage is read from the `response.done` `usage` block. +- **Payload ceiling (Smallest-side).** Hydra caps the `session.configure` payload (~32 KB as of + 2026-07); above it, it silently drops audio output. Instructions are truncated to fit, but for + **tool-heavy domains the tool-schema JSON alone can exceed the ceiling** (e.g. ITSM's 59 tools ≈ + 45 KB), which no instruction truncation can fix — those domains stay degraded until the limit + clears their tool schemas. Airline (~22 KB total) fits comfortably. diff --git a/src/eva/__init__.py b/src/eva/__init__.py index eb2eaa6a..d43e669c 100644 --- a/src/eva/__init__.py +++ b/src/eva/__init__.py @@ -7,7 +7,7 @@ # Bump simulation_version when changes affect benchmark outputs (agent code, # user simulator, orchestrator, simulation prompts, agent configs, tool mocks). -simulation_version = "2.0.1" +simulation_version = "2.0.3" # Bump metrics_version when changes affect metric computation (metrics code, # judge prompts, pricing tables, postprocessor). diff --git a/src/eva/assistant/audio_bridge.py b/src/eva/assistant/audio_bridge.py index af8fc9af..92eeda49 100644 --- a/src/eva/assistant/audio_bridge.py +++ b/src/eva/assistant/audio_bridge.py @@ -10,9 +10,11 @@ import audioop import base64 +import io import json import struct import time +import wave from pathlib import Path import numpy as np @@ -51,6 +53,56 @@ def mulaw_8k_to_pcm16_24k(mulaw_bytes: bytes) -> bytes: return pcm_24k +def mulaw_8k_to_pcm16_48k(mulaw_bytes: bytes) -> bytes: + """Convert 8kHz mu-law audio to 48kHz 16-bit PCM. + + Used by the Smallest Hydra S2S server, which records at 48 kHz (its native + output rate) and must upsample the 8 kHz user track to match. + """ + pcm_8k = audioop.ulaw2lin(mulaw_bytes, 2) + pcm_48k, _ = audioop.ratecv(pcm_8k, 2, 1, 8000, 48000, None) + # Clamp to exactly 6× input length so tracks stay positionally aligned. + expected_bytes = len(pcm_8k) * 6 + if len(pcm_48k) < expected_bytes: + pcm_48k = pcm_48k + b"\x00" * (expected_bytes - len(pcm_48k)) + elif len(pcm_48k) > expected_bytes: + pcm_48k = pcm_48k[:expected_bytes] + return pcm_48k + + +def pcm16_48k_to_mulaw_8k(pcm_bytes: bytes) -> bytes: + """Convert 48kHz 16-bit PCM to 8kHz mu-law. + + Uses soxr VHQ resampling for proper anti-aliasing during the 6:1 + downsampling (audioop.ratecv lacks an anti-aliasing filter and sounds + muffled). Mirrors ``pcm16_24k_to_mulaw_8k`` for Hydra's 48 kHz output. + """ + audio_data = np.frombuffer(pcm_bytes, dtype=np.int16) + resampled = soxr.resample(audio_data, 48000, 8000, quality="VHQ") + expected_samples = round(len(audio_data) * 8000 / 48000) + if len(resampled) < expected_samples: + resampled = np.pad(resampled, (0, expected_samples - len(resampled))) + elif len(resampled) > expected_samples: + resampled = resampled[:expected_samples] + pcm_8k = resampled.astype(np.int16).tobytes() + return audioop.lin2ulaw(pcm_8k, 2) + + +def pcm16_to_wav_bytes(pcm_bytes: bytes, sample_rate: int) -> bytes: + """Wrap raw 16-bit mono PCM in an in-memory WAV container. + + Used to hand a recorded utterance to a batch STT endpoint without touching + disk. 16-bit mono => 2 bytes/sample. + """ + buf = io.BytesIO() + with wave.open(buf, "wb") as wav: + wav.setnchannels(1) + wav.setsampwidth(2) + wav.setframerate(sample_rate) + wav.writeframes(pcm_bytes) + return buf.getvalue() + + def pcm16_24k_to_mulaw_8k(pcm_bytes: bytes) -> bytes: """Convert 24kHz 16-bit PCM to 8kHz mu-law. @@ -72,6 +124,19 @@ def pcm16_24k_to_mulaw_8k(pcm_bytes: bytes) -> bytes: return audioop.lin2ulaw(pcm_8k, 2) +def resample_pcm16_soxr(pcm_bytes: bytes, from_rate: int, to_rate: int) -> bytes: + """Resample 16-bit mono PCM between arbitrary rates using soxr VHQ. + + Anti-aliased (unlike audioop.ratecv), so it is safe for both up- and + down-sampling. Returns the input unchanged when the rates already match. + """ + if from_rate == to_rate or not pcm_bytes: + return pcm_bytes + audio = np.frombuffer(pcm_bytes, dtype=np.int16) + resampled = soxr.resample(audio, from_rate, to_rate, quality="VHQ") + return bytes(resampled.astype(np.int16).tobytes()) + + def sync_buffer_to_position(buffer: bytearray, target_position: int) -> None: """Pad *buffer* with silence bytes so it reaches *target_position*. diff --git a/src/eva/assistant/s2s_transcription.py b/src/eva/assistant/s2s_transcription.py new file mode 100644 index 00000000..0573ed9b --- /dev/null +++ b/src/eva/assistant/s2s_transcription.py @@ -0,0 +1,178 @@ +"""Pluggable batch transcription for S2S frameworks that emit no transcript. + +Some speech-to-speech models (notably Smallest Hydra) stream audio in both +directions but never put text on the wire. EVA's audit log — which nearly all +accuracy/experience metrics read — still needs a per-turn transcript, so the +server transcribes each completed utterance (the PCM it sent or received) with a +separate batch STT call. + +The provider is configurable via the ``transcription`` block inside +``s2s_params``:: + + "transcription": { + "provider": "smallest" | "openai" | "deepgram", + "model": "...", # provider-specific; sensible default per provider + "api_key": "...", # falls back to the S2S api_key (smallest) / env + "language": "en", # falls back to the run language + "base_url": "..." # optional override + } + +All providers are called over a single shared async ``httpx`` client so the +module has no heavy SDK import surface and is straightforward to mock in tests. +""" + +from __future__ import annotations + +import os +from typing import Any + +import httpx + +from eva.assistant.audio_bridge import pcm16_to_wav_bytes +from eva.utils.logging import get_logger + +logger = get_logger(__name__) + +_SMALLEST_STT_URL = "https://api.smallest.ai/waves/v1/stt/" +_OPENAI_STT_URL = "https://api.openai.com/v1/audio/transcriptions" +_DEEPGRAM_STT_URL = "https://api.deepgram.com/v1/listen" + +# Default model per provider. Smallest Pulse-pro is English-only; for any other +# language we fall back to the multilingual Pulse model. +_DEFAULT_MODELS = { + "smallest": "pulse-pro", + "openai": "whisper-1", + "deepgram": "nova-3", +} + + +class BatchTranscriber: + """Transcribe completed utterances via a configurable batch STT provider.""" + + def __init__( + self, + provider: str, + model: str, + api_key: str, + language: str, + base_url: str | None = None, + timeout: float = 30.0, + ) -> None: + self.provider = provider + self.model = model + self.api_key = api_key + self.language = language + self.base_url = base_url + self._client = httpx.AsyncClient(timeout=timeout) + + async def transcribe(self, pcm: bytes, sample_rate: int) -> str: + """Transcribe raw 16-bit mono PCM. Returns "" on any failure (fail-soft). + + Transcription errors must never abort a conversation — a dropped + transcript degrades that turn's text metrics but the audio recording and + tool calls are preserved. + """ + if not pcm: + return "" + wav = pcm16_to_wav_bytes(pcm, sample_rate) + try: + if self.provider == "smallest": + return await self._transcribe_smallest(wav) + if self.provider == "openai": + return await self._transcribe_openai(wav) + if self.provider == "deepgram": + return await self._transcribe_deepgram(wav) + logger.error(f"Unknown transcription provider: {self.provider!r}") + return "" + except Exception as e: # noqa: BLE001 - fail-soft by design + logger.error(f"{self.provider} transcription failed: {e}", exc_info=True) + return "" + + async def _transcribe_smallest(self, wav: bytes) -> str: + url = self.base_url or _SMALLEST_STT_URL + resp = await self._client.post( + url, + params={"model": self.model, "language": self.language}, + headers={ + "Authorization": f"Bearer {self.api_key}", + "Content-Type": "application/octet-stream", + }, + content=wav, + ) + resp.raise_for_status() + return (resp.json().get("transcription") or "").strip() + + async def _transcribe_openai(self, wav: bytes) -> str: + url = self.base_url or _OPENAI_STT_URL + resp = await self._client.post( + url, + headers={"Authorization": f"Bearer {self.api_key}"}, + files={"file": ("audio.wav", wav, "audio/wav")}, + data={"model": self.model, "language": self.language}, + ) + resp.raise_for_status() + return (resp.json().get("text") or "").strip() + + async def _transcribe_deepgram(self, wav: bytes) -> str: + url = self.base_url or _DEEPGRAM_STT_URL + resp = await self._client.post( + url, + params={"model": self.model, "language": self.language, "smart_format": "true"}, + headers={ + "Authorization": f"Token {self.api_key}", + "Content-Type": "audio/wav", + }, + content=wav, + ) + resp.raise_for_status() + alternatives = resp.json()["results"]["channels"][0]["alternatives"] + return (alternatives[0]["transcript"] if alternatives else "").strip() + + async def aclose(self) -> None: + await self._client.aclose() + + +def create_transcriber(s2s_params: dict[str, Any], language: str) -> BatchTranscriber: + """Build a :class:`BatchTranscriber` from ``s2s_params['transcription']``. + + Defaults to the Smallest Pulse provider keyed on the S2S ``api_key`` so an + in-stack transcript works with no extra configuration. The Pulse-pro model is + English-only; for any non-English run the default model falls back to the + multilingual ``pulse`` model. + """ + cfg = dict(s2s_params.get("transcription") or {}) + provider = cfg.get("provider", "smallest") + lang = cfg.get("language") or language or "en" + + model = cfg.get("model") + if not model: + model = _DEFAULT_MODELS.get(provider, "") + # Smallest Pulse-pro is English-only; use multilingual Pulse otherwise. + if provider == "smallest" and not lang.startswith("en"): + model = "pulse" + + # API key resolution: explicit transcription key, else a sensible fallback — + # the shared S2S key for Smallest, or the provider's standard env var. + api_key = cfg.get("api_key") + if not api_key: + if provider == "smallest": + api_key = s2s_params.get("api_key", "") + elif provider == "openai": + api_key = os.environ.get("OPENAI_API_KEY", "") + elif provider == "deepgram": + api_key = os.environ.get("DEEPGRAM_API_KEY", "") + if not api_key: + raise ValueError( + f"No API key for transcription provider {provider!r}. " + f"Set s2s_params.transcription.api_key (or s2s_params.api_key for 'smallest', " + f"OPENAI_API_KEY for 'openai', DEEPGRAM_API_KEY for 'deepgram')." + ) + + logger.info(f"S2S transcription: provider={provider} model={model} language={lang}") + return BatchTranscriber( + provider=provider, + model=model, + api_key=api_key, + language=lang, + base_url=cfg.get("base_url"), + ) diff --git a/src/eva/assistant/smallest_hydra_server.py b/src/eva/assistant/smallest_hydra_server.py new file mode 100644 index 00000000..bc62dca9 --- /dev/null +++ b/src/eva/assistant/smallest_hydra_server.py @@ -0,0 +1,723 @@ +"""Smallest Hydra speech-to-speech AssistantServer for EVA-Bench. + +Bridges between the Twilio-framed WebSocket (user simulator) and Smallest's +Hydra S2S model over a raw WebSocket. Audio flows: + + User simulator (8 kHz mulaw) + -> 16 kHz PCM16 -> Hydra input (input_audio_buffer.append) + Hydra output (48 kHz PCM16, response.output_audio.delta) + -> 8 kHz mulaw -> User simulator + +Hydra's wire protocol is OpenAI-Realtime-shaped: a ``session.configure`` +handshake, ``input_audio_buffer.append`` for audio in, ``response.output_audio +.delta`` for audio out, and ``response.function_call_arguments.done`` / +``conversation.item.create`` (function_call_output) for client-side tools. + +Transcripts: Hydra streams a native **assistant** transcript +(``response.output_audio_transcript.delta``), which we use directly — it is more +accurate than re-transcribing the audio. It emits no **user** transcript, so each +completed user utterance is batch-transcribed with a configurable STT provider +(see ``s2s_transcription.py``). The audit log is timestamp-sorted, so the +asynchronous user transcriptions land in the right order regardless of timing. +""" + +from __future__ import annotations + +import asyncio +import base64 +import contextlib +import json +import time +from pathlib import Path +from typing import Any + +import uvicorn +import websockets +from fastapi import FastAPI, WebSocket, WebSocketDisconnect + +from eva.assistant.audio_bridge import ( + FrameworkLogWriter, + MetricsLogWriter, + create_twilio_media_message, + mulaw_8k_to_pcm16_16k, + mulaw_8k_to_pcm16_48k, + parse_twilio_media_message, + pcm16_48k_to_mulaw_8k, + resample_pcm16_soxr, + sync_buffer_to_position, +) +from eva.assistant.base_server import AbstractAssistantServer +from eva.assistant.s2s_transcription import BatchTranscriber, create_transcriber +from eva.models.agents import AgentConfig +from eva.models.config import ModelConfig +from eva.utils.logging import get_logger +from eva.utils.prompt_manager import PromptManager + +logger = get_logger(__name__) + +# Canonical recording rate. Both tracks (user + assistant) are kept at this +# rate for mixing. Hydra's *output* rate varies by model — V1 emits 48 kHz, +# V1.1 emits 24 kHz — so we read ``output_audio_sample_rate`` from the +# session.configured handshake and resample the assistant track up to this rate. +_RECORDING_SAMPLE_RATE = 48000 +# Rate Hydra expects for input audio. +_HYDRA_INPUT_RATE = 16000 +# Skip batch-transcribing user utterances shorter than this — they are usually +# near-silence that STT models hallucinate full sentences on (16-bit mono => 2 +# bytes/sample, so 300 ms @ 16 kHz = 9600 bytes). +_MIN_USER_UTTERANCE_BYTES = int(_HYDRA_INPUT_RATE * 0.3) * 2 + +_HYDRA_WS_URL = "wss://api.smallest.ai/waves/v1/s2s" +# Voice sets differ by model. V1 (model=hydra) and V1.1 (model=hydra-v1.1) +# expose disjoint voice names. +_VOICES_V1 = {"wren", "sloane", "marlowe", "reed", "knox", "tate"} +_VOICES_V11 = {"zoe", "maya", "elena", "ivy", "grace", "alex", "aria", "leo", "sam", "kai"} +_VALID_VOICES = _VOICES_V1 | _VOICES_V11 + +# Audio output pacing: 160-byte mulaw chunks (20 ms at 8 kHz) at real-time rate +# so the user simulator's silence detection works correctly. +MULAW_CHUNK_SIZE = 160 +MULAW_CHUNK_DURATION_S = 0.02 + + +# --------------------------------------------------------------------------- +# Tool schema helper +# --------------------------------------------------------------------------- + + +def _agent_tools_to_hydra(agent: AgentConfig) -> list[dict[str, Any]] | None: + """Convert EVA AgentConfig tools to Hydra ``session.tools`` (client-side). + + Hydra uses OpenAI-style function declarations. Executing them client-side + (no endpoint) makes Hydra emit ``response.function_call_arguments.done``. + """ + if not agent.tools: + return None + + functions: list[dict[str, Any]] = [] + for tool in agent.tools: + functions.append( + { + "type": "function", + "name": tool.function_name, + "description": f"{tool.name}: {tool.description}", + "parameters": { + "type": "object", + "properties": tool.get_parameter_properties(), + "required": tool.get_required_param_names(), + }, + } + ) + return functions or None + + +# --------------------------------------------------------------------------- +# Smallest Hydra AssistantServer +# --------------------------------------------------------------------------- + + +class SmallestHydraAssistantServer(AbstractAssistantServer): + """Bridges the Twilio WebSocket <-> Smallest Hydra S2S API.""" + + def __init__( + self, + current_date_time: str, + pipeline_config: ModelConfig, + agent: AgentConfig, + agent_config_path: str, + scenario_db_path: str, + output_dir: Path, + port: int, + conversation_id: str, + language: str = "en", + ): + super().__init__( + current_date_time=current_date_time, + pipeline_config=pipeline_config, + agent=agent, + agent_config_path=agent_config_path, + scenario_db_path=scenario_db_path, + output_dir=output_dir, + port=port, + conversation_id=conversation_id, + language=language, + ) + + self._audio_sample_rate = _RECORDING_SAMPLE_RATE + # Hydra's true output rate, learned from session.configured. Default to + # the recording rate (no resample) until the handshake tells us otherwise. + self._hydra_output_rate = _RECORDING_SAMPLE_RATE + + s2s_params = self.pipeline_config.s2s_params or {} + self._model = s2s_params.get("model", "hydra") + self._api_key = s2s_params.get("api_key", "") + if not self._api_key: + raise ValueError("Missing Smallest API key in s2s_params['api_key']") + + self._voice = s2s_params.get("voice", "wren") + expected_voices = _VOICES_V11 if "v1.1" in self._model else _VOICES_V1 + if self._voice not in expected_voices: + logger.warning( + f"Voice {self._voice!r} is not a known {self._model} voice; valid voices: {sorted(expected_voices)}" + ) + self._generate_initial_response = bool(s2s_params.get("generate_initial_response", True)) + + # Build system prompt (same pattern as the other realtime/S2S servers). + prompt_manager = PromptManager() + self._system_prompt = prompt_manager.get_prompt( + "realtime_agent.system_prompt", + agent_personality=agent.description, + agent_instructions=agent.instructions, + datetime=self.current_date_time, + ) + self._hydra_tools = _agent_tools_to_hydra(agent) + + # Hydra generates its own opening line; steer it to EVA's canned greeting + # so the conversation opens consistently with the other frameworks. + greeting_steer = ( + f"\n\nBegin the conversation by greeting the user with: {self.initial_message}" + if self._generate_initial_response + else "" + ) + + # Hydra silently drops audio output when the session.configure payload + # exceeds its limit (~32 KB as of 2026-07), and also when total context + # (instructions + tools + conversation history) grows too large on + # subsequent turns. Cap the instructions so that instructions + tools + # JSON fits comfortably in a single session.configure message and leaves + # headroom for conversation. + _MAX_PAYLOAD_BYTES = 32_000 + _tools_bytes = len(json.dumps(self._hydra_tools)) if self._hydra_tools else 0 + # Reserve room for the greeting steer so it is never truncated away. + max_prompt_chars = max(1000, _MAX_PAYLOAD_BYTES - _tools_bytes - len(greeting_steer) - 500) + if len(self._system_prompt) > max_prompt_chars: + logger.warning( + f"Hydra: truncating system prompt from {len(self._system_prompt)} to " + f"{max_prompt_chars} chars ({len(self._hydra_tools or [])} tools take " + f"{_tools_bytes} bytes) to stay within Hydra's context budget" + ) + self._system_prompt = self._system_prompt[:max_prompt_chars] + # Append the greeting AFTER truncation so it always survives. + self._system_prompt += greeting_steer + + # Per-turn batch transcription (Hydra emits no transcript on the wire). + self._transcriber: BatchTranscriber = create_transcriber(s2s_params, self.language) + self._pending_transcriptions: list[asyncio.Task] = [] + + # ------------------------------------------------------------------ + # Server lifecycle + # ------------------------------------------------------------------ + + async def start(self) -> None: + """Start the FastAPI WebSocket server (non-blocking).""" + if self._running: + logger.warning("Server already running") + return + + self.output_dir.mkdir(parents=True, exist_ok=True) + self._fw_log = FrameworkLogWriter(self.output_dir) + self._metrics_log = MetricsLogWriter(self.output_dir) + + self._app = FastAPI() + + @self._app.websocket("/ws") + async def websocket_endpoint(websocket: WebSocket): + await websocket.accept() + await self._handle_session(websocket) + + @self._app.websocket("/") + async def websocket_root(websocket: WebSocket): + await websocket.accept() + await self._handle_session(websocket) + + config = uvicorn.Config( + self._app, + host="0.0.0.0", + port=self.port, + log_level="warning", + lifespan="off", + ) + self._server = uvicorn.Server(config) + self._running = True + self._server_task = asyncio.create_task(self._server.serve()) + + while not self._server.started: + await asyncio.sleep(0.01) + + logger.info(f"Smallest Hydra server started on ws://localhost:{self.port}") + + async def _shutdown(self) -> None: + """Stop the Hydra server.""" + if not self._running: + return + self._running = False + + await self._transcriber.aclose() + + if self._server: + self._server.should_exit = True + if self._server_task: + try: + await asyncio.wait_for(self._server_task, timeout=5.0) + except TimeoutError: + self._server_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await self._server_task + except (asyncio.CancelledError, KeyboardInterrupt): + pass + self._server = None + self._server_task = None + + logger.info(f"Smallest Hydra server stopped on port {self.port}") + + # ------------------------------------------------------------------ + # Session handler + # ------------------------------------------------------------------ + + async def _handle_session(self, websocket: WebSocket) -> None: # noqa: C901 + """Bridge a single Twilio WebSocket session with Hydra.""" + logger.info("Client connected to Smallest Hydra server") + + stream_sid: str = self.conversation_id + twilio_connected = True + + # Per-turn state (shared across tasks via nonlocal). + _in_model_turn = False + _user_speaking = False + _user_speech_start_ts: str | None = None # simulator VAD: user speech start (wall ms) + _user_speech_stop_ts: str | None = None # simulator VAD: user speech stop (wall ms) + # Hydra's own end-of-user-speech (wall ms when we receive speech_stopped). + # Preferred latency reference: Hydra responds within ~10 ms of this, before + # the simulator's local user_speech_stop fires, so relying on the simulator + # event alone drops nearly every turn's latency (same race as ElevenLabs). + _hydra_speech_stop_wall_ms: str | None = None + _assistant_turn_start_ts: str | None = None # wall ms of first audio chunk this turn + + # User utterances have no native transcript, so we batch-STT the audio we + # sent to Hydra. The assistant transcript comes natively from Hydra's + # response.output_audio_transcript.delta stream (more accurate than STT). + _user_utt_pcm = bytearray() # 16 kHz PCM sent to Hydra for the current user utterance + _asst_text = "" # accumulated native assistant transcript for the current response + + # Outbound mulaw chunks; the pacer drains at real-time rate so the event + # processor never blocks on sleep. + audio_output_queue: asyncio.Queue[bytes] = asyncio.Queue() + + url = f"{_HYDRA_WS_URL}?model={self._model}&api_key={self._api_key}" + + def _transcribe_user_async(pcm: bytes, ts: str | None) -> None: + """Spawn a background STT of a user utterance that appends to the audit log.""" + if len(pcm) < _MIN_USER_UTTERANCE_BYTES: + # Too short to be real speech; skip to avoid STT hallucinations. + return + stamp = ts or str(int(round(time.time() * 1000))) + + async def _run() -> None: + text = await self._transcriber.transcribe(pcm, _HYDRA_INPUT_RATE) + if not text: + return + logger.info(f"User transcript: {text}") + self.audit_log.append_user_input(text, timestamp_ms=stamp) + + self._pending_transcriptions.append(asyncio.create_task(_run())) + + try: + async with websockets.connect(url, max_size=None) as hydra_ws: + logger.info(f"Hydra session connected (model={self._model}, voice={self._voice})") + + # Handshake: server sends session.created, then we configure. + with contextlib.suppress(TimeoutError): + created = await asyncio.wait_for(hydra_ws.recv(), timeout=10.0) + logger.info(f"Hydra handshake: {created!r}") + + session: dict[str, Any] = { + "instructions": self._system_prompt, + "voice": self._voice, + "generate_initial_response": self._generate_initial_response, + "input_audio_format": "pcm16", + "output_audio_format": "pcm16", + } + if self._hydra_tools: + session["tools"] = self._hydra_tools + configure_msg = json.dumps({"type": "session.configure", "session": session}) + logger.info( + f"Sending session.configure ({len(configure_msg)} bytes, {len(session.get('tools', []))} tools)" + ) + await hydra_ws.send(configure_msg) + + # Wait for session.configured before starting concurrent tasks + with contextlib.suppress(TimeoutError): + configured_raw = await asyncio.wait_for(hydra_ws.recv(), timeout=10.0) + configured_msg = json.loads(configured_raw) + sess = configured_msg.get("session", {}) + key_fields = {k: v for k, v in sess.items() if k not in ("instructions", "tools")} + logger.info(f"Hydra session.configured: {key_fields}, tools_count={len(sess.get('tools', []))}") + # Output rate varies by model (V1=48 kHz, V1.1=24 kHz). Read + # it here so the assistant track is recorded at correct pitch. + reported_rate = sess.get("output_audio_sample_rate") + if reported_rate: + self._hydra_output_rate = int(reported_rate) + if self._hydra_output_rate != _RECORDING_SAMPLE_RATE: + logger.info( + f"Hydra output is {self._hydra_output_rate} Hz; resampling " + f"assistant audio to {_RECORDING_SAMPLE_RATE} Hz for recording." + ) + + if self._generate_initial_response: + self._fw_log.turn_start() + + # ----- Concurrent tasks ----- + + async def _forward_user_audio() -> None: + """Read Twilio WS messages, convert audio, send to Hydra.""" + nonlocal stream_sid, twilio_connected + nonlocal _user_speech_start_ts, _user_speech_stop_ts, _user_speaking, _in_model_turn + nonlocal _last_audio_sent + + def _flush_user_utterance() -> None: + nonlocal _user_utt_pcm + if _user_utt_pcm: + _transcribe_user_async(bytes(_user_utt_pcm), _user_speech_start_ts) + _user_utt_pcm = bytearray() + + try: + while twilio_connected and self._running: + try: + raw = await asyncio.wait_for(websocket.receive_text(), timeout=1.0) + except TimeoutError: + continue + + try: + msg = json.loads(raw) + except json.JSONDecodeError: + continue + + event = msg.get("event") + if event == "start": + stream_sid = msg.get("start", {}).get("streamSid", stream_sid) + logger.info(f"Twilio stream started: {stream_sid}") + elif event == "stop": + logger.info("Twilio stream stopped") + twilio_connected = False + break + elif event == "user_speech_start": + # New utterance: flush any unfinished prior one first + # (the simulator's stop event occasionally races). + _flush_user_utterance() + _user_speech_start_ts = msg.get("timestamp_ms") + _user_speaking = True + _in_model_turn = False + logger.info(f"User speech start: {_user_speech_start_ts}") + elif event == "user_speech_stop": + _user_speech_stop_ts = msg.get("timestamp_ms") + _user_speaking = False + logger.info(f"User speech stop: {_user_speech_stop_ts}") + _flush_user_utterance() + elif event == "media": + mulaw_bytes = parse_twilio_media_message(raw) + if mulaw_bytes is None: + continue + + # Record user audio at 48 kHz (recording rate). + pcm_48k = mulaw_8k_to_pcm16_48k(mulaw_bytes) + if not _in_model_turn: + sync_buffer_to_position(self.assistant_audio_buffer, len(self.user_audio_buffer)) + self.user_audio_buffer.extend(pcm_48k) + + # Accumulate the 16 kHz copy for utterance transcription. + pcm_16k = mulaw_8k_to_pcm16_16k(mulaw_bytes) + _user_utt_pcm.extend(pcm_16k) + + # Send 16 kHz PCM16 to Hydra. + await hydra_ws.send( + json.dumps( + { + "type": "input_audio_buffer.append", + "audio": base64.b64encode(pcm_16k).decode("ascii"), + } + ) + ) + _last_audio_sent = time.monotonic() + except WebSocketDisconnect: + logger.info("Twilio WebSocket disconnected") + twilio_connected = False + except asyncio.CancelledError: + pass + except Exception as e: + logger.error(f"Error in user audio forwarder: {e}", exc_info=True) + finally: + _flush_user_utterance() + twilio_connected = False + + async def _pace_audio_output() -> None: + """Drain audio_output_queue and forward chunks at real-time rate.""" + nonlocal twilio_connected + next_send_time = time.monotonic() + try: + while self._running: + try: + chunk = await asyncio.wait_for(audio_output_queue.get(), timeout=1.0) + except TimeoutError: + continue + + try: + await websocket.send_text(create_twilio_media_message(stream_sid, chunk)) + except Exception: + twilio_connected = False + return + + now = time.monotonic() + if next_send_time <= now: + next_send_time = now + next_send_time += MULAW_CHUNK_DURATION_S + sleep_duration = next_send_time - time.monotonic() + if sleep_duration > 0: + await asyncio.sleep(sleep_duration) + except asyncio.CancelledError: + pass + + async def _process_hydra_events() -> None: # noqa: C901 + """Consume events from the Hydra WebSocket.""" + nonlocal _in_model_turn, _user_speaking + nonlocal _user_speech_stop_ts, _hydra_speech_stop_wall_ms + nonlocal _assistant_turn_start_ts, _asst_text + nonlocal twilio_connected + + def _flush_assistant(interrupted: bool) -> None: + nonlocal _asst_text, _in_model_turn, _assistant_turn_start_ts + # Collapse the phrase-level deltas Hydra streams into one line. + text = " ".join(_asst_text.split()) + if text: + stamp = _assistant_turn_start_ts or str(int(round(time.time() * 1000))) + logger.info(f"Assistant transcript: {text}") + label = f"{text} [interrupted]" if interrupted else text + self.audit_log.append_assistant_output(label, timestamp_ms=stamp) + if interrupted: + self._fw_log.s2s_transcript(text) + else: + self._fw_log.llm_response(text) + self._fw_log.turn_end(was_interrupted=interrupted) + _asst_text = "" + _in_model_turn = False + _assistant_turn_start_ts = None + + try: + while self._running: + try: + raw = await asyncio.wait_for(hydra_ws.recv(), timeout=2.0) + except TimeoutError: + continue + except websockets.exceptions.ConnectionClosed: + logger.info("Hydra WebSocket closed") + break + + try: + msg = json.loads(raw) + except (json.JSONDecodeError, TypeError): + if isinstance(raw, bytes): + logger.debug(f"Hydra binary frame: {len(raw)} bytes") + else: + logger.debug(f"Hydra non-JSON text: {raw[:200]!r}") + continue + + etype = msg.get("type") + if etype in ("response.output_audio.delta", "response.output_audio_transcript.delta"): + logger.debug(f"Hydra {etype} ({len(msg.get('delta', ''))} chars)") + elif etype == "session.configured": + sess = msg.get("session", {}) + key_fields = {k: v for k, v in sess.items() if k not in ("instructions", "tools")} + logger.info( + f"Hydra session.configured: {key_fields}, tools={len(sess.get('tools', []))}" + ) + else: + logger.info(f"Hydra event: {etype} — {json.dumps(msg, ensure_ascii=False)[:500]}") + + if etype == "response.output_audio.delta": + pcm_native = base64.b64decode(msg.get("delta", "")) + if len(pcm_native) < 6: + continue + # Resample from Hydra's true output rate up to the + # canonical recording rate (no-op when they match, e.g. + # V1). Keeps pitch/duration correct and the assistant + # track aligned with the 48 kHz user track. + pcm_48k = resample_pcm16_soxr( + pcm_native, self._hydra_output_rate, _RECORDING_SAMPLE_RATE + ) + + if not _in_model_turn: + _in_model_turn = True + _assistant_turn_start_ts = str(int(round(time.time() * 1000))) + self._fw_log.turn_start() + # Model response latency: user speech end -> first audio. + # Prefer Hydra's speech_stopped (fires just before the + # response); fall back to the simulator's user_speech_stop. + # Absent on the initial greeting turn (model-initiated). + _user_end_ref = _hydra_speech_stop_wall_ms or _user_speech_stop_ts + if _user_end_ref and self._metrics_log: + latency_ms = int(_assistant_turn_start_ts) - int(_user_end_ref) + # Floor at 50 ms: Hydra occasionally emits speech_stopped + # only after it has begun generating, collapsing the gap to + # ~0 — sub-RTT values aren't real response latency. + if 50 <= latency_ms < 30_000: + self._metrics_log.write_latency( + "model_response", latency_ms / 1000, self._model + ) + _user_speech_stop_ts = None + _hydra_speech_stop_wall_ms = None + + # Record assistant track (canonical 48 kHz). + if not _user_speaking: + sync_buffer_to_position(self.user_audio_buffer, len(self.assistant_audio_buffer)) + self.assistant_audio_buffer.extend(pcm_48k) + + # Convert to 8 kHz mulaw and enqueue in 20 ms chunks. + if twilio_connected: + try: + mulaw = pcm16_48k_to_mulaw_8k(pcm_48k) + except Exception as conv_err: + logger.warning(f"Audio conversion error ({len(pcm_48k)} bytes): {conv_err}") + continue + offset = 0 + while offset < len(mulaw): + await audio_output_queue.put(mulaw[offset : offset + MULAW_CHUNK_SIZE]) + offset += MULAW_CHUNK_SIZE + + elif etype == "response.output_audio_transcript.delta": + # Native assistant transcript (more accurate than STT). + delta = msg.get("delta", "") + if delta: + _asst_text = f"{_asst_text} {delta}" if _asst_text else delta + + elif etype == "input_audio_buffer.speech_started": + # User barged in; Hydra cancels its in-flight response. + _user_speaking = True + + elif etype == "input_audio_buffer.speech_stopped": + _user_speaking = False + # Anchor for model-response latency (see note at declaration). + _hydra_speech_stop_wall_ms = str(int(round(time.time() * 1000))) + + elif etype == "response.function_call_arguments.done": + name = msg.get("name", "") + call_id = msg.get("call_id", "") + try: + arguments = json.loads(msg.get("arguments") or "{}") + except json.JSONDecodeError: + arguments = {} + logger.info(f"Tool call: {name}({json.dumps(arguments, ensure_ascii=False)})") + result = await self.execute_tool(name, arguments) + await hydra_ws.send( + json.dumps( + { + "type": "conversation.item.create", + "item": { + "type": "function_call_output", + "role": "user", + "call_id": call_id, + "output": json.dumps(result, ensure_ascii=False), + }, + } + ) + ) + # Ask Hydra to narrate the tool result. + await hydra_ws.send(json.dumps({"type": "response.create", "response": {}})) + + elif etype == "response.done": + response = msg.get("response", {}) or {} + status = response.get("status", "completed") + reason = (response.get("status_details") or {}).get("reason", "") + interrupted = status in ("cancelled", "incomplete") or reason in ( + "interrupted", + "client_cancelled", + ) + if _in_model_turn or _asst_text: + _flush_assistant(interrupted) + + usage = response.get("usage") or {} + prompt_tokens = usage.get("input_tokens", 0) or 0 + completion_tokens = usage.get("output_tokens", 0) or 0 + if (prompt_tokens or completion_tokens) and self._metrics_log: + self._metrics_log.write_token_usage( + processor="smallest_hydra", + model=self._model, + prompt_tokens=prompt_tokens, + completion_tokens=completion_tokens, + ) + + elif etype == "error": + err = msg.get("error", {}) + logger.error(f"Hydra error: {err.get('code')} - {err.get('message')}") + + except asyncio.CancelledError: + pass + except Exception as e: + logger.error(f"Error in Hydra event processor: {e}", exc_info=True) + + # Hydra closes the socket after ~30 s of silence. + # Send silence frames every few seconds to keep it alive. + _KEEPALIVE_INTERVAL = 5.0 # seconds + _SILENCE_16K = base64.b64encode(b"\x00" * 3200).decode("ascii") # 100 ms + _last_audio_sent = time.monotonic() + + async def _hydra_keepalive() -> None: + """Send silence to Hydra when no user audio is flowing.""" + nonlocal _last_audio_sent + try: + while self._running: + await asyncio.sleep(_KEEPALIVE_INTERVAL) + elapsed = time.monotonic() - _last_audio_sent + if elapsed >= _KEEPALIVE_INTERVAL: + await hydra_ws.send( + json.dumps({"type": "input_audio_buffer.append", "audio": _SILENCE_16K}) + ) + _last_audio_sent = time.monotonic() + except asyncio.CancelledError: + pass + except Exception: + pass # non-critical; let other tasks drive shutdown + + user_task = asyncio.create_task(_forward_user_audio()) + hydra_task = asyncio.create_task(_process_hydra_events()) + pacer_task = asyncio.create_task(_pace_audio_output()) + keepalive_task = asyncio.create_task(_hydra_keepalive()) + + done, pending = await asyncio.wait( + [user_task, hydra_task, pacer_task, keepalive_task], + return_when=asyncio.FIRST_COMPLETED, + ) + + def _task_name(t: asyncio.Task) -> str: + if t is user_task: + return "user_audio" + if t is hydra_task: + return "hydra_events" + if t is keepalive_task: + return "keepalive" + return "audio_pacer" + + for task in done: + exc = task.exception() if not task.cancelled() else None + if exc: + logger.error(f"Task '{_task_name(task)}' failed: {exc}", exc_info=exc) + else: + logger.info(f"Task '{_task_name(task)}' completed normally") + + for task in pending: + logger.info(f"Cancelling pending task '{_task_name(task)}'") + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task + + except Exception as e: + logger.error(f"Smallest Hydra session error: {e}", exc_info=True) + finally: + # Let outstanding transcriptions finish so the audit log is complete + # before stop()/save_outputs() runs. + if self._pending_transcriptions: + logger.info(f"Awaiting {len(self._pending_transcriptions)} pending transcription(s)") + with contextlib.suppress(TimeoutError): + await asyncio.wait_for( + asyncio.gather(*self._pending_transcriptions, return_exceptions=True), + timeout=60.0, + ) + logger.info("Client disconnected from Smallest Hydra server") diff --git a/src/eva/models/config.py b/src/eva/models/config.py index ea797e8a..80433854 100644 --- a/src/eva/models/config.py +++ b/src/eva/models/config.py @@ -484,16 +484,19 @@ class ModelDeployment(DeploymentTypedDict): ) # Framework selection - framework: Literal["pipecat", "openai_realtime", "gemini_live", "elevenlabs", "grok_voice"] = Field( - "pipecat", - description=( - "Agent framework to use for the assistant server." - "'pipecat' (default): Pipecat pipeline." - "'openai_realtime': OpenAI Realtime API directly." - "'gemini_live': Gemini Live API via google-genai." - "'elevenlabs': ElevenLabs Conversational AI API." - "'grok_voice': xAI Grok voice realtime API." - ), + framework: Literal["pipecat", "openai_realtime", "gemini_live", "elevenlabs", "grok_voice", "smallest_hydra"] = ( + Field( + "pipecat", + description=( + "Agent framework to use for the assistant server." + "'pipecat' (default): Pipecat pipeline." + "'openai_realtime': OpenAI Realtime API directly." + "'gemini_live': Gemini Live API via google-genai." + "'elevenlabs': ElevenLabs Conversational AI API." + "'grok_voice': xAI Grok voice realtime API." + "'smallest_hydra': Smallest Hydra speech-to-speech API." + ), + ) ) # Run identifier diff --git a/src/eva/orchestrator/worker.py b/src/eva/orchestrator/worker.py index 8183ff81..c1baf139 100644 --- a/src/eva/orchestrator/worker.py +++ b/src/eva/orchestrator/worker.py @@ -49,10 +49,14 @@ def _get_server_class(framework: str) -> type[AbstractAssistantServer]: from eva.assistant.grok_voice_server import GrokVoiceAssistantServer return GrokVoiceAssistantServer + elif framework == "smallest_hydra": + from eva.assistant.smallest_hydra_server import SmallestHydraAssistantServer + + return SmallestHydraAssistantServer else: raise ValueError( f"Unknown framework: {framework!r}. " - "Supported: pipecat, openai_realtime, gemini_live, elevenlabs, grok_voice" + "Supported: pipecat, openai_realtime, gemini_live, elevenlabs, grok_voice, smallest_hydra" ) diff --git a/tests/unit/assistant/test_smallest_hydra_server.py b/tests/unit/assistant/test_smallest_hydra_server.py new file mode 100644 index 00000000..871084bb --- /dev/null +++ b/tests/unit/assistant/test_smallest_hydra_server.py @@ -0,0 +1,150 @@ +"""Tests for the Smallest Hydra S2S server helpers and transcription plumbing.""" + +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from eva.assistant.audio_bridge import ( + mulaw_8k_to_pcm16_48k, + pcm16_48k_to_mulaw_8k, + pcm16_to_wav_bytes, +) +from eva.assistant.s2s_transcription import BatchTranscriber, create_transcriber +from eva.assistant.smallest_hydra_server import _agent_tools_to_hydra +from eva.models.agents import AgentConfig, AgentTool, AgentToolParameter + + +def _agent_with_tools() -> AgentConfig: + return AgentConfig( + id="a1", + name="Test Agent", + description="desc", + role="role", + instructions="be helpful", + tool_module_path="eva.assistant.tools.airline_tools", + tools=[ + AgentTool( + id="t1", + name="Lookup Booking", + description="Look up a booking", + required_parameters=[AgentToolParameter(name="booking_id", type="str", description="The booking id")], + optional_parameters=[AgentToolParameter(name="verbose", type="bool")], + ) + ], + ) + + +class TestToolConversion: + def test_no_tools_returns_none(self): + agent = MagicMock() + agent.tools = [] + assert _agent_tools_to_hydra(agent) is None + + def test_openai_style_function_declaration(self): + functions = _agent_tools_to_hydra(_agent_with_tools()) + assert functions is not None and len(functions) == 1 + fn = functions[0] + assert fn["type"] == "function" + assert fn["name"] # function_name derived from the tool + assert "Lookup Booking" in fn["description"] + params = fn["parameters"] + assert params["type"] == "object" + assert "booking_id" in params["properties"] + assert params["required"] == ["booking_id"] + + +class TestAudioHelpers: + def test_mulaw_round_trip_preserves_sample_count(self): + # One 20 ms mulaw chunk: 160 samples @ 8 kHz. + mulaw = b"\xff" * 160 + pcm_48k = mulaw_8k_to_pcm16_48k(mulaw) + assert len(pcm_48k) == 160 * 6 * 2 # 6x upsample, 16-bit + back = pcm16_48k_to_mulaw_8k(pcm_48k) + assert len(back) == 160 # exact inverse sample count + + def test_pcm16_to_wav_has_riff_header(self): + wav = pcm16_to_wav_bytes(b"\x00\x00" * 100, 48000) + assert wav[:4] == b"RIFF" + assert wav[8:12] == b"WAVE" + + +class TestCreateTranscriber: + def test_default_is_smallest_pulse_pro_reusing_s2s_key(self): + t = create_transcriber({"api_key": "SMALL"}, "en") + assert (t.provider, t.model, t.api_key, t.language) == ("smallest", "pulse-pro", "SMALL", "en") + + def test_smallest_non_english_falls_back_to_multilingual_pulse(self): + t = create_transcriber({"api_key": "SMALL", "transcription": {"language": "de"}}, "de") + assert t.model == "pulse" + + def test_explicit_provider_and_model(self): + t = create_transcriber( + {"api_key": "X", "transcription": {"provider": "openai", "model": "gpt-4o-transcribe", "api_key": "OAI"}}, + "en", + ) + assert (t.provider, t.model, t.api_key) == ("openai", "gpt-4o-transcribe", "OAI") + + def test_openai_default_model(self): + t = create_transcriber({"transcription": {"provider": "openai", "api_key": "OAI"}}, "en") + assert t.model == "whisper-1" + + def test_missing_key_for_non_smallest_provider_raises(self, monkeypatch): + monkeypatch.delenv("OPENAI_API_KEY", raising=False) + with pytest.raises(ValueError, match="No API key"): + create_transcriber({"api_key": "SMALL", "transcription": {"provider": "openai"}}, "en") + + def test_openai_falls_back_to_env_key(self, monkeypatch): + monkeypatch.setenv("OPENAI_API_KEY", "ENVOAI") + t = create_transcriber({"transcription": {"provider": "openai"}}, "en") + assert t.api_key == "ENVOAI" + + def test_deepgram_falls_back_to_env_key(self, monkeypatch): + monkeypatch.setenv("DEEPGRAM_API_KEY", "ENVDG") + t = create_transcriber({"transcription": {"provider": "deepgram"}}, "en") + assert t.api_key == "ENVDG" + + +def _mock_response(json_body: dict) -> MagicMock: + resp = MagicMock() + resp.raise_for_status = MagicMock() + resp.json = MagicMock(return_value=json_body) + return resp + + +class TestBatchTranscribe: + @pytest.mark.asyncio + async def test_smallest_parses_transcription_field(self): + t = BatchTranscriber("smallest", "pulse-pro", "K", "en") + t._client.post = AsyncMock(return_value=_mock_response({"transcription": " hello "})) + assert await t.transcribe(b"\x00\x00" * 10, 16000) == "hello" + await t.aclose() + + @pytest.mark.asyncio + async def test_openai_parses_text_field(self): + t = BatchTranscriber("openai", "whisper-1", "K", "en") + t._client.post = AsyncMock(return_value=_mock_response({"text": "hi there"})) + assert await t.transcribe(b"\x00\x00" * 10, 16000) == "hi there" + await t.aclose() + + @pytest.mark.asyncio + async def test_deepgram_parses_nested_transcript(self): + t = BatchTranscriber("deepgram", "nova-3", "K", "en") + body = {"results": {"channels": [{"alternatives": [{"transcript": "deep text"}]}]}} + t._client.post = AsyncMock(return_value=_mock_response(body)) + assert await t.transcribe(b"\x00\x00" * 10, 16000) == "deep text" + await t.aclose() + + @pytest.mark.asyncio + async def test_empty_pcm_skips_request(self): + t = BatchTranscriber("smallest", "pulse-pro", "K", "en") + t._client.post = AsyncMock() + assert await t.transcribe(b"", 16000) == "" + t._client.post.assert_not_called() + await t.aclose() + + @pytest.mark.asyncio + async def test_failure_is_fail_soft(self): + t = BatchTranscriber("smallest", "pulse-pro", "K", "en") + t._client.post = AsyncMock(side_effect=RuntimeError("boom")) + assert await t.transcribe(b"\x00\x00" * 10, 16000) == "" # no exception propagates + await t.aclose()