Module agents.metrics.metrics_schema

Classes

class BaseComponentMetrics (id: str = '', provider_class: str = '', model_name: str = '')

Expand source code

@dataclass
class BaseComponentMetrics:
    """Base for component-specific metrics with provider and model identity."""
    id: str = ""
    provider_class: str = ""
    model_name: str = ""

Base for component-specific metrics with provider and model identity.

Subclasses

• EouMetrics
• KbMetrics
• LlmMetrics
• RealtimeMetrics
• SttMetrics
• TtsMetrics
• VadMetrics

Instance variables

var id : str

var model_name : str

var provider_class : str

class EouMetrics (id: str = '',
provider_class: str = '',
model_name: str = '',
eou_latency: float | None = None,
eou_start_time: float | None = None,
eou_end_time: float | None = None,
eou_probability: float | None = None,
min_speech_wait_timeout: float | None = None,
max_speech_wait_timeout: float | None = None,
eou_avg_delay: float | None = None,
waited_for_additional_speech: bool = False,
eou_wait_ms: float | None = None)

Expand source code

@dataclass
class EouMetrics(BaseComponentMetrics):
    """EOU (End of Utterance) metrics."""
    eou_latency: Optional[float] = None
    eou_start_time: Optional[float] = None
    eou_end_time: Optional[float] = None
    eou_probability: Optional[float] = None
    min_speech_wait_timeout: Optional[float] = None
    max_speech_wait_timeout: Optional[float] = None
    eou_avg_delay: Optional[float] = None
    waited_for_additional_speech: bool = False
    eou_wait_ms: Optional[float] = None

EOU (End of Utterance) metrics.

Ancestors

• BaseComponentMetrics

Instance variables

var eou_avg_delay : float | None

var eou_end_time : float | None

var eou_latency : float | None

var eou_probability : float | None

var eou_start_time : float | None

var eou_wait_ms : float | None

var max_speech_wait_timeout : float | None

var min_speech_wait_timeout : float | None

var waited_for_additional_speech : bool

class FallbackEvent (component_type: str = '',
temporary_disable_sec: float = 0.0,
permanent_disable_after_attempts: int = 0,
recovery_attempt: int = 0,
message: str = '',
start_time: float = <factory>,
end_time: float | None = None,
duration_ms: float | None = None,
original_provider_label: str | None = None,
original_connection_start: float | None = None,
original_connection_end: float | None = None,
original_connection_duration_ms: float | None = None,
new_provider_label: str | None = None,
new_connection_start: float | None = None,
new_connection_end: float | None = None,
new_connection_duration_ms: float | None = None,
is_recovery: bool = False)

Expand source code

@dataclass
class FallbackEvent:
    """Fallback event when a provider fails and switches to backup."""
    component_type: str = ""
    temporary_disable_sec: float = 0.0
    permanent_disable_after_attempts: int = 0
    recovery_attempt: int = 0
    message: str = ""
    start_time: float = field(default_factory=time.perf_counter)
    end_time: Optional[float] = None
    duration_ms: Optional[float] = None
    original_provider_label: Optional[str] = None
    original_connection_start: Optional[float] = None
    original_connection_end: Optional[float] = None
    original_connection_duration_ms: Optional[float] = None
    new_provider_label: Optional[str] = None
    new_connection_start: Optional[float] = None
    new_connection_end: Optional[float] = None
    new_connection_duration_ms: Optional[float] = None
    is_recovery: bool = False

Fallback event when a provider fails and switches to backup.

Instance variables

var component_type : str

var duration_ms : float | None

var end_time : float | None

var is_recovery : bool

var message : str

var new_connection_duration_ms : float | None

var new_connection_end : float | None

var new_connection_start : float | None

var new_provider_label : str | None

var original_connection_duration_ms : float | None

var original_connection_end : float | None

var original_connection_start : float | None

var original_provider_label : str | None

var permanent_disable_after_attempts : int

var recovery_attempt : int

var start_time : float

var temporary_disable_sec : float

class FunctionToolMetrics (tool_name: str = '',
tool_params: Dict[str, Any] | List[Any] = <factory>,
tool_response: Dict[str, Any] = <factory>,
start_time: float | None = None,
end_time: float | None = None,
latency: float | None = None)

Expand source code

@dataclass
class FunctionToolMetrics:
    """Metrics for a single function (agent) tool call in a turn."""
    tool_name: str = ""
    tool_params: Union[Dict[str, Any], List[Any]] = field(default_factory=dict)
    tool_response: Dict[str, Any] = field(default_factory=dict)
    start_time: Optional[float] = None
    end_time: Optional[float] = None
    latency: Optional[float] = None

Metrics for a single function (agent) tool call in a turn.

Instance variables

var end_time : float | None

var latency : float | None

var start_time : float | None

var tool_name : str

var tool_params : Dict[str, Any] | List[Any]

var tool_response : Dict[str, Any]

class InterruptionMetrics (false_interrupt_pause_duration: float | None = None,
resume_on_false_interrupt: bool | None = None,
interrupt_words: int | None = None,
interrupt_duration: float | None = None,
interrupt_reason: List[str] = <factory>,
interrupt_start_time: float | None = None,
interrupt_end_time: float | None = None,
is_false_interrupt: bool = False,
false_interrupt_duration: float | None = None,
false_interrupt_words: int | None = None,
false_interrupt_start_time: float | None = None,
false_interrupt_end_time: float | None = None,
resumed_after_false_interrupt: bool = False,
interrupt_mode: Literal['VAD_ONLY', 'STT_ONLY', 'HYBRID'] | None = None,
interrupt_min_duration: float | None = None,
interrupt_min_words: int | None = None)

Expand source code

@dataclass
class InterruptionMetrics:
    """Interruption and false-interrupt metrics for a turn."""
    false_interrupt_pause_duration: Optional[float] = None
    resume_on_false_interrupt: Optional[bool] = None
    interrupt_words: Optional[int] = None
    interrupt_duration: Optional[float] = None
    interrupt_reason: List[str] = field(default_factory=list)
    interrupt_start_time: Optional[float] = None
    interrupt_end_time: Optional[float] = None
    is_false_interrupt: bool = False
    false_interrupt_duration: Optional[float] = None
    false_interrupt_words: Optional[int] = None
    false_interrupt_start_time: Optional[float] = None
    false_interrupt_end_time: Optional[float] = None
    resumed_after_false_interrupt: bool = False
    interrupt_mode: Optional[Literal["VAD_ONLY", "STT_ONLY", "HYBRID"]] = None
    interrupt_min_duration: Optional[float] = None
    interrupt_min_words: Optional[int] = None

Interruption and false-interrupt metrics for a turn.

Instance variables

var false_interrupt_duration : float | None

var false_interrupt_end_time : float | None

var false_interrupt_pause_duration : float | None

var false_interrupt_start_time : float | None

var false_interrupt_words : int | None

var interrupt_duration : float | None

var interrupt_end_time : float | None

var interrupt_min_duration : float | None

var interrupt_min_words : int | None

var interrupt_mode : Literal['VAD_ONLY', 'STT_ONLY', 'HYBRID'] | None

var interrupt_reason : List[str]

var interrupt_start_time : float | None

var interrupt_words : int | None

var is_false_interrupt : bool

var resume_on_false_interrupt : bool | None

var resumed_after_false_interrupt : bool

class KbMetrics (id: str = '',
provider_class: str = '',
model_name: str = '',
kb_id: str | None = None,
kb_documents: List[str] | None = None,
kb_scores: List[float] | None = None,
kb_retrieval_latency: float | None = None,
kb_start_time: float | None = None,
kb_end_time: float | None = None)

Expand source code

@dataclass
class KbMetrics(BaseComponentMetrics):
    """Knowledge base retrieval metrics."""
    kb_id: Optional[str] = None
    kb_documents: Optional[List[str]] = None
    kb_scores: Optional[List[float]] = None
    kb_retrieval_latency: Optional[float] = None
    kb_start_time: Optional[float] = None
    kb_end_time: Optional[float] = None

Knowledge base retrieval metrics.

Ancestors

• BaseComponentMetrics

Instance variables

var kb_documents : List[str] | None

var kb_end_time : float | None

var kb_id : str | None

var kb_retrieval_latency : float | None

var kb_scores : List[float] | None

var kb_start_time : float | None

class LlmMetrics (id: str = '',
provider_class: str = '',
model_name: str = '',
llm_config: Dict[str, Any] | None = None,
llm_input: str | None = None,
llm_latency: float | None = None,
llm_start_time: float | None = None,
llm_end_time: float | None = None,
llm_duration: float | None = None,
llm_first_token_time: float | None = None,
llm_ttft: float | None = None,
prompt_tokens: int | None = None,
completion_tokens: int | None = None,
total_tokens: int | None = None,
prompt_cached_tokens: int | None = None,
tokens_per_second: float | None = None)

Expand source code

@dataclass
class LlmMetrics(BaseComponentMetrics):
    """LLM metrics."""
    llm_config: Optional[Dict[str, Any]] = None
    llm_input: Optional[str] = None
    llm_latency: Optional[float] = None
    llm_start_time: Optional[float] = None
    llm_end_time: Optional[float] = None
    llm_duration: Optional[float] = None
    llm_first_token_time: Optional[float] = None
    llm_ttft: Optional[float] = None
    prompt_tokens: Optional[int] = None
    completion_tokens: Optional[int] = None
    total_tokens: Optional[int] = None
    prompt_cached_tokens: Optional[int] = None
    tokens_per_second: Optional[float] = None

LLM metrics.

Ancestors

• BaseComponentMetrics

Instance variables

var completion_tokens : int | None

var llm_config : Dict[str, Any] | None

var llm_duration : float | None

var llm_end_time : float | None

var llm_first_token_time : float | None

var llm_input : str | None

var llm_latency : float | None

var llm_start_time : float | None

var llm_ttft : float | None

var prompt_cached_tokens : int | None

var prompt_tokens : int | None

var tokens_per_second : float | None

var total_tokens : int | None

class McpToolMetrics (type: Literal['local', 'http'] = 'local',
tool_url: str | None = None,
tool_params: Dict[str, Any] | List[Any] = <factory>,
tool_response: Dict[str, Any] = <factory>,
start_time: float | None = None,
end_time: float | None = None,
latency: float | None = None)

Expand source code

@dataclass
class McpToolMetrics:
    """Metrics for a single MCP tool call in a turn."""
    type: Literal["local", "http"] = "local"
    tool_url: Optional[str] = None  # endpoint URL for http, or endpoint identifier
    tool_params: Union[Dict[str, Any], List[Any]] = field(default_factory=dict)
    tool_response: Dict[str, Any] = field(default_factory=dict)
    start_time: Optional[float] = None
    end_time: Optional[float] = None
    latency: Optional[float] = None

Metrics for a single MCP tool call in a turn.

Instance variables

var end_time : float | None

var latency : float | None

var start_time : float | None

var tool_params : Dict[str, Any] | List[Any]

var tool_response : Dict[str, Any]

var tool_url : str | None

var type : Literal['local', 'http']

class ParticipantMetrics (participant_id: str | None = None,
kind: Literal['agent', 'user'] | None = None,
sip_user: bool | None = None,
join_time: float | None = None,
meta: Dict[str, Any] | None = None)

Expand source code

@dataclass
class ParticipantMetrics:
    """Participant/connection metrics at session level."""
    participant_id: Optional[str] = None
    kind: Optional[Literal["agent", "user"]] = None
    sip_user: Optional[bool] = None
    join_time: Optional[float] = None
    meta: Optional[Dict[str, Any]] = None

Participant/connection metrics at session level.

Instance variables

var join_time : float | None

var kind : Literal['agent', 'user'] | None

var meta : Dict[str, Any] | None

var participant_id : str | None

var sip_user : bool | None

class RealtimeMetrics (id: str = '',
provider_class: str = '',
model_name: str = '',
realtime_config: Dict[str, Any] | None = None,
realtime_ttfb: float | None = None,
realtime_input_tokens: int | None = None,
realtime_total_tokens: int | None = None,
realtime_output_tokens: int | None = None,
realtime_input_text_tokens: int | None = None,
realtime_input_audio_tokens: int | None = None,
realtime_input_image_tokens: int | None = None,
realtime_input_cached_tokens: int | None = None,
realtime_thoughts_tokens: int | None = None,
realtime_cached_text_tokens: int | None = None,
realtime_cached_audio_tokens: int | None = None,
realtime_cached_image_tokens: int | None = None,
realtime_output_text_tokens: int | None = None,
realtime_output_audio_tokens: int | None = None,
realtime_output_image_tokens: int | None = None)

Expand source code

@dataclass
class RealtimeMetrics(BaseComponentMetrics):
    """Realtime model (full s2s) metrics."""
    realtime_config: Optional[Dict[str, Any]] = None
    realtime_ttfb: Optional[float] = None
    realtime_input_tokens: Optional[int] = None
    realtime_total_tokens: Optional[int] = None
    realtime_output_tokens: Optional[int] = None
    realtime_input_text_tokens: Optional[int] = None
    realtime_input_audio_tokens: Optional[int] = None
    realtime_input_image_tokens: Optional[int] = None
    realtime_input_cached_tokens: Optional[int] = None
    realtime_thoughts_tokens: Optional[int] = None
    realtime_cached_text_tokens: Optional[int] = None
    realtime_cached_audio_tokens: Optional[int] = None
    realtime_cached_image_tokens: Optional[int] = None
    realtime_output_text_tokens: Optional[int] = None
    realtime_output_audio_tokens: Optional[int] = None
    realtime_output_image_tokens: Optional[int] = None

Realtime model (full s2s) metrics.

Ancestors

• BaseComponentMetrics

Instance variables

var realtime_cached_audio_tokens : int | None

var realtime_cached_image_tokens : int | None

var realtime_cached_text_tokens : int | None

var realtime_config : Dict[str, Any] | None

var realtime_input_audio_tokens : int | None

var realtime_input_cached_tokens : int | None

var realtime_input_image_tokens : int | None

var realtime_input_text_tokens : int | None

var realtime_input_tokens : int | None

var realtime_output_audio_tokens : int | None

var realtime_output_image_tokens : int | None

var realtime_output_text_tokens : int | None

var realtime_output_tokens : int | None

var realtime_thoughts_tokens : int | None

var realtime_total_tokens : int | None

var realtime_ttfb : float | None

class SessionMetrics (session_id: str | None = None,
room_id: str | None = None,
system_instruction: str = '',
components: List[str] = <factory>,
pipeline_type: str | None = None,
pipeline_mode: str | None = None,
realtime_mode: str | None = None,
session_start_time: float = <factory>,
session_end_time: float | None = None,
participant_metrics: List[ParticipantMetrics] = <factory>,
errors: List[Dict[str, Any]] = <factory>,
provider_per_component: Dict[str, Dict[str, str]] = <factory>,
eou_config: Dict[str, Any] | None = None,
interrupt_config: Dict[str, Any] | None = None)

Expand source code

@dataclass
class SessionMetrics:
    """Session-level metrics."""
    session_id: Optional[str] = None
    room_id: Optional[str] = None
    system_instruction: str = ""
    components: List[str] = field(default_factory=list)
    pipeline_type: Optional[str] = None
    pipeline_mode: Optional[str] = None
    realtime_mode: Optional[str] = None
    session_start_time: float = field(default_factory=time.time)
    session_end_time: Optional[float] = None
    participant_metrics: List[ParticipantMetrics] = field(default_factory=list)
    errors: List[Dict[str, Any]] = field(default_factory=list)
    provider_per_component: Dict[str, Dict[str, str]] = field(default_factory=dict)
    eou_config: Optional[Dict[str, Any]] = None
    interrupt_config: Optional[Dict[str, Any]] = None

Session-level metrics.

Instance variables

var components : List[str]

var eou_config : Dict[str, Any] | None

var errors : List[Dict[str, Any]]

var interrupt_config : Dict[str, Any] | None

var participant_metrics : List[ParticipantMetrics]

var pipeline_mode : str | None

var pipeline_type : str | None

var provider_per_component : Dict[str, Dict[str, str]]

var realtime_mode : str | None

var room_id : str | None

var session_end_time : float | None

var session_id : str | None

var session_start_time : float

var system_instruction : str

class SttMetrics (id: str = '',
provider_class: str = '',
model_name: str = '',
stt_config: Dict[str, Any] | None = None,
stt_confidence: float | None = None,
stt_input_tokens: int | None = None,
stt_output_tokens: int | None = None,
stt_total_tokens: int | None = None,
stt_latency: float | None = None,
stt_start_time: float | None = None,
stt_end_time: float | None = None,
stt_duration: float | None = None,
stt_preflight_end_time: float | None = None,
stt_preflight_latency: float | None = None,
stt_interim_end_time: float | None = None,
stt_interim_latency: float | None = None,
stt_ttfw: float | None = None,
stt_preemptive_generation_occurred: bool = False,
stt_preemptive_generation_enabled: bool = False,
stt_transcript: str | None = None,
stt_preflight_transcript: str | None = None)

Expand source code

@dataclass
class SttMetrics(BaseComponentMetrics):
    """STT (Speech-to-Text) metrics."""
    stt_config: Optional[Dict[str, Any]] = None
    stt_confidence: Optional[float] = None
    stt_input_tokens: Optional[int] = None
    stt_output_tokens: Optional[int] = None
    stt_total_tokens: Optional[int] = None
    stt_latency: Optional[float] = None
    stt_start_time: Optional[float] = None
    stt_end_time: Optional[float] = None
    stt_duration: Optional[float] = None
    stt_preflight_end_time: Optional[float] = None
    stt_preflight_latency: Optional[float] = None
    stt_interim_end_time: Optional[float] = None
    stt_interim_latency: Optional[float] = None
    stt_ttfw: Optional[float] = None
    stt_preemptive_generation_occurred: bool = False
    stt_preemptive_generation_enabled: bool = False
    stt_transcript: Optional[str] = None
    stt_preflight_transcript: Optional[str] = None

STT (Speech-to-Text) metrics.

Ancestors

• BaseComponentMetrics

Instance variables

var stt_confidence : float | None

var stt_config : Dict[str, Any] | None

var stt_duration : float | None

var stt_end_time : float | None

var stt_input_tokens : int | None

var stt_interim_end_time : float | None

var stt_interim_latency : float | None

var stt_latency : float | None

var stt_output_tokens : int | None

var stt_preemptive_generation_enabled : bool

var stt_preemptive_generation_occurred : bool

var stt_preflight_end_time : float | None

var stt_preflight_latency : float | None

var stt_preflight_transcript : str | None

var stt_start_time : float | None

var stt_total_tokens : int | None

var stt_transcript : str | None

var stt_ttfw : float | None

class TimelineEvent (event_type: str = '',
start_time: float = 0.0,
end_time: float | None = None,
duration_ms: float | None = None,
text: str = '')

Expand source code

@dataclass
class TimelineEvent:
    """Data structure for a single timeline event."""
    event_type: str = ""
    start_time: float = 0.0
    end_time: Optional[float] = None
    duration_ms: Optional[float] = None
    text: str = ""

Data structure for a single timeline event.

Instance variables

var duration_ms : float | None

var end_time : float | None

var event_type : str

var start_time : float

var text : str

class TtsMetrics (id: str = '',
provider_class: str = '',
model_name: str = '',
tts_config: Dict[str, Any] | None = None,
tts_latency: float | None = None,
tts_start_time: float | None = None,
tts_end_time: float | None = None,
tts_duration: float | None = None,
tts_characters: int | None = 0,
tts_first_byte_time: float | None = None,
ttfb: float | None = None)

Expand source code

@dataclass
class TtsMetrics(BaseComponentMetrics):
    """TTS metrics."""
    tts_config: Optional[Dict[str, Any]] = None
    tts_latency: Optional[float] = None
    tts_start_time: Optional[float] = None
    tts_end_time: Optional[float] = None
    tts_duration: Optional[float] = None
    tts_characters: Optional[int] = 0
    tts_first_byte_time: Optional[float] = None
    ttfb: Optional[float] = None

TTS metrics.

Ancestors

• BaseComponentMetrics

Instance variables

var ttfb : float | None

var tts_characters : int | None

var tts_config : Dict[str, Any] | None

var tts_duration : float | None

var tts_end_time : float | None

var tts_first_byte_time : float | None

var tts_latency : float | None

var tts_start_time : float | None

class TurnMetrics (turn_id: str = '',
is_interrupted: bool = False,
user_speech_start_time: float | None = None,
user_speech_end_time: float | None = None,
user_speech_duration: float | None = None,
user_speech: str | None = None,
agent_speech_start_time: float | None = None,
agent_speech_end_time: float | None = None,
agent_speech_duration: float | None = None,
agent_speech: str | None = None,
e2e_latency: float | None = None,
function_tools_called: List[str] = <factory>,
function_tool_timestamps: List[Dict[str, Any]] = <factory>,
is_a2a_enabled: bool = False,
handoff_occurred: bool = False,
errors: List[Dict[str, Any]] = <factory>,
timestamp: float = <factory>,
preemtive_generation_enabled: bool = False,
vad_metrics: List[VadMetrics] = <factory>,
stt_metrics: List[SttMetrics] = <factory>,
eou_metrics: List[EouMetrics] = <factory>,
kb_metrics: List[KbMetrics] = <factory>,
llm_metrics: List[LlmMetrics] = <factory>,
tts_metrics: List[TtsMetrics] = <factory>,
interruption_metrics: InterruptionMetrics | None = None,
realtime_metrics: List[RealtimeMetrics] = <factory>,
timeline_event_metrics: List[TimelineEvent] = <factory>,
fallback_events: List[FallbackEvent] = <factory>,
function_tool_metrics: List[FunctionToolMetrics] = <factory>,
mcp_tool_metrics: List[McpToolMetrics] = <factory>,
session_metrics: SessionMetrics | None = <factory>,
background_audio_file_path: str | None = None,
background_audio_looping: bool | None = None,
thinking_audio_file_path: str | None = None,
thinking_audio_looping: bool | None = None,
thinking_audio_override_thinking: bool | None = None)

Expand source code

@dataclass
class TurnMetrics:
    """Single turn with lists of per-component metrics."""
    turn_id: str = ""
    is_interrupted: bool = False
    user_speech_start_time: Optional[float] = None
    user_speech_end_time: Optional[float] = None
    user_speech_duration: Optional[float] = None
    user_speech: Optional[str] = None
    agent_speech_start_time: Optional[float] = None
    agent_speech_end_time: Optional[float] = None
    agent_speech_duration: Optional[float] = None
    agent_speech: Optional[str] = None
    e2e_latency: Optional[float] = None
    function_tools_called: List[str] = field(default_factory=list)
    function_tool_timestamps: List[Dict[str, Any]] = field(default_factory=list)
    is_a2a_enabled: bool = False
    handoff_occurred: bool = False
    errors: List[Dict[str, Any]] = field(default_factory=list)
    timestamp: float = field(default_factory=time.time)
    preemtive_generation_enabled: bool = False

    vad_metrics: List[VadMetrics] = field(default_factory=list)
    stt_metrics: List[SttMetrics] = field(default_factory=list)
    eou_metrics: List[EouMetrics] = field(default_factory=list)
    kb_metrics: List[KbMetrics] = field(default_factory=list)
    llm_metrics: List[LlmMetrics] = field(default_factory=list)
    tts_metrics: List[TtsMetrics] = field(default_factory=list)
    interruption_metrics: Optional[InterruptionMetrics] = None
    realtime_metrics: List[RealtimeMetrics] = field(default_factory=list)
    timeline_event_metrics: List[TimelineEvent] = field(default_factory=list)
    fallback_events: List[FallbackEvent] = field(default_factory=list)
    function_tool_metrics: List[FunctionToolMetrics] = field(default_factory=list)
    mcp_tool_metrics: List[McpToolMetrics] = field(default_factory=list)
    session_metrics: Optional[SessionMetrics] = field(default_factory=SessionMetrics)

    # Background audio
    background_audio_file_path: Optional[str] = None
    background_audio_looping: Optional[bool] = None

    # Thinking audio
    thinking_audio_file_path: Optional[str] = None
    thinking_audio_looping: Optional[bool] = None
    thinking_audio_override_thinking: Optional[bool] = None

    def compute_e2e_latency(self) -> None:
        """Calculate E2E latency by summing component latencies (STT + EOU + LLM TTFT + TTS)."""
        e2e_components = []

        if self.stt_metrics:
            stt = self.stt_metrics[-1]
            if stt.stt_latency is not None:
                e2e_components.append(stt.stt_latency)

        if self.eou_metrics:
            eou = self.eou_metrics[-1]
            if eou.eou_latency is not None:
                e2e_components.append(eou.eou_latency)
            if eou.eou_wait_ms is not None:
                e2e_components.append(eou.eou_wait_ms)

        if self.llm_metrics:
            llm = self.llm_metrics[-1]
            if llm.llm_ttft is not None:
                e2e_components.append(llm.llm_ttft)

        if self.tts_metrics:
            tts = self.tts_metrics[-1]
            if tts.tts_latency is not None:
                e2e_components.append(tts.tts_latency)

        if self.realtime_metrics:
            rt = self.realtime_metrics[-1]
            if rt.realtime_ttfb is not None:
                e2e_components.append(rt.realtime_ttfb)

        if e2e_components:
            self.e2e_latency = round(sum(e2e_components), 4)

    def to_dict(self) -> Dict[str, Any]:
        """Serialize turn metrics to a dictionary."""
        self.compute_e2e_latency()
        return asdict(self)

Single turn with lists of per-component metrics.

Instance variables

var agent_speech : str | None

var agent_speech_duration : float | None

var agent_speech_end_time : float | None

var agent_speech_start_time : float | None

var background_audio_file_path : str | None

var background_audio_looping : bool | None

var e2e_latency : float | None

var eou_metrics : List[EouMetrics]

var errors : List[Dict[str, Any]]

var fallback_events : List[FallbackEvent]

var function_tool_metrics : List[FunctionToolMetrics]

var function_tool_timestamps : List[Dict[str, Any]]

var function_tools_called : List[str]

var handoff_occurred : bool

var interruption_metrics : InterruptionMetrics | None

var is_a2a_enabled : bool

var is_interrupted : bool

var kb_metrics : List[KbMetrics]

var llm_metrics : List[LlmMetrics]

var mcp_tool_metrics : List[McpToolMetrics]

var preemtive_generation_enabled : bool

var realtime_metrics : List[RealtimeMetrics]

var session_metrics : SessionMetrics | None

var stt_metrics : List[SttMetrics]

var thinking_audio_file_path : str | None

var thinking_audio_looping : bool | None

var thinking_audio_override_thinking : bool | None

var timeline_event_metrics : List[TimelineEvent]

var timestamp : float

var tts_metrics : List[TtsMetrics]

var turn_id : str

var user_speech : str | None

var user_speech_duration : float | None

var user_speech_end_time : float | None

var user_speech_start_time : float | None

var vad_metrics : List[VadMetrics]

Methods

def compute_e2e_latency(self) ‑> None

Expand source code

def compute_e2e_latency(self) -> None:
    """Calculate E2E latency by summing component latencies (STT + EOU + LLM TTFT + TTS)."""
    e2e_components = []

    if self.stt_metrics:
        stt = self.stt_metrics[-1]
        if stt.stt_latency is not None:
            e2e_components.append(stt.stt_latency)

    if self.eou_metrics:
        eou = self.eou_metrics[-1]
        if eou.eou_latency is not None:
            e2e_components.append(eou.eou_latency)
        if eou.eou_wait_ms is not None:
            e2e_components.append(eou.eou_wait_ms)

    if self.llm_metrics:
        llm = self.llm_metrics[-1]
        if llm.llm_ttft is not None:
            e2e_components.append(llm.llm_ttft)

    if self.tts_metrics:
        tts = self.tts_metrics[-1]
        if tts.tts_latency is not None:
            e2e_components.append(tts.tts_latency)

    if self.realtime_metrics:
        rt = self.realtime_metrics[-1]
        if rt.realtime_ttfb is not None:
            e2e_components.append(rt.realtime_ttfb)

    if e2e_components:
        self.e2e_latency = round(sum(e2e_components), 4)

Calculate E2E latency by summing component latencies (STT + EOU + LLM TTFT + TTS).

def to_dict(self) ‑> Dict[str, Any]

Expand source code

def to_dict(self) -> Dict[str, Any]:
    """Serialize turn metrics to a dictionary."""
    self.compute_e2e_latency()
    return asdict(self)

Serialize turn metrics to a dictionary.

class VadMetrics (id: str = '',
provider_class: str = '',
model_name: str = '',
vad_config: Dict[str, Any] | None = None,
user_speech_start_time: float | None = None,
user_speech_end_time: float | None = None,
vad_min_silence_duration: float | None = None,
vad_min_speech_duration: float | None = None,
vad_threshold: float | None = None,
vad_end_of_speech_time: float | None = None)

Expand source code

@dataclass
class VadMetrics(BaseComponentMetrics):
    """VAD (Voice Activity Detection) metrics."""
    vad_config: Optional[Dict[str, Any]] = None
    user_speech_start_time: Optional[float] = None
    user_speech_end_time: Optional[float] = None
    vad_min_silence_duration: Optional[float] = None
    vad_min_speech_duration: Optional[float] = None
    vad_threshold: Optional[float] = None
    vad_end_of_speech_time: Optional[float] = None

VAD (Voice Activity Detection) metrics.

Ancestors

• BaseComponentMetrics

Instance variables

var user_speech_end_time : float | None

var user_speech_start_time : float | None

var vad_config : Dict[str, Any] | None

var vad_end_of_speech_time : float | None

var vad_min_silence_duration : float | None

var vad_min_speech_duration : float | None

var vad_threshold : float | None