little-king-41579
03/20/2026, 10:50 AMtall-airline-16756
03/20/2026, 10:51 AMtall-airline-16756
03/20/2026, 10:51 AMduration, audio_duration, request_id (STT processing time and audio length).
• VAD: idle_time, inference_duration_total, inference_count (VAD inference times, idle detection).
• LLM: duration, ttft (time-to-first-token), prompt_tokens, tokens_per_second.
• TTS: duration, ttfb (time-to-first-byte), audio_duration, characters_count.
• EOU (end-of-utterance): end_of_utterance_delay, transcription_delay (useful to tie STT->LLM handoff).
How to collect metrics in Python (event-handler pattern) 1) Register listeners on each component's metrics_collected (and EOU for STT). Each handler stores the metric record with timestamps so you can compute deltas.
Example (conceptual pattern — adapt names to your SDK objects):
import time
from collections import defaultdict
collected = defaultdict(list)
# STT handler
def on_stt_metrics(metrics):
# metrics.duration, metrics.audio_duration, metrics.request_id, metrics.timestamp
collected['stt'].append({
'ts': time.time(),
'duration': metrics.duration,
'audio_duration': metrics.audio_duration,
'request_id': metrics.request_id,
})
# VAD handler
def on_vad_metrics(metrics):
# metrics.inference_duration_total, metrics.inference_count, metrics.idle_time
collected['vad'].append({
'ts': time.time(),
'inference_total': metrics.inference_duration_total,
'inference_count': metrics.inference_count,
'idle_time': metrics.idle_time,
})
# LLM handler
def on_llm_metrics(metrics):
# metrics.duration, metrics.ttft, metrics.prompt_tokens
collected['llm'].append({
'ts': time.time(),
'duration': metrics.duration,
'ttft': metrics.ttft,
'prompt_tokens': metrics.prompt_tokens,
})
# TTS handler
def on_tts_metrics(metrics):
# metrics.duration, metrics.ttfb, metrics.audio_duration
collected['tts'].append({
'ts': time.time(),
'duration': metrics.duration,
'ttfb': metrics.ttfb,
'audio_duration': metrics.audio_duration,
})
# EOU handler (optional, ties utterance boundaries)
def on_eou_metrics(metrics):
# metrics.end_of_utterance_delay, metrics.transcription_delay
collected['eou'].append({
'ts': time.time(),
'eou_delay': metrics.end_of_utterance_delay,
'transcription_delay': metrics.transcription_delay,
})
Where to hook these handlers
• Use the on("metrics_collected", handler) for each component (e.g., session.stt.on("metrics_collected", on_stt_metrics)).
• Use session.stt.on("eou_metrics_collected", on_eou_metrics) for EOU events.
Per-component latency calculations
• STT latency for an utterance:
◦ Use metrics.duration — represents STT processing time.
◦ If you want capture-to-transcription latency: end_of_utterance_delay + transcription_delay (from EOU metrics) or compute difference between STT start timestamp and STT completion timestamp if both are available.
• VAD latency:
◦ Average inference latency = inference_duration_total / inference_count
◦ Idle detection or buffering delay = idle_time (time spent waiting before detection).
• LLM latency:
◦ Time to first token (TTFT) = metrics.ttft.
◦ Total LLM processing = metrics.duration.
◦ If you want throughput: use metrics.tokens_per_second.
• TTS latency:
◦ Time to first audio data = metrics.ttfb.
◦ Total synthesis time = metrics.duration.
Example: compute simple averages and end-to-end per turn
def avg(values):
return sum(values)/len(values) if values else 0.0
def compute_stats(collected):
stats = {}
# STT
stt_durations = [m['duration'] for m in collected['stt']]
stats['stt_avg'] = avg(stt_durations)
# VAD
vad_avg_infer = avg([m['inference_total']/m['inference_count'] if m['inference_count'] else 0 for m in collected['vad']])
stats['vad_avg_inference'] = vad_avg_infer
# LLM
stats['llm_avg'] = avg([m['duration'] for m in collected['llm']])
stats['llm_avg_ttft'] = avg([m['ttft'] for m in collected['llm']])
# TTS
stats['tts_avg'] = avg([m['duration'] for m in collected['tts']])
stats['tts_avg_ttfb'] = avg([m.get('ttfb', 0) for m in collected['tts']])
return stats
Estimating end-to-end latency for a single turn
• A conservative end-to-end latency approximation (start of speech -> audio out) can be:
total_latency ≈ EOU.end_of_utterance_delay + STT.transcription_delay + LLM.ttft + TTS.ttfb
• If EOU is not available, approximate STT handoff with stt.duration, or use timestamps to compute:
total_latency ≈ (STT processing time) + (LLM time-to-first-token) + (TTS time-to-first-byte)
Notes and tips
• Use the request_id or timestamps to correlate specific STT -> LLM -> TTS events for the same user utterance.
• Prefer the EOU metrics for accurate utterance boundary timings when available.
• For stable numbers, collect many samples and report median and p95 (not just average) because latency distributions are often skewed.
• If you need live dashboards, export these metrics to your monitoring system (Prometheus/Grafana, or LiveKit Cloud Agent Observability dashboard if available).
If you want, I can:
• Provide a full runnable Python example that hooks into your exact session object names.
• Show how to correlate events using request_id or timestamps to produce per-utterance breakdowns.
• Add code to compute median and p95 latency and to export metrics to a CSV or monitoring endpoint.
Which of those would you like next, and can you confirm the exact Python SDK objects/names you’re using (e.g., session.stt, session.llm, etc.)?
Sources: STT Metrics | LiveKit Documentation | LLM Metrics | LiveKit Documentation | VAD Metrics | LiveKit Documentation | TTS Metrics | LiveKit Documentation | Build Your First AI Voice Agent in Python: Complete Tutorial | LiveKit, with interactive elements