BERT_rationale_benchmark/models/pipeline/pipeline_utils.py
def decode_batch(data: List[Tuple[SentenceEvidence, SentenceEvidence]], name: str, score: bool = False,
                     annotations: List[Annotation] = None, class_labels: dict=class_labels) -> dict:
        """Identifies evidence statements and then makes classifications based on it.

        Args:
            data: a paired list of SentenceEvidences, differing only in the kls field.
                  The first corresponds to whether or not something is evidence, and the second corresponds to an evidence class
            name: a name for a results dict
        """

        num_uniques = len(set((x.ann_id, x.docid) for x, _ in data))
        logging.info(f'Decoding dataset {name} with {len(data)} sentences, {num_uniques} annotations')
        identifier_data, classifier_data = zip(*data)
        results = dict()
        with torch.no_grad():
            # make predictions for the evidence_token_identifier
            evidence_token_identifier.eval()
            evidence_classifier.eval()

            _, soft_identification_preds, hard_identification_preds, id_preds_truth = make_token_preds_epoch(evidence_token_identifier,
                                                                                                             identifier_data,
                                                                                                             token_mapping,
                                                                                                             batch_size,
                                                                                                             device,
                                                                                                             tensorize_model_inputs=True)
            assert len(soft_identification_preds) == len(data)
            evidence_only_cls = []
            for id_data, cls_data, soft_id_pred, hard_id_pred in zip(identifier_data,
                                                                     classifier_data,
                                                                     soft_identification_preds,
                                                                     hard_identification_preds):
                assert cls_data.ann_id == id_data.ann_id
                sent = []
                for (start, end) in token_mapping[cls_data.docid][0]:
                    if bool(hard_id_pred[start]):
                        sent.extend(id_data.sentence[start:end])
                #assert len(sent) > 0
                new_cls_data = SentenceEvidence(cls_data.kls,
                                                cls_data.ann_id,
                                                cls_data.query,
                                                cls_data.docid,
                                                cls_data.index,
                                                tuple(sent))
                evidence_only_cls.append(new_cls_data)
            _, soft_classification_preds, hard_classification_preds, classification_truth = make_preds_epoch(
                evidence_classifier, evidence_only_cls, batch_size, device,
                tensorize_model_inputs=True)

            if use_cose_hack:
                logging.info('Reformatting identification and classification results to fit COS-E')
                grouping = 5
                new_soft_identification_preds = []
                new_hard_identification_preds = []
                new_id_preds_truth = []
                new_soft_classification_preds = []
                new_hard_classification_preds = []
                new_classification_truth = []
                new_identifier_data = []
                class_labels = []

                # TODO fix the labels for COS-E
                for i in range(0, len(soft_identification_preds), grouping):
                    cls_scores = torch.stack(soft_classification_preds[i:i + grouping])
                    cls_scores = nn.functional.softmax(cls_scores, dim=-1)
                    cls_scores = cls_scores[:,1]
                    choice = torch.argmax(cls_scores)
                    cls_labels = [x.ann_id.split('_')[-1] for x in evidence_only_cls[i:i + grouping]]
                    class_labels = cls_labels  # we need to update the class labels because of the terrible hackery used to train this
                    cls_truths = [x.kls for x in evidence_only_cls[i:i + grouping]]
                    #cls_choice = evidence_only_cls[i + choice].ann_id.split('_')[-1]
                    cls_truth = np.argmax(cls_truths)
                    new_soft_identification_preds.append(soft_identification_preds[i + choice])
                    new_hard_identification_preds.append(hard_identification_preds[i + choice])
                    new_id_preds_truth.append(id_preds_truth[i + choice])
                    new_soft_classification_preds.append(soft_classification_preds[i + choice])
                    new_hard_classification_preds.append(choice)
                    new_identifier_data.append(identifier_data[i + choice])
                    #new_hard_classification_preds.append(hard_classification_preds[i + choice])
                    #new_classification_truth.append(classification_truth[i + choice])
                    new_classification_truth.append(cls_truth)

                soft_identification_preds = new_soft_identification_preds
                hard_identification_preds = new_hard_identification_preds
                id_preds_truth = new_id_preds_truth
                soft_classification_preds = new_soft_classification_preds
                hard_classification_preds = new_hard_classification_preds
                classification_truth = new_classification_truth
                identifier_data = new_identifier_data
            if score:
                results[f'{name}_f1'] = classification_report(classification_truth, hard_classification_preds,
                                                              target_names=class_labels, output_dict=True)
                results[f'{name}_acc'] = accuracy_score(classification_truth, hard_classification_preds)
                results[f'{name}_token_pred_acc'] = accuracy_score(list(chain.from_iterable(id_preds_truth)),
                                                                   list(chain.from_iterable(hard_identification_preds)))
                results[f'{name}_token_pred_f1'] = classification_report(list(chain.from_iterable(id_preds_truth)),
                                                                         list(chain.from_iterable(hard_identification_preds)),
                                                                         output_dict=True)
                # TODO for token level stuff!
                soft_id_scores = [[1-x, x] for x in chain.from_iterable(soft_identification_preds)]
                results[f'{name}_rationale'] = score_rationales(annotations,
                                                                decoding_docs,
                                                                identifier_data,
                                                                soft_id_scores)
                logging.info(f'Results: {results}')

            # turn the above results into a format suitable for scoring via the rationale scorer
            # n.b. the sentence-level evidence predictions (hard and soft) are
            # broadcast to the token level for scoring. The comprehensiveness class
            # score is also a lie since the pipeline model above is faithful by
            # design.
            decoded = dict()
            scores = []
            assert len(identifier_data) == len(soft_identification_preds)
            for id_data, soft_id_pred, hard_id_pred, soft_cls_preds, hard_cls_pred in zip(identifier_data,
                                                                                          soft_identification_preds,
                                                                                          hard_identification_preds,
                                                                                          soft_classification_preds,
                                                                                          hard_classification_preds):
                docid = id_data.docid
                if use_cose_hack:
                    docid = '_'.join(docid.split('_')[0:-1])
                assert len(docid) > 0
                rationales = {
                    "docid": docid,
                    "hard_rationale_predictions": [],
                    # token level classifications, a value must be provided per-token
                    # in an ideal world, these correspond to the hard-decoding above.
                    "soft_rationale_predictions": [],
                    # sentence level classifications, a value must be provided for every
                    # sentence in each document, or not at all
                    "soft_sentence_predictions": [1.0]
                }
                last = -1
                start_span = -1
                for pos, (start, _) in enumerate(token_mapping[id_data.docid][0]):
                    rationales['soft_rationale_predictions'].append(soft_id_pred[start])
                    if bool(hard_id_pred[start]):
                        if start_span == -1:
                            start_span = pos
                        last = pos
                    else:
                        if start_span != -1:
                            rationales['hard_rationale_predictions'].append({
                                "start_token": start_span,
                                "end_token": last + 1,
                            })
                        last = -1
                        start_span = -1
                if start_span != -1:
                    rationales['hard_rationale_predictions'].append({
                        "start_token": start_span,
                        "end_token": last + 1,
                    })

                ann_id = id_data.ann_id
                if use_cose_hack:
                    ann_id = '_'.join(ann_id.split('_')[0:-1])
                soft_cls_preds = nn.functional.softmax(soft_cls_preds)
                decoded[id_data.ann_id] = {
                    "annotation_id": ann_id,
                    "rationales": [rationales],
                    "classification": class_labels[hard_cls_pred],
                    "classification_scores": {class_labels[i]:score.item() for i,score in enumerate(soft_cls_preds)}
                }
            return results, list(decoded.values())
Identifies evidence statements and then makes classifications based on it.

        Args:
            data: a paired list of SentenceEvidences, differing only in the kls field.
                  The first corresponds to whether or not something is evidence, and the second corresponds to an evidence class
            name: a name for a results dict
        
Identifies evidence statements and then makes classifications based on it.
Identifies evidence statements and then makes classifications based on it .
def decode_batch(data: List[Tuple[SentenceEvidence, SentenceEvidence]], name: str, score: bool = False,
                     annotations: List[Annotation] = None, class_labels: dict=class_labels) -> dict:
        num_uniques = len(set((x.ann_id, x.docid) for x, _ in data))
        logging.info(f'Decoding dataset {name} with {len(data)} sentences, {num_uniques} annotations')
        identifier_data, classifier_data = zip(*data)
        results = dict()
        with torch.no_grad():
            evidence_token_identifier.eval()
            evidence_classifier.eval()
            _, soft_identification_preds, hard_identification_preds, id_preds_truth = make_token_preds_epoch(evidence_token_identifier,
                                                                                                             identifier_data,
                                                                                                             token_mapping,
                                                                                                             batch_size,
                                                                                                             device,
                                                                                                             tensorize_model_inputs=True)
            assert len(soft_identification_preds) == len(data)
            evidence_only_cls = []
            for id_data, cls_data, soft_id_pred, hard_id_pred in zip(identifier_data,
                                                                     classifier_data,
                                                                     soft_identification_preds,
                                                                     hard_identification_preds):
                assert cls_data.ann_id == id_data.ann_id
                sent = []
                for (start, end) in token_mapping[cls_data.docid][0]:
                    if bool(hard_id_pred[start]):
                        sent.extend(id_data.sentence[start:end])
                new_cls_data = SentenceEvidence(cls_data.kls,
                                                cls_data.ann_id,
                                                cls_data.query,
                                                cls_data.docid,
                                                cls_data.index,
                                                tuple(sent))
                evidence_only_cls.append(new_cls_data)
            _, soft_classification_preds, hard_classification_preds, classification_truth = make_preds_epoch(
                evidence_classifier, evidence_only_cls, batch_size, device,
                tensorize_model_inputs=True)
            if use_cose_hack:
                logging.info('Reformatting identification and classification results to fit COS-E')
                grouping = 5
                new_soft_identification_preds = []
                new_hard_identification_preds = []
                new_id_preds_truth = []
                new_soft_classification_preds = []
                new_hard_classification_preds = []
                new_classification_truth = []
                new_identifier_data = []
                class_labels = []
                for i in range(0, len(soft_identification_preds), grouping):
                    cls_scores = torch.stack(soft_classification_preds[i:i + grouping])
                    cls_scores = nn.functional.softmax(cls_scores, dim=-1)
                    cls_scores = cls_scores[:,1]
                    choice = torch.argmax(cls_scores)
                    cls_labels = [x.ann_id.split('_')[-1] for x in evidence_only_cls[i:i + grouping]]
                    class_labels = cls_labels  
                    cls_truths = [x.kls for x in evidence_only_cls[i:i + grouping]]
                    cls_truth = np.argmax(cls_truths)
                    new_soft_identification_preds.append(soft_identification_preds[i + choice])
                    new_hard_identification_preds.append(hard_identification_preds[i + choice])
                    new_id_preds_truth.append(id_preds_truth[i + choice])
                    new_soft_classification_preds.append(soft_classification_preds[i + choice])
                    new_hard_classification_preds.append(choice)
                    new_identifier_data.append(identifier_data[i + choice])
                    new_classification_truth.append(cls_truth)
                soft_identification_preds = new_soft_identification_preds
                hard_identification_preds = new_hard_identification_preds
                id_preds_truth = new_id_preds_truth
                soft_classification_preds = new_soft_classification_preds
                hard_classification_preds = new_hard_classification_preds
                classification_truth = new_classification_truth
                identifier_data = new_identifier_data
            if score:
                results[f'{name}_f1'] = classification_report(classification_truth, hard_classification_preds,
                                                              target_names=class_labels, output_dict=True)
                results[f'{name}_acc'] = accuracy_score(classification_truth, hard_classification_preds)
                results[f'{name}_token_pred_acc'] = accuracy_score(list(chain.from_iterable(id_preds_truth)),
                                                                   list(chain.from_iterable(hard_identification_preds)))
                results[f'{name}_token_pred_f1'] = classification_report(list(chain.from_iterable(id_preds_truth)),
                                                                         list(chain.from_iterable(hard_identification_preds)),
                                                                         output_dict=True)
                soft_id_scores = [[1-x, x] for x in chain.from_iterable(soft_identification_preds)]
                results[f'{name}_rationale'] = score_rationales(annotations,
                                                                decoding_docs,
                                                                identifier_data,
                                                                soft_id_scores)
                logging.info(f'Results: {results}')
            decoded = dict()
            scores = []
            assert len(identifier_data) == len(soft_identification_preds)
            for id_data, soft_id_pred, hard_id_pred, soft_cls_preds, hard_cls_pred in zip(identifier_data,
                                                                                          soft_identification_preds,
                                                                                          hard_identification_preds,
                                                                                          soft_classification_preds,
                                                                                          hard_classification_preds):
                docid = id_data.docid
                if use_cose_hack:
                    docid = '_'.join(docid.split('_')[0:-1])
                assert len(docid) > 0
                rationales = {
                    "docid": docid,
                    "hard_rationale_predictions": [],
                    "soft_rationale_predictions": [],
                    "soft_sentence_predictions": [1.0]
                }
                last = -1
                start_span = -1
                for pos, (start, _) in enumerate(token_mapping[id_data.docid][0]):
                    rationales['soft_rationale_predictions'].append(soft_id_pred[start])
                    if bool(hard_id_pred[start]):
                        if start_span == -1:
                            start_span = pos
                        last = pos
                    else:
                        if start_span != -1:
                            rationales['hard_rationale_predictions'].append({
                                "start_token": start_span,
                                "end_token": last + 1,
                            })
                        last = -1
                        start_span = -1
                if start_span != -1:
                    rationales['hard_rationale_predictions'].append({
                        "start_token": start_span,
                        "end_token": last + 1,
                    })
                ann_id = id_data.ann_id
                if use_cose_hack:
                    ann_id = '_'.join(ann_id.split('_')[0:-1])
                soft_cls_preds = nn.functional.softmax(soft_cls_preds)
                decoded[id_data.ann_id] = {
                    "annotation_id": ann_id,
                    "rationales": [rationales],
                    "classification": class_labels[hard_cls_pred],
                    "classification_scores": {class_labels[i]:score.item() for i,score in enumerate(soft_cls_preds)}
                }
            return results, list(decoded.values())
def decode_batch ( data : List [ Tuple [ SentenceEvidence , SentenceEvidence ] ] , name : str , score : bool = False , annotations : List [ Annotation ] = None , class_labels : dict = class_labels ) -> dict : num_uniques = len ( set ( ( x . ann_id , x . docid ) for x , _ in data ) ) logging . info ( f'Decoding dataset {name} with {len(data)} sentences, {num_uniques} annotations' ) identifier_data , classifier_data = zip ( * data ) results = dict ( ) with torch . no_grad ( ) : evidence_token_identifier . eval ( ) evidence_classifier . eval ( ) _ , soft_identification_preds , hard_identification_preds , id_preds_truth = make_token_preds_epoch ( evidence_token_identifier , identifier_data , token_mapping , batch_size , device , tensorize_model_inputs = True ) assert len ( soft_identification_preds ) == len ( data ) evidence_only_cls = [ ] for id_data , cls_data , soft_id_pred , hard_id_pred in zip ( identifier_data , classifier_data , soft_identification_preds , hard_identification_preds ) : assert cls_data . ann_id == id_data . ann_id sent = [ ] for ( start , end ) in token_mapping [ cls_data . docid ] [ 0 ] : if bool ( hard_id_pred [ start ] ) : sent . extend ( id_data . sentence [ start : end ] ) new_cls_data = SentenceEvidence ( cls_data . kls , cls_data . ann_id , cls_data . query , cls_data . docid , cls_data . index , tuple ( sent ) ) evidence_only_cls . append ( new_cls_data ) _ , soft_classification_preds , hard_classification_preds , classification_truth = make_preds_epoch ( evidence_classifier , evidence_only_cls , batch_size , device , tensorize_model_inputs = True ) if use_cose_hack : logging . info ( 'Reformatting identification and classification results to fit COS-E' ) grouping = 5 new_soft_identification_preds = [ ] new_hard_identification_preds = [ ] new_id_preds_truth = [ ] new_soft_classification_preds = [ ] new_hard_classification_preds = [ ] new_classification_truth = [ ] new_identifier_data = [ ] class_labels = [ ] for i in range ( 0 , len ( soft_identification_preds ) , grouping ) : cls_scores = torch . stack ( soft_classification_preds [ i : i + grouping ] ) cls_scores = nn . functional . softmax ( cls_scores , dim = - 1 ) cls_scores = cls_scores [ : , 1 ] choice = torch . argmax ( cls_scores ) cls_labels = [ x . ann_id . split ( '_' ) [ - 1 ] for x in evidence_only_cls [ i : i + grouping ] ] class_labels = cls_labels cls_truths = [ x . kls for x in evidence_only_cls [ i : i + grouping ] ] cls_truth = np . argmax ( cls_truths ) new_soft_identification_preds . append ( soft_identification_preds [ i + choice ] ) new_hard_identification_preds . append ( hard_identification_preds [ i + choice ] ) new_id_preds_truth . append ( id_preds_truth [ i + choice ] ) new_soft_classification_preds . append ( soft_classification_preds [ i + choice ] ) new_hard_classification_preds . append ( choice ) new_identifier_data . append ( identifier_data [ i + choice ] ) new_classification_truth . append ( cls_truth ) soft_identification_preds = new_soft_identification_preds hard_identification_preds = new_hard_identification_preds id_preds_truth = new_id_preds_truth soft_classification_preds = new_soft_classification_preds hard_classification_preds = new_hard_classification_preds classification_truth = new_classification_truth identifier_data = new_identifier_data if score : results [ f'{name}_f1' ] = classification_report ( classification_truth , hard_classification_preds , target_names = class_labels , output_dict = True ) results [ f'{name}_acc' ] = accuracy_score ( classification_truth , hard_classification_preds ) results [ f'{name}_token_pred_acc' ] = accuracy_score ( list ( chain . from_iterable ( id_preds_truth ) ) , list ( chain . from_iterable ( hard_identification_preds ) ) ) results [ f'{name}_token_pred_f1' ] = classification_report ( list ( chain . from_iterable ( id_preds_truth ) ) , list ( chain . from_iterable ( hard_identification_preds ) ) , output_dict = True ) soft_id_scores = [ [ 1 - x , x ] for x in chain . from_iterable ( soft_identification_preds ) ] results [ f'{name}_rationale' ] = score_rationales ( annotations , decoding_docs , identifier_data , soft_id_scores ) logging . info ( f'Results: {results}' ) decoded = dict ( ) scores = [ ] assert len ( identifier_data ) == len ( soft_identification_preds ) for id_data , soft_id_pred , hard_id_pred , soft_cls_preds , hard_cls_pred in zip ( identifier_data , soft_identification_preds , hard_identification_preds , soft_classification_preds , hard_classification_preds ) : docid = id_data . docid if use_cose_hack : docid = '_' . join ( docid . split ( '_' ) [ 0 : - 1 ] ) assert len ( docid ) > 0 rationales = { "docid" : docid , "hard_rationale_predictions" : [ ] , "soft_rationale_predictions" : [ ] , "soft_sentence_predictions" : [ 1.0 ] } last = - 1 start_span = - 1 for pos , ( start , _ ) in enumerate ( token_mapping [ id_data . docid ] [ 0 ] ) : rationales [ 'soft_rationale_predictions' ] . append ( soft_id_pred [ start ] ) if bool ( hard_id_pred [ start ] ) : if start_span == - 1 : start_span = pos last = pos else : if start_span != - 1 : rationales [ 'hard_rationale_predictions' ] . append ( { "start_token" : start_span , "end_token" : last + 1 , } ) last = - 1 start_span = - 1 if start_span != - 1 : rationales [ 'hard_rationale_predictions' ] . append ( { "start_token" : start_span , "end_token" : last + 1 , } ) ann_id = id_data . ann_id if use_cose_hack : ann_id = '_' . join ( ann_id . split ( '_' ) [ 0 : - 1 ] ) soft_cls_preds = nn . functional . softmax ( soft_cls_preds ) decoded [ id_data . ann_id ] = { "annotation_id" : ann_id , "rationales" : [ rationales ] , "classification" : class_labels [ hard_cls_pred ] , "classification_scores" : { class_labels [ i ] : score . item ( ) for i , score in enumerate ( soft_cls_preds ) } } return results , list ( decoded . values ( ) )
Identifies evidence statements and then makes classifications based on it.
Identifies evidence statements and then makes classifications based on it .
"""Identifies evidence statements and then makes classifications based on it.

        Args:
            data: a paired list of SentenceEvidences, differing only in the kls field.
                  The first corresponds to whether or not something is evidence, and the second corresponds to an evidence class
            name: a name for a results dict
        """ # make predictions for the evidence_token_identifier #assert len(sent) > 0 # TODO fix the labels for COS-E # we need to update the class labels because of the terrible hackery used to train this #cls_choice = evidence_only_cls[i + choice].ann_id.split('_')[-1] #new_hard_classification_preds.append(hard_classification_preds[i + choice]) #new_classification_truth.append(classification_truth[i + choice]) # TODO for token level stuff! # turn the above results into a format suitable for scoring via the rationale scorer # n.b. the sentence-level evidence predictions (hard and soft) are # broadcast to the token level for scoring. The comprehensiveness class # score is also a lie since the pipeline model above is faithful by # design. # token level classifications, a value must be provided per-token # in an ideal world, these correspond to the hard-decoding above. # sentence level classifications, a value must be provided for every # sentence in each document, or not at all
data
List[Tuple[SentenceEvidence, SentenceEvidence]]
a paired list of SentenceEvidences, differing only in the kls field.
The first corresponds to whether or not something is evidence, and the second corresponds to an evidence class
a paired list of SentenceEvidences differing only in the kls field . The first corresponds to whether or not something is evidence and the second corresponds to an evidence class
name
str
a name for a results dict
a name for a results dict
score
bool
annotations
List[Annotation]
class_labels
dict