import json import re import os import spacy from transformers import pipeline, AutoModelForSeq2SeqLM, AutoTokenizer import gradio as gr from huggingface_hub import Repository from datetime import datetime nlp = spacy.load("en_core_web_sm") qg_model = AutoModelForSeq2SeqLM.from_pretrained("valhalla/t5-base-qa-qg-hl") qg_tokenizer = AutoTokenizer.from_pretrained("valhalla/t5-base-qa-qg-hl", use_fast=True) qg_pipeline = pipeline("text2text-generation", model=qg_model, tokenizer=qg_tokenizer) def extract_paragraph_facts(raw_text): return [p.strip() for p in raw_text.strip().split("\n\n") if p.strip()] def extract_noun_phrases(text): doc = nlp(text) return [np.text for np in doc.noun_chunks] def auto_highlight_noun_phrase(text): doc = nlp(text) noun_phrases = sorted(doc.noun_chunks, key=lambda np: len(np.text), reverse=True) for np in noun_phrases: if len(np.text.split()) > 1 or np.root.pos_ == "NOUN": return np.text return text def highlight_selected_phrase(fact, selected_np): return fact.replace(selected_np, f"{selected_np}", 1) def generate_single_qna(fact, noun_phrase, min_len, max_len, temperature, top_k, top_p): hl_fact = highlight_selected_phrase(fact, noun_phrase) try: prompt = f"generate question: {hl_fact}" output = qg_pipeline( prompt, min_length=min_len, max_length=max_len, temperature=temperature, top_k=top_k, top_p=top_p, do_sample=True )[0] question = output.get("generated_text", "").strip() if not question.endswith("?"): question += "?" except Exception as e: question = f"Error generating question: {str(e)}" return {"question": question, "answer": fact} def generate_qna_all(input_text, selected_fact, selected_np, min_len, max_len, temperature, top_k, top_p): facts = extract_paragraph_facts(input_text) results = [] if selected_fact: noun_phrase = selected_np if selected_np else auto_highlight_noun_phrase(selected_fact) result = generate_single_qna(selected_fact, noun_phrase, min_len, max_len, temperature, top_k, top_p) results.append(result) else: for fact in facts: noun_phrase = auto_highlight_noun_phrase(fact) result = generate_single_qna(fact, noun_phrase, min_len, max_len, temperature, top_k, top_p) results.append(result) return json.dumps(results, indent=2, ensure_ascii=False) def save_json_to_dataset(json_str): try: hf_token = os.environ.get("QandA_Generator") if not hf_token: return "❌ HF_TOKEN not found in environment." repo_id = "University_Inquiries_AI_Chatbot" dataset_file = "dataset.json" local_dir = "hf_repo" repo = Repository( local_dir=local_dir, clone_from=f"datasets/{repo_id}", use_auth_token=hf_token ) repo.git_pull() full_path = os.path.join(local_dir, dataset_file) if os.path.exists(full_path): with open(full_path, "r", encoding="utf-8") as f: existing_data = json.load(f) else: existing_data = [] new_data = json.loads(json_str) now = datetime.now() for entry in new_data: entry["month"] = now.strftime("%B") entry["year"] = now.year updated_data = existing_data + new_data with open(full_path, "w", encoding="utf-8") as f: json.dump(updated_data, f, indent=2, ensure_ascii=False) repo.push_to_hub(commit_message="📥 Add new Q&A with timestamp") return "✅ Data with timestamp successfully pushed to HF dataset!" except Exception as e: return f"❌ Error: {str(e)}" def on_extract_facts(text): facts = extract_paragraph_facts(text) default_fact = facts[0] if facts else None return gr.update(choices=facts, value=default_fact), gr.update(choices=[], value=None) def on_select_fact(fact): noun_phrases = extract_noun_phrases(fact) return gr.update(choices=noun_phrases, value=noun_phrases[0] if noun_phrases else None) def main(): with gr.Blocks() as demo: gr.Markdown("## Paragraph-to-Question Generator (Auto Q&A for HF Dataset)") input_text = gr.Textbox(lines=10, label="Enter Data (Seperated by paragraph per question)") with gr.Accordion("⚙️ Customize Question Generation", open=False): extract_btn = gr.Button("Extract & Customize") fact_dropdown = gr.Dropdown(label="Select a Fact", interactive=True) np_dropdown = gr.Dropdown(label="Select Noun Phrase to Highlight (optional)", interactive=True) extract_btn.click(fn=on_extract_facts, inputs=input_text, outputs=[fact_dropdown, np_dropdown]) fact_dropdown.change(fn=on_select_fact, inputs=fact_dropdown, outputs=np_dropdown) gr.Markdown("🔽 **Min Length**: Minimum number of tokens in the generated question.") min_len = gr.Slider(5, 50, value=10, step=1, label="Min Length") gr.Markdown("🔼 **Max Length**: Maximum number of tokens in the generated question.") max_len = gr.Slider(20, 100, value=64, step=1, label="Max Length") gr.Markdown("🌡️ **Temperature**: Controls randomness. Lower = more predictable, higher = more creative.") temperature = gr.Slider(0.1, 1.5, value=1.0, step=0.1, label="Temperature") gr.Markdown("🎯 **Top-k Sampling**: Limits sampling to the top-k most likely words.") top_k = gr.Slider(0, 100, value=50, step=1, label="Top-k") gr.Markdown("🎲 **Top-p (Nucleus Sampling)**: Selects from the smallest set of words with a cumulative probability > p.") top_p = gr.Slider(0.1, 1.0, value=0.95, step=0.05, label="Top-p") gr.Markdown("✏️ You can manually edit the generated JSON here or paste your own in the same format.") output_json = gr.Textbox( lines=14, label="Q&A JSON", interactive=True, placeholder='{\n"question": "Your question?",\n"answer": "Your answer."\n},' ) with gr.Row(): generate_btn = gr.Button("Generate Q&A") send_btn = gr.Button("📤 Send to Dataset") generate_btn.click( fn=generate_qna_all, inputs=[input_text, fact_dropdown, np_dropdown, min_len, max_len, temperature, top_k, top_p], outputs=output_json ) send_status = gr.Textbox(label="Save Status", interactive=False) send_btn.click(fn=save_json_to_dataset, inputs=output_json, outputs=send_status) demo.launch() if __name__ == "__main__": main()