Module agents.inference.turn

VideoSDK Inference Gateway Turn Detection Plugin.

HTTP-based End-of-Utterance (EOU) detector that delegates inference to the VideoSDK Inference Gateway's /v1/turn endpoint.

Three server-side backends are available via factory methods:

All three return a real float probability. No model is loaded locally — the server handles downloads, caching, and inference.

Example

from videosdk.inference import Turn

turn = Turn.namo(language="en") # Namo, English-specific model turn = Turn.turnsense() # TurnSense turn = Turn.videosdk() # VideoSDK BERT

pipeline = CascadingPipeline(stt=stt, llm=llm, tts=tts, turn_detector=turn)

Classes

class Turn (*,
provider: str = 'videosdk',
model_id: str = 'namo-turn-detector-v1',
language: Optional[str] = None,
threshold: float = 0.7,
base_url: Optional[str] = None,
timeout: float = 10.0,
max_connection_attempts: int = 5,
fallback_probability: float = 0.7)
Expand source code 
class Turn(EOU):
    """
    VideoSDK Inference Gateway Turn Detection Plugin.

    Delegates EOU scoring to the inference gateway's ``/v1/turn`` endpoint. No
    model is loaded locally — the server caches every Namo-Turn-Detector-v1
    language variant, so first-request latency per language is bounded by
    tokenize + inference time on the server (typically <50ms).

    Args:
        provider: Inference provider identifier (default: ``"videosdk"``).
        model_id: Turn detection model identifier
            (default: ``"namo-turn-detector-v1"``).
        language: Optional language code (e.g. ``"en"``, ``"hi"``, ``"es"``).
            Omit for the multilingual model.
        threshold: EOU probability threshold (default: ``0.7``).
        base_url: Override for the inference gateway URL.
        timeout: Per-request timeout in seconds.
        max_connection_attempts: After this many consecutive failed requests
            (non-200, timeout, invalid payload, etc.) the client stops calling
            the gateway and returns ``fallback_probability`` for the rest of
            the session. The counter resets on the next successful response.
            Default: 5.
        fallback_probability: Probability returned once the breaker has
            tripped. Default ``0.7`` matches the default threshold, so a
            disabled turn detector lets the pipeline assume end-of-utterance
            rather than stalling.
    """

    def __init__(
        self,
        *,
        provider: str = "videosdk",
        model_id: str = "namo-turn-detector-v1",
        language: Optional[str] = None,
        threshold: float = 0.7,
        base_url: Optional[str] = None,
        timeout: float = DEFAULT_TIMEOUT_SECONDS,
        max_connection_attempts: int = 5,
        fallback_probability: float = 0.7,
    ) -> None:
        super().__init__(threshold=threshold)

        self._videosdk_token = os.getenv("VIDEOSDK_AUTH_TOKEN")
        if not self._videosdk_token:
            raise ValueError(
                "VIDEOSDK_AUTH_TOKEN environment variable must be set for authentication"
            )

        self.provider = provider
        self.model_id = model_id
        self.language = language
        self.base_url = (base_url or DEFAULT_TURN_HTTP_URL).rstrip("/")
        self.timeout = timeout
        self.max_connection_attempts: int = max(1, int(max_connection_attempts))
        self.fallback_probability: float = float(fallback_probability)

        self._session: Optional[requests.Session] = None
        self._logged_first_success: bool = False

        # Circuit breaker: consecutive failed HTTP attempts. When it reaches
        # max_connection_attempts, every subsequent call returns
        # fallback_probability without hitting the network.
        self._consecutive_failures: int = 0
        self._turn_disabled: bool = False

        logger.info(
            f"[InferenceTurn] Configured (base_url={self.base_url}, "
            f"provider={self.provider}, model={self.model_id}, "
            f"language={self.language or 'multilingual'}, threshold={threshold}, "
            f"max_connection_attempts={self.max_connection_attempts}, "
            f"fallback_probability={self.fallback_probability})"
        )

    # ==================== Factory Methods ====================

    @staticmethod
    def namo(
        *,
        language: Optional[str] = None,
        threshold: float = 0.7,
        base_url: Optional[str] = None,
        timeout: float = DEFAULT_TIMEOUT_SECONDS,
        max_connection_attempts: int = 5,
        fallback_probability: float = 0.7,
    ) -> "Turn":
        """
        Create a Turn detector backed by the server-hosted Namo Turn Detector v1.

        Args:
            language: Optional language code (e.g. ``"en"``, ``"hi"``, ``"es"``).
                Omit for the multilingual model.
            threshold: EOU probability threshold (default: ``0.7``).
            base_url: Override for the inference gateway URL.
            timeout: Per-request timeout in seconds.
            max_connection_attempts: Failures before the breaker trips. Default: 5.
            fallback_probability: Probability emitted once disabled. Default: 0.7.
        """
        return Turn(
            provider="videosdk",
            model_id="namo-turn-detector-v1",
            language=language,
            threshold=threshold,
            base_url=base_url,
            timeout=timeout,
            max_connection_attempts=max_connection_attempts,
            fallback_probability=fallback_probability,
        )

    @staticmethod
    def turnsense(
        *,
        threshold: float = 0.7,
        base_url: Optional[str] = None,
        timeout: float = DEFAULT_TIMEOUT_SECONDS,
        max_connection_attempts: int = 5,
        fallback_probability: float = 0.7,
    ) -> "Turn":
        """
        Create a Turn detector backed by the server-hosted TurnSense model
        (latishab/turnsense, SmolLM2-135M, English).

        Args:
            threshold: EOU probability threshold (default: ``0.7``).
            base_url: Override for the inference gateway URL.
            timeout: Per-request timeout in seconds.
            max_connection_attempts: Failures before the breaker trips. Default: 5.
            fallback_probability: Probability emitted once disabled. Default: 0.7.
        """
        return Turn(
            provider="turnsense",
            model_id="latishab/turnsense",
            language=None,
            threshold=threshold,
            base_url=base_url,
            timeout=timeout,
            max_connection_attempts=max_connection_attempts,
            fallback_probability=fallback_probability,
        )

    @staticmethod
    def videosdk(
        *,
        threshold: float = 0.7,
        base_url: Optional[str] = None,
        timeout: float = DEFAULT_TIMEOUT_SECONDS,
        max_connection_attempts: int = 5,
        fallback_probability: float = 0.7,
    ) -> "Turn":
        """
        Create a Turn detector backed by the server-hosted VideoSDK BERT model
        (cdn.videosdk.live, English, binary classifier with softmax probability).

        Args:
            threshold: EOU probability threshold (default: ``0.7``).
            base_url: Override for the inference gateway URL.
            timeout: Per-request timeout in seconds.
            max_connection_attempts: Failures before the breaker trips. Default: 5.
            fallback_probability: Probability emitted once disabled. Default: 0.7.
        """
        return Turn(
            provider="videosdk",
            model_id="videosdk-turn-detector-v1",
            language=None,
            threshold=threshold,
            base_url=base_url,
            timeout=timeout,
            max_connection_attempts=max_connection_attempts,
            fallback_probability=fallback_probability,
        )

    # ==================== Circuit Breaker ====================

    def _record_attempt_failure(self, reason: str) -> float:
        """
        Increment the consecutive-failure counter and trip the breaker once it
        reaches ``max_connection_attempts``. Returns the probability value the
        caller should emit on this turn — ``0.0`` while pre-trip, or
        ``fallback_probability`` once disabled.
        """
        if self._turn_disabled:
            return self.fallback_probability

        self._consecutive_failures += 1
        logger.warning(
            f"[InferenceTurn] Attempt failure "
            f"{self._consecutive_failures}/{self.max_connection_attempts}: {reason}"
        )

        if self._consecutive_failures >= self.max_connection_attempts:
            self._turn_disabled = True
            logger.error(
                f"[InferenceTurn] Disabled for session after "
                f"{self._consecutive_failures} failed attempts — emitting "
                f"probability={self.fallback_probability} for the rest of the "
                f"turns (provider={self.provider}, model={self.model_id})"
            )
            return self.fallback_probability

        return 0.0

    def _record_attempt_success(self) -> None:
        """Reset the consecutive-failure counter on a successful response."""
        if self._consecutive_failures > 0:
            logger.info(
                f"[InferenceTurn] Recovered after {self._consecutive_failures} "
                f"failed attempt(s) — resetting counter"
            )
            self._consecutive_failures = 0

    # ==================== Core EOU Interface ====================

    def get_eou_probability(self, chat_context: ChatContext) -> float:
        """
        Return the EOU probability for the last user message in ``chat_context``.

        This runs a synchronous HTTP POST against the inference gateway because
        the base :class:`EOU` interface is synchronous and the caller invokes it
        from inside an ``asyncio`` task. Requests are expected to be short
        (<100 ms) and the timeout is capped by ``self.timeout``.

        After ``max_connection_attempts`` consecutive failures the client stops
        calling the gateway and returns ``fallback_probability`` for the rest
        of the session.
        """
        if self._turn_disabled:
            return self.fallback_probability

        payload = self._build_payload(chat_context)
        if payload is None:
            return 0.0

        try:
            session = self._get_session()
            url = f"{self.base_url}/v1/turn"
            headers = {
                "Content-Type": "application/json",
                "Authorization": f"Bearer {self._videosdk_token}",
            }

            response = session.post(
                url, json=payload, headers=headers, timeout=self.timeout
            )

            if response.status_code != 200:
                logger.error(
                    f"[InferenceTurn] HTTP {response.status_code}: {response.text}"
                )
                self.emit("error", f"HTTP {response.status_code}: {response.text}")
                return self._record_attempt_failure(
                    f"HTTP {response.status_code}"
                )

            data = response.json()
            probability = data.get("probability", 0.0)
            if not isinstance(probability, (int, float)):
                logger.error(f"[InferenceTurn] Invalid probability: {probability!r}")
                return self._record_attempt_failure("invalid probability payload")

            probability_f = float(probability)

            # A zero probability typically means the server omitted the field
            # (default 0.0 from .get) or returned a placeholder payload with
            # no real model signal. Treat it as a failed attempt for breaker
            # purposes, but keep returning 0.0 downstream while pre-trip so
            # EOU logic still sees "no end-of-utterance" as before.
            if probability_f == 0.0:
                return self._record_attempt_failure("probability=0.0")

            self._record_attempt_success()

            if not self._logged_first_success:
                self._logged_first_success = True
                logger.info(
                    f"[InferenceTurn] Connected successfully to {self.base_url}/v1/turn "
                    f"(first probability={probability_f:.4f})"
                )

            return probability_f

        except requests.Timeout:
            logger.error(f"[InferenceTurn] Request timed out after {self.timeout}s")
            self.emit("error", "turn detection request timed out")
            return self._record_attempt_failure("timeout")
        except requests.RequestException as e:
            logger.error(f"[InferenceTurn] Request failed: {e}")
            self.emit("error", f"turn detection request failed: {e}")
            return self._record_attempt_failure(f"request exception: {e}")
        except Exception as e:
            logger.error(f"[InferenceTurn] Unexpected error: {e}")
            self.emit("error", f"turn detection error: {e}")
            return self._record_attempt_failure(f"unexpected error: {e}")

    # ==================== Helpers ====================

    def _get_session(self) -> requests.Session:
        if self._session is None:
            self._session = requests.Session()
        return self._session

    def _build_payload(self, chat_context: ChatContext) -> Optional[Dict[str, Any]]:
        """Build the /v1/turn request body from a ChatContext."""
        items: List[Dict[str, Any]] = []
        for item in chat_context.items:
            if not isinstance(item, ChatMessage):
                continue
            if item.role != ChatRole.USER:
                continue
            text = self._content_to_text(item.content)
            if text:
                items.append({"role": "user", "content": text})

        if not items:
            return None

        # Only the last user message is inspected by the server, but send the
        # full user-turn list to match the documented request shape.
        payload: Dict[str, Any] = {
            "chatContext": {"items": items},
            "provider": self.provider,
            "modelId": self.model_id,
        }
        if self.language:
            payload["language"] = self.language
        return payload

    @staticmethod
    def _content_to_text(content: Any) -> str:
        if isinstance(content, str):
            return content.strip()
        if isinstance(content, list):
            parts: List[str] = []
            for c in content:
                if hasattr(c, "text"):
                    parts.append(str(getattr(c, "text") or ""))
                elif isinstance(c, str):
                    parts.append(c)
            return " ".join(p for p in parts if p).strip()
        return str(content or "").strip()

    # ==================== Cleanup ====================

    async def aclose(self) -> None:
        logger.info(f"[InferenceTurn] Closing Turn detector (language={self.language})")
        if self._session is not None:
            try:
                self._session.close()
            except Exception as e:
                logger.error(f"[InferenceTurn] Error closing HTTP session: {e}")
            self._session = None
        await super().aclose()

    # ==================== Properties ====================

    @property
    def label(self) -> str:
        lang = self.language or "multilingual"
        return f"videosdk.inference.Turn.{self.provider}.{self.model_id}.{lang}"

VideoSDK Inference Gateway Turn Detection Plugin.

Delegates EOU scoring to the inference gateway's /v1/turn endpoint. No model is loaded locally — the server caches every Namo-Turn-Detector-v1 language variant, so first-request latency per language is bounded by tokenize + inference time on the server (typically <50ms).

Args

provider
Inference provider identifier (default: "videosdk").
model_id
Turn detection model identifier (default: "namo-turn-detector-v1").
language
Optional language code (e.g. "en", "hi", "es"). Omit for the multilingual model.
threshold
EOU probability threshold (default: 0.7).
base_url
Override for the inference gateway URL.
timeout
Per-request timeout in seconds.
max_connection_attempts
After this many consecutive failed requests (non-200, timeout, invalid payload, etc.) the client stops calling the gateway and returns fallback_probability for the rest of the session. The counter resets on the next successful response. Default: 5.
fallback_probability
Probability returned once the breaker has tripped. Default 0.7 matches the default threshold, so a disabled turn detector lets the pipeline assume end-of-utterance rather than stalling.

Ancestors

  • videosdk.agents.eou.EOU
  • videosdk.agents.event_emitter.EventEmitter
  • typing.Generic

Static methods

def namo(*,
language: Optional[str] = None,
threshold: float = 0.7,
base_url: Optional[str] = None,
timeout: float = 10.0,
max_connection_attempts: int = 5,
fallback_probability: float = 0.7) ‑> Turn
Expand source code 
@staticmethod
def namo(
    *,
    language: Optional[str] = None,
    threshold: float = 0.7,
    base_url: Optional[str] = None,
    timeout: float = DEFAULT_TIMEOUT_SECONDS,
    max_connection_attempts: int = 5,
    fallback_probability: float = 0.7,
) -> "Turn":
    """
    Create a Turn detector backed by the server-hosted Namo Turn Detector v1.

    Args:
        language: Optional language code (e.g. ``"en"``, ``"hi"``, ``"es"``).
            Omit for the multilingual model.
        threshold: EOU probability threshold (default: ``0.7``).
        base_url: Override for the inference gateway URL.
        timeout: Per-request timeout in seconds.
        max_connection_attempts: Failures before the breaker trips. Default: 5.
        fallback_probability: Probability emitted once disabled. Default: 0.7.
    """
    return Turn(
        provider="videosdk",
        model_id="namo-turn-detector-v1",
        language=language,
        threshold=threshold,
        base_url=base_url,
        timeout=timeout,
        max_connection_attempts=max_connection_attempts,
        fallback_probability=fallback_probability,
    )

Create a Turn detector backed by the server-hosted Namo Turn Detector v1.

Args

language
Optional language code (e.g. "en", "hi", "es"). Omit for the multilingual model.
threshold
EOU probability threshold (default: 0.7).
base_url
Override for the inference gateway URL.
timeout
Per-request timeout in seconds.
max_connection_attempts
Failures before the breaker trips. Default: 5.
fallback_probability
Probability emitted once disabled. Default: 0.7.
def turnsense(*,
threshold: float = 0.7,
base_url: Optional[str] = None,
timeout: float = 10.0,
max_connection_attempts: int = 5,
fallback_probability: float = 0.7) ‑> Turn
Expand source code 
@staticmethod
def turnsense(
    *,
    threshold: float = 0.7,
    base_url: Optional[str] = None,
    timeout: float = DEFAULT_TIMEOUT_SECONDS,
    max_connection_attempts: int = 5,
    fallback_probability: float = 0.7,
) -> "Turn":
    """
    Create a Turn detector backed by the server-hosted TurnSense model
    (latishab/turnsense, SmolLM2-135M, English).

    Args:
        threshold: EOU probability threshold (default: ``0.7``).
        base_url: Override for the inference gateway URL.
        timeout: Per-request timeout in seconds.
        max_connection_attempts: Failures before the breaker trips. Default: 5.
        fallback_probability: Probability emitted once disabled. Default: 0.7.
    """
    return Turn(
        provider="turnsense",
        model_id="latishab/turnsense",
        language=None,
        threshold=threshold,
        base_url=base_url,
        timeout=timeout,
        max_connection_attempts=max_connection_attempts,
        fallback_probability=fallback_probability,
    )

Create a Turn detector backed by the server-hosted TurnSense model (latishab/turnsense, SmolLM2-135M, English).

Args

threshold
EOU probability threshold (default: 0.7).
base_url
Override for the inference gateway URL.
timeout
Per-request timeout in seconds.
max_connection_attempts
Failures before the breaker trips. Default: 5.
fallback_probability
Probability emitted once disabled. Default: 0.7.
def videosdk(*,
threshold: float = 0.7,
base_url: Optional[str] = None,
timeout: float = 10.0,
max_connection_attempts: int = 5,
fallback_probability: float = 0.7) ‑> Turn
Expand source code 
@staticmethod
def videosdk(
    *,
    threshold: float = 0.7,
    base_url: Optional[str] = None,
    timeout: float = DEFAULT_TIMEOUT_SECONDS,
    max_connection_attempts: int = 5,
    fallback_probability: float = 0.7,
) -> "Turn":
    """
    Create a Turn detector backed by the server-hosted VideoSDK BERT model
    (cdn.videosdk.live, English, binary classifier with softmax probability).

    Args:
        threshold: EOU probability threshold (default: ``0.7``).
        base_url: Override for the inference gateway URL.
        timeout: Per-request timeout in seconds.
        max_connection_attempts: Failures before the breaker trips. Default: 5.
        fallback_probability: Probability emitted once disabled. Default: 0.7.
    """
    return Turn(
        provider="videosdk",
        model_id="videosdk-turn-detector-v1",
        language=None,
        threshold=threshold,
        base_url=base_url,
        timeout=timeout,
        max_connection_attempts=max_connection_attempts,
        fallback_probability=fallback_probability,
    )

Create a Turn detector backed by the server-hosted VideoSDK BERT model (cdn.videosdk.live, English, binary classifier with softmax probability).

Args

threshold
EOU probability threshold (default: 0.7).
base_url
Override for the inference gateway URL.
timeout
Per-request timeout in seconds.
max_connection_attempts
Failures before the breaker trips. Default: 5.
fallback_probability
Probability emitted once disabled. Default: 0.7.

Instance variables

prop label : str
Expand source code 
@property
def label(self) -> str:
    lang = self.language or "multilingual"
    return f"videosdk.inference.Turn.{self.provider}.{self.model_id}.{lang}"

Get the EOU provider label

Methods

async def aclose(self) ‑> None
Expand source code 
async def aclose(self) -> None:
    logger.info(f"[InferenceTurn] Closing Turn detector (language={self.language})")
    if self._session is not None:
        try:
            self._session.close()
        except Exception as e:
            logger.error(f"[InferenceTurn] Error closing HTTP session: {e}")
        self._session = None
    await super().aclose()

Cleanup resources - should be overridden by subclasses to cleanup models

def get_eou_probability(self, chat_context: ChatContext) ‑> float
Expand source code 
def get_eou_probability(self, chat_context: ChatContext) -> float:
    """
    Return the EOU probability for the last user message in ``chat_context``.

    This runs a synchronous HTTP POST against the inference gateway because
    the base :class:`EOU` interface is synchronous and the caller invokes it
    from inside an ``asyncio`` task. Requests are expected to be short
    (<100 ms) and the timeout is capped by ``self.timeout``.

    After ``max_connection_attempts`` consecutive failures the client stops
    calling the gateway and returns ``fallback_probability`` for the rest
    of the session.
    """
    if self._turn_disabled:
        return self.fallback_probability

    payload = self._build_payload(chat_context)
    if payload is None:
        return 0.0

    try:
        session = self._get_session()
        url = f"{self.base_url}/v1/turn"
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self._videosdk_token}",
        }

        response = session.post(
            url, json=payload, headers=headers, timeout=self.timeout
        )

        if response.status_code != 200:
            logger.error(
                f"[InferenceTurn] HTTP {response.status_code}: {response.text}"
            )
            self.emit("error", f"HTTP {response.status_code}: {response.text}")
            return self._record_attempt_failure(
                f"HTTP {response.status_code}"
            )

        data = response.json()
        probability = data.get("probability", 0.0)
        if not isinstance(probability, (int, float)):
            logger.error(f"[InferenceTurn] Invalid probability: {probability!r}")
            return self._record_attempt_failure("invalid probability payload")

        probability_f = float(probability)

        # A zero probability typically means the server omitted the field
        # (default 0.0 from .get) or returned a placeholder payload with
        # no real model signal. Treat it as a failed attempt for breaker
        # purposes, but keep returning 0.0 downstream while pre-trip so
        # EOU logic still sees "no end-of-utterance" as before.
        if probability_f == 0.0:
            return self._record_attempt_failure("probability=0.0")

        self._record_attempt_success()

        if not self._logged_first_success:
            self._logged_first_success = True
            logger.info(
                f"[InferenceTurn] Connected successfully to {self.base_url}/v1/turn "
                f"(first probability={probability_f:.4f})"
            )

        return probability_f

    except requests.Timeout:
        logger.error(f"[InferenceTurn] Request timed out after {self.timeout}s")
        self.emit("error", "turn detection request timed out")
        return self._record_attempt_failure("timeout")
    except requests.RequestException as e:
        logger.error(f"[InferenceTurn] Request failed: {e}")
        self.emit("error", f"turn detection request failed: {e}")
        return self._record_attempt_failure(f"request exception: {e}")
    except Exception as e:
        logger.error(f"[InferenceTurn] Unexpected error: {e}")
        self.emit("error", f"turn detection error: {e}")
        return self._record_attempt_failure(f"unexpected error: {e}")

Return the EOU probability for the last user message in chat_context.

This runs a synchronous HTTP POST against the inference gateway because the base :class:EOU interface is synchronous and the caller invokes it from inside an asyncio task. Requests are expected to be short (<100 ms) and the timeout is capped by self.timeout.

After max_connection_attempts consecutive failures the client stops calling the gateway and returns fallback_probability for the rest of the session.