espnet3.components.metrics.base_metric.BaseMetric

source

class espnet3.components.metrics.base_metric.BaseMetric

Bases: ABC

Base class for metrics that consume inference output paths.

A subclass declares inputs/outputs as tuples of Field. inputs names what the metric reads; the metrics config binds each input to an inference artifact or dataset column, and check_metric_inputs validates that binding. outputs names the result keys, each of kind "number". There is no undeclared fallback: a metric that declares neither raises TypeError as soon as it is instantiated.

A metric whose inputs/outputs never change (most of them - WER, CER, TER) declares them as class attributes. A metric whose contract instead depends on its own configuration (a VERSA-style metric, whose score_config names which measures to compute) sets self.inputs/self.outputs in its own __init__ instead; the instance’s own attributes take priority over the class’s, and the contract is checked once construction finishes (not at class definition, which a configuration-dependent contract cannot satisfy)

• a subclass that does this must call super().__init__() after setting them, so the check sees the instance’s own declaration.

##################

Example

s

A fixed contract (WER-style):

>>> class ExampleMetric(BaseMetric):
...     inputs = (Field("ref", "text"), Field("hyp", "text"))
...     outputs = (Field("score", "number"),)
...     def update(self, **values):
...         pass
...     def compute(self):
...         return {"score": 0.0}
>>> ExampleMetric().inputs[0]
Field(name='ref', kind='text', label='Ref', optional=False, channels=1)

A contract built from the instance’s own configuration (VERSA-style): ref is optional (not every VERSA measure needs a reference), and the output keys follow score_config:

>>> class ExampleVersaMetric(BaseMetric):
...     def __init__(self, score_config):
...         self.score_config = score_config
...         self.inputs = (
...             Field("hyp", "text"),
...             Field("ref", "text", optional=True),
...         )
...         self.outputs = tuple(Field(name, "number") for name in score_config)
...         super().__init__()
...     def update(self, **values):
...         pass
...     def compute(self):
...         return {name: 0.0 for name in self.score_config}
>>> [f.name for f in ExampleVersaMetric(["mcd", "f0"]).outputs]
['mcd', 'f0']

A metric scored against more than one reference (BLEU-style): inputs: does not accept a list of sources for one declared name, so each reference is its own declared input instead; a variable reference count becomes trailing optional fields. Its metrics config binds each one:

inputs:
  hyp: text
  ref1: dataset:text
  ref2: dataset:text_alt

Omitting ref2 entirely is fine, since it is optional.

>>> class ExampleBleuMetric(BaseMetric):
...     inputs = (
...         Field("hyp", "text"),
...         Field("ref1", "text"),
...         Field("ref2", "text", optional=True),
...     )
...     outputs = (Field("score", "number"),)
...     def update(self, **values):
...         pass
...     def compute(self):
...         return {"score": 0.0}
>>> [f.name for f in ExampleBleuMetric().inputs]
['hyp', 'ref1', 'ref2']

A metric that reads more than one audio role (a speaker similarity metric that always needs the target prompt voice and the model’s synthesized audio, but can run without a second reference voice):

inputs:
  target: dataset:prompt_speech
  hyp: wav
  ref: dataset:speech

A dataset: column of kind audio is written as a .wav artifact beside the test set’s SCPs, as measure’s own dataset_artifacts config says:

dataset_artifacts:
  prompt_speech:
    type: wav
    sample_rate: 16000

>>> class ExampleSpeakerMetric(BaseMetric):
...     inputs = (
...         Field("target", "audio"),
...         Field("hyp", "audio"),
...         Field("ref", "audio", optional=True),
...     )
...     outputs = (Field("score", "number"),)
...     def update(self, **values):
...         pass
...     def compute(self):
...         return {"score": 0.0}
>>> [f.name for f in ExampleSpeakerMetric().inputs]
['target', 'hyp', 'ref']

In a training callback (no files): reset, update per batch, gather each rank’s state(), merge, load and compute. Only numbers and lists travel between ranks:

>>> metric.reset()
>>> for batch in batches:
...     metric.update(ref=batch["ref"], hyp=batch["hyp"])
>>> states = all_gather(metric.state())  # one dict per rank
>>> metric.load_state(metric.merge_states(states))
>>> metric.compute()
{'WER': 4.3}

Check this instance’s declared contract.

Reads self.inputs/self.outputs, so it sees a fixed class attribute or an instance attribute a subclass’s own __init__ set before calling this (super().__init__()), whichever the subclass uses.

NOTE

This is a convenience, not the only place the contract is checked: a subclass that overrides __init__ without calling super().__init__() skips it, so measure() calls check_metric_contract again on every metric it instantiates, trusting no declaration this never ran on.

• Raises:TypeError – Neither inputs nor outputs is declared, or the declaration is malformed.

add_state(name: str, default: int | float | List[Any], reduce: Literal['sum', 'cat']) → None

Register a state and set self.<name> to a copy of default.

May be called before or after super().__init__(); the registry is created by the first call.

• Parameters:
  • name – Attribute name of the state.
  • default – The fresh value: an int or float for "sum", a list for "cat". reset() restores a copy.
  • reduce – "sum" or "cat", used by merge_states().
• Raises:ValueError – name is registered or already an attribute, reduce is unknown, or default does not fit it.

##################

Example

s

>>> self.add_state("hits", 0, "sum")
>>> self.add_state("scores", [], "cat")

chunk_size : ClassVar[int] = 1000

Rows per update() call inside __call__().

collect_artifacts : bool = False

True only while __call__() runs with an output_dir. update may keep per-utterance text for write_artifacts() only when it is True; a plain update/compute (a training callback) leaves it False and keeps no text. Not a state: it never appears in state().

abstractmethod compute() → Dict[str, float]

Return the result from the state alone, with no file side effect.

• Returns: One value per declared output name.
• Raises:ValueError – There is nothing to score.

##################

Example

s

>>> metric.compute()
{'WER': 4.3}

inputs : ClassVar[Tuple[Field, ...]]

one Field per SCP input. The name is the key of data (and the default SCP file name); the kind says what the SCP values hold (text: the value itself, audio: a path). A class attribute for a fixed contract; set on self instead, before calling super().__init__(), for one built from the instance’s own configuration.

• Type: What this metric reads

iter_inputs(data: Dict[str, Path], *keys: str) → Iterator[Tuple[str, Dict[str, str]]]

Yield rows from one or more SCP inputs with shared utterance IDs.

This helper reads one or more SCP files in lockstep. For a single key, it behaves as a streaming SCP iterator that returns (utt_id, row) pairs. For multiple keys, it additionally validates that all files contain the same utterance IDs in the same order. It is intended for metrics that want a single streaming API regardless of how many input files they consume.

• Parameters:
  • data (Dict *[*str , Path ]) – Mapping from input aliases to SCP paths.
  • *keys (str) – Input aliases to read together, e.g. ("ref",) or ("ref", "hyp").
• Yields:Iterator[Tuple[str, Dict[str, str]]] – Tuples of (utt_id, row), where row is an alias -> value mapping for that utterance.

  (
      "utt1",
      {"ref": "the cat", "hyp": "the bat"},
  )

• Raises:AssertionError – If no keys are provided, if one file has more entries than another, or if utterance IDs do not match across files.

Example

>>> for utt_id, row in self.iter_inputs(data, "ref", "hyp"):
...     print(utt_id, row["ref"], row["hyp"])
utt1 the cat the bat
utt2 a dog a dog

>>> for utt_id, row in self.iter_inputs(data, "ref"):
...     print(utt_id, row["ref"])
utt1 the cat
utt2 a dog

Notes

• Alignment is checked in file order.
• This helper does not sort utterance IDs.
• Each input is expected to be in SCP format:

  utt1 value
  utt2 another value

load_state(state: Mapping[str, int | float | List[Any]]) → None

Set every registered state from state, such as a merged one.

• Parameters:state – A dict with exactly the registered names.
• Raises:ValueError – state lacks a registered name or has another one.

##################

Example

s

>>> metric.load_state(metric.merge_states(gathered))
>>> metric.compute()
{'WER': 4.3}

merge_states(states: Sequence[Mapping[str, int | float | List[Any]]]) → Dict[str, int | float | List[Any]]

Combine per-rank state dicts by each state’s reduce rule.

Pure: self is not changed. "sum" states are added and "cat" states are concatenated in the order of states. Gathering the dicts across ranks is the caller’s job.

• Parameters:states – One state() dict per rank.
• Returns: The merged dict; the defaults when states is empty.
• Raises:ValueError – A dict lacks a registered name or has another one.

##################

Example

s

>>> metric.merge_states([{"hits": 2}, {"hits": 3}])
{'hits': 5}

outputs : ClassVar[Tuple[Field, ...]]

one Field per key of the result dict, all of kind number. Same class-attribute-or-self choice as inputs.

• Type: What this metric returns

reset() → None

Set every registered state back to a copy of its default.

##################

Example

s

>>> metric.reset()

state() → Dict[str, int | float | List[Any]]

Return a deep copy of every registered state, by name.

• Returns: Plain Python values; {} when nothing is registered.

##################

Example

s

>>> metric.state()
{'hits': 0, 'scores': []}

abstractmethod update(**values: Sequence[Any]) → None

Add one batch to the state.

Gets one list or tuple per declared input name, all of the same length; an optional input may be absent. The batch is checked against the declared inputs before this runs (see __init_subclass__()). Adds counts or appends scores, never rates, and writes no file.

• Parameters:**values – One sequence per declared input name.

##################

Example

s

>>> metric.update(ref=["a b"], hyp=["a c"])

write_artifacts(test_dir: Path) → None

Write files for one test set after compute(); no-op by default.

Runs only from __call__(), and only when it has an output_dir. Override it to write, for example, alignments kept in a "cat" state.

• Parameters:test_dir – output_dir / test_name, already created.

##################

Example

s

>>> metric.write_artifacts(Path("exp/metrics/test-clean"))