elaine14/Automatic_Question-Answer_Generation

1# app.py2import os3import random4import gradio as gr5import pandas as pd6 7import torch8from transformers import T5ForConditionalGeneration, T5Tokenizer, pipeline9import spacy10from rapidfuzz import fuzz11 12# ---------- Config ----------13DEFAULT_MAX_Q = 314MAX_MCQ_OPTIONS = 415DEVICE = "cuda" if torch.cuda.is_available() else "cpu"16 17# ---------- Helper funcs (adapted from your script) ----------18def highlight_answer(context, answer):19    start_idx = context.lower().find(answer.lower())20    if start_idx == -1:21        return None22    end_idx = start_idx + len(answer)23    return context[:start_idx] + "<hl> " + answer + " <hl>" + context[end_idx:]24 25def prepare_qg_input(context, answer):26    highlighted = highlight_answer(context, answer)27    if highlighted:28        return f"generate question: {highlighted}"29    return None30 31def compute_metrics(pred, gold):32    em = int(pred.strip().lower() == gold.strip().lower())33    f1 = fuzz.token_sort_ratio(pred, gold) / 10034    return em, f135 36def make_fill_blank(question, answer):37    if answer.lower() in question.lower():38        return question.replace(answer, "____")39    return f"____ ({question})"40 41def make_mcq(question, correct_answer, context, num_options=MAX_MCQ_OPTIONS, nlp=None):42    """Generate MCQ options from context using spaCy named entities as distractors."""43    candidates = []44    if nlp:45        doc = nlp(context)46        candidates = list({ent.text for ent in doc.ents if ent.text.lower() != correct_answer.lower()})47    random.shuffle(candidates)48    distractors = candidates[: max(0, num_options - 1)]49    # fill with synthetic distractors if not enough50    if len(distractors) < num_options - 1:51        distractors += [f"Option_{i}" for i in range(len(distractors), num_options - 1)]52    options = distractors + [correct_answer]53    random.shuffle(options)54    return question, options55 56def make_true_false(question, answer, context, nlp=None):57    """Return statement and boolean whether it's true."""58    is_true = random.random() > 0.559    used_answer = answer60    if not is_true and nlp:61        doc = nlp(context)62        wrongs = [ent.text for ent in doc.ents if ent.text.lower() != answer.lower()]63        if wrongs:64            used_answer = random.choice(wrongs)65    statement = f"'{used_answer}' is the correct answer to: \"{question}\""66    return statement, is_true67 68def make_matching(context, num_pairs=4, nlp=None):69    if not nlp:70        return []71    doc = nlp(context)72    entities = list({(ent.text, ent.label_) for ent in doc.ents})73    random.shuffle(entities)74    return entities[:num_pairs]75 76# ---------- Model loading (lazy to keep startup responsive) ----------77MODEL_STATE = {"qg_tokenizer": None, "qg_model": None, "qa_pipeline": None, "nlp": None}78 79def load_models():80    if MODEL_STATE["nlp"] is None:81        try:82            MODEL_STATE["nlp"] = spacy.load("en_core_web_sm")83        except Exception:84            # fallback to blank english if core model not available85            MODEL_STATE["nlp"] = spacy.blank("en")86 87    if MODEL_STATE["qg_tokenizer"] is None:88        MODEL_STATE["qg_tokenizer"] = T5Tokenizer.from_pretrained("valhalla/t5-base-qg-hl")89    if MODEL_STATE["qg_model"] is None:90        MODEL_STATE["qg_model"] = T5ForConditionalGeneration.from_pretrained("valhalla/t5-base-qg-hl").to(DEVICE)91    if MODEL_STATE["qa_pipeline"] is None:92        # use device 0 if CUDA else cpu93        device_id = 0 if DEVICE == "cuda" else -194        MODEL_STATE["qa_pipeline"] = pipeline("question-answering", model="deepset/roberta-base-squad2", device=device_id)95 96def generate_wh_questions(context, answers, max_qs=DEFAULT_MAX_Q):97    """98    We expect `answers` as list of candidate answer spans.99    This function returns generated questions and the verified predicted answers by the QA model.100    """101    load_models()102    tokenizer = MODEL_STATE["qg_tokenizer"]103    qg_model = MODEL_STATE["qg_model"]104    qa_pipe = MODEL_STATE["qa_pipeline"]105 106    inputs = []107    keep_pairs = []108    for ans in answers:109        prepared = prepare_qg_input(context, ans)110        if prepared:111            inputs.append(prepared)112            keep_pairs.append(ans)113 114    # limit115    inputs = inputs[:max_qs]116    keep_pairs = keep_pairs[:max_qs]117 118    questions = []119    verified_answers = []120    if inputs:121        enc = tokenizer(inputs, padding=True, truncation=True, max_length=512, return_tensors="pt").to(DEVICE)122        outs = qg_model.generate(**enc, max_length=64, num_beams=4)123        questions = [tokenizer.decode(o, skip_special_tokens=True) for o in outs]124 125        # verify using QA pipeline126        batch = [{"question": q, "context": context} for q in questions]127        res = qa_pipe(batch)128        if isinstance(res, dict):129            res = [res]130        verified_answers = [r.get("answer", "") for r in res]131 132    return list(zip(questions, keep_pairs, verified_answers))133 134# ---------- High-level pipeline called by Gradio ----------135def generate_all_types(context, max_qs=DEFAULT_MAX_Q):136    """137    Given a context, attempt to produce:138      - WH questions (generated from probable answer spans)139      - Fill-in-the-Blank140      - MCQ141      - True/False142      - Matching pairs143    """144    load_models()145    nlp = MODEL_STATE["nlp"]146 147    # 1) pick candidate answers from named entities (simple heuristic)148    doc = nlp(context)149    answers = [ent.text for ent in doc.ents]150    # fall back: split important nouns if no entities found151    if not answers:152        tokens = [t.text for t in doc if t.pos_ in ("PROPN", "NOUN") and len(t.text) > 2]153        answers = list(dict.fromkeys(tokens))  # unique maintain order154 155    # if still empty, just use small text fragment156    if not answers:157        answers = [context.strip().split(".")[0][:50]]158 159    # 2) generate WH questions and verified answers160    wh_pairs = generate_wh_questions(context, answers, max_qs=max_qs)161 162    # 3) for each WH question generate other question types163    rows = []164    for (q, gold_a, pred_a) in wh_pairs:165        # WH row166        rows.append({"Type": "WH", "Question": q, "Gold Answer": gold_a, "Verified Answer": pred_a})167 168        # Fill-Blank169        fib = make_fill_blank(q, gold_a)170        rows.append({"Type": "Fill-Blank", "Question": fib, "Answer": gold_a})171 172        # MCQ173        mcq_q, mcq_opts = make_mcq(q, gold_a, context, nlp=nlp)174        rows.append({"Type": "MCQ", "Question": mcq_q, "Answer": gold_a, "Options": mcq_opts})175 176        # True/False177        tf_stmt, tf_val = make_true_false(q, gold_a, context, nlp=nlp)178        rows.append({"Type": "True/False", "Question": tf_stmt, "Answer": str(tf_val)})179 180        # Matching181        pairs = make_matching(context, nlp=nlp)182        rows.append({"Type": "Matching", "Pairs": pairs})183 184    # If there were fewer wh_pairs than requested, add more matching/MCQ from other answers185    # (optional) — keep it simple for now.186 187    # Build DataFrame for display188    df_rows = []189    for r in rows:190        # normalize display for different question types191        if r["Type"] == "MCQ":192            options = r.get("Options", [])193            df_rows.append({194                "Type": r["Type"],195                "Question": r["Question"],196                "Answer / Gold": r.get("Answer", ""),197                "Options": " | ".join(options)198            })199        elif r["Type"] == "Matching":200            pairs = r.get("Pairs", [])201            df_rows.append({202                "Type": r["Type"],203                "Question": "",204                "Answer / Gold": "",205                "Options": "; ".join([f"{a} -> {b}" for a,b in pairs])206            })207        else:208            df_rows.append({209                "Type": r["Type"],210                "Question": r.get("Question", ""),211                "Answer / Gold": r.get("Answer", r.get("Gold Answer", r.get("Verified Answer", ""))),212                "Options": ""213            })214 215    out_df = pd.DataFrame(df_rows)216    # prepare CSV bytes for download217    csv_bytes = out_df.to_csv(index=False).encode("utf-8")218    return out_df, csv_bytes219 220# ---------- Gradio UI ----------221with gr.Blocks(title="QG & QA — multi-type generator") as demo:222    gr.Markdown("## Question Generation & QA — WH, Fill-Blank, MCQ, True/False, Matching\n"223                "Enter a *context paragraph* and press **Generate**. Models used: `valhalla/t5-base-qg-hl` and `deepset/roberta-base-squad2`.")224 225    with gr.Row():226        context_in = gr.Textbox(lines=8, label="Context paragraph", placeholder="Paste a paragraph here...")227    with gr.Row():228        max_qs_slider = gr.Slider(minimum=1, maximum=6, step=1, value=DEFAULT_MAX_Q, label="Max WH questions to generate")229        generate_btn = gr.Button("Generate")230    with gr.Row():231        output_table = gr.Dataframe(headers=["Type", "Question", "Answer / Gold", "Options"], label="Generated questions")232    with gr.Row():233        download_btn = gr.File(label="Download CSV")234 235    def on_generate(context, max_qs):236        if not context or not context.strip():237            return pd.DataFrame(columns=["Type", "Question", "Answer / Gold", "Options"]), None238        out_df, csv_bytes = generate_all_types(context, int(max_qs))239        # Save CSV to a temporary file so gr.File can serve it240        tmp_path = "/tmp/generated_qa.csv"241        with open(tmp_path, "wb") as f:242            f.write(csv_bytes)243        return out_df, tmp_path244 245    generate_btn.click(fn=on_generate, inputs=[context_in, max_qs_slider], outputs=[output_table, download_btn])246 247# Start server when run as script248if __name__ == "__main__":249    demo.launch(server_name="0.0.0.0", server_port=int(os.environ.get("PORT", 7860)))250