Skip to content

Commit de2077a

Browse files
committed
Add SmallestAI Hydra S2S framework support
Bridge the Twilio user-simulator WebSocket to Smallest's Hydra speech-to-speech model (framework: smallest_hydra, s2s: hydra), modeled on the Gemini Live server: three concurrent tasks (forward user audio / process Hydra events / pace output) with sync_buffer_to_position recording. Audio is 16 kHz in / 48 kHz out; recorded at 48 kHz native. - Assistant transcript comes from Hydra's native response.output_audio_transcript .delta stream (accumulated per response, finalized on response.done). Hydra emits no user transcript, so each user utterance is batch-transcribed by a configurable STT (s2s_transcription.py: smallest default / openai / deepgram), fail-soft, with a short-utterance guard to avoid hallucinations on near-silence. - model_response latency anchored on Hydra's speech_stopped (the simulator's user_speech_stop races), with a 50 ms floor; token usage from response.done. - session.configure handshake, client-side tools, native barge-in, generate_ initial_response greeting; ~18 KB payload cap on instructions (Hydra drops audio above it). - New 48 kHz audio helpers + in-memory WAV builder in audio_bridge.py. - Register framework in worker + config Literal; bump simulation_version to 2.0.6. - Unit tests for tool conversion, audio round-trip, transcriber selection/parsing.
1 parent 8bedd58 commit de2077a

9 files changed

Lines changed: 1122 additions & 4 deletions

File tree

.env.example

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -103,6 +103,14 @@ EVA_MODEL__TTS_PARAMS='{"api_key": "your_cartesia_api_key", "model": "sonic"}'
103103
#x pipeline_mode=S2S
104104
#v EVA_MODEL__S2S_PARAMS='{"model": "gpt-realtime-mini", "api_key": ""}'
105105

106+
# Smallest Hydra S2S (set EVA_FRAMEWORK=smallest_hydra). Hydra emits no transcript
107+
# on the wire, so a batch STT transcribes each turn for the audit log; it defaults
108+
# to Smallest Pulse keyed on the same api_key. Override via the "transcription" block
109+
# (provider: smallest|openai|deepgram). voice: wren|sloane|marlowe|reed|knox|tate.
110+
#x pipeline_mode=S2S
111+
#v EVA_MODEL__S2S=hydra
112+
#v EVA_MODEL__S2S_PARAMS='{"model": "hydra", "api_key": "", "voice": "wren", "generate_initial_response": true, "transcription": {"provider": "smallest", "model": "pulse-pro", "language": "en"}}'
113+
106114
# --- AudioLLM mode ---
107115
#i Audio-input LLM model name.
108116
#d string
@@ -117,7 +125,7 @@ EVA_MODEL__TTS_PARAMS='{"api_key": "your_cartesia_api_key", "model": "sonic"}'
117125
# --- Framework (S2S / AudioLLM) ---
118126
#i Base framework for S2S or AudioLLM pipelines.
119127
#d enum
120-
#e pipecat,openai_realtime,gemini_live,elevenlabs,grok_voice,deepgram
128+
#e pipecat,openai_realtime,gemini_live,elevenlabs,grok_voice,deepgram,smallest_hydra
121129
#v EVA_FRAMEWORK=openai_realtime
122130

123131
# ==============================================

docs/assistant_server_contract.md

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -595,3 +595,34 @@ Notable points specific to Deepgram:
595595
informational.
596596
- **Limitation.** The Voice Agent event stream exposes no token-usage event, so token usage is
597597
not reported for this framework. Latency is still emitted on the first audio chunk per turn.
598+
599+
## 14. Reference implementation: Smallest Hydra (transcript-less S2S)
600+
601+
`src/eva/assistant/smallest_hydra_server.py` (`framework: smallest_hydra`) bridges to Smallest's
602+
**Hydra** speech-to-speech model over a raw WebSocket (no SDK; uses `websockets`). It is scored as a
603+
true **S2S** pipeline (`s2s: hydra`, so `get_pipeline_type → S2S`) and follows the Gemini Live
604+
three-task structure (forward user audio, process model events, pace audio output).
605+
606+
Notable points specific to Hydra:
607+
608+
- **Config.** `framework: smallest_hydra`, `model: {s2s: hydra, s2s_params: {...}}`. Recognised
609+
`s2s_params`: `api_key` (**required**, Smallest key), `model` (default `hydra`), `voice`
610+
(`wren`|`sloane`|`marlowe`|`reed`|`knox`|`tate`, default `wren`, frozen at handshake),
611+
`generate_initial_response` (default `true` — Hydra speaks first; the system prompt is steered to
612+
EVA's canned greeting), and `transcription` (see below).
613+
- **Protocol.** OpenAI-Realtime-shaped JSON over one WebSocket: `session.configure` handshake;
614+
`input_audio_buffer.append` (base64 PCM16 in); `response.output_audio.delta` (base64 PCM16 out);
615+
client-side tools via `response.function_call_arguments.done` → reply with
616+
`conversation.item.create` (`function_call_output`) then `response.create`. Audio is **16 kHz in /
617+
48 kHz out**; recording rate is 48 kHz (assistant native; user track upsampled to match).
618+
- **Transcripts.** Hydra streams a native **assistant** transcript
619+
(`response.output_audio_transcript.delta`, accumulated per response and finalized on
620+
`response.done`) — used directly, since it is more accurate than re-transcribing the audio.
621+
It sends **no user transcript**, so each completed user utterance is batch-transcribed by a
622+
configurable STT (`s2s_transcription.py`) and appended to the audit log; the log's
623+
timestamp-sort keeps these asynchronous writes in chronological order. The `transcription`
624+
block selects the provider (`smallest` default — Pulse-pro for English, Pulse otherwise, keyed
625+
on the Hydra `api_key`; or `openai`/`deepgram` with their own `api_key`). User transcription is
626+
**fail-soft**: an STT error drops that turn's text but never aborts the conversation.
627+
- **Latency / tokens.** `model_response` latency is emitted on the first audio delta per turn
628+
(vs. the simulator's `user_speech_stop`); token usage is read from the `response.done` `usage` block.

src/eva/__init__.py

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@
77

88
# Bump simulation_version when changes affect benchmark outputs (agent code,
99
# user simulator, orchestrator, simulation prompts, agent configs, tool mocks).
10-
simulation_version = "2.0.5"
10+
simulation_version = "2.0.6"
1111

1212
# Bump metrics_version when changes affect metric computation (metrics code,
1313
# judge prompts, pricing tables, postprocessor).

src/eva/assistant/audio_bridge.py

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,9 +10,11 @@
1010

1111
import audioop
1212
import base64
13+
import io
1314
import json
1415
import struct
1516
import time
17+
import wave
1618
from pathlib import Path
1719

1820
import numpy as np
@@ -51,6 +53,56 @@ def mulaw_8k_to_pcm16_24k(mulaw_bytes: bytes) -> bytes:
5153
return pcm_24k
5254

5355

56+
def mulaw_8k_to_pcm16_48k(mulaw_bytes: bytes) -> bytes:
57+
"""Convert 8kHz mu-law audio to 48kHz 16-bit PCM.
58+
59+
Used by the Smallest Hydra S2S server, which records at 48 kHz (its native
60+
output rate) and must upsample the 8 kHz user track to match.
61+
"""
62+
pcm_8k = audioop.ulaw2lin(mulaw_bytes, 2)
63+
pcm_48k, _ = audioop.ratecv(pcm_8k, 2, 1, 8000, 48000, None)
64+
# Clamp to exactly 6× input length so tracks stay positionally aligned.
65+
expected_bytes = len(pcm_8k) * 6
66+
if len(pcm_48k) < expected_bytes:
67+
pcm_48k = pcm_48k + b"\x00" * (expected_bytes - len(pcm_48k))
68+
elif len(pcm_48k) > expected_bytes:
69+
pcm_48k = pcm_48k[:expected_bytes]
70+
return pcm_48k
71+
72+
73+
def pcm16_48k_to_mulaw_8k(pcm_bytes: bytes) -> bytes:
74+
"""Convert 48kHz 16-bit PCM to 8kHz mu-law.
75+
76+
Uses soxr VHQ resampling for proper anti-aliasing during the 6:1
77+
downsampling (audioop.ratecv lacks an anti-aliasing filter and sounds
78+
muffled). Mirrors ``pcm16_24k_to_mulaw_8k`` for Hydra's 48 kHz output.
79+
"""
80+
audio_data = np.frombuffer(pcm_bytes, dtype=np.int16)
81+
resampled = soxr.resample(audio_data, 48000, 8000, quality="VHQ")
82+
expected_samples = round(len(audio_data) * 8000 / 48000)
83+
if len(resampled) < expected_samples:
84+
resampled = np.pad(resampled, (0, expected_samples - len(resampled)))
85+
elif len(resampled) > expected_samples:
86+
resampled = resampled[:expected_samples]
87+
pcm_8k = resampled.astype(np.int16).tobytes()
88+
return audioop.lin2ulaw(pcm_8k, 2)
89+
90+
91+
def pcm16_to_wav_bytes(pcm_bytes: bytes, sample_rate: int) -> bytes:
92+
"""Wrap raw 16-bit mono PCM in an in-memory WAV container.
93+
94+
Used to hand a recorded utterance to a batch STT endpoint without touching
95+
disk. 16-bit mono => 2 bytes/sample.
96+
"""
97+
buf = io.BytesIO()
98+
with wave.open(buf, "wb") as wav:
99+
wav.setnchannels(1)
100+
wav.setsampwidth(2)
101+
wav.setframerate(sample_rate)
102+
wav.writeframes(pcm_bytes)
103+
return buf.getvalue()
104+
105+
54106
def pcm16_24k_to_mulaw_8k(pcm_bytes: bytes) -> bytes:
55107
"""Convert 24kHz 16-bit PCM to 8kHz mu-law.
56108
Lines changed: 178 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,178 @@
1+
"""Pluggable batch transcription for S2S frameworks that emit no transcript.
2+
3+
Some speech-to-speech models (notably Smallest Hydra) stream audio in both
4+
directions but never put text on the wire. EVA's audit log — which nearly all
5+
accuracy/experience metrics read — still needs a per-turn transcript, so the
6+
server transcribes each completed utterance (the PCM it sent or received) with a
7+
separate batch STT call.
8+
9+
The provider is configurable via the ``transcription`` block inside
10+
``s2s_params``::
11+
12+
"transcription": {
13+
"provider": "smallest" | "openai" | "deepgram",
14+
"model": "...", # provider-specific; sensible default per provider
15+
"api_key": "...", # falls back to the S2S api_key (smallest) / env
16+
"language": "en", # falls back to the run language
17+
"base_url": "..." # optional override
18+
}
19+
20+
All providers are called over a single shared async ``httpx`` client so the
21+
module has no heavy SDK import surface and is straightforward to mock in tests.
22+
"""
23+
24+
from __future__ import annotations
25+
26+
import os
27+
from typing import Any
28+
29+
import httpx
30+
31+
from eva.assistant.audio_bridge import pcm16_to_wav_bytes
32+
from eva.utils.logging import get_logger
33+
34+
logger = get_logger(__name__)
35+
36+
_SMALLEST_STT_URL = "https://api.smallest.ai/waves/v1/stt/"
37+
_OPENAI_STT_URL = "https://api.openai.com/v1/audio/transcriptions"
38+
_DEEPGRAM_STT_URL = "https://api.deepgram.com/v1/listen"
39+
40+
# Default model per provider. Smallest Pulse-pro is English-only; for any other
41+
# language we fall back to the multilingual Pulse model.
42+
_DEFAULT_MODELS = {
43+
"smallest": "pulse-pro",
44+
"openai": "whisper-1",
45+
"deepgram": "nova-3",
46+
}
47+
48+
49+
class BatchTranscriber:
50+
"""Transcribe completed utterances via a configurable batch STT provider."""
51+
52+
def __init__(
53+
self,
54+
provider: str,
55+
model: str,
56+
api_key: str,
57+
language: str,
58+
base_url: str | None = None,
59+
timeout: float = 30.0,
60+
) -> None:
61+
self.provider = provider
62+
self.model = model
63+
self.api_key = api_key
64+
self.language = language
65+
self.base_url = base_url
66+
self._client = httpx.AsyncClient(timeout=timeout)
67+
68+
async def transcribe(self, pcm: bytes, sample_rate: int) -> str:
69+
"""Transcribe raw 16-bit mono PCM. Returns "" on any failure (fail-soft).
70+
71+
Transcription errors must never abort a conversation — a dropped
72+
transcript degrades that turn's text metrics but the audio recording and
73+
tool calls are preserved.
74+
"""
75+
if not pcm:
76+
return ""
77+
wav = pcm16_to_wav_bytes(pcm, sample_rate)
78+
try:
79+
if self.provider == "smallest":
80+
return await self._transcribe_smallest(wav)
81+
if self.provider == "openai":
82+
return await self._transcribe_openai(wav)
83+
if self.provider == "deepgram":
84+
return await self._transcribe_deepgram(wav)
85+
logger.error(f"Unknown transcription provider: {self.provider!r}")
86+
return ""
87+
except Exception as e: # noqa: BLE001 - fail-soft by design
88+
logger.error(f"{self.provider} transcription failed: {e}", exc_info=True)
89+
return ""
90+
91+
async def _transcribe_smallest(self, wav: bytes) -> str:
92+
url = self.base_url or _SMALLEST_STT_URL
93+
resp = await self._client.post(
94+
url,
95+
params={"model": self.model, "language": self.language},
96+
headers={
97+
"Authorization": f"Bearer {self.api_key}",
98+
"Content-Type": "application/octet-stream",
99+
},
100+
content=wav,
101+
)
102+
resp.raise_for_status()
103+
return (resp.json().get("transcription") or "").strip()
104+
105+
async def _transcribe_openai(self, wav: bytes) -> str:
106+
url = self.base_url or _OPENAI_STT_URL
107+
resp = await self._client.post(
108+
url,
109+
headers={"Authorization": f"Bearer {self.api_key}"},
110+
files={"file": ("audio.wav", wav, "audio/wav")},
111+
data={"model": self.model, "language": self.language},
112+
)
113+
resp.raise_for_status()
114+
return (resp.json().get("text") or "").strip()
115+
116+
async def _transcribe_deepgram(self, wav: bytes) -> str:
117+
url = self.base_url or _DEEPGRAM_STT_URL
118+
resp = await self._client.post(
119+
url,
120+
params={"model": self.model, "language": self.language, "smart_format": "true"},
121+
headers={
122+
"Authorization": f"Token {self.api_key}",
123+
"Content-Type": "audio/wav",
124+
},
125+
content=wav,
126+
)
127+
resp.raise_for_status()
128+
alternatives = resp.json()["results"]["channels"][0]["alternatives"]
129+
return (alternatives[0]["transcript"] if alternatives else "").strip()
130+
131+
async def aclose(self) -> None:
132+
await self._client.aclose()
133+
134+
135+
def create_transcriber(s2s_params: dict[str, Any], language: str) -> BatchTranscriber:
136+
"""Build a :class:`BatchTranscriber` from ``s2s_params['transcription']``.
137+
138+
Defaults to the Smallest Pulse provider keyed on the S2S ``api_key`` so an
139+
in-stack transcript works with no extra configuration. The Pulse-pro model is
140+
English-only; for any non-English run the default model falls back to the
141+
multilingual ``pulse`` model.
142+
"""
143+
cfg = dict(s2s_params.get("transcription") or {})
144+
provider = cfg.get("provider", "smallest")
145+
lang = cfg.get("language") or language or "en"
146+
147+
model = cfg.get("model")
148+
if not model:
149+
model = _DEFAULT_MODELS.get(provider, "")
150+
# Smallest Pulse-pro is English-only; use multilingual Pulse otherwise.
151+
if provider == "smallest" and not lang.startswith("en"):
152+
model = "pulse"
153+
154+
# API key resolution: explicit transcription key, else a sensible fallback —
155+
# the shared S2S key for Smallest, or the provider's standard env var.
156+
api_key = cfg.get("api_key")
157+
if not api_key:
158+
if provider == "smallest":
159+
api_key = s2s_params.get("api_key", "")
160+
elif provider == "openai":
161+
api_key = os.environ.get("OPENAI_API_KEY", "")
162+
elif provider == "deepgram":
163+
api_key = os.environ.get("DEEPGRAM_API_KEY", "")
164+
if not api_key:
165+
raise ValueError(
166+
f"No API key for transcription provider {provider!r}. "
167+
f"Set s2s_params.transcription.api_key (or s2s_params.api_key for 'smallest', "
168+
f"OPENAI_API_KEY for 'openai', DEEPGRAM_API_KEY for 'deepgram')."
169+
)
170+
171+
logger.info(f"S2S transcription: provider={provider} model={model} language={lang}")
172+
return BatchTranscriber(
173+
provider=provider,
174+
model=model,
175+
api_key=api_key,
176+
language=lang,
177+
base_url=cfg.get("base_url"),
178+
)

0 commit comments

Comments
 (0)