I have a python agent. How can I capture metrics f...
# ask-ai
c
I have a python agent. How can I capture metrics for latency for tts, llm and stt?
t
Thanks for reaching out, Rashmi. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
To capture metrics for latency for TTS, LLM, and STT in your Python agent, you can use the LiveKit Agents metrics system by subscribing to metrics collection events. Below are the steps to implement this: Basic Setup
Copy code
from 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:
Copy code
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
Copy code
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:
Copy code
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
Copy code
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