Source code for lalandre_extraction.metrics

"""Metrics hook interfaces for extraction LLM instrumentation."""

from abc import ABC, abstractmethod




class ExtractionMetricsRecorder(ABC):
    """Backend-agnostic hook interface for extraction LLM metrics."""



    @abstractmethod
    def observe_call(
        self,
        *,
        provider: str,
        model: str,
        outcome: str,
        duration_seconds: float,
    ) -> None:
        """Record one extraction LLM call outcome and latency."""
        ...




    @abstractmethod
    def observe_error(
        self,
        *,
        provider: str,
        model: str,
        error_type: str,
    ) -> None:
        """Record one extraction LLM error bucket."""
        ...




    @abstractmethod
    def observe_json_parse(
        self,
        *,
        provider: str,
        model: str,
        parse_mode: str,
    ) -> None:
        """Record how raw model output was parsed into structured relations."""
        ...




    @abstractmethod
    def observe_relations(
        self,
        *,
        provider: str,
        model: str,
        count: int,
    ) -> None:
        """Record the number of relations produced by a successful call."""
        ...




class _NoopExtractionMetricsRecorder(ExtractionMetricsRecorder):
    def observe_call(
        self,
        *,
        provider: str,
        model: str,
        outcome: str,
        duration_seconds: float,
    ) -> None:
        return None

    def observe_error(
        self,
        *,
        provider: str,
        model: str,
        error_type: str,
    ) -> None:
        return None

    def observe_json_parse(
        self,
        *,
        provider: str,
        model: str,
        parse_mode: str,
    ) -> None:
        return None

    def observe_relations(
        self,
        *,
        provider: str,
        model: str,
        count: int,
    ) -> None:
        return None


_recorder: ExtractionMetricsRecorder = _NoopExtractionMetricsRecorder()




def set_extraction_metrics_recorder(recorder: ExtractionMetricsRecorder) -> None:
    """Register the active extraction metrics recorder backend."""
    global _recorder
    _recorder = recorder





def observe_extraction_llm_call(
    *,
    provider: str,
    model: str,
    outcome: str,
    duration_seconds: float,
) -> None:
    """Forward one extraction-call latency observation to the active recorder."""
    _recorder.observe_call(
        provider=provider,
        model=model,
        outcome=outcome,
        duration_seconds=duration_seconds,
    )





def observe_extraction_llm_error(
    *,
    provider: str,
    model: str,
    error_type: str,
) -> None:
    """Forward one extraction error observation to the active recorder."""
    _recorder.observe_error(
        provider=provider,
        model=model,
        error_type=error_type,
    )





def observe_extraction_llm_relations(
    *,
    provider: str,
    model: str,
    count: int,
) -> None:
    """Forward one extracted-relations count observation to the active recorder."""
    _recorder.observe_relations(
        provider=provider,
        model=model,
        count=count,
    )