Just like how langfuse tracer is able to capture very low-level metrics in the EOU detection, like interruption status, TTFT, TTFB etc., is there a way that I can capture them realtime during the call and push it to some external API service?

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.

so, I'm looking for the below type of metrics:

{
  "id": "item_d78ff4e8824f",
  "type": "message",
  "role": "assistant",
  "content": [
    " Sure, have you ever worked for SASR or"
  ],
  "interrupted": true,
  "extra": {},
  "metrics": {
    "started_speaking_at": 1766574071.152171,
    "stopped_speaking_at": 1766574073.8813922,
    "llm_node_ttft": 1.529258333001053,
    "tts_node_ttfb": 4.312403374999121,
    "e2e_latency": 7.346454858779907
  },
  "created_at": 1766574065.319689
}

which include content, latency, interruption status. I'm using python

Thanks — I’m responding to your example JSON and that you’re using Python. You can build that exact payload in Python by combining these LiveKit metric events in your agent:

LLMMetrics

(for

llm_node_ttft

and

started

),

TTSMetrics

(for

tts_node_ttfb

,

started_speaking_at

,

stopped_speaking_at

,

duration

), and user-state/VAD events (to set

interrupted

). Below is a concise Python example that captures content, timestamps, TTFT, TTFB, computes e2e latency, detects interruption, and POSTs the JSON to an external API. Code (copy into your agent process; adjust model names, endpoints, and IDs):

import asyncio
import time
import aiohttp
from datetime import datetime
from livekit.agents import Agent, AgentSession, inference
from livekit.agents.metrics import LLMMetrics, TTSMetrics
from livekit.agents import UserStateChangedEvent

API_ENDPOINT = "<https://your-api.example.com/metrics>"

class MetricsAgent(Agent):
    def __init__(self):
        super().__init__(instructions="You are helpful")
        self.current_id = None
        self.content_buffer = []
        self.interrupted = False
        self.llm_start_ts = None
        self.tts_start_ts = None
        self.tts_duration = None
        self.llm_ttft = None
        self.tts_ttfb = None

    async def on_enter(self):
        self.current_id = f"item_{int(time.time()*1000)}"
        self.content_buffer = []
        self.interrupted = False
        self.llm_start_ts = None
        self.tts_start_ts = None
        self.llm_ttft = None
        self.tts_ttfb = None
        self.tts_duration = None
        await self.session.generate_reply()  # trigger generation as appropriate

    async def on_llm_metrics(self, metrics: LLMMetrics):
        # capture TTFT and LLM start time
        if getattr(metrics, "ttft", None) is not None:
            self.llm_ttft = float(metrics.ttft)
        if getattr(metrics, "timestamp", None) is not None:
            # timestamp is wall-clock seconds since epoch in metrics
            self.llm_start_ts = float(metrics.timestamp)

    async def on_tts_metrics(self, metrics: TTSMetrics):
        # capture TTFB and TTS start/stop
        if getattr(metrics, "ttfb", None) is not None:
            self.tts_ttfb = float(metrics.ttfb)
        if getattr(metrics, "timestamp", None) is not None:
            self.tts_start_ts = float(metrics.timestamp)
        if getattr(metrics, "duration", None) is not None:
            self.tts_duration = float(metrics.duration)

        # When we have a TTS first byte event, assemble and send payload
        if self.tts_start_ts is not None:
            e2e = None
            if self.llm_start_ts is not None:
                e2e = (self.tts_start_ts - self.llm_start_ts)

            payload = {
                "id": self.current_id,
                "type": "message",
                "role": "assistant",
                "content": ["".join(self.content_buffer)],
                "interrupted": bool(self.interrupted),
                "extra": {},
                "metrics": {
                    "started_speaking_at": self.tts_start_ts,
                    "stopped_speaking_at": self.tts_start_ts + (self.tts_duration or 0.0),
                    "llm_node_ttft": self.llm_ttft or 0.0,
                    "tts_node_ttfb": self.tts_ttfb or 0.0,
                    "e2e_latency": e2e
                },
                "created_at": datetime.now().timestamp()
            }

            # send non-blocking
            asyncio.create_task(self._post_metric(payload))

            # reset buffers for next message
            self.content_buffer = []
            self.interrupted = False
            self.llm_start_ts = None
            self.tts_start_ts = None
            self.llm_ttft = None
            self.tts_ttfb = None
            self.tts_duration = None

    async def on_streamed_response_chunk(self, chunk: str):
        # accumulate LLM streaming content (method name depends on SDK; adapt if needed)
        self.content_buffer.append(chunk)

    async def on_user_state_changed(self, ev: UserStateChangedEvent):
        # if user starts speaking while agent is producing, mark interrupted
        if getattr(ev, "new_state", None) == "speaking":
            self.interrupted = True

    async def _post_metric(self, payload: dict):
        try:
            async with aiohttp.ClientSession() as session:
                async with <http://session.post|session.post>(API_ENDPOINT, json=payload) as resp:
                    if resp.status >= 400:
                        # optionally retry or log
                        print("Failed to post metrics:", resp.status)
        except Exception as e:
            print("Error sending metrics:", e)

# Example session wiring (inside your entrypoint)
async def start_session():
    tts = inference.TTS(model="cartesia/sonic-3", voice="default")
    llm = inference.LLM(model="openai/gpt-5-mini")
    stt = inference.STT(model="deepgram/nova-3-general")
    session = AgentSession(stt=stt, llm=llm, tts=tts)
    agent = MetricsAgent()

    # subscribe SDK events to agent handlers
    session.llm.on("metrics_collected", lambda e: asyncio.create_task(agent.on_llm_metrics(e)))
    session.tts.on("metrics_collected", lambda e: asyncio.create_task(agent.on_tts_metrics(e)))
    session.on("user_state_changed", lambda e: asyncio.create_task(agent.on_user_state_changed(e)))
    # adapt streamed chunk subscription to your SDK method
    session.llm.on("response_chunk", lambda c: asyncio.create_task(agent.on_streamed_response_chunk(c)))

    await session.start(agent=agent, room="your-room")

Notes and tips: • Use the

metrics.timestamp

fields (they are wall-clock epoch seconds) for

started_speaking_at

/

stopped_speaking_at

. For higher accuracy, you can use monotonic clocks locally but then timestamps won’t be absolute. • Interruption detection: mark

interrupted = True

when

user_state_changed

indicates the user started speaking during agent output or when

LLMMetrics.cancelled == True

. • Keep the POST non-blocking (we use

asyncio.create_task

) so audio/turn-taking isn’t delayed. • Tweak where you collect streamed content (final vs streaming chunks) depending on whether you want partial or final content in

content

. If you want, I can: • Return a version that batches metrics or retries failed POSTs. • Adapt the snippet to match the exact event names in your LiveKit SDK version. Which would you prefer? Sources: LLM Metrics | LiveKit docs | TTS Metrics | LiveKit docs | Turns overview | LiveKit docs | Simple Content Filter | LiveKit docs | examples/voice_agents | livekit/livekit_composite | DeepWiki

is there a way to capture the transcription confidence?

Yes — you can capture transcription confidence, but availability depends on the STT provider and how its plugin exposes confidence scores. What to do: • Subscribe to STT events in your agent:

metrics_collected

,

eou_metrics_collected

, and

user_input_transcribed

(or the STT transcript event your SDK emits). • Inspect the emitted transcript/metrics objects for a

confidence

field (some providers include per-utterance or per-word confidence). • If your provider doesn’t include confidence by default, enable it via the provider plugin settings (for example, AssemblyAI and Deepgram often provide confidence options via their plugin configuration). Short Python pattern: 1) Subscribe to STT events:

session.stt.on("metrics_collected", lambda e: asyncio.create_task(on_stt_metrics(e)))
session.stt.on("eou_metrics_collected", lambda e: asyncio.create_task(on_eou_metrics(e)))
session.on("user_input_transcribed", lambda t: asyncio.create_task(on_transcript(t)))

2) Inspect transcript/metrics for confidence:

async def on_transcript(transcript):
    # many providers put confidence on transcript['confidence'] or per-word tokens
    confidence = getattr(transcript, "confidence", None) or transcript.get("confidence")
    # or per-word: transcript.get("tokens", [])
    # include confidence in your payload if present

3) Fallbacks: • Use

EOUMetrics

fields (e.g.,

end_of_utterance_delay

,

transcription_delay

) as indirect signals of turn-detection confidence when explicit scores aren’t provided. • If you need explicit confidence but your current STT plugin doesn’t expose it, switch to or configure a plugin that does (AssemblyAI, Deepgram, etc.) and enable confidence in its

extra_kwargs

or plugin settings. If you tell me which STT provider/plugin you’re using (Deepgram, AssemblyAI, etc.), I can give a concrete Python snippet showing the exact field names and how to include confidence in your metric payload. Sources: STT Metrics | LiveKit docs | AssemblyAI plugin guide | LiveKit docs | Improving voice AI's turn detection with transformers

I'm using the metrics like you said can you tell me how I can have the below as a part of STT metrics: transcription confidence Below as part of TTS metrics:

time to first byte, 
e2e latency

Below for LLM metrics:

ttft

Thanks — I’m responding to each metric you listed and showing a compact Python pattern to capture them and emit a unified payload. Short summary: • STT transcription confidence: captured from the STT transcript/metrics object if your STT provider exposes it (e.g.,

transcript.confidence

or

transcript["confidence"]

). If not present, use EOU-related fields as an indirect signal. • TTS:

ttfb

is available on `TTSMetrics`; compute e2e latency as

tts_start_ts - llm_start_ts

. • LLM:

ttft

is available on

LLMMetrics

. Compact Python example (plug into your agent session; adapt event names to your SDK version and provider):

import asyncio
import time
from datetime import datetime
import aiohttp
from livekit.agents.metrics import LLMMetrics, TTSMetrics
from livekit.agents import UserStateChangedEvent

API_ENDPOINT = "<https://your-api.example.com/metrics>"

class MetricCollector:
    def __init__(self):
        self.reset()

    def reset(self):
        self.current_id = f"item_{int(time.time()*1000)}"
        self.content = []
        self.interrupted = False
        self.llm_start_ts = None
        self.llm_ttft = None
        self.tts_start_ts = None
        self.tts_ttfb = None
        self.tts_duration = None
        self.transcript_confidence = None

    async def on_llm_metrics(self, metrics: LLMMetrics):
        # capture TTFT and LLM start timestamp
        if getattr(metrics, "ttft", None) is not None:
            self.llm_ttft = float(metrics.ttft)
        if getattr(metrics, "timestamp", None) is not None:
            self.llm_start_ts = float(metrics.timestamp)

    async def on_tts_metrics(self, metrics: TTSMetrics):
        # capture TTFB, start timestamp, duration
        if getattr(metrics, "ttfb", None) is not None:
            self.tts_ttfb = float(metrics.ttfb)
        if getattr(metrics, "timestamp", None) is not None:
            self.tts_start_ts = float(metrics.timestamp)
        if getattr(metrics, "duration", None) is not None:
            self.tts_duration = float(metrics.duration)

        # compute e2e latency if possible
        e2e = None
        if self.llm_start_ts is not None and self.tts_start_ts is not None:
            e2e = self.tts_start_ts - self.llm_start_ts

        payload = {
            "id": self.current_id,
            "type": "message",
            "role": "assistant",
            "content": ["".join(self.content)],
            "interrupted": bool(self.interrupted),
            "extra": {},
            "metrics": {
                "started_speaking_at": self.tts_start_ts,
                "stopped_speaking_at": (self.tts_start_ts + (self.tts_duration or 0.0)) if self.tts_start_ts else None,
                "llm_node_ttft": self.llm_ttft or 0.0,
                "tts_node_ttfb": self.tts_ttfb or 0.0,
                "e2e_latency": e2e
            },
            "transcription": {
                "confidence": self.transcript_confidence
            },
            "created_at": datetime.now().timestamp()
        }

        asyncio.create_task(self._post_metric(payload))
        self.reset()

    async def on_transcript(self, transcript_obj):
        # many STT providers put confidence on transcript_obj.confidence or transcript_obj['confidence']
        conf = getattr(transcript_obj, "confidence", None) or transcript_obj.get("confidence") if isinstance(transcript_obj, dict) else None
        self.transcript_confidence = conf
        # also capture final text if provided
        text = getattr(transcript_obj, "text", None) or transcript_obj.get("text") if isinstance(transcript_obj, dict) else None
        if text:
            self.content.append(text)

    async def on_user_state_changed(self, ev: UserStateChangedEvent):
        if getattr(ev, "new_state", None) == "speaking":
            self.interrupted = True

    async def _post_metric(self, payload):
        try:
            async with aiohttp.ClientSession() as s:
                async with s.post(API_ENDPOINT, json=payload) as r:
                    if r.status >= 400:
                        print("Failed to post metrics:", r.status)
        except Exception as e:
            print("Error posting metrics:", e)

Event wiring sketch (inside your session entrypoint): • session.llm.on("metrics_collected", lambda e: asyncio.create_task(collector.on_llm_metrics(e))) • session.tts.on("metrics_collected", lambda e: asyncio.create_task(collector.on_tts_metrics(e))) • session.stt.on("user_input_transcribed", lambda t: asyncio.create_task(collector.on_transcript(t))) • session.on("user_state_changed", lambda ev: asyncio.create_task(collector.on_user_state_changed(ev))) • capture streaming content via whatever LLM chunk/event your SDK emits and append into collector.content Notes and recommendations: • Transcription confidence availability depends on your STT provider (Deepgram, AssemblyAI, etc.). If you don’t see a

confidence

field, enable it in the provider plugin settings or switch to a provider that exposes confidence. • Use

metrics.timestamp

(epoch seconds) for `started_speaking_at`/`stopped_speaking_at`. e2e latency =

tts_start_ts - llm_start_ts

. • Keep HTTP POSTs non-blocking (use background tasks) to avoid affecting audio/turn-taking. • If you tell me which STT provider you use, I’ll give the exact field name and a tailored snippet for that provider.

I was already collecting

metrics_collected

but that was never capturing TTS metrics at all.

Thanks — I hear you: you’re subscribed to

metrics_collected

but not seeing any TTS metrics. Let’s troubleshoot and get those TTS events showing up. Quick checklist — the most common causes: • The TTS instance isn’t created with metrics enabled or isn’t the same TTS object attached to your session. • You subscribed to the wrong event source (e.g., listening on

session

instead of the

session.tts

plugin). • You’re subscribed to

metrics_collected

but the SDK emits TTS metrics under the TTS plugin’s emitter (often

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

). • Your code attaches the handler after TTS events fired, or you replaced the TTS instance later. What to do (short, ordered steps): 1. Ensure you create a TTS instance and attach it to the session you start: ◦ session = AgentSession(..., tts=tts_instance, ...) 1. Subscribe directly on that TTS instance, not on session: ◦ session.tts.on("metrics_collected", handler) ◦ session.tts.on("metrics_collected", lambda e: asyncio.create_task(on_tts_metrics(e))) 1. Confirm you attach the handler before calling

await session.start(...)

or before you trigger generation/speaking. 2. In the handler, inspect the object to confirm fields: ◦ print(dir(metrics)) and print(metrics.dict or metrics) to see

ttfb

,

timestamp

,

duration

. 1. If no TTS metrics appear, verify you’re using a TTS plugin that emits metrics (most LiveKit TTS plugins do). If you constructed your own TTS wrapper, ensure it forwards

metrics_collected

events. Minimal Python example (ensure this wiring is used):

async def on_tts_metrics(metrics):
    print("TTS metrics event:", metrics)
    print("ttfb:", getattr(metrics, "ttfb", None))
    print("timestamp:", getattr(metrics, "timestamp", None))
    print("duration:", getattr(metrics, "duration", None))

tts = inference.TTS(model="cartesia/sonic-3", voice="default")
session = AgentSession(llm=llm, stt=stt, tts=tts)

# attach BEFORE session.start
session.tts.on("metrics_collected", lambda e: asyncio.create_task(on_tts_metrics(e)))

await session.start(agent=agent, room="room-name")

Additional diagnostics if still empty: • After the session is running, force a short TTS utterance and watch logs to confirm the TTS pipeline is invoked. • Dump event names from the TTS object (e.g.,

print(dir(session.tts))

) to ensure

metrics_collected

is present. • Check for

metrics_collected

on

session.tts

vs

session

— TTS metrics are emitted by the TTS plugin. If you share: • the TTS plugin you’re using (model name or provider), and • the small snippet where you create the TTS, attach handlers, and start the session, I’ll point out exactly where the handler should go and adapt the example to your code.

I'm using:

session.tts.on("metrics_collected", lambda e: asyncio.create_task(screening_agent.on_tts_metrics_collected(e)))

this logic but I'm not seeing these metrics being collected anywhere throughout the call. Why is that?

Thanks — I hear you: you attached the handler but TTS metrics never fire. Let’s troubleshoot quickly. Most likely causes and fixes (try these in order): 1. Handler attached too late • Attach before starting the session or before any TTS is invoked. Move your

session.tts.on(...)

call to run immediately after creating the

tts

instance and before

await session.start(...)

. 1. Wrong TTS instance on the session • Confirm

session.tts

is the same object you created. If you recreate/replace the TTS instance after wiring handlers, events won’t reach your handler. 1. TTS not actually producing audio during the call • Verify the TTS flow runs (trigger a short

session.say()

or

agent.say()

during the session) and watch logs for audio packets. If TTS never runs, no metrics will emit. 1. Event name mismatch or plugin doesn’t emit metrics • Confirm

metrics_collected

exists on your TTS:

print(dir(session.tts))

or

hasattr(session.tts, "on")

and inspect available events. Some custom/wrapped plugins might use different event names or need metrics explicitly enabled. 1. Confirm handler code and inspect payload • In the handler, log the full object to discover fields:

async def on_tts_metrics(metrics):
      print("TTS metrics event:", metrics)
      print("fields:", getattr(metrics, "__dict__", metrics))
  session.tts.on("metrics_collected", lambda e: asyncio.create_task(on_tts_metrics(e)))

This shows whether

ttfb

,

timestamp

, or

duration

exist. 1. Race with background threads / async context • Ensure you use

asyncio.create_task(...)

(you are), and that the event loop is running. If the handler raises, exceptions may be swallowed — add try/except and logging inside the handler. 1. Provider/plugin-specific settings • Some TTS providers require enabling metrics or debug mode. Check your TTS plugin options (model args /

extra_kwargs

) for any metric flags. If none of the above reveals the issue, please share: • The short code snippet that creates the TTS instance, wires the handler, and starts the session (3–6 lines is fine). • Which TTS plugin/model you’re using (e.g., cartesia/sonic-3 or a custom provider). I’ll point out the exact line to move or change so metrics start appearing.

In langfuse tracing, the eou_detection span is having all the below data:

{
  "attributes": {
    "room_id": "RM_QVYzo8Sf8NSc",
    "job_id": "AJ_YyPPrWbEhUXY",
    "lk.chat_ctx": {
      "items": [
        {
          "id": "lk.agent_task.instructions",
          "type": "message",
          "role": "system",
          "content": [
            ""
          ],
          "interrupted": false,
          "extra": {},
          "metrics": {},
          "created_at": 1766742365.388652
        },
        {
          "id": "item_a8dbe295953c",
          "type": "message",
          "role": "assistant",
          "content": [
            "<break time=\"3s\"/> Hi Bharath, this is Mellisa from SASR, I'm a virtual recruiter assigned to conduct your phone screening. Do you agree to continue ? "
          ],
          "interrupted": false,
          "extra": {},
          "metrics": {
            "started_speaking_at": 1766742386.8560681,
            "stopped_speaking_at": 1766742397.378554,
            "tts_node_ttfb": 0.12409229100740049
          },
          "created_at": 1766742397.380846
        },
        {
          "id": "item_d4001d767aed",
          "type": "message",
          "role": "user",
          "content": [
            "Yeah."
          ],
          "interrupted": false,
          "transcript_confidence": 1,
          "extra": {},
          "metrics": {
            "started_speaking_at": 1766742400.619081,
            "stopped_speaking_at": 1766742400.717828,
            "transcription_delay": 0.8043649196624756,
            "end_of_turn_delay": 1.5004770755767822,
            "on_user_turn_completed_delay": 0.000002540997229516506
          },
          "created_at": 1766742402.2185578
        },
        {
          "id": "item_6aa944287a53",
          "type": "message",
          "role": "assistant",
          "content": [
            "Perfect, thank you. So, additionally, to help us ensure accurate decision-making and review the full conversation, DHL would like your permission to record this call. This is optional...Um, just let me know if you'd prefer to proceed with or without recording. "
          ],
          "interrupted": false,
          "extra": {},
          "metrics": {
            "started_speaking_at": 1766742403.438019,
            "stopped_speaking_at": 1766742418.5350451,
            "llm_node_ttft": 0.9892517090047477,
            "tts_node_ttfb": 0.2262726670014672,
            "e2e_latency": 2.72019100189209
          },
          "created_at": 1766742402.2412229
        },
        {
          "id": "item_83694eb4749f",
          "type": "message",
          "role": "user",
          "content": [
            "Yeah. So I would like to proceed with recording."
          ],
          "interrupted": false,
          "transcript_confidence": 0.9991777777777777,
          "extra": {},
          "metrics": {
            "started_speaking_at": 1766742421.869071,
            "stopped_speaking_at": 1766742423.2212338,
            "transcription_delay": 1.8774352073669434,
            "end_of_turn_delay": 2.5969951152801514,
            "on_user_turn_completed_delay": 0.0000017080019460991025
          },
          "created_at": 1766742425.8183858
        },
        {
          "id": "item_679c05c439d4",
          "type": "message",
          "role": "assistant",
          "content": [
            "Before we get started, just a quick note: if I happen to interrupt you or go quiet, please keep going and I'll catch up. Alright, let's dive in! Have you ever worked for SASR or Exel Logistics in the past? "
          ],
          "interrupted": false,
          "extra": {},
          "metrics": {
            "started_speaking_at": 1766742427.603677,
            "stopped_speaking_at": 1766742438.115859,
            "llm_node_ttft": 1.151172249999945,
            "tts_node_ttfb": 0.6260351249948144,
            "e2e_latency": 4.382443189620972
          },
          "created_at": 1766742425.8416069
        },
        {
          "id": "item_bf18c76768a2",
          "type": "message",
          "role": "user",
          "content": [
            "No. Unfortunately, I have not."
          ],
          "interrupted": false,
          "extra": {},
          "metrics": {},
          "created_at": 1766742445.726874
        }
      ]
    },
    "lk.eou.probability": "0.619086503982544",
    "lk.eou.unlikely_threshold": "0.011",
    "lk.eou.endpointing_delay": "1.5",
    "lk.eou.language": "en"
  },
  "resourceAttributes": {
    "telemetry.sdk.language": "python",
    "telemetry.sdk.name": "opentelemetry",
    "telemetry.sdk.version": "1.39.0",
    "service.name": "unknown_service"
  },
  "scope": {
    "name": "livekit-agents",
    "attributes": {}
  }
}

how do I capture all this from metrics during the call and push to some external metric capturer?

Thanks — I’m responding to your request to capture the same rich fields shown in your langfuse eou_detection span and push them live to an external service. You can capture and forward those fields by subscribing to these sources during the call, assembling them into the JSON shape you showed, and posting them asynchronously. Key sources and what to extract: • STT (session.stt) ◦ Events:

user_input_transcribed

/ final transcript event, and

metrics_collected

or

eou_metrics_collected

◦ Fields to read: transcript text,

confidence

(if present), `started_speaking_at`/`stopped_speaking_at` (from STT metrics),

transcription_delay

,

end_of_turn_delay

◦ If

confidence

is not present, enable it in the provider plugin or use EOU-related fields as indirect confidence signals • LLM (session.llm) ◦ Event:

metrics_collected

◦ Fields:

ttft

and

timestamp

(LLM start) and

cancelled

(for interruptions) ◦ Also capture streamed response chunks to build

content

• TTS (session.tts) ◦ Event:

metrics_collected

on the TTS plugin (must subscribe on

session.tts

) ◦ Fields:

ttfb

,

timestamp

(tts start),

duration

◦ Compute e2e latency =

tts_start_ts - llm_start_ts

(use metric timestamps) • User/VAD / Turn events ◦ Event:

user_state_changed

and VAD events ◦ Use these to set

interrupted

when the user starts speaking during assistant speech Compact implementation plan (Python): 1. Create a collector object that holds the current item buffer: ◦ ids, content buffer, interrupted flag, llm_start_ts, llm_ttft, tts_start_ts, tts_ttfb, tts_duration, stt_started, stt_stopped, stt_confidence, other metrics. 1. Wire listeners before session.start: ◦ session.llm.on("metrics_collected", lambda e: asyncio.create_task(collector.on_llm_metrics(e))) ◦ session.tts.on("metrics_collected", lambda e: asyncio.create_task(collector.on_tts_metrics(e))) ◦ session.stt.on("metrics_collected", lambda e: asyncio.create_task(collector.on_stt_metrics(e))) ◦ session.on("user_state_changed", lambda e: asyncio.create_task(collector.on_user_state_changed(e))) ◦ session.llm.on("response_chunk", lambda c: asyncio.create_task(collector.on_stream_chunk(c))) ◦ session.stt.on("user_input_transcribed", lambda t: asyncio.create_task(collector.on_transcript(t))) 1. In handlers: ◦ Store timestamps and metric values as they arrive. ◦ Build

content

from streamed LLM chunks (or final LLM output). ◦ Set

interrupted = True

when

user_state_changed

shows

speaking

during assistant output or when

LLMMetrics.cancelled

is True. 1. Emit payload: ◦ Prefer emitting when TTS first byte arrives (TTSMetrics with `ttfb`/`timestamp`) because TTS marks assistant speaking start and you’ll have LL M start to compute e2e. If you need earlier emission, choose a different trigger (LLM end). ◦ Assemble JSON matching your example and POST with an async client (aiohttp) using

asyncio.create_task(...)

to avoid blocking audio. 1. Retry/logging: ◦ Add retry/backoff for failed posts and log handler exceptions (wrap handlers in try/except). Minimal focused code sketch (adapt event names to your SDK version):

import asyncio, time
import aiohttp
from datetime import datetime

API = "<https://your-api/metrics>"

class Collector:
    def __init__(self):
        self.reset()

    def reset(self):
        self.id = f"item_{int(time.time()*1000)}"
        self.content_chunks = []
        self.interrupted = False
        self.llm_start = None
        self.llm_ttft = None
        self.tts_start = None
        self.tts_ttfb = None
        self.tts_duration = None
        self.stt_started = None
        self.stt_stopped = None
        self.stt_confidence = None
        self.other = {}

    async def on_stream_chunk(self, chunk):
        self.content_chunks.append(chunk)

    async def on_llm_metrics(self, metrics):
        if getattr(metrics, "ttft", None) is not None:
            self.llm_ttft = float(metrics.ttft)
        if getattr(metrics, "timestamp", None) is not None:
            self.llm_start = float(metrics.timestamp)
        if getattr(metrics, "cancelled", None):
            self.interrupted = True

    async def on_tts_metrics(self, metrics):
        # capture ttfb/start/duration
        if getattr(metrics, "ttfb", None) is not None:
            self.tts_ttfb = float(metrics.ttfb)
        if getattr(metrics, "timestamp", None) is not None:
            self.tts_start = float(metrics.timestamp)
        if getattr(metrics, "duration", None) is not None:
            self.tts_duration = float(metrics.duration)

        # compute e2e if llm_start available
        e2e = None
        if self.llm_start and self.tts_start:
            e2e = self.tts_start - self.llm_start

        payload = {
            "id": self.id,
            "type": "message",
            "role": "assistant",
            "content": ["".join(self.content_chunks)],
            "interrupted": bool(self.interrupted),
            "extra": {},
            "metrics": {
                "started_speaking_at": self.tts_start,
                "stopped_speaking_at": (self.tts_start + (self.tts_duration or 0)),
                "llm_node_ttft": self.llm_ttft,
                "tts_node_ttfb": self.tts_ttfb,
                "e2e_latency": e2e
            },
            "transcript_confidence": self.stt_confidence,
            "created_at": datetime.now().timestamp()
        }

        asyncio.create_task(self._post(payload))
        self.reset()

    async def on_stt_metrics(self, stt_metrics):
        # save start/stop/confidence if available
        if getattr(stt_metrics, "timestamp", None):
            self.stt_started = float(stt_metrics.timestamp)
        if getattr(stt_metrics, "end_timestamp", None):
            self.stt_stopped = float(stt_metrics.end_timestamp)
        self.stt_confidence = getattr(stt_metrics, "confidence", self.stt_confidence)

    async def on_transcript(self, transcript):
        # many providers put confidence on transcript.confidence or transcript["confidence"]
        conf = getattr(transcript, "confidence", None) or (transcript.get("confidence") if isinstance(transcript, dict) else None)
        if conf is not None:
            self.stt_confidence = float(conf)
        text = getattr(transcript, "text", None) or (transcript.get("text") if isinstance(transcript, dict) else None)
        if text:
            self.content_chunks.append(text)

    async def on_user_state_changed(self, ev):
        if getattr(ev, "new_state", None) == "speaking":
            self.interrupted = True

    async def _post(self, payload):
        try:
            async with aiohttp.ClientSession() as s:
                async with s.post(API, json=payload) as r:
                    if r.status >= 400:
                        print("post failed", r.status)
        except Exception as e:
            print("post error", e)

Wiring (important: attach handlers before start): • session.llm.on("metrics_collected", lambda e: asyncio.create_task(collector.on_llm_metrics(e))) • session.tts.on("metrics_collected", lambda e: asyncio.create_task(collector.on_tts_metrics(e))) • session.stt.on("metrics_collected", lambda e: asyncio.create_task(collector.on_stt_metrics(e))) • session.stt.on("user_input_transcribed", lambda t: asyncio.create_task(collector.on_transcript(t))) • session.llm.on("response_chunk", lambda c: asyncio.create_task(collector.on_stream_chunk(c))) • session.on("user_state_changed", lambda ev: asyncio.create_task(collector.on_user_state_changed(ev))) Troubleshooting tips (TTS metrics not appearing): • Ensure you subscribe on

session.tts

(not

session

) before

session.start(...)

. • Confirm the TTS plugin you use emits

metrics_collected

. Print

dir(session.tts)

to inspect. • Trigger a TTS utterance and log the TTS handler to confirm it runs. • Wrap handlers in try/except and log exceptions — silent failures can drop events. If you want, I can: • Provide a drop-in, end-to-end runnable example tailored to your exact SDK version and TTS/STT plugins, or • Inspect a small snippet of your current wiring (tts creation, handler attach lines, session.start) and point out what to change. Which would you prefer?

I'm using cartesia and so far i never observed it emitting metrics_collected. can you confirm what I could do here?

Thanks — I hear you: you’re using Cartesia TTS and haven’t seen any

metrics_collected

events from it. Here’s what you can do next. Short answer: • Cartesia TTS in your setup is not emitting

metrics_collected

, so you won’t see TTS metrics unless you either enable metrics on the TTS plugin (if supported) or compute the TTS metrics locally from observable events/timestamps. You can still capture LLM

ttft

and STT confidence from their respective events. Concrete next steps 1) Confirm wiring and versions • Ensure your handler is attached before session start: ◦ Create TTS, attach handler to

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

, then call

await session.start(...)

. • Print available events on the TTS object to confirm it exposes `metrics_collected`: ◦

print(dir(session.tts))

or

print(session.tts.__dict__)

2) If Cartesia TTS does not emit

metrics_collected

(what you’re observing) • Compute TTS metrics yourself using timestamps: ◦ Capture LLM start timestamp from

LLMMetrics.timestamp

and

LLMMetrics.ttft

. ◦ Capture the moment you begin sending TTS audio (or first packet) by recording the time when the TTS playback begins in your agent (see option A below). ◦ Treat the difference between TTS-first-audio-time and LLM-start-time as your e2e latency. ◦ For TTFB: measure time from when you call the TTS synth/send function to the time you observe the first audio frame/packet leaving your process (first byte played or sent). Options to capture TTS first-byte / speaking timestamps • Option A (preferred if you control TTS send): instrument the TTS send/play code path ◦ When you call the TTS plugin to start synthesis (or when sending the first audio packet to the RTC track), record

tts_request_time = time.time()

. ◦ When you observe the first audio frame actually emitted or when the SDK signals first-playback, record

tts_first_audio_time = time.time()

(this may be when you write the first buffer to the output or receive a callback). ◦ TTFB ~ (

tts_first_audio_time - tts_request_time

). ◦ E2E ~ (

tts_first_audio_time - llm_start_ts

). • Option B (if you only see session-level events): record wall-clock time when you call

session.say()

(or equivalent) and when audio begins/track becomes active; use those timestamps to approximate

tts_node_ttfb

and

started_speaking_at

. 3) Capture LLM TTFT and STT confidence • LLM: subscribe to

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

and read

metrics.ttft

and

metrics.timestamp

. • STT confidence: subscribe to

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

or

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

and inspect the transcript object for

confidence

(field name may be

confidence

or in tokens). If absent, enable confidence in that STT provider’s plugin settings. 4) Assemble and POST the payload • Build the JSON when you have TTS first-byte (use that as the canonical emit point), include: ◦ content (collected LLM chunks / final text) ◦ interrupted (mark true when

user_state_changed

shows user speaking during assistant output or

LLMMetrics.cancelled==True

) ◦ metrics:

started_speaking_at

(tts_first_audio_time),

stopped_speaking_at

(tts_first_audio_time + tts_duration),

llm_node_ttft

,

tts_node_ttfb

,

e2e_latency

◦ transcription confidence from STT events • Send via async POST (aiohttp + asyncio.create_task) so it won’t block audio. 5) Quick diagnostics you can run now • Immediately after session start run: ◦ print(dir(session.tts)) — confirm

metrics_collected

exists. ◦ Attach a trivial TTS call and log timestamps around the call to capture TTFB yourself. ◦ Log

session.llm

metrics to ensure

ttft

is available. If you want, share the 3–6 lines where you create the TTS, attach handlers, and start the session (the lines around

tts = inference.TTS(...)

,

session = AgentSession(..., tts=tts)

, your

session.tts.on(...)

, and

await session.start(...)

) and I’ll point exactly where to change wiring or add the local timestamp instrumentation.