Skip to content

Metrics#

ragbits.core.audit.metrics.set_metric_handlers #

set_metric_handlers(handlers: Handler | list[Handler]) -> None

Set the global metric handlers.

PARAMETER DESCRIPTION
handlers

List of metric handlers to be used.

TYPE: Handler | list[Handler]

RAISES DESCRIPTION
ValueError

If handler is not found.
TypeError

If handler type is invalid.
Source code in packages/ragbits-core/src/ragbits/core/audit/metrics/__init__.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def set_metric_handlers(handlers: Handler | list[Handler]) -> None:
    """
    Set the global metric handlers.

    Args:
        handlers: List of metric handlers to be used.

    Raises:
        ValueError: If handler is not found.
        TypeError: If handler type is invalid.
    """
    global _metric_handlers  # noqa: PLW0602

    if isinstance(handlers, Handler):
        handlers = [handlers]

    for handler in handlers:
        if isinstance(handler, MetricHandler):
            _metric_handlers.append(handler)
        elif isinstance(handler, str):
            match handler.lower():
                case "otel":
                    from ragbits.core.audit.metrics.otel import OtelMetricHandler

                    if not any(isinstance(item, OtelMetricHandler) for item in _metric_handlers):
                        _metric_handlers.append(OtelMetricHandler())

                case "logfire":
                    from ragbits.core.audit.metrics.logfire import LogfireMetricHandler

                    if not any(isinstance(item, LogfireMetricHandler) for item in _metric_handlers):
                        _metric_handlers.append(LogfireMetricHandler())

                case _:
                    raise ValueError(f"Not found handler: {handler}")
        else:
            raise TypeError(f"Invalid handler type: {type(handler)}")

ragbits.core.audit.metrics.clear_metric_handlers #

clear_metric_handlers() -> None

Clear all metric handlers.

Source code in packages/ragbits-core/src/ragbits/core/audit/metrics/__init__.py 
59
60
61
62
63
64
def clear_metric_handlers() -> None:
    """
    Clear all metric handlers.
    """
    global _metric_handlers  # noqa: PLW0602
    _metric_handlers.clear()

ragbits.core.audit.metrics.register_metric #

register_metric(key: str | Enum, metric: Metric) -> None

Register a new metric in the global registry by type.

PARAMETER DESCRIPTION
key

The metric key (enum value or string)

TYPE: str | Enum

metric

The metric configuration

TYPE: Metric

Source code in packages/ragbits-core/src/ragbits/core/audit/metrics/base.py 
75
76
77
78
79
80
81
82
83
def register_metric(key: str | Enum, metric: Metric) -> None:
    """
    Register a new metric in the global registry by type.

    Args:
        key: The metric key (enum value or string)
        metric: The metric configuration
    """
    METRICS_REGISTRY[metric.type][key] = metric

ragbits.core.audit.metrics.record_metric #

record_metric(metric: str | Enum, value: int | float, metric_type: MetricType, **attributes: Any) -> None

Record a metric of any type using the global metric handlers.

PARAMETER DESCRIPTION
metric

The metric key (name or enum value) to record

TYPE: str | Enum

value

The value to record

TYPE: int | float

metric_type

The type of metric (histogram, counter, gauge)

TYPE: MetricType

**attributes

Additional metadata for the metric

TYPE: Any DEFAULT: {}

Source code in packages/ragbits-core/src/ragbits/core/audit/metrics/__init__.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def record_metric(
    metric: str | Enum,
    value: int | float,
    metric_type: MetricType,
    **attributes: Any,  # noqa: ANN401
) -> None:
    """
    Record a metric of any type using the global metric handlers.

    Args:
        metric: The metric key (name or enum value) to record
        value: The value to record
        metric_type: The type of metric (histogram, counter, gauge)
        **attributes: Additional metadata for the metric
    """
    for handler in _metric_handlers:
        handler.record_metric(metric_key=metric, value=value, attributes=attributes, metric_type=metric_type)

ragbits.core.audit.metrics.base.MetricType #

Bases: Enum

Supported metric types.

HISTOGRAM class-attribute instance-attribute #

HISTOGRAM = 'histogram'

COUNTER class-attribute instance-attribute #

COUNTER = 'counter'

GAUGE class-attribute instance-attribute #

GAUGE = 'gauge'

ragbits.core.audit.metrics.base.Metric dataclass #

Metric(name: str, description: str, unit: str, type: MetricType)

Represents the metric configuration data.

name instance-attribute #

name: str

description instance-attribute #

description: str

unit instance-attribute #

unit: str

type instance-attribute #

type: MetricType

ragbits.core.audit.metrics.base.LLMMetric #

Bases: Enum

LLM-related metrics that can be recorded. Each metric has a predefined type and is registered in the global registry.

PROMPT_THROUGHPUT class-attribute instance-attribute #

PROMPT_THROUGHPUT = auto()

TOKEN_THROUGHPUT class-attribute instance-attribute #

TOKEN_THROUGHPUT = auto()

INPUT_TOKENS class-attribute instance-attribute #

INPUT_TOKENS = auto()

TIME_TO_FIRST_TOKEN class-attribute instance-attribute #

TIME_TO_FIRST_TOKEN = auto()

ragbits.core.audit.metrics.base.MetricHandler #

MetricHandler(metric_prefix: str = 'ragbits')

Bases: ABC

Base class for all metric handlers.

Initialize the MetricHandler instance.

PARAMETER DESCRIPTION
metric_prefix

Prefix for all metric names.

TYPE: str DEFAULT: 'ragbits'

Source code in packages/ragbits-core/src/ragbits/core/audit/metrics/base.py 
105
106
107
108
109
110
111
112
113
114
def __init__(self, metric_prefix: str = "ragbits") -> None:
    """
    Initialize the MetricHandler instance.

    Args:
        metric_prefix: Prefix for all metric names.
    """
    super().__init__()
    self._metric_prefix = metric_prefix
    self._metrics: dict[str, Any] = {}

create_metric abstractmethod #

create_metric(name: str, unit: str = '', description: str = '', metric_type: MetricType = MetricType.HISTOGRAM) -> Any

Create a metric of the given type.

PARAMETER DESCRIPTION
name

The metric name.

TYPE: str

unit

The metric unit.

TYPE: str DEFAULT: ''

description

The metric description.

TYPE: str DEFAULT: ''

metric_type

The type of the metric (histogram, counter, gauge).

TYPE: MetricType DEFAULT: HISTOGRAM

RETURNS DESCRIPTION
Any

The initialized metric.
Source code in packages/ragbits-core/src/ragbits/core/audit/metrics/base.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
@abstractmethod
def create_metric(
    self, name: str, unit: str = "", description: str = "", metric_type: MetricType = MetricType.HISTOGRAM
) -> Any:  # noqa: ANN401
    """
    Create a metric of the given type.

    Args:
        name: The metric name.
        unit: The metric unit.
        description: The metric description.
        metric_type: The type of the metric (histogram, counter, gauge).

    Returns:
        The initialized metric.
    """

register_metric_instance #

register_metric_instance(name: str, unit: str = '', description: str = '', metric_type: MetricType = MetricType.HISTOGRAM) -> None

Register a metric instance.

PARAMETER DESCRIPTION
name

The metric name.

TYPE: str

unit

The metric unit.

TYPE: str DEFAULT: ''

description

The metric description.

TYPE: str DEFAULT: ''

metric_type

The type of the metric (histogram, counter, gauge).

TYPE: MetricType DEFAULT: HISTOGRAM

Source code in packages/ragbits-core/src/ragbits/core/audit/metrics/base.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def register_metric_instance(
    self, name: str, unit: str = "", description: str = "", metric_type: MetricType = MetricType.HISTOGRAM
) -> None:
    """
    Register a metric instance.

    Args:
        name: The metric name.
        unit: The metric unit.
        description: The metric description.
        metric_type: The type of the metric (histogram, counter, gauge).
    """
    self._metrics[name] = self.create_metric(
        name=f"{self._metric_prefix}_{name}",
        unit=unit,
        description=description,
        metric_type=metric_type,
    )

record_metric #

record_metric(metric_key: str | Enum, value: int | float, attributes: dict | None = None, metric_type: MetricType = MetricType.HISTOGRAM) -> None

Record the value for a specified metric.

PARAMETER DESCRIPTION
metric_key

The metric key (name or enum value) to record.

TYPE: str | Enum

value

The value to record for the metric.

TYPE: int | float

attributes

Additional metadata for the metric.

TYPE: dict | None DEFAULT: None

metric_type

The type of the metric (histogram, counter, gauge).

TYPE: MetricType DEFAULT: HISTOGRAM

Source code in packages/ragbits-core/src/ragbits/core/audit/metrics/base.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
def record_metric(
    self,
    metric_key: str | Enum,
    value: int | float,
    attributes: dict | None = None,
    metric_type: MetricType = MetricType.HISTOGRAM,
) -> None:
    """
    Record the value for a specified metric.

    Args:
        metric_key: The metric key (name or enum value) to record.
        value: The value to record for the metric.
        attributes: Additional metadata for the metric.
        metric_type: The type of the metric (histogram, counter, gauge).
    """
    metric_cfg = get_metric(metric_key, metric_type)
    if metric_cfg:
        metric_name = metric_cfg.name
        if metric_name not in self._metrics:
            self.register_metric_instance(
                name=metric_name,
                unit=metric_cfg.unit,
                description=metric_cfg.description,
                metric_type=metric_type,
            )
    else:
        metric_name = str(metric_key)
        if metric_name not in self._metrics:
            self.register_metric_instance(metric_name, metric_type=metric_type)
    self._record(
        metric=self._metrics[metric_name],
        value=value,
        attributes=attributes,
    )

ragbits.core.audit.metrics.otel.OtelMetricHandler #

OtelMetricHandler(provider: MeterProvider | None = None, metric_prefix: str = 'ragbits')

Bases: MetricHandler

OpenTelemetry metric handler.

Initialize the OtelMetricHandler instance.

PARAMETER DESCRIPTION
provider

The meter provider to use.

TYPE: MeterProvider | None DEFAULT: None

metric_prefix

Prefix for all metric names.

TYPE: str DEFAULT: 'ragbits'

Source code in packages/ragbits-core/src/ragbits/core/audit/metrics/otel.py 
13
14
15
16
17
18
19
20
21
22
def __init__(self, provider: MeterProvider | None = None, metric_prefix: str = "ragbits") -> None:
    """
    Initialize the OtelMetricHandler instance.

    Args:
        provider: The meter provider to use.
        metric_prefix: Prefix for all metric names.
    """
    super().__init__(metric_prefix=metric_prefix)
    self._meter = get_meter(name=__name__, meter_provider=provider)

register_metric_instance #

register_metric_instance(name: str, unit: str = '', description: str = '', metric_type: MetricType = MetricType.HISTOGRAM) -> None

Register a metric instance.

PARAMETER DESCRIPTION
name

The metric name.

TYPE: str

unit

The metric unit.

TYPE: str DEFAULT: ''

description

The metric description.

TYPE: str DEFAULT: ''

metric_type

The type of the metric (histogram, counter, gauge).

TYPE: MetricType DEFAULT: HISTOGRAM

Source code in packages/ragbits-core/src/ragbits/core/audit/metrics/base.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def register_metric_instance(
    self, name: str, unit: str = "", description: str = "", metric_type: MetricType = MetricType.HISTOGRAM
) -> None:
    """
    Register a metric instance.

    Args:
        name: The metric name.
        unit: The metric unit.
        description: The metric description.
        metric_type: The type of the metric (histogram, counter, gauge).
    """
    self._metrics[name] = self.create_metric(
        name=f"{self._metric_prefix}_{name}",
        unit=unit,
        description=description,
        metric_type=metric_type,
    )

record_metric #

record_metric(metric_key: str | Enum, value: int | float, attributes: dict | None = None, metric_type: MetricType = MetricType.HISTOGRAM) -> None

Record the value for a specified metric.

PARAMETER DESCRIPTION
metric_key

The metric key (name or enum value) to record.

TYPE: str | Enum

value

The value to record for the metric.

TYPE: int | float

attributes

Additional metadata for the metric.

TYPE: dict | None DEFAULT: None

metric_type

The type of the metric (histogram, counter, gauge).

TYPE: MetricType DEFAULT: HISTOGRAM

Source code in packages/ragbits-core/src/ragbits/core/audit/metrics/base.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
def record_metric(
    self,
    metric_key: str | Enum,
    value: int | float,
    attributes: dict | None = None,
    metric_type: MetricType = MetricType.HISTOGRAM,
) -> None:
    """
    Record the value for a specified metric.

    Args:
        metric_key: The metric key (name or enum value) to record.
        value: The value to record for the metric.
        attributes: Additional metadata for the metric.
        metric_type: The type of the metric (histogram, counter, gauge).
    """
    metric_cfg = get_metric(metric_key, metric_type)
    if metric_cfg:
        metric_name = metric_cfg.name
        if metric_name not in self._metrics:
            self.register_metric_instance(
                name=metric_name,
                unit=metric_cfg.unit,
                description=metric_cfg.description,
                metric_type=metric_type,
            )
    else:
        metric_name = str(metric_key)
        if metric_name not in self._metrics:
            self.register_metric_instance(metric_name, metric_type=metric_type)
    self._record(
        metric=self._metrics[metric_name],
        value=value,
        attributes=attributes,
    )

create_metric #

create_metric(name: str, unit: str = '', description: str = '', metric_type: MetricType = MetricType.HISTOGRAM) -> Any

Create a metric of the specified type.

PARAMETER DESCRIPTION
name

The metric name.

TYPE: str

unit

The metric unit.

TYPE: str DEFAULT: ''

description

The metric description.

TYPE: str DEFAULT: ''

metric_type

The type of the metric (histogram, counter, gauge).

TYPE: MetricType DEFAULT: HISTOGRAM

RETURNS DESCRIPTION
Any

The initialized metric.
Source code in packages/ragbits-core/src/ragbits/core/audit/metrics/otel.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def create_metric(
    self, name: str, unit: str = "", description: str = "", metric_type: MetricType = MetricType.HISTOGRAM
) -> Any:  # noqa: ANN401
    """
    Create a metric of the specified type.

    Args:
        name: The metric name.
        unit: The metric unit.
        description: The metric description.
        metric_type: The type of the metric (histogram, counter, gauge).

    Returns:
        The initialized metric.
    """
    if metric_type == MetricType.HISTOGRAM:
        return self._meter.create_histogram(name=name, unit=unit, description=description)
    elif metric_type == MetricType.COUNTER:
        return self._meter.create_counter(name=name, unit=unit, description=description)
    elif metric_type == MetricType.GAUGE:
        return self._meter.create_gauge(name=name, unit=unit, description=description)
    else:
        raise ValueError(f"Unsupported metric type: {metric_type}")

ragbits.core.audit.metrics.logfire.LogfireMetricHandler #

LogfireMetricHandler(metric_prefix: str = 'ragbits', *args: Any, **kwargs: Any)

Bases: OtelMetricHandler

Logfire metric handler.

Initialize the LogfireMetricHandler instance.

Source code in packages/ragbits-core/src/ragbits/core/audit/metrics/logfire.py 
13
14
15
16
17
18
19
def __init__(self, metric_prefix: str = "ragbits", *args: Any, **kwargs: Any) -> None:  # noqa: ANN401
    """
    Initialize the LogfireMetricHandler instance.
    """
    logfire.configure(*args, **kwargs)
    logfire.instrument_system_metrics()
    super().__init__(metric_prefix=metric_prefix)

create_metric #

create_metric(name: str, unit: str = '', description: str = '', metric_type: MetricType = MetricType.HISTOGRAM) -> Any

Create a metric of the specified type.

PARAMETER DESCRIPTION
name

The metric name.

TYPE: str

unit

The metric unit.

TYPE: str DEFAULT: ''

description

The metric description.

TYPE: str DEFAULT: ''

metric_type

The type of the metric (histogram, counter, gauge).

TYPE: MetricType DEFAULT: HISTOGRAM

RETURNS DESCRIPTION
Any

The initialized metric.
Source code in packages/ragbits-core/src/ragbits/core/audit/metrics/otel.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def create_metric(
    self, name: str, unit: str = "", description: str = "", metric_type: MetricType = MetricType.HISTOGRAM
) -> Any:  # noqa: ANN401
    """
    Create a metric of the specified type.

    Args:
        name: The metric name.
        unit: The metric unit.
        description: The metric description.
        metric_type: The type of the metric (histogram, counter, gauge).

    Returns:
        The initialized metric.
    """
    if metric_type == MetricType.HISTOGRAM:
        return self._meter.create_histogram(name=name, unit=unit, description=description)
    elif metric_type == MetricType.COUNTER:
        return self._meter.create_counter(name=name, unit=unit, description=description)
    elif metric_type == MetricType.GAUGE:
        return self._meter.create_gauge(name=name, unit=unit, description=description)
    else:
        raise ValueError(f"Unsupported metric type: {metric_type}")

register_metric_instance #

register_metric_instance(name: str, unit: str = '', description: str = '', metric_type: MetricType = MetricType.HISTOGRAM) -> None

Register a metric instance.

PARAMETER DESCRIPTION
name

The metric name.

TYPE: str

unit

The metric unit.

TYPE: str DEFAULT: ''

description

The metric description.

TYPE: str DEFAULT: ''

metric_type

The type of the metric (histogram, counter, gauge).

TYPE: MetricType DEFAULT: HISTOGRAM

Source code in packages/ragbits-core/src/ragbits/core/audit/metrics/base.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def register_metric_instance(
    self, name: str, unit: str = "", description: str = "", metric_type: MetricType = MetricType.HISTOGRAM
) -> None:
    """
    Register a metric instance.

    Args:
        name: The metric name.
        unit: The metric unit.
        description: The metric description.
        metric_type: The type of the metric (histogram, counter, gauge).
    """
    self._metrics[name] = self.create_metric(
        name=f"{self._metric_prefix}_{name}",
        unit=unit,
        description=description,
        metric_type=metric_type,
    )

record_metric #

record_metric(metric_key: str | Enum, value: int | float, attributes: dict | None = None, metric_type: MetricType = MetricType.HISTOGRAM) -> None

Record the value for a specified metric.

PARAMETER DESCRIPTION
metric_key

The metric key (name or enum value) to record.

TYPE: str | Enum

value

The value to record for the metric.

TYPE: int | float

attributes

Additional metadata for the metric.

TYPE: dict | None DEFAULT: None

metric_type

The type of the metric (histogram, counter, gauge).

TYPE: MetricType DEFAULT: HISTOGRAM

Source code in packages/ragbits-core/src/ragbits/core/audit/metrics/base.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
def record_metric(
    self,
    metric_key: str | Enum,
    value: int | float,
    attributes: dict | None = None,
    metric_type: MetricType = MetricType.HISTOGRAM,
) -> None:
    """
    Record the value for a specified metric.

    Args:
        metric_key: The metric key (name or enum value) to record.
        value: The value to record for the metric.
        attributes: Additional metadata for the metric.
        metric_type: The type of the metric (histogram, counter, gauge).
    """
    metric_cfg = get_metric(metric_key, metric_type)
    if metric_cfg:
        metric_name = metric_cfg.name
        if metric_name not in self._metrics:
            self.register_metric_instance(
                name=metric_name,
                unit=metric_cfg.unit,
                description=metric_cfg.description,
                metric_type=metric_type,
            )
    else:
        metric_name = str(metric_key)
        if metric_name not in self._metrics:
            self.register_metric_instance(metric_name, metric_type=metric_type)
    self._record(
        metric=self._metrics[metric_name],
        value=value,
        attributes=attributes,
    )