feat(realtime): add response-level retry and recovery for generate_reply

Realtime models had no response-level retry, unlike the pipeline LLM. A
recoverable generate_reply failure (timeout, transient server error, or a
reply discarded by an automatic session reconnection) left the turn silent
with no recovery, surfacing in production as dead air.

- classify errors: RealtimeError.recoverable + shared is_fatal_error()
  (quota/auth/billing are never retried)
- reconnect coordination: session_reconnecting event, reconnecting/
  wait_reconnected(); OpenAI plugin drives _set_reconnecting/_set_reconnected
  and logs reconnection at INFO
- fix infinite reconnect: reset num_retries only after a healthy connection
- cancel_and_wait(): cancel the active response and await its response.done
  before re-issuing, avoiding conversation_already_has_active_response
- retry loop in _realtime_reply_task for pre-response.created failures;
  handle update_chat_ctx push failures; wire GenerationCreatedEvent.done_fut
- tests for the classifier, reconnect state machine, error classification,
  and cancel_and_wait

Fixes #6205

Author: ByteMaster-1

livekit-agents/livekit/agents/llm/__init__.py

@@ -25,6 +25,7 @@
     LLMStream,
 )
 from .realtime import (
+    FATAL_REALTIME_ERROR_CODES,
     GenerationCreatedEvent,
     InputSpeechStartedEvent,
     InputSpeechStoppedEvent,
@@ -36,7 +37,9 @@
     RealtimeModelError,
     RealtimeSession,
     RealtimeSessionReconnectedEvent,
+    RealtimeSessionReconnectingEvent,
     RemoteItemAddedEvent,
+    is_fatal_error,
 )
 from .realtime_fallback_adapter import (
     RealtimeAvailabilityChangedEvent,
@@ -113,7 +116,9 @@
     "GenerationCreatedEvent",
     "MessageGeneration",
     "RealtimeSessionReconnectedEvent",
-    "RealtimeSessionRestoredEvent",
+    "RealtimeSessionReconnectingEvent",
+    "FATAL_REALTIME_ERROR_CODES",
+    "is_fatal_error",
     "LLMError",
     "RemoteItemAddedEvent",
 ]

livekit-agents/livekit/agents/llm/realtime.py

@@ -6,7 +6,7 @@
 from collections.abc import AsyncIterable, Awaitable
 from dataclasses import dataclass
 from types import TracebackType
-from typing import Generic, Literal, TypeVar
+from typing import Any, Generic, Literal, TypeVar
 
 from pydantic import BaseModel, ConfigDict, Field
 
@@ -45,6 +45,15 @@ class GenerationCreatedEvent:
     """True if the message was generated by the user using generate_reply()"""
     response_id: str | None = None
     """The response ID associated with this generation, used for metrics attribution"""
+    done_fut: asyncio.Future[None] | None = None
+    """Resolves when the response finishes.
+
+    Raises a :class:`RealtimeError` if the response failed *after* it started (e.g.
+    ``response.done`` with status ``failed``). This lets the caller learn about a
+    post-``response.created`` failure that the ``generate_reply()`` future — which
+    resolves at ``response.created`` — cannot surface. May be ``None`` for providers
+    that don't wire it.
+    """
 
 
 class RealtimeModelError(BaseModel):
@@ -87,8 +96,60 @@ class RealtimeCapabilities:
 
 
 class RealtimeError(Exception):
-    def __init__(self, message: str) -> None:
+    def __init__(
+        self, message: str, *, recoverable: bool = True, is_timeout: bool = False
+    ) -> None:
         super().__init__(message)
+        self.recoverable = recoverable
+        """Whether re-issuing the request could plausibly succeed.
+
+        ``True`` for transient failures (timeouts, server errors, connection drops);
+        ``False`` for terminal ones (retries exhausted, fatal account errors). Callers
+        should not retry when this is ``False``.
+        """
+        self.is_timeout = is_timeout
+        """``True`` if the request failed because it timed out waiting for the server.
+
+        Providers set this instead of relying on the message text, so callers can detect
+        a timeout without fragile string matching.
+        """
+
+
+# Underlying error codes/types that can never succeed on retry. Retrying these only
+# wastes the retry budget and keeps the reconnect loop spinning, so callers should
+# fail fast instead.
+FATAL_REALTIME_ERROR_CODES = frozenset(
+    {
+        "insufficient_quota",
+        "invalid_api_key",
+        "account_deactivated",
+        "billing_hard_limit_reached",
+    }
+)
+
+
+def is_fatal_error(error: object | None) -> bool:
+    """Return ``True`` for errors that can never succeed on retry.
+
+    Accepts any object (exceptions as well as provider error models) and walks the
+    wrapped-error chain (e.g. ``RealtimeModelError.error`` -> ``APIError.body`` -> ...)
+    matching any ``code``/``type`` attribute against :data:`FATAL_REALTIME_ERROR_CODES`
+    (quota / auth / billing).
+    """
+    seen: set[int] = set()
+    stack: list[Any] = [error]
+    while stack:
+        obj = stack.pop()
+        if obj is None or id(obj) in seen:
+            continue
+        seen.add(id(obj))
+        if getattr(obj, "code", None) in FATAL_REALTIME_ERROR_CODES:
+            return True
+        if getattr(obj, "type", None) in FATAL_REALTIME_ERROR_CODES:
+            return True
+        stack.append(getattr(obj, "error", None))
+        stack.append(getattr(obj, "body", None))
+    return False
 
 
 class RealtimeModel:
@@ -135,6 +196,7 @@ async def __aexit__(
     "input_speech_stopped",  # serverside VAD
     "input_audio_transcription_completed",
     "generation_created",
+    "session_reconnecting",
     "session_reconnected",
     "metrics_collected",
     "remote_item_added",
@@ -155,6 +217,11 @@ class InputTranscriptionCompleted:
     """confidence score of the transcript (0.0 to 1.0), derived from model logprobs"""
 
 
+@dataclass
+class RealtimeSessionReconnectingEvent:
+    pass
+
+
 @dataclass
 class RealtimeSessionReconnectedEvent:
     pass
@@ -170,6 +237,42 @@ class RealtimeSession(ABC, rtc.EventEmitter[EventTypes | TEvent], Generic[TEvent
     def __init__(self, realtime_model: RealtimeModel) -> None:
         super().__init__()
         self._realtime_model = realtime_model
+        self._reconnecting = False
+        self._reconnected_ev = asyncio.Event()
+        self._reconnected_ev.set()  # starts in the "connected" state
+
+    @property
+    def reconnecting(self) -> bool:
+        """True while the underlying session is re-establishing its connection."""
+        return self._reconnecting
+
+    async def wait_reconnected(self, timeout: float | None = None) -> bool:
+        """Wait until the session finishes reconnecting.
+
+        Returns immediately if the session is not currently reconnecting. Returns
+        ``True`` once reconnected, or ``False`` if ``timeout`` elapses first.
+        """
+        if not self._reconnecting:
+            return True
+        try:
+            await asyncio.wait_for(self._reconnected_ev.wait(), timeout)
+            return True
+        except asyncio.TimeoutError:
+            return False
+
+    def _set_reconnecting(self) -> None:
+        """Plugins call this when a reconnect attempt starts."""
+        if self._reconnecting:
+            return
+        self._reconnecting = True
+        self._reconnected_ev.clear()
+        self.emit("session_reconnecting", RealtimeSessionReconnectingEvent())
+
+    def _set_reconnected(self) -> None:
+        """Plugins call this when the session has successfully reconnected."""
+        self._reconnecting = False
+        self._reconnected_ev.set()
+        self.emit("session_reconnected", RealtimeSessionReconnectedEvent())
 
     def _report_connection_acquired(self, acquire_time: float) -> None:
         """Report connection timing as a RealtimeModelMetrics event with zero usage."""
@@ -253,6 +356,16 @@ def clear_audio(self) -> None: ...
     @abstractmethod
     def interrupt(self) -> None: ...
 
+    async def cancel_and_wait(self, timeout: float = 5.0) -> None:
+        """Cancel the current generation and wait for it to clear.
+
+        Re-issuing a reply before the previous response is fully cancelled server-side
+        can be rejected (e.g. ``conversation_already_has_active_response``). The default
+        implementation just calls :meth:`interrupt` without waiting; plugins that can
+        await the cancellation override this.
+        """
+        self.interrupt()
+
     # message_id is the ID of the message to truncate (inside the ChatCtx)
     @abstractmethod
     def truncate(

livekit-agents/livekit/agents/llm/realtime_fallback_adapter.py

@@ -21,7 +21,6 @@
     RealtimeModel,
     RealtimeModelError,
     RealtimeSession,
-    RealtimeSessionReconnectedEvent,
 )
 from .tool_context import Tool, ToolChoice, ToolContext
 
@@ -336,7 +335,7 @@ async def _bring_up(index: int) -> Exception | None:
                 return
 
             # a swap is a reconnect from the caller's perspective
-            self.emit("session_reconnected", RealtimeSessionReconnectedEvent())
+            self._set_reconnected()
 
             # re-issue the interrupted reply on the new session
             if (

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

@@ -103,6 +103,15 @@
 _SpeechHandleContextVar = contextvars.ContextVar["SpeechHandle"]("agents_speech_handle")
 _IdleHoldContextVar = contextvars.ContextVar[bool]("agents_idle_hold", default=False)
 
+# How many times a realtime generate_reply is re-issued when it fails *before the reply
+# starts* (timeout / server error / discarded on reconnection). Fatal errors (quota, auth)
+# are never retried.
+_REALTIME_REPLY_MAX_RETRIES = 3
+# Delay between retry attempts (a reconnect wait, when applicable, happens on top of this).
+_REALTIME_REPLY_RETRY_INTERVAL = 0.5
+# Upper bound on how long a retry waits for an in-progress reconnection before giving up.
+_REALTIME_RECONNECT_WAIT_TIMEOUT = 10.0
+
 
 class ActivityClosedError(Exception):
     """Raised by ``wait_for_idle`` when the target activity/session has closed."""
@@ -3259,7 +3268,19 @@ async def _realtime_reply_task(
         if user_input is not None:
             chat_ctx = self._rt_session.chat_ctx.copy()
             msg = chat_ctx.add_message(role="user", content=user_input)
-            await self._rt_session.update_chat_ctx(chat_ctx)
+            try:
+                await self._rt_session.update_chat_ctx(chat_ctx)
+            except llm.RealtimeError as e:
+                # A timeout means the item events are already on the wire; re-pushing would
+                # duplicate the user message, so proceed and let the reply run. Any other
+                # failure means the message never landed — surface it and don't reply.
+                if e.is_timeout:
+                    logger.warning("update_chat_ctx timed out; assuming the message was queued")
+                else:
+                    logger.error("failed to push user message before reply: %s", str(e))
+                    speech_handle._mark_done(error=e)
+                    self._session._update_agent_state("listening")
+                    return
             self._agent._chat_ctx._upsert_item(msg)
             self._session._conversation_item_added(msg)
 
@@ -3282,38 +3303,73 @@ async def _realtime_reply_task(
                     ori_tools = self._rt_session.tools.flatten()
                     await self._rt_session.update_tools(llm.ToolContext(tools).flatten())
 
-            generate_reply_fut = self._rt_session.generate_reply(
-                instructions=instructions or NOT_GIVEN,
-                tool_choice=(model_settings.tool_choice if per_response_tool_choice else NOT_GIVEN),
-                tools=(
-                    llm.ToolContext(tools).flatten()
-                    if per_response_tool_choice and tools is not None
-                    else NOT_GIVEN
-                ),
-            )
-            await speech_handle.wait_if_not_interrupted([generate_reply_fut])
-            if speech_handle.interrupted:
-                # cancel the pending generation; the plugin emits response.cancel
-                if not generate_reply_fut.done():
-                    generate_reply_fut.cancel()
-                return
-
-            try:
-                generation_ev = await generate_reply_fut
-            except llm.RealtimeError as e:
-                logger.error(
-                    "failed to generate a reply%s: %s",
-                    " after tool execution" if tool_reply else "",
-                    str(e),
+            # Retry the reply while it fails *before it starts* (pre-``response.created``):
+            # timeouts, server errors that produced no output, or a response discarded by a
+            # session reconnection. All of these raise from ``await generate_reply_fut``, so we
+            # re-issue here. Failures *after* the reply starts are handled by the generation
+            # task below and surfaced through ``SpeechHandle.exception()``.
+            reply_ev: llm.GenerationCreatedEvent | None = None
+            last_error: llm.RealtimeError | None = None
+            for attempt in range(_REALTIME_REPLY_MAX_RETRIES + 1):
+                if attempt > 0:
+                    # clear the previous (failed) response and wait for the socket to be
+                    # healthy again before re-issuing, so we neither collide with a still-active
+                    # response nor fire into a reconnecting session
+                    await self._rt_session.cancel_and_wait()
+                    if self._rt_session.reconnecting:
+                        await self._rt_session.wait_reconnected(_REALTIME_RECONNECT_WAIT_TIMEOUT)
+                    await asyncio.sleep(_REALTIME_REPLY_RETRY_INTERVAL)
+
+                generate_reply_fut = self._rt_session.generate_reply(
+                    instructions=instructions or NOT_GIVEN,
+                    tool_choice=(
+                        model_settings.tool_choice if per_response_tool_choice else NOT_GIVEN
+                    ),
+                    tools=(
+                        llm.ToolContext(tools).flatten()
+                        if per_response_tool_choice and tools is not None
+                        else NOT_GIVEN
+                    ),
                 )
-                speech_handle._mark_done(error=e)
+                await speech_handle.wait_if_not_interrupted([generate_reply_fut])
+                if speech_handle.interrupted:
+                    # cancel the pending generation; the plugin emits response.cancel
+                    if not generate_reply_fut.done():
+                        generate_reply_fut.cancel()
+                    return
+
+                try:
+                    reply_ev = await generate_reply_fut
+                    break
+                except llm.RealtimeError as e:
+                    last_error = e
+                    if not e.recoverable:
+                        # fatal / non-recoverable (quota, auth, retries exhausted upstream)
+                        logger.error(
+                            "failed to generate a reply%s (not recoverable, not retrying): %s",
+                            " after tool execution" if tool_reply else "",
+                            str(e),
+                        )
+                        break
+                    logger.warning(
+                        "failed to generate a reply%s (attempt %d/%d): %s",
+                        " after tool execution" if tool_reply else "",
+                        attempt + 1,
+                        _REALTIME_REPLY_MAX_RETRIES + 1,
+                        str(e),
+                    )
+
+            if reply_ev is None:
+                # every attempt failed (or a fatal error): record it so callers can read it
+                # via SpeechHandle.exception() (awaiting the handle never raises)
+                speech_handle._mark_done(error=last_error)
                 self._session._update_agent_state("listening")
                 return
 
             # _realtime_generation_task will clear the authorization
             await self._realtime_generation_task(
                 speech_handle=speech_handle,
-                generation_ev=generation_ev,
+                generation_ev=reply_ev,
                 model_settings=model_settings,
                 instructions=instructions,
             )

livekit-plugins/livekit-plugins-nvidia/livekit/plugins/nvidia/experimental/realtime/realtime_model.py

@@ -383,10 +383,7 @@ async def _main_task(self) -> None:
                 self._msg_ch = utils.aio.Chan[bytes]()
 
                 if restart_wait_task in done:
-                    self.emit(
-                        "session_reconnected",
-                        llm.RealtimeSessionReconnectedEvent(),
-                    )
+                    self._set_reconnected()
 
             except Exception as e:
                 logger.error(f"PersonaPlex WebSocket error: {e}", exc_info=True)