Module livekit.agents.metrics

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

    if isinstance(metrics, PipelineLLMMetrics):
        logger.info(
            f"Pipeline LLM metrics: sequence_id={metrics.sequence_id}, ttft={metrics.ttft:.2f}, input_tokens={metrics.prompt_tokens}, output_tokens={metrics.completion_tokens}, tokens_per_second={metrics.tokens_per_second:.2f}"
        )
    elif isinstance(metrics, LLMMetrics):
        logger.info(
            f"LLM metrics: ttft={metrics.ttft:.2f}, input_tokens={metrics.prompt_tokens}, output_tokens={metrics.completion_tokens}, tokens_per_second={metrics.tokens_per_second:.2f}"
        )
    elif isinstance(metrics, PipelineTTSMetrics):
        logger.info(
            f"Pipeline TTS metrics: sequence_id={metrics.sequence_id}, ttfb={metrics.ttfb}, audio_duration={metrics.audio_duration:.2f}"
        )
    elif isinstance(metrics, TTSMetrics):
        logger.info(
            f"TTS metrics: ttfb={metrics.ttfb}, audio_duration={metrics.audio_duration:.2f}"
        )
    elif isinstance(metrics, PipelineEOUMetrics):
        logger.info(
            f"Pipeline EOU metrics: sequence_id={metrics.sequence_id}, end_of_utterance_delay={metrics.end_of_utterance_delay:.2f}, transcription_delay={metrics.transcription_delay:.2f}"
        )
    elif isinstance(metrics, PipelineSTTMetrics):
        logger.info(
            f"Pipeline STT metrics: duration={metrics.duration:.2f}, audio_duration={metrics.audio_duration:.2f}"
        )
    elif isinstance(metrics, STTMetrics):
        logger.info(f"STT metrics: audio_duration={metrics.audio_duration:.2f}")

@dataclass
class LLMMetrics:
    request_id: str
    timestamp: float
    ttft: float
    duration: float
    label: str
    cancelled: bool
    completion_tokens: int
    prompt_tokens: int
    total_tokens: int
    tokens_per_second: float
    error: Error | None

@dataclass
class MultimodalLLMError(Error):
    type: str | None
    reason: str | None = None
    code: str | None = None
    message: str | None = None

@dataclass
class MultimodalLLMMetrics(LLMMetrics):
    @dataclass
    class CachedTokenDetails:
        text_tokens: int
        audio_tokens: int

    @dataclass
    class InputTokenDetails:
        cached_tokens: int
        text_tokens: int
        audio_tokens: int
        cached_tokens_details: MultimodalLLMMetrics.CachedTokenDetails

    @dataclass
    class OutputTokenDetails:
        text_tokens: int
        audio_tokens: int

    input_token_details: InputTokenDetails
    output_token_details: OutputTokenDetails

@dataclass
class PipelineEOUMetrics:
    sequence_id: str
    """Unique identifier shared across different metrics to combine related STT, LLM, and TTS metrics."""

    timestamp: float
    """Timestamp of when the event was recorded."""

    end_of_utterance_delay: float
    """Amount of time between the end of speech from VAD and the decision to end the user's turn."""

    transcription_delay: float
    """Time taken to obtain the transcript after the end of the user's speech.

    May be 0 if the transcript was already available.
    """

@dataclass
class PipelineLLMMetrics(LLMMetrics):
    sequence_id: str
    """Unique identifier shared across different metrics to combine related STT, LLM, and TTS metrics."""

@dataclass
class PipelineSTTMetrics(STTMetrics):
    pass

@dataclass
class PipelineTTSMetrics(TTSMetrics):
    sequence_id: str
    """Unique identifier shared across different metrics to combine related STT, LLM, and TTS metrics."""

@dataclass
class PipelineVADMetrics(VADMetrics):
    pass

@dataclass
class STTMetrics:
    request_id: str
    timestamp: float
    duration: float
    label: str
    audio_duration: float
    streamed: bool
    error: Error | None

@dataclass
class TTSMetrics:
    request_id: str
    timestamp: float
    ttfb: float
    duration: float
    audio_duration: float
    cancelled: bool
    characters_count: int
    label: str
    streamed: bool
    error: Error | None

class UsageCollector:
    def __init__(self) -> None:
        self._summary = UsageSummary(0, 0, 0, 0.0)

    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_completion_tokens += metrics.completion_tokens

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

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

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

@dataclass
class UsageSummary:
    llm_prompt_tokens: int
    llm_completion_tokens: int
    tts_characters_count: int
    stt_audio_duration: float

@dataclass
class VADMetrics:
    timestamp: float
    idle_time: float
    inference_duration_total: float
    inference_count: int
    label: str