fix(inference): flush text preamble to TTS when a tool call starts

examples/voice_agents/flush_llm_node.py

@@ -88,7 +88,11 @@ async def llm_node(
 
             yield chunk
 
-        # example: fast response conditioned on the tool call name and the presence of a text message
+        # When the model emits its own text preamble before a tool call, the pipeline
+        # already flushes it to TTS automatically at the tool boundary (via the
+        # `tool_call_started` marker), so no manual flush is needed for that case.
+        #
+        # example: scripted fast response for when the model calls the tool with no preamble
         tool_names = [tool.name for tool in called_tools]
         if not has_text_message and "get_weather" in tool_names:
             logger.info("Fast response triggered")

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

@@ -388,6 +388,7 @@ async def _run(self) -> None:
         self._fnc_raw_arguments: str | None = None
         self._tool_extra: dict[str, Any] | None = None
         self._tool_index: int | None = None
+        self._tool_start_signaled = False
         retryable = True
 
         try:
@@ -507,6 +508,19 @@ def _parse_choice(
                     self._fnc_raw_arguments = tool.function.arguments or ""
                     # Extract extra from tool call (e.g., Google thought signatures)
                     self._tool_extra = getattr(tool, "extra_content", None)
+
+                    # Signal the start of the tool call so a buffered text preamble can be
+                    # flushed to TTS now, rather than after the arguments finish streaming
+                    # (which can add ~1s of dead air). The marker carries no tool_calls, so
+                    # it never triggers tool execution; the assembled call is still emitted
+                    # once the arguments are complete. Only the first tool call of a turn
+                    # can have a preamble to flush.
+                    if not self._tool_start_signaled and call_chunk is None:
+                        self._tool_start_signaled = True
+                        return llm.ChatChunk(
+                            id=id,
+                            delta=llm.ChoiceDelta(role="assistant", tool_call_started=True),
+                        )
                 elif tool.function.arguments:
                     self._fnc_raw_arguments += tool.function.arguments  # type: ignore
 

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

@@ -74,6 +74,12 @@ class ChoiceDelta(BaseModel):
     tool_calls: list[FunctionToolCall] = Field(default_factory=list)
     extra: dict[str, Any] | None = None
     """Provider-specific extra data (e.g., Google thought signatures)."""
+    tool_call_started: bool = False
+    """Marks the delta where a tool call begins, before its arguments have streamed in.
+
+    Emitted as an empty marker chunk (no ``content``, no ``tool_calls``) so downstream
+    consumers can flush any buffered text preamble to TTS immediately, instead of waiting
+    for the tool arguments to finish serializing."""
 
 
 class ChatChunk(BaseModel):

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

@@ -221,6 +221,11 @@ async def _llm_inference_task(
                 if not chunk.delta:
                     continue
 
+                # A tool call is starting: flush any buffered text preamble to TTS now,
+                # so it plays while the tool arguments are still streaming in.
+                if chunk.delta.tool_call_started:
+                    text_ch.send_nowait(FlushSentinel())
+
                 if chunk.delta.tool_calls:
                     for tool in chunk.delta.tool_calls:
                         if tool.type != "function":

tests/test_inference_llm_tool_flush.py

@@ -0,0 +1,100 @@
+"""Unit tests for the tool-call preamble flush signal in ``inference.llm.LLMStream``.
+
+Regression coverage for #5826: when a model streams a text preamble followed by a
+tool call in the same turn, ``_parse_choice`` used to emit nothing while the tool
+arguments serialized (~1s), leaving the preamble buffered in TTS and producing
+audible dead air. It now emits a ``tool_call_started`` marker chunk at the tool
+boundary so the preamble can be flushed immediately.
+
+These tests drive ``_parse_choice`` directly with fake OpenAI-shaped deltas, so no
+network or API key is required.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from types import SimpleNamespace
+from typing import Any
+
+import pytest
+
+from livekit.agents.inference.llm import LLMStream
+
+pytestmark = pytest.mark.unit
+
+
+def _stream() -> LLMStream:
+    # Build without __init__ — these tests only exercise _parse_choice and its state.
+    stream = LLMStream.__new__(LLMStream)
+    stream._tool_call_id = None
+    stream._fnc_name = None
+    stream._fnc_raw_arguments = None
+    stream._tool_extra = None
+    stream._tool_index = None
+    stream._tool_start_signaled = False
+    return stream
+
+
+def _tool(*, name: str | None = None, arguments: str | None = None) -> Any:
+    return SimpleNamespace(
+        function=SimpleNamespace(name=name, arguments=arguments),
+        id="call_1" if name else None,
+        index=0,
+        type="function",
+        extra_content=None,
+    )
+
+
+def _choice(*, content: str | None = None, tools: list[Any] | None = None, finish=None) -> Any:
+    return SimpleNamespace(
+        delta=SimpleNamespace(content=content, tool_calls=tools, extra_content=None),
+        finish_reason=finish,
+    )
+
+
+def test_marker_emitted_at_tool_boundary() -> None:
+    stream = _stream()
+    thinking = asyncio.Event()
+
+    preamble = stream._parse_choice("c", _choice(content="Let me check"), thinking)
+    assert preamble is not None and preamble.delta is not None
+    assert preamble.delta.content == "Let me check"
+    assert preamble.delta.tool_call_started is False
+
+    marker = stream._parse_choice(
+        "c", _choice(tools=[_tool(name="get_balance", arguments="")]), thinking
+    )
+    assert marker is not None and marker.delta is not None
+    assert marker.delta.tool_call_started is True
+    # The marker must not carry an executable tool call, or the pipeline would run the
+    # tool with incomplete arguments.
+    assert not marker.delta.tool_calls
+    assert marker.delta.content is None
+
+
+def test_arguments_stream_without_extra_markers() -> None:
+    stream = _stream()
+    thinking = asyncio.Event()
+
+    stream._parse_choice("c", _choice(tools=[_tool(name="get_balance", arguments="")]), thinking)
+    assert stream._parse_choice("c", _choice(tools=[_tool(arguments='{"acc')]), thinking) is None
+    assert stream._parse_choice("c", _choice(tools=[_tool(arguments='ount":1}')]), thinking) is None
+
+    final = stream._parse_choice("c", _choice(finish="tool_calls"), thinking)
+    assert final is not None and final.delta is not None
+    assert final.delta.tool_call_started is False
+    assert len(final.delta.tool_calls) == 1
+    call = final.delta.tool_calls[0]
+    assert call.name == "get_balance"
+    assert call.arguments == '{"account":1}'
+
+
+def test_marker_emitted_once_per_turn() -> None:
+    stream = _stream()
+    thinking = asyncio.Event()
+
+    first = stream._parse_choice("c", _choice(tools=[_tool(name="a", arguments="")]), thinking)
+    assert first is not None and first.delta is not None and first.delta.tool_call_started is True
+
+    # Argument fragments for the same call never re-signal.
+    assert stream._parse_choice("c", _choice(tools=[_tool(arguments="{}")]), thinking) is None