livekit · longcw · Jun 30, 2026 · Jun 9, 2026 · Jun 10, 2026 · Jun 10, 2026
@@ -34,7 +34,7 @@ def __init__(self) -> None:
             "with that in mind keep your responses concise and to the point."
             "do not use emojis, asterisks, markdown, or other special characters in your responses."
             "You are curious and friendly, and have a sense of humor."
-            "you will speak english to the user",
+            "You will speak english to the user over voice.",
             tools=[EndCallTool()],
         )
 
@@ -102,10 +102,20 @@ async def entrypoint(ctx: JobContext) -> None:
             "filter_markdown",
             text_transforms.replace({"LiveKit": "<<ˈ|l|aɪ|v|k|ɪ|t>>"}),
         ],
+        # automatically detect keyterms and apply them to the STT per user turn
+        keyterms_options={
+            "keyterms": ["LiveKit"],
+            "keyterm_detection": {
+                "enabled": True,
+                "turn_interval": 1,  # increase to reduce LLM API calls
+            },
+        },
     )
 
     @session.on("metrics_collected")
     def _on_metrics_collected(ev: MetricsCollectedEvent) -> None:
+        if ev.metrics.type == "stt_metrics":
+            return
         metrics.log_metrics(ev.metrics)
 
     async def log_usage():

@@ -108,6 +108,7 @@
     AMDPredictionEvent,
 )
 from .voice.background_audio import AudioConfig, BackgroundAudioPlayer, BuiltinAudioClip, PlayHandle
+from .voice.keyterm_detection import KeytermDetectionOptions, KeytermsOptions
 from .voice.room_io import RoomInputOptions, RoomIO, RoomOutputOptions
 from .voice.run_result import (
     AgentHandoffEvent,
@@ -256,6 +257,8 @@ def __getattr__(name: str) -> typing.Any:
     "InterruptionOptions",
     "PreemptiveGenerationOptions",
     "UserTurnLimitOptions",
+    "KeytermsOptions",
+    "KeytermDetectionOptions",
     "UserTurnExceededEvent",
 ]
 

@@ -196,6 +196,46 @@ def _diarization_enabled(extra_kwargs: dict[str, Any] | None) -> bool:
     return False
 
 
+def _keyterms_extra_for_model(
+    model: NotGivenOr[str],
+    *,
+    extra_kwargs: dict[str, Any] | None = None,
+    session_keyterms: list[str] | None = None,
+) -> dict[str, Any] | None:
+    """Return the provider's keyterm ``extra`` entry: user keyterms (from ``extra_kwargs``)
+    merged with the framework ``session_keyterms``.
+
+    None if the model has no keyterm prompting, so ``_keyterms_extra_for_model(model) is not
+    None`` is also the capability check.
+    """
+    if not (is_given(model) and isinstance(model, str)):
+        return None
+
+    extra_kwargs = extra_kwargs or {}
+    session_keyterms = session_keyterms or []
+
+    if model.startswith("speechmatics/"):
+        # keep existing entries as-is (they may carry sounds_like etc.); append new session terms
+        existing = list(extra_kwargs.get("additional_vocab", []))
+        seen = {v["content"] for v in existing}
+        additions = set(session_keyterms) - seen
+        return {"additional_vocab": existing + [{"content": term} for term in additions]}
+
+    key: str | None = None
+    if model.startswith("deepgram/"):
+        key = "keyterm"
+    elif model.startswith("assemblyai/"):
+        key = "keyterms_prompt"
+
+    if key is None:
+        return None
+    # deepgram's keyterm may be a bare string; wrap it so it isn't splat char-by-char
+    existing = extra_kwargs.get(key, [])
+    if isinstance(existing, str):
+        existing = [existing]
+    return {key: list(dict.fromkeys([*existing, *session_keyterms]))}
+
+
 STTLanguages = Literal["multi", "en", "de", "es", "fr", "ja", "pt", "zh", "hi"]
 
 
@@ -516,6 +556,7 @@ def __init__(
                 diarization=diarization_enabled,
                 aligned_transcript="word",
                 offline_recognize=False,
+                keyterms=_keyterms_extra_for_model(model) is not None,
             ),
         )
 
@@ -559,6 +600,7 @@ def __init__(
 
         self._session = http_session
         self._vad = vad
+        self._session_keyterms: list[str] = []  # framework-managed; merged into extra_kwargs
         self._streams = weakref.WeakSet[SpeechStream]()
 
     @classmethod
@@ -633,6 +675,10 @@ def update_options(
 
             self._opts.model = model
             self._vad = _resolve_vad_for_model(model, self._vad)
+            self._capabilities = replace(
+                self._capabilities,
+                keyterms=_keyterms_extra_for_model(self._opts.model) is not None,
+            )
         if is_given(language):
             self._opts.language = LanguageCode(language)
         if is_given(extra):
@@ -641,10 +687,37 @@ def update_options(
                 self._capabilities,
                 diarization=_diarization_enabled(self._opts.extra_kwargs),
             )
+            # re-merge the active session keyterms so a user extra update doesn't drop them
+            keyterm_extra = _keyterms_extra_for_model(
+                self._opts.model,
+                extra_kwargs=self._opts.extra_kwargs,
+                session_keyterms=self._session_keyterms,
+            )
+            if keyterm_extra is not None:
+                extra = {**extra, **keyterm_extra}
 
         for stream in self._streams:
             stream.update_options(model=model, language=language, extra=extra)
 
+    def _update_session_keyterms(self, keyterms: list[str]) -> None:
+        if keyterms == self._session_keyterms:
+            return
+        keyterm_extra = _keyterms_extra_for_model(
+            self._opts.model, extra_kwargs=self._opts.extra_kwargs, session_keyterms=keyterms
+        )
+        if keyterm_extra is None:
+            super()._update_session_keyterms(keyterms)  # warn-and-skip for unsupported models
+            return
+
+        self._session_keyterms = list(keyterms)
+        # inference applies extra live via session.update; defer to END_OF_SPEECH since the
+        # gateway may reconnect upstream when the keyterms change
+        for stream in self._streams:
+            if stream._speaking:
+                stream._pending_extra = keyterm_extra
+            else:
+                stream.update_options(extra=keyterm_extra)
+
     def _sanitize_options(
         self, *, language: NotGivenOr[STTLanguages | str] = NOT_GIVEN
     ) -> STTOptions:
@@ -673,6 +746,9 @@ def __init__(
         self._request_id = str(utils.shortuuid("stt_request_"))
 
         self._speaking = False
+        # keyterm extra set while the user is speaking; applied at END_OF_SPEECH (latest wins).
+        # inference applies live, but the gateway may reconnect upstream, so defer to a calm moment.
+        self._pending_extra: dict[str, Any] | None = None
         self._speech_duration: float = 0
         self._ws: aiohttp.ClientWebSocketResponse | None = None
         self._vad: vad.VAD | None = vad_instance
@@ -696,6 +772,7 @@ def update_options(
             self._opts.language = LanguageCode(language)
         if is_given(extra):
             self._opts.extra_kwargs.update(extra)
+            self._pending_extra = None
 
         has_update = is_given(model) or is_given(language) or is_given(extra)
         if has_update and self._ws is not None and not self._ws.closed:
@@ -712,6 +789,11 @@ def update_options(
             }
             asyncio.ensure_future(self._send_session_update(update_msg))
 
+    def _on_end_of_speech(self) -> None:
+        if self._pending_extra is not None:
+            self.update_options(extra=self._pending_extra)
+            self._pending_extra = None
+
     async def _send_session_update(self, msg: dict[str, Any]) -> None:
         try:
             if self._ws is not None and not self._ws.closed:
@@ -846,7 +928,18 @@ async def _connect_ws(
             "settings": {
                 "sample_rate": str(self._opts.sample_rate),
                 "encoding": self._opts.encoding,
-                "extra": self._opts.extra_kwargs,
+                # merge the framework session keyterms into the user's extra_kwargs keyterm key
+                "extra": {
+                    **self._opts.extra_kwargs,
+                    **(
+                        _keyterms_extra_for_model(
+                            self._opts.model,
+                            extra_kwargs=self._opts.extra_kwargs,
+                            session_keyterms=self._stt._session_keyterms,
+                        )
+                        or {}
+                    ),
+                },
             },
         }
 
@@ -973,6 +1066,7 @@ def _process_transcript(self, data: dict, is_final: bool) -> None:
                 self._speaking = False
                 end_event = stt.SpeechEvent(type=stt.SpeechEventType.END_OF_SPEECH)
                 self._event_ch.send_nowait(end_event)
+                self._on_end_of_speech()
         else:
             event = stt.SpeechEvent(
                 type=stt.SpeechEventType.INTERIM_TRANSCRIPT,

@@ -6,7 +6,7 @@
 import time
 from collections.abc import AsyncIterable
 from dataclasses import dataclass
-from typing import Any, Literal
+from typing import TYPE_CHECKING, Any, Literal
 
 from livekit import rtc
 
@@ -19,6 +19,9 @@
 from ..vad import VAD
 from .stt import STT, RecognizeStream, SpeechEvent, SpeechEventType, STTCapabilities
 
+if TYPE_CHECKING:
+    from ..voice.events import ConversationItemAddedEvent
+
 # don't retry when using the fallback adapter
 DEFAULT_FALLBACK_API_CONNECT_OPTIONS = APIConnectOptions(
     max_retry=0, timeout=DEFAULT_API_CONNECT_OPTIONS.timeout
@@ -84,6 +87,8 @@ def __init__(
                 interim_results=all(t.capabilities.interim_results for t in stt),
                 diarization=all(t.capabilities.diarization for t in stt),
                 aligned_transcript=aligned_transcript,
+                keyterms=any(t.capabilities.keyterms for t in stt),
+                chat_context=any(t.capabilities.chat_context for t in stt),
             )
         )
 
@@ -113,6 +118,16 @@ def model(self) -> str:
     def provider(self) -> str:
         return "livekit"
 
+    def _update_session_keyterms(self, keyterms: list[str]) -> None:
+        # forward to every underlying STT; unsupported ones warn-and-skip internally
+        for stt_instance in self._stt_instances:
+            stt_instance._update_session_keyterms(keyterms)
+
+    def _push_conversation_item(self, ev: ConversationItemAddedEvent) -> None:
+        # forward to every underlying STT; unsupported ones warn-and-skip internally
+        for stt_instance in self._stt_instances:
+            stt_instance._push_conversation_item(ev)
+
     async def _try_recognize(
         self,
         *,

@@ -2,13 +2,16 @@
 
 import asyncio
 from collections.abc import AsyncIterable
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from .. import utils
 from ..types import DEFAULT_API_CONNECT_OPTIONS, NOT_GIVEN, APIConnectOptions, NotGivenOr
 from ..vad import VAD, VADEventType
 from .stt import STT, RecognizeStream, SpeechEvent, SpeechEventType, STTCapabilities
 
+if TYPE_CHECKING:
+    from ..voice.events import ConversationItemAddedEvent
+
 # already a retry mechanism in STT.recognize, don't retry in stream adapter
 DEFAULT_STREAM_ADAPTER_API_CONNECT_OPTIONS = APIConnectOptions(
     max_retry=0, timeout=DEFAULT_API_CONNECT_OPTIONS.timeout
@@ -22,6 +25,8 @@ def __init__(self, *, stt: STT, vad: VAD) -> None:
                 streaming=True,
                 interim_results=False,
                 diarization=False,  # diarization requires streaming STT
+                keyterms=stt.capabilities.keyterms,
+                chat_context=stt.capabilities.chat_context,
             )
         )
         self._vad = vad
@@ -42,6 +47,12 @@ def model(self) -> str:
     def provider(self) -> str:
         return self._stt.provider
 
+    def _update_session_keyterms(self, keyterms: list[str]) -> None:
+        self._stt._update_session_keyterms(keyterms)
+
+    def _push_conversation_item(self, item: ConversationItemAddedEvent) -> None:
+        self._stt._push_conversation_item(item)
+
     async def _recognize_impl(
         self,
         buffer: utils.AudioBuffer,

@@ -7,7 +7,7 @@
 from dataclasses import dataclass, field
 from enum import Enum, unique
 from types import TracebackType
-from typing import Any, Generic, Literal, TypeVar
+from typing import TYPE_CHECKING, Any, Generic, Literal, TypeVar
 
 from pydantic import BaseModel, ConfigDict, Field
 
@@ -28,6 +28,9 @@
 from ..utils import AudioBuffer, aio, is_given
 from ..utils.audio import calculate_audio_duration
 
+if TYPE_CHECKING:
+    from ..voice.events import ConversationItemAddedEvent
+
 
 @unique
 class SpeechEventType(str, Enum):
@@ -128,6 +131,10 @@ class STTCapabilities:
     aligned_transcript: Literal["word", "chunk", False] = False
     offline_recognize: bool = True
     """Whether the STT supports batch recognition via recognize() method"""
+    keyterms: bool = False
+    """Whether the STT supports keyterm prompting"""
+    chat_context: bool = False
+    """Whether the STT can natively consume conversation context (see STT._push_conversation_item)"""
 
 
 class STTError(BaseModel):
@@ -152,6 +159,8 @@ def __init__(self, *, capabilities: STTCapabilities) -> None:
         self._capabilities = capabilities
         self._label = f"{type(self).__module__}.{type(self).__name__}"
         self._recognize_metrics_needed = True
+        self._keyterms_unsupported_warned = False
+        self._chat_context_unsupported_warned = False
 
     @property
     def label(self) -> str:
@@ -264,6 +273,37 @@ def _emit_error(self, api_error: Exception, recoverable: bool) -> None:
             ),
         )
 
+    def _update_session_keyterms(self, keyterms: list[str]) -> None:
+        """Set the framework-managed keyterms (session config + auto-detection).
+
+        Internal hook called by the framework, kept separate from the user's own keyterms
+        (constructor / ``update_options``). Plugins that support keyterms override this to
+        store the session set and apply it merged with the user keyterms.
+        """
+        if not self._capabilities.keyterms:
+            if not self._keyterms_unsupported_warned:
+                self._keyterms_unsupported_warned = True
+                logger.warning(
+                    "keyterms are not supported by this STT, ignoring keyterms update",
+                    extra={"stt": self._label},
+                )
+            return
+
+    def _push_conversation_item(self, ev: ConversationItemAddedEvent) -> None:
+        """Feed a new conversation turn to the STT to bias recognition (context carryover).
+
+        Plugins with native context support set ``STTCapabilities.chat_context`` and override
+        this to forward the item to their provider's carryover field.
+        """
+        if not self._capabilities.chat_context:
+            if not self._chat_context_unsupported_warned:
+                self._chat_context_unsupported_warned = True
+                logger.warning(
+                    "chat context is not supported by this STT, ignoring chat context update",
+                    extra={"stt": self._label},
+                )
+            return
+
     def stream(
         self,
         *,

@@ -18,6 +18,7 @@
     UserStateChangedEvent,
     UserTurnExceededEvent,
 )
+from .keyterm_detection import KeytermDetectionOptions, KeytermsOptions
 from .remote_session import RemoteSession
 from .room_io import (
     _ParticipantAudioOutput,
@@ -51,6 +52,8 @@
     "AgentFalseInterruptionEvent",
     "RemoteSession",
     "UserTurnExceededEvent",
+    "KeytermsOptions",
+    "KeytermDetectionOptions",
     "TranscriptSynchronizer",
     "io",
     "room_io",