Source code for cleanlab.multiannotator

# Copyright (C) 2017-2022  Cleanlab Inc.
# This file is part of cleanlab.
#
# cleanlab is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# cleanlab is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with cleanlab.  If not, see <https://www.gnu.org/licenses/>.

"""
Methods for analysis of classification data labeled by multiple annotators, including computation of:

* A consensus label for each example that aggregates the individual annotations more accurately than alternative aggregation via majority-vote or other algorithms used in crowdsourcing.
* A quality score for each consensus label which measures our confidence that this label is correct.
* An analogous label quality score for each individual label chosen by one annotator for a particular example.
* An overall quality score for each annotator which measures our confidence in the overall correctness of labels obtained from this annotator.

The underlying algorithms are described in `the CROWDLAB paper <https://arxiv.org/abs/2210.06812>`_.
"""

import warnings
import numpy as np
import pandas as pd

from typing import List, Dict, Any, Union, Tuple, Optional

from cleanlab.rank import get_label_quality_scores
from cleanlab.internal.util import get_num_classes, value_counts
from cleanlab.internal.multiannotator_utils import assert_valid_inputs_multiannotator


[docs]def get_label_quality_multiannotator( labels_multiannotator: Union[pd.DataFrame, np.ndarray], pred_probs: np.ndarray, *, consensus_method: Union[str, List[str]] = "best_quality", quality_method: str = "crowdlab", return_detailed_quality: bool = True, return_annotator_stats: bool = True, verbose: bool = True, label_quality_score_kwargs: dict = {}, ) -> Dict[str, pd.DataFrame]: """Returns label quality scores for each example and for each annotator. This function is for multiclass classification datasets where examples have been labeled by multiple annotators (not necessarily the same number of annotators per example). It computes one consensus label for each example that best accounts for the labels chosen by each annotator (and their quality), as well as a score for how confident we are that this consensus label is actually correct. It also computes the the quality scores for each annotator's individual labels, and the quality of each annotator. The score is between 0 and 1; lower scores indicate labels less likely to be correct. For example: * 1 - clean label (the given label is likely correct). * 0 - dirty label (the given label is unlikely correct). Parameters ---------- labels_multiannotator : pd.DataFrame of np.ndarray 2D pandas DataFrame or array of multiple given labels for each example with shape ``(N, M)``, where N is the number of examples and M is the number of annotators. ``labels_multiannotator[n][m]`` = label for n-th example given by m-th annotator. For a dataset with K classes, each given label must be an integer in 0, 1, ..., K-1 or ``NaN`` if this annotator did not label a particular example. If you have string or other differently formatted labels, you can convert them to the proper format using :py:func:`format_multiannotator_labels <cleanlab.internal.multiannotator_utils.format_multiannotator_labels>`. If pd.DataFrame, column names should correspond to each annotator's ID. pred_probs : np.ndarray An array of shape ``(N, K)`` of predicted class probabilities from a trained classifier model. Predicted probabilities in the same format expected by the :py:func:`get_label_quality_scores <cleanlab.rank.get_label_quality_scores>`. consensus_method : str or List[str], default = "majority_vote" Specifies the method used to aggregate labels from multiple annotators into a single consensus label. Options include: * ``majority_vote``: consensus obtained using a simple majority vote among annotators, with ties broken via ``pred_probs``. * ``best_quality``: consensus obtained by selecting the label with highest label quality (quality determined by method specified in ``quality_method``). A List may be passed if you want to consider multiple methods for producing consensus labels. If a List is passed, then the 0th element of the list is the method used to produce columns `consensus_label`, `consensus_quality_score`, `annotator_agreement` in the returned DataFrame. The remaning (1st, 2nd, 3rd, etc.) elements of this list are output as extra columns in the returned pandas DataFrame with names formatted as: `consensus_label_SUFFIX`, `consensus_quality_score_SUFFIX` where `SUFFIX` = each element of this list, which must correspond to a valid method for computing consensus labels. quality_method : str, default = "crowdlab" Specifies the method used to calculate the quality of the consensus label. Options include: * ``crowdlab``: an emsemble method that weighs both the annotators' labels as well as the model's prediction. * ``agreement``: the fraction of annotators that agree with the consensus label. return_detailed_quality: bool, default = True Boolean to specify if `detailed_label_quality` is returned. return_annotator_stats : bool, default = True Boolean to specify if `annotator_stats` is returned. verbose : bool, default = True Important warnings and other printed statements may be suppressed if ``verbose`` is set to ``False``. label_quality_score_kwargs : dict, optional Keyword arguments to pass into :py:func:`get_label_quality_scores <cleanlab.rank.get_label_quality_scores>`. Returns ------- labels_info : dict Dictionary containing up to 3 pandas DataFrame with keys as below: ``label_quality`` : pandas.DataFrame pandas DataFrame in which each row corresponds to one example, with columns: * ``num_annotations``: the number of annotators that have labeled each example. * ``consensus_label``: the single label that is best for each example (you can control how it is derived from all annotators' labels via the argument: ``consensus_method``). * ``annotator_agreement``: the fraction of annotators that agree with the consensus label (only consider the annotators that labeled that particular example). * ``consensus_quality_score``: label quality score for consensus label, calculated by the method specified in ``quality_method``. ``detailed_label_quality`` : pandas.DataFrame (returned if `return_detailed_quality=True`) Returns a pandas DataFrame with columns `quality_annotator_1`, `quality_annotator_2`, ..., `quality_annotator_M` where each entry is the label quality score for the labels provided by each annotator (is ``NaN`` for examples which this annotator did not label). ``annotator_stats`` : pandas.DataFrame (returned if `return_annotator_stats=True`) Returns overall statistics about each annotator, sorted by lowest annotator_quality first. pandas DataFrame in which each row corresponds to one annotator (the row IDs correspond to annotator IDs), with columns: * ``annotator_quality``: overall quality of a given annotator's labels, calculated by the method specified in ``quality_method``. * ``num_examples_labeled``: number of examples annotated by a given annotator. * ``agreement_with_consensus``: fraction of examples where a given annotator agrees with the consensus label. * ``worst_class``: the class that is most frequently mislabeled by a given annotator. """ if isinstance(labels_multiannotator, np.ndarray): labels_multiannotator = pd.DataFrame(labels_multiannotator) assert_valid_inputs_multiannotator(labels_multiannotator, pred_probs) # Count number of non-NaN values for each example num_annotations = labels_multiannotator.count(axis=1).to_numpy() if not isinstance(consensus_method, list): consensus_method = [consensus_method] if "best_quality" in consensus_method or "majority_vote" in consensus_method: majority_vote_label = get_majority_vote_label( labels_multiannotator=labels_multiannotator, pred_probs=pred_probs, verbose=False, ) ( MV_annotator_agreement, MV_consensus_quality_score, MV_post_pred_probs, MV_model_weight, MV_annotator_weight, ) = _get_consensus_stats( labels_multiannotator=labels_multiannotator, pred_probs=pred_probs, num_annotations=num_annotations, consensus_label=majority_vote_label, quality_method=quality_method, verbose=verbose, label_quality_score_kwargs=label_quality_score_kwargs, ) label_quality = pd.DataFrame({"num_annotations": num_annotations}) valid_methods = ["majority_vote", "best_quality"] main_method = True for curr_method in consensus_method: # geting consensus label and stats if curr_method == "majority_vote": consensus_label = majority_vote_label annotator_agreement = MV_annotator_agreement consensus_quality_score = MV_consensus_quality_score post_pred_probs = MV_post_pred_probs model_weight = MV_model_weight annotator_weight = MV_annotator_weight elif curr_method == "best_quality": consensus_label = np.full(len(majority_vote_label), np.nan) for i in range(len(consensus_label)): max_pred_probs_ind = np.where( MV_post_pred_probs[i] == np.max(MV_post_pred_probs[i]) )[0] if len(max_pred_probs_ind) == 1: consensus_label[i] = max_pred_probs_ind[0] else: consensus_label[i] = majority_vote_label[i] consensus_label = consensus_label.astype("int64") # convert all label types to int ( annotator_agreement, consensus_quality_score, post_pred_probs, model_weight, annotator_weight, ) = _get_consensus_stats( labels_multiannotator=labels_multiannotator, pred_probs=pred_probs, num_annotations=num_annotations, consensus_label=consensus_label, quality_method=quality_method, verbose=verbose, label_quality_score_kwargs=label_quality_score_kwargs, ) else: raise ValueError( f""" {curr_method} is not a valid consensus method! Please choose a valid consensus_method: {valid_methods} """ ) if verbose: # check if any classes no longer appear in the set of consensus labels _check_consensus_label_classes( labels_multiannotator=labels_multiannotator, consensus_label=consensus_label, consensus_method=curr_method, ) # saving stats into dataframe, computing additional stats if specified if main_method: ( label_quality["consensus_label"], label_quality["consensus_quality_score"], label_quality["annotator_agreement"], ) = ( consensus_label, consensus_quality_score, annotator_agreement, ) label_quality = label_quality.reindex( columns=[ "consensus_label", "consensus_quality_score", "annotator_agreement", "num_annotations", ] ) if return_detailed_quality: # Compute the label quality scores for each annotators' labels detailed_label_quality = labels_multiannotator.apply( _get_annotator_label_quality_score, pred_probs=post_pred_probs, label_quality_score_kwargs=label_quality_score_kwargs, ) detailed_label_quality = detailed_label_quality.add_prefix("quality_annotator_") if return_annotator_stats: annotator_stats = _get_annotator_stats( labels_multiannotator=labels_multiannotator, pred_probs=post_pred_probs, consensus_label=consensus_label, num_annotations=num_annotations, annotator_agreement=annotator_agreement, model_weight=model_weight, annotator_weight=annotator_weight, consensus_quality_score=consensus_quality_score, quality_method=quality_method, ) main_method = False else: ( label_quality[f"consensus_label_{curr_method}"], label_quality[f"consensus_quality_score_{curr_method}"], label_quality[f"annotator_agreement_{curr_method}"], ) = ( consensus_label, consensus_quality_score, annotator_agreement, ) if return_detailed_quality and return_annotator_stats: return { "label_quality": label_quality, "detailed_label_quality": detailed_label_quality, "annotator_stats": annotator_stats, } elif return_detailed_quality: return { "label_quality": label_quality, "detailed_label_quality": detailed_label_quality, } elif return_annotator_stats: return { "label_quality": label_quality, "annotator_stats": annotator_stats, } else: return { "label_quality": label_quality, }
[docs]def get_majority_vote_label( labels_multiannotator: Union[pd.DataFrame, np.ndarray], pred_probs: Optional[np.ndarray] = None, verbose: bool = True, ) -> np.ndarray: """Returns the majority vote label for each example, aggregated from the labels given by multiple annotators. Parameters ---------- labels_multiannotator : pd.DataFrame or np.ndarray 2D pandas DataFrame or array of multiple given labels for each example with shape ``(N, M)``, where N is the number of examples and M is the number of annotators. For more details, labels in the same format expected by the :py:func:`get_label_quality_multiannotator <cleanlab.multiannotator.get_label_quality_multiannotator>`. pred_probs : np.ndarray, optional An array of shape ``(N, K)`` of model-predicted probabilities, ``P(label=k|x)``. For details, predicted probabilities in the same format expected by the :py:func:`get_label_quality_multiannotator <cleanlab.multiannotator.get_label_quality_multiannotator>`. verbose : bool, optional Important warnings and other printed statements may be suppressed if ``verbose`` is set to ``False``. Returns ------- consensus_label: np.ndarray An array of shape ``(N,)`` with the majority vote label aggregated from all annotators. In the event of majority vote ties, ties are broken in the following order: using the model ``pred_probs`` (if provided) and selecting the class with highest predicted probability, using the empirical class frequencies and selecting the class with highest frequency, using an initial annotator quality score and selecting the class that has been labeled by annotators with higher quality, and lastly by random selection. """ if isinstance(labels_multiannotator, np.ndarray): labels_multiannotator = pd.DataFrame(labels_multiannotator) assert_valid_inputs_multiannotator(labels_multiannotator, pred_probs) majority_vote_label = np.full(len(labels_multiannotator), np.nan) mode_labels_multiannotator = labels_multiannotator.mode(axis=1) nontied_idx = [] tied_idx = dict() # obtaining consensus using annotator majority vote for idx in range(len(mode_labels_multiannotator)): label_mode = mode_labels_multiannotator.iloc[idx].dropna().astype(int).to_numpy() if len(label_mode) == 1: majority_vote_label[idx] = label_mode[0] nontied_idx.append(idx) else: tied_idx[idx] = label_mode # tiebreak 1: using pred_probs (if provided) if pred_probs is not None and len(tied_idx) > 0: for idx, label_mode in tied_idx.copy().items(): max_pred_probs = np.where( pred_probs[idx, label_mode] == np.max(pred_probs[idx, label_mode]) )[0] if len(max_pred_probs) == 1: majority_vote_label[idx] = label_mode[max_pred_probs[0]] del tied_idx[idx] else: tied_idx[idx] = label_mode[max_pred_probs] # tiebreak 2: using empirical class frequencies if len(tied_idx) > 0: if pred_probs is not None: num_classes = pred_probs.shape[1] else: num_classes = int( np.nanmax(labels_multiannotator.replace({pd.NA: np.NaN}).astype(float).values) + 1 ) class_frequencies = labels_multiannotator.apply( lambda s: pd.Series(np.bincount(s[s.notna()], minlength=num_classes)), axis=1 ).sum() for idx, label_mode in tied_idx.copy().items(): max_frequency = np.where( class_frequencies[label_mode] == np.max(class_frequencies[label_mode]) )[0] if len(max_frequency) == 1: majority_vote_label[idx] = label_mode[max_frequency[0]] del tied_idx[idx] else: tied_idx[idx] = label_mode[max_frequency] # tiebreak 3: using initial annotator quality scores if len(tied_idx) > 0: nontied_majority_vote_label = majority_vote_label[nontied_idx] nontied_labels_multiannotator = labels_multiannotator.iloc[nontied_idx] annotator_agreement_with_consensus = nontied_labels_multiannotator.apply( lambda s: np.mean(s[pd.notna(s)] == nontied_majority_vote_label[pd.notna(s)]), axis=0, ).to_numpy() for idx, label_mode in tied_idx.copy().items(): label_quality_score = np.array( [ np.mean( annotator_agreement_with_consensus[ labels_multiannotator.iloc[idx][ labels_multiannotator.iloc[idx] == label ].index.values ] ) for label in label_mode ] ) max_score = np.where(label_quality_score == label_quality_score.max())[0] if len(max_score) == 1: majority_vote_label[idx] = label_mode[max_score[0]] del tied_idx[idx] else: tied_idx[idx] = label_mode[max_score] # if still tied, break by random selection if len(tied_idx) > 0: warnings.warn( f"breaking ties of examples {list(tied_idx.keys())} by random selection, you may want to set seed for reproducability" ) for idx, label_mode in tied_idx.items(): majority_vote_label[idx] = np.random.choice(label_mode) if verbose: # check if any classes no longer appear in the set of consensus labels _check_consensus_label_classes( labels_multiannotator=labels_multiannotator, consensus_label=majority_vote_label, consensus_method="majority_vote", ) return majority_vote_label.astype("int64")
[docs]def convert_long_to_wide_dataset( labels_multiannotator_long: pd.DataFrame, ) -> pd.DataFrame: """Converts a long format dataset to wide format which is suitable for passing into :py:func:`get_label_quality_multiannotator <cleanlab.multiannotator.get_label_quality_multiannotator>`. Dataframe must contain three columns named: #. ``task`` representing each example labeled by the annotators #. ``annotator`` representing each annotator #. ``label`` representing the label given by an annotator for the corresponding task (i.e. example) Parameters ---------- labels_multiannotator_long : pd.DataFrame pandas DataFrame in long format with three columns named ``task``, ``annotator`` and ``label`` Returns ------- labels_multiannotator_wide : pd.DataFrame pandas DataFrame of the proper format to be passed as ``labels_multiannotator`` for the other ``cleanlab.multiannotator`` functions. """ labels_multiannotator_wide = labels_multiannotator_long.pivot( index="task", columns="annotator", values="label" ) labels_multiannotator_wide.index.name = None labels_multiannotator_wide.columns.name = None return labels_multiannotator_wide
def _get_consensus_stats( labels_multiannotator: pd.DataFrame, pred_probs: np.ndarray, num_annotations: np.ndarray, consensus_label: np.ndarray, quality_method: str = "crowdlab", verbose: bool = True, label_quality_score_kwargs: dict = {}, ) -> tuple: """Returns a tuple containing the consensus labels, annotator agreement scores, and quality of consensus Parameters ---------- labels_multiannotator : pd.DataFrame 2D pandas DataFrame of multiple given labels for each example with shape ``(N, M)``, where N is the number of examples and M is the number of annotators. For more details, labels in the same format expected by the :py:func:`get_label_quality_multiannotator <cleanlab.multiannotator.get_label_quality_multiannotator>`. pred_probs : np.ndarray An array of shape ``(N, K)`` of model-predicted probabilities, ``P(label=k|x)``. For details, predicted probabilities in the same format expected by the :py:func:`get_label_quality_multiannotator <cleanlab.multiannotator.get_label_quality_multiannotator>`. num_annotations : np.ndarray An array of shape ``(N,)`` with the number of annotators that have labeled each example. consensus_label : np.ndarray An array of shape ``(N,)`` with the consensus labels aggregated from all annotators. quality_method : str, default = "crowdlab" (Options: ["crowdlab", "agreement"]) Specifies the method used to calculate the quality of the consensus label. For valid quality methods, view :py:func:`get_label_quality_multiannotator <cleanlab.multiannotator.get_label_quality_multiannotator>` label_quality_score_kwargs : dict, optional Keyword arguments to pass into ``get_label_quality_scores()``. verbose : bool, default = True Certain warnings and notes will be printed if ``verbose`` is set to ``True``. Returns ------ stats : tuple A tuple of (consensus_label, annotator_agreement, consensus_quality_score, post_pred_probs). """ # compute the fraction of annotator agreeing with the consensus labels annotator_agreement = _get_annotator_agreement_with_consensus( labels_multiannotator=labels_multiannotator, consensus_label=consensus_label, ) # compute posterior predicted probabilites post_pred_probs, model_weight, annotator_weight = _get_post_pred_probs_and_weights( labels_multiannotator=labels_multiannotator, consensus_label=consensus_label, prior_pred_probs=pred_probs, num_annotations=num_annotations, annotator_agreement=annotator_agreement, quality_method=quality_method, verbose=verbose, ) # compute quality of the consensus labels consensus_quality_score = _get_consensus_quality_score( consensus_label=consensus_label, pred_probs=post_pred_probs, num_annotations=num_annotations, annotator_agreement=annotator_agreement, quality_method=quality_method, label_quality_score_kwargs=label_quality_score_kwargs, ) return ( annotator_agreement, consensus_quality_score, post_pred_probs, model_weight, annotator_weight, ) def _get_annotator_stats( labels_multiannotator: pd.DataFrame, pred_probs: np.ndarray, consensus_label: np.ndarray, num_annotations: np.ndarray, annotator_agreement: np.ndarray, model_weight: np.ndarray, annotator_weight: np.ndarray, consensus_quality_score: np.ndarray, quality_method: str = "crowdlab", ) -> pd.DataFrame: """Returns a dictionary containing overall statistics about each annotator. Parameters ---------- labels_multiannotator : pd.DataFrame 2D pandas DataFrame of multiple given labels for each example with shape ``(N, M)``, where N is the number of examples and M is the number of annotators. For more details, labels in the same format expected by the :py:func:`get_label_quality_multiannotator <cleanlab.multiannotator.get_label_quality_multiannotator>`. pred_probs : np.ndarray An array of shape ``(N, K)`` of model-predicted probabilities, ``P(label=k|x)``. For details, predicted probabilities in the same format expected by the :py:func:`get_label_quality_multiannotator <cleanlab.multiannotator.get_label_quality_multiannotator>`. consensus_label : np.ndarray An array of shape ``(N,)`` with the consensus labels aggregated from all annotators. num_annotations : np.ndarray An array of shape ``(N,)`` with the number of annotators that have labeled each example. annotator_agreement : np.ndarray An array of shape ``(N,)`` with the fraction of annotators that agree with each consensus label. model_weight : float float specifying the model weight used in weighted averages, None if model weight is not used to compute quality scores annotator_weight : np.ndarray An array of shape ``(M,)`` where M is the number of annotators, specifying the annotator weights used in weighted averages, None if annotator weights are not used to compute quality scores consensus_quality_score : np.ndarray An array of shape ``(N,)`` with the quality score of the consensus. quality_method : str, default = "crowdlab" (Options: ["crowdlab", "agreement"]) Specifies the method used to calculate the quality of the consensus label. For valid quality methods, view :py:func:`get_label_quality_multiannotator <cleanlab.multiannotator.get_label_quality_multiannotator>` Returns ------- annotator_stats : pd.DataFrame Overall statistics about each annotator. For details, see the documentation of :py:func:`get_label_quality_multiannotator <cleanlab.multiannotator.get_label_quality_multiannotator>`. """ annotator_quality = _get_annotator_quality( labels_multiannotator=labels_multiannotator, pred_probs=pred_probs, consensus_label=consensus_label, num_annotations=num_annotations, annotator_agreement=annotator_agreement, model_weight=model_weight, annotator_weight=annotator_weight, quality_method=quality_method, ) # Compute the number of labels labeled/ by each annotator num_examples_labeled = labels_multiannotator.count() # Compute the fraction of labels annotated by each annotator that agrees with the consensus label # TODO: check if we should drop singleton labels here agreement_with_consensus = labels_multiannotator.apply( lambda s: np.mean(s[pd.notna(s)] == consensus_label[pd.notna(s)]), axis=0, ).to_numpy() # Find the worst labeled class for each annotator worst_class = _get_annotator_worst_class( labels_multiannotator=labels_multiannotator, consensus_label=consensus_label, consensus_quality_score=consensus_quality_score, ) # Create multi-annotator stats DataFrame from its columns annotator_stats = pd.DataFrame( { "annotator_quality": annotator_quality, "agreement_with_consensus": agreement_with_consensus, "worst_class": worst_class, "num_examples_labeled": num_examples_labeled, } ) return annotator_stats.sort_values(by=["annotator_quality", "agreement_with_consensus"]) def _get_annotator_agreement_with_consensus( labels_multiannotator: pd.DataFrame, consensus_label: np.ndarray, ) -> np.ndarray: """Returns the fractions of annotators that agree with the consensus label per example. Note that the fraction for each example only considers the annotators that labeled that particular example. Parameters ---------- labels_multiannotator : pd.DataFrame 2D pandas DataFrame of multiple given labels for each example with shape ``(N, M)``, where N is the number of examples and M is the number of annotators. For more details, labels in the same format expected by the :py:func:`get_label_quality_multiannotator <cleanlab.multiannotator.get_label_quality_multiannotator>`. consensus_label : np.ndarray An array of shape ``(N,)`` with the consensus labels aggregated from all annotators. Returns ------- annotator_agreement : np.ndarray An array of shape ``(N,)`` with the fraction of annotators that agree with each consensus label. """ annotator_agreement = labels_multiannotator.assign(consensus_label=consensus_label).apply( lambda s: np.mean(s.drop("consensus_label").dropna() == s["consensus_label"]), axis=1, ) return annotator_agreement.to_numpy() def _get_annotator_agreement_with_annotators( labels_multiannotator: pd.DataFrame, num_annotations: np.ndarray, verbose: bool = True, ) -> np.ndarray: """Returns the average agreement of each annotator with other annotators that label the same example. Parameters ---------- labels_multiannotator : pd.DataFrame 2D pandas DataFrame of multiple given labels for each example with shape ``(N, M)``, where N is the number of examples and M is the number of annotators. For more details, labels in the same format expected by the :py:func:`get_label_quality_multiannotator <cleanlab.multiannotator.get_label_quality_multiannotator>`. consensus_label : np.ndarray An array of shape ``(N,)`` with the consensus labels aggregated from all annotators. verbose : bool, default = True Certain warnings and notes will be printed if ``verbose`` is set to ``True``. Returns ------- annotator_agreement : np.ndarray An array of shape ``(M,)`` where M is the number of annotators, with the agreement of each annotator with other annotators that labeled the same examples. """ def get_single_annotator_agreement( labels_multiannotator: pd.DataFrame, num_annotations: np.ndarray, annotator_id: Union[int, str], ): annotator_agreement_per_example = labels_multiannotator.apply( lambda s: np.mean(s[pd.notna(s)].drop(annotator_id) == s[annotator_id]), axis=1 ) np.nan_to_num(annotator_agreement_per_example, copy=False, nan=0) try: annotator_agreement = np.average( annotator_agreement_per_example, weights=num_annotations - 1 ) except: annotator_agreement = np.NaN return annotator_agreement annotator_agreement_with_annotators = labels_multiannotator.apply( lambda s: get_single_annotator_agreement( labels_multiannotator[pd.notna(s)], num_annotations[pd.notna(s)], s.name ) ) # impute average annotator accuracy for any annotator that do not overlap with other annotators mask = annotator_agreement_with_annotators.isna() if np.sum(mask) > 0: if verbose: print( f"Annotator(s) {annotator_agreement_with_annotators[mask].index.values} did not annotate any examples that overlap with other annotators, \ \nusing the average annotator agreeement among other annotators as this annotator's agreement." ) avg_annotator_agreement = np.mean(annotator_agreement_with_annotators[~mask]) annotator_agreement_with_annotators[mask] = avg_annotator_agreement return annotator_agreement_with_annotators.to_numpy() def _get_post_pred_probs_and_weights( labels_multiannotator: pd.DataFrame, consensus_label: np.ndarray, prior_pred_probs: np.ndarray, num_annotations: np.ndarray, annotator_agreement: np.ndarray, quality_method: str = "crowdlab", verbose: bool = True, ) -> Tuple[np.ndarray, Any, Any]: """Return the posterior predicted probabilites of each example given a specified quality method. Parameters ---------- labels_multiannotator : pd.DataFrame 2D pandas DataFrame of multiple given labels for each example with shape ``(N, M)``, where N is the number of examples and M is the number of annotators. For more details, labels in the same format expected by the :py:func:`get_label_quality_multiannotator <cleanlab.multiannotator.get_label_quality_multiannotator>`. consensus_label : np.ndarray An array of shape ``(N,)`` with the consensus labels aggregated from all annotators. prior_pred_probs : np.ndarray An array of shape ``(N, K)`` of prior predicted probabilities, ``P(label=k|x)``, usually the out-of-sample predicted probability computed by a model. For details, predicted probabilities in the same format expected by the :py:func:`get_label_quality_multiannotator <cleanlab.multiannotator.get_label_quality_multiannotator>`. num_annotations : np.ndarray An array of shape ``(N,)`` with the number of annotators that have labeled each example. annotator_agreement : np.ndarray An array of shape ``(N,)`` with the fraction of annotators that agree with each consensus label. quality_method : str, default = "crowdlab" (Options: ["crowdlab", "agreement"]) Specifies the method used to calculate the quality of the consensus label. For valid quality methods, view :py:func:`get_label_quality_multiannotator <cleanlab.multiannotator.get_label_quality_multiannotator>` verbose : bool, default = True Certain warnings and notes will be printed if ``verbose`` is set to ``True``. Returns ------- post_pred_probs : np.ndarray An array of shape ``(N, K)`` with the posterior predicted probabilities. model_weight : float float specifying the model weight used in weighted averages, None if model weight is not used to compute quality scores annotator_weight : np.ndarray An array of shape ``(M,)`` where M is the number of annotators, specifying the annotator weights used in weighted averages, None if annotator weights are not used to compute quality scores """ valid_methods = [ "crowdlab", "agreement", ] # setting dummy variables for model and annotator weights that will be returned # only relevant for quality_method == crowdlab, return None for all other methods return_model_weight = None return_annotator_weight = None if quality_method == "crowdlab": num_classes = get_num_classes(pred_probs=prior_pred_probs) likelihood = np.mean( annotator_agreement[num_annotations != 1] ) # likelihood that any annotator will annotate the consensus label for any example # subsetting the dataset to only includes examples with more than one annotation mask = num_annotations != 1 consensus_label_subset = consensus_label[mask] prior_pred_probs_subset = prior_pred_probs[mask] # compute most likely class error most_likely_class_error = np.clip( np.mean( consensus_label_subset != np.argmax(np.bincount(consensus_label_subset, minlength=num_classes)) ), a_min=1e-6, a_max=None, ) # compute adjusted annotator agreement (used as annotator weights) annotator_agreement_with_annotators = _get_annotator_agreement_with_annotators( labels_multiannotator, num_annotations, verbose ) annotator_error = 1 - annotator_agreement_with_annotators adjusted_annotator_agreement = np.clip( 1 - (annotator_error / most_likely_class_error), a_min=1e-6, a_max=None ) # compute model weight model_error = np.mean(np.argmax(prior_pred_probs_subset, axis=1) != consensus_label_subset) model_weight = np.max([(1 - (model_error / most_likely_class_error)), 1e-6]) * np.sqrt( np.mean(num_annotations) ) # compute weighted average post_pred_probs = np.full(prior_pred_probs.shape, np.nan) for i in range(len(labels_multiannotator)): example_mask = labels_multiannotator.iloc[i].notna() example = labels_multiannotator.iloc[i][example_mask] post_pred_probs[i] = [ np.average( [prior_pred_probs[i, true_label]] + [ likelihood if annotator_label == true_label else (1 - likelihood) / (num_classes - 1) for annotator_label in example ], weights=np.concatenate( ([model_weight], adjusted_annotator_agreement[example_mask]) ), ) for true_label in range(num_classes) ] return_model_weight = model_weight return_annotator_weight = adjusted_annotator_agreement elif quality_method == "agreement": num_classes = get_num_classes(pred_probs=prior_pred_probs) label_counts = np.full((len(labels_multiannotator), num_classes), np.NaN) for i in range(len(labels_multiannotator)): s = labels_multiannotator.iloc[i] label_counts[i, :] = value_counts(s[s.notna()], num_classes=num_classes) post_pred_probs = label_counts / num_annotations.reshape(-1, 1) else: raise ValueError( f""" {quality_method} is not a valid quality method! Please choose a valid quality_method: {valid_methods} """ ) return post_pred_probs, return_model_weight, return_annotator_weight def _get_consensus_quality_score( consensus_label: np.ndarray, pred_probs: np.ndarray, num_annotations: np.ndarray, annotator_agreement: np.ndarray, quality_method: str = "crowdlab", label_quality_score_kwargs: dict = {}, ) -> np.ndarray: """Return scores representing quality of the consensus label for each example. Parameters ---------- labels_multiannotator : pd.DataFrame 2D pandas DataFrame of multiple given labels for each example with shape ``(N, M)``, where N is the number of examples and M is the number of annotators. For more details, labels in the same format expected by the :py:func:`get_label_quality_multiannotator <cleanlab.multiannotator.get_label_quality_multiannotator>`. consensus_label : np.ndarray An array of shape ``(N,)`` with the consensus labels aggregated from all annotators. pred_probs : np.ndarray An array of shape ``(N, K)`` of posterior predicted probabilities, ``P(label=k|x)``. For details, predicted probabilities in the same format expected by the :py:func:`get_label_quality_multiannotator <cleanlab.multiannotator.get_label_quality_multiannotator>`. num_annotations : np.ndarray An array of shape ``(N,)`` with the number of annotators that have labeled each example. annotator_agreement : np.ndarray An array of shape ``(N,)`` with the fraction of annotators that agree with each consensus label. quality_method : str, default = "crowdlab" (Options: ["crowdlab", "agreement"]) Specifies the method used to calculate the quality of the consensus label. For valid quality methods, view :py:func:`get_label_quality_multiannotator <cleanlab.multiannotator.get_label_quality_multiannotator>` Returns ------- consensus_quality_score : np.ndarray An array of shape ``(N,)`` with the quality score of the consensus. """ valid_methods = [ "crowdlab", "agreement", ] if quality_method == "crowdlab": consensus_quality_score = get_label_quality_scores( consensus_label, pred_probs, **label_quality_score_kwargs ) elif quality_method == "agreement": consensus_quality_score = annotator_agreement else: raise ValueError( f""" {quality_method} is not a valid consensus quality method! Please choose a valid quality_method: {valid_methods} """ ) return consensus_quality_score def _get_annotator_label_quality_score( annotator_label: pd.Series, pred_probs: np.ndarray, label_quality_score_kwargs: dict = {}, ) -> pd.Series: """Returns quality scores for each datapoint. Very similar functionality as ``_get_consensus_quality_score`` with additional support for annotator labels that contain NaN values. For more info about parameters and returns, see the docstring of :py:func:`_get_consensus_quality_score <cleanlab.multiannotator._get_consensus_quality_score>`. """ mask = pd.notna(annotator_label) annotator_label_quality_score_subset = get_label_quality_scores( labels=annotator_label[mask].to_numpy().astype("int64"), pred_probs=pred_probs[mask], **label_quality_score_kwargs, ) annotator_label_quality_score = np.full(len(annotator_label), np.nan) annotator_label_quality_score[mask] = annotator_label_quality_score_subset return pd.Series(annotator_label_quality_score) def _get_annotator_quality( labels_multiannotator: pd.DataFrame, pred_probs: np.ndarray, consensus_label: np.ndarray, num_annotations: np.ndarray, annotator_agreement: np.ndarray, model_weight: np.ndarray, annotator_weight: np.ndarray, quality_method: str = "crowdlab", ) -> pd.DataFrame: """Returns annotator quality score for each annotator. Parameters ---------- labels_multiannotator : pd.DataFrame 2D pandas DataFrame of multiple given labels for each example with shape ``(N, M)``, where N is the number of examples and M is the number of annotators. For more details, labels in the same format expected by the :py:func:`get_label_quality_multiannotator <cleanlab.multiannotator.get_label_quality_multiannotator>`. pred_probs : np.ndarray An array of shape ``(N, K)`` of model-predicted probabilities, ``P(label=k|x)``. For details, predicted probabilities in the same format expected by the :py:func:`get_label_quality_multiannotator <cleanlab.multiannotator.get_label_quality_multiannotator>`. consensus_label : np.ndarray An array of shape ``(N,)`` with the consensus labels aggregated from all annotators. num_annotations : np.ndarray An array of shape ``(N,)`` with the number of annotators that have labeled each example. annotator_agreement : np.ndarray An array of shape ``(N,)`` with the fraction of annotators that agree with each consensus label. model_weight : float float specifying the model weight used in weighted averages, None if model weight is not used to compute quality scores annotator_weight : np.ndarray An array of shape ``(M,)`` where M is the number of annotators, specifying the annotator weights used in weighted averages, None if annotator weights are not used to compute quality scores quality_method : str, default = "crowdlab" (Options: ["crowdlab", "agreement"]) Specifies the method used to calculate the quality of the annotators. For valid quality methods, view :py:func:`get_label_quality_multiannotator <cleanlab.multiannotator.get_label_quality_multiannotator>` Returns ------- annotator_quality : np.ndarray Quality scores of a given annotator's labels """ valid_methods = [ "crowdlab", "agreement", ] if quality_method == "crowdlab": annotator_lqs = labels_multiannotator.apply( lambda s: np.mean( get_label_quality_scores(s[pd.notna(s)].astype("int64"), pred_probs[pd.notna(s)]) ) ) mask = num_annotations != 1 labels_multiannotator_subset = labels_multiannotator[mask] consensus_label_subset = consensus_label[mask] annotator_agreement = labels_multiannotator_subset.apply( lambda s: np.mean(s[pd.notna(s)] == consensus_label_subset[pd.notna(s)]), axis=0, ) avg_num_annotations_frac = np.mean(num_annotations) / len(annotator_weight) annotator_weight_adjusted = np.sum(annotator_weight) * avg_num_annotations_frac w = model_weight / (model_weight + annotator_weight_adjusted) annotator_quality = w * annotator_lqs + (1 - w) * annotator_agreement elif quality_method == "agreement": mask = num_annotations != 1 labels_multiannotator_subset = labels_multiannotator[mask] consensus_label_subset = consensus_label[mask] annotator_quality = labels_multiannotator_subset.apply( lambda s: np.mean(s[pd.notna(s)] == consensus_label_subset[pd.notna(s)]), axis=0, ) else: raise ValueError( f""" {quality_method} is not a valid annotator quality method! Please choose a valid quality_method: {valid_methods} """ ) return annotator_quality def _get_annotator_worst_class( labels_multiannotator: pd.DataFrame, consensus_label: np.ndarray, consensus_quality_score: np.ndarray, ) -> np.ndarray: """Returns the class which each annotator makes the most errors in. Parameters ---------- labels_multiannotator : pd.DataFrame 2D pandas DataFrame of multiple given labels for each example with shape ``(N, M)``, where N is the number of examples and M is the number of annotators. For more details, labels in the same format expected by the :py:func:`get_label_quality_multiannotator <cleanlab.multiannotator.get_label_quality_multiannotator>`. consensus_label : np.ndarray An array of shape ``(N,)`` with the consensus labels aggregated from all annotators. consensus_quality_score : np.ndarray An array of shape ``(N,)`` with the quality score of the consensus. Returns ------- worst_class : np.ndarray The class that is most frequently mislabeled by a given annotator """ def get_single_annotator_worst_class(s, consensus_quality_score): mask = pd.notna(s) class_accuracies = (s[mask] == consensus_label[mask]).groupby(s).mean() accuracy_min_idx = class_accuracies[class_accuracies == class_accuracies.min()].index.values if len(accuracy_min_idx) == 1: return accuracy_min_idx[0] # tiebreak 1: class counts class_count = s[mask].groupby(s).count()[accuracy_min_idx] count_max_idx = class_count[class_count == class_count.max()].index.values if len(count_max_idx) == 1: return count_max_idx[0] # tiebreak 2: consensus quality scores avg_consensus_quality = ( pd.DataFrame( {"annotator_label": s, "consensus_quality_score": consensus_quality_score} )[mask] .groupby("annotator_label") .mean()["consensus_quality_score"][count_max_idx] ) quality_max_ind = avg_consensus_quality[ avg_consensus_quality == avg_consensus_quality.max() ].index.values # return first item even if there are ties - no better methods to tiebreak return quality_max_ind[0] worst_class = ( labels_multiannotator.apply( lambda s: get_single_annotator_worst_class(s, consensus_quality_score) ) .to_numpy() .astype("int64") ) return worst_class def _check_consensus_label_classes( labels_multiannotator: pd.DataFrame, consensus_label: np.ndarray, consensus_method: str, ) -> None: """Check if any classes no longer appear in the set of consensus labels (established using the consensus_method stated)""" unique_ma_labels = np.unique(labels_multiannotator.replace({pd.NA: np.NaN}).astype(float)) unique_ma_labels = unique_ma_labels[~np.isnan(unique_ma_labels)] labels_set_difference = set(unique_ma_labels) - set(consensus_label) if len(labels_set_difference) > 0: print( f"""CAUTION: Number of unique classes has been reduced from the original data when establishing consensus labels using consensus method "{consensus_method}", likely due to some classes being rarely annotated. If training a classifier on these consensus labels, it will never see any of the omitted classes unless you manually replace some of the consensus labels. Classes in the original data but not in consensus labels: {list(map(int, labels_set_difference))}""" )