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,
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,
        http_session: aiohttp.ClientSession | None = None,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> None:
        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._http_session = http_session

        self._warn_threshold_override()

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

    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,
        )

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

Initialize a new instance of EventEmitter.

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 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,
    )

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