feat(metrics): broadcast llm tps + ttfs + agent_false_interruption

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

@@ -255,6 +255,18 @@ class MetricsReport(TypedDict, total=False):
     Assistant `ChatMessage` only
     """
 
+    llm_node_tps: float
+    """LLM output tokens per second for this turn, measured at the `llm_node`
+
+    Assistant `ChatMessage` only
+    """
+
+    llm_node_ttfs: float
+    """Time from LLM generation start to the first complete sentence
+
+    Assistant `ChatMessage` only
+    """
+
     tts_node_ttfb: float
     """Time taken for the `tts_node` to return the first chunk of audio (after the first text token has been sent)
 

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

@@ -3150,6 +3150,12 @@ async def _next_segment() -> _SpeechSegment | None:
         if llm_gen_data.ttft is not None:
             assistant_metrics["llm_node_ttft"] = llm_gen_data.ttft
 
+        if llm_gen_data.tps is not None:
+            assistant_metrics["llm_node_tps"] = llm_gen_data.tps
+
+        if llm_gen_data.ttfs is not None:
+            assistant_metrics["llm_node_ttfs"] = llm_gen_data.ttfs
+
         if first_tts_gen_data and first_tts_gen_data.ttfb is not None:
             assistant_metrics["tts_node_ttfb"] = first_tts_gen_data.ttfb
 

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

@@ -24,6 +24,7 @@
 from ..llm.chat_context import Instructions
 from ..log import logger
 from ..telemetry import trace_types, tracer
+from ..tokenize import blingfire
 from ..types import (
     USERDATA_TIMED_TRANSCRIPT,
     USERDATA_TTS_STARTED_TIME,
@@ -58,11 +59,20 @@ class _LLMGenerationData:
     id: str = field(default_factory=lambda: utils.shortuuid("item_"))
     started_fut: asyncio.Future[None] = field(default_factory=asyncio.Future)
     ttft: float | None = None
+    tps: float | None = None
+    ttfs: float | None = None
 
 
 # output for an injected in-progress tool call, phrased so the model waits instead of
 # re-issuing the call.
 _RUNNING_TOOL_PLACEHOLDER = "The tool call is still in progress."
+
+
+# Stateless and cheap to construct, min_sentence_len=1
+# keeps short first sentences (e.g. "Hi there.") counted when timing ttfs.
+_SENTENCE_TOKENIZER = blingfire.SentenceTokenizer(min_sentence_len=1)
+
+
 # extra flag marking an injected pair so it can be stripped before the ctx is forwarded.
 _RUNNING_PLACEHOLDER_KEY = "__lk_running_placeholder__"
 
@@ -206,6 +216,20 @@ async def _llm_inference_task(
         return False
 
     # forward llm stream to output channels
+    usage: Any = None
+    # Time the first complete sentence (ttfs) with a streaming tokenizer that consumes
+    # text incrementally without re-tokenizing the whole generation each chunk.
+    sentence_stream = _SENTENCE_TOKENIZER.stream()
+
+    async def _set_time_first_sentence() -> None:
+        try:
+            async for _ in sentence_stream:
+                data.ttfs = time.perf_counter() - start_time
+                return
+        except Exception:
+            logger.exception("failed to time first sentence (llm_node_ttfs)")
+
+    ttfs_task = asyncio.create_task(_set_time_first_sentence())
     try:
         async for chunk in llm_node:
             if data.ttft is None:
@@ -218,6 +242,8 @@ async def _llm_inference_task(
                 content = chunk
 
             elif isinstance(chunk, ChatChunk):
+                if chunk.usage is not None:
+                    usage = chunk.usage
                 if not chunk.delta:
                     continue
 
@@ -261,9 +287,22 @@ async def _llm_inference_task(
             if content:
                 data.generated_text += content
                 text_ch.send_nowait(content)
+                if data.ttfs is None:
+                    sentence_stream.push_text(content)
     finally:
         if isinstance(llm_node, _ACloseable):
             await llm_node.aclose()
+        try:
+            sentence_stream.end_input()  # flush any trailing sentence, then close
+        except RuntimeError:
+            pass
+        await utils.aio.cancel_and_wait(ttfs_task)
+
+    duration = time.perf_counter() - start_time
+    if usage is not None and duration > 0:
+        data.tps = usage.completion_tokens / duration
+    if data.ttfs is None and data.generated_text.strip():
+        data.ttfs = duration
 
     current_span.set_attribute(trace_types.ATTR_RESPONSE_TEXT, data.generated_text)
     current_span.set_attribute(

livekit-agents/livekit/agents/voice/remote_session.py

@@ -37,6 +37,7 @@
 from ..version import __version__
 from ..voice.amd import AMDCategory, AMDPredictionEvent
 from .events import (
+    AgentFalseInterruptionEvent,
     AgentState,
     AgentStateChangedEvent,
     ConversationItemAddedEvent,
@@ -255,6 +256,8 @@ async def __anext__(self) -> agent_pb.AgentSessionMessage:
     "llm_node_ttft",
     "tts_node_ttfb",
     "e2e_latency",
+    "llm_node_tps",
+    "llm_node_ttfs",
 )
 
 _TOOL_CALL_STATUS_MAP: dict[str, agent_pb.ToolCallStatus] = {
@@ -394,6 +397,7 @@ def register_session(self, session: AgentSession) -> None:
             session.on("tool_execution_updated", self._on_tool_execution_updated)
             session.on("session_usage_updated", self._on_session_usage_updated)
             session.on("overlapping_speech", self._on_overlapping_speech)
+            session.on("agent_false_interruption", self._on_agent_false_interruption)
             session.on("error", self._on_error)
             session.on("debug_message", self._on_debug_message)
 
@@ -419,6 +423,7 @@ async def aclose(self) -> None:
             self._session.off("tool_execution_updated", self._on_tool_execution_updated)
             self._session.off("session_usage_updated", self._on_session_usage_updated)
             self._session.off("overlapping_speech", self._on_overlapping_speech)
+            self._session.off("agent_false_interruption", self._on_agent_false_interruption)
             self._session.off("error", self._on_error)
             self._session.off("debug_message", self._on_debug_message)
 
@@ -603,6 +608,10 @@ def _on_overlapping_speech(self, event: OverlappingSpeechEvent) -> None:
 
         self._send_event(agent_pb.AgentSessionEvent(overlapping_speech=pb))
 
+    def _on_agent_false_interruption(self, event: AgentFalseInterruptionEvent) -> None:
+        pb = agent_pb.AgentSessionEvent.AgentFalseInterruption(resumed=event.resumed)
+        self._send_event(agent_pb.AgentSessionEvent(agent_false_interruption=pb))
+
     def _on_amd_prediction(self, event: AMDPredictionEvent) -> None:
         speech_duration = Duration()
         speech_duration.FromNanoseconds(int(event.speech_duration * 1e9))

tests/test_llm_node_metrics.py

@@ -0,0 +1,89 @@
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from livekit.agents.llm import ChatChunk, ChatContext, ChoiceDelta, CompletionUsage
+from livekit.agents.llm.tool_context import ToolContext
+from livekit.agents.tokenize import blingfire
+from livekit.agents.utils import aio
+from livekit.agents.voice.agent import ModelSettings
+from livekit.agents.voice.generation import (
+    _SENTENCE_TOKENIZER,
+    _llm_inference_task,
+    _LLMGenerationData,
+)
+
+pytestmark = pytest.mark.unit
+
+
+def _fake_node(chunks: list[ChatChunk]):
+    # matches the io.LLMNode signature: (chat_ctx, tools, model_settings) -> AsyncIterable
+    async def node(chat_ctx, tools, model_settings):  # type: ignore[no-untyped-def]
+        for chunk in chunks:
+            await asyncio.sleep(0)  # yield so the ttfs consumer task can run
+            yield chunk
+
+    return node
+
+
+async def _run_inference(chunks: list[ChatChunk]) -> _LLMGenerationData:
+    data = _LLMGenerationData(text_ch=aio.Chan(), function_ch=aio.Chan())
+    await _llm_inference_task(
+        _fake_node(chunks),
+        ChatContext.empty(),
+        ToolContext.empty(),
+        ModelSettings(),
+        data,
+    )
+    return data
+
+
+def _content(text: str) -> ChatChunk:
+    return ChatChunk(id="c", delta=ChoiceDelta(content=text))
+
+
+def _usage_chunk(completion_tokens: int) -> ChatChunk:
+    return ChatChunk(
+        id="c",
+        usage=CompletionUsage(
+            completion_tokens=completion_tokens,
+            prompt_tokens=5,
+            total_tokens=completion_tokens + 5,
+        ),
+    )
+
+
+class TestLLMNodeTps:
+    async def test_tps_set_when_usage_reported(self) -> None:
+        data = await _run_inference([_content("Hello there, friend."), _usage_chunk(30)])
+        assert data.tps is not None
+        assert data.tps > 0
+
+    async def test_tps_zero_when_zero_usage_is_reported(self) -> None:
+        data = await _run_inference([_content("Hello there, friend."), _usage_chunk(0)])
+        assert data.tps == 0
+
+    async def test_tps_none_when_no_usage_reported(self) -> None:
+        data = await _run_inference([_content("Hello there, friend.")])  # no usage chunk
+        assert data.tps is None
+
+
+class TestLLMNodeTtfs:
+    async def test_ttfs_set_for_nonempty_generation(self) -> None:
+        data = await _run_inference([_content("First sentence. "), _content("Second one.")])
+        assert data.ttfs is not None
+        assert data.ttfs > 0
+
+    async def test_ttfs_none_for_whitespace_only(self) -> None:
+        data = await _run_inference([_content("   ")])
+        assert data.ttfs is None
+
+
+class TestSentenceTokenizerConfig:
+    def test_short_first_sentence_is_counted(self) -> None:
+        # regression: ttfs must see short openers. blingfire's TTS-oriented default
+        # (min_sentence_len=20) drops them, which silently inflated ttfs to the whole turn.
+        text = "Hi there. How can I help?"
+        assert len(_SENTENCE_TOKENIZER.tokenize(text)) == 2

tests/test_session_host.py

@@ -158,6 +158,18 @@ def test_partial_fields(self) -> None:
         pb = _metrics_to_proto(metrics)
         assert pb.transcription_delay == pytest.approx(0.42)
 
+    @pytest.mark.skipif(
+        "llm_node_tps" not in agent_pb.MetricsReport.DESCRIPTOR.fields_by_name,
+        reason="livekit-protocol < 1.1.18 lacks llm_node_tps/llm_node_ttfs",
+    )
+    def test_llm_node_throughput_fields(self) -> None:
+        # guards the dict-key -> proto-field mapping: a mismatch (e.g. "tps" vs
+        # "llm_node_tps") raises at MetricsReport(**kwargs) instead of silently dropping.
+        metrics = {"llm_node_tps": 12.5, "llm_node_ttfs": 0.6}
+        pb = _metrics_to_proto(metrics)
+        assert pb.llm_node_tps == pytest.approx(12.5)
+        assert pb.llm_node_ttfs == pytest.approx(0.6)
+
 
 class TestSessionUsageToProto:
     def test_llm_usage(self) -> None:
@@ -278,7 +290,7 @@ def mock_session(self) -> MagicMock:
     def test_register_session(self, transport: InMemoryTransport, mock_session: MagicMock) -> None:
         host = SessionHost(transport)
         host.register_session(mock_session)
-        assert mock_session.on.call_count == 10
+        assert mock_session.on.call_count == 11
 
     @pytest.mark.asyncio
     async def test_agent_state_changed(self, transport: InMemoryTransport) -> None:
@@ -660,4 +672,4 @@ async def test_aclose_unregisters_events(self, transport: InMemoryTransport) ->
         host.register_session(session)
         await host.start()
         await host.aclose()
-        assert session.off.call_count == 10
+        assert session.off.call_count == 11