feat(elevenlabs): make realtime STT audio chunk duration configurable

livekit-plugins/livekit-plugins-elevenlabs/README.md

@@ -13,3 +13,24 @@ pip install livekit-plugins-elevenlabs
 ## Pre-requisites
 
 You'll need an API key from ElevenLabs. It can be set as an environment variable: `ELEVEN_API_KEY`
+
+## Realtime speech-to-text audio chunks
+
+Configure the outgoing audio chunk duration when creating the STT instance:
+
+```python
+from livekit.plugins import elevenlabs
+
+stt = elevenlabs.STT(
+    model="scribe_v2_realtime",
+    audio_chunk_duration_ms=100,
+)
+```
+
+`audio_chunk_duration_ms` accepts a positive integer in milliseconds and defaults
+to 50. At 100 ms, continuous audio produces approximately 10 audio messages per
+second instead of 20, at the cost of up to 50 ms additional buffering. Larger
+chunks do not pace reconnect backlogs or guarantee that provider queue errors
+are avoided. Flushes send any remaining partial chunk before the commit.
+
+This option applies only to realtime STT and is set at construction time.

livekit-plugins/livekit-plugins-elevenlabs/livekit/plugins/elevenlabs/stt.py

@@ -99,6 +99,7 @@ class STTOptions:
     no_verbatim: bool
     enable_logging: bool
     previous_text: str | None
+    audio_chunk_duration_ms: int = 50
 
 
 class STT(stt.STT):
@@ -111,6 +112,7 @@ def __init__(
         tag_audio_events: bool = True,
         use_realtime: NotGivenOr[bool] = NOT_GIVEN,  # Deprecated
         sample_rate: STTRealtimeSampleRates = 16000,
+        audio_chunk_duration_ms: int = 50,
         server_vad: NotGivenOr[VADOptions] = NOT_GIVEN,
         include_timestamps: bool = False,
         http_session: aiohttp.ClientSession | None = None,
@@ -134,6 +136,10 @@ def __init__(
             use_realtime (bool): Whether to use "scribe_v2_realtime" model for streaming mode. Default is NOT_GIVEN.
                 Note that this flag is deprecated in favour of explicitly specifying the model id.
             sample_rate (STTRealtimeSampleRates): Audio sample rate in Hz. Default is 16000.
+            audio_chunk_duration_ms (int): Duration of each outgoing realtime audio chunk in
+                milliseconds. Must be a positive integer. Defaults to 50. Larger chunks reduce
+                message frequency but increase buffering latency. Flushes send any shorter
+                remaining chunk before committing. Only used for Scribe v2 realtime.
             server_vad (NotGivenOr[VADOptions]): Server-side VAD options, only supported for Scribe v2 realtime model.
             http_session (aiohttp.ClientSession | None): Custom HTTP session for API requests. Optional.
             model (ElevenLabsSTTModels | str): ElevenLabs STT model to use. If not specified a default model will
@@ -157,6 +163,13 @@ def __init__(
                 audio chunk to improve transcription accuracy. Only supported for Scribe v2 realtime.
         """
 
+        if (
+            isinstance(audio_chunk_duration_ms, bool)
+            or not isinstance(audio_chunk_duration_ms, int)
+            or audio_chunk_duration_ms <= 0
+        ):
+            raise ValueError("audio_chunk_duration_ms must be a positive integer")
+
         if is_given(model_id):
             if is_given(model):
                 logger.warning(
@@ -220,6 +233,7 @@ def __init__(
             language_code=LanguageCode(language_code) if language_code else None,
             tag_audio_events=tag_audio_events,
             sample_rate=sample_rate,
+            audio_chunk_duration_ms=audio_chunk_duration_ms,
             server_vad=server_vad,
             include_timestamps=include_timestamps,
             model_id=model,
@@ -487,18 +501,18 @@ async def keepalive_task(ws: aiohttp.ClientWebSocketResponse) -> None:
         async def send_task(ws: aiohttp.ClientWebSocketResponse) -> None:
             nonlocal closing_ws
 
-            # Buffer audio into chunks (50ms chunks)
-            samples_50ms = self._opts.sample_rate // 20
+            # Buffer audio into chunks of the configured duration.
+            samples_per_chunk = self._opts.sample_rate * self._opts.audio_chunk_duration_ms // 1000
             audio_bstream = utils.audio.AudioByteStream(
                 sample_rate=self._opts.sample_rate,
                 num_channels=1,
-                samples_per_channel=samples_50ms,
+                samples_per_channel=samples_per_chunk,
             )
 
             has_ended = False
             try:
                 async for data in self._input_ch:
-                    # Write audio bytes to buffer and get 50ms frames
+                    # Write audio bytes to the buffer and get complete chunks
                     frames: list[rtc.AudioFrame] = []
                     if isinstance(data, rtc.AudioFrame):
                         frames.extend(audio_bstream.write(data.data.tobytes()))

tests/test_plugin_elevenlabs_stt.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import asyncio
+import base64
 import dataclasses
 import json
 import time
@@ -345,10 +346,10 @@ async def close(self) -> None:
         self._closed.set()
 
 
-def _live_stream(ws: _FakeWS) -> elevenlabs_stt.SpeechStream:
+def _live_stream(ws: _FakeWS, **kwargs: Any) -> elevenlabs_stt.SpeechStream:
     """A real SpeechStream running its real _run loop against a fake socket."""
-    instance = elevenlabs_stt.STT(api_key="test-key", model="scribe_v2_realtime")
-    opts = dataclasses.replace(instance._opts, sample_rate=16000)
+    instance = elevenlabs_stt.STT(api_key="test-key", model="scribe_v2_realtime", **kwargs)
+    opts = dataclasses.replace(instance._opts)
     stream = elevenlabs_stt.SpeechStream(
         stt=instance,
         opts=opts,
@@ -462,3 +463,51 @@ def test_committed_transcript_sets_confidence() -> None:
     final = stream._event_ch.events[1]
     assert final.type == stt.SpeechEventType.FINAL_TRANSCRIPT
     assert final.alternatives[0].confidence > 0.9
+
+
+def test_audio_chunk_duration_defaults_to_50ms() -> None:
+    assert _stt()._opts.audio_chunk_duration_ms == 50
+
+
+@pytest.mark.parametrize("duration", [0, -1, 0.5, 100.0, True, False, None, "100"])
+def test_invalid_audio_chunk_duration_is_rejected(duration: Any) -> None:
+    with pytest.raises(ValueError, match="audio_chunk_duration_ms must be a positive integer"):
+        _stt(audio_chunk_duration_ms=duration)
+
+
+@pytest.mark.parametrize("sample_rate", [8000, 16000, 48000])
+@pytest.mark.parametrize("duration", [1, 50, 75, 100, 200])
+@pytest.mark.parametrize("tail_ms", [0, 1])
+async def test_configured_audio_chunks_preserve_audio_and_commit(
+    sample_rate: int, duration: int, tail_ms: int
+) -> None:
+    ws = _FakeWS()
+    stream = _live_stream(ws, sample_rate=sample_rate, audio_chunk_duration_ms=duration)
+    # Distinct PCM bytes reveal drops, duplication and reordering across input frames.
+    total_samples = sample_rate * (duration * 2 + tail_ms) // 1000
+    audio = bytes(i % 251 for i in range(total_samples * 2))
+    input_frame_bytes = sample_rate * 20 // 1000 * 2
+    chunk_bytes = sample_rate * duration // 1000 * 2
+    expected_chunks = [audio[i : i + chunk_bytes] for i in range(0, len(audio), chunk_bytes)]
+    try:
+        for offset in range(0, len(audio), input_frame_bytes):
+            data = audio[offset : offset + input_frame_bytes]
+            stream.push_frame(
+                rtc.AudioFrame(
+                    data=data,
+                    sample_rate=sample_rate,
+                    num_channels=1,
+                    samples_per_channel=len(data) // 2,
+                )
+            )
+        # Complete chunks must be released before the caller flushes.
+        await _wait_until(lambda: len(ws.sent) >= len(audio) // chunk_bytes)
+        stream.flush()
+        await _wait_until(lambda: len(ws.sent) >= len(expected_chunks) + 1)
+
+        assert [base64.b64decode(msg["audio_base_64"]) for msg in ws.sent[:-1]] == expected_chunks
+        assert [msg["commit"] for msg in ws.sent] == [False] * len(expected_chunks) + [True]
+        assert all(msg["sample_rate"] == sample_rate for msg in ws.sent)
+        assert ws.sent[-1]["audio_base_64"] == ""
+    finally:
+        await stream.aclose()