@synalinks_export("synalinks.metrics.FBetaScore")
class FBetaScore(Metric):
"""Computes F-Beta score.
Formula:
```python
b2 = beta ** 2
f_beta_score = (1 + b2) * (precision * recall) / (precision * b2 + recall)
```
This is the weighted harmonic mean of precision and recall.
Its output range is `[0, 1]`. It operates at a word level
and can be used for **QA systems**.
If `y_true` and `y_pred` contains multiple fields
The JSON object's fields are flattened and the score
computed for each one independently.
Example:
```python
program.compile(
metrics=[
synalinks.metrics.FBetaScore(),
],
)
```
Args:
average (str): Type of averaging to be performed across per-field results
in the multi-field case.
Acceptable values are `None`, `"micro"`, `"macro"` and
`"weighted"`. Defaults to `None`.
If `None`, no averaging is performed and `result()` will return
the score for each class.
If `"micro"`, compute metrics globally by counting the total
true positives, false negatives and false positives.
If `"macro"`, compute metrics for each label,
and return their unweighted mean.
This does not take label imbalance into account.
If `"weighted"`, compute metrics for each label,
and return their average weighted by support
(the number of true instances for each label).
This alters `"macro"` to account for label imbalance.
It can result in an score that is not between precision and recall.
beta (float): Determines the weight of given to recall
in the harmonic mean between precision and recall (see pseudocode
equation above). Defaults to `1`.
name (str): (Optional) string name of the metric instance.
in_mask (list): (Optional) list of keys to keep to compute the metric.
out_mask (list): (Optional) list of keys to remove to compute the metric.
in_mask_pattern (str): (Optional) Regex pattern; fields whose names match
are kept (combined with ``in_mask`` via OR).
out_mask_pattern (str): (Optional) Regex pattern; fields whose names match
are dropped (combined with ``out_mask`` via OR).
"""
direction = "up"
def __init__(
self,
average=None,
beta=1.0,
name="fbeta_score",
in_mask=None,
out_mask=None,
in_mask_pattern=None,
out_mask_pattern=None,
):
super().__init__(
name=name,
in_mask=in_mask,
out_mask=out_mask,
in_mask_pattern=in_mask_pattern,
out_mask_pattern=out_mask_pattern,
)
if average not in (None, "micro", "macro", "weighted"):
raise ValueError(
"Invalid `average` argument value. Expected one of: "
"[None, 'micro', 'macro', 'weighted']. "
f"Received: average={average}"
)
if not isinstance(beta, float):
raise ValueError(
"Invalid `beta` argument value. "
"It should be a Python float. "
f"Received: beta={beta} of type '{type(beta)}'"
)
self.state = self.add_variable(
data_model=FBetaState,
name="state_" + self.name,
)
self.average = average
self.beta = beta
self.axis = None
if self.average != "micro":
self.axis = 0
# Subclasses (Precision, Recall) override this to switch the result
# formula while reusing TP/FP/FN state and update_state.
self._formula = "fbeta"
async def update_state(self, y_true, y_pred):
y_pred = tree.map_structure(lambda x: ops.convert_to_json_data_model(x), y_pred)
y_true = tree.map_structure(lambda x: ops.convert_to_json_data_model(x), y_true)
if self.in_mask or self.in_mask_pattern:
y_pred = tree.map_structure(
lambda x: (
x.in_mask(mask=self.in_mask, pattern=self.in_mask_pattern)
if x is not None
else x
),
y_pred,
)
y_true = tree.map_structure(
lambda x: (
x.in_mask(mask=self.in_mask, pattern=self.in_mask_pattern)
if x is not None
else x
),
y_true,
)
if self.out_mask or self.out_mask_pattern:
y_pred = tree.map_structure(
lambda x: (
x.out_mask(mask=self.out_mask, pattern=self.out_mask_pattern)
if x is not None
else x
),
y_pred,
)
y_true = tree.map_structure(
lambda x: (
x.out_mask(mask=self.out_mask, pattern=self.out_mask_pattern)
if x is not None
else x
),
y_true,
)
if y_true is None or y_pred is None:
# A failed prediction yields `y_pred is None`; there is nothing to
# compare, so skip the sample instead of calling `.get_json()` on None.
return
y_true = tree.flatten(tree.map_structure(lambda x: str(x), y_true.get_json()))
y_pred = tree.flatten(tree.map_structure(lambda x: str(x), y_pred.get_json()))
true_positives = []
false_positives = []
false_negatives = []
intermediate_weights = []
# For each field of y_true and y_pred. SQuAD-style multiset (Counter)
# intersection, needed so identical strings with repeated tokens
# score 1.0.
# zip_longest, not zip: unmatched leaves (pred and gold structures can
# disagree on variable-length arrays) must be scored; the "" fill has
# no tokens, so they land entirely in false positives/negatives.
for yt, yp in zip_longest(y_true, y_pred, fillvalue=""):
y_true_tokens = nlp_utils.normalize_and_tokenize(str(yt))
y_pred_tokens = nlp_utils.normalize_and_tokenize(str(yp))
num_common = sum((Counter(y_true_tokens) & Counter(y_pred_tokens)).values())
true_positives.append(num_common)
false_positives.append(len(y_pred_tokens) - num_common)
false_negatives.append(len(y_true_tokens) - num_common)
intermediate_weights.append(len(y_true_tokens))
true_positives = np.convert_to_numpy(true_positives)
false_positives = np.convert_to_numpy(false_positives)
false_negatives = np.convert_to_numpy(false_negatives)
intermediate_weights = np.convert_to_numpy(intermediate_weights)
current_true_positives = self.state.get("true_positives")
if current_true_positives:
true_positives = ragged_add(current_true_positives, true_positives)
current_false_positives = self.state.get("false_positives")
if current_false_positives:
false_positives = ragged_add(current_false_positives, false_positives)
current_false_negatives = self.state.get("false_negatives")
if current_false_negatives:
false_negatives = ragged_add(current_false_negatives, false_negatives)
current_intermediate_weights = self.state.get("intermediate_weights")
if current_intermediate_weights:
intermediate_weights = ragged_add(
current_intermediate_weights, intermediate_weights
)
self.state.update(
{
"true_positives": true_positives.tolist(),
"false_positives": false_positives.tolist(),
"false_negatives": false_negatives.tolist(),
"intermediate_weights": intermediate_weights.tolist(),
}
)
def result(self):
if (
self.state.get("true_positives") is None
and self.state.get("false_positives") is None
and self.state.get("false_negatives") is None
):
return 0.0
tp = np.convert_to_tensor(self.state.get("true_positives"))
fp = np.convert_to_tensor(self.state.get("false_positives"))
fn = np.convert_to_tensor(self.state.get("false_negatives"))
# Keras/sklearn "micro": aggregate TP/FP/FN across all fields first,
# *then* compute precision/recall. Without this collapse, "micro"
# would degenerate to a mean over per-field scores (i.e. macro).
if self.average == "micro":
tp = np.sum(tp)
fp = np.sum(fp)
fn = np.sum(fn)
precision = np.convert_to_tensor(
np.divide(tp, np.add(tp, fp) + backend.epsilon())
)
recall = np.convert_to_tensor(np.divide(tp, np.add(tp, fn) + backend.epsilon()))
formula = getattr(self, "_formula", "fbeta")
if formula == "precision":
score = precision
elif formula == "recall":
score = recall
else:
mul_value = precision * recall
add_value = ((self.beta**2) * precision) + recall
mean = np.divide(mul_value, add_value + backend.epsilon())
score = mean * (1 + (self.beta**2))
return self._aggregate(score)
def _aggregate(self, score):
"""Apply `average` reduction over per-field scores."""
score = np.convert_to_tensor(score)
if self.average == "weighted":
intermediate_weights = self.state.get("intermediate_weights")
weights = np.divide(
intermediate_weights,
np.sum(intermediate_weights) + backend.epsilon(),
)
score = np.sum(score * weights)
elif self.average is not None: # [micro, macro]
score = np.mean(score, self.axis)
# numpy 1.25+ deprecates float() on a >0-D array even when size == 1,
# so go through .item() / .tolist() to always hand back Python scalars.
score_arr = np.convert_to_numpy(score)
if score_arr.size == 1:
return float(score_arr.item())
return [float(v) for v in score_arr.tolist()]
def get_config(self):
"""Return the serializable config of the metric.
Returns:
(dict): The config dict.
"""
config = {
"name": self.name,
"average": self.average,
"beta": self.beta,
}
base_config = super().get_config()
return {**base_config, **config}