feat(llm): allow a separate attempt timeout for fallback LLMs

livekit-agents/livekit/agents/llm/fallback_adapter.py

@@ -43,6 +43,7 @@ def __init__(
         llm: list[LLM],
         *,
         attempt_timeout: float = 5.0,
+        fallback_attempt_timeout: float | None = None,
         # use fallback instead of retrying
         max_retry_per_llm: int = 0,
         retry_interval: float = 0.5,
@@ -53,6 +54,10 @@ def __init__(
         Args:
             llm (list[LLM]): List of LLM instances to fallback to.
             attempt_timeout (float, optional): Timeout for each LLM attempt. Defaults to 5.0.
+            fallback_attempt_timeout (float, optional): Timeout for attempts on every LLM after
+                the first. Fallback providers typically run cold (e.g. without a warmed prompt
+                cache), so they may need a longer window than the primary's latency target.
+                Defaults to None, meaning ``attempt_timeout`` applies to all providers.
             max_retry_per_llm (int, optional): Internal retries per LLM. Defaults to 0, which means no
                 internal retries, the failed LLM will be skipped and the next LLM will be used.
             retry_interval (float, optional): Interval between retries. Defaults to 0.5.
@@ -69,6 +74,9 @@ def __init__(
 
         self._llm_instances = llm
         self._attempt_timeout = attempt_timeout
+        self._fallback_attempt_timeout = (
+            fallback_attempt_timeout if fallback_attempt_timeout is not None else attempt_timeout
+        )
         self._max_retry_per_llm = max_retry_per_llm
         self._retry_interval = retry_interval
         self._retry_on_chunk_sent = retry_on_chunk_sent
@@ -83,6 +91,13 @@ def __init__(
         for llm_instance in self._llm_instances:
             llm_instance.on("metrics_collected", self._on_metrics_collected)
 
+    def _attempt_timeout_for(self, llm: LLM) -> float:
+        """Attempt timeout for one provider: the primary keeps ``attempt_timeout``,
+        every other provider gets ``fallback_attempt_timeout``."""
+        if llm is self._llm_instances[0]:
+            return self._attempt_timeout
+        return self._fallback_attempt_timeout
+
     @property
     def model(self) -> str:
         return "FallbackAdapter"
@@ -193,7 +208,7 @@ async def _try_generate(
                 conn_options=dataclasses.replace(
                     self._conn_options,
                     max_retry=self._fallback_adapter._max_retry_per_llm,
-                    timeout=self._fallback_adapter._attempt_timeout,
+                    timeout=self._fallback_adapter._attempt_timeout_for(llm),
                     retry_interval=self._fallback_adapter._retry_interval,
                 ),
             ) as stream:

tests/test_llm_fallback.py

@@ -5,8 +5,15 @@
 
 import pytest
 
-from livekit.agents import APIConnectionError
-from livekit.agents.llm import ChatContext, FallbackAdapter, LLMStream, Tool
+from livekit.agents import APIConnectionError, APITimeoutError
+from livekit.agents.llm import (
+    ChatChunk,
+    ChatContext,
+    ChoiceDelta,
+    FallbackAdapter,
+    LLMStream,
+    Tool,
+)
 from livekit.agents.types import DEFAULT_API_CONNECT_OPTIONS, APIConnectOptions
 
 from .fake_llm import FakeLLM, FakeLLMResponse
@@ -141,3 +148,78 @@ async def test_prewarm_forwards_event_loop() -> None:
     finally:
         supplied_loop.close()
         await fallback_adapter.aclose()
+
+
+class _SlowFirstTokenStream(LLMStream):
+    """Enforces ``conn_options.timeout`` on the first token, like provider plugins do."""
+
+    async def _run(self) -> None:
+        assert isinstance(self._llm, _SlowFirstTokenLLM)
+        try:
+            await asyncio.wait_for(asyncio.sleep(self._llm.ttft), self._conn_options.timeout)
+        except asyncio.TimeoutError:
+            raise APITimeoutError(
+                f"{self._llm.model} exceeded the {self._conn_options.timeout}s attempt timeout"
+            ) from None
+        self._event_ch.send_nowait(
+            ChatChunk(id=str(id(self)), delta=ChoiceDelta(role="assistant", content="hello"))
+        )
+
+
+class _SlowFirstTokenLLM(_NamedLLM):
+    """FakeLLM with a fixed time-to-first-token, bounded by the attempt timeout."""
+
+    def __init__(self, *, model: str, ttft: float) -> None:
+        super().__init__(model=model, provider="fake")
+        self.ttft = ttft
+
+    def chat(
+        self,
+        *,
+        chat_ctx: ChatContext,
+        tools: list[Tool] | None = None,
+        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
+        **kwargs: Any,
+    ) -> LLMStream:
+        return _SlowFirstTokenStream(
+            self, chat_ctx=chat_ctx, tools=tools or [], conn_options=conn_options
+        )
+
+
+async def _collect_text(fallback_adapter: FallbackAdapter) -> str:
+    text = ""
+    async with fallback_adapter.chat(chat_ctx=ChatContext.empty()) as stream:
+        async for chunk in stream:
+            if chunk.delta and chunk.delta.content:
+                text += chunk.delta.content
+    return text
+
+
+async def test_fallback_attempt_timeout_gives_fallbacks_a_longer_window() -> None:
+    primary = _SlowFirstTokenLLM(model="primary", ttft=10.0)  # misses its timeout
+    fallback = _SlowFirstTokenLLM(model="fallback", ttft=0.3)  # needs more than the primary's
+
+    fallback_adapter = FallbackAdapter(
+        [primary, fallback], attempt_timeout=0.15, fallback_attempt_timeout=0.5
+    )
+    try:
+        assert await _collect_text(fallback_adapter) == "hello"
+        # the primary was cut at its own timeout, not given the fallback's window
+        assert [status.available for status in fallback_adapter._status] == [False, True]
+        # let the primary's background recovery attempt finish
+        await asyncio.sleep(0.3)
+    finally:
+        await fallback_adapter.aclose()
+
+
+async def test_attempt_timeout_applies_to_all_llms_by_default() -> None:
+    primary = _SlowFirstTokenLLM(model="primary", ttft=10.0)
+    fallback = _SlowFirstTokenLLM(model="fallback", ttft=0.3)
+
+    fallback_adapter = FallbackAdapter([primary, fallback], attempt_timeout=0.15)
+    try:
+        with pytest.raises(APIConnectionError):
+            await _collect_text(fallback_adapter)
+        await asyncio.sleep(0.3)
+    finally:
+        await fallback_adapter.aclose()