automatically collect session usage

exposing this as AgentSession.usage, ensuring collection is performed
for each model & provider pair.

this also updates the UsageSummary fields to be more consistent
- prompt_token -> input_token
- completion_token -> output_token

Author: davidzhao

examples/bank-ivr/ivr_navigator_agent.py

@@ -109,16 +109,12 @@ async def dtmf_session(ctx: JobContext) -> None:
     )
     logger.info(f"==> User request: {user_request}")
 
-    usage_collector = metrics.UsageCollector()
-
     @session.on("metrics_collected")
     def _on_metrics_collected(ev: MetricsCollectedEvent) -> None:
         metrics.log_metrics(ev.metrics)
-        usage_collector.collect(ev.metrics)
 
     async def log_usage() -> None:
-        summary = usage_collector.get_summary()
-        logger.info(f"Usage: {summary}")
+        logger.info(f"Usage: {session.usage}")
 
     ctx.add_shutdown_callback(log_usage)
 

examples/bank-ivr/ivr_system_agent.py

@@ -645,16 +645,12 @@ async def bank_ivr_session(ctx: JobContext) -> None:
         userdata=state,
     )
 
-    usage_collector = metrics.UsageCollector()
-
     @session.on("metrics_collected")
     def _on_metrics(ev: MetricsCollectedEvent) -> None:
         metrics.log_metrics(ev.metrics)
-        usage_collector.collect(ev.metrics)
 
     async def log_usage() -> None:
-        summary = usage_collector.get_summary()
-        logger.info("Usage summary: %s", summary)
+        logger.info("Usage summary: %s", session.usage)
 
     ctx.add_shutdown_callback(log_usage)
 

examples/dtmf/basic_dtmf_agent.py

@@ -142,16 +142,12 @@ async def entrypoint(ctx: JobContext) -> None:
         turn_detection=MultilingualModel(),
     )
 
-    usage_collector = metrics.UsageCollector()
-
     @session.on("metrics_collected")
     def _on_metrics_collected(ev: MetricsCollectedEvent) -> None:
         metrics.log_metrics(ev.metrics)
-        usage_collector.collect(ev.metrics)
 
     async def log_usage() -> None:
-        summary = usage_collector.get_summary()
-        logger.info(f"Usage: {summary}")
+        logger.info(f"Usage: {session.usage}")
 
     ctx.add_shutdown_callback(log_usage)
 

examples/survey/survey_agent.py

@@ -18,7 +18,6 @@
     cli,
     inference,
     llm,
-    metrics,
     room_io,
 )
 from livekit.agents.beta.workflows import GetEmailTask, TaskGroup
@@ -347,11 +346,8 @@ async def entrypoint(ctx: JobContext):
         preemptive_generation=True,
     )
 
-    usage_collector = metrics.UsageCollector()
-
     async def log_usage():
-        summary = usage_collector.get_summary()
-        logger.info(f"Usage: {summary}")
+        logger.info(f"Usage: {session.usage}")
 
     ctx.add_shutdown_callback(log_usage)
 

examples/voice_agents/basic_agent.py

@@ -15,6 +15,7 @@
     metrics,
     room_io,
 )
+from livekit.agents.beta.tools import EndCallTool
 from livekit.agents.llm import function_tool
 from livekit.plugins import silero
 from livekit.plugins.turn_detector.multilingual import MultilingualModel
@@ -35,6 +36,7 @@ def __init__(self) -> None:
             "do not use emojis, asterisks, markdown, or other special characters in your responses."
             "You are curious and friendly, and have a sense of humor."
             "you will speak english to the user",
+            tools=[EndCallTool()],
         )
 
     async def on_enter(self) -> None:
@@ -106,16 +108,12 @@ async def entrypoint(ctx: JobContext) -> None:
     )
 
     # log metrics as they are emitted, and total usage after session is over
-    usage_collector = metrics.UsageCollector()
-
     @session.on("metrics_collected")
     def _on_metrics_collected(ev: MetricsCollectedEvent) -> None:
         metrics.log_metrics(ev.metrics)
-        usage_collector.collect(ev.metrics)
 
-    async def log_usage() -> None:
-        summary = usage_collector.get_summary()
-        logger.info(f"Usage: {summary}")
+    async def log_usage():
+        logger.info(f"Usage: {session.usage}")
 
     # shutdown callbacks are triggered when the session is over
     ctx.add_shutdown_callback(log_usage)

examples/voice_agents/multi_agent.py

@@ -152,16 +152,12 @@ async def entrypoint(ctx: JobContext):
     )
 
     # log metrics as they are emitted, and total usage after session is over
-    usage_collector = metrics.UsageCollector()
-
     @session.on("metrics_collected")
     def _on_metrics_collected(ev: MetricsCollectedEvent):
         metrics.log_metrics(ev.metrics)
-        usage_collector.collect(ev.metrics)
 
     async def log_usage():
-        summary = usage_collector.get_summary()
-        logger.info(f"Usage: {summary}")
+        logger.info(f"Usage: {session.usage}")
 
     ctx.add_shutdown_callback(log_usage)
 

examples/voice_agents/zapier_mcp_integration.py

@@ -61,12 +61,9 @@ async def entrypoint(ctx: JobContext):
     participant = await ctx.wait_for_participant()
     logger.info(f"starting voice assistant for participant {participant.identity}")
 
-    usage_collector = metrics.UsageCollector()
-
-    # Log metrics and collect usage data
+    # Log metrics as they are collected
     def on_metrics_collected(agent_metrics: metrics.AgentMetrics):
         metrics.log_metrics(agent_metrics)
-        usage_collector.collect(agent_metrics)
 
     # Get MCP server URL from environment variable
     zapier_mcp_server = os.getenv("ZAPIER_MCP_SERVER")

livekit-agents/livekit/agents/job.py

@@ -274,6 +274,7 @@ def make_session_report(self, session: AgentSession | None = None) -> SessionRep
             started_at=session._started_at,
             events=session._recorded_events,
             chat_history=session.history.copy(),
+            usage=session.usage,
         )
 
         if recorder_io:

livekit-agents/livekit/agents/metrics/usage_collector.py

@@ -1,3 +1,4 @@
+import dataclasses
 from copy import deepcopy
 from dataclasses import dataclass
 
@@ -6,87 +7,139 @@
 
 @dataclass
 class UsageSummary:
-    llm_prompt_tokens: int = 0
-    llm_prompt_cached_tokens: int = 0
+    """Usage summary for a specific model/provider combination."""
+
+    provider: str = ""
+    """The provider name (e.g., 'openai', 'deepgram', 'elevenlabs')."""
+    model: str = ""
+    """The model name (e.g., 'gpt-4o', 'nova-2', 'eleven_turbo_v2')."""
+
+    llm_input_tokens: int = 0
+    llm_input_cached_tokens: int = 0
     llm_input_audio_tokens: int = 0
     llm_input_cached_audio_tokens: int = 0
     llm_input_text_tokens: int = 0
     llm_input_cached_text_tokens: int = 0
     llm_input_image_tokens: int = 0
     llm_input_cached_image_tokens: int = 0
-    llm_completion_tokens: int = 0
+    llm_output_tokens: int = 0
     llm_output_audio_tokens: int = 0
     llm_output_image_tokens: int = 0
     llm_output_text_tokens: int = 0
     tts_characters_count: int = 0
     tts_audio_duration: float = 0.0
     stt_audio_duration: float = 0.0
 
-    # properties for naming consistency: prompt = input, completion = output
+    # backwards-compatible property aliases
+    @property
+    def llm_prompt_tokens(self) -> int:
+        return self.llm_input_tokens
+
+    @llm_prompt_tokens.setter
+    def llm_prompt_tokens(self, value: int) -> None:
+        self.llm_input_tokens = value
+
     @property
-    def llm_input_tokens(self) -> int:
-        return self.llm_prompt_tokens
+    def llm_prompt_cached_tokens(self) -> int:
+        return self.llm_input_cached_tokens
 
-    @llm_input_tokens.setter
-    def llm_input_tokens(self, value: int) -> None:
-        self.llm_prompt_tokens = value
+    @llm_prompt_cached_tokens.setter
+    def llm_prompt_cached_tokens(self, value: int) -> None:
+        self.llm_input_cached_tokens = value
 
     @property
-    def llm_output_tokens(self) -> int:
-        return self.llm_completion_tokens
+    def llm_completion_tokens(self) -> int:
+        return self.llm_output_tokens
 
-    @llm_output_tokens.setter
-    def llm_output_tokens(self, value: int) -> None:
-        self.llm_completion_tokens = value
+    @llm_completion_tokens.setter
+    def llm_completion_tokens(self, value: int) -> None:
+        self.llm_output_tokens = value
+
+    def to_dict(self) -> dict:
+        """Returns a dict with only non-zero/non-empty values."""
+        return {k: v for k, v in dataclasses.asdict(self).items() if v}
+
+    def __repr__(self) -> str:
+        items = ", ".join(f"{k}={v!r}" for k, v in self.to_dict().items())
+        return f"UsageSummary({items})"
 
 
 class UsageCollector:
+    """Collects and aggregates usage metrics per model/provider combination."""
+
     def __init__(self) -> None:
-        self._summary = UsageSummary()
+        self._summaries: dict[tuple[str, str], UsageSummary] = {}
 
     def __call__(self, metrics: AgentMetrics) -> None:
         self.collect(metrics)
 
+    def _get_summary(self, provider: str, model: str) -> UsageSummary:
+        """Get or create a UsageSummary for the given provider/model combination."""
+        key = (provider, model)
+        if key not in self._summaries:
+            self._summaries[key] = UsageSummary(provider=provider, model=model)
+        return self._summaries[key]
+
+    def _extract_provider_model(
+        self, metrics: LLMMetrics | STTMetrics | TTSMetrics | RealtimeModelMetrics
+    ) -> tuple[str, str]:
+        """Extract provider and model from metrics metadata."""
+        provider = ""
+        model = ""
+        if metrics.metadata:
+            provider = metrics.metadata.model_provider or ""
+            model = metrics.metadata.model_name or ""
+        return provider, model
+
     def collect(self, metrics: AgentMetrics) -> None:
         if isinstance(metrics, LLMMetrics):
-            self._summary.llm_prompt_tokens += metrics.prompt_tokens
-            self._summary.llm_prompt_cached_tokens += metrics.prompt_cached_tokens
-            self._summary.llm_completion_tokens += metrics.completion_tokens
+            provider, model = self._extract_provider_model(metrics)
+            summary = self._get_summary(provider, model)
+            summary.llm_input_tokens += metrics.prompt_tokens
+            summary.llm_input_cached_tokens += metrics.prompt_cached_tokens
+            summary.llm_output_tokens += metrics.completion_tokens
 
         elif isinstance(metrics, RealtimeModelMetrics):
-            self._summary.llm_prompt_tokens += metrics.input_tokens
-            self._summary.llm_prompt_cached_tokens += metrics.input_token_details.cached_tokens
+            provider, model = self._extract_provider_model(metrics)
+            summary = self._get_summary(provider, model)
+            summary.llm_input_tokens += metrics.input_tokens
+            summary.llm_input_cached_tokens += metrics.input_token_details.cached_tokens
 
-            self._summary.llm_input_text_tokens += metrics.input_token_details.text_tokens
-            self._summary.llm_input_cached_text_tokens += (
+            summary.llm_input_text_tokens += metrics.input_token_details.text_tokens
+            summary.llm_input_cached_text_tokens += (
                 metrics.input_token_details.cached_tokens_details.text_tokens
                 if metrics.input_token_details.cached_tokens_details
                 else 0
             )
-            self._summary.llm_input_image_tokens += metrics.input_token_details.image_tokens
-            self._summary.llm_input_cached_image_tokens += (
+            summary.llm_input_image_tokens += metrics.input_token_details.image_tokens
+            summary.llm_input_cached_image_tokens += (
                 metrics.input_token_details.cached_tokens_details.image_tokens
                 if metrics.input_token_details.cached_tokens_details
                 else 0
             )
-            self._summary.llm_input_audio_tokens += metrics.input_token_details.audio_tokens
-            self._summary.llm_input_cached_audio_tokens += (
+            summary.llm_input_audio_tokens += metrics.input_token_details.audio_tokens
+            summary.llm_input_cached_audio_tokens += (
                 metrics.input_token_details.cached_tokens_details.audio_tokens
                 if metrics.input_token_details.cached_tokens_details
                 else 0
             )
 
-            self._summary.llm_output_text_tokens += metrics.output_token_details.text_tokens
-            self._summary.llm_output_image_tokens += metrics.output_token_details.image_tokens
-            self._summary.llm_output_audio_tokens += metrics.output_token_details.audio_tokens
-            self._summary.llm_completion_tokens += metrics.output_tokens
+            summary.llm_output_text_tokens += metrics.output_token_details.text_tokens
+            summary.llm_output_image_tokens += metrics.output_token_details.image_tokens
+            summary.llm_output_audio_tokens += metrics.output_token_details.audio_tokens
+            summary.llm_output_tokens += metrics.output_tokens
 
         elif isinstance(metrics, TTSMetrics):
-            self._summary.tts_characters_count += metrics.characters_count
-            self._summary.tts_audio_duration += metrics.audio_duration
+            provider, model = self._extract_provider_model(metrics)
+            summary = self._get_summary(provider, model)
+            summary.tts_characters_count += metrics.characters_count
+            summary.tts_audio_duration += metrics.audio_duration
 
         elif isinstance(metrics, STTMetrics):
-            self._summary.stt_audio_duration += metrics.audio_duration
+            provider, model = self._extract_provider_model(metrics)
+            summary = self._get_summary(provider, model)
+            summary.stt_audio_duration += metrics.audio_duration
 
-    def get_summary(self) -> UsageSummary:
-        return deepcopy(self._summary)
+    def get_summary(self) -> list[UsageSummary]:
+        """Returns a list of usage summaries, one per model/provider combination."""
+        return [deepcopy(s) for s in self._summaries.values()]

livekit-agents/livekit/agents/telemetry/trace_types.py

@@ -57,10 +57,16 @@
 ATTR_TTS_METRICS = "lk.tts_metrics"
 ATTR_REALTIME_MODEL_METRICS = "lk.realtime_model_metrics"
 
+# latency span attributes
+ATTR_LLM_NODE_TTFT = "lk.ttft"
+ATTR_TTS_NODE_TTFB = "lk.ttfb"
+ATTR_E2E_LATENCY = "lk.e2e_latency"
+
 # OpenTelemetry GenAI attributes
 # OpenTelemetry specification: https://opentelemetry.io/docs/specs/semconv/registry/attributes/gen-ai/
 ATTR_GEN_AI_OPERATION_NAME = "gen_ai.operation.name"
 ATTR_GEN_AI_REQUEST_MODEL = "gen_ai.request.model"
+ATTR_GEN_AI_PROVIDER_NAME = "gen_ai.provider.name"
 ATTR_GEN_AI_USAGE_INPUT_TOKENS = "gen_ai.usage.input_tokens"
 ATTR_GEN_AI_USAGE_OUTPUT_TOKENS = "gen_ai.usage.output_tokens"