Skip to content

metrics

optimus_dl.modules.metrics.metrics

AccuracyMetric

Bases: Metric

Computes Top-1 accuracy for classification tasks.

Source code in optimus_dl/modules/metrics/metrics.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
@register_metric("accuracy", AccuracyMetricConfig)
class AccuracyMetric(Metric):
    """Computes Top-1 accuracy for classification tasks."""

    @property
    def requires(self) -> set[str]:
        return {StandardProtocols.CLASSIFICATION}

    @property
    def accumulators(self) -> dict[str, str]:
        return {"accuracy": "average", "total": "sum"}

    def __call__(
        self, sources_data: dict[str, dict[str, Any]]
    ) -> dict[str, dict[str, Any]]:
        data = sources_data.get(StandardProtocols.CLASSIFICATION)
        if not data:
            return {}

        preds = data["predictions"]
        targets = data["targets"]
        mask = data.get("mask")

        correct = preds == targets
        if mask is not None:
            correct = correct & mask
            total = mask.sum().item()
        else:
            total = targets.numel()

        if total == 0:
            return {}

        return {
            "accuracy": {"value": correct.sum().item() / total, "weight": total},
            "total": {"value": total},
        }

AccuracyMetricConfig dataclass

Bases: MetricConfig

AccuracyMetricConfig(_name: 'str' = 'accuracy', nested_name: 'str | None' = None)

Source code in optimus_dl/modules/metrics/metrics.py 
92
93
94
@dataclass
class AccuracyMetricConfig(MetricConfig):
    _name: str = "accuracy"

LossMetric

Bases: Metric

Simple wrapper to report loss via MetricEngine.

Source code in optimus_dl/modules/metrics/metrics.py 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
@register_metric("loss", LossMetricConfig)
class LossMetric(Metric):
    """Simple wrapper to report loss via MetricEngine."""

    @property
    def requires(self) -> set[str]:
        return {StandardProtocols.LOSS}

    @property
    def accumulators(self) -> dict[str, str]:
        return {"loss": "average"}

    def __call__(
        self, sources_data: dict[str, dict[str, Any]]
    ) -> dict[str, dict[str, Any]]:
        loss = sources_data.get(StandardProtocols.LOSS)
        if loss is None:
            return {}

        weight = 1.0
        classif = sources_data.get(StandardProtocols.CLASSIFICATION)
        if classif and "mask" in classif:
            weight = classif["mask"].sum().item()

        return {
            "loss": {
                "value": loss.item() if hasattr(loss, "item") else loss,
                "weight": weight,
            }
        }

LossMetricConfig dataclass

Bases: MetricConfig

LossMetricConfig(_name: 'str' = 'loss', nested_name: 'str | None' = None)

Source code in optimus_dl/modules/metrics/metrics.py 
173
174
175
@dataclass
class LossMetricConfig(MetricConfig):
    _name: str = "loss"

Metric

Bases: ABC

Stateless definition for computing metrics from model/source data.

A Metric implementation defines: - What data it requires via requires mapping (Role -> set of protocol strings). - How to calculate raw results for a batch, potentially emitting multiple sub-values. - How to finalize those values after they've been aggregated (e.g., F1 from counts).

Source code in optimus_dl/modules/metrics/metrics.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
class Metric(ABC):
    """Stateless definition for computing metrics from model/source data.

    A Metric implementation defines:
    - What data it requires via `requires` mapping (Role -> set of protocol strings).
    - How to calculate raw results for a batch, potentially emitting multiple sub-values.
    - How to finalize those values after they've been aggregated (e.g., F1 from counts).
    """

    def __init__(self, cfg: MetricConfig):
        self.cfg = cfg
        self.nested_name = cfg.nested_name

    @property
    @abstractmethod
    def requires(self) -> set[str]:
        """Mapping from source role name to a set of required protocol strings."""
        raise NotImplementedError

    @property
    def accumulators(self) -> dict[str, str]:
        """Define how each sub-metric should be aggregated across batches.

        Returns a mapping from sub-metric names to accumulator types
        (e.g., 'average', 'sum', 'gather', 'perplexity').
        """
        return {self.cfg._name: "average"}

    @abstractmethod
    def __call__(
        self, sources_data: dict[str, dict[str, Any]]
    ) -> dict[str, dict[str, Any]]:
        """Compute raw metric values for the batch.

        Args:
            batch: The original batch dictionary.
            sources_data: Protocol string -> data.

        Returns:
            Dict mapping sub-metric names to log kwargs (e.g., {'value': ..., 'weight': ...})
            for accumulators.
        """
        raise NotImplementedError

    def finalize(self, aggregated_data: dict[str, Any]) -> dict[str, Any]:
        """Perform final calculations on aggregated data.

        Args:
            aggregated_data: Dict mapping sub-metric names to their
                computed/aggregated values from accumulators.

        Returns:
            Dict of final metrics to be logged/reported.
        """
        return {k: v for k, v in aggregated_data.items() if not k.startswith("_")}

accumulators property

Define how each sub-metric should be aggregated across batches.

Returns a mapping from sub-metric names to accumulator types (e.g., 'average', 'sum', 'gather', 'perplexity').

requires abstractmethod property

Mapping from source role name to a set of required protocol strings.

__call__(sources_data) abstractmethod

Compute raw metric values for the batch.

Parameters:

Name Type Description Default
batch

The original batch dictionary.

 required
sources_data dict[str, dict[str, Any]]

Protocol string -> data.

 required

Returns:

Type Description
dict[str, dict[str, Any]]

Dict mapping sub-metric names to log kwargs (e.g., {'value': ..., 'weight': ...})
dict[str, dict[str, Any]]

for accumulators.
Source code in optimus_dl/modules/metrics/metrics.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
@abstractmethod
def __call__(
    self, sources_data: dict[str, dict[str, Any]]
) -> dict[str, dict[str, Any]]:
    """Compute raw metric values for the batch.

    Args:
        batch: The original batch dictionary.
        sources_data: Protocol string -> data.

    Returns:
        Dict mapping sub-metric names to log kwargs (e.g., {'value': ..., 'weight': ...})
        for accumulators.
    """
    raise NotImplementedError

finalize(aggregated_data)

Perform final calculations on aggregated data.

Parameters:

Name Type Description Default
aggregated_data dict[str, Any]

Dict mapping sub-metric names to their computed/aggregated values from accumulators.

 required

Returns:

Type Description
dict[str, Any]

Dict of final metrics to be logged/reported.
Source code in optimus_dl/modules/metrics/metrics.py 
79
80
81
82
83
84
85
86
87
88
89
def finalize(self, aggregated_data: dict[str, Any]) -> dict[str, Any]:
    """Perform final calculations on aggregated data.

    Args:
        aggregated_data: Dict mapping sub-metric names to their
            computed/aggregated values from accumulators.

    Returns:
        Dict of final metrics to be logged/reported.
    """
    return {k: v for k, v in aggregated_data.items() if not k.startswith("_")}

MetricConfig dataclass

Bases: RegistryConfigStrict

MetricConfig(_name: str | None = None, nested_name: 'str | None' = None)

Parameters:

Name Type Description Default
nested_name str | None

Optional name to nest this metric under in the metrics tree.

 None
Source code in optimus_dl/modules/metrics/metrics.py 
22
23
24
25
26
27
28
29
@dataclass
class MetricConfig(RegistryConfigStrict):
    nested_name: str | None = field(
        default=None,
        metadata={
            "description": "Optional name to nest this metric under in the metrics tree."
        },
    )

PerplexityMetric

Bases: Metric

Computes perplexity (exp(loss)).

Source code in optimus_dl/modules/metrics/metrics.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
@register_metric("perplexity", PerplexityMetricConfig)
class PerplexityMetric(Metric):
    """Computes perplexity (exp(loss))."""

    @property
    def requires(self) -> set[str]:
        return {StandardProtocols.LOSS}

    @property
    def accumulators(self) -> dict[str, str]:
        return {"perplexity": "perplexity"}

    def __call__(
        self, sources_data: dict[str, dict[str, Any]]
    ) -> dict[str, dict[str, Any]]:
        loss = sources_data.get(StandardProtocols.LOSS)
        if loss is None:
            return {}

        weight = 1.0
        classif = sources_data.get(StandardProtocols.CLASSIFICATION)
        if classif and "mask" in classif:
            weight = classif["mask"].sum().item()

        return {
            "perplexity": {
                "value": loss.item() if hasattr(loss, "item") else loss,
                "weight": weight,
            }
        }

PerplexityMetricConfig dataclass

Bases: MetricConfig

PerplexityMetricConfig(_name: 'str' = 'perplexity', nested_name: 'str | None' = None)

Source code in optimus_dl/modules/metrics/metrics.py 
136
137
138
@dataclass
class PerplexityMetricConfig(MetricConfig):
    _name: str = "perplexity"