chilly-ghost-74326
07/29/2025, 3:33 AMtall-airline-16756
07/29/2025, 3:33 AMtall-airline-16756
07/29/2025, 3:34 AMfrom livekit.agents import metrics, MetricsCollectedEvent
@session.on("metrics_collected")
def _on_metrics_collected(ev: MetricsCollectedEvent):
metrics.log_metrics(ev.metrics)
Capturing Individual Component Metrics
For more granular control, set up separate event handlers for each component:
from livekit.agents.metrics import LLMMetrics, STTMetrics, TTSMetrics
class MetricsAgent(Agent):
def __init__(self):
llm = openai.LLM(model="gpt-4o-mini")
stt = deepgram.STT()
tts = openai.TTS()
super().__init__(
instructions="Your agent instructions",
stt=stt,
llm=llm,
tts=tts,
)
def llm_metrics_wrapper(metrics: LLMMetrics):
asyncio.create_task(self.on_llm_metrics_collected(metrics))
llm.on("metrics_collected", llm_metrics_wrapper)
def stt_metrics_wrapper(metrics: STTMetrics):
asyncio.create_task(self.on_stt_metrics_collected(metrics))
stt.on("metrics_collected", stt_metrics_wrapper)
def tts_metrics_wrapper(metrics: TTSMetrics):
asyncio.create_task(self.on_tts_metrics_collected(metrics))
tts.on("metrics_collected", tts_metrics_wrapper)
Key Latency Metrics Available
• LLM Metrics: Time to first token (TTFT), tokens per second, total completion time.
• TTS Metrics: Time to first byte (TTFB), audio duration, total processing time.
• STT Metrics: Total processing time, input audio duration, transcription delay.
Implementing Metric Handlers
async def on_llm_metrics_collected(self, metrics: LLMMetrics):
print(f"LLM TTFT: {metrics.ttft:.4f}s")
print(f"LLM Duration: {metrics.duration:.4f}s")
print(f"Tokens per second: {metrics.tokens_per_second:.2f}")
async def on_tts_metrics_collected(self, metrics: TTSMetrics):
print(f"TTS TTFB: {metrics.ttfb:.4f}s")
print(f"TTS Duration: {metrics.duration:.4f}s")
print(f"Audio Duration: {metrics.audio_duration:.4f}s")
async def on_stt_metrics_collected(self, metrics: STTMetrics):
print(f"STT Duration: {metrics.duration:.4f}s")
print(f"Audio Duration: {metrics.audio_duration:.4f}s")
Aggregating Metrics
Use the UsageCollector to aggregate metrics over a session:
usage_collector = metrics.UsageCollector()
@session.on("metrics_collected")
def _on_metrics_collected(ev: MetricsCollectedEvent):
usage_collector.collect(ev.metrics)
# At session end
summary = usage_collector.get_summary()
print(f"Total TTS audio duration: {summary.tts_audio_duration}s")
print(f"Total STT audio duration: {summary.stt_audio_duration}s")
Measuring Total Conversation Latency
total_latency = eou.end_of_utterance_delay + llm.ttft + tts.ttfb
This approach provides comprehensive visibility into the latency performance of each component in your voice agent pipeline, allowing you to identify bottlenecks and optimize performance.
Sources: Capturing metrics | LiveKit Docs | metrics/metrics_stt.py