fix(inference): keep a stream retryable until it emits generation

livekit-agents/livekit/agents/inference/llm.py

@@ -465,11 +465,11 @@ async def _run(self) -> None:
                     for choice in chunk.choices:
                         chat_chunk = self._parse_choice(chunk.id, choice, thinking_filter)
                         if chat_chunk is not None:
-                            retryable = False
+                            if chat_chunk.carries_generation():
+                                retryable = False
                             self._event_ch.send_nowait(chat_chunk)
 
                     if chunk.usage is not None:
-                        retryable = False
                         tokens_details = chunk.usage.prompt_tokens_details
                         cached_tokens = tokens_details.cached_tokens if tokens_details else 0
                         usage_chunk = llm.ChatChunk(
@@ -486,6 +486,10 @@ async def _run(self) -> None:
 
         except openai.APITimeoutError:
             raise APITimeoutError(retryable=retryable) from None
+        except httpx.TimeoutException as e:
+            # Only the request call runs inside the openai client's error mapping, so a
+            # timeout waiting on the stream body arrives as the raw httpx exception.
+            raise APITimeoutError(retryable=retryable) from e
         except openai.APIStatusError as e:
             raise APIStatusError(
                 e.message,

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

@@ -81,6 +81,15 @@ class ChatChunk(BaseModel):
     delta: ChoiceDelta | None = None
     usage: CompletionUsage | None = None
 
+    def carries_generation(self) -> bool:
+        """Whether this chunk delivered generation the caller can see.
+
+        Token counts and provider metadata (a gateway deployment stamp, a thought
+        signature) reach the caller without being output: they neither start the
+        clock on time-to-first-token nor give a retry anything to duplicate.
+        """
+        return bool(self.delta and (self.delta.content or self.delta.tool_calls))
+
 
 class LLMError(BaseModel):
     model_config = ConfigDict(arbitrary_types_allowed=True)
@@ -328,12 +337,16 @@ async def _metrics_monitor_task(self, event_aiter: AsyncIterable[ChatChunk]) ->
         response_content = ""
         tool_calls: list[FunctionToolCall] = []
         completion_start_time: str | None = None
+        received_chunk = False
 
         async for ev in event_aiter:
+            received_chunk = True
             request_id = ev.id
             if request_id and request_id not in self._provider_request_ids:
                 self._provider_request_ids.append(request_id)
-            if ttft == -1.0:
+            # measured against generation, not the first chunk: a retry that follows a
+            # contentless chunk would otherwise latch the clock on the failed attempt
+            if ttft == -1.0 and ev.carries_generation():
                 ttft = time.perf_counter() - start_time
                 completion_start_time = datetime.now(timezone.utc).isoformat()
 
@@ -348,8 +361,9 @@ async def _metrics_monitor_task(self, event_aiter: AsyncIterable[ChatChunk]) ->
 
         duration = time.perf_counter() - start_time
 
-        # if generation is aborted before any tokens are received, it doesn't make sense to report -1 ttft
-        if self._current_attempt_has_error or ttft < 0:
+        # a request that never yielded a chunk has nothing to report; one that yielded
+        # only metadata still carries token counts, and reports ttft as -1
+        if self._current_attempt_has_error or not received_chunk:
             return
 
         metrics = LLMMetrics(

livekit-agents/livekit/agents/metrics/base.py

@@ -24,6 +24,7 @@ class LLMMetrics(_BaseMetrics):
     timestamp: float
     duration: float
     ttft: float
+    """Time to first generated token in seconds. -1 if the response generated none."""
     cancelled: bool
     completion_tokens: int
     prompt_tokens: int

livekit-agents/livekit/agents/voice/generation.py

@@ -231,19 +231,20 @@ async def _llm_inference_task(
     # forward llm stream to output channels
     try:
         async for chunk in llm_node:
-            if data.ttft is None:
-                data.ttft = time.perf_counter() - start_time
-
             # extract text content from either str or ChatChunk
             content: str | None = None
+            generated = False
 
             if isinstance(chunk, str):
                 content = chunk
+                generated = bool(chunk)
 
             elif isinstance(chunk, ChatChunk):
                 if not chunk.delta:
                     continue
 
+                generated = chunk.carries_generation()
+
                 if chunk.delta.tool_calls:
                     for tool in chunk.delta.tool_calls:
                         if tool.type != "function":
@@ -280,6 +281,9 @@ async def _llm_inference_task(
                 )
                 content = None
 
+            if generated and data.ttft is None:
+                data.ttft = time.perf_counter() - start_time
+
             # route text content to output channels
             if content:
                 data.generated_text += content

livekit-plugins/livekit-plugins-openai/livekit/plugins/openai/responses/llm.py

@@ -484,8 +484,9 @@ async def _run_impl(self) -> None:
                 }
                 async for raw_event in self._llm._ws.generate_response(payload):
                     parsed_ev = self._parse_ws_event(raw_event)
-                    self._process_event(parsed_ev)
-                    retryable = False
+                    chunk = self._process_event(parsed_ev)
+                    if chunk is not None and chunk.carries_generation():
+                        retryable = False
 
                 if not self._response_completed:
                     raise APIConnectionError(retryable=True)
@@ -512,8 +513,9 @@ async def _run_impl(self) -> None:
 
                 async with stream:
                     async for event in stream:
-                        self._process_event(event)
-                        retryable = False
+                        chunk = self._process_event(event)
+                        if chunk is not None and chunk.carries_generation():
+                            retryable = False
 
             except openai.APITimeoutError:
                 raise APITimeoutError(retryable=retryable)  # noqa: B904
@@ -567,9 +569,10 @@ def _parse_ws_event(self, event: dict) -> ResponseStreamEvent | None:
             return ResponseFailedEvent.model_validate(event)
         return None
 
-    def _process_event(self, event: ResponseStreamEvent | None) -> None:
+    def _process_event(self, event: ResponseStreamEvent | None) -> llm.ChatChunk | None:
+        """Handle one stream event, returning the chunk it sent to the caller, if any."""
         if event is None:
-            return
+            return None
         chunk = None
         if isinstance(event, ResponseErrorEvent):
             self._handle_error(event)
@@ -585,6 +588,7 @@ def _process_event(self, event: ResponseStreamEvent | None) -> None:
             self._handle_response_failed(event)
         if chunk is not None:
             self._event_ch.send_nowait(chunk)
+        return chunk
 
     def _handle_error(self, event: ResponseErrorEvent) -> None:
         error_code = -1

tests/test_inference_llm_retry.py

@@ -0,0 +1,236 @@
+"""Retry eligibility, and reported latency, of a failed inference LLM stream.
+
+A stream that dies having emitted nothing the caller can see must be retried:
+provider metadata (a gateway deployment stamp, a thought signature) and token
+counts are not generation. The same line divides the latency the caller actually
+waited from the moment a contentless chunk happened to arrive.
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import AsyncIterator, Callable
+
+import httpx
+import openai
+import pytest
+
+from livekit.agents import APIConnectOptions, APITimeoutError, llm
+from livekit.agents.inference import LLM
+from livekit.agents.metrics import LLMMetrics
+
+pytestmark = pytest.mark.unit
+
+
+def _sse(payload: dict) -> bytes:
+    return f"data: {json.dumps(payload)}\n\n".encode()
+
+
+def _chunk(delta: dict) -> bytes:
+    return _sse(
+        {
+            "id": "chatcmpl-1",
+            "object": "chat.completion.chunk",
+            "created": 0,
+            "model": "google/gemma-4-31b-it",
+            "choices": [{"index": 0, "delta": delta, "finish_reason": None}],
+        }
+    )
+
+
+# The gateway stamps its deployment and billing tier onto the leading delta, which
+# carries no content of its own.
+_METADATA_ONLY = _chunk(
+    {
+        "role": "assistant",
+        "extra_content": {
+            "livekit": {"inference_deployment": "d", "inference_tier_billed": "standard"}
+        },
+    }
+)
+_TEXT = _chunk({"role": "assistant", "content": "hello"})
+_TOOL_NAME = _chunk(
+    {
+        "role": "assistant",
+        "tool_calls": [
+            {"index": 0, "id": "call_1", "type": "function", "function": {"name": "lookup"}}
+        ],
+    }
+)
+_TOOL_ARGS = _chunk(
+    {"role": "assistant", "tool_calls": [{"index": 0, "function": {"arguments": '{"q":"x"}'}}]}
+)
+_TOOL_DONE = _sse(
+    {
+        "id": "chatcmpl-1",
+        "object": "chat.completion.chunk",
+        "created": 0,
+        "model": "google/gemma-4-31b-it",
+        "choices": [{"index": 0, "delta": {}, "finish_reason": "tool_calls"}],
+    }
+)
+_USAGE = _sse(
+    {
+        "id": "chatcmpl-1",
+        "object": "chat.completion.chunk",
+        "created": 0,
+        "model": "google/gemma-4-31b-it",
+        "choices": [],
+        "usage": {"prompt_tokens": 11, "completion_tokens": 7, "total_tokens": 18},
+    }
+)
+
+
+class _StallingStream(httpx.AsyncByteStream):
+    """Yields the given SSE bytes, then stalls out like a provider going quiet."""
+
+    def __init__(self, chunks: list[bytes]) -> None:
+        self._chunks = chunks
+
+    async def __aiter__(self) -> AsyncIterator[bytes]:
+        for chunk in self._chunks:
+            yield chunk
+        raise httpx.ReadTimeout("stalled mid-stream")
+
+
+class _CompletedStream(httpx.AsyncByteStream):
+    """Yields the given SSE bytes, then ends the stream cleanly."""
+
+    def __init__(self, chunks: list[bytes]) -> None:
+        self._chunks = chunks
+
+    async def __aiter__(self) -> AsyncIterator[bytes]:
+        for chunk in self._chunks:
+            yield chunk
+        yield b"data: [DONE]\n\n"
+
+
+def _llm_for(
+    responder: Callable[[int], httpx.AsyncByteStream],
+) -> tuple[LLM, list[httpx.Request], list[LLMMetrics]]:
+    attempts: list[httpx.Request] = []
+    metrics: list[LLMMetrics] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        attempts.append(request)
+        return httpx.Response(
+            200,
+            headers={"content-type": "text/event-stream"},
+            stream=responder(len(attempts)),
+        )
+
+    # Long enough to keep PyJWT's short-key warning out of the suite output.
+    fake_secret = "f" * 32
+    llm_model = LLM(model="google/gemma-4-31b-it", api_key=fake_secret, api_secret=fake_secret)
+    llm_model._client = openai.AsyncClient(
+        api_key=fake_secret,
+        base_url="http://inference.test/v1",
+        max_retries=0,
+        http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)),
+    )
+    llm_model.on("metrics_collected", metrics.append)
+    return llm_model, attempts, metrics
+
+
+def _drain(llm_model: LLM, *, max_retry: int):
+    chat_ctx = llm.ChatContext.empty()
+    chat_ctx.add_message(role="user", content="hi")
+    return llm_model.chat(
+        chat_ctx=chat_ctx,
+        conn_options=APIConnectOptions(max_retry=max_retry, retry_interval=0.0, timeout=5.0),
+    )
+
+
+async def _run(chunks: list[bytes], *, max_retry: int) -> tuple[Exception, list[httpx.Request]]:
+    llm_model, attempts, _ = _llm_for(lambda _: _StallingStream(chunks))
+
+    with pytest.raises(Exception) as exc_info:  # noqa: PT011
+        async with _drain(llm_model, max_retry=max_retry) as stream:
+            async for _ in stream:
+                pass
+
+    return exc_info.value, attempts
+
+
+async def _run_to_completion(
+    responder: Callable[[int], httpx.AsyncByteStream], *, max_retry: int
+) -> tuple[list[LLMMetrics], list[httpx.Request]]:
+    llm_model, attempts, metrics = _llm_for(responder)
+
+    # aclose() awaits the metrics monitor, so metrics are settled once the block exits
+    async with _drain(llm_model, max_retry=max_retry) as stream:
+        async for _ in stream:
+            pass
+
+    return metrics, attempts
+
+
+@pytest.mark.asyncio
+async def test_metadata_only_chunk_stays_retryable() -> None:
+    error, attempts = await _run([_METADATA_ONLY], max_retry=2)
+
+    assert len(attempts) == 3, "a stall after metadata alone must exhaust the retries"
+    assert "after 3 attempts" in str(error)
+
+
+@pytest.mark.asyncio
+async def test_generated_text_is_not_retried() -> None:
+    error, attempts = await _run([_METADATA_ONLY, _TEXT], max_retry=2)
+
+    assert len(attempts) == 1, "text already sent to the caller must not be regenerated"
+    assert "after" not in str(error)
+
+
+@pytest.mark.asyncio
+async def test_completed_tool_call_is_not_retried() -> None:
+    error, attempts = await _run([_METADATA_ONLY, _TOOL_NAME, _TOOL_ARGS, _TOOL_DONE], max_retry=2)
+
+    assert len(attempts) == 1, "a tool call already sent to the caller must not be re-issued"
+    assert "after" not in str(error)
+
+
+@pytest.mark.asyncio
+async def test_partial_tool_arguments_stay_retryable() -> None:
+    _, attempts = await _run([_METADATA_ONLY, _TOOL_NAME, _TOOL_ARGS], max_retry=2)
+
+    assert len(attempts) == 3, "arguments still streaming have reached nobody"
+
+
+@pytest.mark.asyncio
+async def test_stream_read_timeout_is_a_timeout_error() -> None:
+    error, _ = await _run([_TEXT], max_retry=0)
+
+    assert isinstance(error, APITimeoutError), "a stalled stream body is a timeout, not a connect"
+    assert "timed out" in str(error).lower()
+
+
+@pytest.mark.asyncio
+async def test_ttft_spans_the_retry() -> None:
+    """The clock runs until generation, not until the failed attempt's metadata."""
+
+    def responder(attempt: int) -> httpx.AsyncByteStream:
+        if attempt == 1:
+            return _StallingStream([_METADATA_ONLY])
+        return _CompletedStream([_METADATA_ONLY, _TEXT, _USAGE])
+
+    metrics, attempts = await _run_to_completion(responder, max_retry=2)
+
+    assert len(attempts) == 2
+    assert len(metrics) == 1
+    # the first retry always waits APIConnectOptions._interval_for_retry(0) == 0.1s
+    assert metrics[0].ttft > 0.09, "ttft latched on the failed attempt's metadata chunk"
+    assert metrics[0].ttft <= metrics[0].duration
+
+
+@pytest.mark.asyncio
+async def test_token_counts_survive_a_response_that_generates_nothing() -> None:
+    """Metadata and usage but no output: ttft is unmeasurable, the tokens still are."""
+    metrics, attempts = await _run_to_completion(
+        lambda _: _CompletedStream([_METADATA_ONLY, _USAGE]), max_retry=2
+    )
+
+    assert len(attempts) == 1
+    assert len(metrics) == 1
+    assert metrics[0].ttft == -1
+    assert metrics[0].completion_tokens == 7
+    assert metrics[0].total_tokens == 18