Update app.py

app.py

@@ -1,21 +1,16 @@
+import gradio as gr
 import pandas as pd
 import json
-import gradio as gr
 from pathlib import Path
-from tempfile import NamedTemporaryFile
 from sentence_transformers import CrossEncoder
 import numpy as np
 from time import perf_counter
-from backend.semantic_search import table, retriever
-import os
-import logging
-
-# Add Groq imports
+from pydantic import BaseModel, Field
 from phi.agent import Agent
 from phi.model.groq import Groq
-
-VECTOR_COLUMN_NAME = "vector"
-TEXT_COLUMN_NAME = "text"
+from tempfile import NamedTemporaryFile
+import os
+import logging
 
 # Set up logging
 logging.basicConfig(level=logging.INFO)
@@ -26,355 +21,244 @@ api_key = os.getenv("GROQ_API_KEY")
 if not api_key:
     gr.Warning("GROQ_API_KEY not found. Set it in 'Repository secrets'.")
     logger.error("GROQ_API_KEY not found.")
-    api_key = ""
 else:
     os.environ["GROQ_API_KEY"] = api_key
 
-# Initialize Groq Agent for quiz generation
-quiz_agent = Agent(
-    name="Quiz Generation Assistant",
-    role="You are an expert quiz generator for Customs Manual content",
+# Pydantic Model for Quiz Structure
+class QuizItem(BaseModel):
+    question: str = Field(..., description="The quiz question")
+    choices: list[str] = Field(..., description="List of 4 multiple-choice options")
+    correct_answer: str = Field(..., description="The correct choice (e.g., 'C1')")
+
+class QuizOutput(BaseModel):
+    items: list[QuizItem] = Field(..., description="List of 10 quiz items")
+
+# Initialize Quiz Generator Agent
+quiz_generator = Agent(
+    name="Quiz Generator",
+    role="Generates structured quiz questions and answers",
     instructions=[
-        "You are an expert quiz generator specializing in Customs Manual and trade-related topics.",
-        "Create exactly 10 multiple choice questions with 4 options each.",
-        "Base questions strictly on the provided documents from Customs Manual.",
-        "Return the response in the exact JSON format requested.",
-        "Ensure questions match the specified difficulty level.",
-        "Make sure all questions are clear and relevant to customs procedures.",
-        "Focus on practical customs knowledge for training officers."
+        "Create 10 questions with 4 choices each based on the provided topic and documents.",
+        "Use the specified difficulty level (easy, average, hard) to adjust question complexity.",
+        "Ensure questions are derived only from the provided documents.",
+        "Return the output in a structured format using the QuizOutput Pydantic model.",
+        "Each question should have a unique correct answer from the choices (labeled C1, C2, C3, C4)."
     ],
     model=Groq(id="llama3-70b-8192", api_key=api_key),
-    markdown=False
+    response_model=QuizOutput,
+    markdown=True
 )
 
-def retrieve_documents_for_quiz(query, cross_encoder_choice):
-    """Retrieve and rank documents for quiz generation"""
-    top_rerank = 25
-    top_k_rank = 15
-    
-    try:
-        start_time = perf_counter()
-        
-        # Encode query and search documents
-        query_vec = retriever.encode(query)
-        documents = table.search(query_vec, vector_column_name=VECTOR_COLUMN_NAME).limit(top_rerank).to_list()
-        documents = [doc[TEXT_COLUMN_NAME] for doc in documents]
-        
-        # Re-rank documents using cross-encoder
-        if cross_encoder_choice == '(ACCURATE) BGE reranker':
-            cross_encoder_model = CrossEncoder('BAAI/bge-reranker-base')
-        else:
-            cross_encoder_model = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
-        
-        query_doc_pair = [[query, doc] for doc in documents]
-        cross_scores = cross_encoder_model.predict(query_doc_pair)
-        sim_scores_argsort = list(reversed(np.argsort(cross_scores)))
-        documents = [documents[idx] for idx in sim_scores_argsort[:top_k_rank]]
-        
-        logger.info(f"Document retrieval took {perf_counter() - start_time:.2f} seconds")
-        return documents
-    
-    except Exception as e:
-        logger.error(f"Error in document retrieval: {e}")
-        return []
+VECTOR_COLUMN_NAME = "vector"
+TEXT_COLUMN_NAME = "text"
+proj_dir = Path.cwd()
 
-def generate_quiz_with_agent(topic, difficulty, documents):
-    """Generate quiz using Groq agent"""
-    context = "\n\n".join(documents) if documents else ""
-    
-    prompt = f"""You are a great teacher and your task is to create 10 questions with 4 choices with {difficulty} difficulty about the topic "{topic}" only from the below given documents from Customs Manual:
-
-{context}
-
-Create questions and answers in JSON format with the following structure:
-- Questions as "Q1": "question text" to "Q10": "question text"
-- Four choices as "Q1:C1": "choice 1" to "Q1:C4": "choice 4" for each question
-- Answers as "A1": "Q1:C#" to "A10": "Q10:C#" where # is the correct choice number
-
-Example format:
-{{
-    "Q1": "What is the procedure for customs clearance?",
-    "Q1:C1": "Submit documents online",
-    "Q1:C2": "Visit customs office directly", 
-    "Q1:C3": "File declaration and pay duties",
-    "Q1:C4": "Contact shipping agent only",
-    "A1": "Q1:C3"
-}}
-
-Provide complete JSON with all 10 questions, their choices, and answers based strictly on the provided Customs Manual content."""
-    
+# Import backend functions
+from backend.semantic_search import table, retriever
+
+def generate_quiz_data(question_difficulty, topic, documents_str):
+    prompt = f"""Generate a quiz with {question_difficulty} difficulty on topic '{topic}' using only the following documents:\n{documents_str}"""
     try:
-        response = quiz_agent.run(prompt)
-        response_text = response.content if hasattr(response, 'content') else str(response)
-        return response_text
+        response = quiz_generator.run(prompt)
+        return response.content
     except Exception as e:
-        logger.error(f"Error in quiz generation: {e}")
+        logger.error(f"Failed to generate quiz: {e}")
         return None
 
-def json_to_excel(output_json):
-    """Convert quiz JSON to Excel format"""
+def quiz_to_excel(quiz_data):
+    """Convert quiz data to Excel format"""
+    if not quiz_data or not quiz_data.items:
+        return None
+    
+    gr.Warning('Generating Excel file...', duration=10)
     data = []
-    gr.Warning('Generating Shareable file link..', duration=30)
     
-    for i in range(1, 11):
-        question_key = f"Q{i}"
-        answer_key = f"A{i}"
-
-        question = output_json.get(question_key, '')
-        correct_answer_key = output_json.get(answer_key, '')
-        correct_answer = correct_answer_key.split(':')[-1].replace('C', '').strip() if correct_answer_key else ''
-
-        # Extract options
-        option_keys = [f"{question_key}:C{j}" for j in range(1, 5)]
-        options = [output_json.get(key, '') for key in option_keys]
+    for i, item in enumerate(quiz_data.items, 1):
+        # Get correct answer index
+        correct_answer_index = int(item.correct_answer[1]) - 1  # 'C3' -> index 2
         
-        # Add data row
-        data.append([
-            question,                     # Question Text
-            "Multiple Choice",            # Question Type
-            options[0] if len(options) > 0 else '',  # Option 1
-            options[1] if len(options) > 1 else '',  # Option 2
-            options[2] if len(options) > 2 else '',  # Option 3
-            options[3] if len(options) > 3 else '',  # Option 4
-            '',                          # Option 5 (empty)
-            correct_answer,              # Correct Answer
-            30,                          # Time in seconds
-            ''                           # Image Link
-        ])
-
+        # Prepare row data
+        row = [
+            item.question,                      # Question Text
+            "Multiple Choice",                  # Question Type
+            item.choices[0] if len(item.choices) > 0 else '',  # Option 1
+            item.choices[1] if len(item.choices) > 1 else '',  # Option 2
+            item.choices[2] if len(item.choices) > 2 else '',  # Option 3
+            item.choices[3] if len(item.choices) > 3 else '',  # Option 4
+            '',                                 # Option 5 (empty)
+            str(correct_answer_index + 1),      # Correct Answer (1-4)
+            30,                                 # Time in seconds
+            ''                                  # Image Link
+        ]
+        data.append(row)
+    
     # Create DataFrame
     df = pd.DataFrame(data, columns=[
-        "Question Text", "Question Type", "Option 1", "Option 2", "Option 3", 
-        "Option 4", "Option 5", "Correct Answer", "Time in seconds", "Image Link"
+        "Question Text",
+        "Question Type", 
+        "Option 1",
+        "Option 2",
+        "Option 3",
+        "Option 4",
+        "Option 5",
+        "Correct Answer",
+        "Time in seconds",
+        "Image Link"
     ])
-
+    
+    # Save to temporary file
     temp_file = NamedTemporaryFile(delete=False, suffix=".xlsx")
     df.to_excel(temp_file.name, index=False)
     return temp_file.name
 
-# Global variable to store quiz data
-quiz_data = None
+def retrieve_and_generate_quiz(question_difficulty, topic):
+    gr.Warning('Generating quiz may take 1-2 minutes. Please wait.', duration=60)
+    top_k_rank = 10
+    documents = []
 
-def generate_quiz_function(topic, difficulty, cross_encoder_choice):
-    """Main quiz generation function"""
-    global quiz_data
-    
-    if not topic.strip():
-        return "Please enter a topic for the quiz.", [], None
-    
-    gr.Warning('Generating Quiz may take 1-2 minutes. Please wait.', duration=60)
+    document_start = perf_counter()
+    query_vec = retriever.encode(topic)
+    documents = [doc[TEXT_COLUMN_NAME] for doc in table.search(query_vec, vector_column_name=VECTOR_COLUMN_NAME).limit(top_k_rank).to_list()]
     
-    # Retrieve documents
-    documents = retrieve_documents_for_quiz(topic, cross_encoder_choice)
+    # Apply BGE reranker
+    cross_encoder = CrossEncoder('BAAI/bge-reranker-base')
+    query_doc_pair = [[topic, doc] for doc in documents]
+    cross_scores = cross_encoder.predict(query_doc_pair)
+    sim_scores_argsort = list(reversed(np.argsort(cross_scores)))
+    documents = [documents[idx] for idx in sim_scores_argsort[:top_k_rank]]
+
+    documents_str = '\n'.join(documents)
+    quiz_data = generate_quiz_data(question_difficulty, topic, documents_str)
     
-    if not documents:
-        return "No relevant documents found for this topic. Please try a different topic.", [], None
+    # Generate Excel file
+    excel_file = None
+    if quiz_data:
+        excel_file = quiz_to_excel(quiz_data)
     
-    # Generate quiz
-    response_text = generate_quiz_with_agent(topic, difficulty, documents)
+    return quiz_data, excel_file
+
+def update_quiz_components(quiz_data, excel_file):
+    if not quiz_data or not quiz_data.items:
+        return [gr.update(visible=False) for _ in range(10)] + [gr.update(value="Error: Failed to generate quiz.", visible=True), None]
     
-    if not response_text:
-        return "Error generating quiz. Please try again.", [], None
+    radio_updates = []
+    for i, item in enumerate(quiz_data.items[:10]):
+        choices = item.choices
+        radio_update = gr.update(visible=True, choices=choices, label=item.question, value=None)
+        radio_updates.append(radio_update)
     
-    try:
-        # Extract JSON from response
-        start_index = response_text.find('{')
-        end_index = response_text.rfind('}')
-        cleaned_response = response_text[start_index:end_index + 1] if start_index != -1 and end_index != -1 else ''
-        
-        if not cleaned_response:
-            return "Could not extract quiz data. Please try again.", [], None
-        
-        output_json = json.loads(cleaned_response)
-        quiz_data = output_json
-        
-        # Generate Excel file
-        excel_file = json_to_excel(output_json)
-        
-        # Create question components
-        question_components = []
-        for question_num in range(1, 11):
-            question_key = f"Q{question_num}"
-            
-            question_text = output_json.get(question_key)
-            if not question_text:
-                continue
-                
-            choice_keys = [f"{question_key}:C{i}" for i in range(1, 5)]
-            choice_list = [output_json.get(choice_key, f"Choice {i} not found") for i, choice_key in enumerate(choice_keys, 1)]
-            
-            question_components.append((question_text, choice_list))
-        
-        return "Quiz Generated Successfully!", question_components, excel_file
-        
-    except json.JSONDecodeError as e:
-        logger.error(f"Failed to decode JSON: {e}")
-        return "Error parsing quiz data. Please try again.", [], None
-    except Exception as e:
-        logger.error(f"Error in quiz generation: {e}")
-        return "Error generating quiz. Please try again.", [], None
+    return radio_updates + [gr.update(value="Please select answers and click 'Check Score'.", visible=True), excel_file]
 
-def check_answers_function(answers):
-    """Check quiz answers and calculate score"""
-    global quiz_data
+def collect_answers_and_calculate(*all_inputs):
+    print(f"Total inputs received: {len(all_inputs)}")
     
-    if not quiz_data:
-        return "No quiz data available. Please generate a quiz first."
+    # The last input is quiz_data, the first 10 are radio values
+    radio_values = all_inputs[:10]
+    quiz_data = all_inputs[10]
     
+    print(f"Received radio_values: {radio_values}")
+    print(f"Received quiz_data: {quiz_data}")
+    
+    # Calculate score
     score = 0
-    total_questions = 0
+    answered_questions = 0
     
-    for question_num in range(1, 11):
-        answer_key = f"A{question_num}"
-        correct_answer_key = quiz_data.get(answer_key)
-        
-        if not correct_answer_key:
-            continue
+    for i, (user_answer, quiz_item) in enumerate(zip(radio_values, quiz_data.items[:10])):
+        if user_answer is not None:
+            answered_questions += 1
             
-        correct_answer = quiz_data.get(correct_answer_key, "")
-        user_answer = answers[question_num - 1] if question_num - 1 < len(answers) else ""
-        
-        total_questions += 1
-        if user_answer == correct_answer:
-            score += 1
-    
-    if score > 7:
-        message = f"### Excellent! You got {score} out of {total_questions}!"
-    elif score > 5:
-        message = f"### Good! You got {score} out of {total_questions}!"
+            # Convert correct answer code to actual choice text
+            correct_answer_index = int(quiz_item.correct_answer[1]) - 1
+            correct_answer_text = quiz_item.choices[correct_answer_index]
+            
+            print(f"Q{i+1}: User='{user_answer}' vs Correct='{correct_answer_text}'")
+            
+            if user_answer == correct_answer_text:
+                score += 1
+    
+    print(f"Calculated score: {score}/{answered_questions}")
+    
+    # Create simple score message
+    if answered_questions == 0:
+        message = "⚠�� Please answer at least one question!"
+    elif score == answered_questions:
+        message = f"🏆 PERFECT SCORE! You got {score} out of {answered_questions} correct! Outstanding performance! 🌟"
+    elif score > answered_questions * 0.7:
+        message = f"🎉 EXCELLENT! You got {score} out of {answered_questions} correct! Great job! 💪"
+    elif score > answered_questions * 0.5:
+        message = f"👍 GOOD JOB! You got {score} out of {answered_questions} correct! Well done! 📚"
     else:
-        message = f"### You got {score} out of {total_questions}! Don't worry. You can prepare well and try better next time!"
+        message = f"💪 KEEP TRYING! You got {score} out of {answered_questions} correct! Practice makes perfect! 📖✨"
     
     return message
 
-# Gradio Interface
-with gr.Blocks(title="ADWITIYA Quiz Maker", theme=gr.themes.Default(primary_hue="cyan", secondary_hue="yellow", neutral_hue="purple")) as demo:
-    
-    # Header section
+# Define theme
+colorful_theme = gr.themes.Default(primary_hue="cyan", secondary_hue="yellow", neutral_hue="purple")
+
+with gr.Blocks(title="CBSE Gyan Quiz Bot", theme=colorful_theme) as QUIZBOT:
+    # Header
     with gr.Row():
         with gr.Column(scale=2):
-            try:
-                gr.Image(value='logo.png', height=200, width=200)
-            except:
-                gr.HTML("<div style='height: 200px; width: 200px; background-color: #f0f0f0; display: flex; align-items: center; justify-content: center;'>Logo</div>")
+            gr.Image(value='logo.png', height=200, width=200)
         with gr.Column(scale=6):
             gr.HTML("""
             <center>
-                <h1><span style="color: purple;">ADWITIYA</span> Customs Manual Quizbot</h1>
-                <h2>Generative AI-powered Capacity building for Training Officers</h2>
-                <i>⚠️ NACIN Faculties create quiz from any topic dynamically for classroom evaluation after their sessions ! ⚠️</i>
+                <h1><span style="color: purple;">CBSE GYAN</span> Quiz Bot</h1>
+                <h2>Generative AI-powered Learning for CBSE Students</h2>
+                <i>⚠️ Students can create quiz from any topic from CBSE curriculum and evaluate themselves! ⚠️</i>
             </center>
             """)
-    
-    # Input components
-    topic_input = gr.Textbox(
-        label="Enter the Topic for Quiz", 
-        placeholder="Write any topic/details from Customs Manual"
-    )
-    
+
+    # Input controls
+    topic = gr.Textbox(label="Enter the Topic for Quiz", placeholder="Write any CHAPTER NAME from CBSE curriculum")
+
     with gr.Row():
-        difficulty_radio = gr.Radio(
-            choices=["easy", "average", "hard"], 
-            label="How difficult should the quiz be?",
-            value="average"
-        )
-        cross_encoder_radio = gr.Radio(
-            choices=['(FAST) MiniLM-L6v2', '(ACCURATE) BGE reranker'],
-            value='(ACCURATE) BGE reranker',
-            label="Embeddings Model",
-            info="Select the model for document ranking"
-        )
-    
-    generate_btn = gr.Button("Generate Quiz!🚀", variant="primary")
-    status_text = gr.Textbox(label="Status", interactive=False)
-    
-    # Quiz display area
-    quiz_state = gr.State([])
-    question_radios = []
-    
-    with gr.Column(visible=False) as quiz_section:
-        gr.Markdown("## Quiz Questions")
-        for i in range(10):
-            question_radios.append(
-                gr.Radio(label=f"Question {i+1}", visible=False, interactive=True)
-            )
-        
-        check_btn = gr.Button("Check Score", variant="secondary")
-        score_display = gr.Markdown()
+        difficulty_radio = gr.Radio(["easy", "average", "hard"], label="How difficult should the quiz be?")
+        model_radio = gr.Radio(choices=['(ACCURATE) BGE reranker'], value='(ACCURATE) BGE reranker', label="Embeddings")
+
+    # Generate quiz button
+    generate_quiz_btn = gr.Button("Generate Quiz!🚀")
+    quiz_msg = gr.Textbox(label="Status", interactive=False)
     
-    excel_file = gr.File(label="Download Excel", visible=False)
+    # Excel download
+    excel_download = gr.File(label="Download Excel", visible=False)
+
+    # Quiz questions (hidden initially)
+    question_radios = [gr.Radio(visible=False, label="", choices=[""], value=None) for _ in range(10)]
+    quiz_data_state = gr.State(value=None)
     
+    # Check score button
+    check_score_btn = gr.Button("Check Score", variant="primary", size="lg")
+    score_output = gr.Textbox(label="Your Results", visible=False)
+
     # Event handlers
-    def update_quiz_display(topic, difficulty, cross_encoder):
-        status, questions, excel = generate_quiz_function(topic, difficulty, cross_encoder)
-        
-        if not questions:
-            return (
-                status,
-                gr.Column(visible=False),
-                gr.File(visible=False),
-                questions,
-                *[gr.Radio(visible=False) for _ in range(10)]
-            )
-        
-        # Update question radios
-        question_updates = []
-        for i, (question_text, choices) in enumerate(questions):
-            if i < 10:
-                question_updates.append(
-                    gr.Radio(
-                        label=question_text,
-                        choices=choices,
-                        visible=True,
-                        interactive=True
-                    )
-                )
-            else:
-                question_updates.append(gr.Radio(visible=False))
-        
-        # Fill remaining slots if less than 10 questions
-        while len(question_updates) < 10:
-            question_updates.append(gr.Radio(visible=False))
-        
-        return (
-            status,
-            gr.Column(visible=True),
-            gr.File(value=excel, visible=True) if excel else gr.File(visible=False),
-            questions,
-            *question_updates
-        )
-    
-    def check_quiz_answers(*answers):
-        return check_answers_function(list(answers))
-    
-    # Connect events
-    generate_btn.click(
-        fn=update_quiz_display,
-        inputs=[topic_input, difficulty_radio, cross_encoder_radio],
-        outputs=[status_text, quiz_section, excel_file, quiz_state] + question_radios
-    )
-    
-    check_btn.click(
-        fn=check_quiz_answers,
-        inputs=question_radios,
-        outputs=score_display
+    generate_quiz_btn.click(
+        fn=retrieve_and_generate_quiz,
+        inputs=[difficulty_radio, topic],
+        outputs=[quiz_data_state, excel_download]
+    ).then(
+        fn=update_quiz_components,
+        inputs=[quiz_data_state, excel_download],
+        outputs=question_radios + [quiz_msg, excel_download]
+    ).then(
+        fn=lambda: gr.update(visible=True),
+        inputs=[],
+        outputs=[excel_download]
     )
-    
-    # Example questions
-    gr.Examples(
-        examples=[
-            'Customs clearance procedures',
-            'Import duty calculation',
-            'Export documentation requirements',
-            'Customs valuation methods',
-            'Special economic zones regulations'
-        ],
-        inputs=topic_input,
-        label="Try these example topics:"
+
+    check_score_btn.click(
+        fn=collect_answers_and_calculate,
+        inputs=question_radios + [quiz_data_state],
+        outputs=[score_output],
+        api_name="check_score"
+    ).then(
+        fn=lambda: gr.update(visible=True),
+        inputs=[],
+        outputs=[score_output]
     )
 
 if __name__ == "__main__":
-    demo.launch(server_name="0.0.0.0", server_port=7860)
+    QUIZBOT.queue().launch(server_name="0.0.0.0", server_port=7860)
 # import pandas as pd
 # import json
 # import gradio as gr