/livekit-agents/livekit/agents/voice/endpointing.py

from ..log import logger
from ..types import NOT_GIVEN, NotGivenOr
from ..utils import is_given
from ..utils.exp_filter import ExpFilter
from .turn import EndpointingOptions

_AGENT_SPEECH_LEADING_SILENCE_GRACE_PERIOD = 0.25  # seconds


class BaseEndpointing:
    def __init__(self, min_delay: float, max_delay: float):
        self._min_delay = min_delay
        self._max_delay = max_delay
        self._overlapping = False

    def update_options(
        self, *, min_delay: NotGivenOr[float] = NOT_GIVEN, max_delay: NotGivenOr[float] = NOT_GIVEN
    ) -> None:
        if is_given(min_delay):
            self._min_delay = min_delay
        if is_given(max_delay):
            self._max_delay = max_delay

    @property
    def min_delay(self) -> float:
        return self._min_delay

    @property
    def max_delay(self) -> float:
        return self._max_delay

    @property
    def overlapping(self) -> bool:
        return self._overlapping

    def on_start_of_speech(self, started_at: float, overlapping: bool = False) -> None:
        self._overlapping = overlapping

    def on_end_of_speech(self, ended_at: float, should_ignore: bool = False) -> None:
        self._overlapping = False

    def on_start_of_agent_speech(self, started_at: float) -> None:
        pass

    def on_end_of_agent_speech(self, ended_at: float) -> None:
        pass


class DynamicEndpointing(BaseEndpointing):
    def __init__(self, min_delay: float, max_delay: float, alpha: float = 0.9):
        """
        Dynamically adjust the endpointing delay based on the speech activity.

        Args:
            min_delay: Minimum delay in seconds.
            max_delay: Maximum delay in seconds.
            alpha: Exponential moving average coefficient. The higher the value, the more weight is given to the history. Defaults to 0.9.

        The endpointing delay is adjusted based on the following information:

        1. Pauses between utterances:

        [utterance] [pause] [utterance] [pause] [utterance] (<- min delay should cover this)

        2. Pauses between an utterance and next immediate interruption:

        [utterance] [   pause   ] [immediate interruption] (<- this should be a false EOT, and min delay should cover this)
                        [agent speech interrupted]

        3. Pauses between a user utterance and agent speech:

        [utterance] [pause]                  (<- max delay should cover this)
                           [agent speech]    (this could be interrupted later, but that would be the next turn)
        """

        super().__init__(min_delay=min_delay, max_delay=max_delay)

        self._utterance_pause = ExpFilter(
            alpha=alpha, initial=min_delay, min_val=min_delay, max_val=max_delay
        )
        self._turn_pause = ExpFilter(
            alpha=alpha, initial=max_delay, min_val=min_delay, max_val=max_delay
        )

        self._utterance_started_at: float | None = None
        self._utterance_ended_at: float | None = None
        self._agent_speech_started_at: float | None = None
        self._agent_speech_ended_at: float | None = None
        self._speaking = False

    @property
    def min_delay(self) -> float:
        return (
            self._utterance_pause.value
            if self._utterance_pause.value is not None
            else self._min_delay
        )

    @property
    def max_delay(self) -> float:
        turn_val = self._turn_pause.value if self._turn_pause.value is not None else self._max_delay
        return max(turn_val, self.min_delay)

    @property
    def between_utterance_delay(self) -> float:
        if self._utterance_ended_at is None:
            return 0.0
        if self._utterance_started_at is None:
            return 0.0

        return max(0, self._utterance_started_at - self._utterance_ended_at)

    @property
    def between_turn_delay(self) -> float:
        if self._agent_speech_started_at is None:
            return 0.0
        if self._utterance_ended_at is None:
            return 0.0

        return max(0, self._agent_speech_started_at - self._utterance_ended_at)

    @property
    def immediate_interruption_delay(self) -> tuple[float, float]:
        """
        Returns the two pauses in the following case:
        [utterance] [first val][second val] [immediate interruption]
                               [agent speech interrupted]
        """
        if self._utterance_started_at is None:
            return 0.0, 0.0
        if self._agent_speech_started_at is None:
            return 0.0, 0.0

        return (
            self.between_turn_delay,
            abs(self.between_utterance_delay - self.between_turn_delay),
        )

    def on_start_of_agent_speech(self, started_at: float) -> None:
        self._agent_speech_started_at = started_at
        self._agent_speech_ended_at = None
        self._overlapping = False

    def on_end_of_agent_speech(self, ended_at: float) -> None:
        # NOTE: intentionally keep _agent_speech_started_at so that
        # between_turn_delay can be computed in the normal end-of-speech path
        # NOTE: we also guard against duplicate calls from pipeline reply and pipeline reply done
        if self._agent_speech_started_at is not None and (
            self._agent_speech_ended_at is None
            or self._agent_speech_ended_at < self._agent_speech_started_at
        ):
            self._agent_speech_ended_at = ended_at
        self._overlapping = False

    def on_start_of_speech(self, started_at: float, overlapping: bool = False) -> None:
        if self._overlapping:
            # duplicate calls from _interrupt_by_audio_activity and on_start_of_speech
            return

        # VAD interrupt by audio activity is triggered before end of speech is detected
        # adjust the utterance ended time to be just before the agent speech started
        if (
            self._utterance_started_at is not None
            and self._utterance_ended_at is not None
            and self._agent_speech_started_at is not None
            and self._utterance_ended_at < self._utterance_started_at
            and overlapping
        ):
            self._utterance_ended_at = self._agent_speech_started_at - 1e-3
            logger.trace(
                "utterance ended at adjusted: %s",
                self._utterance_ended_at,
            )

        self._utterance_started_at = started_at
        self._overlapping = overlapping
        self._speaking = True

    def on_end_of_speech(self, ended_at: float, should_ignore: bool = False) -> None:
        if should_ignore and self._overlapping:
            # If user speech started within _AGENT_SPEECH_LEADING_SILENCE_GRACE_PERIOD of agent speech,
            # don't ignore — TTS leading silence can cause the agent speech timestamp
            # to precede actual audible audio, making this look like a backchannel
            # when it's really the user speaking before hearing the agent.
            if (
                self._utterance_started_at is not None
                and self._agent_speech_started_at is not None
                and abs(self._utterance_started_at - self._agent_speech_started_at)
                < _AGENT_SPEECH_LEADING_SILENCE_GRACE_PERIOD
            ):
                logger.trace(
                    "ignoring should_ignore=True: user speech started within %.3fs of agent speech "
                    "(within grace period of %.3fs)",
                    abs(self._utterance_started_at - self._agent_speech_started_at),
                    _AGENT_SPEECH_LEADING_SILENCE_GRACE_PERIOD,
                )
            else:
                # skip update because it might be a backchannel
                self._overlapping = False
                self._speaking = False
                self._utterance_started_at = None
                self._utterance_ended_at = None
                return

        if self._overlapping or (
            self._agent_speech_started_at is not None and self._agent_speech_ended_at is None
        ):  # this is an interruption (agent is still speaking)
            # If this is an immediate interruption, update the min delay (case 2)
            turn_delay, interruption_delay = self.immediate_interruption_delay
            if (
                (0 < interruption_delay <= self.min_delay)
                and (0 < turn_delay <= self.max_delay)
                and (pause := self.between_utterance_delay) > 0
            ):
                prev_val = self.min_delay
                self._utterance_pause.apply(1.0, pause)
                logger.debug(
                    "min endpointing delay updated: %s -> %s",
                    prev_val,
                    self.min_delay,
                    extra={
                        "reason": "immediate interruption",
                        "pause": pause,
                        "interruption_delay": interruption_delay,
                        "turn_delay": turn_delay,
                        "max_delay": self.max_delay,
                        "min_delay": self.min_delay,
                    },
                )
            # If this is not an immediate interruption, update the max delay (case 3)
            elif (pause := self.between_turn_delay) > 0:
                prev_val = self.max_delay
                self._turn_pause.apply(1.0, pause)
                logger.debug(
                    "max endpointing delay updated: %s -> %s",
                    prev_val,
                    self.max_delay,
                    extra={
                        "reason": "new turn (interruption)",
                        "pause": pause,
                        "max_delay": self.max_delay,
                        "min_delay": self.min_delay,
                        "between_utterance_delay": self.between_utterance_delay,
                        "between_turn_delay": self.between_turn_delay,
                    },
                )

        else:  # this is a normal end of speech
            if (pause := self.between_turn_delay) > 0:
                prev_val = self.max_delay
                self._turn_pause.apply(1.0, pause)
                logger.debug(
                    "max endpointing delay updated due to pause: %s -> %s",
                    prev_val,
                    self.max_delay,
                    extra={
                        "reason": "new turn",
                        "pause": pause,
                        "max_delay": self.max_delay,
                        "min_delay": self.min_delay,
                    },
                )
            elif (
                (pause := self.between_utterance_delay) > 0
                and self._agent_speech_ended_at is None
                and self._agent_speech_started_at is None
            ):
                prev_val = self.min_delay
                self._utterance_pause.apply(1.0, pause)
                logger.debug(
                    "min endpointing delay updated: %s -> %s",
                    prev_val,
                    self.min_delay,
                    extra={
                        "reason": "pause between utterances",
                        "pause": pause,
                        "max_delay": self.max_delay,
                        "min_delay": self.min_delay,
                    },
                )

        self._utterance_ended_at = ended_at
        self._agent_speech_started_at = None
        self._agent_speech_ended_at = None
        self._speaking = False
        self._overlapping = False

    def update_options(
        self,
        *,
        min_delay: NotGivenOr[float] = NOT_GIVEN,
        max_delay: NotGivenOr[float] = NOT_GIVEN,
        alpha: NotGivenOr[float] = NOT_GIVEN,
    ) -> None:
        if is_given(min_delay):
            self._min_delay = min_delay
            self._utterance_pause.reset(initial=self._min_delay, min_val=self._min_delay)
            self._turn_pause.reset(min_val=self._min_delay)

        if is_given(max_delay):
            self._max_delay = max_delay
            self._turn_pause.reset(initial=self._max_delay, max_val=self._max_delay)
            self._utterance_pause.reset(max_val=self._max_delay)

        if is_given(alpha):
            self._utterance_pause.reset(alpha=alpha)
            self._turn_pause.reset(alpha=alpha)


def create_endpointing(options: EndpointingOptions) -> BaseEndpointing:
    match options.get("mode", "fixed"):
        case "dynamic":
            return DynamicEndpointing(
                min_delay=options["min_delay"],
                max_delay=options["max_delay"],
                alpha=options["alpha"],
            )
        case _:
            return BaseEndpointing(
                min_delay=options["min_delay"],
                max_delay=options["max_delay"],
            )