Module livekit.agents.metrics

def log_metrics(metrics: AgentMetrics, *, logger: logging.Logger | None = None) -> None:
    if logger is None:
        logger = default_logger

    metadata: dict[str, str | float] = {}
    if metrics.metadata:
        metadata |= {
            "model_name": metrics.metadata.model_name or "unknown",
            "model_provider": metrics.metadata.model_provider or "unknown",
        }

    if isinstance(metrics, LLMMetrics):
        logger.info(
            "LLM metrics",
            extra=metadata
            | {
                "ttft": round(metrics.ttft, 2),
                "prompt_tokens": metrics.prompt_tokens,
                "prompt_cached_tokens": metrics.prompt_cached_tokens,
                "completion_tokens": metrics.completion_tokens,
                "tokens_per_second": round(metrics.tokens_per_second, 2),
            },
        )
    elif isinstance(metrics, RealtimeModelMetrics):
        logger.info(
            "RealtimeModel metrics",
            extra=metadata
            | {
                "ttft": round(metrics.ttft, 2),
                "input_tokens": metrics.input_tokens,
                "cached_input_tokens": metrics.input_token_details.cached_tokens,
                "input_text_tokens": metrics.input_token_details.text_tokens,
                "input_cached_text_tokens": metrics.input_token_details.cached_tokens_details.text_tokens
                if metrics.input_token_details.cached_tokens_details
                else 0,
                "input_image_tokens": metrics.input_token_details.image_tokens,
                "input_cached_image_tokens": metrics.input_token_details.cached_tokens_details.image_tokens
                if metrics.input_token_details.cached_tokens_details
                else 0,
                "input_audio_tokens": metrics.input_token_details.audio_tokens,
                "input_cached_audio_tokens": metrics.input_token_details.cached_tokens_details.audio_tokens
                if metrics.input_token_details.cached_tokens_details
                else 0,
                "output_tokens": metrics.output_tokens,
                "output_text_tokens": metrics.output_token_details.text_tokens,
                "output_audio_tokens": metrics.output_token_details.audio_tokens,
                "output_image_tokens": metrics.output_token_details.image_tokens,
                "total_tokens": metrics.total_tokens,
                "tokens_per_second": round(metrics.tokens_per_second, 2),
            },
        )
    elif isinstance(metrics, TTSMetrics):
        logger.info(
            "TTS metrics",
            extra=metadata
            | {
                "ttfb": metrics.ttfb,
                "audio_duration": round(metrics.audio_duration, 2),
            },
        )
    elif isinstance(metrics, EOUMetrics):
        logger.info(
            "EOU metrics",
            extra=metadata
            | {
                "end_of_utterance_delay": round(metrics.end_of_utterance_delay, 2),
                "transcription_delay": round(metrics.transcription_delay, 2),
            },
        )
    elif isinstance(metrics, STTMetrics):
        logger.info(
            "STT metrics",
            extra=metadata
            | {
                "audio_duration": round(metrics.audio_duration, 2),
            },
        )
    elif isinstance(metrics, InterruptionMetrics):
        logger.info(
            "Interruption metrics",
            extra=metadata
            | {
                "total_duration": round(metrics.total_duration, 2),
                "prediction_duration": round(metrics.prediction_duration, 2),
                "detection_delay": round(metrics.detection_delay, 2),
                "num_interruptions": metrics.num_interruptions,
                "num_backchannels": metrics.num_backchannels,
                "num_requests": metrics.num_requests,
            },
        )

@dataclass
class AgentSessionUsage:
    model_usage: list[ModelUsage]

class EOUMetrics(_BaseMetrics):
    type: Literal["eou_metrics"] = "eou_metrics"
    timestamp: float
    end_of_utterance_delay: float
    """Amount of time between the end of speech from VAD and the decision to end the user's turn.
    Set to 0.0 if the end of speech was not detected.
    """

    transcription_delay: float
    """Time taken to obtain the transcript after the end of the user's speech.
    Set to 0.0 if the end of speech was not detected.
    """

    on_user_turn_completed_delay: float
    """Time taken to invoke the user's `Agent.on_user_turn_completed` callback."""

    speech_id: str | None = None

    metadata: Metadata | None = None

class InterruptionMetrics(_BaseMetrics):
    type: Literal["interruption_metrics"] = "interruption_metrics"
    timestamp: float
    total_duration: float
    """Latest RTT (Round Trip Time) time taken to perform the inference, in seconds."""
    prediction_duration: float
    """Latest time taken to perform the inference from the model side, in seconds."""
    detection_delay: float
    """Latest total time from the onset of the speech to the final prediction, in seconds."""
    num_interruptions: int
    """Number of interruptions detected, incrementally counted."""
    num_backchannels: int
    """Number of backchannels detected, incrementally counted."""
    num_requests: int
    """Number of requests sent to the interruption detection model, incrementally counted."""
    metadata: Metadata | None = None

class InterruptionModelUsage(_BaseModelUsage):
    """Usage summary for interruption detection models."""

    type: Literal["interruption_usage"] = "interruption_usage"
    provider: str
    """The provider name (e.g., 'livekit')."""
    model: str
    """The model name (e.g., 'adaptive')."""
    total_requests: int = 0
    """Total number of requests sent to the interruption detection model."""

class LLMMetrics(_BaseMetrics):
    type: Literal["llm_metrics"] = "llm_metrics"
    label: str
    request_id: str
    timestamp: float
    duration: float
    ttft: float
    cancelled: bool
    completion_tokens: int
    prompt_tokens: int
    prompt_cached_tokens: int
    total_tokens: int
    tokens_per_second: float
    speech_id: str | None = None
    metadata: Metadata | None = None

class LLMModelUsage(_BaseModelUsage):
    """Usage summary for LLM models."""

    type: Literal["llm_usage"] = "llm_usage"
    provider: str
    """The provider name (e.g., 'openai', 'anthropic')."""
    model: str
    """The model name (e.g., 'gpt-4o', 'claude-3-5-sonnet')."""

    input_tokens: int = 0
    """Total input tokens."""
    input_cached_tokens: int = 0
    """Input tokens served from cache."""
    input_audio_tokens: int = 0
    """Input audio tokens (for multimodal models)."""
    input_cached_audio_tokens: int = 0
    """Cached input audio tokens."""
    input_text_tokens: int = 0
    """Input text tokens."""
    input_cached_text_tokens: int = 0
    """Cached input text tokens."""
    input_image_tokens: int = 0
    """Input image tokens (for multimodal models)."""
    input_cached_image_tokens: int = 0
    """Cached input image tokens."""

    output_tokens: int = 0
    """Total output tokens."""
    output_audio_tokens: int = 0
    """Output audio tokens (for multimodal models)."""
    output_text_tokens: int = 0
    """Output text tokens."""

    session_duration: float = 0.0
    """Total session connection duration in seconds (for session-based billing like xAI)."""

class ModelUsageCollector:
    """Collects and aggregates usage metrics per model/provider combination."""

    def __init__(self) -> None:
        self._llm_usage: dict[tuple[str, str], LLMModelUsage] = {}
        self._tts_usage: dict[tuple[str, str], TTSModelUsage] = {}
        self._stt_usage: dict[tuple[str, str], STTModelUsage] = {}
        self._interruption_usage: dict[tuple[str, str], InterruptionModelUsage] = {}

    def __call__(self, metrics: AgentMetrics) -> None:
        self.collect(metrics)

    def _extract_provider_model(
        self,
        metrics: LLMMetrics | STTMetrics | TTSMetrics | RealtimeModelMetrics | InterruptionMetrics,
    ) -> tuple[str, str]:
        """Extract provider and model from metrics metadata."""
        provider = ""
        model = ""
        if metrics.metadata:
            provider = metrics.metadata.model_provider or ""
            model = metrics.metadata.model_name or ""
        return provider, model

    def _get_llm_usage(self, provider: str, model: str) -> LLMModelUsage:
        """Get or create an LLMModelUsage for the given provider/model combination."""
        key = (provider, model)
        if key not in self._llm_usage:
            self._llm_usage[key] = LLMModelUsage(provider=provider, model=model)
        return self._llm_usage[key]

    def _get_tts_usage(self, provider: str, model: str) -> TTSModelUsage:
        """Get or create a TTSModelUsage for the given provider/model combination."""
        key = (provider, model)
        if key not in self._tts_usage:
            self._tts_usage[key] = TTSModelUsage(provider=provider, model=model)
        return self._tts_usage[key]

    def _get_stt_usage(self, provider: str, model: str) -> STTModelUsage:
        """Get or create an STTModelUsage for the given provider/model combination."""
        key = (provider, model)
        if key not in self._stt_usage:
            self._stt_usage[key] = STTModelUsage(provider=provider, model=model)
        return self._stt_usage[key]

    def _get_interruption_usage(self, provider: str, model: str) -> InterruptionModelUsage:
        """Get or create an InterruptionModelUsage for the given provider/model combination."""
        key = (provider, model)
        if key not in self._interruption_usage:
            self._interruption_usage[key] = InterruptionModelUsage(provider=provider, model=model)
        return self._interruption_usage[key]

    def collect(self, metrics: AgentMetrics) -> None:
        if isinstance(metrics, LLMMetrics):
            provider, model = self._extract_provider_model(metrics)
            usage = self._get_llm_usage(provider, model)
            usage.input_tokens += metrics.prompt_tokens
            usage.input_cached_tokens += metrics.prompt_cached_tokens
            usage.output_tokens += metrics.completion_tokens

        elif isinstance(metrics, RealtimeModelMetrics):
            provider, model = self._extract_provider_model(metrics)
            usage = self._get_llm_usage(provider, model)
            usage.input_tokens += metrics.input_tokens
            usage.input_cached_tokens += metrics.input_token_details.cached_tokens

            usage.input_text_tokens += metrics.input_token_details.text_tokens
            usage.input_cached_text_tokens += (
                metrics.input_token_details.cached_tokens_details.text_tokens
                if metrics.input_token_details.cached_tokens_details
                else 0
            )
            usage.input_image_tokens += metrics.input_token_details.image_tokens
            usage.input_cached_image_tokens += (
                metrics.input_token_details.cached_tokens_details.image_tokens
                if metrics.input_token_details.cached_tokens_details
                else 0
            )
            usage.input_audio_tokens += metrics.input_token_details.audio_tokens
            usage.input_cached_audio_tokens += (
                metrics.input_token_details.cached_tokens_details.audio_tokens
                if metrics.input_token_details.cached_tokens_details
                else 0
            )

            usage.output_text_tokens += metrics.output_token_details.text_tokens
            usage.output_audio_tokens += metrics.output_token_details.audio_tokens
            usage.output_tokens += metrics.output_tokens
            usage.session_duration += metrics.session_duration

        elif isinstance(metrics, TTSMetrics):
            provider, model = self._extract_provider_model(metrics)
            tts_usage = self._get_tts_usage(provider, model)
            tts_usage.input_tokens += metrics.input_tokens
            tts_usage.output_tokens += metrics.output_tokens
            tts_usage.characters_count += metrics.characters_count
            tts_usage.audio_duration += metrics.audio_duration

        elif isinstance(metrics, STTMetrics):
            provider, model = self._extract_provider_model(metrics)
            stt_usage = self._get_stt_usage(provider, model)
            stt_usage.input_tokens += metrics.input_tokens
            stt_usage.output_tokens += metrics.output_tokens
            stt_usage.audio_duration += metrics.audio_duration
        elif isinstance(metrics, InterruptionMetrics):
            provider, model = self._extract_provider_model(metrics)
            interruption_usage = self._get_interruption_usage(provider, model)
            interruption_usage.total_requests += metrics.num_requests

    def flatten(self) -> list[ModelUsage]:
        """Returns a list of usage summaries, one per model/provider combination."""
        result: list[ModelUsage] = []
        result.extend(u.model_copy(deep=True) for u in self._llm_usage.values())
        result.extend(u.model_copy(deep=True) for u in self._tts_usage.values())
        result.extend(u.model_copy(deep=True) for u in self._stt_usage.values())
        result.extend(u.model_copy(deep=True) for u in self._interruption_usage.values())
        return result

class RealtimeModelMetrics(_BaseMetrics):
    class CachedTokenDetails(BaseModel):
        audio_tokens: int = 0
        text_tokens: int = 0
        image_tokens: int = 0

    class InputTokenDetails(BaseModel):
        audio_tokens: int = 0
        text_tokens: int = 0
        image_tokens: int = 0
        cached_tokens: int = 0
        cached_tokens_details: RealtimeModelMetrics.CachedTokenDetails | None = None

    class OutputTokenDetails(BaseModel):
        text_tokens: int = 0
        audio_tokens: int = 0
        # image_tokens is deprecated, Realtime models no longer emit this metric
        image_tokens: int = 0

    type: Literal["realtime_model_metrics"] = "realtime_model_metrics"
    label: str = ""
    request_id: str
    timestamp: float
    """The timestamp of the response creation."""
    duration: float = 0.0
    """The duration of the response from created to done in seconds."""
    session_duration: float = 0.0
    """The duration of the session connection in seconds (for session-based billing like xAI)."""
    ttft: float = -1
    """Time to first audio token in seconds. -1 if no audio token was sent."""
    cancelled: bool = False
    """Whether the request was cancelled."""
    input_tokens: int = 0
    """The number of input tokens used in the Response, including text and audio tokens."""
    output_tokens: int = 0
    """The number of output tokens sent in the Response, including text and audio tokens."""
    total_tokens: int = 0
    """The total number of tokens in the Response."""
    tokens_per_second: float = 0.0
    """The number of tokens per second."""
    input_token_details: InputTokenDetails
    """Details about the input tokens used in the Response."""
    output_token_details: OutputTokenDetails
    """Details about the output tokens used in the Response."""
    acquire_time: float = 0.0
    """Time in seconds to acquire the connection. (WebSocket only)"""
    connection_reused: bool = False
    """Whether the connection was reused from a pool. (WebSocket only)"""
    metadata: Metadata | None = None

class STTMetrics(_BaseMetrics):
    type: Literal["stt_metrics"] = "stt_metrics"
    label: str
    request_id: str
    timestamp: float
    duration: float
    """The request duration in seconds, 0.0 if the STT is streaming."""
    audio_duration: float
    """The duration of the pushed audio in seconds."""
    input_tokens: int = 0
    """Input audio tokens (for token-based billing)."""
    output_tokens: int = 0
    """Output text tokens (for token-based billing)."""
    streamed: bool
    """Whether the STT is streaming (e.g using websocket)."""
    acquire_time: float = 0.0
    """Time in seconds to acquire the connection. (WebSocket only)"""
    connection_reused: bool = False
    """Whether the connection was reused from a pool. (WebSocket only)"""
    metadata: Metadata | None = None

class STTModelUsage(_BaseModelUsage):
    """Usage summary for STT models."""

    type: Literal["stt_usage"] = "stt_usage"
    provider: str
    """The provider name (e.g., 'deepgram', 'assemblyai')."""
    model: str
    """The model name (e.g., 'nova-2', 'best')."""

    input_tokens: int = 0
    """Input audio tokens (for token-based STT billing)."""
    output_tokens: int = 0
    """Output text tokens (for token-based STT billing)."""
    audio_duration: float = 0.0
    """Duration of processed audio in seconds."""

class TTSMetrics(_BaseMetrics):
    type: Literal["tts_metrics"] = "tts_metrics"
    label: str
    request_id: str
    timestamp: float
    ttfb: float
    duration: float
    audio_duration: float
    cancelled: bool
    characters_count: int
    """Number of characters synthesized (for character-based billing)."""
    input_tokens: int = 0
    """Input text tokens (for token-based billing, e.g., OpenAI TTS)."""
    output_tokens: int = 0
    """Output audio tokens (for token-based billing, e.g., OpenAI TTS)."""
    streamed: bool
    acquire_time: float = 0.0
    """Time in seconds to acquire the connection. (WebSocket only)"""
    connection_reused: bool = False
    """Whether the connection was reused from a pool. (WebSocket only)"""
    segment_id: str | None = None
    speech_id: str | None = None
    metadata: Metadata | None = None

class TTSModelUsage(_BaseModelUsage):
    """Usage summary for TTS models."""

    type: Literal["tts_usage"] = "tts_usage"
    provider: str
    """The provider name (e.g., 'elevenlabs', 'cartesia')."""
    model: str
    """The model name (e.g., 'eleven_turbo_v2', 'sonic')."""

    input_tokens: int = 0
    """Input text tokens (for token-based TTS billing, e.g., OpenAI TTS)."""
    output_tokens: int = 0
    """Output audio tokens (for token-based TTS billing, e.g., OpenAI TTS)."""
    characters_count: int = 0
    """Number of characters synthesized (for character-based TTS billing)."""
    audio_duration: float = 0.0
    """Duration of generated audio in seconds."""

class UsageCollector:
    """
    .. deprecated::
        Use :class:`ModelUsageCollector` instead.
    """

    def __init__(self) -> None:
        warnings.warn(
            "UsageCollector is deprecated. Use ModelUsageCollector from "
            "metrics.model_usage instead.",
            DeprecationWarning,
            stacklevel=2,
        )
        self._summary = UsageSummary()

    def __call__(self, metrics: AgentMetrics) -> None:
        self.collect(metrics)

    def collect(self, metrics: AgentMetrics) -> None:
        if isinstance(metrics, LLMMetrics):
            self._summary.llm_prompt_tokens += metrics.prompt_tokens
            self._summary.llm_prompt_cached_tokens += metrics.prompt_cached_tokens
            self._summary.llm_completion_tokens += metrics.completion_tokens

        elif isinstance(metrics, RealtimeModelMetrics):
            self._summary.llm_prompt_tokens += metrics.input_tokens
            self._summary.llm_prompt_cached_tokens += metrics.input_token_details.cached_tokens

            self._summary.llm_input_text_tokens += metrics.input_token_details.text_tokens
            self._summary.llm_input_cached_text_tokens += (
                metrics.input_token_details.cached_tokens_details.text_tokens
                if metrics.input_token_details.cached_tokens_details
                else 0
            )
            self._summary.llm_input_image_tokens += metrics.input_token_details.image_tokens
            self._summary.llm_input_cached_image_tokens += (
                metrics.input_token_details.cached_tokens_details.image_tokens
                if metrics.input_token_details.cached_tokens_details
                else 0
            )
            self._summary.llm_input_audio_tokens += metrics.input_token_details.audio_tokens
            self._summary.llm_input_cached_audio_tokens += (
                metrics.input_token_details.cached_tokens_details.audio_tokens
                if metrics.input_token_details.cached_tokens_details
                else 0
            )

            self._summary.llm_output_text_tokens += metrics.output_token_details.text_tokens
            self._summary.llm_output_image_tokens += metrics.output_token_details.image_tokens
            self._summary.llm_output_audio_tokens += metrics.output_token_details.audio_tokens
            self._summary.llm_completion_tokens += metrics.output_tokens

        elif isinstance(metrics, TTSMetrics):
            self._summary.tts_characters_count += metrics.characters_count
            self._summary.tts_audio_duration += metrics.audio_duration

        elif isinstance(metrics, STTMetrics):
            self._summary.stt_audio_duration += metrics.audio_duration

    def get_summary(self) -> UsageSummary:
        return deepcopy(self._summary)

@dataclass
class UsageSummary:
    """
    .. deprecated::
        Use :class:`LLMModelUsage`, :class:`TTSModelUsage`, or :class:`STTModelUsage` instead.
    """

    llm_prompt_tokens: int = 0
    llm_prompt_cached_tokens: int = 0
    llm_input_audio_tokens: int = 0
    llm_input_cached_audio_tokens: int = 0
    llm_input_text_tokens: int = 0
    llm_input_cached_text_tokens: int = 0
    llm_input_image_tokens: int = 0
    llm_input_cached_image_tokens: int = 0
    llm_completion_tokens: int = 0
    llm_output_audio_tokens: int = 0
    llm_output_image_tokens: int = 0
    llm_output_text_tokens: int = 0
    tts_characters_count: int = 0
    tts_audio_duration: float = 0.0
    stt_audio_duration: float = 0.0

    def __post_init__(self) -> None:
        warnings.warn(
            "UsageSummary is deprecated. Use LLMModelUsage, TTSModelUsage, "
            "or STTModelUsage from metrics.model_usage instead.",
            DeprecationWarning,
            stacklevel=2,
        )

    # properties for naming consistency: prompt = input, completion = output
    @property
    def llm_input_tokens(self) -> int:
        return self.llm_prompt_tokens

    @llm_input_tokens.setter
    def llm_input_tokens(self, value: int) -> None:
        self.llm_prompt_tokens = value

    @property
    def llm_output_tokens(self) -> int:
        return self.llm_completion_tokens

    @llm_output_tokens.setter
    def llm_output_tokens(self, value: int) -> None:
        self.llm_completion_tokens = value

class VADMetrics(_BaseMetrics):
    type: Literal["vad_metrics"] = "vad_metrics"
    label: str
    timestamp: float
    idle_time: float
    inference_duration_total: float
    inference_count: int
    metadata: Metadata | None = None