Module livekit.agents.inference.eot.detector

Audio end-of-turn detector with cloud → local fallback.

Classes

class TurnDetector (*,
version: NotGivenOr[TurnDetectorVersions] = NOT_GIVEN,
unlikely_threshold: NotGivenOr[float | dict[LanguageCode | str, float]] = NOT_GIVEN,
backchannel_threshold: NotGivenOr[float | dict[LanguageCode | str, float]] = NOT_GIVEN,
base_url: NotGivenOr[str] = NOT_GIVEN,
api_key: NotGivenOr[str] = NOT_GIVEN,
api_secret: NotGivenOr[str] = NOT_GIVEN,
sample_rate: int = 16000,
local_fallback: bool = True,
http_session: aiohttp.ClientSession | None = None,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0))

Expand source code

class TurnDetector(_BaseStreamingTurnDetector):
    def __init__(
        self,
        *,
        version: NotGivenOr[TurnDetectorVersions] = NOT_GIVEN,
        unlikely_threshold: NotGivenOr[float | dict[LanguageCode | str, float]] = NOT_GIVEN,
        backchannel_threshold: NotGivenOr[float | dict[LanguageCode | str, float]] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        sample_rate: int = DEFAULT_SAMPLE_RATE,
        local_fallback: bool = True,
        http_session: aiohttp.ClientSession | None = None,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> None:
        """
        Args:
            local_fallback: Whether a ``v1`` detector may degrade to the local
                ``v1-mini`` model when the gateway fails. False keeps it cloud-only,
                so the mini weights (~108MB, resident for the process' life) are never
                loaded and turns commit on the endpointing delay instead.
        """
        auto = not is_given(version)
        resolved_version: TurnDetectorVersions = (
            version
            if is_given(version)
            else ("v1" if (utils.is_hosted() or utils.is_dev_mode()) else "v1-mini")
        )
        resolved_model: TurnDetectorModels = (
            "turn-detector-v1" if resolved_version == "v1" else "turn-detector-v1-mini"
        )

        cloud_opts: _CloudTransportOptions | None = None

        if resolved_version == "v1":
            lk_base_url = utils.resolve_env_var(
                base_url,
                "LIVEKIT_INFERENCE_URL",
                default=get_default_inference_url(),
            )
            lk_api_key = utils.resolve_env_var(
                api_key, "LIVEKIT_INFERENCE_API_KEY", "LIVEKIT_API_KEY", default=""
            )
            lk_api_secret = utils.resolve_env_var(
                api_secret,
                "LIVEKIT_INFERENCE_API_SECRET",
                "LIVEKIT_API_SECRET",
                default="",
            )
            missing: list[str] = []
            if not lk_base_url:
                missing.append("LIVEKIT_INFERENCE_URL")
            if not lk_api_key:
                missing.append("LIVEKIT_API_KEY")
            if not lk_api_secret:
                missing.append("LIVEKIT_API_SECRET")
            if missing:
                if auto:
                    logger.warning(
                        "LIVEKIT_INFERENCE_URL is set but %s missing; "
                        "falling back to the turn-detector-v1-mini model",
                        ", ".join(missing),
                    )
                    resolved_model = "turn-detector-v1-mini"
                else:
                    raise ValueError(
                        f"TurnDetector(version='v1') requires "
                        f"{', '.join(missing)} (env or constructor argument)."
                    )
            else:
                cloud_opts = _CloudTransportOptions(
                    base_url=lk_base_url,
                    api_key=lk_api_key,
                    api_secret=lk_api_secret,
                    conn_options=conn_options,
                )

        opts = TurnDetectorOptions(
            sample_rate=sample_rate,
            thresholds=ThresholdOptions(resolved_model, unlikely_threshold, backchannel_threshold),
        )
        super().__init__(opts=opts)

        self._model: TurnDetectorModels = resolved_model
        self._cloud_opts = cloud_opts
        self._local_fallback = local_fallback
        self._http_session = http_session

        if not local_fallback and resolved_model == "turn-detector-v1-mini":
            logger.warning(
                "local_fallback=False has no effect on the %s model, which runs locally by design",
                resolved_model,
            )

        self._warn_threshold_override()

    @property
    def model(self) -> TurnDetectorModels:
        return self._model

    def describe_options(self) -> dict[str, Any]:
        """What the session report shows for this detector (``telemetry.DescribesOptions``):
        the model and where it runs, plus the threshold overrides when the user set any.
        Server-calibrated defaults are not repeated here; credentials and endpoints never."""
        options: dict[str, Any] = {
            "model": self.model,
            "provider": self.provider,
            "sample_rate": self._opts.sample_rate,
            "local_fallback": self._local_fallback,
        }
        thresholds = self._opts.thresholds
        if is_given(thresholds.overrides):
            options["threshold_overrides"] = thresholds.overrides
        if is_given(thresholds.backchannel_overrides):
            options["backchannel_threshold_overrides"] = thresholds.backchannel_overrides
        return options

    def _warn_threshold_override(self) -> None:
        thresholds = self._opts.thresholds
        if is_given(overrides := thresholds.overrides):
            logger.warning(
                "a non-default turn detection threshold was provided "
                "(unlikely_threshold=%s); the server provides calibrated defaults and "
                "overriding them may be suboptimal",
                overrides,
            )
        if is_given(bc_overrides := thresholds.backchannel_overrides):
            logger.warning(
                "a non-default backchannel threshold was provided "
                "(backchannel_threshold=%s); the server provides calibrated defaults and "
                "overriding them may be suboptimal",
                bc_overrides,
            )

    def update_options(
        self,
        *,
        unlikely_threshold: NotGivenOr[float | dict[LanguageCode | str, float]] = NOT_GIVEN,
        backchannel_threshold: NotGivenOr[float | dict[LanguageCode | str, float]] = NOT_GIVEN,
    ) -> None:
        if is_given(unlikely_threshold):
            self._opts.thresholds.update_overrides(unlikely_threshold)
        if is_given(backchannel_threshold):
            self._opts.thresholds.update_backchannel_overrides(backchannel_threshold)
        self._warn_threshold_override()

    def stream(
        self,
        *,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> _BaseStreamingTurnDetectorStream:
        cloud_opts = (
            replace(self._cloud_opts, conn_options=conn_options)
            if self._cloud_opts is not None
            else None
        )

        transport: _StreamingTurnDetectionTransport
        if self._model == "turn-detector-v1":
            assert cloud_opts is not None, "turn-detector-v1 requires cloud_opts"
            transport = _CloudTransport(
                detector=self,
                opts=self._opts,
                cloud_opts=cloud_opts,
                http_session=self._http_session,
            )
        else:
            transport = _LocalTransport(opts=self._opts)

        return _BaseStreamingTurnDetectorStream(
            detector=self,
            opts=self._opts,
            transport=transport,
            model=self._model,
            local_fallback=self._local_fallback,
        )

Abstract base class for generic types.

On Python 3.12 and newer, generic classes implicitly inherit from Generic when they declare a parameter list after the class's name::

class Mapping[KT, VT]:
    def __getitem__(self, key: KT) -> VT:
        ...
    # Etc.

On older versions of Python, however, generic classes have to explicitly inherit from Generic.

After a class has been declared to be generic, it can then be used as follows::

def lookup_name[KT, VT](mapping: Mapping[KT, VT], key: KT, default: VT) -> VT:
    try:
        return mapping[key]
    except KeyError:
        return default

Args

local_fallback

Whether a v1 detector may degrade to the local v1-mini model when the gateway fails. False keeps it cloud-only, so the mini weights (~108MB, resident for the process' life) are never loaded and turns commit on the endpointing delay instead.

Ancestors

• livekit.agents.inference.eot.base._BaseStreamingTurnDetector
• EventEmitter
• typing.Generic

Instance variables

prop model : TurnDetectorModels

Expand source code

@property
def model(self) -> TurnDetectorModels:
    return self._model

Methods

def describe_options(self) ‑> dict[str, typing.Any]

Expand source code

def describe_options(self) -> dict[str, Any]:
    """What the session report shows for this detector (``telemetry.DescribesOptions``):
    the model and where it runs, plus the threshold overrides when the user set any.
    Server-calibrated defaults are not repeated here; credentials and endpoints never."""
    options: dict[str, Any] = {
        "model": self.model,
        "provider": self.provider,
        "sample_rate": self._opts.sample_rate,
        "local_fallback": self._local_fallback,
    }
    thresholds = self._opts.thresholds
    if is_given(thresholds.overrides):
        options["threshold_overrides"] = thresholds.overrides
    if is_given(thresholds.backchannel_overrides):
        options["backchannel_threshold_overrides"] = thresholds.backchannel_overrides
    return options

What the session report shows for this detector (telemetry.DescribesOptions): the model and where it runs, plus the threshold overrides when the user set any. Server-calibrated defaults are not repeated here; credentials and endpoints never.

def stream(self,
*,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0)) ‑> livekit.agents.inference.eot.base._BaseStreamingTurnDetectorStream

Expand source code

def stream(
    self,
    *,
    conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
) -> _BaseStreamingTurnDetectorStream:
    cloud_opts = (
        replace(self._cloud_opts, conn_options=conn_options)
        if self._cloud_opts is not None
        else None
    )

    transport: _StreamingTurnDetectionTransport
    if self._model == "turn-detector-v1":
        assert cloud_opts is not None, "turn-detector-v1 requires cloud_opts"
        transport = _CloudTransport(
            detector=self,
            opts=self._opts,
            cloud_opts=cloud_opts,
            http_session=self._http_session,
        )
    else:
        transport = _LocalTransport(opts=self._opts)

    return _BaseStreamingTurnDetectorStream(
        detector=self,
        opts=self._opts,
        transport=transport,
        model=self._model,
        local_fallback=self._local_fallback,
    )

def update_options(self,
*,
unlikely_threshold: NotGivenOr[float | dict[LanguageCode | str, float]] = NOT_GIVEN,
backchannel_threshold: NotGivenOr[float | dict[LanguageCode | str, float]] = NOT_GIVEN) ‑> None

Expand source code

def update_options(
    self,
    *,
    unlikely_threshold: NotGivenOr[float | dict[LanguageCode | str, float]] = NOT_GIVEN,
    backchannel_threshold: NotGivenOr[float | dict[LanguageCode | str, float]] = NOT_GIVEN,
) -> None:
    if is_given(unlikely_threshold):
        self._opts.thresholds.update_overrides(unlikely_threshold)
    if is_given(backchannel_threshold):
        self._opts.thresholds.update_backchannel_overrides(backchannel_threshold)
    self._warn_threshold_override()

Inherited members

• EventEmitter:
  • emit
  • off
  • on
  • once