| # import streamlit as st | |
| # import json | |
| # import re | |
| # import random | |
| # from typing import Dict, List, Optional, Set | |
| # from huggingface_hub import hf_hub_download | |
| # from llama_cpp import Llama | |
| # @st.cache_resource | |
| # def load_model(): | |
| # model_path = hf_hub_download( | |
| # repo_id="Nitish035/gguf_4", | |
| # filename="unsloth.Q4_K_M.gguf", | |
| # local_dir=".", | |
| # ) | |
| # model = Llama( | |
| # model_path=model_path, | |
| # n_ctx=4096, | |
| # n_gpu_layers=35, | |
| # verbose=True, # Set to True temporarily for debugging | |
| # ) | |
| # return model | |
| # def get_difficulty(dok): | |
| # difficulty_map = { | |
| # 1: "Easy", | |
| # 2: "Medium", | |
| # 3: "Hard", | |
| # 4: "Very Hard" | |
| # } | |
| # return difficulty_map.get(dok, "Medium") | |
| # def generate_response(prompt_dict, max_tokens=3024): | |
| # try: | |
| # st.info("Generating response with model...") | |
| # dok_descriptions = { | |
| # 1: "Basic recall of math facts, definitions, or simple calculations", | |
| # 2: "Application of math concepts requiring 1-2 step procedures", | |
| # 3: "Multi-step math problems requiring analysis and justification", | |
| # 4: "Complex math problems needing synthesis and creative approaches" | |
| # } | |
| # difficulty_levels = { | |
| # "Easy": "Multi-step problems requiring reasoning and understanding of concepts - easy; Straightforward problems with obvious approaches.", | |
| # "Medium": "Multi-step problems requiring moderate reasoning and understanding of concepts - medium; Requires careful analysis but standard methods apply.", | |
| # "Hard": "Complex multi-step problems with multiple variables and operations - hard; Demands innovative thinking and multiple concepts.", | |
| # "Very Hard": "Advanced problems requiring systems of equations, conditional logic, and optimization - very hard; Requires advanced reasoning, optimization strategies, and integration of multiple topics." | |
| # } | |
| # contexts = [ | |
| # "a school fundraiser", | |
| # "a community bake sale", | |
| # "a sports team's snack stand", | |
| # "a charity event", | |
| # "a classroom project", | |
| # "real world", | |
| # "", | |
| # "", | |
| # "man", | |
| # "women" | |
| # ] | |
| # operations = [ | |
| # "addition", | |
| # "multiplication", | |
| # "division", | |
| # "subtraction", | |
| # "use numbers", | |
| # "", | |
| # "" | |
| # ] | |
| # prompt = f"""<|im_start|>user | |
| # Generate a {prompt_dict['Difficulty']} difficulty math multiple-choice question with options and correct answer with these specifications: | |
| # * Grade Level: {prompt_dict['Grade']} | |
| # * Topic: {prompt_dict['Topic']} (align with appropriate CCSS standard) | |
| # * Depth of Knowledge (DOK): Level {prompt_dict['DOK']} ({dok_descriptions[prompt_dict['DOK']]}) | |
| # * Difficulty: {prompt_dict['Difficulty']} ({difficulty_levels[prompt_dict['Difficulty']]}) | |
| # * Context: {random.choice(contexts)} | |
| # * Math Operations: {random.choice(operations)} | |
| # 1. Create a unique word problem based on the context and operations | |
| # 2. Design a question that matches DOK level {prompt_dict['DOK']} | |
| # 3. Create four plausible options with one clearly correct answer | |
| # 4. Format as a clean multiple-choice question | |
| # # Requirements: | |
| # 1. The question must be unique and different from previous questions | |
| # 2. Make sure the final answer computed in the explanation is inserted into one of the 4 options | |
| # 3. The `correct_answer` key must match the option letter that holds the correct answer | |
| # 4. Options should reflect common student misconceptions | |
| # 5. Format the response as a JSON object with these keys: 'question', 'options', 'correct_answer', 'explanation' | |
| # IMPORTANT: Only return a valid JSON object with no additional text or formatting outside the JSON structure. | |
| # <|im_end|> | |
| # <|im_start|>assistant | |
| # """ | |
| # model = load_model() | |
| # response = model.create_completion( | |
| # prompt=prompt, | |
| # max_tokens=max_tokens, | |
| # temperature=0.1, | |
| # top_p=0.9, | |
| # echo=False | |
| # ) | |
| # st.success("Response generated!") | |
| # return response['choices'][0]['text'] | |
| # except Exception as e: | |
| # st.error(f"Error generating response: {str(e)}") | |
| # return "" | |
| # def try_parse_json(response_text): | |
| # try: | |
| # # First try to find a JSON object using regex | |
| # json_pattern = r'(\{[\s\S]*\})' | |
| # match = re.search(json_pattern, response_text) | |
| # if match: | |
| # potential_json = match.group(1) | |
| # return json.loads(potential_json) | |
| # # If no match using regex, try direct JSON parse | |
| # return json.loads(response_text) | |
| # except json.JSONDecodeError as e: | |
| # st.error(f"Failed to parse JSON response: {e}") | |
| # st.text("Raw response:") | |
| # st.text(response_text) | |
| # return None | |
| # def main(): | |
| # st.title("DOK-Based Math Question Generator") | |
| # st.write("Generate math questions based on a Depth of Knowledge level") | |
| # col1, col2 = st.columns(2) | |
| # with col1: | |
| # dok_level = st.selectbox("Select DOK Level:", [1, 2, 3, 4]) | |
| # topic = st.selectbox("Select Topic:", [ | |
| # "Number System", | |
| # "Statistics and Probability", | |
| # "Geometry", | |
| # "Expressions and Equations", | |
| # "Functions", | |
| # "Ratios and Proportional" | |
| # ]) | |
| # with col2: | |
| # no_of_questions = st.slider("Number of Questions:", 1, 2) | |
| # grade_level = st.selectbox("Select Grade Level:", [6, 7, 8]) | |
| # if st.button("Generate Questions"): | |
| # with st.spinner("Generating questions..."): | |
| # difficulty = get_difficulty(dok_level) | |
| # all_questions = [] | |
| # generated_questions = set() | |
| # for i in range(no_of_questions): | |
| # st.info(f"Generating question {i+1} of {no_of_questions}...") | |
| # attempts = 0 | |
| # while attempts < 3: | |
| # prompt_dict = { | |
| # "Grade": str(grade_level), | |
| # "Topic": topic, | |
| # "DOK": dok_level, | |
| # "Difficulty": difficulty | |
| # } | |
| # response_text = generate_response(prompt_dict) | |
| # parsed_json = try_parse_json(response_text) | |
| # if parsed_json and parsed_json.get('question') and parsed_json['question'] not in generated_questions: | |
| # generated_questions.add(parsed_json['question']) | |
| # all_questions.append(parsed_json) | |
| # break | |
| # attempts += 1 | |
| # st.warning(f"Attempt {attempts}: Failed to generate unique question or parse response. Retrying...") | |
| # st.subheader("Generated Questions:") | |
| # if all_questions: | |
| # for i, question in enumerate(all_questions): | |
| # with st.expander(f"Question {i+1}"): | |
| # st.write(f"**Question:** {question.get('question', 'No question found')}") | |
| # if 'options' in question: | |
| # st.write("**Options:**") | |
| # options = question['options'] | |
| # if isinstance(options, dict): | |
| # for key, value in options.items(): | |
| # st.write(f"{key}: {value}") | |
| # else: | |
| # st.write(options) | |
| # st.write(f"**Correct Answer:** {question.get('correct_answer', 'No answer found')}") | |
| # st.write(f"**Explanation:** {question.get('explanation', 'No explanation found')}") | |
| # st.download_button( | |
| # label="Download JSON", | |
| # data=json.dumps(all_questions, indent=2), | |
| # file_name="math_questions.json", | |
| # mime="application/json", | |
| # ) | |
| # else: | |
| # st.error("Failed to generate unique questions. Please try again.") | |
| # if __name__ == "__main__": | |
| # main() | |
| import streamlit as st | |
| import json | |
| import re | |
| import random | |
| import os | |
| import time | |
| from typing import Dict, List, Optional, Set | |
| from huggingface_hub import hf_hub_download | |
| from llama_cpp import Llama | |
| # Use /tmp directory which is writable in the Docker container | |
| MODEL_CACHE_DIR = "/tmp/model_cache" | |
| os.makedirs(MODEL_CACHE_DIR, exist_ok=True) | |
| # Global variable to track if model loading was attempted | |
| model_load_attempted = False | |
| def load_model(): | |
| """Load the model with proper error handling""" | |
| try: | |
| st.info("Loading model... This may take a few minutes on first run.") | |
| model_path = hf_hub_download( | |
| repo_id="Nitish035/qwen_gguf_adapter", | |
| filename="unsloth.Q4_K_M.gguf", | |
| local_dir=MODEL_CACHE_DIR, | |
| cache_dir=MODEL_CACHE_DIR, | |
| ) | |
| # Reduce n_gpu_layers if on CPU-only environment | |
| try: | |
| # First attempt with GPU layers | |
| model = Llama( | |
| model_path=model_path, | |
| n_ctx=2048, # Reduced context size for better performance | |
| n_gpu_layers=35, | |
| verbose=False, | |
| ) | |
| except Exception as gpu_error: | |
| st.warning(f"GPU acceleration failed: {str(gpu_error)}. Falling back to CPU.") | |
| # Fall back to CPU-only | |
| model = Llama( | |
| model_path=model_path, | |
| n_ctx=2048, | |
| n_gpu_layers=0, # CPU only | |
| verbose=False, | |
| ) | |
| st.success("Model loaded successfully!") | |
| return model | |
| except Exception as e: | |
| st.error(f"Error loading model: {str(e)}") | |
| return None | |
| def get_difficulty(dok): | |
| difficulty_map = { | |
| 1: "Easy", | |
| 2: "Medium", | |
| 3: "Hard", | |
| 4: "Very Hard" | |
| } | |
| return difficulty_map.get(dok, "Medium") | |
| def generate_response(prompt_dict, max_tokens=2048): | |
| """Generate response with timing and improved error handling""" | |
| try: | |
| # Use global to check if model loading was attempted | |
| global model_load_attempted | |
| if not model_load_attempted: | |
| with st.spinner("Loading the model (first-time only)..."): | |
| model = load_model() | |
| model_load_attempted = True | |
| else: | |
| model = load_model() # This will use cached version if available | |
| if model is None: | |
| st.error("Model failed to load. Please check logs or try again.") | |
| return "" | |
| dok_descriptions = { | |
| 1: "Basic recall of math facts, definitions, or simple calculations", | |
| 2: "Application of math concepts requiring 1-2 step procedures", | |
| 3: "Multi-step math problems requiring analysis and justification", | |
| 4: "Complex math problems needing synthesis and creative approaches" | |
| } | |
| difficulty_levels = { | |
| "Easy": "Multi-step problems requiring reasoning and understanding of concepts - easy", | |
| "Medium": "Multi-step problems requiring moderate reasoning - medium", | |
| "Hard": "Complex multi-step problems with multiple variables - hard", | |
| "Very Hard": "Advanced problems requiring systems of equations - very hard" | |
| } | |
| contexts = [ | |
| "a school fundraiser", | |
| "a community event", | |
| "a sports competition", | |
| "a classroom project", | |
| "real world situation" | |
| ] | |
| operations = [ | |
| "addition", | |
| "multiplication", | |
| "division", | |
| "subtraction", | |
| "basic calculation" | |
| ] | |
| # Simplified prompt for faster generation | |
| prompt = f"""<|im_start|>user | |
| Generate a {prompt_dict['Difficulty']} math multiple-choice question: | |
| * Grade: {prompt_dict['Grade']} | |
| * Topic: {prompt_dict['Topic']} | |
| * DOK: Level {prompt_dict['DOK']} ({dok_descriptions[prompt_dict['DOK']]}) | |
| * Context: {random.choice(contexts)} | |
| * Operations: {random.choice(operations)} | |
| Create a word problem with four options (A-D), one correct answer. Format as JSON with keys: 'question', 'options', 'correct_answer', 'explanation'. | |
| <|im_end|> | |
| <|im_start|>assistant | |
| """ | |
| # Time the generation | |
| start_time = time.time() | |
| response = model.create_completion( | |
| prompt=prompt, | |
| max_tokens=max_tokens, | |
| temperature=0.2, # Lower temperature for more deterministic output | |
| top_p=0.9, | |
| stop=["<|im_end|>", "<|im_start|>"], # Add explicit stop tokens | |
| echo=False | |
| ) | |
| end_time = time.time() | |
| generation_time = end_time - start_time | |
| result_text = response['choices'][0]['text'] | |
| st.session_state["last_generation_time"] = generation_time | |
| return result_text | |
| except Exception as e: | |
| st.error(f"Error generating response: {str(e)}") | |
| st.session_state["last_generation_time"] = 0 | |
| return "" | |
| def try_parse_json(response_text): | |
| """More robust JSON parsing""" | |
| try: | |
| # First try to clean up the response for better parsing | |
| # Remove any trailing commas in JSON objects (common error) | |
| cleaned_text = re.sub(r',\s*}', '}', response_text) | |
| cleaned_text = re.sub(r',\s*]', ']', cleaned_text) | |
| # Find JSON pattern | |
| json_pattern = r'(\{[\s\S]*\})' | |
| match = re.search(json_pattern, cleaned_text) | |
| if match: | |
| potential_json = match.group(1) | |
| return json.loads(potential_json) | |
| # If no match using regex, try direct JSON parse | |
| return json.loads(cleaned_text) | |
| except json.JSONDecodeError as e: | |
| st.error(f"Failed to parse JSON response. Error: {e}") | |
| return None | |
| def main(): | |
| st.title("DOK-Based Math Question Generator") | |
| st.write("Generate math questions based on a Depth of Knowledge level") | |
| # Initialize session state for results | |
| if "results" not in st.session_state: | |
| st.session_state["results"] = [] | |
| if "last_generation_time" not in st.session_state: | |
| st.session_state["last_generation_time"] = 0 | |
| col1, col2 = st.columns(2) | |
| with col1: | |
| dok_level = st.selectbox("Select DOK Level:", [1, 2, 3, 4], index=0) | |
| topic = st.selectbox("Select Topic:", [ | |
| "Number System", | |
| "Statistics and Probability", | |
| "Geometry", | |
| "Expressions and Equations", | |
| "Functions", | |
| "Ratios and Proportional" | |
| ]) | |
| with col2: | |
| no_of_questions = st.slider("Number of Questions:", 1, 2, value=1) | |
| grade_level = st.selectbox("Select Grade Level:", [6, 7, 8]) | |
| if st.button("Generate Questions"): | |
| # Clear previous results | |
| st.session_state["results"] = [] | |
| all_questions = [] | |
| generated_questions = set() | |
| time_stats = [] | |
| progress_bar = st.progress(0) | |
| question_status = st.empty() | |
| for i in range(no_of_questions): | |
| question_status.write(f"Generating question {i+1} of {no_of_questions}...") | |
| attempts = 0 | |
| success = False | |
| while attempts < 3 and not success: | |
| prompt_dict = { | |
| "Grade": str(grade_level), | |
| "Topic": topic, | |
| "DOK": dok_level, | |
| "Difficulty": get_difficulty(dok_level) | |
| } | |
| response_text = generate_response(prompt_dict) | |
| parsed_json = try_parse_json(response_text) | |
| if parsed_json and parsed_json.get('question') and parsed_json['question'] not in generated_questions: | |
| generated_questions.add(parsed_json['question']) | |
| all_questions.append(parsed_json) | |
| # Store generation time | |
| generation_time = st.session_state.get("last_generation_time", 0) | |
| time_stats.append({ | |
| "question_number": i+1, | |
| "generation_time": round(generation_time, 2), | |
| "attempts": attempts+1 | |
| }) | |
| success = True | |
| else: | |
| attempts += 1 | |
| st.warning(f"Attempt {attempts}: Failed to generate valid question. Retrying...") | |
| # Update progress | |
| progress_bar.progress((i+1)/no_of_questions) | |
| # Clear status | |
| question_status.empty() | |
| # Store results in session state | |
| st.session_state["results"] = all_questions | |
| st.session_state["time_stats"] = time_stats | |
| # Display results | |
| display_results(all_questions, time_stats) | |
| def display_results(questions, time_stats): | |
| """Display results and time statistics""" | |
| if not questions: | |
| st.error("Failed to generate any questions. Please try again.") | |
| return | |
| st.subheader("Generated Questions:") | |
| # Display each question in an expander | |
| for i, question in enumerate(questions): | |
| with st.expander(f"Question {i+1}", expanded=True): | |
| st.markdown(f"**Question:** {question.get('question', 'No question found')}") | |
| if 'options' in question: | |
| st.markdown("**Options:**") | |
| options = question['options'] | |
| if isinstance(options, dict): | |
| for key, value in options.items(): | |
| st.markdown(f"**{key}:** {value}") | |
| else: | |
| st.write(options) | |
| st.markdown(f"**Correct Answer:** {question.get('correct_answer', 'No answer found')}") | |
| st.markdown(f"**Explanation:** {question.get('explanation', 'No explanation found')}") | |
| # Display timing information | |
| st.subheader("Generation Time Statistics:") | |
| for stat in time_stats: | |
| st.write(f"Question {stat['question_number']}: Generated in {stat['generation_time']} seconds (after {stat['attempts']} attempt(s))") | |
| # Calculate and display average generation time | |
| if time_stats: | |
| avg_time = sum(stat['generation_time'] for stat in time_stats) / len(time_stats) | |
| st.write(f"Average generation time: {round(avg_time, 2)} seconds per question") | |
| # Add download button | |
| st.download_button( | |
| label="Download Questions as JSON", | |
| data=json.dumps(questions, indent=2), | |
| file_name="math_questions.json", | |
| mime="application/json", | |
| ) | |
| if __name__ == "__main__": | |
| main() |