I have a simple voice agent with Hedra integration. I want to measure end-to-end latency in responses. My current solution is to create a latency tracker and monitor state change of agent with

@session.on("agent_state_changed")
    def on_agent_state_changed(ev: AgentStateChangedEvent):
        if (
            latency_tracker
            and ev.new_state == "speaking"
            and latency_tracker.speech_end_time
        ):
            latency_tracker.mark_response_start()
            latency_tracker.log_latency(logger=json_logger)

however this does not measure the delay introduced by Hedra. How to fix this?

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

speech_created

is emitted when EOU confirmed?

Short answer: No —

speech_created

is emitted when the agent has generated speech (LLM → TTS finished / speech output created), not when EOU is confirmed. Details relevant to your latency measurement: • EOU confirmation (events like

input_speech_stopped

or

input_audio_transcription_completed

, or endpointing configured by `min_silence_duration`/`min_endpointing_delay`) happens earlier and triggers the agent to start generating a response. •

speech_created

fires after the agent produces the speech output (so it already includes LLM + TTS time). • To include Hedra's rendering/streaming and actual playback in your end-to-end measurement, start timing at EOU confirmation (or at `input_speech_stopped`/transcription-complete) and stop timing when the avatar playout actually finishes via `SpeechHandle`'s ended callback. If you want, tell me which EOU event you see in your session (e.g.,

input_speech_stopped

or transcription-complete) and how your

latency_tracker

API looks (`mark_response_start`/`mark_response_end` names), and I’ll give a concrete code snippet that starts at EOU and stops on playout end. Sources: Events and error handling | LiveKit docs | examples/voice_agents | livekit/livekit_composite | DeepWiki | How does end-of-utterance detection work in conversations? | Agent speech and audio | LiveKit docs

I want to measure latency introduced by Hedra live avatar. Give me a working example in Python

Below is a focused, working Python example that measures only the latency introduced by the Hedra live avatar: start timing when user audio endpointing / EOU is detected (

input_speech_stopped

) and stop timing when the Hedra avatar finishes playout via the

SpeechHandle

ended callback. It assumes you have an active

session

(AgentSession/RealtimeSession) and a

latency_tracker

with

mark_start()

/

mark_end()

/

log_latency(logger=...)

methods — adapt method names if yours differ.

import time
import asyncio
import logging

json_logger = logging.getLogger("latency")
json_logger.setLevel(<http://logging.INFO|logging.INFO>)

class SimpleLatencyTracker:
    def __init__(self):
        self.start_time = None
        self.end_time = None

    def mark_start(self):
        self.start_time = time.time()

    def mark_end(self):
        self.end_time = time.time()

    def log_latency(self, logger=json_logger):
        if self.start_time and self.end_time:
            latency_ms = (self.end_time - self.start_time) * 1000.0
            <http://logger.info|logger.info>({"hedra_avatar_latency_ms": latency_ms})
        else:
            <http://logger.info|logger.info>({"error": "start or end time missing"})

latency_tracker = SimpleLatencyTracker()

# 1) Start timing when EOU / input stops (endpointing)
@session.on("input_speech_stopped")
def on_input_speech_stopped(ev):
    # mark start of Hedra-related latency: time between EOU and actual Hedra playout completion
    latency_tracker.mark_start()

# 2) If you detect endpointing via transcription-complete instead, use:
# @session.on("input_audio_transcription_completed")
# def on_transcription_completed(ev):
#     latency_tracker.mark_start()

# 3) Stop timing when avatar playout actually finishes via SpeechHandle
@session.on("speech_created")
def on_speech_created(ev):
    # Only measure Hedra portion for speeches that will be played by Hedra avatar
    if not ev.speech_handle:
        return

    # Define callback to run when playout ends
    async def on_playout_finished():
        # mark end and log
        latency_tracker.mark_end()
        latency_tracker.log_latency(logger=json_logger)

    # If SpeechHandle supports adding ended callbacks
    ev.speech_handle.add_ended_callback(on_playout_finished)

    # If SpeechHandle requires async registration or awaiting, adapt:
    # asyncio.create_task(register_and_wait(ev.speech_handle))

# Optional: If SpeechHandle instead exposes a future/promise named `ended`
# you could do:
# if hasattr(ev.speech_handle, "ended"):
#     async def wait_for_end():
#         await ev.speech_handle.ended
#         latency_tracker.mark_end()
#         latency_tracker.log_latency(logger=json_logger)
#     asyncio.create_task(wait_for_end())

Notes: • This measures the time from EOU detection to avatar playout completion — i.e., Hedra TTS/video rendering, streaming, and playback delay. • If your

SpeechHandle

API uses different callback names (e.g.,

add_ended_callback

vs an

ended

awaitable), replace that part with the correct mechanism. • Tell me your

latency_tracker

method names or the exact

SpeechHandle

API if you want the snippet adapted precisely to your code.

input_speech_stopped

does not exist

You're right — use user state changes and playback completion. Below is a concise, working Python example that measures only the Hedra live-avatar latency: it starts at end-of-user-speech (when the user state transitions from "speaking" → "listening") and stops when avatar audio playback finishes (

playback_finished

). Adapt

LatencyTracker

and any event names if your code uses different method names.

import time
import asyncio
import logging

json_logger = logging.getLogger("latency")
json_logger.setLevel(<http://logging.INFO|logging.INFO>)

class LatencyTracker:
    def __init__(self):
        self.eou_time = None
        self.end_time = None

    def mark_eou(self):
        self.eou_time = time.time()

    def mark_end(self):
        self.end_time = time.time()

    def log(self):
        if self.eou_time and self.end_time:
            ms = (self.end_time - self.eou_time) * 1000.0
            <http://json_logger.info|json_logger.info>({"hedra_avatar_latency_ms": ms})
        else:
            <http://json_logger.info|json_logger.info>({"error": "missing timestamps"})

tracker = LatencyTracker()

# Detect end of user speech via user state change (speaking -> listening)
@session.on("user_state_changed")
def on_user_state_changed(ev):
    if ev.old_state == "speaking" and ev.new_state == "listening":
        tracker.mark_eou()

# When agent speech is created, register for playback finished via SpeechHandle or session output
@session.on("speech_created")
def on_speech_created(ev):
    # prefer SpeechHandle ended callback if available
    if ev.speech_handle:
        async def on_playout_finished():
            tracker.mark_end()
            tracker.log()
        ev.speech_handle.add_ended_callback(on_playout_finished)
        return

    # fallback: listen for room/playback finished event on session output
    @session.output.audio.on("playback_finished")
    def on_playback_finished(evt):
        tracker.mark_end()
        tracker.log()

Notes: • The

user_state_changed

(speaking→listening) is the correct EOU signal;

input_speech_stopped

does not exist. •

speech_created

+

SpeechHandle

ended callback is the most reliable way to detect actual avatar playout completion;

playback_finished

on

session.output.audio

is a usable fallback. • This measures Hedra-related delay: TTS/video rendering, network/streaming, and actual playback completion. Tell me if your

SpeechHandle

uses a different callback API or if you want the example adapted to a specific latency-tracker method names — I’ll update the snippet. Sources: Turn detection and interruptions | LiveKit Docs | Events and error handling | LiveKit docs | How to detect when an agent has finished speaking