when using langfuse traces to capture traces I was able to s LiveKit Community #ask-ai

when using langfuse traces to capture traces, I wa...

cool-lizard-26153

12/25/2025, 12:51 PM

when using langfuse traces to capture traces, I was able to see low level details like interrupted true/false, STT confidence score etc along with the text that was delivered/sent. However, when I try to capture metrics using the usagecollector, I only have very basic details. How do I log low level details in the usage collector just like how langfuse is able to capture it?

tall-airline-16756

12/25/2025, 12:51 PM

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

tall-airline-16756

12/25/2025, 12:51 PM

Thanks — I understand you're asking how to get Langfuse‑level details (interrupted flag, STT confidence, incremental transcripts, etc.) into the UsageCollector, since UsageCollector currently returns only high‑level metrics. Short answer: UsageCollector is intentionally high‑level and does not capture those low‑level fields by default. To get the same detail you see in Langfuse, subscribe to the granular events the agent SDK emits, extract the fields you need, and enrich or forward them alongside the UsageCollector data. How to do that (concise steps): 1. Subscribe to low‑level events that contain the data you want: ◦

user_input_transcribed

— contains

text

is_final

confidence

(partial/final STT info). ◦

playback_finished

— contains

interrupted

flag (speaker interruption). ◦

metrics_collected

on AgentSession components — contains LLM/STT/TTS timing and errors. 1. Log or attach those fields when UsageCollector collects its summary: ◦ Capture the latest STT confidence and transcript in your session handlers. ◦ On

metrics_collected

, add those captured fields to the metric payload before calling

usage_collector.collect(...)

or forward them to your observability backend. Example patterns: • Event handler to capture STT details:

Copy code

@session.on("user_input_transcribed")
def on_transcript(transcript):
    last_stt = {"text": transcript.text, "confidence": transcript.confidence, "is_final": transcript.is_final}

• Playback finished to capture interruption:

Copy code

@session.output.audio.on("playback_finished")
def on_playback_finished(ev):
    <http://logger.info|logger.info>({"interrupted": ev.interrupted})

• Enrich UsageCollector:

Copy code

@session.on("metrics_collected")
def on_metrics(ev):
    ev.metrics.stt_confidence = last_stt.get("confidence")
    ev.metrics.last_transcript = last_stt.get("text")
    usage_collector.collect(ev.metrics)

1. Export/Store: Send these enriched records to your observability backend (or the same place you send UsageCollector data) so you can query them like Langfuse traces. If you’d like, tell me: • which SDK/language you’re using (Python, Node, etc.), and • one example field you want included (e.g., per‑utterance

confidence

interrupted

cool-lizard-26153

12/25/2025, 12:55 PM

yea, please give for python sdk

tall-airline-16756

12/25/2025, 12:55 PM

Thanks — below is a concise Python SDK example that shows how to subscribe to the low‑level events, capture STT confidence and interruption status, and enrich UsageCollector records before collecting/exporting them.

Copy code

# Example: enrich UsageCollector with STT confidence and interrupted flag (Python SDK)

from livekit.agents import AgentSession, UsageCollector
import logging

logger = logging.getLogger("agent")
usage_collector = UsageCollector()

# state holders for latest low-level values
_last_stt = {"text": None, "confidence": None, "is_final": False}
_last_playback = {"interrupted": None}

session = AgentSession(...)

# Capture STT incremental/final transcripts and confidence
@session.on("user_input_transcribed")
def on_transcript(transcript):
    # transcript has .text, .confidence, .is_final
    _last_stt["text"] = transcript.text
    _last_stt["confidence"] = transcript.confidence
    _last_stt["is_final"] = transcript.is_final
    logger.debug("STT: text=%s confidence=%s final=%s", transcript.text, transcript.confidence, transcript.is_final)

# Capture interruption status when playback finishes
@session.output.audio.on("playback_finished")
def on_playback_finished(ev):
    # ev.interrupted is the interruption flag
    _last_playback["interrupted"] = ev.interrupted
    logger.debug("Playback finished: interrupted=%s", ev.interrupted)

# When the high-level metrics are emitted, enrich them with the captured low-level fields
@session.on("metrics_collected")
def on_metrics(ev):
    metrics = ev.metrics  # existing high-level metrics object
    # Attach low-level fields (names are examples; choose fields that fit your storage schema)
    metrics.meta = metrics.meta or {}
    metrics.meta["last_stt_text"] = _last_stt.get("text")
    metrics.meta["last_stt_confidence"] = _last_stt.get("confidence")
    metrics.meta["last_stt_is_final"] = _last_stt.get("is_final")
    metrics.meta["last_playback_interrupted"] = _last_playback.get("interrupted")

    # Optionally log or export enriched metrics
    <http://logger.info|logger.info>("Enriched metrics: %s", metrics.meta)
    usage_collector.collect(metrics)

# Start session as you normally would
session.start()

If you want, I can: • adapt this to your exact session/agent initialization pattern, • show how to persist/export the enriched metrics to a DB or observability backend, • include per-word confidence aggregation. Which would be most helpful?

cool-lizard-26153

12/26/2025, 8:02 AM

confidence attribute deosn't exist on the usertranscript event

tall-airline-16756

12/26/2025, 8:02 AM

Thanks — good catch. You're right:

user_input_transcribed

does not include a

confidence

field. Use one of these approaches in the Python SDK to capture STT confidence and interruption status: • Capture confidence from STT provider metrics: ◦ Subscribe to the STT component's

metrics_collected

event — it can include provider confidence scores. • Inspect transcript alternatives (if provided): ◦

user_input_transcribed

exposes

alternatives

(if the provider returns them); you can read per-alternative confidence there. • Capture interruption from playback/speech handles: ◦ Use

playback_finished

or check

SpeechHandle.interrupted

after

say()

generate_reply()

to get the interruption flag. Updated concise Python pattern:

Copy code

_last_stt = {"text": None, "confidence": None, "is_final": False}
_last_playback = {"interrupted": None}
usage_collector = UsageCollector()
session = AgentSession(...)

# transcripts (use alternatives if confidence present)
@session.on("user_input_transcribed")
def on_transcript(t):
    _last_stt["text"] = getattr(t, "transcript", None) or getattr(t, "text", None)
    alts = getattr(t, "alternatives", None)
    if alts and len(alts) > 0:
        _last_stt["confidence"] = getattr(alts[0], "confidence", None)
    _last_stt["is_final"] = getattr(t, "is_final", False)

# STT provider metrics (preferred if you need authoritative confidence)
@session.stt.on("metrics_collected")
def on_stt_metrics(ev):
    # ev.metrics may contain provider confidence fields
    _last_stt["confidence"] = getattr(ev.metrics, "confidence", _last_stt["confidence"])

# interruption
@session.output.audio.on("playback_finished")
def on_playback_finished(ev):
    _last_playback["interrupted"] = getattr(ev, "interrupted", None)

# enrich UsageCollector
@session.on("metrics_collected")
def on_metrics(ev):
    m = ev.metrics
    m.meta = m.meta or {}
    m.meta["last_stt_text"] = _last_stt["text"]
    m.meta["last_stt_confidence"] = _last_stt["confidence"]
    m.meta["last_stt_is_final"] = _last_stt["is_final"]
    m.meta["last_playback_interrupted"] = _last_playback["interrupted"]
    usage_collector.collect(m)

If you want, tell me which STT provider you’re using and I’ll point to the exact metrics field name (some providers put confidence in the STT metrics, others return it in

alternatives

cool-lizard-26153

12/26/2025, 8:13 AM

This is my logic to capture metrics:

Copy code

@session.on("metrics_collected")
    def _on_metrics_collected(ev: MetricsCollectedEvent):
        _metrics = ev.metrics
        if isinstance(_metrics, STTMetrics):
            _metrics.metadata["transcript"] = _last_stt.get("text")
            _metrics.metadata["confidence"] = _last_stt.get("confidence")
            _metrics.metadata["is_final"] = _last_stt.get("is_final")
        if isinstance(_metrics, TTSMetrics):
            _metrics.metadata["interrupted"] = _last_playback.get("interrupted")
            _metrics.metadata["synchronized_transcript"] = _last_playback.get("synchronized_transcript")
            _metrics.metadata["playback_position"] = _last_playback.get("playback_position")
        <http://logger.info|logger.info>(f"Custom metrics collected here are: {_metrics} for metric type: {type(_metrics)}")
        metrics.log_metrics(ev.metrics)

during the call, I only see EOU and VAD metrics being logged. TTS STT and none of the other metrics are being loged here. Why is that happening and how to fix it?

tall-airline-16756

12/26/2025, 8:13 AM

Thanks — you’re very close. The reason you only see EOU and VAD metrics is that STT and TTS metrics are emitted by their specific component instances (and only when those components actually run and/or the provider emits metrics). Your handler is listening at the session root, so it only receives session-level or forwarded metrics (EOU/VAD), not the STT/TTS component metrics. What to change (concise): 1. Attach listeners to the component instances, not only the session:

Copy code

# attach once after session/tts/stt are created
session.stt.on("metrics_collected", on_stt_metrics)
session.stt.on("eou_metrics_collected", on_eou_metrics)  # if you need EOU from STT
session.tts.on("metrics_collected", on_tts_metrics)

1. Keep your enrichment logic but run it inside those handlers:

Copy code

def on_stt_metrics(ev: STTMetrics):
    ev.metadata["transcript"] = _last_stt.get("text")
    ev.metadata["confidence"] = _last_stt.get("confidence")
    ev.metadata["is_final"] = _last_stt.get("is_final")
    <http://logger.info|logger.info>("STT metrics: %s", ev)
    metrics.log_metrics(ev)

def on_tts_metrics(ev: TTSMetrics):
    ev.metadata["interrupted"] = _last_playback.get("interrupted")
    ev.metadata["synchronized_transcript"] = _last_playback.get("synchronized_transcript")
    ev.metadata["playback_position"] = _last_playback.get("playback_position")
    <http://logger.info|logger.info>("TTS metrics: %s", ev)
    metrics.log_metrics(ev)

1. Verify your components actually run and emit metrics: • Ensure STT runs (speech detected / transcription happened). • Ensure TTS runs (agent produced audio/synthesis). • Some STT/TTS providers or adapters do not emit component metrics (e.g., certain StreamAdapter wrappers). If you use such a provider, switch to or configure a provider known to emit metrics (or instrument the wrapper to forward metrics). 1. Debug tips: • Add temporary handlers to confirm events fire:

Copy code

session.stt.on("metrics_collected", lambda e: logger.debug("stt metrics fired"))
session.tts.on("metrics_collected", lambda e: logger.debug("tts metrics fired"))

• Log

user_speech_started

agent_speaking

cool-lizard-26153

12/26/2025, 9:52 AM

Tried the above shared approach but unfortunately all those are getting logged at the end of the call and not during the call

tall-airline-16756

12/26/2025, 9:53 AM

Thanks — I hear you: STT/TTS metrics should stream during the call, and seeing them only at the end is frustrating. To pinpoint the cause I need a few quick details so I can give a targeted fix. Please paste a minimal snippet showing: • how you construct the AgentSession (STT/TTS providers or adapters used), • where you attach listeners (exact lines for

session.start()

ctx.connect()

, and your

session.stt.on(...)

session.tts.on(...)

attachments), • whether you attach listeners in

on_enter

or after connect, • whether you perform any blocking I/O in the handlers (sync DB writes, HTTP calls), and • whether you’re testing with real speech input or simulated audio. While I wait, try these immediate checks (one‑line actions): • Ensure listeners are attached before connect/start (attach in

on_enter

or immediately after creating the session). • Attach to component instances:

session.stt.on("metrics_collected", ...)

and

session.tts.on("metrics_collected", ...)

. • Offload handler work: wrap heavy work with

asyncio.create_task(...)

. • Confirm provider supports real‑time metrics (e.g., Deepgram, Cartesia); `StreamAdapter`/non‑streaming providers may batch metrics. • Add quick debug prints in the exact handlers to verify when they fire: ◦

session.stt.on("metrics_collected", lambda e: print("stt metric", time.time()))

◦

session.tts.on("metrics_collected", lambda e: print("tts metric", time.time()))

cool-lizard-26153

12/26/2025, 9:57 AM

can you give full snippet on how to use it inside on_enter?

tall-airline-16756

12/26/2025, 9:58 AM

Thanks — below is a concise, complete Python example you can drop into your agent. It attaches STT/TTS metric listeners inside

on_enter

, uses

asyncio.create_task

so handlers don't block, and logs metrics in real time during the call.

Copy code

import asyncio
import logging
from datetime import datetime
from livekit.agents import Agent, AgentSession, inference, AgentServer, JobContext, cli
from livekit.agents.metrics import STTMetrics, TTSMetrics
from livekit.plugins import silero

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

def prewarm(proc):
    proc.userdata["vad"] = silero.VAD.load()

server.setup_fnc = prewarm

class MetricsAgent(Agent):
    def __init__(self):
        super().__init__(instructions="You are a helpful agent that responds to user speech.")

    async def on_enter(self):
        # Handlers must offload work so they do not block audio pipeline
        def stt_metrics_handler(metrics: STTMetrics):
            asyncio.create_task(self._handle_stt_metrics(metrics))

        def tts_metrics_handler(metrics: TTSMetrics):
            asyncio.create_task(self._handle_tts_metrics(metrics))

        # Attach listeners to the component instances (attach here so session is available)
        if self.session.stt:
            self.session.stt.on("metrics_collected", stt_metrics_handler)
        if self.session.tts:
            self.session.tts.on("metrics_collected", tts_metrics_handler)

        # Kick off a reply (optional) after attaching listeners
        self.session.generate_reply(instructions="Hello — speak now to see STT/TTS metrics in real time.")

    async def _handle_stt_metrics(self, metrics: STTMetrics):
        ts = datetime.fromtimestamp(metrics.timestamp).isoformat() if metrics.timestamp else "n/a"
        <http://logger.info|logger.info>("STT metrics | req=%s | ts=%s | dur=%.3fs | audio_dur=%.3fs | streamed=%s | error=%s",
                    getattr(metrics, "request_id", None), ts,
                    getattr(metrics, "duration", 0.0), getattr(metrics, "audio_duration", 0.0),
                    getattr(metrics, "streamed", False), getattr(metrics, "error", None))

        # Example: enrich and forward to your metrics pipeline (non-blocking)
        # await send_to_backend(metrics)  # if async; wrap with create_task if called here

    async def _handle_tts_metrics(self, metrics: TTSMetrics):
        ts = datetime.fromtimestamp(metrics.timestamp).isoformat() if metrics.timestamp else "n/a"
        <http://logger.info|logger.info>("TTS metrics | req=%s | ts=%s | ttfb=%.3fs | dur=%.3fs | chars=%s | error=%s",
                    getattr(metrics, "request_id", None), ts,
                    getattr(metrics, "ttfb", 0.0), getattr(metrics, "duration", 0.0),
                    getattr(metrics, "characters_count", None), getattr(metrics, "error", None))

        # Example: attach interruption info you've captured elsewhere
        # metrics.metadata = metrics.metadata or {}
        # metrics.metadata["interrupted"] = last_playback.get("interrupted")
        # await send_to_backend(metrics)

@server.rtc_session()
async def entrypoint(ctx: JobContext):
    session = AgentSession(
        stt=inference.STT(model="deepgram/nova-3-general"),
        llm=inference.LLM(model="openai/gpt-4o-mini"),
        tts=inference.TTS(model="cartesia/sonic-3", voice="default"),
        vad=ctx.proc.userdata["vad"],
        preemptive_generation=True,
    )

    await session.start(agent=MetricsAgent(), room=ctx.room)
    await ctx.connect()

if __name__ == "__main__":
    cli.run_app(server)

If you want, I can: • add STT alternative parsing to capture per-alternative confidence, • show how to enrich metrics with your

_last_stt

_last_playback

state, • or adapt this to your exact STT/TTS providers. Which would you prefer?

cool-lizard-26153

12/26/2025, 10:30 AM

yes, show how to enrich with

_last_stt

and

_last_playback

. Also, the STT ones are coming but TTS are not. Can you fix that too please?

tall-airline-16756

12/26/2025, 10:31 AM

Thanks — below is a concise, ready-to-run Python agent example that (1) attaches STT and TTS metric listeners inside

on_enter

, (2) enriches metrics with your

_last_stt

and

_last_playback

state, and (3) fixes TTS metrics not appearing by attaching the listener to the actual TTS instance. It also offloads any heavy work so handlers run during the call (not just at the end). Drop this into your agent code and adapt provider/model names as needed.

Copy code

import asyncio
import logging
from datetime import datetime
from typing import Optional

from livekit.agents import Agent, AgentSession, AgentServer, JobContext, cli, inference
from livekit.agents.metrics import STTMetrics, TTSMetrics
from livekit.plugins import silero  # example VAD plugin

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

def prewarm(proc):
    proc.userdata["vad"] = silero.VAD.load()

server.setup_fnc = prewarm

class MetricsAgent(Agent):
    def __init__(self):
        super().__init__(instructions="You are a helpful agent.")
        # state you asked to enrich metrics with
        self._last_stt = {"text": None, "confidence": None, "is_final": False}
        self._last_playback = {"interrupted": None, "synchronized_transcript": None, "playback_position": None}

    async def on_enter(self):
        # STT metrics handler (attach to the STT component)
        def stt_metrics_handler(metrics: STTMetrics):
            asyncio.create_task(self._on_stt_metrics(metrics))

        # TTS metrics handler (attach to the TTS component) — critical to attach to TTS instance
        def tts_metrics_handler(metrics: TTSMetrics):
            asyncio.create_task(self._on_tts_metrics(metrics))

        # Attach listeners to component instances (session.stt and session.tts)
        if self.session.stt:
            self.session.stt.on("metrics_collected", stt_metrics_handler)
        else:
            logger.warning("session.stt is None — STT metrics won't be emitted")

        if self.session.tts:
            self.session.tts.on("metrics_collected", tts_metrics_handler)
        else:
            logger.warning("session.tts is None — TTS metrics won't be emitted")

        # Also attach lower-level events to populate _last_stt/_last_playback state
        # user_input_transcribed -> update _last_stt (alternatives may have confidence)
        @self.session.on("user_input_transcribed")
        def _on_user_transcribed(ev):
            # prefer attributes used by your STT provider; adjust field names if different
            text = getattr(ev, "transcript", None) or getattr(ev, "text", None)
            alts = getattr(ev, "alternatives", None) or []
            confidence = None
            if len(alts) > 0:
                confidence = getattr(alts[0], "confidence", None)
            self._last_stt.update({"text": text, "confidence": confidence, "is_final": getattr(ev, "is_final", False)})

        # playback_finished -> update _last_playback (interruption flag)
        if self.session.output and hasattr(self.session.output, "audio"):
            @self.session.output.audio.on("playback_finished")
            def _on_playback_finished(ev):
                self._last_playback.update({
                    "interrupted": getattr(ev, "interrupted", None),
                    "synchronized_transcript": getattr(ev, "synchronized_transcript", None),
                    "playback_position": getattr(ev, "playback_position", None),
                })

        # Optionally kickoff a reply
        self.session.generate_reply(instructions="Hello — speak now to see STT/TTS metrics in real time.")

    async def _on_stt_metrics(self, metrics: STTMetrics):
        # Enrich STT metrics with _last_stt state
        metrics.metadata = metrics.metadata or {}
        metrics.metadata["last_stt_text"] = self._last_stt.get("text")
        metrics.metadata["last_stt_confidence"] = self._last_stt.get("confidence")
        metrics.metadata["last_stt_is_final"] = self._last_stt.get("is_final")

        <http://logger.info|logger.info>("STT metrics (enriched): request_id=%s duration=%.3f audio_dur=%.3f meta=%s",
                    getattr(metrics, "request_id", None),
                    getattr(metrics, "duration", 0.0),
                    getattr(metrics, "audio_duration", 0.0),
                    metrics.metadata)

        # Example: non-blocking forward to backend
        # asyncio.create_task(send_to_backend(metrics.model_dump()))

    async def _on_tts_metrics(self, metrics: TTSMetrics):
        # Enrich TTS metrics with _last_playback state
        metrics.metadata = metrics.metadata or {}
        metrics.metadata["last_playback_interrupted"] = self._last_playback.get("interrupted")
        metrics.metadata["last_playback_synchronized_transcript"] = self._last_playback.get("synchronized_transcript")
        metrics.metadata["last_playback_position"] = self._last_playback.get("playback_position")

        <http://logger.info|logger.info>("TTS metrics (enriched): request_id=%s ttfb=%.3f duration=%.3f meta=%s",
                    getattr(metrics, "request_id", None),
                    getattr(metrics, "ttfb", 0.0),
                    getattr(metrics, "duration", 0.0),
                    metrics.metadata)

        # Example: non-blocking forward to backend
        # asyncio.create_task(send_to_backend(metrics.model_dump()))

@server.rtc_session()
async def entrypoint(ctx: JobContext):
    session = AgentSession(
        stt=inference.STT(model="deepgram/nova-3-general"),   # ensure provider supports metrics
        llm=inference.LLM(model="openai/gpt-4o-mini"),
        tts=inference.TTS(model="cartesia/sonic-3", voice="default"),  # ensure this TTS emits metrics
        vad=ctx.proc.userdata["vad"],
        preemptive_generation=True,
    )

    await session.start(agent=MetricsAgent(), room=ctx.room)
    await ctx.connect()

if __name__ == "__main__":
    cli.run_app(server)

Quick checklist if TTS metrics still don’t appear: • Confirm you passed a TTS instance that actually emits metrics (some wrappers/adapters do not). Use built-in inference TTS or known plugin (Cartesia/Eleven/OpenAI TTS). • Ensure the listener is attached before TTS runs (attaching in

on_enter

as above does that). • Add a quick debug to verify the TTS instance:

<http://logger.info|logger.info>("tts instance: %s has_on: %s", self.session.tts, hasattr(self.session.tts, "on"))

• Avoid blocking operations in handlers; use

asyncio.create_task

for any I/O. If you tell me which TTS provider/plugin you use (exact class or package), I’ll confirm the exact attachment point and whether that provider emits

metrics_collected

mid-call. Sources: Creating and Publishing Custom Metrics

cool-lizard-26153

12/26/2025, 10:39 AM

the metadata object for TTSMetrics and STTMetrics can't be updated. That's the error I see

tall-airline-16756

12/26/2025, 10:40 AM

Thanks — good catch. The error happens because

TTSMetrics

and

STTMetrics

metadata fields are immutable / not meant to be mutated in-place. Instead of trying to update

metrics.metadata[...]

, create an enriched payload (a new dict) and log or forward that. Don’t mutate the metric object. What to do (concise): 1. Build an enriched record (copy the metric data + your state). 2. Send/log that record (UsageCollector or external backend) — nonblocking. Example pattern:

Copy code

async def _on_stt_metrics(self, metrics: STTMetrics):
    # don't mutate metrics.metadata
    base = metrics.model_dump() if hasattr(metrics, "model_dump") else vars(metrics)
    enriched = {
        "type": "stt_metrics",
        "timestamp": getattr(metrics, "timestamp", None),
        "base": base,
        "last_stt": self._last_stt,
    }
    <http://logger.info|logger.info>("Enriched STT: %s", enriched)
    asyncio.create_task(send_to_backend(enriched))

async def _on_tts_metrics(self, metrics: TTSMetrics):
    base = metrics.model_dump() if hasattr(metrics, "model_dump") else vars(metrics)
    enriched = {
        "type": "tts_metrics",
        "timestamp": getattr(metrics, "timestamp", None),
        "base": base,
        "last_playback": self._last_playback,
    }
    <http://logger.info|logger.info>("Enriched TTS: %s", enriched)
    asyncio.create_task(send_to_backend(enriched))

If you must keep a metrics-like object, make a shallow copy first:

Copy code

copied = dict(base)
copied["metadata"] = dict(copied.get("metadata") or {})
copied["metadata"]["last_stt_text"] = self._last_stt["text"]
# forward copied

Notes/tips: • Use

metrics.model_dump()

vars(metrics)

to serialize the metric contents. • Always offload I/O with

asyncio.create_task()

so handlers run during the call. • Continue attaching listeners to

session.stt

and

session.tts

on_enter

(that part is correct). Tell me where you want to forward enriched records (UsageCollector, Langfuse, DB) and I’ll give the exact nonblocking send snippet for that backend.

2 Views

Open in Slack

Previous Next