feat(elevenlabs): secondary languages and language detection for realtime STT

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

@@ -69,6 +69,15 @@ def _speech_confidence(words: list[dict[str, Any]] | None) -> float:
     return min(1.0, max(0.0, math.exp(sum(logprobs) / len(logprobs))))
 
 
+def _wire_language(language: str) -> str:
+    """Normalize a language to the code the realtime API accepts.
+
+    It takes ISO-639-1 or ISO-639-3 and rejects the whole session on anything else, including
+    the region-tagged tags `LanguageCode` happily produces ("ru-RU"), so the region comes off.
+    """
+    return LanguageCode(language).language
+
+
 class VADOptions(TypedDict, total=False):
     vad_silence_threshold_secs: float | None
     """Silence threshold in seconds for VAD. Default to 1.5"""
@@ -90,6 +99,8 @@ class STTOptions:
     api_key: str
     base_url: str
     language_code: LanguageCode | None
+    secondary_languages: NotGivenOr[list[str]]
+    include_language_detection: NotGivenOr[bool]
     tag_audio_events: bool
     include_timestamps: bool
     sample_rate: STTRealtimeSampleRates
@@ -107,6 +118,8 @@ def __init__(
         api_key: NotGivenOr[str] = NOT_GIVEN,
         base_url: NotGivenOr[str] = NOT_GIVEN,
         language_code: NotGivenOr[str] = NOT_GIVEN,
+        secondary_languages: NotGivenOr[list[str]] = NOT_GIVEN,
+        include_language_detection: NotGivenOr[bool] = NOT_GIVEN,
         tag_audio_events: bool = True,
         use_realtime: NotGivenOr[bool] = NOT_GIVEN,  # Deprecated
         sample_rate: STTRealtimeSampleRates = 16000,
@@ -127,6 +140,16 @@ def __init__(
             api_key (NotGivenOr[str]): ElevenLabs API key. Can be set via argument or `ELEVEN_API_KEY` environment variable.
             base_url (NotGivenOr[str]): Custom base URL for the API. Optional.
             language_code (NotGivenOr[str]): Language code for the STT model. Optional.
+            secondary_languages (NotGivenOr[list[str]]): Additional languages that may be spoken in
+                the audio, on top of `language_code`. Keeps the primary language hinted while still
+                recognizing the others, which is what a code-switching speaker needs. Names and
+                ISO-639-3 codes are accepted and normalized to what the API takes. Only supported
+                for Scribe v2 realtime.
+            include_language_detection (NotGivenOr[bool]): Whether the committed transcript reports
+                the language the model actually heard. Defaults to True when no `language_code` is
+                set and False otherwise. Turning it off while no `language_code` is set leaves the
+                plugin with no language to report and every transcript falls back to "en". Only
+                supported for Scribe v2 realtime.
             tag_audio_events (bool): Whether to tag audio events like (laughter), (footsteps), etc. in the transcription.
                 Only supported for Scribe v1 model. Default is True.
             use_realtime (bool): Whether to use "scribe_v2_realtime" model for streaming mode. Default is NOT_GIVEN.
@@ -177,6 +200,20 @@ def __init__(
         if not use_realtime and is_given(server_vad):
             logger.warning("Server-side VAD is only supported for Scribe v2 realtime model")
 
+        if not use_realtime and is_given(secondary_languages):
+            logger.warning(
+                "`secondary_languages` is only supported for Scribe v2 realtime model "
+                "and will be ignored"
+            )
+            secondary_languages = NOT_GIVEN
+
+        if not use_realtime and is_given(include_language_detection):
+            logger.warning(
+                "`include_language_detection` is only supported for Scribe v2 realtime model "
+                "and will be ignored"
+            )
+            include_language_detection = NOT_GIVEN
+
         resolved_previous_text = previous_text if is_given(previous_text) else None
         if not use_realtime and resolved_previous_text is not None:
             logger.warning(
@@ -203,6 +240,8 @@ def __init__(
             api_key=elevenlabs_api_key,
             base_url=base_url if is_given(base_url) else API_BASE_URL_V1,
             language_code=LanguageCode(language_code) if language_code else None,
+            secondary_languages=secondary_languages,
+            include_language_detection=include_language_detection,
             tag_audio_events=tag_audio_events,
             sample_rate=sample_rate,
             server_vad=server_vad,
@@ -424,6 +463,29 @@ def _on_audio_duration_report(self, duration: float) -> None:
     def _server_vad(self) -> VADOptions | None:
         return self._opts.server_vad if is_given(self._opts.server_vad) else None
 
+    @property
+    def _language_detection(self) -> bool:
+        """Whether the session reports the language the model actually heard.
+
+        Defaults to on when no language was pinned, which is the only case where the plugin
+        used to request it."""
+        if is_given(self._opts.include_language_detection):
+            return self._opts.include_language_detection
+
+        return not self._language
+
+    @property
+    def _final_message_type(self) -> str:
+        """The committed message this session treats as the final transcript.
+
+        ElevenLabs sends every commit twice and puts the word timestamps and the detected
+        language on the delayed copy only, so that copy is the final one whenever either is
+        asked for."""
+        if self._opts.include_timestamps or self._language_detection:
+            return "committed_transcript_with_timestamps"
+
+        return "committed_transcript"
+
     async def _run(self) -> None:
         """Run the streaming transcription session"""
         closing_ws = False
@@ -591,7 +653,7 @@ async def _connect_ws(self) -> aiohttp.ClientWebSocketResponse:
             f"enable_logging={str(self._opts.enable_logging).lower()}",
         ]
 
-        if not self._language:
+        if self._language_detection:
             params.append("include_language_detection=true")
 
         if (server_vad := self._server_vad) is not None:
@@ -607,7 +669,13 @@ async def _connect_ws(self) -> aiohttp.ClientWebSocketResponse:
                 params.append(f"min_silence_duration_ms={min_silence_duration_ms}")
 
         if self._language:
-            params.append(f"language_code={self._language}")
+            params.append(f"language_code={quote(_wire_language(self._language))}")
+
+        if is_given(self._opts.secondary_languages):
+            params.extend(
+                f"secondary_languages={quote(_wire_language(language))}"
+                for language in self._opts.secondary_languages
+            )
 
         if self._opts.include_timestamps:
             params.append("include_timestamps=true")
@@ -662,7 +730,7 @@ def _process_stream_event(self, data: dict) -> None:
             end_time=end_time + self.start_time_offset,
             confidence=_speech_confidence(words),
         )
-        if words:
+        if words and self._opts.include_timestamps:
             speech_data.words = [
                 TimedString(
                     text=word.get("text", ""),
@@ -691,10 +759,8 @@ def _process_stream_event(self, data: dict) -> None:
                 )
                 self._event_ch.send_nowait(interim_event)
 
-        # 11labs sends both when include_timestamps is True or when the model is scribe_v2_realtime :(
-        elif (message_type == "committed_transcript" and not self._opts.include_timestamps) or (
-            message_type == "committed_transcript_with_timestamps" and self._opts.include_timestamps
-        ):
+        # 11labs sends every commit twice; _final_message_type picks the copy this session reads
+        elif message_type == self._final_message_type:
             # Final committed transcripts - these are sent to the LLM/TTS layer in LiveKit agents
             # and trigger agent responses (unlike partial transcripts which are UI-only)
             if text:
@@ -722,8 +788,8 @@ def _process_stream_event(self, data: dict) -> None:
                     self._event_ch.send_nowait(stt.SpeechEvent(type=SpeechEventType.END_OF_SPEECH))
                     self._speaking = False
 
-        elif message_type == "committed_transcript":
-            # if timestamps are included, these will be ignored above since we are handling committed_transcript_with_timestamps
+        elif message_type in ("committed_transcript", "committed_transcript_with_timestamps"):
+            # the other copy of a commit the branch above already emitted
             pass
 
         elif message_type == "session_started":
@@ -749,11 +815,6 @@ def _process_stream_event(self, data: dict) -> None:
                 details_suffix,
             )
             raise APIConnectionError(f"{message_type}: {error_msg}{details_suffix}")
-        elif (
-            message_type == "committed_transcript_with_timestamps"
-            and not self._opts.include_timestamps
-        ):
-            pass
         else:
             logger.warning("ElevenLabs STT unknown message type: %s, data: %s", message_type, data)
 

tests/test_plugin_elevenlabs_stt.py

@@ -15,7 +15,7 @@
 from yarl import URL
 
 from livekit import rtc
-from livekit.agents import DEFAULT_API_CONNECT_OPTIONS, stt
+from livekit.agents import DEFAULT_API_CONNECT_OPTIONS, LanguageCode, stt
 from livekit.agents.types import NOT_GIVEN
 from livekit.plugins.elevenlabs import stt as elevenlabs_stt
 from livekit.plugins.elevenlabs._utils import trace_id_from_headers
@@ -31,39 +31,60 @@ def send_nowait(self, event: stt.SpeechEvent) -> None:
         self.events.append(event)
 
 
-def _new_stream(*, server_vad=NOT_GIVEN) -> elevenlabs_stt.SpeechStream:
+def _new_stream(
+    *,
+    server_vad=NOT_GIVEN,
+    language: str | None = "en",
+    include_timestamps: bool = False,
+    include_language_detection=NOT_GIVEN,
+    secondary_languages=NOT_GIVEN,
+) -> elevenlabs_stt.SpeechStream:
     stream = object.__new__(elevenlabs_stt.SpeechStream)
     stream._opts = elevenlabs_stt.STTOptions(
         model_id="scribe_v2_realtime",
         api_key="test-key",
         base_url=elevenlabs_stt.API_BASE_URL_V1,
         language_code=None,
+        secondary_languages=secondary_languages,
+        include_language_detection=include_language_detection,
         tag_audio_events=True,
-        include_timestamps=False,
+        include_timestamps=include_timestamps,
         sample_rate=16000,
         server_vad=server_vad,
         keyterms=NOT_GIVEN,
         no_verbatim=False,
         enable_logging=True,
         previous_text=None,
     )
-    stream._language = None
+    stream._language = language
     stream._event_ch = _EventSink()
     stream._speaking = False
     stream._start_time_offset = 0.0
     return stream
 
 
-def _committed_transcript(text: str) -> dict:
-    return {
-        "message_type": "committed_transcript",
+def _committed_transcript(
+    text: str,
+    *,
+    with_timestamps: bool = False,
+    language_code: str | None = None,
+) -> dict:
+    message: dict = {
+        "message_type": "committed_transcript_with_timestamps"
+        if with_timestamps
+        else "committed_transcript",
         "text": text,
         "words": [
             {"text": text, "start": 0.1, "end": 0.4},
         ]
         if text
         else [],
     }
+    # the server only carries the detected language on the delayed copy, and only when
+    # language detection is enabled
+    if language_code is not None:
+        message["language_code"] = language_code
+    return message
 
 
 def test_server_vad_commit_emits_end_of_speech() -> None:
@@ -271,6 +292,142 @@ def test_stream_update_options_sets_keyterms_and_requests_reconnect() -> None:
     assert stream._reconnect_event.is_set()
 
 
+async def test_connect_ws_normalizes_the_primary_language() -> None:
+    # LanguageCode keeps the region ("en-US") but the realtime API rejects it, so the primary
+    # language goes on the wire through the same normalization the secondary ones get
+    stream = _new_stream(language=LanguageCode("en_US"), secondary_languages=["ru-RU"])
+
+    url = await _connect_ws_url(stream)
+
+    assert URL(url).query.getall("language_code") == ["en"]
+    assert URL(url).query.getall("secondary_languages") == ["ru"]
+
+
+async def test_connect_ws_includes_secondary_languages() -> None:
+    # secondary languages ride along with the pinned primary one and are sent as
+    # repeated query params, the only serialization the realtime API accepts.
+    stream = _new_stream(language="en", secondary_languages=["ru", "es"])
+
+    url = await _connect_ws_url(stream)
+
+    assert "language_code=en" in url
+    assert URL(url).query.getall("secondary_languages") == ["ru", "es"]
+
+
+async def test_connect_ws_normalizes_secondary_languages() -> None:
+    # the realtime API takes ISO-639-1/3 and rejects anything else, including the region-tagged
+    # tags LanguageCode produces, so names and regions are mapped before the connect URL
+    stream = _new_stream(language="en", secondary_languages=["ru_RU", "french", "spa"])
+
+    url = await _connect_ws_url(stream)
+
+    assert URL(url).query.getall("secondary_languages") == ["ru", "fr", "es"]
+
+
+async def test_connect_ws_omits_secondary_languages_when_not_given() -> None:
+    url = await _connect_ws_url(_new_stream(language="en"))
+
+    assert "secondary_languages=" not in url
+
+
+def test_secondary_languages_ignored_for_batch_model(caplog: pytest.LogCaptureFixture) -> None:
+    with caplog.at_level("WARNING"):
+        instance = elevenlabs_stt.STT(
+            api_key="test-key", model="scribe_v2", secondary_languages=["ru"]
+        )
+
+    assert instance._opts.secondary_languages is NOT_GIVEN
+    assert "only supported for Scribe v2 realtime" in caplog.text
+
+
+async def test_connect_ws_requests_language_detection_when_no_language_is_pinned() -> None:
+    url = await _connect_ws_url(_new_stream(language=None))
+
+    assert "include_language_detection=true" in url
+
+
+async def test_connect_ws_omits_language_detection_when_language_is_pinned() -> None:
+    url = await _connect_ws_url(_new_stream(language="en"))
+
+    assert "include_language_detection" not in url
+
+
+async def test_connect_ws_requests_language_detection_when_explicitly_enabled() -> None:
+    stream = _new_stream(language="en", include_language_detection=True)
+
+    url = await _connect_ws_url(stream)
+
+    assert "include_language_detection=true" in url
+
+
+async def test_connect_ws_omits_language_detection_when_explicitly_disabled() -> None:
+    url = await _connect_ws_url(_new_stream(language=None, include_language_detection=False))
+
+    assert "include_language_detection" not in url
+
+
+def test_final_transcript_reports_the_detected_language() -> None:
+    # with detection on, the detected language only reaches the delayed copy of the
+    # commit, so that copy has to be the final one or every transcript is labelled
+    # with the pinned language (or "en" when nothing is pinned).
+    stream = _new_stream(
+        server_vad={"vad_silence_threshold_secs": 0.5},
+        language="en",
+        include_language_detection=True,
+        secondary_languages=["ru"],
+    )
+
+    stream._process_stream_event(_committed_transcript("привет"))
+    stream._process_stream_event(
+        _committed_transcript("привет", with_timestamps=True, language_code="ru")
+    )
+
+    finals = [
+        event
+        for event in stream._event_ch.events
+        if event.type is stt.SpeechEventType.FINAL_TRANSCRIPT
+    ]
+    assert len(finals) == 1
+    assert finals[0].alternatives[0].language == "ru"
+
+
+def test_autodetected_language_reaches_the_final_transcript() -> None:
+    # without a pinned language the plugin asks the server to detect one; that language
+    # only rides on the delayed copy, so dropping it labelled every transcript "en"
+    stream = _new_stream(server_vad={"vad_silence_threshold_secs": 0.5}, language=None)
+
+    stream._process_stream_event(
+        _committed_transcript("привет", with_timestamps=True, language_code="ru")
+    )
+    stream._process_stream_event(_committed_transcript("привет"))
+
+    finals = [
+        event
+        for event in stream._event_ch.events
+        if event.type is stt.SpeechEventType.FINAL_TRANSCRIPT
+    ]
+    assert len(finals) == 1
+    assert finals[0].alternatives[0].language == "ru"
+    # reading the timestamped copy must not start handing out word timings the caller
+    # never asked for
+    assert finals[0].alternatives[0].words is None
+
+
+def test_final_transcript_keeps_the_plain_copy_without_detection() -> None:
+    stream = _new_stream(server_vad={"vad_silence_threshold_secs": 0.5}, language="es")
+
+    stream._process_stream_event(_committed_transcript("hola"))
+    stream._process_stream_event(_committed_transcript("hola", with_timestamps=True))
+
+    finals = [
+        event
+        for event in stream._event_ch.events
+        if event.type is stt.SpeechEventType.FINAL_TRANSCRIPT
+    ]
+    assert len(finals) == 1
+    assert finals[0].alternatives[0].language == "es"
+
+
 class _FakeWS:
     """Records outgoing messages. receive() parks so recv_task stays alive."""