Update codeexecutor.py

only majority and steps

codeexecutor.py

@@ -1,115 +1,160 @@
-import os
-import re
-import subprocess
-import tempfile
-import multiprocessing
-from collections import Counter
-from contextlib import contextmanager
-from dataclasses import dataclass
+import gradio as gr
+import ctranslate2
+from transformers import AutoTokenizer
+from huggingface_hub import snapshot_download
+from codeexecutor import postprocess_completion, get_majority_vote
 
+# Define the model and tokenizer loading
+model_prompt = "Solve the following mathematical problem: "
+tokenizer = AutoTokenizer.from_pretrained("AI-MO/NuminaMath-7B-TIR")
+model_path = snapshot_download(repo_id="Makima57/deepseek-math-Numina")
+generator = ctranslate2.Generator(model_path, device="cpu", compute_type="int8")
+iterations = 10
 
-class PythonREPL:
-    def __init__(self, timeout=5):
-        self.timeout = timeout
+# Function to generate predictions using the model
+def get_prediction(question):
+    input_text = model_prompt + question
+    input_tokens = tokenizer.tokenize(input_text)
+    results = generator.generate_batch([input_tokens])
+    output_tokens = results[0].sequences[0]
+    predicted_answer = tokenizer.convert_tokens_to_string(output_tokens)
+    return predicted_answer
 
-    @staticmethod
-    def _run_code(temp_file_path):
-        result = subprocess.run(
-            ["python3", temp_file_path],
-            capture_output=True,
-            check=False,
-            text=True
-        )
-        if result.returncode == 0:
-            return True, result.stdout.strip()
-        else:
-            error_msg = result.stderr.strip()
-            msgs = error_msg.split("\n")
-            new_msgs = []
-            want_next = False
-            for m in msgs:
-                if "Traceback" in m:
-                    new_msgs.append(m)
-                elif m == msgs[-1]:
-                    new_msgs.append(m)
-                elif temp_file_path in m:
-                    st = m.index('"/') + 1 if '"/' in m else 0
-                    ed = m.index(temp_file_path) + 1 if temp_file_path in m else None
-                    clr = m[st:ed] if not ed else m[st:]
-                    m = m.replace(clr, "")
-                    new_msgs.append(m)
-                    want_next = True
-                elif want_next:
-                    new_msgs.append(m)
-                    want_next = False
-            return False, "\n".join(new_msgs).strip()
+# Function to perform majority voting and solve the problem with steps
+def majority_vote_with_steps(question, num_iterations=10):
+    all_predictions = []
+    all_answer = []
+    steps_to_solve = []
+    
+    for _ in range(num_iterations):
+        prediction = get_prediction(question)
+        # Process prediction to get steps and answer
+        answer, success = postprocess_completion(prediction, True, True)
+        all_predictions.append(prediction)
+        all_answer.append(answer)
+        if success:
+            steps_to_solve.append(answer)  # Add the steps if code executes successfully
+    
+    majority_voted_ans = get_majority_vote(all_answer)
+    
+    # If steps to solve exist, return them, else fallback to "No steps found"
+    steps_solution = steps_to_solve[0] if steps_to_solve else "No steps found"
+    
+    return majority_voted_ans, steps_solution
 
-    def __call__(self, query):
-        query = "import math\nimport numpy as np\nimport sympy as sp\n" + query
-        query = query.strip().split("\n")
-        if "print(" not in query[-1]:
-            if "#" in query[-1]:
-                query[-1] = query[-1].split("#")[0]
-            query[-1] = "print(" + query[-1] + ")"
-        query = "\n".join(query)
-        
-        with tempfile.TemporaryDirectory() as temp_dir:
-            temp_file_path = os.path.join(temp_dir, "tmp.py")
-            with open(temp_file_path, "w", encoding="utf-8") as f:
-                f.write(query)
+# Gradio interface for user input and output
+def gradio_interface(question, correct_answer):
+    final_answer, steps_solution = majority_vote_with_steps(question, iterations)
+    return {
+        "Question": question,
+        "Majority-Voted Answer": final_answer,
+        "Steps to Solve": steps_solution,
+        "Correct Solution": correct_answer
+    }
 
-            with multiprocessing.Pool(1) as pool:
-                result = pool.apply_async(self._run_code, (temp_file_path,))
-                try:
-                    success, output = result.get(self.timeout)
-                except multiprocessing.TimeoutError:
-                    pool.terminate()
-                    return False, f"Timed out after {self.timeout} seconds."
-        return success, output
+# Custom CSS for enhanced design
+custom_css = """
+    body {
+        background-color: #fafafa;
+        font-family: 'Open Sans', sans-serif;
+    }
+    .gradio-container {
+        background-color: #ffffff;
+        border: 3px solid #007acc;
+        border-radius: 15px;
+        padding: 20px;
+        box-shadow: 0 8px 20px rgba(0, 0, 0, 0.15);
+        max-width: 800px;
+        margin: 50px auto;
+    }
+    h1 {
+        font-family: 'Poppins', sans-serif;
+        color: #007acc;
+        font-weight: bold;
+        font-size: 32px;
+        text-align: center;
+        margin-bottom: 20px;
+    }
+    p {
+        font-family: 'Roboto', sans-serif;
+        font-size: 18px;
+        color: #333;
+        text-align: center;
+        margin-bottom: 15px;
+    }
+    input, textarea {
+        font-family: 'Montserrat', sans-serif;
+        font-size: 16px;
+        padding: 10px;
+        border: 2px solid #007acc;
+        border-radius: 10px;
+        background-color: #f1f8ff;
+        margin-bottom: 15px;
+    }
+    #math_question, #correct_answer {
+        font-size: 20px;
+        font-family: 'Poppins', sans-serif;
+        font-weight: 500px;  /* Apply bold */
+        color: #007acc;
+        margin-bottom: 5px;
+        display: inline-block;
+    }
+    
+    textarea {
+        min-height: 150px;
+    }
+    .gr-button-primary {
+        background-color: #007acc !important;
+        color: white !important;
+        border-radius: 10px !important;
+        font-size: 18px !important;
+        font-weight: bold !important;
+        padding: 10px 20px !important;
+        font-family: 'Montserrat', sans-serif !important;
+        transition: background-color 0.3s ease !important;
+    }
+    .gr-button-primary:hover {
+        background-color: #005f99 !important;
+    }
+    .gr-button-secondary {
+        background-color: #f44336 !important;
+        color: white !important;
+        border-radius: 10px !important;
+        font-size: 18px !important;
+        font-weight: bold !important;
+        padding: 10px 20px !important;
+        font-family: 'Montserrat', sans-serif !important;
+        transition: background-color 0.3s ease !important;
+    }
+    .gr-button-secondary:hover {
+        background-color: #c62828 !important;
+    }
+    .gr-output {
+        background-color: #e0f7fa;
+        border: 2px solid #007acc;
+        border-radius: 10px;
+        padding: 15px;
+        font-size: 16px;
+        font-family: 'Roboto', sans-serif;
+        font-weight: bold;
+        color: #00796b;
+    }
+"""
 
+# Gradio app setup
+interface = gr.Interface(
+    fn=gradio_interface,
+    inputs=[
+        gr.Textbox(label="🧠 Math Question", placeholder="Enter your math question here...", elem_id="math_question"),
+        gr.Textbox(label="✅ Correct Answer", placeholder="Enter the correct answer here...", elem_id="correct_answer"),
+    ],
+    outputs=[
+        gr.JSON(label="📊 Results"),  # Display the results in a JSON format
+    ],
+    title="🔢 Math Question Solver",
+    description="Enter a math question to get the model's majority-voted answer and steps to solve the problem.",
+    css=custom_css  # Apply custom CSS
+)
 
-def execute_completion(executor, completion, return_status, last_code_block):
-    executions = re.findall(r"```python(.*?)```", completion, re.DOTALL)
-    if len(executions) == 0:
-        return completion, False if return_status else completion
-    if last_code_block:
-        executions = [executions[-1]]
-    outputs = []
-    successes = []
-    for code in executions:
-        success = False
-        for lib in ("subprocess", "venv"):
-            if lib in code:
-                output = f"{lib} is not allowed"
-                outputs.append(output)
-                successes.append(success)
-                continue
-        try:
-            success, output = executor(code)
-        except TimeoutError as e:
-            print("Code timed out")
-            output = e
-        if not success and not return_status:
-            output = ""
-        outputs.append(output)
-        successes.append(success)
-    output = str(outputs[-1]).strip()
-    success = successes[-1]
-    if return_status:
-        return output, success
-    return output
-
-
-def postprocess_completion(text, return_status, last_code_block):
-    executor = PythonREPL()
-    result = execute_completion(executor, text, return_status=return_status, last_code_block=last_code_block)
-    del executor
-    return result
-
-
-def get_majority_vote(answers):
-    if not len(answers):
-        return 0
-    c = Counter(answers)
-    value, _ = c.most_common()[0]
-    return value
+if __name__ == "__main__":
+    interface.launch()