fix(elevenlabs): send keyterms on the realtime STT WebSocket

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

@@ -22,6 +22,7 @@
 import weakref
 from dataclasses import dataclass
 from typing import Any, Literal, TypedDict
+from urllib.parse import quote
 
 import aiohttp
 
@@ -137,9 +138,9 @@ def __init__(
                 be selected based on parameters provided.
             model_id (ElevenLabsSTTModels | str): Deprecated alias for `model`. Use `model` instead.
             keyterms (NotGivenOr[list[str]]): A list of keywords or phrases to bias the transcription towards.
-                Each keyterm can contain at most 5 words and must be less than 50 characters.
-                Maximum of 100 keyterms. Only supported for Scribe v2 batch recognition
-                (not realtime streaming). Usage incurs additional costs.
+                Supported for both Scribe v2 (batch) and Scribe v2 realtime. Batch accepts up to
+                1000 keyterms of at most 50 characters each; realtime accepts up to 50 keyterms of
+                at most 20 characters each. Usage incurs additional costs.
             no_verbatim (NotGivenOr[bool]): When True, the model removes filler words, false starts
                 and disfluencies from the transcript, producing cleaner output. Supported for both
                 Scribe v2 (batch) and Scribe v2 realtime. Default is False.
@@ -352,7 +353,7 @@ def update_options(
             self._opts.no_verbatim = no_verbatim
 
         for stream in self._streams:
-            stream.update_options(server_vad=server_vad, no_verbatim=no_verbatim)
+            stream.update_options(server_vad=server_vad, no_verbatim=no_verbatim, keyterms=keyterms)
 
     def stream(
         self,
@@ -399,13 +400,17 @@ def update_options(
         *,
         server_vad: NotGivenOr[VADOptions] = NOT_GIVEN,
         no_verbatim: NotGivenOr[bool] = NOT_GIVEN,
+        keyterms: NotGivenOr[list[str]] = NOT_GIVEN,
     ) -> None:
         if is_given(server_vad):
             self._opts.server_vad = server_vad
             self._reconnect_event.set()
         if is_given(no_verbatim):
             self._opts.no_verbatim = no_verbatim
             self._reconnect_event.set()
+        if is_given(keyterms):
+            self._opts.keyterms = keyterms
+            self._reconnect_event.set()
 
     def _on_audio_duration_report(self, duration: float) -> None:
         usage_event = stt.SpeechEvent(
@@ -610,6 +615,9 @@ async def _connect_ws(self) -> aiohttp.ClientWebSocketResponse:
         if self._opts.no_verbatim:
             params.append("no_verbatim=true")
 
+        if is_given(self._opts.keyterms):
+            params.extend(f"keyterms={quote(keyterm)}" for keyterm in self._opts.keyterms)
+
         query_string = "&".join(params)
 
         # Convert HTTPS URL to WSS

tests/test_plugin_elevenlabs_stt.py

@@ -12,6 +12,7 @@
 
 import pytest
 from multidict import CIMultiDict
+from yarl import URL
 
 from livekit import rtc
 from livekit.agents import DEFAULT_API_CONNECT_OPTIONS, stt
@@ -188,6 +189,88 @@ async def ws_connect(self, url: str, **kwargs: object) -> object:
     assert f"enable_logging={expected}" in captured["url"]
 
 
+async def _connect_ws_url(stream: elevenlabs_stt.SpeechStream) -> str:
+    """Run _connect_ws against a fake session and return the realtime connect URL."""
+
+    class _ConnOptions:
+        timeout = 5.0
+
+    stream._conn_options = _ConnOptions()
+
+    captured: dict[str, str] = {}
+
+    class _FakeSession:
+        async def ws_connect(self, url: str, **kwargs: object) -> object:
+            captured["url"] = url
+            return object()
+
+    stream._session = _FakeSession()
+
+    await stream._connect_ws()
+    return captured["url"]
+
+
+async def test_connect_ws_includes_keyterms() -> None:
+    # keyterms bias the realtime model and are sent as repeated query params on
+    # the connect URL.
+    stream = _new_stream()
+    stream._opts.keyterms = ["nginx", "Grafana Loki", "Ærø"]
+
+    url = await _connect_ws_url(stream)
+
+    assert "keyterms=nginx" in url
+    assert "keyterms=Grafana%20Loki" in url
+    assert "keyterms=%C3%86r%C3%B8" in url
+
+
+async def test_connect_ws_escapes_query_delimiters_in_keyterms() -> None:
+    # Keyterms are free-form text. Unescaped, "&" would split the term and inject
+    # a bogus query param and "#" would truncate it, so both must be encoded and
+    # must survive a round trip through the URL parser aiohttp uses.
+    stream = _new_stream()
+    stream._opts.keyterms = ["Smith & Sons", "C#"]
+
+    url = await _connect_ws_url(stream)
+
+    assert "keyterms=Smith%20%26%20Sons" in url
+    assert "keyterms=C%23" in url
+    assert URL(url).query.getall("keyterms") == ["Smith & Sons", "C#"]
+
+
+async def test_connect_ws_omits_keyterms_when_not_given() -> None:
+    url = await _connect_ws_url(_new_stream())
+
+    assert "keyterms=" not in url
+
+
+def test_update_options_forwards_keyterms_to_active_streams() -> None:
+    # keyterms are a WebSocket query param applied at connect time, so a live
+    # realtime stream must be told to reconnect. Verify STT.update_options
+    # forwards them to active streams (which trigger a reconnect).
+    instance = _stt()
+    captured: dict[str, object] = {}
+
+    class _FakeStream:
+        def update_options(self, **kwargs: object) -> None:
+            captured.update(kwargs)
+
+    # _streams is a WeakSet: keep a strong reference so the fake survives.
+    fake = _FakeStream()
+    instance._streams.add(fake)
+    instance.update_options(keyterms=["nginx"])
+    assert captured.get("keyterms") == ["nginx"]
+
+
+def test_stream_update_options_sets_keyterms_and_requests_reconnect() -> None:
+    stream = _new_stream()
+    stream._reconnect_event = asyncio.Event()
+
+    stream.update_options(keyterms=["nginx"])
+
+    assert stream._opts.keyterms == ["nginx"]
+    assert stream._reconnect_event.is_set()
+
+
 class _FakeWS:
     """Records outgoing messages. receive() parks so recv_task stays alive."""