# def llm_wrapper(metrics: LLMMetrics): # asyncio.create_task(on_llm_metrics(metrics)) # # async def on_llm_metrics(metrics: LLMMetrics): # logger.info(f"Received metrics llm: {metrics}") # # session.llm.on("metrics_collected", llm_wrapper) # # def tts_wrapper(metrics: TTSMetrics): # asyncio.create_task(on_tts_metrics(metrics)) # # async def on_tts_metrics(metrics: TTSMetrics): # logger.info(f"Received metrics tts: {metrics}") # # session.tts.on("metrics_collected", tts_wrapper) # def stt_wrapper(metrics: STTMetrics): # asyncio.create_task(on_stt_metrics(metrics)) # # async def on_stt_metrics(metrics: STTMetrics): # logger.info(f"Received metrics stt: {metrics}") I can use per plugin instance to subscribe over the metrics_collected event, as sessiion level metrics collected event is deprecated. but how will I track eou? there is no plugin instance to track eou

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

what will be the correct way to write this per plugin metrics without repeaing code

Thanks — you're looking for a DRY way to subscribe per‑plugin to

metrics_collected

without repeating code. I can give a concise reusable pattern, but first: which language/SDK and version are you using (Python asyncio, Node, or other)? Also which plugin objects do you need to subscribe (LLM, TTS, STT, others)? , with interactive elements

py

Thanks — you're using Python. Below is a concise, reusable pattern to subscribe multiple plugin instances (

llm

,

tts

,

stt

, etc.) to

metrics_collected

without repeating code. It also shows how to filter by metric type and delegate to async handlers.

import asyncio
from typing import Callable, Iterable

# Generic wrapper factory: creates a sync callback that schedules the async handler
def make_metrics_callback(async_handler: Callable):
    def callback(metrics):
        asyncio.create_task(async_handler(metrics))
    return callback

# Example async handlers for different metric types
async def handle_llm(metrics):
    # process LLM metrics
    logger.info(f"LLM metrics: {metrics}")

async def handle_tts(metrics):
    # process TTS metrics
    logger.info(f"TTS metrics: {metrics}")

async def handle_stt(metrics):
    # process STT metrics
    logger.info(f"STT metrics: {metrics}")

# Register multiple plugins with minimal repetition
def subscribe_metrics(plugins: Iterable, handler_map: dict):
    """
    plugins: iterable of (name, plugin_instance) tuples or plugin instances where you can infer type
    handler_map: mapping from plugin name (or type) -> async handler
    """
    for key, plugin in plugins:
        handler = handler_map.get(key)
        if handler:
            plugin.on("metrics_collected", make_metrics_callback(handler))

# Usage example (adapt keys to how you identify plugins)
plugins = [
    ("llm", session.llm),
    ("tts", session.tts),
    ("stt", session.stt),
]

handler_map = {
    "llm": handle_llm,
    "tts": handle_tts,
    "stt": handle_stt,
}

subscribe_metrics(plugins, handler_map)

Notes and tips: • Use the plugin key you prefer (string name, class name, or other identifier). Adjust

plugins

and

handler_map

accordingly. • If multiple plugins share identical handling, point them to the same async handler to avoid duplication. • For cross‑cutting logic (logging, metrics forwarding), wrap handlers or create a small middleware function to run shared behavior before/after per-plugin handling. If you want, I can adapt this to: • automatically infer plugin keys from plugin types, • include filtering by metric subclass (e.g., only handle

InterruptionMetrics

), or • show an example with structured logging/export to your telemetry sink. Which would you prefer?, with interactive elements

llmmetrics(ttft) is being tracked using session.llm.on("metircs_collected") event which has speech id ttsmetrics(ttfb) is being tracked using session.tts.on("metircs_collected") event which has speech id earlier we tracked eou using metrics_collected event which had speechid so the based on common speech id , these metrics were mapped and were used to calculate cumulative latency (ttfb + ttft + eou_delay) but now eou_delay is being tranck using on_conversation_item_added event's chatMessage.metrics from user's role which has no speech_id to map how will I calculate the cumulative latency of that turn using per plugin ttft, ttfb and eou from new event? how to map it Had the same problem and found this issue. My solution is to maintain a map from

ConversationItemAddedEvent.item.id

to speech_id (from

ev.metrics

)

Thanks — you're asking how to compute cumulative turn latency when LLM/TTS plugin metrics include

speech_id

but EOU now arrives in

conversation_item_added

without

speech_id

. Good question. Short answer: use the per-turn metrics on the conversation item (especially

e2e_latency

) or correlate events by timestamps / conversation item id instead of

speech_id

. Below are two reliable approaches and a concise Python pattern. Recommended (simple & accurate) • Read

e2e_latency

and component fields (

llm_node_ttft

,

tts_node_ttfb

, etc.) from

ev.item.metrics

in

conversation_item_added

.

e2e_latency

already represents end‑to‑end latency for that assistant turn so you don’t need to stitch `speech_id`s. • If you want the breakdown,

ev.item.metrics

includes component metrics so you can compute cumulative =

llm_node_ttft + tts_node_ttfb + eou_delay

(or just trust

e2e_latency

). If you need to correlate separate plugin

metrics_collected

events with the conversation item • Correlate by timestamps: capture

metrics_collected

timestamps (e.g., metric.created_at or time.time()) and match the closest plugin metrics to the conversation item timestamp window. • Correlate by conversation item id (if present): use

ev.item.id

as an anchor if your plugin metrics include or can be mapped to that id. • Use sequence/order heuristics only if traffic is low/serial (risky under concurrency). Minimal Python pattern (timestamp correlation + fallback to e2e)

import time
from collections import defaultdict, deque

# small in-memory buffer keyed by speech_id or plugin type
plugin_buffer = defaultdict(deque)  # {plugin: deque([(ts, metrics), ...])}

def plugin_callback(plugin_name):
    def cb(metrics):
        plugin_buffer[plugin_name].append((time.time(), metrics))
    return cb

# subscribe: session.llm.on("metrics_collected", plugin_callback("llm")) etc.

@session.on("conversation_item_added")
def on_item(ev):
    if ev.item.role != "assistant":
        return
    metrics = ev.item.metrics or {}
    # preferred: use e2e_latency if present
    e2e = metrics.get("e2e_latency")
    if e2e is not None:
        logger.info(f"turn e2e_latency={e2e}")
        # optional: read breakdown fields if you want components
        return

    # fallback: approximate by timestamp matching
    item_ts = ev.item.created_at_epoch or time.time()
    def nearest(plugin):
        candidates = plugin_buffer.get(plugin)
        if not candidates:
            return None
        # pop old entries
        while candidates and item_ts - candidates[0][0] > 10.0:
            candidates.popleft()
        # find closest (simple: rightmost <= item_ts)
        for ts, m in reversed(candidates):
            if ts <= item_ts:
                return m
        return candidates[-1][1] if candidates else None

    llm_m = nearest("llm")
    tts_m = nearest("tts")
    # compute approximate cumulative if available
    if llm_m and tts_m:
        approx = llm_m.get("ttft", 0) + tts_m.get("ttfb", 0) + metrics.get("eou_delay", 0)
        logger.info(f"approx cumulative={approx}")

Notes and tradeoffs • Using

e2e_latency

is the most robust and simplest approach. • Timestamp correlation works but can mis-map under concurrent/parallel turns; tune time windows and clean buffers. • Conversation item id correlation is ideal if you can surface that id in plugin metrics. If you want, tell me: • a sample payload for your

metrics_collected

and

conversation_item_added

events and the SDK version — I’ll adapt the code to your exact field names and time formats. Sources: How to capture real-time agent transcriptions during conversations | Build Your First AI Voice Agent in Python: Complete Tutorial | LiveKit | Build Your First AI Voice Agent in Python: Complete Tutorial | LiveKit, with interactive elements