app.py

import torch
from transformers import AlbertTokenizer, AlbertForSequenceClassification, AlbertForQuestionAnswering
import collections
import math
import gradio as gr

cls_modelPath = "./cls_model"
mrc_modelPath = "./model4"

tokenizer = AlbertTokenizer.from_pretrained(mrc_modelPath)
cls_model = AlbertForSequenceClassification.from_pretrained(cls_modelPath)
cls_model.eval()
mrc_model = AlbertForQuestionAnswering.from_pretrained(mrc_modelPath)
mrc_model.eval()

def _get_best_indexes(logits, n_best_size):
    """Get the n-best logits from a list."""
    index_and_score = sorted(enumerate(logits), key=lambda x: x[1], reverse=True)

    best_indexes = []
    for i in range(len(index_and_score)):
        if i >= n_best_size:
            break
        best_indexes.append(index_and_score[i][0])
    return best_indexes

def _compute_softmax(scores):
    """Compute softmax probability over raw logits."""
    if not scores:
        return []

    max_score = None
    for score in scores:
        if max_score is None or score > max_score:
            max_score = score

    exp_scores = []
    total_sum = 0.0
    for score in scores:
        x = math.exp(score - max_score)
        exp_scores.append(x)
        total_sum += x

    probs = []
    for score in exp_scores:
        probs.append(score / total_sum)
    return probs

def get_qa_nbest(input_ids, start_logits, end_logits, seq_len, n_best_size=20, max_answer_length=30):
    score_null = 1000000  # large and positive
    prelim_predictions = []
    null_start_logit = 0  # the start logit at the slice with min null score
    null_end_logit = 0  # the end logit at the slice with min null score
    _PrelimPrediction = collections.namedtuple(  # pylint: disable=invalid-name
        "PrelimPrediction",
        ["start_index", "end_index", "start_logit", "end_logit"])
    _NbestPrediction = collections.namedtuple(  # pylint: disable=invalid-name
            "NbestPrediction", ["text", "start_logit", "end_logit"])
    
    start_indexes = _get_best_indexes(start_logits, n_best_size)
    end_indexes = _get_best_indexes(end_logits, n_best_size)

    feature_null_score = start_logits[0] + end_logits[0]
    if feature_null_score < score_null:
        score_null = feature_null_score
    for start_index in start_indexes:
        for end_index in end_indexes:
            if end_index < start_index:
                continue
            length = end_index - start_index + 1
            if length > max_answer_length:
                continue
            if start_index >= seq_len:
                        continue
            if end_index >= seq_len:
                continue
            prelim_predictions.append(
                _PrelimPrediction(
                    start_index=start_index,
                    end_index=end_index,
                    start_logit=start_logits[start_index],
                    end_logit=end_logits[end_index]))
    prelim_predictions = sorted(
            prelim_predictions,
            key=lambda x: (x.start_logit + x.start_logit),
            reverse=True)
    
    seen_predictions = {}
    nbest = []
    for pred in prelim_predictions:
        if len(nbest) >= n_best_size:
            break
        
        if pred.start_index > 0:  # this is a non-null prediction\
            predict_answer_tokens = input_ids[0, pred.start_index: (pred.end_index + 1)]
            final_text = tokenizer.decode(predict_answer_tokens)
            if final_text in seen_predictions:
                continue
            seen_predictions[final_text] = True
        else:
            final_text = ""
            seen_predictions[final_text] = True
        

        nbest.append(
                _NbestPrediction(
                    text=final_text,
                    start_logit=pred.start_logit,
                    end_logit=pred.end_logit))
    if "" not in seen_predictions:
        nbest.append(
            _NbestPrediction(
                text="",
                start_logit=null_start_logit,
                end_logit=null_end_logit))

            # In very rare edge cases we could only have single null prediction.
            # So we just create a nonce prediction in this case to avoid failure.
    if len(nbest) == 1:
        nbest.insert(0,
            _NbestPrediction(text="empty", start_logit=0.0, end_logit=0.0))

    # In very rare edge cases we could have no valid predictions. So we
    # just create a nonce prediction in this case to avoid failure.
    if not nbest:
        nbest.append(
            _NbestPrediction(text="empty", start_logit=0.0, end_logit=0.0))


    total_scores = []
    best_non_null_entry = None
    for entry in nbest:
        total_scores.append(entry.start_logit + entry.end_logit)
        if not best_non_null_entry:
            if entry.text:
                best_non_null_entry = entry

    probs = _compute_softmax(total_scores)
    nbest_json = []
    for (i, entry) in enumerate(nbest):
        output = collections.OrderedDict()
        output["text"] = entry.text
        output["probability"] = probs[i]
        output["start_logit"] = entry.start_logit
        output["end_logit"] = entry.end_logit
        nbest_json.append(output)

    score_diff = score_null - best_non_null_entry.start_logit - (
                best_non_null_entry.end_logit)
    
    return nbest_json, score_diff

def inference(context, question, reference):
    inputs = tokenizer(
                question,
                context,
                add_special_tokens=True,
                pad_to_max_length=True,
                max_length=512,
                return_tensors="pt"
            )

    seq_len = inputs.input_ids[0].tolist().index(0)

    with torch.no_grad():
        cls_outputs = cls_model(**inputs)
        qa_outputs = mrc_model(**inputs)

    cls_logits = cls_outputs.logits[0]
    cls_divide = cls_logits[1] - cls_logits[0]


    nbest, score_diff = get_qa_nbest(inputs.input_ids, qa_outputs.start_logits[0], qa_outputs.end_logits[0], seq_len=seq_len)

    thresh = -1.246073067188263

    print(cls_divide, score_diff)

    na_score = (0.5*cls_divide + 0.5*score_diff)*0.5
    if na_score > thresh:
        final_answer = f"<No Answer>. The question is not answerable according to the context."
    else:
        final_answer = nbest[0]["text"]
    return final_answer

demo = gr.Interface(
    fn=inference,
    inputs=[gr.inputs.Textbox(label="Context"), 
            gr.inputs.Textbox(label="Question"),
            gr.inputs.Textbox(label="Reference Answer (Optional)")],
    outputs=gr.outputs.Textbox(label="Output Answer"),
    examples = [
    ["The Norman dynasty had a major political, cultural and military impact on medieval Europe and even the Near East. The Normans were famed for their martial spirit and eventually for their Christian piety, becoming exponents of the Catholic orthodoxy into which they assimilated. They adopted the Gallo-Romance language of the Frankish land they settled, their dialect becoming known as Norman, Normaund or Norman French, an important literary language. The Duchy of Normandy, which they formed by treaty with the French crown, was a great fief of medieval France, and under Richard I of Normandy was forged into a cohesive and formidable principality in feudal tenure. The Normans are noted both for their culture, such as their unique Romanesque architecture and musical traditions, and for their significant military accomplishments and innovations. Norman adventurers founded the Kingdom of Sicily under Roger II after conquering southern Italy on the Saracens and Byzantines, and an expedition on behalf of their duke, William the Conqueror, led to the Norman conquest of England at the Battle of Hastings in 1066. Norman cultural and military influence spread from these new European centres to the Crusader states of the Near East, where their prince Bohemond I founded the Principality of Antioch in the Levant, to Scotland and Wales in Great Britain, to Ireland, and to the coasts of north Africa and the Canary Islands.", "Who was the duke in the battle of Hastings?", "William the Conqueror"],
    ["The Norman dynasty had a major political, cultural and military impact on medieval Europe and even the Near East. The Normans were famed for their martial spirit and eventually for their Christian piety, becoming exponents of the Catholic orthodoxy into which they assimilated. They adopted the Gallo-Romance language of the Frankish land they settled, their dialect becoming known as Norman, Normaund or Norman French, an important literary language. The Duchy of Normandy, which they formed by treaty with the French crown, was a great fief of medieval France, and under Richard I of Normandy was forged into a cohesive and formidable principality in feudal tenure. The Normans are noted both for their culture, such as their unique Romanesque architecture and musical traditions, and for their significant military accomplishments and innovations. Norman adventurers founded the Kingdom of Sicily under Roger II after conquering southern Italy on the Saracens and Byzantines, and an expedition on behalf of their duke, William the Conqueror, led to the Norman conquest of England at the Battle of Hastings in 1066. Norman cultural and military influence spread from these new European centres to the Crusader states of the Near East, where their prince Bohemond I founded the Principality of Antioch in the Levant, to Scotland and Wales in Great Britain, to Ireland, and to the coasts of north Africa and the Canary Islands.", "What type of major impact did the Norman dynasty have on modern Europe?", "<No Answer>"],
    ["Steam engines are external combustion engines, where the working fluid is separate from the combustion products. Non-combustion heat sources such as solar power, nuclear power or geothermal energy may be used. The ideal thermodynamic cycle used to analyze this process is called the Rankine cycle. In the cycle, water is heated and transforms into steam within a boiler operating at a high pressure. When expanded through pistons or turbines, mechanical work is done. The reduced-pressure steam is then condensed and pumped back into the boiler.", "What types of engines are steam engines?", "external combustion engines"],
    ["Steam engines are external combustion engines, where the working fluid is separate from the combustion products. Non-combustion heat sources such as solar power, nuclear power or geothermal energy may be used. The ideal thermodynamic cycle used to analyze this process is called the Rankine cycle. In the cycle, water is heated and transforms into steam within a boiler operating at a high pressure. When expanded through pistons or turbines, mechanical work is done. The reduced-pressure steam is then condensed and pumped back into the boiler.", "What ideal thermodynamic cycle analyzes the process by which solar engines work?", "<No Answer>"],
    ],
    title="Retrospective Reader for Machine Reading Comprehension",
    description=("<div style='text-align: center; margin: 0 auto;'>The model achieved the best performance at the SQuAD2.0 leaderboard. See more details at: <a href='https://aaai.org/papers/14506-retrospective-reader-for-machine-reading-comprehension/'>Paper</a> and <a href='https://github.com/cooelf/AwesomeMRC'>GitHub</a></div>"),
    )

demo.launch(debug=True)