https://livekit.io logo
Join Slack
Powered by
How can i measure the call metrics manually, i hav...
# ask-ai
e
How can i measure the call metrics manually, i have a python agent and i want to measure the p50, p90 and p99, ttfb for llm, tts and stt services, is there anything that i can use or i have to write something custom. Langfuse telemetry is shit, cant understand anything.
t
Thanks for reaching out, Aditya. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
Thanks — I hear your frustration with Langfuse. You don't need to re-invent everything: LiveKit agents already emit the per-request metrics you need (LLMMetrics, TTSMetrics, STTMetrics) including TTFT/TTFB and durations. You can subscribe to those events in your Python agent and compute p50/p90/p99 yourself or push them to a telemetry backend. What you can do (minimal steps): 1) Subscribe to metrics events in your agent • LLMMetrics provides 
ttft
and 
duration
. • TTSMetrics provides 
ttfb
and 
duration
. • STTMetrics provides 
duration
and end-of-utterance timing. Attach an event handler that appends the numeric latencies to in-memory buffers. 2) Compute percentiles • Use numpy (or similar) to compute percentiles: ◦ 
np.percentile(values, 50)
, 
90
, 
99
• Keep a rolling buffer (e.g., 
collections.deque(maxlen=10000)
) to avoid unbounded memory. 3) Track TTFB / TTFT correctly • For LLM: record Time To First Token (
ttft
) and full 
duration
. • For TTS: record Time To First Byte (
ttfb
) and 
duration
. • For STT: record 
duration
and EOU timing as relevant. 4) Production options (avoid Langfuse) • Export metrics via OpenTelemetry (Python SDK) and use Prometheus/Grafana, Datadog, or Influx for built-in percentile aggregation and dashboards. • Or push summarized histograms/percentiles from your agent to your metrics backend directly. If you’d like, I can: • Show a concise Python snippet that subscribes to the LiveKit metric events and computes rolling p50/p90/p99. • Show a short OpenTelemetry example to export histograms for TTFT/TTFB. Which of those would you prefer? Sources: LLM Metrics | LiveKit docs | TTS Metrics | LiveKit docs | STT Metrics | LiveKit docs | Introducing LiveKit Inference: A unified model interface for voice AI | How can I reduce latency in voice agents using STT, TTS and LLM?
e
can you show me the usage of llm metrics etc
t
Below is a concise Python example showing how to subscribe to LiveKit agent metrics (LLM, TTS, STT), collect latencies into rolling buffers, and compute p50/p90/p99 and TTFB/TTFT. Pick the parts you need and integrate into your agent.
Copy code
import asyncio
from collections import deque, defaultdict
import numpy as np

# Rolling buffers per metric (adjust maxlen as needed)
buffers = defaultdict(lambda: deque(maxlen=10000))

# Example handlers for metric events
def handle_llm_metrics(metrics):
    # metrics.ttft and metrics.duration are numeric seconds
    buffers["llm_ttft"].append(metrics.ttft)
    buffers["llm_duration"].append(metrics.duration)

def handle_tts_metrics(metrics):
    # metrics.ttfb and metrics.duration are numeric seconds
    buffers["tts_ttfb"].append(metrics.ttfb)
    buffers["tts_duration"].append(metrics.duration)

def handle_stt_metrics(metrics):
    # metrics.duration (and any EOU timing if present)
    buffers["stt_duration"].append(metrics.duration)

# Percentile helper
def compute_percentiles(deque_obj):
    arr = np.asarray(deque_obj)
    if arr.size == 0:
        return None
    return {
        "p50": float(np.percentile(arr, 50)),
        "p90": float(np.percentile(arr, 90)),
        "p99": float(np.percentile(arr, 99)),
        "count": int(arr.size),
    }

# Example function to print current metrics snapshot
def print_snapshot():
    snap = {
        "llm_ttft": compute_percentiles(buffers["llm_ttft"]),
        "llm_duration": compute_percentiles(buffers["llm_duration"]),
        "tts_ttfb": compute_percentiles(buffers["tts_ttfb"]),
        "tts_duration": compute_percentiles(buffers["tts_duration"]),
        "stt_duration": compute_percentiles(buffers["stt_duration"]),
    }
    for k, v in snap.items():
        if v is None:
            print(f"{k}: no samples")
        else:
            print(f"{k}: p50={v['p50']:.3f}s p90={v['p90']:.3f}s p99={v['p99']:.3f}s count={v['count']}")

# Wiring into your LiveKit agent (pseudo-code, adapt to your agent API)
async def on_enter(session):
    # sync wrapper if LiveKit emits sync callbacks
    def llm_sync(metrics):
        asyncio.create_task(async_handle_llm(metrics))
    async def async_handle_llm(metrics):
        handle_llm_metrics(metrics)

    def tts_sync(metrics):
        asyncio.create_task(async_handle_tts(metrics))
    async def async_handle_tts(metrics):
        handle_tts_metrics(metrics)

    def stt_sync(metrics):
        asyncio.create_task(async_handle_stt(metrics))
    async def async_handle_stt(metrics):
        handle_stt_metrics(metrics)

    # Subscribe - replace with actual event registration from your agent
    session.llm.on("metrics_collected", llm_sync)
    session.tts.on("metrics_collected", tts_sync)
    session.stt.on("metrics_collected", stt_sync)

# Example runtime: periodically print snapshot
async def metrics_loop():
    while True:
        print_snapshot()
        await asyncio.sleep(30)

# Start example (adapt to your app startup)
if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    # hook into actual agent lifecycle; here we just run the printer loop
    loop.create_task(metrics_loop())
    loop.run_forever()
If you want, I can: • Convert the example to use OpenTelemetry histograms (for Prometheus/Grafana). • Add an example that persists aggregated percentiles to a DB or pushes them to your metrics backend. Which next step do you prefer?
2 Views
Open in Slack
PreviousNext
Powered by