feat(voice): add configurable latency budget events

livekit-agents/livekit/agents/__init__.py

@@ -91,6 +91,8 @@
     ErrorEvent,
     ExpressiveOptions,
     FunctionToolsExecutedEvent,
+    LatencyBudgetEvent,
+    LatencyBudgetOptions,
     MetricsCollectedEvent,
     ModelSettings,
     NonverbalOptions,
@@ -224,6 +226,8 @@ def __getattr__(name: str) -> typing.Any:
     "SimulationRun",
     "SimulationVerdict",
     "AgentSession",
+    "LatencyBudgetEvent",
+    "LatencyBudgetOptions",
     "AudioRecognition",
     "ExpressiveOptions",
     "NonverbalOptions",

livekit-agents/livekit/agents/voice/__init__.py

@@ -3,6 +3,7 @@
 from .agent_session import (
     AgentSession,
     ExpressiveOptions,
+    LatencyBudgetOptions,
     NonverbalOptions,
     RecordingOptions,
     SpeechSteeringOptions,
@@ -18,6 +19,7 @@
     ConversationItemAddedEvent,
     ErrorEvent,
     FunctionToolsExecutedEvent,
+    LatencyBudgetEvent,
     MetricsCollectedEvent,
     RunContext,
     SessionUsageUpdatedEvent,
@@ -50,6 +52,7 @@
 __all__ = [
     "AgentSession",
     "ExpressiveOptions",
+    "LatencyBudgetOptions",
     "NonverbalOptions",
     "RecordingOptions",
     "SpeechSteeringOptions",
@@ -63,6 +66,7 @@
     "UserInputTranscribedEvent",
     "AgentEvent",
     "MetricsCollectedEvent",
+    "LatencyBudgetEvent",
     "SessionUsageUpdatedEvent",
     "ConversationItemAddedEvent",
     "SpeechCreatedEvent",

livekit-agents/livekit/agents/voice/agent_activity.py

@@ -337,6 +337,7 @@ def __init__(self, agent: Agent, sess: AgentSession) -> None:
         # placeholder used to hold a RunResult open while waiting for a realtime
         # model to auto-generate a tool reply (auto_tool_reply_generation=True).
         self._pending_auto_tool_reply_fut: asyncio.Future[None] | None = None
+        self._realtime_user_stopped_speaking_at: float | None = None
 
     def _resolve_rt_turn_detection_enabled(self) -> bool:
         """Whether a realtime model's server-side turn detection is on for this session.
@@ -2053,6 +2054,7 @@ def _on_input_speech_started(self, _: llm.InputSpeechStartedEvent) -> None:
                 )
 
     def _on_input_speech_stopped(self, ev: llm.InputSpeechStoppedEvent) -> None:
+        self._realtime_user_stopped_speaking_at = time.time()
         if self.vad is None or self.using_default_vad:
             if self._audio_recognition:
                 self._audio_recognition._on_end_of_speech(
@@ -2113,6 +2115,7 @@ def _on_generation_created(self, ev: llm.GenerationCreatedEvent) -> None:
                 speech_handle=handle,
                 generation_ev=ev,
                 model_settings=ModelSettings(),
+                user_stopped_speaking_at=self._realtime_user_stopped_speaking_at,
             ),
             speech_handle=handle,
             name="AgentActivity.realtime_generation",
@@ -2564,6 +2567,7 @@ async def _user_turn_completed_task(
                         self._agent._chat_ctx.items.append(user_message)
                         self._session._conversation_item_added(user_message)
                     return
+                self._realtime_user_stopped_speaking_at = info.metrics.stopped_speaking_at
                 self._rt_session.commit_audio()
 
         if info.skip_reply:
@@ -3470,6 +3474,9 @@ def _on_first_frame(
                 early_metrics["e2e_latency"] = (
                     started_speaking_at - user_metrics["stopped_speaking_at"]
                 )
+                self._session._evaluate_latency_budget(
+                    latency=early_metrics["e2e_latency"], speech_id=speech_handle.id
+                )
             self._session._early_assistant_metrics = early_metrics
 
             self._session._update_agent_state(
@@ -3924,6 +3931,7 @@ async def _realtime_generation_task(
         generation_ev: llm.GenerationCreatedEvent,
         model_settings: ModelSettings,
         instructions: str | None = None,
+        user_stopped_speaking_at: float | None = None,
     ) -> None:
         with tracer.start_as_current_span(
             "agent_turn", context=self._session._root_span_context
@@ -3948,6 +3956,7 @@ async def _realtime_generation_task(
                     generation_ev=generation_ev,
                     model_settings=model_settings,
                     instructions=instructions,
+                    user_stopped_speaking_at=user_stopped_speaking_at,
                     inference_span=inference_span,
                 )
             finally:
@@ -3963,6 +3972,7 @@ async def _realtime_generation_task_impl(
         generation_ev: llm.GenerationCreatedEvent,
         model_settings: ModelSettings,
         instructions: str | None = None,
+        user_stopped_speaking_at: float | None = None,
         inference_span: trace.Span,
     ) -> None:
         current_span = trace.get_current_span(context=speech_handle._agent_turn_context)
@@ -4081,6 +4091,16 @@ def _on_first_frame(
             except BaseException:
                 return
 
+            if (
+                user_stopped_speaking_at is not None
+                and self._realtime_user_stopped_speaking_at == user_stopped_speaking_at
+            ):
+                self._session._evaluate_latency_budget(
+                    latency=started_speaking_at - user_stopped_speaking_at,
+                    speech_id=speech_handle.id,
+                )
+                self._realtime_user_stopped_speaking_at = None
+
             self._session._update_agent_state(
                 "speaking",
                 start_time=started_speaking_at,

livekit-agents/livekit/agents/voice/agent_session.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import copy
+import math
 import time
 from collections.abc import AsyncIterable, AsyncIterator, Callable, Sequence
 from contextlib import AbstractContextManager, asynccontextmanager, nullcontext
@@ -23,7 +24,7 @@
 from google.protobuf.json_format import ParseDict
 from google.protobuf.struct_pb2 import Struct
 from opentelemetry import context as otel_context, trace
-from typing_extensions import TypedDict
+from typing_extensions import Required, TypedDict
 
 from livekit import rtc
 from livekit.protocol.agent_pb import agent_session as agent_pb
@@ -58,6 +59,7 @@
     CloseReason,
     ConversationItemAddedEvent,
     EventTypes,
+    LatencyBudgetEvent,
     ToolCallEnded,
     ToolExecutionUpdatedEvent,
     UserInputTranscribedEvent,
@@ -144,6 +146,15 @@ class RecordingOptions(TypedDict, total=False):
 }
 
 
+class LatencyBudgetOptions(TypedDict, total=False):
+    """Thresholds for end-of-user-speech to first-agent-output latency."""
+
+    budget: Required[float]
+    """Maximum acceptable latency in seconds. Required when configured."""
+    warning: float
+    """Optional warning threshold in seconds. Must not exceed ``budget``."""
+
+
 def _resolve_recording_options(record: bool | RecordingOptions) -> RecordingOptions:
     if isinstance(record, bool):
         defaults = _RECORDING_ALL_ON if record else _RECORDING_ALL_OFF
@@ -297,6 +308,7 @@ class AgentSessionOptions:
     aec_warmup_duration: float | None
     session_close_transcript_timeout: float
     recording_options: RecordingOptions
+    latency_budget: LatencyBudgetOptions | None
 
     @property
     def endpointing(self) -> EndpointingOptions:
@@ -395,6 +407,7 @@ def __init__(
         user_away_timeout: float | None = 15.0,
         transcription_timeout: float | None = None,
         session_close_transcript_timeout: float = 2.0,
+        latency_budget: LatencyBudgetOptions | None = None,
         # Runtime settings
         conn_options: NotGivenOr[SessionConnectOptions] = NOT_GIVEN,
         loop: asyncio.AbstractEventLoop | None = None,
@@ -500,6 +513,9 @@ def __init__(
             session_close_transcript_timeout (float, optional): Seconds to wait for the
                 final STT transcript when closing the session (after audio is detached).
                 Default ``2.0`` s (independent of ``commit_user_turn``'s ``transcript_timeout``).
+            latency_budget (LatencyBudgetOptions, optional): Emits a ``latency_budget`` event
+                when end-of-user-speech to first-agent-output latency reaches the optional
+                warning threshold or exceeds the required budget. Disabled by default.
             preemptive_generation (NotGivenOr[bool | PreemptiveGenerationOptions]): Deprecated, use turn_handling=TurnHandlingOptions(...) instead.
             min_endpointing_delay (NotGivenOr[float]): Deprecated, use turn_handling=TurnHandlingOptions(...) instead.
             max_endpointing_delay (NotGivenOr[float]): Deprecated, use turn_handling=TurnHandlingOptions(...) instead.
@@ -587,6 +603,7 @@ def __init__(
             aec_warmup_duration=resolved_aec_warmup_duration,
             session_close_transcript_timeout=session_close_transcript_timeout,
             recording_options=_RECORDING_ALL_OFF.copy(),
+            latency_budget=self._resolve_latency_budget(latency_budget),
         )
         self._expressive: bool | ExpressiveOptions = expressive
         self._conn_options = conn_options or SessionConnectOptions()
@@ -734,6 +751,50 @@ def emit(self, event: EventTypes, arg: AgentEvent) -> None:
         self._recorded_events.append(arg)
         super().emit(event, arg)
 
+    @staticmethod
+    def _resolve_latency_budget(
+        options: LatencyBudgetOptions | None,
+    ) -> LatencyBudgetOptions | None:
+        if options is None:
+            return None
+        budget = options.get("budget")
+        warning = options.get("warning")
+        if budget is None or not math.isfinite(budget) or budget <= 0:
+            raise ValueError("latency_budget['budget'] must be finite and greater than zero")
+        if warning is not None and (not math.isfinite(warning) or warning <= 0 or warning > budget):
+            raise ValueError(
+                "latency_budget['warning'] must be finite, greater than zero, and no greater "
+                "than budget"
+            )
+        return LatencyBudgetOptions(**options)
+
+    def _evaluate_latency_budget(self, *, latency: float, speech_id: str) -> None:
+        options = self._opts.latency_budget
+        if options is None:
+            return
+
+        budget = options["budget"]
+        warning = options.get("warning")
+        if latency >= budget:
+            level: Literal["warning", "exceeded"] = "exceeded"
+            threshold = budget
+        elif warning is not None and latency >= warning:
+            level = "warning"
+            threshold = warning
+        else:
+            return
+
+        self.emit(
+            "latency_budget",
+            LatencyBudgetEvent(
+                level=level,
+                latency=latency,
+                threshold=threshold,
+                budget=budget,
+                speech_id=speech_id,
+            ),
+        )
+
     @property
     def userdata(self) -> Userdata_T:
         if self._userdata is None:

livekit-agents/livekit/agents/voice/events.py

@@ -300,6 +300,7 @@ def _make_update_pair(
     "metrics_collected",
     "session_usage_updated",
     "speech_created",
+    "latency_budget",
     "tool_execution_updated",
     "error",
     "close",
@@ -475,6 +476,21 @@ class SpeechCreatedEvent(BaseModel):
     created_at: float = Field(default_factory=time.time)
 
 
+class LatencyBudgetEvent(BaseModel):
+    """Emitted when an agent turn reaches a configured latency threshold."""
+
+    type: Literal["latency_budget"] = "latency_budget"
+    level: Literal["warning", "exceeded"]
+    latency: float
+    """End of user speech to first agent output, in seconds."""
+    threshold: float
+    """The threshold that selected ``level``."""
+    budget: float
+    """The configured maximum latency, in seconds."""
+    speech_id: str
+    created_at: float = Field(default_factory=time.time)
+
+
 class ToolCallStarted(BaseModel):
     """A function tool call was dispatched."""
 
@@ -594,6 +610,7 @@ class CloseEvent(BaseModel):
     | ConversationItemAddedEvent
     | FunctionToolsExecutedEvent
     | SpeechCreatedEvent
+    | LatencyBudgetEvent
     | ToolExecutionUpdatedEvent
     | ErrorEvent
     | CloseEvent

tests/test_agent_session.py

@@ -5,6 +5,7 @@
 import time
 from collections.abc import AsyncIterable
 from types import SimpleNamespace
+from typing import Literal
 from unittest.mock import MagicMock, Mock, patch
 
 import pytest
@@ -20,6 +21,8 @@
     ConversationItemAddedEvent,
     FlushSentinel,
     LanguageCode,
+    LatencyBudgetEvent,
+    LatencyBudgetOptions,
     MetricsCollectedEvent,
     ModelSettings,
     NotGivenOr,
@@ -104,6 +107,46 @@ async def on_user_turn_completed(self, turn_ctx: ChatContext, new_message: ChatM
 SESSION_TIMEOUT = 60.0
 
 
+def test_latency_budget_options_validation() -> None:
+    with pytest.raises(ValueError, match="budget.*greater than zero"):
+        AgentSession(latency_budget={"budget": 0})
+
+    with pytest.raises(ValueError, match="warning.*no greater than budget"):
+        AgentSession(latency_budget={"budget": 1.0, "warning": 1.1})
+
+
+@pytest.mark.parametrize(
+    ("latency_budget", "expected_level", "expected_threshold"),
+    [
+        ({"budget": 0.4, "warning": 0.2}, "exceeded", 0.4),
+        ({"budget": 0.9, "warning": 0.4}, "warning", 0.4),
+    ],
+)
+async def test_latency_budget_event_on_first_output(
+    latency_budget: LatencyBudgetOptions,
+    expected_level: Literal["warning", "exceeded"],
+    expected_threshold: float,
+) -> None:
+    actions = FakeActions()
+    actions.add_user_speech(0.5, 2.5, "Hello", stt_delay=0.2)
+    actions.add_llm("Hi!", ttft=0.1, duration=0.3)
+    actions.add_tts(0.5, ttfb=0.2, duration=0.3)
+
+    session = create_session(actions, speed_factor=1)
+    session._opts.latency_budget = session._resolve_latency_budget(latency_budget)
+    events: list[LatencyBudgetEvent] = []
+    session.on("latency_budget", events.append)
+
+    await asyncio.wait_for(run_session(session, MyAgent()), timeout=SESSION_TIMEOUT)
+
+    assert len(events) == 1
+    assert events[0].level == expected_level
+    assert events[0].threshold == expected_threshold
+    assert events[0].budget == latency_budget["budget"]
+    assert events[0].latency == pytest.approx(0.7, abs=0.01)
+    assert events[0].speech_id
+
+
 def test_realtime_user_input_transcription_preserves_item_id() -> None:
     captured_events: list[UserInputTranscribedEvent] = []