give metrics code for stt, vad, llm, tts

Thanks for reaching out, Laxman. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.

using python

Thanks — I’ll provide concise Python examples that attach to LiveKit Agents-style plugin events for STT, VAD, LLM, and TTS, extract the common fields mentioned (timings, durations, token/character counts, inference counts, EOU timing), log to console, and show an optional Prometheus export. Below are runnable-style snippets you can adapt into your agent. Each section shows: • Event handler that receives the metrics object • How to read common fields • A simple console log • An optional Prometheus metric definition and update call Note: Replace

self.session.stt

,

self.session.vad

,

self.session.llm

, and

self.session.tts

with the actual plugin instances in your agent. 1) Setup shared helpers and Prometheus exporter

from prometheus_client import start_http_server, Summary, Counter, Gauge
import time
import logging

logging.basicConfig(level=<http://logging.INFO|logging.INFO>)
logger = logging.getLogger("metrics")

# Start Prometheus HTTP server on port 8000 (optional)
start_http_server(8000)

# Define Prometheus metrics (example set)
stt_latency = Summary("stt_transcription_latency_seconds", "STT transcription latency (seconds)")
stt_audio_duration = Gauge("stt_audio_duration_seconds", "STT audio input duration (seconds)")
stt_eou_delay = Summary("stt_eou_delay_seconds", "End-of-utterance detection delay (seconds)")

vad_inference_time = Summary("vad_inference_time_seconds", "VAD inference total time (seconds)")
vad_idle_time = Summary("vad_idle_time_seconds", "VAD idle time seconds")
vad_inference_count = Counter("vad_inference_count", "VAD inference count")

llm_ttft = Summary("llm_time_to_first_token_seconds", "LLM time to first token (seconds)")
llm_prompt_tokens = Gauge("llm_prompt_tokens", "LLM prompt token count")
llm_completion_tokens = Gauge("llm_completion_tokens", "LLM completion token count")
llm_tps = Gauge("llm_tokens_per_second", "LLM tokens per second")

tts_ttfb = Summary("tts_time_to_first_byte_seconds", "TTS time to first byte (seconds)")
tts_audio_duration = Gauge("tts_audio_duration_seconds", "TTS audio duration seconds")
tts_characters = Gauge("tts_characters_count", "TTS characters count")

2) STT metrics handler (transcription duration, audio duration, EOU)

from livekit.agents.metrics import STTMetrics, EOUMetrics  # adjust import path as needed

async def on_stt_metrics_collected(metrics: STTMetrics):
    """
    Example fields expected on STTMetrics:
      - metrics.request_time or metrics.duration  (seconds)
      - metrics.audio_duration_seconds
      - metrics.transcription_start_time / end_time
    """
    # Read common fields (field names depend on your SDK version)
    latency = getattr(metrics, "duration", None) or getattr(metrics, "request_time", None)
    audio_dur = getattr(metrics, "audio_duration_seconds", None) or getattr(metrics, "audio_duration", None)

    <http://logger.info|logger.info>(f"STT metrics: latency={latency}s audio_duration={audio_dur}s")

    # Update Prometheus (guard None)
    if latency is not None:
        stt_latency.observe(latency)
    if audio_dur is not None:
        stt_audio_duration.set(audio_dur)

async def on_eou_metrics_collected(metrics: EOUMetrics):
    """
    Example fields on EOUMetrics:
      - metrics.eou_delay_seconds
      - metrics.transcription_delay_seconds
    """
    eou_delay = getattr(metrics, "eou_delay_seconds", None) or getattr(metrics, "eou_delay", None)
    transcription_delay = getattr(metrics, "transcription_delay_seconds", None) or getattr(metrics, "transcription_delay", None)

    <http://logger.info|logger.info>(f"EOU metrics: eou_delay={eou_delay}s transcription_delay={transcription_delay}s")
    if eou_delay is not None:
        stt_eou_delay.observe(eou_delay)

Hook STT handlers into your agent/session:

# inside your agent initialization or on_enter
self.session.stt.on("metrics_collected", on_stt_metrics_collected)
self.session.stt.on("eou_metrics_collected", on_eou_metrics_collected)

3) VAD metrics handler (inference timing, idle time)

from livekit.agents import vad  # adjust import as needed

async def on_vad_event(event: vad.VADEvent):
    """
    Expected fields on VADEvent or related metrics object:
      - event.idle_time
      - event.inference_duration_total
      - event.inference_count
    """
    idle_time = getattr(event, "idle_time", None)
    inf_total = getattr(event, "inference_duration_total", None)
    inf_count = getattr(event, "inference_count", None)

    <http://logger.info|logger.info>(f"VAD metrics: idle_time={idle_time}s inference_total={inf_total}s count={inf_count}")

    if idle_time is not None:
        vad_idle_time.observe(idle_time)
    if inf_total is not None:
        vad_inference_time.observe(inf_total)
    if inf_count is not None:
        vad_inference_count.inc(inf_count)

Hook into your VAD instance:

vad_instance.on("metrics_collected", on_vad_event)
# or if your VAD exposes a prewarmed instance:
# self.session.vad.on("metrics_collected", on_vad_event)

4) LLM metrics handler (TTFT, token counts, throughput)

from livekit.agents.metrics import LLMMetrics  # adjust import path

async def on_llm_metrics_collected(metrics: LLMMetrics):
    """
    Expected LLMMetrics fields:
      - metrics.ttft (time to first token, seconds)
      - metrics.prompt_tokens
      - metrics.completion_tokens
      - metrics.tokens_per_second
    """
    ttft = getattr(metrics, "ttft", None)
    prompt_tokens = getattr(metrics, "prompt_tokens", None)
    completion_tokens = getattr(metrics, "completion_tokens", None)
    tps = getattr(metrics, "tokens_per_second", None)

    <http://logger.info|logger.info>(f"LLM metrics: ttft={ttft}s prompt_tokens={prompt_tokens} completion_tokens={completion_tokens} tps={tps}")

    if ttft is not None:
        llm_ttft.observe(ttft)
    if prompt_tokens is not None:
        llm_prompt_tokens.set(prompt_tokens)
    if completion_tokens is not None:
        llm_completion_tokens.set(completion_tokens)
    if tps is not None:
        llm_tps.set(tps)

# Attach listener
self.session.llm.on("metrics_collected", on_llm_metrics_collected)

5) TTS metrics handler (TTFB, audio duration, characters)

from livekit.agents.metrics import TTSMetrics  # adjust import path

async def on_tts_metrics_collected(metrics: TTSMetrics):
    """
    Expected TTS metrics fields:
      - metrics.ttfb (seconds)
      - metrics.audio_duration (seconds)
      - metrics.characters_count (int)
    """
    ttfb = getattr(metrics, "ttfb", None) or getattr(metrics, "time_to_first_byte", None)
    audio_dur = getattr(metrics, "audio_duration", None)
    chars = getattr(metrics, "characters_count", None) or getattr(metrics, "characters", None)

    <http://logger.info|logger.info>(f"TTS metrics: ttfb={ttfb}s audio_duration={audio_dur}s characters={chars}")

    if ttfb is not None:
        tts_ttfb.observe(ttfb)
    if audio_dur is not None:
        tts_audio_duration.set(audio_dur)
    if chars is not None:
        tts_characters.set(chars)

# Attach listener
self.session.tts.on("metrics_collected", on_tts_metrics_collected)

6) Putting it all together (example agent on_enter)

async def on_enter(self):
    # Start Prometheus server already started above
    # Attach handlers
    self.session.stt.on("metrics_collected", on_stt_metrics_collected)
    self.session.stt.on("eou_metrics_collected", on_eou_metrics_collected)
    self.session.vad.on("metrics_collected", on_vad_event)
    self.session.llm.on("metrics_collected", on_llm_metrics_collected)
    self.session.tts.on("metrics_collected", on_tts_metrics_collected)
    <http://logger.info|logger.info>("Metrics listeners attached")

Notes and next steps • Field names on the metrics objects can vary between SDK versions. If a metric field in the snippets above returns None, print the whole metrics object (e.g.,

<http://logger.info|logger.info>(vars(metrics))

) to confirm available attributes. • If you prefer pushing to a different backend (Datadog, Influx, Cloud Monitoring), I can adapt the exporter calls. • I can also provide a version that formats metrics into a live console table using

rich

if you want a prettier terminal display. Would you like the console-table (rich) version, or do you want me to adapt these snippets to export to Datadog/another backend instead of Prometheus?