To access RealTimeModel usage metrics, you can use the following approaches: 1. Event-Based Metric Collection The RealTimeModel automatically emits

RealtimeModelMetrics

when responses are completed. You can listen for these events:

from livekit.agents import session

@session.on("metrics_collected")
def _on_metrics_collected(ev: MetricsCollectedEvent):
    if isinstance(ev.metrics, RealtimeModelMetrics):
        # Access detailed usage metrics
        print(f"Input tokens: {ev.metrics.input_tokens}")
        print(f"Output tokens: {ev.metrics.output_tokens}")
        print(f"Cached tokens: {ev.metrics.input_token_details.cached_tokens}")

2. Using UsageCollector for Aggregation The

UsageCollector

class provides a convenient way to aggregate usage metrics across an entire session:

from livekit.agents import metrics

usage_collector = metrics.UsageCollector()

@session.on("metrics_collected")
def _on_metrics_collected(ev: MetricsCollectedEvent):
    usage_collector.collect(ev.metrics)

# Get aggregated summary
summary = usage_collector.get_summary()
print(f"Total LLM prompt tokens: {summary.llm_prompt_tokens}")
print(f"Total LLM completion tokens: {summary.llm_completion_tokens}")

3. Available Metrics The

RealtimeModelMetrics

includes comprehensive usage information: • Token counts:

input_tokens

,

output_tokens

,

total_tokens

• Detailed token breakdown: ◦

input_token_details.audio_tokens

◦

input_token_details.text_tokens

◦

input_token_details.cached_tokens

◦

output_token_details.audio_tokens

◦

output_token_details.text_tokens

• Performance metrics:

ttft

(time to first token),

duration

,

tokens_per_second

• Request metadata:

request_id

,

timestamp

,

cancelled

status 4. Complete Example Here’s a complete example showing metric collection in a voice agent:

usage_collector = metrics.UsageCollector()

@session.on("metrics_collected")
def _on_metrics_collected(ev: MetricsCollectedEvent):
    metrics.log_metrics(ev.metrics)  # Log individual metrics
    usage_collector.collect(ev.metrics)  # Aggregate for summary

async def log_usage():
    summary = usage_collector.get_summary()
    <http://logger.info|logger.info>(f"Usage: {summary}")

# Log final usage summary at shutdown
ctx.add_shutdown_callback(log_usage)

These metrics are automatically generated by all RealTimeModel implementations (OpenAI, Google, AWS) and provide detailed insights into token usage, performance, and costs for monitoring and optimization purposes. Sources: examples/voice_agents/multi_agent.py